CultNet: a replica applies the changes of one server commit as one cache commit; a refused commit faults the client

The database publishes commits (WatchCommits); the subscription server and the database server tag the changes a subscription receives from one commit with commitId and commitCount (optional wire fields older peers ignore). The C# subscription client buffers a tagged commit and applies it through one CultCache.Commit, judged on its result, so a swap of [CultIndex] values replicates. A refusal sets Fault, raises Faulted, and stops further applying. Snapshots apply as one commit too.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

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

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": [
     {

src/GameCult.Networking/CultNetDatabase.cs

@@ -315,6 +315,22 @@ namespace GameCult.Networking
         public CultNetSelectionCursorKey? CursorKey { get; set; }
     }
 
+    /// <summary>
+    /// The typed changes (<see cref="CultNetDatabaseChange{T}"/> boxed) one cache commit made, in the order the cache admitted them.
+    /// <see cref="Id"/> is opaque and unique among the commits of one database instance.
+    /// </summary>
+    public sealed class CultNetDatabaseCommit
+    {
+        public CultNetDatabaseCommit(string id, IReadOnlyList<object> changes)
+        {
+            Id = id;
+            Changes = changes;
+        }
+
+        public string Id { get; }
+        public IReadOnlyList<object> Changes { get; }
+    }
+
     /// <summary>
     /// A typed domain change emitted by a CultNet database surface.
     /// </summary>
@@ -587,6 +603,7 @@ namespace GameCult.Networking
         private readonly Dictionary<(string SchemaId, string RecordKey), long> _lastWriteSequence =
             new();
         private readonly Subject<object> _changes = new();
+        private readonly Subject<CultNetDatabaseCommit> _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;
         }
 
+        /// <summary>
+        /// Watches changes as the cache commits that made them: one event per commit, after every change of it was published to
+        /// <see cref="WatchAllChanges"/>. 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.
+        /// </summary>
+        public Observable<CultNetDatabaseCommit> WatchCommits()
+        {
+            ThrowIfDisposed();
+            return _commits;
+        }
+
         /// <summary>
         /// Watches one record.
         /// </summary>
@@ -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<Exception>? 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<CultNetBurnedSequence>? burned = null;
             List<Exception>? 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<object> 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.

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<string>(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<CultNetDatabaseChangeRawMessage> DeliverCommit(
+            CultNetDatabaseCommit commit,
+            string subscriptionId,
+            CultNetSelection selection,
+            HashSet<string> refused,
+            Action<CultNetErrorMessage> sendError)
+        {
+            var outbound = new List<CultNetDatabaseChangeRawMessage>();
+            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(

src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs

@@ -29,6 +29,8 @@ namespace GameCult.Networking
         private readonly ConcurrentDictionary<string, PendingSubscription> _initialSnapshots = new();
         private readonly ConcurrentDictionary<string, CultNetDatabaseSubscriptionDeliveryMode> _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<CultNetDatabaseChangeRawMessage>> _commits = new();
         private readonly object _queueLock = new();
         private Task _queue = Task.CompletedTask;
         private bool _disposed;
@@ -48,6 +50,15 @@ namespace GameCult.Networking
         /// <summary>Raised after a live change has been applied to the local typed cache.</summary>
         public event Action<CultNetReplicatedDocumentChange>? Changed;
 
+        /// <summary>
+        /// Raised when the local cache refuses a replicated commit. The replica then differs from its source; the client applies
+        /// no further changes and reports <see cref="Fault"/> until it is replaced by a new subscription.
+        /// </summary>
+        public event Action<Exception>? Faulted;
+
+        /// <summary>The refusal that stopped replication, or null while the replica is intact.</summary>
+        public Exception? Fault { get; private set; }
+
         /// <summary>Gets the local reactive cache receiving replicated documents.</summary>
         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<CultNetDatabaseChangeRawMessage>();
+                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<CultNetDatabaseChangeRawMessage> 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<object?> 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;
         }
 

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<CultNetDatabaseChangeRawMessage> 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))

src/GameCult.Networking/CultNetDocumentRegistry.cs

@@ -722,6 +722,49 @@ namespace GameCult.Networking
             return document;
         }
 
+        /// <summary>
+        /// 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.
+        /// </summary>
+        internal IReadOnlyList<object?> ApplyRawChangesAsOneCommit(
+            CultCache cache,
+            IReadOnlyList<CultNetDatabaseChangeRawMessage> 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;
+        }
+
         /// <summary>
         /// 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
         }
 
         /// <summary>
-        /// 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.
         /// </summary>
