Ack Cut 1 tests: ordered delivery is held only behind a missing sequence
P1, P1b and P6 as asserting tests in Rust, C#, TypeScript and Python, plus the handshake-seeded watermark, duplicate and reset rules. They fail on the per-channel next pointer and pass on one contiguous watermark. The three tests that fed the receiver an impossible handshake (an Accept at sequence 91 for data that starts at 1) now use a real Connect and Accept, and the Rust channel-bound test holds its frames behind a gap. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
ce0d61fc71
commit
b25effebe4
7 changed files with 488 additions and 47 deletions
|
|
@ -731,12 +731,7 @@ class CultNetTests(unittest.TestCase):
|
|||
def test_cultnet_rudp_session_suppresses_duplicates_and_delivers_reliable_ordered_payloads(self) -> None:
|
||||
sender = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=123, initial_sequence=1))
|
||||
receiver = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=123, initial_sequence=100))
|
||||
sender.receive(
|
||||
CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, 123, 90, 0, 0, "control")
|
||||
)
|
||||
receiver.receive(
|
||||
CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, 123, 91, 0, 0, "control")
|
||||
)
|
||||
sender.receive(receiver.accept_connect(sender.create_connect(0), 0))
|
||||
|
||||
first = sender.send("schema", b"first", CultNetRudpSendOptions(reliable=True, ordered=True))
|
||||
second = sender.send("schema", b"second", CultNetRudpSendOptions(reliable=True, ordered=True))
|
||||
|
|
|
|||
109
packages/cultnet-py/tests/test_rudp_ordered_delivery.py
Normal file
109
packages/cultnet-py/tests/test_rudp_ordered_delivery.py
Normal file
|
|
@ -0,0 +1,109 @@
|
|||
"""Ordered delivery has one owner: the contiguous received-through watermark.
|
||||
|
||||
A frame is held for exactly one reason, a reliable sequence below it has not
|
||||
arrived, and it is delivered in the call that fills the last such gap,
|
||||
whichever channel the gap belonged to.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
from cultnet_py.transport import (
|
||||
CultNetRudpPacket,
|
||||
CultNetRudpPacketType,
|
||||
CultNetRudpSendOptions,
|
||||
CultNetRudpSession,
|
||||
CultNetRudpSessionOptions,
|
||||
)
|
||||
|
||||
|
||||
def handshake() -> tuple[CultNetRudpSession, CultNetRudpSession]:
|
||||
client = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=410, initial_sequence=1))
|
||||
server = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=410, initial_sequence=500))
|
||||
accept = server.accept_connect(client.create_connect(0), 0)
|
||||
client.receive(accept, 0)
|
||||
return client, server
|
||||
|
||||
|
||||
def send(session: CultNetRudpSession, channel: str, payload: str, ordered: bool) -> CultNetRudpPacket:
|
||||
return session.send(channel, payload.encode(), CultNetRudpSendOptions(reliable=True, ordered=ordered))
|
||||
|
||||
|
||||
def names(result) -> list[str]:
|
||||
return [frame.payload.decode() for frame in result.delivered]
|
||||
|
||||
|
||||
class CultNetRudpOrderedDeliveryTests(unittest.TestCase):
|
||||
def test_ordered_frame_waits_for_a_gap_filled_by_another_channel(self) -> None:
|
||||
sender, receiver = handshake()
|
||||
o1 = send(sender, "schema", "o1", True)
|
||||
u = send(sender, "rel", "u", False)
|
||||
o2 = send(sender, "schema", "o2", True)
|
||||
o3 = send(sender, "schema", "o3", True)
|
||||
|
||||
self.assertEqual(names(receiver.receive(o1, 1)), ["o1"])
|
||||
self.assertEqual(names(receiver.receive(o2, 2)), [], "u is missing")
|
||||
self.assertEqual(names(receiver.receive(o3, 3)), [], "u is missing")
|
||||
self.assertEqual(names(receiver.receive(u, 4)), ["u", "o2", "o3"])
|
||||
|
||||
def test_two_ordered_channels_release_each_other_in_sequence_order(self) -> None:
|
||||
sender, receiver = handshake()
|
||||
a1 = send(sender, "schema", "A1", True)
|
||||
b1 = send(sender, "other", "B1", True)
|
||||
a2 = send(sender, "schema", "A2", True)
|
||||
a3 = send(sender, "schema", "A3", True)
|
||||
|
||||
delivered = names(receiver.receive(a1, 1)) + names(receiver.receive(a2, 2)) + names(receiver.receive(a3, 3))
|
||||
self.assertEqual(delivered, ["A1"], "B1 is missing, so A2 and A3 wait")
|
||||
self.assertEqual(names(receiver.receive(b1, 4)), ["B1", "A2", "A3"])
|
||||
|
||||
def test_channel_first_used_after_other_traffic_loses_nothing(self) -> None:
|
||||
sender, receiver = handshake()
|
||||
c1 = send(sender, "late", "C1", True)
|
||||
c2 = send(sender, "late", "C2", True)
|
||||
x = send(sender, "schema", "X", True)
|
||||
|
||||
self.assertEqual(names(receiver.receive(x, 1)), [], "C1 is missing")
|
||||
self.assertEqual(names(receiver.receive(c2, 2)), [], "C1 is missing")
|
||||
self.assertEqual(names(receiver.receive(c1, 3)), ["C1", "C2", "X"])
|
||||
|
||||
def test_accept_seeds_the_watermark_of_the_connecting_side(self) -> None:
|
||||
client, server = handshake()
|
||||
s1 = send(server, "schema", "s1", True)
|
||||
s2 = send(server, "schema", "s2", True)
|
||||
|
||||
self.assertEqual(names(client.receive(s2, 1)), [])
|
||||
self.assertEqual(names(client.receive(s1, 2)), ["s1", "s2"])
|
||||
|
||||
def test_connect_seeds_the_watermark_of_the_accepting_side(self) -> None:
|
||||
client, server = handshake()
|
||||
first = send(client, "schema", "first", True)
|
||||
second = send(client, "schema", "second", True)
|
||||
|
||||
self.assertEqual(names(server.receive(second, 1)), [])
|
||||
self.assertEqual(names(server.receive(first, 2)), ["first", "second"])
|
||||
|
||||
def test_duplicate_of_a_held_frame_is_not_delivered_twice(self) -> None:
|
||||
sender, receiver = handshake()
|
||||
s1 = send(sender, "schema", "s1", True)
|
||||
s2 = send(sender, "schema", "s2", True)
|
||||
|
||||
self.assertEqual(names(receiver.receive(s2, 1)), [])
|
||||
self.assertEqual(names(receiver.receive(s2, 2)), [])
|
||||
self.assertEqual(names(receiver.receive(s1, 3)), ["s1", "s2"])
|
||||
self.assertEqual(names(receiver.receive(s2, 4)), [])
|
||||
|
||||
def test_reset_peer_state_forgets_held_frames_and_the_watermark(self) -> None:
|
||||
sender, receiver = handshake()
|
||||
s1 = send(sender, "schema", "s1", True)
|
||||
s2 = send(sender, "schema", "s2", True)
|
||||
self.assertEqual(names(receiver.receive(s2, 1)), [])
|
||||
|
||||
receiver.reset_peer_state()
|
||||
receiver.receive(CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, 410, 1, 0, 0, "control"), 2)
|
||||
|
||||
self.assertEqual(names(receiver.receive(s1, 3)), ["s1"], "the held s2 died with the reset")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
|
@ -1009,42 +1009,9 @@ fn rudp_session_suppresses_duplicates_and_delivers_reliable_ordered_payloads_in_
|
|||
initial_sequence: 100,
|
||||
..CultNetRudpSessionOptions::default()
|
||||
});
|
||||
sender.receive(
|
||||
&CultNetRudpPacket {
|
||||
packet_type: CultNetRudpPacketType::Accept,
|
||||
connection_id: 123,
|
||||
sequence: 90,
|
||||
ack: 0,
|
||||
ack_mask: 0,
|
||||
channel_id: "control".to_string(),
|
||||
reliable: false,
|
||||
ordered: false,
|
||||
sequenced: false,
|
||||
fragment_id: 0,
|
||||
fragment_index: 0,
|
||||
fragment_count: 0,
|
||||
payload: Vec::new(),
|
||||
},
|
||||
0,
|
||||
)?;
|
||||
receiver.receive(
|
||||
&CultNetRudpPacket {
|
||||
packet_type: CultNetRudpPacketType::Accept,
|
||||
connection_id: 123,
|
||||
sequence: 91,
|
||||
ack: 0,
|
||||
ack_mask: 0,
|
||||
channel_id: "control".to_string(),
|
||||
reliable: false,
|
||||
ordered: false,
|
||||
sequenced: false,
|
||||
fragment_id: 0,
|
||||
fragment_index: 0,
|
||||
fragment_count: 0,
|
||||
payload: Vec::new(),
|
||||
},
|
||||
0,
|
||||
)?;
|
||||
let connect = sender.create_connect(0, Vec::new())?;
|
||||
let accept = receiver.accept_connect(&connect, 0, Vec::new())?;
|
||||
sender.receive(&accept, 0)?;
|
||||
|
||||
let options = CultNetRudpSendOptions {
|
||||
reliable: true,
|
||||
|
|
@ -2667,6 +2634,18 @@ fn rudp_ordered_channel_tracking_is_bounded() -> Result<()> {
|
|||
let accept = receiver.accept_connect(&connect, 1, Vec::new())?;
|
||||
sender.receive(&accept, 2)?;
|
||||
|
||||
// Channels are tracked only while they hold a frame, so the bound is on
|
||||
// channels held behind one missing sequence.
|
||||
let gap = sender.send(
|
||||
"gap",
|
||||
b"lost".to_vec(),
|
||||
CultNetRudpSendOptions {
|
||||
reliable: true,
|
||||
ordered: false,
|
||||
..CultNetRudpSendOptions::default()
|
||||
},
|
||||
)?;
|
||||
sender.receive(&receiver.create_ack_for(gap.sequence), 3)?;
|
||||
let mut error = None;
|
||||
for channel in 0..256u32 {
|
||||
let packet = sender.send(
|
||||
|
|
@ -2679,7 +2658,8 @@ fn rudp_ordered_channel_tracking_is_bounded() -> Result<()> {
|
|||
},
|
||||
)?;
|
||||
match receiver.receive(&packet, 3) {
|
||||
Ok(_) => {
|
||||
Ok(result) => {
|
||||
assert!(result.delivered.is_empty(), "the gap holds every frame");
|
||||
sender.receive(&receiver.create_ack_for(packet.sequence), 3)?;
|
||||
}
|
||||
Err(refused) => {
|
||||
|
|
@ -2743,6 +2723,134 @@ fn rudp_ordered_hold_buffer_is_bounded() -> Result<()> {
|
|||
Ok(())
|
||||
}
|
||||
|
||||
fn rudp_ordered_pair() -> Result<(CultNetRudpSession, CultNetRudpSession)> {
|
||||
let mut client = CultNetRudpSession::new(CultNetRudpSessionOptions {
|
||||
connection_id: 410,
|
||||
initial_sequence: 1,
|
||||
..CultNetRudpSessionOptions::default()
|
||||
});
|
||||
let mut server = CultNetRudpSession::new(CultNetRudpSessionOptions {
|
||||
connection_id: 410,
|
||||
initial_sequence: 500,
|
||||
..CultNetRudpSessionOptions::default()
|
||||
});
|
||||
let connect = client.create_connect(0, Vec::new())?;
|
||||
let accept = server.accept_connect(&connect, 0, Vec::new())?;
|
||||
client.receive(&accept, 0)?;
|
||||
Ok((client, server))
|
||||
}
|
||||
|
||||
fn rudp_send_options(ordered: bool) -> CultNetRudpSendOptions {
|
||||
CultNetRudpSendOptions {
|
||||
reliable: true,
|
||||
ordered,
|
||||
..CultNetRudpSendOptions::default()
|
||||
}
|
||||
}
|
||||
|
||||
fn rudp_delivered_names(delivered: &[cultnet_rs::CultNetRudpDeliveredFrame]) -> Vec<String> {
|
||||
delivered
|
||||
.iter()
|
||||
.map(|frame| String::from_utf8(frame.payload.clone()).unwrap())
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rudp_ordered_frame_waits_for_a_gap_filled_by_another_channel() -> Result<()> {
|
||||
let (mut sender, mut receiver) = rudp_ordered_pair()?;
|
||||
let o1 = sender.send("schema", b"o1".to_vec(), rudp_send_options(true))?;
|
||||
let u = sender.send("media", b"u".to_vec(), rudp_send_options(false))?;
|
||||
let o2 = sender.send("schema", b"o2".to_vec(), rudp_send_options(true))?;
|
||||
let o3 = sender.send("schema", b"o3".to_vec(), rudp_send_options(true))?;
|
||||
|
||||
assert_eq!(rudp_delivered_names(&receiver.receive(&o1, 1)?.delivered), ["o1"]);
|
||||
// `u` is lost, so nothing behind it may be delivered.
|
||||
assert!(receiver.receive(&o2, 2)?.delivered.is_empty());
|
||||
assert!(receiver.receive(&o3, 3)?.delivered.is_empty());
|
||||
// The other channel's packet fills the gap and releases the ordered frames
|
||||
// in the same call.
|
||||
assert_eq!(
|
||||
rudp_delivered_names(&receiver.receive(&u, 4)?.delivered),
|
||||
["u", "o2", "o3"]
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rudp_two_ordered_channels_release_each_other_in_sequence_order() -> Result<()> {
|
||||
let (mut sender, mut receiver) = rudp_ordered_pair()?;
|
||||
let a1 = sender.send("schema", b"A1".to_vec(), rudp_send_options(true))?;
|
||||
let b1 = sender.send("other", b"B1".to_vec(), rudp_send_options(true))?;
|
||||
let a2 = sender.send("schema", b"A2".to_vec(), rudp_send_options(true))?;
|
||||
let a3 = sender.send("schema", b"A3".to_vec(), rudp_send_options(true))?;
|
||||
|
||||
let mut delivered = rudp_delivered_names(&receiver.receive(&a1, 1)?.delivered);
|
||||
delivered.extend(rudp_delivered_names(&receiver.receive(&a2, 2)?.delivered));
|
||||
delivered.extend(rudp_delivered_names(&receiver.receive(&a3, 3)?.delivered));
|
||||
assert_eq!(delivered, ["A1"], "B1 is missing, so A2 and A3 wait");
|
||||
assert_eq!(
|
||||
rudp_delivered_names(&receiver.receive(&b1, 4)?.delivered),
|
||||
["B1", "A2", "A3"]
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rudp_channel_first_used_after_other_traffic_loses_nothing() -> Result<()> {
|
||||
let (mut sender, mut receiver) = rudp_ordered_pair()?;
|
||||
let c1 = sender.send("late", b"C1".to_vec(), rudp_send_options(true))?;
|
||||
let c2 = sender.send("late", b"C2".to_vec(), rudp_send_options(true))?;
|
||||
let x = sender.send("schema", b"X".to_vec(), rudp_send_options(true))?;
|
||||
|
||||
// X, then C2, arrive while C1 is still in flight.
|
||||
let mut delivered = rudp_delivered_names(&receiver.receive(&x, 1)?.delivered);
|
||||
delivered.extend(rudp_delivered_names(&receiver.receive(&c2, 2)?.delivered));
|
||||
assert!(delivered.is_empty(), "C1 is missing, so nothing is delivered");
|
||||
assert_eq!(
|
||||
rudp_delivered_names(&receiver.receive(&c1, 3)?.delivered),
|
||||
["C1", "C2", "X"]
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rudp_accept_seeds_the_watermark_of_the_connecting_side() -> Result<()> {
|
||||
let (mut client, mut server) = rudp_ordered_pair()?;
|
||||
let s1 = server.send("schema", b"s1".to_vec(), rudp_send_options(true))?;
|
||||
let s2 = server.send("schema", b"s2".to_vec(), rudp_send_options(true))?;
|
||||
assert!(client.receive(&s2, 1)?.delivered.is_empty());
|
||||
assert_eq!(
|
||||
rudp_delivered_names(&client.receive(&s1, 2)?.delivered),
|
||||
["s1", "s2"]
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rudp_large_ordered_frame_without_a_gap_is_never_held_or_refused() -> Result<()> {
|
||||
let (mut sender, mut receiver) = rudp_ordered_pair()?;
|
||||
let packets = sender.send_many(
|
||||
"schema",
|
||||
vec![7u8; 5 * 1024 * 1024],
|
||||
rudp_send_options(true),
|
||||
Some(60_000),
|
||||
)?;
|
||||
let mut queue = packets;
|
||||
let mut delivered = Vec::new();
|
||||
while !queue.is_empty() {
|
||||
let mut next = Vec::new();
|
||||
for packet in &queue {
|
||||
delivered.extend(receiver.receive(packet, 1)?.delivered);
|
||||
let ack = receiver.create_ack_for_received(packet.sequence);
|
||||
next.extend(sender.receive(&ack, 1)?.ready_to_send);
|
||||
}
|
||||
queue = next;
|
||||
}
|
||||
assert_eq!(delivered.len(), 1);
|
||||
assert_eq!(delivered[0].payload.len(), 5 * 1024 * 1024);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rudp_far_ahead_sequence_cannot_starve_the_session() -> Result<()> {
|
||||
let mut sender = CultNetRudpSession::new(CultNetRudpSessionOptions {
|
||||
|
|
|
|||
|
|
@ -628,8 +628,7 @@ test("rudp session bounds pending reliable packets before enqueue", () => {
|
|||
test("rudp session suppresses duplicates and delivers reliable ordered payloads in sequence", () => {
|
||||
const sender = new CultNetRudpSession({ connectionId: 123, initialSequence: 1 });
|
||||
const receiver = new CultNetRudpSession({ connectionId: 123, initialSequence: 100 });
|
||||
sender.receive({ packetType: "accept", connectionId: 123, sequence: 90, ack: 0, ackMask: 0, channelId: "control" });
|
||||
receiver.receive({ packetType: "accept", connectionId: 123, sequence: 91, ack: 0, ackMask: 0, channelId: "control" });
|
||||
sender.receive(receiver.acceptConnect(sender.createConnect(0), 0));
|
||||
|
||||
const first = sender.send("schema", Buffer.from("first"), { reliable: true, ordered: true });
|
||||
const second = sender.send("schema", Buffer.from("second"), { reliable: true, ordered: true });
|
||||
|
|
|
|||
103
packages/cultnet-ts/test/rudp-ordered-delivery.test.ts
Normal file
103
packages/cultnet-ts/test/rudp-ordered-delivery.test.ts
Normal file
|
|
@ -0,0 +1,103 @@
|
|||
// Ordered delivery has one owner: the contiguous received-through watermark. A
|
||||
// frame is held for exactly one reason, a reliable sequence below it has not
|
||||
// arrived, and it is delivered in the call that fills the last such gap,
|
||||
// whichever channel the gap belonged to.
|
||||
import assert from "node:assert/strict";
|
||||
import test from "node:test";
|
||||
import { CultNetRudpSession } from "../src";
|
||||
|
||||
const enc = (text: string) => new TextEncoder().encode(text);
|
||||
const dec = (bytes: Uint8Array) => new TextDecoder().decode(bytes);
|
||||
|
||||
function handshake(): { client: CultNetRudpSession; server: CultNetRudpSession } {
|
||||
const client = new CultNetRudpSession({ connectionId: 410, initialSequence: 1 });
|
||||
const server = new CultNetRudpSession({ connectionId: 410, initialSequence: 500 });
|
||||
const accept = server.acceptConnect(client.createConnect(0), 0);
|
||||
client.receive(accept, 0);
|
||||
return { client, server };
|
||||
}
|
||||
|
||||
const names = (result: { delivered: { payload: Uint8Array }[] }) => result.delivered.map((frame) => dec(frame.payload));
|
||||
const send = (session: CultNetRudpSession, channel: string, payload: string, ordered: boolean) =>
|
||||
session.send(channel, enc(payload), { reliable: true, ordered });
|
||||
|
||||
test("an ordered frame waits for a gap filled by another channel", () => {
|
||||
const { client: sender, server: receiver } = handshake();
|
||||
const o1 = send(sender, "schema", "o1", true);
|
||||
const u = send(sender, "rel", "u", false);
|
||||
const o2 = send(sender, "schema", "o2", true);
|
||||
const o3 = send(sender, "schema", "o3", true);
|
||||
|
||||
assert.deepEqual(names(receiver.receive(o1, 1)), ["o1"]);
|
||||
assert.deepEqual(names(receiver.receive(o2, 2)), [], "u is missing");
|
||||
assert.deepEqual(names(receiver.receive(o3, 3)), [], "u is missing");
|
||||
assert.deepEqual(names(receiver.receive(u, 4)), ["u", "o2", "o3"]);
|
||||
});
|
||||
|
||||
test("two ordered channels release each other in sequence order", () => {
|
||||
const { client: sender, server: receiver } = handshake();
|
||||
const a1 = send(sender, "schema", "A1", true);
|
||||
const b1 = send(sender, "other", "B1", true);
|
||||
const a2 = send(sender, "schema", "A2", true);
|
||||
const a3 = send(sender, "schema", "A3", true);
|
||||
|
||||
const delivered = [
|
||||
...names(receiver.receive(a1, 1)),
|
||||
...names(receiver.receive(a2, 2)),
|
||||
...names(receiver.receive(a3, 3)),
|
||||
];
|
||||
assert.deepEqual(delivered, ["A1"], "B1 is missing, so A2 and A3 wait");
|
||||
assert.deepEqual(names(receiver.receive(b1, 4)), ["B1", "A2", "A3"]);
|
||||
});
|
||||
|
||||
test("a channel first used after other traffic loses nothing", () => {
|
||||
const { client: sender, server: receiver } = handshake();
|
||||
const c1 = send(sender, "late", "C1", true);
|
||||
const c2 = send(sender, "late", "C2", true);
|
||||
const x = send(sender, "schema", "X", true);
|
||||
|
||||
assert.deepEqual(names(receiver.receive(x, 1)), [], "C1 is missing");
|
||||
assert.deepEqual(names(receiver.receive(c2, 2)), [], "C1 is missing");
|
||||
assert.deepEqual(names(receiver.receive(c1, 3)), ["C1", "C2", "X"]);
|
||||
});
|
||||
|
||||
test("Accept seeds the watermark of the connecting side", () => {
|
||||
const { client, server } = handshake();
|
||||
const s1 = send(server, "schema", "s1", true);
|
||||
const s2 = send(server, "schema", "s2", true);
|
||||
|
||||
assert.deepEqual(names(client.receive(s2, 1)), []);
|
||||
assert.deepEqual(names(client.receive(s1, 2)), ["s1", "s2"]);
|
||||
});
|
||||
|
||||
test("Connect seeds the watermark of the accepting side", () => {
|
||||
const { client, server } = handshake();
|
||||
const first = send(client, "schema", "first", true);
|
||||
const second = send(client, "schema", "second", true);
|
||||
|
||||
assert.deepEqual(names(server.receive(second, 1)), []);
|
||||
assert.deepEqual(names(server.receive(first, 2)), ["first", "second"]);
|
||||
});
|
||||
|
||||
test("a duplicate of a held frame is not delivered twice", () => {
|
||||
const { client: sender, server: receiver } = handshake();
|
||||
const s1 = send(sender, "schema", "s1", true);
|
||||
const s2 = send(sender, "schema", "s2", true);
|
||||
|
||||
assert.deepEqual(names(receiver.receive(s2, 1)), []);
|
||||
assert.deepEqual(names(receiver.receive(s2, 2)), []);
|
||||
assert.deepEqual(names(receiver.receive(s1, 3)), ["s1", "s2"]);
|
||||
assert.deepEqual(names(receiver.receive(s2, 4)), []);
|
||||
});
|
||||
|
||||
test("resetting peer state forgets held frames and the watermark", () => {
|
||||
const { client: sender, server: receiver } = handshake();
|
||||
const s1 = send(sender, "schema", "s1", true);
|
||||
const s2 = send(sender, "schema", "s2", true);
|
||||
assert.deepEqual(names(receiver.receive(s2, 1)), []);
|
||||
|
||||
receiver.resetPeerState();
|
||||
receiver.receive({ packetType: "accept", connectionId: 410, sequence: 1, ack: 0, ackMask: 0, channelId: "control" }, 2);
|
||||
|
||||
assert.deepEqual(names(receiver.receive(s1, 3)), ["s1"], "the held s2 died with the reset");
|
||||
});
|
||||
|
|
@ -0,0 +1,127 @@
|
|||
#nullable enable
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Text;
|
||||
using NUnit.Framework;
|
||||
|
||||
namespace GameCult.Networking.Tests
|
||||
{
|
||||
/// <summary>
|
||||
/// Ordered delivery has one owner: the contiguous received-through watermark. A frame is held for
|
||||
/// exactly one reason, a reliable sequence below it has not arrived, and it is delivered in the call
|
||||
/// that fills the last such gap, whichever channel the gap belonged to.
|
||||
/// </summary>
|
||||
public sealed class CultNetRudpOrderedDeliveryTests
|
||||
{
|
||||
private static (CultNetRudpSession Client, CultNetRudpSession Server) Handshake()
|
||||
{
|
||||
var client = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = 410, InitialSequence = 1 });
|
||||
var server = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = 410, InitialSequence = 500 });
|
||||
var accept = server.AcceptConnect(client.CreateConnect(0), 0);
|
||||
client.Receive(accept, 0);
|
||||
return (client, server);
|
||||
}
|
||||
|
||||
private static CultNetRudpSendOptions Reliable(bool ordered) => new CultNetRudpSendOptions { Reliable = true, Ordered = ordered };
|
||||
|
||||
private static CultNetRudpPacket Send(CultNetRudpSession session, string channel, string payload, bool ordered)
|
||||
=> session.Send(channel, Encoding.UTF8.GetBytes(payload), Reliable(ordered));
|
||||
|
||||
private static string[] Names(CultNetRudpReceiveResult result)
|
||||
=> result.Delivered.Select(frame => Encoding.UTF8.GetString(frame.Payload)).ToArray();
|
||||
|
||||
[Test]
|
||||
public void OrderedFrameWaitsForAGapFilledByAnotherChannel()
|
||||
{
|
||||
var (sender, receiver) = Handshake();
|
||||
var o1 = Send(sender, "schema", "o1", true);
|
||||
var u = Send(sender, "rel", "u", false);
|
||||
var o2 = Send(sender, "schema", "o2", true);
|
||||
var o3 = Send(sender, "schema", "o3", true);
|
||||
|
||||
Assert.That(Names(receiver.Receive(o1, 1)), Is.EqualTo(new[] { "o1" }));
|
||||
Assert.That(Names(receiver.Receive(o2, 2)), Is.Empty, "u is missing");
|
||||
Assert.That(Names(receiver.Receive(o3, 3)), Is.Empty, "u is missing");
|
||||
Assert.That(Names(receiver.Receive(u, 4)), Is.EqualTo(new[] { "u", "o2", "o3" }));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void TwoOrderedChannelsReleaseEachOtherInSequenceOrder()
|
||||
{
|
||||
var (sender, receiver) = Handshake();
|
||||
var a1 = Send(sender, "schema", "A1", true);
|
||||
var b1 = Send(sender, "other", "B1", true);
|
||||
var a2 = Send(sender, "schema", "A2", true);
|
||||
var a3 = Send(sender, "schema", "A3", true);
|
||||
|
||||
var delivered = new List<string>();
|
||||
delivered.AddRange(Names(receiver.Receive(a1, 1)));
|
||||
delivered.AddRange(Names(receiver.Receive(a2, 2)));
|
||||
delivered.AddRange(Names(receiver.Receive(a3, 3)));
|
||||
Assert.That(delivered, Is.EqualTo(new[] { "A1" }), "B1 is missing, so A2 and A3 wait");
|
||||
Assert.That(Names(receiver.Receive(b1, 4)), Is.EqualTo(new[] { "B1", "A2", "A3" }));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void ChannelFirstUsedAfterOtherTrafficLosesNothing()
|
||||
{
|
||||
var (sender, receiver) = Handshake();
|
||||
var c1 = Send(sender, "late", "C1", true);
|
||||
var c2 = Send(sender, "late", "C2", true);
|
||||
var x = Send(sender, "schema", "X", true);
|
||||
|
||||
Assert.That(Names(receiver.Receive(x, 1)), Is.Empty, "C1 is missing");
|
||||
Assert.That(Names(receiver.Receive(c2, 2)), Is.Empty, "C1 is missing");
|
||||
Assert.That(Names(receiver.Receive(c1, 3)), Is.EqualTo(new[] { "C1", "C2", "X" }));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void AcceptSeedsTheWatermarkOfTheConnectingSide()
|
||||
{
|
||||
var (client, server) = Handshake();
|
||||
var s1 = Send(server, "schema", "s1", true);
|
||||
var s2 = Send(server, "schema", "s2", true);
|
||||
|
||||
Assert.That(Names(client.Receive(s2, 1)), Is.Empty);
|
||||
Assert.That(Names(client.Receive(s1, 2)), Is.EqualTo(new[] { "s1", "s2" }));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void ConnectSeedsTheWatermarkOfTheAcceptingSide()
|
||||
{
|
||||
var (client, server) = Handshake();
|
||||
var first = Send(client, "schema", "first", true);
|
||||
var second = Send(client, "schema", "second", true);
|
||||
|
||||
Assert.That(Names(server.Receive(second, 1)), Is.Empty);
|
||||
Assert.That(Names(server.Receive(first, 2)), Is.EqualTo(new[] { "first", "second" }));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void DuplicateOfAHeldFrameIsNotDeliveredTwice()
|
||||
{
|
||||
var (sender, receiver) = Handshake();
|
||||
var s1 = Send(sender, "schema", "s1", true);
|
||||
var s2 = Send(sender, "schema", "s2", true);
|
||||
|
||||
Assert.That(Names(receiver.Receive(s2, 1)), Is.Empty);
|
||||
Assert.That(Names(receiver.Receive(s2, 2)), Is.Empty);
|
||||
Assert.That(Names(receiver.Receive(s1, 3)), Is.EqualTo(new[] { "s1", "s2" }));
|
||||
Assert.That(Names(receiver.Receive(s2, 4)), Is.Empty);
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void ResetPeerStateForgetsHeldFramesAndTheWatermark()
|
||||
{
|
||||
var (sender, receiver) = Handshake();
|
||||
var s1 = Send(sender, "schema", "s1", true);
|
||||
var s2 = Send(sender, "schema", "s2", true);
|
||||
Assert.That(Names(receiver.Receive(s2, 1)), Is.Empty);
|
||||
|
||||
receiver.ResetPeerState();
|
||||
receiver.Receive(new CultNetRudpPacket { PacketType = CultNetRudpPacketType.Accept, ConnectionId = 410, Sequence = 1, ChannelId = "control" }, 2);
|
||||
|
||||
Assert.That(Names(receiver.Receive(s1, 3)), Is.EqualTo(new[] { "s1" }), "the held s2 died with the reset");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1032,8 +1032,8 @@ namespace GameCult.Networking.Tests
|
|||
ConnectionId = 123,
|
||||
InitialSequence = 100
|
||||
});
|
||||
sender.Receive(new CultNetRudpPacket { PacketType = CultNetRudpPacketType.Accept, ConnectionId = 123, Sequence = 90, ChannelId = "control" });
|
||||
receiver.Receive(new CultNetRudpPacket { PacketType = CultNetRudpPacketType.Accept, ConnectionId = 123, Sequence = 91, ChannelId = "control" });
|
||||
var accept = receiver.AcceptConnect(sender.CreateConnect(0), 0);
|
||||
sender.Receive(accept);
|
||||
|
||||
var first = sender.Send("schema", Encoding.UTF8.GetBytes("first"), new CultNetRudpSendOptions { Reliable = true, Ordered = true });
|
||||
var second = sender.Send("schema", Encoding.UTF8.GetBytes("second"), new CultNetRudpSendOptions { Reliable = true, Ordered = true });
|
||||
|
|
|
|||
Loading…
Reference in a new issue