execution/cache, db/state/execctx, execution/exec: applied-progress watermark for stale-snapshot cache fills

db/state/execctx/domain_shared.go

@@ -1034,6 +1034,9 @@ func (sd *SharedDomains) Commit(ctx context.Context, tx kv.RwTx, validate ...fun
 	}
 	for i := range pending {
 		u := &pending[i]
+		if u.domain != kv.CommitmentDomain {
+			sd.stateCache.NoteApplied(u.domain, u.txN)
+		}
 		switch u.domain {
 		case kv.CommitmentDomain:
 			if len(u.val) == 0 {
@@ -1046,6 +1049,7 @@ func (sd *SharedDomains) Commit(ctx context.Context, tx kv.RwTx, validate ...fun
 				// Deletions are authoritative nil puts — a tombstone here, the
 				// no-code marker for the code binding — so a straddling
 				// pre-delete read-fill defers instead of resurrecting the value.
+				sd.stateCache.NoteApplied(kv.CodeDomain, u.txN)
 				sd.stateCache.Put(kv.AccountsDomain, u.key, nil, u.txN)
 				sd.stateCache.Put(kv.CodeDomain, u.key, nil, u.txN)
 				sd.stateCache.DeleteAddrCodeHash(u.key)
@@ -1241,28 +1245,33 @@ func (sd *SharedDomains) getLatestMetered(domain kv.Domain, tx kv.TemporalTx, k
 
 	// Populate the cache with if-absent semantics: a read-fill never carries
 	// newer information than a flush-apply, so it must not overwrite one
-	// (e.g. an embedded-RPC read straddling an FCU commit). Stamp with the
-	// last txNum of the step the value came from — an upper bound on its
-	// write txNum — so an unwind below it can't leave the entry stale. A
-	// negative carries no step; stamp it with the domain's progress at
-	// observation time so any unwind drops it.
-	if sd.stateCache != nil {
-		readTxNum := (uint64(step)+1)*sd.StepSize() - 1
-		if domain == kv.CodeDomain {
-			if len(v) > 0 {
-				// This SD getter is the single place that populates the code cache
-				// on a read. Key the content-addressed entry by the code's OWN hash,
-				// keccak(v) — NEVER a separately-read account codeHash, which under
-				// parallel exec can be a skewed or cross-account value and would
-				// poison the shared codeHash→code map for every account sharing
-				// that hash.
-				sd.stateCache.PutCodeWithHashIfAbsent(k, v, crypto.Keccak256(v), readTxNum)
-			}
-		} else {
-			if len(v) == 0 && sd.stateCache.GetCache(domain) != nil {
-				readTxNum = tx.Debug().DomainProgress(domain)
+	// (e.g. an embedded-RPC read straddling an FCU commit). Fill only from a
+	// snapshot at least as fresh as the last flush-apply: a staler reader
+	// could resurrect a deleted key once its tombstone is evicted — the
+	// watermark, unlike the tombstone, cannot be. Stamp with the last txNum of
+	// the step the value came from — an upper bound on its write txNum — so an
+	// unwind below it can't leave the entry stale. A negative carries no step;
+	// stamp it with the snapshot's progress so any unwind drops it.
+	if sd.stateCache != nil && sd.stateCache.GetCache(domain) != nil {
+		snapshotProgress := tx.Debug().DomainProgress(domain)
+		if snapshotProgress >= sd.stateCache.AppliedProgress(domain) {
+			readTxNum := (uint64(step)+1)*sd.StepSize() - 1
+			if domain == kv.CodeDomain {
+				if len(v) > 0 {
+					// This SD getter is the single place that populates the code cache
+					// on a read. Key the content-addressed entry by the code's OWN hash,
+					// keccak(v) — NEVER a separately-read account codeHash, which under
+					// parallel exec can be a skewed or cross-account value and would
+					// poison the shared codeHash→code map for every account sharing
+					// that hash.
+					sd.stateCache.PutCodeWithHashIfAbsent(k, v, crypto.Keccak256(v), readTxNum)
+				}
+			} else {
+				if len(v) == 0 {
+					readTxNum = snapshotProgress
+				}
+				sd.stateCache.PutIfAbsent(domain, k, v, readTxNum)
 			}
-			sd.stateCache.PutIfAbsent(domain, k, v, readTxNum)
 		}
 	}
 	// Only cache a branch when the read's txN is known: a txN=0 entry would
@@ -1437,15 +1446,17 @@ func (sd *SharedDomains) codeHashForAddr(tx kv.TemporalTx, addr []byte, txNum ui
 	}
 
 	h := resolve()
-	if sd.stateCache != nil {
+	// Populate — including the zero-hash sentinel for misses, so repeat
+	// lookups skip the whole resolve() chain — but only from a snapshot at
+	// least as fresh as the last flush-apply (see the read-fill gate). txNum
+	// is a conservative upper bound (>= the resolved account's write txNum),
+	// so the mapping drops on any unwind that reverts that account.
+	if sd.stateCache != nil &&
+		tx.Debug().DomainProgress(kv.AccountsDomain) >= sd.stateCache.AppliedProgress(kv.AccountsDomain) {
 		var fixed [32]byte
 		if len(h) == 32 {
 			copy(fixed[:], h)
 		}
-		// Always populate, including the zero-hash sentinel for misses —
-		// repeat lookups skip the whole resolve() chain. txNum is a
-		// conservative upper bound (>= the resolved account's write txNum), so
-		// the mapping drops on any unwind that reverts that account.
 		sd.stateCache.PutAddrCodeHash(addr, fixed, txNum)
 	}
 	return h

db/state/execctx/statecache_readfill_test.go

@@ -352,3 +352,95 @@ func TestReadFill_DoesNotResurrectDeletedCode(t *testing.T) {
 	require.True(t, ok, "the deletion must be cached as a live no-code marker")
 	require.Equal(t, uint64(20), cTxNum)
 }
+
+// commitDelete seeds one key with v at txNum 10 and deletes it at txNum 20,
+// each in its own committed SD, then returns a pre-delete read snapshot.
+func commitDelete(t *testing.T, db kv.TemporalRwDB, sc *cache.StateCache, domain kv.Domain, key, v []byte) kv.TemporalTx {
+	t.Helper()
+	ctx := t.Context()
+
+	rwTx1, err := db.BeginTemporalRw(ctx)
+	require.NoError(t, err)
+	defer rwTx1.Rollback()
+	sd1, err := execctx.NewSharedDomains(ctx, rwTx1, log.New())
+	require.NoError(t, err)
+	defer sd1.Close()
+	sd1.SetStateCacheForTest(sc)
+	sd1.SetTxNum(10)
+	require.NoError(t, sd1.DomainPut(domain, rwTx1, key, v, 10, nil))
+	require.NoError(t, sd1.Commit(ctx, rwTx1))
+
+	roTxOld, err := db.BeginTemporalRo(ctx)
+	require.NoError(t, err)
+	t.Cleanup(roTxOld.Rollback)
+
+	rwTx2, err := db.BeginTemporalRw(ctx)
+	require.NoError(t, err)
+	defer rwTx2.Rollback()
+	sd2, err := execctx.NewSharedDomains(ctx, rwTx2, log.New())
+	require.NoError(t, err)
+	defer sd2.Close()
+	sd2.SetStateCacheForTest(sc)
+	sd2.SetTxNum(20)
+	require.NoError(t, sd2.DomainDel(domain, rwTx2, key, 20, v))
+	require.NoError(t, sd2.Commit(ctx, rwTx2))
+
+	return roTxOld
+}
+
+// A tombstone or no-code marker is an ordinary LRU entry: cache pressure can
+// evict it while a pre-delete snapshot is still alive, and that snapshot's
+// fill then finds the key absent. The applied-progress watermark —
+// unevictable — must reject the stale snapshot's fill regardless.
+func TestReadFill_DoesNotResurrectAfterMarkerEviction(t *testing.T) {
+	if testing.Short() {
+		t.Skip()
+	}
+	t.Parallel()
+
+	for _, domain := range []kv.Domain{kv.AccountsDomain, kv.CodeDomain} {
+		t.Run(domain.String(), func(t *testing.T) {
+			t.Parallel()
+			const stepSize = uint64(16)
+			ctx := t.Context()
+			db := newTestDb(t, stepSize)
+			b := 1 * datasize.KB // entry caps clamp to their minimums — cheap to pressure
+			sc := cache.NewStateCache(b, b, b, b)
+
+			key := make([]byte, 20)
+			key[0] = 0xdd
+			v := encAccount(1)
+
+			roTxOld := commitDelete(t, db, sc, domain, key, v)
+
+			// Evict the deletion marker with cache pressure. Check only after
+			// the burst — a per-insert Get would keep the marker MRU.
+			pressure := make([]byte, 20)
+			for i := 0; i < 8192; i++ {
+				binary.BigEndian.PutUint64(pressure[1:], uint64(i))
+				sc.Put(domain, pressure, v, 10)
+			}
+			_, stillThere := sc.Get(domain, key)
+			require.False(t, stillThere, "pressure must evict the deletion marker")
+
+			// The straddling reader's fill runs against the marker-less cache.
+			sdOld, err := execctx.NewSharedDomains(ctx, roTxOld, log.New())
+			require.NoError(t, err)
+			defer sdOld.Close()
+			sdOld.SetStateCacheForTest(sc)
+			_, _, err = sdOld.GetLatest(domain, roTxOld, key)
+			require.NoError(t, err)
+
+			roTxNew, err := db.BeginTemporalRo(ctx)
+			require.NoError(t, err)
+			defer roTxNew.Rollback()
+			sdNew, err := execctx.NewSharedDomains(ctx, roTxNew, log.New())
+			require.NoError(t, err)
+			defer sdNew.Close()
+			sdNew.SetStateCacheForTest(sc)
+			got, _, err := sdNew.GetLatest(domain, roTxNew, key)
+			require.NoError(t, err)
+			require.Empty(t, got, "a stale snapshot's fill must not resurrect the deleted value once its marker is evicted")
+		})
+	}
+}

execution/cache/cache_test.go

@@ -948,3 +948,22 @@ func TestStateCache_UnwindAssertsWarmupInFlight(t *testing.T) {
 	defer sc.WarmupDone()
 	require.NotPanics(t, func() { sc.Unwind(10) })
 }
+
+// The applied-progress watermark follows flush-applies up (monotonically),
+// unwinds down, and Clear to zero, independently per domain.
+func TestStateCache_AppliedProgressWatermark(t *testing.T) {
+	b := 1 * datasize.MB
+	sc := NewStateCache(b, b, b, b)
+	require.Zero(t, sc.AppliedProgress(kv.AccountsDomain))
+
+	sc.NoteApplied(kv.AccountsDomain, 20)
+	sc.NoteApplied(kv.AccountsDomain, 10)
+	require.Equal(t, uint64(20), sc.AppliedProgress(kv.AccountsDomain))
+	require.Zero(t, sc.AppliedProgress(kv.StorageDomain))
+
+	sc.Unwind(15)
+	require.Equal(t, uint64(15), sc.AppliedProgress(kv.AccountsDomain))
+
+	sc.Clear()
+	require.Zero(t, sc.AppliedProgress(kv.AccountsDomain))
+}

execution/cache/state_cache.go

@@ -61,6 +61,12 @@ type StateCache struct {
 	// racing the epoch bump could stamp a dead-fork value with the post-unwind
 	// epoch and have it served as canonical.
 	warmupsInFlight atomic.Int64
+
+	// appliedProgress is the per-domain txNum watermark of flush-applied
+	// commits. A fill from a snapshot behind it must be skipped: the reader's
+	// view may predate a deletion, and the deletion's tombstone defends its
+	// key only while resident in the LRU — the watermark cannot be evicted.
+	appliedProgress [kv.DomainLen]atomic.Uint64
 }
 
 // NewStateCache creates a new StateCache with the specified byte capacities.
@@ -273,6 +279,9 @@ func (c *StateCache) Clear() {
 			cache.Clear()
 		}
 	}
+	for i := range c.appliedProgress {
+		c.appliedProgress[i].Store(0)
+	}
 }
 
 // Close releases every sub-cache's slot in the shared memory envelope so later
@@ -304,6 +313,27 @@ func (c *StateCache) Unwind(unwindToTxNum uint64) {
 			cache.Unwind(unwindToTxNum)
 		}
 	}
+	for i := range c.appliedProgress {
+		if c.appliedProgress[i].Load() > unwindToTxNum {
+			c.appliedProgress[i].Store(unwindToTxNum)
+		}
+	}
+}
+
+// NoteApplied raises domain's applied-progress watermark to txNum; the flush
+// cache-apply calls it for every update it lands.
+func (c *StateCache) NoteApplied(domain kv.Domain, txNum uint64) {
+	for {
+		cur := c.appliedProgress[domain].Load()
+		if txNum <= cur || c.appliedProgress[domain].CompareAndSwap(cur, txNum) {
+			return
+		}
+	}
+}
+
+// AppliedProgress returns domain's applied-progress watermark; see the field.
+func (c *StateCache) AppliedProgress(domain kv.Domain) uint64 {
+	return c.appliedProgress[domain].Load()
 }
 
 // WarmupStarted and WarmupDone bracket a fire-and-forget cache-populating

execution/exec/blocks_read_ahead.go

@@ -97,38 +97,39 @@ func newCachePopulatingGetter(ttx kv.TemporalTx, sc *cache.StateCache) *cachePop
 
 func (cpg *cachePopulatingGetter) GetLatest(name kv.Domain, k []byte) ([]byte, kv.Step, error) {
 	v, step, err := cpg.g.GetLatest(name, k)
-	if err == nil && cpg.sc != nil {
+	if err == nil && cpg.sc != nil && cpg.progress != nil {
 		// If-absent writes only: this runs in a fire-and-forget goroutine over a
 		// committed snapshot, so an unconditional Put racing an FCU flush's
 		// cache-apply could replace the flushed value with the pre-flush one.
-		if name == kv.CodeDomain {
-			// A live binding makes the conditional put a no-op — skip before
-			// paying the keccak+copy below. Code negatives end here too: they
-			// are not cacheable (CodeCache drops zero-length puts).
-			if len(v) > 0 && !cpg.sc.HasLiveCode(k) {
-				// Key the content cache by keccak(v), the code's own hash — never
-				// a separately read account codeHash, which parallel exec can skew
-				// (see the code-domain read-fill in SharedDomains.getLatestMetered).
-				cpg.sc.PutCodeWithHashIfAbsent(k, v, crypto.Keccak256(v), (uint64(step)+1)*cpg.stepSize-1)
-			}
-		} else {
-			// Cache including nil/empty results: a probe returning no bytes is
-			// a valid negative answer (missing account, empty storage slot) and
-			// caching it lets repeated probes skip the file accessor stack —
-			// revm's CacheAccount { account: None, status: LoadedNotExisting }
-			// pattern. Stamp with the last txNum of the value's step; a
-			// negative has no step — use the domain's progress at observation
-			// time so any unwind drops it.
-			txNum := (uint64(step)+1)*cpg.stepSize - 1
-			if len(v) == 0 {
-				if cpg.progress == nil {
-					// No progress oracle → no honest stamp; skip rather than
-					// cache an unwind-immortal negative.
-					return v, step, err
+		// Fill only from a snapshot at least as fresh as the last flush-apply —
+		// see the SD read-fill gate; a getter without a progress oracle cannot
+		// prove freshness and never fills.
+		snap := cpg.progress(name)
+		if snap >= cpg.sc.AppliedProgress(name) {
+			if name == kv.CodeDomain {
+				// A live binding makes the conditional put a no-op — skip before
+				// paying the keccak+copy below. Code negatives end here too: they
+				// are not cacheable (CodeCache drops zero-length puts).
+				if len(v) > 0 && !cpg.sc.HasLiveCode(k) {
+					// Key the content cache by keccak(v), the code's own hash — never
+					// a separately read account codeHash, which parallel exec can skew
+					// (see the code-domain read-fill in SharedDomains.getLatestMetered).
+					cpg.sc.PutCodeWithHashIfAbsent(k, v, crypto.Keccak256(v), (uint64(step)+1)*cpg.stepSize-1)
+				}
+			} else {
+				// Cache including nil/empty results: a probe returning no bytes is
+				// a valid negative answer (missing account, empty storage slot) and
+				// caching it lets repeated probes skip the file accessor stack —
+				// revm's CacheAccount { account: None, status: LoadedNotExisting }
+				// pattern. Stamp with the last txNum of the value's step; a
+				// negative has no step — use the snapshot's progress so any unwind
+				// drops it.
+				txNum := (uint64(step)+1)*cpg.stepSize - 1
+				if len(v) == 0 {
+					txNum = snap
 				}
-				txNum = cpg.progress(name)
+				cpg.sc.PutIfAbsent(name, k, v, txNum)
 			}
-			cpg.sc.PutIfAbsent(name, k, v, txNum)
 		}
 	}
 	return v, step, err

execution/exec/blocks_read_ahead_test.go

@@ -60,7 +60,7 @@ func TestCachePopulatingGetterKeepsFresherEntry(t *testing.T) {
 	for _, domain := range []kv.Domain{kv.AccountsDomain, kv.StorageDomain} {
 		sc := newTestStateCache()
 		sc.Put(domain, key, fresh, 54)
-		cpg := &cachePopulatingGetter{g: stubTemporalGetter{v: stale}, sc: sc, stepSize: 1_562_500}
+		cpg := &cachePopulatingGetter{g: stubTemporalGetter{v: stale}, sc: sc, stepSize: 1_562_500, progress: zeroProgress}
 
 		v, _, err := cpg.GetLatest(domain, key)
 		require.NoError(t, err)
@@ -80,7 +80,7 @@ func TestCachePopulatingGetterKeepsFresherCodeBinding(t *testing.T) {
 	staleCode := []byte{0xbb, 0x04, 0x05, 0x06}
 	sc := newTestStateCache()
 	sc.PutCodeWithHash(addr, freshCode, crypto.Keccak256(freshCode), 54)
-	cpg := &cachePopulatingGetter{g: stubTemporalGetter{v: staleCode}, sc: sc, stepSize: 1_562_500}
+	cpg := &cachePopulatingGetter{g: stubTemporalGetter{v: staleCode}, sc: sc, stepSize: 1_562_500, progress: zeroProgress}
 
 	_, _, err := cpg.GetLatest(kv.CodeDomain, addr)
 	require.NoError(t, err)
@@ -163,9 +163,9 @@ func TestCachePopulatingGetterNegativeDropsOnUnwind(t *testing.T) {
 	require.False(t, ok, "a negative observed at txNum 10M must not survive an unwind to 5M")
 }
 
-// A getter constructed without a progress oracle must skip caching negatives
-// (an honest stamp is impossible), not panic.
-func TestCachePopulatingGetterNilProgressSkipsNegative(t *testing.T) {
+// A getter constructed without a progress oracle cannot prove its snapshot's
+// freshness and must not fill at all, let alone panic.
+func TestCachePopulatingGetterNilProgressNeverFills(t *testing.T) {
 	key := []byte("\x11\x22\x33\x44\x55\x66\x77\x88\x99\xaa\xbb\xcc\xdd\xee\xff\x00\x11\x22\x33\x44")
 	sc := newTestStateCache()
 	cpg := &cachePopulatingGetter{g: stubTemporalGetter{v: nil}, sc: sc, stepSize: 1_562_500}
@@ -174,7 +174,23 @@ func TestCachePopulatingGetterNilProgressSkipsNegative(t *testing.T) {
 		require.NoError(t, err)
 	})
 	_, ok := sc.Get(kv.AccountsDomain, key)
-	require.False(t, ok, "no progress oracle — the negative must not be cached")
+	require.False(t, ok, "no progress oracle — nothing may be cached")
+}
+
+// The applied-progress watermark rejects fills from snapshots older than the
+// last flush-apply — the eviction-proof backstop behind the tombstones.
+func TestCachePopulatingGetterStaleSnapshotDoesNotFill(t *testing.T) {
+	key := []byte("\x11\x22\x33\x44\x55\x66\x77\x88\x99\xaa\xbb\xcc\xdd\xee\xff\x00\x11\x22\x33\x44")
+	sc := newTestStateCache()
+	sc.NoteApplied(kv.AccountsDomain, 20)
+	cpg := &cachePopulatingGetter{
+		g: stubTemporalGetter{v: []byte("pre-delete-record")}, sc: sc, stepSize: 1_562_500,
+		progress: func(kv.Domain) uint64 { return 10 }, // snapshot older than the applied commit
+	}
+	_, _, err := cpg.GetLatest(kv.AccountsDomain, key)
+	require.NoError(t, err)
+	_, ok := sc.Get(kv.AccountsDomain, key)
+	require.False(t, ok, "a snapshot behind the applied watermark must not fill")
 }
 
 func zeroProgress(kv.Domain) uint64 { return 0 }