diff --git a/cmd/util/cmd/checkpoint-collect-stats/cmd.go b/cmd/util/cmd/checkpoint-collect-stats/cmd.go index 3269f4914cf..4f116cfe8ae 100644 --- a/cmd/util/cmd/checkpoint-collect-stats/cmd.go +++ b/cmd/util/cmd/checkpoint-collect-stats/cmd.go @@ -3,6 +3,7 @@ package checkpoint_collect_stats import ( "cmp" "encoding/hex" + "fmt" "math" "slices" "strings" @@ -315,6 +316,15 @@ func getPayloadStatsFromCheckpoint(payloadCallBack func(payload *ledger.Payload) memAllocBefore := debug.GetHeapAllocsBytes() log.Info().Msgf("loading checkpoint(s) from %v", flagCheckpointDir) + // checkpoint-collect-stats analyzes payload contents (register types, sizes, + // account info). V7 (payloadless) checkpoints store only leaf hashes and contain + // no payloads, so they cannot be processed here. The WAL replay below loads only + // V6 checkpoints and silently ignores V7 files, which would otherwise produce + // misleading (stale or empty) stats. Fail fast with a clear error instead. + if err := requireV6Checkpoint(flagCheckpointDir); err != nil { + log.Fatal().Err(err).Msg("cannot collect stats from checkpoint") + } + diskWal, err := wal.NewDiskWAL(zerolog.Nop(), nil, &metrics.NoopCollector{}, flagCheckpointDir, complete.DefaultCacheSize, pathfinder.PathByteSize, wal.SegmentSize) if err != nil { log.Fatal().Err(err).Msg("cannot create WAL") @@ -369,6 +379,33 @@ func getPayloadStatsFromCheckpoint(payloadCallBack func(payload *ledger.Payload) return ledgerStats } +// requireV6Checkpoint returns an error if the latest checkpoint in dir is a V7 +// (payloadless) checkpoint. checkpoint-collect-stats requires full payloads, +// which V7 checkpoints do not contain. +// +// Only numbered checkpoints are considered (the WAL bootstrap loads the latest +// numbered V6 checkpoint). If the latest numbered checkpoint is V7, this command +// would otherwise silently fall back to an older V6 checkpoint or an empty state, +// reporting misleading stats. +// +// Expected error returns during normal operation: +// - an error when the latest checkpoint in dir is a V7 (payloadless) checkpoint +func requireV6Checkpoint(dir string) error { + _, latest, err := wal.ListCheckpointsWithInfo(dir) + if err != nil { + return fmt.Errorf("cannot list checkpoints in %s: %w", dir, err) + } + + if latest != nil && latest.Version == wal.VersionV7 { + return fmt.Errorf( + "checkpoint %d in %s is a V7 (payloadless) checkpoint, which contains no payloads; "+ + "checkpoint-collect-stats requires a V6 checkpoint", + latest.Number, dir) + } + + return nil +} + func getRegisterStats(valueSizesByType sizesByType) []RegisterStatsByTypes { domainStats := make([]RegisterStatsByTypes, 0, len(common.AllStorageDomains)) var allDomainSizes []float64 diff --git a/cmd/util/cmd/checkpoint-collect-stats/cmd_test.go b/cmd/util/cmd/checkpoint-collect-stats/cmd_test.go new file mode 100644 index 00000000000..72df37ec599 --- /dev/null +++ b/cmd/util/cmd/checkpoint-collect-stats/cmd_test.go @@ -0,0 +1,56 @@ +package checkpoint_collect_stats + +import ( + "testing" + + "github.com/rs/zerolog" + "github.com/stretchr/testify/require" + + "github.com/onflow/flow-go/ledger" + "github.com/onflow/flow-go/ledger/common/testutils" + "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" +) + +// TestRequireV6Checkpoint_EmptyDir verifies that a directory without any numbered +// checkpoint is accepted (the caller proceeds with WAL replay / root checkpoint). +func TestRequireV6Checkpoint_EmptyDir(t *testing.T) { + require.NoError(t, requireV6Checkpoint(t.TempDir())) +} + +// TestRequireV6Checkpoint_V6 verifies that a directory whose latest checkpoint is +// V6 is accepted. +func TestRequireV6Checkpoint_V6(t *testing.T) { + dir := t.TempDir() + + p := testutils.PathByUint8(0) + v := testutils.LightPayload8('A', 'a') + tr, _, err := trie.NewTrieWithUpdatedRegisters( + trie.NewEmptyMTrie(), []ledger.Path{p}, []ledger.Payload{*v}, true) + require.NoError(t, err) + + require.NoError(t, wal.StoreCheckpointV6Concurrently( + []*trie.MTrie{tr}, dir, wal.NumberToFilename(1), zerolog.Nop())) + + require.NoError(t, requireV6Checkpoint(dir)) +} + +// TestRequireV6Checkpoint_V7 verifies that a directory whose latest checkpoint is +// V7 (payloadless) is rejected, since this command requires full payloads. +func TestRequireV6Checkpoint_V7(t *testing.T) { + dir := t.TempDir() + + p := testutils.PathByUint8(0) + v := testutils.LightPayload8('A', 'a') + tr, _, err := payloadless.NewTrieWithUpdatedRegisters( + payloadless.NewEmptyMTrie(), []ledger.Path{p}, [][]byte{v.Value()}, true) + require.NoError(t, err) + + require.NoError(t, wal.StoreCheckpointV7Concurrently( + []*payloadless.MTrie{tr}, dir, wal.NumberToFilenameV7(1), zerolog.Nop())) + + err = requireV6Checkpoint(dir) + require.Error(t, err) + require.Contains(t, err.Error(), "V7") +} diff --git a/cmd/util/cmd/checkpoint-list-tries/cmd.go b/cmd/util/cmd/checkpoint-list-tries/cmd.go index 830075bc5c8..a325db37e6b 100644 --- a/cmd/util/cmd/checkpoint-list-tries/cmd.go +++ b/cmd/util/cmd/checkpoint-list-tries/cmd.go @@ -2,10 +2,14 @@ package checkpoint_list_tries import ( "fmt" + "path/filepath" + "strings" + "github.com/rs/zerolog" "github.com/rs/zerolog/log" "github.com/spf13/cobra" + "github.com/onflow/flow-go/ledger" "github.com/onflow/flow-go/ledger/complete/wal" ) @@ -28,14 +32,32 @@ func init() { func run(*cobra.Command, []string) { - log.Info().Msgf("loading checkpoint %v", flagCheckpoint) - tries, err := wal.LoadCheckpoint(flagCheckpoint, log.Logger) + log.Info().Msgf("reading trie root hashes from checkpoint %v", flagCheckpoint) + + hashes, err := readTrieRootHashes(log.Logger, flagCheckpoint) if err != nil { - log.Fatal().Err(err).Msg("error while loading checkpoint") + log.Fatal().Err(err).Msg("error while reading trie root hashes from checkpoint") + } + log.Info().Msgf("checkpoint read, total tries: %v", len(hashes)) + + for _, h := range hashes { + fmt.Printf("trie root hash: %s\n", h) } - log.Info().Msgf("checkpoint loaded, total tries: %v", len(tries)) +} - for _, trie := range tries { - fmt.Printf("trie root hash: %s\n", trie.RootHash()) +// readTrieRootHashes reads only the trie root hashes from the checkpoint file at +// the given path, without materializing the full trie forest. Only the top-trie +// part file (containing the trie root records) is read. +// +// Both V6 and V7 (payloadless) checkpoints are supported; the version is +// determined by the V7 filename suffix ([wal.V7FileSuffix]). The root hashes are +// returned in the order they are stored in the checkpoint. +// +// No error returns are expected during normal operation. +func readTrieRootHashes(logger zerolog.Logger, checkpointFilePath string) ([]ledger.RootHash, error) { + dir, fileName := filepath.Split(checkpointFilePath) + if strings.HasSuffix(fileName, wal.V7FileSuffix) { + return wal.ReadTriesRootHashV7(logger, dir, fileName) } + return wal.ReadTriesRootHash(logger, dir, fileName) } diff --git a/cmd/util/cmd/checkpoint-list-tries/cmd_test.go b/cmd/util/cmd/checkpoint-list-tries/cmd_test.go new file mode 100644 index 00000000000..138c22d3f07 --- /dev/null +++ b/cmd/util/cmd/checkpoint-list-tries/cmd_test.go @@ -0,0 +1,94 @@ +package checkpoint_list_tries + +import ( + "path/filepath" + "testing" + + "github.com/rs/zerolog" + "github.com/stretchr/testify/require" + + "github.com/onflow/flow-go/ledger" + "github.com/onflow/flow-go/ledger/common/testutils" + "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" +) + +// TestReadTrieRootHashesV6 verifies that the trie root hashes are read from a V6 +// checkpoint in the order they were stored, without loading the full forest. +func TestReadTrieRootHashesV6(t *testing.T) { + dir := t.TempDir() + const fileName = "checkpoint" + + tries := createV6Tries(t) + + err := wal.StoreCheckpointV6Concurrently(tries, dir, fileName, zerolog.Nop()) + require.NoError(t, err) + + hashes, err := readTrieRootHashes(zerolog.Nop(), filepath.Join(dir, fileName)) + require.NoError(t, err) + + expected := make([]ledger.RootHash, len(tries)) + for i, tr := range tries { + expected[i] = tr.RootHash() + } + require.Equal(t, expected, hashes) +} + +// TestReadTrieRootHashesV7 verifies that the trie root hashes are read from a V7 +// (payloadless) checkpoint in the order they were stored, by dispatching on the +// V7 filename suffix. +func TestReadTrieRootHashesV7(t *testing.T) { + dir := t.TempDir() + fileName := "checkpoint" + wal.V7FileSuffix + + tries := createV7Tries(t) + + err := wal.StoreCheckpointV7Concurrently(tries, dir, fileName, zerolog.Nop()) + require.NoError(t, err) + + hashes, err := readTrieRootHashes(zerolog.Nop(), filepath.Join(dir, fileName)) + require.NoError(t, err) + + expected := make([]ledger.RootHash, len(tries)) + for i, tr := range tries { + expected[i] = tr.RootHash() + } + require.Equal(t, expected, hashes) +} + +// createV6Tries builds a chain of two distinct full-payload tries for use as V6 +// checkpoint content. +func createV6Tries(t *testing.T) []*trie.MTrie { + p1 := testutils.PathByUint8(0) + v1 := testutils.LightPayload8('A', 'a') + trie1, _, err := trie.NewTrieWithUpdatedRegisters( + trie.NewEmptyMTrie(), []ledger.Path{p1}, []ledger.Payload{*v1}, true) + require.NoError(t, err) + + p2 := testutils.PathByUint8(1) + v2 := testutils.LightPayload8('B', 'b') + trie2, _, err := trie.NewTrieWithUpdatedRegisters( + trie1, []ledger.Path{p2}, []ledger.Payload{*v2}, true) + require.NoError(t, err) + + return []*trie.MTrie{trie1, trie2} +} + +// createV7Tries builds a chain of two distinct payloadless tries for use as V7 +// checkpoint content. +func createV7Tries(t *testing.T) []*payloadless.MTrie { + p1 := testutils.PathByUint8(0) + v1 := testutils.LightPayload8('A', 'a') + trie1, _, err := payloadless.NewTrieWithUpdatedRegisters( + payloadless.NewEmptyMTrie(), []ledger.Path{p1}, [][]byte{v1.Value()}, true) + require.NoError(t, err) + + p2 := testutils.PathByUint8(1) + v2 := testutils.LightPayload8('B', 'b') + trie2, _, err := payloadless.NewTrieWithUpdatedRegisters( + trie1, []ledger.Path{p2}, [][]byte{v2.Value()}, true) + require.NoError(t, err) + + return []*payloadless.MTrie{trie1, trie2} +} diff --git a/cmd/util/cmd/checkpoint-trie-stats/cmd.go b/cmd/util/cmd/checkpoint-trie-stats/cmd.go deleted file mode 100644 index 327a4cf037b..00000000000 --- a/cmd/util/cmd/checkpoint-trie-stats/cmd.go +++ /dev/null @@ -1,113 +0,0 @@ -package checkpoint_trie_stats - -import ( - "errors" - "fmt" - - "github.com/rs/zerolog" - "github.com/rs/zerolog/log" - "github.com/spf13/cobra" - - "github.com/onflow/flow-go/ledger/complete/mtrie/node" - "github.com/onflow/flow-go/ledger/complete/mtrie/trie" - "github.com/onflow/flow-go/ledger/complete/wal" -) - -var ( - flagCheckpoint string - flagTrieIndex int -) - -var Cmd = &cobra.Command{ - Use: "checkpoint-trie-stats", - Short: "List the trie node count by types in a checkpoint, show total payload size", - Run: run, -} - -func init() { - - Cmd.Flags().StringVar(&flagCheckpoint, "checkpoint", "", - "checkpoint file to read") - _ = Cmd.MarkFlagRequired("checkpoint") - Cmd.Flags().IntVar(&flagTrieIndex, "trie-index", 0, "trie index to read, 0 being the first trie, -1 is the last trie") - -} - -func run(*cobra.Command, []string) { - - log.Info().Msgf("loading checkpoint %v, reading %v-th trie", flagCheckpoint, flagTrieIndex) - res, err := scanCheckpoint(flagCheckpoint, flagTrieIndex, log.Logger) - if err != nil { - log.Fatal().Err(err).Msg("fail to scan checkpoint") - } - log.Info(). - Str("TrieRootHash", res.trieRootHash). - Int("InterimNodeCount", res.interimNodeCount). - Int("LeafNodeCount", res.leafNodeCount). - Int("TotalPayloadSize", res.totalPayloadSize). - Msgf("successfully scanned checkpoint %v", flagCheckpoint) -} - -type result struct { - trieRootHash string - interimNodeCount int - leafNodeCount int - totalPayloadSize int -} - -func readTrie(tries []*trie.MTrie, index int) (*trie.MTrie, error) { - if len(tries) == 0 { - return nil, errors.New("No tries available") - } - - if index < -len(tries) || index >= len(tries) { - return nil, fmt.Errorf("index %d out of range", index) - } - - if index < 0 { - return tries[len(tries)+index], nil - } - - return tries[index], nil -} - -func scanCheckpoint(checkpoint string, trieIndex int, log zerolog.Logger) (result, error) { - tries, err := wal.LoadCheckpoint(flagCheckpoint, log) - if err != nil { - return result{}, fmt.Errorf("error while loading checkpoint: %w", err) - } - - log.Info(). - Int("total_tries", len(tries)). - Msg("checkpoint loaded") - - t, err := readTrie(tries, trieIndex) - if err != nil { - return result{}, fmt.Errorf("error while reading trie: %w", err) - } - - log.Info().Msgf("trie loaded, root hash: %v", t.RootHash()) - - res := &result{ - trieRootHash: t.RootHash().String(), - interimNodeCount: 0, - leafNodeCount: 0, - totalPayloadSize: 0, - } - processNode := func(n *node.Node) error { - if n.IsLeaf() { - res.leafNodeCount++ - res.totalPayloadSize += n.Payload().Size() - } else { - res.interimNodeCount++ - } - return nil - } - - err = trie.TraverseNodes(t, processNode) - if err != nil { - return result{}, fmt.Errorf("fail to traverse the trie: %w", err) - } - - return *res, nil -}