From 2ff82365cb41b6cd152a8678c9f2dfa7b50ea86f Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 19 May 2026 19:55:29 +0100 Subject: [PATCH] Add durable shard mutation log storage --- src/GameCult.Mesh/README.md | 7 + src/GameCult.Mesh/docs/public-api.md | 5 +- .../Contracts/cultnet-distributed-database.md | 16 +- src/GameCult.Networking/CultNetDatabase.cs | 178 +++++++++++++++++- .../CultNetDatabaseServer.cs | 38 +--- .../CultNetShardMutationLogStorage.cs | 109 +++++++++++ src/GameCult.Networking/README.md | 5 +- .../NetworkingTests.cs | 61 ++++++ 8 files changed, 367 insertions(+), 52 deletions(-) create mode 100644 src/GameCult.Networking/CultNetShardMutationLogStorage.cs diff --git a/src/GameCult.Mesh/README.md b/src/GameCult.Mesh/README.md index 06a273c7..6d1a725f 100644 --- a/src/GameCult.Mesh/README.md +++ b/src/GameCult.Mesh/README.md @@ -80,6 +80,13 @@ support weight, total observed weight, confidence, and quorum status. Those candidates are not committed world state by themselves. They are the fast mesh opinion layer that can feed an authoritative shard commit path. +## Durable Catch-Up + +CultMesh can persist authoritative shard logs through +`ICultNetShardMutationLogStore`. The included file store writes one MessagePack +log per shard, so a restarted primary can still serve replica catch-up without +depending on warm process memory. + ## Documentation - [Public API](docs/public-api.md) diff --git a/src/GameCult.Mesh/docs/public-api.md b/src/GameCult.Mesh/docs/public-api.md index fe60e985..f1c7f390 100644 --- a/src/GameCult.Mesh/docs/public-api.md +++ b/src/GameCult.Mesh/docs/public-api.md @@ -56,7 +56,10 @@ authoritative log arrives, the database emits `Reconciled`. ### Replica Catch-Up Shard logs can be requested over schema-v0 and applied by replicas. Replica -cursors can be stored with `ICultNetShardReplicaCursorStore`. +cursors can be stored with `ICultNetShardReplicaCursorStore`. Authoritative +shard logs can be persisted with `ICultNetShardMutationLogStore`; the file +implementation stores one MessagePack log per shard so catch-up survives a +primary restart. ### Simulation Witness Consensus diff --git a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md index ae5e1410..b8f254e7 100644 --- a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md +++ b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md @@ -59,6 +59,9 @@ Implemented: - schema-v0 shard log fetcher for `cultnet://host:port` endpoints - restart-safe replica cursor store, including a local MessagePack file-backed implementation +- durable authoritative shard-log storage behind + `ICultNetShardMutationLogStore`, including a per-shard MessagePack file + implementation - client-authority scopes for locally predicted input documents - predicted and reconciled database change kinds for client-side prediction - simulation witness observations for mesh-side opinions about frame facts @@ -73,7 +76,7 @@ Not implemented yet: - membership/failure detection - leader election or automatic shard failover -- durable log persistence and compaction +- log compaction - snapshot fallback for compacted log history - peer-to-peer fanout for observation gossip and candidate propagation - rollback/resimulation helpers for simulation frames @@ -89,6 +92,8 @@ Not implemented yet: applied. - Raw snapshot/put/delete messages must pass through `CultNetDatabase`, not bypass shard policy. +- The accepted shard log is stored in replica catch-up wire form. Object logs + are for local inspection; wire logs are the durable replication authority. - Realtime subscriptions publish domain changes, not storage envelopes. - Client-owned input documents may be predicted locally only inside explicit `CultNetClientAuthorityScope` declarations. @@ -103,11 +108,12 @@ Not implemented yet: ## Next Coherent Slice -Add durable mutation-log storage and snapshot fallback: +Add compaction-aware snapshot fallback and discovery fanout: -- persist per-shard mutation logs instead of keeping them only in memory - define the compaction boundary that turns old log reads into snapshot reads - return `ResyncRequired` when a replica asks for compacted history +- expose operator-facing defaults for durable shard-log paths in CultMesh nodes +- add Verse discovery over the schema-v0 mesh lane - after that, add simulation-frame rollback/resimulation helpers around predicted input streams - add peer-to-peer fanout for observation gossip and candidate propagation @@ -120,4 +126,6 @@ observations into deterministic consensus candidates for simulation facts like "who shot first," and those reports now have schema-v0 wire contracts. The next useful layer is making authoritative mutation logs durable and compaction-aware. Membership and leader election should wait until basic replica catch-up is -boring. +boring. Authoritative mutation logs can now survive process restart through +`ICultNetShardMutationLogStore`; compaction and snapshot fallback are the next +places where the machine needs an honest boundary. diff --git a/src/GameCult.Networking/CultNetDatabase.cs b/src/GameCult.Networking/CultNetDatabase.cs index f2bd0790..8acb55f9 100644 --- a/src/GameCult.Networking/CultNetDatabase.cs +++ b/src/GameCult.Networking/CultNetDatabase.cs @@ -1,6 +1,7 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Reflection; using System.Threading.Tasks; using GameCult.Caching; using R3; @@ -225,6 +226,10 @@ namespace GameCult.Networking /// Gets or sets the document registry used to create and apply raw CultNet document messages. /// public CultNetDocumentRegistry? DocumentRegistry { get; set; } + /// + /// Gets or sets the durable store for accepted shard mutation logs. + /// + public ICultNetShardMutationLogStore? MutationLogStore { get; set; } } /// @@ -385,6 +390,7 @@ namespace GameCult.Networking private readonly string _runtimeId; private readonly List _shards; private readonly List _clientAuthorityScopes; + private readonly ICultNetShardMutationLogStore? _mutationLogStore; private readonly Dictionary _predictedDocuments = new(StringComparer.Ordinal); private readonly Dictionary> _mutationLogs = new(StringComparer.Ordinal); @@ -402,11 +408,13 @@ namespace GameCult.Networking options ??= new CultNetDatabaseOptions(); _runtimeId = options.RuntimeId; _documents = options.DocumentRegistry ?? new CultNetDocumentRegistry(cache.Registry); + _mutationLogStore = options.MutationLogStore; _shards = (options.Shards == null || options.Shards.Count == 0 ? new[] { CultNetShardDescriptor.PrimaryAll(options.RuntimeId) } : options.Shards) .ToList(); _clientAuthorityScopes = (options.ClientAuthorityScopes ?? Array.Empty()).ToList(); + InitializeLogSequencesFromStore(); _cache.OnUpdate += PublishCacheUpdate; } @@ -452,6 +460,26 @@ namespace GameCult.Networking return query.ToArray(); } + /// + /// Gets replica catch-up log entries for a shard after the supplied sequence. + /// + public IReadOnlyList GetMutationLogMessages( + string shardId, + long afterSequence = 0, + int? limit = null) + { + ThrowIfDisposed(); + if (string.IsNullOrWhiteSpace(shardId)) throw new ArgumentException("Value must be non-empty.", nameof(shardId)); + if (_mutationLogStore != null) + { + return _mutationLogStore.Read(shardId, afterSequence, limit); + } + + return GetMutationLog(shardId, afterSequence, limit) + .Select(ToLogEntryMessage) + .ToArray(); + } + /// /// Gets the last replicated sequence applied for a shard. /// @@ -639,7 +667,12 @@ namespace GameCult.Networking key, descriptor.SchemaId, document, - previous); + previous, + new CultNetShardLogEntryMessage + { + ChangeKind = previous == null ? "added" : "updated", + Put = message + }); PublishUntyped( descriptor.DocumentType, kind, @@ -680,7 +713,12 @@ namespace GameCult.Networking key, descriptor.SchemaId, document: null, - previousDocument: previous); + previousDocument: previous, + wireEntry: new CultNetShardLogEntryMessage + { + ChangeKind = "removed", + Delete = message + }); PublishUntyped( descriptor.DocumentType, CultNetDatabaseChangeKind.Removed, @@ -877,22 +915,30 @@ namespace GameCult.Networking CultRecordKey key, string schemaId, object? document, - object? previousDocument) + object? previousDocument, + CultNetShardLogEntryMessage? wireEntry = null) { var sequence = NextMutationLogSequence(shard.ShardId); - RecordMutationLogEntry(new CultNetShardMutationLogEntry( + var committedAt = DateTimeOffset.UtcNow.ToString("O"); + var entry = new CultNetShardMutationLogEntry( shard.ShardId, shard.Epoch, sequence, - DateTimeOffset.UtcNow.ToString("O"), + committedAt, kind, schemaId, key, document, - previousDocument)); + previousDocument); + var storedWireEntry = _mutationLogStore == null + ? null + : wireEntry ?? ToLogEntryMessage(entry); + RecordMutationLogEntry(entry, storedWireEntry); } - private void RecordMutationLogEntry(CultNetShardMutationLogEntry entry) + private void RecordMutationLogEntry( + CultNetShardMutationLogEntry entry, + CultNetShardLogEntryMessage? wireEntry = null) { if (!_mutationLogs.TryGetValue(entry.ShardId, out var entries)) { @@ -906,6 +952,11 @@ namespace GameCult.Networking { _nextLogSequences[entry.ShardId] = entry.Sequence + 1; } + + if (_mutationLogStore != null && wireEntry != null) + { + _mutationLogStore.Append(entry.ShardId, NormalizeWireLogEntry(entry, wireEntry)); + } } private long NextMutationLogSequence(string shardId) @@ -917,6 +968,26 @@ namespace GameCult.Networking return sequence; } + private void InitializeLogSequencesFromStore() + { + if (_mutationLogStore == null) + { + return; + } + + foreach (var shard in _shards) + { + var highest = _mutationLogStore.Read(shard.ShardId) + .Select(entry => entry.Sequence) + .DefaultIfEmpty(0) + .Max(); + if (highest > 0) + { + _nextLogSequences[shard.ShardId] = highest + 1; + } + } + } + private async Task ApplyCommittedShardLogEntryAsync( CultNetShardDescriptor shard, CultNetShardLogEntryMessage entry) @@ -983,7 +1054,8 @@ namespace GameCult.Networking descriptor.SchemaId, key, document, - previous)); + previous), + entry); PublishUntyped( descriptor.DocumentType, kind, @@ -1029,7 +1101,8 @@ namespace GameCult.Networking descriptor.SchemaId, key, document: null, - previousDocument: previous)); + previousDocument: previous), + entry); PublishUntyped( descriptor.DocumentType, CultNetDatabaseChangeKind.Removed, @@ -1040,6 +1113,93 @@ namespace GameCult.Networking previousDocument: previous); } + private CultNetShardLogEntryMessage ToLogEntryMessage(CultNetShardMutationLogEntry entry) + { + if (entry.Kind == CultNetDatabaseChangeKind.Removed || entry.Document == null) + { + return new CultNetShardLogEntryMessage + { + Sequence = entry.Sequence, + CommittedAt = entry.CommittedAt, + ChangeKind = "removed", + Delete = new CultNetDocumentDeleteMessage + { + MessageId = Guid.NewGuid().ToString("N"), + SchemaId = entry.SchemaId, + RecordKey = entry.Key.Value, + ShardId = entry.ShardId, + ShardEpoch = entry.ShardEpoch + } + }; + } + + return new CultNetShardLogEntryMessage + { + Sequence = entry.Sequence, + CommittedAt = entry.CommittedAt, + ChangeKind = entry.Kind == CultNetDatabaseChangeKind.Added ? "added" : "updated", + Put = new CultNetDocumentPutRawMessage + { + MessageId = Guid.NewGuid().ToString("N"), + Document = CreateRawDocumentRecord(entry.Key, entry.Document), + ShardId = entry.ShardId, + ShardEpoch = entry.ShardEpoch + } + }; + } + + private CultNetShardLogEntryMessage NormalizeWireLogEntry( + CultNetShardMutationLogEntry entry, + CultNetShardLogEntryMessage wireEntry) + { + return new CultNetShardLogEntryMessage + { + Sequence = entry.Sequence, + CommittedAt = string.IsNullOrWhiteSpace(entry.CommittedAt) ? wireEntry.CommittedAt : entry.CommittedAt, + ChangeKind = entry.Kind == CultNetDatabaseChangeKind.Removed + ? "removed" + : entry.Kind == CultNetDatabaseChangeKind.Added ? "added" : "updated", + Put = wireEntry.Put == null + ? null + : new CultNetDocumentPutRawMessage + { + MessageId = string.IsNullOrWhiteSpace(wireEntry.Put.MessageId) + ? Guid.NewGuid().ToString("N") + : wireEntry.Put.MessageId, + Document = wireEntry.Put.Document, + ShardId = entry.ShardId, + ShardEpoch = entry.ShardEpoch + }, + Delete = wireEntry.Delete == null + ? null + : new CultNetDocumentDeleteMessage + { + MessageId = string.IsNullOrWhiteSpace(wireEntry.Delete.MessageId) + ? Guid.NewGuid().ToString("N") + : wireEntry.Delete.MessageId, + SchemaId = wireEntry.Delete.SchemaId, + RecordKey = wireEntry.Delete.RecordKey, + ShardId = entry.ShardId, + ShardEpoch = entry.ShardEpoch + } + }; + } + + private CultNetRawDocumentRecord CreateRawDocumentRecord(CultRecordKey key, object document) + { + var method = typeof(CultNetDocumentRegistry) + .GetMethods(BindingFlags.Public | BindingFlags.Instance) + .Single(candidate => candidate.Name == nameof(CultNetDocumentRegistry.CreateRawDocumentPutMessage) && + candidate.IsGenericMethodDefinition); + var documentType = document.GetType(); + var handleType = typeof(CultRecordHandle<>).MakeGenericType(documentType); + var handle = Activator.CreateInstance(handleType, new object[] { key }); + var message = method + .MakeGenericMethod(documentType) + .Invoke(_documents, new[] { Guid.NewGuid().ToString("N"), handle, document, null }); + return ((CultNetDocumentPutRawMessage)message!).Document; + } + private static CultNetDatabaseChangeKind ChangeKindFromWire(string changeKind) { return changeKind switch diff --git a/src/GameCult.Networking/CultNetDatabaseServer.cs b/src/GameCult.Networking/CultNetDatabaseServer.cs index 7ea11894..321d1c8e 100644 --- a/src/GameCult.Networking/CultNetDatabaseServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseServer.cs @@ -132,8 +132,7 @@ namespace GameCult.Networking }; } - var entries = _database.GetMutationLog(request.ShardId, request.AfterSequence, request.Limit) - .Select(ToLogEntryMessage) + var entries = _database.GetMutationLogMessages(request.ShardId, request.AfterSequence, request.Limit) .ToArray(); return new CultNetShardLogResponseMessage @@ -334,41 +333,6 @@ namespace GameCult.Networking }; } - private CultNetShardLogEntryMessage ToLogEntryMessage(CultNetShardMutationLogEntry entry) - { - if (entry.Kind == CultNetDatabaseChangeKind.Removed || entry.Document == null) - { - return new CultNetShardLogEntryMessage - { - Sequence = entry.Sequence, - CommittedAt = entry.CommittedAt, - ChangeKind = "removed", - Delete = new CultNetDocumentDeleteMessage - { - MessageId = Guid.NewGuid().ToString("N"), - SchemaId = entry.SchemaId, - RecordKey = entry.Key.Value, - ShardId = entry.ShardId, - ShardEpoch = entry.ShardEpoch - } - }; - } - - return new CultNetShardLogEntryMessage - { - Sequence = entry.Sequence, - CommittedAt = entry.CommittedAt, - ChangeKind = entry.Kind == CultNetDatabaseChangeKind.Added ? "added" : "updated", - Put = new CultNetDocumentPutRawMessage - { - MessageId = Guid.NewGuid().ToString("N"), - Document = CreateRawDocumentRecord(entry.Key, entry.Document), - ShardId = entry.ShardId, - ShardEpoch = entry.ShardEpoch - } - }; - } - private CultNetRawDocumentRecord CreateRawDocumentRecord(CultRecordKey key, object document) { var method = typeof(CultNetDocumentRegistry) diff --git a/src/GameCult.Networking/CultNetShardMutationLogStorage.cs b/src/GameCult.Networking/CultNetShardMutationLogStorage.cs new file mode 100644 index 00000000..231c43df --- /dev/null +++ b/src/GameCult.Networking/CultNetShardMutationLogStorage.cs @@ -0,0 +1,109 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Security.Cryptography; +using MessagePack; + +namespace GameCult.Networking +{ + /// + /// Stores accepted shard-log entries in their replica catch-up wire shape. + /// + public interface ICultNetShardMutationLogStore + { + /// + /// Reads accepted log entries for a shard after the supplied sequence. + /// + IReadOnlyList Read(string shardId, long afterSequence = 0, int? limit = null); + + /// + /// Appends or replaces one accepted log entry for a shard. + /// + void Append(string shardId, CultNetShardLogEntryMessage entry); + } + + /// + /// Persists shard-log entries as one MessagePack file per shard. + /// + public sealed class CultNetFileShardMutationLogStore : ICultNetShardMutationLogStore + { + private readonly string _rootPath; + private readonly object _gate = new(); + + /// + /// Creates a file-backed shard mutation log store. + /// + public CultNetFileShardMutationLogStore(string rootPath) + { + if (string.IsNullOrWhiteSpace(rootPath)) throw new ArgumentException("Value must be non-empty.", nameof(rootPath)); + _rootPath = rootPath; + Directory.CreateDirectory(_rootPath); + } + + /// + public IReadOnlyList Read(string shardId, long afterSequence = 0, int? limit = null) + { + if (string.IsNullOrWhiteSpace(shardId)) throw new ArgumentException("Value must be non-empty.", nameof(shardId)); + if (afterSequence < 0) throw new ArgumentOutOfRangeException(nameof(afterSequence), "Sequence must be non-negative."); + + lock (_gate) + { + IEnumerable entries = ReadAll(shardId) + .Where(entry => entry.Sequence > afterSequence) + .OrderBy(entry => entry.Sequence); + if (limit.HasValue) + { + entries = entries.Take(limit.Value); + } + + return entries.ToArray(); + } + } + + /// + public void Append(string shardId, CultNetShardLogEntryMessage entry) + { + if (string.IsNullOrWhiteSpace(shardId)) throw new ArgumentException("Value must be non-empty.", nameof(shardId)); + if (entry == null) throw new ArgumentNullException(nameof(entry)); + if (entry.Sequence <= 0) throw new ArgumentOutOfRangeException(nameof(entry), "Log entry sequence must be positive."); + + lock (_gate) + { + var entries = ReadAll(shardId) + .Where(existing => existing.Sequence != entry.Sequence) + .Append(entry) + .OrderBy(existing => existing.Sequence) + .ToArray(); + var payload = MessagePackSerializer.Serialize(entries, CultNetSchemaMessageSerialization.Options); + File.WriteAllBytes(GetShardPath(shardId), payload); + } + } + + private CultNetShardLogEntryMessage[] ReadAll(string shardId) + { + var path = GetShardPath(shardId); + if (!File.Exists(path)) + { + return Array.Empty(); + } + + var payload = File.ReadAllBytes(path); + return MessagePackSerializer.Deserialize( + payload, + CultNetSchemaMessageSerialization.Options) ?? Array.Empty(); + } + + private string GetShardPath(string shardId) + { + return Path.Combine(_rootPath, HashShardId(shardId) + ".mpack"); + } + + private static string HashShardId(string shardId) + { + using var sha = SHA256.Create(); + var hash = sha.ComputeHash(System.Text.Encoding.UTF8.GetBytes(shardId)); + return string.Concat(hash.Select(value => value.ToString("x2"))); + } + } +} diff --git a/src/GameCult.Networking/README.md b/src/GameCult.Networking/README.md index 6c25ad32..d80b2013 100644 --- a/src/GameCult.Networking/README.md +++ b/src/GameCult.Networking/README.md @@ -156,7 +156,10 @@ primary endpoints, using `CultNetSchemaShardLogFetcher` for the schema-v0 transport path. Replica cursors can be stored through `ICultNetShardReplicaCursorStore`; `CultNetFileShardReplicaCursorStore` keeps those cursors in one local MessagePack file so replicas can resume after a -restart. +restart. Authoritative shard logs can be stored through +`ICultNetShardMutationLogStore`; `CultNetFileShardMutationLogStore` keeps one +MessagePack log file per shard and lets a restarted primary continue answering +`cultnet.shard_log_request.v0` from durable wire entries. For client-side prediction, `CultNetClientAuthorityScope` declares which input documents a runtime may author optimistically. `PutPredictedAsync` updates the diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index e41c6984..3fc31720 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -961,6 +961,67 @@ namespace GameCult.Networking.Tests Assert.That(response.Entries[0].Delete!.ShardEpoch, Is.EqualTo(6)); } + [Test] + public async Task CultNetDatabaseServer_Reads_DurableShardLog_AfterRestart() + { + var rootPath = Path.Combine(TestContext.CurrentContext.WorkDirectory, "shard-logs", Guid.NewGuid().ToString("N")); + var schemaId = new CultCache().Registry.GetRequired().SchemaId; + var shard = new CultNetShardDescriptor( + "players-durable-log", + "runtime-a", + epoch: 7, + isPrimary: true, + schemaIds: [schemaId], + keyPrefix: "player:"); + + var sourceCache = new CultCache(); + var sourceRegistry = new CultNetDocumentRegistry(sourceCache.Registry) + .Register(CultNetDocumentBinding.ForDocument( + sourceCache.Registry, + payloadSerializer: SerializePlayerDataPayload, + payloadDeserializer: DeserializePlayerDataPayload)); + var sourceDatabase = new CultNetDatabase(sourceCache, new CultNetDatabaseOptions + { + DocumentRegistry = sourceRegistry, + MutationLogStore = new CultNetFileShardMutationLogStore(rootPath), + Shards = [shard] + }); + var player = new PlayerData + { + PlayerId = Guid.NewGuid(), + Email = "durable@example.test", + PasswordHash = "hash", + Username = "Durable" + }; + + await sourceDatabase.PutAsync(new CultRecordKey("player:durable"), player); + sourceDatabase.Dispose(); + + var restartedCache = new CultCache(); + var restartedDatabase = new CultNetDatabase(restartedCache, new CultNetDatabaseOptions + { + MutationLogStore = new CultNetFileShardMutationLogStore(rootPath), + Shards = [shard] + }); + using var server = new Server(restartedCache, DevelopmentServerSecurity); + using var databaseServer = new CultNetDatabaseServer(server, restartedDatabase); + + var response = databaseServer.CreateShardLogResponse(new CultNetShardLogRequestMessage + { + MessageId = "durable-log", + ShardId = "players-durable-log", + ShardEpoch = 7 + }); + + Assert.That(response.ResyncRequired, Is.False); + Assert.That(response.Entries, Has.Length.EqualTo(1)); + Assert.That(response.Entries[0].Sequence, Is.EqualTo(1)); + Assert.That(response.Entries[0].Put, Is.Not.Null); + Assert.That(response.Entries[0].Put!.Document.SchemaId, Is.EqualTo(schemaId)); + Assert.That(response.Entries[0].Put!.Document.RecordKey, Is.EqualTo("player:durable")); + Assert.That(response.Entries[0].Put!.Document.Payload, Is.EqualTo(SerializePlayerDataPayload(player))); + } + [Test] public async Task CultNetDatabase_Applies_ShardLogResponse_ToReplica() {