GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 2 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();