[EN Performance] Reuse ledger state for about -200GB peak RAM, -160GB disk i/o, and about -32 minutes duration

cmd/bootstrap/run/execution_state.go

@@ -41,14 +41,14 @@ func GenerateExecutionState(
 	if err != nil {
 		return flow.DummyStateCommitment, err
 	}
-	defer func() {
-		<-diskWal.Done()
-	}()
 
 	ledgerStorage, err := ledger.NewLedger(diskWal, 100, metricsCollector, zerolog.Nop(), ledger.DefaultPathFinderVersion)
 	if err != nil {
 		return flow.DummyStateCommitment, err
 	}
+	defer func() {
+		<-ledgerStorage.Done()
+	}()
 
 	return bootstrap.NewBootstrapper(
 		zerolog.Nop()).BootstrapLedger(

cmd/execution_builder.go

@@ -343,12 +343,6 @@ func (e *ExecutionNodeBuilder) LoadComponentsAndModules() {
 			}
 			return nil
 		}).
-		Component("Write-Ahead Log", func(node *NodeConfig) (module.ReadyDoneAware, error) {
-			var err error
-			diskWAL, err = wal.NewDiskWAL(node.Logger.With().Str("subcomponent", "wal").Logger(),
-				node.MetricsRegisterer, collector, e.exeConf.triedir, int(e.exeConf.mTrieCacheSize), pathfinder.PathByteSize, wal.SegmentSize)
-			return diskWAL, err
-		}).
 		Component("execution state ledger", func(node *NodeConfig) (module.ReadyDoneAware, error) {
 
 			// check if the execution database already exists
@@ -384,23 +378,27 @@ func (e *ExecutionNodeBuilder) LoadComponentsAndModules() {
 				}
 			}
 
-			ledgerStorage, err = ledger.NewLedger(diskWAL, int(e.exeConf.mTrieCacheSize), collector, node.Logger.With().Str("subcomponent",
+			// Ledger is responsible for starting and shutdowning DiskWAL component.
+			// This ensures that all WAL updates are completed before closing opened WAL segment.
+			diskWAL, err = wal.NewDiskWAL(node.Logger.With().Str("subcomponent", "wal").Logger(),
+				node.MetricsRegisterer, collector, e.exeConf.triedir, int(e.exeConf.mTrieCacheSize), pathfinder.PathByteSize, wal.SegmentSize)
+			if err != nil {
+				return nil, fmt.Errorf("failed to initialize wal: %w", err)
+			}
+
+			ledgerStorage, err = ledger.NewSyncLedger(diskWAL, int(e.exeConf.mTrieCacheSize), collector, node.Logger.With().Str("subcomponent",
 				"ledger").Logger(), ledger.DefaultPathFinderVersion)
 			return ledgerStorage, err
 		}).
 		Component("execution state ledger WAL compactor", func(node *NodeConfig) (module.ReadyDoneAware, error) {
 
-			checkpointer, err := ledgerStorage.Checkpointer()
-			if err != nil {
-				return nil, fmt.Errorf("cannot create checkpointer: %w", err)
-			}
-			compactor := wal.NewCompactor(checkpointer,
-				10*time.Second,
+			return ledger.NewCompactor(
+				ledgerStorage,
+				diskWAL,
 				e.exeConf.checkpointDistance,
 				e.exeConf.checkpointsToKeep,
-				node.Logger.With().Str("subcomponent", "checkpointer").Logger())
-
-			return compactor, nil
+				node.Logger.With().Str("subcomponent", "checkpointer").Logger(),
+			)
 		}).
 		Component("execution data service", func(node *NodeConfig) (module.ReadyDoneAware, error) {
 			err := os.MkdirAll(e.exeConf.executionDataDir, 0700)

cmd/util/cmd/exec-data-json-export/ledger_exporter.go

@@ -24,13 +24,13 @@ func ExportLedger(ledgerPath string, targetstate string, outputPath string) erro
 	if err != nil {
 		return fmt.Errorf("cannot create WAL: %w", err)
 	}
-	defer func() {
-		<-diskWal.Done()
-	}()
 	led, err := complete.NewLedger(diskWal, complete.DefaultCacheSize, &metrics.NoopCollector{}, log.Logger, 0)
 	if err != nil {
 		return fmt.Errorf("cannot create ledger from write-a-head logs and checkpoints: %w", err)
 	}
+	defer func() {
+		<-led.Done()
+	}()
 	stateBytes, err := hex.DecodeString(targetstate)
 	if err != nil {
 		return fmt.Errorf("failed to decode hex code of state: %w", err)

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

@@ -43,9 +43,6 @@ func extractExecutionState(
 	if err != nil {
 		return fmt.Errorf("cannot create disk WAL: %w", err)
 	}
-	defer func() {
-		<-diskWal.Done()
-	}()
 
 	led, err := complete.NewLedger(
 		diskWal,
@@ -56,6 +53,9 @@ func extractExecutionState(
 	if err != nil {
 		return fmt.Errorf("cannot create ledger from write-a-head logs and checkpoints: %w", err)
 	}
+	defer func() {
+		<-led.Done()
+	}()
 
 	var migrations []ledger.Migration
 	var preCheckpointReporters, postCheckpointReporters []ledger.Reporter

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

@@ -120,7 +120,6 @@ func TestExtractExecutionState(t *testing.T) {
 				blocksInOrder[i] = blockID
 			}
 
-			<-diskWal.Done()
 			<-f.Done()
 			err = db.Close()
 			require.NoError(t, err)
@@ -181,7 +180,6 @@ func TestExtractExecutionState(t *testing.T) {
 						require.Error(t, err)
 					}
 
-					<-diskWal.Done()
 					<-storage.Done()
 				})
 			}

ledger/complete/checkpoint_benchmark_test.go

@@ -284,7 +284,6 @@ func benchmarkNewCheckpointRandomData(b *testing.B, segmentCount int) {
 		b.Fatal(err)
 	}
 
-	<-wal1.Done()
 	<-led.Done()
 
 	wal2, err := wal.NewDiskWAL(