GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

CultNet selection cut1, commit 2d: the CultMesh collapse

Browse source 
Collapses the rest of the Mesh selector surface named in docs/cultnet-selection-cut.md
section 4/11 onto CultNetSelection:

- CultMeshSnapshotRequestOptions becomes a record carrying Selection in place of
  SchemaIds/RecordKeys; its three hand-written clone functions
  (CloneSnapshotRequestOptions x2, CloneSnapshotFacadeRequestOptions) are gone in
  favor of `with`.
- CultMeshPeerSnapshotDocumentOptions becomes a record for the same reason;
  WithPublicationBindingOptions and ToSnapshotRequestOptions collapse to `with` and
  one inlined Selection literal at their single call site.
- CleanSnapshotFilter and the three-times "no schema filter when keys are given"
  rule collapse into two owners: CultMesh.ResolveDefaultSelection (the typed-document
  default) and CultMesh.OverlaySelection (explicit override over defaults).
- CultMeshHotBodySubscription carries one Selection instead of parallel
  RecordKeys/SchemaIds allowlists.
- ReadDocumentFromSnapshotResponse's four-tier fallback collapses to one
  recordKey-anchored read: exact schema/alias match, or (foreign/runtime-generated
  schema id) a payload that decodes as TDocument. The two tiers that matched by
  schema alone or recordKey alone - the ones that could return a record under the
  wrong key - are gone; DocumentHandle_ReadsRemotePeerSnapshotsAsTypedDocuments
  pins the surviving recordKey-anchored decode fallback.

Two items in section 11's table are not cut here, by finding rather than oversight:

- IsSameCultDocumentSchema (CultMesh.cs) already delegates to the one
  CultNetSchemaAliasMatching owner (no second engine); deleting the wrapper would
  duplicate its type-to-descriptor resolution at four call sites instead of
  collapsing it. Left in place.
- The selector literals in CultMeshClient.cs (574-604, 830-860, 932-962) call
  CultNetDatabaseSubscriptionClient.SubscribeAsync, which stays a v0
  (schemaIds/recordKeys) API in this cut - that client is not named in section 7's
  per-file plan. There is no Selection to move these onto without adding a v1
  subscribe path to Networking, which is out of this commit's scope. Left in place.

Tests: GameCult.Mesh.Tests 252/253 (1 pre-existing unrelated skip), 0 failed.
GameCult.Networking.Tests 161/161 excluding one pre-existing flaky test
(DatabaseSubscriptionServer_ReconcilesDeliveredProjectionAndBodyDemandWhenAuthorityChanges,
a background RUDP poll thread racing a disposed socket in code this commit does not
touch; it crashes the test host the same way on this branch before this commit too).
GameCult.Caching.Tests 188/188.

