diff --git a/Directory.Build.props b/Directory.Build.props index 7e4fd1b7..45b0cfa3 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -14,7 +14,7 @@ MIT © Meta Vens and contributors true - 1.0.29 + 1.0.30 true diff --git a/src/GameCult.Mesh/CultMeshBodyDemand.cs b/src/GameCult.Mesh/CultMeshBodyDemand.cs index add14b04..55fa3807 100644 --- a/src/GameCult.Mesh/CultMeshBodyDemand.cs +++ b/src/GameCult.Mesh/CultMeshBodyDemand.cs @@ -317,6 +317,7 @@ namespace GameCult.Mesh public void Observe(CultNetDatabaseSubscriptionDemand demand) { if (demand == null) throw new ArgumentNullException(nameof(demand)); + if (demand.BodyIds.Count == 0) return; var key = demand.ConsumerRuntimeId + "\u001f" + demand.SubscriptionId; lock (_gate) { diff --git a/src/GameCult.Mesh/docs/public-api.md b/src/GameCult.Mesh/docs/public-api.md index 429860e0..5804b59d 100644 --- a/src/GameCult.Mesh/docs/public-api.md +++ b/src/GameCult.Mesh/docs/public-api.md @@ -528,6 +528,11 @@ shared-file mapping, or network from observed locality and available adapters. Body bytes never become subscription snapshot payload. Either subscription may legitimately start with `HasValue == false`; it remains active until the provider publishes the exact record, avoiding a startup retry loop or broad snapshot. +`CultNetDatabaseSubscriptionServer.DemandChanged` exposes the exact requested +record and schema set as well as optional body negotiation. A provider can +therefore materialize a computed record only while it has consumers. The watch +is installed before demand is announced, so the first publication travels as a +normal live change and does not require a bootstrap snapshot. ### Client Prediction diff --git a/src/GameCult.Mesh/docs/reliability-migration-plan.md b/src/GameCult.Mesh/docs/reliability-migration-plan.md index feb025f2..acddd1b9 100644 --- a/src/GameCult.Mesh/docs/reliability-migration-plan.md +++ b/src/GameCult.Mesh/docs/reliability-migration-plan.md @@ -920,6 +920,18 @@ stream demand merely by existing. and server disposal all update the same typed demand projection. Stable peer identity, not per-message wrapper identity, keys subscription lifetime. +Exact reactive document demand carries the requested record and schema set even +when no hot body is involved. Providers may use that demand to materialize +computed documents lazily; the subscription watch exists before activation is +reported, so frame zero is delivered on the live path. Dynamic state is not +forced back through a broad snapshot merely because the record did not exist at +subscribe time. + +The client installs the declared delivery mode before sending the subscription. +A provider may publish synchronously from demand activation, so a live change +can correctly precede the empty bootstrap acknowledgement without briefly +becoming durable replica state. + `CultMesh.SubscribeLiveBodyAsync(...)` is the narrow public construction path. Its `CultMeshLiveBodySubscription` names only the logical body, consumer, and subscription lifetime. `CultMeshLiveBody` owns the ephemeral latest-publication diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs index 51cda8f0..39cc4cf3 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs @@ -116,6 +116,7 @@ namespace GameCult.Networking var completion = new TaskCompletionSource>(TaskCreationOptions.RunContinuationsAsynchronously); if (!_initialSnapshots.TryAdd(messageId, new PendingSubscription(subscriptionId, deliveryMode, completion))) throw new InvalidOperationException($"Duplicate database subscription message id '{messageId}'."); + _deliveryModes[subscriptionId] = deliveryMode; CancellationTokenRegistration cancellation = default; if (cancellationToken.CanBeCanceled) @@ -151,6 +152,7 @@ namespace GameCult.Networking } catch { + _deliveryModes.TryRemove(subscriptionId, out _); if (_initialSnapshots.TryRemove(messageId, out var pending)) pending.Completion.TrySetException(new InvalidOperationException("CultNet subscription request could not be sent.")); throw; @@ -198,7 +200,6 @@ namespace GameCult.Networking { applied = await _documents.ApplyRawSnapshotResponseAsync(_cache, message).ConfigureAwait(false); } - _deliveryModes[pending.SubscriptionId] = pending.DeliveryMode; pending.Completion.TrySetResult(applied); } catch (Exception error) diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs index 75028899..e55895e9 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs @@ -41,7 +41,11 @@ namespace GameCult.Networking _peerLifecycle.PeerDisconnected += HandlePeerDisconnected; } - /// Raised when exact document subscriptions add or remove typed hot-body demand. + /// + /// Raised when an exact subscription is activated or withdrawn. Providers can use + /// the requested records and schemas to materialize reactive state only while a + /// consumer needs it; body publishers additionally use the negotiated body route. + /// public event Action? DemandChanged; /// Detaches handlers and releases all active watches. @@ -142,11 +146,21 @@ namespace GameCult.Networking SubscriptionKey key, bool active) { - if (request.BodyIds is not { Length: > 0 }) return; DemandChanged?.Invoke(new CultNetDatabaseSubscriptionDemand( string.IsNullOrWhiteSpace(request.ConsumerRuntimeId) ? key.Id : request.ConsumerRuntimeId!, key.Id, - request.BodyIds.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray(), + (request.RecordKeys ?? Array.Empty()) + .Where(value => !string.IsNullOrWhiteSpace(value)) + .Distinct(StringComparer.Ordinal) + .ToArray(), + (request.SchemaIds ?? Array.Empty()) + .Where(value => !string.IsNullOrWhiteSpace(value)) + .Distinct(StringComparer.Ordinal) + .ToArray(), + (request.BodyIds ?? Array.Empty()) + .Where(value => !string.IsNullOrWhiteSpace(value)) + .Distinct(StringComparer.Ordinal) + .ToArray(), (request.SupportedBodyTransports ?? Array.Empty()) .Where(value => !string.IsNullOrWhiteSpace(value)) .Distinct(StringComparer.OrdinalIgnoreCase) @@ -284,7 +298,7 @@ namespace GameCult.Networking } } - /// One active or withdrawn consumer demand for logical hot bodies. + /// One active or withdrawn exact state subscription and its optional hot-body route. public sealed class CultNetDatabaseSubscriptionDemand { public CultNetDatabaseSubscriptionDemand( @@ -294,9 +308,32 @@ namespace GameCult.Networking IReadOnlyList supportedBodyTransports, bool sameMachine, bool active) + : this( + consumerRuntimeId, + subscriptionId, + Array.Empty(), + Array.Empty(), + bodyIds, + supportedBodyTransports, + sameMachine, + active) + { + } + + public CultNetDatabaseSubscriptionDemand( + string consumerRuntimeId, + string subscriptionId, + IReadOnlyList recordKeys, + IReadOnlyList schemaIds, + IReadOnlyList bodyIds, + IReadOnlyList supportedBodyTransports, + bool sameMachine, + bool active) { ConsumerRuntimeId = consumerRuntimeId; SubscriptionId = subscriptionId; + RecordKeys = recordKeys; + SchemaIds = schemaIds; BodyIds = bodyIds; SupportedBodyTransports = supportedBodyTransports; SameMachine = sameMachine; @@ -305,6 +342,8 @@ namespace GameCult.Networking public string ConsumerRuntimeId { get; } public string SubscriptionId { get; } + public IReadOnlyList RecordKeys { get; } + public IReadOnlyList SchemaIds { get; } public IReadOnlyList BodyIds { get; } public IReadOnlyList SupportedBodyTransports { get; } public bool SameMachine { get; } diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 0edc9a39..e7f4a8b4 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -1509,6 +1509,8 @@ namespace GameCult.Networking.Tests var observed = await AwaitWithTimeout(active.Task, TimeSpan.FromSeconds(2)); Assert.That(observed.ConsumerRuntimeId, Is.EqualTo("eve-unity")); Assert.That(observed.SubscriptionId, Is.EqualTo("world-body")); + Assert.That(observed.RecordKeys, Is.EqualTo(new[] { "world:entities", "mesh:body:world:latest" })); + Assert.That(observed.SchemaIds, Is.Empty); Assert.That(observed.BodyIds, Is.EqualTo(new[] { "world" })); Assert.That(observed.SupportedBodyTransports, Is.EqualTo(new[] { "SharedMemory", "Network" })); Assert.That(observed.SameMachine, Is.True); @@ -1524,6 +1526,53 @@ namespace GameCult.Networking.Tests Assert.That(removed.Active, Is.False); } + [Test] + public async Task DatabaseSubscriptionServer_ProjectsExactReactiveStateDemandWithoutBodyDemand() + { + var database = new CultNetDatabase(new CultCache()); + using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions + { + RuntimeId = "database-reactive-demand-server", + Socket = BindUdpSocket() + }); + using var subscriptions = new CultNetDatabaseSubscriptionServer(server, database); + var active = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + subscriptions.DemandChanged += demand => + { + if (demand.Active) active.TrySetResult(demand); + }; + using var cancellation = new CancellationTokenSource(); + var serverThread = new Thread(() => + { + while (!cancellation.IsCancellationRequested) + { + _ = server.PollOnceAsync().GetAwaiter().GetResult(); + Thread.Sleep(1); + } + }) { IsBackground = true }; + serverThread.Start(); + + using var client = CultNetSchemaClients.CreateRudp("database-reactive-demand-client"); + client.Connect("127.0.0.1", server.LocalEndPoint.Port); + await WaitUntilAsync(() => client.Connected, TimeSpan.FromSeconds(2)); + client.SendCultNet(new CultNetDatabaseSubscribeMessage + { + MessageId = "subscribe-reactive", + SubscriptionId = "fog-field", + RecordKeys = ["world:field:fog"], + SchemaIds = ["gamecult.fields.splats.v1"], + ConsumerRuntimeId = "eve-unity", + IncludeSnapshot = false + }); + + var observed = await AwaitWithTimeout(active.Task, TimeSpan.FromSeconds(2)); + cancellation.Cancel(); + Assert.That(observed.RecordKeys, Is.EqualTo(new[] { "world:field:fog" })); + Assert.That(observed.SchemaIds, Is.EqualTo(new[] { "gamecult.fields.splats.v1" })); + Assert.That(observed.BodyIds, Is.Empty); + } + [Test] public async Task DatabaseSubscriptionServer_WithdrawsDemandWhenPeerDisconnects() { @@ -1757,6 +1806,56 @@ namespace GameCult.Networking.Tests Assert.That(value.Current.Text, Is.EqualTo("updated")); } + [Test] + public async Task DatabaseSubscriptionClient_ReceivesProviderMaterializedValueFromExactDemand() + { + var sourceDatabase = new CultNetDatabase(new CultCache()); + const string recordKey = "tests:subscription-client:demanded-note"; + using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions + { + RuntimeId = "database-demanded-value-server", + Socket = BindUdpSocket() + }); + using var subscriptions = new CultNetDatabaseSubscriptionServer(server, sourceDatabase); + subscriptions.DemandChanged += demand => + { + if (!demand.Active || !demand.RecordKeys.Contains(recordKey, StringComparer.Ordinal)) + return; + sourceDatabase.PutAsync(new CultRecordKey(recordKey), new NetworkSchemaNote + { + Schema = "tests.networking_note.v1", + Text = "materialized-on-demand" + }).GetAwaiter().GetResult(); + }; + using var cancellation = new CancellationTokenSource(); + var serverThread = new Thread(() => + { + while (!cancellation.IsCancellationRequested) + { + _ = server.PollOnceAsync().GetAwaiter().GetResult(); + Thread.Sleep(1); + } + }) { IsBackground = true }; + serverThread.Start(); + + var targetCache = new CultCache(); + var targetDocuments = new CultNetDocumentRegistry(targetCache.Registry) + .Register(CultNetDocumentBinding.ForDocument(targetCache.Registry)); + var transport = CultNetSchemaClients.CreateRudp("database-demanded-value-client"); + using var client = new CultNetDatabaseSubscriptionClient(transport, targetCache, targetDocuments); + transport.Connect("127.0.0.1", server.LocalEndPoint.Port); + await WaitUntilAsync(() => transport.Connected, TimeSpan.FromSeconds(2)); + + using var value = await AwaitWithTimeout( + client.SubscribeLiveValueAsync("demanded-note", recordKey), + TimeSpan.FromSeconds(2)); + await WaitUntilAsync(() => value.HasValue, TimeSpan.FromSeconds(2)); + + cancellation.Cancel(); + Assert.That(value.Current.Text, Is.EqualTo("materialized-on-demand")); + Assert.That(targetCache.Get(new CultRecordKey(recordKey)), Is.Null); + } + [Test] public async Task DatabaseSubscription_FiltersLiveChangesByWireSchemaBinding() { diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.dll index 7c5433f6..38a637ee 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.pdb index 068b16c2..a1b0d789 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.pdb differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.dll index 679c71b1..335ee453 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.pdb index 4ccae173..0884bc32 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.pdb differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.dll index aa2b1d84..597059b3 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.pdb index 706813e8..1ad4d85d 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.pdb differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.dll index d5aa167e..5b2616c6 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.pdb index 9a992554..c37f0ea6 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.pdb differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.dll index 5c66a8af..babfd9e5 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.pdb index a68c3148..dc343f0d 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.pdb differ diff --git a/unity/org.gamecult.cultlib/package.json b/unity/org.gamecult.cultlib/package.json index bafe2360..f80cda7b 100644 --- a/unity/org.gamecult.cultlib/package.json +++ b/unity/org.gamecult.cultlib/package.json @@ -1,7 +1,7 @@ { "name": "org.gamecult.cultlib", "displayName": "CultLib", - "version": "1.0.29", + "version": "1.0.30", "unity": "2021.3", "description": "CultCache, CultNet, and CultMesh managed runtime assemblies for Unity.", "license": "MIT",