GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Refuse variants after the selection is applied, and never tell a subscriber about the removal of one it was only refused

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-29 08:28:24 +02:00
commit b572823d0d
4 changed files with 240 additions and 24 deletions
Download patch file Download diff file Expand all files Collapse all files

52
src/GameCult.Networking/CultNetDatabaseServer.cs
View file

@ -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<string>(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<string> refused,
Action<CultNetErrorMessage> 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"))
};
}

24
src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs
View file

@ -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;

28
src/GameCult.Networking/CultNetDocumentRegistry.cs
View file

@ -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<CultStoredDocument> entries, IReadOnlyList<CultNetSelectionEvaluator.Row> 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);
}

160
tests/GameCult.Networking.Tests/CultNetDocumentRegistryVariantTests.cs
View file

@ -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<VariantWireFixture>(registry))
.Register(CultNetDocumentBinding.ForDocument<VariantCiterFixture>(registry));
var wireSchemaId = registry.GetRequired<VariantWireFixture>().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<CultNetSnapshotResponseRawV1Message>();
var changes = new ConcurrentQueue<CultNetDatabaseChangeRawMessage>();
var errors = new ConcurrentQueue<CultNetErrorMessage>();
client.OnCultNet<CultNetSnapshotResponseRawV1Message>(snapshots.Enqueue);
client.OnCultNet<CultNetDatabaseChangeRawMessage>(changes.Enqueue);
client.OnCultNet<CultNetErrorMessage>(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<VariantWireFixture>(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<string> refused,
System.Collections.Generic.List<CultNetErrorMessage> errors)
{
var method = typeof(CultNetDatabaseServer).GetMethod("DeliverChange", BindingFlags.Instance | BindingFlags.NonPublic)!;
return (CultNetDatabaseChangeRawMessage?)method.Invoke(server, new object[]
{
change, "sub", selection, refused, (Action<CultNetErrorMessage>)errors.Add
});
}
private static VariantCiterFixture Citer(string name, CultRecordKey target) =>
new() { Name = name, Ref = new CultRecordRef<VariantWireFixture>(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<VariantWireFixture>().SchemaId;
var selection = new CultNetSelection { Projection = CultNetSelectionProjections.Document };
var refused = new System.Collections.Generic.HashSet<string>();
var errors = new System.Collections.Generic.List<CultNetErrorMessage>();
var added = new CultNetDatabaseChange<VariantWireFixture>(
CultNetDatabaseChangeKind.Added, VariantKey, schemaId, database.Shards[0], cache.Get<VariantWireFixture>(VariantKey)!, previousDocument: null);
Assert.That(Deliver(databaseServer, added, selection, refused, errors), Is.Null);
Assert.That(errors, Has.Count.EqualTo(1));
var removed = new CultNetDatabaseChange<VariantWireFixture>(
CultNetDatabaseChangeKind.Removed, VariantKey, schemaId, database.Shards[0], document: null, previousDocument: cache.Get<VariantWireFixture>(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<VariantWireFixture>(
CultNetDatabaseChangeKind.Removed, BaseKey, schemaId, database.Shards[0], document: null, previousDocument: cache.Get<VariantWireFixture>(BaseKey)!);
Assert.That(Deliver(databaseServer, baseRemoved, selection, refused, errors)?.ChangeKind, Is.EqualTo("removed"));
}
private static async Task WaitUntilAsync(Func<bool> condition, Func<string>? 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<VariantWireFixture> Ref;
}
}
}