CultLib/tests/GameCult.Networking.Tests/CultNetJournalTests.cs

#nullable enable
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using GameCult.Caching;
using GameCult.Caching.MessagePack;
using NUnit.Framework;
using R3;
using static GameCult.Networking.Tests.NetworkingTests;

namespace GameCult.Networking.Tests
{
    // The database's shard log is written by a cache journal, under the cache gate, in cache order. A primary that
    // cannot log a committed change burns that change's sequence and compacts past it, so the log can lag the cache but
    // never lead it, and every replica behind the hole resynchronizes by snapshot.
    public sealed class CultNetJournalTests
    {
        private const string ShardId = "log-shard";

        private static readonly CultRecordKey One = new("tests:journal:one");
        private static readonly CultRecordKey Two = new("tests:journal:two");
        private static readonly CultRecordKey Three = new("tests:journal:three");

        private static NetworkSchemaNote Note(string text) => new() { Schema = "tests.networking_note.v1", Text = text };

        private static string SchemaId(CultCache cache) => cache.Registry.GetRequired<NetworkSchemaNote>().SchemaId;

        private static CultNetShardDescriptor Shard(CultCache cache, bool primary) =>
            new(ShardId, "primary", epoch: 1, isPrimary: primary, schemaIds: [SchemaId(cache)]);

        private static CultNetDatabase Database(CultCache cache, bool primary, FlakyLogStore? store) =>
            new(cache, new CultNetDatabaseOptions { Shards = [Shard(cache, primary)], MutationLogStore = store });

        private static string KeyOfChange(object change) => ((dynamic)change).Key.Value;

        private static List<string> Record(CultNetDatabase database)
        {
            var seen = new List<string>();
            database.WatchAllChanges().Subscribe(change => seen.Add(KeyOfChange(change)));
            return seen;
        }

        private static long[] Sequences(FlakyLogStore store) => store.Read(ShardId).Select(entry => entry.Sequence).ToArray();

        // A store whose refusals and side effects the test scripts. One shard per store.
        private sealed class FlakyLogStore : ICultNetShardMutationLogStore
        {
            private readonly object _gate = new();
            private readonly SortedDictionary<long, CultNetShardLogEntryMessage> _entries = new();
            private long _compactedThrough;

            public Func<long, bool> RefuseAppend { get; set; } = _ => false;
            public bool RefuseCompact { get; set; }
            public Action? OnRead { get; set; }
            public Action? OnAppend { get; set; }

            public IReadOnlyList<CultNetShardLogEntryMessage> Read(string shardId, long afterSequence = 0, int? limit = null)
            {
                OnRead?.Invoke();
                lock (_gate)
                {
                    var entries = _entries.Values.Where(entry => entry.Sequence > afterSequence);
                    return (limit.HasValue ? entries.Take(limit.Value) : entries).ToArray();
                }
            }

            public void Append(string shardId, CultNetShardLogEntryMessage entry)
            {
                OnAppend?.Invoke();
                lock (_gate)
                {
                    if (RefuseAppend(entry.Sequence))
                        throw new IOException($"The log refused sequence {entry.Sequence}.");
                    _entries[entry.Sequence] = entry;
                }
            }

            public long GetCompactedThrough(string shardId)
            {
                lock (_gate)
                    return _compactedThrough;
            }

            public void CompactThrough(string shardId, long sequence)
            {
                lock (_gate)
                {
                    if (RefuseCompact)
                        throw new IOException("The log refused to compact.");
                    _compactedThrough = Math.Max(_compactedThrough, sequence);
                    foreach (var stale in _entries.Keys.Where(key => key <= sequence).ToArray())
                        _entries.Remove(stale);
                }
            }
        }

        // The primary as a replica meets it: the database server's own log answer, and the database's own snapshot.
        private sealed class PrimaryPeer : ICultNetShardLogFetcher, ICultNetShardSnapshotFetcher, IDisposable
        {
            private readonly Server _server;
            private readonly CultNetDatabaseServer _databaseServer;
            private readonly CultNetDatabase _database;

            public PrimaryPeer(CultCache cache, CultNetDatabase database)
            {
                _database = database;
                _server = new Server(cache, ServerSecurityOptions.Development());
                _databaseServer = new CultNetDatabaseServer(_server, database);
            }

            public CultNetShardLogResponseMessage Log(long afterSequence) =>
                _databaseServer.CreateShardLogResponse(new CultNetShardLogRequestMessage
                {
                    MessageId = "pull",
                    ShardId = ShardId,
                    ShardEpoch = 1,
                    AfterSequence = afterSequence
                });

            public Task<CultNetShardLogResponseMessage> FetchAsync(CultNetShardDescriptor shard, long afterSequence, int? limit = null) =>
                Task.FromResult(Log(afterSequence));

            public Task<CultNetSnapshotResponseRawMessage> FetchAsync(CultNetShardDescriptor shard) =>
                Task.FromResult(_database.CreateShardSnapshotResponse(_database.Shards[0], "snapshot"));

            public void Dispose()
            {
                _databaseServer.Dispose();
                _server.Dispose();
            }
        }

        private static CultNetShardReplicator Replicator(CultNetDatabase replica, PrimaryPeer peer) =>
            new(replica, new CultNetShardReplicatorOptions { Fetcher = peer, SnapshotFetcher = peer });

        private static string? TextOf(CultCache cache, CultRecordKey key) => cache.Get<NetworkSchemaNote>(key)?.Text;

