execution, db: bind cache views to state versions and file views

cmd/integration/commands/stages.go

@@ -714,15 +714,11 @@ func stageExec(db kv.TemporalRwDB, ctx context.Context, logger log.Logger) error
 			return err
 		}
 		defer doms.Close()
+		doms.SetCanonicalCaches(nil)
 		if err := stagedsync.UnwindExecutionStage(u, s, doms, tx, ctx, cfg, logger); err != nil {
 			return err
 		}
-		if err := doms.Flush(ctx, tx); err != nil {
-			return err
-		}
-		err = tx.Commit()
-		tx = nil
-		return err
+		return commitExecUnwind(ctx, doms, tx)
 	}
 
 	if pruneTo > 0 {
@@ -824,13 +820,18 @@ func stageExec(db kv.TemporalRwDB, ctx context.Context, logger log.Logger) error
 	return nil
 }
 
+func commitExecUnwind(ctx context.Context, doms *execctx.SharedDomains, tx kv.TemporalRwTx) error {
+	return doms.Commit(ctx, tx)
+}
+
 // execBlocksBatch runs one stage_exec batch in its own rwtx and SharedDomains:
 // exec up to toBlock (or the batch limit), then doms.Commit. Commit (not Flush)
-// refreshes the aggregator BranchCache to match committed state — a stale cache
-// makes the next batch compute a wrong trie root — and commits the tx. A fresh
-// SharedDomains per call avoids reusing a committed (spent) one. Pruning and
-// file-building are the caller's job (agg.CollateAndPrune). Returns the Execution
-// stage progress after the batch.
+// publishes StateCache and BranchCache only after the database commit and
+// consumes the transaction. BranchCache must match durable state because the
+// next batch uses its branches to compute the trie root. A fresh SharedDomains
+// per call avoids reusing the spent transaction. Pruning and file building are
+// the caller's job. The function returns Execution stage progress after the
+// batch.
 func execBlocksBatch(ctx context.Context, db kv.TemporalRwDB, st *stagedsync.Sync, cfg stagedsync.ExecuteBlockCfg, toBlock uint64, initialCycle bool, stateCache *cache.StateCache, codeStore *cache.CodeStore, logger log.Logger) (uint64, error) {
 	tx, err := db.BeginTemporalRw(ctx)
 	if err != nil {
@@ -844,9 +845,9 @@ func execBlocksBatch(ctx context.Context, db kv.TemporalRwDB, st *stagedsync.Syn
 	}
 	defer doms.Close()
 	doms.SetInMemHistoryReads(false)
-	doms.SetStateCache(stateCache)
+	doms.SetCanonicalCaches(stateCache)
 	doms.SetCodeStore(codeStore)
-	execctx.GuardAggregatorForCache(db, stateCache)
+	execctx.BindStateCacheToAggregator(db, stateCache)
 
 	s, err := st.StageState(stages.Execution, tx, initialCycle, false)
 	if err != nil {

cmd/integration/commands/stages_test.go

@@ -0,0 +1,96 @@
+// 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 commands
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/erigontech/erigon/common/dbg"
+	"github.com/erigontech/erigon/common/log/v3"
+	"github.com/erigontech/erigon/db/datadir"
+	"github.com/erigontech/erigon/db/kv"
+	"github.com/erigontech/erigon/db/kv/temporal/temporaltest"
+	"github.com/erigontech/erigon/db/rawdb"
+	"github.com/erigontech/erigon/db/state/execctx"
+	"github.com/erigontech/erigon/execution/cache"
+	"github.com/erigontech/erigon/execution/commitment"
+)
+
+func branchGeneration(t *testing.T, tx kv.TemporalTx) cache.Generation {
+	t.Helper()
+	stateVersion, err := rawdb.GetStateVersion(tx)
+	require.NoError(t, err)
+	return cache.BranchGeneration(stateVersion, tx.Debug().TxNumsInFiles(kv.CommitmentDomain))
+}
+
+func TestCommitExecUnwindDoesNotRepublishDiscardedBranches(t *testing.T) {
+	previous := dbg.UseStateCache
+	dbg.SetUseStateCache(true)
+	t.Cleanup(func() { dbg.SetUseStateCache(previous) })
+
+	ctx := t.Context()
+	logger := log.New()
+	db := temporaltest.NewTestDBWithStepSize(t, datadir.New(t.TempDir()), 100)
+
+	seedTx, err := db.BeginTemporalRw(ctx)
+	require.NoError(t, err)
+	defer seedTx.Rollback()
+	seedDomains, err := execctx.NewSharedDomains(ctx, seedTx, logger)
+	require.NoError(t, err)
+	seedDomains.SetCanonicalCaches(nil)
+	require.NoError(t, seedDomains.Commit(ctx, seedTx))
+	seedDomains.Close()
+
+	unwindTx, err := db.BeginTemporalRw(ctx)
+	require.NoError(t, err)
+	defer unwindTx.Rollback()
+	unwindDomains, err := execctx.NewSharedDomains(ctx, unwindTx, logger)
+	require.NoError(t, err)
+	unwindDomains.SetCanonicalCaches(nil)
+
+	provider, ok := unwindTx.AggTx().(commitment.BranchCacheProvider)
+	require.True(t, ok)
+	branchCache := provider.BranchCache()
+	require.NotNil(t, branchCache)
+	discardedKey := []byte{0xa0, 0xb0}
+	oldView := branchCache.View(branchGeneration(t, unwindTx))
+	oldView.Fill(discardedKey, []byte("discarded-fork"), 1)
+	_, _, ok = oldView.Get(discardedKey)
+	require.True(t, ok, "precondition: discarded branch is cached")
+
+	var diffs [kv.DomainLen][]kv.DomainEntryDiff
+	unwindDomains.Unwind(0, &diffs)
+	require.NoError(t, commitExecUnwind(ctx, unwindDomains, unwindTx))
+	unwindDomains.Close()
+
+	nextTx, err := db.BeginTemporalRw(ctx)
+	require.NoError(t, err)
+	defer nextTx.Rollback()
+	nextDomains, err := execctx.NewSharedDomains(ctx, nextTx, logger)
+	require.NoError(t, err)
+	nextDomains.SetCanonicalCaches(nil)
+	require.NoError(t, nextDomains.Commit(ctx, nextTx))
+	nextDomains.Close()
+
+	readTx, err := db.BeginTemporalRo(ctx)
+	require.NoError(t, err)
+	defer readTx.Rollback()
+	_, _, ok = branchCache.View(branchGeneration(t, readTx)).Get(discardedKey)
+	require.False(t, ok, "a later commit must not republish a branch discarded by the unwind")
+}

cmd/integration/commands/state_stages.go

@@ -169,6 +169,7 @@ func syncBySmallSteps(db kv.TemporalRwDB, builderConfig buildercfg.BuilderConfig
 	}
 	defer func() { sd.Close() }() // closes whichever SD is current after the commit loop swaps it
 	sd.SetInMemHistoryReads(false)
+	sd.SetCanonicalCaches(nil)
 
 	var batchSize datasize.ByteSize
 	must(batchSize.UnmarshalText([]byte(batchSizeStr)))
@@ -296,6 +297,7 @@ func syncBySmallSteps(db kv.TemporalRwDB, builderConfig buildercfg.BuilderConfig
 				return err
 			}
 			sd.SetInMemHistoryReads(false)
+			sd.SetCanonicalCaches(nil)
 		}
 
 		//receiptsInDB := rawdb.ReadReceiptsByNumber(tx, progress(tx, stages.Execution)+1)

db/kv/kv_interface.go

@@ -525,6 +525,9 @@ type TemporalDebugTx interface {
 	// DomainVisibleEnd returns the exact exclusive txNum bound of the tx's
 	// domain read view. ok is false when the backend cannot provide an exact bound.
 	DomainVisibleEnd(domain Domain) (visibleEnd uint64, ok bool)
+	// HasExactDomainVisibleEnd reports DomainVisibleEnd's ok result without
+	// resolving the bound, which may require a database cursor.
+	HasExactDomainVisibleEnd(domain Domain) bool
 	IIProgress(name InvertedIdx) (txNum uint64)
 	StepSize() uint64
 	// Retire retires frozen history files entirely below their

db/kv/remotedb/kv_remote.go

@@ -258,6 +258,7 @@ func (tx *tx) DomainProgress(domain kv.Domain) uint64          { panic("not impl
 func (tx *tx) DomainVisibleEnd(domain kv.Domain) (uint64, bool) {
 	return 0, false
 }
+func (tx *tx) HasExactDomainVisibleEnd(domain kv.Domain) bool { return false }
 func (tx *tx) GetLatestFromDB(domain kv.Domain, k []byte) (v []byte, step kv.Step, found bool, err error) {
 	panic("not implemented")
 }

db/kv/temporal/kv_temporal.go

@@ -277,13 +277,9 @@ type RwTx struct {
 }
 
 type domainVisibleEnds struct {
-	// ends is atomic so a lock-free read can overlap a reset-and-reload of
-	// the same slot without a data race. A torn read (state bit from one
-	// generation, end from another) can only be stale-low, which merely
-	// 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).
+	// ends is atomic because a lock-free read may overlap reset after the files
+	// transaction reopens. state publishes a slot only after its end is stored;
+	// reset takes mu so an in-flight load cannot republish old data.
 	ends  [kv.DomainLen]atomic.Uint64
 	mu    sync.Mutex
 	state atomic.Uint32
@@ -796,6 +792,12 @@ func (tx *Tx) DomainVisibleEnd(domain kv.Domain) (uint64, bool) {
 func (tx *RwTx) DomainVisibleEnd(domain kv.Domain) (uint64, bool) {
 	return tx.aggtx.DomainVisibleEnd(domain, tx.RwTx)
 }
+func (tx *Tx) HasExactDomainVisibleEnd(domain kv.Domain) bool {
+	return tx.aggtx.HasExactDomainVisibleEnd(domain)
+}
+func (tx *RwTx) HasExactDomainVisibleEnd(domain kv.Domain) bool {
+	return tx.aggtx.HasExactDomainVisibleEnd(domain)
+}
 func (tx *Tx) IIProgress(domain kv.InvertedIdx) uint64 {
 	return tx.aggtx.IIProgress(domain, tx.Tx)
 }

db/kv/temporal/kv_temporal_test.go

@@ -294,6 +294,7 @@ func TestTemporalTx_DomainVisibleEndConcurrent(t *testing.T) {
 	defer baseTtx.Rollback()
 	for d := range kv.DomainLen {
 		expectedEnd[d], expectedOk[d] = baseTtx.Debug().DomainVisibleEnd(d)
+		require.Equal(t, expectedOk[d], baseTtx.Debug().HasExactDomainVisibleEnd(d))
 	}
 	baseTtx.Rollback()
 	require.Equal(t, uint64(2), expectedEnd[kv.StorageDomain])