From 6ad259a105f56cad997458cbe2b641b4b2c72e04 Mon Sep 17 00:00:00 2001 From: awskii Date: Wed, 26 Aug 2026 17:03:13 +0700 Subject: [PATCH 1/5] execution/commitment: report commitment progress from the parallel engine so commit_* metrics stop reading zero --- execution/commitment/hex_patricia_hashed.go | 15 +-- execution/commitment/metrics.go | 50 ++++++++- execution/commitment/parallel_metrics_test.go | 106 ++++++++++++++++++ execution/commitment/parallel_mount.go | 1 + .../commitment/parallel_patricia_hashed.go | 20 +++- execution/stagedsync/committer.go | 44 +++++++- execution/stagedsync/exec3.go | 7 +- execution/stagedsync/exec3_metrics.go | 2 +- execution/stagedsync/exec3_parallel.go | 11 +- execution/stagedsync/exec3_serial.go | 24 +++- 10 files changed, 244 insertions(+), 36 deletions(-) create mode 100644 execution/commitment/parallel_metrics_test.go diff --git a/execution/commitment/hex_patricia_hashed.go b/execution/commitment/hex_patricia_hashed.go index 2b93a5adcb4..8417c960edf 100644 --- a/execution/commitment/hex_patricia_hashed.go +++ b/execution/commitment/hex_patricia_hashed.go @@ -2242,10 +2242,7 @@ func (hph *HexPatriciaHashed) RootHash() ([]byte, error) { func (hph *HexPatriciaHashed) unfoldKeyPath(hashedKey, plainKey []byte) error { for unfolding := hph.needUnfolding(hashedKey); unfolding > 0; unfolding = hph.needUnfolding(hashedKey) { printLater := hph.currentKeyLen == 0 && hph.mounted && hph.traceW != nil - var unfoldDone func() - if dbg.KVReadLevelledMetrics { - unfoldDone = hph.metrics.StartUnfolding(plainKey) - } + unfoldDone := hph.metrics.StartUnfolding(plainKey) if err := hph.unfold(hashedKey, unfolding); err != nil { return fmt.Errorf("unfold: %w", err) } @@ -2265,10 +2262,7 @@ func (hph *HexPatriciaHashed) followAndUpdate(hashedKey, plainKey []byte, stateU //} // Keep folding until the currentKey is the prefix of the key we modify for hph.needFolding(hashedKey) { - var foldDone func() - if dbg.KVReadLevelledMetrics { - foldDone = hph.metrics.StartFolding(plainKey) - } + foldDone := hph.metrics.StartFolding(plainKey) if err := hph.fold(); err != nil { return fmt.Errorf("fold: %w", err) } @@ -2567,10 +2561,7 @@ func (hph *HexPatriciaHashed) Process(ctx context.Context, updates *Updates, log // Folding everything up to the root for hph.activeRows > 0 { - var foldDone func() - if dbg.KVReadLevelledMetrics { - foldDone = hph.metrics.StartFolding(nil) - } + foldDone := hph.metrics.StartFolding(nil) if err = hph.fold(); err != nil { return nil, fmt.Errorf("final fold: %w", err) } diff --git a/execution/commitment/metrics.go b/execution/commitment/metrics.go index f00321190bf..e2c92defb99 100644 --- a/execution/commitment/metrics.go +++ b/execution/commitment/metrics.go @@ -41,6 +41,7 @@ type Metrics struct { updateBranch atomic.Uint64 loadDepths [10]uint64 unfolds atomic.Uint64 + folds atomic.Uint64 spentUnfolding atomic.Int64 spentFolding atomic.Int64 spentProcessing atomic.Int64 @@ -68,6 +69,7 @@ type MetricValues struct { UpdateBranch uint64 LoadDepths [10]uint64 Unfolds uint64 + Folds uint64 SpentUnfolding time.Duration SpentFolding time.Duration SpentProcessing time.Duration @@ -142,6 +144,7 @@ func (m *Metrics) AsValues() MetricValues { UpdateBranch: m.updateBranch.Load(), LoadDepths: m.loadDepths, Unfolds: m.unfolds.Load(), + Folds: m.folds.Load(), SpentUnfolding: time.Duration(m.spentUnfolding.Load()), SpentFolding: time.Duration(m.spentFolding.Load()), SpentProcessing: time.Duration(m.spentProcessing.Load()), @@ -172,7 +175,7 @@ func (m *Metrics) logMetrics() []any { "cs", common.PrettyCounter(m.cacheStorage.Load()), "mb", common.PrettyCounter(m.missBranch.Load()), "ma", common.PrettyCounter(m.missAccount.Load()), "ms", common.PrettyCounter(m.missStorage.Load()), - "fld", common.PrettyCounter(m.unfolds.Load()), "pdur", common.Round(time.Duration(m.spentProcessing.Load()), 0).String(), + "fld", common.PrettyCounter(m.folds.Load()), "ufld", common.PrettyCounter(m.unfolds.Load()), "pdur", common.Round(time.Duration(m.spentProcessing.Load()), 0).String(), "fdur", common.Round(time.Duration(m.spentFolding.Load()), 0).String(), "ufdur", common.Round(time.Duration(m.spentUnfolding.Load()), 0), } } @@ -306,7 +309,14 @@ func (m *Metrics) Reset() { m.missBranch.Store(0) m.missAccount.Store(0) m.missStorage.Store(0) + m.cacheBranch.Store(0) + m.cacheAccount.Store(0) + m.cacheStorage.Store(0) m.unfolds.Store(0) + m.folds.Store(0) + m.spentUnfolding.Store(0) + m.spentFolding.Store(0) + m.spentProcessing.Store(0) if !m.collectCommitmentMetrics { return @@ -314,9 +324,6 @@ func (m *Metrics) Reset() { m.Accounts.Reset() m.Branches.Reset() - m.spentUnfolding.Store(0) - m.spentFolding.Store(0) - m.spentProcessing.Store(0) } func (m *Metrics) CollectFileDepthStats(endTxNumStats map[uint64]skipStat) { @@ -375,6 +382,10 @@ func (m *Metrics) BranchLoad(plainKey []byte) { } } +// StartUnfolding counts the unfold always — one atomic add on a worker-private +// line — and returns a timing closure only when per-key metrics are on, since +// that costs a time.Now() and an escaping closure per call. A nil return means +// there is nothing to stop. func (m *Metrics) StartUnfolding(plainKey []byte) func() { m.unfolds.Add(1) if m.collectCommitmentMetrics { @@ -387,10 +398,11 @@ func (m *Metrics) StartUnfolding(plainKey []byte) func() { }) } } - return func() {} + return nil } func (m *Metrics) StartFolding(plainKey []byte) func() { + m.folds.Add(1) if m.collectCommitmentMetrics { start := time.Now() return func() { @@ -401,7 +413,33 @@ func (m *Metrics) StartFolding(plainKey []byte) func() { }) } } - return func() {} + return nil +} + +// Merge folds src's counters into m. The parallel trie gives every mount +// worker its own Metrics — an atomic add on a shared line in the fold loop +// would cost more than the counter is worth — and merges once per round. +func (m *Metrics) Merge(src *Metrics) { + if src == nil || m == src { + return + } + m.addressKeys.Add(src.addressKeys.Load()) + m.storageKeys.Add(src.storageKeys.Load()) + m.loadBranch.Add(src.loadBranch.Load()) + m.loadAccount.Add(src.loadAccount.Load()) + m.loadStorage.Add(src.loadStorage.Load()) + m.cacheBranch.Add(src.cacheBranch.Load()) + m.cacheAccount.Add(src.cacheAccount.Load()) + m.cacheStorage.Add(src.cacheStorage.Load()) + m.missBranch.Add(src.missBranch.Load()) + m.missAccount.Add(src.missAccount.Load()) + m.missStorage.Add(src.missStorage.Load()) + m.updateBranch.Add(src.updateBranch.Load()) + m.unfolds.Add(src.unfolds.Load()) + m.folds.Add(src.folds.Load()) + m.spentUnfolding.Add(src.spentUnfolding.Load()) + m.spentFolding.Add(src.spentFolding.Load()) + m.spentProcessing.Add(src.spentProcessing.Load()) } func (m *Metrics) TotalProcessingTimeInc(t time.Time) { diff --git a/execution/commitment/parallel_metrics_test.go b/execution/commitment/parallel_metrics_test.go new file mode 100644 index 00000000000..02ae7e618d6 --- /dev/null +++ b/execution/commitment/parallel_metrics_test.go @@ -0,0 +1,106 @@ +// 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 commitment + +import ( + "context" + "math/rand" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// buildNibbleSpread returns a corpus whose accounts land under distinct root +// nibbles, so processMounted actually fans out across workers. +func buildNibbleSpread(t *testing.T, nibbles, slots int) ([][]byte, []Update) { + t.Helper() + rnd := rand.New(rand.NewSource(9931)) + ub := NewUpdateBuilder() + for n := range nibbles { + addNibbleAccount(ub, rnd, n, n, slots) + } + return ub.Build() +} + +func TestParallelPatriciaHashedReportsProgress(t *testing.T) { + ms := NewMockState(t) + keys, upds := buildNibbleSpread(t, 16, 4) + require.NoError(t, ms.applyPlainUpdates(keys, upds)) + + tr := newParTrie(t, ms, 4) + defer tr.Release() + ut := NewUpdates(ModeParallel, t.TempDir(), KeyToHexNibbleHash) + defer ut.Close() + for _, k := range keys { + ut.TouchPlainKey(string(k), nil, nil) + } + + var got []*CommitProgress + _, err := tr.Process(context.Background(), ut, "", func(p *CommitProgress) { + got = append(got, p) + }, WarmupConfig{}) + require.NoError(t, err) + + require.Len(t, got, 1, "parallel Process must report the round exactly once") + p := got[0] + assert.Equal(t, p.UpdateCount, p.KeyIndex, "terminal callback reports a finished round") + + // The counters must describe the whole round, not one worker's slice. + m := p.Metrics + assert.Positive(t, m.AddressKeys) + assert.Positive(t, m.StorageKeys) + assert.Positive(t, m.Folds, "fold count is recorded") + assert.Positive(t, m.UpdateBranch, "deferred branch writes are counted") + + // The parallel engine re-traverses some subtrees (mount+replay), so its key + // counts are traversals, not distinct keys, and legitimately exceed the + // sequential engine's. Guard the direction: under-counting would mean a + // worker's Metrics never got merged. + seqMS := NewMockState(t) + require.NoError(t, seqMS.applyPlainUpdates(keys, upds)) + seq := newSeqTrie(t, seqMS) + defer seq.Release() + sut := WrapKeyUpdates(t, ModeDirect, KeyToHexNibbleHash, keys, upds) + defer sut.Close() + _, err = seq.Process(context.Background(), sut, "", nil, WarmupConfig{}) + require.NoError(t, err) + assert.GreaterOrEqual(t, m.AddressKeys, seq.metrics.addressKeys.Load(), + "every worker's address keys are merged in") + assert.GreaterOrEqual(t, m.StorageKeys, seq.metrics.storageKeys.Load(), + "every worker's storage keys are merged in") +} + +func TestMetricsResetClearsEveryCounter(t *testing.T) { + m := NewMetrics("") + m.cacheBranch.Add(3) + m.cacheAccount.Add(4) + m.cacheStorage.Add(5) + m.folds.Add(6) + m.unfolds.Add(7) + m.spentFolding.Add(8) + + m.Reset() + + v := m.AsValues() + assert.Zero(t, v.CacheBranch) + assert.Zero(t, v.CacheAccount) + assert.Zero(t, v.CacheStorage) + assert.Zero(t, v.Folds) + assert.Zero(t, v.Unfolds) + assert.Zero(t, v.SpentFolding) +} diff --git a/execution/commitment/parallel_mount.go b/execution/commitment/parallel_mount.go index 0d24e0e20d0..eda7cc14e02 100644 --- a/execution/commitment/parallel_mount.go +++ b/execution/commitment/parallel_mount.go @@ -138,6 +138,7 @@ func (p *ParallelPatriciaHashed) processMounted(ctx context.Context, updates *Up ni, ch := nib, child g.Go(func() error { w := NewHexPatriciaHashed(p.accountKeyLen, nil, p.cfg) + defer p.metrics.Merge(w.metrics) w.mountTo(base, ni) if p.template != nil && p.template.traceW != nil { w.traceW = tracePrefix(p.template.traceW, fmt.Sprintf("[mnt %x] ", ni)) diff --git a/execution/commitment/parallel_patricia_hashed.go b/execution/commitment/parallel_patricia_hashed.go index 7efc4ee4036..517206623d5 100644 --- a/execution/commitment/parallel_patricia_hashed.go +++ b/execution/commitment/parallel_patricia_hashed.go @@ -41,6 +41,11 @@ type ParallelPatriciaHashed struct { leaveDeferredForCaller bool deferredForCaller []*DeferredBranchUpdate + + // metrics is the round's aggregate: each mount worker counts into its own + // Metrics and merges here when it finishes, so the fold loop never touches + // a shared cache line. + metrics *Metrics } func (p *ParallelPatriciaHashed) DeepLocalFolds() uint64 { return p.deepLocalFolds.Load() } @@ -53,6 +58,9 @@ func NewParallelPatriciaHashed(ctxFactory TrieContextFactory, accountKeyLen int1 numWorkers: runtime.NumCPU(), cfg: cfg, } + // Its own, not the template's: the template traverses the skeleton over the + // same keys the workers do, so aliasing them counts every key twice. + p.metrics = NewMetrics("") return p } @@ -250,6 +258,10 @@ func (p *ParallelPatriciaHashed) Process( copy(out, rh) p.rootHash.Store(&out) flushTrieStateRates() + if onProgress != nil && p.metrics != nil { + n := updates.Size() + onProgress(&CommitProgress{KeyIndex: n, UpdateCount: n, Metrics: p.metrics.AsValues()}) + } return out, nil } @@ -305,8 +317,14 @@ func (p *ParallelPatriciaHashed) applyDeferredUpdates(ctx context.Context, pu *p return errors.New("ParallelPatriciaHashed: trieCtxFactory returned nil context for deferred apply") } - if _, err := ApplyDeferredBranchUpdates(deferred, p.numWorkers, applyCtx.PutBranch); err != nil { + // This path calls PutBranch directly rather than through a BranchEncoder, + // so it is the only place the parallel engine's branch writes get counted. + n, err := ApplyDeferredBranchUpdates(deferred, p.numWorkers, applyCtx.PutBranch) + if err != nil { return fmt.Errorf("apply deferred branch updates: %w", err) } + if p.metrics != nil { + p.metrics.updateBranch.Add(uint64(n)) + } return nil } diff --git a/execution/stagedsync/committer.go b/execution/stagedsync/committer.go index bdf7b2f48ec..da1b2c1e631 100644 --- a/execution/stagedsync/committer.go +++ b/execution/stagedsync/committer.go @@ -9,6 +9,7 @@ import ( "runtime/pprof" "sync" "sync/atomic" + "time" "github.com/erigontech/erigon/common" "github.com/erigontech/erigon/common/dbg" @@ -107,6 +108,12 @@ type commitmentCalculator struct { // Opened at start, lives for the calculator's lifetime. roTx kv.TemporalTx + // commitProgress holds the most recent CommitProgress the trie reported, + // and firstCommitAtNs the wall time of the first round. The parallel + // executor reads both to log commitment stats on its own ticker. + commitProgress atomic.Pointer[commitment.CommitProgress] + firstCommitAtNs atomic.Int64 + // lastTarget tracks the most recent block boundary so that // computeAndPublish knows which block to compute for. lastTarget commitTarget @@ -279,6 +286,34 @@ func newCommitmentCalculator( }, nil } +// onCommitProgress is handed to ComputeCommitment so the trie's counters +// reach the executor. Called from the calculator goroutine. +func (cc *commitmentCalculator) onCommitProgress(p *commitment.CommitProgress) { + if p == nil { + return + } + cc.commitProgress.Store(p) + cc.firstCommitAtNs.CompareAndSwap(0, time.Now().UnixNano()) +} + +// LastCommitProgress returns the most recent round's counters, or the zero +// value if no round has completed. +func (cc *commitmentCalculator) LastCommitProgress() commitment.CommitProgress { + if p := cc.commitProgress.Load(); p != nil { + return *p + } + return commitment.CommitProgress{} +} + +// FirstCommitAt reports when the first round ran; zero if none has. +func (cc *commitmentCalculator) FirstCommitAt() time.Time { + ns := cc.firstCommitAtNs.Load() + if ns == 0 { + return time.Time{} + } + return time.Unix(0, ns) +} + func (cc *commitmentCalculator) Start(ctx context.Context) { cc.wg.Go(func() { cc.loop(ctx) @@ -855,7 +890,7 @@ func (cc *commitmentCalculator) computeIsolated(ctx context.Context, t commitTar defer cc.doms.UnlockChangesetAccumulator() defer cc.doms.DetachChangesetAccumulatorLocked()() - rh, err := cc.doms.ComputeCommitmentLocked(ctx, cc.roTx, true, t.blockNum, t.lastTxNum, cc.logPrefix, nil) + rh, err := cc.doms.ComputeCommitmentLocked(ctx, cc.roTx, true, t.blockNum, t.lastTxNum, cc.logPrefix, cc.onCommitProgress) if err != nil { return nil, err } @@ -977,7 +1012,7 @@ func (cc *commitmentCalculator) computeWithBlockAccumulator(ctx context.Context, defer cc.doms.UnlockChangesetAccumulator() cs := cc.doms.GetChangesetByHash(t.blockNum, t.blockHash) if cs == nil { - return cc.doms.ComputeCommitmentLocked(ctx, cc.roTx, true, t.blockNum, t.lastTxNum, cc.logPrefix, nil) + return cc.doms.ComputeCommitmentLocked(ctx, cc.roTx, true, t.blockNum, t.lastTxNum, cc.logPrefix, cc.onCommitProgress) } // LOAD-BEARING swap under the outer lock (already taken above). The // swap below mutates the global current-accumulator pointer; the @@ -990,7 +1025,7 @@ func (cc *commitmentCalculator) computeWithBlockAccumulator(ctx context.Context, // Inside the lock we must use the *Locked variants — the public // counterparts re-acquire the same Mutex and would self-deadlock. defer cc.doms.SwapCommitmentDiffLocked(cs)() - return cc.doms.ComputeCommitmentLocked(ctx, cc.roTx, true, t.blockNum, t.lastTxNum, cc.logPrefix, nil) + return cc.doms.ComputeCommitmentLocked(ctx, cc.roTx, true, t.blockNum, t.lastTxNum, cc.logPrefix, cc.onCommitProgress) } // asOfStateReader reads account/storage/code at a specific txNum via @@ -1059,6 +1094,3 @@ func (r *asOfStateReader) CloneForWorker(workerCtx context.Context, tx kv.Tempor } return &asOfStateReader{sd: r.sd, roTx: tx, getter: r.sd.AsStateGetter(tx, getterOpts), txNum: r.txNum} } - -// Keep imports used. -var _ = commitment.CommitProgress{} diff --git a/execution/stagedsync/exec3.go b/execution/stagedsync/exec3.go index d4be2915373..cce54d21e4c 100644 --- a/execution/stagedsync/exec3.go +++ b/execution/stagedsync/exec3.go @@ -39,7 +39,6 @@ import ( "github.com/erigontech/erigon/db/rawdb/rawdbhelpers" "github.com/erigontech/erigon/db/rawdb/rawtemporaldb" "github.com/erigontech/erigon/db/state/execctx" - "github.com/erigontech/erigon/execution/commitment" "github.com/erigontech/erigon/execution/exec" "github.com/erigontech/erigon/execution/protocol" "github.com/erigontech/erigon/execution/protocol/rules" @@ -228,7 +227,7 @@ func execV3(ctx context.Context, isApplyingBlocks: isApplyingBlocks, logger: logger, logPrefix: logPrefix, - progress: NewProgress(blockNum, inputTxNum, commitThreshold, false, logPrefix, logger), + progress: NewProgress(blockNum, inputTxNum, commitThreshold, logPrefix, logger), enableChaosMonkey: initialCycle, hooks: hooks, blockSrc: blockSrc, @@ -342,7 +341,7 @@ func execV3Serial(ctx context.Context, applyTx: applyTx, logger: logger, logPrefix: execStage.LogPrefix(), - progress: NewProgress(blockNum, inputTxNum, commitThreshold, false, execStage.LogPrefix(), logger), + progress: NewProgress(blockNum, inputTxNum, commitThreshold, execStage.LogPrefix(), logger), enableChaosMonkey: initialCycle, hooks: hooks, }} @@ -382,7 +381,7 @@ func execV3Serial(ctx context.Context, stepsInDb = rawdbhelpers.IdxStepsCountV3(applyTx, doms.StepSize()) if initialCycle { - se.LogCommitments(committedTransactions, stepsInDb, commitment.CommitProgress{}) + se.LogCommitments(committedTransactions, stepsInDb, se.LastCommitProgress()) } case errors.Is(execErr, ErrWrongTrieRoot): execErr = handleIncorrectRootHashError( diff --git a/execution/stagedsync/exec3_metrics.go b/execution/stagedsync/exec3_metrics.go index 5a7cd318a42..8dfe130000f 100644 --- a/execution/stagedsync/exec3_metrics.go +++ b/execution/stagedsync/exec3_metrics.go @@ -445,7 +445,7 @@ func updateExecDomainMetrics(metrics *kvmetrics.DomainMetrics, prevMetrics *kvme return prevMetrics } -func NewProgress(initialBlockNum, initialTxNum, commitThreshold uint64, updateMetrics bool, logPrefix string, logger log.Logger) *Progress { +func NewProgress(initialBlockNum, initialTxNum, commitThreshold uint64, logPrefix string, logger log.Logger) *Progress { now := time.Now() return &Progress{ initialTime: now, diff --git a/execution/stagedsync/exec3_parallel.go b/execution/stagedsync/exec3_parallel.go index 5486178a1e0..6468e07358e 100644 --- a/execution/stagedsync/exec3_parallel.go +++ b/execution/stagedsync/exec3_parallel.go @@ -381,9 +381,6 @@ func (pe *parallelExecutor) execImpl(ctx context.Context, var uncommittedGas int64 var hasLoggedExecution bool var hasLoggedCommittments atomic.Bool - var commitStart time.Time - - var lastProgress commitment.CommitProgress execErr := func() (err error) { defer func() { @@ -832,6 +829,10 @@ func (pe *parallelExecutor) execImpl(ctx context.Context, hasLoggedExecution = true lastExecutedLog = time.Now() pe.LogExecution() + if !calculator.FirstCommitAt().IsZero() { + hasLoggedCommittments.Store(true) + pe.LogCommitments(0, stepsInDb, calculator.LastCommitProgress()) + } agg := pe.cfg.db.(dbstate.HasAgg).Agg().(*dbstate.Aggregator) if agg.HasBackgroundFilesBuild() { pe.logger.Info(fmt.Sprintf("[%s] Background files build", pe.logPrefix), "progress", agg.BackgroundProgress()) @@ -859,8 +860,8 @@ func (pe *parallelExecutor) execImpl(ctx context.Context, // Commitment is computed per-block by the calculator. Stage progress // is updated in handleCommitResult when results are consumed. - if !hasLoggedCommittments.Load() && !commitStart.IsZero() { - pe.LogCommitments(0, stepsInDb, lastProgress) + if !hasLoggedCommittments.Load() && !calculator.FirstCommitAt().IsZero() { + pe.LogCommitments(0, stepsInDb, calculator.LastCommitProgress()) } if execErr != nil { diff --git a/execution/stagedsync/exec3_serial.go b/execution/stagedsync/exec3_serial.go index 0713d5e2239..4bebda74da6 100644 --- a/execution/stagedsync/exec3_serial.go +++ b/execution/stagedsync/exec3_serial.go @@ -18,6 +18,8 @@ import ( "github.com/erigontech/erigon/db/rawdb/rawtemporaldb" "github.com/erigontech/erigon/db/state/changeset" "github.com/erigontech/erigon/db/state/execctx/execctxapi" + "sync/atomic" + "github.com/erigontech/erigon/execution/commitment" "github.com/erigontech/erigon/execution/exec" "github.com/erigontech/erigon/execution/protocol" @@ -39,6 +41,10 @@ type serialExecutor struct { blobGasUsed uint64 worker *exec.Worker + // commitProgress holds the most recent CommitProgress the trie reported, + // so the caller can log real commitment counters instead of a zero value. + commitProgress atomic.Pointer[commitment.CommitProgress] + // accumulator for the current block; set at StartChange and used by the // block-end stateWriter so that AuRa system-call nonce changes are // included in the txpool state-diff batch. @@ -186,7 +192,7 @@ func (se *serialExecutor) exec(ctx context.Context, execStage *StageState, u Unw se.doms.GetCommitmentCtx().SetTraceWriter(os.Stderr) } // Warmup is enabled via EnableTrieWarmup at executor init - rh, err := se.doms.ComputeCommitment(ctx, se.applyTx, true, blockNum, inputTxNum-1, se.logPrefix, nil) + rh, err := se.doms.ComputeCommitment(ctx, se.applyTx, true, blockNum, inputTxNum-1, se.logPrefix, se.onCommitProgress) if traceBlk { se.doms.GetCommitmentCtx().SetTraceWriter(nil) } @@ -281,6 +287,22 @@ func (se *serialExecutor) LogExecution() { se.progress.LogExecution(se.rs.StateV3, se) } +// onCommitProgress records the trie's counters for the round just finished. +func (se *serialExecutor) onCommitProgress(p *commitment.CommitProgress) { + if p != nil { + se.commitProgress.Store(p) + } +} + +// LastCommitProgress returns the most recent round's counters, or the zero +// value if no round has completed. +func (se *serialExecutor) LastCommitProgress() commitment.CommitProgress { + if p := se.commitProgress.Load(); p != nil { + return *p + } + return commitment.CommitProgress{} +} + func (se *serialExecutor) LogCommitments(committedTransactions uint64, stepsInDb float64, lastProgress commitment.CommitProgress) { se.txExecutor.lastCommittedTxNum.Add(committedTransactions) se.progress.LogCommitments(se.rs.StateV3, se, stepsInDb, lastProgress) From eda32c5ce3c666e47f241e2878336ef1e08a096c Mon Sep 17 00:00:00 2001 From: awskii Date: Wed, 26 Aug 2026 17:25:15 +0700 Subject: [PATCH 2/5] execution/commitment: replace commit_* rate gauges with counters, round histogram and branch IO bytes --- execution/commitment/commitment.go | 6 + execution/commitment/hex_patricia_hashed.go | 3 + execution/commitment/metrics.go | 116 ++++++++++------- execution/commitment/parallel_metrics_test.go | 26 ++++ .../commitment/parallel_patricia_hashed.go | 11 ++ execution/commitment/prom_metrics.go | 40 ++++++ execution/stagedsync/exec3_metrics.go | 123 ++++++++++++------ 7 files changed, 243 insertions(+), 82 deletions(-) create mode 100644 execution/commitment/prom_metrics.go diff --git a/execution/commitment/commitment.go b/execution/commitment/commitment.go index dce122acc6a..9db6e1f9e9a 100644 --- a/execution/commitment/commitment.go +++ b/execution/commitment/commitment.go @@ -372,6 +372,11 @@ func (be *BranchEncoder) ApplyDeferredUpdates( } if be.metrics != nil { be.metrics.updateBranch.Add(uint64(written)) + var bytesOut int + for _, upd := range be.deferred { + bytesOut += len(upd.encoded) + } + be.metrics.AddBranchWrite(bytesOut) } return nil } @@ -502,6 +507,7 @@ func (be *BranchEncoder) CollectUpdate( } if be.metrics != nil { be.metrics.updateBranch.Add(1) + be.metrics.AddBranchWrite(len(updateCopy)) } mxTrieBranchesUpdated.Inc() return nil diff --git a/execution/commitment/hex_patricia_hashed.go b/execution/commitment/hex_patricia_hashed.go index 8417c960edf..b0a9b47e477 100644 --- a/execution/commitment/hex_patricia_hashed.go +++ b/execution/commitment/hex_patricia_hashed.go @@ -1464,6 +1464,7 @@ func (hph *HexPatriciaHashed) unfoldBranchNode(row int, depth int16, deleted boo if err != nil { return err } + hph.metrics.AddBranchRead(len(branchData)) // depthsToTxNum is used for per-file metrics; step is no longer available // from the cache-or-DB helper (cache never had a meaningful step anyway). @@ -2483,6 +2484,8 @@ func (hph *HexPatriciaHashed) Process(ctx context.Context, updates *Updates, log hph.metrics.Reset() hph.metrics.updates.Store(updatesCount) + hph.metrics.AddRoundKeys(updatesCount) + defer ObserveRound(time.Now()) if hph.metrics.collectCommitmentMetrics { defer func() { hph.metrics.TotalProcessingTimeInc(start) diff --git a/execution/commitment/metrics.go b/execution/commitment/metrics.go index e2c92defb99..08d7f68a0d6 100644 --- a/execution/commitment/metrics.go +++ b/execution/commitment/metrics.go @@ -42,6 +42,9 @@ type Metrics struct { loadDepths [10]uint64 unfolds atomic.Uint64 folds atomic.Uint64 + roundKeys atomic.Uint64 + branchReadBytes atomic.Uint64 + branchWriteBytes atomic.Uint64 spentUnfolding atomic.Int64 spentFolding atomic.Int64 spentProcessing atomic.Int64 @@ -51,28 +54,34 @@ type Metrics struct { } type MetricValues struct { - mu *sync.RWMutex - Accounts map[string]*AccountStats - Branches map[string]*BranchStats - Updates uint64 - AddressKeys uint64 - StorageKeys uint64 - LoadBranch uint64 - LoadAccount uint64 - LoadStorage uint64 - CacheBranch uint64 - CacheAccount uint64 - CacheStorage uint64 - MissBranch uint64 - MissAccount uint64 - MissStorage uint64 - UpdateBranch uint64 - LoadDepths [10]uint64 - Unfolds uint64 - Folds uint64 - SpentUnfolding time.Duration - SpentFolding time.Duration - SpentProcessing time.Duration + mu *sync.RWMutex + Accounts map[string]*AccountStats + Branches map[string]*BranchStats + Updates uint64 + AddressKeys uint64 + StorageKeys uint64 + LoadBranch uint64 + LoadAccount uint64 + LoadStorage uint64 + CacheBranch uint64 + CacheAccount uint64 + CacheStorage uint64 + MissBranch uint64 + MissAccount uint64 + MissStorage uint64 + UpdateBranch uint64 + LoadDepths [10]uint64 + Unfolds uint64 + Folds uint64 + // RoundKeys counts distinct keys handed to the trie, summed over rounds. + // AddressKeys/StorageKeys count cell traversals instead, which the parallel + // engine inflates by re-walking subtrees on mount+replay. + RoundKeys uint64 + BranchReadBytes uint64 + BranchWriteBytes uint64 + SpentUnfolding time.Duration + SpentFolding time.Duration + SpentProcessing time.Duration } func (m MetricValues) RLock() { @@ -126,28 +135,31 @@ func (m *Metrics) EnableCsvMetrics(filePathPrefix string) { func (m *Metrics) AsValues() MetricValues { return MetricValues{ - mu: &m.Accounts.m, - Accounts: m.Accounts.AccountStats, - Branches: m.Branches.BranchStats, - Updates: m.updates.Load(), - AddressKeys: m.addressKeys.Load(), - StorageKeys: m.storageKeys.Load(), - LoadBranch: m.loadBranch.Load(), - LoadAccount: m.loadAccount.Load(), - LoadStorage: m.loadStorage.Load(), - CacheBranch: m.cacheBranch.Load(), - CacheAccount: m.cacheAccount.Load(), - CacheStorage: m.cacheStorage.Load(), - MissBranch: m.missBranch.Load(), - MissAccount: m.missAccount.Load(), - MissStorage: m.missStorage.Load(), - UpdateBranch: m.updateBranch.Load(), - LoadDepths: m.loadDepths, - Unfolds: m.unfolds.Load(), - Folds: m.folds.Load(), - SpentUnfolding: time.Duration(m.spentUnfolding.Load()), - SpentFolding: time.Duration(m.spentFolding.Load()), - SpentProcessing: time.Duration(m.spentProcessing.Load()), + mu: &m.Accounts.m, + Accounts: m.Accounts.AccountStats, + Branches: m.Branches.BranchStats, + Updates: m.updates.Load(), + AddressKeys: m.addressKeys.Load(), + StorageKeys: m.storageKeys.Load(), + LoadBranch: m.loadBranch.Load(), + LoadAccount: m.loadAccount.Load(), + LoadStorage: m.loadStorage.Load(), + CacheBranch: m.cacheBranch.Load(), + CacheAccount: m.cacheAccount.Load(), + CacheStorage: m.cacheStorage.Load(), + MissBranch: m.missBranch.Load(), + MissAccount: m.missAccount.Load(), + MissStorage: m.missStorage.Load(), + UpdateBranch: m.updateBranch.Load(), + LoadDepths: m.loadDepths, + Unfolds: m.unfolds.Load(), + Folds: m.folds.Load(), + RoundKeys: m.roundKeys.Load(), + BranchReadBytes: m.branchReadBytes.Load(), + BranchWriteBytes: m.branchWriteBytes.Load(), + SpentUnfolding: time.Duration(m.spentUnfolding.Load()), + SpentFolding: time.Duration(m.spentFolding.Load()), + SpentProcessing: time.Duration(m.spentProcessing.Load()), } } @@ -314,6 +326,9 @@ func (m *Metrics) Reset() { m.cacheStorage.Store(0) m.unfolds.Store(0) m.folds.Store(0) + m.roundKeys.Store(0) + m.branchReadBytes.Store(0) + m.branchWriteBytes.Store(0) m.spentUnfolding.Store(0) m.spentFolding.Store(0) m.spentProcessing.Store(0) @@ -416,6 +431,17 @@ func (m *Metrics) StartFolding(plainKey []byte) func() { return nil } +// AddBranchRead records one branch read of n bytes. +func (m *Metrics) AddBranchRead(n int) { m.branchReadBytes.Add(uint64(n)) } + +// AddBranchWrite records one branch write of n bytes. +func (m *Metrics) AddBranchWrite(n int) { m.branchWriteBytes.Add(uint64(n)) } + +// AddRoundKeys records the distinct-key count of one finished round. Not +// merged between tries: the engine that ran the round owns this number, while +// each worker only sees its own subtree. +func (m *Metrics) AddRoundKeys(n uint64) { m.roundKeys.Add(n) } + // Merge folds src's counters into m. The parallel trie gives every mount // worker its own Metrics — an atomic add on a shared line in the fold loop // would cost more than the counter is worth — and merges once per round. @@ -437,6 +463,8 @@ func (m *Metrics) Merge(src *Metrics) { m.updateBranch.Add(src.updateBranch.Load()) m.unfolds.Add(src.unfolds.Load()) m.folds.Add(src.folds.Load()) + m.branchReadBytes.Add(src.branchReadBytes.Load()) + m.branchWriteBytes.Add(src.branchWriteBytes.Load()) m.spentUnfolding.Add(src.spentUnfolding.Load()) m.spentFolding.Add(src.spentFolding.Load()) m.spentProcessing.Add(src.spentProcessing.Load()) diff --git a/execution/commitment/parallel_metrics_test.go b/execution/commitment/parallel_metrics_test.go index 02ae7e618d6..9ef8ffb5c84 100644 --- a/execution/commitment/parallel_metrics_test.go +++ b/execution/commitment/parallel_metrics_test.go @@ -104,3 +104,29 @@ func TestMetricsResetClearsEveryCounter(t *testing.T) { assert.Zero(t, v.Unfolds) assert.Zero(t, v.SpentFolding) } + +func TestRoundKeysAreDistinctNotTraversals(t *testing.T) { + ms := NewMockState(t) + keys, upds := buildNibbleSpread(t, 16, 4) + require.NoError(t, ms.applyPlainUpdates(keys, upds)) + + tr := newParTrie(t, ms, 4) + defer tr.Release() + ut := NewUpdates(ModeParallel, t.TempDir(), KeyToHexNibbleHash) + defer ut.Close() + for _, k := range keys { + ut.TouchPlainKey(string(k), nil, nil) + } + + var got *CommitProgress + _, err := tr.Process(context.Background(), ut, "", func(p *CommitProgress) { got = p }, WarmupConfig{}) + require.NoError(t, err) + require.NotNil(t, got) + + m := got.Metrics + assert.EqualValues(t, len(keys), m.RoundKeys, + "RoundKeys is the distinct key count handed to the trie") + assert.GreaterOrEqual(t, m.AddressKeys+m.StorageKeys, m.RoundKeys, + "traversals count cell visits, so they never undercount distinct keys") + assert.Positive(t, m.BranchWriteBytes, "branch write bytes are counted") +} diff --git a/execution/commitment/parallel_patricia_hashed.go b/execution/commitment/parallel_patricia_hashed.go index 517206623d5..ed89668c763 100644 --- a/execution/commitment/parallel_patricia_hashed.go +++ b/execution/commitment/parallel_patricia_hashed.go @@ -25,6 +25,7 @@ import ( "runtime" "sync" "sync/atomic" + "time" ) type ParallelPatriciaHashed struct { @@ -232,6 +233,9 @@ func (p *ParallelPatriciaHashed) Process( return rh, nil } + defer ObserveRound(time.Now()) + p.metrics.AddRoundKeys(updates.Size()) + rh, mErr := p.processMounted(ctx, updates) if mErr != nil { pu.deferredMu.Lock() @@ -325,6 +329,13 @@ func (p *ParallelPatriciaHashed) applyDeferredUpdates(ctx context.Context, pu *p } if p.metrics != nil { p.metrics.updateBranch.Add(uint64(n)) + // encoded is filled by the merge inside the apply, so this only sums + // after it returns; the pool recycles these on the deferred cleanup. + var bytesOut int + for _, upd := range deferred { + bytesOut += len(upd.encoded) + } + p.metrics.AddBranchWrite(bytesOut) } return nil } diff --git a/execution/commitment/prom_metrics.go b/execution/commitment/prom_metrics.go new file mode 100644 index 00000000000..48516b0d044 --- /dev/null +++ b/execution/commitment/prom_metrics.go @@ -0,0 +1,40 @@ +// 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 commitment + +import ( + "time" + + "github.com/erigontech/erigon/diagnostics/metrics" +) + +// Round-level commitment metrics. Counters and a histogram rather than gauges +// holding pre-divided rates: rate() belongs in the query, so the window stays +// the reader's choice and a missed scrape costs a sample instead of an interval. +var ( + mxRounds = metrics.GetOrCreateCounter("commitment_rounds_total") + + // Buckets span a fast incremental block through a whale fold. + mxRoundDuration = metrics.NewHistogram("commitment_round_duration_seconds", + []float64{0.001, 0.005, 0.01, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60}) +) + +// ObserveRound records one finished commitment round. +func ObserveRound(start time.Time) { + mxRounds.Inc() + mxRoundDuration.ObserveDuration(start) +} diff --git a/execution/stagedsync/exec3_metrics.go b/execution/stagedsync/exec3_metrics.go index 8dfe130000f..908fcc325d3 100644 --- a/execution/stagedsync/exec3_metrics.go +++ b/execution/stagedsync/exec3_metrics.go @@ -106,19 +106,26 @@ var ( mxExecCodeDomainFileReads = metrics.NewGauge(`exec_domain_file_read_rate{domain="code"}`) mxExecCodeDomainFileReadDuration = metrics.NewGauge(`exec_domain_file_read_dur{domain="code"}`) - mxCommitmentTransactions = metrics.NewGauge(`commit_txns`) - mxCommitmentBlocks = metrics.NewGauge("commit_blocks") - mxCommitmentBlockDuration = metrics.NewGauge("commit_block_dur") - mxCommitmentReadRate = metrics.NewGauge("commit_read_rate") - mxCommitmentAccountReadRate = metrics.NewGauge("commit_account_read_rate") - mxCommitmentStorageReadRate = metrics.NewGauge("commit_storage_read_rate") - mxCommitmentBranchReadRate = metrics.NewGauge("commit_branch_read_rate") - mxCommitmentBranchWriteRate = metrics.NewGauge("commit_branch_write_rate") - mxCommitmentKeyRate = metrics.NewGauge("commit_key_rate") - mxCommitmentAccountKeyRate = metrics.NewGauge("commit_account_key_rate") - mxCommitmentStorageKeyRate = metrics.NewGauge("commit_storage_key_rate") - mxCommitmentFoldRate = metrics.NewGauge("commit_fold_rate") - mxCommitmentUnfoldRate = metrics.NewGauge("commit_unfold_rate") + // Commitment counters. rate() belongs in the query, so these are monotonic + // totals rather than gauges holding a pre-divided rate over the log interval. + mxCommitmentBlocks = metrics.GetOrCreateCounter("commitment_blocks_total") + mxCommitmentTxns = metrics.GetOrCreateCounter("commitment_txns_total") + mxCommitmentKeys = metrics.GetOrCreateCounter("commitment_keys_total") + mxCommitmentFolds = metrics.GetOrCreateCounter("commitment_folds_total") + mxCommitmentUnfolds = metrics.GetOrCreateCounter("commitment_unfolds_total") + mxCommitmentBranchPuts = metrics.GetOrCreateCounter("commitment_branch_writes_total") + mxCommitmentReadBytes = metrics.GetOrCreateCounter("commitment_branch_read_bytes_total") + mxCommitmentWriteBytes = metrics.GetOrCreateCounter("commitment_branch_write_bytes_total") + + // kind=address|storage. Traversals, not distinct keys: the parallel engine + // re-walks subtrees on mount+replay, so this exceeds commitment_keys_total. + mxCommitmentTraversals = metrics.GetOrCreateCounterVec("commitment_key_traversals_total", + []string{"kind"}, "cell traversals during commitment, by key kind") + + // kind=account|storage|branch + mxCommitmentReads = metrics.GetOrCreateCounterVec("commitment_reads_total", []string{"kind"}, "PatriciaContext reads during commitment") + mxCommitmentCacheHits = metrics.GetOrCreateCounterVec("commitment_cache_hits_total", []string{"kind"}, "commitment cache hits") + mxCommitmentCacheMisses = metrics.GetOrCreateCounterVec("commitment_cache_misses_total", []string{"kind"}, "commitment cache misses") mxCommitmentDomainReads = metrics.NewGauge(`exec_domain_read_rate{domain="commitment"}`) mxCommitmentDomainReadDuration = metrics.NewGauge(`exec_domain_read_dur{domain="commitment"}`) mxCommitmentDomainCacheReads = metrics.NewGauge(`exec_domain_cache_read_rate{domain="commitment"}`) @@ -216,9 +223,6 @@ func resetCommitmentGauges(ctx context.Context) { } else { commitResetTask.Timer = time.NewTimer(resetDelay) commitResetTask.ctx = ctx - commitResetTask.gauges = []metrics.Gauge{ - mxCommitmentTransactions, mxCommitmentBlocks, mxCommitmentBlockDuration, - } commitResetTask.run(ctx) } } @@ -248,9 +252,8 @@ func resetDomainGauges(ctx context.Context) { mxExecDomainPutKeySize, mxExecDomainPutValueSize, mxExecAccountDomainPutRate, mxExecAccountDomainPutSize, mxExecAccountDomainPutKeySize, mxExecAccountDomainPutValueSize, mxExecStorageDomainPutRate, mxExecStorageDomainPutSize, mxExecStorageDomainPutKeySize, mxExecStorageDomainPutValueSize, mxExecCodeDomainPutRate, mxExecCodeDomainPutSize, - mxExecCodeDomainPutKeySize, mxExecCodeDomainPutValueSize, mxCommitmentReadRate, mxCommitmentAccountReadRate, - mxCommitmentStorageReadRate, mxCommitmentBranchReadRate, mxCommitmentBranchWriteRate, mxCommitmentKeyRate, - mxCommitmentAccountKeyRate, mxCommitmentStorageKeyRate, mxCommitmentFoldRate, mxCommitmentUnfoldRate, mxCommitmentDomainReads, + mxExecCodeDomainPutKeySize, + mxCommitmentDomainReads, mxCommitmentDomainReadDuration, mxCommitmentDomainCacheReads, mxCommitmentDomainCacheReadDuration, mxCommitmentDomainDbReads, mxCommitmentDomainDbReadDuration, mxCommitmentDomainFileReads, mxCommitmentDomainFileReadDuration, mxCommitmentDomainPutRate, mxCommitmentDomainPutSize, mxCommitmentDomainPutKeySize, mxCommitmentDomainPutValueSize, @@ -498,6 +501,16 @@ type Progress struct { prevCommitmentStorageReadCount uint64 prevBranchReadCount uint64 prevBranchWriteCount uint64 + prevBranchReadBytes uint64 + prevBranchWriteBytes uint64 + prevFoldCount uint64 + prevUnfoldCount uint64 + prevCacheAccountHits uint64 + prevCacheStorageHits uint64 + prevCacheBranchHits uint64 + prevMissAccount uint64 + prevMissStorage uint64 + prevMissBranch uint64 commitThreshold uint64 prevDomainMetrics *kvmetrics.DomainMetrics logPrefix string @@ -776,7 +789,6 @@ func (p *Progress) LogCommitments(rs *state.StateV3, ex executor, stepsInDb floa lastProgress.Metrics.RLock() accountKeyCount := lastProgress.Metrics.AddressKeys storageKeyCount := lastProgress.Metrics.StorageKeys - keyCount := accountKeyCount + storageKeyCount accountReadCount := lastProgress.Metrics.LoadAccount storageReadCount := lastProgress.Metrics.LoadStorage branchReadCount := lastProgress.Metrics.LoadBranch @@ -787,34 +799,69 @@ func (p *Progress) LogCommitments(rs *state.StateV3, ex executor, stepsInDb floa missBranchCount := lastProgress.Metrics.MissBranch missAccountCount := lastProgress.Metrics.MissAccount missStorageCount := lastProgress.Metrics.MissStorage + roundKeyCount := lastProgress.Metrics.RoundKeys + branchReadBytes := lastProgress.Metrics.BranchReadBytes + branchWriteBytes := lastProgress.Metrics.BranchWriteBytes + foldCount := lastProgress.Metrics.Folds + unfoldCount := lastProgress.Metrics.Unfolds lastProgress.Metrics.RUnlock() - curKeyCount := int64(keyCount - p.prevCommitmentKeyCount) curAccountKeyCount := int64(accountKeyCount - p.prevCommitmentAccountKeyCount) curStorageKeyCount := int64(storageKeyCount - p.prevCommitmentStorageKeyCount) - - mxCommitmentKeyRate.Set(float64(curKeyCount) / interval.Seconds()) - mxCommitmentAccountKeyRate.Set(float64(curAccountKeyCount) / interval.Seconds()) - mxCommitmentStorageKeyRate.Set(float64(curStorageKeyCount) / interval.Seconds()) - curAccountReadCount := int64(accountReadCount - p.prevCommitmentAccountReadCount) curStorageReadCount := int64(storageReadCount - p.prevCommitmentStorageReadCount) curBranchReadCount := int64(branchReadCount - p.prevBranchReadCount) curBranchWriteCount := int64(branchWriteCount - p.prevBranchWriteCount) - curReadCount := curAccountReadCount + curStorageReadCount + curBranchReadCount - curReadRate := uint64(float64(curReadCount) / interval.Seconds()) - curBranchWriteRate := uint64(float64(curBranchWriteCount) / interval.Seconds()) - - mxCommitmentReadRate.SetUint64(curReadRate) - mxCommitmentAccountReadRate.Set(float64(curAccountReadCount) / interval.Seconds()) - mxCommitmentStorageReadRate.Set(float64(curStorageReadCount) / interval.Seconds()) - mxCommitmentBranchReadRate.Set(float64(curBranchReadCount) / interval.Seconds()) - mxCommitmentBranchWriteRate.SetUint64(curBranchWriteRate) + // The trie's counters are cumulative; Prometheus wants the increment. + addCounter := func(c metrics.Counter, delta int64) { + if delta > 0 { + c.AddUint64(uint64(delta)) + } + } + addVec := func(v *metrics.CounterVec, kind string, delta int64) { + if delta > 0 { + v.WithLabelValues(kind).Add(float64(delta)) + } + } - mxCommitmentTransactions.Set(float64(committedTxSec)) - mxCommitmentBlocks.Set(float64(committedDiffBlocks)) - mxCommitmentBlockDuration.Set(float64(commitedBlockDur)) + addCounter(mxCommitmentKeys, int64(roundKeyCount-p.prevCommitmentKeyCount)) + addVec(mxCommitmentTraversals, "address", curAccountKeyCount) + addVec(mxCommitmentTraversals, "storage", curStorageKeyCount) + addVec(mxCommitmentReads, "account", curAccountReadCount) + addVec(mxCommitmentReads, "storage", curStorageReadCount) + addVec(mxCommitmentReads, "branch", curBranchReadCount) + addCounter(mxCommitmentBranchPuts, curBranchWriteCount) + addCounter(mxCommitmentReadBytes, int64(branchReadBytes-p.prevBranchReadBytes)) + addCounter(mxCommitmentWriteBytes, int64(branchWriteBytes-p.prevBranchWriteBytes)) + addCounter(mxCommitmentFolds, int64(foldCount-p.prevFoldCount)) + addCounter(mxCommitmentUnfolds, int64(unfoldCount-p.prevUnfoldCount)) + addVec(mxCommitmentCacheHits, "account", int64(cacheAccountHits-p.prevCacheAccountHits)) + addVec(mxCommitmentCacheHits, "storage", int64(cacheStorageHits-p.prevCacheStorageHits)) + addVec(mxCommitmentCacheHits, "branch", int64(cacheBranchHits-p.prevCacheBranchHits)) + addVec(mxCommitmentCacheMisses, "account", int64(missAccountCount-p.prevMissAccount)) + addVec(mxCommitmentCacheMisses, "storage", int64(missStorageCount-p.prevMissStorage)) + addVec(mxCommitmentCacheMisses, "branch", int64(missBranchCount-p.prevMissBranch)) + addCounter(mxCommitmentBlocks, committedDiffBlocks) + addCounter(mxCommitmentTxns, int64(te.lastCommittedTxNum.Load()-p.prevCommittedTxNum)) + + p.prevCommitmentKeyCount = roundKeyCount + p.prevCommitmentAccountKeyCount = accountKeyCount + p.prevCommitmentStorageKeyCount = storageKeyCount + p.prevCommitmentAccountReadCount = accountReadCount + p.prevCommitmentStorageReadCount = storageReadCount + p.prevBranchReadCount = branchReadCount + p.prevBranchWriteCount = branchWriteCount + p.prevBranchReadBytes = branchReadBytes + p.prevBranchWriteBytes = branchWriteBytes + p.prevFoldCount = foldCount + p.prevUnfoldCount = unfoldCount + p.prevCacheAccountHits = cacheAccountHits + p.prevCacheStorageHits = cacheStorageHits + p.prevCacheBranchHits = cacheBranchHits + p.prevMissAccount = missAccountCount + p.prevMissStorage = missStorageCount + p.prevMissBranch = missBranchCount totalCacheHits := cacheBranchHits + cacheAccountHits + cacheStorageHits totalCacheMisses := missBranchCount + missAccountCount + missStorageCount From c79ebbab3d3fa166e6c3bc3ab8f60a50ed5344d0 Mon Sep 17 00:00:00 2001 From: awskii Date: Wed, 26 Aug 2026 18:04:28 +0700 Subject: [PATCH 3/5] execution/commitment: publish commitment counters per round from the trie, and port the dashboards --- .../dashboards/erigon_internals.json | 22 +- .../erigon_custom_metrics.internal.json | 24 +-- db/state/execctx/domain_shared.go | 6 +- execution/commitment/commitment.go | 32 ++- execution/commitment/commitment_test.go | 2 +- .../commitmentdb/commitment_context.go | 2 + execution/commitment/hex_patricia_hashed.go | 7 +- execution/commitment/parallel_metrics_test.go | 35 ++++ execution/commitment/parallel_mount.go | 3 + .../commitment/parallel_patricia_hashed.go | 26 +-- execution/commitment/prom_metrics.go | 61 +++++- execution/stagedsync/exec3.go | 2 - execution/stagedsync/exec3_metrics.go | 188 +++--------------- execution/stagedsync/exec3_serial.go | 1 - 14 files changed, 194 insertions(+), 217 deletions(-) diff --git a/cmd/prometheus/dashboards/erigon_internals.json b/cmd/prometheus/dashboards/erigon_internals.json index e7459459dd0..c7625d3900f 100644 --- a/cmd/prometheus/dashboards/erigon_internals.json +++ b/cmd/prometheus/dashboards/erigon_internals.json @@ -5144,7 +5144,7 @@ "targets": [ { "editorMode": "code", - "expr": "commit_key_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_keys_total{instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "total keys", "range": true, "refId": "A", @@ -5155,7 +5155,7 @@ }, { "editorMode": "code", - "expr": "commit_account_key_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_key_traversals_total{kind=\"address\",instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "account keys", "range": true, "refId": "B", @@ -5166,7 +5166,7 @@ }, { "editorMode": "code", - "expr": "commit_storage_key_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_key_traversals_total{kind=\"storage\",instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "storage keys", "range": true, "refId": "C", @@ -5265,7 +5265,7 @@ "targets": [ { "editorMode": "code", - "expr": "commit_fold_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_folds_total{instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "fold", "range": true, "refId": "A", @@ -5276,7 +5276,7 @@ }, { "editorMode": "code", - "expr": "commit_unfold_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_unfolds_total{instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "unfold", "range": true, "refId": "B", @@ -5375,7 +5375,7 @@ "targets": [ { "editorMode": "code", - "expr": "commit_branch_read_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_reads_total{kind=\"branch\",instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "branch reads", "range": true, "refId": "A", @@ -5386,7 +5386,7 @@ }, { "editorMode": "code", - "expr": "commit_branch_write_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_branch_writes_total{instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "branch writes", "range": true, "refId": "B", @@ -5397,7 +5397,7 @@ }, { "editorMode": "code", - "expr": "commit_read_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_reads_total{instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "total reads", "range": true, "refId": "C", @@ -5408,7 +5408,7 @@ }, { "editorMode": "code", - "expr": "commit_account_read_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_reads_total{kind=\"account\",instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "account reads", "range": true, "refId": "D", @@ -5419,7 +5419,7 @@ }, { "editorMode": "code", - "expr": "commit_storage_read_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_reads_total{kind=\"storage\",instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "storage reads", "range": true, "refId": "E", @@ -5535,7 +5535,7 @@ "targets": [ { "editorMode": "code", - "expr": "commit_block_dur{instance=~\"$instance\"}", + "expr": "1e9 * histogram_quantile(0.9, sum by (le) (rate(commitment_round_duration_seconds_bucket{instance=~\"$instance\"}[$__rate_interval])))", "legendFormat": "block dur (ns)", "range": true, "refId": "A", diff --git a/dashboards/erigon_custom_metrics/erigon_custom_metrics.internal.json b/dashboards/erigon_custom_metrics/erigon_custom_metrics.internal.json index 184cc4e86b6..c5caec558f8 100644 --- a/dashboards/erigon_custom_metrics/erigon_custom_metrics.internal.json +++ b/dashboards/erigon_custom_metrics/erigon_custom_metrics.internal.json @@ -5997,7 +5997,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "builder", - "expr": "commit_block_dur{instance=\"$instance\"}", + "expr": "1e9 * histogram_quantile(0.9, sum by (le) (rate(commitment_round_duration_seconds_bucket{instance=\"$instance\"}[$__rate_interval])))", "instant": false, "legendFormat": "commitment: {{instance}}", "range": true, @@ -9290,7 +9290,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "commit_key_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_keys_total{instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "total keys {{instance}}", "range": true, "refId": "A" @@ -9301,7 +9301,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "commit_account_key_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_key_traversals_total{kind=\"address\",instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "account keys {{instance}}", "range": true, "refId": "B" @@ -9312,7 +9312,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "commit_storage_key_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_key_traversals_total{kind=\"storage\",instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "storage keys {{instance}}", "range": true, "refId": "C" @@ -9417,7 +9417,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "commit_fold_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_folds_total{instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "fold {{instance}}", "range": true, "refId": "A" @@ -9428,7 +9428,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "commit_unfold_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_unfolds_total{instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "unfold {{instance}}", "range": true, "refId": "B" @@ -9533,7 +9533,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "commit_branch_read_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_reads_total{kind=\"branch\",instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "reads {{instance}}", "range": true, "refId": "A" @@ -9544,7 +9544,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "commit_branch_write_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_branch_writes_total{instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "writes {{instance}}", "range": true, "refId": "B" @@ -9555,7 +9555,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "commit_read_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_reads_total{instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "total reads {{instance}}", "range": true, "refId": "C" @@ -9566,7 +9566,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "commit_account_read_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_reads_total{kind=\"account\",instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "account reads {{instance}}", "range": true, "refId": "D" @@ -9577,7 +9577,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "commit_storage_read_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_reads_total{kind=\"storage\",instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "storage reads {{instance}}", "range": true, "refId": "E" @@ -9699,7 +9699,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "commit_block_dur{instance=~\"$instance\"}", + "expr": "1e9 * histogram_quantile(0.9, sum by (le) (rate(commitment_round_duration_seconds_bucket{instance=~\"$instance\"}[$__rate_interval])))", "legendFormat": "block dur (ns) {{instance}}", "range": true, "refId": "A" diff --git a/db/state/execctx/domain_shared.go b/db/state/execctx/domain_shared.go index c43838b3fe0..954b9e4b684 100644 --- a/db/state/execctx/domain_shared.go +++ b/db/state/execctx/domain_shared.go @@ -539,7 +539,7 @@ func (sd *SharedDomains) flushPendingUpdates(ctx context.Context, tx kv.Temporal switcher, ok := sd.mem.(changesetSwitcher) if !ok { - _, err := commitment.ApplyDeferredBranchUpdates(upd.Deferred, runtime.NumCPU(), putBranch) + _, err := commitment.ApplyDeferredBranchUpdates(upd.Deferred, runtime.NumCPU(), putBranch, upd.Metrics) return err } @@ -563,7 +563,7 @@ func (sd *SharedDomains) flushPendingUpdates(ctx context.Context, tx kv.Temporal // see concurrency contract on the wrappers above. defer sd.SwapCommitmentDiffLocked(cs)() - if _, err := commitment.ApplyDeferredBranchUpdates(upd.Deferred, runtime.NumCPU(), putBranch); err != nil { + if _, err := commitment.ApplyDeferredBranchUpdates(upd.Deferred, runtime.NumCPU(), putBranch, upd.Metrics); err != nil { return err } @@ -572,7 +572,7 @@ func (sd *SharedDomains) flushPendingUpdates(ctx context.Context, tx kv.Temporal } // No past changeset found — write into whatever is current. - _, err := commitment.ApplyDeferredBranchUpdates(upd.Deferred, runtime.NumCPU(), putBranch) + _, err := commitment.ApplyDeferredBranchUpdates(upd.Deferred, runtime.NumCPU(), putBranch, upd.Metrics) return err } diff --git a/execution/commitment/commitment.go b/execution/commitment/commitment.go index 9db6e1f9e9a..68738e0622f 100644 --- a/execution/commitment/commitment.go +++ b/execution/commitment/commitment.go @@ -280,6 +280,9 @@ type PendingCommitmentUpdate struct { BlockHash common.Hash TxNum uint64 Deferred []*DeferredBranchUpdate + // Metrics is the trie's, carried so the caller-owned apply still counts + // against the round that produced these writes. + Metrics *Metrics } func (p *PendingCommitmentUpdate) Clear() { @@ -366,18 +369,9 @@ func (be *BranchEncoder) ApplyDeferredUpdates( numWorkers int, putBranch func(prefix []byte, data []byte, prevData []byte) error, ) error { - written, err := ApplyDeferredBranchUpdates(be.deferred, numWorkers, putBranch) - if err != nil { + if _, err := ApplyDeferredBranchUpdates(be.deferred, numWorkers, putBranch, be.metrics); err != nil { return err } - if be.metrics != nil { - be.metrics.updateBranch.Add(uint64(written)) - var bytesOut int - for _, upd := range be.deferred { - bytesOut += len(upd.encoded) - } - be.metrics.AddBranchWrite(bytesOut) - } return nil } @@ -386,11 +380,27 @@ var workerMergerPool = sync.Pool{New: func() any { return NewHexBranchMerger(512 // Returns the number of updates written. putBranch must copy prefix and data rather than // retain them: they are pooled and reused for a later, unrelated update. prevData is // cloned per update and carries no such constraint. +// ApplyDeferredBranchUpdates applies the queued branch writes and, when m is +// non-nil, accounts them. Accounting lives here because this is the one place +// every deferred path passes through — including the caller-owned one, which +// applies from SharedDomains long after the trie's round has ended. func ApplyDeferredBranchUpdates( deferred []*DeferredBranchUpdate, numWorkers int, putBranch func(prefix []byte, data []byte, prevData []byte) error, -) (int, error) { + m *Metrics, +) (n int, err error) { + if m != nil { + defer func() { + m.updateBranch.Add(uint64(n)) + // encoded is filled by the merge above, so this only reads after it. + var bytesOut int + for _, upd := range deferred { + bytesOut += len(upd.encoded) + } + m.AddBranchWrite(bytesOut) + }() + } if len(deferred) == 0 { return 0, nil } diff --git a/execution/commitment/commitment_test.go b/execution/commitment/commitment_test.go index 80ad0b64a05..7b1db5e9b3b 100644 --- a/execution/commitment/commitment_test.go +++ b/execution/commitment/commitment_test.go @@ -1105,7 +1105,7 @@ func TestApplyDeferred_CallbackSeesInputDerivedCapacity(t *testing.T) { require.Equal(t, len(data), cap(data), "data carries leftover pool capacity") require.Equal(t, len(prevData), cap(prevData), "prevData carries leftover pool capacity") return nil - }) + }, nil) require.NoError(t, err) require.Equal(t, tc.updates, written) require.Equal(t, tc.updates, seen, "callback must run for every update") diff --git a/execution/commitment/commitmentdb/commitment_context.go b/execution/commitment/commitmentdb/commitment_context.go index 982fb0ab4c4..ff2bd1b5074 100644 --- a/execution/commitment/commitmentdb/commitment_context.go +++ b/execution/commitment/commitmentdb/commitment_context.go @@ -601,6 +601,7 @@ func (sdc *SharedDomainsCommitmentContext) ComputeCommitment(ctx context.Context BlockNum: blockNum, TxNum: txNum, Deferred: trie.TakeDeferredUpdates(), + Metrics: trie.Metrics(), } } case *commitment.ParallelPatriciaHashed: @@ -609,6 +610,7 @@ func (sdc *SharedDomainsCommitmentContext) ComputeCommitment(ctx context.Context BlockNum: blockNum, TxNum: txNum, Deferred: trie.TakeDeferredUpdates(), + Metrics: trie.Metrics(), } } } diff --git a/execution/commitment/hex_patricia_hashed.go b/execution/commitment/hex_patricia_hashed.go index b0a9b47e477..e6745f13264 100644 --- a/execution/commitment/hex_patricia_hashed.go +++ b/execution/commitment/hex_patricia_hashed.go @@ -233,6 +233,10 @@ func newHexPatriciaHashed() *HexPatriciaHashed { return hph } +// Metrics exposes the trie's counters so a caller applying its deferred writes +// can account them against the round that produced them. +func (hph *HexPatriciaHashed) Metrics() *Metrics { return hph.metrics } + // SetCollapseTracer sets a callback that will be invoked when a node collapse occurs // during commitment calculation. This is used by witness generation to capture paths // to HashNodes that need resolution when a FullNode is reduced to a single child. @@ -2485,7 +2489,8 @@ func (hph *HexPatriciaHashed) Process(ctx context.Context, updates *Updates, log hph.metrics.Reset() hph.metrics.updates.Store(updatesCount) hph.metrics.AddRoundKeys(updatesCount) - defer ObserveRound(time.Now()) + roundStart := time.Now() + defer func() { observeRound(hph.metrics, roundStart) }() if hph.metrics.collectCommitmentMetrics { defer func() { hph.metrics.TotalProcessingTimeInc(start) diff --git a/execution/commitment/parallel_metrics_test.go b/execution/commitment/parallel_metrics_test.go index 9ef8ffb5c84..444a1e21191 100644 --- a/execution/commitment/parallel_metrics_test.go +++ b/execution/commitment/parallel_metrics_test.go @@ -130,3 +130,38 @@ func TestRoundKeysAreDistinctNotTraversals(t *testing.T) { "traversals count cell visits, so they never undercount distinct keys") assert.Positive(t, m.BranchWriteBytes, "branch write bytes are counted") } + +// Two rounds on one trie must report the second round's own numbers. Tries come +// from a pool whose Release does not clear counters, and the parallel trie used +// to accumulate across rounds, so both ends of the merge could carry history in. +func TestRoundCountersDoNotAccumulateAcrossRounds(t *testing.T) { + ms := NewMockState(t) + keys, upds := buildNibbleSpread(t, 16, 4) + require.NoError(t, ms.applyPlainUpdates(keys, upds)) + + tr := newParTrie(t, ms, 4) + defer tr.Release() + + round := func() *CommitProgress { + t.Helper() + ut := NewUpdates(ModeParallel, t.TempDir(), KeyToHexNibbleHash) + defer ut.Close() + for _, k := range keys { + ut.TouchPlainKey(string(k), nil, nil) + } + var got *CommitProgress + _, err := tr.Process(context.Background(), ut, "", func(p *CommitProgress) { got = p }, WarmupConfig{}) + require.NoError(t, err) + require.NotNil(t, got) + return got + } + + first := round() + second := round() + + assert.EqualValues(t, len(keys), first.Metrics.RoundKeys) + assert.EqualValues(t, len(keys), second.Metrics.RoundKeys, + "the second round reports its own key count, not the running total") + assert.LessOrEqual(t, second.Metrics.AddressKeys, first.Metrics.AddressKeys*2, + "traversals are per-round; a pooled worker must not carry its last round in") +} diff --git a/execution/commitment/parallel_mount.go b/execution/commitment/parallel_mount.go index eda7cc14e02..ae957ae28a3 100644 --- a/execution/commitment/parallel_mount.go +++ b/execution/commitment/parallel_mount.go @@ -138,6 +138,9 @@ func (p *ParallelPatriciaHashed) processMounted(ctx context.Context, updates *Up ni, ch := nib, child g.Go(func() error { w := NewHexPatriciaHashed(p.accountKeyLen, nil, p.cfg) + // Tries come from a pool and Release does not clear their counters, + // so a checkout carries the previous round's numbers into the merge. + w.metrics.Reset() defer p.metrics.Merge(w.metrics) w.mountTo(base, ni) if p.template != nil && p.template.traceW != nil { diff --git a/execution/commitment/parallel_patricia_hashed.go b/execution/commitment/parallel_patricia_hashed.go index ed89668c763..ade3ea0acd7 100644 --- a/execution/commitment/parallel_patricia_hashed.go +++ b/execution/commitment/parallel_patricia_hashed.go @@ -51,6 +51,9 @@ type ParallelPatriciaHashed struct { func (p *ParallelPatriciaHashed) DeepLocalFolds() uint64 { return p.deepLocalFolds.Load() } +// Metrics exposes the round's counters; see HexPatriciaHashed.Metrics. +func (p *ParallelPatriciaHashed) Metrics() *Metrics { return p.metrics } + func NewParallelPatriciaHashed(ctxFactory TrieContextFactory, accountKeyLen int16, cfg TrieConfig) *ParallelPatriciaHashed { p := &ParallelPatriciaHashed{ template: NewHexPatriciaHashed(accountKeyLen, nil, cfg), @@ -222,6 +225,13 @@ func (p *ParallelPatriciaHashed) Process( p.rootHash.Store(nil) p.deepLocalFolds.Store(0) + // Per-round, matching HexPatriciaHashed.Process: the counters published for + // a round have to describe that round alone. + p.metrics.Reset() + p.metrics.AddRoundKeys(updates.Size()) + roundStart := time.Now() + defer func() { observeRound(p.metrics, roundStart) }() + pu := updates.parallel if pu.trie == nil || pu.trie.root == nil || pu.trie.root.subtreeCount == 0 { // A consumed (or never-touched) collection must return the carried root; folding @@ -233,9 +243,6 @@ func (p *ParallelPatriciaHashed) Process( return rh, nil } - defer ObserveRound(time.Now()) - p.metrics.AddRoundKeys(updates.Size()) - rh, mErr := p.processMounted(ctx, updates) if mErr != nil { pu.deferredMu.Lock() @@ -323,19 +330,8 @@ func (p *ParallelPatriciaHashed) applyDeferredUpdates(ctx context.Context, pu *p // This path calls PutBranch directly rather than through a BranchEncoder, // so it is the only place the parallel engine's branch writes get counted. - n, err := ApplyDeferredBranchUpdates(deferred, p.numWorkers, applyCtx.PutBranch) - if err != nil { + if _, err := ApplyDeferredBranchUpdates(deferred, p.numWorkers, applyCtx.PutBranch, p.metrics); err != nil { return fmt.Errorf("apply deferred branch updates: %w", err) } - if p.metrics != nil { - p.metrics.updateBranch.Add(uint64(n)) - // encoded is filled by the merge inside the apply, so this only sums - // after it returns; the pool recycles these on the deferred cleanup. - var bytesOut int - for _, upd := range deferred { - bytesOut += len(upd.encoded) - } - p.metrics.AddBranchWrite(bytesOut) - } return nil } diff --git a/execution/commitment/prom_metrics.go b/execution/commitment/prom_metrics.go index 48516b0d044..e5765d6ceb6 100644 --- a/execution/commitment/prom_metrics.go +++ b/execution/commitment/prom_metrics.go @@ -22,19 +22,70 @@ import ( "github.com/erigontech/erigon/diagnostics/metrics" ) -// Round-level commitment metrics. Counters and a histogram rather than gauges -// holding pre-divided rates: rate() belongs in the query, so the window stays -// the reader's choice and a missed scrape costs a sample instead of an interval. +// Commitment metrics are counters and a histogram rather than gauges holding a +// pre-divided rate: rate() belongs in the query, so the averaging window stays +// the reader's choice and a missed scrape costs a sample, not an interval. +// +// They are emitted per round, from the trie itself, because a round's Metrics +// is per-round on both engines and there is no cumulative series to difference: +// HexPatriciaHashed.Process resets at the top of every round, and the parallel +// trie now does the same. Publishing from the executor's log ticker instead +// would both mis-difference that snapshot and drop every round between ticks. var ( mxRounds = metrics.GetOrCreateCounter("commitment_rounds_total") // Buckets span a fast incremental block through a whale fold. mxRoundDuration = metrics.NewHistogram("commitment_round_duration_seconds", []float64{0.001, 0.005, 0.01, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60}) + + mxKeys = metrics.GetOrCreateCounter("commitment_keys_total") + mxFolds = metrics.GetOrCreateCounter("commitment_folds_total") + mxUnfolds = metrics.GetOrCreateCounter("commitment_unfolds_total") + mxBranchPuts = metrics.GetOrCreateCounter("commitment_branch_writes_total") + mxReadBytes = metrics.GetOrCreateCounter("commitment_branch_read_bytes_total") + mxWriteBytes = metrics.GetOrCreateCounter("commitment_branch_write_bytes_total") + + // kind=address|storage. Cell traversals, not distinct keys: the parallel + // engine re-walks subtrees on mount+replay, so this exceeds commitment_keys_total. + mxTraversals = metrics.GetOrCreateCounterVec("commitment_key_traversals_total", + []string{"kind"}, "cell traversals during commitment, by key kind") + + // kind=account|storage|branch + mxReads = metrics.GetOrCreateCounterVec("commitment_reads_total", + []string{"kind"}, "PatriciaContext reads during commitment") ) -// ObserveRound records one finished commitment round. -func ObserveRound(start time.Time) { +func addU64(c metrics.Counter, v uint64) { + if v > 0 { + c.AddUint64(v) + } +} + +func addVec(v *metrics.CounterVec, kind string, n uint64) { + if n > 0 { + v.WithLabelValues(kind).Add(float64(n)) + } +} + +// observeRound publishes one finished round. Called from Trie.Process on both +// engines, after the final fold and the deferred apply, so the counts include +// the work those do. +func observeRound(m *Metrics, start time.Time) { mxRounds.Inc() mxRoundDuration.ObserveDuration(start) + if m == nil { + return + } + v := m.AsValues() + addU64(mxKeys, v.RoundKeys) + addU64(mxFolds, v.Folds) + addU64(mxUnfolds, v.Unfolds) + addU64(mxBranchPuts, v.UpdateBranch) + addU64(mxReadBytes, v.BranchReadBytes) + addU64(mxWriteBytes, v.BranchWriteBytes) + addVec(mxTraversals, "address", v.AddressKeys) + addVec(mxTraversals, "storage", v.StorageKeys) + addVec(mxReads, "account", v.LoadAccount) + addVec(mxReads, "storage", v.LoadStorage) + addVec(mxReads, "branch", v.LoadBranch) } diff --git a/execution/stagedsync/exec3.go b/execution/stagedsync/exec3.go index cce54d21e4c..5f10da51c61 100644 --- a/execution/stagedsync/exec3.go +++ b/execution/stagedsync/exec3.go @@ -171,7 +171,6 @@ func execV3(ctx context.Context, logEvery := time.NewTicker(20 * time.Second) defer logEvery.Stop() defer resetExecGauges(ctx) - defer resetCommitmentGauges(ctx) defer resetDomainGauges(ctx) stepsInDb := rawdbhelpers.IdxStepsCountV3(applyTx, doms.StepSize()) @@ -299,7 +298,6 @@ func execV3Serial(ctx context.Context, logEvery := time.NewTicker(20 * time.Second) defer logEvery.Stop() defer resetExecGauges(ctx) - defer resetCommitmentGauges(ctx) defer resetDomainGauges(ctx) stepsInDb := rawdbhelpers.IdxStepsCountV3(applyTx, doms.StepSize()) diff --git a/execution/stagedsync/exec3_metrics.go b/execution/stagedsync/exec3_metrics.go index 908fcc325d3..907e97205f9 100644 --- a/execution/stagedsync/exec3_metrics.go +++ b/execution/stagedsync/exec3_metrics.go @@ -106,26 +106,6 @@ var ( mxExecCodeDomainFileReads = metrics.NewGauge(`exec_domain_file_read_rate{domain="code"}`) mxExecCodeDomainFileReadDuration = metrics.NewGauge(`exec_domain_file_read_dur{domain="code"}`) - // Commitment counters. rate() belongs in the query, so these are monotonic - // totals rather than gauges holding a pre-divided rate over the log interval. - mxCommitmentBlocks = metrics.GetOrCreateCounter("commitment_blocks_total") - mxCommitmentTxns = metrics.GetOrCreateCounter("commitment_txns_total") - mxCommitmentKeys = metrics.GetOrCreateCounter("commitment_keys_total") - mxCommitmentFolds = metrics.GetOrCreateCounter("commitment_folds_total") - mxCommitmentUnfolds = metrics.GetOrCreateCounter("commitment_unfolds_total") - mxCommitmentBranchPuts = metrics.GetOrCreateCounter("commitment_branch_writes_total") - mxCommitmentReadBytes = metrics.GetOrCreateCounter("commitment_branch_read_bytes_total") - mxCommitmentWriteBytes = metrics.GetOrCreateCounter("commitment_branch_write_bytes_total") - - // kind=address|storage. Traversals, not distinct keys: the parallel engine - // re-walks subtrees on mount+replay, so this exceeds commitment_keys_total. - mxCommitmentTraversals = metrics.GetOrCreateCounterVec("commitment_key_traversals_total", - []string{"kind"}, "cell traversals during commitment, by key kind") - - // kind=account|storage|branch - mxCommitmentReads = metrics.GetOrCreateCounterVec("commitment_reads_total", []string{"kind"}, "PatriciaContext reads during commitment") - mxCommitmentCacheHits = metrics.GetOrCreateCounterVec("commitment_cache_hits_total", []string{"kind"}, "commitment cache hits") - mxCommitmentCacheMisses = metrics.GetOrCreateCounterVec("commitment_cache_misses_total", []string{"kind"}, "commitment cache misses") mxCommitmentDomainReads = metrics.NewGauge(`exec_domain_read_rate{domain="commitment"}`) mxCommitmentDomainReadDuration = metrics.NewGauge(`exec_domain_read_dur{domain="commitment"}`) mxCommitmentDomainCacheReads = metrics.NewGauge(`exec_domain_cache_read_rate{domain="commitment"}`) @@ -182,7 +162,6 @@ func (g *gaugeResetTask) reset() { } var execResetTask = gaugeResetTask{} -var commitResetTask = gaugeResetTask{} var domainResetTask = gaugeResetTask{} // enough time to alow the sampler to scrape @@ -211,22 +190,6 @@ func resetExecGauges(ctx context.Context) { } } -func resetCommitmentGauges(ctx context.Context) { - commitResetTask.Lock() - defer commitResetTask.Unlock() - if commitResetTask.Timer != nil { - if commitResetTask.stopped { - commitResetTask.Timer = time.NewTimer(resetDelay) - } else { - commitResetTask.Reset(resetDelay) - } - } else { - commitResetTask.Timer = time.NewTimer(resetDelay) - commitResetTask.ctx = ctx - commitResetTask.run(ctx) - } -} - func resetDomainGauges(ctx context.Context) { domainResetTask.Lock() defer domainResetTask.Unlock() @@ -252,7 +215,7 @@ func resetDomainGauges(ctx context.Context) { mxExecDomainPutKeySize, mxExecDomainPutValueSize, mxExecAccountDomainPutRate, mxExecAccountDomainPutSize, mxExecAccountDomainPutKeySize, mxExecAccountDomainPutValueSize, mxExecStorageDomainPutRate, mxExecStorageDomainPutSize, mxExecStorageDomainPutKeySize, mxExecStorageDomainPutValueSize, mxExecCodeDomainPutRate, mxExecCodeDomainPutSize, - mxExecCodeDomainPutKeySize, + mxExecCodeDomainPutKeySize, mxExecCodeDomainPutValueSize, mxCommitmentDomainReads, mxCommitmentDomainReadDuration, mxCommitmentDomainCacheReads, mxCommitmentDomainCacheReadDuration, mxCommitmentDomainDbReads, mxCommitmentDomainDbReadDuration, mxCommitmentDomainFileReads, mxCommitmentDomainFileReadDuration, mxCommitmentDomainPutRate, @@ -466,55 +429,38 @@ func NewProgress(initialBlockNum, initialTxNum, commitThreshold uint64, logPrefi } type Progress struct { - initialTime time.Time - initialTxNum uint64 - initialBlockNum uint64 - prevExecTime time.Time - prevExecutedBlockNum int64 - prevExecutedTxNum uint64 - prevExecutedGas int64 - prevExecCount uint64 - prevActivations int64 - prevTaskDuration time.Duration - prevTaskReadDuration time.Duration - prevAccountReadDuration time.Duration - prevStorageReadDuration time.Duration - prevCodeReadDuration time.Duration - prevTaskGas int64 - prevBlockCount int64 - prevBlockDuration time.Duration - prevAbortCount uint64 - prevInvalidCount uint64 - prevReadCount int64 - prevAccountReadCount int64 - prevStorageReadCount int64 - prevCodeReadCount int64 - prevWriteCount uint64 - prevCommitTime time.Time - prevCommittedBlockNum uint64 - prevCommittedTxNum uint64 - prevCommitLogGas int64 - prevCommitmentKeyCount uint64 - prevCommitmentAccountKeyCount uint64 - prevCommitmentStorageKeyCount uint64 - prevCommitmentAccountReadCount uint64 - prevCommitmentStorageReadCount uint64 - prevBranchReadCount uint64 - prevBranchWriteCount uint64 - prevBranchReadBytes uint64 - prevBranchWriteBytes uint64 - prevFoldCount uint64 - prevUnfoldCount uint64 - prevCacheAccountHits uint64 - prevCacheStorageHits uint64 - prevCacheBranchHits uint64 - prevMissAccount uint64 - prevMissStorage uint64 - prevMissBranch uint64 - commitThreshold uint64 - prevDomainMetrics *kvmetrics.DomainMetrics - logPrefix string - logger log.Logger + initialTime time.Time + initialTxNum uint64 + initialBlockNum uint64 + prevExecTime time.Time + prevExecutedBlockNum int64 + prevExecutedTxNum uint64 + prevExecutedGas int64 + prevExecCount uint64 + prevActivations int64 + prevTaskDuration time.Duration + prevTaskReadDuration time.Duration + prevAccountReadDuration time.Duration + prevStorageReadDuration time.Duration + prevCodeReadDuration time.Duration + prevTaskGas int64 + prevBlockCount int64 + prevBlockDuration time.Duration + prevAbortCount uint64 + prevInvalidCount uint64 + prevReadCount int64 + prevAccountReadCount int64 + prevStorageReadCount int64 + prevCodeReadCount int64 + prevWriteCount uint64 + prevCommitTime time.Time + prevCommittedBlockNum uint64 + prevCommittedTxNum uint64 + prevCommitLogGas int64 + commitThreshold uint64 + prevDomainMetrics *kvmetrics.DomainMetrics + logPrefix string + logger log.Logger } type executor interface { @@ -787,82 +733,14 @@ func (p *Progress) LogCommitments(rs *state.StateV3, ex executor, stepsInDb floa } lastProgress.Metrics.RLock() - accountKeyCount := lastProgress.Metrics.AddressKeys - storageKeyCount := lastProgress.Metrics.StorageKeys - accountReadCount := lastProgress.Metrics.LoadAccount - storageReadCount := lastProgress.Metrics.LoadStorage - branchReadCount := lastProgress.Metrics.LoadBranch - branchWriteCount := lastProgress.Metrics.UpdateBranch cacheBranchHits := lastProgress.Metrics.CacheBranch cacheAccountHits := lastProgress.Metrics.CacheAccount cacheStorageHits := lastProgress.Metrics.CacheStorage missBranchCount := lastProgress.Metrics.MissBranch missAccountCount := lastProgress.Metrics.MissAccount missStorageCount := lastProgress.Metrics.MissStorage - roundKeyCount := lastProgress.Metrics.RoundKeys - branchReadBytes := lastProgress.Metrics.BranchReadBytes - branchWriteBytes := lastProgress.Metrics.BranchWriteBytes - foldCount := lastProgress.Metrics.Folds - unfoldCount := lastProgress.Metrics.Unfolds lastProgress.Metrics.RUnlock() - curAccountKeyCount := int64(accountKeyCount - p.prevCommitmentAccountKeyCount) - curStorageKeyCount := int64(storageKeyCount - p.prevCommitmentStorageKeyCount) - curAccountReadCount := int64(accountReadCount - p.prevCommitmentAccountReadCount) - curStorageReadCount := int64(storageReadCount - p.prevCommitmentStorageReadCount) - curBranchReadCount := int64(branchReadCount - p.prevBranchReadCount) - curBranchWriteCount := int64(branchWriteCount - p.prevBranchWriteCount) - - // The trie's counters are cumulative; Prometheus wants the increment. - addCounter := func(c metrics.Counter, delta int64) { - if delta > 0 { - c.AddUint64(uint64(delta)) - } - } - addVec := func(v *metrics.CounterVec, kind string, delta int64) { - if delta > 0 { - v.WithLabelValues(kind).Add(float64(delta)) - } - } - - addCounter(mxCommitmentKeys, int64(roundKeyCount-p.prevCommitmentKeyCount)) - addVec(mxCommitmentTraversals, "address", curAccountKeyCount) - addVec(mxCommitmentTraversals, "storage", curStorageKeyCount) - addVec(mxCommitmentReads, "account", curAccountReadCount) - addVec(mxCommitmentReads, "storage", curStorageReadCount) - addVec(mxCommitmentReads, "branch", curBranchReadCount) - addCounter(mxCommitmentBranchPuts, curBranchWriteCount) - addCounter(mxCommitmentReadBytes, int64(branchReadBytes-p.prevBranchReadBytes)) - addCounter(mxCommitmentWriteBytes, int64(branchWriteBytes-p.prevBranchWriteBytes)) - addCounter(mxCommitmentFolds, int64(foldCount-p.prevFoldCount)) - addCounter(mxCommitmentUnfolds, int64(unfoldCount-p.prevUnfoldCount)) - addVec(mxCommitmentCacheHits, "account", int64(cacheAccountHits-p.prevCacheAccountHits)) - addVec(mxCommitmentCacheHits, "storage", int64(cacheStorageHits-p.prevCacheStorageHits)) - addVec(mxCommitmentCacheHits, "branch", int64(cacheBranchHits-p.prevCacheBranchHits)) - addVec(mxCommitmentCacheMisses, "account", int64(missAccountCount-p.prevMissAccount)) - addVec(mxCommitmentCacheMisses, "storage", int64(missStorageCount-p.prevMissStorage)) - addVec(mxCommitmentCacheMisses, "branch", int64(missBranchCount-p.prevMissBranch)) - addCounter(mxCommitmentBlocks, committedDiffBlocks) - addCounter(mxCommitmentTxns, int64(te.lastCommittedTxNum.Load()-p.prevCommittedTxNum)) - - p.prevCommitmentKeyCount = roundKeyCount - p.prevCommitmentAccountKeyCount = accountKeyCount - p.prevCommitmentStorageKeyCount = storageKeyCount - p.prevCommitmentAccountReadCount = accountReadCount - p.prevCommitmentStorageReadCount = storageReadCount - p.prevBranchReadCount = branchReadCount - p.prevBranchWriteCount = branchWriteCount - p.prevBranchReadBytes = branchReadBytes - p.prevBranchWriteBytes = branchWriteBytes - p.prevFoldCount = foldCount - p.prevUnfoldCount = unfoldCount - p.prevCacheAccountHits = cacheAccountHits - p.prevCacheStorageHits = cacheStorageHits - p.prevCacheBranchHits = cacheBranchHits - p.prevMissAccount = missAccountCount - p.prevMissStorage = missStorageCount - p.prevMissBranch = missBranchCount - totalCacheHits := cacheBranchHits + cacheAccountHits + cacheStorageHits totalCacheMisses := missBranchCount + missAccountCount + missStorageCount diff --git a/execution/stagedsync/exec3_serial.go b/execution/stagedsync/exec3_serial.go index 4bebda74da6..d253a5834be 100644 --- a/execution/stagedsync/exec3_serial.go +++ b/execution/stagedsync/exec3_serial.go @@ -251,7 +251,6 @@ func (se *serialExecutor) exec(ctx context.Context, execStage *StageState, u Unw if !ok { return b.HeaderNoCopy(), rwTx, nil } - resetCommitmentGauges(ctx) se.txExecutor.lastCommittedBlockNum.Store(b.NumberU64()) se.txExecutor.lastCommittedTxNum.Store(inputTxNum) se.logger.Info( From 3515aa20563c0d2ce0958a1374114076cbbe078d Mon Sep 17 00:00:00 2001 From: awskii Date: Wed, 26 Aug 2026 18:38:06 +0700 Subject: [PATCH 4/5] execution/commitment: bill branch writes where they land, and fix the ported dashboard queries --- .../dashboards/erigon_internals.json | 4 +- .../erigon_custom_metrics.internal.json | 6 +-- execution/commitment/commitment.go | 43 ++++++++----------- execution/commitment/metrics.go | 7 +-- execution/commitment/parallel_metrics_test.go | 5 ++- execution/commitment/parallel_mount.go | 13 ++++-- execution/commitment/prom_metrics.go | 23 ++++++++-- 7 files changed, 61 insertions(+), 40 deletions(-) diff --git a/cmd/prometheus/dashboards/erigon_internals.json b/cmd/prometheus/dashboards/erigon_internals.json index c7625d3900f..94c79ebc4be 100644 --- a/cmd/prometheus/dashboards/erigon_internals.json +++ b/cmd/prometheus/dashboards/erigon_internals.json @@ -5397,7 +5397,7 @@ }, { "editorMode": "code", - "expr": "rate(commitment_reads_total{instance=~\"$instance\"}[$__rate_interval])", + "expr": "sum without (kind) (rate(commitment_reads_total{instance=~\"$instance\"}[$__rate_interval]))", "legendFormat": "total reads", "range": true, "refId": "C", @@ -5535,7 +5535,7 @@ "targets": [ { "editorMode": "code", - "expr": "1e9 * histogram_quantile(0.9, sum by (le) (rate(commitment_round_duration_seconds_bucket{instance=~\"$instance\"}[$__rate_interval])))", + "expr": "1e9 * histogram_quantile(0.9, sum by (le, instance) (rate(commitment_round_duration_seconds_bucket{instance=~\"$instance\"}[$__rate_interval])))", "legendFormat": "block dur (ns)", "range": true, "refId": "A", diff --git a/dashboards/erigon_custom_metrics/erigon_custom_metrics.internal.json b/dashboards/erigon_custom_metrics/erigon_custom_metrics.internal.json index c5caec558f8..1025f8b9b21 100644 --- a/dashboards/erigon_custom_metrics/erigon_custom_metrics.internal.json +++ b/dashboards/erigon_custom_metrics/erigon_custom_metrics.internal.json @@ -5997,7 +5997,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "builder", - "expr": "1e9 * histogram_quantile(0.9, sum by (le) (rate(commitment_round_duration_seconds_bucket{instance=\"$instance\"}[$__rate_interval])))", + "expr": "1e9 * histogram_quantile(0.9, sum by (le, instance) (rate(commitment_round_duration_seconds_bucket{instance=~\"$instance\"}[$__rate_interval])))", "instant": false, "legendFormat": "commitment: {{instance}}", "range": true, @@ -9555,7 +9555,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "rate(commitment_reads_total{instance=~\"$instance\"}[$__rate_interval])", + "expr": "sum without (kind) (rate(commitment_reads_total{instance=~\"$instance\"}[$__rate_interval]))", "legendFormat": "total reads {{instance}}", "range": true, "refId": "C" @@ -9699,7 +9699,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "1e9 * histogram_quantile(0.9, sum by (le) (rate(commitment_round_duration_seconds_bucket{instance=~\"$instance\"}[$__rate_interval])))", + "expr": "1e9 * histogram_quantile(0.9, sum by (le, instance) (rate(commitment_round_duration_seconds_bucket{instance=~\"$instance\"}[$__rate_interval])))", "legendFormat": "block dur (ns) {{instance}}", "range": true, "refId": "A" diff --git a/execution/commitment/commitment.go b/execution/commitment/commitment.go index 68738e0622f..4b8f13ce8bc 100644 --- a/execution/commitment/commitment.go +++ b/execution/commitment/commitment.go @@ -377,30 +377,21 @@ func (be *BranchEncoder) ApplyDeferredUpdates( var workerMergerPool = sync.Pool{New: func() any { return NewHexBranchMerger(512) }} -// Returns the number of updates written. putBranch must copy prefix and data rather than -// retain them: they are pooled and reused for a later, unrelated update. prevData is -// cloned per update and carries no such constraint. -// ApplyDeferredBranchUpdates applies the queued branch writes and, when m is -// non-nil, accounts them. Accounting lives here because this is the one place -// every deferred path passes through — including the caller-owned one, which -// applies from SharedDomains long after the trie's round has ended. +// ApplyDeferredBranchUpdates applies the queued branch writes and returns how many +// were written. Writes are published to the branch-write counters as they land, +// not against a round: the caller-owned path applies from SharedDomains after the +// producing round has already closed, so there is no round left to bill. m, when +// non-nil, additionally carries them into that trie's log and CSV counters. +// +// putBranch must copy prefix and data rather than retain them: they are pooled and +// reused for a later, unrelated update. prevData is cloned per update and carries +// no such constraint. func ApplyDeferredBranchUpdates( deferred []*DeferredBranchUpdate, numWorkers int, putBranch func(prefix []byte, data []byte, prevData []byte) error, m *Metrics, ) (n int, err error) { - if m != nil { - defer func() { - m.updateBranch.Add(uint64(n)) - // encoded is filled by the merge above, so this only reads after it. - var bytesOut int - for _, upd := range deferred { - bytesOut += len(upd.encoded) - } - m.AddBranchWrite(bytesOut) - }() - } if len(deferred) == 0 { return 0, nil } @@ -412,20 +403,24 @@ func ApplyDeferredBranchUpdates( merger := workerMergerPool.Get().(*BranchMerger) defer workerMergerPool.Put(merger) - var written int + var written, bytesOut int for _, upd := range deferred { if err := mergeDeferredUpdate(upd, merger); err != nil { + publishBranchWrites(written, bytesOut, m) return written, err } if upd.encoded == nil { continue } if err := putBranch(capLen(upd.prefix), capLen(upd.encoded), capLen(upd.prev)); err != nil { + publishBranchWrites(written, bytesOut, m) return written, err } written++ + bytesOut += len(upd.encoded) } mxTrieBranchesUpdated.AddInt(written) + publishBranchWrites(written, bytesOut, m) return written, nil } @@ -457,17 +452,20 @@ func ApplyDeferredBranchUpdates( } } - var written int + var written, bytesOut int for _, upd := range deferred { if upd.encoded == nil { continue } if err := putBranch(capLen(upd.prefix), capLen(upd.encoded), capLen(upd.prev)); err != nil { + publishBranchWrites(written, bytesOut, m) return written, err } written++ + bytesOut += len(upd.encoded) } mxTrieBranchesUpdated.AddInt(written) + publishBranchWrites(written, bytesOut, m) return written, nil } @@ -515,10 +513,7 @@ func (be *BranchEncoder) CollectUpdate( if err := ctx.PutBranch(prefixCopy, updateCopy, prev); err != nil { return err } - if be.metrics != nil { - be.metrics.updateBranch.Add(1) - be.metrics.AddBranchWrite(len(updateCopy)) - } + publishBranchWrites(1, len(updateCopy), be.metrics) mxTrieBranchesUpdated.Inc() return nil } diff --git a/execution/commitment/metrics.go b/execution/commitment/metrics.go index 08d7f68a0d6..c18223272cd 100644 --- a/execution/commitment/metrics.go +++ b/execution/commitment/metrics.go @@ -73,9 +73,10 @@ type MetricValues struct { LoadDepths [10]uint64 Unfolds uint64 Folds uint64 - // RoundKeys counts distinct keys handed to the trie, summed over rounds. - // AddressKeys/StorageKeys count cell traversals instead, which the parallel - // engine inflates by re-walking subtrees on mount+replay. + // RoundKeys is the distinct key count of one round — Process resets the + // counters on both engines, so nothing here spans rounds. AddressKeys and + // StorageKeys count cell traversals instead, which the parallel engine + // inflates by re-walking subtrees on mount+replay. RoundKeys uint64 BranchReadBytes uint64 BranchWriteBytes uint64 diff --git a/execution/commitment/parallel_metrics_test.go b/execution/commitment/parallel_metrics_test.go index 444a1e21191..77b45258887 100644 --- a/execution/commitment/parallel_metrics_test.go +++ b/execution/commitment/parallel_metrics_test.go @@ -162,6 +162,9 @@ func TestRoundCountersDoNotAccumulateAcrossRounds(t *testing.T) { assert.EqualValues(t, len(keys), first.Metrics.RoundKeys) assert.EqualValues(t, len(keys), second.Metrics.RoundKeys, "the second round reports its own key count, not the running total") - assert.LessOrEqual(t, second.Metrics.AddressKeys, first.Metrics.AddressKeys*2, + // Strictly less than 2x: with the pooled reset removed a worker enters the + // second round holding the first's traversals and adds its own, landing on + // exactly 2x — which an inclusive bound would admit. + assert.Less(t, second.Metrics.AddressKeys, first.Metrics.AddressKeys*2, "traversals are per-round; a pooled worker must not carry its last round in") } diff --git a/execution/commitment/parallel_mount.go b/execution/commitment/parallel_mount.go index ae957ae28a3..d0ab3715a4d 100644 --- a/execution/commitment/parallel_mount.go +++ b/execution/commitment/parallel_mount.go @@ -141,7 +141,12 @@ func (p *ParallelPatriciaHashed) processMounted(ctx context.Context, updates *Up // Tries come from a pool and Release does not clear their counters, // so a checkout carries the previous round's numbers into the merge. w.metrics.Reset() - defer p.metrics.Merge(w.metrics) + // Merge before releasing: Release pools the trie, after which another + // goroutine may check it out and write these same counters. + release := func() { + p.metrics.Merge(w.metrics) + w.Release() + } w.mountTo(base, ni) if p.template != nil && p.template.traceW != nil { w.traceW = tracePrefix(p.template.traceW, fmt.Sprintf("[mnt %x] ", ni)) @@ -172,7 +177,7 @@ func (p *ParallelPatriciaHashed) processMounted(ctx context.Context, updates *Up return sr, err }) if buildErr != nil { - w.Release() + release() return fmt.Errorf("mount[%x] build: %w", ni, buildErr) } var tf time.Time @@ -185,7 +190,7 @@ func (p *ParallelPatriciaHashed) processMounted(ctx context.Context, updates *Up foldDur[ni] = time.Since(tf) } if err != nil { - w.Release() + release() return fmt.Errorf("mount[%x] fold: %w", ni, err) } cells[ni] = c @@ -193,7 +198,7 @@ func (p *ParallelPatriciaHashed) processMounted(ctx context.Context, updates *Up if deferred := w.TakeDeferredUpdates(); len(deferred) > 0 { pu.appendDeferred(deferred) } - w.Release() + release() return nil }) childIdx++ diff --git a/execution/commitment/prom_metrics.go b/execution/commitment/prom_metrics.go index e5765d6ceb6..21c280300e4 100644 --- a/execution/commitment/prom_metrics.go +++ b/execution/commitment/prom_metrics.go @@ -67,9 +67,26 @@ func addVec(v *metrics.CounterVec, kind string, n uint64) { } } -// observeRound publishes one finished round. Called from Trie.Process on both -// engines, after the final fold and the deferred apply, so the counts include -// the work those do. +// publishBranchWrites bills n branch writes of bytesOut bytes. Called where the +// write lands rather than at a round boundary, because deferred writes can be +// applied after their round has closed. m, when non-nil, also gets them for the +// trie's own log and CSV counters. +func publishBranchWrites(n, bytesOut int, m *Metrics) { + if n <= 0 { + return + } + mxBranchPuts.AddInt(n) + if bytesOut > 0 { + mxWriteBytes.AddInt(bytesOut) + } + if m != nil { + m.updateBranch.Add(uint64(n)) + m.AddBranchWrite(bytesOut) + } +} + +// observeRound publishes one finished round. Branch writes are not published +// here — publishBranchWrites bills those where they land. func observeRound(m *Metrics, start time.Time) { mxRounds.Inc() mxRoundDuration.ObserveDuration(start) From 4a9ad204258b135da4314cc6433d54f7b8b13815 Mon Sep 17 00:00:00 2001 From: awskii Date: Wed, 26 Aug 2026 18:54:13 +0700 Subject: [PATCH 5/5] execution/commitment: stop observeRound republishing branch writes already billed at the write site --- .../dashboards/erigon_internals.json | 2 +- execution/commitment/commitment.go | 5 +-- execution/commitment/hex_patricia_hashed.go | 2 +- execution/commitment/parallel_metrics_test.go | 31 +++++++++++++++++++ execution/commitment/prom_metrics.go | 2 -- 5 files changed, 36 insertions(+), 6 deletions(-) diff --git a/cmd/prometheus/dashboards/erigon_internals.json b/cmd/prometheus/dashboards/erigon_internals.json index 94c79ebc4be..d176e6d52ad 100644 --- a/cmd/prometheus/dashboards/erigon_internals.json +++ b/cmd/prometheus/dashboards/erigon_internals.json @@ -5536,7 +5536,7 @@ { "editorMode": "code", "expr": "1e9 * histogram_quantile(0.9, sum by (le, instance) (rate(commitment_round_duration_seconds_bucket{instance=~\"$instance\"}[$__rate_interval])))", - "legendFormat": "block dur (ns)", + "legendFormat": "block dur (ns) {{instance}}", "range": true, "refId": "A", "datasource": { diff --git a/execution/commitment/commitment.go b/execution/commitment/commitment.go index 4b8f13ce8bc..d7b76d293ee 100644 --- a/execution/commitment/commitment.go +++ b/execution/commitment/commitment.go @@ -280,8 +280,9 @@ type PendingCommitmentUpdate struct { BlockHash common.Hash TxNum uint64 Deferred []*DeferredBranchUpdate - // Metrics is the trie's, carried so the caller-owned apply still counts - // against the round that produced these writes. + // Metrics is the producing trie's, carried so the later apply still reaches + // that trie's log and CSV counters. The Prometheus counters do not depend on + // it — publishBranchWrites bills those where the write lands. Metrics *Metrics } diff --git a/execution/commitment/hex_patricia_hashed.go b/execution/commitment/hex_patricia_hashed.go index e6745f13264..8aceffb6817 100644 --- a/execution/commitment/hex_patricia_hashed.go +++ b/execution/commitment/hex_patricia_hashed.go @@ -234,7 +234,7 @@ func newHexPatriciaHashed() *HexPatriciaHashed { } // Metrics exposes the trie's counters so a caller applying its deferred writes -// can account them against the round that produced them. +// can carry them into this trie's log and CSV totals. func (hph *HexPatriciaHashed) Metrics() *Metrics { return hph.metrics } // SetCollapseTracer sets a callback that will be invoked when a node collapse occurs diff --git a/execution/commitment/parallel_metrics_test.go b/execution/commitment/parallel_metrics_test.go index 77b45258887..07fe71b722e 100644 --- a/execution/commitment/parallel_metrics_test.go +++ b/execution/commitment/parallel_metrics_test.go @@ -168,3 +168,34 @@ func TestRoundCountersDoNotAccumulateAcrossRounds(t *testing.T) { assert.Less(t, second.Metrics.AddressKeys, first.Metrics.AddressKeys*2, "traversals are per-round; a pooled worker must not carry its last round in") } + +// A branch write must reach the Prometheus counter exactly once. Billing it both +// where it lands and again from the round's snapshot is invisible in the trie's +// own MetricValues, so this reads the published counter instead. +func TestBranchWritesArePublishedOnce(t *testing.T) { + ms := NewMockState(t) + keys, upds := buildNibbleSpread(t, 16, 4) + require.NoError(t, ms.applyPlainUpdates(keys, upds)) + + tr := newParTrie(t, ms, 4) + defer tr.Release() + ut := NewUpdates(ModeParallel, t.TempDir(), KeyToHexNibbleHash) + defer ut.Close() + for _, k := range keys { + ut.TouchPlainKey(string(k), nil, nil) + } + + beforePuts := mxBranchPuts.GetValueUint64() + beforeBytes := mxWriteBytes.GetValueUint64() + + var got *CommitProgress + _, err := tr.Process(context.Background(), ut, "", func(p *CommitProgress) { got = p }, WarmupConfig{}) + require.NoError(t, err) + require.NotNil(t, got) + + require.Positive(t, got.Metrics.UpdateBranch, "the round wrote branches at all") + assert.EqualValues(t, got.Metrics.UpdateBranch, mxBranchPuts.GetValueUint64()-beforePuts, + "commitment_branch_writes_total counts each write once") + assert.EqualValues(t, got.Metrics.BranchWriteBytes, mxWriteBytes.GetValueUint64()-beforeBytes, + "commitment_branch_write_bytes_total counts each write once") +} diff --git a/execution/commitment/prom_metrics.go b/execution/commitment/prom_metrics.go index 21c280300e4..a507343ab3b 100644 --- a/execution/commitment/prom_metrics.go +++ b/execution/commitment/prom_metrics.go @@ -97,9 +97,7 @@ func observeRound(m *Metrics, start time.Time) { addU64(mxKeys, v.RoundKeys) addU64(mxFolds, v.Folds) addU64(mxUnfolds, v.Unfolds) - addU64(mxBranchPuts, v.UpdateBranch) addU64(mxReadBytes, v.BranchReadBytes) - addU64(mxWriteBytes, v.BranchWriteBytes) addVec(mxTraversals, "address", v.AddressKeys) addVec(mxTraversals, "storage", v.StorageKeys) addVec(mxReads, "account", v.LoadAccount)