From 10e1e56eb61be98b29b926605cefaf15a068784b Mon Sep 17 00:00:00 2001 From: "Leo Zhang (zhangchiqing)" Date: Mon, 15 Jun 2026 11:41:02 -0700 Subject: [PATCH 1/4] add convert_stream --- cmd/util/cmd/checkpoint-convert-v7/cmd.go | 57 +- .../wal/checkpoint_v7_convert_stream.go | 540 ++++++++++++++++++ .../wal/checkpoint_v7_convert_stream_test.go | 186 ++++++ 3 files changed, 770 insertions(+), 13 deletions(-) create mode 100644 ledger/complete/wal/checkpoint_v7_convert_stream.go create mode 100644 ledger/complete/wal/checkpoint_v7_convert_stream_test.go diff --git a/cmd/util/cmd/checkpoint-convert-v7/cmd.go b/cmd/util/cmd/checkpoint-convert-v7/cmd.go index 675ac442c04..c2f9794468f 100644 --- a/cmd/util/cmd/checkpoint-convert-v7/cmd.go +++ b/cmd/util/cmd/checkpoint-convert-v7/cmd.go @@ -11,11 +11,13 @@ import ( ) var ( - flagCheckpointDir string - flagCheckpoint string - flagOutputDir string - flagOutput string - flagNWorker uint + flagCheckpointDir string + flagCheckpoint string + flagOutputDir string + flagOutput string + flagNWorker uint + flagStream bool + flagVerifyLeafHash bool ) // Cmd converts a V6 checkpoint to a V7 (payloadless) checkpoint by reading @@ -54,6 +56,15 @@ func init() { Cmd.Flags().UintVar(&flagNWorker, "nworker", 16, "number of subtrie files to encode in parallel (valid range [1, 16])") + + Cmd.Flags().BoolVar(&flagStream, "stream", false, + "stream part files node-by-node instead of loading the full trie forest into memory "+ + "(constant memory, preserves node hashes without re-deriving root hashes)") + + Cmd.Flags().BoolVar(&flagVerifyLeafHash, "verify-leaf-hash", false, + "in --stream mode, verify every allocated leaf by comparing the derived V7 node hash "+ + "against the V6 node hash on disk (ignored without --stream, which already cross-checks "+ + "root hashes)") } func run(*cobra.Command, []string) { @@ -73,16 +84,36 @@ func run(*cobra.Command, []string) { Str("output_dir", outputDir). Str("output", outputFile). Uint("nworker", flagNWorker). + Bool("stream", flagStream). + Bool("verify_leaf_hash", flagVerifyLeafHash). Msg("converting V6 checkpoint to V7") - err := wal.ConvertCheckpointV6ToV7( - flagCheckpointDir, - flagCheckpoint, - outputDir, - outputFile, - log.Logger, - flagNWorker, - ) + var err error + if flagStream { + err = wal.ConvertCheckpointV6ToV7Stream( + flagCheckpointDir, + flagCheckpoint, + outputDir, + outputFile, + log.Logger, + flagNWorker, + flagVerifyLeafHash, + ) + } else { + if flagVerifyLeafHash { + // The in-memory converter already cross-checks every trie root hash, + // which transitively verifies all leaf hashes, so the flag is a no-op here. + log.Warn().Msg("--verify-leaf-hash has no effect without --stream; the in-memory converter already verifies root hashes") + } + err = wal.ConvertCheckpointV6ToV7( + flagCheckpointDir, + flagCheckpoint, + outputDir, + outputFile, + log.Logger, + flagNWorker, + ) + } if err != nil { log.Fatal().Err(err).Msg("checkpoint conversion failed") } diff --git a/ledger/complete/wal/checkpoint_v7_convert_stream.go b/ledger/complete/wal/checkpoint_v7_convert_stream.go new file mode 100644 index 00000000000..8eeaca4eac6 --- /dev/null +++ b/ledger/complete/wal/checkpoint_v7_convert_stream.go @@ -0,0 +1,540 @@ +package wal + +import ( + "bufio" + "encoding/binary" + "fmt" + "io" + "os" + + "github.com/rs/zerolog" + + "github.com/onflow/flow-go/ledger" + "github.com/onflow/flow-go/ledger/common/hash" + "github.com/onflow/flow-go/ledger/complete/mtrie/flattener" + "github.com/onflow/flow-go/ledger/complete/mtrie/node" + "github.com/onflow/flow-go/ledger/complete/payloadless" +) + +// Encoded node field sizes shared by the V6 and V7 on-disk node formats. They +// mirror the (unexported) constants in the mtrie/flattener and payloadless +// flatteners; they are duplicated here because the streaming converter operates +// on the raw byte stream rather than through either flattener. +const ( + encNodeTypeSize = 1 + encHeightSize = 2 + encHashSize = hash.HashLen + encPathSize = ledger.PathLen + encNodeIndexSize = 8 + encPayloadLengthSize = 4 + + // fixedNodePrefixSize is the size of the leading bytes shared by every + // encoded node (leaf or interim): node type + height + node hash. + fixedNodePrefixSize = encNodeTypeSize + encHeightSize + encHashSize + + // leafNodeTypeByte and interimNodeTypeByte are the node-type tags. They are + // identical in the V6 and V7 encodings, so an interim node's bytes can be + // copied verbatim. + leafNodeTypeByte = byte(0) + interimNodeTypeByte = byte(1) + + // payloadEncodingVersion is the payload encoding version used by the V6 + // leaf node encoding. + payloadEncodingVersion = 1 +) + +// ConvertCheckpointV6ToV7Stream converts a V6 checkpoint at (inputDir, inputFileName) +// into a V7 (payloadless) checkpoint at (outputDir, outputFileName) by streaming +// each part file node-by-node, without ever materializing the full trie forest in +// memory. +// +// How it works: +// - The V6 and V7 on-disk layouts are byte-identical except for (a) the version +// bytes in every part file, (b) the leaf node encoding — V6 stores the full +// payload, V7 stores a 32-byte leaf hash — and (c) the trie root records in the +// top-trie part file, where V7 drops V6's 8-byte allocated-register-size field. +// Interim nodes are byte-identical. +// - Each of the 16 subtrie part files is a pure node stream: interim nodes are +// copied verbatim and leaf nodes are projected to their payloadless form. +// - The top-trie part file additionally re-encodes each trie root record to drop +// the register-size field. +// - Node count and ordering are unchanged by the conversion, so every interim +// node's child indices remain valid without rewriting. +// - Per-part-file CRC32 checksums are recomputed during the write and collected +// into a freshly written V7 header. +// +// Peak memory is independent of checkpoint size: a single node plus reusable +// scratch buffers per part file. The 16 subtrie part files are converted in +// parallel using up to nWorker goroutines; valid range is [1, subtrieCount]. +// +// Unlike [ConvertCheckpointV6ToV7], this function does not load the forest and +// therefore does not re-derive or cross-check trie root hashes. Node hashes are +// carried over verbatim from the V6 stream, so root hashes are structurally +// preserved. +// +// When verifyLeafHash is true, every allocated leaf is additionally checked: the +// node hash derived from (path, value) is compared against the leaf's V6 node hash +// read from disk, and a mismatch aborts the conversion. This is the streaming, +// per-leaf equivalent of the root-hash cross-check performed by the in-memory +// [ConvertCheckpointV6ToV7] (a wrong leaf hash would otherwise only surface as a +// root-hash mismatch when the forest is loaded). It adds a hash recomputation per +// allocated leaf but no extra memory. +// +// The output filename must carry the V7 suffix and no output part file may already +// exist; otherwise the call is rejected. On any failure, partially written output +// files are removed. +// +// No error returns are expected during normal operation; all error returns indicate +// a malformed input, a clobbering output, an IO failure, or a leaf-hash mismatch +// when verifyLeafHash is enabled. +func ConvertCheckpointV6ToV7Stream( + inputDir string, + inputFileName string, + outputDir string, + outputFileName string, + logger zerolog.Logger, + nWorker uint, + verifyLeafHash bool, +) error { + err := convertCheckpointV6ToV7Stream(inputDir, inputFileName, outputDir, outputFileName, logger, nWorker, verifyLeafHash) + if err != nil { + cleanupErr := deleteCheckpointFiles(outputDir, outputFileName) + if cleanupErr != nil { + return fmt.Errorf("fail to cleanup temp file %s, after running into error: %w", cleanupErr, err) + } + return err + } + return nil +} + +func convertCheckpointV6ToV7Stream( + inputDir string, + inputFileName string, + outputDir string, + outputFileName string, + logger zerolog.Logger, + nWorker uint, + verifyLeafHash bool, +) error { + if nWorker == 0 || nWorker > subtrieCount { + return fmt.Errorf("invalid nWorker %v, valid range is [1, %v]", nWorker, subtrieCount) + } + + // Reject obvious filename misuse so converted files can coexist with the V6 source. + if err := requireV7Filename(outputFileName); err != nil { + return err + } + + // Validate V6 input exists (header + part files). + v6Header := filePathCheckpointHeader(inputDir, inputFileName) + if _, err := os.Stat(v6Header); err != nil { + return fmt.Errorf("V6 checkpoint header not found at %s: %w", v6Header, err) + } + subtrieChecksums, topTrieChecksum, err := readCheckpointHeader(v6Header, logger) + if err != nil { + return fmt.Errorf("could not read V6 checkpoint header: %w", err) + } + if err := allPartFileExist(inputDir, inputFileName, len(subtrieChecksums)); err != nil { + return fmt.Errorf("V6 part files incomplete for %s/%s: %w", inputDir, inputFileName, err) + } + + // Validate V7 output is not present (any of the part files). + v7Existing, err := findCheckpointPartFiles(outputDir, outputFileName) + if err != nil { + return fmt.Errorf("could not check existing V7 output files: %w", err) + } + if len(v7Existing) != 0 { + return fmt.Errorf("V7 output already exists: %v", v7Existing) + } + + logger.Info(). + Str("v6_dir", inputDir). + Str("v6_file", inputFileName). + Str("v7_dir", outputDir). + Str("v7_file", outputFileName). + Uint("nworker", nWorker). + Bool("verify_leaf_hash", verifyLeafHash). + Msg("starting streaming V6→V7 checkpoint conversion") + + // Convert the 16 subtrie part files concurrently, recomputing each checksum. + newSubtrieChecksums, err := convertSubTriesV6ToV7StreamConcurrently( + inputDir, inputFileName, outputDir, outputFileName, subtrieChecksums, logger, nWorker, verifyLeafHash) + if err != nil { + return fmt.Errorf("could not convert subtrie files: %w", err) + } + + // Convert the top-trie part file. + newTopTrieChecksum, err := convertTopTrieFileV6ToV7Stream( + inputDir, inputFileName, outputDir, outputFileName, topTrieChecksum, logger, verifyLeafHash) + if err != nil { + return fmt.Errorf("could not convert top-trie file: %w", err) + } + + // Write the V7 header referencing the freshly computed checksums. + if err := storeCheckpointHeaderV7(newSubtrieChecksums, newTopTrieChecksum, outputDir, outputFileName, logger); err != nil { + return fmt.Errorf("could not write V7 checkpoint header: %w", err) + } + + logger.Info().Msg("stream V6→V7 checkpoint conversion complete") + return nil +} + +type streamSubtrieResult struct { + index int + checksum uint32 + err error +} + +// convertSubTriesV6ToV7StreamConcurrently streams all subtrieCount subtrie part +// files through the V6→V7 conversion using up to nWorker goroutines, and returns +// the recomputed per-file checksums in subtrie-index order. +func convertSubTriesV6ToV7StreamConcurrently( + inputDir string, + inputFileName string, + outputDir string, + outputFileName string, + subtrieChecksums []uint32, + logger zerolog.Logger, + nWorker uint, + verifyLeafHash bool, +) ([]uint32, error) { + jobs := make(chan int, subtrieCount) + for i := 0; i < subtrieCount; i++ { + jobs <- i + } + close(jobs) + + // Buffered to subtrieCount so workers never block on send, even if the + // collector returns early after the first error. + results := make(chan streamSubtrieResult, subtrieCount) + + for w := 0; w < int(nWorker); w++ { + go func() { + for i := range jobs { + sum, err := convertSubTrieFileV6ToV7Stream( + inputDir, inputFileName, outputDir, outputFileName, i, subtrieChecksums[i], logger, verifyLeafHash) + results <- streamSubtrieResult{index: i, checksum: sum, err: err} + } + }() + } + + checksums := make([]uint32, subtrieCount) + for k := 0; k < subtrieCount; k++ { + r := <-results + if r.err != nil { + return nil, fmt.Errorf("fail to convert %v-th subtrie: %w", r.index, r.err) + } + checksums[r.index] = r.checksum + } + return checksums, nil +} + +// convertSubTrieFileV6ToV7Stream streams the subtrie part file at the given index, +// writing the converted V7 subtrie part file, and returns the recomputed checksum. +// +// expectedSum is the checksum recorded in the V6 header for this subtrie; it is +// verified against the checksum embedded in the V6 subtrie file before conversion. +func convertSubTrieFileV6ToV7Stream( + inputDir string, + inputFileName string, + outputDir string, + outputFileName string, + index int, + expectedSum uint32, + logger zerolog.Logger, + verifyLeafHash bool, +) (checksum uint32, errToReturn error) { + inPath, _, err := filePathSubTries(inputDir, inputFileName, index) + if err != nil { + return 0, err + } + + inFile, err := os.Open(inPath) + if err != nil { + return 0, fmt.Errorf("could not open subtrie file %v: %w", inPath, err) + } + defer func() { + errToReturn = closeAndMergeError(inFile, errToReturn) + }() + + nodeCount, embeddedSum, err := readSubTriesFooter(inFile) + if err != nil { + return 0, fmt.Errorf("could not read subtrie footer: %w", err) + } + if embeddedSum != expectedSum { + return 0, fmt.Errorf("mismatch checksum in subtrie file %v: header has %v, file has %v", + index, expectedSum, embeddedSum) + } + + if _, err := inFile.Seek(0, io.SeekStart); err != nil { + return 0, fmt.Errorf("could not seek to start of subtrie file: %w", err) + } + if err := validateFileHeader(MagicBytesCheckpointSubtrie, VersionV6, inFile); err != nil { + return 0, fmt.Errorf("invalid subtrie file header: %w", err) + } + reader := bufio.NewReaderSize(inFile, defaultBufioReadSize) + + closable, err := createWriterForSubtrie(outputDir, outputFileName, logger, index) + if err != nil { + return 0, fmt.Errorf("could not create writer for subtrie: %w", err) + } + defer func() { + errToReturn = closeAndMergeError(closable, errToReturn) + }() + + writer := NewCRC32Writer(closable) + if _, err := writer.Write(encodeVersion(MagicBytesCheckpointSubtrie, VersionV7)); err != nil { + return 0, fmt.Errorf("cannot write version into subtrie file: %w", err) + } + + logging := logProgress(fmt.Sprintf("converting %v-th sub trie (streaming)", index), int(nodeCount), logger) + conv := newV6ToV7NodeConverter(verifyLeafHash) + for i := uint64(0); i < nodeCount; i++ { + if err := conv.convertNode(reader, writer); err != nil { + return 0, fmt.Errorf("cannot convert node %d of subtrie %d: %w", i, index, err) + } + logging(i) + } + + sum, err := storeSubtrieFooter(nodeCount, writer) + if err != nil { + return 0, fmt.Errorf("could not store subtrie footer: %w", err) + } + return sum, nil +} + +// convertTopTrieFileV6ToV7Stream streams the top-trie part file, converting its +// top-level nodes and re-encoding each trie root record to drop V6's register-size +// field, and returns the recomputed checksum. +// +// expectedSum is the top-trie checksum recorded in the V6 header; it is verified +// against the checksum embedded in the V6 top-trie file before conversion. +func convertTopTrieFileV6ToV7Stream( + inputDir string, + inputFileName string, + outputDir string, + outputFileName string, + expectedSum uint32, + logger zerolog.Logger, + verifyLeafHash bool, +) (checksum uint32, errToReturn error) { + inPath, _ := filePathTopTries(inputDir, inputFileName) + + inFile, err := os.Open(inPath) + if err != nil { + return 0, fmt.Errorf("could not open top-trie file %v: %w", inPath, err) + } + defer func() { + errToReturn = closeAndMergeError(inFile, errToReturn) + }() + + topLevelNodesCount, triesCount, embeddedSum, err := readTopTriesFooter(inFile) + if err != nil { + return 0, fmt.Errorf("could not read top-trie footer: %w", err) + } + if embeddedSum != expectedSum { + return 0, fmt.Errorf("mismatch top-trie checksum: header has %v, file has %v", + expectedSum, embeddedSum) + } + + if _, err := inFile.Seek(0, io.SeekStart); err != nil { + return 0, fmt.Errorf("could not seek to start of top-trie file: %w", err) + } + if err := validateFileHeader(MagicBytesCheckpointToptrie, VersionV6, inFile); err != nil { + return 0, fmt.Errorf("invalid top-trie file header: %w", err) + } + reader := bufio.NewReaderSize(inFile, defaultBufioReadSize) + + // Read the subtrie node count and carry it over verbatim (unchanged by conversion). + subtrieNodeCountBuf := make([]byte, encNodeCountSize) + if _, err := io.ReadFull(reader, subtrieNodeCountBuf); err != nil { + return 0, fmt.Errorf("could not read subtrie node count: %w", err) + } + + closable, err := createWriterForTopTries(outputDir, outputFileName, logger) + if err != nil { + return 0, fmt.Errorf("could not create writer for top tries: %w", err) + } + defer func() { + errToReturn = closeAndMergeError(closable, errToReturn) + }() + + writer := NewCRC32Writer(closable) + if _, err := writer.Write(encodeVersion(MagicBytesCheckpointToptrie, VersionV7)); err != nil { + return 0, fmt.Errorf("cannot write version into top-trie file: %w", err) + } + if _, err := writer.Write(subtrieNodeCountBuf); err != nil { + return 0, fmt.Errorf("cannot write subtrie node count: %w", err) + } + + // Convert the top-level nodes (above subtrieLevel). + conv := newV6ToV7NodeConverter(verifyLeafHash) + for i := uint64(0); i < topLevelNodesCount; i++ { + if err := conv.convertNode(reader, writer); err != nil { + return 0, fmt.Errorf("cannot convert top-level node %d: %w", i, err) + } + } + + // Re-encode each trie root record from V6 (index + regCount + regSize + hash) + // to V7 (index + regCount + hash), dropping the register-size field. + readScratch := make([]byte, flattener.EncodedTrieSize) + trieBuf := make([]byte, payloadless.EncodedTrieSize) + for i := uint16(0); i < triesCount; i++ { + encTrie, err := flattener.ReadEncodedTrie(reader, readScratch) + if err != nil { + return 0, fmt.Errorf("cannot read trie root record %d: %w", i, err) + } + + pos := 0 + binary.BigEndian.PutUint64(trieBuf[pos:], encTrie.RootIndex) + pos += encNodeIndexSize + binary.BigEndian.PutUint64(trieBuf[pos:], encTrie.RegCount) + pos += encNodeIndexSize + copy(trieBuf[pos:], encTrie.RootHash[:]) + + if _, err := writer.Write(trieBuf); err != nil { + return 0, fmt.Errorf("cannot write converted trie root record %d: %w", i, err) + } + } + + sum, err := storeTopLevelTrieFooter(topLevelNodesCount, triesCount, writer) + if err != nil { + return 0, fmt.Errorf("could not store top-trie footer: %w", err) + } + return sum, nil +} + +// v6ToV7NodeConverter streams individual V6-encoded nodes into V7-encoded nodes, +// reusing internal scratch buffers across calls to avoid per-node allocations. +// +// NOT CONCURRENCY SAFE! A single converter must be used by one goroutine at a time. +type v6ToV7NodeConverter struct { + prefix []byte // node type + height + hash (fixedNodePrefixSize) + childIndex []byte // interim left + right child indices + path []byte // leaf path + lenBuf []byte // leaf payload length prefix + payload []byte // leaf payload bytes (grows as needed) + enc []byte // scratch for the payloadless leaf encoding + + // verifyLeafHash, when true, makes convertLeaf compare each derived V7 leaf + // node hash against the V6 leaf node hash read from disk. + verifyLeafHash bool +} + +// newV6ToV7NodeConverter returns a converter with preallocated scratch buffers. +func newV6ToV7NodeConverter(verifyLeafHash bool) *v6ToV7NodeConverter { + return &v6ToV7NodeConverter{ + prefix: make([]byte, fixedNodePrefixSize), + childIndex: make([]byte, 2*encNodeIndexSize), + path: make([]byte, encPathSize), + lenBuf: make([]byte, encPayloadLengthSize), + payload: make([]byte, 1024), + enc: make([]byte, 1024*4), + verifyLeafHash: verifyLeafHash, + } +} + +// convertNode reads one V6-encoded node from reader and writes its V7 encoding to +// writer. Interim nodes are copied verbatim (their on-disk format is identical in +// V7); leaf nodes are projected via [FromV6LeafNode] and re-encoded with the +// payloadless flattener. +// +// No error returns are expected during normal operation; all error returns indicate +// a malformed input stream or an IO failure. +func (c *v6ToV7NodeConverter) convertNode(reader io.Reader, writer io.Writer) error { + if _, err := io.ReadFull(reader, c.prefix); err != nil { + return fmt.Errorf("cannot read node prefix: %w", err) + } + + switch c.prefix[0] { + case interimNodeTypeByte: + // Interim node: read the two child indices and copy the whole record verbatim. + if _, err := io.ReadFull(reader, c.childIndex); err != nil { + return fmt.Errorf("cannot read interim node child indices: %w", err) + } + if _, err := writer.Write(c.prefix); err != nil { + return fmt.Errorf("cannot write interim node prefix: %w", err) + } + if _, err := writer.Write(c.childIndex); err != nil { + return fmt.Errorf("cannot write interim node child indices: %w", err) + } + return nil + + case leafNodeTypeByte: + return c.convertLeaf(reader, writer) + + default: + return fmt.Errorf("failed to decode node type %d", c.prefix[0]) + } +} + +// convertLeaf reads the remainder of a V6 leaf node (path + payload) from reader, +// having already consumed the shared prefix into c.prefix, and writes its V7 +// payloadless encoding to writer. +// +// When c.verifyLeafHash is set, the V7 node hash derived from (path, value) is +// compared against the V6 node hash read from disk, and a mismatch is reported as +// an error. +// +// No error returns are expected during normal operation; all error returns indicate +// a malformed input stream, an IO failure, or (when verifying) a leaf-hash mismatch. +func (c *v6ToV7NodeConverter) convertLeaf(reader io.Reader, writer io.Writer) error { + height := binary.BigEndian.Uint16(c.prefix[encNodeTypeSize:]) + nodeHash, err := hash.ToHash(c.prefix[encNodeTypeSize+encHeightSize:]) + if err != nil { + return fmt.Errorf("failed to decode leaf node hash: %w", err) + } + + // Read path (32 bytes). + if _, err := io.ReadFull(reader, c.path); err != nil { + return fmt.Errorf("cannot read leaf path: %w", err) + } + path, err := ledger.ToPath(c.path) + if err != nil { + return fmt.Errorf("failed to decode leaf path: %w", err) + } + + // Read payload length prefix (4 bytes) and payload bytes. + if _, err := io.ReadFull(reader, c.lenBuf); err != nil { + return fmt.Errorf("cannot read leaf payload length: %w", err) + } + size := binary.BigEndian.Uint32(c.lenBuf) + if uint32(cap(c.payload)) < size { + c.payload = make([]byte, size) + } + payloadBuf := c.payload[:size] + if _, err := io.ReadFull(reader, payloadBuf); err != nil { + return fmt.Errorf("cannot read leaf payload: %w", err) + } + + // DecodePayloadWithoutPrefix with zeroCopy=false returns a copy, so reusing + // payloadBuf on the next iteration is safe. + payload, err := ledger.DecodePayloadWithoutPrefix(payloadBuf, false, payloadEncodingVersion) + if err != nil { + return fmt.Errorf("failed to decode leaf payload: %w", err) + } + + // Reuse the tested V6→V7 leaf projection to keep a single source of truth for + // the leaf-hash / empty-payload handling. + v6leaf := node.NewNode(int(height), nil, nil, path, payload, nodeHash) + v7leaf, err := FromV6LeafNode(v6leaf) + if err != nil { + return fmt.Errorf("cannot convert leaf node: %w", err) + } + + // Optionally verify that the V7 node hash derived from (path, value) matches + // the V6 node hash carried on disk. This is the per-leaf, streaming equivalent + // of the forest-level root-hash cross-check in ConvertCheckpointV6ToV7. + if c.verifyLeafHash { + if derived := v7leaf.Hash(); derived != nodeHash { + return fmt.Errorf("leaf hash verification failed for path %x: derived node hash %x does not match V6 node hash %x", + path, derived, nodeHash) + } + } + + encoded := payloadless.EncodeNode(v7leaf, 0, 0, c.enc) + if _, err := writer.Write(encoded); err != nil { + return fmt.Errorf("cannot write converted leaf node: %w", err) + } + return nil +} diff --git a/ledger/complete/wal/checkpoint_v7_convert_stream_test.go b/ledger/complete/wal/checkpoint_v7_convert_stream_test.go new file mode 100644 index 00000000000..0a6dec8506b --- /dev/null +++ b/ledger/complete/wal/checkpoint_v7_convert_stream_test.go @@ -0,0 +1,186 @@ +package wal + +import ( + "bytes" + "fmt" + "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/flattener" + "github.com/onflow/flow-go/ledger/complete/mtrie/trie" + "github.com/onflow/flow-go/utils/unittest" +) + +// TestConvertCheckpointV6ToV7Stream_MatchesNonStream verifies that the streaming +// converter produces byte-identical V7 part files to the in-memory +// converter. Both preserve the V6 on-disk node ordering and use the same leaf +// projection and encoding, so their output must match exactly. The check is run +// with leaf-hash verification both off and on, since verification must not alter +// the output bytes. +func TestConvertCheckpointV6ToV7Stream_MatchesNonStream(t *testing.T) { + for _, verify := range []bool{false, true} { + t.Run(fmt.Sprintf("verifyLeafHash=%v", verify), func(t *testing.T) { + unittest.RunWithTempDir(t, func(dir string) { + logger := zerolog.Nop() + v6Tries := createMultipleRandomTries(t) + v6Name := "checkpoint.00000300" + require.NoError(t, StoreCheckpointV6Concurrently(v6Tries, dir, v6Name, logger)) + + // Path A: in-memory converter. + nonStreamName := v6Name + ".nonstream" + V7FileSuffix + require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, nonStreamName, logger, 16)) + + // Path B: streaming converter. + streamName := v6Name + ".stream" + V7FileSuffix + require.NoError(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, streamName, logger, 16, verify)) + + nonStreamFiles := filePaths(dir, nonStreamName, subtrieLevel) + streamFiles := filePaths(dir, streamName, subtrieLevel) + require.Equal(t, len(nonStreamFiles), len(streamFiles)) + for i, nf := range nonStreamFiles { + require.NoError(t, compareFiles(nf, streamFiles[i]), + "stream converter output differs from non-stream at part %d", i) + } + }) + }) + } +} + +// TestConvertCheckpointV6ToV7Stream_PreservesRootHashes writes a V6 checkpoint, +// runs the stream converter (with leaf-hash verification enabled), then reads the +// V7 result back and verifies every trie root hash matches. +func TestConvertCheckpointV6ToV7Stream_PreservesRootHashes(t *testing.T) { + unittest.RunWithTempDir(t, func(dir string) { + logger := zerolog.Nop() + v6Tries := createMultipleRandomTries(t) + v6Name := "checkpoint.00000301" + require.NoError(t, StoreCheckpointV6Concurrently(v6Tries, dir, v6Name, logger)) + + v7Name := v6Name + V7FileSuffix + require.NoError(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, v7Name, logger, 16, true)) + + v7Tries, err := OpenAndReadCheckpointV7(dir, v7Name, logger) + require.NoError(t, err) + require.Equal(t, len(v6Tries), len(v7Tries)) + for i, v6 := range v6Tries { + require.Equal(t, v6.RootHash(), v7Tries[i].RootHash(), "trie %d root hash mismatch", i) + } + }) +} + +// TestConvertCheckpointV6ToV7Stream_NWorkerVariants covers the minimum, an +// intermediate, and the maximum worker counts, with leaf-hash verification on. +func TestConvertCheckpointV6ToV7Stream_NWorkerVariants(t *testing.T) { + unittest.RunWithTempDir(t, func(dir string) { + logger := zerolog.Nop() + v6Tries := createMultipleRandomTries(t) + v6Name := "checkpoint.00000302" + require.NoError(t, StoreCheckpointV6Concurrently(v6Tries, dir, v6Name, logger)) + + for _, nWorker := range []uint{1, 3, 16} { + v7Name := fmt.Sprintf("%s.nw%d%s", v6Name, nWorker, V7FileSuffix) + require.NoError(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, v7Name, logger, nWorker, true)) + + v7Tries, err := OpenAndReadCheckpointV7(dir, v7Name, logger) + require.NoError(t, err) + for i, v6 := range v6Tries { + require.Equal(t, v6.RootHash(), v7Tries[i].RootHash(), + "trie %d root hash mismatch at nWorker=%d", i, nWorker) + } + } + }) +} + +// TestConvertCheckpointV6ToV7Stream_EmptyTrie verifies the stream converter handles +// an empty-trie checkpoint. +func TestConvertCheckpointV6ToV7Stream_EmptyTrie(t *testing.T) { + unittest.RunWithTempDir(t, func(dir string) { + logger := zerolog.Nop() + v6Tries := []*trie.MTrie{trie.NewEmptyMTrie()} + v6Name := "checkpoint.00000303" + require.NoError(t, StoreCheckpointV6Concurrently(v6Tries, dir, v6Name, logger)) + + v7Name := v6Name + V7FileSuffix + require.NoError(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, v7Name, logger, 16, true)) + + v7Tries, err := OpenAndReadCheckpointV7(dir, v7Name, logger) + require.NoError(t, err) + require.Len(t, v7Tries, 1) + require.True(t, v7Tries[0].IsEmpty()) + }) +} + +// TestV6ToV7NodeConverter_VerifyLeafHash confirms that the per-leaf verification +// path actually detects a tampered leaf. A leaf whose stored V6 node hash no +// longer matches its (path, value) is rejected when verification is on, and is +// silently converted when it is off (the streaming path otherwise re-derives the +// leaf hash from the payload and ignores the stored node hash). +func TestV6ToV7NodeConverter_VerifyLeafHash(t *testing.T) { + // A single-register trie compactifies to a leaf root, giving a real V6 leaf + // node with a correct node hash. + emptyTrie := trie.NewEmptyMTrie() + p := testutils.PathByUint8(7) + v := testutils.LightPayload8('A', 'a') + updated, _, err := trie.NewTrieWithUpdatedRegisters(emptyTrie, []ledger.Path{p}, []ledger.Payload{*v}, true) + require.NoError(t, err) + leaf := updated.RootNode() + require.True(t, leaf.IsLeaf(), "test setup expects a compactified leaf root") + + scratch := make([]byte, 1024*4) + shared := flattener.EncodeNode(leaf, 0, 0, scratch) + encoded := make([]byte, len(shared)) // copy: EncodeNode shares the scratch buffer + copy(encoded, shared) + + // A correct leaf converts cleanly with verification enabled. + var out bytes.Buffer + require.NoError(t, newV6ToV7NodeConverter(true).convertNode(bytes.NewReader(encoded), &out)) + + // Corrupt the last byte of the stored node hash (offset within the fixed + // prefix: type(1) + height(2) + hash(32)). + corrupt := make([]byte, len(encoded)) + copy(corrupt, encoded) + corrupt[fixedNodePrefixSize-1] ^= 0xFF + + // With verification on, the stored-vs-derived hash mismatch is detected. + var outVerify bytes.Buffer + err = newV6ToV7NodeConverter(true).convertNode(bytes.NewReader(corrupt), &outVerify) + require.Error(t, err) + require.Contains(t, err.Error(), "leaf hash verification failed") + + // With verification off, the corrupt node hash is ignored and conversion + // succeeds (the V7 leaf hash is derived from the intact payload). + var outNoVerify bytes.Buffer + require.NoError(t, newV6ToV7NodeConverter(false).convertNode(bytes.NewReader(corrupt), &outNoVerify)) +} + +// TestConvertCheckpointV6ToV7Stream_Validation verifies argument and filename +// validation: invalid worker counts, a non-V7 output filename, refusing to +// clobber an existing output, and a missing V6 input. +func TestConvertCheckpointV6ToV7Stream_Validation(t *testing.T) { + unittest.RunWithTempDir(t, func(dir string) { + logger := zerolog.Nop() + + require.Error(t, ConvertCheckpointV6ToV7Stream(dir, "x", dir, "out"+V7FileSuffix, logger, 0, false), + "nWorker=0 must be rejected") + require.Error(t, ConvertCheckpointV6ToV7Stream(dir, "x", dir, "out"+V7FileSuffix, logger, 17, false), + "nWorker > subtrieCount must be rejected") + require.Error(t, ConvertCheckpointV6ToV7Stream(dir, "missing", dir, "missing"+V7FileSuffix, logger, 4, false), + "missing V6 input must be reported") + + v6Tries := createSimpleTrie(t) + v6Name := "checkpoint.00000304" + require.NoError(t, StoreCheckpointV6Concurrently(v6Tries, dir, v6Name, logger)) + + require.Error(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, "no-suffix", logger, 4, false), + "output filename without V7 suffix must be rejected") + + v7Name := v6Name + V7FileSuffix + require.NoError(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, v7Name, logger, 4, false)) + require.Error(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, v7Name, logger, 4, false), + "second conversion to the same V7 output must be rejected") + }) +} From 1f10e17379f5505d0174299f93a17b53d89d99ef Mon Sep 17 00:00:00 2001 From: "Leo Zhang (zhangchiqing)" Date: Tue, 16 Jun 2026 16:10:35 -0700 Subject: [PATCH 2/4] remove temp files --- cmd/util/cmd/checkpoint-convert-v7/cmd.go | 4 +- ledger/complete/wal/checkpoint_v6_writer.go | 30 +++++++++ .../complete/wal/checkpoint_v6_writer_test.go | 66 +++++++++++++++++++ ledger/complete/wal/checkpoint_v7_convert.go | 6 ++ .../wal/checkpoint_v7_convert_stream.go | 6 ++ 5 files changed, 111 insertions(+), 1 deletion(-) create mode 100644 ledger/complete/wal/checkpoint_v6_writer_test.go diff --git a/cmd/util/cmd/checkpoint-convert-v7/cmd.go b/cmd/util/cmd/checkpoint-convert-v7/cmd.go index c2f9794468f..6d7784e447f 100644 --- a/cmd/util/cmd/checkpoint-convert-v7/cmd.go +++ b/cmd/util/cmd/checkpoint-convert-v7/cmd.go @@ -118,7 +118,9 @@ func run(*cobra.Command, []string) { log.Fatal().Err(err).Msg("checkpoint conversion failed") } - log.Info().Msgf("wrote V7 checkpoint to %s", filepath.Join(outputDir, outputFile)) + log.Info(). + Str("output", filepath.Join(outputDir, outputFile)). + Msg("✅ V6→V7 checkpoint conversion completed successfully") } // defaultV7Filename returns the default V7 output filename for a given V6 diff --git a/ledger/complete/wal/checkpoint_v6_writer.go b/ledger/complete/wal/checkpoint_v6_writer.go index b72eff4392e..3d2250906a5 100644 --- a/ledger/complete/wal/checkpoint_v6_writer.go +++ b/ledger/complete/wal/checkpoint_v6_writer.go @@ -582,6 +582,36 @@ func storeTries( return nil } +// removeStaleTempFiles removes leftover "writing-*" temporary part +// files in outputDir. +// +// createClosableWriter writes each checkpoint part to such a temp file and renames +// it to the target on success (or removes it on a handled write error). A process +// killed mid-write — e.g. OOM or Ctrl-C — leaves the temp file behind, and a +// subsequent run uses a fresh random suffix rather than reusing it, so orphaned +// temp files accumulate. Removing them at the start of a run reclaims that space. +// +// Only temp files for outputFile are matched. Final part files lack the "writing-" +// prefix and so are never touched. +// +// No error returns are expected during normal operation. +func removeStaleTempFiles(outputDir string, outputFile string, logger zerolog.Logger) error { + pattern := path.Join(outputDir, fmt.Sprintf("writing-%v*", outputFile)) + filesToRemove, err := filepath.Glob(pattern) + if err != nil { + return fmt.Errorf("could not glob stale temp files with pattern %v: %w", pattern, err) + } + + for _, file := range filesToRemove { + if err := os.Remove(file); err != nil { + return fmt.Errorf("could not remove stale temp file %v: %w", file, err) + } + logger.Info().Msgf("removed stale checkpoint temp file %v", file) + } + + return nil +} + // deleteCheckpointFiles removes any checkpoint files with given checkpoint prefix in the outputDir. func deleteCheckpointFiles(outputDir string, outputFile string) error { pattern := filePathPattern(outputDir, outputFile) diff --git a/ledger/complete/wal/checkpoint_v6_writer_test.go b/ledger/complete/wal/checkpoint_v6_writer_test.go new file mode 100644 index 00000000000..fe0b8f158ca --- /dev/null +++ b/ledger/complete/wal/checkpoint_v6_writer_test.go @@ -0,0 +1,66 @@ +package wal + +import ( + "os" + "path" + "testing" + + "github.com/rs/zerolog" + "github.com/stretchr/testify/require" + + "github.com/onflow/flow-go/utils/unittest" +) + +// TestRemoveStaleTempFiles verifies that removeStaleTempFiles deletes only the +// "writing-*" temp files for the given output, while leaving final +// part files, the header, and temp files belonging to other outputs untouched. +func TestRemoveStaleTempFiles(t *testing.T) { + unittest.RunWithTempDir(t, func(dir string) { + outputFile := "root.checkpoint.v7" + + // Stale temp files for outputFile: subtries, top-trie, and header. + // These mirror the names produced by createClosableWriter + // ("writing--"). + staleTempFiles := []string{ + "writing-root.checkpoint.v7.000-1234567890", + "writing-root.checkpoint.v7.000-9876543210", // a second orphan for the same part + "writing-root.checkpoint.v7.016-1720029787", // top-trie part + "writing-root.checkpoint.v7-246069680", // header + } + + // Files that must NOT be removed: final part files, the header, and a temp + // file for a different output (e.g. a V6 checkpoint with a different name). + keepFiles := []string{ + "root.checkpoint.v7", // final header + "root.checkpoint.v7.000", // final subtrie part + "root.checkpoint.v7.016", // final top-trie part + "writing-root.checkpoint.v6.000-111222333", // temp for a different output + "root.checkpoint.v6", // unrelated final file + } + + for _, name := range append(append([]string{}, staleTempFiles...), keepFiles...) { + require.NoError(t, os.WriteFile(path.Join(dir, name), []byte("x"), 0644)) + } + + require.NoError(t, removeStaleTempFiles(dir, outputFile, zerolog.Nop())) + + for _, name := range staleTempFiles { + require.NoFileExists(t, path.Join(dir, name), "stale temp file should have been removed: %s", name) + } + for _, name := range keepFiles { + require.FileExists(t, path.Join(dir, name), "file should have been kept: %s", name) + } + }) +} + +// TestRemoveStaleTempFiles_NoMatches verifies that removeStaleTempFiles is a +// no-op (no error) when there are no matching temp files. +func TestRemoveStaleTempFiles_NoMatches(t *testing.T) { + unittest.RunWithTempDir(t, func(dir string) { + require.NoError(t, os.WriteFile(path.Join(dir, "root.checkpoint.v7.000"), []byte("x"), 0644)) + + require.NoError(t, removeStaleTempFiles(dir, "root.checkpoint.v7", zerolog.Nop())) + + require.FileExists(t, path.Join(dir, "root.checkpoint.v7.000")) + }) +} diff --git a/ledger/complete/wal/checkpoint_v7_convert.go b/ledger/complete/wal/checkpoint_v7_convert.go index 689a2dea316..4c11c122738 100644 --- a/ledger/complete/wal/checkpoint_v7_convert.go +++ b/ledger/complete/wal/checkpoint_v7_convert.go @@ -192,6 +192,12 @@ func ConvertCheckpointV6ToV7( return fmt.Errorf("V7 output already exists: %v", v7Existing) } + // Remove any leftover temp part files from a previously interrupted conversion + // to this output; they are never reused and would otherwise accumulate. + if err := removeStaleTempFiles(outputDir, outputFileName, logger); err != nil { + return fmt.Errorf("could not remove stale temp files: %w", err) + } + logger.Info(). Str("v6_dir", inputDir). Str("v6_file", inputFileName). diff --git a/ledger/complete/wal/checkpoint_v7_convert_stream.go b/ledger/complete/wal/checkpoint_v7_convert_stream.go index 8eeaca4eac6..1db23065c16 100644 --- a/ledger/complete/wal/checkpoint_v7_convert_stream.go +++ b/ledger/complete/wal/checkpoint_v7_convert_stream.go @@ -147,6 +147,12 @@ func convertCheckpointV6ToV7Stream( return fmt.Errorf("V7 output already exists: %v", v7Existing) } + // Remove any leftover temp part files from a previously interrupted conversion + // to this output; they are never reused and would otherwise accumulate. + if err := removeStaleTempFiles(outputDir, outputFileName, logger); err != nil { + return fmt.Errorf("could not remove stale temp files: %w", err) + } + logger.Info(). Str("v6_dir", inputDir). Str("v6_file", inputFileName). From a2d38efa001ca5d27cc603a75de0c035d497d5c0 Mon Sep 17 00:00:00 2001 From: "Leo Zhang (zhangchiqing)" Date: Wed, 17 Jun 2026 20:01:38 -0700 Subject: [PATCH 3/4] remove verify leaf hash flag --- cmd/util/cmd/checkpoint-convert-v7/cmd.go | 13 -- .../wal/checkpoint_v7_convert_stream.go | 64 +++------ .../wal/checkpoint_v7_convert_stream_test.go | 123 +++++------------- 3 files changed, 50 insertions(+), 150 deletions(-) diff --git a/cmd/util/cmd/checkpoint-convert-v7/cmd.go b/cmd/util/cmd/checkpoint-convert-v7/cmd.go index 6d7784e447f..a082b17a8c2 100644 --- a/cmd/util/cmd/checkpoint-convert-v7/cmd.go +++ b/cmd/util/cmd/checkpoint-convert-v7/cmd.go @@ -17,7 +17,6 @@ var ( flagOutput string flagNWorker uint flagStream bool - flagVerifyLeafHash bool ) // Cmd converts a V6 checkpoint to a V7 (payloadless) checkpoint by reading @@ -60,11 +59,6 @@ func init() { Cmd.Flags().BoolVar(&flagStream, "stream", false, "stream part files node-by-node instead of loading the full trie forest into memory "+ "(constant memory, preserves node hashes without re-deriving root hashes)") - - Cmd.Flags().BoolVar(&flagVerifyLeafHash, "verify-leaf-hash", false, - "in --stream mode, verify every allocated leaf by comparing the derived V7 node hash "+ - "against the V6 node hash on disk (ignored without --stream, which already cross-checks "+ - "root hashes)") } func run(*cobra.Command, []string) { @@ -85,7 +79,6 @@ func run(*cobra.Command, []string) { Str("output", outputFile). Uint("nworker", flagNWorker). Bool("stream", flagStream). - Bool("verify_leaf_hash", flagVerifyLeafHash). Msg("converting V6 checkpoint to V7") var err error @@ -97,14 +90,8 @@ func run(*cobra.Command, []string) { outputFile, log.Logger, flagNWorker, - flagVerifyLeafHash, ) } else { - if flagVerifyLeafHash { - // The in-memory converter already cross-checks every trie root hash, - // which transitively verifies all leaf hashes, so the flag is a no-op here. - log.Warn().Msg("--verify-leaf-hash has no effect without --stream; the in-memory converter already verifies root hashes") - } err = wal.ConvertCheckpointV6ToV7( flagCheckpointDir, flagCheckpoint, diff --git a/ledger/complete/wal/checkpoint_v7_convert_stream.go b/ledger/complete/wal/checkpoint_v7_convert_stream.go index 1db23065c16..9d454062080 100644 --- a/ledger/complete/wal/checkpoint_v7_convert_stream.go +++ b/ledger/complete/wal/checkpoint_v7_convert_stream.go @@ -72,21 +72,12 @@ const ( // carried over verbatim from the V6 stream, so root hashes are structurally // preserved. // -// When verifyLeafHash is true, every allocated leaf is additionally checked: the -// node hash derived from (path, value) is compared against the leaf's V6 node hash -// read from disk, and a mismatch aborts the conversion. This is the streaming, -// per-leaf equivalent of the root-hash cross-check performed by the in-memory -// [ConvertCheckpointV6ToV7] (a wrong leaf hash would otherwise only surface as a -// root-hash mismatch when the forest is loaded). It adds a hash recomputation per -// allocated leaf but no extra memory. -// // The output filename must carry the V7 suffix and no output part file may already // exist; otherwise the call is rejected. On any failure, partially written output // files are removed. // // No error returns are expected during normal operation; all error returns indicate -// a malformed input, a clobbering output, an IO failure, or a leaf-hash mismatch -// when verifyLeafHash is enabled. +// a malformed input, a clobbering output, or an IO failure. func ConvertCheckpointV6ToV7Stream( inputDir string, inputFileName string, @@ -94,9 +85,8 @@ func ConvertCheckpointV6ToV7Stream( outputFileName string, logger zerolog.Logger, nWorker uint, - verifyLeafHash bool, ) error { - err := convertCheckpointV6ToV7Stream(inputDir, inputFileName, outputDir, outputFileName, logger, nWorker, verifyLeafHash) + err := convertCheckpointV6ToV7Stream(inputDir, inputFileName, outputDir, outputFileName, logger, nWorker) if err != nil { cleanupErr := deleteCheckpointFiles(outputDir, outputFileName) if cleanupErr != nil { @@ -114,7 +104,6 @@ func convertCheckpointV6ToV7Stream( outputFileName string, logger zerolog.Logger, nWorker uint, - verifyLeafHash bool, ) error { if nWorker == 0 || nWorker > subtrieCount { return fmt.Errorf("invalid nWorker %v, valid range is [1, %v]", nWorker, subtrieCount) @@ -159,19 +148,18 @@ func convertCheckpointV6ToV7Stream( Str("v7_dir", outputDir). Str("v7_file", outputFileName). Uint("nworker", nWorker). - Bool("verify_leaf_hash", verifyLeafHash). Msg("starting streaming V6→V7 checkpoint conversion") // Convert the 16 subtrie part files concurrently, recomputing each checksum. newSubtrieChecksums, err := convertSubTriesV6ToV7StreamConcurrently( - inputDir, inputFileName, outputDir, outputFileName, subtrieChecksums, logger, nWorker, verifyLeafHash) + inputDir, inputFileName, outputDir, outputFileName, subtrieChecksums, logger, nWorker) if err != nil { return fmt.Errorf("could not convert subtrie files: %w", err) } // Convert the top-trie part file. newTopTrieChecksum, err := convertTopTrieFileV6ToV7Stream( - inputDir, inputFileName, outputDir, outputFileName, topTrieChecksum, logger, verifyLeafHash) + inputDir, inputFileName, outputDir, outputFileName, topTrieChecksum, logger) if err != nil { return fmt.Errorf("could not convert top-trie file: %w", err) } @@ -202,7 +190,6 @@ func convertSubTriesV6ToV7StreamConcurrently( subtrieChecksums []uint32, logger zerolog.Logger, nWorker uint, - verifyLeafHash bool, ) ([]uint32, error) { jobs := make(chan int, subtrieCount) for i := 0; i < subtrieCount; i++ { @@ -218,7 +205,7 @@ func convertSubTriesV6ToV7StreamConcurrently( go func() { for i := range jobs { sum, err := convertSubTrieFileV6ToV7Stream( - inputDir, inputFileName, outputDir, outputFileName, i, subtrieChecksums[i], logger, verifyLeafHash) + inputDir, inputFileName, outputDir, outputFileName, i, subtrieChecksums[i], logger) results <- streamSubtrieResult{index: i, checksum: sum, err: err} } }() @@ -248,7 +235,6 @@ func convertSubTrieFileV6ToV7Stream( index int, expectedSum uint32, logger zerolog.Logger, - verifyLeafHash bool, ) (checksum uint32, errToReturn error) { inPath, _, err := filePathSubTries(inputDir, inputFileName, index) if err != nil { @@ -294,7 +280,7 @@ func convertSubTrieFileV6ToV7Stream( } logging := logProgress(fmt.Sprintf("converting %v-th sub trie (streaming)", index), int(nodeCount), logger) - conv := newV6ToV7NodeConverter(verifyLeafHash) + conv := newV6ToV7NodeConverter() for i := uint64(0); i < nodeCount; i++ { if err := conv.convertNode(reader, writer); err != nil { return 0, fmt.Errorf("cannot convert node %d of subtrie %d: %w", i, index, err) @@ -322,7 +308,6 @@ func convertTopTrieFileV6ToV7Stream( outputFileName string, expectedSum uint32, logger zerolog.Logger, - verifyLeafHash bool, ) (checksum uint32, errToReturn error) { inPath, _ := filePathTopTries(inputDir, inputFileName) @@ -374,7 +359,7 @@ func convertTopTrieFileV6ToV7Stream( } // Convert the top-level nodes (above subtrieLevel). - conv := newV6ToV7NodeConverter(verifyLeafHash) + conv := newV6ToV7NodeConverter() for i := uint64(0); i < topLevelNodesCount; i++ { if err := conv.convertNode(reader, writer); err != nil { return 0, fmt.Errorf("cannot convert top-level node %d: %w", i, err) @@ -421,22 +406,17 @@ type v6ToV7NodeConverter struct { lenBuf []byte // leaf payload length prefix payload []byte // leaf payload bytes (grows as needed) enc []byte // scratch for the payloadless leaf encoding - - // verifyLeafHash, when true, makes convertLeaf compare each derived V7 leaf - // node hash against the V6 leaf node hash read from disk. - verifyLeafHash bool } // newV6ToV7NodeConverter returns a converter with preallocated scratch buffers. -func newV6ToV7NodeConverter(verifyLeafHash bool) *v6ToV7NodeConverter { +func newV6ToV7NodeConverter() *v6ToV7NodeConverter { return &v6ToV7NodeConverter{ - prefix: make([]byte, fixedNodePrefixSize), - childIndex: make([]byte, 2*encNodeIndexSize), - path: make([]byte, encPathSize), - lenBuf: make([]byte, encPayloadLengthSize), - payload: make([]byte, 1024), - enc: make([]byte, 1024*4), - verifyLeafHash: verifyLeafHash, + prefix: make([]byte, fixedNodePrefixSize), + childIndex: make([]byte, 2*encNodeIndexSize), + path: make([]byte, encPathSize), + lenBuf: make([]byte, encPayloadLengthSize), + payload: make([]byte, 1024), + enc: make([]byte, 1024*4), } } @@ -478,12 +458,8 @@ func (c *v6ToV7NodeConverter) convertNode(reader io.Reader, writer io.Writer) er // having already consumed the shared prefix into c.prefix, and writes its V7 // payloadless encoding to writer. // -// When c.verifyLeafHash is set, the V7 node hash derived from (path, value) is -// compared against the V6 node hash read from disk, and a mismatch is reported as -// an error. -// // No error returns are expected during normal operation; all error returns indicate -// a malformed input stream, an IO failure, or (when verifying) a leaf-hash mismatch. +// a malformed input stream or an IO failure. func (c *v6ToV7NodeConverter) convertLeaf(reader io.Reader, writer io.Writer) error { height := binary.BigEndian.Uint16(c.prefix[encNodeTypeSize:]) nodeHash, err := hash.ToHash(c.prefix[encNodeTypeSize+encHeightSize:]) @@ -528,16 +504,6 @@ func (c *v6ToV7NodeConverter) convertLeaf(reader io.Reader, writer io.Writer) er return fmt.Errorf("cannot convert leaf node: %w", err) } - // Optionally verify that the V7 node hash derived from (path, value) matches - // the V6 node hash carried on disk. This is the per-leaf, streaming equivalent - // of the forest-level root-hash cross-check in ConvertCheckpointV6ToV7. - if c.verifyLeafHash { - if derived := v7leaf.Hash(); derived != nodeHash { - return fmt.Errorf("leaf hash verification failed for path %x: derived node hash %x does not match V6 node hash %x", - path, derived, nodeHash) - } - } - encoded := payloadless.EncodeNode(v7leaf, 0, 0, c.enc) if _, err := writer.Write(encoded); err != nil { return fmt.Errorf("cannot write converted leaf node: %w", err) diff --git a/ledger/complete/wal/checkpoint_v7_convert_stream_test.go b/ledger/complete/wal/checkpoint_v7_convert_stream_test.go index 0a6dec8506b..77d7db328d6 100644 --- a/ledger/complete/wal/checkpoint_v7_convert_stream_test.go +++ b/ledger/complete/wal/checkpoint_v7_convert_stream_test.go @@ -1,16 +1,12 @@ package wal import ( - "bytes" "fmt" "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/flattener" "github.com/onflow/flow-go/ledger/complete/mtrie/trie" "github.com/onflow/flow-go/utils/unittest" ) @@ -18,41 +14,35 @@ import ( // TestConvertCheckpointV6ToV7Stream_MatchesNonStream verifies that the streaming // converter produces byte-identical V7 part files to the in-memory // converter. Both preserve the V6 on-disk node ordering and use the same leaf -// projection and encoding, so their output must match exactly. The check is run -// with leaf-hash verification both off and on, since verification must not alter -// the output bytes. +// projection and encoding, so their output must match exactly. func TestConvertCheckpointV6ToV7Stream_MatchesNonStream(t *testing.T) { - for _, verify := range []bool{false, true} { - t.Run(fmt.Sprintf("verifyLeafHash=%v", verify), func(t *testing.T) { - unittest.RunWithTempDir(t, func(dir string) { - logger := zerolog.Nop() - v6Tries := createMultipleRandomTries(t) - v6Name := "checkpoint.00000300" - require.NoError(t, StoreCheckpointV6Concurrently(v6Tries, dir, v6Name, logger)) - - // Path A: in-memory converter. - nonStreamName := v6Name + ".nonstream" + V7FileSuffix - require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, nonStreamName, logger, 16)) - - // Path B: streaming converter. - streamName := v6Name + ".stream" + V7FileSuffix - require.NoError(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, streamName, logger, 16, verify)) - - nonStreamFiles := filePaths(dir, nonStreamName, subtrieLevel) - streamFiles := filePaths(dir, streamName, subtrieLevel) - require.Equal(t, len(nonStreamFiles), len(streamFiles)) - for i, nf := range nonStreamFiles { - require.NoError(t, compareFiles(nf, streamFiles[i]), - "stream converter output differs from non-stream at part %d", i) - } - }) - }) - } + unittest.RunWithTempDir(t, func(dir string) { + logger := zerolog.Nop() + v6Tries := createMultipleRandomTries(t) + v6Name := "checkpoint.00000300" + require.NoError(t, StoreCheckpointV6Concurrently(v6Tries, dir, v6Name, logger)) + + // Path A: in-memory converter. + nonStreamName := v6Name + ".nonstream" + V7FileSuffix + require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, nonStreamName, logger, 16)) + + // Path B: streaming converter. + streamName := v6Name + ".stream" + V7FileSuffix + require.NoError(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, streamName, logger, 16)) + + nonStreamFiles := filePaths(dir, nonStreamName, subtrieLevel) + streamFiles := filePaths(dir, streamName, subtrieLevel) + require.Equal(t, len(nonStreamFiles), len(streamFiles)) + for i, nf := range nonStreamFiles { + require.NoError(t, compareFiles(nf, streamFiles[i]), + "stream converter output differs from non-stream at part %d", i) + } + }) } // TestConvertCheckpointV6ToV7Stream_PreservesRootHashes writes a V6 checkpoint, -// runs the stream converter (with leaf-hash verification enabled), then reads the -// V7 result back and verifies every trie root hash matches. +// runs the stream converter, then reads the V7 result back and verifies every +// trie root hash matches. func TestConvertCheckpointV6ToV7Stream_PreservesRootHashes(t *testing.T) { unittest.RunWithTempDir(t, func(dir string) { logger := zerolog.Nop() @@ -61,7 +51,7 @@ func TestConvertCheckpointV6ToV7Stream_PreservesRootHashes(t *testing.T) { require.NoError(t, StoreCheckpointV6Concurrently(v6Tries, dir, v6Name, logger)) v7Name := v6Name + V7FileSuffix - require.NoError(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, v7Name, logger, 16, true)) + require.NoError(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, v7Name, logger, 16)) v7Tries, err := OpenAndReadCheckpointV7(dir, v7Name, logger) require.NoError(t, err) @@ -73,7 +63,7 @@ func TestConvertCheckpointV6ToV7Stream_PreservesRootHashes(t *testing.T) { } // TestConvertCheckpointV6ToV7Stream_NWorkerVariants covers the minimum, an -// intermediate, and the maximum worker counts, with leaf-hash verification on. +// intermediate, and the maximum worker counts. func TestConvertCheckpointV6ToV7Stream_NWorkerVariants(t *testing.T) { unittest.RunWithTempDir(t, func(dir string) { logger := zerolog.Nop() @@ -83,7 +73,7 @@ func TestConvertCheckpointV6ToV7Stream_NWorkerVariants(t *testing.T) { for _, nWorker := range []uint{1, 3, 16} { v7Name := fmt.Sprintf("%s.nw%d%s", v6Name, nWorker, V7FileSuffix) - require.NoError(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, v7Name, logger, nWorker, true)) + require.NoError(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, v7Name, logger, nWorker)) v7Tries, err := OpenAndReadCheckpointV7(dir, v7Name, logger) require.NoError(t, err) @@ -105,7 +95,7 @@ func TestConvertCheckpointV6ToV7Stream_EmptyTrie(t *testing.T) { require.NoError(t, StoreCheckpointV6Concurrently(v6Tries, dir, v6Name, logger)) v7Name := v6Name + V7FileSuffix - require.NoError(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, v7Name, logger, 16, true)) + require.NoError(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, v7Name, logger, 16)) v7Tries, err := OpenAndReadCheckpointV7(dir, v7Name, logger) require.NoError(t, err) @@ -114,49 +104,6 @@ func TestConvertCheckpointV6ToV7Stream_EmptyTrie(t *testing.T) { }) } -// TestV6ToV7NodeConverter_VerifyLeafHash confirms that the per-leaf verification -// path actually detects a tampered leaf. A leaf whose stored V6 node hash no -// longer matches its (path, value) is rejected when verification is on, and is -// silently converted when it is off (the streaming path otherwise re-derives the -// leaf hash from the payload and ignores the stored node hash). -func TestV6ToV7NodeConverter_VerifyLeafHash(t *testing.T) { - // A single-register trie compactifies to a leaf root, giving a real V6 leaf - // node with a correct node hash. - emptyTrie := trie.NewEmptyMTrie() - p := testutils.PathByUint8(7) - v := testutils.LightPayload8('A', 'a') - updated, _, err := trie.NewTrieWithUpdatedRegisters(emptyTrie, []ledger.Path{p}, []ledger.Payload{*v}, true) - require.NoError(t, err) - leaf := updated.RootNode() - require.True(t, leaf.IsLeaf(), "test setup expects a compactified leaf root") - - scratch := make([]byte, 1024*4) - shared := flattener.EncodeNode(leaf, 0, 0, scratch) - encoded := make([]byte, len(shared)) // copy: EncodeNode shares the scratch buffer - copy(encoded, shared) - - // A correct leaf converts cleanly with verification enabled. - var out bytes.Buffer - require.NoError(t, newV6ToV7NodeConverter(true).convertNode(bytes.NewReader(encoded), &out)) - - // Corrupt the last byte of the stored node hash (offset within the fixed - // prefix: type(1) + height(2) + hash(32)). - corrupt := make([]byte, len(encoded)) - copy(corrupt, encoded) - corrupt[fixedNodePrefixSize-1] ^= 0xFF - - // With verification on, the stored-vs-derived hash mismatch is detected. - var outVerify bytes.Buffer - err = newV6ToV7NodeConverter(true).convertNode(bytes.NewReader(corrupt), &outVerify) - require.Error(t, err) - require.Contains(t, err.Error(), "leaf hash verification failed") - - // With verification off, the corrupt node hash is ignored and conversion - // succeeds (the V7 leaf hash is derived from the intact payload). - var outNoVerify bytes.Buffer - require.NoError(t, newV6ToV7NodeConverter(false).convertNode(bytes.NewReader(corrupt), &outNoVerify)) -} - // TestConvertCheckpointV6ToV7Stream_Validation verifies argument and filename // validation: invalid worker counts, a non-V7 output filename, refusing to // clobber an existing output, and a missing V6 input. @@ -164,23 +111,23 @@ func TestConvertCheckpointV6ToV7Stream_Validation(t *testing.T) { unittest.RunWithTempDir(t, func(dir string) { logger := zerolog.Nop() - require.Error(t, ConvertCheckpointV6ToV7Stream(dir, "x", dir, "out"+V7FileSuffix, logger, 0, false), + require.Error(t, ConvertCheckpointV6ToV7Stream(dir, "x", dir, "out"+V7FileSuffix, logger, 0), "nWorker=0 must be rejected") - require.Error(t, ConvertCheckpointV6ToV7Stream(dir, "x", dir, "out"+V7FileSuffix, logger, 17, false), + require.Error(t, ConvertCheckpointV6ToV7Stream(dir, "x", dir, "out"+V7FileSuffix, logger, 17), "nWorker > subtrieCount must be rejected") - require.Error(t, ConvertCheckpointV6ToV7Stream(dir, "missing", dir, "missing"+V7FileSuffix, logger, 4, false), + require.Error(t, ConvertCheckpointV6ToV7Stream(dir, "missing", dir, "missing"+V7FileSuffix, logger, 4), "missing V6 input must be reported") v6Tries := createSimpleTrie(t) v6Name := "checkpoint.00000304" require.NoError(t, StoreCheckpointV6Concurrently(v6Tries, dir, v6Name, logger)) - require.Error(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, "no-suffix", logger, 4, false), + require.Error(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, "no-suffix", logger, 4), "output filename without V7 suffix must be rejected") v7Name := v6Name + V7FileSuffix - require.NoError(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, v7Name, logger, 4, false)) - require.Error(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, v7Name, logger, 4, false), + require.NoError(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, v7Name, logger, 4)) + require.Error(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, v7Name, logger, 4), "second conversion to the same V7 output must be rejected") }) } From 91eb0a7cbeef425f82c410eab8baf186134904de Mon Sep 17 00:00:00 2001 From: "Leo Zhang (zhangchiqing)" Date: Thu, 18 Jun 2026 09:08:53 -0700 Subject: [PATCH 4/4] fix lint --- cmd/util/cmd/checkpoint-convert-v7/cmd.go | 12 ++++++------ integration/localnet/builder/bootstrap.go | 1 - 2 files changed, 6 insertions(+), 7 deletions(-) diff --git a/cmd/util/cmd/checkpoint-convert-v7/cmd.go b/cmd/util/cmd/checkpoint-convert-v7/cmd.go index a082b17a8c2..4780be2755c 100644 --- a/cmd/util/cmd/checkpoint-convert-v7/cmd.go +++ b/cmd/util/cmd/checkpoint-convert-v7/cmd.go @@ -11,12 +11,12 @@ import ( ) var ( - flagCheckpointDir string - flagCheckpoint string - flagOutputDir string - flagOutput string - flagNWorker uint - flagStream bool + flagCheckpointDir string + flagCheckpoint string + flagOutputDir string + flagOutput string + flagNWorker uint + flagStream bool ) // Cmd converts a V6 checkpoint to a V7 (payloadless) checkpoint by reading diff --git a/integration/localnet/builder/bootstrap.go b/integration/localnet/builder/bootstrap.go index a9a5ee5d635..69e90ea8e83 100644 --- a/integration/localnet/builder/bootstrap.go +++ b/integration/localnet/builder/bootstrap.go @@ -1024,4 +1024,3 @@ func prepareTestExecutionService(dockerServices Services, flowNodeContainerConfi return dockerServices } -