From 56af9520bd5131e7bd36d288a12e5b39f7a79a15 Mon Sep 17 00:00:00 2001 From: Lotus CI Date: Wed, 28 Oct 2026 06:40:51 -0400 Subject: [PATCH] test(e2e): scheduled messages and offline outbox; retry rate-limited scheduled-message calls New end-to-end coverage for two Lotus features the October audit listed as untested, checked against the server at each step: - scheduled-messages.spec: schedule (tray + one delayed event on the server), Send now (arrives once, delayed event gone), cancel (gone from the server, never sent). - offline-outbox.spec: offline, three messages show "Queued. Will send when you are back online"; back online, the server has each exactly once and in order; the queued marks clear. The scheduled-messages test found a real bug: Synapse rate-limits delayed event management per user, and a quick Send now then Cancel got a 429, so the tray showed "Could not cancel this message" although nothing was wrong. All scheduled-message calls (schedule, list, send now, cancel, restart) now wait out a 429 (Retry-After, bounded) and retry; scheduling reuses its txnId. Unit test for the retry helper. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_017PCb59uY3UKqDqt7B6hzd5 --- e2e/offline-outbox.spec.ts | 72 +++++++++++++++++ e2e/scheduled-messages.spec.ts | 102 ++++++++++++++++++++++++ src/app/utils/scheduledMessages.test.ts | 40 ++++++++++ src/app/utils/scheduledMessages.ts | 102 +++++++++++++++++------- 4 files changed, 287 insertions(+), 29 deletions(-) create mode 100644 e2e/offline-outbox.spec.ts create mode 100644 e2e/scheduled-messages.spec.ts diff --git a/e2e/offline-outbox.spec.ts b/e2e/offline-outbox.spec.ts new file mode 100644 index 000000000..6bcb31880 --- /dev/null +++ b/e2e/offline-outbox.spec.ts @@ -0,0 +1,72 @@ +import { test, expect } from '@playwright/test'; +import { + api, + createRoom, + enc, + ensureUser, + hsReachable, + loginUI, + openRoom, + sendText, + uniq, +} from './localHs'; + +// Offline outbox (#112): messages sent while offline are queued, then +// delivered once each, in order, when the connection returns. Checked on the +// server, so duplicates or reordering would show. +test.describe('offline outbox', () => { + test.beforeAll(async () => { + test.skip(!(await hsReachable()), 'local homeserver not running'); + }); + + test('queued while offline, delivered once each and in order when back', async ({ + page, + context, + }) => { + test.setTimeout(120_000); + const me = await ensureUser(uniq('outbox')); + const room = await createRoom(me, 'Outbox Room'); + await sendText(me, room, 'before going offline'); + await loginUI(page, me); + await openRoom(page, room); + await expect(page.getByText('before going offline')).toBeVisible(); + + await context.setOffline(true); + const composer = page.getByRole('textbox', { name: /message/i }).last(); + const texts = [uniq('one '), uniq('two '), uniq('three ')]; + for (const text of texts) { + // eslint-disable-next-line no-await-in-loop + await composer.fill(text); + // eslint-disable-next-line no-await-in-loop + await composer.press('Enter'); + } + for (const text of texts) { + const item = page.locator('[data-message-item]', { hasText: text }); + // eslint-disable-next-line no-await-in-loop + await expect(item).toHaveCount(1); + // eslint-disable-next-line no-await-in-loop + await expect(item.getByLabel(/^Queued\. Will send when you're back online/)).toBeVisible({ + timeout: 20_000, + }); + } + + await context.setOffline(false); + const bodiesOnServer = async () => { + const res = await api<{ chunk: { type: string; content?: { body?: string } }[] }>( + 'GET', + `/_matrix/client/v3/rooms/${enc(room)}/messages?dir=f&limit=100`, + me.token, + ); + return res.chunk + .filter((e) => e.type === 'm.room.message' && texts.includes(e.content?.body ?? '')) + .map((e) => e.content?.body); + }; + await expect.poll(bodiesOnServer, { timeout: 60_000, intervals: [1000] }).toEqual(texts); + for (const text of texts) { + // eslint-disable-next-line no-await-in-loop + await expect( + page.locator('[data-message-item]', { hasText: text }).getByLabel(/^Queued/), + ).toHaveCount(0, { timeout: 20_000 }); + } + }); +}); diff --git a/e2e/scheduled-messages.spec.ts b/e2e/scheduled-messages.spec.ts new file mode 100644 index 000000000..c527286fb --- /dev/null +++ b/e2e/scheduled-messages.spec.ts @@ -0,0 +1,102 @@ +import { test, expect, Page } from '@playwright/test'; +import { + api, + createRoom, + enc, + ensureUser, + hsReachable, + loginUI, + openRoom, + sendText, + uniq, +} from './localHs'; + +// Scheduled messages (MSC4140 delayed events). Each step is checked on the +// server too, so this proves the delayed event really exists, fires and is +// cancelled, not just that the tray looks right. +type Delayed = { delayed_events?: { delay_id: string; room_id: string }[] }; +const delayedOnServer = (token: string) => + api('GET', '/_matrix/client/unstable/org.matrix.msc4140/delayed_events', token).then( + (r) => r.delayed_events ?? [], + ); + +// The tray remembers whether it is open; open it only if its rows are hidden. +const openTray = async (page: Page) => { + const rowButton = page.getByRole('button', { name: /^Cancel scheduled message: / }); + if (await rowButton.count()) return; + await page.getByRole('button', { name: /^\d+ scheduled messages?$/ }).click(); + await expect(rowButton.first()).toBeVisible(); +}; + +const schedule = async (page: Page, text: string) => { + await page.getByRole('button', { name: 'Schedule message' }).click(); + const dialog = page.getByRole('dialog', { name: 'Schedule message' }); + await dialog.locator('#schedule-message-body').fill(text); + // Two hours ahead, in the browser's own time zone. + const { date, time } = await page.evaluate(() => { + const d = new Date(Date.now() + 2 * 60 * 60 * 1000); + const p = (n: number) => String(n).padStart(2, '0'); + return { + date: `${d.getFullYear()}-${p(d.getMonth() + 1)}-${p(d.getDate())}`, + time: `${p(d.getHours())}:${p(d.getMinutes())}`, + }; + }); + await dialog.locator('#schedule-date').fill(date); + await dialog.locator('#schedule-time').fill(time); + await dialog.getByRole('button', { name: 'Schedule', exact: true }).click(); + await expect(dialog).toHaveCount(0); +}; + +test.describe('scheduled messages', () => { + test.beforeAll(async () => { + test.skip(!(await hsReachable()), 'local homeserver not running'); + }); + + test('schedule, send now, and cancel — the server agrees at each step', async ({ page }) => { + test.setTimeout(120_000); + const me = await ensureUser(uniq('sched')); + const room = await createRoom(me, 'Schedule Room'); + await sendText(me, room, 'room is ready'); + await loginUI(page, me); + await openRoom(page, room); + + // 1. Schedule: tray shows it, the server holds one delayed event. + const first = uniq('later '); + await schedule(page, first); + await expect(page.getByRole('button', { name: '1 scheduled message' })).toBeVisible(); + await expect + .poll(async () => (await delayedOnServer(me.token)).length, { intervals: [1000] }) + .toBe(1); + expect((await delayedOnServer(me.token))[0].room_id).toBe(room); + await expect(page.locator('[data-message-item]', { hasText: first })).toHaveCount(0); + + // 2. Send now: it arrives once, the delayed event is gone. + await openTray(page); + await page.getByRole('button', { name: /^Send scheduled message now: / }).click(); + await expect(page.locator('[data-message-item]', { hasText: first })).toHaveCount(1, { + timeout: 20_000, + }); + await expect + .poll(async () => (await delayedOnServer(me.token)).length, { intervals: [1000] }) + .toBe(0); + + // 3. Cancel: removed from the server and never sent. + const second = uniq('never '); + await schedule(page, second); + await expect + .poll(async () => (await delayedOnServer(me.token)).length, { intervals: [1000] }) + .toBe(1); + await openTray(page); + await page.getByRole('button', { name: /^Cancel scheduled message: / }).click(); + await expect + .poll(async () => (await delayedOnServer(me.token)).length, { intervals: [1000] }) + .toBe(0); + await expect(page.getByRole('button', { name: /scheduled message/ })).toHaveCount(0); + const msgs = await api<{ chunk: { content?: { body?: string } }[] }>( + 'GET', + `/_matrix/client/v3/rooms/${enc(room)}/messages?dir=b&limit=50`, + me.token, + ); + expect(msgs.chunk.some((e) => e.content?.body === second)).toBe(false); + }); +}); diff --git a/src/app/utils/scheduledMessages.test.ts b/src/app/utils/scheduledMessages.test.ts index 45c2e7fb3..25dbde9b9 100644 --- a/src/app/utils/scheduledMessages.test.ts +++ b/src/app/utils/scheduledMessages.test.ts @@ -191,3 +191,43 @@ test('[#275] the server limit maps to friendly text; other errors keep theirs', assert.equal(describeScheduleError(new Error('Boom')), 'Boom'); assert.equal(describeScheduleError('nope'), 'Failed to schedule message.'); }); + +test('withRateLimitRetry: waits out a 429 then succeeds; other errors and over-budget waits throw', async () => { + const { withRateLimitRetry } = await import('./scheduledMessages'); + const { MatrixError } = await import('matrix-js-sdk'); + const limited = () => new MatrixError({ errcode: 'M_LIMIT_EXCEEDED', retry_after_ms: 1500 }, 429); + const sleeps: number[] = []; + const sleepFn = async (ms: number) => { + sleeps.push(ms); + }; + + let calls = 0; + const ok = await withRateLimitRetry( + async () => { + calls += 1; + if (calls < 3) throw limited(); + return 'done'; + }, + { sleepFn }, + ); + assert.equal(ok, 'done'); + assert.deepEqual(sleeps, [1500, 1500]); + + await assert.rejects( + withRateLimitRetry( + async () => { + throw new MatrixError({ errcode: 'M_FORBIDDEN' }, 403); + }, + { sleepFn }, + ), + /M_FORBIDDEN|403/, + ); + await assert.rejects( + withRateLimitRetry( + async () => { + throw limited(); + }, + { sleepFn, maxWaitMs: 2000 }, + ), + ); +}); diff --git a/src/app/utils/scheduledMessages.ts b/src/app/utils/scheduledMessages.ts index 0b76e1ed9..189b1811f 100644 --- a/src/app/utils/scheduledMessages.ts +++ b/src/app/utils/scheduledMessages.ts @@ -1,4 +1,4 @@ -import { IContent, MatrixClient, Method } from 'matrix-js-sdk'; +import { IContent, MatrixClient, MatrixError, Method } from 'matrix-js-sdk'; import type { ScheduledMessage } from '../state/scheduledMessages'; /** @@ -17,6 +17,40 @@ export const describeScheduleError = (err: unknown): string => { return err instanceof Error ? err.message : 'Failed to schedule message.'; }; +/** + * Synapse rate-limits delayed-event management per user (listing, send now, + * cancel, restart all count). A quick "Send now" then "Cancel" could get a 429 + * and show "Could not cancel" although nothing was wrong. Wait out the + * Retry-After (bounded) and try again. Retrying is safe: a 429 means the + * request was not processed, and scheduling reuses its txnId. + */ +export async function withRateLimitRetry( + fn: () => Promise, + opts: { maxWaitMs?: number; sleepFn?: (ms: number) => Promise } = {}, +): Promise { + const maxWaitMs = opts.maxWaitMs ?? 15_000; + const sleepFn = + opts.sleepFn ?? + ((ms: number) => + new Promise((r) => { + setTimeout(r, ms); + })); + let waited = 0; + for (;;) { + try { + // eslint-disable-next-line no-await-in-loop + return await fn(); + } catch (e) { + if (!(e instanceof MatrixError) || e.httpStatus !== 429) throw e; + const wait = Math.min(Math.max(e.getRetryAfterMs() ?? 1000, 200), 10_000); + if (waited + wait > maxWaitMs) throw e; + waited += wait; + // eslint-disable-next-line no-await-in-loop + await sleepFn(wait); + } + } +} + /** * Schedule a message via MSC4140 (delayed messages). * Synapse implementation: pass org.matrix.msc4140.delay on the standard @@ -42,13 +76,15 @@ export async function scheduleMessage( const txnId = `sched_${Date.now()}_${Math.random().toString(36).slice(2)}`; const path = `/rooms/${encodeURIComponent(roomId)}/send/m.room.message/${txnId}`; // Use standard v3 prefix with the MSC4140 delay query param. - const res = (await mx.http.authedRequest( - Method.Put, - path, - { - 'org.matrix.msc4140.delay': delayMs, - }, - content, + const res = (await withRateLimitRetry(() => + mx.http.authedRequest( + Method.Put, + path, + { + 'org.matrix.msc4140.delay': delayMs, + }, + content, + ), )) as { delay_id: string }; return res.delay_id; } @@ -58,23 +94,27 @@ export async function scheduleMessage( */ export async function cancelScheduledMessage(mx: MatrixClient, delayId: string): Promise { const path = `/delayed_events/${encodeURIComponent(delayId)}`; - await mx.http.authedRequest( - Method.Post, - path, - undefined, - { action: 'cancel' }, - { prefix: '/_matrix/client/unstable/org.matrix.msc4140' }, + await withRateLimitRetry(() => + mx.http.authedRequest( + Method.Post, + path, + undefined, + { action: 'cancel' }, + { prefix: '/_matrix/client/unstable/org.matrix.msc4140' }, + ), ); } export async function restartScheduledMessage(mx: MatrixClient, delayId: string): Promise { const path = `/delayed_events/${encodeURIComponent(delayId)}`; - await mx.http.authedRequest( - Method.Post, - path, - undefined, - { action: 'restart' }, - { prefix: '/_matrix/client/unstable/org.matrix.msc4140' }, + await withRateLimitRetry(() => + mx.http.authedRequest( + Method.Post, + path, + undefined, + { action: 'restart' }, + { prefix: '/_matrix/client/unstable/org.matrix.msc4140' }, + ), ); } @@ -85,12 +125,14 @@ export async function restartScheduledMessage(mx: MatrixClient, delayId: string) */ export async function sendScheduledMessageNow(mx: MatrixClient, delayId: string): Promise { const path = `/delayed_events/${encodeURIComponent(delayId)}`; - await mx.http.authedRequest( - Method.Post, - path, - undefined, - { action: 'send' }, - { prefix: '/_matrix/client/unstable/org.matrix.msc4140' }, + await withRateLimitRetry(() => + mx.http.authedRequest( + Method.Post, + path, + undefined, + { action: 'send' }, + { prefix: '/_matrix/client/unstable/org.matrix.msc4140' }, + ), ); } @@ -138,9 +180,11 @@ export function parseDelayedEvents(res: unknown): ScheduledMessage[] { /** The user's pending scheduled messages, as the server has them. */ export async function fetchScheduledMessages(mx: MatrixClient): Promise { - const res = await mx.http.authedRequest(Method.Get, '/delayed_events', undefined, undefined, { - prefix: '/_matrix/client/unstable/org.matrix.msc4140', - }); + const res = await withRateLimitRetry(() => + mx.http.authedRequest(Method.Get, '/delayed_events', undefined, undefined, { + prefix: '/_matrix/client/unstable/org.matrix.msc4140', + }), + ); return parseDelayedEvents(res); }