diff --git a/block/manager.go b/block/manager.go index 9d29e8cde4..c731d9d957 100644 --- a/block/manager.go +++ b/block/manager.go @@ -715,6 +715,10 @@ func (m *Manager) publishBlock(ctx context.Context) error { return ErrNotProposer } + if m.conf.MaxPendingBlocks != 0 && m.pendingBlocks.numPendingBlocks() >= m.conf.MaxPendingBlocks { + return fmt.Errorf("number of blocks pending DA submission (%d) reached configured limit (%d)", m.pendingBlocks.numPendingBlocks(), m.conf.MaxPendingBlocks) + } + var ( lastCommit *types.Commit lastHeaderHash types.Hash diff --git a/block/pending_blocks.go b/block/pending_blocks.go index 8d52d54b89..0013df87c0 100644 --- a/block/pending_blocks.go +++ b/block/pending_blocks.go @@ -79,6 +79,10 @@ func (pb *PendingBlocks) isEmpty() bool { return pb.store.Height() == pb.lastSubmittedHeight.Load() } +func (pb *PendingBlocks) numPendingBlocks() uint64 { + return pb.store.Height() - pb.lastSubmittedHeight.Load() +} + func (pb *PendingBlocks) setLastSubmittedHeight(ctx context.Context, newLastSubmittedHeight uint64) { lsh := pb.lastSubmittedHeight.Load() diff --git a/block/pending_blocks_test.go b/block/pending_blocks_test.go index 65b9bbd4a9..8669d8b279 100644 --- a/block/pending_blocks_test.go +++ b/block/pending_blocks_test.go @@ -97,6 +97,7 @@ func checkRequirements(ctx context.Context, t *testing.T, pb *PendingBlocks, nBl blocks, err := pb.getPendingBlocks(ctx) require.NoError(t, err) require.Len(t, blocks, nBlocks) + require.Equal(t, uint64(len(blocks)), pb.numPendingBlocks()) require.True(t, sort.SliceIsSorted(blocks, func(i, j int) bool { return blocks[i].Height() < blocks[j].Height() })) diff --git a/cmd/rollkit/docs/rollkit_start.md b/cmd/rollkit/docs/rollkit_start.md index fb6ae452b6..cb5112f24a 100644 --- a/cmd/rollkit/docs/rollkit_start.md +++ b/cmd/rollkit/docs/rollkit_start.md @@ -40,6 +40,7 @@ rollkit start [flags] --rollkit.da_start_height uint starting DA block height (for syncing) --rollkit.lazy_aggregator wait for transactions, don't build empty blocks --rollkit.light run light client + --rollkit.max_pending_blocks uint limit of blocks pending DA submission (0 for no limit) --rollkit.trusted_hash string initial trusted hash to start the header exchange service --rpc.grpc_laddr string GRPC listen address (BroadcastTx only). Port required --rpc.laddr string RPC listen address. Port required (default "tcp://127.0.0.1:26657") diff --git a/config/config.go b/config/config.go index bb560c7da3..df826ae3d3 100644 --- a/config/config.go +++ b/config/config.go @@ -34,6 +34,8 @@ const ( FlagTrustedHash = "rollkit.trusted_hash" // FlagLazyAggregator is a flag for enabling lazy aggregation FlagLazyAggregator = "rollkit.lazy_aggregator" + // FlagMaxPendingBlocks is a flag to pause aggregator in case of large number of blocks pending DA submission + FlagMaxPendingBlocks = "rollkit.max_pending_blocks" ) // NodeConfig stores Rollkit node configuration. @@ -74,6 +76,9 @@ type BlockManagerConfig struct { DAStartHeight uint64 `mapstructure:"da_start_height"` // DAMempoolTTL is the number of DA blocks until transaction is dropped from the mempool. DAMempoolTTL uint64 `mapstructure:"da_mempool_ttl"` + // MaxPendingBlocks defines limit of blocks pending DA submission. 0 means no limit. + // When limit is reached, aggregator pauses block production. + MaxPendingBlocks uint64 `mapstructure:"max_pending_blocks"` } // GetNodeConfig translates Tendermint's configuration into Rollkit configuration. @@ -120,6 +125,7 @@ func (nc *NodeConfig) GetViperConfig(v *viper.Viper) error { nc.Light = v.GetBool(FlagLight) nc.TrustedHash = v.GetString(FlagTrustedHash) nc.TrustedHash = v.GetString(FlagTrustedHash) + nc.MaxPendingBlocks = v.GetUint64(FlagMaxPendingBlocks) return nil } @@ -140,4 +146,5 @@ func AddFlags(cmd *cobra.Command) { cmd.Flags().String(FlagDANamespace, def.DANamespace, "DA namespace to submit blob transactions") cmd.Flags().Bool(FlagLight, def.Light, "run light client") cmd.Flags().String(FlagTrustedHash, def.TrustedHash, "initial trusted hash to start the header exchange service") + cmd.Flags().Uint64(FlagMaxPendingBlocks, def.MaxPendingBlocks, "limit of blocks pending DA submission (0 for no limit)") } diff --git a/node/full_node_integration_test.go b/node/full_node_integration_test.go index e62f8148a8..4cd1d731d4 100644 --- a/node/full_node_integration_test.go +++ b/node/full_node_integration_test.go @@ -3,6 +3,7 @@ package node import ( "context" "crypto/rand" + "crypto/sha256" "errors" "fmt" mrand "math/rand" @@ -24,6 +25,7 @@ import ( "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" + goDA "github.com/rollkit/go-da" "github.com/rollkit/rollkit/config" "github.com/rollkit/rollkit/da" test "github.com/rollkit/rollkit/test/log" @@ -406,6 +408,89 @@ func TestSubmitBlocksToDA(t *testing.T) { } } +func TestMaxPending(t *testing.T) { + cases := []struct { + name string + maxPending uint64 + }{ + { + name: "no limit", + maxPending: 0, + }, + { + name: "10 pending blocks limit", + maxPending: 10, + }, + { + name: "50 pending blocks limit", + maxPending: 50, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + doTestMaxPending(tc.maxPending, t) + }) + } +} + +func doTestMaxPending(maxPending uint64, t *testing.T) { + require := require.New(t) + + clientNodes := 1 + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + nodes, _ := createNodes( + ctx, + context.Background(), + clientNodes, + config.BlockManagerConfig{ + DABlockTime: 20 * time.Millisecond, + BlockTime: 10 * time.Millisecond, + MaxPendingBlocks: maxPending, + }, + t, + ) + seq := nodes[0] + mockDA := &mocks.DA{} + + // make sure mock DA is not accepting any submissions + mockDA.On("MaxBlobSize", mock.Anything).Return(uint64(123456789), nil) + mockDA.On("Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil, errors.New("DA not available")) + + dalc := da.NewDAClient(mockDA, 1234, 5678, goDA.Namespace(MockDANamespace), log.NewNopLogger()) + require.NotNil(dalc) + seq.dalc = dalc + seq.blockManager.SetDALC(dalc) + + startNodeWithCleanup(t, seq) + + if maxPending == 0 { // if there is no limit, sequencer should produce blocks even DA is unavailable + require.NoError(waitForAtLeastNBlocks(seq, 3, Store)) + return + } else { // if there is a limit, sequencer should produce exactly maxPending blocks and pause + require.NoError(waitForAtLeastNBlocks(seq, int(maxPending), Store)) + // wait few block times and ensure that new blocks are not produced + time.Sleep(3 * seq.nodeConfig.BlockTime) + require.EqualValues(maxPending, seq.Store.Height()) + } + + // change mock function to start "accepting" blobs + mockDA.On("Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Unset() + mockDA.On("Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return( + func(ctx context.Context, blobs [][]byte, gasPrice float64, namespace []byte) ([][]byte, error) { + hashes := make([][]byte, len(blobs)) + for i, blob := range blobs { + sha := sha256.Sum256(blob) + hashes[i] = sha[:] + } + return hashes, nil + }) + + // wait for next block to ensure that sequencer is producing blocks again + require.NoError(waitForAtLeastNBlocks(seq, int(maxPending+1), Store)) +} + func testSingleAggregatorSingleFullNode(t *testing.T, source Source) { require := require.New(t)