diff --git a/src/GameCult.Networking/CultNetDatabase.cs b/src/GameCult.Networking/CultNetDatabase.cs
index d9e7ec09..af561ea5 100644
--- a/src/GameCult.Networking/CultNetDatabase.cs
+++ b/src/GameCult.Networking/CultNetDatabase.cs
@@ -1017,6 +1017,8 @@ namespace GameCult.Networking
///
/// 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 , which does.
+ /// Returning normally means the put is committed and visible to readers: a receipt-capable peer is told
+ /// write_admitted on the strength of it, so a write that has only been queued must not return.
///
/// No shard owns the schema and key; nothing was written.
/// The change committed and was published, but the shard log could not record it.
@@ -1047,6 +1049,7 @@ namespace GameCult.Networking
///
/// 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 , which does.
+ /// Returning normally means the delete is committed and visible to readers (see ).
///
/// No shard owns the schema and key; nothing was written.
/// The change committed and was published, but the shard log could not record it.
diff --git a/src/GameCult.Networking/CultNetDatabaseServer.cs b/src/GameCult.Networking/CultNetDatabaseServer.cs
index f515b09d..a6e55c3d 100644
--- a/src/GameCult.Networking/CultNetDatabaseServer.cs
+++ b/src/GameCult.Networking/CultNetDatabaseServer.cs
@@ -345,7 +345,7 @@ namespace GameCult.Networking
}
// The write's receipt state, or null when its connection did not opt in (it is then answered exactly as before).
- private static CultNetReceiptedWrite? BeginReceipted(
+ private CultNetReceiptedWrite? BeginReceipted(
CultNetServerPeer peer,
string operation,
string? messageId,
@@ -362,8 +362,22 @@ namespace GameCult.Networking
: new CultNetReceiptedWrite(session, message => Answer(peer, message), operation, messageId, schemaId, recordKey, payload);
}
- // Every frame a write can be answered with. Mock point for tests: the answer leaves through the peer's transport.
- internal static void Answer(CultNetServerPeer peer, ICultNetSchemaMessage message)
+ // Every frame a write can be answered with. Mock point for tests: the answer leaves through the peer's transport. An
+ // answer that cannot be queued ends the connection: the peer settles its write as session_ended, never as silence.
+ private void Answer(CultNetServerPeer peer, ICultNetSchemaMessage message)
+ {
+ try
+ {
+ Send(peer, message);
+ }
+ catch (Exception ex)
+ {
+ _server.Logger.LogWarning($"CultNet write answer could not be sent; ending the connection: {ex.Message}");
+ peer.Transport?.End();
+ }
+ }
+
+ private static void Send(CultNetServerPeer peer, ICultNetSchemaMessage message)
{
switch (message)
{
@@ -393,7 +407,10 @@ namespace GameCult.Networking
}
_server.Logger.LogWarning($"CultNet forwarded write was not answered by its primary: {failure.Message}");
- write.Unanswered();
+ // Silence past the window means the primary may have applied it: never forward the id again on this session.
+ // Any other failure never reached a primary that could have.
+ if (failure is CultNetWriteUnconfirmedException) write.LeftSilent();
+ else write.Unanswered();
}
private async Task HandlePutAsync(CultNetDocumentPutRawMessage message, CultNetServerPeer peer)
@@ -405,16 +422,14 @@ namespace GameCult.Networking
return;
}
+ if (write != null && !write.Begin()) return;
+
var authorize = Authorize(peer, CultNetRemoteWriteOperation.Put, message.ShardId, message.ShardEpoch);
try
{
- await _database.ApplyPutAsync(message, write == null ? authorize : write.Guard(authorize)).ConfigureAwait(false);
+ await _database.ApplyPutAsync(message, authorize).ConfigureAwait(false);
write?.Admitted();
}
- catch (CultNetWriteResendException resend) when (write != null)
- {
- write.Resent(resend);
- }
catch (CultNetRemoteWriteRefusedException ex)
{
_server.Logger.LogWarning($"CultNet raw put refused: {ex.Message}");
@@ -462,7 +477,7 @@ namespace GameCult.Networking
catch (Exception ex)
{
_server.Logger.LogError($"CultNet raw put failed: {ex.Message}");
- Refuse(peer, write, new CultNetErrorMessage { Error = ex.Message });
+ Refuse(peer, write, new CultNetErrorMessage { Error = CultNetWriteReceiptSession.WriteFailedText });
}
}
@@ -475,16 +490,14 @@ namespace GameCult.Networking
return;
}
+ if (write != null && !write.Begin()) return;
+
var authorize = Authorize(peer, CultNetRemoteWriteOperation.Delete, message.ShardId, message.ShardEpoch);
try
{
- await _database.ApplyDeleteAsync(message, write == null ? authorize : write.Guard(authorize)).ConfigureAwait(false);
+ await _database.ApplyDeleteAsync(message, authorize).ConfigureAwait(false);
write?.Admitted();
}
- catch (CultNetWriteResendException resend) when (write != null)
- {
- write.Resent(resend);
- }
catch (CultNetRemoteWriteRefusedException ex)
{
_server.Logger.LogWarning($"CultNet raw delete refused: {ex.Message}");
@@ -528,7 +541,7 @@ namespace GameCult.Networking
}
catch (Exception ex)
{
- Refuse(peer, write, new CultNetErrorMessage { Error = ex.Message });
+ Refuse(peer, write, new CultNetErrorMessage { Error = CultNetWriteReceiptSession.WriteFailedText });
}
}
diff --git a/src/GameCult.Networking/CultNetSchemaMessages.cs b/src/GameCult.Networking/CultNetSchemaMessages.cs
index 0a3f9adb..350ef43b 100644
--- a/src/GameCult.Networking/CultNetSchemaMessages.cs
+++ b/src/GameCult.Networking/CultNetSchemaMessages.cs
@@ -663,6 +663,18 @@ namespace GameCult.Networking
};
}
+ ///
+ /// The refusal of a write a session has no room to remember (write-receipt.md, rule 5): code limit_reached,
+ /// details present and all nil. Exists only as error.v1; v0 has no such code.
+ ///
+ public static CultNetErrorV1Message LimitReached(string? messageId) => new CultNetErrorV1Message
+ {
+ Error = "limit_reached: the session's remembered-write limit is full",
+ Code = "limit_reached",
+ Details = new CultNetErrorDetails(),
+ MessageId = string.IsNullOrEmpty(messageId) ? null : messageId
+ };
+
/// Gets the v0 form, dropping the id.
public CultNetErrorMessage ToV0() => new CultNetErrorMessage
{
diff --git a/src/GameCult.Networking/CultNetTransport.cs b/src/GameCult.Networking/CultNetTransport.cs
index 06e2d93b..03cfef83 100644
--- a/src/GameCult.Networking/CultNetTransport.cs
+++ b/src/GameCult.Networking/CultNetTransport.cs
@@ -3534,6 +3534,7 @@ namespace GameCult.Networking
public sealed class LiteNetLibTransportConnection
{
private readonly Action? _deliver;
+ private readonly Action? _end;
private readonly CultNetTransportStats _stats = new CultNetTransportStats();
///
@@ -3543,21 +3544,41 @@ namespace GameCult.Networking
{
if (peer == null) throw new ArgumentNullException(nameof(peer));
_deliver = payload => peer.Send(payload, DeliveryMethod.ReliableOrdered);
+ _end = peer.Disconnect;
Profile = profile ?? CultNetTransportProfiles.CreateLiteNetLib("csharp-litenetlib-peer");
}
// The mock point for what leaves the connection: the peer's reliable ordered send.
- internal LiteNetLibTransportConnection(Action deliver)
+ internal LiteNetLibTransportConnection(Action deliver, Action? end = null)
{
_deliver = deliver ?? throw new ArgumentNullException(nameof(deliver));
+ _end = end;
Profile = CultNetTransportProfiles.CreateLiteNetLib("csharp-litenetlib-peer");
}
- /// Answers a schema frame that did not decode with a message_invalid error.v1 naming its id when it holds one.
- internal void AnswerUnreadableSchemaFrame(byte[] payload) =>
- SendSchema(CultNetErrorV1Message.From(
- CultNetErrorMessage.ForMessageInvalid("message", "the message could not be read"),
- CultNetReadableMessageId.From(payload)));
+ ///
+ /// Ends the connection. A write that can be neither answered nor refused is answered by this goodbye: the peer sees
+ /// its session end and settles the write as unconfirmed (session_ended), never as silence.
+ ///
+ internal void End() => _end?.Invoke();
+
+ ///
+ /// Answers a schema frame that did not decode with a message_invalid error.v1 naming its id when it holds one;
+ /// when that answer cannot be sent the connection ends instead.
+ ///
+ internal void AnswerUnreadableSchemaFrame(byte[] payload)
+ {
+ try
+ {
+ SendSchema(CultNetErrorV1Message.From(
+ CultNetErrorMessage.ForMessageInvalid("message", "the message could not be read"),
+ CultNetReadableMessageId.From(payload)));
+ }
+ catch (Exception)
+ {
+ End();
+ }
+ }
private readonly object _writeFramingGate = new();
private bool _writeFramingSettled;
diff --git a/src/GameCult.Networking/CultNetWriteReceipts.cs b/src/GameCult.Networking/CultNetWriteReceipts.cs
index 0d09a92f..87f2b286 100644
--- a/src/GameCult.Networking/CultNetWriteReceipts.cs
+++ b/src/GameCult.Networking/CultNetWriteReceipts.cs
@@ -48,14 +48,17 @@ namespace GameCult.Networking
}
///
- /// One receipt-capable peer's writes: message id to content hash, so a resend of the same content is re-receipted
- /// without being committed again and the same id with other content is refused.
+ /// 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.
///
internal sealed class CultNetWriteReceiptSession
{
- /// The most writes a peer is remembered for; a peer at the limit has new writes refused.
+ /// The most writes a peer is remembered for, refused ones included; a peer at the limit has new writes refused.
internal const int MaxRememberedWrites = 1024;
+ /// What a write is refused with when it failed for a reason no typed refusal names. Never an exception's own text.
+ internal const string WriteFailedText = "the write could not be applied";
+
private readonly object _gate = new();
private readonly Dictionary _writes = new(StringComparer.Ordinal);
@@ -63,20 +66,31 @@ namespace GameCult.Networking
{
public Entry(byte[] hash) { Hash = hash; }
public byte[] Hash { get; }
- public bool Admitted { get; set; }
+ public bool Silent { get; set; }
+ public ICultNetSchemaMessage? Answer { get; set; }
}
- public enum Begin { New, AdmittedBefore, InFlight, DifferentContent, Full }
+ public enum Begin { New, Replay, InFlight, Silent, DifferentContent, Full }
- /// Claims for a write with this content, or says what already holds it.
- public Begin TryBegin(string messageId, byte[] hash)
+ ///
+ /// Looks up first. An id not seen is claimed for this content ();
+ /// otherwise the outcome says what already holds it, and is the answer to replay.
+ ///
+ public Begin TryBegin(string messageId, byte[] hash, out ICultNetSchemaMessage? answer)
{
lock (_gate)
{
+ answer = null;
if (_writes.TryGetValue(messageId, out var entry))
{
if (!CryptographicOperations.FixedTimeEquals(entry.Hash, hash)) return Begin.DifferentContent;
- return entry.Admitted ? Begin.AdmittedBefore : Begin.InFlight;
+ if (entry.Answer != null)
+ {
+ answer = entry.Answer;
+ return Begin.Replay;
+ }
+
+ return entry.Silent ? Begin.Silent : Begin.InFlight;
}
if (_writes.Count >= MaxRememberedWrites) return Begin.Full;
@@ -85,16 +99,28 @@ namespace GameCult.Networking
}
}
- /// The write committed: remember it so a resend is re-receipted.
- public void Admitted(string messageId)
+ /// The write was answered, admitted or refused: remember the answer so a resend repeats it.
+ public void Answered(string messageId, ICultNetSchemaMessage answer)
{
lock (_gate)
{
- if (_writes.TryGetValue(messageId, out var entry)) entry.Admitted = true;
+ if (_writes.TryGetValue(messageId, out var entry)) entry.Answer = answer;
}
}
- /// The write was refused or unanswered: forget it so a corrected or repeated attempt is judged afresh.
+ ///
+ /// The primary left a forward unanswered: it may have applied it, so the id is never forwarded again on this session and
+ /// a resend is answered with nothing (the client settles it unconfirmed).
+ ///
+ public void LeftSilent(string messageId)
+ {
+ lock (_gate)
+ {
+ if (_writes.TryGetValue(messageId, out var entry)) entry.Silent = true;
+ }
+ }
+
+ /// The write never reached anyone who could have applied it: forget it so a repeat is judged afresh.
public void Abandoned(string messageId)
{
lock (_gate)
@@ -139,18 +165,6 @@ namespace GameCult.Networking
}
}
- /// A resend the receipt session answered without applying it again, or refused as a reused id.
- internal sealed class CultNetWriteResendException : Exception
- {
- public CultNetWriteResendException(CultNetWriteReceiptSession.Begin outcome)
- : base("write answered from the connection's remembered writes")
- {
- Outcome = outcome;
- }
-
- public CultNetWriteReceiptSession.Begin Outcome { get; }
- }
-
///
/// One write on a receipt-capable connection: its dedupe claim and its single answer (write-receipt.md, rules 1 to 5).
///
@@ -185,94 +199,87 @@ namespace GameCult.Networking
}
///
- /// Wraps the server's authorization so the connection's remembered writes are consulted only after the actor is
- /// authorized: authorize first, then claim the id. A resend of an admitted write throws
- /// instead of applying it again.
+ /// Looks the write's id up before anything else is decided (rule 5). True when the id is new and the write should go on
+ /// to authorization and the commit; false when it was answered here (a replay, a reused id, a full table) or is owed no
+ /// further answer (still in flight, or left silent by its primary).
///
- public Action Guard(Action authorize) => (schemaId, key) =>
+ public bool Begin()
{
- authorize(schemaId, key);
- if (_messageId.Length == 0) return;
- var outcome = _session.TryBegin(_messageId, CultNetWriteContentHash.Compute(_operation, _schemaId, _recordKey, _payload));
- if (outcome == CultNetWriteReceiptSession.Begin.New)
+ var outcome = _session.TryBegin(_messageId, CultNetWriteContentHash.Compute(_operation, _schemaId, _recordKey, _payload), out var answer);
+ switch (outcome)
{
- _claimed = true;
- return;
+ case CultNetWriteReceiptSession.Begin.New:
+ _claimed = true;
+ return true;
+ case CultNetWriteReceiptSession.Begin.Replay:
+ _answered = true;
+ _send(answer!);
+ return false;
+ case CultNetWriteReceiptSession.Begin.InFlight:
+ // The first attempt is still being applied and will answer this id.
+ _answered = true;
+ return false;
+ case CultNetWriteReceiptSession.Begin.Silent:
+ // The first attempt went to a primary that never answered; this id is never sent again, and stays unanswered.
+ _answered = true;
+ return false;
+ case CultNetWriteReceiptSession.Begin.DifferentContent:
+ Refused(CultNetErrorMessage.ForMessageInvalid("messageId", "the message id was already used for different content"));
+ return false;
+ default:
+ _answered = true;
+ _send(CultNetErrorV1Message.LimitReached(_messageId));
+ return false;
}
+ }
- throw new CultNetWriteResendException(outcome);
- };
-
- /// The write committed: remember it and send its one receipt.
+ /// The write committed: remember the receipt and send it.
public void Admitted()
{
if (_answered) return;
_answered = true;
- if (_messageId.Length > 0) _session.Admitted(_messageId);
- _send(new CultNetWriteAdmittedMessage
+ var receipt = new CultNetWriteAdmittedMessage
{
MessageId = _messageId,
Operation = _operation,
SchemaId = _schemaId,
RecordKey = _recordKey
- });
+ };
+ if (_claimed) _session.Answered(_messageId, receipt);
+ _send(receipt);
}
- /// The write was refused: forget its claim so a repeat is judged afresh, and send its one error.v1.
+ /// The write was refused: remember the refusal (a resend repeats it) and send its one error.v1.
public void Refused(CultNetErrorMessage error)
{
if (_answered) return;
_answered = true;
- Release();
- _send(CultNetErrorV1Message.From(error, _messageId));
+ var refusal = CultNetErrorV1Message.From(error, _messageId);
+ if (_claimed) _session.Answered(_messageId, refusal);
+ _send(refusal);
}
- /// No answer is owed or possible (a forward the primary never answered): forget the claim, send nothing.
+ ///
+ /// A forward its primary left unanswered: no answer is owed or possible, and the id is never forwarded again on this
+ /// session. Sends nothing.
+ ///
+ public void LeftSilent()
+ {
+ if (_answered) return;
+ _answered = true;
+ if (_claimed) _session.LeftSilent(_messageId);
+ }
+
+ /// A forward that never reached a primary that could have applied it: forget the claim, send nothing.
public void Unanswered()
{
if (_answered) return;
_answered = true;
- Release();
- }
-
- /// Answers a resend per rule 5.
- public void Resent(CultNetWriteResendException resend)
- {
- switch (resend.Outcome)
- {
- case CultNetWriteReceiptSession.Begin.AdmittedBefore:
- _answered = true;
- _send(new CultNetWriteAdmittedMessage
- {
- MessageId = _messageId,
- Operation = _operation,
- SchemaId = _schemaId,
- RecordKey = _recordKey
- });
- break;
- case CultNetWriteReceiptSession.Begin.InFlight:
- // The first attempt is still being applied and will answer this id.
- _answered = true;
- break;
- case CultNetWriteReceiptSession.Begin.DifferentContent:
- Refused(CultNetErrorMessage.ForMessageInvalid("messageId", "the message id was already used for different content"));
- break;
- default:
- Refused(new CultNetErrorMessage { Error = "too many writes are remembered for this connection" });
- break;
- }
- }
-
- /// Whether the write has had its answer (or has none owed).
- public bool Settled => _answered;
-
- /// A missing id cannot be answered under rule 1; refuse it with no id.
- public bool HasId => _messageId.Length > 0;
-
- private void Release()
- {
if (_claimed) _session.Abandoned(_messageId);
_claimed = false;
}
+
+ /// A missing id cannot be answered under rule 1; refuse it with no id.
+ public bool HasId => _messageId.Length > 0;
}
}
diff --git a/tests/GameCult.Networking.Tests/CultNetWriteReceiptServerTests.cs b/tests/GameCult.Networking.Tests/CultNetWriteReceiptServerTests.cs
index 9625c364..b93b43b3 100644
--- a/tests/GameCult.Networking.Tests/CultNetWriteReceiptServerTests.cs
+++ b/tests/GameCult.Networking.Tests/CultNetWriteReceiptServerTests.cs
@@ -23,20 +23,32 @@ namespace GameCult.Networking.Tests
private sealed class Authority : ICultNetRemoteWriteAuthority
{
public HashSet Denied { get; } = new();
- public CultNetRemoteWriteDecision Decide(CultNetRemoteWriteRequest request) =>
- Denied.Contains(request.RecordKey.Value) ? CultNetRemoteWriteDecision.Refuse("denied") : CultNetRemoteWriteDecision.Grant();
+ public int Asked { get; private set; }
+ public CultNetRemoteWriteDecision Decide(CultNetRemoteWriteRequest request)
+ {
+ Asked++;
+ return Denied.Contains(request.RecordKey.Value) ? CultNetRemoteWriteDecision.Refuse("denied") : CultNetRemoteWriteDecision.Grant();
+ }
}
private sealed class Wire
{
public List Sent { get; } = new();
+ /// When set, the connection cannot queue what it is sent.
+ public bool Unsendable { get; set; }
+ public int Ended { get; private set; }
public LiteNetLibTransportConnection Transport { get; }
public CultNetServerPeer Peer { get; }
public Wire()
{
- Transport = new LiteNetLibTransportConnection(bytes =>
- Sent.Add(LiteNetLibTransportConnection.DecodeSchema(LiteNetLibTransportConnection.Decode(bytes))));
+ Transport = new LiteNetLibTransportConnection(
+ bytes =>
+ {
+ if (Unsendable) throw new InvalidOperationException("the send queue is full");
+ Sent.Add(LiteNetLibTransportConnection.DecodeSchema(LiteNetLibTransportConnection.Decode(bytes)));
+ },
+ () => Ended++);
Peer = (CultNetServerPeer)RuntimeHelpers.GetUninitializedObject(typeof(CultNetServerPeer));
typeof(CultNetServerPeer).GetField("k__BackingField", BindingFlags.NonPublic | BindingFlags.Instance)!
.SetValue(Peer, Transport);
@@ -273,7 +285,7 @@ namespace GameCult.Networking.Tests
}
[Test]
- public async Task AuthorizationIsDecidedBeforeAReusedIdIsLookedAt()
+ public async Task AReusedIdIsJudgedBeforeAuthorityIsAskedAgain()
{
var (rig, wire) = await Capable();
await rig.Put(wire, "m1", KeyA, "one");
@@ -281,11 +293,25 @@ namespace GameCult.Networking.Tests
await rig.Put(wire, "m1", KeyB, "other");
- AssertRefused(wire.Sent[1], "m1", "write_not_authorized");
+ AssertRefused(wire.Sent[1], "m1", "message_invalid");
+ Assert.That(rig.Authority.Asked, Is.EqualTo(1), "authority is asked on an id's first sighting only");
}
[Test]
- public async Task ARefusedWriteIsForgottenSoARepeatIsJudgedAfresh()
+ public async Task AResendIsAnsweredFromMemoryWithoutAskingAuthorityAgain()
+ {
+ var (rig, wire) = await Capable();
+ await rig.Put(wire, "m1", KeyA, "one");
+ rig.Authority.Denied.Add(KeyA.Value);
+
+ await rig.Put(wire, "m1", KeyA, "one");
+
+ AssertAdmitted(wire.Sent[1], "m1", "put", KeyA.Value);
+ Assert.That(rig.Authority.Asked, Is.EqualTo(1));
+ }
+
+ [Test]
+ public async Task ARefusedWriteIsRememberedSoItsResendReplaysTheRefusal()
{
var (rig, wire) = await Capable();
rig.Authority.Denied.Add(KeyA.Value);
@@ -294,9 +320,131 @@ namespace GameCult.Networking.Tests
await rig.Put(wire, "m1", KeyA, "one");
- AssertRefused(wire.Sent[0], "m1", "write_not_authorized");
- AssertAdmitted(wire.Sent[1], "m1", "put", KeyA.Value);
- Assert.That(rig.TextOf(KeyA), Is.EqualTo("one"));
+ var first = AssertRefused(wire.Sent[0], "m1", "write_not_authorized");
+ var again = AssertRefused(wire.Sent[1], "m1", "write_not_authorized");
+ Assert.That(again.Error, Is.EqualTo(first.Error));
+ Assert.That(rig.TextOf(KeyA), Is.Null, "the resend was not decided afresh");
+ Assert.That(rig.Authority.Asked, Is.EqualTo(1));
+ }
+
+ [Test]
+ public async Task ARefusedIdReusedForOtherContentIsMessageInvalid()
+ {
+ var (rig, wire) = await Capable();
+ rig.Authority.Denied.Add(KeyA.Value);
+ await rig.Put(wire, "m1", KeyA, "one");
+
+ await rig.Put(wire, "m1", KeyB, "other");
+
+ AssertRefused(wire.Sent[1], "m1", "message_invalid");
+ Assert.That(rig.TextOf(KeyB), Is.Null);
+ }
+
+ [Test]
+ public async Task RefusedWritesCountAgainstTheRememberedWriteLimit()
+ {
+ var (rig, wire) = await Capable();
+ rig.Authority.Denied.Add(KeyA.Value);
+ for (var index = 0; index < CultNetWriteReceiptSession.MaxRememberedWrites; index++)
+ {
+ await rig.Put(wire, $"r{index}", KeyA, $"v{index}");
+ }
+
+ wire.Sent.Clear();
+ await rig.Put(wire, "over", KeyB, "x");
+ await rig.Put(wire, "r0", KeyA, "v0");
+
+ AssertRefused(wire.Sent[0], "over", "limit_reached");
+ AssertRefused(wire.Sent[1], "r0", "write_not_authorized");
+ Assert.That(rig.TextOf(KeyB), Is.Null);
+ }
+
+ [Test]
+ public async Task APutAndADeleteNeverShareAnId()
+ {
+ var (rig, wire) = await Capable();
+ await rig.Put(wire, "p1", KeyA, "one");
+ await rig.Delete(wire, "d1", KeyA);
+ wire.Sent.Clear();
+
+ await rig.Delete(wire, "p1", KeyA);
+ await rig.Put(wire, "d1", KeyA, "one");
+
+ AssertRefused(wire.Sent[0], "p1", "message_invalid");
+ AssertRefused(wire.Sent[1], "d1", "message_invalid");
+ Assert.That(rig.TextOf(KeyA), Is.Null, "neither reuse was applied");
+ }
+
+ [Test]
+ public async Task ADeleteAsTheFirstFrameSettlesTheConnectionAsLegacy()
+ {
+ var rig = Build();
+ var wire = new Wire();
+ await rig.Delete(wire, "d1", KeyA);
+ await rig.Hello(wire, BothFrames);
+ await rig.Put(wire, "m1", KeyA, "one");
+
+ Assert.That(wire.Sent, Is.Empty, "a hello after the first delete does not opt the connection in");
+ }
+
+ [Test]
+ public async Task AnAnswerThatCannotBeSentEndsTheConnectionInsteadOfLeavingTheWriteSilent()
+ {
+ var (rig, wire) = await Capable();
+ wire.Unsendable = true;
+ await rig.Put(wire, "m1", KeyA, "one");
+ Assert.That(wire.Ended, Is.EqualTo(1), "a receipt that cannot be queued");
+
+ rig.Authority.Denied.Add(KeyB.Value);
+ await rig.Put(wire, "m2", KeyB, "two");
+ Assert.That(wire.Ended, Is.EqualTo(2), "a refusal that cannot be queued");
+
+ wire.Transport.AnswerUnreadableSchemaFrame(new byte[] { 0xC1 });
+ Assert.That(wire.Ended, Is.EqualTo(3), "an unreadable-frame answer that cannot be queued");
+ }
+
+ [Test]
+ public async Task AnUnexpectedFailureIsRefusedWithFixedTextNotTheExceptionsOwn()
+ {
+ var (rig, wire) = await Capable();
+ var message = rig.PutOf("m1", KeyA, "one");
+ message.Document!.Payload = [0xC1];
+ var legacy = new Wire();
+
+ await rig.PutHandler(message, wire.Peer);
+ await rig.PutHandler(message, legacy.Peer);
+ await rig.DeleteHandler(new CultNetDocumentDeleteMessage { MessageId = "d1", SchemaId = "no.such.schema", RecordKey = "k" }, legacy.Peer);
+
+ Assert.That(((CultNetErrorV1Message)wire.Sent.Single()).Error, Is.EqualTo(CultNetWriteReceiptSession.WriteFailedText));
+ Assert.That(((CultNetErrorMessage)legacy.Sent[0]).Error, Is.EqualTo(CultNetWriteReceiptSession.WriteFailedText));
+ }
+
+ [Test]
+ public void TheLimitReachedRefusalIsTheSpecVectorByteForByte()
+ {
+ const string vector = "86AD736368656D6156657273696F6EB063756C746E65742E6572726F722E7631A56572726F72D93B6C696D69745F726561636865643A207468652073657373696F6E27732072656D656D62657265642D7772697465206C696D69742069732066756C6CAB726F7574696E6748696E74C0A4636F6465AD6C696D69745F72656163686564A764657461696C7384A56669656C64C0A576616C7565C0A461734F66C0A763757272656E74C0A96D6573736167654964A87075742D31303235";
+
+ var bytes = CultNetSchemaMessageSerialization.Serialize(CultNetErrorV1Message.LimitReached("put-1025"));
+
+ Assert.That(Convert.ToHexString(bytes), Is.EqualTo(vector));
+ }
+
+ [Test]
+ public void TheSpecWireVectorsDecodeToTheirExpectedFields()
+ {
+ var admitted = (CultNetWriteAdmittedMessage)CultNetSchemaMessageSerialization.Deserialize(Convert.FromHexString(
+ "85AD736368656D6156657273696F6EB963756C746E65742E77726974655F61646D69747465642E7630A96D6573736167654964A57075742D31A96F7065726174696F6EA3707574A8736368656D614964A76E6F74652E7630A97265636F72644B6579A66E6F74653A31"));
+ Assert.That((admitted.MessageId, admitted.Operation, admitted.SchemaId, admitted.RecordKey),
+ Is.EqualTo(("put-1", "put", "note.v0", "note:1")));
+
+ var refused = (CultNetErrorV1Message)CultNetSchemaMessageSerialization.Deserialize(Convert.FromHexString(
+ "86AD736368656D6156657273696F6EB063756C746E65742E6572726F722E7631A56572726F72D93677726974655F6E6F745F617574686F72697A65643A2074686520636174616C6F6720726566757365642074686520646F63756D656E74AB726F7574696E6748696E74C0A4636F6465B477726974655F6E6F745F617574686F72697A6564A764657461696C7384A56669656C64A97265636F72644B6579A576616C7565A66E6F74653A32A461734F66C0A763757272656E74C0A96D6573736167654964A57075742D32"));
+ Assert.That(refused.Error, Is.EqualTo("write_not_authorized: the catalog refused the document"));
+ Assert.That(refused.Code, Is.EqualTo("write_not_authorized"));
+ Assert.That(refused.MessageId, Is.EqualTo("put-2"));
+ Assert.That((refused.Details!.Field, refused.Details.Value), Is.EqualTo(("recordKey", "note:2")));
+ Assert.That(refused.Details.AsOf, Is.Null);
+ Assert.That(refused.Details.Current, Is.Null);
}
[Test]
@@ -312,7 +460,8 @@ namespace GameCult.Networking.Tests
await rig.Put(wire, "over", KeyA, "x");
await rig.Put(wire, "w0", new CultRecordKey("tests:receipt:0"), "x");
- AssertRefused(wire.Sent[0], "over", null);
+ var limit = AssertRefused(wire.Sent[0], "over", "limit_reached");
+ Assert.That(limit.Error, Is.EqualTo("limit_reached: the session's remembered-write limit is full"));
AssertAdmitted(wire.Sent[1], "w0", "put", "tests:receipt:0");
Assert.That(rig.TextOf(KeyA), Is.Null);
}
@@ -446,11 +595,50 @@ namespace GameCult.Networking.Tests
AssertRefused(wire.Sent[1], "fwd-2", "message_invalid");
}
- [TestCase(true)]
- [TestCase(false)]
- public async Task AForwardThePrimaryNeverAnsweredRelaysNothingAndMayBeRetried(bool unconfirmed)
+ [Test]
+ public async Task AForwardThePrimaryLeftSilentIsNeverForwardedAgainOnTheSession()
{
- Exception failure = unconfirmed ? new CultNetWriteUnconfirmedException("fwd-1") : new TimeoutException("primary unreachable");
+ var forwarder = new ScriptedForwarder { OnPut = _ => Task.FromException(new CultNetWriteUnconfirmedException("fwd-1")) };
+ var rig = Build(primary: false, forwarder: forwarder);
+ var wire = new Wire();
+ await rig.Hello(wire, BothFrames);
+ wire.Sent.Clear();
+
+ await rig.Put(wire, "fwd-1", KeyA, "one");
+ forwarder.OnPut = _ => Task.CompletedTask;
+ await rig.Put(wire, "fwd-1", KeyA, "one");
+
+ Assert.That(wire.Sent, Is.Empty, "no receipt, no error: the write stays unconfirmed");
+ Assert.That(forwarder.Forwarded, Is.EqualTo(1), "the primary may have applied it, so it is not sent again");
+
+ await rig.Put(wire, "fwd-1", KeyB, "other");
+ AssertRefused(wire.Sent.Single(), "fwd-1", "message_invalid");
+ }
+
+ [Test]
+ public async Task AForwardRefusedByThePrimaryReplaysItsRefusalOnResend()
+ {
+ var forwarder = new ScriptedForwarder
+ {
+ OnPut = _ => Task.FromException(new CultNetWriteRefusedException(new CultNetErrorMessage { Error = "primary says no", Code = "write_not_authorized" }))
+ };
+ var rig = Build(primary: false, forwarder: forwarder);
+ var wire = new Wire();
+ await rig.Hello(wire, BothFrames);
+ wire.Sent.Clear();
+
+ await rig.Put(wire, "fwd-1", KeyA, "one");
+ 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(forwarder.Forwarded, Is.EqualTo(1));
+ }
+
+ [Test]
+ public async Task AForwardThatNeverReachedThePrimaryMayBeRetried()
+ {
+ Exception failure = new TimeoutException("primary unreachable");
var forwarder = new ScriptedForwarder { OnPut = _ => Task.FromException(failure) };
var rig = Build(primary: false, forwarder: forwarder);
var wire = new Wire();