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

Route each document type to one home store; attach hydrates; commit explicit batches

Browse source 
Cut 3 of the CultCache migration (Aetheria docs/cultcache-migration-cut.md, 2.1
and 2.2). One commit, because routing, the batch that replaces the ambient
transaction, and conditional commit share one admission path.

Deleted first:
- globals: InitializeGlobals, MaterializeMissingGlobals, the two-argument
  constructor, the initializeGlobals plumbing in CultCacheMessagePack,
  ContainsDurableRecord (cache and directory store)
- replication: the push-to-every-store loops and the attach-time push of
  existing records
- typeof(T) as the schema owner in AddAsync/UpsertAsync and in
  CultNetDatabase.PutAsync/PutPredictedAsync
- the ambient transaction: ExecuteTransactionAsync x2, CultCacheTransaction,
  CommitTransaction, VisibleStoredDocuments, the AsyncLocal, the SemaphoreSlim
  gate, and every ambient check
- CultCacheOpenOptions.PullOnOpen; OpenAsync is Task.FromResult(Create(...))
- DirectoryMessagePackBackingStore.ResolveLegacyUncataloguedRecordCatalog: the
  catalog precommit (fe55bff) is an ancestor of the v4 page format (eea8248),
  so no v4 manifest can reach it
- CultCache.DeleteAsync<T>; its one caller, CultNetDatabaseSubscriptionClient,
  called it by reflection and now calls Remove(key)

Added:
- AddBackingStore(store, params Type[] homes): attach validates (clean, not
  already attached, one untyped store, no claimed home type, no admitted
  record changes home) and then pulls under the gate
- Home(type): most derived routed home, else the untyped store
- Admit: every add and remove, from a write, a batch or a load, validates
  (home, read-only, loaded-from-home, global singleton), lands durably, then
  applies and publishes; a pull hands its whole delta over in one call, so a
  refused record leaves the store and the cache unchanged
- read-only stores (CacheBackingStore(readOnly), both store constructors,
  CultCacheOpenOptions.ReadOnly); flush skips them and writes only dirty stores
- assignable GetGlobal<T>/GetByName<T>/GetByIndex<T>/Watch<T>; more than one
  candidate throws naming them
- Commit(Action<CultCacheBatch>) returning bool, TryCommit returning
  CultCommitOutcome, CultCacheBatch (Upsert, Remove, Expect, ExpectUnchanged,
  sealed after commit), CultCommitRequest; CacheBackingStore.CommitBatch
  evaluates conditions under the store's exclusive lock
- single-file store: <path>.lock sidecar for CommitBatch and PushAll; a clean
  store commits onto the file as it is, so records it never loaded survive
- storedAt minted in one place and bumped a tick when not later than the record
  it replaces

