[CompactSealedState] T2: add compact-execution-state command with shared WAL/rollback helpers

cmd/util/cmd/common/checkpoint.go

@@ -5,6 +5,8 @@ import (
 	"fmt"
 	"os"
 
+	"github.com/rs/zerolog"
+
 	"github.com/onflow/flow-go/ledger/complete/wal"
 	"github.com/onflow/flow-go/module/irrecoverable"
 	utilsio "github.com/onflow/flow-go/utils/io"
@@ -69,3 +71,46 @@ func MoveCheckpointFiles(sourceDir, sourceName, destDir, destName string) error
 
 	return nil
 }
+
+// BackupCheckpointsFrom moves all checkpoint files in dir whose checkpoint number is
+// greater than or equal to minNumber into backupDir, preserving their file names.
+//
+// A checkpoint number identifies the highest WAL segment the checkpoint was created up
+// to. After the WAL has been trimmed so that segment minNumber is its last segment, any
+// checkpoint at or beyond minNumber references state newer than the trim target and is
+// inconsistent with the remaining WAL; moving those checkpoint files to backupDir keeps
+// the execution-state directory free of checkpoints the node could load over the trimmed
+// WAL, while preserving the files for possible restore.
+//
+// Each checkpoint is moved via [MoveCheckpointFiles]. A partial checkpoint (a subset of
+// the 18 V6 checkpoint files) cannot be moved and is skipped with a warning, since the
+// node ignores checkpoints that fail to load. If backupDir does not exist it is created.
+//
+// No error returns are expected during normal operation.
+func BackupCheckpointsFrom(lg zerolog.Logger, dir, backupDir string, minNumber int) error {
+	checkpoints, err := wal.Checkpoints(dir)
+	if err != nil {
+		return fmt.Errorf("cannot list checkpoints in %s: %w", dir, err)
+	}
+
+	for _, n := range checkpoints {
+		if n < minNumber {
+			continue
+		}
+
+		name := wal.NumberToFilename(n)
+		if err := MoveCheckpointFiles(dir, name, backupDir, name); err != nil {
+			if errors.Is(err, ErrCheckpointFileMissing) {
+				lg.Warn().Int("checkpoint", n).
+					Msg("skipping partial checkpoint in execution state dir")
+				continue
+			}
+			return fmt.Errorf("cannot move checkpoint %d to backup dir: %w", n, err)
+		}
+
+		lg.Info().Int("checkpoint", n).Str("backup-dir", backupDir).
+			Msg("moved checkpoint to backup dir")
+	}
+
+	return nil
+}

cmd/util/cmd/common/checkpoint_test.go

