Keep exact live subscriptions active before publication

Directory.Build.props

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

src/GameCult.Mesh/CultMesh.cs

@@ -2921,11 +2921,12 @@ namespace GameCult.Mesh
                         cancellationToken: cancellationToken)
                     .ConfigureAwait(false);
                 var publications = initial.OfType<CultMeshBodyPublicationDocument>().ToArray();
-                if (publications.Length != 1)
+                if (publications.Length > 1)
                     throw new InvalidOperationException(
-                        $"Live body subscription '{subscription.SubscriptionId}' expected one publication for " +
+                        $"Live body subscription '{subscription.SubscriptionId}' expected at most one publication for " +
                         $"'{subscription.BodyId}' but received {publications.Length}.");
-                liveBody.Initialize(publications[0]);
+                if (publications.Length == 1)
+                    liveBody.Initialize(publications[0]);
                 return liveBody;
             }
             catch

src/GameCult.Mesh/CultMeshBodyDemand.cs

@@ -141,8 +141,7 @@ namespace GameCult.Mesh
         private void OnChanged(CultNetReplicatedDocumentChange change)
         {
             if (!string.Equals(change.SubscriptionId, _subscription.SubscriptionId, StringComparison.Ordinal) ||
-                !string.Equals(change.RecordKey, _subscription.PublicationRecordKey, StringComparison.Ordinal) ||
-                !string.Equals(change.SchemaId, CultMeshBodyPublicationSchemaVersions.Publication, StringComparison.Ordinal))
+                !string.Equals(change.RecordKey, _subscription.PublicationRecordKey, StringComparison.Ordinal))
                 return;
 
             if (string.Equals(change.ChangeKind, "removed", StringComparison.Ordinal))

src/GameCult.Mesh/README.md

@@ -403,14 +403,20 @@ using var input = await subscriptions.SubscribeLiveValueAsync<EveInputCapability
 
 // Open after each descriptor change. CultMesh chooses shared memory, shared-file
 // mapping, or network from observed locality and the resolver's real adapters.
