From 851cc3dcddef17102de17923b2ad62aa41da51b2 Mon Sep 17 00:00:00 2001 From: Martin Tomazic Date: Thu, 4 Dec 2025 19:48:53 +0100 Subject: [PATCH 1/5] go/oasis-node/cmd/storage: Add create and import checkpoint cmd --- .changelog/6454.trivial.md | 0 go/oasis-node/cmd/storage/checkpoint.go | 623 +++++++++++++++++++ go/oasis-node/cmd/storage/checkpoint_test.go | 323 ++++++++++ go/oasis-node/cmd/storage/dbs.go | 22 +- go/oasis-node/cmd/storage/storage.go | 1 + go/storage/mkvs/checkpoint/chunk.go | 2 +- go/storage/mkvs/checkpoint/restorer.go | 2 +- 7 files changed, 963 insertions(+), 10 deletions(-) create mode 100644 .changelog/6454.trivial.md create mode 100644 go/oasis-node/cmd/storage/checkpoint.go create mode 100644 go/oasis-node/cmd/storage/checkpoint_test.go diff --git a/.changelog/6454.trivial.md b/.changelog/6454.trivial.md new file mode 100644 index 00000000000..e69de29bb2d diff --git a/go/oasis-node/cmd/storage/checkpoint.go b/go/oasis-node/cmd/storage/checkpoint.go new file mode 100644 index 00000000000..fb19cf155e7 --- /dev/null +++ b/go/oasis-node/cmd/storage/checkpoint.go @@ -0,0 +1,623 @@ +package storage + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "runtime" + + cmtCfg "github.com/cometbft/cometbft/config" + cmtProtoState "github.com/cometbft/cometbft/proto/tendermint/state" + cmtstore "github.com/cometbft/cometbft/proto/tendermint/store" + cmtProto "github.com/cometbft/cometbft/proto/tendermint/types" + cmtState "github.com/cometbft/cometbft/state" + "github.com/cometbft/cometbft/store" + cmttypes "github.com/cometbft/cometbft/types" + "github.com/cometbft/cometbft/version" + "github.com/cosmos/gogoproto/proto" + "github.com/spf13/cobra" + + "github.com/oasisprotocol/oasis-core/go/common" + "github.com/oasisprotocol/oasis-core/go/common/cbor" + cmdCommon "github.com/oasisprotocol/oasis-core/go/oasis-node/cmd/common" + "github.com/oasisprotocol/oasis-core/go/storage/mkvs/checkpoint" + "github.com/oasisprotocol/oasis-core/go/storage/mkvs/db/api" + "github.com/oasisprotocol/oasis-core/go/storage/mkvs/node" +) + +const ( + consensusSubdir = "consensus" + runtimesSubdir = "runtimes" + + consensusMetaFilename = "bootstrap.cbor" +) + +func newCheckpointCmd() *cobra.Command { + cmd := &cobra.Command{ + Use: "checkpoint", + Short: "create and import storage checkpoints", + PersistentPreRunE: func(_ *cobra.Command, args []string) error { + if err := cmdCommon.Init(); err != nil { + cmdCommon.EarlyLogAndExit(err) + } + running, err := cmdCommon.IsNodeRunning() + if err != nil { + return fmt.Errorf("failed to ensure the node is not running: %w", err) + } + if running { + return fmt.Errorf("checkpoint operations can only be done when the node is not running") + } + return nil + }, + } + + cmd.AddCommand(newCreateCmd()) + cmd.AddCommand(newImportCmd()) + + return cmd +} + +func newCreateCmd() *cobra.Command { + var ( + height uint64 + runtimeID string + round uint64 + outDir string + ) + + cmd := &cobra.Command{ + Use: "create", + Args: cobra.NoArgs, + Short: "create storage checkpoints", + RunE: func(cmd *cobra.Command, args []string) error { + ctx := cmd.Context() + dataDir := cmdCommon.DataDir() + + // Consensus checkpoint: + createConsensusCp := func(outputDir string) error { + ndb, close, err := openConsensusNodeDB(dataDir) + if err != nil { + return fmt.Errorf("failed to open consensus state DB: %w", err) + } + defer close() + return createCheckpoints(ctx, ndb, common.Namespace{}, height, outputDir) + } + if height != 0 { + consensusOutDir := filepath.Join(outDir, consensusSubdir) + logger.Info("creating consensus checkpoint", + "height", height, + "output_dir", consensusOutDir, + ) + if err := createConsensusCp(consensusOutDir); err != nil { + return fmt.Errorf("failed to create consensus checkpoint (height: %d): %w", height, err) + } + logger.Info("consensus checkpoint created") + + if err := createCometBFTBootstrapMeta(dataDir, height, consensusOutDir); err != nil { + return fmt.Errorf("failed to write bootstrap metadata: %w", err) + } + logger.Info("bootstrap metadata created") + } + + // Runtime Checkpoints: + createRuntimeCps := func(ns common.Namespace, outputDir string) error { + ndb, err := openRuntimeStateDB(dataDir, ns) + if err != nil { + return fmt.Errorf("failed to open runtime state DB: %w", err) + } + defer ndb.Close() + return createCheckpoints(ctx, ndb, ns, round, outputDir) + } + if runtimeID != "" { + var ns common.Namespace + if err := ns.UnmarshalHex(runtimeID); err != nil { + return fmt.Errorf("malformed source runtime ID: %q: %w", runtimeID, err) + } + rtOutDir := filepath.Join(outDir, runtimesSubdir, ns.Hex()) + logger.Info("creating runtime checkpoints (may take minutes to hours depending on the db size)", + "round", round, + "output_dir", rtOutDir, + "runtime_id", ns, + ) + if err := createRuntimeCps(ns, rtOutDir); err != nil { + return fmt.Errorf("failed to create runtime checkpoints (runtime: %s, round: %d): %w", runtimeID, round, err) + } + logger.Info("runtime checkpoints created", "runtime_id", ns) + } + + return nil + }, + } + + cmd.Flags().Uint64Var(&height, "height", 0, "consensus height") + cmd.Flags().StringVar(&runtimeID, "runtime", "", "hex encoded runtime ID") + cmd.Flags().Uint64Var(&round, "round", 0, "round for which checkpoints will be created") + cmd.Flags().StringVar(&outDir, "output-dir", "", "output directory") + + return cmd +} + +func newImportCmd() *cobra.Command { + var inputDir string + + cmd := &cobra.Command{ + Use: "import", + Args: cobra.NoArgs, + Short: "import storage checkpoints", + RunE: func(cmd *cobra.Command, args []string) error { + ctx := cmd.Context() + dataDir := cmdCommon.DataDir() + + // Consensus checkpoint: + importConsensusCp := func(inputDir string) error { + ndb, close, err := openConsensusNodeDB(dataDir) + if err != nil { + return fmt.Errorf("failed to open consensus state DB: %w", err) + } + defer close() + + return importCheckpoints(ctx, ndb, common.Namespace{}, inputDir) + } + consensusInputDir := filepath.Join(inputDir, consensusSubdir) + _, err := os.ReadDir(consensusInputDir) + switch { + case err == nil: + logger.Info("importing consensus checkpoint", "input_dir", inputDir) + if err := importConsensusCp(consensusInputDir); err != nil { + return fmt.Errorf("failed to import consensus checkpoint (input dir: %s): %w", consensusInputDir, err) + } + logger.Info("consensus checkpoint imported") + + meta, err := readCometBFTBootstrapMeta(consensusInputDir) + if err != nil { + return fmt.Errorf("failed to read bootstrap metadata: %w", err) + } + if err := bootstrapTrustedState(dataDir, meta); err != nil { + return fmt.Errorf("failed to bootstrap CometBFT trusted state: %w", err) + } + logger.Info("bootstrap metadata imported") + case os.IsNotExist(err): + // No consensus checkpoints to import + default: + return fmt.Errorf("failed to stat: %w", err) + } + + // Runtime checkpoints: + importRuntimeCp := func(inputDir string, ns common.Namespace) error { + ndb, err := openRuntimeStateDB(dataDir, ns) + if err != nil { + return err + } + defer ndb.Close() + + return importCheckpoints(ctx, ndb, ns, inputDir) + } + runtimeInputDir := filepath.Join(inputDir, runtimesSubdir) + entries, err := os.ReadDir(runtimeInputDir) + switch { + case err == nil: + for _, entry := range entries { + var ns common.Namespace + if err := ns.UnmarshalHex(entry.Name()); err != nil { + return fmt.Errorf("malformed source runtime ID: %s: %w", entry.Name(), err) + } + rtCpsDir := filepath.Join(runtimeInputDir, entry.Name()) + + logger.Info("importing runtime checkpoints (may take few minutes)", + "runtime_id", ns, + "input_dir", inputDir, + ) + if err := importRuntimeCp(rtCpsDir, ns); err != nil { + return fmt.Errorf("failed to import checkpoints (checkpoint dir: %s): %w", rtCpsDir, err) + } + logger.Info("runtime checkpoints imported", "runtime_id", ns) + } + case os.IsNotExist(err): + // No runtime checkpoints to import + default: + return fmt.Errorf("failed to stat: %w", err) + } + + return nil + }, + } + + cmd.Flags().StringVar(&inputDir, "input-dir", "", "directory with checkpoints to import") + + return cmd +} + +func createCheckpoints(ctx context.Context, ndb api.NodeDB, ns common.Namespace, version uint64, outputDir string) error { + if err := ensureEmptyDir(outputDir); err != nil { + return err + } + + latest, ok := ndb.GetLatestVersion() + if !ok { + return fmt.Errorf("empty state DB") + } + earliest := ndb.GetEarliestVersion() + if version < earliest || version > latest { + return fmt.Errorf("version not found (available range: %d-%d)", earliest, latest) + } + + roots, err := ndb.GetRootsForVersion(version) + if err != nil { + return fmt.Errorf("failed to get roots %w", err) + } + if len(roots) == 0 { + // Empty roots are implicit in NodeDB and therefore not returned by GetRootsForVersion. + // If at this height state DB has an empty state, create a checkpoint for the empty state root, + // so that importing it will still finalize this version (making NodeDB non-empty). + // + // In case of state DB with multiple empty roots (e.g. state and IO root both empty) + // this will create only one empty checkpoint (duplicate hash), which is safe as we only need + // it to finalize version with implicitly empty state and IO root. + stateRoot := emptyRoot(ns, version, node.RootTypeState) + roots = []node.Root{stateRoot} + } + + creator, err := checkpoint.NewFileCreator(outputDir, ndb) + if err != nil { + return fmt.Errorf("failed to create file creator: %w", err) + } + + const chunkSize uint64 = 8 * 1024 * 1024 // 8MiB chunks + // Use parallel chunking algorithm, but limit concurrency. + chunkerThreads := uint16(runtime.GOMAXPROCS(0) / 2) + for _, root := range roots { + _, err := creator.CreateCheckpoint(ctx, root, chunkSize, chunkerThreads) + if err != nil { + return fmt.Errorf("failed to create checkpoint (rootType: %s): %w", root.Type, err) + } + } + + return nil +} + +func importCheckpoints(ctx context.Context, ndb api.NodeDB, ns common.Namespace, inputDir string) error { + isEmpty, err := isEmptyDir(inputDir) + if err != nil { + return fmt.Errorf("failed to inspect input directory: %w", err) + } + if isEmpty { + return fmt.Errorf("input directory is empty: %s", inputDir) + } + if _, ok := ndb.GetLatestVersion(); ok { + return fmt.Errorf("state db not empty") + } + + provider, err := checkpoint.NewFileCreator(inputDir, nil) // ndb = nil since we will only import checkpoints. + if err != nil { + return fmt.Errorf("failed to create checkpoint file creator: %w", err) + } + cps, err := provider.GetCheckpoints(ctx, &checkpoint.GetCheckpointsRequest{Version: 1, Namespace: ns}) + if err != nil { + return fmt.Errorf("failed to read checkpoints: %w", err) + } + + var roots []node.Root + for _, cp := range cps { + if err := importCp(ctx, ndb, provider, cp); err != nil { + return fmt.Errorf("failed to import checkpoint (root hash: %s): %w", cp.Root.Hash, err) + } + roots = append(roots, cp.Root) + } + if err := ndb.Finalize(roots); err != nil { + return fmt.Errorf("failed to finalize: %w", err) + } + + return nil +} + +func importCp(ctx context.Context, ndb api.NodeDB, provider checkpoint.Creator, cp *checkpoint.Metadata) error { + if err := ndb.StartMultipartInsert(cp.Root.Version); err != nil { + return fmt.Errorf("failed to start multipart insert: %w", err) + } + defer func() { + if err := ndb.AbortMultipartInsert(); err != nil { + logger.Error("failed to abort multi-part insert", "err", err) + } + }() + + for idx := range cp.Chunks { + chunk, err := cp.GetChunkMetadata(uint64(idx)) + if err != nil { + return fmt.Errorf("failed to get chunk metadata: %w", err) + } + var buf bytes.Buffer + if err := provider.GetCheckpointChunk(ctx, chunk, &buf); err != nil { + return fmt.Errorf("failed to read checkpoint chunk (idx: %d): %w", idx, err) + } + if err := checkpoint.RestoreChunk(ctx, ndb, chunk, &buf); err != nil { + return fmt.Errorf("failed to restore chunk: %w", err) + } + } + return nil +} + +type bootstrapMeta struct { + State []byte `json:"state"` + Commit []byte `json:"commit"` +} + +func createCometBFTBootstrapMeta(dataDir string, height uint64, outputDir string) error { + stateStore, err := openConsensusStatestore(dataDir) + if err != nil { + return fmt.Errorf("failed to open cometbft state store: %w", err) + } + defer stateStore.Close() + + blockStore, err := openConsensusBlockstore(dataDir) + if err != nil { + return fmt.Errorf("failed to open consensus blockstore: %w", err) + } + defer blockStore.Close() + + state, err := getState(height, stateStore, blockStore) + if err != nil { + return fmt.Errorf("failed to load consensus state at height %d: %w", height, err) + } + statePB, err := state.ToProto() + if err != nil { + return fmt.Errorf("failed to convert consensus state to proto: %w", err) + } + stateBytes, err := proto.Marshal(statePB) + if err != nil { + return fmt.Errorf("failed to marshal consensus state: %w", err) + } + + commit, err := getCommit(blockStore, height) + if err != nil { + return fmt.Errorf("failed to load consensus commit at height %d: %w", height, err) + } + commitBytes, err := proto.Marshal(commit.ToProto()) + if err != nil { + return fmt.Errorf("failed to marshal consensus commit: %w", err) + } + + meta := bootstrapMeta{ + State: stateBytes, + Commit: commitBytes, + } + if err := os.WriteFile(filepath.Join(outputDir, consensusMetaFilename), cbor.Marshal(meta), 0o600); err != nil { + return fmt.Errorf("failed to write bootstrap metadata: %w", err) + } + + return nil +} + +func readCometBFTBootstrapMeta(inputDir string) (bootstrapMeta, error) { + data, err := os.ReadFile(filepath.Join(inputDir, consensusMetaFilename)) + if err != nil { + return bootstrapMeta{}, err + } + + var meta bootstrapMeta + if err := cbor.Unmarshal(data, &meta); err != nil { + return bootstrapMeta{}, fmt.Errorf("failed to decode bootstrap metadata: %w", err) + } + + return meta, nil +} + +// bootstrapTrustedState synchronizes the cometbft databases after the state sync +// has been performed offline. +// +// It is expected that the block store and state store are empty at the time the +// function is called. +// +// Adapted from https://github.com/oasisprotocol/cometbft/blob/08e22df73d354512fc27bd0c5731b3dcf1f8fef7/node/node.go#L198. +func bootstrapTrustedState(dataDir string, meta bootstrapMeta) error { + stateDB, err := openConsensusStateDB(dataDir) + if err != nil { + return fmt.Errorf("failed to open cometbft state store: %w", err) + } + defer stateDB.Close() + + blockStoreDB, err := openConsensusBlockstoreDB(dataDir) + if err != nil { + return fmt.Errorf("failed to open consensus blockstore: %w", err) + } + defer blockStoreDB.Close() + blockStore := store.NewBlockStore(blockStoreDB) + + if !blockStore.IsEmpty() { + return fmt.Errorf("blockstore not empty, trying to initialize non empty state") + } + + stateStore := cmtState.NewBootstrapStore(stateDB, cmtState.StoreOptions{ + DiscardABCIResponses: cmtCfg.DefaultConfig().Storage.DiscardABCIResponses, + }) + defer stateStore.Close() + + state, err := stateStore.Load() + if err != nil { + return err + } + + if !state.IsEmpty() { + return fmt.Errorf("state not empty, trying to initialize non empty state") + } + + var statePB cmtProtoState.State + if err := proto.Unmarshal(meta.State, &statePB); err != nil { + return fmt.Errorf("failed to unmarshal consensus state: %w", err) + } + metaState, err := cmtState.FromProto(&statePB) + if err != nil { + return fmt.Errorf("failed to parse consensus state: %w", err) + } + + var commitPB cmtProto.Commit + if err := proto.Unmarshal(meta.Commit, &commitPB); err != nil { + return fmt.Errorf("failed to unmarshal consensus commit: %w", err) + } + commit, err := cmttypes.CommitFromProto(&commitPB) + if err != nil { + return fmt.Errorf("failed to parse consensus commit: %w", err) + } + + if err = stateStore.Bootstrap(*metaState); err != nil { + return err + } + + err = blockStore.SaveSeenCommit(metaState.LastBlockHeight, commit) + if err != nil { + return err + } + + // SaveSeenCommit (as used by the upstream) does not persist the BlockStoreState (height/base), + // so we save it explicitly here. This way the blockstore reports the correct height after bootstrap, + // so that status works as expected. If this is not done, this blockstore base is kept as uninitialized, + // and is initialized as soon as the first block is fetched. However, we would cut last retained height by 1, + // which is the current case for consensus checkpoint sync. + store.SaveBlockStoreState( + &cmtstore.BlockStoreState{ + Base: metaState.LastBlockHeight, + Height: metaState.LastBlockHeight, + }, + blockStoreDB, + ) + + // Once the stores are bootstrapped, we need to set the height at which the node has finished + // statesyncing. This will allow the blocksync reactor to fetch blocks at a proper height. + // In case this operation fails, it is equivalent to a failure in online state sync where the operator + // needs to manually delete the state and blockstores and rerun the bootstrapping process. + err = stateStore.SetOfflineStateSyncHeight(metaState.LastBlockHeight) + if err != nil { + return fmt.Errorf("failed to set synced height: %w", err) + } + + return nil +} + +// getCommit is adapted and simplified and mimics StateProvider behaviour used in the upstream BootstrapState. +func getCommit(blockStore *store.BlockStore, height uint64) (*cmttypes.Commit, error) { + commit := blockStore.LoadBlockCommit(int64(height)) + if commit == nil { + return nil, fmt.Errorf("commit not found at height %d", height) + } + return commit, nil +} + +// getState is adapted and mimics StateProvider behaviour used in the upstream BootstrapState. +func getState(height uint64, stateStore cmtState.Store, blockStore *store.BlockStore) (cmtState.State, error) { + // The snapshot height maps onto the state heights as follows: + // + // height: last block, i.e. the snapshotted height + // height+1: current block, i.e. the first block we'll process after the snapshot + // height+2: next block, i.e. the second block after the snapshot + // + // We need to fetch the NextValidators from height+2 because if the application changed + // the validator set at the snapshot height then this only takes effect at height+2. + h := int64(height) + lastMeta := blockStore.LoadBlockMeta(h) + if lastMeta == nil { + return cmtState.State{}, fmt.Errorf("block meta not found at height %d", h) + } + currentMeta := blockStore.LoadBlockMeta(h + 1) + if currentMeta == nil { + return cmtState.State{}, fmt.Errorf("block meta not found at height %d", h+1) + } + + lastVals, err := stateStore.LoadValidators(h) + if err != nil { + return cmtState.State{}, err + } + currentVals, err := stateStore.LoadValidators(h + 1) + if err != nil { + return cmtState.State{}, err + } + nextVals, err := stateStore.LoadValidators(h + 2) + if err != nil { + return cmtState.State{}, err + } + + consensusParams, err := stateStore.LoadConsensusParams(h + 1) + if err != nil { + return cmtState.State{}, err + } + + storeState, err := stateStore.Load() + if err != nil { + return cmtState.State{}, err + } + if storeState.IsEmpty() { + return cmtState.State{}, fmt.Errorf("state store is empty") + } + + state := cmtState.State{ + ChainID: storeState.ChainID, + Version: cmtProtoState.Version{ + Consensus: currentMeta.Header.Version, + Software: version.TMCoreSemVer, + }, + InitialHeight: storeState.InitialHeight, + } + if state.InitialHeight == 0 { + state.InitialHeight = 1 + } + + state.LastBlockHeight = lastMeta.Header.Height + state.LastBlockTime = lastMeta.Header.Time + state.LastBlockID = lastMeta.BlockID + state.AppHash = currentMeta.Header.AppHash + state.LastResultsHash = currentMeta.Header.LastResultsHash + state.LastValidators = lastVals + state.Validators = currentVals + state.NextValidators = nextVals + state.LastHeightValidatorsChanged = h + 2 + state.ConsensusParams = consensusParams + state.LastHeightConsensusParamsChanged = currentMeta.Header.Height + + return state, nil +} + +func isEmptyDir(dir string) (bool, error) { + // Avoid using os.ReadDir as it reads the whole dir. + f, err := os.Open(dir) + if err != nil { + return false, fmt.Errorf("failed to open directory %q: %w", dir, err) + } + defer f.Close() + + switch _, err = f.Readdir(1); { + case err == nil: + return false, nil + case errors.Is(err, io.EOF): + default: + return false, fmt.Errorf("failed to read directory %q: %w", dir, err) + } + return true, nil +} + +func ensureEmptyDir(dir string) error { + if err := os.MkdirAll(dir, 0o700); err != nil { + return fmt.Errorf("failed to create output directory: %w", err) + } + + isEmpty, err := isEmptyDir(dir) + if err != nil { + return err + } + if !isEmpty { + return fmt.Errorf("output directory is not empty: %s", dir) + } + + return nil +} + +func emptyRoot(ns common.Namespace, version uint64, rootType node.RootType) node.Root { + root := node.Root{ + Namespace: ns, + Version: version, + Type: rootType, + } + root.Hash.Empty() + return root +} diff --git a/go/oasis-node/cmd/storage/checkpoint_test.go b/go/oasis-node/cmd/storage/checkpoint_test.go new file mode 100644 index 00000000000..a0d08e5980e --- /dev/null +++ b/go/oasis-node/cmd/storage/checkpoint_test.go @@ -0,0 +1,323 @@ +package storage + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/oasisprotocol/oasis-core/go/common" + "github.com/oasisprotocol/oasis-core/go/storage/mkvs" + dbAPI "github.com/oasisprotocol/oasis-core/go/storage/mkvs/db/api" + "github.com/oasisprotocol/oasis-core/go/storage/mkvs/db/pathbadger" + "github.com/oasisprotocol/oasis-core/go/storage/mkvs/node" +) + +const testVersion uint64 = 10 + +var testNs common.Namespace = common.NewTestNamespaceFromSeed([]byte("test namespace"), 0) + +func TestCreateCheckpoints(t *testing.T) { + t.Run("fails on non-empty output dir", func(t *testing.T) { + t.Parallel() + + ctx := t.Context() + ns := common.Namespace{} + ndb, err := newTestNodeDB(t, ns) + require.NoError(t, err) + defer ndb.Close() + + stateRoot := commitRoot(ctx, t, ndb, ns, testVersion, node.RootTypeState, map[string]string{ + "key": "value", + }) + require.NoError(t, ndb.Finalize([]node.Root{stateRoot})) + + outDir := filepath.Join(t.TempDir(), "checkpoints") + require.NoError(t, os.Mkdir(outDir, 0o700)) + require.NoError(t, os.WriteFile(filepath.Join(outDir, "existing"), []byte{}, 0o600)) + + err = createCheckpoints(ctx, ndb, ns, testVersion, outDir) + require.ErrorContains(t, err, "output directory is not empty") + }) + + t.Run("fails on empty node DB", func(t *testing.T) { + t.Parallel() + + ctx := t.Context() + ns := common.Namespace{} + ndb, err := newTestNodeDB(t, ns) + require.NoError(t, err) + defer ndb.Close() + + cpDir := filepath.Join(t.TempDir(), "checkpoint") + err = createCheckpoints(ctx, ndb, ns, testVersion, cpDir) + require.ErrorContains(t, err, "empty state DB") + }) + + t.Run("fails when version is before earliest", func(t *testing.T) { + t.Parallel() + + ctx := t.Context() + ns := common.Namespace{} + ndb, err := newTestNodeDB(t, ns) + require.NoError(t, err) + defer ndb.Close() + + stateRoot := commitRoot(ctx, t, ndb, ns, testVersion, node.RootTypeState, map[string]string{ + "key": "value", + }) + require.NoError(t, ndb.Finalize([]node.Root{stateRoot})) + + cpDir := filepath.Join(t.TempDir(), "checkpoint") + err = createCheckpoints(ctx, ndb, ns, testVersion-1, cpDir) + require.ErrorContains(t, err, "version not found") + }) + + t.Run("fails when version is after latest", func(t *testing.T) { + t.Parallel() + + ctx := t.Context() + ns := common.Namespace{} + ndb, err := newTestNodeDB(t, ns) + require.NoError(t, err) + defer ndb.Close() + + stateRoot := commitRoot(ctx, t, ndb, ns, testVersion, node.RootTypeState, map[string]string{ + "key": "value", + }) + require.NoError(t, ndb.Finalize([]node.Root{stateRoot})) + + cpDir := filepath.Join(t.TempDir(), "checkpoint") + err = createCheckpoints(ctx, ndb, ns, testVersion+1, cpDir) + require.ErrorContains(t, err, "version not found") + }) +} + +func TestImportCheckpoints(t *testing.T) { + t.Run("fails on non-empty destination DB", func(t *testing.T) { + t.Parallel() + + ctx := t.Context() + ns := common.Namespace{} + + srcNDB, err := newTestNodeDB(t, ns) + require.NoError(t, err) + defer srcNDB.Close() + + stateRoot := commitRoot(ctx, t, srcNDB, ns, testVersion, node.RootTypeState, map[string]string{ + "key": "value", + }) + require.NoError(t, srcNDB.Finalize([]node.Root{stateRoot})) + + cpDir := filepath.Join(t.TempDir(), "checkpoint") + require.NoError(t, createCheckpoints(ctx, srcNDB, ns, testVersion, cpDir)) + + dstNDB, err := newTestNodeDB(t, ns) + require.NoError(t, err) + defer dstNDB.Close() + + // NodeDB allows importing checkpoint even if the target NodeDB is not empty, + // as long as the checkpoint is younger than the the last finalized version. + // We don't want that, thus finalizing testVersion-1 to ensure createCheckpoints + // guard is tested and that the test does not rely on mkvs version already finalized. + err = dstNDB.Finalize([]node.Root{emptyRoot(ns, testVersion-1, node.RootTypeState)}) + require.NoError(t, err) + + err = importCheckpoints(ctx, dstNDB, ns, cpDir) + require.ErrorContains(t, err, "state db not empty") + }) + + t.Run("fails on empty input dir", func(t *testing.T) { + t.Parallel() + + ctx := t.Context() + ns := common.Namespace{} + ndb, err := newTestNodeDB(t, ns) + require.NoError(t, err) + defer ndb.Close() + + err = importCheckpoints(ctx, ndb, ns, t.TempDir()) + require.ErrorContains(t, err, "input directory is empty") + }) +} + +func TestCreateImportRoundtrip(t *testing.T) { + t.Run("single non-empty checkpoint", func(t *testing.T) { + t.Parallel() + + ctx := t.Context() + ns := common.Namespace{} + + srcNDB, err := newTestNodeDB(t, ns) + require.NoError(t, err) + defer srcNDB.Close() + + stateRoot := commitRoot(ctx, t, srcNDB, ns, testVersion, node.RootTypeState, map[string]string{ + "key": "value", + }) + require.NoError(t, srcNDB.Finalize([]node.Root{stateRoot})) + + cpDir := filepath.Join(t.TempDir(), "checkpoint") + require.NoError(t, createCheckpoints(ctx, srcNDB, ns, testVersion, cpDir)) + + dstNDB, err := newTestNodeDB(t, ns) + require.NoError(t, err) + defer dstNDB.Close() + + require.NoError(t, importCheckpoints(ctx, dstNDB, ns, cpDir)) + + roots, err := dstNDB.GetRootsForVersion(testVersion) + require.NoError(t, err) + require.Equal(t, []node.Root{stateRoot}, roots) + require.True(t, dstNDB.HasRoot(stateRoot)) + }) + + t.Run("empty root is imported", func(t *testing.T) { + t.Parallel() + + ctx := t.Context() + ns := common.Namespace{} + stateRoot := emptyRoot(ns, testVersion, node.RootTypeState) + + srcNDB, err := newTestNodeDB(t, ns) + require.NoError(t, err) + defer srcNDB.Close() + + require.NoError(t, srcNDB.Finalize([]node.Root{stateRoot})) + + latest, ok := srcNDB.GetLatestVersion() + require.True(t, ok) + require.EqualValues(t, testVersion, latest) + + // Empty roots are implicit yet GetRootForVersion does not return them. + roots, err := srcNDB.GetRootsForVersion(testVersion) + require.NoError(t, err) + require.Empty(t, roots) + require.True(t, srcNDB.HasRoot(stateRoot)) + + cpDir := filepath.Join(t.TempDir(), "checkpoint") + require.NoError(t, createCheckpoints(ctx, srcNDB, ns, testVersion, cpDir)) + + dstNDB, err := newTestNodeDB(t, ns) + require.NoError(t, err) + defer dstNDB.Close() + + require.NoError(t, importCheckpoints(ctx, dstNDB, ns, cpDir)) + + latest, ok = dstNDB.GetLatestVersion() + require.True(t, ok) + require.EqualValues(t, testVersion, latest) + + roots, err = dstNDB.GetRootsForVersion(testVersion) + require.NoError(t, err) + // Via checkpoint roundtrip, empty state can be materialized as an explicit root, + // which technically makes checkpoint restoration not 1:1 match. Same issue would + // happen with checkpointer if this case would be ever triggered. + require.Contains(t, roots, stateRoot) + require.True(t, dstNDB.HasRoot(stateRoot)) + }) + + t.Run("two non-empty roots (state and io)", func(t *testing.T) { + t.Parallel() + + ctx := t.Context() + + srcNDB, err := newTestNodeDB(t, testNs) + require.NoError(t, err) + defer srcNDB.Close() + + stateRoot := commitRoot(ctx, t, srcNDB, testNs, testVersion, node.RootTypeState, map[string]string{ + "state-key": "state-value", + }) + ioRoot := commitRoot(ctx, t, srcNDB, testNs, testVersion, node.RootTypeIO, map[string]string{ + "io-key": "io-value", + }) + require.NoError(t, srcNDB.Finalize([]node.Root{stateRoot, ioRoot})) + + cpDir := filepath.Join(t.TempDir(), "checkpoint") + require.NoError(t, createCheckpoints(ctx, srcNDB, testNs, testVersion, cpDir)) + + dstNDB, err := newTestNodeDB(t, testNs) + require.NoError(t, err) + defer dstNDB.Close() + + require.NoError(t, importCheckpoints(ctx, dstNDB, testNs, cpDir)) + + roots, err := dstNDB.GetRootsForVersion(testVersion) + require.NoError(t, err) + require.ElementsMatch(t, []node.Root{stateRoot, ioRoot}, roots) + }) + + t.Run("non-empty state root and empty io root", func(t *testing.T) { + t.Parallel() + + ctx := t.Context() + + srcNDB, err := newTestNodeDB(t, testNs) + require.NoError(t, err) + defer srcNDB.Close() + + stateRoot := commitRoot(ctx, t, srcNDB, testNs, testVersion, node.RootTypeState, map[string]string{ + "state-key": "state-value", + }) + ioRoot := emptyRoot(testNs, testVersion, node.RootTypeIO) + require.NoError(t, srcNDB.Finalize([]node.Root{stateRoot, ioRoot})) + + cpDir := filepath.Join(t.TempDir(), "checkpoint") + require.NoError(t, createCheckpoints(ctx, srcNDB, testNs, testVersion, cpDir)) + + dstNDB, err := newTestNodeDB(t, testNs) + require.NoError(t, err) + defer dstNDB.Close() + + require.NoError(t, importCheckpoints(ctx, dstNDB, testNs, cpDir)) + + roots, err := dstNDB.GetRootsForVersion(testVersion) + require.NoError(t, err) + require.Contains(t, roots, stateRoot) + require.True(t, dstNDB.HasRoot(stateRoot)) + require.True(t, dstNDB.HasRoot(ioRoot)) + }) +} + +func newTestNodeDB(t *testing.T, ns common.Namespace) (dbAPI.NodeDB, error) { + t.Helper() + + return pathbadger.New(&dbAPI.Config{ + DB: filepath.Join(t.TempDir()), + Namespace: ns, + MemoryOnly: true, + }) +} + +func commitRoot( + ctx context.Context, + t *testing.T, + ndb dbAPI.NodeDB, + ns common.Namespace, + version uint64, + rootType node.RootType, + data map[string]string, +) node.Root { + t.Helper() + + tree := mkvs.New(nil, ndb, rootType) + defer tree.Close() + + for k, v := range data { + err := tree.Insert(ctx, []byte(k), []byte(v)) + require.NoError(t, err) + } + + _, rootHash, err := tree.Commit(ctx, ns, version) + require.NoError(t, err) + + return node.Root{ + Namespace: ns, + Version: version, + Type: rootType, + Hash: rootHash, + } +} diff --git a/go/oasis-node/cmd/storage/dbs.go b/go/oasis-node/cmd/storage/dbs.go index 5a7f4f79b3a..95ac5147a98 100644 --- a/go/oasis-node/cmd/storage/dbs.go +++ b/go/oasis-node/cmd/storage/dbs.go @@ -4,8 +4,9 @@ import ( "fmt" "path/filepath" + cometbftDB "github.com/cometbft/cometbft-db" cmtCfg "github.com/cometbft/cometbft/config" - "github.com/cometbft/cometbft/state" + cmtState "github.com/cometbft/cometbft/state" "github.com/cometbft/cometbft/store" "github.com/oasisprotocol/oasis-core/go/common" @@ -43,7 +44,7 @@ func openConsensusNodeDB(dataDir string) (api.NodeDB, func(), error) { return ndb, close, nil } -func openConsensusBlockstore(dataDir string) (*store.BlockStore, error) { +func openConsensusBlockstoreDB(dataDir string) (cometbftDB.DB, error) { cmtConfig := cmtCfg.DefaultConfig() cmtConfig.SetRoot(filepath.Join(dataDir, cmtCommon.StateDir)) @@ -51,17 +52,18 @@ func openConsensusBlockstore(dataDir string) (*store.BlockStore, error) { if err != nil { return nil, fmt.Errorf("failed to obtain db provider: %w", err) } + return cmtDB.OpenBlockstoreDB(dbProvider, cmtConfig) +} - blockstoreDB, err := cmtDB.OpenBlockstoreDB(dbProvider, cmtConfig) +func openConsensusBlockstore(dataDir string) (*store.BlockStore, error) { + blockstoreDB, err := openConsensusBlockstoreDB(dataDir) if err != nil { return nil, fmt.Errorf("failed to open blockstore: %w", err) } - blockstore := store.NewBlockStore(blockstoreDB) - - return blockstore, nil + return store.NewBlockStore(blockstoreDB), nil } -func openConsensusStatestore(dataDir string) (state.Store, error) { +func openConsensusStateDB(dataDir string) (cometbftDB.DB, error) { cmtConfig := cmtCfg.DefaultConfig() cmtConfig.SetRoot(filepath.Join(dataDir, cmtCommon.StateDir)) @@ -69,7 +71,11 @@ func openConsensusStatestore(dataDir string) (state.Store, error) { if err != nil { return nil, fmt.Errorf("failed to obtain db provider: %w", err) } - stateDB, err := cmtDB.OpenStateDB(dbProvider, cmtConfig) + return cmtDB.OpenStateDB(dbProvider, cmtConfig) +} + +func openConsensusStatestore(dataDir string) (cmtState.Store, error) { + stateDB, err := openConsensusStateDB(dataDir) if err != nil { return nil, fmt.Errorf("failed to open state db: %w", err) } diff --git a/go/oasis-node/cmd/storage/storage.go b/go/oasis-node/cmd/storage/storage.go index 71bb69d680a..8e40de32fff 100644 --- a/go/oasis-node/cmd/storage/storage.go +++ b/go/oasis-node/cmd/storage/storage.go @@ -603,5 +603,6 @@ func Register(parentCmd *cobra.Command) { storageCmd.AddCommand(storageCompactCmd) storageCmd.AddCommand(pruneCmd) storageCmd.AddCommand(newInspectCmd()) + storageCmd.AddCommand(newCheckpointCmd()) parentCmd.AddCommand(storageCmd) } diff --git a/go/storage/mkvs/checkpoint/chunk.go b/go/storage/mkvs/checkpoint/chunk.go index dec60bcfb9f..984bff7172e 100644 --- a/go/storage/mkvs/checkpoint/chunk.go +++ b/go/storage/mkvs/checkpoint/chunk.go @@ -262,7 +262,7 @@ func writeChunk(proof *syncer.Proof, w io.Writer) (hash.Hash, error) { return hb.Build(), nil } -func restoreChunk(ctx context.Context, ndb db.NodeDB, chunk *ChunkMetadata, r io.Reader) error { +func RestoreChunk(ctx context.Context, ndb db.NodeDB, chunk *ChunkMetadata, r io.Reader) error { hb := hash.NewBuilder() tr := io.TeeReader(r, hb) sr := snappy.NewReader(tr) diff --git a/go/storage/mkvs/checkpoint/restorer.go b/go/storage/mkvs/checkpoint/restorer.go index e00a21d3ab8..db71d09e6bc 100644 --- a/go/storage/mkvs/checkpoint/restorer.go +++ b/go/storage/mkvs/checkpoint/restorer.go @@ -83,7 +83,7 @@ func (rs *restorer) RestoreChunk(ctx context.Context, idx uint64, r io.Reader) ( return false, err } - err = restoreChunk(ctx, rs.ndb, chunk, r) + err = RestoreChunk(ctx, rs.ndb, chunk, r) switch { case err == nil: case errors.Is(err, ErrChunkProofVerificationFailed): From 44005e24c8aa88e7c67688a262db660e856fc70b Mon Sep 17 00:00:00 2001 From: Martin Tomazic Date: Sat, 21 Feb 2026 15:58:49 +0100 Subject: [PATCH 2/5] go/oasis-test-runner/e2e: Add create import checkpoint test The test should be ideally hardened by also making sure the target node also syncs up to the tip of the runtime chain and not just consensus. --- .../e2e/runtime/checkpoint_create_import.go | 149 ++++++++++++++++++ .../scenario/e2e/runtime/scenario.go | 2 + 2 files changed, 151 insertions(+) create mode 100644 go/oasis-test-runner/scenario/e2e/runtime/checkpoint_create_import.go diff --git a/go/oasis-test-runner/scenario/e2e/runtime/checkpoint_create_import.go b/go/oasis-test-runner/scenario/e2e/runtime/checkpoint_create_import.go new file mode 100644 index 00000000000..705585eedd6 --- /dev/null +++ b/go/oasis-test-runner/scenario/e2e/runtime/checkpoint_create_import.go @@ -0,0 +1,149 @@ +package runtime + +import ( + "context" + "fmt" + "path/filepath" + + consensus "github.com/oasisprotocol/oasis-core/go/consensus/api" + "github.com/oasisprotocol/oasis-core/go/oasis-test-runner/env" + "github.com/oasisprotocol/oasis-core/go/oasis-test-runner/oasis" + "github.com/oasisprotocol/oasis-core/go/oasis-test-runner/oasis/cli" + "github.com/oasisprotocol/oasis-core/go/oasis-test-runner/scenario" + roothash "github.com/oasisprotocol/oasis-core/go/roothash/api" +) + +// CheckpointCreateImport is the checkpoint create/import e2e scenario. +var CheckpointCreateImport scenario.Scenario = newCheckpointCreateImportImpl() + +type checkpointCreateImportImpl struct { + Scenario +} + +func newCheckpointCreateImportImpl() scenario.Scenario { + return &checkpointCreateImportImpl{ + Scenario: *NewScenario( + "checkpoint-create-import", + NewTestClient().WithScenario(SimpleScenario), + ), + } +} + +func (sc *checkpointCreateImportImpl) Clone() scenario.Scenario { + return &checkpointCreateImportImpl{ + Scenario: *sc.Scenario.Clone().(*Scenario), + } +} + +func (sc *checkpointCreateImportImpl) Fixture() (*oasis.NetworkFixture, error) { + return sc.Scenario.Fixture() +} + +func (sc *checkpointCreateImportImpl) Run(ctx context.Context, childEnv *env.Env) error { + // Start the network and run the test client to populate state. + if err := sc.StartNetworkAndTestClient(ctx, childEnv); err != nil { + return err + } + if err := sc.WaitTestClient(); err != nil { + return err + } + + // Use height - 3 so that blocks at h, h+1, h+2 all exist in the block store. + blk, err := sc.Net.Controller().Consensus.GetBlock(ctx, consensus.HeightLatest) + if err != nil { + return fmt.Errorf("failed to get latest consensus block: %w", err) + } + height := blk.Height - 3 + + rtBlk, err := sc.Net.ClientController().Roothash.GetLatestBlock(ctx, &roothash.RuntimeRequest{ + RuntimeID: KeyValueRuntimeID, + Height: height, + }) + if err != nil { + return fmt.Errorf("failed to get runtime block at height %d: %w", height, err) + } + round := rtBlk.Header.Round + + sc.Logger.Info("creating checkpoints", + "height", height, + "round", round, + "runtime_id", KeyValueRuntimeID, + ) + + cpDir := filepath.Join(childEnv.Dir(), "checkpoint") + + // Stop compute worker 0 (source node). + source := sc.Net.ComputeWorkers()[0] + if err := source.StopGracefully(); err != nil { + return fmt.Errorf("failed to stop source compute worker: %w", err) + } + + // Create checkpoints from the source node's data. + args := []string{ + "storage", "checkpoint", "create", + "--config", source.ConfigFile(), + "--height", fmt.Sprintf("%d", height), + "--runtime", KeyValueRuntimeID.Hex(), + "--round", fmt.Sprintf("%d", round), + "--output-dir", cpDir, + "--debug.dont_blame_oasis", + "--debug.allow_test_keys", + } + if err := cli.RunSubCommand(childEnv, sc.Logger, "checkpoint-create", sc.Net.Config().NodeBinary, args); err != nil { + return fmt.Errorf("failed to create checkpoints: %w", err) + } + + sc.Logger.Info("checkpoints created successfully") + + // Start the source compute worker again. + if err := source.Start(); err != nil { + return fmt.Errorf("failed to restart source compute worker: %w", err) + } + + // Stop compute worker 2 (target node). + target := sc.Net.ComputeWorkers()[2] + if err := target.StopGracefully(); err != nil { + return fmt.Errorf("failed to stop target compute worker: %w", err) + } + + // Reset the target node's state completely. Ideally we would use NoAutoStart, + // however if we do so no config is created until the node is started and import + // command therefore fails. + sc.Logger.Info("resetting target node state") + cliHelpers := cli.New(childEnv, sc.Net, sc.Logger) + if err := cliHelpers.UnsafeReset(target.DataDir(), false, false, true); err != nil { + return fmt.Errorf("failed to reset target node state: %w", err) + } + + // Import checkpoints into the target node. + sc.Logger.Info("importing checkpoints into target node") + args = []string{ + "storage", "checkpoint", "import", + "--config", target.ConfigFile(), + "--input-dir", cpDir, + "--debug.dont_blame_oasis", + "--debug.allow_test_keys", + } + if err := cli.RunSubCommand(childEnv, sc.Logger, "checkpoint-import", sc.Net.Config().NodeBinary, args); err != nil { + return fmt.Errorf("failed to import checkpoints: %w", err) + } + + sc.Logger.Info("checkpoints imported, starting target node") + + // Start the target node. + if err := target.Start(); err != nil { + return fmt.Errorf("failed to start target node: %w", err) + } + + // Wait for the target node to sync. + sc.Logger.Info("waiting for target node to sync") + ctrl, err := oasis.NewController(target.SocketPath()) + if err != nil { + return fmt.Errorf("failed to create controller for target node: %w", err) + } + if err := ctrl.WaitReady(ctx); err != nil { + return fmt.Errorf("target node failed to sync: %w", err) + } + + return nil +} diff --git a/go/oasis-test-runner/scenario/e2e/runtime/scenario.go b/go/oasis-test-runner/scenario/e2e/runtime/scenario.go index db4d8143004..317173173fd 100644 --- a/go/oasis-test-runner/scenario/e2e/runtime/scenario.go +++ b/go/oasis-test-runner/scenario/e2e/runtime/scenario.go @@ -347,6 +347,8 @@ func RegisterScenarios() error { StorageSyncFromRegistered, StorageSyncInconsistent, StorageEarlyStateSync, + // Checkpoint create/import test. + CheckpointCreateImport, // Sentry test. Sentry, // Keymanager tests. From 7a3bee8091774a7485cfc0d39ab4f6aaf77647f9 Mon Sep 17 00:00:00 2001 From: Martin Tomazic Date: Tue, 31 Mar 2026 11:07:16 +0200 Subject: [PATCH 3/5] go/oasis-test-runner: Add wait methods to controller --- go/oasis-test-runner/oasis/controller.go | 50 ++++++++++++++++++++++++ 1 file changed, 50 insertions(+) diff --git a/go/oasis-test-runner/oasis/controller.go b/go/oasis-test-runner/oasis/controller.go index 51f9000fb3d..fd2a4a68336 100644 --- a/go/oasis-test-runner/oasis/controller.go +++ b/go/oasis-test-runner/oasis/controller.go @@ -1,10 +1,14 @@ package oasis import ( + "context" + "fmt" + "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" beacon "github.com/oasisprotocol/oasis-core/go/beacon/api" + "github.com/oasisprotocol/oasis-core/go/common" cmnGrpc "github.com/oasisprotocol/oasis-core/go/common/grpc" consensus "github.com/oasisprotocol/oasis-core/go/consensus/api" control "github.com/oasisprotocol/oasis-core/go/control/api" @@ -46,6 +50,52 @@ func (c *Controller) Close() { c.conn.Close() } +// WaitConsensusHeight waits until the controller observes at least specified consensus height. +func (c *Controller) WaitConsensusHeight(ctx context.Context, height int64) error { + blkCh, sub, err := c.Consensus.WatchBlocks(ctx) + if err != nil { + return fmt.Errorf("failed to watch consensus blocks: %w", err) + } + defer sub.Close() + + for { + select { + case blk, ok := <-blkCh: + if !ok { + return fmt.Errorf("consensus block channel closed") + } + if blk.Height >= height { + return nil + } + case <-ctx.Done(): + return ctx.Err() + } + } +} + +// WaitRuntimeRound waits until the controller observes at least specified runtime round. +func (c *Controller) WaitRuntimeRound(ctx context.Context, runtimeID common.Namespace, round uint64) error { + blkCh, sub, err := c.RuntimeClient.WatchBlocks(ctx, runtimeID) + if err != nil { + return fmt.Errorf("failed to watch runtime blocks: %w", err) + } + defer sub.Close() + + for { + select { + case annBlk, ok := <-blkCh: + if !ok { + return fmt.Errorf("runtime block channel closed") + } + if annBlk.Block.Header.Round >= round { + return nil + } + case <-ctx.Done(): + return ctx.Err() + } + } +} + // NewController creates a new node controller given the path to // a node's internal socket. func NewController(socketPath string) (*Controller, error) { From 296298bf10cda1be4b9d23618e86fbbab65667d5 Mon Sep 17 00:00:00 2001 From: Martin Tomazic Date: Tue, 31 Mar 2026 11:19:25 +0200 Subject: [PATCH 4/5] Fixup: Fix flaky test and ensure sync from imported checkpoint Prevously compute may not sync the latest round and checkpoint might fail. This is fixed by querying compute explicitly. --- .../e2e/runtime/checkpoint_create_import.go | 93 ++++++++++++++----- 1 file changed, 70 insertions(+), 23 deletions(-) diff --git a/go/oasis-test-runner/scenario/e2e/runtime/checkpoint_create_import.go b/go/oasis-test-runner/scenario/e2e/runtime/checkpoint_create_import.go index 705585eedd6..3eea25fc9dd 100644 --- a/go/oasis-test-runner/scenario/e2e/runtime/checkpoint_create_import.go +++ b/go/oasis-test-runner/scenario/e2e/runtime/checkpoint_create_import.go @@ -48,43 +48,57 @@ func (sc *checkpointCreateImportImpl) Run(ctx context.Context, childEnv *env.Env return err } + src := sc.Net.ComputeWorkers()[0] + srcCtrl, err := oasis.NewController(src.SocketPath()) + if err != nil { + return fmt.Errorf("failed to create controller for the source node: %w", err) + } + // Use height - 3 so that blocks at h, h+1, h+2 all exist in the block store. - blk, err := sc.Net.Controller().Consensus.GetBlock(ctx, consensus.HeightLatest) + blk, err := srcCtrl.Consensus.GetBlock(ctx, consensus.HeightLatest) if err != nil { return fmt.Errorf("failed to get latest consensus block: %w", err) } - height := blk.Height - 3 - - rtBlk, err := sc.Net.ClientController().Roothash.GetLatestBlock(ctx, &roothash.RuntimeRequest{ + candidateHeight := blk.Height - 3 + rtState, err := srcCtrl.Roothash.GetRuntimeState(ctx, &roothash.RuntimeRequest{ RuntimeID: KeyValueRuntimeID, - Height: height, + Height: candidateHeight, }) if err != nil { - return fmt.Errorf("failed to get runtime block at height %d: %w", height, err) + return fmt.Errorf("failed to get runtime state for height %d: %w", candidateHeight, err) } - round := rtBlk.Header.Round - sc.Logger.Info("creating checkpoints", - "height", height, - "round", round, - "runtime_id", KeyValueRuntimeID, - ) + // Pick runtime state's LastBlockHeight as the consensus checkpoint height else + // runtime light history indexer might miss authoritative light block for the + // corresponding runtime round. + cpRound := rtState.LastBlock.Header.Round + cpHeight := rtState.LastBlockHeight - cpDir := filepath.Join(childEnv.Dir(), "checkpoint") + // Ensure runtime round is synced before stopping the node and creating a checkpoint for it. + if err := srcCtrl.WaitRuntimeRound(ctx, KeyValueRuntimeID, cpRound); err != nil { + return fmt.Errorf("waiting runtime round %d: %w", cpRound, err) + } // Stop compute worker 0 (source node). - source := sc.Net.ComputeWorkers()[0] - if err := source.StopGracefully(); err != nil { + if err := src.StopGracefully(); err != nil { return fmt.Errorf("failed to stop source compute worker: %w", err) } // Create checkpoints from the source node's data. + cpDir := filepath.Join(childEnv.Dir(), "checkpoint") + + sc.Logger.Info("creating checkpoints", + "height", cpHeight, + "round", cpRound, + "runtime_id", KeyValueRuntimeID, + ) + args := []string{ "storage", "checkpoint", "create", - "--config", source.ConfigFile(), - "--height", fmt.Sprintf("%d", height), + "--config", src.ConfigFile(), + "--height", fmt.Sprintf("%d", cpHeight), "--runtime", KeyValueRuntimeID.Hex(), - "--round", fmt.Sprintf("%d", round), + "--round", fmt.Sprintf("%d", cpRound), "--output-dir", cpDir, "--debug.dont_blame_oasis", "--debug.allow_test_keys", @@ -96,7 +110,7 @@ func (sc *checkpointCreateImportImpl) Run(ctx context.Context, childEnv *env.Env sc.Logger.Info("checkpoints created successfully") // Start the source compute worker again. - if err := source.Start(); err != nil { + if err := src.Start(); err != nil { return fmt.Errorf("failed to restart source compute worker: %w", err) } @@ -135,15 +149,48 @@ func (sc *checkpointCreateImportImpl) Run(ctx context.Context, childEnv *env.Env return fmt.Errorf("failed to start target node: %w", err) } - // Wait for the target node to sync. - sc.Logger.Info("waiting for target node to sync") - ctrl, err := oasis.NewController(target.SocketPath()) + targetCtrl, err := oasis.NewController(target.SocketPath()) if err != nil { return fmt.Errorf("failed to create controller for target node: %w", err) } - if err := ctrl.WaitReady(ctx); err != nil { + + // Ensure target node syncs to the tip of the chain from the imported checkpoints. + sc.Logger.Info("waiting for target node to sync") + if err := targetCtrl.WaitReady(ctx); err != nil { return fmt.Errorf("target node failed to sync: %w", err) } + sc.Logger.Info("target node is ready") + + // Manually ensure that runtime state was synced up to the latest round as + // WaitReady only guarantees consensus sync. + latestBlk, err := sc.Net.ClientController().Roothash.GetLatestBlock(ctx, &roothash.RuntimeRequest{ + RuntimeID: KeyValueRuntimeID, + Height: consensus.HeightLatest, + }) + if err != nil { + return fmt.Errorf("failed to get latest runtime block: %w", err) + } + latestRound := latestBlk.Header.Round + sc.Logger.Info("waiting the target node to have runtime state synced") + if err := targetCtrl.WaitRuntimeRound(ctx, KeyValueRuntimeID, latestRound); err != nil { + return fmt.Errorf("waiting synced runtime round %d: %w", latestRound, err) + } + sc.Logger.Info("target node has runtime state synced") + + // Ensure target synced from the imported checkpoint and not from the genesis. + status, err := targetCtrl.NodeController.GetStatus(ctx) + if err != nil { + return fmt.Errorf("failed to get target node status: %w", err) + } + if lastRetainedHeight := status.Consensus.LastRetainedHeight; lastRetainedHeight != cpHeight { + sc.Logger.Info("last retained height is not equal to the imported checkpoint height", + "cp_height", cpHeight, + "last_retained_height", lastRetainedHeight) + return fmt.Errorf("failed to ensure consensus synced from the imported checkpoint") + } + // No need to assert target node didn't sync runtime state from the genesis, + // since runtime genesis sync cannot succeed with a missing runtime light history + // (the case when consensus is synced using an imported checkpoint, asserted above). return nil } From 194a24fbb2e116b64db0159117e43ddd01663bf4 Mon Sep 17 00:00:00 2001 From: Martin Tomazic Date: Tue, 31 Mar 2026 20:57:47 +0200 Subject: [PATCH 5/5] Fixup: Make start and abort multipart insert version scoped --- go/oasis-node/cmd/storage/checkpoint.go | 21 ++++++++++++--------- 1 file changed, 12 insertions(+), 9 deletions(-) diff --git a/go/oasis-node/cmd/storage/checkpoint.go b/go/oasis-node/cmd/storage/checkpoint.go index fb19cf155e7..eb79632661d 100644 --- a/go/oasis-node/cmd/storage/checkpoint.go +++ b/go/oasis-node/cmd/storage/checkpoint.go @@ -300,6 +300,18 @@ func importCheckpoints(ctx context.Context, ndb api.NodeDB, ns common.Namespace, return fmt.Errorf("failed to read checkpoints: %w", err) } + if len(cps) == 0 { + return fmt.Errorf("no checkpoints found") + } + + if err := ndb.StartMultipartInsert(cps[0].Root.Version); err != nil { + return fmt.Errorf("failed to start multipart insert: %w", err) + } + defer func() { + if err := ndb.AbortMultipartInsert(); err != nil { + logger.Error("failed to abort multi-part insert", "err", err) + } + }() var roots []node.Root for _, cp := range cps { if err := importCp(ctx, ndb, provider, cp); err != nil { @@ -315,15 +327,6 @@ func importCheckpoints(ctx context.Context, ndb api.NodeDB, ns common.Namespace, } func importCp(ctx context.Context, ndb api.NodeDB, provider checkpoint.Creator, cp *checkpoint.Metadata) error { - if err := ndb.StartMultipartInsert(cp.Root.Version); err != nil { - return fmt.Errorf("failed to start multipart insert: %w", err) - } - defer func() { - if err := ndb.AbortMultipartInsert(); err != nil { - logger.Error("failed to abort multi-part insert", "err", err) - } - }() - for idx := range cp.Chunks { chunk, err := cp.GetChunkMetadata(uint64(idx)) if err != nil {