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

Make CultCache transactions atomic to observers

Browse source
This commit is contained in:
Meta Vens 2026-08-19 01:36:23 +02:00
commit e80383c33f
20 changed files with 773 additions and 145 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -253,6 +253,52 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
}
}
/// <inheritdoc />
public override void CommitBatch(
IReadOnlyCollection<CultStoredDocument> upserts,
IReadOnlyCollection<CultStoredDocument> 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;
}
}
}
/// <inheritdoc />
public override void PushAll(bool soft = false)
{

15
src/GameCult.Caching/Contracts/cultcache-persistence-format.md
View file

@ -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:

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

@ -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<object, DocumentHandleBox> _documentHandles = new();
private readonly Subject<object> _changes = new();
private readonly CultCacheSoaStore _soa = new();
private readonly SemaphoreSlim _transactionGate = new(1, 1);
private readonly AsyncLocal<CultCacheTransaction?> _ambientTransaction = new();
private readonly object _stateGate = new();
private ILogger _logger = new NullLogger();
private bool _hasUnflushedMutations;
@ -1348,14 +1352,29 @@ namespace GameCult.Caching
/// <summary>
/// Gets all document instances currently held by the cache.
/// </summary>
public IEnumerable<object> AllEntries => _entries.Values.Select(entry => entry.Document);
public IEnumerable<object> AllEntries
{
get
{
lock (_stateGate)
return VisibleStoredDocuments().Select(entry => entry.Document).ToArray();
}
}
/// <summary>
/// Gets all stored document records currently held by the cache.
/// </summary>
public IEnumerable<CultStoredDocument> AllStoredDocuments =>
_entries.Values.OrderBy(entry => entry.Descriptor.SchemaName, StringComparer.Ordinal)
.ThenBy(entry => entry.Key.Value, StringComparer.Ordinal);
public IEnumerable<CultStoredDocument> AllStoredDocuments
{
get
{
lock (_stateGate)
return VisibleStoredDocuments()
.OrderBy(entry => entry.Descriptor.SchemaName, StringComparer.Ordinal)
.ThenBy(entry => entry.Key.Value, StringComparer.Ordinal)
.ToArray();
}
}
/// <summary>
/// Gets the document registry used by this cache.
@ -1457,15 +1476,22 @@ namespace GameCult.Caching
/// </summary>
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;
}
/// <summary>
/// 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.
/// </summary>
public async Task ExecuteTransactionAsync(Func<Task> 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);
}
/// <summary>Runs a value-producing buffered cache mutation.</summary>
public async Task<T> ExecuteTransactionAsync<T>(Func<Task<T>> 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;
}
/// <summary>
/// Adds or replaces a typed document and returns its record handle.
/// </summary>
public async Task<CultRecordHandle<T>> AddAsync<T>(T document, CultRecordHandle<T>? 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<T>(stored.Key);
if (_ambientTransaction.Value is { } transaction)
{
var staged = CreateStoredDocument(typeof(T), document, handle?.Key);
transaction.Stage(staged);
return new CultRecordHandle<T>(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<T>(stored.Key);
}
finally
{
_transactionGate.Release();
}
}
/// <summary>
@ -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();
}
}
/// <summary>
@ -1572,9 +1689,15 @@ namespace GameCult.Caching
/// </summary>
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;
}
}
/// <summary>
@ -1607,17 +1730,13 @@ namespace GameCult.Caching
/// </summary>
public IEnumerable<CultStoredDocument> GetStoredDocuments<T>() where T : class
{
var type = typeof(T);
if (type.IsSealed)
lock (_stateGate)
{
return _typeMaps.TryGetValue(type, out var exact)
? exact.Values
: Enumerable.Empty<CultStoredDocument>();
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);
}
/// <summary>
@ -1690,9 +1809,21 @@ namespace GameCult.Caching
/// </summary>
public void Remove<T>(CultRecordHandle<T> 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
/// </summary>
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<CultStoredDocument> 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<CultStoredDocument>()
.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<CultStoredDocument>()
.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<string, CultStoredDocument?> _mutations = new(StringComparer.Ordinal);
public IReadOnlyDictionary<string, CultStoredDocument?> 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
/// </summary>
public abstract void Delete(CultStoredDocument entry);
/// <summary>
/// Atomically stages and durably commits a buffered record batch for this store.
/// Implementations must restore their prior staged view when finality fails.
/// </summary>
public virtual void CommitBatch(
IReadOnlyCollection<CultStoredDocument> upserts,
IReadOnlyCollection<CultStoredDocument> 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;
}
}
/// <summary>
/// Persists all current backing store entries.
/// </summary>
public abstract void PushAll(bool soft = false);

