services/ingestion/block_tracking_subscriber.go (1)

137-142: Simplify loop by removing redundant context check.

Since you're already handling context cancellation within the select statement using <-ctx.Done(), the loop condition for ctx.Err() == nil is redundant. Simplify the loop by using for {} to improve code readability.

Apply this diff to simplify the loop:

-       for ctx.Err() == nil {
+       for {
            select {
            case <-ctx.Done():
                r.logger.Info().Msg("event ingestion received done signal")
                return
            // ...

services/requester/cross-spork_client.go (1)

262-265: Avoid type assertion to *grpc.Client; use an interface to ensure compatibility.

Direct type assertion to *grpc.Client can be fragile and may break if the underlying implementation changes. Consider defining an interface that includes the SubscribeBlockHeadersFromStartHeight method and updating your clients to implement this interface. This approach enhances flexibility and maintainability.

For example, define an interface:

type BlockHeaderSubscriber interface {
    SubscribeBlockHeadersFromStartHeight(ctx context.Context, startHeight uint64, blockStatus flow.BlockStatus) (<-chan flow.BlockHeader, <-chan error, error)
}

And ensure that your clients implement this interface. Then, modify the code:

-       grpcClient, ok := (client).(*grpc.Client)
-       if !ok {
-           return nil, nil, fmt.Errorf("unable to convert to Flow grpc.Client")
-       }

-       return grpcClient.SubscribeBlockHeadersFromStartHeight(ctx, startHeight, blockStatus)
+       subscriberClient, ok := client.(BlockHeaderSubscriber)
+       if !ok {
+           return nil, nil, fmt.Errorf("client does not implement BlockHeaderSubscriber interface")
+       }

+       return subscriberClient.SubscribeBlockHeadersFromStartHeight(ctx, startHeight, blockStatus)

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

Makefile (2)

236-238: Maintain configuration consistency with existing patterns.

The Makefile already defines variables for spork hosts (e.g., TESTNET_ACCESS_NODE_SPORK_HOSTS). Follow this pattern for other configuration values like access node hosts.

Apply this diff to improve configuration consistency:

 # Add at the top with other variables
+TESTNET_ACCESS_NODE_HOST := access.devnet.nodes.onflow.org:9000
+MAINNET_ACCESS_NODE_HOST := access.mainnet.nodes.onflow.org:9000

 .PHONY: start-testnet
 start-testnet:
        ...
-               --access-node-grpc-host=access.devnet.nodes.onflow.org:9000 \
-               --access-node-spork-hosts=access-001.devnet51.nodes.onflow.org:9000 \
+               --access-node-grpc-host=$(TESTNET_ACCESS_NODE_HOST) \
+               --access-node-spork-hosts=$(TESTNET_ACCESS_NODE_SPORK_HOSTS) \
        ...

 .PHONY: start-mainnet
 start-mainnet:
        ...
-               --access-node-grpc-host=access.mainnet.nodes.onflow.org:9000 \
-               --access-node-spork-hosts=access-001.mainnet25.nodes.onflow.org:9000 \
+               --access-node-grpc-host=$(MAINNET_ACCESS_NODE_HOST) \
+               --access-node-spork-hosts=$(MAINNET_ACCESS_NODE_SPORK_HOSTS) \
        ...

Also applies to: 253-254

---

239-239: Consider making initialization heights configurable.

The initialization heights are hardcoded. Consider making them configurable through variables to simplify future updates.

Apply this diff to improve configurability:

 # Add at the top with other variables
+TESTNET_INIT_CADENCE_HEIGHT := 211176670
+MAINNET_INIT_CADENCE_HEIGHT := 85981135

 .PHONY: start-testnet
 start-testnet:
        ...
-               --init-cadence-height=211176670 \
+               --init-cadence-height=$(TESTNET_INIT_CADENCE_HEIGHT) \
        ...

 .PHONY: start-mainnet
 start-mainnet:
        ...
-               --init-cadence-height=85981135 \
+               --init-cadence-height=$(MAINNET_INIT_CADENCE_HEIGHT) \
        ...

Also applies to: 256-256

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

services/ingestion/block_tracking_subscriber.go (2)

21-29: ⚠️ Potential issue

Add documentation about limitations and ensure thread safety.

The struct requires important documentation about its limitations and potential risks. Additionally, the height field needs thread-safe access as it's modified from multiple goroutines.

Apply this diff to add documentation and ensure thread-safety:

+// RPCBlockTrackingSubscriber subscribes to new EVM block events for unsealed finalized blocks.
+// This is accomplished by following finalized blocks from the upstream Access node, and using the
+// polling endpoint to fetch the events for each finalized block.
+//
+// IMPORTANT: Since data is downloaded and processed from unsealed blocks, it's possible for the
+// data that was downloaded to be incorrect. This subscriber provides no handling or detection for
+// cases where the received data differs from the data that was ultimately sealed. The operator must
+// handle this manually.
+// Since it's not reasonable to expect operators to do this manual tracking, this features should NOT
+// be used outside of a limited Proof of Concept. Use at own risk.
+//
+// A future version of the RPCEventSubscriber will provide this detection and handling functionality
+// at which point this subscriber will be removed.
 type RPCBlockTrackingSubscriber struct {
 	*RPCEventSubscriber
 
-	logger  zerolog.Logger
-	client  *requester.CrossSporkClient
-	chain   flowGo.ChainID
-	keyLock requester.KeyLock
-	height  uint64
+	logger   zerolog.Logger
+	client   *requester.CrossSporkClient
+	chain    flowGo.ChainID
+	keyLock  requester.KeyLock
+	height   uint64
+	heightMu sync.RWMutex
}

---

192-194: ⚠️ Potential issue

Use error types instead of string matching for error handling.

Using strings.Contains to check error messages is not robust and can lead to unexpected behavior if the error messages change.

Apply this diff to improve error handling:

-				if strings.Contains(errors.Unwrap(err).Error(), "DeadlineExceeded") ||
-					strings.Contains(errors.Unwrap(err).Error(), "unexpected EOF") {
+				unwrappedErr := errors.Unwrap(err)
+				if errors.Is(unwrappedErr, context.DeadlineExceeded) ||
+					errors.Is(unwrappedErr, io.EOF) {

services/ingestion/block_tracking_subscriber.go (3)

31-55: Optimize constructor to avoid field duplication.

The constructor duplicates fields that are already available from the embedded RPCEventSubscriber.

Apply this diff to simplify the constructor:

 func NewRPCBlockTrackingSubscriber(
 	logger zerolog.Logger,
 	client *requester.CrossSporkClient,
 	chainID flowGo.ChainID,
 	keyLock requester.KeyLock,
 	startHeight uint64,
 ) *RPCBlockTrackingSubscriber {
-	eventSubscriber := NewRPCEventSubscriber(
-		logger,
-		client,
-		chainID,
-		keyLock,
-		startHeight,
-	)
-	logger = logger.With().Str("component", "subscriber").Logger()
-
 	return &RPCBlockTrackingSubscriber{
-		RPCEventSubscriber: eventSubscriber,
-		logger:             logger,
-		client:             client,
-		chain:              chainID,
-		keyLock:            keyLock,
-		height:             startHeight,
+		RPCEventSubscriber: NewRPCEventSubscriber(
+			logger.With().Str("component", "subscriber").Logger(),
+			client,
+			chainID,
+			keyLock,
+			startHeight,
+		),
+		height: startHeight,
 	}
 }

---

62-64: Document the rationale for channel buffer size.

The channel buffer size of 1000 seems arbitrary. Consider documenting why this specific size was chosen and what happens when the buffer is full.

+// buffered channel with size 1000 to handle bursts of events during backfilling
+// without blocking the producer. If the buffer is full, the producer will block
+// until the consumer catches up.
 eventsChan := make(chan models.BlockEvents, 1000)

---

220-283: Refactor the method for better readability.

The method is complex and could benefit from being broken down into smaller, more focused functions.

Consider refactoring as suggested in the past review comments to improve readability and maintainability. The refactored version should:

1. Split event retrieval into a separate function
2. Handle block and transaction events separately
3. Improve error messages for clarity

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

services/ingestion/block_tracking_subscriber.go (3)

85-88: ⚠️ Potential issue

Use mutex when updating height field during backfill.

The height field is updated without proper synchronization during backfill.

Add mutex protection:

+				r.heightMu.Lock()
 				r.height = ev.Events.CadenceHeight()
+				r.heightMu.Unlock()

---

91-93: ⚠️ Potential issue

Use mutex when incrementing height field.

The height field is incremented without proper synchronization.

Add mutex protection:

+			r.heightMu.Lock()
 			r.height = r.height + 1
+			r.heightMu.Unlock()

---

226-231: ⚠️ Potential issue

Use GetEventsForBlockIDs instead of GetEventsForHeightRange.

Using GetEventsForHeightRange restricts the request to sealed blocks, which defeats the purpose of this change.

Replace with GetEventsForBlockIDs:

-	evts, err := r.client.GetEventsForHeightRange(
+	blockID, err := r.client.GetBlockIDAtHeight(ctx, height)
+	if err != nil {
+		return flow.BlockEvents{}, err
+	}
+
+	evts, err := r.client.GetEventsForBlockIDs(
 		ctx,
 		evmBlockEvent,
-		height,
-		height,
+		[]flow.Identifier{blockID},
 	)

services/ingestion/block_tracking_subscriber.go (3)

38-54: Simplify constructor to avoid field duplication.

The constructor duplicates fields that are already available from the embedded RPCEventSubscriber.

Simplify the constructor:

-	eventSubscriber := NewRPCEventSubscriber(
-		logger,
-		client,
-		chainID,
-		keyLock,
-		startHeight,
-	)
-	logger = logger.With().Str("component", "subscriber").Logger()
-
-	return &RPCBlockTrackingSubscriber{
-		RPCEventSubscriber: eventSubscriber,
-		logger:             logger,
-		client:             client,
-		chain:              chainID,
-		keyLock:            keyLock,
-		height:             startHeight,
-	}
+	return &RPCBlockTrackingSubscriber{
+		RPCEventSubscriber: NewRPCEventSubscriber(
+			logger.With().Str("component", "subscriber").Logger(),
+			client,
+			chainID,
+			keyLock,
+			startHeight,
+		),
+	}

---

64-64: Document the channel buffer size choice.

The channel buffer size of 1000 seems arbitrary. Consider documenting the reasoning behind this choice or making it configurable.

Add a comment explaining the buffer size choice:

+	// Use a buffer size of 1000 to allow for parallel processing without blocking
+	// while maintaining reasonable memory usage
 	eventsChan := make(chan models.BlockEvents, 1000)

---

219-282: Refactor method for better clarity and maintainability.

The method could be split into smaller, more focused functions for better readability and maintainability.

Consider refactoring as suggested in the past review:

+func (r *RPCBlockTrackingSubscriber) evmEventsForBlock(
+	ctx context.Context,
+	blockID flow.Identifier,
+) (flow.BlockEvents, error) {
+	eventTypes := blocksFilter(r.chain).EventTypes
+
+	// evm Block events
+	blockEvents, err := r.getEventsByType(ctx, blockID, eventTypes[0])
+	if err != nil {
+		return flow.BlockEvents{}, err
+	}
+
+	payload, err := events.DecodeBlockEventPayload(blockEvents.Events[0].Value)
+	if err != nil {
+		return flow.BlockEvents{}, err
+	}
+
+	if payload.TransactionHashRoot == types.EmptyTxsHash {
+		return blockEvents, nil
+	}
+
+	// evm TX events
+	txEvents, err := r.getEventsByType(ctx, blockID, eventTypes[1])
+	if err != nil {
+		return flow.BlockEvents{}, err
+	}
+
+	// combine block and tx events to be processed together
+	blockEvents.Events = append(blockEvents.Events, txEvents.Events...)
+
+	return blockEvents, nil
+}
+
+func (r *RPCBlockTrackingSubscriber) getEventsByType(
+	ctx context.Context,
+	blockID flow.Identifier,
+	eventType string,
+) (flow.BlockEvents, error) {
+	evts, err := r.client.GetEventsForBlockIDs(
+		ctx,
+		eventType,
+		[]flow.Identifier{blockID},
+	)
+	if err != nil {
+		return flow.BlockEvents{}, err
+	}
+
+	if len(evts) != 1 {
+		return flow.BlockEvents{}, fmt.Errorf(
+			"received unexpected number of block events: got: %d, expected: 1",
+			len(evts),
+		)
+	}
+
+	if len(evts[0].Events) != 1 {
+		return flow.BlockEvents{}, fmt.Errorf(
+			"received unexpected number of events in block: got: %d, expected: 1",
+			len(evts[0].Events),
+		)
+	}
+
+	return evts[0], nil
+}

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

services/ingestion/block_tracking_subscriber.go (2)

21-29: ⚠️ Potential issue

Ensure thread-safe access to height field.

The height field is accessed concurrently in multiple goroutines without proper synchronization, which could lead to race conditions.

Add a mutex to protect the height field:

 type RPCBlockTrackingSubscriber struct {
     *RPCEventSubscriber

     logger  zerolog.Logger
     client  *requester.CrossSporkClient
     chain   flowGo.ChainID
     keyLock requester.KeyLock
+    heightMu sync.RWMutex
     height  uint64
 }

---

211-274: 🛠️ Refactor suggestion

⚠️ Potential issue

Refactor the method and use GetEventsForBlockIDs.

The method needs several improvements:

1. Use GetEventsForBlockIDs instead of GetEventsForHeightRange
2. Extract common error handling logic
3. Break down the method into smaller functions

Apply this diff to improve the implementation:

 func (r *RPCBlockTrackingSubscriber) evmEventsForHeight(
     ctx context.Context,
     height uint64,
 ) (flow.BlockEvents, error) {
     eventTypes := blocksFilter(r.chain).EventTypes
-    evmBlockEvent := eventTypes[0]
-
-    evts, err := r.client.GetEventsForHeightRange(
+    blockID, err := r.client.GetBlockIDAtHeight(ctx, height)
+    if err != nil {
+        return flow.BlockEvents{}, err
+    }
+
+    blockEvents, err := r.getEventsByType(
         ctx,
-        evmBlockEvent,
-        height,
-        height,
+        blockID,
+        eventTypes[0],
     )
     if err != nil {
         return flow.BlockEvents{}, err
     }

-    if len(evts) != 1 || len(evts[0].Events) != 1 {
-        return flow.BlockEvents{}, fmt.Errorf(
-            "received unexpected number of EVM events for height: %d, got: %d, expected: 1",
-            height,
-            len(evts),
-        )
-    }
-
-    blockEvents := evts[0]
     payload, err := events.DecodeBlockEventPayload(blockEvents.Events[0].Value)
     if err != nil {
         return flow.BlockEvents{}, err
     }

     if payload.TransactionHashRoot == types.EmptyTxsHash {
         return blockEvents, nil
     }

-    evmTxEvent := eventTypes[1]
-    evts, err = r.client.GetEventsForHeightRange(
+    txEvents, err := r.getEventsByType(
         ctx,
-        evmTxEvent,
-        height,
-        height,
+        blockID,
+        eventTypes[1],
     )
     if err != nil {
         return flow.BlockEvents{}, err
     }

-    if len(evts) != 1 {
-        return flow.BlockEvents{}, fmt.Errorf(
-            "received unexpected number of EVM events for height: %d, got: %d, expected: 1",
-            height,
-            len(evts),
-        )
-    }
-    txEvents := evts[0]
-
     blockEvents.Events = append(blockEvents.Events, txEvents.Events...)

     return blockEvents, nil
 }
+
+func (r *RPCBlockTrackingSubscriber) getEventsByType(
+    ctx context.Context,
+    blockID flow.Identifier,
+    eventType string,
+) (flow.BlockEvents, error) {
+    evts, err := r.client.GetEventsForBlockIDs(
+        ctx,
+        eventType,
+        []flow.Identifier{blockID},
+    )
+    if err != nil {
+        return flow.BlockEvents{}, err
+    }
+
+    if len(evts) != 1 {
+        return flow.BlockEvents{}, fmt.Errorf(
+            "received unexpected number of block events: got: %d, expected: 1",
+            len(evts),
+        )
+    }
+
+    if len(evts[0].Events) != 1 {
+        return flow.BlockEvents{}, fmt.Errorf(
+            "received unexpected number of events in block: got: %d, expected: 1",
+            len(evts[0].Events),
+        )
+    }
+
+    return evts[0], nil
+}

services/ingestion/block_tracking_subscriber.go (1)

102-209: Consider breaking down the subscribe method.

The method is quite long and handles multiple responsibilities. Consider extracting the event processing and error handling into separate methods for better maintainability.

Here's a suggested refactoring approach:

+func (r *RPCBlockTrackingSubscriber) handleBlockHeader(
+    ctx context.Context,
+    blockHeader *flow.BlockHeader,
+    eventsChan chan<- models.BlockEvents,
+) (uint64, bool) {
+    blockEvents, err := r.evmEventsForHeight(ctx, blockHeader.Height)
+    if err != nil {
+        eventsChan <- models.NewBlockEventsError(err)
+        return 0, false
+    }
+
+    evmEvents := models.NewSingleBlockEvents(blockEvents)
+    if evmEvents.Err != nil || r.recovery {
+        evmEvents = r.recover(ctx, blockEvents, evmEvents.Err)
+        if r.recovery {
+            return blockHeader.Height, true
+        }
+    }
+
+    for _, evt := range blockEvents.Events {
+        r.keyLock.UnlockKey(evt.TransactionID)
+    }
+    r.keyLock.Notify(blockHeader.Height)
+
+    eventsChan <- evmEvents
+    return blockHeader.Height, true
+}
+
+func (r *RPCBlockTrackingSubscriber) handleSubscriptionError(
+    ctx context.Context,
+    err error,
+    lastHeight uint64,
+) (chan *flow.BlockHeader, chan error, error) {
+    if status.Code(err) == codes.DeadlineExceeded || status.Code(err) == codes.Internal {
+        return r.client.SubscribeBlockHeadersFromStartHeight(
+            ctx,
+            lastHeight+1,
+            flow.BlockStatusFinalized,
+        )
+    }
+    return nil, nil, fmt.Errorf("%w: %w", errs.ErrDisconnected, err)
+}

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

services/requester/cross-spork_client.go (1)

265-280: Consider adding error recovery for gRPC connection issues.

The method correctly handles client retrieval and type assertion. However, consider adding retry logic for transient gRPC connection issues.

 func (c *CrossSporkClient) SubscribeBlockHeadersFromStartHeight(
     ctx context.Context,
     startHeight uint64,
     blockStatus flow.BlockStatus,
 ) (<-chan flow.BlockHeader, <-chan error, error) {
     client, err := c.getClientForHeight(startHeight)
     if err != nil {
         return nil, nil, err
     }
     grpcClient, ok := (client).(*grpc.Client)
     if !ok {
         return nil, nil, fmt.Errorf("unable to convert to Flow grpc.Client")
     }
+
+    // Add retry with backoff for transient gRPC connection issues
+    var attempt int
+    for attempt < 3 {
+        headerChan, errChan, err := grpcClient.SubscribeBlockHeadersFromStartHeight(ctx, startHeight, blockStatus)
+        if err == nil {
+            return headerChan, errChan, nil
+        }
+        if !isTransientError(err) {
+            return nil, nil, err
+        }
+        attempt++
+        time.Sleep(time.Duration(attempt) * time.Second)
+    }

     return grpcClient.SubscribeBlockHeadersFromStartHeight(ctx, startHeight, blockStatus)
 }

services/ingestion/block_tracking_subscriber.go (2)

62-108: Consider adding metrics for backfilling progress.

While the implementation correctly handles backfilling events from previous sporks, adding metrics would help monitor the progress.

 func (r *RPCBlockTrackingSubscriber) Subscribe(ctx context.Context) <-chan models.BlockEvents {
     eventsChan := make(chan models.BlockEvents, 1000)
+    
+    // Add metrics for monitoring backfill progress
+    var backfilledEvents uint64
+    startTime := time.Now()

     go func() {
         defer func() {
             close(eventsChan)
+            if backfilledEvents > 0 {
+                r.logger.Info().
+                    Uint64("total_events", backfilledEvents).
+                    Dur("duration", time.Since(startTime)).
+                    Msg("backfill completed")
+            }
         }()

         if r.client.IsPastSpork(r.height) {
             for ev := range r.backfill(ctx, r.height) {
                 eventsChan <- ev
+                backfilledEvents++
+                
+                // Log progress every 1000 events
+                if backfilledEvents%1000 == 0 {
+                    r.logger.Info().
+                        Uint64("events_processed", backfilledEvents).
+                        Msg("backfill in progress")
+                }

                 if ev.Err != nil {
                     return
                 }
                 r.height = ev.Events.CadenceHeight()
             }
             r.height = r.height + 1
         }

---

110-217: Consider adding circuit breaker for repeated failures.

The error handling and recovery logic is solid, but consider adding a circuit breaker to prevent infinite retries on persistent failures.

 func (r *RPCBlockTrackingSubscriber) subscribe(ctx context.Context, height uint64) <-chan models.BlockEvents {
     eventsChan := make(chan models.BlockEvents)
+    
+    // Add circuit breaker to prevent infinite retries
+    var consecutiveFailures int
+    const maxConsecutiveFailures = 5
+    const resetDuration = 5 * time.Minute
+    lastFailureTime := time.Now()

     blockHeadersChan, errChan, err := r.client.SubscribeBlockHeadersFromStartHeight(
         ctx,
         height,
         flow.BlockStatusFinalized,
     )
     if err != nil {
+        consecutiveFailures++
         eventsChan <- models.NewBlockEventsError(
             fmt.Errorf(
                 "failed to subscribe for finalized block headers on height: %d, with: %w",
                 height,
                 err,
             ),
         )
         close(eventsChan)
         return eventsChan
     }

     go func() {
         defer func() {
             close(eventsChan)
         }()

         for ctx.Err() == nil {
             select {
             case err, ok := <-errChan:
                 if !ok {
                     // ... existing error handling ...
                 }

                 if status.Code(err) == codes.DeadlineExceeded || status.Code(err) == codes.Internal {
+                    consecutiveFailures++
+                    if consecutiveFailures >= maxConsecutiveFailures {
+                        if time.Since(lastFailureTime) < resetDuration {
+                            eventsChan <- models.NewBlockEventsError(
+                                fmt.Errorf("circuit breaker triggered: too many consecutive failures"),
+                            )
+                            return
+                        }
+                        consecutiveFailures = 0
+                        lastFailureTime = time.Now()
+                    }
                     
                     blockHeadersChan, errChan, err = r.client.SubscribeBlockHeadersFromStartHeight(
                         ctx,
                         lastReceivedHeight+1,
                         flow.BlockStatusFinalized,
                     )
                     if err != nil {
                         eventsChan <- models.NewBlockEventsError(
                             fmt.Errorf(
                                 "failed to subscribe for finalized block headers on height: %d, with: %w",
                                 height,
                                 err,
                             ),
                         )
                         return
                     }
+                    consecutiveFailures = 0
                 } else {
                     eventsChan <- models.NewBlockEventsError(fmt.Errorf("%w: %w", errs.ErrDisconnected, err))
                     return
                 }
             }
         }
     }()

     return eventsChan
 }

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

config/config.go (1)

92-95: Consider enhancing the documentation with specific scenarios.

The warning about incorrect data is good, but it would be more helpful to document specific scenarios where the data might be incorrect. This helps users make informed decisions about using this feature.

Consider expanding the comment to:

-	// ExperimentalSoftFinalityEnabled enables the experimental soft finality feature which syncs
-	// EVM block and transaction data from the upstream Access node before the block is sealed.
-	// CAUTION: This feature is experimental and may return incorrect data in certain circumstances.
+	// ExperimentalSoftFinalityEnabled enables the experimental soft finality feature which syncs
+	// EVM block and transaction data from the upstream Access node before the block is sealed.
+	// CAUTION: This feature is experimental and may return incorrect data in the following scenarios:
+	// - When the block is reverted before being sealed
+	// - When there are concurrent updates to the same state
+	// - When the node experiences network partitioning

cmd/run/cmd.go (1)

287-287: Consider enhancing the warning message with specific scenarios.

While the warning message is good, it would be more helpful to list specific scenarios where incorrect results might occur. This aligns with the enhanced documentation suggested for the config file.

-	Cmd.Flags().BoolVar(&experimentalSoftFinalityEnabled, "experimental-soft-finality-enabled", false, "Sets whether the gateway should use the experimental soft finality feature. WARNING: This may result in incorrect results being returned in certain circumstances. Use only if you know what you are doing.")
+	Cmd.Flags().BoolVar(&experimentalSoftFinalityEnabled, "experimental-soft-finality-enabled", false, "Sets whether the gateway should use the experimental soft finality feature. WARNING: This may return incorrect results when blocks are reverted before sealing, during concurrent state updates, or network partitioning. Use only if you understand these risks.")

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

services/ingestion/block_tracking_subscriber.go (5)

23-38: Fix typo in documentation.

There's a typo in the documentation: "this features should NOT" should be "this feature should NOT".

-// Since it's not reasonable to expect operators to do this manual tracking, this features should NOT
+// Since it's not reasonable to expect operators to do this manual tracking, this feature should NOT

---

40-56: Add input validation to constructor.

Consider adding validation for nil parameters to prevent potential panics.

 func NewRPCBlockTrackingSubscriber(
     logger zerolog.Logger,
     client *requester.CrossSporkClient,
     chainID flowGo.ChainID,
     keyLock requester.KeyLock,
     startHeight uint64,
 ) *RPCBlockTrackingSubscriber {
+    if client == nil {
+        panic("client cannot be nil")
+    }
+    if keyLock == nil {
+        panic("keyLock cannot be nil")
+    }
     return &RPCBlockTrackingSubscriber{
         RPCEventSubscriber: NewRPCEventSubscriber(
             logger.With().Str("component", "subscriber").Logger(),

---

111-235: Consider extracting error handling logic.

The error handling logic in the select statement is complex and could be more maintainable if extracted into separate methods.

Consider refactoring like this:

+func (r *RPCBlockTrackingSubscriber) handleBlockHeader(
+    ctx context.Context,
+    blockHeader flow.BlockHeader,
+    eventsChan chan<- models.BlockEvents,
+) (uint64, bool) {
+    blockEvents, err := r.evmEventsForBlock(ctx, blockHeader)
+    if err != nil {
+        eventsChan <- models.NewBlockEventsError(err)
+        return 0, false
+    }
+    // ... rest of block header handling logic
+    return blockHeader.Height, true
+}

+func (r *RPCBlockTrackingSubscriber) handleError(
+    ctx context.Context,
+    err error,
+    eventsChan chan<- models.BlockEvents,
+    lastReceivedHeight uint64,
+) bool {
+    // ... error handling logic
+}

 func (r *RPCBlockTrackingSubscriber) subscribe(ctx context.Context, height uint64) <-chan models.BlockEvents {
     // ... existing code ...
     select {
     case blockHeader, ok := <-blockHeadersChan:
         if !ok {
             // ... channel closed handling ...
         }
-        blockEvents, err := r.evmEventsForBlock(ctx, blockHeader)
-        if err != nil {
-            eventsChan <- models.NewBlockEventsError(err)
-            return
-        }
-        // ... rest of block header handling
+        if height, ok := r.handleBlockHeader(ctx, blockHeader, eventsChan); ok {
+            lastReceivedHeight = height
+        } else {
+            return
+        }
     case err, ok := <-errChan:
-        // ... error handling logic
+        if !r.handleError(ctx, err, eventsChan, lastReceivedHeight) {
+            return
+        }
     }

---

237-268: Enhance error messages for better debugging.

The error messages could be more descriptive to aid in debugging issues.

 func (r *RPCBlockTrackingSubscriber) evmEventsForBlock(
     ctx context.Context,
     blockHeader flow.BlockHeader,
 ) (flow.BlockEvents, error) {
     eventTypes := blocksFilter(r.chain).EventTypes
 
     // evm Block events
     blockEvents, err := r.getEventsByType(ctx, blockHeader, eventTypes[0])
     if err != nil {
-        return flow.BlockEvents{}, err
+        return flow.BlockEvents{}, fmt.Errorf("failed to get block events at height %d: %w", blockHeader.Height, err)
     }
 
     payload, err := events.DecodeBlockEventPayload(blockEvents.Events[0].Value)
     if err != nil {
-        return flow.BlockEvents{}, err
+        return flow.BlockEvents{}, fmt.Errorf("failed to decode block event payload at height %d: %w", blockHeader.Height, err)
     }

---

270-303: Extract magic strings into constants.

The event type string comparison uses a magic string that should be extracted into a constant.

+const (
+    evmBlockExecutedEvent = "EVM.BlockExecuted"
+)

 func (r *RPCBlockTrackingSubscriber) getEventsByType(
     ctx context.Context,
     blockHeader flow.BlockHeader,
     eventType string,
 ) (flow.BlockEvents, error) {
     // ...
-    if strings.Contains(eventType, string(events.EventTypeBlockExecuted)) {
+    if strings.Contains(eventType, evmBlockExecutedEvent) {

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro