mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
a5d209afd0
The pure swap is gone: a reload restarts every live object of a changed plugin (documents and tables kept). Calls running at the commit and every call they cause (the ctx call chain; an old flag on call from trusted links) run on the open old attempt, which seals once its work drained and its successor's recorded requests are ready, bounded by reloadDrainTimeout (5 s); then the successor's factory runs. New work waits at the gate; same-host consumers see the draining object live; a reload cycle fails at once with unavailable/reload_cycle. Example 64, design section 21 rewritten.
756 lines
32 KiB
TypeScript
756 lines
32 KiB
TypeScript
import { describe, expect, it } from "vitest";
|
|
import { BACKGROUND_CONTEXT as context } from "../../src/context/index.ts";
|
|
import {
|
|
binaryCodec,
|
|
checkMessage,
|
|
DeflatedPart,
|
|
describeMessage,
|
|
jsonPart,
|
|
type Message,
|
|
parsePart,
|
|
parsePayload,
|
|
payloadPart,
|
|
textCodec,
|
|
} from "../../src/link/internal.ts";
|
|
import type { Part } from "../../src/link/part.ts";
|
|
import { forEachCodec, testCodec } from "../support.ts";
|
|
|
|
/** A part as JSON text: a received part's bytes decoded (a deflated part stays as it is). */
|
|
function partText(part: Part): Part {
|
|
return part instanceof Uint8Array ? new TextDecoder().decode(part) : part;
|
|
}
|
|
|
|
const incarnation = "0190f5a2-7c3e-7d4a-9b1c-2d3e4f5a6b7c";
|
|
|
|
/** The address a hold names: `Server()` on the wire. */
|
|
const server = { kind: "server", params: {} };
|
|
const error = { kind: "chord", error: { code: "released", message: "gone" } } as const;
|
|
|
|
/** One message of every kind, with the optional fields both present and absent; parts are JSON text. */
|
|
const EVERY_KIND: Message[] = [
|
|
{ kind: "hello", protocol: 1, incarnation, calls: 256 },
|
|
{
|
|
kind: "hold",
|
|
hold: 300,
|
|
type: "app.notes",
|
|
version: 2,
|
|
fingerprint: "0123456789abcdef",
|
|
params: '{"id":1}',
|
|
address: server,
|
|
},
|
|
{
|
|
kind: "hold",
|
|
hold: 1,
|
|
type: "app.notes",
|
|
version: 1,
|
|
fingerprint: "0123456789abcdef",
|
|
params: '{"session":"S1"}',
|
|
address: { kind: "session", params: { id: "S1" } },
|
|
principal: { kind: "guest" },
|
|
incarnation,
|
|
},
|
|
{
|
|
kind: "hold",
|
|
hold: 2,
|
|
type: "app.notes",
|
|
version: 1,
|
|
fingerprint: "0123456789abcdef",
|
|
params: "{}",
|
|
address: server,
|
|
incarnation,
|
|
},
|
|
{ kind: "release", hold: 0 },
|
|
{ kind: "call", call: 3, hold: 0, method: 1, args: '{"text":"x"}' },
|
|
{ kind: "call", call: 4, hold: 0, method: 0 },
|
|
{ kind: "abort", call: 3 },
|
|
{
|
|
kind: "bound",
|
|
holds: [{ hold: 0, variant: 1, members: [{ member: 0, value: '[1,"a"]' }], failed: [{ member: 1, error }] }],
|
|
},
|
|
{
|
|
kind: "bound",
|
|
holds: [
|
|
{ hold: 0, variant: 0, members: [], failed: [] },
|
|
{ hold: 5, variant: 2, members: [{ member: 1, value: "null" }], failed: [] },
|
|
],
|
|
},
|
|
{
|
|
kind: "state",
|
|
holds: [
|
|
{ hold: 0, members: [{ member: 2, ops: '[["s",["n"],1]]' }], failed: [], retired: [] },
|
|
{ hold: 1, members: [], failed: [{ member: 0, error }], retired: [] },
|
|
// Retired members (decision "One retirement mechanism"), alone and with a failure.
|
|
{ hold: 2, members: [{ member: 1, ops: '[["r",{}]]' }], failed: [], retired: [1, 3] },
|
|
{ hold: 3, members: [], failed: [{ member: 0, error }], retired: [2] },
|
|
],
|
|
},
|
|
{
|
|
kind: "failed",
|
|
hold: 0,
|
|
cause: { kind: "transient", error: { kind: "chord", error: { code: "unavailable", message: "m" } } },
|
|
},
|
|
{
|
|
kind: "failed",
|
|
hold: 0,
|
|
cause: { kind: "broken", error: { kind: "chord", error: { code: "unavailable", message: "m" } } },
|
|
},
|
|
{
|
|
kind: "failed",
|
|
hold: 0,
|
|
cause: { kind: "transient", error: { kind: "chord", error: { code: "unavailable", message: "m" } } },
|
|
retryIn: 0,
|
|
},
|
|
// Not the object's fault (dependency model §2): no backoff window, an event wakes it.
|
|
{
|
|
kind: "failed",
|
|
hold: 0,
|
|
cause: { kind: "waiting", error: { kind: "chord", error: { code: "contract_mismatch", message: "m" } } },
|
|
},
|
|
// The wire error format in `failed` and `gone`: a plugin's error travels too.
|
|
{ kind: "failed", hold: 0, cause: { kind: "broken", error: { kind: "plugin", code: "config", message: "m" } } },
|
|
{ kind: "gone", hold: 1, error: { kind: "plugin", code: "config", message: "m", details: { at: 1 } } },
|
|
{
|
|
kind: "failed",
|
|
hold: 9,
|
|
cause: { kind: "transient", error: { kind: "chord", error: { code: "unavailable", message: "m" } } },
|
|
retryIn: 60_000,
|
|
},
|
|
// With its hop budget (audit A5): the first host's, and the last one passed on.
|
|
{ kind: "retry", hold: 300, hops: 16 },
|
|
{ kind: "retry", hold: 300, hops: 1 },
|
|
{ kind: "loading", hold: 300 },
|
|
{ kind: "result", call: 3, value: '{"n":1}' },
|
|
{ kind: "result", call: 3 },
|
|
{ kind: "error", call: 3, error: { kind: "plugin", code: "c", message: "m", details: [1] } },
|
|
{ kind: "gone", hold: 0, error: { kind: "chord", error: { code: "released", message: "m" } } },
|
|
{ kind: "ack" },
|
|
{ kind: "ack", origin: 1 },
|
|
{ kind: "closed", origin: 300 },
|
|
{
|
|
kind: "hold",
|
|
origin: 2,
|
|
hold: 0,
|
|
type: "t",
|
|
version: 1,
|
|
fingerprint: "0123456789abcdef",
|
|
params: "{}",
|
|
address: server,
|
|
},
|
|
{ kind: "release", origin: 2, hold: 3 },
|
|
{ kind: "call", origin: 2, call: 1, hold: 3, method: 0, args: "[1]" },
|
|
{ kind: "abort", origin: 2, call: 1 },
|
|
{ kind: "retry", origin: 2, hold: 3, hops: 16 },
|
|
{ kind: "bound", origin: 2, holds: [{ hold: 3, variant: 0, members: [], failed: [] }] },
|
|
{ kind: "state", origin: 2, holds: [] },
|
|
{
|
|
kind: "failed",
|
|
origin: 2,
|
|
hold: 3,
|
|
cause: { kind: "broken", error: { kind: "chord", error: { code: "unavailable", message: "m" } } },
|
|
},
|
|
{ kind: "loading", origin: 2, hold: 3 },
|
|
{ kind: "result", origin: 2, call: 1 },
|
|
{ kind: "error", origin: 2, call: 1, error: { kind: "plugin", code: "c", message: "m" } },
|
|
{ kind: "gone", origin: 2, hold: 3, error: { kind: "chord", error: { code: "released", message: "m" } } },
|
|
{
|
|
kind: "hold",
|
|
hold: 3,
|
|
type: "app.notes",
|
|
version: 1,
|
|
fingerprint: "0123456789abcdef",
|
|
params: "{}",
|
|
address: server,
|
|
chain: [incarnation, incarnation],
|
|
limit: 65_536,
|
|
},
|
|
{
|
|
kind: "hold",
|
|
hold: 4,
|
|
type: "t",
|
|
version: 1,
|
|
fingerprint: "0123456789abcdef",
|
|
params: "{}",
|
|
address: server,
|
|
limit: 0,
|
|
},
|
|
{ kind: "revoke", revocation: { grant: "g1" } },
|
|
{ kind: "ping" },
|
|
{ kind: "pong" },
|
|
// Trace carriers (R5, slice 9) on a hold and a call, with and without an origin, arguments and other fields.
|
|
{
|
|
kind: "hold",
|
|
hold: 5,
|
|
type: "app.notes",
|
|
version: 1,
|
|
fingerprint: "0123456789abcdef",
|
|
params: "{}",
|
|
address: server,
|
|
trace: { "pi-memory-span": "1", "pi-memory-trace": "s:1" },
|
|
},
|
|
{
|
|
kind: "hold",
|
|
origin: 3,
|
|
hold: 6,
|
|
type: "app.notes",
|
|
version: 1,
|
|
fingerprint: "0123456789abcdef",
|
|
params: "{}",
|
|
address: server,
|
|
limit: 10,
|
|
trace: { traceparent: "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01" },
|
|
},
|
|
{ kind: "call", call: 7, hold: 0, method: 1, args: '{"text":"x"}', trace: { a: "b" } },
|
|
{ kind: "call", origin: 2, call: 8, hold: 3, method: 0, trace: {} },
|
|
{ kind: "call", call: 9, hold: 0, method: 1, args: '{"text":"x"}', old: true },
|
|
{ kind: "call", origin: 2, call: 10, hold: 3, method: 0, trace: { a: "b" }, old: true },
|
|
];
|
|
|
|
/** A carrier at the bounds: 8 keys, keys and values of 256 characters. */
|
|
const LARGEST_CARRIER = Object.fromEntries(
|
|
Array.from({ length: 8 }, (_, index) => [`${index}`.padEnd(256, "k"), "v".repeat(256)]),
|
|
);
|
|
|
|
/** A decoded message with every part as its text, to compare with one built from text parts. */
|
|
function plain(message: Message): Message {
|
|
switch (message.kind) {
|
|
case "hold":
|
|
return { ...message, params: partText(message.params) };
|
|
case "call":
|
|
return message.args === undefined ? message : { ...message, args: partText(message.args) };
|
|
case "result":
|
|
return message.value === undefined ? message : { ...message, value: partText(message.value) };
|
|
case "bound":
|
|
return {
|
|
...message,
|
|
holds: message.holds.map((entry) => ({
|
|
...entry,
|
|
members: entry.members.map(({ member, value }) => ({ member, value: partText(value) })),
|
|
})),
|
|
};
|
|
case "state":
|
|
return {
|
|
...message,
|
|
holds: message.holds.map((entry) => ({
|
|
...entry,
|
|
members: entry.members.map(({ member, ops }) => ({ member, ops: partText(ops) })),
|
|
})),
|
|
};
|
|
default:
|
|
return message;
|
|
}
|
|
}
|
|
|
|
forEachCodec(() => {
|
|
describe("messages", () => {
|
|
it.each(EVERY_KIND)("round-trip %j and pass the message schema", (message) => {
|
|
const codec = testCodec();
|
|
const decoded = checkMessage(codec.decode(codec.encode(message)), context);
|
|
expect(plain(decoded)).toEqual(message);
|
|
});
|
|
|
|
it("round-trip deflated parts in every part's place, still deflated (a relay forwards them so)", () => {
|
|
const codec = testCodec();
|
|
const deflated = () => new DeflatedPart(new Uint8Array([0, 1, 2, 255]));
|
|
const messages: Message[] = [
|
|
{
|
|
kind: "hold",
|
|
hold: 0,
|
|
type: "t",
|
|
version: 1,
|
|
fingerprint: "0123456789abcdef",
|
|
params: deflated(),
|
|
address: server,
|
|
},
|
|
{ kind: "call", call: 0, hold: 0, method: 0, args: deflated() },
|
|
{ kind: "result", call: 0, value: deflated(), segments: [new Uint8Array([7])] },
|
|
{
|
|
kind: "bound",
|
|
holds: [{ hold: 0, variant: 0, members: [{ member: 0, value: deflated() }], failed: [] }],
|
|
},
|
|
{ kind: "state", holds: [{ hold: 0, members: [{ member: 0, ops: deflated() }], failed: [], retired: [] }] },
|
|
];
|
|
for (const message of messages) {
|
|
const decoded = checkMessage(codec.decode(codec.encode(message)), context);
|
|
expect(decoded).toEqual(message);
|
|
}
|
|
});
|
|
|
|
it("round-trip parts received as bytes (a relay forwards them as they came)", () => {
|
|
const codec = testCodec();
|
|
const message: Message = { kind: "result", call: 0, value: new TextEncoder().encode('{"n":1}') };
|
|
expect(plain(codec.decode(codec.encode(message)))).toEqual({ kind: "result", call: 0, value: '{"n":1}' });
|
|
});
|
|
|
|
it("carry byte segments beside a payload, as views of the message", () => {
|
|
const codec = testCodec();
|
|
const data = new Uint8Array(1024).fill(7);
|
|
const thumbs = [new Uint8Array([1, 2]), new Uint8Array([3])];
|
|
const { json, segments } = payloadPart({ name: "a", data, thumbs }, true);
|
|
const message: Message = { kind: "call", call: 0, hold: 0, method: 0, args: json, segments };
|
|
const bytes = codec.encode(message);
|
|
const decoded = codec.decode(bytes) as Extract<Message, { kind: "call" }>;
|
|
expect(decoded.segments).toHaveLength(3);
|
|
expect(decoded.segments![0]!.buffer).toBe(bytes.buffer);
|
|
const args = parsePayload(decoded.args as Uint8Array, decoded.segments) as {
|
|
data: Uint8Array;
|
|
thumbs: Uint8Array[];
|
|
};
|
|
expect(args.data).toEqual(data);
|
|
expect(args.data.buffer).toBe(bytes.buffer);
|
|
expect(args.thumbs).toEqual(thumbs);
|
|
// The segments travel raw: the message is barely larger than the bytes it carries.
|
|
expect(bytes.byteLength).toBeLessThan(1024 + 3 + 100);
|
|
});
|
|
|
|
it("refuse an empty message, an unknown kind and trailing bytes", () => {
|
|
const codec = testCodec();
|
|
const release = codec.encode({ kind: "release", hold: 1 });
|
|
const trailing = new Uint8Array([...release, ...(codec === binaryCodec ? [0] : [0x20, 0x31])]);
|
|
expect(() => codec.decode(new Uint8Array())).toThrow();
|
|
expect(() =>
|
|
codec.decode(codec === binaryCodec ? new Uint8Array([31]) : new TextEncoder().encode('["?"]')),
|
|
).toThrow("unknown kind");
|
|
expect(() => codec.decode(trailing)).toThrow();
|
|
});
|
|
|
|
it("refuse an origin on a kind without one, an origin 0, and a closed without one (the message schema)", () => {
|
|
const codec = testCodec();
|
|
const check = (message: unknown) => () =>
|
|
checkMessage(codec.decode(codec.encode(message as Message)), context);
|
|
expect(check({ kind: "ping", origin: 1 })).toThrow("message schema");
|
|
expect(check({ kind: "ack", origin: 0 })).toThrow("message schema");
|
|
const closed = codec === binaryCodec ? new Uint8Array([14]) : new TextEncoder().encode('["C"]');
|
|
expect(() => checkMessage(codec.decode(closed), context)).toThrow("message schema");
|
|
});
|
|
|
|
it("carry a trace carrier up to its bounds; past them, of another shape or on another kind, the schema refuses it", () => {
|
|
const codec = testCodec();
|
|
const check = (message: unknown) => () =>
|
|
checkMessage(codec.decode(codec.encode(message as Message)), context);
|
|
const call = { kind: "call", call: 0, hold: 0, method: 0 } as const;
|
|
expect(check({ ...call, trace: LARGEST_CARRIER })()).toEqual({ ...call, trace: LARGEST_CARRIER });
|
|
expect(check({ ...call, trace: { ...LARGEST_CARRIER, nine: "x" } })).toThrow("message schema");
|
|
expect(check({ ...call, trace: { ["k".repeat(257)]: "v" } })).toThrow("message schema");
|
|
expect(check({ ...call, trace: { k: "v".repeat(257) } })).toThrow("message schema");
|
|
expect(check({ ...call, trace: { k: 1 } })).toThrow("message schema");
|
|
expect(check({ ...call, trace: ["k"] })).toThrow("message schema");
|
|
// Printable ASCII only (review M2): no control character, no space in a key; a value may hold one (W3C's
|
|
// `tracestate`).
|
|
expect(check({ ...call, trace: { k: "a\u001b[2J" } })).toThrow("message schema");
|
|
expect(check({ ...call, trace: { "k\u0007": "v" } })).toThrow("message schema");
|
|
expect(check({ ...call, trace: { "a b": "v" } })).toThrow("message schema");
|
|
expect(check({ ...call, trace: { k: "é" } })).toThrow("message schema");
|
|
expect(check({ ...call, trace: { k: "a=1, b=2" } })()).toEqual({ ...call, trace: { k: "a=1, b=2" } });
|
|
// The codecs write it only on a hold or a call; a decoder's forgery fails the schema.
|
|
expect(() => checkMessage({ kind: "abort", call: 0, trace: { k: "v" } }, context)).toThrow("message schema");
|
|
});
|
|
|
|
it("carry a trace carrier beside a call's byte segments", () => {
|
|
const codec = testCodec();
|
|
const { json, segments } = payloadPart({ data: new Uint8Array([1, 2, 3]) }, true);
|
|
const message: Message = {
|
|
kind: "call",
|
|
call: 0,
|
|
hold: 0,
|
|
method: 0,
|
|
args: json,
|
|
segments,
|
|
trace: { a: "b" },
|
|
};
|
|
const decoded = codec.decode(codec.encode(message)) as Extract<Message, { kind: "call" }>;
|
|
expect(decoded.trace).toEqual({ a: "b" });
|
|
expect(decoded.segments).toEqual(segments);
|
|
});
|
|
|
|
it("refuse a type or host kind past 256 characters (review M1: a relay records them unparsed)", () => {
|
|
const codec = testCodec();
|
|
const check = (message: unknown) => () =>
|
|
checkMessage(codec.decode(codec.encode(message as Message)), context);
|
|
const hold = {
|
|
kind: "hold",
|
|
hold: 0,
|
|
type: "t".repeat(256),
|
|
version: 1,
|
|
fingerprint: "0123456789abcdef",
|
|
params: "{}",
|
|
address: { kind: "k".repeat(256), params: {} },
|
|
};
|
|
expect(check(hold)).not.toThrow();
|
|
expect(check({ ...hold, type: "t".repeat(257) })).toThrow("message schema");
|
|
expect(check({ ...hold, address: { kind: "k".repeat(257), params: {} } })).toThrow("message schema");
|
|
});
|
|
|
|
it("keep a leading U+FEFF in strings and parts (it is data, not a byte order mark)", () => {
|
|
const codec = testCodec();
|
|
const message: Message = {
|
|
kind: "hold",
|
|
hold: 0,
|
|
type: "\uFEFFthing",
|
|
version: 1,
|
|
fingerprint: "0123456789abcdef",
|
|
params: jsonPart({ "\uFEFFkey": "\uFEFFvalue" }),
|
|
address: server,
|
|
};
|
|
const decoded = codec.decode(codec.encode(message)) as Extract<Message, { kind: "hold" }>;
|
|
expect(plain(decoded)).toEqual(message);
|
|
expect(parsePart(decoded.params as string | Uint8Array)).toEqual({ "\uFEFFkey": "\uFEFFvalue" });
|
|
const result: Message = { kind: "result", call: 0, value: jsonPart("\uFEFF") };
|
|
expect(plain(codec.decode(codec.encode(result)))).toEqual(result);
|
|
});
|
|
|
|
it("leave the envelope to the message schema: a misshapen error passes the codec and fails the schema", () => {
|
|
const codec = testCodec();
|
|
const decoded = codec.decode(codec.encode({ kind: "error", call: 0, error: { kind: "nope" } } as never));
|
|
expect(() => checkMessage(decoded, context)).toThrow("message: fails the message schema");
|
|
expect(() => checkMessage({ kind: "release", hold: -1 }, context)).toThrow(
|
|
"message: fails the message schema",
|
|
);
|
|
});
|
|
});
|
|
});
|
|
|
|
describe("the binary codec", () => {
|
|
it("writes ids as varints, the incarnation as 16 and the fingerprint as 8 raw bytes, no outer length", () => {
|
|
expect(binaryCodec.encode({ kind: "hello", protocol: 1, incarnation, calls: 1 })).toHaveLength(1 + 1 + 16 + 1);
|
|
expect([...binaryCodec.encode({ kind: "release", hold: 300 })]).toEqual([2, 0xac, 0x02]);
|
|
const hold = binaryCodec.encode({
|
|
kind: "hold",
|
|
hold: 0,
|
|
type: "t",
|
|
version: 1,
|
|
fingerprint: "0123456789abcdef",
|
|
params: "{}",
|
|
address: server,
|
|
});
|
|
// kind, id, type, version, fingerprint, params, the address's kind and params, the optional fields' flags
|
|
expect(hold).toHaveLength(1 + 1 + 2 + 1 + 8 + 3 + 7 + 3 + 1);
|
|
expect([...binaryCodec.encode({ kind: "ack" })]).toEqual([13]);
|
|
// An origin: the kind byte's high bit, then the origin as a varint.
|
|
expect([...binaryCodec.encode({ kind: "ack", origin: 300 })]).toEqual([13 | 0x80, 0xac, 0x02]);
|
|
expect([...binaryCodec.encode({ kind: "closed", origin: 1 })]).toEqual([14 | 0x80, 1]);
|
|
});
|
|
|
|
it("writes a failure's cause as one byte, 0 transient, 1 broken, 2 waiting (dependency model §2); refuses another", () => {
|
|
const error = { kind: "chord", error: { code: "unavailable", message: "m" } } as const;
|
|
const failed = (kind: "transient" | "broken" | "waiting"): Message => ({
|
|
kind: "failed",
|
|
hold: 0,
|
|
cause: { kind, error },
|
|
});
|
|
expect(
|
|
["transient", "broken", "waiting"].map((kind) => binaryCodec.encode(failed(kind as "waiting"))[2]),
|
|
).toEqual([0, 1, 2]);
|
|
const unknown = binaryCodec.encode(failed("waiting"));
|
|
unknown[2] = 3;
|
|
expect(() => binaryCodec.decode(unknown)).toThrow("an unknown failure kind");
|
|
});
|
|
|
|
it("costs a state entry without retirements no byte: the failure count is doubled, plus one when they follow", () => {
|
|
const state = (retired: number[]): Message => ({
|
|
kind: "state",
|
|
holds: [{ hold: 0, members: [], failed: [], retired }],
|
|
});
|
|
// kind, entry count, hold, member count, failures (0 doubled)
|
|
expect([...binaryCodec.encode(state([]))]).toEqual([6, 1, 0, 0, 0]);
|
|
// the low bit set, then the retired count and member
|
|
expect([...binaryCodec.encode(state([3]))]).toEqual([6, 1, 0, 0, 1, 1, 3]);
|
|
const first = binaryCodec.decode(binaryCodec.encode(state([])));
|
|
const second = binaryCodec.decode(binaryCodec.encode(state([])));
|
|
const retired = (message: Message) => (message as Extract<Message, { kind: "state" }>).holds[0]!.retired;
|
|
// No allocation per message: decoded entries without retirements share one empty list.
|
|
expect(retired(first)).toBe(retired(second));
|
|
});
|
|
|
|
it("prefixes a part with its length doubled, plus one when it is deflated", () => {
|
|
const plain = binaryCodec.encode({ kind: "result", call: 0, value: "12" });
|
|
const deflated = binaryCodec.encode({ kind: "result", call: 0, value: new DeflatedPart(new Uint8Array([7, 8])) });
|
|
// kind, call, payload flags, prefix, bytes
|
|
expect([...plain]).toEqual([8, 0, 1, 4, 0x31, 0x32]);
|
|
expect([...deflated]).toEqual([8, 0, 1, 5, 7, 8]);
|
|
});
|
|
|
|
it("decodes parts as views of the message, never parsing them: a malformed part decodes", () => {
|
|
const part = new TextEncoder().encode("{not json");
|
|
const bytes = binaryCodec.encode({ kind: "result", call: 0, value: part });
|
|
const decoded = binaryCodec.decode(bytes) as Extract<Message, { kind: "result" }>;
|
|
const value = decoded.value as Uint8Array;
|
|
expect(value.buffer).toBe(bytes.buffer);
|
|
expect(() => parsePart(value)).toThrow();
|
|
});
|
|
|
|
it("refuses varints past a safe integer and lengths past the message", () => {
|
|
expect(() => binaryCodec.decode(new Uint8Array([2, ...new Array(8).fill(0xff), 0x7f]))).toThrow(
|
|
"a varint does not fit",
|
|
);
|
|
expect(() => binaryCodec.decode(new Uint8Array([1, 0, 9, 0x61]))).toThrow("the message ends early");
|
|
});
|
|
|
|
it("refuses a varint of more than 8 bytes, which would read as no number", () => {
|
|
const state = new Uint8Array([6, ...new Array(150).fill(0x80), 0]);
|
|
expect(() => binaryCodec.decode(state)).toThrow("a varint does not fit");
|
|
});
|
|
|
|
it("flags a trace carrier in the kind byte (0x40) and writes it as JSON after the origin", () => {
|
|
const call = binaryCodec.encode({ kind: "call", origin: 2, call: 0, hold: 0, method: 0, trace: { a: "b" } });
|
|
const json = new TextEncoder().encode('{"a":"b"}');
|
|
expect([...call.subarray(0, 3 + json.byteLength)]).toEqual([3 | 0x80 | 0x40, 2, json.byteLength, ...json]);
|
|
const plainCall = binaryCodec.encode({ kind: "call", call: 0, hold: 0, method: 0 });
|
|
expect(plainCall[0]).toBe(3);
|
|
});
|
|
|
|
it("flags a call that is old work of a reload in the kind byte (0x20); another kind with that flag is unknown", () => {
|
|
const call = binaryCodec.encode({ kind: "call", call: 0, hold: 0, method: 0, old: true });
|
|
expect(call[0]).toBe(3 | 0x20);
|
|
expect(binaryCodec.decode(call)).toEqual({ kind: "call", call: 0, hold: 0, method: 0, old: true });
|
|
const release = binaryCodec.encode({ kind: "release", hold: 1 });
|
|
release[0]! |= 0x20;
|
|
expect(() => binaryCodec.decode(release)).toThrow("unknown kind");
|
|
});
|
|
|
|
it("refuses unknown hold flags", () => {
|
|
const hold = binaryCodec.encode({
|
|
kind: "hold",
|
|
hold: 0,
|
|
type: "t",
|
|
version: 1,
|
|
fingerprint: "0123456789abcdef",
|
|
params: "{}",
|
|
address: server,
|
|
});
|
|
const forged = hold.slice();
|
|
forged[forged.length - 1] = 32;
|
|
expect(() => binaryCodec.decode(forged)).toThrow("unknown hold flags");
|
|
});
|
|
|
|
it("returns a right-sized message: a large one keeps no larger backing store", () => {
|
|
const value = jsonPart("x".repeat(500_000));
|
|
// Small fields after the large value: they must not double the buffer the message keeps.
|
|
const members = [
|
|
{ member: 0, value },
|
|
{ member: 1, value: "1" },
|
|
];
|
|
const bytes = binaryCodec.encode({ kind: "bound", holds: [{ hold: 0, variant: 0, members, failed: [] }] });
|
|
expect(bytes.buffer.byteLength - bytes.byteLength).toBeLessThanOrEqual(4096);
|
|
const small = binaryCodec.encode({ kind: "release", hold: 1 });
|
|
expect(small.buffer.byteLength).toBe(small.byteLength);
|
|
// A message of about two thirds of the starting buffer keeps no spare room either.
|
|
const medium = binaryCodec.encode({ kind: "result", call: 0, value: jsonPart("x".repeat(160)) });
|
|
expect(medium.buffer.byteLength).toBe(medium.byteLength);
|
|
});
|
|
|
|
it("refuses invalid UTF-8 in strings", () => {
|
|
expect(() => binaryCodec.decode(new Uint8Array([1, 0, 1, 0xff]))).toThrow();
|
|
});
|
|
|
|
it("writes retry as its kind, hold and hops, loading as its kind and hold, and failed's retryIn as one more than it (0: absent)", () => {
|
|
expect([...binaryCodec.encode({ kind: "retry", hold: 300, hops: 16 })]).toEqual([11, 0xac, 0x02, 16]);
|
|
expect([...binaryCodec.encode({ kind: "loading", hold: 1 })]).toEqual([12, 1]);
|
|
const cause = { kind: "broken", error: { kind: "chord", error: { code: "released", message: "" } } } as const;
|
|
const absent = binaryCodec.encode({ kind: "failed", hold: 0, cause });
|
|
const present = binaryCodec.encode({ kind: "failed", hold: 0, cause, retryIn: 0 });
|
|
expect(absent.at(-1)).toBe(0);
|
|
expect(present.at(-1)).toBe(1);
|
|
});
|
|
});
|
|
|
|
describe("the text codec", () => {
|
|
const text = (message: Message) => new TextDecoder().decode(textCodec.encode(message));
|
|
|
|
it("writes one JSON tuple per message with short tags and positional fields, parts spliced in", () => {
|
|
expect(text({ kind: "release", hold: 3 })).toBe('["R",3]');
|
|
expect(text({ kind: "call", call: 1, hold: 0, method: 2 })).toBe('["c",1,0,2]');
|
|
const members = [{ member: 0, value: "5" }];
|
|
expect(text({ kind: "bound", holds: [{ hold: 0, variant: 1, members, failed: [] }] })).toBe(
|
|
'["b",[[0,1,[[0,[0,5]]]]]]',
|
|
);
|
|
expect(text({ kind: "retry", hold: 2, hops: 16 })).toBe('["T",2,16]');
|
|
expect(text({ kind: "loading", hold: 2 })).toBe('["l",2]');
|
|
const cause = {
|
|
kind: "transient",
|
|
error: { kind: "chord", error: { code: "unavailable", message: "m" } },
|
|
} as const;
|
|
expect(text({ kind: "failed", hold: 2, cause })).toBe(
|
|
'["f",2,"t",{"kind":"chord","error":{"code":"unavailable","message":"m"}}]',
|
|
);
|
|
expect(text({ kind: "failed", hold: 2, cause, retryIn: 750 })).toBe(
|
|
'["f",2,"t",{"kind":"chord","error":{"code":"unavailable","message":"m"}},750]',
|
|
);
|
|
expect(text({ kind: "failed", hold: 2, cause: { ...cause, kind: "waiting" } })).toBe(
|
|
'["f",2,"w",{"kind":"chord","error":{"code":"unavailable","message":"m"}}]',
|
|
);
|
|
expect(() => textCodec.decode(new TextEncoder().encode('["f",2,"x",{}]'))).toThrow("an unknown failure kind");
|
|
expect(text({ kind: "ack" })).toBe('["a"]');
|
|
// An origin wraps the message's own tuple.
|
|
expect(text({ kind: "release", origin: 4, hold: 3 })).toBe('["@",4,["R",3]]');
|
|
expect(text({ kind: "closed", origin: 4 })).toBe('["@",4,["C"]]');
|
|
});
|
|
|
|
it("tags every part with its codec, [0, json] or [1, base64] for a deflated one, so no value is reserved", () => {
|
|
const message: Message = { kind: "result", call: 1, value: new DeflatedPart(new Uint8Array([1, 2, 3])) };
|
|
expect(text(message)).toBe('["r",1,[1,"AQID"]]');
|
|
expect(textCodec.decode(textCodec.encode(message))).toEqual(message);
|
|
// A value of any shape is data, the one-key `$deflated` object the first builder reserved included.
|
|
const shaped: Message = { kind: "result", call: 1, value: '{"$deflated":"AQID"}' };
|
|
expect(text(shaped)).toBe('["r",1,[0,{"$deflated":"AQID"}]]');
|
|
expect(textCodec.decode(textCodec.encode(shaped))).toEqual(shaped);
|
|
expect(text({ kind: "revoke", revocation: { grant: "g" } })).toBe('["V",{"grant":"g"}]');
|
|
});
|
|
|
|
it("refuses a part that is no [codec, payload] pair, or of an unknown codec", () => {
|
|
const decode = (tuple: string) => textCodec.decode(new TextEncoder().encode(tuple));
|
|
expect(() => decode('["r",1,{"n":1}]')).toThrow("no [codec, payload] pair");
|
|
expect(() => decode('["r",1,[2,"AQID"]]')).toThrow("unknown codec");
|
|
expect(() => decode('["r",1,[1,5]]')).toThrow("unknown codec");
|
|
});
|
|
|
|
it("carries a binary link's deflated part over a text hop and back unchanged (a relay between codecs)", () => {
|
|
const part = new DeflatedPart(new Uint8Array([9, 8, 7]));
|
|
const message: Message = {
|
|
kind: "state",
|
|
holds: [{ hold: 0, members: [{ member: 0, ops: part }], failed: [], retired: [] }],
|
|
};
|
|
const overText = textCodec.decode(textCodec.encode(binaryCodec.decode(binaryCodec.encode(message))));
|
|
expect(binaryCodec.decode(binaryCodec.encode(overText))).toEqual(message);
|
|
});
|
|
|
|
it("leaves out a hold's absent optional fields at the end, null in the middle", () => {
|
|
const hold = {
|
|
kind: "hold",
|
|
hold: 0,
|
|
type: "t",
|
|
version: 1,
|
|
fingerprint: "0123456789abcdef",
|
|
params: "{}",
|
|
address: server,
|
|
} as const;
|
|
const head = '["H",0,"t",1,"0123456789abcdef",[0,{}],["server",{}]';
|
|
expect(text(hold)).toBe(`${head}]`);
|
|
expect(text({ ...hold, incarnation })).toBe(`${head},null,"${incarnation}"]`);
|
|
expect(text({ ...hold, limit: 5 })).toBe(`${head},null,null,null,5]`);
|
|
expect(text({ ...hold, trace: { a: "b" } })).toBe(`${head},null,null,null,null,{"a":"b"}]`);
|
|
});
|
|
|
|
it("writes a call's trace carrier last, its payload's fields positional then (null when absent)", () => {
|
|
const call = { kind: "call", call: 0, hold: 1, method: 2 } as const;
|
|
expect(text({ ...call, trace: { a: "b" } })).toBe('["c",0,1,2,null,null,{"a":"b"}]');
|
|
expect(text({ ...call, args: "[1]", trace: { a: "b" } })).toBe('["c",0,1,2,[0,[1]],null,{"a":"b"}]');
|
|
expect(text({ ...call, args: "[1]" })).toBe('["c",0,1,2,[0,[1]]]');
|
|
});
|
|
|
|
it("writes a call's old-work flag after its carrier (null when absent), its payload's fields positional then", () => {
|
|
const call = { kind: "call", call: 0, hold: 1, method: 2 } as const;
|
|
expect(text({ ...call, old: true })).toBe('["c",0,1,2,null,null,null,true]');
|
|
expect(text({ ...call, args: "[1]", trace: { a: "b" }, old: true })).toBe(
|
|
'["c",0,1,2,[0,[1]],null,{"a":"b"},true]',
|
|
);
|
|
});
|
|
|
|
it("leaves the old-work flag's value to the message schema: only true passes", () => {
|
|
const decoded = textCodec.decode(new TextEncoder().encode('["c",0,1,2,null,null,null,1]'));
|
|
expect(() => checkMessage(decoded, context)).toThrow("message schema");
|
|
});
|
|
|
|
it("puts byte segments after the tuple and a newline, their lengths the tuple's last field", () => {
|
|
const bytes = textCodec.encode({
|
|
kind: "result",
|
|
call: 0,
|
|
value: '{"$bytes":0}',
|
|
segments: [new Uint8Array([1, 2])],
|
|
});
|
|
expect([...bytes]).toEqual([...new TextEncoder().encode('["r",0,[0,{"$bytes":0}],[2]]\n'), 1, 2]);
|
|
});
|
|
|
|
it("refuses a tuple of the wrong arity and segments without a payload carrying them", () => {
|
|
expect(() => textCodec.decode(new TextEncoder().encode('["R",1,2]'))).toThrow('a "R" message with 2 fields');
|
|
expect(() => textCodec.decode(new TextEncoder().encode('["s",[[0,[],[],[],9]]]'))).toThrow(
|
|
"a state entry with 5 fields",
|
|
);
|
|
expect(() => textCodec.decode(new TextEncoder().encode('["s",[[0]]]'))).toThrow("a state entry with 1 fields");
|
|
expect(() => textCodec.decode(new Uint8Array([...new TextEncoder().encode('["R",1]\n'), 1]))).toThrow(
|
|
"byte segments without a payload",
|
|
);
|
|
expect(() => textCodec.decode(new TextEncoder().encode('["@",1,["@",2,["a"]]]'))).toThrow("wraps one message");
|
|
expect(() => textCodec.decode(new TextEncoder().encode('["@",1]'))).toThrow("wraps one message");
|
|
});
|
|
|
|
it("refuses a hold without its address: every hold names one", () => {
|
|
const decode = (tuple: string) => textCodec.decode(new TextEncoder().encode(tuple));
|
|
expect(() => decode('["H",0,"t",1,"0123456789abcdef",[0,{}]]')).toThrow('a "H" message with 5 fields');
|
|
expect(() => decode('["H",0,"t",1,"0123456789abcdef",[0,{}],null]')).toThrow("expected a list");
|
|
expect(() => decode('["H",0,"t",1,"0123456789abcdef",[0,{}],["server"]]')).toThrow("an address is a pair");
|
|
const hold = { kind: "hold", hold: 0, type: "t", version: 1, fingerprint: "0123456789abcdef", params: "{}" };
|
|
expect(() => checkMessage(hold, context)).toThrow("message schema");
|
|
});
|
|
|
|
it("refuses a retry without its hop budget, or with one outside 1..16 (audit A5)", () => {
|
|
const decode = (tuple: string) => textCodec.decode(new TextEncoder().encode(tuple));
|
|
expect(() => decode('["T",2]')).toThrow('a "T" message with 1 fields');
|
|
for (const hops of [0, 17]) {
|
|
expect(() => checkMessage({ kind: "retry", hold: 2, hops }, context)).toThrow("message schema");
|
|
}
|
|
});
|
|
});
|
|
|
|
describe("describeMessage", () => {
|
|
it("prints a message in the text tuple form", () => {
|
|
expect(describeMessage({ kind: "abort", call: 7 })).toBe('["x",7]');
|
|
expect(describeMessage({ kind: "result", call: 0, value: '{"$bytes":0}', segments: [new Uint8Array(3)] })).toBe(
|
|
'["r",0,[0,{"$bytes":0}],[3]]',
|
|
);
|
|
});
|
|
});
|
|
|
|
describe("a codec mismatch", () => {
|
|
it("decodes the other codec's hello as no message of its own", () => {
|
|
const hello: Message = { kind: "hello", protocol: 1, incarnation, calls: 256 };
|
|
expect(() => binaryCodec.decode(textCodec.encode(hello))).toThrow();
|
|
expect(() => textCodec.decode(binaryCodec.encode(hello))).toThrow();
|
|
});
|
|
});
|
|
|
|
describe("the binary codec's strings", () => {
|
|
it("survive the buffer growing around them, at every offset near the initial size", () => {
|
|
for (let length = 0; length < 400; length++) {
|
|
// A long value of three-byte characters (filling what it reserved), then a short one that lands at every offset around the buffer's first growth.
|
|
const members = [
|
|
{ member: 0, value: jsonPart("€".repeat(length)) },
|
|
{ member: 1, value: "1" },
|
|
];
|
|
const message: Message = { kind: "bound", holds: [{ hold: 1, variant: 0, members, failed: [] }] };
|
|
expect(plain(binaryCodec.decode(binaryCodec.encode(message)))).toEqual(message);
|
|
}
|
|
});
|
|
});
|
|
|
|
describe("the binary codec's long strings", () => {
|
|
const texts = [
|
|
["ASCII", "x".repeat(10_000)],
|
|
["three-byte characters", "€".repeat(10_000)],
|
|
["surrogate pairs", "😀".repeat(5_000)],
|
|
["mixed, at the in-place threshold", `${"a".repeat(4_093)}😀€`],
|
|
["a leading U+FEFF", `\uFEFF${"y".repeat(5_000)}`],
|
|
];
|
|
|
|
it.each(texts)("round-trip %s, encoded in place behind a padded length", (_, text) => {
|
|
const message: Message = { kind: "result", call: 0, value: jsonPart(text) };
|
|
const bytes = binaryCodec.encode(message);
|
|
expect(plain(binaryCodec.decode(bytes))).toEqual(message);
|
|
expect(bytes.buffer.byteLength - bytes.byteLength).toBeLessThanOrEqual(4096);
|
|
});
|
|
});
|
|
|
|
describe("the binary codec's short strings", () => {
|
|
it.each([0, 1, 41, 42, 43])("round-trip at %i three-byte characters, around the one-byte length", (length) => {
|
|
const type = "€".repeat(length);
|
|
const message: Message = {
|
|
kind: "hold",
|
|
hold: 0,
|
|
type,
|
|
version: 1,
|
|
fingerprint: "0123456789abcdef",
|
|
params: jsonPart(type),
|
|
address: server,
|
|
};
|
|
expect(plain(binaryCodec.decode(binaryCodec.encode(message)))).toEqual(message);
|
|
});
|
|
});
|