Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

CallMany per txn gran #13399

Open
wants to merge 6 commits into
base: main
Choose a base branch
from
Open

CallMany per txn gran #13399

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
ae202bf
commit
Jan 12, 2025
b08607d
commit
Jan 12, 2025
54f5dd2
commit
Jan 12, 2025
9acfea1
commit
Jan 12, 2025
b6c10ee
commit
Jan 13, 2025
883f820
commit
Jan 13, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
281 changes: 237 additions & 44 deletions turbo/jsonrpc/trace_adhoc.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,8 @@ import (
"encoding/json"
"errors"
"fmt"
"github.com/erigontech/erigon-lib/kv/rawdbv3"
"github.com/erigontech/erigon/turbo/snapshotsync/freezeblocks"
"math"
"strings"

Expand Down Expand Up @@ -817,7 +819,7 @@ func (api *TraceAPIImpl) ReplayTransaction(ctx context.Context, txHash libcommon
}

var isBorStateSyncTxn bool
blockNum, _, ok, err := api.txnLookup(ctx, tx, txHash)
blockNum, txNum, ok, err := api.txnLookup(ctx, tx, txHash)
if err != nil {
return nil, err
}
Expand All @@ -843,30 +845,31 @@ func (api *TraceAPIImpl) ReplayTransaction(ctx context.Context, txHash libcommon
isBorStateSyncTxn = true
}

block, err := api.blockByNumberWithSenders(ctx, tx, blockNum)
header, err := api.headerByRPCNumber(ctx, rpc.BlockNumber(blockNum), tx)
if err != nil {
return nil, err
}
if block == nil {
return nil, nil

txNumsReader := rawdbv3.TxNums.WithCustomReadTxNumFunc(freezeblocks.ReadTxNumFuncFromBlockReader(ctx, api._blockReader))

txNumMin, err := txNumsReader.Min(tx, blockNum)
if err != nil {
return nil, err
}

var txnIndex int
for idx := 0; idx < block.Transactions().Len() && !isBorStateSyncTxn; idx++ {
txn := block.Transactions()[idx]
if txn.Hash() == txHash {
txnIndex = idx
break
}
if txNumMin+2 > txNum {
return nil, fmt.Errorf("uint underflow txnums error txNum: %d, txNumMin: %d, blockNum: %d", txNum, txNumMin, blockNum)
}

var txnIndex = int(txNum - txNumMin - 2)

if isBorStateSyncTxn {
txnIndex = block.Transactions().Len()
txnIndex = -1
}

signer := types.MakeSigner(chainConfig, blockNum, block.Time())
signer := types.MakeSigner(chainConfig, blockNum, header.Time)
// Returns an array of trace arrays, one trace array for each transaction
traces, _, err := api.callManyTransactions(ctx, tx, block, traceTypes, txnIndex, *gasBailOut, signer, chainConfig, traceConfig)
trace, _, err := api.callTransaction(ctx, tx, header, traceTypes, txnIndex, *gasBailOut, signer, chainConfig, traceConfig)
if err != nil {
return nil, err
}
Expand All @@ -886,25 +889,18 @@ func (api *TraceAPIImpl) ReplayTransaction(ctx context.Context, txHash libcommon
}
result := &TraceCallResult{}

for txno, trace := range traces {
// We're only looking for a specific transaction
if txno == txnIndex {
result.Output = trace.Output
if traceTypeTrace {
result.Trace = trace.Trace
}
if traceTypeStateDiff {
result.StateDiff = trace.StateDiff
}
if traceTypeVmTrace {
result.VmTrace = trace.VmTrace
}

return trace, nil
}
result.Output = trace.Output
if traceTypeTrace {
result.Trace = trace.Trace
}
if traceTypeStateDiff {
result.StateDiff = trace.StateDiff
}
if traceTypeVmTrace {
result.VmTrace = trace.VmTrace
}

return result, nil
return trace, nil
}

func (api *TraceAPIImpl) ReplayBlockTransactions(ctx context.Context, blockNrOrHash rpc.BlockNumberOrHash, traceTypes []string, gasBailOut *bool, traceConfig *config.TraceConfig) ([]*TraceCallResult, error) {
Expand Down Expand Up @@ -950,7 +946,7 @@ func (api *TraceAPIImpl) ReplayBlockTransactions(ctx context.Context, blockNrOrH

signer := types.MakeSigner(chainConfig, blockNumber, block.Time())
// Returns an array of trace arrays, one trace array for each transaction
traces, _, err := api.callManyTransactions(ctx, tx, block, traceTypes, -1 /* all txn indices */, *gasBailOut, signer, chainConfig, traceConfig)
traces, _, err := api.callBlock(ctx, tx, block, traceTypes, *gasBailOut, signer, chainConfig, traceConfig)
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -1212,14 +1208,14 @@ func (api *TraceAPIImpl) CallMany(ctx context.Context, calls json.RawMessage, pa
cachedWriter := state.NewCachedWriter(noop, stateCache)
ibs := state.New(cachedReader)

return api.doCallMany(ctx, dbtx, stateReader, stateCache, cachedWriter, ibs,
msgs, callParams, parentNrOrHash, nil, true /* gasBailout */, -1 /* all txn indices */, traceConfig)
return api.doCallBlock(ctx, dbtx, stateReader, stateCache, cachedWriter, ibs,
msgs, callParams, parentNrOrHash, nil, true /* gasBailout */, traceConfig)
}

func (api *TraceAPIImpl) doCallMany(ctx context.Context, dbtx kv.Tx, stateReader state.StateReader,
func (api *TraceAPIImpl) doCallBlock(ctx context.Context, dbtx kv.Tx, stateReader state.StateReader,
stateCache *shards.StateCache, cachedWriter state.StateWriter, ibs *state.IntraBlockState,
msgs []types.Message, callParams []TraceCallParam,
parentNrOrHash *rpc.BlockNumberOrHash, header *types.Header, gasBailout bool, txIndexNeeded int,
parentNrOrHash *rpc.BlockNumberOrHash, header *types.Header, gasBailout bool,
traceConfig *config.TraceConfig,
) ([]*TraceCallResult, error) {
chainConfig, err := api.chainConfig(ctx, dbtx)
Expand Down Expand Up @@ -1299,7 +1295,7 @@ func (api *TraceAPIImpl) doCallMany(ctx context.Context, dbtx kv.Tx, stateReader

traceResult := &TraceCallResult{Trace: []*ParityTrace{}, TransactionHash: args.txHash}
vmConfig := vm.Config{}
if (traceTypeTrace && (txIndexNeeded == -1 || txIndex == txIndexNeeded)) || traceTypeVmTrace {
if traceTypeTrace || traceTypeVmTrace {
var ot OeTracer
ot.config, err = parseOeTracerConfig(traceConfig)
if err != nil {
Expand All @@ -1308,7 +1304,7 @@ func (api *TraceAPIImpl) doCallMany(ctx context.Context, dbtx kv.Tx, stateReader
ot.compat = api.compatibility
ot.r = traceResult
ot.idx = []string{fmt.Sprintf("%d-", txIndex)}
if traceTypeTrace && (txIndexNeeded == -1 || txIndex == txIndexNeeded) {
if traceTypeTrace {
ot.traceAddr = []int{}
}
if traceTypeVmTrace {
Expand Down Expand Up @@ -1389,7 +1385,11 @@ func (api *TraceAPIImpl) doCallMany(ctx context.Context, dbtx kv.Tx, stateReader
return nil, err
}
}
sd.CompareStates(initialIbs, ibs)
if sd != nil {
if err = sd.CompareStates(initialIbs, ibs); err != nil {
return nil, err
}
}
if err = ibs.CommitBlock(chainRules, cachedWriter); err != nil {
return nil, err
}
Expand All @@ -1407,16 +1407,209 @@ func (api *TraceAPIImpl) doCallMany(ctx context.Context, dbtx kv.Tx, stateReader
traceResult.Trace = []*ParityTrace{}
}
results = append(results, traceResult)
// When txIndexNeeded is not -1, we are tracing specific transaction in the block and not the entire block, so we stop after we've traced
// the required transaction
if txIndexNeeded != -1 && txIndex == txIndexNeeded {
break
}
}

return results, nil
}

func (api *TraceAPIImpl) doCall(ctx context.Context, dbtx kv.Tx, stateReader state.StateReader,
stateCache *shards.StateCache, cachedWriter state.StateWriter, ibs *state.IntraBlockState,
msg types.Message, callParam TraceCallParam,
parentNrOrHash *rpc.BlockNumberOrHash, header *types.Header, gasBailout bool, txIndex int,
traceConfig *config.TraceConfig,
) (*TraceCallResult, error) {
chainConfig, err := api.chainConfig(ctx, dbtx)
if err != nil {
return nil, err
}
engine := api.engine()

if parentNrOrHash == nil {
var num = rpc.LatestBlockNumber
parentNrOrHash = &rpc.BlockNumberOrHash{BlockNumber: &num}
}
blockNumber, hash, _, err := rpchelper.GetBlockNumber(ctx, *parentNrOrHash, dbtx, api._blockReader, api.filters)
if err != nil {
return nil, err
}
noop := state.NewNoopWriter()

parentHeader, err := api.headerByRPCNumber(ctx, rpc.BlockNumber(blockNumber), dbtx)
if err != nil {
return nil, err
}
if parentHeader == nil {
return nil, fmt.Errorf("parent header %d(%x) not found", blockNumber, hash)
}

// 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()

useParent := false
if header == nil {
header = parentHeader
useParent = true
}

var baseTxNum uint64
historicalStateReader, isHistoricalStateReader := stateReader.(state.HistoricalStateReader)
if isHistoricalStateReader {
baseTxNum = historicalStateReader.GetTxNum()
}

blockCtx := transactions.NewEVMBlockContext(engine, header, parentNrOrHash.RequireCanonical, dbtx, api._blockReader, chainConfig)

if isHistoricalStateReader {
historicalStateReader.SetTxNum(baseTxNum + uint64(txIndex))
}
if err := libcommon.Stopped(ctx.Done()); err != nil {
return nil, err
}

var traceTypeTrace, traceTypeStateDiff, traceTypeVmTrace bool
args := callParam
for _, traceType := range args.traceTypes {
switch traceType {
case TraceTypeTrace:
traceTypeTrace = true
case TraceTypeStateDiff:
traceTypeStateDiff = true
case TraceTypeVmTrace:
traceTypeVmTrace = true
default:
return nil, fmt.Errorf("unrecognized trace type: %s", traceType)
}
}

traceResult := &TraceCallResult{Trace: []*ParityTrace{}, TransactionHash: args.txHash}
vmConfig := vm.Config{}
if traceTypeTrace || traceTypeVmTrace {
var ot OeTracer
ot.config, err = parseOeTracerConfig(traceConfig)
if err != nil {
return nil, err
}
ot.compat = api.compatibility
ot.r = traceResult
ot.idx = []string{fmt.Sprintf("%d-", txIndex)}
if traceTypeTrace {
ot.traceAddr = []int{}
}
if traceTypeVmTrace {
traceResult.VmTrace = &VmTrace{Ops: []*VmTraceOp{}}
}
vmConfig.Debug = true
vmConfig.Tracer = &ot
}

if useParent {
blockCtx.GasLimit = math.MaxUint64
blockCtx.MaxGasLimit = true
}

// Clone the state cache before applying the changes for diff after transaction execution, clone is discarded
var cloneReader state.StateReader
var sd *StateDiff
if traceTypeStateDiff {
cloneCache := stateCache.Clone()
cloneReader = state.NewCachedReader(stateReader, cloneCache)
//cloneReader = stateReader
if isHistoricalStateReader {
historicalStateReader.SetTxNum(baseTxNum + uint64(txIndex))
}
sdMap := make(map[libcommon.Address]*StateDiffAccount)
traceResult.StateDiff = sdMap
sd = &StateDiff{sdMap: sdMap}
}

ibs.Reset()
var finalizeTxStateWriter state.StateWriter
if sd != nil {
finalizeTxStateWriter = sd
} else {
finalizeTxStateWriter = noop
}

var txFinalized bool
var execResult *evmtypes.ExecutionResult
if args.isBorStateSyncTxn {
txFinalized = true
var stateSyncEvents []*types.Message
stateSyncEvents, err = api.stateSyncEvents(ctx, dbtx, header.Hash(), blockNumber, chainConfig)
if err != nil {
return nil, err
}

execResult, err = tracer.TraceBorStateSyncTxnTraceAPI(
ctx,
&vmConfig,
chainConfig,
ibs,
finalizeTxStateWriter,
blockCtx,
header.Hash(),
header.Number.Uint64(),
header.Time,
stateSyncEvents,
)
} else {
ibs.SetTxContext(txIndex)
txCtx := core.NewEVMTxContext(msg)
evm := vm.NewEVM(blockCtx, txCtx, ibs, chainConfig, vmConfig)
gp := new(core.GasPool).AddGas(msg.Gas()).AddBlobGas(msg.BlobGas())

execResult, err = core.ApplyMessage(evm, msg, gp, true /* refunds */, gasBailout /*gasBailout*/)
}
if err != nil {
return nil, fmt.Errorf("first run for txIndex %d error: %w", txIndex, err)
}

chainRules := chainConfig.Rules(blockCtx.BlockNumber, blockCtx.Time)
traceResult.Output = libcommon.CopyBytes(execResult.ReturnData)
if traceTypeStateDiff {
initialIbs := state.New(cloneReader)
if !txFinalized {
if err = ibs.FinalizeTx(chainRules, sd); err != nil {
return nil, err
}
}

if sd != nil {
if err = sd.CompareStates(initialIbs, ibs); err != nil {
return nil, err
}
}

if err = ibs.CommitBlock(chainRules, cachedWriter); err != nil {
return nil, err
}
} else {
if !txFinalized {
if err = ibs.FinalizeTx(chainRules, noop); err != nil {
return nil, err
}
}
if err = ibs.CommitBlock(chainRules, cachedWriter); err != nil {
return nil, err
}
}
if !traceTypeTrace {
traceResult.Trace = []*ParityTrace{}
}

return traceResult, nil
}

// RawTransaction implements trace_rawTransaction.
func (api *TraceAPIImpl) RawTransaction(ctx context.Context, txHash libcommon.Hash, traceTypes []string) ([]interface{}, error) {
var stub []interface{}
Expand Down
Loading
Loading