From a3cc39424166c05b73b0da8235ad9a16cd615161 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 13 Sep 2026 17:57:58 +0200 Subject: [PATCH] Route each document type to one home store; attach hydrates; commit explicit batches 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; 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/GetByName/GetByIndex/Watch; more than one candidate throws naming them - Commit(Action) 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: .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 --- .../CultCacheMessagePack.cs | 55 +- .../CultDocumentMessagePackSerialization.cs | 2 +- .../DirectoryMessagePackBackingStore.cs | 186 +-- src/GameCult.Caching/CultCache.cs | 1347 ++++++++--------- src/GameCult.Networking/CultNetDatabase.cs | 4 +- .../CultNetDatabaseSubscriptionClient.cs | 16 +- .../Caching/Editor/CultCacheStudioWindow.cs | 11 +- .../BackingStoreTests.cs | 223 +-- .../PreCut2StoreFormatTests.cs | 5 +- .../CultMeshStreamingTests.cs | 27 +- 10 files changed, 822 insertions(+), 1054 deletions(-) diff --git a/src/GameCult.Caching.MessagePack/CultCacheMessagePack.cs b/src/GameCult.Caching.MessagePack/CultCacheMessagePack.cs index e464e5e6..04bdc220 100644 --- a/src/GameCult.Caching.MessagePack/CultCacheMessagePack.cs +++ b/src/GameCult.Caching.MessagePack/CultCacheMessagePack.cs @@ -8,7 +8,7 @@ namespace GameCult.Caching.MessagePack { public CultDocumentRegistry? Registry { get; set; } - public bool PullOnOpen { get; set; } = true; + public bool ReadOnly { get; set; } public bool FlushOnDispose { get; set; } @@ -19,57 +19,32 @@ namespace GameCult.Caching.MessagePack public static class CultCacheMessagePack { + // Opening hydrates. A consumer that wants a fresh store removes the records it does not keep, upserts, and + // flushes; the atomic replace does the rest. public static CultCache Create(string filePath, CultCacheOpenOptions? options = null) - { - options ??= new CultCacheOpenOptions(); - return CreateCore(filePath, options, initializeGlobals: true); - } - - public static async Task OpenAsync(string filePath, CultCacheOpenOptions? options = null) - { - options ??= new CultCacheOpenOptions(); - var cache = CreateCore(filePath, options, initializeGlobals: false); - if (options.PullOnOpen) - { - await cache.PullAllBackingStoresAsync().ConfigureAwait(false); - } - - cache.MaterializeMissingGlobals(); - - return cache; - } - - private static CultCache CreateCore( - string filePath, - CultCacheOpenOptions options, - bool initializeGlobals) { if (string.IsNullOrWhiteSpace(filePath)) { throw new ArgumentException("File path must be non-empty.", nameof(filePath)); } - var cache = new CultCache(options.Registry, initializeGlobals) + options ??= new CultCacheOpenOptions(); + var cache = new CultCache(options.Registry) { FlushAttachedStoresOnDispose = options.FlushOnDispose }; - if (options.UseDirectoryStore || Directory.Exists(DirectoryMessagePackBackingStore.DefaultRecordDirectoryPath(filePath))) - { - cache.AddBackingStore(new DirectoryMessagePackBackingStore(filePath) - { - FlushOnDispose = options.StoreFlushOnDispose - }); - } - else - { - cache.AddBackingStore(new SingleFileMessagePackBackingStore(filePath) - { - FlushOnDispose = options.StoreFlushOnDispose - }); - } - + CacheBackingStore store = options.UseDirectoryStore || Directory.Exists(DirectoryMessagePackBackingStore.DefaultRecordDirectoryPath(filePath)) + ? new DirectoryMessagePackBackingStore(filePath, readOnly: options.ReadOnly) + : new SingleFileMessagePackBackingStore(filePath, options.ReadOnly); + store.FlushOnDispose = options.StoreFlushOnDispose; + cache.AddBackingStore(store); return cache; } + + public static Task OpenAsync(string filePath, CultCacheOpenOptions? options = null) + { + return Task.FromResult(Create(filePath, options)); + } } } diff --git a/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs b/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs index b5a621fb..223d0310 100644 --- a/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs +++ b/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs @@ -353,7 +353,7 @@ public static class CultDocumentMessagePackSerialization public class SingleFileMessagePackBackingStore : SingleFileBackingStore { - public SingleFileMessagePackBackingStore(string filePath) : base(filePath) + public SingleFileMessagePackBackingStore(string filePath, bool readOnly = false) : base(filePath, readOnly) { } diff --git a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs index 13264e6f..f2a7faee 100644 --- a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs +++ b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs @@ -24,7 +24,8 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore private Dictionary _durableIndex = new(StringComparer.Ordinal); private CultSchemaCatalogEntry[] _durableCatalog = Array.Empty(); - 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(); var loaded = new Dictionary(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 upserts, - IReadOnlyCollection 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 ResolveLegacyUncataloguedRecordCatalog( - CultPersistedRecord record, - IReadOnlyCollection 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(), - Members = Array.Empty() - }) - .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); diff --git a/src/GameCult.Caching/CultCache.cs b/src/GameCult.Caching/CultCache.cs index 609bda17..ff2314ce 100644 --- a/src/GameCult.Caching/CultCache.cs +++ b/src/GameCult.Caching/CultCache.cs @@ -1,6 +1,8 @@ using System; using System.Collections.Concurrent; using System.Collections.Generic; +using System.Diagnostics; +using System.Globalization; using System.IO; using System.Linq; using System.Reflection; @@ -1053,56 +1055,167 @@ namespace GameCult.Caching } } + public enum CultCommitOutcome + { + Committed, + Mismatch, + Contended + } + + public sealed class CultCommitRequest + { + internal CultCommitRequest( + IReadOnlyList upserts, + IReadOnlyList deletes, + IReadOnlyList<(CultRecordKey Key, string? SchemaId, string? StoredAt)> expected, + bool expectUnchanged) + { + Upserts = upserts; + Deletes = deletes; + Expected = expected; + ExpectUnchanged = expectUnchanged; + } + + public IReadOnlyList Upserts { get; } + public IReadOnlyList Deletes { get; } + public IReadOnlyList<(CultRecordKey Key, string? SchemaId, string? StoredAt)> Expected { get; } + public bool ExpectUnchanged { get; } + public bool HasConditions => Expected.Count > 0 || ExpectUnchanged; + + // durable is what the store holds on disk now; observed is what its cache last loaded or committed. + // Identity is (schemaId, storedAt), sound because every write to a key mints a later storedAt. + public bool ConditionsHold(IReadOnlyCollection durable, IEnumerable observed) + { + var byKey = durable.ToDictionary(record => record.Key, StringComparer.Ordinal); + foreach (var (key, schemaId, storedAt) in Expected) + { + byKey.TryGetValue(key.Value, out var record); + var holds = schemaId == null + ? record == null + : record != null && record.SchemaId == schemaId && record.StoredAt == storedAt; + if (!holds) + return false; + } + + return !ExpectUnchanged || + durable.Select(record => Identity(record.Key, record.SchemaId, record.StoredAt)) + .OrderBy(identity => identity, StringComparer.Ordinal) + .SequenceEqual(observed + .Select(stored => Identity(stored.Key.Value, stored.Descriptor.SchemaId, stored.StoredAt)) + .OrderBy(identity => identity, StringComparer.Ordinal)); + } + + private static string Identity(string key, string schemaId, string storedAt) => $"{key}\n{schemaId}\n{storedAt}"; + } + + // An explicit value: nothing staged here is visible to anyone until Commit returns. + public sealed class CultCacheBatch + { + private readonly CultCache _cache; + internal readonly Dictionary Operations = new(StringComparer.Ordinal); + internal readonly List<(CultRecordKey Key, string? SchemaId, string? StoredAt)> Expected = new(); + internal bool ExpectsUnchanged; + internal bool Sealed; + + internal CultCacheBatch(CultCache cache) + { + _cache = cache; + } + + public CultRecordHandle Upsert(T document, CultRecordHandle? handle = null) + { + if (document == null) throw new ArgumentNullException(nameof(document)); + return new CultRecordHandle(Stage(document, handle?.Key)); + } + + public CultRecordKey Upsert(Type type, object document, CultRecordKey? key = null) + { + CultCache.RequireInstanceOf(type, document); + return Stage(document, key); + } + + public void Remove(CultRecordKey key) + { + ThrowIfSealed(); + Operations[key.Value] = (key, null); + } + + public void Expect(CultRecordKey key, object? current) + { + ThrowIfSealed(); + var observed = _cache.Observe(key, current); + Expected.Add((key, observed?.Descriptor.SchemaId, observed?.StoredAt)); + } + + public void ExpectUnchanged() + { + ThrowIfSealed(); + ExpectsUnchanged = true; + } + + private CultRecordKey Stage(object document, CultRecordKey? key) + { + ThrowIfSealed(); + var stored = _cache.CreateStoredDocument(document, key); + Operations[stored.Key.Value] = (stored.Key, stored); + return stored.Key; + } + + private void ThrowIfSealed() + { + if (Sealed) + throw new InvalidOperationException("This batch has already been committed or abandoned."); + } + } + public sealed class CultCache : IDisposable { private readonly CultDocumentRegistry _registry; - private readonly List _backingStores = new(); - private readonly ConcurrentDictionary _entries = new(StringComparer.Ordinal); - private readonly ConcurrentDictionary> _typeMaps = new(); - private readonly ConcurrentDictionary> _nameMaps = new(); - private readonly ConcurrentDictionary<(Type Type, string Alias), ConcurrentDictionary> _indexMaps = new(); - private readonly ConcurrentDictionary _globalKeys = new(); - private readonly ConditionalWeakTable _documentHandles = new(); - private readonly Subject _changes = new(); - private readonly SemaphoreSlim _transactionGate = new(1, 1); - private readonly AsyncLocal _ambientTransaction = new(); - private readonly object _stateGate = new(); - private bool _hasUnflushedMutations; + private readonly List<(CacheBackingStore Store, Type[] Homes)> _stores = new(); + private readonly Dictionary _entries = new(StringComparer.Ordinal); + private readonly Dictionary> _names = new(); + private readonly Dictionary<(Type Type, string Alias), Dictionary> _indexes = new(); + private readonly Dictionary _globals = new(); + private readonly ConditionalWeakTable _handles = new(); + private readonly Subject _changes = new(); + private readonly object _gate = new(); + private bool _dirtyInMemory; public CultCache(CultDocumentRegistry? registry = null) - : this(registry, initializeGlobals: true) - { - } - - public CultCache(CultDocumentRegistry? registry, bool initializeGlobals) { _registry = registry ?? CultDocumentRegistry.Shared; - if (initializeGlobals) + } + + public CultDocumentRegistry Registry => _registry; + + public bool IsDirty + { + get { - MaterializeMissingGlobals(); + lock (_gate) + return _stores.Count == 0 ? _dirtyInMemory : _stores.Any(entry => entry.Store.IsDirty); } } - // Durable open paths call this after hydration so persisted globals remain authoritative. - public void MaterializeMissingGlobals() - { - InitializeGlobals(); - } - - public bool IsDirty => _hasUnflushedMutations || _backingStores.Any(store => store.IsDirty); - public bool FlushAttachedStoresOnDispose { get; set; } public event Action? OnUpdate; - public IReadOnlyList BackingStores => _backingStores; + public IReadOnlyList BackingStores + { + get + { + lock (_gate) + return _stores.Select(entry => entry.Store).ToArray(); + } + } public IEnumerable AllEntries { get { - lock (_stateGate) - return VisibleStoredDocuments().Select(entry => entry.Document).ToArray(); + lock (_gate) + return _entries.Values.Select(entry => entry.Document).ToArray(); } } @@ -1110,21 +1223,23 @@ namespace GameCult.Caching { get { - lock (_stateGate) - return VisibleStoredDocuments() + lock (_gate) + return _entries.Values .OrderBy(entry => entry.Descriptor.SchemaName, StringComparer.Ordinal) .ThenBy(entry => entry.Key.Value, StringComparer.Ordinal) .ToArray(); } } - public CultDocumentRegistry Registry => _registry; - public Observable> Watch() where T : class { return _changes - .Where(change => change is CultCacheDocumentChange) - .Select(change => (CultCacheDocumentChange)change); + .Where(change => typeof(T).IsAssignableFrom(change.Stored.Descriptor.DocumentType)) + .Select(change => new CultCacheDocumentChange( + change.Kind, + change.Stored.Key, + change.Document as T, + change.Previous as T)); } public Observable> WatchRecord(CultRecordKey key) where T : class @@ -1132,59 +1247,73 @@ namespace GameCult.Caching return Watch().Where(change => change.Key.Equals(key)); } - public void AddBackingStore(CacheBackingStore store) + // Attaching reads the store; there is no interval in which it is attached but unread. + public void AddBackingStore(CacheBackingStore store, params Type[] homes) { if (store == null) throw new ArgumentNullException(nameof(store)); - store.AttachRegistry(_registry); - store.Loaded = entry => AddStoredDocumentInternal(entry, store, raiseUpdate: true); - store.Unloaded = entry => RemoveStoredDocumentInternal(entry, store, raiseUpdate: true); - - foreach (var entry in _entries.Values.OrderBy(entry => entry.Key.Value, StringComparer.Ordinal)) + homes ??= Array.Empty(); + lock (_gate) { - store.Push(entry); - } + 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) + { + 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}."); + } - _backingStores.Add(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) + 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.Loaded = (loaded, dropped) => Admit(loaded, dropped, store, _ => CultCommitOutcome.Committed); + _stores.Add(candidate); + try + { + store.PullAll(); + } + catch + { + _stores.Remove(candidate); + store.Loaded = null; + throw; + } + } } - public async Task PullAllBackingStoresAsync() + public Task PullAllBackingStoresAsync() { - if (_ambientTransaction.Value != null) - throw new InvalidOperationException( - "CultCache hydration cannot run inside a mutation transaction; hydrate before opening the commit scope."); - - await _transactionGate.WaitAsync().ConfigureAwait(false); - try - { - foreach (var store in _backingStores) - store.PullAll(); - - _hasUnflushedMutations = _backingStores.Any(store => store.IsDirty); - } - finally - { - _transactionGate.Release(); - } + foreach (var store in BackingStores) + store.PullAll(); + return Task.CompletedTask; } public void FlushAllBackingStores() { - if (_ambientTransaction.Value != null) - throw new InvalidOperationException("A CultCache transaction owns durable commit; do not flush inside its stage callback."); - - _transactionGate.Wait(); - try + lock (_gate) { - foreach (var store in _backingStores) + foreach (var (store, _) in _stores) { - store.PushAll(); - RecomputeDirtyState(); + if (!store.IsReadOnly && store.IsDirty) + store.PushAll(); } } - finally - { - _transactionGate.Release(); - } } // soft is dead. CultCacheStudioWindow reflects on FlushAsync(bool) until Cut 6 rewrites it. @@ -1194,127 +1323,66 @@ namespace GameCult.Caching return Task.CompletedTask; } - // Staged records are visible only to the executing async flow. The durable backing store - // commits before the live cache and its observers advance; an exception discards the whole batch. - public async Task ExecuteTransactionAsync(Func stageAsync) + public Task> AddAsync(T document, CultRecordHandle? handle = null) { - if (stageAsync == null) throw new ArgumentNullException(nameof(stageAsync)); - if (_ambientTransaction.Value != null) - { - await stageAsync().ConfigureAwait(false); - return; - } - - await _transactionGate.WaitAsync().ConfigureAwait(false); - var transaction = new CultCacheTransaction(); - IReadOnlyList<(CultStoredDocument Stored, object? Previous, bool Removed)> changes; - _ambientTransaction.Value = transaction; - try - { - await stageAsync().ConfigureAwait(false); - transaction.Seal(); - changes = CommitTransaction(transaction); - } - finally - { - transaction.Seal(); - _ambientTransaction.Value = null; - _transactionGate.Release(); - } - - foreach (var change in changes) - PublishChange(change.Stored, change.Previous, change.Removed); - } - - public async Task ExecuteTransactionAsync(Func> stageAsync) - { - if (stageAsync == null) throw new ArgumentNullException(nameof(stageAsync)); - T result = default!; - await ExecuteTransactionAsync(async () => - { - result = await stageAsync().ConfigureAwait(false); - }).ConfigureAwait(false); - return result; - } - - public async Task> AddAsync(T document, CultRecordHandle? handle = null) - { - if (document == null) throw new ArgumentNullException(nameof(document)); - if (_ambientTransaction.Value is { } transaction) - { - var staged = CreateStoredDocument(typeof(T), document, handle?.Key); - transaction.Stage(staged); - return new CultRecordHandle(staged.Key); - } - - await _transactionGate.WaitAsync().ConfigureAwait(false); - try - { - var stored = CreateStoredDocument(typeof(T), document, handle?.Key); - AddStoredDocumentInternal(stored, source: null, raiseUpdate: false); - return new CultRecordHandle(stored.Key); - } - finally - { - _transactionGate.Release(); - } + return UpsertAsync(document, handle); } public Task> UpsertAsync(T document, CultRecordHandle? handle = null) { - return AddAsync(document, handle); + if (document == null) throw new ArgumentNullException(nameof(document)); + return Task.FromResult(new CultRecordHandle(Write(document, handle?.Key))); } - public async Task UpsertAsync(Type documentType, object document, CultRecordKey? key = null) + public Task UpsertAsync(Type documentType, object document, CultRecordKey? key = null) { - if (documentType == null) throw new ArgumentNullException(nameof(documentType)); - if (document == null) throw new ArgumentNullException(nameof(document)); - if (!documentType.IsInstanceOfType(document)) - { - throw new ArgumentException( - $"Document instance must be assignable to {documentType.FullName}.", - nameof(document)); - } + RequireInstanceOf(documentType, document); + return Task.FromResult(Write(document, key)); + } - if (_ambientTransaction.Value is { } transaction) - { - var staged = CreateStoredDocument(documentType, document, key); - transaction.Stage(staged); - return staged.Key; - } + public bool Remove(CultRecordKey key) + { + CultStoredDocument? existing; + lock (_gate) + _entries.TryGetValue(key.Value, out existing); + if (existing == null) + return false; - await _transactionGate.WaitAsync().ConfigureAwait(false); - try + Admit(Array.Empty(), new[] { existing }, null, home => { - var stored = CreateStoredDocument(documentType, document, key); - AddStoredDocumentInternal(stored, source: null, raiseUpdate: false); - return stored.Key; - } - finally - { - _transactionGate.Release(); - } + home?.Delete(existing); + return CultCommitOutcome.Committed; + }); + return true; + } + + public void Remove(CultRecordHandle handle) + { + Remove(handle.Key); + } + + // false: a condition failed; nothing was written, changed in memory, or published. + public bool Commit(Action stage) + { + return Land(stage, wait: true) == CultCommitOutcome.Committed; + } + + public CultCommitOutcome TryCommit(Action stage) + { + return Land(stage, wait: false); } public CultRecordHandle? TryGetHandle(T document) { if (document == null) throw new ArgumentNullException(nameof(document)); - return _documentHandles.TryGetValue(document, out var box) - ? new CultRecordHandle(box.Key) - : null; + lock (_gate) + return _handles.TryGetValue(document, out var box) ? new CultRecordHandle(box.Key) : null; } public object? Get(CultRecordKey key) { - lock (_stateGate) - { - if (_ambientTransaction.Value is { } transaction && - transaction.TryGet(key, out var staged)) - return staged?.Document; - return _entries.TryGetValue(key.Value, out var stored) - ? stored.Document - : null; - } + lock (_gate) + return _entries.TryGetValue(key.Value, out var stored) ? stored.Document : null; } public T? Get(CultRecordKey key) where T : class @@ -1330,469 +1398,352 @@ namespace GameCult.Caching public IEnumerable GetAll() where T : class { - lock (_stateGate) - { - var type = typeof(T); - return VisibleStoredDocuments() - .Where(entry => type.IsAssignableFrom(entry.Descriptor.DocumentType)) - .Select(entry => (T)entry.Document) - .ToArray(); - } + lock (_gate) + return _entries.Values.Select(entry => entry.Document).OfType().ToArray(); } public T? GetGlobal() where T : class { - return _globalKeys.TryGetValue(typeof(T), out var key) - ? Get(new CultRecordKey(key)) - : null; + return Single("global", () => _globals + .Where(pair => typeof(T).IsAssignableFrom(pair.Key)) + .Select(pair => pair.Value)); } public T? GetByName(string name) where T : class { - var type = typeof(T); - if (_nameMaps.TryGetValue(type, out var map) && - map.TryGetValue(name, out var key)) - { - return Get(new CultRecordKey(key)); - } - - return null; + return Single($"name '{name}'", () => _names + .Where(pair => typeof(T).IsAssignableFrom(pair.Key)) + .SelectMany(pair => pair.Value.TryGetValue(name, out var key) ? new[] { key } : Array.Empty())); } public T? GetByIndex(string alias, string value) where T : class { - if (_indexMaps.TryGetValue((typeof(T), alias), out var map) && - map.TryGetValue(value, out var key)) - { - return Get(new CultRecordKey(key)); - } - - return null; - } - - public void Remove(CultRecordHandle handle) - { - if (_ambientTransaction.Value is { } transaction) - { - transaction.Delete(handle.Key); - return; - } - - _transactionGate.Wait(); - try - { - if (_entries.TryGetValue(handle.Key.Value, out var stored)) - RemoveStoredDocumentInternal(stored, source: null, raiseUpdate: false); - } - finally - { - _transactionGate.Release(); - } - } - - public bool Remove(CultRecordKey key) - { - if (Get(key) == null) - { - return false; - } - - if (_ambientTransaction.Value is { } transaction) - { - transaction.Delete(key); - return true; - } - - _transactionGate.Wait(); - try - { - if (_entries.TryGetValue(key.Value, out var stored)) - RemoveStoredDocumentInternal(stored, source: null, raiseUpdate: false); - } - finally - { - _transactionGate.Release(); - } - return true; - } - - public Task DeleteAsync(CultRecordHandle handle) - { - Remove(handle); - return Task.CompletedTask; + return Single($"index {alias}='{value}'", () => _indexes + .Where(pair => pair.Key.Alias == alias && typeof(T).IsAssignableFrom(pair.Key.Type)) + .SelectMany(pair => pair.Value.TryGetValue(value, out var key) ? new[] { key } : Array.Empty())); } public void Dispose() { if (FlushAttachedStoresOnDispose && IsDirty) - { FlushAllBackingStores(); - } - foreach (var store in _backingStores.OfType()) - { + foreach (var store in BackingStores) store.Dispose(); - } - _transactionGate.Dispose(); _changes.Dispose(); } - private IEnumerable VisibleStoredDocuments() + internal static void RequireInstanceOf(Type documentType, object document) { - if (_ambientTransaction.Value == null) - return _entries.Values.ToArray(); - - var visible = _entries.ToDictionary(pair => pair.Key, pair => pair.Value, StringComparer.Ordinal); - foreach (var pair in _ambientTransaction.Value.Mutations) - { - if (pair.Value == null) - visible.Remove(pair.Key); - else - visible[pair.Key] = pair.Value; - } - return visible.Values.ToArray(); + if (documentType == null) throw new ArgumentNullException(nameof(documentType)); + if (document == null) throw new ArgumentNullException(nameof(document)); + if (!documentType.IsInstanceOfType(document)) + throw new ArgumentException($"Document instance must be assignable to {documentType.FullName}.", nameof(document)); } - private IReadOnlyList<(CultStoredDocument Stored, object? Previous, bool Removed)> CommitTransaction( - CultCacheTransaction transaction) + // The runtime type decides the schema; a minted storedAt is always later than the record it replaces. + internal CultStoredDocument CreateStoredDocument(object document, CultRecordKey? key) { - var mutations = transaction.SnapshotForCommit() - .OrderBy(pair => pair.Key, StringComparer.Ordinal) - .ToArray(); - if (mutations.Length == 0) - return Array.Empty<(CultStoredDocument, object?, bool)>(); - - if (_backingStores.Count > 1) + var descriptor = _registry.GetRequired(document.GetType()); + lock (_gate) { - throw new InvalidOperationException( - "A CultCache transaction requires zero or one durable backing store; multiple stores cannot share one atomic commit boundary."); + 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); + } + } + + internal CultStoredDocument? Observe(CultRecordKey key, object? current) + { + if (current == null) + return null; + lock (_gate) + { + if (_handles.TryGetValue(current, out var box) && box.Key.Equals(key) && + _entries.TryGetValue(key.Value, out var stored) && ReferenceEquals(stored.Document, current)) + return stored; } - CultStoredDocument[] previous; - lock (_stateGate) + throw new ArgumentException($"Expect({key.Value}) was given an instance this cache does not hold at that key.", nameof(current)); + } + + private static string MintStoredAt(string? previous) + { + var now = DateTimeOffset.UtcNow; + if (previous != null && + DateTimeOffset.TryParse(previous, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind, out var last) && + now <= last) + now = last.ToUniversalTime().AddTicks(1); + return now.ToString("O", CultureInfo.InvariantCulture); + } + + private CultRecordKey Write(object document, CultRecordKey? key) + { + var stored = CreateStoredDocument(document, key); + Admit(new[] { stored }, Array.Empty(), null, home => { - previous = mutations - .Select(pair => _entries.TryGetValue(pair.Key, out var stored) ? stored : null) - .Where(stored => stored != null) - .Cast() + home?.Push(stored); + return CultCommitOutcome.Committed; + }); + return stored.Key; + } + + private CultCommitOutcome Land(Action stage, bool wait) + { + if (stage == null) throw new ArgumentNullException(nameof(stage)); + var batch = new CultCacheBatch(this); + try + { + stage(batch); + } + finally + { + batch.Sealed = true; + } + + if (!Monitor.TryEnter(_gate, wait ? Timeout.Infinite : 0)) + return CultCommitOutcome.Contended; + 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 upserts = mutations.Where(pair => pair.Value != null).Select(pair => pair.Value!).ToArray(); - var deleted = mutations - .Where(pair => pair.Value == null) - .Select(pair => previous.FirstOrDefault(stored => string.Equals(stored.Key.Value, pair.Key, StringComparison.Ordinal))) - .Where(stored => stored != null) - .Cast() - .ToArray(); - - foreach (var store in _backingStores) - store.CommitBatch(upserts, deleted); - - var changes = new List<(CultStoredDocument Stored, object? Previous, bool Removed)>(); - lock (_stateGate) - { - foreach (var pair in mutations) + var request = new CultCommitRequest(upserts, deletes, batch.Expected.ToArray(), batch.ExpectsUnchanged); + return Admit(upserts, deletes, null, home => { - _entries.TryGetValue(pair.Key, out var existing); - if (existing != null) + if (home != null) { - RemoveIndexes(existing); - _documentHandles.Remove(existing.Document); + 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 (pair.Value == null) - { - if (existing != null) - { - _entries.TryRemove(pair.Key, out _); - changes.Add((existing, existing.Document, true)); - } - continue; - } + 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 + { + Monitor.Exit(_gate); + } + } - var stored = pair.Value; - _entries[pair.Key] = stored; - _documentHandles.Remove(stored.Document); - _documentHandles.Add(stored.Document, new DocumentHandleBox(stored.Key)); - AddIndexes(stored); - changes.Add((stored, existing?.Document, false)); - } - _hasUnflushedMutations = _backingStores.Any(store => store.IsDirty); + // Every add and remove passes here: single writes, committed batches, and loads (source set). + // Nothing lands unless all of it is admissible; land is the durable step between validation and memory. + private CultCommitOutcome Admit( + IReadOnlyList admitted, + IReadOnlyList evicted, + CacheBackingStore? source, + Func land) + { + List changes; + lock (_gate) + { + var outcome = land(Validate(admitted, evicted, source)); + if (outcome != CultCommitOutcome.Committed) + return outcome; + changes = Apply(admitted, evicted, source); } + foreach (var change in changes) + { + _changes.OnNext(change); + if (source != null) + OnUpdate?.Invoke(change.Previous, change.Document); + } + + return CultCommitOutcome.Committed; + } + + private CacheBackingStore? Validate( + IReadOnlyList admitted, + IReadOnlyList evicted, + CacheBackingStore? source) + { + var homes = new List(); + var globals = new Dictionary(_globals); + foreach (var stored in evicted) + { + var type = stored.Descriptor.DocumentType; + if (globals.TryGetValue(type, out var key) && key == stored.Key.Value) + globals.Remove(type); + if (source == null) + Claim(stored); + } + + foreach (var stored in admitted) + { + var descriptor = stored.Descriptor; + if (source == null) + Claim(stored); + else if (Home(descriptor.DocumentType) is var home && home != source) + throw new InvalidOperationException( + $"{descriptor.SchemaName} record {stored.Key.Value} was loaded from {source} but its home is {home?.ToString() ?? "no store"}."); + + if (!descriptor.IsGlobal) + continue; + if (globals.TryGetValue(descriptor.DocumentType, out var existing) && existing != stored.Key.Value) + throw new InvalidOperationException( + $"{descriptor.SchemaName} is a global and already has record {existing}; {stored.Key.Value} would be a second."); + globals[descriptor.DocumentType] = stored.Key.Value; + } + + var distinct = homes.Distinct().ToArray(); + if (distinct.Length > 1) + throw new InvalidOperationException( + $"The batch spans {distinct[0]} and {distinct[1]}; a commit lands in one home store."); + return distinct.FirstOrDefault(); + + void Claim(CultStoredDocument stored) + { + var home = Home(stored.Descriptor.DocumentType); + if (home == null) + { + if (_stores.Count > 0) + throw new InvalidOperationException($"No backing store is home to {stored.Descriptor.SchemaName}."); + return; + } + + if (home.IsReadOnly) + throw new InvalidOperationException( + $"{stored.Descriptor.SchemaName} record {stored.Key.Value} routes to read-only backing store {home}."); + homes.Add(home); + } + } + + private List Apply( + IReadOnlyList admitted, + IReadOnlyList evicted, + CacheBackingStore? source) + { + var changes = new List(admitted.Count + evicted.Count); + foreach (var stored in evicted) + { + if (!_entries.TryGetValue(stored.Key.Value, out var existing)) + continue; + Unindex(existing); + _entries.Remove(stored.Key.Value); + _handles.Remove(existing.Document); + changes.Add(new Change(CultCacheDocumentChangeKind.Removed, existing, null, existing.Document)); + } + + foreach (var stored in admitted) + { + _entries.TryGetValue(stored.Key.Value, out var previous); + if (previous != null) + { + Unindex(previous); + _handles.Remove(previous.Document); + } + + _entries[stored.Key.Value] = stored; + _handles.AddOrUpdate(stored.Document, new KeyBox(stored.Key)); + Index(stored); + changes.Add(new Change( + previous == null ? CultCacheDocumentChangeKind.Added : CultCacheDocumentChangeKind.Updated, + stored, + stored.Document, + previous?.Document)); + } + + if (source == null && _stores.Count == 0) + _dirtyInMemory = true; return changes; } - private sealed class CultCacheTransaction + // The most derived routed home type assignable from the document type, else the untyped store. + private CacheBackingStore? Home(Type documentType) { - private readonly object _gate = new(); - private readonly Dictionary _mutations = new(StringComparer.Ordinal); - private bool _sealed; - - public IReadOnlyDictionary Mutations + CacheBackingStore? untyped = null; + CacheBackingStore? routed = null; + Type? best = null; + foreach (var (store, homes) in _stores) { - get + if (homes.Length == 0) { - lock (_gate) + untyped = store; + continue; + } + + foreach (var home in homes) + { + if (home.IsAssignableFrom(documentType) && (best == null || best.IsAssignableFrom(home))) { - ThrowIfSealedForAmbientAccess(); - return new Dictionary(_mutations, StringComparer.Ordinal); + best = home; + routed = store; } } } - public void Stage(CultStoredDocument stored) - { - lock (_gate) - { - ThrowIfSealedForAmbientAccess(); - _mutations[stored.Key.Value] = stored; - } - } + return routed ?? untyped; + } - public void Delete(CultRecordKey key) + private T? Single(string lookup, Func> keys) where T : class + { + lock (_gate) { - lock (_gate) - { - ThrowIfSealedForAmbientAccess(); - _mutations[key.Value] = null; - } - } - - public bool TryGet(CultRecordKey key, out CultStoredDocument? stored) - { - lock (_gate) - { - ThrowIfSealedForAmbientAccess(); - return _mutations.TryGetValue(key.Value, out stored); - } - } - - public void Seal() - { - lock (_gate) - _sealed = true; - } - - public IReadOnlyDictionary SnapshotForCommit() - { - lock (_gate) - { - if (!_sealed) - throw new InvalidOperationException("A CultCache transaction must be sealed before commit."); - return new Dictionary(_mutations, StringComparer.Ordinal); - } - } - - private void ThrowIfSealedForAmbientAccess() - { - if (_sealed) + var found = keys().Distinct().ToArray(); + if (found.Length > 1) throw new InvalidOperationException( - "This CultCache transaction has already completed; escaped async work cannot read or mutate it."); + $"{typeof(T).Name} {lookup} matches records {string.Join(", ", found)}; look it up by its concrete type."); + return found.Length == 1 && _entries.TryGetValue(found[0], out var stored) ? stored.Document as T : null; } } - internal CultStoredDocument CreateStoredDocument(Type documentType, object document, CultRecordKey? key = null, string? storedAt = null) + private void Index(CultStoredDocument stored) { - var descriptor = _registry.GetRequired(documentType); - var resolvedKey = key ?? ResolveKey(document, descriptor); - return new CultStoredDocument( - resolvedKey, - storedAt ?? DateTimeOffset.UtcNow.ToString("O"), - descriptor, - document); - } - - private void AddStoredDocumentInternal( - CultStoredDocument stored, - CacheBackingStore? source, - bool raiseUpdate) - { - CultStoredDocument? existing = null; - lock (_stateGate) - { - _entries.TryGetValue(stored.Key.Value, out existing); - if (existing != null) - RemoveIndexes(existing); - - _entries[stored.Key.Value] = stored; - _documentHandles.Remove(stored.Document); - _documentHandles.Add(stored.Document, new DocumentHandleBox(stored.Key)); - AddIndexes(stored); - - foreach (var store in _backingStores) - { - if (store != source) - store.Push(stored); - } - - if (source == null) - _hasUnflushedMutations = true; - else - RecomputeDirtyState(); - } - - PublishChange(stored, existing?.Document); - - if (raiseUpdate) - { - OnUpdate?.Invoke(existing?.Document, stored.Document); - } - } - - private void RemoveStoredDocumentInternal( - CultStoredDocument stored, - CacheBackingStore? source, - bool raiseUpdate) - { - CultStoredDocument? existing; - lock (_stateGate) - { - if (!_entries.TryRemove(stored.Key.Value, out existing)) - return; - - RemoveIndexes(existing); - _documentHandles.Remove(existing.Document); - - foreach (var store in _backingStores) - { - if (store != source) - store.Delete(existing); - } - - if (source == null) - _hasUnflushedMutations = true; - else - RecomputeDirtyState(); - } - - PublishChange(existing, existing.Document, removed: true); - - if (raiseUpdate) - { - OnUpdate?.Invoke(existing.Document, null); - } - } - - private void InitializeGlobals() - { - foreach (var descriptor in _registry.AllDescriptors.Where(candidate => candidate.IsGlobal)) - { - var key = new CultRecordKey($"global:{descriptor.SchemaId}"); - if (_entries.ContainsKey(key.Value) || - _backingStores.Any(store => store.ContainsDurableRecord(key))) - { - continue; - } - - if (descriptor.DocumentType.GetConstructor(Type.EmptyTypes) == null) - { - continue; - } - - var instance = Activator.CreateInstance(descriptor.DocumentType); - if (instance == null) - { - continue; - } - - AddStoredDocumentInternal( - new CultStoredDocument( - key, - DateTimeOffset.UtcNow.ToString("O"), - descriptor, - instance), - source: null, - raiseUpdate: false); - } - } - - private CultRecordKey ResolveKey(object document, CultDocumentDescriptor descriptor) - { - if (_documentHandles.TryGetValue(document, out var existing)) - { - return existing.Key; - } - - if (descriptor.IsGlobal) - { - return new CultRecordKey($"global:{descriptor.SchemaId}"); - } - - return new CultRecordKey(Guid.NewGuid().ToString("N")); - } - - private void AddIndexes(CultStoredDocument stored) - { - var typeMap = _typeMaps.GetOrAdd( - stored.Descriptor.DocumentType, - _ => new ConcurrentDictionary(StringComparer.Ordinal)); - typeMap[stored.Key.Value] = stored; - + var type = stored.Descriptor.DocumentType; if (stored.Descriptor.IsGlobal) - { - _globalKeys[stored.Descriptor.DocumentType] = stored.Key.Value; - } - + _globals[type] = stored.Key.Value; if (stored.Descriptor.NameAccessor?.Invoke(stored.Document) is { Length: > 0 } name) - { - var map = _nameMaps.GetOrAdd( - stored.Descriptor.DocumentType, - _ => new ConcurrentDictionary(StringComparer.Ordinal)); - map[name] = stored.Key.Value; - } - + MapOf(_names, type)[name] = stored.Key.Value; foreach (var pair in stored.Descriptor.IndexAccessors) { var value = pair.Value(stored.Document); - if (string.IsNullOrWhiteSpace(value)) - { - continue; - } - - var map = _indexMaps.GetOrAdd( - (stored.Descriptor.DocumentType, pair.Key), - _ => new ConcurrentDictionary(StringComparer.Ordinal)); - map[value] = stored.Key.Value; + if (!string.IsNullOrWhiteSpace(value)) + MapOf(_indexes, (type, pair.Key))[value] = stored.Key.Value; } } - private void RemoveIndexes(CultStoredDocument stored) + // Documents are mutable, so a name or index value may have changed since it was indexed: drop by key. + private void Unindex(CultStoredDocument stored) { - if (_typeMaps.TryGetValue(stored.Descriptor.DocumentType, out var typeMap)) - typeMap.TryRemove(stored.Key.Value, out _); - - if (stored.Descriptor.IsGlobal) - { - _globalKeys.TryRemove(stored.Descriptor.DocumentType, out _); - } - - if (stored.Descriptor.NameAccessor?.Invoke(stored.Document) is { Length: > 0 } name && - _nameMaps.TryGetValue(stored.Descriptor.DocumentType, out var nameMap)) - { - nameMap.TryRemove(name, out _); - } - - foreach (var pair in stored.Descriptor.IndexAccessors) - { - var value = pair.Value(stored.Document); - if (string.IsNullOrWhiteSpace(value)) - { - continue; - } - - if (_indexMaps.TryGetValue((stored.Descriptor.DocumentType, pair.Key), out var map)) - { - map.TryRemove(value, out _); - } - } + var type = stored.Descriptor.DocumentType; + if (_globals.TryGetValue(type, out var globalKey) && globalKey == stored.Key.Value) + _globals.Remove(type); + if (_names.TryGetValue(type, out var names)) + RemoveKey(names, stored.Key.Value); + foreach (var pair in _indexes.Where(pair => pair.Key.Type == type)) + RemoveKey(pair.Value, stored.Key.Value); } - private sealed class DocumentHandleBox + private static Dictionary MapOf(Dictionary> maps, TKey key) + where TKey : notnull { - public DocumentHandleBox(CultRecordKey key) + if (!maps.TryGetValue(key, out var map)) + maps[key] = map = new Dictionary(StringComparer.Ordinal); + return map; + } + + private static void RemoveKey(Dictionary map, string key) + { + foreach (var stale in map.Where(pair => pair.Value == key).Select(pair => pair.Key).ToArray()) + map.Remove(stale); + } + + private sealed class KeyBox + { + public KeyBox(CultRecordKey key) { Key = key; } @@ -1800,56 +1751,50 @@ namespace GameCult.Caching public CultRecordKey Key { get; } } - private void RecomputeDirtyState() + private sealed class Change { - _hasUnflushedMutations = _backingStores.Count == 0 - ? _hasUnflushedMutations - : _backingStores.Any(store => store.IsDirty); - } - - private void PublishChange(CultStoredDocument stored, object? previousDocument, bool removed = false) - { - var changeType = typeof(CultCacheDocumentChange<>).MakeGenericType(stored.Descriptor.DocumentType); - var change = Activator.CreateInstance( - changeType, - removed ? CultCacheDocumentChangeKind.Removed : - previousDocument == null ? CultCacheDocumentChangeKind.Added : - CultCacheDocumentChangeKind.Updated, - stored.Key, - removed ? null : stored.Document, - previousDocument); - if (change != null) + public Change(CultCacheDocumentChangeKind kind, CultStoredDocument stored, object? document, object? previous) { - _changes.OnNext(change); + Kind = kind; + Stored = stored; + Document = document; + Previous = previous; } + + public CultCacheDocumentChangeKind Kind { get; } + public CultStoredDocument Stored { get; } + public object? Document { get; } + public object? Previous { get; } } } public abstract class CacheBackingStore : IDisposable { private CultDocumentRegistry? _registry; - private bool _isDirty; private CultSchemaMigrationReport[] _lastSchemaMigrationReports = Array.Empty(); + protected CacheBackingStore(bool readOnly = false) + { + IsReadOnly = readOnly; + } + protected CultDocumentRegistry Registry => _registry ?? throw new InvalidOperationException("Backing store is not attached to a CultDocumentRegistry."); protected ConcurrentDictionary Entries { get; } = new(StringComparer.Ordinal); - public bool IsDirty - { - get => _isDirty; - protected set => _isDirty = value; - } + public bool IsReadOnly { get; } + + public bool IsDirty { get; protected set; } public bool FlushOnDispose { get; set; } public IReadOnlyList LastSchemaMigrationReports => _lastSchemaMigrationReports; - // Set by the cache at attach; a store calls them for records it loads or drops. - protected internal Action? Loaded; - protected internal Action? Unloaded; + // Set by the cache at attach. A pull hands over everything it loaded and dropped in one call before it adopts + // any of it; if the cache refuses a record the call throws and the store keeps its previous view. + protected internal Action, IReadOnlyList>? Loaded; internal void AttachRegistry(CultDocumentRegistry registry) { @@ -1857,46 +1802,28 @@ namespace GameCult.Caching } public abstract void PullAll(); - public virtual bool ContainsDurableRecord(CultRecordKey key) - { - return Entries.ContainsKey(key.Value); - } public abstract void Push(CultStoredDocument entry); public abstract void Delete(CultStoredDocument entry); - // Stages and durably commits a batch as one step. Implementations must restore their prior - // staged view when finality fails. - public virtual void CommitBatch( - IReadOnlyCollection upserts, - IReadOnlyCollection deletes) - { - var previousEntries = Entries.ToDictionary(pair => pair.Key, pair => pair.Value, StringComparer.Ordinal); - var wasDirty = IsDirty; - try - { - foreach (var entry in deletes) - Delete(entry); - foreach (var entry in upserts) - Push(entry); - PushAll(); - } - catch - { - Entries.Clear(); - foreach (var pair in previousEntries) - Entries[pair.Key] = pair.Value; - IsDirty = wasDirty; - throw; - } - } public abstract void PushAll(); + + // One durable step under the store's exclusive lock; conditions are evaluated against what is on disk then. + // A failure restores the store's staged view. + public abstract CultCommitOutcome CommitBatch(CultCommitRequest request, bool wait); + public virtual void Dispose() { - if (FlushOnDispose && IsDirty) + if (FlushOnDispose && IsDirty && !IsReadOnly) { PushAll(); } } + protected void ThrowIfReadOnly() + { + if (IsReadOnly) + throw new InvalidOperationException($"Backing store {this} is read-only."); + } + protected CultPersistedRecord ToPersistedRecord(CultStoredDocument entry, Func serializePayload) { return new CultPersistedRecord @@ -1935,7 +1862,8 @@ namespace GameCult.Caching public abstract class SingleFileBackingStore : CacheBackingStore { - protected SingleFileBackingStore(string filePath) + protected SingleFileBackingStore(string filePath, bool readOnly = false) + : base(readOnly) { FileInfo = new FileInfo(filePath); } @@ -1946,93 +1874,164 @@ namespace GameCult.Caching protected abstract byte[] SerializePayload(object document); protected abstract object DeserializePayload(Type documentType, byte[] payload); + public override string ToString() => FileInfo.FullName; + public override void PullAll() { - FileInfo.Refresh(); - if (!FileInfo.Exists) - { - SetLastSchemaMigrationReports(Array.Empty()); - IsDirty = false; - return; - } - - // 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 those mutations before the next flush. Flush first; clean readers can - // still poll external snapshots incrementally. + // 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. if (IsDirty) return; - CultPersistedStoreSnapshot snapshot; - try + var snapshot = ReadSnapshot(); + if (snapshot == null) { - snapshot = DeserializeSnapshot(ReadAllBytesShared(FileInfo.FullName)); - } - catch (FileNotFoundException) - { - FileInfo.Refresh(); - if (!FileInfo.Exists) - return; - snapshot = DeserializeSnapshot(ReadAllBytesShared(FileInfo.FullName)); + SetLastSchemaMigrationReports(Array.Empty()); + return; } + var reports = new List(snapshot.Records.Length); - var persistedKeys = new HashSet(StringComparer.Ordinal); + var persisted = new Dictionary(StringComparer.Ordinal); foreach (var record in snapshot.Records) { - persistedKeys.Add(record.Key); reports.Add(Registry.ResolvePersistedSchemaReport(record.SchemaId, snapshot.SchemaCatalog)); var stored = ToStoredDocument(record, snapshot.SchemaCatalog, DeserializePayload); - if (Entries.TryGetValue(stored.Key.Value, out var existing) && - string.Equals(existing.StoredAt, stored.StoredAt, StringComparison.Ordinal) && - string.Equals(existing.Descriptor.SchemaId, stored.Descriptor.SchemaId, StringComparison.Ordinal)) - { - continue; - } + persisted[stored.Key.Value] = stored; + } + + var loaded = persisted.Values + .Where(stored => !Entries.TryGetValue(stored.Key.Value, out var existing) || + existing.StoredAt != stored.StoredAt || + existing.Descriptor.SchemaId != stored.Descriptor.SchemaId) + .ToArray(); + var dropped = Entries.Values.Where(existing => !persisted.ContainsKey(existing.Key.Value)).ToArray(); + if (loaded.Length > 0 || dropped.Length > 0) + Loaded?.Invoke(loaded, dropped); + + foreach (var stored in dropped) + Entries.TryRemove(stored.Key.Value, out _); + foreach (var stored in loaded) Entries[stored.Key.Value] = stored; - Loaded?.Invoke(stored); - } - - foreach (var removedKey in Entries.Keys.Where(key => !persistedKeys.Contains(key)).ToArray()) - { - if (Entries.TryRemove(removedKey, out var removed)) - Unloaded?.Invoke(removed); - } - SetLastSchemaMigrationReports(reports); - IsDirty = false; } public override void Push(CultStoredDocument entry) { + ThrowIfReadOnly(); Entries[entry.Key.Value] = entry; IsDirty = true; } public override void Delete(CultStoredDocument entry) { + ThrowIfReadOnly(); 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 + // one wins. Processes sharing a store must use conditional commit. public override void PushAll() { + ThrowIfReadOnly(); + using (AcquireLock(wait: true)) + { + WriteSnapshot( + Entries.Values.Select(entry => ToPersistedRecord(entry, SerializePayload)), + Entries.Values.Select(entry => entry.Descriptor.ToCatalogEntry())); + } + + MarkFlushSucceeded(); + } + + public override CultCommitOutcome CommitBatch(CultCommitRequest request, bool wait) + { + ThrowIfReadOnly(); + using var fileLock = AcquireLock(wait); + if (fileLock == null) + return CultCommitOutcome.Contended; + + var disk = ReadSnapshot() ?? new CultPersistedStoreSnapshot(); + 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; + foreach (var entry in request.Deletes) + records.Remove(entry.Key.Value); + foreach (var entry in request.Upserts) + records[entry.Key.Value] = ToPersistedRecord(entry, SerializePayload); + + WriteSnapshot(records.Values, catalog.Concat(request.Upserts.Select(entry => entry.Descriptor.ToCatalogEntry()))); + foreach (var entry in request.Deletes) + Entries.TryRemove(entry.Key.Value, out _); + foreach (var entry in request.Upserts) + Entries[entry.Key.Value] = entry; + MarkFlushSucceeded(); + return CultCommitOutcome.Committed; + } + + private CultPersistedStoreSnapshot? ReadSnapshot() + { + FileInfo.Refresh(); + if (!FileInfo.Exists) + return null; + try + { + return DeserializeSnapshot(ReadAllBytesShared(FileInfo.FullName)); + } + catch (FileNotFoundException) + { + return null; + } + } + + private void WriteSnapshot(IEnumerable records, IEnumerable catalog) + { + var ordered = records.OrderBy(record => record.Key, StringComparer.Ordinal).ToArray(); + var used = new HashSet(ordered.Select(record => record.SchemaId), StringComparer.Ordinal); var snapshot = new CultPersistedStoreSnapshot { - SchemaCatalog = Entries.Values - .Select(entry => entry.Descriptor.ToCatalogEntry()) + SchemaCatalog = catalog .GroupBy(entry => entry.SchemaId, StringComparer.Ordinal) .Select(group => group.First()) + .Where(entry => used.Contains(entry.SchemaId)) .OrderBy(entry => entry.SchemaName, StringComparer.Ordinal) .ToArray(), - Records = Entries.Values - .OrderBy(entry => entry.Key.Value, StringComparer.Ordinal) - .Select(entry => ToPersistedRecord(entry, SerializePayload)) - .ToArray() + Records = ordered }; Directory.CreateDirectory(FileInfo.DirectoryName!); WriteSnapshotAtomically(FileInfo.FullName, SerializeSnapshot(snapshot)); - MarkFlushSucceeded(); + } + + // The lock is a sidecar opened exclusively, which excludes other handles in this process and in others alike. + private FileStream? AcquireLock(bool wait) + { + Directory.CreateDirectory(FileInfo.DirectoryName!); + var lockPath = FileInfo.FullName + ".lock"; + var started = Stopwatch.StartNew(); + while (true) + { + try + { + return new FileStream(lockPath, FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.None, bufferSize: 1); + } + catch (IOException) when (!wait) + { + return null; + } + catch (IOException) when (started.Elapsed < TimeSpan.FromSeconds(30)) + { + Thread.Sleep(10); + } + } } private static void WriteSnapshotAtomically(string path, byte[] payload) diff --git a/src/GameCult.Networking/CultNetDatabase.cs b/src/GameCult.Networking/CultNetDatabase.cs index 74932da9..996e85f5 100644 --- a/src/GameCult.Networking/CultNetDatabase.cs +++ b/src/GameCult.Networking/CultNetDatabase.cs @@ -805,7 +805,7 @@ namespace GameCult.Networking ThrowIfDisposed(); if (document == null) throw new ArgumentNullException(nameof(document)); - var descriptor = _cache.Registry.GetRequired(); + var descriptor = _cache.Registry.GetRequired(document.GetType()); var shard = ResolveShardInternal(descriptor, key); EnsurePrimary(shard, descriptor.SchemaId, key); @@ -840,7 +840,7 @@ namespace GameCult.Networking ThrowIfDisposed(); if (document == null) throw new ArgumentNullException(nameof(document)); - var descriptor = _cache.Registry.GetRequired(); + var descriptor = _cache.Registry.GetRequired(document.GetType()); var shard = ResolveShardInternal(descriptor, key); EnsureClientAuthority(descriptor, key); diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs index 39cc4cf3..db1e2907 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs @@ -248,18 +248,12 @@ namespace GameCult.Networking }); } - private async Task DeleteAsync(CultNetDatabaseChangeRawMessage message) + private Task DeleteAsync(CultNetDatabaseChangeRawMessage message) { - if (string.IsNullOrWhiteSpace(message.RecordKey) || string.IsNullOrWhiteSpace(message.SchemaId)) return; - var descriptor = _documents.ResolveDescriptorForSchemaId(message.SchemaId!); - var handle = Activator.CreateInstance( - typeof(CultRecordHandle<>).MakeGenericType(descriptor.DocumentType), - new object[] { new CultRecordKey(message.RecordKey!) }); - var task = (Task)typeof(CultCache) - .GetMethod(nameof(CultCache.DeleteAsync), BindingFlags.Public | BindingFlags.Instance)! - .MakeGenericMethod(descriptor.DocumentType) - .Invoke(_cache, new[] { handle })!; - await task.ConfigureAwait(false); + if (string.IsNullOrWhiteSpace(message.RecordKey) || string.IsNullOrWhiteSpace(message.SchemaId)) return Task.CompletedTask; + _documents.ResolveDescriptorForSchemaId(message.SchemaId!); + _cache.Remove(new CultRecordKey(message.RecordKey!)); + return Task.CompletedTask; } private void Enqueue(Func operation) diff --git a/src/GameCult.Unity/Assets/Caching/Editor/CultCacheStudioWindow.cs b/src/GameCult.Unity/Assets/Caching/Editor/CultCacheStudioWindow.cs index e50ff1e6..6647c858 100644 --- a/src/GameCult.Unity/Assets/Caching/Editor/CultCacheStudioWindow.cs +++ b/src/GameCult.Unity/Assets/Caching/Editor/CultCacheStudioWindow.cs @@ -517,7 +517,7 @@ namespace GameCult.Unity.Caching.Editor if (!string.IsNullOrEmpty(selected)) { _path = selected; - OpenPath(true); + OpenPath(); } } @@ -528,16 +528,16 @@ namespace GameCult.Unity.Caching.Editor if (!string.IsNullOrEmpty(selected)) { _path = selected; - OpenPath(false); + OpenPath(); SaveCurrent(); } } - private void OpenPath(bool pullOnOpen) + private void OpenPath() { try { - _cache = CultCacheBridge.Open(_path, pullOnOpen); + _cache = CultCacheBridge.Open(_path); _selectedType = CultCacheBridge.GetDescriptors(_cache).FirstOrDefault()?.DocumentType; _selectedKey = string.Empty; EditorPrefs.SetString(LastPathKey, _path); @@ -680,11 +680,10 @@ namespace GameCult.Unity.Caching.Editor public static bool IsAvailable => CacheType != null && OpenOptionsType != null && MessagePackType != null && RecordKeyType != null; - public static object Open(string path, bool pullOnOpen) + public static object Open(string path) { EnsureAvailable(); var options = Activator.CreateInstance(OpenOptionsType); - OpenOptionsType.GetProperty("PullOnOpen").SetValue(options, pullOnOpen, null); var method = MessagePackType.GetMethod("OpenAsync", new[] { typeof(string), OpenOptionsType }); var task = (Task)method.Invoke(null, new[] { path, options }); task.GetAwaiter().GetResult(); diff --git a/tests/GameCult.Caching.Tests/BackingStoreTests.cs b/tests/GameCult.Caching.Tests/BackingStoreTests.cs index 668df1ed..85fc1d3f 100644 --- a/tests/GameCult.Caching.Tests/BackingStoreTests.cs +++ b/tests/GameCult.Caching.Tests/BackingStoreTests.cs @@ -285,14 +285,11 @@ namespace GameCult.Caching.Tests } [Test] - public async Task CultCacheMessagePack_OpenAsync_HydratesPersistedGlobalBeforeDefault() + public async Task CultCacheMessagePack_OpenAsync_HydratesPersistedGlobal() { var filePath = Path.Combine(Path.GetTempPath(), $"cultlib-tests-{Guid.NewGuid():N}.cc"); var recordsPath = DirectoryMessagePackBackingStore.DefaultRecordDirectoryPath(filePath); var registry = CultDocumentRegistry.ForTypes(new[] { typeof(GlobalTestEntry) }); - typeof(CultDocumentDescriptor) - .GetField("k__BackingField", BindingFlags.Instance | BindingFlags.NonPublic)! - .SetValue(registry.GetRequired(), true); try { @@ -300,10 +297,8 @@ namespace GameCult.Caching.Tests filePath, new CultCacheOpenOptions { Registry = registry, UseDirectoryStore = true })) { - var global = seed.GetGlobal(); - Assert.That(global, Is.Not.Null); - global!.Value = "durable authority"; - await seed.UpsertAsync(global); + Assert.That(seed.GetGlobal(), Is.Null, "opening a store invented a global"); + await seed.UpsertAsync(new GlobalTestEntry { Value = "durable authority" }); await seed.FlushAsync(); } @@ -312,8 +307,7 @@ namespace GameCult.Caching.Tests new CultCacheOpenOptions { Registry = registry, UseDirectoryStore = true }); Assert.That(reopened.GetGlobal()?.Value, Is.EqualTo("durable authority")); - Assert.That(reopened.IsDirty, Is.False, - "a hydrated durable global must not stage its schema default as a write"); + Assert.That(reopened.IsDirty, Is.False, "hydrating a global staged a write"); } finally { @@ -446,15 +440,16 @@ namespace GameCult.Caching.Tests var recordsPath = DirectoryMessagePackBackingStore.DefaultRecordDirectoryPath(filePath); try { + var readerStore = new DirectoryMessagePackBackingStore(filePath, recordsPath); + var readerCache = new CultCache(); + readerCache.AddBackingStore(readerStore); + var writerStore = new DirectoryMessagePackBackingStore(filePath, recordsPath); var writerCache = new CultCache(); writerCache.AddBackingStore(writerStore); await writerCache.UpsertAsync(new NamedTestEntry { Name = "source", Value = "one" }); writerStore.PushAll(); - var readerStore = new DirectoryMessagePackBackingStore(filePath, recordsPath); - var readerCache = new CultCache(); - readerCache.AddBackingStore(readerStore); var observerRan = false; using var subscription = readerCache.Watch().Subscribe(_ => { @@ -885,49 +880,40 @@ namespace GameCult.Caching.Tests } [Test] - public async Task CultCache_Transaction_Hides_Staged_Records_Until_Whole_Batch_Is_Durable() + public async Task CultCache_Commit_Hides_Batch_Until_The_Directory_Store_Has_Committed_It() { - var root = Path.Combine(Path.GetTempPath(), $"cultcache-transaction-{Guid.NewGuid():N}"); + var root = Path.Combine(Path.GetTempPath(), $"cultcache-commit-{Guid.NewGuid():N}"); var manifest = Path.Combine(root, "state.cc"); Directory.CreateDirectory(root); try { using var cache = new CultCache(); cache.AddBackingStore(new DirectoryMessagePackBackingStore(manifest)); - var firstKey = new CultRecordKey("transaction:first"); - var secondKey = new CultRecordKey("transaction:second"); - var staged = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var firstKey = new CultRecordKey("commit:first"); + var secondKey = new CultRecordKey("commit:second"); var observed = 0; using var subscription = cache.Watch().Subscribe(_ => observed++); - var transaction = cache.ExecuteTransactionAsync(async () => + var committed = cache.Commit(batch => { - await cache.UpsertAsync( - new NamedTestEntry { Name = "first", Value = "one" }, - new CultRecordHandle(firstKey)); - staged.SetResult(true); - await release.Task; - await cache.UpsertAsync( - new NamedTestEntry { Name = "second", Value = "two" }, - new CultRecordHandle(secondKey)); + batch.Upsert(new NamedTestEntry { Name = "first", Value = "one" }, new CultRecordHandle(firstKey)); + Assert.That(cache.Get(firstKey), Is.Null); + Assert.That(File.Exists(manifest), Is.False); + batch.Upsert(new NamedTestEntry { Name = "second", Value = "two" }, new CultRecordHandle(secondKey)); + Assert.That(observed, Is.Zero); }); - await staged.Task; - Assert.That(cache.Get(firstKey), Is.Null); - Assert.That(observed, Is.Zero); - - release.SetResult(true); - await transaction; + Assert.That(committed, Is.True); Assert.That(cache.Get(firstKey)?.Value, Is.EqualTo("one")); Assert.That(cache.Get(secondKey)?.Value, Is.EqualTo("two")); Assert.That(observed, Is.EqualTo(2)); + Assert.That(cache.IsDirty, Is.False); using var reopened = new CultCache(); reopened.AddBackingStore(new DirectoryMessagePackBackingStore(manifest)); - await reopened.PullAllBackingStoresAsync(); Assert.That(reopened.Get(firstKey)?.Value, Is.EqualTo("one")); Assert.That(reopened.Get(secondKey)?.Value, Is.EqualTo("two")); + await Task.CompletedTask; } finally { @@ -936,7 +922,7 @@ namespace GameCult.Caching.Tests } [Test] - public async Task CultCache_Transaction_Abort_Cannot_Leak_Into_Later_Commit() + public void CultCache_Commit_Stage_Failure_Cannot_Leak_Into_Later_Commit() { var root = Path.Combine(Path.GetTempPath(), $"cultcache-abort-{Guid.NewGuid():N}"); var manifest = Path.Combine(root, "state.cc"); @@ -945,33 +931,25 @@ namespace GameCult.Caching.Tests { using var cache = new CultCache(); cache.AddBackingStore(new DirectoryMessagePackBackingStore(manifest)); - var abortedKey = new CultRecordKey("transaction:aborted"); - var committedKey = new CultRecordKey("transaction:committed"); + var abortedKey = new CultRecordKey("commit:aborted"); + var committedKey = new CultRecordKey("commit:committed"); var observed = 0; using var subscription = cache.Watch().Subscribe(_ => observed++); - Assert.ThrowsAsync(async () => - await cache.ExecuteTransactionAsync(async () => - { - await cache.UpsertAsync( - new NamedTestEntry { Name = "aborted", Value = "must-not-leak" }, - new CultRecordHandle(abortedKey)); - throw new InvalidOperationException("abort probe"); - })); + Assert.Throws(() => cache.Commit(batch => + { + batch.Upsert(new NamedTestEntry { Name = "aborted", Value = "must-not-leak" }, new CultRecordHandle(abortedKey)); + throw new InvalidOperationException("abort probe"); + })); Assert.That(cache.Get(abortedKey), Is.Null); Assert.That(observed, Is.Zero); - await cache.ExecuteTransactionAsync(async () => - { - await cache.UpsertAsync( - new NamedTestEntry { Name = "committed", Value = "survives" }, - new CultRecordHandle(committedKey)); - }); + Assert.That(cache.Commit(batch => + batch.Upsert(new NamedTestEntry { Name = "committed", Value = "survives" }, new CultRecordHandle(committedKey))), Is.True); Assert.That(observed, Is.EqualTo(1)); using var reopened = new CultCache(); reopened.AddBackingStore(new DirectoryMessagePackBackingStore(manifest)); - await reopened.PullAllBackingStoresAsync(); Assert.That(reopened.Get(abortedKey), Is.Null); Assert.That(reopened.Get(committedKey)?.Value, Is.EqualTo("survives")); } @@ -982,45 +960,33 @@ namespace GameCult.Caching.Tests } [Test] - public async Task CultCache_Transaction_Rejects_Escaped_Async_Writes_After_Commit() + public void CultCache_Commit_Batch_Used_After_Commit_Throws() { - var root = Path.Combine(Path.GetTempPath(), $"cultcache-escaped-transaction-{Guid.NewGuid():N}"); + var root = Path.Combine(Path.GetTempPath(), $"cultcache-escaped-batch-{Guid.NewGuid():N}"); var manifest = Path.Combine(root, "state.cc"); Directory.CreateDirectory(root); try { using var cache = new CultCache(); cache.AddBackingStore(new DirectoryMessagePackBackingStore(manifest)); - var committedKey = new CultRecordKey("transaction:committed-parent"); - var escapedKey = new CultRecordKey("transaction:escaped-child"); - var releaseChild = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - Task? escapedWrite = null; + var committedKey = new CultRecordKey("commit:committed-parent"); + var escapedKey = new CultRecordKey("commit:escaped-child"); + CultCacheBatch? escaped = null; - await cache.ExecuteTransactionAsync(async () => + cache.Commit(batch => { - await cache.UpsertAsync( - new NamedTestEntry { Name = "parent", Value = "committed" }, - new CultRecordHandle(committedKey)); - escapedWrite = Task.Run(async () => - { - await releaseChild.Task; - await cache.UpsertAsync( - new NamedTestEntry { Name = "child", Value = "must-fail" }, - new CultRecordHandle(escapedKey)); - }); + batch.Upsert(new NamedTestEntry { Name = "parent", Value = "committed" }, new CultRecordHandle(committedKey)); + escaped = batch; }); - releaseChild.SetResult(true); - Assert.That(escapedWrite, Is.Not.Null); Assert.That( - async () => await escapedWrite!, - Throws.TypeOf().With.Message.Contains("already completed")); + () => escaped!.Upsert(new NamedTestEntry { Name = "child", Value = "must-fail" }, new CultRecordHandle(escapedKey)), + Throws.TypeOf().With.Message.Contains("already been committed")); Assert.That(cache.Get(committedKey)?.Value, Is.EqualTo("committed")); Assert.That(cache.Get(escapedKey), Is.Null); using var reopened = new CultCache(); reopened.AddBackingStore(new DirectoryMessagePackBackingStore(manifest)); - await reopened.PullAllBackingStoresAsync(); Assert.That(reopened.Get(committedKey)?.Value, Is.EqualTo("committed")); Assert.That(reopened.Get(escapedKey), Is.Null); } @@ -1031,118 +997,32 @@ namespace GameCult.Caching.Tests } [Test] - public async Task CultCache_Transaction_Rejects_Hydration_Inside_Its_Mutation_Scope() + public void CultCache_Commit_Observer_Writes_In_A_New_Commit() { - var root = Path.Combine(Path.GetTempPath(), $"cultcache-transaction-hydration-{Guid.NewGuid():N}"); - var manifest = Path.Combine(root, "state.cc"); - Directory.CreateDirectory(root); - try - { - using (var writer = new CultCache()) - { - writer.AddBackingStore(new DirectoryMessagePackBackingStore(manifest)); - await writer.UpsertAsync( - new NamedTestEntry { Name = "external", Value = "durable" }, - new CultRecordHandle(new CultRecordKey("transaction:external"))); - writer.FlushAllBackingStores(); - } - - using var reader = new CultCache(); - reader.AddBackingStore(new DirectoryMessagePackBackingStore(manifest)); - var observed = 0; - using var subscription = reader.Watch().Subscribe(_ => observed++); - - Assert.That( - async () => await reader.ExecuteTransactionAsync( - () => reader.PullAllBackingStoresAsync()), - Throws.TypeOf().With.Message.Contains("hydrate before")); - Assert.That(reader.Get(new CultRecordKey("transaction:external")), Is.Null); - Assert.That(observed, Is.Zero); - - await reader.PullAllBackingStoresAsync(); - Assert.That(reader.Get(new CultRecordKey("transaction:external"))?.Value, - Is.EqualTo("durable")); - Assert.That(observed, Is.EqualTo(1)); - } - finally - { - if (Directory.Exists(root)) Directory.Delete(root, recursive: true); - } - } - - [Test] - public async Task CultCache_Hydration_Waits_For_The_Active_Mutation_Transaction() - { - var root = Path.Combine(Path.GetTempPath(), $"cultcache-hydration-gate-{Guid.NewGuid():N}"); + var root = Path.Combine(Path.GetTempPath(), $"cultcache-observer-commit-{Guid.NewGuid():N}"); var manifest = Path.Combine(root, "state.cc"); Directory.CreateDirectory(root); try { using var cache = new CultCache(); cache.AddBackingStore(new DirectoryMessagePackBackingStore(manifest)); - var staged = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - - var transaction = cache.ExecuteTransactionAsync(async () => - { - await cache.UpsertAsync( - new NamedTestEntry { Name = "staged", Value = "committed" }, - new CultRecordHandle(new CultRecordKey("transaction:hydration-gate"))); - staged.SetResult(true); - await release.Task; - }); - - await staged.Task; - var hydration = cache.PullAllBackingStoresAsync(); - Assert.That(hydration.IsCompleted, Is.False, - "hydration must wait until the active mutation transaction has committed"); - - release.SetResult(true); - await transaction; - await hydration; - Assert.That(cache.Get(new CultRecordKey("transaction:hydration-gate"))?.Value, - Is.EqualTo("committed")); - } - finally - { - if (Directory.Exists(root)) Directory.Delete(root, recursive: true); - } - } - - [Test] - public async Task CultCache_Transaction_Observer_Writes_In_A_New_Commit() - { - var root = Path.Combine(Path.GetTempPath(), $"cultcache-observer-transaction-{Guid.NewGuid():N}"); - var manifest = Path.Combine(root, "state.cc"); - Directory.CreateDirectory(root); - try - { - using var cache = new CultCache(); - cache.AddBackingStore(new DirectoryMessagePackBackingStore(manifest)); - var firstKey = new CultRecordKey("transaction:observer-source"); - var secondKey = new CultRecordKey("transaction:observer-write"); + var firstKey = new CultRecordKey("commit:observer-source"); + var secondKey = new CultRecordKey("commit:observer-write"); using var subscription = cache.Watch().Subscribe(change => { if (!change.Key.Equals(firstKey)) return; - cache.ExecuteTransactionAsync(async () => - { - await cache.UpsertAsync( - new NamedTestEntry { Name = "observer", Value = "separate-commit" }, - new CultRecordHandle(secondKey)); - }).GetAwaiter().GetResult(); + cache.Commit(batch => batch.Upsert( + new NamedTestEntry { Name = "observer", Value = "separate-commit" }, + new CultRecordHandle(secondKey))); }); - await cache.ExecuteTransactionAsync(async () => - { - await cache.UpsertAsync( - new NamedTestEntry { Name = "source", Value = "committed-first" }, - new CultRecordHandle(firstKey)); - }); + cache.Commit(batch => batch.Upsert( + new NamedTestEntry { Name = "source", Value = "committed-first" }, + new CultRecordHandle(firstKey))); Assert.That(cache.Get(secondKey)?.Value, Is.EqualTo("separate-commit")); using var reopened = new CultCache(); reopened.AddBackingStore(new DirectoryMessagePackBackingStore(manifest)); - await reopened.PullAllBackingStoresAsync(); Assert.That(reopened.Get(firstKey)?.Value, Is.EqualTo("committed-first")); Assert.That(reopened.Get(secondKey)?.Value, Is.EqualTo("separate-commit")); } @@ -1170,6 +1050,7 @@ namespace GameCult.Caching.Tests } [CultDocument("tests.global_entry", "tests.global_entry.v1")] + [CultGlobal] internal sealed class GlobalTestEntry { [Key(0)] diff --git a/tests/GameCult.Caching.Tests/PreCut2StoreFormatTests.cs b/tests/GameCult.Caching.Tests/PreCut2StoreFormatTests.cs index 3b442cf3..6089b94e 100644 --- a/tests/GameCult.Caching.Tests/PreCut2StoreFormatTests.cs +++ b/tests/GameCult.Caching.Tests/PreCut2StoreFormatTests.cs @@ -104,11 +104,10 @@ namespace GameCult.Caching.Tests File.WriteAllBytes(manifest, bytes); using var cache = new CultCache(); - var store = new DirectoryMessagePackBackingStore(manifest); - cache.AddBackingStore(store); - Assert.That(() => store.PullAll(), + Assert.That(() => cache.AddBackingStore(new DirectoryMessagePackBackingStore(manifest)), Throws.TypeOf().With.Message.Contains(formatVersion)); + Assert.That(cache.BackingStores, Is.Empty); Assert.That(File.ReadAllBytes(manifest), Is.EqualTo(bytes)); } finally diff --git a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs index 87f1a6b5..cd80f6e5 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs @@ -1761,8 +1761,7 @@ public sealed class CultMeshStreamingTests StartServer = false, CacheOptions = new CultCacheOpenOptions { - Registry = sourceCache.Registry, - PullOnOpen = false + Registry = sourceCache.Registry }, DatabaseOptions = new CultNetDatabaseOptions { @@ -1820,8 +1819,7 @@ public sealed class CultMeshStreamingTests StartServer = false, CacheOptions = new CultCacheOpenOptions { - Registry = sourceCache.Registry, - PullOnOpen = false + Registry = sourceCache.Registry }, DatabaseOptions = new CultNetDatabaseOptions { @@ -1846,8 +1844,7 @@ public sealed class CultMeshStreamingTests StartServer = false, CacheOptions = new CultCacheOpenOptions { - Registry = sourceCache.Registry, - PullOnOpen = false + Registry = sourceCache.Registry }, DatabaseOptions = new CultNetDatabaseOptions { @@ -1899,8 +1896,7 @@ public sealed class CultMeshStreamingTests StartServer = false, CacheOptions = new CultCacheOpenOptions { - Registry = sourceCache.Registry, - PullOnOpen = false + Registry = sourceCache.Registry }, DatabaseOptions = new CultNetDatabaseOptions { @@ -2760,8 +2756,7 @@ public sealed class CultMeshStreamingTests StartServer = false, CacheOptions = new CultCacheOpenOptions { - Registry = sourceCache.Registry, - PullOnOpen = false + Registry = sourceCache.Registry }, DatabaseOptions = new CultNetDatabaseOptions { @@ -2835,8 +2830,7 @@ public sealed class CultMeshStreamingTests StartServer = false, CacheOptions = new CultCacheOpenOptions { - Registry = targetCache.Registry, - PullOnOpen = false + Registry = targetCache.Registry }, DatabaseOptions = new CultNetDatabaseOptions { @@ -2993,8 +2987,7 @@ public sealed class CultMeshStreamingTests StartServer = false, CacheOptions = new CultCacheOpenOptions { - Registry = cacheRegistry, - PullOnOpen = false + Registry = cacheRegistry }, DatabaseOptions = new CultNetDatabaseOptions { @@ -3030,8 +3023,7 @@ public sealed class CultMeshStreamingTests StartServer = false, CacheOptions = new CultCacheOpenOptions { - Registry = sourceCache.Registry, - PullOnOpen = false + Registry = sourceCache.Registry }, DatabaseOptions = new CultNetDatabaseOptions { @@ -3170,8 +3162,7 @@ public sealed class CultMeshStreamingTests StartServer = false, CacheOptions = new CultCacheOpenOptions { - Registry = sourceCache.Registry, - PullOnOpen = false + Registry = sourceCache.Registry }, DatabaseOptions = new CultNetDatabaseOptions {