fix: wait 30 seconds before looking up a send, bound the waits at a start, and keep opened files only as the chosen scope allows

This commit is contained in:
Linus Rath
2026-11-03 09:00:52 +01:00
parent 62a47434e3
commit 18efb9f4c3
13 changed files with 178 additions and 16 deletions
-1
View File
@@ -7,7 +7,6 @@ import { useConfirmDialog } from '@/hooks/use-confirm-dialog';
import { addSignOutGate } from '@/lib/sign-out-cleanup';
import { reloadToNewBuild } from '@/lib/app-update';
import { confirmLoss, setOfflineNotifier, startOfflineMode, unsentCount, type OfflineNotice } from '@/lib/offline/offline-mode';
import { useOfflineStore } from '@/lib/offline/state';
import { toast } from '@/stores/toast-store';
/**
+2
View File
@@ -264,6 +264,8 @@ describe('a send whose response got lost', () => {
const entry = (await readQueue('acc')).outbox[0];
await store.area('acc', 'outbox.claim', { id: entry.id });
await store.area('acc', 'outbox.update', { id: entry.id, status: 'uncertain' });
// The wait before a lookup has passed.
store.dbs.get('acc')!.run('UPDATE outbox SET last_attempt_at = ?', [Date.now() - 31_000]);
return entry;
}
+70 -2
View File
@@ -27,13 +27,14 @@ import {
unsentCount,
wipeEverything,
} from '../offline-mode';
import { flushQueue, settleStaleSends, SENDING_STALE_MS } from '../flush';
import { flushQueue, LOOKUP_WAIT_MS, settleStaleSends, SENDING_STALE_MS } from '../flush';
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';
import { shellFiles } from '../shell';
import { isOfflineNow, OFFLINE_ACCOUNTS_KEY, OFFLINE_EVICTED_KEY, setServerDown, useOfflineStore, writeEnabled } from '../state';
import { DEFAULT_SCOPE } from '@/lib/store/area';
import { isOfflineNow, OFFLINE_ACCOUNTS_KEY, OFFLINE_EVICTED_KEY, OFFLINE_SCOPE_KEY, offlineScopeFor, setServerDown, useOfflineStore, writeEnabled } from '../state';
function fakeClient(server = 'https://mail.example.org', overrides: Record<string, unknown> = {}) {
return {
@@ -288,6 +289,41 @@ describe('11: starting from the device', () => {
expect(await offlineSessionFor('acc')).toMatchObject({ primaryAccounts: { 'urn:ietf:params:jmap:mail': JMAP_ACCOUNT } });
});
it('gives up on a policy fetch that never answers', async () => {
writeEnabled(['acc']);
usePolicyStore.setState({ loaded: false, fetching: false, source: 'none' });
vi.useFakeTimers();
vi.stubGlobal('fetch', vi.fn(() => new Promise(() => {})));
try {
const session = offlineSessionFor('acc');
await vi.advanceTimersByTimeAsync(10_000);
expect(await session).toBeNull();
} finally {
vi.unstubAllGlobals();
vi.useRealTimers();
}
});
it('doesn\'t wait for the old database to finish moving', async () => {
writeEnabled(['acc']);
forgetLegacyMove();
const legacy = indexedDB.open;
vi.spyOn(indexedDB, 'open').mockImplementation(((name: string, version?: number) => {
// A database that never opens.
if (name === 'bulwark-offline') return { result: null } as never;
return legacy.call(indexedDB, name, version);
}) as never);
vi.useFakeTimers();
try {
const session = offlineSessionFor('acc');
await vi.advanceTimersByTimeAsync(10_000);
expect(await session).toMatchObject({ primaryAccounts: { 'urn:ietf:params:jmap:mail': JMAP_ACCOUNT } });
} finally {
vi.useRealTimers();
vi.restoreAllMocks();
}
});
it('doesn\'t offer a quick start before the store has its folders', async () => {
store = new InProcessStore(new FakeJmapMail());
setStoreBackend(store);
@@ -430,6 +466,38 @@ describe('a send the server couldn\'t take right now', () => {
});
});
describe('looking up a send that may have gone', () => {
it('waits until about 30 seconds after the last try', async () => {
writeEnabled(['acc']);
const id = await queueSend('acc', { subject: 'Hello', to: [] }, { kind: 'email', args: [] }, undefined, 'fixed@example.org');
await claimForSending('acc', id);
await updateOutboxEntry({ account: 'acc', id }, { status: 'uncertain' });
const requestNetworkOnly = vi.fn(async () => { throw new TypeError('Failed to fetch'); });
const client = fakeClient('https://mail.example.org', { requestNetworkOnly, resolveSendAccountId: async () => 'j1', submissionAccountFor: () => 'j1' }) as unknown as JMAPClient;
const retryLater = vi.fn();
await flushQueue('acc', client, { notify: vi.fn(), allowed: () => true, retryLater });
expect(requestNetworkOnly).not.toHaveBeenCalled();
expect(retryLater).toHaveBeenCalled();
const later = Date.now() + LOOKUP_WAIT_MS + 1000;
const clock = vi.spyOn(Date, 'now').mockReturnValue(later);
try {
await flushQueue('acc', client, { notify: vi.fn(), allowed: () => true, retryLater });
expect(requestNetworkOnly).toHaveBeenCalled();
} finally {
clock.mockRestore();
}
});
});
describe('what the store is told to keep', () => {
it('follows what the account chose, on top of the defaults', () => {
expect(offlineScopeFor('acc')).toEqual(DEFAULT_SCOPE);
localStorage.setItem(OFFLINE_SCOPE_KEY, JSON.stringify({ acc: { attachments: false, days: 7 } }));
expect(offlineScopeFor('acc')).toEqual({ ...DEFAULT_SCOPE, attachments: false, days: 7 });
expect(offlineScopeFor('other')).toEqual(DEFAULT_SCOPE);
});
});
describe('a send on the local copy', () => {
it('queues, before the swap too, while the probes say the server is up', () => {
writeEnabled(['acc']);
+7
View File
@@ -58,6 +58,9 @@ export interface FlushDeps {
retryLater?: () => void;
}
/** A message that just went may not show in a lookup yet, which would send it twice. */
export const LOOKUP_WAIT_MS = 30_000;
/** A send is taken for at most this long before a closed tab's send counts as uncertain. */
export const SENDING_STALE_MS = 60_000;
@@ -225,6 +228,10 @@ async function settleUncertain(account: string, client: JMAPClient, deps: FlushD
for (const entry of outbox) {
if (entry.status !== 'uncertain') continue;
if (stillOffline() || !deps.allowed()) return false;
if (entry.lastAttemptAt && Date.now() - entry.lastAttemptAt < LOOKUP_WAIT_MS) {
deps.retryLater?.();
continue;
}
let decision: RetryDecision;
try {
decision = await lookUp(client, entry);
+16 -7
View File
@@ -54,8 +54,8 @@ import {
type PolicyFingerprint,
} from './queue';
import { clearHandoffs } from '@/lib/launch-handoff';
import { clearEvicted, EVICTED_KEEP_MS, forgetMailSnapshot, isEvicted, isOfflineNow, OFFLINE_ACCOUNTS_KEY, readEnabledFrom, readEvicted, visibleAccounts, serverKey, setServerDown, useOfflineStore, writeEnabled, type OutboxRow } from './state';
import { flushQueue, settleStaleSends } from './flush';
import { clearEvicted, EVICTED_KEEP_MS, forgetMailSnapshot, isEvicted, offlineScopeFor, isOfflineNow, OFFLINE_ACCOUNTS_KEY, readEnabledFrom, readEvicted, visibleAccounts, serverKey, setServerDown, useOfflineStore, writeEnabled, type OutboxRow } from './state';
import { flushQueue, LOOKUP_WAIT_MS, settleStaleSends } from './flush';
import { LEGACY_DB_NAME, moveLegacyOutbox } from './migrate-legacy';
import { deleteShellCaches, tellServiceWorker } from './shell';
import { rememberLost } from './lost';
@@ -159,6 +159,7 @@ const storeHooks: StorePageHooks = {
session: (client?.getSessionSnapshot() as StoreSession | undefined) ?? null,
auth: client?.getAuthHeader() ?? null,
invertedKeywordSort: client?.keywordSortInverted() ?? null,
scope: offlineScopeFor(account),
};
},
async refreshAuth(account) {
@@ -571,7 +572,7 @@ export function scheduleFlushAll(delayMs = 1_000): void {
/** How long after a failed policy fetch the outbox asks again. */
const POLICY_RETRY_MS = 60_000;
/** A lost response with the server still there: nothing else would start the lookup. */
const LOOKUP_AGAIN_MS = 30_000;
const LOOKUP_AGAIN_MS = LOOKUP_WAIT_MS + 1_000;
/** A send held by the server for undo, from the outbox: the undo toast as after any send. */
function offerUndo(entry: OutboxEntry, result: SendEmailResult): void {
@@ -1021,16 +1022,24 @@ export async function serverUnreachable(serverUrl: string): Promise<boolean> {
const SESSION_READ_TIMEOUT_MS = 5_000;
/** The session kept for an account, for starting from the local copy. */
/** A start doesn't hang on a black-holed fetch or a stuck database. */
const BOOT_WAIT_MS = 3_000;
function withinBootWait(work: Promise<unknown>): Promise<unknown> {
return Promise.race([work.catch(() => {}), new Promise((resolve) => setTimeout(resolve, BOOT_WAIT_MS))]);
}
export async function offlineSessionFor(account: string, options: { withFolders?: boolean } = {}): Promise<unknown | null> {
// The organisation's last word: the policy kept from the last fetch (the
// fetch itself fails without a network), never the defaults.
const policy = usePolicyStore.getState();
if (!policy.loaded && !policy.fetching) await policy.fetchPolicy();
else await whenPolicySettled(3000);
if (!offlineActive(account)) return null;
await withinBootWait(!policy.loaded && !policy.fetching ? policy.fetchPolicy() : whenPolicySettled(BOOT_WAIT_MS));
// Still nothing to go on: the usual start, which has its own errors.
if (!usePolicyStore.getState().loaded || !offlineActive(account)) return null;
registerOfflineHooks();
try {
await moveLegacyOutbox();
// What's in the old database goes on moving behind the start.
await withinBootWait(moveLegacyOutbox());
return await Promise.race([
(async () => {
if (options.withFolders && !(await getStore().area<boolean>(account, 'mail.hasFolders'))) return null;
+13
View File
@@ -7,6 +7,7 @@
*/
import { create } from 'zustand';
import { DEFAULT_SCOPE, type StoreScope } from '@/lib/store/area';
import type { ConflictEntry, OutboxEntry, OutboxError } from './queue';
/** localStorage: the local account ids that keep mail on this device. */
@@ -74,6 +75,18 @@ export function writeEnabled(enabled: string[]): void {
}
}
/** localStorage: what each account chose to keep offline, on top of the defaults. */
export const OFFLINE_SCOPE_KEY = 'bulwark-offline-scope';
export function offlineScopeFor(account: string): StoreScope {
try {
const chosen = (JSON.parse(localStorage.getItem(OFFLINE_SCOPE_KEY) ?? '{}') as Record<string, Partial<StoreScope>>)[account];
return { ...DEFAULT_SCOPE, ...chosen };
} catch {
return DEFAULT_SCOPE;
}
}
/** localStorage: accounts the app dropped by itself, and when. Their data waits for a new sign-in. */
export const OFFLINE_EVICTED_KEY = 'bulwark-offline-evicted';
/** How long that data waits. */
+20
View File
@@ -471,6 +471,26 @@ describe('attachments', () => {
expect(refused.downloads).toEqual(['b-gone']);
});
it('files a blob the user opened only as far as the chosen scope allows', async () => {
const server = new Files();
server.addEmail({ id: 'm', attachments: [{ blobId: 'b-kept', size: 10, type: 'image/png' }] });
const data = new Uint8Array(10);
const record = { jmap: JMAP_ACCOUNT, blobId: 'b-kept', type: 'image/png', data };
const none = engineFor(server, { ...EVERYTHING, attachments: false });
await none.engine.requestSync();
expect(await none.engine.areaCall('mail.recordBlob', record)).toBe(false);
expect(await none.engine.areaCall('mail.getBlob', { jmap: JMAP_ACCOUNT, blobId: 'b-kept' })).toBeNull();
const small = engineFor(server, { ...EVERYTHING, attachments: true, maxAttachmentBytes: 9 });
await small.engine.requestSync();
expect(await small.engine.areaCall('mail.recordBlob', record)).toBe(false);
const fits = engineFor(server, { ...EVERYTHING, attachments: true, maxAttachmentBytes: 10 });
await fits.engine.requestSync();
expect(await fits.engine.areaCall('mail.recordBlob', record)).toBe(true);
});
it('files a blob the user opened only under a kept message: never a stray file', async () => {
const server = new Files();
server.addEmail({ id: 'm', attachments: [{ blobId: 'b-kept', size: 99_999_999, type: 'image/png' }] });
+20 -4
View File
@@ -34,7 +34,7 @@ function setup() {
transport: server,
areas: [mailArea, outboxArea],
scope: { ...DEFAULT_SCOPE, days: null },
now: () => NOW,
now: () => clock,
sleep: async () => {},
});
return { server, db, engine };
@@ -43,6 +43,7 @@ function setup() {
let t: ReturnType<typeof setup>;
beforeEach(async () => {
clock = NOW;
t = setup();
await t.engine.requestSync();
});
@@ -57,6 +58,8 @@ async function local(id: string) {
return answer?.methodResponses[0][1].list[0];
}
let clock = NOW;
async function boxes() {
const answer = await t.engine.answer([['Mailbox/get', {}, 'm']]);
return Object.fromEntries((answer?.methodResponses[0][1].list as Array<{ id: string; totalEmails: number; unreadEmails: number }>).map((box) => [box.id, [box.totalEmails, box.unreadEmails]]));
@@ -430,7 +433,7 @@ describe('sends', () => {
expect(await t.engine.areaCall('outbox.claim', { id })).toBe(true);
expect(await t.engine.areaCall('outbox.claim', { id })).toBe(false);
const entry = (await read()).entries[0];
expect(entry).toMatchObject({ status: 'sending', sendingSince: NOW, firstAttemptAt: NOW, messageId: 'fixed@example.org', item });
expect(entry).toMatchObject({ status: 'sending', sendingSince: NOW, lastAttemptAt: NOW, messageId: 'fixed@example.org', item });
});
it('marks what a closed tab left behind uncertain', async () => {
@@ -439,15 +442,28 @@ describe('sends', () => {
await t.engine.areaCall('outbox.settleStale', { before: NOW });
expect((await read()).entries[0].status).toBe('sending');
await t.engine.areaCall('outbox.settleStale', { before: NOW + 1 });
expect((await read()).entries[0].status).toBe('uncertain');
expect((await read()).entries[0]).toMatchObject({ status: 'uncertain', firstAttemptAt: NOW });
});
it('keeps the first attempt time across retries, so the lookup covers the first copy', async () => {
const id = await t.engine.areaCall('outbox.queueSend', { summary, item }) as number;
await t.engine.areaCall('outbox.claim', { id });
await t.engine.areaCall('outbox.update', { id, status: 'uncertain' });
expect((await read()).entries[0].firstAttemptAt).toBe(NOW);
clock = NOW + 60_000;
await t.engine.areaCall('outbox.update', { id, status: 'waiting' });
await t.engine.areaCall('outbox.claim', { id });
expect((await read()).entries[0].firstAttemptAt).toBe(NOW);
await t.engine.areaCall('outbox.update', { id, status: 'uncertain' });
const entry = (await read()).entries[0];
expect(entry.firstAttemptAt).toBe(NOW);
expect(entry.lastAttemptAt).toBe(NOW + 60_000);
});
it('doesn\'t call a try that never started an attempt', async () => {
const id = await t.engine.areaCall('outbox.queueSend', { summary, item }) as number;
await t.engine.areaCall('outbox.claim', { id });
await t.engine.areaCall('outbox.update', { id, status: 'waiting' });
expect((await read()).entries[0].firstAttemptAt).toBeUndefined();
});
it('keeps order across changes and sends, and a later merge moves a change later', async () => {
+10
View File
@@ -55,6 +55,8 @@ class PageStore implements StoreBackend {
private readonly opening = new Map<string, Promise<void>>();
/** The engine can't start in this browser: don't keep every caller waiting for an owner. */
private broken = false;
/** What the owner's engine was last told to keep. */
private readonly sentScope = new Map<string, string>();
constructor(private readonly hooks: () => StorePageHooks | null) {}
@@ -90,6 +92,7 @@ class PageStore implements StoreBackend {
private reset(): void {
this.client = null;
this.sentScope.clear();
this.wanted.clear();
this.opening.clear();
this.broken = false;
@@ -103,6 +106,7 @@ class PageStore implements StoreBackend {
const client = this.started();
const { keyHex, record } = await openAccountKey(account, this.hooks()?.keyLayers?.() ?? []);
const inputs = this.hooks()?.inputs(account) ?? { session: null, auth: null, invertedKeywordSort: null };
if (inputs.scope) this.sentScope.set(account, JSON.stringify(inputs.scope));
await client.openAccount({
account,
file: record.file,
@@ -132,6 +136,12 @@ class PageStore implements StoreBackend {
if (!inputs) return;
if (inputs.session) client.setSession(account, inputs.session);
client.setAuth(account, inputs.auth);
// Only a real change: setting it starts over what a full disk made the store give up on.
const scope = inputs.scope ? JSON.stringify(inputs.scope) : null;
if (inputs.scope && scope !== this.sentScope.get(account)) {
if (this.sentScope.has(account)) client.setScope(account, inputs.scope);
this.sentScope.set(account, scope!);
}
if (inputs.invertedKeywordSort !== null) client.setInvertedKeywordSort(account, inputs.invertedKeywordSort);
}
+1
View File
@@ -34,6 +34,7 @@ function recordOpened(ctx: StoreContext, params: { jmap: string; emails: EmailOb
/** A file of a kept message the user opened. Nothing else is kept: not an export, not a stray file. */
function recordBlob(ctx: StoreContext, params: { jmap: string; blobId: string; type: string | null; data: Uint8Array }): boolean {
if (!ctx.scope.attachments || params.data.byteLength > ctx.scope.maxAttachmentBytes) return false;
const owner = ctx.db.get<{ received_at: number }>('SELECT received_at FROM email_part WHERE jmap = ? AND blob_id = ?', [params.jmap, params.blobId]);
if (!owner) return false;
putBlob(ctx.db, params.jmap, params.blobId, params.type, params.data, Number(owner.received_at));
+8 -2
View File
@@ -27,6 +27,7 @@ interface OutboxRow {
status: string;
sending_since: number | null;
first_attempt_at: number | null;
last_attempt_at: number | null;
payload: string;
summary: string;
message_id: string | null;
@@ -46,6 +47,7 @@ function toEntry(row: OutboxRow): StoredOutboxEntry {
...(payload.fingerprint ? { fingerprint: payload.fingerprint } : {}),
...(row.sending_since ? { sendingSince: Number(row.sending_since) } : {}),
...(row.first_attempt_at ? { firstAttemptAt: Number(row.first_attempt_at) } : {}),
...(row.last_attempt_at ? { lastAttemptAt: Number(row.last_attempt_at) } : {}),
...(row.error ? { error: JSON.parse(row.error) as OutboxError } : {}),
...(row.message_id ? { messageId: row.message_id } : {}),
...(row.copy_id ? { copyId: row.copy_id } : {}),
@@ -101,7 +103,7 @@ function read(ctx: StoreContext): { patches: number; entries: StoredOutboxEntry[
function claim(ctx: StoreContext, params: { id: number }): boolean {
const now = ctx.now();
return ctx.db.run(
`UPDATE outbox SET status = 'sending', sending_since = ?, first_attempt_at = COALESCE(first_attempt_at, ?) WHERE id = ? AND status = 'waiting'`,
`UPDATE outbox SET status = 'sending', sending_since = ?, last_attempt_at = ? WHERE id = ? AND status = 'waiting'`,
[now, now, params.id],
).changes === 1;
}
@@ -110,11 +112,14 @@ function claim(ctx: StoreContext, params: { id: number }): boolean {
function update(ctx: StoreContext, params: { id: number; status: OutboxStatus; error?: OutboxError; copyId?: string | null }): void {
ctx.db.run(
`UPDATE outbox SET status = ?, error = ?, sending_since = NULL,
first_attempt_at = CASE WHEN ? THEN COALESCE(first_attempt_at, last_attempt_at) ELSE first_attempt_at END,
copy_id = CASE WHEN ? THEN ? ELSE copy_id END
WHERE id = ?`,
[
params.status,
params.error ? JSON.stringify(params.error) : null,
// A try that may have reached the server is what a later lookup has to cover.
params.status === 'uncertain' ? 1 : 0,
params.copyId === undefined ? 0 : 1,
params.copyId ?? null,
params.id,
@@ -125,7 +130,8 @@ function update(ctx: StoreContext, params: { id: number; status: OutboxStatus; e
/** A send a closed tab left behind may have gone out. */
function settleStale(ctx: StoreContext, params: { before: number }): number {
return ctx.db.run(
`UPDATE outbox SET status = 'uncertain', sending_since = NULL WHERE status = 'sending' AND COALESCE(sending_since, 0) < ?`,
`UPDATE outbox SET status = 'uncertain', sending_since = NULL, first_attempt_at = COALESCE(first_attempt_at, last_attempt_at)
WHERE status = 'sending' AND COALESCE(sending_since, 0) < ?`,
[params.before],
).changes;
}
+2
View File
@@ -62,7 +62,9 @@ export interface StoredOutboxEntry {
summary: OutboxSummary;
status: OutboxStatus;
sendingSince?: number;
/** When the first try that may have reached the server began. */
firstAttemptAt?: number;
lastAttemptAt?: number;
error?: OutboxError;
item: SendItem;
fingerprint?: PolicyFingerprint;
+9
View File
@@ -1,6 +1,15 @@
import type { Migration } from '../storage';
export const OUTBOX_MIGRATIONS: readonly Migration[] = [
{
id: 201,
name: 'outbox_last_attempt',
sql: `
-- When the latest try began. It becomes first_attempt_at only once a try
-- may have reached the server.
ALTER TABLE outbox ADD COLUMN last_attempt_at INTEGER;
`,
},
{
id: 200,
name: 'outbox',