execution, db: keep StateCache coherent through startup catchup; batch applies

cmd/integration/commands/stages.go

@@ -851,9 +851,9 @@ func execBlocksBatch(ctx context.Context, db kv.TemporalRwDB, st *stagedsync.Syn
 	}
 	defer doms.Close()
 	doms.SetInMemHistoryReads(false)
+	stateCache.BindAggregator(db)
 	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

@@ -658,6 +658,9 @@ type TemporalRwDB interface {
 	BeginTemporalRw(ctx context.Context) (TemporalRwTx, error)
 	BeginTemporalRwNosync(ctx context.Context) (TemporalRwTx, error)
 	UpdateTemporal(ctx context.Context, f func(tx TemporalRwTx) error) error
+	// Agg returns the DB's state-files aggregator as `any` (the concrete type
+	// lives above the kv layer); nil when the DB has none.
+	Agg() any
 }
 
 // ---- non-important utilities

db/kv/membatchwithdb/memory_mutation.go

@@ -1341,6 +1341,8 @@ func (td temporaldb) BeginTemporalRwNosync(ctx context.Context) (kv.TemporalRwTx
 	return td.memoryMutation, nil
 }
 
+func (td temporaldb) Agg() any { return nil }
+
 func (td temporaldb) Debug() kv.TemporalDebugDB {
 	panic("not implemented")
 }

db/seg/decompress.go

@@ -196,7 +196,7 @@ type Decompressor struct {
 	readAheadRefcnt atomic.Int32 // ref-counter: allow enable/disable read-ahead from goroutines. only when refcnt=0 - disable read-ahead once
 
 	residency     atomic.Pointer[residencyBitmap] // page-residency bitmap for the async-io gate; nil unless enabled
-	residencyOnce sync.Once
+	residencyOnce sync.Once                       //nolint:unused // Used by the Linux residency gate.
 }
 
 const (

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,17 @@ 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)
+}
+
+// debugDomainVisibleEnd tolerates txs without a debug backend (MemoryMutation
+// over a nil db): no exact frontier means no fills, reads still work.
+func debugDomainVisibleEnd(tx kv.TemporalTx, domain kv.Domain) (uint64, bool) {
+	dbgTx := tx.Debug()
+	if dbgTx == nil {
+		return 0, false
+	}
+	return dbgTx.DomainVisibleEnd(domain)
 }
 
 // sdFrontier adapts one (SharedDomains, tx) pair to cache.Frontier: writable
@@ -934,6 +944,9 @@ func (sd *SharedDomains) SetStateCache(stateCache *cache.StateCache) {
 	if !dbg.UseStateCache || stateCache == nil {
 		return
 	}
+	if stateCache.FillsEnabled() && !stateCache.AggregatorBound() {
+		panic("assert: fill-enabled StateCache wired before BindAggregator — the visibility-lowering guard is not bound")
+	}
 	sd.BindStateCache(stateCache)
 }
 
@@ -946,29 +959,6 @@ 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.
-func GuardAggregatorForCache(db any, sc *cache.StateCache) {
-	if sc == nil || !sc.FillsEnabled() {
-		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))
-	}
-	agg := h.Agg()
-	f, ok := agg.(interface{ ForbidVisibilityLowering() })
-	if !ok {
-		panic(fmt.Sprintf("assert: aggregator %T lacks ForbidVisibilityLowering — the visibility-lowering guard would be silently dropped", agg))
-	}
-	f.ForbidVisibilityLowering()
-}
-
 // SetCodeStore sets the persistent codehash-keyed code cache.
 func (sd *SharedDomains) SetCodeStore(codeStore *cache.CodeStore) {
 	sd.codeStore = codeStore
@@ -1070,22 +1060,15 @@ func (sd *SharedDomains) Close() {
 // admission.
 
 // Flush writes the in-memory batch into tx without committing. It deliberately
-// does NOT touch the caches: plain Flush leaves the commit to the caller (who
-// may still roll back), so it must not warm a cache with state that could be
-// rolled back. Cache entries are populated elsewhere — by Commit after a
-// successful commit, and by reads (GetLatest) — each stamped with a
-// conservative upper-bound txNum. It is that txNum stamp, not population
-// timing, that keeps the cache correct: an unwind lowers the floor so every
-// entry reflecting a now-dead fork is evicted, and mem-first masking means a
-// later in-memory write shadows a stale cached read.
-//
-// An SD with an attached state cache must route every flush through Commit:
-// Flush neither applies nor invalidates, so a populated cache would keep
-// serving pre-flush values for the flushed keys after the caller's own
-// commit — and Commit collects its cache updates only from its own flush, so
-// an earlier plain Flush's keys would never be applied. Cache-less callers
-// may Flush and commit themselves.
+// does not touch the caches — the caller may still roll back. An SD with a
+// state cache must route every flush through Commit: a plain Flush would
+// leave the cache serving pre-flush values for the flushed keys forever, so
+// it panics here, like the SetStateCache assert for the neighbouring wiring
+// bug — an error return can be swallowed.
 func (sd *SharedDomains) Flush(ctx context.Context, tx kv.RwTx) error {
+	if sd.stateCache != nil {
+		panic("assert: SharedDomains with a state cache must flush through Commit")
+	}
 	defer mxFlushTook.ObserveDuration(time.Now())
 	return sd.flushMem(ctx, tx)
 }
@@ -1513,7 +1496,8 @@ func (sd *SharedDomains) getLatestMetered(domain kv.Domain, tx kv.TemporalTx, k
 	}
 
 	// A bounded read observes a staged unwind, not stable committed state.
-	if maxStep == kv.NoStepBound && sd.stateCache != nil && sd.stateCache.Caches(domain) {
+	// Apply-only mode skips frontier binding for fills that cannot happen.
+	if maxStep == kv.NoStepBound && sd.stateCache != nil && sd.stateCache.FillsEnabled() && sd.stateCache.Caches(domain) {
 		readTxNum := (uint64(step)+1)*sd.StepSize() - 1
 		fillView := view
 		if fillView.NeedsFrontier() {
@@ -1752,7 +1736,7 @@ func (sd *SharedDomains) codeHashForAddr(tx kv.TemporalTx, view cache.ReadView,
 	}
 
 	h, fromReadView := resolve()
-	if fromReadView && sd.stateCache != nil {
+	if fromReadView && sd.stateCache != nil && sd.stateCache.FillsEnabled() {
 		var fixed [32]byte
 		if len(h) == 32 {
 			copy(fixed[:], h)