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

Delete CultNetDatabase's transaction wrapper

Browse source 
ExecuteTransactionAsync (both overloads), RequireTransactionsForAuthoritativeWrites,
EnsureAuthoritativeTransaction and the AfterCommit deferral had no consumer
outside CultNetDatabase_RequiredTransaction_DefersRawPutPublicationUntilCommit,
deleted with them. Writes publish immediately, as they already did outside a
transaction. The cache's own transaction machinery stays until Cut 3 replaces
it with the explicit batch commit.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-13 16:35:56 +02:00
commit 88e1c8e157
2 changed files with 5 additions and 140 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -298,11 +298,6 @@ 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>
@ -470,8 +465,6 @@ 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>
@ -484,7 +477,6 @@ 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)
@ -512,49 +504,6 @@ 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
{
@ -854,7 +803,6 @@ 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>();
@ -880,10 +828,7 @@ namespace GameCult.Networking
document,
previous));
}
if (_ambientTransaction.Value is { } transaction)
transaction.AfterCommit.Add(PublishCommittedPut);
else
PublishCommittedPut();
PublishCommittedPut();
return handle;
}
@ -918,7 +863,6 @@ 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);
@ -944,10 +888,7 @@ namespace GameCult.Networking
document: null,
previousDocument: previous));
}
if (_ambientTransaction.Value is { } transaction)
transaction.AfterCommit.Add(PublishCommittedDelete);
else
PublishCommittedDelete();
PublishCommittedDelete();
}
return Task.CompletedTask;
@ -959,7 +900,6 @@ namespace GameCult.Networking
public async Task<object> ApplyPutAsync(CultNetDocumentPutRawMessage message)
{
ThrowIfDisposed();
EnsureAuthoritativeTransaction();
if (message == null) throw new ArgumentNullException(nameof(message));
if (message.Document == null)
{
@ -1005,10 +945,7 @@ namespace GameCult.Networking
document,
previous);
}
if (_ambientTransaction.Value is { } transaction)
transaction.AfterCommit.Add(PublishCommittedPut);
else
PublishCommittedPut();
PublishCommittedPut();
return document;
}
@ -1018,7 +955,6 @@ namespace GameCult.Networking
public Task ApplyDeleteAsync(CultNetDocumentDeleteMessage message)
{
ThrowIfDisposed();
EnsureAuthoritativeTransaction();
if (message == null) throw new ArgumentNullException(nameof(message));
var key = new CultRecordKey(message.RecordKey);
@ -1055,10 +991,7 @@ namespace GameCult.Networking
document: null,
previousDocument: previous);
}
if (_ambientTransaction.Value is { } transaction)
transaction.AfterCommit.Add(PublishCommittedDelete);
else
PublishCommittedDelete();
PublishCommittedDelete();
return Task.CompletedTask;
}
@ -1771,20 +1704,6 @@ 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)

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

@ -2957,60 +2957,6 @@ 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()
{
@ -4171,7 +4117,7 @@ namespace GameCult.Networking.Tests
{
Assert.That(new CultMeshRudpSocketOptions().BindHost, Is.EqualTo("0.0.0.0"));
}
[Test]
public void CultMeshRudpFacade_CreatesAuthorizedClientFromPeerEndpoint()
{