diff --git a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs index 4f1a2521..454a8ed8 100644 --- a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs +++ b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs @@ -253,6 +253,52 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore } } + /// + public override void CommitBatch( + IReadOnlyCollection upserts, + IReadOnlyCollection deletes, + bool soft) + { + lock (_mutationGate) + { + 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) + { + Entries.TryRemove(entry.Key.Value, out _); + _dirtyKeys.TryRemove(entry.Key.Value, out _); + _deletedKeys[entry.Key.Value] = true; + } + foreach (var entry in upserts) + { + Entries[entry.Key.Value] = entry; + _dirtyKeys[entry.Key.Value] = true; + _deletedKeys.TryRemove(entry.Key.Value, out _); + } + IsDirty = true; + PushAllCore(soft); + } + catch + { + Entries.Clear(); + foreach (var pair in previousEntries) + Entries[pair.Key] = pair.Value; + _dirtyKeys.Clear(); + foreach (var key in previousDirtyKeys) + _dirtyKeys[key] = true; + _deletedKeys.Clear(); + foreach (var key in previousDeletedKeys) + _deletedKeys[key] = true; + IsDirty = wasDirty; + throw; + } + } + } + /// public override void PushAll(bool soft = false) { diff --git a/src/GameCult.Caching/Contracts/cultcache-persistence-format.md b/src/GameCult.Caching/Contracts/cultcache-persistence-format.md index 66d0e135..e7ffd54f 100644 --- a/src/GameCult.Caching/Contracts/cultcache-persistence-format.md +++ b/src/GameCult.Caching/Contracts/cultcache-persistence-format.md @@ -39,6 +39,21 @@ These are **not** domain fields and should not be exposed on app-facing types: Those belong to the CultCache persistence layer. +## Transaction Visibility + +An authoritative multi-record change uses `CultCache.ExecuteTransactionAsync`. +The executing async flow sees its buffered overlay while staging; other readers +and all observers continue to see the prior committed generation. The backing +store commits the complete batch before the live cache swaps to it. Observer +notifications run only after the transaction context and commit lock have been +released, so an observer-triggered write begins a separate transaction. + +An exception before durable finality discards the overlay. It cannot become +visible through a later flush. A transaction currently permits at most one +durable backing store because independent stores cannot provide one atomic +commit boundary. Applications that require authoritative writes may configure +`CultNetDatabase` to reject record-at-a-time writes outside this primitive. + ## Canonical Store Shape A persisted CultCache file should contain: diff --git a/src/GameCult.Caching/CultCache.cs b/src/GameCult.Caching/CultCache.cs index 21b437af..1d1dab81 100644 --- a/src/GameCult.Caching/CultCache.cs +++ b/src/GameCult.Caching/CultCache.cs @@ -7,6 +7,7 @@ using System.Reflection; using System.Runtime.CompilerServices; using System.Security.Cryptography; using System.Text; +using System.Threading; using System.Threading.Tasks; using GameCult.Logging; using R3; @@ -1278,6 +1279,9 @@ namespace GameCult.Caching private readonly ConditionalWeakTable _documentHandles = new(); private readonly Subject _changes = new(); private readonly CultCacheSoaStore _soa = new(); + private readonly SemaphoreSlim _transactionGate = new(1, 1); + private readonly AsyncLocal _ambientTransaction = new(); + private readonly object _stateGate = new(); private ILogger _logger = new NullLogger(); private bool _hasUnflushedMutations; @@ -1348,14 +1352,29 @@ namespace GameCult.Caching /// /// Gets all document instances currently held by the cache. /// - public IEnumerable AllEntries => _entries.Values.Select(entry => entry.Document); + public IEnumerable AllEntries + { + get + { + lock (_stateGate) + return VisibleStoredDocuments().Select(entry => entry.Document).ToArray(); + } + } /// /// Gets all stored document records currently held by the cache. /// - public IEnumerable AllStoredDocuments => - _entries.Values.OrderBy(entry => entry.Descriptor.SchemaName, StringComparer.Ordinal) - .ThenBy(entry => entry.Key.Value, StringComparer.Ordinal); + public IEnumerable AllStoredDocuments + { + get + { + lock (_stateGate) + return VisibleStoredDocuments() + .OrderBy(entry => entry.Descriptor.SchemaName, StringComparer.Ordinal) + .ThenBy(entry => entry.Key.Value, StringComparer.Ordinal) + .ToArray(); + } + } /// /// Gets the document registry used by this cache. @@ -1457,15 +1476,22 @@ namespace GameCult.Caching /// public void FlushAllBackingStores(bool soft = false) { - foreach (var store in _backingStores) - { - FlushBackingStore(store, soft); - } + if (_ambientTransaction.Value != null) + throw new InvalidOperationException("A CultCache transaction owns durable commit; do not flush inside its stage callback."); - RecomputeDirtyState(); - if (!IsDirty) + _transactionGate.Wait(); + try { - LastSuccessfulFlushAtUtc = DateTimeOffset.UtcNow; + foreach (var store in _backingStores) + FlushBackingStoreCore(store, soft); + + RecomputeDirtyState(); + if (!IsDirty) + LastSuccessfulFlushAtUtc = DateTimeOffset.UtcNow; + } + finally + { + _transactionGate.Release(); } } @@ -1489,6 +1515,22 @@ namespace GameCult.Caching throw new InvalidOperationException("Backing store is not attached to this cache."); } + if (_ambientTransaction.Value != null) + throw new InvalidOperationException("A CultCache transaction owns durable commit; do not flush inside its stage callback."); + + _transactionGate.Wait(); + try + { + FlushBackingStoreCore(store, soft); + } + finally + { + _transactionGate.Release(); + } + } + + private void FlushBackingStoreCore(CacheBackingStore store, bool soft) + { store.PushAll(soft); RecomputeDirtyState(); if (!store.IsDirty) @@ -1514,17 +1556,77 @@ namespace GameCult.Caching return Task.CompletedTask; } + /// + /// Runs a buffered cache mutation. Staged records are visible only to the executing + /// async flow. The durable backing store is committed before the live cache and its + /// observers advance; an exception discards the entire staged batch. + /// + public async Task ExecuteTransactionAsync(Func stageAsync, bool soft = false) + { + 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); + changes = CommitTransaction(transaction, soft); + } + finally + { + _ambientTransaction.Value = null; + _transactionGate.Release(); + } + + foreach (var change in changes) + PublishChange(change.Stored, change.Previous, change.Removed); + } + + /// Runs a value-producing buffered cache mutation. + public async Task ExecuteTransactionAsync(Func> stageAsync, bool soft = false) + { + if (stageAsync == null) throw new ArgumentNullException(nameof(stageAsync)); + T result = default!; + await ExecuteTransactionAsync(async () => + { + result = await stageAsync().ConfigureAwait(false); + }, soft).ConfigureAwait(false); + return result; + } + /// /// Adds or replaces a typed document and returns its record handle. /// public async Task> AddAsync(T document, CultRecordHandle? handle = null) { if (document == null) throw new ArgumentNullException(nameof(document)); - var stored = await AddStoredDocumentInternal( - CreateStoredDocument(typeof(T), document, handle?.Key), - source: null, - raiseUpdate: false); - return new CultRecordHandle(stored.Key); + 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 = await AddStoredDocumentInternal( + CreateStoredDocument(typeof(T), document, handle?.Key), + source: null, + raiseUpdate: false).ConfigureAwait(false); + return new CultRecordHandle(stored.Key); + } + finally + { + _transactionGate.Release(); + } } /// @@ -1549,11 +1651,26 @@ namespace GameCult.Caching nameof(document)); } - var stored = await AddStoredDocumentInternal( - CreateStoredDocument(documentType, document, key), - source: null, - raiseUpdate: false).ConfigureAwait(false); - return stored.Key; + if (_ambientTransaction.Value is { } transaction) + { + var staged = CreateStoredDocument(documentType, document, key); + transaction.Stage(staged); + return staged.Key; + } + + await _transactionGate.WaitAsync().ConfigureAwait(false); + try + { + var stored = await AddStoredDocumentInternal( + CreateStoredDocument(documentType, document, key), + source: null, + raiseUpdate: false).ConfigureAwait(false); + return stored.Key; + } + finally + { + _transactionGate.Release(); + } } /// @@ -1572,9 +1689,15 @@ namespace GameCult.Caching /// public object? Get(CultRecordKey key) { - return _entries.TryGetValue(key.Value, out var stored) - ? stored.Document - : null; + 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; + } } /// @@ -1607,17 +1730,13 @@ namespace GameCult.Caching /// public IEnumerable GetStoredDocuments() where T : class { - var type = typeof(T); - if (type.IsSealed) + lock (_stateGate) { - return _typeMaps.TryGetValue(type, out var exact) - ? exact.Values - : Enumerable.Empty(); + var type = typeof(T); + return VisibleStoredDocuments() + .Where(entry => type.IsAssignableFrom(entry.Descriptor.DocumentType)) + .ToArray(); } - - return _typeMaps - .Where(pair => type.IsAssignableFrom(pair.Key)) - .SelectMany(pair => pair.Value.Values); } /// @@ -1690,9 +1809,21 @@ namespace GameCult.Caching /// public void Remove(CultRecordHandle handle) { - if (_entries.TryGetValue(handle.Key.Value, out var stored)) + if (_ambientTransaction.Value is { } transaction) { - RemoveStoredDocumentInternal(stored, source: null, raiseUpdate: false); + 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(); } } @@ -1701,12 +1832,27 @@ namespace GameCult.Caching /// public bool Remove(CultRecordKey key) { - if (!_entries.TryGetValue(key.Value, out var stored)) + if (Get(key) == null) { return false; } - RemoveStoredDocumentInternal(stored, source: null, raiseUpdate: 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; } @@ -1734,9 +1880,116 @@ namespace GameCult.Caching store.Dispose(); } + _transactionGate.Dispose(); _changes.Dispose(); } + private IEnumerable VisibleStoredDocuments() + { + 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(); + } + + private IReadOnlyList<(CultStoredDocument Stored, object? Previous, bool Removed)> CommitTransaction( + CultCacheTransaction transaction, + bool soft) + { + var mutations = transaction.Mutations + .OrderBy(pair => pair.Key, StringComparer.Ordinal) + .ToArray(); + if (mutations.Length == 0) + return Array.Empty<(CultStoredDocument, object?, bool)>(); + + if (_backingStores.Count > 1) + { + throw new InvalidOperationException( + "A CultCache transaction requires zero or one durable backing store; multiple stores cannot share one atomic commit boundary."); + } + + CultStoredDocument[] previous; + lock (_stateGate) + { + previous = mutations + .Select(pair => _entries.TryGetValue(pair.Key, out var stored) ? stored : null) + .Where(stored => stored != null) + .Cast() + .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, soft); + + var changes = new List<(CultStoredDocument Stored, object? Previous, bool Removed)>(); + lock (_stateGate) + { + foreach (var pair in mutations) + { + _entries.TryGetValue(pair.Key, out var existing); + if (existing != null) + { + RemoveIndexes(existing); + _documentHandles.Remove(existing.Document); + _soa.Remove(existing); + } + + if (pair.Value == null) + { + if (existing != null) + { + _entries.TryRemove(pair.Key, out _); + changes.Add((existing, existing.Document, true)); + } + continue; + } + + var stored = pair.Value; + _entries[pair.Key] = stored; + _documentHandles.Remove(stored.Document); + _documentHandles.Add(stored.Document, new DocumentHandleBox(stored.Key)); + _soa.Upsert(stored); + AddIndexes(stored); + changes.Add((stored, existing?.Document, false)); + } + _hasUnflushedMutations = _backingStores.Any(store => store.IsDirty); + if (!IsDirty) + LastSuccessfulFlushAtUtc = DateTimeOffset.UtcNow; + } + + return changes; + } + + private sealed class CultCacheTransaction + { + private readonly Dictionary _mutations = new(StringComparer.Ordinal); + + public IReadOnlyDictionary Mutations => _mutations; + + public void Stage(CultStoredDocument stored) => _mutations[stored.Key.Value] = stored; + + public void Delete(CultRecordKey key) => _mutations[key.Value] = null; + + public bool TryGet(CultRecordKey key, out CultStoredDocument? stored) => + _mutations.TryGetValue(key.Value, out stored); + } + internal CultStoredDocument CreateStoredDocument(Type documentType, object document, CultRecordKey? key = null, string? storedAt = null) { var descriptor = _registry.GetRequired(documentType); @@ -1754,33 +2007,28 @@ namespace GameCult.Caching bool raiseUpdate) { CultStoredDocument? existing = null; - _entries.TryGetValue(stored.Key.Value, out existing); - if (existing != null) + lock (_stateGate) { - RemoveIndexes(existing); - } + _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)); - _soa.Upsert(stored); - AddIndexes(stored); + _entries[stored.Key.Value] = stored; + _documentHandles.Remove(stored.Document); + _documentHandles.Add(stored.Document, new DocumentHandleBox(stored.Key)); + _soa.Upsert(stored); + AddIndexes(stored); - foreach (var store in _backingStores) - { - if (store != source) + foreach (var store in _backingStores) { - store.Push(stored); + if (store != source) + store.Push(stored); } - } - if (source == null) - { - _hasUnflushedMutations = true; - } - else - { - RecomputeDirtyState(); + if (source == null) + _hasUnflushedMutations = true; + else + RecomputeDirtyState(); } PublishChange(stored, existing?.Document); @@ -1799,30 +2047,26 @@ namespace GameCult.Caching CacheBackingStore? source, bool raiseUpdate) { - if (!_entries.TryRemove(stored.Key.Value, out var existing)) + CultStoredDocument? existing; + lock (_stateGate) { - return; - } + if (!_entries.TryRemove(stored.Key.Value, out existing)) + return; - RemoveIndexes(existing); - _documentHandles.Remove(existing.Document); - _soa.Remove(existing); + RemoveIndexes(existing); + _documentHandles.Remove(existing.Document); + _soa.Remove(existing); - foreach (var store in _backingStores) - { - if (store != source) + foreach (var store in _backingStores) { - store.Delete(existing); + if (store != source) + store.Delete(existing); } - } - if (source == null) - { - _hasUnflushedMutations = true; - } - else - { - RecomputeDirtyState(); + if (source == null) + _hasUnflushedMutations = true; + else + RecomputeDirtyState(); } PublishChange(existing, existing.Document, removed: true); @@ -2104,6 +2348,34 @@ namespace GameCult.Caching /// public abstract void Delete(CultStoredDocument entry); /// + /// Atomically stages and durably commits a buffered record batch for this store. + /// Implementations must restore their prior staged view when finality fails. + /// + public virtual void CommitBatch( + IReadOnlyCollection upserts, + IReadOnlyCollection deletes, + bool soft) + { + 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(soft); + } + catch + { + Entries.Clear(); + foreach (var pair in previousEntries) + Entries[pair.Key] = pair.Value; + IsDirty = wasDirty; + throw; + } + } + /// /// Persists all current backing store entries. /// public abstract void PushAll(bool soft = false); diff --git a/src/GameCult.Networking/CultNetDatabase.cs b/src/GameCult.Networking/CultNetDatabase.cs index 8000e45b..9c1d1480 100644 --- a/src/GameCult.Networking/CultNetDatabase.cs +++ b/src/GameCult.Networking/CultNetDatabase.cs @@ -2,6 +2,7 @@ using System; using System.Collections.Generic; using System.Linq; using System.Reflection; +using System.Threading; using System.Threading.Tasks; using GameCult.Caching; using R3; @@ -297,6 +298,11 @@ namespace GameCult.Networking /// Gets or sets the durable store for accepted shard mutation logs. /// public ICultNetShardMutationLogStore? MutationLogStore { get; set; } + /// + /// Gets or sets whether authoritative writes must occur inside ExecuteTransactionAsync. + /// Use this for state owners that forbid record-at-a-time mutation bypasses. + /// + public bool RequireTransactionsForAuthoritativeWrites { get; set; } } /// @@ -464,6 +470,8 @@ namespace GameCult.Networking private readonly Dictionary _nextLogSequences = new(StringComparer.Ordinal); private readonly Dictionary _appliedShardSequences = new(StringComparer.Ordinal); private readonly Subject _changes = new(); + private readonly AsyncLocal _ambientTransaction = new(); + private readonly bool _requireTransactionsForAuthoritativeWrites; private bool _disposed; /// @@ -476,6 +484,7 @@ namespace GameCult.Networking _runtimeId = options.RuntimeId; _documents = options.DocumentRegistry ?? new CultNetDocumentRegistry(cache.Registry); _mutationLogStore = options.MutationLogStore; + _requireTransactionsForAuthoritativeWrites = options.RequireTransactionsForAuthoritativeWrites; _shards = (options.Shards == null || options.Shards.Count == 0 ? new[] { CultNetShardDescriptor.PrimaryAll(options.RuntimeId) } : options.Shards) @@ -503,6 +512,49 @@ namespace GameCult.Networking /// public IReadOnlyList ClientAuthorityScopes => _clientAuthorityScopes; + /// + /// Runs one buffered authoritative database mutation. Cache state, durable state, + /// mutation logs, and database observers advance only after the complete stage succeeds. + /// + public async Task ExecuteTransactionAsync(Func stageAsync, bool soft = false) + { + ThrowIfDisposed(); + if (stageAsync == null) throw new ArgumentNullException(nameof(stageAsync)); + if (_ambientTransaction.Value != null) + { + await stageAsync().ConfigureAwait(false); + return; + } + + var transaction = new CultNetDatabaseTransaction(); + IReadOnlyList afterCommit; + _ambientTransaction.Value = transaction; + try + { + await _cache.ExecuteTransactionAsync(stageAsync, soft).ConfigureAwait(false); + afterCommit = transaction.AfterCommit.ToArray(); + } + finally + { + _ambientTransaction.Value = null; + } + + foreach (var publish in afterCommit) + publish(); + } + + /// Runs one value-producing buffered authoritative database mutation. + public async Task ExecuteTransactionAsync(Func> stageAsync, bool soft = false) + { + if (stageAsync == null) throw new ArgumentNullException(nameof(stageAsync)); + T result = default!; + await ExecuteTransactionAsync(async () => + { + result = await stageAsync().ConfigureAwait(false); + }, soft).ConfigureAwait(false); + return result; + } + /// Gets whether this database instance can authoritatively replace the document at a key. public bool CanWriteAuthoritatively(CultRecordKey key) where T : class { @@ -817,6 +869,7 @@ namespace GameCult.Networking public async Task> PutAsync(CultRecordKey key, T document) where T : class { ThrowIfDisposed(); + EnsureAuthoritativeTransaction(); if (document == null) throw new ArgumentNullException(nameof(document)); var descriptor = _cache.Registry.GetRequired(); @@ -825,20 +878,27 @@ namespace GameCult.Networking var previous = _cache.Get(key); var handle = await _cache.UpsertAsync(document, new CultRecordHandle(key)).ConfigureAwait(false); - AppendMutationLogEntry( - shard, - previous == null ? CultNetDatabaseChangeKind.Added : CultNetDatabaseChangeKind.Updated, - key, - descriptor.SchemaId, - document, - previous); - Publish(new CultNetDatabaseChange( - previous == null ? CultNetDatabaseChangeKind.Added : CultNetDatabaseChangeKind.Updated, - key, - descriptor.SchemaId, - shard, - document, - previous)); + void PublishCommittedPut() + { + AppendMutationLogEntry( + shard, + previous == null ? CultNetDatabaseChangeKind.Added : CultNetDatabaseChangeKind.Updated, + key, + descriptor.SchemaId, + document, + previous); + Publish(new CultNetDatabaseChange( + previous == null ? CultNetDatabaseChangeKind.Added : CultNetDatabaseChangeKind.Updated, + key, + descriptor.SchemaId, + shard, + document, + previous)); + } + if (_ambientTransaction.Value is { } transaction) + transaction.AfterCommit.Add(PublishCommittedPut); + else + PublishCommittedPut(); return handle; } @@ -873,6 +933,7 @@ namespace GameCult.Networking public Task DeleteAsync(CultRecordKey key) where T : class { ThrowIfDisposed(); + EnsureAuthoritativeTransaction(); var descriptor = _cache.Registry.GetRequired(); var shard = ResolveShardInternal(descriptor, key); EnsurePrimary(shard, descriptor.SchemaId, key); @@ -881,20 +942,27 @@ namespace GameCult.Networking if (previous != null) { _cache.Remove(new CultRecordHandle(key)); - AppendMutationLogEntry( - shard, - CultNetDatabaseChangeKind.Removed, - key, - descriptor.SchemaId, - document: null, - previousDocument: previous); - Publish(new CultNetDatabaseChange( - CultNetDatabaseChangeKind.Removed, - key, - descriptor.SchemaId, - shard, - document: null, - previousDocument: previous)); + void PublishCommittedDelete() + { + AppendMutationLogEntry( + shard, + CultNetDatabaseChangeKind.Removed, + key, + descriptor.SchemaId, + document: null, + previousDocument: previous); + Publish(new CultNetDatabaseChange( + CultNetDatabaseChangeKind.Removed, + key, + descriptor.SchemaId, + shard, + document: null, + previousDocument: previous)); + } + if (_ambientTransaction.Value is { } transaction) + transaction.AfterCommit.Add(PublishCommittedDelete); + else + PublishCommittedDelete(); } return Task.CompletedTask; @@ -906,6 +974,7 @@ namespace GameCult.Networking public async Task ApplyPutAsync(CultNetDocumentPutRawMessage message) { ThrowIfDisposed(); + EnsureAuthoritativeTransaction(); if (message == null) throw new ArgumentNullException(nameof(message)); if (message.Document == null) { @@ -920,31 +989,41 @@ namespace GameCult.Networking var document = await _documents.ApplyRawDocumentPutMessageAsync(_cache, message).ConfigureAwait(false); var kind = previous == null ? CultNetDatabaseChangeKind.Added : CultNetDatabaseChangeKind.Updated; var predictionKey = PredictionKey(descriptor.SchemaId, key); - if (_predictedDocuments.Remove(predictionKey)) + var reconcilesPrediction = _predictedDocuments.ContainsKey(predictionKey); + if (reconcilesPrediction) { kind = CultNetDatabaseChangeKind.Reconciled; } - AppendMutationLogEntry( - shard, - kind == CultNetDatabaseChangeKind.Reconciled ? CultNetDatabaseChangeKind.Updated : kind, - key, - descriptor.SchemaId, - document, - previous, - new CultNetShardLogEntryMessage - { - ChangeKind = previous == null ? "added" : "updated", - Put = message - }); - PublishUntyped( - descriptor.DocumentType, - kind, - key, - descriptor.SchemaId, - shard, - document, - previous); + void PublishCommittedPut() + { + if (reconcilesPrediction) + _predictedDocuments.Remove(predictionKey); + AppendMutationLogEntry( + shard, + kind == CultNetDatabaseChangeKind.Reconciled ? CultNetDatabaseChangeKind.Updated : kind, + key, + descriptor.SchemaId, + document, + previous, + new CultNetShardLogEntryMessage + { + ChangeKind = previous == null ? "added" : "updated", + Put = message + }); + PublishUntyped( + descriptor.DocumentType, + kind, + key, + descriptor.SchemaId, + shard, + document, + previous); + } + if (_ambientTransaction.Value is { } transaction) + transaction.AfterCommit.Add(PublishCommittedPut); + else + PublishCommittedPut(); return document; } @@ -954,6 +1033,7 @@ namespace GameCult.Networking public Task ApplyDeleteAsync(CultNetDocumentDeleteMessage message) { ThrowIfDisposed(); + EnsureAuthoritativeTransaction(); if (message == null) throw new ArgumentNullException(nameof(message)); var key = new CultRecordKey(message.RecordKey); @@ -967,26 +1047,33 @@ namespace GameCult.Networking } _cache.Remove(key); - AppendMutationLogEntry( - shard, - CultNetDatabaseChangeKind.Removed, - key, - descriptor.SchemaId, - document: null, - previousDocument: previous, - wireEntry: new CultNetShardLogEntryMessage - { - ChangeKind = "removed", - Delete = message - }); - PublishUntyped( - descriptor.DocumentType, - CultNetDatabaseChangeKind.Removed, - key, - descriptor.SchemaId, - shard, - document: null, - previousDocument: previous); + void PublishCommittedDelete() + { + AppendMutationLogEntry( + shard, + CultNetDatabaseChangeKind.Removed, + key, + descriptor.SchemaId, + document: null, + previousDocument: previous, + wireEntry: new CultNetShardLogEntryMessage + { + ChangeKind = "removed", + Delete = message + }); + PublishUntyped( + descriptor.DocumentType, + CultNetDatabaseChangeKind.Removed, + key, + descriptor.SchemaId, + shard, + document: null, + previousDocument: previous); + } + if (_ambientTransaction.Value is { } transaction) + transaction.AfterCommit.Add(PublishCommittedDelete); + else + PublishCommittedDelete(); return Task.CompletedTask; } @@ -1699,6 +1786,20 @@ namespace GameCult.Networking }; } + private sealed class CultNetDatabaseTransaction + { + public List AfterCommit { get; } = new(); + } + + private void EnsureAuthoritativeTransaction() + { + if (_requireTransactionsForAuthoritativeWrites && _ambientTransaction.Value == null) + { + throw new InvalidOperationException( + "This CultNet database requires authoritative writes to use ExecuteTransactionAsync."); + } + } + private void ThrowIfDisposed() { if (_disposed) diff --git a/tests/GameCult.Caching.Tests/BackingStoreTests.cs b/tests/GameCult.Caching.Tests/BackingStoreTests.cs index ee833f17..39fe52ca 100644 --- a/tests/GameCult.Caching.Tests/BackingStoreTests.cs +++ b/tests/GameCult.Caching.Tests/BackingStoreTests.cs @@ -1335,6 +1335,146 @@ namespace GameCult.Caching.Tests Throws.TypeOf().With.Message.Contains("changed target schema")); } + [Test] + public async Task CultCache_Transaction_Hides_Staged_Records_Until_Whole_Batch_Is_Durable() + { + var root = Path.Combine(Path.GetTempPath(), $"cultcache-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:first"); + var secondKey = new CultRecordKey("transaction:second"); + var staged = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var observed = 0; + using var subscription = cache.Watch().Subscribe(_ => observed++); + + var transaction = cache.ExecuteTransactionAsync(async () => + { + 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)); + }); + + await staged.Task; + Assert.That(cache.Get(firstKey), Is.Null); + Assert.That(observed, Is.Zero); + + release.SetResult(true); + await transaction; + Assert.That(cache.Get(firstKey)?.Value, Is.EqualTo("one")); + Assert.That(cache.Get(secondKey)?.Value, Is.EqualTo("two")); + Assert.That(observed, Is.EqualTo(2)); + + 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")); + } + finally + { + if (Directory.Exists(root)) Directory.Delete(root, recursive: true); + } + } + + [Test] + public async Task CultCache_Transaction_Abort_Cannot_Leak_Into_Later_Commit() + { + var root = Path.Combine(Path.GetTempPath(), $"cultcache-abort-{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 abortedKey = new CultRecordKey("transaction:aborted"); + var committedKey = new CultRecordKey("transaction: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.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(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")); + } + 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"); + 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(); + }); + + await cache.ExecuteTransactionAsync(async () => + { + await cache.UpsertAsync( + 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")); + } + finally + { + if (Directory.Exists(root)) Directory.Delete(root, recursive: true); + } + } + private const string NamedFixtureCanonicalSchemaJson = "{\"schemaName\":\"tests.named_entry\",\"schemaVersion\":\"tests.named_entry.v1\",\"members\":[{\"slot\":0,\"name\":\"Name\",\"type\":\"System.String\",\"isReference\":false,\"many\":false,\"targetSchemaName\":null,\"indexAlias\":null,\"isName\":true},{\"slot\":1,\"name\":\"Value\",\"type\":\"System.String\",\"isReference\":false,\"many\":false,\"targetSchemaName\":null,\"indexAlias\":null,\"isName\":false}]}"; private const string NamedFixtureSchemaId = "sha256:e7b97801b94190f3159012ede45b0069bb09ebf7920f7432c971bc86a0e08de8"; diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 03b4f76c..28245a3d 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -2701,6 +2701,60 @@ namespace GameCult.Networking.Tests Assert.That(changes[0].Document!.Username, Is.EqualTo("Raw")); } + [Test] + public async Task CultNetDatabase_RequiredTransaction_DefersRawPutPublicationUntilCommit() + { + var cache = new CultCache(); + var registry = new CultNetDocumentRegistry(cache.Registry) + .Register(CultNetDocumentBinding.ForDocument( + cache.Registry, + payloadSerializer: SerializePlayerDataPayload, + payloadDeserializer: DeserializePlayerDataPayload)); + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions + { + DocumentRegistry = registry, + RequireTransactionsForAuthoritativeWrites = true + }); + var key = new CultRecordKey("player:transactional-raw"); + var message = registry.CreateRawDocumentPutMessage( + "put-transactional-raw", + new CultRecordHandle(key), + new PlayerData + { + PlayerId = Guid.NewGuid(), + Email = "transactional@example.test", + PasswordHash = "hash", + Username = "Transactional" + }); + var changes = new List>(); + using var subscription = database.WatchRecord(key) + .Subscribe(change => changes.Add(change)); + + Assert.That( + async () => await database.ApplyPutAsync(message), + Throws.TypeOf()); + + var staged = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var transaction = database.ExecuteTransactionAsync(async () => + { + await database.ApplyPutAsync(message); + staged.SetResult(true); + await release.Task; + }); + await staged.Task; + + Assert.That(cache.Get(key), Is.Null); + Assert.That(changes, Is.Empty); + + release.SetResult(true); + await transaction; + + Assert.That(cache.Get(key)?.Username, Is.EqualTo("Transactional")); + Assert.That(changes, Has.Count.EqualTo(1)); + Assert.That(changes[0].Kind, Is.EqualTo(CultNetDatabaseChangeKind.Added)); + } + [Test] public async Task CultNetDatabase_ApplyRawPut_ResolvesForeignSchemaIdFromPayload() { diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.dll index 1deb8004..7284604e 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.pdb index edd92539..c881cae5 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.pdb differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.dll index 931e42dc..7158570c 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.pdb index 60ce5583..6b4ea8d7 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.pdb differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.dll index bc67fc77..b6116f32 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.pdb index f702015c..4fdfbb00 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.pdb differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.dll index 96f9481b..a5468f11 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.pdb index fc8890e2..2d70b0eb 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.pdb differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.dll index 65f566e9..78476742 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.pdb index ca39b949..09bb1b13 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.pdb differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.WebSockets.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.WebSockets.dll index 7cfbbf13..a01a7704 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.WebSockets.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.WebSockets.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.dll index 40d4ec74..a68c628d 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.pdb index e1aa9f03..e13f706d 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.pdb differ diff --git a/unity/org.gamecult.cultlib/package.json b/unity/org.gamecult.cultlib/package.json index b441d31b..1b939c02 100644 --- a/unity/org.gamecult.cultlib/package.json +++ b/unity/org.gamecult.cultlib/package.json @@ -1,7 +1,7 @@ { "name": "org.gamecult.cultlib", "displayName": "CultLib", - "version": "1.0.53", + "version": "1.0.54", "unity": "2021.3", "description": "CultCache, CultNet, and CultMesh managed runtime assemblies for Unity.", "license": "MIT",