Add CultNet schema write forwarder

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.

src/GameCult.Networking/Client.cs

@@ -87,6 +87,11 @@ namespace GameCult.Networking
         /// </summary>
         public bool LogSensitivePayloads { get; set; }
 
+        /// <summary>
+        /// Gets or sets whether CultNet schema-v0 messages may be sent before user session verification.
+        /// </summary>
+        public bool AllowUnverifiedCultNetMessages { get; set; }
+
         /// <summary>
         /// Gets a value indicating whether the client currently has a verified session.
         /// </summary>
@@ -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);

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.

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
         /// </summary>
         public ICultNetShardWriteForwarder? WriteForwarder { get; set; }
     }
+
+    /// <summary>
+    /// Options for the schema-v0 client-based shard write forwarder.
+    /// </summary>
+    public sealed class CultNetSchemaWriteForwarderOptions
+    {
+        /// <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 the forwarded write.
+        /// </summary>
+        public TimeSpan ConnectTimeout { get; set; } = TimeSpan.FromSeconds(5);
+
+        /// <summary>
+        /// Gets or sets a callback used to customize each ephemeral forwarding client.
+        /// </summary>
+        public Action<Client>? ConfigureClient { get; set; }
+    }
+
+    /// <summary>
+    /// Forwards shard writes over the CultNet schema-v0 client transport.
+    /// </summary>
+    public sealed class CultNetSchemaWriteForwarder : ICultNetShardWriteForwarder
+    {
+        private readonly CultNetSchemaWriteForwarderOptions _options;
+
+        /// <summary>
+        /// Creates a schema-v0 write forwarder.
+        /// </summary>
+        public CultNetSchemaWriteForwarder(CultNetSchemaWriteForwarderOptions? options = null)
+        {
+            _options = options ?? new CultNetSchemaWriteForwarderOptions();
+        }
+
+        /// <inheritdoc />
+        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);
+        }
+
+        /// <inheritdoc />
+        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);
+        }
+
+        /// <summary>
+        /// Parses a CultNet endpoint URI into host and port.
+        /// </summary>
+        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<T>(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.");
+        }
+    }
 }

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

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<FormatException>());
+        }
+
         [Test]
         public void CultNetDatabaseServer_Creates_Filtered_SubscriptionChange()
         {