Matrix mouth: m.read follows a chat_read report, presence follows the turn #35
5 changed files with 353 additions and 143 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. 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.
|
||||
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 posts nothing to Matrix. That Persona's public `m.read` receipt follows `gamecult.persona.chat_read.v1` (`[persona, room_alias, event_id, at_unix_ms]`), which Huginn sends when a delivery has gone into the model's input: the mouth posts `m.read` at that event id only, for an event it delivered to that Persona in that room (to the delivered room id, whatever alias the report named), never behind the last receipt that landed there, one report per room at a time, and only while her brake is released (re-read after the presence write; fire and forget on the endpoint: a failure is logged once and not retried; pending polls and acks post none). Her Matrix presence follows a turn and her brake: `online` while a read report, a send or an ack came within `presenceGraceMs` (5 minutes), `unavailable` after that, `offline` while the brake is held. Each `/sync` carries the state as `set_presence`, and a change is also `PUT /presence/<user>/status` at once (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.
|
||||
|
||||
|
|
|
|||
|
|
@ -134,10 +134,11 @@ test("two clients served at once each get their own answer on their own session"
|
|||
});
|
||||
|
||||
test("the RUDP endpoint serves pending, ack and send over CultNet, and answers a held brake with a Brake refusal", async () => {
|
||||
const calls = { acks: [], sends: [] };
|
||||
const calls = { acks: [], sends: [], reads: [] };
|
||||
const mouth = {
|
||||
pendingAfter: after => (after < 7 ? [delivery] : []),
|
||||
ack: async through => { calls.acks.push(through); },
|
||||
read: async request => { calls.reads.push(request); return null; },
|
||||
send: async request => { calls.sends.push(request); return request.text === "muted" ? { requestId: request.requestId, eventId: null, refusal: "Brake" } : { requestId: request.requestId, eventId: "$sent", refusal: null }; },
|
||||
};
|
||||
const key = "connection-key-for-test";
|
||||
|
|
@ -183,6 +184,16 @@ test("the RUDP endpoint serves pending, ack and send over CultNet, and answers a
|
|||
assert.deepEqual([document.schemaId, docs.parseRefusal(msgpack.decode(document.payload))], [docs.kinds.refusal, { persona: "other", refusal: "UnknownPersona" }], "an ack for an unlisted Persona is refused, typed, and not applied");
|
||||
assert.deepEqual(calls.acks, [7]);
|
||||
|
||||
reply = nextReply();
|
||||
put(docs.kinds.read, "other:#operator:gamecult.org", docs.readTuple({ persona: "other", roomAlias: "#operator:gamecult.org", eventId: "$e1", atUnixMs: 5 }));
|
||||
document = await reply;
|
||||
assert.deepEqual([document.schemaId, docs.parseRefusal(msgpack.decode(document.payload))], [docs.kinds.refusal, { persona: "other", refusal: "UnknownPersona" }], "a read report for an unlisted Persona is refused, typed, and not applied");
|
||||
assert.deepEqual(calls.reads, []);
|
||||
|
||||
put(docs.kinds.read, "eureka:#operator:gamecult.org", docs.readTuple({ persona: "eureka", roomAlias: "#operator:gamecult.org", eventId: "$e1", atUnixMs: 5 }));
|
||||
for (let i = 0; i < 100 && calls.reads.length < 1; i++) await new Promise(r => setTimeout(r, 20));
|
||||
assert.deepEqual(calls.reads, [{ persona: "eureka", roomAlias: "#operator:gamecult.org", eventId: "$e1", atUnixMs: 5 }], "her own report reaches her mouth with its persona");
|
||||
|
||||
reply = nextReply();
|
||||
put(docs.kinds.pendingRequest, "other", docs.pendingRequestTuple({ persona: "other", afterSeq: 0 }));
|
||||
document = await reply;
|
||||
|
|
@ -373,7 +384,7 @@ test("the RUDP endpoint answers a v1 send with the three-field receipt and a v2
|
|||
}
|
||||
});
|
||||
|
||||
test("over the RUDP endpoint: pending posts no receipt, and an ack never holds up the same peer's next request", async () => {
|
||||
test("over the RUDP endpoint: pending and an ack post no receipt, a read report posts one, and it never holds up the same peer's next request", async () => {
|
||||
const receipts = [];
|
||||
let release;
|
||||
const held = new Promise(r => { release = r; });
|
||||
|
|
@ -412,11 +423,14 @@ test("over the RUDP endpoint: pending posts no receipt, and an ack never holds u
|
|||
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.read, "eureka:#r:t", docs.readTuple({ persona: "eureka", roomAlias: "#r:t", eventId: "$m1", atUnixMs: 6 }));
|
||||
await waitFor(() => receipts.length === 1, "the read report posted her receipt");
|
||||
assert.ok(receipts[0].endsWith("/receipt/m.read/%24m1"), "at the reported event");
|
||||
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");
|
||||
assert.equal(receipts.length, 1, "the ack posted no receipt of its own");
|
||||
} finally {
|
||||
release();
|
||||
server.close();
|
||||
|
|
|
|||
|
|
@ -18,8 +18,8 @@ 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";
|
||||
// The one write a round makes by design: her presence state (any state; the brake and the grace decide which).
|
||||
const offlinePresence = r => r.method === "PUT" && r.path.includes("/presence/");
|
||||
const tick = () => new Promise(r => setImmediate(r));
|
||||
|
||||
// A fake Matrix client-server API: just the endpoints the mouth uses.
|
||||
|
|
@ -953,7 +953,8 @@ test("image_delivery: a store that cannot write leaves the line and a nil image,
|
|||
} finally { await hs.close(); }
|
||||
});
|
||||
|
||||
// Read receipts and presence: her ack posts her m.read; her brake sets her presence.
|
||||
// Read receipts and presence: Huginn's chat_read report posts her m.read; an ack posts nothing; her presence follows a
|
||||
// turn and her brake.
|
||||
const A = "!swarm:gamecult.test", B = "!operator:gamecult.test";
|
||||
const receiptScript = since => since === null ? { next_batch: "b0", ...joined({}) } : since === "b0" ? {
|
||||
next_batch: "b1",
|
||||
|
|
@ -970,78 +971,240 @@ const withMail = async (hs, overrides) => {
|
|||
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]);
|
||||
const presencesOf = hs => hs.presencePuts().map(r => r.body.presence);
|
||||
const syncsOf = hs => hs.seen.filter(r => r.path.endsWith("/sync")).map(r => r.query.get("set_presence"));
|
||||
const reportRead = (mouth, roomAlias, eventId, persona = "eureka") => mouth.read({ persona, roomAlias, eventId, atUnixMs: 1 });
|
||||
|
||||
test("an ack posts one m.read per room at the newest covered event, as that Persona", async () => {
|
||||
test("delivery, pending and an ack post no receipt: she has not read what the model has not seen", 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");
|
||||
for (const after of [0, 1, 2, 3]) mouth.pendingAfter(after);
|
||||
await mouth.ack(3);
|
||||
await until(() => hs.receipts().length >= 2);
|
||||
await new Promise(r => setTimeout(r, 30));
|
||||
assert.deepEqual(hs.receipts(), [], "mail handed to her queue, polled and acked: still no m.read");
|
||||
assert.equal(mouth.pendingCount, 0);
|
||||
});
|
||||
|
||||
test("a read report posts m.read at exactly that event id, as that Persona", async () => {
|
||||
const hs = await fakeHomeserver({ sync: receiptScript });
|
||||
const mouth = await withMail(hs);
|
||||
assert.equal(await reportRead(mouth, A, "$a1"), null);
|
||||
assert.deepEqual(receiptsOf(hs), [[A, "$a1", `Bearer ${TOKEN}`, {}]], "the reported event, not the newest in the room");
|
||||
assert.equal(await reportRead(mouth, B, "$b1"), null);
|
||||
assert.deepEqual(receiptsOf(hs).slice(1), [[B, "$b1", `Bearer ${TOKEN}`, {}]]);
|
||||
assert.equal(hs.receipts().length, 2);
|
||||
});
|
||||
|
||||
test("pending posts no receipt", async () => {
|
||||
test("a read report for an event the mouth never delivered, or in another room, posts nothing", 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.equal(await reportRead(mouth, A, "$made-up"), "UnknownEvent");
|
||||
assert.equal(await reportRead(mouth, B, "$a1"), "RoomMismatch");
|
||||
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;
|
||||
test("a read report never moves her receipt backward or repeats it in a room", async () => {
|
||||
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");
|
||||
const mouth = await withMail(hs);
|
||||
assert.equal(await reportRead(mouth, A, "$a2"), null);
|
||||
assert.equal(await reportRead(mouth, A, "$a1"), "Behind");
|
||||
assert.equal(await reportRead(mouth, A, "$a2"), "Behind");
|
||||
assert.equal(hs.receipts().length, 1);
|
||||
});
|
||||
|
||||
test("a read report posts nothing while braked, and nothing is queued for the release", async () => {
|
||||
let open = false;
|
||||
const hs = await fakeHomeserver({ sync: receiptScript });
|
||||
const lines = [];
|
||||
const mouth = await withMail(hs, { brake: async () => open, log: l => lines.push(l) });
|
||||
assert.equal(await reportRead(mouth, A, "$a1"), "Brake");
|
||||
assert.ok(lines.some(l => l.includes("receipt not posted: brake held")));
|
||||
open = true;
|
||||
await mouth.syncOnce();
|
||||
assert.equal(syncs().at(-1), "online");
|
||||
open = false;
|
||||
await new Promise(r => setTimeout(r, 20));
|
||||
assert.deepEqual(hs.receipts(), []);
|
||||
assert.equal(await reportRead(mouth, A, "$a1"), null, "a fresh report after the release posts");
|
||||
});
|
||||
|
||||
test("a read report names her persona: another persona's report posts nothing", async () => {
|
||||
const hs = await fakeHomeserver({ sync: receiptScript });
|
||||
const mouth = await withMail(hs);
|
||||
assert.equal(await reportRead(mouth, A, "$a1", "someone-else"), "PersonaMismatch");
|
||||
assert.deepEqual(hs.receipts(), []);
|
||||
});
|
||||
|
||||
test("the receipt goes to the delivered room id, not to the alias the report named", async () => {
|
||||
const hs = await fakeHomeserver({ sync: receiptScript });
|
||||
const mouth = await withMail(hs);
|
||||
assert.equal(await reportRead(mouth, `#swarm:${HS_NAME}`, "$a1"), null);
|
||||
assert.deepEqual(receiptsOf(hs), [[A, "$a1", `Bearer ${TOKEN}`, {}]]);
|
||||
});
|
||||
|
||||
test("concurrent reports for one room are judged in arrival order: the newer lands and the older is behind", async () => {
|
||||
const hs = await fakeHomeserver({ sync: receiptScript });
|
||||
const mouth = await withMail(hs);
|
||||
const [newer, older] = await Promise.all([reportRead(mouth, A, "$a2"), reportRead(mouth, A, "$a1")]);
|
||||
assert.deepEqual([newer, older], [null, "Behind"]);
|
||||
assert.deepEqual(receiptsOf(hs).map(r => r[1]), ["$a2"], "never a1 after a2");
|
||||
const [first, second] = await Promise.all([reportRead(mouth, B, "$b1"), reportRead(mouth, B, "$b1")]);
|
||||
assert.deepEqual([first, second], [null, "Behind"], "the same event twice posts once");
|
||||
});
|
||||
|
||||
test("a failed receipt leaves the last landed receipt in place: an older event is still behind", async () => {
|
||||
let receiptCalls = 0;
|
||||
const hs = await fakeHomeserver({ sync: receiptScript });
|
||||
const fetch = (url, init) => String(url).includes("/receipt/") && ++receiptCalls === 2 ? Promise.resolve(new Response("{}", { status: 500 })) : hs.fetch(url, init);
|
||||
const mouth = await withMail(hs, { fetch });
|
||||
assert.equal(await reportRead(mouth, A, "$a1"), null);
|
||||
assert.equal(await reportRead(mouth, A, "$a2"), "Failed");
|
||||
assert.equal(await reportRead(mouth, A, "$a1"), "Behind", "a1 landed; the failure of a2 does not reopen the room to a backward post");
|
||||
assert.equal(await reportRead(mouth, A, "$a2"), null, "a2 itself can still land");
|
||||
assert.deepEqual(receiptsOf(hs).map(r => r[1]), ["$a1", "$a2"]);
|
||||
});
|
||||
|
||||
test("a brake placed after the first brake read stops the receipt and no present state is PUT while braked", async () => {
|
||||
let used = 0, allowed = Infinity;
|
||||
const hs = await fakeHomeserver({ sync: receiptScript });
|
||||
const mouth = await withMail(hs, { brake: async () => used++ < allowed });
|
||||
used = 0; allowed = 1; // read() sees it released once; every read after that finds it held
|
||||
assert.equal(await reportRead(mouth, A, "$a1"), "Brake");
|
||||
assert.deepEqual(hs.receipts(), [], "the receipt re-reads the brake after the presence write");
|
||||
assert.ok(!presencesOf(hs).includes("online"), "online is never PUT on a stale brake read");
|
||||
assert.equal(presencesOf(hs).at(-1), "offline");
|
||||
});
|
||||
test("a slow presence write after a send holds no other send in the room", async () => {
|
||||
const hs = await fakeHomeserver({ sync: receiptScript });
|
||||
const fetch = (url, init) => String(url).includes("/presence/") && init?.method === "PUT"
|
||||
? new Promise((_, reject) => init.signal.addEventListener("abort", () => reject(new Error("aborted")))) : hs.fetch(url, init);
|
||||
const mouth = await withMail(hs, { fetch, sendTimeoutMs: 400 });
|
||||
const send = id => mouth.send({ requestId: id, persona: "eureka", roomAlias: `#swarm:${HS_NAME}`, text: id });
|
||||
void send("s1"); // its presence write hangs
|
||||
const second = await Promise.race([send("s2").then(r => r.refusal), new Promise(r => setTimeout(() => r("held behind the first send's presence write"), 150))]);
|
||||
assert.equal(second, null);
|
||||
});
|
||||
|
||||
test("the grace ends exactly at its boundary: one millisecond inside is online, the boundary is unavailable", async () => {
|
||||
let t = 9_000_000;
|
||||
const hs = await fakeHomeserver({ sync: receiptScript });
|
||||
const mouth = await withMail(hs, { now: () => t, presenceGraceMs: 60_000 });
|
||||
await mouth.ack(3); // touched at t
|
||||
t += 59_999;
|
||||
await mouth.syncOnce();
|
||||
assert.equal(hs.presencePuts().length, 2, "a second transition PUTs once more");
|
||||
assert.equal(syncsOf(hs).at(-1), "online");
|
||||
t += 1;
|
||||
await mouth.syncOnce();
|
||||
assert.equal(syncsOf(hs).at(-1), "unavailable");
|
||||
});
|
||||
|
||||
test("the state a poll carries is fixed before the poll: a grace that passes while it is open is PUT next round", async () => {
|
||||
let t = 3_000_000;
|
||||
const hs = await fakeHomeserver({ sync: receiptScript });
|
||||
const fetch = async (url, init) => { const res = await hs.fetch(url, init); if (String(url).includes("/sync")) t += 100_000; return res; };
|
||||
const mouth = mouthFor(hs, { now: () => t, presenceGraceMs: 60_000, fetch });
|
||||
await mouth.ack(0); // a turn: online
|
||||
await mouth.syncOnce();
|
||||
assert.equal(syncsOf(hs).at(-1), "online");
|
||||
await mouth.syncOnce();
|
||||
assert.deepEqual(presencesOf(hs), ["online", "unavailable"], "the homeserver was told online by the poll; the next round must tell it unavailable");
|
||||
});
|
||||
|
||||
test("a failed receipt is logged once, not retried, and the next report for the room posts", async () => {
|
||||
const hs = await fakeHomeserver({ sync: receiptScript, failReceipts: true });
|
||||
const lines = [];
|
||||
const mouth = await withMail(hs, { log: l => lines.push(l) });
|
||||
assert.equal(await reportRead(mouth, A, "$a1"), "Failed");
|
||||
const failures = lines.filter(l => l.startsWith("receipt in "));
|
||||
assert.equal(failures.length, 1);
|
||||
assert.ok(failures[0].includes("not posted") && !failures[0].includes(TOKEN));
|
||||
await new Promise(r => setTimeout(r, 20));
|
||||
assert.equal(hs.receipts().length, 1, "no retry");
|
||||
assert.equal(await reportRead(mouth, A, "$a1"), "Failed", "the claim was released: the room is not stuck behind a failed post");
|
||||
});
|
||||
|
||||
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) });
|
||||
assert.equal(await reportRead(mouth, A, "$a1"), "Failed");
|
||||
assert.ok(lines.some(l => l.startsWith("receipt in ") && l.includes("not posted")));
|
||||
assert.equal(await reportRead(mouth, B, "$b1"), "Failed", "the next room's receipt went out after the first was abandoned");
|
||||
assert.equal(hs.receipts().length, 2);
|
||||
});
|
||||
|
||||
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 reportRead(mouth, A, "$a1").catch(() => {});
|
||||
await new Promise(r => setTimeout(r, 50));
|
||||
} finally { process.off("unhandledRejection", note); }
|
||||
assert.deepEqual(rejected, []);
|
||||
});
|
||||
|
||||
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();
|
||||
assert.equal(await reportRead(mouth, ROOM, EVENT), null);
|
||||
assert.equal(hs.receipts()[0].rawPath, `/_matrix/client/v3/rooms/${encodeURIComponent(ROOM)}/receipt/m.read/${encodeURIComponent(EVENT)}`);
|
||||
});
|
||||
|
||||
test("presence: unavailable while released and idle, online through a turn, unavailable after the grace", async () => {
|
||||
let t = 1_000_000;
|
||||
const hs = await fakeHomeserver({ sync: receiptScript });
|
||||
const mouth = await withMail(hs, { now: () => t, presenceGraceMs: 60_000 });
|
||||
assert.deepEqual(syncsOf(hs), ["unavailable", "unavailable"], "mail alone does not make her present");
|
||||
assert.deepEqual(presencesOf(hs), ["unavailable"], "one PUT, not one per round");
|
||||
await reportRead(mouth, A, "$a1");
|
||||
assert.deepEqual(presencesOf(hs), ["unavailable", "online"], "the report puts her online at once, not at the next poll");
|
||||
t += 59_000;
|
||||
await mouth.syncOnce();
|
||||
assert.equal(syncsOf(hs).at(-1), "online", "inside the grace");
|
||||
t += 2_000;
|
||||
await mouth.syncOnce();
|
||||
assert.equal(syncsOf(hs).at(-1), "unavailable", "past the grace");
|
||||
assert.deepEqual(presencesOf(hs), ["unavailable", "online", "unavailable"]);
|
||||
});
|
||||
|
||||
test("a send or an ack within the grace keeps her online", async () => {
|
||||
let t = 5_000_000;
|
||||
const hs = await fakeHomeserver({ sync: receiptScript });
|
||||
const mouth = await withMail(hs, { now: () => t, presenceGraceMs: 60_000 });
|
||||
await mouth.ack(3);
|
||||
await mouth.syncOnce();
|
||||
assert.equal(syncsOf(hs).at(-1), "online", "an ack is a sign of a turn");
|
||||
t += 61_000;
|
||||
await mouth.syncOnce();
|
||||
assert.equal(syncsOf(hs).at(-1), "unavailable");
|
||||
const sent = await mouth.send({ requestId: "p1", persona: "eureka", roomAlias: `#swarm:${HS_NAME}`, text: "hi" });
|
||||
assert.equal(sent.refusal, null);
|
||||
await mouth.syncOnce();
|
||||
assert.equal(syncsOf(hs).at(-1), "online", "a send is a sign of a turn");
|
||||
});
|
||||
|
||||
test("presence under the brake is offline whatever the turn: one PUT on the transition, and a report cannot lift it", async () => {
|
||||
let open = true;
|
||||
const hs = await fakeHomeserver({ sync: receiptScript });
|
||||
const mouth = await withMail(hs, { brake: async () => open });
|
||||
await reportRead(mouth, A, "$a1");
|
||||
await mouth.syncOnce();
|
||||
assert.equal(syncsOf(hs).at(-1), "online");
|
||||
open = false;
|
||||
await mouth.syncOnce(); await mouth.syncOnce(); await mouth.syncOnce();
|
||||
assert.deepEqual(syncsOf(hs).slice(-3), ["offline", "offline", "offline"]);
|
||||
assert.equal(presencesOf(hs).at(-1), "offline");
|
||||
assert.equal(presencesOf(hs).filter(p => p === "offline").length, 1, "exactly one offline PUT");
|
||||
assert.equal(hs.presencePuts().at(-1).rawPath, `/_matrix/client/v3/presence/${encodeURIComponent(SELF)}/status`, "the user id is percent-encoded on the wire");
|
||||
assert.equal(await reportRead(mouth, B, "$b1"), "Brake");
|
||||
await mouth.syncOnce();
|
||||
assert.equal(syncsOf(hs).at(-1), "offline");
|
||||
open = true;
|
||||
await mouth.syncOnce();
|
||||
assert.equal(syncsOf(hs).at(-1), "online", "still inside the grace of the earlier report");
|
||||
});
|
||||
|
||||
test("the first braked round of a fresh mouth PUTs offline once; a failed PUT is tried again next round", async () => {
|
||||
|
|
@ -1063,15 +1226,14 @@ test("two Personas on one homeserver: receipts and presence carry only her own b
|
|||
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");
|
||||
await reportRead(a, A, "$a1", "pa");
|
||||
assert.equal(await reportRead(b, A, "$a1", "pb"), "Brake");
|
||||
assert.ok(hs.receipts().every(r => r.auth === "Bearer tokA"), "only A read");
|
||||
bOpen = true; // her receipts go out only while released
|
||||
await b.ack(3);
|
||||
await until(() => hs.receipts().length >= 4);
|
||||
assert.equal(await reportRead(b, A, "$a1", "pb"), null);
|
||||
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"));
|
||||
assert.ok(hs.presencePuts().some(r => r.path.endsWith(`/presence/${BOB}/status`) && r.auth === "Bearer tokB"));
|
||||
assert.ok(hs.presencePuts().filter(r => r.path.endsWith(`/presence/${BOB}/status`)).every(r => r.auth === "Bearer tokB"));
|
||||
});
|
||||
|
||||
// ---- review fixes ------------------------------------------------------------------
|
||||
|
|
@ -1087,59 +1249,6 @@ test("a brake placed during the long-poll stops the join the poll's invite asks
|
|||
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 });
|
||||
|
|
@ -1150,7 +1259,7 @@ test("a full backlog still sets presence from the brake", async () => {
|
|||
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" }]);
|
||||
assert.deepEqual(hs.presencePuts().map(r => r.body), [{ presence: "unavailable" }, { presence: "offline" }]);
|
||||
});
|
||||
|
||||
test("a brake placed during the first of two joins stops the second join and its m.direct write", async () => {
|
||||
|
|
@ -1167,3 +1276,30 @@ test("a brake placed during the first of two joins stops the second join and its
|
|||
await mouth.syncOnce();
|
||||
assert.deepEqual(joinsOf(hs), [DMROOM, DM2], "the second waits for the release");
|
||||
});
|
||||
|
||||
test("chat_read round-trips and refuses a malformed tuple", () => {
|
||||
const tuple = docs.readTuple({ persona: "eureka", roomAlias: "#swarm:gamecult.test", eventId: "$e1", atUnixMs: 12 });
|
||||
assert.deepEqual(tuple, ["eureka", "#swarm:gamecult.test", "$e1", 12]);
|
||||
assert.deepEqual(docs.parseRead(tuple), { persona: "eureka", roomAlias: "#swarm:gamecult.test", eventId: "$e1", atUnixMs: 12 });
|
||||
assert.throws(() => docs.parseRead(["eureka", "#r", "", 1]), /event_id/);
|
||||
assert.throws(() => docs.parseRead(["eureka", "#r", "$e", -1]), /at_unix_ms/);
|
||||
assert.throws(() => docs.parseRead(["eureka", "#r", "$e"]));
|
||||
});
|
||||
|
||||
test("a read report that throws does not wedge the room: the next report still posts and nothing is left unhandled", async () => {
|
||||
const hs = await fakeHomeserver({ sync: receiptScript });
|
||||
let boom = false;
|
||||
const fetch = (url, init) => boom && String(url).includes("/presence/") ? Promise.reject(new Error("net")) : hs.fetch(url, init);
|
||||
const mouth = await withMail(hs, { fetch, log: line => { if (boom && line.startsWith("presence not set")) throw new Error("logger down"); } }); // the presence failure and a throwing logger make run() reject
|
||||
const unhandled = [], note = error => unhandled.push(String(error));
|
||||
process.on("unhandledRejection", note);
|
||||
try {
|
||||
boom = true;
|
||||
await assert.rejects(reportRead(mouth, A, "$a1"), /logger down/);
|
||||
boom = false;
|
||||
assert.equal(await reportRead(mouth, A, "$a2"), null, "the room's next report is judged and posted");
|
||||
await new Promise(resolve => setTimeout(resolve, 50));
|
||||
} finally { process.off("unhandledRejection", note); }
|
||||
assert.deepEqual(unhandled, [], "no rejection escapes the tail's cleanup");
|
||||
assert.equal(hs.receipts().length, 1);
|
||||
});
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ export const kinds = Object.freeze({
|
|||
pendingRequest: "gamecult.persona.chat_pending_request.v1",
|
||||
pending: "gamecult.persona.chat_pending.v2",
|
||||
ack: "gamecult.persona.chat_ack.v1",
|
||||
read: "gamecult.persona.chat_read.v1",
|
||||
sendRequest: "gamecult.persona.chat_send_request.v1",
|
||||
sendReceipt: "gamecult.persona.chat_send_receipt.v1",
|
||||
sendRequestV2: "gamecult.persona.chat_send_request.v2",
|
||||
|
|
@ -52,6 +53,16 @@ export function parseAck(t) {
|
|||
return { persona: text(persona, "persona"), throughSeq: seq(throughSeq, "through_seq") };
|
||||
}
|
||||
|
||||
// Huginn's report that a delivery has gone into the model's input: she has read through this event of this room.
|
||||
// at_unix_ms is when the model took it in; the mouth does not trust it for anything but the log.
|
||||
export function readTuple({ persona, roomAlias, eventId, atUnixMs }) {
|
||||
return [text(persona, "persona"), text(roomAlias, "room_alias"), text(eventId, "event_id"), seq(atUnixMs, "at_unix_ms")];
|
||||
}
|
||||
export function parseRead(t) {
|
||||
const [persona, roomAlias, eventId, atUnixMs] = tuple(t, 4, "chat_read");
|
||||
return { persona: text(persona, "persona"), roomAlias: text(roomAlias, "room_alias"), eventId: text(eventId, "event_id"), atUnixMs: seq(atUnixMs, "at_unix_ms") };
|
||||
}
|
||||
|
||||
export function sendRequestTuple({ requestId, persona, roomAlias, text: body }) {
|
||||
return [text(requestId, "request_id"), text(persona, "persona"), text(roomAlias, "room_alias"), String(body)];
|
||||
}
|
||||
|
|
|
|||
|
|
@ -6,12 +6,15 @@
|
|||
// 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. 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.
|
||||
// with a non-consuming chat_pending_request and decides who runs. An ack posts nothing to Matrix. Her public
|
||||
// m.read receipt follows a chat_read document from Huginn, sent when a delivery has gone into the model's
|
||||
// input: the mouth posts m.read at that event id only, for an event it delivered to her, never behind one
|
||||
// already posted in that room (fire and forget on the endpoint: a failure is logged once, not retried).
|
||||
// Her presence: online while a turn is going on (a read report, a send or an ack within presenceGraceMs),
|
||||
// unavailable once the grace has passed, offline while her brake is held. Every /sync carries the state
|
||||
// and a change is also PUT at once, so a transition does not wait out the long poll. 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
|
||||
|
|
@ -53,6 +56,8 @@ const V1 = "/_matrix/client/v1"; // /relations lives only under v1
|
|||
const MAX_PENDING = 500;
|
||||
const SLUG = /^[A-Za-z0-9_-]{1,64}$/;
|
||||
const SEND_TIMEOUT_MS = 10000; // the head read and the PUT of one gated send
|
||||
const PRESENCE_GRACE_MS = 5 * 60 * 1000; // a turn's last sign of life keeps her online this long
|
||||
const DELIVERED_INDEX = 4000; // delivered event ids remembered so a read report can be checked against them
|
||||
const HEAD_PAGE = 50; // the room head a gated send reads; also the most ids a RoomMoved receipt names
|
||||
|
||||
// A message line: a text event with a sender, not an edit.
|
||||
|
|
@ -107,6 +112,7 @@ export function createRoomGate() {
|
|||
export function createMouth({
|
||||
persona, homeserver, serverName, rooms, token, fetch: httpFetch = globalThis.fetch, state, brake,
|
||||
now = Date.now, log = () => {}, syncTimeoutMs = 30000, maxPending = MAX_PENDING, gate = createRoomGate(), sendTimeoutMs = SEND_TIMEOUT_MS,
|
||||
presenceGraceMs = PRESENCE_GRACE_MS,
|
||||
images = null, // this Persona's image store (createImageStore), or null: images stay the typed text line
|
||||
}) {
|
||||
if (!persona || !homeserver || !token || !rooms?.length) throw new Error("persona, homeserver, rooms and token are required");
|
||||
|
|
@ -124,7 +130,11 @@ 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
|
||||
let lastPresence = null; // 'online' | 'unavailable' | 'offline' | null: command-only, suppresses repeat presence PUTs
|
||||
let activeUntil = 0; // her last sign of life in a turn, plus the grace
|
||||
const delivered = new Map(); // event id -> { roomId, roomAlias, seq }: what the mouth handed her, newest last, bounded
|
||||
const readThrough = new Map(); // room id -> { seq, eventId } of the last receipt that landed
|
||||
const readTails = new Map(); // room id -> the read report in flight: reports for one room run one at a time
|
||||
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
|
||||
|
|
@ -281,21 +291,31 @@ export function createMouth({
|
|||
}
|
||||
}
|
||||
|
||||
// Her presence: the brake decides offline; otherwise a turn in the last presenceGraceMs decides online.
|
||||
const presenceFor = free => !free ? "offline" : now() < activeUntil ? "online" : "unavailable";
|
||||
const touch = () => { activeUntil = now() + presenceGraceMs; };
|
||||
async function setPresence(free) {
|
||||
let wanted = presenceFor(free);
|
||||
if (wanted === lastPresence) return;
|
||||
if (wanted !== "offline" && !(await released())) wanted = "offline"; // never PUT a present state on a stale brake read
|
||||
if (wanted === lastPresence) return;
|
||||
try {
|
||||
await matrix("PUT", `/presence/${encodeURIComponent(selfId)}/status`, { presence: wanted }, API, AbortSignal.timeout(sendTimeoutMs));
|
||||
lastPresence = wanted;
|
||||
} catch (error) { say(`presence not set: ${error.message}`); }
|
||||
}
|
||||
|
||||
// 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}`); }
|
||||
}
|
||||
await setPresence(free);
|
||||
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}&set_presence=${free ? "online" : "offline"}&filter=${filter}${since}`);
|
||||
if (free) lastPresence = "online";
|
||||
const stateSent = presenceFor(free); // the state this poll carries, fixed before the poll: the grace may pass while it is open
|
||||
const reply = await matrix("GET", `/sync?timeout=${cursor === null ? 0 : syncTimeoutMs}&set_presence=${stateSent}&filter=${filter}${since}`);
|
||||
if (free) lastPresence = stateSent; // a braked round's offline is only believed once its own PUT lands
|
||||
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);
|
||||
|
|
@ -336,6 +356,10 @@ export function createMouth({
|
|||
}
|
||||
}
|
||||
pending.push(...made);
|
||||
for (const d of made) {
|
||||
delivered.set(d.eventId, { roomId: d.roomId, roomAlias: d.roomAlias, seq: d.seq });
|
||||
if (delivered.size > DELIVERED_INDEX) delivered.delete(delivered.keys().next().value);
|
||||
}
|
||||
if (!truncated) { // a truncated round leaves the position where it was: the rest stays on the server
|
||||
cursor = reply.next_batch;
|
||||
batches.push({ token: reply.next_batch, maxSeq: seqCounter });
|
||||
|
|
@ -358,26 +382,44 @@ 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
|
||||
// Her m.read receipt, display-only: public, unthreaded, at the event Huginn reports she has taken in. Only for her own
|
||||
// persona, an event this mouth delivered to her, in the room it was delivered for (the receipt goes to the delivered
|
||||
// room id, never to the reported alias), and only forward of the last receipt that landed in that room. Reports for
|
||||
// one room run one at a time, so the order they arrive in is the order they are judged in. Posted only while her
|
||||
// brake is released, read fresh before the presence write and again before the POST. Returns the reason nothing
|
||||
// was posted, or null.
|
||||
function read({ persona: asked, roomAlias, eventId }) {
|
||||
const quiet = message => { try { say(message); } catch { /* a throwing logger must not reject a caller that does not await */ } };
|
||||
if (asked !== persona) return Promise.resolve("PersonaMismatch");
|
||||
const known = delivered.get(eventId);
|
||||
if (!known) { quiet("read report names an event this mouth never delivered"); return Promise.resolve("UnknownEvent"); }
|
||||
if (roomAlias !== known.roomAlias && roomAlias !== known.roomId) { quiet("read report names another room than the delivery"); return Promise.resolve("RoomMismatch"); }
|
||||
const run = async () => {
|
||||
if ((readThrough.get(known.roomId)?.seq ?? 0) >= known.seq) return "Behind";
|
||||
if (!(await released())) { quiet("receipt not posted: brake held"); return "Brake"; } // none is queued
|
||||
touch();
|
||||
await setPresence(true);
|
||||
if (!(await released())) { quiet("receipt not posted: brake held"); return "Brake"; }
|
||||
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}`); }
|
||||
}
|
||||
await matrix("POST", `/rooms/${encodeURIComponent(known.roomId)}/receipt/m.read/${encodeURIComponent(eventId)}`, {}, API, AbortSignal.timeout(sendTimeoutMs));
|
||||
} catch (error) {
|
||||
quiet(`receipt in ${known.roomId} not posted: ${error.message}`);
|
||||
return "Failed"; // the last receipt that landed stays what it was: a later report for an older event is still Behind
|
||||
}
|
||||
readThrough.set(known.roomId, { seq: known.seq, eventId });
|
||||
return null;
|
||||
};
|
||||
const tail = (readTails.get(known.roomId) ?? Promise.resolve()).then(run, run); // a report that threw must not wedge the room's later ones
|
||||
readTails.set(known.roomId, tail);
|
||||
tail.then(() => { if (readTails.get(known.roomId) === tail) readTails.delete(known.roomId); }, () => { if (readTails.get(known.roomId) === tail) readTails.delete(known.roomId); });
|
||||
return tail;
|
||||
}
|
||||
|
||||
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);
|
||||
touch();
|
||||
}
|
||||
|
||||
function pendingAfter(afterSeq) {
|
||||
|
|
@ -407,7 +449,7 @@ export function createMouth({
|
|||
if (!started || unjoined()) await start();
|
||||
const roomId = joinedAliases.has(roomAlias) ? roomIdByAlias.get(roomAlias) : directRooms.has(roomAlias) ? roomAlias : null;
|
||||
if (!roomId) return refuse("UnknownRoom");
|
||||
return await gate.run(roomId, async () => {
|
||||
const result = await gate.run(roomId, async () => {
|
||||
if (basis) {
|
||||
const newer = await newerThan(roomId, basis);
|
||||
if (newer.length > 0) return refuse("RoomMoved", newer);
|
||||
|
|
@ -416,6 +458,8 @@ export function createMouth({
|
|||
const sent = await matrix("PUT", `/rooms/${encodeURIComponent(roomId)}/send/m.room.message/${encodeURIComponent(requestId)}`, { msgtype: "m.text", body: text }, API, AbortSignal.timeout(sendTimeoutMs));
|
||||
return { requestId, eventId: sent.event_id, refusal: null, newer: [] };
|
||||
});
|
||||
if (result.refusal === null) { touch(); setPresence(true).catch(() => {}); } // outside the room gate and not awaited: a slow presence write holds neither another send nor this receipt
|
||||
return result;
|
||||
} catch (error) {
|
||||
say(`send failed: ${error.message}`);
|
||||
return refuse("SendFailed");
|
||||
|
|
@ -435,7 +479,7 @@ export function createMouth({
|
|||
}
|
||||
}
|
||||
|
||||
return { start, syncOnce, ack, pendingAfter, send, run, get pendingCount() { return pending.length; } };
|
||||
return { start, syncOnce, ack, read, pendingAfter, send, run, get pendingCount() { return pending.length; } };
|
||||
}
|
||||
|
||||
// Replace every secret in a message. `secrets` is one string or a list.
|
||||
|
|
@ -517,7 +561,7 @@ export async function startChatEndpoint({ mouths, runtime, port, connectionKey,
|
|||
try {
|
||||
if (channelId !== "schema") throw new Error("chat request arrived on a foreign channel");
|
||||
const message = cultnet.parseCultNetMessage(msgpack.decode(payload), "cultnet.schema.v0");
|
||||
const document = docs.checkEnvelope(message, [docs.kinds.pendingRequest, docs.kinds.ack, docs.kinds.sendRequest, docs.kinds.sendRequestV2]);
|
||||
const document = docs.checkEnvelope(message, [docs.kinds.pendingRequest, docs.kinds.ack, docs.kinds.read, docs.kinds.sendRequest, docs.kinds.sendRequestV2]);
|
||||
const tuple = msgpack.decode(document.payload);
|
||||
if (document.schemaId === docs.kinds.pendingRequest) {
|
||||
const request = docs.parsePendingRequest(tuple);
|
||||
|
|
@ -529,6 +573,11 @@ export async function startChatEndpoint({ mouths, runtime, port, connectionKey,
|
|||
const mouth = mouths.get(request.persona);
|
||||
if (!mouth) return refuseUnknown(peer, request.persona);
|
||||
await mouth.ack(request.throughSeq);
|
||||
} else if (document.schemaId === docs.kinds.read) {
|
||||
const request = docs.parseRead(tuple);
|
||||
const mouth = mouths.get(request.persona);
|
||||
if (!mouth) return refuseUnknown(peer, request.persona);
|
||||
mouth.read(request).catch(error => log(`read report failed: ${error?.message ?? error}`)); // the peer's next request does not wait on Matrix
|
||||
} else {
|
||||
const v2 = document.schemaId === docs.kinds.sendRequestV2;
|
||||
const request = v2 ? docs.parseSendRequestV2(tuple) : docs.parseSendRequest(tuple);
|
||||
|
|
|
|||
Loading…
Reference in a new issue