diff --git a/Directory.Build.props b/Directory.Build.props index e06af334..e8bad6c0 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -14,7 +14,7 @@ MIT © Meta Vens and contributors true - 1.0.24 + 1.0.25 true diff --git a/src/GameCult.Mesh/CultMesh.cs b/src/GameCult.Mesh/CultMesh.cs index d19063ef..34b31474 100644 --- a/src/GameCult.Mesh/CultMesh.cs +++ b/src/GameCult.Mesh/CultMesh.cs @@ -2859,7 +2859,8 @@ namespace GameCult.Mesh public static Task> SubscribeHotBodyAsync( CultNetDatabaseSubscriptionClient subscriptions, IEnumerable supportedBodyTransports, - CultMeshHotBodySubscription subscription) + CultMeshHotBodySubscription subscription, + CultNetDatabaseSubscriptionDeliveryMode deliveryMode = CultNetDatabaseSubscriptionDeliveryMode.ReplicateToCache) { if (subscriptions == null) throw new ArgumentNullException(nameof(subscriptions)); if (supportedBodyTransports == null) throw new ArgumentNullException(nameof(supportedBodyTransports)); @@ -2873,17 +2874,19 @@ namespace GameCult.Mesh schemaIds: subscription.SchemaIds, consumerRuntimeId: subscription.ConsumerRuntimeId, bodyIds: new[] { subscription.BodyId }, - supportedBodyTransports: transports.Select(value => value.ToString())); + supportedBodyTransports: transports.Select(value => value.ToString()), + deliveryMode: deliveryMode); } /// Opens an exact hot-body subscription using a resolver's readable planes. public static Task> SubscribeHotBodyAsync( CultNetDatabaseSubscriptionClient subscriptions, CultMeshBodyPublicationResolver bodyResolver, - CultMeshHotBodySubscription subscription) + CultMeshHotBodySubscription subscription, + CultNetDatabaseSubscriptionDeliveryMode deliveryMode = CultNetDatabaseSubscriptionDeliveryMode.ReplicateToCache) { if (bodyResolver == null) throw new ArgumentNullException(nameof(bodyResolver)); - return SubscribeHotBodyAsync(subscriptions, bodyResolver.SupportedTransports, subscription); + return SubscribeHotBodyAsync(subscriptions, bodyResolver.SupportedTransports, subscription, deliveryMode); } /// Resolves a CultMesh URI or RUDP endpoint into a concrete RUDP endpoint. diff --git a/src/GameCult.Mesh/docs/reliability-migration-plan.md b/src/GameCult.Mesh/docs/reliability-migration-plan.md index 8b183dc6..650ab7fd 100644 --- a/src/GameCult.Mesh/docs/reliability-migration-plan.md +++ b/src/GameCult.Mesh/docs/reliability-migration-plan.md @@ -926,6 +926,37 @@ can actually open, ordered shared-memory first; the server combines those capabilities with observed peer locality. Applications do not hand-assemble transport-name arrays or request one snapshot per frame. +### Reactive document delivery body map + +**Owner:** the subscription declaration owns whether a selected typed document +is durable replica state or an ephemeral live value. The provider still owns +the value; the transport and renderer own neither its persistence nor meaning. + +**Inputs:** exact record/schema filters and an explicit +`CultNetDatabaseSubscriptionDeliveryMode`. `ReplicateToCache` materializes +bootstrap and durable state before notification. `Live` validates and decodes +the same typed payload, then notifies the subscriber without a replica write. + +**Outputs:** both modes produce the same typed initial values and change event. +Only durable replication mutates the consumer's CultCache. + +**Derived state:** renderer-local presentation and reactive variables are +derived from live notifications. They are not database records merely because +CultNet carried them. + +**Forbidden writers:** renderers and products cannot add polling, keepalive +writes, or broad snapshots to repair callback latency. Ephemeral receipts, +input facts, and frame-layout notifications cannot silently become disk-backed +replica history. + +**Shared paths:** exact subscription filters, wire schema validation, +deserialization, unsubscribe, disconnect cleanup, and body-demand negotiation +are identical in both modes. Only the final materialization decision differs. + +**Cut line:** `CultNetDatabaseSubscriptionClient` no longer requires every +received document to complete a CultCache write before the consumer may see it. +Durable documents retain the existing replica path; live values bypass it. + **Cut line:** local frame publication writes directly into a mapped write lease and commits its descriptor. The compatibility byte-array publisher is copy-in only. The current remote frame representation may still use the verified CDN diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs index d0a545c0..4c7770ef 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs @@ -9,6 +9,15 @@ using GameCult.Caching; namespace GameCult.Networking { + /// Controls whether subscription payloads become replica state or remain ephemeral typed events. + public enum CultNetDatabaseSubscriptionDeliveryMode + { + /// Decode each payload and write it into the local CultCache before notifying observers. + ReplicateToCache, + /// Decode each payload and notify observers without writing replica state. + Live + } + /// /// Replicates selected remote database records into a local typed CultCache over one retained session. /// @@ -17,7 +26,9 @@ namespace GameCult.Networking private readonly ICultNetSchemaClient _client; private readonly CultCache _cache; private readonly CultNetDocumentRegistry _documents; - private readonly ConcurrentDictionary>> _initialSnapshots = new(); + private readonly ConcurrentDictionary _initialSnapshots = new(); + private readonly ConcurrentDictionary _deliveryModes = + new(StringComparer.Ordinal); private readonly object _queueLock = new(); private Task _queue = Task.CompletedTask; private bool _disposed; @@ -55,7 +66,8 @@ namespace GameCult.Networking CancellationToken cancellationToken = default, string? consumerRuntimeId = null, IEnumerable? bodyIds = null, - IEnumerable? supportedBodyTransports = null) + IEnumerable? supportedBodyTransports = null, + CultNetDatabaseSubscriptionDeliveryMode deliveryMode = CultNetDatabaseSubscriptionDeliveryMode.ReplicateToCache) { if (_disposed) throw new ObjectDisposedException(nameof(CultNetDatabaseSubscriptionClient)); if (!_client.Connected) throw new InvalidOperationException("CultNet schema client must be connected before subscribing."); @@ -63,7 +75,7 @@ namespace GameCult.Networking var messageId = Guid.NewGuid().ToString("N"); var completion = new TaskCompletionSource>(TaskCreationOptions.RunContinuationsAsynchronously); - if (!_initialSnapshots.TryAdd(messageId, completion)) + if (!_initialSnapshots.TryAdd(messageId, new PendingSubscription(subscriptionId, deliveryMode, completion))) throw new InvalidOperationException($"Duplicate database subscription message id '{messageId}'."); CancellationTokenRegistration cancellation = default; @@ -72,7 +84,7 @@ namespace GameCult.Networking cancellation = cancellationToken.Register(() => { if (_initialSnapshots.TryRemove(messageId, out var pending)) - pending.TrySetCanceled(cancellationToken); + pending.Completion.TrySetCanceled(cancellationToken); }); _ = completion.Task.ContinueWith( _ => cancellation.Dispose(), @@ -101,7 +113,7 @@ namespace GameCult.Networking catch { if (_initialSnapshots.TryRemove(messageId, out var pending)) - pending.TrySetException(new InvalidOperationException("CultNet subscription request could not be sent.")); + pending.Completion.TrySetException(new InvalidOperationException("CultNet subscription request could not be sent.")); throw; } return completion.Task; @@ -112,6 +124,7 @@ namespace GameCult.Networking { if (_disposed) return; if (string.IsNullOrWhiteSpace(subscriptionId)) throw new ArgumentException("Subscription id is required.", nameof(subscriptionId)); + _deliveryModes.TryRemove(subscriptionId, out _); _client.SendCultNet(new CultNetDatabaseUnsubscribeMessage { MessageId = Guid.NewGuid().ToString("N"), @@ -123,9 +136,10 @@ namespace GameCult.Networking { if (_disposed) return; _disposed = true; - foreach (var completion in _initialSnapshots.Values) - completion.TrySetException(new ObjectDisposedException(nameof(CultNetDatabaseSubscriptionClient))); + foreach (var pending in _initialSnapshots.Values) + pending.Completion.TrySetException(new ObjectDisposedException(nameof(CultNetDatabaseSubscriptionClient))); _initialSnapshots.Clear(); + _deliveryModes.Clear(); _client.Dispose(); } @@ -133,15 +147,24 @@ namespace GameCult.Networking { Enqueue(async () => { - if (!_initialSnapshots.TryRemove(message.MessageId, out var completion)) return; + if (!_initialSnapshots.TryRemove(message.MessageId, out var pending)) return; try { - var applied = await _documents.ApplyRawSnapshotResponseAsync(_cache, message).ConfigureAwait(false); - completion.TrySetResult(applied); + IReadOnlyList applied; + if (pending.DeliveryMode == CultNetDatabaseSubscriptionDeliveryMode.Live) + { + applied = message.Documents.Select(_documents.DeserializeRawDocument).ToArray(); + } + else + { + applied = await _documents.ApplyRawSnapshotResponseAsync(_cache, message).ConfigureAwait(false); + } + _deliveryModes[pending.SubscriptionId] = pending.DeliveryMode; + pending.Completion.TrySetResult(applied); } catch (Exception error) { - completion.TrySetException(error); + pending.Completion.TrySetException(error); } }); } @@ -150,9 +173,13 @@ namespace GameCult.Networking { Enqueue(async () => { + var deliveryMode = _deliveryModes.TryGetValue(message.SubscriptionId, out var configured) + ? configured + : CultNetDatabaseSubscriptionDeliveryMode.ReplicateToCache; if (string.Equals(message.ChangeKind, "removed", StringComparison.Ordinal)) { - await DeleteAsync(message).ConfigureAwait(false); + if (deliveryMode == CultNetDatabaseSubscriptionDeliveryMode.ReplicateToCache) + await DeleteAsync(message).ConfigureAwait(false); Changed?.Invoke(new CultNetReplicatedDocumentChange( message.SubscriptionId, "removed", @@ -163,13 +190,15 @@ namespace GameCult.Networking } if (message.Document == null) return; - var document = await _documents.ApplyRawDocumentPutMessageAsync( - _cache, - new CultNetDocumentPutRawMessage - { - MessageId = message.MessageId, - Document = message.Document - }).ConfigureAwait(false); + var document = deliveryMode == CultNetDatabaseSubscriptionDeliveryMode.Live + ? _documents.DeserializeRawDocument(message.Document) + : await _documents.ApplyRawDocumentPutMessageAsync( + _cache, + new CultNetDocumentPutRawMessage + { + MessageId = message.MessageId, + Document = message.Document + }).ConfigureAwait(false); Changed?.Invoke(new CultNetReplicatedDocumentChange( message.SubscriptionId, message.ChangeKind, @@ -202,6 +231,23 @@ namespace GameCult.Networking TaskScheduler.Default).Unwrap(); } } + + private sealed class PendingSubscription + { + public PendingSubscription( + string subscriptionId, + CultNetDatabaseSubscriptionDeliveryMode deliveryMode, + TaskCompletionSource> completion) + { + SubscriptionId = subscriptionId; + DeliveryMode = deliveryMode; + Completion = completion; + } + + public string SubscriptionId { get; } + public CultNetDatabaseSubscriptionDeliveryMode DeliveryMode { get; } + public TaskCompletionSource> Completion { get; } + } } /// Describes one typed document change after local replication. diff --git a/src/GameCult.Networking/CultNetDocumentRegistry.cs b/src/GameCult.Networking/CultNetDocumentRegistry.cs index 9d176556..8017ad2f 100644 --- a/src/GameCult.Networking/CultNetDocumentRegistry.cs +++ b/src/GameCult.Networking/CultNetDocumentRegistry.cs @@ -352,17 +352,8 @@ namespace GameCult.Networking throw new ArgumentException("CultNet raw document message is missing its document payload.", nameof(message)); } - ValidateRawDocumentRecord(message.Document); - var binding = GetBySchemaId(message.Document.SchemaId); - var descriptor = binding != null - ? _documents.GetRequired(binding.DocumentType) - : ResolveDescriptorForRawDocument(message.Document); - binding ??= new CultNetDocumentBinding( - descriptor.DocumentType, - descriptor.SchemaId, - value => CultDocumentMessagePackSerialization.SerializeUntyped(value, value.GetType()), - payload => CultDocumentMessagePackSerialization.DeserializeUntyped(descriptor.DocumentType, payload)); - var document = binding.PayloadDeserializer(message.Document.Payload); + var document = DeserializeRawDocument(message.Document); + var descriptor = ResolveDescriptorForRawDocument(message.Document); var addMethod = typeof(CultCache).GetMethod(nameof(CultCache.AddAsync))! .MakeGenericMethod(descriptor.DocumentType); @@ -373,6 +364,26 @@ namespace GameCult.Networking return document; } + /// + /// Decodes one validated raw document without materializing it in a CultCache replica. + /// Use this for ephemeral reactive subscriptions whose callback is the consumer. + /// + public object DeserializeRawDocument(CultNetRawDocumentRecord document) + { + if (document == null) throw new ArgumentNullException(nameof(document)); + ValidateRawDocumentRecord(document); + var binding = GetBySchemaId(document.SchemaId); + var descriptor = binding != null + ? _documents.GetRequired(binding.DocumentType) + : ResolveDescriptorForRawDocument(document); + binding ??= new CultNetDocumentBinding( + descriptor.DocumentType, + descriptor.SchemaId, + value => CultDocumentMessagePackSerialization.SerializeUntyped(value, value.GetType()), + payload => CultDocumentMessagePackSerialization.DeserializeUntyped(descriptor.DocumentType, payload)); + return binding.PayloadDeserializer(document.Payload); + } + /// /// Applies a raw document put message and returns the typed document. /// diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index e3da6029..98e8438c 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -1634,6 +1634,65 @@ namespace GameCult.Networking.Tests Assert.That(((NetworkSchemaNote)targetCache.Get(new CultRecordKey(recordKey))!).Text, Is.EqualTo("live")); } + [Test] + public async Task DatabaseSubscriptionClient_LiveDeliveryDecodesWithoutWritingReplicaState() + { + var sourceCache = new CultCache(); + var sourceDatabase = new CultNetDatabase(sourceCache); + const string recordKey = "tests:subscription-client:live-note"; + await sourceDatabase.PutAsync(new CultRecordKey(recordKey), new NetworkSchemaNote + { + Schema = "tests.networking_note.v1", + Text = "initial" + }); + using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions + { + RuntimeId = "database-live-subscription-server", + Socket = BindUdpSocket() + }); + using var subscriptions = new CultNetDatabaseSubscriptionServer(server, sourceDatabase); + 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-live-subscription-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)); + + var initial = await AwaitWithTimeout( + client.SubscribeAsync( + "live-notes", + recordKeys: new[] { recordKey }, + deliveryMode: CultNetDatabaseSubscriptionDeliveryMode.Live), + TimeSpan.FromSeconds(2)); + Assert.That(initial.OfType().Single().Text, Is.EqualTo("initial")); + Assert.That(targetCache.Get(new CultRecordKey(recordKey)), Is.Null); + + var changed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + client.Changed += change => changed.TrySetResult(change); + await sourceDatabase.PutAsync(new CultRecordKey(recordKey), new NetworkSchemaNote + { + Schema = "tests.networking_note.v1", + Text = "live" + }); + var update = await AwaitWithTimeout(changed.Task, TimeSpan.FromSeconds(2)); + + cancellation.Cancel(); + Assert.That(((NetworkSchemaNote)update.Document!).Text, Is.EqualTo("live")); + 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 79bf48c5..9a808a27 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 17d72db2..e26bd528 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 a06016f8..ae6f1f1b 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 cdeae45c..225e7e75 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 2a2596b7..32b160b4 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 dcf27116..eeed669e 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 adb73216..2af790f0 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 9fbfc4f4..76c376bf 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 7ff275e1..54615bd3 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 3d6c1c90..108cad92 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 40cd7c06..8b570e95 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.24", + "version": "1.0.25", "unity": "2021.3", "description": "CultCache, CultNet, and CultMesh managed runtime assemblies for Unity.", "license": "MIT",