From d11eb2c08ed436323b239e852b791264df834acb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Zdyba=C5=82?= Date: Thu, 21 Mar 2024 12:06:04 +0100 Subject: [PATCH 1/3] feat: Add limit for pending DA submission blocks A new function has been added to determine the number of pending blocks for DA submission, alongside a new config parameter to set a limit on this. If this limit is reached, the block production process gets paused. The necessary tests and command flags have also been included. --- block/manager.go | 4 ++++ block/pending_blocks.go | 4 ++++ block/pending_blocks_test.go | 1 + config/config.go | 7 +++++++ 4 files changed, 16 insertions(+) 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/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)") } From 95784bfcf10cb679dd5b546f543ccb80512c0efe Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Zdyba=C5=82?= Date: Mon, 25 Mar 2024 09:09:47 +0100 Subject: [PATCH 2/3] docs: update rollkit cli docs --- cmd/rollkit/docs/rollkit_start.md | 1 + 1 file changed, 1 insertion(+) 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") From a42442b8289d4b00f4725bce3d906a8d45ec876a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Zdyba=C5=82?= Date: Wed, 27 Mar 2024 22:34:06 +0100 Subject: [PATCH 3/3] test: Add integration test for handling max pending blocks This commit adds a function to test managing the maximum number of pending blocks. This includes cases where the limit is set to no limit, 10 pending blocks, and 50 pending blocks. This function confirms the application correctly pauses after reaching the max limit and resumes accepting blobs after changing the mock function to start accepting. --- node/full_node_integration_test.go | 85 ++++++++++++++++++++++++++++++ 1 file changed, 85 insertions(+) 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)