Ack Cut 1d batch 2 (TypeScript): an Ack does not retire a Connect awaiting its Accept; the fresh attempt jumps by the receive window less one
While a Connect awaits its Accept, only an Accept the client honours retires it. An Ack that names the Connect (a server's reply to a repeat or a stale copy) no longer wedges the client with nothing to resend. The Connect payload lives on the session, so a fresh attempt can always be built. The fresh attempt uses the abandoned sequence plus the receive window less one (serial), so a late copy of the abandoned Connect stays inside the new one's stale window; only a session's first Connect draws at random. The cultmesh-ts document server has a test that it does not acknowledge by name a reliable packet its session did not receive. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
7b46b4c481
commit
9c562adaf2
3 changed files with 180 additions and 22 deletions
|
|
@ -360,6 +360,41 @@ test("a client restarted on the same address with the same Connect payload repla
|
|||
}
|
||||
});
|
||||
|
||||
test("the RUDP document server does not acknowledge by name a reliable packet its session did not receive", async () => {
|
||||
const connectionId = 0x43554c55;
|
||||
const server = CultMesh.createRudpDocumentServer("odin-ack-test", connectionId, {
|
||||
bindPort: 0,
|
||||
documents: new CultNetDocumentRegistry(),
|
||||
onError: () => undefined,
|
||||
});
|
||||
await server.start();
|
||||
const socket = createSocket("udp4");
|
||||
const received: CultNetRudpPacket[] = [];
|
||||
socket.on("message", wire => received.push(decodeRudpPacket(wire)));
|
||||
const toServer = (packet: CultNetRudpPacket) =>
|
||||
socket.send(encodeRudpPacket(packet), server.bind.port, "127.0.0.1");
|
||||
try {
|
||||
const client = new CultNetRudpSession({ connectionId, initialSequence: 50_000 });
|
||||
const connect = client.createConnect(Date.now());
|
||||
toServer(connect);
|
||||
await waitFor(() => received.some(packet => packet.packetType === "accept"), "the Accept");
|
||||
received.length = 0;
|
||||
|
||||
// A reliable packet the session ignores (a client does not send an Accept).
|
||||
// The session never received it, so the Ack must not name it.
|
||||
const ignored: CultNetRudpPacket = { ...client.createAck(), packetType: "accept", reliable: true, sequence: connect.sequence + 5 };
|
||||
toServer(ignored);
|
||||
await waitFor(() => received.some(packet => packet.packetType === "ack"), "the acknowledgement");
|
||||
const ack = received.find(packet => packet.packetType === "ack")!;
|
||||
const names = ack.ack === ignored.sequence
|
||||
|| Array.from({ length: 32 }, (_, bit) => bit).some(bit => (ack.ackMask & (1 << bit)) !== 0 && ack.ack - bit - 1 === ignored.sequence);
|
||||
assert.equal(names, false, "the server acknowledged by name a packet its session refused");
|
||||
} finally {
|
||||
socket.close();
|
||||
server.close();
|
||||
}
|
||||
});
|
||||
|
||||
function deferred<T>(): { promise: Promise<T>; resolve(value: T): void } {
|
||||
let resolve!: (value: T) => void;
|
||||
return { promise: new Promise<T>(done => resolve = done), resolve };
|
||||
|
|
|
|||
|
|
@ -25,14 +25,11 @@ const RUDP_RECEIVED_SEQUENCE_WINDOW = 4_096;
|
|||
/// retransmitting the same sequence would never change its mind.
|
||||
const RUDP_CONNECT_ATTEMPT_MS = 3_000;
|
||||
|
||||
/// A sequence drawn from a secure source in [floor, 2^31). The Connect's
|
||||
/// sequence is what tells a peer whether a Connect repeats one it already
|
||||
/// accepted or starts a new session, so two sessions must not share one by
|
||||
/// default. A floor above the range draws nothing new: the caller keeps its own
|
||||
/// sequence.
|
||||
function drawSequence(floor: number): number {
|
||||
const low = Math.max(1, floor);
|
||||
return low >= 2 ** 31 ? low : randomInt(low, 2 ** 31);
|
||||
/// A sequence drawn from a secure source in [1, 2^31). The Connect's sequence
|
||||
/// is what tells a peer whether a Connect repeats one it already accepted or
|
||||
/// starts a new session, so two sessions must not share one by default.
|
||||
function drawSequence(): number {
|
||||
return randomInt(1, 2 ** 31);
|
||||
}
|
||||
|
||||
/// Whether `sequence` is at or before `mark` in serial order, within the
|
||||
|
|
@ -189,6 +186,9 @@ export class CultNetRudpSession {
|
|||
#connectSequence: number | undefined;
|
||||
/// A Connect this side sent is unanswered. Only then is an Accept honoured.
|
||||
#awaitingAccept = false;
|
||||
/// The payload of the Connect this side sent, kept so a fresh attempt can
|
||||
/// always be built whatever became of the pending packet.
|
||||
#connectPayload = new Uint8Array();
|
||||
/// When the unanswered Connect first went out; `dueResends` abandons it for a
|
||||
/// fresh attempt once `RUDP_CONNECT_ATTEMPT_MS` has passed.
|
||||
#connectStartedAtMs = 0;
|
||||
|
|
@ -218,7 +218,7 @@ export class CultNetRudpSession {
|
|||
|
||||
constructor(options: CultNetRudpSessionOptions) {
|
||||
this.connectionId = toUint32(options.connectionId, "connectionId");
|
||||
this.#nextSequence = toUint32(options.initialSequence ?? drawSequence(1), "initialSequence");
|
||||
this.#nextSequence = toUint32(options.initialSequence ?? drawSequence(), "initialSequence");
|
||||
if (this.#nextSequence === 0xffff_ffff) {
|
||||
throw new Error("RUDP initialSequence must leave room for a reliable packet.");
|
||||
}
|
||||
|
|
@ -317,6 +317,7 @@ export class CultNetRudpSession {
|
|||
this.#connectSequence = packet.sequence;
|
||||
this.#awaitingAccept = true;
|
||||
this.#connectStartedAtMs = nowMs;
|
||||
this.#connectPayload = new Uint8Array(payload);
|
||||
this.#trackReliable(packet, nowMs);
|
||||
return packet;
|
||||
}
|
||||
|
|
@ -489,7 +490,12 @@ export class CultNetRudpSession {
|
|||
if (packet.packetType === "accept" && !honoursAccept) {
|
||||
return { delivered: [], readyToSend: [] };
|
||||
}
|
||||
this.#applyAcknowledgements(packet);
|
||||
// While this side's Connect awaits its Accept, only the Accept it honours
|
||||
// retires it: an Ack that names the Connect (a server's reply to a repeat or
|
||||
// a stale copy) says the server did not start a session.
|
||||
if (honoursAccept || !this.#awaitingAccept) {
|
||||
this.#applyAcknowledgements(packet);
|
||||
}
|
||||
const readyToSend = this.#promoteQueuedReliable(nowMs);
|
||||
this.#lastReceivedAtMs = nowMs;
|
||||
|
||||
|
|
@ -666,24 +672,24 @@ export class CultNetRudpSession {
|
|||
}
|
||||
|
||||
/// A Connect unanswered for `RUDP_CONNECT_ATTEMPT_MS` is replaced, not
|
||||
/// retransmitted further, by a Connect with a newly drawn sequence and the
|
||||
/// same payload. Nothing was ever sent in an unanswered generation, so no
|
||||
/// sequence issued so far is owed; the draw stays above them, so a frame the
|
||||
/// peer still remembers from an earlier generation of this session stays at or
|
||||
/// before the new Connect.
|
||||
/// retransmitted further, by a Connect with the same payload and the
|
||||
/// abandoned sequence plus the receive window less one. Nothing was ever sent
|
||||
/// in an unanswered generation, so no sequence issued so far is owed. The jump
|
||||
/// keeps a late copy of the abandoned Connect inside the new one's stale
|
||||
/// window, so a server that took the new Connect answers the copy with an Ack
|
||||
/// instead of restarting; and it leaves the stale window of whatever
|
||||
/// generation the server holds, unless that generation sits exactly at the
|
||||
/// jump, and then the next attempt leaves it. The first Connect of a session
|
||||
/// is the only one drawn at random.
|
||||
#abandonUnansweredConnect(nowMs: number): CultNetRudpPacket | undefined {
|
||||
if (!this.#awaitingAccept || nowMs - this.#connectStartedAtMs < RUDP_CONNECT_ATTEMPT_MS) {
|
||||
return undefined;
|
||||
}
|
||||
const pending = this.#connectSequence === undefined
|
||||
? undefined
|
||||
: this.#pendingReliable.get(this.#connectSequence);
|
||||
if (!pending) {
|
||||
if (this.#connectSequence === undefined) {
|
||||
return undefined;
|
||||
}
|
||||
const payload = new Uint8Array(pending.packet.payload ?? []);
|
||||
this.#nextSequence = drawSequence(this.#nextSequence);
|
||||
return this.createConnect(nowMs, payload);
|
||||
this.#nextSequence = Math.min(this.#connectSequence + RUDP_RECEIVED_SEQUENCE_WINDOW - 1, 0xffff_fffe);
|
||||
return this.createConnect(nowMs, this.#connectPayload);
|
||||
}
|
||||
|
||||
#createPacket(packet: {
|
||||
|
|
|
|||
|
|
@ -416,3 +416,120 @@ test("an accepted peer that goes silent times out", () => {
|
|||
server.acceptConnect(session(10).createConnect(0), 0);
|
||||
assert.equal(server.checkTimeout(100_000, 1_000), true);
|
||||
});
|
||||
|
||||
// Ack Cut 1d, batch 2: the connect-attempt timeout.
|
||||
|
||||
test("an Ack naming the pending Connect does not retire it", () => {
|
||||
const server = session(500);
|
||||
const old = session(1);
|
||||
const oldAccept = server.acceptConnect(old.createConnect(0), 0);
|
||||
old.receive(oldAccept, 0);
|
||||
server.receive(old.createAckForReceived(oldAccept.sequence), 0);
|
||||
|
||||
const restarted = session(1);
|
||||
const first = restarted.createConnect(0, enc("join"));
|
||||
const reply = server.acceptConnect(first, 1);
|
||||
assert.equal(reply.packetType, "ack");
|
||||
assert.ok(namesSequence(reply, first.sequence), "the Ack must name the pending Connect for this test to bite");
|
||||
restarted.receive(reply, 1);
|
||||
assert.equal(restarted.connected, false);
|
||||
assert.deepEqual(restarted.pendingReliableSequences, [first.sequence], "an Ack retired the Connect");
|
||||
|
||||
const fresh = restarted.dueResends(3_000);
|
||||
assert.equal(fresh.length, 1, "the client waits for ever with nothing to resend");
|
||||
assert.equal(dec(fresh[0]!.payload ?? new Uint8Array()), "join");
|
||||
const accept = server.acceptConnect(fresh[0]!, 3_000);
|
||||
assert.equal(accept.packetType, "accept");
|
||||
restarted.receive(accept, 3_000);
|
||||
assert.ok(restarted.connected);
|
||||
});
|
||||
|
||||
test("a late copy of an abandoned Connect is stale once its replacement is accepted", () => {
|
||||
const server = session(500);
|
||||
const client = session(10);
|
||||
const abandoned = client.createConnect(0);
|
||||
const fresh = client.dueResends(3_000);
|
||||
assert.equal(fresh.length, 1);
|
||||
assert.equal(fresh[0]!.sequence, abandoned.sequence + 4_095);
|
||||
const accept = server.acceptConnect(fresh[0]!, 3_001);
|
||||
client.receive(accept, 3_002);
|
||||
server.receive(client.createAckForReceived(accept.sequence), 3_002);
|
||||
assert.ok(client.connected);
|
||||
const a = send(client, "a");
|
||||
assert.deepEqual(names(server.receive(a, 3_003)), ["a"]);
|
||||
client.receive(server.createAckForReceived(a.sequence), 3_003);
|
||||
|
||||
assert.equal(server.connectRepeats(abandoned), true, "the late copy would restart the server");
|
||||
const reply = server.acceptConnect(abandoned, 3_004);
|
||||
assert.equal(reply.packetType, "ack");
|
||||
client.receive(reply, 3_005);
|
||||
assert.ok(client.connected);
|
||||
assert.deepEqual(names(server.receive(send(client, "b"), 3_006)), ["b"]);
|
||||
});
|
||||
|
||||
test("a server sitting at the jump is left by the next attempt", () => {
|
||||
const server = session(500);
|
||||
const old = session(5_095);
|
||||
const oldAccept = server.acceptConnect(old.createConnect(0), 0);
|
||||
old.receive(oldAccept, 0);
|
||||
server.receive(old.createAckForReceived(oldAccept.sequence), 0);
|
||||
|
||||
const restarted = session(1_000);
|
||||
const first = restarted.createConnect(0, enc("join"));
|
||||
restarted.receive(server.acceptConnect(first, 1), 1);
|
||||
assert.equal(restarted.connected, false);
|
||||
|
||||
const second = restarted.dueResends(3_000)[0]!;
|
||||
assert.equal(second.sequence, 5_095, "the jump lands on the server's generation");
|
||||
const reply = server.acceptConnect(second, 3_001);
|
||||
assert.equal(reply.packetType, "ack");
|
||||
restarted.receive(reply, 3_001);
|
||||
assert.equal(restarted.connected, false);
|
||||
|
||||
const third = restarted.dueResends(6_000)[0]!;
|
||||
assert.equal(third.sequence, 5_095 + 4_095);
|
||||
const accept = server.acceptConnect(third, 6_001);
|
||||
assert.equal(accept.packetType, "accept");
|
||||
restarted.receive(accept, 6_001);
|
||||
assert.ok(restarted.connected);
|
||||
});
|
||||
|
||||
test("server mode admits a pinned client that restarts on the same address", async () => {
|
||||
const serverSocket = await bind();
|
||||
const peer = await bind();
|
||||
const server = new CultNetRudpSocketTransportConnection({
|
||||
runtimeId: "srv", socket: serverSocket, mode: "server", connectionId, resendPollMs: 20, resendDelayMs: 20,
|
||||
});
|
||||
const port = (serverSocket.address() as { port: number }).port;
|
||||
const received: CultNetRudpPacket[] = [];
|
||||
peer.on("message", (wire) => received.push(decodeRudpPacket(wire)));
|
||||
const take = async (type: string) => {
|
||||
for (let index = 0; index < 50; index += 1) {
|
||||
const found = received.findIndex((packet) => packet.packetType === type);
|
||||
if (found >= 0) return received.splice(found, 1)[0]!;
|
||||
await sleep(10);
|
||||
}
|
||||
throw new Error(`no ${type} arrived`);
|
||||
};
|
||||
try {
|
||||
const first = session(1);
|
||||
peer.send(encodeRudpPacket(first.createConnect(0, enc("join"))), port, "127.0.0.1");
|
||||
const accept = await take("accept");
|
||||
first.receive(accept, 0);
|
||||
peer.send(encodeRudpPacket(first.createAckForReceived(accept.sequence)), port, "127.0.0.1");
|
||||
await sleep(50);
|
||||
received.length = 0;
|
||||
|
||||
const restarted = session(1);
|
||||
peer.send(encodeRudpPacket(restarted.createConnect(0, enc("join"))), port, "127.0.0.1");
|
||||
restarted.receive(await take("ack"), 1);
|
||||
assert.equal(restarted.connected, false);
|
||||
|
||||
peer.send(encodeRudpPacket(restarted.dueResends(3_000)[0]!), port, "127.0.0.1");
|
||||
restarted.receive(await take("accept"), 3_001);
|
||||
assert.ok(restarted.connected, "the restarted pinned client was never admitted");
|
||||
} finally {
|
||||
server.close();
|
||||
peer.close();
|
||||
}
|
||||
});
|
||||
|
|
|
|||
Loading…
Reference in a new issue