239
src/GameCult.Networking/CultNetDatabase.cs
View file

@ -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.
/// </summary>
public ICultNetShardMutationLogStore? MutationLogStore { get; set; }
/// <summary>
/// Gets or sets whether authoritative writes must occur inside ExecuteTransactionAsync.
/// Use this for state owners that forbid record-at-a-time mutation bypasses.
/// </summary>
public bool RequireTransactionsForAuthoritativeWrites { get; set; }
}
/// <summary>
@ -464,6 +470,8 @@ namespace GameCult.Networking
private readonly Dictionary<string, long> _nextLogSequences = new(StringComparer.Ordinal);
private readonly Dictionary<string, long> _appliedShardSequences = new(StringComparer.Ordinal);
private readonly Subject<object> _changes = new();
private readonly AsyncLocal<CultNetDatabaseTransaction?> _ambientTransaction = new();
private readonly bool _requireTransactionsForAuthoritativeWrites;
private bool _disposed;
/// <summary>
@ -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
/// </summary>
public IReadOnlyList<CultNetClientAuthorityScope> ClientAuthorityScopes => _clientAuthorityScopes;
/// <summary>
/// Runs one buffered authoritative database mutation. Cache state, durable state,
/// mutation logs, and database observers advance only after the complete stage succeeds.
/// </summary>
public async Task ExecuteTransactionAsync(Func<Task> 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<Action> 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();
}
/// <summary>Runs one value-producing buffered authoritative database mutation.</summary>
public async Task<T> ExecuteTransactionAsync<T>(Func<Task<T>> 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;
}
/// <summary>Gets whether this database instance can authoritatively replace the document at a key.</summary>
public bool CanWriteAuthoritatively<T>(CultRecordKey key) where T : class
{
@ -817,6 +869,7 @@ namespace GameCult.Networking
public async Task<CultRecordHandle<T>> PutAsync<T>(CultRecordKey key, T document) where T : class
{
ThrowIfDisposed();
EnsureAuthoritativeTransaction();
if (document == null) throw new ArgumentNullException(nameof(document));
var descriptor = _cache.Registry.GetRequired<T>();
@ -825,20 +878,27 @@ namespace GameCult.Networking
var previous = _cache.Get<T>(key);
var handle = await _cache.UpsertAsync(document, new CultRecordHandle<T>(key)).ConfigureAwait(false);
AppendMutationLogEntry(
shard,
previous == null ? CultNetDatabaseChangeKind.Added : CultNetDatabaseChangeKind.Updated,
key,
descriptor.SchemaId,
document,
previous);
Publish(new CultNetDatabaseChange<T>(
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<T>(
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<T>(CultRecordKey key) where T : class
{
ThrowIfDisposed();
EnsureAuthoritativeTransaction();
var descriptor = _cache.Registry.GetRequired<T>();
var shard = ResolveShardInternal(descriptor, key);
EnsurePrimary(shard, descriptor.SchemaId, key);
@ -881,20 +942,27 @@ namespace GameCult.Networking
if (previous != null)
{
_cache.Remove(new CultRecordHandle<T>(key));
AppendMutationLogEntry(
shard,
CultNetDatabaseChangeKind.Removed,
key,
descriptor.SchemaId,
document: null,
previousDocument: previous);
Publish(new CultNetDatabaseChange<T>(
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<T>(
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<object> 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<Action> 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)

140
tests/GameCult.Caching.Tests/BackingStoreTests.cs
View file

@ -1335,6 +1335,146 @@ namespace GameCult.Caching.Tests
Throws.TypeOf<InvalidOperationException>().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<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
var observed = 0;
using var subscription = cache.Watch<NamedTestEntry>().Subscribe(_ => observed++);
var transaction = cache.ExecuteTransactionAsync(async () =>
{
await cache.UpsertAsync(
new NamedTestEntry { Name = "first", Value = "one" },
new CultRecordHandle<NamedTestEntry>(firstKey));
staged.SetResult(true);
await release.Task;
await cache.UpsertAsync(
new NamedTestEntry { Name = "second", Value = "two" },
new CultRecordHandle<NamedTestEntry>(secondKey));
});
await staged.Task;
Assert.That(cache.Get<NamedTestEntry>(firstKey), Is.Null);
Assert.That(observed, Is.Zero);
release.SetResult(true);
await transaction;
Assert.That(cache.Get<NamedTestEntry>(firstKey)?.Value, Is.EqualTo("one"));
Assert.That(cache.Get<NamedTestEntry>(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<NamedTestEntry>(firstKey)?.Value, Is.EqualTo("one"));
Assert.That(reopened.Get<NamedTestEntry>(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<NamedTestEntry>().Subscribe(_ => observed++);
Assert.ThrowsAsync<InvalidOperationException>(async () =>
await cache.ExecuteTransactionAsync(async () =>
{
await cache.UpsertAsync(
new NamedTestEntry { Name = "aborted", Value = "must-not-leak" },
new CultRecordHandle<NamedTestEntry>(abortedKey));
throw new InvalidOperationException("abort probe");
}));
Assert.That(cache.Get<NamedTestEntry>(abortedKey), Is.Null);
Assert.That(observed, Is.Zero);
await cache.ExecuteTransactionAsync(async () =>
{
await cache.UpsertAsync(
new NamedTestEntry { Name = "committed", Value = "survives" },
new CultRecordHandle<NamedTestEntry>(committedKey));
});
Assert.That(observed, Is.EqualTo(1));
using var reopened = new CultCache();
reopened.AddBackingStore(new DirectoryMessagePackBackingStore(manifest));
await reopened.PullAllBackingStoresAsync();
Assert.That(reopened.Get<NamedTestEntry>(abortedKey), Is.Null);
Assert.That(reopened.Get<NamedTestEntry>(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<NamedTestEntry>().Subscribe(change =>
{
if (!change.Key.Equals(firstKey)) return;
cache.ExecuteTransactionAsync(async () =>
{
await cache.UpsertAsync(
new NamedTestEntry { Name = "observer", Value = "separate-commit" },
new CultRecordHandle<NamedTestEntry>(secondKey));
}).GetAwaiter().GetResult();
});
await cache.ExecuteTransactionAsync(async () =>
{
await cache.UpsertAsync(
new NamedTestEntry { Name = "source", Value = "committed-first" },
new CultRecordHandle<NamedTestEntry>(firstKey));
});
Assert.That(cache.Get<NamedTestEntry>(secondKey)?.Value, Is.EqualTo("separate-commit"));
using var reopened = new CultCache();
reopened.AddBackingStore(new DirectoryMessagePackBackingStore(manifest));
await reopened.PullAllBackingStoresAsync();
Assert.That(reopened.Get<NamedTestEntry>(firstKey)?.Value, Is.EqualTo("committed-first"));
Assert.That(reopened.Get<NamedTestEntry>(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";

54
tests/GameCult.Networking.Tests/NetworkingTests.cs
View file

@ -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<PlayerData>(
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<PlayerData>(key),
new PlayerData
{
PlayerId = Guid.NewGuid(),
Email = "transactional@example.test",
PasswordHash = "hash",
Username = "Transactional"
});
var changes = new List<CultNetDatabaseChange<PlayerData>>();
using var subscription = database.WatchRecord<PlayerData>(key)
.Subscribe(change => changes.Add(change));
Assert.That(
async () => await database.ApplyPutAsync(message),
Throws.TypeOf<InvalidOperationException>());
var staged = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
var transaction = database.ExecuteTransactionAsync(async () =>
{
await database.ApplyPutAsync(message);
staged.SetResult(true);
await release.Task;
});
await staged.Task;
Assert.That(cache.Get<PlayerData>(key), Is.Null);
Assert.That(changes, Is.Empty);
release.SetResult(true);
await transaction;
Assert.That(cache.Get<PlayerData>(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()
{

BIN
unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.dll
View file

Binary file not shown.

BIN
unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.pdb
View file

Binary file not shown.

BIN
unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.dll
View file

Binary file not shown.

BIN
unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.pdb
View file

Binary file not shown.

BIN
unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.dll
View file

Binary file not shown.

BIN
unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.pdb
View file

Binary file not shown.

BIN
unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.dll
View file

Binary file not shown.

BIN
unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.pdb
View file

Binary file not shown.

BIN
unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.dll
View file

Binary file not shown.

BIN
unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.pdb
View file

Binary file not shown.

BIN
unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.WebSockets.dll
View file

Binary file not shown.

BIN
unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.dll
View file

Binary file not shown.

BIN
unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.pdb
View file

Binary file not shown.

2
unity/org.gamecult.cultlib/package.json
View file

@ -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",