@@ -7,6 +7,7 @@ import (
 	"path/filepath"
 	"testing"
 
+	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/require"
 
 	"github.com/onflow/flow-go/cmd/util/cmd/common"
@@ -162,6 +163,76 @@ func TestCheckpointV6AllFilePaths_Count(t *testing.T) {
 	require.Len(t, paths, 18)
 }
 
+// TestBackupCheckpointsFrom_FiltersByNumber verifies that only checkpoints with number
+// greater than or equal to minNumber are moved to the backup directory, while older
+// checkpoints remain in place, and that all 18 files of each moved checkpoint are
+// relocated with their names preserved.
+func TestBackupCheckpointsFrom_FiltersByNumber(t *testing.T) {
+	unittest.RunWithTempDir(t, func(base string) {
+		dir := filepath.Join(base, "exec")
+		backupDir := filepath.Join(base, "backup")
+		require.NoError(t, os.MkdirAll(dir, 0755))
+
+		createFakeCheckpoint(t, dir, "checkpoint.00000002")
+		createFakeCheckpoint(t, dir, "checkpoint.00000004")
+		createFakeCheckpoint(t, dir, "checkpoint.00000006")
+
+		logger := zerolog.Nop()
+		require.NoError(t, common.BackupCheckpointsFrom(logger, dir, backupDir, 4))
+
+		// Number 4 and 6 are at or beyond minNumber 4: moved.
+		for _, name := range []string{"checkpoint.00000004", "checkpoint.00000006"} {
+			for i, dst := range wal.CheckpointV6AllFilePaths(backupDir, name) {
+				_, err := os.Stat(dst)
+				require.NoError(t, err, "expected backed up file %d of %s in %s", i, name, dst)
+			}
+			for i, src := range wal.CheckpointV6AllFilePaths(dir, name) {
+				_, err := os.Stat(src)
+				require.True(t, os.IsNotExist(err), "expected moved file %d of %s to be gone: %s", i, name, src)
+			}
+		}
+
+		// Number 2 is below minNumber 4: untouched.
+		for i, src := range wal.CheckpointV6AllFilePaths(dir, "checkpoint.00000002") {
+			_, err := os.Stat(src)
+			require.NoError(t, err, "expected older checkpoint file %d to remain: %s", i, src)
+		}
+	})
+}
+
+// TestBackupCheckpointsFrom_PartialCheckpointSkipped verifies that a partial checkpoint
+// (only a subset of the 18 V6 files present) is skipped with no error and left in place,
+// while a complete checkpoint at or beyond minNumber is moved.
+func TestBackupCheckpointsFrom_PartialCheckpointSkipped(t *testing.T) {
+	unittest.RunWithTempDir(t, func(base string) {
+		dir := filepath.Join(base, "exec")
+		backupDir := filepath.Join(base, "backup")
+		require.NoError(t, os.MkdirAll(dir, 0755))
+
+		// Create only the header file of checkpoint 3, a full checkpoint 4, and a full
+		// older checkpoint 1.
+		header := wal.CheckpointV6AllFilePaths(dir, "checkpoint.00000003")[0]
+		require.NoError(t, os.WriteFile(header, []byte("partial"), 0644))
+		createFakeCheckpoint(t, dir, "checkpoint.00000004")
+		createFakeCheckpoint(t, dir, "checkpoint.00000001")
+
+		logger := zerolog.Nop()
+		require.NoError(t, common.BackupCheckpointsFrom(logger, dir, backupDir, 3))
+
+		// The partial checkpoint 3 remains in place, untouched.
+		_, err := os.Stat(filepath.Join(dir, "checkpoint.00000003"))
+		require.NoError(t, err, "expected partial checkpoint to stay in source")
+		_, err = os.Stat(filepath.Join(backupDir, "checkpoint.00000003"))
+		require.True(t, os.IsNotExist(err), "expected partial checkpoint not to be moved")
+
+		// Complete checkpoint 4 is moved; older checkpoint 1 stayed.
+		_, err = os.Stat(wal.CheckpointV6AllFilePaths(backupDir, "checkpoint.00000004")[0])
+		require.NoError(t, err, "expected complete checkpoint to be moved to backup")
+		_, err = os.Stat(wal.CheckpointV6AllFilePaths(dir, "checkpoint.00000001")[0])
+		require.NoError(t, err, "expected older checkpoint to remain")
+	})
+}
+
 // TestCheckpointV6AllFilePaths_Suffixes verifies the returned paths follow the expected
 // naming scheme: header has no suffix, part files end in .000–.016.
 func TestCheckpointV6AllFilePaths_Suffixes(t *testing.T) {

cmd/util/cmd/common/rollback.go

@@ -0,0 +1,165 @@
+package common
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/rs/zerolog/log"
+
+	"github.com/onflow/flow-go/model/flow"
+	"github.com/onflow/flow-go/state/protocol"
+	"github.com/onflow/flow-go/storage"
+)
+
+// RemoveExecutionResultsFromHeight removes all execution results and related data for
+// every block at or above fromHeight, including both finalized and pending blocks.
+// It returns the set of chunk IDs that were removed from the protocol-state DB so
+// that the caller can delete the corresponding chunk-data-packs from the chunk DB.
+//
+// No error returns are expected during normal operation.
+func RemoveExecutionResultsFromHeight(
+	protocolDBBatch storage.Batch,
+	protoState protocol.State,
+	transactionResults storage.TransactionResults,
+	commits storage.Commits,
+	results storage.ExecutionResults,
+	myReceipts storage.MyExecutionReceipts,
+	events storage.Events,
+	serviceEvents storage.ServiceEvents,
+	fromHeight uint64,
+) ([]flow.Identifier, error) {
+	log.Info().Msgf("removing results for blocks from height: %v", fromHeight)
+
+	root := protoState.Params().FinalizedRoot()
+
+	if fromHeight <= root.Height {
+		return nil, fmt.Errorf("can only remove results for blocks above root: fromHeight %v, rootHeight %v",
+			fromHeight, root.Height)
+	}
+
+	final, err := protoState.Final().Head()
+	if err != nil {
+		return nil, fmt.Errorf("could not get finalized head: %w", err)
+	}
+
+	if fromHeight > final.Height {
+		return nil, fmt.Errorf("cannot remove results for unfinalized height %v (finalized: %v)",
+			fromHeight, final.Height)
+	}
+
+	var allChunkIDs []flow.Identifier
+
+	pendings, err := protoState.Final().Descendants()
+	if err != nil {
+		return nil, fmt.Errorf("could not get pending descendants: %w", err)
+	}
+
+	// Remove pending descendants before finalized blocks, and iterate in reverse only
+	// so that deeper descendants are logged before their ancestors in progress messages.
+	// All removals are staged in protocolDBBatch and committed once by the caller.
+	for i := len(pendings) - 1; i >= 0; i-- {
+		pending := pendings[i]
+		chunkIDs, err := RemoveExecutionResultsForBlock(
+			protocolDBBatch, commits, transactionResults, results,
+			myReceipts, events, serviceEvents, pending)
+		if err != nil {
+			return nil, fmt.Errorf("could not remove result for pending block %v: %w", pending, err)
+		}
+
+		allChunkIDs = append(allChunkIDs, chunkIDs...)
+		log.Info().Msgf("removed result for pending block %v (%v/%v)", pending, len(pendings)-i, len(pendings))
+	}
+
+	total := int(final.Height-fromHeight) + 1
+	finalRemoved := 0
+
+	// Iterate from highest to lowest only for clearer progress logging.
+	// All removals are staged in protocolDBBatch and committed once by the caller.
+	for height := final.Height; height >= fromHeight; height-- {
+		head, err := protoState.AtHeight(height).Head()
+		if err != nil {
+			return nil, fmt.Errorf("could not get header at height %v: %w", height, err)
+		}
+
+		chunkIDs, err := RemoveExecutionResultsForBlock(
+			protocolDBBatch, commits, transactionResults, results,
+			myReceipts, events, serviceEvents, head.ID())
+		if err != nil {
+			return nil, fmt.Errorf("could not remove result for finalized block at height %v: %w", height, err)
+		}
+
+		allChunkIDs = append(allChunkIDs, chunkIDs...)
+		finalRemoved++
+		log.Info().Msgf("removed result at height %v (%v/%v)", height, finalRemoved, total)
+	}
+
+	log.Info().Msgf("removed execution results from height %v: %v finalized, %v pending blocks",
+		fromHeight, finalRemoved, len(pendings))
+
+	return allChunkIDs, nil
+}
+
+// RemoveExecutionResultsForBlock removes all execution-related storage entries for a
+// single block in a single batch write and returns the chunk IDs that should be
+// removed from the chunk-data-pack database.
+//
+// No error returns are expected during normal operation.
+func RemoveExecutionResultsForBlock(
+	protocolDBBatch storage.Batch,
+	commits storage.Commits,
+	transactionResults storage.TransactionResults,
+	results storage.ExecutionResults,
+	myReceipts storage.MyExecutionReceipts,
+	events storage.Events,
+	serviceEvents storage.ServiceEvents,
+	blockID flow.Identifier,
+) ([]flow.Identifier, error) {
+	result, err := results.ByBlockID(blockID)
+	if errors.Is(err, storage.ErrNotFound) {
+		log.Info().Msgf("no execution result for block %v — skipping", blockID)
+		return nil, nil
+	}
+	if err != nil {
+		return nil, fmt.Errorf("could not get execution result for block %v: %w", blockID, err)
+	}
+
+	chunkIDs := make([]flow.Identifier, 0, len(result.Chunks))
+	for _, chunk := range result.Chunks {
+		chunkIDs = append(chunkIDs, chunk.ID())
+	}
+
+	if err = commits.BatchRemoveByBlockID(blockID, protocolDBBatch); err != nil {
+		if !errors.Is(err, storage.ErrNotFound) {
+			return nil, fmt.Errorf("could not remove state commitment for block %v: %w", blockID, err)
+		}
+		log.Warn().Msgf("state commitment not found for block %v", blockID)
+	}
+
+	if err = transactionResults.BatchRemoveByBlockID(blockID, protocolDBBatch); err != nil {
+		return nil, fmt.Errorf("could not remove transaction results for block %v: %w", blockID, err)
+	}
+
+	if err = myReceipts.BatchRemoveIndexByBlockID(blockID, protocolDBBatch); err != nil {
+		if !errors.Is(err, storage.ErrNotFound) {
+			return nil, fmt.Errorf("could not remove own receipt index for block %v: %w", blockID, err)
+		}
+		log.Warn().Msgf("own receipt not found for block %v", blockID)
+	}
+
+	if err = events.BatchRemoveByBlockID(blockID, protocolDBBatch); err != nil {
+		return nil, fmt.Errorf("could not remove events for block %v: %w", blockID, err)
+	}
+
+	if err = serviceEvents.BatchRemoveByBlockID(blockID, protocolDBBatch); err != nil {
+		return nil, fmt.Errorf("could not remove service events for block %v: %w", blockID, err)
+	}
+
+	if err = results.BatchRemoveIndexByBlockID(blockID, protocolDBBatch); err != nil {
+		if !errors.Is(err, storage.ErrNotFound) {
+			return nil, fmt.Errorf("could not remove execution result index for block %v: %w", blockID, err)
+		}
+		log.Warn().Msgf("execution result index not found for block %v", blockID)
+	}
+
+	return chunkIDs, nil
+}