CultNetDatabaseServer: a peer's put or delete commits only when its write authority grants it

CultNetDatabaseServerOptions.RemoteWriteAuthority is asked once per received
write, after the schema resolves and before the shard does; none configured
refuses every remote write; an authority that throws refuses and is logged.
A refusal applies, publishes, logs and forwards nothing. The host's own
CultNetDatabase.ApplyPutAsync/ApplyDeleteAsync ask no remote authority.
Existing server tests whose subject is not the grant pass a test-local
granting authority.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/GameCult.Networking/CultNetDatabase.cs

@@ -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);

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<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 _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)
             {

src/GameCult.Networking/CultNetRemoteWriteAuthority.cs

@@ -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; }
+    }
+}

src/GameCult.Networking/CultNetSchemaMessages.cs

@@ -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.

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.
         /// </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>

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<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)!;

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<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)
+        {
+            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
+            });
+            var 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 = 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<MeshQuickstartNote>(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<MeshQuickstartNote>(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<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) { }
+        }
+    }
+}

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<CultNetDocumentPutRawMessage, CultNetServerPeer, Task>)typeof(CultNetDatabaseServer)
                 .GetField("_putHandler", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance)!
                 .GetValue(databaseServer)!;