From d778c36c273d36e008a1ad77c4e7452bc72c8a69 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 19 May 2026 18:12:24 +0100 Subject: [PATCH] Add CultNet shard replication puller --- .../cultnet-distributed-database/summary.md | 5 +- .../Contracts/cultnet-distributed-database.md | 16 +- .../CultNetShardReplication.cs | 305 ++++++++++++++++++ src/GameCult.Networking/README.md | 3 + .../NetworkingTests.cs | 92 ++++++ 5 files changed, 412 insertions(+), 9 deletions(-) create mode 100644 src/GameCult.Networking/CultNetShardReplication.cs diff --git a/research/cultnet-distributed-database/summary.md b/research/cultnet-distributed-database/summary.md index febfa8a1..698f9a92 100644 --- a/research/cultnet-distributed-database/summary.md +++ b/research/cultnet-distributed-database/summary.md @@ -168,14 +168,15 @@ Shard catalog exchange now exists: - wire-level shard mutation log catch-up with raw put/delete entries - replica-side application of shard log responses with epoch checks, gap rejection, and idempotent replay +- background shard log replicator plus schema-v0 fetcher for primary endpoints ## Next Cut -Build durable replica catch-up loops: +Build durable replica catch-up state: - persist per-shard logs and last-applied replica sequence - define the snapshot fallback boundary for compacted history -- add a background pull loop from catalog-advertised primary endpoints +- store replica cursors across restarts 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 e69f49f9..eb159b96 100644 --- a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md +++ b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md @@ -54,13 +54,16 @@ Implemented: - raw put/delete representation for accepted mutation-log entries - replica-side shard log application with epoch checks, gap rejection, and idempotent replay +- shard log replicator that pulls non-primary shards from catalog-advertised + primary endpoints +- schema-v0 shard log fetcher for `cultnet://host:port` endpoints Not implemented yet: - membership/failure detection - leader election or automatic shard failover - durable log persistence and compaction -- background replica pull loops +- durable replica cursor storage across restarts - declared CRDT merge policies ## Live Invariants @@ -77,14 +80,13 @@ Not implemented yet: ## Next Coherent Slice -Add durable log storage and background replica pull loops: +Add durable log storage and restart-safe replica cursors: - persist per-shard mutation logs instead of keeping them only in memory - define the compaction boundary that turns old log reads into snapshot reads - store last applied sequence per shard across restarts -- let replicas periodically pull from primary endpoints in the shard catalog -The log is now wire-readable and replicas can apply it explicitly. The next -useful layer is making that state survive restart and giving replicas a pull -loop. Membership and leader election should wait until basic replica catch-up is -boring. +The log is now wire-readable, replicas can apply it explicitly, and a pull loop +can drive non-primary shards from primary endpoints. The next useful layer is +making that state survive restart. Membership and leader election should wait +until basic replica catch-up is boring. diff --git a/src/GameCult.Networking/CultNetShardReplication.cs b/src/GameCult.Networking/CultNetShardReplication.cs new file mode 100644 index 00000000..e03e01cf --- /dev/null +++ b/src/GameCult.Networking/CultNetShardReplication.cs @@ -0,0 +1,305 @@ +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; + +namespace GameCult.Networking +{ + /// + /// Fetches committed shard-log entries from an authoritative shard owner. + /// + public interface ICultNetShardLogFetcher + { + /// + /// Fetches shard-log entries after the supplied sequence. + /// + Task FetchAsync( + CultNetShardDescriptor shard, + long afterSequence, + int? limit = null); + } + + /// + /// Options for background shard-log replication. + /// + public sealed class CultNetShardReplicatorOptions + { + /// + /// Gets or sets the interval used by the background pull loop. + /// + public TimeSpan PollInterval { get; set; } = TimeSpan.FromSeconds(1); + + /// + /// Gets or sets the maximum number of log entries requested per pull. + /// + public int? BatchSize { get; set; } = 256; + + /// + /// Gets or sets the fetcher used to read from shard primaries. + /// + public ICultNetShardLogFetcher? Fetcher { get; set; } + + /// + /// Gets or sets a callback for background pull failures. + /// + public Action? OnError { get; set; } + } + + /// + /// Pulls committed shard logs from primaries and applies them to a local replica database. + /// + public sealed class CultNetShardReplicator : IDisposable + { + private readonly CultNetDatabase _database; + private readonly CultNetShardReplicatorOptions _options; + private readonly List _subscriptions = new(); + private readonly ConcurrentDictionary _pullsInFlight = new(StringComparer.Ordinal); + private bool _disposed; + + /// + /// Creates a shard replicator. + /// + public CultNetShardReplicator( + CultNetDatabase database, + CultNetShardReplicatorOptions? options = null) + { + _database = database ?? throw new ArgumentNullException(nameof(database)); + _options = options ?? new CultNetShardReplicatorOptions(); + } + + /// + /// Starts background pull loops for non-primary shards with advertised primaries. + /// + public void Start() + { + ThrowIfDisposed(); + if (_options.Fetcher == null) + { + throw new InvalidOperationException("A shard log fetcher is required before starting replication."); + } + + if (_subscriptions.Count > 0) + { + return; + } + + foreach (var shard in _database.Shards.Where(shard => !shard.IsPrimary && shard.PrimaryEndpoints.Count > 0)) + { + _subscriptions.Add(new Timer( + _ => _ = PullIfIdleAsync(shard), + state: null, + dueTime: TimeSpan.Zero, + period: _options.PollInterval)); + } + } + + /// + /// Pulls and applies one batch for a shard id. + /// + public Task PullOnceAsync(string shardId) + { + ThrowIfDisposed(); + if (string.IsNullOrWhiteSpace(shardId)) throw new ArgumentException("Value must be non-empty.", nameof(shardId)); + var shard = _database.Shards.FirstOrDefault(candidate => + string.Equals(candidate.ShardId, shardId, StringComparison.Ordinal)); + if (shard == null) + { + throw new InvalidOperationException($"Shard '{shardId}' is not known by this database."); + } + + return PullOnceAsync(shard); + } + + /// + /// Pulls and applies one batch for a shard. + /// + public async Task PullOnceAsync(CultNetShardDescriptor shard) + { + ThrowIfDisposed(); + if (shard == null) throw new ArgumentNullException(nameof(shard)); + if (_options.Fetcher == null) + { + throw new InvalidOperationException("A shard log fetcher is required before pulling replication."); + } + + if (shard.IsPrimary) + { + throw new InvalidOperationException($"Shard '{shard.ShardId}' is primary on this node and does not need replica pulling."); + } + + var afterSequence = _database.GetAppliedShardSequence(shard.ShardId); + var response = await _options.Fetcher + .FetchAsync(shard, afterSequence, _options.BatchSize) + .ConfigureAwait(false); + return await _database.ApplyShardLogResponseAsync(response).ConfigureAwait(false); + } + + /// + public void Dispose() + { + if (_disposed) + { + return; + } + + _disposed = true; + foreach (var subscription in _subscriptions) + { + subscription.Dispose(); + } + + _subscriptions.Clear(); + _pullsInFlight.Clear(); + } + + private async Task PullIfIdleAsync(CultNetShardDescriptor shard) + { + if (!_pullsInFlight.TryAdd(shard.ShardId, 0)) + { + return; + } + + try + { + await PullOnceAsync(shard).ConfigureAwait(false); + } + catch (Exception ex) + { + _options.OnError?.Invoke(ex); + } + finally + { + _pullsInFlight.TryRemove(shard.ShardId, out _); + } + } + + private void ThrowIfDisposed() + { + if (_disposed) + { + throw new ObjectDisposedException(nameof(CultNetShardReplicator)); + } + } + } + + /// + /// Options for the schema-v0 client-based shard-log fetcher. + /// + public sealed class CultNetSchemaShardLogFetcherOptions + { + /// + /// Gets or sets client security options used to connect to primary endpoints. + /// + public ClientSecurityOptions? Security { get; set; } + + /// + /// Gets or sets how long to wait for a connection before failing. + /// + public TimeSpan ConnectTimeout { get; set; } = TimeSpan.FromSeconds(5); + + /// + /// Gets or sets how long to wait for a log response before failing. + /// + public TimeSpan ResponseTimeout { get; set; } = TimeSpan.FromSeconds(5); + + /// + /// Gets or sets a callback used to customize each ephemeral fetch client. + /// + public Action? ConfigureClient { get; set; } + } + + /// + /// Fetches shard-log batches over the CultNet schema-v0 client transport. + /// + public sealed class CultNetSchemaShardLogFetcher : ICultNetShardLogFetcher + { + private readonly CultNetSchemaShardLogFetcherOptions _options; + + /// + /// Creates a schema-v0 shard-log fetcher. + /// + public CultNetSchemaShardLogFetcher(CultNetSchemaShardLogFetcherOptions? options = null) + { + _options = options ?? new CultNetSchemaShardLogFetcherOptions(); + } + + /// + public async Task FetchAsync( + CultNetShardDescriptor shard, + long afterSequence, + int? limit = null) + { + if (shard == null) throw new ArgumentNullException(nameof(shard)); + var endpoint = ResolvePrimaryEndpoint(shard); + var (host, port) = CultNetSchemaWriteForwarder.ParseEndpoint(endpoint); + var messageId = Guid.NewGuid().ToString("N"); + var completion = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + + using var client = new Client(_options.Security ?? ClientSecurityOptions.Development()) + { + AllowUnverifiedCultNetMessages = true + }; + _options.ConfigureClient?.Invoke(client); + client.OnCultNet(response => + { + if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal)) + { + completion.TrySetResult(response); + } + }); + client.OnCultNet(error => + completion.TrySetException(new InvalidOperationException(error.Error))); + + client.Connect(host, port); + await WaitForConnectionAsync(client, endpoint).ConfigureAwait(false); + client.SendCultNet(new CultNetShardLogRequestMessage + { + MessageId = messageId, + ShardId = shard.ShardId, + ShardEpoch = shard.Epoch, + AfterSequence = afterSequence, + Limit = limit + }); + + return await WaitForResponseAsync(completion.Task, endpoint).ConfigureAwait(false); + } + + private async Task WaitForConnectionAsync(Client client, string endpoint) + { + var deadline = DateTimeOffset.UtcNow + _options.ConnectTimeout; + while (!client.Connected) + { + if (DateTimeOffset.UtcNow >= deadline) + { + throw new TimeoutException($"Timed out connecting to shard primary endpoint {endpoint}."); + } + + await Task.Delay(TimeSpan.FromMilliseconds(25)).ConfigureAwait(false); + } + } + + private async Task WaitForResponseAsync( + Task responseTask, + string endpoint) + { + var timeoutTask = Task.Delay(_options.ResponseTimeout); + var completed = await Task.WhenAny(responseTask, timeoutTask).ConfigureAwait(false); + if (completed != responseTask) + { + throw new TimeoutException($"Timed out waiting for shard log response from {endpoint}."); + } + + return await responseTask.ConfigureAwait(false); + } + + private static string ResolvePrimaryEndpoint(CultNetShardDescriptor shard) + { + return shard.PrimaryEndpoints.FirstOrDefault() + ?? throw new InvalidOperationException( + $"Shard '{shard.ShardId}' does not advertise a primary endpoint."); + } + } +} diff --git a/src/GameCult.Networking/README.md b/src/GameCult.Networking/README.md index 38741618..030623c1 100644 --- a/src/GameCult.Networking/README.md +++ b/src/GameCult.Networking/README.md @@ -141,6 +141,9 @@ mutations in per-shard ordered logs and exposes those logs over catch-up. Replicas apply those committed entries through `CultNetDatabase.ApplyShardLogResponseAsync`, which checks shard epoch, rejects sequence gaps, and treats replayed entries as already applied. +`CultNetShardReplicator` can then pull non-primary shard logs from advertised +primary endpoints, using `CultNetSchemaShardLogFetcher` for the schema-v0 +transport path. 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 5af57b71..6eea0179 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -972,6 +972,71 @@ namespace GameCult.Networking.Tests Throws.TypeOf().With.Message.Contains("log has a gap")); } + [Test] + public async Task CultNetShardReplicator_Pulls_AndApplies_NextBatch() + { + 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 shard = new CultNetShardDescriptor( + "players-pull", + "runtime-a", + epoch: 2, + isPrimary: false, + schemaIds: [schemaId], + keyPrefix: "player:", + primaryEndpoints: ["cultnet://primary.example.test:3075"]); + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions + { + DocumentRegistry = registry, + Shards = [shard] + }); + var key = new CultRecordKey("player:pull"); + var player = new PlayerData + { + PlayerId = Guid.NewGuid(), + Email = "pull@example.test", + PasswordHash = "hash", + Username = "Pull" + }; + var put = registry.CreateRawDocumentPutMessage("put-pull", new CultRecordHandle(key), player); + put.ShardId = "players-pull"; + put.ShardEpoch = 2; + var fetcher = new CapturingShardLogFetcher(new CultNetShardLogResponseMessage + { + ShardId = "players-pull", + ShardEpoch = 2, + Entries = + [ + new CultNetShardLogEntryMessage + { + Sequence = 1, + CommittedAt = "2026-05-19T12:00:00.0000000Z", + ChangeKind = "added", + Put = put + } + ] + }); + using var replicator = new CultNetShardReplicator(database, new CultNetShardReplicatorOptions + { + Fetcher = fetcher, + BatchSize = 7 + }); + + var sequence = await replicator.PullOnceAsync("players-pull"); + + Assert.That(sequence, Is.EqualTo(1)); + Assert.That(fetcher.FetchCount, Is.EqualTo(1)); + Assert.That(fetcher.LastShard!.ShardId, Is.EqualTo("players-pull")); + Assert.That(fetcher.LastAfterSequence, Is.EqualTo(0)); + Assert.That(fetcher.LastLimit, Is.EqualTo(7)); + Assert.That(cache.Get(key)!.Username, Is.EqualTo("Pull")); + } + [Test] public void CultNetDatabaseServer_Creates_Filtered_SubscriptionChange() { @@ -1201,6 +1266,33 @@ namespace GameCult.Networking.Tests } } + private sealed class CapturingShardLogFetcher : ICultNetShardLogFetcher + { + private readonly CultNetShardLogResponseMessage _response; + + public CapturingShardLogFetcher(CultNetShardLogResponseMessage response) + { + _response = response; + } + + public int FetchCount { get; private set; } + public CultNetShardDescriptor? LastShard { get; private set; } + public long LastAfterSequence { get; private set; } + public int? LastLimit { get; private set; } + + public Task FetchAsync( + CultNetShardDescriptor shard, + long afterSequence, + int? limit = null) + { + FetchCount++; + LastShard = shard; + LastAfterSequence = afterSequence; + LastLimit = limit; + return Task.FromResult(_response); + } + } + [MessagePack.MessagePackObject] public sealed class PlayerDataPayload {