[Access] Access ingestion error handle

cmd/access/node_builder/access_node_builder.go

@@ -2281,6 +2281,7 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				notNil(builder.CollectionSyncer),
 				notNil(builder.CollectionIndexer),
 				notNil(builder.collectionExecutedMetric),
+				builder.AccessMetrics,
 				notNil(builder.TxResultErrorMessagesCore),
 				builder.FollowerDistributor,
 			)

engine/access/access_test.go

@@ -787,6 +787,7 @@ func (suite *Suite) TestGetSealedTransaction() {
 			collectionSyncer,
 			collectionIndexer,
 			collectionExecutedMetric,
+			suite.metrics,
 			nil,
 			followerDistributor,
 		)
@@ -1052,6 +1053,7 @@ func (suite *Suite) TestGetTransactionResult() {
 			collectionSyncer,
 			collectionIndexer,
 			collectionExecutedMetric,
+			suite.metrics,
 			nil,
 			followerDistributor,
 		)
@@ -1324,6 +1326,7 @@ func (suite *Suite) TestExecuteScript() {
 			collectionSyncer,
 			collectionIndexer,
 			collectionExecutedMetric,
+			suite.metrics,
 			nil,
 			followerDistributor,
 		)

engine/access/ingestion/engine.go

@@ -2,6 +2,7 @@ package ingestion
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"github.com/jordanschalm/lockctx"
@@ -73,6 +74,7 @@ type Engine struct {
 	// TODO: There's still a need for this metric to be in the ingestion engine rather than collection syncer.
 	// Maybe it is a good idea to split it up?
 	collectionExecutedMetric module.CollectionExecutedMetric
+	accessMetrics            module.AccessMetrics
 
 	txErrorMessagesCore *tx_error_messages.TxErrorMessagesCore
 }
@@ -96,6 +98,7 @@ func New(
 	collectionSyncer *collections.Syncer,
 	collectionIndexer *collections.Indexer,
 	collectionExecutedMetric module.CollectionExecutedMetric,
+	accessMetrics module.AccessMetrics,
 	txErrorMessagesCore *tx_error_messages.TxErrorMessagesCore,
 	registrar hotstuff.FinalizationRegistrar,
 ) (*Engine, error) {
@@ -130,6 +133,7 @@ func New(
 		executionReceipts:        executionReceipts,
 		maxReceiptHeight:         0,
 		collectionExecutedMetric: collectionExecutedMetric,
+		accessMetrics:            accessMetrics,
 		finalizedBlockNotifier:   engine.NewNotifier(),
 
 		// queue / notifier for execution receipts
@@ -230,12 +234,15 @@ func (e *Engine) processFinalizedBlockJob(ctx irrecoverable.SignalerContext, job
 	}
 
 	err = e.processFinalizedBlock(block)
-	if err == nil {
-		done()
+	if err != nil {
+		ctx.Throw(
+			fmt.Errorf(
+				"fatal error when ingestion building col->block index for finalized block (job: %s, height: %v): %w",
+				job.ID(), block.Height, err))
 		return
 	}
 
-	e.log.Error().Err(err).Str("job_id", string(job.ID())).Msg("error during finalized block processing job")
+	done()
 }
 
 // processExecutionReceipts is responsible for processing the execution receipts.
@@ -353,10 +360,10 @@ func (e *Engine) onFinalizedBlock(*model.Block) {
 
 // processFinalizedBlock handles an incoming finalized block.
 // It processes the block, indexes it for further processing, and requests missing collections if necessary.
+// If the block is already indexed (storage.ErrAlreadyExists), it logs a warning and continues processing.
 //
 // Expected errors during normal operation:
 //   - storage.ErrNotFound - if last full block height does not exist in the database.
-//   - storage.ErrAlreadyExists - if the collection within block or an execution result ID already exists in the database.
 //   - generic error in case of unexpected failure from the database layer, or failure
 //     to decode an existing database value.
 func (e *Engine) processFinalizedBlock(block *flow.Block) error {
@@ -386,14 +393,23 @@ func (e *Engine) processFinalizedBlock(block *flow.Block) error {
 		})
 	})
 	if err != nil {
-		return fmt.Errorf("could not index block for collections: %w", err)
+		if !errors.Is(err, storage.ErrAlreadyExists) {
+			return fmt.Errorf("could not index block for collections: %w", err)
+		}
+		// the job queue processed index is updated in a separate db update, so it's possible that the above index
+		// has been built, but the jobqueue index has not been updated yet. In this case, we can safely skip processing.
+		e.log.Warn().
+			Uint64("height", block.Height).
+			Str("block_id", block.ID().String()).
+			Msg("block already indexed, skipping indexing")
 	}
 
 	err = e.collectionSyncer.RequestCollectionsForBlock(block.Height, block.Payload.Guarantees)
 	if err != nil {
 		return fmt.Errorf("could not request collections for block: %w", err)
 	}
 	e.collectionExecutedMetric.BlockFinalized(block)
+	e.accessMetrics.UpdateIngestionFinalizedBlockHeight(block.Height)
 
 	return nil
 }

