[Storehouse] 010 Util convert v6 -> v7 checkpoint

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)

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

integration/localnet/builder/bootstrap.go

@@ -906,6 +906,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))
 		}

ledger/complete/wal/checkpoint_v6_test.go

@@ -463,7 +463,7 @@ func compareFiles(file1, file2 string) error {
 
 	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()
@@ -475,25 +475,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 {

ledger/complete/wal/checkpoint_v6_writer.go

@@ -583,6 +583,74 @@ func storeTries(
 	return nil
 }
 
+// removeStaleTempFiles removes leftover "writing-<outputFile>-*" (header) and
+// "writing-<outputFile>.<NNN>-*" (part file) temporary 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.
+//
+// The patterns pin the character following outputFile ("-" for the header, "."
+// plus the three-digit part index for a part file) so temp files belonging to a
+// different output whose name merely starts with outputFile are left alone; a
+// bare "writing-<outputFile>*" glob would let overlapping conversions remove each
+// other's in-progress files. Final checkpoint 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 {
+	patterns := []string{
+		path.Join(outputDir, fmt.Sprintf("writing-%v-*", outputFile)),
+		path.Join(outputDir, fmt.Sprintf("writing-%v.[0-9][0-9][0-9]-*", outputFile)),
+	}
+
+	var merror *multierror.Error
+	for _, pattern := range patterns {
+		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 && !os.IsNotExist(err) {
+				merror = multierror.Append(merror, err)
+			}
+			logger.Info().Msgf("removed stale checkpoint temp file %v", file)
+		}
+	}
+
+	return merror.ErrorOrNil()
+}
+
+// deleteCheckpointPartFiles removes the checkpoint files (header and the 17 part
+// files) that currently exist for the given fileName in outputDir.
+//
+// Unlike [deleteCheckpointFiles], it treats fileName as an exact name rather than
+// a prefix, so it never matches a different checkpoint whose name merely starts
+// with fileName. This matters when a conversion reads its input from and writes
+// its output to the same directory: a failed V6→V7 conversion cleans up after
+// itself, and a prefix glob would also delete the V6 input when its name shares
+// the output's prefix.
+//
+// No error returns are expected during normal operation.
+func deleteCheckpointPartFiles(outputDir string, fileName string) error {
+	existing, err := findCheckpointPartFiles(outputDir, fileName)
+	if err != nil {
+		return fmt.Errorf("could not locate checkpoint files to delete: %w", err)
+	}
+
+	var merror *multierror.Error
+	for _, file := range existing {
+		if err := os.Remove(file); err != nil && !os.IsNotExist(err) {
+			merror = multierror.Append(merror, err)
+		}
+	}
+
+	return merror.ErrorOrNil()
+}
+
 // deleteCheckpointFiles removes any checkpoint files with given checkpoint prefix in the outputDir.
 //
 // A V6 checkpoint name is a prefix of the same-numbered V7 checkpoint name (e.g. "checkpoint.00000100"

ledger/complete/wal/checkpoint_v6_writer_test.go

@@ -0,0 +1,69 @@
+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-<outputFile>*" 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-<fileName>-<random>").
+		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, a temp file
+		// for a different output (e.g. a V6 checkpoint with a different name), and a
+		// temp file for a different V7 output whose name starts with outputFile.
+		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
+			// temp for a different V7 output that shares outputFile's prefix
+			"writing-root.checkpoint.v7.00000100.v7-5150",
+		}
+
+		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"))
+	})
+}