[Storehouse] 015 - add util to extract payloadless

cmd/util/cmd/execution-state-extract-payloadless/cmd.go

@@ -0,0 +1,160 @@
+package extractpayloadless
+
+import (
+	"encoding/hex"
+	"fmt"
+	"os"
+	"path"
+	"path/filepath"
+
+	"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
+	flagUseWalSegmentNumber bool
+)
+
+// 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 + `"). With --use-wal-segment-number, the
+output file is instead named after the WAL segment the commitment was recovered from
+(e.g. "` + wal.NumberToFilenameV7(2) + `").
+
+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)")
+
+	Cmd.Flags().BoolVar(&flagUseWalSegmentNumber, "use-wal-segment-number", false,
+		"name the output checkpoint after the WAL segment number where the state commitment was found "+
+			"(e.g. \""+wal.NumberToFilenameV7(2)+"\") instead of the default \""+
+			bootstrap.FilenameWALRootCheckpoint+wal.V7FileSuffix+"\"")
+}
+
+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)
+	}
+
+	// The execution state directory is the source WAL directory and must not be
+	// written to: extracting into it would add the output checkpoint to the WAL
+	// directory the node loads on startup (and an interrupted write would delete
+	// source checkpoint files sharing the output name).
+	if filepath.Clean(flagOutputDir) == filepath.Clean(flagExecutionStateDir) {
+		return fmt.Errorf(
+			"--output-dir and --execution-state-dir must differ, but both are %s; refusing to write the extracted checkpoint into the execution state directory",
+			flagOutputDir)
+	}
+
+	log.Info().
+		Str("execution-state-dir", flagExecutionStateDir).
+		Str("output-dir", flagOutputDir).
+		Str("state-commitment", stateCommitment.String()).
+		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, sourceNumber, err := util.ReadPayloadlessTrie(flagExecutionStateDir, stateCommitment, int(flagMTrieCacheSize))
+	if err != nil {
+		return fmt.Errorf("cannot read payloadless trie for state commitment %s: %w", stateCommitment, err)
+	}
+
+	// By default the output uses the canonical V7 root checkpoint name. With
+	// --use-wal-segment-number, it is named after the WAL segment (or numbered
+	// checkpoint) the state commitment was recovered from, e.g.
+	// "checkpoint.00000002.v7". A sourceNumber of -1 means the target was found
+	// in the unnumbered V7 root checkpoint, so the default name is kept.
+	outputFile := bootstrap.FilenameWALRootCheckpoint + wal.V7FileSuffix
+	if flagUseWalSegmentNumber && sourceNumber >= 0 {
+		outputFile = wal.NumberToFilenameV7(sourceNumber)
+	}
+
+	// Refuse to overwrite: storing a checkpoint writes the header file plus part
+	// files named "<outputFile>.NNN", so fail if any file with that prefix
+	// already exists in the output directory.
+	outputPath := path.Join(flagOutputDir, outputFile)
+	exists, err := wal.AnyCheckpointFileExists(flagOutputDir, outputFile)
+	if err != nil {
+		return fmt.Errorf("cannot check for existing checkpoint files %s: %w", outputPath, err)
+	}
+	if exists {
+		return fmt.Errorf("output checkpoint %s already exists in %s; refusing to overwrite",
+			outputFile, flagOutputDir)
+	}
+
+	log.Info().
+		Str("root_hash", trie.RootHash().String()).
+		Uint64("allocated_reg_count", trie.AllocatedRegCount()).
+		Int("source_number", sourceNumber).
+		Str("output", outputPath).
+		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", outputPath).
+		Msg("✅ payloadless (V7) state extraction completed successfully")
+	return nil
+}

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)

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,78 @@ 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.
+//
+// `sourceNumber` identifies where the trie was recovered from: the number of the loaded V7 checkpoint
+// when the target was already one of its tries, or the number of the WAL segment whose replay produced
+// it. It is -1 when the target was found in the unnumbered V7 root checkpoint.
+//
+// No error returns are expected during normal operation.
+func ReadPayloadlessTrie(dir string, targetHash flow.StateCommitment, capacity int) (*payloadless.MTrie, int, 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, -1, 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, -1, 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, sourceNumber, err := diskWal.ReplayOnPayloadlessForestUntil(forest, targetRootHash)
+	if err != nil {
+		return nil, -1, fmt.Errorf("cannot replay payloadless WAL: %w", err)
+	}
+	if !found {
+		return nil, -1, 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, -1, fmt.Errorf("cannot get payloadless trie at state commitment %x: %w", targetHash[:], err)
+	}
+
+	return trie, sourceNumber, nil
+}
+
 func ReadTrieForPayloads(dir string, targetHash flow.StateCommitment) ([]*ledger.Payload, error) {
 	trie, err := ReadTrie(dir, targetHash)
 	if err != nil {

ledger/complete/wal/checkpoint_v6_reader.go

@@ -170,6 +170,30 @@ func filePathPattern(dir string, fileName string) string {
 	return fmt.Sprintf("%v*", filePathCheckpointHeader(dir, fileName))
 }
 
+// AnyCheckpointFileExists reports whether any file belonging to a checkpoint
+// with the given fileName exists under dir: the checkpoint header, any of its
+// part files, or any stray file sharing the same prefix (for example a leftover
+// from an interrupted write). It is intended for callers that must refuse to
+// overwrite the checkpoint output, since writing a checkpoint recreates all of
+// these files.
+//
+// No error returns are expected during normal operation.
+func AnyCheckpointFileExists(dir string, fileName string) (bool, error) {
+	// Enumerate the directory and compare literal name prefixes instead of using
+	// filepath.Glob: `dir` is an arbitrary path, so any glob metacharacter in it
+	// would be interpreted as pattern syntax rather than a literal directory name.
+	entries, err := os.ReadDir(dir)
+	if err != nil {
+		return false, fmt.Errorf("could not find checkpoint files: %w", err)
+	}
+	for _, entry := range entries {
+		if strings.HasPrefix(entry.Name(), fileName) {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
 // readCheckpointHeader takes a file path and returns subtrieChecksums and topTrieChecksum
 // any error returned are exceptions
 func readCheckpointHeader(filepath string, logger zerolog.Logger) (

ledger/complete/wal/checkpoint_v6_test.go

@@ -756,3 +756,44 @@ func TestCheckpointHasRootHash(t *testing.T) {
 		require.Error(t, CheckpointHasRootHash(logger, dir, fileName, nonExist))
 	})
 }
+
+// TestAnyCheckpointFileExists verifies that [AnyCheckpointFileExists] reports
+// true whenever any file with the checkpoint's prefix exists — the header, a
+// single part file, or a stray leftover — and false only when nothing with that
+// prefix is present. Callers use this to refuse overwriting a checkpoint.
+func TestAnyCheckpointFileExists(t *testing.T) {
+	unittest.RunWithTempDir(t, func(dir string) {
+		fileName := "checkpoint.00000001.v7"
+
+		exists, err := AnyCheckpointFileExists(dir, fileName)
+		require.NoError(t, err)
+		require.False(t, exists, "nothing exists yet")
+
+		// Header file alone.
+		require.NoError(t, os.WriteFile(path.Join(dir, fileName), []byte("header"), 0644))
+		exists, err = AnyCheckpointFileExists(dir, fileName)
+		require.NoError(t, err)
+		require.True(t, exists, "header file must be detected")
+
+		// A single part file alone.
+		require.NoError(t, os.Remove(path.Join(dir, fileName)))
+		require.NoError(t, os.WriteFile(path.Join(dir, partFileName(fileName, 5)), []byte("part"), 0644))
+		exists, err = AnyCheckpointFileExists(dir, fileName)
+		require.NoError(t, err)
+		require.True(t, exists, "any part file must be detected")
+
+		// A stray file sharing the prefix (leftover from an interrupted write).
+		require.NoError(t, os.Remove(path.Join(dir, partFileName(fileName, 5))))
+		require.NoError(t, os.WriteFile(path.Join(dir, fileName+".tmp"), []byte("partial"), 0644))
+		exists, err = AnyCheckpointFileExists(dir, fileName)
+		require.NoError(t, err)
+		require.True(t, exists, "stray file with the prefix must be detected")
+
+		// A file without the prefix must not trigger the check.
+		require.NoError(t, os.Remove(path.Join(dir, fileName+".tmp")))
+		require.NoError(t, os.WriteFile(path.Join(dir, "other.txt"), []byte("x"), 0644))
+		exists, err = AnyCheckpointFileExists(dir, fileName)
+		require.NoError(t, err)
+		require.False(t, exists, "unrelated file must be ignored")
+	})
+}