fix(block manage): reducing batch overhead size when reaping Txs from…

… mempool (#886)

block/executor.go

@@ -96,11 +96,11 @@ func (e *Executor) InitChain(genesis *tmtypes.GenesisDoc, validators []*tmtypes.
 }
 
 // CreateBlock reaps transactions from mempool and builds a block.
-func (e *Executor) CreateBlock(height uint64, lastCommit *types.Commit, lastHeaderHash [32]byte, state *types.State, maxBytes uint64) *types.Block {
+func (e *Executor) CreateBlock(height uint64, lastCommit *types.Commit, lastHeaderHash [32]byte, state *types.State, maxBlockDataSizeBytes uint64) *types.Block {
 	if state.ConsensusParams.Block.MaxBytes > 0 {
-		maxBytes = min(maxBytes, uint64(state.ConsensusParams.Block.MaxBytes))
+		maxBlockDataSizeBytes = min(maxBlockDataSizeBytes, uint64(state.ConsensusParams.Block.MaxBytes))
 	}
-	mempoolTxs := e.mempool.ReapMaxBytesMaxGas(int64(maxBytes), state.ConsensusParams.Block.MaxGas)
+	mempoolTxs := e.mempool.ReapMaxBytesMaxGas(int64(maxBlockDataSizeBytes), state.ConsensusParams.Block.MaxGas)
 
 	block := &types.Block{
 		Header: types.Header{

block/produce.go

@@ -128,7 +128,9 @@ func (m *Manager) produceBlock(allowEmpty bool) (*types.Block, *types.Commit, er
 	} else if !errors.Is(err, gerr.ErrNotFound) {
 		return nil, nil, fmt.Errorf("load block: height: %d: %w: %w", newHeight, err, ErrNonRecoverable)
 	} else {
-		block = m.Executor.CreateBlock(newHeight, lastCommit, lastHeaderHash, m.State, m.Conf.BlockBatchMaxSizeBytes)
+		// limit to the max block data, so we don't create a block that is too big to fit in a batch
+		maxBlockDataSize := uint64(float64(m.Conf.BlockBatchMaxSizeBytes) * types.MaxBlockSizeAdjustment)
+		block = m.Executor.CreateBlock(newHeight, lastCommit, lastHeaderHash, m.State, maxBlockDataSize)
 		if !allowEmpty && len(block.Data.Txs) == 0 {
 			return nil, nil, fmt.Errorf("%w: %w", types.ErrSkippedEmptyBlock, ErrRecoverable)
 		}

block/submit_test.go

@@ -23,6 +23,81 @@ import (
 	"github.com/dymensionxyz/dymint/types"
 )
 
+// TestBatchOverhead tests the scenario where we have a single block that is very large, and occupies the entire batch size.
+// This test is to ensure the value of 90% for types.MaxBlockSizeAdjustment is valid
+// 2 use cases:
+// 1. single block with single large tx
+// 2. single block with multiple small tx
+func TestBatchOverhead(t *testing.T) {
+	manager, err := testutil.GetManager(testutil.GetManagerConfig(), nil, nil, 1, 1, 0, nil, nil)
+	require.NoError(t, err)
+	require.NotNil(t, manager)
+
+	maxBatchSize := uint64(10_000)                                            // 10KB
+	maxTxData := uint64(float64(maxBatchSize) * types.MaxBlockSizeAdjustment) // 90% of maxBatchSize
+
+	// first batch with single block with single large tx
+	var tcases = []struct {
+		name string
+		nTxs int
+	}{
+		{
+			name: "single block with single large tx",
+			nTxs: 1,
+		},
+		{
+			name: "single block with multiple small tx",
+			nTxs: 100,
+		},
+	}
+
+	for _, tcase := range tcases {
+		blocks, err := testutil.GenerateBlocks(1, 1, manager.ProposerKey)
+		require.NoError(t, err)
+		block := blocks[0]
+
+		mallete := func(nTxs int) {
+			txSize := maxTxData / uint64(nTxs)
+
+			block.Data = types.Data{
+				Txs: make(types.Txs, nTxs),
+			}
+
+			for i := 0; i < nTxs; i++ {
+				block.Data.Txs[0] = testutil.GetRandomBytes(txSize)
+			}
+		}
+
+		mallete(tcase.nTxs)
+
+		commits, err := testutil.GenerateCommits(blocks, manager.ProposerKey)
+		require.NoError(t, err)
+		commit := commits[0]
+
+		batch := types.Batch{
+			StartHeight: 1,
+			EndHeight:   1,
+			Blocks:      blocks,
+			Commits:     commits,
+		}
+
+		batchSize := batch.ToProto().Size()
+
+		var blocksize, commitSize int
+		blocksize = block.ToProto().Size()
+		commitSize = commit.ToProto().Size()
+
+		// we assert that the batch size is not larger than the maxBatchSize
+		assert.LessOrEqual(t, batchSize, int(maxBatchSize), tcase.name)
+
+		t.Log("Batch size:", batchSize, "Max batch size:", maxBatchSize, tcase.name)
+		t.Log("Commit size:", commitSize, tcase.name)
+		t.Log("Block size:", blocksize, tcase.name)
+		t.Log("Tx size:", maxTxData, "num of txs:", len(blocks[0].Data.Txs), tcase.name)
+		t.Log("Overhead:", batchSize-int(maxTxData), tcase.name)
+	}
+}
+
 func TestBatchSubmissionHappyFlow(t *testing.T) {
 	require := require.New(t)
 	app := testutil.GetAppMock()

block/types.go

@@ -8,6 +8,7 @@ import (
 )
 
 // TODO: move to types package
+
 type blockSource string
 
 const (

config/config_test.go

@@ -193,7 +193,7 @@ func fullNodeConfig() config.NodeConfig {
 			BatchSubmitMaxTime:     20 * time.Second,
 			MaxSupportedBatchSkew:  10,
 			NamespaceID:            "test",
-			BlockBatchMaxSizeBytes: 1,
+			BlockBatchMaxSizeBytes: 10000,
 		},
 		DALayer:         "celestia",
 		DAConfig:        "da-config",

testutil/types.go

@@ -38,11 +38,11 @@ func createRandomHashes() [][32]byte {
 
 func GetRandomTx() types.Tx {
 	n, _ := rand.Int(rand.Reader, big.NewInt(100))
-	size := int(n.Int64()) + 100
+	size := uint64(n.Int64()) + 100
 	return types.Tx(GetRandomBytes(size))
 }
 
-func GetRandomBytes(n int) []byte {
+func GetRandomBytes(n uint64) []byte {
 	data := make([]byte, n)
 	_, _ = rand.Read(data)
 	return data

types/batch.go

@@ -1,5 +1,9 @@
 package types
 
+const (
+	MaxBlockSizeAdjustment = 0.9 // have a safety margin of 10% in regard of MaxBlockBatchSizeBytes
+)
+
 // Batch defines a struct for block aggregation for support of batching.
 // TODO: maybe change to BlockBatch
 type Batch struct {