Net line count, src/GameCult.Mesh only: +143/-226, net -83. Section 11 estimated
about -326 for this cut's slice. The gap is mostly the two items above (about -49
of the estimate) plus one already-ruled keep this map's Progress block already
named (the foreign-schema-id decode fallback, -41 in the table, kept per the prior
Hands' ruling and untouched here) and the four-tier fallback surviving as a
two-tier one rather than a zero-fallback exact read (an honest test-enforced
requirement, not a padding choice).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-22 03:38:56 +02:00
commit 5ba48f0ebe
5 changed files with 149 additions and 232 deletions
Download patch file Download diff file Expand all files Collapse all files

144
src/GameCult.Mesh/CultMesh.cs
View file

@ -112,7 +112,7 @@ namespace GameCult.Mesh
/// <summary>
/// Options for opening one typed document from a remote CultNet snapshot endpoint.
/// </summary>
public sealed class CultMeshPeerSnapshotDocumentOptions
public sealed record CultMeshPeerSnapshotDocumentOptions
{
/// <summary>Gets or sets the semantic document id. Defaults to the record key.</summary>
public string? DocumentId { get; set; }
@ -135,8 +135,8 @@ namespace GameCult.Mesh
/// <summary>Gets or sets the message id prefix used for snapshot requests.</summary>
public string? MessageIdPrefix { get; set; }
/// <summary>Gets or sets schema ids to request. Empty means the document type schema is requested.</summary>
public IReadOnlyList<string>? SchemaIds { get; set; }
/// <summary>Gets or sets the selection to request. Absent means the document type's own schema.</summary>
public CultNetSelection? Selection { get; set; }
/// <summary>Gets or sets the target shard id, when the endpoint is shard-aware.</summary>
public string? ShardId { get; set; }
@ -1944,28 +1944,12 @@ namespace GameCult.Mesh
string? documentId,
string? sourceId)
{
return new CultMeshPeerSnapshotDocumentOptions
{
DocumentId = string.IsNullOrWhiteSpace(documentId) ? key.Value : documentId,
SourceId = string.IsNullOrWhiteSpace(sourceId)
? (string.IsNullOrWhiteSpace(documentId) ? key.Value : documentId)
: sourceId,
RouteHint = options?.RouteHint,
ResponseTimeout = options?.ResponseTimeout ?? TimeSpan.FromSeconds(5),
ConnectTimeout = options?.ConnectTimeout ?? TimeSpan.FromSeconds(5),
PollInterval = options?.PollInterval ?? TimeSpan.FromMilliseconds(250),
MessageIdPrefix = options?.MessageIdPrefix,
SchemaIds = options?.SchemaIds,
ShardId = options?.ShardId,
ShardEpoch = options?.ShardEpoch,
Security = options?.Security,
ConfigureClient = options?.ConfigureClient,
RudpRuntimeId = options?.RudpRuntimeId,
RudpConnectionId = options?.RudpConnectionId ?? 0x43554c54,
RudpConnectPayload = options?.RudpConnectPayload ?? "cultnet-schema-rudp",
RudpMaxFragmentBytes = options?.RudpMaxFragmentBytes ?? 1024,
RudpResendDelayMs = options?.RudpResendDelayMs ?? 25
};
var resolved = options is null ? new CultMeshPeerSnapshotDocumentOptions() : options with { };
resolved.DocumentId = string.IsNullOrWhiteSpace(documentId) ? key.Value : documentId;
resolved.SourceId = string.IsNullOrWhiteSpace(sourceId)
? (string.IsNullOrWhiteSpace(documentId) ? key.Value : documentId)
: sourceId;
return resolved;
}
/// <summary>
@ -2707,46 +2691,42 @@ namespace GameCult.Mesh
where TDocument : class
{
var descriptor = CultDocumentRegistry.Shared.GetRequired<TDocument>();
var selection = new CultNetSelection
{
Schemas = options.Selection?.Schemas ?? new[] { descriptor.SchemaId },
Keys = new[] { recordKey }
};
return await FetchSnapshotAsync(
createClient,
endpoint,
ToSnapshotRequestOptions(
options,
context,
options.SchemaIds ?? new[] { descriptor.SchemaId },
new[] { recordKey }))
new CultMeshSnapshotRequestOptions
{
Selection = selection,
ShardId = options.ShardId,
ShardEpoch = options.ShardEpoch,
ResponseTimeout = options.ResponseTimeout,
ConnectTimeout = options.ConnectTimeout,
MessageIdPrefix = string.IsNullOrWhiteSpace(options.MessageIdPrefix)
? $"cultmesh:{context.RuntimeId}:snapshot"
: options.MessageIdPrefix,
Security = options.Security,
ConfigureClient = options.ConfigureClient,
RudpRuntimeId = string.IsNullOrWhiteSpace(options.RudpRuntimeId)
? context.RuntimeId
: options.RudpRuntimeId,
RudpConnectionId = options.RudpConnectionId,
RudpConnectPayload = options.RudpConnectPayload,
RudpMaxFragmentBytes = options.RudpMaxFragmentBytes,
RudpResendDelayMs = options.RudpResendDelayMs
})
.ConfigureAwait(false);
}
private static CultMeshSnapshotRequestOptions ToSnapshotRequestOptions(
CultMeshPeerSnapshotDocumentOptions options,
CultMeshQueryContext context,
IReadOnlyList<string>? schemaIds,
IReadOnlyList<string>? recordKeys)
{
return new CultMeshSnapshotRequestOptions
{
SchemaIds = schemaIds,
RecordKeys = recordKeys,
ShardId = options.ShardId,
ShardEpoch = options.ShardEpoch,
ResponseTimeout = options.ResponseTimeout,
ConnectTimeout = options.ConnectTimeout,
MessageIdPrefix = string.IsNullOrWhiteSpace(options.MessageIdPrefix)
? $"cultmesh:{context.RuntimeId}:snapshot"
: options.MessageIdPrefix,
Security = options.Security,
ConfigureClient = options.ConfigureClient,
RudpRuntimeId = string.IsNullOrWhiteSpace(options.RudpRuntimeId)
? context.RuntimeId
: options.RudpRuntimeId,
RudpConnectionId = options.RudpConnectionId,
RudpConnectPayload = options.RudpConnectPayload,
RudpMaxFragmentBytes = options.RudpMaxFragmentBytes,
RudpResendDelayMs = options.RudpResendDelayMs
};
}
// One exact read (docs/cultnet-selection-cut.md, section 4/7): the record this recordKey
// names, decoded as TDocument when its schema is this schema, a declared alias of it, or
// (a foreign/runtime-generated schema id this runtime does not recognize by any alias) its
// payload decodes as TDocument. No re-filter of the server's answer by schema alone, and no
// fallback that returns a record under the wrong key.
private static TDocument ReadDocumentFromSnapshotResponse<TDocument>(
CultNetSnapshotResponseRawMessage response,
string schemaId,
@ -2754,16 +2734,21 @@ namespace GameCult.Mesh
where TDocument : class
{
if (response == null) throw new ArgumentNullException(nameof(response));
var record = response.Documents.FirstOrDefault(candidate =>
string.Equals(candidate.SchemaId, schemaId, StringComparison.Ordinal) &&
string.Equals(candidate.RecordKey, recordKey, StringComparison.Ordinal))
?? response.Documents.FirstOrDefault(candidate =>
string.Equals(candidate.RecordKey, recordKey, StringComparison.Ordinal) &&
TryDecodeSnapshotDocument(candidate, out TDocument? _))
?? response.Documents.FirstOrDefault(candidate =>
string.Equals(candidate.SchemaId, schemaId, StringComparison.Ordinal))
?? response.Documents.FirstOrDefault(candidate =>
string.Equals(candidate.RecordKey, recordKey, StringComparison.Ordinal));
var descriptor = CultDocumentRegistry.Shared.GetRequired<TDocument>();
CultNetRawDocumentRecord? record = null;
foreach (var candidate in response.Documents)
{
if (!string.Equals(candidate.RecordKey, recordKey, StringComparison.Ordinal))
continue;
if (string.Equals(candidate.SchemaId, schemaId, StringComparison.Ordinal) ||
CultNetSchemaAliasMatching.Matches(candidate.SchemaId, descriptor) ||
TryDecodeAsMessagePack<TDocument>(candidate, out _))
{
record = candidate;
break;
}
}
if (record == null)
{
throw new InvalidOperationException(
@ -2776,21 +2761,18 @@ namespace GameCult.Mesh
$"CultNet raw document payloadEncoding must be \"messagepack\", not \"{record.PayloadEncoding}\".");
}
return DecodeSnapshotDocument<TDocument>(record);
return (TDocument)CultDocumentMessagePackSerialization.DeserializeUntyped(typeof(TDocument), record.Payload);
}
private static bool TryDecodeSnapshotDocument<TDocument>(
CultNetRawDocumentRecord record,
out TDocument? document)
private static bool TryDecodeAsMessagePack<TDocument>(CultNetRawDocumentRecord record, out TDocument? document)
where TDocument : class
{
document = null;
if (!string.Equals(record.PayloadEncoding, "messagepack", StringComparison.Ordinal))
return false;
try
{
document = DecodeSnapshotDocument<TDocument>(record);
document = (TDocument)CultDocumentMessagePackSerialization.DeserializeUntyped(typeof(TDocument), record.Payload);
return true;
}
catch
@ -2799,14 +2781,6 @@ namespace GameCult.Mesh
}
}
private static TDocument DecodeSnapshotDocument<TDocument>(CultNetRawDocumentRecord record)
where TDocument : class
{
return (TDocument)CultDocumentMessagePackSerialization.DeserializeUntyped(
typeof(TDocument),
record.Payload);
}
/// <summary>
/// Parses a rudp://host:port endpoint into its host/port parts.
/// </summary>
@ -2849,8 +2823,8 @@ namespace GameCult.Mesh
throw new ArgumentException("At least one readable body transport is required.", nameof(supportedBodyTransports));
return subscriptions.SubscribeAsync(
subscription.SubscriptionId,
recordKeys: subscription.RecordKeys,
schemaIds: subscription.SchemaIds,
recordKeys: subscription.Selection.Keys,
schemaIds: subscription.Selection.Schemas,
consumerRuntimeId: subscription.ConsumerRuntimeId,
bodyIds: new[] { subscription.BodyId },
supportedBodyTransports: transports.Select(value => value.ToString()),

44
src/GameCult.Mesh/CultMeshBodyDemand.cs
View file

@ -188,31 +188,35 @@ namespace GameCult.Mesh
SubscriptionId = Require(subscriptionId, nameof(subscriptionId));
ConsumerRuntimeId = Require(consumerRuntimeId, nameof(consumerRuntimeId));
BodyId = Require(bodyId, nameof(bodyId));
RecordKeys = new[]
{
Require(viewRecordKey, nameof(viewRecordKey)),
CultMeshBodyPublicationDocument.CreateLatestRecordKey(bodyId).Value
}
.Concat(additionalRecordKeys ?? Array.Empty<string>())
.Where(value => !string.IsNullOrWhiteSpace(value))
.Distinct(StringComparer.Ordinal)
.ToArray();
SchemaIds = new[]
{
Require(viewSchemaId, nameof(viewSchemaId)),
CultMeshBodyPublicationSchemaVersions.Publication
}
.Concat(additionalSchemaIds ?? Array.Empty<string>())
.Where(value => !string.IsNullOrWhiteSpace(value))
.Distinct(StringComparer.Ordinal)
.ToArray();
Selection = new CultNetSelection
{
Keys = new[]
{
Require(viewRecordKey, nameof(viewRecordKey)),
CultMeshBodyPublicationDocument.CreateLatestRecordKey(bodyId).Value
}
.Concat(additionalRecordKeys ?? Array.Empty<string>())
.Where(value => !string.IsNullOrWhiteSpace(value))
.Distinct(StringComparer.Ordinal)
.ToArray(),
Schemas = new[]
{
Require(viewSchemaId, nameof(viewSchemaId)),
CultMeshBodyPublicationSchemaVersions.Publication
}
.Concat(additionalSchemaIds ?? Array.Empty<string>())
.Where(value => !string.IsNullOrWhiteSpace(value))
.Distinct(StringComparer.Ordinal)
.ToArray()
};
}
public string SubscriptionId { get; }
public string ConsumerRuntimeId { get; }
public string BodyId { get; }
public IReadOnlyList<string> RecordKeys { get; }
public IReadOnlyList<string> SchemaIds { get; }
/// <summary>The view record, the body's latest-publication record, and any additional keys/schemas, as one selection.</summary>
public CultNetSelection Selection { get; }
private static string Require(string value, string parameterName) =>
string.IsNullOrWhiteSpace(value)

