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 8f7bd916..d59c4898 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) + + // 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) + } + 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() + } } } } @@ -613,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 @@ -633,6 +752,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())) } @@ -656,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 } @@ -691,12 +814,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,7 +841,19 @@ 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) { + 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)) @@ -746,6 +876,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 } @@ -754,8 +890,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) @@ -763,7 +898,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 } @@ -806,12 +951,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 +973,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 +987,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 +1012,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 +1087,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 +1201,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..0bdb593a --- /dev/null +++ b/ethmonitor/prefetch_internal_test.go @@ -0,0 +1,196 @@ +package ethmonitor + +import ( + "context" + "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" +) + +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 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) { + 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..d1713576 --- /dev/null +++ b/ethmonitor/prefetch_test.go @@ -0,0 +1,972 @@ +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) + } + }() + + // 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() + + var reorgAt uint64 + if prefetched { + require.Eventually(t, func() bool { + 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() + 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) + } + + // 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 +// 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, foundBlocks, filterLogs, farAheadBlocks 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(), + foundBlocks: chain.foundBlockByNumberCalls.Load(), + filterLogs: chain.filterLogsCalls.Load(), + farAheadBlocks: chain.farAheadBlockByNumberCalls.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) + 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) + } 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)) + } + + // 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) + }) + } +} + +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 _, 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) + + 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) + } + sub := monitor.Subscribe("TestMonitorParentMismatchReorg") + defer sub.Unsubscribe() + runMonitorForTest(t, monitor) + 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 { + 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("real reorg did not recover to new canonical block 1001") + } + } + }) + } +} + +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 + foundBlockByNumberCalls atomic.Int64 + farAheadBlockByNumberCalls 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 + held map[uint64]chan struct{} // block numbers not served until released +} + +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() + } +} + +// 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() + 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 num != nil && num.Uint64() > p.chain.head()+1 { + p.chain.farAheadBlockByNumberCalls.Add(1) + } + 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 + } + p.chain.foundBlockByNumberCalls.Add(1) + 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 + removed int +} + +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] + } + 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() + 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 +}