execution, db: bind StateCache fills to transaction views and reject stale fills

cmd/integration/commands/stages.go

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

db/kv/kv_interface.go

@@ -517,7 +517,14 @@ type TemporalDebugTx interface {
 	// HistoryStartFrom return the earliest known txnum in history of a given domain
 	HistoryStartFrom(domainName Domain) uint64
 
+	// DomainProgress is a best-effort progress number for reporting: it mixes
+	// an exclusive files end with an inclusive DB txNum (so it is ±1 depending
+	// on which side wins) and falls back to step granularity when history is
+	// disabled. For an exact bound use DomainVisibleEnd.
 	DomainProgress(domain Domain) (txNum uint64)
+	// 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)
 	IIProgress(name InvertedIdx) (txNum uint64)
 	StepSize() uint64
 	// Retire retires frozen history files entirely below their

db/kv/remotedb/kv_remote.go

@@ -255,6 +255,9 @@ func (tx *tx) Retire(ctx context.Context, cutoffs kv.RetireCutoffs) (int, error)
 }
 func (tx *tx) DomainFiles(domain ...kv.Domain) kv.VisibleFiles { panic("not implemented") }
 func (tx *tx) DomainProgress(domain kv.Domain) uint64          { panic("not implemented") }
+func (tx *tx) DomainVisibleEnd(domain kv.Domain) (uint64, bool) {
+	return 0, 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

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/erigontech/erigon/db/datadir"
@@ -267,13 +268,71 @@ type tx struct {
 type Tx struct {
 	kv.Tx
 	tx
+	visibleEnds domainVisibleEnds
 }
 
 type RwTx struct {
 	kv.RwTx
 	tx
 }
 
+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  [kv.DomainLen]atomic.Uint64
+	mu    sync.Mutex
+	state atomic.Uint32
+}
+
+// state packs two bits per domain into one word so a single atomic load
+// returns a consistent (loaded, ok) pair: loadedBit says ends[domain] is
+// memoized, okBit is the memoized ok answer of DomainVisibleEnd. The array
+// size asserts at compile time that both halves fit in uint32.
+var _ [32 - 2*int(kv.DomainLen)]struct{}
+
+func visibleEndBits(domain kv.Domain) (loadedBit, okBit uint32) {
+	loadedBit = uint32(1) << uint32(domain)
+	return loadedBit, loadedBit << uint32(kv.DomainLen)
+}
+
+func (v *domainVisibleEnds) get(tx *Tx, domain kv.Domain) (uint64, bool) {
+	loadedBit, okBit := visibleEndBits(domain)
+	state := v.state.Load()
+	if state&loadedBit != 0 {
+		return v.ends[domain].Load(), state&okBit != 0
+	}
+	return v.load(tx, domain, loadedBit, okBit)
+}
+
+func (v *domainVisibleEnds) load(tx *Tx, domain kv.Domain, loadedBit, okBit uint32) (uint64, bool) {
+	v.mu.Lock()
+	defer v.mu.Unlock()
+
+	state := v.state.Load()
+	if state&loadedBit == 0 {
+		end, ok := tx.aggtx.DomainVisibleEnd(domain, tx.Tx)
+		v.ends[domain].Store(end)
+		state |= loadedBit
+		if ok {
+			state |= okBit
+		}
+		v.state.Store(state)
+	}
+	return v.ends[domain].Load(), state&okBit != 0
+}
+
+// reset takes mu so an in-flight load can't re-store pre-reset bits.
+func (v *domainVisibleEnds) reset() {
+	v.mu.Lock()
+	defer v.mu.Unlock()
+	v.state.Store(0)
+}
+
 func (tx *tx) ForceReopenUnderlyingFilesTx() {
 	if tx.blocktx != nil {
 		tx.blocktx.Close()
@@ -284,6 +343,13 @@ func (tx *tx) ForceReopenUnderlyingFilesTx() {
 	}
 	tx.aggtx = tx.Agg().BeginFilesRo()
 }
+
+// ForceReopenUnderlyingFilesTx swaps in a fresh files view, which can extend
+// the visible frontier — drop the memoized ends so they are re-derived.
+func (tx *Tx) ForceReopenUnderlyingFilesTx() {
+	tx.tx.ForceReopenUnderlyingFilesTx()
+	tx.visibleEnds.reset()
+}
 func (tx *tx) FreezeInfo() kv.FreezeInfo { return tx.aggtx }
 
 func (tx *tx) AggTx() any             { return tx.aggtx }
@@ -724,6 +790,12 @@ func (tx *Tx) DomainProgress(domain kv.Domain) uint64 {
 func (tx *RwTx) DomainProgress(domain kv.Domain) uint64 {
 	return tx.aggtx.DomainProgress(domain, tx.RwTx)
 }
+func (tx *Tx) DomainVisibleEnd(domain kv.Domain) (uint64, bool) {
+	return tx.visibleEnds.get(tx, domain)
+}
+func (tx *RwTx) DomainVisibleEnd(domain kv.Domain) (uint64, bool) {
+	return tx.aggtx.DomainVisibleEnd(domain, tx.RwTx)
+}
 func (tx *Tx) IIProgress(domain kv.InvertedIdx) uint64 {
 	return tx.aggtx.IIProgress(domain, tx.Tx)
 }

db/kv/temporal/kv_temporal_test.go

@@ -2,6 +2,7 @@ package temporal
 
 import (
 	"encoding/binary"
+	"sync"
 	"testing"
 	"time"
 
@@ -257,6 +258,132 @@ func TestTemporalTx_PinsBlockFilesView(t *testing.T) {
 	require.NotNil(t, roTx2.(*Tx).blocktx)
 }
 
+// DomainVisibleEnd's memo serves repeat readers lock-free while first loads
+// run under the memo mutex. Fresh txs each round make the two paths
+// interleave across goroutines; results must stay stable (run with -race).
+func TestTemporalTx_DomainVisibleEndConcurrent(t *testing.T) {
+	t.Parallel()
+	ctx := t.Context()
+
+	mdbxDb := memdb.NewTestDB(t, dbcfg.ChainDB)
+	dirs := datadir.New(t.TempDir())
+	agg := state.NewTest(dirs).StepSize(1).MustOpen(ctx, mdbxDb)
+	defer agg.Close()
+	temporalDb, err := New(mdbxDb, agg, nil)
+	require.NoError(t, err)
+	defer temporalDb.Close()
+
+	acc := common.HexToAddress("0x1234567890123456789012345678901234567890")
+	slot := common.HexToHash("0x0000000000000000000000000000000000000000000000000000000000000001")
+	storageK := append(append([]byte{}, acc[:]...), slot[:]...)
+
+	rwTtx, err := temporalDb.BeginTemporalRw(ctx)
+	require.NoError(t, err)
+	defer rwTtx.Rollback()
+	sd, err := execctx.NewSharedDomains(ctx, rwTtx, log.Root())
+	require.NoError(t, err)
+	defer sd.Close()
+	require.NoError(t, sd.DomainPut(kv.StorageDomain, rwTtx, storageK, []byte{1}, 1, nil))
+	require.NoError(t, sd.Flush(ctx, rwTtx))
+	require.NoError(t, rwTtx.Commit())
+
+	var expectedEnd [kv.DomainLen]uint64
+	var expectedOk [kv.DomainLen]bool
+	baseTtx, err := temporalDb.BeginTemporalRo(ctx)
+	require.NoError(t, err)
+	defer baseTtx.Rollback()
+	for d := range kv.DomainLen {
+		expectedEnd[d], expectedOk[d] = baseTtx.Debug().DomainVisibleEnd(d)
+	}
+	baseTtx.Rollback()
+	require.Equal(t, uint64(2), expectedEnd[kv.StorageDomain])
+	require.True(t, expectedOk[kv.StorageDomain])
+
+	for range 25 {
+		require.NoError(t, temporalDb.ViewTemporal(ctx, func(roTtx kv.TemporalTx) error {
+			var wg sync.WaitGroup
+			for range 8 {
+				wg.Go(func() {
+					for range 4 {
+						for d := range kv.DomainLen {
+							end, ok := roTtx.Debug().DomainVisibleEnd(d)
+							if end != expectedEnd[d] || ok != expectedOk[d] {
+								t.Errorf("domain %v: got (%d, %t), want (%d, %t)", d, end, ok, expectedEnd[d], expectedOk[d])
+							}
+						}
+					}
+				})
+			}
+			wg.Wait()
+			return nil
+		}))
+	}
+}
+
+// A read-only temporal tx memoizes DomainVisibleEnd, while
+// ForceReopenUnderlyingFilesTx swaps in a fresh files view that can extend the
+// frontier — the memo must be re-derived after the swap.
+func TestTemporalTx_ForceReopenRefreshesDomainVisibleEnd(t *testing.T) {
+	t.Parallel()
+	ctx := t.Context()
+
+	mdbxDb := memdb.NewTestDB(t, dbcfg.ChainDB)
+	dirs := datadir.New(t.TempDir())
+	agg := state.NewTest(dirs).StepSize(1).MustOpen(ctx, mdbxDb)
+	defer agg.Close()
+	temporalDb, err := New(mdbxDb, agg, nil)
+	require.NoError(t, err)
+	defer temporalDb.Close()
+
+	acc := common.HexToAddress("0x1234567890123456789012345678901234567890")
+	slot := common.HexToHash("0x0000000000000000000000000000000000000000000000000000000000000001")
+	storageK := append(append([]byte{}, acc[:]...), slot[:]...)
+
+	rwTtx1, err := temporalDb.BeginTemporalRw(ctx)
+	require.NoError(t, err)
+	defer rwTtx1.Rollback()
+	sd, err := execctx.NewSharedDomains(ctx, rwTtx1, log.Root())
+	require.NoError(t, err)
+	defer sd.Close()
+	require.NoError(t, sd.DomainPut(kv.StorageDomain, rwTtx1, storageK, []byte{1}, 1, nil))
+	require.NoError(t, sd.Flush(ctx, rwTtx1))
+	require.NoError(t, rwTtx1.Commit())
+
+	roTtx, err := temporalDb.BeginTemporalRo(ctx)
+	require.NoError(t, err)
+	defer roTtx.Rollback()
+	end, ok := roTtx.Debug().DomainVisibleEnd(kv.StorageDomain)
+	require.True(t, ok)
+	require.Equal(t, uint64(2), end)
+
+	// Write past the RO tx's MVCC view and move the data into files, which are
+	// visible regardless of the DB read view.
+	for txNum := uint64(2); txNum <= 3; txNum++ {
+		rwTtx, err := temporalDb.BeginTemporalRw(ctx)
+		require.NoError(t, err)
+		defer rwTtx.Rollback()
+		require.NoError(t, sd.DomainPut(kv.StorageDomain, rwTtx, storageK, []byte{byte(txNum)}, txNum, nil))
+		require.NoError(t, sd.Flush(ctx, rwTtx))
+		require.NoError(t, rwTtx.Commit())
+	}
+	require.NoError(t, agg.BuildFiles(3))
+
+	freshRoTtx, err := temporalDb.BeginTemporalRo(ctx)
+	require.NoError(t, err)
+	defer freshRoTtx.Rollback()
+	filesEnd := freshRoTtx.Debug().TxNumsInFiles(kv.StorageDomain)
+	require.Greater(t, filesEnd, uint64(2), "the new files must extend past the memoized frontier")
+
+	end, ok = roTtx.Debug().DomainVisibleEnd(kv.StorageDomain)
+	require.True(t, ok)
+	require.Equal(t, uint64(2), end, "the pinned files view cannot see the new files before reopen")
+
+	roTtx.(*Tx).ForceReopenUnderlyingFilesTx()
+	end, ok = roTtx.Debug().DomainVisibleEnd(kv.StorageDomain)
+	require.True(t, ok)
+	require.Equal(t, filesEnd, end, "the frontier must reflect the fresh files view after reopen")
+}
+
 func TestTemporalTx_RangeAsOf_StorageDomain(t *testing.T) {
 	t.Parallel()
 	ctx := t.Context()

db/state/aggregator.go

@@ -88,9 +88,15 @@ type Aggregator struct {
 	oldestVisible *aggregatorVisible
 	// unaligned entities are left out of the shared visible-file ceiling while tooling
 	// regenerates them. Guarded by dirtyFilesLock.
-	unalignedDomain   [kv.DomainLen]bool
-	unalignedIdx      [kv.StandaloneIdxLen]bool
-	snapshotBuildSema *semaphore.Weighted
+	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 atomic.Bool
+	snapshotBuildSema           *semaphore.Weighted
 
 	disableHistory      bool
 	branchCacheDisabled bool
@@ -542,6 +548,17 @@ 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.
+// Serialized with recalcVisibleFiles via dirtyFilesLock so "from then on"
+// holds against a recalculation already in flight.
+func (a *Aggregator) ForbidVisibilityLowering() {
+	a.dirtyFilesLock.Lock()
+	defer a.dirtyFilesLock.Unlock()
+	a.visibilityLoweringForbidden.Store(true)
+}
+
 func (a *Aggregator) setUnalignedDomain(d kv.Domain, v bool) {
 	a.dirtyFilesLock.Lock()
 	defer a.dirtyFilesLock.Unlock()
@@ -683,6 +700,9 @@ func (a *Aggregator) WaitForFiles() {
 }
 
 func (a *Aggregator) Close() {
+	a.dirtyFilesLock.Lock()
+	a.visibilityLoweringForbidden.Store(false) // shutdown is not a fill window
+	a.dirtyFilesLock.Unlock()
 	a.WaitForFiles()
 	if !a.background.BeginClose() { // idempotent: safe to call Close multiple times
 		return
@@ -1852,6 +1872,28 @@ func (a *Aggregator) recalcVisibleFiles(retired retiredFiles) {
 	}
 	next.minimaxTxNum = next.stateMinimaxTxNum()
 
+	if a.visibilityLoweringForbidden.Load() {
+		prev := a.visible.Load()
+		for _, d := range []kv.Domain{kv.AccountsDomain, kv.StorageDomain, kv.CodeDomain} {
+			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))
+			}
+			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))
+			}
+		}
+	}
+
 	old := a.visible.Load()
 	old.retired = retired
 	old.next = next
@@ -2596,6 +2638,21 @@ func (at *AggregatorRoTx) DomainProgress(name kv.Domain, tx kv.Tx) uint64 {
 	}
 	return at.d[name].ht.iit.Progress(tx)
 }
+func (at *AggregatorRoTx) DomainVisibleEnd(name kv.Domain, tx kv.Tx) (uint64, bool) {
+	d := at.d[name]
+	if d.d.HistoryDisabled {
+		return 0, false
+	}
+	// A dependency checker can clamp the values view below the history-II end.
+	// Such a view has no exact frontier: reads mix fresh DB-resident keys with
+	// older file values for the gap, and raising the dependent file's
+	// visibility later reveals state without any cache apply — a fill made
+	// during the clamp would never be invalidated.
+	if d.files.EndTxNum() < d.ht.iit.files.EndTxNum() {
+		return 0, false
+	}
+	return d.ht.iit.visibleEnd(tx), true
+}
 func (at *AggregatorRoTx) IIProgress(name kv.InvertedIdx, tx kv.Tx) uint64 {
 	return at.searchII(name).Progress(tx)
 }