From a8c36a96fce5ea8409bcadb89276eb13f95645cd Mon Sep 17 00:00:00 2001 From: py-zoid Date: Wed, 21 Oct 2026 10:22:50 +0400 Subject: [PATCH 1/7] fix(ethmonitor): handle Avalanche settled block headers (ACP-194) Avalanche Helicon headers carry settledHeight, and their logsBloom covers the blocks they settle rather than the block's own logs. Detect them from the raw block JSON, stop using the bloom to validate their logs, and check an empty eth_getLogs result against the block's receipts, which wait for execution, using the receipt logs when they disagree. eth_getBlockReceipts is tried first; per-transaction receipts are fetched in bounded batches when the node doesn't support it, or for a single call when it answers with a non-2xx error. --- ethmonitor/ethmonitor.go | 63 ++++++++++-- ethmonitor/fetch_test.go | 4 +- ethmonitor/prefetch.go | 2 +- ethmonitor/settled.go | 200 +++++++++++++++++++++++++++++++++++++++ 4 files changed, 261 insertions(+), 8 deletions(-) create mode 100644 ethmonitor/settled.go diff --git a/ethmonitor/ethmonitor.go b/ethmonitor/ethmonitor.go index 311b61b6..0954e656 100644 --- a/ethmonitor/ethmonitor.go +++ b/ethmonitor/ethmonitor.go @@ -161,6 +161,14 @@ type Monitor struct { // event or, in polling mode, the prefetcher's head poll. latestHead atomic.Uint64 + // settledHeaders is set once a block header carrying settledHeight is seen. + settledHeaders atomic.Bool + + // blockReceiptsUnsupported is set once the node answers eth_getBlockReceipts + // with a method-not-found error, after which receiptLogs uses + // per-transaction receipts. + blockReceiptsUnsupported atomic.Bool + cache cachestore.Store[[]byte] prefetch *prefetcher @@ -820,10 +828,11 @@ func (m *Monitor) addLogs(ctx context.Context, blocks Blocks) { blockHash := block.Hash() - logs, _, err := m.filterLogs(tctx, blockHash, m.logTopics(), block.Bloom()) + logs, _, err := m.filterLogs(tctx, block.Block, m.logTopics()) if err == nil { - // filterLogs validates emptiness against the bloom and query topics. + // filterLogs validates emptiness against the bloom, or against + // receipts on settled-header chains, and the query topics. block.Logs = logs block.OK = true continue @@ -849,11 +858,14 @@ func (m *Monitor) logTopics() [][]common.Hash { return topics } -func (m *Monitor) filterLogs(ctx context.Context, blockHash common.Hash, topics [][]common.Hash, blockBloom types.Bloom) ([]types.Log, []byte, error) { +func (m *Monitor) filterLogs(ctx context.Context, block *types.Block, topics [][]common.Hash) ([]types.Log, []byte, error) { var fetchedLogs []types.Log + blockHash := block.Hash() // The block bloom covers all logs, so unrelated topics may set it even - // when a topic-filtered query correctly returns no logs. - expectLogs := len(topics) == 0 && blockBloom != (types.Bloom{}) + // when a topic-filtered query correctly returns no logs. Settled headers + // carry the bloom of other blocks, so it says nothing about this block. + settled := m.usesSettledHeaders() + expectLogs := len(topics) == 0 && !settled && block.Bloom() != (types.Bloom{}) getter := func(ctx context.Context, _ string) ([]byte, error) { if m.options.DebugLogging { @@ -872,10 +884,36 @@ func (m *Monitor) filterLogs(ctx context.Context, blockHash common.Hash, topics } // Validate before caching so a malformed response cannot block log // backfilling until cache expiry. - fetchedLogs, err = m.unmarshalLogs(logsPayload, expectLogs) + logs, err := m.unmarshalLogs(logsPayload, expectLogs) if err != nil { return nil, err } + + // On settled-header chains eth_getLogs does not wait for execution, so + // an empty result may be premature. Receipt calls do wait, so confirm + // it against them before it can be cached. + if settled && len(topics) == 0 && len(logs) == 0 { + rctx, rcancel := context.WithTimeout(ctx, 4*time.Second) + defer rcancel() + + receiptLogs, err := m.receiptLogs(rctx, block) + if err != nil { + return nil, fmt.Errorf("%w: %w", errReceiptsFailed, err) + } + if len(receiptLogs) > 0 { + m.log.Warn(fmt.Sprintf("ethmonitor: settled block %d %s: eth_getLogs returned no logs but receipts have %d, using receipt logs", block.NumberU64(), blockHash.Hex(), len(receiptLogs))) + // Cache the receipt logs so a cache hit replays them. + logsPayload, err = json.Marshal(receiptLogs) + if err != nil { + return nil, err + } + logs = receiptLogs + } else if m.options.DebugLogging { + m.log.Debug(fmt.Sprintf("ethmonitor: settled block %s: empty logs confirmed via receipts", blockHash.Hex())) + } + } + + fetchedLogs = logs return logsPayload, nil } @@ -1442,9 +1480,22 @@ func (m *Monitor) unmarshalBlock(blockPayload []byte) (*types.Block, error) { if err != nil { return nil, err } + + if !m.settledHeaders.Load() && hasSettledHeight(blockPayload) { + if m.settledHeaders.CompareAndSwap(false, true) { + m.log.Info("ethmonitor: settled block headers detected (settledHeight present), header logsBloom will not be used to validate logs") + } + } return block, nil } +// usesSettledHeaders reports whether settled block headers (ACP-194 / Avalanche +// Helicon) were detected, in which case header logsBloom does not describe the +// block's own logs and must not be trusted to validate them. +func (m *Monitor) usesSettledHeaders() bool { + return m.settledHeaders.Load() +} + func (m *Monitor) unmarshalLogs(logsPayload []byte, expectLogs bool) ([]types.Log, error) { var logs []types.Log err := json.Unmarshal(logsPayload, &logs) diff --git a/ethmonitor/fetch_test.go b/ethmonitor/fetch_test.go index 96d2479e..8e3eb461 100644 --- a/ethmonitor/fetch_test.go +++ b/ethmonitor/fetch_test.go @@ -196,7 +196,9 @@ func TestMonitorCacheFallback(t *testing.T) { var response []byte if method == "logs" { var logs []types.Log - logs, response, err = monitor.filterLogs(ctx, hash, nil, types.CreateBloom(&types.Receipt{Logs: []*types.Log{&log}})) + block := types.NewBlockWithHeader(&types.Header{Bloom: types.CreateBloom(&types.Receipt{Logs: []*types.Log{&log}})}) + block.SetHash(hash) + logs, response, err = monitor.filterLogs(ctx, block, nil) if failure == "backend_failure" { require.Equal(t, []types.Log{log}, logs) } else { diff --git a/ethmonitor/prefetch.go b/ethmonitor/prefetch.go index 5709105f..7bc5c4b7 100644 --- a/ethmonitor/prefetch.go +++ b/ethmonitor/prefetch.go @@ -279,7 +279,7 @@ func (p *prefetcher) fetch(ctx context.Context, job prefetchJob) { // 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 { + if _, _, err := m.filterLogs(tctx, block, m.logTopics()); err != nil { m.log.Debug(fmt.Sprintf("ethmonitor: prefetch of logs for block %d failed: %v", job.num, err)) } } diff --git a/ethmonitor/settled.go b/ethmonitor/settled.go new file mode 100644 index 00000000..721f4200 --- /dev/null +++ b/ethmonitor/settled.go @@ -0,0 +1,200 @@ +package ethmonitor + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "strings" + + "github.com/0xsequence/ethkit/ethrpc" + "github.com/0xsequence/ethkit/ethrpc/jsonrpc" + "github.com/0xsequence/ethkit/go-ethereum/common" + "github.com/0xsequence/ethkit/go-ethereum/common/hexutil" + "github.com/0xsequence/ethkit/go-ethereum/core/types" + "github.com/goware/superr" +) + +// hasSettledHeight reports whether a raw block payload carries a non-null +// top-level settledHeight field (ACP-194 / Avalanche Helicon). +func hasSettledHeight(payload []byte) bool { + if !bytes.Contains(payload, []byte(`"settledHeight"`)) { + return false + } + var header struct { + SettledHeight json.RawMessage `json:"settledHeight"` + } + if err := json.Unmarshal(payload, &header); err != nil { + return false + } + return len(header.SettledHeight) > 0 && !bytes.Equal(bytes.TrimSpace(header.SettledHeight), []byte("null")) +} + +// receiptsBatchSize bounds each batch of per-transaction receipt calls, since +// RPC gateways and providers cap batch sizes (some reject batches over 40). +const receiptsBatchSize = 20 + +// errReceiptsFailed marks a filterLogs failure where the block's receipts +// were already fetched, and failed, in the same call. +var errReceiptsFailed = errors.New("ethmonitor: receipts failed") + +// logsReceipt is the subset of a transaction receipt that receiptLogs needs. +// It is not decoded into types.Receipt, as chain-specific receipt fields may +// fail its strict decoding. +type logsReceipt struct { + Type hexutil.Uint64 `json:"type"` + BlockHash common.Hash `json:"blockHash"` + TransactionHash common.Hash `json:"transactionHash"` + Logs []types.Log `json:"logs"` +} + +// receiptLogs returns the block's logs as recorded in its receipts, which, +// unlike eth_getLogs on settled-header chains, wait for execution. It is +// independent of the settled-header flag. Every failure returns an error. +func (m *Monitor) receiptLogs(ctx context.Context, block *types.Block) ([]types.Log, error) { + txs := block.Transactions() + blockHash := block.Hash() + + // ethrpc drops transaction types it can't decode, so an empty transaction + // list only means an empty block when the header's root agrees. A zero + // root means the node did not report one. + if len(txs) == 0 && (block.TxHash() == types.EmptyTxsHash || block.TxHash() == (common.Hash{})) { + return []types.Log{}, nil + } + + if !m.blockReceiptsUnsupported.Load() { + var receipts []*logsReceipt + _, err := m.provider.Do(ctx, ethrpc.NewCallBuilder[[]*logsReceipt]("eth_getBlockReceipts", nil, blockHash).Into(&receipts)) + switch { + case err == nil: + if err := checkBlockReceipts(receipts, blockHash, len(txs)); err != nil { + return nil, fmt.Errorf("ethmonitor: eth_getBlockReceipts for block %d %s: %w", block.NumberU64(), blockHash.Hex(), err) + } + return flattenReceiptLogs(receipts), nil + case isMethodNotFoundError(err): + // The node doesn't support the method, so stop asking. + if m.blockReceiptsUnsupported.CompareAndSwap(false, true) { + m.log.Info(fmt.Sprintf("ethmonitor: eth_getBlockReceipts is not supported (%v), using eth_getTransactionReceipt from now on", err)) + } + case isHTTPJSONRPCError(err): + // A non-2xx response with a JSON-RPC error body, for example a + // gateway rejecting the method or rate limiting. It may be + // transient, so fall back without memoising. + default: + return nil, fmt.Errorf("ethmonitor: eth_getBlockReceipts for block %d %s: %w", block.NumberU64(), blockHash.Hex(), err) + } + } + + // Fall back to one receipt per transaction, in batches of at most + // receiptsBatchSize calls. The block's transaction count is fetched + // alongside, as receipts can only be requested for transactions ethrpc + // decoded. + var txCount *hexutil.Uint + receipts := make([]*logsReceipt, len(txs)) + calls := make([]ethrpc.Call, 0, len(txs)+1) + calls = append(calls, ethrpc.NewCallBuilder[*hexutil.Uint]("eth_getBlockTransactionCountByHash", nil, blockHash).Into(&txCount)) + for i, tx := range txs { + calls = append(calls, ethrpc.NewCallBuilder[*logsReceipt]("eth_getTransactionReceipt", nil, tx.Hash()).Into(&receipts[i])) + } + for start := 0; start < len(calls); start += receiptsBatchSize { + end := min(start+receiptsBatchSize, len(calls)) + if _, err := m.provider.Do(ctx, calls[start:end]...); err != nil { + return nil, fmt.Errorf("ethmonitor: eth_getTransactionReceipt for block %d %s: %w", block.NumberU64(), blockHash.Hex(), err) + } + } + if txCount == nil { + return nil, fmt.Errorf("ethmonitor: eth_getBlockTransactionCountByHash for block %d %s returned null", block.NumberU64(), blockHash.Hex()) + } + if int(*txCount) != len(txs) { + return nil, fmt.Errorf("ethmonitor: block %d %s has %d transactions but %d were decoded, per-transaction receipts would miss logs", block.NumberU64(), blockHash.Hex(), uint(*txCount), len(txs)) + } + for i, receipt := range receipts { + if receipt == nil { + return nil, fmt.Errorf("ethmonitor: receipt for transaction %s in block %d %s is null (not executed yet)", txs[i].Hash().Hex(), block.NumberU64(), blockHash.Hex()) + } + if receipt.BlockHash != blockHash { + return nil, fmt.Errorf("ethmonitor: receipt for transaction %s has block hash %s, expected block %d %s", txs[i].Hash().Hex(), receipt.BlockHash.Hex(), block.NumberU64(), blockHash.Hex()) + } + } + return flattenReceiptLogs(receipts), nil +} + +// checkBlockReceipts validates an eth_getBlockReceipts result against the +// block. Receipts beyond the decoded transactions are allowed only for +// transaction types ethrpc does not decode. +func checkBlockReceipts(receipts []*logsReceipt, blockHash common.Hash, numTxs int) error { + // Only reached for blocks with transactions, so null (unknown or not yet + // executed block) and [] are both errors. + if len(receipts) == 0 { + return fmt.Errorf("no receipts for a block with transactions") + } + decodable := 0 + for i, receipt := range receipts { + if receipt == nil { + return fmt.Errorf("receipt %d is null", i) + } + if receipt.BlockHash != blockHash { + return fmt.Errorf("receipt %d has block hash %s", i, receipt.BlockHash.Hex()) + } + if receipt.Type <= types.DynamicFeeTxType { + decodable++ + } + } + if len(receipts) < numTxs || decodable > numTxs { + return fmt.Errorf("%d receipts for %d transactions", len(receipts), numTxs) + } + return nil +} + +// flattenReceiptLogs returns the receipts' logs in receipt order, which is +// logIndex order, as a non-nil slice. +func flattenReceiptLogs(receipts []*logsReceipt) []types.Log { + logs := []types.Log{} + for _, receipt := range receipts { + logs = append(logs, receipt.Logs...) + } + return logs +} + +// isMethodNotFoundError reports whether err is a JSON-RPC error returned by the +// node in a successful HTTP response saying the method isn't supported: code +// -32601, or a message about the method itself, as some nodes answer with a +// generic code (e.g. "the method eth_getBlockReceipts does not exist/is not +// available"). Any other JSON-RPC error may be transient (a timeout, a missing +// trie node, a lagging backend answering that a new block does not exist), so +// it is returned to the caller to be retried instead. +func isMethodNotFoundError(err error) bool { + var rpcErr jsonrpc.Error + if !errors.As(err, &rpcErr) || errors.Is(err, ethrpc.ErrRequestFail) { + return false + } + if rpcErr.Code == -32601 { + return true + } + msg := strings.ToLower(rpcErr.Message) + if !strings.Contains(msg, "method") { + return false + } + for _, s := range []string{"not found", "does not exist", "not supported", "unsupported", "not available"} { + if strings.Contains(msg, s) { + return true + } + } + return false +} + +// isHTTPJSONRPCError reports whether err is a non-2xx HTTP response whose body +// carried a JSON-RPC error object. ethrpc wraps those with superr, whose +// stack is not reachable through errors.As. +func isHTTPJSONRPCError(err error) bool { + if !errors.Is(err, ethrpc.ErrRequestFail) { + return false + } + for _, e := range superr.GetErrorStack(err) { + if _, ok := e.(*jsonrpc.Error); ok { + return true + } + } + return false +} From 95a6ca944e01aba836736ed2eb2c9b27c04dcff2 Mon Sep 17 00:00:00 2001 From: py-zoid Date: Wed, 21 Oct 2026 10:24:49 +0400 Subject: [PATCH 2/7] feat(ethmonitor): resolve stuck blocks from receipts instead of failing on a full queue A block whose logs keep failing used to hold the publish queue until it overflowed into ErrFatal(ErrQueueFull), on any chain. After MaxLogsAttempts (default 10) failed fetches its logs are now taken from its receipts; outside settled-header chains, empty receipts for a block with a set bloom are rejected. A block still unresolved when it leaves the chain cache is published with empty logs, with an Error log and an alert, instead of stopping the monitor. This changes behaviour for every consumer: queue overflow is no longer fatal by default. Options.StrictLogs restores the previous behaviour for consumers that would rather stop than publish unconfirmed logs. --- ethmonitor/chain.go | 3 + ethmonitor/ethmonitor.go | 31 ++++++++++ ethmonitor/prefetch_test.go | 1 + ethmonitor/pubsub.go | 16 ++++++ ethmonitor/resolve.go | 112 ++++++++++++++++++++++++++++++++++++ 5 files changed, 163 insertions(+) create mode 100644 ethmonitor/resolve.go diff --git a/ethmonitor/chain.go b/ethmonitor/chain.go index 7233b5d0..353a5dc2 100644 --- a/ethmonitor/chain.go +++ b/ethmonitor/chain.go @@ -244,6 +244,9 @@ type Block struct { OK bool canonicalState *blockCanonicalState + + // logsAttempts counts failed log fetches for this block instance. + logsAttempts int } type blockCanonicalState struct { diff --git a/ethmonitor/ethmonitor.go b/ethmonitor/ethmonitor.go index 0954e656..d20cd96e 100644 --- a/ethmonitor/ethmonitor.go +++ b/ethmonitor/ethmonitor.go @@ -45,6 +45,8 @@ var DefaultOptions = Options{ PrefetchConcurrency: 0, // Prefetching is disabled by default PrefetchWindow: 0, // 4x PrefetchConcurrency set on init Alerter: util.NoopAlerter(), + MaxLogsAttempts: 10, + StrictLogs: false, // publish unresolved blocks without logs once evicted } type Options struct { @@ -125,6 +127,18 @@ type Options struct { // Alerter config via github.com/goware/alerter Alerter util.Alerter + // MaxLogsAttempts is the number of failed eth_getLogs attempts for a block, + // after which its logs are taken from its receipts. Defaults to 10. + MaxLogsAttempts int + + // StrictLogs, when true, never publishes a block whose logs can't be + // confirmed by eth_getLogs or its receipts. The monitor keeps retrying and + // returns ErrFatal(ErrQueueFull) if the publish queue overflows, which was + // the behaviour before this option existed. When false (the default), such + // a block is published with empty logs once it leaves the chain cache + // (BlockRetentionLimit), with an Error log and an alert. + StrictLogs bool + // DebugLogging toggle DebugLogging bool } @@ -207,6 +221,10 @@ 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.MaxLogsAttempts <= 0 { + opts.MaxLogsAttempts = 10 + } + if opts.PrefetchConcurrency < 0 { opts.PrefetchConcurrency = 0 } @@ -701,6 +719,7 @@ func (m *Monitor) monitor(ctx context.Context, wg *sync.WaitGroup) error { if m.options.WithLogs { m.addLogs(ctx, events) m.backfillChainLogs(ctx, events) + m.resolveEvictedBlocks(ctx) } else { for _, b := range events { b.Logs = nil // nil it out to be clear to subscribers @@ -838,6 +857,18 @@ func (m *Monitor) addLogs(ctx context.Context, blocks Blocks) { continue } + // After MaxLogsAttempts failed fetches, take the logs from the block's + // receipts instead, which are authoritative. Skip it when filterLogs + // already failed on receipts in this attempt. + block.logsAttempts++ + if block.logsAttempts >= m.options.MaxLogsAttempts && !errors.Is(err, errReceiptsFailed) { + rerr := m.resolveFromReceipts(tctx, block) + if rerr == nil { + continue + } + err = fmt.Errorf("%w (receipts: %v)", err, rerr) + } + // mark for backfilling block.Logs = nil block.OK = false diff --git a/ethmonitor/prefetch_test.go b/ethmonitor/prefetch_test.go index d1713576..599463c5 100644 --- a/ethmonitor/prefetch_test.go +++ b/ethmonitor/prefetch_test.go @@ -477,6 +477,7 @@ func TestMonitorPrefetchFatalExitStopsWorkers(t *testing.T) { initial := newTestMonitor(t, chain, false, 4, 0, &invalidLogsBackend{backend}) opts := initial.Options() opts.BlockRetentionLimit = 2 // Four queued events; failed logs prevent dequeue. + opts.StrictLogs = true // the fatal queue overflow is opt-in monitor, err := ethmonitor.NewMonitor(initial.Provider().(ethrpc.RawInterface), opts) require.NoError(t, err) sub := monitor.Subscribe("TestMonitorPrefetchFatalExitStopsWorkers") diff --git a/ethmonitor/pubsub.go b/ethmonitor/pubsub.go index d6a2af9b..a366fd81 100644 --- a/ethmonitor/pubsub.go +++ b/ethmonitor/pubsub.go @@ -150,6 +150,22 @@ func (c *queue) dequeue(maxBlockNum uint64) (Blocks, bool) { return events, true } +// unresolved returns the queued Added events which are not ready and are +// numbered below the given block number, in queue order. The blocks are shared +// with the chain and are resolved in place by the monitor loop. +func (c *queue) unresolved(below uint64) Blocks { + c.mu.Lock() + defer c.mu.Unlock() + + var blocks Blocks + for _, ev := range c.events { + if ev.Event == Added && !ev.OK && ev.NumberU64() < below { + blocks = append(blocks, ev) + } + } + return blocks +} + func (c *queue) head() *Block { c.mu.Lock() defer c.mu.Unlock() diff --git a/ethmonitor/resolve.go b/ethmonitor/resolve.go new file mode 100644 index 00000000..59325085 --- /dev/null +++ b/ethmonitor/resolve.go @@ -0,0 +1,112 @@ +package ethmonitor + +import ( + "context" + "fmt" + "slices" + "time" + + "github.com/0xsequence/ethkit/go-ethereum/common" + "github.com/0xsequence/ethkit/go-ethereum/core/types" +) + +// resolveFromReceipts sets the block's logs from its receipts, filtered by the +// monitor's log topics, and marks it ready. It doesn't write to the cache. On +// failure the block is left unchanged. +func (m *Monitor) resolveFromReceipts(ctx context.Context, block *Block) error { + tctx, cancel := context.WithTimeout(ctx, 4*time.Second) + defer cancel() + + logs, err := m.receiptLogs(tctx, block.Block) + if err != nil { + return err + } + + // Outside settled-header chains the header bloom covers the block's own + // logs, so receipts without any logs contradict a non-zero bloom, the same + // way an empty eth_getLogs result does. + if len(logs) == 0 && !m.usesSettledHeaders() && block.Bloom() != (types.Bloom{}) { + return fmt.Errorf("ethmonitor: block %d %s: receipts have no logs but block bloom is set", block.NumberU64(), block.Hash().Hex()) + } + + // apply the same topic filter as the eth_getLogs query + topics := m.logTopics() + if len(topics) > 0 { + filtered := make([]types.Log, 0, len(logs)) + for _, log := range logs { + if matchTopics(log.Topics, topics) { + filtered = append(filtered, log) + } + } + logs = filtered + } + + block.Logs = logs + block.OK = true + + m.log.Warn(fmt.Sprintf("ethmonitor: block %d %s: eth_getLogs failed %d times, using logs from receipts", block.NumberU64(), block.Hash().Hex(), block.logsAttempts)) + return nil +} + +// matchTopics reports whether log topics match an eth_getLogs topic filter, +// as geth's filter does: each position matches any of its values, an empty +// position matches anything, and the log needs a topic for every position. +func matchTopics(logTopics []common.Hash, topics [][]common.Hash) bool { + if len(topics) > len(logTopics) { + return false + } + for i, want := range topics { + if len(want) > 0 && !slices.Contains(want, logTopics[i]) { + return false + } + } + return true +} + +// resolveEvictedBlocks resolves queued blocks whose logs are still missing and +// which have left the chain cache, where backfillChainLogs no longer retries +// them. They are taken from receipts, or, unless StrictLogs is set, published +// with empty logs, so they can't hold up the publish queue. +// +// NOTE: called with m.chain.mu held, so it reads m.chain.blocks directly and +// must not call Chain methods which take the lock. +func (m *Monitor) resolveEvictedBlocks(ctx context.Context) { + if len(m.chain.blocks) == 0 { + return + } + oldest := m.chain.blocks[0].NumberU64() + + for _, block := range m.publishQueue.unresolved(oldest) { + select { + case <-ctx.Done(): + return + default: + } + + // A reorg deeper than the chain cache can pop a queued block below the + // oldest retained one. Its Removed event, published in this same loop + // iteration, drops it from the queue, so leave it alone. + if _, canonical := block.CanonicalState(); !canonical { + continue + } + + err := m.resolveFromReceipts(ctx, block) + if err == nil { + continue + } + + if m.options.StrictLogs { + // Blocks are published in order, so later ones can't be published + // before this one is resolved. Retry on the next iteration. + return + } + + block.Logs = []types.Log{} + block.OK = true + + m.log.Error(fmt.Sprintf("ethmonitor: block %d %s: publishing without logs, eth_getLogs and receipts both failed after %d attempts: %v", block.NumberU64(), block.Hash().Hex(), block.logsAttempts, err)) + if m.alert != nil { + m.alert.Alert(ctx, "ethmonitor: block %d %s: publishing without logs, eth_getLogs and receipts both failed after %d attempts: %v", block.NumberU64(), block.Hash().Hex(), block.logsAttempts, err) + } + } +} From 2a7e529761f3c600b59b7b896e5ea15cde86e461 Mon Sep 17 00:00:00 2001 From: py-zoid Date: Wed, 21 Oct 2026 10:26:16 +0400 Subject: [PATCH 3/7] test(ethmonitor): cover settled headers and unresolved blocks --- ethmonitor/logs_test.go | 389 +++++++++++++++++++++++++++++++++ ethmonitor/resolve_test.go | 428 +++++++++++++++++++++++++++++++++++++ ethmonitor/settled_test.go | 205 ++++++++++++++++++ 3 files changed, 1022 insertions(+) create mode 100644 ethmonitor/resolve_test.go create mode 100644 ethmonitor/settled_test.go diff --git a/ethmonitor/logs_test.go b/ethmonitor/logs_test.go index 305fe354..d4346799 100644 --- a/ethmonitor/logs_test.go +++ b/ethmonitor/logs_test.go @@ -1,19 +1,24 @@ package ethmonitor import ( + "bytes" "context" "encoding/json" "fmt" "math/big" "net/http" "net/http/httptest" + "sync" "sync/atomic" "testing" "github.com/0xsequence/ethkit/ethrpc" "github.com/0xsequence/ethkit/ethrpc/jsonrpc" "github.com/0xsequence/ethkit/go-ethereum/common" + "github.com/0xsequence/ethkit/go-ethereum/common/hexutil" "github.com/0xsequence/ethkit/go-ethereum/core/types" + "github.com/0xsequence/ethkit/go-ethereum/crypto" + memcache "github.com/goware/cachestore-mem" "github.com/stretchr/testify/require" ) @@ -120,3 +125,387 @@ func TestMonitorEmptyLogResponses(t *testing.T) { } } } + +// testNode is an httptest JSON-RPC node which counts the calls it serves per +// method. Batch requests are served call by call and counted the same way. +type testNode struct { + *httptest.Server + mu sync.Mutex + counts map[string]int + + maxBatch int // when set, larger batches are rejected with HTTP 400, as some RPC gateways do + largestBatch int +} + +func newTestNode(t *testing.T, handle func(method string, params []any) (json.RawMessage, *jsonrpc.Error)) *testNode { + t.Helper() + node := &testNode{counts: map[string]int{}} + node.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body bytes.Buffer + if _, err := body.ReadFrom(r.Body); err != nil { + t.Error(err) + return + } + var requests []jsonrpc.Message + batch := bytes.HasPrefix(bytes.TrimSpace(body.Bytes()), []byte("[")) + if batch { + if err := json.Unmarshal(body.Bytes(), &requests); err != nil { + t.Error(err) + return + } + } else { + var request jsonrpc.Message + if err := json.Unmarshal(body.Bytes(), &request); err != nil { + t.Error(err) + return + } + requests = append(requests, request) + } + + node.mu.Lock() + node.largestBatch = max(node.largestBatch, len(requests)) + maxBatch := node.maxBatch + node.mu.Unlock() + if maxBatch > 0 && len(requests) > maxBatch { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + if err := json.NewEncoder(w).Encode(jsonrpc.Message{Version: "2.0", Error: &jsonrpc.Error{Code: -32600, Message: fmt.Sprintf("too many batch requests, max is %d", maxBatch)}}); err != nil { + t.Error(err) + } + return + } + + responses := make([]jsonrpc.Message, 0, len(requests)) + for _, request := range requests { + node.mu.Lock() + node.counts[request.Method]++ + node.mu.Unlock() + result, rpcErr := handle(request.Method, request.Params) + responses = append(responses, jsonrpc.Message{Version: "2.0", ID: request.ID, Result: result, Error: rpcErr}) + } + + w.Header().Set("Content-Type", "application/json") + var err error + if batch { + err = json.NewEncoder(w).Encode(responses) + } else { + err = json.NewEncoder(w).Encode(responses[0]) + } + if err != nil { + t.Error(err) + } + })) + t.Cleanup(node.Close) + return node +} + +func (n *testNode) calls(method string) int { + n.mu.Lock() + defer n.mu.Unlock() + return n.counts[method] +} + +// receiptsBehavior configures what a receiptsNode answers. +type receiptsBehavior struct { + logs json.RawMessage // eth_getLogs result, [] by default + logsErr *jsonrpc.Error // eth_getLogs JSON-RPC error + receiptsErr *jsonrpc.Error // JSON-RPC error from every receipts method + blockRcpErr *jsonrpc.Error // JSON-RPC error from eth_getBlockReceipts only + + receiptLogs map[common.Hash][]types.Log // receipt logs by block hash, held by the first tx's receipt + wrongBlockHash bool // receipts carry another block's hash + nullTxReceipt bool // eth_getTransactionReceipt returns null + dropReceipts int // eth_getBlockReceipts omits this many receipts + txCountDelta int // eth_getBlockTransactionCountByHash is off by this much +} + +var errReceiptsDown = &jsonrpc.Error{Code: -32000, Message: "receipts down"} + +// newReceiptsNode serves eth_getLogs and the receipts methods for blocks. +func newReceiptsNode(t *testing.T, blocks []*types.Block, b *receiptsBehavior) *testNode { + t.Helper() + byHash := map[common.Hash]*types.Block{} + byTx := map[common.Hash]struct { + block *types.Block + index int + }{} + for _, block := range blocks { + byHash[block.Hash()] = block + for i, tx := range block.Transactions() { + byTx[tx.Hash()] = struct { + block *types.Block + index int + }{block, i} + } + } + + receipt := func(block *types.Block, i int) map[string]any { + blockHash := block.Hash() + if b.wrongBlockHash { + blockHash = common.HexToHash("0xdead") + } + logs := []types.Log{} + if i == 0 { + logs = b.receiptLogs[block.Hash()] + } + return map[string]any{ + "type": "0x0", + "blockHash": blockHash, + "transactionHash": block.Transactions()[i].Hash(), + "logs": logs, + } + } + hashParam := func(params []any) common.Hash { + s, _ := params[0].(string) + return common.HexToHash(s) + } + + return newTestNode(t, func(method string, params []any) (json.RawMessage, *jsonrpc.Error) { + switch method { + case "eth_getLogs": + if b.logsErr != nil { + return nil, b.logsErr + } + if b.logs != nil { + return b.logs, nil + } + return json.RawMessage(`[]`), nil + case "eth_getBlockReceipts", "eth_getTransactionReceipt", "eth_getBlockTransactionCountByHash": + if b.receiptsErr != nil { + return nil, b.receiptsErr + } + } + + var result any + switch method { + case "eth_getBlockReceipts": + if b.blockRcpErr != nil { + return nil, b.blockRcpErr + } + block, ok := byHash[hashParam(params)] + if !ok { + return json.RawMessage(`null`), nil + } + receipts := []map[string]any{} + for i := 0; i < len(block.Transactions())-b.dropReceipts; i++ { + receipts = append(receipts, receipt(block, i)) + } + result = receipts + case "eth_getTransactionReceipt": + found, ok := byTx[hashParam(params)] + if !ok || b.nullTxReceipt { + return json.RawMessage(`null`), nil + } + result = receipt(found.block, found.index) + case "eth_getBlockTransactionCountByHash": + block, ok := byHash[hashParam(params)] + if !ok { + return json.RawMessage(`null`), nil + } + result = hexutil.Uint(len(block.Transactions()) + b.txCountDelta) + default: + t.Errorf("unexpected RPC method %q", method) + return nil, &jsonrpc.Error{Code: -32601, Message: "method not found"} + } + encoded, err := json.Marshal(result) + if err != nil { + t.Error(err) + return nil, &jsonrpc.Error{Code: -32603, Message: err.Error()} + } + return encoded, nil + }) +} + +var testSigningKey, _ = crypto.HexToECDSA("4c0883a69102937d6231471b5dbb6204fe5129617082792ae468d01a3f362318") + +// newTxBlock builds a block with txCount signed transactions. A block with +// transactions gets a non-zero header TxHash, as a real one would. +func newTxBlock(t *testing.T, number uint64, txCount int, bloom types.Bloom) *types.Block { + t.Helper() + signer := types.LatestSignerForChainID(big.NewInt(1)) + txs := make([]*types.Transaction, 0, txCount) + for i := 0; i < txCount; i++ { + tx, err := types.SignTx(types.NewTx(&types.LegacyTx{ + Nonce: number*100 + uint64(i), + GasPrice: big.NewInt(1), + Gas: 21000, + To: &common.Address{1}, + Value: big.NewInt(1), + }), signer, testSigningKey) + require.NoError(t, err) + txs = append(txs, tx) + } + header := &types.Header{ + Number: new(big.Int).SetUint64(number), + ParentHash: crypto.Keccak256Hash(new(big.Int).SetUint64(number - 1).Bytes()), + Difficulty: big.NewInt(0), + Bloom: bloom, + } + header.SetHash(crypto.Keccak256Hash(new(big.Int).SetUint64(number).Bytes())) + if txCount > 0 { + header.TxHash = common.HexToHash("0xbeef") + } + return types.NewBlockWithHeader(header).WithBody(types.Body{Transactions: txs}) +} + +// newTestLog builds a log of block's first transaction, in the shape a node returns. +func newTestLog(block *types.Block, index uint, topic common.Hash) types.Log { + return types.Log{ + Address: common.HexToAddress("0x1234"), + Topics: []common.Hash{topic}, + Data: []byte{1, 2, 3}, + BlockNumber: block.NumberU64(), + TxHash: block.Transactions()[0].Hash(), + BlockHash: block.Hash(), + Index: index, + } +} + +var nonzeroBloom = types.BytesToBloom([]byte{1}) + +func TestSettledEmptyLogResponses(t *testing.T) { + topicA := common.HexToHash("0xaa") + + type counts struct{ getLogs, blockReceipts, txCount, txReceipt int } + cases := []struct { + name string + settled bool // when false, the monitor is not a settled-header one + bloom types.Bloom + txs int + blocks int // blocks to run addLogs for, 1 by default + cache bool + numLogs int // receipt logs per block + logs string + behavior receiptsBehavior + maxBatch int // node batch limit, such as an RPC gateway's 40 + ready bool + want counts // totals across all blocks + }{ + {name: "a_nonzero_bloom_empty_no_txs", settled: true, bloom: nonzeroBloom, ready: true, + want: counts{getLogs: 1}}, + {name: "b_nonzero_bloom_empty_two_txs_no_receipt_logs", settled: true, bloom: nonzeroBloom, txs: 2, ready: true, + want: counts{getLogs: 1, blockReceipts: 1}}, + {name: "c_zero_bloom_receipts_have_logs", settled: true, txs: 1, numLogs: 2, ready: true, + want: counts{getLogs: 1, blockReceipts: 1}}, + {name: "c_zero_bloom_receipts_have_logs_cached", settled: true, txs: 1, numLogs: 2, cache: true, ready: true, + want: counts{getLogs: 1, blockReceipts: 1}}, + {name: "d_block_receipts_unsupported_falls_back_and_is_memoised", settled: true, txs: 2, blocks: 2, ready: true, + behavior: receiptsBehavior{blockRcpErr: &jsonrpc.Error{Code: -32601, Message: "method not found"}}, + want: counts{getLogs: 2, blockReceipts: 1, txCount: 2, txReceipt: 4}}, + {name: "e_receipt_block_hash_mismatch", settled: true, txs: 1, numLogs: 1, + behavior: receiptsBehavior{wrongBlockHash: true}, + want: counts{getLogs: 1, blockReceipts: 1}}, + {name: "e_tx_receipt_block_hash_mismatch", settled: true, txs: 1, numLogs: 1, + behavior: receiptsBehavior{wrongBlockHash: true, blockRcpErr: &jsonrpc.Error{Code: -32601, Message: "method not found"}}, + want: counts{getLogs: 1, blockReceipts: 1, txCount: 1, txReceipt: 1}}, + {name: "f_null_tx_receipt", settled: true, txs: 2, + behavior: receiptsBehavior{nullTxReceipt: true, blockRcpErr: &jsonrpc.Error{Code: -32601, Message: "method not found"}}, + want: counts{getLogs: 1, blockReceipts: 1, txCount: 1, txReceipt: 2}}, + {name: "g_logs_returned_no_receipt_calls", settled: true, bloom: nonzeroBloom, txs: 1, numLogs: 1, ready: true, + logs: "log", want: counts{getLogs: 1}}, + {name: "h_not_settled_nonzero_bloom_empty", bloom: nonzeroBloom, txs: 1, numLogs: 1, + want: counts{getLogs: 1}}, + {name: "i_block_receipts_count_below_tx_count", settled: true, txs: 2, + behavior: receiptsBehavior{dropReceipts: 1}, + want: counts{getLogs: 1, blockReceipts: 1}}, + {name: "i_tx_count_differs_from_decoded_txs", settled: true, txs: 1, + behavior: receiptsBehavior{txCountDelta: 1, blockRcpErr: &jsonrpc.Error{Code: -32601, Message: "method not found"}}, + want: counts{getLogs: 1, blockReceipts: 1, txCount: 1, txReceipt: 1}}, + {name: "receipts_unavailable", settled: true, txs: 1, + behavior: receiptsBehavior{receiptsErr: errReceiptsDown}, + want: counts{getLogs: 1, blockReceipts: 1}}, + {name: "d_transient_block_receipts_error_is_not_memoised", settled: true, txs: 2, blocks: 2, + behavior: receiptsBehavior{blockRcpErr: &jsonrpc.Error{Code: -32000, Message: "request timed out"}}, + want: counts{getLogs: 2, blockReceipts: 2}}, + {name: "d_fallback_batches_stay_within_gateway_limit", settled: true, txs: 45, numLogs: 2, maxBatch: 40, ready: true, + behavior: receiptsBehavior{blockRcpErr: &jsonrpc.Error{Code: -32601, Message: "method not found"}}, + want: counts{getLogs: 1, blockReceipts: 1, txCount: 1, txReceipt: 45}}, + {name: "d_block_not_found_message_is_not_memoised", settled: true, txs: 2, blocks: 2, + behavior: receiptsBehavior{blockRcpErr: &jsonrpc.Error{Code: -32000, Message: "block does not exist"}}, + want: counts{getLogs: 2, blockReceipts: 2}}, + {name: "d_method_unavailable_message_is_memoised", settled: true, txs: 2, blocks: 2, ready: true, + behavior: receiptsBehavior{blockRcpErr: &jsonrpc.Error{Code: -32000, Message: "the method eth_getBlockReceipts does not exist/is not available"}}, + want: counts{getLogs: 2, blockReceipts: 1, txCount: 2, txReceipt: 4}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if tc.blocks == 0 { + tc.blocks = 1 + } + blocks := make([]*types.Block, 0, tc.blocks) + behavior := tc.behavior + behavior.receiptLogs = map[common.Hash][]types.Log{} + wantLogs := map[common.Hash][]types.Log{} + for i := 0; i < tc.blocks; i++ { + block := newTxBlock(t, 1000+uint64(i), tc.txs, tc.bloom) + blocks = append(blocks, block) + var logs []types.Log + for j := 0; j < tc.numLogs; j++ { + logs = append(logs, newTestLog(block, uint(j), topicA)) + } + behavior.receiptLogs[block.Hash()] = logs + wantLogs[block.Hash()] = logs + } + if tc.logs == "log" { + payload, err := json.Marshal(wantLogs[blocks[0].Hash()]) + require.NoError(t, err) + behavior.logs = payload + } + + node := newReceiptsNode(t, blocks, &behavior) + node.maxBatch = tc.maxBatch + provider, err := ethrpc.NewProvider(node.URL) + require.NoError(t, err) + opts := DefaultOptions + opts.WithLogs = true + if tc.cache { + backend, err := memcache.NewBackend(64) + require.NoError(t, err) + opts.CacheBackend = backend + } + monitor, err := NewMonitor(provider, opts) + require.NoError(t, err) + monitor.chainID = big.NewInt(1) + monitor.settledHeaders.Store(tc.settled) + + for _, block := range blocks { + b := &Block{Block: block, Event: Added} + monitor.addLogs(context.Background(), Blocks{b}) + require.Equal(t, tc.ready, b.OK) + if !tc.ready { + require.Nil(t, b.Logs) + continue + } + require.NotNil(t, b.Logs) + if want := wantLogs[block.Hash()]; len(want) == 0 { + require.Empty(t, b.Logs) + } else { + require.Equal(t, want, b.Logs) + } + } + + got := counts{ + getLogs: node.calls("eth_getLogs"), + blockReceipts: node.calls("eth_getBlockReceipts"), + txCount: node.calls("eth_getBlockTransactionCountByHash"), + txReceipt: node.calls("eth_getTransactionReceipt"), + } + require.Equal(t, tc.want, got) + if tc.maxBatch > 0 { + node.mu.Lock() + require.LessOrEqual(t, node.largestBatch, receiptsBatchSize) + node.mu.Unlock() + } + + if tc.cache { + // the receipt logs were cached, so a second read replays them + // without asking the node + logs, _, err := monitor.filterLogs(context.Background(), blocks[0], monitor.logTopics()) + require.NoError(t, err) + require.Equal(t, wantLogs[blocks[0].Hash()], logs) + require.Equal(t, tc.want.getLogs, node.calls("eth_getLogs")) + require.Equal(t, tc.want.blockReceipts, node.calls("eth_getBlockReceipts")) + } + }) + } +} diff --git a/ethmonitor/resolve_test.go b/ethmonitor/resolve_test.go new file mode 100644 index 00000000..6f1a05d3 --- /dev/null +++ b/ethmonitor/resolve_test.go @@ -0,0 +1,428 @@ +package ethmonitor + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log/slog" + "math/big" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/0xsequence/ethkit/ethmonitor/internal/mocks" + "github.com/0xsequence/ethkit/ethrpc" + "github.com/0xsequence/ethkit/ethrpc/jsonrpc" + "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 TestLogsAttemptsFallBackToReceipts(t *testing.T) { + topicA, topicB := common.HexToHash("0xaa"), common.HexToHash("0xbb") + logsDown := &jsonrpc.Error{Code: -32000, Message: "logs down"} + + cases := []struct { + name string + settled bool + txs int + bloom types.Bloom + topics []common.Hash + behavior receiptsBehavior + receipts func(block *types.Block) []types.Log + calls int + wantReady []bool // after each call + wantLogs func(receipts []types.Log) []types.Log + + wantBlockReceipts int // eth_getBlockReceipts calls in total, checked when non-zero + }{ + { + name: "a_empty_logs_with_nonzero_bloom", txs: 1, bloom: nonzeroBloom, calls: 3, + receipts: func(b *types.Block) []types.Log { + return []types.Log{newTestLog(b, 0, topicA), newTestLog(b, 1, topicB)} + }, + wantReady: []bool{false, false, true}, + wantLogs: func(r []types.Log) []types.Log { return r }, + }, + { + name: "b_logs_error_receipts_empty", txs: 2, behavior: receiptsBehavior{logsErr: logsDown}, calls: 3, + receipts: func(b *types.Block) []types.Log { return nil }, + wantReady: []bool{false, false, true}, + wantLogs: func(r []types.Log) []types.Log { return []types.Log{} }, + }, + { + name: "c_topic_filtered_receipts", txs: 1, topics: []common.Hash{topicA}, behavior: receiptsBehavior{logsErr: logsDown}, calls: 3, + receipts: func(b *types.Block) []types.Log { + return []types.Log{newTestLog(b, 0, topicA), newTestLog(b, 1, topicB)} + }, + wantReady: []bool{false, false, true}, + wantLogs: func(r []types.Log) []types.Log { return r[:1] }, + }, + { + // a transient JSON-RPC error doesn't disable eth_getBlockReceipts: + // it's asked again on each of attempts 3, 4 and 5 + name: "d_receipts_fail_too", txs: 1, behavior: receiptsBehavior{logsErr: logsDown, receiptsErr: errReceiptsDown}, calls: 5, + receipts: func(b *types.Block) []types.Log { return nil }, + wantReady: []bool{false, false, false, false, false}, + wantBlockReceipts: 3, + }, + { + // receipts without logs contradict a non-zero bloom outside + // settled-header chains, so they aren't trusted either + name: "e_empty_receipts_contradict_nonzero_bloom", txs: 1, bloom: nonzeroBloom, calls: 5, + receipts: func(b *types.Block) []types.Log { return nil }, + wantReady: []bool{false, false, false, false, false}, + wantBlockReceipts: 3, + }, + { + // on settled-header chains filterLogs already asks for receipts on + // every attempt, so addLogs doesn't ask a second time once the + // attempts are used up: one receipts call per attempt + name: "f_settled_receipts_fail_one_call_per_attempt", settled: true, txs: 1, behavior: receiptsBehavior{receiptsErr: errReceiptsDown}, calls: 5, + receipts: func(b *types.Block) []types.Log { return nil }, + wantReady: []bool{false, false, false, false, false}, + wantBlockReceipts: 5, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + block := newTxBlock(t, 1000, tc.txs, tc.bloom) + receiptLogs := tc.receipts(block) + behavior := tc.behavior + behavior.receiptLogs = map[common.Hash][]types.Log{block.Hash(): receiptLogs} + node := newReceiptsNode(t, []*types.Block{block}, &behavior) + provider, err := ethrpc.NewProvider(node.URL) + require.NoError(t, err) + + opts := DefaultOptions + opts.WithLogs = true + opts.MaxLogsAttempts = 3 + opts.LogTopics = tc.topics + monitor, err := NewMonitor(provider, opts) + require.NoError(t, err) + monitor.chainID = big.NewInt(1) + monitor.settledHeaders.Store(tc.settled) + + b := &Block{Block: block, Event: Added} + for i := 0; i < tc.calls; i++ { + monitor.addLogs(context.Background(), Blocks{b}) + require.Equal(t, tc.wantReady[i], b.OK, "call %d", i+1) + if i < 2 && !tc.settled { + // receipts are only consulted once the attempts are used up + require.Zero(t, node.calls("eth_getBlockReceipts"), "call %d", i+1) + } + } + if tc.wantBlockReceipts > 0 { + require.Equal(t, tc.wantBlockReceipts, node.calls("eth_getBlockReceipts")) + require.Zero(t, node.calls("eth_getTransactionReceipt"), "eth_getBlockReceipts must not be given up on") + } + if tc.wantLogs == nil { + require.Nil(t, b.Logs) + return + } + want := tc.wantLogs(receiptLogs) + require.NotNil(t, b.Logs) + if len(want) == 0 { + require.Empty(t, b.Logs) + } else { + require.Equal(t, want, b.Logs) + } + }) + } +} + +// recordingHandler is a slog.Handler which keeps every record. +type recordingHandler struct { + mu sync.Mutex + records []slog.Record +} + +func (h *recordingHandler) Enabled(context.Context, slog.Level) bool { return true } +func (h *recordingHandler) Handle(_ context.Context, r slog.Record) error { + h.mu.Lock() + defer h.mu.Unlock() + h.records = append(h.records, r) + return nil +} +func (h *recordingHandler) WithAttrs([]slog.Attr) slog.Handler { return h } +func (h *recordingHandler) WithGroup(string) slog.Handler { return h } + +// messages returns the messages at level which contain substr. +func (h *recordingHandler) messages(level slog.Level, substr string) []string { + h.mu.Lock() + defer h.mu.Unlock() + var out []string + for _, r := range h.records { + if r.Level == level && strings.Contains(r.Message, substr) { + out = append(out, r.Message) + } + } + return out +} + +// recordingAlerter is a util.Alerter which keeps every alert. +type recordingAlerter struct { + mu sync.Mutex + alerts []string +} + +func (a *recordingAlerter) Alert(_ context.Context, format string, v ...interface{}) { + a.mu.Lock() + defer a.mu.Unlock() + a.alerts = append(a.alerts, fmt.Sprintf(format, v...)) +} + +func (a *recordingAlerter) messages() []string { + a.mu.Lock() + defer a.mu.Unlock() + return append([]string{}, a.alerts...) +} + +// testBlockPayload is the JSON a node returns for a block built by newTxBlock, +// optionally with a settledHeight field. +func testBlockPayload(t *testing.T, number uint64, txCount int, bloom types.Bloom, settled bool) json.RawMessage { + t.Helper() + block := newTxBlock(t, number, txCount, bloom) + headerJSON, err := json.Marshal(block.Header()) + require.NoError(t, err) + var payload map[string]json.RawMessage + require.NoError(t, json.Unmarshal(headerJSON, &payload)) + + txs := make([]json.RawMessage, 0, txCount) + for _, tx := range block.Transactions() { + txJSON, err := json.Marshal(tx) + require.NoError(t, err) + txs = append(txs, txJSON) + } + payload["transactions"], err = json.Marshal(txs) + require.NoError(t, err) + payload["uncles"] = json.RawMessage(`[]`) + if settled { + payload["settledHeight"] = json.RawMessage(fmt.Sprintf(`"0x%x"`, number-1)) + } + out, err := json.Marshal(payload) + require.NoError(t, err) + return out +} + +// newMockChainProvider serves blocks first..last, built by testBlockPayload. +// Their hashes are those of newTxBlock. eth_getLogs answers with logs and +// every receipts call with receipts, as given. +func newMockChainProvider(t *testing.T, first, last uint64, txCount int, bloom types.Bloom, settled bool, + logs func() (json.RawMessage, error), receipts func() error) *mocks.MockRawInterface { + t.Helper() + + payloads := map[uint64]json.RawMessage{} + hashes := map[common.Hash]uint64{} + for n := first; n <= last; n++ { + payloads[n] = testBlockPayload(t, n, txCount, bloom, settled) + hashes[newTxBlock(t, n, 0, bloom).Hash()] = n + } + + 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 + } + n := last + if number != nil { + n = number.Uint64() + } + payload, ok := payloads[n] + if !ok { + return nil, ethereum.NotFound + } + return payload, nil + }).AnyTimes() + provider.EXPECT().RawBlockByHash(gomock.Any(), gomock.Any()).DoAndReturn( + func(ctx context.Context, hash common.Hash) (json.RawMessage, error) { + payload, ok := payloads[hashes[hash]] + if !ok { + return nil, ethereum.NotFound + } + return payload, nil + }).AnyTimes() + provider.EXPECT().RawFilterLogs(gomock.Any(), gomock.Any()).DoAndReturn( + func(ctx context.Context, _ ethereum.FilterQuery) (json.RawMessage, error) { + return logs() + }).AnyTimes() + provider.EXPECT().Do(gomock.Any(), gomock.Any()).DoAndReturn( + func(ctx context.Context, _ ...ethrpc.Call) ([]byte, error) { + return nil, receipts() + }).AnyTimes() + return provider +} + +// runTestMonitor runs the monitor until the test ends. The returned channel +// yields Run's error, and is closed after. +func runTestMonitor(t *testing.T, monitor *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 +} + +// collectBlocks records every block the subscription receives. +type collectedBlocks struct { + mu sync.Mutex + blocks []*Block +} + +func collectSubscription(t *testing.T, monitor *Monitor) *collectedBlocks { + t.Helper() + sub := monitor.Subscribe(t.Name()) + stop := make(chan struct{}) + collected := &collectedBlocks{} + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case blocks := <-sub.Blocks(): + collected.mu.Lock() + collected.blocks = append(collected.blocks, blocks...) + collected.mu.Unlock() + case <-stop: + return + } + } + }() + t.Cleanup(func() { + close(stop) + wg.Wait() + sub.Unsubscribe() + }) + return collected +} + +func (c *collectedBlocks) snapshot() []*Block { + c.mu.Lock() + defer c.mu.Unlock() + return append([]*Block{}, c.blocks...) +} + +func TestMonitorUnresolvedBlocksAreNotFatal(t *testing.T) { + // Blocks carry transactions and both eth_getLogs and every receipts call + // fail, so no block can be resolved. The chain is served by a mocked + // provider, as the fake chain's blocks have no transactions. + const ( + first = uint64(1000) + retention = 10 + numBlocks = 3*retention + 5 + last = first + numBlocks - 1 + ) + newMonitor := func(t *testing.T, strict bool) (*Monitor, *recordingHandler, *recordingAlerter) { + provider := newMockChainProvider(t, first, last, 1, types.Bloom{}, false, + func() (json.RawMessage, error) { return nil, errors.New("logs down") }, + func() error { return errors.New("receipts down") }) + handler, alerter := &recordingHandler{}, &recordingAlerter{} + opts := DefaultOptions + opts.Logger = slog.New(handler) + opts.Alerter = alerter + opts.WithLogs = true + opts.StrictLogs = strict + opts.BlockRetentionLimit = retention + opts.MaxLogsAttempts = 3 + opts.StartBlockNumber = new(big.Int).SetUint64(first) + opts.PollingInterval = time.Millisecond + opts.Timeout = 2 * time.Second + monitor, err := NewMonitor(provider, opts) + require.NoError(t, err) + return monitor, handler, alerter + } + + t.Run("default", func(t *testing.T) { + monitor, handler, alerter := newMonitor(t, false) + collected := collectSubscription(t, monitor) + done := runTestMonitor(t, monitor) + + // A block is published without logs once it has left the chain cache, + // which holds the last `retention` blocks. + published := int(numBlocks - retention) + var runErr atomic.Value + require.Eventually(t, func() bool { + select { + case err := <-done: + runErr.Store(fmt.Sprintf("%v", err)) + return true + default: + } + return len(collected.snapshot()) >= published + }, 10*time.Second, 5*time.Millisecond) + require.Nil(t, runErr.Load(), "Run returned") + select { + case err := <-done: + t.Fatalf("Run returned: %v", err) + case <-time.After(100 * time.Millisecond): + } + + blocks := collected.snapshot() + require.Len(t, blocks, published) + require.Greater(t, len(blocks), 2*retention, "more blocks than the publish queue holds") + for i, block := range blocks { + require.Equal(t, first+uint64(i), block.NumberU64()) + require.Equal(t, Added, block.Event) + require.True(t, block.OK) + require.NotNil(t, block.Logs) + require.Empty(t, block.Logs) + } + + errorLogs := handler.messages(slog.LevelError, "publishing without logs") + require.Len(t, errorLogs, published) + alerts := alerter.messages() + require.Len(t, alerts, published) + for _, block := range blocks { + hash := block.Hash().Hex() + require.Len(t, filterContaining(errorLogs, hash), 1, "Error logs for block %d", block.NumberU64()) + require.Len(t, filterContaining(alerts, hash), 1, "alerts for block %d", block.NumberU64()) + } + }) + + t.Run("strict", func(t *testing.T) { + monitor, _, alerter := newMonitor(t, true) + collected := collectSubscription(t, monitor) + done := runTestMonitor(t, monitor) + + select { + case err := <-done: + require.ErrorIs(t, err, ErrFatal) + require.ErrorIs(t, err, ErrQueueFull) + case <-time.After(10 * time.Second): + t.Fatal("expected the publish queue to overflow") + } + require.Empty(t, collected.snapshot()) + require.Empty(t, alerter.messages()) + }) +} + +func filterContaining(messages []string, substr string) []string { + var out []string + for _, m := range messages { + if strings.Contains(m, substr) { + out = append(out, m) + } + } + return out +} diff --git a/ethmonitor/settled_test.go b/ethmonitor/settled_test.go new file mode 100644 index 00000000..9c48c0a2 --- /dev/null +++ b/ethmonitor/settled_test.go @@ -0,0 +1,205 @@ +package ethmonitor + +import ( + "encoding/json" + "errors" + "fmt" + "log/slog" + "math/big" + "sync/atomic" + "testing" + "time" + + "github.com/0xsequence/ethkit/ethmonitor/internal/mocks" + "github.com/0xsequence/ethkit/ethrpc" + "github.com/0xsequence/ethkit/ethrpc/jsonrpc" + "github.com/goware/superr" + "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" +) + +func TestHasSettledHeight(t *testing.T) { + cases := []struct { + name string + payload string + want bool + }{ + {name: "hex", payload: `{"number":"0x1","settledHeight":"0x5"}`, want: true}, + {name: "number", payload: `{"number":"0x1","settledHeight":5}`, want: true}, + {name: "zero", payload: `{"number":"0x1","settledHeight":"0x0"}`, want: true}, + {name: "spaced", payload: `{"number":"0x1", "settledHeight" : "0x5"}`, want: true}, + {name: "absent", payload: `{"number":"0x1","hash":"0x2"}`}, + {name: "null", payload: `{"number":"0x1","settledHeight":null}`}, + {name: "spaced_null", payload: `{"number":"0x1","settledHeight": null }`}, + {name: "malformed_json", payload: `{"number":"0x1","settledHeight":"0x5"`}, + {name: "not_json", payload: `"settledHeight"`}, + {name: "empty", payload: ``}, + // the key inside a string, which the byte pre-check alone can't rule out + {name: "key_in_transaction_input", payload: `{"number":"0x1","transactions":[{"input":"{\"settledHeight\":\"0x5\"}"}]}`}, + {name: "key_in_string_value", payload: `{"number":"0x1","extraData":"\"settledHeight\""}`}, + // the pre-check passes and the decoder must not accept a nested key + {name: "nested_key_only", payload: `{"number":"0x1","transactions":[{"settledHeight":"0x5"}]}`}, + {name: "nested_object_only", payload: `{"number":"0x1","extra":{"settledHeight":"0x5"}}`}, + {name: "top_level_and_nested", payload: `{"extra":{"settledHeight":"0x5"},"settledHeight":"0x6"}`, want: true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.want, hasSettledHeight([]byte(tc.payload))) + }) + } +} + +func TestSettledHeadersDetection(t *testing.T) { + const detected = "settled block headers detected" + + newMonitor := func(t *testing.T) (*Monitor, *recordingHandler) { + handler := &recordingHandler{} + opts := DefaultOptions + opts.Logger = slog.New(handler) + monitor, err := NewMonitor(mocks.NewMockRawInterface(gomock.NewController(t)), opts) + require.NoError(t, err) + return monitor, handler + } + + t.Run("settledHeight_flips_flag_and_logs_once", func(t *testing.T) { + monitor, handler := newMonitor(t) + require.False(t, monitor.usesSettledHeaders()) + + for n := uint64(1000); n < 1002; n++ { + block, err := monitor.unmarshalBlock(testBlockPayload(t, n, 0, nonzeroBloom, true)) + require.NoError(t, err) + require.Equal(t, n, block.NumberU64()) + require.True(t, monitor.usesSettledHeaders()) + } + require.Len(t, handler.messages(slog.LevelInfo, detected), 1) + }) + + t.Run("no_settledHeight_leaves_flag_false", func(t *testing.T) { + monitor, handler := newMonitor(t) + for n := uint64(1000); n < 1002; n++ { + _, err := monitor.unmarshalBlock(testBlockPayload(t, n, 0, nonzeroBloom, false)) + require.NoError(t, err) + require.False(t, monitor.usesSettledHeaders()) + } + require.Empty(t, handler.messages(slog.LevelInfo, detected)) + }) + + t.Run("invalid_payload_does_not_flip_flag", func(t *testing.T) { + monitor, _ := newMonitor(t) + _, err := monitor.unmarshalBlock([]byte(`{"settledHeight":"0x5"`)) + require.Error(t, err) + require.False(t, monitor.usesSettledHeaders()) + }) +} + +func TestMonitorSettledChainDoesNotStall(t *testing.T) { + // Every block has a non-zero bloom and eth_getLogs answers []. On a + // settled-header chain that is a premature answer for a block which + // has no logs of its own, and must not stall the monitor. + const ( + first = uint64(1000) + count = 25 + last = first + count - 1 + ) + newMonitor := func(t *testing.T, settled bool, mutate func(*Options)) (*Monitor, *atomic.Int64) { + var receiptCalls atomic.Int64 + provider := newMockChainProvider(t, first, last, 0, nonzeroBloom, settled, + func() (json.RawMessage, error) { return json.RawMessage(`[]`), nil }, + func() error { + receiptCalls.Add(1) + return errors.New("receipts down") + }) + opts := DefaultOptions + opts.WithLogs = true + opts.StartBlockNumber = new(big.Int).SetUint64(first) + opts.PollingInterval = time.Millisecond + opts.Timeout = 2 * time.Second + mutate(&opts) + monitor, err := NewMonitor(provider, opts) + require.NoError(t, err) + return monitor, &receiptCalls + } + + t.Run("settled", func(t *testing.T) { + monitor, receiptCalls := newMonitor(t, true, func(o *Options) { + o.BlockRetentionLimit = 6 // the publish queue holds 12, fewer than the 25 blocks + }) + collected := collectSubscription(t, monitor) + done := runTestMonitor(t, monitor) + + var runErr atomic.Value + require.Eventually(t, func() bool { + select { + case err := <-done: + runErr.Store(err) + return true + default: + } + return len(collected.snapshot()) >= count + }, 10*time.Second, 5*time.Millisecond) + require.Nil(t, runErr.Load(), "Run returned") + + blocks := collected.snapshot() + require.Len(t, blocks, count) + for i, block := range blocks { + require.Equal(t, first+uint64(i), block.NumberU64()) + require.Equal(t, Added, block.Event) + require.True(t, block.OK) + require.Empty(t, block.Logs) + } + require.Zero(t, receiptCalls.Load(), "blocks without transactions need no receipts") + require.True(t, monitor.IsRunning()) + }) + + // Without settledHeight the non-zero bloom makes the empty logs invalid, as + // before. StrictLogs and a high MaxLogsAttempts turn off the receipts + // fallback, which would otherwise resolve these transaction-less blocks. + // The queue holds 4 events, fewer than the 10 blocks the chain retains, so + // it overflows before any block is evicted. + t.Run("control_not_settled_stalls", func(t *testing.T) { + monitor, receiptCalls := newMonitor(t, false, func(o *Options) { + o.BlockRetentionLimit = 2 + o.StrictLogs = true + o.MaxLogsAttempts = 1000 + }) + collected := collectSubscription(t, monitor) + done := runTestMonitor(t, monitor) + + select { + case err := <-done: + require.ErrorIs(t, err, ErrFatal) + require.ErrorIs(t, err, ErrQueueFull) + case <-time.After(10 * time.Second): + t.Fatal("expected the publish queue to overflow") + } + require.Empty(t, collected.snapshot(), "no event may be published for a block without logs") + require.Zero(t, receiptCalls.Load()) + }) +} + +func TestIsMethodNotFoundError(t *testing.T) { + cases := []struct { + name string + err error + want bool + }{ + {name: "code_32601", err: jsonrpc.Error{Code: -32601, Message: "anything"}, want: true}, + {name: "method_not_found_message", err: jsonrpc.Error{Code: -32000, Message: "Method not found"}, want: true}, + {name: "avalanche_public_rpc", err: jsonrpc.Error{Code: -32000, Message: "the method eth_getBlockReceipts does not exist/is not available"}, want: true}, + {name: "method_not_supported", err: jsonrpc.Error{Code: -32000, Message: "method eth_getBlockReceipts is not supported"}, want: true}, + {name: "wrapped", err: fmt.Errorf("batch: %w", jsonrpc.Error{Code: -32601, Message: "method not found"}), want: true}, + + {name: "block_does_not_exist", err: jsonrpc.Error{Code: -32000, Message: "block 0xabc does not exist"}}, + {name: "header_not_found", err: jsonrpc.Error{Code: -32000, Message: "header not found"}}, + {name: "state_not_available", err: jsonrpc.Error{Code: -32000, Message: "historical state not available"}}, + {name: "unsupported_block_tag", err: jsonrpc.Error{Code: -32602, Message: "unsupported block number"}}, + {name: "timeout", err: jsonrpc.Error{Code: -32000, Message: "request timed out"}}, + {name: "http_failure", err: superr.Wrap(ethrpc.ErrRequestFail, &jsonrpc.Error{Code: -32601, Message: "method not found"})}, + {name: "plain_error", err: errors.New("method not found")}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.want, isMethodNotFoundError(tc.err)) + }) + } +} From c7206b3007a71c124120752e9aaad2f413d72282 Mon Sep 17 00:00:00 2001 From: py-zoid Date: Thu, 22 Oct 2026 16:22:37 +0400 Subject: [PATCH 4/7] fix(ethmonitor): recheck cached empty logs, flag unconfirmed blocks, retry rate-limited receipts - A cached empty logs result for a settled-header block is now checked against the block's receipts, and replaced with the receipt logs when they disagree. Entries written by monitors that did not confirm empty results, such as earlier versions sharing the cache, were otherwise accepted until they expired. - Block.LogsUnconfirmed is set on blocks published without confirmed logs, after eth_getLogs and receipts both failed for the block's whole stay in the chain cache, so consumers can tell them apart from blocks without logs. It is kept in block snapshots and by Blocks.Copy. - eth_getBlockReceipts errors in non-2xx responses no longer trigger the per-transaction receipts fallback, which multiplied requests against rate-limited providers. A method-not-found error is recognised at any HTTP status; any other error is retried. --- ethmonitor/bootstrap.go | 19 +++++++----- ethmonitor/chain.go | 17 +++++++--- ethmonitor/ethmonitor.go | 63 ++++++++++++++++++++++++++++---------- ethmonitor/logs_test.go | 35 +++++++++++++++++++-- ethmonitor/resolve.go | 1 + ethmonitor/resolve_test.go | 20 ++++++++++++ ethmonitor/settled.go | 46 +++++++++++++++------------- ethmonitor/settled_test.go | 5 ++- 8 files changed, 151 insertions(+), 55 deletions(-) diff --git a/ethmonitor/bootstrap.go b/ethmonitor/bootstrap.go index d616c6e2..5559828f 100644 --- a/ethmonitor/bootstrap.go +++ b/ethmonitor/bootstrap.go @@ -77,18 +77,20 @@ func (c *Chain) Snapshot() ([]byte, error) { } type blockSnapshot struct { - Block *types.Block `json:"block"` - Event Event `json:"event"` - Logs []types.Log `json:"logs"` - OK bool `json:"ok"` + Block *types.Block `json:"block"` + Event Event `json:"event"` + Logs []types.Log `json:"logs"` + OK bool `json:"ok"` + LogsUnconfirmed bool `json:"logsUnconfirmed,omitempty"` } func (b *Block) MarshalJSON() ([]byte, error) { return json.Marshal(&blockSnapshot{ - Block: b.Block, - Event: b.Event, - Logs: b.Logs, - OK: b.OK, + Block: b.Block, + Event: b.Event, + Logs: b.Logs, + OK: b.OK, + LogsUnconfirmed: b.LogsUnconfirmed, }) } @@ -102,6 +104,7 @@ func (b *Block) UnmarshalJSON(data []byte) error { b.Event = s.Event b.Logs = s.Logs b.OK = s.OK + b.LogsUnconfirmed = s.LogsUnconfirmed b.canonicalState = nil return nil } diff --git a/ethmonitor/chain.go b/ethmonitor/chain.go index 353a5dc2..959d3976 100644 --- a/ethmonitor/chain.go +++ b/ethmonitor/chain.go @@ -243,6 +243,12 @@ type Block struct { // OK flag which represents the block is ready for broadcasting OK bool + // LogsUnconfirmed is set when the block was published without confirmed + // logs: eth_getLogs and receipts both failed for its whole stay in the + // chain cache. Logs is empty, but the block may have had logs. Never set + // when StrictLogs is enabled. + LogsUnconfirmed bool + canonicalState *blockCanonicalState // logsAttempts counts failed log fetches for this block instance. @@ -597,11 +603,12 @@ func (blocks Blocks) Copy() Blocks { } nb[i] = &Block{ - Block: b.Block, - Event: b.Event, - Logs: logs, - OK: b.OK, - canonicalState: b.canonicalState, + Block: b.Block, + Event: b.Event, + Logs: logs, + OK: b.OK, + LogsUnconfirmed: b.LogsUnconfirmed, + canonicalState: b.canonicalState, } } diff --git a/ethmonitor/ethmonitor.go b/ethmonitor/ethmonitor.go index d20cd96e..662db229 100644 --- a/ethmonitor/ethmonitor.go +++ b/ethmonitor/ethmonitor.go @@ -924,23 +924,12 @@ func (m *Monitor) filterLogs(ctx context.Context, block *types.Block, topics [][ // an empty result may be premature. Receipt calls do wait, so confirm // it against them before it can be cached. if settled && len(topics) == 0 && len(logs) == 0 { - rctx, rcancel := context.WithTimeout(ctx, 4*time.Second) - defer rcancel() - - receiptLogs, err := m.receiptLogs(rctx, block) + receiptLogs, receiptsPayload, err := m.confirmEmptyLogs(ctx, block) if err != nil { - return nil, fmt.Errorf("%w: %w", errReceiptsFailed, err) + return nil, err } - if len(receiptLogs) > 0 { - m.log.Warn(fmt.Sprintf("ethmonitor: settled block %d %s: eth_getLogs returned no logs but receipts have %d, using receipt logs", block.NumberU64(), blockHash.Hex(), len(receiptLogs))) - // Cache the receipt logs so a cache hit replays them. - logsPayload, err = json.Marshal(receiptLogs) - if err != nil { - return nil, err - } - logs = receiptLogs - } else if m.options.DebugLogging { - m.log.Debug(fmt.Sprintf("ethmonitor: settled block %s: empty logs confirmed via receipts", blockHash.Hex())) + if receiptsPayload != nil { + logs, logsPayload = receiptLogs, receiptsPayload } } @@ -979,8 +968,50 @@ func (m *Monitor) filterLogs(ctx context.Context, block *types.Block, topics [][ 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 + } + + // Peers or older monitors may have cached a premature empty result for a + // settled-header block without checking it against receipts. + if settled && len(topics) == 0 && len(logs) == 0 { + receiptLogs, receiptsPayload, err := m.confirmEmptyLogs(ctx, block) + if err != nil { + return nil, resp, err + } + if receiptsPayload != nil { + if err := m.cache.SetEx(ctx, key, receiptsPayload, m.options.CacheExpiry); err != nil { + m.log.Warn(fmt.Sprintf("ethmonitor: error replacing logs cache for block hash %s due to: '%v'", blockHash.Hex(), err)) + } + return receiptLogs, receiptsPayload, nil + } + } + return logs, resp, nil +} + +// confirmEmptyLogs checks an empty eth_getLogs result for a settled-header +// block against the block's receipts. It returns the receipt logs and their +// payload when the receipts have logs, and a nil payload when they confirm +// the block has none. +func (m *Monitor) confirmEmptyLogs(ctx context.Context, block *types.Block) ([]types.Log, []byte, error) { + ctx, cancel := context.WithTimeout(ctx, 4*time.Second) + defer cancel() + + logs, err := m.receiptLogs(ctx, block) + if err != nil { + return nil, nil, fmt.Errorf("%w: %w", errReceiptsFailed, err) + } + if len(logs) == 0 { + if m.options.DebugLogging { + m.log.Debug(fmt.Sprintf("ethmonitor: settled block %s: empty logs confirmed via receipts", block.Hash().Hex())) + } + return logs, nil, nil + } + m.log.Warn(fmt.Sprintf("ethmonitor: settled block %d %s: eth_getLogs returned no logs but receipts have %d, using receipt logs", block.NumberU64(), block.Hash().Hex(), len(logs))) + payload, err := json.Marshal(logs) + if err != nil { + return nil, nil, err } - return logs, resp, err + return logs, payload, nil } func (m *Monitor) backfillChainLogs(ctx context.Context, newBlocks Blocks) { diff --git a/ethmonitor/logs_test.go b/ethmonitor/logs_test.go index d4346799..11779cf0 100644 --- a/ethmonitor/logs_test.go +++ b/ethmonitor/logs_test.go @@ -133,7 +133,8 @@ type testNode struct { mu sync.Mutex counts map[string]int - maxBatch int // when set, larger batches are rejected with HTTP 400, as some RPC gateways do + maxBatch int // when set, larger batches are rejected with HTTP 400, as some RPC gateways do + status map[string]int // HTTP status for single requests to a method, 200 by default largestBatch int } @@ -185,6 +186,12 @@ func newTestNode(t *testing.T, handle func(method string, params []any) (json.Ra } w.Header().Set("Content-Type", "application/json") + node.mu.Lock() + status := node.status[requests[0].Method] + node.mu.Unlock() + if !batch && status != 0 { + w.WriteHeader(status) + } var err error if batch { err = json.NewEncoder(w).Encode(responses) @@ -377,7 +384,9 @@ func TestSettledEmptyLogResponses(t *testing.T) { numLogs int // receipt logs per block logs string behavior receiptsBehavior - maxBatch int // node batch limit, such as an RPC gateway's 40 + maxBatch int // node batch limit, such as an RPC gateway's 40 + status map[string]int // HTTP status by method, for single requests + preCache string // logs cache entry written beforehand, e.g. by an older monitor ready bool want counts // totals across all blocks }{ @@ -423,6 +432,19 @@ func TestSettledEmptyLogResponses(t *testing.T) { {name: "d_block_not_found_message_is_not_memoised", settled: true, txs: 2, blocks: 2, behavior: receiptsBehavior{blockRcpErr: &jsonrpc.Error{Code: -32000, Message: "block does not exist"}}, want: counts{getLogs: 2, blockReceipts: 2}}, + {name: "d_rate_limited_block_receipts_is_retried_without_fallback", settled: true, txs: 2, blocks: 2, + status: map[string]int{"eth_getBlockReceipts": http.StatusTooManyRequests}, + behavior: receiptsBehavior{blockRcpErr: &jsonrpc.Error{Code: -32005, Message: "rate limit exceeded"}}, + want: counts{getLogs: 2, blockReceipts: 2}}, + {name: "d_http_method_not_found_is_memoised", settled: true, txs: 2, blocks: 2, ready: true, + status: map[string]int{"eth_getBlockReceipts": http.StatusBadRequest}, + behavior: receiptsBehavior{blockRcpErr: &jsonrpc.Error{Code: -32601, Message: "method not found"}}, + want: counts{getLogs: 2, blockReceipts: 1, txCount: 2, txReceipt: 4}}, + {name: "j_premature_empty_cached_by_peer_is_replaced", settled: true, txs: 1, numLogs: 2, cache: true, preCache: `[]`, ready: true, + want: counts{blockReceipts: 1}}, + {name: "j_premature_empty_cached_by_peer_receipts_unavailable", settled: true, txs: 1, numLogs: 2, preCache: `[]`, + behavior: receiptsBehavior{receiptsErr: errReceiptsDown}, + want: counts{blockReceipts: 1}}, {name: "d_method_unavailable_message_is_memoised", settled: true, txs: 2, blocks: 2, ready: true, behavior: receiptsBehavior{blockRcpErr: &jsonrpc.Error{Code: -32000, Message: "the method eth_getBlockReceipts does not exist/is not available"}}, want: counts{getLogs: 2, blockReceipts: 1, txCount: 2, txReceipt: 4}}, @@ -454,11 +476,12 @@ func TestSettledEmptyLogResponses(t *testing.T) { node := newReceiptsNode(t, blocks, &behavior) node.maxBatch = tc.maxBatch + node.status = tc.status provider, err := ethrpc.NewProvider(node.URL) require.NoError(t, err) opts := DefaultOptions opts.WithLogs = true - if tc.cache { + if tc.cache || tc.preCache != "" { backend, err := memcache.NewBackend(64) require.NoError(t, err) opts.CacheBackend = backend @@ -467,6 +490,12 @@ func TestSettledEmptyLogResponses(t *testing.T) { require.NoError(t, err) monitor.chainID = big.NewInt(1) monitor.settledHeaders.Store(tc.settled) + if tc.preCache != "" { + for _, block := range blocks { + key := CacheKeyBlockLogs(monitor.chainID, block.Hash(), monitor.logTopics()) + require.NoError(t, monitor.cache.Set(context.Background(), key, []byte(tc.preCache))) + } + } for _, block := range blocks { b := &Block{Block: block, Event: Added} diff --git a/ethmonitor/resolve.go b/ethmonitor/resolve.go index 59325085..659e34ad 100644 --- a/ethmonitor/resolve.go +++ b/ethmonitor/resolve.go @@ -102,6 +102,7 @@ func (m *Monitor) resolveEvictedBlocks(ctx context.Context) { } block.Logs = []types.Log{} + block.LogsUnconfirmed = true block.OK = true m.log.Error(fmt.Sprintf("ethmonitor: block %d %s: publishing without logs, eth_getLogs and receipts both failed after %d attempts: %v", block.NumberU64(), block.Hash().Hex(), block.logsAttempts, err)) diff --git a/ethmonitor/resolve_test.go b/ethmonitor/resolve_test.go index 6f1a05d3..457a2ed2 100644 --- a/ethmonitor/resolve_test.go +++ b/ethmonitor/resolve_test.go @@ -385,6 +385,7 @@ func TestMonitorUnresolvedBlocksAreNotFatal(t *testing.T) { require.Equal(t, first+uint64(i), block.NumberU64()) require.Equal(t, Added, block.Event) require.True(t, block.OK) + require.True(t, block.LogsUnconfirmed) require.NotNil(t, block.Logs) require.Empty(t, block.Logs) } @@ -426,3 +427,22 @@ func filterContaining(messages []string, substr string) []string { } return out } + +func TestBlockLogsUnconfirmedSurvivesCopyAndSnapshot(t *testing.T) { + for _, unconfirmed := range []bool{false, true} { + t.Run(fmt.Sprintf("unconfirmed=%v", unconfirmed), func(t *testing.T) { + block := canonicalTestBlock(100) + block.OK = true + block.Logs = []types.Log{} + block.LogsUnconfirmed = unconfirmed + + require.Equal(t, unconfirmed, Blocks{block}.Copy()[0].LogsUnconfirmed) + + data, err := json.Marshal(block) + require.NoError(t, err) + var decoded Block + require.NoError(t, json.Unmarshal(data, &decoded)) + require.Equal(t, unconfirmed, decoded.LogsUnconfirmed) + }) + } +} diff --git a/ethmonitor/settled.go b/ethmonitor/settled.go index 721f4200..52b10de9 100644 --- a/ethmonitor/settled.go +++ b/ethmonitor/settled.go @@ -77,11 +77,9 @@ func (m *Monitor) receiptLogs(ctx context.Context, block *types.Block) ([]types. if m.blockReceiptsUnsupported.CompareAndSwap(false, true) { m.log.Info(fmt.Sprintf("ethmonitor: eth_getBlockReceipts is not supported (%v), using eth_getTransactionReceipt from now on", err)) } - case isHTTPJSONRPCError(err): - // A non-2xx response with a JSON-RPC error body, for example a - // gateway rejecting the method or rate limiting. It may be - // transient, so fall back without memoising. default: + // Includes non-2xx responses such as rate limiting, which the + // per-tx fallback would only add load to. return nil, fmt.Errorf("ethmonitor: eth_getBlockReceipts for block %d %s: %w", block.NumberU64(), blockHash.Hex(), err) } } @@ -157,16 +155,16 @@ func flattenReceiptLogs(receipts []*logsReceipt) []types.Log { return logs } -// isMethodNotFoundError reports whether err is a JSON-RPC error returned by the -// node in a successful HTTP response saying the method isn't supported: code -// -32601, or a message about the method itself, as some nodes answer with a -// generic code (e.g. "the method eth_getBlockReceipts does not exist/is not -// available"). Any other JSON-RPC error may be transient (a timeout, a missing -// trie node, a lagging backend answering that a new block does not exist), so -// it is returned to the caller to be retried instead. +// isMethodNotFoundError reports whether err is a JSON-RPC error saying the +// method isn't supported, in a response of any HTTP status: code -32601, or a +// message about the method itself, as some nodes answer with a generic code +// (e.g. "the method eth_getBlockReceipts does not exist/is not available"). +// Any other JSON-RPC error may be transient (a timeout, rate limiting, a +// missing trie node, a lagging backend answering that a new block does not +// exist), so it is returned to the caller to be retried instead. func isMethodNotFoundError(err error) bool { - var rpcErr jsonrpc.Error - if !errors.As(err, &rpcErr) || errors.Is(err, ethrpc.ErrRequestFail) { + rpcErr, ok := jsonRPCError(err) + if !ok { return false } if rpcErr.Code == -32601 { @@ -184,17 +182,21 @@ func isMethodNotFoundError(err error) bool { return false } -// isHTTPJSONRPCError reports whether err is a non-2xx HTTP response whose body -// carried a JSON-RPC error object. ethrpc wraps those with superr, whose -// stack is not reachable through errors.As. -func isHTTPJSONRPCError(err error) bool { - if !errors.Is(err, ethrpc.ErrRequestFail) { - return false +// jsonRPCError returns the JSON-RPC error in err. ethrpc wraps the error body +// of a non-2xx response with superr, whose stack is not reachable through +// errors.As. +func jsonRPCError(err error) (jsonrpc.Error, bool) { + var rpcErr jsonrpc.Error + if errors.As(err, &rpcErr) { + return rpcErr, true } for _, e := range superr.GetErrorStack(err) { - if _, ok := e.(*jsonrpc.Error); ok { - return true + switch e := e.(type) { + case *jsonrpc.Error: + return *e, true + case jsonrpc.Error: + return e, true } } - return false + return jsonrpc.Error{}, false } diff --git a/ethmonitor/settled_test.go b/ethmonitor/settled_test.go index 9c48c0a2..11b3f903 100644 --- a/ethmonitor/settled_test.go +++ b/ethmonitor/settled_test.go @@ -145,6 +145,7 @@ func TestMonitorSettledChainDoesNotStall(t *testing.T) { require.Equal(t, first+uint64(i), block.NumberU64()) require.Equal(t, Added, block.Event) require.True(t, block.OK) + require.False(t, block.LogsUnconfirmed) require.Empty(t, block.Logs) } require.Zero(t, receiptCalls.Load(), "blocks without transactions need no receipts") @@ -194,7 +195,9 @@ func TestIsMethodNotFoundError(t *testing.T) { {name: "state_not_available", err: jsonrpc.Error{Code: -32000, Message: "historical state not available"}}, {name: "unsupported_block_tag", err: jsonrpc.Error{Code: -32602, Message: "unsupported block number"}}, {name: "timeout", err: jsonrpc.Error{Code: -32000, Message: "request timed out"}}, - {name: "http_failure", err: superr.Wrap(ethrpc.ErrRequestFail, &jsonrpc.Error{Code: -32601, Message: "method not found"})}, + {name: "http_method_not_found", err: superr.Wrap(ethrpc.ErrRequestFail, &jsonrpc.Error{Code: -32601, Message: "method not found"}), want: true}, + {name: "http_rate_limited", err: superr.Wrap(ethrpc.ErrRequestFail, &jsonrpc.Error{Code: -32005, Message: "rate limit exceeded"})}, + {name: "http_failure_without_body", err: superr.Wrap(ethrpc.ErrRequestFail, errors.New("non-200 response with status code: 503"))}, {name: "plain_error", err: errors.New("method not found")}, } for _, tc := range cases { From a87d0078bca41b0a0f31bed6c60058565e53995c Mon Sep 17 00:00:00 2001 From: py-zoid Date: Thu, 22 Oct 2026 18:40:33 +0400 Subject: [PATCH 5/7] refactor(ethmonitor): drop zero transactions root case, document sticky receipts fallback - An empty transaction list now counts as an empty block only when the header carries the empty transactions root. Headers always include transactionsRoot, so the zero-root case only served test blocks, which now use the real empty root. - Note that once eth_getBlockReceipts is found unsupported, the monitor uses per-transaction receipts for the rest of its lifetime. --- ethmonitor/logs_test.go | 6 ++++-- ethmonitor/settled.go | 10 ++++++---- 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/ethmonitor/logs_test.go b/ethmonitor/logs_test.go index 11779cf0..5df18f59 100644 --- a/ethmonitor/logs_test.go +++ b/ethmonitor/logs_test.go @@ -325,8 +325,9 @@ func newReceiptsNode(t *testing.T, blocks []*types.Block, b *receiptsBehavior) * var testSigningKey, _ = crypto.HexToECDSA("4c0883a69102937d6231471b5dbb6204fe5129617082792ae468d01a3f362318") -// newTxBlock builds a block with txCount signed transactions. A block with -// transactions gets a non-zero header TxHash, as a real one would. +// newTxBlock builds a block with txCount signed transactions. Its header +// TxHash is the empty root without transactions and non-zero with them, as a +// real one would be. func newTxBlock(t *testing.T, number uint64, txCount int, bloom types.Bloom) *types.Block { t.Helper() signer := types.LatestSignerForChainID(big.NewInt(1)) @@ -349,6 +350,7 @@ func newTxBlock(t *testing.T, number uint64, txCount int, bloom types.Bloom) *ty Bloom: bloom, } header.SetHash(crypto.Keccak256Hash(new(big.Int).SetUint64(number).Bytes())) + header.TxHash = types.EmptyTxsHash if txCount > 0 { header.TxHash = common.HexToHash("0xbeef") } diff --git a/ethmonitor/settled.go b/ethmonitor/settled.go index 52b10de9..87447969 100644 --- a/ethmonitor/settled.go +++ b/ethmonitor/settled.go @@ -57,9 +57,8 @@ func (m *Monitor) receiptLogs(ctx context.Context, block *types.Block) ([]types. blockHash := block.Hash() // ethrpc drops transaction types it can't decode, so an empty transaction - // list only means an empty block when the header's root agrees. A zero - // root means the node did not report one. - if len(txs) == 0 && (block.TxHash() == types.EmptyTxsHash || block.TxHash() == (common.Hash{})) { + // list only means an empty block when the header's root agrees. + if len(txs) == 0 && block.TxHash() == types.EmptyTxsHash { return []types.Log{}, nil } @@ -73,7 +72,10 @@ func (m *Monitor) receiptLogs(ctx context.Context, block *types.Block) ([]types. } return flattenReceiptLogs(receipts), nil case isMethodNotFoundError(err): - // The node doesn't support the method, so stop asking. + // The node doesn't support the method, so stop asking. This lasts + // for the monitor's lifetime: behind a load balancer with mixed + // backends, one answer from a backend without the method moves + // every later block to the per-tx fallback. if m.blockReceiptsUnsupported.CompareAndSwap(false, true) { m.log.Info(fmt.Sprintf("ethmonitor: eth_getBlockReceipts is not supported (%v), using eth_getTransactionReceipt from now on", err)) } From b38f14f3a5e0dc84c837d53ceb14a8108541b6c6 Mon Sep 17 00:00:00 2001 From: py-zoid Date: Thu, 22 Oct 2026 19:44:13 +0400 Subject: [PATCH 6/7] fix(ethmonitor): require a receipt for every decoded transaction eth_getBlockReceipts results were only checked by count, so a partial response could pass when a receipt for a transaction type ethrpc does not decode stood in for a missing one, and the block was accepted without that transaction's logs. Every decoded transaction must now have a receipt, matched by transaction hash. --- ethmonitor/logs_test.go | 22 +++++++++++++++++----- ethmonitor/settled.go | 22 ++++++++++++++++------ 2 files changed, 33 insertions(+), 11 deletions(-) diff --git a/ethmonitor/logs_test.go b/ethmonitor/logs_test.go index 5df18f59..267c5f47 100644 --- a/ethmonitor/logs_test.go +++ b/ethmonitor/logs_test.go @@ -219,11 +219,12 @@ type receiptsBehavior struct { receiptsErr *jsonrpc.Error // JSON-RPC error from every receipts method blockRcpErr *jsonrpc.Error // JSON-RPC error from eth_getBlockReceipts only - receiptLogs map[common.Hash][]types.Log // receipt logs by block hash, held by the first tx's receipt - wrongBlockHash bool // receipts carry another block's hash - nullTxReceipt bool // eth_getTransactionReceipt returns null - dropReceipts int // eth_getBlockReceipts omits this many receipts - txCountDelta int // eth_getBlockTransactionCountByHash is off by this much + receiptLogs map[common.Hash][]types.Log // receipt logs by block hash, held by the first tx's receipt + wrongBlockHash bool // receipts carry another block's hash + nullTxReceipt bool // eth_getTransactionReceipt returns null + dropReceipts int // eth_getBlockReceipts omits this many receipts + swapLastForTyped bool // eth_getBlockReceipts replaces the last tx's receipt with one for an undecoded tx type + txCountDelta int // eth_getBlockTransactionCountByHash is off by this much } var errReceiptsDown = &jsonrpc.Error{Code: -32000, Message: "receipts down"} @@ -297,6 +298,14 @@ func newReceiptsNode(t *testing.T, blocks []*types.Block, b *receiptsBehavior) * for i := 0; i < len(block.Transactions())-b.dropReceipts; i++ { receipts = append(receipts, receipt(block, i)) } + if b.swapLastForTyped && len(receipts) > 0 { + receipts[len(receipts)-1] = map[string]any{ + "type": "0x3", + "blockHash": block.Hash(), + "transactionHash": common.HexToHash("0xb10b"), + "logs": []types.Log{}, + } + } result = receipts case "eth_getTransactionReceipt": found, ok := byTx[hashParam(params)] @@ -419,6 +428,9 @@ func TestSettledEmptyLogResponses(t *testing.T) { {name: "i_block_receipts_count_below_tx_count", settled: true, txs: 2, behavior: receiptsBehavior{dropReceipts: 1}, want: counts{getLogs: 1, blockReceipts: 1}}, + {name: "i_block_receipts_missing_decoded_tx_with_typed_receipt", settled: true, txs: 2, numLogs: 1, + behavior: receiptsBehavior{swapLastForTyped: true}, + want: counts{getLogs: 1, blockReceipts: 1}}, {name: "i_tx_count_differs_from_decoded_txs", settled: true, txs: 1, behavior: receiptsBehavior{txCountDelta: 1, blockRcpErr: &jsonrpc.Error{Code: -32601, Message: "method not found"}}, want: counts{getLogs: 1, blockReceipts: 1, txCount: 1, txReceipt: 1}}, diff --git a/ethmonitor/settled.go b/ethmonitor/settled.go index 87447969..37c23e71 100644 --- a/ethmonitor/settled.go +++ b/ethmonitor/settled.go @@ -67,7 +67,7 @@ func (m *Monitor) receiptLogs(ctx context.Context, block *types.Block) ([]types. _, err := m.provider.Do(ctx, ethrpc.NewCallBuilder[[]*logsReceipt]("eth_getBlockReceipts", nil, blockHash).Into(&receipts)) switch { case err == nil: - if err := checkBlockReceipts(receipts, blockHash, len(txs)); err != nil { + if err := checkBlockReceipts(receipts, blockHash, txs); err != nil { return nil, fmt.Errorf("ethmonitor: eth_getBlockReceipts for block %d %s: %w", block.NumberU64(), blockHash.Hex(), err) } return flattenReceiptLogs(receipts), nil @@ -121,15 +121,17 @@ func (m *Monitor) receiptLogs(ctx context.Context, block *types.Block) ([]types. } // checkBlockReceipts validates an eth_getBlockReceipts result against the -// block. Receipts beyond the decoded transactions are allowed only for -// transaction types ethrpc does not decode. -func checkBlockReceipts(receipts []*logsReceipt, blockHash common.Hash, numTxs int) error { +// block. Every decoded transaction must have a receipt. Receipts beyond the +// decoded transactions are allowed only for transaction types ethrpc does +// not decode. +func checkBlockReceipts(receipts []*logsReceipt, blockHash common.Hash, txs types.Transactions) error { // Only reached for blocks with transactions, so null (unknown or not yet // executed block) and [] are both errors. if len(receipts) == 0 { return fmt.Errorf("no receipts for a block with transactions") } decodable := 0 + txHashes := make(map[common.Hash]struct{}, len(receipts)) for i, receipt := range receipts { if receipt == nil { return fmt.Errorf("receipt %d is null", i) @@ -140,9 +142,17 @@ func checkBlockReceipts(receipts []*logsReceipt, blockHash common.Hash, numTxs i if receipt.Type <= types.DynamicFeeTxType { decodable++ } + txHashes[receipt.TransactionHash] = struct{}{} } - if len(receipts) < numTxs || decodable > numTxs { - return fmt.Errorf("%d receipts for %d transactions", len(receipts), numTxs) + if len(receipts) < len(txs) || decodable > len(txs) { + return fmt.Errorf("%d receipts for %d transactions", len(receipts), len(txs)) + } + // The counts can match while a decoded transaction's receipt is missing, + // when the response has a receipt for a type ethrpc does not decode. + for _, tx := range txs { + if _, ok := txHashes[tx.Hash()]; !ok { + return fmt.Errorf("no receipt for transaction %s", tx.Hash().Hex()) + } } return nil } From 013e2e9d3e79dc103f1ecf5f8b7f7445e479d1ee Mon Sep 17 00:00:00 2001 From: py-zoid Date: Thu, 22 Oct 2026 20:09:12 +0400 Subject: [PATCH 7/7] docs(ethmonitor): note the unchecked transaction count in block receipts validation --- ethmonitor/settled.go | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/ethmonitor/settled.go b/ethmonitor/settled.go index 37c23e71..0363960d 100644 --- a/ethmonitor/settled.go +++ b/ethmonitor/settled.go @@ -144,6 +144,12 @@ func checkBlockReceipts(receipts []*logsReceipt, blockHash common.Hash, txs type } txHashes[receipt.TransactionHash] = struct{}{} } + // NOTE: known limitation. The block's actual transaction count is not + // checked here, so a partial response that omits the receipt of a + // transaction ethrpc did not decode passes, and that transaction's logs + // are missed. Checking it would cost an eth_getBlockTransactionCountByHash + // call per lookup, as the per-tx fallback in receiptLogs does, for a case that + // needs both a node returning partial receipts and an undecoded tx type. if len(receipts) < len(txs) || decodable > len(txs) { return fmt.Errorf("%d receipts for %d transactions", len(receipts), len(txs)) }