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 fix batch: R-A/B/C/D/E/F/G/H/J/M (C#)

Browse source 
Wires v1 into both C# servers (R-A): CultNetDatabaseServer registers
CultNetSnapshotRequestV1Message/CultNetDatabaseSubscribeV1Message
listeners and Server.cs's pre-verification allowlist gains them;
CultNetDatabaseSubscriptionServer's subscribe path is generalised to a
shared SubscriptionRequest so v0 (lowered) and v1 selections run the
same Subscribe/Watch/Reconcile core, with D6's rule wired in: a
hop-bearing selection runs the full reconcile diff on every change
instead of the single-row fast path. CultNetDatabase.CurrentAsOf gives
both servers a real snapshot watermark.

Fixes the confirmed parity and correctness defects from Soul's
whole-cut pass:
- R-B: cited/cites edges are anchored to the row the direction actually
  reports against (the citer under cites, the cited row under cited),
  not filtered by "From is on the page" regardless of direction; order
  is page-row order then (from, role, to) in code-point order.
- R-C: a new CultNetCodePointComparer replaces every StringComparer.Ordinal/
  string.CompareOrdinal in the row tiebreak, cursor position and edge
  order, so an astral key sorts the same way here as it does under
  Rust's UTF-8 byte comparison.
- R-D: CultCache renders a float/double's exact decimal expansion from
  its IEEE-754 bits via BigInteger (RenderExactFloat/RenderExactDouble),
  never .NET's shortest round-trip form, which rounded a tie the
  opposite way from Rust at some values. The now-dead ExpandScientificNotation
  is deleted.
- R-E: MatchesCitation compares a cites target through
  CultNetSchemaAliasMatching instead of an exact SchemaId string
  compare; ToEdge and ToRawRecord now emit the same wire schema id
  (WireSchemaId) for the same row, where ToEdge previously used the
  descriptor's own id and ToRawRecord the binding's override.
- R-F: CultNetSelectionEvaluator.Select validates first, unconditionally
  - the validate:false/true flag on SelectPage/CreateRawSnapshotResponse
  is gone, so no public evaluation path can skip the door. Mesh's
  EnsureV0Compatible runs the new CultNetSelectionValidation.ValidateShape
  (the declaration-independent half of the door) before its own v0
  transport refusals, so Keys=[]/[""] is refused there too.
- R-G: Evaluation.TotalMatched is the selection's whole matching count;
  SelectPage's Matched field now reports it instead of the page's row
  count. CreateRawSnapshotResponse and CreateShardSnapshotResponse
  (v0/shard full walks) call the new CultNetDocumentRegistry.SelectAll,
  which evaluates once via CultNetSelectionEvaluator.EvaluateAll and
  projects every matched row, instead of looping SelectPage per 200-row
  page and re-filtering/re-sorting the whole row set each time.
- R-H: the cursor digest is length-prefixed for every string and list
  (AppendString/AppendList) instead of delimiter-joined, closing the
  collision ["a|b"] == ["a","b"] and ["a,b"] == ["a","b"].
- R-J: any_of on a numeric alias compares TryGetIndexNumber's canonical
  rendering, not the string getter's culture-dependent ToString().
- R-M: WithBindingSchemaAlias is deleted; both servers' single-change
  fast path calls CultNetDocumentRegistry.ExpandSchemaBindingAliases
  instead, the one binding-alias reconciler. The duplicated v0/shard
  lower-and-page loop is gone with SelectAll above.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-22 09:39:04 +02:00
commit 790385364c
10 changed files with 752 additions and 188 deletions
Download patch file Download diff file Expand all files Collapse all files

118
src/GameCult.Caching/CultCache.cs
View file

@ -5,6 +5,7 @@ using System.Diagnostics;
using System.Globalization;
using System.IO;
using System.Linq;
using System.Numerics;
using System.Reflection;
using System.Runtime.CompilerServices;
using System.Runtime.ExceptionServices;
@ -1230,44 +1231,105 @@ namespace GameCult.Caching
// "a NaN or infinite member matches no comparison."
private static string? RenderCanonicalNumber(object raw) => raw switch
{
float f => float.IsNaN(f) || float.IsInfinity(f) ? null : CanonicalizeDecimalDigits(ExpandScientificNotation(f.ToString(CultureInfo.InvariantCulture))),
double d => double.IsNaN(d) || double.IsInfinity(d) ? null : CanonicalizeDecimalDigits(ExpandScientificNotation(d.ToString(CultureInfo.InvariantCulture))),
float f => float.IsNaN(f) || float.IsInfinity(f) ? null : RenderExactFloat(f),
double d => double.IsNaN(d) || double.IsInfinity(d) ? null : RenderExactDouble(d),
decimal m => CanonicalizeDecimalDigits(m.ToString(CultureInfo.InvariantCulture)),
_ => CanonicalizeDecimalDigits(Convert.ToString(raw, CultureInfo.InvariantCulture)!)
};
// double/float's default ToString(CultureInfo.InvariantCulture) is .NET's shortest round-trip
// form since .NET Core 3.0, but it uses exponent notation outside a middle magnitude range
// ("1E+20", "1.5E-05"). Rewritten to positional digits before CanonicalizeDecimalDigits runs, so
// the wire never carries an exponent (S16's float64-rendering mutant: a long of 2^53+1 must
// compare greater than "9007199254740992", which a double-precision render would fail).
private static string ExpandScientificNotation(string value)
// R-D (docs/cultnet-selection-cut.md): the wire's canonical decimal is the value's exact decimal
// expansion, not a shortest round-trip form. .NET's default float/double ToString has used the
// shortest round-trip form since .NET Core 3.0, which rounds a tie between two decimals to
// whichever is "shorter" - Rust's shortest form breaks the same tie the other way at some
// values, so the two runtimes rendered different strings for the same bit pattern (394/200k f32,
// 48/200k f64 in the parity vectors). Every finite float and double is an exact dyadic rational
// (significand * 2^exponent for an integer significand and exponent), so it has a finite exact
// decimal expansion computed with BigInteger: multiply the significand by 5^-exponent when the
// exponent is negative (1/2^k == 5^k/10^k) to get the exact digits, and place the decimal point.
// No CLR floating-point arithmetic and no string round-tripping is on this path.
private static string RenderExactFloat(float value)
{
var eIndex = value.IndexOfAny(ExponentMarkers);
if (eIndex < 0) return value;
var bits = BitConverter.SingleToInt32Bits(value);
var negative = bits < 0;
var biasedExponent = (bits >> 23) & 0xFF;
var mantissaBits = bits & 0x7FFFFF;
if (biasedExponent == 0 && mantissaBits == 0)
return "0"; // canonical zero: a single "0", never "-0" (Q-J).
var negative = value.Length > 0 && value[0] == '-';
var unsigned = negative ? value.Substring(1) : value;
eIndex = unsigned.IndexOfAny(ExponentMarkers);
var mantissa = unsigned.Substring(0, eIndex);
var exponent = int.Parse(unsigned.Substring(eIndex + 1), NumberStyles.AllowLeadingSign, CultureInfo.InvariantCulture);
var dot = mantissa.IndexOf('.');
var digits = dot < 0 ? mantissa : mantissa.Remove(dot, 1);
var pointPosition = (dot < 0 ? digits.Length : dot) + exponent;
string expanded;
if (pointPosition <= 0)
expanded = "0." + new string('0', -pointPosition) + digits;
else if (pointPosition >= digits.Length)
expanded = digits + new string('0', pointPosition - digits.Length);
BigInteger significand;
int exponent;
if (biasedExponent == 0)
{
significand = mantissaBits;
exponent = -126 - 23; // subnormal: no implicit leading 1, minimum exponent.
}
else
expanded = digits.Substring(0, pointPosition) + "." + digits.Substring(pointPosition);
{
significand = mantissaBits | (1 << 23);
exponent = biasedExponent - 127 - 23;
}
return negative ? "-" + expanded : expanded;
return RenderExactDecimal(negative, significand, exponent);
}
private static readonly char[] ExponentMarkers = { 'E', 'e' };
private static string RenderExactDouble(double value)
{
var bits = BitConverter.DoubleToInt64Bits(value);
var negative = bits < 0;
var biasedExponent = (int)((bits >> 52) & 0x7FF);
var mantissaBits = bits & 0xFFFFFFFFFFFFFL;
if (biasedExponent == 0 && mantissaBits == 0)
return "0"; // canonical zero: a single "0", never "-0" (Q-J).
BigInteger significand;
int exponent;
if (biasedExponent == 0)
{
significand = mantissaBits;
exponent = -1022 - 52; // subnormal: no implicit leading 1, minimum exponent.
}
else
{
significand = mantissaBits | (1L << 52);
exponent = biasedExponent - 1023 - 52;
}
return RenderExactDecimal(negative, significand, exponent);
}
// value = significand * 2^exponent, significand > 0. exponent >= 0 is an exact integer shift;
// exponent < 0 rewrites 2^exponent as 5^-exponent / 10^-exponent, so the digits are exact and
// the decimal point lands -exponent places from the right - no precision limit, no rounding.
private static string RenderExactDecimal(bool negative, BigInteger significand, int exponent)
{
string digits;
int fractionDigits;
if (exponent >= 0)
{
digits = (significand << exponent).ToString(CultureInfo.InvariantCulture);
fractionDigits = 0;
}
else
{
fractionDigits = -exponent;
digits = (significand * BigInteger.Pow(5, fractionDigits)).ToString(CultureInfo.InvariantCulture);
}
string positional;
if (fractionDigits == 0)
{
positional = digits;
}
else
{
if (digits.Length <= fractionDigits)
digits = digits.PadLeft(fractionDigits + 1, '0');
var splitAt = digits.Length - fractionDigits;
positional = digits.Substring(0, splitAt) + "." + digits.Substring(splitAt);
}
return CanonicalizeDecimalDigits(negative ? "-" + positional : positional);
}
// The one canonicalizer every rendered number passes through: strips leading integer zeros and
// trailing fractional zeros, and collapses a negative value that reduces to zero into "0" - the

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

@ -2755,9 +2755,7 @@ namespace GameCult.Mesh
exact = candidate;
else if (aliased == null && CultNetSchemaAliasMatching.Matches(candidate.SchemaId, descriptor))
aliased = candidate;
else if (byPayload == null &&
CultNetDocumentRegistry.TryReadSchemaVersion(candidate.Payload) is { } payloadSchemaVersion &&
CultNetSchemaAliasMatching.Matches(payloadSchemaVersion, descriptor))
else if (byPayload == null && CultNetDocumentRegistry.PayloadMatchesSchema(candidate.Payload, descriptor))
byPayload = candidate;
}

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

