From ccdfa7e5ac1078e5e251ae5b3ffd544a5249ab1e Mon Sep 17 00:00:00 2001 From: "Leo Zhang (zhangchiqing)" Date: Mon, 6 Jul 2026 15:52:27 -0700 Subject: [PATCH] add util to extract payloadless --- .../cmd.go | 119 ++++++++++++++++++ cmd/util/cmd/root.go | 2 + cmd/util/ledger/util/state.go | 69 ++++++++++ .../complete/wal/payloadless_replay_test.go | 106 ++++++++++++++++ ledger/complete/wal/wal.go | 91 ++++++++++++++ 5 files changed, 387 insertions(+) create mode 100644 cmd/util/cmd/execution-state-extract-payloadless/cmd.go diff --git a/cmd/util/cmd/execution-state-extract-payloadless/cmd.go b/cmd/util/cmd/execution-state-extract-payloadless/cmd.go new file mode 100644 index 00000000000..b3f50cda8b6 --- /dev/null +++ b/cmd/util/cmd/execution-state-extract-payloadless/cmd.go @@ -0,0 +1,119 @@ +package extractpayloadless + +import ( + "encoding/hex" + "fmt" + "os" + "path" + + "github.com/rs/zerolog/log" + "github.com/spf13/cobra" + + "github.com/onflow/flow-go/cmd/util/ledger/util" + "github.com/onflow/flow-go/ledger" + "github.com/onflow/flow-go/ledger/complete/payloadless" + "github.com/onflow/flow-go/ledger/complete/wal" + "github.com/onflow/flow-go/model/bootstrap" + "github.com/onflow/flow-go/model/flow" +) + +var ( + flagExecutionStateDir string + flagOutputDir string + flagStateCommitment string + flagNWorker uint + flagMTrieCacheSize uint32 +) + +// Cmd extracts the payloadless (V7) trie at a given state commitment from a WAL directory and writes +// it as a single-trie V7 root checkpoint. It is the payloadless counterpart of +// execution-state-extract: no migration is performed and no payloads are read, because a payloadless +// trie stores only leaf hashes. +var Cmd = &cobra.Command{ + Use: "execution-state-extract-payloadless", + Short: "Extract a payloadless (V7) trie at a state commitment into a V7 root checkpoint", + Long: `Extract the payloadless (V7) trie at a given state commitment and write it as a V7 root checkpoint. + +The trie is loaded from the WAL directory (--execution-state-dir), recovering in-memory state from the +latest V7 checkpoint plus any newer WAL segments, exactly like the node does at startup. The trie whose +root hash matches --state-commitment is written to --output-dir as a single-trie V7 root checkpoint +("` + bootstrap.FilenameWALRootCheckpoint + wal.V7FileSuffix + `"). + +Because a payloadless trie carries only leaf hashes and no payloads, no migration is possible or needed; +this command only re-checkpoints the selected trie. It acquires an exclusive lock on the WAL directory, +so it must be run against a stopped node's data directory.`, + RunE: runE, +} + +func init() { + Cmd.Flags().StringVar(&flagExecutionStateDir, "execution-state-dir", "", + "Execution Node state dir (where the V7 checkpoint and WAL logs are written)") + _ = Cmd.MarkFlagRequired("execution-state-dir") + + Cmd.Flags().StringVar(&flagOutputDir, "output-dir", "", + "Directory to write the V7 root checkpoint to") + _ = Cmd.MarkFlagRequired("output-dir") + + Cmd.Flags().StringVar(&flagStateCommitment, "state-commitment", "", + "state commitment of the trie to extract (hex-encoded, 64 characters)") + _ = Cmd.MarkFlagRequired("state-commitment") + + Cmd.Flags().UintVar(&flagNWorker, "nworker", 16, + "number of subtrie files to encode in parallel (valid range [1, 16])") + + Cmd.Flags().Uint32Var(&flagMTrieCacheSize, "mtrie-cache-size", ledger.DefaultMTrieCacheSize, + "number of tries retained in the forest during WAL replay; match the node's --mtrie-cache-size. "+ + "This is the main driver of peak memory; lower it to reduce memory (at the risk of failing to "+ + "resolve tries across WAL forks)") +} + +func runE(*cobra.Command, []string) error { + stateCommitmentBytes, err := hex.DecodeString(flagStateCommitment) + if err != nil { + return fmt.Errorf("cannot decode state commitment: %w", err) + } + stateCommitment, err := flow.ToStateCommitment(stateCommitmentBytes) + if err != nil { + return fmt.Errorf("invalid state commitment length: %w", err) + } + + outputFile := bootstrap.FilenameWALRootCheckpoint + wal.V7FileSuffix + + log.Info(). + Str("execution-state-dir", flagExecutionStateDir). + Str("output-dir", flagOutputDir). + Str("state-commitment", stateCommitment.String()). + Str("output", path.Join(flagOutputDir, outputFile)). + Msg("extracting payloadless (V7) trie at state commitment") + + if err := os.MkdirAll(flagOutputDir, 0755); err != nil { + return fmt.Errorf("cannot create output directory %s: %w", flagOutputDir, err) + } + + trie, err := util.ReadPayloadlessTrie(flagExecutionStateDir, stateCommitment, int(flagMTrieCacheSize)) + if err != nil { + return fmt.Errorf("cannot read payloadless trie for state commitment %s: %w", stateCommitment, err) + } + + log.Info(). + Str("root_hash", trie.RootHash().String()). + Uint64("allocated_reg_count", trie.AllocatedRegCount()). + Msg("loaded payloadless trie, storing V7 root checkpoint") + + err = wal.StoreCheckpointV7( + []*payloadless.MTrie{trie}, + flagOutputDir, + outputFile, + log.Logger, + flagNWorker, + ) + if err != nil { + return fmt.Errorf("cannot store V7 root checkpoint: %w", err) + } + + log.Info(). + Str("state-commitment", ledger.State(trie.RootHash()).String()). + Str("output", path.Join(flagOutputDir, outputFile)). + Msg("✅ payloadless (V7) state extraction completed successfully") + return nil +} diff --git a/cmd/util/cmd/root.go b/cmd/util/cmd/root.go index 5514af7197e..32899b990de 100644 --- a/cmd/util/cmd/root.go +++ b/cmd/util/cmd/root.go @@ -29,6 +29,7 @@ import ( export "github.com/onflow/flow-go/cmd/util/cmd/exec-data-json-export" edbs "github.com/onflow/flow-go/cmd/util/cmd/execution-data-blobstore/cmd" extract "github.com/onflow/flow-go/cmd/util/cmd/execution-state-extract" + extractpayloadless "github.com/onflow/flow-go/cmd/util/cmd/execution-state-extract-payloadless" evm_state_exporter "github.com/onflow/flow-go/cmd/util/cmd/export-evm-state" ledger_json_exporter "github.com/onflow/flow-go/cmd/util/cmd/export-json-execution-state" export_json_transactions "github.com/onflow/flow-go/cmd/util/cmd/export-json-transactions" @@ -106,6 +107,7 @@ func init() { func addCommands() { rootCmd.AddCommand(version.Cmd) rootCmd.AddCommand(extract.Cmd) + rootCmd.AddCommand(extractpayloadless.Cmd) rootCmd.AddCommand(export.Cmd) rootCmd.AddCommand(checkpoint_list_tries.Cmd) rootCmd.AddCommand(checkpoint_collect_stats.Cmd) diff --git a/cmd/util/ledger/util/state.go b/cmd/util/ledger/util/state.go index 7ef36270040..0a0112639d1 100644 --- a/cmd/util/ledger/util/state.go +++ b/cmd/util/ledger/util/state.go @@ -12,6 +12,7 @@ import ( "github.com/onflow/flow-go/ledger/common/pathfinder" "github.com/onflow/flow-go/ledger/complete" mtrie "github.com/onflow/flow-go/ledger/complete/mtrie/trie" + "github.com/onflow/flow-go/ledger/complete/payloadless" "github.com/onflow/flow-go/ledger/complete/wal" "github.com/onflow/flow-go/model/flow" "github.com/onflow/flow-go/module/metrics" @@ -96,6 +97,74 @@ func ReadTrie(dir string, targetHash flow.StateCommitment) (*mtrie.MTrie, error) return trie, nil } +// ReadPayloadlessTrie loads the payloadless (V7) trie at the given state commitment from the WAL +// directory, recovering in-memory state from the latest V7 checkpoint plus any newer WAL segments. +// It is the payloadless counterpart of [ReadTrie]: the returned trie's leaves carry a 32-byte leaf +// hash, not a full payload. +// +// `capacity` bounds the number of tries retained in the forest during replay (the peak-memory +// driver). It should match the node's `--mtrie-cache-size` ([ledger.DefaultMTrieCacheSize]) so this +// tool's memory footprint matches a node booting at the same state; a smaller value trades safety +// against WAL forks for lower memory. +// +// WAL replay stops as soon as the target trie is produced (see +// [wal.DiskWAL.ReplayOnPayloadlessForestUntil]), so it does not read segments past the target. +// This is what lets an older state commitment be extracted at all: replaying to the WAL tip would +// evict the target from the LRU-bounded forest before it could be read. +// +// This is a read-only load: no checkpoint is written and no compactor is started. The exclusive WAL +// directory lock acquired on open is released before this returns, and only the returned trie's +// reachable nodes stay resident for any downstream checkpoint writing. +// +// No error returns are expected during normal operation. +func ReadPayloadlessTrie(dir string, targetHash flow.StateCommitment, capacity int) (*payloadless.MTrie, error) { + log.Info().Msg("init WAL") + + diskWal, err := wal.NewDiskWAL( + log.Logger, + nil, + metrics.NewNoopCollector(), + dir, + capacity, + pathfinder.PathByteSize, + wal.SegmentSize, + ) + if err != nil { + return nil, fmt.Errorf("cannot create disk WAL: %w", err) + } + + // Done closes the WAL and releases the exclusive directory lock. + defer func() { + <-diskWal.Done() + }() + + forest, err := payloadless.NewForest(capacity, metrics.NewNoopCollector(), nil) + if err != nil { + return nil, fmt.Errorf("cannot create payloadless forest: %w", err) + } + + targetRootHash := ledger.RootHash(targetHash) + + log.Info().Msg("loading V7 checkpoint and replaying WAL until the target trie is found") + + found, err := diskWal.ReplayOnPayloadlessForestUntil(forest, targetRootHash) + if err != nil { + return nil, fmt.Errorf("cannot replay payloadless WAL: %w", err) + } + if !found { + return nil, fmt.Errorf( + "no payloadless trie with state commitment %x was found in %s; check the --state-commitment and --execution-state-dir flags", + targetHash[:], dir) + } + + trie, err := forest.GetTrie(targetRootHash) + if err != nil { + return nil, fmt.Errorf("cannot get payloadless trie at state commitment %x: %w", targetHash[:], err) + } + + return trie, nil +} + func ReadTrieForPayloads(dir string, targetHash flow.StateCommitment) ([]*ledger.Payload, error) { trie, err := ReadTrie(dir, targetHash) if err != nil { diff --git a/ledger/complete/wal/payloadless_replay_test.go b/ledger/complete/wal/payloadless_replay_test.go index 9be58d6f7ff..0fef933c1d4 100644 --- a/ledger/complete/wal/payloadless_replay_test.go +++ b/ledger/complete/wal/payloadless_replay_test.go @@ -119,3 +119,109 @@ func TestReplayOnPayloadlessForest_ReplaysWALSegments(t *testing.T) { require.True(t, forest.HasTrie(root1), "forest must contain the root produced by replaying the WAL segment") }) } + +// TestReplayOnPayloadlessForestUntil verifies the early-stop replay: it stops as +// soon as the target trie is produced (or is already in the V7 checkpoint), and +// reports whether the target was found. Stopping early is what lets an older +// state commitment be extracted without being evicted from the LRU forest by a +// full replay to the WAL tip. +func TestReplayOnPayloadlessForestUntil(t *testing.T) { + unittest.RunWithTempDir(t, func(dir string) { + logger := zerolog.Nop() + + // Seed state (root0) captured as the V7 root checkpoint. + fullForest, err := mtrie.NewForest(100, &metrics.NoopCollector{}, nil) + require.NoError(t, err) + + paths0, payloads0 := randNPathPayloads(10) + root0, err := fullForest.Update(&ledger.TrieUpdate{ + RootHash: fullForest.GetEmptyRootHash(), + Paths: paths0, + Payloads: toPayloadPtrs(payloads0), + }) + require.NoError(t, err) + + v6Tries, err := fullForest.GetTries() + require.NoError(t, err) + v7Tries, err := FromV6Tries(v6Tries) + require.NoError(t, err) + require.NoError(t, StoreCheckpointV7Concurrently(v7Tries, dir, RootCheckpointFilenameV7(), logger)) + + // Three chained updates recorded into the WAL (but NOT the checkpoint): + // root0 -> root1 -> root2 -> root3. + recordWAL, err := NewDiskWAL(logger, nil, metrics.NewNoopCollector(), dir, 10, pathByteSize, segmentSize) + require.NoError(t, err) + + parent := root0 + roots := make([]ledger.RootHash, 0, 3) + for i := 0; i < 3; i++ { + pathsi, payloadsi := randNPathPayloads(10) + update := &ledger.TrieUpdate{ + RootHash: parent, + Paths: pathsi, + Payloads: toPayloadPtrs(payloadsi), + } + root, err := fullForest.Update(update) + require.NoError(t, err) + _, _, err = recordWAL.RecordUpdate(update) + require.NoError(t, err) + roots = append(roots, root) + parent = root + } + <-recordWAL.Done() + root1, root2, root3 := roots[0], roots[1], roots[2] + + // A fourth update built on root3 but never recorded: a valid root hash that + // is present neither in the checkpoint nor in the WAL. + pathsAbsent, payloadsAbsent := randNPathPayloads(10) + rootAbsent, err := fullForest.Update(&ledger.TrieUpdate{ + RootHash: root3, + Paths: pathsAbsent, + Payloads: toPayloadPtrs(payloadsAbsent), + }) + require.NoError(t, err) + + // replayUntil runs a fresh DiskWAL + forest and returns the found flag plus + // the populated forest, so each case is independent. + replayUntil := func(t *testing.T, target ledger.RootHash) (bool, *payloadless.Forest) { + w, err := NewDiskWAL(logger, nil, metrics.NewNoopCollector(), dir, 10, pathByteSize, segmentSize) + require.NoError(t, err) + t.Cleanup(func() { <-w.Done() }) + + forest, err := payloadless.NewForest(100, &metrics.NoopCollector{}, nil) + require.NoError(t, err) + + found, err := w.ReplayOnPayloadlessForestUntil(forest, target) + require.NoError(t, err) + return found, forest + } + + t.Run("stops at a mid-WAL target", func(t *testing.T) { + found, forest := replayUntil(t, root1) + require.True(t, found, "root1 is produced by replaying the first WAL update") + require.True(t, forest.HasTrie(root1), "target trie must be present") + // Proof of early-stop: updates producing root2/root3 must NOT be applied. + require.False(t, forest.HasTrie(root2), "replay must stop at the target, before producing root2") + require.False(t, forest.HasTrie(root3), "replay must stop at the target, before producing root3") + }) + + t.Run("target already in checkpoint replays no segments", func(t *testing.T) { + found, forest := replayUntil(t, root0) + require.True(t, found, "root0 is a checkpoint trie") + require.True(t, forest.HasTrie(root0)) + require.False(t, forest.HasTrie(root1), "no WAL segment should be replayed when the target is in the checkpoint") + }) + + t.Run("target reachable only at the WAL tip", func(t *testing.T) { + found, forest := replayUntil(t, root3) + require.True(t, found, "root3 is produced by replaying all recorded WAL updates") + require.True(t, forest.HasTrie(root3)) + }) + + t.Run("absent target returns not found without error", func(t *testing.T) { + found, forest := replayUntil(t, rootAbsent) + require.False(t, found, "rootAbsent is present neither in the checkpoint nor the WAL") + require.False(t, forest.HasTrie(rootAbsent)) + }) + }) +} diff --git a/ledger/complete/wal/wal.go b/ledger/complete/wal/wal.go index 1c3137fa107..0ef50105b22 100644 --- a/ledger/complete/wal/wal.go +++ b/ledger/complete/wal/wal.go @@ -1,6 +1,7 @@ package wal import ( + "errors" "fmt" "sort" @@ -234,6 +235,96 @@ func (w *DiskWAL) replaySegmentsForPayloadlessForest( return nil } +// errStopPayloadlessReplay is a sentinel used to break out of segment replay in +// [DiskWAL.ReplayOnPayloadlessForestUntil] once the target trie has been +// produced. It never escapes that method. +var errStopPayloadlessReplay = errors.New("target payloadless trie found; stopping replay") + +// ReplayOnPayloadlessForestUntil reconstructs payloadless state like +// [DiskWAL.ReplayOnPayloadlessForest], but stops replaying WAL segments as soon +// as an update produces a trie whose root hash equals `targetRootHash` (or the +// target is already one of the loaded V7 checkpoint tries). +// +// Stopping early bounds both time and memory to the segments up to the target. +// This also avoids a correctness pitfall of replaying to the end: the forest is +// LRU-bounded, so a target more than `capacity` tries before the WAL tip would +// be evicted before it could be read. +// +// It returns true when the target trie is present after loading the V7 +// checkpoint or during segment replay, and false when all segments were replayed +// without producing it. The caller reads the trie back via [payloadless.Forest.GetTrie]. +// +// Expected error returns during normal operation: +// - error containing "no V7 checkpoint found": when the WAL directory contains +// no V7 checkpoint of either kind, so the forest cannot be seeded. +func (w *DiskWAL) ReplayOnPayloadlessForestUntil( + forest *payloadless.Forest, + targetRootHash ledger.RootHash, +) (bool, error) { + checkpointer, err := w.NewCheckpointer() + if err != nil { + return false, fmt.Errorf("cannot create checkpointer: %w", err) + } + + tries, loadedCheckpoint, err := checkpointer.LoadLatestCheckpointV7() + if err != nil { + return false, fmt.Errorf("cannot load latest V7 checkpoint: %w", err) + } + + // Mirrors [DiskWAL.ReplayOnPayloadlessForest]: a payloadless forest cannot be + // seeded by WAL replay alone, so a V7 checkpoint of either kind is required. + if loadedCheckpoint < 0 && len(tries) == 0 { + return false, fmt.Errorf( + "no V7 checkpoint found in %s; a V7 checkpoint is required to start a payloadless ledger", + w.wal.Dir(), + ) + } + + if err := forest.AddTries(tries); err != nil { + return false, fmt.Errorf("failed to seed payloadless forest from V7 checkpoint: %w", err) + } + + // The target may already be one of the checkpoint tries; if so, no segment + // replay is needed. + if forest.HasTrie(targetRootHash) { + return true, nil + } + + firstSeg, lastSeg, err := w.Segments() + if err != nil { + return false, fmt.Errorf("could not find segments: %w", err) + } + from := firstSeg + if loadedCheckpoint >= from { + from = loadedCheckpoint + 1 + } + if from > lastSeg { + // V7 checkpoint already covers everything on disk and did not contain the target. + return false, nil + } + + found := false + err = w.replaySegments(from, lastSeg, + func(update *ledger.TrieUpdate) error { + rootHash, err := forest.Update(update) + if err != nil { + return err + } + if rootHash.Equals(targetRootHash) { + found = true + return errStopPayloadlessReplay + } + return nil + }, + func(rootHash ledger.RootHash) error { return nil }, + ) + if err != nil && !errors.Is(err, errStopPayloadlessReplay) { + return false, fmt.Errorf("could not replay WAL segments [%v:%v] for payloadless forest: %w", from, lastSeg, err) + } + + return found, nil +} + func (w *DiskWAL) Segments() (first, last int, err error) { return prometheusWAL.Segments(w.wal.Dir()) }