From b572823d0d9ed7744d59f0cef03ccb647d3e7779 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 29 Sep 2026 08:28:24 +0200 Subject: [PATCH] Refuse variants after the selection is applied, and never tell a subscriber about the removal of one it was only refused The selection paths (SelectPage, SelectAll) refused every stored variant before the selection ran, so once any variant existed a hop-bearing subscription threw out of its change handler and went silent. They now refuse only a variant the selection selects; the subscription server delivers that as a typed error and keeps its projection. The database server no longer sends removed for a key the subscriber only ever received as an error. A flatten racing the delivery check is recorded in a comment. Co-Authored-By: Claude Sonnet 5 --- .../CultNetDatabaseServer.cs | 52 ++++-- .../CultNetDatabaseSubscriptionServer.cs | 24 ++- .../CultNetDocumentRegistry.cs | 28 ++- .../CultNetDocumentRegistryVariantTests.cs | 160 ++++++++++++++++++ 4 files changed, 240 insertions(+), 24 deletions(-) diff --git a/src/GameCult.Networking/CultNetDatabaseServer.cs b/src/GameCult.Networking/CultNetDatabaseServer.cs index 0dc489ed..d53315ed 100644 --- a/src/GameCult.Networking/CultNetDatabaseServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseServer.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Concurrent; +using System.Collections.Generic; using System.Linq; using System.Reflection; using System.Threading.Tasks; @@ -502,27 +503,48 @@ namespace GameCult.Networking string subscriptionId, CultNetServerPeer peer) { + // Keys this subscriber only ever received as a refusal: it was never given them as records. + var refused = new HashSet(StringComparer.Ordinal); return _database.WatchAllChanges().Subscribe(change => { - 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; - } - + var outbound = DeliverChange(change, subscriptionId, selection, refused, error => peer.SendCultNet(error)); if (outbound != null) - { peer.SendCultNet(outbound); - } }); } + // One subscriber's view of one change. A variant is not a record CultNet can carry (Q6): the subscriber is told, by key, and + // is not later told the variant was removed, because it never held it. + internal CultNetDatabaseChangeRawMessage? DeliverChange( + object change, + string subscriptionId, + CultNetSelection selection, + HashSet refused, + Action sendError) + { + CultNetDatabaseChangeRawMessage? outbound; + try + { + outbound = CreateChangeMessage(change, subscriptionId, selection); + } + catch (NotSupportedException refusal) + { + var refusedKey = change.GetType().GetProperty("Key")?.GetValue(change) is CultRecordKey changed ? changed.Value : string.Empty; + lock (refused) + refused.Add(refusedKey); + sendError(new CultNetErrorMessage { Error = refusal.Message }); + return null; + } + + if (outbound == null) + return null; + var recordKey = outbound.RecordKey ?? outbound.Document?.RecordKey ?? string.Empty; + bool wasRefused; + lock (refused) + wasRefused = refused.Remove(recordKey); + return wasRefused && outbound.ChangeKind == "removed" ? null : outbound; + } + internal CultNetDatabaseChangeRawMessage? CreateChangeMessage( object change, string subscriptionId, @@ -565,7 +587,7 @@ namespace GameCult.Networking { 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 359a9dfb..bb1f3900 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs @@ -204,6 +204,13 @@ namespace GameCult.Networking peer.SendCultNet(CultNetErrorMessage.ForReferenceOutsideTarget(ex)); return Task.CompletedTask; } + catch (NotSupportedException refusal) + { + // Q6: a snapshot that selects a variant is refused to this subscriber as a typed error. + Withdraw(key, sendRemovals: false, forgetRequest: true); + peer.SendCultNet(new CultNetErrorMessage { Error = refusal.Message }); + return Task.CompletedTask; + } catch { Withdraw(key, sendRemovals: false, forgetRequest: true); @@ -342,9 +349,20 @@ namespace GameCult.Networking if (!_projections.TryGetValue(key, out var projection)) return; var legacyRequest = request.ToLegacyShape(key.Id); var authorized = _authorizeRequest?.Invoke(legacyRequest, key.Peer) != false; - var next = authorized - ? CreateProjectedSnapshot(request, key.Peer) - : ProjectedSnapshot.Empty; + ProjectedSnapshot next; + try + { + next = authorized + ? CreateProjectedSnapshot(request, key.Peer) + : ProjectedSnapshot.Empty; + } + 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. + key.Peer.SendCultNet(new CultNetErrorMessage { Error = refusal.Message }); + return; + } // With _projectRecord gone (docs/cultnet-selection-cut.md, S2-2), a projection's dictionary // key is always the record's own RecordKey/SchemaId, so the two can never disagree here; diff --git a/src/GameCult.Networking/CultNetDocumentRegistry.cs b/src/GameCult.Networking/CultNetDocumentRegistry.cs index 356761f5..5df34e98 100644 --- a/src/GameCult.Networking/CultNetDocumentRegistry.cs +++ b/src/GameCult.Networking/CultNetDocumentRegistry.cs @@ -391,11 +391,13 @@ namespace GameCult.Networking var stored = cache.AllStoredDocuments; if (rowFilter != null) stored = stored.Where(entry => rowFilter(entry.Descriptor, entry.Key)); - var rows = stored + var entries = stored.ToArray(); + var rows = entries .Select(entry => new CultNetSelectionEvaluator.Row( - entry.Descriptor, entry.Key, RefuseVariant(entry).Document, ordinalOf(entry.Descriptor.SchemaId, entry.Key), entry.StoredAt)) + 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); // 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 @@ -485,12 +487,14 @@ namespace GameCult.Networking var stored = cache.AllStoredDocuments; if (rowFilter != null) stored = stored.Where(entry => rowFilter(entry.Descriptor, entry.Key)); - var rows = stored + var entries = stored.ToArray(); + var rows = entries .Select(entry => new CultNetSelectionEvaluator.Row( - entry.Descriptor, entry.Key, RefuseVariant(entry).Document, ordinalOf(entry.Descriptor.SchemaId, entry.Key), entry.StoredAt)) + 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); var records = full.Ordered.Select(row => ToRawRecord(row, options)).ToArray(); var wantDocument = selection.Projection == CultNetSelectionProjections.Document; var edges = selection.HasHop @@ -550,12 +554,24 @@ 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 VariantRefusal(entry); + // 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) + { + if (selected.Count == 0 || !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) + { + if (variants.TryGetValue(row.Key.Value, out var variant)) + throw VariantRefusal(variant); + } + } // 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) { + // Recorded, not fixed: a Flatten committed on another thread between the change's admission and this read makes + // the record plain here, and its resolved view is delivered. The window is one thread hop wide. if (cache.GetStored(key) is { Variant: not null } entry) throw VariantRefusal(entry); } diff --git a/tests/GameCult.Networking.Tests/CultNetDocumentRegistryVariantTests.cs b/tests/GameCult.Networking.Tests/CultNetDocumentRegistryVariantTests.cs index 8c70c31e..2ae6dfab 100644 --- a/tests/GameCult.Networking.Tests/CultNetDocumentRegistryVariantTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetDocumentRegistryVariantTests.cs @@ -182,6 +182,153 @@ namespace GameCult.Networking.Tests Assert.That(refused.InnerException!.Message, Does.Contain(VariantKey.Value).And.Contain(BaseKey.Value)); } + + // A hop-bearing selection reconciles the whole store on every change. The refusal follows the selection: a subscriber + // whose selection never selects the variant keeps receiving; one that does is told, as a typed error, not a silent stop. + [Test] + public async Task AHopBearingSubscriptionReconcilesPastAVariantItDoesNotSelectAndIsToldWhenItDoes() + { + 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)); + var wireSchemaId = registry.GetRequired().SchemaId; + 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 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-hop-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-hop-client"); + var snapshots = new ConcurrentQueue(); + var changes = new ConcurrentQueue(); + var errors = new ConcurrentQueue(); + client.OnCultNet(snapshots.Enqueue); + client.OnCultNet(changes.Enqueue); + client.OnCultNet(errors.Enqueue); + client.Connect("127.0.0.1", server.LocalEndPoint.Port); + await WaitUntilAsync(() => client.Connected); + string Diagnostic() => + $"changes={string.Join(",", changes.Select(c => c.SubscriptionId + ":" + c.ChangeKind + ":" + (c.Document?.RecordKey ?? c.RecordKey)))} " + + $"errors={string.Join(",", errors.Select(e => e.Error))}"; + + // "citers": rows citing the base. It never selects the variant. + client.SendCultNet(new CultNetDatabaseSubscribeV1Message + { + MessageId = "citers", + SubscriptionId = "citers", + IncludeSnapshot = true, + Selection = new CultNetSelection + { + Cites = new CultNetCitation { Target = new CultNetRecordRef { SchemaId = wireSchemaId, RecordKey = BaseKey.Value }, Role = "Ref" } + } + }); + await WaitUntilAsync(() => snapshots.Count == 1, Diagnostic); + + cache.Commit(batch => batch.UpsertVariant(VariantKey, BaseKey, new[] + { + cache.Override(nameof(VariantWireFixture.Name), "big") + })); + cache.Commit(batch => batch.Upsert(typeof(VariantCiterFixture), Citer("citer-two", BaseKey), new CultRecordKey("citer-two"))); + + await WaitUntilAsync(() => changes.Any(change => change.SubscriptionId == "citers" && change.Document?.RecordKey == "citer-two"), Diagnostic); + Assert.That(errors, Is.Empty, "a subscriber that never selects the variant is never told about it"); + + // "cited": wire rows some citer cites. Once a citer cites the variant, it selects it. + client.SendCultNet(new CultNetDatabaseSubscribeV1Message + { + MessageId = "cited", + SubscriptionId = "cited", + IncludeSnapshot = true, + Selection = new CultNetSelection + { + Schemas = new[] { wireSchemaId }, + Cited = new CultNetIncoming { Role = "Ref", Exists = true } + } + }); + await WaitUntilAsync(() => snapshots.Count == 2, Diagnostic); + Assert.That(errors, Is.Empty, "only the base is cited yet"); + cache.Commit(batch => batch.Upsert(typeof(VariantCiterFixture), Citer("citer-three", VariantKey), new CultRecordKey("citer-three"))); + + await WaitUntilAsync(() => errors.Any(error => error.Error.Contains(VariantKey.Value)), Diagnostic); + 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. + cache.Commit(batch => batch.Upsert(typeof(VariantCiterFixture), Citer("citer-four", BaseKey), new CultRecordKey("citer-four"))); + await WaitUntilAsync(() => changes.Any(change => change.SubscriptionId == "citers" && change.Document?.RecordKey == "citer-four"), Diagnostic); + } + finally + { + cancellation.Cancel(); + serverThread.Join(); + } + } + + private static CultNetDatabaseChangeRawMessage? Deliver( + CultNetDatabaseServer server, + object change, + CultNetSelection selection, + System.Collections.Generic.HashSet refused, + System.Collections.Generic.List errors) + { + var method = typeof(CultNetDatabaseServer).GetMethod("DeliverChange", BindingFlags.Instance | BindingFlags.NonPublic)!; + return (CultNetDatabaseChangeRawMessage?)method.Invoke(server, new object[] + { + change, "sub", selection, refused, (Action)errors.Add + }); + } + + private static VariantCiterFixture Citer(string name, CultRecordKey target) => + new() { Name = name, Ref = new CultRecordRef(target) }; + + // A subscriber told a variant only as an error is not later told it was removed: it never held it. + [Test] + public async Task RemovingAVariantTheSubscriberWasOnlyToldAboutAsAnErrorSendsNoRemoval() + { + 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 selection = new CultNetSelection { Projection = CultNetSelectionProjections.Document }; + var refused = new System.Collections.Generic.HashSet(); + var errors = new System.Collections.Generic.List(); + + var added = new CultNetDatabaseChange( + CultNetDatabaseChangeKind.Added, VariantKey, schemaId, database.Shards[0], cache.Get(VariantKey)!, previousDocument: null); + Assert.That(Deliver(databaseServer, added, selection, refused, errors), Is.Null); + Assert.That(errors, Has.Count.EqualTo(1)); + + var removed = new CultNetDatabaseChange( + CultNetDatabaseChangeKind.Removed, VariantKey, schemaId, database.Shards[0], document: null, previousDocument: cache.Get(VariantKey)!); + Assert.That(Deliver(databaseServer, removed, selection, refused, errors), Is.Null, + "no removal for a key the subscriber never held as a record"); + + // A record it did hold is still told about. + var baseRemoved = new CultNetDatabaseChange( + CultNetDatabaseChangeKind.Removed, BaseKey, schemaId, database.Shards[0], document: null, previousDocument: cache.Get(BaseKey)!); + Assert.That(Deliver(databaseServer, baseRemoved, selection, refused, errors)?.ChangeKind, Is.EqualTo("removed")); + } + private static async Task WaitUntilAsync(Func condition, Func? diagnostic = null) { var deadline = DateTimeOffset.UtcNow + TimeSpan.FromSeconds(5); @@ -205,5 +352,18 @@ namespace GameCult.Networking.Tests [Key(1)] public int Power; } + + [CultDocument("variants.wire-citer-fixture", "variants.wire-citer-fixture.v1")] + [MessagePackObject(AllowPrivate = true)] + public sealed class VariantCiterFixture + { + [Key(0)] + [CultName] + public string Name = string.Empty; + + [Key(1)] + [CultReference(typeof(VariantWireFixture))] + public CultRecordRef Ref; + } } }