execution, db: bind cache views to database and file generations

cmd/integration/commands/stages.go

@@ -844,9 +844,9 @@ func execBlocksBatch(ctx context.Context, db kv.TemporalRwDB, st *stagedsync.Syn
 	}
 	defer doms.Close()
 	doms.SetInMemHistoryReads(false)
-	doms.SetStateCache(stateCache)
+	doms.SetCanonicalStateCache(stateCache)
 	doms.SetCodeStore(codeStore)
-	execctx.GuardAggregatorForCache(db, stateCache)
+	execctx.BindStateCacheToAggregator(db, stateCache)
 
 	s, err := st.StageState(stages.Execution, tx, initialCycle, false)
 	if err != nil {

db/kv/temporal/kv_temporal.go

@@ -283,7 +283,7 @@ type domainVisibleEnds struct {
 	// over-rejects fills: a view's frontier never decreases in a process that
 	// fills a cache — the DB component is frozen at tx begin, and a files
 	// reopen only extends it, an invariant the aggregator enforces once a
-	// fill-enabled cache is wired over it (ForbidVisibilityLowering).
+	// shared latest-state cache is bound (BindStateCache).
 	ends  [kv.DomainLen]atomic.Uint64
 	mu    sync.Mutex
 	state atomic.Uint32

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"
 )
 
@@ -90,13 +91,14 @@ type Aggregator struct {
 	// regenerates them. Guarded by dirtyFilesLock.
 	unalignedDomain [kv.DomainLen]bool
 	unalignedIdx    [kv.StandaloneIdxLen]bool
-	// visibilityLoweringForbidden: a fill-enabled StateCache is wired over
-	// this aggregator, and its fill admission relies on view frontiers never
-	// decreasing. recalcVisibleFiles refuses to lower the cached state
-	// domains' visible ends while set; Close clears it (shutdown is not a
-	// fill window).
+	// visibilityLoweringForbidden: a single-version cache is wired over this
+	// aggregator, and PlainStateVersion does not encode changes to file
+	// visibility. Close clears it because shutdown is not a cache-read window.
 	visibilityLoweringForbidden atomic.Bool
-	snapshotBuildSema           *semaphore.Weighted
+	// boundStateCache is reconciled before a new files view becomes visible.
+	// Guarded by dirtyFilesLock.
+	boundStateCache   *cache.StateCache
+	snapshotBuildSema *semaphore.Weighted
 
 	disableHistory      bool
 	branchCacheDisabled bool
@@ -549,17 +551,35 @@ func (a *Aggregator) UnalignIdx(name kv.InvertedIdx) (realign func()) {
 	return func() {}
 }
 
-// ForbidVisibilityLowering marks this aggregator as backing a fill-enabled
-// StateCache: from then on recalcVisibleFiles panics instead of lowering a
-// cached state domain's visible end, whichever entry point caused it.
+// ForbidVisibilityLowering marks this aggregator as backing a single-version
+// cache. From then on recalcVisibleFiles rejects lowering a cached domain's
+// visible end because PlainStateVersion does not identify that change.
 // Serialized with recalcVisibleFiles via dirtyFilesLock so "from then on"
 // holds against a recalculation already in flight.
 func (a *Aggregator) ForbidVisibilityLowering() {
+	if a.visibilityLoweringForbidden.Load() {
+		return
+	}
 	a.dirtyFilesLock.Lock()
 	defer a.dirtyFilesLock.Unlock()
 	a.visibilityLoweringForbidden.Store(true)
 }
 
+// BindStateCache prevents visibility lowering and reconciles the cache with
+// files that are already visible. Future file publications are reconciled by
+// recalcVisibleFiles before readers can observe their new backing view.
+func (a *Aggregator) BindStateCache(stateCache *cache.StateCache) {
+	if stateCache == nil {
+		return
+	}
+	a.dirtyFilesLock.Lock()
+	defer a.dirtyFilesLock.Unlock()
+	a.visibilityLoweringForbidden.Store(true)
+	a.boundStateCache = stateCache
+	change := stateCache.BeginFilesPublication(visibleStateFilesEnd(a.visible.Load()))
+	change.Finish()
+}
+
 func (a *Aggregator) setUnalignedDomain(d kv.Domain, v bool) {
 	a.dirtyFilesLock.Lock()
 	defer a.dirtyFilesLock.Unlock()
@@ -709,9 +729,17 @@ func (a *Aggregator) ReloadFiles() error {
 // closeDirtyFilesNoReopen drops all dirty-file mmaps without re-scanning the
 // snapshots dir, so a caller can rename the underlying files (Windows forbids
 // renaming a mapped file); a later ReloadFiles re-opens them.
+// closeDirtyFilesNoReopen is an exclusive tooling operation: it temporarily
+// removes all visible files and invalidates cache guarantees tied to them.
 func (a *Aggregator) closeDirtyFilesNoReopen() {
 	a.dirtyFilesLock.Lock()
 	defer a.dirtyFilesLock.Unlock()
+	// This path removes every visible file before replacing them, so no cache
+	// view may remain live across the reset.
+	a.visibilityLoweringForbidden.Store(false)
+	if cd := a.d[kv.CommitmentDomain]; cd != nil && cd.branchCache != nil {
+		cd.branchCache.Reset()
+	}
 	a.closeDirtyFiles()
 	a.recalcVisibleFiles(nil)
 }
@@ -1876,6 +1904,47 @@ type aggregatorVisible struct {
 	next    *aggregatorVisible // oldest→newest linked-list link (set under dirtyFilesLock)
 }
 
+func visibleStateFilesEnd(visible *aggregatorVisible) (ends [kv.DomainLen]uint64) {
+	if visible == nil {
+		return ends
+	}
+	for domain, domainVisible := range visible.d {
+		if domainVisible != nil {
+			ends[domain] = visibleFiles(domainVisible.files).EndTxNum()
+		}
+	}
+	return ends
+}
+
+type cacheFilesPublication struct {
+	state  *cache.PlainStateVersionBackingChange
+	branch *cache.PlainStateVersionBackingChange
+}
+
+func (a *Aggregator) beginCacheFilesPublication(visible *aggregatorVisible) cacheFilesPublication {
+	var publication cacheFilesPublication
+	// SharedDomains.Commit acquires cache publication in the same order.
+	if domain := a.d[kv.CommitmentDomain]; domain != nil && domain.branchCache != nil {
+		if commitmentVisible := visible.d[kv.CommitmentDomain]; commitmentVisible != nil {
+			publication.branch = domain.branchCache.BeginFilesPublication(visibleFiles(commitmentVisible.files).EndTxNum())
+		}
+	}
+	if a.boundStateCache != nil {
+		publication.state = a.boundStateCache.BeginFilesPublication(visibleStateFilesEnd(visible))
+	}
+	return publication
+}
+
+func (p *cacheFilesPublication) Finish() {
+	if p == nil {
+		return
+	}
+	p.state.Finish()
+	p.state = nil
+	p.branch.Finish()
+	p.branch = nil
+}
+
 // recalcVisibleFiles must be called with dirtyFilesLock held (writers are
 // serialized by it; readers take no lock and instead load a.visible). It builds
 // a fresh immutable aggregatorVisible bundle via the per-entity calcVisibleFiles
@@ -1901,30 +1970,34 @@ func (a *Aggregator) recalcVisibleFiles(retired retiredFiles) {
 
 	if a.visibilityLoweringForbidden.Load() {
 		prev := a.visible.Load()
-		for _, d := range []kv.Domain{kv.AccountsDomain, kv.StorageDomain, kv.CodeDomain} {
+		for _, d := range []kv.Domain{kv.AccountsDomain, kv.StorageDomain, kv.CodeDomain, kv.CommitmentDomain} {
 			if prev.d[d] == nil || next.d[d] == nil {
 				continue
 			}
 			prevEnd := visibleFiles(prev.d[d].files).EndTxNum()
 			nextEnd := visibleFiles(next.d[d].files).EndTxNum()
 			if nextEnd < prevEnd {
-				panic(fmt.Sprintf("assert: %s visible end lowered %d -> %d while a fill-enabled StateCache is wired — fill admission relies on view frontiers never decreasing", d, prevEnd, nextEnd))
+				panic(fmt.Sprintf("assert: %s visible end lowered %d -> %d while a single-version cache is wired — PlainStateVersion does not identify file-visibility changes", d, prevEnd, nextEnd))
 			}
 			if prev.dhii[d] == nil || next.dhii[d] == nil {
 				continue
 			}
 			prevII := prev.dhii[d].files.EndTxNum()
 			nextII := next.dhii[d].files.EndTxNum()
 			if nextII < prevII {
-				panic(fmt.Sprintf("assert: %s history-II visible end lowered %d -> %d while a fill-enabled StateCache is wired — DomainVisibleEnd derives view frontiers from it", d, prevII, nextII))
+				panic(fmt.Sprintf("assert: %s history-II visible end lowered %d -> %d while a single-version cache is wired — exact cache-view eligibility derives its frontier from history-II", d, prevII, nextII))
 			}
 		}
 	}
 
+	cachePublication := a.beginCacheFilesPublication(next)
+	defer cachePublication.Finish()
+
 	old := a.visible.Load()
 	old.retired = retired
 	old.next = next
 	a.visible.Store(next)
+	cachePublication.Finish()
 
 	// `recalcVisibleFiles` is rare background operation under `dirtyFilesLock`
 	// it's good idea to delete files here, then hot reader-Close path will more likely be lock-free
@@ -2675,6 +2748,14 @@ func (at *AggregatorRoTx) MetricsCollector() *kvmetrics.Collector {
 	return at.a.metricsCollector
 }
 
+func (at *AggregatorRoTx) ForbidVisibilityLowering() {
+	at.a.ForbidVisibilityLowering()
+}
+
+func (at *AggregatorRoTx) BindStateCache(stateCache *cache.StateCache) {
+	at.a.BindStateCache(stateCache)
+}
+
 func (at *AggregatorRoTx) Dirs() datadir.Dirs                  { return at.a.dirs }
 func (at *AggregatorRoTx) standaloneIIs() []*InvertedIndexRoTx { return at.iis[:at.iisCount] }
 

db/state/aggregator_align_test.go

@@ -24,6 +24,8 @@ import (
 
 	"github.com/erigontech/erigon/db/datadir"
 	"github.com/erigontech/erigon/db/kv"
+	"github.com/erigontech/erigon/db/state/execctx"
+	"github.com/erigontech/erigon/execution/cache"
 )
 
 // generateStandaloneIIFile writes files with a hardcoded step size of 10.
@@ -55,6 +57,10 @@ func requireVisibleEnd(t *testing.T, agg *Aggregator, end uint64) {
 	}
 }
 
+type cacheAggregatorHolder struct{ agg *Aggregator }
+
+func (h cacheAggregatorHolder) Agg() any { return h.agg }
+
 // state visible past commitment's files = state no commitment covers
 func TestVisibleFilesAligned_LaggingCommitmentClampsEveryone(t *testing.T) {
 	t.Parallel()
@@ -260,3 +266,107 @@ 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 TestVisibilityLowering_GuardsCommitmentDomain(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())
+
+	agg.Unalign(kv.CommitmentDomain)
+	agg.ForbidVisibilityLowering()
+
+	agg.dirtyFilesLock.Lock()
+	defer agg.dirtyFilesLock.Unlock()
+	dropped := 0
+	agg.d[kv.CommitmentDomain].dirtyFiles.CloseIf(func(item *FilesItem) bool {
+		if item.endTxNum == 2*alignStepSize {
+			dropped++
+			return true
+		}
+		return false
+	})
+	require.Equal(t, 1, dropped)
+
+	require.Panics(t, func() { agg.recalcVisibleFiles(nil) },
+		"BranchCache validity requires the commitment frontier to remain monotonic")
+}
+
+func TestFilePublicationRevokesCacheGenerations(t *testing.T) {
+	t.Parallel()
+	_, agg := testDbAndAggregatorv3(t, alignStepSize)
+
+	stateCache := cache.NewStateCache(1<<20, 1<<20, 1<<20, 1<<20)
+	t.Cleanup(stateCache.Close)
+	statePublisher := stateCache.Publisher()
+	statePublisher.Initialize(1)
+	execctx.BindStateCacheToAggregator(cacheAggregatorHolder{agg}, stateCache)
+
+	accountKey := make([]byte, 20)
+	accountKey[0] = 1
+	stateView := stateCache.View(1)
+	stateView.Fill(kv.AccountsDomain, accountKey, []byte{1}, 1)
+	_, ok := stateView.Get(kv.AccountsDomain, accountKey)
+	require.True(t, ok)
+
+	branchCache := agg.d[kv.CommitmentDomain].branchCache
+	require.NotNil(t, branchCache)
+	branchPublisher := branchCache.Publisher()
+	branchPublisher.Initialize(1)
+	branchKey := []byte{0x01}
+	branchView := branchCache.View(1)
+	branchView.Fill(branchKey, []byte{0xbb}, 1)
+	_, _, ok = branchView.Get(branchKey)
+	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 = stateView.Get(kv.AccountsDomain, accountKey)
+	require.False(t, ok, "file publication must revoke pre-publication state views")
+	stateView.Fill(kv.AccountsDomain, accountKey, []byte{1}, 1)
+	statePublisher.Initialize(1)
+	_, ok = stateCache.View(1).Get(kv.AccountsDomain, accountKey)
+	require.False(t, ok, "a revoked state view must not refill after file publication")
+
+	_, _, ok = branchView.Get(branchKey)
+	require.False(t, ok, "file publication must revoke pre-publication branch views")
+	branchView.Fill(branchKey, []byte{0xbb}, 1)
+	branchPublisher.Initialize(1)
+	_, _, ok = branchCache.View(1).Get(branchKey)
+	require.False(t, ok, "a revoked branch view must not refill after file publication")
+}
+
+func TestCacheBindingAbsorbsExistingFileVisibility(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())
+
+	stateCache := cache.NewStateCache(1<<20, 1<<20, 1<<20, 1<<20)
+	t.Cleanup(stateCache.Close)
+	statePublisher := stateCache.Publisher()
+	statePublisher.Initialize(1)
+	accountKey := make([]byte, 20)
+	accountKey[0] = 1
+	oldView := stateCache.View(1)
+	oldView.Fill(kv.AccountsDomain, accountKey, []byte{1}, 1)
+	_, ok := oldView.Get(kv.AccountsDomain, accountKey)
+	require.True(t, ok)
+
+	execctx.BindStateCacheToAggregator(cacheAggregatorHolder{agg}, stateCache)
+
+	_, ok = oldView.Get(kv.AccountsDomain, accountKey)
+	require.False(t, ok, "binding must revoke entries created before the visible files were absorbed")
+	statePublisher.Initialize(1)
+	_, ok = stateCache.View(1).Get(kv.AccountsDomain, accountKey)
+	require.False(t, ok)
+}

db/state/aggregator_close_test.go

@@ -181,7 +181,7 @@ func TestAggregatorCloseReleasesBranchCache(t *testing.T) {
 	require.NotNil(t, cd.branchCache, "precondition: BranchCache is set when USE_STATE_CACHE is on")
 
 	prefix := []byte{0x01, 0x02}
-	cd.branchCache.Put(prefix, []byte{0xaa, 0xbb}, 1, 1)
+	cd.branchCache.Put(prefix, []byte{0xaa, 0xbb}, 1)
 	_, _, ok := cd.branchCache.Get(prefix)
 	require.True(t, ok, "precondition: entry is cached before Close")
 

db/state/commitment_convert_blackbox_test.go

@@ -474,7 +474,7 @@ func computeCommitmentRoot(t *testing.T, db kv.TemporalRwDB) []byte {
 	tx, err := db.BeginTemporalRw(t.Context())
 	require.NoError(t, err)
 	defer tx.Rollback()
-	domains, err := execctx.NewSharedDomains(t.Context(), tx, log.New())
+	domains, err := execctx.NewSharedDomains(t.Context(), tx, log.New(), execctx.WithoutSharedBranchCache())
 	require.NoError(t, err)
 	defer domains.Close()
 	rh, err := domains.ComputeCommitment(t.Context(), tx, false, 0, 0, "", nil)