diff --git a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs index f2a7faee..7c32500f 100644 --- a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs +++ b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs @@ -17,7 +17,6 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore private const string IndexedFormatVersion = "cultcache.store.v4.directory-content-addressed-pages"; private readonly FileInfo _manifestFile; private readonly DirectoryInfo _recordDirectory; - private readonly object _mutationGate = new(); private readonly ConcurrentDictionary _dirtyKeys = new(StringComparer.Ordinal); private readonly ConcurrentDictionary _deletedKeys = new(StringComparer.Ordinal); private readonly HashSet _hydratedKeys = new(StringComparer.Ordinal); @@ -42,7 +41,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore public override void PullAll() { - lock (_mutationGate) + lock (Gate) PullAllCore(); } @@ -112,7 +111,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore public override void Push(CultStoredDocument entry) { ThrowIfReadOnly(); - lock (_mutationGate) + lock (Gate) { Entries[entry.Key.Value] = entry; _dirtyKeys[entry.Key.Value] = true; @@ -124,7 +123,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore public override void Delete(CultStoredDocument entry) { ThrowIfReadOnly(); - lock (_mutationGate) + lock (Gate) { Entries.TryRemove(entry.Key.Value, out _); _dirtyKeys.TryRemove(entry.Key.Value, out _); @@ -136,7 +135,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore public override CultCommitOutcome CommitBatch(CultCommitRequest request, bool wait) { ThrowIfReadOnly(); - lock (_mutationGate) + lock (Gate) { Directory.CreateDirectory(_manifestFile.DirectoryName!); using var commitLease = AcquireCommitLease(wait); @@ -188,7 +187,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore public override void PushAll() { ThrowIfReadOnly(); - lock (_mutationGate) + lock (Gate) { if (!IsDirty) return; diff --git a/src/GameCult.Caching/Contracts/cultcache-persistence-format.md b/src/GameCult.Caching/Contracts/cultcache-persistence-format.md index d65a4919..e8231c10 100644 --- a/src/GameCult.Caching/Contracts/cultcache-persistence-format.md +++ b/src/GameCult.Caching/Contracts/cultcache-persistence-format.md @@ -52,9 +52,17 @@ under the store's exclusive lock against what is durably on disk. A failed condition is a lost race, not an error: the commit reports it and writes nothing. -**Conditional commit is the only safe multi-process write.** A plain flush -writes the whole snapshot and is last-writer-wins, so processes sharing a store -must all use conditional commit. +**Conditional commit is the only safe multi-process write.** A plain flush and +an unconditional commit are last-writer-wins and write the same bytes: a single +file is replaced by the writer's whole view, a directory manifest takes the +writer's staged keys over the current manifest. An unconditional commit also +persists writes staged earlier and leaves the store clean. Processes sharing a +store must all use conditional commit. + +A store that fails to hydrate on open (corrupt bytes, unresolvable schema) +makes the open throw and is left byte-identical; it is never silently +overwritten. Within one process the lock order is the cache's gate, then the +store's file lock; a store's I/O blocks its cache's readers. C# implements batch and conditional commit (`CultCache.Commit`, `TryCommit`, `CultCacheBatch`). Rust implements both. TypeScript and Python implement @@ -358,8 +366,8 @@ The v1 concurrent single-file policy is: 2. Writers take an exclusive sidecar lock derived from the `.cc` path. 3. Writers re-read the current snapshot after taking the lock. 4. Writers using conditional commit evaluate their conditions against the - latest snapshot and write nothing if one fails. A plain flush compares - nothing and is last-writer-wins. + latest snapshot and write nothing if one fails. A plain flush and an + unconditional commit compare nothing and are last-writer-wins. 5. Writers write a temp file, flush it, atomically replace the `.cc` file, and release the lock. diff --git a/src/GameCult.Caching/Contracts/cultcache-store-composition.md b/src/GameCult.Caching/Contracts/cultcache-store-composition.md index ccb3c0ec..0dc29d44 100644 --- a/src/GameCult.Caching/Contracts/cultcache-store-composition.md +++ b/src/GameCult.Caching/Contracts/cultcache-store-composition.md @@ -40,7 +40,24 @@ which route wrote it. - A record loaded from a store that is not its home is refused at load, naming the key and both stores. - Pulling again after attach is a re-pull. It does not erase mutations staged in - a dirty store. + a dirty store. A dirty single-file store skips the re-pull entirely; a missing + 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. +- 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 + open. + +## Locking + +- One lock order: the cache's gate, then the store's lock. A store attached to a + cache takes that cache's gate as its own lock, so a store's load callback into + the cache cannot take the two out of order. +- A store doing I/O on behalf of its cache (pull, flush, commit) holds the gate, + 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. ## Dirtiness: loading never writes @@ -123,10 +140,20 @@ replaces; writers bump a minted timestamp by one tick when it is not later. - Single-file stores lock a sidecar `.lock` opened exclusively. Directory stores use their existing commit lease. -**A plain flush is last-writer-wins.** It writes the whole snapshot under the -same lock, so two writers never interleave bytes, but it compares nothing and -overwrites whatever another process committed. Conditional commit is the only -safe multi-process write: processes sharing a store must all use it. +**A plain flush and an unconditional commit are last-writer-wins.** 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. + +- A single-file store writes this cache's whole view of the file: a record + another writer added since this cache last pulled is gone. +- A directory store writes this cache's staged keys onto the current manifest: + another writer's unrelated keys survive, a key both wrote holds the last + write. +- An unconditional commit writes exactly what a flush of the same staged state + plus the batch would write. It also persists any single writes staged earlier + in that store, and leaves the store clean. +- A conditional commit lands the batch onto the file as it is under the lock. A store is written by one runtime at a time. Whether a Rust `fs2` lock and a C# exclusive open of the same lock file exclude each other is not established, and diff --git a/src/GameCult.Caching/CultCache.cs b/src/GameCult.Caching/CultCache.cs index 6ee87c81..f6a5c484 100644 --- a/src/GameCult.Caching/CultCache.cs +++ b/src/GameCult.Caching/CultCache.cs @@ -1282,6 +1282,7 @@ namespace GameCult.Caching } store.AttachRegistry(_registry); + store.Gate = _gate; store.Loaded = (loaded, dropped) => Admit(loaded, dropped, store, _ => CultCommitOutcome.Committed); _stores.Add(candidate); try @@ -1292,6 +1293,7 @@ namespace GameCult.Caching { _stores.Remove(candidate); store.Loaded = null; + store.Gate = new object(); throw; } } @@ -1300,7 +1302,11 @@ namespace GameCult.Caching public Task PullAllBackingStoresAsync() { foreach (var store in BackingStores) - store.PullAll(); + { + lock (_gate) + store.PullAll(); + } + return Task.CompletedTask; } @@ -1800,6 +1806,10 @@ namespace GameCult.Caching // any of it; if the cache refuses a record the call throws and the store keeps its previous view. protected internal Action, IReadOnlyList>? Loaded; + // Once attached, the cache's gate: the store and its cache share one lock, so a load calling back into the cache + // can never take the gate after the store lock. File locks are always taken inside it. + protected internal object Gate { get; internal set; } = new(); + internal void AttachRegistry(CultDocumentRegistry registry) { _registry = registry ?? throw new ArgumentNullException(nameof(registry)); @@ -1881,6 +1891,12 @@ namespace GameCult.Caching public override string ToString() => FileInfo.FullName; public override void PullAll() + { + lock (Gate) + PullAllCore(); + } + + private void PullAllCore() { // A single-file snapshot cannot distinguish local dirty keys from clean keys. Pulling while local mutations // are staged would let an older disk snapshot erase them before the next flush. Flush first. @@ -1959,14 +1975,15 @@ namespace GameCult.Caching if (!request.ConditionsHold(disk.Records, Entries.Values)) return CultCommitOutcome.Mismatch; - // A clean store commits onto the file as it is now, so records this store never loaded survive. A dirty - // store's staged view is the snapshot its flush would write, so the batch lands on that. - var records = IsDirty - ? Entries.Values.Select(entry => ToPersistedRecord(entry, SerializePayload)).ToDictionary(record => record.Key, StringComparer.Ordinal) - : disk.Records.ToDictionary(record => record.Key, StringComparer.Ordinal); - var catalog = IsDirty - ? Entries.Values.Select(entry => entry.Descriptor.ToCatalogEntry()) - : disk.SchemaCatalog; + // An unconditional commit writes what a flush would, this store's whole view plus the batch: last-writer-wins, + // and staged single writes land with it. A conditional commit (store clean) lands the batch onto the file as it is. + var ontoDisk = request.HasConditions && !IsDirty; + var records = ontoDisk + ? disk.Records.ToDictionary(record => record.Key, StringComparer.Ordinal) + : Entries.Values.Select(entry => ToPersistedRecord(entry, SerializePayload)).ToDictionary(record => record.Key, StringComparer.Ordinal); + var catalog = ontoDisk + ? disk.SchemaCatalog + : Entries.Values.Select(entry => entry.Descriptor.ToCatalogEntry()); foreach (var entry in request.Deletes) records.Remove(entry.Key.Value); foreach (var entry in request.Upserts) diff --git a/tests/GameCult.Caching.Tests/ConditionalCommitTests.cs b/tests/GameCult.Caching.Tests/ConditionalCommitTests.cs index c7c42b66..4a3ce596 100644 --- a/tests/GameCult.Caching.Tests/ConditionalCommitTests.cs +++ b/tests/GameCult.Caching.Tests/ConditionalCommitTests.cs @@ -303,6 +303,143 @@ namespace GameCult.Caching.Tests Assert.That(System.IO.Directory.GetFiles(_directory), Is.Empty, "a store file or lock file was touched"); } + // Cache A pulls and upserts on two threads while cache B commits to the same directory store; the pull's load + // callback and the upsert's push must take the cache gate and the store lock in one order. A is not disposed on + // failure because a deadlocked gate would hang the test thread too. + [Test] + public void PullAndUpsertOnDirectoryStoreDoNotDeadlock() + { + var store = DirectoryStore(PathOf("deadlock.cc")); + var a = store.Open(); + using var b = store.Open(); + var pulling = Task.Run(() => + { + for (var i = 0; i < 300; i++) + a.PullAllBackingStoresAsync(); + }); + var upserting = Task.Run(() => + { + for (var i = 0; i < 3000; i++) + a.UpsertAsync(new Counter { Name = $"a:{i}" }, new CultRecordHandle(new CultRecordKey($"a:{i}"))); + }); + var writing = Task.Run(() => + { + for (var i = 0; i < 150; i++) + { + b.UpsertAsync(new Counter { Name = $"b:{i}" }, new CultRecordHandle(new CultRecordKey($"b:{i}"))); + b.FlushAllBackingStores(); + } + }); + + Assert.That(Task.WaitAll(new[] { pulling, upserting, writing }, TimeSpan.FromSeconds(60)), Is.True, "pull and upsert deadlocked"); + a.Dispose(); + } + + [Test] + public void RepullNeverErasesAStagedWrite() + { + var path = PathOf("repull.cc"); + Create(SingleFile(path), 0); + var store = new PausingStore(path); + 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); + } + + Assert.That(cache.Get(staged), Is.Not.Null, "the re-pull erased a staged write from the cache"); + cache.FlushAllBackingStores(); + using var reader = SingleFile(path).Open(); + Assert.That(reader.Get(staged), Is.Not.Null, "the flush after the re-pull lost the staged write"); + } + + [Test] + public void UnconditionalCommitIsLastWriterWinsLikeFlush() + { + var committed = LastWriter(SingleFile(PathOf("commit.cc")), (cache, key) => + Assert.That(cache.Commit(batch => batch.Upsert(new Counter { Name = key.Value }, new CultRecordHandle(key))), Is.True)); + var flushed = LastWriter(SingleFile(PathOf("flush.cc")), (cache, key) => + { + cache.UpsertAsync(new Counter { Name = key.Value }, new CultRecordHandle(key)); + cache.FlushAllBackingStores(); + }); + + Assert.That(committed, Is.EqualTo(new[] { "a", "x" }), "an unconditional commit kept a record its cache never saw"); + Assert.That(committed, Is.EqualTo(flushed)); + + // A directory store's flush lands staged keys onto the current manifest, and its commit does exactly the same. + var directoryCommitted = LastWriter(DirectoryStore(PathOf("commit-dir.cc")), (cache, key) => + Assert.That(cache.Commit(batch => batch.Upsert(new Counter { Name = key.Value }, new CultRecordHandle(key))), Is.True)); + var directoryFlushed = LastWriter(DirectoryStore(PathOf("flush-dir.cc")), (cache, key) => + { + cache.UpsertAsync(new Counter { Name = key.Value }, new CultRecordHandle(key)); + cache.FlushAllBackingStores(); + }); + Assert.That(directoryCommitted, Is.EqualTo(directoryFlushed)); + } + + [Test] + public void UnconditionalCommitPersistsStagedWrites() + { + var path = PathOf("staged.cc"); + using var cache = SingleFile(path).Open(); + cache.UpsertAsync(new Counter { Name = "s" }, new CultRecordHandle(new CultRecordKey("s"))); + Assert.That(cache.Commit(batch => batch.Upsert(new Counter { Name = "y" }, new CultRecordHandle(new CultRecordKey("y")))), Is.True); + + Assert.That(cache.IsDirty, Is.False); + using var reader = SingleFile(path).Open(); + Assert.That(reader.AllStoredDocuments.Select(stored => stored.Key.Value), Is.EqualTo(new[] { "s", "y" })); + } + + [Test] + public void OpeningACorruptStoreThrowsAndLeavesItUntouched() + { + foreach (var store in new[] { SingleFile(PathOf("corrupt.cc")), DirectoryStore(PathOf("corrupt-dir.cc")) }) + { + File.WriteAllBytes(store.Path, new byte[] { 0xC1, 0x00, 0xFF, 0x13 }); + Assert.That(() => store.Open(), Throws.Exception, $"{store.Path}: a corrupt store opened"); + Assert.That(File.ReadAllBytes(store.Path), Is.EqualTo(new byte[] { 0xC1, 0x00, 0xFF, 0x13 }), $"{store.Path}: opening rewrote a corrupt store"); + } + + var unknown = PathOf("unknown-schema.cc"); + using (var writer = SingleFile(unknown).Open()) + { + writer.UpsertAsync(new Tally { Name = "t" }); + writer.FlushAllBackingStores(); + } + + var bytes = File.ReadAllBytes(unknown); + using var narrow = new CultCache(CultDocumentRegistry.ForTypes(new[] { typeof(Counter) })); + Assert.That(() => narrow.AddBackingStore(new SingleFileMessagePackBackingStore(unknown)), Throws.Exception, "a store with an unknown schema opened"); + Assert.That(File.ReadAllBytes(unknown), Is.EqualTo(bytes), "opening rewrote a store with an unknown schema"); + } + + // A pulls {a}; B commits c; A writes x. Returns the keys on disk afterwards. + private static string[] LastWriter(StoreFactory store, Action write) + { + using (var seed = store.Open()) + Assert.That(seed.Commit(batch => batch.Upsert(new Counter { Name = "a" }, new CultRecordHandle(new CultRecordKey("a")))), Is.True); + using var a = store.Open(); + using (var b = store.Open()) + Assert.That(b.Commit(batch => batch.Upsert(new Counter { Name = "c" }, new CultRecordHandle(new CultRecordKey("c")))), Is.True); + write(a, new CultRecordKey("x")); + using var reader = store.Open(); + return reader.AllStoredDocuments.Select(stored => stored.Key.Value).ToArray(); + } + private static void StaleExpect(StoreFactory store) { Create(store, 0); @@ -390,6 +527,29 @@ namespace GameCult.Caching.Tests private static StoreFactory DirectoryStore(string path) => new(path, () => new DirectoryMessagePackBackingStore(path)); + // Pauses a pull between reading the file and handing the records to the cache. + private sealed class PausingStore : SingleFileMessagePackBackingStore + { + public PausingStore(string path) : base(path) + { + } + + public ManualResetEventSlim? Pause; + public readonly ManualResetEventSlim Entered = new(); + + protected override CultPersistedStoreSnapshot DeserializeSnapshot(byte[] data) + { + var snapshot = base.DeserializeSnapshot(data); + if (Pause is { } pause) + { + Entered.Set(); + pause.Wait(); + } + + return snapshot; + } + } + private sealed class StoreFactory { private readonly Func _create;