        [Test]
        public async Task ARefusedPrimaryAppendPublishesTheChangeTheWriterHearsOfItAndAReplicaBehindItResyncs()
        {
            var store = new FlakyLogStore { RefuseAppend = sequence => sequence == 2 };
            var cache = new CultCache();
            var primary = Database(cache, primary: true, store);
            using var peer = new PrimaryPeer(cache, primary);
            var replicaCache = new CultCache();
            var replica = Database(replicaCache, primary: false, store: null);
            using var replicator = Replicator(replica, peer);
            await primary.PutAsync(One, Note("one"));
            Assert.That(await replicator.PullOnceAsync(ShardId), Is.EqualTo(1));
            var published = Record(primary);

            var failure = Assert.ThrowsAsync<CultNetShardLogException>(async () => await primary.PutAsync(Two, Note("two")))!;

            Assert.That(failure.Burned, Is.EqualTo(new[] { new CultNetBurnedSequence(ShardId, 2) }));
            Assert.That(failure, Is.Not.InstanceOf<InvalidOperationException>(), "a committed write must not read as a refused one");
            Assert.That(published, Is.EqualTo(new[] { Two.Value }), "the change was published before the writer heard");
            Assert.That(TextOf(cache, Two), Is.EqualTo("two"), "the commit stands");
            Assert.That(primary.LastWriteSequence(SchemaId(cache), Two), Is.EqualTo(2L), "the row was written at the burned sequence");
            await primary.PutAsync(Three, Note("three"));
            Assert.That(Sequences(store), Is.EqualTo(new[] { 3L }), "sequence 2 stays a hole below the compaction point");
            Assert.That(peer.Log(afterSequence: 1).Reason, Is.EqualTo("compacted_history"));

            Assert.That(await replicator.PullOnceAsync(ShardId), Is.EqualTo(3));
            Assert.That(new[] { One, Two, Three }.Select(key => TextOf(replicaCache, key)), Is.EqualTo(new[] { "one", "two", "three" }));
            Assert.That(await replicator.PullOnceAsync(ShardId), Is.EqualTo(3), "a later pull is a quiet no-op");
        }

        [Test]
        public async Task ARestartAfterARefusedTailAppendReusesNoSequence()
        {
            var store = new FlakyLogStore { RefuseAppend = sequence => sequence == 2 };
            var cache = new CultCache();
            var primary = Database(cache, primary: true, store);
            var replicaCache = new CultCache();
            var replica = Database(replicaCache, primary: false, store: null);
            using (var before = new PrimaryPeer(cache, primary))
            {
                await primary.PutAsync(One, Note("one"));
                using var replicator = Replicator(replica, before);
                await replicator.PullOnceAsync(ShardId);
            }

            Assert.ThrowsAsync<CultNetShardLogException>(async () => await primary.PutAsync(Two, Note("two")));
            primary.Dispose();

            var restarted = Database(cache, primary: true, store);
            await restarted.PutAsync(Three, Note("three"));

            Assert.That(Sequences(store), Is.EqualTo(new[] { 3L }), "the burned sequence was not minted again");
            using var after = new PrimaryPeer(cache, restarted);
            using var resumed = Replicator(replica, after);
            await resumed.PullOnceAsync(ShardId);
            Assert.That(new[] { One, Two, Three }.Select(key => TextOf(replicaCache, key)), Is.EqualTo(new[] { "one", "two", "three" }));
        }

        [Test]
        public async Task AStoreThatRefusesTheAppendAndTheCompactionStillCompactsInMemoryAndNamesBothFailures()
        {
            var store = new FlakyLogStore { RefuseAppend = sequence => sequence == 2, RefuseCompact = true };
            var cache = new CultCache();
            var primary = Database(cache, primary: true, store);
            using var peer = new PrimaryPeer(cache, primary);
            var replicaCache = new CultCache();
            var replica = Database(replicaCache, primary: false, store: null);
            using var replicator = Replicator(replica, peer);
            await primary.PutAsync(One, Note("one"));
            await replicator.PullOnceAsync(ShardId);

            var failure = Assert.ThrowsAsync<CultNetShardLogException>(async () => await primary.PutAsync(Two, Note("two")))!;

            Assert.That(failure.InnerException, Is.InstanceOf<AggregateException>());
            Assert.That(((AggregateException)failure.InnerException!).InnerExceptions, Has.Count.EqualTo(2));
            Assert.That(primary.GetCompactedMutationLogSequence(ShardId), Is.EqualTo(2), "the primary's own floor covers a hole its store never compacted");
            Assert.That(primary.GetLatestMutationLogSequence(ShardId), Is.EqualTo(2), "the latest sequence is never below the floor");
            Assert.That(peer.Log(afterSequence: 1).Reason, Is.EqualTo("compacted_history"));
            Assert.That(await replicator.PullOnceAsync(ShardId), Is.EqualTo(2));
            Assert.That(TextOf(replicaCache, Two), Is.EqualTo("two"));
        }

        [Test]
        public async Task CompactingAndRestartingResumesTheNextSequenceAboveTheCompactionPoint()
        {
            var store = new FlakyLogStore();
            var cache = new CultCache();
            var primary = Database(cache, primary: true, store);
            await primary.PutAsync(One, Note("one"));
            await primary.PutAsync(Two, Note("two"));
            store.CompactThrough(ShardId, 2);
            primary.Dispose();

            var restarted = Database(cache, primary: true, store);
            Assert.That(restarted.GetLatestMutationLogSequence(ShardId), Is.EqualTo(2), "the latest sequence is never below the compaction point");
            await restarted.PutAsync(Three, Note("three"));

            Assert.That(Sequences(store), Is.EqualTo(new[] { 3L }));
            Assert.That(restarted.GetLatestMutationLogSequence(ShardId), Is.EqualTo(3));
        }

