From cdda75d13b79eb20e5396608d773a635b0fbfc03 Mon Sep 17 00:00:00 2001 From: Linus Rath <139418639+rathlinus@users.noreply.github.com> Date: Tue, 3 Nov 2026 08:49:53 +0100 Subject: [PATCH] fix: keep queued changes and mail through rate limits, logins that need renewing and proxy errors, and show the server's copy after a real refusal --- lib/offline/__tests__/review-fixes.test.ts | 62 +++++++++++++++++++++- lib/offline/flush.ts | 49 ++++++++++++----- lib/store/__tests__/outbox.test.ts | 53 +++++++++++++++--- lib/store/outbox/flush-patches.ts | 50 +++++++++++++---- 4 files changed, 183 insertions(+), 31 deletions(-) diff --git a/lib/offline/__tests__/review-fixes.test.ts b/lib/offline/__tests__/review-fixes.test.ts index 11bd3cfd5..3801d334e 100644 --- a/lib/offline/__tests__/review-fixes.test.ts +++ b/lib/offline/__tests__/review-fixes.test.ts @@ -25,7 +25,7 @@ import { wipeEverything, } from '../offline-mode'; import { flushQueue, settleStaleSends, SENDING_STALE_MS } from '../flush'; -import { claimForSending, holdFile, queueSend, readQueue } from '../queue'; +import { claimForSending, holdFile, LOCAL_BLOB_PREFIX, queueSend, readQueue, updateOutboxEntry } from '../queue'; import { forgetLegacyMove } from '../migrate-legacy'; import { consumeLost } from '../lost'; import { getOutboxHooks } from '../outbox-hooks'; @@ -295,6 +295,66 @@ describe('11: starting from the device', () => { }); }); +describe('a send the server couldn\'t take right now', () => { + const statusAfter = async (error: Error) => { + writeEnabled(['acc']); + await queueSend('acc', { subject: 'Hello', to: [] }, { kind: 'email', args: [] }); + const notify = vi.fn(); + const retryLater = vi.fn(); + const sendEmail = vi.fn(async () => { throw error; }); + await flushQueue('acc', fakeClient('https://mail.example.org', { sendEmail }) as unknown as JMAPClient, { notify, allowed: () => true, retryLater }); + const row = (await readQueue('acc')).outbox[0]; + return { row, notify, retryLater }; + }; + + it.each([ + ['a rate limit', Object.assign(new Error('Rate limited by server'), { name: 'RateLimitError' })], + ['a login that needs renewing', new Error('Request failed: 401 - ')], + ['a store that went away', Object.assign(new Error('gone'), { name: 'StoreOwnerGoneError' })], + ['a full disk', Object.assign(new Error('full'), { name: 'QuotaExceededError' })], + ])('goes back to waiting, on a timer, after %s', async (_name, error) => { + const { row, notify, retryLater } = await statusAfter(error); + expect(row.status).toBe('waiting'); + expect(retryLater).toHaveBeenCalled(); + expect(notify).not.toHaveBeenCalled(); + }); + + it.each(['Request failed: 500 - oops', 'Request failed: 501 - nope'])('is looked up after %s, which may have got through', async (message) => { + const { row, notify, retryLater } = await statusAfter(new Error(message)); + expect(row.status).toBe('uncertain'); + expect(retryLater).toHaveBeenCalled(); + expect(notify).not.toHaveBeenCalled(); + }); + + it('fails, for the user to see, on a refusal', async () => { + const { row, notify } = await statusAfter(new Error('Request failed: 400 - {"type":"urn:ietf:params:jmap:error:notRequest"}')); + expect(row.status).toBe('failed'); + expect(notify).toHaveBeenCalledWith(expect.objectContaining({ kind: 'send-failed' })); + }); + + it('keeps the copy on the server through a wait', async () => { + writeEnabled(['acc']); + const id = await queueSend('acc', { subject: 'Hello', to: [] }, { kind: 'email', args: [] }); + const [entry] = (await readQueue('acc')).outbox; + await updateOutboxEntry(entry, { status: 'waiting', copyId: 'copy-1' }); + const sendEmail = vi.fn(async () => { throw new Error('Request failed: 429 - '); }); + await flushQueue('acc', fakeClient('https://mail.example.org', { sendEmail }) as unknown as JMAPClient, { notify: vi.fn(), allowed: () => true }); + const row = (await readQueue('acc')).outbox.find((candidate) => candidate.id === id); + expect(row).toMatchObject({ status: 'waiting', copyId: 'copy-1' }); + }); + + it('doesn\'t send a stand-in for a file the upload held again', async () => { + writeEnabled(['acc']); + const { blobId } = await holdFile('acc', new File(['x'], 'x.txt', { type: 'text/plain' }), 'x.txt'); + await queueSend('acc', { subject: 'Hello', to: [] }, { kind: 'email', args: [undefined, undefined, undefined, undefined, undefined, undefined, undefined, undefined, undefined, undefined, [{ blobId, name: 'x.txt', type: 'text/plain', size: 1 }]] }); + const sendEmail = vi.fn(async () => ({ scheduled: false, emailId: 'e', emailSubmissionId: 's' })); + const uploadBlob = vi.fn(async () => ({ blobId: `${LOCAL_BLOB_PREFIX}stand-in`, size: 1, type: 'text/plain' })); + await flushQueue('acc', fakeClient('https://mail.example.org', { sendEmail, uploadBlob }) as unknown as JMAPClient, { notify: vi.fn(), allowed: () => true }); + expect(sendEmail).not.toHaveBeenCalled(); + expect((await readQueue('acc')).outbox[0].status).toBe('waiting'); + }); +}); + describe('a send on the local copy', () => { it('queues, before the swap too, while the probes say the server is up', () => { writeEnabled(['acc']); diff --git a/lib/offline/flush.ts b/lib/offline/flush.ts index 53e250173..291004300 100644 --- a/lib/offline/flush.ts +++ b/lib/offline/flush.ts @@ -11,9 +11,10 @@ * - A message is sent as it was finished at Send (queue.ts); files attached * offline are uploaded first. A hold for undo-send that has passed is * dropped, so the message goes at once. - * - Only a failure before the submission request (SendNotStartedError, a - * file upload on a dead network) puts a message back to waiting. A send - * whose connection broke after that is marked uncertain. A refusal by the + * - A failure before the submission request (SendNotStartedError, a file + * upload on a dead network), a rate limit, a login that needs renewing and + * trouble with this device's store put a message back to waiting. A send whose + * connection broke, or that got a 5xx, is marked uncertain. A refusal by the * server is marked failed, with the reason. * - An uncertain message is looked up when the server is back, by the * Message-ID fixed when it was queued (lib/store/outbox/dedupe.ts): it's @@ -29,6 +30,7 @@ import type { JMAPClient } from '@/lib/jmap/client'; import { isNetworkFailure } from '@/lib/jmap/offline-hooks'; import type { SendEmailResult } from '@/lib/jmap/types'; import { getStore } from '@/lib/store/client/api'; +import { isStorageFailure } from '@/lib/store/storage'; import type { MethodCall } from '@/lib/store/engine/transport'; import { decideRetry, type LookupIo, type RetryDecision } from '@/lib/store/outbox/dedupe'; import { flushPatches, type PatchIo, type PatchOutcome, type PendingPatch } from '@/lib/store/outbox/flush-patches'; @@ -72,6 +74,23 @@ class OutboxPrepError extends Error { } } +/** + * Not the server's word about this message: it's rate limiting, the login needs + * renewing, or this device's store is in the way. The next try goes as usual. + */ +function waitsForNextTry(error: unknown): boolean { + if (!(error instanceof Error)) return false; + if (error.name === 'RateLimitError' || error.name === 'StoreOwnerGoneError' || error.name === 'StoreKeyError') return true; + if (isStorageFailure(error)) return true; + return /^Request failed: (401|408|429)\b/.test(error.message); +} + +/** A server error can come after the message went through: looked up before it goes again. */ +function mayHaveArrived(error: unknown): boolean { + if (isNetworkFailure(error)) return true; + return error instanceof Error && (error.name === 'SendUnconfirmedError' || /^Request failed: 5\d\d\b/.test(error.message)); +} + function notStartedError(message: string): Error { return Object.assign(new Error(message), { name: 'SendNotStartedError' }); } @@ -89,12 +108,16 @@ function notStarted(error: unknown): boolean { async function uploadHeld(account: string, client: JMAPClient, blobId: string): Promise { const file = await heldFile(account, blobId); if (!file) throw new OutboxPrepError('file_missing'); + let uploaded: string; try { - return (await client.uploadBlob(file)).blobId; + uploaded = (await client.uploadBlob(file)).blobId; } catch (error) { if (isNetworkFailure(error)) throw notStartedError('upload failed'); throw error; } + // The server went away just now, and offline mode held the file again. + if (isLocalBlobId(uploaded)) throw notStartedError('upload held'); + return uploaded; } async function sendEntry(account: string, client: JMAPClient, entry: OutboxEntry): Promise { @@ -153,12 +176,8 @@ export async function flushQueue(account: string, client: JMAPClient, deps: Flus /** Marks sends of closed tabs uncertain (they may have gone out). */ export async function settleStaleSends(account?: string, now = Date.now()): Promise { const accounts = account ? [account] : visibleAccounts(); - await Promise.all(accounts.map((id) => getStore().area(id, 'outbox.settleStale', { before: now - SENDING_STALE_MS }))); -} - -/** The server refused the request itself: more tries won't change that. Outages and auth wait instead. */ -function isRequestRefusal(error: unknown): boolean { - return error instanceof Error && /^Request failed: (400|403|404|413|422)\b/.test(error.message); + // One account's store in trouble doesn't hold up another's. + await Promise.allSettled(accounts.map((id) => getStore().area(id, 'outbox.settleStale', { before: now - SENDING_STALE_MS }))); } /** The changes queued before `upToSeq` (all, with null). False when the run stopped on an outage. */ @@ -167,7 +186,6 @@ async function sendPatches(account: string, client: JMAPClient, deps: FlushDeps, if (pending.length === 0) return true; const io: PatchIo = { request: (calls: MethodCall[], using: string[]) => client.requestNetworkOnly(calls, using), - isRefusal: isRequestRefusal, }; return flushPatches(io, pending, async (outcomes: PatchOutcome[]) => { const conflicts = await getStore().area(account, 'outbox.patchResults', { outcomes }); @@ -271,8 +289,13 @@ async function flushLocked(account: string, client: JMAPClient, deps: FlushDeps) deps.notify({ kind: 'send-failed', subject: entry.summary.subject, error: { code } }); continue; } - if (isNetworkFailure(error) || (error instanceof Error && error.name === 'SendUnconfirmedError')) { - // It may have reached the server: looked up before it goes again. + if (waitsForNextTry(error)) { + // Nothing the user can do about it, and the copy on the server stays useful. + await updateOutboxEntry(entry, { status: 'waiting' }); + deps.retryLater?.(); + return; + } + if (mayHaveArrived(error)) { await updateOutboxEntry(entry, { status: 'uncertain' }); deps.retryLater?.(); return; diff --git a/lib/store/__tests__/outbox.test.ts b/lib/store/__tests__/outbox.test.ts index 0edf2cc39..af80990e5 100644 --- a/lib/store/__tests__/outbox.test.ts +++ b/lib/store/__tests__/outbox.test.ts @@ -62,9 +62,8 @@ async function boxes() { return Object.fromEntries((answer?.methodResponses[0][1].list as Array<{ id: string; totalEmails: number; unreadEmails: number }>).map((box) => [box.id, [box.totalEmails, box.unreadEmails]])); } -const io = (server: FakeJmapMail = t.server, isRefusal: (error: unknown) => boolean = () => false) => ({ +const io = (server: FakeJmapMail = t.server) => ({ request: (calls: MethodCall[]) => server.request(calls), - isRefusal, }); async function flush(ioOverride = io()) { @@ -358,16 +357,54 @@ describe('when the server can\'t be reached during a flush', () => { expect(await pending()).toHaveLength(1); }); - it('turns a request the server refuses for good into a conflict, not a stuck queue', async () => { + /** Refuses the Email/set with this answer, and reads go through. */ + const refusingSet = (message: string) => ({ + request: async (calls: MethodCall[]) => { + if (calls[0][0] === 'Email/set') throw new Error(message); + return t.server.request(calls); + }, + }); + const PROBLEM = 'Request failed: 400 - {"type":"urn:ietf:params:jmap:error:notRequest","status":400}'; + + it('turns a request the server refuses for good into a conflict, and shows what the server has', async () => { await update({ a: { 'keywords/$seen': true } }); - const refusing = { - request: async () => { throw new Error('Request failed: 400 - bad'); }, - isRefusal: (error: unknown) => /^Request failed: 400/.test((error as Error).message), - }; - const { outcomes, finished } = await flush(refusing); + const { outcomes, finished } = await flush(refusingSet(PROBLEM)); expect(finished).toBe(true); expect(outcomes[0]).toMatchObject({ outcome: 'conflict', reason: 'refused' }); expect(await pending()).toEqual([]); + // The change is gone, and so is its effect here. + const answer = await t.engine.answer([['Email/get', { ids: ['a'], properties: ['keywords'] }, 'g']]); + expect(answer?.methodResponses[0][1].list[0].keywords).toEqual({}); + expect((await read()).conflicts).toHaveLength(1); + }); + + it.each([ + ['the parallel-request limit', 'Request failed: 400 - {"type":"urn:ietf:params:jmap:error:limit","limit":"maxConcurrentRequests"}'], + ['a proxy\'s error page', 'Request failed: 403 - Forbidden'], + ['a body that\'s no JMAP problem', 'Request failed: 413 - Payload too large'], + ['a server error', 'Request failed: 500 - oops'], + ])('keeps the change for later on %s', async (_name, message) => { + await update({ a: { 'keywords/$seen': true } }); + const { outcomes, finished } = await flush(refusingSet(message)); + expect(finished).toBe(false); + expect(outcomes).toEqual([]); + expect(await pending()).toHaveLength(1); + expect((await read()).conflicts).toEqual([]); + }); + + it('keeps the change for later when the read after a refusal fails', async () => { + await update({ a: { 'keywords/$seen': true } }); + let calls = 0; + const { finished } = await flush({ + request: async (batch: MethodCall[]) => { + calls += 1; + if (calls === 1) return t.server.request(batch); + if (batch[0][0] === 'Email/set') throw new Error(PROBLEM); + throw new TypeError('Failed to fetch'); + }, + }); + expect(finished).toBe(false); + expect(await pending()).toHaveLength(1); }); it('keeps a change made while the flush ran, and doesn\'t compare it with the state the flush caused', async () => { diff --git a/lib/store/outbox/flush-patches.ts b/lib/store/outbox/flush-patches.ts index 361cf817d..ae0901828 100644 --- a/lib/store/outbox/flush-patches.ts +++ b/lib/store/outbox/flush-patches.ts @@ -42,17 +42,26 @@ export type PatchOutcome = export interface PatchIo { request(calls: MethodCall[], using: string[]): Promise; - /** The server refused the request itself, for good: not an outage, and no use retrying. */ - isRefusal(error: unknown): boolean; } const BATCH = 200; /** Method errors that mean "not now" rather than "no". */ const RETRY_LATER = new Set(['serverUnavailable', 'serverFail', 'serverPartialFail', 'rateLimit', 'requestTooLarge']); -function isLater(error: unknown, io: PatchIo): boolean { +/** + * The server refused the request itself, for good: a JMAP problem document with + * a status that says so. `limit` is the parallel-request ceiling, which passes, + * and an error page says something about a proxy, not about this request. + */ +export function isRequestRefusal(error: unknown): boolean { + if (!(error instanceof Error)) return false; + const problem = /^Request failed: (?:400|403|404|413|422)\b[\s\S]*?"type"\s*:\s*"urn:ietf:params:jmap:error:([A-Za-z]+)"/.exec(error.message); + return !!problem && problem[1] !== 'limit'; +} + +function isLater(error: unknown): boolean { if (error instanceof JmapMethodError) return RETRY_LATER.has(error.type); - return !io.isRefusal(error); + return !isRequestRefusal(error); } function conflict(row: PendingPatch, reason: ConflictReason, server?: EmailObject | null, detail?: string): PatchOutcome { @@ -67,14 +76,36 @@ function errorText(error: unknown): string { return error instanceof Error ? error.message : String(error); } -async function flushBatch(io: PatchIo, jmap: string, rows: PendingPatch[]): Promise { +async function readRecords(io: PatchIo, jmap: string, rows: PendingPatch[]): Promise> { const read = await io.request([['Email/get', { accountId: jmap, ids: rows.map((row) => row.recordId), properties: EMAIL_LIST_PROPERTIES, }, 'g']], MAIL_USING); const [, got] = responseFor(read, 'g'); - const current = new Map((got.list as EmailObject[]).map((email) => [email.id, email])); + return new Map((got.list as EmailObject[]).map((email) => [email.id, email])); +} + +/** + * The whole batch was refused. The server's copy of each record goes with the + * conflict so that this device shows what's true again. When even that read is + * refused the record stays as it is here until the next sync. + */ +async function refusedOutcomes(io: PatchIo, jmap: string, rows: PendingPatch[], refusal: unknown): Promise { + let current: Map | null = null; + try { + current = await readRecords(io, jmap, rows); + } catch (error) { + if (isLater(error)) return null; + } + return rows.map((row) => { + if (current && !current.has(row.recordId)) return row.op === 'destroy' ? done(row) : conflict(row, 'gone', null); + return conflict(row, 'refused', current?.get(row.recordId), errorText(refusal)); + }); +} + +async function flushBatch(io: PatchIo, jmap: string, rows: PendingPatch[]): Promise { + const current = await readRecords(io, jmap, rows); const outcomes: PatchOutcome[] = []; const update: Record = {}; @@ -129,12 +160,13 @@ export async function flushPatches( for (const [jmap, rows] of byAccount) { for (let i = 0; i < rows.length; i += BATCH) { const batch = rows.slice(i, i + BATCH); - let outcomes: PatchOutcome[]; + let outcomes: PatchOutcome[] | null; try { outcomes = await flushBatch(io, jmap, batch); } catch (error) { - if (isLater(error, io)) return false; - outcomes = batch.map((row) => conflict(row, 'refused', undefined, errorText(error))); + if (isLater(error)) return false; + outcomes = await refusedOutcomes(io, jmap, batch, error); + if (!outcomes) return false; } await report(outcomes); }