Merge pull request onflow#5151 from onflow/leo/enable-unfinalized-loader

[Storehouse] Use unfinalized loader when storehouse is enabled

cmd/execution_builder.go

@@ -958,7 +958,12 @@ func (exeNode *ExecutionNode) LoadIngestionEngine(
 	}
 
 	fetcher := fetcher.NewCollectionFetcher(node.Logger, exeNode.collectionRequester, node.State, exeNode.exeConf.onflowOnlyLNs)
-	loader := loader.NewUnexecutedLoader(node.Logger, node.State, node.Storage.Headers, exeNode.executionState)
+	var blockLoader ingestion.BlockLoader
+	if exeNode.exeConf.enableStorehouse {
+		blockLoader = loader.NewUnfinalizedLoader(node.Logger, node.State, node.Storage.Headers, exeNode.executionState)
+	} else {
+		blockLoader = loader.NewUnexecutedLoader(node.Logger, node.State, node.Storage.Headers, exeNode.executionState)
+	}
 
 	exeNode.ingestionEng, err = ingestion.New(
 		exeNode.ingestionUnit,
@@ -978,7 +983,7 @@ func (exeNode *ExecutionNode) LoadIngestionEngine(
 		exeNode.executionDataPruner,
 		exeNode.blockDataUploader,
 		exeNode.stopControl,
-		loader,
+		blockLoader,
 	)
 
 	// TODO: we should solve these mutual dependencies better

engine/execution/ingestion/engine.go

@@ -202,7 +202,7 @@ func (e *Engine) reloadUnexecutedBlocks() error {
 			e.log.Debug().Hex("block_id", blockID[:]).Msg("reloaded block")
 		}
 
-		log.Info().Msg("all unexecuted have been successfully reloaded")
+		e.log.Info().Int("count", len(unexecuted)).Msg("all unexecuted have been successfully reloaded")
 
 		return nil
 	})
@@ -396,6 +396,8 @@ func (e *Engine) enqueueBlockAndCheckExecutable(
 		Uint64("first_unexecuted_in_queue", firstUnexecutedHeight).
 		Bool("complete", complete).
 		Bool("head_of_queue", head).
+		Int("cols", len(executableBlock.Block.Payload.Guarantees)).
+		Int("missing_cols", len(missingCollections)).
 		Msg("block is enqueued")
 
 	return missingCollections, nil

engine/execution/ingestion/loader/unfinalized_loader.go

@@ -46,6 +46,13 @@ func (e *UnfinalizedLoader) LoadUnexecuted(ctx context.Context) ([]flow.Identifi
 		return nil, fmt.Errorf("could not get finalized block: %w", err)
 	}
 
+	lg := e.log.With().
+		Uint64("last_finalized", final.Height).
+		Uint64("last_finalized_executed", lastExecuted).
+		Logger()
+
+	lg.Info().Msgf("start loading unfinalized blocks")
+
 	// TODO: dynamically bootstrapped execution node will reload blocks from
 	unexecutedFinalized := make([]flow.Identifier, 0)
 
@@ -69,9 +76,7 @@ func (e *UnfinalizedLoader) LoadUnexecuted(ctx context.Context) ([]flow.Identifi
 
 	unexecuted := append(unexecutedFinalized, pendings...)
 
-	e.log.Info().
-		Uint64("last_finalized", final.Height).
-		Uint64("last_finalized_executed", lastExecuted).
+	lg.Info().
 		// Uint64("sealed_root_height", rootBlock.Height).
 		// Hex("sealed_root_id", logging.Entity(rootBlock)).
 		Int("total_finalized_unexecuted", len(unexecutedFinalized)).

engine/execution/state/state.go

@@ -84,6 +84,10 @@ type ExecutionState interface {
 		ctx context.Context,
 		result *execution.ComputationResult,
 	) error
+
+	// only available with storehouse enabled
+	// panic when called with storehouse disabled (which should be a bug)
+	GetHighestFinalizedExecuted() uint64
 }
 
 type state struct {