Share the cache gate with its stores and make unconditional commit last-writer-wins

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 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-13 18:45:27 +02:00
commit 5aad137e61
5 changed files with 236 additions and 25 deletions

View file

@ -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<string, bool> _dirtyKeys = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<string, bool> _deletedKeys = new(StringComparer.Ordinal);
private readonly HashSet<string> _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;

View file

@ -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.

View file

@ -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 (`<path>.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 `<path>.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

View file

@ -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<CultStoredDocument>, IReadOnlyList<CultStoredDocument>>? 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)

View file

@ -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<Counter>(new CultRecordKey($"a:{i}")));
});
var writing = Task.Run(() =>
{
for (var i = 0; i < 150; i++)
{
b.UpsertAsync(new Counter { Name = $"b:{i}" }, new CultRecordHandle<Counter>(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<Counter>(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<Counter>(key))), Is.True));
var flushed = LastWriter(SingleFile(PathOf("flush.cc")), (cache, key) =>
{
cache.UpsertAsync(new Counter { Name = key.Value }, new CultRecordHandle<Counter>(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<Counter>(key))), Is.True));
var directoryFlushed = LastWriter(DirectoryStore(PathOf("flush-dir.cc")), (cache, key) =>
{
cache.UpsertAsync(new Counter { Name = key.Value }, new CultRecordHandle<Counter>(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<Counter>(new CultRecordKey("s")));
Assert.That(cache.Commit(batch => batch.Upsert(new Counter { Name = "y" }, new CultRecordHandle<Counter>(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<CultCache, CultRecordKey> write)
{
using (var seed = store.Open())
Assert.That(seed.Commit(batch => batch.Upsert(new Counter { Name = "a" }, new CultRecordHandle<Counter>(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<Counter>(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<CacheBackingStore> _create;