From 852bb7c3f2b13ebcf9846b066a08bdc62b1e9bd4 Mon Sep 17 00:00:00 2001 From: "Leo Zhang (zhangchiqing)" Date: Mon, 15 Jun 2026 11:41:02 -0700 Subject: [PATCH 1/5] 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 ca072b2f4f67b9b4a8ae746f6ec0df532ba91e06 Mon Sep 17 00:00:00 2001 From: "Leo Zhang (zhangchiqing)" Date: Tue, 16 Jun 2026 16:10:35 -0700 Subject: [PATCH 2/5] 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 15cf534cdb99b5411ec630101961007c2d08eb25 Mon Sep 17 00:00:00 2001 From: "Leo Zhang (zhangchiqing)" Date: Wed, 17 Jun 2026 20:01:38 -0700 Subject: [PATCH 3/5] 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 1c62fec0aca2e3bccb60149d7dfc766e510bb7c7 Mon Sep 17 00:00:00 2001 From: "Leo Zhang (zhangchiqing)" Date: Thu, 18 Jun 2026 09:08:53 -0700 Subject: [PATCH 4/5] fix lint --- cmd/util/cmd/checkpoint-convert-v7/cmd.go | 12 ++++++------ 1 file changed, 6 insertions(+), 6 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 From b82aac51bc6f13d74f27115baad87f34a57d2db1 Mon Sep 17 00:00:00 2001 From: "Leo Zhang (zhangchiqing)" Date: Thu, 30 Jul 2026 21:30:53 -0700 Subject: [PATCH 5/5] address review comments: version-aware latest-checkpoint path, partial V7 conversion cleanup at bootstrap, logger and godoc fixes, and tests for the leaf-hash flag and V7 subtrie paths --- cmd/execution_builder.go | 1 + cmd/util/cmd/checkpoint-convert-v7/cmd.go | 29 +-- integration/localnet/builder/bootstrap.go | 1 + ledger/complete/wal/checkpoint_v6_test.go | 39 ++-- ledger/complete/wal/checkpoint_v7_convert.go | 130 +++++++++++--- .../wal/checkpoint_v7_convert_stream.go | 166 ++++++++++-------- .../wal/checkpoint_v7_convert_stream_test.go | 132 ++++++++++++-- .../wal/checkpoint_v7_convert_test.go | 36 ++-- ledger/factory/factory_test.go | 4 +- 9 files changed, 376 insertions(+), 162 deletions(-) diff --git a/cmd/execution_builder.go b/cmd/execution_builder.go index d0b90e2c159..0f74ac9c8b5 100644 --- a/cmd/execution_builder.go +++ b/cmd/execution_builder.go @@ -1531,6 +1531,7 @@ func (exeNode *ExecutionNode) LoadBootstrapper(node *NodeConfig) error { v7RootFileName, node.Logger, 16, + false, ) if err != nil { return fmt.Errorf("could not convert V6 root checkpoint to V7 for payloadless node: %w", err) diff --git a/cmd/util/cmd/checkpoint-convert-v7/cmd.go b/cmd/util/cmd/checkpoint-convert-v7/cmd.go index 4780be2755c..b0bf76d57bc 100644 --- a/cmd/util/cmd/checkpoint-convert-v7/cmd.go +++ b/cmd/util/cmd/checkpoint-convert-v7/cmd.go @@ -81,26 +81,15 @@ func run(*cobra.Command, []string) { Bool("stream", flagStream). Msg("converting V6 checkpoint to V7") - var err error - if flagStream { - err = wal.ConvertCheckpointV6ToV7Stream( - flagCheckpointDir, - flagCheckpoint, - outputDir, - outputFile, - log.Logger, - flagNWorker, - ) - } else { - err = wal.ConvertCheckpointV6ToV7( - flagCheckpointDir, - flagCheckpoint, - outputDir, - outputFile, - log.Logger, - flagNWorker, - ) - } + err := wal.ConvertCheckpointV6ToV7( + flagCheckpointDir, + flagCheckpoint, + outputDir, + outputFile, + log.Logger, + flagNWorker, + flagStream, + ) if err != nil { log.Fatal().Err(err).Msg("checkpoint conversion failed") } diff --git a/integration/localnet/builder/bootstrap.go b/integration/localnet/builder/bootstrap.go index b7e9b91c1a3..61a4efc6dc2 100644 --- a/integration/localnet/builder/bootstrap.go +++ b/integration/localnet/builder/bootstrap.go @@ -890,6 +890,7 @@ func prepareLedgerService(dockerServices Services, flowNodeContainerConfigs []te v7Filename, logger, 16, + false, ); convertErr != nil { panic(fmt.Errorf("failed to convert V6 root checkpoint to V7 for payloadless ledger service: %w", convertErr)) } diff --git a/ledger/complete/wal/checkpoint_v6_test.go b/ledger/complete/wal/checkpoint_v6_test.go index 1e036d3adf6..d2a0b64129c 100644 --- a/ledger/complete/wal/checkpoint_v6_test.go +++ b/ledger/complete/wal/checkpoint_v6_test.go @@ -448,9 +448,9 @@ func compareFiles(file1, file2 string) error { f.Close() }(closable1) - closable2, err := os.Open(file1) + closable2, err := os.Open(file2) if err != nil { - return fmt.Errorf("could not open file 2 %v: %w", closable2, err) + return fmt.Errorf("could not open file 2 %v: %w", file2, err) } defer func(f *os.File) { f.Close() @@ -462,25 +462,38 @@ func compareFiles(file1, file2 string) error { buf1 := make([]byte, defaultBufioReadSize) buf2 := make([]byte, defaultBufioReadSize) for { - _, err1 := reader1.Read(buf1) - _, err2 := reader2.Read(buf2) - if errors.Is(err1, io.EOF) && errors.Is(err2, io.EOF) { - break + // io.ReadFull fills the entire buffer unless the file ends, so the number of + // bytes read only differs between the two files when their sizes differ + n1, err1 := io.ReadFull(reader1, buf1) + n2, err2 := io.ReadFull(reader2, buf2) + + if !bytes.Equal(buf1[:n1], buf2[:n2]) { + return fmt.Errorf("bytes are different: %x, %x", buf1[:n1], buf2[:n2]) + } + + // both files ended at the same offset with identical content + if isEOF(err1) && isEOF(err2) { + return nil } - if err1 != nil { - return err1 + if err1 != nil && !isEOF(err1) { + return fmt.Errorf("could not read file 1 %v: %w", file1, err1) } - if err2 != nil { - return err2 + if err2 != nil && !isEOF(err2) { + return fmt.Errorf("could not read file 2 %v: %w", file2, err2) } - if !bytes.Equal(buf1, buf2) { - return fmt.Errorf("bytes are different: %x, %x", buf1, buf2) + // exactly one of the files ended here, so they have different lengths + if isEOF(err1) != isEOF(err2) { + return fmt.Errorf("files have different length: %v, %v", file1, file2) } } +} - return nil +// isEOF returns true if the given error signals that the end of the file was reached, +// which io.ReadFull reports as io.EOF (nothing read) or io.ErrUnexpectedEOF (partial read). +func isEOF(err error) bool { + return errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) } func storeCheckpointV5(tries []*trie.MTrie, dir string, fileName string, logger zerolog.Logger) error { diff --git a/ledger/complete/wal/checkpoint_v7_convert.go b/ledger/complete/wal/checkpoint_v7_convert.go index 4c11c122738..c0ebfac3b5a 100644 --- a/ledger/complete/wal/checkpoint_v7_convert.go +++ b/ledger/complete/wal/checkpoint_v7_convert.go @@ -134,22 +134,28 @@ func FromV6Tries(v6Tries []*trie.MTrie) ([]*payloadless.MTrie, error) { // - The output filename must use the V7 suffix (e.g. "checkpoint.00000100.v7"); // a missing or wrong suffix is rejected. // - No output file (including any part file) with the same name may already -// exist; otherwise the call is rejected. +// exist; otherwise the call is rejected and the existing output is left +// untouched. // - The conversion preserves trie root hashes: a V7 checkpoint round-tripped // through this function matches the V6 root hashes exactly. +// - On any failure after the checks above, the partially written output is +// removed. // -// nWorker controls how many of the 16 subtrie part files are encoded in -// parallel during the V7 write step; valid range is [1, 16]. The V6 read step +// `stream` selects the conversion strategy: +// - false: read the entire V6 forest into memory, convert it, and write the V7 +// checkpoint. Peak memory is approximately the sum of the V6 trie set and the +// V7 trie set, so mainnet-scale checkpoints need a host with memory headroom. +// - true: stream each part file node-by-node (see +// [convertCheckpointV6ToV7Stream]). Peak memory is independent of checkpoint +// size, at the cost of not re-deriving the trie root hashes from the +// converted nodes. +// +// nWorker controls how many of the 16 subtrie part files are processed in +// parallel; valid range is [1, 16]. In the non-streaming mode, the V6 read step // also reads the 16 subtrie part files concurrently using its own internal worker // pool (this function does not gate that), so the total parallelism while // running may exceed nWorker briefly during the read→write hand-off. // -// Memory: this implementation reads the entire V6 forest into memory before -// emitting V7 — peak memory is approximately the sum of the V6 trie set and the -// V7 trie set. For mainnet-scale checkpoints, run this on a host with enough -// memory headroom. Streaming subtrie-by-subtrie conversion is a possible future -// optimization but is not implemented here. -// // Expected error returns during normal operation: // - none — all error returns indicate a malformed input, a clobbering output, // or a write failure, which are treated as exceptions. @@ -160,55 +166,128 @@ func ConvertCheckpointV6ToV7( outputFileName string, logger zerolog.Logger, nWorker uint, + stream bool, ) error { + subtrieChecksums, topTrieChecksum, err := validateV6ToV7Conversion( + inputDir, inputFileName, outputDir, outputFileName, logger, nWorker) + if err != nil { + return err + } + + logger.Info(). + Str("v6_dir", inputDir). + Str("v6_file", inputFileName). + Str("v7_dir", outputDir). + Str("v7_file", outputFileName). + Uint("nworker", nWorker). + Bool("stream", stream). + Msg("starting V6→V7 checkpoint conversion") + + if stream { + err = convertCheckpointV6ToV7Stream( + inputDir, inputFileName, outputDir, outputFileName, logger, nWorker, subtrieChecksums, topTrieChecksum) + } else { + err = convertCheckpointV6ToV7InMemory(inputDir, inputFileName, outputDir, outputFileName, logger, nWorker) + } + + if err != nil { + // validateV6ToV7Conversion established that no output file existed before this + // call, so every file matching the output name now was written by this failed + // call and is safe to remove. + cleanupErr := deleteCheckpointFiles(outputDir, outputFileName) + if cleanupErr != nil { + return fmt.Errorf("fail to cleanup partially written output %s, after running into error: %w", + cleanupErr, err) + } + return err + } + + logger.Info().Msg("V6→V7 checkpoint conversion complete") + return nil +} + +// validateV6ToV7Conversion performs the pre-conversion checks shared by both +// conversion strategies and returns the per-subtrie checksums and the top-trie +// checksum recorded in the V6 checkpoint header. +// +// This function must run before any output file is created, and it must not +// create any itself: a failure here means this call wrote nothing, so the caller +// must not run output cleanup - which would delete a pre-existing V7 checkpoint +// belonging to a previous, successful conversion. +// +// No error returns are expected during normal operation. +func validateV6ToV7Conversion( + inputDir string, + inputFileName string, + outputDir string, + outputFileName string, + logger zerolog.Logger, + nWorker uint, +) ([]uint32, uint32, error) { if nWorker == 0 || nWorker > subtrieCount { - return fmt.Errorf("invalid nWorker %v, valid range is [1, %v]", nWorker, subtrieCount) + return nil, 0, 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 + return nil, 0, 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) + return nil, 0, fmt.Errorf("V6 checkpoint header not found at %s: %w", v6Header, err) } - subtrieChecksums, _, err := readCheckpointHeader(v6Header, logger) + subtrieChecksums, topTrieChecksum, err := readCheckpointHeader(v6Header, logger) if err != nil { - return fmt.Errorf("could not read V6 checkpoint header: %w", err) + return nil, 0, fmt.Errorf("could not read V6 checkpoint header: %w", err) + } + // The converters address the subtrie part files by index in [0, subtrieCount), + // so a header declaring a different number of subtries cannot be converted. + if len(subtrieChecksums) != subtrieCount { + return nil, 0, fmt.Errorf("V6 checkpoint header declares %v subtrie checksums, expected %v", + len(subtrieChecksums), subtrieCount) } if err := allPartFileExist(inputDir, inputFileName, len(subtrieChecksums)); err != nil { - return fmt.Errorf("V6 part files incomplete for %s/%s: %w", inputDir, inputFileName, err) + return nil, 0, 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) + return nil, 0, fmt.Errorf("could not check existing V7 output files: %w", err) } if len(v7Existing) != 0 { - return fmt.Errorf("V7 output already exists: %v", v7Existing) + return nil, 0, fmt.Errorf("V7 output already exists: %v", v7Existing) } + return subtrieChecksums, topTrieChecksum, nil +} + +// convertCheckpointV6ToV7InMemory converts a V6 checkpoint by loading the entire +// V6 forest into memory, converting it to payloadless tries, and writing them out +// with the V7 writer. Inputs are expected to have been checked by +// [validateV6ToV7Conversion]. +// +// No error returns are expected during normal operation. +func convertCheckpointV6ToV7InMemory( + inputDir string, + inputFileName string, + outputDir string, + outputFileName string, + logger zerolog.Logger, + nWorker uint, +) error { // 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). - Str("v7_dir", outputDir). - Str("v7_file", outputFileName). - Uint("nworker", nWorker). - Msg("starting V6→V7 checkpoint conversion") - // Read the V6 checkpoint fully — the V6 reader already reads the 16 subtrie // part files concurrently. The resulting tries share sub-tries via Go pointer // identity, which lets FromV6Tries memoize and avoid redundant conversion. + v6Header := filePathCheckpointHeader(inputDir, inputFileName) v6Tries, err := LoadCheckpoint(v6Header, logger) if err != nil { return fmt.Errorf("could not load V6 checkpoint: %w", err) @@ -237,7 +316,6 @@ func ConvertCheckpointV6ToV7( return fmt.Errorf("could not write V7 checkpoint: %w", err) } - logger.Info().Msg("V6→V7 checkpoint conversion complete") return nil } diff --git a/ledger/complete/wal/checkpoint_v7_convert_stream.go b/ledger/complete/wal/checkpoint_v7_convert_stream.go index 9d454062080..29df3f96881 100644 --- a/ledger/complete/wal/checkpoint_v7_convert_stream.go +++ b/ledger/complete/wal/checkpoint_v7_convert_stream.go @@ -7,6 +7,7 @@ import ( "io" "os" + "github.com/hashicorp/go-multierror" "github.com/rs/zerolog" "github.com/onflow/flow-go/ledger" @@ -43,10 +44,11 @@ const ( payloadEncodingVersion = 1 ) -// ConvertCheckpointV6ToV7Stream converts a V6 checkpoint at (inputDir, inputFileName) +// 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. +// memory. Inputs are expected to have been checked by [validateV6ToV7Conversion], +// which also supplies the V6 header's checksums. // // How it works: // - The V6 and V7 on-disk layouts are byte-identical except for (a) the version @@ -60,6 +62,8 @@ const ( // the register-size field. // - Node count and ordering are unchanged by the conversion, so every interim // node's child indices remain valid without rewriting. +// - Every input part file is fully CRC32-verified while being read, so input +// corruption is detected rather than carried into the V7 output. // - Per-part-file CRC32 checksums are recomputed during the write and collected // into a freshly written V7 header. // @@ -67,36 +71,13 @@ const ( // 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 +// Unlike the in-memory conversion, 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. // -// 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, or an IO failure. -func ConvertCheckpointV6ToV7Stream( - inputDir string, - inputFileName string, - outputDir string, - outputFileName string, - logger zerolog.Logger, - nWorker uint, -) error { - err := convertCheckpointV6ToV7Stream(inputDir, inputFileName, outputDir, outputFileName, logger, nWorker) - 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 -} - +// a malformed input or an IO failure. func convertCheckpointV6ToV7Stream( inputDir string, inputFileName string, @@ -104,52 +85,15 @@ func convertCheckpointV6ToV7Stream( outputFileName string, logger zerolog.Logger, nWorker uint, + subtrieChecksums []uint32, + topTrieChecksum uint32, ) 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) - } - // 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). - Str("v7_dir", outputDir). - Str("v7_file", outputFileName). - Uint("nworker", nWorker). - 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) @@ -169,7 +113,6 @@ func convertCheckpointV6ToV7Stream( return fmt.Errorf("could not write V7 checkpoint header: %w", err) } - logger.Info().Msg("stream V6→V7 checkpoint conversion complete") return nil } @@ -182,6 +125,11 @@ type streamSubtrieResult struct { // 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. +// +// subtrieChecksums are the checksums recorded in the V6 checkpoint header, one per +// subtrie part file; it must have exactly subtrieCount entries. +// +// No error returns are expected during normal operation. func convertSubTriesV6ToV7StreamConcurrently( inputDir string, inputFileName string, @@ -191,6 +139,13 @@ func convertSubTriesV6ToV7StreamConcurrently( logger zerolog.Logger, nWorker uint, ) ([]uint32, error) { + // The workers index subtrieChecksums by subtrie index, so a shorter slice would + // panic inside a goroutine. Callers validate this via validateV6ToV7Conversion; + // checking here keeps the indexing below provably safe. + if len(subtrieChecksums) != subtrieCount { + return nil, fmt.Errorf("expect %v subtrie checksums, but got %v", subtrieCount, len(subtrieChecksums)) + } + jobs := make(chan int, subtrieCount) for i := 0; i < subtrieCount; i++ { jobs <- i @@ -211,14 +166,22 @@ func convertSubTriesV6ToV7StreamConcurrently( }() } + // Drain all results before returning: a worker only renames its temp file to the + // final part file when it finishes, so returning early on the first error would + // let stragglers create output files after the caller has cleaned up. checksums := make([]uint32, subtrieCount) + var merr *multierror.Error 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) + merr = multierror.Append(merr, fmt.Errorf("fail to convert %v-th subtrie: %w", r.index, r.err)) + continue } checksums[r.index] = r.checksum } + if err := merr.ErrorOrNil(); err != nil { + return nil, err + } return checksums, nil } @@ -258,13 +221,18 @@ func convertSubTrieFileV6ToV7Stream( index, expectedSum, embeddedSum) } + // Restart from the beginning of the file and read everything through a + // Crc32Reader, so the bytes we convert are themselves CRC-verified (against the + // checksum stored in the file) rather than only the two stored checksums being + // compared. Without this, input corruption would be copied into the V7 output + // and covered up by a freshly computed, valid V7 checksum. 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 { + reader := NewCRC32Reader(bufio.NewReaderSize(inFile, defaultBufioReadSize)) + if err := validateFileHeader(MagicBytesCheckpointSubtrie, VersionV6, reader); 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 { @@ -288,6 +256,13 @@ func convertSubTrieFileV6ToV7Stream( logging(i) } + // Read the input's footer (node count) through the CRC reader, which completes + // the checksummed byte range, and verify the input file's integrity before + // finalizing the output. + if err := verifyInputChecksum(reader, encNodeCountSize, embeddedSum); err != nil { + return 0, fmt.Errorf("could not verify subtrie file %v: %w", index, err) + } + sum, err := storeSubtrieFooter(nodeCount, writer) if err != nil { return 0, fmt.Errorf("could not store subtrie footer: %w", err) @@ -328,13 +303,16 @@ func convertTopTrieFileV6ToV7Stream( expectedSum, embeddedSum) } + // Restart from the beginning of the file and read everything through a + // Crc32Reader, so the converted bytes are CRC-verified against the checksum + // stored in the input file (see convertSubTrieFileV6ToV7Stream). 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 { + reader := NewCRC32Reader(bufio.NewReaderSize(inFile, defaultBufioReadSize)) + if err := validateFileHeader(MagicBytesCheckpointToptrie, VersionV6, reader); 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) @@ -388,6 +366,12 @@ func convertTopTrieFileV6ToV7Stream( } } + // Read the input's footer (top-level node count + trie count) through the CRC + // reader and verify the input file's integrity before finalizing the output. + if err := verifyInputChecksum(reader, encNodeCountSize+encTrieCountSize, embeddedSum); err != nil { + return 0, fmt.Errorf("could not verify top-trie file: %w", err) + } + sum, err := storeTopLevelTrieFooter(topLevelNodesCount, triesCount, writer) if err != nil { return 0, fmt.Errorf("could not store top-trie footer: %w", err) @@ -395,6 +379,44 @@ func convertTopTrieFileV6ToV7Stream( return sum, nil } +// verifyInputChecksum completes the checksummed byte range of a V6 part file and +// verifies its integrity. It is called after all nodes (and, for the top-trie +// file, all trie root records) have been read from `reader`: it consumes the +// `footerSize` footer bytes — which are part of the checksummed range — compares +// the CRC32 computed over everything read so far against `expectedSum`, then +// consumes the stored checksum and asserts that the file ends there. +// +// This detects corruption of the input bytes themselves. Comparing the checksum +// stored in the part file against the one recorded in the checkpoint header is not +// sufficient: both are stored values and neither is derived from the bytes read. +// +// No error returns are expected during normal operation; all error returns +// indicate a corrupted or truncated input file, or an IO failure. +func verifyInputChecksum(reader *Crc32Reader, footerSize int, expectedSum uint32) error { + scratch := make([]byte, footerSize+crc32SumSize) + + // read the footer and discard it, the converted output writes its own + if _, err := io.ReadFull(reader, scratch[:footerSize]); err != nil { + return fmt.Errorf("cannot read footer: %w", err) + } + + actualSum := reader.Crc32() + if actualSum != expectedSum { + return fmt.Errorf("invalid checksum, expected %v, actual %v", expectedSum, actualSum) + } + + // read the stored checksum and discard it, we only care about reaching EOF + if _, err := io.ReadFull(reader, scratch[:crc32SumSize]); err != nil { + return fmt.Errorf("could not read stored checksum: %w", err) + } + + if err := ensureReachedEOF(reader); err != nil { + return fmt.Errorf("fail to reach end of file: %w", err) + } + + return nil +} + // v6ToV7NodeConverter streams individual V6-encoded nodes into V7-encoded nodes, // reusing internal scratch buffers across calls to avoid per-node allocations. // diff --git a/ledger/complete/wal/checkpoint_v7_convert_stream_test.go b/ledger/complete/wal/checkpoint_v7_convert_stream_test.go index 77d7db328d6..8ed2d494c48 100644 --- a/ledger/complete/wal/checkpoint_v7_convert_stream_test.go +++ b/ledger/complete/wal/checkpoint_v7_convert_stream_test.go @@ -2,6 +2,7 @@ package wal import ( "fmt" + "os" "testing" "github.com/rs/zerolog" @@ -24,11 +25,11 @@ func TestConvertCheckpointV6ToV7Stream_MatchesNonStream(t *testing.T) { // Path A: in-memory converter. nonStreamName := v6Name + ".nonstream" + V7FileSuffix - require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, nonStreamName, logger, 16)) + require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, nonStreamName, logger, 16, false)) // Path B: streaming converter. streamName := v6Name + ".stream" + V7FileSuffix - require.NoError(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, streamName, logger, 16)) + require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, streamName, logger, 16, true)) nonStreamFiles := filePaths(dir, nonStreamName, subtrieLevel) streamFiles := filePaths(dir, streamName, subtrieLevel) @@ -51,7 +52,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)) + require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, 16, true)) v7Tries, err := OpenAndReadCheckpointV7(dir, v7Name, logger) require.NoError(t, err) @@ -73,7 +74,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)) + require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, nWorker, true)) v7Tries, err := OpenAndReadCheckpointV7(dir, v7Name, logger) require.NoError(t, err) @@ -95,7 +96,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)) + require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, 16, true)) v7Tries, err := OpenAndReadCheckpointV7(dir, v7Name, logger) require.NoError(t, err) @@ -111,23 +112,132 @@ 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), + require.Error(t, ConvertCheckpointV6ToV7(dir, "x", dir, "out"+V7FileSuffix, logger, 0, true), "nWorker=0 must be rejected") - require.Error(t, ConvertCheckpointV6ToV7Stream(dir, "x", dir, "out"+V7FileSuffix, logger, 17), + require.Error(t, ConvertCheckpointV6ToV7(dir, "x", dir, "out"+V7FileSuffix, logger, 17, true), "nWorker > subtrieCount must be rejected") - require.Error(t, ConvertCheckpointV6ToV7Stream(dir, "missing", dir, "missing"+V7FileSuffix, logger, 4), + require.Error(t, ConvertCheckpointV6ToV7(dir, "missing", dir, "missing"+V7FileSuffix, logger, 4, true), "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), + require.Error(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, "no-suffix", logger, 4, true), "output filename without V7 suffix must be rejected") v7Name := v6Name + V7FileSuffix - require.NoError(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, v7Name, logger, 4)) - require.Error(t, ConvertCheckpointV6ToV7Stream(dir, v6Name, dir, v7Name, logger, 4), + require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, 4, true)) + require.Error(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, 4, true), "second conversion to the same V7 output must be rejected") }) } + +// TestConvertCheckpointV6ToV7_RejectedRerunKeepsOutput verifies that a conversion +// rejected because its output already exists leaves that output intact: the +// failure happens before anything is written, so the cleanup of partial output +// must not run and delete a previously converted checkpoint. +func TestConvertCheckpointV6ToV7_RejectedRerunKeepsOutput(t *testing.T) { + for _, stream := range []bool{false, true} { + t.Run(fmt.Sprintf("stream=%v", stream), func(t *testing.T) { + unittest.RunWithTempDir(t, func(dir string) { + logger := zerolog.Nop() + v6Tries := createMultipleRandomTries(t) + v6Name := "checkpoint.00000305" + require.NoError(t, StoreCheckpointV6Concurrently(v6Tries, dir, v6Name, logger)) + + v7Name := v6Name + V7FileSuffix + require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, 16, stream)) + + require.Error(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, 16, stream), + "second conversion to the same V7 output must be rejected") + + // the rejected re-run must not have touched the existing V7 checkpoint + v7Tries, err := OpenAndReadCheckpointV7(dir, v7Name, logger) + require.NoError(t, err, "existing V7 output must survive a rejected re-run") + 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_DetectsCorruptedInput verifies that the stream +// converter CRC-verifies the input bytes it converts: flipping a single byte of a +// V6 part file - leaving both stored checksums intact - must fail the conversion +// rather than produce a V7 checkpoint carrying corrupted data under a freshly +// computed, valid checksum. +func TestConvertCheckpointV6ToV7Stream_DetectsCorruptedInput(t *testing.T) { + // index of the V6 part file to corrupt: the largest subtrie file, and the + // top-trie file (always the (subtrieCount)-th part file) + for _, partFile := range []string{"subtrie", "toptrie"} { + t.Run(partFile, func(t *testing.T) { + unittest.RunWithTempDir(t, func(dir string) { + logger := zerolog.Nop() + v6Tries := createMultipleRandomTries(t) + v6Name := "checkpoint.00000306" + require.NoError(t, StoreCheckpointV6Concurrently(v6Tries, dir, v6Name, logger)) + + var path string + if partFile == "toptrie" { + path, _ = filePathTopTries(dir, v6Name) + } else { + path = largestSubTrieFilePath(t, dir, v6Name) + } + + // flip the last byte of the file's content: it belongs to the last + // encoded node (or trie root record) and precedes the footer and the + // stored checksum, so both stored checksums remain unchanged + footerSize := encNodeCountSize + crc32SumSize + if partFile == "toptrie" { + footerSize = encNodeCountSize + encTrieCountSize + crc32SumSize + } + corruptByteAt(t, path, -(int64(footerSize) + 1)) + + v7Name := v6Name + V7FileSuffix + err := ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, 16, true) + require.Error(t, err, "corrupted V6 input must be detected") + require.Contains(t, err.Error(), "invalid checksum") + + // no V7 output must be left behind + files, err := findCheckpointPartFiles(dir, v7Name) + require.NoError(t, err) + require.Empty(t, files, "failed conversion must not leave output files behind") + }) + }) + } +} + +// largestSubTrieFilePath returns the path of the V6 subtrie part file with the most +// content, i.e. the one guaranteed to hold encoded nodes. +func largestSubTrieFilePath(t *testing.T, dir string, fileName string) string { + var largestPath string + var largestSize int64 + for i := 0; i < subtrieCount; i++ { + path, _, err := filePathSubTries(dir, fileName, i) + require.NoError(t, err) + info, err := os.Stat(path) + require.NoError(t, err) + if info.Size() > largestSize { + largestSize, largestPath = info.Size(), path + } + } + require.NotEmpty(t, largestPath) + return largestPath +} + +// corruptByteAt flips all bits of a single byte of the given file. A negative +// offset is interpreted relative to the end of the file. +func corruptByteAt(t *testing.T, path string, offset int64) { + content, err := os.ReadFile(path) + require.NoError(t, err) + if offset < 0 { + offset += int64(len(content)) + } + require.GreaterOrEqual(t, offset, int64(0)) + require.Less(t, offset, int64(len(content))) + content[offset] ^= 0xFF + require.NoError(t, os.WriteFile(path, content, 0644)) +} diff --git a/ledger/complete/wal/checkpoint_v7_convert_test.go b/ledger/complete/wal/checkpoint_v7_convert_test.go index 419fd345aea..ae6cc08a38f 100644 --- a/ledger/complete/wal/checkpoint_v7_convert_test.go +++ b/ledger/complete/wal/checkpoint_v7_convert_test.go @@ -115,7 +115,7 @@ func TestConvertCheckpointV6ToV7_PreservesRootHashes(t *testing.T) { require.NoError(t, StoreCheckpointV6Concurrently(v6Tries, dir, v6Name, logger)) v7Name := v6Name + V7FileSuffix - require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, 16)) + require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, 16, false)) v7Tries, err := OpenAndReadCheckpointV7(dir, v7Name, logger) require.NoError(t, err) @@ -137,7 +137,7 @@ func TestConvertCheckpointV6ToV7_NWorkerOne(t *testing.T) { require.NoError(t, StoreCheckpointV6Concurrently(v6Tries, dir, v6Name, logger)) v7Name := v6Name + V7FileSuffix - require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, 1)) + require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, 1, false)) v7Tries, err := OpenAndReadCheckpointV7(dir, v7Name, logger) require.NoError(t, err) @@ -151,10 +151,10 @@ func TestConvertCheckpointV6ToV7_NWorkerOne(t *testing.T) { func TestConvertCheckpointV6ToV7_InvalidNWorker(t *testing.T) { unittest.RunWithTempDir(t, func(dir string) { logger := zerolog.Nop() - err := ConvertCheckpointV6ToV7(dir, "doesnt-matter", dir, "out"+V7FileSuffix, logger, 0) + err := ConvertCheckpointV6ToV7(dir, "doesnt-matter", dir, "out"+V7FileSuffix, logger, 0, false) require.Error(t, err, "nWorker=0 must be rejected") - err = ConvertCheckpointV6ToV7(dir, "doesnt-matter", dir, "out"+V7FileSuffix, logger, 17) + err = ConvertCheckpointV6ToV7(dir, "doesnt-matter", dir, "out"+V7FileSuffix, logger, 17, false) require.Error(t, err, "nWorker > subtrieCount must be rejected") }) } @@ -168,7 +168,7 @@ func TestConvertCheckpointV6ToV7_RequiresV7Suffix(t *testing.T) { v6Name := "checkpoint.00000001" require.NoError(t, StoreCheckpointV6Concurrently(v6Tries, dir, v6Name, logger)) - err := ConvertCheckpointV6ToV7(dir, v6Name, dir, "no-suffix", logger, 4) + err := ConvertCheckpointV6ToV7(dir, v6Name, dir, "no-suffix", logger, 4, false) require.Error(t, err, "output filename without V7 suffix must be rejected") }) } @@ -183,9 +183,9 @@ func TestConvertCheckpointV6ToV7_RejectsClobber(t *testing.T) { require.NoError(t, StoreCheckpointV6Concurrently(v6Tries, dir, v6Name, logger)) v7Name := v6Name + V7FileSuffix - require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, 4)) + require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, 4, false)) - err := ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, 4) + err := ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, 4, false) require.Error(t, err, "second conversion to the same V7 output must be rejected") }) } @@ -212,13 +212,13 @@ func TestDeleteCheckpointFilesClearsPartialV7Conversion(t *testing.T) { hasV7Root, err := HasRootCheckpointV7(dir) require.NoError(t, err) require.False(t, hasV7Root) - require.Error(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, 4), + require.Error(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, 4, false), "leftover part files must block a retry") // clearing the partial output unblocks the retry require.NoError(t, DeleteCheckpointFiles(dir, v7Name)) require.NoFileExists(t, partialPart) - require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, 4)) + require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, 4, false)) hasV7Root, err = HasRootCheckpointV7(dir) require.NoError(t, err) @@ -238,7 +238,7 @@ func TestDeleteCheckpointFilesClearsPartialV7Conversion(t *testing.T) { func TestConvertCheckpointV6ToV7_MissingV6Input(t *testing.T) { unittest.RunWithTempDir(t, func(dir string) { logger := zerolog.Nop() - err := ConvertCheckpointV6ToV7(dir, "missing", dir, "missing"+V7FileSuffix, logger, 4) + err := ConvertCheckpointV6ToV7(dir, "missing", dir, "missing"+V7FileSuffix, logger, 4, false) require.Error(t, err, "missing V6 input must be reported") }) } @@ -254,7 +254,7 @@ func TestConvertCheckpointV6ToV7_DifferentOutputDir(t *testing.T) { require.NoError(t, StoreCheckpointV6Concurrently(v6Tries, srcDir, v6Name, logger)) v7Name := v6Name + V7FileSuffix - require.NoError(t, ConvertCheckpointV6ToV7(srcDir, v6Name, dstDir, v7Name, logger, 4)) + require.NoError(t, ConvertCheckpointV6ToV7(srcDir, v6Name, dstDir, v7Name, logger, 4, false)) // V7 files exist in dstDir, not in srcDir. v7Tries, err := OpenAndReadCheckpointV7(dstDir, v7Name, logger) @@ -280,7 +280,7 @@ func TestConvertCheckpointV6ToV7_EmptyTrie(t *testing.T) { require.NoError(t, StoreCheckpointV6Concurrently(v6Tries, dir, v6Name, logger)) v7Name := v6Name + V7FileSuffix - require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, 16)) + require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, 16, false)) v7Tries, err := OpenAndReadCheckpointV7(dir, v7Name, logger) require.NoError(t, err) @@ -405,7 +405,7 @@ func TestFullVsPayloadlessForest_LoadConvertedCheckpoint(t *testing.T) { // Convert V6 → V7. v7Name := v6Name + V7FileSuffix - require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, 16)) + require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, 16, false)) // Reload V7 into a fresh payloadless forest. v7Tries, err := OpenAndReadCheckpointV7(dir, v7Name, logger) @@ -508,8 +508,8 @@ func TestConvertCheckpointV6ToV7_Deterministic(t *testing.T) { require.NoError(t, StoreCheckpointV6Concurrently(v6Tries, srcDir, v6Name, logger)) v7Name := v6Name + V7FileSuffix - require.NoError(t, ConvertCheckpointV6ToV7(srcDir, v6Name, dst1, v7Name, logger, 16)) - require.NoError(t, ConvertCheckpointV6ToV7(srcDir, v6Name, dst2, v7Name, logger, 16)) + require.NoError(t, ConvertCheckpointV6ToV7(srcDir, v6Name, dst1, v7Name, logger, 16, false)) + require.NoError(t, ConvertCheckpointV6ToV7(srcDir, v6Name, dst2, v7Name, logger, 16, false)) files1 := filePaths(dst1, v7Name, subtrieLevel) files2 := filePaths(dst2, v7Name, subtrieLevel) @@ -533,7 +533,7 @@ func TestConvertCheckpointV6ToV7_IntermediateNWorker(t *testing.T) { for _, nWorker := range []uint{2, 4, 8} { v7Name := fmt.Sprintf("%s.nw%d%s", v6Name, nWorker, V7FileSuffix) - require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, nWorker)) + require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, v7Name, logger, nWorker, false)) v7Tries, err := OpenAndReadCheckpointV7(dir, v7Name, logger) require.NoError(t, err) @@ -558,7 +558,7 @@ func TestConvertCheckpointV6ToV7_MatchesDirectV7Write(t *testing.T) { // Path A: converter. convertedName := v6Name + ".converted" + V7FileSuffix - require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, convertedName, logger, 16)) + require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, convertedName, logger, 16, false)) // Path B: convert tries in-memory and write directly. v7Tries, err := FromV6Tries(v6Tries) @@ -586,7 +586,7 @@ func TestConvertCheckpointV6ToV7_JunkInput(t *testing.T) { junkPath := filepath.Join(dir, v6Name) require.NoError(t, writeBytes(junkPath, []byte("not a checkpoint header"))) - err := ConvertCheckpointV6ToV7(dir, v6Name, dir, v6Name+V7FileSuffix, logger, 16) + err := ConvertCheckpointV6ToV7(dir, v6Name, dir, v6Name+V7FileSuffix, logger, 16, false) require.Error(t, err, "junk V6 header file must be rejected") }) } diff --git a/ledger/factory/factory_test.go b/ledger/factory/factory_test.go index 8d3d0685b09..76b11258909 100644 --- a/ledger/factory/factory_test.go +++ b/ledger/factory/factory_test.go @@ -700,7 +700,7 @@ func TestNewPayloadlessLedger_LoadsConvertedV6(t *testing.T) { )) v7Name := v6Name + wal.V7FileSuffix - require.NoError(t, wal.ConvertCheckpointV6ToV7(tempDir, v6Name, tempDir, v7Name, logger, 16)) + require.NoError(t, wal.ConvertCheckpointV6ToV7(tempDir, v6Name, tempDir, v7Name, logger, 16, false)) plLedger, err := NewPayloadlessLedger(Config{ Triedir: tempDir, @@ -748,7 +748,7 @@ func TestNewPayloadlessLedger_LoadsV7RootCheckpoint(t *testing.T) { require.NoError(t, wal.ConvertCheckpointV6ToV7( tempDir, bootstrap.FilenameWALRootCheckpoint, tempDir, bootstrap.FilenameWALRootCheckpoint+wal.V7FileSuffix, - logger, 16, + logger, 16, false, )) // Ensure the test actually exercises the root-checkpoint path: no numbered