diff --git a/contracts/cultnet/cultnet-distributed-database.md b/contracts/cultnet/cultnet-distributed-database.md index 932ba861..1ddc80e8 100644 --- a/contracts/cultnet/cultnet-distributed-database.md +++ b/contracts/cultnet/cultnet-distributed-database.md @@ -140,6 +140,12 @@ A malformed put or delete is `message_invalid`, not a per-operation code: the pe - A shard snapshot represents a log sequence. Applying it replaces the local shard view and advances the replica cursor to that sequence. - Realtime subscriptions publish domain changes, not storage envelopes. +- A subscription's `database_change_raw` messages may carry `commitId` and `commitCount`: the changes of one cache + commit that this subscription delivers (`commitCount` of them, at least two). Both fields are optional and additive; a + peer that does not know them ignores them and applies each change alone. A replica that knows them applies the + commit's changes as one commit, so the replica judges the result (a swap of two unique-index values) and not the state + between records. A replica whose cache refuses a commit is faulted, never silent. `commitId` is opaque and valid only + within the subscription. The C# subscription server and database server tag commits; other runtimes adopt next. - Client-owned input documents may be predicted locally only inside explicit `CultNetClientAuthorityScope` declarations. - Predicted input state is not authoritative shared state. The shard log still diff --git a/contracts/cultnet/cultnet.database-change-raw.schema.json b/contracts/cultnet/cultnet.database-change-raw.schema.json index 2725a5af..d214ed6b 100644 --- a/contracts/cultnet/cultnet.database-change-raw.schema.json +++ b/contracts/cultnet/cultnet.database-change-raw.schema.json @@ -12,7 +12,9 @@ "changeKind": { "enum": ["added", "updated", "removed"] }, "document": { "$ref": "./cultnet.raw-document-record.schema.json" }, "recordKey": { "type": "string", "minLength": 1 }, - "schemaId": { "type": "string", "minLength": 1 } + "schemaId": { "type": "string", "minLength": 1 }, + "commitId": { "type": "string", "minLength": 1 }, + "commitCount": { "type": "integer", "minimum": 2 } }, "allOf": [ { diff --git a/src/GameCult.Networking/CultNetDatabase.cs b/src/GameCult.Networking/CultNetDatabase.cs index af561ea5..9db3cd1b 100644 --- a/src/GameCult.Networking/CultNetDatabase.cs +++ b/src/GameCult.Networking/CultNetDatabase.cs @@ -315,6 +315,22 @@ namespace GameCult.Networking public CultNetSelectionCursorKey? CursorKey { get; set; } } + /// + /// The typed changes ( boxed) one cache commit made, in the order the cache admitted them. + /// is opaque and unique among the commits of one database instance. + /// + public sealed class CultNetDatabaseCommit + { + public CultNetDatabaseCommit(string id, IReadOnlyList changes) + { + Id = id; + Changes = changes; + } + + public string Id { get; } + public IReadOnlyList Changes { get; } + } + /// /// A typed domain change emitted by a CultNet database surface. /// @@ -587,6 +603,7 @@ namespace GameCult.Networking private readonly Dictionary<(string SchemaId, string RecordKey), long> _lastWriteSequence = new(); private readonly Subject _changes = new(); + private readonly Subject _commits = new(); private readonly IDisposable _cacheChanges; private readonly IDisposable _cacheJournal; // What a write door of this database knows about the change it is admitting and the cache does not: the wire @@ -1148,6 +1165,17 @@ namespace GameCult.Networking return _changes; } + /// + /// Watches changes as the cache commits that made them: one event per commit, after every change of it was published to + /// . A consumer that forwards changes to a replica forwards a commit as a unit, so the replica + /// can judge the result of the commit and not each record's intermediate state. + /// + public Observable WatchCommits() + { + ThrowIfDisposed(); + return _commits; + } + /// /// Watches one record. /// @@ -1215,6 +1243,7 @@ namespace GameCult.Networking } _changes.Dispose(); + _commits.Dispose(); } // The cache is the one owner of "a change was committed"; the journal is this database's one decision point and its @@ -1231,6 +1260,8 @@ namespace GameCult.Networking } List? failures = null; + // One journal call is one admission, so one commit; its first Sequence names it. + var commit = new CommitGroup(changes.Count == 0 ? "0" : changes[0].Sequence.ToString(System.Globalization.CultureInfo.InvariantCulture)); List? burned = null; List? burnCauses = null; foreach (var change in changes) @@ -1288,7 +1319,10 @@ namespace GameCult.Networking } lock (_logGate) - _stash[change.Sequence] = new Publication(documentType, kind, change.Key, descriptor.SchemaId, shard, change.Document, change.PreviousDocument); + { + _stash[change.Sequence] = new Publication(documentType, kind, change.Key, descriptor.SchemaId, shard, change.Document, change.PreviousDocument, commit); + commit.Unpublished++; + } } // Every change is stashed and will be published; the writer hears of a failed log afterwards, in one exception. @@ -1321,8 +1355,9 @@ namespace GameCult.Networking Publish(publication); } - private void Publish(Publication publication) => - PublishUntyped( + private void Publish(Publication publication) + { + var change = PublishUntyped( publication.DocumentType, publication.Kind, publication.Key, @@ -1330,6 +1365,19 @@ namespace GameCult.Networking publication.Shard, publication.Document, publication.PreviousDocument); + CultNetDatabaseCommit? finished = null; + lock (_logGate) + { + var commit = publication.Commit; + if (change != null) + commit.Published.Add(change); + if (--commit.Unpublished == 0 && commit.Published.Count > 0) + finished = new CultNetDatabaseCommit(commit.Id, commit.Published.ToArray()); + } + + if (finished != null) + _commits.OnNext(finished); + } private void RecordReplicatedEntry( CultNetShardDescriptor shard, @@ -1467,8 +1515,10 @@ namespace GameCult.Networking string schemaId, CultNetShardDescriptor shard, object? document, - object? previousDocument) + object? previousDocument, + CommitGroup commit) { + Commit = commit; DocumentType = documentType; Kind = kind; Key = key; @@ -1485,6 +1535,16 @@ namespace GameCult.Networking public CultNetShardDescriptor Shard { get; } public object? Document { get; } public object? PreviousDocument { get; } + public CommitGroup Commit { get; } + } + + // The changes of one cache admission, counted down as they are published. Guarded by _logGate. + private sealed class CommitGroup + { + public CommitGroup(string id) => Id = id; + public string Id { get; } + public int Unpublished { get; set; } + public List Published { get; } = new(); } private sealed class DoorContext @@ -1964,7 +2024,7 @@ namespace GameCult.Networking return $"{schemaId}:{key.Value}"; } - private void PublishUntyped( + private object? PublishUntyped( Type documentType, CultNetDatabaseChangeKind kind, CultRecordKey key, @@ -1987,6 +2047,8 @@ namespace GameCult.Networking { _changes.OnNext(change); } + + return change; } // No shard owns a schema and key that no descriptor matches, and nothing is written to one: every write path refuses. diff --git a/src/GameCult.Networking/CultNetDatabaseServer.cs b/src/GameCult.Networking/CultNetDatabaseServer.cs index e1694131..32f302fc 100644 --- a/src/GameCult.Networking/CultNetDatabaseServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseServer.cs @@ -765,14 +765,41 @@ namespace GameCult.Networking { // Keys this subscriber only ever received as a refusal: it was never given them as records. var refused = new HashSet(StringComparer.Ordinal); - return _database.WatchAllChanges().Subscribe(change => + return _database.WatchCommits().Subscribe(commit => { - var outbound = DeliverChange(change, subscriptionId, selection, refused, error => peer.SendCultNet(error)); - if (outbound != null) - peer.SendCultNet(outbound); + foreach (var message in DeliverCommit(commit, subscriptionId, selection, refused, error => peer.SendCultNet(error))) + peer.SendCultNet(message); }); } + // One subscriber's view of one commit. Its changes go out together, tagged with the commit, so a replica applies them as one. + internal IReadOnlyList DeliverCommit( + CultNetDatabaseCommit commit, + string subscriptionId, + CultNetSelection selection, + HashSet refused, + Action sendError) + { + var outbound = new List(); + foreach (var change in commit.Changes) + { + var message = DeliverChange(change, subscriptionId, selection, refused, sendError); + if (message != null) + outbound.Add(message); + } + + if (outbound.Count > 1) + { + foreach (var message in outbound) + { + message.CommitId = commit.Id; + message.CommitCount = outbound.Count; + } + } + + return 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( diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs index db1e2907..7df7349a 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs @@ -29,6 +29,8 @@ namespace GameCult.Networking private readonly ConcurrentDictionary _initialSnapshots = new(); private readonly ConcurrentDictionary _deliveryModes = new(StringComparer.Ordinal); + // Changes of a commit that has not fully arrived, by subscription and commit id. Touched only inside the queue. + private readonly Dictionary<(string SubscriptionId, string CommitId), List> _commits = new(); private readonly object _queueLock = new(); private Task _queue = Task.CompletedTask; private bool _disposed; @@ -48,6 +50,15 @@ namespace GameCult.Networking /// Raised after a live change has been applied to the local typed cache. public event Action? Changed; + /// + /// Raised when the local cache refuses a replicated commit. The replica then differs from its source; the client applies + /// no further changes and reports until it is replaced by a new subscription. + /// + public event Action? Faulted; + + /// The refusal that stopped replication, or null while the replica is intact. + public Exception? Fault { get; private set; } + /// Gets the local reactive cache receiving replicated documents. public CultCache Cache => _cache; @@ -211,48 +222,61 @@ namespace GameCult.Networking private void HandleChange(CultNetDatabaseChangeRawMessage message) { - Enqueue(async () => + Enqueue(() => { var deliveryMode = _deliveryModes.TryGetValue(message.SubscriptionId, out var configured) ? configured : CultNetDatabaseSubscriptionDeliveryMode.ReplicateToCache; - if (string.Equals(message.ChangeKind, "removed", StringComparison.Ordinal)) - { - if (deliveryMode == CultNetDatabaseSubscriptionDeliveryMode.ReplicateToCache) - await DeleteAsync(message).ConfigureAwait(false); - Changed?.Invoke(new CultNetReplicatedDocumentChange( - message.SubscriptionId, - "removed", - message.RecordKey ?? string.Empty, - message.SchemaId ?? string.Empty, - null)); - return; - } + // Live delivery applies nothing, so there is no commit to judge; each change is reported as it arrives. + if (deliveryMode != CultNetDatabaseSubscriptionDeliveryMode.ReplicateToCache || message.CommitId == null || !(message.CommitCount > 1)) + return Apply(deliveryMode, new[] { message }); - if (message.Document == null) return; - var document = deliveryMode == CultNetDatabaseSubscriptionDeliveryMode.Live - ? _documents.DeserializeRawDocument(message.Document) - : await _documents.ApplyRawDocumentPutMessageAsync( - _cache, - new CultNetDocumentPutRawMessage - { - MessageId = message.MessageId, - Document = message.Document - }).ConfigureAwait(false); - Changed?.Invoke(new CultNetReplicatedDocumentChange( - message.SubscriptionId, - message.ChangeKind, - message.Document.RecordKey, - message.Document.SchemaId, - document)); + // The commit's other changes follow on this connection; the replica applies them as one commit. + var held = (message.SubscriptionId, message.CommitId); + if (!_commits.TryGetValue(held, out var changes)) + _commits[held] = changes = new List(); + changes.Add(message); + if (changes.Count < message.CommitCount) + return Task.CompletedTask; + _commits.Remove(held); + return Apply(deliveryMode, changes); }); } - private Task DeleteAsync(CultNetDatabaseChangeRawMessage message) + private Task Apply(CultNetDatabaseSubscriptionDeliveryMode deliveryMode, IReadOnlyList changes) { - if (string.IsNullOrWhiteSpace(message.RecordKey) || string.IsNullOrWhiteSpace(message.SchemaId)) return Task.CompletedTask; - _documents.ResolveDescriptorForSchemaId(message.SchemaId!); - _cache.Remove(new CultRecordKey(message.RecordKey!)); + // A replica that refused a commit no longer matches its source. Applying later commits on top would hide that, so + // nothing more is applied: the fault is surfaced and the replica is rebuilt by subscribing again. + if (Fault != null) + return Task.CompletedTask; + IReadOnlyList applied; + try + { + applied = deliveryMode == CultNetDatabaseSubscriptionDeliveryMode.Live + ? changes.Select(change => change.Document == null ? null : _documents.DeserializeRawDocument(change.Document)).ToArray() + : _documents.ApplyRawChangesAsOneCommit(_cache, changes); + } + catch (Exception error) + { + Fault = error; + Faulted?.Invoke(error); + return Task.CompletedTask; + } + + for (var index = 0; index < changes.Count; index++) + { + var change = changes[index]; + var removed = string.Equals(change.ChangeKind, "removed", StringComparison.Ordinal); + if (!removed && change.Document == null) + continue; + Changed?.Invoke(new CultNetReplicatedDocumentChange( + change.SubscriptionId, + change.ChangeKind, + removed ? change.RecordKey ?? string.Empty : change.Document!.RecordKey, + removed ? change.SchemaId ?? string.Empty : change.Document!.SchemaId, + removed ? null : applied[index])); + } + return Task.CompletedTask; } diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs index 8398feba..1325f85e 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs @@ -300,50 +300,76 @@ namespace GameCult.Networking SubscriptionKey key) { var legacyRequest = request.ToLegacyShape(subscriptionId); - return _database.WatchAllChanges().Subscribe(change => + return _database.WatchCommits().Subscribe(commit => { lock (_lifecycleGate) { if (_disposed || !_requests.ContainsKey(key) || !_projections.TryGetValue(key, out var projection)) return; - if (_authorizeRequest?.Invoke(legacyRequest, peer) == false) - { - Reconcile(key, request); - return; - } - - // D6: cites/cited are set-dependent - a change to row B can change whether row A - // matches - so a hop-bearing selection runs the full diff-against-delivered - // reconcile on every change instead of the single-row fast path. - if (request.Selection.HasHop) - { - Reconcile(key, request); - return; - } - - var sourceRecordKey = ResolveChangeRecordKey(change); - CultNetRawDocumentRecord? matched; + // What this subscription sends for the commit is held and sent together, tagged with the commit, so the replica + // applies it as one commit and judges the result. + var outbox = _outbox = new CommitOutbox(commit.Id); try { - matched = CreateMatchedRecord(change, request, peer); + ForwardCommit(commit, request, key, projection, legacyRequest, peer, subscriptionId); } - catch (NotSupportedException refusal) + finally { - // Q6: a variant is not a record CultNet can carry; the subscriber that would have received it is told, by key. - peer.SendCultNet(CultNetErrorMessage.ForVariantUnsupported(refusal.Message)); - return; + _outbox = null; + outbox.Send(peer); } - - ApplyProjectedChange( - projection, - sourceRecordKey, - matched, - peer, - subscriptionId); } }); } + private void ForwardCommit( + CultNetDatabaseCommit commit, + SubscriptionRequest request, + SubscriptionKey key, + SubscriptionProjectionState projection, + CultNetDatabaseSubscribeMessage legacyRequest, + ICultNetSchemaServerPeer peer, + string subscriptionId) + { + if (_authorizeRequest?.Invoke(legacyRequest, peer) == false) + { + Reconcile(key, request); + return; + } + + // D6: cites/cited are set-dependent - a change to row B can change whether row A + // matches - so a hop-bearing selection runs the full diff-against-delivered + // reconcile once per commit instead of the single-row fast path. + if (request.Selection.HasHop) + { + Reconcile(key, request); + return; + } + + foreach (var change in commit.Changes) + { + var sourceRecordKey = ResolveChangeRecordKey(change); + CultNetRawDocumentRecord? matched; + try + { + matched = CreateMatchedRecord(change, request, peer); + } + 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(CultNetErrorMessage.ForVariantUnsupported(refusal.Message)); + continue; + } + + ApplyProjectedChange( + projection, + sourceRecordKey, + matched, + peer, + subscriptionId); + } + } + private void Reconcile(SubscriptionKey key, SubscriptionRequest request) { if (!_projections.TryGetValue(key, out var projection)) return; @@ -487,12 +513,15 @@ namespace GameCult.Networking string.Equals(left.SourceAgentId, right.SourceAgentId, StringComparison.Ordinal) && string.Equals(left.SourceRole, right.SourceRole, StringComparison.Ordinal); - private static void SendUpsert( + // The commit being forwarded, under _lifecycleGate: the change messages it produces are held here, not sent one by one. + private CommitOutbox? _outbox; + + private void SendUpsert( ICultNetSchemaServerPeer peer, string subscriptionId, CultNetRawDocumentRecord document, bool added) => - peer.SendCultNet(new CultNetDatabaseChangeRawMessage + Deliver(peer, new CultNetDatabaseChangeRawMessage { MessageId = Guid.NewGuid().ToString("N"), SubscriptionId = subscriptionId, @@ -500,11 +529,11 @@ namespace GameCult.Networking Document = document }); - private static void SendRemoval( + private void SendRemoval( ICultNetSchemaServerPeer peer, string subscriptionId, CultNetRawDocumentRecord document) => - peer.SendCultNet(new CultNetDatabaseChangeRawMessage + Deliver(peer, new CultNetDatabaseChangeRawMessage { MessageId = Guid.NewGuid().ToString("N"), SubscriptionId = subscriptionId, @@ -513,6 +542,38 @@ namespace GameCult.Networking SchemaId = document.SchemaId }); + private void Deliver(ICultNetSchemaServerPeer peer, CultNetDatabaseChangeRawMessage message) + { + if (_outbox is { } outbox) + outbox.Messages.Add(message); + else + peer.SendCultNet(message); + } + + private sealed class CommitOutbox + { + private readonly string _commitId; + + public CommitOutbox(string commitId) => _commitId = commitId; + + public List Messages { get; } = new(); + + public void Send(ICultNetSchemaServerPeer peer) + { + if (Messages.Count > 1) + { + foreach (var message in Messages) + { + message.CommitId = _commitId; + message.CommitCount = Messages.Count; + } + } + + foreach (var message in Messages) + peer.SendCultNet(message); + } + } + private void Withdraw(SubscriptionKey key, bool sendRemovals, bool forgetRequest) { if (_subscriptions.TryRemove(key, out var subscription)) diff --git a/src/GameCult.Networking/CultNetDocumentRegistry.cs b/src/GameCult.Networking/CultNetDocumentRegistry.cs index 58b7ca36..19f17c8c 100644 --- a/src/GameCult.Networking/CultNetDocumentRegistry.cs +++ b/src/GameCult.Networking/CultNetDocumentRegistry.cs @@ -722,6 +722,49 @@ namespace GameCult.Networking return document; } + /// + /// Applies change messages to a cache as one commit, so the cache judges the result of the changes together and not the + /// state between them. Returns the applied document of each message, null for a removal. A refusal by the cache throws + /// and nothing is applied. + /// + internal IReadOnlyList ApplyRawChangesAsOneCommit( + CultCache cache, + IReadOnlyList changes) + { + var applied = new object?[changes.Count]; + var staged = new List<(Type? Type, object? Document, CultRecordKey Key)>(changes.Count); + for (var index = 0; index < changes.Count; index++) + { + var change = changes[index]; + if (string.Equals(change.ChangeKind, "removed", StringComparison.Ordinal)) + { + if (string.IsNullOrWhiteSpace(change.RecordKey) || string.IsNullOrWhiteSpace(change.SchemaId)) + continue; + ResolveDescriptorForSchemaId(change.SchemaId!); + staged.Add((null, null, new CultRecordKey(change.RecordKey!))); + continue; + } + + if (change.Document == null) + continue; + var document = DeserializeRawDocument(change.Document); + applied[index] = document; + staged.Add((ResolveDescriptorForRawDocument(change.Document).DocumentType, document, new CultRecordKey(change.Document.RecordKey))); + } + + cache.Commit(batch => + { + foreach (var (type, document, key) in staged) + { + if (type == null) + batch.Remove(key); + else + batch.Upsert(type, document!, key); + } + }); + return applied; + } + /// /// Decodes one validated raw document without materializing it in a CultCache replica. /// Use this for ephemeral reactive subscriptions whose callback is the consumer. @@ -754,25 +797,22 @@ namespace GameCult.Networking } /// - /// Applies all documents from a raw snapshot response. + /// Applies all documents from a raw snapshot response as one commit, so a snapshot that moves a unique-index value + /// between records is judged on its result. /// - public async Task> ApplyRawSnapshotResponseAsync( + public Task> ApplyRawSnapshotResponseAsync( CultCache cache, CultNetSnapshotResponseRawMessage response) { - var applied = new List(response.Documents.Length); - foreach (var document in response.Documents) - { - applied.Add(await ApplyRawDocumentPutMessageAsync( - cache, - new CultNetDocumentPutRawMessage - { - MessageId = response.MessageId, - Document = document - }).ConfigureAwait(false)); - } - - return applied; + var applied = ApplyRawChangesAsOneCommit( + cache, + response.Documents.Select(document => new CultNetDatabaseChangeRawMessage + { + SubscriptionId = response.MessageId, + ChangeKind = "added", + Document = document + }).ToArray()); + return Task.FromResult>(applied.Select(document => document!).ToArray()); } internal CultDocumentDescriptor ResolveDescriptorForRawDocument(CultNetRawDocumentRecord document) diff --git a/src/GameCult.Networking/CultNetSchemaMessages.cs b/src/GameCult.Networking/CultNetSchemaMessages.cs index a2211672..bc59b007 100644 --- a/src/GameCult.Networking/CultNetSchemaMessages.cs +++ b/src/GameCult.Networking/CultNetSchemaMessages.cs @@ -1179,6 +1179,17 @@ namespace GameCult.Networking /// Gets or sets the deleted schema id for removed changes. /// [Key("schemaId")] public string? SchemaId { get; set; } + /// + /// Gets or sets the identity of the cache commit this change belongs to. Optional and additive: a peer that does not + /// know the field ignores it and applies the change alone. Opaque to the receiver; equal ids on one subscription + /// name changes of one commit. + /// + [Key("commitId")] public string? CommitId { get; set; } + /// + /// Gets or sets how many change messages of this commit this subscription carries. Set with + /// when that is more than one; the receiver applies the commit once all have arrived. + /// + [Key("commitCount")] public int? CommitCount { get; set; } } /// diff --git a/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs b/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs new file mode 100644 index 00000000..e6aca14a --- /dev/null +++ b/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs @@ -0,0 +1,306 @@ +#nullable enable +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using GameCult.Caching; +using R3; +using MessagePack; +using NUnit.Framework; + +namespace GameCult.Networking.Tests +{ + // variants:ruling:replica-commit-grouped-changes. The authoritative cache judges a commit on its result; a replica applying the + // changes one record at a time would judge each intermediate state, where a swap of two [CultIndex] values holds two holders. + // The subscription server tags the changes of one cache commit, the client applies them as one commit, and a replica that + // refuses a commit says so. No transport: the server and client are joined in memory, and every message crosses MessagePack. + public sealed class CultNetReplicaCommitTests + { + [CultDocument("tests.replica_commit_item", "tests.replica_commit_item.v1")] + [MessagePackObject(AllowPrivate = true)] + public sealed class ReplicaItem + { + [Key(0)] + [CultIndex("sku")] + public string Sku = string.Empty; + + [Key(1)] + public string Note = string.Empty; + } + + // The change message as a peer from before commit identity knew it. + [MessagePackObject] + public sealed class LegacyChange + { + [Key("schemaVersion")] public string SchemaVersion { get; set; } = string.Empty; + [Key("messageId")] public string MessageId { get; set; } = string.Empty; + [Key("subscriptionId")] public string SubscriptionId { get; set; } = string.Empty; + [Key("changeKind")] public string ChangeKind { get; set; } = string.Empty; + [Key("document")] public CultNetRawDocumentRecord? Document { get; set; } + [Key("recordKey")] public string? RecordKey { get; set; } + [Key("schemaId")] public string? SchemaId { get; set; } + } + + private sealed class LoopbackServer : ICultNetSchemaServer + { + private readonly Dictionary _handlers = new(); + public void OnCultNet(Func callback) where T : ICultNetSchemaMessage => _handlers[typeof(T)] = callback; + public void RemoveCultNetMessageListener(Delegate callback) where T : ICultNetSchemaMessage => _handlers.Remove(typeof(T)); + public Task Dispatch(T message, ICultNetSchemaServerPeer peer) where T : ICultNetSchemaMessage => + ((Func)_handlers[typeof(T)])(message, peer); + } + + private sealed class Client : ICultNetSchemaClient + { + private readonly LoopbackServer _server; + private readonly Dictionary> _handlers = new(); + private readonly Peer _peer; + public Client(LoopbackServer server, Peer peer) { _server = server; _peer = peer; peer.Client = this; } + public bool Connected => true; + public void Connect(string host, int port) { } + public void SendCultNet(T message) where T : ICultNetSchemaMessage + { + if (message is CultNetDatabaseSubscribeMessage subscribe) + _server.Dispatch(subscribe, _peer).GetAwaiter().GetResult(); + } + public void OnCultNet(Action callback) where T : ICultNetSchemaMessage + { + if (!_handlers.TryGetValue(typeof(T), out var list)) _handlers[typeof(T)] = list = new List(); + list.Add(callback); + } + public void Emit(T message) where T : ICultNetSchemaMessage + { + if (_handlers.TryGetValue(typeof(T), out var list)) + foreach (var handler in list.ToArray()) ((Action)handler)(message); + } + public void Dispose() { } + } + + private sealed class Peer : ICultNetSchemaServerPeer + { + public Client? Client; + public ConcurrentQueue Sent { get; } = new(); + public void SendCultNet(T message) where T : ICultNetSchemaMessage + { + // The wire: serialized, then read back as the receiver would. + var bytes = CultNetSchemaMessageSerialization.Serialize(message); + Sent.Enqueue(bytes); + Client!.Emit(CultNetSchemaMessageSerialization.Deserialize(bytes)); + } + public CultNetDatabaseChangeRawMessage[] Changes => Sent + .Select(bytes => CultNetSchemaMessageSerialization.Deserialize(bytes)) + .OfType().ToArray(); + } + + private sealed class Replication : IDisposable + { + public CultCache Source { get; } = new(); + public CultCache Replica { get; } = new(); + public Peer Peer { get; } = new(); + public CultNetDatabaseSubscriptionClient Client { get; } + public List Faults { get; } = new(); + public CultNetDatabase Database { get; } + private readonly CultNetDatabaseSubscriptionServer _server; + + public Replication(Action? seedSource = null, Action? seedReplica = null) + { + seedSource?.Invoke(Source); + seedReplica?.Invoke(Replica); + Database = new CultNetDatabase(Source); + var wire = new LoopbackServer(); + _server = new CultNetDatabaseSubscriptionServer(wire, Database); + var documents = new CultNetDocumentRegistry(Replica.Registry) + .Register(CultNetDocumentBinding.ForDocument(Replica.Registry)); + Client = new CultNetDatabaseSubscriptionClient(new Client(wire, Peer), Replica, documents); + Client.Faulted += Faults.Add; + Client.SubscribeAsync("items", schemaIds: new[] { Source.Registry.GetRequired().SchemaId }) + .GetAwaiter().GetResult(); + } + + public ReplicaItem? ReplicaItem(string key) => Replica.Get(new CultRecordKey(key)); + + public void Dispose() + { + Client.Dispose(); + _server.Dispose(); + Database.Dispose(); + } + } + + private static ReplicaItem Item(string sku) => new() { Sku = sku }; + + private static void Stage(CultCache cache, params (string Key, string Sku)[] items) => + Assert.That(cache.Commit(batch => + { + foreach (var (key, sku) in items) + batch.Upsert(typeof(ReplicaItem), Item(sku), new CultRecordKey(key)); + }), Is.True); + + private static async Task Until(Func condition) + { + var deadline = DateTime.UtcNow.AddSeconds(5); + while (!condition()) + { + if (DateTime.UtcNow > deadline) Assert.Fail("The replica did not reach the expected state."); + await Task.Delay(5); + } + } + + [Test] + public async Task ASwapOfTwoIndexValuesReachesTheReplicaIntact() + { + using var run = new Replication(source => Stage(source, ("a", "one"), ("b", "two"))); + Assert.That(run.ReplicaItem("a")!.Sku, Is.EqualTo("one")); + + Stage(run.Source, ("a", "two"), ("b", "one")); + await Until(() => run.ReplicaItem("a")!.Sku == "two"); + + Assert.That(run.ReplicaItem("b")!.Sku, Is.EqualTo("one")); + Assert.That(run.Client.Fault, Is.Null); + Assert.That(run.Faults, Is.Empty); + Assert.That(run.Replica.GetByIndex("sku", "two"), Is.SameAs(run.ReplicaItem("a"))); + var swap = run.Peer.Changes; + Assert.That(swap.Select(change => change.CommitId).Distinct().Single(), Is.Not.Null, "the commit is named on the wire"); + Assert.That(swap.Select(change => change.CommitCount), Is.All.EqualTo(2)); + } + + [Test] + public async Task AValueMovedToANewRecordWhileItsOldHolderIsRemovedReachesTheReplica() + { + using var run = new Replication(source => Stage(source, ("a", "one"))); + + Assert.That(run.Source.Commit(batch => + { + batch.Remove(new CultRecordKey("a")); + batch.Upsert(typeof(ReplicaItem), Item("one"), new CultRecordKey("c")); + }), Is.True); + await Until(() => run.ReplicaItem("c") != null); + + Assert.That(run.ReplicaItem("a"), Is.Null); + Assert.That(run.Client.Fault, Is.Null); + Assert.That(run.Replica.GetByIndex("sku", "one"), Is.SameAs(run.ReplicaItem("c"))); + } + + [Test] + public async Task ACommitOfOneChangeCarriesNoCommitFieldsAndIsApplied() + { + using var run = new Replication(); + + Stage(run.Source, ("a", "one")); + await Until(() => run.ReplicaItem("a") != null); + + var change = run.Peer.Changes.Single(); + Assert.That(change.CommitId, Is.Null); + Assert.That(change.CommitCount, Is.Null); + } + + [Test] + public async Task AReplicaThatRefusesACommitFaultsLoudlyAndAppliesNothingMore() + { + // The replica holds a record the source does not, under the value the source now gives to another record: a conflict + // the replica cannot resolve, not an intermediate state. + using var run = new Replication(seedReplica: replica => Stage(replica, ("local", "taken"))); + + Stage(run.Source, ("a", "taken"), ("b", "free")); + await Until(() => run.Client.Fault != null); + + Assert.That(run.Faults, Has.Count.EqualTo(1)); + Assert.That(run.Faults[0], Is.SameAs(run.Client.Fault)); + Assert.That(run.Faults[0].Message, Does.Contain("sku")); + Assert.That(run.ReplicaItem("a"), Is.Null, "the refused commit is not partly applied"); + Assert.That(run.ReplicaItem("b"), Is.Null, "the refused commit is not partly applied"); + + Stage(run.Source, ("c", "later")); + await Task.Delay(100); + Assert.That(run.ReplicaItem("c"), Is.Null, "a replica known to differ takes nothing on top"); + Assert.That(run.Faults, Has.Count.EqualTo(1)); + } + + [Test] + public void AnOlderPeerReadsTheChangeAndIgnoresTheCommitFields() + { + var tagged = new CultNetDatabaseChangeRawMessage + { + MessageId = "m", + SubscriptionId = "items", + ChangeKind = "removed", + RecordKey = "a", + SchemaId = "tests.replica_commit_item.v1", + CommitId = "7", + CommitCount = 2 + }; + + var legacy = MessagePackSerializer.Deserialize( + CultNetSchemaMessageSerialization.Serialize(tagged), CultNetSchemaMessageSerialization.Options); + + Assert.That(legacy.ChangeKind, Is.EqualTo("removed")); + Assert.That(legacy.RecordKey, Is.EqualTo("a")); + Assert.That(legacy.SubscriptionId, Is.EqualTo("items")); + } + + [Test] + public async Task ASwapFromAServerThatDoesNotTagCommitsStillFaultsInsteadOfDiverging() + { + // An older server sends the two halves of a swap untagged: each is judged alone and refused. The refusal is surfaced. + using var run = new Replication(source => Stage(source, ("a", "one"), ("b", "two"))); + var raw = (string key, string sku) => run.Database.Documents.ToRawRecord( + run.Source.Registry.GetRequired(), new CultRecordKey(key), Item(sku), "2026-01-01T00:00:00Z"); + + run.Peer.Client!.Emit(new CultNetDatabaseChangeRawMessage + { + MessageId = "x", SubscriptionId = "items", ChangeKind = "updated", Document = raw("a", "two") + }); + await Until(() => run.Client.Fault != null); + + Assert.That(run.ReplicaItem("a")!.Sku, Is.EqualTo("one")); + } + + [Test] + public async Task ARotationOfThreeValuesReachesTheReplicaIntact() + { + using var run = new Replication(source => Stage(source, ("a", "one"), ("b", "two"), ("c", "three"))); + + Stage(run.Source, ("a", "two"), ("b", "three"), ("c", "one")); + await Until(() => run.ReplicaItem("a")!.Sku == "two"); + + Assert.That(new[] { "a", "b", "c" }.Select(key => run.ReplicaItem(key)!.Sku), Is.EqualTo(new[] { "two", "three", "one" })); + Assert.That(run.Client.Fault, Is.Null); + Assert.That(run.Peer.Changes.Select(change => change.CommitCount), Is.All.EqualTo(3)); + } + + // The database server is the other producer of change messages; it names a commit the same way. + [Test] + public void TheDatabaseServerTagsTheChangesOfOneCommitAndNotALoneChange() + { + var cache = new CultCache(); + Stage(cache, ("a", "one"), ("b", "two")); + using var database = new CultNetDatabase(cache); + using var server = new Server(cache, ServerSecurityOptions.Development()); + using var databaseServer = new CultNetDatabaseServer(server, database); + var commits = new List(); + using var watch = database.WatchCommits().Subscribe(commits.Add); + var selection = new CultNetSelection + { + Schemas = new[] { cache.Registry.GetRequired().SchemaId }, + Projection = CultNetSelectionProjections.Document + }; + IReadOnlyList Deliver(CultNetDatabaseCommit commit) => + (IReadOnlyList)typeof(CultNetDatabaseServer) + .GetMethod("DeliverCommit", System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)! + .Invoke(databaseServer, new object[] { commit, "sub", selection, new HashSet(), (Action)(_ => { }) })!; + + Stage(cache, ("a", "two"), ("b", "one")); + Stage(cache, ("c", "three")); + + var swap = Deliver(commits[0]); + Assert.That(swap, Has.Count.EqualTo(2)); + Assert.That(swap.Select(change => change.CommitId).Distinct().Single(), Is.EqualTo(commits[0].Id)); + Assert.That(swap.Select(change => change.CommitCount), Is.All.EqualTo(2)); + var lone = Deliver(commits[1]).Single(); + Assert.That(lone.CommitId, Is.Null); + Assert.That(lone.CommitCount, Is.Null); + Assert.That(commits[0].Id, Is.Not.EqualTo(commits[1].Id)); + } + } +}