Skip to content

rpc: use pre-configured txNumReader (it's thread-safe) #14575

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

Merged
merged 2 commits into from
Apr 13, 2025
Merged

rpc: use pre-configured txNumReader (it's thread-safe) #14575

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
19b352e
save
AskAlexSharov Apr 12, 2025
b7d7ce1
save
AskAlexSharov Apr 12, 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
3 changes: 2 additions & 1 deletion cmd/state/commands/opcode_tracer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -585,6 +585,7 @@ func OpcodeTracer(genesis *types.Genesis, blockNum uint64, chaindata string, num

timeLastBlock := startTime
blockNumLastReport := blockNum
txNumReader := rawdbv3.TxNums.WithCustomReadTxNumFunc(freezeblocks.ReadTxNumFuncFromBlockReader(context.Background(), blockReader))

for !interrupt {
var block *types.Block
Expand All @@ -608,7 +609,7 @@ func OpcodeTracer(genesis *types.Genesis, blockNum uint64, chaindata string, num
ot.fsumWriter = bufio.NewWriter(fsum)
}

dbstate, err := rpchelper.CreateHistoryStateReader(historyTx, rawdbv3.TxNums.WithCustomReadTxNumFunc(freezeblocks.ReadTxNumFuncFromBlockReader(context.TODO(), blockReader)),
dbstate, err := rpchelper.CreateHistoryStateReader(historyTx, txNumReader,
block.NumberU64(), 0, chainConfig.ChainName)
if err != nil {
return err
Expand Down
4 changes: 2 additions & 2 deletions eth/stagedsync/stage_custom_trace.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,11 +68,11 @@ func StageCustomTraceCfg(db kv.TemporalRwDB, prune prune.Mode, dirs datadir.Dirs
}

func SpawnCustomTrace(cfg CustomTraceCfg, ctx context.Context, logger log.Logger) error {
txNumsReader := rawdbv3.TxNums.WithCustomReadTxNumFunc(freezeblocks.ReadTxNumFuncFromBlockReader(ctx, cfg.execArgs.BlockReader))

var startBlock, endBlock uint64
stepSize := cfg.db.(state2.HasAgg).Agg().(*state2.Aggregator).StepSize()
if err := cfg.db.View(ctx, func(tx kv.Tx) (err error) {
txNumsReader := rawdbv3.TxNums.WithCustomReadTxNumFunc(freezeblocks.ReadTxNumFuncFromBlockReader(ctx, cfg.execArgs.BlockReader))

ac := tx.(state2.HasAggTx).AggTx().(*state2.AggregatorRoTx)
txNum := ac.DbgDomain(kv.AccountsDomain).FirstStepNotInFiles() * stepSize
var ok bool
Expand Down
23 changes: 7 additions & 16 deletions rpc/jsonrpc/debug_api.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,15 +92,14 @@ func (api *PrivateDebugAPIImpl) StorageRangeAt(ctx context.Context, blockHash co
}
defer tx.Rollback()

txNumsReader := rawdbv3.TxNums.WithCustomReadTxNumFunc(freezeblocks.ReadTxNumFuncFromBlockReader(ctx, api._blockReader))
number, err := api._blockReader.HeaderNumber(ctx, tx, blockHash)
if err != nil {
return StorageRangeResult{}, err
}
if number == nil {
return StorageRangeResult{}, nil
}
minTxNum, err := txNumsReader.Min(tx, *number)
minTxNum, err := api._txNumReader.Min(tx, *number)
if err != nil {
return StorageRangeResult{}, err
}
Expand Down Expand Up @@ -188,8 +187,6 @@ func (api *PrivateDebugAPIImpl) GetModifiedAccountsByNumber(ctx context.Context,
return nil, err
}

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

// forces negative numbers to fail (too large) but allows zero
startNum := uint64(startNumber.Int64())
if startNum > latestBlock {
Expand All @@ -211,11 +208,11 @@ func (api *PrivateDebugAPIImpl) GetModifiedAccountsByNumber(ctx context.Context,
return nil, fmt.Errorf("start block (%d) must be less than or equal to end block (%d)", startNum, endNum)
}
//[from, to)
startTxNum, err := txNumsReader.Min(tx, startNum)
startTxNum, err := api._txNumReader.Min(tx, startNum)
if err != nil {
return nil, err
}
endTxNum, err := txNumsReader.Min(tx, endNum)
endTxNum, err := api._txNumReader.Min(tx, endNum)
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -255,8 +252,6 @@ func (api *PrivateDebugAPIImpl) GetModifiedAccountsByHash(ctx context.Context, s
}
defer tx.Rollback()

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

startBlock, err := api.blockByHashWithSenders(ctx, tx, startHash)
if err != nil {
return nil, err
Expand All @@ -283,11 +278,11 @@ func (api *PrivateDebugAPIImpl) GetModifiedAccountsByHash(ctx context.Context, s
}

//[from, to)
startTxNum, err := txNumsReader.Min(tx, startNum)
startTxNum, err := api._txNumReader.Min(tx, startNum)
if err != nil {
return nil, err
}
endTxNum, err := txNumsReader.Min(tx, endNum)
endTxNum, err := api._txNumReader.Min(tx, endNum)
if err != nil {
return nil, err
}
Expand All @@ -301,8 +296,6 @@ func (api *PrivateDebugAPIImpl) AccountAt(ctx context.Context, blockHash common.
}
defer tx.Rollback()

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

number, err := api._blockReader.HeaderNumber(ctx, tx, blockHash)
if err != nil {
return &AccountResult{}, err
Expand All @@ -322,7 +315,7 @@ func (api *PrivateDebugAPIImpl) AccountAt(ctx context.Context, blockHash common.
return nil, errors.New("block hash is not canonical")
}

minTxNum, err := txNumsReader.Min(tx, *number)
minTxNum, err := api._txNumReader.Min(tx, *number)
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -506,8 +499,6 @@ func (api *PrivateDebugAPIImpl) GetRawTransaction(ctx context.Context, txnHash c
return nil, err
}

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

// Private API returns 0 if transaction is not found.
isBorStateSyncTx := blockNum == 0 && chainConfig.Bor != nil
if isBorStateSyncTx {
Expand All @@ -526,7 +517,7 @@ func (api *PrivateDebugAPIImpl) GetRawTransaction(ctx context.Context, txnHash c
return nil, nil
}

txNumMin, err := txNumsReader.Min(tx, blockNum)
txNumMin, err := api._txNumReader.Min(tx, blockNum)
if err != nil {
return nil, err
}
Expand Down
7 changes: 2 additions & 5 deletions rpc/jsonrpc/erigon_block.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,15 +28,13 @@ import (
"github.com/erigontech/erigon-lib/common/hexutil"
"github.com/erigontech/erigon-lib/kv"
"github.com/erigontech/erigon-lib/kv/order"
"github.com/erigontech/erigon-lib/kv/rawdbv3"
"github.com/erigontech/erigon-lib/types/accounts"
"github.com/erigontech/erigon/core/rawdb"
"github.com/erigontech/erigon/core/types"
"github.com/erigontech/erigon/rpc"
"github.com/erigontech/erigon/rpc/ethapi"
"github.com/erigontech/erigon/rpc/rpchelper"
"github.com/erigontech/erigon/turbo/services"
"github.com/erigontech/erigon/turbo/snapshotsync/freezeblocks"
)

// GetHeaderByNumber implements erigon_getHeaderByNumber. Returns a block's header given a block number ignoring the block's transaction and uncle list (may be faster).
Expand Down Expand Up @@ -218,7 +216,6 @@ func (api *ErigonImpl) GetBalanceChangesInBlock(ctx context.Context, blockNrOrHa
}
defer tx.Rollback()

txNumsReader := rawdbv3.TxNums.WithCustomReadTxNumFunc(freezeblocks.ReadTxNumFuncFromBlockReader(ctx, api._blockReader))
balancesMapping := make(map[common.Address]*hexutil.Big)
latestState, err := rpchelper.CreateStateReader(ctx, tx, api._blockReader, blockNrOrHash, 0, api.filters, api.stateCache, "")
if err != nil {
Expand All @@ -230,11 +227,11 @@ func (api *ErigonImpl) GetBalanceChangesInBlock(ctx context.Context, blockNrOrHa
return nil, err
}

minTxNum, err := txNumsReader.Min(tx, blockNumber)
minTxNum, err := api._txNumReader.Min(tx, blockNumber)
if err != nil {
return nil, err
}
maxTxNum, err := txNumsReader.Max(tx, blockNumber)
maxTxNum, err := api._txNumReader.Max(tx, blockNumber)
if err != nil {
return nil, err
}
Expand Down
6 changes: 2 additions & 4 deletions rpc/jsonrpc/erigon_receipts.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,6 @@ import (
bortypes "github.com/erigontech/erigon/polygon/bor/types"
"github.com/erigontech/erigon/rpc"
"github.com/erigontech/erigon/rpc/rpchelper"
"github.com/erigontech/erigon/turbo/snapshotsync/freezeblocks"
)

// GetLogsByHash implements erigon_getLogsByHash. Returns an array of arrays of logs generated by the transactions in the block given by the block's hash.
Expand Down Expand Up @@ -192,8 +191,7 @@ func (api *ErigonImpl) GetLatestLogs(ctx context.Context, crit filters.FilterCri
exec := exec3.NewTraceWorker(tx, chainConfig, api.engine(), api._blockReader, nil)
defer exec.Close()

txNumsReader := rawdbv3.TxNums.WithCustomReadTxNumFunc(freezeblocks.ReadTxNumFuncFromBlockReader(ctx, api._blockReader))
txNumbers, err := applyFiltersV3(txNumsReader, tx, begin, end, crit)
txNumbers, err := applyFiltersV3(api._txNumReader, tx, begin, end, crit)
if err != nil {
return erigonLogs, err
}
Expand All @@ -211,7 +209,7 @@ func (api *ErigonImpl) GetLatestLogs(ctx context.Context, crit filters.FilterCri

// latest logs that match the filter crit
it := rawdbv3.TxNums2BlockNums(tx,
txNumsReader,
api._txNumReader,
txNumbers, order.Asc)
defer it.Close()

Expand Down
4 changes: 4 additions & 0 deletions rpc/jsonrpc/eth_api.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ import (
"github.com/erigontech/erigon-lib/kv"
"github.com/erigontech/erigon-lib/kv/kvcache"
"github.com/erigontech/erigon-lib/kv/prune"
"github.com/erigontech/erigon-lib/kv/rawdbv3"
"github.com/erigontech/erigon-lib/log/v3"
"github.com/erigontech/erigon-lib/types/accounts"
"github.com/erigontech/erigon/core"
Expand All @@ -52,6 +53,7 @@ import (
"github.com/erigontech/erigon/rpc/jsonrpc/receipts"
"github.com/erigontech/erigon/rpc/rpchelper"
"github.com/erigontech/erigon/turbo/services"
"github.com/erigontech/erigon/turbo/snapshotsync/freezeblocks"
)

// EthAPI is a collection of functions that are exposed in the
Expand Down Expand Up @@ -134,6 +136,7 @@ type BaseAPI struct {
_pruneMode atomic.Pointer[prune.Mode]

_blockReader services.FullBlockReader
_txNumReader rawdbv3.TxNumsReader
_txnReader services.TxnReader
_engine consensus.EngineReader

Expand Down Expand Up @@ -165,6 +168,7 @@ func NewBaseApi(f *rpchelper.Filters, stateCache kvcache.Cache, blockReader serv
blocksLRU: blocksLRU,
_blockReader: blockReader,
_txnReader: blockReader,
_txNumReader: rawdbv3.TxNums.WithCustomReadTxNumFunc(freezeblocks.ReadTxNumFuncFromBlockReader(context.Background(), blockReader)),
evmCallTimeout: evmCallTimeout,
_engine: engine,
receiptsGenerator: receipts.NewGenerator(blockReader, engine),
Expand Down
5 changes: 1 addition & 4 deletions rpc/jsonrpc/eth_block.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,6 @@ import (
"github.com/erigontech/erigon-lib/common/math"
"github.com/erigontech/erigon-lib/crypto/cryptopool"
"github.com/erigontech/erigon-lib/kv"
"github.com/erigontech/erigon-lib/kv/rawdbv3"
"github.com/erigontech/erigon-lib/log/v3"
"github.com/erigontech/erigon/core"
"github.com/erigontech/erigon/core/rawdb"
Expand All @@ -38,7 +37,6 @@ import (
"github.com/erigontech/erigon/rpc"
"github.com/erigontech/erigon/rpc/ethapi"
"github.com/erigontech/erigon/rpc/rpchelper"
"github.com/erigontech/erigon/turbo/snapshotsync/freezeblocks"
"github.com/erigontech/erigon/turbo/transactions"
)

Expand Down Expand Up @@ -90,7 +88,6 @@ func (api *APIImpl) CallBundle(ctx context.Context, txHashes []common.Hash, stat
}
defer func(start time.Time) { log.Trace("Executing EVM call finished", "runtime", time.Since(start)) }(time.Now())

txNumsReader := rawdbv3.TxNums.WithCustomReadTxNumFunc(freezeblocks.ReadTxNumFuncFromBlockReader(ctx, api._blockReader))
stateBlockNumber, hash, latest, err := rpchelper.GetBlockNumber(ctx, stateBlockNumberOrHash, tx, api._blockReader, api.filters)
if err != nil {
return nil, err
Expand All @@ -103,7 +100,7 @@ func (api *APIImpl) CallBundle(ctx context.Context, txHashes []common.Hash, stat
}
stateReader = rpchelper.CreateLatestCachedStateReader(cacheView, tx)
} else {
stateReader, err = rpchelper.CreateHistoryStateReader(tx, txNumsReader, stateBlockNumber+1, 0, chainConfig.ChainName)
stateReader, err = rpchelper.CreateHistoryStateReader(tx, api._txNumReader, stateBlockNumber+1, 0, chainConfig.ChainName)
if err != nil {
return nil, err
}
Expand Down
6 changes: 2 additions & 4 deletions rpc/jsonrpc/eth_call.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -177,8 +177,7 @@ func (api *APIImpl) EstimateGas(ctx context.Context, argsOrNil *ethapi2.CallArgs
}
header := block.HeaderNoCopy()

txNumsReader := rawdbv3.TxNums.WithCustomReadTxNumFunc(freezeblocks.ReadTxNumFuncFromBlockReader(ctx, api._blockReader))
stateReader, err := rpchelper.CreateStateReaderFromBlockNumber(ctx, dbtx, txNumsReader, blockNum, isLatest, 0, api.stateCache, chainConfig.ChainName)
stateReader, err := rpchelper.CreateStateReaderFromBlockNumber(ctx, dbtx, api._txNumReader, blockNum, isLatest, 0, api.stateCache, chainConfig.ChainName)
if err != nil {
return 0, err
}
Expand Down Expand Up @@ -745,7 +744,6 @@ func (api *APIImpl) CreateAccessList(ctx context.Context, args ethapi2.CallArgs,
return nil, nil
}
var stateReader state.StateReader
txNumsReader := rawdbv3.TxNums.WithCustomReadTxNumFunc(freezeblocks.ReadTxNumFuncFromBlockReader(ctx, api._blockReader))

if latest {
cacheView, err := api.stateCache.View(ctx, tx)
Expand All @@ -754,7 +752,7 @@ func (api *APIImpl) CreateAccessList(ctx context.Context, args ethapi2.CallArgs,
}
stateReader = rpchelper.CreateLatestCachedStateReader(cacheView, tx)
} else {
stateReader, err = rpchelper.CreateHistoryStateReader(tx, txNumsReader, blockNumber+1, 0, chainConfig.ChainName)
stateReader, err = rpchelper.CreateHistoryStateReader(tx, api._txNumReader, blockNumber+1, 0, chainConfig.ChainName)
if err != nil {
return nil, err
}
Expand Down
15 changes: 5 additions & 10 deletions rpc/jsonrpc/eth_receipts.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,6 @@ import (
bortypes "github.com/erigontech/erigon/polygon/bor/types"
"github.com/erigontech/erigon/rpc"
"github.com/erigontech/erigon/rpc/rpchelper"
"github.com/erigontech/erigon/turbo/snapshotsync/freezeblocks"
)

// getReceipts - checking in-mem cache, or else fallback to db, or else fallback to re-exec of block to re-gen receipts
Expand All @@ -56,8 +55,7 @@ func (api *BaseAPI) getReceipt(ctx context.Context, cc *chain.Config, tx kv.Temp
}

func (api *BaseAPI) getReceiptsGasUsed(ctx context.Context, tx kv.TemporalTx, block *types.Block) (types.Receipts, error) {
txNumsReader := rawdbv3.TxNums.WithCustomReadTxNumFunc(freezeblocks.ReadTxNumFuncFromBlockReader(ctx, api._blockReader))
return api.receiptsGenerator.GetReceiptsGasUsed(tx, block, txNumsReader)
return api.receiptsGenerator.GetReceiptsGasUsed(tx, block, api._txNumReader)
}

func (api *BaseAPI) getCachedReceipt(ctx context.Context, hash common.Hash) (*types.Receipt, bool) {
Expand Down Expand Up @@ -222,14 +220,13 @@ func (api *BaseAPI) getLogsV3(ctx context.Context, tx kv.TemporalTx, begin, end
var blockHash common.Hash
var header *types.Header

txNumsReader := rawdbv3.TxNums.WithCustomReadTxNumFunc(freezeblocks.ReadTxNumFuncFromBlockReader(ctx, api._blockReader))
txNumbers, err := applyFiltersV3(txNumsReader, tx, begin, end, crit)
txNumbers, err := applyFiltersV3(api._txNumReader, tx, begin, end, crit)
if err != nil {
return logs, err
}

it := rawdbv3.TxNums2BlockNums(tx,
txNumsReader,
api._txNumReader,
txNumbers, order.Asc)
defer it.Close()
for it.HasNext() {
Expand Down Expand Up @@ -258,7 +255,7 @@ func (api *BaseAPI) getLogsV3(ctx context.Context, tx kv.TemporalTx, begin, end
continue
}

borLogs, err := api.borReceiptGenerator.GenerateBorLogs(ctx, events, txNumsReader, tx, header, chainConfig, txIndex, len(logs))
borLogs, err := api.borReceiptGenerator.GenerateBorLogs(ctx, events, api._txNumReader, tx, header, chainConfig, txIndex, len(logs))
if err != nil {
return logs, err
}
Expand Down Expand Up @@ -417,9 +414,7 @@ func (api *APIImpl) GetTransactionReceipt(ctx context.Context, txnHash common.Ha
return nil, nil
}

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

txNumMin, err := txNumsReader.Min(tx, blockNum)
txNumMin, err := api._txNumReader.Min(tx, blockNum)
if err != nil {
return nil, err
}
Expand Down
6 changes: 1 addition & 5 deletions rpc/jsonrpc/eth_txs.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,14 +27,12 @@ import (
"github.com/erigontech/erigon-lib/gointerfaces"
txpool "github.com/erigontech/erigon-lib/gointerfaces/txpoolproto"
types "github.com/erigontech/erigon-lib/gointerfaces/typesproto"
"github.com/erigontech/erigon-lib/kv/rawdbv3"
"github.com/erigontech/erigon/core/rawdb"
types2 "github.com/erigontech/erigon/core/types"
bortypes "github.com/erigontech/erigon/polygon/bor/types"
"github.com/erigontech/erigon/rpc"
"github.com/erigontech/erigon/rpc/ethapi"
"github.com/erigontech/erigon/rpc/rpchelper"
"github.com/erigontech/erigon/turbo/snapshotsync/freezeblocks"
)

// GetTransactionByHash implements eth_getTransactionByHash. Returns information about a transaction given the transaction's hash.
Expand All @@ -55,8 +53,6 @@ func (api *APIImpl) GetTransactionByHash(ctx context.Context, txnHash common.Has
return nil, err
}

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

// Private API returns 0 if transaction is not found.
isBorStateSyncTx := blockNum == 0 && chainConfig.Bor != nil
if isBorStateSyncTx {
Expand All @@ -71,7 +67,7 @@ func (api *APIImpl) GetTransactionByHash(ctx context.Context, txnHash common.Has
}
}
if ok {
txNumMin, err := txNumsReader.Min(tx, blockNum)
txNumMin, err := api._txNumReader.Min(tx, blockNum)
if err != nil {
return nil, err
}
Expand Down
Loading
Loading