From b1a8ad20404e13cd7dd16903e7933570c143eb87 Mon Sep 17 00:00:00 2001 From: Sivert Date: Mon, 28 Sep 2026 10:39:21 +0200 Subject: [PATCH] Keep MLS replies to ten binary parts (GRYT-1528) socket.io-parser closes the connection on a packet with more than ten binary parts. mls:sync sent every waiting Welcome in one reply, so a device that eleven DMs added while it was offline got dropped on every connect and never got past it. mls:sync now sends the oldest nine and sets moreWelcomes. Acking them brings the next nine. mls:log:fetch pages at ten entries whatever limit is asked for, and ten is the default now too. mls:keypackages:claim claims for nine devices at most and lists the rest in more. A DM has nine other devices at most, so that one can't come up yet. Co-Authored-By: Claude Opus 5.5 --- src/socket/handlers/mls.test.ts | 43 ++++++++++++++++++++++++++++++++- src/socket/handlers/mls.ts | 21 ++++++++++------ 2 files changed, 56 insertions(+), 8 deletions(-) diff --git a/src/socket/handlers/mls.test.ts b/src/socket/handlers/mls.test.ts index bf101d55..1a844f2b 100644 --- a/src/socket/handlers/mls.test.ts +++ b/src/socket/handlers/mls.test.ts @@ -12,7 +12,7 @@ import { blockUser, unblockUser } from "../../db/sqlite/blocks"; import { getSqliteDb, initSqlite } from "../../db/sqlite/connection"; import { setContactPrefs } from "../../db/sqlite/contactPrefs"; import { createGroupConversation, openDirectConversation } from "../../db/sqlite/conversations"; -import { getMlsGroupForConversation } from "../../db/sqlite/mls"; +import { appendMlsMessage, createMlsGroup, getMlsGroupForConversation } from "../../db/sqlite/mls"; import { createServerConfigIfNotExists, setServerRole, updateServerConfig } from "../../db/sqlite/servers"; import { hasWrittenTo } from "../../db/sqlite/contactPrefs"; import { listMessages } from "../../db/sqlite/messages"; @@ -407,6 +407,7 @@ describe("KeyPackages", () => { const claimed = await gina.call("mls:keypackages:claim", { conversationId: conv, deviceId: ginaLaptop.id }); assert.deepEqual(claimed.keyPackages, []); assert.deepEqual(claimed.missing, [{ serverUserId: hans.serverUserId, deviceId: hansPhone.id }]); + assert.deepEqual(claimed.more, [], "nothing left over: a DM never has more than nine other devices"); const sync = await hans.call("mls:sync", { deviceId: hansPhone.id }); assert.equal((sync.keyPackages as { unclaimed: number }).unclaimed, 0, "so the phone knows to upload more"); @@ -709,6 +710,46 @@ describe("a DM over MLS", () => { }); }); +describe("packets socket.io will carry", () => { + /* socket.io-parser closes the connection on more than ten binary parts (GRYT-1528). */ + it("hands out waiting Welcomes nine at a time, and says when there are more", async () => { + const kim = await connect("Kim"); + const phone = await makeDevice("kim-phone", 1); + await publish(kim, phone); + const insert = getSqliteDb().prepare( + `INSERT INTO mls_welcomes (welcome_id, server_user_id, device_id, group_id, data, created_at) VALUES (?, ?, ?, ?, ?, ?)`, + ); + for (let i = 0; i < 11; i++) { + insert.run(`kim-welcome-${i}`, kim.serverUserId, phone.id, `ab${i}`, new Uint8Array([i]), new Date(Date.now() + i).toISOString()); + } + + const first = await kim.call("mls:sync", { deviceId: phone.id }); + const ids = (first.welcomes as { welcomeId: string }[]).map((w) => w.welcomeId); + assert.deepEqual(ids, Array.from({ length: 9 }, (_, i) => `kim-welcome-${i}`), "oldest first"); + assert.equal(first.moreWelcomes, true); + + await kim.call("mls:welcome:ack", { deviceId: phone.id, welcomeIds: ids }); + const rest = await kim.call("mls:sync", { deviceId: phone.id }); + assert.equal((rest.welcomes as unknown[]).length, 2); + assert.equal(rest.moreWelcomes, false); + }); + + it("pages the log ten entries at a time, whatever limit is asked for", async () => { + const lou = await connect("Lou"); + const max = await connect("Max"); + const conv = (await openDirectConversation(lou.serverUserId, max.serverUserId)).conversation_id; + createMlsGroup("cafe01", conv, lou.serverUserId); + for (let i = 0; i < 12; i++) { + appendMlsMessage("application", { groupId: "cafe01", epoch: 0, senderServerUserId: lou.serverUserId, senderDeviceId: "d", data: new Uint8Array([i, 1]) }); + } + for (const limit of [200, undefined]) { + const page = await max.call("mls:log:fetch", { conversationId: conv, after: 0, limit }); + assert.equal((page.entries as unknown[]).length, 10); + assert.equal(page.hasMore, true); + } + }); +}); + describe("retention", () => { it("is 30 days, and a host can only make it shorter", () => { assert.equal(mlsRetentionDays({}), 30); diff --git a/src/socket/handlers/mls.ts b/src/socket/handlers/mls.ts index 0a515423..f3044788 100644 --- a/src/socket/handlers/mls.ts +++ b/src/socket/handlers/mls.ts @@ -64,8 +64,12 @@ const GROUP_ID = /^(?:[0-9a-f]{2}){1,64}$/; const MAX_KEY_PACKAGE_BYTES = 16 * 1024; /** A DM commit adds at most ten devices; a Welcome carries the tree for all of them. */ const MAX_HANDSHAKE_BYTES = 256 * 1024; -const MAX_LOG_PAGE = 200; -const MAX_CLAIM_DEVICES = 50; +/** socket.io-parser closes the connection on a packet with more than ten binary parts, + so no reply carries more (GRYT-1528). Nine where the count is ours to pick. */ +const MAX_BINARY_PARTS = 10; +const MAX_LOG_PAGE = MAX_BINARY_PARTS; +const MAX_CLAIM_DEVICES = MAX_BINARY_PARTS - 1; +const MAX_SYNC_WELCOMES = MAX_BINARY_PARTS - 1; const RL_PUBLISH: RateLimitRule = { limit: 10, windowMs: 60_000, scorePerAction: 1, maxScore: 10, scoreDecayMs: 6000 }; /** Every claim uses up somebody else's packages, so it is the one to hold back. */ @@ -386,6 +390,8 @@ export function registerMlsHandlers(ctx: HandlerContext): EventHandlerMap { const asked = new Set(payload.devices.map((d) => `${d?.serverUserId}\u0000${d?.deviceId}`)); targets = targets.filter((d) => asked.has(`${d.serverUserId}\u0000${d.deviceId}`)); } + // A DM has nine other devices at most, so `more` is for later stages; name them to get them. + const more = targets.slice(MAX_CLAIM_DEVICES).map((d) => ({ serverUserId: d.serverUserId, deviceId: d.deviceId })); targets = targets.slice(0, MAX_CLAIM_DEVICES); const peers = new Set(targets.map((d) => d.serverUserId).filter((id) => id !== self)); @@ -415,7 +421,7 @@ export function registerMlsHandlers(ctx: HandlerContext): EventHandlerMap { missing.push({ serverUserId: d.serverUserId, deviceId: d.deviceId }); } } - ack({ ok: true, keyPackages, missing }); + ack({ ok: true, keyPackages, missing, more }); } catch (err) { consola.error("mls:keypackages:claim failed", err); ack(fail("failed", "Could not hand out KeyPackages")); @@ -676,7 +682,7 @@ export function registerMlsHandlers(ctx: HandlerContext): EventHandlerMap { ack(fail("invalid_payload", "after has to be a seq, 0 for the start.")); return; } - const limit = typeof payload.limit === "number" && payload.limit >= 1 ? Math.min(Math.floor(payload.limit), MAX_LOG_PAGE) : 100; + const limit = typeof payload.limit === "number" && payload.limit >= 1 ? Math.min(Math.floor(payload.limit), MAX_LOG_PAGE) : MAX_LOG_PAGE; const group = getMlsGroupForConversation(payload.conversationId); if (!group) { ack({ ok: true, group: null, entries: [], nextCursor: after, hasMore: false, gap: false }); @@ -714,14 +720,15 @@ export function registerMlsHandlers(ctx: HandlerContext): EventHandlerMap { const self = auth.tokenPayload.serverUserId; const registered = isMlsDevice(self, payload.deviceId); if (registered) touchMlsDevice(self, payload.deviceId); + // Oldest first, a page at a time. Acking them is what brings the next ones. + const waiting = registered ? listMlsWelcomes(self, payload.deviceId) : []; ack({ ok: true, registered, groups: listMlsGroupsForMember(self).map((g) => ({ ...groupView(g), oldestSeq: g.oldestSeq })), - welcomes: registered - ? listMlsWelcomes(self, payload.deviceId).map((w) => ({ ...w, data: Buffer.from(w.data) })) - : [], + welcomes: waiting.slice(0, MAX_SYNC_WELCOMES).map((w) => ({ ...w, data: Buffer.from(w.data) })), + moreWelcomes: waiting.length > MAX_SYNC_WELCOMES, keyPackages: { ...countMlsKeyPackages(self, payload.deviceId), target: MLS_MAX_KEY_PACKAGES }, }); } catch (err) {