Skip to content

Commit c65f742

Browse files
pkieltykaclaude
andauthored
ethmonitor: parallel block prefetch for fast chains (#219)
* 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) <noreply@anthropic.com> * test(ethmonitor): avoid timing-dependent polling counts * fix(ethmonitor): reject malformed logs before caching * 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) <noreply@anthropic.com> * fix(ethmonitor): confirm cached parents and track canonical incarnations * Harden ethreceipts rollback, filter ownership and cancellation (#220) * fix(ethreceipts): harden rollback, filter ownership and cancellation * fix(ethreceipts): preserve retry candidates and filter identity Keep successful canceled retries tied to their learned block generation, reopen canonical adopted blocks after retention, and support stable custom value filter removal. Document supported limits and add regression coverage. * ci: test Go 1.25 and 1.26 * fix(ethreceipts): retry current candidates after stale RPC receipts * refactor(ethreceipts): pass block processing contexts explicitly * refactor(ethreceipts): restore breaker retries with v0.3.2 * test(ethreceipts): organize regressions by behavior * refactor(ethreceipts): clarify pending receipt retries * fix(ethreceipts): harden subscription cleanup and finality * fix(ethreceipts): preserve fetch filter compatibility * fix(ethmonitor): reject invalid empty log responses * fix(ethmonitor): preserve empty-log provider compatibility * fix(ethmonitor): retry origin after cache fetch failures * fix(ethmonitor): keep log publication alive during cache outages * fix(ethreceipts): release orphaned LimitOne selection at finality Rollback keeps a LimitOne owner's claim on its delivered txn so the same txn can be re-mined. Nothing released that claim once the orphaned delivery was pruned at finality, so a filter whose txn never came back silently dropped every later match: forever with MaxWait 0, or until ErrFilterExhausted otherwise. Release the claim when its delivery is pruned at finality. releaseClaim still keeps it while the owner has queued finality, in-flight, delivered or pending work, so a txn re-mined before then stays selected. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(ethreceipts): harden delivery and channel handling Upgrade goware/channel to v0.6.0 for safe concurrent Send/Close and backed-off queue alerts. Raise the subscriber warning threshold to 10. Release receiptMu before channel sends while deliveryMu continues to serialize publication, rollback, and finality. Recheck cancellation after block validation, accumulate matches across added blocks, and let custom filters manage their own MaxWait exhaustion. Preserve block generations needed by pending retries and late RPCs beyond monitor retention. Cover receipt recovery, orphan rejection, cancellation, blocked sends, batch matching, and custom exhaustion with regression tests. Bound the blocked-send test's startup and cleanup waits. Validation: full ethreceipts and ethmonitor suites with -race; focused regressions repeated 20 times; go build ./... and package vet passed. --------- Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
1 parent 72f37f6 commit c65f742

28 files changed

Lines changed: 7098 additions & 1136 deletions

‎.github/workflows/ci.yml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@ jobs:
1717

1818
strategy:
1919
matrix:
20-
go-version: [1.24.x,1.25.x]
20+
go-version: [1.25.x,1.26.x]
2121
os: [ubuntu-latest, macos-latest]
2222

2323
runs-on: ${{ matrix.os }}

‎Makefile‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -56,13 +56,14 @@ install:
5656
# Run baseline tests
5757
test: check-testchain-running go-test
5858

59+
# Packages share one testchain and deployment sender, so serialize package binaries.
5960
# Go test short-hand, and skip testing go-ethereum
6061
go-test: test-clean
61-
GOGC=off go test $(TEST_FLAGS) $(MOD_VENDOR) -race -run=$(TEST) `go list ./... | grep -v go-ethereum`
62+
GOGC=off go test $(TEST_FLAGS) $(MOD_VENDOR) -p=1 -race -run=$(TEST) `go list ./... | grep -v go-ethereum`
6263

6364
# Go test short-hand, including go-ethereum
6465
go-test-all: test-clean
65-
GOGC=off go test $(TEST_FLAGS) $(MOD_VENDOR) -run=$(TEST) ./...
66+
GOGC=off go test $(TEST_FLAGS) $(MOD_VENDOR) -p=1 -run=$(TEST) ./...
6667

6768
test-clean:
6869
GOGC=off go clean -testcache

‎ethmonitor/bootstrap.go‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -38,8 +38,9 @@ func (c *Chain) bootstrapBlocks(blocks Blocks) error {
3838
return nil
3939
}
4040

41-
if len(blocks) == 1 {
41+
if len(blocks) == 1 && blocks[0].Event != Added {
4242
c.blocks = blocks.Copy()
43+
c.blocks[0].canonicalState = nil
4344
return nil
4445
}
4546

@@ -51,7 +52,7 @@ func (c *Chain) bootstrapBlocks(blocks Blocks) error {
5152

5253
for _, b := range blocks {
5354
if b.Event == Added {
54-
err := c.push(b)
55+
_, err := c.push(b)
5556
if err != nil {
5657
return fmt.Errorf("ethmonitor: bootstrap failed to build canonical chain: %w", err)
5758
}
@@ -101,5 +102,6 @@ func (b *Block) UnmarshalJSON(data []byte) error {
101102
b.Event = s.Event
102103
b.Logs = s.Logs
103104
b.OK = s.OK
105+
b.canonicalState = nil
104106
return nil
105107
}

‎ethmonitor/cache_test.go‎

Lines changed: 131 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,131 @@
1+
package ethmonitor_test
2+
3+
import (
4+
"context"
5+
"encoding/json"
6+
"errors"
7+
"fmt"
8+
"math/big"
9+
"strings"
10+
"sync/atomic"
11+
"testing"
12+
"time"
13+
14+
"github.com/0xsequence/ethkit/ethmonitor"
15+
"github.com/0xsequence/ethkit/ethmonitor/internal/mocks"
16+
"github.com/0xsequence/ethkit/go-ethereum"
17+
"github.com/0xsequence/ethkit/go-ethereum/common"
18+
"github.com/0xsequence/ethkit/go-ethereum/core/types"
19+
memcache "github.com/goware/cachestore-mem"
20+
cachestore "github.com/goware/cachestore2"
21+
"github.com/stretchr/testify/require"
22+
"go.uber.org/mock/gomock"
23+
)
24+
25+
// Fail before invoking the getter, as an unreachable shared cache would.
26+
type outageBackend struct {
27+
cachestore.Backend
28+
down atomic.Bool
29+
failures atomic.Int64
30+
}
31+
32+
func (b *outageBackend) GetOrSetWithLockEx(ctx context.Context, key string, getter func(context.Context, string) (any, error), ttl time.Duration) (any, error) {
33+
if b.down.Load() {
34+
b.failures.Add(1)
35+
return nil, errors.New("cache unavailable")
36+
}
37+
return b.Backend.GetOrSetWithLockEx(ctx, key, getter, ttl)
38+
}
39+
40+
// Give each fake block a nonzero bloom and an actual log to verify that
41+
// outage recovery preserves logs, rather than just marking blocks ready.
42+
type cacheTestProvider struct {
43+
*fakeProvider
44+
log types.Log
45+
}
46+
47+
func (p *cacheTestProvider) RawBlockByNumber(ctx context.Context, num *big.Int) (json.RawMessage, error) {
48+
payload, err := p.fakeProvider.RawBlockByNumber(ctx, num)
49+
if err != nil {
50+
return nil, err
51+
}
52+
bloom := types.CreateBloom(&types.Receipt{Logs: []*types.Log{&p.log}})
53+
return json.RawMessage(strings.Replace(string(payload), strings.Repeat("0", 512), common.Bytes2Hex(bloom[:]), 1)), nil
54+
}
55+
56+
func (p *cacheTestProvider) RawFilterLogs(ctx context.Context, q ethereum.FilterQuery) (json.RawMessage, error) {
57+
if err := p.chain.wait(ctx); err != nil {
58+
return nil, err
59+
}
60+
block, ok := p.chain.byHash(*q.BlockHash)
61+
if !ok {
62+
return nil, ethereum.NotFound
63+
}
64+
log := p.log
65+
log.BlockHash, log.BlockNumber = block.hash, block.num
66+
return json.Marshal([]types.Log{log})
67+
}
68+
69+
func TestMonitorCacheOutageWithLogs(t *testing.T) {
70+
for _, concurrency := range []int{0, 2} {
71+
t.Run(fmt.Sprintf("prefetch=%d", concurrency), func(t *testing.T) {
72+
chain := newFakeChain(1000, 40, 0)
73+
chain.hold(1020)
74+
provider := &cacheTestProvider{
75+
fakeProvider: &fakeProvider{MockRawInterface: mocks.NewMockRawInterface(gomock.NewController(t)), chain: chain},
76+
log: types.Log{Address: common.HexToAddress("0x1234"), Topics: []common.Hash{common.HexToHash("0xabcd")}, Data: []byte{1, 2, 3}},
77+
}
78+
mem, err := memcache.NewBackend(512)
79+
require.NoError(t, err)
80+
backend := &outageBackend{Backend: mem}
81+
backend.down.Store(true)
82+
opts := ethmonitor.DefaultOptions
83+
opts.PollingInterval = 5 * time.Millisecond
84+
opts.Timeout = time.Second
85+
opts.WithLogs = true
86+
opts.BlockRetentionLimit = 5 // Queue capacity is 10; publish 20 blocks while down.
87+
opts.StartBlockNumber = big.NewInt(1000)
88+
opts.PrefetchConcurrency = concurrency
89+
opts.CacheBackend = backend
90+
monitor, err := ethmonitor.NewMonitor(provider, opts)
91+
require.NoError(t, err)
92+
sub := monitor.Subscribe()
93+
defer sub.Unsubscribe()
94+
done := runMonitorForTest(t, monitor)
95+
timer := time.NewTimer(3 * time.Second)
96+
defer timer.Stop()
97+
next := uint64(1000)
98+
for next <= 1039 {
99+
select {
100+
case blocks := <-sub.Blocks():
101+
for _, block := range blocks {
102+
require.Equal(t, next, block.NumberU64())
103+
require.Equal(t, ethmonitor.Added, block.Event)
104+
require.True(t, block.OK)
105+
require.NotZero(t, block.Bloom())
106+
log := provider.log
107+
log.BlockHash, log.BlockNumber = block.Hash(), next
108+
require.Equal(t, []types.Log{log}, block.Logs)
109+
next++
110+
if next == 1020 {
111+
require.True(t, backend.down.Load())
112+
require.GreaterOrEqual(t, backend.failures.Load(), int64(40))
113+
backend.down.Store(false)
114+
chain.release(1020)
115+
}
116+
}
117+
case err := <-done:
118+
t.Fatalf("Run returned during cache outage/recovery: %v (head=%v)", err, monitor.LatestBlockNum())
119+
case <-timer.C:
120+
t.Fatalf("monitor stopped publishing at block %d", next-1)
121+
}
122+
}
123+
// After recovery, ordinary reads fill the cache again without a restart.
124+
key := ethmonitor.CacheKeyBlockLogs(big.NewInt(1), chain.hashAt(1039), nil)
125+
_, found, err := backend.Get(context.Background(), key)
126+
require.NoError(t, err)
127+
require.True(t, found)
128+
require.True(t, monitor.IsRunning())
129+
})
130+
}
131+
}

‎ethmonitor/canonical_state_test.go‎

Lines changed: 179 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,179 @@
1+
package ethmonitor
2+
3+
import (
4+
"encoding/json"
5+
"fmt"
6+
"math/big"
7+
"testing"
8+
9+
"github.com/0xsequence/ethkit/go-ethereum/common"
10+
"github.com/0xsequence/ethkit/go-ethereum/core/types"
11+
"github.com/stretchr/testify/require"
12+
)
13+
14+
func canonicalTestBlock(num int64) *Block {
15+
return &Block{
16+
Block: types.NewBlockWithHeader(&types.Header{
17+
Number: big.NewInt(num),
18+
BlockHash: common.BigToHash(big.NewInt(num)),
19+
ParentHash: common.BigToHash(big.NewInt(num - 1)),
20+
Time: uint64(num),
21+
}),
22+
Event: Added,
23+
OK: true,
24+
}
25+
}
26+
27+
func pushCanonicalTestBlock(t *testing.T, chain *Chain, input *Block) *Block {
28+
t.Helper()
29+
chain.push(input)
30+
block := chain.Head()
31+
require.NotNil(t, block)
32+
require.Equal(t, input.Hash(), block.Hash())
33+
return block
34+
}
35+
36+
// The interface keeps the before-fix proof executable without the new method.
37+
func canonicalTestState(t *testing.T, block *Block) (uint64, bool) {
38+
t.Helper()
39+
state, ok := any(block).(interface{ CanonicalState() (uint64, bool) })
40+
if !ok {
41+
return 0, false
42+
}
43+
return state.CanonicalState()
44+
}
45+
46+
func TestBlockCanonicalStateRetention(t *testing.T) {
47+
for _, depth := range []int64{1, 3} {
48+
t.Run(fmt.Sprintf("evictionDepth=%d", depth), func(t *testing.T) {
49+
chain := newChain(10, false)
50+
block := pushCanonicalTestBlock(t, chain, canonicalTestBlock(100))
51+
copy := chain.Blocks().Copy()[0]
52+
incarnation, canonical := canonicalTestState(t, block)
53+
require.Positive(t, incarnation)
54+
require.True(t, canonical)
55+
for num := int64(101); num <= 109+depth; num++ {
56+
pushCanonicalTestBlock(t, chain, canonicalTestBlock(num))
57+
}
58+
require.Nil(t, chain.GetBlock(block.Hash()))
59+
for _, snapshot := range []*Block{block, copy} {
60+
got, canonical := canonicalTestState(t, snapshot)
61+
require.Equal(t, incarnation, got)
62+
require.True(t, canonical, "retention eviction was mistaken for removal")
63+
}
64+
})
65+
}
66+
}
67+
68+
func TestBlockCanonicalStateReadoption(t *testing.T) {
69+
chain := newChain(10, false)
70+
input := canonicalTestBlock(100)
71+
block := pushCanonicalTestBlock(t, chain, input)
72+
shallow := *block
73+
copy := chain.Blocks().Copy()[0]
74+
incarnation, canonical := canonicalTestState(t, block)
75+
require.Positive(t, incarnation)
76+
require.True(t, canonical)
77+
removed := *chain.pop()
78+
removed.Event = Removed
79+
for _, snapshot := range []*Block{block, &shallow, copy, &removed} {
80+
got, canonical := canonicalTestState(t, snapshot)
81+
require.Equal(t, incarnation, got)
82+
require.False(t, canonical, "snapshot did not observe the actual removal")
83+
}
84+
85+
// Reusing either the original input or a removed snapshot must create a fresh
86+
// owned incarnation without mutating old Added/Removed event copies.
87+
for _, reused := range []*Block{input, copy} {
88+
fresh := pushCanonicalTestBlock(t, chain, reused)
89+
got, canonical := canonicalTestState(t, fresh)
90+
require.Greater(t, got, incarnation)
91+
require.True(t, canonical)
92+
for _, snapshot := range []*Block{block, &shallow, copy, &removed} {
93+
got, canonical := canonicalTestState(t, snapshot)
94+
require.Equal(t, incarnation, got)
95+
require.False(t, canonical, "fresh readoption revived an old event")
96+
}
97+
chain.pop()
98+
}
99+
}
100+
101+
func TestBlockCanonicalStateBootstrap(t *testing.T) {
102+
for _, count := range []int{1, 3} {
103+
for _, serialized := range []bool{false, true} {
104+
t.Run(fmt.Sprintf("blocks=%d/JSON=%v", count, serialized), func(t *testing.T) {
105+
inputs := make(Blocks, count)
106+
for i := range inputs {
107+
inputs[i] = canonicalTestBlock(int64(100 + i))
108+
}
109+
chain := newChain(10, true)
110+
if serialized {
111+
data, err := json.Marshal(inputs)
112+
require.NoError(t, err)
113+
require.NoError(t, chain.BootstrapFromBlocksJSON(data))
114+
} else {
115+
require.NoError(t, chain.BootstrapFromBlocks(inputs))
116+
}
117+
for _, block := range chain.Blocks().Copy() {
118+
incarnation, canonical := canonicalTestState(t, block)
119+
require.Positive(t, incarnation)
120+
require.True(t, canonical)
121+
data, err := json.Marshal(block)
122+
require.NoError(t, err)
123+
require.NotContains(t, string(data), "incarnation")
124+
require.NoError(t, json.Unmarshal(data, block))
125+
incarnation, canonical = canonicalTestState(t, block)
126+
require.Zero(t, incarnation, "serialized state preserved runtime ownership")
127+
require.False(t, canonical)
128+
}
129+
})
130+
}
131+
}
132+
}
133+
134+
func TestBlockCanonicalStateConcurrentRemoval(t *testing.T) {
135+
chain := newChain(10, false)
136+
input := canonicalTestBlock(100)
137+
block := pushCanonicalTestBlock(t, chain, input)
138+
incarnation, _ := canonicalTestState(t, block)
139+
require.Positive(t, incarnation)
140+
state := any(block).(interface{ CanonicalState() (uint64, bool) })
141+
start, done := make(chan struct{}), make(chan struct{})
142+
result := make(chan error, 1)
143+
go func() {
144+
close(start)
145+
removed := false
146+
for {
147+
got, canonical := state.CanonicalState()
148+
if got != incarnation || (removed && canonical) {
149+
result <- fmt.Errorf("old incarnation changed or revived: id=%d canonical=%v", got, canonical)
150+
return
151+
}
152+
removed = removed || !canonical
153+
select {
154+
case <-done:
155+
result <- nil
156+
return
157+
default:
158+
}
159+
}
160+
}()
161+
<-start
162+
for i := 0; i < 100; i++ {
163+
chain.pop()
164+
pushCanonicalTestBlock(t, chain, input)
165+
}
166+
close(done)
167+
require.NoError(t, <-result)
168+
got, canonical := state.CanonicalState()
169+
require.Equal(t, incarnation, got)
170+
require.False(t, canonical)
171+
}
172+
173+
func TestBlockCanonicalStateUntracked(t *testing.T) {
174+
for _, block := range []*Block{nil, canonicalTestBlock(100)} {
175+
incarnation, canonical := canonicalTestState(t, block)
176+
require.Zero(t, incarnation)
177+
require.False(t, canonical)
178+
}
179+
}

0 commit comments

Comments
 (0)