Mouth: id shape on every delivery field, mentions bounded in bytes, round cap counts every target GET
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Heimdall
parent
7231cdb977
commit
d300ba1c8c
4 changed files with 99 additions and 22 deletions
File diff suppressed because one or more lines are too long
|
|
@ -2020,3 +2020,72 @@ test("reply link: the scrub removes U+2065, U+FFF0-FFF8, U+E0080-E00FF and U+E01
|
|||
]);
|
||||
assert.deepEqual(made.map(d => d.replyTo.excerpt), ["ab", "cde", "fgh", "ijk"]);
|
||||
});
|
||||
|
||||
// ---- Lows after PR #68: id shape on every delivery field, mentions bounded in bytes, a read cap on all target GETs.
|
||||
const ID_OK = { seq: 1, roomAlias: "#swarm:gamecult.test", roomId: ROOM, eventId: "$e", sender: ALICE_ID, senderDisplay: "alice", text: "hi", atUnixMs: 1, context: [{ sender: ALICE_ID, text: "x", atUnixMs: 0 }], mentions: [SELF], isDirect: false, image: null };
|
||||
|
||||
test("round cap: 12 plain replies to 12 distinct targets make at most 8 /event/ GETs a round (basis reads count), each delivered once, in order, with reply_to", async () => {
|
||||
const ids = Array.from({ length: 12 }, (_, i) => `$t${i}`);
|
||||
const hs = await fakeHomeserver({ preload: { [ROOM]: ids.map(t => target(t, ALICE_ID, `text ${t}`)) }, sync: since => since === null ? { next_batch: "b0" } : since === "b0" ? { next_batch: "b1", ...joined({ [ROOM]: { timeline: { events: ids.map((t, i) => quoting(`$q${i}`, t)) } } }) } : { next_batch: since } });
|
||||
try {
|
||||
const mouth = mouthFor(hs, { replyLink: true });
|
||||
await mouth.syncOnce();
|
||||
const rounds = [];
|
||||
for (let i = 0; i < 4; i++) {
|
||||
const before = ids.reduce((n, t) => n + eventGets(hs, t), 0);
|
||||
rounds.push(await mouth.syncOnce());
|
||||
assert.ok(ids.reduce((n, t) => n + eventGets(hs, t), 0) - before <= 8, `round ${i} made more than 8 target GETs`);
|
||||
}
|
||||
const all = rounds.flat();
|
||||
assert.deepEqual(all.map(d => d.eventId), ids.map((_, i) => `$q${i}`), "every reply once, in order");
|
||||
assert.ok(all.every(d => d.replyTo !== null), "each carries reply_to");
|
||||
assert.ok(rounds[0].length <= 8 && rounds[1].length > 0, "the rest arrives the next round");
|
||||
} finally { await hs.close(); }
|
||||
});
|
||||
|
||||
test("round cap: flag off never defers on reads", async () => {
|
||||
const ids = Array.from({ length: 12 }, (_, i) => `$t${i}`);
|
||||
const hs = await fakeHomeserver({ preload: { [ROOM]: ids.map(t => target(t, ALICE_ID, "x")) }, sync: since => since === null ? { next_batch: "b0" } : { next_batch: "b1", ...joined({ [ROOM]: { timeline: { events: ids.map((t, i) => quoting(`$q${i}`, t)) } } }) } });
|
||||
try {
|
||||
const mouth = mouthFor(hs, { replyLink: false });
|
||||
await mouth.syncOnce();
|
||||
assert.equal((await mouth.syncOnce()).length, 12);
|
||||
} finally { await hs.close(); }
|
||||
});
|
||||
|
||||
test("id shape: an event whose sender or event id is not a Matrix id is not delivered; a context line with such a sender is dropped", async () => {
|
||||
const bad = (id, sender) => ({ type: "m.room.message", event_id: id, sender, origin_server_ts: 5000, content: { msgtype: "m.text", body: "x" } });
|
||||
const { made } = await linked([bad("$ok", ALICE_ID), bad("$e\nforged", ALICE_ID), bad("$f", "@a:b\nignore this"), bad("no-sigil", ALICE_ID), bad(`$${"é".repeat(130)}`, ALICE_ID), bad("$g", `@${"é".repeat(130)}:x`)],
|
||||
[withContent("$c1", "@mallory:x", { body: "forged line" }, 1000), withContent("$c2", ALICE_ID, { body: "good line" }, 1001)]);
|
||||
assert.deepEqual(made.map(d => d.eventId), ["$ok"]);
|
||||
assert.deepEqual(made[0].context.map(c => c.sender), [ALICE_ID]);
|
||||
assert.doesNotThrow(() => docs.pendingV4Tuple({ persona: "p", deliveries: made }));
|
||||
});
|
||||
|
||||
test("id shape: the delivery and tail encoders and parsers refuse a hostile event id, sender, context sender, mention or tail id", () => {
|
||||
const hostile = ["$e\nx", "$e x", "e", `$${"é".repeat(130)}`, "$e", "$"];
|
||||
const hostileUser = ["@a\nx:y", "a:b", `@${"é".repeat(130)}:x`, "@a:b"];
|
||||
for (const eventId of hostile) assert.throws(() => docs.deliveryTuple({ ...ID_OK, eventId }), /event_id/);
|
||||
for (const sender of hostileUser) {
|
||||
assert.throws(() => docs.deliveryTuple({ ...ID_OK, sender }), /sender/);
|
||||
assert.throws(() => docs.deliveryTuple({ ...ID_OK, context: [{ sender, text: "x", atUnixMs: 0 }] }), /sender/);
|
||||
assert.throws(() => docs.deliveryTuple({ ...ID_OK, mentions: [sender] }), /mention/);
|
||||
assert.throws(() => docs.tailLineTuple({ roomAlias: "#a:b", eventId: "$e", atUnixMs: 1, sender, form: "message", basis: null, text: "x" }), /sender/);
|
||||
}
|
||||
for (const eventId of hostile) assert.throws(() => docs.tailLineTuple({ roomAlias: "#a:b", eventId, atUnixMs: 1, sender: ALICE_ID, form: "message", basis: null, text: "x" }), /event_id/);
|
||||
const good = docs.deliveryV4Tuple({ ...ID_OK, basis: null, form: "message", replyTo: null });
|
||||
assert.doesNotThrow(() => docs.parseDeliveryV4(good));
|
||||
const parsed = (i, v) => { const t = structuredClone(good); if (i === 8) t[8][0][0] = v; else t[i] = v; return t; };
|
||||
assert.throws(() => docs.parseDeliveryV4(parsed(3, "$e\nx")), /event_id/);
|
||||
assert.throws(() => docs.parseDeliveryV4(parsed(4, "@a\nb:c")), /sender/);
|
||||
assert.throws(() => docs.parseDeliveryV4(parsed(8, "@a\nb:c")), /sender/);
|
||||
const pending = docs.pendingV3Tuple({ persona: "p", deliveries: [], tail: [{ roomAlias: "#a:b", eventId: "$e", atUnixMs: 1, sender: ALICE_ID, form: "message", basis: null, text: "x" }] });
|
||||
assert.doesNotThrow(() => docs.parsePendingV3(pending));
|
||||
assert.throws(() => docs.parsePendingV3(JSON.parse(JSON.stringify(pending).replace('"$e"', '"$e\\nx"'))), /event_id/);
|
||||
assert.throws(() => docs.parsePendingV3(JSON.parse(JSON.stringify(pending).replace(`"${ALICE_ID}"`, '"@a\\nb:c"'))), /sender/);
|
||||
});
|
||||
|
||||
test("id shape: a mention longer than 255 bytes, though under 255 characters, is dropped; the delivery goes", async () => {
|
||||
const { made } = await linked([withContent("$m", META, { body: "hi", "m.mentions": { user_ids: [SELF, `@${"é".repeat(200)}:x`, "@a b:x"] } })], []);
|
||||
assert.deepEqual(made.map(d => d.mentions), [[SELF]]);
|
||||
});
|
||||
|
|
|
|||
|
|
@ -174,10 +174,10 @@ export function parseOpenDirectReceipt(t) {
|
|||
|
||||
export function deliveryTuple(d) {
|
||||
return [
|
||||
seq(d.seq, "seq"), text(d.roomAlias, "room_alias"), text(d.roomId, "room_id"), text(d.eventId, "event_id"),
|
||||
text(d.sender, "sender"), text(d.senderDisplay, "sender_display"), String(d.text), seq(d.atUnixMs, "at_unix_ms"),
|
||||
d.context.slice(-maxContext).map(c => [text(c.sender, "sender"), String(c.text), seq(c.atUnixMs, "at_unix_ms")]),
|
||||
d.mentions.map(id => text(id, "mention")), flag(d.isDirect, "is_direct"), imageTuple(d.image ?? null),
|
||||
seq(d.seq, "seq"), text(d.roomAlias, "room_alias"), text(d.roomId, "room_id"), matrixId("$", d.eventId, "event_id"),
|
||||
matrixId("@", d.sender, "sender"), text(d.senderDisplay, "sender_display"), String(d.text), seq(d.atUnixMs, "at_unix_ms"),
|
||||
d.context.slice(-maxContext).map(c => [matrixId("@", c.sender, "sender"), String(c.text), seq(c.atUnixMs, "at_unix_ms")]),
|
||||
d.mentions.map(id => matrixId("@", id, "mention")), flag(d.isDirect, "is_direct"), imageTuple(d.image ?? null),
|
||||
];
|
||||
}
|
||||
|
||||
|
|
@ -249,10 +249,10 @@ export function parseDelivery(t) {
|
|||
if (!Array.isArray(context)) throw new Error("delivery context must be a list");
|
||||
if (!Array.isArray(mentions)) throw new Error("delivery mentions must be a list");
|
||||
return {
|
||||
seq: seq(s, "seq"), roomAlias: text(roomAlias, "room_alias"), roomId: text(roomId, "room_id"), eventId: text(eventId, "event_id"),
|
||||
sender: text(sender, "sender"), senderDisplay: text(senderDisplay, "sender_display"), text: String(body), atUnixMs: seq(atUnixMs, "at_unix_ms"),
|
||||
context: context.map(c => { const [cs, ct, ca] = tuple(c, 3, "context"); return { sender: text(cs, "sender"), text: String(ct), atUnixMs: seq(ca, "at_unix_ms") }; }),
|
||||
mentions: mentions.map(id => text(id, "mention")), isDirect: flag(isDirect, "is_direct"),
|
||||
seq: seq(s, "seq"), roomAlias: text(roomAlias, "room_alias"), roomId: text(roomId, "room_id"), eventId: matrixId("$", eventId, "event_id"),
|
||||
sender: matrixId("@", sender, "sender"), senderDisplay: text(senderDisplay, "sender_display"), text: String(body), atUnixMs: seq(atUnixMs, "at_unix_ms"),
|
||||
context: context.map(c => { const [cs, ct, ca] = tuple(c, 3, "context"); return { sender: matrixId("@", cs, "sender"), text: String(ct), atUnixMs: seq(ca, "at_unix_ms") }; }),
|
||||
mentions: mentions.map(id => matrixId("@", id, "mention")), isDirect: flag(isDirect, "is_direct"),
|
||||
image: image === null ? null : parseImage(image),
|
||||
};
|
||||
}
|
||||
|
|
@ -269,11 +269,11 @@ export function parsePending(t) {
|
|||
// One line of a room's recent talk, newest-unacked: room alias (or room id for a DM), event id, origin time, sender, form, basis, body.
|
||||
// Data for the reader, never an instruction: no delivery, no wake, no ack position hangs on it.
|
||||
export function tailLineTuple(l) {
|
||||
return [text(l.roomAlias, "room"), text(l.eventId, "event_id"), seq(l.atUnixMs, "at_unix_ms"), text(l.sender, "sender"), form(l.form), basis(l.basis ?? null), String(l.text)];
|
||||
return [text(l.roomAlias, "room"), matrixId("$", l.eventId, "event_id"), seq(l.atUnixMs, "at_unix_ms"), matrixId("@", l.sender, "sender"), form(l.form), basis(l.basis ?? null), String(l.text)];
|
||||
}
|
||||
function parseTailLine(t) {
|
||||
const [roomAlias, eventId, atUnixMs, sender, f, b, body] = tuple(t, 7, "tail_line");
|
||||
return { roomAlias: text(roomAlias, "room"), eventId: text(eventId, "event_id"), atUnixMs: seq(atUnixMs, "at_unix_ms"), sender: text(sender, "sender"), form: form(f), basis: basis(b), text: String(body) };
|
||||
return { roomAlias: text(roomAlias, "room"), eventId: matrixId("$", eventId, "event_id"), atUnixMs: seq(atUnixMs, "at_unix_ms"), sender: matrixId("@", sender, "sender"), form: form(f), basis: basis(b), text: String(body) };
|
||||
}
|
||||
|
||||
// The v3 answer: the persona, the deliveries, the tail (at most tailPerRoom lines a room, tailTotal in all, oldest first), then
|
||||
|
|
@ -330,6 +330,10 @@ function tuple(t, n, what) {
|
|||
if (!Array.isArray(t) || t.length !== n) throw new Error(`${what} must be a ${n}-field tuple`);
|
||||
return t;
|
||||
}
|
||||
function matrixId(sigil, v, what) {
|
||||
if (!isMatrixId(sigil, v)) throw new Error(`${what} must be a Matrix id (${sigil}..., at most 255 bytes, no whitespace or control characters)`);
|
||||
return v;
|
||||
}
|
||||
function text(v, what) {
|
||||
if (typeof v !== "string" || v === "") throw new Error(`${what} must be a non-empty string`);
|
||||
return v;
|
||||
|
|
|
|||
|
|
@ -288,6 +288,7 @@ export function createMouth({
|
|||
const lines = [];
|
||||
let bytes = 0;
|
||||
for (const e of events.slice(0, cut).reverse()) {
|
||||
if (!docs.isMatrixId("@", e.sender)) continue; // a line the tuple would refuse is left out, not the whole delivery
|
||||
const l = lineOf(e);
|
||||
const size = Buffer.byteLength(l.text) + Buffer.byteLength(l.sender);
|
||||
if (lines.length >= CONTEXT_LINES || bytes + size > CONTEXT_BYTES) break;
|
||||
|
|
@ -313,7 +314,7 @@ export function createMouth({
|
|||
const pillFor = html => typeof html === "string" && [...html.replace(/<mx-reply>[\s\S]*?<\/mx-reply>/gi, "").matchAll(/<a\s[^>]*?href\s*=\s*(["'])(.*?)\1/gi)].some(([, , href]) => [selfId, encodeURIComponent(selfId)].some(id => href === `https://matrix.to/#/${id}`));
|
||||
const authors = new Map(); // `${room id} ${event id}` -> { sender, excerpt, at }: read results, kept TARGET_TTL_MS, bounded; the key holds the room so another room's reply cannot borrow a hit
|
||||
const TARGET_TTL_MS = 60 * 1000; // how long a read of a target is kept: a failure always; a hit only with replyLink (without it a hit is kept until bounded out, as before v4)
|
||||
const REPLY_READS_PER_ROUND = 8; // distinct uncached targets replyToOf may read in one sync round; past it reply_to is nil (the tuple needs a sender, which only a read gives)
|
||||
const REPLY_READS_PER_ROUND = 8; // target GETs one sync round may make, whoever asks (basis or reply_to): an event whose uncached targets would pass it waits for the next round
|
||||
const KEEP_HIT_MS = 10 * 60 * 1000; // how long transient failures may keep a hit's sender alive after its last good read
|
||||
let replyReads = 0;
|
||||
const cachedTarget = key => {
|
||||
|
|
@ -353,6 +354,7 @@ export function createMouth({
|
|||
if (cached) return cached; // with replyLink a hit is short too: a later redaction or edit of the target must show at a re-read
|
||||
const known = authors.get(key);
|
||||
let sender = null, excerpt = null, transient = false;
|
||||
replyReads++;
|
||||
try {
|
||||
const target = await matrix("GET", `/rooms/${encodeURIComponent(roomId)}/event/${encodeURIComponent(eventId)}`, undefined, API, AbortSignal.timeout(sendTimeoutMs));
|
||||
if (target?.room_id === undefined || target.room_id === roomId) { // an id from another room is not an answer about this one: no sender, no text
|
||||
|
|
@ -374,16 +376,18 @@ export function createMouth({
|
|||
const authorOf = async (roomId, eventId) => (await targetOf(roomId, eventId)).sender;
|
||||
// The event a message replies to: its own m.in_reply_to, unless that is a thread's fallback pointer (is_falling_back), which
|
||||
// names no one. nil when the target cannot be read or sits in another room.
|
||||
const DEFER = Symbol("defer"); // replyToOf's answer when the round's read cap is spent: the delivery waits for the next round, as at the pending cap
|
||||
// The GETs an event could cost in targetOf: its uncached reply target and thread root. An over-count is harmless (the event waits a round); an under-count would pass the cap.
|
||||
const targetsToRead = (roomId, event) => {
|
||||
const relates = event.content["m.relates_to"];
|
||||
if (!replyLink || formOf(event.content) !== "message") return 0;
|
||||
const ids = new Set([relates?.["m.in_reply_to"]?.event_id, relates?.rel_type === "m.thread" ? relates.event_id : undefined].filter(id => typeof id === "string" && id !== ""));
|
||||
return [...ids].filter(id => !cachedTarget(`${roomId} ${id}`)).length;
|
||||
};
|
||||
async function replyToOf(roomId, event) {
|
||||
if (!replyLink || formOf(event.content) !== "message") return null;
|
||||
const r = event.content["m.relates_to"]?.["m.in_reply_to"];
|
||||
const id = r?.event_id;
|
||||
if (typeof id !== "string" || id === "" || event.content["m.relates_to"]["is_falling_back"] === true) return null;
|
||||
if (!cachedTarget(`${roomId} ${id}`)) {
|
||||
if (replyReads >= REPLY_READS_PER_ROUND) { say(`reply target ${id} not read: round cap, the rest of the round waits for the next`); return DEFER; }
|
||||
replyReads++;
|
||||
}
|
||||
const { sender, excerpt } = await targetOf(roomId, id);
|
||||
return sender === null || !docs.isMatrixId("$", id) ? null : { eventId: id, sender, excerpt };
|
||||
}
|
||||
|
|
@ -450,15 +454,15 @@ export function createMouth({
|
|||
if ((room.timeline?.events ?? []).length > 0) tailReads.delete(roomId);
|
||||
for (const event of room.timeline?.events ?? []) {
|
||||
if (event.type !== "m.room.message" || typeof event.content?.body !== "string") continue;
|
||||
if (typeof event.sender !== "string" || event.sender === "" || typeof event.event_id !== "string" || event.event_id === "") continue;
|
||||
if (!docs.isMatrixId("@", event.sender) || !docs.isMatrixId("$", event.event_id)) continue; // an id the tuple would refuse is no event: it could not be delivered, and one such event must not wedge the round
|
||||
if (processed.has(event.event_id)) continue; // read already in a round the cap kept from advancing
|
||||
const ts = event.origin_server_ts;
|
||||
const atUnixMs = Number.isSafeInteger(ts) && ts >= 0 ? ts : now();
|
||||
const deliverable = !baseline && !freshRooms.has(roomId) && event.sender !== selfId;
|
||||
if (deliverable && pending.length + made.length >= maxPending) { truncated = true; break rounds; }
|
||||
if (deliverable && replyLink && replyReads + targetsToRead(roomId, event) > REPLY_READS_PER_ROUND) { say(`event ${event.event_id} not read: round cap on target reads, the rest of the round waits for the next`); truncated = true; break rounds; } // not processed: the next round meets it again, with its read count reset
|
||||
const basis = deliverable ? await basisOf(roomId, event) : null;
|
||||
const replyTo = deliverable ? await replyToOf(roomId, event) : null;
|
||||
if (replyTo === DEFER) { truncated = true; break rounds; } // not processed: the next round meets this event again, with its read count reset
|
||||
processed.add(event.event_id);
|
||||
if (deliverable) {
|
||||
const context = await contextFor(roomId, event, round);
|
||||
|
|
@ -469,7 +473,7 @@ export function createMouth({
|
|||
seq: ++seqCounter, roomAlias: alias, roomId, eventId: event.event_id, sender: event.sender,
|
||||
senderDisplay: displayNames.get(event.sender) ?? localpart(event.sender), text: bodyText(event.content, docs.limits.textBytes, refused), atUnixMs,
|
||||
context, image,
|
||||
mentions: Array.isArray(mentioned) ? mentioned.filter(id => typeof id === "string" && id !== "" && id.length <= 255).slice(0, docs.limits.mentions) : [],
|
||||
mentions: Array.isArray(mentioned) ? mentioned.filter(id => docs.isMatrixId("@", id)).slice(0, docs.limits.mentions) : [],
|
||||
isDirect: directRooms.has(roomId), basis, form: formOf(event.content), replyTo,
|
||||
});
|
||||
if (image) images.hold(image.path, seqCounter);
|
||||
|
|
@ -576,7 +580,7 @@ export function createMouth({
|
|||
if (truncated || events.length > docs.limits.tailPerRoom) gapped.add(alias);
|
||||
const lines = [];
|
||||
for (const e of events.slice(-docs.limits.tailPerRoom)) {
|
||||
if (typeof e.event_id !== "string" || e.event_id === "") { gapped.add(alias); continue; }
|
||||
if (!docs.isMatrixId("$", e.event_id) || !docs.isMatrixId("@", e.sender)) { gapped.add(alias); continue; }
|
||||
const at = Number.isSafeInteger(e.origin_server_ts) && e.origin_server_ts >= 0 ? e.origin_server_ts : 0;
|
||||
lines.push({ roomAlias: alias, eventId: e.event_id, atUnixMs: at, sender: e.sender, form: formOf(e.content), basis: await basisOf(roomId, e, false), text: bodyText(e.content, docs.limits.contextBytes) });
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue