db/state, execution: reconcile caches when immutable files become visible

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,6 +97,7 @@ type Aggregator struct {
 	// domains' visible ends while set; Close clears it (shutdown is not a
 	// fill window).
 	visibilityLoweringForbidden atomic.Bool
+	boundStateCache             *cache.StateCache
 	snapshotBuildSema           *semaphore.Weighted
 
 	disableHistory      bool
@@ -560,6 +562,25 @@ func (a *Aggregator) ForbidVisibilityLowering() {
 	a.visibilityLoweringForbidden.Store(true)
 }
 
+// BindStateCache reconciles a cache with current and future file visibility.
+func (a *Aggregator) BindStateCache(stateCache *cache.StateCache) {
+	if stateCache == nil {
+		return
+	}
+	a.dirtyFilesLock.Lock()
+	defer a.dirtyFilesLock.Unlock()
+	if a.boundStateCache != nil && a.boundStateCache != stateCache {
+		panic("assert: aggregator already has a different StateCache")
+	}
+	if stateCache.FillsEnabled() {
+		a.visibilityLoweringForbidden.Store(true)
+	}
+	a.boundStateCache = stateCache
+	if visible := a.visible.Load(); visible != nil {
+		a.reconcileCachesLocked(visible)
+	}
+}
+
 func (a *Aggregator) setUnalignedDomain(d kv.Domain, v bool) {
 	a.dirtyFilesLock.Lock()
 	defer a.dirtyFilesLock.Unlock()
@@ -729,6 +750,7 @@ func (a *Aggregator) WaitForFiles() {
 func (a *Aggregator) Close() {
 	a.dirtyFilesLock.Lock()
 	a.visibilityLoweringForbidden.Store(false) // shutdown is not a fill window
+	a.boundStateCache = nil
 	a.dirtyFilesLock.Unlock()
 	a.WaitForFiles()
 	if !a.background.BeginClose() { // idempotent: safe to call Close multiple times
@@ -1920,6 +1942,7 @@ func (a *Aggregator) recalcVisibleFiles(retired retiredFiles) {
 			}
 		}
 	}
+	a.reconcileCachesLocked(next)
 
 	old := a.visible.Load()
 	old.retired = retired
@@ -1931,6 +1954,24 @@ func (a *Aggregator) recalcVisibleFiles(retired retiredFiles) {
 	reclaimFiles(a.reclaimRetiredLocked())
 }
 
+// Cache reconciliation precedes visible.Store so readers cannot combine new
+// files with entries from the previous visibility generation.
+func (a *Aggregator) reconcileCachesLocked(visible *aggregatorVisible) {
+	if a.boundStateCache != nil {
+		a.boundStateCache.Applier().ReconcileFiles(cache.FrontierFunc(func(domain kv.Domain) (uint64, bool) {
+			domainVisible := visible.d[domain]
+			if domainVisible == nil {
+				return 0, false
+			}
+			return visibleFiles(domainVisible.files).EndTxNum(), true
+		}))
+	}
+	commitmentDomain := a.d[kv.CommitmentDomain]
+	if commitmentDomain != nil && commitmentDomain.branchCache != nil && visible.d[kv.CommitmentDomain] != nil {
+		commitmentDomain.branchCache.ReconcileFiles(visibleFiles(visible.d[kv.CommitmentDomain].files).EndTxNum())
+	}
+}
+
 // stateMinimaxTxNum returns min(EndTxNum) across kv.StateDomains. Mirrors
 // AggregatorRoTx.TxNumsInFiles but operates directly on the bundle so the
 // writer can compute it without spinning up a throwaway RoTx.

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,71 @@ 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")
 }
+
+func TestFilePublicationClearsStaleBranchCache(t *testing.T) {
+	t.Parallel()
+	_, agg := testDbAndAggregatorv3(t, alignStepSize)
+
+	branchCache := agg.d[kv.CommitmentDomain].branchCache
+	require.NotNil(t, branchCache)
+	prefix := []byte{0x01}
+	branchCache.Put(prefix, []byte{0xbb}, 0, 5)
+	_, _, ok := branchCache.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 = branchCache.Get(prefix)
+	require.False(t, ok)
+}
+
+func TestFilePublicationReconcilesStateCache(t *testing.T) {
+	t.Setenv("STATE_CACHE_FILLS", "true")
+	_, agg := testDbAndAggregatorv3(t, alignStepSize)
+
+	stateCache := cache.NewStateCache(1<<20, 1<<20, 1<<20, 1<<20)
+	t.Cleanup(stateCache.Close)
+	agg.BindStateCache(stateCache)
+
+	key := make([]byte, 20)
+	key[0] = 1
+	older := stateCache.View(cache.FrontierFunc(func(kv.Domain) (uint64, bool) { return 10, true }))
+	older.Fill(kv.AccountsDomain, key, []byte{1}, 5)
+	_, ok := stateCache.View(nil).Get(kv.AccountsDomain, key)
+	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 = stateCache.View(nil).Get(kv.AccountsDomain, key)
+	require.False(t, ok)
+	older.Fill(kv.AccountsDomain, key, []byte{1}, 5)
+	_, ok = stateCache.View(nil).Get(kv.AccountsDomain, key)
+	require.False(t, ok)
+}
+
+func TestBindStateCacheReconcilesExistingFiles(t *testing.T) {
+	t.Setenv("STATE_CACHE_FILLS", "true")
+	_, 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())
+
+	stateCache := cache.NewStateCache(1<<20, 1<<20, 1<<20, 1<<20)
+	t.Cleanup(stateCache.Close)
+	agg.BindStateCache(stateCache)
+
+	key := make([]byte, 20)
+	key[0] = 1
+	stateCache.View(cache.FrontierFunc(func(kv.Domain) (uint64, bool) { return 10, true })).Fill(
+		kv.AccountsDomain, key, []byte{1}, 5)
+	_, ok := stateCache.View(nil).Get(kv.AccountsDomain, key)
+	require.False(t, ok)
+}

db/state/execctx/domain_shared.go

@@ -136,7 +136,7 @@ func (m *domainVisibleEndMemo) load(tx kv.TemporalTx, domain kv.Domain, viewID u
 		state = 0
 		m.viewID.Store(viewID)
 	}
-	end, ok := tx.Debug().DomainVisibleEnd(domain)
+	end, ok := debugDomainVisibleEnd(tx, domain)
 	m.ends[domain].Store(end)
 	state |= loadedBit
 	if ok {
@@ -159,7 +159,15 @@ func (sd *SharedDomains) domainVisibleEnd(tx kv.TemporalTx, domain kv.Domain) (u
 	if _, ok := tx.(kv.TemporalRwTx); ok {
 		return sd.visibleEnds.get(tx, domain)
 	}
-	return tx.Debug().DomainVisibleEnd(domain)
+	return debugDomainVisibleEnd(tx, domain)
+}
+
+func debugDomainVisibleEnd(tx kv.TemporalTx, domain kv.Domain) (uint64, bool) {
+	debugTx := tx.Debug()
+	if debugTx == nil {
+		return 0, false
+	}
+	return debugTx.DomainVisibleEnd(domain)
 }
 
 // sdFrontier adapts one (SharedDomains, tx) pair to cache.Frontier: writable
@@ -946,27 +954,22 @@ func (sd *SharedDomains) BindStateCache(stateCache *cache.StateCache) {
 	sd.cacheApplier.Initialize(sd.baseStateVersion)
 }
 
-// GuardAggregatorForCache forbids visibility lowering on db's aggregator when
-// sc is a fill-enabled StateCache: fill admission relies on view frontiers
-// never decreasing. This is the one place that binds the invariant — call it
-// wherever a fill-enabled cache is wired over a DB. Duck-typed so the storage
-// layer need not know the cache type (and vice versa) — but load-bearing, so
-// a db that cannot produce its aggregator fails loudly instead of silently
-// dropping the guard. A nil or apply-only cache needs no guard.
+// GuardAggregatorForCache binds a StateCache to immutable-file
+// publication. An incompatible DB panics because omitting this fence is unsafe.
 func GuardAggregatorForCache(db any, sc *cache.StateCache) {
-	if sc == nil || !sc.FillsEnabled() {
+	if sc == nil {
 		return
 	}
 	h, ok := db.(interface{ Agg() any })
 	if !ok {
-		panic(fmt.Sprintf("assert: fill-enabled StateCache wired over %T, which cannot produce its aggregator — the visibility-lowering guard would be silently dropped", db))
+		panic(fmt.Sprintf("assert: StateCache wired over %T, which cannot produce its aggregator — file-publication coherence would be silently dropped", db))
 	}
 	agg := h.Agg()
-	f, ok := agg.(interface{ ForbidVisibilityLowering() })
+	binder, ok := agg.(interface{ BindStateCache(*cache.StateCache) })
 	if !ok {
-		panic(fmt.Sprintf("assert: aggregator %T lacks ForbidVisibilityLowering — the visibility-lowering guard would be silently dropped", agg))
+		panic(fmt.Sprintf("assert: aggregator %T lacks BindStateCache — file-publication coherence would be silently dropped", agg))
 	}
-	f.ForbidVisibilityLowering()
+	binder.BindStateCache(sc)
 }
 
 // SetCodeStore sets the persistent codehash-keyed code cache.
@@ -1319,17 +1322,21 @@ func (sd *SharedDomains) Commit(ctx context.Context, tx kv.RwTx, validate ...fun
 	for i := range pendingBranches {
 		u := &pendingBranches[i]
 		if len(u.val) == 0 {
-			sd.branchCache.Invalidate(u.key)
+			sd.branchCache.Delete(u.key, u.txN)
 		} else {
 			sd.branchCache.Put(u.key, u.val, uint64(u.step), u.txN)
 		}
 	}
+	if sd.branchCache != nil {
+		sd.branchCache.AdvanceCommit(sd.txNum)
+	}
 	if sd.stateCache != nil {
 		if sd.cacheUnwind.pending {
 			sd.cacheApplier.PublishUnwind(sourceStateVersion, committedStateVersion, sd.cacheUnwind.toTxNum, pendingState)
 		} else {
 			sd.cacheApplier.Publish(sourceStateVersion, committedStateVersion, pendingState)
 		}
+		sd.cacheApplier.AdvanceCommit(sd.txNum)
 		sd.cacheUnwind = cacheUnwindState{}
 	}
 	return nil
@@ -1485,8 +1492,10 @@ func (sd *SharedDomains) getLatestMetered(domain kv.Domain, tx kv.TemporalTx, k
 	// branchCache sits between sd.mem/parent.mem and the aggTx files for
 	// CommitmentDomain only. Snapshot-isolated readers must disable it because
 	// concurrent commits can advance the cache beyond their transaction view.
+	var branchView commitment.BranchCacheView
 	if domain == kv.CommitmentDomain && sd.branchCache != nil {
-		if cv, cStepU64, ok := sd.branchCache.Get(k); ok {
+		branchView = sd.branchCache.View()
+		if cv, cStepU64, ok := branchView.Get(k); ok {
 			// Get returns the on-disk step index directly — do NOT divide by
 			// StepSize (that double-division collapsed cStep to ~0, defeating the
 			// gate).
@@ -1527,11 +1536,12 @@ func (sd *SharedDomains) getLatestMetered(domain kv.Domain, tx kv.TemporalTx, k
 			fillView.Fill(domain, k, v, readTxNum)
 		}
 	}
-	// Only cache a branch when the read's txN is known: a txN=0 entry would
-	// be treated as immortal by UnwindTo, so skip the Put rather than insert
-	// an entry that can never be unwind-evicted.
+	// Only cache a branch when its txN and the read view's exact frontier are
+	// known. Both are needed to fence unwinds and file publication.
 	if domain == kv.CommitmentDomain && sd.branchCache != nil && len(v) > 0 && txNKnown {
-		sd.branchCache.Put(k, v, uint64(step), readTxN)
+		if visibleEnd, ok := sd.domainVisibleEnd(tx, domain); ok {
+			branchView.Fill(k, v, uint64(step), readTxN, visibleEnd)
+		}
 	}
 
 	return v, step, nil

db/state/execctx/domain_visible_end_memo_test.go

@@ -39,10 +39,20 @@ type stubVisibleEndDebug struct {
 	viewID uint64
 }
 
+type nilDebugTemporalTx struct{ kv.TemporalTx }
+
+func (nilDebugTemporalTx) Debug() kv.TemporalDebugTx { return nil }
+
 func (d stubVisibleEndDebug) DomainVisibleEnd(kv.Domain) (uint64, bool) {
 	return d.viewID * 100, true
 }
 
+func TestDebugDomainVisibleEndWithoutDebugBackend(t *testing.T) {
+	end, ok := debugDomainVisibleEnd(nilDebugTemporalTx{}, kv.AccountsDomain)
+	require.False(t, ok)
+	require.Zero(t, end)
+}
+
 // Parallel-exec workers share one SharedDomains and one view, so the memo
 // must tolerate concurrent gets interleaved with resets, and must re-derive
 // after a sequential view rotation.

db/state/execctx/statecache_readfill_test.go

@@ -720,45 +720,50 @@ func TestReadFill_NegativeUsesLastVisibleTxNum(t *testing.T) {
 	require.False(t, ok, "an unwind of the view's last included txNum must invalidate the negative")
 }
 
-type fakeForbidder struct{ called bool }
+type fakeCacheBinder struct {
+	called bool
+	cache  *cache.StateCache
+}
 
-func (f *fakeForbidder) ForbidVisibilityLowering() { f.called = true }
+func (f *fakeCacheBinder) BindStateCache(stateCache *cache.StateCache) {
+	f.called = true
+	f.cache = stateCache
+}
 
-type fakeHasAgg struct{ f *fakeForbidder }
+type fakeHasAgg struct{ f *fakeCacheBinder }
 
 func (h fakeHasAgg) Agg() any { return h.f }
 
 type fakeHasBadAgg struct{}
 
 func (fakeHasBadAgg) Agg() any { return struct{}{} }
 
-// The guard is load-bearing: for a fill-enabled cache it must either bind the
-// invariant or fail loudly — never silently drop it on a DB shape mismatch.
-// A nil or apply-only cache needs no guard at all.
+// Every non-nil cache must bind or fail loudly because file publication can
+// bypass authoritative applies even when reader fills are disabled.
 func TestGuardAggregatorForCache(t *testing.T) {
 	sc := newSmallStateCache()
 	t.Cleanup(sc.Close)
 
-	f := &fakeForbidder{}
+	f := &fakeCacheBinder{}
 	execctx.GuardAggregatorForCache(fakeHasAgg{f}, sc)
 	require.True(t, f.called)
+	require.Same(t, sc, f.cache)
 
 	require.NotPanics(t, func() { execctx.GuardAggregatorForCache(struct{}{}, nil) },
 		"no cache, no invariant to bind — shape is irrelevant")
 	require.Panics(t, func() { execctx.GuardAggregatorForCache(struct{}{}, sc) },
 		"a db that cannot produce its aggregator must fail loudly, not drop the guard")
 	require.Panics(t, func() { execctx.GuardAggregatorForCache(fakeHasBadAgg{}, sc) },
-		"an aggregator without ForbidVisibilityLowering must fail loudly, not drop the guard")
+		"an aggregator without BindStateCache must fail loudly, not drop the guard")
 }
 
-// An apply-only cache (STATE_CACHE_FILLS=false) has no fills for a lowered
-// frontier to poison, so the guard must not constrain the aggregator.
-func TestGuardAggregatorForCache_ApplyOnlySkips(t *testing.T) {
+func TestGuardAggregatorForCache_ApplyOnlyStillBindsPublication(t *testing.T) {
 	t.Setenv("STATE_CACHE_FILLS", "false")
 	sc := newSmallStateCache()
 	t.Cleanup(sc.Close)
 
-	f := &fakeForbidder{}
+	f := &fakeCacheBinder{}
 	execctx.GuardAggregatorForCache(fakeHasAgg{f}, sc)
-	require.False(t, f.called)
+	require.True(t, f.called)
+	require.Same(t, sc, f.cache)
 }