From a327ffe9b35289719ac3c484b7332584985b598a Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Mon, 14 Jul 2025 14:07:43 +0200 Subject: [PATCH 1/8] params: EIP-7892 - Blob Parameter Only Hardforks (#32193) This is a resubmit of https://github.com/ethereum/go-ethereum/pull/31820 against the `master` branch. --------- Co-authored-by: Marius van der Wijden Co-authored-by: Gary Rong --- consensus/misc/eip4844/eip4844.go | 72 ++++++++++++++++------------- params/config.go | 75 +++++++++++++++++++++++++++++++ 2 files changed, 115 insertions(+), 32 deletions(-) diff --git a/consensus/misc/eip4844/eip4844.go b/consensus/misc/eip4844/eip4844.go index 049043e3ce..fc143027dd 100644 --- a/consensus/misc/eip4844/eip4844.go +++ b/consensus/misc/eip4844/eip4844.go @@ -23,7 +23,6 @@ import ( "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/params" - "github.com/ethereum/go-ethereum/params/forks" ) var ( @@ -100,42 +99,53 @@ func CalcExcessBlobGas(config *params.ChainConfig, parent *types.Header, headTim // CalcBlobFee calculates the blobfee from the header's excess blob gas field. func CalcBlobFee(config *params.ChainConfig, header *types.Header) *big.Int { - var frac uint64 - switch config.LatestFork(header.Time) { - case forks.Osaka: - frac = config.BlobScheduleConfig.Osaka.UpdateFraction - case forks.Prague: - frac = config.BlobScheduleConfig.Prague.UpdateFraction - case forks.Cancun: - frac = config.BlobScheduleConfig.Cancun.UpdateFraction - default: + blobConfig := latestBlobConfig(config, header.Time) + if blobConfig == nil { panic("calculating blob fee on unsupported fork") } - return fakeExponential(minBlobGasPrice, new(big.Int).SetUint64(*header.ExcessBlobGas), new(big.Int).SetUint64(frac)) + return fakeExponential(minBlobGasPrice, new(big.Int).SetUint64(*header.ExcessBlobGas), new(big.Int).SetUint64(blobConfig.UpdateFraction)) } // MaxBlobsPerBlock returns the max blobs per block for a block at the given timestamp. func MaxBlobsPerBlock(cfg *params.ChainConfig, time uint64) int { - if cfg.BlobScheduleConfig == nil { + blobConfig := latestBlobConfig(cfg, time) + if blobConfig == nil { return 0 } + return blobConfig.Max +} + +func latestBlobConfig(cfg *params.ChainConfig, time uint64) *params.BlobConfig { + if cfg.BlobScheduleConfig == nil { + return nil + } var ( london = cfg.LondonBlock s = cfg.BlobScheduleConfig ) switch { + case cfg.IsBPO5(london, time) && s.BPO5 != nil: + return s.BPO5 + case cfg.IsBPO4(london, time) && s.BPO4 != nil: + return s.BPO4 + case cfg.IsBPO3(london, time) && s.BPO3 != nil: + return s.BPO3 + case cfg.IsBPO2(london, time) && s.BPO2 != nil: + return s.BPO2 + case cfg.IsBPO1(london, time) && s.BPO1 != nil: + return s.BPO1 case cfg.IsOsaka(london, time) && s.Osaka != nil: - return s.Osaka.Max + return s.Osaka case cfg.IsPrague(london, time) && s.Prague != nil: - return s.Prague.Max + return s.Prague case cfg.IsCancun(london, time) && s.Cancun != nil: - return s.Cancun.Max + return s.Cancun default: - return 0 + return nil } } -// MaxBlobsPerBlock returns the maximum blob gas that can be spent in a block at the given timestamp. +// MaxBlobGasPerBlock returns the maximum blob gas that can be spent in a block at the given timestamp. func MaxBlobGasPerBlock(cfg *params.ChainConfig, time uint64) uint64 { return uint64(MaxBlobsPerBlock(cfg, time)) * params.BlobTxBlobGasPerBlob } @@ -148,6 +158,16 @@ func LatestMaxBlobsPerBlock(cfg *params.ChainConfig) int { return 0 } switch { + case s.BPO5 != nil: + return s.BPO5.Max + case s.BPO4 != nil: + return s.BPO4.Max + case s.BPO3 != nil: + return s.BPO3.Max + case s.BPO2 != nil: + return s.BPO2.Max + case s.BPO1 != nil: + return s.BPO1.Max case s.Osaka != nil: return s.Osaka.Max case s.Prague != nil: @@ -161,23 +181,11 @@ func LatestMaxBlobsPerBlock(cfg *params.ChainConfig) int { // targetBlobsPerBlock returns the target number of blobs in a block at the given timestamp. func targetBlobsPerBlock(cfg *params.ChainConfig, time uint64) int { - if cfg.BlobScheduleConfig == nil { - return 0 - } - var ( - london = cfg.LondonBlock - s = cfg.BlobScheduleConfig - ) - switch { - case cfg.IsOsaka(london, time) && s.Osaka != nil: - return s.Osaka.Target - case cfg.IsPrague(london, time) && s.Prague != nil: - return s.Prague.Target - case cfg.IsCancun(london, time) && s.Cancun != nil: - return s.Cancun.Target - default: + blobConfig := latestBlobConfig(cfg, time) + if blobConfig == nil { return 0 } + return blobConfig.Target } // fakeExponential approximates factor * e ** (numerator / denominator) using diff --git a/params/config.go b/params/config.go index 58a0550303..85619bbe22 100644 --- a/params/config.go +++ b/params/config.go @@ -411,6 +411,11 @@ type ChainConfig struct { PragueTime *uint64 `json:"pragueTime,omitempty"` // Prague switch time (nil = no fork, 0 = already on prague) OsakaTime *uint64 `json:"osakaTime,omitempty"` // Osaka switch time (nil = no fork, 0 = already on osaka) VerkleTime *uint64 `json:"verkleTime,omitempty"` // Verkle switch time (nil = no fork, 0 = already on verkle) + BPO1Time *uint64 `json:"bpo1Time,omitempty"` // BPO1 switch time (nil = no fork, 0 = already on bpo1) + BPO2Time *uint64 `json:"bpo2Time,omitempty"` // BPO2 switch time (nil = no fork, 0 = already on bpo2) + BPO3Time *uint64 `json:"bpo3Time,omitempty"` // BPO3 switch time (nil = no fork, 0 = already on bpo3) + BPO4Time *uint64 `json:"bpo4Time,omitempty"` // BPO4 switch time (nil = no fork, 0 = already on bpo4) + BPO5Time *uint64 `json:"bpo5Time,omitempty"` // BPO5 switch time (nil = no fork, 0 = already on bpo5) // TerminalTotalDifficulty is the amount of total difficulty reached by // the network that triggers the consensus upgrade. @@ -531,6 +536,21 @@ func (c *ChainConfig) Description() string { if c.VerkleTime != nil { banner += fmt.Sprintf(" - Verkle: @%-10v\n", *c.VerkleTime) } + if c.BPO1Time != nil { + banner += fmt.Sprintf(" - BPO1: @%-10v\n", *c.BPO1Time) + } + if c.BPO2Time != nil { + banner += fmt.Sprintf(" - BPO2: @%-10v\n", *c.BPO2Time) + } + if c.BPO3Time != nil { + banner += fmt.Sprintf(" - BPO3: @%-10v\n", *c.BPO3Time) + } + if c.BPO4Time != nil { + banner += fmt.Sprintf(" - BPO4: @%-10v\n", *c.BPO4Time) + } + if c.BPO5Time != nil { + banner += fmt.Sprintf(" - BPO5: @%-10v\n", *c.BPO5Time) + } return banner } @@ -547,6 +567,11 @@ type BlobScheduleConfig struct { Prague *BlobConfig `json:"prague,omitempty"` Osaka *BlobConfig `json:"osaka,omitempty"` Verkle *BlobConfig `json:"verkle,omitempty"` + BPO1 *BlobConfig `json:"bpo1,omitempty"` + BPO2 *BlobConfig `json:"bpo2,omitempty"` + BPO3 *BlobConfig `json:"bpo3,omitempty"` + BPO4 *BlobConfig `json:"bpo4,omitempty"` + BPO5 *BlobConfig `json:"bpo5,omitempty"` } // IsHomestead returns whether num is either equal to the homestead block or greater. @@ -654,6 +679,31 @@ func (c *ChainConfig) IsVerkle(num *big.Int, time uint64) bool { return c.IsLondon(num) && isTimestampForked(c.VerkleTime, time) } +// IsBPO1 returns whether time is either equal to the BPO1 fork time or greater. +func (c *ChainConfig) IsBPO1(num *big.Int, time uint64) bool { + return c.IsLondon(num) && isTimestampForked(c.BPO1Time, time) +} + +// IsBPO2 returns whether time is either equal to the BPO2 fork time or greater. +func (c *ChainConfig) IsBPO2(num *big.Int, time uint64) bool { + return c.IsLondon(num) && isTimestampForked(c.BPO2Time, time) +} + +// IsBPO3 returns whether time is either equal to the BPO3 fork time or greater. +func (c *ChainConfig) IsBPO3(num *big.Int, time uint64) bool { + return c.IsLondon(num) && isTimestampForked(c.BPO3Time, time) +} + +// IsBPO4 returns whether time is either equal to the BPO4 fork time or greater. +func (c *ChainConfig) IsBPO4(num *big.Int, time uint64) bool { + return c.IsLondon(num) && isTimestampForked(c.BPO4Time, time) +} + +// IsBPO5 returns whether time is either equal to the BPO5 fork time or greater. +func (c *ChainConfig) IsBPO5(num *big.Int, time uint64) bool { + return c.IsLondon(num) && isTimestampForked(c.BPO5Time, time) +} + // IsVerkleGenesis checks whether the verkle fork is activated at the genesis block. // // Verkle mode is considered enabled if the verkle fork time is configured, @@ -729,6 +779,11 @@ func (c *ChainConfig) CheckConfigForkOrder() error { {name: "pragueTime", timestamp: c.PragueTime, optional: true}, {name: "osakaTime", timestamp: c.OsakaTime, optional: true}, {name: "verkleTime", timestamp: c.VerkleTime, optional: true}, + {name: "bpo1", timestamp: c.BPO1Time, optional: true}, + {name: "bpo2", timestamp: c.BPO2Time, optional: true}, + {name: "bpo3", timestamp: c.BPO3Time, optional: true}, + {name: "bpo4", timestamp: c.BPO4Time, optional: true}, + {name: "bpo5", timestamp: c.BPO5Time, optional: true}, } { if lastFork.name != "" { switch { @@ -778,6 +833,11 @@ func (c *ChainConfig) CheckConfigForkOrder() error { {name: "cancun", timestamp: c.CancunTime, config: bsc.Cancun}, {name: "prague", timestamp: c.PragueTime, config: bsc.Prague}, {name: "osaka", timestamp: c.OsakaTime, config: bsc.Osaka}, + {name: "bpo1", timestamp: c.BPO1Time, config: bsc.BPO1}, + {name: "bpo2", timestamp: c.BPO2Time, config: bsc.BPO2}, + {name: "bpo3", timestamp: c.BPO3Time, config: bsc.BPO3}, + {name: "bpo4", timestamp: c.BPO4Time, config: bsc.BPO4}, + {name: "bpo5", timestamp: c.BPO5Time, config: bsc.BPO5}, } { if cur.config != nil { if err := cur.config.validate(); err != nil { @@ -878,6 +938,21 @@ func (c *ChainConfig) checkCompatible(newcfg *ChainConfig, headNumber *big.Int, if isForkTimestampIncompatible(c.VerkleTime, newcfg.VerkleTime, headTimestamp) { return newTimestampCompatError("Verkle fork timestamp", c.VerkleTime, newcfg.VerkleTime) } + if isForkTimestampIncompatible(c.BPO1Time, newcfg.BPO1Time, headTimestamp) { + return newTimestampCompatError("BPO1 fork timestamp", c.BPO1Time, newcfg.BPO1Time) + } + if isForkTimestampIncompatible(c.BPO2Time, newcfg.BPO2Time, headTimestamp) { + return newTimestampCompatError("BPO2 fork timestamp", c.BPO2Time, newcfg.BPO2Time) + } + if isForkTimestampIncompatible(c.BPO3Time, newcfg.BPO3Time, headTimestamp) { + return newTimestampCompatError("BPO3 fork timestamp", c.BPO3Time, newcfg.BPO3Time) + } + if isForkTimestampIncompatible(c.BPO4Time, newcfg.BPO4Time, headTimestamp) { + return newTimestampCompatError("BPO4 fork timestamp", c.BPO4Time, newcfg.BPO4Time) + } + if isForkTimestampIncompatible(c.BPO5Time, newcfg.BPO5Time, headTimestamp) { + return newTimestampCompatError("BPO5 fork timestamp", c.BPO5Time, newcfg.BPO5Time) + } return nil } From e9e12a97d27ccaa3a11b80f03b897a515ed44aa2 Mon Sep 17 00:00:00 2001 From: Bosul Mun Date: Mon, 14 Jul 2025 15:33:24 +0200 Subject: [PATCH 2/8] eth/fetcher: fix announcement drop logic (#32210) This PR fixes an issue in the tx_fetcher DoS prevention logic where the code keeps the overflow amount (`want - maxTxAnnounces`) instead of the allowed amount (`maxTxAnnounces - used`). The specific changes are: - Correct slice indexing in the announcement drop logic - Extend the overflow test case to cover the inversion scenario --- eth/fetcher/tx_fetcher.go | 4 ++-- eth/fetcher/tx_fetcher_test.go | 27 +++++++++++++++++++++++++++ 2 files changed, 29 insertions(+), 2 deletions(-) diff --git a/eth/fetcher/tx_fetcher.go b/eth/fetcher/tx_fetcher.go index 98a1c6e9a6..3e050320e9 100644 --- a/eth/fetcher/tx_fetcher.go +++ b/eth/fetcher/tx_fetcher.go @@ -439,8 +439,8 @@ func (f *TxFetcher) loop() { if want > maxTxAnnounces { txAnnounceDOSMeter.Mark(int64(want - maxTxAnnounces)) - ann.hashes = ann.hashes[:want-maxTxAnnounces] - ann.metas = ann.metas[:want-maxTxAnnounces] + ann.hashes = ann.hashes[:maxTxAnnounces-used] + ann.metas = ann.metas[:maxTxAnnounces-used] } // All is well, schedule the remainder of the transactions var ( diff --git a/eth/fetcher/tx_fetcher_test.go b/eth/fetcher/tx_fetcher_test.go index c4c8cac56e..0f05a1c995 100644 --- a/eth/fetcher/tx_fetcher_test.go +++ b/eth/fetcher/tx_fetcher_test.go @@ -1179,6 +1179,24 @@ func TestTransactionFetcherDoSProtection(t *testing.T) { size: 111, }) } + var ( + hashesC []common.Hash + typesC []byte + sizesC []uint32 + announceC []announce + ) + for i := 0; i < maxTxAnnounces+2; i++ { + hash := common.Hash{0x03, byte(i / 256), byte(i % 256)} + hashesC = append(hashesC, hash) + typesC = append(typesC, types.LegacyTxType) + sizesC = append(sizesC, 111) + + announceC = append(announceC, announce{ + hash: hash, + kind: types.LegacyTxType, + size: 111, + }) + } testTransactionFetcherParallel(t, txFetcherTest{ init: func() *TxFetcher { return NewTxFetcher( @@ -1192,43 +1210,52 @@ func TestTransactionFetcherDoSProtection(t *testing.T) { // Announce half of the transaction and wait for them to be scheduled doTxNotify{peer: "A", hashes: hashesA[:maxTxAnnounces/2], types: typesA[:maxTxAnnounces/2], sizes: sizesA[:maxTxAnnounces/2]}, doTxNotify{peer: "B", hashes: hashesB[:maxTxAnnounces/2-1], types: typesB[:maxTxAnnounces/2-1], sizes: sizesB[:maxTxAnnounces/2-1]}, + doTxNotify{peer: "C", hashes: hashesC[:maxTxAnnounces/2-1], types: typesC[:maxTxAnnounces/2-1], sizes: sizesC[:maxTxAnnounces/2-1]}, doWait{time: txArriveTimeout, step: true}, // Announce the second half and keep them in the wait list doTxNotify{peer: "A", hashes: hashesA[maxTxAnnounces/2 : maxTxAnnounces], types: typesA[maxTxAnnounces/2 : maxTxAnnounces], sizes: sizesA[maxTxAnnounces/2 : maxTxAnnounces]}, doTxNotify{peer: "B", hashes: hashesB[maxTxAnnounces/2-1 : maxTxAnnounces-1], types: typesB[maxTxAnnounces/2-1 : maxTxAnnounces-1], sizes: sizesB[maxTxAnnounces/2-1 : maxTxAnnounces-1]}, + doTxNotify{peer: "C", hashes: hashesC[maxTxAnnounces/2-1 : maxTxAnnounces-1], types: typesC[maxTxAnnounces/2-1 : maxTxAnnounces-1], sizes: sizesC[maxTxAnnounces/2-1 : maxTxAnnounces-1]}, // Ensure the hashes are split half and half isWaiting(map[string][]announce{ "A": announceA[maxTxAnnounces/2 : maxTxAnnounces], "B": announceB[maxTxAnnounces/2-1 : maxTxAnnounces-1], + "C": announceC[maxTxAnnounces/2-1 : maxTxAnnounces-1], }), isScheduled{ tracking: map[string][]announce{ "A": announceA[:maxTxAnnounces/2], "B": announceB[:maxTxAnnounces/2-1], + "C": announceC[:maxTxAnnounces/2-1], }, fetching: map[string][]common.Hash{ "A": hashesA[:maxTxRetrievals], "B": hashesB[:maxTxRetrievals], + "C": hashesC[:maxTxRetrievals], }, }, // Ensure that adding even one more hash results in dropping the hash doTxNotify{peer: "A", hashes: []common.Hash{hashesA[maxTxAnnounces]}, types: []byte{typesA[maxTxAnnounces]}, sizes: []uint32{sizesA[maxTxAnnounces]}}, doTxNotify{peer: "B", hashes: hashesB[maxTxAnnounces-1 : maxTxAnnounces+1], types: typesB[maxTxAnnounces-1 : maxTxAnnounces+1], sizes: sizesB[maxTxAnnounces-1 : maxTxAnnounces+1]}, + doTxNotify{peer: "C", hashes: hashesC[maxTxAnnounces-1 : maxTxAnnounces+2], types: typesC[maxTxAnnounces-1 : maxTxAnnounces+2], sizes: sizesC[maxTxAnnounces-1 : maxTxAnnounces+2]}, isWaiting(map[string][]announce{ "A": announceA[maxTxAnnounces/2 : maxTxAnnounces], "B": announceB[maxTxAnnounces/2-1 : maxTxAnnounces], + "C": announceC[maxTxAnnounces/2-1 : maxTxAnnounces], }), isScheduled{ tracking: map[string][]announce{ "A": announceA[:maxTxAnnounces/2], "B": announceB[:maxTxAnnounces/2-1], + "C": announceC[:maxTxAnnounces/2-1], }, fetching: map[string][]common.Hash{ "A": hashesA[:maxTxRetrievals], "B": hashesB[:maxTxRetrievals], + "C": hashesC[:maxTxRetrievals], }, }, }, From 1a5f399e301baf795d7d4c38d9a53ea5d7940a1b Mon Sep 17 00:00:00 2001 From: maskpp Date: Mon, 14 Jul 2025 22:06:57 +0800 Subject: [PATCH 3/8] miner: set sidecar version when recomputing proofs (#32199) - If the block number is `osaka` fork and needs to recompute some `blob proofs` to `cell proofs`, here also needs to set version to `1`. --- miner/worker.go | 1 + 1 file changed, 1 insertion(+) diff --git a/miner/worker.go b/miner/worker.go index 799107f3bf..21e73058ce 100644 --- a/miner/worker.go +++ b/miner/worker.go @@ -440,6 +440,7 @@ func (miner *Miner) commitTransactions(env *environment, plainTxs, blobTxs *tran } sidecar.Proofs = append(sidecar.Proofs, cellProofs...) } + sidecar.Version = 1 } } } From 5bce990891435ff9881192cf52373bf93317b704 Mon Sep 17 00:00:00 2001 From: FT <140458077+zeevick10@users.noreply.github.com> Date: Tue, 15 Jul 2025 00:08:06 +0200 Subject: [PATCH 4/8] all: fix outdated ethereum wiki json-rpc json-rpc doc links (#32209) Replace outdated wiki reference with ethereum.org documentation links --- internal/ethapi/api.go | 2 +- internal/ethapi/errors.go | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/internal/ethapi/api.go b/internal/ethapi/api.go index 21c35b7665..8857f1c04f 100644 --- a/internal/ethapi/api.go +++ b/internal/ethapi/api.go @@ -1549,7 +1549,7 @@ func (api *TransactionAPI) SendRawTransaction(ctx context.Context, input hexutil // // The account associated with addr must be unlocked. // -// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_sign +// https://ethereum.org/en/developers/docs/apis/json-rpc/#eth_sign func (api *TransactionAPI) Sign(addr common.Address, data hexutil.Bytes) (hexutil.Bytes, error) { // Look up the wallet containing the requested signer account := accounts.Account{Address: addr} diff --git a/internal/ethapi/errors.go b/internal/ethapi/errors.go index ae38061234..154938fa0e 100644 --- a/internal/ethapi/errors.go +++ b/internal/ethapi/errors.go @@ -34,7 +34,7 @@ type revertError struct { } // ErrorCode returns the JSON error code for a revert. -// See: https://github.com/ethereum/wiki/wiki/JSON-RPC-Error-Codes-Improvement-Proposal +// See: https://ethereum.org/en/developers/docs/apis/json-rpc/#error-codes func (e *revertError) ErrorCode() int { return 3 } @@ -71,7 +71,7 @@ func (e *TxIndexingError) Error() string { } // ErrorCode returns the JSON error code for a revert. -// See: https://github.com/ethereum/wiki/wiki/JSON-RPC-Error-Codes-Improvement-Proposal +// See: https://ethereum.org/en/developers/docs/apis/json-rpc/#error-codes func (e *TxIndexingError) ErrorCode() int { return -32000 // to be decided } From fe0ae06c779bc62901dcd9f6fe8b1967409c1bf5 Mon Sep 17 00:00:00 2001 From: maskpp Date: Tue, 15 Jul 2025 09:07:23 +0800 Subject: [PATCH 5/8] core/types: fix CellProofsAt method (#32198) --- core/types/tx_blob.go | 24 +++++++++++++----------- eth/catalyst/api.go | 7 ++++++- 2 files changed, 19 insertions(+), 12 deletions(-) diff --git a/core/types/tx_blob.go b/core/types/tx_blob.go index 93a76a28b7..774d6d14d2 100644 --- a/core/types/tx_blob.go +++ b/core/types/tx_blob.go @@ -22,7 +22,6 @@ import ( "errors" "fmt" "math/big" - "slices" "github.com/ethereum/go-ethereum/common" @@ -75,17 +74,19 @@ func (sc *BlobTxSidecar) BlobHashes() []common.Hash { } // CellProofsAt returns the cell proofs for blob with index idx. -func (sc *BlobTxSidecar) CellProofsAt(idx int) []kzg4844.Proof { - var cellProofs []kzg4844.Proof - for i := range kzg4844.CellProofsPerBlob { - index := idx*kzg4844.CellProofsPerBlob + i - if index > len(sc.Proofs) { - return nil - } - proof := sc.Proofs[index] - cellProofs = append(cellProofs, proof) +// This method is only valid for sidecars with version 1. +func (sc *BlobTxSidecar) CellProofsAt(idx int) ([]kzg4844.Proof, error) { + if sc.Version != 1 { + return nil, fmt.Errorf("cell proof unsupported, version: %d", sc.Version) } - return cellProofs + if idx < 0 || idx >= len(sc.Blobs) { + return nil, fmt.Errorf("cell proof out of bounds, index: %d, blobs: %d", idx, len(sc.Blobs)) + } + index := idx * kzg4844.CellProofsPerBlob + if len(sc.Proofs) < index+kzg4844.CellProofsPerBlob { + return nil, fmt.Errorf("cell proof is corrupted, index: %d, proofs: %d", idx, len(sc.Proofs)) + } + return sc.Proofs[index : index+kzg4844.CellProofsPerBlob], nil } // encodedSize computes the RLP size of the sidecar elements. This does NOT return the @@ -217,6 +218,7 @@ func (tx *BlobTx) copy() TxData { } if tx.Sidecar != nil { cpy.Sidecar = &BlobTxSidecar{ + Version: tx.Sidecar.Version, Blobs: slices.Clone(tx.Sidecar.Blobs), Commitments: slices.Clone(tx.Sidecar.Commitments), Proofs: slices.Clone(tx.Sidecar.Proofs), diff --git a/eth/catalyst/api.go b/eth/catalyst/api.go index 409899d582..c52ffb09ba 100644 --- a/eth/catalyst/api.go +++ b/eth/catalyst/api.go @@ -564,7 +564,12 @@ func (api *ConsensusAPI) GetBlobsV2(hashes []common.Hash) ([]*engine.BlobAndProo blobHashes := sidecar.BlobHashes() for bIdx, hash := range blobHashes { if idxes, ok := index[hash]; ok { - proofs := sidecar.CellProofsAt(bIdx) + proofs, err := sidecar.CellProofsAt(bIdx) + if err != nil { + // TODO @rjl @marius we should return an error + log.Info("Failed to get cell proof", "err", err) + return nil, nil + } var cellProofs []hexutil.Bytes for _, proof := range proofs { cellProofs = append(cellProofs, proof[:]) From 17903fedf0374be940f1e75260a8eacbbef62d0a Mon Sep 17 00:00:00 2001 From: Delweng Date: Tue, 15 Jul 2025 11:45:20 +0800 Subject: [PATCH 6/8] triedb/pathdb: introduce file-based state journal (#32060) Introduce file-based state journal in path database, fixing the Pebble restriction when the journal size exceeds 4GB. --------- Signed-off-by: jsvisa Co-authored-by: Gary Rong --- cmd/utils/flags.go | 6 ++ core/blockchain.go | 10 +-- core/rawdb/accessors_state.go | 8 --- eth/backend.go | 6 +- triedb/pathdb/database.go | 22 ++++++- triedb/pathdb/database_test.go | 52 +++++++++++----- triedb/pathdb/fileutils_unix.go | 57 +++++++++++++++++ triedb/pathdb/fileutils_windows.go | 25 ++++++++ triedb/pathdb/history_reader_test.go | 2 +- triedb/pathdb/journal.go | 93 +++++++++++++++++++++++++--- 10 files changed, 240 insertions(+), 41 deletions(-) create mode 100644 triedb/pathdb/fileutils_unix.go create mode 100644 triedb/pathdb/fileutils_windows.go diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index d3e842149a..b86970651f 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -2198,6 +2198,12 @@ func MakeChain(ctx *cli.Context, stack *node.Node, readonly bool) (*core.BlockCh StateHistory: ctx.Uint64(StateHistoryFlag.Name), // Disable transaction indexing/unindexing. TxLookupLimit: -1, + + // Enables file journaling for the trie database. The journal files will be stored + // within the data directory. The corresponding paths will be either: + // - DATADIR/triedb/merkle.journal + // - DATADIR/triedb/verkle.journal + TrieJournalDirectory: stack.ResolvePath("triedb"), } if options.ArchiveMode && !options.Preimages { options.Preimages = true diff --git a/core/blockchain.go b/core/blockchain.go index 2290b6d3cd..d52990ec5a 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -162,10 +162,11 @@ const ( // BlockChainConfig contains the configuration of the BlockChain object. type BlockChainConfig struct { // Trie database related options - TrieCleanLimit int // Memory allowance (MB) to use for caching trie nodes in memory - TrieDirtyLimit int // Memory limit (MB) at which to start flushing dirty trie nodes to disk - TrieTimeLimit time.Duration // Time limit after which to flush the current in-memory trie to disk - TrieNoAsyncFlush bool // Whether the asynchronous buffer flushing is disallowed + TrieCleanLimit int // Memory allowance (MB) to use for caching trie nodes in memory + TrieDirtyLimit int // Memory limit (MB) at which to start flushing dirty trie nodes to disk + TrieTimeLimit time.Duration // Time limit after which to flush the current in-memory trie to disk + TrieNoAsyncFlush bool // Whether the asynchronous buffer flushing is disallowed + TrieJournalDirectory string // Directory path to the journal used for persisting trie data across node restarts Preimages bool // Whether to store preimage of trie key to the disk StateHistory uint64 // Number of blocks from head whose state histories are reserved. @@ -246,6 +247,7 @@ func (cfg *BlockChainConfig) triedbConfig(isVerkle bool) *triedb.Config { EnableStateIndexing: cfg.ArchiveMode, TrieCleanSize: cfg.TrieCleanLimit * 1024 * 1024, StateCleanSize: cfg.SnapshotLimit * 1024 * 1024, + JournalDirectory: cfg.TrieJournalDirectory, // TODO(rjl493456442): The write buffer represents the memory limit used // for flushing both trie data and state data to disk. The config name diff --git a/core/rawdb/accessors_state.go b/core/rawdb/accessors_state.go index 7d7b37641b..44f041d82e 100644 --- a/core/rawdb/accessors_state.go +++ b/core/rawdb/accessors_state.go @@ -157,14 +157,6 @@ func WriteTrieJournal(db ethdb.KeyValueWriter, journal []byte) { } } -// DeleteTrieJournal deletes the serialized in-memory trie nodes of layers saved at -// the last shutdown. -func DeleteTrieJournal(db ethdb.KeyValueWriter) { - if err := db.Delete(trieJournalKey); err != nil { - log.Crit("Failed to remove tries journal", "err", err) - } -} - // ReadStateHistoryMeta retrieves the metadata corresponding to the specified // state history. Compute the position of state history in freezer by minus // one since the id of first state history starts from one(zero for initial diff --git a/eth/backend.go b/eth/backend.go index 80ffd301ce..7616ec9d31 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -236,9 +236,13 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) { VmConfig: vm.Config{ EnablePreimageRecording: config.EnablePreimageRecording, }, + // Enables file journaling for the trie database. The journal files will be stored + // within the data directory. The corresponding paths will be either: + // - DATADIR/triedb/merkle.journal + // - DATADIR/triedb/verkle.journal + TrieJournalDirectory: stack.ResolvePath("triedb"), } ) - if config.VMTrace != "" { traceConfig := json.RawMessage("{}") if config.VMTraceJsonConfig != "" { diff --git a/triedb/pathdb/database.go b/triedb/pathdb/database.go index 23a7d383b5..e323a7449e 100644 --- a/triedb/pathdb/database.go +++ b/triedb/pathdb/database.go @@ -21,6 +21,7 @@ import ( "errors" "fmt" "io" + "path/filepath" "sync" "time" @@ -120,6 +121,7 @@ type Config struct { StateCleanSize int // Maximum memory allowance (in bytes) for caching clean state data WriteBufferSize int // Maximum memory allowance (in bytes) for write buffer ReadOnly bool // Flag whether the database is opened in read only mode + JournalDirectory string // Absolute path of journal directory (null means the journal data is persisted in key-value store) // Testing configurations SnapshotNoBuild bool // Flag Whether the state generation is allowed @@ -156,6 +158,9 @@ func (c *Config) fields() []interface{} { } else { list = append(list, "history", fmt.Sprintf("last %d blocks", c.StateHistory)) } + if c.JournalDirectory != "" { + list = append(list, "journal-dir", c.JournalDirectory) + } return list } @@ -493,7 +498,6 @@ func (db *Database) Enable(root common.Hash) error { // Drop the stale state journal in persistent database and // reset the persistent state id back to zero. batch := db.diskdb.NewBatch() - rawdb.DeleteTrieJournal(batch) rawdb.DeleteSnapshotRoot(batch) rawdb.WritePersistentStateID(batch, 0) if err := batch.Write(); err != nil { @@ -573,8 +577,6 @@ func (db *Database) Recover(root common.Hash) error { // disk layer won't be accessible from outside. db.tree.init(dl) } - rawdb.DeleteTrieJournal(db.diskdb) - // Explicitly sync the key-value store to ensure all recent writes are // flushed to disk. This step is crucial to prevent a scenario where // recent key-value writes are lost due to an application panic, while @@ -680,6 +682,20 @@ func (db *Database) modifyAllowed() error { return nil } +// journalPath returns the absolute path of journal for persisting state data. +func (db *Database) journalPath() string { + if db.config.JournalDirectory == "" { + return "" + } + var fname string + if db.isVerkle { + fname = fmt.Sprintf("verkle.journal") + } else { + fname = fmt.Sprintf("merkle.journal") + } + return filepath.Join(db.config.JournalDirectory, fname) +} + // AccountHistory inspects the account history within the specified range. // // Start: State ID of the first history object for the query. 0 implies the first diff --git a/triedb/pathdb/database_test.go b/triedb/pathdb/database_test.go index 2982202009..e9a1850ee0 100644 --- a/triedb/pathdb/database_test.go +++ b/triedb/pathdb/database_test.go @@ -21,12 +21,16 @@ import ( "errors" "fmt" "math/rand" + "os" + "path/filepath" + "strconv" "testing" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/internal/testrand" "github.com/ethereum/go-ethereum/rlp" "github.com/ethereum/go-ethereum/trie" @@ -121,7 +125,7 @@ type tester struct { snapStorages map[common.Hash]map[common.Hash]map[common.Hash][]byte // Keyed by the hash of account address and the hash of storage key } -func newTester(t *testing.T, historyLimit uint64, isVerkle bool, layers int, enableIndex bool) *tester { +func newTester(t *testing.T, historyLimit uint64, isVerkle bool, layers int, enableIndex bool, journalDir string) *tester { var ( disk, _ = rawdb.Open(rawdb.NewMemoryDatabase(), rawdb.OpenOptions{Ancient: t.TempDir()}) db = New(disk, &Config{ @@ -131,6 +135,7 @@ func newTester(t *testing.T, historyLimit uint64, isVerkle bool, layers int, ena StateCleanSize: 256 * 1024, WriteBufferSize: 256 * 1024, NoAsyncFlush: true, + JournalDirectory: journalDir, }, isVerkle) obj = &tester{ @@ -466,7 +471,7 @@ func TestDatabaseRollback(t *testing.T) { }() // Verify state histories - tester := newTester(t, 0, false, 32, false) + tester := newTester(t, 0, false, 32, false, "") defer tester.release() if err := tester.verifyHistory(); err != nil { @@ -500,7 +505,7 @@ func TestDatabaseRecoverable(t *testing.T) { }() var ( - tester = newTester(t, 0, false, 12, false) + tester = newTester(t, 0, false, 12, false, "") index = tester.bottomIndex() ) defer tester.release() @@ -544,7 +549,7 @@ func TestDisable(t *testing.T) { maxDiffLayers = 128 }() - tester := newTester(t, 0, false, 32, false) + tester := newTester(t, 0, false, 32, false, "") defer tester.release() stored := crypto.Keccak256Hash(rawdb.ReadAccountTrieNode(tester.db.diskdb, nil)) @@ -586,7 +591,7 @@ func TestCommit(t *testing.T) { maxDiffLayers = 128 }() - tester := newTester(t, 0, false, 12, false) + tester := newTester(t, 0, false, 12, false, "") defer tester.release() if err := tester.db.Commit(tester.lastHash(), false); err != nil { @@ -610,20 +615,25 @@ func TestCommit(t *testing.T) { } func TestJournal(t *testing.T) { + testJournal(t, "") + testJournal(t, filepath.Join(t.TempDir(), strconv.Itoa(rand.Intn(10000)))) +} + +func testJournal(t *testing.T, journalDir string) { // Redefine the diff layer depth allowance for faster testing. maxDiffLayers = 4 defer func() { maxDiffLayers = 128 }() - tester := newTester(t, 0, false, 12, false) + tester := newTester(t, 0, false, 12, false, journalDir) defer tester.release() if err := tester.db.Journal(tester.lastHash()); err != nil { t.Errorf("Failed to journal, err: %v", err) } tester.db.Close() - tester.db = New(tester.db.diskdb, nil, false) + tester.db = New(tester.db.diskdb, tester.db.config, false) // Verify states including disk layer and all diff on top. for i := 0; i < len(tester.roots); i++ { @@ -640,13 +650,30 @@ func TestJournal(t *testing.T) { } func TestCorruptedJournal(t *testing.T) { + testCorruptedJournal(t, "", func(db ethdb.Database) { + // Mutate the journal in disk, it should be regarded as invalid + blob := rawdb.ReadTrieJournal(db) + blob[0] = 0xa + rawdb.WriteTrieJournal(db, blob) + }) + + directory := filepath.Join(t.TempDir(), strconv.Itoa(rand.Intn(10000))) + testCorruptedJournal(t, directory, func(_ ethdb.Database) { + f, _ := os.OpenFile(filepath.Join(directory, "merkle.journal"), os.O_WRONLY, 0644) + f.WriteAt([]byte{0xa}, 0) + f.Sync() + f.Close() + }) +} + +func testCorruptedJournal(t *testing.T, journalDir string, modifyFn func(database ethdb.Database)) { // Redefine the diff layer depth allowance for faster testing. maxDiffLayers = 4 defer func() { maxDiffLayers = 128 }() - tester := newTester(t, 0, false, 12, false) + tester := newTester(t, 0, false, 12, false, journalDir) defer tester.release() if err := tester.db.Journal(tester.lastHash()); err != nil { @@ -655,13 +682,10 @@ func TestCorruptedJournal(t *testing.T) { tester.db.Close() root := crypto.Keccak256Hash(rawdb.ReadAccountTrieNode(tester.db.diskdb, nil)) - // Mutate the journal in disk, it should be regarded as invalid - blob := rawdb.ReadTrieJournal(tester.db.diskdb) - blob[0] = 0xa - rawdb.WriteTrieJournal(tester.db.diskdb, blob) + modifyFn(tester.db.diskdb) // Verify states, all not-yet-written states should be discarded - tester.db = New(tester.db.diskdb, nil, false) + tester.db = New(tester.db.diskdb, tester.db.config, false) for i := 0; i < len(tester.roots); i++ { if tester.roots[i] == root { if err := tester.verifyState(root); err != nil { @@ -694,7 +718,7 @@ func TestTailTruncateHistory(t *testing.T) { maxDiffLayers = 128 }() - tester := newTester(t, 10, false, 12, false) + tester := newTester(t, 10, false, 12, false, "") defer tester.release() tester.db.Close() diff --git a/triedb/pathdb/fileutils_unix.go b/triedb/pathdb/fileutils_unix.go new file mode 100644 index 0000000000..fde0bf50fa --- /dev/null +++ b/triedb/pathdb/fileutils_unix.go @@ -0,0 +1,57 @@ +// Copyright 2025 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library 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. +// +// The go-ethereum library 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 the go-ethereum library. If not, see . + +//go:build !windows +// +build !windows + +package pathdb + +import ( + "errors" + "os" + "syscall" +) + +func isErrInvalid(err error) bool { + if errors.Is(err, os.ErrInvalid) { + return true + } + // Go >= 1.8 returns *os.PathError instead + if patherr, ok := err.(*os.PathError); ok && patherr.Err == syscall.EINVAL { + return true + } + return false +} + +func syncDir(name string) error { + // As per fsync manpage, Linux seems to expect fsync on directory, however + // some system don't support this, so we will ignore syscall.EINVAL. + // + // From fsync(2): + // Calling fsync() does not necessarily ensure that the entry in the + // directory containing the file has also reached disk. For that an + // explicit fsync() on a file descriptor for the directory is also needed. + f, err := os.Open(name) + if err != nil { + return err + } + defer f.Close() + + if err := f.Sync(); err != nil && !isErrInvalid(err) { + return err + } + return nil +} diff --git a/triedb/pathdb/fileutils_windows.go b/triedb/pathdb/fileutils_windows.go new file mode 100644 index 0000000000..e4c644d757 --- /dev/null +++ b/triedb/pathdb/fileutils_windows.go @@ -0,0 +1,25 @@ +// Copyright 2025 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library 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. +// +// The go-ethereum library 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 the go-ethereum library. If not, see . + +//go:build windows +// +build windows + +package pathdb + +func syncDir(name string) error { + // On Windows, fsync on directories is not supported + return nil +} diff --git a/triedb/pathdb/history_reader_test.go b/triedb/pathdb/history_reader_test.go index 4356490f23..4eb93fb9c9 100644 --- a/triedb/pathdb/history_reader_test.go +++ b/triedb/pathdb/history_reader_test.go @@ -126,7 +126,7 @@ func testHistoryReader(t *testing.T, historyLimit uint64) { }() //log.SetDefault(log.NewLogger(log.NewTerminalHandlerWithLevel(os.Stderr, log.LevelDebug, true))) - env := newTester(t, historyLimit, false, 64, true) + env := newTester(t, historyLimit, false, 64, true, "") defer env.release() waitIndexing(env.db) diff --git a/triedb/pathdb/journal.go b/triedb/pathdb/journal.go index e88b3e062f..4639932763 100644 --- a/triedb/pathdb/journal.go +++ b/triedb/pathdb/journal.go @@ -21,6 +21,7 @@ import ( "errors" "fmt" "io" + "os" "time" "github.com/ethereum/go-ethereum/common" @@ -31,6 +32,8 @@ import ( "github.com/ethereum/go-ethereum/rlp" ) +const tempJournalSuffix = ".tmp" + var ( errMissJournal = errors.New("journal not found") errMissVersion = errors.New("version not found") @@ -51,11 +54,25 @@ const journalVersion uint64 = 3 // loadJournal tries to parse the layer journal from the disk. func (db *Database) loadJournal(diskRoot common.Hash) (layer, error) { - journal := rawdb.ReadTrieJournal(db.diskdb) - if len(journal) == 0 { - return nil, errMissJournal + var reader io.Reader + if path := db.journalPath(); path != "" && common.FileExist(path) { + // If a journal file is specified, read it from there + log.Info("Load database journal from file", "path", path) + f, err := os.OpenFile(path, os.O_RDONLY, 0644) + if err != nil { + return nil, fmt.Errorf("failed to read journal file %s: %w", path, err) + } + defer f.Close() + reader = f + } else { + log.Info("Load database journal from disk") + journal := rawdb.ReadTrieJournal(db.diskdb) + if len(journal) == 0 { + return nil, errMissJournal + } + reader = bytes.NewReader(journal) } - r := rlp.NewStream(bytes.NewReader(journal), 0) + r := rlp.NewStream(reader, 0) // Firstly, resolve the first element as the journal version version, err := r.Uint64() @@ -297,9 +314,9 @@ func (db *Database) Journal(root common.Hash) error { } disk := db.tree.bottom() if l, ok := l.(*diffLayer); ok { - log.Info("Persisting dirty state to disk", "head", l.block, "root", root, "layers", l.id-disk.id+disk.buffer.layers) + log.Info("Persisting dirty state", "head", l.block, "root", root, "layers", l.id-disk.id+disk.buffer.layers) } else { // disk layer only on noop runs (likely) or deep reorgs (unlikely) - log.Info("Persisting dirty state to disk", "root", root, "layers", disk.buffer.layers) + log.Info("Persisting dirty state", "root", root, "layers", disk.buffer.layers) } // Block until the background flushing is finished and terminate // the potential active state generator. @@ -316,8 +333,37 @@ func (db *Database) Journal(root common.Hash) error { if db.readOnly { return errDatabaseReadOnly } + + // Store the journal into the database and return + var ( + file *os.File + journal io.Writer + journalPath = db.journalPath() + ) + if journalPath != "" { + // Write into a temp file first + err := os.MkdirAll(db.config.JournalDirectory, 0755) + if err != nil { + return err + } + tmp := journalPath + tempJournalSuffix + file, err = os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644) + if err != nil { + return fmt.Errorf("failed to open journal file %s: %w", tmp, err) + } + defer func() { + if file != nil { + file.Close() + os.Remove(tmp) // Clean up temp file if we didn't successfully rename it + log.Warn("Removed leftover temporary journal file", "path", tmp) + } + }() + journal = file + } else { + journal = new(bytes.Buffer) + } + // Firstly write out the metadata of journal - journal := new(bytes.Buffer) if err := rlp.Encode(journal, journalVersion); err != nil { return err } @@ -334,11 +380,38 @@ func (db *Database) Journal(root common.Hash) error { if err := l.journal(journal); err != nil { return err } - // Store the journal into the database and return - rawdb.WriteTrieJournal(db.diskdb, journal.Bytes()) + // Store the journal into the database and return + if file == nil { + data := journal.(*bytes.Buffer) + size := data.Len() + rawdb.WriteTrieJournal(db.diskdb, data.Bytes()) + log.Info("Persisted dirty state to disk", "size", common.StorageSize(size), "elapsed", common.PrettyDuration(time.Since(start))) + } else { + stat, err := file.Stat() + if err != nil { + return err + } + size := int(stat.Size()) + + // Close the temporary file and atomically rename it + if err := file.Sync(); err != nil { + return fmt.Errorf("failed to fsync the journal, %v", err) + } + if err := file.Close(); err != nil { + return fmt.Errorf("failed to close the journal: %v", err) + } + // Replace the live journal with the newly generated one + if err := os.Rename(journalPath+tempJournalSuffix, journalPath); err != nil { + return fmt.Errorf("failed to rename the journal: %v", err) + } + if err := syncDir(db.config.JournalDirectory); err != nil { + return fmt.Errorf("failed to fsync the dir: %v", err) + } + file = nil + log.Info("Persisted dirty state to file", "path", journalPath, "size", common.StorageSize(size), "elapsed", common.PrettyDuration(time.Since(start))) + } // Set the db in read only mode to reject all following mutations db.readOnly = true - log.Info("Persisted dirty state to disk", "size", common.StorageSize(journal.Len()), "elapsed", common.PrettyDuration(time.Since(start))) return nil } From 7364e63ef90f0dc4589d96a38e450dcfe7183caa Mon Sep 17 00:00:00 2001 From: rjl493456442 Date: Tue, 15 Jul 2025 13:50:52 +0800 Subject: [PATCH 7/8] core/rawdb: change the mechanism to schedule freezer sync (#32135) This pull request slightly improves the freezer fsync mechanism by scheduling the Sync operation based on the number of uncommitted items and original time interval. Originally, freezer.Sync was triggered every 30 seconds, which worked well during active chain synchronization. However, once the initial state sync is complete, the fixed interval causes Sync to be scheduled too frequently. To address this, the scheduling logic has been improved to consider both the time interval and the number of uncommitted items. This additional condition helps avoid unnecessary Sync operations when the chain is idle. --- core/rawdb/freezer_batch.go | 18 ++++++++++++++---- core/rawdb/freezer_table.go | 5 +++-- 2 files changed, 17 insertions(+), 6 deletions(-) diff --git a/core/rawdb/freezer_batch.go b/core/rawdb/freezer_batch.go index 99b63df4dc..7e46e49f43 100644 --- a/core/rawdb/freezer_batch.go +++ b/core/rawdb/freezer_batch.go @@ -25,9 +25,16 @@ import ( "github.com/golang/snappy" ) -// This is the maximum amount of data that will be buffered in memory -// for a single freezer table batch. -const freezerBatchBufferLimit = 2 * 1024 * 1024 +const ( + // This is the maximum amount of data that will be buffered in memory + // for a single freezer table batch. + freezerBatchBufferLimit = 2 * 1024 * 1024 + + // freezerTableFlushThreshold defines the threshold for triggering a freezer + // table sync operation. If the number of accumulated uncommitted items exceeds + // this value, a sync will be scheduled. + freezerTableFlushThreshold = 512 +) // freezerBatch is a write operation of multiple items on a freezer. type freezerBatch struct { @@ -201,6 +208,7 @@ func (batch *freezerTableBatch) commit() error { // Update headBytes of table. batch.t.headBytes += dataSize + items := batch.curItem - batch.t.items.Load() batch.t.items.Store(batch.curItem) // Update metrics. @@ -208,7 +216,9 @@ func (batch *freezerTableBatch) commit() error { batch.t.writeMeter.Mark(dataSize + indexSize) // Periodically sync the table, todo (rjl493456442) make it configurable? - if time.Since(batch.t.lastSync) > 30*time.Second { + batch.t.uncommitted += items + if batch.t.uncommitted > freezerTableFlushThreshold && time.Since(batch.t.lastSync) > 30*time.Second { + batch.t.uncommitted = 0 batch.t.lastSync = time.Now() return batch.t.Sync() } diff --git a/core/rawdb/freezer_table.go b/core/rawdb/freezer_table.go index 3900b5d558..19c40cc16e 100644 --- a/core/rawdb/freezer_table.go +++ b/core/rawdb/freezer_table.go @@ -112,8 +112,9 @@ type freezerTable struct { headId uint32 // number of the currently active head file tailId uint32 // number of the earliest file - metadata *freezerTableMeta // metadata of the table - lastSync time.Time // Timestamp when the last sync was performed + metadata *freezerTableMeta // metadata of the table + uncommitted uint64 // Count of items written without flushing to file + lastSync time.Time // Timestamp when the last sync was performed headBytes int64 // Number of bytes written to the head file readMeter *metrics.Meter // Meter for measuring the effective amount of data read From d7db10ddbde2b1a80485e25808410a2a3b44b836 Mon Sep 17 00:00:00 2001 From: asamuj <105436033+asamuj@users.noreply.github.com> Date: Tue, 15 Jul 2025 14:20:45 +0800 Subject: [PATCH 8/8] eth/protocols/snap, p2p/discover: improve zero time checks (#32214) --- eth/protocols/snap/sync.go | 6 ++---- p2p/discover/table.go | 2 +- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/eth/protocols/snap/sync.go b/eth/protocols/snap/sync.go index 9e079f540f..9982d0f55c 100644 --- a/eth/protocols/snap/sync.go +++ b/eth/protocols/snap/sync.go @@ -615,7 +615,7 @@ func (s *Syncer) Sync(root common.Hash, cancel chan struct{}) error { s.statelessPeers = make(map[string]struct{}) s.lock.Unlock() - if s.startTime == (time.Time{}) { + if s.startTime.IsZero() { s.startTime = time.Now() } // Retrieve the previous sync status from LevelDB and abort if already synced @@ -1987,9 +1987,7 @@ func (s *Syncer) processAccountResponse(res *accountResponse) { func (s *Syncer) processBytecodeResponse(res *bytecodeResponse) { batch := s.db.NewBatch() - var ( - codes uint64 - ) + var codes uint64 for i, hash := range res.hashes { code := res.codes[i] diff --git a/p2p/discover/table.go b/p2p/discover/table.go index 6a3d1af3af..b6c35aaaa9 100644 --- a/p2p/discover/table.go +++ b/p2p/discover/table.go @@ -562,7 +562,7 @@ func (tab *Table) addReplacement(b *bucket, n *enode.Node) { } func (tab *Table) nodeAdded(b *bucket, n *tableNode) { - if n.addedToTable == (time.Time{}) { + if n.addedToTable.IsZero() { n.addedToTable = time.Now() } n.addedToBucket = time.Now()