C#: receipt server follows Heimdall's review of #61
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Failing after 5m30s
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Successful in 21s

Look the id up first and authorize only on first sighting; remember refusals within the cap; never re-forward an id whose primary stayed silent; end the connection when an answer cannot be queued; fixed refusal texts; limit_reached cap refusal; commit-point doc on ApplyPut/Delete.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Libby 2026-11-08 14:32:39 +00:00
commit ed5f9d30d1
6 changed files with 362 additions and 118 deletions

View file

@ -1017,6 +1017,8 @@ namespace GameCult.Networking
/// <summary>
/// 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.
/// Returning normally means the put is committed and visible to readers: a receipt-capable peer is told
/// <c>write_admitted</c> on the strength of it, so a write that has only been queued must not return.
/// </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>
@ -1047,6 +1049,7 @@ namespace GameCult.Networking
/// <summary>
/// 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.
/// Returning normally means the delete is committed and visible to readers (see <see cref="ApplyPutAsync(CultNetDocumentPutRawMessage)"/>).
/// </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>

View file

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

View file

@ -663,6 +663,18 @@ namespace GameCult.Networking
};
}
/// <summary>
/// The refusal of a write a session has no room to remember (write-receipt.md, rule 5): code <c>limit_reached</c>,
/// details present and all nil. Exists only as error.v1; v0 has no such code.
/// </summary>
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
};
/// <summary>Gets the v0 form, dropping the id.</summary>
public CultNetErrorMessage ToV0() => new CultNetErrorMessage
{

View file

@ -3534,6 +3534,7 @@ namespace GameCult.Networking
public sealed class LiteNetLibTransportConnection
{
private readonly Action<byte[]>? _deliver;
private readonly Action? _end;
private readonly CultNetTransportStats _stats = new CultNetTransportStats();
/// <summary>
@ -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<byte[]> deliver)
internal LiteNetLibTransportConnection(Action<byte[]> deliver, Action? end = null)
{
_deliver = deliver ?? throw new ArgumentNullException(nameof(deliver));
_end = end;
Profile = CultNetTransportProfiles.CreateLiteNetLib("csharp-litenetlib-peer");
}
/// <summary>Answers a schema frame that did not decode with a <c>message_invalid</c> error.v1 naming its id when it holds one.</summary>
internal void AnswerUnreadableSchemaFrame(byte[] payload) =>
SendSchema(CultNetErrorV1Message.From(
CultNetErrorMessage.ForMessageInvalid("message", "the message could not be read"),
CultNetReadableMessageId.From(payload)));
/// <summary>
/// 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 (<c>session_ended</c>), never as silence.
/// </summary>
internal void End() => _end?.Invoke();
/// <summary>
/// Answers a schema frame that did not decode with a <c>message_invalid</c> error.v1 naming its id when it holds one;
/// when that answer cannot be sent the connection ends instead.
/// </summary>
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;

View file

@ -48,14 +48,17 @@ namespace GameCult.Networking
}
/// <summary>
/// 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.
/// </summary>
internal sealed class CultNetWriteReceiptSession
{
/// <summary>The most writes a peer is remembered for; a peer at the limit has new writes refused.</summary>
/// <summary>The most writes a peer is remembered for, refused ones included; a peer at the limit has new writes refused.</summary>
internal const int MaxRememberedWrites = 1024;
/// <summary>What a write is refused with when it failed for a reason no typed refusal names. Never an exception's own text.</summary>
internal const string WriteFailedText = "the write could not be applied";
private readonly object _gate = new();
private readonly Dictionary<string, Entry> _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 }
/// <summary>Claims <paramref name="messageId"/> for a write with this content, or says what already holds it.</summary>
public Begin TryBegin(string messageId, byte[] hash)
/// <summary>
/// Looks <paramref name="messageId"/> up first. An id not seen is claimed for this content (<see cref="Begin.New"/>);
/// otherwise the outcome says what already holds it, and <paramref name="answer"/> is the answer to replay.
/// </summary>
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
}
}
/// <summary>The write committed: remember it so a resend is re-receipted.</summary>
public void Admitted(string messageId)
/// <summary>The write was answered, admitted or refused: remember the answer so a resend repeats it.</summary>
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;
}
}
/// <summary>The write was refused or unanswered: forget it so a corrected or repeated attempt is judged afresh.</summary>
/// <summary>
/// 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).
/// </summary>
public void LeftSilent(string messageId)
{
lock (_gate)
{
if (_writes.TryGetValue(messageId, out var entry)) entry.Silent = true;
}
}
/// <summary>The write never reached anyone who could have applied it: forget it so a repeat is judged afresh.</summary>
public void Abandoned(string messageId)
{
lock (_gate)
@ -139,18 +165,6 @@ namespace GameCult.Networking
}
}
/// <summary>A resend the receipt session answered without applying it again, or refused as a reused id.</summary>
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; }
}
/// <summary>
/// One write on a receipt-capable connection: its dedupe claim and its single answer (write-receipt.md, rules 1 to 5).
/// </summary>
@ -185,94 +199,87 @@ namespace GameCult.Networking
}
/// <summary>
/// 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
/// <see cref="CultNetWriteResendException"/> 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).
/// </summary>
public Action<string, CultRecordKey> Guard(Action<string, CultRecordKey> 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);
};
/// <summary>The write committed: remember it and send its one receipt.</summary>
/// <summary>The write committed: remember the receipt and send it.</summary>
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);
}
/// <summary>The write was refused: forget its claim so a repeat is judged afresh, and send its one error.v1.</summary>
/// <summary>The write was refused: remember the refusal (a resend repeats it) and send its one error.v1.</summary>
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);
}
/// <summary>No answer is owed or possible (a forward the primary never answered): forget the claim, send nothing.</summary>
/// <summary>
/// A forward its primary left unanswered: no answer is owed or possible, and the id is never forwarded again on this
/// session. Sends nothing.
/// </summary>
public void LeftSilent()
{
if (_answered) return;
_answered = true;
if (_claimed) _session.LeftSilent(_messageId);
}
/// <summary>A forward that never reached a primary that could have applied it: forget the claim, send nothing.</summary>
public void Unanswered()
{
if (_answered) return;
_answered = true;
Release();
}
/// <summary>Answers a resend per rule 5.</summary>
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;
}
}
/// <summary>Whether the write has had its answer (or has none owed).</summary>
public bool Settled => _answered;
/// <summary>A missing id cannot be answered under rule 1; refuse it with no id.</summary>
public bool HasId => _messageId.Length > 0;
private void Release()
{
if (_claimed) _session.Abandoned(_messageId);
_claimed = false;
}
/// <summary>A missing id cannot be answered under rule 1; refuse it with no id.</summary>
public bool HasId => _messageId.Length > 0;
}
}

View file

@ -23,20 +23,32 @@ namespace GameCult.Networking.Tests
private sealed class Authority : ICultNetRemoteWriteAuthority
{
public HashSet<string> 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<ICultNetSchemaMessage> Sent { get; } = new();
/// <summary>When set, the connection cannot queue what it is sent.</summary>
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("<Transport>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();