diff --git a/Directory.Build.props b/Directory.Build.props index 0817976e..7e4fd1b7 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -14,7 +14,7 @@ MIT © Meta Vens and contributors true - 1.0.28 + 1.0.29 true diff --git a/src/GameCult.Mesh/CultMesh.cs b/src/GameCult.Mesh/CultMesh.cs index 7efa0aa1..fbb08f22 100644 --- a/src/GameCult.Mesh/CultMesh.cs +++ b/src/GameCult.Mesh/CultMesh.cs @@ -2921,11 +2921,12 @@ namespace GameCult.Mesh cancellationToken: cancellationToken) .ConfigureAwait(false); var publications = initial.OfType().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 diff --git a/src/GameCult.Mesh/CultMeshBodyDemand.cs b/src/GameCult.Mesh/CultMeshBodyDemand.cs index 416cc13c..add14b04 100644 --- a/src/GameCult.Mesh/CultMeshBodyDemand.cs +++ b/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)) diff --git a/src/GameCult.Mesh/README.md b/src/GameCult.Mesh/README.md index 2539f83e..f6e0ea30 100644 --- a/src/GameCult.Mesh/README.md +++ b/src/GameCult.Mesh/README.md @@ -403,14 +403,20 @@ using var input = await subscriptions.SubscribeLiveValueAsync` 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. diff --git a/src/GameCult.Mesh/docs/public-api.md b/src/GameCult.Mesh/docs/public-api.md index 876a56f6..429860e0 100644 --- a/src/GameCult.Mesh/docs/public-api.md +++ b/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 diff --git a/src/GameCult.Mesh/docs/reliability-migration-plan.md b/src/GameCult.Mesh/docs/reliability-migration-plan.md index 11437238..feb025f2 100644 --- a/src/GameCult.Mesh/docs/reliability-migration-plan.md +++ b/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` 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 diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs index 1491e45c..51cda8f0 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs +++ b/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().SchemaId; - var value = new CultNetLiveValue(this, subscriptionId, recordKey, schemaId); + var value = new CultNetLiveValue(this, subscriptionId, recordKey); try { var initial = await SubscribeAsync( @@ -80,10 +80,11 @@ namespace GameCult.Networking cancellationToken: cancellationToken) .ConfigureAwait(false); var documents = initial.OfType().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)) diff --git a/tests/GameCult.Mesh.Tests/CultMeshBodyPublicationTests.cs b/tests/GameCult.Mesh.Tests/CultMeshBodyPublicationTests.cs index afd591ba..eb8eca2e 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshBodyPublicationTests.cs +++ b/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( + 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)) diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index e44f0018..0edc9a39 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/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("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(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(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] 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 2382c1b7..7c5433f6 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 95c153bd..068b16c2 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 1f5f832c..679c71b1 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 1f1c16ea..4ccae173 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 a338e289..aa2b1d84 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 89e7b13a..706813e8 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 14c28698..d5aa167e 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 4e448e4a..9a992554 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 15d64a5f..5c66a8af 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 e6a6a632..a68c3148 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 36c08d91..bafe2360 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.28", + "version": "1.0.29", "unity": "2021.3", "description": "CultCache, CultNet, and CultMesh managed runtime assemblies for Unity.", "license": "MIT",