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 675ac442c04..b0bf76d57bc 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,6 +78,7 @@ 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( @@ -82,12 +88,15 @@ func run(*cobra.Command, []string) { outputFile, log.Logger, flagNWorker, + flagStream, ) 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 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_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..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,49 +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) } - 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") + 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) + } // 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) @@ -231,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 new file mode 100644 index 00000000000..29df3f96881 --- /dev/null +++ b/ledger/complete/wal/checkpoint_v7_convert_stream.go @@ -0,0 +1,534 @@ +package wal + +import ( + "bufio" + "encoding/binary" + "fmt" + "io" + "os" + + "github.com/hashicorp/go-multierror" + "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. 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 +// 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. +// - 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. +// +// 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 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. +// +// No error returns are expected during normal operation; all error returns indicate +// a malformed input or an IO failure. +func convertCheckpointV6ToV7Stream( + inputDir string, + inputFileName string, + outputDir string, + outputFileName string, + logger zerolog.Logger, + nWorker uint, + subtrieChecksums []uint32, + topTrieChecksum uint32, +) 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) + } + + // 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) + } + + 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. +// +// 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, + outputDir string, + outputFileName string, + subtrieChecksums []uint32, + 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 + } + 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} + } + }() + } + + // 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 { + 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 +} + +// 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) + } + + // 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) + } + reader := NewCRC32Reader(bufio.NewReaderSize(inFile, defaultBufioReadSize)) + if err := validateFileHeader(MagicBytesCheckpointSubtrie, VersionV6, reader); err != nil { + return 0, fmt.Errorf("invalid subtrie file header: %w", err) + } + + 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) + } + + // 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) + } + 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) + } + + // 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) + } + 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) + } + + // 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) + } + } + + // 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) + } + 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. +// +// 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..8ed2d494c48 --- /dev/null +++ b/ledger/complete/wal/checkpoint_v7_convert_stream_test.go @@ -0,0 +1,243 @@ +package wal + +import ( + "fmt" + "os" + "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, false)) + + // Path B: streaming converter. + streamName := v6Name + ".stream" + V7FileSuffix + require.NoError(t, ConvertCheckpointV6ToV7(dir, v6Name, dir, streamName, logger, 16, true)) + + 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, ConvertCheckpointV6ToV7(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. +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, ConvertCheckpointV6ToV7(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, ConvertCheckpointV6ToV7(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()) + }) +} + +// 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, ConvertCheckpointV6ToV7(dir, "x", dir, "out"+V7FileSuffix, logger, 0, true), + "nWorker=0 must be rejected") + require.Error(t, ConvertCheckpointV6ToV7(dir, "x", dir, "out"+V7FileSuffix, logger, 17, true), + "nWorker > subtrieCount must be rejected") + 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, ConvertCheckpointV6ToV7(dir, v6Name, dir, "no-suffix", logger, 4, true), + "output filename without V7 suffix must be rejected") + + v7Name := v6Name + V7FileSuffix + 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