diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs index bb1f3900..60bb66a5 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs @@ -358,8 +358,10 @@ namespace GameCult.Networking } catch (NotSupportedException refusal) { - // Q6: the selection selects a variant, which CultNet cannot carry. This subscriber is told, by key, and keeps what - // it was delivered; a change never throws out of the change handler. + // Q6: the selection selects a variant, which CultNet cannot carry. Refused as the subscribe path refuses it: the + // subscription is withdrawn, so the peer is told once, not once per store change; a change never throws out of + // the change handler. + Withdraw(key, sendRemovals: false, forgetRequest: true); key.Peer.SendCultNet(new CultNetErrorMessage { Error = refusal.Message }); return; } diff --git a/src/GameCult.Networking/CultNetDocumentRegistry.cs b/src/GameCult.Networking/CultNetDocumentRegistry.cs index 5df34e98..eb23d503 100644 --- a/src/GameCult.Networking/CultNetDocumentRegistry.cs +++ b/src/GameCult.Networking/CultNetDocumentRegistry.cs @@ -397,7 +397,7 @@ namespace GameCult.Networking entry.Descriptor, entry.Key, entry.Document, ordinalOf(entry.Descriptor.SchemaId, entry.Key), entry.StoredAt)) .ToArray(); var full = CultNetSelectionEvaluator.EvaluateAll(_documents, rows, selection); - RefuseSelectedVariants(entries, full.Ordered); + RefuseVariants(entries, full.Ordered.Select(row => row.Key)); // R-Q: a page's asOf is exact only when every matched row is committed through one shard's // log - a caller that knows about shards (CultNetDatabaseServer) supplies shardIdOf; one // that does not (the raw v0/shard-scoped rowFilter paths, which are already bounded to a @@ -417,6 +417,7 @@ namespace GameCult.Networking } var evaluation = CultNetSelectionEvaluator.Page(full, selection, asOf, cursorKey); + RefuseVariants(entries, evaluation.Edges.Select(edge => edge.From.Key)); var records = evaluation.Rows.Select(row => ToRawRecord(row, options)).ToArray(); var wantDocument = selection.Projection == CultNetSelectionProjections.Document; @@ -494,12 +495,13 @@ namespace GameCult.Networking .ToArray(); var full = CultNetSelectionEvaluator.EvaluateAll(_documents, rows, selection); - RefuseSelectedVariants(entries, full.Ordered); + RefuseVariants(entries, full.Ordered.Select(row => row.Key)); var records = full.Ordered.Select(row => ToRawRecord(row, options)).ToArray(); var wantDocument = selection.Projection == CultNetSelectionProjections.Document; - var edges = selection.HasHop - ? CultNetSelectionEvaluator.EdgesFor(full.Ordered, full).Select(edge => ToEdge(edge, wantDocument)).ToArray() - : null; + var hopEdges = selection.HasHop ? CultNetSelectionEvaluator.EdgesFor(full.Ordered, full) : null; + if (hopEdges != null) + RefuseVariants(entries, hopEdges.Select(edge => edge.From.Key)); + var edges = hopEdges?.Select(edge => ToEdge(edge, wantDocument)).ToArray(); return new CultNetSelectionPage { @@ -554,15 +556,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. - // The refusal follows the selection: a variant is refused only when the selection selects it, never for existing. - private static void RefuseSelectedVariants(IReadOnlyList entries, IReadOnlyList selected) + // The refusal follows the selection: a variant is refused only when the selection selects it, never for existing. An + // edge carries its citer's reference payload, read from the citer's resolved document, so a variant citing is selected + // by the selection's hop exactly as a variant row is. + private static void RefuseVariants(IReadOnlyList entries, IEnumerable selected) { - if (selected.Count == 0 || !entries.Any(entry => entry.Variant != null)) + if (!entries.Any(entry => entry.Variant != null)) return; var variants = entries.Where(entry => entry.Variant != null).ToDictionary(entry => entry.Key.Value, StringComparer.Ordinal); - foreach (var row in selected) + foreach (var key in selected) { - if (variants.TryGetValue(row.Key.Value, out var variant)) + if (variants.TryGetValue(key.Value, out var variant)) throw VariantRefusal(variant); } } diff --git a/tests/GameCult.Networking.Tests/CultNetDocumentRegistryVariantTests.cs b/tests/GameCult.Networking.Tests/CultNetDocumentRegistryVariantTests.cs index 7aeadc15..a4658117 100644 --- a/tests/GameCult.Networking.Tests/CultNetDocumentRegistryVariantTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetDocumentRegistryVariantTests.cs @@ -275,9 +275,103 @@ namespace GameCult.Networking.Tests Assert.That(changes.Where(change => change.SubscriptionId == "cited").Select(change => change.Document?.RecordKey), Does.Not.Contain(VariantKey.Value), "a variant never reaches a subscriber as a record"); - // The first subscriber is still alive after the refusal. + // The first subscriber is still alive after the refusal; the refused one is withdrawn, not told again per change. + var told = errors.Count; await database.PutAsync(new CultRecordKey("citer-five"), Citer("citer-five", BaseKey)); await WaitUntilAsync(() => changes.Any(change => change.SubscriptionId == "citers" && change.Document?.RecordKey == "citer-five"), Diagnostic); + await Task.Delay(200); + Assert.That(errors.Count, Is.EqualTo(told), "a refused subscription is withdrawn: one error, not one per store change"); + } + finally + { + cancellation.Cancel(); + serverThread.Join(); + } + } + + // An edge carries its citer's reference payload, read from the citer's resolved document. A selection whose hop + // reaches a variant citer is refused as a selection of a variant row is, though only the cited row is selected. + private static async Task<(CultCache Cache, CultNetDocumentRegistry Documents, CultNetSelection Selection, CultRecordKey CiterVariant)> BuildCitedByVariantFixtureAsync() + { + var registry = CultDocumentRegistry.Shared; + var cache = new CultCache(registry, CultCacheMessagePack.CreateCodec(registry)); + var documents = new CultNetDocumentRegistry(registry) + .Register(CultNetDocumentBinding.ForDocument(registry)) + .Register(CultNetDocumentBinding.ForDocument(registry)); + cache.Commit(batch => + { + batch.Upsert(typeof(VariantWireFixture), new VariantWireFixture { Name = "base", Power = 1 }, BaseKey); + batch.Upsert(typeof(VariantCiterFixture), Citer("citer-one", BaseKey), new CultRecordKey("citer-one")); + }); + var citerVariant = new CultRecordKey("citer-variant"); + await cache.UpsertVariantAsync(citerVariant, new CultRecordKey("citer-one"), new[] + { + cache.Override(nameof(VariantCiterFixture.Name), "citer variant") + }); + var selection = new CultNetSelection + { + Schemas = new[] { cache.Registry.GetRequired().SchemaId }, + Cited = new CultNetIncoming { Role = "Ref", Exists = true }, + Projection = CultNetSelectionProjections.Document + }; + return (cache, documents, selection, citerVariant); + } + + [Test] + public async Task AHopEdgeWhoseCiterIsAVariantIsRefusedNamingItAndAStoreWithoutOneAnswers() + { + var (cache, documents, selection, citerVariant) = await BuildCitedByVariantFixtureAsync(); + + var error = Assert.Throws(() => + documents.CreateSelectionResponse(cache, "m1", selection, ordinalOf: (_, _) => 1, asOf: 1))!; + Assert.That(error.Message, Does.Contain(citerVariant.Value)); + + cache.Commit(batch => batch.Remove(citerVariant)); + var answered = documents.CreateSelectionResponse(cache, "m2", selection, ordinalOf: (_, _) => 1, asOf: 1); + Assert.That(answered.Edges, Has.Length.EqualTo(1)); + Assert.That(answered.Edges![0].From.RecordKey, Is.EqualTo("citer-one")); + } + + [Test] + public async Task AHopSubscriptionWhoseEdgeCiterIsAVariantIsRefusedOnceAndWithdrawn() + { + var (cache, documents, selection, citerVariant) = await BuildCitedByVariantFixtureAsync(); + 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-edge-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-edge-client"); + var snapshots = new ConcurrentQueue(); + var errors = new ConcurrentQueue(); + client.OnCultNet(snapshots.Enqueue); + client.OnCultNet(errors.Enqueue); + client.Connect("127.0.0.1", server.LocalEndPoint.Port); + await WaitUntilAsync(() => client.Connected); + + client.SendCultNet(new CultNetDatabaseSubscribeV1Message + { + MessageId = "edges", + SubscriptionId = "edges", + IncludeSnapshot = true, + Selection = selection + }); + await WaitUntilAsync(() => errors.Any(error => error.Error.Contains(citerVariant.Value))); + Assert.That(snapshots, Is.Empty, "the snapshot's edges would carry the variant's resolved reference"); + Assert.That(errors, Has.Count.EqualTo(1)); } finally {