From 4af500a87d14994dae64e888aed88a5f2879381d Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 19 May 2026 17:34:49 +0100 Subject: [PATCH] Add CultNet shard catalog routing --- .../cultnet-distributed-database/summary.md | 18 ++- .../Contracts/cultnet-distributed-database.md | 23 +-- src/GameCult.Networking/CultNetDatabase.cs | 107 +++++++++++++- .../CultNetDatabaseServer.cs | 39 +++++ .../CultNetSchemaMessageSerialization.cs | 2 + .../CultNetSchemaMessages.cs | 136 ++++++++++++++++++ src/GameCult.Networking/README.md | 6 +- .../NetworkingTests.cs | 129 +++++++++++++++++ 8 files changed, 440 insertions(+), 20 deletions(-) diff --git a/research/cultnet-distributed-database/summary.md b/research/cultnet-distributed-database/summary.md index a88c4af6..5c85b739 100644 --- a/research/cultnet-distributed-database/summary.md +++ b/research/cultnet-distributed-database/summary.md @@ -152,9 +152,9 @@ The coherent path is: details. - Conflict policy must be declared before conflicting writes are accepted. -## Next Cut +## Current Cut -Build shard catalog exchange: +Shard catalog exchange now exists: - `cultnet.shard_catalog_request.v0` - `cultnet.shard_catalog_response.v0` @@ -162,5 +162,15 @@ Build shard catalog exchange: - stale epoch rejection on writes - routing error that tells clients where the primary lives when known -That is the next foundation needed before forwarding, membership, Raft-style -replication, or CRDT policy work can be honest. +## Next Cut + +Build write forwarding: + +- non-primary nodes forward raw put/delete messages to primary endpoints when + policy allows +- forwarded writes preserve shard id and epoch +- stale epochs still fail +- clients can opt into direct retry using routing hints + +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 ea4fbc2a..93fa4e17 100644 --- a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md +++ b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md @@ -39,10 +39,14 @@ Implemented: - `CultNetHost.Database` and `CultNetHost.DatabaseServer` - subscription request/cancel/change messages - server-side live change fanout for schema/key-filtered subscriptions +- shard catalog request/response messages +- shard descriptors with owner runtime id, epoch, schema filters, key prefix, + primary endpoints, replica endpoints, read replica endpoints, and region +- epoch-aware raw put/delete messages +- routing hints on shard authority errors Not implemented yet: -- shard catalog exchange - membership/failure detection - leader election or automatic shard failover - replicated log @@ -62,13 +66,14 @@ Not implemented yet: ## Next Coherent Slice -Add shard catalog exchange: +Add write forwarding: -- advertise shard ids, schema filters, key ranges/prefixes, owner runtime ids, - and epochs -- let clients ask any node where a write belongs -- reject stale-epoch writes explicitly -- prepare forwarding without making forwarding mandatory +- 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 -That gives the subscription surface a real routing model before membership, -leader election, or replicated logs enter the room and start charging rent. +Forwarding is the next useful layer now that shard catalogs exist. 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 bc1eeed6..bf03285a 100644 --- a/src/GameCult.Networking/CultNetDatabase.cs +++ b/src/GameCult.Networking/CultNetDatabase.cs @@ -48,7 +48,11 @@ namespace GameCult.Networking long epoch, bool isPrimary, IEnumerable? schemaIds = null, - string? keyPrefix = null) + string? keyPrefix = null, + IEnumerable? primaryEndpoints = null, + IEnumerable? replicaEndpoints = null, + IEnumerable? readReplicaEndpoints = null, + string? region = null) { ShardId = RequireNonEmpty(shardId, nameof(shardId)); OwnerRuntimeId = RequireNonEmpty(ownerRuntimeId, nameof(ownerRuntimeId)); @@ -57,6 +61,10 @@ namespace GameCult.Networking SchemaIds = schemaIds?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray() ?? Array.Empty(); KeyPrefix = keyPrefix; + PrimaryEndpoints = CleanEndpoints(primaryEndpoints); + ReplicaEndpoints = CleanEndpoints(replicaEndpoints); + ReadReplicaEndpoints = CleanEndpoints(readReplicaEndpoints); + Region = region; } /// @@ -83,6 +91,22 @@ namespace GameCult.Networking /// Gets the optional record-key prefix governed by this shard. /// public string? KeyPrefix { get; } + /// + /// Gets endpoints that can accept authoritative writes for this shard. + /// + public IReadOnlyList PrimaryEndpoints { get; } + /// + /// Gets endpoints that replicate authoritative shard mutations. + /// + public IReadOnlyList ReplicaEndpoints { get; } + /// + /// Gets endpoints intended for low-latency read and subscription traffic. + /// + public IReadOnlyList ReadReplicaEndpoints { get; } + /// + /// Gets an optional locality label for this shard owner. + /// + public string? Region { get; } /// /// Creates a primary shard that accepts every schema and key. @@ -122,6 +146,12 @@ namespace GameCult.Networking return value; } + + private static string[] CleanEndpoints(IEnumerable? endpoints) + { + return endpoints?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray() + ?? Array.Empty(); + } } /// @@ -257,6 +287,35 @@ namespace GameCult.Networking /// public IReadOnlyList Shards => _shards; + /// + /// Creates a shard catalog response for optional schema/key filters. + /// + public CultNetShardCatalogResponseMessage CreateShardCatalogResponse(CultNetShardCatalogRequestMessage request) + { + ThrowIfDisposed(); + if (request == null) throw new ArgumentNullException(nameof(request)); + + var schemaIds = request.SchemaIds == null || request.SchemaIds.Length == 0 + ? null + : request.SchemaIds; + var recordKeys = request.RecordKeys == null || request.RecordKeys.Length == 0 + ? null + : request.RecordKeys.Select(key => new CultRecordKey(key)).ToArray(); + + var descriptors = _shards + .Where(shard => MatchesCatalogFilter(shard, schemaIds, recordKeys)) + .Select(ToMessage) + .ToArray(); + + return new CultNetShardCatalogResponseMessage + { + MessageId = string.IsNullOrWhiteSpace(request.MessageId) + ? Guid.NewGuid().ToString("N") + : request.MessageId, + Shards = descriptors + }; + } + /// /// Gets a document by key. /// @@ -330,7 +389,7 @@ namespace GameCult.Networking var key = new CultRecordKey(message.Document.RecordKey); var shard = ResolveShard(message.Document.SchemaId, key); - EnsurePrimary(shard, 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); var document = await _documents.ApplyRawDocumentPutMessageAsync(_cache, message).ConfigureAwait(false); @@ -355,7 +414,7 @@ namespace GameCult.Networking var key = new CultRecordKey(message.RecordKey); var shard = ResolveShard(message.SchemaId, key); - EnsurePrimary(shard, message.SchemaId, key); + EnsurePrimary(shard, message.SchemaId, key, message.ShardEpoch); var descriptor = _cache.Registry.GetRequiredBySchemaId(message.SchemaId); var previous = _cache.Get(key); if (previous == null) @@ -537,8 +596,15 @@ namespace GameCult.Networking return _shards.FirstOrDefault(shard => shard.Matches(schemaId, key)) ?? _shards[0]; } - private static void EnsurePrimary(CultNetShardDescriptor shard, string schemaId, CultRecordKey key) + private static void EnsurePrimary(CultNetShardDescriptor shard, string schemaId, CultRecordKey key, long? expectedEpoch = null) { + if (expectedEpoch.HasValue && expectedEpoch.Value != shard.Epoch) + { + throw new CultNetShardAuthorityException( + shard, + $"Shard '{shard.ShardId}' is at epoch {shard.Epoch}, not requested epoch {expectedEpoch.Value}."); + } + if (shard.IsPrimary) { return; @@ -549,6 +615,39 @@ namespace GameCult.Networking $"Shard '{shard.ShardId}' owned by '{shard.OwnerRuntimeId}' does not accept local writes for schema '{schemaId}' key '{key.Value}'."); } + private static bool MatchesCatalogFilter( + CultNetShardDescriptor shard, + IReadOnlyList? schemaIds, + IReadOnlyList? recordKeys) + { + var schemaMatches = schemaIds == null || + schemaIds.Count == 0 || + shard.SchemaIds.Count == 0 || + schemaIds.Any(schemaId => shard.SchemaIds.Contains(schemaId, StringComparer.Ordinal)); + var keyMatches = recordKeys == null || + recordKeys.Count == 0 || + recordKeys.Any(key => string.IsNullOrEmpty(shard.KeyPrefix) || + key.Value.StartsWith(shard.KeyPrefix!, StringComparison.Ordinal)); + return schemaMatches && keyMatches; + } + + internal static CultNetShardDescriptorMessage ToMessage(CultNetShardDescriptor shard) + { + return new CultNetShardDescriptorMessage + { + ShardId = shard.ShardId, + OwnerRuntimeId = shard.OwnerRuntimeId, + Epoch = shard.Epoch, + IsPrimary = shard.IsPrimary, + SchemaIds = shard.SchemaIds.ToArray(), + KeyPrefix = shard.KeyPrefix, + PrimaryEndpoints = shard.PrimaryEndpoints.ToArray(), + ReplicaEndpoints = shard.ReplicaEndpoints.ToArray(), + ReadReplicaEndpoints = shard.ReadReplicaEndpoints.ToArray(), + Region = shard.Region + }; + } + private void ThrowIfDisposed() { if (_disposed) diff --git a/src/GameCult.Networking/CultNetDatabaseServer.cs b/src/GameCult.Networking/CultNetDatabaseServer.cs index b46613c2..589376a7 100644 --- a/src/GameCult.Networking/CultNetDatabaseServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseServer.cs @@ -19,6 +19,7 @@ namespace GameCult.Networking private readonly Func _snapshotHandler; private readonly Func _putHandler; private readonly Func _deleteHandler; + private readonly Func _shardCatalogHandler; private readonly Func _subscribeHandler; private readonly Func _unsubscribeHandler; private readonly ConcurrentDictionary _subscriptions = new(StringComparer.Ordinal); @@ -34,12 +35,14 @@ namespace GameCult.Networking _snapshotHandler = HandleSnapshotRequestAsync; _putHandler = HandlePutAsync; _deleteHandler = HandleDeleteAsync; + _shardCatalogHandler = HandleShardCatalogRequestAsync; _subscribeHandler = HandleSubscribeAsync; _unsubscribeHandler = HandleUnsubscribeAsync; _server.OnCultNet(_snapshotHandler); _server.OnCultNet(_putHandler); _server.OnCultNet(_deleteHandler); + _server.OnCultNet(_shardCatalogHandler); _server.OnCultNet(_subscribeHandler); _server.OnCultNet(_unsubscribeHandler); } @@ -77,6 +80,14 @@ namespace GameCult.Networking request); } + /// + /// Creates a shard catalog response from the database shard map. + /// + public CultNetShardCatalogResponseMessage CreateShardCatalogResponse(CultNetShardCatalogRequestMessage request) + { + return _database.CreateShardCatalogResponse(request); + } + /// /// Detaches handlers from the server. /// @@ -91,6 +102,7 @@ namespace GameCult.Networking _server.RemoveCultNetMessageListener(_snapshotHandler); _server.RemoveCultNetMessageListener(_putHandler); _server.RemoveCultNetMessageListener(_deleteHandler); + _server.RemoveCultNetMessageListener(_shardCatalogHandler); _server.RemoveCultNetMessageListener(_subscribeHandler); _server.RemoveCultNetMessageListener(_unsubscribeHandler); foreach (var subscription in _subscriptions.Values) @@ -107,12 +119,22 @@ namespace GameCult.Networking return Task.CompletedTask; } + private Task HandleShardCatalogRequestAsync(CultNetShardCatalogRequestMessage request, NetPeer peer) + { + peer.SendCultNet(CreateShardCatalogResponse(request)); + return Task.CompletedTask; + } + private async Task HandlePutAsync(CultNetDocumentPutRawMessage message, NetPeer peer) { try { await ApplyPutAsync(message).ConfigureAwait(false); } + catch (CultNetShardAuthorityException ex) + { + peer.SendCultNet(CreateRoutingError(ex)); + } catch (Exception ex) { peer.SendCultNet(new CultNetErrorMessage { Error = ex.Message }); @@ -125,6 +147,10 @@ namespace GameCult.Networking { await ApplyDeleteAsync(message).ConfigureAwait(false); } + catch (CultNetShardAuthorityException ex) + { + peer.SendCultNet(CreateRoutingError(ex)); + } catch (Exception ex) { peer.SendCultNet(new CultNetErrorMessage { Error = ex.Message }); @@ -260,5 +286,18 @@ namespace GameCult.Networking { return $"{peer.Id}:{subscriptionId}"; } + + private static CultNetErrorMessage CreateRoutingError(CultNetShardAuthorityException exception) + { + return new CultNetErrorMessage + { + Error = exception.Message, + RoutingHint = new CultNetShardRoutingHint + { + Reason = exception.Shard.IsPrimary ? "stale_epoch" : "not_primary", + Shard = CultNetDatabase.ToMessage(exception.Shard) + } + }; + } } } diff --git a/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs b/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs index 6abfe32b..2cda2d20 100644 --- a/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs +++ b/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs @@ -62,6 +62,8 @@ namespace GameCult.Networking CultNetSchemaVersions.DatabaseSubscribe => Deserialize(payload), CultNetSchemaVersions.DatabaseUnsubscribe => Deserialize(payload), CultNetSchemaVersions.DatabaseChangeRaw => Deserialize(payload), + CultNetSchemaVersions.ShardCatalogRequest => Deserialize(payload), + CultNetSchemaVersions.ShardCatalogResponse => Deserialize(payload), CultNetSchemaVersions.SchemaCatalogRequest => Deserialize(payload), CultNetSchemaVersions.SchemaCatalogResponse => Deserialize(payload), _ => throw new MessagePackSerializationException( diff --git a/src/GameCult.Networking/CultNetSchemaMessages.cs b/src/GameCult.Networking/CultNetSchemaMessages.cs index 21ca81db..0c579547 100644 --- a/src/GameCult.Networking/CultNetSchemaMessages.cs +++ b/src/GameCult.Networking/CultNetSchemaMessages.cs @@ -91,6 +91,14 @@ namespace GameCult.Networking /// database subscription change contract identifier. /// public const string DatabaseChangeRaw = "cultnet.database_change_raw.v0"; + /// + /// shard catalog request contract identifier. + /// + public const string ShardCatalogRequest = "cultnet.shard_catalog_request.v0"; + /// + /// shard catalog response contract identifier. + /// + public const string ShardCatalogResponse = "cultnet.shard_catalog_response.v0"; } /// @@ -321,6 +329,26 @@ namespace GameCult.Networking /// Gets or sets the error. /// [Key("error")] public string Error { get; set; } = string.Empty; + /// + /// Gets or sets optional shard routing details for authority failures. + /// + [Key("routingHint")] public CultNetShardRoutingHint? RoutingHint { get; set; } + } + + /// + /// Describes where a shard write should be routed. + /// + [MessagePackObject] + public class CultNetShardRoutingHint + { + /// + /// Gets or sets the shard descriptor. + /// + [Key("shard")] public CultNetShardDescriptorMessage? Shard { get; set; } + /// + /// Gets or sets a machine-readable routing reason. + /// + [Key("reason")] public string Reason { get; set; } = string.Empty; } /// @@ -421,6 +449,14 @@ namespace GameCult.Networking /// Gets or sets the record key. /// [Key("recordKey")] public string RecordKey { get; set; } = string.Empty; + /// + /// Gets or sets the shard id the sender targeted. + /// + [Key("shardId")] public string? ShardId { get; set; } + /// + /// Gets or sets the shard epoch the sender targeted. + /// + [Key("shardEpoch")] public long? ShardEpoch { get; set; } } /// @@ -441,6 +477,14 @@ namespace GameCult.Networking /// Gets or sets the document. /// [Key("document")] public CultNetRawDocumentRecord Document { get; set; } = new CultNetRawDocumentRecord(); + /// + /// Gets or sets the shard id the sender targeted. + /// + [Key("shardId")] public string? ShardId { get; set; } + /// + /// Gets or sets the shard epoch the sender targeted. + /// + [Key("shardEpoch")] public long? ShardEpoch { get; set; } } /// @@ -575,6 +619,98 @@ namespace GameCult.Networking [Key("schemaId")] public string? SchemaId { get; set; } } + /// + /// Describes one shard advertised by a CultNet database node. + /// + [MessagePackObject] + public class CultNetShardDescriptorMessage + { + /// + /// Gets or sets the shard id. + /// + [Key("shardId")] public string ShardId { get; set; } = string.Empty; + /// + /// Gets or sets the owner runtime id. + /// + [Key("ownerRuntimeId")] public string OwnerRuntimeId { get; set; } = string.Empty; + /// + /// Gets or sets the shard authority epoch. + /// + [Key("epoch")] public long Epoch { get; set; } + /// + /// Gets or sets whether the responding node is primary for this shard. + /// + [Key("isPrimary")] public bool IsPrimary { get; set; } + /// + /// Gets or sets schema ids governed by this shard. Empty means all schemas. + /// + [Key("schemaIds")] public string[] SchemaIds { get; set; } = Array.Empty(); + /// + /// Gets or sets the record-key prefix governed by this shard. + /// + [Key("keyPrefix")] public string? KeyPrefix { get; set; } + /// + /// Gets or sets endpoints that accept authoritative writes. + /// + [Key("primaryEndpoints")] public string[] PrimaryEndpoints { get; set; } = Array.Empty(); + /// + /// Gets or sets authoritative replica endpoints. + /// + [Key("replicaEndpoints")] public string[] ReplicaEndpoints { get; set; } = Array.Empty(); + /// + /// Gets or sets low-latency read replica endpoints. + /// + [Key("readReplicaEndpoints")] public string[] ReadReplicaEndpoints { get; set; } = Array.Empty(); + /// + /// Gets or sets an optional locality label. + /// + [Key("region")] public string? Region { get; set; } + } + + /// + /// Requests shard ownership and routing metadata. + /// + [MessagePackObject] + public class CultNetShardCatalogRequestMessage : ICultNetSchemaMessage + { + /// + /// Gets or sets the schema version. + /// + [Key("schemaVersion")] public string SchemaVersion { get; set; } = CultNetSchemaVersions.ShardCatalogRequest; + /// + /// Gets or sets the request id. + /// + [Key("messageId")] public string MessageId { get; set; } = string.Empty; + /// + /// Gets or sets optional schema filters. + /// + [Key("schemaIds")] public string[]? SchemaIds { get; set; } + /// + /// Gets or sets optional record-key filters. + /// + [Key("recordKeys")] public string[]? RecordKeys { get; set; } + } + + /// + /// Returns shard ownership and routing metadata. + /// + [MessagePackObject] + public class CultNetShardCatalogResponseMessage : ICultNetSchemaMessage + { + /// + /// Gets or sets the schema version. + /// + [Key("schemaVersion")] public string SchemaVersion { get; set; } = CultNetSchemaVersions.ShardCatalogResponse; + /// + /// Gets or sets the request id. + /// + [Key("messageId")] public string MessageId { get; set; } = string.Empty; + /// + /// Gets or sets shard descriptors. + /// + [Key("shards")] public CultNetShardDescriptorMessage[] Shards { get; set; } = Array.Empty(); + } + /// /// Describes one schema exposed through the CultNet schema catalog. /// diff --git a/src/GameCult.Networking/README.md b/src/GameCult.Networking/README.md index 4b8c6870..49fd7242 100644 --- a/src/GameCult.Networking/README.md +++ b/src/GameCult.Networking/README.md @@ -131,9 +131,9 @@ CultNet becomes the distribution model: 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, and -schema/key-filtered live subscriptions. Shard catalog exchange is the next -missing organ. +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. 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 971bfd51..fbe813e2 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -241,6 +241,41 @@ namespace GameCult.Networking.Tests Assert.That(roundTrip.Document!.RecordKey, Is.EqualTo("record-1")); } + [Test] + public void CultNetSchemaMessageSerialization_RoundTrips_ShardCatalogResponse() + { + var message = new CultNetShardCatalogResponseMessage + { + MessageId = "shards-1", + Shards = + [ + new CultNetShardDescriptorMessage + { + ShardId = "players-eu", + OwnerRuntimeId = "runtime-a", + Epoch = 12, + IsPrimary = true, + SchemaIds = ["schema-player"], + KeyPrefix = "player:", + PrimaryEndpoints = ["cultnet://runtime-a:3075"], + ReplicaEndpoints = ["cultnet://runtime-b:3075"], + ReadReplicaEndpoints = ["cultnet://edge-1:3075"], + Region = "eu-west" + } + ] + }; + + var payload = CultNetSchemaMessageSerialization.Serialize(message); + var roundTrip = (CultNetShardCatalogResponseMessage)CultNetSchemaMessageSerialization.Deserialize(payload); + + Assert.That(roundTrip.MessageId, Is.EqualTo("shards-1")); + Assert.That(roundTrip.Shards, Has.Length.EqualTo(1)); + Assert.That(roundTrip.Shards[0].ShardId, Is.EqualTo("players-eu")); + Assert.That(roundTrip.Shards[0].Epoch, Is.EqualTo(12)); + Assert.That(roundTrip.Shards[0].PrimaryEndpoints, Is.EqualTo(["cultnet://runtime-a:3075"])); + Assert.That(roundTrip.Shards[0].Region, Is.EqualTo("eu-west")); + } + [Test] public async Task CultNetDocumentRegistry_RawSnapshotReplication_PreservesPayloadBytes() { @@ -342,6 +377,53 @@ namespace GameCult.Networking.Tests .EqualTo("remote-players")); } + [Test] + public async Task CultNetDatabase_Rejects_RawPut_With_StaleShardEpoch() + { + 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", + "runtime-a", + epoch: 7, + isPrimary: true, + schemaIds: [schemaId], + keyPrefix: "player:", + primaryEndpoints: ["cultnet://runtime-a:3075"]) + ] + }); + var key = new CultRecordKey("player:stale"); + var message = registry.CreateRawDocumentPutMessage( + "put-stale", + new CultRecordHandle(key), + new PlayerData + { + PlayerId = Guid.NewGuid(), + Email = "stale@example.test", + PasswordHash = "hash", + Username = "Stale" + }); + message.ShardId = "players"; + message.ShardEpoch = 6; + + Assert.That( + async () => await database.ApplyPutAsync(message), + Throws.TypeOf() + .With.Property(nameof(CultNetShardAuthorityException.Shard)) + .Property(nameof(CultNetShardDescriptor.Epoch)) + .EqualTo(7)); + } + [Test] public async Task CultNetDatabase_ApplyRawPut_Publishes_DomainChange() { @@ -416,6 +498,53 @@ namespace GameCult.Networking.Tests Assert.That(response.Documents[0].Payload, Is.EqualTo(SerializePlayerDataPayload(player))); } + [Test] + public void CultNetDatabaseServer_Creates_Filtered_ShardCatalogResponse() + { + var cache = new CultCache(); + var schemaId = cache.Registry.GetRequired().SchemaId; + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions + { + Shards = + [ + new CultNetShardDescriptor( + "players", + "runtime-a", + epoch: 3, + isPrimary: true, + schemaIds: [schemaId], + keyPrefix: "player:", + primaryEndpoints: ["cultnet://runtime-a:3075"], + replicaEndpoints: ["cultnet://runtime-b:3075"], + readReplicaEndpoints: ["cultnet://edge-a:3075"], + region: "eu-west"), + new CultNetShardDescriptor( + "world", + "runtime-c", + epoch: 1, + isPrimary: false, + schemaIds: ["schema-world"], + keyPrefix: "world:") + ] + }); + using var server = new Server(cache, DevelopmentServerSecurity); + using var databaseServer = new CultNetDatabaseServer(server, database); + + var response = databaseServer.CreateShardCatalogResponse(new CultNetShardCatalogRequestMessage + { + MessageId = "catalog-players", + SchemaIds = [schemaId], + RecordKeys = ["player:one"] + }); + + Assert.That(response.MessageId, Is.EqualTo("catalog-players")); + Assert.That(response.Shards, Has.Length.EqualTo(1)); + Assert.That(response.Shards[0].ShardId, Is.EqualTo("players")); + Assert.That(response.Shards[0].Epoch, Is.EqualTo(3)); + Assert.That(response.Shards[0].PrimaryEndpoints, Is.EqualTo(["cultnet://runtime-a:3075"])); + Assert.That(response.Shards[0].ReadReplicaEndpoints, Is.EqualTo(["cultnet://edge-a:3075"])); + } + [Test] public void CultNetDatabaseServer_Creates_Filtered_SubscriptionChange() {