New mode of head-polling using client-side state management only

config.md

@@ -111,8 +111,9 @@
 |catchupDownscaleRegex|An error pattern to check for from JSON/RPC providers if they limit response sizes to eth_getLogs(). If an error is returned from eth_getLogs() and that error matches the configured pattern, the number of logs requested (catchupPageSize) will be reduced automatically.|string|`Response size is larger than.*limit`
 |catchupPageSize|Number of blocks to query per poll when catching up to the head of the blockchain|`int`|`500`
 |catchupThreshold|How many blocks behind the chain head an event stream or listener must be on startup, to enter catchup mode|`int`|`500`
-|checkpointBlockGap|The number of blocks at the head of the chain that should be considered unstable (could be dropped from the canonical chain after a re-org). Unless events with a full set of confirmations are detected, the restart checkpoint will this many blocks behind the chain head.|`int`|`50`
+|checkpointBlockGap|The number of blocks at the head of the chain that should be considered unstable (could be dropped from the canonical chain after a re-org). Unless events with a full set of confirmations are detected, the restart checkpoint will be at least this many blocks behind the chain head. Note the target confirmation count can be smaller than this (must never be larger), and the target confirmation count is the only guaranteed re-delivery window at the head of a chain that can have re-orgs (non-immediate finality)|`int`|`50`
 |filterPollingInterval|The interval between polling calls to a filter, when checking for newly arrived events|[`time.Duration`](https://pkg.go.dev/time#Duration)|`1s`
+|filterPollingMode|How events are polled for in the steady state event loop, once caught up with the head of the chain. 'server' establishes a node-side filter with eth_newFilter and polls it with eth_getFilterChanges. 'client' uses stateless eth_getLogs range queries with the connector tracking its own poll position, avoiding node-side filter state entirely. In 'client' mode with chainTrackingMode 'light', the checkpointBlockGap window of blocks behind the head is re-scanned on each polling interval to detect events moved by a re-org, with block-hash de-duplication of events already delivered (set events.blockTimestamps false to avoid per-block fetches)|server,client|`server`
 
 ## connector.net
 

internal/ethereum/config.go

@@ -37,6 +37,7 @@ const (
 	EventsCheckpointBlockGap    = "events.checkpointBlockGap"
 	EventsBlockTimestamps       = "events.blockTimestamps"
 	EventsFilterPollingInterval = "events.filterPollingInterval"
+	EventsFilterPollingMode     = "events.filterPollingMode"
 	RetryInitDelay              = "queryLoopRetry.initialDelay"
 	RetryMaxDelay               = "queryLoopRetry.maxDelay"
 	RetryFactor                 = "queryLoopRetry.factor"
@@ -57,6 +58,19 @@ const (
 	UseGetBlockReceipts           = "useGetBlockReceipts"
 )
 
+// filterPollingMode determines how the steady state loop of an event stream polls for new events,
+// once it has caught up with the head of the chain.
+type filterPollingMode string
+
+const (
+	// FilterPollingModeServer uses a node-side filter, established with eth_newFilter and polled
+	// with eth_getFilterChanges, so the node tracks which logs are new since the last poll
+	FilterPollingModeServer filterPollingMode = "server"
+	// FilterPollingModeClient uses stateless eth_getLogs range queries, with the connector tracking
+	// its own in-memory poll position - avoiding node-side filter state entirely
+	FilterPollingModeClient filterPollingMode = "client"
+)
+
 const (
 	DefaultListenerPort        = 5102
 	DefaultGasEstimationFactor = 1.5
@@ -84,6 +98,7 @@ func InitConfig(conf config.Section) {
 	conf.AddKnownKey(ConfigGasEstimationFactor, DefaultGasEstimationFactor)
 	conf.AddKnownKey(EventsBlockTimestamps, true)
 	conf.AddKnownKey(EventsFilterPollingInterval, "1s")
+	conf.AddKnownKey(EventsFilterPollingMode, string(FilterPollingModeServer))
 	conf.AddKnownKey(EventsCatchupPageSize, DefaultCatchupPageSize)
 	conf.AddKnownKey(EventsCatchupThreshold, DefaultEventsCatchupThreshold)
 	conf.AddKnownKey(EventsCatchupDownscaleRegex, DefaultEventsCatchupDownscaleRegex)

internal/ethereum/ethereum.go

@@ -55,6 +55,7 @@ type ethConnector struct {
 	eventBlockTimestamps       bool
 	blockListener              ethblocklistener.BlockListener
 	eventFilterPollingInterval time.Duration
+	eventFilterPollingMode     filterPollingMode
 	traceTXForRevertReason     bool
 	chainID                    string
 
@@ -117,6 +118,14 @@ func NewEthereumConnectorWithRPC(ctx context.Context, conf config.Section, rpc e
 		return nil, i18n.NewError(ctx, msgs.MsgInvalidChainTrackingMode, chainTrackingMode)
 	}
 
+	eventFilterPollingMode := filterPollingMode(conf.GetString(EventsFilterPollingMode))
+	if eventFilterPollingMode == "" {
+		eventFilterPollingMode = FilterPollingModeServer
+	}
+	if eventFilterPollingMode != FilterPollingModeServer && eventFilterPollingMode != FilterPollingModeClient {
+		return nil, i18n.NewError(ctx, msgs.MsgInvalidFilterPollingMode, eventFilterPollingMode)
+	}
+
 	c := &ethConnector{
 		rpc:                        rpc,
 		eventStreams:               make(map[fftypes.UUID]*eventStream),
@@ -125,6 +134,7 @@ func NewEthereumConnectorWithRPC(ctx context.Context, conf config.Section, rpc e
 		checkpointBlockGap:         conf.GetInt64(EventsCheckpointBlockGap),
 		eventBlockTimestamps:       conf.GetBool(EventsBlockTimestamps),
 		eventFilterPollingInterval: conf.GetDuration(EventsFilterPollingInterval),
+		eventFilterPollingMode:     eventFilterPollingMode,
 		traceTXForRevertReason:     conf.GetBool(TraceTXForRevertReason),
 		chainTrackingMode:          chainTrackingMode,
 		retry:                      retryutil.RetryWrapper{Retry: &retry.Retry{}},

internal/ethereum/ethereum_test.go

@@ -105,6 +105,11 @@ func TestConnectorInit(t *testing.T) {
 
 	conf.Set(RPCRoutingMode, ethrpc.RoutingModeAuto)
 	conf.Set(ChainTrackingMode, "")
+	conf.Set(EventsFilterPollingMode, "wrong")
+	_, err = NewEthereumConnector(context.Background(), conf)
+	assert.Regexp(t, "FF23078.*wrong", err)
+
+	conf.Set(EventsFilterPollingMode, string(FilterPollingModeClient))
 	conf.Set(WebSocketsEnabled, true)
 	conf.Set(EventsCatchupThreshold, 1)
 	conf.Set(EventsCatchupPageSize, 500)

internal/ethereum/event_listener.go

@@ -114,7 +114,7 @@ func (l *listener) ensureHWM(ctx context.Context) error {
 			return err
 		}
 		// HWM is the configured fromBlock
-		l.hwmBlock = int64(firstBlock) //nolint:gosec // convert to int64 to match the type of hwmBlock, we should change the type of hwmBlock to uint64
+		l.hwmBlock = blockNumberToInt64(firstBlock)
 	}
 	return nil
 }

internal/ethereum/event_stream.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"math"
 	"sort"
 	"strings"
 	"sync"
@@ -272,15 +273,26 @@ func (es *eventStream) leadGroupCatchup() bool {
 			}
 		}
 
-		// Check if we're ready to exit catchup mode
-		headGap := (int64(chainHeadBlock) - fromBlock) //nolint:gosec // convert to int64 to match the type of headGap
+		// Catchup only polls blocks that are outside the re-org unstable window at the head of
+		// the chain (checkpointBlockGap behind the head).
+		// The steady-state loops own delivery of the unstable window.
+		// We stop on the first page where the end lands between catchupThreshold+checkpointBlockGap
+		// (say 550) and the checkpointBlockGap (say 50) before the head to do the switch.
+		pollableHead := blockNumberToInt64(chainHeadBlock) - es.c.checkpointBlockGap
+		if pollableHead < 0 {
+			pollableHead = 0
+		}
+		headGap := pollableHead - fromBlock
 		if headGap < es.c.catchupThreshold {
 			log.L(es.ctx).Infof("Stream head is up to date with chain fromBlock=%d chainHead=%d headGap=%d", fromBlock, chainHeadBlock, headGap)
 			return false
 		}
 
 		// Poll in the range for events
 		toBlock := fromBlock + es.c.catchupPageSize - 1
+		if toBlock > pollableHead {
+			toBlock = pollableHead
+		}
 		events, err := es.getBlockRangeEvents(es.ctx, ag, fromBlock, toBlock)
 		if err != nil {
 			log.L(es.ctx).Errorf("Failed to query block range fromBlock=%d toBlock=%d headBlock=%d: %s", fromBlock, toBlock, chainHeadBlock, err)
@@ -289,8 +301,12 @@ func (es *eventStream) leadGroupCatchup() bool {
 		}
 		log.L(es.ctx).Infof("Stream catchup fromBlock=%d toBlock=%d headBlock=%d events=%d listeners=%d", fromBlock, toBlock, chainHeadBlock, len(events), len(ag.listeners))
 
+		// The poll position never enters the unstable window, so the HWM for the restart
+		// checkpoint is simply the next block to poll
+		hwmBlock := toBlock + 1
+
 		// Dispatch the events
-		if es.dispatchSetHWMCheckExit(ag, events, toBlock+1 /* hwm is the next block after our poll */) {
+		if es.dispatchSetHWMCheckExit(ag, events, hwmBlock) {
 			log.L(es.ctx).Debugf("Stream catchup loop exiting")
 			return true
 		}
@@ -338,7 +354,7 @@ func (es *eventStream) leadGroupSteadyState() bool {
 			// High water mark is a point safely behind the head of the chain in this case,
 			// where re-orgs are not expected.
 			bh, _ := es.c.blockListener.GetHighestBlock(es.ctx) /* note we know we're initialized here and will not block */
-			hwmBlock := int64(bh) - es.c.checkpointBlockGap     //nolint:gosec // convert to int64 to match the type of hwmBlock
+			hwmBlock := blockNumberToInt64(bh) - es.c.checkpointBlockGap
 			if hwmBlock < 0 {
 				hwmBlock = 0
 			}
@@ -359,9 +375,12 @@ func (es *eventStream) leadGroupSteadyState() bool {
 					}
 				}
 
-				// Check we're not outside of the steady state window, and need to fall back to catchup mode
+				// Check we're not outside of the steady state window, and need to fall back to
+				// catchup mode. Catchup only polls up to the stability horizon (checkpointBlockGap
+				// behind the head), so we measure against the same point - the two loops can never
+				// disagree and bounce control between each other.
 				chainHeadBlock, _ := es.c.blockListener.GetHighestBlock(es.ctx) /* note we know we're initialized here and will not block */
-				blockGapEstimate := (int64(chainHeadBlock) - fromBlock)         //nolint:gosec // convert to int64 to match the type of blockGapEstimate
+				blockGapEstimate := (blockNumberToInt64(chainHeadBlock) - es.c.checkpointBlockGap - fromBlock)
 				if blockGapEstimate > es.c.catchupThreshold {
 					log.L(es.ctx).Warnf("Block gap estimate reached %d (above threshold of %d) - reverting to catchup mode", blockGapEstimate, es.c.catchupThreshold)
 					return false
@@ -429,6 +448,17 @@ func (es *eventStream) leadGroupSteadyState() bool {
 	}
 }
 
+// blockNumberToInt64 converts a block number from the node into the int64 type we use for all
+// block range arithmetic, with a bounds check to avoid wraparound. A block number large enough
+// to overflow an int64 cannot occur on a real chain and cannot be handled, so a panic is
+// acceptable in that case.
+func blockNumberToInt64(blockNumber uint64) int64 {
+	if blockNumber > math.MaxInt64 {
+		panic(fmt.Sprintf("block number %d too large", blockNumber))
+	}
+	return int64(blockNumber)
+}
+
 func (es *eventStream) preStartProcessing() {
 	ctx := es.ctx
 	chainHead, ok := es.c.blockListener.GetHighestBlock(ctx)
@@ -439,7 +469,7 @@ func (es *eventStream) preStartProcessing() {
 	// The lead group never advances past checkpointBlockGap behind the chain head, as those blocks
 	// are re-org unstable. We establish our head position on the same basis, so that a listener
 	// held in catchup clamps against a safe ceiling from the moment it is established.
-	safeHead := int64(chainHead) - es.c.checkpointBlockGap //nolint:gosec // convert to int64 to match the type of headBlock
+	safeHead := blockNumberToInt64(chainHead) - es.c.checkpointBlockGap
 	if safeHead < 0 {
 		safeHead = 0
 	}
@@ -489,7 +519,13 @@ func (es *eventStream) streamLoop() {
 
 		// We then transition to our steady state, filtering from the front of the chain.
 		// But we might fall behind and need to go back to the catchup mode.
-		if es.leadGroupSteadyState() {
+		var exiting bool
+		if es.c.eventFilterPollingMode == FilterPollingModeClient {
+			exiting = es.leadGroupSteadyStateGetLogs()
+		} else {
+			exiting = es.leadGroupSteadyState()
+		}
+		if exiting {
 			return
 		}
 	}
@@ -582,7 +618,7 @@ func (es *eventStream) filterEnrichSort(ctx context.Context, ag *aggregatedListe
 	return updates, nil
 }
 
-func (es *eventStream) getBlockRangeEvents(ctx context.Context, ag *aggregatedListener, fromBlock, toBlock int64) (ffcapi.ListenerEvents, error) {
+func (es *eventStream) getBlockRangeLogs(ctx context.Context, ag *aggregatedListener, fromBlock, toBlock int64) ([]*ethrpc.LogJSONRPC, error) {
 	var ethLogs []*ethrpc.LogJSONRPC
 	logFilterJSONRPCReq := &ethrpc.LogFilterJSONRPC{
 		FromBlock: ethtypes.NewHexInteger64(fromBlock),
@@ -600,6 +636,14 @@ func (es *eventStream) getBlockRangeEvents(ctx context.Context, ag *aggregatedLi
 	if rpcErr != nil {
 		return nil, rpcErr.Error()
 	}
+	return ethLogs, nil
+}
+
+func (es *eventStream) getBlockRangeEvents(ctx context.Context, ag *aggregatedListener, fromBlock, toBlock int64) (ffcapi.ListenerEvents, error) {
+	ethLogs, err := es.getBlockRangeLogs(ctx, ag, fromBlock, toBlock)
+	if err != nil {
+		return nil, err
+	}
 	return es.filterEnrichSort(ctx, ag, ethLogs)
 }