-        public async Task<IReadOnlyList<object>> ApplyRawSnapshotResponseAsync(
+        public Task<IReadOnlyList<object>> ApplyRawSnapshotResponseAsync(
             CultCache cache,
             CultNetSnapshotResponseRawMessage response)
         {
-            var applied = new List<object>(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<IReadOnlyList<object>>(applied.Select(document => document!).ToArray());
         }
 
         internal CultDocumentDescriptor ResolveDescriptorForRawDocument(CultNetRawDocumentRecord document)

src/GameCult.Networking/CultNetSchemaMessages.cs

@@ -1179,6 +1179,17 @@ namespace GameCult.Networking
         /// Gets or sets the deleted schema id for removed changes.
         /// </summary>
         [Key("schemaId")] public string? SchemaId { get; set; }
+        /// <summary>
+        /// 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.
+        /// </summary>
+        [Key("commitId")] public string? CommitId { get; set; }
+        /// <summary>
+        /// Gets or sets how many change messages of this commit this subscription carries. Set with
+        /// <see cref="CommitId"/> when that is more than one; the receiver applies the commit once all have arrived.
+        /// </summary>
+        [Key("commitCount")] public int? CommitCount { get; set; }
     }
 
     /// <summary>

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<Type, Delegate> _handlers = new();
+            public void OnCultNet<T>(Func<T, ICultNetSchemaServerPeer, Task> callback) where T : ICultNetSchemaMessage => _handlers[typeof(T)] = callback;
+            public void RemoveCultNetMessageListener<T>(Delegate callback) where T : ICultNetSchemaMessage => _handlers.Remove(typeof(T));
+            public Task Dispatch<T>(T message, ICultNetSchemaServerPeer peer) where T : ICultNetSchemaMessage =>
+                ((Func<T, ICultNetSchemaServerPeer, Task>)_handlers[typeof(T)])(message, peer);
+        }
+
+        private sealed class Client : ICultNetSchemaClient
+        {
+            private readonly LoopbackServer _server;
+            private readonly Dictionary<Type, List<Delegate>> _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>(T message) where T : ICultNetSchemaMessage
+            {
+                if (message is CultNetDatabaseSubscribeMessage subscribe)
+                    _server.Dispatch(subscribe, _peer).GetAwaiter().GetResult();
+            }
+            public void OnCultNet<T>(Action<T> callback) where T : ICultNetSchemaMessage
+            {
+                if (!_handlers.TryGetValue(typeof(T), out var list)) _handlers[typeof(T)] = list = new List<Delegate>();
+                list.Add(callback);
+            }
+            public void Emit<T>(T message) where T : ICultNetSchemaMessage
+            {
+                if (_handlers.TryGetValue(typeof(T), out var list))
+                    foreach (var handler in list.ToArray()) ((Action<T>)handler)(message);
+            }
+            public void Dispose() { }
+        }
+
+        private sealed class Peer : ICultNetSchemaServerPeer
+        {
+            public Client? Client;
+            public ConcurrentQueue<byte[]> Sent { get; } = new();
+            public void SendCultNet<T>(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<T>(bytes));
+            }
+            public CultNetDatabaseChangeRawMessage[] Changes => Sent
+                .Select(bytes => CultNetSchemaMessageSerialization.Deserialize(bytes))
+                .OfType<CultNetDatabaseChangeRawMessage>().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<Exception> Faults { get; } = new();
+            public CultNetDatabase Database { get; }
+            private readonly CultNetDatabaseSubscriptionServer _server;
+
+            public Replication(Action<CultCache>? seedSource = null, Action<CultCache>? 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<ReplicaItem>(Replica.Registry));
+                Client = new CultNetDatabaseSubscriptionClient(new Client(wire, Peer), Replica, documents);
+                Client.Faulted += Faults.Add;
+                Client.SubscribeAsync("items", schemaIds: new[] { Source.Registry.GetRequired<ReplicaItem>().SchemaId })
+                    .GetAwaiter().GetResult();
+            }
+
+            public ReplicaItem? ReplicaItem(string key) => Replica.Get<ReplicaItem>(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<bool> 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<ReplicaItem>("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<ReplicaItem>("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<LegacyChange>(
+                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<ReplicaItem>(), 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<CultNetDatabaseCommit>();
+            using var watch = database.WatchCommits().Subscribe(commits.Add);
+            var selection = new CultNetSelection
+            {
+                Schemas = new[] { cache.Registry.GetRequired<ReplicaItem>().SchemaId },
+                Projection = CultNetSelectionProjections.Document
+            };
+            IReadOnlyList<CultNetDatabaseChangeRawMessage> Deliver(CultNetDatabaseCommit commit) =>
+                (IReadOnlyList<CultNetDatabaseChangeRawMessage>)typeof(CultNetDatabaseServer)
+                    .GetMethod("DeliverCommit", System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)!
+                    .Invoke(databaseServer, new object[] { commit, "sub", selection, new HashSet<string>(), (Action<CultNetErrorMessage>)(_ => { }) })!;
+
+            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));
+        }
+    }
+}