        [Test]
        public async Task ARefusedReplicaAppendRecordsNothingAdvancesNoCursorAndTheRetryConverges()
        {
            var refusals = 0;
            var replicaStore = new FlakyLogStore { RefuseAppend = sequence => sequence == 1 && refusals++ == 0 };
            var cache = new CultCache();
            var primary = Database(cache, primary: true, new FlakyLogStore());
            await primary.PutAsync(One, Note("one"));
            await primary.PutAsync(Two, Note("two"));
            using var peer = new PrimaryPeer(cache, primary);
            var replicaCache = new CultCache();
            var replica = Database(replicaCache, primary: false, replicaStore);
            using var replicator = Replicator(replica, peer);

            Assert.ThrowsAsync<IOException>(async () => await replicator.PullOnceAsync(ShardId));

            Assert.That(replica.GetAppliedShardSequence(ShardId), Is.EqualTo(0));
            Assert.That(replica.GetMutationLog(ShardId), Is.Empty, "memory never holds what the durable log refused");
            Assert.That(await replicator.PullOnceAsync(ShardId), Is.EqualTo(2));
            Assert.That(replica.GetMutationLog(ShardId).Select(entry => entry.Sequence), Is.EqualTo(new[] { 1L, 2L }));
            Assert.That(Sequences(replicaStore), Is.EqualTo(new[] { 1L, 2L }));
        }

        [Test]
        public async Task AReentrantShardLogApplyFromAHandlerKeepsTheReplicaLogInPrimaryOrder()
        {
            var cache = new CultCache();
            var primary = Database(cache, primary: true, new FlakyLogStore());
            await primary.PutAsync(One, Note("one"));
            await primary.PutAsync(Two, Note("two"));
            await primary.PutAsync(Three, Note("three"));
            var response = new CultNetShardLogResponseMessage
            {
                ShardId = ShardId,
                ShardEpoch = 1,
                Entries = primary.GetMutationLogMessages(ShardId).ToArray()
            };
            var replicaStore = new FlakyLogStore();
            var replicaCache = new CultCache();
            var replica = Database(replicaCache, primary: false, replicaStore);
            var nested = false;
            long nestedApplied = 0;
            Exception? nestedFailure = null;
            replica.Watch<NetworkSchemaNote>().Where(change => change.Key.Equals(One)).Subscribe(_ =>
            {
                if (nested) return;
                nested = true;
                try
                {
                    nestedApplied = replica.ApplyShardLogResponseAsync(response).GetAwaiter().GetResult();
                }
                catch (Exception exception)
                {
                    nestedFailure = exception;
                }
            });

            await replica.ApplyShardLogResponseAsync(response);

            Assert.That(nested, Is.True);
            Assert.That(nestedFailure, Is.Null, "the nested apply re-records entries the outer apply already logged");
            Assert.That(nestedApplied, Is.EqualTo(3));
            Assert.That(replica.GetMutationLog(ShardId).Select(entry => entry.Sequence), Is.EqualTo(new[] { 1L, 2L, 3L }));
            Assert.That(Sequences(replicaStore), Is.EqualTo(new[] { 1L, 2L, 3L }));
            Assert.That(replica.GetAppliedShardSequence(ShardId), Is.EqualTo(3));
            Assert.That(replica.CurrentAsOf(ShardId), Is.EqualTo(3UL), "the replica's own watermark follows the primary's sequences");
        }

        [Test]
        public async Task AReentrantDatabaseWriteFromAHandlerIsLoggedWithItsContextAndPublishedOnce()
        {
            var store = new FlakyLogStore();
            var cache = new CultCache();
            var database = Database(cache, primary: true, store);
            var message = database.Documents.CreateRawDocumentPutMessage(
                "reentrant-two", new CultRecordHandle<NetworkSchemaNote>(Two), Note("two"));
            var published = Record(database);
            database.Watch<NetworkSchemaNote>().Where(change => change.Key.Equals(One)).Subscribe(_ =>
                database.ApplyPutAsync(message).GetAwaiter().GetResult());

            await database.PutAsync(One, Note("one"));

            Assert.That(published, Is.EqualTo(new[] { One.Value, Two.Value }));
            Assert.That(database.GetMutationLog(ShardId).Select(entry => entry.Key.Value), Is.EqualTo(new[] { One.Value, Two.Value }));
            Assert.That(store.Read(ShardId).Single(entry => entry.Sequence == 2).Put!.MessageId, Is.EqualTo("reentrant-two"),
                "the nested write's own wire message is what the log stored");
        }

        [Test]
        public async Task ALogStoreThatWritesTheCacheFailsThroughTheReentryGuardAndTheAdmissionStands()
        {
            var cache = new CultCache();
            var store = new FlakyLogStore();
            store.OnAppend = () =>
            {
                store.OnAppend = null;
                cache.UpsertAsync(Note("from the store"), new CultRecordHandle<NetworkSchemaNote>(Three)).GetAwaiter().GetResult();
            };
            var database = Database(cache, primary: true, store);
            var published = Record(database);

            var failure = Assert.ThrowsAsync<CultNetShardLogException>(async () => await database.PutAsync(One, Note("one")))!;

            Assert.That(failure.InnerException, Is.InstanceOf<InvalidOperationException>(), "the cache's guard refused the store's write");
            Assert.That(TextOf(cache, One), Is.EqualTo("one"));
            Assert.That(TextOf(cache, Three), Is.Null);
            Assert.That(published, Is.EqualTo(new[] { One.Value }));
        }

        [Test]
        public void SeveralRefusedAppendsInOneAdmissionAreAllPublishedAndAllReachTheWriter()
        {
            var store = new FlakyLogStore { RefuseAppend = _ => true };
            var cache = new CultCache();
            var database = Database(cache, primary: true, store);
            var published = Record(database);

            var failure = Assert.Throws<CultNetShardLogException>(() => cache.Commit(batch =>
            {
                batch.Upsert(Note("one"), new CultRecordHandle<NetworkSchemaNote>(One));
                batch.Upsert(Note("two"), new CultRecordHandle<NetworkSchemaNote>(Two));
            }))!;

            Assert.That(failure.Burned, Is.EqualTo(new[] { new CultNetBurnedSequence(ShardId, 1), new CultNetBurnedSequence(ShardId, 2) }), "one exception carries every burned sequence");
            Assert.That(((AggregateException)failure.InnerException!).InnerExceptions, Has.Count.EqualTo(2), "and every store failure");
            Assert.That(published, Is.EqualTo(new[] { One.Value, Two.Value }));
            Assert.That(database.GetCompactedMutationLogSequence(ShardId), Is.EqualTo(2));
        }

