diff --git a/config.md b/config.md index 0611223..4f755b3 100644 --- a/config.md +++ b/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 diff --git a/internal/ethereum/config.go b/internal/ethereum/config.go index 9be0239..3c0a7db 100644 --- a/internal/ethereum/config.go +++ b/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) diff --git a/internal/ethereum/ethereum.go b/internal/ethereum/ethereum.go index f1be593..4e7bdfe 100644 --- a/internal/ethereum/ethereum.go +++ b/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 := ðConnector{ 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{}}, diff --git a/internal/ethereum/ethereum_test.go b/internal/ethereum/ethereum_test.go index 105b79d..eb430b6 100644 --- a/internal/ethereum/ethereum_test.go +++ b/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) diff --git a/internal/ethereum/event_listener.go b/internal/ethereum/event_listener.go index a80da35..35715b8 100644 --- a/internal/ethereum/event_listener.go +++ b/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 } diff --git a/internal/ethereum/event_stream.go b/internal/ethereum/event_stream.go index 96fb007..43abe6f 100644 --- a/internal/ethereum/event_stream.go +++ b/internal/ethereum/event_stream.go @@ -20,6 +20,7 @@ import ( "context" "encoding/json" "fmt" + "math" "sort" "strings" "sync" @@ -272,8 +273,16 @@ 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 @@ -281,6 +290,9 @@ func (es *eventStream) leadGroupCatchup() bool { // 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 := ðrpc.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) } diff --git a/internal/ethereum/event_stream_getlogs.go b/internal/ethereum/event_stream_getlogs.go new file mode 100644 index 0000000..7df6996 --- /dev/null +++ b/internal/ethereum/event_stream_getlogs.go @@ -0,0 +1,374 @@ +// Copyright © 2026 Kaleido, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package ethereum + +import ( + "bytes" + "context" + "time" + + "github.com/hyperledger-firefly/common/pkg/log" + "github.com/hyperledger-firefly/evmconnect/pkg/ethrpc" + "github.com/hyperledger-firefly/signer/pkg/ethtypes" + "github.com/hyperledger-firefly/transaction-manager/pkg/ffcapi" +) + +// State required when doing all management of polling position client-side +type getLogsPollState struct { + fromBlock int64 // full mode: the next block to poll. light mode: the committed base of the re-scan window + scanBlock int64 // light mode only: sweep cursor when paging a window wider than one page (-1 = start the next sweep at fromBlock) + polledChain []*ethrpc.BlockInfoJSONRPC // full mode only: sparse ascending (number, hash) records of polled blocks in the unstable window + deliveredBlocks map[string]int64 // light mode only: blockHash->number for block-versions whose logs we have already delivered +} + +// reset (re-)establishes the poll position, discarding any recorded hash continuity or +// delivered-block records (redelivery of the unstable window is de-duplicated downstream) +func (ps *getLogsPollState) reset(fromBlock int64) { + ps.fromBlock = fromBlock + ps.scanBlock = -1 + ps.polledChain = nil + ps.deliveredBlocks = nil +} + +// filterDelivered strips logs from block-versions we already delivered on a previous sweep of the +// re-scan window (light mode). The block hash commits to the entire content of a block, so logs +// returned for a block hash we have seen are guaranteed identical to the ones we already +// processed - and a re-org replacing a block gives its logs a new block hash, so they pass +// through as new detections. +func (ps *getLogsPollState) filterDelivered(logs []*ethrpc.LogJSONRPC) []*ethrpc.LogJSONRPC { + if len(ps.deliveredBlocks) == 0 { + return logs + } + newLogs := make([]*ethrpc.LogJSONRPC, 0, len(logs)) + for _, l := range logs { + if _, delivered := ps.deliveredBlocks[string(l.BlockHash)]; !delivered { + newLogs = append(newLogs, l) + } + } + return newLogs +} + +// advanceRescan moves the light mode poll state forwards after successfully dispatching a page: +// the newly delivered block-versions are recorded for de-duplication on later sweeps, the +// committed window base moves to newBase (capped at the stability horizon), records for blocks +// that have become stable are dropped (they are never re-scanned), and the sweep cursor either +// pages onwards or resets ready to re-scan the whole window on the next cycle. +func (ps *getLogsPollState) advanceRescan(newLogs []*ethrpc.LogJSONRPC, newBase, toBlock, chainHead int64) { + for _, l := range newLogs { + if ps.deliveredBlocks == nil { + ps.deliveredBlocks = map[string]int64{} + } + ps.deliveredBlocks[string(l.BlockHash)] = trimUint64(l.BlockNumber.Uint64()) + } + ps.fromBlock = newBase + for h, n := range ps.deliveredBlocks { + if n < newBase { + delete(ps.deliveredBlocks, h) + } + } + if toBlock >= chainHead { + ps.scanBlock = -1 // sweep complete - re-scan the whole window from fromBlock next cycle + } else { + ps.scanBlock = toBlock + 1 // page onwards through this sweep without waiting + } +} + +// checkReorgRewind is used in full chain tracking mode only (light mode re-scans the unstable +// window and de-duplicates instead - see leadGroupSteadyStateGetLogs). +// It compares the canonical chain view recorded at the time we scanned each range, against the +// block listener's current canonical chain view. On a mismatch the view of the chain has changed +// behind our poll position (a re-org, or a stale view at scan time), so we rewind to the earliest +// diverging block to re-scan from there. +func (ps *getLogsPollState) checkReorgRewind(ctx context.Context, headChain []*ethrpc.BlockInfoJSONRPC) { + if len(headChain) == 0 || len(ps.polledChain) == 0 { + return + } + // Prune records that have aged out below the base of the monitored window - those blocks are + // now considered stable, and we have nothing to compare them against + baseBlock := blockNumberToInt64(headChain[0].Number.Uint64()) + firstInWindow := 0 + for firstInWindow < len(ps.polledChain) && blockNumberToInt64(ps.polledChain[firstInWindow].Number.Uint64()) < baseBlock { + firstInWindow++ + } + ps.polledChain = ps.polledChain[firstInWindow:] + // Find the earliest block we polled whose hash is no longer canonical (lists are short and ordered, so linear scan is efficient) + for i, polled := range ps.polledChain { + polledNumber := blockNumberToInt64(polled.Number.Uint64()) + canonicalHash := blockHashInHeadChain(headChain, polledNumber) + if canonicalHash == nil { + continue // above the top of the current window - nothing to compare against + } + if !bytes.Equal(canonicalHash, polled.Hash) { + log.L(ctx).Infof("Re-org detected at block %d (polled hash %s, now %s) - rewinding poll position from %d to %d", polledNumber, polled.Hash, canonicalHash, ps.fromBlock, polledNumber) + ps.fromBlock = polledNumber + ps.polledChain = ps.polledChain[:i] // records at/after the divergence are no longer valid + return + } + } +} + +// advance moves the poll position forwards after successfully processing blocks up to toBlock, +// recording the canonical hashes we hold for the polled range so a re-org behind the new position +// can be detected by checkReorgRewind on a later cycle. +// +// Note the hashes come from the headChain snapshot taken before the eth_getLogs query - if the +// chain re-organizes in between, the recorded hash and the queried logs can disagree, but the next +// cycle's continuity check then mismatches the new canonical view and rewinds us to re-poll. +func (ps *getLogsPollState) advance(headChain []*ethrpc.BlockInfoJSONRPC, toBlock int64) { + for _, bi := range headChain { + if n := blockNumberToInt64(bi.Number.Uint64()); n >= ps.fromBlock && n <= toBlock { + ps.polledChain = append(ps.polledChain, bi) + } + } + ps.fromBlock = toBlock + 1 +} + +// blockHashInHeadChain returns the hash of the given block number in the supplied canonical chain +// snapshot, or nil if that block number is not within the snapshot +func blockHashInHeadChain(headChain []*ethrpc.BlockInfoJSONRPC, blockNumber int64) ethtypes.HexBytes0xPrefix { + for _, bi := range headChain { + if blockNumberToInt64(bi.Number.Uint64()) == blockNumber { + return bi.Hash + } + } + return nil +} + +// steadyStateScanCeiling returns the highest block the steady-state scan is allowed to poll to. +// Both modes poll all the way to the head, but in full mode we must never pass a block above the +// stability horizon that the canonical view does not cover: the hashes recorded from the snapshot +// at scan time are what checkReorgRewind compares on later cycles, and a block scanned without a +// recorded hash could never be invalidated. The view is contiguous and sized to the +// checkpointBlockGap, so in steady operation its top IS the head (the head number itself comes +// from reconciled blocks) and this ceiling never binds - it holds the scan back only while the +// view is back-filling, such as at startup when it is seeded with a single anchor block. +func steadyStateScanCeiling(lightMode bool, chainHead, stableHead int64, headChain []*ethrpc.BlockInfoJSONRPC) int64 { + if lightMode { + return chainHead + } + verifiableTo := stableHead + if len(headChain) > 0 { + if snapTop := blockNumberToInt64(headChain[len(headChain)-1].Number.Uint64()); snapTop > verifiableTo { + verifiableTo = snapTop + } + } + if chainHead < verifiableTo { + return chainHead + } + return verifiableTo +} + +// leadGroupSteadyStateGetLogs is the alternative steady state to leadGroupSteadyState, selected with +// events.filterPollingMode: client. +// +// Instead of establishing a node-side filter, we track our own in-memory poll position and page +// forwards with stateless eth_getLogs range queries. +// +// Detection in this function is decoupled from confirmation (in FFTM). +// The confirmation manager waits for its own configured number of confirmations after each +// event's block (immediate for an event that arrives already deep enough), +// delivers to the application, and waits for the ack. +// +// For light mode that is based just on a comparison of event vs. head block numbers. +// For full mode there is a client-side tracking of the full unstable head and the confirmation +// list is re-calculated client-side. +// +// Details on polling approach: +// +// - FULL: We take a snapshot of the block listener's current view of the unstable head +// of the chain before we poll for events. Each poll we check if a new fork is apparent +// and re-poll from the position of the new fork in the chain. +// While we cannot be certain the chain we poll is the same we had a client-side view of +// before the poll, we know we will detect further divergence on subsequent poll cycles +// (as long as it occurs within the checkpointBlockGap). +// +// - LIGHT: Full blocks are not available to compare, so instead we re-scan the whole unstable +// window (the checkpointBlockGap blocks below the head) on every sweep, de-duplicating what +// we already delivered by block hash. +// +// Checkpoints come from two places: +// +// - Ack-based: each delivered event carries its own checkpoint, persisted by FFTM as batches +// are acknowledged. When events are flowing, the checkpoint moves forwards with delivery. +// +// - Inactivity: when no events are flowing, FFTM periodically polls our high water mark (see +// getHWM) recording how far we have scanned, held back checkpointBlockGap from the head. +// The LastDetected floor in that response stops an inactivity checkpoint overtaking a +// detected-but-unacknowledged event, so anything FFTM had not finished delivering is +// re-detected after a crash, at any confirmation count. +// +// This means on restart, processing will either: +// +// - Continue from the most recent confirmed acknowledged event, which could be the +// confirmation count back from the head (configured to 6 or 10 etc.). +// - Continue at least checkpointBlockGap behind wherever the head was before the restart, +// possibly significantly longer based on when the inactivity checkpoint was captured. +func (es *eventStream) leadGroupSteadyStateGetLogs() bool { + var ag *aggregatedListener + lastUpdate := -1 + failCount := 0 + poll := &getLogsPollState{fromBlock: -1, scanBlock: -1} + for { + if es.c.retry.DoFailureDelay(es.ctx, failCount) { + log.L(es.ctx).Debugf("Stream loop exiting") + return true + } + + // Build the aggregated listener list if it has changed + listenerChanged := es.buildReuseLeadGroupListener(&lastUpdate, &ag) + + caughtUpToHead := true + + // No need to poll for events, if we don't have any listeners + if len(ag.signatureSet) > 0 { + + // The block listener maintains a view of the highest block (in light mode this + // can go down as well as up). This call is just grabbing the current in-memory value. + chainHeadBlock, ok := es.c.blockListener.GetHighestBlock(es.ctx) + if !ok { + log.L(es.ctx).Debugf("Stream loop exiting (closed checking block height)") + return true + } + chainHead := blockNumberToInt64(chainHeadBlock) + + // (Re-)establish the poll position from the earliest listener HWM if we need to, + // just as filter mode (re-)establishes the fromBlock of its filter + if poll.fromBlock < 0 || listenerChanged { + fromBlock := int64(-1) + for _, l := range ag.listeners { + if lHWM := l.getHWMBlock(); fromBlock < 0 || lHWM < fromBlock { + fromBlock = lHWM + } + } + poll.reset(fromBlock) + } + + // The stability horizon is the point checkpointBlockGap behind the head, below which + // re-orgs are not expected + stableHead := chainHead - es.c.checkpointBlockGap + if stableHead < 0 { + stableHead = 0 + } + + // 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, so we measure against the same + // point - the two loops can never disagree and bounce control between each other. + if (stableHead - poll.fromBlock) > es.c.catchupThreshold { + log.L(es.ctx).Warnf("Block gap reached %d (above threshold of %d) - reverting to catchup mode", stableHead-poll.fromBlock, es.c.catchupThreshold) + return false + } + + // Both modes poll all the way to the head. What differs is how a re-org behind the + // poll position is repaired: + // In full chain tracking mode we check the blocks we already polled are still + // canonical against the block listener's chain view, rewinding our position if not. + // In light chain tracking mode there is no canonical chain view to check against, so + // instead we re-scan the whole unstable window on every sweep, de-duplicating what we + // already delivered by block hash - the committed position (poll.fromBlock) only ever + // advances to the stability horizon, and the sweep cursor pages beyond it to the head + lightMode := es.c.chainTrackingMode == ffcapi.ChainTrackingModeLight + var headChain []*ethrpc.BlockInfoJSONRPC + scanFrom := poll.fromBlock + if lightMode { + if poll.scanBlock > poll.fromBlock { + scanFrom = poll.scanBlock // mid-sweep - continue paging from the cursor + } + } else { + headChain = es.c.blockListener.SnapshotMonitoredHeadChain() + poll.checkReorgRewind(es.ctx, headChain) + scanFrom = poll.fromBlock // may have been rewound + } + + // Poll the next page of blocks, if there are any we haven't polled yet. + // Note if the scan ceiling holds us below the head, caughtUpToHead stays true: + // we wait a poll interval for the view to extend, we don't spin. + toBlock := steadyStateScanCeiling(lightMode, chainHead, stableHead, headChain) + if maxToBlock := scanFrom + es.c.catchupPageSize - 1; toBlock > maxToBlock { + toBlock = maxToBlock + caughtUpToHead = false // page again immediately, rather than waiting the polling interval + } + if toBlock >= scanFrom { + ethLogs, err := es.getBlockRangeLogs(es.ctx, ag, scanFrom, toBlock) + if err != nil { + log.L(es.ctx).Errorf("Failed to query block range fromBlock=%d toBlock=%d headBlock=%d: %s", scanFrom, toBlock, chainHead, err) + failCount++ + continue + } + if lightMode { + // Drop logs from block-versions already delivered on a previous sweep + ethLogs = poll.filterDelivered(ethLogs) + } + events, err := es.filterEnrichSort(es.ctx, ag, ethLogs) + if err != nil { + log.L(es.ctx).Errorf("Failed to filter/enrich events fromBlock=%d toBlock=%d headBlock=%d: %s", scanFrom, toBlock, chainHead, err) + failCount++ + continue + } + + // High water mark for the restart checkpoint is min(scan position, stability horizon): + // the scan runs to the head, but blocks in the unstable window can still change and the + // re-org repair state (recorded hashes / delivered blocks) is in-memory only, so the + // checkpoint holds at the horizon and a restart re-scans the window (redelivery is + // de-duplicated downstream). Light mode heads can also move backwards when the chain + // shortens, so there we additionally never move the committed base backwards. + hwmBlock := toBlock + 1 + if stableHead < hwmBlock { + hwmBlock = stableHead + } + if lightMode && hwmBlock < poll.fromBlock { + hwmBlock = poll.fromBlock + } + + // Dispatch the events + if es.dispatchSetHWMCheckExit(ag, events, hwmBlock) { + log.L(es.ctx).Debugf("Stream loop exiting") + return true + } + + // Update the head block to be the hwm block + es.headBlock.Store(hwmBlock) + + if lightMode { + // Record the block-versions we just delivered, advance the committed window + // base, and page or reset the sweep cursor + poll.advanceRescan(ethLogs, hwmBlock, toBlock, chainHead) + } else { + // Advance our poll position, recording the hashes of the blocks we polled so we + // can detect a re-org behind us on a later cycle + poll.advance(headChain, toBlock) + } + } else if lightMode { + // Nothing scannable (the head is at/below our committed base) - restart the sweep + // from the base when the chain grows again + poll.scanBlock = -1 + } + } + + // Reset failure count if we reach here + failCount = 0 + + // Sleep for the polling interval, unless we are paging through a backlog + if caughtUpToHead { + select { + case <-time.After(es.c.eventFilterPollingInterval): + case <-es.ctx.Done(): + log.L(es.ctx).Debugf("Stream loop stopping") + return true + } + } + } +} diff --git a/internal/ethereum/event_stream_test.go b/internal/ethereum/event_stream_test.go index 3817da8..9bf7589 100644 --- a/internal/ethereum/event_stream_test.go +++ b/internal/ethereum/event_stream_test.go @@ -18,14 +18,20 @@ package ethereum import ( "context" + "encoding/json" + "fmt" + "math" "strconv" "sync" "testing" "time" + "github.com/hyperledger-firefly/common/pkg/config" "github.com/hyperledger-firefly/common/pkg/fftypes" + "github.com/hyperledger-firefly/evmconnect/mocks/ethblocklistenermocks" "github.com/hyperledger-firefly/evmconnect/mocks/rpcbackendmocks" "github.com/hyperledger-firefly/evmconnect/pkg/ethrpc" + "github.com/hyperledger-firefly/signer/pkg/abi" "github.com/hyperledger-firefly/signer/pkg/ethtypes" "github.com/hyperledger-firefly/signer/pkg/rpcbackend" "github.com/hyperledger-firefly/transaction-manager/pkg/ffcapi" @@ -1282,3 +1288,1187 @@ func TestDispatchSetHWMDetectionBeforeScanPosition(t *testing.T) { assert.True(t, lastDetected.LessThan(scanned)) } + +func TestLeadGroupDeliverEventsGetLogsMode(t *testing.T) { + + lID := fftypes.NewUUID() + l1req := &ffcapi.EventListenerAddRequest{ + ListenerID: lID, + EventListenerOptions: ffcapi.EventListenerOptions{ + Filters: []fftypes.JSONAny{ + *fftypes.JSONAnyPtr(`{"address":"0xc89E46EEED41b777ca6625d37E1Cc87C5c037828","event":` + abiTransferEvent + `}`), + }, + Options: fftypes.JSONAnyPtr(`{}`), + FromBlock: strconv.Itoa(testHighBlock), + }, + } + + ctx, c, mRPC, done := newTestConnector(t, func(conf config.Section) { + conf.Set(EventsFilterPollingMode, string(FilterPollingModeClient)) + // A single-block monitored window: the real block listener seeds its canonical view with + // just the head block (mocked below), so the view covers the head and the scan can poll there + // - this test exercises delivery at the head, not re-org repair + conf.Set(EventsCheckpointBlockGap, 1) + }) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_blockNumber").Return(nil).Run(func(args mock.Arguments) { + *args[1].(*ethtypes.HexInteger) = *ethtypes.NewHexInteger64(testHighBlock) + }) + // The block listener seeds its monitored head view with this block, so the view covers the + // head and the getLogs scan can poll it. Deliberately a different hash to the event's block, so the + // enricher's eth_getBlockByHash below cannot be satisfied from the block cache. + seedBlockNumber := ethtypes.HexUint64(testHighBlock) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getBlockByNumber", seedBlockNumber.String(), false).Return(nil).Run(func(args mock.Arguments) { + *args[1].(**ethrpc.EVMBlockWithTxHashesJSONRPC) = ðrpc.EVMBlockWithTxHashesJSONRPC{BlockHeaderJSONRPC: ethrpc.BlockHeaderJSONRPC{ + Number: ethtypes.HexUint64(testHighBlock), + Hash: ethtypes.MustNewHexBytes0xPrefix("0x81f5bd39dbe293bb2a3467a29a30f16ec7c69fbef7a1eec9067a4f76de259e9a"), + ParentHash: ethtypes.MustNewHexBytes0xPrefix("0xd7f9ce8c2fd39a41d0cbdcd4a2c8c2ab8b58bdf5bbcae665b433e6a4f00e350f"), + }} + }) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getBlockByNumber", mock.Anything, false).Return(nil).Maybe() + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_newBlockFilter").Return(nil).Run(func(args mock.Arguments) { + *args[1].(*string) = testBlockFilterID1 + }).Maybe() + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getFilterChanges", testBlockFilterID1).Return(nil).Run(func(args mock.Arguments) { + *args[1].(*[]ethtypes.HexBytes0xPrefix) = nil + }).Maybe() + // Note there are deliberately no mocks for eth_newFilter/eth_getFilterLogs/eth_uninstallFilter - + // getLogs mode must never establish a node-side log filter + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getLogs", mock.Anything).Return(nil).Run(func(args mock.Arguments) { + filter := args[3].(*ethrpc.LogFilterJSONRPC) + assert.Equal(t, int64(testHighBlock), filter.FromBlock.BigInt().Int64()) + assert.Equal(t, int64(testHighBlock), filter.ToBlock.BigInt().Int64()) + *args[1].(*[]*ethrpc.LogJSONRPC) = []*ethrpc.LogJSONRPC{ + { + BlockNumber: ethtypes.HexUint64(testHighBlock), + TransactionIndex: ethtypes.HexUint64(64), + LogIndex: ethtypes.HexUint64(2), + BlockHash: ethtypes.MustNewHexBytes0xPrefix("0x6b012339fbb85b70c58ecfd97b31950c4a28bcef5226e12dbe551cb1abaf3b4c"), + Address: ethtypes.MustNewAddress("0xc89E46EEED41b777ca6625d37E1Cc87C5c037828"), + Topics: []ethtypes.HexBytes0xPrefix{ + ethtypes.MustNewHexBytes0xPrefix("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"), + ethtypes.MustNewHexBytes0xPrefix("0x0000000000000000000000003968ef051b422d3d1cdc182a88bba8dd922e6fa4"), + ethtypes.MustNewHexBytes0xPrefix("0x000000000000000000000000d0f2f5103fd050739a9fb567251bc460cc24d091"), + }, + Data: ethtypes.MustNewHexBytes0xPrefix("0x00000000000000000000000000000000000000000000000000000000000003e8"), + }, + } + }).Once() + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getLogs", mock.Anything).Return(nil).Run(func(args mock.Arguments) { + *args[1].(*[]*ethrpc.LogJSONRPC) = []*ethrpc.LogJSONRPC{} + }).Maybe() + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getBlockByHash", "0x6b012339fbb85b70c58ecfd97b31950c4a28bcef5226e12dbe551cb1abaf3b4c", false).Return(nil).Run(func(args mock.Arguments) { + *args[1].(**ethrpc.EVMBlockWithTxHashesJSONRPC) = ðrpc.EVMBlockWithTxHashesJSONRPC{BlockHeaderJSONRPC: ethrpc.BlockHeaderJSONRPC{ + Number: ethtypes.HexUint64(testHighBlock), + Hash: ethtypes.MustNewHexBytes0xPrefix("0x6b012339fbb85b70c58ecfd97b31950c4a28bcef5226e12dbe551cb1abaf3b4c"), + }} + }) + + es, events, _, done := testEventStreamExistingConnector(t, ctx, done, c, mRPC, l1req) + defer done() + + e := <-events + assert.Equal(t, fftypes.FFuint64(testHighBlock), e.Event.ID.BlockNumber) + assert.Equal(t, fftypes.FFuint64(64), e.Event.ID.TransactionIndex) + assert.Equal(t, fftypes.FFuint64(2), e.Event.ID.LogIndex) + assert.Equal(t, int64(testHighBlock), e.Checkpoint.(*listenerCheckpoint).Block) + assert.Equal(t, int64(64), e.Checkpoint.(*listenerCheckpoint).TransactionIndex) + assert.Equal(t, int64(2), e.Checkpoint.(*listenerCheckpoint).LogIndex) + assert.Equal(t, "0x3968ef051b422d3d1cdc182a88bba8dd922e6fa4", e.Event.Data.JSONObject().GetString("from")) + assert.Equal(t, "0xd0f2f5103fd050739a9fb567251bc460cc24d091", e.Event.Data.JSONObject().GetString("to")) + assert.Equal(t, "1000", e.Event.Data.JSONObject().GetString("value")) + + // The detection point must be recorded before the dispatch we received + _, lastDetected := es.listeners[*lID].getHWM() + assert.Equal(t, &listenerCheckpoint{Block: testHighBlock, TransactionIndex: 64, LogIndex: 2}, lastDetected) + + mRPC.AssertExpectations(t) +} + +func TestLeadGroupSteadyStateGetLogsFallbackToCatchup(t *testing.T) { + + ctx, c, mRPC, done := newTestConnector(t) + mockStreamLoopEmpty(mRPC) + defer done() + + es := &eventStream{ + id: fftypes.NewUUID(), + c: c, + ctx: ctx, + events: make(chan<- *ffcapi.ListenerEvent), + listeners: map[fftypes.UUID]*listener{ + *fftypes.NewUUID(): { + id: fftypes.NewUUID(), + config: listenerConfig{ + filters: []*eventFilter{ + {}, + }, + }, + }, + }, + streamLoopDone: make(chan struct{}), + } + es.headBlock.Store(-1) + + // The listener HWM of zero is way behind the chain head, so we must fall back to catchup + endedDueToExit := es.leadGroupSteadyStateGetLogs() + assert.False(t, endedDueToExit) +} + +func TestLeadGroupGetLogsRetry(t *testing.T) { + + l1req := &ffcapi.EventListenerAddRequest{ + ListenerID: fftypes.NewUUID(), + EventListenerOptions: ffcapi.EventListenerOptions{ + Filters: []fftypes.JSONAny{ + *fftypes.JSONAnyPtr(`{"event":` + abiTransferEvent + `}`), + }, + Options: fftypes.JSONAnyPtr(`{}`), + FromBlock: strconv.Itoa(testHighBlock), + }, + } + ctx, c, mRPC, done := newTestConnector(t, func(conf config.Section) { + conf.Set(EventsFilterPollingMode, string(FilterPollingModeClient)) + // A single-block monitored window: the real block listener seeds its canonical view with + // just the head block (mocked below), so the view covers the head and the scan can poll there + // - this test exercises the getLogs retry path + conf.Set(EventsCheckpointBlockGap, 1) + }) + + retried := make(chan struct{}) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_blockNumber").Return(nil).Run(func(args mock.Arguments) { + *args[1].(*ethtypes.HexInteger) = *ethtypes.NewHexInteger64(testHighBlock) + }) + // The block listener seeds its monitored head view with this block, so the view covers the + // head and the getLogs scan can poll it + seedBlockNumber := ethtypes.HexUint64(testHighBlock) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getBlockByNumber", seedBlockNumber.String(), false).Return(nil).Run(func(args mock.Arguments) { + *args[1].(**ethrpc.EVMBlockWithTxHashesJSONRPC) = ðrpc.EVMBlockWithTxHashesJSONRPC{BlockHeaderJSONRPC: ethrpc.BlockHeaderJSONRPC{ + Number: ethtypes.HexUint64(testHighBlock), + Hash: ethtypes.MustNewHexBytes0xPrefix("0x81f5bd39dbe293bb2a3467a29a30f16ec7c69fbef7a1eec9067a4f76de259e9a"), + ParentHash: ethtypes.MustNewHexBytes0xPrefix("0xd7f9ce8c2fd39a41d0cbdcd4a2c8c2ab8b58bdf5bbcae665b433e6a4f00e350f"), + }} + }) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getBlockByNumber", mock.Anything, false).Return(nil).Maybe() + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_newBlockFilter").Return(nil).Run(func(args mock.Arguments) { + *args[1].(*string) = testBlockFilterID1 + }).Maybe() + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getFilterChanges", testBlockFilterID1).Return(nil).Run(func(args mock.Arguments) { + *args[1].(*[]ethtypes.HexBytes0xPrefix) = nil + }).Maybe() + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getLogs", mock.Anything).Return(&rpcbackend.RPCError{Message: "pop"}). + Run(func(args mock.Arguments) { + close(retried) + }).Once() + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getLogs", mock.Anything).Return(&rpcbackend.RPCError{Message: "pop"}).Maybe() + + _, _, mRPC, done = testEventStreamExistingConnector(t, ctx, done, c, mRPC, l1req) + defer done() + + <-retried + +} + +// testGetLogsModeStream builds an eventStream with a mocked block listener for driving +// leadGroupSteadyStateGetLogs directly with precise control of the chain head and the +// canonical chain snapshot +func testGetLogsModeStream(t *testing.T, hwmBlock int64) (*eventStream, *listener, *rpcbackendmocks.Backend, *ethblocklistenermocks.BlockListener, context.CancelFunc, func()) { + _, c, mRPC, done := newTestConnector(t) + ctx, cancelCtx := context.WithCancel(context.Background()) + + c.catchupPageSize = 10 + c.catchupThreshold = 2000 + c.checkpointBlockGap = 50 + c.eventFilterPollingInterval = 1 * time.Millisecond + c.retry.MaximumDelay = 1 * time.Microsecond + + mbl := ethblocklistenermocks.NewBlockListener(t) + mbl.On("WaitClosed").Return().Maybe() + c.blockListener = mbl + + es := &eventStream{ + id: fftypes.NewUUID(), + c: c, + ctx: ctx, + events: make(chan<- *ffcapi.ListenerEvent), + listeners: map[fftypes.UUID]*listener{}, + streamLoopDone: make(chan struct{}), + } + lID := fftypes.NewUUID() + l := &listener{ + id: lID, + c: c, + es: es, + hwmBlock: hwmBlock, + config: listenerConfig{ + filters: []*eventFilter{ + {}, + }, + }, + } + es.listeners[*lID] = l + + return es, l, mRPC, mbl, cancelCtx, func() { + cancelCtx() + done() + } +} + +// testHeadChain builds a deterministic canonical chain snapshot - blocks at/above forkBlock get a +// different hash to the same block number below forkBlock, simulating a re-org fork at that point +func testHeadChain(fromBlock, toBlock, forkBlock int64) []*ethrpc.BlockInfoJSONRPC { + chain := make([]*ethrpc.BlockInfoJSONRPC, 0, toBlock-fromBlock+1) + for b := fromBlock; b <= toBlock; b++ { + fork := 0 + if b >= forkBlock { + fork = 1 + } + chain = append(chain, ðrpc.BlockInfoJSONRPC{ + Number: ethtypes.HexUint64(b), //nolint:gosec + Hash: ethtypes.MustNewHexBytes0xPrefix(fmt.Sprintf("0x%060x%04x", b, fork)), + }) + } + return chain +} + +func TestLeadGroupGetLogsPaginationAndHWMClamp(t *testing.T) { + + es, l, mRPC, mbl, cancelCtx, done := testGetLogsModeStream(t, 900) + defer done() + + mbl.On("GetHighestBlock", mock.Anything).Return(uint64(1000), true) + // The monitored view covers the whole unstable window (checkpointBlockGap 50 behind head + // 1000) - the scan may only pass blocks above the stability horizon that the view covers + mbl.On("SnapshotMonitoredHeadChain").Return(testHeadChain(951, 1000, 1001 /* no fork */)) + + type pollRange struct{ from, to, hwmAtCall int64 } + polls := make(chan pollRange, 20) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getLogs", mock.Anything).Return(nil).Run(func(args mock.Arguments) { + filter := args[3].(*ethrpc.LogFilterJSONRPC) + polls <- pollRange{ + from: filter.FromBlock.BigInt().Int64(), + to: filter.ToBlock.BigInt().Int64(), + hwmAtCall: l.getHWMBlock(), + } + *args[1].(*[]*ethrpc.LogJSONRPC) = []*ethrpc.LogJSONRPC{} + }) + + loopDone := make(chan bool, 1) + go func() { + loopDone <- es.leadGroupSteadyStateGetLogs() + }() + + // We page forwards in catchupPageSize pages until we reach the head of the chain. The HWM + // trails checkpointBlockGap (50) behind the chain head, but during pagination is clamped so + // it never passes the blocks we have not yet polled. + expected := []pollRange{ + {from: 900, to: 909, hwmAtCall: 900}, + {from: 910, to: 919, hwmAtCall: 910}, + {from: 920, to: 929, hwmAtCall: 920}, + {from: 930, to: 939, hwmAtCall: 930}, + {from: 940, to: 949, hwmAtCall: 940}, + {from: 950, to: 959, hwmAtCall: 950}, + {from: 960, to: 969, hwmAtCall: 950}, + {from: 970, to: 979, hwmAtCall: 950}, + {from: 980, to: 989, hwmAtCall: 950}, + {from: 990, to: 999, hwmAtCall: 950}, + {from: 1000, to: 1000, hwmAtCall: 950}, + } + for i, e := range expected { + select { + case p := <-polls: + assert.Equal(t, e, p, "poll %d", i) + case <-time.After(5 * time.Second): + t.Fatalf("timed out waiting for poll %d", i) + } + } + + cancelCtx() + assert.True(t, <-loopDone) + + // The HWM never advances past the reorg-safe point behind the head of the chain + assert.Equal(t, int64(950), l.getHWMBlock()) + assert.Equal(t, int64(950), es.headBlock.Load()) +} + +func TestLeadGroupGetLogsReorgRewind(t *testing.T) { + + es, _, mRPC, mbl, cancelCtx, done := testGetLogsModeStream(t, 995) + defer done() + + var mux sync.Mutex + reorged := false + mbl.On("GetHighestBlock", mock.Anything).Return(uint64(1000), true) + mbl.On("SnapshotMonitoredHeadChain").Return(func() []*ethrpc.BlockInfoJSONRPC { + mux.Lock() + defer mux.Unlock() + if reorged { + // Blocks 998 and above have been replaced on a new fork + return testHeadChain(990, 1000, 998) + } + return testHeadChain(990, 1000, 1001 /* no fork */) + }) + + polls := make(chan []int64, 20) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getLogs", mock.Anything).Return(nil).Run(func(args mock.Arguments) { + filter := args[3].(*ethrpc.LogFilterJSONRPC) + polls <- []int64{filter.FromBlock.BigInt().Int64(), filter.ToBlock.BigInt().Int64()} + *args[1].(*[]*ethrpc.LogJSONRPC) = []*ethrpc.LogJSONRPC{} + }) + + loopDone := make(chan bool, 1) + go func() { + loopDone <- es.leadGroupSteadyStateGetLogs() + }() + + // First poll takes us to the head of the chain, recording the hashes of blocks 995-1000 + assert.Equal(t, []int64{995, 1000}, <-polls) + + // Re-org blocks 998-1000 - the hash continuity check must find the earliest diverging + // block, and rewind the poll position to exactly there (not the whole unstable window) + mux.Lock() + reorged = true + mux.Unlock() + assert.Equal(t, []int64{998, 1000}, <-polls) + + cancelCtx() + assert.True(t, <-loopDone) +} + +func TestLeadGroupGetLogsFullModeHoldsAtViewCoverage(t *testing.T) { + + es, l, mRPC, mbl, cancelCtx, done := testGetLogsModeStream(t, 960) + defer done() + es.c.checkpointBlockGap = 6 + + // The re-org repair for blocks above the stability horizon relies on the hashes + // recorded from the monitored view at scan time, so the scan must never pass a block above + // the horizon that the view does not cover. The view back-fills here in three stages: + // empty (startup, before the seed), covering the window base only, then the full window. + var mux sync.Mutex + headChain := []*ethrpc.BlockInfoJSONRPC{} + mbl.On("GetHighestBlock", mock.Anything).Return(uint64(1000), true) + mbl.On("SnapshotMonitoredHeadChain").Return(func() []*ethrpc.BlockInfoJSONRPC { + mux.Lock() + defer mux.Unlock() + return headChain + }) + setHeadChain := func(chain []*ethrpc.BlockInfoJSONRPC) { + mux.Lock() + defer mux.Unlock() + headChain = chain + } + + type pollRange struct{ from, to, hwmAtCall int64 } + polls := make(chan pollRange, 20) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getLogs", mock.Anything).Return(nil).Run(func(args mock.Arguments) { + filter := args[3].(*ethrpc.LogFilterJSONRPC) + polls <- pollRange{ + from: filter.FromBlock.BigInt().Int64(), + to: filter.ToBlock.BigInt().Int64(), + hwmAtCall: l.getHWMBlock(), + } + *args[1].(*[]*ethrpc.LogJSONRPC) = []*ethrpc.LogJSONRPC{} + }) + + loopDone := make(chan bool, 1) + go func() { + loopDone <- es.leadGroupSteadyStateGetLogs() + }() + + readPoll := func(what string) pollRange { + select { + case p := <-polls: + return p + case <-time.After(5 * time.Second): + t.Fatalf("timed out waiting for %s", what) + return pollRange{} + } + } + + // With an empty view we page up to the stability horizon (994) and no further - blocks at or + // below the horizon are stable by definition and need no recorded hash + assert.Equal(t, pollRange{from: 960, to: 969, hwmAtCall: 960}, readPoll("page 1")) + assert.Equal(t, pollRange{from: 970, to: 979, hwmAtCall: 970}, readPoll("page 2")) + assert.Equal(t, pollRange{from: 980, to: 989, hwmAtCall: 980}, readPoll("page 3")) + assert.Equal(t, pollRange{from: 990, to: 994, hwmAtCall: 990}, readPoll("scan to horizon")) + + // The view now covers the base of the unstable window - the scan follows it, exactly + setHeadChain(testHeadChain(995, 996, 1001 /* no fork */)) + assert.Equal(t, pollRange{from: 995, to: 996, hwmAtCall: 994}, readPoll("scan to view coverage")) + + // The view completes to the head - the scan completes with it, and the HWM/checkpoint stays + // at the stability horizon throughout + setHeadChain(testHeadChain(995, 1000, 1001 /* no fork */)) + assert.Equal(t, pollRange{from: 997, to: 1000, hwmAtCall: 994}, readPoll("scan to head")) + + cancelCtx() + assert.True(t, <-loopDone) + assert.Equal(t, int64(994), l.getHWMBlock()) + assert.Equal(t, int64(994), es.headBlock.Load()) +} + +func TestGetLogsPollStateCheckReorgRewind(t *testing.T) { + + ctx := context.Background() + canonicalChain := testHeadChain(990, 1000, 1001 /* no fork */) + + // Nothing recorded - nothing to check + ps := &getLogsPollState{fromBlock: 1001} + ps.checkReorgRewind(ctx, canonicalChain) + assert.Equal(t, int64(1001), ps.fromBlock) + + // Empty snapshot - nothing to compare against + ps = &getLogsPollState{fromBlock: 1001, polledChain: testHeadChain(995, 1000, 1001)} + ps.checkReorgRewind(ctx, []*ethrpc.BlockInfoJSONRPC{}) + assert.Equal(t, int64(1001), ps.fromBlock) + assert.Len(t, ps.polledChain, 6) + + // Records below the monitored window are pruned as stable; matching hashes cause no rewind + ps = &getLogsPollState{fromBlock: 1001, polledChain: testHeadChain(985, 1000, 1001)} + ps.checkReorgRewind(ctx, canonicalChain) + assert.Equal(t, int64(1001), ps.fromBlock) + assert.Len(t, ps.polledChain, 11) // 990-1000 + assert.Equal(t, int64(990), int64(ps.polledChain[0].Number)) + + // Records above the top of the window are skipped (nothing to compare against) + ps = &getLogsPollState{fromBlock: 1006, polledChain: testHeadChain(995, 1005, 1006)} + ps.checkReorgRewind(ctx, canonicalChain) + assert.Equal(t, int64(1006), ps.fromBlock) + assert.Len(t, ps.polledChain, 11) // 995-1005 + + // The earliest diverging block wins - records forked at 997 rewind exactly there, and + // the records at/after the divergence are discarded + ps = &getLogsPollState{fromBlock: 1001, polledChain: testHeadChain(995, 1000, 997)} + ps.checkReorgRewind(ctx, canonicalChain) + assert.Equal(t, int64(997), ps.fromBlock) + assert.Len(t, ps.polledChain, 2) // 995, 996 + assert.Equal(t, int64(996), int64(ps.polledChain[1].Number)) +} + +func TestGetLogsPollStateAdvance(t *testing.T) { + + // Only blocks in the polled range [fromBlock, toBlock] are recorded + ps := &getLogsPollState{fromBlock: 995} + ps.advance(testHeadChain(990, 1000, 1001), 998) + assert.Equal(t, int64(999), ps.fromBlock) + assert.Len(t, ps.polledChain, 4) // 995-998 + assert.Equal(t, int64(995), int64(ps.polledChain[0].Number)) + assert.Equal(t, int64(998), int64(ps.polledChain[3].Number)) + + // Blocks polled below the monitored window leave no record (they are already stable) + ps = &getLogsPollState{fromBlock: 900} + ps.advance(testHeadChain(990, 1000, 1001), 950) + assert.Equal(t, int64(951), ps.fromBlock) + assert.Empty(t, ps.polledChain) +} + +func TestGetLogsPollStateAdvanceRescan(t *testing.T) { + + testLog := func(blockNumber int64, blockHash string) *ethrpc.LogJSONRPC { + return ðrpc.LogJSONRPC{ + BlockNumber: ethtypes.HexUint64(blockNumber), //nolint:gosec + BlockHash: ethtypes.MustNewHexBytes0xPrefix(blockHash), + } + } + hash995 := "0x00000000000000000000000000000000000000000000000000000000000003e3" + hash996 := "0x00000000000000000000000000000000000000000000000000000000000003e4" + hash1000 := "0x00000000000000000000000000000000000000000000000000000000000003e8" + + // A page mid-sweep: delivered block-versions are recorded, the committed base holds at the + // stability horizon, and the sweep cursor pages onwards without waiting + ps := &getLogsPollState{fromBlock: 994, scanBlock: -1} + ps.advanceRescan([]*ethrpc.LogJSONRPC{testLog(995, hash995), testLog(996, hash996)}, 994, 996, 1000) + assert.Equal(t, int64(994), ps.fromBlock) + assert.Equal(t, int64(997), ps.scanBlock) + assert.Equal(t, map[string]int64{ + string(ethtypes.MustNewHexBytes0xPrefix(hash995)): 995, + string(ethtypes.MustNewHexBytes0xPrefix(hash996)): 996, + }, ps.deliveredBlocks) + + // The final page of the sweep reaches the head - the cursor resets so the next sweep + // re-scans the whole window from the committed base + ps.advanceRescan([]*ethrpc.LogJSONRPC{testLog(1000, hash1000)}, 994, 1000, 1000) + assert.Equal(t, int64(994), ps.fromBlock) + assert.Equal(t, int64(-1), ps.scanBlock) + assert.Len(t, ps.deliveredBlocks, 3) + + // The chain grows and the committed base advances with the stability horizon - records for + // blocks that fall below the base are pruned (those blocks are stable, never re-scanned) + ps.advanceRescan(nil, 996, 1002, 1002) + assert.Equal(t, int64(996), ps.fromBlock) + assert.Equal(t, int64(-1), ps.scanBlock) + assert.Equal(t, map[string]int64{ + string(ethtypes.MustNewHexBytes0xPrefix(hash996)): 996, + string(ethtypes.MustNewHexBytes0xPrefix(hash1000)): 1000, + }, ps.deliveredBlocks) +} + +func TestBlockNumberToInt64Overflow(t *testing.T) { + assert.Equal(t, int64(12345), blockNumberToInt64(12345)) + assert.Panics(t, func() { + blockNumberToInt64(uint64(math.MaxInt64) + 1) + }) +} + +func TestLeadGroupGetLogsExitGettingHighBlock(t *testing.T) { + + es, _, _, mbl, _, done := testGetLogsModeStream(t, 0) + defer done() + + // The block listener closing while we check the chain head means we are shutting down + mbl.On("GetHighestBlock", mock.Anything).Return(uint64(0), false) + assert.True(t, es.leadGroupSteadyStateGetLogs()) +} + +func TestLeadGroupGetLogsHWMClampAtZero(t *testing.T) { + + es, l, mRPC, mbl, cancelCtx, done := testGetLogsModeStream(t, 0) + defer done() + + mbl.On("GetHighestBlock", mock.Anything).Return(uint64(5), true) + // The whole chain is inside the monitored window (head 5, checkpointBlockGap 50) + mbl.On("SnapshotMonitoredHeadChain").Return(testHeadChain(0, 5, 6 /* no fork */)) + + polls := make(chan []int64, 10) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getLogs", mock.Anything).Return(nil).Run(func(args mock.Arguments) { + filter := args[3].(*ethrpc.LogFilterJSONRPC) + polls <- []int64{filter.FromBlock.BigInt().Int64(), filter.ToBlock.BigInt().Int64()} + *args[1].(*[]*ethrpc.LogJSONRPC) = []*ethrpc.LogJSONRPC{} + }) + + loopDone := make(chan bool, 1) + go func() { + loopDone <- es.leadGroupSteadyStateGetLogs() + }() + + assert.Equal(t, []int64{0, 5}, <-polls) + + cancelCtx() + assert.True(t, <-loopDone) + + // The whole chain (head 5) is within checkpointBlockGap (50) of genesis, so the HWM clamps at zero + assert.Equal(t, int64(0), l.getHWMBlock()) + assert.Equal(t, int64(0), es.headBlock.Load()) +} + +func TestLeadGroupGetLogsExitDuringDispatch(t *testing.T) { + + lID := fftypes.NewUUID() + l1req := &ffcapi.EventListenerAddRequest{ + ListenerID: lID, + EventListenerOptions: ffcapi.EventListenerOptions{ + Filters: []fftypes.JSONAny{ + *fftypes.JSONAnyPtr(`{"address":"0xc89E46EEED41b777ca6625d37E1Cc87C5c037828","event":` + abiTransferEvent + `}`), + }, + Options: fftypes.JSONAnyPtr(`{}`), + FromBlock: strconv.Itoa(testHighBlock), + }, + } + + ctx, c, mRPC, done := newTestConnector(t, func(conf config.Section) { + conf.Set(EventsFilterPollingMode, string(FilterPollingModeClient)) + // A single-block monitored window: the real block listener seeds its canonical view with + // just the head block (mocked below), so the view covers the head and the scan can poll there + // - this test exercises the exit-during-dispatch path + conf.Set(EventsCheckpointBlockGap, 1) + }) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_blockNumber").Return(nil).Run(func(args mock.Arguments) { + *args[1].(*ethtypes.HexInteger) = *ethtypes.NewHexInteger64(testHighBlock) + }) + // The block listener seeds its monitored head view with this block, so the view covers the + // head and the getLogs scan can poll it + seedBlockNumber := ethtypes.HexUint64(testHighBlock) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getBlockByNumber", seedBlockNumber.String(), false).Return(nil).Run(func(args mock.Arguments) { + *args[1].(**ethrpc.EVMBlockWithTxHashesJSONRPC) = ðrpc.EVMBlockWithTxHashesJSONRPC{BlockHeaderJSONRPC: ethrpc.BlockHeaderJSONRPC{ + Number: ethtypes.HexUint64(testHighBlock), + Hash: ethtypes.MustNewHexBytes0xPrefix("0x81f5bd39dbe293bb2a3467a29a30f16ec7c69fbef7a1eec9067a4f76de259e9a"), + ParentHash: ethtypes.MustNewHexBytes0xPrefix("0xd7f9ce8c2fd39a41d0cbdcd4a2c8c2ab8b58bdf5bbcae665b433e6a4f00e350f"), + }} + }) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getBlockByNumber", mock.Anything, false).Return(nil).Maybe() + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_newBlockFilter").Return(nil).Run(func(args mock.Arguments) { + *args[1].(*string) = testBlockFilterID1 + }).Maybe() + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getFilterChanges", testBlockFilterID1).Return(nil).Run(func(args mock.Arguments) { + *args[1].(*[]ethtypes.HexBytes0xPrefix) = nil + }).Maybe() + polled := make(chan struct{}) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getLogs", mock.Anything).Return(nil).Run(func(args mock.Arguments) { + close(polled) + *args[1].(*[]*ethrpc.LogJSONRPC) = []*ethrpc.LogJSONRPC{ + { + BlockNumber: ethtypes.HexUint64(testHighBlock), + TransactionIndex: ethtypes.HexUint64(64), + LogIndex: ethtypes.HexUint64(2), + BlockHash: ethtypes.MustNewHexBytes0xPrefix("0x6b012339fbb85b70c58ecfd97b31950c4a28bcef5226e12dbe551cb1abaf3b4c"), + Address: ethtypes.MustNewAddress("0xc89E46EEED41b777ca6625d37E1Cc87C5c037828"), + Topics: []ethtypes.HexBytes0xPrefix{ + ethtypes.MustNewHexBytes0xPrefix("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"), + ethtypes.MustNewHexBytes0xPrefix("0x0000000000000000000000003968ef051b422d3d1cdc182a88bba8dd922e6fa4"), + ethtypes.MustNewHexBytes0xPrefix("0x000000000000000000000000d0f2f5103fd050739a9fb567251bc460cc24d091"), + }, + Data: ethtypes.MustNewHexBytes0xPrefix("0x00000000000000000000000000000000000000000000000000000000000003e8"), + }, + } + }).Once() + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getLogs", mock.Anything).Return(nil).Run(func(args mock.Arguments) { + *args[1].(*[]*ethrpc.LogJSONRPC) = []*ethrpc.LogJSONRPC{} + }).Maybe() + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getBlockByHash", mock.Anything, false).Return(nil).Run(func(args mock.Arguments) { + *args[1].(**ethrpc.EVMBlockWithTxHashesJSONRPC) = ðrpc.EVMBlockWithTxHashesJSONRPC{BlockHeaderJSONRPC: ethrpc.BlockHeaderJSONRPC{ + Number: ethtypes.HexUint64(testHighBlock), + Hash: ethtypes.MustNewHexBytes0xPrefix("0x6b012339fbb85b70c58ecfd97b31950c4a28bcef5226e12dbe551cb1abaf3b4c"), + }} + }).Maybe() + + // Note we never read from the events channel, so the dispatch of the event blocks until + // the stream context is cancelled - covering the exit path during dispatch + es, _, mRPC, done := testEventStreamExistingConnector(t, ctx, done, c, mRPC, l1req) + + <-polled + done() + <-es.streamLoopDone +} + +func TestLeadGroupGetLogsLightModeRescansUnstableWindow(t *testing.T) { + + es, l, mRPC, mbl, cancelCtx, done := testGetLogsModeStream(t, 960) + defer done() + es.c.chainTrackingMode = ffcapi.ChainTrackingModeLight + es.c.checkpointBlockGap = 6 // tuned a small margin above the confirmation target in this mode + + var headMux sync.Mutex + head := uint64(1000) + mbl.On("GetHighestBlock", mock.Anything).Return(func(context.Context) (uint64, bool) { + headMux.Lock() + defer headMux.Unlock() + return head, true + }) + // Note no SnapshotMonitoredHeadChain expectation - in light mode there is no canonical chain + // view, and the loop must never ask for one + + type pollRange struct{ from, to, hwmAtCall int64 } + polls := make(chan pollRange, 20) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getLogs", mock.Anything).Return(nil).Run(func(args mock.Arguments) { + filter := args[3].(*ethrpc.LogFilterJSONRPC) + polls <- pollRange{ + from: filter.FromBlock.BigInt().Int64(), + to: filter.ToBlock.BigInt().Int64(), + hwmAtCall: l.getHWMBlock(), + } + *args[1].(*[]*ethrpc.LogJSONRPC) = []*ethrpc.LogJSONRPC{} + }) + + loopDone := make(chan bool, 1) + go func() { + loopDone <- es.leadGroupSteadyStateGetLogs() + }() + + // Light mode pages all the way to the head just like full mode, but the committed position + // (and with it the HWM/checkpoint) holds at the stability horizon - checkpointBlockGap (6) + // behind the head + expected := []pollRange{ + {from: 960, to: 969, hwmAtCall: 960}, + {from: 970, to: 979, hwmAtCall: 970}, + {from: 980, to: 989, hwmAtCall: 980}, + {from: 990, to: 999, hwmAtCall: 990}, + {from: 1000, to: 1000, hwmAtCall: 994}, + } + for i, e := range expected { + select { + case p := <-polls: + assert.Equal(t, e, p, "poll %d", i) + case <-time.After(5 * time.Second): + t.Fatalf("timed out waiting for poll %d", i) + } + } + + // In the steady state the whole unstable window is re-scanned on every polling interval - + // a re-org inside the window is invisible to light mode (it only sees the head number), so + // the changed block hashes in the re-scan are the only signal for replaced events + select { + case p := <-polls: + assert.Equal(t, pollRange{from: 994, to: 1000, hwmAtCall: 994}, p) + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for window re-scan") + } + + // When the head advances, the window slides with it + headMux.Lock() + head = 1001 + headMux.Unlock() + deadline := time.After(5 * time.Second) + for { + var p pollRange + select { + case p = <-polls: + case <-deadline: + t.Fatal("timed out waiting for the window to slide") + } + if p.from == 995 { + assert.Equal(t, pollRange{from: 995, to: 1001, hwmAtCall: 995}, p) + break + } + // Re-scans of the old window (including the first to span to the new head) come first + assert.Equal(t, int64(994), p.from) + } + + cancelCtx() + assert.True(t, <-loopDone) + assert.Equal(t, int64(995), l.getHWMBlock()) + assert.Equal(t, int64(995), es.headBlock.Load()) +} + +func TestLeadGroupGetLogsLightModeHeadBelowGap(t *testing.T) { + + es, l, mRPC, mbl, cancelCtx, done := testGetLogsModeStream(t, 0) + defer done() + es.c.chainTrackingMode = ffcapi.ChainTrackingModeLight + es.c.checkpointBlockGap = 6 + + mbl.On("GetHighestBlock", mock.Anything).Return(uint64(5), true) + + polls := make(chan []int64, 10) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getLogs", mock.Anything).Return(nil).Run(func(args mock.Arguments) { + filter := args[3].(*ethrpc.LogFilterJSONRPC) + polls <- []int64{filter.FromBlock.BigInt().Int64(), filter.ToBlock.BigInt().Int64()} + *args[1].(*[]*ethrpc.LogJSONRPC) = []*ethrpc.LogJSONRPC{} + }) + + loopDone := make(chan bool, 1) + go func() { + loopDone <- es.leadGroupSteadyStateGetLogs() + }() + + // The whole chain (head 5) is within checkpointBlockGap (6), so nothing is stable yet: we + // still poll to the head, but the committed position (and with it the HWM/checkpoint) clamps + // at genesis, and the whole chain is re-scanned each cycle + assert.Equal(t, []int64{0, 5}, <-polls) + assert.Equal(t, []int64{0, 5}, <-polls) + + cancelCtx() + assert.True(t, <-loopDone) + assert.Equal(t, int64(0), l.getHWMBlock()) + assert.Equal(t, int64(0), es.headBlock.Load()) +} + +func TestLeadGroupGetLogsLightModeZeroGapPollsToHead(t *testing.T) { + + es, l, mRPC, mbl, cancelCtx, done := testGetLogsModeStream(t, 0) + defer done() + es.c.chainTrackingMode = ffcapi.ChainTrackingModeLight + es.c.checkpointBlockGap = 0 // an instant-finality chain - every block is stable immediately + + mbl.On("GetHighestBlock", mock.Anything).Return(uint64(5), true) + + polls := make(chan []int64, 10) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getLogs", mock.Anything).Return(nil).Run(func(args mock.Arguments) { + filter := args[3].(*ethrpc.LogFilterJSONRPC) + polls <- []int64{filter.FromBlock.BigInt().Int64(), filter.ToBlock.BigInt().Int64()} + *args[1].(*[]*ethrpc.LogJSONRPC) = []*ethrpc.LogJSONRPC{} + }) + + loopDone := make(chan bool, 1) + go func() { + loopDone <- es.leadGroupSteadyStateGetLogs() + }() + + // With gap 0 every block is stable the moment it is polled, so after the first scan to the + // head only the head block itself is re-scanned (the horizon formula min(scan position, + // head - gap) is uniform with full mode), and the HWM/checkpoint sits at the head + assert.Equal(t, []int64{0, 5}, <-polls) + assert.Equal(t, []int64{5, 5}, <-polls) + + cancelCtx() + assert.True(t, <-loopDone) + assert.Equal(t, int64(5), l.getHWMBlock()) + assert.Equal(t, int64(5), es.headBlock.Load()) +} + +func TestLeadGroupGetLogsLightModeNoCatchupOscillation(t *testing.T) { + + es, _, mRPC, mbl, cancelCtx, done := testGetLogsModeStream(t, 905) + defer done() + es.c.chainTrackingMode = ffcapi.ChainTrackingModeLight + es.c.catchupThreshold = 90 + es.c.checkpointBlockGap = 6 + + mbl.On("GetHighestBlock", mock.Anything).Return(uint64(1000), true) + + polls := make(chan []int64, 10) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getLogs", mock.Anything).Return(nil).Run(func(args mock.Arguments) { + filter := args[3].(*ethrpc.LogFilterJSONRPC) + polls <- []int64{filter.FromBlock.BigInt().Int64(), filter.ToBlock.BigInt().Int64()} + *args[1].(*[]*ethrpc.LogJSONRPC) = []*ethrpc.LogJSONRPC{} + }) + + loopDone := make(chan bool, 1) + go func() { + loopDone <- es.leadGroupSteadyStateGetLogs() + }() + + // The raw gap to the head (1000-905=95) is over the catchup threshold (90), but the gap to + // the stability horizon that catchup would poll to (994-905=89) is not - we must measure the + // same way and stay in steady state, or we would bounce between the two loops without making + // progress + assert.Equal(t, []int64{905, 914}, <-polls) + + cancelCtx() + assert.True(t, <-loopDone) +} + +func TestLeadGroupGetLogsLightModeRescanDedupAndReorgRedetect(t *testing.T) { + + // The block hash before and after a re-org of block 998 - same block number, same transaction, + // but any change to the block content changes its hash + const blockHashA = "0x6b012339fbb85b70c58ecfd97b31950c4a28bcef5226e12dbe551cb1abaf3b4a" + const blockHashB = "0x6b012339fbb85b70c58ecfd97b31950c4a28bcef5226e12dbe551cb1abaf3b4b" + + es, l, mRPC, mbl, cancelCtx, done := testGetLogsModeStream(t, 994) + defer done() + es.c.chainTrackingMode = ffcapi.ChainTrackingModeLight + es.c.checkpointBlockGap = 6 + es.c.chainID = "12345" + es.c.eventBlockTimestamps = false + + // A real filter and enricher on the listener, and a readable events channel, so the test can + // observe exactly what gets delivered + var transferEvent *abi.Entry + err := json.Unmarshal([]byte(abiTransferEvent), &transferEvent) + require.NoError(t, err) + l.config.filters = []*eventFilter{{ + Event: transferEvent, + Topic0: ethtypes.MustNewHexBytes0xPrefix("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"), + }} + l.config.options = &listenerOptions{} + l.ee = &eventEnricher{connector: es.c} + events := make(chan *ffcapi.ListenerEvent) + es.events = events + + mbl.On("GetHighestBlock", mock.Anything).Return(uint64(1000), true) + + makeTransferLog := func(blockHash string) *ethrpc.LogJSONRPC { + return ðrpc.LogJSONRPC{ + BlockNumber: ethtypes.HexUint64(998), + TransactionIndex: ethtypes.HexUint64(0), + LogIndex: ethtypes.HexUint64(0), + TransactionHash: ethtypes.MustNewHexBytes0xPrefix("0x1a5df31d1371f7fc9f242e2b19d287d32e1205cad392ce6ab4b1cf87dbdc9b74"), + BlockHash: ethtypes.MustNewHexBytes0xPrefix(blockHash), + Address: ethtypes.MustNewAddress("0xc89E46EEED41b777ca6625d37E1Cc87C5c037828"), + Topics: []ethtypes.HexBytes0xPrefix{ + ethtypes.MustNewHexBytes0xPrefix("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"), + ethtypes.MustNewHexBytes0xPrefix("0x0000000000000000000000003968ef051b422d3d1cdc182a88bba8dd922e6fa4"), + ethtypes.MustNewHexBytes0xPrefix("0x000000000000000000000000d0f2f5103fd050739a9fb567251bc460cc24d091"), + }, + Data: ethtypes.MustNewHexBytes0xPrefix("0x00000000000000000000000000000000000000000000000000000000000003e8"), + } + } + + // Every scan of the window returns the same event - on the original fork until the test + // re-orgs it, then on the replacement fork (same block number, new block hash) + var mux sync.Mutex + reorged := false + polls := make(chan []int64, 20) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getLogs", mock.Anything).Return(nil).Run(func(args mock.Arguments) { + filter := args[3].(*ethrpc.LogFilterJSONRPC) + select { + case polls <- []int64{filter.FromBlock.BigInt().Int64(), filter.ToBlock.BigInt().Int64()}: + default: // drop poll records rather than block the loop - the test only samples them + } + mux.Lock() + defer mux.Unlock() + if reorged { + *args[1].(*[]*ethrpc.LogJSONRPC) = []*ethrpc.LogJSONRPC{makeTransferLog(blockHashB)} + } else { + *args[1].(*[]*ethrpc.LogJSONRPC) = []*ethrpc.LogJSONRPC{makeTransferLog(blockHashA)} + } + }) + + loopDone := make(chan bool, 1) + go func() { + loopDone <- es.leadGroupSteadyStateGetLogs() + }() + + readEvent := func(what string) *ffcapi.ListenerEvent { + select { + case e := <-events: + return e + case <-time.After(5 * time.Second): + t.Fatalf("timed out waiting for %s", what) + return nil + } + } + + // The first scan of the window [994,1000] detects and delivers the event + assert.Equal(t, []int64{994, 1000}, <-polls) + ev := readEvent("first detection") + assert.Equal(t, blockHashA, ev.Event.ID.BlockHash) + assert.Equal(t, fftypes.FFuint64(998), ev.Event.ID.BlockNumber) + assert.Equal(t, int64(998), ev.Checkpoint.(*listenerCheckpoint).Block) + + // Re-scans return the identical block-version, which is de-duplicated - wait until we have + // seen at least two more full window scans without any event arriving (a duplicate delivery + // would be sitting in the unbuffered events channel, and would be received below instead of + // the replacement event) + for i := 0; i < 2; i++ { + assert.Equal(t, []int64{994, 1000}, <-polls) + } + + // Re-org block 998 - in light mode the head number alone cannot signal this, but the re-scan + // returns the replacement block-version, whose new hash makes its events new detections + mux.Lock() + reorged = true + mux.Unlock() + ev = readEvent("re-detection after re-org") + assert.Equal(t, blockHashB, ev.Event.ID.BlockHash) + assert.Equal(t, fftypes.FFuint64(998), ev.Event.ID.BlockNumber) + + // The committed position (and HWM/checkpoint) held at the stability horizon throughout + assert.Equal(t, int64(994), l.getHWMBlock()) + + cancelCtx() + assert.True(t, <-loopDone) +} + +func TestLeadGroupGetLogsLightModeHeadShrink(t *testing.T) { + + es, l, mRPC, mbl, cancelCtx, done := testGetLogsModeStream(t, 994) + defer done() + es.c.chainTrackingMode = ffcapi.ChainTrackingModeLight + es.c.checkpointBlockGap = 6 + + // A light mode head can move backwards (each poll asks eth_blockNumber, and nodes/gateways can + // disagree or fork-switch to a shorter chain). One head value per loop cycle: + // 1000 (normal), 995 (horizon drops below our committed base), 990 (head drops below the + // committed base - nothing to scan at all), then 1001 (chain grows past where it was) + heads := []uint64{1000, 995, 990, 1001} + var headMux sync.Mutex + headIdx := 0 + mbl.On("GetHighestBlock", mock.Anything).Return(func(context.Context) (uint64, bool) { + headMux.Lock() + defer headMux.Unlock() + head := heads[headIdx] + if headIdx < len(heads)-1 { + headIdx++ + } + return head, true + }) + + type pollRange struct{ from, to, hwmAtCall int64 } + polls := make(chan pollRange, 20) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getLogs", mock.Anything).Return(nil).Run(func(args mock.Arguments) { + filter := args[3].(*ethrpc.LogFilterJSONRPC) + polls <- pollRange{ + from: filter.FromBlock.BigInt().Int64(), + to: filter.ToBlock.BigInt().Int64(), + hwmAtCall: l.getHWMBlock(), + } + *args[1].(*[]*ethrpc.LogJSONRPC) = []*ethrpc.LogJSONRPC{} + }) + + loopDone := make(chan bool, 1) + go func() { + loopDone <- es.leadGroupSteadyStateGetLogs() + }() + + expected := []pollRange{ + // Head 1000: normal scan of the window, committed base at the horizon (994) + {from: 994, to: 1000, hwmAtCall: 994}, + // Head 995: the horizon (989) is now below our committed base, but the base (and with it + // the HWM/checkpoint) never moves backwards - we re-scan [994,995] and hold at 994 + {from: 994, to: 995, hwmAtCall: 994}, + // Head 990 produced no poll at all (nothing scannable above the committed base). + // Head 1001: the chain regrew - the sweep restarts from the held base, which then + // advances to the new horizon (995) + {from: 994, to: 1001, hwmAtCall: 994}, + {from: 995, to: 1001, hwmAtCall: 995}, + } + for i, e := range expected { + select { + case p := <-polls: + assert.Equal(t, e, p, "poll %d", i) + case <-time.After(5 * time.Second): + t.Fatalf("timed out waiting for poll %d", i) + } + } + + cancelCtx() + assert.True(t, <-loopDone) + assert.Equal(t, int64(995), l.getHWMBlock()) + assert.Equal(t, int64(995), es.headBlock.Load()) +} + +func TestLeadGroupGetLogsLightModeEnrichFailRetry(t *testing.T) { + + const blockHash998 = "0x6b012339fbb85b70c58ecfd97b31950c4a28bcef5226e12dbe551cb1abaf3b4a" + + es, l, mRPC, mbl, cancelCtx, done := testGetLogsModeStream(t, 994) + defer done() + es.c.chainTrackingMode = ffcapi.ChainTrackingModeLight + es.c.checkpointBlockGap = 6 + es.c.chainID = "12345" + es.c.eventBlockTimestamps = true // enrichment fetches the block for its timestamp + + var transferEvent *abi.Entry + err := json.Unmarshal([]byte(abiTransferEvent), &transferEvent) + require.NoError(t, err) + l.config.filters = []*eventFilter{{ + Event: transferEvent, + Topic0: ethtypes.MustNewHexBytes0xPrefix("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"), + }} + l.config.options = &listenerOptions{} + l.ee = &eventEnricher{connector: es.c} + events := make(chan *ffcapi.ListenerEvent) + es.events = events + + mbl.On("GetHighestBlock", mock.Anything).Return(uint64(1000), true) + + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getLogs", mock.Anything).Return(nil).Run(func(args mock.Arguments) { + *args[1].(*[]*ethrpc.LogJSONRPC) = []*ethrpc.LogJSONRPC{{ + BlockNumber: ethtypes.HexUint64(998), + TransactionIndex: ethtypes.HexUint64(0), + LogIndex: ethtypes.HexUint64(0), + TransactionHash: ethtypes.MustNewHexBytes0xPrefix("0x1a5df31d1371f7fc9f242e2b19d287d32e1205cad392ce6ab4b1cf87dbdc9b74"), + BlockHash: ethtypes.MustNewHexBytes0xPrefix(blockHash998), + Address: ethtypes.MustNewAddress("0xc89E46EEED41b777ca6625d37E1Cc87C5c037828"), + Topics: []ethtypes.HexBytes0xPrefix{ + ethtypes.MustNewHexBytes0xPrefix("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"), + ethtypes.MustNewHexBytes0xPrefix("0x0000000000000000000000003968ef051b422d3d1cdc182a88bba8dd922e6fa4"), + ethtypes.MustNewHexBytes0xPrefix("0x000000000000000000000000d0f2f5103fd050739a9fb567251bc460cc24d091"), + }, + Data: ethtypes.MustNewHexBytes0xPrefix("0x00000000000000000000000000000000000000000000000000000000000003e8"), + }} + }) + + // The enrichment fails on the first cycle - nothing must be delivered or recorded as + // delivered, so the retry re-detects the same event + mbl.On("GetBlockInfoByHash", mock.Anything, blockHash998).Return(nil, fmt.Errorf("pop")).Once() + mbl.On("GetBlockInfoByHash", mock.Anything, blockHash998).Return(ðrpc.BlockInfoJSONRPC{ + Number: ethtypes.HexUint64(998), + Hash: ethtypes.MustNewHexBytes0xPrefix(blockHash998), + Timestamp: ethtypes.HexUint64(1700000000), + }, nil) + + loopDone := make(chan bool, 1) + go func() { + loopDone <- es.leadGroupSteadyStateGetLogs() + }() + + select { + case ev := <-events: + assert.Equal(t, blockHash998, ev.Event.ID.BlockHash) + assert.Equal(t, fftypes.FFuint64(998), ev.Event.ID.BlockNumber) + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for the event after the enrich retry") + } + + cancelCtx() + assert.True(t, <-loopDone) +} + +func TestLeadGroupGetLogsFullModeNoCatchupOscillation(t *testing.T) { + + es, _, mRPC, mbl, cancelCtx, done := testGetLogsModeStream(t, 905) + defer done() + es.c.catchupThreshold = 90 + + mbl.On("GetHighestBlock", mock.Anything).Return(uint64(1000), true) + mbl.On("SnapshotMonitoredHeadChain").Return([]*ethrpc.BlockInfoJSONRPC{}).Maybe() + + polls := make(chan []int64, 10) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getLogs", mock.Anything).Return(nil).Run(func(args mock.Arguments) { + filter := args[3].(*ethrpc.LogFilterJSONRPC) + polls <- []int64{filter.FromBlock.BigInt().Int64(), filter.ToBlock.BigInt().Int64()} + *args[1].(*[]*ethrpc.LogJSONRPC) = []*ethrpc.LogJSONRPC{} + }) + + loopDone := make(chan bool, 1) + go func() { + loopDone <- es.leadGroupSteadyStateGetLogs() + }() + + // The raw gap to the head (1000-905=95) is over the catchup threshold (90), but catchup only + // polls to the stability horizon (950), which we are within the threshold of (45) - so it + // would exit straight back to us without polling. We must measure the reversion check the + // same way and stay in steady state (still polling to the head - full mode delivers the + // unstable window with hash tracking), or the two loops would bounce control forever + assert.Equal(t, []int64{905, 914}, <-polls) + + cancelCtx() + assert.True(t, <-loopDone) +} + +func TestLeadGroupCatchupCapsAtStableHead(t *testing.T) { + + es, l, mRPC, mbl, _, done := testGetLogsModeStream(t, 900) + defer done() + es.c.catchupThreshold = 90 + es.c.catchupPageSize = 500 + es.c.checkpointBlockGap = 6 + + mbl.On("GetHighestBlock", mock.Anything).Return(uint64(1000), true) + + polls := make(chan []int64, 10) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getLogs", mock.Anything).Return(nil).Run(func(args mock.Arguments) { + filter := args[3].(*ethrpc.LogFilterJSONRPC) + polls <- []int64{filter.FromBlock.BigInt().Int64(), filter.ToBlock.BigInt().Int64()} + *args[1].(*[]*ethrpc.LogJSONRPC) = []*ethrpc.LogJSONRPC{} + }) + + // The gap to the pollable head (994-900=94) is over the catchup threshold (90), so catchup + // runs one page - but that page is capped at head-checkpointBlockGap (994) rather than + // running to fromBlock+catchupPageSize-1 (1399), so the unstable window is left for the + // steady state loop to deliver, and the HWM (994+1) follows the poll position exactly + exited := es.leadGroupCatchup() + assert.False(t, exited) + assert.Equal(t, []int64{900, 994}, <-polls) + assert.Equal(t, int64(995), l.getHWMBlock()) +} + +func TestLeadGroupCatchupCaughtUpToStableHead(t *testing.T) { + + es, _, _, mbl, _, done := testGetLogsModeStream(t, 980) + defer done() + es.c.catchupThreshold = 20 + es.c.checkpointBlockGap = 6 + + mbl.On("GetHighestBlock", mock.Anything).Return(uint64(1000), true) + + // The raw gap to the head (20) is at the catchup threshold, but the gap to the pollable head + // (994-980=14) is below it - so catchup exits back to steady state without polling at all + // (no eth_getLogs expectation is registered - a poll fails the test) + exited := es.leadGroupCatchup() + assert.False(t, exited) +} + +func TestLeadGroupCatchupHeadBelowGap(t *testing.T) { + + es, _, _, mbl, _, done := testGetLogsModeStream(t, 0) + defer done() + es.c.catchupThreshold = 1 + es.c.checkpointBlockGap = 6 + + mbl.On("GetHighestBlock", mock.Anything).Return(uint64(5), true) + + // The whole chain (head 5) is within checkpointBlockGap (6), so the pollable head + // clamps at genesis and catchup exits to steady state without polling + exited := es.leadGroupCatchup() + assert.False(t, exited) +} + +func TestLeadGroupCatchupGapLargerThanThresholdExitsToSteadyState(t *testing.T) { + + es, _, _, mbl, _, done := testGetLogsModeStream(t, 960) + defer done() + es.c.catchupThreshold = 30 + + mbl.On("GetHighestBlock", mock.Anything).Return(uint64(1000), true) + + // Degenerate config: checkpointBlockGap (50) is larger than catchupThreshold (30), so from + // hwm 960 the pollable head (950) is already behind us. Catchup must hand straight over to + // the steady state loop (which measures its reversion check against the same pollable head, + // so control cannot bounce back) rather than polling or spinning here (no eth_getLogs + // expectation is registered - a poll fails the test) + exited := es.leadGroupCatchup() + assert.False(t, exited) +} diff --git a/internal/msgs/en_config_descriptions.go b/internal/msgs/en_config_descriptions.go index 5269fc8..05f0606 100644 --- a/internal/msgs/en_config_descriptions.go +++ b/internal/msgs/en_config_descriptions.go @@ -46,8 +46,9 @@ var ( _ = ffc("config.connector.events.catchupPageSize", "Number of blocks to query per poll when catching up to the head of the blockchain", i18n.IntType) _ = ffc("config.connector.events.catchupThreshold", "How many blocks behind the chain head an event stream or listener must be on startup, to enter catchup mode", i18n.IntType) _ = ffc("config.connector.events.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") - _ = ffc("config.connector.events.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.", i18n.IntType) + _ = ffc("config.connector.events.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)", i18n.IntType) _ = ffc("config.connector.events.filterPollingInterval", "The interval between polling calls to a filter, when checking for newly arrived events", i18n.TimeDurationType) + _ = ffc("config.connector.events.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") _ = ffc("config.connector.txCacheSize", "Maximum of transactions to hold in the transaction info cache", i18n.IntType) _ = ffc("config.connector.maxConcurrentRequests", "Maximum of concurrent requests to be submitted to the blockchain", i18n.IntType) _ = ffc("config.connector.hederaCompatibilityMode", "Compatibility mode for Hedera, allowing non-standard block header hashes to be processed", i18n.BooleanType) diff --git a/internal/msgs/en_error_messages.go b/internal/msgs/en_error_messages.go index ac114b7..c254f7a 100644 --- a/internal/msgs/en_error_messages.go +++ b/internal/msgs/en_error_messages.go @@ -95,4 +95,5 @@ var ( MsgInvalidRPCRoutingMode = ffe("FF23075", "Invalid JSON/RPC routing mode '%s': must be 'http', 'ws', 'auto' or 'legacy'") MsgWebSocketNotConfigured = ffe("FF23076", "A WebSocket connection is not configured") MsgRPCClientClosed = ffe("FF23077", "The JSON/RPC client is closed") + MsgInvalidFilterPollingMode = ffe("FF23078", "Invalid filter polling mode '%s': must be 'server' or 'client'") ) diff --git a/pkg/ethblocklistener/blocklistener.go b/pkg/ethblocklistener/blocklistener.go index f5f24d7..16c1089 100644 --- a/pkg/ethblocklistener/blocklistener.go +++ b/pkg/ethblocklistener/blocklistener.go @@ -390,13 +390,14 @@ func (bl *blockListener) listenLoop() { continue } // In light mode there is no canonical chain being built, so the head we dispatch to - // consumers is what we report as the canonical height + // consumers is what we report as the canonical height - both through GetHeadBlockNumber + // (used by FFTM's head-number confirmation checks) and GetHighestBlock (used by event streams) if head == bl.currentChainHead { failCount = 0 continue } bl.currentChainHead = head - bl.setBlockHeightMetric(metricCanonicalBlockHeight, bl.currentChainHead) + bl.setHighestBlock(head) update := &ffcapi.BlockHashEvent{GapPotential: false, Created: fftypes.Now(), HeadBlockNumber: bl.currentChainHead} bl.consumerMux.Lock() consumers := make([]*BlockUpdateConsumer, 0, len(bl.consumers)) diff --git a/pkg/ethblocklistener/blocklistener_test.go b/pkg/ethblocklistener/blocklistener_test.go index 2aae1d8..6eb4b45 100644 --- a/pkg/ethblocklistener/blocklistener_test.go +++ b/pkg/ethblocklistener/blocklistener_test.go @@ -1262,6 +1262,9 @@ func TestBlockListenerHeadBlockNumber_DispatchesAndSkipsDuplicateHead(t *testing <-bl.listenLoopDone assert.Equal(t, uint64(1001), bl.currentChainHead) + // GetHighestBlock must track the live head in light mode too (not freeze at the startup height), + // as it drives the poll position of event streams + assert.Equal(t, uint64(1001), bl.highestBlock) mRPC.AssertExpectations(t) }