From 61257d5a2072876315ecf08f595f4edf20d0ea26 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Mon, 13 Jul 2026 04:19:54 +0100 Subject: [PATCH] Harden CultMesh subscription readiness --- src/GameCult.Mesh/CultMeshClient.cs | 87 ++++++++++++++++--- .../CultNetDatabaseServer.cs | 8 ++ .../CultNetDatabaseSubscriptionClient.cs | 12 +-- .../CultNetDatabaseSubscriptionServer.cs | 8 ++ 4 files changed, 96 insertions(+), 19 deletions(-) diff --git a/src/GameCult.Mesh/CultMeshClient.cs b/src/GameCult.Mesh/CultMeshClient.cs index 08260e8a..ddc55a38 100644 --- a/src/GameCult.Mesh/CultMeshClient.cs +++ b/src/GameCult.Mesh/CultMeshClient.cs @@ -115,17 +115,25 @@ namespace GameCult.Mesh } /// Opens a live collection of one typed schema by stable provider identity. + public Task> CollectionAsync( + string endpointId, + CancellationToken cancellationToken) + where TDocument : class => + CollectionAsync(endpointId, includeInitialSnapshot: true, cancellationToken); + + /// Opens a live collection and optionally skips replaying its initial snapshot. public async Task> CollectionAsync( string endpointId, + bool includeInitialSnapshot = true, CancellationToken cancellationToken = default) where TDocument : class { ThrowIfDisposed(); if (string.IsNullOrWhiteSpace(endpointId)) throw new ArgumentException("Endpoint identity is required.", nameof(endpointId)); var identity = endpointId.Trim(); - var key = identity + "\u001f" + typeof(TDocument).AssemblyQualifiedName; + var key = identity + "\u001f" + typeof(TDocument).AssemblyQualifiedName + "\u001f" + includeInitialSnapshot; var lazy = _collections.GetOrAdd(key, _ => new Lazy>( - async () => await OpenCollectionAsync(identity).ConfigureAwait(false), + async () => await OpenCollectionAsync(identity, includeInitialSnapshot).ConfigureAwait(false), LazyThreadSafetyMode.ExecutionAndPublication)); try { @@ -157,12 +165,44 @@ namespace GameCult.Mesh $"CultMesh endpoint '{endpointId}' did not publish {typeof(TDocument).FullName} record '{recordKey}'."); } + /// Reads one typed record once with an explicit response deadline. + public async Task ReadAsync( + string endpointId, + string recordKey, + TimeSpan responseTimeout, + CancellationToken cancellationToken = default) + where TDocument : class + { + var documents = await ReadManyCoreAsync( + endpointId, + new[] { recordKey }, + responseTimeout, + cancellationToken) + .ConfigureAwait(false); + return documents.FirstOrDefault() + ?? throw new InvalidOperationException( + $"CultMesh endpoint '{endpointId}' did not publish {typeof(TDocument).FullName} record '{recordKey}'."); + } + /// Reads typed records once in one request through the reusable document session. public async Task> ReadManyAsync( string endpointId, IReadOnlyList recordKeys, CancellationToken cancellationToken = default) where TDocument : class + => await ReadManyCoreAsync( + endpointId, + recordKeys, + TimeSpan.FromSeconds(10), + cancellationToken) + .ConfigureAwait(false); + + private async Task> ReadManyCoreAsync( + string endpointId, + IReadOnlyList recordKeys, + TimeSpan responseTimeout, + CancellationToken cancellationToken) + where TDocument : class { ThrowIfDisposed(); if (string.IsNullOrWhiteSpace(endpointId)) throw new ArgumentException("Endpoint identity is required.", nameof(endpointId)); @@ -177,7 +217,7 @@ namespace GameCult.Mesh new CultMeshSnapshotRequestOptions { ConnectTimeout = TimeSpan.FromSeconds(5), - ResponseTimeout = TimeSpan.FromSeconds(10), + ResponseTimeout = responseTimeout, MessageIdPrefix = "cultmesh-client-read" }, networkRegistry, @@ -223,11 +263,11 @@ namespace GameCult.Mesh } } - private async Task OpenCollectionAsync(string endpointId) + private async Task OpenCollectionAsync(string endpointId, bool includeInitialSnapshot) where TDocument : class { var session = await ConnectAsync(endpointId, CultMeshProtocols.Documents).ConfigureAwait(false); - var binding = new RemoteCollectionBinding(session, endpointId); + var binding = new RemoteCollectionBinding(session, endpointId, includeInitialSnapshot); _documentResources.Add(binding); try { @@ -261,6 +301,8 @@ namespace GameCult.Mesh private readonly CultCache _cache; private readonly CultNetDatabaseSubscriptionClient _subscription; private readonly TaskCompletionSource _ready = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly object _subscribeLock = new(); + private Task? _subscribeTask; private IDisposable? _stateWatch; private bool _disposed; @@ -286,11 +328,20 @@ namespace GameCult.Mesh { _stateWatch = _session.WatchState() .Where(state => state.Status == CultMeshSessionStatus.Online) - .Subscribe(state => { _ = SubscribeAsync(); }); - _ = SubscribeAsync(); + .Subscribe(state => { _ = EnsureSubscribedAsync(); }); + _ = EnsureSubscribedAsync(); await _ready.Task.ConfigureAwait(false); } + private Task EnsureSubscribedAsync() + { + lock (_subscribeLock) + { + if (_subscribeTask is { IsCompleted: false }) return _subscribeTask; + return _subscribeTask = SubscribeAsync(); + } + } + private async Task SubscribeAsync() { try @@ -325,14 +376,18 @@ namespace GameCult.Mesh private readonly string _subscriptionId; private readonly CultCache _cache; private readonly CultNetDatabaseSubscriptionClient _subscription; + private readonly bool _includeInitialSnapshot; private readonly TaskCompletionSource _ready = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly object _subscribeLock = new(); + private Task? _subscribeTask; private IDisposable? _stateWatch; private bool _disposed; - public RemoteCollectionBinding(CultMeshSession session, string endpointId) + public RemoteCollectionBinding(CultMeshSession session, string endpointId, bool includeInitialSnapshot) { _session = session; _subscriptionId = "cultmesh-collection:" + endpointId + ":" + typeof(TDocument).FullName; + _includeInitialSnapshot = includeInitialSnapshot; var cacheRegistry = CultMesh.CreateCultCacheDocumentRegistry(typeof(TDocument)); var networkRegistry = CultMesh.CreateCultNetDocumentRegistry(new[] { typeof(TDocument) }, cacheRegistry); _cache = new CultCache(cacheRegistry); @@ -348,11 +403,20 @@ namespace GameCult.Mesh { _stateWatch = _session.WatchState() .Where(state => state.Status == CultMeshSessionStatus.Online) - .Subscribe(state => { _ = SubscribeAsync(); }); - _ = SubscribeAsync(); + .Subscribe(state => { _ = EnsureSubscribedAsync(); }); + _ = EnsureSubscribedAsync(); await _ready.Task.ConfigureAwait(false); } + private Task EnsureSubscribedAsync() + { + lock (_subscribeLock) + { + if (_subscribeTask is { IsCompleted: false }) return _subscribeTask; + return _subscribeTask = SubscribeAsync(); + } + } + private async Task SubscribeAsync() { try @@ -360,7 +424,8 @@ namespace GameCult.Mesh if (_disposed) return; await _subscription.SubscribeAsync( _subscriptionId, - schemaIds: new[] { CultDocumentRegistry.Shared.GetRequired().SchemaId }) + schemaIds: new[] { CultDocumentRegistry.Shared.GetRequired().SchemaId }, + includeSnapshot: _includeInitialSnapshot) .ConfigureAwait(false); _ready.TrySetResult(true); } diff --git a/src/GameCult.Networking/CultNetDatabaseServer.cs b/src/GameCult.Networking/CultNetDatabaseServer.cs index a136e1fc..c2926da3 100644 --- a/src/GameCult.Networking/CultNetDatabaseServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseServer.cs @@ -303,6 +303,14 @@ namespace GameCult.Networking RecordKeys = message.RecordKeys })); } + else + { + peer.SendCultNet(new CultNetSnapshotResponseRawMessage + { + MessageId = message.MessageId, + Documents = Array.Empty() + }); + } return Task.CompletedTask; } diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs index 6540e9f9..b44e28e9 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs @@ -53,13 +53,9 @@ namespace GameCult.Networking if (string.IsNullOrWhiteSpace(subscriptionId)) throw new ArgumentException("Subscription id is required.", nameof(subscriptionId)); var messageId = Guid.NewGuid().ToString("N"); - TaskCompletionSource>? completion = null; - if (includeSnapshot) - { - completion = new TaskCompletionSource>(TaskCreationOptions.RunContinuationsAsynchronously); - if (!_initialSnapshots.TryAdd(messageId, completion)) - throw new InvalidOperationException($"Duplicate database subscription message id '{messageId}'."); - } + var completion = new TaskCompletionSource>(TaskCreationOptions.RunContinuationsAsynchronously); + if (!_initialSnapshots.TryAdd(messageId, completion)) + throw new InvalidOperationException($"Duplicate database subscription message id '{messageId}'."); _client.SendCultNet(new CultNetDatabaseSubscribeMessage { @@ -69,7 +65,7 @@ namespace GameCult.Networking SchemaIds = schemaIds?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray(), IncludeSnapshot = includeSnapshot }); - return completion?.Task ?? Task.FromResult>(Array.Empty()); + return completion.Task; } /// Stops one live remote subscription. diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs index 785b35aa..f61f7b93 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs @@ -77,6 +77,14 @@ namespace GameCult.Networking RecordKeys = request.RecordKeys })); } + else + { + peer.SendCultNet(new CultNetSnapshotResponseRawMessage + { + MessageId = request.MessageId, + Documents = Array.Empty() + }); + } return Task.CompletedTask; }