        [Test]
        public async Task APullOnAPrimaryIsLoggedSoAReplicaConverges()
        {
            var path = Path.Combine(Path.GetTempPath(), $"cultnet-journal-pull-{Guid.NewGuid():N}.cc");
            try
            {
                var writer = new CultCache();
                writer.AddBackingStore(new SingleFileMessagePackBackingStore(path));
                await writer.UpsertAsync(Note("seed"), new CultRecordHandle<NetworkSchemaNote>(One));
                writer.FlushAllBackingStores();
                var cache = new CultCache();
                cache.AddBackingStore(new SingleFileMessagePackBackingStore(path));
                var primary = Database(cache, primary: true, new FlakyLogStore());
                using var peer = new PrimaryPeer(cache, primary);
                var replicaCache = new CultCache();
                var replica = Database(replicaCache, primary: false, store: null);
                using var replicator = Replicator(replica, peer);

                await writer.UpsertAsync(Note("pulled"), new CultRecordHandle<NetworkSchemaNote>(Two));
                writer.FlushAllBackingStores();
                await cache.PullAllBackingStoresAsync();
                await replicator.PullOnceAsync(ShardId);

                Assert.That(primary.GetMutationLog(ShardId).Select(entry => entry.Key.Value), Does.Contain(Two.Value));
                Assert.That(TextOf(replicaCache, Two), Is.EqualTo("pulled"));
            }
            finally
            {
                if (File.Exists(path)) File.Delete(path);
                if (File.Exists(path + ".lock")) File.Delete(path + ".lock");
            }
        }

        [Test]
        public async Task AStoreAttachedAfterTheDatabaseExistsHasItsHydrationLogged()
        {
            var path = Path.Combine(Path.GetTempPath(), $"cultnet-journal-attach-{Guid.NewGuid():N}.cc");
            try
            {
                var writer = new CultCache();
                writer.AddBackingStore(new SingleFileMessagePackBackingStore(path));
                await writer.UpsertAsync(Note("one"), new CultRecordHandle<NetworkSchemaNote>(One));
                await writer.UpsertAsync(Note("two"), new CultRecordHandle<NetworkSchemaNote>(Two));
                writer.FlushAllBackingStores();
                var cache = new CultCache();
                var database = Database(cache, primary: true, new FlakyLogStore());

                cache.AddBackingStore(new SingleFileMessagePackBackingStore(path));

                Assert.That(database.GetMutationLog(ShardId).Select(entry => entry.Key.Value), Is.EquivalentTo(new[] { One.Value, Two.Value }));
            }
            finally
            {
                if (File.Exists(path)) File.Delete(path);
                if (File.Exists(path + ".lock")) File.Delete(path + ".lock");
            }
        }

        [Test]
        public async Task ASnapshotReadsItsLogSequenceBeforeItsDocumentsSoARacingWriteConverges()
        {
            var store = new FlakyLogStore();
            var cache = new CultCache();
            var primary = Database(cache, primary: true, store);
            await primary.PutAsync(One, Note("one"));
            using var peer = new PrimaryPeer(cache, primary);
            var replicaCache = new CultCache();
            var replica = Database(replicaCache, primary: false, store: null);
            using var replicator = Replicator(replica, peer);
            store.OnRead = () =>
            {
                store.OnRead = null;
                primary.PutAsync(Two, Note("two")).GetAwaiter().GetResult();
            };

            var snapshot = primary.CreateShardSnapshotResponse(primary.Shards[0], "snapshot");
            await replica.ApplyShardSnapshotResponseAsync(replica.Shards[0], snapshot);
            await replicator.PullOnceAsync(ShardId);

            Assert.That(TextOf(replicaCache, Two), Is.EqualTo("two"), "a write racing the snapshot reaches the replica by one road or the other");
        }

        [Test]
        public async Task ASnapshotAppliedToAPrimaryShardLogsNothing()
        {
            var cache = new CultCache();
            await cache.UpsertAsync(Note("stale"), new CultRecordHandle<NetworkSchemaNote>(One));
            var database = Database(cache, primary: true, new FlakyLogStore());
            var put = database.Documents.CreateRawDocumentPutMessage("snapshot-two", new CultRecordHandle<NetworkSchemaNote>(Two), Note("two"));

            await database.ApplyShardSnapshotResponseAsync(database.Shards[0], new CultNetSnapshotResponseRawMessage
            {
                ShardId = ShardId,
                ShardEpoch = 1,
                Documents = [put.Document!],
                ShardLogSequence = 5
            });

            Assert.That(TextOf(cache, One), Is.Null);
            Assert.That(TextOf(cache, Two), Is.EqualTo("two"));
            Assert.That(database.GetMutationLog(ShardId), Is.Empty, "the snapshot's removal and its put are both a resync, not a commit");
        }

        [Test]
        public async Task ARestartOverARetainedLogResumesAboveItsHighestSequence()
        {
            var store = new FlakyLogStore();
            var cache = new CultCache();
            var primary = Database(cache, primary: true, store);
            await primary.PutAsync(One, Note("one"));
            await primary.PutAsync(Two, Note("two"));
            primary.Dispose();

            var restarted = Database(cache, primary: true, store);
            Assert.That(restarted.GetLatestMutationLogSequence(ShardId), Is.EqualTo(2), "a restarted database reads its latest sequence from the durable log");
            await restarted.PutAsync(Three, Note("three"));

            Assert.That(Sequences(store), Is.EqualTo(new[] { 1L, 2L, 3L }));
        }

