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(); + } + } + } + } } }