mirror of
https://github.com/bulwarkmail/native.git
synced 2026-11-02 00:06:23 +00:00
fix: send a message queued or retried while the outbox is already sending
This commit is contained in:
@@ -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'));
|
||||
|
||||
@@ -258,6 +258,12 @@ function opFamily(op: OutboxOp): string {
|
||||
|
||||
let retryTimer: ReturnType<typeof setTimeout> | 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<OutboxState>((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<OutboxState>((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<OutboxState>((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();
|
||||
},
|
||||
|
||||
|
||||
Reference in New Issue
Block a user