Merge pull request 'test(e2e): scheduled messages and offline outbox; retry rate-limited scheduled-message calls' (#476) from test/lotus-feature-e2e into lotus

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 },
+    ),
+  );
+});

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<T>(
+  fn: () => Promise<T>,
+  opts: { maxWaitMs?: number; sleepFn?: (ms: number) => Promise<void> } = {},
+): Promise<T> {
+  const maxWaitMs = opts.maxWaitMs ?? 15_000;
+  const sleepFn =
+    opts.sleepFn ??
+    ((ms: number) =>
+      new Promise<void>((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<void> {
   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<void> {
   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<void> {
   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<ScheduledMessage[]> {
-  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);
 }