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/block/manager.go b/block/manager.go index 727a7f1d7d..f9b6c9ff41 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(), + pendingBlocks: pendingBlocks, metrics: seqMetrics, } return agg, nil @@ -784,9 +789,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 { @@ -840,7 +842,20 @@ 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(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 attempt := 0 maxBlobSize, err := m.dalc.DA.MaxBlobSize(ctx) @@ -863,7 +878,11 @@ func (m *Manager) submitBlocksToDA(ctx context.Context) error { for _, block := range submittedBlocks { m.blockCache.setDAIncluded(block.Hash().String()) } - m.pendingBlocks.removeSubmittedBlocks(submittedBlocks) + lastSubmittedHeight := uint64(0) + if l := len(submittedBlocks); l > 0 { + lastSubmittedHeight = submittedBlocks[l-1].Height() + } + m.pendingBlocks.setLastSubmittedHeight(ctx, lastSubmittedHeight) blocksToSubmit = notSubmittedBlocks // reset submission options when successful // scale back gasPrice gradually diff --git a/block/manager_test.go b/block/manager_test.go index 589397e0aa..787c5f94e0 100644 --- a/block/manager_test.go +++ b/block/manager_test.go @@ -3,9 +3,13 @@ package block import ( "bytes" "context" + "os" + "strconv" "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 +142,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 @@ -162,8 +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.pendingBlocks.addPendingBlock(block) + m.pendingBlocks, err = NewPendingBlocks(m.store, m.logger) + require.NoError(t, err) err = m.submitBlocksToDA(ctx) require.NoError(t, err) mockDA.AssertExpectations(t) @@ -171,6 +182,7 @@ func TestSubmitBlocksToMockDA(t *testing.T) { } func TestSubmitBlocksToDA(t *testing.T) { + assert := assert.New(t) require := require.New(t) ctx := context.Background() @@ -230,14 +242,41 @@ func TestSubmitBlocksToDA(t *testing.T) { } for _, tc := range testCases { - m.pendingBlocks = NewPendingBlocks() + // 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, 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 { - m.pendingBlocks.addPendingBlock(block) + require.NoError(m.store.SaveBlock(ctx, block, &types.Commit{})) } + m.store.SetHeight(ctx, uint64(len(tc.blocks))) + err := m.submitBlocksToDA(ctx) - assert.Equal(t, tc.isErrExpected, err != nil) - assert.Equal(t, tc.expectedPendingBlocksLength, len(m.pendingBlocks.getPendingBlocks())) + assert.Equal(tc.isErrExpected, err != nil) + blocks, err := m.pendingBlocks.getPendingBlocks(ctx) + assert.NoError(err) + assert.Equal(tc.expectedPendingBlocksLength, len(blocks)) + + // ensure that metadata is updated in KV store + raw, err := m.store.GetMetadata(ctx, LastSubmittedHeightKey) + require.NoError(err) + lshInKV, err := strconv.ParseUint(string(raw), 10, 64) + require.NoError(err) + assert.Equal(m.store.Height(), lshInKV+uint64(tc.expectedPendingBlocksLength)) }) } } + +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 0a04e55b49..8d52d54b89 100644 --- a/block/pending_blocks.go +++ b/block/pending_blocks.go @@ -1,64 +1,112 @@ package block import ( - "sort" - "sync" + "context" + "errors" + "fmt" + "strconv" + "sync/atomic" + ds "github.com/ipfs/go-datastore" + + "github.com/rollkit/rollkit/store" + "github.com/rollkit/rollkit/third_party/log" "github.com/rollkit/rollkit/types" ) +// 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 +// +// 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): we shouldn't try to push all pending blocks at once; this should depend on max blob size type PendingBlocks struct { - pendingBlocks map[uint64]*types.Block - mtx *sync.RWMutex + 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() *PendingBlocks { - return &PendingBlocks{ - pendingBlocks: make(map[uint64]*types.Block), - mtx: new(sync.RWMutex), +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() []*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(ctx context.Context) ([]*types.Block, error) { + lastSubmitted := pb.lastSubmittedHeight.Load() + height := pb.store.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)) + } -// 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 { + blocks := make([]*types.Block, 0, height-lastSubmitted) + for i := lastSubmitted + 1; i <= height; i++ { + block, err := pb.store.GetBlock(ctx, 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) setLastSubmittedHeight(ctx context.Context, newLastSubmittedHeight uint64) { + lsh := pb.lastSubmittedHeight.Load() + + if newLastSubmittedHeight > lsh && pb.lastSubmittedHeight.CompareAndSwap(lsh, newLastSubmittedHeight) { + 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) + } + } } -func (pb *PendingBlocks) removeSubmittedBlocks(blocks []*types.Block) { - pb.mtx.Lock() - defer pb.mtx.Unlock() - for _, block := range blocks { - delete(pb.pendingBlocks, block.Height()) +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 + 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 ef43130ca7..65b9bbd4a9 100644 --- a/block/pending_blocks_test.go +++ b/block/pending_blocks_test.go @@ -1,74 +1,103 @@ package block import ( + "context" "sort" "testing" + "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) - pb := NewPendingBlocks() - for i := uint64(0); i < 5; i++ { - pb.addPendingBlock(types.GetRandomBlock(i, 0)) +const ( + numBlocks = 5 + testHeight = 3 +) + +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, + }, } - blocks := pb.getPendingBlocks() - require.True(sort.SliceIsSorted(blocks, func(i, j int) bool { - return blocks[i].Height() < blocks[j].Height() - })) -} -func TestRemoveSubmittedBlocks(t *testing.T) { - require := require.New(t) - pb := NewPendingBlocks() - for i := uint64(0); i < 5; i++ { - pb.addPendingBlock(types.GetRandomBlock(i, 0)) + 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) + + tc.init(ctx, t, pb) + checkRequirements(ctx, t, pb, tc.expectedBlocksAfterInit) + + tc.exec(ctx, t, pb) + checkRequirements(ctx, t, pb, tc.expectedBlocksAfterExec) + }) } - blocks := pb.getPendingBlocks() - pb.removeSubmittedBlocks(blocks) - require.True(pb.isEmpty()) } -func TestRemoveSubsetOfBlocks(t *testing.T) { - require := require.New(t) - pb := NewPendingBlocks() - for i := uint64(0); i < 5; i++ { - pb.addPendingBlock(types.GetRandomBlock(i, 0)) - } - // Remove blocks with height 1 and 2 - pb.removeSubmittedBlocks([]*types.Block{ - types.GetRandomBlock(1, 0), - types.GetRandomBlock(2, 0), - }) - remainingBlocks := pb.getPendingBlocks() - 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") - } +func newPendingBlocks(t *testing.T) *PendingBlocks { + kv, err := store.NewDefaultInMemoryKVStore() + require.NoError(t, err) + pendingBlocks, err := NewPendingBlocks(store.New(kv), test.NewLogger(t)) + require.NoError(t, err) + return pendingBlocks } -func TestRemoveAllBlocksAndVerifyEmpty(t *testing.T) { - require := require.New(t) - pb := NewPendingBlocks() - for i := uint64(0); i < 5; i++ { - pb.addPendingBlock(types.GetRandomBlock(i, 0)) +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) } - // Remove all blocks - pb.removeSubmittedBlocks(pb.getPendingBlocks()) - require.True(pb.isEmpty(), "PendingBlocks should be empty after removing all blocks") } -func TestRemoveBlocksFromEmptyPendingBlocks(t *testing.T) { - require := require.New(t) - pb := NewPendingBlocks() - // Attempt to remove blocks from an empty PendingBlocks - require.NotPanics(func() { - pb.removeSubmittedBlocks([]*types.Block{ - types.GetRandomBlock(1, 0), - types.GetRandomBlock(2, 0), - }) - }, "Removing blocks from an empty PendingBlocks should not cause a panic") +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() + })) } 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..41f13e0b56 100644 --- a/node/full_node_test.go +++ b/node/full_node_test.go @@ -3,10 +3,21 @@ package node import ( "context" "crypto/rand" + "crypto/sha256" + "errors" "fmt" + "os" + "strconv" "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 +160,139 @@ 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 +// +// 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() + + 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 + secondRunBlocks = 5 + ) + + 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 collect all blobs in order + // node will be stopped after producing at least firstRunBlocks blocks + // 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[:] + } + allBlobs = append(allBlobs, blobs...) + return hashes, nil + }) + + err = node.Start() + assert.NoError(t, err) + + // let node produce few more blocks + err = waitForAtLeastNBlocks(node, firstRunBlocks+secondRunBlocks, 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) + 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) { + 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/store/store.go b/store/store.go index c6c5ce6875..12f80bc497 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,26 @@ 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 { + 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) { + 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 func (s *DefaultStore) loadHashFromIndex(ctx context.Context, height uint64) (header.Hash, error) { blob, err := s.db.Get(ctx, ds.NewKey(getIndexKey(height))) @@ -239,3 +260,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..578f7b1491 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,37 @@ func TestBlockResponses(t *testing.T) { assert.NotNil(resp) assert.Equal(expected, resp) } + +func TestMetadata(t *testing.T) { + t.Parallel() + 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++ { + require.NoError(s.SetMetadata(ctx, getKey(i), getValue(i))) + } + + for i := 0; i < n; i++ { + value, err := s.GetMetadata(ctx, getKey(i)) + require.NoError(err) + require.Equal(getValue(i), value) + } + + v, err := s.GetMetadata(ctx, "unused key") + require.Error(err) + require.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 } 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 +}