Consumer adjustments in this repo: Mesh streaming tests drop PullOnOpen; the
Studio bridge stops reflecting on it; BackingStoreTests' global test creates
its global explicitly and its transaction tests are rewritten against Commit
(the two that proved ambient hydration rules are deleted); the pre-Cut 2
refusal test expects the attach to throw.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-13 17:57:58 +02:00
commit a3cc394241
10 changed files with 767 additions and 999 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -24,7 +24,8 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
private Dictionary<string, CultPersistedRecord> _durableIndex = new(StringComparer.Ordinal);
private CultSchemaCatalogEntry[] _durableCatalog = Array.Empty<CultSchemaCatalogEntry>();
public DirectoryMessagePackBackingStore(string manifestPath, string? recordDirectoryPath = null)
public DirectoryMessagePackBackingStore(string manifestPath, string? recordDirectoryPath = null, bool readOnly = false)
: base(readOnly)
{
if (string.IsNullOrWhiteSpace(manifestPath))
{
@ -37,6 +38,8 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
public static string DefaultRecordDirectoryPath(string manifestPath) => manifestPath + ".records";
public override string ToString() => _manifestFile.FullName;
public override void PullAll()
{
lock (_mutationGate)
@ -59,7 +62,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
var reports = new List<CultSchemaMigrationReport>();
var loaded = new Dictionary<string, CultStoredDocument>(StringComparer.Ordinal);
using (AcquireCommitLease())
using (AcquireCommitLease(wait: true))
{
// Only pages named by the manifest read under the lease are loaded; orphaned pages are never loaded.
var manifest = ReadManifest();
@ -78,32 +81,26 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
Trace($"indexed-pages loaded={loaded.Count}");
}
foreach (var pair in loaded)
{
if (_dirtyKeys.ContainsKey(pair.Key) || _deletedKeys.ContainsKey(pair.Key))
continue;
if (Entries.TryGetValue(pair.Key, out var existing) &&
string.Equals(existing.StoredAt, pair.Value.StoredAt, StringComparison.Ordinal) &&
string.Equals(existing.Descriptor.SchemaId, pair.Value.Descriptor.SchemaId, StringComparison.Ordinal))
{
continue;
}
Entries[pair.Key] = pair.Value;
Loaded?.Invoke(pair.Value);
}
foreach (var removedKey in _hydratedKeys.Where(key => !loaded.ContainsKey(key)).ToArray())
{
if (_dirtyKeys.ContainsKey(removedKey) || _deletedKeys.ContainsKey(removedKey))
continue;
if (Entries.TryRemove(removedKey, out var removed))
Unloaded?.Invoke(removed);
_hydratedKeys.Remove(removedKey);
}
bool Staged(string key) => _dirtyKeys.ContainsKey(key) || _deletedKeys.ContainsKey(key);
var arrived = loaded
.Where(pair => !Staged(pair.Key) &&
(!Entries.TryGetValue(pair.Key, out var existing) ||
existing.StoredAt != pair.Value.StoredAt ||
existing.Descriptor.SchemaId != pair.Value.Descriptor.SchemaId))
.Select(pair => pair.Value)
.ToArray();
var departed = _hydratedKeys
.Where(key => !loaded.ContainsKey(key) && !Staged(key) && Entries.ContainsKey(key))
.Select(key => Entries[key])
.ToArray();
if (arrived.Length > 0 || departed.Length > 0)
Loaded?.Invoke(arrived, departed);
foreach (var stored in departed)
Entries.TryRemove(stored.Key.Value, out _);
_hydratedKeys.RemoveWhere(key => !loaded.ContainsKey(key) && !Staged(key));
foreach (var stored in arrived)
Entries[stored.Key.Value] = stored;
foreach (var key in loaded.Keys)
_hydratedKeys.Add(key);
@ -112,14 +109,9 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
Trace("publish");
}
public override bool ContainsDurableRecord(CultRecordKey key)
{
lock (_mutationGate)
return _durableIndex.ContainsKey(key.Value) || base.ContainsDurableRecord(key);
}
public override void Push(CultStoredDocument entry)
{
ThrowIfReadOnly();
lock (_mutationGate)
{
Entries[entry.Key.Value] = entry;
@ -131,6 +123,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
public override void Delete(CultStoredDocument entry)
{
ThrowIfReadOnly();
lock (_mutationGate)
{
Entries.TryRemove(entry.Key.Value, out _);
@ -140,32 +133,40 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
}
}
public override void CommitBatch(
IReadOnlyCollection<CultStoredDocument> upserts,
IReadOnlyCollection<CultStoredDocument> deletes)
public override CultCommitOutcome CommitBatch(CultCommitRequest request, bool wait)
{
ThrowIfReadOnly();
lock (_mutationGate)
{
Directory.CreateDirectory(_manifestFile.DirectoryName!);
using var commitLease = AcquireCommitLease(wait);
if (commitLease == null)
return CultCommitOutcome.Contended;
var manifest = ReadManifest();
if (!request.ConditionsHold(manifest.Records, Entries.Values))
return CultCommitOutcome.Mismatch;
var previousEntries = Entries.ToDictionary(pair => pair.Key, pair => pair.Value, StringComparer.Ordinal);
var previousDirtyKeys = _dirtyKeys.Keys.ToArray();
var previousDeletedKeys = _deletedKeys.Keys.ToArray();
var wasDirty = IsDirty;
try
{
foreach (var entry in deletes)
foreach (var entry in request.Deletes)
{
Entries.TryRemove(entry.Key.Value, out _);
_dirtyKeys.TryRemove(entry.Key.Value, out _);
_deletedKeys[entry.Key.Value] = true;
}
foreach (var entry in upserts)
foreach (var entry in request.Upserts)
{
Entries[entry.Key.Value] = entry;
_dirtyKeys[entry.Key.Value] = true;
_deletedKeys.TryRemove(entry.Key.Value, out _);
}
IsDirty = true;
PushAllCore();
WriteGeneration(manifest);
return CultCommitOutcome.Committed;
}
catch
{
@ -186,22 +187,22 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
public override void PushAll()
{
ThrowIfReadOnly();
lock (_mutationGate)
PushAllCore();
{
if (!IsDirty)
return;
Directory.CreateDirectory(_manifestFile.DirectoryName!);
using var commitLease = AcquireCommitLease(wait: true);
WriteGeneration(ReadManifest());
}
}
private void PushAllCore()
// Runs under the commit lease: pages first, then the manifest that names them.
private void WriteGeneration(CultPersistedStoreSnapshot currentManifest)
{
if (!IsDirty)
{
return;
}
Directory.CreateDirectory(_manifestFile.DirectoryName!);
Directory.CreateDirectory(_recordDirectory.FullName);
using var commitLease = AcquireCommitLease();
var currentManifest = ReadManifest();
var currentIndex = currentManifest.Records.ToDictionary(record => record.Key, record => record, StringComparer.Ordinal);
foreach (var key in _deletedKeys.Keys)
currentIndex.Remove(key);
@ -292,11 +293,10 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
var started = tracePages ? Stopwatch.GetTimestamp() : 0L;
var metadata = records[index];
var record = ReadPersistedRecordPage(metadata, out var pagePayload);
var catalog = ResolveLegacyUncataloguedRecordCatalog(record, catalogEntries);
recordReports[index] = Registry.ResolvePersistedSchemaReport(record.SchemaId, catalog);
recordReports[index] = Registry.ResolvePersistedSchemaReport(record.SchemaId, catalogEntries);
storedRecords[index] = ToStoredDocument(
record,
catalog,
catalogEntries,
(type, payload) => CultDocumentMessagePackSerialization.DeserializeUntyped(type, payload, Registry));
if (tracePages)
{
@ -381,80 +381,6 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
return snapshot;
}
// Explicit compatibility path for records written before catalog precommit existed.
private IReadOnlyCollection<CultSchemaCatalogEntry> ResolveLegacyUncataloguedRecordCatalog(
CultPersistedRecord record,
IReadOnlyCollection<CultSchemaCatalogEntry> manifestCatalog)
{
if (manifestCatalog.Any(entry => string.Equals(entry.SchemaId, record.SchemaId, StringComparison.Ordinal)))
{
return manifestCatalog;
}
var schemaVersion = TryReadPayloadSchemaVersion(record.Payload);
if (string.IsNullOrWhiteSpace(schemaVersion))
{
return manifestCatalog;
}
var schemaName = InferSchemaName(schemaVersion);
if (string.IsNullOrWhiteSpace(schemaName))
{
return manifestCatalog;
}
return manifestCatalog
.Append(new CultSchemaCatalogEntry
{
SchemaId = record.SchemaId,
SchemaName = schemaName,
SchemaVersion = schemaVersion,
ContentHash = record.SchemaId,
CanonicalSchemaJson = "",
CompatibleSchemaIds = Array.Empty<string>(),
Members = Array.Empty<CultSchemaMemberCatalogEntry>()
})
.ToArray();
}
private static string TryReadPayloadSchemaVersion(byte[] payload)
{
if (payload == null || payload.Length == 0)
{
return "";
}
try
{
var reader = new MessagePackReader(payload);
if (reader.NextMessagePackType != MessagePackType.Array)
{
return "";
}
if (reader.ReadArrayHeader() <= 0)
{
return "";
}
return reader.NextMessagePackType == MessagePackType.String
? reader.ReadString() ?? ""
: "";
}
catch
{
return "";
}
}
private static string InferSchemaName(string schemaVersion)
{
var versionMarker = schemaVersion.LastIndexOf(".v", StringComparison.Ordinal);
return versionMarker > 0 && versionMarker + 2 < schemaVersion.Length && char.IsDigit(schemaVersion[versionMarker + 2])
? schemaVersion.Substring(0, versionMarker)
: "";
}
private string ContentAddressedRecordPath(CultPersistedRecord metadata)
{
if (metadata.Payload == null || metadata.Payload.Length != 32)
@ -553,7 +479,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
return buffer.ToArray();
}
private FileStream AcquireCommitLease()
private FileStream? AcquireCommitLease(bool wait)
{
Directory.CreateDirectory(_recordDirectory.FullName);
var lockPath = Path.Combine(_recordDirectory.FullName, ".commit.lock");
@ -570,6 +496,10 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
bufferSize: 1,
FileOptions.WriteThrough);
}
catch (IOException) when (!wait)
{
return null;
}
catch (IOException) when (started.Elapsed < TimeSpan.FromSeconds(30))
{
Thread.Sleep(10);