diff --git a/cmd/util/cmd/checkpoint-convert-v7/cmd.go b/cmd/util/cmd/checkpoint-convert-v7/cmd.go index 675ac442c04..4780be2755c 100644 --- a/cmd/util/cmd/checkpoint-convert-v7/cmd.go +++ b/cmd/util/cmd/checkpoint-convert-v7/cmd.go @@ -16,6 +16,7 @@ var ( flagOutputDir string flagOutput string flagNWorker uint + flagStream bool ) // Cmd converts a V6 checkpoint to a V7 (payloadless) checkpoint by reading @@ -54,6 +55,10 @@ 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)") } func run(*cobra.Command, []string) { @@ -73,21 +78,36 @@ func run(*cobra.Command, []string) { Str("output_dir", outputDir). Str("output", outputFile). Uint("nworker", flagNWorker). + Bool("stream", flagStream). 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, + ) + } else { + err = wal.ConvertCheckpointV6ToV7( + flagCheckpointDir, + flagCheckpoint, + outputDir, + outputFile, + log.Logger, + flagNWorker, + ) + } if err != nil { 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/integration/localnet/builder/bootstrap.go b/integration/localnet/builder/bootstrap.go index a9a5ee5d635..69e90ea8e83 100644 --- a/integration/localnet/builder/bootstrap.go +++ b/integration/localnet/builder/bootstrap.go @@ -1024,4 +1024,3 @@ func prepareTestExecutionService(dockerServices Services, flowNodeContainerConfi return dockerServices } - 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 new file mode 100644 index 00000000000..9d454062080 --- /dev/null +++ b/ledger/complete/wal/checkpoint_v7_convert_stream.go @@ -0,0 +1,512 @@ +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. +// +// 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 +} + +func convertCheckpointV6ToV7Stream( + inputDir string, + inputFileName string, + outputDir string, + outputFileName string, + logger zerolog.Logger, + nWorker uint, +) 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) + 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) + 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, +) ([]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) + 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, +) (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() + 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, +) (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() + 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 +} + +// newV6ToV7NodeConverter returns a converter with preallocated scratch buffers. +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), + } +} + +// 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. +// +// No error returns are expected during normal operation; all error returns indicate +// 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:]) + 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) + } + + 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..77d7db328d6 --- /dev/null +++ b/ledger/complete/wal/checkpoint_v7_convert_stream_test.go @@ -0,0 +1,133 @@ +package wal + +import ( + "fmt" + "testing" + + "github.com/rs/zerolog" + "github.com/stretchr/testify/require" + + "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. +func TestConvertCheckpointV6ToV7Stream_MatchesNonStream(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)) + + 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, 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)) + + 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. +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)) + + 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)) + + v7Tries, err := OpenAndReadCheckpointV7(dir, v7Name, logger) + require.NoError(t, err) + require.Len(t, v7Tries, 1) + require.True(t, v7Tries[0].IsEmpty()) + }) +} + +// 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), + "nWorker=0 must be rejected") + 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), + "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), + "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), + "second conversion to the same V7 output must be rejected") + }) +}