diff --git a/src/GameCult.Networking/CultNetDatabase.cs b/src/GameCult.Networking/CultNetDatabase.cs index ef37f17a..13600a04 100644 --- a/src/GameCult.Networking/CultNetDatabase.cs +++ b/src/GameCult.Networking/CultNetDatabase.cs @@ -638,9 +638,7 @@ namespace GameCult.Networking ThrowIfDisposed(); if (shard == null) throw new ArgumentNullException(nameof(shard)); - var requestedSchemaIds = filter?.SchemaIds != null - ? new HashSet(filter.SchemaIds, StringComparer.Ordinal) - : null; + var requestedSchemaIds = filter?.SchemaIds; var requestedRecordKeys = filter?.RecordKeys != null ? new HashSet(filter.RecordKeys, StringComparer.Ordinal) : null; @@ -652,7 +650,7 @@ namespace GameCult.Networking var key = GetTrackedKey(document, documentType); if (string.IsNullOrWhiteSpace(key.Value) || !shard.Matches(descriptor.SchemaId, key) || - (requestedSchemaIds != null && !MatchesRequestedSchema(descriptor, requestedSchemaIds)) || + (requestedSchemaIds != null && !CultNetSchemaAliasMatching.MatchesAny(requestedSchemaIds, descriptor)) || (requestedRecordKeys != null && !requestedRecordKeys.Contains(key.Value))) { continue; @@ -1692,7 +1690,7 @@ namespace GameCult.Networking { try { - if (MatchesRequestedSchema(_cache.Registry.GetRequiredBySchemaId(shardSchemaId), requestedSchemaIds)) + if (CultNetSchemaAliasMatching.MatchesAny(schemaIds, _cache.Registry.GetRequiredBySchemaId(shardSchemaId))) return true; } catch (InvalidOperationException) @@ -1704,37 +1702,6 @@ namespace GameCult.Networking return false; } - private static bool MatchesRequestedSchema( - CultDocumentDescriptor descriptor, - ISet requestedSchemaIds) - { - if (requestedSchemaIds.Count == 0) - return true; - - if (requestedSchemaIds.Contains(descriptor.SchemaId)) - return true; - - if (descriptor.ToCatalogEntry().CompatibleSchemaIds.Any(requestedSchemaIds.Contains)) - return true; - - return requestedSchemaIds - .Select(InferSchemaName) - .Where(schemaName => !string.IsNullOrWhiteSpace(schemaName)) - .Any(schemaName => string.Equals(schemaName, descriptor.SchemaName, StringComparison.Ordinal)); - } - - private static string? InferSchemaName(string schemaVersion) - { - var marker = schemaVersion.LastIndexOf(".v", StringComparison.Ordinal); - if (marker <= 0 || marker + 2 >= schemaVersion.Length) - return null; - - var version = schemaVersion.Substring(marker + 2); - return version.All(char.IsDigit) - ? schemaVersion.Substring(0, marker) - : null; - } - internal static CultNetShardDescriptorMessage ToMessage(CultNetShardDescriptor shard) { return new CultNetShardDescriptorMessage diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs index 04b4a0dd..7d5378ef 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs @@ -13,8 +13,9 @@ namespace GameCult.Networking { /// /// Publishes live database changes through any schema-v0 server transport. - /// Optional record projection runs after authorization and owns both initial - /// snapshot and live-update delivery for a peer. + /// Selection match runs after authorization and owns both initial + /// snapshot and live-update delivery for a peer. Projection (header vs. + /// document) is a value on the request's selection, not a delegate. /// public sealed class CultNetDatabaseSubscriptionServer : IDisposable { @@ -24,7 +25,6 @@ namespace GameCult.Networking private readonly Func _unsubscribe; private readonly Func? _authorizeRequest; private readonly Func? _authorizeRecord; - private readonly Func? _projectRecord; private readonly ICultNetSchemaServerPeerLifecycle? _peerLifecycle; private readonly ConcurrentDictionary _subscriptions = new ConcurrentDictionary(SubscriptionKeyComparer.Instance); @@ -40,14 +40,12 @@ namespace GameCult.Networking ICultNetSchemaServer server, CultNetDatabase database, Func? authorizeRequest = null, - Func? authorizeRecord = null, - Func? projectRecord = null) + Func? authorizeRecord = null) { _server = server ?? throw new ArgumentNullException(nameof(server)); _database = database ?? throw new ArgumentNullException(nameof(database)); _authorizeRequest = authorizeRequest; _authorizeRecord = authorizeRecord; - _projectRecord = projectRecord; _subscribe = HandleSubscribeAsync; _unsubscribe = HandleUnsubscribeAsync; _server.OnCultNet(_subscribe); @@ -235,13 +233,10 @@ namespace GameCult.Networking var sourceRecordKey = ResolveChangeRecordKey(change); var matched = CreateMatchedRecord(change, request, peer); - var projected = matched; - if (projected != null && _projectRecord != null) - projected = _projectRecord(request, peer, projected); ApplyProjectedChange( projection, sourceRecordKey, - projected, + matched, peer, subscriptionId); } @@ -305,18 +300,16 @@ namespace GameCult.Networking RecordKeys = request.RecordKeys }); var bySourceRecordKey = new Dictionary(StringComparer.Ordinal); - var projectedRecordKeys = new HashSet(StringComparer.Ordinal); + var matchedRecordKeys = new HashSet(StringComparer.Ordinal); foreach (var source in snapshot.Documents) { var sourceRecordKey = source.RecordKey; if (_authorizeRecord?.Invoke(request, peer, sourceRecordKey, source.SchemaId) == false) continue; - var projected = _projectRecord == null ? source : _projectRecord(request, peer, source); - if (projected == null) continue; - if (!projectedRecordKeys.Add(projected.RecordKey)) + if (!matchedRecordKeys.Add(source.RecordKey)) throw new InvalidOperationException( - $"Database subscription projection produced duplicate record key '{projected.RecordKey}'."); - bySourceRecordKey[sourceRecordKey] = projected; + $"Database subscription snapshot produced duplicate record key '{source.RecordKey}'."); + bySourceRecordKey[sourceRecordKey] = source; } return new ProjectedSnapshot(bySourceRecordKey); } diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index ae04a9ee..d420eb6d 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -1597,12 +1597,7 @@ namespace GameCult.Networking.Tests server, database, authorizeRequest: (request, _) => request.ConsumerRuntimeId == "allowed-runtime", - authorizeRecord: (_, _, recordKey, _) => recordKey.StartsWith("tests:public:", StringComparison.Ordinal), - projectRecord: (request, _, record) => - { - record.SourceRuntimeId = "projection:" + request.ConsumerRuntimeId; - return record; - }); + authorizeRecord: (_, _, recordKey, _) => recordKey.StartsWith("tests:public:", StringComparison.Ordinal)); using var cancellation = new CancellationTokenSource(); var serverThread = new Thread(() => { @@ -1645,7 +1640,6 @@ namespace GameCult.Networking.Tests var snapshot = await AwaitWithTimeout(subscribed.Task, TimeSpan.FromSeconds(2)); Assert.That(snapshot.Documents.Select(document => document.RecordKey), Is.EqualTo(new[] { "tests:public:initial" })); - Assert.That(snapshot.Documents.Single().SourceRuntimeId, Is.EqualTo("projection:allowed-runtime")); await database.PutAsync(new CultRecordKey("tests:private:live"), new NetworkSchemaNote { @@ -1662,19 +1656,19 @@ namespace GameCult.Networking.Tests cancellation.Cancel(); Assert.That(update.Document, Is.Not.Null); Assert.That(update.Document!.RecordKey, Is.EqualTo("tests:public:live")); - Assert.That(update.Document.SourceRuntimeId, Is.EqualTo("projection:allowed-runtime")); } [Test] public async Task DatabaseSubscriptionServer_ReconcilesDeliveredProjectionAndBodyDemandWhenAuthorityChanges() { - const string sourceRecordKey = "tests:authority:source"; + const string schemaId = "tests.networking_note"; + const string recordKeyOne = "tests:authority:one"; + const string recordKeyTwo = "tests:authority:two"; const string bodyId = "tests:authority:body"; var authorized = true; - var projectionName = "one"; var sourceCache = new CultCache(); var sourceDatabase = new CultNetDatabase(sourceCache); - await sourceDatabase.PutAsync(new CultRecordKey(sourceRecordKey), new NetworkSchemaNote + await sourceDatabase.PutAsync(new CultRecordKey(recordKeyOne), new NetworkSchemaNote { Schema = "tests.networking_note.v1", Text = "initial" @@ -1688,12 +1682,7 @@ namespace GameCult.Networking.Tests server, sourceDatabase, authorizeRequest: (_, _) => authorized, - authorizeRecord: (_, _, _, _) => authorized, - projectRecord: (_, _, record) => - { - record.RecordKey = $"tests:authority:projected:{projectionName}"; - return record; - }); + authorizeRecord: (_, _, _, _) => authorized); using var bodyDemand = new CultMeshBodyDemandTracker(subscriptions); using var cancellation = new CancellationTokenSource(); var serverThread = new Thread(() => @@ -1718,33 +1707,42 @@ namespace GameCult.Networking.Tests await AwaitWithTimeout( client.SubscribeAsync( "authority", - recordKeys: [sourceRecordKey], + schemaIds: [schemaId], consumerRuntimeId: "authority-client", bodyIds: [bodyId], supportedBodyTransports: [CultMeshBodyTransportKind.SharedMemory.ToString()]), TimeSpan.FromSeconds(2)); - Assert.That(targetCache.Get(new CultRecordKey("tests:authority:projected:one")), Is.Not.Null); + Assert.That(targetCache.Get(new CultRecordKey(recordKeyOne)), Is.Not.Null); Assert.That(bodyDemand.Plan(bodyId).HasConsumers, Is.True); - projectionName = "two"; + // Drive the record's identity change through the selection itself + // (a real rename: the old key's row is deleted, a new key's row is + // put), not through a projection delegate, then force the full + // reconcile diff that D6 gives the subscription server. + await sourceDatabase.PutAsync(new CultRecordKey(recordKeyTwo), new NetworkSchemaNote + { + Schema = "tests.networking_note.v1", + Text = "renamed" + }); + await sourceDatabase.DeleteAsync(new CultRecordKey(recordKeyOne)); subscriptions.Reconcile(); await WaitUntilAsync( - () => targetCache.Get(new CultRecordKey("tests:authority:projected:one")) == null && - targetCache.Get(new CultRecordKey("tests:authority:projected:two")) != null, + () => targetCache.Get(new CultRecordKey(recordKeyOne)) == null && + targetCache.Get(new CultRecordKey(recordKeyTwo)) != null, TimeSpan.FromSeconds(2)); Assert.That(changes.Any(change => change.ChangeKind == "removed" && - change.RecordKey == "tests:authority:projected:one"), Is.True); + change.RecordKey == recordKeyOne), Is.True); Assert.That(changes.Any(change => change.ChangeKind == "added" && - change.RecordKey == "tests:authority:projected:two"), Is.True); + change.RecordKey == recordKeyTwo), Is.True); authorized = false; subscriptions.Reconcile(); await WaitUntilAsync( - () => targetCache.Get(new CultRecordKey("tests:authority:projected:two")) == null, + () => targetCache.Get(new CultRecordKey(recordKeyTwo)) == null, TimeSpan.FromSeconds(2)); Assert.That(bodyDemand.Plan(bodyId).HasConsumers, Is.False); var changeCountAfterRevocation = changes.Count; - await sourceDatabase.PutAsync(new CultRecordKey(sourceRecordKey), new NetworkSchemaNote + await sourceDatabase.PutAsync(new CultRecordKey(recordKeyTwo), new NetworkSchemaNote { Schema = "tests.networking_note.v1", Text = "must-not-escape" @@ -1753,7 +1751,7 @@ namespace GameCult.Networking.Tests cancellation.Cancel(); Assert.That(changes.Count, Is.EqualTo(changeCountAfterRevocation)); - Assert.That(targetCache.Get(new CultRecordKey("tests:authority:projected:two")), Is.Null); + Assert.That(targetCache.Get(new CultRecordKey(recordKeyTwo)), Is.Null); } [Test]