-using var frame = entities.OpenCurrentReadOnly().Lease;
-Present(frame, input.Current);
+if (entities.HasValue && input.HasValue)
+{
+    using var frame = entities.OpenCurrentReadOnly().Lease;
+    Present(frame, input.Current);
+}
 ```
 
 `CultMeshLiveBody` watches only the small latest-publication record and declares
 the logical body demand. It never routes the body through the subscription
 snapshot. `CultNetLiveValue<T>` does the corresponding job for exact scalar or
-small structured state without creating a renderer-owned replica cache.
+small structured state without creating a renderer-owned replica cache. Both
+subscriptions may begin before the provider's first publication: `HasValue`
+remains false while the retained demand stays active, then the first typed value
+arrives through the normal reactive change path.
 
 Snapshots remain bootstrap and recovery machinery. Reactive document changes
 carry small control state; mapped or negotiated network bodies carry hot data.

src/GameCult.Mesh/docs/public-api.md

@@ -525,7 +525,9 @@ and own their unsubscribe lifetime. Hot body state uses
 `CultMesh.SubscribeLiveBodyAsync(...)`: the reactive value is only the current
 body descriptor, while `OpenCurrentReadOnly()` negotiates shared memory,
 shared-file mapping, or network from observed locality and available adapters.
-Body bytes never become subscription snapshot payload.
+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.
 
 ### Client Prediction
 

src/GameCult.Mesh/docs/reliability-migration-plan.md

@@ -926,7 +926,9 @@ subscription lifetime. `CultMeshLiveBody` owns the ephemeral latest-publication
 value and opens the current generation through the fastest valid representation
 the resolver can actually import. Layout and other small state are independent
 exact reactive values, so changing one variable does not redeliver an unrelated
-view document or body.
+view document or body. A subscription may predate frame zero: absence is an
+uninitialized live value, not a failed subscription, and demand remains active
+until the first descriptor arrives.
 
 `CultMesh.SubscribeHotBodyAsync(...)` remains the grouped compatibility path. Its
 `CultMeshHotBodySubscription` keeps the exact view record, the body's stable
@@ -962,7 +964,8 @@ single-record ergonomic path. Its `CultNetLiveValue<T>` owns the exact
 record/schema filter, current value, typed change/removal notifications, and
 unsubscribe lifetime. Consumers no longer need a process-wide change handler,
 an object type switch, or a renderer-local replica cache to model one reactive
-variable.
+variable. The value may start absent and become initialized on its first matching
+change without resubscribing.
 
 **Derived state:** renderer-local presentation and reactive variables are
 derived from live notifications. They are not database records merely because

src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs

@@ -69,7 +69,7 @@ namespace GameCult.Networking
                 throw new ArgumentException("Record key is required.", nameof(recordKey));
             var schemaId = _documents.GetByDocumentType(typeof(TDocument))?.SchemaId ??
                 _cache.Registry.GetRequired<TDocument>().SchemaId;
-            var value = new CultNetLiveValue<TDocument>(this, subscriptionId, recordKey, schemaId);
+            var value = new CultNetLiveValue<TDocument>(this, subscriptionId, recordKey);
             try
             {
                 var initial = await SubscribeAsync(
@@ -80,10 +80,11 @@ namespace GameCult.Networking
                         cancellationToken: cancellationToken)
                     .ConfigureAwait(false);
                 var documents = initial.OfType<TDocument>().ToArray();
-                if (documents.Length != 1)
+                if (documents.Length > 1)
                     throw new InvalidOperationException(
-                        $"Live value subscription '{subscriptionId}' expected one '{schemaId}' record '{recordKey}' but received {documents.Length}.");
-                value.Initialize(documents[0]);
+                        $"Live value subscription '{subscriptionId}' expected at most one '{schemaId}' record '{recordKey}' but received {documents.Length}.");
+                if (documents.Length == 1)
+                    value.Initialize(documents[0]);
                 return value;
             }
             catch
@@ -294,7 +295,6 @@ namespace GameCult.Networking
         private readonly CultNetDatabaseSubscriptionClient _owner;
         private readonly string _subscriptionId;
         private readonly string _recordKey;
-        private readonly string _schemaId;
         private readonly object _gate = new();
         private TDocument? _current;
         private bool _hasValue;
@@ -303,15 +303,13 @@ namespace GameCult.Networking
         internal CultNetLiveValue(
             CultNetDatabaseSubscriptionClient owner,
             string subscriptionId,
-            string recordKey,
-            string schemaId)
+            string recordKey)
         {
             _owner = owner ?? throw new ArgumentNullException(nameof(owner));
             _subscriptionId = string.IsNullOrWhiteSpace(subscriptionId)
                 ? throw new ArgumentException("Subscription id is required.", nameof(subscriptionId))
                 : subscriptionId;
             _recordKey = recordKey;
-            _schemaId = schemaId;
             _owner.Changed += OnChanged;
         }
 
@@ -364,8 +362,7 @@ namespace GameCult.Networking
         private void OnChanged(CultNetReplicatedDocumentChange change)
         {
             if (!string.Equals(change.SubscriptionId, _subscriptionId, StringComparison.Ordinal) ||
-                !string.Equals(change.RecordKey, _recordKey, StringComparison.Ordinal) ||
-                !string.Equals(change.SchemaId, _schemaId, StringComparison.Ordinal))
+                !string.Equals(change.RecordKey, _recordKey, StringComparison.Ordinal))
                 return;
 
             if (string.Equals(change.ChangeKind, "removed", StringComparison.Ordinal))

tests/GameCult.Mesh.Tests/CultMeshBodyPublicationTests.cs

@@ -2,6 +2,9 @@ using System;
 using System.Collections.Generic;
 using System.IO;
 using System.Linq;
+using System.Net;
+using System.Net.Sockets;
+using System.Threading;
 using System.Threading.Tasks;
 using FluentAssertions;
 using GameCult.Caching;
@@ -220,6 +223,71 @@ public sealed class CultMeshBodyPublicationTests
         restored.Should().Equal(bytes);
     }
 
+    [Test]
+    public async Task LiveBody_SubscriptionCanPrecedeTheFirstProviderPublication()
+    {
+        var now = DateTimeOffset.UtcNow;
+        var bytes = new byte[] { 2, 7, 1, 8 };
+        var publication = Publication(bytes, now);
+        var sourceCache = new CultCache();
+        var sourceDocuments = CultMesh.CreateBodyPublicationDocumentRegistry(sourceCache.Registry);
+        var sourceDatabase = new CultNetDatabase(sourceCache, new CultNetDatabaseOptions
+        {
+            DocumentRegistry = sourceDocuments
+        });
+        using var socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
+        socket.Bind(new IPEndPoint(IPAddress.Loopback, 0));
+        using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions
+        {
+            RuntimeId = "live-body-before-publication-server",
+            Socket = socket
+        });
+        using var serverSubscriptions = 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 documents = CultMesh.CreateBodyPublicationDocumentRegistry(targetCache.Registry);
+        var transport = CultNetSchemaClients.CreateRudp("live-body-before-publication-client");
+        using var subscriptions = new CultNetDatabaseSubscriptionClient(transport, targetCache, documents);
+        transport.Connect("127.0.0.1", server.LocalEndPoint.Port);
+        var connectedBy = DateTimeOffset.UtcNow + TimeSpan.FromSeconds(2);
+        while (!transport.Connected && DateTimeOffset.UtcNow < connectedBy)
+            await Task.Delay(5);
+        transport.Connected.Should().BeTrue();
+
+        using var body = await CultMesh.SubscribeLiveBodyAsync(
+            subscriptions,
+            Resolver(bytes, (_, _) => true),
+            new CultMeshLiveBodySubscription("pilot-body", "eve-unity", publication.BodyId));
+
+        body.HasValue.Should().BeFalse(
+            "body demand must remain active while the producer has not published frame zero yet");
+        var changed = new TaskCompletionSource<CultMeshBodyPublicationDocument>(
+            TaskCreationOptions.RunContinuationsAsynchronously);
+        body.Changed += value => changed.TrySetResult(value);
+
+        await sourceDatabase.PutAsync(
+            CultMeshBodyPublicationDocument.CreateLatestRecordKey(publication.BodyId),
+            publication);
+        var completed = await Task.WhenAny(changed.Task, Task.Delay(TimeSpan.FromSeconds(2)));
+
+        cancellation.Cancel();
+        completed.Should().Be(changed.Task);
+        body.HasValue.Should().BeTrue();
+        body.Current.Should().BeEquivalentTo(publication);
+        using var opened = body.OpenCurrentReadOnly(now).Lease;
+        opened.TransportKind.Should().Be(CultMeshBodyTransportKind.SharedFileMapping);
+    }
+
     private CultMeshBodyPublicationDocument Publication(byte[] bytes, DateTimeOffset now)
     {
         var local = new CultMeshMappedBodyPublisher(_root, TimeSpan.FromMinutes(1))

tests/GameCult.Networking.Tests/NetworkingTests.cs

@@ -1699,11 +1699,6 @@ namespace GameCult.Networking.Tests
             var sourceCache = new CultCache();
             var sourceDatabase = new CultNetDatabase(sourceCache);
             const string recordKey = "tests:subscription-client:reactive-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-value-server",
@@ -1732,8 +1727,7 @@ namespace GameCult.Networking.Tests
             using var value = await AwaitWithTimeout(
                 client.SubscribeLiveValueAsync<NetworkSchemaNote>("reactive-note", recordKey),
                 TimeSpan.FromSeconds(2));
-            Assert.That(value.HasValue, Is.True);
-            Assert.That(value.Current.Text, Is.EqualTo("initial"));
+            Assert.That(value.HasValue, Is.False);
             Assert.That(targetCache.Get(new CultRecordKey(recordKey)), Is.Null);
 
             var changed = new TaskCompletionSource<NetworkSchemaNote>(TaskCreationOptions.RunContinuationsAsynchronously);
@@ -1741,14 +1735,26 @@ namespace GameCult.Networking.Tests
             await sourceDatabase.PutAsync(new CultRecordKey(recordKey), new NetworkSchemaNote
             {
                 Schema = "tests.networking_note.v1",
-                Text = "updated"
+                Text = "initial"
             });
             var update = await AwaitWithTimeout(changed.Task, TimeSpan.FromSeconds(2));
 
+            Assert.That(update.Text, Is.EqualTo("initial"));
+            Assert.That(value.Current.Text, Is.EqualTo("initial"));
+            Assert.That(targetCache.Get(new CultRecordKey(recordKey)), Is.Null);
+
+            var updated = new TaskCompletionSource<NetworkSchemaNote>(TaskCreationOptions.RunContinuationsAsynchronously);
+            value.Changed += document => updated.TrySetResult(document);
+            await sourceDatabase.PutAsync(new CultRecordKey(recordKey), new NetworkSchemaNote
+            {
+                Schema = "tests.networking_note.v1",
+                Text = "updated"
+            });
+            update = await AwaitWithTimeout(updated.Task, TimeSpan.FromSeconds(2));
+
             cancellation.Cancel();
             Assert.That(update.Text, Is.EqualTo("updated"));
             Assert.That(value.Current.Text, Is.EqualTo("updated"));
-            Assert.That(targetCache.Get(new CultRecordKey(recordKey)), Is.Null);
         }
 
         [Test]

unity/org.gamecult.cultlib/package.json

@@ -1,7 +1,7 @@
 {
   "name": "org.gamecult.cultlib",
   "displayName": "CultLib",
-  "version": "1.0.28",
+  "version": "1.0.29",
   "unity": "2021.3",
   "description": "CultCache, CultNet, and CultMesh managed runtime assemblies for Unity.",
   "license": "MIT",