From 4ee918a98fbd0b355fd7a19081893a479f76ae87 Mon Sep 17 00:00:00 2001 From: Linus Rath <139418639+rathlinus@users.noreply.github.com> Date: Tue, 3 Nov 2026 05:46:55 +0100 Subject: [PATCH] fix: list a large folder off its index and sync message bodies in bigger batches --- lib/store/__tests__/mail-sync.test.ts | 46 +++++++++++++++++++ lib/store/adapters/wasm.ts | 2 + lib/store/mail/methods.ts | 29 +++++++++--- lib/store/mail/schema.ts | 4 +- lib/store/mail/sync.ts | 63 +++++++++++++++++---------- lib/store/storage.ts | 8 +++- 6 files changed, 120 insertions(+), 32 deletions(-) diff --git a/lib/store/__tests__/mail-sync.test.ts b/lib/store/__tests__/mail-sync.test.ts index 80e1c69fe..a5ddae149 100644 --- a/lib/store/__tests__/mail-sync.test.ts +++ b/lib/store/__tests__/mail-sync.test.ts @@ -6,6 +6,8 @@ import path from 'node:path'; import { DEFAULT_SCOPE, type StoreContext, type StoreEvent } from '../area'; import { AccountEngine } from '../engine/account-engine'; import { runChangesLoop } from '../engine/changes'; +import { answerRequest, buildMethodTable } from '../engine/responder'; +import type { SqlValue } from '../storage'; import { TransportError, type MethodCall } from '../engine/transport'; import { mailArea } from '../mail'; import { FAKE_SESSION, FakeJmap, JMAP_ACCOUNT } from './helpers/fake-jmap'; @@ -320,6 +322,50 @@ describe('when the server is busy or gone', () => { }); }); +describe('query plans', () => { + it('reads a newest-first folder page off the index, without sorting the folder', async () => { + const server = new FakeJmap(); + for (let i = 0; i < 30; i += 1) server.addEmail(); + const db = openMemoryStore(); + const engine = engineFor(server, db); + await engine.requestSync(); + const seen: string[] = []; + const spied = { + ...db, + all: (sql: string, params?: readonly SqlValue[]) => { + seen.push(sql); + return db.all(sql, params); + }, + }; + const answer = await answerRequest( + { account: 'a', db: spied as typeof db, session: FAKE_SESSION, scope: EVERYTHING, now: Date.now, emit: () => {}, defaultJmap: JMAP_ACCOUNT, invertedKeywordSort: false }, + buildMethodTable([mailArea]), + [['Email/query', { filter: INBOX, sort: [{ property: 'receivedAt', isAscending: false }], limit: 5 }, 'q']], + ); + expect(answer?.methodResponses[0][1].ids).toHaveLength(5); + const [sql] = seen; + const plan = db.all<{ detail: string }>(`EXPLAIN QUERY PLAN ${sql}`, [JMAP_ACCOUNT, 'inbox', 5, 0]).map((row) => row.detail).join('; '); + expect(plan).not.toContain('TEMP B-TREE'); + expect(plan).toContain('email_mailbox_list'); + // The folder condition isn't looked up again for every row. + expect(sql).not.toContain('m.mailbox_id'); + }); +}); + +describe('a failure inside a transaction', () => { + it('reports the real error even when SQLite has already rolled back', () => { + const db = openMemoryStore(); + db.exec('CREATE TABLE t (id INTEGER PRIMARY KEY)'); + expect(() => db.transaction(() => { + db.exec('ROLLBACK'); + throw new Error('disk is full'); + })).toThrow('disk is full'); + // And the connection still works. + db.transaction(() => db.run('INSERT INTO t (id) VALUES (1)')); + expect(db.all('SELECT id FROM t')).toHaveLength(1); + }); +}); + describe('calls into an area', () => { const noteArea = { name: 'note', diff --git a/lib/store/adapters/wasm.ts b/lib/store/adapters/wasm.ts index e85bf14f4..2af1373fe 100644 --- a/lib/store/adapters/wasm.ts +++ b/lib/store/adapters/wasm.ts @@ -165,6 +165,8 @@ export class WasmStorage { const pageSize = Number(db.selectValue('PRAGMA page_size')); if (pageSize !== REQUIRED_PAGE_SIZE) throw new Error(`Refusing an unsafe page size: ${pageSize}`); db.exec('PRAGMA temp_store = MEMORY'); + // Every page read is a decryption, so a miss costs more than usual. + db.exec('PRAGMA cache_size = -65536'); db.exec('PRAGMA foreign_keys = OFF'); db.exec('PRAGMA synchronous = NORMAL'); } catch (error) { diff --git a/lib/store/mail/methods.ts b/lib/store/mail/methods.ts index e91bdd45c..65493b7ef 100644 --- a/lib/store/mail/methods.ts +++ b/lib/store/mail/methods.ts @@ -15,10 +15,14 @@ function jmapOf(ctx: MethodContext, args: Record): string { return (args.accountId as string | undefined) ?? ctx.defaultJmap; } -function filterSql(filter: Filter, jmap: string): Sql { +/** + * `drivenBy` is the folder the query already starts from: repeating that + * condition would look the membership up again for every row. + */ +function filterSql(filter: Filter, jmap: string, drivenBy: string | null = null): Sql { if (typeof filter.operator === 'string') { const conditions = Array.isArray(filter.conditions) ? filter.conditions as Filter[] : []; - const parts = conditions.map((c) => filterSql(c, jmap)); + const parts = conditions.map((c) => filterSql(c, jmap, filter.operator === 'AND' ? drivenBy : null)); const params = parts.flatMap((p) => p.params); const joined = (joiner: string) => parts.map((p) => `(${p.sql})`).join(joiner); if (filter.operator === 'AND') return { sql: parts.length ? joined(' AND ') : '1', params }; @@ -31,6 +35,7 @@ function filterSql(filter: Filter, jmap: string): Sql { for (const [key, value] of Object.entries(filter)) { switch (key) { case 'inMailbox': + if (value === drivenBy) break; conditions.push('EXISTS (SELECT 1 FROM email_mailbox m WHERE m.email_rowid = e.rowid AND m.mailbox_id = ?)'); params.push(String(value)); break; @@ -87,6 +92,15 @@ function filterSql(filter: Filter, jmap: string): Sql { interface Comparator { property?: string; isAscending?: boolean; keyword?: string } +/** + * Newest first reads straight off an index. Any other order has to look at + * every row of the folder, which in a large mailbox takes seconds. + */ +function isNewestFirst(sort: Comparator[]): boolean { + if (sort.length === 0) return true; + return sort.length === 1 && sort[0].property === 'receivedAt' && sort[0].isAscending === false; +} + function orderSql(ctx: MethodContext, sort: Comparator[]): Sql { const levels = sort.length > 0 ? sort : [{ property: 'receivedAt', isAscending: false }]; const parts: string[] = []; @@ -134,9 +148,12 @@ const emailQuery: MethodHandler = (ctx, args) => { const jmap = jmapOf(ctx, args); const filter = (args.filter ?? undefined) as Filter | undefined; if (args.anchor !== undefined) throw new MethodError('unsupportedFilter', 'anchor needs the server'); - const where = filter ? filterSql(filter, jmap) : { sql: '1', params: [] }; - const order = orderSql(ctx, (args.sort as Comparator[]) ?? []); const box = drivingMailbox(filter); + const where = filter ? filterSql(filter, jmap, box) : { sql: '1', params: [] }; + const sort = (args.sort as Comparator[]) ?? []; + const order: Sql = !isNewestFirst(sort) + ? orderSql(ctx, sort) + : { sql: box ? 'd.received_at DESC, d.email_rowid DESC' : 'e.received_at DESC, e.rowid DESC', params: [] }; // Starting from the folder's own index keeps small folders fast in a large account. const from = box ? 'email_mailbox d INDEXED BY email_mailbox_list CROSS JOIN email e ON e.rowid = d.email_rowid' @@ -162,8 +179,10 @@ const emailQuery: MethodHandler = (ctx, args) => { }; if (args.calculateTotal) { const count = collapse ? 'COUNT(DISTINCT COALESCE(e.thread_id, e.id))' : 'COUNT(*)'; + // A plain folder counts from its index alone, without reading a message. + const countFrom = box && !collapse && where.sql === '1' ? 'email_mailbox d INDEXED BY email_mailbox_list' : from; result.total = Number(ctx.db.get<{ n: number }>( - `SELECT ${count} AS n FROM ${from} WHERE ${base.sql} AND (${where.sql})`, + `SELECT ${count} AS n FROM ${countFrom} WHERE ${base.sql} AND (${where.sql})`, [...base.params, ...where.params], )?.n ?? 0); } diff --git a/lib/store/mail/schema.ts b/lib/store/mail/schema.ts index 52ef16dde..98f29576d 100644 --- a/lib/store/mail/schema.ts +++ b/lib/store/mail/schema.ts @@ -23,7 +23,7 @@ export const MAIL_MIGRATIONS: readonly Migration[] = [ gen INTEGER NOT NULL DEFAULT 0, UNIQUE (jmap, id) ); - CREATE INDEX email_received ON email (jmap, received_at DESC); + CREATE INDEX email_received ON email (jmap, received_at DESC, rowid DESC); CREATE INDEX email_thread ON email (jmap, thread_id); CREATE INDEX email_backfill ON email (jmap, body_state, received_at DESC); CREATE TABLE email_mailbox ( @@ -33,7 +33,7 @@ export const MAIL_MIGRATIONS: readonly Migration[] = [ received_at INTEGER NOT NULL, PRIMARY KEY (email_rowid, mailbox_id) ) WITHOUT ROWID; - CREATE INDEX email_mailbox_list ON email_mailbox (jmap, mailbox_id, received_at DESC); + CREATE INDEX email_mailbox_list ON email_mailbox (jmap, mailbox_id, received_at DESC, email_rowid DESC); CREATE TABLE email_keyword ( email_rowid INTEGER NOT NULL REFERENCES email (rowid) ON DELETE CASCADE, keyword TEXT NOT NULL, diff --git a/lib/store/mail/sync.ts b/lib/store/mail/sync.ts index 21ad9bb0d..30b8c7b7b 100644 --- a/lib/store/mail/sync.ts +++ b/lib/store/mail/sync.ts @@ -26,6 +26,8 @@ const PAGE = 500; const PAIRS_PER_REQUEST = 4; const BODY_BATCH_IDS = 100; const BODY_BATCH_BYTES = 8 * 1024 * 1024; +/** Bodies fetched before one commit. Every commit writes search index segments, so fewer is much cheaper. */ +const BODY_REQUESTS_PER_COMMIT = 5; const MAX_BODY_VALUE_BYTES = 1_000_000; const DELTA_BATCH = 500; @@ -83,11 +85,15 @@ async function syncMailboxes(ctx: SyncContext, jmap: string): Promise { interface ListCursor { position: number; gen: number; captured: string } +/** The generation of the last full listing. One row is enough: every row of a pass has it or an older one. */ +function currentGen(ctx: SyncContext, jmap: string): number { + return Number(ctx.db.get<{ gen: number }>('SELECT gen FROM email WHERE jmap = ? ORDER BY gen DESC LIMIT 1', [jmap])?.gen ?? 0); +} + async function startFullList(ctx: SyncContext, jmap: string): Promise { const response = await ctx.transport.request([['Email/get', { accountId: jmap, ids: [], properties: ['id'] }, 's']], MAIL_USING); const captured = String(responseFor(response, 's')[1].state); - const gen = (ctx.db.get<{ gen: number | null }>('SELECT MAX(gen) AS gen FROM email WHERE jmap = ?', [jmap])?.gen ?? 0) + 1; - const cursor: ListCursor = { position: 0, gen: Number(gen), captured }; + const cursor: ListCursor = { position: 0, gen: currentGen(ctx, jmap) + 1, captured }; ctx.db.transaction(() => ctx.setSyncState(jmap, 'Email', { state: null, phase: 'list', cursor: { ...cursor } })); return cursor; } @@ -171,11 +177,11 @@ async function syncEmailDelta(ctx: SyncContext, jmap: string, since: string, tar const created = new Set(page.created); const newIds: string[] = []; ctx.db.transaction(() => { + const gen = fetched.length > 0 ? currentGen(ctx, jmap) : 0; for (const email of fetched) { const known = ctx.db.get('SELECT 1 AS x FROM email WHERE jmap = ? AND id = ?', [jmap, email.id]); // An old message changing doesn't pull it into the window. if (!known && !inWindow(ctx, email)) continue; - const gen = Number(ctx.db.get<{ gen: number | null }>('SELECT MAX(gen) AS gen FROM email WHERE jmap = ?', [jmap])?.gen ?? 0); upsertEmail(ctx.db, jmap, email, gen); if (!known && created.has(email.id)) newIds.push(email.id); } @@ -254,39 +260,48 @@ async function backfillBodies(ctx: SyncContext, jmap: string): Promise { let done = 0; const totalRow = ctx.db.get<{ n: number }>('SELECT COUNT(*) AS n FROM email WHERE jmap = ? AND body_state = 0', [jmap]); const total = Number(totalRow?.n ?? 0); + if (total === 0) return; for (;;) { if (ctx.stopped()) return; const candidates = ctx.db.all<{ rowid: number; id: string; size: number; json: string }>( 'SELECT rowid, id, size, json FROM email WHERE jmap = ? AND body_state = 0 ORDER BY received_at DESC LIMIT ?', - [jmap, BODY_BATCH_IDS], + [jmap, BODY_BATCH_IDS * BODY_REQUESTS_PER_COMMIT], ); if (candidates.length === 0) return; - let bytes = 0; - const batch: typeof candidates = []; - for (const row of candidates) { - if (batch.length > 0 && bytes + Number(row.size) > BODY_BATCH_BYTES) break; - bytes += Number(row.size); - batch.push(row); + const fetched = new Map(); + const asked: typeof candidates = []; + let next = 0; + for (let request = 0; request < BODY_REQUESTS_PER_COMMIT && next < candidates.length; request += 1) { + let bytes = 0; + const batch: typeof candidates = []; + while (next < candidates.length && batch.length < BODY_BATCH_IDS) { + const row = candidates[next]; + if (batch.length > 0 && bytes + Number(row.size) > BODY_BATCH_BYTES) break; + bytes += Number(row.size); + batch.push(row); + next += 1; + } + const response = await ctx.transport.request([['Email/get', { + accountId: jmap, + ids: batch.map((row) => String(row.id)), + properties: ['id', ...EMAIL_BODY_PROPERTIES], + fetchTextBodyValues: true, + fetchHTMLBodyValues: true, + fetchAllBodyValues: true, + maxBodyValueBytes: MAX_BODY_VALUE_BYTES, + }, 'b']], MAIL_USING); + for (const email of responseFor(response, 'b')[1].list as EmailObject[]) fetched.set(email.id, email); + asked.push(...batch); + if (ctx.stopped()) break; } - const response = await ctx.transport.request([['Email/get', { - accountId: jmap, - ids: batch.map((row) => String(row.id)), - properties: ['id', ...EMAIL_BODY_PROPERTIES], - fetchTextBodyValues: true, - fetchHTMLBodyValues: true, - fetchAllBodyValues: true, - maxBodyValueBytes: MAX_BODY_VALUE_BYTES, - }, 'b']], MAIL_USING); - const list = responseFor(response, 'b')[1].list as EmailObject[]; - const byId = new Map(list.map((email) => [email.id, email])); ctx.db.transaction(() => { - for (const row of batch) { - const full = byId.get(String(row.id)); + for (const row of asked) { + const full = fetched.get(String(row.id)); if (full) storeBody(ctx.db, Number(row.rowid), String(row.json), full); else ctx.db.run('UPDATE email SET body_state = 2 WHERE rowid = ?', [row.rowid]); } }); - done += batch.length; + done += asked.length; ctx.emit({ type: 'progress', account: ctx.account, area: 'mail', phase: 'bodies', done, total }); } } diff --git a/lib/store/storage.ts b/lib/store/storage.ts index e8cde1216..5ca4c29af 100644 --- a/lib/store/storage.ts +++ b/lib/store/storage.ts @@ -39,7 +39,13 @@ export function runTransaction( exec('COMMIT'); return result; } catch (error) { - exec('ROLLBACK'); + // SQLite has already rolled back after some errors (a full disk, a failed + // write). A second failure here must not hide the real one. + try { + exec('ROLLBACK'); + } catch { + // Nothing left to roll back. + } throw error; } finally { state.depth -= 1;