diff --git a/src/GameCult.Networking/CultNetTransport.cs b/src/GameCult.Networking/CultNetTransport.cs
index 03cfef83..31ccc428 100644
--- a/src/GameCult.Networking/CultNetTransport.cs
+++ b/src/GameCult.Networking/CultNetTransport.cs
@@ -3580,6 +3580,35 @@ namespace GameCult.Networking
}
}
+ ///
+ /// A put or delete that arrived on an unreliable channel of a receipt session (write-receipt.md, Reliable channel only):
+ /// answered message_invalid, details channel, fixed text, carrying the write's id, and not applied. True when
+ /// the frame was such a write and has been answered here.
+ ///
+ internal bool RefuseUnreliableWrite(ICultNetSchemaMessage message, DeliveryMethod method)
+ {
+ if (WriteReceipts == null) return false;
+ if (method is DeliveryMethod.ReliableOrdered or DeliveryMethod.ReliableUnordered or DeliveryMethod.ReliableSequenced) return false;
+ var messageId = message switch
+ {
+ CultNetDocumentPutRawMessage put => put.MessageId,
+ CultNetDocumentDeleteMessage delete => delete.MessageId,
+ _ => null
+ };
+ if (messageId == null && message is not (CultNetDocumentPutRawMessage or CultNetDocumentDeleteMessage)) return false;
+ try
+ {
+ SendSchema(CultNetErrorV1Message.From(
+ CultNetErrorMessage.ForMessageInvalid("channel", "receipt sessions require the reliable channel"), messageId));
+ }
+ catch (Exception)
+ {
+ End();
+ }
+
+ return true;
+ }
+
private readonly object _writeFramingGate = new();
private bool _writeFramingSettled;
diff --git a/src/GameCult.Networking/CultNetWriteReceipts.cs b/src/GameCult.Networking/CultNetWriteReceipts.cs
index 7e984287..a713c771 100644
--- a/src/GameCult.Networking/CultNetWriteReceipts.cs
+++ b/src/GameCult.Networking/CultNetWriteReceipts.cs
@@ -70,10 +70,51 @@ namespace GameCult.Networking
public static bool operator !=(CultNetDigest left, CultNetDigest right) => !left.Equals(right);
}
+ ///
+ /// What a session keeps of a refusal (write-receipt.md, What is remembered): its code, the field its details name, and its
+ /// routing hint (a shard descriptor the server owns). Never the write's strings or an authority's reason: the replay is built
+ /// from fixed text and the fields of the resend, which carries the same content.
+ ///
+ internal readonly struct CultNetRememberedRefusal
+ {
+ public string? Code { get; }
+ public bool HasDetails { get; }
+ public string? Field { get; }
+ public CultNetShardRoutingHint? Hint { get; }
+
+ public CultNetRememberedRefusal(string? code, bool hasDetails, string? field, CultNetShardRoutingHint? hint)
+ {
+ Code = code; HasDetails = hasDetails; Field = field; Hint = hint;
+ }
+
+ public static CultNetRememberedRefusal Of(CultNetErrorMessage error) =>
+ new(error.Code, error.Details != null, error.Details?.Field, error.RoutingHint);
+
+ /// The refusal, in fixed text, naming the write's own record key or schema id where its details do.
+ public CultNetErrorMessage ToError(string schemaId, string recordKey) => new()
+ {
+ Error = Code switch
+ {
+ "unowned_schema" => "unowned_schema: no shard owns the schema and key",
+ null => Hint != null ? CultNetDatabaseServer.RoutingRefusalText : CultNetWriteReceiptSession.WriteFailedText,
+ _ => Code
+ },
+ Code = Code,
+ RoutingHint = Hint,
+ Details = HasDetails
+ ? new CultNetErrorDetails
+ {
+ Field = Field,
+ Value = Field switch { "recordKey" => recordKey, "schemaId" => schemaId, _ => null }
+ }
+ : null
+ };
+ }
+
///
/// One receipt-capable peer's writes: message id to content hash and the answer given, refusals included, so a resend of the
/// same content is answered again without being decided again and the same id with other content is refused. Ids and content
- /// hashes are held as fixed-size digests; the answer is held as its kind (and, for a refusal, its code, text and details),
+ /// hashes are held as fixed-size digests; the answer is held as its kind (and, for a refusal, its code and the field its details name),
/// never the strings of the write.
///
internal sealed class CultNetWriteReceiptSession
@@ -100,7 +141,7 @@ namespace GameCult.Networking
public CultNetDigest Hash { get; set; }
public bool Abandoned { get; set; }
public bool Admitted { get; set; }
- public CultNetErrorMessage? Refusal { get; set; }
+ public CultNetRememberedRefusal? Refusal { get; set; }
}
public enum Begin { New, ReplayAdmitted, ReplayRefusal, Unanswered, DifferentContent, Full }
@@ -111,7 +152,7 @@ namespace GameCult.Networking
/// Looks up first. An id not seen is claimed for this content ();
/// otherwise the outcome says what already holds it, and is the refusal to repeat.
///
- public Begin TryBegin(string messageId, byte[] hash, out CultNetErrorMessage? refusal)
+ public Begin TryBegin(string messageId, byte[] hash, out CultNetRememberedRefusal? refusal)
{
var id = IdDigest(messageId);
var content = CultNetDigest.Of(hash);
@@ -162,7 +203,7 @@ namespace GameCult.Networking
}
/// The write was refused: remember the refusal so a resend repeats it.
- public void Refused(string messageId, CultNetErrorMessage refusal)
+ public void Refused(string messageId, CultNetRememberedRefusal refusal)
{
lock (_gate)
{
@@ -270,7 +311,7 @@ namespace GameCult.Networking
return false;
case CultNetWriteReceiptSession.Begin.ReplayRefusal:
_answered = true;
- _send(CultNetErrorV1Message.From(refusal!, _messageId));
+ _send(CultNetErrorV1Message.From(refusal!.Value.ToError(_schemaId, _recordKey), _messageId));
return false;
case CultNetWriteReceiptSession.Begin.Unanswered:
// The first attempt is still being applied and will answer this id, or its primary left it unanswered:
@@ -309,9 +350,15 @@ namespace GameCult.Networking
{
if (_answered) return;
_answered = true;
- var refusal = CultNetErrorV1Message.From(error, _messageId);
- if (_claimed) _session.Refused(_messageId, error);
- _send(refusal);
+ if (_claimed)
+ {
+ // A remembered refusal is answered from what is remembered, so the replay is the refusal again.
+ var kept = CultNetRememberedRefusal.Of(error);
+ _session.Refused(_messageId, kept);
+ error = kept.ToError(_schemaId, _recordKey);
+ }
+
+ _send(CultNetErrorV1Message.From(error, _messageId));
}
///
diff --git a/src/GameCult.Networking/Server.cs b/src/GameCult.Networking/Server.cs
index d9b763b9..08af0cc4 100644
--- a/src/GameCult.Networking/Server.cs
+++ b/src/GameCult.Networking/Server.cs
@@ -415,6 +415,7 @@ namespace GameCult.Networking
{
var cultNetMessage = LiteNetLibTransportConnection.DecodeSchema(frame);
Logger.LogDebug($"Received CultNet schema message {cultNetMessage.SchemaVersion}");
+ if (transport.RefuseUnreliableWrite(cultNetMessage, method)) return;
await HandleCultNetSchemaMessageAsync(peer, user, cultNetMessage).ConfigureAwait(false);
return;
}
diff --git a/tests/GameCult.Networking.Tests/CultNetWriteReceiptLoopbackTests.cs b/tests/GameCult.Networking.Tests/CultNetWriteReceiptLoopbackTests.cs
index 8a1f0b46..19180459 100644
--- a/tests/GameCult.Networking.Tests/CultNetWriteReceiptLoopbackTests.cs
+++ b/tests/GameCult.Networking.Tests/CultNetWriteReceiptLoopbackTests.cs
@@ -47,6 +47,9 @@ namespace GameCult.Networking.Tests
public void Send(T message) where T : ICultNetSchemaMessage =>
_peer.Send(CultNetSchemaMessageSerialization.Serialize(message), DeliveryMethod.ReliableOrdered);
+ public void SendUnreliable(T message) where T : ICultNetSchemaMessage =>
+ _peer.Send(CultNetSchemaMessageSerialization.Serialize(message), DeliveryMethod.Unreliable);
+
public void SendRaw(byte[] bytes) => _peer.Send(bytes, DeliveryMethod.ReliableOrdered);
public T Next() where T : ICultNetSchemaMessage
@@ -175,6 +178,69 @@ namespace GameCult.Networking.Tests
Assert.That(running.Cache.Get(KeyB), Is.Null);
}
+ // write-receipt.md, Reliable channel only: a write on an unreliable channel of a receipt session is refused
+ // message_invalid, details.field "channel", fixed text, carrying the write's id; it is not applied.
+ [Test]
+ public async Task AWriteOnAnUnreliableChannelOfAReceiptPeerIsRefusedAndNotApplied()
+ {
+ using var running = Running.Start();
+ using var client = running.Connect();
+ client.Send(HelloOf(BothFrames));
+ client.Next();
+
+ client.SendUnreliable(running.PutOf("u1", KeyA, "lost"));
+
+ var error = client.Next();
+ Assert.That(error.Code, Is.EqualTo("message_invalid"));
+ Assert.That(error.Details!.Field, Is.EqualTo("channel"));
+ Assert.That(error.Error, Is.EqualTo("message_invalid: receipt sessions require the reliable channel"));
+ Assert.That(error.MessageId, Is.EqualTo("u1"));
+
+ // The session goes on, and the same id on the reliable channel is a new write, not a replayed refusal.
+ client.Send(running.PutOf("u1", KeyA, "kept"));
+ Assert.That(client.Next().MessageId, Is.EqualTo("u1"));
+ await WaitFor(() => running.Cache.Get(KeyA)?.Text == "kept");
+ }
+
+ [Test]
+ public void ADeleteOnAnUnreliableChannelOfAReceiptPeerIsRefusedWithItsId()
+ {
+ using var running = Running.Start();
+ using var client = running.Connect();
+ client.Send(HelloOf(BothFrames));
+ client.Next();
+
+ client.SendUnreliable(new CultNetDocumentDeleteMessage
+ {
+ MessageId = "ud1", SchemaId = running.SchemaId, RecordKey = KeyA.Value, ShardId = ShardId, ShardEpoch = 1
+ });
+
+ var error = client.Next();
+ Assert.That(error.Code, Is.EqualTo("message_invalid"));
+ Assert.That(error.Details!.Field, Is.EqualTo("channel"));
+ Assert.That(error.MessageId, Is.EqualTo("ud1"));
+ }
+
+ [Test]
+ public async Task OnlyWritesAreRefusedOnTheUnreliableChannelAndALegacyPeerIsLeftAsBefore()
+ {
+ using var running = Running.Start();
+ using var receipt = running.Connect();
+ receipt.Send(HelloOf(BothFrames));
+ receipt.Next();
+ receipt.SendUnreliable(HelloOf(BothFrames)); // not a write: answered with nothing on the unreliable channel
+ receipt.Send(running.PutOf("r1", KeyA, "one"));
+ Assert.That(receipt.Next().MessageId, Is.EqualTo("r1"), "nothing else was sent ahead of the receipt");
+
+ using var legacy = running.Connect();
+ legacy.Send(HelloOf());
+ legacy.SendUnreliable(running.PutOf("l1", KeyB, "two")); // refused by the authority, answered as v0 as before
+ Assert.That(legacy.Next().Code, Is.EqualTo("write_not_authorized"));
+ legacy.SendUnreliable(running.PutOf("l2", KeyA, "three"));
+ await WaitFor(() => running.Cache.Get(KeyA)?.Text == "three");
+ legacy.ExpectNothingMore("a legacy peer is never sent a receipt, error.v1 or a channel refusal");
+ }
+
private static async Task WaitFor(Func condition)
{
var deadline = DateTime.UtcNow + Cap;
diff --git a/tests/GameCult.Networking.Tests/CultNetWriteReceiptServerTests.cs b/tests/GameCult.Networking.Tests/CultNetWriteReceiptServerTests.cs
index 09188c5f..7c6537a3 100644
--- a/tests/GameCult.Networking.Tests/CultNetWriteReceiptServerTests.cs
+++ b/tests/GameCult.Networking.Tests/CultNetWriteReceiptServerTests.cs
@@ -591,7 +591,7 @@ namespace GameCult.Networking.Tests
await rig.Delete(wire, "fwd-2", KeyA);
AssertRefused(wire.Sent[0], "fwd-1", "write_not_authorized");
- Assert.That(((CultNetErrorV1Message)wire.Sent[0]).Error, Is.EqualTo("primary says no"));
+ Assert.That(((CultNetErrorV1Message)wire.Sent[0]).Error, Is.EqualTo("write_not_authorized"), "a relayed refusal is in fixed text, the same one a resend replays");
AssertRefused(wire.Sent[1], "fwd-2", "message_invalid");
}
@@ -631,7 +631,7 @@ namespace GameCult.Networking.Tests
await rig.Put(wire, "fwd-1", KeyA, "one");
Assert.That(wire.Sent, Has.Count.EqualTo(2));
- Assert.That(((CultNetErrorV1Message)wire.Sent[1]).Error, Is.EqualTo("primary says no"));
+ Assert.That(((CultNetErrorV1Message)wire.Sent[1]).Error, Is.EqualTo("write_not_authorized"), "a remembered refusal is replayed in fixed text, not the primary's own");
Assert.That(forwarder.Forwarded, Is.EqualTo(1));
}
@@ -1119,5 +1119,57 @@ namespace GameCult.Networking.Tests
var error = (CultNetErrorMessage)wire.Sent.Single();
Assert.That(error.Error, Is.EqualTo("the shard log request could not be answered"));
}
+
+ private static CultNetErrorMessage DeniedWith(string recordKey, string reason) => new()
+ {
+ Error = $"write_not_authorized: {reason}",
+ Code = "write_not_authorized",
+ Details = new CultNetErrorDetails { Field = "recordKey", Value = recordKey }
+ };
+
+ // write-receipt.md, What is remembered: a refusal is remembered as its code (and the field it names), never the strings of
+ // the write or of the authority's reason; the replay is built from fixed text and the resend's own fields.
+ [Test]
+ public void ARefusalIsReplayedFromFixedTextAndTheResendsOwnFieldsNotFromRememberedStrings()
+ {
+ var session = new CultNetWriteReceiptSession();
+ var sent = new List();
+ var longKey = new string('k', 50_000);
+ CultNetReceiptedWrite Write() => new(session, sent.Add, "put", "m1", "schema.v1", longKey, [1]);
+
+ var first = Write();
+ Assert.That(first.Begin(), Is.True);
+ first.Refused(DeniedWith(longKey, new string('r', 50_000)));
+ Assert.That(Write().Begin(), Is.False);
+
+ var again = (CultNetErrorV1Message)sent[1];
+ Assert.That(again.MessageId, Is.EqualTo("m1"));
+ Assert.That(again.Code, Is.EqualTo("write_not_authorized"));
+ Assert.That(again.Error, Is.EqualTo("write_not_authorized"));
+ Assert.That(again.Details!.Field, Is.EqualTo("recordKey"));
+ Assert.That(again.Details.Value, Is.EqualTo(longKey));
+ var firstAnswer = (CultNetErrorV1Message)sent[0];
+ Assert.That(firstAnswer.Error, Is.EqualTo(again.Error), "the replay is the refusal again");
+ Assert.That(firstAnswer.Details!.Value, Is.EqualTo(again.Details.Value));
+ }
+
+ [Test]
+ public void ThousandsOfRefusalsOf50kCharacterKeysAreRememberedInBoundedMemory()
+ {
+ var session = new CultNetWriteReceiptSession();
+ var before = GC.GetTotalMemory(true);
+ for (var i = 0; i < CultNetWriteReceiptSession.MaxRememberedWrites; i++)
+ {
+ var key = new string('k', 50_000) + i;
+ var write = new CultNetReceiptedWrite(session, _ => { }, "put", $"m{i}", "schema.v1", key, [1]);
+ Assert.That(write.Begin(), Is.True);
+ write.Refused(DeniedWith(key, key));
+ }
+
+ // Held as strings, the 1024 keys and reasons alone would be over 200 MB.
+ var growth = GC.GetTotalMemory(true) - before;
+ GC.KeepAlive(session);
+ Assert.That(growth, Is.LessThan(8L * 1024 * 1024));
+ }
}
}