From 3f942a57e777da3458a5d7c707ebcfe88a02f74a Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 08:03:09 +0200 Subject: [PATCH] Cut 4 Soul 4d: a skipped entry is relayed as the primary sent it; probes, wire cases and the closed-enum rule committed F1: RecordUnappliedEntry keeps the primary's wire entry on the in-memory log entry (internal Relayed) and ToLogEntryMessage returns it through the same normalisation the store uses, so a replica with no store no longer relays a skipped put as a delete; memory and disk relay identically. Soul's 4d probes are committed (chained replica keeps the row, key-prefixed reads serve only owned keys, primary-all serves everything; the two measurements are explicit) with a memory/disk relay-parity test. Rust: captured-bytes cases for unowned_schema and variant_unsupported (with and without a subscription). The error-code set is stated as a closed enum in the schema description (both copies) and in cultnet-distributed-database.md: adding a code is one coordinated change across runtimes, since Rust and TypeScript refuse an unknown code. The Soul 4c probe file's header says what it guards. Co-Authored-By: Claude Sonnet 5.5 --- .../cultnet/cultnet-distributed-database.md | 10 + contracts/cultnet/cultnet.error.schema.json | 2 +- .../cultnet/cultnet.error.schema.json | 2 +- packages/cultnet-rs/tests/error_message.rs | 40 ++++ src/GameCult.Networking/CultNetDatabase.cs | 11 +- .../SoulP4cProbeTests.cs | 4 +- .../SoulP4dProbeTests.cs | 201 ++++++++++++++++++ 7 files changed, 266 insertions(+), 4 deletions(-) create mode 100644 tests/GameCult.Networking.Tests/SoulP4dProbeTests.cs diff --git a/contracts/cultnet/cultnet-distributed-database.md b/contracts/cultnet/cultnet-distributed-database.md index 8d2172c0..66bb450e 100644 --- a/contracts/cultnet/cultnet-distributed-database.md +++ b/contracts/cultnet/cultnet-distributed-database.md @@ -104,6 +104,16 @@ Not implemented yet: - cross-runtime rollback/resimulation helpers for simulation frames - declared CRDT merge policies +## Error Codes + +`cultnet.error.v0` carries an optional machine-readable `code`. The set of codes is a closed enum +(`cultnet.error.schema.json`): Rust and TypeScript refuse a message whose code they do not know, they do +not guess. Adding a code is therefore a coordinated change across every runtime, made together: the schema (both +copies), the C# `CultNetErrorMessage` factory, the Rust `CultNetErrorCode`, and a captured-bytes case in +`packages/cultnet-rs/tests/error_message.rs` for it. TypeScript reads the schema. The current codes are +`selection_invalid`, `cursor_stale`, `cursor_invalid`, `reference_outside_target`, `variant_unsupported` and +`unowned_schema` (a write, or a remote put or delete, for a schema and key no shard owns). + ## Live Invariants - CultCache owns document schema, identity, local indexes, and local diffing. diff --git a/contracts/cultnet/cultnet.error.schema.json b/contracts/cultnet/cultnet.error.schema.json index 0d541262..59764338 100644 --- a/contracts/cultnet/cultnet.error.schema.json +++ b/contracts/cultnet/cultnet.error.schema.json @@ -26,7 +26,7 @@ } }, "code": { - "description": "Machine-readable refusal code (R-N). Additive: a v0 peer that does not know this field ignores it.", + "description": "Machine-readable refusal code (R-N). Additive: a v0 peer that does not know this field ignores it. The set is a closed enum: Rust and TypeScript refuse a message whose code is not listed, so adding a code is one coordinated change across every runtime (this schema, the C# factory, the Rust CultNetErrorCode, and a captured-bytes case for each).", "type": ["string", "null"], "enum": [null, "selection_invalid", "cursor_stale", "cursor_invalid", "reference_outside_target", "variant_unsupported", "unowned_schema"] }, diff --git a/packages/cultnet-rs/contracts/cultnet/cultnet.error.schema.json b/packages/cultnet-rs/contracts/cultnet/cultnet.error.schema.json index 0d541262..59764338 100644 --- a/packages/cultnet-rs/contracts/cultnet/cultnet.error.schema.json +++ b/packages/cultnet-rs/contracts/cultnet/cultnet.error.schema.json @@ -26,7 +26,7 @@ } }, "code": { - "description": "Machine-readable refusal code (R-N). Additive: a v0 peer that does not know this field ignores it.", + "description": "Machine-readable refusal code (R-N). Additive: a v0 peer that does not know this field ignores it. The set is a closed enum: Rust and TypeScript refuse a message whose code is not listed, so adding a code is one coordinated change across every runtime (this schema, the C# factory, the Rust CultNetErrorCode, and a captured-bytes case for each).", "type": ["string", "null"], "enum": [null, "selection_invalid", "cursor_stale", "cursor_invalid", "reference_outside_target", "variant_unsupported", "unowned_schema"] }, diff --git a/packages/cultnet-rs/tests/error_message.rs b/packages/cultnet-rs/tests/error_message.rs index 011ff147..5c633087 100644 --- a/packages/cultnet-rs/tests/error_message.rs +++ b/packages/cultnet-rs/tests/error_message.rs @@ -118,6 +118,46 @@ fn cases() -> Vec { details: None, }, }, + Case { + name: "variant_unsupported_subscription_ended", + // C#: ForVariantUnsupported("variant unsupported here", "sub-1") + csharp_hex: "85AD736368656D6156657273696F6EB063756C746E65742E6572726F722E7630A56572726F72B876617269616E7420756E737570706F727465642068657265AB726F7574696E6748696E74C0A4636F6465B376617269616E745F756E737570706F72746564A764657461696C7384A56669656C64AE737562736372697074696F6E4964A576616C7565A57375622D31A461734F66C0A763757272656E74C0", + message: || CultNetMessage::Error { + error: "variant unsupported here".to_string(), + code: Some(CultNetErrorCode::VariantUnsupported), + details: Some(CultNetErrorDetails { + field: Some("subscriptionId".to_string()), + value: Some("sub-1".to_string()), + as_of: None, + current: None, + }), + }, + }, + Case { + name: "variant_unsupported_no_subscription", + // C#: ForVariantUnsupported("variant unsupported here") - no subscription ended, so Details stays null. + csharp_hex: "85AD736368656D6156657273696F6EB063756C746E65742E6572726F722E7630A56572726F72B876617269616E7420756E737570706F727465642068657265AB726F7574696E6748696E74C0A4636F6465B376617269616E745F756E737570706F72746564A764657461696C73C0", + message: || CultNetMessage::Error { + error: "variant unsupported here".to_string(), + code: Some(CultNetErrorCode::VariantUnsupported), + details: None, + }, + }, + Case { + name: "unowned_schema", + // C#: ForUnownedSchema(new CultNetUnownedSchemaException("some.schema.v1", new CultRecordKey("some:key"))) + csharp_hex: "85AD736368656D6156657273696F6EB063756C746E65742E6572726F722E7630A56572726F72D95C756E6F776E65645F736368656D613A204E6F207368617264206F776E7320736368656D612027736F6D652E736368656D612E763127206B65792027736F6D653A6B6579273B207468652077726974652077617320726566757365642EAB726F7574696E6748696E74C0A4636F6465AE756E6F776E65645F736368656D61A764657461696C7384A56669656C64A8736368656D614964A576616C7565AE736F6D652E736368656D612E7631A461734F66C0A763757272656E74C0", + message: || CultNetMessage::Error { + error: "unowned_schema: No shard owns schema 'some.schema.v1' key 'some:key'; the write was refused.".to_string(), + code: Some(CultNetErrorCode::UnownedSchema), + details: Some(CultNetErrorDetails { + field: Some("schemaId".to_string()), + value: Some("some.schema.v1".to_string()), + as_of: None, + current: None, + }), + }, + }, ] } diff --git a/src/GameCult.Networking/CultNetDatabase.cs b/src/GameCult.Networking/CultNetDatabase.cs index bedf91c6..668c088e 100644 --- a/src/GameCult.Networking/CultNetDatabase.cs +++ b/src/GameCult.Networking/CultNetDatabase.cs @@ -437,6 +437,10 @@ namespace GameCult.Networking /// Gets the previous document, when present. /// public object? PreviousDocument { get; } + + // The primary's own wire entry, kept for an entry this replica changed nothing for (a row its shard does not own, an + // absent key): it is relayed downstream exactly as the primary sent it, in memory and on disk alike. + internal CultNetShardLogEntryMessage? Relayed { get; set; } } /// @@ -1819,13 +1823,18 @@ namespace GameCult.Networking { RecordMutationLogEntry(new CultNetShardMutationLogEntry( shard.ShardId, shard.Epoch, entry.Sequence, entry.CommittedAt, kind, - schemaId, key, document: null, previousDocument: null), entry); + schemaId, key, document: null, previousDocument: null) { Relayed = entry }, entry); _lastWriteSequence[(schemaId, key.Value)] = entry.Sequence; } } private CultNetShardLogEntryMessage ToLogEntryMessage(CultNetShardMutationLogEntry entry) { + if (entry.Relayed != null) + { + return NormalizeWireLogEntry(entry, entry.Relayed); + } + if (entry.Kind == CultNetDatabaseChangeKind.Removed || entry.Document == null) { return new CultNetShardLogEntryMessage diff --git a/tests/GameCult.Networking.Tests/SoulP4cProbeTests.cs b/tests/GameCult.Networking.Tests/SoulP4cProbeTests.cs index 98956d54..a4796e7d 100644 --- a/tests/GameCult.Networking.Tests/SoulP4cProbeTests.cs +++ b/tests/GameCult.Networking.Tests/SoulP4cProbeTests.cs @@ -11,7 +11,9 @@ using static GameCult.Networking.Tests.NetworkingTests; namespace GameCult.Networking.Tests { - // Soul pass 4c scratch probes (not for merge). + // Soul pass 4c probes. They guard: a committed put whose shard log failed is not answered as a refusal (Q1); reads and watch agree + // on what a database serves (Q2); a replica log apply leaves out a row its shard does not own (Q3); the unowned refusal is not a + // shard-authority exception (Q4); two databases over one cache aggregate their log failures (Q5). public sealed class SoulP4cProbeTests { private const string ShardId = "soulc-shard"; diff --git a/tests/GameCult.Networking.Tests/SoulP4dProbeTests.cs b/tests/GameCult.Networking.Tests/SoulP4dProbeTests.cs new file mode 100644 index 00000000..6623f8a1 --- /dev/null +++ b/tests/GameCult.Networking.Tests/SoulP4dProbeTests.cs @@ -0,0 +1,201 @@ +#nullable enable +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.IO; +using System.Linq; +using System.Threading.Tasks; +using GameCult.Caching; +using NUnit.Framework; +using static GameCult.Networking.Tests.NetworkingTests; + +namespace GameCult.Networking.Tests +{ + // Soul pass 4d probes. They guard the replica log relay (a skipped entry is relayed as the primary sent it, in memory and on + // disk), key-prefixed shard reads, and measure two costs; the measurements are explicit. + public sealed class SoulP4dProbeTests + { + private const string ShardId = "soupd-shard"; + private static readonly string Now = DateTimeOffset.UtcNow.ToString("O"); + + private static string NoteSchema(CultCache cache) => cache.Registry.GetRequired().SchemaId; + private static string MeshSchema(CultCache cache) => cache.Registry.GetRequired().SchemaId; + + private static CultNetDatabase Db(CultCache cache, bool primary, string[] schemas, string? keyPrefix = null, ICultNetShardMutationLogStore? store = null) => + new(cache, new CultNetDatabaseOptions + { + Shards = [new CultNetShardDescriptor(ShardId, "primary", epoch: 1, isPrimary: primary, schemaIds: schemas, keyPrefix: keyPrefix)], + MutationLogStore = store + }); + + private static CultNetShardLogResponseMessage UnownedPutLog(CultNetDatabase anyDb, string key) + { + var put = anyDb.Documents.CreateRawDocumentPutMessage("wire", new CultRecordHandle(new CultRecordKey(key)), new MeshQuickstartNote { NoteId = "p", Body = "primary-put" }); + put.ShardId = ShardId; + put.ShardEpoch = 1; + return new CultNetShardLogResponseMessage + { + MessageId = "log", + ShardId = ShardId, + ShardEpoch = 1, + Entries = [new CultNetShardLogEntryMessage { Sequence = 1, CommittedAt = Now, ChangeKind = "added", Put = put }] + }; + } + + // Q3a: a replica (no store) that skips an unowned put relays it to a chained replica as a delete. + [Test, Category("SoulFinding")] + public async Task Q3a_MemoryLogRelaysASkippedPutAsARemoval() + { + var cache = new CultCache(); + var replica = Db(cache, primary: false, [NoteSchema(cache)]); + await replica.ApplyShardLogResponseAsync(UnownedPutLog(replica, "relay:k")); + var relayed = replica.GetMutationLogMessages(ShardId); + TestContext.Out.WriteLine($"memory relay: kind={relayed[0].ChangeKind} put={relayed[0].Put != null} delete={relayed[0].Delete != null}"); + Assert.That(relayed[0].ChangeKind, Is.EqualTo("added"), "the relayed entry keeps the primary's kind"); + } + + // Q3b: the same replica with a durable store relays the primary's put (disk and memory disagree). + [Test, Category("SoulFinding")] + public async Task Q3b_StoreBackedLogRelaysTheSkippedPutAsAPut() + { + var dir = Path.Combine(Path.GetTempPath(), "soulp4d-" + Guid.NewGuid().ToString("N")); + var cache = new CultCache(); + var replica = Db(cache, primary: false, [NoteSchema(cache)], store: new CultNetFileShardMutationLogStore(dir)); + await replica.ApplyShardLogResponseAsync(UnownedPutLog(replica, "relay:k")); + var relayed = replica.GetMutationLogMessages(ShardId); + var memory = replica.GetMutationLog(ShardId); + TestContext.Out.WriteLine($"store relay: kind={relayed[0].ChangeKind} put={relayed[0].Put != null}; memory entry kind={memory[0].Kind} doc={(memory[0].Document == null ? "null" : "set")}"); + Assert.That(relayed[0].ChangeKind, Is.EqualTo("added")); + } + + // Q3c: a chained replica that does own the row (config drift) and holds it deletes it on the relayed entry. + [Test, Category("SoulFinding")] + public async Task Q3c_AChainedReplicaThatOwnsTheRowDeletesWhatThePrimaryPut() + { + var cacheA = new CultCache(); + var a = Db(cacheA, primary: false, [NoteSchema(cacheA)]); + await a.ApplyShardLogResponseAsync(UnownedPutLog(a, "relay:k")); + + var cacheC = new CultCache(); + var c = Db(cacheC, primary: false, [NoteSchema(cacheC), MeshSchema(cacheC)]); + await cacheC.UpsertAsync(new MeshQuickstartNote { NoteId = "p", Body = "earlier" }, new CultRecordHandle(new CultRecordKey("relay:k"))); + await c.ApplyShardLogResponseAsync(new CultNetShardLogResponseMessage + { + MessageId = "chain", + ShardId = ShardId, + ShardEpoch = 1, + Entries = a.GetMutationLogMessages(ShardId).ToArray() + }); + var row = cacheC.Get(new CultRecordKey("relay:k")); + TestContext.Out.WriteLine($"chained row after relay: {row?.Body ?? ""}"); + Assert.That(row, Is.Not.Null, "the primary put this row; a relay must not delete it"); + } + + // A skipped entry is relayed exactly as the primary sent it, whether the replica keeps its log in memory or on disk. + [Test] + public async Task SkippedEntriesAreRelayedIdenticallyFromMemoryAndFromDisk() + { + var dir = Path.Combine(Path.GetTempPath(), "soulp4d-parity-" + Guid.NewGuid().ToString("N")); + try + { + var memoryCache = new CultCache(); + var inMemory = Db(memoryCache, primary: false, [NoteSchema(memoryCache)]); + var diskCache = new CultCache(); + var onDisk = Db(diskCache, primary: false, [NoteSchema(diskCache)], store: new CultNetFileShardMutationLogStore(dir)); + var log = UnownedPutLog(inMemory, "relay:k"); + var removal = new CultNetShardLogEntryMessage + { + Sequence = 2, + CommittedAt = Now, + ChangeKind = "removed", + Delete = new CultNetDocumentDeleteMessage + { + MessageId = "d", + SchemaId = memoryCache.Registry.GetRequired().SchemaId, + RecordKey = "relay:k", + ShardId = ShardId, + ShardEpoch = 1 + } + }; + log.Entries = [log.Entries[0], removal]; + + await inMemory.ApplyShardLogResponseAsync(log); + await onDisk.ApplyShardLogResponseAsync(log); + + var memory = inMemory.GetMutationLogMessages(ShardId); + var disk = onDisk.GetMutationLogMessages(ShardId); + Assert.That(memory.Select(e => e.ChangeKind), Is.EqualTo(new[] { "added", "removed" })); + Assert.That(memory.Select(e => e.ChangeKind), Is.EqualTo(disk.Select(e => e.ChangeKind))); + Assert.That(memory.Select(e => e.Put != null), Is.EqualTo(new[] { true, false })); + Assert.That(memory.Select(e => e.Put != null), Is.EqualTo(disk.Select(e => e.Put != null))); + Assert.That(memory.Select(e => e.Delete != null), Is.EqualTo(new[] { false, true })); + Assert.That(memory.Select(e => e.Delete != null), Is.EqualTo(disk.Select(e => e.Delete != null))); + Assert.That(memory[0].Put!.Document!.RecordKey, Is.EqualTo(disk[0].Put!.Document!.RecordKey)); + Assert.That(memory[0].Put!.Document!.Payload, Is.EqualTo(disk[0].Put!.Document!.Payload), "the document bytes are the primary's"); + } + finally + { + if (Directory.Exists(dir)) Directory.Delete(dir, true); + } + } + + // Q2: key-prefixed shard: the schema is owned, a key outside the prefix is not. + [Test, Category("SoulPin")] + public async Task Q2_KeyPrefixedReadsServeOnlyTheOwnedKeys() + { + var cache = new CultCache(); + var database = Db(cache, primary: true, [NoteSchema(cache)], keyPrefix: "a:"); + await cache.UpsertAsync(new NetworkSchemaNote { Text = "a" }, new CultRecordHandle(new CultRecordKey("a:1"))); + await cache.UpsertAsync(new NetworkSchemaNote { Text = "b" }, new CultRecordHandle(new CultRecordKey("b:1"))); + Assert.That(database.GetAll().Select(n => n.Text), Is.EqualTo(new[] { "a" })); + Assert.That(await database.GetAsync(new CultRecordKey("b:1")), Is.Null); + Assert.That(await database.GetAsync(new CultRecordKey("a:1")), Is.Not.Null); + using var server = new Server(cache, ServerSecurityOptions.Development()); + using var databaseServer = new CultNetDatabaseServer(server, database); + var snapshot = databaseServer.CreateSnapshotResponse(new CultNetSnapshotRequestMessage { MessageId = "all" }); + Assert.That(snapshot.Documents.Select(d => d.RecordKey), Is.EqualTo(new[] { "a:1" })); + } + + [Test, Category("SoulPin")] + public async Task Q2_PrimaryAllServesEveryCachedRow() + { + var cache = new CultCache(); + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions { Shards = [CultNetShardDescriptor.PrimaryAll()] }); + await cache.UpsertAsync(new MeshQuickstartNote { NoteId = "x", Body = "x" }, new CultRecordHandle(new CultRecordKey("any:1"))); + Assert.That(database.GetAll().Count(), Is.EqualTo(1)); + Assert.That(database.GetByName("x"), Is.Not.Null); + } + + // Q2 cost: GetAll through the database versus the cache, scoped shard (reflection per row). + [Test, Category("SoulMeasure"), Explicit("Measurement, no assertion.")] + public void Q2_GetAllCost() + { + var cache = new CultCache(); + var database = Db(cache, primary: false, [NoteSchema(cache)], keyPrefix: "n:"); + cache.Commit(batch => + { + for (var i = 0; i < 20000; i++) + batch.Upsert(new NetworkSchemaNote { Text = "t" + i }, new CultRecordHandle(new CultRecordKey("n:" + i))); + }); + _ = cache.GetAll().Count(); + _ = database.GetAll().Count(); + var sw = Stopwatch.StartNew(); + for (var r = 0; r < 5; r++) _ = cache.GetAll().Count(); + var cacheMs = sw.Elapsed.TotalMilliseconds / 5; + sw.Restart(); + for (var r = 0; r < 5; r++) _ = database.GetAll().Count(); + var dbMs = sw.Elapsed.TotalMilliseconds / 5; + TestContext.Out.WriteLine($"GetAll 20000 rows: cache={cacheMs:F2}ms database={dbMs:F2}ms ratio={dbMs / Math.Max(cacheMs, 0.001):F1}x"); + } + + // Wire: the bytes a peer receives for the unowned refusal. + [Test, Category("SoulMeasure"), Explicit("Prints the wire bytes the Rust captured-bytes case pins.")] + public void Wire_UnownedSchemaBytes() + { + var message = CultNetErrorMessage.ForUnownedSchema(new CultNetUnownedSchemaException("some.schema.v1", new CultRecordKey("some:key"))); + var bytes = MessagePack.MessagePackSerializer.Serialize(message, CultNetSchemaMessageSerialization.Options); + TestContext.Out.WriteLine("HEX=" + Convert.ToHexString(bytes)); + TestContext.Out.WriteLine("JSON=" + MessagePack.MessagePackSerializer.ConvertToJson(bytes)); + } + } +}