diff --git a/src/GameCult.Mesh/CultMeshClient.cs b/src/GameCult.Mesh/CultMeshClient.cs
index ed712521..2a5a2012 100644
--- a/src/GameCult.Mesh/CultMeshClient.cs
+++ b/src/GameCult.Mesh/CultMeshClient.cs
@@ -24,6 +24,9 @@ namespace GameCult.Mesh
/// Gets or sets optional transport connectors. The CultNet schema connector is used by default.
public IReadOnlyList? Connectors { get; set; }
+
+ /// Gets or sets the deadline before an unanswered subscription intent is replayed.
+ public TimeSpan SubscriptionResponseTimeout { get; set; } = TimeSpan.FromSeconds(2);
}
///
@@ -34,6 +37,7 @@ 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();
@@ -42,6 +46,8 @@ namespace GameCult.Mesh
public CultMeshClient(CultMeshClientOptions options)
{
if (options == null) throw new ArgumentNullException(nameof(options));
+ if (options.SubscriptionResponseTimeout <= TimeSpan.Zero)
+ throw new ArgumentOutOfRangeException(nameof(options.SubscriptionResponseTimeout));
var endpoints = options.RendezvousEndpoints
.Where(endpoint => !string.IsNullOrWhiteSpace(endpoint))
.Distinct(StringComparer.Ordinal)
@@ -65,6 +71,7 @@ namespace GameCult.Mesh
new CultMeshSchemaTransportConnector(clock: options.Sessions.Clock)
},
options.Sessions);
+ _subscriptionResponseTimeout = options.SubscriptionResponseTimeout;
}
/// Connects to a stable endpoint identity using one application protocol.
@@ -249,7 +256,7 @@ namespace GameCult.Mesh
where TDocument : class
{
var session = await ConnectAsync(endpointId, CultMeshProtocols.Documents).ConfigureAwait(false);
- var binding = new RemoteDocumentBinding(session, endpointId, recordKey);
+ var binding = new RemoteDocumentBinding(session, endpointId, recordKey, _subscriptionResponseTimeout);
_documentResources.Add(binding);
try
{
@@ -267,7 +274,7 @@ namespace GameCult.Mesh
where TDocument : class
{
var session = await ConnectAsync(endpointId, CultMeshProtocols.Documents).ConfigureAwait(false);
- var binding = new RemoteCollectionBinding(session, endpointId, includeInitialSnapshot);
+ var binding = new RemoteCollectionBinding(session, endpointId, includeInitialSnapshot, _subscriptionResponseTimeout);
_documentResources.Add(binding);
try
{
@@ -300,6 +307,8 @@ namespace GameCult.Mesh
private readonly string _subscriptionId;
private readonly CultCache _cache;
private readonly CultNetDatabaseSubscriptionClient _subscription;
+ private readonly TimeSpan _responseTimeout;
+ private readonly CancellationTokenSource _lifetime = new();
private readonly TaskCompletionSource _ready = new(TaskCreationOptions.RunContinuationsAsynchronously);
private readonly object _subscribeLock = new();
private Task? _subscribeTask;
@@ -307,9 +316,14 @@ namespace GameCult.Mesh
private IDisposable? _stateWatch;
private bool _disposed;
- public RemoteDocumentBinding(CultMeshSession session, string endpointId, string recordKey)
+ public RemoteDocumentBinding(
+ CultMeshSession session,
+ string endpointId,
+ string recordKey,
+ TimeSpan responseTimeout)
{
_session = session;
+ _responseTimeout = responseTimeout;
_recordKey = recordKey;
_subscriptionId = "cultmesh-document:" + endpointId + ":" + recordKey;
var cacheRegistry = CultMesh.CreateCultCacheDocumentRegistry(typeof(TDocument));
@@ -349,13 +363,28 @@ namespace GameCult.Mesh
{
try
{
- if (_disposed) return;
- await _subscription.SubscribeAsync(
- _subscriptionId,
- recordKeys: new[] { _recordKey },
- schemaIds: new[] { CultDocumentRegistry.Shared.GetRequired().SchemaId })
- .ConfigureAwait(false);
- _ready.TrySetResult(true);
+ while (!_disposed)
+ {
+ using var deadline = CancellationTokenSource.CreateLinkedTokenSource(_lifetime.Token);
+ deadline.CancelAfter(_responseTimeout);
+ try
+ {
+ await _subscription.SubscribeAsync(
+ _subscriptionId,
+ recordKeys: new[] { _recordKey },
+ schemaIds: new[] { CultDocumentRegistry.Shared.GetRequired().SchemaId },
+ cancellationToken: deadline.Token)
+ .ConfigureAwait(false);
+ _ready.TrySetResult(true);
+ return;
+ }
+ catch (OperationCanceledException) when (!_lifetime.IsCancellationRequested)
+ {
+ }
+ }
+ }
+ catch (OperationCanceledException) when (_lifetime.IsCancellationRequested)
+ {
}
catch (ObjectDisposedException) when (_disposed)
{
@@ -366,10 +395,12 @@ namespace GameCult.Mesh
{
if (_disposed) return;
_disposed = true;
+ _lifetime.Cancel();
_stateWatch?.Dispose();
_subscription.Dispose();
_ready.TrySetException(new ObjectDisposedException(GetType().FullName));
_cache.Dispose();
+ _lifetime.Dispose();
}
}
@@ -379,6 +410,8 @@ namespace GameCult.Mesh
private readonly string _subscriptionId;
private readonly CultCache _cache;
private readonly CultNetDatabaseSubscriptionClient _subscription;
+ private readonly TimeSpan _responseTimeout;
+ private readonly CancellationTokenSource _lifetime = new();
private readonly bool _includeInitialSnapshot;
private readonly TaskCompletionSource _ready = new(TaskCreationOptions.RunContinuationsAsynchronously);
private readonly object _subscribeLock = new();
@@ -387,9 +420,14 @@ namespace GameCult.Mesh
private IDisposable? _stateWatch;
private bool _disposed;
- public RemoteCollectionBinding(CultMeshSession session, string endpointId, bool includeInitialSnapshot)
+ public RemoteCollectionBinding(
+ CultMeshSession session,
+ string endpointId,
+ bool includeInitialSnapshot,
+ TimeSpan responseTimeout)
{
_session = session;
+ _responseTimeout = responseTimeout;
_subscriptionId = "cultmesh-collection:" + endpointId + ":" + typeof(TDocument).FullName;
_includeInitialSnapshot = includeInitialSnapshot;
var cacheRegistry = CultMesh.CreateCultCacheDocumentRegistry(typeof(TDocument));
@@ -427,13 +465,28 @@ namespace GameCult.Mesh
{
try
{
- if (_disposed) return;
- await _subscription.SubscribeAsync(
- _subscriptionId,
- schemaIds: new[] { CultDocumentRegistry.Shared.GetRequired().SchemaId },
- includeSnapshot: _includeInitialSnapshot)
- .ConfigureAwait(false);
- _ready.TrySetResult(true);
+ while (!_disposed)
+ {
+ using var deadline = CancellationTokenSource.CreateLinkedTokenSource(_lifetime.Token);
+ deadline.CancelAfter(_responseTimeout);
+ try
+ {
+ await _subscription.SubscribeAsync(
+ _subscriptionId,
+ schemaIds: new[] { CultDocumentRegistry.Shared.GetRequired().SchemaId },
+ includeSnapshot: _includeInitialSnapshot,
+ cancellationToken: deadline.Token)
+ .ConfigureAwait(false);
+ _ready.TrySetResult(true);
+ return;
+ }
+ catch (OperationCanceledException) when (!_lifetime.IsCancellationRequested)
+ {
+ }
+ }
+ }
+ catch (OperationCanceledException) when (_lifetime.IsCancellationRequested)
+ {
}
catch (ObjectDisposedException) when (_disposed)
{
@@ -444,10 +497,12 @@ namespace GameCult.Mesh
{
if (_disposed) return;
_disposed = true;
+ _lifetime.Cancel();
_stateWatch?.Dispose();
_subscription.Dispose();
_ready.TrySetException(new ObjectDisposedException(GetType().FullName));
_cache.Dispose();
+ _lifetime.Dispose();
}
}
diff --git a/src/GameCult.Mesh/docs/reliability-migration-plan.md b/src/GameCult.Mesh/docs/reliability-migration-plan.md
index d6e6c9cd..65e97a9c 100644
--- a/src/GameCult.Mesh/docs/reliability-migration-plan.md
+++ b/src/GameCult.Mesh/docs/reliability-migration-plan.md
@@ -429,7 +429,10 @@ cached snapshots, and online/reconnecting/offline projections. The session
increments the generation only after replacing its physical channel. A binding
may coalesce duplicate subscription attempts within that generation, but a
pending request from an older generation cannot suppress replay. No individual
-request owns the logical handle.
+request owns the logical handle. Within one healthy generation, an unanswered
+subscribe request expires on the client policy deadline and the binding
+replays the same idempotent subscription intent; the caller does not own this
+retry loop.
**Forbidden writers:** bindings and schema clients cannot choose endpoints,
replace channels, reconnect, or declare session state. An abandoned initial
@@ -448,7 +451,9 @@ waiting for readiness, so disposal terminates pending opens.
**Verification layer:** Mesh tests lose the first document and collection
snapshot, fail the physical client, and prove the replacement snapshot opens
the same logical handle exactly once. A disposal test proves a pending open
-cannot survive its owning client. The Aetheria/EveUnity witness proves provider
+cannot survive its owning client. Another test drops a subscription response
+without dropping the physical path and proves the same binding becomes ready
+after bounded replay. The Aetheria/EveUnity witness proves provider
discovery, live SoA state, commands, receipts, and assets through the migrated
session path.
diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs
index b44e28e9..53da60d5 100644
--- a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs
+++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs
@@ -3,6 +3,7 @@ using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
+using System.Threading;
using System.Threading.Tasks;
using GameCult.Caching;
@@ -46,7 +47,8 @@ namespace GameCult.Networking
string subscriptionId,
IEnumerable? recordKeys = null,
IEnumerable? schemaIds = null,
- bool includeSnapshot = true)
+ bool includeSnapshot = true,
+ CancellationToken cancellationToken = default)
{
if (_disposed) throw new ObjectDisposedException(nameof(CultNetDatabaseSubscriptionClient));
if (!_client.Connected) throw new InvalidOperationException("CultNet schema client must be connected before subscribing.");
@@ -57,14 +59,38 @@ namespace GameCult.Networking
if (!_initialSnapshots.TryAdd(messageId, completion))
throw new InvalidOperationException($"Duplicate database subscription message id '{messageId}'.");
- _client.SendCultNet(new CultNetDatabaseSubscribeMessage
+ CancellationTokenRegistration cancellation = default;
+ if (cancellationToken.CanBeCanceled)
{
- MessageId = messageId,
- SubscriptionId = subscriptionId,
- RecordKeys = recordKeys?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray(),
- SchemaIds = schemaIds?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray(),
- IncludeSnapshot = includeSnapshot
- });
+ cancellation = cancellationToken.Register(() =>
+ {
+ if (_initialSnapshots.TryRemove(messageId, out var pending))
+ pending.TrySetCanceled(cancellationToken);
+ });
+ _ = completion.Task.ContinueWith(
+ _ => cancellation.Dispose(),
+ CancellationToken.None,
+ TaskContinuationOptions.ExecuteSynchronously,
+ TaskScheduler.Default);
+ }
+
+ try
+ {
+ _client.SendCultNet(new CultNetDatabaseSubscribeMessage
+ {
+ MessageId = messageId,
+ SubscriptionId = subscriptionId,
+ RecordKeys = recordKeys?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray(),
+ SchemaIds = schemaIds?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray(),
+ IncludeSnapshot = includeSnapshot
+ });
+ }
+ catch
+ {
+ if (_initialSnapshots.TryRemove(messageId, out var pending))
+ pending.TrySetException(new InvalidOperationException("CultNet subscription request could not be sent."));
+ throw;
+ }
return completion.Task;
}
diff --git a/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs b/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs
index d4f38dfc..b7c41dc9 100644
--- a/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs
+++ b/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs
@@ -84,6 +84,7 @@ public sealed class CultMeshClientTests
connector.Clients[0].Fail(new InvalidOperationException("path lost"));
await WaitUntilAsync(() => connector.Clients.Count == 2 && connector.Clients[1].SubscribeCount == 1);
+ await WaitForTextAsync(document, "pilot surface 2");
(await document.LatestAsync()).Text.Should().Be("pilot surface 2");
connector.Clients[0].SubscribeCount.Should().Be(1);
@@ -112,6 +113,25 @@ public sealed class CultMeshClientTests
connector.Clients[1].SubscribeCount.Should().Be(1);
}
+ [Test]
+ public async Task Document_ReplaysUnansweredIntentOnSamePhysicalSession()
+ {
+ var connector = new DocumentConnector { SuppressFirstSubscriptionSnapshot = true };
+ using var mesh = new CultMeshClient(new CultMeshClientOptions
+ {
+ RendezvousEndpoints = new[] { "rudp://odin:3076" },
+ Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
+ Connectors = new[] { connector },
+ SubscriptionResponseTimeout = TimeSpan.FromMilliseconds(20)
+ });
+
+ var document = await mesh.DocumentAsync("aetheria", "surface:pilot");
+
+ (await document.LatestAsync()).Text.Should().Be("pilot surface 1");
+ connector.Clients.Should().ContainSingle();
+ connector.Clients[0].SubscribeCount.Should().Be(2);
+ }
+
[Test]
public async Task Collection_ReplicatesAllDocumentsOfTypedSchema()
{
@@ -232,6 +252,7 @@ public sealed class CultMeshClientTests
{
public List Clients { get; } = new();
public bool SuppressFirstSnapshot { get; set; }
+ public bool SuppressFirstSubscriptionSnapshot { get; set; }
public string ConnectorId => "document-test";
public int SubscribeCount => Clients.Sum(client => client.SubscribeCount);
public bool CanConnect(CultMeshTransportCandidate candidate) => true;
@@ -242,7 +263,9 @@ public sealed class CultMeshClientTests
{
var client = new DocumentClient(
"pilot surface " + (Clients.Count + 1),
- respondToSubscribe: !SuppressFirstSnapshot || Clients.Count > 0);
+ snapshotsToSuppress: SuppressFirstSnapshot && Clients.Count == 0
+ ? int.MaxValue
+ : SuppressFirstSubscriptionSnapshot && Clients.Count == 0 ? 1 : 0);
Clients.Add(client);
return Task.FromResult(client);
}
@@ -254,11 +277,11 @@ public sealed class CultMeshClientTests
private readonly CultNetDocumentRegistry _documents;
private readonly string _text;
private readonly TaskCompletionSource _failure = new(TaskCreationOptions.RunContinuationsAsynchronously);
- private readonly bool _respondToSubscribe;
- public DocumentClient(string text, bool respondToSubscribe = true)
+ private int _snapshotsToSuppress;
+ public DocumentClient(string text, int snapshotsToSuppress = 0)
{
_text = text;
- _respondToSubscribe = respondToSubscribe;
+ _snapshotsToSuppress = snapshotsToSuppress;
var cache = CultMesh.CreateCultCacheDocumentRegistry(typeof(ClientTestDocument));
_documents = CultMesh.CreateCultNetDocumentRegistry(new[] { typeof(ClientTestDocument) }, cache);
}
@@ -270,7 +293,11 @@ public sealed class CultMeshClientTests
{
if (message is not CultNetDatabaseSubscribeMessage request) return;
SubscribeCount++;
- if (!_respondToSubscribe) return;
+ if (_snapshotsToSuppress > 0)
+ {
+ _snapshotsToSuppress--;
+ return;
+ }
var put = _documents.CreateRawDocumentPutMessage(
"test-put",
new CultRecordHandle(new CultRecordKey("surface:pilot")),
@@ -298,6 +325,19 @@ public sealed class CultMeshClientTests
condition().Should().BeTrue();
}
+ private static async Task WaitForTextAsync(
+ CultMeshDocumentHandle document,
+ string expected)
+ {
+ var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(2);
+ while (DateTime.UtcNow < deadline)
+ {
+ if (string.Equals((await document.LatestAsync()).Text, expected, StringComparison.Ordinal)) return;
+ await Task.Delay(1);
+ }
+ (await document.LatestAsync()).Text.Should().Be(expected);
+ }
+
[CultDocument("tests.cultmesh_client_document", "tests.cultmesh_client_document.v1")]
[MessagePackObject]
public sealed class ClientTestDocument