        [Test]
        public async Task APutAndDeleteDoorLogTheirOwnWireMessagesAndChangeKinds()
        {
            var store = new FlakyLogStore();
            var cache = new CultCache();
            var database = Database(cache, primary: true, store);
            var added = database.Documents.CreateRawDocumentPutMessage("put-added", new CultRecordHandle<NetworkSchemaNote>(One), Note("one"));
            var updated = database.Documents.CreateRawDocumentPutMessage("put-updated", new CultRecordHandle<NetworkSchemaNote>(One), Note("uno"));

            await database.ApplyPutAsync(added);
            await database.ApplyPutAsync(updated);
            await database.ApplyDeleteAsync(new CultNetDocumentDeleteMessage { MessageId = "delete-one", SchemaId = SchemaId(cache), RecordKey = One.Value });

            var entries = store.Read(ShardId);
            Assert.That(entries.Select(entry => entry.ChangeKind), Is.EqualTo(new[] { "added", "updated", "removed" }));
            Assert.That(entries[0].Put!.MessageId, Is.EqualTo("put-added"));
            Assert.That(entries[1].Put!.MessageId, Is.EqualTo("put-updated"));
            Assert.That(entries[2].Delete!.MessageId, Is.EqualTo("delete-one"));
            Assert.That(DateTimeOffset.TryParse(entries[0].CommittedAt, out _), Is.True, "a primary stamps its entries");
        }

        [Test]
        public async Task AReconciledPutIsLoggedAsTheUpdateItIs()
        {
            var cache = new CultCache();
            var database = new CultNetDatabase(cache, new CultNetDatabaseOptions
            {
                RuntimeId = "local",
                ClientAuthorityScopes = [new CultNetClientAuthorityScope("local")]
            });
            var shard = database.Shards[0];
            var message = database.Documents.CreateRawDocumentPutMessage("authoritative", new CultRecordHandle<NetworkSchemaNote>(One), Note("authoritative"));
            message.ShardId = shard.ShardId;
            message.ShardEpoch = shard.Epoch;
            var published = new List<CultNetDatabaseChangeKind>();
            database.Watch<NetworkSchemaNote>().Subscribe(change => published.Add(change.Kind));

            await database.PutPredictedAsync(One, Note("predicted"));
            await database.ApplyPutAsync(message);

            Assert.That(published, Is.EqualTo(new[] { CultNetDatabaseChangeKind.Predicted, CultNetDatabaseChangeKind.Reconciled }));
            Assert.That(database.GetMutationLog(shard.ShardId).Select(entry => entry.Kind), Is.EqualTo(new[] { CultNetDatabaseChangeKind.Updated }),
                "a prediction is never logged and its reconciliation is an ordinary update");
        }

        [Test]
        public async Task AnAdmissionRacingDisposalLogsNothingAfterTheDatabaseIsDisposed()
        {
            var store = new FlakyLogStore();
            var cache = new CultCache();
            var entered = new System.Threading.ManualResetEventSlim();
            CultNetDatabase? database = null;
            // Registered before the database, so it runs first inside the writer's hold, while another thread disposes the database.
            using var holdUntilDisposed = cache.AddJournal(_ =>
            {
                entered.Set();
                var deadline = DateTime.UtcNow.AddSeconds(5);
                while (DateTime.UtcNow < deadline)
                {
                    try
                    {
                        database!.GetLatestMutationLogSequence(ShardId);
                    }
                    catch (ObjectDisposedException)
                    {
                        return;
                    }

                    System.Threading.Thread.Sleep(1);
                }
            });
            database = Database(cache, primary: true, store);
            var writer = Task.Run(() => database.PutAsync(One, Note("one")));
            Assert.That(entered.Wait(TimeSpan.FromSeconds(5)), Is.True);

            database.Dispose();
            await writer;

            Assert.That(Sequences(store), Is.Empty, "the admission was journaled after disposal began and must not reach the durable log");
        }

        [Test]
        public void ABatchAcrossTwoShardsWhoseStoreRefusesReachesTheWriterAsOneExceptionNamingBothShards()
        {
            var store = new FlakyLogStore { RefuseAppend = _ => true };
            var cache = new CultCache();
            _ = new CultNetDatabase(cache, new CultNetDatabaseOptions
            {
                Shards =
                [
                    new CultNetShardDescriptor("shard-a", "primary", epoch: 1, isPrimary: true, schemaIds: [SchemaId(cache)], keyPrefix: "a:"),
                    new CultNetShardDescriptor("shard-b", "primary", epoch: 1, isPrimary: true, schemaIds: [SchemaId(cache)], keyPrefix: "b:")
                ],
                MutationLogStore = store
            });

            var failure = Assert.Throws<CultNetShardLogException>(() => cache.Commit(batch =>
            {
                batch.Upsert(Note("a"), new CultRecordHandle<NetworkSchemaNote>(new CultRecordKey("a:1")));
                batch.Upsert(Note("b"), new CultRecordHandle<NetworkSchemaNote>(new CultRecordKey("b:1")));
            }))!;

            Assert.That(failure.Burned, Is.EqualTo(new[] { new CultNetBurnedSequence("shard-a", 1), new CultNetBurnedSequence("shard-b", 1) }));
        }

        // The cache aggregates whatever else fails in the same hold; the documented unwrap finds the log failure inside.
        [Test]
        public void ALogFailureBesideAnotherFailureInTheSameHoldIsFoundByUnwrappingTheAggregate()
        {
            var store = new FlakyLogStore { RefuseAppend = _ => true };
            var cache = new CultCache();
            _ = Database(cache, primary: true, store);
            using var second = cache.AddJournal(_ => throw new InvalidOperationException("second journal"));

            var failure = Assert.Throws<AggregateException>(() => cache.Commit(batch =>
            {
                batch.Upsert(Note("one"), new CultRecordHandle<NetworkSchemaNote>(One));
                batch.Upsert(Note("two"), new CultRecordHandle<NetworkSchemaNote>(Two));
            }))!;

            var logFailure = failure.InnerExceptions.OfType<CultNetShardLogException>().Single();
            Assert.That(logFailure.Burned.Select(burned => burned.Sequence), Is.EqualTo(new[] { 1L, 2L }));
            Assert.That(failure.InnerExceptions.OfType<InvalidOperationException>().Single().Message, Is.EqualTo("second journal"));
        }

