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() {