Deliver reactive subscriptions without replica writes

Directory.Build.props

@@ -14,7 +14,7 @@
     <PackageLicenseExpression>MIT</PackageLicenseExpression>
     <Copyright>© Meta Vens and contributors</Copyright>
     <PublishRepositoryUrl>true</PublishRepositoryUrl>
-	<Version>1.0.24</Version>
+	<Version>1.0.25</Version>
 
     <!-- Symbols/debugging -->
     <EmbedUntrackedSources>true</EmbedUntrackedSources>

src/GameCult.Mesh/CultMesh.cs

@@ -2859,7 +2859,8 @@ namespace GameCult.Mesh
         public static Task<IReadOnlyList<object>> SubscribeHotBodyAsync(
             CultNetDatabaseSubscriptionClient subscriptions,
             IEnumerable<CultMeshBodyTransportKind> 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);
         }
 
         /// <summary>Opens an exact hot-body subscription using a resolver's readable planes.</summary>
         public static Task<IReadOnlyList<object>> 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);
         }
 
         /// <summary>Resolves a CultMesh URI or RUDP endpoint into a concrete RUDP endpoint.</summary>

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

src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs

@@ -9,6 +9,15 @@ using GameCult.Caching;
 
 namespace GameCult.Networking
 {
+    /// <summary>Controls whether subscription payloads become replica state or remain ephemeral typed events.</summary>
+    public enum CultNetDatabaseSubscriptionDeliveryMode
+    {
+        /// <summary>Decode each payload and write it into the local CultCache before notifying observers.</summary>
+        ReplicateToCache,
+        /// <summary>Decode each payload and notify observers without writing replica state.</summary>
+        Live
+    }
+
     /// <summary>
     /// Replicates selected remote database records into a local typed CultCache over one retained session.
     /// </summary>
@@ -17,7 +26,9 @@ namespace GameCult.Networking
         private readonly ICultNetSchemaClient _client;
         private readonly CultCache _cache;
         private readonly CultNetDocumentRegistry _documents;
-        private readonly ConcurrentDictionary<string, TaskCompletionSource<IReadOnlyList<object>>> _initialSnapshots = new();
+        private readonly ConcurrentDictionary<string, PendingSubscription> _initialSnapshots = new();
+        private readonly ConcurrentDictionary<string, CultNetDatabaseSubscriptionDeliveryMode> _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<string>? bodyIds = null,
-            IEnumerable<string>? supportedBodyTransports = null)
+            IEnumerable<string>? 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<IReadOnlyList<object>>(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<object> 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<IReadOnlyList<object>> completion)
+            {
+                SubscriptionId = subscriptionId;
+                DeliveryMode = deliveryMode;
+                Completion = completion;
+            }
+
+            public string SubscriptionId { get; }
+            public CultNetDatabaseSubscriptionDeliveryMode DeliveryMode { get; }
+            public TaskCompletionSource<IReadOnlyList<object>> Completion { get; }
+        }
     }
 
     /// <summary>Describes one typed document change after local replication.</summary>

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;
         }
 
+        /// <summary>
+        /// Decodes one validated raw document without materializing it in a CultCache replica.
+        /// Use this for ephemeral reactive subscriptions whose callback is the consumer.
+        /// </summary>
+        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);
+        }
+
         /// <summary>
         /// Applies a raw document put message and returns the typed document.
         /// </summary>

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<NetworkSchemaNote>(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<NetworkSchemaNote>().Single().Text, Is.EqualTo("initial"));
+            Assert.That(targetCache.Get(new CultRecordKey(recordKey)), Is.Null);
+
+            var changed = new TaskCompletionSource<CultNetReplicatedDocumentChange>(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()
         {

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",