        [Test]
        public async Task ABareWriteOfASchemaNoShardOwnsIsNeitherLoggedNorPublishedNorReplicated()
        {
            var store = new FlakyLogStore();
            var cache = new CultCache();
            var database = Database(cache, primary: true, store);
            var published = Record(database);

            await cache.UpsertAsync(new MeshQuickstartNote { NoteId = "n", Body = "unrelated" }, new CultRecordHandle<MeshQuickstartNote>(new CultRecordKey("unrelated:n")));
            await cache.UpsertAsync(Note("one"), new CultRecordHandle<NetworkSchemaNote>(One));

            Assert.That(Sequences(store), Is.EqualTo(new[] { 1L }), "the unrelated write minted no sequence");
            Assert.That(database.GetMutationLog(ShardId).Select(entry => entry.Key.Value), Is.EqualTo(new[] { One.Value }));
            Assert.That(published, Is.EqualTo(new[] { One.Value }), "no shard owns the unrelated write, so the database does not publish it");
        }

        // The database is the sharded write door: a schema and key no shard owns cannot be written on any path, and a refused
        // write caches and logs nothing.
        private static (CultNetDatabase Database, CultCache Cache, FlakyLogStore Store, string SchemaId) OwnedNothingOfNote()
        {
            var store = new FlakyLogStore();
            var cache = new CultCache();
            return (Database(cache, primary: true, store), cache, store, cache.Registry.GetRequired<MeshQuickstartNote>().SchemaId);
        }

        private static readonly CultRecordKey Unowned = new("unrelated:note");

        private static CultNetDocumentPutRawMessage UnownedPut(CultNetDatabase database)
        {
            var message = database.Documents.CreateRawDocumentPutMessage("wire", new CultRecordHandle<MeshQuickstartNote>(Unowned), new MeshQuickstartNote { NoteId = "w", Body = "wire" });
            message.ShardId = database.Shards[0].ShardId;
            message.ShardEpoch = database.Shards[0].Epoch;
            return message;
        }

        [Test]
        public void ADatabasePutOrDeleteOfASchemaNoShardOwnsIsRefusedNamingTheSchema()
        {
            var (database, cache, store, unowned) = OwnedNothingOfNote();

            var put = Assert.ThrowsAsync<CultNetUnownedSchemaException>(async () =>
                await database.PutAsync(Unowned, new MeshQuickstartNote { NoteId = "p", Body = "plain" }))!;
            Assert.That(put.SchemaId, Is.EqualTo(unowned));
            Assert.That(put.Message, Does.Contain(unowned));
            Assert.That(put, Is.InstanceOf<InvalidOperationException>(), "a refused write reads as one");
            Assert.That(cache.Get<MeshQuickstartNote>(Unowned), Is.Null, "nothing was cached");
            Assert.ThrowsAsync<CultNetUnownedSchemaException>(async () => await database.DeleteAsync<MeshQuickstartNote>(Unowned));
            Assert.That(Sequences(store), Is.Empty);
        }

        [Test]
        public void ARemoteAuthoritativePutOfASchemaNoShardOwnsIsRefusedAndCachesNothing()
        {
            var (database, cache, store, unowned) = OwnedNothingOfNote();

            var refusal = Assert.ThrowsAsync<CultNetUnownedSchemaException>(async () => await database.ApplyPutAsync(UnownedPut(database)))!;

            Assert.That(refusal.SchemaId, Is.EqualTo(unowned));
            Assert.That(cache.Get<MeshQuickstartNote>(Unowned), Is.Null);
            Assert.That(Sequences(store), Is.Empty);
        }

        [Test]
        public async Task ARemoteDeleteOfASchemaNoShardOwnsIsRefusedEvenWhenTheKeyIsCached()
        {
            var (database, cache, store, unowned) = OwnedNothingOfNote();
            await cache.UpsertAsync(new MeshQuickstartNote { NoteId = "c", Body = "cached" }, new CultRecordHandle<MeshQuickstartNote>(Unowned));

            var refusal = Assert.ThrowsAsync<CultNetUnownedSchemaException>(async () => await database.ApplyDeleteAsync(new CultNetDocumentDeleteMessage
            {
                MessageId = "d",
                SchemaId = unowned,
                RecordKey = Unowned.Value,
                ShardId = database.Shards[0].ShardId,
                ShardEpoch = database.Shards[0].Epoch
            }))!;

            Assert.That(refusal.SchemaId, Is.EqualTo(unowned));
            Assert.That(cache.Get<MeshQuickstartNote>(Unowned), Is.Not.Null, "the refused delete removed nothing");
            Assert.That(Sequences(store), Is.Empty);
        }

        [Test]
        public void APredictionOfASchemaNoShardOwnsIsRefused()
        {
            var cache = new CultCache();
            var database = new CultNetDatabase(cache, new CultNetDatabaseOptions
            {
                RuntimeId = "local",
                Shards = [new CultNetShardDescriptor("only-notes", "local", epoch: 1, isPrimary: true, schemaIds: [SchemaId(cache)])],
                ClientAuthorityScopes = [new CultNetClientAuthorityScope("local")]
            });

            Assert.ThrowsAsync<CultNetUnownedSchemaException>(async () =>
                await database.PutPredictedAsync(Unowned, new MeshQuickstartNote { NoteId = "p", Body = "predicted" }));
            Assert.That(cache.Get<MeshQuickstartNote>(Unowned), Is.Null);
        }

