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
{