diff --git a/contracts/cultnet/cultnet.error.schema.json b/contracts/cultnet/cultnet.error.schema.json index 15880fa1..0d541262 100644 --- a/contracts/cultnet/cultnet.error.schema.json +++ b/contracts/cultnet/cultnet.error.schema.json @@ -28,7 +28,7 @@ "code": { "description": "Machine-readable refusal code (R-N). Additive: a v0 peer that does not know this field ignores it.", "type": ["string", "null"], - "enum": [null, "selection_invalid", "cursor_stale", "cursor_invalid", "reference_outside_target", "variant_unsupported"] + "enum": [null, "selection_invalid", "cursor_stale", "cursor_invalid", "reference_outside_target", "variant_unsupported", "unowned_schema"] }, "details": { "description": "Code-specific structured detail (R-N). Every field is populated only for the code that defines it; the others are null, not omitted.", @@ -36,8 +36,8 @@ "additionalProperties": false, "required": ["field", "value", "asOf", "current"], "properties": { - "field": { "description": "selection_invalid: the selection field that failed. variant_unsupported: \"subscriptionId\", when a subscription was ended.", "type": ["string", "null"] }, - "value": { "description": "selection_invalid: the offending value, when there is one string worth naming. variant_unsupported: the id of the subscription that ended.", "type": ["string", "null"] }, + "field": { "description": "selection_invalid: the selection field that failed. variant_unsupported: \"subscriptionId\", when a subscription was ended. unowned_schema: \"schemaId\".", "type": ["string", "null"] }, + "value": { "description": "selection_invalid: the offending value, when there is one string worth naming. variant_unsupported: the id of the subscription that ended. unowned_schema: the schema id no shard owns.", "type": ["string", "null"] }, "asOf": { "description": "cursor_stale: the cursor's own minted asOf.", "type": ["integer", "null"], "minimum": 0 }, "current": { "description": "cursor_stale: the answering server's current asOf.", "type": ["integer", "null"], "minimum": 0 } } diff --git a/packages/cultnet-rs/contracts/cultnet/cultnet.error.schema.json b/packages/cultnet-rs/contracts/cultnet/cultnet.error.schema.json index 15880fa1..0d541262 100644 --- a/packages/cultnet-rs/contracts/cultnet/cultnet.error.schema.json +++ b/packages/cultnet-rs/contracts/cultnet/cultnet.error.schema.json @@ -28,7 +28,7 @@ "code": { "description": "Machine-readable refusal code (R-N). Additive: a v0 peer that does not know this field ignores it.", "type": ["string", "null"], - "enum": [null, "selection_invalid", "cursor_stale", "cursor_invalid", "reference_outside_target", "variant_unsupported"] + "enum": [null, "selection_invalid", "cursor_stale", "cursor_invalid", "reference_outside_target", "variant_unsupported", "unowned_schema"] }, "details": { "description": "Code-specific structured detail (R-N). Every field is populated only for the code that defines it; the others are null, not omitted.", @@ -36,8 +36,8 @@ "additionalProperties": false, "required": ["field", "value", "asOf", "current"], "properties": { - "field": { "description": "selection_invalid: the selection field that failed. variant_unsupported: \"subscriptionId\", when a subscription was ended.", "type": ["string", "null"] }, - "value": { "description": "selection_invalid: the offending value, when there is one string worth naming. variant_unsupported: the id of the subscription that ended.", "type": ["string", "null"] }, + "field": { "description": "selection_invalid: the selection field that failed. variant_unsupported: \"subscriptionId\", when a subscription was ended. unowned_schema: \"schemaId\".", "type": ["string", "null"] }, + "value": { "description": "selection_invalid: the offending value, when there is one string worth naming. variant_unsupported: the id of the subscription that ended. unowned_schema: the schema id no shard owns.", "type": ["string", "null"] }, "asOf": { "description": "cursor_stale: the cursor's own minted asOf.", "type": ["integer", "null"], "minimum": 0 }, "current": { "description": "cursor_stale: the answering server's current asOf.", "type": ["integer", "null"], "minimum": 0 } } diff --git a/packages/cultnet-rs/src/contracts.rs b/packages/cultnet-rs/src/contracts.rs index f31ade85..1808c62f 100644 --- a/packages/cultnet-rs/src/contracts.rs +++ b/packages/cultnet-rs/src/contracts.rs @@ -250,8 +250,8 @@ impl CultNetShardDescriptor { } } -/// Machine-readable `cultnet.error.v0` refusal code (R-N). Mirrors the five strings -/// `contracts/cultnet/cultnet.error.schema.json` enumerates for `code`; a peer that receives a fifth +/// Machine-readable `cultnet.error.v0` refusal code (R-N). Mirrors the strings +/// `contracts/cultnet/cultnet.error.schema.json` enumerates for `code`; a peer that receives a /// string it does not know refuses to decode rather than guess a meaning for it. #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] @@ -261,6 +261,7 @@ pub enum CultNetErrorCode { CursorInvalid, ReferenceOutsideTarget, VariantUnsupported, + UnownedSchema, } impl CultNetErrorCode { @@ -271,6 +272,7 @@ impl CultNetErrorCode { Self::CursorInvalid => "cursor_invalid", Self::ReferenceOutsideTarget => "reference_outside_target", Self::VariantUnsupported => "variant_unsupported", + Self::UnownedSchema => "unowned_schema", } } @@ -281,6 +283,7 @@ impl CultNetErrorCode { "cursor_invalid" => Ok(Self::CursorInvalid), "reference_outside_target" => Ok(Self::ReferenceOutsideTarget), "variant_unsupported" => Ok(Self::VariantUnsupported), + "unowned_schema" => Ok(Self::UnownedSchema), other => Err(anyhow!("ErrorMessage.Code {other:?} is not a recognized cultnet.error.v0 code")), } } diff --git a/src/GameCult.Caching/CultCache.cs b/src/GameCult.Caching/CultCache.cs index 68fbfb1b..6365261f 100644 --- a/src/GameCult.Caching/CultCache.cs +++ b/src/GameCult.Caching/CultCache.cs @@ -2124,10 +2124,15 @@ namespace GameCult.Caching Remove(handle.Key); } - // false: a condition failed; nothing was written, changed in memory, or published. A commit that succeeds and then - // fails in a journal or observer still stands and is published; the writer receives the failure. One failure arrives - // as itself, several as an AggregateException. A writer that wants a particular failure (CultNetShardLogException, when - // a database's log refused the commit) catches that type, or unwraps AggregateException.InnerExceptions. + /// + /// Commits the staged batch. Returns false when a condition failed: nothing was written, changed in memory, or published. + /// + /// + /// A commit that succeeds and then fails in a journal or observer still stands and is published, and the writer + /// receives the failure. One failure arrives as itself, several as an . A writer that + /// wants a particular failure (a CultNetShardLogException, when a database's log refused the commit) catches that type, + /// or unwraps and looks for it. + /// public bool Commit(Action stage) { return Land(stage, wait: true) == CultCommitOutcome.Committed; diff --git a/src/GameCult.Mesh/CultMesh.cs b/src/GameCult.Mesh/CultMesh.cs index 52c67482..bdc1dbdc 100644 --- a/src/GameCult.Mesh/CultMesh.cs +++ b/src/GameCult.Mesh/CultMesh.cs @@ -3370,6 +3370,20 @@ namespace GameCult.Mesh $"CultMesh document '{key.Value}' was not found as {typeof(TDocument).FullName}."); } + // A database serves only what its shards own; the cache behind it may hold more. + private static bool DatabaseServes(CultNetDatabase database, object document, CultRecordKey key) + { + try + { + database.ResolveShard(CultDocumentRegistry.Shared.GetRequired(document.GetType()).SchemaId, key); + return true; + } + catch (CultNetUnownedSchemaException) + { + return false; + } + } + private static async Task ReadDatabaseDocumentRequiredAsync( CultNetDatabase database, CultRecordKey key) @@ -3380,7 +3394,7 @@ namespace GameCult.Mesh return document; var untyped = database.Cache.Get(key); - if (untyped != null && IsSameCultDocumentSchema(untyped.GetType())) + if (untyped != null && IsSameCultDocumentSchema(untyped.GetType()) && DatabaseServes(database, untyped, key)) return ConvertUntypedDocument(untyped); throw new KeyNotFoundException( diff --git a/src/GameCult.Networking/CultNetDatabase.cs b/src/GameCult.Networking/CultNetDatabase.cs index 07336488..1cb1a1e0 100644 --- a/src/GameCult.Networking/CultNetDatabase.cs +++ b/src/GameCult.Networking/CultNetDatabase.cs @@ -920,16 +920,18 @@ namespace GameCult.Networking public Task GetAsync(CultRecordKey key) where T : class { ThrowIfDisposed(); - return Task.FromResult(_cache.Get(key)); + var document = _cache.Get(key); + return Task.FromResult(document != null && Owns(_cache.Registry.GetRequired(document.GetType()), key) ? document : null); } /// - /// Gets all documents assignable to the requested type. + /// Gets all documents assignable to the requested type. A database serves only what its shards own: a cached row no + /// shard owns is left out here, as it is from every read and projection. The cache itself still holds it. /// public IEnumerable GetAll() where T : class { ThrowIfDisposed(); - return _cache.GetAll(); + return _cache.GetAll().Where(Serves); } /// @@ -938,7 +940,8 @@ namespace GameCult.Networking public T? GetByName(string name) where T : class { ThrowIfDisposed(); - return _cache.GetByName(name); + var document = _cache.GetByName(name); + return document != null && Serves(document) ? document : null; } /// @@ -947,12 +950,15 @@ namespace GameCult.Networking public T? GetByIndex(string alias, string value) where T : class { ThrowIfDisposed(); - return _cache.GetByIndex(alias, value); + var document = _cache.GetByIndex(alias, value); + return document != null && Serves(document) ? document : null; } /// /// Adds or replaces a document at a specific key. /// + /// No shard owns the schema and key; nothing was written. + /// The change committed and was published, but the shard log could not record it. public async Task> PutAsync(CultRecordKey key, T document) where T : class { ThrowIfDisposed(); @@ -967,6 +973,7 @@ namespace GameCult.Networking /// /// Applies a locally predicted document for a client-owned input scope. /// + /// No shard owns the schema and key; nothing was written. public async Task> PutPredictedAsync(CultRecordKey key, T document) where T : class { ThrowIfDisposed(); @@ -991,6 +998,8 @@ namespace GameCult.Networking /// /// Deletes a document by key. /// + /// No shard owns the schema and key; nothing was written. + /// The change committed and was published, but the shard log could not record it. public Task DeleteAsync(CultRecordKey key) where T : class { ThrowIfDisposed(); @@ -1004,6 +1013,8 @@ namespace GameCult.Networking /// /// Applies a raw document mutation after checking shard authority. /// + /// No shard owns the schema and key; nothing was written. + /// The change committed and was published, but the shard log could not record it. public async Task ApplyPutAsync(CultNetDocumentPutRawMessage message) { ThrowIfDisposed(); @@ -1026,6 +1037,8 @@ namespace GameCult.Networking /// /// Applies a raw document delete after checking shard authority. /// + /// No shard owns the schema and key; nothing was written. + /// The change committed and was published, but the shard log could not record it. public Task ApplyDeleteAsync(CultNetDocumentDeleteMessage message) { ThrowIfDisposed(); @@ -1729,6 +1742,12 @@ namespace GameCult.Networking // The replica's log records the primary's own entry under the primary's own sequence, in the same hold that // admits the change; the journal does both. A retry after a refused append is a fresh instance, so it matches // its door again and re-records the entry. + if (!shard.Matches(descriptor, key)) + { + RecordUnappliedEntry(shard, kind, descriptor.SchemaId, key, entry); + return; + } + var document = _documents.DeserializeRawDocument(message.Document); using (Door(document, new DoorContext(shard) { Kind = kind, Replica = entry, PredictionKey = PredictionKey(descriptor.SchemaId, key) })) await _cache.UpsertAsync(descriptor.DocumentType, document, key).ConfigureAwait(false); @@ -1750,6 +1769,12 @@ namespace GameCult.Networking var key = new CultRecordKey(message.RecordKey); var descriptor = _cache.Registry.GetRequiredBySchemaId(message.SchemaId); + if (!shard.Matches(descriptor, key)) + { + RecordUnappliedEntry(shard, CultNetDatabaseChangeKind.Removed, descriptor.SchemaId, key, entry); + return; + } + var admitted = false; var previous = _cache.Get(key); if (previous != null) @@ -1761,15 +1786,26 @@ namespace GameCult.Networking if (!admitted) { - // The key is not here, so nothing was admitted to log. This replica's log must still hold every entry of the - // primary's, or a replica chained on it sees a gap. - lock (_logGate) - { - RecordMutationLogEntry(new CultNetShardMutationLogEntry( - shard.ShardId, shard.Epoch, entry.Sequence, entry.CommittedAt, CultNetDatabaseChangeKind.Removed, - descriptor.SchemaId, key, document: null, previousDocument: null), entry); - _lastWriteSequence[(descriptor.SchemaId, key.Value)] = entry.Sequence; - } + // The key is not here, so nothing was admitted to log. + RecordUnappliedEntry(shard, CultNetDatabaseChangeKind.Removed, descriptor.SchemaId, key, entry); + } + } + + // An entry that changed nothing here (an absent key, or a row this shard does not own - the snapshot path leaves such + // rows out too). This replica's log must still hold every entry of the primary's, or a replica chained on it sees a gap. + private void RecordUnappliedEntry( + CultNetShardDescriptor shard, + CultNetDatabaseChangeKind kind, + string schemaId, + CultRecordKey key, + CultNetShardLogEntryMessage entry) + { + lock (_logGate) + { + RecordMutationLogEntry(new CultNetShardMutationLogEntry( + shard.ShardId, shard.Epoch, entry.Sequence, entry.CommittedAt, kind, + schemaId, key, document: null, previousDocument: null), entry); + _lastWriteSequence[(schemaId, key.Value)] = entry.Sequence; } } @@ -1924,6 +1960,19 @@ namespace GameCult.Networking internal bool Owns(CultDocumentDescriptor descriptor, CultRecordKey key) => FindShard(descriptor, key) != null; + // Whether the database serves this cached instance: a shard owns its schema and key. A shard that owns everything makes + // the lookup of the instance's key unnecessary. + private bool Serves(object document) + { + if (_shards.Any(shard => shard.SchemaIds.Count == 0 && string.IsNullOrEmpty(shard.KeyPrefix))) + { + return true; + } + + var type = document.GetType(); + return Owns(_cache.Registry.GetRequired(type), GetTrackedKey(document, type)); + } + private CultNetShardDescriptor? FindShard(CultDocumentDescriptor descriptor, CultRecordKey key) => _shards.FirstOrDefault(shard => shard.Matches(descriptor, key)); diff --git a/src/GameCult.Networking/CultNetDatabaseServer.cs b/src/GameCult.Networking/CultNetDatabaseServer.cs index 78124acc..77b03070 100644 --- a/src/GameCult.Networking/CultNetDatabaseServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseServer.cs @@ -112,10 +112,13 @@ namespace GameCult.Networking request); } + // A database serves only what its shards own. return _database.Documents.CreateRawSnapshotResponse( _database.Cache, string.IsNullOrWhiteSpace(request.MessageId) ? Guid.NewGuid().ToString("N") : request.MessageId, - request); + request, + options: null, + rowFilter: _database.Owns); } /// @@ -136,6 +139,7 @@ namespace GameCult.Networking // one shard's log is refused rather than answered against a watermark that is not // exact for all of them. // A row whose schema no shard owns is not part of what this database serves: it is left out, not an error. + options: null, rowFilter: _database.Owns, shardIdOf: (schemaId, key) => _database.ResolveShard(schemaId, key).ShardId, cursorKey: _database.CursorKey, @@ -314,6 +318,16 @@ namespace GameCult.Networking peer.SendCultNet(CreateRoutingError(ex)); } } + catch (CultNetUnownedSchemaException ex) + { + _server.Logger.LogWarning($"CultNet raw put refused: {ex.Message}"); + peer.SendCultNet(CultNetErrorMessage.ForUnownedSchema(ex)); + } + catch (Exception ex) when (CommittedWithLogFailure(ex)) + { + // The write committed and was published; only its shard log failed, so the peer is not told it was refused. + _server.Logger.LogWarning($"CultNet raw put committed but its shard log failed: {ex.Message}"); + } catch (Exception ex) { _server.Logger.LogError($"CultNet raw put failed: {ex.Message}"); @@ -334,6 +348,15 @@ namespace GameCult.Networking peer.SendCultNet(CreateRoutingError(ex)); } } + catch (CultNetUnownedSchemaException ex) + { + _server.Logger.LogWarning($"CultNet raw delete refused: {ex.Message}"); + peer.SendCultNet(CultNetErrorMessage.ForUnownedSchema(ex)); + } + catch (Exception ex) when (CommittedWithLogFailure(ex)) + { + _server.Logger.LogWarning($"CultNet raw delete committed but its shard log failed: {ex.Message}"); + } catch (Exception ex) { peer.SendCultNet(new CultNetErrorMessage { Error = ex.Message }); @@ -606,6 +629,12 @@ namespace GameCult.Networking : messageId; } + // A CultNetShardLogException means the write committed and was published (the writer hears of a failed log afterwards); + // the cache wraps it in an AggregateException when something else failed in the same hold. + private static bool CommittedWithLogFailure(Exception exception) => + exception is CultNetShardLogException || + exception is AggregateException aggregate && aggregate.Flatten().InnerExceptions.Any(inner => inner is CultNetShardLogException); + private static CultNetErrorMessage CreateRoutingError(CultNetShardAuthorityException exception) { return new CultNetErrorMessage diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs index f3150740..7ae3c67e 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs @@ -421,7 +421,8 @@ namespace GameCult.Networking documentSelection, ordinalOf: (schemaId, key) => _database.LastWriteSequence(schemaId, key) ?? 0, asOf: _database.CurrentAsOf(), - options: null); + options: null, + rowFilter: _database.Owns); // The snapshot is one row per record key by construction, so the duplicate-key throw this // used to guard can no longer fire (docs/cultnet-selection-cut.md, S2-2); deleted rather // than tested around. diff --git a/src/GameCult.Networking/CultNetDocumentRegistry.cs b/src/GameCult.Networking/CultNetDocumentRegistry.cs index a47f28b3..62691a3e 100644 --- a/src/GameCult.Networking/CultNetDocumentRegistry.cs +++ b/src/GameCult.Networking/CultNetDocumentRegistry.cs @@ -280,7 +280,16 @@ namespace GameCult.Networking CultCache cache, string messageId, CultNetSnapshotRequestMessage? filter = null, - CultNetDocumentMessageOptions? options = null) + CultNetDocumentMessageOptions? options = null) => + CreateRawSnapshotResponse(cache, messageId, filter, options, rowFilter: null); + + // rowFilter leaves out the cached rows it rejects before evaluation (a database serves only what its shards own). + internal CultNetSnapshotResponseRawMessage CreateRawSnapshotResponse( + CultCache cache, + string messageId, + CultNetSnapshotRequestMessage? filter, + CultNetDocumentMessageOptions? options, + Func? rowFilter) { if (cache == null) throw new ArgumentNullException(nameof(cache)); var lowSchemas = CultNetV0SelectionLowering.Lower(filter?.SchemaIds); @@ -323,7 +332,7 @@ namespace GameCult.Networking ordinalOf = static (_, _) => 0; } - var page = SelectAll(cache, selection, ordinalOf, asOf: 0, options); + var page = SelectAll(cache, selection, ordinalOf, asOf: 0, options, rowFilter); return new CultNetSnapshotResponseRawMessage { @@ -338,7 +347,6 @@ namespace GameCult.Networking /// row's ordinal (the reference: ); /// 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). - /// , when supplied, leaves out the cached rows it rejects before evaluation. /// public CultNetSnapshotResponseRawV1Message CreateSelectionResponse( CultCache cache, @@ -349,8 +357,21 @@ namespace GameCult.Networking CultNetDocumentMessageOptions? options = null, Func? shardIdOf = null, CultNetSelectionCursorKey? cursorKey = null, - Func? asOfForShard = null, - Func? rowFilter = null) + Func? asOfForShard = null) => + CreateSelectionResponse(cache, messageId, selection, ordinalOf, asOf, options, shardIdOf, cursorKey, asOfForShard, rowFilter: null); + + // rowFilter leaves out the cached rows it rejects before evaluation (a database serves only what its shards own). + internal CultNetSnapshotResponseRawV1Message CreateSelectionResponse( + CultCache cache, + string messageId, + CultNetSelection selection, + Func ordinalOf, + ulong asOf, + CultNetDocumentMessageOptions? options, + Func? shardIdOf, + CultNetSelectionCursorKey? cursorKey, + Func? asOfForShard, + Func? rowFilter) { if (cache == null) throw new ArgumentNullException(nameof(cache)); if (selection == null) throw new ArgumentNullException(nameof(selection)); diff --git a/src/GameCult.Networking/CultNetSchemaMessages.cs b/src/GameCult.Networking/CultNetSchemaMessages.cs index 4daa76d5..e7f4cfd9 100644 --- a/src/GameCult.Networking/CultNetSchemaMessages.cs +++ b/src/GameCult.Networking/CultNetSchemaMessages.cs @@ -511,7 +511,7 @@ namespace GameCult.Networking [Key("routingHint")] public CultNetShardRoutingHint? RoutingHint { get; set; } /// /// Gets or sets the machine-readable refusal code (R-N), e.g. selection_invalid, - /// cursor_stale, cursor_invalid, reference_outside_target. Additive to + /// cursor_stale, cursor_invalid, reference_outside_target, unowned_schema. Additive to /// cultnet.error.v0; a v0 peer that does not know this field ignores it. /// [Key("code")] public string? Code { get; set; } @@ -532,6 +532,17 @@ namespace GameCult.Networking Details = new CultNetErrorDetails { Field = ex.Field, Value = ex.Value } }; + /// + /// Builds the wire error for a write no shard owns: code unowned_schema, details { field: "schemaId", value }. + /// + public static CultNetErrorMessage ForUnownedSchema(CultNetUnownedSchemaException ex) => + new CultNetErrorMessage + { + Error = $"unowned_schema: {ex.Message}", + Code = "unowned_schema", + Details = new CultNetErrorDetails { Field = "schemaId", Value = ex.SchemaId } + }; + /// /// Builds the wire error for a cursor refusal (R-N): code cursor_stale with details /// { asOf, current }, or code cursor_invalid with no details. diff --git a/tests/GameCult.Networking.Tests/CultNetJournalTests.cs b/tests/GameCult.Networking.Tests/CultNetJournalTests.cs index 3bbb72ef..7ba70dc8 100644 --- a/tests/GameCult.Networking.Tests/CultNetJournalTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetJournalTests.cs @@ -741,7 +741,8 @@ namespace GameCult.Networking.Tests try { await handler(UnownedPut(database), peer); } catch (Exception) { } - Assert.That(logger.Errors, Has.Some.Contains("CultNet raw put failed").And.Contains(unowned)); + Assert.That(logger.Warnings, Has.Some.Contains("raw put refused").And.Contains(unowned)); + Assert.That(logger.Errors, Is.Empty, "a refusal is not a fault"); Assert.That(cache.Get(Unowned), Is.Null); } @@ -766,11 +767,296 @@ namespace GameCult.Networking.Tests Assert.That(response.Matched, Is.EqualTo(1u)); } + [CultDocument("tests.journal_indexed", "tests.journal_indexed.v1")] + [MessagePack.MessagePackObject] + public sealed class JournalIndexedNote + { + [MessagePack.Key(0)] + [CultName] + public string Name { get; set; } = string.Empty; + + [MessagePack.Key(1)] + [CultIndex("tag")] + public string Tag { get; set; } = string.Empty; + } + + // A database serves only what its shards own: reads, the legacy and shard-less snapshot, selections, subscription + // snapshots and the watch agree. The cache behind it still holds the row. + [Test] + public async Task ADatabaseReadsServeOnlyWhatItsShardsOwnWhileTheCacheStillHoldsTheRest() + { + var (database, cache, _, _) = OwnedNothingOfNote(); + var seen = Record(database); + await database.PutAsync(One, Note("owned")); + await cache.UpsertAsync(new MeshQuickstartNote { NoteId = "u", Body = "unowned" }, new CultRecordHandle(Unowned)); + await cache.UpsertAsync(new JournalIndexedNote { Name = "n", Tag = "t" }, new CultRecordHandle(new CultRecordKey("unrelated:indexed"))); + + Assert.That(database.GetAll(), Is.Empty); + Assert.That(database.GetAll().Count(), Is.EqualTo(1), "only the owned row"); + Assert.That(database.GetAll().Count(), Is.EqualTo(1)); + Assert.That(await database.GetAsync(Unowned), Is.Null); + Assert.That(await database.GetAsync(One), Is.Not.Null); + Assert.That(database.GetByName("u"), Is.Null); + Assert.That(database.GetByName("n"), Is.Null); + Assert.That(database.GetByIndex("tag", "t"), Is.Null); + Assert.That(cache.Get(Unowned), Is.Not.Null, "the cache itself still holds it"); + Assert.That(seen, Is.EqualTo(new[] { One.Value }), "what the database reads out is what it publishes"); + } + + [Test] + public async Task ADatabaseCollectionOfASchemaNoShardOwnsIsEmpty() + { + var (database, cache, _, _) = OwnedNothingOfNote(); + await cache.UpsertAsync(new MeshQuickstartNote { NoteId = "u", Body = "unowned" }, new CultRecordHandle(Unowned)); + + var collection = GameCult.Mesh.CultMesh.Collection(database); + + Assert.That(await collection.LatestAsync(), Is.Empty); + } + + [Test] + public async Task TheShardlessSnapshotOmitsARowNoShardOwns() + { + var (database, cache, _, _) = OwnedNothingOfNote(); + await database.PutAsync(One, Note("owned")); + await cache.UpsertAsync(new MeshQuickstartNote { NoteId = "u", Body = "unowned" }, new CultRecordHandle(Unowned)); + using var server = new Server(cache, ServerSecurityOptions.Development()); + using var databaseServer = new CultNetDatabaseServer(server, database); + + var response = databaseServer.CreateSnapshotResponse(new CultNetSnapshotRequestMessage { MessageId = "all" }); + + Assert.That(response.Documents.Select(document => document.RecordKey), Is.EqualTo(new[] { One.Value })); + } + + [Test] + public async Task ASubscriptionSnapshotOmitsARowNoShardOwns() + { + var (database, cache, _, _) = OwnedNothingOfNote(); + await database.PutAsync(One, Note("owned")); + await cache.UpsertAsync(new MeshQuickstartNote { NoteId = "u", Body = "unowned" }, new CultRecordHandle(Unowned)); + var socket = new System.Net.Sockets.Socket(System.Net.Sockets.AddressFamily.InterNetwork, System.Net.Sockets.SocketType.Dgram, System.Net.Sockets.ProtocolType.Udp); + socket.Bind(new System.Net.IPEndPoint(System.Net.IPAddress.Loopback, 0)); + socket.ReceiveTimeout = 20; + using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions { RuntimeId = "owned-reads-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("owned-reads-client"); + CultNetSnapshotResponseRawMessage? snapshot = null; + client.OnCultNet(message => snapshot = message); + client.Connect("127.0.0.1", server.LocalEndPoint.Port); + for (var attempt = 0; attempt < 300 && !client.Connected; attempt++) + await Task.Delay(10); + client.SendCultNet(new CultNetDatabaseSubscribeMessage { MessageId = "subscribe-all", SubscriptionId = "all", IncludeSnapshot = true }); + for (var attempt = 0; attempt < 300 && snapshot == null; attempt++) + await Task.Delay(10); + + Assert.That(snapshot, Is.Not.Null, "the subscription answered a snapshot"); + Assert.That(snapshot!.Documents.Select(document => document.RecordKey), Is.EqualTo(new[] { One.Value })); + } + finally + { + cancellation.Cancel(); + serverThread.Join(); + } + } + + // A replica applies the log the way it applies a snapshot: a row its shard does not own is left out. The entry is + // still recorded, so a replica chained on this one sees no gap. + [Test] + public async Task AReplicaLogApplySkipsARowItsShardDoesNotOwnButRecordsTheEntry() + { + var cache = new CultCache(); + var replica = Database(cache, primary: false, store: null); + await cache.UpsertAsync(new MeshQuickstartNote { NoteId = "c", Body = "cached" }, new CultRecordHandle(Unowned)); + var put = replica.Documents.CreateRawDocumentPutMessage("wire", new CultRecordHandle(new CultRecordKey("unrelated:put")), new MeshQuickstartNote { NoteId = "p", Body = "replicated" }); + put.ShardId = ShardId; + put.ShardEpoch = 1; + var owned = replica.Documents.CreateRawDocumentPutMessage("wire", new CultRecordHandle(One), Note("owned")); + owned.ShardId = ShardId; + owned.ShardEpoch = 1; + var now = DateTimeOffset.UtcNow.ToString("O"); + + await replica.ApplyShardLogResponseAsync(new CultNetShardLogResponseMessage + { + MessageId = "log", + ShardId = ShardId, + ShardEpoch = 1, + Entries = + [ + new CultNetShardLogEntryMessage { Sequence = 1, CommittedAt = now, ChangeKind = "added", Put = put }, + new CultNetShardLogEntryMessage + { + Sequence = 2, + CommittedAt = now, + ChangeKind = "removed", + Delete = new CultNetDocumentDeleteMessage + { + MessageId = "d", + SchemaId = cache.Registry.GetRequired().SchemaId, + RecordKey = Unowned.Value, + ShardId = ShardId, + ShardEpoch = 1 + } + }, + new CultNetShardLogEntryMessage { Sequence = 3, CommittedAt = now, ChangeKind = "added", Put = owned } + ] + }); + + Assert.That(cache.Get(new CultRecordKey("unrelated:put")), Is.Null, "the put was left out"); + Assert.That(cache.Get(Unowned), Is.Not.Null, "the delete removed nothing the shard owns"); + Assert.That(TextOf(cache, One), Is.EqualTo("owned"), "owned entries still apply"); + Assert.That(replica.GetMutationLog(ShardId).Select(entry => entry.Sequence), Is.EqualTo(new[] { 1L, 2L, 3L })); + Assert.That(replica.GetAppliedShardSequence(ShardId), Is.EqualTo(3)); + } + + // The server's put and delete handlers are reached through their private delegates with an uninitialized peer (as + // the R-AM test does): anything the handler tries to send throws on that peer, so a handler that returns quietly + // sent nothing. + private static (CultNetDatabaseServer Server, Func Put, Func Delete, CapturingLogger Logger, Server Host) WireHandlers(CultNetDatabase database, CultCache cache) + { + var host = new Server(cache, ServerSecurityOptions.Development()); + var logger = new CapturingLogger(); + host.Logger = logger; + var databaseServer = new CultNetDatabaseServer(host, database); + var peer = (CultNetServerPeer)System.Runtime.CompilerServices.RuntimeHelpers.GetUninitializedObject(typeof(CultNetServerPeer)); + const System.Reflection.BindingFlags flags = System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance; + var put = (Func)typeof(CultNetDatabaseServer).GetField("_putHandler", flags)!.GetValue(databaseServer)!; + var delete = (Func)typeof(CultNetDatabaseServer).GetField("_deleteHandler", flags)!.GetValue(databaseServer)!; + return (databaseServer, message => put(message, peer), message => delete(message, peer), logger, host); + } + + private static CultNetDocumentPutRawMessage OwnedPut(CultNetDatabase database) + { + var message = database.Documents.CreateRawDocumentPutMessage("wire", new CultRecordHandle(One), Note("wire")); + message.ShardId = ShardId; + message.ShardEpoch = 1; + return message; + } + + [Test] + public async Task ACommittedRemotePutWhoseLogFailedIsNotAnsweredAsARefusal() + { + var cache = new CultCache(); + var database = Database(cache, primary: true, new FlakyLogStore { RefuseAppend = _ => true }); + var (databaseServer, put, _, logger, host) = WireHandlers(database, cache); + using var _1 = databaseServer; + using var _2 = host; + + await put(OwnedPut(database)); + + Assert.That(TextOf(cache, One), Is.EqualTo("wire"), "the put committed"); + Assert.That(logger.Errors, Is.Empty); + Assert.That(logger.Warnings, Has.Some.Contains("committed but its shard log failed")); + } + + [Test] + public async Task ACommittedRemotePutWhoseLogFailedBesideAnotherFailureIsStillNotAnsweredAsARefusal() + { + var cache = new CultCache(); + var database = Database(cache, primary: true, new FlakyLogStore { RefuseAppend = _ => true }); + using var second = cache.AddJournal(_ => throw new InvalidOperationException("second journal")); + var (databaseServer, put, _, logger, host) = WireHandlers(database, cache); + using var _1 = databaseServer; + using var _2 = host; + + await put(OwnedPut(database)); + + Assert.That(TextOf(cache, One), Is.EqualTo("wire")); + Assert.That(logger.Errors, Is.Empty); + Assert.That(logger.Warnings, Has.Some.Contains("committed but its shard log failed")); + } + + [Test] + public async Task ACommittedRemoteDeleteWhoseLogFailedIsNotAnsweredAsARefusal() + { + var cache = new CultCache(); + var database = Database(cache, primary: true, new FlakyLogStore { RefuseAppend = sequence => sequence == 2 }); + await database.PutAsync(One, Note("one")); + var (databaseServer, _, delete, logger, host) = WireHandlers(database, cache); + using var _1 = databaseServer; + using var _2 = host; + + await delete(new CultNetDocumentDeleteMessage + { + MessageId = "d", + SchemaId = SchemaId(cache), + RecordKey = One.Value, + ShardId = ShardId, + ShardEpoch = 1 + }); + + Assert.That(TextOf(cache, One), Is.Null, "the delete committed"); + Assert.That(logger.Warnings, Has.Some.Contains("committed but its shard log failed")); + } + + [Test] + public void ARemoteDeleteOfASchemaNoShardOwnsIsRefusedOnTheWireAtWarningLevel() + { + var (database, cache, _, unowned) = OwnedNothingOfNote(); + var (databaseServer, _, delete, logger, host) = WireHandlers(database, cache); + using var _1 = databaseServer; + using var _2 = host; + + // The handler answers the peer with the refusal; the uninitialized peer cannot send, which is how the test knows it tried. + Assert.CatchAsync(async () => await delete(new CultNetDocumentDeleteMessage + { + MessageId = "d", + SchemaId = unowned, + RecordKey = Unowned.Value, + ShardId = database.Shards[0].ShardId, + ShardEpoch = database.Shards[0].Epoch + })); + + Assert.That(logger.Warnings, Has.Some.Contains("raw delete refused").And.Contains(unowned)); + Assert.That(logger.Errors, Is.Empty); + } + + // The journal is disposed before the observer: an admission the journal logs while the database is being disposed is + // still published, never logged and then dropped. + [Test] + public async Task ADatabaseDisposedWhileAWriteIsInItsLogAppendStillPublishesWhatItLogged() + { + var store = new FlakyLogStore(); + var cache = new CultCache(); + var database = Database(cache, primary: true, store); + var seen = Record(database); + using var inAppend = new ManualResetEventSlim(); + using var release = new ManualResetEventSlim(); + store.OnAppend = () => + { + inAppend.Set(); + release.Wait(TimeSpan.FromSeconds(10)); + }; + var writer = Task.Run(async () => await database.PutAsync(One, Note("one"))); + Assert.That(inAppend.Wait(TimeSpan.FromSeconds(10)), Is.True); + + var dispose = Task.Run(database.Dispose); + await Task.Delay(300); + release.Set(); + await writer; + await dispose; + + Assert.That(Sequences(store), Is.EqualTo(new[] { 1L }), "the write was logged"); + Assert.That(seen, Is.EqualTo(new[] { One.Value }), "and published"); + } + private sealed class CapturingLogger : GameCult.Logging.ILogger { public List Errors { get; } = new(); + public List Warnings { get; } = new(); public void LogInfo(string message) { } - public void LogWarning(string message) { } + public void LogWarning(string message) => Warnings.Add(message); public void LogError(string message) => Errors.Add(message); public void LogDebug(string message) { } } diff --git a/tests/GameCult.Networking.Tests/CultNetSchemaContractTests.cs b/tests/GameCult.Networking.Tests/CultNetSchemaContractTests.cs index b5f4f08f..3aa2705d 100644 --- a/tests/GameCult.Networking.Tests/CultNetSchemaContractTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetSchemaContractTests.cs @@ -195,6 +195,17 @@ namespace GameCult.Networking.Tests AssertErrorMatchesSchema(CultNetErrorMessage.ForReferenceOutsideTarget(ex)); } + [Test] + public void ErrorMessage_UnownedSchemaRealBytes_MatchesItsSchema() + { + var message = CultNetErrorMessage.ForUnownedSchema(new CultNetUnownedSchemaException("some.schema.v1", new CultRecordKey("some:key"))); + + Assert.That(message.Code, Is.EqualTo("unowned_schema")); + Assert.That(message.Details!.Field, Is.EqualTo("schemaId")); + Assert.That(message.Details.Value, Is.EqualTo("some.schema.v1")); + AssertErrorMatchesSchema(message); + } + private static void AssertErrorMatchesSchema(CultNetErrorMessage message) { var validator = new MiniJsonSchemaValidator(SchemaDir());