From cf8c4c191a3e6acafd92d7ac1d3dd80bb8071303 Mon Sep 17 00:00:00 2001 From: Peter Kieltyka Date: Wed, 30 Sep 2026 13:23:47 -0400 Subject: [PATCH 1/5] ethmonitor: parallel block prefetch for fast chains The monitor fetches each block and its logs serially, which caps ingestion at roughly 1 / (2 x node latency) blocks per second. On chains producing blocks faster than that (robinhood mainnet at ~10 blocks/s) the monitor falls behind and never recovers: node-gateway on dev was serving a head ~14 minutes stale. Add an opt-in prefetcher (Options.PrefetchConcurrency, default 0 = off, and Options.PrefetchWindow, default 4x concurrency). While the monitor trails the head, N workers fetch the blocks past its next block, and their logs, into the cache under the existing cache keys. The serial run loop is unchanged and still the only place the canonical chain is built; it just reads prefetched payloads as cache hits. At the head the prefetcher is idle, and in polling mode it only polls the head while the run loop is catching up, so slow chains see no extra node calls. Reorg safety: - with prefetch on, a block that does not extend the head is confirmed with a direct, uncached node fetch before it is treated as a reorg - a reorg resets the prefetcher and drops its cached blocks above the reverted block; a generation counter discards in-flight stale results Also, affecting all monitors: - block payloads are decoded before caching, and undecodable cache entries are dropped, so one bad node response can't poison a block number until cache expiry - Run owns a child context and joins its publisher, head listeners and prefetcher before returning, so goroutines don't outlive a fatal exit and a later Run - fetchNextBlock reads nextBlockNumber under its lock Co-Authored-By: Claude Opus 5.5 (1M context) --- ethmonitor/ethmonitor.go | 244 ++++++-- ethmonitor/prefetch.go | 286 ++++++++++ ethmonitor/prefetch_internal_test.go | 61 ++ ethmonitor/prefetch_test.go | 812 +++++++++++++++++++++++++++ 4 files changed, 1360 insertions(+), 43 deletions(-) create mode 100644 ethmonitor/prefetch.go create mode 100644 ethmonitor/prefetch_internal_test.go create mode 100644 ethmonitor/prefetch_test.go diff --git a/ethmonitor/ethmonitor.go b/ethmonitor/ethmonitor.go index 8f7bd916..c1c70b84 100644 --- a/ethmonitor/ethmonitor.go +++ b/ethmonitor/ethmonitor.go @@ -19,6 +19,7 @@ import ( "github.com/0xsequence/ethkit/go-ethereum/core/types" "github.com/0xsequence/ethkit/util" "github.com/goware/breaker" + memcache "github.com/goware/cachestore-mem" cachestore "github.com/goware/cachestore2" "github.com/goware/channel" "github.com/goware/superr" @@ -41,6 +42,8 @@ var DefaultOptions = Options{ LogTopics: []common.Hash{}, // all logs DebugLogging: false, CacheExpiry: 600 * time.Second, + PrefetchConcurrency: 0, // Prefetching is disabled by default + PrefetchWindow: 0, // 4x PrefetchConcurrency set on init Alerter: util.NoopAlerter(), } @@ -104,6 +107,18 @@ type Options struct { // CacheExpiry is how long to keep each record in cache CacheExpiry time.Duration + // PrefetchConcurrency is the number of workers fetching blocks (and their + // logs, when WithLogs is set) ahead of the monitor while it trails the chain + // head. Prefetched payloads land in the cache, so the monitor's serial loop + // reads them as cache hits instead of paying node round-trips per block. + // Useful on chains whose block rate outpaces a serial fetch. If no + // CacheBackend is set, an in-memory cache is used. 0 disables prefetching. + PrefetchConcurrency int + + // PrefetchWindow is how many blocks past the monitor's next block the + // prefetcher may fetch. Defaults to 4x PrefetchConcurrency. + PrefetchWindow int + // Alerter config via github.com/goware/alerter Alerter util.Alerter @@ -135,7 +150,16 @@ type Monitor struct { pollInterval atomic.Int64 isStreamingMode atomic.Bool - cache cachestore.Store[[]byte] + // hitStreak counts the run loop's consecutive fetches which found the + // next block without a miss, capped to avoid overflow. + hitStreak atomic.Int32 + + // latestHead is the most recent chain head number seen, from the + // newHeads stream or, in polling mode, the prefetcher's head poll. + latestHead atomic.Uint64 + + cache cachestore.Store[[]byte] + prefetch *prefetcher publishCh chan Blocks publishQueue *queue @@ -172,6 +196,25 @@ func NewMonitor(provider ethrpc.RawInterface, options ...Options) (*Monitor, err // with slog, we can't modify the log level after the logger is created. } + if opts.PrefetchConcurrency < 0 { + opts.PrefetchConcurrency = 0 + } + if opts.PrefetchConcurrency > 0 && opts.PrefetchWindow <= 0 { + opts.PrefetchWindow = 4 * opts.PrefetchConcurrency + } + + // prefetching hands payloads to the monitor loop through the cache, + // so it needs one even when the caller didn't configure a backend. + if opts.PrefetchConcurrency > 0 && opts.CacheBackend == nil { + // room for a block-by-number and a logs entry per block in the + // window, plus headroom for by-hash lookups during reorgs. + backend, err := memcache.NewBackend(uint32(4*opts.PrefetchWindow + 256)) + if err != nil { + return nil, fmt.Errorf("ethmonitor: creating prefetch cache: %w", err) + } + opts.CacheBackend = backend + } + var cache cachestore.Store[[]byte] if opts.CacheBackend != nil { if opts.CacheExpiry == 0 { @@ -180,7 +223,7 @@ func NewMonitor(provider ethrpc.RawInterface, options ...Options) (*Monitor, err cache = cachestore.OpenStore[[]byte](opts.CacheBackend, cachestore.WithDefaultKeyExpiry(opts.CacheExpiry)) } - return &Monitor{ + m := &Monitor{ options: opts, log: opts.Logger, alert: opts.Alerter, @@ -191,7 +234,11 @@ func NewMonitor(provider ethrpc.RawInterface, options ...Options) (*Monitor, err publishCh: make(chan Blocks), publishQueue: newQueue(opts.BlockRetentionLimit * 2), subscribers: make([]*subscriber, 0), - }, nil + } + if opts.PrefetchConcurrency > 0 { + m.prefetch = newPrefetcher(m, opts.PrefetchConcurrency, opts.PrefetchWindow) + } + return m, nil } func (m *Monitor) lazyInit(ctx context.Context) error { @@ -221,10 +268,19 @@ func (m *Monitor) Run(ctx context.Context) error { return fmt.Errorf("ethmonitor: already running") } - m.ctx, m.ctxStop = context.WithCancel(ctx) + ctx, cancel := context.WithCancel(ctx) + m.ctx, m.ctxStop = ctx, cancel atomic.StoreInt32(&m.running, 1) defer atomic.StoreInt32(&m.running, 0) + var wg sync.WaitGroup + defer func() { + cancel() + wg.Wait() + }() + m.hitStreak.Store(0) + m.latestHead.Store(0) + m.isStreamingMode.Store(false) if err := m.lazyInit(ctx); err != nil { return err @@ -265,7 +321,9 @@ func (m *Monitor) Run(ctx context.Context) error { } // Broadcast published events to all subscribers + wg.Add(1) go func() { + defer wg.Done() defer func() { if r := recover(); r != nil { m.log.Error(fmt.Sprintf("ethmonitor: panic in publish loop: %v - stack: %s", r, string(debug.Stack()))) @@ -289,7 +347,7 @@ func (m *Monitor) Run(ctx context.Context) error { }() // Monitor the chain for canonical representation - err := m.monitor() + err := m.monitor(ctx, &wg) if m.options.UnsubscribeOnStop { m.UnsubscribeAll(err) } @@ -325,13 +383,33 @@ func (m *Monitor) IsStreamingEnabled() bool { func (m *Monitor) IsStreamingMode() bool { return m.isStreamingMode.Load() } -func (m *Monitor) listenNewHead() <-chan uint64 { + +// nextBlockNum returns the number of the next block the run loop will fetch, +// or false while the monitor has not settled on one yet. +func (m *Monitor) nextBlockNum() (uint64, bool) { + m.nextBlockNumberMu.Lock() + defer m.nextBlockNumberMu.Unlock() + if m.nextBlockNumber == nil || !m.nextBlockNumber.IsUint64() { + return 0, false + } + return m.nextBlockNumber.Uint64(), true +} + +// isCatchingUp reports whether the run loop found its next block on at least +// its last two fetches in a row, ie. the chain has been ahead of it. +func (m *Monitor) isCatchingUp() bool { + return m.hitStreak.Load() >= 2 +} + +func (m *Monitor) listenNewHead(ctx context.Context, wg *sync.WaitGroup) <-chan uint64 { ch := make(chan uint64) var latestHeadBlock atomic.Uint64 nextBlock := make(chan uint64) + wg.Add(2) go func() { + defer wg.Done() defer func() { if r := recover(); r != nil { m.log.Error(fmt.Sprintf("ethmonitor: panic in new head loop: %v - stack: %s", r, string(debug.Stack()))) @@ -376,18 +454,23 @@ func (m *Monitor) listenNewHead() <-chan uint64 { m.isStreamingMode.Store(true) newHeads := make(chan *types.Header) - sub, err := m.provider.SubscribeNewHeads(m.ctx, newHeads) + sub, err := m.provider.SubscribeNewHeads(ctx, newHeads) if err != nil { m.log.Warn(fmt.Sprintf("ethmonitor (chain %s): websocket connect failed: %v", m.chainID.String(), err)) m.alert.Alert(context.Background(), "ethmonitor (chain %s): websocket connect failed: %v", m.chainID.String(), err) - time.Sleep(2000 * time.Millisecond) + select { + case <-ctx.Done(): + close(nextBlock) + return + case <-time.After(2 * time.Second): + } streamingErrLastTime = time.Now() goto reconnect } for { select { - case <-m.ctx.Done(): + case <-ctx.Done(): // if we're done, we'll unsubscribe and close the nextBlock channel sub.Unsubscribe() close(nextBlock) @@ -404,6 +487,10 @@ func (m *Monitor) listenNewHead() <-chan uint64 { case newHead := <-newHeads: latestHeadBlock.Store(newHead.Number.Uint64()) + m.latestHead.Store(newHead.Number.Uint64()) + if m.prefetch != nil { + m.prefetch.notify() + } select { case nextBlock <- newHead.Number.Uint64(): default: @@ -434,7 +521,7 @@ func (m *Monitor) listenNewHead() <-chan uint64 { // Polling mode, where we poll for the latest block number select { - case <-m.ctx.Done(): + case <-ctx.Done(): // if we're done, we'll close the nextBlock channel close(nextBlock) retryStreamingTimer.Stop() @@ -443,7 +530,7 @@ func (m *Monitor) listenNewHead() <-chan uint64 { case <-time.After(time.Duration(m.pollInterval.Load())): select { case nextBlock <- 0: - case <-m.ctx.Done(): + case <-ctx.Done(): } } } @@ -452,6 +539,7 @@ func (m *Monitor) listenNewHead() <-chan uint64 { // The main loop which notifies the monitor to continue to the next block go func() { + defer wg.Done() defer func() { if r := recover(); r != nil { m.log.Error(fmt.Sprintf("ethmonitor: panic in next block loop: %v - stack: %s", r, string(debug.Stack()))) @@ -461,7 +549,7 @@ func (m *Monitor) listenNewHead() <-chan uint64 { for { select { - case <-m.ctx.Done(): + case <-ctx.Done(): return default: } @@ -479,20 +567,23 @@ func (m *Monitor) listenNewHead() <-chan uint64 { // waiting on the nextBlock channel select { case ch <- nextBlockNumber: - case <-m.ctx.Done(): + case <-ctx.Done(): return } continue } else { // wait for the next block select { - case <-nextBlock: - case <-m.ctx.Done(): + case _, ok := <-nextBlock: + if !ok { + return + } + case <-ctx.Done(): return } select { case ch <- latestBlockNum: - case <-m.ctx.Done(): + case <-ctx.Done(): return } } @@ -502,8 +593,7 @@ func (m *Monitor) listenNewHead() <-chan uint64 { return ch } -func (m *Monitor) monitor() error { - ctx := m.ctx +func (m *Monitor) monitor(ctx context.Context, wg *sync.WaitGroup) error { events := Blocks{} // minLoopInterval is time we monitor between cycles. It's a fast @@ -512,13 +602,22 @@ func (m *Monitor) monitor() error { minLoopInterval := 5 * time.Millisecond // listen for new heads either via streaming or polling - listenNewHead := m.listenNewHead() + listenNewHead := m.listenNewHead(ctx, wg) + + // fetch blocks ahead of the run loop into the cache while we trail the head + if m.prefetch != nil { + wg.Add(1) + go func() { + defer wg.Done() + m.prefetch.run(ctx) + }() + } // monitor run loop for { select { - case <-m.ctx.Done(): + case <-ctx.Done(): return nil case newHeadNum := <-listenNewHead: @@ -539,8 +638,17 @@ func (m *Monitor) monitor() error { } // fetch the next block, either via the stream or via a poll - nextBlock, nextBlockPayload, miss, err := m.fetchNextBlock(ctx) + nextBlock, nextBlockPayload, miss, err := m.fetchNextBlock(ctx, false) + + // with prefetching, a cached block may be a stale entry from an abandoned + // fork, written by a prefetch worker of this or a peer monitor. confirm a + // parent mismatch with the node before treating it as a reorg. + if err == nil && m.prefetch != nil && headBlock != nil && nextBlock.ParentHash() != headBlock.Hash() { + nextBlock, nextBlockPayload, miss, err = m.refetchNextBlock(ctx, nextBlock) + } + if err != nil { + m.hitStreak.Store(0) if errors.Is(err, context.DeadlineExceeded) { m.log.Info(fmt.Sprintf("ethmonitor: fetchNextBlock timed out: '%v', for blockNum:%v, retrying..", err, m.nextBlockNumber)) } else { @@ -556,8 +664,12 @@ func (m *Monitor) monitor() error { // we speed up the polling interval if miss { m.pollInterval.Store(int64(m.options.PollingInterval)) + m.hitStreak.Store(0) } else { m.pollInterval.Store(int64(clampDuration(minLoopInterval, time.Duration(m.pollInterval.Load())/4))) + if m.hitStreak.Load() < 1<<20 { + m.hitStreak.Add(1) + } } // build deterministic set of add/remove events which construct the canonical chain @@ -593,6 +705,11 @@ func (m *Monitor) monitor() error { // clear events sink events = Blocks{} + + // the monitor advanced, which opens up room in the prefetch window + if m.prefetch != nil { + m.prefetch.notify() + } } } } @@ -633,6 +750,11 @@ func (m *Monitor) buildCanonicalChain(ctx context.Context, nextBlock *types.Bloc } } + // blocks prefetched above the popped block may be from the abandoned fork + if m.prefetch != nil { + m.prefetch.reset(ctx, poppedBlock.NumberU64()) + } + if m.options.DebugLogging { m.log.Debug(fmt.Sprintf("ethmonitor: block reorg, reverting block #%d hash:%s prevHash:%s", poppedBlock.NumberU64(), poppedBlock.Hash().Hex(), poppedBlock.ParentHash().Hex())) } @@ -691,12 +813,7 @@ func (m *Monitor) addLogs(ctx context.Context, blocks Blocks) { blockHash := block.Hash() - topics := [][]common.Hash{} - if len(m.options.LogTopics) > 0 { - topics = append(topics, m.options.LogTopics) - } - - logs, _, err := m.filterLogs(tctx, blockHash, topics, block.Bloom()) + logs, _, err := m.filterLogs(tctx, blockHash, m.logTopics(), block.Bloom()) if err == nil { // check the logsBloom from the block to check if we should be expecting logs. logsBloom @@ -723,6 +840,16 @@ func (m *Monitor) addLogs(ctx context.Context, blocks Blocks) { } } +// logTopics returns the topic filter for block log queries, which is also +// part of the logs cache key. +func (m *Monitor) logTopics() [][]common.Hash { + topics := [][]common.Hash{} + if len(m.options.LogTopics) > 0 { + topics = append(topics, m.options.LogTopics) + } + return topics +} + func (m *Monitor) filterLogs(ctx context.Context, blockHash common.Hash, topics [][]common.Hash, blockBloom types.Bloom) ([]types.Log, []byte, error) { getter := func(ctx context.Context, _ string) ([]byte, error) { if m.options.DebugLogging { @@ -806,12 +933,20 @@ func (m *Monitor) backfillChainLogs(ctx context.Context, newBlocks Blocks) { } } -func (m *Monitor) fetchNextBlock(ctx context.Context) (*types.Block, []byte, bool, error) { +func (m *Monitor) fetchNextBlock(ctx context.Context, bypassCache bool) (*types.Block, []byte, bool, error) { miss := false + var fetchedBlock *types.Block + + var nextBlockNumber *big.Int + m.nextBlockNumberMu.Lock() + if m.nextBlockNumber != nil { + nextBlockNumber = big.NewInt(0).Set(m.nextBlockNumber) + } + m.nextBlockNumberMu.Unlock() getter := func(ctx context.Context, _ string) ([]byte, error) { if m.options.DebugLogging { - m.log.Debug(fmt.Sprintf("ethmonitor: fetchNextBlock is calling origin for number %s", m.nextBlockNumber)) + m.log.Debug(fmt.Sprintf("ethmonitor: fetchNextBlock is calling origin for number %s", nextBlockNumber)) } for { select { @@ -820,10 +955,11 @@ func (m *Monitor) fetchNextBlock(ctx context.Context) (*types.Block, []byte, boo default: } - nextBlockPayload, err := m.fetchRawBlockByNumber(ctx, m.nextBlockNumber) + nextBlockPayload, err := m.fetchRawBlockByNumber(ctx, nextBlockNumber) if err != nil { - m.log.Debug(fmt.Sprintf("ethmonitor: [retrying] failed to fetch next block # %d, due to: %v", m.nextBlockNumber, err)) + m.log.Debug(fmt.Sprintf("ethmonitor: [retrying] failed to fetch next block # %d, due to: %v", nextBlockNumber, err)) miss = true + m.hitStreak.Store(0) if m.IsStreamingMode() { // in streaming mode, we'll use a shorter time to pause before we refetch time.Sleep(200 * time.Millisecond) @@ -833,25 +969,23 @@ func (m *Monitor) fetchNextBlock(ctx context.Context) (*types.Block, []byte, boo continue } + // Validate before caching so a bad node response cannot poison this + // block number until cache expiry. Let the run loop retry decode errors. + fetchedBlock, err = m.unmarshalBlock(nextBlockPayload) + if err != nil { + return nil, err + } return nextBlockPayload, nil } } - var nextBlockNumber *big.Int - m.nextBlockNumberMu.Lock() - if m.nextBlockNumber != nil { - nextBlockNumber = big.NewInt(0).Set(m.nextBlockNumber) - } - m.nextBlockNumberMu.Unlock() - - // skip cache if isn't provided, or in case when nextBlockNumber is nil (latest) - if m.cache == nil || nextBlockNumber == nil { + // Reorg confirmation must bypass cache reads and any in-flight cache getter. + if bypassCache || m.cache == nil || nextBlockNumber == nil { resp, err := getter(ctx, "") if err != nil { return nil, resp, miss, err } - block, err := m.unmarshalBlock(resp) - return block, resp, miss, err + return fetchedBlock, resp, miss, nil } // fetch with distributed mutex @@ -860,10 +994,32 @@ func (m *Monitor) fetchNextBlock(ctx context.Context) (*types.Block, []byte, boo if err != nil { return nil, resp, miss, err } + if fetchedBlock != nil { + return fetchedBlock, resp, miss, nil + } block, err := m.unmarshalBlock(resp) + if err != nil { + // A peer or an older monitor may have cached an invalid payload. + if deleteErr := m.cache.Delete(ctx, key); deleteErr != nil { + m.log.Warn(fmt.Sprintf("ethmonitor: error deleting invalid block cache for block num %s due to: '%v'", nextBlockNumber, deleteErr)) + } + } return block, resp, miss, err } +// refetchNextBlock drops the cached by-number entry for block and confirms the +// next block directly with the node, even if a peer repopulates the cache. +func (m *Monitor) refetchNextBlock(ctx context.Context, block *types.Block) (*types.Block, []byte, bool, error) { + key := CacheKeyBlockByNumber(m.chainID, block.Number()) + if err := m.cache.Delete(ctx, key); err != nil { + m.log.Warn(fmt.Sprintf("ethmonitor: error deleting block cache for block num %d due to: '%v'", block.NumberU64(), err)) + } + if m.options.DebugLogging { + m.log.Debug(fmt.Sprintf("ethmonitor: block #%d hash:%s does not extend head, refetching from origin", block.NumberU64(), block.Hash().Hex())) + } + return m.fetchNextBlock(ctx, true) +} + func CacheKeyBlockByNumber(chainID *big.Int, num *big.Int) string { return fmt.Sprintf("ethmonitor:%s:BlockNum:%s", chainID.String(), num.String()) } @@ -913,6 +1069,8 @@ func (m *Monitor) fetchRawBlockByNumber(ctx context.Context, num *big.Int) ([]by cancel() if err != nil { + // Clear the catch-up signal before retries, which may wait at the head. + m.hitStreak.Store(0) if errors.Is(err, ethereum.NotFound) { return nil, ethereum.NotFound } else { @@ -1025,7 +1183,7 @@ func (m *Monitor) publish(ctx context.Context, events Blocks) error { if ok { select { case m.publishCh <- pubEvents: - case <-m.ctx.Done(): + case <-ctx.Done(): } } diff --git a/ethmonitor/prefetch.go b/ethmonitor/prefetch.go new file mode 100644 index 00000000..9b6c0565 --- /dev/null +++ b/ethmonitor/prefetch.go @@ -0,0 +1,286 @@ +package ethmonitor + +import ( + "context" + "fmt" + "math/big" + "runtime/debug" + "sync" + "sync/atomic" + "time" + + "github.com/0xsequence/ethkit/go-ethereum" + "github.com/0xsequence/ethkit/go-ethereum/core/types" +) + +// prefetchTick is how often the prefetcher reconsiders its window when +// nothing else wakes it up. +const prefetchTick = 25 * time.Millisecond + +// prefetcher fetches blocks, and their logs when the monitor runs WithLogs, +// ahead of the monitor's run loop while the monitor trails the chain head. +// +// It only fills the cache, using the same keys the run loop reads, so the run +// loop stays the single place the canonical chain is built and validated; +// prefetching only changes where a payload comes from. When the monitor is at +// the head there is nothing past its next block to fetch, and the prefetcher +// is idle. +type prefetcher struct { + m *Monitor + concurrency int + window uint64 + + jobs chan prefetchJob + wake chan struct{} + + // gen is bumped on every reset, so a job scheduled before a reorg can + // tell its payload may belong to the abandoned fork. + gen atomic.Uint64 + + mu sync.Mutex + cursor uint64 // next block number to schedule + highWater uint64 // highest block number scheduled since the last reset +} + +type prefetchJob struct { + num uint64 + gen uint64 +} + +func newPrefetcher(m *Monitor, concurrency, window int) *prefetcher { + return &prefetcher{ + m: m, + concurrency: concurrency, + window: uint64(window), + jobs: make(chan prefetchJob, concurrency), + wake: make(chan struct{}, 1), + } +} + +// notify asks the prefetcher to reconsider its window, without blocking. +func (p *prefetcher) notify() { + select { + case p.wake <- struct{}{}: + default: + } +} + +// run schedules prefetch jobs until ctx is done, and returns once all of its +// workers have exited. +func (p *prefetcher) run(ctx context.Context) { + defer func() { + if r := recover(); r != nil { + p.m.log.Error(fmt.Sprintf("ethmonitor: panic in prefetch loop: %v - stack: %s", r, string(debug.Stack()))) + p.m.alert.Alert(context.Background(), "ethmonitor: panic in prefetch loop: %v", r) + } + }() + + // start from a clean slate, as Run may be called again after a failure + p.mu.Lock() + p.gen.Add(1) + p.cursor, p.highWater = 0, 0 + p.mu.Unlock() + + ctx, cancel := context.WithCancel(ctx) + var wg sync.WaitGroup + defer func() { + cancel() + wg.Wait() + }() + for i := 0; i < p.concurrency; i++ { + wg.Add(1) + go func() { + defer wg.Done() + p.worker(ctx) + }() + } + + ticker := time.NewTicker(prefetchTick) + defer ticker.Stop() + + var lastHeadPoll time.Time + + for { + select { + case <-ctx.Done(): + return + case <-p.wake: + case <-ticker.C: + } + + // in streaming mode the newHeads stream keeps latestHead current, + // in polling mode nothing does, so we ask the node for it. but only + // while the monitor is catching up: at the head there is nothing to + // prefetch, and on a slow chain we'd just double the polling load. + if !p.m.IsStreamingMode() && p.m.isCatchingUp() && time.Since(lastHeadPoll) >= p.m.options.PollingInterval { + lastHeadPoll = time.Now() + p.pollHead(ctx) + } + + p.schedule() + } +} + +func (p *prefetcher) pollHead(ctx context.Context) { + tctx, cancel := context.WithTimeout(ctx, 4*time.Second) + defer cancel() + + head, err := p.m.provider.BlockNumber(tctx) + if err != nil { + p.m.log.Debug(fmt.Sprintf("ethmonitor: prefetch failed to poll head block number: %v", err)) + return + } + p.m.latestHead.Store(head) +} + +// schedule hands the workers the block numbers in the window past the +// monitor's next block, as far as they have room for. +func (p *prefetcher) schedule() { + next, ok := p.m.nextBlockNum() + if !ok { + return + } + + // The run loop fetches `next` itself, so we only look past it. We also + // leave the newest head block to the run loop: it is the one most likely + // not yet served by the node, and the run loop already retries it. + head := p.m.latestHead.Load() + if head < next+2 { + return + } + hi := min(head-1, next+p.window) + + p.mu.Lock() + defer p.mu.Unlock() + + gen := p.gen.Load() + for n := max(p.cursor, next+1); n <= hi; n++ { + select { + case p.jobs <- prefetchJob{num: n, gen: gen}: + p.cursor = n + 1 + p.highWater = max(p.highWater, n) + default: + // workers are busy, carry on from the cursor next time + return + } + } +} + +// reset is called when the run loop pops the block fromNum off the canonical +// chain during a reorg. Any block prefetched above it may belong to the +// abandoned fork, so their by-number cache entries are dropped and scheduling +// resumes from fromNum. +// +// Entries a worker writes after the reset are dropped by the worker itself, +// and any stale entry that slips through is caught by the run loop, which +// confirms a block that does not extend its head with the node before +// treating it as a reorg. +func (p *prefetcher) reset(ctx context.Context, fromNum uint64) { + p.mu.Lock() + p.gen.Add(1) + hi := p.highWater + p.cursor = fromNum + p.highWater = fromNum + p.mu.Unlock() + + for n := fromNum + 1; n <= hi; n++ { + key := CacheKeyBlockByNumber(p.m.chainID, new(big.Int).SetUint64(n)) + if err := p.m.cache.Delete(ctx, key); err != nil { + p.m.log.Warn(fmt.Sprintf("ethmonitor: error deleting prefetched block cache for block num %d due to: '%v'", n, err)) + } + } +} + +func (p *prefetcher) worker(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + case job := <-p.jobs: + p.fetch(ctx, job) + p.notify() + } + } +} + +// fetch loads one block, and its logs, into the cache. It makes a single +// attempt: on any failure the run loop fetches the block itself when it gets +// there. +func (p *prefetcher) fetch(ctx context.Context, job prefetchJob) { + defer func() { + if r := recover(); r != nil { + p.m.log.Error(fmt.Sprintf("ethmonitor: panic in prefetch worker: %v - stack: %s", r, string(debug.Stack()))) + p.m.alert.Alert(context.Background(), "ethmonitor: panic in prefetch worker: %v", r) + } + }() + + m := p.m + if job.gen != p.gen.Load() { + return + } + + num := new(big.Int).SetUint64(job.num) + key := CacheKeyBlockByNumber(m.chainID, num) + var block *types.Block + + // NOTE: the getter must not retry, as it runs while holding the lock for the key + getter := func(ctx context.Context, _ string) ([]byte, error) { + if m.options.DebugLogging { + m.log.Debug(fmt.Sprintf("ethmonitor: prefetch is calling origin for number %d", job.num)) + } + tctx, cancel := context.WithTimeout(ctx, m.options.Timeout) + defer cancel() + + payload, err := m.provider.RawBlockByNumber(tctx, num) + if err != nil { + return nil, err + } + if len(payload) == 0 || string(payload) == "null" { + return nil, ethereum.NotFound + } + // Validate even without logs: the serial loop must never inherit an + // undecodable response from a successful cache write. + block, err = m.unmarshalBlock(payload) + if err != nil { + return nil, err + } + return payload, nil + } + + payload, err := m.cache.GetOrSetWithLockEx(ctx, key, getter, m.options.CacheExpiry) + if err != nil { + m.log.Debug(fmt.Sprintf("ethmonitor: prefetch of block %d failed: %v", job.num, err)) + return + } + + if job.gen != p.gen.Load() { + // a reorg reset ran while we were fetching, so this payload may be + // from the abandoned fork + if err := m.cache.Delete(ctx, key); err != nil { + m.log.Warn(fmt.Sprintf("ethmonitor: error deleting prefetched block cache for block num %d due to: '%v'", job.num, err)) + } + return + } + + if block == nil { + block, err = m.unmarshalBlock(payload) + if err != nil { + m.log.Debug(fmt.Sprintf("ethmonitor: prefetch failed to decode block %d: %v", job.num, err)) + if deleteErr := m.cache.Delete(ctx, key); deleteErr != nil { + m.log.Warn(fmt.Sprintf("ethmonitor: error deleting invalid block cache for block num %d due to: '%v'", job.num, deleteErr)) + } + return + } + } + + if !m.options.WithLogs { + return + } + + // logs are keyed by block hash, so they stay valid across reorgs + tctx, cancel := context.WithTimeout(ctx, m.options.Timeout) + defer cancel() + if _, _, err := m.filterLogs(tctx, block.Hash(), m.logTopics(), block.Bloom()); err != nil { + m.log.Debug(fmt.Sprintf("ethmonitor: prefetch of logs for block %d failed: %v", job.num, err)) + } +} diff --git a/ethmonitor/prefetch_internal_test.go b/ethmonitor/prefetch_internal_test.go new file mode 100644 index 00000000..259ae05e --- /dev/null +++ b/ethmonitor/prefetch_internal_test.go @@ -0,0 +1,61 @@ +package ethmonitor + +import ( + "context" + "encoding/json" + "fmt" + "math/big" + "testing" + "time" + + "github.com/0xsequence/ethkit/ethmonitor/internal/mocks" + "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" +) + +func TestPrefetchRejectsInvalidBlockBeforeCaching(t *testing.T) { + for _, withLogs := range []bool{false, true} { + t.Run(fmt.Sprintf("withLogs=%v", withLogs), func(t *testing.T) { + provider := mocks.NewMockRawInterface(gomock.NewController(t)) + provider.EXPECT().RawBlockByNumber(gomock.Any(), big.NewInt(1000)). + Return(json.RawMessage(`{"number":"0x3e8"}`), nil) + opts := DefaultOptions + opts.PrefetchConcurrency = 1 + opts.WithLogs = withLogs + monitor, err := NewMonitor(provider, opts) + require.NoError(t, err) + monitor.chainID = big.NewInt(1) + + monitor.prefetch.fetch(context.Background(), prefetchJob{num: 1000}) + key := CacheKeyBlockByNumber(monitor.chainID, big.NewInt(1000)) + _, found, err := monitor.cache.Get(context.Background(), key) + require.NoError(t, err) + require.False(t, found, "worker cached an undecodable block") + }) + } +} + +func TestPrefetchPanicStopsWorkers(t *testing.T) { + provider := mocks.NewMockRawInterface(gomock.NewController(t)) + provider.EXPECT().BlockNumber(gomock.Any()).DoAndReturn(func(context.Context) (uint64, error) { + panic("simulated head poll panic") + }) + opts := DefaultOptions + opts.PrefetchConcurrency = 4 + monitor, err := NewMonitor(provider, opts) + require.NoError(t, err) + monitor.hitStreak.Store(2) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + monitor.prefetch.run(ctx) + }() + defer func() { cancel(); <-done }() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("prefetch panic did not cancel and join its workers") + } +} diff --git a/ethmonitor/prefetch_test.go b/ethmonitor/prefetch_test.go new file mode 100644 index 00000000..51d4216d --- /dev/null +++ b/ethmonitor/prefetch_test.go @@ -0,0 +1,812 @@ +package ethmonitor_test + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "math/big" + "runtime" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/0xsequence/ethkit/ethmonitor" + "github.com/0xsequence/ethkit/ethmonitor/internal/mocks" + "github.com/0xsequence/ethkit/ethrpc" + "github.com/0xsequence/ethkit/go-ethereum" + "github.com/0xsequence/ethkit/go-ethereum/common" + "github.com/0xsequence/ethkit/go-ethereum/core/types" + "github.com/0xsequence/ethkit/go-ethereum/crypto" + memcache "github.com/goware/cachestore-mem" + cachestore "github.com/goware/cachestore2" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" +) + +// TestMonitorPrefetchKeepsUp reproduces a chain producing blocks faster than +// the monitor can fetch them one at a time, and checks prefetching keeps the +// monitor at the head, in both polling and streaming mode. +func TestMonitorPrefetchKeepsUp(t *testing.T) { + if testing.Short() { + t.Skip("Skipping in short mode") + } + + // 100 blocks/s, with 15ms per node call: a serial fetch of a block and + // its logs manages ~30 blocks/s at best. + const blockInterval = 10 * time.Millisecond + const latency = 15 * time.Millisecond + const runFor = 2 * time.Second + + for _, streaming := range []bool{false, true} { + for _, concurrency := range []int{0, 4, 8} { + t.Run(fmt.Sprintf("streaming=%v/prefetch=%d", streaming, concurrency), func(t *testing.T) { + chain := newFakeChain(1000, 1, latency) + monitor := newTestMonitor(t, chain, streaming, concurrency, 0) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go chain.produce(ctx, blockInterval) + go monitor.Run(ctx) + defer monitor.Stop() + + time.Sleep(runFor) + lag := chain.head() - monitor.LatestBlockNum().Uint64() + t.Logf("head:%d monitor:%d lag:%d blocks", chain.head(), monitor.LatestBlockNum().Uint64(), lag) + + if concurrency == 0 { + // the bug: without prefetching, the monitor falls behind + assert.Greater(t, lag, uint64(60)) + } else { + assert.Less(t, lag, uint64(15)) + } + }) + } + } +} + +// TestMonitorPrefetchReorg reorgs the chain while the prefetcher is ahead of +// the monitor, and checks the monitor ends on the new fork with a consistent +// stream of events. +func TestMonitorPrefetchReorg(t *testing.T) { + if testing.Short() { + t.Skip("Skipping in short mode") + } + + // prefetched: the reorg starts at a block the prefetcher already cached + // from the abandoned fork, ahead of the monitor head. otherwise it starts + // below the monitor head, which has already published blocks from it. + // + // the window is kept small, as the monitor pauses 2s for every block it + // reverts, and stale prefetched blocks can deepen a reorg up to the window. + for _, concurrency := range []int{2, 4} { + for _, prefetched := range []bool{true, false} { + t.Run(fmt.Sprintf("prefetch=%d/prefetched=%v", concurrency, prefetched), func(t *testing.T) { + testMonitorPrefetchReorg(t, concurrency, prefetched) + }) + } + } +} + +func testMonitorPrefetchReorg(t *testing.T, concurrency int, prefetched bool) { + chain := newFakeChain(1000, 200, 15*time.Millisecond) + backend, err := memcache.NewBackend(1024) + require.NoError(t, err) + cache := cachestore.OpenStore[[]byte](backend) + monitor := newTestMonitor(t, chain, false, concurrency, 4, backend) + + sub := monitor.Subscribe("TestMonitorPrefetchReorg") + defer sub.Unsubscribe() + events := newEventLog() + go func() { + for blocks := range sub.Blocks() { + events.apply(t, blocks) + } + }() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go monitor.Run(ctx) + defer monitor.Stop() + + require.Eventually(t, func() bool { + return monitor.LatestBlockNum().Uint64() >= 1050 + }, 10*time.Second, time.Millisecond) + + var reorgAt uint64 + if prefetched { + // the monitor takes ~15ms a block here, so a block cached two or + // more past its head is still ahead of it when we reorg + require.Eventually(t, func() bool { + head := monitor.LatestBlockNum().Uint64() + for n := head + 4; n >= head+2; n-- { + key := ethmonitor.CacheKeyBlockByNumber(big.NewInt(1), new(big.Int).SetUint64(n)) + if _, ok, _ := cache.Get(context.Background(), key); ok { + reorgAt = n + return true + } + } + return false + }, 5*time.Second, time.Millisecond) + } else { + reorgAt = monitor.LatestBlockNum().Uint64() - 2 + } + chain.reorgFrom(reorgAt) + t.Logf("reorged chain from block %d, monitor head %d", reorgAt, monitor.LatestBlockNum().Uint64()) + + require.Eventually(t, func() bool { + head := monitor.LatestReadyBlock() + return head != nil && head.Hash() == chain.hashAt(chain.head()) + }, 60*time.Second, 10*time.Millisecond) + + // the retained chain must be the new canonical chain + for _, b := range monitor.Chain().Blocks() { + assert.Equal(t, chain.hashAt(b.NumberU64()), b.Hash(), "block %d", b.NumberU64()) + } + + // and so must the chain subscribers built from the events + require.Eventually(t, func() bool { + return events.head() == chain.hashAt(chain.head()) + }, 5*time.Second, 10*time.Millisecond) + for num, hash := range events.blocks() { + assert.Equal(t, chain.hashAt(num), hash, "subscriber block %d", num) + } +} + +// TestMonitorPrefetchHeadAhead has the node announce heads it cannot serve +// yet, as when a websocket is ahead of the http node. The prefetcher's misses +// must not wedge the monitor. +func TestMonitorPrefetchHeadAhead(t *testing.T) { + if testing.Short() { + t.Skip("Skipping in short mode") + } + + chain := newFakeChain(1000, 1, 5*time.Millisecond) + chain.announceAhead = 5 + monitor := newTestMonitor(t, chain, true, 4, 0) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go chain.produce(ctx, 10*time.Millisecond) + go monitor.Run(ctx) + defer monitor.Stop() + + time.Sleep(1500 * time.Millisecond) + assert.Greater(t, monitor.LatestBlockNum().Uint64(), uint64(1050)) +} + +// TestMonitorPrefetchSlowChainNoExtraCalls checks the prefetcher stays out +// of the way of a monitor at the head of a slow chain: workers never ask for +// blocks the chain hasn't made, and in polling mode the head isn't polled. +func TestMonitorPrefetchSlowChainNoExtraCalls(t *testing.T) { + if testing.Short() { + t.Skip("Skipping in short mode") + } + + type calls struct{ blockNumber, blockByNumber, filterLogs int64 } + + run := func(streaming bool, concurrency int) calls { + chain := newFakeChain(1000, 1, 5*time.Millisecond) + monitor := newTestMonitor(t, chain, streaming, concurrency, 0) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go chain.produce(ctx, 200*time.Millisecond) + go monitor.Run(ctx) + defer monitor.Stop() + + time.Sleep(2 * time.Second) + return calls{ + blockNumber: chain.blockNumberCalls.Load(), + blockByNumber: chain.blockByNumberCalls.Load(), + filterLogs: chain.filterLogsCalls.Load(), + } + } + + for _, streaming := range []bool{false, true} { + t.Run(fmt.Sprintf("streaming=%v", streaming), func(t *testing.T) { + off := run(streaming, 0) + on := run(streaming, 4) + t.Logf("prefetch=0: %+v", off) + t.Logf("prefetch=4: %+v", on) + + assert.Zero(t, off.blockNumber) + if streaming { + // the stream supplies the head + assert.Zero(t, on.blockNumber) + } else { + // at most the one poll at startup, before the monitor has found + // out it's at the head + assert.LessOrEqual(t, on.blockNumber, int64(1)) + } + + // ~10 blocks made in the run: allow a couple of calls of jitter + // between the two runs, well short of a block per worker + assert.InDelta(t, off.blockByNumber, on.blockByNumber, 3) + assert.InDelta(t, off.filterLogs, on.filterLogs, 3) + }) + } +} + +func TestMonitorPrefetchShutdownNoGoroutineLeak(t *testing.T) { + if testing.Short() { + t.Skip("Skipping in short mode") + } + + chain := newFakeChain(1000, 100, 5*time.Millisecond) + baseline := runtime.NumGoroutine() + + monitor := newTestMonitor(t, chain, false, 4, 0) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { + done <- monitor.Run(ctx) + }() + + time.Sleep(300 * time.Millisecond) + cancel() + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("Monitor.Run() didn't return within timeout") + } + + // Run joins its goroutines; allow runtime cleanup before comparing counts. + // NOTE: poll here rather than with assert.Eventually, whose condition runs + // in a goroutine of its own and so always counts one extra. + deadline := time.Now().Add(2 * time.Second) + for runtime.NumGoroutine() > baseline && time.Now().Before(deadline) { + runtime.GC() + time.Sleep(50 * time.Millisecond) + } + if n := runtime.NumGoroutine(); n > baseline { + buf := make([]byte, 1<<20) + t.Fatalf("%d goroutine(s) leaked:\n%s", n-baseline, buf[:runtime.Stack(buf, true)]) + } +} + +// A peer can complete an old-fork fetch between DEL and the confirmation read, +// and a cache deletion can fail. Neither may cause a false canonical removal. +func TestMonitorPrefetchRefetchBypassesCache(t *testing.T) { + for _, deleteFails := range []bool{false, true} { + t.Run(fmt.Sprintf("deleteFails=%v", deleteFails), func(t *testing.T) { + chain := newFakeChain(1000, 20, time.Millisecond) + oldNext, ok := chain.byNumber(big.NewInt(1001)) + require.True(t, ok) + chain.reorgFrom(1000) + + backend, err := memcache.NewBackend(512) + require.NoError(t, err) + key := ethmonitor.CacheKeyBlockByNumber(big.NewInt(1), big.NewInt(1001)) + racingBackend := &repopulatingBackend{ + Backend: backend, key: key, payload: oldNext.payload(), deleteFails: deleteFails, + } + require.NoError(t, racingBackend.SetEx(context.Background(), key, []byte(oldNext.payload()), time.Minute)) + monitor := newTestMonitor(t, chain, false, 1, 1, racingBackend) + sub := monitor.Subscribe("TestMonitorPrefetchRefetchBypassesCache") + defer sub.Unsubscribe() + runMonitorForTest(t, monitor) + + timer := time.NewTimer(5 * time.Second) + defer timer.Stop() + for { + select { + case blocks := <-sub.Blocks(): + for _, b := range blocks { + require.Equal(t, ethmonitor.Added, b.Event, "removed canonical block %d", b.NumberU64()) + require.Equal(t, chain.hashAt(b.NumberU64()), b.Hash()) + if b.NumberU64() >= 1001 { + require.Positive(t, racingBackend.deletes.Load(), "did not exercise the mismatch confirmation") + return + } + } + case <-timer.C: + t.Fatal("monitor did not advance through the stale cache entry") + } + } + }) + } +} + +func TestMonitorPrefetchInvalidBlockRecovery(t *testing.T) { + for _, withLogs := range []bool{false, true} { + for _, cached := range []bool{false, true} { + t.Run(fmt.Sprintf("withLogs=%v/cached=%v", withLogs, cached), func(t *testing.T) { + chain := newFakeChain(1000, 30, time.Millisecond) + initial := newTestMonitor(t, chain, false, 4, 0) + opts := initial.Options() + opts.WithLogs = withLogs + provider := &invalidBlockProvider{fakeProvider: initial.Provider().(*fakeProvider), target: 1000} + if cached { + key := ethmonitor.CacheKeyBlockByNumber(big.NewInt(1), big.NewInt(1000)) + require.NoError(t, opts.CacheBackend.SetEx(context.Background(), key, []byte(`{"number":"0x3e8"}`), time.Minute)) + // The origin is healthy; only the existing cache entry is invalid. + provider.injected.Store(true) + } else { + // Exercise the automatic memory cache with one invalid origin response. + opts.CacheBackend = nil + } + monitor, err := ethmonitor.NewMonitor(provider, opts) + require.NoError(t, err) + runMonitorForTest(t, monitor) + require.Eventually(t, func() bool { + return monitor.LatestReadyBlock() != nil && monitor.LatestBlockNum().Uint64() == chain.head() + }, 2*time.Second, time.Millisecond) + require.True(t, provider.injected.Load()) + for _, block := range monitor.Chain().Blocks() { + require.Equal(t, chain.hashAt(block.NumberU64()), block.Hash()) + } + }) + } + } +} + +func TestMonitorPrefetchIdleAfterBacklog(t *testing.T) { + chain := newFakeChain(1000, 12, time.Millisecond) + monitor := newTestMonitor(t, chain, false, 4, 0) + provider := &idleHeadProvider{fakeProvider: monitor.Provider().(*fakeProvider), missed: make(chan struct{})} + monitor, err := ethmonitor.NewMonitor(provider, monitor.Options()) + require.NoError(t, err) + runMonitorForTest(t, monitor) + + // Wait for the serial loop's first miss after draining the backlog, rather + // than sampling during the last block's processing. + select { + case <-provider.missed: + case <-time.After(2 * time.Second): + t.Fatal("monitor did not drain the backlog") + } + before := chain.blockNumberCalls.Load() + time.Sleep(150 * time.Millisecond) + // Permit a head poll that was already selected when the miss occurred. + require.LessOrEqual(t, chain.blockNumberCalls.Load()-before, int64(1)) +} + +func TestMonitorPrefetchFatalExitStopsWorkers(t *testing.T) { + chain := newFakeChain(1000, 20, time.Millisecond) + backend, err := memcache.NewBackend(512) + require.NoError(t, err) + initial := newTestMonitor(t, chain, false, 4, 0, &invalidLogsBackend{backend}) + opts := initial.Options() + opts.BlockRetentionLimit = 2 // Four queued events; failed logs prevent dequeue. + monitor, err := ethmonitor.NewMonitor(initial.Provider().(ethrpc.RawInterface), opts) + require.NoError(t, err) + sub := monitor.Subscribe("TestMonitorPrefetchFatalExitStopsWorkers") + defer sub.Unsubscribe() + + for i := 0; i < 2; i++ { + done := runMonitorForTest(t, monitor) + select { + case err := <-done: + require.ErrorIs(t, err, ethmonitor.ErrFatal) + case <-time.After(2 * time.Second): + t.Fatal("expected a fatal publish error") + } + require.False(t, monitor.IsRunning()) + before := chain.blockNumberCalls.Load() + time.Sleep(150 * time.Millisecond) + require.Equal(t, before, chain.blockNumberCalls.Load(), "prefetch continued after Run returned") + } +} + +func TestMonitorPrefetchRestart(t *testing.T) { + for _, streaming := range []bool{false, true} { + t.Run(fmt.Sprintf("streaming=%v", streaming), func(t *testing.T) { + chain := newFakeChain(1000, 10, time.Millisecond) + monitor := newTestMonitor(t, chain, streaming, 4, 0) + ctx, cancel := context.WithCancel(context.Background()) + produced := make(chan struct{}) + go func() { + defer close(produced) + chain.produce(ctx, 20*time.Millisecond) + }() + defer func() { cancel(); <-produced }() + + for i := 0; i < 2; i++ { + target := chain.head() + done := runMonitorForTest(t, monitor) + require.Eventually(t, func() bool { + return monitor.LatestBlockNum().Uint64() >= target + }, 2*time.Second, time.Millisecond) + monitor.Stop() + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(2 * time.Second): + t.Fatal("monitor did not stop") + } + // Make a gap that the next run must ingest. + require.Eventually(t, func() bool { + return chain.head() >= monitor.LatestBlockNum().Uint64()+3 + }, time.Second, time.Millisecond) + } + }) + } +} + +func runMonitorForTest(t *testing.T, monitor *ethmonitor.Monitor) <-chan error { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { + done <- monitor.Run(ctx) + close(done) + }() + t.Cleanup(func() { + cancel() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Error("monitor did not shut down") + } + }) + return done +} + +type repopulatingBackend struct { + cachestore.Backend + key string + payload []byte + deleteFails bool + once sync.Once + deletes atomic.Int64 +} + +func (b *repopulatingBackend) Delete(ctx context.Context, key string) error { + if key != b.key { + return b.Backend.Delete(ctx, key) + } + b.deletes.Add(1) + if b.deleteFails { + return errors.New("simulated cache deletion failure") + } + if err := b.Backend.Delete(ctx, key); err != nil { + return err + } + b.once.Do(func() { _ = b.Backend.SetEx(ctx, key, b.payload, time.Minute) }) + return nil +} + +type invalidBlockProvider struct { + *fakeProvider + target uint64 + injected atomic.Bool +} + +func (p *invalidBlockProvider) RawBlockByNumber(ctx context.Context, num *big.Int) (json.RawMessage, error) { + payload, err := p.fakeProvider.RawBlockByNumber(ctx, num) + if err == nil && num != nil && num.Uint64() == p.target && p.injected.CompareAndSwap(false, true) { + return json.RawMessage(fmt.Sprintf(`{"number":"0x%x"}`, p.target)), nil + } + return payload, err +} + +type idleHeadProvider struct { + *fakeProvider + missed chan struct{} + once sync.Once +} + +func (p *idleHeadProvider) RawBlockByNumber(ctx context.Context, num *big.Int) (json.RawMessage, error) { + payload, err := p.fakeProvider.RawBlockByNumber(ctx, num) + if errors.Is(err, ethereum.NotFound) { + p.once.Do(func() { close(p.missed) }) + } + return payload, err +} + +type invalidLogsBackend struct{ cachestore.Backend } + +func (b *invalidLogsBackend) GetOrSetWithLockEx(ctx context.Context, key string, getter func(context.Context, string) (any, error), ttl time.Duration) (any, error) { + if strings.Contains(key, ":Logs:") { + return []byte("invalid logs"), nil + } + return b.Backend.GetOrSetWithLockEx(ctx, key, getter, ttl) +} + +func newTestMonitor(t *testing.T, chain *fakeChain, streaming bool, prefetchConcurrency, prefetchWindow int, cacheBackend ...cachestore.Backend) *ethmonitor.Monitor { + t.Helper() + + provider := &fakeProvider{ + MockRawInterface: mocks.NewMockRawInterface(gomock.NewController(t)), + chain: chain, + streaming: streaming, + } + + opts := ethmonitor.DefaultOptions + opts.PollingInterval = 20 * time.Millisecond + opts.Timeout = 2 * time.Second + opts.WithLogs = true + opts.StartBlockNumber = new(big.Int).SetUint64(chain.base) + opts.PrefetchConcurrency = prefetchConcurrency + opts.PrefetchWindow = prefetchWindow + if len(cacheBackend) > 0 { + opts.CacheBackend = cacheBackend[0] + } + + monitor, err := ethmonitor.NewMonitor(provider, opts) + require.NoError(t, err) + return monitor +} + +// fakeChain is an in-memory chain served with a fixed latency per call. +type fakeChain struct { + base uint64 + latency time.Duration + + // announceAhead makes the chain announce heads this many blocks past the + // blocks it serves. + announceAhead uint64 + + // node calls served, by method + blockNumberCalls atomic.Int64 + blockByNumberCalls atomic.Int64 + filterLogsCalls atomic.Int64 + + mu sync.Mutex + canonical []common.Hash // canonical[i] is block base+i + blocks map[common.Hash]fakeBlock + fork int + heads []*fakeSubscription +} + +type fakeBlock struct { + num uint64 + hash common.Hash + parent common.Hash +} + +func newFakeChain(base uint64, n int, latency time.Duration) *fakeChain { + c := &fakeChain{base: base, latency: latency, blocks: map[common.Hash]fakeBlock{}} + for i := 0; i < n; i++ { + c.appendLocked() + } + return c +} + +func (c *fakeChain) appendLocked() fakeBlock { + num := c.base + uint64(len(c.canonical)) + b := fakeBlock{ + num: num, + hash: crypto.Keccak256Hash([]byte(fmt.Sprintf("fork:%d/block:%d", c.fork, num))), + } + if len(c.canonical) > 0 { + b.parent = c.canonical[len(c.canonical)-1] + } + c.canonical = append(c.canonical, b.hash) + c.blocks[b.hash] = b + return b +} + +func (c *fakeChain) produce(ctx context.Context, interval time.Duration) { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + c.mu.Lock() + b := c.appendLocked() + heads := append([]*fakeSubscription{}, c.heads...) + c.mu.Unlock() + + header := &types.Header{Number: new(big.Int).SetUint64(b.num + c.announceAhead)} + for _, sub := range heads { + sub.send(header) + } + } + } +} + +// reorgFrom replaces the canonical chain from block num onwards with a new +// fork of the same length. +func (c *fakeChain) reorgFrom(num uint64) { + c.mu.Lock() + defer c.mu.Unlock() + + n := len(c.canonical) + c.canonical = c.canonical[:num-c.base] + c.fork++ + for len(c.canonical) < n { + c.appendLocked() + } +} + +func (c *fakeChain) head() uint64 { + c.mu.Lock() + defer c.mu.Unlock() + return c.base + uint64(len(c.canonical)) - 1 +} + +func (c *fakeChain) hashAt(num uint64) common.Hash { + c.mu.Lock() + defer c.mu.Unlock() + return c.canonical[num-c.base] +} + +func (c *fakeChain) byNumber(num *big.Int) (fakeBlock, bool) { + c.mu.Lock() + defer c.mu.Unlock() + i := uint64(len(c.canonical) - 1) + if num != nil { + if num.Uint64() < c.base || num.Uint64()-c.base > i { + return fakeBlock{}, false + } + i = num.Uint64() - c.base + } + return c.blocks[c.canonical[i]], true +} + +func (c *fakeChain) byHash(hash common.Hash) (fakeBlock, bool) { + c.mu.Lock() + defer c.mu.Unlock() + b, ok := c.blocks[hash] + return b, ok +} + +func (c *fakeChain) wait(ctx context.Context) error { + select { + case <-time.After(c.latency): + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func (b fakeBlock) payload() json.RawMessage { + zero := common.Hash{}.Hex() + return json.RawMessage(fmt.Sprintf(`{`+ + `"number":"0x%x","hash":"%s","parentHash":"%s",`+ + `"sha3Uncles":"%s","miner":"0x0000000000000000000000000000000000000000",`+ + `"stateRoot":"%s","transactionsRoot":"%s","receiptsRoot":"%s",`+ + `"logsBloom":"0x%s","difficulty":"0x0","gasLimit":"0x1c9c380","gasUsed":"0x0",`+ + `"timestamp":"0x%x","extraData":"0x","transactions":[],"uncles":[]}`, + b.num, b.hash.Hex(), b.parent.Hex(), + zero, zero, zero, zero, + strings.Repeat("0", 512), + b.num, + )) +} + +// fakeProvider serves a fakeChain. Methods the monitor is not expected to +// call fall through to the gomock mock, which fails the test. +type fakeProvider struct { + *mocks.MockRawInterface + chain *fakeChain + streaming bool +} + +var _ ethrpc.RawInterface = &fakeProvider{} + +func (p *fakeProvider) ChainID(ctx context.Context) (*big.Int, error) { + return big.NewInt(1), nil +} + +func (p *fakeProvider) IsStreamingEnabled() bool { + return p.streaming +} + +func (p *fakeProvider) BlockNumber(ctx context.Context) (uint64, error) { + p.chain.blockNumberCalls.Add(1) + if err := p.chain.wait(ctx); err != nil { + return 0, err + } + return p.chain.head() + p.chain.announceAhead, nil +} + +func (p *fakeProvider) RawBlockByNumber(ctx context.Context, num *big.Int) (json.RawMessage, error) { + p.chain.blockByNumberCalls.Add(1) + if err := p.chain.wait(ctx); err != nil { + return nil, err + } + b, ok := p.chain.byNumber(num) + if !ok { + return nil, ethereum.NotFound + } + return b.payload(), nil +} + +func (p *fakeProvider) RawBlockByHash(ctx context.Context, hash common.Hash) (json.RawMessage, error) { + if err := p.chain.wait(ctx); err != nil { + return nil, err + } + b, ok := p.chain.byHash(hash) + if !ok { + return nil, ethereum.NotFound + } + return b.payload(), nil +} + +func (p *fakeProvider) RawFilterLogs(ctx context.Context, q ethereum.FilterQuery) (json.RawMessage, error) { + p.chain.filterLogsCalls.Add(1) + if err := p.chain.wait(ctx); err != nil { + return nil, err + } + return json.RawMessage(`[]`), nil +} + +func (p *fakeProvider) SubscribeNewHeads(ctx context.Context, ch chan<- *types.Header) (ethereum.Subscription, error) { + sub := &fakeSubscription{ch: ch, err: make(chan error), done: make(chan struct{})} + p.chain.mu.Lock() + p.chain.heads = append(p.chain.heads, sub) + p.chain.mu.Unlock() + return sub, nil +} + +type fakeSubscription struct { + ch chan<- *types.Header + err chan error + done chan struct{} + once sync.Once +} + +func (s *fakeSubscription) send(header *types.Header) { + select { + case s.ch <- header: + case <-s.done: + } +} + +func (s *fakeSubscription) Unsubscribe() { + s.once.Do(func() { close(s.done) }) +} + +func (s *fakeSubscription) Err() <-chan error { + return s.err +} + +// eventLog rebuilds the chain a subscriber sees from the monitor's events, +// checking each event is consistent with what came before. +type eventLog struct { + mu sync.Mutex + chain []fakeBlock +} + +func newEventLog() *eventLog { + return &eventLog{} +} + +func (l *eventLog) apply(t *testing.T, blocks ethmonitor.Blocks) { + l.mu.Lock() + defer l.mu.Unlock() + + for _, b := range blocks { + switch b.Event { + case ethmonitor.Added: + if n := len(l.chain); n > 0 { + assert.Equal(t, l.chain[n-1].hash, b.ParentHash(), "added block %d does not extend the chain", b.NumberU64()) + } + l.chain = append(l.chain, fakeBlock{num: b.NumberU64(), hash: b.Hash(), parent: b.ParentHash()}) + case ethmonitor.Removed: + n := len(l.chain) + if assert.Greater(t, n, 0, "removed block %d from an empty chain", b.NumberU64()) { + assert.Equal(t, l.chain[n-1].hash, b.Hash(), "removed block %d is not the head", b.NumberU64()) + l.chain = l.chain[:n-1] + } + } + } +} + +func (l *eventLog) head() common.Hash { + l.mu.Lock() + defer l.mu.Unlock() + if len(l.chain) == 0 { + return common.Hash{} + } + return l.chain[len(l.chain)-1].hash +} + +func (l *eventLog) blocks() map[uint64]common.Hash { + l.mu.Lock() + defer l.mu.Unlock() + out := make(map[uint64]common.Hash, len(l.chain)) + for _, b := range l.chain { + out[b.num] = b.hash + } + return out +} From 1c355320ee6f02939bbf513bdbd465659d345b57 Mon Sep 17 00:00:00 2001 From: Peter Kieltyka Date: Wed, 30 Sep 2026 14:08:14 -0400 Subject: [PATCH 2/5] test(ethmonitor): avoid timing-dependent polling counts --- ethmonitor/prefetch_test.go | 30 ++++++++++++++++++++---------- 1 file changed, 20 insertions(+), 10 deletions(-) diff --git a/ethmonitor/prefetch_test.go b/ethmonitor/prefetch_test.go index 51d4216d..1c84edc0 100644 --- a/ethmonitor/prefetch_test.go +++ b/ethmonitor/prefetch_test.go @@ -186,7 +186,7 @@ func TestMonitorPrefetchSlowChainNoExtraCalls(t *testing.T) { t.Skip("Skipping in short mode") } - type calls struct{ blockNumber, blockByNumber, filterLogs int64 } + type calls struct{ blockNumber, foundBlocks, filterLogs, farAheadBlocks int64 } run := func(streaming bool, concurrency int) calls { chain := newFakeChain(1000, 1, 5*time.Millisecond) @@ -200,9 +200,10 @@ func TestMonitorPrefetchSlowChainNoExtraCalls(t *testing.T) { time.Sleep(2 * time.Second) return calls{ - blockNumber: chain.blockNumberCalls.Load(), - blockByNumber: chain.blockByNumberCalls.Load(), - filterLogs: chain.filterLogsCalls.Load(), + blockNumber: chain.blockNumberCalls.Load(), + foundBlocks: chain.foundBlockByNumberCalls.Load(), + filterLogs: chain.filterLogsCalls.Load(), + farAheadBlocks: chain.farAheadBlockByNumberCalls.Load(), } } @@ -214,6 +215,8 @@ func TestMonitorPrefetchSlowChainNoExtraCalls(t *testing.T) { t.Logf("prefetch=4: %+v", on) assert.Zero(t, off.blockNumber) + assert.Zero(t, off.farAheadBlocks) + assert.Zero(t, on.farAheadBlocks, "prefetch asked for a block beyond the next block at the head") if streaming { // the stream supplies the head assert.Zero(t, on.blockNumber) @@ -223,9 +226,10 @@ func TestMonitorPrefetchSlowChainNoExtraCalls(t *testing.T) { assert.LessOrEqual(t, on.blockNumber, int64(1)) } - // ~10 blocks made in the run: allow a couple of calls of jitter - // between the two runs, well short of a block per worker - assert.InDelta(t, off.blockByNumber, on.blockByNumber, 3) + // Compare requests that found blocks. Ordinary polling retries for the + // next missing block depend on timer scheduling and cannot be compared + // across independent runs. Allow jitter in the number of produced blocks. + assert.InDelta(t, off.foundBlocks, on.foundBlocks, 3) assert.InDelta(t, off.filterLogs, on.filterLogs, 3) }) } @@ -543,9 +547,11 @@ type fakeChain struct { announceAhead uint64 // node calls served, by method - blockNumberCalls atomic.Int64 - blockByNumberCalls atomic.Int64 - filterLogsCalls atomic.Int64 + blockNumberCalls atomic.Int64 + blockByNumberCalls atomic.Int64 + foundBlockByNumberCalls atomic.Int64 + farAheadBlockByNumberCalls atomic.Int64 + filterLogsCalls atomic.Int64 mu sync.Mutex canonical []common.Hash // canonical[i] is block base+i @@ -701,6 +707,9 @@ func (p *fakeProvider) BlockNumber(ctx context.Context) (uint64, error) { func (p *fakeProvider) RawBlockByNumber(ctx context.Context, num *big.Int) (json.RawMessage, error) { p.chain.blockByNumberCalls.Add(1) + if num != nil && num.Uint64() > p.chain.head()+1 { + p.chain.farAheadBlockByNumberCalls.Add(1) + } if err := p.chain.wait(ctx); err != nil { return nil, err } @@ -708,6 +717,7 @@ func (p *fakeProvider) RawBlockByNumber(ctx context.Context, num *big.Int) (json if !ok { return nil, ethereum.NotFound } + p.chain.foundBlockByNumberCalls.Add(1) return b.payload(), nil } From 987262d69722820de861cc166cf4672f63fc1e56 Mon Sep 17 00:00:00 2001 From: Peter Kieltyka Date: Fri, 2 Oct 2026 14:39:30 -0400 Subject: [PATCH 3/5] fix(ethmonitor): reject malformed logs before caching --- ethmonitor/ethmonitor.go | 21 ++++- ethmonitor/prefetch_internal_test.go | 135 +++++++++++++++++++++++++++ 2 files changed, 154 insertions(+), 2 deletions(-) diff --git a/ethmonitor/ethmonitor.go b/ethmonitor/ethmonitor.go index c1c70b84..041c7104 100644 --- a/ethmonitor/ethmonitor.go +++ b/ethmonitor/ethmonitor.go @@ -851,6 +851,8 @@ func (m *Monitor) logTopics() [][]common.Hash { } func (m *Monitor) filterLogs(ctx context.Context, blockHash common.Hash, topics [][]common.Hash, blockBloom types.Bloom) ([]types.Log, []byte, error) { + var fetchedLogs []types.Log + getter := func(ctx context.Context, _ string) ([]byte, error) { if m.options.DebugLogging { m.log.Debug(fmt.Sprintf("ethmonitor: filterLogs is calling origin for block hash %s", blockHash)) @@ -873,6 +875,12 @@ func (m *Monitor) filterLogs(ctx context.Context, blockHash common.Hash, topics // ensure we do not incorrectly cache an empty block-logs response as valid. return nil, fmt.Errorf("ethmonitor: filterLogs detected empty block-logs response but block bloom is set, ignoring node response") } + // Validate before caching so a malformed response cannot block log + // backfilling until cache expiry. + fetchedLogs, err = m.unmarshalLogs(logsPayload) + if err != nil { + return nil, err + } return logsPayload, nil } @@ -881,8 +889,7 @@ func (m *Monitor) filterLogs(ctx context.Context, blockHash common.Hash, topics if err != nil { return nil, resp, err } - logs, err := m.unmarshalLogs(resp) - return logs, resp, err + return fetchedLogs, resp, nil } key := CacheKeyBlockLogs(m.chainID, blockHash, topics) @@ -890,7 +897,17 @@ func (m *Monitor) filterLogs(ctx context.Context, blockHash common.Hash, topics if err != nil { return nil, resp, err } + if fetchedLogs != nil { + return fetchedLogs, resp, nil + } logs, err := m.unmarshalLogs(resp) + if err != nil { + // Recover entries cached by peers or older monitors that did not + // validate logs before writing them. + if deleteErr := m.cache.Delete(ctx, key); deleteErr != nil { + m.log.Warn(fmt.Sprintf("ethmonitor: error deleting invalid logs cache for block hash %s due to: '%v'", blockHash.Hex(), deleteErr)) + } + } return logs, resp, err } diff --git a/ethmonitor/prefetch_internal_test.go b/ethmonitor/prefetch_internal_test.go index 259ae05e..0bdb593a 100644 --- a/ethmonitor/prefetch_internal_test.go +++ b/ethmonitor/prefetch_internal_test.go @@ -5,10 +5,14 @@ import ( "encoding/json" "fmt" "math/big" + "sync/atomic" "testing" "time" "github.com/0xsequence/ethkit/ethmonitor/internal/mocks" + "github.com/0xsequence/ethkit/go-ethereum" + "github.com/0xsequence/ethkit/go-ethereum/common" + "github.com/0xsequence/ethkit/go-ethereum/core/types" "github.com/stretchr/testify/require" "go.uber.org/mock/gomock" ) @@ -35,6 +39,137 @@ func TestPrefetchRejectsInvalidBlockBeforeCaching(t *testing.T) { } } +func TestMonitorRecoversInvalidPrefetchedLogs(t *testing.T) { + for _, payload := range []string{`{}`, `[{}]`} { + for _, cached := range []bool{false, true} { + t.Run(fmt.Sprintf("payload=%s/cached=%v", payload, cached), func(t *testing.T) { + const first, target, last = uint64(1000), uint64(1002), uint64(1004) + targetHash := common.BigToHash(new(big.Int).SetUint64(target)) + expectedLogs := []types.Log{{ + Address: common.HexToAddress("0x1234"), + Topics: []common.Hash{common.HexToHash("0xabcd")}, + Data: []byte{1, 2}, + TxHash: common.HexToHash("0x5678"), + BlockHash: targetHash, + BlockNumber: target, + }} + validPayload, err := json.Marshal(expectedLogs) + require.NoError(t, err) + + provider := mocks.NewMockRawInterface(gomock.NewController(t)) + provider.EXPECT().ChainID(gomock.Any()).Return(big.NewInt(1), nil).AnyTimes() + provider.EXPECT().IsStreamingEnabled().Return(false).AnyTimes() + provider.EXPECT().BlockNumber(gomock.Any()).Return(last, nil).AnyTimes() + provider.EXPECT().RawBlockByNumber(gomock.Any(), gomock.Any()).DoAndReturn( + func(ctx context.Context, number *big.Int) (json.RawMessage, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + if number.Uint64() > last { + return nil, ethereum.NotFound + } + var bloom types.Bloom + if number.Uint64() == target { + bloom = types.BytesToBloom([]byte{1}) + } + header := &types.Header{ + Number: number, + ParentHash: common.BigToHash(new(big.Int).Sub(number, big.NewInt(1))), + Difficulty: big.NewInt(0), + GasLimit: 30_000_000, + Bloom: bloom, + } + header.SetHash(common.BigToHash(number)) + return json.Marshal(header) + }, + ).AnyTimes() + var originCalls atomic.Int64 + provider.EXPECT().RawFilterLogs(gomock.Any(), gomock.Any()).DoAndReturn( + func(ctx context.Context, query ethereum.FilterQuery) (json.RawMessage, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + if *query.BlockHash != targetHash { + return json.RawMessage(`[]`), nil + } + if originCalls.Add(1) == 1 && !cached { + return json.RawMessage(payload), nil + } + return json.RawMessage(validPayload), nil + }, + ).AnyTimes() + + opts := DefaultOptions + opts.WithLogs = true + opts.PrefetchConcurrency = 1 + opts.StartBlockNumber = new(big.Int).SetUint64(first) + opts.PollingInterval = 10 * time.Millisecond + opts.CacheExpiry = time.Hour + // Leave CacheBackend unset to exercise automatic memory caching. + monitor, err := NewMonitor(provider, opts) + require.NoError(t, err) + monitor.chainID = big.NewInt(1) + key := CacheKeyBlockLogs(monitor.chainID, targetHash, monitor.logTopics()) + if cached { + require.NoError(t, monitor.cache.SetEx(context.Background(), key, []byte(payload), time.Hour)) + } else { + // Complete the speculative fetch before starting the serial loop, + // so the worker deterministically receives the malformed response. + monitor.prefetch.fetch(context.Background(), prefetchJob{num: target}) + require.Equal(t, int64(1), originCalls.Load()) + _, found, err := monitor.cache.Get(context.Background(), key) + require.NoError(t, err) + require.False(t, found, "failed prefetch must not cache malformed logs") + } + + sub := monitor.Subscribe("TestMonitorRecoversInvalidPrefetchedLogs") + defer sub.Unsubscribe() + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- monitor.Run(ctx) }() + defer func() { + cancel() + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(5 * time.Second): + t.Error("monitor did not stop") + } + }() + + timer := time.NewTimer(3 * time.Second) + defer timer.Stop() + next := first + for next <= last { + select { + case blocks := <-sub.Blocks(): + for _, block := range blocks { + require.Equal(t, Added, block.Event) + require.True(t, block.OK) + require.Equal(t, next, block.NumberU64(), "blocks must publish in order") + if next == target { + require.Equal(t, expectedLogs, block.Logs) + } + next++ + } + case <-timer.C: + t.Fatalf("publication stalled at block %d; origin calls: %d", next, originCalls.Load()) + } + } + if !cached { + require.GreaterOrEqual(t, originCalls.Load(), int64(2), "malformed prefetch must allow an origin retry") + } else { + require.Positive(t, originCalls.Load(), "invalid cache entry must allow an origin retry") + } + stored, found, err := monitor.cache.Get(context.Background(), key) + require.NoError(t, err) + require.True(t, found) + require.JSONEq(t, string(validPayload), string(stored)) + }) + } + } +} + func TestPrefetchPanicStopsWorkers(t *testing.T) { provider := mocks.NewMockRawInterface(gomock.NewController(t)) provider.EXPECT().BlockNumber(gomock.Any()).DoAndReturn(func(context.Context) (uint64, error) { From 247c2e19340d297056992200e4bf2fd675bc3622 Mon Sep 17 00:00:00 2001 From: Peter Kieltyka Date: Mon, 5 Oct 2026 10:10:50 -0400 Subject: [PATCH 4/5] ethmonitor: make the prefetched reorg test deterministic TestMonitorPrefetchReorg/prefetched=true raced the monitor: it waited for a block 2-4 past the monitor head to show up in the cache, assuming the monitor (~15ms a block) couldn't reach it first. On slow CI runners (macos-latest) the monitor either drained the fixed chain before the condition was seen, failing the test, or reached the reorg block first, passing without exercising the stale prefetched block at all. Hold block 1060 on the fake chain instead, so the monitor stalls before it while the workers prefetch past it. Reorg from 1062 once it is cached from the old fork, then release. Also assert the reorg was actually exercised: a block from the abandoned fork must be published and removed. Co-Authored-By: Claude Opus 5.5 (1M context) --- ethmonitor/prefetch_test.go | 94 ++++++++++++++++++++++++++++++------- 1 file changed, 76 insertions(+), 18 deletions(-) diff --git a/ethmonitor/prefetch_test.go b/ethmonitor/prefetch_test.go index 1c84edc0..06f86822 100644 --- a/ethmonitor/prefetch_test.go +++ b/ethmonitor/prefetch_test.go @@ -107,35 +107,38 @@ func testMonitorPrefetchReorg(t *testing.T, concurrency int, prefetched bool) { } }() + // To reorg at a block the prefetcher cached from the abandoned fork, the + // monitor must not reach that block first. Rather than racing it, hold + // block 1060 so the monitor stalls before it while the workers fill the + // window past it, then reorg from 1062 and let the monitor continue. + const heldBlock, stalePrefetched = 1060, 1062 + if prefetched { + chain.hold(heldBlock) + } + ctx, cancel := context.WithCancel(context.Background()) defer cancel() go monitor.Run(ctx) defer monitor.Stop() - require.Eventually(t, func() bool { - return monitor.LatestBlockNum().Uint64() >= 1050 - }, 10*time.Second, time.Millisecond) - var reorgAt uint64 if prefetched { - // the monitor takes ~15ms a block here, so a block cached two or - // more past its head is still ahead of it when we reorg require.Eventually(t, func() bool { - head := monitor.LatestBlockNum().Uint64() - for n := head + 4; n >= head+2; n-- { - key := ethmonitor.CacheKeyBlockByNumber(big.NewInt(1), new(big.Int).SetUint64(n)) - if _, ok, _ := cache.Get(context.Background(), key); ok { - reorgAt = n - return true - } - } - return false - }, 5*time.Second, time.Millisecond) + key := ethmonitor.CacheKeyBlockByNumber(big.NewInt(1), big.NewInt(stalePrefetched)) + _, ok, _ := cache.Get(context.Background(), key) + return ok + }, 10*time.Second, time.Millisecond, "block %d was never prefetched", stalePrefetched) + require.Less(t, monitor.LatestBlockNum().Uint64(), uint64(heldBlock)) + reorgAt = stalePrefetched } else { + require.Eventually(t, func() bool { + return monitor.LatestBlockNum().Uint64() >= 1050 + }, 10*time.Second, time.Millisecond) reorgAt = monitor.LatestBlockNum().Uint64() - 2 } chain.reorgFrom(reorgAt) t.Logf("reorged chain from block %d, monitor head %d", reorgAt, monitor.LatestBlockNum().Uint64()) + chain.release(heldBlock) require.Eventually(t, func() bool { head := monitor.LatestReadyBlock() @@ -154,6 +157,11 @@ func testMonitorPrefetchReorg(t *testing.T, concurrency int, prefetched bool) { for num, hash := range events.blocks() { assert.Equal(t, chain.hashAt(num), hash, "subscriber block %d", num) } + + // both cases publish abandoned-fork blocks before the reorg is seen: in + // the prefetched case the stale block extends the head, so it is accepted + // and must be reverted once the new fork shows up. + assert.Positive(t, events.removedCount(), "the reorg was never exercised") } // TestMonitorPrefetchHeadAhead has the node announce heads it cannot serve @@ -558,6 +566,7 @@ type fakeChain struct { blocks map[common.Hash]fakeBlock fork int heads []*fakeSubscription + held map[uint64]chan struct{} // block numbers not served until released } type fakeBlock struct { @@ -623,6 +632,44 @@ func (c *fakeChain) reorgFrom(num uint64) { } } +// hold stops the chain serving block num by number until release(num). +func (c *fakeChain) hold(num uint64) { + c.mu.Lock() + defer c.mu.Unlock() + if c.held == nil { + c.held = map[uint64]chan struct{}{} + } + c.held[num] = make(chan struct{}) +} + +func (c *fakeChain) release(num uint64) { + c.mu.Lock() + defer c.mu.Unlock() + if ch, ok := c.held[num]; ok { + close(ch) + delete(c.held, num) + } +} + +// waitIfHeld blocks while block num is held. +func (c *fakeChain) waitIfHeld(ctx context.Context, num *big.Int) error { + if num == nil { + return nil + } + c.mu.Lock() + ch := c.held[num.Uint64()] + c.mu.Unlock() + if ch == nil { + return nil + } + select { + case <-ch: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + func (c *fakeChain) head() uint64 { c.mu.Lock() defer c.mu.Unlock() @@ -713,6 +760,9 @@ func (p *fakeProvider) RawBlockByNumber(ctx context.Context, num *big.Int) (json if err := p.chain.wait(ctx); err != nil { return nil, err } + if err := p.chain.waitIfHeld(ctx, num); err != nil { + return nil, err + } b, ok := p.chain.byNumber(num) if !ok { return nil, ethereum.NotFound @@ -773,8 +823,9 @@ func (s *fakeSubscription) Err() <-chan error { // eventLog rebuilds the chain a subscriber sees from the monitor's events, // checking each event is consistent with what came before. type eventLog struct { - mu sync.Mutex - chain []fakeBlock + mu sync.Mutex + chain []fakeBlock + removed int } func newEventLog() *eventLog { @@ -798,10 +849,17 @@ func (l *eventLog) apply(t *testing.T, blocks ethmonitor.Blocks) { assert.Equal(t, l.chain[n-1].hash, b.Hash(), "removed block %d is not the head", b.NumberU64()) l.chain = l.chain[:n-1] } + l.removed++ } } } +func (l *eventLog) removedCount() int { + l.mu.Lock() + defer l.mu.Unlock() + return l.removed +} + func (l *eventLog) head() common.Hash { l.mu.Lock() defer l.mu.Unlock() From 77bbef8b2ac2a3fcb70b360a9b2ff2ffacd1be69 Mon Sep 17 00:00:00 2001 From: Peter Kieltyka Date: Mon, 5 Oct 2026 15:39:01 -0400 Subject: [PATCH 5/5] fix(ethmonitor): confirm cached parents and track canonical incarnations --- ethmonitor/bootstrap.go | 6 +- ethmonitor/canonical_state_test.go | 179 +++++++++++++++++++++++++++++ ethmonitor/chain.go | 53 +++++++-- ethmonitor/ethmonitor.go | 19 +-- ethmonitor/prefetch_test.go | 134 +++++++++++++++++---- 5 files changed, 350 insertions(+), 41 deletions(-) create mode 100644 ethmonitor/canonical_state_test.go diff --git a/ethmonitor/bootstrap.go b/ethmonitor/bootstrap.go index c4c28f6d..d616c6e2 100644 --- a/ethmonitor/bootstrap.go +++ b/ethmonitor/bootstrap.go @@ -38,8 +38,9 @@ func (c *Chain) bootstrapBlocks(blocks Blocks) error { return nil } - if len(blocks) == 1 { + if len(blocks) == 1 && blocks[0].Event != Added { c.blocks = blocks.Copy() + c.blocks[0].canonicalState = nil return nil } @@ -51,7 +52,7 @@ func (c *Chain) bootstrapBlocks(blocks Blocks) error { for _, b := range blocks { if b.Event == Added { - err := c.push(b) + _, err := c.push(b) if err != nil { return fmt.Errorf("ethmonitor: bootstrap failed to build canonical chain: %w", err) } @@ -101,5 +102,6 @@ func (b *Block) UnmarshalJSON(data []byte) error { b.Event = s.Event b.Logs = s.Logs b.OK = s.OK + b.canonicalState = nil return nil } diff --git a/ethmonitor/canonical_state_test.go b/ethmonitor/canonical_state_test.go new file mode 100644 index 00000000..e6ca0e64 --- /dev/null +++ b/ethmonitor/canonical_state_test.go @@ -0,0 +1,179 @@ +package ethmonitor + +import ( + "encoding/json" + "fmt" + "math/big" + "testing" + + "github.com/0xsequence/ethkit/go-ethereum/common" + "github.com/0xsequence/ethkit/go-ethereum/core/types" + "github.com/stretchr/testify/require" +) + +func canonicalTestBlock(num int64) *Block { + return &Block{ + Block: types.NewBlockWithHeader(&types.Header{ + Number: big.NewInt(num), + BlockHash: common.BigToHash(big.NewInt(num)), + ParentHash: common.BigToHash(big.NewInt(num - 1)), + Time: uint64(num), + }), + Event: Added, + OK: true, + } +} + +func pushCanonicalTestBlock(t *testing.T, chain *Chain, input *Block) *Block { + t.Helper() + chain.push(input) + block := chain.Head() + require.NotNil(t, block) + require.Equal(t, input.Hash(), block.Hash()) + return block +} + +// The interface keeps the before-fix proof executable without the new method. +func canonicalTestState(t *testing.T, block *Block) (uint64, bool) { + t.Helper() + state, ok := any(block).(interface{ CanonicalState() (uint64, bool) }) + if !ok { + return 0, false + } + return state.CanonicalState() +} + +func TestBlockCanonicalStateRetention(t *testing.T) { + for _, depth := range []int64{1, 3} { + t.Run(fmt.Sprintf("evictionDepth=%d", depth), func(t *testing.T) { + chain := newChain(10, false) + block := pushCanonicalTestBlock(t, chain, canonicalTestBlock(100)) + copy := chain.Blocks().Copy()[0] + incarnation, canonical := canonicalTestState(t, block) + require.Positive(t, incarnation) + require.True(t, canonical) + for num := int64(101); num <= 109+depth; num++ { + pushCanonicalTestBlock(t, chain, canonicalTestBlock(num)) + } + require.Nil(t, chain.GetBlock(block.Hash())) + for _, snapshot := range []*Block{block, copy} { + got, canonical := canonicalTestState(t, snapshot) + require.Equal(t, incarnation, got) + require.True(t, canonical, "retention eviction was mistaken for removal") + } + }) + } +} + +func TestBlockCanonicalStateReadoption(t *testing.T) { + chain := newChain(10, false) + input := canonicalTestBlock(100) + block := pushCanonicalTestBlock(t, chain, input) + shallow := *block + copy := chain.Blocks().Copy()[0] + incarnation, canonical := canonicalTestState(t, block) + require.Positive(t, incarnation) + require.True(t, canonical) + removed := *chain.pop() + removed.Event = Removed + for _, snapshot := range []*Block{block, &shallow, copy, &removed} { + got, canonical := canonicalTestState(t, snapshot) + require.Equal(t, incarnation, got) + require.False(t, canonical, "snapshot did not observe the actual removal") + } + + // Reusing either the original input or a removed snapshot must create a fresh + // owned incarnation without mutating old Added/Removed event copies. + for _, reused := range []*Block{input, copy} { + fresh := pushCanonicalTestBlock(t, chain, reused) + got, canonical := canonicalTestState(t, fresh) + require.Greater(t, got, incarnation) + require.True(t, canonical) + for _, snapshot := range []*Block{block, &shallow, copy, &removed} { + got, canonical := canonicalTestState(t, snapshot) + require.Equal(t, incarnation, got) + require.False(t, canonical, "fresh readoption revived an old event") + } + chain.pop() + } +} + +func TestBlockCanonicalStateBootstrap(t *testing.T) { + for _, count := range []int{1, 3} { + for _, serialized := range []bool{false, true} { + t.Run(fmt.Sprintf("blocks=%d/JSON=%v", count, serialized), func(t *testing.T) { + inputs := make(Blocks, count) + for i := range inputs { + inputs[i] = canonicalTestBlock(int64(100 + i)) + } + chain := newChain(10, true) + if serialized { + data, err := json.Marshal(inputs) + require.NoError(t, err) + require.NoError(t, chain.BootstrapFromBlocksJSON(data)) + } else { + require.NoError(t, chain.BootstrapFromBlocks(inputs)) + } + for _, block := range chain.Blocks().Copy() { + incarnation, canonical := canonicalTestState(t, block) + require.Positive(t, incarnation) + require.True(t, canonical) + data, err := json.Marshal(block) + require.NoError(t, err) + require.NotContains(t, string(data), "incarnation") + require.NoError(t, json.Unmarshal(data, block)) + incarnation, canonical = canonicalTestState(t, block) + require.Zero(t, incarnation, "serialized state preserved runtime ownership") + require.False(t, canonical) + } + }) + } + } +} + +func TestBlockCanonicalStateConcurrentRemoval(t *testing.T) { + chain := newChain(10, false) + input := canonicalTestBlock(100) + block := pushCanonicalTestBlock(t, chain, input) + incarnation, _ := canonicalTestState(t, block) + require.Positive(t, incarnation) + state := any(block).(interface{ CanonicalState() (uint64, bool) }) + start, done := make(chan struct{}), make(chan struct{}) + result := make(chan error, 1) + go func() { + close(start) + removed := false + for { + got, canonical := state.CanonicalState() + if got != incarnation || (removed && canonical) { + result <- fmt.Errorf("old incarnation changed or revived: id=%d canonical=%v", got, canonical) + return + } + removed = removed || !canonical + select { + case <-done: + result <- nil + return + default: + } + } + }() + <-start + for i := 0; i < 100; i++ { + chain.pop() + pushCanonicalTestBlock(t, chain, input) + } + close(done) + require.NoError(t, <-result) + got, canonical := state.CanonicalState() + require.Equal(t, incarnation, got) + require.False(t, canonical) +} + +func TestBlockCanonicalStateUntracked(t *testing.T) { + for _, block := range []*Block{nil, canonicalTestBlock(100)} { + incarnation, canonical := canonicalTestState(t, block) + require.Zero(t, incarnation) + require.False(t, canonical) + } +} diff --git a/ethmonitor/chain.go b/ethmonitor/chain.go index 8685a862..7233b5d0 100644 --- a/ethmonitor/chain.go +++ b/ethmonitor/chain.go @@ -7,6 +7,7 @@ import ( "math/big" "slices" "sync" + "sync/atomic" "github.com/0xsequence/ethkit/go-ethereum" "github.com/0xsequence/ethkit/go-ethereum/common" @@ -28,6 +29,7 @@ type Chain struct { mu sync.RWMutex averageBlockTime float64 // in seconds + lastIncarnation uint64 } func newChain(retentionLimit int, bootstrapMode bool) *Chain { @@ -59,7 +61,7 @@ func newChain(retentionLimit int, bootstrapMode bool) *Chain { // } // Push to the top of the stack -func (c *Chain) push(nextBlock *Block) error { +func (c *Chain) push(nextBlock *Block) (*Block, error) { c.mu.Lock() defer c.mu.Unlock() @@ -70,12 +72,12 @@ func (c *Chain) push(nextBlock *Block) error { // Assert pointing at prev block if nextBlock.ParentHash() != headBlock.Hash() { - return ErrUnexpectedParentHash + return nil, ErrUnexpectedParentHash } // Assert block numbers are in sequence if nextBlock.NumberU64() != headBlock.NumberU64()+1 { - return ErrUnexpectedBlockNumber + return nil, ErrUnexpectedBlockNumber } // Update average block time @@ -86,14 +88,20 @@ func (c *Chain) push(nextBlock *Block) error { } } + // Each adoption owns its state so reusing an input cannot revive old events. + c.lastIncarnation++ + block := *nextBlock + block.canonicalState = &blockCanonicalState{incarnation: c.lastIncarnation} + block.canonicalState.canonical.Store(true) + // Add to head of stack - c.blocks = append(c.blocks, nextBlock) + c.blocks = append(c.blocks, &block) if len(c.blocks) > c.retentionLimit { c.blocks[0] = nil c.blocks = c.blocks[1:] } - return nil + return &block, nil } // Pop from the top of the stack @@ -107,6 +115,9 @@ func (c *Chain) pop() *Block { n := len(c.blocks) - 1 block := c.blocks[n] + if block.canonicalState != nil { + block.canonicalState.canonical.Store(false) + } c.blocks[n] = nil c.blocks = c.blocks[:n] return block @@ -215,6 +226,9 @@ const ( Removed ) +// Block contains a monitored block and its event data. +// Construct values with keyed composite literals: private canonical state makes +// positional literals unsupported. type Block struct { *types.Block @@ -228,6 +242,26 @@ type Block struct { // OK flag which represents the block is ready for broadcasting OK bool + + canonicalState *blockCanonicalState +} + +type blockCanonicalState struct { + incarnation uint64 + canonical atomic.Bool +} + +// CanonicalState reports the monitor-assigned incarnation and whether it has +// remained canonical without a known removal. Zero means the block is untracked. +// Retention eviction preserves this state; in-memory copies share removal updates. +// Incarnations are local to a monitor chain. Serialized blocks are untracked +// until accepted by a monitor. +func (b *Block) CanonicalState() (incarnation uint64, canonical bool) { + if b == nil || b.canonicalState == nil { + return 0, false + } + state := b.canonicalState + return state.incarnation, state.canonical.Load() } type Blocks []*Block @@ -560,10 +594,11 @@ func (blocks Blocks) Copy() Blocks { } nb[i] = &Block{ - Block: b.Block, - Event: b.Event, - Logs: logs, - OK: b.OK, + Block: b.Block, + Event: b.Event, + Logs: logs, + OK: b.OK, + canonicalState: b.canonicalState, } } diff --git a/ethmonitor/ethmonitor.go b/ethmonitor/ethmonitor.go index 041c7104..d59c4898 100644 --- a/ethmonitor/ethmonitor.go +++ b/ethmonitor/ethmonitor.go @@ -640,10 +640,10 @@ func (m *Monitor) monitor(ctx context.Context, wg *sync.WaitGroup) error { // fetch the next block, either via the stream or via a poll nextBlock, nextBlockPayload, miss, err := m.fetchNextBlock(ctx, false) - // with prefetching, a cached block may be a stale entry from an abandoned - // fork, written by a prefetch worker of this or a peer monitor. confirm a - // parent mismatch with the node before treating it as a reorg. - if err == nil && m.prefetch != nil && headBlock != nil && nextBlock.ParentHash() != headBlock.Hash() { + // A cached block may be a stale entry from an abandoned fork, written by + // this or a peer monitor even when local prefetching is disabled. Confirm + // a parent mismatch with the node before treating it as a reorg. + if err == nil && m.cache != nil && headBlock != nil && nextBlock.ParentHash() != headBlock.Hash() { nextBlock, nextBlockPayload, miss, err = m.refetchNextBlock(ctx, nextBlock) } @@ -730,9 +730,11 @@ func (m *Monitor) buildCanonicalChain(ctx context.Context, nextBlock *types.Bloc if headBlock == nil || nextBlock.ParentHash() == headBlock.Hash() { // block-chaining it up - block := &Block{Event: Added, Block: nextBlock} - events = append(events, block) - return events, m.chain.push(block) + block, err := m.chain.push(&Block{Event: Added, Block: nextBlock}) + if err != nil { + return events, err + } + return append(events, block), nil } // next block doest match prevHash, therefore we must pop our previous block and recursively @@ -778,8 +780,7 @@ func (m *Monitor) buildCanonicalChain(ctx context.Context, nextBlock *types.Bloc return events, err } - block := &Block{Event: Added, Block: nextBlock} - err = m.chain.push(block) + block, err := m.chain.push(&Block{Event: Added, Block: nextBlock}) if err != nil { return events, err } diff --git a/ethmonitor/prefetch_test.go b/ethmonitor/prefetch_test.go index 06f86822..d1713576 100644 --- a/ethmonitor/prefetch_test.go +++ b/ethmonitor/prefetch_test.go @@ -284,40 +284,132 @@ func TestMonitorPrefetchShutdownNoGoroutineLeak(t *testing.T) { // A peer can complete an old-fork fetch between DEL and the confirmation read, // and a cache deletion can fail. Neither may cause a false canonical removal. func TestMonitorPrefetchRefetchBypassesCache(t *testing.T) { - for _, deleteFails := range []bool{false, true} { - t.Run(fmt.Sprintf("deleteFails=%v", deleteFails), func(t *testing.T) { - chain := newFakeChain(1000, 20, time.Millisecond) - oldNext, ok := chain.byNumber(big.NewInt(1001)) - require.True(t, ok) - chain.reorgFrom(1000) + for _, concurrency := range []int{0, 1} { + for _, deleteFails := range []bool{false, true} { + t.Run(fmt.Sprintf("prefetch=%d/deleteFails=%v", concurrency, deleteFails), func(t *testing.T) { + chain := newFakeChain(1000, 20, time.Millisecond) + oldNext, ok := chain.byNumber(big.NewInt(1001)) + require.True(t, ok) + chain.reorgFrom(1000) + + backend, err := memcache.NewBackend(512) + require.NoError(t, err) + key := ethmonitor.CacheKeyBlockByNumber(big.NewInt(1), big.NewInt(1001)) + racingBackend := &repopulatingBackend{ + Backend: backend, key: key, payload: oldNext.payload(), deleteFails: deleteFails, + } + require.NoError(t, racingBackend.SetEx(context.Background(), key, []byte(oldNext.payload()), time.Minute)) + monitor := newTestMonitor(t, chain, false, concurrency, 1, racingBackend) + sub := monitor.Subscribe("TestMonitorPrefetchRefetchBypassesCache") + defer sub.Unsubscribe() + runMonitorForTest(t, monitor) - backend, err := memcache.NewBackend(512) - require.NoError(t, err) - key := ethmonitor.CacheKeyBlockByNumber(big.NewInt(1), big.NewInt(1001)) - racingBackend := &repopulatingBackend{ - Backend: backend, key: key, payload: oldNext.payload(), deleteFails: deleteFails, + timer := time.NewTimer(5 * time.Second) + defer timer.Stop() + for { + select { + case blocks := <-sub.Blocks(): + for _, b := range blocks { + require.Equal(t, ethmonitor.Added, b.Event, "removed canonical block %d", b.NumberU64()) + require.Equal(t, chain.hashAt(b.NumberU64()), b.Hash()) + if b.NumberU64() >= 1001 { + require.Positive(t, racingBackend.deletes.Load(), "did not exercise the mismatch confirmation") + return + } + } + case <-timer.C: + t.Fatal("monitor did not advance through the stale cache entry") + } + } + }) + } + } +} + +// Cache confirmation must preserve genuine reorgs, including when prefetching +// is disabled. An uncached monitor still follows the direct-origin path. +func TestMonitorParentMismatchReorg(t *testing.T) { + cases := []struct { + concurrency int + cached bool + }{{0, false}, {0, true}, {1, true}} + for _, tc := range cases { + t.Run(fmt.Sprintf("prefetch=%d/cached=%v", tc.concurrency, tc.cached), func(t *testing.T) { + chain := newFakeChain(1000, 1, time.Millisecond) + oldHash := chain.hashAt(1000) + var monitor *ethmonitor.Monitor + var tracker *repopulatingBackend + if tc.cached { + backend, err := memcache.NewBackend(512) + require.NoError(t, err) + tracker = &repopulatingBackend{Backend: backend, key: ethmonitor.CacheKeyBlockByNumber(big.NewInt(1), big.NewInt(1001)), deleteFails: true} + monitor = newTestMonitor(t, chain, false, tc.concurrency, 1, tracker) + } else { + monitor = newTestMonitor(t, chain, false, tc.concurrency, 1) + require.Nil(t, monitor.Options().CacheBackend) } - require.NoError(t, racingBackend.SetEx(context.Background(), key, []byte(oldNext.payload()), time.Minute)) - monitor := newTestMonitor(t, chain, false, 1, 1, racingBackend) - sub := monitor.Subscribe("TestMonitorPrefetchRefetchBypassesCache") + sub := monitor.Subscribe("TestMonitorParentMismatchReorg") defer sub.Unsubscribe() runMonitorForTest(t, monitor) - - timer := time.NewTimer(5 * time.Second) + state := func(block *ethmonitor.Block) (uint64, bool) { + t.Helper() + witness, ok := any(block).(interface{ CanonicalState() (uint64, bool) }) + require.True(t, ok) + return witness.CanonicalState() + } + var initial *ethmonitor.Block + var oldIncarnation uint64 + select { + case blocks := <-sub.Blocks(): + require.Len(t, blocks, 1) + require.Equal(t, ethmonitor.Added, blocks[0].Event) + require.Equal(t, oldHash, blocks[0].Hash()) + initial = blocks[0] + var canonical bool + oldIncarnation, canonical = state(initial) + require.Positive(t, oldIncarnation) + require.True(t, canonical) + case <-time.After(5 * time.Second): + t.Fatal("missing initial canonical block") + } + chain.reorgFrom(1000) + chain.mu.Lock() + chain.appendLocked() + chain.mu.Unlock() + timer := time.NewTimer(8 * time.Second) defer timer.Stop() + removed, added := 0, 0 for { select { case blocks := <-sub.Blocks(): for _, b := range blocks { - require.Equal(t, ethmonitor.Added, b.Event, "removed canonical block %d", b.NumberU64()) - require.Equal(t, chain.hashAt(b.NumberU64()), b.Hash()) - if b.NumberU64() >= 1001 { - require.Positive(t, racingBackend.deletes.Load(), "did not exercise the mismatch confirmation") + if b.Event == ethmonitor.Removed { + require.Equal(t, uint64(1000), b.NumberU64()) + require.Equal(t, oldHash, b.Hash()) + incarnation, canonical := state(b) + require.Equal(t, oldIncarnation, incarnation) + require.False(t, canonical) + _, canonical = state(initial) + require.False(t, canonical, "queued Added did not observe removal") + removed++ + } else { + require.Equal(t, chain.hashAt(b.NumberU64()), b.Hash()) + incarnation, canonical := state(b) + require.Greater(t, incarnation, oldIncarnation) + require.True(t, canonical) + added++ + } + if b.Event == ethmonitor.Added && b.NumberU64() == 1001 { + require.Equal(t, 1, removed) + require.Equal(t, 2, added) + if tracker != nil { + require.Positive(t, tracker.deletes.Load(), "cached parent mismatch was not confirmed") + } return } } case <-timer.C: - t.Fatal("monitor did not advance through the stale cache entry") + t.Fatal("real reorg did not recover to new canonical block 1001") } } })