From fe7bca0cb5b30ba1fff43077d694b98c3dceb429 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 29 Sep 2026 23:48:50 +0200 Subject: [PATCH] CultNetDatabase: log on the cache journal and publish from Watch, no queue Cut 4b-4d of the publication map. The reorder buffer, drainer, prediction gate and the logging inside the Watch handler are gone. The database registers the Watch observer, then one cache journal. The journal runs under the gate in Sequence order: it takes the door, decides the change kind and prediction bookkeeping, logs (a primary mints, a replica records the primary's entry) and stashes the change; the Watch observer publishes the stash entry by Sequence. A primary that cannot log a committed change burns that sequence, raises its own compaction floor, tries CompactThrough, and throws CultNetShardLogException (not an InvalidOperationException) after publication. A replica's refused append reaches the pull loop and the cursor does not advance. Fixes four defects the map named: the log is appended to disk before memory (and memory is a SortedList keyed by sequence); the snapshot reads its log sequence before its documents; a restart resumes above the compaction point; GetCompactedMutationLogSequence and GetLatestMutationLogSequence honour the in-memory floor. The snapshot removal door sets Removal. Co-Authored-By: Claude Sonnet 5.5 --- .../Contracts/cultcache-store-composition.md | 25 +- src/GameCult.Networking/CultNetDatabase.cs | 491 ++++++++++-------- .../CultNetShardMutationLogStorage.cs | 5 + .../CultNetJournalTests.cs | 456 ++++++++++++++++ 4 files changed, 751 insertions(+), 226 deletions(-) create mode 100644 tests/GameCult.Networking.Tests/CultNetJournalTests.cs diff --git a/src/GameCult.Caching/Contracts/cultcache-store-composition.md b/src/GameCult.Caching/Contracts/cultcache-store-composition.md index 931088eb..a19a8839 100644 --- a/src/GameCult.Caching/Contracts/cultcache-store-composition.md +++ b/src/GameCult.Caching/Contracts/cultcache-store-composition.md @@ -47,14 +47,19 @@ which route wrote it. - Changes published through `Watch`/`WatchRecord` carry a `Sequence`: a per-cache, in-memory number assigned when the cache admits the change, under its gate. It increases in admission order and is not persisted. - `OnUpdate(previous, current)` carries none. CultNet derives its - streams (`CultNetDatabase.WatchAllChanges`, the subscription server, - database-backed Mesh handles) from `Watch`, so a load, a commit and a variant - dependent's re-resolution all reach subscribers and the mutation log through - one handler over `Watch`; a database write door only hands that handler the - wire message or replicated entry for the change it admitted. That handler releases - changes in cache `Sequence` order, so the log and the stream agree with the - cache. A replica publishes but never mints log sequences. The + `OnUpdate(previous, current)` carries none. CultNet derives its streams + (`CultNetDatabase.WatchAllChanges`, the subscription server, database-backed + Mesh handles) from `Watch` plus one journal (`AddJournal`), so a load, a + commit and a variant dependent's re-resolution all reach subscribers and the + mutation log. The journal runs under the gate in `Sequence` order, so the + shard log is written in cache order; the `Watch` observer publishes what the + journal decided. A database write door only hands the journal the wire message + or replicated entry for the change it admitted. A replica publishes but never + mints log sequences. A primary logs every change it admits, loads and pulls + included; a store attached after the database exists has its hydration logged + too. A primary that cannot log a change compacts through that change's + sequence and throws `CultNetShardLogException` to the writer after publication: + the commit stands, and a replica behind it resynchronizes from a snapshot. The streams carry no `Sequence` and have no stale protection. - `GetWithSequence(key)` returns the document and the cache's current `Sequence`, read together under the gate: every change with a `Sequence` at @@ -131,6 +136,10 @@ which route wrote it. needs the cache or its gate: another thread's write, a lock a writer holds. Its own leaf locks, taken only inside the journal, are the only locks that may nest under the gate. +- A shard mutation log store (`ICultNetShardMutationLogStore`) runs inside the + CultNet journal, under the cache's gate: it must not call the cache or wait + for a thread that does. A store that writes its own cache throws through the + journal's reentry guard, and the admission stands. - A thread holds one cache's gate at a time: entering a cache's hold inside another cache's hold throws. - A store doing I/O on behalf of its cache (pull, flush, commit) holds the gate, diff --git a/src/GameCult.Networking/CultNetDatabase.cs b/src/GameCult.Networking/CultNetDatabase.cs index 359049f1..202f88b5 100644 --- a/src/GameCult.Networking/CultNetDatabase.cs +++ b/src/GameCult.Networking/CultNetDatabase.cs @@ -463,6 +463,36 @@ namespace GameCult.Networking public string Reason { get; } } + /// + /// Raised to the writer when a shard primary committed a change and could not log it. The commit stands and the + /// change was published; the shard's log burned and compacted past it, so every replica + /// behind it resynchronizes from a snapshot. It is deliberately not an : + /// callers that read that type as a refused write must not mistake a committed one for it. When several things + /// fail in one admission, the writer sees an that contains this one. + /// + public sealed class CultNetShardLogException : Exception + { + /// + /// Creates a shard log exception. + /// + public CultNetShardLogException(string shardId, long sequence, Exception cause) + : base($"Shard '{shardId}' committed a change but could not log it; sequence {sequence} is burned and compacted past.", cause) + { + ShardId = shardId; + Sequence = sequence; + } + + /// + /// Gets the shard whose log refused the change. + /// + public string ShardId { get; } + + /// + /// Gets the shard-log sequence that was minted for the change and burned. + /// + public long Sequence { get; } + } + /// /// Database-style CultNet facade over a CultCache shard set. /// @@ -474,38 +504,39 @@ namespace GameCult.Networking private readonly List _shards; private readonly List _clientAuthorityScopes; private readonly ICultNetShardMutationLogStore? _mutationLogStore; - private readonly HashSet _predictedDocuments = new(StringComparer.Ordinal); - private readonly Dictionary> _mutationLogs = + // Keyed by sequence, like the durable store: recording an entry twice replaces it. + private readonly Dictionary> _mutationLogs = new(StringComparer.Ordinal); private readonly Dictionary _nextLogSequences = new(StringComparer.Ordinal); + // The highest sequence this primary burned per shard (a change it committed and could not log). Serves as the + // shard's compaction floor beside the store's own, so history the log cannot serve is answered compacted_history. + private readonly Dictionary _burnedThrough = new(StringComparer.Ordinal); private readonly Dictionary _appliedShardSequences = new(StringComparer.Ordinal); // CultNet typed selection, section 2: "ordinal is the sequence of the commit that last wrote - // the row". Kept on every append (AppendMutationLogEntry is the one site), keyed by (schemaId, + // the row". Kept on every append (LogPrimaryChange and RecordReplicatedEntry are the sites), keyed by (schemaId, // recordKey) so the evaluator's order and cursor never read a clock. private readonly Dictionary<(string SchemaId, string RecordKey), long> _lastWriteSequence = new(); private readonly Subject _changes = 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 - // message it carries, a prediction it reconciles, the replicated log entry it applies. The cache's Watch stream - // is the only publisher and the only logger; a door registers this context under the identity of the change it - // admits (the instance it writes, or the instance it removes) and the stream handler consumes it once, for that - // change alone. A write to the same key by anyone else is a different instance and gets no context. + // message it carries, a prediction it reconciles, the replicated log entry it applies. The journal is the only + // logger and consumes the context under the cache gate; a door registers it under the identity of the change it + // admits (the instance it writes, or the instance it removes), and the journal takes it once, for that change + // alone. A write to the same key by anyone else is a different instance and gets no context. private readonly ConcurrentDictionary _doors = new(ReferenceIdentity.Instance); - // One gate over the mutation log, its sequence counters and the last-write map. Publication stays outside it: a - // subscriber may write the cache from its handler. + // One gate over the mutation log, its sequence counters, the last-write map and the publication stash. Lock order: + // the cache gate, then this, then the log store's own lock. Nothing holding this calls the cache, and a log store + // must not either (ICultNetShardMutationLogStore). Publication stays outside it: a subscriber may write the cache + // from its handler. private readonly object _logGate = new(); - // The cache admits a change to a key under its gate, but each thread reaches this handler after leaving it, in any - // order. Changes are therefore released strictly in cache Sequence order: a change ahead of a gap is parked until the - // gap fills, and one thread at a time drains the ready queue, outside every lock, so a subscriber that writes the - // cache from its handler queues behind the change it is handling instead of running inside it. - private readonly SortedDictionary _parked = new(); - private readonly Queue _ready = new(); - private long _nextSequence; - private bool _draining; - // Predictions this runtime made and has not seen reconciled. Every read and write takes _predictionGate. - private readonly object _predictionGate = new(); - private bool _disposed; + // Changes the journal has decided and the Watch observer has not yet published, by cache Sequence. The journal + // runs under the cache gate before any observer, so each change is stashed before it can be published. + private readonly Dictionary _stash = new(); + // Predictions this runtime made and has not seen reconciled. Only the journal reads or writes it, under the cache gate. + private readonly HashSet _predictedDocuments = new(StringComparer.Ordinal); + private volatile bool _disposed; /// /// Creates a database surface over a CultCache instance. @@ -524,12 +555,9 @@ namespace GameCult.Networking _clientAuthorityScopes = (options.ClientAuthorityScopes ?? Array.Empty()).ToList(); CursorKey = options.CursorKey ?? CultNetSelectionCursorKey.Random(); InitializeLogSequencesFromStore(); - lock (_logGate) - { - _cacheChanges = _cache.Watch().Subscribe(OnCacheChange); - // Changes admitted before this read are released as they arrive; the next one is the first this handler waits for. - _nextSequence = _cache.GetWithSequence(new CultRecordKey(string.Empty)).Sequence + 1; - } + // The observer first, the journal second: a change the journal stashed always finds its publisher. + _cacheChanges = _cache.Watch().Subscribe(Publish); + _cacheJournal = _cache.AddJournal(Journal); } /// @@ -591,7 +619,7 @@ namespace GameCult.Networking return Array.Empty(); } - var query = entries.Where(entry => entry.Sequence > afterSequence); + var query = entries.Values.Where(entry => entry.Sequence > afterSequence); if (limit.HasValue) { query = query.Take(limit.Value); @@ -628,7 +656,15 @@ namespace GameCult.Networking { ThrowIfDisposed(); if (string.IsNullOrWhiteSpace(shardId)) throw new ArgumentException("Value must be non-empty.", nameof(shardId)); - return _mutationLogStore?.GetCompactedThrough(shardId) ?? 0; + lock (_logGate) + return CompactedFloor(shardId); + } + + // The store's compaction point or this primary's own burned sequences, whichever is higher. Under _logGate. + private long CompactedFloor(string shardId) + { + var burned = _burnedThrough.TryGetValue(shardId, out var through) ? through : 0; + return Math.Max(burned, _mutationLogStore?.GetCompactedThrough(shardId) ?? 0); } /// @@ -638,11 +674,13 @@ namespace GameCult.Networking { ThrowIfDisposed(); if (string.IsNullOrWhiteSpace(shardId)) throw new ArgumentException("Value must be non-empty.", nameof(shardId)); - var compactedThrough = GetCompactedMutationLogSequence(shardId); - var retained = _mutationLogStore != null - ? _mutationLogStore.Read(shardId).Select(entry => entry.Sequence) - : GetMutationLog(shardId).Select(entry => entry.Sequence); - return retained.DefaultIfEmpty(compactedThrough).Max(); + lock (_logGate) + { + var retained = _mutationLogStore != null + ? _mutationLogStore.Read(shardId).Select(entry => entry.Sequence) + : GetMutationLog(shardId).Select(entry => entry.Sequence); + return retained.Append(CompactedFloor(shardId)).Max(); + } } /// @@ -659,6 +697,10 @@ namespace GameCult.Networking ThrowIfDisposed(); if (shard == null) throw new ArgumentNullException(nameof(shard)); + // The log sequence is read before any document: a write that lands in between is then replayed from the log on + // top of documents that already hold it (whole-document puts and deletes, applied in order), and a snapshot can + // never claim a sequence whose write its documents lack. + var logSequence = GetLatestMutationLogSequence(shard.ShardId); var lowSchemas = CultNetV0SelectionLowering.Lower(filter?.SchemaIds); var lowKeys = CultNetV0SelectionLowering.Lower(filter?.RecordKeys); @@ -672,7 +714,7 @@ namespace GameCult.Networking Documents = Array.Empty(), ShardId = shard.ShardId, ShardEpoch = shard.Epoch, - ShardLogSequence = GetLatestMutationLogSequence(shard.ShardId) + ShardLogSequence = logSequence }; } @@ -695,7 +737,7 @@ namespace GameCult.Networking Documents = page.Documents ?? Array.Empty(), ShardId = shard.ShardId, ShardEpoch = shard.Epoch, - ShardLogSequence = GetLatestMutationLogSequence(shard.ShardId) + ShardLogSequence = logSequence }; } @@ -730,7 +772,7 @@ namespace GameCult.Networking var local = GetLocalShardDocuments(shard).ToArray(); foreach (var existing in local.Where(item => !incomingKeys.Contains(item.Key.Value))) { - using (Door(existing.Document, new DoorContext(shard) { NoLog = true })) + using (Door(existing.Document, new DoorContext(shard) { NoLog = true, Removal = true })) _cache.Remove(existing.Key); } @@ -912,14 +954,7 @@ namespace GameCult.Networking var shard = ResolveShardInternal(descriptor, key); EnsurePrimary(shard, descriptor.SchemaId, key, message.ShardEpoch); var document = _documents.DeserializeRawDocument(message.Document); - var predictionKey = PredictionKey(descriptor.SchemaId, key); - var reconcilesPrediction = HasPrediction(predictionKey); - using (Door(document, new DoorContext(shard) - { - Put = message, - Kind = reconcilesPrediction ? CultNetDatabaseChangeKind.Reconciled : null, - PredictionKey = reconcilesPrediction ? predictionKey : null - })) + using (Door(document, new DoorContext(shard) { Put = message, PredictionKey = PredictionKey(descriptor.SchemaId, key) })) await _cache.UpsertAsync(descriptor.DocumentType, document, key).ConfigureAwait(false); return document; } @@ -1057,7 +1092,6 @@ namespace GameCult.Networking (change.Document != null && ReferenceEquals(_cache.GetByIndex(alias, value), change.Document)) || (change.PreviousDocument != null && ReferenceEquals(_cache.GetByIndex(alias, value), change.PreviousDocument))); } - /// /// Releases database subscriptions. /// @@ -1070,146 +1104,182 @@ namespace GameCult.Networking _disposed = true; _cacheChanges.Dispose(); + _cacheJournal.Dispose(); + lock (_logGate) + _stash.Clear(); _changes.Dispose(); } - // The cache is the one owner of "a change was committed", and this is its one publisher and one logger: every admitted - // change reaches subscribers and, when this runtime is the shard's primary, the mutation log, exactly once, in cache - // Sequence order. A replica publishes what it admits but never mints log sequences: its log holds only the primary's - // own entries, which the replica door hands over. - private void OnCacheChange(CultCacheDocumentChange change) + // The cache is the one owner of "a change was committed"; the journal is this database's one decision point and its + // one logger. It runs under the cache gate, in cache Sequence order, before any observer: it takes the change's door, + // settles the change kind and the prediction bookkeeping, logs the change when this runtime is the shard's primary + // (a replica records only the primary's own entry, which its door carries), and stashes what the publisher needs. + // A change with no door was written by someone else (a bare cache write, a load, a pull) and is logged like any + // unattributed commit. It never enters the cache and calls nothing that does. + private void Journal(IReadOnlyList> changes) { - var pending = new PendingChange(change, TakeDoor(change)); - lock (_logGate) - { - if (_disposed) - { - return; - } - - if (change.Sequence < _nextSequence) - { - _ready.Enqueue(pending); - } - else - { - _parked[change.Sequence] = pending; - while (_parked.Remove(_nextSequence, out var next)) - { - _ready.Enqueue(next); - _nextSequence++; - } - } - - if (_draining) - { - return; - } - - _draining = true; - } - - try - { - while (true) - { - PendingChange? item; - lock (_logGate) - { - if (_disposed || !_ready.TryDequeue(out item)) - { - _draining = false; - return; - } - } - - Deliver(item); - } - } - catch - { - lock (_logGate) - _draining = false; - throw; - } - } - - private void Deliver(PendingChange item) - { - var change = item.Change; - var door = item.Door; - var identity = change.Document ?? change.PreviousDocument; - if (identity == null) + if (_disposed) { return; } - var documentType = identity.GetType(); - var descriptor = _cache.Registry.GetRequired(documentType); - var shard = door?.Shard ?? ResolveShardInternal(descriptor, change.Key); - var kind = door?.Kind ?? change.Kind switch + List? failures = null; + foreach (var change in changes) { - CultCacheDocumentChangeKind.Removed => CultNetDatabaseChangeKind.Removed, - CultCacheDocumentChangeKind.Added => CultNetDatabaseChangeKind.Added, - _ => CultNetDatabaseChangeKind.Updated - }; - var logKind = kind == CultNetDatabaseChangeKind.Reconciled ? CultNetDatabaseChangeKind.Updated : kind; - if (door?.PredictionKey is { } predictionKey) - { - lock (_predictionGate) + var identity = change.Document ?? change.PreviousDocument; + if (identity == null) + { + continue; + } + + var door = TakeDoor(change); + var documentType = identity.GetType(); + var descriptor = _cache.Registry.GetRequired(documentType); + var shard = door?.Shard ?? ResolveShardInternal(descriptor, change.Key); + var kind = door?.Kind ?? change.Kind switch + { + CultCacheDocumentChangeKind.Removed => CultNetDatabaseChangeKind.Removed, + CultCacheDocumentChangeKind.Added => CultNetDatabaseChangeKind.Added, + _ => CultNetDatabaseChangeKind.Updated + }; + if (door?.PredictionKey is { } predictionKey) { if (door.Predicts) _predictedDocuments.Add(predictionKey); - else - _predictedDocuments.Remove(predictionKey); + else if (_predictedDocuments.Remove(predictionKey)) + kind = CultNetDatabaseChangeKind.Reconciled; } - } - if (door?.Replica is { } replicated) - { - lock (_logGate) + // The log records a reconciled change as the update it is. + var logKind = kind == CultNetDatabaseChangeKind.Reconciled ? CultNetDatabaseChangeKind.Updated : kind; + try { - RecordMutationLogEntry(new CultNetShardMutationLogEntry( - shard.ShardId, - shard.Epoch, - replicated.Sequence, - replicated.CommittedAt, - logKind, - descriptor.SchemaId, - change.Key, - change.Document, - change.PreviousDocument), - replicated); - _lastWriteSequence[(descriptor.SchemaId, change.Key.Value)] = replicated.Sequence; + if (door?.Replica is { } replicated) + { + RecordReplicatedEntry(shard, logKind, descriptor.SchemaId, change, replicated); + } + else if (door?.NoLog != true && shard.IsPrimary) + { + LogPrimaryChange(shard, logKind, descriptor.SchemaId, change, door); + } } - } - else if (door?.NoLog != true && shard.IsPrimary) - { - AppendMutationLogEntry( - shard, - logKind, - change.Key, - descriptor.SchemaId, - change.Document, - change.PreviousDocument, - door?.Put != null - ? new CultNetShardLogEntryMessage - { - ChangeKind = change.Kind == CultCacheDocumentChangeKind.Added ? "added" : "updated", - Put = door.Put - } - : door?.Delete != null - ? new CultNetShardLogEntryMessage { ChangeKind = "removed", Delete = door.Delete } - : null); + catch (Exception exception) + { + (failures ??= new List()).Add(exception); + } + + lock (_logGate) + _stash[change.Sequence] = new Publication(documentType, kind, change.Key, descriptor.SchemaId, shard, change.Document, change.PreviousDocument); } - PublishUntyped(documentType, kind, change.Key, descriptor.SchemaId, shard, change.Document, change.PreviousDocument); + // Every change is stashed and will be published; the writer hears of a failed log afterwards. + if (failures != null) + { + throw failures.Count == 1 ? failures[0] : new AggregateException(failures); + } } - private bool HasPrediction(string predictionKey) + // The Watch observer publishes, on the writer's thread after the cache gate opens and before the write returns, what + // the journal stashed for this change. A change with no stash was admitted before this database existed. + private void Publish(CultCacheDocumentChange change) { - lock (_predictionGate) - return _predictedDocuments.Contains(predictionKey); + Publication? publication; + lock (_logGate) + { + if (!_stash.Remove(change.Sequence, out publication)) + { + return; + } + } + + PublishUntyped( + publication.DocumentType, + publication.Kind, + publication.Key, + publication.SchemaId, + publication.Shard, + publication.Document, + publication.PreviousDocument); + } + + private void RecordReplicatedEntry( + CultNetShardDescriptor shard, + CultNetDatabaseChangeKind logKind, + string schemaId, + CultCacheDocumentChange change, + CultNetShardLogEntryMessage replicated) + { + lock (_logGate) + { + RecordMutationLogEntry(new CultNetShardMutationLogEntry( + shard.ShardId, + shard.Epoch, + replicated.Sequence, + replicated.CommittedAt, + logKind, + schemaId, + change.Key, + change.Document, + change.PreviousDocument), + replicated); + _lastWriteSequence[(schemaId, change.Key.Value)] = replicated.Sequence; + } + } + + // A primary mints the sequence, appends durably, then records in memory. If it cannot log the change, the change is + // still committed: the sequence is burned and compacted past, so a replica behind it is told compacted_history and + // resynchronizes by snapshot, and the writer is told with a CultNetShardLogException. + private void LogPrimaryChange( + CultNetShardDescriptor shard, + CultNetDatabaseChangeKind logKind, + string schemaId, + CultCacheDocumentChange change, + DoorContext? door) + { + lock (_logGate) + { + var sequence = NextMutationLogSequence(shard.ShardId); + _lastWriteSequence[(schemaId, change.Key.Value)] = sequence; + try + { + var entry = new CultNetShardMutationLogEntry( + shard.ShardId, + shard.Epoch, + sequence, + DateTimeOffset.UtcNow.ToString("O"), + logKind, + schemaId, + change.Key, + change.Document, + change.PreviousDocument); + var wireEntry = _mutationLogStore == null + ? null + : door?.Put != null + ? new CultNetShardLogEntryMessage + { + ChangeKind = change.Kind == CultCacheDocumentChangeKind.Added ? "added" : "updated", + Put = door.Put + } + : door?.Delete != null + ? new CultNetShardLogEntryMessage { ChangeKind = "removed", Delete = door.Delete } + : ToLogEntryMessage(entry); + RecordMutationLogEntry(entry, wireEntry); + } + catch (Exception cause) + { + _burnedThrough[shard.ShardId] = Math.Max(_burnedThrough.TryGetValue(shard.ShardId, out var burned) ? burned : 0, sequence); + try + { + _mutationLogStore?.CompactThrough(shard.ShardId, sequence); + } + catch (Exception compactCause) + { + cause = new AggregateException(cause, compactCause); + } + + throw new CultNetShardLogException(shard.ShardId, sequence, cause); + } + } } // A door's context belongs to one specific change, recognised by shape: a put door wrote an instance no cache had @@ -1252,7 +1322,7 @@ namespace GameCult.Networking _owned = owned; } - // True when the stream handler took the context: the door's write was admitted. + // True when the journal took the context: the door's write was admitted. public bool Consumed => !_owned || !_doors.TryGetValue(_identity, out var held) || !ReferenceEquals(held, _context); public void Dispose() @@ -1262,16 +1332,34 @@ namespace GameCult.Networking } } - private sealed class PendingChange + // What the publisher needs of a change the journal decided. + private sealed class Publication { - public PendingChange(CultCacheDocumentChange change, DoorContext? door) + public Publication( + Type documentType, + CultNetDatabaseChangeKind kind, + CultRecordKey key, + string schemaId, + CultNetShardDescriptor shard, + object? document, + object? previousDocument) { - Change = change; - Door = door; + DocumentType = documentType; + Kind = kind; + Key = key; + SchemaId = schemaId; + Shard = shard; + Document = document; + PreviousDocument = previousDocument; } - public CultCacheDocumentChange Change { get; } - public DoorContext? Door { get; } + public Type DocumentType { get; } + public CultNetDatabaseChangeKind Kind { get; } + public CultRecordKey Key { get; } + public string SchemaId { get; } + public CultNetShardDescriptor Shard { get; } + public object? Document { get; } + public object? PreviousDocument { get; } } private sealed class DoorContext @@ -1365,36 +1453,6 @@ namespace GameCult.Networking public object Document { get; } } - private void AppendMutationLogEntry( - CultNetShardDescriptor shard, - CultNetDatabaseChangeKind kind, - CultRecordKey key, - string schemaId, - object? document, - object? previousDocument, - CultNetShardLogEntryMessage? wireEntry = null) - { - lock (_logGate) - { - var sequence = NextMutationLogSequence(shard.ShardId); - var committedAt = DateTimeOffset.UtcNow.ToString("O"); - var entry = new CultNetShardMutationLogEntry( - shard.ShardId, - shard.Epoch, - sequence, - committedAt, - kind, - schemaId, - key, - document, - previousDocument); - var storedWireEntry = _mutationLogStore == null - ? null - : wireEntry ?? ToLogEntryMessage(entry); - RecordMutationLogEntry(entry, storedWireEntry); - _lastWriteSequence[(schemaId, key.Value)] = sequence; - } - } /// /// The shard-log sequence of the commit that last wrote this row, or null when the row has @@ -1444,21 +1502,22 @@ namespace GameCult.Networking { if (!_mutationLogs.TryGetValue(entry.ShardId, out var entries)) { - entries = new List(); + entries = new SortedList(); _mutationLogs[entry.ShardId] = entries; } - entries.Add(entry); + // Disk first: memory never serves an entry the durable log lacks, and a refused append records nothing. + if (_mutationLogStore != null && wireEntry != null) + { + _mutationLogStore.Append(entry.ShardId, NormalizeWireLogEntry(entry, wireEntry)); + } + + entries[entry.Sequence] = entry; if (!_nextLogSequences.TryGetValue(entry.ShardId, out var next) || next <= entry.Sequence) { _nextLogSequences[entry.ShardId] = entry.Sequence + 1; } - - if (_mutationLogStore != null && wireEntry != null) - { - _mutationLogStore.Append(entry.ShardId, NormalizeWireLogEntry(entry, wireEntry)); - } } private long NextMutationLogSequence(string shardId) @@ -1481,7 +1540,7 @@ namespace GameCult.Networking { // R-Q: the log is the durable record of every commit, replica-applied ones included, so // the last-write sequence a restart resumes with is rebuilt from it exactly as - // AppendMutationLogEntry/the replica apply paths maintain it live - a row's ordinal for + // LogPrimaryChange/the replica apply paths maintain it live - a row's ordinal for // the selection evaluator's order and cursor must not reset to "never written" on restart. var entries = _mutationLogStore.Read(shard.ShardId); var highest = 0L; @@ -1499,6 +1558,8 @@ namespace GameCult.Networking } } + // Sequences at or below the compaction point are spent even when the log holds none of them. + highest = Math.Max(highest, _mutationLogStore.GetCompactedThrough(shard.ShardId)); if (highest > 0) { _nextLogSequences[shard.ShardId] = highest + 1; @@ -1570,17 +1631,11 @@ namespace GameCult.Networking $"Shard '{shard.ShardId}' log entry {entry.Sequence} has a removed kind with a put payload."); } - var predictionKey = PredictionKey(descriptor.SchemaId, key); - var reconcilesPrediction = HasPrediction(predictionKey); - if (reconcilesPrediction) - { - kind = CultNetDatabaseChangeKind.Reconciled; - } - // The replica's log records the primary's own entry under the primary's own sequence, in the same hold that - // admits the change; the stream handler does both. + // admits the change; the journal does both. A retry after a refused append is a fresh instance, so it matches + // its door again and re-records the entry. var document = _documents.DeserializeRawDocument(message.Document); - using (Door(document, new DoorContext(shard) { Kind = kind, Replica = entry, PredictionKey = reconcilesPrediction ? predictionKey : null })) + using (Door(document, new DoorContext(shard) { Kind = kind, Replica = entry, PredictionKey = PredictionKey(descriptor.SchemaId, key) })) await _cache.UpsertAsync(descriptor.DocumentType, document, key).ConfigureAwait(false); } diff --git a/src/GameCult.Networking/CultNetShardMutationLogStorage.cs b/src/GameCult.Networking/CultNetShardMutationLogStorage.cs index 5b3a9c32..6325a0a0 100644 --- a/src/GameCult.Networking/CultNetShardMutationLogStorage.cs +++ b/src/GameCult.Networking/CultNetShardMutationLogStorage.cs @@ -10,6 +10,11 @@ namespace GameCult.Networking /// /// Stores accepted shard-log entries in their replica catch-up wire shape. /// + /// + /// A database calls its store from a cache journal, under that cache's gate, so a slow store slows every writer of + /// the cache. A store must not call the cache, or wait for a thread that does. Appends must throw rather than + /// swallow a failure: the database burns the sequence and compacts past it, and tells the writer. + /// public interface ICultNetShardMutationLogStore { /// diff --git a/tests/GameCult.Networking.Tests/CultNetJournalTests.cs b/tests/GameCult.Networking.Tests/CultNetJournalTests.cs new file mode 100644 index 00000000..c0251af5 --- /dev/null +++ b/tests/GameCult.Networking.Tests/CultNetJournalTests.cs @@ -0,0 +1,456 @@ +#nullable enable +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; +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().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 Record(CultNetDatabase database) + { + var seen = new List(); + 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 _entries = new(); + private long _compactedThrough; + + public Func RefuseAppend { get; set; } = _ => false; + public bool RefuseCompact { get; set; } + public Action? OnRead { get; set; } + public Action? OnAppend { get; set; } + + public IReadOnlyList 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 FetchAsync(CultNetShardDescriptor shard, long afterSequence, int? limit = null) => + Task.FromResult(Log(afterSequence)); + + public Task 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(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(async () => await primary.PutAsync(Two, Note("two")))!; + + Assert.That(failure.ShardId, Is.EqualTo(ShardId)); + Assert.That(failure.Sequence, Is.EqualTo(2)); + Assert.That(failure, Is.Not.InstanceOf(), "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(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(async () => await primary.PutAsync(Two, Note("two")))!; + + Assert.That(failure.InnerException, Is.InstanceOf()); + 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(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; + replica.Watch().Where(change => change.Key.Equals(One)).Subscribe(_ => + { + if (nested) return; + nested = true; + replica.ApplyShardLogResponseAsync(response).GetAwaiter().GetResult(); + }); + + await replica.ApplyShardLogResponseAsync(response); + + Assert.That(nested, Is.True); + 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)); + } + + [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(Two), Note("two")); + database.Watch().Where(change => change.Key.Equals(One)).Subscribe(_ => + database.ApplyPutAsync(message).GetAwaiter().GetResult()); + var published = Record(database); + + 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 { OnAppend = () => cache.UpsertAsync(Note("from the store"), new CultRecordHandle(Three)).GetAwaiter().GetResult() }; + var database = Database(cache, primary: true, store); + var published = Record(database); + + var failure = Assert.ThrowsAsync(async () => await database.PutAsync(One, Note("one")))!; + + Assert.That(failure.InnerException, Is.InstanceOf(), "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(() => cache.Commit(batch => + { + batch.Upsert(Note("one"), new CultRecordHandle(One)); + batch.Upsert(Note("two"), new CultRecordHandle(Two)); + }))!; + + Assert.That(failure.InnerExceptions.OfType().Select(exception => exception.Sequence), Is.EqualTo(new[] { 1L, 2L })); + 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(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(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(One)); + await writer.UpsertAsync(Note("two"), new CultRecordHandle(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(One)); + var database = Database(cache, primary: true, new FlakyLogStore()); + var put = database.Documents.CreateRawDocumentPutMessage("snapshot-two", new CultRecordHandle(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"); + } + } +}