From 75213f6a551278c295a6e95246dc15a6cbf1e101 Mon Sep 17 00:00:00 2001 From: Linus Rath <139418639+rathlinus@users.noreply.github.com> Date: Sat, 31 Oct 2026 18:32:21 +0100 Subject: [PATCH] fix: send a message queued or retried while the outbox is already sending --- .../__tests__/outbox-send-queue.test.ts | 31 +++++++++++++++++++ src/stores/outbox-store.ts | 15 +++++++-- 2 files changed, 44 insertions(+), 2 deletions(-) diff --git a/src/stores/__tests__/outbox-send-queue.test.ts b/src/stores/__tests__/outbox-send-queue.test.ts index 11655c32..c7555456 100644 --- a/src/stores/__tests__/outbox-send-queue.test.ts +++ b/src/stores/__tests__/outbox-send-queue.test.ts @@ -141,6 +141,37 @@ describe('offline send queue', () => { expect(runQueuedSend).toHaveBeenCalledTimes(1); }); + it('sends a message queued while another one is on its way', async () => { + let release!: () => void; + runQueuedSend.mockImplementationOnce(() => new Promise((r) => { release = () => r({ kind: 'sent', result: {} }); })); + useNetworkStore.setState({ online: true, connected: true }); + useOutboxStore.getState().enqueueSend(payload('first')); + await flushed(); + expect(runQueuedSend).toHaveBeenCalledTimes(1); + useOutboxStore.getState().enqueueSend(payload('second')); + release(); + await vi.waitFor(() => expect(useOutboxStore.getState().sends).toEqual([])); + expect(runQueuedSend.mock.calls.map(([s]) => s.payload.outgoing.subject)).toEqual(['first', 'second']); + }); + + it('sends a message retried while another one is on its way', async () => { + useNetworkStore.setState({ online: true, connected: true }); + runQueuedSend.mockRejectedValueOnce(new Error('Invalid envelope')); + const failed = useOutboxStore.getState().enqueueSend(payload('failed'))!; + await vi.waitFor(() => expect(useOutboxStore.getState().sends[0]?.status).toBe('failed')); + + let release!: () => void; + runQueuedSend.mockImplementationOnce(() => new Promise((r) => { release = () => r({ kind: 'sent', result: {} }); })); + useOutboxStore.getState().enqueueSend(payload('other')); + await flushed(); + expect(runQueuedSend).toHaveBeenCalledTimes(2); + const retried = useOutboxStore.getState().retrySend(failed); + release(); + await retried; + await vi.waitFor(() => expect(useOutboxStore.getState().sends).toEqual([])); + expect(runQueuedSend.mock.calls.map(([s]) => s.payload.outgoing.subject)).toEqual(['failed', 'other', 'failed']); + }); + it('parks a refused message with its reason and leaves it until the user tries again', async () => { useNetworkStore.setState({ online: true, connected: true }); runQueuedSend.mockRejectedValueOnce(new Error('Invalid envelope')); diff --git a/src/stores/outbox-store.ts b/src/stores/outbox-store.ts index ab4c7350..a052bb2c 100644 --- a/src/stores/outbox-store.ts +++ b/src/stores/outbox-store.ts @@ -258,6 +258,12 @@ function opFamily(op: OutboxOp): string { let retryTimer: ReturnType | null = null; let retryAttempt = 0; +/** + * A flush was asked for while one ran (a message queued or retried then): + * the running one works from lists it took when it started, so another + * follows it. + */ +let flushAgain = false; // Reads of a bucket from storage. A read still running when the account // changes or its queue is cleared must not put the old entries back, so @@ -436,7 +442,11 @@ export const useOutboxStore = create((set, get) => ({ pendingForEmail: (emailId) => get().entries.filter((e) => e.op.emailId === emailId), flush: async () => { - if (get().flushing || get().paused || !get().resumed || !get().hydrated) return; + if (get().flushing) { + flushAgain = true; + return; + } + if (get().paused || !get().resumed || !get().hydrated) return; const accountId = get().activeAccountId; if (!accountId) return; if (get().entries.length === 0 && !hasWaitingSends()) return; @@ -445,6 +455,7 @@ export const useOutboxStore = create((set, get) => ({ if (!useNetworkStore.getState().online) return; set({ flushing: true }); + flushAgain = false; let brokeTransient = false; let dropped = false; // An entry that took on a later edit while its op was out runs again. @@ -509,7 +520,7 @@ export const useOutboxStore = create((set, get) => ({ set({ flushing: false }); } if (brokeTransient && (get().entries.length > 0 || hasWaitingSends())) scheduleRetry(); - else if (changedInFlight) void get().flush(); + else if (changedInFlight || flushAgain) void get().flush(); if (dropped) void refreshAfterDrop(); },