@ -833,6 +833,12 @@ namespace GameCult.Mesh
{
if (selection == null)
return;
// The door runs first (docs/cultnet-selection-cut.md, R-F): Mesh has no descriptor list to
// check field/role reachability against, but the declaration-independent half of the door -
// an empty or blank-only schemas/keys list, an unrecognised projection - applies here just
// as it does inside CultNetSelectionEvaluator.Select, so Keys=[]/[""] is refused rather than
// silently lowered to "every key".
CultNetSelectionValidation.ValidateShape(selection);
if (selection.Fields != null)
throw new CultNetSelectionInvalidException("fields", null, "CultMesh sends cultnet.snapshot_request.v0 in this cut; selection.fields cannot reach the wire (FU-Mesh-v1).");
if (selection.Cites != null)
@ -974,8 +980,7 @@ namespace GameCult.Mesh
var canDeserializeAsSchemaAlias =
CultNetSchemaAliasMatching.Matches(record.SchemaId, descriptor) ||
(binding != null && CultNetSchemaAliasMatching.Matches(binding.SchemaId, descriptor)) ||
(CultNetDocumentRegistry.TryReadSchemaVersion(record.Payload) is { } payloadSchemaVersion &&
CultNetSchemaAliasMatching.Matches(payloadSchemaVersion, descriptor));
CultNetDocumentRegistry.PayloadMatchesSchema(record.Payload, descriptor);
if (!canDeserializeWithBinding && !canDeserializeAsSchemaAlias)
continue;

59
src/GameCult.Networking/CultNetDatabase.cs
View file

@ -98,36 +98,9 @@ namespace GameCult.Networking
: null;
}
/// <summary>
/// A selection whose schema filter also matches a document's wire binding id, when it carries
/// one different from its descriptor's own. <see cref="CultDocumentDescriptor"/> knows nothing
/// of <c>CultNetDocumentBinding</c>'s optional schema-id override, so a caller that has both in
/// hand (the two subscription servers) reconciles them here rather than in the evaluator, which
/// stays decoupled from CultNetDocumentRegistry by design.
/// </summary>
public static CultNetSelection WithBindingSchemaAlias(CultNetSelection selection, CultDocumentDescriptor descriptor, string? bindingSchemaId)
{
if (selection.Schemas is not { Length: > 0 } ||
string.IsNullOrEmpty(bindingSchemaId) ||
!selection.Schemas.Contains(bindingSchemaId, StringComparer.Ordinal) ||
MatchesAny(selection.Schemas, descriptor))
{
return selection;
}
return new CultNetSelection
{
Schemas = null,
Keys = selection.Keys,
Fields = selection.Fields,
Cites = selection.Cites,
Cited = selection.Cited,
Projection = selection.Projection,
Descending = selection.Descending,
Limit = selection.Limit,
Cursor = selection.Cursor
};
}
// R-M: WithBindingSchemaAlias (a second binding-alias reconciler, scoped to one descriptor) is
// deleted - CultNetDocumentRegistry.ExpandSchemaBindingAliases is the one copy every caller
// uses, CultNetDatabaseServer and CultNetDatabaseSubscriptionServer included.
}
/// <summary>
@ -645,26 +618,19 @@ namespace GameCult.Networking
{
Schemas = CultNetV0SelectionLowering.Lower(filter?.SchemaIds),
Keys = CultNetV0SelectionLowering.Lower(filter?.RecordKeys),
Projection = CultNetSelectionProjections.Document,
Limit = CultNetSelectionEvaluator.LimitMax
Projection = CultNetSelectionProjections.Document
};
bool RowFilter(CultDocumentDescriptor descriptor, CultRecordKey key) =>
!string.IsNullOrWhiteSpace(key.Value) && shard.Matches(descriptor.SchemaId, key);
var documents = new List<CultNetRawDocumentRecord>();
string? cursor = null;
do
{
selection.Cursor = cursor;
var page = _documents.SelectPage(_cache, selection, ordinalOf: static (_, _) => 0, asOf: 0, validate: false, options: null, rowFilter: RowFilter);
documents.AddRange(page.Documents ?? Array.Empty<CultNetRawDocumentRecord>());
cursor = page.Next;
} while (cursor != null);
// R-G: one evaluation answers the whole shard snapshot, not a loop that re-filters and
// re-sorts the shard's row set once per 200-row page.
var page = _documents.SelectAll(_cache, selection, ordinalOf: static (_, _) => 0, asOf: 0, options: null, rowFilter: RowFilter);
return new CultNetSnapshotResponseRawMessage
{
MessageId = string.IsNullOrWhiteSpace(messageId) ? Guid.NewGuid().ToString("N") : messageId,
Documents = documents.ToArray(),
Documents = page.Documents ?? Array.Empty<CultNetRawDocumentRecord>(),
ShardId = shard.ShardId,
ShardEpoch = shard.Epoch,
ShardLogSequence = GetLatestMutationLogSequence(shard.ShardId)
@ -1304,6 +1270,15 @@ namespace GameCult.Networking
public long? LastWriteSequence(string schemaId, CultRecordKey key) =>
_lastWriteSequence.TryGetValue((schemaId, key.Value), out var sequence) ? sequence : null;
/// <summary>
/// The current write-sequence watermark this database can present an exact snapshot for (R-A):
/// the highest <see cref="LastWriteSequence"/> assigned to any row, 0 when nothing has been
/// committed yet. This is the evaluator's <c>asOf</c> for a v1 snapshot or subscription page - it
/// advances on every committed write, so a cursor minted against an older watermark refuses as
/// <c>cursor_stale</c> once a page's answer could differ.
/// </summary>
public ulong CurrentAsOf() => _lastWriteSequence.Count == 0 ? 0UL : (ulong)_lastWriteSequence.Values.Max();
private void RecordMutationLogEntry(
CultNetShardMutationLogEntry entry,
CultNetShardLogEntryMessage? wireEntry = null)

122
src/GameCult.Networking/CultNetDatabaseServer.cs
View file

@ -18,11 +18,13 @@ namespace GameCult.Networking
private readonly CultNetDatabase _database;
private readonly CultNetDatabaseServerOptions _options;
private readonly Func<CultNetSnapshotRequestMessage, CultNetServerPeer, Task> _snapshotHandler;
private readonly Func<CultNetSnapshotRequestV1Message, CultNetServerPeer, Task> _snapshotHandlerV1;
private readonly Func<CultNetDocumentPutRawMessage, CultNetServerPeer, Task> _putHandler;
private readonly Func<CultNetDocumentDeleteMessage, CultNetServerPeer, Task> _deleteHandler;
private readonly Func<CultNetShardCatalogRequestMessage, CultNetServerPeer, Task> _shardCatalogHandler;
private readonly Func<CultNetShardLogRequestMessage, CultNetServerPeer, Task> _shardLogHandler;
private readonly Func<CultNetDatabaseSubscribeMessage, CultNetServerPeer, Task> _subscribeHandler;
private readonly Func<CultNetDatabaseSubscribeV1Message, CultNetServerPeer, Task> _subscribeHandlerV1;
private readonly Func<CultNetDatabaseUnsubscribeMessage, CultNetServerPeer, Task> _unsubscribeHandler;
private readonly ConcurrentDictionary<string, IDisposable> _subscriptions = new(StringComparer.Ordinal);
private bool _disposed;
@ -39,19 +41,23 @@ namespace GameCult.Networking
_database = database ?? throw new ArgumentNullException(nameof(database));
_options = options ?? new CultNetDatabaseServerOptions();
_snapshotHandler = HandleSnapshotRequestAsync;
_snapshotHandlerV1 = HandleSnapshotRequestV1Async;
_putHandler = HandlePutAsync;
_deleteHandler = HandleDeleteAsync;
_shardCatalogHandler = HandleShardCatalogRequestAsync;
_shardLogHandler = HandleShardLogRequestAsync;
_subscribeHandler = HandleSubscribeAsync;
_subscribeHandlerV1 = HandleSubscribeV1Async;
_unsubscribeHandler = HandleUnsubscribeAsync;
_server.OnCultNet(_snapshotHandler);
_server.OnCultNet(_snapshotHandlerV1);
_server.OnCultNet(_putHandler);
_server.OnCultNet(_deleteHandler);
_server.OnCultNet(_shardCatalogHandler);
_server.OnCultNet(_shardLogHandler);
_server.OnCultNet(_subscribeHandler);
_server.OnCultNet(_subscribeHandlerV1);
_server.OnCultNet(_unsubscribeHandler);
}
@ -111,6 +117,22 @@ namespace GameCult.Networking
request);
}
/// <summary>
/// Answers a typed-selection snapshot request (R-A, docs/cultnet-selection-cut.md): the door,
/// then one evaluation over the database's rows in last-write order, through
/// <see cref="CultNetDocumentRegistry.CreateSelectionResponse"/>.
/// </summary>
public CultNetSnapshotResponseRawV1Message CreateSelectionResponse(CultNetSnapshotRequestV1Message request)
{
if (request == null) throw new ArgumentNullException(nameof(request));
return _database.Documents.CreateSelectionResponse(
_database.Cache,
string.IsNullOrWhiteSpace(request.MessageId) ? Guid.NewGuid().ToString("N") : request.MessageId,
request.Selection,
ordinalOf: (schemaId, key) => _database.LastWriteSequence(schemaId, key) ?? 0,
asOf: _database.CurrentAsOf());
}
/// <summary>
/// Creates a shard catalog response from the database shard map.
/// </summary>
@ -193,11 +215,13 @@ namespace GameCult.Networking
_disposed = true;
_server.RemoveCultNetMessageListener<CultNetSnapshotRequestMessage>(_snapshotHandler);
_server.RemoveCultNetMessageListener<CultNetSnapshotRequestV1Message>(_snapshotHandlerV1);
_server.RemoveCultNetMessageListener<CultNetDocumentPutRawMessage>(_putHandler);
_server.RemoveCultNetMessageListener<CultNetDocumentDeleteMessage>(_deleteHandler);
_server.RemoveCultNetMessageListener<CultNetShardCatalogRequestMessage>(_shardCatalogHandler);
_server.RemoveCultNetMessageListener<CultNetShardLogRequestMessage>(_shardLogHandler);
_server.RemoveCultNetMessageListener<CultNetDatabaseSubscribeMessage>(_subscribeHandler);
_server.RemoveCultNetMessageListener<CultNetDatabaseSubscribeV1Message>(_subscribeHandlerV1);
_server.RemoveCultNetMessageListener<CultNetDatabaseUnsubscribeMessage>(_unsubscribeHandler);
foreach (var subscription in _subscriptions.Values)
{
@ -213,6 +237,28 @@ namespace GameCult.Networking
return Task.CompletedTask;
}
private Task HandleSnapshotRequestV1Async(CultNetSnapshotRequestV1Message request, CultNetServerPeer peer)
{
try
{
peer.SendCultNet(CreateSelectionResponse(request));
}
catch (CultNetSelectionInvalidException ex)
{
peer.SendCultNet(new CultNetErrorMessage { Error = $"selection_invalid: {ex.Message}" });
}
catch (CultNetSelectionCursorException ex)
{
peer.SendCultNet(new CultNetErrorMessage { Error = $"{ex.Code}: {ex.Message}" });
}
catch (CultNetSelectionReferenceOutsideTargetException ex)
{
peer.SendCultNet(new CultNetErrorMessage { Error = $"reference_outside_target: {ex.Message}" });
}
return Task.CompletedTask;
}
private Task HandleShardCatalogRequestAsync(CultNetShardCatalogRequestMessage request, CultNetServerPeer peer)
{
peer.SendCultNet(CreateShardCatalogResponse(request));
@ -315,6 +361,68 @@ namespace GameCult.Networking
return Task.CompletedTask;
}
private Task HandleSubscribeV1Async(CultNetDatabaseSubscribeV1Message message, CultNetServerPeer peer)
{
var subscriptionId = string.IsNullOrWhiteSpace(message.SubscriptionId)
? message.MessageId
: message.SubscriptionId;
if (string.IsNullOrWhiteSpace(subscriptionId))
{
peer.SendCultNet(new CultNetErrorMessage { Error = "Database subscription requires a subscriptionId or messageId." });
return Task.CompletedTask;
}
// This server delivers live changes through the single-row fast path (CreateChangeMessage /
// CultNetSelectionEvaluator.Matches), which is set-independent by construction - the same
// ceiling v0 subscriptions on this server already have. A hop-bearing selection is
// set-dependent (D6) and needs the subscription server's reconcile loop instead, so it is
// refused here rather than reaching Matches' own InvalidOperationException on the first change.
if (message.Selection.HasHop)
{
peer.SendCultNet(new CultNetErrorMessage { Error = "selection_invalid: selection.cites/selection.cited need the subscription server's reconcile loop (D6); this server only fast-matches a single row." });
return Task.CompletedTask;
}
try
{
message.Selection.Validate(_database.Cache.Registry.AllDescriptors.ToArray());
}
catch (CultNetSelectionInvalidException ex)
{
peer.SendCultNet(new CultNetErrorMessage { Error = $"selection_invalid: {ex.Message}" });
return Task.CompletedTask;
}
var key = SubscriptionKey(peer.Peer, subscriptionId);
_subscriptions.AddOrUpdate(
key,
_ => CreateSubscription(message.Selection, subscriptionId, peer),
(_, existing) =>
{
existing.Dispose();
return CreateSubscription(message.Selection, subscriptionId, peer);
});
if (message.IncludeSnapshot)
{
peer.SendCultNet(CreateSelectionResponse(new CultNetSnapshotRequestV1Message
{
MessageId = message.MessageId,
Selection = message.Selection
}));
}
else
{
peer.SendCultNet(new CultNetSnapshotResponseRawV1Message
{
MessageId = message.MessageId,
AsOf = _database.CurrentAsOf()
});
}
return Task.CompletedTask;
}
private Task HandleUnsubscribeAsync(CultNetDatabaseUnsubscribeMessage message, CultNetServerPeer peer)
{
var subscriptionId = string.IsNullOrWhiteSpace(message.SubscriptionId)
@ -335,13 +443,21 @@ namespace GameCult.Networking
CultNetServerPeer peer)
{
// v0 has no engine of its own (docs/cultnet-selection-cut.md, D3/D4): it lowers into a
// Selection and is matched by the one evaluator, same as a v1 subscribe would be.
// Selection and is matched by the one evaluator, same as a v1 subscribe.
var selection = new CultNetSelection
{
Schemas = request.SchemaIds,
Keys = request.RecordKeys,
Projection = CultNetSelectionProjections.Document
};
return CreateSubscription(selection, subscriptionId, peer);
}
private IDisposable CreateSubscription(
CultNetSelection selection,
string subscriptionId,
CultNetServerPeer peer)
{
return _database.WatchAllChanges().Subscribe(change =>
{
var outbound = CreateChangeMessage(change, subscriptionId, selection);
@ -370,8 +486,8 @@ namespace GameCult.Networking
return null;
}
var binding = _database.Documents.GetByDocumentType(descriptor.DocumentType);
var effectiveSelection = CultNetSchemaAliasMatching.WithBindingSchemaAlias(selection, descriptor, binding?.SchemaId);
// R-M: the one binding-alias reconciler (CultNetDocumentRegistry.ExpandSchemaBindingAliases).
var effectiveSelection = _database.Documents.ExpandSchemaBindingAliases(selection);
if (!CultNetSelectionEvaluator.Matches(descriptor, key, document, effectiveSelection))
{
return null;

214
src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs
View file

@ -22,14 +22,15 @@ namespace GameCult.Networking
private readonly ICultNetSchemaServer _server;
private readonly CultNetDatabase _database;
private readonly Func<CultNetDatabaseSubscribeMessage, ICultNetSchemaServerPeer, Task> _subscribe;
private readonly Func<CultNetDatabaseSubscribeV1Message, ICultNetSchemaServerPeer, Task> _subscribeV1;
private readonly Func<CultNetDatabaseUnsubscribeMessage, ICultNetSchemaServerPeer, Task> _unsubscribe;
private readonly Func<CultNetDatabaseSubscribeMessage, ICultNetSchemaServerPeer, bool>? _authorizeRequest;
private readonly Func<CultNetDatabaseSubscribeMessage, ICultNetSchemaServerPeer, string, string, bool>? _authorizeRecord;
private readonly ICultNetSchemaServerPeerLifecycle? _peerLifecycle;
private readonly ConcurrentDictionary<SubscriptionKey, IDisposable> _subscriptions =
new ConcurrentDictionary<SubscriptionKey, IDisposable>(SubscriptionKeyComparer.Instance);
private readonly ConcurrentDictionary<SubscriptionKey, CultNetDatabaseSubscribeMessage> _requests =
new ConcurrentDictionary<SubscriptionKey, CultNetDatabaseSubscribeMessage>(SubscriptionKeyComparer.Instance);
private readonly ConcurrentDictionary<SubscriptionKey, SubscriptionRequest> _requests =
new ConcurrentDictionary<SubscriptionKey, SubscriptionRequest>(SubscriptionKeyComparer.Instance);
private readonly ConcurrentDictionary<SubscriptionKey, SubscriptionProjectionState> _projections =
new ConcurrentDictionary<SubscriptionKey, SubscriptionProjectionState>(SubscriptionKeyComparer.Instance);
private readonly object _lifecycleGate = new();
@ -47,8 +48,10 @@ namespace GameCult.Networking
_authorizeRequest = authorizeRequest;
_authorizeRecord = authorizeRecord;
_subscribe = HandleSubscribeAsync;
_subscribeV1 = HandleSubscribeV1Async;
_unsubscribe = HandleUnsubscribeAsync;
_server.OnCultNet(_subscribe);
_server.OnCultNet(_subscribeV1);
_server.OnCultNet(_unsubscribe);
_peerLifecycle = server as ICultNetSchemaServerPeerLifecycle;
if (_peerLifecycle != null)
@ -86,6 +89,7 @@ namespace GameCult.Networking
if (_disposed) return;
_disposed = true;
_server.RemoveCultNetMessageListener<CultNetDatabaseSubscribeMessage>(_subscribe);
_server.RemoveCultNetMessageListener<CultNetDatabaseSubscribeV1Message>(_subscribeV1);
_server.RemoveCultNetMessageListener<CultNetDatabaseUnsubscribeMessage>(_unsubscribe);
if (_peerLifecycle != null)
_peerLifecycle.PeerDisconnected -= HandlePeerDisconnected;
@ -108,6 +112,33 @@ namespace GameCult.Networking
}
private Task HandleSubscribeAsync(CultNetDatabaseSubscribeMessage request, ICultNetSchemaServerPeer peer)
{
// v0 has no engine of its own (docs/cultnet-selection-cut.md, D3/D4): it lowers into a
// Selection and is matched by the one evaluator, same as a v1 subscribe.
var selection = new CultNetSelection
{
Schemas = request.SchemaIds,
Keys = request.RecordKeys,
Projection = CultNetSelectionProjections.Document
};
var subscriptionRequest = new SubscriptionRequest(
selection, request.MessageId, request.SubscriptionId, request.IncludeSnapshot,
request.ConsumerRuntimeId, request.BodyIds, request.SupportedBodyTransports, isV1: false);
return Subscribe(subscriptionRequest, peer);
}
private Task HandleSubscribeV1Async(CultNetDatabaseSubscribeV1Message request, ICultNetSchemaServerPeer peer)
{
var subscriptionRequest = new SubscriptionRequest(
request.Selection, request.MessageId, request.SubscriptionId, request.IncludeSnapshot,
request.ConsumerRuntimeId, request.BodyIds, request.SupportedBodyTransports, isV1: true);
return Subscribe(subscriptionRequest, peer);
}
// R-A: the one subscribe core both v0 (lowered) and v1 selections run through - the door
// (validate), the watch (D6: reconcile on every change for a hop-bearing selection, the fast
// single-row path otherwise), the initial snapshot in each version's own wire shape, and demand.
private Task Subscribe(SubscriptionRequest request, ICultNetSchemaServerPeer peer)
{
var subscriptionId = string.IsNullOrWhiteSpace(request.SubscriptionId)
? request.MessageId
@ -117,7 +148,19 @@ namespace GameCult.Networking
peer.SendCultNet(new CultNetErrorMessage { Error = "Database subscription requires a subscriptionId or messageId." });
return Task.CompletedTask;
}
if (_authorizeRequest?.Invoke(request, peer) == false)
try
{
request.Selection.Validate(_database.Cache.Registry.AllDescriptors.ToArray());
}
catch (CultNetSelectionInvalidException ex)
{
peer.SendCultNet(new CultNetErrorMessage { Error = $"selection_invalid: {ex.Message}" });
return Task.CompletedTask;
}
var legacyRequest = request.ToLegacyShape(subscriptionId);
if (_authorizeRequest?.Invoke(legacyRequest, peer) == false)
{
peer.SendCultNet(new CultNetErrorMessage { Error = "Database subscription is not authorized for this peer." });
return Task.CompletedTask;
@ -138,19 +181,11 @@ namespace GameCult.Networking
var snapshot = CreateProjectedSnapshot(request, peer);
foreach (var entry in snapshot.BySourceRecordKey)
projection.DeliveredBySourceRecordKey[entry.Key] = entry.Value;
peer.SendCultNet(new CultNetSnapshotResponseRawMessage
{
MessageId = request.MessageId,
Documents = snapshot.Documents
});
SendSnapshot(peer, request, snapshot);
}
else
{
peer.SendCultNet(new CultNetSnapshotResponseRawMessage
{
MessageId = request.MessageId,
Documents = Array.Empty<CultNetRawDocumentRecord>()
});
SendSnapshot(peer, request, ProjectedSnapshot.Empty);
}
PublishDemand(request, key, active: true);
projection.DemandActive = true;
@ -164,6 +199,30 @@ namespace GameCult.Networking
return Task.CompletedTask;
}
private static void SendSnapshot(ICultNetSchemaServerPeer peer, SubscriptionRequest request, ProjectedSnapshot snapshot)
{
if (request.IsV1)
{
var wantDocument = request.Selection.Projection == CultNetSelectionProjections.Document;
peer.SendCultNet(new CultNetSnapshotResponseRawV1Message
{
MessageId = request.MessageId,
Matched = (uint)snapshot.Documents.Length,
Documents = wantDocument ? snapshot.Documents : null,
Headers = wantDocument ? null : snapshot.Documents.Select(CultNetRawDocumentHeader.FromRecord).ToArray(),
Edges = request.Selection.HasHop ? snapshot.Edges : null
});
}
else
{
peer.SendCultNet(new CultNetSnapshotResponseRawMessage
{
MessageId = request.MessageId,
Documents = snapshot.Documents
});
}
}
private Task HandleUnsubscribeAsync(CultNetDatabaseUnsubscribeMessage request, ICultNetSchemaServerPeer peer)
{
var subscriptionId = string.IsNullOrWhiteSpace(request.SubscriptionId)
@ -179,18 +238,18 @@ namespace GameCult.Networking
}
private void PublishDemand(
CultNetDatabaseSubscribeMessage request,
SubscriptionRequest request,
SubscriptionKey key,
bool active)
{
DemandChanged?.Invoke(new CultNetDatabaseSubscriptionDemand(
string.IsNullOrWhiteSpace(request.ConsumerRuntimeId) ? key.Id : request.ConsumerRuntimeId!,
key.Id,
(request.RecordKeys ?? Array.Empty<string>())
(request.Selection.Keys ?? Array.Empty<string>())
.Where(value => !string.IsNullOrWhiteSpace(value))
.Distinct(StringComparer.Ordinal)
.ToArray(),
(request.SchemaIds ?? Array.Empty<string>())
(request.Selection.Schemas ?? Array.Empty<string>())
.Where(value => !string.IsNullOrWhiteSpace(value))
.Distinct(StringComparer.Ordinal)
.ToArray(),
@ -214,18 +273,28 @@ namespace GameCult.Networking
}
private IDisposable Watch(
CultNetDatabaseSubscribeMessage request,
SubscriptionRequest request,
string subscriptionId,
ICultNetSchemaServerPeer peer,
SubscriptionKey key)
{
var legacyRequest = request.ToLegacyShape(subscriptionId);
return _database.WatchAllChanges().Subscribe(change =>
{
lock (_lifecycleGate)
{
if (_disposed || !_requests.ContainsKey(key) || !_projections.TryGetValue(key, out var projection))
return;
if (_authorizeRequest?.Invoke(request, peer) == false)
if (_authorizeRequest?.Invoke(legacyRequest, peer) == false)
{
Reconcile(key, request);
return;
}
// D6: cites/cited are set-dependent - a change to row B can change whether row A
// matches - so a hop-bearing selection runs the full diff-against-delivered
// reconcile on every change instead of the single-row fast path.
if (request.Selection.HasHop)
{
Reconcile(key, request);
return;
@ -243,10 +312,11 @@ namespace GameCult.Networking
});
}
private void Reconcile(SubscriptionKey key, CultNetDatabaseSubscribeMessage request)
private void Reconcile(SubscriptionKey key, SubscriptionRequest request)
{
if (!_projections.TryGetValue(key, out var projection)) return;
var authorized = _authorizeRequest?.Invoke(request, key.Peer) != false;
var legacyRequest = request.ToLegacyShape(key.Id);
var authorized = _authorizeRequest?.Invoke(legacyRequest, key.Peer) != false;
var next = authorized
? CreateProjectedSnapshot(request, key.Peer)
: ProjectedSnapshot.Empty;
@ -284,30 +354,40 @@ namespace GameCult.Networking
}
private ProjectedSnapshot CreateProjectedSnapshot(
CultNetDatabaseSubscribeMessage request,
SubscriptionRequest request,
ICultNetSchemaServerPeer peer)
{
var snapshot = _database.Documents.CreateRawSnapshotResponse(
var legacyRequest = request.ToLegacyShape(request.SubscriptionId);
// Both v0 (lowered) and v1 selections evaluate through SelectAll: one evaluation over the
// whole matching set (R-G), documents always (this projects the wire record regardless of
// the selection's own header/document projection - SendSnapshot applies that afterward).
var documentSelection = new CultNetSelection
{
Schemas = request.Selection.Schemas,
Keys = request.Selection.Keys,
Fields = request.Selection.Fields,
Cites = request.Selection.Cites,
Cited = request.Selection.Cited,
Projection = CultNetSelectionProjections.Document
};
var page = _database.Documents.SelectAll(
_database.Cache,
request.MessageId,
new CultNetSnapshotRequestMessage
{
MessageId = request.MessageId,
SchemaIds = request.SchemaIds,
RecordKeys = request.RecordKeys
});
documentSelection,
ordinalOf: (schemaId, key) => _database.LastWriteSequence(schemaId, key) ?? 0,
asOf: _database.CurrentAsOf(),
options: null);
// The snapshot is one row per record key by construction, so the duplicate-key throw this
// used to guard can no longer fire (docs/cultnet-selection-cut.md, S2-2); deleted rather
// than tested around.
var bySourceRecordKey = new Dictionary<string, CultNetRawDocumentRecord>(StringComparer.Ordinal);
foreach (var source in snapshot.Documents)
foreach (var source in page.Documents ?? Array.Empty<CultNetRawDocumentRecord>())
{
var sourceRecordKey = source.RecordKey;
if (_authorizeRecord?.Invoke(request, peer, sourceRecordKey, source.SchemaId) == false)
if (_authorizeRecord?.Invoke(legacyRequest, peer, sourceRecordKey, source.SchemaId) == false)
continue;
bySourceRecordKey[sourceRecordKey] = source;
}
return new ProjectedSnapshot(bySourceRecordKey);
return new ProjectedSnapshot(bySourceRecordKey, page.Edges ?? Array.Empty<CultNetEdge>());
}
private void ApplyProjectedChange(
@ -406,10 +486,12 @@ namespace GameCult.Networking
// One evaluator (docs/cultnet-selection-cut.md, D3/D4): a v0 subscribe request lowers into a
// Selection with no engine of its own, matched and projected exactly as a v1 request would be.
// Only called for a non-hop selection (Watch routes a hop-bearing one to Reconcile instead, per
// D6) - CultNetSelectionEvaluator.Matches refuses a hop-bearing selection on this fast path.
// Record-level authorization (_authorizeRecord) is not selection and stays layered on top.
private CultNetRawDocumentRecord? CreateMatchedRecord(
object change,
CultNetDatabaseSubscribeMessage request,
SubscriptionRequest request,
ICultNetSchemaServerPeer peer)
{
var changeType = change.GetType();
@ -423,15 +505,17 @@ namespace GameCult.Networking
var descriptor = _database.Cache.Registry.GetRequired(forDescriptor.GetType());
var selection = new CultNetSelection
{
Schemas = request.SchemaIds,
Keys = request.RecordKeys,
Schemas = request.Selection.Schemas,
Keys = request.Selection.Keys,
Fields = request.Selection.Fields,
Projection = CultNetSelectionProjections.Document
};
var binding = _database.Documents.GetByDocumentType(descriptor.DocumentType);
var effectiveSelection = CultNetSchemaAliasMatching.WithBindingSchemaAlias(selection, descriptor, binding?.SchemaId);
// R-M: the one binding-alias reconciler (CultNetDocumentRegistry.ExpandSchemaBindingAliases).
var effectiveSelection = _database.Documents.ExpandSchemaBindingAliases(selection);
if (!CultNetSelectionEvaluator.Matches(descriptor, key, document, effectiveSelection))
return null;
if (_authorizeRecord?.Invoke(request, peer, key.Value, descriptor.SchemaId) == false)
var legacyRequest = request.ToLegacyShape(request.SubscriptionId);
if (_authorizeRecord?.Invoke(legacyRequest, peer, key.Value, descriptor.SchemaId) == false)
return null;
if (kind == CultNetDatabaseChangeKind.Removed || document == null)
@ -440,6 +524,56 @@ namespace GameCult.Networking
return _database.Documents.ToRawRecord(descriptor, key, document, DateTimeOffset.UtcNow.ToString("O"));
}
// R-A: the one internal shape both a lowered v0 subscribe and a v1 subscribe carry through
// Subscribe/Watch/Reconcile/CreateProjectedSnapshot/CreateMatchedRecord/PublishDemand. The
// authorize/authorizeRecord delegates keep their existing public constructor contract (typed
// against the v0 message) rather than becoming a breaking API change in this fix batch;
// ToLegacyShape carries every field those delegates read (schemas/keys/consumer/body demand) -
// fields/cites/cited reach neither delegate under v0 today either, so nothing regresses.
private sealed class SubscriptionRequest
{
public SubscriptionRequest(
CultNetSelection selection,
string messageId,
string subscriptionId,
bool includeSnapshot,
string? consumerRuntimeId,
string[]? bodyIds,
string[]? supportedBodyTransports,
bool isV1)
{
Selection = selection;
MessageId = messageId;
SubscriptionId = subscriptionId;
IncludeSnapshot = includeSnapshot;
ConsumerRuntimeId = consumerRuntimeId;
BodyIds = bodyIds;
SupportedBodyTransports = supportedBodyTransports;
IsV1 = isV1;
}
public CultNetSelection Selection { get; }
public string MessageId { get; }
public string SubscriptionId { get; }
public bool IncludeSnapshot { get; }
public string? ConsumerRuntimeId { get; }
public string[]? BodyIds { get; }
public string[]? SupportedBodyTransports { get; }
public bool IsV1 { get; }
public CultNetDatabaseSubscribeMessage ToLegacyShape(string subscriptionId) => new()
{
MessageId = MessageId,
SubscriptionId = subscriptionId,
SchemaIds = Selection.Schemas,
RecordKeys = Selection.Keys,
IncludeSnapshot = IncludeSnapshot,
ConsumerRuntimeId = ConsumerRuntimeId,
BodyIds = BodyIds,
SupportedBodyTransports = SupportedBodyTransports
};
}
private sealed class SubscriptionProjectionState
{
public Dictionary<string, CultNetRawDocumentRecord> DeliveredBySourceRecordKey { get; } =
@ -449,15 +583,17 @@ namespace GameCult.Networking
private sealed class ProjectedSnapshot
{
public ProjectedSnapshot(Dictionary<string, CultNetRawDocumentRecord> bySourceRecordKey)
public ProjectedSnapshot(Dictionary<string, CultNetRawDocumentRecord> bySourceRecordKey, CultNetEdge[]? edges = null)
{
BySourceRecordKey = bySourceRecordKey;
Documents = bySourceRecordKey.Values.ToArray();
Edges = edges ?? Array.Empty<CultNetEdge>();
}
public static ProjectedSnapshot Empty { get; } =
new(new Dictionary<string, CultNetRawDocumentRecord>(StringComparer.Ordinal));
public Dictionary<string, CultNetRawDocumentRecord> BySourceRecordKey { get; }
public CultNetEdge[] Edges { get; }
public CultNetRawDocumentRecord[] Documents { get; }
}

120
src/GameCult.Networking/CultNetDocumentRegistry.cs
View file

@ -273,8 +273,8 @@ namespace GameCult.Networking
/// <summary>
/// Creates a raw snapshot response from the cache. v0 has no selector engine of its own
/// (docs/cultnet-selection-cut.md, D3/D4): the filter lowers into a <see cref="CultNetSelection"/>
/// and is answered by <see cref="CultNetSelectionEvaluator"/>, walking every page it returns so a
/// v0 caller still sees its whole matching set in one message.
/// and is answered by <see cref="CultNetSelectionEvaluator"/> in one evaluation (R-G), so a v0
/// caller sees its whole matching set in one message without re-filtering the cache per page.
/// </summary>
public CultNetSnapshotResponseRawMessage CreateRawSnapshotResponse(
CultCache cache,
@ -287,24 +287,15 @@ namespace GameCult.Networking
{
Schemas = CultNetV0SelectionLowering.Lower(filter?.SchemaIds),
Keys = CultNetV0SelectionLowering.Lower(filter?.RecordKeys),
Projection = CultNetSelectionProjections.Document,
Limit = CultNetSelectionEvaluator.LimitMax
Projection = CultNetSelectionProjections.Document
};
var documents = new List<CultNetRawDocumentRecord>();
string? cursor = null;
do
{
selection.Cursor = cursor;
var page = SelectPage(cache, selection, ordinalOf: static (_, _) => 0, asOf: 0, validate: false, options);
documents.AddRange(page.Documents ?? Array.Empty<CultNetRawDocumentRecord>());
cursor = page.Next;
} while (cursor != null);
var page = SelectAll(cache, selection, ordinalOf: static (_, _) => 0, asOf: 0, options);
return new CultNetSnapshotResponseRawMessage
{
MessageId = RequireNonEmpty(messageId, nameof(messageId)),
Documents = documents.ToArray()
Documents = page.Documents ?? Array.Empty<CultNetRawDocumentRecord>()
};
}
@ -326,7 +317,7 @@ namespace GameCult.Networking
if (selection == null) throw new ArgumentNullException(nameof(selection));
if (ordinalOf == null) throw new ArgumentNullException(nameof(ordinalOf));
var page = SelectPage(cache, selection, ordinalOf, asOf, validate: true, options);
var page = SelectPage(cache, selection, ordinalOf, asOf, options);
return new CultNetSnapshotResponseRawV1Message
{
MessageId = RequireNonEmpty(messageId, nameof(messageId)),
@ -349,13 +340,13 @@ namespace GameCult.Networking
CultNetSelection selection,
Func<string, CultRecordKey, long> ordinalOf,
ulong asOf,
bool validate,
CultNetDocumentMessageOptions? options,
Func<CultDocumentDescriptor, CultRecordKey, bool>? rowFilter = null)
{
selection = ExpandSchemaBindingAliases(selection);
if (validate)
selection.Validate(_documents.AllDescriptors.ToArray());
// R-F: CultNetSelectionEvaluator.Select validates first, every time - there is no separate
// validate flag here any more, because a public evaluation path that can skip the door is
// exactly what R-F closes.
var stored = cache.AllStoredDocuments;
if (rowFilter != null)
@ -370,7 +361,8 @@ namespace GameCult.Networking
return new CultNetSelectionPage
{
Matched = (uint)records.Length,
// R-G/C5: the selection's whole matching count, not this page's row count.
Matched = (uint)evaluation.TotalMatched,
AsOf = asOf,
Next = evaluation.NextCursor,
Documents = wantDocument ? records : null,
@ -379,12 +371,56 @@ namespace GameCult.Networking
};
}
/// <summary>
/// Evaluates a selection over every row the cache holds and projects every matched row - no
/// cursor, no 200-row cap. The v0 lowering and the shard-bounded snapshot both answer their
/// whole matching set in one message, so they call this once instead of paging through
/// <see cref="SelectPage"/> in a loop that re-filters and re-sorts the same row set per page
/// (R-G: "answer v0 and shard paging from one evaluation").
/// </summary>
internal CultNetSelectionPage SelectAll(
CultCache cache,
CultNetSelection selection,
Func<string, CultRecordKey, long> ordinalOf,
ulong asOf,
CultNetDocumentMessageOptions? options,
Func<CultDocumentDescriptor, CultRecordKey, bool>? rowFilter = null)
{
selection = ExpandSchemaBindingAliases(selection);
var stored = cache.AllStoredDocuments;
if (rowFilter != null)
stored = stored.Where(entry => rowFilter(entry.Descriptor, entry.Key));
var rows = stored
.Select(entry => new CultNetSelectionEvaluator.Row(
entry.Descriptor, entry.Key, entry.Document, ordinalOf(entry.Descriptor.SchemaId, entry.Key), entry.StoredAt))
.ToArray();
var full = CultNetSelectionEvaluator.EvaluateAll(_documents, rows, selection);
var records = full.Ordered.Select(row => ToRawRecord(row, options)).ToArray();
var wantDocument = selection.Projection == CultNetSelectionProjections.Document;
var edges = selection.HasHop
? CultNetSelectionEvaluator.EdgesFor(full.Ordered, full).Select(edge => ToEdge(edge, wantDocument)).ToArray()
: null;
return new CultNetSelectionPage
{
Matched = (uint)full.Ordered.Count,
AsOf = asOf,
Next = null,
Documents = wantDocument ? records : null,
Headers = wantDocument ? null : records.Select(CultNetRawDocumentHeader.FromRecord).ToArray(),
Edges = edges
};
}
// A CultDocumentDescriptor knows nothing of CultNetDocumentBinding's optional schema-id
// override; a caller filtering by that wire id (rather than the descriptor's own) needs the
// descriptor's own id present in the selection too, since CultNetSelectionEvaluator only ever
// reads descriptors. Mirrors CultNetSchemaAliasMatching.WithBindingSchemaAlias, at the layer
// that owns every binding rather than one row's.
private CultNetSelection ExpandSchemaBindingAliases(CultNetSelection selection)
// reads descriptors. R-M: this is the one copy - both CultNetDatabaseServer's and
// CultNetDatabaseSubscriptionServer's single-change fast path call this too instead of carrying
// their own CultNetSchemaAliasMatching.WithBindingSchemaAlias (deleted).
internal CultNetSelection ExpandSchemaBindingAliases(CultNetSelection selection)
{
if (selection.Schemas is not { Length: > 0 } || _bindingsByType.Count == 0)
return selection;
@ -459,21 +495,31 @@ namespace GameCult.Networking
private CultNetEdge ToEdge(CultNetSelectionEvaluator.EdgeMatch edge, bool wantDocument)
{
// R-E: ToRawRecord and ToEdge emit the same id for the same row - the wire schema id a
// binding overrides to (WireSchemaId), never the descriptor's own id when the two differ.
var wire = new CultNetEdge
{
From = new CultNetRecordRef { SchemaId = edge.From.Descriptor.SchemaId, RecordKey = edge.From.Key.Value },
From = new CultNetRecordRef { SchemaId = WireSchemaId(edge.From.Descriptor), RecordKey = edge.From.Key.Value },
Role = edge.Role,
To = new CultNetRecordRef { SchemaId = edge.To.Descriptor.SchemaId, RecordKey = edge.To.Key.Value }
To = new CultNetRecordRef { SchemaId = WireSchemaId(edge.To.Descriptor), RecordKey = edge.To.Key.Value }
};
if (wantDocument && edge.Payload != null)
{
wire.PayloadEncoding = "messagepack";
wire.Payload = CultDocumentMessagePackSerialization.SerializeUntyped(edge.Payload, edge.Payload.GetType(), _documents);
// A dictionary reference's payload is typed by the member's declared value type (D11) -
// routinely a plain value like float, not a registered CultDocument - so this serializes
// it directly rather than through CultDocumentMessagePackSerialization.SerializeUntyped,
// which requires the type to be a registered document and throws otherwise.
wire.Payload = MessagePackSerializer.Serialize(edge.Payload.GetType(), edge.Payload, CultNetSchemaMessageSerialization.Options);
}
return wire;
}
/// <summary>The wire schema id this registry answers with for a descriptor - its binding's override when one is registered, else the descriptor's own id (R-E: the same id <see cref="ToRawRecord(CultNetSelectionEvaluator.Row, CultNetDocumentMessageOptions?)"/> emits).</summary>
private string WireSchemaId(CultDocumentDescriptor descriptor) =>
GetByDocumentType(descriptor.DocumentType)?.SchemaId ?? descriptor.SchemaId;
/// <summary>
/// Applies a raw document put message to a cache.
/// </summary>
@ -640,16 +686,28 @@ namespace GameCult.Networking
.Select(group => group.First());
}
/// <summary>
/// True when a raw payload's embedded <c>schemaVersion</c> stamp (when it carries one) aliases
/// the given descriptor - the "foreign schema id" fallback shared by CultMesh's peer-snapshot
/// read and its typed-document decode (docs/cultnet-selection-cut.md, R-L: one registry method
/// taking the descriptor, replacing two call sites that each read the payload and alias-matched
/// it by hand). <see cref="TryReadSchemaVersion"/> stays internal: a caller with an already-known
/// descriptor never needs the raw schema-version string itself, only this answer.
/// </summary>
public static bool PayloadMatchesSchema(byte[] payload, CultDocumentDescriptor descriptor) =>
TryReadSchemaVersion(payload) is { } schemaVersion &&
CultNetSchemaAliasMatching.Matches(schemaVersion, descriptor);
/// <summary>
/// Reads a raw payload's embedded <c>schemaVersion</c> stamp, when it carries one - the parse
/// step of the registry's payload-schema decode rule, shared with
/// <see cref="TryResolveDescriptorByPayloadSchema"/> and every "foreign schema id" fallback in
/// CultNet and CultMesh (docs/cultnet-selection-cut.md, section 4/S2-5). A caller matching
/// against one already-known descriptor (rather than resolving "whichever registered type owns
/// this schema string", which is ambiguous when two document types alias the same schema id)
/// should alias-match this string with <see cref="CultNetSchemaAliasMatching"/> directly.
/// <see cref="TryResolveDescriptorByPayloadSchema"/> and <see cref="PayloadMatchesSchema"/>
/// (docs/cultnet-selection-cut.md, section 4/S2-5, R-L). Internal: a caller outside this
/// assembly matching against one already-known descriptor uses <see cref="PayloadMatchesSchema"/>;
/// resolving "whichever registered type owns this schema string" (ambiguous when two document
/// types alias the same schema id) is <see cref="TryResolveDescriptorByPayloadSchema"/>'s job.
/// </summary>
public static string? TryReadSchemaVersion(byte[] payload)
internal static string? TryReadSchemaVersion(byte[] payload)
{
try
{

24
src/GameCult.Networking/CultNetSelection.cs
View file

@ -70,8 +70,12 @@ namespace GameCult.Networking
/// <summary>
/// Forbids leading zeros, a trailing fractional zero or bare point, an explicit '+', exponent
/// notation, and whitespace. Does not by itself forbid "-0"; <see cref="IsCanonical"/> does.
/// Anchored with <c>\z</c>, not <c>$</c>: .NET's <c>$</c> matches immediately before a single
/// trailing newline even without <see cref="System.Text.RegularExpressions.RegexOptions.Multiline"/>,
/// so a naive <c>$</c> here would accept "5\n" as canonical (R-D/Q-J; the JSON schema's ECMA-262
/// <c>$</c> has no such exception and already refused it).
/// </summary>
public const string Pattern = @"^-?(0|[1-9][0-9]*)(\.[0-9]*[1-9])?$";
public const string Pattern = @"\A-?(0|[1-9][0-9]*)(\.[0-9]*[1-9])?\z";
private static readonly Regex CanonicalRegex = new(Pattern, RegexOptions.Compiled | RegexOptions.CultureInvariant);
@ -318,10 +322,17 @@ namespace GameCult.Networking
/// </summary>
public static class CultNetSelectionValidation
{
public static void Validate(this CultNetSelection selection, IReadOnlyList<CultDocumentDescriptor> allDescriptors)
/// <summary>
/// The declaration-independent half of the door: empty/blank <c>schemas</c>/<c>keys</c> and an
/// unrecognised <c>projection</c>. Runs first inside <see cref="Validate"/>, and is also the
/// whole door for a caller with no descriptor list in hand - CultMesh's
/// <c>EnsureV0Compatible</c> (docs/cultnet-selection-cut.md, R-F) runs this before its own v0
/// transport refusals, so <c>Keys=[]</c>/<c>[""]</c> is refused there too, not only in the
/// reference runtime's full evaluator path.
/// </summary>
public static void ValidateShape(CultNetSelection selection)
{
if (selection == null) throw new ArgumentNullException(nameof(selection));
if (allDescriptors == null) throw new ArgumentNullException(nameof(allDescriptors));
if (selection.Schemas != null && selection.Schemas.Length == 0)
throw new CultNetSelectionInvalidException("schemas", null, "selection.schemas is present and empty; omit it to reach every schema.");
@ -345,6 +356,13 @@ namespace GameCult.Networking
}
if (selection.Projection is not (CultNetSelectionProjections.Header or CultNetSelectionProjections.Document))
throw new CultNetSelectionInvalidException("projection", selection.Projection, $"selection.projection \"{selection.Projection}\" is neither \"header\" nor \"document\".");
}
public static void Validate(this CultNetSelection selection, IReadOnlyList<CultDocumentDescriptor> allDescriptors)
{
if (allDescriptors == null) throw new ArgumentNullException(nameof(allDescriptors));
ValidateShape(selection);
var reachable = ReachableSchemas(selection, allDescriptors);
if (selection.Fields != null)

259
src/GameCult.Networking/CultNetSelectionEvaluator.cs
View file

@ -63,6 +63,8 @@ namespace GameCult.Networking
public IReadOnlyList<Row> Rows { get; set; } = Array.Empty<Row>();
public IReadOnlyList<EdgeMatch> Edges { get; set; } = Array.Empty<EdgeMatch>();
public string? NextCursor { get; set; }
/// <summary>The selection's whole matching count, not this page's row count (R-G/C5).</summary>
public int TotalMatched { get; set; }
}
/// <summary>
@ -96,10 +98,15 @@ namespace GameCult.Networking
return false;
if (op == CultNetSelectionOperator.AnyOf)
{
if (!descriptor.TryGetIndexValue(document, field.Index, out var value) ||
value == null ||
field.Values == null ||
!field.Values.Contains(value, StringComparer.Ordinal))
// R-J: on a numeric alias, any_of compares the canonical rendering
// (TryGetIndexNumber), not the string getter's culture-dependent ToString() - the
// same rendering a comparison operator uses, so a numeric member's any_of and its
// comparisons agree on what the row's value spells.
var isNumeric = descriptor.DeclaredMembers.Any(member => member.IndexAlias == field.Index && member.IsNumeric);
var value = isNumeric
? (descriptor.TryGetIndexNumber(document, field.Index, out var numeric) ? numeric : null)
: (descriptor.TryGetIndexValue(document, field.Index, out var raw) ? raw : null);
if (value == null || field.Values == null || !field.Values.Contains(value, StringComparer.Ordinal))
return false;
}
else
@ -131,46 +138,100 @@ namespace GameCult.Networking
};
}
/// <summary>One evaluation's full matching, ordered row set, with the edges its hop traversed anchored to the row that carries each one.</summary>
public sealed class FullEvaluation
{
/// <summary>Every matching row, in order, with no cursor or limit applied.</summary>
public IReadOnlyList<Row> Ordered { get; set; } = Array.Empty<Row>();
/// <summary>Every edge the hop traversed, each paired with the page-eligible row it is reported against: the citer under <c>cites</c>, the cited row under <c>cited</c> (R-B).</summary>
public IReadOnlyList<(EdgeMatch Edge, Row Anchor)> Edges { get; set; } = Array.Empty<(EdgeMatch, Row)>();
}
/// <summary>
/// Evaluates a selection over the full row set: schemas/keys/fields, the hop (cites/cited,
/// with its edges and the reference_outside_target refusal), order, cursor and paging. The
/// caller projects the result to wire records (<see cref="CultNetDocumentRegistry"/> owns that).
/// Evaluates a selection's full matching, ordered row set and the edges its hop traversed, with
/// no cursor or limit applied - the door (R-F), schemas/keys/fields, the hop (cites/cited, with
/// the reference_outside_target refusal) and order all run exactly once here. This is the one
/// evaluation both a single v1 page (<see cref="Select"/>) and a v0/shard full walk page over,
/// instead of each page re-filtering and re-sorting the whole row set (R-G).
/// </summary>
public static Evaluation Select(
public static FullEvaluation EvaluateAll(
CultDocumentRegistry registry,
IReadOnlyList<Row> allRows,
CultNetSelection selection,
ulong asOf)
CultNetSelection selection)
{
if (registry == null) throw new ArgumentNullException(nameof(registry));
if (allRows == null) throw new ArgumentNullException(nameof(allRows));
if (selection == null) throw new ArgumentNullException(nameof(selection));
// R-F: the door is inside select. This is the one public entry point that evaluates a
// selection over a full row set, so validation cannot be skipped by a caller that forgets
// to gate it - a public entry point that answers an unvalidated selection does not exist.
selection.Validate(registry.AllDescriptors.ToArray());
var byKey = new Dictionary<string, Row>(StringComparer.Ordinal);
foreach (var row in allRows)
byKey[row.Key.Value] = row;
var candidates = allRows.Where(row => MatchesSchemaKeysFields(row.Descriptor, row.Key, row.Document, selection));
var edges = new List<EdgeMatch>();
// R-B: cites and cited are opposite hop directions, and each keeps its own edge sink so an
// edge is always anchored to the row it will be reported against - the citer under cites,
// the cited row under cited - never filtered by "From is on the page" regardless of which
// direction the hop ran.
var citedEdges = new List<EdgeMatch>();
if (selection.Cited != null)
{
var incoming = BuildIncomingIndex(registry, byKey, selection.Cited.Role, edges);
var incoming = BuildIncomingIndex(registry, byKey, selection.Cited.Role, citedEdges);
var exists = selection.Cited.Exists;
candidates = candidates.Where(row => incoming.Contains(row.Key.Value) == exists);
}
var citesEdges = new List<EdgeMatch>();
if (selection.Cites != null)
{
candidates = candidates.Where(row => MatchesCitation(registry, byKey, row, selection.Cites, edges));
candidates = candidates.Where(row => MatchesCitation(registry, byKey, row, selection.Cites, citesEdges));
}
var matched = candidates.ToArray();
var ordered = (selection.Descending
? matched.OrderByDescending(row => row.Ordinal).ThenByDescending(row => row.Descriptor.SchemaId, StringComparer.Ordinal).ThenByDescending(row => row.Key.Value, StringComparer.Ordinal)
: matched.OrderBy(row => row.Ordinal).ThenBy(row => row.Descriptor.SchemaId, StringComparer.Ordinal).ThenBy(row => row.Key.Value, StringComparer.Ordinal))
? matched.OrderByDescending(row => row.Ordinal).ThenByDescending(row => row.Descriptor.SchemaId, CultNetCodePointComparer.Instance).ThenByDescending(row => row.Key.Value, CultNetCodePointComparer.Instance)
: matched.OrderBy(row => row.Ordinal).ThenBy(row => row.Descriptor.SchemaId, CultNetCodePointComparer.Instance).ThenBy(row => row.Key.Value, CultNetCodePointComparer.Instance))
.ToArray();
var edges = new List<(EdgeMatch, Row)>(citesEdges.Count + citedEdges.Count);
foreach (var edge in citesEdges) edges.Add((edge, edge.From));
foreach (var edge in citedEdges) edges.Add((edge, edge.To));
return new FullEvaluation { Ordered = ordered, Edges = edges };
}
/// <summary>
/// Evaluates a selection over the full row set and returns one cursor/limit page of it
/// (docs/cultnet-selection-cut.md, section 2/6). <see cref="Evaluation.TotalMatched"/> is the
/// selection's whole matching count, not this page's length (R-G/C5). The caller projects the
/// result to wire records (<see cref="CultNetDocumentRegistry"/> owns that).
/// </summary>
public static Evaluation Select(
CultDocumentRegistry registry,
IReadOnlyList<Row> allRows,
CultNetSelection selection,
ulong asOf)
{
var full = EvaluateAll(registry, allRows, selection);
return Page(full, selection, asOf);
}
/// <summary>
/// Slices one cursor/limit page out of an already-evaluated, already-ordered row set (R-G): the
/// v0/shard full walk calls this once per 200-row chunk of a single <see cref="EvaluateAll"/>
/// result instead of re-evaluating per page.
/// </summary>
public static Evaluation Page(FullEvaluation full, CultNetSelection selection, ulong asOf)
{
if (full == null) throw new ArgumentNullException(nameof(full));
if (selection == null) throw new ArgumentNullException(nameof(selection));
var ordered = full.Ordered;
var startIndex = 0;
if (!string.IsNullOrEmpty(selection.Cursor))
{
@ -184,17 +245,39 @@ namespace GameCult.Networking
var limit = (int)Math.Clamp(selection.Limit ?? LimitMax, LimitMin, LimitMax);
var page = ordered.Skip(startIndex).Take(limit).ToArray();
var hasNext = startIndex + page.Length < ordered.Length;
var hasNext = startIndex + page.Length < ordered.Count;
var nextCursor = hasNext && page.Length > 0
? CultNetSelectionCursor.Mint(asOf, page[^1], selection)
: null;
var pageKeys = new HashSet<string>(page.Select(row => row.Key.Value), StringComparer.Ordinal);
var pageEdges = selection.HasHop
? edges.Where(edge => pageKeys.Contains(edge.From.Key.Value)).ToArray()
? EdgesFor(page, full)
: Array.Empty<EdgeMatch>();
return new Evaluation { Rows = page, Edges = pageEdges, NextCursor = nextCursor };
return new Evaluation { Rows = page, Edges = pageEdges, NextCursor = nextCursor, TotalMatched = ordered.Count };
}
// R-B: the order is deterministic - the page-row order, then (from, role, to) in code-point
// order. Grouping by the row an edge is anchored to (LINQ's OrderBy is stable, so the given
// rows' own order survives as the primary key) and breaking ties by the edge's own identity
// gives both without a second pass. Public because a v0/shard full walk (R-G, one evaluation
// answering every page) reports edges against its whole matched set, not one 200-row page.
public static EdgeMatch[] EdgesFor(IReadOnlyList<Row> rows, FullEvaluation full)
{
var pageRowIndex = new Dictionary<string, int>(StringComparer.Ordinal);
for (var i = 0; i < rows.Count; i++)
pageRowIndex[rows[i].Key.Value] = i;
return full.Edges
.Where(pair => pageRowIndex.ContainsKey(pair.Anchor.Key.Value))
.OrderBy(pair => pageRowIndex[pair.Anchor.Key.Value])
.ThenBy(pair => pair.Edge.From.Descriptor.SchemaId, CultNetCodePointComparer.Instance)
.ThenBy(pair => pair.Edge.From.Key.Value, CultNetCodePointComparer.Instance)
.ThenBy(pair => pair.Edge.Role, CultNetCodePointComparer.Instance)
.ThenBy(pair => pair.Edge.To.Descriptor.SchemaId, CultNetCodePointComparer.Instance)
.ThenBy(pair => pair.Edge.To.Key.Value, CultNetCodePointComparer.Instance)
.Select(pair => pair.Edge)
.ToArray();
}
private static int FindCursorPosition(IReadOnlyList<Row> ordered, CultNetSelectionCursor cursor, bool descending)
@ -216,8 +299,8 @@ namespace GameCult.Networking
{
var ordinal = row.Ordinal.CompareTo(cursor.Ordinal);
if (ordinal != 0) return ordinal;
var schema = string.CompareOrdinal(row.Descriptor.SchemaId, cursor.SchemaId);
return schema != 0 ? schema : string.CompareOrdinal(row.Key.Value, cursor.RecordKey);
var schema = CultNetCodePointComparer.Instance.Compare(row.Descriptor.SchemaId, cursor.SchemaId);
return schema != 0 ? schema : CultNetCodePointComparer.Instance.Compare(row.Key.Value, cursor.RecordKey);
}
// D9: a reference's target set is every registered leaf assignable to its declared target type.
@ -251,7 +334,10 @@ namespace GameCult.Networking
if (!byKey.TryGetValue(targetKey.Value, out var resolved))
continue;
EnsureWithinDeclaredTarget(registry, member.TargetType, citer, role, resolved);
if (resolved.Descriptor.SchemaId != citation.Target.SchemaId)
// R-E: one schema-identity rule everywhere - a cites target goes through the alias
// matcher, the same as `schemas`, instead of an exact CultDocumentDescriptor.SchemaId
// compare that a schema alias or version string could never satisfy.
if (!CultNetSchemaAliasMatching.Matches(citation.Target.SchemaId, resolved.Descriptor))
continue;
edgeSink.Add(new EdgeMatch(citer, role, resolved, payload));
found = true;
@ -386,24 +472,129 @@ namespace GameCult.Networking
return new CultNetSelectionCursor(asOf, ordinal, parts[2], parts[3], parts[4]);
}
/// <summary>A digest of the selection with cursor and limit cleared, so a cursor is bound to the selection that minted it.</summary>
/// <summary>
/// A digest of the selection with cursor and limit cleared, so a cursor is bound to the
/// selection that minted it. R-H: every string and list is length-prefixed, so no delimiter
/// choice can make two different selections collide - the prior delimiter-joined form digested
/// values ["a|b"] the same as ["a","b"], and keys ["a,b"] the same as ["a","b"].
/// </summary>
public static string ComputeDigest(CultNetSelection selection)
{
var canonical = string.Join(
"",
string.Join(",", (selection.Schemas ?? Array.Empty<string>()).OrderBy(v => v, StringComparer.Ordinal)),
string.Join(",", (selection.Keys ?? Array.Empty<string>()).OrderBy(v => v, StringComparer.Ordinal)),
string.Join(";", (selection.Fields ?? Array.Empty<CultNetFieldPredicate>())
.Select(f => $"{f.Index}:{f.Op}:{string.Join("|", f.Values ?? Array.Empty<string>())}:{f.Number}")),
selection.Cites == null ? "" : $"{selection.Cites.Target.SchemaId}/{selection.Cites.Target.RecordKey}:{selection.Cites.Role}",
selection.Cited == null ? "" : $"{selection.Cited.Role}:{selection.Cited.Exists}",
selection.Projection,
selection.Descending.ToString());
var sb = new StringBuilder();
AppendList(sb, selection.Schemas);
AppendList(sb, selection.Keys);
var fields = selection.Fields ?? Array.Empty<CultNetFieldPredicate>();
sb.Append(fields.Length).Append(':');
foreach (var field in fields)
{
AppendString(sb, field.Index);
AppendString(sb, field.Op);
AppendList(sb, field.Values);
AppendString(sb, field.Number ?? string.Empty);
sb.Append(field.Number == null ? '0' : '1');
}
if (selection.Cites != null)
{
sb.Append('1');
AppendString(sb, selection.Cites.Target.SchemaId);
AppendString(sb, selection.Cites.Target.RecordKey);
AppendString(sb, selection.Cites.Role ?? string.Empty);
sb.Append(selection.Cites.Role == null ? '0' : '1');
}
else
{
sb.Append('0');
}
if (selection.Cited != null)
{
sb.Append('1');
AppendString(sb, selection.Cited.Role);
sb.Append(selection.Cited.Exists ? '1' : '0');
}
else
{
sb.Append('0');
}
AppendString(sb, selection.Projection);
sb.Append(selection.Descending ? '1' : '0');
using var sha = SHA256.Create();
var bytes = sha.ComputeHash(Encoding.UTF8.GetBytes(canonical));
var bytes = sha.ComputeHash(Encoding.UTF8.GetBytes(sb.ToString()));
var hex = new StringBuilder(bytes.Length * 2);
foreach (var b in bytes) hex.Append(b.ToString("x2", CultureInfo.InvariantCulture));
return hex.ToString();
}
/// <summary>Appends one length-prefixed string: unambiguous regardless of what the string itself contains.</summary>
private static void AppendString(StringBuilder sb, string value)
{
sb.Append(value.Length).Append(':').Append(value);
}
/// <summary>Appends a count-prefixed list of length-prefixed strings, sorted by code point so the digest does not depend on wire order.</summary>
private static void AppendList(StringBuilder sb, IReadOnlyList<string>? values)
{
var ordered = (values ?? Array.Empty<string>()).OrderBy(v => v, CultNetCodePointComparer.Instance).ToArray();
sb.Append(ordered.Length).Append(':');
foreach (var value in ordered)
AppendString(sb, value);
}
}
/// <summary>
/// Compares two strings by Unicode code point, not UTF-16 code unit (R-C, docs/cultnet-selection-cut.md).
/// .NET's ordinal comparers (StringComparer.Ordinal, string.CompareOrdinal) compare UTF-16 code
/// units, which sorts every astral character (encoded as a surrogate pair whose high half is in
/// U+D800-U+DBFF) before every BMP character at or above U+E000 - the opposite of code-point order,
/// and the opposite of what Rust gets by comparing UTF-8 bytes. Every ordered comparison the
/// selection vocabulary makes - the row tiebreak, the cursor position, and edge order - uses this
/// instead, so an astral key sorts the same way in both runtimes.
/// </summary>
public sealed class CultNetCodePointComparer : IComparer<string>
{
public static readonly CultNetCodePointComparer Instance = new();
private CultNetCodePointComparer() { }
public int Compare(string? x, string? y)
{
if (ReferenceEquals(x, y)) return 0;
if (x == null) return -1;
if (y == null) return 1;
var xi = 0;
var yi = 0;
while (xi < x.Length && yi < y.Length)
{
var xCodePoint = CodePointAt(x, xi, out var xSize);
var yCodePoint = CodePointAt(y, yi, out var ySize);
if (xCodePoint != yCodePoint)
return xCodePoint < yCodePoint ? -1 : 1;
xi += xSize;
yi += ySize;
}
return (x.Length - xi).CompareTo(y.Length - yi);
}
// A lone (unpaired) surrogate is not valid UTF-16, but a comparer must not throw on it - it is
// treated as its own code unit value. It never round-trips to the wire as anything but
// well-formed text, so this only has to not crash.
private static int CodePointAt(string value, int index, out int size)
{
var unit = value[index];
if (char.IsHighSurrogate(unit) && index + 1 < value.Length && char.IsLowSurrogate(value[index + 1]))
{
size = 2;
return char.ConvertToUtf32(unit, value[index + 1]);
}
size = 1;
return unit;
}
}
}

7
src/GameCult.Networking/Server.cs
View file

@ -567,7 +567,12 @@ namespace GameCult.Networking
or CultNetDocumentDeleteMessage
or CultNetDatabaseSubscribeMessage
or CultNetDatabaseUnsubscribeMessage
or CultNetSnapshotRequestMessage;
or CultNetSnapshotRequestMessage
// R-A: v1 requests reach the same pre-verification handling as their v0 counterparts -
// otherwise a v1 snapshot/subscribe from an unverified peer is silently dropped before
// any listener runs, unlike v0 (docs/cultnet-selection-cut.md, Soul's whole-cut pass F1).
or CultNetSnapshotRequestV1Message
or CultNetDatabaseSubscribeV1Message;
}
/// <inheritdoc />