        [Test]
        public void ASchemaNoShardOwnsCannotBeWrittenAuthoritativelyAndHasNoShardToResolve()
        {
            var (database, cache, _, unowned) = OwnedNothingOfNote();

            Assert.That(database.CanWriteAuthoritatively<MeshQuickstartNote>(Unowned), Is.False);
            Assert.That(database.CanWriteAuthoritatively<NetworkSchemaNote>(One), Is.True, "an owned schema is unaffected");
            var refusal = Assert.Throws<CultNetUnownedSchemaException>(() => database.ResolveShard(unowned, Unowned))!;
            Assert.That(refusal.SchemaId, Is.EqualTo(unowned));
            Assert.That(database.ResolveShard(SchemaId(cache), One).ShardId, Is.EqualTo(ShardId));
        }

        // The server answers a refused remote put the way it answers any other: it logs the failure and sends the peer a
        // CultNetErrorMessage carrying the refusal. The handler is reached through its private delegate with an
        // uninitialized peer, as the R-AM test does; the send then fails on that peer, after the failure was logged.
        [Test]
        public async Task TheServerAnswersARemotePutOfASchemaNoShardOwnsWithAnApplicationRejection()
        {
            var (database, cache, _, unowned) = OwnedNothingOfNote();
            using var server = new Server(cache, ServerSecurityOptions.Development());
            var logger = new CapturingLogger();
            server.Logger = logger;
            using var databaseServer = new CultNetDatabaseServer(server, database);
            var handler = (Func<CultNetDocumentPutRawMessage, CultNetServerPeer, Task>)typeof(CultNetDatabaseServer)
                .GetField("_putHandler", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance)!
                .GetValue(databaseServer)!;
            var peer = (CultNetServerPeer)System.Runtime.CompilerServices.RuntimeHelpers.GetUninitializedObject(typeof(CultNetServerPeer));

            try { await handler(UnownedPut(database), peer); } catch (Exception) { }

            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<MeshQuickstartNote>(Unowned), Is.Null);
        }

        // One cached row of a schema no shard owns is not part of what this database serves: the selection answers with the
        // rest, it does not fail as a whole.
        [Test]
        public async Task ASelectionSkipsACachedRowNoShardOwnsAndAnswersWithTheRest()
        {
            var (database, cache, _, _) = OwnedNothingOfNote();
            await database.PutAsync(One, Note("owned"));
            await cache.UpsertAsync(new MeshQuickstartNote { NoteId = "u", Body = "unowned" }, new CultRecordHandle<MeshQuickstartNote>(Unowned));
            using var server = new Server(cache, ServerSecurityOptions.Development());
            using var databaseServer = new CultNetDatabaseServer(server, database);

            var response = databaseServer.CreateSelectionResponse(new CultNetSnapshotRequestV1Message
            {
                MessageId = "select-all",
                Selection = new CultNetSelection { Projection = CultNetSelectionProjections.Document }
            });

            Assert.That(response.Documents!.Select(document => document.RecordKey), Is.EqualTo(new[] { One.Value }));
            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<MeshQuickstartNote>(Unowned));
            await cache.UpsertAsync(new JournalIndexedNote { Name = "n", Tag = "t" }, new CultRecordHandle<JournalIndexedNote>(new CultRecordKey("unrelated:indexed")));

