From 1f527ae99b05160b041799cd12c875bcff7bdcad Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 19 May 2026 17:40:19 +0100 Subject: [PATCH] Add CultNet shard write forwarding policy --- .../cultnet-distributed-database/summary.md | 10 +-- .../Contracts/cultnet-distributed-database.md | 19 ++--- src/GameCult.Networking/CultNetDatabase.cs | 34 +++++--- .../CultNetDatabaseServer.cs | 59 +++++++++++++- src/GameCult.Networking/CultNetLocal.cs | 7 +- .../CultNetShardWriteForwarding.cs | 36 +++++++++ src/GameCult.Networking/README.md | 4 +- .../NetworkingTests.cs | 78 +++++++++++++++++++ 8 files changed, 218 insertions(+), 29 deletions(-) create mode 100644 src/GameCult.Networking/CultNetShardWriteForwarding.cs diff --git a/research/cultnet-distributed-database/summary.md b/research/cultnet-distributed-database/summary.md index 5c85b739..b0dcd4e0 100644 --- a/research/cultnet-distributed-database/summary.md +++ b/research/cultnet-distributed-database/summary.md @@ -161,16 +161,16 @@ Shard catalog exchange now exists: - per-shard id, owner runtime id, epoch, schema ids, key prefix/range - stale epoch rejection on writes - routing error that tells clients where the primary lives when known +- injectable non-primary write forwarding policy ## Next Cut -Build write forwarding: +Build a concrete schema-v0 write forwarder: -- non-primary nodes forward raw put/delete messages to primary endpoints when - policy allows +- dial the selected primary endpoint from the shard catalog - forwarded writes preserve shard id and epoch -- stale epochs still fail -- clients can opt into direct retry using routing hints +- stale epochs still fail before forwarding +- failed forwarding returns endpoint-aware routing errors That is the next foundation needed before membership, Raft-style replication, 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 93fa4e17..807ea2c1 100644 --- a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md +++ b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md @@ -44,12 +44,14 @@ Implemented: primary endpoints, replica endpoints, read replica endpoints, and region - epoch-aware raw put/delete messages - routing hints on shard authority errors +- injectable non-primary write forwarding policy 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 @@ -66,14 +68,13 @@ Not implemented yet: ## Next Coherent Slice -Add write forwarding: +Add a concrete endpoint dialer for write forwarding: -- non-primary nodes can forward raw put/delete messages to the advertised - primary endpoint when policy allows it -- forwarded writes must preserve the original shard id and epoch -- stale-epoch writes still fail instead of being "helpfully" retried against - unknown authority -- clients that opt out of forwarding receive routing hints and retry directly +- 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 -Forwarding is the next useful layer now that shard catalogs exist. Membership, -leader election, and replicated logs should wait until routing is boring. +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. diff --git a/src/GameCult.Networking/CultNetDatabase.cs b/src/GameCult.Networking/CultNetDatabase.cs index bf03285a..f4bbdd52 100644 --- a/src/GameCult.Networking/CultNetDatabase.cs +++ b/src/GameCult.Networking/CultNetDatabase.cs @@ -237,15 +237,20 @@ namespace GameCult.Networking /// /// Creates a shard authority exception. /// - public CultNetShardAuthorityException(CultNetShardDescriptor shard, string message) : base(message) + public CultNetShardAuthorityException(CultNetShardDescriptor shard, string message, string reason = "not_primary") : base(message) { Shard = shard; + Reason = reason; } /// /// Gets the shard that rejected the write. /// public CultNetShardDescriptor Shard { get; } + /// + /// Gets the machine-readable authority rejection reason. + /// + public string Reason { get; } } /// @@ -316,6 +321,15 @@ namespace GameCult.Networking }; } + /// + /// Resolves the shard that governs the supplied schema and key. + /// + public CultNetShardDescriptor ResolveShard(string schemaId, CultRecordKey key) + { + ThrowIfDisposed(); + return ResolveShardInternal(schemaId, key); + } + /// /// Gets a document by key. /// @@ -334,7 +348,7 @@ namespace GameCult.Networking if (document == null) throw new ArgumentNullException(nameof(document)); var descriptor = _cache.Registry.GetRequired(); - var shard = ResolveShard(descriptor.SchemaId, key); + var shard = ResolveShardInternal(descriptor.SchemaId, key); EnsurePrimary(shard, descriptor.SchemaId, key); var previous = _cache.Get(key); @@ -356,7 +370,7 @@ namespace GameCult.Networking { ThrowIfDisposed(); var descriptor = _cache.Registry.GetRequired(); - var shard = ResolveShard(descriptor.SchemaId, key); + var shard = ResolveShardInternal(descriptor.SchemaId, key); EnsurePrimary(shard, descriptor.SchemaId, key); var previous = _cache.Get(key); @@ -388,7 +402,7 @@ namespace GameCult.Networking } var key = new CultRecordKey(message.Document.RecordKey); - var shard = ResolveShard(message.Document.SchemaId, key); + var shard = ResolveShardInternal(message.Document.SchemaId, key); EnsurePrimary(shard, message.Document.SchemaId, key, message.ShardEpoch); var descriptor = _cache.Registry.GetRequiredBySchemaId(message.Document.SchemaId); var previous = _cache.Get(key); @@ -413,7 +427,7 @@ namespace GameCult.Networking if (message == null) throw new ArgumentNullException(nameof(message)); var key = new CultRecordKey(message.RecordKey); - var shard = ResolveShard(message.SchemaId, key); + var shard = ResolveShardInternal(message.SchemaId, key); EnsurePrimary(shard, message.SchemaId, key, message.ShardEpoch); var descriptor = _cache.Registry.GetRequiredBySchemaId(message.SchemaId); var previous = _cache.Get(key); @@ -526,7 +540,7 @@ namespace GameCult.Networking var key = current != null ? GetTrackedKey(current, documentType) : new CultRecordKey(string.Empty); - var shard = ResolveShard(descriptor.SchemaId, key); + var shard = ResolveShardInternal(descriptor.SchemaId, key); var changeType = typeof(CultNetDatabaseChange<>).MakeGenericType(documentType); var change = Activator.CreateInstance( changeType, @@ -591,7 +605,7 @@ namespace GameCult.Networking } } - private CultNetShardDescriptor ResolveShard(string schemaId, CultRecordKey key) + private CultNetShardDescriptor ResolveShardInternal(string schemaId, CultRecordKey key) { return _shards.FirstOrDefault(shard => shard.Matches(schemaId, key)) ?? _shards[0]; } @@ -602,7 +616,8 @@ namespace GameCult.Networking { throw new CultNetShardAuthorityException( shard, - $"Shard '{shard.ShardId}' is at epoch {shard.Epoch}, not requested epoch {expectedEpoch.Value}."); + $"Shard '{shard.ShardId}' is at epoch {shard.Epoch}, not requested epoch {expectedEpoch.Value}.", + "stale_epoch"); } if (shard.IsPrimary) @@ -612,7 +627,8 @@ namespace GameCult.Networking throw new CultNetShardAuthorityException( shard, - $"Shard '{shard.ShardId}' owned by '{shard.OwnerRuntimeId}' does not accept local writes for schema '{schemaId}' key '{key.Value}'."); + $"Shard '{shard.ShardId}' owned by '{shard.OwnerRuntimeId}' does not accept local writes for schema '{schemaId}' key '{key.Value}'.", + "not_primary"); } private static bool MatchesCatalogFilter( diff --git a/src/GameCult.Networking/CultNetDatabaseServer.cs b/src/GameCult.Networking/CultNetDatabaseServer.cs index 589376a7..a8f3b598 100644 --- a/src/GameCult.Networking/CultNetDatabaseServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseServer.cs @@ -16,6 +16,7 @@ namespace GameCult.Networking { private readonly Server _server; private readonly CultNetDatabase _database; + private readonly CultNetDatabaseServerOptions _options; private readonly Func _snapshotHandler; private readonly Func _putHandler; private readonly Func _deleteHandler; @@ -28,10 +29,14 @@ namespace GameCult.Networking /// /// Creates and attaches a database message bridge to a server. /// - public CultNetDatabaseServer(Server server, CultNetDatabase database) + public CultNetDatabaseServer( + Server server, + CultNetDatabase database, + CultNetDatabaseServerOptions? options = null) { _server = server ?? throw new ArgumentNullException(nameof(server)); _database = database ?? throw new ArgumentNullException(nameof(database)); + _options = options ?? new CultNetDatabaseServerOptions(); _snapshotHandler = HandleSnapshotRequestAsync; _putHandler = HandlePutAsync; _deleteHandler = HandleDeleteAsync; @@ -133,7 +138,10 @@ namespace GameCult.Networking } catch (CultNetShardAuthorityException ex) { - peer.SendCultNet(CreateRoutingError(ex)); + if (!await TryForwardPutAsync(ex, message).ConfigureAwait(false)) + { + peer.SendCultNet(CreateRoutingError(ex)); + } } catch (Exception ex) { @@ -149,7 +157,10 @@ namespace GameCult.Networking } catch (CultNetShardAuthorityException ex) { - peer.SendCultNet(CreateRoutingError(ex)); + if (!await TryForwardDeleteAsync(ex, message).ConfigureAwait(false)) + { + peer.SendCultNet(CreateRoutingError(ex)); + } } catch (Exception ex) { @@ -294,10 +305,50 @@ namespace GameCult.Networking Error = exception.Message, RoutingHint = new CultNetShardRoutingHint { - Reason = exception.Shard.IsPrimary ? "stale_epoch" : "not_primary", + Reason = exception.Reason, Shard = CultNetDatabase.ToMessage(exception.Shard) } }; } + + /// + /// Attempts to forward a rejected raw put to the shard owner. + /// + public async Task TryForwardPutAsync( + CultNetShardAuthorityException exception, + CultNetDocumentPutRawMessage message) + { + if (!_options.ForwardNonPrimaryWrites || + _options.WriteForwarder == null || + exception.Reason != "not_primary") + { + return false; + } + + message.ShardId ??= exception.Shard.ShardId; + message.ShardEpoch ??= exception.Shard.Epoch; + await _options.WriteForwarder.ForwardPutAsync(exception.Shard, message).ConfigureAwait(false); + return true; + } + + /// + /// Attempts to forward a rejected raw delete to the shard owner. + /// + public async Task TryForwardDeleteAsync( + CultNetShardAuthorityException exception, + CultNetDocumentDeleteMessage message) + { + if (!_options.ForwardNonPrimaryWrites || + _options.WriteForwarder == null || + exception.Reason != "not_primary") + { + return false; + } + + message.ShardId ??= exception.Shard.ShardId; + message.ShardEpoch ??= exception.Shard.Epoch; + await _options.WriteForwarder.ForwardDeleteAsync(exception.Shard, message).ConfigureAwait(false); + return true; + } } } diff --git a/src/GameCult.Networking/CultNetLocal.cs b/src/GameCult.Networking/CultNetLocal.cs index 1227aa6f..a134cfcf 100644 --- a/src/GameCult.Networking/CultNetLocal.cs +++ b/src/GameCult.Networking/CultNetLocal.cs @@ -31,6 +31,11 @@ namespace GameCult.Networking /// public CultNetDatabaseOptions? DatabaseOptions { get; set; } + /// + /// Gets or sets the database server bridge options used for shard routing and forwarding. + /// + public CultNetDatabaseServerOptions? DatabaseServerOptions { get; set; } + /// /// Gets or sets an optional callback used to customize the server before start. /// @@ -130,7 +135,7 @@ namespace GameCult.Networking var server = new Server(cache, options.Security ?? ServerSecurityOptions.Development()); var databaseOptions = options.DatabaseOptions ?? new CultNetDatabaseOptions(); var database = new CultNetDatabase(cache, databaseOptions); - var databaseServer = new CultNetDatabaseServer(server, database); + var databaseServer = new CultNetDatabaseServer(server, database, options.DatabaseServerOptions); options.ConfigureServer?.Invoke(server); var host = new CultNetHost(cache, store, server, database, databaseServer); diff --git a/src/GameCult.Networking/CultNetShardWriteForwarding.cs b/src/GameCult.Networking/CultNetShardWriteForwarding.cs new file mode 100644 index 00000000..97c68676 --- /dev/null +++ b/src/GameCult.Networking/CultNetShardWriteForwarding.cs @@ -0,0 +1,36 @@ +using System.Threading.Tasks; + +namespace GameCult.Networking +{ + /// + /// Forwards shard writes to an authoritative owner. + /// + public interface ICultNetShardWriteForwarder + { + /// + /// Forwards a raw document put to the shard owner. + /// + Task ForwardPutAsync(CultNetShardDescriptor shard, CultNetDocumentPutRawMessage message); + + /// + /// Forwards a raw document delete to the shard owner. + /// + Task ForwardDeleteAsync(CultNetShardDescriptor shard, CultNetDocumentDeleteMessage message); + } + + /// + /// Options controlling shard write forwarding from a database server bridge. + /// + public sealed class CultNetDatabaseServerOptions + { + /// + /// Gets or sets whether non-primary writes may be forwarded to shard owners. + /// + public bool ForwardNonPrimaryWrites { get; set; } + + /// + /// Gets or sets the forwarder used when non-primary write forwarding is enabled. + /// + public ICultNetShardWriteForwarder? WriteForwarder { get; set; } + } +} diff --git a/src/GameCult.Networking/README.md b/src/GameCult.Networking/README.md index 49fd7242..b6122498 100644 --- a/src/GameCult.Networking/README.md +++ b/src/GameCult.Networking/README.md @@ -133,7 +133,9 @@ The current first slice exposes this as `CultNetDatabase` and wires hosted 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. Write forwarding is the next missing organ. +write rejection. Non-primary write forwarding is available behind an injectable +`ICultNetShardWriteForwarder`; a concrete schema-v0 endpoint dialer is 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 fbe813e2..a02a1ab4 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -545,6 +545,64 @@ namespace GameCult.Networking.Tests Assert.That(response.Shards[0].ReadReplicaEndpoints, Is.EqualTo(["cultnet://edge-a:3075"])); } + [Test] + public async Task CultNetDatabaseServer_Forwards_NonPrimaryWrites_WhenConfigured() + { + 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 database = new CultNetDatabase(cache, new CultNetDatabaseOptions + { + DocumentRegistry = registry, + Shards = + [ + new CultNetShardDescriptor( + "players-remote", + "runtime-owner", + epoch: 9, + isPrimary: false, + schemaIds: [schemaId], + keyPrefix: "player:", + primaryEndpoints: ["cultnet://runtime-owner:3075"]) + ] + }); + var forwarder = new CapturingShardWriteForwarder(); + using var server = new Server(cache, DevelopmentServerSecurity); + using var databaseServer = new CultNetDatabaseServer( + server, + database, + new CultNetDatabaseServerOptions + { + ForwardNonPrimaryWrites = true, + WriteForwarder = forwarder + }); + var key = new CultRecordKey("player:forward"); + var message = registry.CreateRawDocumentPutMessage( + "put-forward", + new CultRecordHandle(key), + new PlayerData + { + PlayerId = Guid.NewGuid(), + Email = "forward@example.test", + PasswordHash = "hash", + Username = "Forward" + }); + var exception = Assert.ThrowsAsync( + async () => await database.ApplyPutAsync(message)); + + var forwarded = await databaseServer.TryForwardPutAsync(exception!, message); + + Assert.That(forwarded, Is.True); + Assert.That(forwarder.PutCount, Is.EqualTo(1)); + Assert.That(forwarder.LastPutShard!.ShardId, Is.EqualTo("players-remote")); + Assert.That(forwarder.LastPutMessage!.ShardId, Is.EqualTo("players-remote")); + Assert.That(forwarder.LastPutMessage.ShardEpoch, Is.EqualTo(9)); + } + [Test] public void CultNetDatabaseServer_Creates_Filtered_SubscriptionChange() { @@ -754,6 +812,26 @@ namespace GameCult.Networking.Tests } } + private sealed class CapturingShardWriteForwarder : ICultNetShardWriteForwarder + { + public int PutCount { get; private set; } + public CultNetShardDescriptor? LastPutShard { get; private set; } + public CultNetDocumentPutRawMessage? LastPutMessage { get; private set; } + + public Task ForwardPutAsync(CultNetShardDescriptor shard, CultNetDocumentPutRawMessage message) + { + PutCount++; + LastPutShard = shard; + LastPutMessage = message; + return Task.CompletedTask; + } + + public Task ForwardDeleteAsync(CultNetShardDescriptor shard, CultNetDocumentDeleteMessage message) + { + return Task.CompletedTask; + } + } + [MessagePack.MessagePackObject] public sealed class PlayerDataPayload {