From 4b0f1996e7cc15501f2b540378980f246e1b9be1 Mon Sep 17 00:00:00 2001 From: yperbasis Date: Wed, 5 Aug 2026 23:16:57 +0200 Subject: [PATCH 1/2] db/state, execution/cache, execution/commitment: reconcile caches at the file-publication boundary MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Snapshot downloads publish state that never flows through cache applies. The PR-23033 reconciliation ran at one caller (ProcessFrozenBlocks) on its normal exits only: an error between publication and the absorb skipped it while Start logs and continues, and the commitment BranchCache — whose Get guards only against unwind staleness — was not reconciled at all, so a branch warmed before a download could restore the trie to the wrong state. Move reconciliation to the chokepoint every publication funnels through: recalcVisibleFiles absorbs the bound StateCache (per-domain values-files ends, no tx involved) and the aggregator-owned BranchCache, under dirtyFilesLock and before the new bundle is published, so readers never see extended files while stale entries live. BranchCache gains the applied-watermark it lacked (putWatermark, bumped on every Put): own-built files are a no-op, foreign state clears. BindAggregator's duck becomes BindStateCache — the aggregator holds the cache and reconciles immediately at bind, covering wired-after-files-visible ordering; the PFB-local absorb is removed as superseded. Error paths and mid-run downloads are covered by construction. Closes #23028. --- db/state/aggregator.go | 44 +++++++++++- db/state/aggregator_align_test.go | 70 ++++++++++++++++++++ db/state/execctx/statecache_readfill_test.go | 2 +- execution/cache/state_cache.go | 20 +++--- execution/commitment/branch_cache.go | 32 +++++++++ execution/execmodule/executor.go | 5 -- 6 files changed, 156 insertions(+), 17 deletions(-) diff --git a/db/state/aggregator.go b/db/state/aggregator.go index c6bce6a926f..153452cfba3 100644 --- a/db/state/aggregator.go +++ b/db/state/aggregator.go @@ -55,6 +55,7 @@ import ( "github.com/erigontech/erigon/db/state/kvmetrics" "github.com/erigontech/erigon/db/state/statecfg" "github.com/erigontech/erigon/db/version" + "github.com/erigontech/erigon/execution/cache" "github.com/erigontech/erigon/execution/commitment" ) @@ -96,7 +97,10 @@ type Aggregator struct { // domains' visible ends while set; Close clears it (shutdown is not a // fill window). visibilityLoweringForbidden atomic.Bool - snapshotBuildSema *semaphore.Weighted + // boundStateCache, when set, is reconciled with file publications at every + // visibility recalculation (guarded by dirtyFilesLock). + boundStateCache *cache.StateCache + snapshotBuildSema *semaphore.Weighted disableHistory bool branchCacheDisabled bool @@ -560,6 +564,42 @@ func (a *Aggregator) ForbidVisibilityLowering() { a.visibilityLoweringForbidden.Store(true) } +// BindStateCache binds a fill-enabled StateCache to this aggregator: forbids +// visibility lowering and reconciles the cache with every file publication — +// at the visibility recalculation itself, so no caller or error path can +// publish files without the caches absorbing the extension. Reconciles +// immediately: files may already be visible when the cache is wired. +func (a *Aggregator) BindStateCache(sc *cache.StateCache) { + a.dirtyFilesLock.Lock() + defer a.dirtyFilesLock.Unlock() + a.visibilityLoweringForbidden.Store(true) + a.boundStateCache = sc + if v := a.visible.Load(); v != nil { + a.reconcileCachesLocked(v) + } +} + +// reconcileCachesLocked absorbs the bundle's file visibility into the bound +// StateCache and the aggregator-owned BranchCache. Runs under dirtyFilesLock, +// before a new bundle is published, so readers never see extended files while +// stale cache entries live. +func (a *Aggregator) reconcileCachesLocked(v *aggregatorVisible) { + if sc := a.boundStateCache; sc != nil { + sc.Applier().AbsorbFilesExtension(cache.FrontierFunc(func(domain kv.Domain) (uint64, bool) { + dv := v.d[domain] + if dv == nil { + return 0, false + } + return visibleFiles(dv.files).EndTxNum(), true + })) + } + if cd := a.d[kv.CommitmentDomain]; cd != nil && cd.branchCache != nil { + if cv := v.d[kv.CommitmentDomain]; cv != nil { + cd.branchCache.AbsorbFilesExtension(visibleFiles(cv.files).EndTxNum()) + } + } +} + func (a *Aggregator) setUnalignedDomain(d kv.Domain, v bool) { a.dirtyFilesLock.Lock() defer a.dirtyFilesLock.Unlock() @@ -1921,6 +1961,8 @@ func (a *Aggregator) recalcVisibleFiles(retired retiredFiles) { } } + a.reconcileCachesLocked(next) + old := a.visible.Load() old.retired = retired old.next = next diff --git a/db/state/aggregator_align_test.go b/db/state/aggregator_align_test.go index 2d4be50b798..27b37fcf760 100644 --- a/db/state/aggregator_align_test.go +++ b/db/state/aggregator_align_test.go @@ -24,6 +24,7 @@ import ( "github.com/erigontech/erigon/db/datadir" "github.com/erigontech/erigon/db/kv" + "github.com/erigontech/erigon/execution/cache" ) // generateStandaloneIIFile writes files with a hardcoded step size of 10. @@ -260,3 +261,72 @@ func TestVisibilityLowering_GuardsHistoryIIEnd(t *testing.T) { require.Panics(t, func() { agg.recalcVisibleFiles(nil) }, "lowering a history-II end while values ends stay put must trip the forbid assert") } + +// File publication brings state that never flows through cache applies. The +// visibility recalculation is the one chokepoint every publication funnels +// through, so it reconciles both caches there — no caller or error path can +// skip it. +func TestFilePublicationReconcilesCaches(t *testing.T) { + t.Parallel() + _, agg := testDbAndAggregatorv3(t, alignStepSize) + + sc := cache.NewStateCache(1<<20, 1<<20, 1<<20, 1<<20) + t.Cleanup(sc.Close) + agg.BindStateCache(sc) + + key := make([]byte, 20) + key[0] = 1 + preView := sc.View(cache.FrontierFunc(func(kv.Domain) (uint64, bool) { return 10, true })) + preView.Fill(kv.AccountsDomain, key, []byte{1}, 5) + _, ok := sc.View(nil).Get(kv.AccountsDomain, key) + require.True(t, ok) + + prefix := []byte{0x01} + bc := agg.d[kv.CommitmentDomain].branchCache + require.NotNil(t, bc) + bc.Put(prefix, []byte{0xbb}, 0, 5) + _, _, ok = bc.Get(prefix) + require.True(t, ok) + + generateStateFiles(t, agg.Dirs(), []testFileRange{{0, 1}, {1, 2}}) + generateCommitmentFile(t, agg.Dirs(), []testFileRange{{0, 1}, {1, 2}}) + generateStandaloneIIFiles(t, agg.Dirs(), []testFileRange{{0, 1}, {1, 2}}) + require.NoError(t, agg.OpenFolder()) + + _, ok = sc.View(nil).Get(kv.AccountsDomain, key) + require.False(t, ok, "publication must drop pre-publication state entries") + preView.Fill(kv.AccountsDomain, key, []byte{1}, 5) + _, ok = sc.View(nil).Get(kv.AccountsDomain, key) + require.False(t, ok, "a pre-publication view must not refill past the publication") + + _, _, ok = bc.Get(prefix) + require.False(t, ok, "publication must drop pre-publication branch entries") +} + +// A cache bound after files are already visible starts with its admission +// frontier at the published ends, so views older than the files cannot fill. +func TestBindStateCacheAbsorbsExistingVisibility(t *testing.T) { + t.Parallel() + _, agg := testDbAndAggregatorv3(t, alignStepSize) + + generateStateFiles(t, agg.Dirs(), []testFileRange{{0, 1}, {1, 2}}) + generateCommitmentFile(t, agg.Dirs(), []testFileRange{{0, 1}, {1, 2}}) + generateStandaloneIIFiles(t, agg.Dirs(), []testFileRange{{0, 1}, {1, 2}}) + require.NoError(t, agg.OpenFolder()) + + sc := cache.NewStateCache(1<<20, 1<<20, 1<<20, 1<<20) + t.Cleanup(sc.Close) + agg.BindStateCache(sc) + + key := make([]byte, 20) + key[0] = 1 + older := sc.View(cache.FrontierFunc(func(kv.Domain) (uint64, bool) { return 10, true })) + older.Fill(kv.AccountsDomain, key, []byte{1}, 5) + _, ok := sc.View(nil).Get(kv.AccountsDomain, key) + require.False(t, ok, "a view older than the already-published files must not fill") + + current := sc.View(cache.FrontierFunc(func(kv.Domain) (uint64, bool) { return 2 * alignStepSize, true })) + current.Fill(kv.AccountsDomain, key, []byte{2}, 15) + _, ok = sc.View(nil).Get(kv.AccountsDomain, key) + require.True(t, ok, "a view at the published ends fills normally") +} diff --git a/db/state/execctx/statecache_readfill_test.go b/db/state/execctx/statecache_readfill_test.go index 11476a2b902..03f99cbc9ba 100644 --- a/db/state/execctx/statecache_readfill_test.go +++ b/db/state/execctx/statecache_readfill_test.go @@ -428,7 +428,7 @@ func TestReadFill_NegativeUsesLastVisibleTxNum(t *testing.T) { type fakeForbidder struct{ called bool } -func (f *fakeForbidder) ForbidVisibilityLowering() { f.called = true } +func (f *fakeForbidder) BindStateCache(*cache.StateCache) { f.called = true } // fakeTemporalDB satisfies kv.TemporalRwDB by embedding (the interface now // carries Agg, so a DB shape without it no longer compiles); only Agg is diff --git a/execution/cache/state_cache.go b/execution/cache/state_cache.go index 37adc1fa9f1..246989db1ea 100644 --- a/execution/cache/state_cache.go +++ b/execution/cache/state_cache.go @@ -459,13 +459,13 @@ func (c *StateCache) absorbFilesExtension(f Frontier) { } } -// BindAggregator forbids visibility lowering on db's aggregator for a -// fill-enabled cache: fill admission relies on view frontiers never -// decreasing. SharedDomains.SetStateCache asserts this binding, so wiring -// cannot forget it. The aggregator side is duck-typed (the concrete type -// lives in db/state, above this package) but load-bearing: an aggregator -// without the forbid fails loudly. A nil or apply-only cache needs no -// binding. +// BindAggregator hands this fill-enabled cache to db's aggregator, which +// forbids visibility lowering (fill admission relies on view frontiers never +// decreasing) and reconciles the cache with every file publication. +// SharedDomains.SetStateCache asserts this binding, so wiring cannot forget +// it. The aggregator side is duck-typed (the concrete type lives in db/state, +// above this package) but load-bearing: an aggregator that cannot take the +// binding fails loudly. A nil or apply-only cache needs no binding. func (c *StateCache) BindAggregator(db kv.TemporalRwDB) { if c == nil || !c.FillsEnabled() { return @@ -474,11 +474,11 @@ func (c *StateCache) BindAggregator(db kv.TemporalRwDB) { if agg == nil { panic("assert: fill-enabled StateCache bound to a DB without an aggregator — the visibility-lowering guard would be silently dropped") } - f, ok := agg.(interface{ ForbidVisibilityLowering() }) + b, ok := agg.(interface{ BindStateCache(*StateCache) }) if !ok { - panic(fmt.Sprintf("assert: fill-enabled StateCache bound to a DB whose aggregator %T lacks ForbidVisibilityLowering — the visibility-lowering guard would be silently dropped", agg)) + panic(fmt.Sprintf("assert: fill-enabled StateCache bound to a DB whose aggregator %T lacks BindStateCache — the visibility-lowering guard would be silently dropped", agg)) } - f.ForbidVisibilityLowering() + b.BindStateCache(c) c.aggBound.Store(true) } diff --git a/execution/commitment/branch_cache.go b/execution/commitment/branch_cache.go index 3332d633763..e260859e95e 100644 --- a/execution/commitment/branch_cache.go +++ b/execution/commitment/branch_cache.go @@ -51,6 +51,10 @@ func isCommitmentStateKey(prefix []byte) bool { // writer stripes only make stamped publications atomic with Clear; callers must // still ensure one logical mutation per prefix at the orchestrator. type BranchCache struct { + // putWatermark is the exclusive txNum end of the trie's own writes; a file + // extension at or below it covers state this cache already reflects. + putWatermark atomic.Uint64 + // Root tier — single slot for the root branch (always hottest, always // present). Atomic-pointer access so no lock is needed for the hot // read path. @@ -702,6 +706,12 @@ func (c *BranchCache) Put(prefix []byte, data []byte, step, txN uint64) { dataCopy := make([]byte, len(data)) copy(dataCopy, data) + for w := c.putWatermark.Load(); txN+1 > w; w = c.putWatermark.Load() { + if c.putWatermark.CompareAndSwap(w, txN+1) { + break + } + } + stripe := c.putStripe(prefix) stripe.Lock() defer stripe.Unlock() @@ -758,6 +768,28 @@ func (c *BranchCache) Unwind(unwindToTxN uint64) { // tiers are empty and coherence is reset, so a publication cannot cross // generations. Reset runs after every tier is cleared, so a reader cannot pair a // retired entry with the lifted unwind floor. +// AbsorbFilesExtension reconciles the cache with commitment state published +// by files rather than the trie's own writes (snapshot download): files +// covering the process's writes stay at or below the put watermark and are a +// no-op; anything beyond it carries foreign branch data that nothing would +// ever overwrite, so drop every entry — a stale branch restores the trie to +// the wrong state. +func (c *BranchCache) AbsorbFilesExtension(filesEnd uint64) { + if c == nil { + return + } + for { + w := c.putWatermark.Load() + if filesEnd <= w { + return + } + if c.putWatermark.CompareAndSwap(w, filesEnd) { + break + } + } + c.Clear() +} + func (c *BranchCache) Clear() { c.lockAllPutStripes() defer c.unlockAllPutStripes() diff --git a/execution/execmodule/executor.go b/execution/execmodule/executor.go index 81c0b4f3322..874bdea671d 100644 --- a/execution/execmodule/executor.go +++ b/execution/execmodule/executor.go @@ -210,11 +210,6 @@ func (pe *PipelineExecutor) ProcessFrozenBlocks(ctx context.Context, hook *stage if err := pe.sync.RunSnapshots(nil, tx); err != nil { return err } - // Downloaded state files publish without applies; reconcile the cache - // before anything reads through it. - stateCache.Applier().AbsorbFilesExtension(cache.FrontierFunc(func(domain kv.Domain) (uint64, bool) { - return tx.Debug().DomainVisibleEnd(domain) - })) if onlySnapDownload { return nil } From 0566925869a45aa22e9cbba4af426cc516247749 Mon Sep 17 00:00:00 2001 From: yperbasis Date: Wed, 5 Aug 2026 23:28:57 +0200 Subject: [PATCH 2/2] execution/commitment: add the missed BranchCache absorb unit test The watermark test was written with the boundary-hook commit but never staged (git add -u skips new files); it sat untracked, breaking compilation on sibling branches. --- .../commitment/branch_cache_absorb_test.go | 49 +++++++++++++++++++ 1 file changed, 49 insertions(+) create mode 100644 execution/commitment/branch_cache_absorb_test.go diff --git a/execution/commitment/branch_cache_absorb_test.go b/execution/commitment/branch_cache_absorb_test.go new file mode 100644 index 00000000000..b215fb3bdc6 --- /dev/null +++ b/execution/commitment/branch_cache_absorb_test.go @@ -0,0 +1,49 @@ +// 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 ( + "testing" + + "github.com/stretchr/testify/require" +) + +// Commitment files built from the process's own writes cover txNums at or +// below the put watermark and must not churn the cache; files from a snapshot +// download exceed it and must clear everything — a stale branch restores the +// trie to the wrong state. +func TestBranchCacheAbsorbFilesExtension(t *testing.T) { + t.Parallel() + + c := NewBranchCache(64) + t.Cleanup(c.Close) + prefix := []byte{0x01} + c.Put(prefix, []byte{0xbb}, 0, 100) + + c.AbsorbFilesExtension(101) + _, _, ok := c.Get(prefix) + require.True(t, ok, "files covering the process's own writes must not clear the cache") + + c.AbsorbFilesExtension(150) + _, _, ok = c.Get(prefix) + require.False(t, ok, "files beyond the put watermark carry foreign state — clear") + + c.Put(prefix, []byte{0xcc}, 0, 200) + c.AbsorbFilesExtension(150) + _, _, ok = c.Get(prefix) + require.True(t, ok, "an already-absorbed extension must not clear again") +}