Garden path: remote writes need a granting authority (cut remote-write-authority)

Merges hands/garden-path-remote-write-authority (51ca81a1..1e73e206).
cultmesh-garden-path:cut_report:cut-remote-write-authority.h1/h2; Soul s1 and s2 hold.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-28 02:07:44 +01:00
commit a924512e01
15 changed files with 687 additions and 35 deletions

View file

@ -111,8 +111,9 @@ Not implemented yet:
not guess. Adding a code is therefore a coordinated change across every runtime, made together: the schema (both
copies), the C# `CultNetErrorMessage` factory, the Rust `CultNetErrorCode`, and a captured-bytes case in
`packages/cultnet-rs/tests/error_message.rs` for it. TypeScript reads the schema. The current codes are
`selection_invalid`, `cursor_stale`, `cursor_invalid`, `reference_outside_target`, `variant_unsupported` and
`unowned_schema` (a write, or a remote put or delete, for a schema and key no shard owns).
`selection_invalid`, `cursor_stale`, `cursor_invalid`, `reference_outside_target`, `variant_unsupported`,
`unowned_schema` (a write, or a remote put or delete, for a schema and key no shard owns) and
`write_not_authorized` (a peer's put or delete its server's remote write authority did not grant).
## Live Invariants
@ -122,6 +123,9 @@ copies), the C# `CultNetErrorMessage` factory, the Rust `CultNetErrorCode`, and
- One primary owns writes for a shard until a stronger policy exists.
- Non-owner writes are rejected or explicitly forwarded. They are not silently
applied.
- A peer's put or delete is applied or forwarded only when the server's
`RemoteWriteAuthority` grants it; none configured refuses every remote write.
The connection key admits a peer to the transport and is never a grant.
- Raw snapshot/put/delete messages must pass through `CultNetDatabase`, not
bypass shard policy.
- The accepted shard log is stored in replica catch-up wire form. Object logs

View file

@ -28,7 +28,7 @@
"code": {
"description": "Machine-readable refusal code (R-N). Additive: a v0 peer that does not know this field ignores it. The set is a closed enum: Rust and TypeScript refuse a message whose code is not listed, so adding a code is one coordinated change across every runtime (this schema, the C# factory, the Rust CultNetErrorCode, and a captured-bytes case for each).",
"type": ["string", "null"],
"enum": [null, "selection_invalid", "cursor_stale", "cursor_invalid", "reference_outside_target", "variant_unsupported", "unowned_schema"]
"enum": [null, "selection_invalid", "cursor_stale", "cursor_invalid", "reference_outside_target", "variant_unsupported", "unowned_schema", "write_not_authorized"]
},
"details": {
"description": "Code-specific structured detail (R-N). Every field is populated only for the code that defines it; the others are null, not omitted.",
@ -36,8 +36,8 @@
"additionalProperties": false,
"required": ["field", "value", "asOf", "current"],
"properties": {
"field": { "description": "selection_invalid: the selection field that failed. variant_unsupported: \"subscriptionId\", when a subscription was ended. unowned_schema: \"schemaId\".", "type": ["string", "null"] },
"value": { "description": "selection_invalid: the offending value, when there is one string worth naming. variant_unsupported: the id of the subscription that ended. unowned_schema: the schema id no shard owns.", "type": ["string", "null"] },
"field": { "description": "selection_invalid: the selection field that failed. variant_unsupported: \"subscriptionId\", when a subscription was ended. unowned_schema: \"schemaId\". write_not_authorized: \"recordKey\".", "type": ["string", "null"] },
"value": { "description": "selection_invalid: the offending value, when there is one string worth naming. variant_unsupported: the id of the subscription that ended. unowned_schema: the schema id no shard owns. write_not_authorized: the key of the refused write.", "type": ["string", "null"] },
"asOf": { "description": "cursor_stale: the cursor's own minted asOf.", "type": ["integer", "null"], "minimum": 0 },
"current": { "description": "cursor_stale: the answering server's current asOf.", "type": ["integer", "null"], "minimum": 0 }
}

View file

@ -28,7 +28,7 @@
"code": {
"description": "Machine-readable refusal code (R-N). Additive: a v0 peer that does not know this field ignores it. The set is a closed enum: Rust and TypeScript refuse a message whose code is not listed, so adding a code is one coordinated change across every runtime (this schema, the C# factory, the Rust CultNetErrorCode, and a captured-bytes case for each).",
"type": ["string", "null"],
"enum": [null, "selection_invalid", "cursor_stale", "cursor_invalid", "reference_outside_target", "variant_unsupported", "unowned_schema"]
"enum": [null, "selection_invalid", "cursor_stale", "cursor_invalid", "reference_outside_target", "variant_unsupported", "unowned_schema", "write_not_authorized"]
},
"details": {
"description": "Code-specific structured detail (R-N). Every field is populated only for the code that defines it; the others are null, not omitted.",
@ -36,8 +36,8 @@
"additionalProperties": false,
"required": ["field", "value", "asOf", "current"],
"properties": {
"field": { "description": "selection_invalid: the selection field that failed. variant_unsupported: \"subscriptionId\", when a subscription was ended. unowned_schema: \"schemaId\".", "type": ["string", "null"] },
"value": { "description": "selection_invalid: the offending value, when there is one string worth naming. variant_unsupported: the id of the subscription that ended. unowned_schema: the schema id no shard owns.", "type": ["string", "null"] },
"field": { "description": "selection_invalid: the selection field that failed. variant_unsupported: \"subscriptionId\", when a subscription was ended. unowned_schema: \"schemaId\". write_not_authorized: \"recordKey\".", "type": ["string", "null"] },
"value": { "description": "selection_invalid: the offending value, when there is one string worth naming. variant_unsupported: the id of the subscription that ended. unowned_schema: the schema id no shard owns. write_not_authorized: the key of the refused write.", "type": ["string", "null"] },
"asOf": { "description": "cursor_stale: the cursor's own minted asOf.", "type": ["integer", "null"], "minimum": 0 },
"current": { "description": "cursor_stale: the answering server's current asOf.", "type": ["integer", "null"], "minimum": 0 }
}

View file

@ -246,6 +246,7 @@ pub enum CultNetErrorCode {
ReferenceOutsideTarget,
VariantUnsupported,
UnownedSchema,
WriteNotAuthorized,
}
impl CultNetErrorCode {
@ -257,6 +258,7 @@ impl CultNetErrorCode {
Self::ReferenceOutsideTarget => "reference_outside_target",
Self::VariantUnsupported => "variant_unsupported",
Self::UnownedSchema => "unowned_schema",
Self::WriteNotAuthorized => "write_not_authorized",
}
}
@ -268,6 +270,7 @@ impl CultNetErrorCode {
"reference_outside_target" => Ok(Self::ReferenceOutsideTarget),
"variant_unsupported" => Ok(Self::VariantUnsupported),
"unowned_schema" => Ok(Self::UnownedSchema),
"write_not_authorized" => Ok(Self::WriteNotAuthorized),
_ => Err(anyhow!("ErrorMessage.Code is not a recognized cultnet.error.v0 code")),
}
}

View file

@ -158,6 +158,21 @@ fn cases() -> Vec<Case> {
}),
},
},
Case {
name: "write_not_authorized",
// C#: ForRemoteWriteRefused(new CultNetRemoteWriteRequest(null, Put, "some.schema.v1", new CultRecordKey("some:key"), null, null), "not granted")
csharp_hex: "85AD736368656D6156657273696F6EB063756C746E65742E6572726F722E7630A56572726F72D92177726974655F6E6F745F617574686F72697A65643A206E6F74206772616E746564AB726F7574696E6748696E74C0A4636F6465B477726974655F6E6F745F617574686F72697A6564A764657461696C7384A56669656C64A97265636F72644B6579A576616C7565A8736F6D653A6B6579A461734F66C0A763757272656E74C0",
message: || CultNetMessage::Error {
error: "write_not_authorized: not granted".to_string(),
code: Some(CultNetErrorCode::WriteNotAuthorized),
details: Some(CultNetErrorDetails {
field: Some("recordKey".to_string()),
value: Some("some:key".to_string()),
as_of: None,
current: None,
}),
},
},
]
}

