Matrix mouth: read receipts on ack, presence from the brake #13
4 changed files with 320 additions and 14 deletions
|
|
@ -196,7 +196,7 @@ The command is a dry-run shape probe for public surfaces that are real enough to
|
|||
|
||||
Bifrost's native Discord gateway is the conversation observer. A direct mention of the Bifrost bot, or a reply to one of its messages, becomes an immutable typed `bifrost.discord.persona_conversation_event.v0` only when the guild and channel have one exact Persona target binding. Bot-authored, ambient, empty, over-1200-character, unbound, and ambiguously bound messages do not enter the conversation path. The 1200-character cap is the shared signed alpha boundary with Epiphany. The observation store is Bifrost-private working state, never an actuator queue, and insert collisions must contain exactly equal content. VoidBot does not observe, schedule, or answer for the target Persona.
|
||||
|
||||
Persona speech goes out through Matrix, not Discord. `tools/persona-matrix-mouth.mjs` is the one process for every Persona (`PERSONAS`, a list of slugs): it holds each Persona's Matrix access token as its own systemd credential (`matrix-token-<persona>`) and nobody else holds any. Per Persona it keeps one `/sync` loop over the configured rooms and the account's direct rooms (a DM invitation from an account on `MATRIX_SERVER_NAME` is accepted while that Persona's brake is released, and one from any other server is declined), one cursor and one delivery sequence. A delivery carries the sender's Matrix id, the event's `m.mentions` user ids and whether the room is a DM. A media event (`m.image`, `m.file`, `m.audio`, `m.video`) is delivered as a typed text line, `[image filename="cat.png" mime=image/png size=2048 url=mxc://server/id]`, followed by its caption when it has one, in `text` and in context. Only `m.image` is also fetched, and only when `IMAGE_STORE_ROOT` is set (unset, the line above is all there is and nothing is downloaded). The fetch is made as the receiving Persona, with its own token, from `/_matrix/client/v1/media/download/<server>/<mediaId>` on its homeserver, for a message the Persona is delivered (not its own words, a room it is not joined to, or history before the ear), and only from the event's own `content.url`; a URL in the message text is never followed. An encrypted attachment (`m.room.encrypted`, or `content.file`) is never fetched. The type is decided by the magic bytes alone (png, jpeg, gif, webp; SVG and anything else is refused, the declared mime is ignored), the body is capped at 5 MiB by `Content-Length` and by the bytes actually read, and the pixel count read from the image header (never decoded) is capped at 40 MP. A refused or failed fetch leaves the typed line with a fixed-word reason (`refused=encrypted|no-url|fetch-failed|too-big|wrong-type|bad-header|too-many-pixels|store-failed`) and no image. A stored image is written to `IMAGE_STORE_ROOT/<persona>/<hex sha256 of the room id>/<hex sha256 of the bytes>[.png|.jpg|.gif|.webp]`, the name generated by the mouth, never the sender's: files `0640`, room directories `2750` (the Persona directory is setgid to the Persona's group, so the Persona's reader opens them by group; the mouth is the only writer), each written under a temporary name and renamed. It is deleted once it is at least 24 hours old and its delivery has been acked, whichever is later; a room keeps at most 50 MiB, the oldest evicted first. The sweep runs when an image is stored and when an ack arrives. A delivery is the 12-field tuple of `gamecult.persona.chat_pending.v2`: field 11, after `is_direct`, is nil or `[path, mime, size, sha256]`, a reference and never bytes; `path` is the store-relative `<room hash>/<file>` above, `mime` the sniffed type, `size` the file length (at most 5 MiB) and `sha256` its lowercase hex digest. The reader re-checks all four against the file. A Persona turn that has no reader for it still sees the typed line. Reading wakes nothing and acks nothing: the heartbeat polls with the non-consuming `gamecult.persona.chat_pending_request.v1`, and only `chat_ack` advances a position. The mouth speaks on a `gamecult.persona.chat_send_request.v1` and joins rooms only while that Persona's brake (`/etc/gamecult/minds/<persona>/control.cc`) is released. One loopback CultNet RUDP endpoint serves all of them; a request naming an unlisted Persona gets a typed `chat_refusal` (`UnknownPersona`). Document shapes: `tools/persona-chat-documents.mjs`; golden bytes: `tests/fixtures/persona-chat-v1/`, `-v2/` and `-v3/` (the pending answer; Huginn pins the same bytes). The Discord posting actuator, its permit and its RUDP listener are removed.
|
||||
Persona speech goes out through Matrix, not Discord. `tools/persona-matrix-mouth.mjs` is the one process for every Persona (`PERSONAS`, a list of slugs): it holds each Persona's Matrix access token as its own systemd credential (`matrix-token-<persona>`) and nobody else holds any. Per Persona it keeps one `/sync` loop over the configured rooms and the account's direct rooms (a DM invitation from an account on `MATRIX_SERVER_NAME` is accepted while that Persona's brake is released, and one from any other server is declined), one cursor and one delivery sequence. A delivery carries the sender's Matrix id, the event's `m.mentions` user ids and whether the room is a DM. A media event (`m.image`, `m.file`, `m.audio`, `m.video`) is delivered as a typed text line, `[image filename="cat.png" mime=image/png size=2048 url=mxc://server/id]`, followed by its caption when it has one, in `text` and in context. Only `m.image` is also fetched, and only when `IMAGE_STORE_ROOT` is set (unset, the line above is all there is and nothing is downloaded). The fetch is made as the receiving Persona, with its own token, from `/_matrix/client/v1/media/download/<server>/<mediaId>` on its homeserver, for a message the Persona is delivered (not its own words, a room it is not joined to, or history before the ear), and only from the event's own `content.url`; a URL in the message text is never followed. An encrypted attachment (`m.room.encrypted`, or `content.file`) is never fetched. The type is decided by the magic bytes alone (png, jpeg, gif, webp; SVG and anything else is refused, the declared mime is ignored), the body is capped at 5 MiB by `Content-Length` and by the bytes actually read, and the pixel count read from the image header (never decoded) is capped at 40 MP. A refused or failed fetch leaves the typed line with a fixed-word reason (`refused=encrypted|no-url|fetch-failed|too-big|wrong-type|bad-header|too-many-pixels|store-failed`) and no image. A stored image is written to `IMAGE_STORE_ROOT/<persona>/<hex sha256 of the room id>/<hex sha256 of the bytes>[.png|.jpg|.gif|.webp]`, the name generated by the mouth, never the sender's: files `0640`, room directories `2750` (the Persona directory is setgid to the Persona's group, so the Persona's reader opens them by group; the mouth is the only writer), each written under a temporary name and renamed. It is deleted once it is at least 24 hours old and its delivery has been acked, whichever is later; a room keeps at most 50 MiB, the oldest evicted first. The sweep runs when an image is stored and when an ack arrives. A delivery is the 12-field tuple of `gamecult.persona.chat_pending.v2`: field 11, after `is_direct`, is nil or `[path, mime, size, sha256]`, a reference and never bytes; `path` is the store-relative `<room hash>/<file>` above, `mime` the sniffed type, `size` the file length (at most 5 MiB) and `sha256` its lowercase hex digest. The reader re-checks all four against the file. A Persona turn that has no reader for it still sees the typed line. Reading wakes nothing and acks nothing: the heartbeat polls with the non-consuming `gamecult.persona.chat_pending_request.v1`, and only `chat_ack` advances a position. An ack also posts that Persona's public `m.read` receipt, per room, at the newest acked event (fire and forget: the ack never waits on it, a failure is logged once and not retried; the room's next ack covers it); the heartbeat's and a turn's pending calls post none. Her Matrix presence follows her brake: each `/sync` carries `set_presence=online` while it is released and `offline` while it is held, and a round that finds her braked makes one `PUT /presence/<user>/status` to offline first (a brake placed mid long-poll shows within `syncTimeoutMs`, 30 s, plus one PUT; a stopped mouth reads offline after Synapse's sync timeout). The mouth speaks on a `gamecult.persona.chat_send_request.v1` and joins rooms only while that Persona's brake (`/etc/gamecult/minds/<persona>/control.cc`) is released. One loopback CultNet RUDP endpoint serves all of them; a request naming an unlisted Persona gets a typed `chat_refusal` (`UnknownPersona`). Document shapes: `tools/persona-chat-documents.mjs`; golden bytes: `tests/fixtures/persona-chat-v1/`, `-v2/` and `-v3/` (the pending answer; Huginn pins the same bytes). The Discord posting actuator, its permit and its RUDP listener are removed.
|
||||
|
||||
Bifrost admits an observation only when guild, channel, target runtime, canonical repository, Persona, producer, and producer runtime match a preconfigured `bifrost.persona_feedback.target_binding.v0` document in Bifrost's private provider store. That binding also owns source visibility, data classification, and the delivery-store target. Bifrost never infers public visibility from Discord identifiers. It classifies the author using its own `bifrost.discord.actor_link.v0` state: a match yields `linked_governance_feedback`; absence yields `unlinked_social_feedback`. Neither class grants adoption, work, release, or deployment authority.
|
||||
|
||||
|
|
|
|||
|
|
@ -10,7 +10,7 @@ import { mkdir, mkdtemp, readFile, writeFile } from "node:fs/promises";
|
|||
import { tmpdir } from "node:os";
|
||||
import { resolve } from "node:path";
|
||||
import * as docs from "../tools/persona-chat-documents.mjs";
|
||||
import { cultcacheState, connectionIdFor, createBrakeReader, loadRuntime, pageOf, startChatEndpoint } from "../tools/persona-matrix-mouth.mjs";
|
||||
import { createMouth, cultcacheState, connectionIdFor, createBrakeReader, loadRuntime, pageOf, startChatEndpoint } from "../tools/persona-matrix-mouth.mjs";
|
||||
|
||||
const root = resolve(import.meta.dirname, "..");
|
||||
const runtime = loadRuntime(resolve(process.env.CULTLIB_ROOT ?? resolve(root, "..", "CultLib")));
|
||||
|
|
@ -370,3 +370,55 @@ test("the RUDP endpoint answers a v1 send with the three-field receipt and a v2
|
|||
try { client.close(); } catch { /* the connection already closed its socket */ }
|
||||
}
|
||||
});
|
||||
|
||||
test("over the RUDP endpoint: pending posts no receipt, and an ack never holds up the same peer's next request", async () => {
|
||||
const receipts = [];
|
||||
let release;
|
||||
const held = new Promise(r => { release = r; });
|
||||
const json = (body, status = 200) => new Response(JSON.stringify(body), { status, headers: { "content-type": "application/json" } });
|
||||
const message = { type: "m.room.message", event_id: "$m1", sender: "@alice:t", origin_server_ts: 5, content: { msgtype: "m.text", body: "hi" } };
|
||||
const fetch = async (url, init = {}) => {
|
||||
const u = new URL(url), p = u.pathname.replace("/_matrix/client/v3", "");
|
||||
if (p === "/account/whoami") return json({ user_id: "@eureka:t" });
|
||||
if (p === "/joined_rooms") return json({ joined_rooms: ["!r:t"] });
|
||||
if (p.startsWith("/directory/room/")) return json({ room_id: "!r:t" });
|
||||
if (p.includes("/account_data/")) return json({}, 404);
|
||||
if (p === "/sync") return json(u.searchParams.get("since") === "b0" ? { next_batch: "b1", rooms: { join: { "!r:t": { timeline: { events: [message] } } } } } : { next_batch: "b0" });
|
||||
if (p.includes("/messages")) return json({ chunk: [] });
|
||||
if (p.includes("/receipt/")) { receipts.push(p); await held; return json({}); }
|
||||
return json({}, 404);
|
||||
};
|
||||
const state = { load: async () => null, save: async () => {} };
|
||||
const mouth = createMouth({ persona: "eureka", homeserver: "http://hs.test", rooms: ["#r:t"], token: "tok", state, fetch, serverName: "t", brake: async () => true, syncTimeoutMs: 1 });
|
||||
await mouth.syncOnce(); await mouth.syncOnce();
|
||||
const key = "connection-key-for-receipts";
|
||||
const server = await startChatEndpoint({ mouths: new Map([["eureka", mouth]]), runtime, port: 0, connectionKey: key, runtimeId: "mouth-test" });
|
||||
const client = dgram.createSocket("udp4");
|
||||
await new Promise(ok => client.bind(0, "127.0.0.1", ok));
|
||||
const connection = new runtime.cultnet.CultNetRudpSocketTransportConnection({
|
||||
runtimeId: "huginn-test", socket: client, mode: "client", remoteHost: "127.0.0.1", remotePort: server.address.port,
|
||||
connectionId: connectionIdFor(key), initialSequence: 1, resendDelayMs: 50, resendPollMs: 10, maxFragmentBytes: 1200,
|
||||
});
|
||||
const replies = [];
|
||||
connection.on("frame", value => replies.push(runtime.cultnet.parseCultNetMessage(msgpack.decode(value.payload), "cultnet.schema.v0").document));
|
||||
const put = (kind, recordKey, tuple) => connection.send("schema", msgpack.encode(runtime.cultnet.encodeCultNetMessageForWire(
|
||||
docs.envelope({ kind, recordKey, payload: msgpack.encode(tuple), sourceRuntimeId: "huginn-test", sourceAgentId: "eureka", sourceRole: "persona-organ" }), "cultnet.schema.v0")));
|
||||
const waitFor = async (check, what) => { for (let i = 0; i < 250 && !check(); i++) await new Promise(r => setTimeout(r, 20)); assert.ok(check(), what); };
|
||||
try {
|
||||
connection.connect();
|
||||
await waitFor(() => connection.connected, "session opens");
|
||||
put(docs.kinds.pendingRequest, "eureka", docs.pendingRequestTuple({ persona: "eureka", afterSeq: 0 }));
|
||||
await waitFor(() => replies.length === 1, "pending answered");
|
||||
assert.deepEqual(receipts, [], "a pending request posts no receipt");
|
||||
put(docs.kinds.ack, "eureka:1", docs.ackTuple({ persona: "eureka", throughSeq: 1 }));
|
||||
await waitFor(() => receipts.length === 1, "the ack posted her receipt");
|
||||
put(docs.kinds.pendingRequest, "eureka", docs.pendingRequestTuple({ persona: "eureka", afterSeq: 0 }));
|
||||
await waitFor(() => replies.length === 2, "the next request is answered while the receipt response is still held open");
|
||||
assert.deepEqual(docs.parsePending(msgpack.decode(replies[1].payload)).deliveries, [], "the ack applied");
|
||||
} finally {
|
||||
release();
|
||||
server.close();
|
||||
connection.close();
|
||||
try { client.close(); } catch { /* already closed */ }
|
||||
}
|
||||
});
|
||||
|
|
|
|||
|
|
@ -18,12 +18,14 @@ const ROOMS = { [`#swarm:${HS_NAME}`]: "!swarm:gamecult.test", [`#operator:${HS_
|
|||
const SELF = `@eureka:${HS_NAME}`;
|
||||
|
||||
const message = (id, sender, body, ts) => ({ type: "m.room.message", event_id: id, sender, origin_server_ts: ts, content: { msgtype: "m.text", body } });
|
||||
// The one write a braked round makes by design; an online presence PUT or any other write is not exempt.
|
||||
const offlinePresence = r => r.method === "PUT" && r.path.includes("/presence/") && r.body?.presence === "offline";
|
||||
const tick = () => new Promise(r => setImmediate(r));
|
||||
|
||||
// A fake Matrix client-server API: just the endpoints the mouth uses.
|
||||
// It is also the record of every room: events the sync script emits and messages sent are kept, and
|
||||
// `preload` (room id -> events, oldest first) stands for history from before the mouth's process began.
|
||||
async function fakeHomeserver({ sync, failWith, users, member, direct, preload = {}, failContext = false, media } = {}) {
|
||||
async function fakeHomeserver({ sync, failWith, users, member, direct, preload = {}, failContext = false, media, failReceipts = false, deferReceipts = null, failPresence = 0 } = {}) {
|
||||
const seen = [];
|
||||
const record_ = new Map(Object.entries(preload).map(([room, events]) => [room, [...events]]));
|
||||
const remember = (room, event) => { const list = record_.get(room) ?? []; if (!list.some(e => e.event_id === event.event_id)) list.push(event); record_.set(room, list); };
|
||||
|
|
@ -34,12 +36,17 @@ async function fakeHomeserver({ sync, failWith, users, member, direct, preload =
|
|||
req.on("data", c => (raw += c));
|
||||
req.on("end", () => {
|
||||
const url = new URL(req.url, "http://x");
|
||||
const record = { method: req.method, path: decodeURIComponent(url.pathname), query: url.searchParams, auth: req.headers.authorization, body: raw ? JSON.parse(raw) : null };
|
||||
const record = { method: req.method, rawPath: url.pathname, path: decodeURIComponent(url.pathname), query: url.searchParams, auth: req.headers.authorization, body: raw ? JSON.parse(raw) : null };
|
||||
seen.push(record);
|
||||
const reply = (status, body) => { res.writeHead(status, { "content-type": "application/json" }); res.end(JSON.stringify(body)); };
|
||||
if (failWith) return reply(failWith, { errcode: "M_UNKNOWN_TOKEN", error: `bad token ${record.auth}` });
|
||||
const p = record.path.replace("/_matrix/client/v3", "").replace("/_matrix/client/v1", "");
|
||||
const bearer = (record.auth ?? "").replace("Bearer ", "");
|
||||
if (req.method === "POST" && p.includes("/receipt/m.read/")) {
|
||||
if (deferReceipts) return deferReceipts.then(() => reply(200, {}));
|
||||
return failReceipts ? reply(500, { errcode: "M_UNKNOWN", error: `down ${record.auth}` }) : reply(200, {});
|
||||
}
|
||||
if (req.method === "PUT" && p.includes("/presence/")) return failPresence-- > 0 ? reply(500, { errcode: "M_UNKNOWN" }) : reply(200, {});
|
||||
if (p.startsWith("/media/download/")) {
|
||||
const served = media?.(bearer, ...p.slice("/media/download/".length).split("/")) ?? { status: 404, body: "" };
|
||||
res.writeHead(served.status ?? 200, served.headers ?? {});
|
||||
|
|
@ -77,7 +84,9 @@ async function fakeHomeserver({ sync, failWith, users, member, direct, preload =
|
|||
reply(404, { errcode: "M_NOT_FOUND" });
|
||||
});
|
||||
};
|
||||
const fetch = (url, init = {}) => new Promise(resolve => {
|
||||
const fetch = (url, init = {}) => new Promise((resolve, reject) => {
|
||||
if (init.signal?.aborted) return reject(init.signal.reason);
|
||||
init.signal?.addEventListener("abort", () => reject(init.signal.reason), { once: true });
|
||||
const u = new URL(url);
|
||||
const req = Object.assign(new EventEmitter(), { method: init.method ?? "GET", url: u.pathname + u.search, headers: init.headers ?? {} });
|
||||
const res = {
|
||||
|
|
@ -91,7 +100,9 @@ async function fakeHomeserver({ sync, failWith, users, member, direct, preload =
|
|||
url: "http://hs.test",
|
||||
fetch,
|
||||
seen,
|
||||
puts: () => seen.filter(r => r.method === "PUT"),
|
||||
puts: () => seen.filter(r => r.method === "PUT" && !offlinePresence(r)),
|
||||
receipts: () => seen.filter(r => r.path.includes("/receipt/")),
|
||||
presencePuts: () => seen.filter(r => r.path.includes("/presence/")),
|
||||
downloads: () => seen.filter(r => r.path.includes("/media/download/")),
|
||||
close: async () => {},
|
||||
};
|
||||
|
|
@ -443,7 +454,7 @@ test("join_while_braked: no join and no send while the brake is held; joining ha
|
|||
const mouth = mouthFor(hs, { brake: async () => open, log: l => logs.push(l) });
|
||||
await mouth.syncOnce();
|
||||
await mouth.send({ requestId: "r", persona: "eureka", roomAlias: `#swarm:${HS_NAME}`, text: "speak" });
|
||||
assert.equal(hs.seen.filter(r => r.method === "POST" || r.method === "PUT").length, 0, "braked: nothing is joined or sent");
|
||||
assert.equal(hs.seen.filter(r => (r.method === "POST" || r.method === "PUT") && !offlinePresence(r)).length, 0, "braked: nothing is joined or sent");
|
||||
open = true;
|
||||
await mouth.syncOnce();
|
||||
assert.deepEqual(hs.seen.filter(r => r.method === "POST").map(r => r.path.split("/join/")[1]).sort(), ["!operator:gamecult.test", "!swarm:gamecult.test"]);
|
||||
|
|
@ -588,7 +599,7 @@ test("dm_invite_waits_for_release: an invitation received while braked is not jo
|
|||
try {
|
||||
const mouth = mouthFor(hs, { brake: async () => open });
|
||||
await mouth.syncOnce(); await mouth.syncOnce(); await mouth.syncOnce();
|
||||
assert.deepEqual(hs.seen.filter(r => r.method === "POST" || r.method === "PUT"), [], "braked: no join, no leave, nothing written");
|
||||
assert.deepEqual(hs.seen.filter(r => (r.method === "POST" || r.method === "PUT") && !offlinePresence(r)), [], "braked: no join, no leave, nothing written");
|
||||
open = true;
|
||||
await mouth.syncOnce();
|
||||
assert.deepEqual(joinsOf(hs), [DMROOM]);
|
||||
|
|
@ -941,3 +952,218 @@ test("image_delivery: a store that cannot write leaves the line and a nil image,
|
|||
assert.ok(lines.some(l => l.includes("ENOSPC")));
|
||||
} finally { await hs.close(); }
|
||||
});
|
||||
|
||||
// Read receipts and presence: her ack posts her m.read; her brake sets her presence.
|
||||
const A = "!swarm:gamecult.test", B = "!operator:gamecult.test";
|
||||
const receiptScript = since => since === null ? { next_batch: "b0", ...joined({}) } : since === "b0" ? {
|
||||
next_batch: "b1",
|
||||
...joined({
|
||||
[A]: { timeline: { events: [message("$a1", `@alice:${HS_NAME}`, "one", 10), message("$a2", `@alice:${HS_NAME}`, "two", 11)] } },
|
||||
[B]: { timeline: { events: [message("$b1", `@bob:${HS_NAME}`, "three", 12)] } },
|
||||
}),
|
||||
} : { next_batch: since };
|
||||
const until = async check => { for (let i = 0; i < 200 && !check(); i++) await new Promise(r => setImmediate(r)); };
|
||||
const withMail = async (hs, overrides) => {
|
||||
const mouth = mouthFor(hs, overrides);
|
||||
await mouth.syncOnce();
|
||||
assert.equal((await mouth.syncOnce()).map(d => d.seq).join(), "1,2,3");
|
||||
return mouth;
|
||||
};
|
||||
const receiptsOf = hs => hs.receipts().map(r => [r.path.split("/rooms/")[1].split("/receipt/")[0], r.path.split("/m.read/")[1], r.auth, r.body]);
|
||||
|
||||
test("an ack posts one m.read per room at the newest covered event, as that Persona", async () => {
|
||||
const hs = await fakeHomeserver({ sync: receiptScript });
|
||||
const mouth = await withMail(hs);
|
||||
await mouth.ack(2);
|
||||
await until(() => hs.receipts().length >= 1);
|
||||
await new Promise(r => setTimeout(r, 20));
|
||||
assert.deepEqual(receiptsOf(hs), [[A, "$a2", `Bearer ${TOKEN}`, {}]], "room A at its second event; room B untouched");
|
||||
await mouth.ack(3);
|
||||
await until(() => hs.receipts().length >= 2);
|
||||
assert.deepEqual(receiptsOf(hs).slice(1), [[B, "$b1", `Bearer ${TOKEN}`, {}]]);
|
||||
assert.equal(hs.receipts().length, 2);
|
||||
});
|
||||
|
||||
test("pending posts no receipt", async () => {
|
||||
const hs = await fakeHomeserver({ sync: receiptScript });
|
||||
const mouth = await withMail(hs);
|
||||
for (const after of [0, 1, 2, 3]) mouth.pendingAfter(after);
|
||||
await new Promise(r => setTimeout(r, 20));
|
||||
assert.deepEqual(hs.receipts(), []);
|
||||
});
|
||||
|
||||
test("a failed receipt leaves the ack whole", async () => {
|
||||
const hs = await fakeHomeserver({ sync: receiptScript, failReceipts: true });
|
||||
const lines = [];
|
||||
const state = memoryState();
|
||||
const mouth = await withMail(hs, { state, log: l => lines.push(l) });
|
||||
const saves = [];
|
||||
const save = state.save;
|
||||
state.save = async v => { saves.push(v); return save(v); };
|
||||
await mouth.ack(3);
|
||||
await until(() => lines.some(l => l.includes("receipt in")));
|
||||
assert.equal(mouth.pendingCount, 0);
|
||||
assert.equal(state.box.value.nextBatch, "b1", "the cursor committed");
|
||||
assert.ok(saves.length >= 1, "state.save called");
|
||||
const failures = lines.filter(l => l.startsWith("receipt in "));
|
||||
assert.equal(failures.length, 2, "one line per room, nothing retried");
|
||||
assert.ok(failures.every(l => l.includes("not posted") && !l.includes(TOKEN)));
|
||||
await new Promise(r => setTimeout(r, 20));
|
||||
assert.equal(hs.receipts().length, 2, "no retry");
|
||||
});
|
||||
|
||||
test("the ack does not wait for the receipt", async () => {
|
||||
let release;
|
||||
const hs = await fakeHomeserver({ sync: receiptScript, deferReceipts: new Promise(r => { release = r; }) });
|
||||
const mouth = await withMail(hs);
|
||||
const outcome = await Promise.race([mouth.ack(3).then(() => "acked"), new Promise(r => setTimeout(() => r("waited on the receipt"), 500))]);
|
||||
assert.equal(outcome, "acked", "the ack resolves while the receipt response is held open");
|
||||
assert.equal(mouth.pendingCount, 0);
|
||||
release();
|
||||
await until(() => hs.receipts().length >= 2);
|
||||
});
|
||||
|
||||
test("presence follows the brake: online while released, offline and one PUT on the transition", async () => {
|
||||
let open = true;
|
||||
const hs = await fakeHomeserver({ sync: receiptScript });
|
||||
const mouth = mouthFor(hs, { brake: async () => open });
|
||||
const syncs = () => hs.seen.filter(r => r.path.endsWith("/sync")).map(r => r.query.get("set_presence"));
|
||||
await mouth.syncOnce(); await mouth.syncOnce();
|
||||
assert.deepEqual(syncs(), ["online", "online"]);
|
||||
assert.equal(hs.presencePuts().length, 0, "released: no PUT");
|
||||
open = false;
|
||||
await mouth.syncOnce(); await mouth.syncOnce(); await mouth.syncOnce();
|
||||
assert.deepEqual(syncs().slice(2), ["offline", "offline", "offline"]);
|
||||
assert.deepEqual(hs.presencePuts().map(r => [r.path, r.body]), [[`/_matrix/client/v3/presence/${SELF}/status`, { presence: "offline" }]], "exactly one PUT");
|
||||
assert.equal(hs.presencePuts()[0].rawPath, `/_matrix/client/v3/presence/${encodeURIComponent(SELF)}/status`, "the user id is percent-encoded on the wire");
|
||||
open = true;
|
||||
await mouth.syncOnce();
|
||||
assert.equal(syncs().at(-1), "online");
|
||||
open = false;
|
||||
await mouth.syncOnce();
|
||||
assert.equal(hs.presencePuts().length, 2, "a second transition PUTs once more");
|
||||
});
|
||||
|
||||
test("the first braked round of a fresh mouth PUTs offline once; a failed PUT is tried again next round", async () => {
|
||||
const hs = await fakeHomeserver({ sync: receiptScript, failPresence: 1 });
|
||||
const lines = [];
|
||||
const mouth = mouthFor(hs, { brake: async () => false, log: l => lines.push(l) });
|
||||
await mouth.syncOnce();
|
||||
assert.equal(hs.presencePuts().length, 1);
|
||||
assert.ok(lines.some(l => l.startsWith("presence not set")), "one log line");
|
||||
await mouth.syncOnce();
|
||||
assert.equal(hs.presencePuts().length, 2, "retried the next round");
|
||||
await mouth.syncOnce();
|
||||
assert.equal(hs.presencePuts().length, 2, "then settled");
|
||||
});
|
||||
|
||||
test("two Personas on one homeserver: receipts and presence carry only her own bearer and user id", async () => {
|
||||
const ALICE = `@pa:${HS_NAME}`, BOB = `@pb:${HS_NAME}`;
|
||||
const hs = await fakeHomeserver({ sync: receiptScript, users: { tokA: ALICE, tokB: BOB } });
|
||||
const a = await withMail(hs, { persona: "pa", token: "tokA" });
|
||||
let bOpen = false;
|
||||
const b = await withMail(hs, { persona: "pb", token: "tokB", brake: async () => bOpen });
|
||||
await a.ack(3);
|
||||
await b.syncOnce();
|
||||
await until(() => hs.receipts().length >= 2);
|
||||
assert.ok(hs.receipts().every(r => r.auth === "Bearer tokA"), "only A acked");
|
||||
bOpen = true; // her receipts go out only while released
|
||||
await b.ack(3);
|
||||
await until(() => hs.receipts().length >= 4);
|
||||
assert.deepEqual([...new Set(hs.receipts().map(r => r.auth))].sort(), ["Bearer tokA", "Bearer tokB"]);
|
||||
assert.ok(hs.presencePuts().length >= 1 && hs.presencePuts().every(r => r.path.endsWith(`/presence/${BOB}/status`) && r.auth === "Bearer tokB"));
|
||||
});
|
||||
|
||||
// ---- review fixes ------------------------------------------------------------------
|
||||
|
||||
test("a brake placed during the long-poll stops the join the poll's invite asks for", async () => {
|
||||
let open = true;
|
||||
const script = inviteSync(`@meta:${HS_NAME}`);
|
||||
const hs = await fakeHomeserver({ sync: since => { const out = script(since); if (since === "b0") open = false; return out; } }); // the brake lands while this poll is open
|
||||
const mouth = mouthFor(hs, { brake: async () => open });
|
||||
await mouth.syncOnce(); // baseline, released
|
||||
await mouth.syncOnce(); // the poll that carries the invite
|
||||
assert.deepEqual(joinsOf(hs), [], "no join");
|
||||
assert.equal(hs.seen.filter(r => r.path.endsWith("/account_data/m.direct") && r.method === "PUT").length, 0, "no m.direct write");
|
||||
});
|
||||
|
||||
test("no m.read receipt while braked: the ack proceeds, nothing is queued", async () => {
|
||||
let open = true;
|
||||
const hs = await fakeHomeserver({ sync: receiptScript });
|
||||
const lines = [];
|
||||
const state = memoryState();
|
||||
const mouth = await withMail(hs, { brake: async () => open, state, log: l => lines.push(l) });
|
||||
open = false;
|
||||
await mouth.ack(3);
|
||||
await until(() => lines.some(l => l.includes("receipts not posted")));
|
||||
assert.equal(mouth.pendingCount, 0);
|
||||
assert.equal(state.box.value.nextBatch, "b1", "the cursor committed");
|
||||
assert.equal(lines.filter(l => l.includes("receipts not posted")).length, 1, "one line");
|
||||
open = true;
|
||||
await mouth.syncOnce();
|
||||
await new Promise(r => setTimeout(r, 20));
|
||||
assert.deepEqual(hs.receipts(), [], "released again: nothing deferred is posted");
|
||||
});
|
||||
|
||||
test("the receipt path is percent-encoded on the wire", async () => {
|
||||
const ROOM = "!od d/room:gamecult.test", EVENT = "$ev/with+odd:chars";
|
||||
const hs = await fakeHomeserver({ sync: since => since === null ? { next_batch: "b0" } : { next_batch: "b1", account_data: { events: [{ type: "m.direct", content: { [`@meta:${HS_NAME}`]: [ROOM] } }] }, ...joined({ [ROOM]: { timeline: { events: [message(EVENT, `@meta:${HS_NAME}`, "x", 9)] } } }) } });
|
||||
const mouth = mouthFor(hs);
|
||||
await mouth.syncOnce(); await mouth.syncOnce();
|
||||
await mouth.ack(1);
|
||||
await until(() => hs.receipts().length >= 1);
|
||||
assert.equal(hs.receipts()[0].rawPath, `/_matrix/client/v3/rooms/${encodeURIComponent(ROOM)}/receipt/m.read/${encodeURIComponent(EVENT)}`);
|
||||
});
|
||||
|
||||
test("a receipt that never answers is abandoned after sendTimeoutMs", async () => {
|
||||
const hs = await fakeHomeserver({ sync: receiptScript, deferReceipts: new Promise(() => {}) });
|
||||
const lines = [];
|
||||
const mouth = await withMail(hs, { sendTimeoutMs: 40, log: l => lines.push(l) });
|
||||
await mouth.ack(3);
|
||||
await until(() => lines.length >= 1);
|
||||
await new Promise(r => setTimeout(r, 150));
|
||||
assert.ok(lines.some(l => l.startsWith("receipt in ") && l.includes("not posted")), "abandoned and logged");
|
||||
assert.equal(hs.receipts().length, 2, "the next room's receipt went out after the first was abandoned");
|
||||
});
|
||||
|
||||
test("a throwing logger on a failed receipt causes no unhandled rejection", async () => {
|
||||
const hs = await fakeHomeserver({ sync: receiptScript, failReceipts: true });
|
||||
const mouth = await withMail(hs, { log: () => { throw new Error("logger down"); } });
|
||||
const rejected = [];
|
||||
const note = e => rejected.push(e);
|
||||
process.on("unhandledRejection", note);
|
||||
try {
|
||||
await mouth.ack(3);
|
||||
await until(() => hs.receipts().length >= 2);
|
||||
await new Promise(r => setTimeout(r, 50));
|
||||
} finally { process.off("unhandledRejection", note); }
|
||||
assert.deepEqual(rejected, []);
|
||||
});
|
||||
|
||||
test("a full backlog still sets presence from the brake", async () => {
|
||||
let open = true;
|
||||
const hs = await fakeHomeserver({ sync: receiptScript });
|
||||
const mouth = mouthFor(hs, { brake: async () => open, maxPending: 3 });
|
||||
await mouth.syncOnce();
|
||||
assert.equal((await mouth.syncOnce()).length, 3);
|
||||
const syncsBefore = hs.seen.filter(r => r.path.endsWith("/sync")).length;
|
||||
open = false;
|
||||
assert.deepEqual(await mouth.syncOnce(), [], "backlog full: no /sync");
|
||||
assert.equal(hs.seen.filter(r => r.path.endsWith("/sync")).length, syncsBefore);
|
||||
assert.deepEqual(hs.presencePuts().map(r => r.body), [{ presence: "offline" }]);
|
||||
});
|
||||
|
||||
test("a brake placed during the first of two joins stops the second join and its m.direct write", async () => {
|
||||
const DM2 = "!newdm2:gamecult.test";
|
||||
const invites2 = { ...invite(`@meta:${HS_NAME}`), [DM2]: invite(`@meta:${HS_NAME}`)[DMROOM] };
|
||||
const hs = await fakeHomeserver({ sync: since => since === null ? { next_batch: "b0" } : since === "b0" ? { next_batch: "b1", rooms: { invite: invites2 } } : { next_batch: since } });
|
||||
let open = true;
|
||||
const fetch = async (url, init) => { const res = await hs.fetch(url, init); if (init?.method === "POST" && String(url).includes("/join/")) open = false; return res; }; // the brake lands during the first join
|
||||
const mouth = mouthFor(hs, { fetch, brake: async () => open });
|
||||
await mouth.syncOnce(); await mouth.syncOnce();
|
||||
assert.deepEqual(joinsOf(hs), [DMROOM], "only the join already started");
|
||||
assert.equal(hs.seen.filter(r => r.method === "PUT" && r.path.endsWith("/account_data/m.direct")).length, 1, "one m.direct, for the first join");
|
||||
open = true;
|
||||
await mouth.syncOnce();
|
||||
assert.deepEqual(joinsOf(hs), [DMROOM, DM2], "the second waits for the release");
|
||||
});
|
||||
|
|
|
|||
|
|
@ -6,7 +6,12 @@
|
|||
// its m.mentions user ids and whether the room is a DM. The /sync position
|
||||
// (next_batch) is persisted only when an ack covers it, so a restart before the
|
||||
// ack re-delivers instead of losing. Reading wakes nothing: the heartbeat polls
|
||||
// with a non-consuming chat_pending_request and decides who runs.
|
||||
// with a non-consuming chat_pending_request and decides who runs. An ack also posts this Persona's public
|
||||
// m.read receipt per room, at the newest event the ack covers (fire and forget: the ack never waits on it,
|
||||
// a failure is logged once and not retried, the room's next ack covers it). Her presence follows her
|
||||
// brake: every /sync says online while it is released and offline while it is held, and a round that
|
||||
// finds her braked makes one presence PUT to offline first. A brake placed mid long-poll shows at the next
|
||||
// round, within syncTimeoutMs plus one PUT; a stopped mouth reads offline after the homeserver's sync timeout.
|
||||
// Mouth: a send request becomes one PUT with txnId = request_id (a retry is idempotent), and a room
|
||||
// is joined, only while that Persona's brake is released. A v2 send carries the basis of her turn (the
|
||||
// newest room event she saw): inside the room gate, one step per room across every Persona, the mouth
|
||||
|
|
@ -119,6 +124,7 @@ export function createMouth({
|
|||
let batches = []; // { token, maxSeq }: positions awaiting an ack
|
||||
let committed = null; // the position that is persisted
|
||||
let started = false;
|
||||
let lastPresence = null; // 'online' | 'offline' | null: command-only, suppresses repeat presence PUTs
|
||||
const directContent = {}; // the account's m.direct content: user id -> room ids
|
||||
const invites = new Map(); // room id -> { inviter, direct }: invitations not yet answered
|
||||
const processed = new Set(); // event ids read since the position last advanced
|
||||
|
|
@ -159,7 +165,6 @@ export function createMouth({
|
|||
// released; one received while braked waits for the release. An invitation from any other server is
|
||||
// declined. A same-server invitation to a room that is not a DM is not ours to accept and is dropped.
|
||||
async function answerInvites() {
|
||||
let free = null;
|
||||
for (const [roomId, invite] of [...invites]) {
|
||||
try {
|
||||
const room = encodeURIComponent(roomId);
|
||||
|
|
@ -169,8 +174,7 @@ export function createMouth({
|
|||
} else if (!invite.direct) {
|
||||
invites.delete(roomId);
|
||||
} else {
|
||||
free ??= await released();
|
||||
if (!free) continue;
|
||||
if (!(await released())) continue; // read fresh before every join: a brake placed mid-pass stops the next one
|
||||
await matrix("POST", `/join/${room}`, {});
|
||||
invites.delete(roomId);
|
||||
directRooms.add(roomId);
|
||||
|
|
@ -280,14 +284,22 @@ export function createMouth({
|
|||
// One /sync round. Returns the deliveries it made.
|
||||
async function syncOnce() {
|
||||
if (!started || unjoined()) await start();
|
||||
const free = await released(); // the round's brake read for presence, taken before any early return
|
||||
if (!free && lastPresence !== "offline") {
|
||||
try {
|
||||
await matrix("PUT", `/presence/${encodeURIComponent(selfId)}/status`, { presence: "offline" }, API, AbortSignal.timeout(sendTimeoutMs));
|
||||
lastPresence = "offline";
|
||||
} catch (error) { say(`presence not set: ${error.message}`); }
|
||||
}
|
||||
if (pending.length >= maxPending) return []; // unanswered mail is not outrun: the position stays where it is
|
||||
const filter = encodeURIComponent(JSON.stringify({ room: { timeline: { limit: CONTEXT } } }));
|
||||
const since = cursor === null ? "" : `&since=${encodeURIComponent(cursor)}`;
|
||||
const reply = await matrix("GET", `/sync?timeout=${cursor === null ? 0 : syncTimeoutMs}&filter=${filter}${since}`);
|
||||
const reply = await matrix("GET", `/sync?timeout=${cursor === null ? 0 : syncTimeoutMs}&set_presence=${free ? "online" : "offline"}&filter=${filter}${since}`);
|
||||
if (free) lastPresence = "online";
|
||||
const baseline = cursor === null; // first ever sync: history is context, not mail
|
||||
for (const event of reply.account_data?.events ?? []) if (event.type === "m.direct") noteDirect(event.content);
|
||||
noteInvites(reply.rooms?.invite);
|
||||
await answerInvites();
|
||||
await answerInvites(); // each join re-reads the brake, as send does
|
||||
const made = [];
|
||||
const round = new Map();
|
||||
let truncated = false; // the pending cap was reached inside this round
|
||||
|
|
@ -346,10 +358,26 @@ export function createMouth({
|
|||
batches = batches.filter(b => b.maxSeq > throughSeq);
|
||||
}
|
||||
|
||||
// Her m.read receipt, display-only: public, unthreaded, at the newest event each room's ack covers.
|
||||
// Never awaited by the ack; a failure is one log line and nothing else. Posted only while her brake is
|
||||
// released, read fresh when each receipt would go out; a held brake drops it (the next ack covers it).
|
||||
async function readUpTo(newest) {
|
||||
const quiet = message => { try { say(message); } catch { /* a throwing logger must not reject this unawaited call */ } };
|
||||
for (const { roomId, eventId } of newest.values()) {
|
||||
if (!(await released())) { quiet("receipts not posted: brake held"); return; } // no receipt while braked; none is queued
|
||||
try {
|
||||
await matrix("POST", `/rooms/${encodeURIComponent(roomId)}/receipt/m.read/${encodeURIComponent(eventId)}`, {}, API, AbortSignal.timeout(sendTimeoutMs));
|
||||
} catch (error) { quiet(`receipt in ${roomId} not posted: ${error.message}`); }
|
||||
}
|
||||
}
|
||||
|
||||
async function ack(throughSeq) {
|
||||
const newest = new Map(); // roomId -> the highest-seq delivery this ack covers
|
||||
for (const d of pending) if (d.seq <= throughSeq && (newest.get(d.roomId)?.seq ?? 0) < d.seq) newest.set(d.roomId, d);
|
||||
pending = pending.filter(d => d.seq > throughSeq);
|
||||
images?.acked(throughSeq);
|
||||
await commit(pending.length ? pending[0].seq - 1 : seqCounter);
|
||||
readUpTo(newest);
|
||||
}
|
||||
|
||||
function pendingAfter(afterSeq) {
|
||||
|
|
|
|||
Loading…
Reference in a new issue