diff --git a/block/manager.go b/block/manager.go index 517a855fe5..1e3c0ee6c1 100644 --- a/block/manager.go +++ b/block/manager.go @@ -827,35 +827,42 @@ func (m *Manager) recordMetrics(block *types.Block) { } func (m *Manager) submitBlocksToDA(ctx context.Context) error { - submitted := false + submittedAllBlocks := false backoff := initialBackoff - blocks := m.pendingBlocks.getPendingBlocks() - for attempt := 1; ctx.Err() == nil && !submitted && attempt <= maxSubmitAttempts; attempt++ { - res := m.dalc.SubmitBlocks(ctx, blocks) + blocksToSubmit := m.pendingBlocks.getPendingBlocks() + numTotalBlocks := len(blocksToSubmit) + numSubmittedBlocks := 0 + attempt := 0 + for ctx.Err() == nil && !submittedAllBlocks && attempt < maxSubmitAttempts { + res := m.dalc.SubmitBlocks(ctx, blocksToSubmit) switch res.Code { case da.StatusSuccess: - m.logger.Info("successfully submitted Rollkit block to DA layer", "daHeight", res.DAHeight, "count", res.SubmittedCount) - if int(res.SubmittedCount) == len(blocks) { - submitted = true + m.logger.Info("successfully submitted Rollkit blocks to DA layer", "daHeight", res.DAHeight, "count", res.SubmittedCount) + if res.SubmittedCount == uint64(len(blocksToSubmit)) { + submittedAllBlocks = true } - submittedBlocks := blocks[:res.SubmittedCount] + submittedBlocks, notSubmittedBlocks := blocksToSubmit[:res.SubmittedCount], blocksToSubmit[res.SubmittedCount:] + numSubmittedBlocks += len(submittedBlocks) for _, block := range submittedBlocks { m.blockCache.setDAIncluded(block.Hash().String()) } m.pendingBlocks.removeSubmittedBlocks(submittedBlocks) - case da.StatusError, da.StatusNotFound: - m.logger.Error("DA layer submission failed", "error", res.Message, "attempt", attempt) - time.Sleep(backoff) - backoff = m.exponentialBackoff(backoff) + blocksToSubmit = notSubmittedBlocks default: - m.logger.Error("DA layer unknown status", "error", res.Message, "attempt", attempt) + m.logger.Error("DA layer submission failed", "error", res.Message, "attempt", attempt) time.Sleep(backoff) backoff = m.exponentialBackoff(backoff) } + attempt += 1 } - if !submitted { - return fmt.Errorf("failed to submit block to DA layer after %d attempts", maxSubmitAttempts) + if !submittedAllBlocks { + return fmt.Errorf( + "failed to submit all blocks to DA layer, submitted %d of %d blocks after %d attempts", + numSubmittedBlocks, + numTotalBlocks, + maxSubmitAttempts, + ) } return nil } diff --git a/block/manager_test.go b/block/manager_test.go index 76347d8b19..b7db58824d 100644 --- a/block/manager_test.go +++ b/block/manager_test.go @@ -5,12 +5,42 @@ import ( "testing" cmtypes "github.com/cometbft/cometbft/types" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + goDATest "github.com/rollkit/go-da/test" + + "github.com/rollkit/rollkit/da" "github.com/rollkit/rollkit/store" + test "github.com/rollkit/rollkit/test/log" "github.com/rollkit/rollkit/types" ) +// Returns a minimalistic block manager +func getManager(t *testing.T) *Manager { + logger := test.NewFileLoggerCustom(t, test.TempLogFileName(t, t.Name())) + return &Manager{ + dalc: &da.DAClient{DA: goDATest.NewDummyDA(), GasPrice: -1, Logger: logger}, + blockCache: NewBlockCache(), + logger: logger, + } +} + +// getBlockBiggerThan generates a block with the given height bigger than the specified limit. +func getBlockBiggerThan(blockHeight, limit uint64) (*types.Block, error) { + for numTxs := 0; ; numTxs += 100 { + block := types.GetRandomBlock(blockHeight, numTxs) + blob, err := block.MarshalBinary() + if err != nil { + return nil, err + } + + if uint64(len(blob)) > limit { + return block, nil + } + } +} + func TestInitialStateClean(t *testing.T) { require := require.New(t) genesisDoc, _ := types.GetGenesisWithPrivkey() @@ -94,3 +124,75 @@ func TestIsDAIncluded(t *testing.T) { m.blockCache.setDAIncluded(hash.String()) require.True(m.IsDAIncluded(hash)) } + +func TestSubmitBlocksToDA(t *testing.T) { + require := require.New(t) + ctx := context.Background() + + m := getManager(t) + + maxDABlobSizeLimit, err := m.dalc.DA.MaxBlobSize(ctx) + require.NoError(err) + + testCases := []struct { + name string + blocks []*types.Block + isErrExpected bool + expectedPendingBlocksLength int + }{ + { + name: "happy path, all blocks A, B, C combine to less than maxDABlobSize", + blocks: []*types.Block{types.GetRandomBlock(1, 5), types.GetRandomBlock(2, 5), types.GetRandomBlock(3, 5)}, + isErrExpected: false, + expectedPendingBlocksLength: 0, + }, + { + name: "blocks A and B are submitted together without C because including C triggers blob size limit. C is submitted in a separate round", + blocks: func() []*types.Block { + // Find three blocks where two of them are under blob size limit + // but adding the third one exceeds the blob size limit + block1 := types.GetRandomBlock(1, 100) + blob1, err := block1.MarshalBinary() + require.NoError(err) + + block2 := types.GetRandomBlock(2, 100) + blob2, err := block2.MarshalBinary() + require.NoError(err) + + block3, err := getBlockBiggerThan(3, maxDABlobSizeLimit-uint64(len(blob1)+len(blob2))) + require.NoError(err) + + return []*types.Block{block1, block2, block3} + }(), + isErrExpected: false, + expectedPendingBlocksLength: 0, + }, + { + name: "A and B are submitted successfully but C is too big on its own, so C never gets submitted", + blocks: func() []*types.Block { + numBlocks, numTxs := 3, 5 + blocks := make([]*types.Block, numBlocks) + for i := 0; i < numBlocks-1; i++ { + blocks[i] = types.GetRandomBlock(uint64(i+1), numTxs) + } + blocks[2], err = getBlockBiggerThan(3, maxDABlobSizeLimit) + require.NoError(err) + return blocks + }(), + isErrExpected: true, + expectedPendingBlocksLength: 1, + }, + } + + for _, tc := range testCases { + m.pendingBlocks = NewPendingBlocks() + t.Run(tc.name, func(t *testing.T) { + for _, block := range tc.blocks { + m.pendingBlocks.addPendingBlock(block) + } + err := m.submitBlocksToDA(ctx) + assert.Equal(t, tc.isErrExpected, err != nil) + assert.Equal(t, tc.expectedPendingBlocksLength, len(m.pendingBlocks.getPendingBlocks())) + }) + } +}