From afc20286fbdfbb3f42650d632d7e31b264df6e88 Mon Sep 17 00:00:00 2001 From: Daniel Liu <139250065@qq.com> Date: Tue, 9 Sep 2025 10:18:10 +0800 Subject: [PATCH] eth: add request cancellation checks #26320 (#1314) This ensures that RPC method handlers will react to a timeout or cancelled request soon after the event occurs. Co-authored-by: Sina Mahmoodi --- eth/api.go | 2 +- eth/api_backend.go | 4 ++-- eth/state_accessor.go | 13 ++++++++++--- eth/tracers/api.go | 22 +++++++++++++++++----- 4 files changed, 30 insertions(+), 11 deletions(-) diff --git a/eth/api.go b/eth/api.go index bc2b26a70f..d6515c945b 100644 --- a/eth/api.go +++ b/eth/api.go @@ -404,7 +404,7 @@ func (api *DebugAPI) StorageRangeAt(ctx context.Context, blockHash common.Hash, if block == nil { return StorageRangeResult{}, fmt.Errorf("block %#x not found", blockHash) } - _, _, statedb, release, err := api.eth.stateAtTransaction(block, txIndex, 0) + _, _, statedb, release, err := api.eth.stateAtTransaction(ctx, block, txIndex, 0) if err != nil { return StorageRangeResult{}, err } diff --git a/eth/api_backend.go b/eth/api_backend.go index 87400638c8..999dcbfe35 100644 --- a/eth/api_backend.go +++ b/eth/api_backend.go @@ -415,11 +415,11 @@ func (b *EthAPIBackend) Miner() *miner.Miner { } func (b *EthAPIBackend) StateAtBlock(ctx context.Context, block *types.Block, reexec uint64, base *state.StateDB, readOnly bool, preferDisk bool) (*state.StateDB, tracers.StateReleaseFunc, error) { - return b.eth.StateAtBlock(block, reexec, base, readOnly, preferDisk) + return b.eth.StateAtBlock(ctx, block, reexec, base, readOnly, preferDisk) } func (b *EthAPIBackend) StateAtTransaction(ctx context.Context, block *types.Block, txIndex int, reexec uint64) (core.Message, vm.BlockContext, *state.StateDB, tracers.StateReleaseFunc, error) { - return b.eth.stateAtTransaction(block, txIndex, reexec) + return b.eth.stateAtTransaction(ctx, block, txIndex, reexec) } func (b *EthAPIBackend) GetIPCClient() (bind.ContractBackend, error) { diff --git a/eth/state_accessor.go b/eth/state_accessor.go index 2cdfc4c9e0..a5fe1872f8 100644 --- a/eth/state_accessor.go +++ b/eth/state_accessor.go @@ -17,6 +17,7 @@ package eth import ( + "context" "errors" "fmt" "math/big" @@ -57,7 +58,7 @@ var noopReleaser = tracers.StateReleaseFunc(func() {}) // - preferDisk: this arg can be used by the caller to signal that even though the 'base' is // provided, it would be preferable to start from a fresh state, if we have it // on disk. -func (eth *Ethereum) StateAtBlock(block *types.Block, reexec uint64, base *state.StateDB, readOnly bool, preferDisk bool) (statedb *state.StateDB, release tracers.StateReleaseFunc, err error) { +func (eth *Ethereum) StateAtBlock(ctx context.Context, block *types.Block, reexec uint64, base *state.StateDB, readOnly bool, preferDisk bool) (statedb *state.StateDB, release tracers.StateReleaseFunc, err error) { var ( current *types.Block database state.Database @@ -112,6 +113,9 @@ func (eth *Ethereum) StateAtBlock(block *types.Block, reexec uint64, base *state } // Database does not have the state for the given block, try to regenerate for i := uint64(0); i < reexec; i++ { + if err := ctx.Err(); err != nil { + return nil, nil, err + } if current.NumberU64() == 0 { return nil, nil, errors.New("genesis state is missing") } @@ -143,6 +147,9 @@ func (eth *Ethereum) StateAtBlock(block *types.Block, reexec uint64, base *state parent common.Hash ) for current.NumberU64() < origin { + if err := ctx.Err(); err != nil { + return nil, nil, err + } // Print progress logs if long enough time elapsed if time.Since(logged) > 8*time.Second && report { log.Info("Regenerating historical state", "block", current.NumberU64()+1, "target", origin, "remaining", origin-current.NumberU64()-1, "elapsed", time.Since(start)) @@ -183,7 +190,7 @@ func (eth *Ethereum) StateAtBlock(block *types.Block, reexec uint64, base *state } // stateAtTransaction returns the execution environment of a certain transaction. -func (eth *Ethereum) stateAtTransaction(block *types.Block, txIndex int, reexec uint64) (core.Message, vm.BlockContext, *state.StateDB, tracers.StateReleaseFunc, error) { +func (eth *Ethereum) stateAtTransaction(ctx context.Context, block *types.Block, txIndex int, reexec uint64) (core.Message, vm.BlockContext, *state.StateDB, tracers.StateReleaseFunc, error) { // Short circuit if it's genesis block. if block.NumberU64() == 0 { return nil, vm.BlockContext{}, nil, nil, errors.New("no transaction in genesis") @@ -195,7 +202,7 @@ func (eth *Ethereum) stateAtTransaction(block *types.Block, txIndex int, reexec } // Lookup the statedb of parent block from the live database, // otherwise regenerate it on the flight. - statedb, release, err := eth.StateAtBlock(parent, reexec, nil, true, false) + statedb, release, err := eth.StateAtBlock(ctx, parent, reexec, nil, true, false) if err != nil { return nil, vm.BlockContext{}, nil, nil, err } diff --git a/eth/tracers/api.go b/eth/tracers/api.go index 19438e3eb5..b5a4c40f63 100644 --- a/eth/tracers/api.go +++ b/eth/tracers/api.go @@ -526,6 +526,9 @@ func (api *API) IntermediateRoots(ctx context.Context, hash common.Hash, config ) feeCapacity := state.GetTRC21FeeCapacityFromState(statedb) for i, tx := range block.Transactions() { + if err := ctx.Err(); err != nil { + return nil, err + } var balance *big.Int if tx.To() != nil { // Bypass the validation for trading and lending transactions as their nonce are not incremented @@ -590,14 +593,13 @@ func (api *API) traceBlock(ctx context.Context, block *types.Block, config *Trac signer = types.MakeSigner(api.backend.ChainConfig(), block.Number()) txs = block.Transactions() results = make([]*txTraceResult, len(txs)) - - pend = new(sync.WaitGroup) - jobs = make(chan *txTraceTask, len(txs)) + pend sync.WaitGroup ) threads := runtime.NumCPU() if threads > len(txs) { threads = len(txs) } + jobs := make(chan *txTraceTask, threads) blockHash := block.Hash() for th := 0; th < threads; th++ { blockCtx := core.NewEVMBlockContext(block.Header(), api.chainContext(ctx), nil) @@ -629,13 +631,22 @@ func (api *API) traceBlock(ctx context.Context, block *types.Block, config *Trac } }() } + // Feed the transactions into the tracers and return feeCapacity := state.GetTRC21FeeCapacityFromState(statedb) var failed error blockCtx := core.NewEVMBlockContext(block.Header(), api.chainContext(ctx), nil) +txloop: for i, tx := range txs { // Send the trace task over for execution - jobs <- &txTraceTask{statedb: statedb.Copy(), index: i} + task := &txTraceTask{statedb: statedb.Copy(), index: i} + select { + case <-ctx.Done(): + failed = ctx.Err() + break txloop + case jobs <- task: + } + var balance *big.Int if tx.To() != nil { // Bypass the validation for trading and lending transactions as their nonce are not incremented @@ -654,12 +665,13 @@ func (api *API) traceBlock(ctx context.Context, block *types.Block, config *Trac owner := common.Address{} if _, err := core.ApplyMessage(vmenv, msg, new(core.GasPool).AddGas(msg.Gas()), owner); err != nil { failed = err - break + break txloop } // Finalize the state so any modifications are written to the trie // Only delete empty objects if EIP158/161 (a.k.a Spurious Dragon) is in effect statedb.Finalise(vmenv.ChainConfig().IsEIP158(block.Number())) } + close(jobs) pend.Wait()