From 376c08a155ae4b6ee879b73c2cd4f754fed301d7 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Mon, 17 Aug 2026 20:39:24 +0200 Subject: [PATCH] Make CultMesh client resources explicit --- src/GameCult.Mesh/CultMeshClient.cs | 273 +++++++++++++++--- .../getting-started/02-connect-by-identity.md | 7 +- .../CultMeshClientTests.cs | 84 +++++- 3 files changed, 313 insertions(+), 51 deletions(-) diff --git a/src/GameCult.Mesh/CultMeshClient.cs b/src/GameCult.Mesh/CultMeshClient.cs index 12ccd470..027cc68e 100644 --- a/src/GameCult.Mesh/CultMeshClient.cs +++ b/src/GameCult.Mesh/CultMeshClient.cs @@ -42,6 +42,40 @@ namespace GameCult.Mesh public TimeSpan SubscriptionResponseTimeout { get; set; } = TimeSpan.FromSeconds(2); } + /// Owns one shared live document subscription until disposed. + public sealed class CultMeshDocumentLease : IDisposable where TDocument : class + { + private Action? _release; + + internal CultMeshDocumentLease(CultMeshDocumentHandle handle, Action release) + { + Handle = handle ?? throw new ArgumentNullException(nameof(handle)); + _release = release ?? throw new ArgumentNullException(nameof(release)); + } + + /// Gets the live typed document handle owned by this lease. + public CultMeshDocumentHandle Handle { get; } + + public void Dispose() => Interlocked.Exchange(ref _release, null)?.Invoke(); + } + + /// Owns one shared live collection subscription until disposed. + public sealed class CultMeshCollectionLease : IDisposable where TDocument : class + { + private Action? _release; + + internal CultMeshCollectionLease(CultMeshCollectionHandle handle, Action release) + { + Handle = handle ?? throw new ArgumentNullException(nameof(handle)); + _release = release ?? throw new ArgumentNullException(nameof(release)); + } + + /// Gets the live typed collection handle owned by this lease. + public CultMeshCollectionHandle Handle { get; } + + public void Dispose() => Interlocked.Exchange(ref _release, null)?.Invoke(); + } + /// /// 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>> _documents = new(StringComparer.Ordinal); - private readonly ConcurrentDictionary>> _collections = new(StringComparer.Ordinal); - private readonly ConcurrentBag _documentResources = new(); + private readonly ConcurrentDictionary _documents = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary _collections = new(StringComparer.Ordinal); private bool _disposed; public CultMeshClient(CultMeshClientOptions options) @@ -147,8 +180,8 @@ namespace GameCult.Mesh options); } - /// Opens one live typed document by stable provider identity and record key. - public async Task> DocumentAsync( + /// Leases one shared live typed document by stable provider identity and record key. + public async Task> LeaseDocumentAsync( 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>( - async () => await OpenDocumentAsync(endpointId.Trim(), recordKey.Trim()).ConfigureAwait(false), - LazyThreadSafetyMode.ExecutionAndPublication)); - try + while (true) { - var binding = (RemoteDocumentBinding)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(endpointId.Trim(), recordKey.Trim(), owner).ConfigureAwait(false))); + if (!resource.TryAcquire()) + { + RemoveExact(_documents, key, resource); + continue; + } + try + { + var binding = (RemoteDocumentBinding)await AwaitForCallerAsync(resource.Value, cancellationToken) + .ConfigureAwait(false); + return new CultMeshDocumentLease( + binding.Handle, + () => Release(_documents, key, resource)); + } + catch + { + Release(_documents, key, resource); + throw; + } } } - /// Opens a live collection of one typed schema by stable provider identity. - public Task> CollectionAsync( + /// Leases a shared live collection of one typed schema by stable provider identity. + public Task> LeaseCollectionAsync( string endpointId, CancellationToken cancellationToken) where TDocument : class => - CollectionAsync(endpointId, includeInitialSnapshot: true, cancellationToken); + LeaseCollectionAsync(endpointId, includeInitialSnapshot: true, cancellationToken); - /// Opens a live collection and optionally skips replaying its initial snapshot. - public async Task> CollectionAsync( + /// Leases a live collection and optionally skips replaying its initial snapshot. + public async Task> LeaseCollectionAsync( 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>( - async () => await OpenCollectionAsync(identity, includeInitialSnapshot).ConfigureAwait(false), - LazyThreadSafetyMode.ExecutionAndPublication)); - try + while (true) { - var binding = (RemoteCollectionBinding)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(identity, includeInitialSnapshot, owner).ConfigureAwait(false))); + if (!resource.TryAcquire()) + { + RemoveExact(_collections, key, resource); + continue; + } + try + { + var binding = (RemoteCollectionBinding)await AwaitForCallerAsync(resource.Value, cancellationToken) + .ConfigureAwait(false); + return new CultMeshCollectionLease( + binding.Handle, + () => Release(_collections, key, resource)); + } + catch + { + Release(_collections, key, resource); + throw; + } } } + /// + /// 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. + /// + public async Task SubmitDocumentAsync( + 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(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); + } + + /// Gets the number of currently leased shared document subscriptions. + public int ActiveDocumentResourceCount => _documents.Count; + + /// Gets the number of currently leased shared collection subscriptions. + public int ActiveCollectionResourceCount => _collections.Count; + /// Reads one typed record once through the reusable document session. public async Task ReadAsync( 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 OpenDocumentAsync(string endpointId, string recordKey) + private async Task OpenDocumentAsync( + string endpointId, + string recordKey, + SharedResource owner) where TDocument : class { var session = await ConnectAsync(endpointId, CultMeshProtocols.Documents).ConfigureAwait(false); var binding = new RemoteDocumentBinding(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 OpenCollectionAsync(string endpointId, bool includeInitialSnapshot) + private async Task OpenCollectionAsync( + string endpointId, + bool includeInitialSnapshot, + SharedResource owner) where TDocument : class { var session = await ConnectAsync(endpointId, CultMeshProtocols.Documents).ConfigureAwait(false); var binding = new RemoteCollectionBinding(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 resources, + string key, + SharedResource resource) + { + if (!resource.Release()) return; + RemoveExact(resources, key, resource); + resource.Dispose(); + } + + private static void RemoveExact( + ConcurrentDictionary resources, + string key, + SharedResource resource) + { + ((ICollection>)resources) + .Remove(new KeyValuePair(key, resource)); + } + + private sealed class SharedResource : IDisposable + { + private readonly object _gate = new(); + private readonly Lazy> _value; + private IDisposable? _attached; + private int _leases; + private bool _retired; + + public SharedResource(Func> open) + { + if (open == null) throw new ArgumentNullException(nameof(open)); + _value = new Lazy>(() => open(this), LazyThreadSafetyMode.ExecutionAndPublication); + } + + public Task 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 : IDisposable where TDocument : class { private readonly CultMeshSession _session; diff --git a/src/GameCult.Mesh/docs/getting-started/02-connect-by-identity.md b/src/GameCult.Mesh/docs/getting-started/02-connect-by-identity.md index 9301b49f..dd7a1abd 100644 --- a/src/GameCult.Mesh/docs/getting-started/02-connect-by-identity.md +++ b/src/GameCult.Mesh/docs/getting-started/02-connect-by-identity.md @@ -12,15 +12,18 @@ using var mesh = new CultMeshClient(new CultMeshClientOptions RendezvousEndpoints = new[] { "rudp://odin.gamecult.net:3076" } }); -var pilotSurface = await mesh.DocumentAsync( +using var pilotSurfaceLease = await mesh.LeaseDocumentAsync( "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. diff --git a/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs b/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs index 1596b93a..bf5de6a7 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs @@ -62,8 +62,10 @@ public sealed class CultMeshClientTests Connectors = new[] { connector } }); - var first = await mesh.DocumentAsync("aetheria", "surface:pilot"); - var second = await mesh.DocumentAsync("aetheria", "surface:pilot"); + using var firstLease = await mesh.LeaseDocumentAsync("aetheria", "surface:pilot"); + using var secondLease = await mesh.LeaseDocumentAsync("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("aetheria", "surface:pilot"); + using var lease = await mesh.LeaseDocumentAsync("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("aetheria", "surface:pilot"); + var opening = mesh.LeaseDocumentAsync("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("aetheria", "surface:pilot"); + using var lease = await mesh.LeaseDocumentAsync("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("aetheria"); + using var lease = await mesh.LeaseCollectionAsync("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("aetheria"); + var opening = mesh.LeaseCollectionAsync("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("aetheria", "surface:pilot"); + var second = await mesh.LeaseDocumentAsync("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("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("aetheria", "surface:pilot"); + var opening = mesh.LeaseDocumentAsync("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 BackgroundFailure => _failure.Task; public int SubscribeCount { get; private set; } + public List Submitted { get; } = new(); public void Connect(string host, int port) { } public void SendCultNet(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)