diff --git a/rpc/jsonrpc/debug_api.go b/rpc/jsonrpc/debug_api.go index 2a998c5a8a1..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)) @@ -807,18 +797,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/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..8fbd5828eae 100644 --- a/rpc/jsonrpc/erigon_receipts.go +++ b/rpc/jsonrpc/erigon_receipts.go @@ -20,12 +20,14 @@ import ( "context" "errors" "fmt" + "math/big" "github.com/RoaringBitmap/roaring/v2" "github.com/erigontech/erigon/common" "github.com/erigontech/erigon/common/hexutil" "github.com/erigontech/erigon/common/log/v3" + "github.com/erigontech/erigon/db/kv" "github.com/erigontech/erigon/db/kv/order" "github.com/erigontech/erigon/db/kv/rawdbv3" "github.com/erigontech/erigon/execution/exec" @@ -77,6 +79,37 @@ func (api *ErigonImpl) GetLogsByHash(ctx context.Context, hash common.Hash) ([][ return logs, nil } +// logRangeLatestOnly resolves a filter range where the only negative tag accepted for +// FromBlock/ToBlock is "latest"; other tags (pending, safe, finalized) are rejected. +func logRangeLatestOnly(tx kv.Tx, crit filters.FilterCriteria) (begin, end uint64, err error) { + latest, err := rpchelper.GetLatestBlockNumber(tx) + if err != nil { + return 0, 0, err + } + begin, err = resolveLogBound(crit.FromBlock, 0, "FromBlock") + if err != nil { + return 0, 0, err + } + 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 @@ -97,28 +130,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 +208,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} @@ -420,7 +418,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 +429,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_api.go b/rpc/jsonrpc/eth_api.go index 98163d39a39..22cc2572224 100644 --- a/rpc/jsonrpc/eth_api.go +++ b/rpc/jsonrpc/eth_api.go @@ -239,6 +239,44 @@ 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. 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 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 +} + 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..76a0255ce0d 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" @@ -75,14 +74,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 @@ -147,30 +142,15 @@ 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() + _, storeEVM, cleanup := setupEVMTimeout(ctx, timeout) + defer cleanup() + storeEVM(evm) // Setup the gas pool (also for unmetered requests) // and apply the message. @@ -189,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 64a3973380a..df9e4836bf3 100644 --- a/rpc/jsonrpc/eth_callMany.go +++ b/rpc/jsonrpc/eth_callMany.go @@ -37,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 { @@ -49,6 +50,47 @@ type StateContext struct { TransactionIndex *int } +func validateBundles(bundles []Bundle) error { + for _, bundle := range bundles { + if len(bundle.Transactions) != 0 { + return nil + } + } + return errors.New("empty bundles") +} + +// 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 + if timeout > 0 { + ctx, cancel = context.WithTimeout(ctx, timeout) + } else { + ctx, cancel = context.WithCancel(ctx) + } + + stop := context.AfterFunc(ctx, func() { + if evm := evmPtr.Load(); evm != nil { + evm.Cancel() + } + }) + store := func(evm *vm.EVM) { + evmPtr.Store(evm) + if ctx.Err() != nil { + evm.Cancel() + } + } + return ctx, store, 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 +111,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,50 +165,25 @@ 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 + timeout := api.evmCallTimeout + + if timeoutMilliSecondsPtr != nil && *timeoutMilliSecondsPtr > 0 { + timeout = time.Duration(*timeoutMilliSecondsPtr) * time.Millisecond } + ctx, storeEVM, 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{}) + storeEVM(evm) 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() - // Setup the gas pool (also for unmetered requests) // and apply the message. gp := new(protocol.GasPool).AddGas(math.MaxUint64).AddBlobGas(math.MaxUint64) @@ -188,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 { @@ -227,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 9e1746e9ae6..710a449a967 100644 --- a/rpc/jsonrpc/eth_callMany_test.go +++ b/rpc/jsonrpc/eth_callMany_test.go @@ -17,15 +17,20 @@ package jsonrpc import ( + "bytes" "context" "encoding/hex" "fmt" "math/big" "strconv" "testing" + "time" "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" @@ -34,12 +39,56 @@ 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" + "github.com/erigontech/erigon/rpc/jsonstream" "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) + 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/eth_filters.go b/rpc/jsonrpc/eth_filters.go index ce90d4e7779..3501d4edc60 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,13 @@ 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. 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 } notifier, supported := rpc.NotifierFromContext(ctx) @@ -164,21 +169,22 @@ 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() + + 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 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: if !ok { - log.Warn("[rpc] new heads channel was closed") + log.Warn(closedWarn) return } + notify(emit, item) case <-rpcSub.Err(): return } @@ -188,169 +194,78 @@ 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 - } - case <-rpcSub.Err(): - 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(emit func(payload any), h *types.Header) { + if h != nil { + emit(h) } - } - }() + }, + "[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() - - go func() { - defer dbg.LogPanic() - txsCh, id := api.filters.SubscribePendingTxs(512) - defer api.filters.UnsubscribePendingTxs(id) + return api.subscribePendingTransactions(ctx, 512, true) +} - 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) - } +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(emit func(payload any), txs []types.Transaction) { + for _, t := range txs { + if t != nil { + if fullTx { + emit(newRPCPendingTransaction(t, nil, nil)) + } else { + emit(t.Hash()) } } - 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 - } - case <-rpcSub.Err(): - 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(emit func(payload any), h *types.Log) { + if h != nil { + emit(h) } - } - }() - - 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) - } - } - if !ok { - log.Warn("[rpc] receipts channel was closed") - return - } - case <-rpcSub.Err(): - return + 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(emit func(payload any), protoReceipt *remoteproto.SubscribeReceiptsReply) { + if protoReceipt != nil { + receipt := ethutils.MarshalSubscribeReceipt(protoReceipt) + emit([]map[string]any{receipt}) } - } - }() - - 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") diff --git a/rpc/jsonrpc/eth_receipts.go b/rpc/jsonrpc/eth_receipts.go index 0868ecfb4d7..7c98c22b450 100644 --- a/rpc/jsonrpc/eth_receipts.go +++ b/rpc/jsonrpc/eth_receipts.go @@ -77,6 +77,37 @@ 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 + } + borReceipt, err := api.borReceiptForBlock(ctx, tx, chainConfig, block) + if err != nil { + return nil, nil, err + } + 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, err + } + 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) { return api.receiptsGenerator.GetReceipt(ctx, cc, tx, header, txn, index, txNum, postState) } @@ -89,9 +120,68 @@ 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 + } + + 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) @@ -113,63 +203,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 { @@ -510,14 +550,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 +569,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 @@ -547,25 +583,17 @@ 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 } - var txnIndex = int(txNum - txNumMin - 1) - txn, err := api._blockReader.TxnByIdxInBlock(ctx, overlayTx, header.Number.Uint64(), txnIndex) if err != nil { return nil, err @@ -640,7 +668,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 } @@ -650,20 +678,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 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 61451cc5172..495282e3605 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" @@ -50,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 { @@ -70,15 +60,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 +92,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 +150,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/overlay_api.go b/rpc/jsonrpc/overlay_api.go index 97e09af02f5..8b5e1fe4f29 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,16 +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 := 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 := transactions.MakeBlockHashProvider(ctx, tx, api._blockReader, overrideBlockHash) blockCtx = protocol.NewEVMBlockContext(header, getHash, api.engine(), accounts.NilAddress, chainConfig) @@ -462,43 +454,22 @@ 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 - } + timeout := api.OverlayReplayBlockTimeout + ctx, storeEVM, cleanup := setupEVMTimeout(ctx, timeout) + defer cleanup() + + getHash := transactions.MakeBlockHashProvider(ctx, tx, api._blockReader, overrideBlockHash) blockCtx = protocol.NewEVMBlockContext(header, getHash, api.engine(), accounts.NilAddress, chainConfig) signer := types.MakeSigner(chainConfig, blockNum, blockCtx.Time) 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) - - stop := context.AfterFunc(ctx, evm.Cancel) - defer stop() + evm = vm.NewEVM(blockCtx, evmtypes.TxContext{}, statedb, chainConfig, vm.Config{}) + storeEVM(evm) receipts, err := api.getReceipts(ctx, tx, block) if err != nil { return nil, err @@ -575,54 +546,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) +} diff --git a/rpc/jsonrpc/trace_adhoc.go b/rpc/jsonrpc/trace_adhoc.go index cf9433eeb4c..543c9cfd1e3 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,21 +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 - } - 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) @@ -1154,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 @@ -1231,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 @@ -1853,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 @@ -1910,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 dd0921a7895..88af147c097 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,21 +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 - } - bn := hexutil.Uint64(blockNumber) hash := header.Hash() signer := types.MakeSigner(chainConfig, blockNumber, header.Time) diff --git a/rpc/jsonrpc/tracing.go b/rpc/jsonrpc/tracing.go index ed0c1e892c8..52a9c42a93f 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) } @@ -503,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()) @@ -562,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 := 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 }