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 7233b5d0..959d3976 100644 --- a/ethmonitor/chain.go +++ b/ethmonitor/chain.go @@ -243,7 +243,16 @@ 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. + logsAttempts int } type blockCanonicalState struct { @@ -594,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 311b61b6..662db229 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 } @@ -161,6 +175,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 @@ -199,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 } @@ -693,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 @@ -820,15 +847,28 @@ 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 } + // 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 @@ -849,11 +889,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 +915,25 @@ 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 { + receiptLogs, receiptsPayload, err := m.confirmEmptyLogs(ctx, block) + if err != nil { + return nil, err + } + if receiptsPayload != nil { + logs, logsPayload = receiptLogs, receiptsPayload + } + } + + fetchedLogs = logs return logsPayload, nil } @@ -910,8 +968,50 @@ func (m *Monitor) filterLogs(ctx context.Context, blockHash common.Hash, 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) { @@ -1442,9 +1542,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/logs_test.go b/ethmonitor/logs_test.go index 305fe354..267c5f47 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,430 @@ 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 + status map[string]int // HTTP status for single requests to a method, 200 by default + 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") + 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) + } 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 + 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"} + +// 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)) + } + 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)] + 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. 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)) + 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())) + header.TxHash = types.EmptyTxsHash + 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 + 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 + }{ + {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_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}}, + {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_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}}, + } + 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 + node.status = tc.status + provider, err := ethrpc.NewProvider(node.URL) + require.NoError(t, err) + opts := DefaultOptions + opts.WithLogs = true + if tc.cache || tc.preCache != "" { + 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) + 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} + 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/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/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..659e34ad --- /dev/null +++ b/ethmonitor/resolve.go @@ -0,0 +1,113 @@ +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.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)) + 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) + } + } +} diff --git a/ethmonitor/resolve_test.go b/ethmonitor/resolve_test.go new file mode 100644 index 00000000..457a2ed2 --- /dev/null +++ b/ethmonitor/resolve_test.go @@ -0,0 +1,448 @@ +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.True(t, block.LogsUnconfirmed) + 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 +} + +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 new file mode 100644 index 00000000..0363960d --- /dev/null +++ b/ethmonitor/settled.go @@ -0,0 +1,220 @@ +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. + if len(txs) == 0 && block.TxHash() == types.EmptyTxsHash { + 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, 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. 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)) + } + 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) + } + } + + // 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. 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) + } + if receipt.BlockHash != blockHash { + return fmt.Errorf("receipt %d has block hash %s", i, receipt.BlockHash.Hex()) + } + if receipt.Type <= types.DynamicFeeTxType { + decodable++ + } + 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)) + } + // 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 +} + +// 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 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 { + rpcErr, ok := jsonRPCError(err) + if !ok { + 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 +} + +// 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) { + switch e := e.(type) { + case *jsonrpc.Error: + return *e, true + case jsonrpc.Error: + return e, true + } + } + return jsonrpc.Error{}, false +} diff --git a/ethmonitor/settled_test.go b/ethmonitor/settled_test.go new file mode 100644 index 00000000..11b3f903 --- /dev/null +++ b/ethmonitor/settled_test.go @@ -0,0 +1,208 @@ +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.False(t, block.LogsUnconfirmed) + 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_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 { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.want, isMethodNotFoundError(tc.err)) + }) + } +}