diff --git a/src/GameCult.Networking/CultNetDatabaseServer.cs b/src/GameCult.Networking/CultNetDatabaseServer.cs index 1a66dc2a..0dc489ed 100644 --- a/src/GameCult.Networking/CultNetDatabaseServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseServer.cs @@ -504,7 +504,18 @@ namespace GameCult.Networking { return _database.WatchAllChanges().Subscribe(change => { - var outbound = CreateChangeMessage(change, subscriptionId, selection); + CultNetDatabaseChangeRawMessage? outbound; + try + { + outbound = CreateChangeMessage(change, subscriptionId, selection); + } + catch (NotSupportedException refusal) + { + // Q6: a variant is not a record CultNet can carry; the subscriber that would have received it is told, by key. + peer.SendCultNet(new CultNetErrorMessage { Error = refusal.Message }); + return; + } + if (outbound != null) { peer.SendCultNet(outbound); @@ -549,11 +560,12 @@ namespace GameCult.Networking }; } + CultNetDocumentRegistry.RefuseVariantChange(_database.Cache, key); return new CultNetDatabaseChangeRawMessage { MessageId = Guid.NewGuid().ToString("N"), SubscriptionId = subscriptionId, - ChangeKind = kind == CultNetDatabaseChangeKind.Added ? "added" : "updated", + ChangeKind = kind ==CultNetDatabaseChangeKind.Added ? "added" : "updated", Document = _database.Documents.ToRawRecord(descriptor, key, document, DateTimeOffset.UtcNow.ToString("O")) }; } diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs index 2ccb031c..359a9dfb 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs @@ -315,7 +315,18 @@ namespace GameCult.Networking } var sourceRecordKey = ResolveChangeRecordKey(change); - var matched = CreateMatchedRecord(change, request, peer); + CultNetRawDocumentRecord? matched; + try + { + matched = CreateMatchedRecord(change, request, peer); + } + catch (NotSupportedException refusal) + { + // Q6: a variant is not a record CultNet can carry; the subscriber that would have received it is told, by key. + peer.SendCultNet(new CultNetErrorMessage { Error = refusal.Message }); + return; + } + ApplyProjectedChange( projection, sourceRecordKey, @@ -541,6 +552,7 @@ namespace GameCult.Networking if (kind == CultNetDatabaseChangeKind.Removed || document == null) return null; + CultNetDocumentRegistry.RefuseVariantChange(_database.Cache, key); return _database.Documents.ToRawRecord(descriptor, key, document, DateTimeOffset.UtcNow.ToString("O")); } diff --git a/src/GameCult.Networking/CultNetDocumentRegistry.cs b/src/GameCult.Networking/CultNetDocumentRegistry.cs index 60446d31..356761f5 100644 --- a/src/GameCult.Networking/CultNetDocumentRegistry.cs +++ b/src/GameCult.Networking/CultNetDocumentRegistry.cs @@ -551,10 +551,17 @@ namespace GameCult.Networking // CultNet carries no variant deltas yet, and a resolved view sent as a plain record would be a lie about the store. private static CultStoredDocument RefuseVariant(CultStoredDocument entry) => - entry.Variant == null - ? entry - : throw new NotSupportedException( - $"Record '{entry.Key.Value}' is a variant of '{entry.Variant.BaseKey}'; CultNet does not carry document variants yet."); + entry.Variant == null ? entry : throw VariantRefusal(entry); + + // The live-change funnel's refusal: a load or write that lands a variant must not reach a subscriber as a record. + internal static void RefuseVariantChange(CultCache cache, CultRecordKey key) + { + if (cache.GetStored(key) is { Variant: not null } entry) + throw VariantRefusal(entry); + } + + private static NotSupportedException VariantRefusal(CultStoredDocument entry) => + new($"Record '{entry.Key.Value}' is a variant of '{entry.Variant!.BaseKey}'; CultNet does not carry document variants yet."); /// /// Builds the wire record for one evaluated row. The shared path under snapshot and change diff --git a/tests/GameCult.Networking.Tests/CultNetDocumentRegistryVariantTests.cs b/tests/GameCult.Networking.Tests/CultNetDocumentRegistryVariantTests.cs index 46f91438..2a8ddc58 100644 --- a/tests/GameCult.Networking.Tests/CultNetDocumentRegistryVariantTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetDocumentRegistryVariantTests.cs @@ -1,5 +1,12 @@ #nullable enable using System; +using System.Collections.Concurrent; +using System.IO; +using System.Linq; +using System.Net; +using System.Net.Sockets; +using System.Reflection; +using System.Threading; using System.Threading.Tasks; using GameCult.Caching; using GameCult.Caching.MessagePack; @@ -63,6 +70,128 @@ namespace GameCult.Networking.Tests Assert.That(documents.CreateRawSnapshotResponse(cache, "m2").Documents, Has.Length.EqualTo(1)); } + // The live-change funnel: a load that lands a variant reaches a subscriber as neither a record nor silence. + [Test] + public async Task ALoadThatLandsAVariantNeverReachesASubscriberAsARecordAndTheSubscriberIsToldByKey() + { + var registry = CultDocumentRegistry.Shared; + var directory = Path.Combine(Path.GetTempPath(), $"cultlib-variant-live-{Guid.NewGuid():N}"); + Directory.CreateDirectory(directory); + try + { + var path = Path.Combine(directory, "wire.cc"); + using (var writer = CultCacheMessagePack.Create(path, new CultCacheOpenOptions { Registry = registry })) + { + writer.Commit(batch => batch.Upsert(typeof(VariantWireFixture), new VariantWireFixture { Name = "base", Power = 1 }, BaseKey)); + writer.UpsertVariantAsync(VariantKey, BaseKey, new[] + { + writer.Override(nameof(VariantWireFixture.Name), "big"), + writer.Override(nameof(VariantWireFixture.Power), 9) + }).GetAwaiter().GetResult(); + } + + var cache = new CultCache(registry, CultCacheMessagePack.CreateCodec(registry)); + var documents = new CultNetDocumentRegistry(registry) + .Register(CultNetDocumentBinding.ForDocument(registry)); + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions { DocumentRegistry = documents }); + var socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp); + socket.Bind(new IPEndPoint(IPAddress.Loopback, 0)); + socket.ReceiveTimeout = 20; + using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions + { + RuntimeId = "variant-live-server", + Socket = socket + }); + using var subscriptions = new CultNetDatabaseSubscriptionServer(server, database); + using var cancellation = new CancellationTokenSource(); + var serverThread = new Thread(() => + { + while (!cancellation.IsCancellationRequested) + { + _ = server.PollOnceAsync().GetAwaiter().GetResult(); + Thread.Sleep(1); + } + }) { IsBackground = true }; + serverThread.Start(); + try + { + using var client = CultNetSchemaClients.CreateRudp("variant-live-client"); + var snapshot = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var changes = new ConcurrentQueue(); + var errors = new ConcurrentQueue(); + client.OnCultNet(_ => snapshot.TrySetResult(true)); + client.OnCultNet(changes.Enqueue); + client.OnCultNet(errors.Enqueue); + client.Connect("127.0.0.1", server.LocalEndPoint.Port); + await WaitUntilAsync(() => client.Connected); + client.SendCultNet(new CultNetDatabaseSubscribeMessage + { + MessageId = "subscribe-all", + SubscriptionId = "all", + IncludeSnapshot = true + }); + await WaitUntilAsync(() => snapshot.Task.IsCompleted); + + cache.AddBackingStore(new SingleFileMessagePackBackingStore(path)); + + await WaitUntilAsync(() => !errors.IsEmpty || changes.Any(change => change.Document?.RecordKey == VariantKey.Value)); + await Task.Delay(200); // let anything else already in flight arrive + + Assert.That(changes.Select(change => change.Document?.RecordKey), Does.Not.Contain(VariantKey.Value), + "a variant must not reach a subscriber as a plain record"); + Assert.That(errors.Select(error => error.Error), Has.Some.Contains(VariantKey.Value), + "the subscriber is told, naming the variant"); + } + finally + { + cancellation.Cancel(); + serverThread.Join(); + } + } + finally + { + Directory.Delete(directory, recursive: true); + } + } + + [Test] + public async Task TheDatabaseServersLiveChangeForAVariantIsRefusedNamingIt() + { + var (cache, documents) = await BuildFixtureAsync(withVariant: true); + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions { DocumentRegistry = documents }); + using var server = new Server(cache, ServerSecurityOptions.Development()); + using var databaseServer = new CultNetDatabaseServer(server, database); + var schemaId = cache.Registry.GetRequired().SchemaId; + var change = new CultNetDatabaseChange( + CultNetDatabaseChangeKind.Added, + VariantKey, + schemaId, + database.Shards[0], + cache.Get(VariantKey)!, + previousDocument: null); + var method = typeof(CultNetDatabaseServer).GetMethod("CreateChangeMessage", BindingFlags.Instance | BindingFlags.NonPublic)!; + + var refused = Assert.Throws(() => method.Invoke(databaseServer, new object[] + { + change, "sub-1", new CultNetSelection { Projection = CultNetSelectionProjections.Document } + }))!; + + Assert.That(refused.InnerException, Is.TypeOf()); + Assert.That(refused.InnerException!.Message, Does.Contain(VariantKey.Value).And.Contain(BaseKey.Value)); + } + + private static async Task WaitUntilAsync(Func condition) + { + var deadline = DateTimeOffset.UtcNow + TimeSpan.FromSeconds(5); + while (!condition()) + { + if (DateTimeOffset.UtcNow > deadline) + Assert.Fail("Condition was not satisfied before timeout."); + await Task.Delay(5); + } + } + + [CultDocument("variants.wire-fixture", "variants.wire-fixture.v1")] [MessagePackObject(AllowPrivate = true)] public sealed class VariantWireFixture