181
src/GameCult.Mesh/CultMeshSnapshots.cs
View file

@ -13,13 +13,10 @@ namespace GameCult.Mesh
/// <summary>
/// Options for one scoped CultNet snapshot request.
/// </summary>
public sealed class CultMeshSnapshotRequestOptions
public sealed record CultMeshSnapshotRequestOptions
{
/// <summary>Gets or sets schema ids to request. Empty means no schema filter.</summary>
public IReadOnlyList<string>? SchemaIds { get; set; }
/// <summary>Gets or sets record keys to request. Empty means no record-key filter.</summary>
public IReadOnlyList<string>? RecordKeys { get; set; }
/// <summary>Gets or sets the selection to request. Absent selects every schema and key.</summary>
public CultNetSelection? Selection { get; set; }
/// <summary>Gets or sets the target shard id, when the endpoint is shard-aware.</summary>
public string? ShardId { get; set; }
@ -87,7 +84,7 @@ namespace GameCult.Mesh
_endpoint = string.IsNullOrWhiteSpace(endpoint)
? throw new ArgumentException("Value must be non-empty.", nameof(endpoint))
: endpoint;
_defaults = CultMesh.CloneSnapshotRequestOptions(options);
_defaults = options with { };
_registry = registry ?? new CultNetDocumentRegistry();
_client = CultMesh.CreateSnapshotClient(endpoint, _defaults)();
_client.OnCultNet<CultNetSnapshotResponseRawMessage>(OnResponse);
@ -103,7 +100,7 @@ namespace GameCult.Mesh
{
_sharedSession = session ?? throw new ArgumentNullException(nameof(session));
_endpoint = session.State.Path?.Endpoint ?? session.Target.AuthorityRuntimeId;
_defaults = CultMesh.CloneSnapshotRequestOptions(options);
_defaults = options with { };
_registry = registry ?? new CultNetDocumentRegistry();
_client = session.Channel;
_responseSubscription = session.OnCultNet<CultNetSnapshotResponseRawMessage>(OnResponse);
@ -150,9 +147,8 @@ namespace GameCult.Mesh
_defaults.ConnectTimeout,
(_client as ICultNetSchemaClientHealth)?.BackgroundFailure)
.ConfigureAwait(false);
var options = CultMesh.CloneSnapshotRequestOptions(_defaults);
options.SchemaIds = schemaIds ?? options.SchemaIds;
options.RecordKeys = recordKeys ?? options.RecordKeys;
var selection = CultMesh.OverlaySelection(_defaults.Selection, schemaIds, recordKeys);
var options = _defaults with { Selection = selection };
var messageId = CultMesh.CreateSnapshotMessageId(options);
var completion = new TaskCompletionSource<CultNetSnapshotResponseRawMessage>(
TaskCreationOptions.RunContinuationsAsynchronously);
@ -164,8 +160,8 @@ namespace GameCult.Mesh
_client.SendCultNet(new CultNetSnapshotRequestMessage
{
MessageId = messageId,
SchemaIds = CultMesh.CleanSnapshotFilter(options.SchemaIds),
RecordKeys = CultMesh.CleanSnapshotFilter(options.RecordKeys),
SchemaIds = selection.Schemas,
RecordKeys = selection.Keys,
ShardId = string.IsNullOrWhiteSpace(options.ShardId) ? null : options.ShardId,
ShardEpoch = options.ShardEpoch
});
@ -195,8 +191,8 @@ namespace GameCult.Mesh
where TDocument : class
{
var descriptor = CultDocumentRegistry.Shared.GetRequired<TDocument>();
var resolvedSchemas = schemaIds ?? (recordKeys is { Count: > 0 } ? null : new[] { descriptor.SchemaId });
var snapshot = await FetchSnapshotAsync(resolvedSchemas, recordKeys).ConfigureAwait(false);
var selection = CultMesh.ResolveDefaultSelection(schemaIds, recordKeys, descriptor);
var snapshot = await FetchSnapshotAsync(selection.Schemas, selection.Keys).ConfigureAwait(false);
return CultMesh.DecodeSnapshotDocuments<TDocument>(snapshot, _registry);
}
@ -397,7 +393,7 @@ namespace GameCult.Mesh
var resolvedOptions = options ?? new CultMeshSnapshotEndpointOptions();
Context = resolvedOptions.Context ?? CultMesh.Verse("remote", "cultmesh-snapshot-client").Context;
DocumentRegistry = resolvedOptions.DocumentRegistry ?? new CultNetDocumentRegistry();
Request = CloneSnapshotRequestOptions(resolvedOptions.Request);
Request = resolvedOptions.Request is null ? new CultMeshSnapshotRequestOptions() : resolvedOptions.Request with { };
RouteHint = resolvedOptions.RouteHint ?? new CultMeshRouteHint(CultMeshLocalityKind.Network, Endpoint);
SourceId = string.IsNullOrWhiteSpace(resolvedOptions.SourceId) ? Endpoint : resolvedOptions.SourceId!;
PollInterval = resolvedOptions.PollInterval;
@ -463,9 +459,10 @@ namespace GameCult.Mesh
where TDocument : class
{
var descriptor = CultDocumentRegistry.Shared.GetRequired<TDocument>();
var selection = CultMesh.ResolveDefaultSelection(schemaIds, recordKeys, descriptor);
return CultMesh.FetchSnapshotDocumentsAsync<TDocument>(
Endpoint,
CreateRequest(ResolveDefaultSchemaFilter(schemaIds, recordKeys, descriptor), recordKeys),
CreateRequest(selection.Schemas, selection.Keys),
DocumentRegistry);
}
@ -480,10 +477,11 @@ namespace GameCult.Mesh
if (node == null) throw new ArgumentNullException(nameof(node));
var descriptor = CultDocumentRegistry.Shared.GetRequired<TDocument>();
var selection = CultMesh.ResolveDefaultSelection(schemaIds, recordKeys, descriptor);
var result = await SyncSnapshotAsync(
node,
ResolveDefaultSchemaFilter(schemaIds, recordKeys, descriptor),
recordKeys,
selection.Schemas,
selection.Keys,
flush)
.ConfigureAwait(false);
return CultMesh.DecodeSnapshotDocuments<TDocument>(result.Snapshot, DocumentRegistry);
@ -612,53 +610,13 @@ namespace GameCult.Mesh
IReadOnlyList<string>? schemaIds,
IReadOnlyList<string>? recordKeys)
{
var request = CloneSnapshotRequestOptions(Request);
request.SchemaIds = schemaIds ?? request.SchemaIds;
request.RecordKeys = recordKeys ?? request.RecordKeys;
var request = Request with { Selection = CultMesh.OverlaySelection(Request.Selection, schemaIds, recordKeys) };
if (string.IsNullOrWhiteSpace(request.RudpRuntimeId))
request.RudpRuntimeId = Context.RuntimeId;
if (string.IsNullOrWhiteSpace(request.MessageIdPrefix))
request.MessageIdPrefix = $"cultmesh:{Context.RuntimeId}:snapshot";
return request;
}
private static IReadOnlyList<string>? ResolveDefaultSchemaFilter(
IReadOnlyList<string>? schemaIds,
IReadOnlyList<string>? recordKeys,
CultDocumentDescriptor descriptor)
{
if (schemaIds != null)
return schemaIds;
return recordKeys is { Count: > 0 }
? null
: new[] { descriptor.SchemaId };
}
private static CultMeshSnapshotRequestOptions CloneSnapshotRequestOptions(CultMeshSnapshotRequestOptions? source)
{
if (source == null)
return new CultMeshSnapshotRequestOptions();
return new CultMeshSnapshotRequestOptions
{
SchemaIds = source.SchemaIds,
RecordKeys = source.RecordKeys,
ShardId = source.ShardId,
ShardEpoch = source.ShardEpoch,
ResponseTimeout = source.ResponseTimeout,
ConnectTimeout = source.ConnectTimeout,
MessageIdPrefix = source.MessageIdPrefix,
Security = source.Security,
ConfigureClient = source.ConfigureClient,
CreateClient = source.CreateClient,
RudpRuntimeId = source.RudpRuntimeId,
RudpConnectionId = source.RudpConnectionId,
RudpConnectPayload = source.RudpConnectPayload,
RudpMaxFragmentBytes = source.RudpMaxFragmentBytes,
RudpResendDelayMs = source.RudpResendDelayMs
};
}
}
public static partial class CultMesh
@ -733,7 +691,7 @@ namespace GameCult.Mesh
if (string.IsNullOrWhiteSpace(endpoint)) throw new ArgumentException("Value must be non-empty.", nameof(endpoint));
if (string.IsNullOrWhiteSpace(recordKey)) throw new ArgumentException("Value must be non-empty.", nameof(recordKey));
var request = CloneSnapshotFacadeRequestOptions(options?.Request);
var request = options?.Request is null ? new CultMeshSnapshotRequestOptions() : options.Request with { };
request.CreateClient = createClient;
var resolvedOptions = new CultMeshSnapshotEndpointOptions
{
@ -747,31 +705,6 @@ namespace GameCult.Mesh
return SnapshotEndpoint(endpoint, resolvedOptions).SyncDocumentAsync<TDocument>(node, recordKey, flush);
}
private static CultMeshSnapshotRequestOptions CloneSnapshotFacadeRequestOptions(CultMeshSnapshotRequestOptions? source)
{
if (source == null)
return new CultMeshSnapshotRequestOptions();
return new CultMeshSnapshotRequestOptions
{
SchemaIds = source.SchemaIds,
RecordKeys = source.RecordKeys,
ShardId = source.ShardId,
ShardEpoch = source.ShardEpoch,
ResponseTimeout = source.ResponseTimeout,
ConnectTimeout = source.ConnectTimeout,
MessageIdPrefix = source.MessageIdPrefix,
Security = source.Security,
ConfigureClient = source.ConfigureClient,
CreateClient = source.CreateClient,
RudpRuntimeId = source.RudpRuntimeId,
RudpConnectionId = source.RudpConnectionId,
RudpConnectPayload = source.RudpConnectPayload,
RudpMaxFragmentBytes = source.RudpMaxFragmentBytes,
RudpResendDelayMs = source.RudpResendDelayMs
};
}
/// <summary>
/// Fetches one scoped raw CultNet snapshot from an endpoint.
/// </summary>
@ -819,8 +752,8 @@ namespace GameCult.Mesh
client.SendCultNet(new CultNetSnapshotRequestMessage
{
MessageId = messageId,
SchemaIds = CleanSnapshotFilter(resolvedOptions.SchemaIds),
RecordKeys = CleanSnapshotFilter(resolvedOptions.RecordKeys),
SchemaIds = resolvedOptions.Selection?.Schemas,
RecordKeys = resolvedOptions.Selection?.Keys,
ShardId = string.IsNullOrWhiteSpace(resolvedOptions.ShardId) ? null : resolvedOptions.ShardId,
ShardEpoch = resolvedOptions.ShardEpoch
});
@ -886,30 +819,48 @@ namespace GameCult.Mesh
};
}
internal static CultMeshSnapshotRequestOptions CloneSnapshotRequestOptions(
CultMeshSnapshotRequestOptions? source)
/// <summary>
/// Resolves one selection for a typed convenience call: explicit <paramref name="schemaIds"/>
/// win; otherwise an explicit, non-empty <paramref name="recordKeys"/> means no schema filter;
/// otherwise the selection defaults to the document type's own schema. The one owner of that
/// rule (docs/cultnet-selection-cut.md, section 4).
/// </summary>
internal static CultNetSelection ResolveDefaultSelection(
IReadOnlyList<string>? schemaIds,
IReadOnlyList<string>? recordKeys,
CultDocumentDescriptor descriptor)
{
if (source == null) return new CultMeshSnapshotRequestOptions();
return new CultMeshSnapshotRequestOptions
var schemas = schemaIds != null
? Clean(schemaIds)
: (recordKeys is { Count: > 0 } ? null : new[] { descriptor.SchemaId });
return new CultNetSelection { Schemas = schemas, Keys = Clean(recordKeys) };
}
/// <summary>Overlays explicit schema/record-key overrides onto a default selection.</summary>
internal static CultNetSelection OverlaySelection(
CultNetSelection? defaults,
IReadOnlyList<string>? schemaIds,
IReadOnlyList<string>? recordKeys)
{
return new CultNetSelection
{
SchemaIds = source.SchemaIds,
RecordKeys = source.RecordKeys,
ShardId = source.ShardId,
ShardEpoch = source.ShardEpoch,
ResponseTimeout = source.ResponseTimeout,
ConnectTimeout = source.ConnectTimeout,
MessageIdPrefix = source.MessageIdPrefix,
Security = source.Security,
ConfigureClient = source.ConfigureClient,
CreateClient = source.CreateClient,
RudpRuntimeId = source.RudpRuntimeId,
RudpConnectionId = source.RudpConnectionId,
RudpConnectPayload = source.RudpConnectPayload,
RudpMaxFragmentBytes = source.RudpMaxFragmentBytes,
RudpResendDelayMs = source.RudpResendDelayMs
Schemas = schemaIds != null ? Clean(schemaIds) : defaults?.Schemas,
Keys = recordKeys != null ? Clean(recordKeys) : defaults?.Keys
};
}
private static string[]? Clean(IReadOnlyList<string>? values)
{
if (values is not { Count: > 0 })
return null;
var filtered = values
.Where(value => !string.IsNullOrWhiteSpace(value))
.Distinct(StringComparer.Ordinal)
.ToArray();
return filtered.Length == 0 ? null : filtered;
}
internal static string CreateSnapshotMessageId(CultMeshSnapshotRequestOptions options)
{
var prefix = string.IsNullOrWhiteSpace(options.MessageIdPrefix)
@ -963,25 +914,13 @@ namespace GameCult.Mesh
{
throw new TimeoutException(
$"Timed out waiting for CultNet snapshot response '{messageId}' from {endpoint} " +
$"for schemas [{string.Join(", ", CleanSnapshotFilter(options.SchemaIds) ?? Array.Empty<string>())}] " +
$"and records [{string.Join(", ", CleanSnapshotFilter(options.RecordKeys) ?? Array.Empty<string>())}].");
$"for schemas [{string.Join(", ", options.Selection?.Schemas ?? Array.Empty<string>())}] " +
$"and records [{string.Join(", ", options.Selection?.Keys ?? Array.Empty<string>())}].");
}
return await responseTask.ConfigureAwait(false);
}
internal static string[]? CleanSnapshotFilter(IReadOnlyList<string>? values)
{
if (values is not { Count: > 0 })
return null;
var filtered = values
.Where(value => !string.IsNullOrWhiteSpace(value))
.Distinct(StringComparer.Ordinal)
.ToArray();
return filtered.Length == 0 ? null : filtered;
}
internal static IReadOnlyList<TDocument> DecodeSnapshotDocuments<TDocument>(
CultNetSnapshotResponseRawMessage snapshot,
CultNetDocumentRegistry registry)

