feat: Add limit for pending DA submission blocks (#1609)

<!--
Please read and fill out this form before submitting your PR.

Please make sure you have reviewed our contributors guide before
submitting your
first PR.
-->

## Overview

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.

Resolves #1524

<!-- 
Please provide an explanation of the PR, including the appropriate
context,
background, goal, and rationale. If there is an issue with this
information,
please provide a tl;dr and link the issue. 
-->

## Checklist

<!-- 
Please complete the checklist to ensure that the PR is ready to be
reviewed.

IMPORTANT:
PRs should be left in Draft until the below checklist is completed.
-->

- [x] New and updated code has appropriate documentation
- [x] New and updated code has new and/or updated testing
- [x] Required CI checks are passing
- [x] Visual proof for any user facing features like CLI or
documentation updates
- [x] Linked issues closed with keywords


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

- **New Features**
- Introduced a limit on the number of blocks pending submission to
enhance system stability and manageability.
- Added a new configuration option to set the maximum number of pending
blocks during system initialization.
- **Tests**
- Implemented new integration tests to verify the block submission limit
functionality.
- **Documentation**
- Updated documentation to include guidance on configuring the maximum
number of pending blocks.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: tzdybal

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

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()
 

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()
 	}))

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")

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)")
 }

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)