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"); + } + } +}