From 4eacd876bf0d0abfb255f63915c6623055fe6363 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 13 Sep 2026 19:23:44 +0200 Subject: [PATCH] Publish observers after the gate, gate every store entry point, stamp under one hold Observers and OnUpdate never run under the cache gate. Admit queues changes while the gate is held; the outermost entry point (attach, pull, commit, upsert, remove) publishes them after releasing it. A store's direct PullAll hands over an empty load outside its lock so the cache publishes then. A throwing observer can no longer split store from cache: publication starts only after the store has adopted its loaded view. The directory store adopts its durable catalog only after the cache admits the load; the write-only _durableIndex is deleted. Single-file Push, Delete, PushAll and CommitBatch take the gate like the directory store; Dispose flushes under it. Write mints storedAt and admits under one gate hold, and Land re-stamps staged upserts under the gate that admits them, so a concurrent pull cannot land a later storedAt in between. Tests: ObserverDoesNotRunUnderTheGate, ThrowingOnUpdateHandlerDoesNotResurrectDeletedRecord, RepullNeverErasesAStagedWriteOnDirectoryStore, DirectCallsOnAnAttachedStoreShareTheCacheGate; the directory last-writer result is pinned to {a,c,x}; the single-file re-pull test waits for the write to reach the store or block instead of sleeping. Pausing tests always release the pause so a failure cannot deadlock Dispose. Docs: observer rule, direct store calls share the gate, directory TryCommit contention and gate wait, per-key last-writer-wins, directory flush merge. Co-Authored-By: Claude Opus 5 --- .../DirectoryMessagePackBackingStore.cs | 14 +- .../Contracts/cultcache-store-composition.md | 25 +- src/GameCult.Caching/CultCache.cs | 302 +++++++++++------- .../ConditionalCommitTests.cs | 208 +++++++++++- 4 files changed, 417 insertions(+), 132 deletions(-) diff --git a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs index 7c32500f..f72c2343 100644 --- a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs +++ b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs @@ -20,7 +20,6 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore private readonly ConcurrentDictionary _dirtyKeys = new(StringComparer.Ordinal); private readonly ConcurrentDictionary _deletedKeys = new(StringComparer.Ordinal); private readonly HashSet _hydratedKeys = new(StringComparer.Ordinal); - private Dictionary _durableIndex = new(StringComparer.Ordinal); private CultSchemaCatalogEntry[] _durableCatalog = Array.Empty(); public DirectoryMessagePackBackingStore(string manifestPath, string? recordDirectoryPath = null, bool readOnly = false) @@ -43,6 +42,8 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore { lock (Gate) PullAllCore(); + // Outside the gate, an empty hand-over lets the cache publish what this pull admitted. + Loaded?.Invoke(Array.Empty(), Array.Empty()); } private void PullAllCore() @@ -61,17 +62,13 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore var reports = new List(); var loaded = new Dictionary(StringComparer.Ordinal); + CultSchemaCatalogEntry[] catalog; using (AcquireCommitLease(wait: true)) { // Only pages named by the manifest read under the lease are loaded; orphaned pages are never loaded. var manifest = ReadManifest(); Trace($"manifest records={manifest.Records.Length}"); - _durableCatalog = manifest.SchemaCatalog; - _durableIndex = manifest.Records - .Where(record => !string.IsNullOrWhiteSpace(record.Key)) - .GroupBy(record => record.Key, StringComparer.Ordinal) - .ToDictionary(group => group.Key, group => group.Last(), StringComparer.Ordinal); - + catalog = manifest.SchemaCatalog; LoadRecordPages( manifest.Records.OrderBy(record => record.Key, StringComparer.Ordinal).ToArray(), manifest.SchemaCatalog, @@ -95,6 +92,8 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore if (arrived.Length > 0 || departed.Length > 0) Loaded?.Invoke(arrived, departed); + // Adopted only once the cache admitted the load; a refused load leaves the store's durable view untouched. + _durableCatalog = catalog; foreach (var stored in departed) Entries.TryRemove(stored.Key.Value, out _); _hydratedKeys.RemoveWhere(key => !loaded.ContainsKey(key) && !Staged(key)); @@ -252,7 +251,6 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore DeleteUnreferencedRecordPages(currentIndex.Values); _durableCatalog = targetCatalog; - _durableIndex = currentIndex; _dirtyKeys.Clear(); _deletedKeys.Clear(); MarkFlushSucceeded(); diff --git a/src/GameCult.Caching/Contracts/cultcache-store-composition.md b/src/GameCult.Caching/Contracts/cultcache-store-composition.md index 0dc29d44..645468c2 100644 --- a/src/GameCult.Caching/Contracts/cultcache-store-composition.md +++ b/src/GameCult.Caching/Contracts/cultcache-store-composition.md @@ -44,6 +44,10 @@ which route wrote it. single file keeps staged writes dirty and does not drop loaded records. - A load publishes changes to `Watch` subscribers and fires `OnUpdate`. A local write or commit publishes to `Watch` only; `OnUpdate` fires for loads alone. +- Observers and `OnUpdate` run after the cache releases its gate, never under + it: an observer may read or write the cache from any thread. A throwing + observer cannot undo the store's adoption of a load; the store and cache + already agree when publication starts. - Hydration failure on open is loud: a corrupt store file, or a record whose schema the registry cannot resolve, makes the open throw and leaves the file byte-identical. Consumers never delete and rewrite a store they failed to @@ -58,6 +62,9 @@ which route wrote it. so that cache's readers wait for the I/O. The file lock (`.lock` or the directory commit lease) is always taken inside the gate and released before a load is handed to the cache. +- Direct calls on an attached store (`Push`, `Delete`, `PushAll`, `CommitBatch`, + `PullAll`) take the same gate, so they wait for that cache's pull, flush or + commit, and the cache's flush on dispose runs under it. ## Dirtiness: loading never writes @@ -134,13 +141,19 @@ replaces; writers bump a minted timestamp by one tick when it is not later. - A failed condition is a lost race, not an error: `Commit` returns false and nothing is written, changed in memory, or published. `TryCommit` makes one - non-blocking lock attempt and reports `Contended` instead of waiting. + non-blocking lock attempt and reports `Contended` instead of waiting. On a + directory store the lock is the commit lease that pulls also take, so + `Contended` can mean another cache or process is reading or committing. The + attempt itself runs inside the cache gate, so `TryCommit` may first block + behind a pull or flush of the same cache, for up to that operation's lease + wait bound (30 seconds). - A conditional commit on a store with staged single-record writes throws: flush first. - Single-file stores lock a sidecar `.lock` opened exclusively. Directory stores use their existing commit lease. -**A plain flush and an unconditional commit are last-writer-wins.** Both run +**A plain flush and an unconditional commit are last-writer-wins: per file for +single-file stores, per key for directory stores.** Both run under the same lock, so two writers never interleave bytes, but they compare nothing. Only a conditional commit (`Expect` or `ExpectUnchanged`) protects against another writer; processes sharing a store must all use it. @@ -197,6 +210,8 @@ no serialization; a CultMath encoding, if added, uses this shape. A "contract" cell means this document is ahead of the C# code. Opening a store always hydrates it. A consumer that wants a store to hold only -what it writes removes the records it does not keep, upserts, and flushes; the -flush replaces the file in one atomic step. There is no open-without-reading -option. +what it writes removes the records it does not keep, upserts, and flushes. On a +single-file store the flush replaces the file in one atomic step. On a directory +store the flush writes pages, then the manifest, under the commit lease, merging +its staged keys and removals onto the current manifest; a key another writer +added since the pull survives. There is no open-without-reading option. diff --git a/src/GameCult.Caching/CultCache.cs b/src/GameCult.Caching/CultCache.cs index f6a5c484..428ebf93 100644 --- a/src/GameCult.Caching/CultCache.cs +++ b/src/GameCult.Caching/CultCache.cs @@ -1179,6 +1179,7 @@ namespace GameCult.Caching private readonly ConditionalWeakTable _handles = new(); private readonly Subject _changes = new(); private readonly object _gate = new(); + private List<(Change Change, bool Loaded)> _unpublished = new(); private bool _dirtyInMemory; public CultCache(CultDocumentRegistry? registry = null) @@ -1252,59 +1253,74 @@ namespace GameCult.Caching { if (store == null) throw new ArgumentNullException(nameof(store)); homes ??= Array.Empty(); - lock (_gate) + try { - if (store.Loaded != null) - throw new InvalidOperationException($"Backing store {store} is already attached to a cache."); - if (store.IsDirty) - throw new InvalidOperationException($"Backing store {store} has staged writes; attach it clean."); - if (homes.Length == 0 && _stores.Any(entry => entry.Homes.Length == 0)) - throw new InvalidOperationException( - $"Backing store {store} would be a second untyped store; name the types it is home to."); - foreach (var home in homes) + lock (_gate) { - var owner = _stores.FirstOrDefault(entry => entry.Homes.Contains(home)).Store; - if (owner != null) - throw new InvalidOperationException($"{home.FullName} is already routed to {owner}; it cannot also route to {store}."); - } - - var candidate = (store, homes); - foreach (var type in _entries.Values.Select(entry => entry.Descriptor).Distinct()) - { - var before = Home(type.DocumentType); - _stores.Add(candidate); - var after = Home(type.DocumentType); - _stores.RemoveAt(_stores.Count - 1); - if (before != after) + if (store.Loaded != null) + throw new InvalidOperationException($"Backing store {store} is already attached to a cache."); + if (store.IsDirty) + throw new InvalidOperationException($"Backing store {store} has staged writes; attach it clean."); + if (homes.Length == 0 && _stores.Any(entry => entry.Homes.Length == 0)) throw new InvalidOperationException( - $"Attaching {store} would move {type.SchemaName} from {before?.ToString() ?? "memory"} to {after}; " + - "attach routed stores before the untyped store."); - } + $"Backing store {store} would be a second untyped store; name the types it is home to."); + foreach (var home in homes) + { + var owner = _stores.FirstOrDefault(entry => entry.Homes.Contains(home)).Store; + if (owner != null) + throw new InvalidOperationException($"{home.FullName} is already routed to {owner}; it cannot also route to {store}."); + } - store.AttachRegistry(_registry); - store.Gate = _gate; - store.Loaded = (loaded, dropped) => Admit(loaded, dropped, store, _ => CultCommitOutcome.Committed); - _stores.Add(candidate); - try - { - store.PullAll(); - } - catch - { - _stores.Remove(candidate); - store.Loaded = null; - store.Gate = new object(); - throw; + var candidate = (store, homes); + foreach (var type in _entries.Values.Select(entry => entry.Descriptor).Distinct()) + { + var before = Home(type.DocumentType); + _stores.Add(candidate); + var after = Home(type.DocumentType); + _stores.RemoveAt(_stores.Count - 1); + if (before != after) + throw new InvalidOperationException( + $"Attaching {store} would move {type.SchemaName} from {before?.ToString() ?? "memory"} to {after}; " + + "attach routed stores before the untyped store."); + } + + store.AttachRegistry(_registry); + store.Gate = _gate; + store.Loaded = (loaded, dropped) => Admit(loaded, dropped, store, _ => CultCommitOutcome.Committed); + _stores.Add(candidate); + try + { + store.PullAll(); + } + catch + { + _stores.Remove(candidate); + store.Loaded = null; + store.Gate = new object(); + throw; + } } } + finally + { + Publish(); + } } public Task PullAllBackingStoresAsync() { foreach (var store in BackingStores) { - lock (_gate) - store.PullAll(); + try + { + // Our stores take the gate themselves; a third-party store might not, so the cache takes it here. + lock (_gate) + store.PullAll(); + } + finally + { + Publish(); + } } return Task.CompletedTask; @@ -1348,18 +1364,24 @@ namespace GameCult.Caching public bool Remove(CultRecordKey key) { - CultStoredDocument? existing; - lock (_gate) - _entries.TryGetValue(key.Value, out existing); - if (existing == null) - return false; - - Admit(Array.Empty(), new[] { existing }, null, home => + try { - home?.Delete(existing); - return CultCommitOutcome.Committed; - }); - return true; + lock (_gate) + { + if (!_entries.TryGetValue(key.Value, out var existing)) + return false; + Admit(Array.Empty(), new[] { existing }, null, home => + { + home?.Delete(existing); + return CultCommitOutcome.Committed; + }); + return true; + } + } + finally + { + Publish(); + } } public void Remove(CultRecordHandle handle) @@ -1431,8 +1453,11 @@ namespace GameCult.Caching public void Dispose() { - if (FlushAttachedStoresOnDispose && IsDirty) - FlushAllBackingStores(); + lock (_gate) + { + if (FlushAttachedStoresOnDispose && IsDirty) + FlushAllBackingStores(); + } foreach (var store in BackingStores) store.Dispose(); @@ -1453,16 +1478,18 @@ namespace GameCult.Caching { var descriptor = _registry.GetRequired(document.GetType()); lock (_gate) - { - var resolved = key - ?? (_handles.TryGetValue(document, out var box) ? box.Key - : descriptor.IsGlobal ? new CultRecordKey($"global:{descriptor.SchemaId}") - : new CultRecordKey(Guid.NewGuid().ToString("N"))); - _entries.TryGetValue(resolved.Value, out var previous); - return new CultStoredDocument(resolved, MintStoredAt(previous?.StoredAt), descriptor, document); - } + return Stamp( + key ?? (_handles.TryGetValue(document, out var box) ? box.Key + : descriptor.IsGlobal ? new CultRecordKey($"global:{descriptor.SchemaId}") + : new CultRecordKey(Guid.NewGuid().ToString("N"))), + descriptor, + document); } + // Call under the gate that admits the result, so no load can land a later storedAt in between. + private CultStoredDocument Stamp(CultRecordKey key, CultDocumentDescriptor descriptor, object document) => + new(key, MintStoredAt(_entries.TryGetValue(key.Value, out var previous) ? previous.StoredAt : null), descriptor, document); + internal CultStoredDocument? Observe(CultRecordKey key, object? current) { if (current == null) @@ -1489,13 +1516,23 @@ namespace GameCult.Caching private CultRecordKey Write(object document, CultRecordKey? key) { - var stored = CreateStoredDocument(document, key); - Admit(new[] { stored }, Array.Empty(), null, home => + try { - home?.Push(stored); - return CultCommitOutcome.Committed; - }); - return stored.Key; + lock (_gate) + { + var stored = CreateStoredDocument(document, key); + Admit(new[] { stored }, Array.Empty(), null, home => + { + home?.Push(stored); + return CultCommitOutcome.Committed; + }); + return stored.Key; + } + } + finally + { + Publish(); + } } private CultCommitOutcome Land(Action stage, bool wait) @@ -1512,30 +1549,41 @@ namespace GameCult.Caching } // The in-process gate is always entered; only the store lock attempt honors wait, so Contended means another writer holds the store. - lock (_gate) + try { - var upserts = batch.Operations.Values.Where(op => op.Stored != null).Select(op => op.Stored!).ToArray(); - var deletes = batch.Operations.Values - .Where(op => op.Stored == null && _entries.ContainsKey(op.Key.Value)) - .Select(op => _entries[op.Key.Value]) - .ToArray(); - var request = new CultCommitRequest(upserts, deletes, batch.Expected.ToArray(), batch.ExpectsUnchanged); - return Admit(upserts, deletes, null, home => + lock (_gate) { - if (home != null) - { - if (request.HasConditions && home.IsDirty) - throw new InvalidOperationException($"Backing store {home} has staged writes; flush before a conditional commit."); - return home.CommitBatch(request, wait); - } - - if (request.HasConditions && _stores.Count > 0) - throw new InvalidOperationException("A conditional batch names its home store through a record it upserts or removes."); - var inMemory = _entries.Values - .Select(entry => new CultPersistedRecord { Key = entry.Key.Value, SchemaId = entry.Descriptor.SchemaId, StoredAt = entry.StoredAt }) + // Staging stamped outside this gate; a pull since then may hold a later storedAt, so stamp again. + var upserts = batch.Operations.Values + .Where(op => op.Stored != null) + .Select(op => Stamp(op.Key, op.Stored!.Descriptor, op.Stored.Document)) .ToArray(); - return request.ConditionsHold(inMemory, _entries.Values) ? CultCommitOutcome.Committed : CultCommitOutcome.Mismatch; - }); + var deletes = batch.Operations.Values + .Where(op => op.Stored == null && _entries.ContainsKey(op.Key.Value)) + .Select(op => _entries[op.Key.Value]) + .ToArray(); + var request = new CultCommitRequest(upserts, deletes, batch.Expected.ToArray(), batch.ExpectsUnchanged); + return Admit(upserts, deletes, null, home => + { + if (home != null) + { + if (request.HasConditions && home.IsDirty) + throw new InvalidOperationException($"Backing store {home} has staged writes; flush before a conditional commit."); + return home.CommitBatch(request, wait); + } + + if (request.HasConditions && _stores.Count > 0) + throw new InvalidOperationException("A conditional batch names its home store through a record it upserts or removes."); + var inMemory = _entries.Values + .Select(entry => new CultPersistedRecord { Key = entry.Key.Value, SchemaId = entry.Descriptor.SchemaId, StoredAt = entry.StoredAt }) + .ToArray(); + return request.ConditionsHold(inMemory, _entries.Values) ? CultCommitOutcome.Committed : CultCommitOutcome.Mismatch; + }); + } + } + finally + { + Publish(); } } @@ -1547,23 +1595,44 @@ namespace GameCult.Caching CacheBackingStore? source, Func land) { - List changes; + try + { + lock (_gate) + { + var outcome = land(Validate(admitted, evicted, source)); + if (outcome != CultCommitOutcome.Committed) + return outcome; + foreach (var change in Apply(admitted, evicted, source)) + _unpublished.Add((change, source != null)); + return CultCommitOutcome.Committed; + } + } + finally + { + Publish(); + } + } + + // Observers never run under the gate: admitted changes wait until this thread has left its outermost hold. + private void Publish() + { + if (Monitor.IsEntered(_gate)) + return; + List<(Change Change, bool Loaded)> pending; lock (_gate) { - var outcome = land(Validate(admitted, evicted, source)); - if (outcome != CultCommitOutcome.Committed) - return outcome; - changes = Apply(admitted, evicted, source); + if (_unpublished.Count == 0) + return; + pending = _unpublished; + _unpublished = new(); } - foreach (var change in changes) + foreach (var (change, loaded) in pending) { _changes.OnNext(change); - if (source != null) + if (loaded) OnUpdate?.Invoke(change.Previous, change.Document); } - - return CultCommitOutcome.Committed; } private CacheBackingStore? Validate( @@ -1894,6 +1963,8 @@ namespace GameCult.Caching { lock (Gate) PullAllCore(); + // Outside the gate, an empty hand-over lets the cache publish what this pull admitted. + Loaded?.Invoke(Array.Empty(), Array.Empty()); } private void PullAllCore() @@ -1938,15 +2009,21 @@ namespace GameCult.Caching public override void Push(CultStoredDocument entry) { ThrowIfReadOnly(); - Entries[entry.Key.Value] = entry; - IsDirty = true; + lock (Gate) + { + Entries[entry.Key.Value] = entry; + IsDirty = true; + } } public override void Delete(CultStoredDocument entry) { ThrowIfReadOnly(); - Entries.TryRemove(entry.Key.Value, out _); - IsDirty = true; + lock (Gate) + { + Entries.TryRemove(entry.Key.Value, out _); + IsDirty = true; + } } // A plain flush writes this store's whole view under the lock: two writers never interleave bytes, but the last @@ -1954,19 +2031,28 @@ namespace GameCult.Caching public override void PushAll() { ThrowIfReadOnly(); - using (AcquireLock(wait: true)) + lock (Gate) { - WriteSnapshot( - Entries.Values.Select(entry => ToPersistedRecord(entry, SerializePayload)), - Entries.Values.Select(entry => entry.Descriptor.ToCatalogEntry())); - } + using (AcquireLock(wait: true)) + { + WriteSnapshot( + Entries.Values.Select(entry => ToPersistedRecord(entry, SerializePayload)), + Entries.Values.Select(entry => entry.Descriptor.ToCatalogEntry())); + } - MarkFlushSucceeded(); + MarkFlushSucceeded(); + } } public override CultCommitOutcome CommitBatch(CultCommitRequest request, bool wait) { ThrowIfReadOnly(); + lock (Gate) + return CommitBatchCore(request, wait); + } + + private CultCommitOutcome CommitBatchCore(CultCommitRequest request, bool wait) + { using var fileLock = AcquireLock(wait); if (fileLock == null) return CultCommitOutcome.Contended; diff --git a/tests/GameCult.Caching.Tests/ConditionalCommitTests.cs b/tests/GameCult.Caching.Tests/ConditionalCommitTests.cs index 4a3ce596..e3a70c9a 100644 --- a/tests/GameCult.Caching.Tests/ConditionalCommitTests.cs +++ b/tests/GameCult.Caching.Tests/ConditionalCommitTests.cs @@ -344,20 +344,24 @@ namespace GameCult.Caching.Tests using var cache = new CultCache(Registry); cache.AddBackingStore(store); var staged = new CultRecordKey("staged"); - Task pulling; - Task writing; using (var release = new ManualResetEventSlim()) { store.Pause = release; - pulling = Task.Run(() => cache.PullAllBackingStoresAsync()); - Assert.That(store.Entered.Wait(TimeSpan.FromSeconds(10)), Is.True, "the pull never read the file"); - writing = Task.Run(() => cache.UpsertAsync(new Counter { Name = "staged" }, new CultRecordHandle(staged))); - // Before the fix the write lands inside the paused pull; after it, the write waits for the pull. - writing.Wait(200); - store.Pause = null; - release.Set(); - Assert.That(Task.WaitAll(new[] { pulling, writing }, TimeSpan.FromSeconds(10)), Is.True); + try + { + var pulling = Task.Run(() => cache.PullAllBackingStoresAsync()); + Assert.That(store.Entered.Wait(TimeSpan.FromSeconds(10)), Is.True, "the pull never read the file"); + var writer = StartWriter(() => cache.UpsertAsync(new Counter { Name = "staged" }, new CultRecordHandle(staged)), () => store.Pushed.IsSet); + store.Pause = null; + release.Set(); + Assert.That(pulling.Wait(TimeSpan.FromSeconds(10)) && writer(), Is.True); + } + finally + { + store.Pause = null; + release.Set(); + } } Assert.That(cache.Get(staged), Is.Not.Null, "the re-pull erased a staged write from the cache"); @@ -366,6 +370,152 @@ namespace GameCult.Caching.Tests Assert.That(reader.Get(staged), Is.Not.Null, "the flush after the re-pull lost the staged write"); } + // The pull pauses inside the directory store's own page load, under the commit lease. + [Test] + public void RepullNeverErasesAStagedWriteOnDirectoryStore() + { + var store = DirectoryStore(PathOf("repull-dir.cc")); + using (var seed = store.Open()) + Assert.That(seed.Commit(batch => batch.Upsert(new Slow { Name = "slow", Blocks = true })), Is.True); + using var cache = store.Open(); + var staged = new CultRecordKey("staged"); + + using (var release = new ManualResetEventSlim()) + { + Slow.Entered.Reset(); + Slow.Pause = release; + try + { + var pulling = Task.Run(() => cache.PullAllBackingStoresAsync()); + Assert.That(Slow.Entered.Wait(TimeSpan.FromSeconds(10)), Is.True, "the pull never loaded a page"); + var writer = StartWriter(() => cache.UpsertAsync(new Counter { Name = "staged" }, new CultRecordHandle(staged)), () => false); + Slow.Pause = null; + release.Set(); + Assert.That(pulling.Wait(TimeSpan.FromSeconds(10)) && writer(), Is.True); + } + finally + { + Slow.Pause = null; + release.Set(); + } + } + + Assert.That(cache.Get(staged), Is.Not.Null, "the re-pull erased a staged write from the cache"); + cache.FlushAllBackingStores(); + using var reader = store.Open(); + Assert.That(reader.Get(staged), Is.Not.Null, "the flush after the re-pull lost the staged write"); + } + + // Gate sharing is what makes a direct store call wait for its cache's I/O instead of interleaving with it. + [Test] + public void DirectCallsOnAnAttachedStoreShareTheCacheGate() + { + var path = PathOf("direct.cc"); + Create(SingleFile(path), 0); + var store = new PausingStore(path); + using var cache = new CultCache(Registry); + cache.AddBackingStore(store); + + using var release = new ManualResetEventSlim(); + store.Pause = release; + try + { + var pulling = Task.Run(() => cache.PullAllBackingStoresAsync()); + Assert.That(store.Entered.Wait(TimeSpan.FromSeconds(10)), Is.True, "the pull never read the file"); + var pushing = Task.Run(() => store.PushAll()); + Assert.That(pushing.Wait(200), Is.False, "a direct PushAll ran while the cache's pull held the gate"); + store.Pause = null; + release.Set(); + Assert.That(Task.WaitAll(new[] { pulling, pushing }, TimeSpan.FromSeconds(10)), Is.True); + } + finally + { + // A failed assert must not leave the pull parked on the gate that Dispose needs. + store.Pause = null; + release.Set(); + } + } + + [Test] + public void ObserverDoesNotRunUnderTheGate() + { + var path = PathOf("observer.cc"); + using var cache = SingleFile(path).Open(); + var gated = new System.Collections.Concurrent.ConcurrentQueue(); + var seen = 0; + using var subscription = cache.Watch().Subscribe(change => + { + Interlocked.Increment(ref seen); + if (!Task.Run(() => cache.Get(change.Key)).Wait(TimeSpan.FromSeconds(5))) + gated.Enqueue(change.Key.Value); + }); + + cache.UpsertAsync(new Counter { Name = "upserted" }, new CultRecordHandle(new CultRecordKey("upserted"))); + Assert.That(cache.Commit(batch => batch.Upsert(new Counter { Name = "committed" }, new CultRecordHandle(new CultRecordKey("committed")))), Is.True); + using (var other = SingleFile(path).Open()) + Assert.That(other.Commit(batch => batch.Upsert(new Counter { Name = "pulled" }, new CultRecordHandle(new CultRecordKey("pulled")))), Is.True); + cache.PullAllBackingStoresAsync(); + + Assert.That(seen, Is.EqualTo(3)); + Assert.That(gated, Is.Empty, "an observer ran while the cache gate was held"); + } + + [Test] + public void ThrowingOnUpdateHandlerDoesNotResurrectDeletedRecord() + { + var store = SingleFile(PathOf("resurrect.cc")); + var doomed = new CultRecordKey("doomed"); + var later = new CultRecordKey("later"); + using (var seed = store.Open()) + Assert.That(seed.Commit(batch => + { + batch.Upsert(new Counter { Name = "doomed" }, new CultRecordHandle(doomed)); + batch.Upsert(new Counter { Name = "kept" }); + }), Is.True); + using var a = store.Open(); + using (var b = store.Open()) + { + Assert.That(b.Remove(doomed), Is.True); + b.FlushAllBackingStores(); + } + + var throwing = true; + a.OnUpdate += (_, _) => + { + if (throwing) + throw new InvalidOperationException("observer failed"); + }; + Assert.That(() => a.PullAllBackingStoresAsync(), Throws.InvalidOperationException.With.Message.EqualTo("observer failed")); + throwing = false; + Assert.That(a.Get(doomed), Is.Null); + + a.UpsertAsync(new Counter { Name = "later" }, new CultRecordHandle(later)); + a.FlushAllBackingStores(); + using var reader = store.Open(); + Assert.That(reader.Get(later), Is.Not.Null); + Assert.That(reader.Get(doomed), Is.Null, "a throwing observer left the store holding a record the cache dropped"); + } + + // Runs write on its own thread and returns once it has reached the store or is blocked waiting; the result joins it. + private static Func StartWriter(Action write, Func reachedStore) + { + Exception? failure = null; + var thread = new Thread(() => + { + try { write(); } + catch (Exception exception) { failure = exception; } + }); + thread.Start(); + var waited = System.Diagnostics.Stopwatch.StartNew(); + while (!reachedStore() && thread.IsAlive && (thread.ThreadState & ThreadState.WaitSleepJoin) == 0) + { + Assert.That(waited.Elapsed, Is.LessThan(TimeSpan.FromSeconds(10)), "the write neither reached the store nor blocked"); + Thread.Yield(); + } + + return () => thread.Join(TimeSpan.FromSeconds(10)) && failure == null; + } + [Test] public void UnconditionalCommitIsLastWriterWinsLikeFlush() { @@ -388,6 +538,7 @@ namespace GameCult.Caching.Tests cache.UpsertAsync(new Counter { Name = key.Value }, new CultRecordHandle(key)); cache.FlushAllBackingStores(); }); + Assert.That(directoryCommitted, Is.EqualTo(new[] { "a", "c", "x" }), "a directory commit dropped a key another writer added"); Assert.That(directoryCommitted, Is.EqualTo(directoryFlushed)); } @@ -521,7 +672,7 @@ namespace GameCult.Caching.Tests private string PathOf(string fileName) => System.IO.Path.Combine(_directory, fileName); - private static readonly CultDocumentRegistry Registry = CultDocumentRegistry.ForTypes(new[] { typeof(Counter), typeof(Tally) }); + private static readonly CultDocumentRegistry Registry = CultDocumentRegistry.ForTypes(new[] { typeof(Counter), typeof(Tally), typeof(Slow) }); private static StoreFactory SingleFile(string path) => new(path, () => new SingleFileMessagePackBackingStore(path)); @@ -536,6 +687,13 @@ namespace GameCult.Caching.Tests public ManualResetEventSlim? Pause; public readonly ManualResetEventSlim Entered = new(); + public readonly ManualResetEventSlim Pushed = new(); + + public override void Push(CultStoredDocument entry) + { + base.Push(entry); + Pushed.Set(); + } protected override CultPersistedStoreSnapshot DeserializeSnapshot(byte[] data) { @@ -588,5 +746,33 @@ namespace GameCult.Caching.Tests [CultName] public string Name = string.Empty; } + + // Loading a record with Blocks set waits on Pause, which parks a directory pull inside its page load. + [CultDocument("tests.conditional_slow", "tests.conditional_slow.v1")] + internal sealed class Slow + { + internal static volatile ManualResetEventSlim? Pause; + internal static readonly ManualResetEventSlim Entered = new(); + private bool _blocks; + + [Key(0)] + [CultName] + public string Name = string.Empty; + + [Key(1)] + public bool Blocks + { + get => _blocks; + set + { + _blocks = value; + if (value && Pause is { } pause) + { + Entered.Set(); + pause.Wait(); + } + } + } + } } }