[Storehouse] 010 Util convert v6 -> v7 checkpoint

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

integration/localnet/builder/bootstrap.go

@@ -1024,4 +1024,3 @@ func prepareTestExecutionService(dockerServices Services, flowNodeContainerConfi
 
 	return dockerServices
 }
-

ledger/complete/wal/checkpoint_v6_writer.go

@@ -582,6 +582,36 @@ func storeTries(
 	return nil
 }
 
+// removeStaleTempFiles removes leftover "writing-<outputFile>*" 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)

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-<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, 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"))
+	})
+}

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).