From 5aad137e61a987b41bef5b30206e98de03ff8e80 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 13 Sep 2026 18:45:27 +0200 Subject: [PATCH] =?UTF-8?q?=EF=BB=BFShare=20the=20cache=20gate=20with=20it?= =?UTF-8?q?s=20stores=20and=20make=20unconditional=20commit=20last-writer-?= =?UTF-8?q?wins?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An attached store takes its cache's gate as its own lock (CacheBackingStore.Gate), so a pull's load callback into the cache can no longer take the gate after the store lock. The directory store's private mutation gate is deleted; the single-file pull now runs under the gate; PullAllBackingStoresAsync takes the gate around each store's pull. Fixes the pull/upsert deadlock on directory stores and the re-pull that erased a staged single-file write. An unconditional single-file Commit writes the store's whole view plus the batch, exactly as a flush would, instead of merging onto the file; it persists earlier staged writes and leaves the store clean. Conditional commits still land onto the file under the lock. Directory-store commit and flush already share one generation writer (staged keys onto the current manifest) and are unchanged. Contracts state both rules, the lock order, that OnUpdate fires for loads only, that a missing single file keeps staged writes, and that a store failing to hydrate on open throws and is left byte-identical. Tests: PullAndUpsertOnDirectoryStoreDoNotDeadlock, RepullNeverErasesAStagedWrite, UnconditionalCommitIsLastWriterWinsLikeFlush, UnconditionalCommitPersistsStagedWrites, OpeningACorruptStoreThrowsAndLeavesItUntouched. Co-Authored-By: Claude Opus 5 --- .../DirectoryMessagePackBackingStore.cs | 11 +- .../Contracts/cultcache-persistence-format.md | 18 +- .../Contracts/cultcache-store-composition.md | 37 +++- src/GameCult.Caching/CultCache.cs | 35 +++- .../ConditionalCommitTests.cs | 160 ++++++++++++++++++ 5 files changed, 236 insertions(+), 25 deletions(-) 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;