From a090b25eddd56ac01c2b54e2e1775ce0e78cca67 Mon Sep 17 00:00:00 2001 From: yperbasis Date: Mon, 17 Aug 2026 16:20:14 +0200 Subject: [PATCH] db/state, execution: reconcile caches at file publication --- db/state/aggregator.go | 41 ++++ db/state/aggregator_align_test.go | 69 ++++++ db/state/execctx/domain_shared.go | 50 +++-- .../execctx/domain_visible_end_memo_test.go | 10 + db/state/execctx/statecache_readfill_test.go | 31 +-- execution/cache/cache_test.go | 56 +++++ execution/cache/state_cache.go | 72 ++++++- execution/cache/view.go | 18 ++ execution/commitment/branch_cache.go | 158 +++++++++++++- execution/commitment/branch_cache_test.go | 197 +++++++++++++++++- execution/commitment/preload.go | 3 +- execution/commitment/preload_parallel.go | 3 +- 12 files changed, 655 insertions(+), 53 deletions(-) diff --git a/db/state/aggregator.go b/db/state/aggregator.go index eb946252347..ae289a3a667 100644 --- a/db/state/aggregator.go +++ b/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" ) @@ -96,6 +97,7 @@ type Aggregator struct { // domains' visible ends while set; Close clears it (shutdown is not a // fill window). visibilityLoweringForbidden atomic.Bool + boundStateCache *cache.StateCache snapshotBuildSema *semaphore.Weighted disableHistory bool @@ -560,6 +562,25 @@ func (a *Aggregator) ForbidVisibilityLowering() { a.visibilityLoweringForbidden.Store(true) } +// BindStateCache reconciles a cache with current and future file visibility. +func (a *Aggregator) BindStateCache(stateCache *cache.StateCache) { + if stateCache == nil { + return + } + a.dirtyFilesLock.Lock() + defer a.dirtyFilesLock.Unlock() + if a.boundStateCache != nil && a.boundStateCache != stateCache { + panic("assert: aggregator already has a different StateCache") + } + if stateCache.FillsEnabled() { + a.visibilityLoweringForbidden.Store(true) + } + a.boundStateCache = stateCache + if visible := a.visible.Load(); visible != nil { + a.reconcileCachesLocked(visible) + } +} + func (a *Aggregator) setUnalignedDomain(d kv.Domain, v bool) { a.dirtyFilesLock.Lock() defer a.dirtyFilesLock.Unlock() @@ -729,6 +750,7 @@ func (a *Aggregator) WaitForFiles() { func (a *Aggregator) Close() { a.dirtyFilesLock.Lock() a.visibilityLoweringForbidden.Store(false) // shutdown is not a fill window + a.boundStateCache = nil a.dirtyFilesLock.Unlock() a.WaitForFiles() if !a.background.BeginClose() { // idempotent: safe to call Close multiple times @@ -1920,6 +1942,7 @@ func (a *Aggregator) recalcVisibleFiles(retired retiredFiles) { } } } + a.reconcileCachesLocked(next) old := a.visible.Load() old.retired = retired @@ -1931,6 +1954,24 @@ func (a *Aggregator) recalcVisibleFiles(retired retiredFiles) { reclaimFiles(a.reclaimRetiredLocked()) } +// Cache reconciliation precedes visible.Store so readers cannot combine new +// files with entries from the previous visibility generation. +func (a *Aggregator) reconcileCachesLocked(visible *aggregatorVisible) { + if a.boundStateCache != nil { + a.boundStateCache.Applier().ReconcileFiles(cache.FrontierFunc(func(domain kv.Domain) (uint64, bool) { + domainVisible := visible.d[domain] + if domainVisible == nil { + return 0, false + } + return visibleFiles(domainVisible.files).EndTxNum(), true + })) + } + commitmentDomain := a.d[kv.CommitmentDomain] + if commitmentDomain != nil && commitmentDomain.branchCache != nil && visible.d[kv.CommitmentDomain] != nil { + commitmentDomain.branchCache.ReconcileFiles(visibleFiles(visible.d[kv.CommitmentDomain].files).EndTxNum()) + } +} + // stateMinimaxTxNum returns min(EndTxNum) across kv.StateDomains. Mirrors // AggregatorRoTx.TxNumsInFiles but operates directly on the bundle so the // writer can compute it without spinning up a throwaway RoTx. diff --git a/db/state/aggregator_align_test.go b/db/state/aggregator_align_test.go index 2d4be50b798..7d6d3891e57 100644 --- a/db/state/aggregator_align_test.go +++ b/db/state/aggregator_align_test.go @@ -24,6 +24,7 @@ import ( "github.com/erigontech/erigon/db/datadir" "github.com/erigontech/erigon/db/kv" + "github.com/erigontech/erigon/execution/cache" ) // generateStandaloneIIFile writes files with a hardcoded step size of 10. @@ -260,3 +261,71 @@ 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 TestFilePublicationClearsStaleBranchCache(t *testing.T) { + t.Parallel() + _, agg := testDbAndAggregatorv3(t, alignStepSize) + + branchCache := agg.d[kv.CommitmentDomain].branchCache + require.NotNil(t, branchCache) + prefix := []byte{0x01} + branchCache.Put(prefix, []byte{0xbb}, 0, 5) + _, _, ok := branchCache.Get(prefix) + 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 = branchCache.Get(prefix) + require.False(t, ok) +} + +func TestFilePublicationReconcilesStateCache(t *testing.T) { + t.Setenv("STATE_CACHE_FILLS", "true") + _, agg := testDbAndAggregatorv3(t, alignStepSize) + + stateCache := cache.NewStateCache(1<<20, 1<<20, 1<<20, 1<<20) + t.Cleanup(stateCache.Close) + agg.BindStateCache(stateCache) + + key := make([]byte, 20) + key[0] = 1 + older := stateCache.View(cache.FrontierFunc(func(kv.Domain) (uint64, bool) { return 10, true })) + older.Fill(kv.AccountsDomain, key, []byte{1}, 5) + _, ok := stateCache.View(nil).Get(kv.AccountsDomain, key) + 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 = stateCache.View(nil).Get(kv.AccountsDomain, key) + require.False(t, ok) + older.Fill(kv.AccountsDomain, key, []byte{1}, 5) + _, ok = stateCache.View(nil).Get(kv.AccountsDomain, key) + require.False(t, ok) +} + +func TestBindStateCacheReconcilesExistingFiles(t *testing.T) { + t.Setenv("STATE_CACHE_FILLS", "true") + _, 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) + agg.BindStateCache(stateCache) + + key := make([]byte, 20) + key[0] = 1 + stateCache.View(cache.FrontierFunc(func(kv.Domain) (uint64, bool) { return 10, true })).Fill( + kv.AccountsDomain, key, []byte{1}, 5) + _, ok := stateCache.View(nil).Get(kv.AccountsDomain, key) + require.False(t, ok) +} diff --git a/db/state/execctx/domain_shared.go b/db/state/execctx/domain_shared.go index 3a5429e6d54..930b9fa10cc 100644 --- a/db/state/execctx/domain_shared.go +++ b/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,15 @@ 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) +} + +func debugDomainVisibleEnd(tx kv.TemporalTx, domain kv.Domain) (uint64, bool) { + debugTx := tx.Debug() + if debugTx == nil { + return 0, false + } + return debugTx.DomainVisibleEnd(domain) } // sdFrontier adapts one (SharedDomains, tx) pair to cache.Frontier: writable @@ -946,27 +954,22 @@ 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. +// GuardAggregatorForCache binds a StateCache to immutable-file +// publication. An incompatible DB panics because omitting this fence is unsafe. func GuardAggregatorForCache(db any, sc *cache.StateCache) { - if sc == nil || !sc.FillsEnabled() { + if sc == nil { 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)) + panic(fmt.Sprintf("assert: StateCache wired over %T, which cannot produce its aggregator — file-publication coherence would be silently dropped", db)) } agg := h.Agg() - f, ok := agg.(interface{ ForbidVisibilityLowering() }) + binder, ok := agg.(interface{ BindStateCache(*cache.StateCache) }) if !ok { - panic(fmt.Sprintf("assert: aggregator %T lacks ForbidVisibilityLowering — the visibility-lowering guard would be silently dropped", agg)) + panic(fmt.Sprintf("assert: aggregator %T lacks BindStateCache — file-publication coherence would be silently dropped", agg)) } - f.ForbidVisibilityLowering() + binder.BindStateCache(sc) } // SetCodeStore sets the persistent codehash-keyed code cache. @@ -1319,17 +1322,21 @@ func (sd *SharedDomains) Commit(ctx context.Context, tx kv.RwTx, validate ...fun for i := range pendingBranches { u := &pendingBranches[i] if len(u.val) == 0 { - sd.branchCache.Invalidate(u.key) + sd.branchCache.Delete(u.key, u.txN) } else { sd.branchCache.Put(u.key, u.val, uint64(u.step), u.txN) } } + if sd.branchCache != nil { + sd.branchCache.AdvanceCommit(sd.txNum) + } if sd.stateCache != nil { if sd.cacheUnwind.pending { sd.cacheApplier.PublishUnwind(sourceStateVersion, committedStateVersion, sd.cacheUnwind.toTxNum, pendingState) } else { sd.cacheApplier.Publish(sourceStateVersion, committedStateVersion, pendingState) } + sd.cacheApplier.AdvanceCommit(sd.txNum) sd.cacheUnwind = cacheUnwindState{} } return nil @@ -1485,8 +1492,10 @@ func (sd *SharedDomains) getLatestMetered(domain kv.Domain, tx kv.TemporalTx, k // branchCache sits between sd.mem/parent.mem and the aggTx files for // CommitmentDomain only. Snapshot-isolated readers must disable it because // concurrent commits can advance the cache beyond their transaction view. + var branchView commitment.BranchCacheView if domain == kv.CommitmentDomain && sd.branchCache != nil { - if cv, cStepU64, ok := sd.branchCache.Get(k); ok { + branchView = sd.branchCache.View() + if cv, cStepU64, ok := branchView.Get(k); ok { // Get returns the on-disk step index directly — do NOT divide by // StepSize (that double-division collapsed cStep to ~0, defeating the // gate). @@ -1527,11 +1536,12 @@ func (sd *SharedDomains) getLatestMetered(domain kv.Domain, tx kv.TemporalTx, k fillView.Fill(domain, k, v, readTxNum) } } - // Only cache a branch when the read's txN is known: a txN=0 entry would - // be treated as immortal by UnwindTo, so skip the Put rather than insert - // an entry that can never be unwind-evicted. + // Only cache a branch when its txN and the read view's exact frontier are + // known. Both are needed to fence unwinds and file publication. if domain == kv.CommitmentDomain && sd.branchCache != nil && len(v) > 0 && txNKnown { - sd.branchCache.Put(k, v, uint64(step), readTxN) + if visibleEnd, ok := sd.domainVisibleEnd(tx, domain); ok { + branchView.Fill(k, v, uint64(step), readTxN, visibleEnd) + } } return v, step, nil diff --git a/db/state/execctx/domain_visible_end_memo_test.go b/db/state/execctx/domain_visible_end_memo_test.go index 0ad82a7132a..a37b4f26634 100644 --- a/db/state/execctx/domain_visible_end_memo_test.go +++ b/db/state/execctx/domain_visible_end_memo_test.go @@ -39,10 +39,20 @@ type stubVisibleEndDebug struct { viewID uint64 } +type nilDebugTemporalTx struct{ kv.TemporalTx } + +func (nilDebugTemporalTx) Debug() kv.TemporalDebugTx { return nil } + func (d stubVisibleEndDebug) DomainVisibleEnd(kv.Domain) (uint64, bool) { return d.viewID * 100, true } +func TestDebugDomainVisibleEndWithoutDebugBackend(t *testing.T) { + end, ok := debugDomainVisibleEnd(nilDebugTemporalTx{}, kv.AccountsDomain) + require.False(t, ok) + require.Zero(t, end) +} + // Parallel-exec workers share one SharedDomains and one view, so the memo // must tolerate concurrent gets interleaved with resets, and must re-derive // after a sequential view rotation. diff --git a/db/state/execctx/statecache_readfill_test.go b/db/state/execctx/statecache_readfill_test.go index cd5c0f2167c..3a5cc68e5c7 100644 --- a/db/state/execctx/statecache_readfill_test.go +++ b/db/state/execctx/statecache_readfill_test.go @@ -720,11 +720,17 @@ func TestReadFill_NegativeUsesLastVisibleTxNum(t *testing.T) { require.False(t, ok, "an unwind of the view's last included txNum must invalidate the negative") } -type fakeForbidder struct{ called bool } +type fakeCacheBinder struct { + called bool + cache *cache.StateCache +} -func (f *fakeForbidder) ForbidVisibilityLowering() { f.called = true } +func (f *fakeCacheBinder) BindStateCache(stateCache *cache.StateCache) { + f.called = true + f.cache = stateCache +} -type fakeHasAgg struct{ f *fakeForbidder } +type fakeHasAgg struct{ f *fakeCacheBinder } func (h fakeHasAgg) Agg() any { return h.f } @@ -732,33 +738,32 @@ type fakeHasBadAgg struct{} func (fakeHasBadAgg) Agg() any { return struct{}{} } -// The guard is load-bearing: for a fill-enabled cache it must either bind the -// invariant or fail loudly — never silently drop it on a DB shape mismatch. -// A nil or apply-only cache needs no guard at all. +// Every non-nil cache must bind or fail loudly because file publication can +// bypass authoritative applies even when reader fills are disabled. func TestGuardAggregatorForCache(t *testing.T) { sc := newSmallStateCache() t.Cleanup(sc.Close) - f := &fakeForbidder{} + f := &fakeCacheBinder{} execctx.GuardAggregatorForCache(fakeHasAgg{f}, sc) require.True(t, f.called) + require.Same(t, sc, f.cache) require.NotPanics(t, func() { execctx.GuardAggregatorForCache(struct{}{}, nil) }, "no cache, no invariant to bind — shape is irrelevant") require.Panics(t, func() { execctx.GuardAggregatorForCache(struct{}{}, sc) }, "a db that cannot produce its aggregator must fail loudly, not drop the guard") require.Panics(t, func() { execctx.GuardAggregatorForCache(fakeHasBadAgg{}, sc) }, - "an aggregator without ForbidVisibilityLowering must fail loudly, not drop the guard") + "an aggregator without BindStateCache must fail loudly, not drop the guard") } -// An apply-only cache (STATE_CACHE_FILLS=false) has no fills for a lowered -// frontier to poison, so the guard must not constrain the aggregator. -func TestGuardAggregatorForCache_ApplyOnlySkips(t *testing.T) { +func TestGuardAggregatorForCache_ApplyOnlyStillBindsPublication(t *testing.T) { t.Setenv("STATE_CACHE_FILLS", "false") sc := newSmallStateCache() t.Cleanup(sc.Close) - f := &fakeForbidder{} + f := &fakeCacheBinder{} execctx.GuardAggregatorForCache(fakeHasAgg{f}, sc) - require.False(t, f.called) + require.True(t, f.called) + require.Same(t, sc, f.cache) } diff --git a/execution/cache/cache_test.go b/execution/cache/cache_test.go index f0d8a7e02cd..96a84791398 100644 --- a/execution/cache/cache_test.go +++ b/execution/cache/cache_test.go @@ -995,6 +995,62 @@ func TestStateCache_ReaderAheadOfApplyWindowCanFill(t *testing.T) { "a reader ahead of appliedEnd is the normal commit-then-apply window, not a dead fork") } +func TestStateCacheReconcileFilesClearsAndFencesOlderViews(t *testing.T) { + t.Setenv("STATE_CACHE_FILLS", "true") + b := 1 * datasize.MB + sc := NewStateCache(b, b, b, b) + t.Cleanup(sc.Close) + + key := makeAddr(1) + older := sc.View(frontierAt(100)) + older.Fill(kv.AccountsDomain, key, makeValue(1), 90) + _, ok := sc.get(kv.AccountsDomain, key) + require.True(t, ok) + + sc.Applier().ReconcileFiles(frontierAt(150)) + _, ok = sc.get(kv.AccountsDomain, key) + require.False(t, ok) + + older.Fill(kv.AccountsDomain, key, makeValue(1), 90) + _, ok = sc.get(kv.AccountsDomain, key) + require.False(t, ok) + + sc.View(frontierAt(150)).Fill(kv.AccountsDomain, key, makeValue(2), 140) + value, ok := sc.get(kv.AccountsDomain, key) + require.True(t, ok) + require.Equal(t, makeValue(2), value) +} + +func TestStateCacheReconcileFilesPreservesAppliedState(t *testing.T) { + b := 1 * datasize.MB + sc := NewStateCache(b, b, b, b) + t.Cleanup(sc.Close) + + key := makeAddr(1) + sc.apply(kv.AccountsDomain, key, makeValue(1), 50) + sc.Applier().AdvanceCommit(149) + sc.Applier().ReconcileFiles(frontierAt(150)) + + value, ok := sc.get(kv.AccountsDomain, key) + require.True(t, ok) + require.Equal(t, makeValue(1), value) +} + +func TestStateCacheInitializePreservesFileFrontier(t *testing.T) { + t.Setenv("STATE_CACHE_FILLS", "true") + b := 1 * datasize.MB + sc := NewStateCache(b, b, b, b) + t.Cleanup(sc.Close) + + sc.Applier().ReconcileFiles(frontierAt(150)) + sc.Applier().Initialize(1) + key := makeAddr(1) + sc.View(frontierAtVersion(100, 1)).Fill(kv.AccountsDomain, key, makeValue(1), 90) + + _, ok := sc.get(kv.AccountsDomain, key) + require.False(t, ok) +} + func TestStateCache_PreReorgViewCannotFillAfterUnwind(t *testing.T) { b := 1 * datasize.MB sc := NewStateCache(b, b, b, b) diff --git a/execution/cache/state_cache.go b/execution/cache/state_cache.go index 6511ee31fe7..fc2823a4aad 100644 --- a/execution/cache/state_cache.go +++ b/execution/cache/state_cache.go @@ -61,13 +61,17 @@ type StateCache struct { applierMu sync.Mutex // admissionMu protects fill eligibility and publication identity. // publishing disables admission-gated fills while an update is incomplete. - admissionMu sync.RWMutex - publishing bool + admissionMu sync.RWMutex + publishing bool + // appliedEnd gates read fills after domain mutations. coveredEnd also tracks + // quiet commits, while filesEnd survives state-version resets. appliedEnd [kv.DomainLen]uint64 + coveredEnd [kv.DomainLen]uint64 + filesEnd [kv.DomainLen]uint64 stateVersion uint64 stateVersionKnown bool - // readViewEpoch lets an unwind or state-version discontinuity revoke fill - // authority from all older ReadViews. Per-cache entry epochs instead stamp + // readViewEpoch lets an unwind, file publication, or state-version + // discontinuity revoke fill authority from older ReadViews. Entry epochs stamp // stored values and also advance on Clear; sharing them would make an // ordinary Clear revoke otherwise valid read views. readViewEpoch atomic.Uint64 @@ -201,10 +205,8 @@ func (c *StateCache) putCodeSizeByHash(codeHash []byte, size int, txNum uint64) } // FillsEnabled reports whether reader fills are active (STATE_CACHE_FILLS). -// Wire-up code uses it to decide whether the backing aggregator must forbid -// visibility lowering: fill admission relies on view frontiers never -// decreasing, and apply-only caches have nothing for a lowered frontier to -// poison. +// The backing aggregator uses it to decide whether visibility lowering must be +// forbidden; immutable-file publication is reconciled in either mode. func (c *StateCache) FillsEnabled() bool { return !c.disableFills } // getAddrCodeHash returns the Ethereum codeHash for addr without an @@ -389,6 +391,9 @@ func (c *StateCache) noteApplied(domain kv.Domain, txNum uint64) { if end > c.appliedEnd[domain] { c.appliedEnd[domain] = end } + if end > c.coveredEnd[domain] { + c.coveredEnd[domain] = end + } } // clear removes all mutable entries from all caches. The admission frontier @@ -410,12 +415,60 @@ func (c *StateCache) clearLocked() { } } +func (c *StateCache) reconcileFiles(frontier Frontier) { + c.applierMu.Lock() + defer c.applierMu.Unlock() + c.admissionMu.Lock() + defer c.admissionMu.Unlock() + + extended := false + for _, domain := range kv.StateDomains { + if c.caches[domain] == nil { + continue + } + end, ok := frontier.DomainVisibleEnd(domain) + if !ok { + continue + } + if end > c.coveredEnd[domain] { + c.coveredEnd[domain] = end + c.appliedEnd[domain] = max(c.appliedEnd[domain], end) + extended = true + } + c.filesEnd[domain] = end + } + if extended { + c.readViewEpoch.Add(1) + c.clearLocked() + } +} + +func (c *StateCache) advanceCommit(txN uint64) { + c.applierMu.Lock() + defer c.applierMu.Unlock() + c.admissionMu.Lock() + defer c.admissionMu.Unlock() + + end := txN + if end < math.MaxUint64 { + end++ + } + for _, domain := range kv.StateDomains { + if c.caches[domain] != nil && end > c.coveredEnd[domain] { + c.coveredEnd[domain] = end + } + } +} + func (c *StateCache) resetForStateVersionLocked() { // Clearing entries is not enough: views bound to the previous state could // otherwise refill them after continuity was lost. c.readViewEpoch.Add(1) c.clearLocked() - clear(c.appliedEnd[:]) + for i := range c.appliedEnd { + c.appliedEnd[i] = c.filesEnd[i] + c.coveredEnd[i] = c.filesEnd[i] + } } // Close releases every sub-cache's slot in the shared memory envelope so later @@ -450,6 +503,7 @@ func (c *StateCache) unwindLocked(unwindToTxNum uint64) { } for i := range c.appliedEnd { c.appliedEnd[i] = min(c.appliedEnd[i], unwindToTxNum) + c.coveredEnd[i] = min(c.coveredEnd[i], unwindToTxNum) } } diff --git a/execution/cache/view.go b/execution/cache/view.go index dfa62543cea..c843aaafc50 100644 --- a/execution/cache/view.go +++ b/execution/cache/view.go @@ -380,3 +380,21 @@ func (a Applier) Clear() { } a.c.clear() } + +// ReconcileFiles advances admission past state published through immutable +// files and clears entries when the cache has not already applied that state. +func (a Applier) ReconcileFiles(frontier Frontier) { + if a.c == nil || frontier == nil { + return + } + a.c.reconcileFiles(frontier) +} + +// AdvanceCommit records that no domain updates were missed through txN without +// changing the per-domain fill-admission frontiers. +func (a Applier) AdvanceCommit(txN uint64) { + if a.c == nil { + return + } + a.c.advanceCommit(txN) +} diff --git a/execution/commitment/branch_cache.go b/execution/commitment/branch_cache.go index 3e980ab98c3..3d6937cce5a 100644 --- a/execution/commitment/branch_cache.go +++ b/execution/commitment/branch_cache.go @@ -19,6 +19,7 @@ package commitment import ( "bytes" "fmt" + "math" "os" "sync" "sync/atomic" @@ -76,7 +77,13 @@ type BranchCache struct { putStripes [256]sync.Mutex - coh coherence.Gen + // appliedEnd gates read-sourced writes. coveredEnd also tracks quiet commits + // so locally built files do not churn the cache. + filesEnd atomic.Uint64 + appliedEnd atomic.Uint64 + coveredEnd atomic.Uint64 + fillEpoch atomic.Uint64 + coh coherence.Gen } type branchCacheEntry struct { @@ -86,6 +93,12 @@ type branchCacheEntry struct { epoch uint32 } +// BranchCacheView binds read fills to one file-visibility generation. +type BranchCacheView struct { + cache *BranchCache + fillEpoch uint64 +} + // MissCallback runs on the hot read path when lookup misses every tier that // applies to a prefix; implementations must be lock-free. type MissCallback func(prefix []byte) @@ -504,8 +517,17 @@ func (c *BranchCache) store(prefix []byte, entry *branchCacheEntry) { c.tailForWrite().Add(maphash.Hash(prefix), entry) } -// PinEntry copies data; safe to mutate the input after the call. +// PinEntry copies data. txN identifies the state view used to load the entry. func (c *BranchCache) PinEntry(prefix []byte, data []byte, step, txN uint64) { + c.View().PinEntry(prefix, data, step, txN) +} + +// PinEntry adds a pinned entry read through this view. +func (v BranchCacheView) PinEntry(prefix []byte, data []byte, step, txN uint64) { + if v.cache == nil { + return + } + c := v.cache if isCommitmentStateKey(prefix) { return } @@ -515,6 +537,9 @@ func (c *BranchCache) PinEntry(prefix []byte, data []byte, step, txN uint64) { stripe := c.putStripe(prefix) stripe.Lock() defer stripe.Unlock() + if c.closed.Load() || v.fillEpoch != c.fillEpoch.Load() || exclusiveTxEnd(txN) < c.appliedEnd.Load() { + return + } entry := &branchCacheEntry{data: dataCopy, step: step, txN: txN, epoch: c.coh.Epoch()} var nibBuf [4]byte @@ -559,8 +584,36 @@ func (c *BranchCache) Get(prefix []byte) ([]byte, uint64, bool) { return entry.data, entry.step, true } -// Put copies the input data. +// View returns a cache handle for one backing read view. +func (c *BranchCache) View() BranchCacheView { + if c == nil { + return BranchCacheView{} + } + return BranchCacheView{cache: c, fillEpoch: c.fillEpoch.Load()} +} + +func (v BranchCacheView) Get(prefix []byte) ([]byte, uint64, bool) { + if v.cache == nil { + return nil, 0, false + } + return v.cache.Get(prefix) +} + +// Put applies an authoritative branch value and copies the input data. func (c *BranchCache) Put(prefix []byte, data []byte, step, txN uint64) { + c.put(prefix, data, step, txN, exclusiveTxEnd(txN), true, 0) +} + +// Fill offers an entry read through this view. It is dropped if the view's +// frontier or file-visibility generation is behind cache publication. +func (v BranchCacheView) Fill(prefix []byte, data []byte, step, txN, visibleEnd uint64) { + if v.cache == nil { + return + } + v.cache.put(prefix, data, step, txN, visibleEnd, false, v.fillEpoch) +} + +func (c *BranchCache) put(prefix []byte, data []byte, step, txN, sourceEnd uint64, authoritative bool, fillEpoch uint64) { if isCommitmentStateKey(prefix) { return } @@ -570,6 +623,15 @@ func (c *BranchCache) Put(prefix []byte, data []byte, step, txN uint64) { stripe := c.putStripe(prefix) stripe.Lock() defer stripe.Unlock() + if c.closed.Load() || !authoritative && fillEpoch != c.fillEpoch.Load() { + return + } + if sourceEnd < c.appliedEnd.Load() { + return + } + if authoritative { + c.advanceAppliedEnd(sourceEnd) + } c.store(prefix, &branchCacheEntry{ data: dataCopy, @@ -579,6 +641,68 @@ func (c *BranchCache) Put(prefix []byte, data []byte, step, txN uint64) { }) } +func exclusiveTxEnd(txN uint64) uint64 { + if txN == math.MaxUint64 { + return txN + } + return txN + 1 +} + +func (c *BranchCache) advanceAppliedEnd(end uint64) { + for current := c.appliedEnd.Load(); end > current; current = c.appliedEnd.Load() { + if c.appliedEnd.CompareAndSwap(current, end) { + break + } + } + c.advanceCoveredEnd(end) +} + +func (c *BranchCache) advanceCoveredEnd(end uint64) { + for current := c.coveredEnd.Load(); end > current; current = c.coveredEnd.Load() { + if c.coveredEnd.CompareAndSwap(current, end) { + return + } + } +} + +func (c *BranchCache) lowerAppliedEnd(end uint64) { + for current := c.appliedEnd.Load(); end < current; current = c.appliedEnd.Load() { + if c.appliedEnd.CompareAndSwap(current, end) { + break + } + } + for current := c.coveredEnd.Load(); end < current; current = c.coveredEnd.Load() { + if c.coveredEnd.CompareAndSwap(current, end) { + return + } + } +} + +// AdvanceCommit records a successful authoritative commit without treating a +// quiet range as a branch mutation. +func (c *BranchCache) AdvanceCommit(txN uint64) { + if c == nil { + return + } + c.advanceCoveredEnd(exclusiveTxEnd(txN)) +} + +// Delete applies an authoritative branch deletion. +func (c *BranchCache) Delete(prefix []byte, txN uint64) { + if c == nil || isCommitmentStateKey(prefix) { + return + } + end := exclusiveTxEnd(txN) + stripe := c.putStripe(prefix) + stripe.Lock() + defer stripe.Unlock() + if end < c.appliedEnd.Load() { + return + } + c.advanceAppliedEnd(end) + c.Invalidate(prefix) +} + func (c *BranchCache) Invalidate(prefix []byte) { if isRootPrefix(prefix) { c.root.Store(nil) @@ -606,14 +730,42 @@ func (c *BranchCache) Invalidate(prefix []byte) { // Unwind is O(1) and scan-free: it bumps the epoch and lowers the unwind // floor, so stale entries drop lazily on their next Get. func (c *BranchCache) Unwind(unwindToTxN uint64) { + c.lockAllPutStripes() + defer c.unlockAllPutStripes() + c.fillEpoch.Add(1) c.coh.Unwind(unwindToTxN) + c.lowerAppliedEnd(unwindToTxN) +} + +// ReconcileFiles invalidates entries when visible files advance beyond the +// latest authoritative state observed by this cache. +func (c *BranchCache) ReconcileFiles(filesEnd uint64) { + if c == nil || c.closed.Load() || filesEnd == c.filesEnd.Load() { + return + } + c.lockAllPutStripes() + defer c.unlockAllPutStripes() + if c.closed.Load() || filesEnd == c.filesEnd.Load() { + return + } + if filesEnd < c.filesEnd.Load() || filesEnd > c.coveredEnd.Load() { + c.fillEpoch.Add(1) + c.clearLocked() + c.advanceAppliedEnd(filesEnd) + } + c.filesEnd.Store(filesEnd) + c.advanceCoveredEnd(filesEnd) } // Clear holds every writer stripe so a publication cannot cross generations. func (c *BranchCache) Clear() { c.lockAllPutStripes() defer c.unlockAllPutStripes() + c.fillEpoch.Add(1) + c.clearLocked() +} +func (c *BranchCache) clearLocked() { c.root.Store(nil) c.clearTrunk() c.pinned.Store(nil) diff --git a/execution/commitment/branch_cache_test.go b/execution/commitment/branch_cache_test.go index 7fba0094be4..f5f4fe7b5ed 100644 --- a/execution/commitment/branch_cache_test.go +++ b/execution/commitment/branch_cache_test.go @@ -200,7 +200,7 @@ func TestBranchCache_ClearRacingPut_EpochAlias(t *testing.T) { require.False(t, ok, "pre-Clear epoch must not alias the live epoch after a later unwind") } -func clearDuringBlockedBranchCacheWrite(c *BranchCache, block *sync.Mutex, write func()) { +func fenceDuringBlockedBranchCacheWrite(block *sync.Mutex, write, fence func()) { // GOMAXPROCS=1 makes each Gosched yield deterministically to the queued goroutine. previousProcs := runtime.GOMAXPROCS(1) defer runtime.GOMAXPROCS(previousProcs) @@ -218,7 +218,7 @@ func clearDuringBlockedBranchCacheWrite(c *BranchCache, block *sync.Mutex, write clearDone := make(chan struct{}) go func() { - c.Clear() + fence() close(clearDone) }() runtime.Gosched() @@ -234,9 +234,9 @@ func TestBranchCache_ClearFencesStartedPut(t *testing.T) { c.Unwind(300) key := []byte{0x12, 0x34, 0x56} - clearDuringBlockedBranchCacheWrite(c, &c.tailMu, func() { + fenceDuringBlockedBranchCacheWrite(&c.tailMu, func() { c.Put(key, []byte("dead-fork-branch"), 0, 200) - }) + }, c.Clear) _, _, ok := c.Get(key) require.False(t, ok, "Clear must remove a Put that started in the retiring generation") @@ -249,14 +249,199 @@ func TestBranchCache_ClearFencesStartedPinEntry(t *testing.T) { key := make([]byte, 33) key[32] = 1 - clearDuringBlockedBranchCacheWrite(c, &c.pinnedMu, func() { + fenceDuringBlockedBranchCacheWrite(&c.pinnedMu, func() { c.PinEntry(key, []byte("dead-fork-branch"), 0, 200) - }) + }, c.Clear) _, _, ok := c.Get(key) require.False(t, ok, "Clear must remove a PinEntry that started in the retiring generation") } +func TestBranchCacheReconcileFilesPreservesAppliedState(t *testing.T) { + t.Parallel() + c := NewBranchCache(64) + t.Cleanup(c.Close) + + prefix := []byte{0x01} + c.Put(prefix, []byte("branch"), 0, 100) + c.ReconcileFiles(101) + + _, _, ok := c.Get(prefix) + require.True(t, ok) +} + +func TestBranchCacheReconcileFilesClearsReadFills(t *testing.T) { + t.Parallel() + c := NewBranchCache(64) + t.Cleanup(c.Close) + + prefix := []byte{0x01} + c.ReconcileFiles(100) + c.View().Fill(prefix, []byte("branch"), 0, 90, 100) + _, _, ok := c.Get(prefix) + require.True(t, ok) + + c.ReconcileFiles(150) + _, _, ok = c.Get(prefix) + require.False(t, ok) +} + +func TestBranchCacheRejectsReadFillBehindPublication(t *testing.T) { + t.Parallel() + c := NewBranchCache(64) + t.Cleanup(c.Close) + + prefix := []byte{0x01} + c.ReconcileFiles(150) + c.View().Fill(prefix, []byte("stale"), 0, 90, 100) + _, _, ok := c.Get(prefix) + require.False(t, ok) + + c.View().Fill(prefix, []byte("current"), 0, 140, 150) + value, _, ok := c.Get(prefix) + require.True(t, ok) + require.Equal(t, []byte("current"), value) +} + +func TestBranchCacheRejectsAuthoritativePutBehindPublication(t *testing.T) { + t.Parallel() + c := NewBranchCache(64) + t.Cleanup(c.Close) + + prefix := []byte{0x01} + c.ReconcileFiles(150) + c.Put(prefix, []byte("stale"), 0, 100) + + _, _, ok := c.Get(prefix) + require.False(t, ok) +} + +func TestBranchCacheRejectsPinBehindPublication(t *testing.T) { + t.Parallel() + c := NewBranchCache(64) + t.Cleanup(c.Close) + + prefix := make([]byte, 33) + prefix[32] = 1 + c.ReconcileFiles(150) + c.PinEntry(prefix, []byte("stale"), 0, 100) + _, _, ok := c.Get(prefix) + require.False(t, ok) + + c.PinEntry(prefix, []byte("current"), 0, 150) + value, _, ok := c.Get(prefix) + require.True(t, ok) + require.Equal(t, []byte("current"), value) +} + +func TestBranchCacheDeleteAdvancesAppliedFrontier(t *testing.T) { + t.Parallel() + c := NewBranchCache(64) + t.Cleanup(c.Close) + + kept := []byte{0x01} + deleted := []byte{0x02} + c.Put(kept, []byte("kept"), 0, 50) + c.Put(deleted, []byte("deleted"), 0, 50) + c.Delete(deleted, 100) + c.ReconcileFiles(101) + + _, _, ok := c.Get(kept) + require.True(t, ok) + _, _, ok = c.Get(deleted) + require.False(t, ok) +} + +func TestBranchCacheAdvanceCommitPreservesStateAcrossQuietRange(t *testing.T) { + t.Parallel() + c := NewBranchCache(64) + t.Cleanup(c.Close) + + prefix := []byte{0x01} + c.Put(prefix, []byte("branch"), 0, 50) + c.AdvanceCommit(100) + c.ReconcileFiles(101) + + _, _, ok := c.Get(prefix) + require.True(t, ok) +} + +func TestBranchCacheQuietCommitKeepsUnchangedFillEligible(t *testing.T) { + t.Parallel() + c := NewBranchCache(64) + t.Cleanup(c.Close) + + prefix := []byte{0x01} + c.ReconcileFiles(100) + c.AdvanceCommit(149) + c.View().Fill(prefix, []byte("branch"), 0, 90, 100) + + _, _, ok := c.Get(prefix) + require.True(t, ok) +} + +func TestBranchCacheReconcileFilesFencesStartedFill(t *testing.T) { + c := NewBranchCache(100) + t.Cleanup(c.Close) + c.ReconcileFiles(100) + + key := []byte{0x12, 0x34, 0x56} + view := c.View() + fenceDuringBlockedBranchCacheWrite(&c.tailMu, func() { + view.Fill(key, []byte("stale"), 0, 90, 100) + }, func() { + c.ReconcileFiles(150) + }) + + _, _, ok := c.Get(key) + require.False(t, ok) +} + +func TestBranchCacheRejectsReadFillAcrossVisibilityLowering(t *testing.T) { + t.Parallel() + c := NewBranchCache(64) + t.Cleanup(c.Close) + + c.ReconcileFiles(100) + older := c.View() + c.ReconcileFiles(50) + older.Fill([]byte{0x01}, []byte("stale"), 0, 90, 100) + + _, _, ok := c.Get([]byte{0x01}) + require.False(t, ok) +} + +func TestBranchCacheRejectsPinAcrossVisibilityLowering(t *testing.T) { + t.Parallel() + c := NewBranchCache(64) + t.Cleanup(c.Close) + + prefix := make([]byte, 33) + prefix[32] = 1 + c.ReconcileFiles(100) + older := c.View() + c.ReconcileFiles(50) + older.PinEntry(prefix, []byte("stale"), 0, 100) + + _, _, ok := c.Get(prefix) + require.False(t, ok) +} + +func TestBranchCacheUnwindLowersAppliedFrontier(t *testing.T) { + t.Parallel() + c := NewBranchCache(64) + t.Cleanup(c.Close) + + prefix := []byte{0x01} + c.Put(prefix, []byte("old fork"), 0, 100) + c.Unwind(50) + c.Put(prefix, []byte("new fork"), 0, 50) + + value, _, ok := c.Get(prefix) + require.True(t, ok) + require.Equal(t, []byte("new fork"), value) +} + func TestBranchCache_Stats(t *testing.T) { c := NewBranchCache(100) tailHit := []byte{0x12, 0x34, 0x56} diff --git a/execution/commitment/preload.go b/execution/commitment/preload.go index 3d07fe59357..8aac8010896 100644 --- a/execution/commitment/preload.go +++ b/execution/commitment/preload.go @@ -69,6 +69,7 @@ func (p *ContractTrunkPreload) Run( if additionalBudgetBytes <= 0 { return 0, len(p.queue) == 0, nil } + cacheView := cache.View() chunkUsedBytes := 0 chunkPinned := 0 @@ -94,7 +95,7 @@ func (p *ContractTrunkPreload) Run( } p.queue = p.queue[1:] - cache.PinEntry(prefix, v, step, p.pinTxNum) + cacheView.PinEntry(prefix, v, step, p.pinTxNum) p.pinnedPrefixes = append(p.pinnedPrefixes, prefix) chunkUsedBytes += entryCost chunkPinned++ diff --git a/execution/commitment/preload_parallel.go b/execution/commitment/preload_parallel.go index 2a1a3eaf2d3..2e641d0b280 100644 --- a/execution/commitment/preload_parallel.go +++ b/execution/commitment/preload_parallel.go @@ -132,6 +132,7 @@ func (p *ContractTrunkPreloadParallel) Run( if stepBudgetBytes <= 0 { return 0, p.queueEmpty(), nil } + cacheView := cache.View() defer p.releaseScratch() stepCap := p.usedBytes + stepBudgetBytes @@ -148,7 +149,7 @@ func (p *ContractTrunkPreloadParallel) Run( // source step, and the pinTxNum stamp already gives unwind coherence — the // floor drops a preloaded pin before the cStep<=maxStep gate is consulted, // so leaving step unset only keeps that gate trivially true for live pins. - cache.PinEntry(pk.key, v, 0, p.pinTxNum) + cacheView.PinEntry(pk.key, v, 0, p.pinTxNum) p.pinnedPrefixes = append(p.pinnedPrefixes, pk.key) p.usedBytes += cost p.pinned++