From 65078c54a7cbacd3b8d26c1a0613f8a0ac8cddbb Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 19 May 2026 17:52:29 +0100 Subject: [PATCH] Add CultNet shard mutation logs --- .../cultnet-distributed-database/summary.md | 11 +- .../Contracts/cultnet-distributed-database.md | 24 ++- src/GameCult.Networking/CultNetDatabase.cs | 156 ++++++++++++++++++ src/GameCult.Networking/README.md | 5 +- .../NetworkingTests.cs | 75 +++++++++ 5 files changed, 255 insertions(+), 16 deletions(-) diff --git a/research/cultnet-distributed-database/summary.md b/research/cultnet-distributed-database/summary.md index 449c12a0..417b56ef 100644 --- a/research/cultnet-distributed-database/summary.md +++ b/research/cultnet-distributed-database/summary.md @@ -164,15 +164,16 @@ Shard catalog exchange now exists: - injectable non-primary write forwarding policy - concrete schema-v0 write forwarder for `cultnet://host:port` primary endpoints +- in-memory per-shard mutation logs and catch-up by last seen sequence ## Next Cut -Build per-shard replica logs: +Build wire-level replica log catch-up: -- assign sequence numbers to accepted shard mutations -- stream log entries to replicas and subscribers -- compact logs into snapshots -- make missed entries trigger explicit resync +- `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 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 1cdc7016..b99b8f70 100644 --- a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md +++ b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md @@ -47,12 +47,16 @@ Implemented: - injectable non-primary write forwarding policy - schema-v0 write forwarder that dials advertised `cultnet://host:port` primary endpoints +- 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 Not implemented yet: - membership/failure detection - leader election or automatic shard failover -- replicated log +- wire-level replica log request/response messages +- durable log persistence and compaction - declared CRDT merge policies ## Live Invariants @@ -69,13 +73,15 @@ Not implemented yet: ## Next Coherent Slice -Add per-shard replica logs: +Add wire-level replica log catch-up: -- assign monotonically increasing sequence numbers to accepted shard mutations -- stream log entries to replicas and subscribers -- compact log entries into CultCache snapshots -- reject or resync replicas that miss entries +- `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 -Forwarding now has both a policy seam and a concrete schema-v0 dialer. The next -useful layer is ordered replication. Membership and leader election should wait -until the replication log shape is boring. +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. diff --git a/src/GameCult.Networking/CultNetDatabase.cs b/src/GameCult.Networking/CultNetDatabase.cs index f4bbdd52..8887bf65 100644 --- a/src/GameCult.Networking/CultNetDatabase.cs +++ b/src/GameCult.Networking/CultNetDatabase.cs @@ -229,6 +229,74 @@ namespace GameCult.Networking public string? Message { get; } } + /// + /// One ordered mutation accepted for a shard. + /// + public sealed class CultNetShardMutationLogEntry + { + /// + /// Creates a shard mutation log entry. + /// + public CultNetShardMutationLogEntry( + string shardId, + long shardEpoch, + long sequence, + string committedAt, + CultNetDatabaseChangeKind kind, + string schemaId, + CultRecordKey key, + object? document, + object? previousDocument) + { + ShardId = shardId; + ShardEpoch = shardEpoch; + Sequence = sequence; + CommittedAt = committedAt; + Kind = kind; + SchemaId = schemaId; + Key = key; + Document = document; + PreviousDocument = previousDocument; + } + + /// + /// Gets the shard id. + /// + public string ShardId { get; } + /// + /// Gets the shard epoch. + /// + public long ShardEpoch { get; } + /// + /// Gets the per-shard mutation sequence. + /// + public long Sequence { get; } + /// + /// Gets the commit timestamp. + /// + public string CommittedAt { get; } + /// + /// Gets the mutation kind. + /// + public CultNetDatabaseChangeKind Kind { get; } + /// + /// Gets the schema id. + /// + public string SchemaId { get; } + /// + /// Gets the record key. + /// + public CultRecordKey Key { get; } + /// + /// Gets the current document, when present. + /// + public object? Document { get; } + /// + /// Gets the previous document, when present. + /// + public object? PreviousDocument { get; } + } + /// /// Raised when a database write targets a shard this node does not own. /// @@ -261,6 +329,9 @@ namespace GameCult.Networking private readonly CultCache _cache; private readonly CultNetDocumentRegistry _documents; private readonly List _shards; + private readonly Dictionary> _mutationLogs = + new(StringComparer.Ordinal); + private readonly Dictionary _nextLogSequences = new(StringComparer.Ordinal); private readonly Subject _changes = new(); private bool _disposed; @@ -292,6 +363,30 @@ namespace GameCult.Networking /// public IReadOnlyList Shards => _shards; + /// + /// Gets mutation log entries for a shard after the supplied sequence. + /// + public IReadOnlyList GetMutationLog( + string shardId, + long afterSequence = 0, + int? limit = null) + { + ThrowIfDisposed(); + if (string.IsNullOrWhiteSpace(shardId)) throw new ArgumentException("Value must be non-empty.", nameof(shardId)); + if (!_mutationLogs.TryGetValue(shardId, out var entries)) + { + return Array.Empty(); + } + + var query = entries.Where(entry => entry.Sequence > afterSequence); + if (limit.HasValue) + { + query = query.Take(limit.Value); + } + + return query.ToArray(); + } + /// /// Creates a shard catalog response for optional schema/key filters. /// @@ -353,6 +448,13 @@ namespace GameCult.Networking var previous = _cache.Get(key); var handle = await _cache.UpsertAsync(document, new CultRecordHandle(key)).ConfigureAwait(false); + AppendMutationLogEntry( + shard, + previous == null ? CultNetDatabaseChangeKind.Added : CultNetDatabaseChangeKind.Updated, + key, + descriptor.SchemaId, + document, + previous); Publish(new CultNetDatabaseChange( previous == null ? CultNetDatabaseChangeKind.Added : CultNetDatabaseChangeKind.Updated, key, @@ -377,6 +479,13 @@ namespace GameCult.Networking if (previous != null) { _cache.Remove(new CultRecordHandle(key)); + AppendMutationLogEntry( + shard, + CultNetDatabaseChangeKind.Removed, + key, + descriptor.SchemaId, + document: null, + previousDocument: previous); Publish(new CultNetDatabaseChange( CultNetDatabaseChangeKind.Removed, key, @@ -407,6 +516,13 @@ namespace GameCult.Networking var descriptor = _cache.Registry.GetRequiredBySchemaId(message.Document.SchemaId); var previous = _cache.Get(key); var document = await _documents.ApplyRawDocumentPutMessageAsync(_cache, message).ConfigureAwait(false); + AppendMutationLogEntry( + shard, + previous == null ? CultNetDatabaseChangeKind.Added : CultNetDatabaseChangeKind.Updated, + key, + descriptor.SchemaId, + document, + previous); PublishUntyped( descriptor.DocumentType, previous == null ? CultNetDatabaseChangeKind.Added : CultNetDatabaseChangeKind.Updated, @@ -441,6 +557,13 @@ namespace GameCult.Networking var handleType = typeof(CultRecordHandle<>).MakeGenericType(descriptor.DocumentType); var handle = Activator.CreateInstance(handleType, new object[] { key }); removeMethod.Invoke(_cache, new[] { handle }); + AppendMutationLogEntry( + shard, + CultNetDatabaseChangeKind.Removed, + key, + descriptor.SchemaId, + document: null, + previousDocument: previous); PublishUntyped( descriptor.DocumentType, CultNetDatabaseChangeKind.Removed, @@ -580,6 +703,39 @@ namespace GameCult.Networking _changes.OnNext(change); } + private void AppendMutationLogEntry( + CultNetShardDescriptor shard, + CultNetDatabaseChangeKind kind, + CultRecordKey key, + string schemaId, + object? document, + object? previousDocument) + { + var sequence = _nextLogSequences.TryGetValue(shard.ShardId, out var next) + ? next + : 1; + _nextLogSequences[shard.ShardId] = sequence + 1; + + var entry = new CultNetShardMutationLogEntry( + shard.ShardId, + shard.Epoch, + sequence, + DateTimeOffset.UtcNow.ToString("O"), + kind, + schemaId, + key, + document, + previousDocument); + + if (!_mutationLogs.TryGetValue(shard.ShardId, out var entries)) + { + entries = new List(); + _mutationLogs[shard.ShardId] = entries; + } + + entries.Add(entry); + } + private void PublishUntyped( Type documentType, CultNetDatabaseChangeKind kind, diff --git a/src/GameCult.Networking/README.md b/src/GameCult.Networking/README.md index cfb30b9d..23919e56 100644 --- a/src/GameCult.Networking/README.md +++ b/src/GameCult.Networking/README.md @@ -135,8 +135,9 @@ R3 watch streams, raw document puts/deletes, raw snapshot responses, 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. Per-shard replica logs are the -next missing organ. +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. 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 108cf461..c0e695ae 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -629,6 +629,81 @@ namespace GameCult.Networking.Tests Throws.TypeOf()); } + [Test] + public async Task CultNetDatabase_Appends_PerShardMutationLog() + { + var cache = new CultCache(); + var schemaId = cache.Registry.GetRequired().SchemaId; + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions + { + Shards = + [ + new CultNetShardDescriptor( + "players-log", + "runtime-a", + epoch: 4, + isPrimary: true, + schemaIds: [schemaId], + keyPrefix: "player:") + ] + }); + var key = new CultRecordKey("player:log"); + var player = new PlayerData + { + PlayerId = Guid.NewGuid(), + Email = "log@example.test", + PasswordHash = "hash", + Username = "Log" + }; + + await database.PutAsync(key, player); + player.Username = "LogUpdated"; + await database.PutAsync(key, player); + await database.DeleteAsync(key); + + var entries = database.GetMutationLog("players-log"); + + Assert.That(entries, Has.Count.EqualTo(3)); + Assert.That(new[] { entries[0].Sequence, entries[1].Sequence, entries[2].Sequence }, Is.EqualTo([1, 2, 3])); + Assert.That(entries[0].Kind, Is.EqualTo(CultNetDatabaseChangeKind.Added)); + Assert.That(entries[1].Kind, Is.EqualTo(CultNetDatabaseChangeKind.Updated)); + Assert.That(entries[2].Kind, Is.EqualTo(CultNetDatabaseChangeKind.Removed)); + Assert.That(entries[0].ShardEpoch, Is.EqualTo(4)); + } + + [Test] + public async Task CultNetDatabase_MutationLog_CanCatchUpAfterSequence() + { + var cache = new CultCache(); + var schemaId = cache.Registry.GetRequired().SchemaId; + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions + { + Shards = + [ + new CultNetShardDescriptor( + "players-catchup", + "runtime-a", + epoch: 1, + isPrimary: true, + schemaIds: [schemaId], + keyPrefix: "player:") + ] + }); + + await database.PutAsync( + new CultRecordKey("player:one"), + new PlayerData { PlayerId = Guid.NewGuid(), Email = "one@example.test", PasswordHash = "hash", Username = "One" }); + await database.PutAsync( + new CultRecordKey("player:two"), + new PlayerData { PlayerId = Guid.NewGuid(), Email = "two@example.test", PasswordHash = "hash", Username = "Two" }); + + var entries = database.GetMutationLog("players-catchup", afterSequence: 1); + + Assert.That(entries, Has.Count.EqualTo(1)); + Assert.That(entries[0].Sequence, Is.EqualTo(2)); + Assert.That(entries[0].Key.Value, Is.EqualTo("player:two")); + } + [Test] public void CultNetDatabaseServer_Creates_Filtered_SubscriptionChange() {