From fe7eabfe06720365900e8d3fb47070f0f5fa24f2 Mon Sep 17 00:00:00 2001
From: Ardit Marku <markoupetr@gmail.com>
Date: Wed, 30 Jul 2025 11:18:54 +0300
Subject: [PATCH 1/2] Add reconnect logic to RPCEventSubscriber

---
 services/ingestion/event_subscriber.go | 63 +++++++++++++++++++++-----
 1 file changed, 52 insertions(+), 11 deletions(-)

diff --git a/services/ingestion/event_subscriber.go b/services/ingestion/event_subscriber.go
index 8e182a927..1ed0b7d0c 100644
--- a/services/ingestion/event_subscriber.go
+++ b/services/ingestion/event_subscriber.go
@@ -5,9 +5,12 @@ import (
 	"errors"
 	"fmt"
 	"sort"
+	"time"
 
 	"github.com/onflow/cadence/common"
 	"github.com/onflow/flow-go/fvm/evm/events"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 
 	"github.com/onflow/flow-evm-gateway/models"
 	errs "github.com/onflow/flow-evm-gateway/models/errors"
@@ -128,17 +131,30 @@ func (r *RPCEventSubscriber) subscribe(ctx context.Context, height uint64) <-cha
 		return eventsChan
 	}
 
-	// we always use heartbeat interval of 1 to have the least amount of delay from the access node
-	eventStream, errChan, err := r.client.SubscribeEventsByBlockHeight(
-		ctx,
-		height,
-		blocksFilter(r.chain),
-		access.WithHeartbeatInterval(1),
-	)
-	if err != nil {
+	var blockEventsStream <-chan flow.BlockEvents
+	var errChan <-chan error
+
+	lastReceivedHeight := height
+	connect := func(height uint64) error {
+		var err error
+
+		// we always use heartbeat interval of 1 to have the
+		// least amount of delay from the access node
+		blockEventsStream, errChan, err = r.client.SubscribeEventsByBlockHeight(
+			ctx,
+			height,
+			blocksFilter(r.chain),
+			access.WithHeartbeatInterval(1),
+		)
+
+		return err
+	}
+
+	if err := connect(lastReceivedHeight); err != nil {
 		eventsChan <- models.NewBlockEventsError(
 			fmt.Errorf("failed to subscribe to events by block height: %d, with: %w", height, err),
 		)
+		close(eventsChan)
 		return eventsChan
 	}
 
@@ -153,8 +169,9 @@ func (r *RPCEventSubscriber) subscribe(ctx context.Context, height uint64) <-cha
 				r.logger.Info().Msg("event ingestion received done signal")
 				return
 
-			case blockEvents, ok := <-eventStream:
+			case blockEvents, ok := <-blockEventsStream:
 				if !ok {
+					// typically we receive an error in the errChan before the channels are closed
 					var err error
 					err = errs.ErrDisconnected
 					if ctx.Err() != nil {
@@ -183,10 +200,13 @@ func (r *RPCEventSubscriber) subscribe(ctx context.Context, height uint64) <-cha
 					},
 				)
 
+				lastReceivedHeight = blockEvents.Height
+
 				eventsChan <- evmEvents
 
 			case err, ok := <-errChan:
 				if !ok {
+					// typically we receive an error in the errChan before the channels are closed
 					var err error
 					err = errs.ErrDisconnected
 					if ctx.Err() != nil {
@@ -196,8 +216,29 @@ func (r *RPCEventSubscriber) subscribe(ctx context.Context, height uint64) <-cha
 					return
 				}
 
-				eventsChan <- models.NewBlockEventsError(fmt.Errorf("%w: %w", errs.ErrDisconnected, err))
-				return
+				switch status.Code(err) {
+				case codes.NotFound:
+					// we can get not found when reconnecting after a disconnect/restart before the
+					// next block is finalized. just wait briefly and try again
+					time.Sleep(200 * time.Millisecond)
+				case codes.DeadlineExceeded, codes.Internal:
+					// these are sometimes returned when the stream is disconnected by a middleware or the server
+				default:
+					// skip reconnect on all other errors
+					eventsChan <- models.NewBlockEventsError(fmt.Errorf("%w: %w", errs.ErrDisconnected, err))
+					return
+				}
+
+				if err := connect(lastReceivedHeight + 1); err != nil {
+					eventsChan <- models.NewBlockEventsError(
+						fmt.Errorf(
+							"failed to resubscribe for events on height: %d, with: %w",
+							lastReceivedHeight+1,
+							err,
+						),
+					)
+					return
+				}
 			}
 		}
 	}()

From 29b5098ba9c2191ea6e4001c94fc43822e4d9a03 Mon Sep 17 00:00:00 2001
From: Ardit Marku <markoupetr@gmail.com>
Date: Wed, 6 Aug 2025 15:25:57 +0300
Subject: [PATCH 2/2] Fix error handling on subscribe method

---
 services/ingestion/event_subscriber.go | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/services/ingestion/event_subscriber.go b/services/ingestion/event_subscriber.go
index 1ed0b7d0c..90f206031 100644
--- a/services/ingestion/event_subscriber.go
+++ b/services/ingestion/event_subscriber.go
@@ -122,12 +122,15 @@ func (r *RPCEventSubscriber) Subscribe(ctx context.Context) <-chan models.BlockE
 // Subscribing to EVM specific events and handle any disconnection errors
 // as well as context cancellations.
 func (r *RPCEventSubscriber) subscribe(ctx context.Context, height uint64) <-chan models.BlockEvents {
-	eventsChan := make(chan models.BlockEvents)
+	// create the channel with a buffer size of 1,
+	// to avoid blocking on the two error cases below
+	eventsChan := make(chan models.BlockEvents, 1)
 
 	_, err := r.client.GetBlockHeaderByHeight(ctx, height)
 	if err != nil {
 		err = fmt.Errorf("failed to subscribe for events, the block height %d doesn't exist: %w", height, err)
 		eventsChan <- models.NewBlockEventsError(err)
+		close(eventsChan)
 		return eventsChan
 	}