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

Make CultMesh client resources explicit

Browse source
This commit is contained in:
Meta Vens 2026-08-17 20:39:24 +02:00
commit 376c08a155
3 changed files with 313 additions and 51 deletions
Download patch file Download diff file Expand all files Collapse all files

273
src/GameCult.Mesh/CultMeshClient.cs
View file

@ -42,6 +42,40 @@ namespace GameCult.Mesh
public TimeSpan SubscriptionResponseTimeout { get; set; } = TimeSpan.FromSeconds(2);
}
/// <summary>Owns one shared live document subscription until disposed.</summary>
public sealed class CultMeshDocumentLease<TDocument> : IDisposable where TDocument : class
{
private Action? _release;
internal CultMeshDocumentLease(CultMeshDocumentHandle<TDocument> handle, Action release)
{
Handle = handle ?? throw new ArgumentNullException(nameof(handle));
_release = release ?? throw new ArgumentNullException(nameof(release));
}
/// <summary>Gets the live typed document handle owned by this lease.</summary>
public CultMeshDocumentHandle<TDocument> Handle { get; }
public void Dispose() => Interlocked.Exchange(ref _release, null)?.Invoke();
}
/// <summary>Owns one shared live collection subscription until disposed.</summary>
public sealed class CultMeshCollectionLease<TDocument> : IDisposable where TDocument : class
{
private Action? _release;
internal CultMeshCollectionLease(CultMeshCollectionHandle<TDocument> handle, Action release)
{
Handle = handle ?? throw new ArgumentNullException(nameof(handle));
_release = release ?? throw new ArgumentNullException(nameof(release));
}
/// <summary>Gets the live typed collection handle owned by this lease.</summary>
public CultMeshCollectionHandle<TDocument> Handle { get; }
public void Dispose() => Interlocked.Exchange(ref _release, null)?.Invoke();
}
/// <summary>
/// Owns discovery and reusable sessions for one application lifetime.
/// Applications address stable identities; CultMesh owns physical routes and reconnection.
@ -51,9 +85,8 @@ namespace GameCult.Mesh
private readonly CultMeshDiscoveryService _discovery;
private readonly CultMeshSessionManager _sessions;
private readonly TimeSpan _subscriptionResponseTimeout;
private readonly ConcurrentDictionary<string, Lazy<Task<object>>> _documents = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<string, Lazy<Task<object>>> _collections = new(StringComparer.Ordinal);
private readonly ConcurrentBag<IDisposable> _documentResources = new();
private readonly ConcurrentDictionary<string, SharedResource> _documents = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<string, SharedResource> _collections = new(StringComparer.Ordinal);
private bool _disposed;
public CultMeshClient(CultMeshClientOptions options)
@ -147,8 +180,8 @@ namespace GameCult.Mesh
options);
}
/// <summary>Opens one live typed document by stable provider identity and record key.</summary>
public async Task<CultMeshDocumentHandle<TDocument>> DocumentAsync<TDocument>(
/// <summary>Leases one shared live typed document by stable provider identity and record key.</summary>
public async Task<CultMeshDocumentLease<TDocument>> LeaseDocumentAsync<TDocument>(
string endpointId,
string recordKey,
CancellationToken cancellationToken = default)
@ -158,31 +191,40 @@ namespace GameCult.Mesh
if (string.IsNullOrWhiteSpace(endpointId)) throw new ArgumentException("Endpoint identity is required.", nameof(endpointId));
if (string.IsNullOrWhiteSpace(recordKey)) throw new ArgumentException("Record key is required.", nameof(recordKey));
var key = endpointId.Trim() + "\u001f" + typeof(TDocument).AssemblyQualifiedName + "\u001f" + recordKey.Trim();
var lazy = _documents.GetOrAdd(key, _ => new Lazy<Task<object>>(
async () => await OpenDocumentAsync<TDocument>(endpointId.Trim(), recordKey.Trim()).ConfigureAwait(false),
LazyThreadSafetyMode.ExecutionAndPublication));
try
while (true)
{
var binding = (RemoteDocumentBinding<TDocument>)await AwaitForCallerAsync(lazy.Value, cancellationToken)
.ConfigureAwait(false);
return binding.Handle;
}
catch
{
_documents.TryRemove(key, out _);
throw;
var resource = _documents.GetOrAdd(key, _ => new SharedResource(
async owner => await OpenDocumentAsync<TDocument>(endpointId.Trim(), recordKey.Trim(), owner).ConfigureAwait(false)));
if (!resource.TryAcquire())
{
RemoveExact(_documents, key, resource);
continue;
}
try
{
var binding = (RemoteDocumentBinding<TDocument>)await AwaitForCallerAsync(resource.Value, cancellationToken)
.ConfigureAwait(false);
return new CultMeshDocumentLease<TDocument>(
binding.Handle,
() => Release(_documents, key, resource));
}
catch
{
Release(_documents, key, resource);
throw;
}
}
}
/// <summary>Opens a live collection of one typed schema by stable provider identity.</summary>
public Task<CultMeshCollectionHandle<TDocument>> CollectionAsync<TDocument>(
/// <summary>Leases a shared live collection of one typed schema by stable provider identity.</summary>
public Task<CultMeshCollectionLease<TDocument>> LeaseCollectionAsync<TDocument>(
string endpointId,
CancellationToken cancellationToken)
where TDocument : class =>
CollectionAsync<TDocument>(endpointId, includeInitialSnapshot: true, cancellationToken);
LeaseCollectionAsync<TDocument>(endpointId, includeInitialSnapshot: true, cancellationToken);
/// <summary>Opens a live collection and optionally skips replaying its initial snapshot.</summary>
public async Task<CultMeshCollectionHandle<TDocument>> CollectionAsync<TDocument>(
/// <summary>Leases a live collection and optionally skips replaying its initial snapshot.</summary>
public async Task<CultMeshCollectionLease<TDocument>> LeaseCollectionAsync<TDocument>(
string endpointId,
bool includeInitialSnapshot = true,
CancellationToken cancellationToken = default)
@ -192,22 +234,74 @@ namespace GameCult.Mesh
if (string.IsNullOrWhiteSpace(endpointId)) throw new ArgumentException("Endpoint identity is required.", nameof(endpointId));
var identity = endpointId.Trim();
var key = identity + "\u001f" + typeof(TDocument).AssemblyQualifiedName + "\u001f" + includeInitialSnapshot;
var lazy = _collections.GetOrAdd(key, _ => new Lazy<Task<object>>(
async () => await OpenCollectionAsync<TDocument>(identity, includeInitialSnapshot).ConfigureAwait(false),
LazyThreadSafetyMode.ExecutionAndPublication));
try
while (true)
{
var binding = (RemoteCollectionBinding<TDocument>)await AwaitForCallerAsync(lazy.Value, cancellationToken)
.ConfigureAwait(false);
return binding.Handle;
}
catch
{
_collections.TryRemove(key, out _);
throw;
var resource = _collections.GetOrAdd(key, _ => new SharedResource(
async owner => await OpenCollectionAsync<TDocument>(identity, includeInitialSnapshot, owner).ConfigureAwait(false)));
if (!resource.TryAcquire())
{
RemoveExact(_collections, key, resource);
continue;
}
try
{
var binding = (RemoteCollectionBinding<TDocument>)await AwaitForCallerAsync(resource.Value, cancellationToken)
.ConfigureAwait(false);
return new CultMeshCollectionLease<TDocument>(
binding.Handle,
() => Release(_collections, key, resource));
}
catch
{
Release(_collections, key, resource);
throw;
}
}
}
/// <summary>
/// Submits one typed document to a provider by stable identity. Submission is uncommitted input;
/// only a provider-authored receipt or resulting state can establish acceptance.
/// </summary>
public async Task SubmitDocumentAsync<TDocument>(
string endpointId,
string recordKey,
TDocument document,
string sourceRuntimeId,
string sourceRole,
CancellationToken cancellationToken = default)
where TDocument : class
{
ThrowIfDisposed();
if (string.IsNullOrWhiteSpace(endpointId)) throw new ArgumentException("Endpoint identity is required.", nameof(endpointId));
if (string.IsNullOrWhiteSpace(recordKey)) throw new ArgumentException("Record key is required.", nameof(recordKey));
if (document == null) throw new ArgumentNullException(nameof(document));
if (string.IsNullOrWhiteSpace(sourceRuntimeId)) throw new ArgumentException("Source runtime identity is required.", nameof(sourceRuntimeId));
if (string.IsNullOrWhiteSpace(sourceRole)) throw new ArgumentException("Source role is required.", nameof(sourceRole));
var cacheRegistry = CultMesh.CreateCultCacheDocumentRegistry(typeof(TDocument));
var networkRegistry = CultMesh.CreateCultNetDocumentRegistry(new[] { typeof(TDocument) }, cacheRegistry);
var message = networkRegistry.CreateRawDocumentPutMessage(
"cultmesh-client-submit-" + Guid.NewGuid().ToString("N"),
new CultRecordHandle<TDocument>(new CultRecordKey(recordKey.Trim())),
document,
new CultNetDocumentMessageOptions
{
SourceRuntimeId = sourceRuntimeId.Trim(),
SourceRole = sourceRole.Trim()
});
var session = await ConnectAsync(endpointId.Trim(), CultMeshProtocols.Documents, cancellationToken)
.ConfigureAwait(false);
cancellationToken.ThrowIfCancellationRequested();
session.OpenSchemaClient().SendCultNet(message);
}
/// <summary>Gets the number of currently leased shared document subscriptions.</summary>
public int ActiveDocumentResourceCount => _documents.Count;
/// <summary>Gets the number of currently leased shared collection subscriptions.</summary>
public int ActiveCollectionResourceCount => _collections.Count;
/// <summary>Reads one typed record once through the reusable document session.</summary>
public async Task<TDocument> ReadAsync<TDocument>(
string endpointId,
@ -293,7 +387,7 @@ namespace GameCult.Mesh
{
if (_disposed) return;
_disposed = true;
foreach (var resource in _documentResources) resource.Dispose();
foreach (var resource in _documents.Values.Concat(_collections.Values).Distinct()) resource.Dispose();
_documents.Clear();
_collections.Clear();
_sessions.Dispose();
@ -305,12 +399,15 @@ namespace GameCult.Mesh
if (_disposed) throw new ObjectDisposedException(nameof(CultMeshClient));
}
private async Task<object> OpenDocumentAsync<TDocument>(string endpointId, string recordKey)
private async Task<object> OpenDocumentAsync<TDocument>(
string endpointId,
string recordKey,
SharedResource owner)
where TDocument : class
{
var session = await ConnectAsync(endpointId, CultMeshProtocols.Documents).ConfigureAwait(false);
var binding = new RemoteDocumentBinding<TDocument>(session, endpointId, recordKey, _subscriptionResponseTimeout);
_documentResources.Add(binding);
owner.Attach(binding);
try
{
await binding.StartAsync().ConfigureAwait(false);
@ -318,17 +415,21 @@ namespace GameCult.Mesh
}
catch
{
owner.Detach(binding);
binding.Dispose();
throw;
}
}
private async Task<object> OpenCollectionAsync<TDocument>(string endpointId, bool includeInitialSnapshot)
private async Task<object> OpenCollectionAsync<TDocument>(
string endpointId,
bool includeInitialSnapshot,
SharedResource owner)
where TDocument : class
{
var session = await ConnectAsync(endpointId, CultMeshProtocols.Documents).ConfigureAwait(false);
var binding = new RemoteCollectionBinding<TDocument>(session, endpointId, includeInitialSnapshot, _subscriptionResponseTimeout);
_documentResources.Add(binding);
owner.Attach(binding);
try
{
await binding.StartAsync().ConfigureAwait(false);
@ -336,6 +437,7 @@ namespace GameCult.Mesh
}
catch
{
owner.Detach(binding);
binding.Dispose();
throw;
}
@ -353,6 +455,99 @@ namespace GameCult.Mesh
return await shared.ConfigureAwait(false);
}
private static void Release(
ConcurrentDictionary<string, SharedResource> resources,
string key,
SharedResource resource)
{
if (!resource.Release()) return;
RemoveExact(resources, key, resource);
resource.Dispose();
}
private static void RemoveExact(
ConcurrentDictionary<string, SharedResource> resources,
string key,
SharedResource resource)
{
((ICollection<KeyValuePair<string, SharedResource>>)resources)
.Remove(new KeyValuePair<string, SharedResource>(key, resource));
}
private sealed class SharedResource : IDisposable
{
private readonly object _gate = new();
private readonly Lazy<Task<object>> _value;
private IDisposable? _attached;
private int _leases;
private bool _retired;
public SharedResource(Func<SharedResource, Task<object>> open)
{
if (open == null) throw new ArgumentNullException(nameof(open));
_value = new Lazy<Task<object>>(() => open(this), LazyThreadSafetyMode.ExecutionAndPublication);
}
public Task<object> Value => _value.Value;
public bool TryAcquire()
{
lock (_gate)
{
if (_retired) return false;
_leases++;
return true;
}
}
public bool Release()
{
lock (_gate)
{
if (_leases <= 0) return false;
_leases--;
if (_leases != 0) return false;
_retired = true;
return true;
}
}
public void Attach(IDisposable resource)
{
if (resource == null) throw new ArgumentNullException(nameof(resource));
lock (_gate)
{
if (_retired)
{
resource.Dispose();
throw new ObjectDisposedException(nameof(SharedResource));
}
if (_attached != null) throw new InvalidOperationException("CultMesh shared resource already has a live binding.");
_attached = resource;
}
}
public void Detach(IDisposable resource)
{
lock (_gate)
{
if (ReferenceEquals(_attached, resource)) _attached = null;
}
}
public void Dispose()
{
IDisposable? attached;
lock (_gate)
{
_retired = true;
attached = _attached;
_attached = null;
}
attached?.Dispose();
}
}
private sealed class RemoteDocumentBinding<TDocument> : IDisposable where TDocument : class
{
private readonly CultMeshSession _session;

7
src/GameCult.Mesh/docs/getting-started/02-connect-by-identity.md
View file

@ -12,15 +12,18 @@ using var mesh = new CultMeshClient(new CultMeshClientOptions
RendezvousEndpoints = new[] { "rudp://odin.gamecult.net:3076" }
});
var pilotSurface = await mesh.DocumentAsync<EveSurfaceDocument>(
using var pilotSurfaceLease = await mesh.LeaseDocumentAsync<EveSurfaceDocument>(
"aetheria.daemon",
"eve:surface:aetheria.pilot",
cancellationToken);
var pilotSurface = pilotSurfaceLease.Handle;
using var surfaceWatch = pilotSurface.Watch().Subscribe(Render);
```
Keep `mesh` for the application lifetime. The document handle is stable: when
Keep `mesh` for the application lifetime and dispose each document lease when
the consuming screen or system closes. Leases for the same identity share one
subscription. The document handle is stable while leased: when
Odin advertises a new physical route after a partition, CultMesh migrates the
session, restores the server-side subscription, refreshes the typed snapshot,
and continues the same watch.

84
tests/GameCult.Mesh.Tests/CultMeshClientTests.cs
View file

@ -62,8 +62,10 @@ public sealed class CultMeshClientTests
Connectors = new[] { connector }
});
var first = await mesh.DocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
var second = await mesh.DocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
using var firstLease = await mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
using var secondLease = await mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
var first = firstLease.Handle;
var second = secondLease.Handle;
first.Should().BeSameAs(second);
(await first.LatestAsync()).Text.Should().Be("pilot surface 1");
@ -80,7 +82,8 @@ public sealed class CultMeshClientTests
Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
Connectors = new[] { connector }
});
var document = await mesh.DocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
using var lease = await mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
var document = lease.Handle;
connector.Clients[0].Fail(new InvalidOperationException("path lost"));
await WaitUntilAsync(() => connector.Clients.Count == 2 && connector.Clients[1].SubscribeCount == 1);
@ -102,11 +105,12 @@ public sealed class CultMeshClientTests
Connectors = new[] { connector }
});
var opening = mesh.DocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
var opening = mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
await WaitUntilAsync(() => connector.Clients.Count == 1 && connector.Clients[0].SubscribeCount == 1);
connector.Clients[0].Fail(new InvalidOperationException("path lost before initial snapshot"));
var document = await opening;
using var lease = await opening;
var document = lease.Handle;
(await document.LatestAsync()).Text.Should().Be("pilot surface 2");
connector.Clients.Should().HaveCount(2);
@ -125,7 +129,8 @@ public sealed class CultMeshClientTests
SubscriptionResponseTimeout = TimeSpan.FromMilliseconds(20)
});
var document = await mesh.DocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
using var lease = await mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
var document = lease.Handle;
(await document.LatestAsync()).Text.Should().Be("pilot surface 1");
connector.Clients.Should().ContainSingle();
@ -143,7 +148,8 @@ public sealed class CultMeshClientTests
Connectors = new[] { connector }
});
var collection = await mesh.CollectionAsync<ClientTestDocument>("aetheria");
using var lease = await mesh.LeaseCollectionAsync<ClientTestDocument>("aetheria");
var collection = lease.Handle;
(await collection.LatestAsync()).Should().ContainSingle()
.Which.Text.Should().Be("pilot surface 1");
@ -161,11 +167,12 @@ public sealed class CultMeshClientTests
Connectors = new[] { connector }
});
var opening = mesh.CollectionAsync<ClientTestDocument>("aetheria");
var opening = mesh.LeaseCollectionAsync<ClientTestDocument>("aetheria");
await WaitUntilAsync(() => connector.Clients.Count == 1 && connector.Clients[0].SubscribeCount == 1);
connector.Clients[0].Fail(new InvalidOperationException("path lost before initial collection snapshot"));
var collection = await opening;
using var lease = await opening;
var collection = lease.Handle;
(await collection.LatestAsync()).Should().ContainSingle()
.Which.Text.Should().Be("pilot surface 2");
@ -173,6 +180,57 @@ public sealed class CultMeshClientTests
connector.Clients[1].SubscribeCount.Should().Be(1);
}
[Test]
public async Task Document_ResourceIsSharedAndReleasedWithLastLease()
{
var connector = new DocumentConnector();
using var mesh = new CultMeshClient(new CultMeshClientOptions
{
RendezvousEndpoints = new[] { "rudp://odin:3076" },
Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
Connectors = new[] { connector }
});
var first = await mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
var second = await mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
mesh.ActiveDocumentResourceCount.Should().Be(1);
first.Dispose();
mesh.ActiveDocumentResourceCount.Should().Be(1);
second.Dispose();
mesh.ActiveDocumentResourceCount.Should().Be(0);
using var replacement = await mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
replacement.Handle.Should().NotBeSameAs(first.Handle);
connector.SubscribeCount.Should().Be(2);
}
[Test]
public async Task SubmitDocument_SendsUncommittedInputThroughStableIdentitySession()
{
var connector = new DocumentConnector();
using var mesh = new CultMeshClient(new CultMeshClientOptions
{
RendezvousEndpoints = new[] { "rudp://odin:3076" },
Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
Connectors = new[] { connector }
});
var request = new ClientTestDocument { Id = "command:1", Text = "move" };
await mesh.SubmitDocumentAsync(
"aetheria",
request.Id,
request,
sourceRuntimeId: "pilot-one",
sourceRole: "pilot");
connector.Clients.Should().ContainSingle();
var submitted = connector.Clients[0].Submitted.Should().ContainSingle().Subject;
submitted.Document.RecordKey.Should().Be("command:1");
submitted.Document.SourceRuntimeId.Should().Be("pilot-one");
submitted.Document.SourceRole.Should().Be("pilot");
}
[Test]
public async Task Dispose_TerminatesDocumentWaitingForInitialSnapshot()
{
@ -183,7 +241,7 @@ public sealed class CultMeshClientTests
Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
Connectors = new[] { connector }
});
var opening = mesh.DocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
var opening = mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
await WaitUntilAsync(() => connector.Clients.Count == 1 && connector.Clients[0].SubscribeCount == 1);
mesh.Dispose();
@ -290,9 +348,15 @@ public sealed class CultMeshClientTests
public bool Connected => true;
public Task<Exception> BackgroundFailure => _failure.Task;
public int SubscribeCount { get; private set; }
public List<CultNetDocumentPutRawMessage> Submitted { get; } = new();
public void Connect(string host, int port) { }
public void SendCultNet<T>(T message) where T : ICultNetSchemaMessage
{
if (message is CultNetDocumentPutRawMessage submitted)
{
Submitted.Add(submitted);
return;
}
if (message is not CultNetDatabaseSubscribeMessage request) return;
SubscribeCount++;
if (_snapshotsToSuppress > 0)