From a900687751a8be8096dc62977b3c226b606dd0a3 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 19 May 2026 18:01:11 +0100 Subject: [PATCH] Add CultNet shard log catch-up --- .../cultnet-distributed-database/summary.md | 10 +- .../Contracts/cultnet-distributed-database.md | 24 +-- .../CultNetDatabaseServer.cs | 109 +++++++++++++ .../CultNetSchemaMessageSerialization.cs | 2 + .../CultNetSchemaMessages.cs | 104 +++++++++++++ src/GameCult.Networking/README.md | 5 +- .../NetworkingTests.cs | 145 ++++++++++++++++++ 7 files changed, 381 insertions(+), 18 deletions(-) diff --git a/research/cultnet-distributed-database/summary.md b/research/cultnet-distributed-database/summary.md index 417b56ef..22199224 100644 --- a/research/cultnet-distributed-database/summary.md +++ b/research/cultnet-distributed-database/summary.md @@ -165,15 +165,15 @@ Shard catalog exchange now exists: - concrete schema-v0 write forwarder for `cultnet://host:port` primary endpoints - in-memory per-shard mutation logs and catch-up by last seen sequence +- wire-level shard mutation log catch-up with raw put/delete entries ## Next Cut -Build wire-level replica log catch-up: +Build durable replica catch-up: -- `cultnet.shard_log_request.v0` -- `cultnet.shard_log_response.v0` -- raw put/delete representation for log entries -- explicit resync-required failure when a replica asks for compacted history +- persist per-shard logs and last-applied replica sequence +- define the snapshot fallback boundary for compacted history +- add a replica-side apply loop for shard log responses That is the next foundation needed before membership, Raft-style failover, or CRDT policy work can be honest. diff --git a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md index b99b8f70..9f9afd4f 100644 --- a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md +++ b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md @@ -50,13 +50,15 @@ Implemented: - in-memory per-shard mutation logs with sequence numbers, epochs, mutation kind, schema id, key, and document references - mutation-log catch-up by last seen sequence +- wire-level shard log request/response messages +- raw put/delete representation for accepted mutation-log entries Not implemented yet: - membership/failure detection - leader election or automatic shard failover -- wire-level replica log request/response messages - durable log persistence and compaction +- replica-side log application loop - declared CRDT merge policies ## Live Invariants @@ -73,15 +75,15 @@ Not implemented yet: ## Next Coherent Slice -Add wire-level replica log catch-up: +Add durable log storage and replica application: -- `cultnet.shard_log_request.v0` -- `cultnet.shard_log_response.v0` -- serialize accepted mutations as raw put/delete records -- let replicas ask for entries after a known sequence -- return an explicit resync-required error when the requested sequence has - already been compacted away +- persist per-shard mutation logs instead of keeping them only in memory +- define the compaction boundary that turns old log reads into snapshot reads +- apply `cultnet.shard_log_response.v0` entries on replicas through + `CultNetDatabase` +- track last applied sequence per shard -The in-memory log shape now exists. The next useful layer is moving that log -over the wire. Membership and leader election should wait until replication -catch-up is boring. +The log is now wire-readable. The next useful layer is making a replica consume +that stream durably enough that restart/reconnect behavior can be explained +without hand waving. Membership and leader election should wait until basic +replica catch-up is boring. diff --git a/src/GameCult.Networking/CultNetDatabaseServer.cs b/src/GameCult.Networking/CultNetDatabaseServer.cs index a8f3b598..7ea11894 100644 --- a/src/GameCult.Networking/CultNetDatabaseServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseServer.cs @@ -21,6 +21,7 @@ namespace GameCult.Networking private readonly Func _putHandler; private readonly Func _deleteHandler; private readonly Func _shardCatalogHandler; + private readonly Func _shardLogHandler; private readonly Func _subscribeHandler; private readonly Func _unsubscribeHandler; private readonly ConcurrentDictionary _subscriptions = new(StringComparer.Ordinal); @@ -41,6 +42,7 @@ namespace GameCult.Networking _putHandler = HandlePutAsync; _deleteHandler = HandleDeleteAsync; _shardCatalogHandler = HandleShardCatalogRequestAsync; + _shardLogHandler = HandleShardLogRequestAsync; _subscribeHandler = HandleSubscribeAsync; _unsubscribeHandler = HandleUnsubscribeAsync; @@ -48,6 +50,7 @@ namespace GameCult.Networking _server.OnCultNet(_putHandler); _server.OnCultNet(_deleteHandler); _server.OnCultNet(_shardCatalogHandler); + _server.OnCultNet(_shardLogHandler); _server.OnCultNet(_subscribeHandler); _server.OnCultNet(_unsubscribeHandler); } @@ -93,6 +96,55 @@ namespace GameCult.Networking return _database.CreateShardCatalogResponse(request); } + /// + /// Creates a shard mutation-log response from the database log. + /// + public CultNetShardLogResponseMessage CreateShardLogResponse(CultNetShardLogRequestMessage request) + { + if (request == null) throw new ArgumentNullException(nameof(request)); + if (string.IsNullOrWhiteSpace(request.ShardId)) + { + throw new ArgumentException("Shard log request requires a shardId.", nameof(request)); + } + + var shard = _database.Shards.FirstOrDefault(candidate => + string.Equals(candidate.ShardId, request.ShardId, StringComparison.Ordinal)); + if (shard == null) + { + return new CultNetShardLogResponseMessage + { + MessageId = RequestMessageId(request.MessageId), + ShardId = request.ShardId, + ResyncRequired = true, + Reason = "unknown_shard" + }; + } + + if (request.ShardEpoch.HasValue && request.ShardEpoch.Value != shard.Epoch) + { + return new CultNetShardLogResponseMessage + { + MessageId = RequestMessageId(request.MessageId), + ShardId = request.ShardId, + ShardEpoch = shard.Epoch, + ResyncRequired = true, + Reason = "stale_epoch" + }; + } + + var entries = _database.GetMutationLog(request.ShardId, request.AfterSequence, request.Limit) + .Select(ToLogEntryMessage) + .ToArray(); + + return new CultNetShardLogResponseMessage + { + MessageId = RequestMessageId(request.MessageId), + ShardId = request.ShardId, + ShardEpoch = shard.Epoch, + Entries = entries + }; + } + /// /// Detaches handlers from the server. /// @@ -108,6 +160,7 @@ namespace GameCult.Networking _server.RemoveCultNetMessageListener(_putHandler); _server.RemoveCultNetMessageListener(_deleteHandler); _server.RemoveCultNetMessageListener(_shardCatalogHandler); + _server.RemoveCultNetMessageListener(_shardLogHandler); _server.RemoveCultNetMessageListener(_subscribeHandler); _server.RemoveCultNetMessageListener(_unsubscribeHandler); foreach (var subscription in _subscriptions.Values) @@ -130,6 +183,20 @@ namespace GameCult.Networking return Task.CompletedTask; } + private Task HandleShardLogRequestAsync(CultNetShardLogRequestMessage request, NetPeer peer) + { + try + { + peer.SendCultNet(CreateShardLogResponse(request)); + } + catch (Exception ex) + { + peer.SendCultNet(new CultNetErrorMessage { Error = ex.Message }); + } + + return Task.CompletedTask; + } + private async Task HandlePutAsync(CultNetDocumentPutRawMessage message, NetPeer peer) { try @@ -267,6 +334,41 @@ 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) @@ -298,6 +400,13 @@ namespace GameCult.Networking return $"{peer.Id}:{subscriptionId}"; } + private static string RequestMessageId(string messageId) + { + return string.IsNullOrWhiteSpace(messageId) + ? Guid.NewGuid().ToString("N") + : messageId; + } + private static CultNetErrorMessage CreateRoutingError(CultNetShardAuthorityException exception) { return new CultNetErrorMessage diff --git a/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs b/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs index 2cda2d20..786b3868 100644 --- a/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs +++ b/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs @@ -64,6 +64,8 @@ namespace GameCult.Networking CultNetSchemaVersions.DatabaseChangeRaw => Deserialize(payload), CultNetSchemaVersions.ShardCatalogRequest => Deserialize(payload), CultNetSchemaVersions.ShardCatalogResponse => Deserialize(payload), + CultNetSchemaVersions.ShardLogRequest => Deserialize(payload), + CultNetSchemaVersions.ShardLogResponse => Deserialize(payload), CultNetSchemaVersions.SchemaCatalogRequest => Deserialize(payload), CultNetSchemaVersions.SchemaCatalogResponse => Deserialize(payload), _ => throw new MessagePackSerializationException( diff --git a/src/GameCult.Networking/CultNetSchemaMessages.cs b/src/GameCult.Networking/CultNetSchemaMessages.cs index 0c579547..e09bf3ba 100644 --- a/src/GameCult.Networking/CultNetSchemaMessages.cs +++ b/src/GameCult.Networking/CultNetSchemaMessages.cs @@ -99,6 +99,14 @@ namespace GameCult.Networking /// shard catalog response contract identifier. /// public const string ShardCatalogResponse = "cultnet.shard_catalog_response.v0"; + /// + /// shard mutation log request contract identifier. + /// + public const string ShardLogRequest = "cultnet.shard_log_request.v0"; + /// + /// shard mutation log response contract identifier. + /// + public const string ShardLogResponse = "cultnet.shard_log_response.v0"; } /// @@ -711,6 +719,102 @@ namespace GameCult.Networking [Key("shards")] public CultNetShardDescriptorMessage[] Shards { get; set; } = Array.Empty(); } + /// + /// Requests accepted mutation log entries for a shard. + /// + [MessagePackObject] + public class CultNetShardLogRequestMessage : ICultNetSchemaMessage + { + /// + /// Gets or sets the schema version. + /// + [Key("schemaVersion")] public string SchemaVersion { get; set; } = CultNetSchemaVersions.ShardLogRequest; + /// + /// Gets or sets the request id. + /// + [Key("messageId")] public string MessageId { get; set; } = string.Empty; + /// + /// Gets or sets the shard id. + /// + [Key("shardId")] public string ShardId { get; set; } = string.Empty; + /// + /// Gets or sets the shard epoch expected by the requester. + /// + [Key("shardEpoch")] public long? ShardEpoch { get; set; } + /// + /// Gets or sets the last sequence already held by the requester. + /// + [Key("afterSequence")] public long AfterSequence { get; set; } + /// + /// Gets or sets the maximum number of entries to return. + /// + [Key("limit")] public int? Limit { get; set; } + } + + /// + /// Returns accepted mutation log entries for a shard. + /// + [MessagePackObject] + public class CultNetShardLogResponseMessage : ICultNetSchemaMessage + { + /// + /// Gets or sets the schema version. + /// + [Key("schemaVersion")] public string SchemaVersion { get; set; } = CultNetSchemaVersions.ShardLogResponse; + /// + /// Gets or sets the request id. + /// + [Key("messageId")] public string MessageId { get; set; } = string.Empty; + /// + /// Gets or sets the shard id. + /// + [Key("shardId")] public string ShardId { get; set; } = string.Empty; + /// + /// Gets or sets the shard epoch. + /// + [Key("shardEpoch")] public long ShardEpoch { get; set; } + /// + /// Gets or sets mutation log entries. + /// + [Key("entries")] public CultNetShardLogEntryMessage[] Entries { get; set; } = Array.Empty(); + /// + /// Gets or sets whether the requester must fall back to a shard snapshot. + /// + [Key("resyncRequired")] public bool ResyncRequired { get; set; } + /// + /// Gets or sets a machine-readable resync reason. + /// + [Key("reason")] public string? Reason { get; set; } + } + + /// + /// Describes one accepted shard mutation for replica catch-up. + /// + [MessagePackObject] + public class CultNetShardLogEntryMessage + { + /// + /// Gets or sets the per-shard sequence. + /// + [Key("sequence")] public long Sequence { get; set; } + /// + /// Gets or sets the commit timestamp. + /// + [Key("committedAt")] public string CommittedAt { get; set; } = string.Empty; + /// + /// Gets or sets the mutation kind. + /// + [Key("changeKind")] public string ChangeKind { get; set; } = string.Empty; + /// + /// Gets or sets a raw put message for added/updated entries. + /// + [Key("put")] public CultNetDocumentPutRawMessage? Put { get; set; } + /// + /// Gets or sets a raw delete message for removed entries. + /// + [Key("delete")] public CultNetDocumentDeleteMessage? Delete { get; set; } + } + /// /// Describes one schema exposed through the CultNet schema catalog. /// diff --git a/src/GameCult.Networking/README.md b/src/GameCult.Networking/README.md index 23919e56..10168ee9 100644 --- a/src/GameCult.Networking/README.md +++ b/src/GameCult.Networking/README.md @@ -136,8 +136,9 @@ schema/key-filtered live subscriptions, shard catalog exchange, and stale-epoch write rejection. Non-primary write forwarding is available behind an injectable `ICultNetShardWriteForwarder`, with `CultNetSchemaWriteForwarder` as the schema-v0 `cultnet://host:port` endpoint dialer. It also records accepted -mutations in per-shard ordered logs for local replica catch-up. Wire-level log -catch-up is the next missing organ. +mutations in per-shard ordered logs and exposes those logs over +`cultnet.shard_log_request.v0` / `cultnet.shard_log_response.v0` for replica +catch-up. The first coherent shard policy is primary ownership. Each shard has one authoritative writer at a time. Clients may connect to any node; a non-owner diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index c0e695ae..f044f787 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -276,6 +276,49 @@ namespace GameCult.Networking.Tests Assert.That(roundTrip.Shards[0].Region, Is.EqualTo("eu-west")); } + [Test] + public void CultNetSchemaMessageSerialization_RoundTrips_ShardLogResponse() + { + var message = new CultNetShardLogResponseMessage + { + MessageId = "log-1", + ShardId = "players-eu", + ShardEpoch = 12, + Entries = + [ + new CultNetShardLogEntryMessage + { + Sequence = 42, + CommittedAt = "2026-05-19T12:00:00.0000000Z", + ChangeKind = "updated", + Put = new CultNetDocumentPutRawMessage + { + MessageId = "put-42", + ShardId = "players-eu", + ShardEpoch = 12, + Document = new CultNetRawDocumentRecord + { + SchemaId = "schema-player", + RecordKey = "player:42", + Payload = [0x91, 0x2A] + } + } + } + ] + }; + + var payload = CultNetSchemaMessageSerialization.Serialize(message); + var roundTrip = (CultNetShardLogResponseMessage)CultNetSchemaMessageSerialization.Deserialize(payload); + + Assert.That(roundTrip.MessageId, Is.EqualTo("log-1")); + Assert.That(roundTrip.ShardId, Is.EqualTo("players-eu")); + Assert.That(roundTrip.ShardEpoch, Is.EqualTo(12)); + Assert.That(roundTrip.Entries, Has.Length.EqualTo(1)); + Assert.That(roundTrip.Entries[0].Sequence, Is.EqualTo(42)); + Assert.That(roundTrip.Entries[0].Put, Is.Not.Null); + Assert.That(roundTrip.Entries[0].Put!.Document.RecordKey, Is.EqualTo("player:42")); + } + [Test] public async Task CultNetDocumentRegistry_RawSnapshotReplication_PreservesPayloadBytes() { @@ -704,6 +747,108 @@ namespace GameCult.Networking.Tests Assert.That(entries[0].Key.Value, Is.EqualTo("player:two")); } + [Test] + public async Task CultNetDatabaseServer_Creates_ShardLogResponse_AfterSequence() + { + var cache = new CultCache(); + var registry = new CultNetDocumentRegistry(cache.Registry) + .Register(CultNetDocumentBinding.ForDocument( + cache.Registry, + payloadSerializer: SerializePlayerDataPayload, + payloadDeserializer: DeserializePlayerDataPayload)); + var schemaId = cache.Registry.GetRequired().SchemaId; + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions + { + DocumentRegistry = registry, + Shards = + [ + new CultNetShardDescriptor( + "players-wire-log", + "runtime-a", + epoch: 5, + isPrimary: true, + schemaIds: [schemaId], + keyPrefix: "player:") + ] + }); + using var server = new Server(cache, DevelopmentServerSecurity); + using var databaseServer = new CultNetDatabaseServer(server, database); + var first = new PlayerData { PlayerId = Guid.NewGuid(), Email = "one@example.test", PasswordHash = "hash", Username = "One" }; + var second = new PlayerData { PlayerId = Guid.NewGuid(), Email = "two@example.test", PasswordHash = "hash", Username = "Two" }; + + await database.PutAsync(new CultRecordKey("player:one"), first); + await database.PutAsync(new CultRecordKey("player:two"), second); + + var response = databaseServer.CreateShardLogResponse(new CultNetShardLogRequestMessage + { + MessageId = "wire-log", + ShardId = "players-wire-log", + ShardEpoch = 5, + AfterSequence = 1 + }); + + Assert.That(response.MessageId, Is.EqualTo("wire-log")); + Assert.That(response.ShardId, Is.EqualTo("players-wire-log")); + Assert.That(response.ShardEpoch, Is.EqualTo(5)); + Assert.That(response.ResyncRequired, Is.False); + Assert.That(response.Entries, Has.Length.EqualTo(1)); + Assert.That(response.Entries[0].Sequence, Is.EqualTo(2)); + Assert.That(response.Entries[0].ChangeKind, Is.EqualTo("added")); + Assert.That(response.Entries[0].Put, Is.Not.Null); + Assert.That(response.Entries[0].Put!.ShardId, Is.EqualTo("players-wire-log")); + Assert.That(response.Entries[0].Put!.ShardEpoch, Is.EqualTo(5)); + Assert.That(response.Entries[0].Put!.Document.Payload, Is.EqualTo(SerializePlayerDataPayload(second))); + } + + [Test] + public async Task CultNetDatabaseServer_Creates_ShardLogDeleteEntry() + { + var cache = new CultCache(); + var registry = new CultNetDocumentRegistry(cache.Registry) + .Register(CultNetDocumentBinding.ForDocument( + cache.Registry, + payloadSerializer: SerializePlayerDataPayload, + payloadDeserializer: DeserializePlayerDataPayload)); + var schemaId = cache.Registry.GetRequired().SchemaId; + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions + { + DocumentRegistry = registry, + Shards = + [ + new CultNetShardDescriptor( + "players-wire-delete", + "runtime-a", + epoch: 6, + isPrimary: true, + schemaIds: [schemaId], + keyPrefix: "player:") + ] + }); + using var server = new Server(cache, DevelopmentServerSecurity); + using var databaseServer = new CultNetDatabaseServer(server, database); + var key = new CultRecordKey("player:delete-log"); + + await database.PutAsync( + key, + new PlayerData { PlayerId = Guid.NewGuid(), Email = "delete@example.test", PasswordHash = "hash", Username = "Delete" }); + await database.DeleteAsync(key); + + var response = databaseServer.CreateShardLogResponse(new CultNetShardLogRequestMessage + { + ShardId = "players-wire-delete", + AfterSequence = 1 + }); + + Assert.That(response.Entries, Has.Length.EqualTo(1)); + Assert.That(response.Entries[0].Sequence, Is.EqualTo(2)); + Assert.That(response.Entries[0].ChangeKind, Is.EqualTo("removed")); + Assert.That(response.Entries[0].Put, Is.Null); + Assert.That(response.Entries[0].Delete, Is.Not.Null); + Assert.That(response.Entries[0].Delete!.SchemaId, Is.EqualTo(schemaId)); + Assert.That(response.Entries[0].Delete!.RecordKey, Is.EqualTo(key.Value)); + Assert.That(response.Entries[0].Delete!.ShardEpoch, Is.EqualTo(6)); + } + [Test] public void CultNetDatabaseServer_Creates_Filtered_SubscriptionChange() {