db/state, execution: reconcile StateCache and BranchCache at the file-publication boundary

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

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")
+}

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

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)
 }
 

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

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 <http://www.gnu.org/licenses/>.
+
+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")
+}

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
 	}