From 82d70278c6058a3437fad9bf6b88635d835bad57 Mon Sep 17 00:00:00 2001 From: yperbasis Date: Thu, 2 Jul 2026 22:23:15 +0200 Subject: [PATCH 1/8] rpc/jsonrpc: extract txn lookup+bor-fallback and txn-index helpers Collapse the repeated txnLookup -> EventTxnLookup fallback -> _txNumReader.Min -> underflow check -> txnIndex derivation dance into two BaseAPI helpers: - txnLookupWithBorFallback: the !ok-triggered bor state-sync fallback shared verbatim by debug_traceTransaction, trace_replayTransaction and trace_transaction. GetTransactionReceipt, GetTransactionByHash and debug_getRawTransaction keep their inline fallbacks: their trigger is blockNum==0-based (fires also for ok lookups) and interleaves differently with prune/receipts checks, so folding them in would not be behavior-preserving. - txnIndexInBlock: Min + 'uint underflow txnums error' + index derivation, now shared by all eight call sites at the exact position the Min call used to occupy, so error precedence wrt prune checks and header fetches is unchanged. In GetTransactionReceipt the underflow check moves from after the bor fallback to the helper (before it); this is equivalent because TxnLookup returns blockNum==0 on every miss (both local BlockReader and RemoteBlockReader enforce this), so a failed lookup on a bor chain always sets isBorStateSyncTx and the check is skipped either way. Behavior-preserving: identical JSON responses, error strings and nil,nil not-found quirks at every site. --- rpc/jsonrpc/debug_api.go | 10 ++-------- rpc/jsonrpc/eth_api.go | 35 ++++++++++++++++++++++++++++++++++ rpc/jsonrpc/eth_block.go | 6 +----- rpc/jsonrpc/eth_receipts.go | 14 ++++---------- rpc/jsonrpc/eth_txs.go | 19 ++++-------------- rpc/jsonrpc/trace_adhoc.go | 27 +++----------------------- rpc/jsonrpc/trace_filtering.go | 27 +++----------------------- rpc/jsonrpc/tracing.go | 33 +++++++------------------------- 8 files changed, 59 insertions(+), 112 deletions(-) diff --git a/rpc/jsonrpc/debug_api.go b/rpc/jsonrpc/debug_api.go index 2a998c5a8a1..62a4663e740 100644 --- a/rpc/jsonrpc/debug_api.go +++ b/rpc/jsonrpc/debug_api.go @@ -807,18 +807,12 @@ func (api *DebugAPIImpl) GetRawTransaction(ctx context.Context, txnHash common.H } } - txNumMin, err := api._txNumReader.Min(ctx, tx, blockNum) + txnIndex, err := api.txnIndexInBlock(ctx, tx, blockNum, txNum, isBorStateSyncTx) if err != nil { return nil, err } - if txNumMin+1 > txNum && !isBorStateSyncTx { - return nil, fmt.Errorf("uint underflow txnums error txNum: %d, txNumMin: %d, blockNum: %d", txNum, txNumMin, blockNum) - } - - var txnIndex = txNum - txNumMin - 1 - - txn, err := api._txnReader.TxnByIdxInBlock(ctx, tx, blockNum, int(txnIndex)) + txn, err := api._txnReader.TxnByIdxInBlock(ctx, tx, blockNum, txnIndex) if err != nil { return nil, err } diff --git a/rpc/jsonrpc/eth_api.go b/rpc/jsonrpc/eth_api.go index 98163d39a39..093dfae03ef 100644 --- a/rpc/jsonrpc/eth_api.go +++ b/rpc/jsonrpc/eth_api.go @@ -239,6 +239,41 @@ func (api *BaseAPI) txnLookup(ctx context.Context, tx kv.Tx, txnHash common.Hash return api._txnReader.TxnLookup(ctx, overlayTx, txnHash) } +func (api *BaseAPI) txnLookupWithBorFallback(ctx context.Context, tx kv.Tx, txnHash common.Hash, chainConfig *chain.Config) (blockNum uint64, txNum uint64, isBorStateSyncTxn bool, ok bool, err error) { + blockNum, txNum, ok, err = api.txnLookup(ctx, tx, txnHash) + if err != nil { + return 0, 0, false, false, err + } + if ok { + return blockNum, txNum, false, true, nil + } + if chainConfig.Bor == nil { + return 0, 0, false, false, nil + } + blockNum, ok, err = api.bridgeReader.EventTxnLookup(ctx, txnHash) + if err != nil { + return 0, 0, false, false, err + } + if !ok { + return 0, 0, false, false, nil + } + return blockNum, txNum, true, true, nil +} + +// txnIndexInBlock derives the in-block txn index from a global txNum. For bor state sync +// txns txNum comes from a missed lookup, so the consistency check is skipped and the +// returned index is meaningless. +func (api *BaseAPI) txnIndexInBlock(ctx context.Context, tx kv.Tx, blockNum, txNum uint64, isBorStateSyncTxn bool) (int, error) { + txNumMin, err := api._txNumReader.Min(ctx, tx, blockNum) + if err != nil { + return 0, err + } + if txNumMin+1 > txNum && !isBorStateSyncTxn { + return 0, fmt.Errorf("uint underflow txnums error txNum: %d, txNumMin: %d, blockNum: %d", txNum, txNumMin, blockNum) + } + return int(txNum - txNumMin - 1), nil +} + func (api *BaseAPI) blockByNumberWithSenders(ctx context.Context, tx kv.Tx, number uint64) (*types.Block, error) { blockNumber, hash, _, err := rpchelper.GetBlockNumber(ctx, rpc.BlockNumberOrHashWithNumber(rpc.BlockNumber(number)), tx, api._blockReader, api.filters) if err != nil { diff --git a/rpc/jsonrpc/eth_block.go b/rpc/jsonrpc/eth_block.go index 9346bc45107..d8dca05e992 100644 --- a/rpc/jsonrpc/eth_block.go +++ b/rpc/jsonrpc/eth_block.go @@ -75,14 +75,10 @@ func (api *APIImpl) CallBundle(ctx context.Context, txHashes []common.Hash, stat return nil, err } - txNumMin, err := api._txNumReader.Min(ctx, tx, blockNumber) + txnIndex, err := api.txnIndexInBlock(ctx, tx, blockNumber, txNum, false) if err != nil { return nil, err } - if txNumMin+1 > txNum { - return nil, fmt.Errorf("uint underflow txnums error txNum: %d, txNumMin: %d, blockNum: %d", txNum, txNumMin, blockNumber) - } - txnIndex := int(txNum - txNumMin - 1) txn, err := api._txnReader.TxnByIdxInBlock(ctx, tx, blockNumber, txnIndex) if err != nil { return nil, err diff --git a/rpc/jsonrpc/eth_receipts.go b/rpc/jsonrpc/eth_receipts.go index 0868ecfb4d7..bd2a3f634b7 100644 --- a/rpc/jsonrpc/eth_receipts.go +++ b/rpc/jsonrpc/eth_receipts.go @@ -510,14 +510,14 @@ func (api *APIImpl) GetTransactionReceipt(ctx context.Context, txnHash common.Ha return nil, err } - txNumMin, err := api._txNumReader.Min(ctx, overlayTx, blockNum) + // Private API returns 0 if transaction is not found. + isBorStateSyncTx := blockNum == 0 && chainConfig.Bor != nil + + txnIndex, err := api.txnIndexInBlock(ctx, overlayTx, blockNum, txNum, isBorStateSyncTx) if err != nil { return nil, err } - // Private API returns 0 if transaction is not found. - isBorStateSyncTx := blockNum == 0 && chainConfig.Bor != nil - if isBorStateSyncTx { blockNum, ok, err = api.bridgeReader.EventTxnLookup(ctx, txnHash) if err != nil { @@ -529,10 +529,6 @@ func (api *APIImpl) GetTransactionReceipt(ctx context.Context, txnHash common.Ha return nil, nil } - if txNumMin+1 > txNum && !isBorStateSyncTx { - return nil, fmt.Errorf("uint underflow txnums error txNum: %d, txNumMin: %d, blockNum: %d", txNum, txNumMin, blockNum) - } - header, err := api._blockReader.HeaderByNumber(ctx, overlayTx, blockNum) if err != nil { return nil, err @@ -564,8 +560,6 @@ func (api *APIImpl) GetTransactionReceipt(ctx context.Context, txnHash common.Ha return ethutils.MarshalReceipt(borReceipt, bortypes.NewBorTransaction(), chainConfig, block.HeaderNoCopy(), txnHash, false, false), nil } - var txnIndex = int(txNum - txNumMin - 1) - txn, err := api._blockReader.TxnByIdxInBlock(ctx, overlayTx, header.Number.Uint64(), txnIndex) if err != nil { return nil, err diff --git a/rpc/jsonrpc/eth_txs.go b/rpc/jsonrpc/eth_txs.go index 61451cc5172..41d81e8637d 100644 --- a/rpc/jsonrpc/eth_txs.go +++ b/rpc/jsonrpc/eth_txs.go @@ -20,7 +20,6 @@ import ( "bytes" "context" "errors" - "fmt" "github.com/holiman/uint256" @@ -70,15 +69,11 @@ func (api *APIImpl) GetTransactionByHash(ctx context.Context, txnHash common.Has } overlayTx := api.filters.WithOverlay(tx) - txNumMin, err := api._txNumReader.Min(ctx, overlayTx, blockNum) + txnIndex, err := api.txnIndexInBlock(ctx, overlayTx, blockNum, txNum, isBorStateSyncTx) if err != nil { return nil, err } - if txNumMin+1 > txNum && !isBorStateSyncTx { - return nil, fmt.Errorf("uint underflow txnums error txNum: %d, txNumMin: %d, blockNum: %d", txNum, txNumMin, blockNum) - } - header, err := api._blockReader.HeaderByNumber(ctx, overlayTx, blockNum) if err != nil { return nil, err @@ -106,14 +101,12 @@ func (api *APIImpl) GetTransactionByHash(ctx context.Context, txnHash common.Has return ethapi.NewRPCBorTransaction(borTx, txnHash, blockHash, blockNum, uint64(txCount), chainConfig.ChainID), nil } - var txnIndex = txNum - txNumMin - 1 - - txn, err := api._txnReader.TxnByIdxInBlock(ctx, tx, blockNum, int(txnIndex)) + txn, err := api._txnReader.TxnByIdxInBlock(ctx, tx, blockNum, txnIndex) if err != nil { return nil, err } - return ethapi.NewRPCTransaction(txn, blockHash, blockTime, blockNum, txnIndex, baseFee), nil + return ethapi.NewRPCTransaction(txn, blockHash, blockTime, blockNum, uint64(txnIndex), baseFee), nil } curHeader := rawdb.ReadCurrentHeader(tx) @@ -166,14 +159,10 @@ func (api *APIImpl) GetRawTransactionByHash(ctx context.Context, hash common.Has return nil, err } - txNumMin, err := api._txNumReader.Min(ctx, tx, blockNum) + txnIndex, err := api.txnIndexInBlock(ctx, tx, blockNum, txNum, false) if err != nil { return nil, err } - if txNumMin+1 > txNum { - return nil, fmt.Errorf("uint underflow txnums error txNum: %d, txNumMin: %d, blockNum: %d", txNum, txNumMin, blockNum) - } - txnIndex := int(txNum - txNumMin - 1) txn, err := api._txnReader.TxnByIdxInBlock(ctx, tx, blockNum, txnIndex) if err != nil { return nil, err diff --git a/rpc/jsonrpc/trace_adhoc.go b/rpc/jsonrpc/trace_adhoc.go index cf9433eeb4c..c14bce80fb4 100644 --- a/rpc/jsonrpc/trace_adhoc.go +++ b/rpc/jsonrpc/trace_adhoc.go @@ -895,27 +895,12 @@ func (api *TraceAPIImpl) ReplayTransaction(ctx context.Context, txHash common.Ha return nil, err } - var isBorStateSyncTxn bool - blockNum, txNum, ok, err := api.txnLookup(ctx, tx, txHash) + blockNum, txNum, isBorStateSyncTxn, ok, err := api.txnLookupWithBorFallback(ctx, tx, txHash, chainConfig) if err != nil { return nil, err } - if !ok { - if chainConfig.Bor == nil { - return nil, nil - } - - // otherwise this may be a bor state sync transaction - check - blockNum, ok, err = api.bridgeReader.EventTxnLookup(ctx, txHash) - if err != nil { - return nil, err - } - if !ok { - return nil, nil - } - - isBorStateSyncTxn = true + return nil, nil } err = api.BaseAPI.checkPruneHistory(ctx, tx, blockNum) @@ -928,17 +913,11 @@ func (api *TraceAPIImpl) ReplayTransaction(ctx context.Context, txHash common.Ha return nil, err } - txNumMin, err := api._txNumReader.Min(ctx, tx, blockNum) + txnIndex, err := api.txnIndexInBlock(ctx, tx, blockNum, txNum, isBorStateSyncTxn) if err != nil { return nil, err } - if txNumMin+1 > txNum && !isBorStateSyncTxn { - return nil, fmt.Errorf("uint underflow txnums error txNum: %d, txNumMin: %d, blockNum: %d", txNum, txNumMin, blockNum) - } - - var txnIndex = int(txNum - txNumMin - 1) - if isBorStateSyncTxn { txnIndex = -1 } diff --git a/rpc/jsonrpc/trace_filtering.go b/rpc/jsonrpc/trace_filtering.go index dd0921a7895..5dc78928ab2 100644 --- a/rpc/jsonrpc/trace_filtering.go +++ b/rpc/jsonrpc/trace_filtering.go @@ -84,27 +84,12 @@ func (api *TraceAPIImpl) Transaction(ctx context.Context, txHash common.Hash, ga return nil, err } - var isBorStateSyncTxn bool - blockNumber, txNum, ok, err := api.txnLookup(ctx, tx, txHash) + blockNumber, txNum, isBorStateSyncTxn, ok, err := api.txnLookupWithBorFallback(ctx, tx, txHash, chainConfig) if err != nil { return nil, err } - if !ok { - if chainConfig.Bor == nil { - return nil, nil - } - - // otherwise this may be a bor state sync transaction - check - blockNumber, ok, err = api.bridgeReader.EventTxnLookup(ctx, txHash) - if err != nil { - return nil, err - } - if !ok { - return nil, nil - } - - isBorStateSyncTxn = true + return nil, nil } err = api.BaseAPI.checkPruneHistory(ctx, tx, blockNumber) @@ -120,17 +105,11 @@ func (api *TraceAPIImpl) Transaction(ctx context.Context, txHash common.Hash, ga return nil, nil } - txNumMin, err := api._txNumReader.Min(ctx, tx, blockNumber) + txIndex, err := api.txnIndexInBlock(ctx, tx, blockNumber, txNum, isBorStateSyncTxn) if err != nil { return nil, err } - if txNumMin+1 > txNum && !isBorStateSyncTxn { - return nil, fmt.Errorf("uint underflow txnums error txNum: %d, txNumMin: %d, blockNum: %d", txNum, txNumMin, blockNumber) - } - - var txIndex = int(txNum - txNumMin - 1) - if isBorStateSyncTxn { txIndex = -1 } diff --git a/rpc/jsonrpc/tracing.go b/rpc/jsonrpc/tracing.go index ed0c1e892c8..5f6dd9bba5e 100644 --- a/rpc/jsonrpc/tracing.go +++ b/rpc/jsonrpc/tracing.go @@ -258,31 +258,16 @@ func (api *DebugAPIImpl) TraceTransaction(ctx context.Context, hash common.Hash, return err } // Retrieve the transaction and assemble its EVM context - var isBorStateSyncTxn bool - blockNum, txNum, ok, err := api.txnLookup(ctx, tx, hash) + blockNum, txNum, isBorStateSyncTxn, ok, err := api.txnLookupWithBorFallback(ctx, tx, hash, chainConfig) if err != nil { return err } - if !ok { - if chainConfig.Bor == nil { - return fmt.Errorf("transaction not found") - } - - // otherwise this may be a bor state sync transaction - check - blockNum, ok, err = api.bridgeReader.EventTxnLookup(ctx, hash) - if err != nil { - return err - } - if !ok { - return fmt.Errorf("transaction not found") - } - if config == nil || config.BorTraceEnabled == nil || !*config.BorTraceEnabled { - stream.WriteEmptyArray() // matches maticnetwork/bor API behaviour for consistency - return nil - } - - isBorStateSyncTxn = true + return fmt.Errorf("transaction not found") + } + if isBorStateSyncTxn && (config == nil || config.BorTraceEnabled == nil || !*config.BorTraceEnabled) { + stream.WriteEmptyArray() // matches maticnetwork/bor API behaviour for consistency + return nil } if blockNum == 0 { @@ -309,14 +294,10 @@ func (api *DebugAPIImpl) TraceTransaction(ctx context.Context, hash common.Hash, // bor state sync txn is appended at the end of the block txnIndex = block.Transactions().Len() } else { - txNumMin, err := api._txNumReader.Min(ctx, tx, blockNum) + txnIndex, err = api.txnIndexInBlock(ctx, tx, blockNum, txNum, false) if err != nil { return err } - if txNumMin+1 > txNum { - return fmt.Errorf("uint underflow txnums error txNum: %d, txNumMin: %d, blockNum: %d", txNum, txNumMin, blockNum) - } - txnIndex = int(txNum - txNumMin - 1) if txnIndex >= block.Transactions().Len() { return fmt.Errorf("transaction %#x not found", hash) } From 22c17aa3e18f60e291fa555df942bbde0fef3333 Mon Sep 17 00:00:00 2001 From: yperbasis Date: Thu, 2 Jul 2026 22:27:56 +0200 Subject: [PATCH 2/8] rpc/jsonrpc: extract getReceiptsWithBor helper eth_getBlockReceipts, erigon_getBlockReceiptsByBlockHash and debug_getRawReceipts shared the same tail: getReceipts, then on bor chains fetch state sync events and generate the synthetic bor receipt. Collapse it into BaseAPI.getReceiptsWithBor. The bor receipt is returned separately rather than appended because the three call sites marshal it differently (eth marshals with block timestamp, erigon uses the legacy format without it, debug appends it raw); marshalling stays at the call sites with their exact flags. debug_getRawReceipts now reads chainConfig before generating receipts instead of after; both are adjacent error returns, and chainConfig is also read inside getReceipts itself, so the response is unchanged. TestGetRawReceipts pins the previously untested debug_getRawReceipts non-bor path against getReceipts+MarshalBinary; the bor tail is not reachable from this package's harness (no bor test chain) and is covered by code review. --- rpc/jsonrpc/debug_api.go | 18 ++++---------- rpc/jsonrpc/debug_api_test.go | 34 +++++++++++++++++++++++++++ rpc/jsonrpc/erigon_receipts.go | 18 +++----------- rpc/jsonrpc/eth_receipts.go | 43 ++++++++++++++++++++++------------ 4 files changed, 69 insertions(+), 44 deletions(-) diff --git a/rpc/jsonrpc/debug_api.go b/rpc/jsonrpc/debug_api.go index 62a4663e740..f8515afde2d 100644 --- a/rpc/jsonrpc/debug_api.go +++ b/rpc/jsonrpc/debug_api.go @@ -699,26 +699,16 @@ func (api *DebugAPIImpl) GetRawReceipts(ctx context.Context, blockNrOrHash rpc.B if block == nil { return nil, nil } - receipts, err := api.getReceipts(ctx, tx, block) + chainConfig, err := api.chainConfig(ctx, tx) if err != nil { return nil, err } - chainConfig, err := api.chainConfig(ctx, tx) + receipts, borReceipt, err := api.getReceiptsWithBor(ctx, tx, chainConfig, block) if err != nil { return nil, err } - if chainConfig.Bor != nil { - events, err := api.bridgeReader.Events(ctx, block.Hash(), blockNum) - if err != nil { - return nil, err - } - if len(events) != 0 { - borReceipt, err := api.borReceiptGenerator.GenerateBorReceipt(ctx, tx, block, events, chainConfig) - if err != nil { - return nil, err - } - receipts = append(receipts, borReceipt) - } + if borReceipt != nil { + receipts = append(receipts, borReceipt) } result := make([]hexutil.Bytes, len(receipts)) diff --git a/rpc/jsonrpc/debug_api_test.go b/rpc/jsonrpc/debug_api_test.go index baa1b98dbe3..85e83d71bbe 100644 --- a/rpc/jsonrpc/debug_api_test.go +++ b/rpc/jsonrpc/debug_api_test.go @@ -993,6 +993,40 @@ func TestGetRawTransaction(t *testing.T) { require.True(testedOnce, "Test flow didn't touch the target flow") } +func TestGetRawReceipts(t *testing.T) { + m, _, _ := rpcdaemontest.CreateTestExecModule(t) + api := NewPrivateDebugAPI(newBaseApiForTest(m), m.DB, nil, 5000000, false) + ctx := context.Background() + + require := require.New(t) + tx, err := m.DB.BeginTemporalRo(ctx) + require.NoError(err) + defer tx.Rollback() + number := *rawdb.ReadCurrentBlockNumber(tx) + + testedNonEmpty := false + for i := uint64(0); i <= number; i++ { + block, err := api.blockByNumberWithSenders(ctx, tx, i) + require.NoError(err) + receipts, err := api.getReceipts(ctx, tx, block) + require.NoError(err) + expected := make([]hexutil.Bytes, len(receipts)) + for j, receipt := range receipts { + b, err := receipt.MarshalBinary() + require.NoError(err) + expected[j] = b + } + + raw, err := api.GetRawReceipts(ctx, rpc.BlockNumberOrHashWithNumber(rpc.BlockNumber(i))) + require.NoError(err) + require.Equal(expected, raw) + if len(raw) > 0 { + testedNonEmpty = true + } + } + require.True(testedNonEmpty, "test chain has no receipts") +} + func TestExecutionWitness(t *testing.T) { // Enable historical commitment schema so the test aggregator maintains per-block history. previousSchema := statecfg.Schema diff --git a/rpc/jsonrpc/erigon_receipts.go b/rpc/jsonrpc/erigon_receipts.go index 80e25c05d19..fb2a5f63703 100644 --- a/rpc/jsonrpc/erigon_receipts.go +++ b/rpc/jsonrpc/erigon_receipts.go @@ -420,7 +420,7 @@ func (api *ErigonImpl) GetBlockReceiptsByBlockHash(ctx context.Context, cannonic if err != nil { return nil, err } - receipts, err := api.getReceipts(ctx, tx, block) + receipts, borReceipt, err := api.getReceiptsWithBor(ctx, tx, chainConfig, block) if err != nil { return nil, err } @@ -431,20 +431,8 @@ func (api *ErigonImpl) GetBlockReceiptsByBlockHash(ctx context.Context, cannonic result = append(result, ethutils.MarshalReceipt(receipt, txn, chainConfig, block.HeaderNoCopy(), txn.Hash(), true, false)) } - if chainConfig.Bor != nil { - events, err := api.bridgeReader.Events(ctx, block.Hash(), blockNum) - if err != nil { - return nil, err - } - - if len(events) != 0 { - borReceipt, err := api.borReceiptGenerator.GenerateBorReceipt(ctx, tx, block, events, chainConfig) - if err != nil { - return nil, err - } - - result = append(result, ethutils.MarshalReceipt(borReceipt, bortypes.NewBorTransaction(), chainConfig, block.HeaderNoCopy(), borReceipt.TxHash, false, false)) - } + if borReceipt != nil { + result = append(result, ethutils.MarshalReceipt(borReceipt, bortypes.NewBorTransaction(), chainConfig, block.HeaderNoCopy(), borReceipt.TxHash, false, false)) } return result, nil diff --git a/rpc/jsonrpc/eth_receipts.go b/rpc/jsonrpc/eth_receipts.go index bd2a3f634b7..07f5f1f9b81 100644 --- a/rpc/jsonrpc/eth_receipts.go +++ b/rpc/jsonrpc/eth_receipts.go @@ -77,6 +77,31 @@ func (api *BaseAPI) getReceipts(ctx context.Context, tx kv.TemporalTx, block *ty return api.receiptsGenerator.GetReceipts(ctx, chainConfig, tx, block, eth.ReceiptsOpts{CommitmentHistoryEnabled: commitmentHistoryEnabled}) } +// getReceiptsWithBor returns the block's receipts plus, on bor chains with state sync +// events in this block, the synthetic bor receipt (nil otherwise). The bor receipt is +// returned separately because call sites marshal it differently from regular receipts. +func (api *BaseAPI) getReceiptsWithBor(ctx context.Context, tx kv.TemporalTx, chainConfig *chain.Config, block *types.Block) (types.Receipts, *types.Receipt, error) { + receipts, err := api.getReceipts(ctx, tx, block) + if err != nil { + return nil, nil, err + } + if chainConfig.Bor == nil { + return receipts, nil, nil + } + events, err := api.bridgeReader.Events(ctx, block.Hash(), block.NumberU64()) + if err != nil { + return nil, nil, err + } + if len(events) == 0 { + return receipts, nil, nil + } + borReceipt, err := api.borReceiptGenerator.GenerateBorReceipt(ctx, tx, block, events, chainConfig) + if err != nil { + return nil, nil, err + } + return receipts, borReceipt, nil +} + func (api *BaseAPI) getReceipt(ctx context.Context, cc *chain.Config, tx kv.TemporalTx, header *types.Header, txn types.Transaction, index int, txNum uint64, postState *receipts.PostStateInfo) (*types.Receipt, error) { return api.receiptsGenerator.GetReceipt(ctx, cc, tx, header, txn, index, txNum, postState) } @@ -634,7 +659,7 @@ func (api *APIImpl) GetBlockReceipts(ctx context.Context, numberOrHash rpc.Block if err != nil { return nil, err } - receipts, err := api.getReceipts(ctx, tx, block) + receipts, borReceipt, err := api.getReceiptsWithBor(ctx, tx, chainConfig, block) if err != nil { return nil, err } @@ -644,20 +669,8 @@ func (api *APIImpl) GetBlockReceipts(ctx context.Context, numberOrHash rpc.Block result = append(result, ethutils.MarshalReceipt(receipt, txn, chainConfig, block.HeaderNoCopy(), txn.Hash(), true, true)) } - if chainConfig.Bor != nil { - events, err := api.bridgeReader.Events(ctx, block.Hash(), blockNum) - if err != nil { - return nil, err - } - - if len(events) != 0 { - borReceipt, err := api.borReceiptGenerator.GenerateBorReceipt(ctx, tx, block, events, chainConfig) - if err != nil { - return nil, err - } - - result = append(result, ethutils.MarshalReceipt(borReceipt, bortypes.NewBorTransaction(), chainConfig, block.HeaderNoCopy(), borReceipt.TxHash, false, true)) - } + if borReceipt != nil { + result = append(result, ethutils.MarshalReceipt(borReceipt, bortypes.NewBorTransaction(), chainConfig, block.HeaderNoCopy(), borReceipt.TxHash, false, true)) } return result, nil From 01d7083d236efd2abf4389720ba1ecdcbee0f61b Mon Sep 17 00:00:00 2001 From: yperbasis Date: Thu, 2 Jul 2026 22:33:41 +0200 Subject: [PATCH 3/8] rpc/jsonrpc: extract log block-range resolution helpers Two deliberately separate helpers, matching the two distinct dialects in use: - logRangeLatestOnly (erigon_receipts.go): the byte-identical blocks in erigon_getLogs and erigon_getLatestLogs that accept only 'latest' as a negative tag and reject the rest with a CustomError. - BaseAPI.resolveLogsRange: the resolving dialect from overlay's getBeginEnd, now also used by eth_getLogs. Negative tags are resolved via rpchelper.GetBlockNumber against the latest executed block. The one real delta is parameterized: eth_getLogs rejects ranges past the head (checkFuture) as each bound is resolved, overlay does not. eth_getLogs' blockHash+range guard is hoisted above the helper call (fires under exactly the same condition as before), and each caller keeps its own end= 0 { + begin = crit.FromBlock.Uint64() + } else if !crit.FromBlock.IsInt64() || crit.FromBlock.Int64() != int64(rpc.LatestBlockNumber) { + return 0, 0, &rpc.CustomError{Message: fmt.Sprintf("negative value for FromBlock: %v", crit.FromBlock), Code: rpc.ErrCodeInvalidParams} + } + } + end = latest + if crit.ToBlock != nil { + if crit.ToBlock.Sign() >= 0 { + end = crit.ToBlock.Uint64() + } else if !crit.ToBlock.IsInt64() || crit.ToBlock.Int64() != int64(rpc.LatestBlockNumber) { + return 0, 0, &rpc.CustomError{Message: fmt.Sprintf("negative value for ToBlock: %v", crit.ToBlock), Code: rpc.ErrCodeInvalidParams} + } + } + return begin, end, nil +} + // GetLogs implements erigon_getLogs. Returns an array of logs matching a given filter object. func (api *ErigonImpl) GetLogs(ctx context.Context, crit filters.FilterCriteria) (types.ErigonLogs, error) { var begin, end uint64 @@ -97,28 +125,11 @@ func (api *ErigonImpl) GetLogs(ctx context.Context, crit filters.FilterCriteria) end = header.Number.Uint64() } else { - // Convert the RPC block numbers into internal representations - latest, err := rpchelper.GetLatestBlockNumber(tx) + var err error + begin, end, err = logRangeLatestOnly(tx, crit) if err != nil { return nil, err } - - begin = 0 - if crit.FromBlock != nil { - if crit.FromBlock.Sign() >= 0 { - begin = crit.FromBlock.Uint64() - } else if !crit.FromBlock.IsInt64() || crit.FromBlock.Int64() != int64(rpc.LatestBlockNumber) { - return nil, &rpc.CustomError{Message: fmt.Sprintf("negative value for FromBlock: %v", crit.FromBlock), Code: rpc.ErrCodeInvalidParams} - } - } - end = latest - if crit.ToBlock != nil { - if crit.ToBlock.Sign() >= 0 { - end = crit.ToBlock.Uint64() - } else if !crit.ToBlock.IsInt64() || crit.ToBlock.Int64() != int64(rpc.LatestBlockNumber) { - return nil, &rpc.CustomError{Message: fmt.Sprintf("negative value for ToBlock: %v", crit.ToBlock), Code: rpc.ErrCodeInvalidParams} - } - } } if len(crit.Topics) > maxTopics { @@ -192,28 +203,10 @@ func (api *ErigonImpl) GetLatestLogs(ctx context.Context, crit filters.FilterCri begin = header.Number.Uint64() end = header.Number.Uint64() } else { - // Convert the RPC block numbers into internal representations - latest, err := rpchelper.GetLatestBlockNumber(tx) + begin, end, err = logRangeLatestOnly(tx, crit) if err != nil { return nil, err } - - begin = 0 - if crit.FromBlock != nil { - if crit.FromBlock.Sign() >= 0 { - begin = crit.FromBlock.Uint64() - } else if !crit.FromBlock.IsInt64() || crit.FromBlock.Int64() != int64(rpc.LatestBlockNumber) { - return nil, &rpc.CustomError{Message: fmt.Sprintf("negative value for FromBlock: %v", crit.FromBlock), Code: rpc.ErrCodeInvalidParams} - } - } - end = latest - if crit.ToBlock != nil { - if crit.ToBlock.Sign() >= 0 { - end = crit.ToBlock.Uint64() - } else if !crit.ToBlock.IsInt64() || crit.ToBlock.Int64() != int64(rpc.LatestBlockNumber) { - return nil, &rpc.CustomError{Message: fmt.Sprintf("negative value for ToBlock: %v", crit.ToBlock), Code: rpc.ErrCodeInvalidParams} - } - } } if end < begin { return nil, &rpc.CustomError{Message: fmt.Sprintf("invalid block range: end (%d) < begin (%d)", end, begin), Code: rpc.ErrCodeInvalidParams} diff --git a/rpc/jsonrpc/eth_receipts.go b/rpc/jsonrpc/eth_receipts.go index 07f5f1f9b81..af5cd0ffb80 100644 --- a/rpc/jsonrpc/eth_receipts.go +++ b/rpc/jsonrpc/eth_receipts.go @@ -114,9 +114,69 @@ func (api *BaseAPI) getCachedReceipts(ctx context.Context, hash common.Hash) (ty return api.receiptsGenerator.GetCachedReceipts(ctx, hash) } +// resolveLogsRange resolves a filter's block range. A BlockHash pins the range to that +// block; otherwise negative tags are resolved against the chain, defaulting to the +// latest executed block. With checkFuture, ranges past the latest executed block are +// rejected as they are resolved. +func (api *BaseAPI) resolveLogsRange(ctx context.Context, tx kv.Tx, crit filters.FilterCriteria, checkFuture bool) (begin, end uint64, err error) { + if crit.BlockHash != nil { + block, err := api.blockByHashWithSenders(ctx, tx, *crit.BlockHash) + if err != nil { + return 0, 0, err + } + if block == nil { + return 0, 0, fmt.Errorf("block not found: %x", *crit.BlockHash) + } + + num := block.NumberU64() + return num, num, nil + } + + // Convert the RPC block numbers into internal representations + latest, _, _, err := rpchelper.GetBlockNumber(ctx, rpc.BlockNumberOrHashWithNumber(rpc.LatestExecutedBlockNumber), tx, api._blockReader, nil) + if err != nil { + return 0, 0, err + } + + begin = latest + if crit.FromBlock != nil { + fromBlock := crit.FromBlock.Int64() + if fromBlock > 0 { + begin = uint64(fromBlock) + } else { + blockNum := rpc.BlockNumber(fromBlock) + begin, _, _, err = rpchelper.GetBlockNumber(ctx, rpc.BlockNumberOrHashWithNumber(blockNum), tx, api._blockReader, api.filters) + if err != nil { + return 0, 0, err + } + } + + if checkFuture && begin > latest { + return 0, 0, &rpc.CustomError{Message: ErrBlockRangeIntoFuture, Code: rpc.ErrCodeInvalidParams} + } + } + end = latest + if crit.ToBlock != nil { + toBlock := crit.ToBlock.Int64() + if toBlock > 0 { + end = uint64(toBlock) + } else { + blockNum := rpc.BlockNumber(toBlock) + end, _, _, err = rpchelper.GetBlockNumber(ctx, rpc.BlockNumberOrHashWithNumber(blockNum), tx, api._blockReader, api.filters) + if err != nil { + return 0, 0, err + } + } + + if checkFuture && end > latest { + return 0, 0, &rpc.CustomError{Message: ErrBlockRangeIntoFuture, Code: rpc.ErrCodeInvalidParams} + } + } + return begin, end, nil +} + // GetLogs implements eth_getLogs. Returns an array of logs matching a given filter object. func (api *APIImpl) GetLogs(ctx context.Context, crit filters.FilterCriteria) (types.RPCLogs, error) { - var begin, end uint64 logs := types.RPCLogs{} tx, beginErr := api.db.BeginTemporalRo(ctx) @@ -138,63 +198,13 @@ func (api *APIImpl) GetLogs(ctx context.Context, crit filters.FilterCriteria) (t } } - if crit.BlockHash != nil { - if crit.FromBlock != nil || crit.ToBlock != nil { - return nil, &rpc.CustomError{Message: errBlockHashWithRange, Code: rpc.ErrCodeInvalidParams} - } - - block, err := api.blockByHashWithSenders(ctx, tx, *crit.BlockHash) - if err != nil { - return nil, err - } - if block == nil { - return nil, fmt.Errorf("block not found: %x", *crit.BlockHash) - } - - num := block.NumberU64() - begin = num - end = num - } else { - // Convert the RPC block numbers into internal representations - latest, _, _, err := rpchelper.GetBlockNumber(ctx, rpc.BlockNumberOrHashWithNumber(rpc.LatestExecutedBlockNumber), tx, api._blockReader, nil) - if err != nil { - return nil, err - } - - begin = latest - if crit.FromBlock != nil { - fromBlock := crit.FromBlock.Int64() - if fromBlock > 0 { - begin = uint64(fromBlock) - } else { - blockNum := rpc.BlockNumber(fromBlock) - begin, _, _, err = rpchelper.GetBlockNumber(ctx, rpc.BlockNumberOrHashWithNumber(blockNum), tx, api._blockReader, api.filters) - if err != nil { - return nil, err - } - } - - if begin > latest { - return nil, &rpc.CustomError{Message: ErrBlockRangeIntoFuture, Code: rpc.ErrCodeInvalidParams} - } - } - end = latest - if crit.ToBlock != nil { - toBlock := crit.ToBlock.Int64() - if toBlock > 0 { - end = uint64(toBlock) - } else { - blockNum := rpc.BlockNumber(toBlock) - end, _, _, err = rpchelper.GetBlockNumber(ctx, rpc.BlockNumberOrHashWithNumber(blockNum), tx, api._blockReader, api.filters) - if err != nil { - return nil, err - } - } + if crit.BlockHash != nil && (crit.FromBlock != nil || crit.ToBlock != nil) { + return nil, &rpc.CustomError{Message: errBlockHashWithRange, Code: rpc.ErrCodeInvalidParams} + } - if end > latest { - return nil, &rpc.CustomError{Message: ErrBlockRangeIntoFuture, Code: rpc.ErrCodeInvalidParams} - } - } + begin, end, err := api.resolveLogsRange(ctx, tx, crit, true) + if err != nil { + return nil, err } if end < begin { diff --git a/rpc/jsonrpc/overlay_api.go b/rpc/jsonrpc/overlay_api.go index 97e09af02f5..f6b173cdd8f 100644 --- a/rpc/jsonrpc/overlay_api.go +++ b/rpc/jsonrpc/overlay_api.go @@ -575,54 +575,9 @@ func (api *OverlayAPIImpl) replayBlock(ctx context.Context, blockNum uint64, sta } func getBeginEnd(ctx context.Context, tx kv.Tx, api *OverlayAPIImpl, crit filters.FilterCriteria) (uint64, uint64, error) { - var begin, end uint64 - if crit.BlockHash != nil { - block, err := api.blockByHashWithSenders(ctx, tx, *crit.BlockHash) - if err != nil { - return 0, 0, err - } - - if block == nil { - return 0, 0, fmt.Errorf("block not found: %x", *crit.BlockHash) - } - - num := block.NumberU64() - begin = num - end = num - } else { - // Convert the RPC block numbers into internal representations - latest, _, _, err := rpchelper.GetBlockNumber(ctx, rpc.BlockNumberOrHashWithNumber(rpc.LatestExecutedBlockNumber), tx, api._blockReader, nil) - if err != nil { - return 0, 0, err - } - - begin = latest - if crit.FromBlock != nil { - fromBlock := crit.FromBlock.Int64() - if fromBlock > 0 { - begin = uint64(fromBlock) - } else { - blockNum := rpc.BlockNumber(fromBlock) - begin, _, _, err = rpchelper.GetBlockNumber(ctx, rpc.BlockNumberOrHashWithNumber(blockNum), tx, api._blockReader, api.filters) - if err != nil { - return 0, 0, err - } - } - - } - end = latest - if crit.ToBlock != nil { - toBlock := crit.ToBlock.Int64() - if toBlock > 0 { - end = uint64(toBlock) - } else { - blockNum := rpc.BlockNumber(toBlock) - end, _, _, err = rpchelper.GetBlockNumber(ctx, rpc.BlockNumberOrHashWithNumber(blockNum), tx, api._blockReader, api.filters) - if err != nil { - return 0, 0, err - } - } - } + begin, end, err := api.resolveLogsRange(ctx, tx, crit, false) + if err != nil { + return 0, 0, err } if end < begin { diff --git a/rpc/jsonrpc/overlay_api_test.go b/rpc/jsonrpc/overlay_api_test.go new file mode 100644 index 00000000000..7f7f517d807 --- /dev/null +++ b/rpc/jsonrpc/overlay_api_test.go @@ -0,0 +1,61 @@ +// Copyright 2026 The Erigon Authors +// This file is part of Erigon. +// +// Erigon is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// Erigon is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with Erigon. If not, see . + +package jsonrpc + +import ( + "math/big" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/erigontech/erigon/cmd/rpcdaemon/rpcdaemontest" + "github.com/erigontech/erigon/rpc" + "github.com/erigontech/erigon/rpc/filters" + "github.com/erigontech/erigon/rpc/rpchelper" +) + +func TestOverlayGetBeginEnd(t *testing.T) { + m, _, _ := rpcdaemontest.CreateTestExecModule(t) + api := &OverlayAPIImpl{BaseAPI: newBaseApiForTest(m)} + tx, err := m.DB.BeginTemporalRo(m.Ctx) + require.NoError(t, err) + defer tx.Rollback() + + latestExecuted, _, _, err := rpchelper.GetBlockNumber(m.Ctx, rpc.BlockNumberOrHashWithNumber(rpc.LatestExecutedBlockNumber), tx, api._blockReader, nil) + require.NoError(t, err) + + begin, end, err := getBeginEnd(m.Ctx, tx, api, filters.FilterCriteria{FromBlock: big.NewInt(2), ToBlock: big.NewInt(5)}) + require.NoError(t, err) + require.Equal(t, uint64(2), begin) + require.Equal(t, uint64(5), end) + + begin, end, err = getBeginEnd(m.Ctx, tx, api, filters.FilterCriteria{}) + require.NoError(t, err) + require.Equal(t, latestExecuted, begin) + require.Equal(t, latestExecuted, end) + + _, _, err = getBeginEnd(m.Ctx, tx, api, filters.FilterCriteria{FromBlock: big.NewInt(5), ToBlock: big.NewInt(2)}) + require.EqualError(t, err, "end (2) < begin (5)") + + block, err := api._blockReader.BlockByNumber(m.Ctx, tx, 1) + require.NoError(t, err) + blockHash := block.Hash() + begin, end, err = getBeginEnd(m.Ctx, tx, api, filters.FilterCriteria{BlockHash: &blockHash}) + require.NoError(t, err) + require.Equal(t, uint64(1), begin) + require.Equal(t, uint64(1), end) +} From 1db3ff834974e5f7157b0de5cbc3c84f0a10071c Mon Sep 17 00:00:00 2001 From: yperbasis Date: Thu, 2 Jul 2026 22:38:01 +0200 Subject: [PATCH 4/8] rpc/jsonrpc: extract generic subscription skeleton in eth_filters The five subscription methods (eth_newHeads, eth_newPendingTransactions, eth_newPendingTransactionsWithBody, eth_logs, eth_transactionReceipts) shared the same wrapper: nil-filters guard, NotifierFromContext guard, CreateSubscription, and a goroutine with dbg.LogPanic pumping the filter channel into the notifier with defer-unsubscribe and channel-closed warning. Collapse it into subscribeRPC[T]; each method supplies its subscribe closure, per-item notify logic and closed-channel message. NewPendingTransactionsWithBody delegates to the same core as NewPendingTransactions with fullTx=true; their bodies differed only in that flag and the channel buffer size (256 vs 512), which stays a parameter. TestSubscriptionsRequireFiltersAndNotifier pins the guard behavior (ErrNotificationsUnsupported + empty subscription) for all five methods with both nil filters and a notifier-less context; the notify path is covered end-to-end by TestSendRawTransactionSync's eth_newPendingTransactions subscription. --- rpc/jsonrpc/eth_filters.go | 231 +++++++++++--------------------- rpc/jsonrpc/eth_filters_test.go | 30 +++++ 2 files changed, 107 insertions(+), 154 deletions(-) diff --git a/rpc/jsonrpc/eth_filters.go b/rpc/jsonrpc/eth_filters.go index ce90d4e7779..e5d916bf3f3 100644 --- a/rpc/jsonrpc/eth_filters.go +++ b/rpc/jsonrpc/eth_filters.go @@ -24,6 +24,7 @@ import ( "github.com/erigontech/erigon/common/log/v3" "github.com/erigontech/erigon/execution/types" "github.com/erigontech/erigon/execution/types/ethutils" + "github.com/erigontech/erigon/node/gointerfaces/remoteproto" "github.com/erigontech/erigon/rpc" "github.com/erigontech/erigon/rpc/filters" "github.com/erigontech/erigon/rpc/rpchelper" @@ -150,9 +151,12 @@ func (api *APIImpl) GetFilterLogs(_ context.Context, index string) ([]*types.Log return []*types.Log{}, nil } -// NewHeads send a notification each time a new (header) block is appended to the chain. -func (api *APIImpl) NewHeads(ctx context.Context) (*rpc.Subscription, error) { - if api.filters == nil { +// subscribeRPC runs the shared subscription skeleton: guard checks, subscription +// creation, and a goroutine that pumps items from the filter channel into notify until +// the channel closes or the client goes away. subscribe is called inside the goroutine +// and must return the item channel plus an unsubscribe func. +func subscribeRPC[T any](ctx context.Context, apiFilters *rpchelper.Filters, subscribe func() (<-chan T, func()), notify func(notifier rpc.Notifier, rpcSub *rpc.Subscription, item T), closedWarn string) (*rpc.Subscription, error) { + if apiFilters == nil { return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported } notifier, supported := rpc.NotifierFromContext(ctx) @@ -164,19 +168,15 @@ func (api *APIImpl) NewHeads(ctx context.Context) (*rpc.Subscription, error) { go func() { defer dbg.LogPanic() - headers, id := api.filters.SubscribeNewHeads(32) - defer api.filters.UnsubscribeHeads(id) + ch, unsubscribe := subscribe() + defer unsubscribe() + for { select { - case h, ok := <-headers: - if h != nil { - err := notifier.Notify(rpcSub.ID, h) - if err != nil { - log.Warn("[rpc] error while notifying subscription", "err", err) - } - } + case item, ok := <-ch: + notify(notifier, rpcSub, item) if !ok { - log.Warn("[rpc] new heads channel was closed") + log.Warn(closedWarn) return } case <-rpcSub.Err(): @@ -188,169 +188,92 @@ func (api *APIImpl) NewHeads(ctx context.Context) (*rpc.Subscription, error) { return rpcSub, nil } -// NewPendingTransactions send a notification each time when a transaction had added into mempool. -func (api *APIImpl) NewPendingTransactions(ctx context.Context, fullTx *bool) (*rpc.Subscription, error) { - if api.filters == nil { - return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported - } - notifier, supported := rpc.NotifierFromContext(ctx) - if !supported { - return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported - } - - rpcSub := notifier.CreateSubscription() - - go func() { - defer dbg.LogPanic() - txsCh, id := api.filters.SubscribePendingTxs(256) - defer api.filters.UnsubscribePendingTxs(id) - - for { - select { - case txs, ok := <-txsCh: - for _, t := range txs { - if t != nil { - var err error - if fullTx != nil && *fullTx { - err = notifier.Notify(rpcSub.ID, newRPCPendingTransaction(t, nil, nil)) - } else { - err = notifier.Notify(rpcSub.ID, t.Hash()) - } - - if err != nil { - log.Warn("[rpc] error while notifying subscription", "err", err) - } - } - } - if !ok { - log.Warn("[rpc] new pending transactions channel was closed") - return +// NewHeads send a notification each time a new (header) block is appended to the chain. +func (api *APIImpl) NewHeads(ctx context.Context) (*rpc.Subscription, error) { + return subscribeRPC(ctx, api.filters, + func() (<-chan *types.Header, func()) { + headers, id := api.filters.SubscribeNewHeads(32) + return headers, func() { api.filters.UnsubscribeHeads(id) } + }, + func(notifier rpc.Notifier, rpcSub *rpc.Subscription, h *types.Header) { + if h != nil { + err := notifier.Notify(rpcSub.ID, h) + if err != nil { + log.Warn("[rpc] error while notifying subscription", "err", err) } - case <-rpcSub.Err(): - return } - } - }() + }, + "[rpc] new heads channel was closed") +} - return rpcSub, nil +// NewPendingTransactions send a notification each time when a transaction had added into mempool. +func (api *APIImpl) NewPendingTransactions(ctx context.Context, fullTx *bool) (*rpc.Subscription, error) { + return api.subscribePendingTransactions(ctx, 256, fullTx != nil && *fullTx) } // NewPendingTransactionsWithBody send a notification each time when a transaction had added into mempool. func (api *APIImpl) NewPendingTransactionsWithBody(ctx context.Context) (*rpc.Subscription, error) { - if api.filters == nil { - return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported - } - notifier, supported := rpc.NotifierFromContext(ctx) - if !supported { - return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported - } - - rpcSub := notifier.CreateSubscription() + return api.subscribePendingTransactions(ctx, 512, true) +} - go func() { - defer dbg.LogPanic() - txsCh, id := api.filters.SubscribePendingTxs(512) - defer api.filters.UnsubscribePendingTxs(id) +func (api *APIImpl) subscribePendingTransactions(ctx context.Context, chanSize int, fullTx bool) (*rpc.Subscription, error) { + return subscribeRPC(ctx, api.filters, + func() (<-chan []types.Transaction, func()) { + txsCh, id := api.filters.SubscribePendingTxs(chanSize) + return txsCh, func() { api.filters.UnsubscribePendingTxs(id) } + }, + func(notifier rpc.Notifier, rpcSub *rpc.Subscription, txs []types.Transaction) { + for _, t := range txs { + if t != nil { + var err error + if fullTx { + err = notifier.Notify(rpcSub.ID, newRPCPendingTransaction(t, nil, nil)) + } else { + err = notifier.Notify(rpcSub.ID, t.Hash()) + } - for { - select { - case txs, ok := <-txsCh: - for _, t := range txs { - if t != nil { - err := notifier.Notify(rpcSub.ID, newRPCPendingTransaction(t, nil, nil)) - if err != nil { - log.Warn("[rpc] error while notifying subscription", "err", err) - } + if err != nil { + log.Warn("[rpc] error while notifying subscription", "err", err) } } - if !ok { - log.Warn("[rpc] new pending transactions channel was closed") - return - } - case <-rpcSub.Err(): - return } - } - }() - - return rpcSub, nil + }, + "[rpc] new pending transactions channel was closed") } // Logs send a notification each time a new log appears. func (api *APIImpl) Logs(ctx context.Context, crit filters.FilterCriteria) (*rpc.Subscription, error) { - if api.filters == nil { - return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported - } - notifier, supported := rpc.NotifierFromContext(ctx) - if !supported { - return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported - } - - rpcSub := notifier.CreateSubscription() - - go func() { - defer dbg.LogPanic() - logs, id := api.filters.SubscribeLogs(api.SubscribeLogsChannelSize, crit) - defer api.filters.UnsubscribeLogs(id) - - for { - select { - case h, ok := <-logs: - if h != nil { - err := notifier.Notify(rpcSub.ID, h) - if err != nil { - log.Warn("[rpc] error while notifying subscription", "err", err) - } - } - if !ok { - log.Warn("[rpc] log channel was closed") - return + return subscribeRPC(ctx, api.filters, + func() (<-chan *types.Log, func()) { + logs, id := api.filters.SubscribeLogs(api.SubscribeLogsChannelSize, crit) + return logs, func() { api.filters.UnsubscribeLogs(id) } + }, + func(notifier rpc.Notifier, rpcSub *rpc.Subscription, h *types.Log) { + if h != nil { + err := notifier.Notify(rpcSub.ID, h) + if err != nil { + log.Warn("[rpc] error while notifying subscription", "err", err) } - case <-rpcSub.Err(): - return } - } - }() - - return rpcSub, nil + }, + "[rpc] log channel was closed") } // TransactionReceipts send a notification each time a new receipt appears. func (api *APIImpl) TransactionReceipts(ctx context.Context, crit filters.ReceiptsFilterCriteria) (*rpc.Subscription, error) { - if api.filters == nil { - return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported - } - notifier, supported := rpc.NotifierFromContext(ctx) - if !supported { - return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported - } - - rpcSub := notifier.CreateSubscription() - - go func() { - defer dbg.LogPanic() - receipts, id := api.filters.SubscribeReceipts(api.SubscribeLogsChannelSize, crit) - defer api.filters.UnsubscribeReceipts(id) - - for { - select { - case protoReceipt, ok := <-receipts: - if protoReceipt != nil { - receipt := ethutils.MarshalSubscribeReceipt(protoReceipt) - err := notifier.Notify(rpcSub.ID, []map[string]any{receipt}) - if err != nil { - log.Warn("[rpc] error while notifying subscription", "err", err) - } + return subscribeRPC(ctx, api.filters, + func() (<-chan *remoteproto.SubscribeReceiptsReply, func()) { + receipts, id := api.filters.SubscribeReceipts(api.SubscribeLogsChannelSize, crit) + return receipts, func() { api.filters.UnsubscribeReceipts(id) } + }, + func(notifier rpc.Notifier, rpcSub *rpc.Subscription, protoReceipt *remoteproto.SubscribeReceiptsReply) { + if protoReceipt != nil { + receipt := ethutils.MarshalSubscribeReceipt(protoReceipt) + err := notifier.Notify(rpcSub.ID, []map[string]any{receipt}) + if err != nil { + log.Warn("[rpc] error while notifying subscription", "err", err) } - if !ok { - log.Warn("[rpc] receipts channel was closed") - return - } - case <-rpcSub.Err(): - return } - } - }() - - return rpcSub, nil + }, + "[rpc] receipts channel was closed") } diff --git a/rpc/jsonrpc/eth_filters_test.go b/rpc/jsonrpc/eth_filters_test.go index b0b5cc458eb..950b14c6376 100644 --- a/rpc/jsonrpc/eth_filters_test.go +++ b/rpc/jsonrpc/eth_filters_test.go @@ -44,6 +44,36 @@ func newBaseApiWithFiltersForTest(f *rpchelper.Filters, stateCache *kvcache.Cohe return NewBaseApi(f, stateCache, m.BlockReader, false, rpccfg.DefaultEvmCallTimeout, m.Engine, m.Dirs, nil, 0, 0) } +func TestSubscriptionsRequireFiltersAndNotifier(t *testing.T) { + m := execmoduletester.New(t) + ctx, conn := rpcdaemontest.CreateTestGrpcConn(t, m) + mining := txpoolproto.NewMiningClient(conn) + ff := rpchelper.New(ctx, rpchelper.DefaultFiltersConfig, nil, nil, mining, func() {}, m.Log, nil) + stateCache := kvcache.New(kvcache.DefaultCoherentConfig) + + apis := map[string]*APIImpl{ + "withFilters": newEthApiForTest(newBaseApiWithFiltersForTest(ff, stateCache, m), m.DB, nil, nil), + "nilFilters": newEthApiForTest(newBaseApiForTest(m), m.DB, nil, nil), + } + for apiName, api := range apis { + // ctx carries no rpc notifier, so every subscription method must refuse + subscriptions := map[string]func() (*rpc.Subscription, error){ + "newHeads": func() (*rpc.Subscription, error) { return api.NewHeads(ctx) }, + "newPendingTransactions": func() (*rpc.Subscription, error) { return api.NewPendingTransactions(ctx, nil) }, + "newPendingTransactionsWithBody": func() (*rpc.Subscription, error) { return api.NewPendingTransactionsWithBody(ctx) }, + "logs": func() (*rpc.Subscription, error) { return api.Logs(ctx, filters.FilterCriteria{}) }, + "transactionReceipts": func() (*rpc.Subscription, error) { + return api.TransactionReceipts(ctx, filters.ReceiptsFilterCriteria{}) + }, + } + for name, subscribe := range subscriptions { + sub, err := subscribe() + require.ErrorIs(t, err, rpc.ErrNotificationsUnsupported, "%s/%s", apiName, name) + require.Equal(t, &rpc.Subscription{}, sub, "%s/%s", apiName, name) + } + } +} + func TestNewFilters(t *testing.T) { if testing.Short() { t.Skip("slow test") From 4cf3871128bbaf6bcdf832c2e19eb554b7f80ecf Mon Sep 17 00:00:00 2001 From: yperbasis Date: Thu, 2 Jul 2026 22:43:45 +0200 Subject: [PATCH 5/8] rpc/jsonrpc: extract replay-env helpers Three helpers for the replay/simulation scaffolds: - BaseAPI.blockHashGetter: the BLOCKHASH resolver closure (overrideBlockHash map, then CanonicalHash with a debug log on miss) that appeared in eth_callMany, debug_traceCallMany and twice in the overlay API. The traceCallMany variant returned common.Hash{} explicitly on miss, but both BlockReader and RemoteBlockReader already return the zero hash whenever ok is false or err is set, so the variants are equivalent and merge into one. - validateBundles: the double empty-bundles check shared verbatim by eth_callMany and debug_traceCallMany, same 'empty bundles' error. - setupEVMTimeout: the evmPtr + WithTimeout/WithCancel + AfterFunc-cancel scaffold shared by eth_callBundle and eth_callMany; the returned cleanup preserves the original stop-before-cancel defer order, and per-call timeout derivation (5s literal vs evmCallTimeout default with different nil/positive guards) stays at the call sites. TestCallManyEmptyBundles pins the empty-bundles error for both eth_callMany and debug_traceCallMany before the loop+flag logic was rewritten as early returns; trace_adhoc/trace_filtering block stacks and overlay replayBlock's simpler timeout block are intentionally untouched. --- rpc/jsonrpc/eth_block.go | 21 +------ rpc/jsonrpc/eth_callMany.go | 95 ++++++++++++++++++-------------- rpc/jsonrpc/eth_callMany_test.go | 30 ++++++++++ rpc/jsonrpc/overlay_api.go | 22 +------- rpc/jsonrpc/tracing.go | 26 +-------- 5 files changed, 91 insertions(+), 103 deletions(-) diff --git a/rpc/jsonrpc/eth_block.go b/rpc/jsonrpc/eth_block.go index d8dca05e992..ccd69d268f4 100644 --- a/rpc/jsonrpc/eth_block.go +++ b/rpc/jsonrpc/eth_block.go @@ -20,7 +20,6 @@ import ( "context" "errors" "fmt" - "sync/atomic" "time" "github.com/erigontech/erigon/common" @@ -143,30 +142,14 @@ func (api *APIImpl) CallBundle(ctx context.Context, txHashes []common.Hash, stat // Get a new instance of the EVM evm := vm.NewEVM(blockCtx, txCtx, ibs, chainConfig, vm.Config{}) - // evmPtr is updated atomically each time evm is recreated in the loop, - // so the AfterFunc callback always cancels the current instance. - var evmPtr atomic.Pointer[vm.EVM] - evmPtr.Store(evm) - timeoutMilliSeconds := int64(5000) if timeoutMilliSecondsPtr != nil { timeoutMilliSeconds = *timeoutMilliSecondsPtr } timeout := time.Millisecond * time.Duration(timeoutMilliSeconds) - // Setup context so it may be cancelled the call has completed - // or, in case of unmetered gas, setup a context with a timeout. - var cancel context.CancelFunc - if timeout > 0 { - ctx, cancel = context.WithTimeout(ctx, timeout) - } else { - ctx, cancel = context.WithCancel(ctx) - } - // Make sure the context is cancelled when the call has completed - // this makes sure resources are cleaned up. - defer cancel() - stop := context.AfterFunc(ctx, func() { evmPtr.Load().Cancel() }) - defer stop() + evmPtr, cleanup := setupEVMTimeout(ctx, evm, timeout) + defer cleanup() // Setup the gas pool (also for unmetered requests) // and apply the message. diff --git a/rpc/jsonrpc/eth_callMany.go b/rpc/jsonrpc/eth_callMany.go index 64a3973380a..147ad41a51a 100644 --- a/rpc/jsonrpc/eth_callMany.go +++ b/rpc/jsonrpc/eth_callMany.go @@ -28,6 +28,7 @@ import ( "github.com/erigontech/erigon/common/hexutil" "github.com/erigontech/erigon/common/log/v3" "github.com/erigontech/erigon/common/math" + "github.com/erigontech/erigon/db/kv" "github.com/erigontech/erigon/execution/protocol" "github.com/erigontech/erigon/execution/state" "github.com/erigontech/erigon/execution/types" @@ -49,6 +50,54 @@ type StateContext struct { TransactionIndex *int } +func validateBundles(bundles []Bundle) error { + if len(bundles) == 0 { + return errors.New("empty bundles") + } + for _, bundle := range bundles { + if len(bundle.Transactions) != 0 { + return nil + } + } + return errors.New("empty bundles") +} + +// blockHashGetter returns a BLOCKHASH resolver that prefers overrideBlockHash entries +// over the canonical chain. +func (api *BaseAPI) blockHashGetter(ctx context.Context, tx kv.Tx, overrideBlockHash map[uint64]common.Hash) func(i uint64) (common.Hash, error) { + return func(i uint64) (common.Hash, error) { + if hash, ok := overrideBlockHash[i]; ok { + return hash, nil + } + hash, ok, err := api._blockReader.CanonicalHash(ctx, tx, i) + if err != nil || !ok { + log.Debug("Can't get block hash by number", "number", i, "only-canonical", true, "err", err, "ok", ok) + } + return hash, err + } +} + +// setupEVMTimeout cancels the EVM held by the returned pointer once timeout elapses (or +// ctx is cancelled). Callers recreating the EVM in a loop must Store the new instance so +// the cancellation always hits the current one. The returned cleanup must be deferred. +func setupEVMTimeout(ctx context.Context, evm *vm.EVM, timeout time.Duration) (*atomic.Pointer[vm.EVM], func()) { + evmPtr := &atomic.Pointer[vm.EVM]{} + evmPtr.Store(evm) + + var cancel context.CancelFunc + if timeout > 0 { + ctx, cancel = context.WithTimeout(ctx, timeout) + } else { + ctx, cancel = context.WithCancel(ctx) + } + + stop := context.AfterFunc(ctx, func() { evmPtr.Load().Cancel() }) + return evmPtr, func() { + stop() + cancel() + } +} + func (api *APIImpl) CallMany(ctx context.Context, bundles []Bundle, simulateContext StateContext, stateOverride *ethapi.StateOverrides, timeoutMilliSecondsPtr *int64) ([][]map[string]any, error) { var ( hash common.Hash @@ -69,18 +118,8 @@ func (api *APIImpl) CallMany(ctx context.Context, bundles []Bundle, simulateCont if err != nil { return nil, err } - if len(bundles) == 0 { - return nil, errors.New("empty bundles") - } - empty := true - for _, bundle := range bundles { - if len(bundle.Transactions) != 0 { - empty = false - } - } - - if empty { - return nil, errors.New("empty bundles") + if err := validateBundles(bundles); err != nil { + return nil, err } defer func(start time.Time) { log.Trace("Executing EVM callMany finished", "runtime", time.Since(start)) }(time.Now()) @@ -133,16 +172,7 @@ func (api *APIImpl) CallMany(ctx context.Context, bundles []Bundle, simulateCont return nil, fmt.Errorf("block %d(%x) not found", blockNum, hash) } - getHash := func(i uint64) (common.Hash, error) { - if hash, ok := overrideBlockHash[i]; ok { - return hash, nil - } - hash, ok, err := api._blockReader.CanonicalHash(ctx, tx, i) - if err != nil || !ok { - log.Debug("Can't get block hash by number", "number", i, "only-canonical", true, "err", err, "ok", ok) - } - return hash, err - } + getHash := api.blockHashGetter(ctx, tx, overrideBlockHash) blockCtx = protocol.NewEVMBlockContext(header, getHash, api.engine(), accounts.NilAddress /* author */, chainConfig) @@ -151,31 +181,14 @@ func (api *APIImpl) CallMany(ctx context.Context, bundles []Bundle, simulateCont signer := types.MakeSigner(chainConfig, blockNum, blockCtx.Time) rules := evm.ChainRules() - // evmPtr is updated atomically each time evm is recreated in the loop, - // so the AfterFunc callback always cancels the current instance. - var evmPtr atomic.Pointer[vm.EVM] - evmPtr.Store(evm) - timeout := api.evmCallTimeout if timeoutMilliSecondsPtr != nil && *timeoutMilliSecondsPtr > 0 { timeout = time.Duration(*timeoutMilliSecondsPtr) * time.Millisecond } - // Setup context so it may be cancelled the call has completed - // or, in case of unmetered gas, setup a context with a timeout. - var cancel context.CancelFunc - if timeout > 0 { - ctx, cancel = context.WithTimeout(ctx, timeout) - } else { - ctx, cancel = context.WithCancel(ctx) - } - // Make sure the context is cancelled when the call has completed - // this makes sure resources are cleaned up. - defer cancel() - - stop := context.AfterFunc(ctx, func() { evmPtr.Load().Cancel() }) - defer stop() + evmPtr, cleanup := setupEVMTimeout(ctx, evm, timeout) + defer cleanup() // Setup the gas pool (also for unmetered requests) // and apply the message. diff --git a/rpc/jsonrpc/eth_callMany_test.go b/rpc/jsonrpc/eth_callMany_test.go index 9e1746e9ae6..8b79ce68eed 100644 --- a/rpc/jsonrpc/eth_callMany_test.go +++ b/rpc/jsonrpc/eth_callMany_test.go @@ -17,6 +17,7 @@ package jsonrpc import ( + "bytes" "context" "encoding/hex" "fmt" @@ -25,7 +26,10 @@ import ( "testing" "github.com/holiman/uint256" + jsoniter "github.com/json-iterator/go" + "github.com/stretchr/testify/require" + "github.com/erigontech/erigon/cmd/rpcdaemon/rpcdaemontest" "github.com/erigontech/erigon/common/crypto" "github.com/erigontech/erigon/common/hexutil" "github.com/erigontech/erigon/db/datadir" @@ -37,9 +41,35 @@ import ( "github.com/erigontech/erigon/rpc" "github.com/erigontech/erigon/rpc/ethapi" "github.com/erigontech/erigon/rpc/jsonrpc/contracts" + "github.com/erigontech/erigon/rpc/jsonstream" "github.com/erigontech/erigon/rpc/rpccfg" ) +func TestCallManyEmptyBundles(t *testing.T) { + m, _, _ := rpcdaemontest.CreateTestExecModule(t) + baseApi := newBaseApiForTest(m) + api := newEthApiForTest(baseApi, m.DB, nil, nil) + debugApi := NewPrivateDebugAPI(baseApi, m.DB, nil, 5000000, false) + ctx := context.Background() + + txIndex := -1 + stateCtx := StateContext{BlockNumber: rpc.BlockNumberOrHashWithNumber(rpc.LatestBlockNumber), TransactionIndex: &txIndex} + + for name, bundles := range map[string][]Bundle{ + "noBundles": {}, + "noTransactions": {{}, {}}, + "emptyTransactions": {{Transactions: []ethapi.CallArgs{}}}, + } { + _, err := api.CallMany(ctx, bundles, stateCtx, nil, nil) + require.EqualError(t, err, "empty bundles", name) + + var buf bytes.Buffer + stream := jsonstream.New(jsoniter.NewStream(jsoniter.ConfigDefault, &buf, 4096)) + err = debugApi.TraceCallMany(ctx, bundles, stateCtx, nil, stream) + require.EqualError(t, err, "empty bundles", name) + } +} + // block 1 contains 3 Transactions // 1. deploy token A // 2. mint address 2 100 token diff --git a/rpc/jsonrpc/overlay_api.go b/rpc/jsonrpc/overlay_api.go index f6b173cdd8f..9a8ebe94a5c 100644 --- a/rpc/jsonrpc/overlay_api.go +++ b/rpc/jsonrpc/overlay_api.go @@ -171,16 +171,7 @@ func (api *OverlayAPIImpl) CallConstructor(ctx context.Context, address common.A return nil, fmt.Errorf("block %d(%x) not found", blockNum, block.Hash()) } - getHash := func(i uint64) (common.Hash, error) { - if hash, ok := overrideBlockHash[i]; ok { - return hash, nil - } - hash, ok, err := api._blockReader.CanonicalHash(ctx, tx, i) - if err != nil || !ok { - log.Debug("Can't get block hash by number", "number", i, "only-canonical", true, "err", err, "ok", ok) - } - return hash, err - } + getHash := api.blockHashGetter(ctx, tx, overrideBlockHash) blockCtx = protocol.NewEVMBlockContext(header, getHash, api.engine(), accounts.NilAddress, chainConfig) @@ -462,16 +453,7 @@ func (api *OverlayAPIImpl) replayBlock(ctx context.Context, blockNum uint64, sta return nil, fmt.Errorf("block %d(%x) not found", blockNum, hash) } - getHash := func(i uint64) (common.Hash, error) { - if hash, ok := overrideBlockHash[i]; ok { - return hash, nil - } - hash, ok, err := api._blockReader.CanonicalHash(ctx, tx, i) - if err != nil || !ok { - log.Debug("Can't get block hash by number", "number", i, "only-canonical", true, "err", err, "ok", ok) - } - return hash, err - } + getHash := api.blockHashGetter(ctx, tx, overrideBlockHash) blockCtx = protocol.NewEVMBlockContext(header, getHash, api.engine(), accounts.NilAddress, chainConfig) diff --git a/rpc/jsonrpc/tracing.go b/rpc/jsonrpc/tracing.go index 5f6dd9bba5e..ad4682735fc 100644 --- a/rpc/jsonrpc/tracing.go +++ b/rpc/jsonrpc/tracing.go @@ -484,18 +484,8 @@ func (api *DebugAPIImpl) TraceCallMany(ctx context.Context, bundles []Bundle, si if err != nil { return err } - if len(bundles) == 0 { - return errors.New("empty bundles") - } - empty := true - for _, bundle := range bundles { - if len(bundle.Transactions) != 0 { - empty = false - } - } - - if empty { - return errors.New("empty bundles") + if err := validateBundles(bundles); err != nil { + return err } defer func(start time.Time) { log.Trace("Tracing CallMany finished", "runtime", time.Since(start)) }(time.Now()) @@ -543,17 +533,7 @@ func (api *DebugAPIImpl) TraceCallMany(ctx context.Context, bundles []Bundle, si ibs := state.New(stateReader) - getHash := func(i uint64) (common.Hash, error) { - if hash, ok := overrideBlockHash[i]; ok { - return hash, nil - } - hash, ok, err := api._blockReader.CanonicalHash(ctx, tx, i) - if err != nil || !ok { - log.Debug("Can't get block hash by number", "number", i, "only-canonical", true, "err", err, "ok", ok) - return common.Hash{}, err - } - return hash, nil - } + getHash := api.blockHashGetter(ctx, tx, overrideBlockHash) blockCtx = protocol.NewEVMBlockContext(header, getHash, api.engine(), accounts.NilAddress /* author */, chainConfig) // Apply global block overrides as the baseline for all bundles. From 846665c63ca11d830d6d83869b77db269128b877 Mon Sep 17 00:00:00 2001 From: yperbasis Date: Sat, 4 Jul 2026 11:46:09 +0200 Subject: [PATCH 6/8] rpc/jsonrpc, rpc/transactions: apply review cleanups to helper families - reuse transactions.MakeBlockHashProvider instead of adding blockHashGetter, unifying the canonical-hash-miss log at Debug level - subscribeRPC: hand notify callbacks a pre-bound emit func, folding the Notify+Warn boilerplate into the skeleton - extract borReceiptForBlock, shared by getReceiptsWithBor and GetTransactionReceipt's bor state-sync path - setupEVMTimeout: return the derived context; convert replayBlock's inline timeout copy - logRangeLatestOnly: collapse the FromBlock/ToBlock twins into resolveLogBound - drop a comment restating the code in resolveLogsRange --- rpc/jsonrpc/erigon_receipts.go | 35 +++++++++++++++------------ rpc/jsonrpc/eth_block.go | 2 +- rpc/jsonrpc/eth_callMany.go | 30 +++++++---------------- rpc/jsonrpc/eth_filters.go | 44 ++++++++++++++-------------------- rpc/jsonrpc/eth_receipts.go | 31 ++++++++++++------------ rpc/jsonrpc/overlay_api.go | 23 +++++------------- rpc/jsonrpc/tracing.go | 2 +- rpc/transactions/call.go | 2 +- 8 files changed, 70 insertions(+), 99 deletions(-) diff --git a/rpc/jsonrpc/erigon_receipts.go b/rpc/jsonrpc/erigon_receipts.go index 3e537c18fa6..8fbd5828eae 100644 --- a/rpc/jsonrpc/erigon_receipts.go +++ b/rpc/jsonrpc/erigon_receipts.go @@ -20,6 +20,7 @@ import ( "context" "errors" "fmt" + "math/big" "github.com/RoaringBitmap/roaring/v2" @@ -85,26 +86,30 @@ func logRangeLatestOnly(tx kv.Tx, crit filters.FilterCriteria) (begin, end uint6 if err != nil { return 0, 0, err } - - begin = 0 - if crit.FromBlock != nil { - if crit.FromBlock.Sign() >= 0 { - begin = crit.FromBlock.Uint64() - } else if !crit.FromBlock.IsInt64() || crit.FromBlock.Int64() != int64(rpc.LatestBlockNumber) { - return 0, 0, &rpc.CustomError{Message: fmt.Sprintf("negative value for FromBlock: %v", crit.FromBlock), Code: rpc.ErrCodeInvalidParams} - } + begin, err = resolveLogBound(crit.FromBlock, 0, "FromBlock") + if err != nil { + return 0, 0, err } - end = latest - if crit.ToBlock != nil { - if crit.ToBlock.Sign() >= 0 { - end = crit.ToBlock.Uint64() - } else if !crit.ToBlock.IsInt64() || crit.ToBlock.Int64() != int64(rpc.LatestBlockNumber) { - return 0, 0, &rpc.CustomError{Message: fmt.Sprintf("negative value for ToBlock: %v", crit.ToBlock), Code: rpc.ErrCodeInvalidParams} - } + end, err = resolveLogBound(crit.ToBlock, latest, "ToBlock") + if err != nil { + return 0, 0, err } return begin, end, nil } +func resolveLogBound(bound *big.Int, def uint64, name string) (uint64, error) { + switch { + case bound == nil: + return def, nil + case bound.Sign() >= 0: + return bound.Uint64(), nil + case bound.IsInt64() && bound.Int64() == int64(rpc.LatestBlockNumber): + return def, nil + default: + return 0, &rpc.CustomError{Message: fmt.Sprintf("negative value for %s: %v", name, bound), Code: rpc.ErrCodeInvalidParams} + } +} + // GetLogs implements erigon_getLogs. Returns an array of logs matching a given filter object. func (api *ErigonImpl) GetLogs(ctx context.Context, crit filters.FilterCriteria) (types.ErigonLogs, error) { var begin, end uint64 diff --git a/rpc/jsonrpc/eth_block.go b/rpc/jsonrpc/eth_block.go index ccd69d268f4..d10923aab8c 100644 --- a/rpc/jsonrpc/eth_block.go +++ b/rpc/jsonrpc/eth_block.go @@ -148,7 +148,7 @@ func (api *APIImpl) CallBundle(ctx context.Context, txHashes []common.Hash, stat } timeout := time.Millisecond * time.Duration(timeoutMilliSeconds) - evmPtr, cleanup := setupEVMTimeout(ctx, evm, timeout) + _, evmPtr, cleanup := setupEVMTimeout(ctx, evm, timeout) defer cleanup() // Setup the gas pool (also for unmetered requests) diff --git a/rpc/jsonrpc/eth_callMany.go b/rpc/jsonrpc/eth_callMany.go index 147ad41a51a..ca892a0288e 100644 --- a/rpc/jsonrpc/eth_callMany.go +++ b/rpc/jsonrpc/eth_callMany.go @@ -28,7 +28,6 @@ import ( "github.com/erigontech/erigon/common/hexutil" "github.com/erigontech/erigon/common/log/v3" "github.com/erigontech/erigon/common/math" - "github.com/erigontech/erigon/db/kv" "github.com/erigontech/erigon/execution/protocol" "github.com/erigontech/erigon/execution/state" "github.com/erigontech/erigon/execution/types" @@ -38,6 +37,7 @@ import ( "github.com/erigontech/erigon/rpc" "github.com/erigontech/erigon/rpc/ethapi" "github.com/erigontech/erigon/rpc/rpchelper" + "github.com/erigontech/erigon/rpc/transactions" ) type Bundle struct { @@ -62,25 +62,11 @@ func validateBundles(bundles []Bundle) error { return errors.New("empty bundles") } -// blockHashGetter returns a BLOCKHASH resolver that prefers overrideBlockHash entries -// over the canonical chain. -func (api *BaseAPI) blockHashGetter(ctx context.Context, tx kv.Tx, overrideBlockHash map[uint64]common.Hash) func(i uint64) (common.Hash, error) { - return func(i uint64) (common.Hash, error) { - if hash, ok := overrideBlockHash[i]; ok { - return hash, nil - } - hash, ok, err := api._blockReader.CanonicalHash(ctx, tx, i) - if err != nil || !ok { - log.Debug("Can't get block hash by number", "number", i, "only-canonical", true, "err", err, "ok", ok) - } - return hash, err - } -} - // setupEVMTimeout cancels the EVM held by the returned pointer once timeout elapses (or -// ctx is cancelled). Callers recreating the EVM in a loop must Store the new instance so -// the cancellation always hits the current one. The returned cleanup must be deferred. -func setupEVMTimeout(ctx context.Context, evm *vm.EVM, timeout time.Duration) (*atomic.Pointer[vm.EVM], func()) { +// ctx is cancelled), and returns the deadline-carrying context. Callers recreating the +// EVM in a loop must Store the new instance so the cancellation always hits the current +// one. The returned cleanup must be deferred. +func setupEVMTimeout(ctx context.Context, evm *vm.EVM, timeout time.Duration) (context.Context, *atomic.Pointer[vm.EVM], func()) { evmPtr := &atomic.Pointer[vm.EVM]{} evmPtr.Store(evm) @@ -92,7 +78,7 @@ func setupEVMTimeout(ctx context.Context, evm *vm.EVM, timeout time.Duration) (* } stop := context.AfterFunc(ctx, func() { evmPtr.Load().Cancel() }) - return evmPtr, func() { + return ctx, evmPtr, func() { stop() cancel() } @@ -172,7 +158,7 @@ func (api *APIImpl) CallMany(ctx context.Context, bundles []Bundle, simulateCont return nil, fmt.Errorf("block %d(%x) not found", blockNum, hash) } - getHash := api.blockHashGetter(ctx, tx, overrideBlockHash) + getHash := transactions.MakeBlockHashProvider(ctx, tx, api._blockReader, overrideBlockHash) blockCtx = protocol.NewEVMBlockContext(header, getHash, api.engine(), accounts.NilAddress /* author */, chainConfig) @@ -187,7 +173,7 @@ func (api *APIImpl) CallMany(ctx context.Context, bundles []Bundle, simulateCont timeout = time.Duration(*timeoutMilliSecondsPtr) * time.Millisecond } - evmPtr, cleanup := setupEVMTimeout(ctx, evm, timeout) + _, evmPtr, cleanup := setupEVMTimeout(ctx, evm, timeout) defer cleanup() // Setup the gas pool (also for unmetered requests) diff --git a/rpc/jsonrpc/eth_filters.go b/rpc/jsonrpc/eth_filters.go index e5d916bf3f3..8afdc45ba84 100644 --- a/rpc/jsonrpc/eth_filters.go +++ b/rpc/jsonrpc/eth_filters.go @@ -154,8 +154,9 @@ func (api *APIImpl) GetFilterLogs(_ context.Context, index string) ([]*types.Log // subscribeRPC runs the shared subscription skeleton: guard checks, subscription // creation, and a goroutine that pumps items from the filter channel into notify until // the channel closes or the client goes away. subscribe is called inside the goroutine -// and must return the item channel plus an unsubscribe func. -func subscribeRPC[T any](ctx context.Context, apiFilters *rpchelper.Filters, subscribe func() (<-chan T, func()), notify func(notifier rpc.Notifier, rpcSub *rpc.Subscription, item T), closedWarn string) (*rpc.Subscription, error) { +// and must return the item channel plus an unsubscribe func. notify receives an emit +// func that sends a payload to the client, logging on failure. +func subscribeRPC[T any](ctx context.Context, apiFilters *rpchelper.Filters, subscribe func() (<-chan T, func()), notify func(emit func(payload any), item T), closedWarn string) (*rpc.Subscription, error) { if apiFilters == nil { return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported } @@ -171,10 +172,15 @@ func subscribeRPC[T any](ctx context.Context, apiFilters *rpchelper.Filters, sub ch, unsubscribe := subscribe() defer unsubscribe() + emit := func(payload any) { + if err := notifier.Notify(rpcSub.ID, payload); err != nil { + log.Warn("[rpc] error while notifying subscription", "err", err) + } + } for { select { case item, ok := <-ch: - notify(notifier, rpcSub, item) + notify(emit, item) if !ok { log.Warn(closedWarn) return @@ -195,12 +201,9 @@ func (api *APIImpl) NewHeads(ctx context.Context) (*rpc.Subscription, error) { headers, id := api.filters.SubscribeNewHeads(32) return headers, func() { api.filters.UnsubscribeHeads(id) } }, - func(notifier rpc.Notifier, rpcSub *rpc.Subscription, h *types.Header) { + func(emit func(payload any), h *types.Header) { if h != nil { - err := notifier.Notify(rpcSub.ID, h) - if err != nil { - log.Warn("[rpc] error while notifying subscription", "err", err) - } + emit(h) } }, "[rpc] new heads channel was closed") @@ -222,18 +225,13 @@ func (api *APIImpl) subscribePendingTransactions(ctx context.Context, chanSize i txsCh, id := api.filters.SubscribePendingTxs(chanSize) return txsCh, func() { api.filters.UnsubscribePendingTxs(id) } }, - func(notifier rpc.Notifier, rpcSub *rpc.Subscription, txs []types.Transaction) { + func(emit func(payload any), txs []types.Transaction) { for _, t := range txs { if t != nil { - var err error if fullTx { - err = notifier.Notify(rpcSub.ID, newRPCPendingTransaction(t, nil, nil)) + emit(newRPCPendingTransaction(t, nil, nil)) } else { - err = notifier.Notify(rpcSub.ID, t.Hash()) - } - - if err != nil { - log.Warn("[rpc] error while notifying subscription", "err", err) + emit(t.Hash()) } } } @@ -248,12 +246,9 @@ func (api *APIImpl) Logs(ctx context.Context, crit filters.FilterCriteria) (*rpc logs, id := api.filters.SubscribeLogs(api.SubscribeLogsChannelSize, crit) return logs, func() { api.filters.UnsubscribeLogs(id) } }, - func(notifier rpc.Notifier, rpcSub *rpc.Subscription, h *types.Log) { + func(emit func(payload any), h *types.Log) { if h != nil { - err := notifier.Notify(rpcSub.ID, h) - if err != nil { - log.Warn("[rpc] error while notifying subscription", "err", err) - } + emit(h) } }, "[rpc] log channel was closed") @@ -266,13 +261,10 @@ func (api *APIImpl) TransactionReceipts(ctx context.Context, crit filters.Receip receipts, id := api.filters.SubscribeReceipts(api.SubscribeLogsChannelSize, crit) return receipts, func() { api.filters.UnsubscribeReceipts(id) } }, - func(notifier rpc.Notifier, rpcSub *rpc.Subscription, protoReceipt *remoteproto.SubscribeReceiptsReply) { + func(emit func(payload any), protoReceipt *remoteproto.SubscribeReceiptsReply) { if protoReceipt != nil { receipt := ethutils.MarshalSubscribeReceipt(protoReceipt) - err := notifier.Notify(rpcSub.ID, []map[string]any{receipt}) - if err != nil { - log.Warn("[rpc] error while notifying subscription", "err", err) - } + emit([]map[string]any{receipt}) } }, "[rpc] receipts channel was closed") diff --git a/rpc/jsonrpc/eth_receipts.go b/rpc/jsonrpc/eth_receipts.go index af5cd0ffb80..7c98c22b450 100644 --- a/rpc/jsonrpc/eth_receipts.go +++ b/rpc/jsonrpc/eth_receipts.go @@ -88,18 +88,24 @@ func (api *BaseAPI) getReceiptsWithBor(ctx context.Context, tx kv.TemporalTx, ch if chainConfig.Bor == nil { return receipts, nil, nil } - events, err := api.bridgeReader.Events(ctx, block.Hash(), block.NumberU64()) + borReceipt, err := api.borReceiptForBlock(ctx, tx, chainConfig, block) if err != nil { return nil, nil, err } - if len(events) == 0 { - return receipts, nil, nil - } - borReceipt, err := api.borReceiptGenerator.GenerateBorReceipt(ctx, tx, block, events, chainConfig) + return receipts, borReceipt, nil +} + +// borReceiptForBlock returns the synthetic bor receipt for the block's state sync +// events, or nil when the block has none. +func (api *BaseAPI) borReceiptForBlock(ctx context.Context, tx kv.TemporalTx, chainConfig *chain.Config, block *types.Block) (*types.Receipt, error) { + events, err := api.bridgeReader.Events(ctx, block.Hash(), block.NumberU64()) if err != nil { - return nil, nil, err + return nil, err } - return receipts, borReceipt, nil + if len(events) == 0 { + return nil, nil + } + return api.borReceiptGenerator.GenerateBorReceipt(ctx, tx, block, events, chainConfig) } func (api *BaseAPI) getReceipt(ctx context.Context, cc *chain.Config, tx kv.TemporalTx, header *types.Header, txn types.Transaction, index int, txNum uint64, postState *receipts.PostStateInfo) (*types.Receipt, error) { @@ -132,7 +138,6 @@ func (api *BaseAPI) resolveLogsRange(ctx context.Context, tx kv.Tx, crit filters return num, num, nil } - // Convert the RPC block numbers into internal representations latest, _, _, err := rpchelper.GetBlockNumber(ctx, rpc.BlockNumberOrHashWithNumber(rpc.LatestExecutedBlockNumber), tx, api._blockReader, nil) if err != nil { return 0, 0, err @@ -578,20 +583,14 @@ func (api *APIImpl) GetTransactionReceipt(ctx context.Context, txnHash common.Ha return nil, nil // not error, see https://github.com/erigontech/erigon/issues/1645 } - events, err := api.bridgeReader.Events(ctx, block.Hash(), blockNum) + borReceipt, err := api.borReceiptForBlock(ctx, tx, chainConfig, block) if err != nil { return nil, err } - - if len(events) == 0 { + if borReceipt == nil { return nil, errors.New("tx not found") } - borReceipt, err := api.borReceiptGenerator.GenerateBorReceipt(ctx, tx, block, events, chainConfig) - if err != nil { - return nil, err - } - return ethutils.MarshalReceipt(borReceipt, bortypes.NewBorTransaction(), chainConfig, block.HeaderNoCopy(), txnHash, false, false), nil } diff --git a/rpc/jsonrpc/overlay_api.go b/rpc/jsonrpc/overlay_api.go index 9a8ebe94a5c..ca93fe9f629 100644 --- a/rpc/jsonrpc/overlay_api.go +++ b/rpc/jsonrpc/overlay_api.go @@ -44,6 +44,7 @@ import ( "github.com/erigontech/erigon/rpc/ethapi" "github.com/erigontech/erigon/rpc/filters" "github.com/erigontech/erigon/rpc/rpchelper" + "github.com/erigontech/erigon/rpc/transactions" ) type OverlayAPI interface { @@ -171,7 +172,7 @@ func (api *OverlayAPIImpl) CallConstructor(ctx context.Context, address common.A return nil, fmt.Errorf("block %d(%x) not found", blockNum, block.Hash()) } - getHash := api.blockHashGetter(ctx, tx, overrideBlockHash) + getHash := transactions.MakeBlockHashProvider(ctx, tx, api._blockReader, overrideBlockHash) blockCtx = protocol.NewEVMBlockContext(header, getHash, api.engine(), accounts.NilAddress, chainConfig) @@ -453,7 +454,7 @@ func (api *OverlayAPIImpl) replayBlock(ctx context.Context, blockNum uint64, sta return nil, fmt.Errorf("block %d(%x) not found", blockNum, hash) } - getHash := api.blockHashGetter(ctx, tx, overrideBlockHash) + getHash := transactions.MakeBlockHashProvider(ctx, tx, api._blockReader, overrideBlockHash) blockCtx = protocol.NewEVMBlockContext(header, getHash, api.engine(), accounts.NilAddress, chainConfig) @@ -461,26 +462,14 @@ func (api *OverlayAPIImpl) replayBlock(ctx context.Context, blockNum uint64, sta rules := blockCtx.Rules(chainConfig) timeout := api.OverlayReplayBlockTimeout - // Setup context so it may be cancelled the call has completed - // or, in case of unmetered gas, setup a context with a timeout. - var cancel context.CancelFunc - if timeout > 0 { - ctx, cancel = context.WithTimeout(ctx, timeout) - } else { - ctx, cancel = context.WithCancel(ctx) - } - // Make sure the context is cancelled when the call has completed - // this makes sure resources are cleaned up. - defer cancel() // Setup the gas pool (also for unmetered requests) // and apply the message. gp := new(protocol.GasPool).AddGas(math.MaxUint64).AddBlobGas(math.MaxUint64) - vmConfig := vm.Config{} - evm = vm.NewEVM(blockCtx, evmtypes.TxContext{}, statedb, chainConfig, vmConfig) + evm = vm.NewEVM(blockCtx, evmtypes.TxContext{}, statedb, chainConfig, vm.Config{}) - stop := context.AfterFunc(ctx, evm.Cancel) - defer stop() + ctx, _, cleanup := setupEVMTimeout(ctx, evm, timeout) + defer cleanup() receipts, err := api.getReceipts(ctx, tx, block) if err != nil { return nil, err diff --git a/rpc/jsonrpc/tracing.go b/rpc/jsonrpc/tracing.go index ad4682735fc..52a9c42a93f 100644 --- a/rpc/jsonrpc/tracing.go +++ b/rpc/jsonrpc/tracing.go @@ -533,7 +533,7 @@ func (api *DebugAPIImpl) TraceCallMany(ctx context.Context, bundles []Bundle, si ibs := state.New(stateReader) - getHash := api.blockHashGetter(ctx, tx, overrideBlockHash) + getHash := transactions.MakeBlockHashProvider(ctx, tx, api._blockReader, overrideBlockHash) blockCtx = protocol.NewEVMBlockContext(header, getHash, api.engine(), accounts.NilAddress /* author */, chainConfig) // Apply global block overrides as the baseline for all bundles. diff --git a/rpc/transactions/call.go b/rpc/transactions/call.go index ce243c97201..33f22f4ff52 100644 --- a/rpc/transactions/call.go +++ b/rpc/transactions/call.go @@ -156,7 +156,7 @@ func MakeBlockHashProvider(ctx context.Context, tx kv.Getter, reader services.Ca } blockHash, ok, err := reader.CanonicalHash(ctx, tx, blockNum) if err != nil || !ok { - log.Error("[evm] canonical hash not found", "blockNum", blockNum, "ok", ok, "err", err) + log.Debug("[evm] canonical hash not found", "blockNum", blockNum, "ok", ok, "err", err) } return blockHash, err } From ce492bbfc0cb706c4df9cea23be53a1cba147f58 Mon Sep 17 00:00:00 2001 From: yperbasis Date: Sat, 4 Jul 2026 11:57:36 +0200 Subject: [PATCH 7/8] rpc/jsonrpc: address Copilot review on helper extractions - subscribeRPC: check channel closure before invoking notify, so the skeleton no longer relies on callbacks guarding the zero value - setupEVMTimeout: derive the context before the EVM exists (callers Store the instance after construction) and build the BLOCKHASH provider from the derived context, restoring the pre-extraction behavior where CallMany/replayBlock hash lookups ran under the EVM-timeout deadline (the old inline closures late-bound the reassigned ctx variable) --- rpc/jsonrpc/eth_block.go | 3 ++- rpc/jsonrpc/eth_callMany.go | 34 +++++++++++++++++++--------------- rpc/jsonrpc/eth_filters.go | 2 +- rpc/jsonrpc/overlay_api.go | 10 +++++----- 4 files changed, 27 insertions(+), 22 deletions(-) diff --git a/rpc/jsonrpc/eth_block.go b/rpc/jsonrpc/eth_block.go index d10923aab8c..695080f95a7 100644 --- a/rpc/jsonrpc/eth_block.go +++ b/rpc/jsonrpc/eth_block.go @@ -148,8 +148,9 @@ func (api *APIImpl) CallBundle(ctx context.Context, txHashes []common.Hash, stat } timeout := time.Millisecond * time.Duration(timeoutMilliSeconds) - _, evmPtr, cleanup := setupEVMTimeout(ctx, evm, timeout) + _, evmPtr, cleanup := setupEVMTimeout(ctx, timeout) defer cleanup() + evmPtr.Store(evm) // Setup the gas pool (also for unmetered requests) // and apply the message. diff --git a/rpc/jsonrpc/eth_callMany.go b/rpc/jsonrpc/eth_callMany.go index ca892a0288e..3d567e6d1ed 100644 --- a/rpc/jsonrpc/eth_callMany.go +++ b/rpc/jsonrpc/eth_callMany.go @@ -63,12 +63,11 @@ func validateBundles(bundles []Bundle) error { } // setupEVMTimeout cancels the EVM held by the returned pointer once timeout elapses (or -// ctx is cancelled), and returns the deadline-carrying context. Callers recreating the -// EVM in a loop must Store the new instance so the cancellation always hits the current -// one. The returned cleanup must be deferred. -func setupEVMTimeout(ctx context.Context, evm *vm.EVM, timeout time.Duration) (context.Context, *atomic.Pointer[vm.EVM], func()) { +// ctx is cancelled), and returns the deadline-carrying context. Callers must Store the +// EVM once created — and again whenever it is recreated — so the cancellation always +// hits the current instance. The returned cleanup must be deferred. +func setupEVMTimeout(ctx context.Context, timeout time.Duration) (context.Context, *atomic.Pointer[vm.EVM], func()) { evmPtr := &atomic.Pointer[vm.EVM]{} - evmPtr.Store(evm) var cancel context.CancelFunc if timeout > 0 { @@ -77,7 +76,11 @@ func setupEVMTimeout(ctx context.Context, evm *vm.EVM, timeout time.Duration) (c ctx, cancel = context.WithCancel(ctx) } - stop := context.AfterFunc(ctx, func() { evmPtr.Load().Cancel() }) + stop := context.AfterFunc(ctx, func() { + if evm := evmPtr.Load(); evm != nil { + evm.Cancel() + } + }) return ctx, evmPtr, func() { stop() cancel() @@ -158,24 +161,25 @@ func (api *APIImpl) CallMany(ctx context.Context, bundles []Bundle, simulateCont return nil, fmt.Errorf("block %d(%x) not found", blockNum, hash) } + timeout := api.evmCallTimeout + + if timeoutMilliSecondsPtr != nil && *timeoutMilliSecondsPtr > 0 { + timeout = time.Duration(*timeoutMilliSecondsPtr) * time.Millisecond + } + + ctx, evmPtr, cleanup := setupEVMTimeout(ctx, timeout) + defer cleanup() + getHash := transactions.MakeBlockHashProvider(ctx, tx, api._blockReader, overrideBlockHash) blockCtx = protocol.NewEVMBlockContext(header, getHash, api.engine(), accounts.NilAddress /* author */, chainConfig) // Get a new instance of the EVM evm = vm.NewEVM(blockCtx, txCtx, st, chainConfig, vm.Config{}) + evmPtr.Store(evm) signer := types.MakeSigner(chainConfig, blockNum, blockCtx.Time) rules := evm.ChainRules() - timeout := api.evmCallTimeout - - if timeoutMilliSecondsPtr != nil && *timeoutMilliSecondsPtr > 0 { - timeout = time.Duration(*timeoutMilliSecondsPtr) * time.Millisecond - } - - _, evmPtr, cleanup := setupEVMTimeout(ctx, evm, timeout) - defer cleanup() - // Setup the gas pool (also for unmetered requests) // and apply the message. gp := new(protocol.GasPool).AddGas(math.MaxUint64).AddBlobGas(math.MaxUint64) diff --git a/rpc/jsonrpc/eth_filters.go b/rpc/jsonrpc/eth_filters.go index 8afdc45ba84..3501d4edc60 100644 --- a/rpc/jsonrpc/eth_filters.go +++ b/rpc/jsonrpc/eth_filters.go @@ -180,11 +180,11 @@ func subscribeRPC[T any](ctx context.Context, apiFilters *rpchelper.Filters, sub for { select { case item, ok := <-ch: - notify(emit, item) if !ok { log.Warn(closedWarn) return } + notify(emit, item) case <-rpcSub.Err(): return } diff --git a/rpc/jsonrpc/overlay_api.go b/rpc/jsonrpc/overlay_api.go index ca93fe9f629..f0c88d939f5 100644 --- a/rpc/jsonrpc/overlay_api.go +++ b/rpc/jsonrpc/overlay_api.go @@ -454,6 +454,10 @@ func (api *OverlayAPIImpl) replayBlock(ctx context.Context, blockNum uint64, sta return nil, fmt.Errorf("block %d(%x) not found", blockNum, hash) } + timeout := api.OverlayReplayBlockTimeout + ctx, evmPtr, cleanup := setupEVMTimeout(ctx, timeout) + defer cleanup() + getHash := transactions.MakeBlockHashProvider(ctx, tx, api._blockReader, overrideBlockHash) blockCtx = protocol.NewEVMBlockContext(header, getHash, api.engine(), accounts.NilAddress, chainConfig) @@ -461,15 +465,11 @@ func (api *OverlayAPIImpl) replayBlock(ctx context.Context, blockNum uint64, sta signer := types.MakeSigner(chainConfig, blockNum, blockCtx.Time) rules := blockCtx.Rules(chainConfig) - timeout := api.OverlayReplayBlockTimeout - // Setup the gas pool (also for unmetered requests) // and apply the message. gp := new(protocol.GasPool).AddGas(math.MaxUint64).AddBlobGas(math.MaxUint64) evm = vm.NewEVM(blockCtx, evmtypes.TxContext{}, statedb, chainConfig, vm.Config{}) - - ctx, _, cleanup := setupEVMTimeout(ctx, evm, timeout) - defer cleanup() + evmPtr.Store(evm) receipts, err := api.getReceipts(ctx, tx, block) if err != nil { return nil, err From 0ff24142eaf8540f2acfd3ef2637538a7d1a1cd8 Mon Sep 17 00:00:00 2001 From: yperbasis Date: Sat, 4 Jul 2026 14:17:52 +0200 Subject: [PATCH 8/8] rpc/jsonrpc: close EVM cancellation window and apply further review fixes setupEVMTimeout armed its one-shot AfterFunc before any EVM was stored, so a cancellation firing in the arm-to-Store window was silently dropped (the nil-guarded Load skips, AfterFunc never re-fires) and the call then ran with no timeout at all: ApplyMessage takes no ctx and the loops abort only via evm.Cancelled(). The helper now returns a store func that re-checks ctx.Err() after each store and cancels the just-stored EVM if the deadline has already fired. The re-check runs on every store, so the EVM-recreation loops in CallBundle/CallMany are also covered between iterations (a hole that predated the extraction). Pinned by TestSetupEVMTimeoutCancelsEVMStoredAfterExpiry, verified red against the plain-Store semantics. Further review fixes: - txnIndexInBlock returns the established -1 bor state-sync sentinel instead of a documented-meaningless wraparound value (RemoteBlockReader already guards i < 0, the local snapshot path guards == -1, and trace_filtering decodes bor-ness from txIndex == -1); the two trace-site -1 overwrites become redundant and are removed. The Min call stays first so error propagation is unchanged. - GetTransactionByHash adopts txnLookupWithBorFallback - the one remaining site where the collapse is behavior-preserving: its prune check runs post-fallback inside "if ok", and both TxnLookup implementations return blockNum 0 exactly on miss, so the blockNum==0 trigger is equivalent to the helper's !ok trigger. - simulateCall, trace_call and trace_rawTransaction adopt setupEVMTimeout. They previously bound evm.Cancel after the EVM existed, so converting them without the store re-check would have imported the cancellation window; with it, the conversion is a strict dedup. - validateBundles drops its redundant len==0 branch: an empty slice skips the loop and falls through to the identical error, pinned by TestCallManyEmptyBundles. --- rpc/jsonrpc/eth_api.go | 11 ++++++---- rpc/jsonrpc/eth_block.go | 6 +++--- rpc/jsonrpc/eth_callMany.go | 30 +++++++++++++++------------ rpc/jsonrpc/eth_callMany_test.go | 19 +++++++++++++++++ rpc/jsonrpc/eth_simulation.go | 14 +++---------- rpc/jsonrpc/eth_txs.go | 11 +--------- rpc/jsonrpc/overlay_api.go | 4 ++-- rpc/jsonrpc/trace_adhoc.go | 35 ++++++-------------------------- rpc/jsonrpc/trace_filtering.go | 4 ---- 9 files changed, 58 insertions(+), 76 deletions(-) diff --git a/rpc/jsonrpc/eth_api.go b/rpc/jsonrpc/eth_api.go index 093dfae03ef..22cc2572224 100644 --- a/rpc/jsonrpc/eth_api.go +++ b/rpc/jsonrpc/eth_api.go @@ -260,15 +260,18 @@ func (api *BaseAPI) txnLookupWithBorFallback(ctx context.Context, tx kv.Tx, txnH return blockNum, txNum, true, true, nil } -// txnIndexInBlock derives the in-block txn index from a global txNum. For bor state sync -// txns txNum comes from a missed lookup, so the consistency check is skipped and the -// returned index is meaningless. +// txnIndexInBlock derives the in-block txn index from a global txNum. Bor state sync +// txns are not part of the block body, so they yield the -1 sentinel and the consistency +// check is skipped (their txNum comes from a missed lookup). func (api *BaseAPI) txnIndexInBlock(ctx context.Context, tx kv.Tx, blockNum, txNum uint64, isBorStateSyncTxn bool) (int, error) { txNumMin, err := api._txNumReader.Min(ctx, tx, blockNum) if err != nil { return 0, err } - if txNumMin+1 > txNum && !isBorStateSyncTxn { + if isBorStateSyncTxn { + return -1, nil + } + if txNumMin+1 > txNum { return 0, fmt.Errorf("uint underflow txnums error txNum: %d, txNumMin: %d, blockNum: %d", txNum, txNumMin, blockNum) } return int(txNum - txNumMin - 1), nil diff --git a/rpc/jsonrpc/eth_block.go b/rpc/jsonrpc/eth_block.go index 695080f95a7..76a0255ce0d 100644 --- a/rpc/jsonrpc/eth_block.go +++ b/rpc/jsonrpc/eth_block.go @@ -148,9 +148,9 @@ func (api *APIImpl) CallBundle(ctx context.Context, txHashes []common.Hash, stat } timeout := time.Millisecond * time.Duration(timeoutMilliSeconds) - _, evmPtr, cleanup := setupEVMTimeout(ctx, timeout) + _, storeEVM, cleanup := setupEVMTimeout(ctx, timeout) defer cleanup() - evmPtr.Store(evm) + storeEVM(evm) // Setup the gas pool (also for unmetered requests) // and apply the message. @@ -169,7 +169,7 @@ func (api *APIImpl) CallBundle(ctx context.Context, txHashes []common.Hash, stat msg.SetCheckGas(false) // Recreate EVM with the correct txCtx for this transaction evm = vm.NewEVM(blockCtx, protocol.NewEVMTxContext(msg), ibs, chainConfig, vm.Config{}) - evmPtr.Store(evm) + storeEVM(evm) // Execute the transaction message result, err := protocol.ApplyMessage(evm, msg, gp, true /* refunds */, false /* gasBailout */, engine) if err != nil { diff --git a/rpc/jsonrpc/eth_callMany.go b/rpc/jsonrpc/eth_callMany.go index 3d567e6d1ed..df9e4836bf3 100644 --- a/rpc/jsonrpc/eth_callMany.go +++ b/rpc/jsonrpc/eth_callMany.go @@ -51,9 +51,6 @@ type StateContext struct { } func validateBundles(bundles []Bundle) error { - if len(bundles) == 0 { - return errors.New("empty bundles") - } for _, bundle := range bundles { if len(bundle.Transactions) != 0 { return nil @@ -62,11 +59,12 @@ func validateBundles(bundles []Bundle) error { return errors.New("empty bundles") } -// setupEVMTimeout cancels the EVM held by the returned pointer once timeout elapses (or -// ctx is cancelled), and returns the deadline-carrying context. Callers must Store the -// EVM once created — and again whenever it is recreated — so the cancellation always -// hits the current instance. The returned cleanup must be deferred. -func setupEVMTimeout(ctx context.Context, timeout time.Duration) (context.Context, *atomic.Pointer[vm.EVM], func()) { +// setupEVMTimeout cancels the EVM registered via the returned store func once timeout +// elapses (or ctx is cancelled), and returns the deadline-carrying context. Callers must +// store the EVM once created — and again whenever it is recreated — so cancellation +// always hits the current instance, even when it fires before the EVM exists. The +// returned cleanup must be deferred. +func setupEVMTimeout(ctx context.Context, timeout time.Duration) (context.Context, func(*vm.EVM), func()) { evmPtr := &atomic.Pointer[vm.EVM]{} var cancel context.CancelFunc @@ -81,7 +79,13 @@ func setupEVMTimeout(ctx context.Context, timeout time.Duration) (context.Contex evm.Cancel() } }) - return ctx, evmPtr, func() { + store := func(evm *vm.EVM) { + evmPtr.Store(evm) + if ctx.Err() != nil { + evm.Cancel() + } + } + return ctx, store, func() { stop() cancel() } @@ -167,7 +171,7 @@ func (api *APIImpl) CallMany(ctx context.Context, bundles []Bundle, simulateCont timeout = time.Duration(*timeoutMilliSecondsPtr) * time.Millisecond } - ctx, evmPtr, cleanup := setupEVMTimeout(ctx, timeout) + ctx, storeEVM, cleanup := setupEVMTimeout(ctx, timeout) defer cleanup() getHash := transactions.MakeBlockHashProvider(ctx, tx, api._blockReader, overrideBlockHash) @@ -176,7 +180,7 @@ func (api *APIImpl) CallMany(ctx context.Context, bundles []Bundle, simulateCont // Get a new instance of the EVM evm = vm.NewEVM(blockCtx, txCtx, st, chainConfig, vm.Config{}) - evmPtr.Store(evm) + storeEVM(evm) signer := types.MakeSigner(chainConfig, blockNum, blockCtx.Time) rules := evm.ChainRules() @@ -191,7 +195,7 @@ func (api *APIImpl) CallMany(ctx context.Context, bundles []Bundle, simulateCont } txCtx = protocol.NewEVMTxContext(msg) evm = vm.NewEVM(blockCtx, txCtx, evm.IntraBlockState(), chainConfig, vm.Config{}) - evmPtr.Store(evm) + storeEVM(evm) // Execute the transaction message _, err = protocol.ApplyMessage(evm, msg, gp, true /* refunds */, false /* gasBailout */, api.engine()) if err != nil { @@ -230,7 +234,7 @@ func (api *APIImpl) CallMany(ctx context.Context, bundles []Bundle, simulateCont } txCtx = protocol.NewEVMTxContext(msg) evm = vm.NewEVM(blockCtx, txCtx, evm.IntraBlockState(), chainConfig, vm.Config{}) - evmPtr.Store(evm) + storeEVM(evm) result, err := protocol.ApplyMessage(evm, msg, gp, true /* refunds */, false /* gasBailout */, api.engine()) if err != nil { return nil, err diff --git a/rpc/jsonrpc/eth_callMany_test.go b/rpc/jsonrpc/eth_callMany_test.go index 8b79ce68eed..710a449a967 100644 --- a/rpc/jsonrpc/eth_callMany_test.go +++ b/rpc/jsonrpc/eth_callMany_test.go @@ -24,6 +24,7 @@ import ( "math/big" "strconv" "testing" + "time" "github.com/holiman/uint256" jsoniter "github.com/json-iterator/go" @@ -38,6 +39,8 @@ import ( "github.com/erigontech/erigon/execution/abi/bind/backends" "github.com/erigontech/erigon/execution/chain" "github.com/erigontech/erigon/execution/types" + "github.com/erigontech/erigon/execution/vm" + "github.com/erigontech/erigon/execution/vm/evmtypes" "github.com/erigontech/erigon/rpc" "github.com/erigontech/erigon/rpc/ethapi" "github.com/erigontech/erigon/rpc/jsonrpc/contracts" @@ -45,6 +48,22 @@ import ( "github.com/erigontech/erigon/rpc/rpccfg" ) +// Pins that an EVM stored after the deadline has already fired is still cancelled. +func TestSetupEVMTimeoutCancelsEVMStoredAfterExpiry(t *testing.T) { + parent, cancelParent := context.WithCancel(context.Background()) + cancelParent() + + _, storeEVM, cleanup := setupEVMTimeout(parent, time.Hour) + defer cleanup() + + // let the one-shot AfterFunc fire before any EVM is registered + time.Sleep(50 * time.Millisecond) + + evm := vm.NewEVM(evmtypes.BlockContext{}, evmtypes.TxContext{}, nil, chain.AllProtocolChanges, vm.Config{}) + storeEVM(evm) + require.True(t, evm.Cancelled()) +} + func TestCallManyEmptyBundles(t *testing.T) { m, _, _ := rpcdaemontest.CreateTestExecModule(t) baseApi := newBaseApiForTest(m) diff --git a/rpc/jsonrpc/eth_simulation.go b/rpc/jsonrpc/eth_simulation.go index 780811060f3..ee5c1d59a37 100644 --- a/rpc/jsonrpc/eth_simulation.go +++ b/rpc/jsonrpc/eth_simulation.go @@ -774,14 +774,8 @@ func (s *simulator) simulateCall( vmConfig vm.Config, precompiles vm.PrecompiledContracts, ) (*CallResult, types.Transaction, *types.Receipt, error) { - // Setup context, so it may be cancelled after the call has completed or in case of unmetered gas use a timeout. - var cancel context.CancelFunc - if s.evmCallTimeout > 0 { - ctx, cancel = context.WithTimeout(ctx, s.evmCallTimeout) - } else { - ctx, cancel = context.WithCancel(ctx) - } - defer cancel() + _, storeEVM, cleanup := setupEVMTimeout(ctx, s.evmCallTimeout) + defer cleanup() err := s.sanitizeCall(call, intraBlockState, &blockCtx, header.BaseFee, *cumulativeGasUsed, s.gasPool.Gas()) if err != nil { @@ -808,9 +802,7 @@ func (s *simulator) simulateCall( // It is possible to override precompiles with EVM bytecode or move them to another address. evm.SetPrecompiles(precompiles) - - stop := context.AfterFunc(ctx, evm.Cancel) - defer stop() + storeEVM(evm) s.gasPool.AddBlobGas(msg.BlobGas()) result, err := protocol.ApplyMessage(evm, msg, s.gasPool, true, false, s.engine) diff --git a/rpc/jsonrpc/eth_txs.go b/rpc/jsonrpc/eth_txs.go index 41d81e8637d..495282e3605 100644 --- a/rpc/jsonrpc/eth_txs.go +++ b/rpc/jsonrpc/eth_txs.go @@ -49,19 +49,10 @@ func (api *APIImpl) GetTransactionByHash(ctx context.Context, txnHash common.Has } // https://www.quicknode.com/docs/ethereum/eth_getTransactionByHash - blockNum, txNum, ok, err := api.txnLookup(ctx, tx, txnHash) + blockNum, txNum, isBorStateSyncTx, ok, err := api.txnLookupWithBorFallback(ctx, tx, txnHash, chainConfig) if err != nil { return nil, err } - - // Private API returns 0 if transaction is not found. - isBorStateSyncTx := blockNum == 0 && chainConfig.Bor != nil - if isBorStateSyncTx { - blockNum, ok, err = api.bridgeReader.EventTxnLookup(ctx, txnHash) - if err != nil { - return nil, err - } - } if ok { err = api.BaseAPI.checkPruneBlocks(ctx, tx, blockNum) if err != nil { diff --git a/rpc/jsonrpc/overlay_api.go b/rpc/jsonrpc/overlay_api.go index f0c88d939f5..8b5e1fe4f29 100644 --- a/rpc/jsonrpc/overlay_api.go +++ b/rpc/jsonrpc/overlay_api.go @@ -455,7 +455,7 @@ func (api *OverlayAPIImpl) replayBlock(ctx context.Context, blockNum uint64, sta } timeout := api.OverlayReplayBlockTimeout - ctx, evmPtr, cleanup := setupEVMTimeout(ctx, timeout) + ctx, storeEVM, cleanup := setupEVMTimeout(ctx, timeout) defer cleanup() getHash := transactions.MakeBlockHashProvider(ctx, tx, api._blockReader, overrideBlockHash) @@ -469,7 +469,7 @@ func (api *OverlayAPIImpl) replayBlock(ctx context.Context, blockNum uint64, sta // and apply the message. gp := new(protocol.GasPool).AddGas(math.MaxUint64).AddBlobGas(math.MaxUint64) evm = vm.NewEVM(blockCtx, evmtypes.TxContext{}, statedb, chainConfig, vm.Config{}) - evmPtr.Store(evm) + storeEVM(evm) receipts, err := api.getReceipts(ctx, tx, block) if err != nil { return nil, err diff --git a/rpc/jsonrpc/trace_adhoc.go b/rpc/jsonrpc/trace_adhoc.go index c14bce80fb4..543c9cfd1e3 100644 --- a/rpc/jsonrpc/trace_adhoc.go +++ b/rpc/jsonrpc/trace_adhoc.go @@ -918,10 +918,6 @@ func (api *TraceAPIImpl) ReplayTransaction(ctx context.Context, txHash common.Ha return nil, err } - if isBorStateSyncTxn { - txnIndex = -1 - } - signer := types.MakeSigner(chainConfig, blockNum, header.Time) // Returns an array of trace arrays, one trace array for each transaction trace, err := api.callTransaction(ctx, tx, header, traceTypes, txnIndex, *gasBailOut, signer, chainConfig, traceConfig) @@ -1133,18 +1129,8 @@ func (api *TraceAPIImpl) Call(ctx context.Context, args TraceCallParam, traceTyp ibs := state.New(stateReader) - // Setup context so it may be cancelled the call has completed - // or, in case of unmetered gas, setup a context with a timeout. - var cancel context.CancelFunc - if api.evmCallTimeout > 0 { - ctx, cancel = context.WithTimeout(ctx, api.evmCallTimeout) - } else { - ctx, cancel = context.WithCancel(ctx) - } - - // Make sure the context is cancelled when the call has completed - // this makes sure resources are cleaned up. - defer cancel() + _, storeEVM, cleanup := setupEVMTimeout(ctx, api.evmCallTimeout) + defer cleanup() traceResult := &TraceCallResult{Trace: []*ParityTrace{}} var traceTypeTrace, traceTypeStateDiff, traceTypeVmTrace bool @@ -1210,9 +1196,7 @@ func (api *TraceAPIImpl) Call(ctx context.Context, args TraceCallParam, traceTyp if precompiles != nil { evm.SetPrecompiles(precompiles) } - - stop := context.AfterFunc(ctx, evm.Cancel) - defer stop() + storeEVM(evm) gp := new(protocol.GasPool).AddGas(msg.Gas()).AddBlobGas(msg.BlobGas()) var execResult *evmtypes.ExecutionResult @@ -1832,13 +1816,8 @@ func (api *TraceAPIImpl) RawTransaction(ctx context.Context, encodedTx hexutil.B return nil, err } - var cancel context.CancelFunc - if api.evmCallTimeout > 0 { - ctx, cancel = context.WithTimeout(ctx, api.evmCallTimeout) - } else { - ctx, cancel = context.WithCancel(ctx) - } - defer cancel() + _, storeEVM, cleanup := setupEVMTimeout(ctx, api.evmCallTimeout) + defer cleanup() traceResult := &TraceCallResult{Trace: []*ParityTrace{}} var traceTypeTrace, traceTypeStateDiff, traceTypeVmTrace bool @@ -1889,9 +1868,7 @@ func (api *TraceAPIImpl) RawTransaction(ctx context.Context, encodedTx hexutil.B blockCtx.MaxGasLimit = true evm := vm.NewEVM(blockCtx, txCtx, ibs, chainConfig, vm.Config{Tracer: ot.Tracer().Hooks}) - - stop := context.AfterFunc(ctx, evm.Cancel) - defer stop() + storeEVM(evm) gp := new(protocol.GasPool).AddGas(msg.Gas()).AddBlobGas(msg.BlobGas()) var execResult *evmtypes.ExecutionResult diff --git a/rpc/jsonrpc/trace_filtering.go b/rpc/jsonrpc/trace_filtering.go index 5dc78928ab2..88af147c097 100644 --- a/rpc/jsonrpc/trace_filtering.go +++ b/rpc/jsonrpc/trace_filtering.go @@ -110,10 +110,6 @@ func (api *TraceAPIImpl) Transaction(ctx context.Context, txHash common.Hash, ga return nil, err } - if isBorStateSyncTxn { - txIndex = -1 - } - bn := hexutil.Uint64(blockNumber) hash := header.Hash() signer := types.MakeSigner(chainConfig, blockNumber, header.Time)