Checkpointing V6 - support concurrent checkpoint encoding/decoding

cmd/execution_builder.go

@@ -609,6 +609,10 @@ func (exeNode *ExecutionNode) LoadExecutionStateLedger(
 		return nil, fmt.Errorf("failed to initialize wal: %w", err)
 	}
 
+	if exeNode.exeConf.outputCheckpointV5 {
+		exeNode.diskWAL.UseCheckpointVersion5()
+	}
+
 	exeNode.ledgerStorage, err = ledger.NewLedger(exeNode.diskWAL, int(exeNode.exeConf.mTrieCacheSize), exeNode.collector, node.Logger.With().Str("subcomponent",
 		"ledger").Logger(), ledger.DefaultPathFinderVersion)
 	return exeNode.ledgerStorage, err

cmd/execution_config.go

@@ -26,6 +26,7 @@ type ExecutionConfig struct {
 	transactionResultsCacheSize          uint
 	checkpointDistance                   uint
 	checkpointsToKeep                    uint
+	outputCheckpointV5                   bool
 	stateDeltasLimit                     uint
 	chunkDataPackCacheSize               uint
 	chunkDataPackRequestsCacheSize       uint32
@@ -63,6 +64,7 @@ func (exeConf *ExecutionConfig) SetupFlags(flags *pflag.FlagSet) {
 	flags.Uint32Var(&exeConf.mTrieCacheSize, "mtrie-cache-size", 500, "cache size for MTrie")
 	flags.UintVar(&exeConf.checkpointDistance, "checkpoint-distance", 20, "number of WAL segments between checkpoints")
 	flags.UintVar(&exeConf.checkpointsToKeep, "checkpoints-to-keep", 5, "number of recent checkpoints to keep (0 to keep all)")
+	flags.BoolVar(&exeConf.outputCheckpointV5, "outputCheckpointV5", false, "output checkpoint file in v5")
 	flags.UintVar(&exeConf.stateDeltasLimit, "state-deltas-limit", 100, "maximum number of state deltas in the memory pool")
 	flags.UintVar(&exeConf.computationConfig.ProgramsCacheSize, "cadence-execution-cache", programs.DefaultProgramsCacheSize,
 		"cache size for Cadence execution")

cmd/util/cmd/execution-state-extract/cmd.go

@@ -26,6 +26,7 @@ var (
 	flagChain             string
 	flagNoMigration       bool
 	flagNoReport          bool
+	flagVersion           int
 )
 
 func getChain(chainName string) (chain flow.Chain, err error) {
@@ -70,6 +71,8 @@ func init() {
 
 	Cmd.Flags().BoolVar(&flagNoReport, "no-report", false,
 		"don't report the state")
+
+	Cmd.Flags().IntVar(&flagVersion, "version", 6, "checkpoint version")
 }
 
 func run(*cobra.Command, []string) {
@@ -86,6 +89,8 @@ func run(*cobra.Command, []string) {
 			log.Fatal().Err(err).Msg("malformed block hash")
 		}
 
+		log.Info().Msgf("extracting state by block ID: %v", blockID)
+
 		db := common.InitStorage(flagDatadir)
 		defer db.Close()
 
@@ -94,7 +99,7 @@ func run(*cobra.Command, []string) {
 
 		stateCommitment, err = getStateCommitment(commits, blockID)
 		if err != nil {
-			log.Fatal().Err(err).Msg("cannot get state commitment for block")
+			log.Fatal().Err(err).Msgf("cannot get state commitment for block %v", blockID)
 		}
 	}
 
@@ -108,6 +113,8 @@ func run(*cobra.Command, []string) {
 		if err != nil {
 			log.Fatal().Err(err).Msg("invalid state commitment length")
 		}
+
+		log.Info().Msgf("extracting state by state commitment: %x", stateCommitment)
 	}
 
 	if len(flagBlockHash) == 0 && len(flagStateCommitment) == 0 {
@@ -130,7 +137,20 @@ func run(*cobra.Command, []string) {
 		}
 	}
 
-	log.Info().Msgf("Block state commitment: %s", hex.EncodeToString(stateCommitment[:]))
+	log.Info().Msgf("Extracting state from %s, exporting root checkpoint to %s, version: %v",
+		flagExecutionStateDir,
+		path.Join(flagOutputDir, bootstrap.FilenameWALRootCheckpoint),
+		flagVersion)
+
+	log.Info().Msgf("Block state commitment: %s from %v, output dir: %s",
+		hex.EncodeToString(stateCommitment[:]),
+		flagExecutionStateDir,
+		flagOutputDir)
+
+	// err := ensureCheckpointFileExist(flagExecutionStateDir)
+	// if err != nil {
+	// 	log.Fatal().Err(err).Msgf("cannot ensure checkpoint file exist in folder %v", flagExecutionStateDir)
+	// }
 
 	chain, err := getChain(flagChain)
 	if err != nil {
@@ -143,6 +163,7 @@ func run(*cobra.Command, []string) {
 		flagOutputDir,
 		log.Logger,
 		chain,
+		flagVersion,
 		!flagNoMigration,
 		!flagNoReport,
 	)
@@ -151,3 +172,27 @@ func run(*cobra.Command, []string) {
 		log.Fatal().Err(err).Msgf("error extracting the execution state: %s", err.Error())
 	}
 }
+
+// func ensureCheckpointFileExist(dir string) error {
+// 	checkpoints, err := wal.Checkpoints(dir)
+// 	if err != nil {
+// 		return fmt.Errorf("could not find checkpoint files: %v", err)
+// 	}
+//
+// 	if len(checkpoints) != 0 {
+// 		log.Info().Msgf("found checkpoint %v files: %v", len(checkpoints), checkpoints)
+// 		return nil
+// 	}
+//
+// 	has, err := wal.HasRootCheckpoint(dir)
+// 	if err != nil {
+// 		return fmt.Errorf("could not check has root checkpoint: %w", err)
+// 	}
+//
+// 	if has {
+// 		log.Info().Msg("found root checkpoint file")
+// 		return nil
+// 	}
+//
+// 	return fmt.Errorf("no checkpoint file was found, no root checkpoint file was found")
+// }

cmd/util/cmd/execution-state-extract/execution_state_extract.go

@@ -28,12 +28,15 @@ func extractExecutionState(
 	outputDir string,
 	log zerolog.Logger,
 	chain flow.Chain,
+	version int, // to be removed after next spork
 	migrate bool,
 	report bool,
 ) error {
 
+	log.Info().Msg("init WAL")
+
 	diskWal, err := wal.NewDiskWAL(
-		zerolog.Nop(),
+		log,
 		nil,
 		metrics.NewNoopCollector(),
 		dir,
@@ -45,6 +48,8 @@ func extractExecutionState(
 		return fmt.Errorf("cannot create disk WAL: %w", err)
 	}
 
+	log.Info().Msg("init ledger")
+
 	led, err := complete.NewLedger(
 		diskWal,
 		complete.DefaultCacheSize,
@@ -60,11 +65,15 @@ func extractExecutionState(
 		checkpointsToKeep  = 1
 	)
 
-	compactor, err := complete.NewCompactor(led, diskWal, zerolog.Nop(), complete.DefaultCacheSize, checkpointDistance, checkpointsToKeep, atomic.NewBool(false))
+	log.Info().Msg("init compactor")
+
+	compactor, err := complete.NewCompactor(led, diskWal, log, complete.DefaultCacheSize, checkpointDistance, checkpointsToKeep, atomic.NewBool(false))
 	if err != nil {
 		return fmt.Errorf("cannot create compactor: %w", err)
 	}
 
+	log.Info().Msgf("waiting for compactor to load checkpoint and WAL")
+
 	<-compactor.Ready()
 
 	defer func() {
@@ -116,6 +125,7 @@ func extractExecutionState(
 		complete.DefaultPathFinderVersion,
 		outputDir,
 		bootstrap.FilenameWALRootCheckpoint,
+		version,
 	)
 	if err != nil {
 		return fmt.Errorf("cannot generate the output checkpoint: %w", err)

cmd/util/cmd/execution-state-extract/execution_state_extract_test.go

@@ -65,6 +65,7 @@ func TestExtractExecutionState(t *testing.T) {
 				outdir,
 				zerolog.Nop(),
 				flow.Emulator.Chain(),
+				6,
 				false,
 				false,
 			)