From b3a8f6d1954eeeb001e1f3bc79b1c51370e4baca Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 22 Sep 2026 21:12:07 +0200 Subject: [PATCH] Fix batch 4 (S-9/S-11): shard-scoped asOf, and the v1 subscribe door runs before the hop refusal S-9: CurrentAsOf() returned the maximum shard-log sequence across every shard the database holds, but that sequence is a per-shard counter - mixing shards' counters together could report a page as exact "as of" a sequence a different, untouched shard advanced to while the page's own shard sat still. CultNetDatabase gains CurrentAsOf(shardId), the watermark of one shard's own log, and CultNetDocumentRegistry.SelectPage now scopes asOf to the single shard EnsureSingleShard already resolves the matched rows to, falling back to the caller-supplied value only when the selection matches no rows. S-11: CultNetDatabaseServer.HandleSubscribeV1Async refused a hop-bearing selection before running the door (Validate), so a selection invalid for an unrelated reason (an empty keys list, an undeclared index, ...) got a hop refusal that had nothing to do with why it was actually rejected. The door now runs first, unconditionally. The hop refusal also always named the field "cites" even when the selection carried Cited instead; it now names whichever hop is actually set. Co-Authored-By: Claude Sonnet 5 --- src/GameCult.Networking/CultNetDatabase.cs | 27 ++++++++-- .../CultNetDatabaseServer.cs | 43 +++++++++------- .../CultNetDocumentRegistry.cs | 27 +++++++--- .../NetworkingTests.cs | 49 +++++++++++++++++++ 4 files changed, 118 insertions(+), 28 deletions(-) diff --git a/src/GameCult.Networking/CultNetDatabase.cs b/src/GameCult.Networking/CultNetDatabase.cs index 43144333..d59cd6c7 100644 --- a/src/GameCult.Networking/CultNetDatabase.cs +++ b/src/GameCult.Networking/CultNetDatabase.cs @@ -1303,14 +1303,31 @@ namespace GameCult.Networking _lastWriteSequence.TryGetValue((schemaId, key.Value), out var sequence) ? sequence : null; /// - /// The current write-sequence watermark this database can present an exact snapshot for (R-A): - /// the highest assigned to any row, 0 when nothing has been - /// committed yet. This is the evaluator's asOf for a v1 snapshot or subscription page - it - /// advances on every committed write, so a cursor minted against an older watermark refuses as - /// cursor_stale once a page's answer could differ. + /// The current write-sequence watermark across every shard this database holds, 0 when nothing + /// has been committed yet. Only meaningful for an answer that is not scoped to any particular + /// shard's log - a selection's own asOf must use instead + /// (S-9): the shard-log sequence is a per-shard counter (), + /// so taking the maximum across every shard's rows mixes counters that do not compare to one + /// another and can report a page as exact "as of" a sequence another, unrelated shard advanced to + /// while the page's own shard sat still. /// public ulong CurrentAsOf() => _lastWriteSequence.Count == 0 ? 0UL : (ulong)_lastWriteSequence.Values.Max(); + /// + /// The write-sequence watermark of one shard's own log (S-9): the sequence of the last commit + /// that shard's log recorded, or 0 when the shard has never taken a write. This is the evaluator's + /// asOf for a v1 snapshot or subscription page whose matched rows all come from one shard + /// ('s single-shard check) - it advances only when that + /// shard's own log advances, so a cursor minted against it refuses cursor_stale exactly + /// when that shard's log has moved, never because an unrelated shard elsewhere took a write the + /// selection never touched. + /// + public ulong CurrentAsOf(string shardId) + { + if (string.IsNullOrWhiteSpace(shardId)) throw new ArgumentException("Value must be non-empty.", nameof(shardId)); + return _nextLogSequences.TryGetValue(shardId, out var next) ? (ulong)(next - 1) : 0UL; + } + private void RecordMutationLogEntry( CultNetShardMutationLogEntry entry, CultNetShardLogEntryMessage? wireEntry = null) diff --git a/src/GameCult.Networking/CultNetDatabaseServer.cs b/src/GameCult.Networking/CultNetDatabaseServer.cs index 1e53abf1..6b86f4aa 100644 --- a/src/GameCult.Networking/CultNetDatabaseServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseServer.cs @@ -135,7 +135,10 @@ namespace GameCult.Networking // one shard's log is refused rather than answered against a watermark that is not // exact for all of them. shardIdOf: (schemaId, key) => _database.ResolveShard(schemaId, key).ShardId, - cursorKey: _database.CursorKey); + cursorKey: _database.CursorKey, + // S-9: once the matched rows resolve to one shard, asOf is that shard's own watermark, + // not CurrentAsOf()'s database-wide maximum across every shard. + asOfForShard: _database.CurrentAsOf); } /// @@ -377,22 +380,9 @@ namespace GameCult.Networking return Task.CompletedTask; } - // This server delivers live changes through the single-row fast path (CreateChangeMessage / - // CultNetSelectionEvaluator.Matches), which is set-independent by construction - the same - // ceiling v0 subscriptions on this server already have. A hop-bearing selection is - // set-dependent (D6) and needs the subscription server's reconcile loop instead, so it is - // refused here rather than reaching Matches' own InvalidOperationException on the first change. - if (message.Selection.HasHop) - { - peer.SendCultNet(new CultNetErrorMessage - { - Error = "selection_invalid: selection.cites/selection.cited need the subscription server's reconcile loop (D6); this server only fast-matches a single row.", - Code = "selection_invalid", - Details = new CultNetErrorDetails { Field = "cites" } - }); - return Task.CompletedTask; - } - + // S-11: the door runs first, unconditionally (R-F) - a selection that is invalid for reasons + // unrelated to its hop (an empty keys list, an undeclared index, ...) reports that refusal, + // not a hop refusal that has nothing to do with why it was actually rejected. try { message.Selection.Validate(_database.Cache.Registry.AllDescriptors.ToArray()); @@ -403,6 +393,25 @@ namespace GameCult.Networking return Task.CompletedTask; } + // This server delivers live changes through the single-row fast path (CreateChangeMessage / + // CultNetSelectionEvaluator.Matches), which is set-independent by construction - the same + // ceiling v0 subscriptions on this server already have. A hop-bearing selection is + // set-dependent (D6) and needs the subscription server's reconcile loop instead, so it is + // refused here, after the door, rather than reaching Matches' own InvalidOperationException + // on the first change. The refused field names whichever hop is actually set - cited, not + // always cites - so the refusal describes the selection that was actually sent. + if (message.Selection.HasHop) + { + var hopField = message.Selection.Cites != null ? "cites" : "cited"; + peer.SendCultNet(new CultNetErrorMessage + { + Error = $"selection_invalid: selection.{hopField} needs the subscription server's reconcile loop (D6); this server only fast-matches a single row.", + Code = "selection_invalid", + Details = new CultNetErrorDetails { Field = hopField } + }); + return Task.CompletedTask; + } + var key = SubscriptionKey(peer.Peer, subscriptionId); _subscriptions.AddOrUpdate( key, diff --git a/src/GameCult.Networking/CultNetDocumentRegistry.cs b/src/GameCult.Networking/CultNetDocumentRegistry.cs index dd85ffec..be114002 100644 --- a/src/GameCult.Networking/CultNetDocumentRegistry.cs +++ b/src/GameCult.Networking/CultNetDocumentRegistry.cs @@ -336,7 +336,8 @@ namespace GameCult.Networking /// Evaluates a typed selection against the cache and returns its page /// (docs/cultnet-selection-cut.md, section 2/6). supplies each /// row's ordinal (the reference: ); - /// is the snapshot the page is exact for. + /// is the snapshot the page is exact for when the selection is not shard-scoped, or the fallback + /// used only when is null or the selection matches no rows (S-9). /// public CultNetSnapshotResponseRawV1Message CreateSelectionResponse( CultCache cache, @@ -346,13 +347,14 @@ namespace GameCult.Networking ulong asOf, CultNetDocumentMessageOptions? options = null, Func? shardIdOf = null, - CultNetSelectionCursorKey? cursorKey = null) + CultNetSelectionCursorKey? cursorKey = null, + Func? asOfForShard = null) { if (cache == null) throw new ArgumentNullException(nameof(cache)); if (selection == null) throw new ArgumentNullException(nameof(selection)); if (ordinalOf == null) throw new ArgumentNullException(nameof(ordinalOf)); - var page = SelectPage(cache, selection, ordinalOf, asOf, options, shardIdOf: shardIdOf, cursorKey: cursorKey); + var page = SelectPage(cache, selection, ordinalOf, asOf, options, shardIdOf: shardIdOf, cursorKey: cursorKey, asOfForShard: asOfForShard); return new CultNetSnapshotResponseRawV1Message { MessageId = RequireNonEmpty(messageId, nameof(messageId)), @@ -378,7 +380,8 @@ namespace GameCult.Networking CultNetDocumentMessageOptions? options, Func? rowFilter = null, Func? shardIdOf = null, - CultNetSelectionCursorKey? cursorKey = null) + CultNetSelectionCursorKey? cursorKey = null, + Func? asOfForShard = null) { selection = ExpandSchemaBindingAliases(selection); // R-F: CultNetSelectionEvaluator.Select validates first, every time - there is no separate @@ -399,7 +402,16 @@ namespace GameCult.Networking // single shard by construction) passes null and skips the check rather than pay for it. if (shardIdOf != null) { - EnsureSingleShard(full.Ordered, shardIdOf); + var shardId = EnsureSingleShard(full.Ordered, shardIdOf); + // S-9: asOf is the watermark of the shard the selection's matched rows actually come + // from, not the database-wide maximum across every shard - a selection matching only one + // shard's rows must not claim exactness as of a sequence a different, untouched shard + // happened to advance to. A selection matching no rows has no shard to scope to, so it + // keeps the caller-supplied fallback. + if (shardId != null && asOfForShard != null) + { + asOf = asOfForShard(shardId); + } } var evaluation = CultNetSelectionEvaluator.Page(full, selection, asOf, cursorKey); @@ -428,7 +440,8 @@ namespace GameCult.Networking /// than adding a second refusal shape for what is, from the wire's point of view, the same /// "this selection cannot be answered as asked" refusal a bad field or an empty list gets. /// - private static void EnsureSingleShard( + /// Returns the one shard every matched row belongs to, or null when the selection matched no rows. + private static string? EnsureSingleShard( IReadOnlyList rows, Func shardIdOf) { @@ -448,6 +461,8 @@ namespace GameCult.Networking $"The selection matches rows from more than one shard's log (at least '{shardId}' and '{rowShardId}'); asOf is not defined for a shard-spanning page."); } } + + return shardId; } /// diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 5ab0d13b..40bf43eb 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -5637,6 +5637,55 @@ namespace GameCult.Networking.Tests Assert.That(ex!.Field, Is.EqualTo("asOf")); } + // S-9 (docs/cultnet-selection-cut.md, fix batch 4): asOf is the watermark of the shard(s) the + // selection's matched rows actually come from, not CurrentAsOf()'s database-wide maximum across + // every shard - a shard-log sequence is a per-shard counter, so a selection matching only + // shard-a's row must not claim exactness as of a sequence shard-b's unrelated writes advanced to. + [Test] + public async Task CultNetDatabaseServer_SelectionResponseAsOf_IsTheMatchedShardsOwnWatermarkNotTheDatabaseWideMaximum() + { + var cache = new CultCache(); + var registry = new CultNetDocumentRegistry(cache.Registry) + .Register(CultNetDocumentBinding.ForDocument( + cache.Registry, + payloadSerializer: SerializePlayerDataPayload, + payloadDeserializer: DeserializePlayerDataPayload)); + var schemaId = cache.Registry.GetRequired().SchemaId; + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions + { + DocumentRegistry = registry, + Shards = + [ + new CultNetShardDescriptor("shard-a", "runtime-a", epoch: 1, isPrimary: true, schemaIds: [schemaId], keyPrefix: "a:"), + new CultNetShardDescriptor("shard-b", "runtime-a", epoch: 1, isPrimary: true, schemaIds: [schemaId], keyPrefix: "b:") + ] + }); + using var server = new Server(cache, DevelopmentServerSecurity); + using var databaseServer = new CultNetDatabaseServer(server, database); + + await database.PutAsync( + new CultRecordKey("a:one"), + new PlayerData { PlayerId = Guid.NewGuid(), Email = "a@example.test", PasswordHash = "hash", Username = "A" }); + // shard-b takes several more writes than shard-a, so the database-wide maximum runs well + // ahead of shard-a's own watermark. + for (var i = 0; i < 4; i++) + { + await database.PutAsync( + new CultRecordKey("b:one"), + new PlayerData { PlayerId = Guid.NewGuid(), Email = $"b{i}@example.test", PasswordHash = "hash", Username = "B" }); + } + + Assert.That(database.CurrentAsOf(), Is.EqualTo(4UL), "sanity: the database-wide maximum is shard-b's watermark"); + Assert.That(database.CurrentAsOf("shard-a"), Is.EqualTo(1UL), "sanity: shard-a's own watermark is unaffected by shard-b's writes"); + + var response = databaseServer.CreateSelectionResponse(new CultNetSnapshotRequestV1Message + { + Selection = new CultNetSelection { Keys = ["a:one"] } + }); + + Assert.That(response.AsOf, Is.EqualTo(1UL)); + } + [Test] public async Task CultNetDatabase_Predicts_ClientOwnedInput_AndReconciles_AuthoritativeLog() {