View file

@ -1015,11 +1015,16 @@ namespace GameCult.Networking
}
/// <summary>
/// Applies a raw document mutation after checking shard authority.
/// Applies a raw document mutation after checking shard authority. This is the host's own door (and replication's):
/// it asks no remote write authority. A peer's put goes through <see cref="CultNetDatabaseServer"/>, which does.
/// </summary>
/// <exception cref="CultNetUnownedSchemaException">No shard owns the schema and key; nothing was written.</exception>
/// <exception cref="CultNetShardLogException">The change committed and was published, but the shard log could not record it.</exception>
public async Task<object> ApplyPutAsync(CultNetDocumentPutRawMessage message)
public Task<object> ApplyPutAsync(CultNetDocumentPutRawMessage message) => ApplyPutAsync(message, authorize: null);
// The peer door: the same apply, with the server's authorize callback asked once the schema is resolved and before the
// shard is. The callback throws to refuse. The public overload is the host's own door and asks no remote authority.
internal async Task<object> ApplyPutAsync(CultNetDocumentPutRawMessage message, Action<string, CultRecordKey>? authorize)
{
ThrowIfDisposed();
if (message == null) throw new ArgumentNullException(nameof(message));
@ -1030,6 +1035,7 @@ namespace GameCult.Networking
var key = new CultRecordKey(message.Document.RecordKey);
var descriptor = _documents.ResolveDescriptorForRawDocument(message.Document);
authorize?.Invoke(descriptor.SchemaId, key);
var shard = ResolveShardInternal(descriptor, key);
EnsurePrimary(shard, descriptor.SchemaId, key, message.ShardEpoch);
var document = _documents.DeserializeRawDocument(message.Document);
@ -1039,17 +1045,22 @@ namespace GameCult.Networking
}
/// <summary>
/// Applies a raw document delete after checking shard authority.
/// Applies a raw document delete after checking shard authority. This is the host's own door: it asks no remote
/// write authority. A peer's delete goes through <see cref="CultNetDatabaseServer"/>, which does.
/// </summary>
/// <exception cref="CultNetUnownedSchemaException">No shard owns the schema and key; nothing was written.</exception>
/// <exception cref="CultNetShardLogException">The change committed and was published, but the shard log could not record it.</exception>
public Task ApplyDeleteAsync(CultNetDocumentDeleteMessage message)
public Task ApplyDeleteAsync(CultNetDocumentDeleteMessage message) => ApplyDeleteAsync(message, authorize: null);
// The peer door; see the put overload.
internal Task ApplyDeleteAsync(CultNetDocumentDeleteMessage message, Action<string, CultRecordKey>? authorize)
{
ThrowIfDisposed();
if (message == null) throw new ArgumentNullException(nameof(message));
var key = new CultRecordKey(message.RecordKey);
var descriptor = _documents.ResolveDescriptorForSchemaId(message.SchemaId);
authorize?.Invoke(descriptor.SchemaId, key);
var shard = ResolveShardInternal(descriptor, key);
EnsurePrimary(shard, descriptor.SchemaId, key, message.ShardEpoch);
var previous = _cache.Get(key);

View file

@ -67,22 +67,6 @@ namespace GameCult.Networking
/// </summary>
public CultNetDatabase Database => _database;
/// <summary>
/// Applies a raw put message through the database shard policy.
/// </summary>
public Task<object> ApplyPutAsync(CultNetDocumentPutRawMessage message)
{
return _database.ApplyPutAsync(message);
}
/// <summary>
/// Applies a raw delete message through the database shard policy.
/// </summary>
public Task ApplyDeleteAsync(CultNetDocumentDeleteMessage message)
{
return _database.ApplyDeleteAsync(message);
}
/// <summary>
/// Creates a raw snapshot response from the database cache.
/// </summary>
@ -304,11 +288,52 @@ namespace GameCult.Networking
return Task.CompletedTask;
}
// The one question asked of the write authority per received write. None configured refuses; an authority that throws
// refuses too (fail closed) and the fault is logged as an error.
private Action<string, CultRecordKey> Authorize(
CultNetServerPeer peer,
CultNetRemoteWriteOperation operation,
string? shardId,
long? shardEpoch)
{
return (schemaId, key) =>
{
var request = new CultNetRemoteWriteRequest(peer, operation, schemaId, key, shardId, shardEpoch);
var authority = _options.RemoteWriteAuthority;
if (authority == null)
{
throw new CultNetRemoteWriteRefusedException(request, "no remote write authority is configured");
}
CultNetRemoteWriteDecision decision;
try
{
decision = authority.Decide(request);
}
catch (Exception ex)
{
_server.Logger.LogError($"CultNet remote write authority failed: {ex.Message}");
throw new CultNetRemoteWriteRefusedException(request, "the write authority failed");
}
if (decision == null || !decision.IsGranted)
{
throw new CultNetRemoteWriteRefusedException(request, decision?.Reason);
}
};
}
private async Task HandlePutAsync(CultNetDocumentPutRawMessage message, CultNetServerPeer peer)
{
try
{
await ApplyPutAsync(message).ConfigureAwait(false);
await _database.ApplyPutAsync(message, Authorize(peer, CultNetRemoteWriteOperation.Put, message.ShardId, message.ShardEpoch))
.ConfigureAwait(false);
}
catch (CultNetRemoteWriteRefusedException ex)
{
_server.Logger.LogWarning($"CultNet raw put refused: {ex.Message}");
peer.SendCultNet(CultNetErrorMessage.ForRemoteWriteRefused(ex.Request, ex.Reason));
}
catch (CultNetShardAuthorityException ex)
{
@ -339,7 +364,13 @@ namespace GameCult.Networking
{
try
{
await ApplyDeleteAsync(message).ConfigureAwait(false);
await _database.ApplyDeleteAsync(message, Authorize(peer, CultNetRemoteWriteOperation.Delete, message.ShardId, message.ShardEpoch))
.ConfigureAwait(false);
}
catch (CultNetRemoteWriteRefusedException ex)
{
_server.Logger.LogWarning($"CultNet raw delete refused: {ex.Message}");
peer.SendCultNet(CultNetErrorMessage.ForRemoteWriteRefused(ex.Request, ex.Reason));
}
catch (CultNetShardAuthorityException ex)
{

View file

@ -0,0 +1,113 @@
using System;
using GameCult.Caching;
namespace GameCult.Networking
{
/// <summary>
/// Decides whether a connected peer may put or delete a record in the database a
/// <see cref="CultNetDatabaseServer"/> serves. Transport admission (the connection key, a completed handshake) is
/// never a grant; this is the one owner of that decision.
/// </summary>
public interface ICultNetRemoteWriteAuthority
{
/// <summary>
/// Decides one peer write. Called once per received put or delete, after the schema is resolved and before the
/// shard is, so a refusal applies, publishes, logs and forwards nothing.
/// </summary>
CultNetRemoteWriteDecision Decide(CultNetRemoteWriteRequest request);
}
/// <summary>
/// The kind of write a peer asked for.
/// </summary>
public enum CultNetRemoteWriteOperation
{
/// <summary>Put a record.</summary>
Put,
/// <summary>Delete a record.</summary>
Delete
}
/// <summary>
/// One peer write, as the authority sees it.
/// </summary>
public sealed class CultNetRemoteWriteRequest
{
/// <summary>
/// Creates a request.
/// </summary>
public CultNetRemoteWriteRequest(
CultNetServerPeer peer,
CultNetRemoteWriteOperation operation,
string schemaId,
CultRecordKey recordKey,
string? shardId,
long? shardEpoch)
{
Peer = peer;
Operation = operation;
SchemaId = schemaId;
RecordKey = recordKey;
ShardId = shardId;
ShardEpoch = shardEpoch;
}
/// <summary>Gets the peer connection the message arrived on (endpoint and transport only).</summary>
public CultNetServerPeer Peer { get; }
/// <summary>Gets the operation.</summary>
public CultNetRemoteWriteOperation Operation { get; }
/// <summary>Gets the canonical schema id the database resolved for the write, never the alias the peer sent.</summary>
public string SchemaId { get; }
/// <summary>Gets the record key.</summary>
public CultRecordKey RecordKey { get; }
/// <summary>Gets the shard id as the peer sent it.</summary>
public string? ShardId { get; }
/// <summary>Gets the shard epoch as the peer sent it.</summary>
public long? ShardEpoch { get; }
}
/// <summary>
/// An authority's answer to one <see cref="CultNetRemoteWriteRequest"/>.
/// </summary>
public sealed class CultNetRemoteWriteDecision
{
private CultNetRemoteWriteDecision(bool isGranted, string? reason)
{
IsGranted = isGranted;
Reason = reason;
}
/// <summary>Gets whether the write may proceed.</summary>
public bool IsGranted { get; }
/// <summary>Gets the refusal reason, when there is one.</summary>
public string? Reason { get; }
/// <summary>Grants the write.</summary>
public static CultNetRemoteWriteDecision Grant() => new(true, null);
/// <summary>Refuses the write.</summary>
public static CultNetRemoteWriteDecision Refuse(string? reason = null) => new(false, reason);
}
// Thrown by the server's authorize callback through the database's remote apply when the authority refuses; the
// server's handlers are its one catcher.
internal sealed class CultNetRemoteWriteRefusedException : Exception
{
public CultNetRemoteWriteRefusedException(CultNetRemoteWriteRequest request, string? reason)
: base(reason ?? "the write was not authorized")
{
Request = request;
Reason = reason;
}
public CultNetRemoteWriteRequest Request { get; }
public string? Reason { get; }
}
}

View file

@ -511,7 +511,7 @@ namespace GameCult.Networking
[Key("routingHint")] public CultNetShardRoutingHint? RoutingHint { get; set; }
/// <summary>
/// Gets or sets the machine-readable refusal code (R-N), e.g. <c>selection_invalid</c>,
/// <c>cursor_stale</c>, <c>cursor_invalid</c>, <c>reference_outside_target</c>, <c>unowned_schema</c>. Additive to
/// <c>cursor_stale</c>, <c>cursor_invalid</c>, <c>reference_outside_target</c>, <c>unowned_schema</c>, <c>write_not_authorized</c>. Additive to
/// <c>cultnet.error.v0</c>; a v0 peer that does not know this field ignores it.
/// </summary>
[Key("code")] public string? Code { get; set; }
@ -543,6 +543,18 @@ namespace GameCult.Networking
Details = new CultNetErrorDetails { Field = "schemaId", Value = ex.SchemaId }
};
/// <summary>
/// Builds the wire error for a peer write its authority refused: code <c>write_not_authorized</c>, details
/// <c>{ field: "recordKey", value }</c>.
/// </summary>
public static CultNetErrorMessage ForRemoteWriteRefused(CultNetRemoteWriteRequest request, string? reason) =>
new CultNetErrorMessage
{
Error = string.IsNullOrWhiteSpace(reason) ? "write_not_authorized" : $"write_not_authorized: {reason}",
Code = "write_not_authorized",
Details = new CultNetErrorDetails { Field = "recordKey", Value = request.RecordKey.Value }
};
/// <summary>
/// Builds the wire error for a cursor refusal (R-N): code <c>cursor_stale</c> with details
/// <c>{ asOf, current }</c>, or code <c>cursor_invalid</c> with no details.

View file

@ -34,6 +34,13 @@ namespace GameCult.Networking
/// Gets or sets the forwarder used when non-primary write forwarding is enabled.
/// </summary>
public ICultNetShardWriteForwarder? WriteForwarder { get; set; }
/// <summary>
/// Gets or sets the authority asked for every peer put and delete. None configured refuses them all: a connection
/// key admits a peer to the transport, never to write. A write forwarded to a primary is judged here for the
/// original peer and again at the primary for the forwarder's connection.
/// </summary>
public ICultNetRemoteWriteAuthority? RemoteWriteAuthority { get; set; }
}
/// <summary>

View file

@ -197,7 +197,10 @@ reply.
The first coherent shard policy is primary ownership. Each shard has one
authoritative writer at a time. Clients may connect to any node; a non-owner
node forwards a write to the owner or rejects it honestly when forwarding is
not available. Followers subscribe to committed shard mutations and apply them
not available. A node applies or forwards a peer's write only after its
`RemoteWriteAuthority` grants it; with none configured it refuses all of them, and a
write forwarded to a primary is judged again there for the forwarder's connection.
Followers subscribe to committed shard mutations and apply them
through the same CultCache reconciliation path as local file changes.
This should feel like the local-first parts of Firebase or RethinkDB without

View file

@ -733,7 +733,7 @@ namespace GameCult.Networking.Tests
using var server = new Server(cache, ServerSecurityOptions.Development());
var logger = new CapturingLogger();
server.Logger = logger;
using var databaseServer = new CultNetDatabaseServer(server, database);
using var databaseServer = new CultNetDatabaseServer(server, database, new CultNetDatabaseServerOptions { RemoteWriteAuthority = new TestGrantsEverything() });
var handler = (Func<CultNetDocumentPutRawMessage, CultNetServerPeer, Task>)typeof(CultNetDatabaseServer)
.GetField("_putHandler", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance)!
.GetValue(databaseServer)!;
@ -929,7 +929,7 @@ namespace GameCult.Networking.Tests
var host = new Server(cache, ServerSecurityOptions.Development());
var logger = new CapturingLogger();
host.Logger = logger;
var databaseServer = new CultNetDatabaseServer(host, database);
var databaseServer = new CultNetDatabaseServer(host, database, new CultNetDatabaseServerOptions { RemoteWriteAuthority = new TestGrantsEverything() });
var peer = (CultNetServerPeer)System.Runtime.CompilerServices.RuntimeHelpers.GetUninitializedObject(typeof(CultNetServerPeer));
const System.Reflection.BindingFlags flags = System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance;
var put = (Func<CultNetDocumentPutRawMessage, CultNetServerPeer, Task>)typeof(CultNetDatabaseServer).GetField("_putHandler", flags)!.GetValue(databaseServer)!;

View file

@ -206,6 +206,20 @@ namespace GameCult.Networking.Tests
AssertErrorMatchesSchema(message);
}
[Test]
public void ErrorMessage_RemoteWriteRefusedRealBytes_MatchesItsSchema()
{
var request = new CultNetRemoteWriteRequest(null!, CultNetRemoteWriteOperation.Put, "some.schema.v1", new CultRecordKey("some:key"), null, null);
var message = CultNetErrorMessage.ForRemoteWriteRefused(request, "not granted");
Assert.That(message.Code, Is.EqualTo("write_not_authorized"));
Assert.That(message.Details!.Field, Is.EqualTo("recordKey"));
Assert.That(message.Details.Value, Is.EqualTo("some:key"));
AssertErrorMatchesSchema(message);
// The bytes cultnet-rs's error_message.rs write_not_authorized case decodes and re-encodes.
Assert.That(Convert.ToHexString(CultNetSchemaMessageSerialization.Serialize(message)), Is.EqualTo("85AD736368656D6156657273696F6EB063756C746E65742E6572726F722E7630A56572726F72D92177726974655F6E6F745F617574686F72697A65643A206E6F74206772616E746564AB726F7574696E6748696E74C0A4636F6465B477726974655F6E6F745F617574686F72697A6564A764657461696C7384A56669656C64A97265636F72644B6579A576616C7565A8736F6D653A6B6579A461734F66C0A763757272656E74C0"));
}
private static void AssertErrorMatchesSchema(CultNetErrorMessage message)
{
var validator = new MiniJsonSchemaValidator(SchemaDir());

View file

@ -0,0 +1,439 @@
#nullable enable
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Reflection;
using System.Runtime.CompilerServices;
using System.Threading.Tasks;
using GameCult.Caching;
using LiteNetLib;
using NUnit.Framework;
using R3;
using static GameCult.Networking.Tests.NetworkingTests;
namespace GameCult.Networking.Tests
{
// A peer's put or delete is applied or forwarded only when the server's RemoteWriteAuthority grants it. The wire handlers
// are reached through their private delegates with an uninitialized peer, as CultNetJournalTests.WireHandlers does:
// anything the handler tries to send throws on that peer, so a handler that throws answered the peer with a refusal and a
// handler that returns quietly sent nothing.
// The tests' own grant-everyone authority, for tests whose subject is not the grant. The library ships none.
internal sealed class TestGrantsEverything : ICultNetRemoteWriteAuthority
{
public CultNetRemoteWriteDecision Decide(CultNetRemoteWriteRequest request) => CultNetRemoteWriteDecision.Grant();
}
public sealed class RemoteWriteAuthorityTests
{
private const string ShardId = "authority-shard";
private static readonly CultRecordKey KeyA = new("tests:authority:a");
private static readonly CultRecordKey KeyB = new("tests:authority:b");
private static readonly CultRecordKey Unowned = new("unrelated:authority");
private static NetworkSchemaNote Note(string text) => new() { Schema = "tests.networking_note.v1", Text = text };
private sealed class Rig
{
public CultCache Cache = null!;
public CultNetDatabase Database = null!;
public CultNetDatabaseServer DatabaseServer = null!;
public Server Host = null!;
public CultNetServerPeer Peer = null!;
public CapturingLogger Logger = null!;
public Func<CultNetDocumentPutRawMessage, Task> Put = null!;
public Func<CultNetDocumentDeleteMessage, Task> Delete = null!;
public string SchemaId = string.Empty;
public List<object> Published = new();
public IDisposable Subscription = null!;
public string? TextOf(CultRecordKey key) => Cache.Get<NetworkSchemaNote>(key)?.Text;
public int LogLength => Database.GetMutationLog(ShardId).Count;
public CultNetDocumentPutRawMessage PutOf(CultRecordKey key, string text)
{
var message = Database.Documents.CreateRawDocumentPutMessage("wire", new CultRecordHandle<NetworkSchemaNote>(key), Note(text));
message.ShardId = ShardId;
message.ShardEpoch = 1;
return message;
}
public CultNetDocumentDeleteMessage DeleteOf(CultRecordKey key) =>
new() { MessageId = "wire-delete", SchemaId = SchemaId, RecordKey = key.Value, ShardId = ShardId, ShardEpoch = 1 };
}
private static Rig Build(ICultNetRemoteWriteAuthority? authority, bool primary = true, ICultNetShardWriteForwarder? forwarder = null, CultNetServerPeer? peer = null)
{
var cache = new CultCache();
var schemaId = cache.Registry.GetRequired<NetworkSchemaNote>().SchemaId;
var database = new CultNetDatabase(cache, new CultNetDatabaseOptions
{
Shards =
[
new CultNetShardDescriptor(ShardId, "primary", epoch: 1, isPrimary: primary, schemaIds: [schemaId],
primaryEndpoints: primary ? null : ["cultnet://runtime-owner:3075"])
]
});
var host = new Server(cache, ServerSecurityOptions.Development());
var logger = new CapturingLogger();
host.Logger = logger;
var databaseServer = new CultNetDatabaseServer(host, database, new CultNetDatabaseServerOptions
{
RemoteWriteAuthority = authority,
ForwardNonPrimaryWrites = forwarder != null,
WriteForwarder = forwarder
});
peer ??= (CultNetServerPeer)RuntimeHelpers.GetUninitializedObject(typeof(CultNetServerPeer));
const BindingFlags flags = BindingFlags.NonPublic | BindingFlags.Instance;
var put = (Func<CultNetDocumentPutRawMessage, CultNetServerPeer, Task>)typeof(CultNetDatabaseServer).GetField("_putHandler", flags)!.GetValue(databaseServer)!;
var delete = (Func<CultNetDocumentDeleteMessage, CultNetServerPeer, Task>)typeof(CultNetDatabaseServer).GetField("_deleteHandler", flags)!.GetValue(databaseServer)!;
var rig = new Rig
{
Cache = cache,
Database = database,
DatabaseServer = databaseServer,
Host = host,
Peer = peer,
Logger = logger,
SchemaId = schemaId,
Put = message => put(message, peer!),
Delete = message => delete(message, peer!)
};
rig.Subscription = database.WatchAllChanges().Subscribe(change => rig.Published.Add(change));
return rig;
}
// The handler answered the peer with a refusal: the send reached the (uninitialized) peer and failed there, so the
// refusal message was built first. A failure earlier, in building it, would not pass through the peer's send.
private static void AssertAnswered(Func<Task> handler, string because)
{
var failure = Assert.CatchAsync(async () => await handler(), because)!;
Assert.That(failure.ToString(), Does.Contain("CultNetServerPeer.SendCultNet").And.Not.Contain("ForRemoteWriteRefused"), because);
}
// The server writes its own record through the host's door, which asks no remote authority.
private static async Task Seed(Rig rig, CultRecordKey key, string text) => await rig.Database.ApplyPutAsync(rig.PutOf(key, text));
[Test]
public async Task WithNoAuthorityAPeersPutAndDeleteAreRefusedAndChangeNothing()
{
var rig = Build(authority: null);
await Seed(rig, KeyA, "server");
var publishedBefore = rig.Published.Count;
var logBefore = rig.LogLength;
Assert.That(rig.TextOf(KeyA), Is.EqualTo("server"), "the host's own door needs no remote authority");
AssertAnswered(() => rig.Put(rig.PutOf(KeyA, "peer")), "the peer is answered with a refusal");
AssertAnswered(() => rig.Delete(rig.DeleteOf(KeyA)), "the peer is answered with a refusal");
AssertAnswered(() => rig.Put(rig.PutOf(KeyB, "peer")), "a new key is refused too");
Assert.That(rig.TextOf(KeyA), Is.EqualTo("server"));
Assert.That(rig.TextOf(KeyB), Is.Null);
Assert.That(rig.Published, Has.Count.EqualTo(publishedBefore), "a refusal publishes no change");
Assert.That(rig.LogLength, Is.EqualTo(logBefore), "a refusal appends no mutation log entry");
Assert.That(rig.Logger.Warnings, Has.Some.Contains("raw put refused"));
Assert.That(rig.Logger.Warnings, Has.Some.Contains("raw delete refused"));
Assert.That(rig.Logger.Errors, Is.Empty, "a refusal is not a fault");
}
[Test]
public async Task AGrantedPutAndDeleteCommitAndTheAuthoritySawThisPeer()
{
var authority = new RecordingAuthority(_ => CultNetRemoteWriteDecision.Grant());
var rig = Build(authority);
await rig.Put(rig.PutOf(KeyA, "peer"));
Assert.That(rig.TextOf(KeyA), Is.EqualTo("peer"));
Assert.That(rig.Published, Has.Count.EqualTo(1));
Assert.That(rig.LogLength, Is.EqualTo(1));
await rig.Delete(rig.DeleteOf(KeyA));
Assert.That(rig.TextOf(KeyA), Is.Null);
Assert.That(authority.Requests, Has.Count.EqualTo(2), "asked once per received write");
Assert.That(authority.Requests.Select(request => request.Peer), Has.All.SameAs(rig.Peer));
Assert.That(rig.Logger.Errors, Is.Empty);
}
[Test]
public async Task TheRequestCarriesTheResolvedWriteAndScopedGrantsRefuseTheirNeighbours()
{
// The put names a foreign alias of the schema; the payload routes it to the canonical descriptor.
var recorded = new RecordingAuthority(request =>
request.Operation == CultNetRemoteWriteOperation.Put && request.RecordKey.Equals(KeyA)
? CultNetRemoteWriteDecision.Grant()
: CultNetRemoteWriteDecision.Refuse("scoped"));
var rig = Build(recorded);
await Seed(rig, KeyA, "server");
var aliased = rig.PutOf(KeyA, "aliased");
aliased.Document.SchemaId = "runtime.generated.network-note.ui.42";
await rig.Put(aliased);
var request = recorded.Requests.Single();
Assert.That(request.SchemaId, Is.EqualTo(rig.SchemaId), "the authority sees the canonical id, not the alias as sent");
Assert.That(request.Operation, Is.EqualTo(CultNetRemoteWriteOperation.Put));
Assert.That(request.RecordKey, Is.EqualTo(KeyA));
Assert.That(request.ShardId, Is.EqualTo(ShardId));
Assert.That(request.ShardEpoch, Is.EqualTo(1));
Assert.That(rig.TextOf(KeyA), Is.EqualTo("aliased"));
AssertAnswered(() => rig.Put(rig.PutOf(KeyB, "other-key")), "the peer is answered with a refusal");
AssertAnswered(() => rig.Delete(rig.DeleteOf(KeyA)), "the peer is answered with a refusal");
Assert.That(rig.TextOf(KeyB), Is.Null, "a grant scoped to key a refuses key b");
Assert.That(rig.TextOf(KeyA), Is.EqualTo("aliased"), "a grant scoped to Put refuses Delete");
Assert.That(recorded.Requests.Last().Operation, Is.EqualTo(CultNetRemoteWriteOperation.Delete));
Assert.That(recorded.Requests.Last().SchemaId, Is.EqualTo(rig.SchemaId));
}
[Test]
public async Task ARefusedWriteIsNeverForwardedAndAGrantedOneIs()
{
var forwarder = new CountingForwarder();
var refusing = Build(new RecordingAuthority(_ => CultNetRemoteWriteDecision.Refuse("no")), primary: false, forwarder);
AssertAnswered(() => refusing.Put(refusing.PutOf(KeyA, "peer")), "the peer is answered with a refusal");
AssertAnswered(() => refusing.Delete(refusing.DeleteOf(KeyA)), "the peer is answered with a refusal");
Assert.That(forwarder.Puts, Is.EqualTo(0), "a refused put is not forwarded");
Assert.That(forwarder.Deletes, Is.EqualTo(0), "a refused delete is not forwarded");
var unconfigured = Build(authority: null, primary: false, forwarder);
AssertAnswered(() => unconfigured.Put(unconfigured.PutOf(KeyA, "peer")), "the peer is answered with a refusal");
Assert.That(forwarder.Puts, Is.EqualTo(0), "no authority forwards nothing");
var granting = Build(new RecordingAuthority(_ => CultNetRemoteWriteDecision.Grant()), primary: false, forwarder);
await granting.Put(granting.PutOf(KeyA, "peer"));
await granting.Delete(granting.DeleteOf(KeyA));
Assert.That(forwarder.Puts, Is.EqualTo(1), "a granted write on a non-primary shard is forwarded as before");
Assert.That(forwarder.Deletes, Is.EqualTo(1));
}
[Test]
public void RefusalPrecedesTheUnownedSchemaCheck()
{
var unowned = new MeshQuickstartNote { NoteId = "u", Body = "unowned" };
var rig = Build(authority: null);
var put = rig.Database.Documents.CreateRawDocumentPutMessage("wire", new CultRecordHandle<MeshQuickstartNote>(Unowned), unowned);
put.ShardId = ShardId;
put.ShardEpoch = 1;
AssertAnswered(() => rig.Put(put), "the peer is answered with a refusal");
Assert.That(rig.Logger.Warnings, Has.Some.Contains("authority"), "an ungranted put of an unowned schema is answered write_not_authorized");
Assert.That(rig.Logger.Warnings, Has.None.Contains("No shard owns"));
var granting = Build(new RecordingAuthority(_ => CultNetRemoteWriteDecision.Grant()));
var grantedPut = granting.Database.Documents.CreateRawDocumentPutMessage("wire", new CultRecordHandle<MeshQuickstartNote>(Unowned), unowned);
grantedPut.ShardId = ShardId;
grantedPut.ShardEpoch = 1;
AssertAnswered(() => granting.Put(grantedPut), "the peer is answered with a refusal");
Assert.That(granting.Logger.Warnings, Has.Some.Contains("No shard owns"), "a granted put still meets the unowned-schema refusal");
}
[Test]
public async Task AnAuthorityThatThrowsRefusesAndIsLoggedAsAnError()
{
var rig = Build(new RecordingAuthority(_ => throw new InvalidOperationException("authority exploded")));
await Seed(rig, KeyA, "server");
AssertAnswered(() => rig.Put(rig.PutOf(KeyA, "peer")), "the peer is answered with a refusal");
AssertAnswered(() => rig.Delete(rig.DeleteOf(KeyA)), "the peer is answered with a refusal");
Assert.That(rig.TextOf(KeyA), Is.EqualTo("server"), "fail closed");
Assert.That(rig.Logger.Errors, Has.Some.Contains("authority exploded"));
}
[Test]
public void ARefusalWithNoReasonStillSaysWhy()
{
var rig = Build(new RecordingAuthority(_ => CultNetRemoteWriteDecision.Refuse()));
AssertAnswered(() => rig.Put(rig.PutOf(KeyA, "peer")), "refused");
Assert.That(rig.Logger.Warnings, Has.Some.Contains("raw put refused: the write was not authorized"));
}
[Test]
public void TheRefusalNamesTheKeyAndTheCode()
{
var request = new CultNetRemoteWriteRequest(null!, CultNetRemoteWriteOperation.Delete, "s", KeyA, null, null);
var message = CultNetErrorMessage.ForRemoteWriteRefused(request, "why not");
Assert.That(message.Code, Is.EqualTo("write_not_authorized"));
Assert.That(message.Error, Is.EqualTo("write_not_authorized: why not"));
Assert.That(message.Details!.Field, Is.EqualTo("recordKey"));
Assert.That(message.Details.Value, Is.EqualTo(KeyA.Value));
Assert.That(message.RoutingHint, Is.Null);
Assert.That(CultNetErrorMessage.ForRemoteWriteRefused(request, null).Error, Is.EqualTo("write_not_authorized"));
}
// A real LiteNetLib pair on loopback with both sockets bound to ephemeral ports (Server and Client bind fixed ports,
// which a parallel suite cannot share). The server side is a real NetPeer wrapped as the CultNetServerPeer the handlers
// answer; the client side decodes what arrives on the wire with the production schema decoder.
private sealed class WirePair : IDisposable
{
private readonly NetManager _serverManager;
private readonly NetManager _clientManager;
public CultNetServerPeer Peer { get; }
public ConcurrentQueue<CultNetErrorMessage> Errors { get; } = new();
public WirePair()
{
var serverListener = new EventBasedNetListener();
_serverManager = new NetManager(serverListener) { UnsyncedEvents = true };
var clientListener = new EventBasedNetListener();
_clientManager = new NetManager(clientListener) { UnsyncedEvents = true };
var connected = new TaskCompletionSource<NetPeer>(TaskCreationOptions.RunContinuationsAsynchronously);
serverListener.ConnectionRequestEvent += request => request.AcceptIfKey("rwa-wire");
serverListener.PeerConnectedEvent += serverSide => connected.TrySetResult(serverSide);
var decoder = new LiteNetLibTransportConnection();
clientListener.NetworkReceiveEvent += (_, reader, _, _) =>
{
var frame = decoder.Receive(reader.GetRemainingBytes());
if (LiteNetLibTransportConnection.DecodeSchema(frame) is CultNetErrorMessage error) Errors.Enqueue(error);
};
_serverManager.Start(0);
_clientManager.Start(0);
_clientManager.Connect("127.0.0.1", _serverManager.LocalPort, "rwa-wire");
if (!connected.Task.Wait(TimeSpan.FromSeconds(10))) Assert.Fail("the wire pair did not connect");
var serverPeer = connected.Task.Result;
Peer = new CultNetServerPeer(serverPeer, new LiteNetLibTransportConnection(serverPeer));
}
public async Task<CultNetErrorMessage> NextErrorAsync()
{
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10);
CultNetErrorMessage? error;
while (!Errors.TryDequeue(out error))
{
if (DateTime.UtcNow > deadline) Assert.Fail("no reply arrived on the wire");
await Task.Delay(5);
}
return error;
}
public void Dispose()
{
_clientManager.Stop();
_serverManager.Stop();
}
}
[Test]
public async Task ThePeerReceivesTheTypedRefusalOnTheWire()
{
using var wire = new WirePair();
var rig = Build(authority: null, peer: wire.Peer);
await Seed(rig, KeyA, "server");
var logBefore = rig.LogLength;
await rig.Put(rig.PutOf(KeyA, "peer"));
var putReply = await wire.NextErrorAsync();
await rig.Delete(rig.DeleteOf(KeyA));
var deleteReply = await wire.NextErrorAsync();
await rig.Put(rig.PutOf(KeyB, "peer"));
var newKeyReply = await wire.NextErrorAsync();
foreach (var (reply, key) in new[] { (putReply, KeyA), (deleteReply, KeyA), (newKeyReply, KeyB) })
{
Assert.That(reply.Code, Is.EqualTo("write_not_authorized"));
Assert.That(reply.Error, Does.StartWith("write_not_authorized: "));
Assert.That(reply.Details!.Field, Is.EqualTo("recordKey"));
Assert.That(reply.Details.Value, Is.EqualTo(key.Value));
}
Assert.That(rig.TextOf(KeyA), Is.EqualTo("server"));
Assert.That(rig.TextOf(KeyB), Is.Null);
Assert.That(rig.LogLength, Is.EqualTo(logBefore));
Assert.That(wire.Errors, Is.Empty, "one reply per refused write");
}
[Test]
public async Task AnAuthoritysRefusalReasonReachesThePeerAndAGrantSendsNothingBack()
{
using var wire = new WirePair();
var rig = Build(new RecordingAuthority(request => request.RecordKey.Equals(KeyA)
? CultNetRemoteWriteDecision.Grant()
: CultNetRemoteWriteDecision.Refuse("only a")), peer: wire.Peer);
await rig.Put(rig.PutOf(KeyA, "peer"));
await rig.Put(rig.PutOf(KeyB, "peer"));
var reply = await wire.NextErrorAsync();
Assert.That(reply.Code, Is.EqualTo("write_not_authorized"));
Assert.That(reply.Error, Is.EqualTo("write_not_authorized: only a"));
Assert.That(reply.Details!.Value, Is.EqualTo(KeyB.Value));
Assert.That(rig.TextOf(KeyA), Is.EqualTo("peer"));
await Task.Delay(100);
Assert.That(wire.Errors, Is.Empty, "the granted write was not answered with an error");
}
[Test]
public async Task ADeleteSentUnderASchemaAliasAsksTheAuthorityWithTheCanonicalId()
{
var authority = new RecordingAuthority(_ => CultNetRemoteWriteDecision.Grant());
var rig = Build(authority);
await Seed(rig, KeyA, "server");
var aliased = rig.DeleteOf(KeyA);
aliased.SchemaId = "tests.networking_note.v1";
Assert.That(aliased.SchemaId, Is.Not.EqualTo(rig.SchemaId), "the fixture really sends an alias");
await rig.Delete(aliased);
var request = authority.Requests.Single();
Assert.That(request.Operation, Is.EqualTo(CultNetRemoteWriteOperation.Delete));
Assert.That(request.SchemaId, Is.EqualTo(rig.SchemaId), "the authority sees the canonical id, not the alias as sent");
Assert.That(rig.TextOf(KeyA), Is.Null, "the aliased delete was granted and applied");
}
[Test]
public async Task ANullDecisionRefusesPutAndDelete()
{
var rig = Build(new RecordingAuthority(_ => null!));
await Seed(rig, KeyA, "server");
AssertAnswered(() => rig.Put(rig.PutOf(KeyA, "peer")), "a null decision is a refusal");
AssertAnswered(() => rig.Delete(rig.DeleteOf(KeyA)), "a null decision is a refusal");
AssertAnswered(() => rig.Put(rig.PutOf(KeyB, "peer")), "a null decision is a refusal");
Assert.That(rig.TextOf(KeyA), Is.EqualTo("server"));
Assert.That(rig.TextOf(KeyB), Is.Null);
Assert.That(rig.Logger.Warnings, Has.Some.Contains("raw put refused: the write was not authorized"));
Assert.That(rig.Logger.Warnings, Has.Some.Contains("raw delete refused: the write was not authorized"));
Assert.That(rig.Logger.Errors, Is.Empty, "a null decision is a refusal, not a fault");
}
private sealed class RecordingAuthority : ICultNetRemoteWriteAuthority
{
private readonly Func<CultNetRemoteWriteRequest, CultNetRemoteWriteDecision> _decide;
public List<CultNetRemoteWriteRequest> Requests { get; } = new();
public RecordingAuthority(Func<CultNetRemoteWriteRequest, CultNetRemoteWriteDecision> decide) => _decide = decide;
public CultNetRemoteWriteDecision Decide(CultNetRemoteWriteRequest request)
{
Requests.Add(request);
return _decide(request);
}
}
private sealed class CountingForwarder : ICultNetShardWriteForwarder
{
public int Puts { get; private set; }
public int Deletes { get; private set; }
public Task ForwardPutAsync(CultNetShardDescriptor shard, CultNetDocumentPutRawMessage message) { Puts++; return Task.CompletedTask; }
public Task ForwardDeleteAsync(CultNetShardDescriptor shard, CultNetDocumentDeleteMessage message) { Deletes++; return Task.CompletedTask; }
}
private sealed class CapturingLogger : GameCult.Logging.ILogger
{
public List<string> Errors { get; } = new();
public List<string> Warnings { get; } = new();
public void LogInfo(string message) { }
public void LogWarning(string message) => Warnings.Add(message);
public void LogError(string message) => Errors.Add(message);
public void LogDebug(string message) { }
}
}
}

View file

@ -56,7 +56,7 @@ namespace GameCult.Networking.Tests
using var server = new Server(cache, ServerSecurityOptions.Development());
var logger = new Logger();
server.Logger = logger;
using var databaseServer = new CultNetDatabaseServer(server, database);
using var databaseServer = new CultNetDatabaseServer(server, database, new CultNetDatabaseServerOptions { RemoteWriteAuthority = new TestGrantsEverything() });
var handler = (Func<CultNetDocumentPutRawMessage, CultNetServerPeer, Task>)typeof(CultNetDatabaseServer)
.GetField("_putHandler", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance)!
.GetValue(databaseServer)!;