diff --git a/research/cultnet-distributed-database/summary.md b/research/cultnet-distributed-database/summary.md
index b0dcd4e0..449c12a0 100644
--- a/research/cultnet-distributed-database/summary.md
+++ b/research/cultnet-distributed-database/summary.md
@@ -162,15 +162,17 @@ Shard catalog exchange now exists:
- stale epoch rejection on writes
- routing error that tells clients where the primary lives when known
- injectable non-primary write forwarding policy
+- concrete schema-v0 write forwarder for `cultnet://host:port` primary
+ endpoints
## Next Cut
-Build a concrete schema-v0 write forwarder:
+Build per-shard replica logs:
-- dial the selected primary endpoint from the shard catalog
-- forwarded writes preserve shard id and epoch
-- stale epochs still fail before forwarding
-- failed forwarding returns endpoint-aware routing errors
+- assign sequence numbers to accepted shard mutations
+- stream log entries to replicas and subscribers
+- compact logs into snapshots
+- make missed entries trigger explicit resync
-That is the next foundation needed before membership, Raft-style replication,
-or CRDT policy work can be honest.
+That is the next foundation needed before membership, Raft-style failover, or
+CRDT policy work can be honest.
diff --git a/src/GameCult.Networking/Client.cs b/src/GameCult.Networking/Client.cs
index 67ca9fe5..fdfe4ba3 100644
--- a/src/GameCult.Networking/Client.cs
+++ b/src/GameCult.Networking/Client.cs
@@ -87,6 +87,11 @@ namespace GameCult.Networking
///
public bool LogSensitivePayloads { get; set; }
+ ///
+ /// Gets or sets whether CultNet schema-v0 messages may be sent before user session verification.
+ ///
+ public bool AllowUnverifiedCultNetMessages { get; set; }
+
///
/// Gets a value indicating whether the client currently has a verified session.
///
@@ -152,7 +157,7 @@ namespace GameCult.Networking
if (Connected && _peer != null)
{
- if (Verified || message is CultNetLoginMessage or CultNetRegisterMessage or CultNetVerifyMessage)
+ if (AllowUnverifiedCultNetMessages || Verified || message is CultNetLoginMessage or CultNetRegisterMessage or CultNetVerifyMessage)
{
Logger.LogDebug($"Sending CultNet schema message {message.SchemaVersion}");
_peer.SendCultNet(message);
diff --git a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md
index 807ea2c1..1cdc7016 100644
--- a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md
+++ b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md
@@ -45,13 +45,14 @@ Implemented:
- epoch-aware raw put/delete messages
- routing hints on shard authority errors
- injectable non-primary write forwarding policy
+- schema-v0 write forwarder that dials advertised `cultnet://host:port`
+ primary endpoints
Not implemented yet:
- membership/failure detection
- leader election or automatic shard failover
- replicated log
-- concrete endpoint dialer for forwarded writes
- declared CRDT merge policies
## Live Invariants
@@ -68,13 +69,13 @@ Not implemented yet:
## Next Coherent Slice
-Add a concrete endpoint dialer for write forwarding:
+Add per-shard replica logs:
-- implement `ICultNetShardWriteForwarder` over the schema-v0 client transport
-- choose the best primary endpoint from the shard descriptor
-- preserve shard id and epoch on forwarded messages
-- surface forwarding failures as routing errors with endpoint context
+- 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
-Forwarding now has the policy seam. The next useful layer is the actual network
-dialer. Membership, leader election, and replicated logs should wait until
-routing is boring.
+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.
diff --git a/src/GameCult.Networking/CultNetShardWriteForwarding.cs b/src/GameCult.Networking/CultNetShardWriteForwarding.cs
index 97c68676..d5114d69 100644
--- a/src/GameCult.Networking/CultNetShardWriteForwarding.cs
+++ b/src/GameCult.Networking/CultNetShardWriteForwarding.cs
@@ -1,3 +1,5 @@
+using System;
+using System.Linq;
using System.Threading.Tasks;
namespace GameCult.Networking
@@ -33,4 +35,116 @@ namespace GameCult.Networking
///
public ICultNetShardWriteForwarder? WriteForwarder { get; set; }
}
+
+ ///
+ /// Options for the schema-v0 client-based shard write forwarder.
+ ///
+ public sealed class CultNetSchemaWriteForwarderOptions
+ {
+ ///
+ /// 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 the forwarded write.
+ ///
+ public TimeSpan ConnectTimeout { get; set; } = TimeSpan.FromSeconds(5);
+
+ ///
+ /// Gets or sets a callback used to customize each ephemeral forwarding client.
+ ///
+ public Action? ConfigureClient { get; set; }
+ }
+
+ ///
+ /// Forwards shard writes over the CultNet schema-v0 client transport.
+ ///
+ public sealed class CultNetSchemaWriteForwarder : ICultNetShardWriteForwarder
+ {
+ private readonly CultNetSchemaWriteForwarderOptions _options;
+
+ ///
+ /// Creates a schema-v0 write forwarder.
+ ///
+ public CultNetSchemaWriteForwarder(CultNetSchemaWriteForwarderOptions? options = null)
+ {
+ _options = options ?? new CultNetSchemaWriteForwarderOptions();
+ }
+
+ ///
+ public async Task ForwardPutAsync(CultNetShardDescriptor shard, CultNetDocumentPutRawMessage message)
+ {
+ if (message == null) throw new ArgumentNullException(nameof(message));
+ var endpoint = ResolvePrimaryEndpoint(shard);
+ message.ShardId ??= shard.ShardId;
+ message.ShardEpoch ??= shard.Epoch;
+ await SendAsync(endpoint, message).ConfigureAwait(false);
+ }
+
+ ///
+ public async Task ForwardDeleteAsync(CultNetShardDescriptor shard, CultNetDocumentDeleteMessage message)
+ {
+ if (message == null) throw new ArgumentNullException(nameof(message));
+ var endpoint = ResolvePrimaryEndpoint(shard);
+ message.ShardId ??= shard.ShardId;
+ message.ShardEpoch ??= shard.Epoch;
+ await SendAsync(endpoint, message).ConfigureAwait(false);
+ }
+
+ ///
+ /// Parses a CultNet endpoint URI into host and port.
+ ///
+ public static (string Host, int Port) ParseEndpoint(string endpoint)
+ {
+ if (string.IsNullOrWhiteSpace(endpoint))
+ {
+ throw new ArgumentException("Endpoint must be non-empty.", nameof(endpoint));
+ }
+
+ if (!Uri.TryCreate(endpoint, UriKind.Absolute, out var uri) ||
+ !string.Equals(uri.Scheme, "cultnet", StringComparison.OrdinalIgnoreCase) ||
+ string.IsNullOrWhiteSpace(uri.Host))
+ {
+ throw new FormatException($"CultNet endpoint '{endpoint}' must use cultnet://host:port.");
+ }
+
+ return (uri.Host, uri.IsDefaultPort ? 3075 : uri.Port);
+ }
+
+ private async Task SendAsync(string endpoint, T message) where T : ICultNetSchemaMessage
+ {
+ var (host, port) = ParseEndpoint(endpoint);
+ using var client = new Client(_options.Security ?? ClientSecurityOptions.Development())
+ {
+ AllowUnverifiedCultNetMessages = true
+ };
+ _options.ConfigureClient?.Invoke(client);
+ client.Connect(host, port);
+ await WaitForConnectionAsync(client, endpoint).ConfigureAwait(false);
+ client.SendCultNet(message);
+ }
+
+ 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 static string ResolvePrimaryEndpoint(CultNetShardDescriptor shard)
+ {
+ if (shard == null) throw new ArgumentNullException(nameof(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 b6122498..cfb30b9d 100644
--- a/src/GameCult.Networking/README.md
+++ b/src/GameCult.Networking/README.md
@@ -134,8 +134,9 @@ servers through `CultNetDatabaseServer`. It handles primary-shard writes, typed
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`; a concrete schema-v0 endpoint dialer is the next
-missing organ.
+`ICultNetShardWriteForwarder`, with `CultNetSchemaWriteForwarder` as the
+schema-v0 `cultnet://host:port` endpoint dialer. Per-shard replica logs are 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 a02a1ab4..108cf461 100644
--- a/tests/GameCult.Networking.Tests/NetworkingTests.cs
+++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs
@@ -603,6 +603,32 @@ namespace GameCult.Networking.Tests
Assert.That(forwarder.LastPutMessage.ShardEpoch, Is.EqualTo(9));
}
+ [Test]
+ public void CultNetSchemaWriteForwarder_Parses_CultNetEndpoints()
+ {
+ var parsed = CultNetSchemaWriteForwarder.ParseEndpoint("cultnet://primary.example.test:4075");
+
+ Assert.That(parsed.Host, Is.EqualTo("primary.example.test"));
+ Assert.That(parsed.Port, Is.EqualTo(4075));
+ }
+
+ [Test]
+ public void CultNetSchemaWriteForwarder_Uses_DefaultPort()
+ {
+ var parsed = CultNetSchemaWriteForwarder.ParseEndpoint("cultnet://primary.example.test");
+
+ Assert.That(parsed.Host, Is.EqualTo("primary.example.test"));
+ Assert.That(parsed.Port, Is.EqualTo(3075));
+ }
+
+ [Test]
+ public void CultNetSchemaWriteForwarder_Rejects_InvalidEndpoint()
+ {
+ Assert.That(
+ () => CultNetSchemaWriteForwarder.ParseEndpoint("http://primary.example.test:3075"),
+ Throws.TypeOf());
+ }
+
[Test]
public void CultNetDatabaseServer_Creates_Filtered_SubscriptionChange()
{