chore: merge the store's speed work into the mail branch

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: <T extends object>(sql: string, params?: readonly SqlValue[]) => {
+        seen.push(sql);
+        return db.all<T>(sql, params);
+      },
+    };
+    const answer = await answerRequest(
+      { account: 'a', db: spied as typeof db, session: FAKE_SESSION, scope: EVERYTHING, now: Date.now, emit: () => {}, overlay: (_jmap, _type, records) => records, 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',

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) {

lib/store/mail/methods.ts

@@ -28,10 +28,14 @@ function likeTerm(value: unknown): string {
   return `%${String(value).replace(/[\\%_]/g, (c) => `\\${c}`)}%`;
 }
 
-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 };
@@ -44,6 +48,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;
@@ -118,6 +123,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[] = [];
@@ -165,9 +179,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'
@@ -194,8 +211,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);
   }

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,

lib/store/mail/sync.ts

@@ -27,6 +27,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;
 
@@ -84,11 +86,15 @@ async function syncMailboxes(ctx: SyncContext, jmap: string): Promise<void> {
 
 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<ListCursor> {
   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;
 }
@@ -172,11 +178,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 ctx.overlay(jmap, 'Email', 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);
         }
@@ -255,39 +261,48 @@ async function backfillBodies(ctx: SyncContext, jmap: string): Promise<void> {
   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<string, EmailObject>();
+    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 });
   }
 }

lib/store/storage.ts

@@ -39,7 +39,13 @@ export function runTransaction<T>(
     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;