4
tests/GameCult.Mesh.Tests/CultMeshBodyTransportTests.cs
View file

@ -463,11 +463,11 @@ public sealed class CultMeshBodyTransportTests
new[] { "eve:field:gravity" },
new[] { "gamecult.eve.fields_splats.v1" });
subscription.RecordKeys.Should().Equal(
subscription.Selection.Keys.Should().Equal(
"eve:view:pilot",
CultMeshBodyPublicationDocument.CreateLatestRecordKey("eve:body:pilot").Value,
"eve:field:gravity");
subscription.SchemaIds.Should().Equal(
subscription.Selection.Schemas.Should().Equal(
"gamecult.eve.entity_soa_view.v2",
CultMeshBodyPublicationSchemaVersions.Publication,
"gamecult.eve.fields_splats.v1");

8
tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs
View file

@ -2936,7 +2936,7 @@ public sealed class CultMeshStreamingTests
var endpoint = "cultnet://snapshot-helper.test:3075";
var options = new CultMeshSnapshotRequestOptions
{
RecordKeys = new[] { key.Value },
Selection = new CultNetSelection { Keys = new[] { key.Value } },
ShardId = "primary",
ShardEpoch = 7,
MessageIdPrefix = "mesh-test-scoped-snapshot",
@ -2986,7 +2986,7 @@ public sealed class CultMeshStreamingTests
var options = new CultMeshSnapshotRequestOptions
{
ResponseTimeout = TimeSpan.FromSeconds(30),
RecordKeys = new[] { "eve:surface:test" },
Selection = new CultNetSelection { Keys = new[] { "eve:surface:test" } },
CreateClient = () => new FailedBackgroundSchemaClient(
new FormatException("malformed fragment"))
};
@ -3051,7 +3051,7 @@ public sealed class CultMeshStreamingTests
"cultnet://foreign-schema.test:3075",
new CultMeshSnapshotRequestOptions
{
RecordKeys = new[] { key.Value },
Selection = new CultNetSelection { Keys = new[] { key.Value } },
CreateClient = () => new MeshSnapshotSchemaClient(request =>
{
snapshot.MessageId = request.MessageId;
@ -3236,7 +3236,7 @@ public sealed class CultMeshStreamingTests
endpoint,
new CultMeshSnapshotRequestOptions
{
RecordKeys = new[] { key.Value },
Selection = new CultNetSelection { Keys = new[] { key.Value } },
CreateClient = () => new MeshSnapshotSchemaClient(request =>
networkRegistry.CreateRawSnapshotResponse(sourceCache, request.MessageId, request))
},