GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Publish observers after the gate, gate every store entry point, stamp under one hold

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-13 19:23:44 +02:00
commit 4eacd876bf
4 changed files with 417 additions and 132 deletions
Download patch file Download diff file Expand all files Collapse all files

14
src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs
View file

@ -20,7 +20,6 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
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);
private Dictionary<string, CultPersistedRecord> _durableIndex = new(StringComparer.Ordinal);
private CultSchemaCatalogEntry[] _durableCatalog = Array.Empty<CultSchemaCatalogEntry>();
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<CultStoredDocument>(), Array.Empty<CultStoredDocument>());
}
private void PullAllCore()
@ -61,17 +62,13 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
var reports = new List<CultSchemaMigrationReport>();
var loaded = new Dictionary<string, CultStoredDocument>(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();

25
src/GameCult.Caching/Contracts/cultcache-store-composition.md
View file

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

302
src/GameCult.Caching/CultCache.cs
View file

@ -1179,6 +1179,7 @@ namespace GameCult.Caching
private readonly ConditionalWeakTable<object, KeyBox> _handles = new();
private readonly Subject<Change> _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<Type>();
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<CultStoredDocument>(), 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<CultStoredDocument>(), new[] { existing }, null, home =>
{
home?.Delete(existing);
return CultCommitOutcome.Committed;
});
return true;
}
}
finally
{
Publish();
}
}
public void Remove<T>(CultRecordHandle<T> 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<CultStoredDocument>(), null, home =>
try
{
home?.Push(stored);
return CultCommitOutcome.Committed;
});
return stored.Key;
lock (_gate)
{
var stored = CreateStoredDocument(document, key);
Admit(new[] { stored }, Array.Empty<CultStoredDocument>(), null, home =>
{
home?.Push(stored);
return CultCommitOutcome.Committed;
});
return stored.Key;
}
}
finally
{
Publish();
}
}
private CultCommitOutcome Land(Action<CultCacheBatch> 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<CacheBackingStore?, CultCommitOutcome> land)
{
List<Change> 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<CultStoredDocument>(), Array.Empty<CultStoredDocument>());
}
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;

208
tests/GameCult.Caching.Tests/ConditionalCommitTests.cs
View file

@ -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<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);
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<Counter>(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<Counter>(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<string>();
var seen = 0;
using var subscription = cache.Watch<Counter>().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<Counter>(new CultRecordKey("upserted")));
Assert.That(cache.Commit(batch => batch.Upsert(new Counter { Name = "committed" }, new CultRecordHandle<Counter>(new CultRecordKey("committed")))), Is.True);
using (var other = SingleFile(path).Open())
Assert.That(other.Commit(batch => batch.Upsert(new Counter { Name = "pulled" }, new CultRecordHandle<Counter>(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<Counter>(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<Counter>(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<bool> StartWriter(Action write, Func<bool> 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<Counter>(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();
}
}
}
}
}
}