            Assert.That(database.GetAll<MeshQuickstartNote>(), Is.Empty);
            Assert.That(database.GetAll<object>().Count(), Is.EqualTo(1), "only the owned row");
            Assert.That(database.GetAll<NetworkSchemaNote>().Count(), Is.EqualTo(1));
            Assert.That(await database.GetAsync<MeshQuickstartNote>(Unowned), Is.Null);
            Assert.That(await database.GetAsync<NetworkSchemaNote>(One), Is.Not.Null);
            Assert.That(database.GetByName<MeshQuickstartNote>("u"), Is.Null);
            Assert.That(database.GetByName<JournalIndexedNote>("n"), Is.Null);
            Assert.That(database.GetByIndex<JournalIndexedNote>("tag", "t"), Is.Null);
            Assert.That(cache.Get<MeshQuickstartNote>(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<MeshQuickstartNote>(Unowned));

            var collection = GameCult.Mesh.CultMesh.Collection<MeshQuickstartNote>(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<MeshQuickstartNote>(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<MeshQuickstartNote>(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<CultNetSnapshotResponseRawMessage>(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<MeshQuickstartNote>(Unowned));
            var put = replica.Documents.CreateRawDocumentPutMessage("wire", new CultRecordHandle<MeshQuickstartNote>(new CultRecordKey("unrelated:put")), new MeshQuickstartNote { NoteId = "p", Body = "replicated" });
            put.ShardId = ShardId;
            put.ShardEpoch = 1;
            var owned = replica.Documents.CreateRawDocumentPutMessage("wire", new CultRecordHandle<NetworkSchemaNote>(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<MeshQuickstartNote>().SchemaId,
                            RecordKey = Unowned.Value,
                            ShardId = ShardId,
                            ShardEpoch = 1
                        }
                    },
                    new CultNetShardLogEntryMessage { Sequence = 3, CommittedAt = now, ChangeKind = "added", Put = owned }
                ]
            });

            Assert.That(cache.Get<MeshQuickstartNote>(new CultRecordKey("unrelated:put")), Is.Null, "the put was left out");
            Assert.That(cache.Get<MeshQuickstartNote>(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<CultNetDocumentPutRawMessage, Task> Put, Func<CultNetDocumentDeleteMessage, Task> 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<CultNetDocumentPutRawMessage, CultNetServerPeer, Task>)typeof(CultNetDatabaseServer).GetField("_putHandler", flags)!.GetValue(databaseServer)!;
            var delete = (Func<CultNetDocumentDeleteMessage, CultNetServerPeer, Task>)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<NetworkSchemaNote>(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);
        }

        // A change the log holds is published, however Dispose interleaves with the write: Dispose waits for a journal in
        // flight (disposing the journal first) and publishes whatever the writer had logged and not yet published.
        [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<string> Errors { get; } = new();
            public List<string> Warnings { get; } = new();
            public void LogInfo(string message) { }
            public void LogWarning(string message) => Warnings.Add(message);
            public void LogError(string message) => Errors.Add(message);
            public void LogDebug(string message) { }
        }

        private static async Task ApplyAuthoritativeAsync(CultNetDatabase database, CultRecordKey key, string text)
        {
            var shard = database.Shards[0];
            var message = database.Documents.CreateRawDocumentPutMessage("authoritative", new CultRecordHandle<NetworkSchemaNote>(key), Note(text));
            message.ShardId = shard.ShardId;
            message.ShardEpoch = shard.Epoch;
            await database.ApplyPutAsync(message);
        }

        private static CultNetDatabaseOptions ClientOptions() => new()
        {
            RuntimeId = "local",
            ClientAuthorityScopes = [new CultNetClientAuthorityScope("local")]
        };

        [Test]
        public async Task APredictionOverAPredictionIsStillAPredictionAndTheAuthoritativeWriteReconcilesOnce()
        {
            var cache = new CultCache();
            var database = new CultNetDatabase(cache, ClientOptions());
            var kinds = new List<CultNetDatabaseChangeKind>();
            database.Watch<NetworkSchemaNote>().Subscribe(change => kinds.Add(change.Kind));

            await database.PutPredictedAsync(One, Note("p1"));
            await database.PutPredictedAsync(One, Note("p2"));
            await ApplyAuthoritativeAsync(database, One, "authoritative");
            await ApplyAuthoritativeAsync(database, One, "again");

            Assert.That(kinds, Is.EqualTo(new[]
            {
                CultNetDatabaseChangeKind.Predicted,
                CultNetDatabaseChangeKind.Predicted,
                CultNetDatabaseChangeKind.Reconciled,
                CultNetDatabaseChangeKind.Updated
            }));
        }

        [Test]
        public async Task AGameSessionPredictionOfTheInstanceTheCacheAlreadyHoldsIsAPredictionThatIsNeverLogged()
        {
            var rootPath = Path.Combine(TestContext.CurrentContext.WorkDirectory, "journal-session", Guid.NewGuid().ToString("N"));
            Directory.CreateDirectory(rootPath);
            using var node = await GameCult.Mesh.CultMesh.CreateNodeAsync(Path.Combine(rootPath, "world.ccmp"), new GameCult.Mesh.CultMeshNodeOptions
            {
                StartServer = false,
                DatabaseOptions = ClientOptions()
            });
            using var session = GameCult.Mesh.CultMesh.CreateGameSession(node, new GameCult.Mesh.CultMeshGameSessionOptions
            {
                ServeSimulationObservations = false,
                ServeVerseDiscovery = false,
                ServePeerExchange = false
            });
            var kinds = new List<CultNetDatabaseChangeKind>();
            node.Database.Watch<NetworkSchemaNote>().Subscribe(change => kinds.Add(change.Kind));

            await session.PredictAsync(One, Note("p1"));
            var held = node.Cache.Get<NetworkSchemaNote>(One)!;
            held.Text = "p2";
            await session.PredictAsync(One, held);
            await ApplyAuthoritativeAsync(node.Database, One, "authoritative");

            Assert.That(kinds, Is.EqualTo(new[] { CultNetDatabaseChangeKind.Predicted, CultNetDatabaseChangeKind.Predicted, CultNetDatabaseChangeKind.Reconciled }));
            Assert.That(node.Database.GetMutationLog(node.Database.Shards[0].ShardId).Select(entry => entry.Kind), Is.EqualTo(new[] { CultNetDatabaseChangeKind.Updated }));
        }

        [Test]
        public async Task ADatabaseRecordsEveryDeleteEntryItAppliesWhileAnotherThreadRemovesTheSameKeys()
        {
            const int rounds = 20000;
            var replicaCache = new CultCache();
            var replica = Database(replicaCache, primary: false, store: null);
            var schemaId = SchemaId(replicaCache);
            for (var i = 1; i <= rounds; i++)
                await replicaCache.UpsertAsync(Note("n"), new CultRecordHandle<NetworkSchemaNote>(new CultRecordKey("race:" + i)));

            var current = 0;
            var stop = false;
            var remover = Task.Run(() =>
            {
                var next = 1;
                while (next <= rounds && !Volatile.Read(ref stop))
                {
                    if (Volatile.Read(ref current) >= next)
                    {
                        replicaCache.Remove(new CultRecordKey("race:" + next));
                        next++;
                    }
                }
            });
            for (var i = 1; i <= rounds; i++)
            {
                Volatile.Write(ref current, i);
                await replica.ApplyShardLogResponseAsync(new CultNetShardLogResponseMessage
                {
                    ShardId = ShardId,
                    ShardEpoch = 1,
                    Entries =
                    [
                        new CultNetShardLogEntryMessage
                        {
                            Sequence = i,
                            CommittedAt = DateTimeOffset.UtcNow.ToString("O"),
                            ChangeKind = "removed",
                            Delete = new CultNetDocumentDeleteMessage
                            {
                                MessageId = "d" + i,
                                SchemaId = schemaId,
                                RecordKey = "race:" + i,
                                ShardId = ShardId,
                                ShardEpoch = 1
                            }
                        }
                    ]
                });
            }

            Volatile.Write(ref stop, true);
            await remover;

            Assert.That(replica.GetMutationLog(ShardId).Select(entry => entry.Sequence), Is.EqualTo(Enumerable.Range(1, rounds).Select(i => (long)i)),
                "a replica's log holds every entry of the primary's, whoever removed the key first");
        }
    }
}