From 8ec09c1d901d9a7789f850bec6bc0dd5860dee63 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Zdyba=C5=82?= Date: Tue, 27 Feb 2024 15:06:39 +0100 Subject: [PATCH 01/20] test: Add test for pending blocks and refactor code TestPendingBlocks should fail to proof the existence of the bug. This commit introduces a new test to check pending blocks as described in issue #1548. It mimics the behavior of a node producing blocks, stopping and restarting, then producing more blocks. --- Makefile | 1 + node/full_node_integration_test.go | 19 ++- node/full_node_test.go | 116 ++++++++++++++ test/mocks/DA.go | 236 +++++++++++++++++++++++++++++ 4 files changed, 365 insertions(+), 7 deletions(-) create mode 100644 test/mocks/DA.go diff --git a/Makefile b/Makefile index 815501d69f..69d461a1cd 100644 --- a/Makefile +++ b/Makefile @@ -83,6 +83,7 @@ mock-gen: @echo "-> Generating mocks" mockery --output test/mocks --srcpkg github.com/cometbft/cometbft/rpc/client --name Client mockery --output test/mocks --srcpkg github.com/cometbft/cometbft/abci/types --name Application + mockery --output test/mocks --srcpkg github.com/rollkit/go-da --name DA .PHONY: mock-gen diff --git a/node/full_node_integration_test.go b/node/full_node_integration_test.go index 874073326f..649a6f3658 100644 --- a/node/full_node_integration_test.go +++ b/node/full_node_integration_test.go @@ -509,6 +509,17 @@ func testSingleAggregatorSingleFullNodeSingleLightNode(t *testing.T) { require.NoError(verifyNodesSynced(fullNode, lightNode, Header)) } +func getMockApplication() *mocks.Application { + app := &mocks.Application{} + app.On("InitChain", mock.Anything, mock.Anything).Return(&abci.ResponseInitChain{}, nil) + app.On("CheckTx", mock.Anything, mock.Anything).Return(&abci.ResponseCheckTx{}, nil) + app.On("Commit", mock.Anything, mock.Anything).Return(&abci.ResponseCommit{}, nil) + app.On("PrepareProposal", mock.Anything, mock.Anything).Return(prepareProposalResponse).Maybe() + app.On("ProcessProposal", mock.Anything, mock.Anything).Return(&abci.ResponseProcessProposal{Status: abci.ResponseProcessProposal_ACCEPT}, nil) + app.On("FinalizeBlock", mock.Anything, mock.Anything).Return(finalizeBlockResponse) + return app +} + // Starts the given nodes using the given wait group to synchronize them // and wait for them to gossip transactions func startNodes(nodes []*FullNode, apps []*mocks.Application, t *testing.T) { @@ -613,13 +624,7 @@ func createNode(ctx context.Context, n int, aggregator bool, isLight bool, keys } p2pConfig.Seeds = strings.TrimSuffix(p2pConfig.Seeds, ",") - app := &mocks.Application{} - app.On("InitChain", mock.Anything, mock.Anything).Return(&abci.ResponseInitChain{}, nil) - app.On("CheckTx", mock.Anything, mock.Anything).Return(&abci.ResponseCheckTx{}, nil) - app.On("Commit", mock.Anything, mock.Anything).Return(&abci.ResponseCommit{}, nil) - app.On("PrepareProposal", mock.Anything, mock.Anything).Return(prepareProposalResponse).Maybe() - app.On("ProcessProposal", mock.Anything, mock.Anything).Return(&abci.ResponseProcessProposal{Status: abci.ResponseProcessProposal_ACCEPT}, nil) - app.On("FinalizeBlock", mock.Anything, mock.Anything).Return(finalizeBlockResponse) + app := getMockApplication() if ctx == nil { ctx = context.Background() diff --git a/node/full_node_test.go b/node/full_node_test.go index 9fb46b4069..f1099b080c 100644 --- a/node/full_node_test.go +++ b/node/full_node_test.go @@ -3,10 +3,19 @@ package node import ( "context" "crypto/rand" + "crypto/sha256" + "errors" "fmt" + "os" "testing" "time" + cmconfig "github.com/cometbft/cometbft/config" + "github.com/cometbft/cometbft/proxy" + goDA "github.com/rollkit/go-da" + "github.com/rollkit/rollkit/config" + test "github.com/rollkit/rollkit/test/log" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" @@ -149,6 +158,113 @@ func TestInvalidBlocksIgnored(t *testing.T) { require.False(t, manager.IsBlockHashSeen(junkProposerBlock.Hash().String())) } +// TestPendingBlocks is a test for bug described in https://github.com/rollkit/rollkit/issues/1548 +func TestPendingBlocks(t *testing.T) { + ctx := context.Background() + + mockDA := new(mocks.DA) + mockDA.On("MaxBlobSize", mock.Anything).Return(uint64(10240), nil) + mockDA.On("Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil, errors.New("DA not available")) + + dac := &da.DAClient{ + DA: mockDA, + Namespace: goDA.Namespace(MockNamespace), + GasPrice: 1234, + } + dbPath, err := os.MkdirTemp("", "testdb") + require.NoError(t, err) + defer func() { + _ = os.RemoveAll(dbPath) + }() + + node, _ := createAggregatorWithPersistence(ctx, dbPath, dac, t) + err = node.Start() + assert.NoError(t, err) + + const firstRunBlocks = 10 + + err = waitForAtLeastNBlocks(node, firstRunBlocks, Store) + assert.NoError(t, err) + + err = node.Stop() + assert.NoError(t, err) + + // create & start new node + node, _ = createAggregatorWithPersistence(ctx, dbPath, dac, t) + + // reset DA mock to ensure that Submit was called + mockDA.On("Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Unset() + + // mock submit function to just return some hash and count the number of submitted blobs + // node will be stopped after producing at least firstRunBlocks blocks + // restarted node should get blocks from first and second run (more than firstRunBlocks + // TODO(tzdybal): this seems fragile and probably should be improved (to deserialize blocks and check heights) + uniqueBlobs := make(map[string]uint64) + 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[:] + uniqueBlobs[string(sha[:])]++ + } + return hashes, nil + }) + + err = node.Start() + assert.NoError(t, err) + + // let node produce few more blocks + err = waitForAtLeastNBlocks(node, firstRunBlocks+5, Store) + assert.NoError(t, err) + + err = node.Stop() + assert.NoError(t, err) + + assert.Greater(t, len(uniqueBlobs), firstRunBlocks) + mock.AssertExpectationsForObjects(t, mockDA) +} + +func createAggregatorWithPersistence(ctx context.Context, dbPath string, dalc *da.DAClient, t *testing.T) (Node, *mocks.Application) { + t.Helper() + + key, _, _ := crypto.GenerateEd25519Key(rand.Reader) + genesis, genesisValidatorKey := types.GetGenesisWithPrivkey() + signingKey, err := types.PrivKeyToSigningKey(genesisValidatorKey) + require.NoError(t, err) + + app := getMockApplication() + + node, err := NewNode( + ctx, + config.NodeConfig{ + DBPath: dbPath, + DAAddress: MockServerAddr, + DANamespace: MockNamespace, + Aggregator: true, + BlockManagerConfig: config.BlockManagerConfig{ + BlockTime: 100 * time.Millisecond, + DABlockTime: 300 * time.Millisecond, + }, + Light: false, + }, + key, + signingKey, + proxy.NewLocalClientCreator(app), + genesis, + DefaultMetricsProvider(cmconfig.DefaultInstrumentationConfig()), + test.NewFileLoggerCustom(t, test.TempLogFileName(t, "")), + ) + require.NoError(t, err) + require.NotNil(t, node) + + fullNode := node.(*FullNode) + fullNode.dalc = dalc + fullNode.blockManager.SetDALC(dalc) + + return fullNode, app +} + // setupMockApplication initializes a mock application func setupMockApplication() *mocks.Application { app := &mocks.Application{} diff --git a/test/mocks/DA.go b/test/mocks/DA.go new file mode 100644 index 0000000000..28562b9063 --- /dev/null +++ b/test/mocks/DA.go @@ -0,0 +1,236 @@ +// Code generated by mockery v2.38.0. DO NOT EDIT. + +package mocks + +import ( + context "context" + + mock "github.com/stretchr/testify/mock" +) + +// DA is an autogenerated mock type for the DA type +type DA struct { + mock.Mock +} + +// Commit provides a mock function with given fields: ctx, blobs, namespace +func (_m *DA) Commit(ctx context.Context, blobs [][]byte, namespace []byte) ([][]byte, error) { + ret := _m.Called(ctx, blobs, namespace) + + if len(ret) == 0 { + panic("no return value specified for Commit") + } + + var r0 [][]byte + var r1 error + if rf, ok := ret.Get(0).(func(context.Context, [][]byte, []byte) ([][]byte, error)); ok { + return rf(ctx, blobs, namespace) + } + if rf, ok := ret.Get(0).(func(context.Context, [][]byte, []byte) [][]byte); ok { + r0 = rf(ctx, blobs, namespace) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).([][]byte) + } + } + + if rf, ok := ret.Get(1).(func(context.Context, [][]byte, []byte) error); ok { + r1 = rf(ctx, blobs, namespace) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// Get provides a mock function with given fields: ctx, ids, namespace +func (_m *DA) Get(ctx context.Context, ids [][]byte, namespace []byte) ([][]byte, error) { + ret := _m.Called(ctx, ids, namespace) + + if len(ret) == 0 { + panic("no return value specified for Get") + } + + var r0 [][]byte + var r1 error + if rf, ok := ret.Get(0).(func(context.Context, [][]byte, []byte) ([][]byte, error)); ok { + return rf(ctx, ids, namespace) + } + if rf, ok := ret.Get(0).(func(context.Context, [][]byte, []byte) [][]byte); ok { + r0 = rf(ctx, ids, namespace) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).([][]byte) + } + } + + if rf, ok := ret.Get(1).(func(context.Context, [][]byte, []byte) error); ok { + r1 = rf(ctx, ids, namespace) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// GetIDs provides a mock function with given fields: ctx, height, namespace +func (_m *DA) GetIDs(ctx context.Context, height uint64, namespace []byte) ([][]byte, error) { + ret := _m.Called(ctx, height, namespace) + + if len(ret) == 0 { + panic("no return value specified for GetIDs") + } + + var r0 [][]byte + var r1 error + if rf, ok := ret.Get(0).(func(context.Context, uint64, []byte) ([][]byte, error)); ok { + return rf(ctx, height, namespace) + } + if rf, ok := ret.Get(0).(func(context.Context, uint64, []byte) [][]byte); ok { + r0 = rf(ctx, height, namespace) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).([][]byte) + } + } + + if rf, ok := ret.Get(1).(func(context.Context, uint64, []byte) error); ok { + r1 = rf(ctx, height, namespace) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// GetProofs provides a mock function with given fields: ctx, ids, namespace +func (_m *DA) GetProofs(ctx context.Context, ids [][]byte, namespace []byte) ([][]byte, error) { + ret := _m.Called(ctx, ids, namespace) + + if len(ret) == 0 { + panic("no return value specified for GetProofs") + } + + var r0 [][]byte + var r1 error + if rf, ok := ret.Get(0).(func(context.Context, [][]byte, []byte) ([][]byte, error)); ok { + return rf(ctx, ids, namespace) + } + if rf, ok := ret.Get(0).(func(context.Context, [][]byte, []byte) [][]byte); ok { + r0 = rf(ctx, ids, namespace) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).([][]byte) + } + } + + if rf, ok := ret.Get(1).(func(context.Context, [][]byte, []byte) error); ok { + r1 = rf(ctx, ids, namespace) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// MaxBlobSize provides a mock function with given fields: ctx +func (_m *DA) MaxBlobSize(ctx context.Context) (uint64, error) { + ret := _m.Called(ctx) + + if len(ret) == 0 { + panic("no return value specified for MaxBlobSize") + } + + var r0 uint64 + var r1 error + if rf, ok := ret.Get(0).(func(context.Context) (uint64, error)); ok { + return rf(ctx) + } + if rf, ok := ret.Get(0).(func(context.Context) uint64); ok { + r0 = rf(ctx) + } else { + r0 = ret.Get(0).(uint64) + } + + if rf, ok := ret.Get(1).(func(context.Context) error); ok { + r1 = rf(ctx) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// Submit provides a mock function with given fields: ctx, blobs, gasPrice, namespace +func (_m *DA) Submit(ctx context.Context, blobs [][]byte, gasPrice float64, namespace []byte) ([][]byte, error) { + ret := _m.Called(ctx, blobs, gasPrice, namespace) + + if len(ret) == 0 { + panic("no return value specified for Submit") + } + + var r0 [][]byte + var r1 error + if rf, ok := ret.Get(0).(func(context.Context, [][]byte, float64, []byte) ([][]byte, error)); ok { + return rf(ctx, blobs, gasPrice, namespace) + } + if rf, ok := ret.Get(0).(func(context.Context, [][]byte, float64, []byte) [][]byte); ok { + r0 = rf(ctx, blobs, gasPrice, namespace) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).([][]byte) + } + } + + if rf, ok := ret.Get(1).(func(context.Context, [][]byte, float64, []byte) error); ok { + r1 = rf(ctx, blobs, gasPrice, namespace) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// Validate provides a mock function with given fields: ctx, ids, proofs, namespace +func (_m *DA) Validate(ctx context.Context, ids [][]byte, proofs [][]byte, namespace []byte) ([]bool, error) { + ret := _m.Called(ctx, ids, proofs, namespace) + + if len(ret) == 0 { + panic("no return value specified for Validate") + } + + var r0 []bool + var r1 error + if rf, ok := ret.Get(0).(func(context.Context, [][]byte, [][]byte, []byte) ([]bool, error)); ok { + return rf(ctx, ids, proofs, namespace) + } + if rf, ok := ret.Get(0).(func(context.Context, [][]byte, [][]byte, []byte) []bool); ok { + r0 = rf(ctx, ids, proofs, namespace) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).([]bool) + } + } + + if rf, ok := ret.Get(1).(func(context.Context, [][]byte, [][]byte, []byte) error); ok { + r1 = rf(ctx, ids, proofs, namespace) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// NewDA creates a new instance of DA. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. +// The first argument is typically a *testing.T value. +func NewDA(t interface { + mock.TestingT + Cleanup(func()) +}) *DA { + mock := &DA{} + mock.Mock.Test(t) + + t.Cleanup(func() { mock.AssertExpectations(t) }) + + return mock +} From 054a1c013fa9996e8be6869ceada8b4edff6ec46 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Zdyba=C5=82?= Date: Wed, 28 Feb 2024 10:45:13 +0100 Subject: [PATCH 02/20] wip: reimplement PendingBlocks using store and 'high-water-mark' Pending blocks are retrieved directly from store. Next step is to save latest submitted block height to store (store interface needs to be modified). Tests are expected to fail. --- block/manager.go | 10 ++++- block/manager_test.go | 11 +++-- block/pending_blocks.go | 83 ++++++++++++++++++++++-------------- block/pending_blocks_test.go | 29 +++++++++---- node/full_node_test.go | 1 + 5 files changed, 87 insertions(+), 47 deletions(-) diff --git a/block/manager.go b/block/manager.go index 727a7f1d7d..b50c5dbde7 100644 --- a/block/manager.go +++ b/block/manager.go @@ -219,7 +219,7 @@ func NewManager( validatorSet: &valSet, txsAvailable: txsAvailableCh, buildingBlock: false, - pendingBlocks: NewPendingBlocks(), + pendingBlocks: NewPendingBlocks(store), metrics: seqMetrics, } return agg, nil @@ -840,7 +840,13 @@ func (m *Manager) recordMetrics(block *types.Block) { func (m *Manager) submitBlocksToDA(ctx context.Context) error { submittedAllBlocks := false backoff := initialBackoff - blocksToSubmit := m.pendingBlocks.getPendingBlocks() + blocksToSubmit, err := m.pendingBlocks.getPendingBlocks() + if len(blocksToSubmit) == 0 { + return err + } + if err != nil { + m.logger.Error("error while fetching blocks pending DA", "err", err) + } numSubmittedBlocks := 0 attempt := 0 maxBlobSize, err := m.dalc.DA.MaxBlobSize(ctx) diff --git a/block/manager_test.go b/block/manager_test.go index 589397e0aa..ddd05cd63f 100644 --- a/block/manager_test.go +++ b/block/manager_test.go @@ -162,7 +162,7 @@ func TestSubmitBlocksToMockDA(t *testing.T) { On("Submit", blobs, 1.0*1.2*1.2, []byte(nil)). Return([][]byte{bytes.Repeat([]byte{0x00}, 8)}, nil) - m.pendingBlocks = NewPendingBlocks() + m.pendingBlocks = NewPendingBlocks(m.store) m.pendingBlocks.addPendingBlock(block) err = m.submitBlocksToDA(ctx) require.NoError(t, err) @@ -171,6 +171,7 @@ func TestSubmitBlocksToMockDA(t *testing.T) { } func TestSubmitBlocksToDA(t *testing.T) { + assert := assert.New(t) require := require.New(t) ctx := context.Background() @@ -230,14 +231,18 @@ func TestSubmitBlocksToDA(t *testing.T) { } for _, tc := range testCases { - m.pendingBlocks = NewPendingBlocks() + kvStore, err := store.NewDefaultInMemoryKVStore() + require.NoError(err) + m.pendingBlocks = NewPendingBlocks(store.New(kvStore)) 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())) + blocks, err := m.pendingBlocks.getPendingBlocks() + assert.NoError(err) + assert.Equal(tc.expectedPendingBlocksLength, len(blocks)) }) } } diff --git a/block/pending_blocks.go b/block/pending_blocks.go index 0a04e55b49..c8cfdac3b2 100644 --- a/block/pending_blocks.go +++ b/block/pending_blocks.go @@ -1,64 +1,81 @@ package block import ( - "sort" - "sync" + "context" + "sync/atomic" + + "github.com/rollkit/rollkit/store" "github.com/rollkit/rollkit/types" ) // PendingBlocks maintains blocks that need to be published to DA layer +// +// Important assertions: +// - blocks are safely stored in database before submission to DA +// - blocks are always pushed to DA in order (by height) +// - DA submission of multiple blocks is atomic - it's impossible to submit only part of a batch +// +// lastSubmittedHeight is updated only after receiving confirmation from DA. +// Worst case scenario is when blocks was successfully submitted to DA, but confirmation was not received (e.g. node was +// restarted, networking issue occurred). In this case blocks are re-submitted to DA (it's extra cost). +// rollkit is able to skip duplicate blocks so this shouldn't affect full nodes. +// TODO(tzdybal): batch size type PendingBlocks struct { - pendingBlocks map[uint64]*types.Block - mtx *sync.RWMutex + store store.Store + + // lastSubmittedHeight holds information about last block successfully submitted to DA + lastSubmittedHeight atomic.Uint64 } // NewPendingBlocks returns a new PendingBlocks struct -func NewPendingBlocks() *PendingBlocks { +func NewPendingBlocks(store store.Store) *PendingBlocks { return &PendingBlocks{ - pendingBlocks: make(map[uint64]*types.Block), - mtx: new(sync.RWMutex), + store: store, + // TODO(tzdybal): lastSubmittedHeight from store } } // getPendingBlocks returns a sorted slice of pending blocks // that need to be published to DA layer in order of block height -func (pb *PendingBlocks) getPendingBlocks() []*types.Block { - blocks := copyBlocks(pb) - sort.Slice(blocks, func(i, j int) bool { - return blocks[i].Height() < blocks[j].Height() - }) - return blocks -} +func (pb *PendingBlocks) getPendingBlocks() ([]*types.Block, error) { + height := pb.store.Height() + lastSubmitted := pb.lastSubmittedHeight.Load() -// copyBlocks creates a copy of the pending blocks in a thread-safe manner. -// It returns a slice of pointers to the copied blocks. -func copyBlocks(pb *PendingBlocks) []*types.Block { - pb.mtx.RLock() - defer pb.mtx.RUnlock() - blocks := make([]*types.Block, 0, len(pb.pendingBlocks)) - for _, block := range pb.pendingBlocks { + // TODO(tzdybal) - lastSubmitted should never be > than height in final implementation + if lastSubmitted >= height { + return nil, nil + } + + blocks := make([]*types.Block, 0, height-lastSubmitted) + for i := lastSubmitted + 1; i <= height; i++ { + block, err := pb.store.GetBlock(context.TODO(), i) + if err != nil { + // return as much as possible + error information + return blocks, err + } blocks = append(blocks, block) } - return blocks + return blocks, nil } func (pb *PendingBlocks) isEmpty() bool { - pb.mtx.RLock() - defer pb.mtx.RUnlock() - return len(pb.pendingBlocks) == 0 + return pb.store.Height() == pb.lastSubmittedHeight.Load() } -func (pb *PendingBlocks) addPendingBlock(block *types.Block) { - pb.mtx.Lock() - defer pb.mtx.Unlock() - pb.pendingBlocks[block.Height()] = block +func (pb *PendingBlocks) addPendingBlock(_ *types.Block) { + // TODO(tzdybal): remove this method } +// TODO(tzdybal): change signature (accept height) func (pb *PendingBlocks) removeSubmittedBlocks(blocks []*types.Block) { - pb.mtx.Lock() - defer pb.mtx.Unlock() - for _, block := range blocks { - delete(pb.pendingBlocks, block.Height()) + if len(blocks) == 0 { + return + } + height := blocks[len(blocks)-1].Height() + lastSubmitted := pb.lastSubmittedHeight.Load() + + if height > lastSubmitted { + pb.lastSubmittedHeight.CompareAndSwap(lastSubmitted, height) } } diff --git a/block/pending_blocks_test.go b/block/pending_blocks_test.go index ef43130ca7..583b469b22 100644 --- a/block/pending_blocks_test.go +++ b/block/pending_blocks_test.go @@ -4,6 +4,8 @@ import ( "sort" "testing" + "github.com/rollkit/rollkit/store" + "github.com/stretchr/testify/require" "github.com/rollkit/rollkit/types" @@ -11,11 +13,11 @@ import ( func TestGetPendingBlocks(t *testing.T) { require := require.New(t) - pb := NewPendingBlocks() + pb := newPendingBlocks(t) for i := uint64(0); i < 5; i++ { pb.addPendingBlock(types.GetRandomBlock(i, 0)) } - blocks := pb.getPendingBlocks() + blocks, _ := pb.getPendingBlocks() require.True(sort.SliceIsSorted(blocks, func(i, j int) bool { return blocks[i].Height() < blocks[j].Height() })) @@ -23,18 +25,18 @@ func TestGetPendingBlocks(t *testing.T) { func TestRemoveSubmittedBlocks(t *testing.T) { require := require.New(t) - pb := NewPendingBlocks() + pb := newPendingBlocks(t) for i := uint64(0); i < 5; i++ { pb.addPendingBlock(types.GetRandomBlock(i, 0)) } - blocks := pb.getPendingBlocks() + blocks, _ := pb.getPendingBlocks() pb.removeSubmittedBlocks(blocks) require.True(pb.isEmpty()) } func TestRemoveSubsetOfBlocks(t *testing.T) { require := require.New(t) - pb := NewPendingBlocks() + pb := newPendingBlocks(t) for i := uint64(0); i < 5; i++ { pb.addPendingBlock(types.GetRandomBlock(i, 0)) } @@ -43,7 +45,8 @@ func TestRemoveSubsetOfBlocks(t *testing.T) { types.GetRandomBlock(1, 0), types.GetRandomBlock(2, 0), }) - remainingBlocks := pb.getPendingBlocks() + remainingBlocks, err := pb.getPendingBlocks() + require.NoError(err) require.Len(remainingBlocks, 3, "There should be 3 blocks remaining") for _, block := range remainingBlocks { require.Contains([]uint64{0, 3, 4}, block.Height(), "Only blocks with height 0, 3, and 4 should remain") @@ -52,18 +55,20 @@ func TestRemoveSubsetOfBlocks(t *testing.T) { func TestRemoveAllBlocksAndVerifyEmpty(t *testing.T) { require := require.New(t) - pb := NewPendingBlocks() + pb := newPendingBlocks(t) for i := uint64(0); i < 5; i++ { pb.addPendingBlock(types.GetRandomBlock(i, 0)) } // Remove all blocks - pb.removeSubmittedBlocks(pb.getPendingBlocks()) + blocks, err := pb.getPendingBlocks() + require.NoError(err) + pb.removeSubmittedBlocks(blocks) require.True(pb.isEmpty(), "PendingBlocks should be empty after removing all blocks") } func TestRemoveBlocksFromEmptyPendingBlocks(t *testing.T) { require := require.New(t) - pb := NewPendingBlocks() + pb := newPendingBlocks(t) // Attempt to remove blocks from an empty PendingBlocks require.NotPanics(func() { pb.removeSubmittedBlocks([]*types.Block{ @@ -72,3 +77,9 @@ func TestRemoveBlocksFromEmptyPendingBlocks(t *testing.T) { }) }, "Removing blocks from an empty PendingBlocks should not cause a panic") } + +func newPendingBlocks(t *testing.T) *PendingBlocks { + kv, err := store.NewDefaultInMemoryKVStore() + require.NoError(t, err) + return NewPendingBlocks(store.New(kv)) +} diff --git a/node/full_node_test.go b/node/full_node_test.go index f1099b080c..48b1eceb4f 100644 --- a/node/full_node_test.go +++ b/node/full_node_test.go @@ -12,6 +12,7 @@ import ( cmconfig "github.com/cometbft/cometbft/config" "github.com/cometbft/cometbft/proxy" + goDA "github.com/rollkit/go-da" "github.com/rollkit/rollkit/config" test "github.com/rollkit/rollkit/test/log" From d634794a8a69b57990cdcf74d2e5c02835c55260 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Zdyba=C5=82?= Date: Wed, 28 Feb 2024 11:13:15 +0100 Subject: [PATCH 03/20] feat: add metadata handling to store interface Two new methods have been added to the store interface: SetMetadata and GetMetadata. These methods allow for the storage and retrieval of arbitrary data associated with a specified key. Demonstration of usage and functionality is provided through additional tests in store_test.go. --- store/store.go | 17 +++++++++++++++++ store/store_test.go | 36 ++++++++++++++++++++++++++++++++++++ store/types.go | 8 ++++++++ 3 files changed, 61 insertions(+) diff --git a/store/store.go b/store/store.go index c6c5ce6875..7282bc33b7 100644 --- a/store/store.go +++ b/store/store.go @@ -22,6 +22,7 @@ var ( commitPrefix = "c" statePrefix = "s" responsesPrefix = "r" + metaPrefix = "m" ) // DefaultStore is a default store implmementation. @@ -207,6 +208,18 @@ func (s *DefaultStore) GetState(ctx context.Context) (types.State, error) { return state, err } +// SetMetadata saves arbitrary value in the store. +// +// Metadata is separated from other data by using prefix in KV. +func (s *DefaultStore) SetMetadata(ctx context.Context, key string, value []byte) error { + return s.db.Put(ctx, ds.NewKey(getMetaKey(key)), value) +} + +// GetMetadata returns values stored for given key with SetMetadata. +func (s *DefaultStore) GetMetadata(ctx context.Context, key string) ([]byte, error) { + return s.db.Get(ctx, ds.NewKey(getMetaKey(key))) +} + // loadHashFromIndex returns the hash of a block given its height func (s *DefaultStore) loadHashFromIndex(ctx context.Context, height uint64) (header.Hash, error) { blob, err := s.db.Get(ctx, ds.NewKey(getIndexKey(height))) @@ -239,3 +252,7 @@ func getStateKey() string { func getResponsesKey(height uint64) string { return GenerateKey([]interface{}{responsesPrefix, height}) } + +func getMetaKey(key string) string { + return GenerateKey([]interface{}{metaPrefix, key}) +} diff --git a/store/store_test.go b/store/store_test.go index 4762586677..12d685b21e 100644 --- a/store/store_test.go +++ b/store/store_test.go @@ -2,6 +2,7 @@ package store import ( "context" + "fmt" "os" "testing" @@ -206,3 +207,38 @@ func TestBlockResponses(t *testing.T) { assert.NotNil(resp) assert.Equal(expected, resp) } + +func TestMetadata(t *testing.T) { + t.Parallel() + assert := assert.New(t) + require := require.New(t) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + kv, err := NewDefaultInMemoryKVStore() + require.NoError(err) + s := New(kv) + + getKey := func(i int) string { + return fmt.Sprintf("key %d", i) + } + getValue := func(i int) []byte { + return []byte(fmt.Sprintf("value %d", i)) + } + + const n = 5 + for i := 0; i < n; i++ { + assert.NoError(s.SetMetadata(ctx, getKey(i), getValue(i))) + } + + for i := 0; i < n; i++ { + value, err := s.GetMetadata(ctx, getKey(i)) + assert.NoError(err) + assert.Equal(getValue(i), value) + } + + v, err := s.GetMetadata(ctx, "unused key") + assert.Error(err) + assert.Nil(v) +} diff --git a/store/types.go b/store/types.go index 2c303d2206..bed4a4f327 100644 --- a/store/types.go +++ b/store/types.go @@ -41,6 +41,14 @@ type Store interface { // GetState returns last state saved with UpdateState. GetState(ctx context.Context) (types.State, error) + // SetMetadata saves arbitrary value in the store. + // + // This method enables rollkit to safely persist any information. + SetMetadata(ctx context.Context, key string, value []byte) error + + // GetMetadata returns values stored for given key with SetMetadata. + GetMetadata(ctx context.Context, key string) ([]byte, error) + // Close safely closes underlying data storage, to ensure that data is actually saved. Close() error } From d973790789b8a28327eaa5f9889848e167410c80 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Zdyba=C5=82?= Date: Wed, 28 Feb 2024 22:56:19 +0100 Subject: [PATCH 04/20] feat: load lastBlockHeight from store tests are improved to reflect added assertions (block submission always in order) and code changes (in memory kv couldn't handle big values O_o) --- block/manager_test.go | 34 ++++++++++++++++++++++++---- block/pending_blocks.go | 44 ++++++++++++++++++++++++++++++++---- block/pending_blocks_test.go | 11 ++++++--- 3 files changed, 78 insertions(+), 11 deletions(-) diff --git a/block/manager_test.go b/block/manager_test.go index ddd05cd63f..ae984e738e 100644 --- a/block/manager_test.go +++ b/block/manager_test.go @@ -3,9 +3,12 @@ package block import ( "bytes" "context" + "os" "testing" "time" + ds "github.com/ipfs/go-datastore" + cmtypes "github.com/cometbft/cometbft/types" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -138,13 +141,20 @@ func TestSubmitBlocksToMockDA(t *testing.T) { m.conf.DAMempoolTTL = 1 m.dalc.GasPrice = 1.0 m.dalc.GasMultiplier = 1.2 + kvStore, err := store.NewDefaultInMemoryKVStore() + require.NoError(t, err) + m.store = store.New(kvStore) t.Run("handle_tx_already_in_mempool", func(t *testing.T) { var blobs [][]byte block := types.GetRandomBlock(1, 5) blob, err := block.MarshalBinary() + require.NoError(t, err) + err = m.store.SaveBlock(ctx, block, &types.Commit{}) require.NoError(t, err) + m.store.SetHeight(ctx, 1) + blobs = append(blobs, blob) // Set up the mock to // * throw timeout waiting for tx to be included exactly once @@ -231,18 +241,34 @@ func TestSubmitBlocksToDA(t *testing.T) { } for _, tc := range testCases { - kvStore, err := store.NewDefaultInMemoryKVStore() - require.NoError(err) - m.pendingBlocks = NewPendingBlocks(store.New(kvStore)) + // there is a limitation of value size for underlying in-memory KV store, so (temporary) on-disk store is needed + kvStore := getTempKVStore(t) + m.store = store.New(kvStore) + m.pendingBlocks = NewPendingBlocks(m.store) t.Run(tc.name, func(t *testing.T) { + // PendingBlocks depend on store, so blocks needs to be saved and height updated for _, block := range tc.blocks { + require.NoError(m.store.SaveBlock(ctx, block, &types.Commit{})) m.pendingBlocks.addPendingBlock(block) } + m.store.SetHeight(ctx, uint64(len(tc.blocks))) + err := m.submitBlocksToDA(ctx) - assert.Equal(t, tc.isErrExpected, err != nil) + assert.Equal(tc.isErrExpected, err != nil) blocks, err := m.pendingBlocks.getPendingBlocks() assert.NoError(err) assert.Equal(tc.expectedPendingBlocksLength, len(blocks)) }) } } + +func getTempKVStore(t *testing.T) ds.TxnDatastore { + dbPath, err := os.MkdirTemp("", t.Name()) + require.NoError(t, err) + t.Cleanup(func() { + _ = os.RemoveAll(dbPath) + }) + kvStore, err := store.NewDefaultKVStore(os.TempDir(), dbPath, t.Name()) + require.NoError(t, err) + return kvStore +} diff --git a/block/pending_blocks.go b/block/pending_blocks.go index c8cfdac3b2..fb57f95581 100644 --- a/block/pending_blocks.go +++ b/block/pending_blocks.go @@ -2,13 +2,21 @@ package block import ( "context" + "errors" + "fmt" + "strconv" "sync/atomic" + ds "github.com/ipfs/go-datastore" + "github.com/rollkit/rollkit/store" "github.com/rollkit/rollkit/types" ) +// lshKey is the key used for persisting the last submitted height in store. +const lshKey = "last submitted" + // PendingBlocks maintains blocks that need to be published to DA layer // // Important assertions: @@ -32,20 +40,30 @@ type PendingBlocks struct { func NewPendingBlocks(store store.Store) *PendingBlocks { return &PendingBlocks{ store: store, - // TODO(tzdybal): lastSubmittedHeight from store } } // getPendingBlocks returns a sorted slice of pending blocks // that need to be published to DA layer in order of block height func (pb *PendingBlocks) getPendingBlocks() ([]*types.Block, error) { - height := pb.store.Height() lastSubmitted := pb.lastSubmittedHeight.Load() + if lastSubmitted == 0 { + // TODO(tzdybal): think about passing context here + err := pb.loadFromStore(context.TODO()) + if err != nil { + return nil, err + } + lastSubmitted = pb.lastSubmittedHeight.Load() + } + height := pb.store.Height() - // TODO(tzdybal) - lastSubmitted should never be > than height in final implementation - if lastSubmitted >= height { + if lastSubmitted == height { return nil, nil } + if lastSubmitted > height { + panic(fmt.Sprintf("height of last block submitted to DA (%d) is greater than height of last block (%d)", + lastSubmitted, height)) + } blocks := make([]*types.Block, 0, height-lastSubmitted) for i := lastSubmitted + 1; i <= height; i++ { @@ -79,3 +97,21 @@ func (pb *PendingBlocks) removeSubmittedBlocks(blocks []*types.Block) { pb.lastSubmittedHeight.CompareAndSwap(lastSubmitted, height) } } + +func (pb *PendingBlocks) loadFromStore(ctx context.Context) error { + raw, err := pb.store.GetMetadata(context.TODO(), lshKey) + if errors.Is(err, ds.ErrNotFound) { + // lshKey was never used, it's special case not actual error + // we don't need to modify lastSubmittedHeight + return nil + } + if err != nil { + return err + } + lsh, err := strconv.ParseUint(string(raw), 10, 64) + if err != nil { + return err + } + pb.lastSubmittedHeight.CompareAndSwap(0, lsh) + return nil +} diff --git a/block/pending_blocks_test.go b/block/pending_blocks_test.go index 583b469b22..7cb5948f3f 100644 --- a/block/pending_blocks_test.go +++ b/block/pending_blocks_test.go @@ -1,6 +1,7 @@ package block import ( + "context" "sort" "testing" @@ -36,9 +37,13 @@ func TestRemoveSubmittedBlocks(t *testing.T) { func TestRemoveSubsetOfBlocks(t *testing.T) { require := require.New(t) + ctx := context.Background() pb := newPendingBlocks(t) - for i := uint64(0); i < 5; i++ { - pb.addPendingBlock(types.GetRandomBlock(i, 0)) + for i := uint64(1); i <= 5; i++ { + block := types.GetRandomBlock(i, 0) + pb.addPendingBlock(block) + require.NoError(pb.store.SaveBlock(ctx, block, &types.Commit{})) + pb.store.SetHeight(ctx, i) } // Remove blocks with height 1 and 2 pb.removeSubmittedBlocks([]*types.Block{ @@ -49,7 +54,7 @@ func TestRemoveSubsetOfBlocks(t *testing.T) { require.NoError(err) require.Len(remainingBlocks, 3, "There should be 3 blocks remaining") for _, block := range remainingBlocks { - require.Contains([]uint64{0, 3, 4}, block.Height(), "Only blocks with height 0, 3, and 4 should remain") + require.Contains([]uint64{3, 4, 5}, block.Height(), "Only blocks with height 3, 4 and 5 should remain") } } From bded6219dd98968f6c5fa34bcf8df4f440ee5d9a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Zdyba=C5=82?= Date: Thu, 29 Feb 2024 09:42:27 +0100 Subject: [PATCH 05/20] refactor: remove addPendingBlock function and update tests As PendingBlocks depend on store, it is safe to remove addPendingBlock function - all tests were updated to save blocks to store instead of calling addPendingBlock. --- block/manager.go | 3 --- block/manager_test.go | 4 +--- block/pending_blocks.go | 4 ---- block/pending_blocks_test.go | 19 ++++++++++++------- 4 files changed, 13 insertions(+), 17 deletions(-) diff --git a/block/manager.go b/block/manager.go index b50c5dbde7..4dee66bb34 100644 --- a/block/manager.go +++ b/block/manager.go @@ -784,9 +784,6 @@ func (m *Manager) publishBlock(ctx context.Context) error { return err } - // Submit block to be published to the DA layer - m.pendingBlocks.addPendingBlock(block) - // Commit the new state and block which writes to disk on the proxy app appHash, _, err := m.executor.Commit(ctx, newState, block, responses) if err != nil { diff --git a/block/manager_test.go b/block/manager_test.go index ae984e738e..276749763b 100644 --- a/block/manager_test.go +++ b/block/manager_test.go @@ -172,8 +172,7 @@ func TestSubmitBlocksToMockDA(t *testing.T) { On("Submit", blobs, 1.0*1.2*1.2, []byte(nil)). Return([][]byte{bytes.Repeat([]byte{0x00}, 8)}, nil) - m.pendingBlocks = NewPendingBlocks(m.store) - m.pendingBlocks.addPendingBlock(block) + m.pendingBlocks = NewPendingBlocks(m.store, m.logger) err = m.submitBlocksToDA(ctx) require.NoError(t, err) mockDA.AssertExpectations(t) @@ -249,7 +248,6 @@ func TestSubmitBlocksToDA(t *testing.T) { // PendingBlocks depend on store, so blocks needs to be saved and height updated for _, block := range tc.blocks { require.NoError(m.store.SaveBlock(ctx, block, &types.Commit{})) - m.pendingBlocks.addPendingBlock(block) } m.store.SetHeight(ctx, uint64(len(tc.blocks))) diff --git a/block/pending_blocks.go b/block/pending_blocks.go index fb57f95581..e70e6fd8bf 100644 --- a/block/pending_blocks.go +++ b/block/pending_blocks.go @@ -81,10 +81,6 @@ func (pb *PendingBlocks) isEmpty() bool { return pb.store.Height() == pb.lastSubmittedHeight.Load() } -func (pb *PendingBlocks) addPendingBlock(_ *types.Block) { - // TODO(tzdybal): remove this method -} - // TODO(tzdybal): change signature (accept height) func (pb *PendingBlocks) removeSubmittedBlocks(blocks []*types.Block) { if len(blocks) == 0 { diff --git a/block/pending_blocks_test.go b/block/pending_blocks_test.go index 7cb5948f3f..218514466f 100644 --- a/block/pending_blocks_test.go +++ b/block/pending_blocks_test.go @@ -14,9 +14,11 @@ import ( func TestGetPendingBlocks(t *testing.T) { require := require.New(t) + ctx := context.Background() pb := newPendingBlocks(t) for i := uint64(0); i < 5; i++ { - pb.addPendingBlock(types.GetRandomBlock(i, 0)) + require.NoError(pb.store.SaveBlock(ctx, types.GetRandomBlock(i, 0), &types.Commit{})) + pb.store.SetHeight(ctx, i) } blocks, _ := pb.getPendingBlocks() require.True(sort.SliceIsSorted(blocks, func(i, j int) bool { @@ -26,11 +28,14 @@ func TestGetPendingBlocks(t *testing.T) { func TestRemoveSubmittedBlocks(t *testing.T) { require := require.New(t) + ctx := context.Background() pb := newPendingBlocks(t) for i := uint64(0); i < 5; i++ { - pb.addPendingBlock(types.GetRandomBlock(i, 0)) + require.NoError(pb.store.SaveBlock(ctx, types.GetRandomBlock(i, 0), &types.Commit{})) + pb.store.SetHeight(ctx, i) } - blocks, _ := pb.getPendingBlocks() + blocks, err := pb.getPendingBlocks() + require.NoError(err) pb.removeSubmittedBlocks(blocks) require.True(pb.isEmpty()) } @@ -40,9 +45,7 @@ func TestRemoveSubsetOfBlocks(t *testing.T) { ctx := context.Background() pb := newPendingBlocks(t) for i := uint64(1); i <= 5; i++ { - block := types.GetRandomBlock(i, 0) - pb.addPendingBlock(block) - require.NoError(pb.store.SaveBlock(ctx, block, &types.Commit{})) + require.NoError(pb.store.SaveBlock(ctx, types.GetRandomBlock(i, 0), &types.Commit{})) pb.store.SetHeight(ctx, i) } // Remove blocks with height 1 and 2 @@ -60,9 +63,11 @@ func TestRemoveSubsetOfBlocks(t *testing.T) { func TestRemoveAllBlocksAndVerifyEmpty(t *testing.T) { require := require.New(t) + ctx := context.Background() pb := newPendingBlocks(t) for i := uint64(0); i < 5; i++ { - pb.addPendingBlock(types.GetRandomBlock(i, 0)) + require.NoError(pb.store.SaveBlock(ctx, types.GetRandomBlock(i, 0), &types.Commit{})) + pb.store.SetHeight(ctx, i) } // Remove all blocks blocks, err := pb.getPendingBlocks() From 632935b36c7077acc3008479c6bbfeddc0aca4d1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Zdyba=C5=82?= Date: Thu, 29 Feb 2024 12:57:14 +0100 Subject: [PATCH 06/20] feat: save lastSubmittedHeight into store --- block/manager_test.go | 8 ++++++++ block/pending_blocks.go | 10 ++++++---- 2 files changed, 14 insertions(+), 4 deletions(-) diff --git a/block/manager_test.go b/block/manager_test.go index 276749763b..0970e18025 100644 --- a/block/manager_test.go +++ b/block/manager_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "os" + "strconv" "testing" "time" @@ -256,6 +257,13 @@ func TestSubmitBlocksToDA(t *testing.T) { blocks, err := m.pendingBlocks.getPendingBlocks() assert.NoError(err) assert.Equal(tc.expectedPendingBlocksLength, len(blocks)) + + // ensure that metadata is updated in KV store + raw, err := m.store.GetMetadata(ctx, lshKey) + require.NoError(err) + lshInKV, err := strconv.ParseUint(string(raw), 10, 64) + require.NoError(err) + assert.Equal(m.store.Height(), lshInKV+uint64(tc.expectedPendingBlocksLength)) }) } } diff --git a/block/pending_blocks.go b/block/pending_blocks.go index e70e6fd8bf..8fe27a99ee 100644 --- a/block/pending_blocks.go +++ b/block/pending_blocks.go @@ -86,11 +86,13 @@ func (pb *PendingBlocks) removeSubmittedBlocks(blocks []*types.Block) { if len(blocks) == 0 { return } - height := blocks[len(blocks)-1].Height() - lastSubmitted := pb.lastSubmittedHeight.Load() + latestBlockHeight := blocks[len(blocks)-1].Height() + lsh := pb.lastSubmittedHeight.Load() - if height > lastSubmitted { - pb.lastSubmittedHeight.CompareAndSwap(lastSubmitted, height) + if latestBlockHeight > lsh { + if pb.lastSubmittedHeight.CompareAndSwap(lsh, latestBlockHeight) { + pb.store.SetMetadata(context.TODO(), lshKey, []byte(strconv.FormatUint(latestBlockHeight, 10))) + } } } From e8eb37bc98c1d11c8ca33d7ecb87bf763155038e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Zdyba=C5=82?= Date: Thu, 29 Feb 2024 14:01:17 +0100 Subject: [PATCH 07/20] refactor: rename const and improve tests --- block/manager_test.go | 2 +- block/pending_blocks.go | 12 ++++++------ node/full_node_test.go | 8 ++++++++ 3 files changed, 15 insertions(+), 7 deletions(-) diff --git a/block/manager_test.go b/block/manager_test.go index 0970e18025..b1147e78f9 100644 --- a/block/manager_test.go +++ b/block/manager_test.go @@ -259,7 +259,7 @@ func TestSubmitBlocksToDA(t *testing.T) { assert.Equal(tc.expectedPendingBlocksLength, len(blocks)) // ensure that metadata is updated in KV store - raw, err := m.store.GetMetadata(ctx, lshKey) + raw, err := m.store.GetMetadata(ctx, LastSubmittedHeightKey) require.NoError(err) lshInKV, err := strconv.ParseUint(string(raw), 10, 64) require.NoError(err) diff --git a/block/pending_blocks.go b/block/pending_blocks.go index 8fe27a99ee..08784b3336 100644 --- a/block/pending_blocks.go +++ b/block/pending_blocks.go @@ -14,8 +14,8 @@ import ( "github.com/rollkit/rollkit/types" ) -// lshKey is the key used for persisting the last submitted height in store. -const lshKey = "last submitted" +// LastSubmittedHeightKey is the key used for persisting the last submitted height in store. +const LastSubmittedHeightKey = "last submitted" // PendingBlocks maintains blocks that need to be published to DA layer // @@ -81,7 +81,7 @@ func (pb *PendingBlocks) isEmpty() bool { return pb.store.Height() == pb.lastSubmittedHeight.Load() } -// TODO(tzdybal): change signature (accept height) +// TODO(tzdybal): change signature (accept height, maybe context?) func (pb *PendingBlocks) removeSubmittedBlocks(blocks []*types.Block) { if len(blocks) == 0 { return @@ -91,15 +91,15 @@ func (pb *PendingBlocks) removeSubmittedBlocks(blocks []*types.Block) { if latestBlockHeight > lsh { if pb.lastSubmittedHeight.CompareAndSwap(lsh, latestBlockHeight) { - pb.store.SetMetadata(context.TODO(), lshKey, []byte(strconv.FormatUint(latestBlockHeight, 10))) + pb.store.SetMetadata(context.TODO(), LastSubmittedHeightKey, []byte(strconv.FormatUint(latestBlockHeight, 10))) } } } func (pb *PendingBlocks) loadFromStore(ctx context.Context) error { - raw, err := pb.store.GetMetadata(context.TODO(), lshKey) + raw, err := pb.store.GetMetadata(ctx, LastSubmittedHeightKey) if errors.Is(err, ds.ErrNotFound) { - // lshKey was never used, it's special case not actual error + // LastSubmittedHeightKey was never used, it's special case not actual error // we don't need to modify lastSubmittedHeight return nil } diff --git a/node/full_node_test.go b/node/full_node_test.go index 48b1eceb4f..684ef005d7 100644 --- a/node/full_node_test.go +++ b/node/full_node_test.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "os" + "strconv" "testing" "time" @@ -219,6 +220,13 @@ func TestPendingBlocks(t *testing.T) { err = waitForAtLeastNBlocks(node, firstRunBlocks+5, Store) assert.NoError(t, err) + // assert that LastSubmittedHeight was updated in store + raw, err := node.(*FullNode).Store.GetMetadata(context.Background(), block.LastSubmittedHeightKey) + require.NoError(t, err) + lsh, err := strconv.ParseUint(string(raw), 10, 64) + require.NoError(t, err) + assert.Greater(t, lsh, uint64(firstRunBlocks)) + err = node.Stop() assert.NoError(t, err) From d567f778cae0a2e6d34bdc8ace8986b731599676 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Zdyba=C5=82?= Date: Thu, 29 Feb 2024 14:11:37 +0100 Subject: [PATCH 08/20] refactor: updated getPendingBlocks to accept context --- block/manager.go | 2 +- block/manager_test.go | 2 +- block/pending_blocks.go | 8 ++++---- block/pending_blocks_test.go | 8 ++++---- 4 files changed, 10 insertions(+), 10 deletions(-) diff --git a/block/manager.go b/block/manager.go index 4dee66bb34..cc13555d2d 100644 --- a/block/manager.go +++ b/block/manager.go @@ -837,7 +837,7 @@ func (m *Manager) recordMetrics(block *types.Block) { func (m *Manager) submitBlocksToDA(ctx context.Context) error { submittedAllBlocks := false backoff := initialBackoff - blocksToSubmit, err := m.pendingBlocks.getPendingBlocks() + blocksToSubmit, err := m.pendingBlocks.getPendingBlocks(ctx) if len(blocksToSubmit) == 0 { return err } diff --git a/block/manager_test.go b/block/manager_test.go index b1147e78f9..b09fd1279c 100644 --- a/block/manager_test.go +++ b/block/manager_test.go @@ -254,7 +254,7 @@ func TestSubmitBlocksToDA(t *testing.T) { err := m.submitBlocksToDA(ctx) assert.Equal(tc.isErrExpected, err != nil) - blocks, err := m.pendingBlocks.getPendingBlocks() + blocks, err := m.pendingBlocks.getPendingBlocks(ctx) assert.NoError(err) assert.Equal(tc.expectedPendingBlocksLength, len(blocks)) diff --git a/block/pending_blocks.go b/block/pending_blocks.go index 08784b3336..12f9f70e98 100644 --- a/block/pending_blocks.go +++ b/block/pending_blocks.go @@ -45,11 +45,10 @@ func NewPendingBlocks(store store.Store) *PendingBlocks { // getPendingBlocks returns a sorted slice of pending blocks // that need to be published to DA layer in order of block height -func (pb *PendingBlocks) getPendingBlocks() ([]*types.Block, error) { +func (pb *PendingBlocks) getPendingBlocks(ctx context.Context) ([]*types.Block, error) { lastSubmitted := pb.lastSubmittedHeight.Load() if lastSubmitted == 0 { - // TODO(tzdybal): think about passing context here - err := pb.loadFromStore(context.TODO()) + err := pb.loadFromStore(ctx) if err != nil { return nil, err } @@ -67,7 +66,7 @@ func (pb *PendingBlocks) getPendingBlocks() ([]*types.Block, error) { blocks := make([]*types.Block, 0, height-lastSubmitted) for i := lastSubmitted + 1; i <= height; i++ { - block, err := pb.store.GetBlock(context.TODO(), i) + block, err := pb.store.GetBlock(ctx, i) if err != nil { // return as much as possible + error information return blocks, err @@ -91,6 +90,7 @@ func (pb *PendingBlocks) removeSubmittedBlocks(blocks []*types.Block) { if latestBlockHeight > lsh { if pb.lastSubmittedHeight.CompareAndSwap(lsh, latestBlockHeight) { + // TODO(tzdybal): handle error vs just ignore it - even if there is an issue, there is not much we can do about it pb.store.SetMetadata(context.TODO(), LastSubmittedHeightKey, []byte(strconv.FormatUint(latestBlockHeight, 10))) } } diff --git a/block/pending_blocks_test.go b/block/pending_blocks_test.go index 218514466f..278f872712 100644 --- a/block/pending_blocks_test.go +++ b/block/pending_blocks_test.go @@ -20,7 +20,7 @@ func TestGetPendingBlocks(t *testing.T) { require.NoError(pb.store.SaveBlock(ctx, types.GetRandomBlock(i, 0), &types.Commit{})) pb.store.SetHeight(ctx, i) } - blocks, _ := pb.getPendingBlocks() + blocks, _ := pb.getPendingBlocks(ctx) require.True(sort.SliceIsSorted(blocks, func(i, j int) bool { return blocks[i].Height() < blocks[j].Height() })) @@ -34,7 +34,7 @@ func TestRemoveSubmittedBlocks(t *testing.T) { require.NoError(pb.store.SaveBlock(ctx, types.GetRandomBlock(i, 0), &types.Commit{})) pb.store.SetHeight(ctx, i) } - blocks, err := pb.getPendingBlocks() + blocks, err := pb.getPendingBlocks(ctx) require.NoError(err) pb.removeSubmittedBlocks(blocks) require.True(pb.isEmpty()) @@ -53,7 +53,7 @@ func TestRemoveSubsetOfBlocks(t *testing.T) { types.GetRandomBlock(1, 0), types.GetRandomBlock(2, 0), }) - remainingBlocks, err := pb.getPendingBlocks() + remainingBlocks, err := pb.getPendingBlocks(ctx) require.NoError(err) require.Len(remainingBlocks, 3, "There should be 3 blocks remaining") for _, block := range remainingBlocks { @@ -70,7 +70,7 @@ func TestRemoveAllBlocksAndVerifyEmpty(t *testing.T) { pb.store.SetHeight(ctx, i) } // Remove all blocks - blocks, err := pb.getPendingBlocks() + blocks, err := pb.getPendingBlocks(ctx) require.NoError(err) pb.removeSubmittedBlocks(blocks) require.True(pb.isEmpty(), "PendingBlocks should be empty after removing all blocks") From 222d89cf7dd21db9936b0bdafb35cd4315b6c67d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Zdyba=C5=82?= Date: Thu, 29 Feb 2024 14:17:52 +0100 Subject: [PATCH 09/20] refactor: add logger to PendingBlocks --- block/manager.go | 2 +- block/manager_test.go | 2 +- block/pending_blocks.go | 9 ++++++--- block/pending_blocks_test.go | 3 ++- 4 files changed, 10 insertions(+), 6 deletions(-) diff --git a/block/manager.go b/block/manager.go index cc13555d2d..c1866a054b 100644 --- a/block/manager.go +++ b/block/manager.go @@ -219,7 +219,7 @@ func NewManager( validatorSet: &valSet, txsAvailable: txsAvailableCh, buildingBlock: false, - pendingBlocks: NewPendingBlocks(store), + pendingBlocks: NewPendingBlocks(store, logger), metrics: seqMetrics, } return agg, nil diff --git a/block/manager_test.go b/block/manager_test.go index b09fd1279c..4c80f289fd 100644 --- a/block/manager_test.go +++ b/block/manager_test.go @@ -244,7 +244,7 @@ func TestSubmitBlocksToDA(t *testing.T) { // there is a limitation of value size for underlying in-memory KV store, so (temporary) on-disk store is needed kvStore := getTempKVStore(t) m.store = store.New(kvStore) - m.pendingBlocks = NewPendingBlocks(m.store) + m.pendingBlocks = NewPendingBlocks(m.store, m.logger) t.Run(tc.name, func(t *testing.T) { // PendingBlocks depend on store, so blocks needs to be saved and height updated for _, block := range tc.blocks { diff --git a/block/pending_blocks.go b/block/pending_blocks.go index 12f9f70e98..4c0efca932 100644 --- a/block/pending_blocks.go +++ b/block/pending_blocks.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "github.com/rollkit/rollkit/third_party/log" "strconv" "sync/atomic" @@ -30,16 +31,18 @@ const LastSubmittedHeightKey = "last submitted" // rollkit is able to skip duplicate blocks so this shouldn't affect full nodes. // TODO(tzdybal): batch size type PendingBlocks struct { - store store.Store + store store.Store + logger log.Logger // lastSubmittedHeight holds information about last block successfully submitted to DA lastSubmittedHeight atomic.Uint64 } // NewPendingBlocks returns a new PendingBlocks struct -func NewPendingBlocks(store store.Store) *PendingBlocks { +func NewPendingBlocks(store store.Store, logger log.Logger) *PendingBlocks { return &PendingBlocks{ - store: store, + store: store, + logger: logger, } } diff --git a/block/pending_blocks_test.go b/block/pending_blocks_test.go index 278f872712..f4519fd731 100644 --- a/block/pending_blocks_test.go +++ b/block/pending_blocks_test.go @@ -2,6 +2,7 @@ package block import ( "context" + test "github.com/rollkit/rollkit/test/log" "sort" "testing" @@ -91,5 +92,5 @@ func TestRemoveBlocksFromEmptyPendingBlocks(t *testing.T) { func newPendingBlocks(t *testing.T) *PendingBlocks { kv, err := store.NewDefaultInMemoryKVStore() require.NoError(t, err) - return NewPendingBlocks(store.New(kv)) + return NewPendingBlocks(store.New(kv), test.NewLogger(t)) } From 1d9348119a629627f499055083259ecd38cb0ae5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Zdyba=C5=82?= Date: Thu, 29 Feb 2024 14:27:09 +0100 Subject: [PATCH 10/20] fix: add error logging when it's not possible to store latest submitted height --- block/pending_blocks.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/block/pending_blocks.go b/block/pending_blocks.go index 4c0efca932..e4712b5fb8 100644 --- a/block/pending_blocks.go +++ b/block/pending_blocks.go @@ -93,8 +93,8 @@ func (pb *PendingBlocks) removeSubmittedBlocks(blocks []*types.Block) { if latestBlockHeight > lsh { if pb.lastSubmittedHeight.CompareAndSwap(lsh, latestBlockHeight) { - // TODO(tzdybal): handle error vs just ignore it - even if there is an issue, there is not much we can do about it - pb.store.SetMetadata(context.TODO(), LastSubmittedHeightKey, []byte(strconv.FormatUint(latestBlockHeight, 10))) + err := pb.store.SetMetadata(context.TODO(), LastSubmittedHeightKey, []byte(strconv.FormatUint(latestBlockHeight, 10))) + pb.logger.Error("failed to store height of latest block submitted to DA", "err", err) } } } From 8530658ade773088d8d70d8b7b99ec2f906254d1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Zdyba=C5=82?= Date: Thu, 29 Feb 2024 19:26:37 +0100 Subject: [PATCH 11/20] refactor: change signature of removeSubmittedBlocks setLastSubmittedHeight now accepts only height, and the name better reflects it's --- block/manager.go | 6 +++++- block/pending_blocks.go | 15 ++++++--------- block/pending_blocks_test.go | 22 ++++++++-------------- 3 files changed, 19 insertions(+), 24 deletions(-) diff --git a/block/manager.go b/block/manager.go index c1866a054b..3f5f5575e6 100644 --- a/block/manager.go +++ b/block/manager.go @@ -866,7 +866,11 @@ func (m *Manager) submitBlocksToDA(ctx context.Context) error { for _, block := range submittedBlocks { m.blockCache.setDAIncluded(block.Hash().String()) } - m.pendingBlocks.removeSubmittedBlocks(submittedBlocks) + lsh := uint64(0) + if l := len(submittedBlocks); l > 0 { + lsh = submittedBlocks[l-1].Height() + } + m.pendingBlocks.setLastSubmittedHeight(lsh) blocksToSubmit = notSubmittedBlocks // reset submission options when successful // scale back gasPrice gradually diff --git a/block/pending_blocks.go b/block/pending_blocks.go index e4712b5fb8..324e3a0e11 100644 --- a/block/pending_blocks.go +++ b/block/pending_blocks.go @@ -4,10 +4,11 @@ import ( "context" "errors" "fmt" - "github.com/rollkit/rollkit/third_party/log" "strconv" "sync/atomic" + "github.com/rollkit/rollkit/third_party/log" + ds "github.com/ipfs/go-datastore" "github.com/rollkit/rollkit/store" @@ -84,16 +85,12 @@ func (pb *PendingBlocks) isEmpty() bool { } // TODO(tzdybal): change signature (accept height, maybe context?) -func (pb *PendingBlocks) removeSubmittedBlocks(blocks []*types.Block) { - if len(blocks) == 0 { - return - } - latestBlockHeight := blocks[len(blocks)-1].Height() +func (pb *PendingBlocks) setLastSubmittedHeight(newLastSubmittedHeight uint64) { lsh := pb.lastSubmittedHeight.Load() - if latestBlockHeight > lsh { - if pb.lastSubmittedHeight.CompareAndSwap(lsh, latestBlockHeight) { - err := pb.store.SetMetadata(context.TODO(), LastSubmittedHeightKey, []byte(strconv.FormatUint(latestBlockHeight, 10))) + if newLastSubmittedHeight > lsh && pb.lastSubmittedHeight.CompareAndSwap(lsh, newLastSubmittedHeight) { + err := pb.store.SetMetadata(context.TODO(), LastSubmittedHeightKey, []byte(strconv.FormatUint(newLastSubmittedHeight, 10))) + if err != nil { pb.logger.Error("failed to store height of latest block submitted to DA", "err", err) } } diff --git a/block/pending_blocks_test.go b/block/pending_blocks_test.go index f4519fd731..9951426711 100644 --- a/block/pending_blocks_test.go +++ b/block/pending_blocks_test.go @@ -2,10 +2,11 @@ package block import ( "context" - test "github.com/rollkit/rollkit/test/log" "sort" "testing" + test "github.com/rollkit/rollkit/test/log" + "github.com/rollkit/rollkit/store" "github.com/stretchr/testify/require" @@ -31,13 +32,12 @@ func TestRemoveSubmittedBlocks(t *testing.T) { require := require.New(t) ctx := context.Background() pb := newPendingBlocks(t) - for i := uint64(0); i < 5; i++ { + const nBlocks = 5 + for i := uint64(1); i <= nBlocks; i++ { require.NoError(pb.store.SaveBlock(ctx, types.GetRandomBlock(i, 0), &types.Commit{})) pb.store.SetHeight(ctx, i) } - blocks, err := pb.getPendingBlocks(ctx) - require.NoError(err) - pb.removeSubmittedBlocks(blocks) + pb.setLastSubmittedHeight(nBlocks) require.True(pb.isEmpty()) } @@ -50,10 +50,7 @@ func TestRemoveSubsetOfBlocks(t *testing.T) { pb.store.SetHeight(ctx, i) } // Remove blocks with height 1 and 2 - pb.removeSubmittedBlocks([]*types.Block{ - types.GetRandomBlock(1, 0), - types.GetRandomBlock(2, 0), - }) + pb.setLastSubmittedHeight(2) remainingBlocks, err := pb.getPendingBlocks(ctx) require.NoError(err) require.Len(remainingBlocks, 3, "There should be 3 blocks remaining") @@ -73,7 +70,7 @@ func TestRemoveAllBlocksAndVerifyEmpty(t *testing.T) { // Remove all blocks blocks, err := pb.getPendingBlocks(ctx) require.NoError(err) - pb.removeSubmittedBlocks(blocks) + pb.setLastSubmittedHeight(blocks[len(blocks)-1].Height()) require.True(pb.isEmpty(), "PendingBlocks should be empty after removing all blocks") } @@ -82,10 +79,7 @@ func TestRemoveBlocksFromEmptyPendingBlocks(t *testing.T) { pb := newPendingBlocks(t) // Attempt to remove blocks from an empty PendingBlocks require.NotPanics(func() { - pb.removeSubmittedBlocks([]*types.Block{ - types.GetRandomBlock(1, 0), - types.GetRandomBlock(2, 0), - }) + pb.setLastSubmittedHeight(2) }, "Removing blocks from an empty PendingBlocks should not cause a panic") } From e43432fe6c6753c38580f64fdcf1606b13b94bf5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Zdyba=C5=82?= Date: Thu, 29 Feb 2024 20:27:34 +0100 Subject: [PATCH 12/20] chore: minor improvements --- block/pending_blocks.go | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/block/pending_blocks.go b/block/pending_blocks.go index 324e3a0e11..ae281194f7 100644 --- a/block/pending_blocks.go +++ b/block/pending_blocks.go @@ -7,12 +7,10 @@ import ( "strconv" "sync/atomic" - "github.com/rollkit/rollkit/third_party/log" - ds "github.com/ipfs/go-datastore" "github.com/rollkit/rollkit/store" - + "github.com/rollkit/rollkit/third_party/log" "github.com/rollkit/rollkit/types" ) @@ -30,7 +28,7 @@ const LastSubmittedHeightKey = "last submitted" // Worst case scenario is when blocks was successfully submitted to DA, but confirmation was not received (e.g. node was // restarted, networking issue occurred). In this case blocks are re-submitted to DA (it's extra cost). // rollkit is able to skip duplicate blocks so this shouldn't affect full nodes. -// TODO(tzdybal): batch size +// TODO(tzdybal): we shouldn't try to push all pending blocks at once; this should depend on max blob size type PendingBlocks struct { store store.Store logger log.Logger @@ -84,7 +82,6 @@ func (pb *PendingBlocks) isEmpty() bool { return pb.store.Height() == pb.lastSubmittedHeight.Load() } -// TODO(tzdybal): change signature (accept height, maybe context?) func (pb *PendingBlocks) setLastSubmittedHeight(newLastSubmittedHeight uint64) { lsh := pb.lastSubmittedHeight.Load() From 90bebada12836853899f933d218713d12db53320 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Zdyba=C5=82?= Date: Thu, 29 Feb 2024 21:09:39 +0100 Subject: [PATCH 13/20] refactor: improve TestPendingBlocks The "TestPendingBlocks" test function was modified to provide more specific testing for block submissions. The test now verifies that blocks from the first and second runs are submitted in order and that this information is persisted in the store. Additionally, the test scenario has been detailed for clarity. --- node/full_node_test.go | 37 ++++++++++++++++++++++++++++--------- 1 file changed, 28 insertions(+), 9 deletions(-) diff --git a/node/full_node_test.go b/node/full_node_test.go index 684ef005d7..41f13e0b56 100644 --- a/node/full_node_test.go +++ b/node/full_node_test.go @@ -161,6 +161,16 @@ func TestInvalidBlocksIgnored(t *testing.T) { } // TestPendingBlocks is a test for bug described in https://github.com/rollkit/rollkit/issues/1548 +// +// Test scenario: +// - mock DA to refuse all submissions (returning error) +// - run aggregator to produce some blocks +// - stop aggregator node +// - all blocks should be considered as pending DA submission (because of mock DA behaviour) +// - change mock to accept all submissions +// - start aggregator node again (using the same store, to simulate restart) +// - verify that blocks from first run was submitted to DA +// - additionally - ensure that information was persisted in store (TODO: this should be tested separately) func TestPendingBlocks(t *testing.T) { ctx := context.Background() @@ -183,7 +193,10 @@ func TestPendingBlocks(t *testing.T) { err = node.Start() assert.NoError(t, err) - const firstRunBlocks = 10 + const ( + firstRunBlocks = 10 + secondRunBlocks = 5 + ) err = waitForAtLeastNBlocks(node, firstRunBlocks, Store) assert.NoError(t, err) @@ -197,19 +210,18 @@ func TestPendingBlocks(t *testing.T) { // reset DA mock to ensure that Submit was called mockDA.On("Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Unset() - // mock submit function to just return some hash and count the number of submitted blobs + // mock submit function to just return some hash and collect all blobs in order // node will be stopped after producing at least firstRunBlocks blocks - // restarted node should get blocks from first and second run (more than firstRunBlocks - // TODO(tzdybal): this seems fragile and probably should be improved (to deserialize blocks and check heights) - uniqueBlobs := make(map[string]uint64) + // restarted node should submit to DA blocks from first and second run (more than firstRunBlocks) + allBlobs := make([][]byte, 0, firstRunBlocks+secondRunBlocks) 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[:] - uniqueBlobs[string(sha[:])]++ } + allBlobs = append(allBlobs, blobs...) return hashes, nil }) @@ -217,7 +229,7 @@ func TestPendingBlocks(t *testing.T) { assert.NoError(t, err) // let node produce few more blocks - err = waitForAtLeastNBlocks(node, firstRunBlocks+5, Store) + err = waitForAtLeastNBlocks(node, firstRunBlocks+secondRunBlocks, Store) assert.NoError(t, err) // assert that LastSubmittedHeight was updated in store @@ -229,9 +241,16 @@ func TestPendingBlocks(t *testing.T) { err = node.Stop() assert.NoError(t, err) - - assert.Greater(t, len(uniqueBlobs), firstRunBlocks) mock.AssertExpectationsForObjects(t, mockDA) + + // ensure that all blocks were submitted in order + for i := 0; i < len(allBlobs); i++ { + b := &types.Block{} + err := b.UnmarshalBinary(allBlobs[i]) + require.NoError(t, err) + require.Equal(t, uint64(i+1), b.Height()) // '+1' because blocks start at genesis with height 1 + } + } func createAggregatorWithPersistence(ctx context.Context, dbPath string, dalc *da.DAClient, t *testing.T) (Node, *mocks.Application) { From e79287bdd856c2191ff48a38b18c0bb1f3db6b30 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Zdyba=C5=82?= Date: Thu, 29 Feb 2024 21:52:23 +0100 Subject: [PATCH 14/20] refactor: wrap errors in SetMetadata and GetMetadata --- store/store.go | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/store/store.go b/store/store.go index 7282bc33b7..12f80bc497 100644 --- a/store/store.go +++ b/store/store.go @@ -212,12 +212,20 @@ func (s *DefaultStore) GetState(ctx context.Context) (types.State, error) { // // Metadata is separated from other data by using prefix in KV. func (s *DefaultStore) SetMetadata(ctx context.Context, key string, value []byte) error { - return s.db.Put(ctx, ds.NewKey(getMetaKey(key)), value) + err := s.db.Put(ctx, ds.NewKey(getMetaKey(key)), value) + if err != nil { + return fmt.Errorf("failed to set metadata for key '%s': %w", key, err) + } + return nil } // GetMetadata returns values stored for given key with SetMetadata. func (s *DefaultStore) GetMetadata(ctx context.Context, key string) ([]byte, error) { - return s.db.Get(ctx, ds.NewKey(getMetaKey(key))) + data, err := s.db.Get(ctx, ds.NewKey(getMetaKey(key))) + if err != nil { + return nil, fmt.Errorf("failed to get metadata for key '%s': %w", key, err) + } + return data, nil } // loadHashFromIndex returns the hash of a block given its height From 0b74b42c8a4be3507bc7524a177eab7ccc265bac Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Zdyba=C5=82?= Date: Fri, 1 Mar 2024 21:57:17 +0100 Subject: [PATCH 15/20] fix: address review comments --- block/manager.go | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/block/manager.go b/block/manager.go index 3f5f5575e6..7eaef5aebb 100644 --- a/block/manager.go +++ b/block/manager.go @@ -839,9 +839,16 @@ func (m *Manager) submitBlocksToDA(ctx context.Context) error { backoff := initialBackoff blocksToSubmit, err := m.pendingBlocks.getPendingBlocks(ctx) if len(blocksToSubmit) == 0 { + // There are no pending blocks; return because there's nothing to do, but: + // - it might be caused by error, then err != nil + // - all pending blocks are processed, then err == nil + // whatever the reason, error information is propagated correctly to the caller return err } if err != nil { + // There are some pending blocks but also an error. It's very unlikely case - probably some error while reading + // blocks from the store. + // The error is logged and normal processing of pending blocks continues. m.logger.Error("error while fetching blocks pending DA", "err", err) } numSubmittedBlocks := 0 @@ -866,11 +873,11 @@ func (m *Manager) submitBlocksToDA(ctx context.Context) error { for _, block := range submittedBlocks { m.blockCache.setDAIncluded(block.Hash().String()) } - lsh := uint64(0) + lastSubmittedHeight := uint64(0) if l := len(submittedBlocks); l > 0 { - lsh = submittedBlocks[l-1].Height() + lastSubmittedHeight = submittedBlocks[l-1].Height() } - m.pendingBlocks.setLastSubmittedHeight(lsh) + m.pendingBlocks.setLastSubmittedHeight(lastSubmittedHeight) blocksToSubmit = notSubmittedBlocks // reset submission options when successful // scale back gasPrice gradually From 8f90acce35eacc1a6eac771a638d22819c620098 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Zdyba=C5=82?= Date: Wed, 6 Mar 2024 22:00:23 +0100 Subject: [PATCH 16/20] fix: move initialization of `pendingBlocks` to constructor --- block/manager.go | 7 ++++++- block/manager_test.go | 6 ++++-- block/pending_blocks.go | 19 ++++++++----------- block/pending_blocks_test.go | 4 +++- 4 files changed, 21 insertions(+), 15 deletions(-) diff --git a/block/manager.go b/block/manager.go index 7eaef5aebb..d91816f053 100644 --- a/block/manager.go +++ b/block/manager.go @@ -197,6 +197,11 @@ func NewManager( txsAvailableCh = nil } + pendingBlocks, err := NewPendingBlocks(store, logger) + if err != nil { + return nil, err + } + agg := &Manager{ proposerKey: proposerKey, conf: conf, @@ -219,7 +224,7 @@ func NewManager( validatorSet: &valSet, txsAvailable: txsAvailableCh, buildingBlock: false, - pendingBlocks: NewPendingBlocks(store, logger), + pendingBlocks: pendingBlocks, metrics: seqMetrics, } return agg, nil diff --git a/block/manager_test.go b/block/manager_test.go index 4c80f289fd..787c5f94e0 100644 --- a/block/manager_test.go +++ b/block/manager_test.go @@ -173,7 +173,8 @@ func TestSubmitBlocksToMockDA(t *testing.T) { On("Submit", blobs, 1.0*1.2*1.2, []byte(nil)). Return([][]byte{bytes.Repeat([]byte{0x00}, 8)}, nil) - m.pendingBlocks = NewPendingBlocks(m.store, m.logger) + m.pendingBlocks, err = NewPendingBlocks(m.store, m.logger) + require.NoError(t, err) err = m.submitBlocksToDA(ctx) require.NoError(t, err) mockDA.AssertExpectations(t) @@ -244,7 +245,8 @@ func TestSubmitBlocksToDA(t *testing.T) { // there is a limitation of value size for underlying in-memory KV store, so (temporary) on-disk store is needed kvStore := getTempKVStore(t) m.store = store.New(kvStore) - m.pendingBlocks = NewPendingBlocks(m.store, m.logger) + m.pendingBlocks, err = NewPendingBlocks(m.store, m.logger) + require.NoError(err) t.Run(tc.name, func(t *testing.T) { // PendingBlocks depend on store, so blocks needs to be saved and height updated for _, block := range tc.blocks { diff --git a/block/pending_blocks.go b/block/pending_blocks.go index ae281194f7..cf06c1525d 100644 --- a/block/pending_blocks.go +++ b/block/pending_blocks.go @@ -38,24 +38,21 @@ type PendingBlocks struct { } // NewPendingBlocks returns a new PendingBlocks struct -func NewPendingBlocks(store store.Store, logger log.Logger) *PendingBlocks { - return &PendingBlocks{ +func NewPendingBlocks(store store.Store, logger log.Logger) (*PendingBlocks, error) { + pb := &PendingBlocks{ store: store, logger: logger, } + if err := pb.init(); err != nil { + return nil, err + } + return pb, nil } // getPendingBlocks returns a sorted slice of pending blocks // that need to be published to DA layer in order of block height func (pb *PendingBlocks) getPendingBlocks(ctx context.Context) ([]*types.Block, error) { lastSubmitted := pb.lastSubmittedHeight.Load() - if lastSubmitted == 0 { - err := pb.loadFromStore(ctx) - if err != nil { - return nil, err - } - lastSubmitted = pb.lastSubmittedHeight.Load() - } height := pb.store.Height() if lastSubmitted == height { @@ -93,8 +90,8 @@ func (pb *PendingBlocks) setLastSubmittedHeight(newLastSubmittedHeight uint64) { } } -func (pb *PendingBlocks) loadFromStore(ctx context.Context) error { - raw, err := pb.store.GetMetadata(ctx, LastSubmittedHeightKey) +func (pb *PendingBlocks) init() error { + raw, err := pb.store.GetMetadata(context.Background(), LastSubmittedHeightKey) if errors.Is(err, ds.ErrNotFound) { // LastSubmittedHeightKey was never used, it's special case not actual error // we don't need to modify lastSubmittedHeight diff --git a/block/pending_blocks_test.go b/block/pending_blocks_test.go index 9951426711..ee90ba93d1 100644 --- a/block/pending_blocks_test.go +++ b/block/pending_blocks_test.go @@ -86,5 +86,7 @@ func TestRemoveBlocksFromEmptyPendingBlocks(t *testing.T) { func newPendingBlocks(t *testing.T) *PendingBlocks { kv, err := store.NewDefaultInMemoryKVStore() require.NoError(t, err) - return NewPendingBlocks(store.New(kv), test.NewLogger(t)) + pendingBlocks, err := NewPendingBlocks(store.New(kv), test.NewLogger(t)) + require.NoError(t, err) + return pendingBlocks } From 2fde0b189668282e6453ed5f462c179b46d32dab Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Zdyba=C5=82?= Date: Wed, 6 Mar 2024 22:03:16 +0100 Subject: [PATCH 17/20] test: replace assert with require in metadata tests --- store/store_test.go | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/store/store_test.go b/store/store_test.go index 12d685b21e..578f7b1491 100644 --- a/store/store_test.go +++ b/store/store_test.go @@ -210,7 +210,6 @@ func TestBlockResponses(t *testing.T) { func TestMetadata(t *testing.T) { t.Parallel() - assert := assert.New(t) require := require.New(t) ctx, cancel := context.WithCancel(context.Background()) @@ -229,16 +228,16 @@ func TestMetadata(t *testing.T) { const n = 5 for i := 0; i < n; i++ { - assert.NoError(s.SetMetadata(ctx, getKey(i), getValue(i))) + require.NoError(s.SetMetadata(ctx, getKey(i), getValue(i))) } for i := 0; i < n; i++ { value, err := s.GetMetadata(ctx, getKey(i)) - assert.NoError(err) - assert.Equal(getValue(i), value) + require.NoError(err) + require.Equal(getValue(i), value) } v, err := s.GetMetadata(ctx, "unused key") - assert.Error(err) - assert.Nil(v) + require.Error(err) + require.Nil(v) } From 5c906390d36fd607418b105400701ef3c1c274e9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Zdyba=C5=82?= Date: Wed, 6 Mar 2024 22:04:34 +0100 Subject: [PATCH 18/20] test: add error handling in TestRemoveSubsetOfBlocks --- block/pending_blocks_test.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/block/pending_blocks_test.go b/block/pending_blocks_test.go index ee90ba93d1..d2e160e37e 100644 --- a/block/pending_blocks_test.go +++ b/block/pending_blocks_test.go @@ -22,7 +22,8 @@ func TestGetPendingBlocks(t *testing.T) { require.NoError(pb.store.SaveBlock(ctx, types.GetRandomBlock(i, 0), &types.Commit{})) pb.store.SetHeight(ctx, i) } - blocks, _ := pb.getPendingBlocks(ctx) + blocks, err := pb.getPendingBlocks(ctx) + require.NoError(err) require.True(sort.SliceIsSorted(blocks, func(i, j int) bool { return blocks[i].Height() < blocks[j].Height() })) From 7b52fa513a764e7dc8167002c9fc572fb1170d84 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Zdyba=C5=82?= Date: Wed, 6 Mar 2024 23:01:28 +0100 Subject: [PATCH 19/20] test: refactor PendingBlocks tests functionally they are the same, but unified into neat tabular test --- block/pending_blocks_test.go | 144 +++++++++++++++++++---------------- 1 file changed, 77 insertions(+), 67 deletions(-) diff --git a/block/pending_blocks_test.go b/block/pending_blocks_test.go index d2e160e37e..65b9bbd4a9 100644 --- a/block/pending_blocks_test.go +++ b/block/pending_blocks_test.go @@ -4,84 +4,77 @@ import ( "context" "sort" "testing" - - test "github.com/rollkit/rollkit/test/log" - - "github.com/rollkit/rollkit/store" + "time" "github.com/stretchr/testify/require" + "github.com/rollkit/rollkit/store" + test "github.com/rollkit/rollkit/test/log" "github.com/rollkit/rollkit/types" ) -func TestGetPendingBlocks(t *testing.T) { - require := require.New(t) - ctx := context.Background() - pb := newPendingBlocks(t) - for i := uint64(0); i < 5; i++ { - require.NoError(pb.store.SaveBlock(ctx, types.GetRandomBlock(i, 0), &types.Commit{})) - pb.store.SetHeight(ctx, i) - } - blocks, err := pb.getPendingBlocks(ctx) - require.NoError(err) - require.True(sort.SliceIsSorted(blocks, func(i, j int) bool { - return blocks[i].Height() < blocks[j].Height() - })) -} +const ( + numBlocks = 5 + testHeight = 3 +) -func TestRemoveSubmittedBlocks(t *testing.T) { - require := require.New(t) - ctx := context.Background() - pb := newPendingBlocks(t) - const nBlocks = 5 - for i := uint64(1); i <= nBlocks; i++ { - require.NoError(pb.store.SaveBlock(ctx, types.GetRandomBlock(i, 0), &types.Commit{})) - pb.store.SetHeight(ctx, i) +func TestPendingBlocks(t *testing.T) { + cases := []struct { + name string + init func(context.Context, *testing.T, *PendingBlocks) + exec func(context.Context, *testing.T, *PendingBlocks) + expectedBlocksAfterInit int + expectedBlocksAfterExec int + }{ + {name: "empty store", + init: func(context.Context, *testing.T, *PendingBlocks) {}, + exec: func(context.Context, *testing.T, *PendingBlocks) {}, + expectedBlocksAfterInit: 0, + expectedBlocksAfterExec: 0, + }, + { + name: "mock successful DA submission of some blocks by manually setting last submitted height", + init: fillWithBlocks, + exec: func(ctx context.Context, t *testing.T, pb *PendingBlocks) { + pb.lastSubmittedHeight.Store(testHeight) + }, + expectedBlocksAfterInit: numBlocks, + expectedBlocksAfterExec: numBlocks - testHeight, + }, + { + name: "mock successful DA submission of all blocks by manually setting last submitted height", + init: fillWithBlocks, + exec: func(ctx context.Context, t *testing.T, pb *PendingBlocks) { + pb.lastSubmittedHeight.Store(numBlocks) + }, + expectedBlocksAfterInit: numBlocks, + expectedBlocksAfterExec: 0, + }, + { + name: "mock successful DA submission of all blocks by setting last submitted height using store", + init: fillWithBlocks, + exec: func(ctx context.Context, t *testing.T, pb *PendingBlocks) { + pb.lastSubmittedHeight.Store(pb.store.Height()) + }, + expectedBlocksAfterInit: numBlocks, + expectedBlocksAfterExec: 0, + }, } - pb.setLastSubmittedHeight(nBlocks) - require.True(pb.isEmpty()) -} -func TestRemoveSubsetOfBlocks(t *testing.T) { - require := require.New(t) - ctx := context.Background() - pb := newPendingBlocks(t) - for i := uint64(1); i <= 5; i++ { - require.NoError(pb.store.SaveBlock(ctx, types.GetRandomBlock(i, 0), &types.Commit{})) - pb.store.SetHeight(ctx, i) - } - // Remove blocks with height 1 and 2 - pb.setLastSubmittedHeight(2) - remainingBlocks, err := pb.getPendingBlocks(ctx) - require.NoError(err) - require.Len(remainingBlocks, 3, "There should be 3 blocks remaining") - for _, block := range remainingBlocks { - require.Contains([]uint64{3, 4, 5}, block.Height(), "Only blocks with height 3, 4 and 5 should remain") - } -} + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + // use timeout to ensure tests will end + ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second) + defer cancel() + pb := newPendingBlocks(t) -func TestRemoveAllBlocksAndVerifyEmpty(t *testing.T) { - require := require.New(t) - ctx := context.Background() - pb := newPendingBlocks(t) - for i := uint64(0); i < 5; i++ { - require.NoError(pb.store.SaveBlock(ctx, types.GetRandomBlock(i, 0), &types.Commit{})) - pb.store.SetHeight(ctx, i) - } - // Remove all blocks - blocks, err := pb.getPendingBlocks(ctx) - require.NoError(err) - pb.setLastSubmittedHeight(blocks[len(blocks)-1].Height()) - require.True(pb.isEmpty(), "PendingBlocks should be empty after removing all blocks") -} + tc.init(ctx, t, pb) + checkRequirements(ctx, t, pb, tc.expectedBlocksAfterInit) -func TestRemoveBlocksFromEmptyPendingBlocks(t *testing.T) { - require := require.New(t) - pb := newPendingBlocks(t) - // Attempt to remove blocks from an empty PendingBlocks - require.NotPanics(func() { - pb.setLastSubmittedHeight(2) - }, "Removing blocks from an empty PendingBlocks should not cause a panic") + tc.exec(ctx, t, pb) + checkRequirements(ctx, t, pb, tc.expectedBlocksAfterExec) + }) + } } func newPendingBlocks(t *testing.T) *PendingBlocks { @@ -91,3 +84,20 @@ func newPendingBlocks(t *testing.T) *PendingBlocks { require.NoError(t, err) return pendingBlocks } + +func fillWithBlocks(ctx context.Context, t *testing.T, pb *PendingBlocks) { + for i := uint64(1); i <= numBlocks; i++ { + require.NoError(t, pb.store.SaveBlock(ctx, types.GetRandomBlock(i, 0), &types.Commit{})) + pb.store.SetHeight(ctx, i) + } +} + +func checkRequirements(ctx context.Context, t *testing.T, pb *PendingBlocks, nBlocks int) { + require.Equal(t, pb.isEmpty(), nBlocks == 0) + blocks, err := pb.getPendingBlocks(ctx) + require.NoError(t, err) + require.Len(t, blocks, nBlocks) + require.True(t, sort.SliceIsSorted(blocks, func(i, j int) bool { + return blocks[i].Height() < blocks[j].Height() + })) +} From 95c074b425ba24bae0155a180cc911514239fa64 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Zdyba=C5=82?= Date: Wed, 6 Mar 2024 23:27:46 +0100 Subject: [PATCH 20/20] fix: improve setLastSubmittedHeight Add context as parameter and add comment on error handling. --- block/manager.go | 2 +- block/pending_blocks.go | 7 +++++-- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/block/manager.go b/block/manager.go index d91816f053..f9b6c9ff41 100644 --- a/block/manager.go +++ b/block/manager.go @@ -882,7 +882,7 @@ func (m *Manager) submitBlocksToDA(ctx context.Context) error { if l := len(submittedBlocks); l > 0 { lastSubmittedHeight = submittedBlocks[l-1].Height() } - m.pendingBlocks.setLastSubmittedHeight(lastSubmittedHeight) + m.pendingBlocks.setLastSubmittedHeight(ctx, lastSubmittedHeight) blocksToSubmit = notSubmittedBlocks // reset submission options when successful // scale back gasPrice gradually diff --git a/block/pending_blocks.go b/block/pending_blocks.go index cf06c1525d..8d52d54b89 100644 --- a/block/pending_blocks.go +++ b/block/pending_blocks.go @@ -79,12 +79,15 @@ func (pb *PendingBlocks) isEmpty() bool { return pb.store.Height() == pb.lastSubmittedHeight.Load() } -func (pb *PendingBlocks) setLastSubmittedHeight(newLastSubmittedHeight uint64) { +func (pb *PendingBlocks) setLastSubmittedHeight(ctx context.Context, newLastSubmittedHeight uint64) { lsh := pb.lastSubmittedHeight.Load() if newLastSubmittedHeight > lsh && pb.lastSubmittedHeight.CompareAndSwap(lsh, newLastSubmittedHeight) { - err := pb.store.SetMetadata(context.TODO(), LastSubmittedHeightKey, []byte(strconv.FormatUint(newLastSubmittedHeight, 10))) + err := pb.store.SetMetadata(ctx, LastSubmittedHeightKey, []byte(strconv.FormatUint(newLastSubmittedHeight, 10))) if err != nil { + // This indicates IO error in KV store. We can't do much about this. + // After next successful DA submission, update will be re-attempted (with new value). + // If store is not updated, after node restart some blocks will be re-submitted to DA. pb.logger.Error("failed to store height of latest block submitted to DA", "err", err) } }