engine/access/ingestion/engine_test.go

@@ -221,6 +221,7 @@ func (s *Suite) initEngineAndSyncer() (*Engine, *collections.Syncer, *collection
 		syncer,
 		indexer,
 		s.collectionExecutedMetric,
+		metrics.NewNoopCollector(),
 		nil,
 		s.distributor,
 	)
@@ -567,6 +568,77 @@ func (s *Suite) TestCollectionSyncing() {
 	}, 2*time.Second, 100*time.Millisecond, "last full block height never updated")
 }
 
+// TestOnFinalizedBlockAlreadyIndexed checks that when a block has already been indexed
+// (storage.ErrAlreadyExists), the engine logs a warning and continues processing by
+// requesting collections and updating metrics. This can happen when the job queue processed
+// index hasn't been updated yet after a previous indexing operation.
+func (s *Suite) TestOnFinalizedBlockAlreadyIndexed() {
+	cluster := new(protocolmock.Cluster)
+	epoch := new(protocolmock.CommittedEpoch)
+	epochs := new(protocolmock.EpochQuery)
+	snap := new(protocolmock.Snapshot)
+
+	finalSnapshot := protocolmock.NewSnapshot(s.T())
+	finalSnapshot.On("Head").Return(s.finalizedBlock, nil).Twice()
+	s.proto.state.On("Final").Return(finalSnapshot, nil).Twice()
+
+	epoch.On("ClusterByChainID", mock.Anything).Return(cluster, nil)
+	epochs.On("Current").Return(epoch, nil)
+	snap.On("Epochs").Return(epochs)
+
+	// prepare cluster committee members
+	clusterCommittee := unittest.IdentityListFixture(32 * 4).Filter(filter.HasRole[flow.Identity](flow.RoleCollection)).ToSkeleton()
+	cluster.On("Members").Return(clusterCommittee, nil)
+
+	eng, _, _ := s.initEngineAndSyncer()
+
+	irrecoverableCtx, cancel := irrecoverable.NewMockSignalerContextWithCancel(s.T(), s.ctx)
+	eng.ComponentManager.Start(irrecoverableCtx)
+	unittest.RequireCloseBefore(s.T(), eng.Ready(), 100*time.Millisecond, "could not start worker")
+	defer func() {
+		cancel()
+		unittest.RequireCloseBefore(s.T(), eng.Done(), 100*time.Millisecond, "could not stop worker")
+	}()
+
+	block := s.generateBlock(clusterCommittee, snap)
+	block.Height = s.finalizedBlock.Height + 1
+	s.blockMap[block.Height] = block
+	s.mockCollectionsForBlock(block)
+	s.finalizedBlock = block.ToHeader()
+
+	hotstuffBlock := hotmodel.Block{
+		BlockID: block.ID(),
+	}
+
+	// simulate that the block has already been indexed (e.g., by a previous job queue run)
+	// by returning storage.ErrAlreadyExists
+	s.blocks.On("BatchIndexBlockContainingCollectionGuarantees", mock.Anything, mock.Anything, block.ID(), []flow.Identifier(flow.GetIDs(block.Payload.Guarantees))).
+		Return(storage.ErrAlreadyExists).Once()
+
+	missingCollectionCount := 4
+	wg := sync.WaitGroup{}
+	wg.Add(missingCollectionCount)
+
+	// even though block is already indexed, collections should still be requested
+	for _, cg := range block.Payload.Guarantees {
+		s.request.On("EntityByID", cg.CollectionID, mock.Anything).Return().Run(func(args mock.Arguments) {
+			wg.Done()
+		}).Once()
+	}
+
+	// force should be called once
+	s.request.On("Force").Return().Once()
+
+	// process the block through the finalized callback
+	s.distributor.OnFinalizedBlock(&hotstuffBlock)
+
+	unittest.RequireReturnsBefore(s.T(), wg.Wait, 100*time.Millisecond, "expect to process new block before timeout")
+
+	// assert that collections were still requested despite the block being already indexed
+	s.headers.AssertExpectations(s.T())
+	s.request.AssertNumberOfCalls(s.T(), "EntityByID", len(block.Payload.Guarantees))
+}
+
 func (s *Suite) mockGuarantorsForCollection(guarantee *flow.CollectionGuarantee, members flow.IdentitySkeletonList) {
 	cluster := protocolmock.NewCluster(s.T())
 	cluster.On("Members").Return(members, nil).Once()

module/metrics.go

@@ -940,6 +940,9 @@ type AccessMetrics interface {
 
 	// UpdateLastFullBlockHeight tracks the height of the last block for which all collections were received
 	UpdateLastFullBlockHeight(height uint64)
+
+	// UpdateIngestionFinalizedBlockHeight tracks the latest finalized block height processed by ingestion
+	UpdateIngestionFinalizedBlockHeight(height uint64)
 }
 
 type ExecutionResultStats struct {

module/metrics/access.go

@@ -40,15 +40,16 @@ type AccessCollector struct {
 	module.TransactionValidationMetrics
 	module.BackendScriptsMetrics
 
-	connectionReused      prometheus.Counter
-	connectionsInPool     *prometheus.GaugeVec
-	connectionAdded       prometheus.Counter
-	connectionEstablished prometheus.Counter
-	connectionInvalidated prometheus.Counter
-	connectionUpdated     prometheus.Counter
-	connectionEvicted     prometheus.Counter
-	lastFullBlockHeight   prometheus.Gauge
-	maxReceiptHeight      prometheus.Gauge
+	connectionReused              prometheus.Counter
+	connectionsInPool             *prometheus.GaugeVec
+	connectionAdded               prometheus.Counter
+	connectionEstablished         prometheus.Counter
+	connectionInvalidated         prometheus.Counter
+	connectionUpdated             prometheus.Counter
+	connectionEvicted             prometheus.Counter
+	lastFullBlockHeight           prometheus.Gauge
+	ingestionFinalizedBlockHeight prometheus.Gauge
+	maxReceiptHeight              prometheus.Gauge
 
 	// used to skip heights that are lower than the current max height
 	maxReceiptHeightValue counters.StrictMonotonicCounter
@@ -106,6 +107,12 @@ func NewAccessCollector(opts ...AccessCollectorOpts) *AccessCollector {
 			Subsystem: subsystemIngestion,
 			Help:      "gauge to track the highest consecutive finalized block height with all collections indexed",
 		}),
+		ingestionFinalizedBlockHeight: promauto.NewGauge(prometheus.GaugeOpts{
+			Name:      "ingestion_finalized_block_height",
+			Namespace: namespaceAccess,
+			Subsystem: subsystemIngestion,
+			Help:      "gauge to track the latest finalized block height processed by ingestion",
+		}),
 		maxReceiptHeight: promauto.NewGauge(prometheus.GaugeOpts{
 			Name:      "max_receipt_height",
 			Namespace: namespaceAccess,
@@ -155,6 +162,10 @@ func (ac *AccessCollector) UpdateLastFullBlockHeight(height uint64) {
 	ac.lastFullBlockHeight.Set(float64(height))
 }
 
+func (ac *AccessCollector) UpdateIngestionFinalizedBlockHeight(height uint64) {
+	ac.ingestionFinalizedBlockHeight.Set(float64(height))
+}
+
 func (ac *AccessCollector) UpdateExecutionReceiptMaxHeight(height uint64) {
 	if ac.maxReceiptHeightValue.Set(height) {
 		ac.maxReceiptHeight.Set(float64(height))

module/metrics/noop.go

@@ -229,6 +229,7 @@ func (nc *NoopCollector) TransactionValidationSkipped()
 func (nc *NoopCollector) TransactionSubmissionFailed()                                          {}
 func (nc *NoopCollector) UpdateExecutionReceiptMaxHeight(height uint64)                         {}
 func (nc *NoopCollector) UpdateLastFullBlockHeight(height uint64)                               {}
+func (nc *NoopCollector) UpdateIngestionFinalizedBlockHeight(height uint64)                     {}
 func (nc *NoopCollector) ChunkDataPackRequestProcessed()                                        {}
 func (nc *NoopCollector) ExecutionSync(syncing bool)                                            {}
 func (nc *NoopCollector) ExecutionBlockDataUploadStarted()                                      {}