diff --git a/src/GameCult.Networking/CultNetDatabase.cs b/src/GameCult.Networking/CultNetDatabase.cs
index 668c088e..d9e7ec09 100644
--- a/src/GameCult.Networking/CultNetDatabase.cs
+++ b/src/GameCult.Networking/CultNetDatabase.cs
@@ -1015,11 +1015,16 @@ namespace GameCult.Networking
}
///
- /// 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
/// No shard owns the schema and key; nothing was written.
/// The change committed and was published, but the shard log could not record it.
- public async Task ApplyPutAsync(CultNetDocumentPutRawMessage message)
+ public Task 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 ApplyPutAsync(CultNetDocumentPutRawMessage message, Action? 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
}
///
- /// 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
/// No shard owns the schema and key; nothing was written.
/// The change committed and was published, but the shard log could not record it.
- 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? 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);
diff --git a/src/GameCult.Networking/CultNetDatabaseServer.cs b/src/GameCult.Networking/CultNetDatabaseServer.cs
index f5bbb464..149b9619 100644
--- a/src/GameCult.Networking/CultNetDatabaseServer.cs
+++ b/src/GameCult.Networking/CultNetDatabaseServer.cs
@@ -288,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 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 _database.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)
{
@@ -323,7 +364,13 @@ namespace GameCult.Networking
{
try
{
- await _database.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)
{
diff --git a/src/GameCult.Networking/CultNetRemoteWriteAuthority.cs b/src/GameCult.Networking/CultNetRemoteWriteAuthority.cs
new file mode 100644
index 00000000..b3083f19
--- /dev/null
+++ b/src/GameCult.Networking/CultNetRemoteWriteAuthority.cs
@@ -0,0 +1,113 @@
+using System;
+using GameCult.Caching;
+
+namespace GameCult.Networking
+{
+ ///
+ /// Decides whether a connected peer may put or delete a record in the database a
+ ///
+ public interface ICultNetRemoteWriteAuthority
+ {
+ ///
+ /// 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.
+ ///
+ CultNetRemoteWriteDecision Decide(CultNetRemoteWriteRequest request);
+ }
+
+ ///
+ /// The kind of write a peer asked for.
+ ///
+ public enum CultNetRemoteWriteOperation
+ {
+ /// Put a record.
+ Put,
+ /// Delete a record.
+ Delete
+ }
+
+ ///
+ /// One peer write, as the authority sees it.
+ ///
+ public sealed class CultNetRemoteWriteRequest
+ {
+ ///
+ /// Creates a request.
+ ///
+ 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;
+ }
+
+ /// Gets the peer connection the message arrived on (endpoint and transport only).
+ public CultNetServerPeer Peer { get; }
+
+ /// Gets the operation.
+ public CultNetRemoteWriteOperation Operation { get; }
+
+ /// Gets the canonical schema id the database resolved for the write, never the alias the peer sent.
+ public string SchemaId { get; }
+
+ /// Gets the record key.
+ public CultRecordKey RecordKey { get; }
+
+ /// Gets the shard id as the peer sent it.
+ public string? ShardId { get; }
+
+ /// Gets the shard epoch as the peer sent it.
+ public long? ShardEpoch { get; }
+ }
+
+ ///
+ /// An authority's answer to one
+ public sealed class CultNetRemoteWriteDecision
+ {
+ private CultNetRemoteWriteDecision(bool isGranted, string? reason)
+ {
+ IsGranted = isGranted;
+ Reason = reason;
+ }
+
+ /// Gets whether the write may proceed.
+ public bool IsGranted { get; }
+
+ /// Gets the refusal reason, when there is one.
+ public string? Reason { get; }
+
+ /// Grants the write.
+ public static CultNetRemoteWriteDecision Grant() => new(true, null);
+
+ /// Refuses the write.
+ 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; }
+ }
+}
diff --git a/src/GameCult.Networking/CultNetSchemaMessages.cs b/src/GameCult.Networking/CultNetSchemaMessages.cs
index e7f4cfd9..baf0b6d5 100644
--- a/src/GameCult.Networking/CultNetSchemaMessages.cs
+++ b/src/GameCult.Networking/CultNetSchemaMessages.cs
@@ -511,7 +511,7 @@ namespace GameCult.Networking
[Key("routingHint")] public CultNetShardRoutingHint? RoutingHint { get; set; }
///
/// Gets or sets the machine-readable refusal code (R-N), e.g. selection_invalid ,
- /// cursor_stale , cursor_invalid , reference_outside_target , unowned_schema . Additive to
+ /// cursor_stale , cursor_invalid , reference_outside_target , unowned_schema , write_not_authorized . Additive to
/// cultnet.error.v0 ; a v0 peer that does not know this field ignores it.
///
[Key("code")] public string? Code { get; set; }
@@ -543,6 +543,18 @@ namespace GameCult.Networking
Details = new CultNetErrorDetails { Field = "schemaId", Value = ex.SchemaId }
};
+ ///
+ /// Builds the wire error for a peer write its authority refused: code write_not_authorized , details
+ /// { field: "recordKey", value } .
+ ///
+ 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 }
+ };
+
///
/// Builds the wire error for a cursor refusal (R-N): code cursor_stale with details
/// { asOf, current } , or code cursor_invalid with no details.
diff --git a/src/GameCult.Networking/CultNetShardWriteForwarding.cs b/src/GameCult.Networking/CultNetShardWriteForwarding.cs
index b2b9e7a2..7038b0a7 100644
--- a/src/GameCult.Networking/CultNetShardWriteForwarding.cs
+++ b/src/GameCult.Networking/CultNetShardWriteForwarding.cs
@@ -34,6 +34,13 @@ namespace GameCult.Networking
/// Gets or sets the forwarder used when non-primary write forwarding is enabled.
///
public ICultNetShardWriteForwarder? WriteForwarder { get; set; }
+
+ ///
+ /// 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.
+ ///
+ public ICultNetRemoteWriteAuthority? RemoteWriteAuthority { get; set; }
}
///
diff --git a/tests/GameCult.Networking.Tests/CultNetJournalTests.cs b/tests/GameCult.Networking.Tests/CultNetJournalTests.cs
index 6285d674..eb68b1fc 100644
--- a/tests/GameCult.Networking.Tests/CultNetJournalTests.cs
+++ b/tests/GameCult.Networking.Tests/CultNetJournalTests.cs
@@ -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)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)typeof(CultNetDatabaseServer).GetField("_putHandler", flags)!.GetValue(databaseServer)!;
diff --git a/tests/GameCult.Networking.Tests/RemoteWriteAuthorityTests.cs b/tests/GameCult.Networking.Tests/RemoteWriteAuthorityTests.cs
new file mode 100644
index 00000000..198abb6d
--- /dev/null
+++ b/tests/GameCult.Networking.Tests/RemoteWriteAuthorityTests.cs
@@ -0,0 +1,280 @@
+#nullable enable
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Reflection;
+using System.Runtime.CompilerServices;
+using System.Threading.Tasks;
+using GameCult.Caching;
+using NUnit.Framework;
+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 Put = null!;
+ public Func Delete = null!;
+ public string SchemaId = string.Empty;
+ public List Published = new();
+ public IDisposable Subscription = null!;
+
+ public string? TextOf(CultRecordKey key) => Cache.Get(key)?.Text;
+ public int LogLength => Database.GetMutationLog(ShardId).Count;
+
+ public CultNetDocumentPutRawMessage PutOf(CultRecordKey key, string text)
+ {
+ var message = Database.Documents.CreateRawDocumentPutMessage("wire", new CultRecordHandle(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)
+ {
+ var cache = new CultCache();
+ var schemaId = cache.Registry.GetRequired().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
+ });
+ var peer = (CultNetServerPeer)RuntimeHelpers.GetUninitializedObject(typeof(CultNetServerPeer));
+ const BindingFlags flags = BindingFlags.NonPublic | BindingFlags.Instance;
+ var put = (Func)typeof(CultNetDatabaseServer).GetField("_putHandler", flags)!.GetValue(databaseServer)!;
+ var delete = (Func)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 = R3.ObservableExtensions.Subscribe(database.WatchAllChanges(), change => rig.Published.Add(change));
+ return rig;
+ }
+
+ // 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");
+
+ Assert.CatchAsync(async () => await rig.Put(rig.PutOf(KeyA, "peer")), "the peer is answered with a refusal");
+ Assert.CatchAsync(async () => await rig.Delete(rig.DeleteOf(KeyA)), "the peer is answered with a refusal");
+ Assert.CatchAsync(async () => await 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"));
+
+ Assert.CatchAsync(async () => await rig.Put(rig.PutOf(KeyB, "other-key")));
+ Assert.CatchAsync(async () => await rig.Delete(rig.DeleteOf(KeyA)));
+ 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);
+
+ Assert.CatchAsync(async () => await refusing.Put(refusing.PutOf(KeyA, "peer")));
+ Assert.CatchAsync(async () => await refusing.Delete(refusing.DeleteOf(KeyA)));
+ 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);
+ Assert.CatchAsync(async () => await unconfigured.Put(unconfigured.PutOf(KeyA, "peer")));
+ 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(Unowned), unowned);
+ put.ShardId = ShardId;
+ put.ShardEpoch = 1;
+
+ Assert.CatchAsync(async () => await rig.Put(put));
+
+ 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(Unowned), unowned);
+ grantedPut.ShardId = ShardId;
+ grantedPut.ShardEpoch = 1;
+ Assert.CatchAsync(async () => await granting.Put(grantedPut));
+ 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");
+
+ Assert.CatchAsync(async () => await rig.Put(rig.PutOf(KeyA, "peer")), "the peer is answered with a refusal");
+ Assert.CatchAsync(async () => await rig.Delete(rig.DeleteOf(KeyA)));
+
+ Assert.That(rig.TextOf(KeyA), Is.EqualTo("server"), "fail closed");
+ Assert.That(rig.Logger.Errors, Has.Some.Contains("authority exploded"));
+ }
+
+ [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"));
+ }
+
+ private sealed class RecordingAuthority : ICultNetRemoteWriteAuthority
+ {
+ private readonly Func _decide;
+ public List Requests { get; } = new();
+ public RecordingAuthority(Func 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 Errors { get; } = new();
+ public List 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) { }
+ }
+ }
+}
diff --git a/tests/GameCult.Networking.Tests/SoulP4cProbeTests.cs b/tests/GameCult.Networking.Tests/SoulP4cProbeTests.cs
index a4796e7d..726dfea4 100644
--- a/tests/GameCult.Networking.Tests/SoulP4cProbeTests.cs
+++ b/tests/GameCult.Networking.Tests/SoulP4cProbeTests.cs
@@ -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)typeof(CultNetDatabaseServer)
.GetField("_putHandler", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance)!
.GetValue(databaseServer)!;