Add CultNet shard replication puller

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.

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.

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
+{
+    /// <summary>
+    /// Fetches committed shard-log entries from an authoritative shard owner.
+    /// </summary>
+    public interface ICultNetShardLogFetcher
+    {
+        /// <summary>
+        /// Fetches shard-log entries after the supplied sequence.
+        /// </summary>
+        Task<CultNetShardLogResponseMessage> FetchAsync(
+            CultNetShardDescriptor shard,
+            long afterSequence,
+            int? limit = null);
+    }
+
+    /// <summary>
+    /// Options for background shard-log replication.
+    /// </summary>
+    public sealed class CultNetShardReplicatorOptions
+    {
+        /// <summary>
+        /// Gets or sets the interval used by the background pull loop.
+        /// </summary>
+        public TimeSpan PollInterval { get; set; } = TimeSpan.FromSeconds(1);
+
+        /// <summary>
+        /// Gets or sets the maximum number of log entries requested per pull.
+        /// </summary>
+        public int? BatchSize { get; set; } = 256;
+
+        /// <summary>
+        /// Gets or sets the fetcher used to read from shard primaries.
+        /// </summary>
+        public ICultNetShardLogFetcher? Fetcher { get; set; }
+
+        /// <summary>
+        /// Gets or sets a callback for background pull failures.
+        /// </summary>
+        public Action<Exception>? OnError { get; set; }
+    }
+
+    /// <summary>
+    /// Pulls committed shard logs from primaries and applies them to a local replica database.
+    /// </summary>
+    public sealed class CultNetShardReplicator : IDisposable
+    {
+        private readonly CultNetDatabase _database;
+        private readonly CultNetShardReplicatorOptions _options;
+        private readonly List<IDisposable> _subscriptions = new();
+        private readonly ConcurrentDictionary<string, byte> _pullsInFlight = new(StringComparer.Ordinal);
+        private bool _disposed;
+
+        /// <summary>
+        /// Creates a shard replicator.
+        /// </summary>
+        public CultNetShardReplicator(
+            CultNetDatabase database,
+            CultNetShardReplicatorOptions? options = null)
+        {
+            _database = database ?? throw new ArgumentNullException(nameof(database));
+            _options = options ?? new CultNetShardReplicatorOptions();
+        }
+
+        /// <summary>
+        /// Starts background pull loops for non-primary shards with advertised primaries.
+        /// </summary>
+        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));
+            }
+        }
+
+        /// <summary>
+        /// Pulls and applies one batch for a shard id.
+        /// </summary>
+        public Task<long> 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);
+        }
+
+        /// <summary>
+        /// Pulls and applies one batch for a shard.
+        /// </summary>
+        public async Task<long> 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);
+        }
+
+        /// <inheritdoc />
+        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));
+            }
+        }
+    }
+
+    /// <summary>
+    /// Options for the schema-v0 client-based shard-log fetcher.
+    /// </summary>
+    public sealed class CultNetSchemaShardLogFetcherOptions
+    {
+        /// <summary>
+        /// Gets or sets client security options used to connect to primary endpoints.
+        /// </summary>
+        public ClientSecurityOptions? Security { get; set; }
+
+        /// <summary>
+        /// Gets or sets how long to wait for a connection before failing.
+        /// </summary>
+        public TimeSpan ConnectTimeout { get; set; } = TimeSpan.FromSeconds(5);
+
+        /// <summary>
+        /// Gets or sets how long to wait for a log response before failing.
+        /// </summary>
+        public TimeSpan ResponseTimeout { get; set; } = TimeSpan.FromSeconds(5);
+
+        /// <summary>
+        /// Gets or sets a callback used to customize each ephemeral fetch client.
+        /// </summary>
+        public Action<Client>? ConfigureClient { get; set; }
+    }
+
+    /// <summary>
+    /// Fetches shard-log batches over the CultNet schema-v0 client transport.
+    /// </summary>
+    public sealed class CultNetSchemaShardLogFetcher : ICultNetShardLogFetcher
+    {
+        private readonly CultNetSchemaShardLogFetcherOptions _options;
+
+        /// <summary>
+        /// Creates a schema-v0 shard-log fetcher.
+        /// </summary>
+        public CultNetSchemaShardLogFetcher(CultNetSchemaShardLogFetcherOptions? options = null)
+        {
+            _options = options ?? new CultNetSchemaShardLogFetcherOptions();
+        }
+
+        /// <inheritdoc />
+        public async Task<CultNetShardLogResponseMessage> 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<CultNetShardLogResponseMessage>(
+                TaskCreationOptions.RunContinuationsAsynchronously);
+
+            using var client = new Client(_options.Security ?? ClientSecurityOptions.Development())
+            {
+                AllowUnverifiedCultNetMessages = true
+            };
+            _options.ConfigureClient?.Invoke(client);
+            client.OnCultNet<CultNetShardLogResponseMessage>(response =>
+            {
+                if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal))
+                {
+                    completion.TrySetResult(response);
+                }
+            });
+            client.OnCultNet<CultNetErrorMessage>(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<CultNetShardLogResponseMessage> WaitForResponseAsync(
+            Task<CultNetShardLogResponseMessage> 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.");
+        }
+    }
+}

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

tests/GameCult.Networking.Tests/NetworkingTests.cs

@@ -972,6 +972,71 @@ namespace GameCult.Networking.Tests
                 Throws.TypeOf<InvalidOperationException>().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<PlayerData>(
+                    cache.Registry,
+                    payloadSerializer: SerializePlayerDataPayload,
+                    payloadDeserializer: DeserializePlayerDataPayload));
+            var schemaId = cache.Registry.GetRequired<PlayerData>().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<PlayerData>(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<PlayerData>(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<CultNetShardLogResponseMessage> 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
         {