feat: merge offline contacts and files

lib/store/__tests__/files.test.ts

@@ -54,6 +54,9 @@ async function call(engine: AccountEngine, name: string, args: Record<string, un
 
 const ref = (id: string, jmap = JMAP_ACCOUNT) => ({ jmap, id });
 
+const bytesOf = async (kept: unknown): Promise<Uint8Array> => new Uint8Array(await (kept as { blob: Blob }).blob.arrayBuffer());
+const keptIn = (db: StorageAdapter, blobId: string, scope: StoreScope = DEFAULT_SCOPE) => fileBlob({ db, scope }, JMAP_ACCOUNT, blobId);
+
 function names(rows: Array<{ name?: unknown }>): unknown[] {
   return rows.map((row) => row.name).sort();
 }
@@ -110,7 +113,7 @@ describe('file metadata sync', () => {
     server.calls.length = 0;
     await rig.engine.requestSync();
     expect(Number(rig.db.get<{ n: number }>('SELECT COUNT(*) AS n FROM file_node')?.n)).toBe(6000);
-    expect(server.calls.find(([n]) => n === 'FileNode/query')?.[1].position).toBe(5000);
+    expect(server.calls.find(([n, a]) => n === 'FileNode/query' && a.accountId === JMAP_ACCOUNT)?.[1].anchor).toBeTruthy();
     expect(server.calls.filter(([n, a]) => n === 'FileNode/get' && a.ids === null && a.accountId === JMAP_ACCOUNT)).toHaveLength(0);
   });
 
@@ -159,6 +162,60 @@ describe('file answers', () => {
     expect(await engine.answer([['FileNode/get', { accountId: JMAP_ACCOUNT, ids: null }, '0']])).toBeNull();
   });
 
+  it('answers serverUnavailable until the first full pass is done', async () => {
+    const server = new FakeFiles({ getAll: 'refuse' });
+    for (let i = 0; i < 6000; i += 1) server.addFile(`f${i}`);
+    const rig = rigFor(server);
+    let requests = 0;
+    server.beforeRequest = () => {
+      requests += 1;
+      if (requests === 6) server.offline = true;
+    };
+    await rig.engine.requestSync();
+    expect(Number(rig.db.get<{ n: number }>('SELECT COUNT(*) AS n FROM file_node')?.n)).toBe(5000);
+    expect(await rig.engine.answer([['FileNode/get', { accountId: JMAP_ACCOUNT, ids: null }, '0']])).toBeNull();
+    expect(await rig.engine.answer([['FileNode/query', { accountId: JMAP_ACCOUNT, filter: {} }, '0']])).toBeNull();
+    server.offline = false;
+    server.beforeRequest = null;
+    await rig.engine.requestSync();
+    expect((await call(rig.engine, 'FileNode/query', { filter: {}, calculateTotal: true })).total).toBe(6000);
+  });
+
+  it('goes on answering from the old copy while a later resync runs', async () => {
+    const server = new FakeFiles({ getAll: 'refuse' });
+    for (let i = 0; i < 6000; i += 1) server.addFile(`f${i}`);
+    const { db, engine } = await synced(server);
+    db.run("UPDATE sync_state SET state = 'bad state!' WHERE type = 'FileNode'");
+    let requests = 0;
+    server.beforeRequest = () => {
+      requests += 1;
+      if (requests === 3) server.offline = true;
+    };
+    await engine.requestSync();
+    expect((await call(engine, 'FileNode/query', { filter: {}, calculateTotal: true })).total).toBe(6000);
+  });
+
+  it('does not skip a node when an earlier one is deleted between pages, and starts over if the anchor goes', async () => {
+    for (const victim of [0, 4999]) {
+      const server = new FakeFiles({ getAll: 'refuse' });
+      const files = Array.from({ length: 6000 }, (_, i) => server.addFile(`f${i}`));
+      const rig = rigFor(server);
+      let queries = 0;
+      const original = server.request.bind(server);
+      server.request = async (calls) => {
+        if (calls.some(([name, args]) => name === 'FileNode/query' && args.accountId === JMAP_ACCOUNT)) {
+          queries += 1;
+          if (queries === 2) server.removeNode(files[victim].id);
+        }
+        return original(calls);
+      };
+      await rig.engine.requestSync();
+      const kept = new Set(rig.db.all<{ id: string }>('SELECT id FROM file_node').map((r) => r.id));
+      expect(kept.size).toBe(5999);
+      expect(kept.has(files[victim].id)).toBe(false);
+    }
+  });
+
   it('gives the last server state and only asked-for properties', async () => {
     const server = new FakeFiles();
     const file = server.addFile('a.txt');
@@ -234,8 +291,8 @@ describe('pinned files', () => {
     const read = await engine.areaCall('files.read', ref(file.id)) as any;
     expect(read.name).toBe('big.bin');
     expect(read.type).toBe('text/plain');
-    expect(Buffer.compare(read.data, content)).toBe(0);
-    expect(Buffer.compare(fileBlob({ db }, JMAP_ACCOUNT, file.blobId as string)?.data as Uint8Array, content)).toBe(0);
+    expect(Buffer.compare(await bytesOf(read), content)).toBe(0);
+    expect(Buffer.compare(await bytesOf(keptIn(db, file.blobId as string)), content)).toBe(0);
     expect(await engine.areaCall('files.blob', { jmap: JMAP_ACCOUNT, blobId: file.blobId })).toMatchObject({ size: content.length });
     // Already kept: not fetched again.
     await engine.requestSync();
@@ -260,7 +317,7 @@ describe('pinned files', () => {
     const file = server.addFile('a.txt');
     const { db, engine } = await synced(server);
     expect(await engine.areaCall('files.read', ref(file.id))).toBeNull();
-    expect(fileBlob({ db }, JMAP_ACCOUNT, file.blobId as string)).toBeNull();
+    expect(keptIn(db, file.blobId as string)).toBeNull();
   });
 
   it('refuses to pin a folder or an unknown id', async () => {
@@ -436,15 +493,219 @@ describe('pinned files', () => {
     expect((await engine.areaCall('files.kept', undefined) as any).files[0].error).toBe('tooLarge');
   });
 
+  it('stops a download at the limit when the size is missing or wrong', async () => {
+    const server = new FakeFiles();
+    server.partBytes = 10;
+    const noSize = server.addFile('noSize', null, new Uint8Array(1000), { size: null });
+    const wrong = server.addFile('wrong', null, new Uint8Array(1000), { size: 1 });
+    const { db, engine } = await synced(server, { maxAttachmentBytes: 50 });
+    await engine.areaCall('files.noteOpened', ref(noSize.id));
+    await engine.areaCall('files.noteOpened', ref(wrong.id));
+    await engine.requestSync();
+    expect(server.delivered.get(noSize.blobId as string)).toBeLessThanOrEqual(60);
+    expect(server.delivered.get(wrong.blobId as string)).toBeLessThanOrEqual(60);
+    expect(Number(db.get<{ n: number }>('SELECT COUNT(*) AS n FROM file_blob_chunk')?.n)).toBe(0);
+    expect(Number(db.get<{ n: number }>('SELECT COUNT(*) AS n FROM file_blob')?.n)).toBe(0);
+    // Noted, so the next pass doesn't start it again.
+    const count = server.downloads.length;
+    await engine.requestSync();
+    expect(server.downloads).toHaveLength(count);
+  });
+
+  it('stops a pinned download at the total limit too', async () => {
+    const server = new FakeFiles();
+    server.partBytes = 10;
+    const big = server.addFile('big', null, new Uint8Array(1000), { size: null });
+    const { engine } = await synced(server, { maxTotalBytes: 100 });
+    await engine.areaCall('files.pin', ref(big.id));
+    await engine.requestSync();
+    expect(server.delivered.get(big.blobId as string)).toBeLessThanOrEqual(110);
+    expect((await engine.areaCall('files.kept', undefined) as any).files[0]).toMatchObject({ kept: false, error: 'overLimit' });
+  });
+
+  it('stops a pinned download at the organisation limit, whatever the size says', async () => {
+    const server = new FakeFiles();
+    server.partBytes = 10;
+    const big = server.addFile('big', null, new Uint8Array(1000), { size: null });
+    const { engine } = await synced(server, { orgMaxFileBytes: 50 });
+    await engine.areaCall('files.pin', ref(big.id));
+    await engine.requestSync();
+    expect(server.delivered.get(big.blobId as string)).toBeLessThanOrEqual(60);
+    expect((await engine.areaCall('files.kept', undefined) as any).files[0]).toMatchObject({ kept: false, error: 'tooLarge' });
+  });
+
+  it('writes a download in pieces and gives nothing out until it is complete', async () => {
+    const server = new FakeFiles();
+    const file = server.addFile('big', null, new Uint8Array(3 * 1024 * 1024).fill(7));
+    const { db, engine } = await synced(server);
+    await engine.areaCall('files.pin', ref(file.id));
+    const seen: number[] = [];
+    let readDuring: unknown = 'not read';
+    server.onPart = () => {
+      seen.push(Number(db.get<{ n: number }>('SELECT COUNT(*) AS n FROM file_blob_chunk')?.n));
+      readDuring = keptIn(db, file.blobId as string);
+    };
+    await engine.requestSync();
+    expect(Math.max(...seen)).toBeGreaterThanOrEqual(2);
+    expect(readDuring).toBeNull();
+    expect((await bytesOf(keptIn(db, file.blobId as string))).length).toBe(3 * 1024 * 1024);
+  });
+
+  it('leaves no pieces behind after a broken download, and gives none out', async () => {
+    const server = new FakeFiles();
+    const file = server.addFile('big', null, new Uint8Array(3 * 1024 * 1024).fill(7));
+    // Twice, since the pin itself starts a pass and the sync below asks for another.
+    server.failMidBody.set(file.blobId as string, { after: 2 * 1024 * 1024, times: 2 });
+    const { db, engine, clock } = await synced(server);
+    await engine.areaCall('files.pin', ref(file.id));
+    await engine.requestSync();
+    expect(server.failMidBody.get(file.blobId as string)?.times).toBe(0);
+    expect(await engine.areaCall('files.read', ref(file.id))).toBeNull();
+    expect(Number(db.get<{ n: number }>('SELECT COUNT(*) AS n FROM file_blob_chunk')?.n)).toBe(0);
+    clock.at += 60 * 60 * 1000;
+    await engine.requestSync();
+    expect((await bytesOf(await engine.areaCall('files.read', ref(file.id)))).length).toBe(3 * 1024 * 1024);
+  });
+
+  it('clears the pieces a closed app left behind', async () => {
+    const server = new FakeFiles();
+    const file = server.addFile('a');
+    const { db, engine } = await synced(server);
+    db.run('INSERT INTO file_blob (jmap, blob_id, type, size, kept_at, complete) VALUES (?, ?, ?, 0, 0, 0)', [JMAP_ACCOUNT, 'unfinished', 'x']);
+    db.run('INSERT INTO file_blob_chunk (jmap, blob_id, seq, data) VALUES (?, ?, 0, ?)', [JMAP_ACCOUNT, 'unfinished', new Uint8Array(5)]);
+    expect(keptIn(db, 'unfinished')).toBeNull();
+    await engine.areaCall('files.pin', ref(file.id));
+    await engine.requestSync();
+    expect(Number(db.get<{ n: number }>("SELECT COUNT(*) AS n FROM file_blob_chunk WHERE blob_id = 'unfinished'")?.n)).toBe(0);
+    expect(Number(db.get<{ n: number }>("SELECT COUNT(*) AS n FROM file_blob WHERE blob_id = 'unfinished'")?.n)).toBe(0);
+  });
+
+  it('marks a file that keeps breaking and gets on with the others', async () => {
+    const server = new FakeFiles();
+    const bad = server.addFile('bad', null, new Uint8Array(100));
+    const good = server.addFile('good', null, 10);
+    server.partBytes = 10;
+    server.failMidBody.set(bad.blobId as string, { after: 50, times: 3 });
+    const rig = await synced(server);
+    await rig.engine.areaCall('files.pin', ref(bad.id));
+    await rig.engine.areaCall('files.pin', ref(good.id));
+    await rig.engine.requestSync();
+    expect(await rig.engine.areaCall('files.read', ref(good.id))).not.toBeNull();
+    const badRow = async () => (await rig.engine.areaCall('files.kept', undefined) as any).files.find((f: { name: string }) => f.name === 'bad');
+    expect(await badRow()).toMatchObject({ kept: false, error: 'download' });
+    // Failed files go last, and the second failure starts a wait.
+    await rig.engine.requestSync();
+    const tries = server.downloads.filter((d) => d.blobId === bad.blobId).length;
+    await rig.engine.requestSync();
+    expect(server.downloads.filter((d) => d.blobId === bad.blobId)).toHaveLength(tries);
+    rig.clock.at += 60 * 60 * 1000;
+    await rig.engine.requestSync();
+    rig.clock.at += 2 * 60 * 60 * 1000;
+    await rig.engine.requestSync();
+    expect(await rig.engine.areaCall('files.read', ref(bad.id))).not.toBeNull();
+    expect(await badRow()).toMatchObject({ kept: true, error: null });
+  });
+
+  it('keeps the old copy of a pinned file until the new one is stored', async () => {
+    const server = new FakeFiles();
+    const file = server.addFile('a.txt', null, new Uint8Array([1, 2, 3]));
+    const { db, engine } = await synced(server);
+    await engine.areaCall('files.pin', ref(file.id));
+    await engine.requestSync();
+    const old = file.blobId as string;
+    const fresh = server.replaceContent(file.id, new Uint8Array([9, 9]));
+    server.failMidBody.set(fresh, { after: 0, times: 1 });
+    await engine.requestSync();
+    // The new one didn't arrive: what's asked for by its new id is the last copy, marked as old.
+    expect([...await bytesOf(keptIn(db, old))]).toEqual([1, 2, 3]);
+    const stale = await engine.areaCall('files.read', ref(file.id)) as any;
+    expect([...await bytesOf(stale)]).toEqual([1, 2, 3]);
+    expect(stale.stale).toBe(true);
+    expect(keptIn(db, fresh)).toMatchObject({ stale: true });
+    await engine.requestSync();
+    expect(keptIn(db, old)).toBeNull();
+    const current = await engine.areaCall('files.read', ref(file.id)) as any;
+    expect([...await bytesOf(current)]).toEqual([9, 9]);
+    expect(current.stale).toBe(false);
+  });
+
+  it('drops pins newest first when a policy lowers the limit, as soon as it changes', async () => {
+    const server = new FakeFiles();
+    const a = server.addFile('a', null, 60);
+    const b = server.addFile('b', null, 60);
+    const c = server.addFile('c', null, 30);
+    const rig = await synced(server);
+    for (const file of [a, b, c]) {
+      rig.clock.at += 1000;
+      await rig.engine.areaCall('files.pin', ref(file.id));
+      await rig.engine.requestSync();
+    }
+    rig.engine.setScope({ ...DEFAULT_SCOPE, maxTotalBytes: 100 });
+    const status = await rig.engine.areaCall('files.kept', undefined) as any;
+    const row = (name: string) => status.files.find((f: { name: string }) => f.name === name);
+    expect(row('a')).toMatchObject({ kept: true, pinned: true });
+    expect(row('b')).toMatchObject({ kept: false, pinned: true, error: 'overLimit' });
+    expect(row('c')).toMatchObject({ kept: true, pinned: true });
+    expect(status.bytes).toBe(90);
+    const count = server.downloads.length;
+    await rig.engine.requestSync();
+    expect(server.downloads).toHaveLength(count);
+
+    rig.engine.setScope(DEFAULT_SCOPE);
+    await rig.engine.requestSync();
+    expect((await rig.engine.areaCall('files.kept', undefined) as any).bytes).toBe(150);
+  });
+
+  it('applies a tighter policy at start too, before any sync', async () => {
+    const server = new FakeFiles();
+    const file = server.addFile('a');
+    const rig = await synced(server);
+    await rig.engine.areaCall('files.pin', ref(file.id));
+    await rig.engine.requestSync();
+    await rig.engine.close();
+    server.offline = true;
+    new AccountEngine({ account: 'a@fake.test', db: rig.db, session: FILES_SESSION, transport: server, areas: [filesArea], scope: { ...DEFAULT_SCOPE, attachments: false }, sleep: async () => {} });
+    expect(Number(rig.db.get<{ n: number }>('SELECT COUNT(*) AS n FROM file_blob_chunk')?.n)).toBe(0);
+  });
+
+  it('refuses kept contents with attachments off, and drops them when the scope changes', async () => {
+    const server = new FakeFiles();
+    const file = server.addFile('a');
+    const rig = await synced(server);
+    await rig.engine.areaCall('files.pin', ref(file.id));
+    await rig.engine.requestSync();
+    const off = { ...DEFAULT_SCOPE, attachments: false };
+    expect(keptIn(rig.db, file.blobId as string, off)).toBeNull();
+    expect(keptIn(rig.db, file.blobId as string)).not.toBeNull();
+    rig.engine.setScope(off);
+    expect(await rig.engine.areaCall('files.read', ref(file.id))).toBeNull();
+    expect(Number(rig.db.get<{ n: number }>('SELECT COUNT(*) AS n FROM file_blob_chunk')?.n)).toBe(0);
+    expect((await rig.engine.areaCall('files.kept', undefined) as any).files[0]).toMatchObject({ kept: false, error: 'notAllowed' });
+  });
+
+  it('lets a close end a download at once', async () => {
+    const server = new FakeFiles();
+    const file = server.addFile('a');
+    const { engine } = await synced(server);
+    await engine.areaCall('files.pin', ref(file.id));
+    server.hangDownloads = true;
+    const pass = engine.requestSync();
+    await new Promise((resolve) => setTimeout(resolve, 5));
+    expect(server.downloading).toBe(1);
+    const closed = await Promise.race([engine.close().then(() => 'closed'), new Promise((resolve) => setTimeout(() => resolve('still waiting'), 200))]);
+    expect(closed).toBe('closed');
+    await pass;
+  });
+
   it('does not store a file unpinned while it was downloading', async () => {
     const server = new FakeFiles();
     const file = server.addFile('a');
     const { engine } = await synced(server);
     await engine.areaCall('files.pin', ref(file.id));
-    const original = server.download.bind(server);
-    server.download = async (jmap, blobId) => {
+    const original = server.downloadStream.bind(server);
+    server.downloadStream = async (jmap, blobId, type, name, sink) => {
       await engine.areaCall('files.unpin', ref(file.id));
-      return original(jmap, blobId);
+      return original(jmap, blobId, type, name, sink);
     };
     await engine.requestSync();
     expect(server.downloads).toHaveLength(1);
@@ -477,9 +738,9 @@ describe('pinned files', () => {
     const old = file.blobId as string;
     const fresh = server.replaceContent(file.id, new Uint8Array([9, 9]));
     await engine.requestSync();
-    expect(fileBlob({ db }, JMAP_ACCOUNT, old)).toBeNull();
-    expect([...(fileBlob({ db }, JMAP_ACCOUNT, fresh)?.data ?? [])]).toEqual([9, 9]);
-    expect(Array.from((await engine.areaCall('files.read', ref(file.id)) as any).data)).toEqual([9, 9]);
+    expect(keptIn(db, old)).toBeNull();
+    expect([...await bytesOf(keptIn(db, fresh))]).toEqual([9, 9]);
+    expect([...await bytesOf(await engine.areaCall('files.read', ref(file.id)))]).toEqual([9, 9]);
   });
 
   it('drops the content when the file is removed on the server, unless another node shares it', async () => {
@@ -493,10 +754,10 @@ describe('pinned files', () => {
     expect(server.downloads).toHaveLength(1);
     server.removeNode(a.id);
     await engine.requestSync();
-    expect(fileBlob({ db }, JMAP_ACCOUNT, 'shared-blob')).not.toBeNull();
+    expect(keptIn(db, 'shared-blob')).not.toBeNull();
     server.removeNode(b.id);
     await engine.requestSync();
-    expect(fileBlob({ db }, JMAP_ACCOUNT, 'shared-blob')).toBeNull();
+    expect(keptIn(db, 'shared-blob')).toBeNull();
     expect(Number(db.get<{ n: number }>('SELECT COUNT(*) AS n FROM file_blob_chunk')?.n)).toBe(0);
   });
 
@@ -521,8 +782,8 @@ describe('pinned files', () => {
     const file = server.addFile('a');
     const rig = await synced(server);
     await rig.engine.areaCall('files.pin', ref(file.id));
-    const original = server.download.bind(server);
-    server.download = async () => {
+    const original = server.downloadStream.bind(server);
+    server.downloadStream = async () => {
       throw new TransportError('network', 'Failed to fetch');
     };
     const syncedBefore = rig.events.filter((e) => e.type === 'synced').length;
@@ -531,7 +792,7 @@ describe('pinned files', () => {
     expect(rig.events.filter((e) => e.type === 'synced').length).toBe(syncedBefore);
     expect((await rig.engine.areaCall('files.kept', undefined) as any).files[0]).toMatchObject({ kept: false, error: null });
 
-    server.download = original;
+    server.downloadStream = original;
     await rig.engine.requestSync();
     expect(await rig.engine.areaCall('files.read', ref(file.id))).not.toBeNull();
   });
@@ -601,11 +862,11 @@ describe('recently opened files', () => {
     const file = server.addFile('new', null, 10);
     const rig = await synced(server, { maxTotalBytes: 2 * FILE_CACHE_CAP_BYTES });
     // As if an older file already took all of it: only the row counts here.
-    rig.db.run('INSERT INTO file_blob (jmap, blob_id, type, size, kept_at) VALUES (?, ?, ?, ?, ?)', [JMAP_ACCOUNT, 'huge', 'x', FILE_CACHE_CAP_BYTES, NOW - 1000]);
+    rig.db.run('INSERT INTO file_blob (jmap, blob_id, type, size, kept_at, complete) VALUES (?, ?, ?, ?, ?, 1)', [JMAP_ACCOUNT, 'huge', 'x', FILE_CACHE_CAP_BYTES, NOW - 1000]);
     rig.db.run("INSERT INTO file_node (jmap, id, name, kind, blob_id, size, json, opened_at) VALUES (?, 'hugenode', 'huge', 'file', 'huge', ?, '{}', ?)", [JMAP_ACCOUNT, FILE_CACHE_CAP_BYTES, NOW - 1000]);
     await rig.engine.areaCall('files.noteOpened', ref(file.id));
     await rig.engine.requestSync();
-    expect(fileBlob({ db: rig.db }, JMAP_ACCOUNT, 'huge')).toBeNull();
+    expect(keptIn(rig.db, 'huge')).toBeNull();
     expect(await rig.engine.areaCall('files.read', ref(file.id))).not.toBeNull();
     expect((await rig.engine.areaCall('files.kept', undefined) as any).capBytes).toBe(FILE_CACHE_CAP_BYTES);
   });
@@ -621,7 +882,8 @@ describe('recently opened files', () => {
     await rig.engine.requestSync();
     expect(await rig.engine.areaCall('files.read', ref(b.id))).not.toBeNull();
     expect(await rig.engine.areaCall('files.read', ref(a.id))).toBeNull();
-    expect(server.downloads.map((d) => d.blobId)).toEqual([b.blobId, a.blobId]);
+    // It can see that from the sizes, so it doesn't even download it.
+    expect(server.downloads.map((d) => d.blobId)).toEqual([b.blobId]);
     const count = server.downloads.length;
     await rig.engine.requestSync();
     expect(server.downloads).toHaveLength(count);

lib/store/__tests__/helpers/fake-files.ts

@@ -59,6 +59,17 @@ export class FakeFiles extends FakeJmap {
   readonly downloads: Array<{ jmap: string; blobId: string }> = [];
   downloading = 0;
   mostAtOnce = 0;
+  /** How a body arrives. */
+  partBytes = 256 * 1024;
+  /** Blob ids whose body breaks after this many bytes, with how many more times it should. */
+  readonly failMidBody = new Map<string, { after: number; times: number }>();
+  /** Bytes each blob's body actually delivered, last time. */
+  readonly delivered = new Map<string, number>();
+  /** Called after each part of a body, to look at the store meanwhile. */
+  onPart: ((blobId: string) => void) | null = null;
+  /** Set to make a download hang until the transport is aborted. */
+  hangDownloads = false;
+  private readonly hanging = new Set<() => void>();
 
   constructor(options: FakeFilesOptions = {}) {
     super();
@@ -162,7 +173,12 @@ export class FakeFiles extends FakeJmap {
         let list = [...account.nodes.values()];
         if (this.options.queryOmitsFolders) list = list.filter((n) => n.blobId !== null);
         if (filter.parentId) list = list.filter((n) => n.parentId === filter.parentId);
-        const position = Number(args.position ?? 0);
+        let position = Number(args.position ?? 0);
+        if (args.anchor !== undefined) {
+          const at = list.findIndex((n) => n.id === args.anchor);
+          if (at < 0) throw Object.assign(new Error('anchorNotFound'), { jmapError: 'anchorNotFound' });
+          position = at + Number(args.anchorOffset ?? 0);
+        }
         const limit = Math.min(Number(args.limit ?? 5000) || 5000, 5000);
         return {
           accountId: args.accountId,
@@ -178,15 +194,37 @@ export class FakeFiles extends FakeJmap {
     }
   }
 
-  override async download(jmap: string, blobId: string): Promise<Uint8Array> {
+  abort(): void {
+    for (const stop of this.hanging) stop();
+  }
+
+  override async downloadStream(jmap: string, blobId: string, _type: string, _name: string, sink: (part: Uint8Array) => void): Promise<void> {
     if (this.offline) throw new TransportError('network', 'Failed to fetch');
     this.downloads.push({ jmap, blobId });
     this.downloading += 1;
     this.mostAtOnce = Math.max(this.mostAtOnce, this.downloading);
-    await Promise.resolve();
-    this.downloading -= 1;
-    const data = this.blobs.get(blobId);
-    if (!data) throw new TransportError('http', 'Download failed: 404', 404);
-    return data;
+    try {
+      if (this.hangDownloads) {
+        await new Promise<void>((resolve) => this.hanging.add(resolve));
+        throw new TransportError('network', 'Aborted');
+      }
+      const data = this.blobs.get(blobId);
+      if (!data) throw new TransportError('http', 'Download failed: 404', 404);
+      const failure = this.failMidBody.get(blobId);
+      this.delivered.set(blobId, 0);
+      for (let at = 0; at < data.length; at += this.partBytes) {
+        await Promise.resolve();
+        if (failure && failure.times > 0 && at >= failure.after) {
+          failure.times -= 1;
+          throw new TransportError('network', 'Connection reset', undefined, undefined, undefined, true);
+        }
+        const part = data.subarray(at, at + this.partBytes);
+        this.delivered.set(blobId, at + part.length);
+        sink(part);
+        this.onPart?.(blobId);
+      }
+    } finally {
+      this.downloading -= 1;
+    }
   }
 }

lib/store/__tests__/helpers/fake-jmap.ts

@@ -290,6 +290,10 @@ export class FakeJmap implements JmapTransport {
     return new Uint8Array();
   }
 
+  async downloadStream(jmap: string, blobId: string, type: string, name: string, sink: (part: Uint8Array) => void): Promise<void> {
+    sink(await this.download(jmap, blobId, type, name));
+  }
+
   async upload(): Promise<{ blobId: string; type: string; size: number }> {
     return { blobId: 'uploaded', type: 'application/octet-stream', size: 0 };
   }

lib/store/__tests__/transport.test.ts

@@ -0,0 +1,129 @@
+// @vitest-environment node
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import { FetchTransport, TransportError, type StoreSession } from '../engine/transport';
+
+const SESSION: StoreSession = {
+  apiUrl: 'https://fake.test/jmap/',
+  downloadUrl: 'https://fake.test/d/{accountId}/{blobId}/{name}?type={type}',
+  uploadUrl: 'https://fake.test/u/{accountId}',
+  primaryAccounts: {},
+  accounts: {},
+  capabilities: {},
+};
+
+type Pull = (controller: ReadableStreamDefaultController<Uint8Array>) => void | Promise<void>;
+
+/** A response whose body comes as the test says, and that fails when the request is aborted, as fetch does. */
+function bodyResponse(signal: AbortSignal | null | undefined, pull: Pull): Response {
+  const stream = new ReadableStream<Uint8Array>({
+    start(controller) {
+      signal?.addEventListener('abort', () => controller.error(new DOMException('Aborted', 'AbortError')));
+    },
+    pull,
+  });
+  return new Response(stream, { status: 200 });
+}
+
+function transportWith(respond: (init: RequestInit) => Response | Promise<Response>): FetchTransport {
+  return new FetchTransport({
+    session: () => SESSION,
+    auth: () => 'Basic x',
+    refreshAuth: async () => 'Basic x',
+    fetch: (async (_url: string, init: RequestInit) => respond(init)) as unknown as typeof fetch,
+  });
+}
+
+afterEach(() => vi.useRealTimers());
+
+describe('downloads', () => {
+  it('gives a body piece by piece and adds them up for a plain download', async () => {
+    const parts = [new Uint8Array([1, 2]), new Uint8Array([3])];
+    const open = () => transportWith((init) => {
+      let i = 0;
+      return bodyResponse(init.signal, (c) => {
+        if (i < parts.length) c.enqueue(parts[i++]);
+        else c.close();
+      });
+    });
+    const seen: number[] = [];
+    await open().downloadStream('a', 'b', 't', 'n', (part) => seen.push(part.length));
+    expect(seen).toEqual([2, 1]);
+    expect([...await open().download('a', 'b', 't', 'n')]).toEqual([1, 2, 3]);
+  });
+
+  it('turns a body that breaks into a TransportError saying so', async () => {
+    const open = () => transportWith((init) => {
+      let i = 0;
+      return bodyResponse(init.signal, (c) => {
+        if (i++ === 0) c.enqueue(new Uint8Array(1));
+        else c.error(new TypeError('network error'));
+      });
+    });
+    const broken = await open().download('a', 'b', 't', 'n').catch((e: unknown) => e);
+    expect(broken).toBeInstanceOf(TransportError);
+    expect(broken).toMatchObject({ kind: 'network', interrupted: true });
+  });
+
+  it('lets a slow download run past the old total limit while bytes keep coming', async () => {
+    vi.useFakeTimers();
+    let sent = 0;
+    const transport = transportWith((init) => bodyResponse(init.signal, async (c) => {
+      await new Promise((resolve) => setTimeout(resolve, 30_000));
+      if (sent++ < 10) c.enqueue(new Uint8Array(1));
+      else c.close();
+    }));
+    let bytes = 0;
+    const done = transport.downloadStream('a', 'b', 't', 'n', (part) => { bytes += part.length; });
+    await vi.advanceTimersByTimeAsync(11 * 30_000);
+    await done;
+    expect(bytes).toBe(10);
+  });
+
+  it('gives up when no bytes come for a while, and says it was this download', async () => {
+    vi.useFakeTimers();
+    const transport = transportWith((init) => bodyResponse(init.signal, () => new Promise(() => {})));
+    const done = transport.downloadStream('a', 'b', 't', 'n', () => {}).catch((e: unknown) => e);
+    await vi.advanceTimersByTimeAsync(5 * 60_000);
+    expect(await done).toMatchObject({ kind: 'network', interrupted: true });
+  });
+
+  it('lets an error from the sink through and stops reading', async () => {
+    let cancelled = false;
+    const transport = transportWith(() => new Response(new ReadableStream<Uint8Array>({
+      pull: (c) => c.enqueue(new Uint8Array(1)),
+      cancel: () => { cancelled = true; },
+    })));
+    const stop = new Error('over the limit');
+    await expect(transport.downloadStream('a', 'b', 't', 'n', () => { throw stop; })).rejects.toBe(stop);
+    expect(cancelled).toBe(true);
+  });
+
+  it('keeps a plain network failure apart from a broken download', async () => {
+    const transport = transportWith(() => { throw new TypeError('Failed to fetch'); });
+    const failed = await transport.downloadStream('a', 'b', 't', 'n', () => {}).catch((e: unknown) => e);
+    expect(failed).toMatchObject({ kind: 'network' });
+    expect((failed as TransportError).interrupted).toBeFalsy();
+  });
+});
+
+describe('requests', () => {
+  it('turns a broken response body into a TransportError', async () => {
+    const transport = transportWith((init) => bodyResponse(init.signal, (c) => c.error(new TypeError('network error'))));
+    const failed = await transport.request([['Core/echo', {}, '0']], ['urn:ietf:params:jmap:core']).catch((e: unknown) => e);
+    expect(failed).toBeInstanceOf(TransportError);
+    expect(failed).toMatchObject({ kind: 'network' });
+  });
+});
+
+describe('abort', () => {
+  it('ends what is running at once, and anything after it', async () => {
+    const transport = transportWith((init) => bodyResponse(init.signal, () => new Promise(() => {})));
+    const running = transport.downloadStream('a', 'b', 't', 'n', () => {}).catch((e: unknown) => e);
+    const request = transport.request([['Core/echo', {}, '0']], ['x']).catch((e: unknown) => e);
+    await new Promise((resolve) => setTimeout(resolve, 5));
+    transport.abort();
+    expect(await running).toBeInstanceOf(TransportError);
+    expect(await request).toBeInstanceOf(TransportError);
+    await expect(transport.download('a', 'b', 't', 'n')).rejects.toBeInstanceOf(TransportError);
+  });
+});

lib/store/files/index.ts

@@ -12,6 +12,7 @@ export const filesArea: StoreArea = {
   methods: FILE_METHODS,
   syncers: [filesSyncer],
   applyScope: applyFilesScope,
+  scopeChanged: applyFilesScope,
   usage: filesUsage,
   keptBytes: (ctx) => filesUsage(ctx)[0].bytes,
   calls: {

lib/store/files/keep.ts

@@ -6,22 +6,38 @@
  * - A pinned file stays until it's unpinned or removed on the server. The cap
  *   doesn't push it out and the user's own size limit doesn't apply. The
  *   organisation's limit on one file (scope.orgMaxFileBytes) and the account's
- *   total (scope.maxTotalBytes, which a policy can lower) do.
+ *   total (scope.maxTotalBytes, which a policy can lower) do: past the total, the
+ *   newest pins lose their content and say `overLimit`.
  * - A recently opened file is kept if it's within scope.maxAttachmentBytes, and
  *   pushed out oldest first when the cap is reached. Its opened_at goes with
  *   it, so it isn't fetched straight back.
- * - With scope.attachments off, nothing is kept.
+ * - With scope.attachments off, nothing is kept or given out.
+ *
+ * A file comes in as a stream, written in 1 MB chunks that each commit on their
+ * own, and is held nowhere whole. `complete` says the last chunk is in, so a
+ * broken download or a closed app leaves nothing readable. Reading goes chunk by
+ * chunk into a Blob, which the browser can keep out of the worker's memory.
+ * A new version's content replaces the old one only when it's all there.
  */
 
 import { TransportError } from '../engine/transport';
-import { MethodError, type ScopeChange, type StoreContext, type StoreUsagePart, type SyncContext } from '../area';
+import { MethodError, type StoreContext, type StoreUsagePart, type SyncContext } from '../area';
 import type { StorageAdapter } from '../storage';
 import { nodeMediaType, type WireNode } from './records';
 
 export const FILE_CACHE_CAP_BYTES = 500 * 1024 * 1024;
 const CHUNK_BYTES = 1024 * 1024;
+/** After the second failure in a row a file waits this long, doubling, before the next try. */
+const RETRY_WAIT_MS = 15 * 60 * 1000;
+const MOST_RETRY_WAIT_MS = 24 * 60 * 60 * 1000;
 
-export interface KeptFile { data: Uint8Array; type: string; size: number }
+export interface KeptFile {
+  blob: Blob;
+  type: string;
+  size: number;
+  /** This is the version before the one asked for, whose content hasn't arrived. */
+  stale: boolean;
+}
 
 export interface NodeRef { jmap: string; id: string }
 
@@ -31,6 +47,7 @@ interface NodeRow {
   name: string;
   kind: string;
   blob_id: string | null;
+  stale_blob_id: string | null;
   size: number | null;
   json: string;
   pinned: number;
@@ -44,6 +61,7 @@ interface BlobRow {
   size: number;
   kept_at: number;
   pinned: number;
+  pinned_at: number | null;
   opened_at: number | null;
 }
 
@@ -51,41 +69,52 @@ function capBytes(ctx: Pick<StoreContext, 'scope'>): number {
   return Math.min(FILE_CACHE_CAP_BYTES, ctx.scope.maxTotalBytes);
 }
 
-/** A kept file's content, for the client's blob lookup. */
-export function fileBlob(ctx: Pick<StoreContext, 'db'>, jmap: string, blobId: string): KeptFile | null {
-  const row = ctx.db.get<{ type: string; size: number }>('SELECT type, size FROM file_blob WHERE jmap = ? AND blob_id = ?', [jmap, blobId]);
-  if (!row) return null;
-  const data = new Uint8Array(Number(row.size));
-  let at = 0;
-  for (const chunk of ctx.db.all<{ data: Uint8Array }>('SELECT data FROM file_blob_chunk WHERE jmap = ? AND blob_id = ? ORDER BY seq', [jmap, blobId])) {
-    data.set(chunk.data, at);
-    at += chunk.data.length;
+function openedLimit(ctx: Pick<StoreContext, 'scope'>): number {
+  return Math.min(ctx.scope.maxAttachmentBytes, capBytes(ctx));
+}
+
+function assemble(db: StorageAdapter, jmap: string, blobId: string, type: string): Blob {
+  let blob = new Blob([], { type });
+  for (let seq = 0; ; seq += 1) {
+    const chunk = db.get<{ data: Uint8Array }>('SELECT data FROM file_blob_chunk WHERE jmap = ? AND blob_id = ? AND seq = ?', [jmap, blobId, seq]);
+    if (!chunk) return blob;
+    // Each step only refers to the last Blob, so one chunk at a time is in memory here.
+    blob = new Blob([blob, chunk.data as Uint8Array<ArrayBuffer>], { type });
   }
-  return { data, type: row.type, size: Number(row.size) };
+}
+
+/**
+ * A kept file's content, for the client's blob lookup. If this version hasn't
+ * arrived and the one before is still kept, that one, marked `stale`.
+ */
+export function fileBlob(ctx: Pick<StoreContext, 'db' | 'scope'>, jmap: string, blobId: string): KeptFile | null {
+  if (!ctx.scope.attachments) return null;
+  const read = (id: string, stale: boolean): KeptFile | null => {
+    const row = ctx.db.get<{ type: string; size: number }>('SELECT type, size FROM file_blob WHERE jmap = ? AND blob_id = ? AND complete = 1', [jmap, id]);
+    return row ? { blob: assemble(ctx.db, jmap, id, row.type), type: row.type, size: Number(row.size), stale } : null;
+  };
+  const current = read(blobId, false);
+  if (current) return current;
+  const older = ctx.db.get<{ stale_blob_id: string }>('SELECT stale_blob_id FROM file_node WHERE jmap = ? AND blob_id = ? AND stale_blob_id IS NOT NULL LIMIT 1', [jmap, blobId]);
+  return older ? read(older.stale_blob_id, true) : null;
 }
 
 function hasBlob(db: StorageAdapter, jmap: string, blobId: string): boolean {
-  return Boolean(db.get('SELECT 1 AS x FROM file_blob WHERE jmap = ? AND blob_id = ?', [jmap, blobId]));
+  return Boolean(db.get('SELECT 1 AS x FROM file_blob WHERE jmap = ? AND blob_id = ? AND complete = 1', [jmap, blobId]));
 }
 
 function dropBlob(db: StorageAdapter, jmap: string, blobId: string): void {
   db.run('DELETE FROM file_blob_chunk WHERE jmap = ? AND blob_id = ?', [jmap, blobId]);
   db.run('DELETE FROM file_blob WHERE jmap = ? AND blob_id = ?', [jmap, blobId]);
+  db.run('UPDATE file_node SET stale_blob_id = NULL WHERE jmap = ? AND stale_blob_id = ?', [jmap, blobId]);
 }
 
-function storeBlob(ctx: StoreContext, jmap: string, blobId: string, type: string, data: Uint8Array): void {
-  dropBlob(ctx.db, jmap, blobId);
-  ctx.db.run('INSERT INTO file_blob (jmap, blob_id, type, size, kept_at) VALUES (?, ?, ?, ?, ?)', [jmap, blobId, type, data.length, ctx.now()]);
-  for (let seq = 0, at = 0; at < data.length; seq += 1, at += CHUNK_BYTES) {
-    ctx.db.run('INSERT INTO file_blob_chunk (jmap, blob_id, seq, data) VALUES (?, ?, ?, ?)', [jmap, blobId, seq, data.subarray(at, at + CHUNK_BYTES)]);
-  }
-}
-
-/** All kept contents in the store, with what the nodes pointing at each say about it. */
+/** All complete contents in the store, with what the nodes pointing at each say about it. */
 function keptBlobs(db: StorageAdapter): BlobRow[] {
   return db.all<BlobRow>(
-    `SELECT b.jmap, b.blob_id, b.size, b.kept_at, COALESCE(MAX(n.pinned), 0) AS pinned, MAX(n.opened_at) AS opened_at
-     FROM file_blob b LEFT JOIN file_node n ON n.jmap = b.jmap AND n.blob_id = b.blob_id
+    `SELECT b.jmap, b.blob_id, b.size, b.kept_at, COALESCE(MAX(n.pinned), 0) AS pinned, MAX(n.pinned_at) AS pinned_at, MAX(n.opened_at) AS opened_at
+     FROM file_blob b LEFT JOIN file_node n ON n.jmap = b.jmap AND (n.blob_id = b.blob_id OR n.stale_blob_id = b.blob_id)
+     WHERE b.complete = 1
      GROUP BY b.jmap, b.blob_id`,
   );
 }
@@ -95,16 +124,16 @@ function ageOf(blob: BlobRow): number {
 }
 
 function pushOut(db: StorageAdapter, blob: BlobRow): void {
+  db.run('UPDATE file_node SET opened_at = NULL WHERE jmap = ? AND (blob_id = ? OR stale_blob_id = ?)', [blob.jmap, blob.blob_id, blob.blob_id]);
   dropBlob(db, blob.jmap, blob.blob_id);
-  db.run('UPDATE file_node SET opened_at = NULL WHERE jmap = ? AND blob_id = ?', [blob.jmap, blob.blob_id]);
 }
 
 /**
  * Pushes out unpinned contents, oldest first, so `size` more fits under the
  * cap. A recently opened file only pushes out older ones and says false when
- * that isn't enough. A pinned one always gets in.
+ * that isn't enough. A pinned one always gets in. With `dryRun` it only says.
  */
-function makeRoom(ctx: StoreContext, size: number, incoming: { pinned: boolean; openedAt: number }): boolean {
+function makeRoom(ctx: StoreContext, size: number, incoming: { pinned: boolean; openedAt: number }, dryRun = false): boolean {
   const cap = capBytes(ctx);
   const blobs = keptBlobs(ctx.db);
   let used = blobs.reduce((sum, blob) => sum + Number(blob.size), 0);
@@ -118,7 +147,7 @@ function makeRoom(ctx: StoreContext, size: number, incoming: { pinned: boolean;
     chosen.push(blob);
   }
   if (used + size > cap && !incoming.pinned) return false;
-  for (const blob of chosen) pushOut(ctx.db, blob);
+  if (!dryRun) for (const blob of chosen) pushOut(ctx.db, blob);
   return true;
 }
 
@@ -128,162 +157,275 @@ function overOrgLimit(ctx: Pick<StoreContext, 'scope'>, size: number): boolean {
   return typeof cap === 'number' && size > cap;
 }
 
-function pinnedBytes(db: StorageAdapter): number {
-  return keptBlobs(db).filter((blob) => blob.pinned).reduce((sum, blob) => sum + Number(blob.size), 0);
+/** What pinned contents take. Leaves out the old version a new one will replace. */
+function pinnedBytes(db: StorageAdapter, except: string | null = null): number {
+  return keptBlobs(db).filter((blob) => blob.pinned && blob.blob_id !== except).reduce((sum, blob) => sum + Number(blob.size), 0);
 }
 
 function setError(db: StorageAdapter, jmap: string, blobId: string, error: string | null): void {
-  db.run('UPDATE file_node SET keep_error = ? WHERE jmap = ? AND blob_id = ?', [error, jmap, blobId]);
+  db.run('UPDATE file_node SET keep_error = ? WHERE jmap = ? AND (blob_id = ? OR stale_blob_id = ?)', [error, jmap, blobId, blobId]);
 }
 
-/** Contents no node points at any more: the file was removed or has a new version. */
+/**
+ * Makes the store fit the scope as it is now, without the network. Runs at
+ * start, when the scope changes and before fetching. The pins stay, so a loosened
+ * scope fetches them back at the next pass.
+ */
+export function applyFilesScope(ctx: StoreContext): void {
+  const { db } = ctx;
+  const before = keptBlobs(db).length;
+  db.transaction(() => {
+    if (!ctx.scope.attachments) {
+      for (const row of db.all<{ jmap: string; blob_id: string }>('SELECT jmap, blob_id FROM file_blob')) dropBlob(db, row.jmap, row.blob_id);
+      db.run("UPDATE file_node SET keep_error = 'notAllowed' WHERE pinned = 1 OR opened_at IS NOT NULL");
+      return;
+    }
+    db.run("UPDATE file_node SET keep_error = NULL WHERE keep_error = 'notAllowed'");
+    for (const blob of keptBlobs(db)) {
+      if (blob.pinned ? !overOrgLimit(ctx, Number(blob.size)) : Number(blob.size) <= openedLimit(ctx)) continue;
+      if (blob.pinned) {
+        setError(db, blob.jmap, blob.blob_id, 'tooLarge');
+        dropBlob(db, blob.jmap, blob.blob_id);
+      } else {
+        pushOut(db, blob);
+      }
+    }
+    // Oldest pins first, so the newest are the ones to lose out. One that still
+    // fits stays: dropping it would only have the next pass fetch it back.
+    const pins = keptBlobs(db).filter((blob) => blob.pinned).sort((a, b) => Number(a.pinned_at ?? 0) - Number(b.pinned_at ?? 0));
+    let used = 0;
+    for (const blob of pins) {
+      if (used + Number(blob.size) <= ctx.scope.maxTotalBytes) {
+        used += Number(blob.size);
+      } else {
+        setError(db, blob.jmap, blob.blob_id, 'overLimit');
+        dropBlob(db, blob.jmap, blob.blob_id);
+      }
+    }
+    makeRoom(ctx, 0, { pinned: true, openedAt: 0 });
+  });
+  if (keptBlobs(db).length !== before) ctx.emit({ type: 'changed', account: ctx.account, collections: ['FileContent'] });
+}
+
+export function filesUsage(ctx: Pick<StoreContext, 'db'>): StoreUsagePart[] {
+  const row = ctx.db.get<{ bytes: number | null; items: number }>('SELECT SUM(size) AS bytes, COUNT(*) AS items FROM file_blob WHERE complete = 1');
+  return [{ part: 'files', bytes: Number(row?.bytes ?? 0), items: Number(row?.items ?? 0) }];
+}
+
+/** Contents no node points at any more: the file was removed or has a new version in place. */
 export function dropOrphanBlobs(db: StorageAdapter): void {
   for (const blob of keptBlobs(db)) {
-    const used = db.get('SELECT 1 AS x FROM file_node WHERE jmap = ? AND blob_id = ?', [blob.jmap, blob.blob_id]);
+    const used = db.get('SELECT 1 AS x FROM file_node WHERE jmap = ? AND (blob_id = ? OR stale_blob_id = ?)', [blob.jmap, blob.blob_id, blob.blob_id]);
     if (!used) dropBlob(db, blob.jmap, blob.blob_id);
   }
 }
 
-interface Wanted { blob_id: string; id: string; name: string; size: number | null; json: string; pinned: number; opened_at: number | null }
+interface Wanted {
+  blob_id: string;
+  id: string;
+  name: string;
+  size: number | null;
+  json: string;
+  pinned: number;
+  opened_at: number | null;
+  stale_blob_id: string | null;
+  keep_failures: number;
+  keep_failed_at: number | null;
+}
 
 function wantedRow(db: StorageAdapter, jmap: string, blobId: string): Wanted | undefined {
   return db.get<Wanted>(
-    `SELECT blob_id, id, name, size, json, pinned, opened_at FROM file_node
+    `SELECT blob_id, id, name, size, json, pinned, opened_at, stale_blob_id, keep_failures, keep_failed_at FROM file_node
      WHERE jmap = ? AND blob_id = ? AND (pinned = 1 OR opened_at IS NOT NULL)
      ORDER BY pinned DESC, opened_at DESC LIMIT 1`,
     [jmap, blobId],
   );
 }
 
-/** Fetches what should be kept and isn't yet, one file at a time, pinned files first. */
+/** A file that failed once goes again next pass. After that it waits, so it can't eat every pass. */
+function stillWaiting(ctx: StoreContext, want: Wanted): boolean {
+  const failures = Number(want.keep_failures);
+  if (failures < 2 || want.keep_failed_at === null) return false;
+  const wait = Math.min(RETRY_WAIT_MS * 2 ** (failures - 2), MOST_RETRY_WAIT_MS);
+  return ctx.now() < Number(want.keep_failed_at) + wait;
+}
+
+/** Thrown from the sink to end a download that's no longer wanted. */
+class DownloadStopped extends Error {}
+
+/** Thrown from the sink when a download passes its limit, whatever the metadata said. */
+class DownloadTooLarge extends Error {
+  constructor(readonly seen: number) {
+    super('Over the limit');
+  }
+}
+
+/** Takes a body piece by piece and writes it in chunks, each committed on its own. */
+function chunkWriter(ctx: SyncContext, jmap: string, blobId: string, limit: number) {
+  const buffer = new Uint8Array(CHUNK_BYTES);
+  let filled = 0;
+  let seq = 0;
+  let total = 0;
+  const flush = () => {
+    if (filled === 0) return;
+    if (ctx.stopped() || !ctx.scope.attachments || !wantedRow(ctx.db, jmap, blobId)) throw new DownloadStopped();
+    ctx.db.run('INSERT INTO file_blob_chunk (jmap, blob_id, seq, data) VALUES (?, ?, ?, ?)', [jmap, blobId, seq, buffer.subarray(0, filled)]);
+    seq += 1;
+    filled = 0;
+  };
+  return {
+    push(part: Uint8Array): void {
+      total += part.length;
+      if (total > limit) throw new DownloadTooLarge(total);
+      for (let at = 0; at < part.length;) {
+        const take = Math.min(part.length - at, CHUNK_BYTES - filled);
+        buffer.set(part.subarray(at, at + take), filled);
+        filled += take;
+        at += take;
+        if (filled === CHUNK_BYTES) flush();
+      }
+    },
+    finish(): number {
+      flush();
+      return total;
+    },
+  };
+}
+
+function markFailed(ctx: StoreContext, jmap: string, blobId: string): void {
+  ctx.db.run(
+    "UPDATE file_node SET keep_error = 'download', keep_failures = keep_failures + 1, keep_failed_at = ? WHERE jmap = ? AND blob_id = ?",
+    [ctx.now(), jmap, blobId],
+  );
+}
+
+/** The last chunk is in: the file becomes readable if it still fits and is still wanted. */
+function completeBlob(ctx: SyncContext, jmap: string, blobId: string, size: number): void {
+  const { db } = ctx;
+  db.transaction(() => {
+    const now = wantedRow(db, jmap, blobId);
+    if (!now || !ctx.scope.attachments) {
+      dropBlob(db, jmap, blobId);
+      return;
+    }
+    if (now.pinned && (overOrgLimit(ctx, size) || pinnedBytes(db, now.stale_blob_id) + size > ctx.scope.maxTotalBytes)) {
+      dropBlob(db, jmap, blobId);
+      setError(db, jmap, blobId, overOrgLimit(ctx, size) ? 'tooLarge' : 'overLimit');
+      return;
+    }
+    if (!makeRoom(ctx, size, { pinned: Boolean(now.pinned), openedAt: Number(now.opened_at ?? 0) })) {
+      dropBlob(db, jmap, blobId);
+      db.run('UPDATE file_node SET opened_at = NULL WHERE jmap = ? AND blob_id = ? AND pinned = 0', [jmap, blobId]);
+      return;
+    }
+    db.run('UPDATE file_blob SET complete = 1, size = ?, kept_at = ? WHERE jmap = ? AND blob_id = ?', [size, ctx.now(), jmap, blobId]);
+    // The new version is all here, so the old one can go.
+    const olds = db.all<{ stale_blob_id: string }>('SELECT DISTINCT stale_blob_id FROM file_node WHERE jmap = ? AND blob_id = ? AND stale_blob_id IS NOT NULL', [jmap, blobId]);
+    db.run('UPDATE file_node SET stale_blob_id = NULL, keep_error = NULL, keep_failures = 0, keep_failed_at = NULL WHERE jmap = ? AND blob_id = ?', [jmap, blobId]);
+    for (const { stale_blob_id: old } of olds) {
+      if (!db.get('SELECT 1 AS x FROM file_node WHERE jmap = ? AND (blob_id = ? OR stale_blob_id = ?)', [jmap, old, old])) dropBlob(db, jmap, old);
+    }
+  });
+}
+
+async function fetchOne(ctx: SyncContext, jmap: string, want: Wanted): Promise<boolean> {
+  const { db } = ctx;
+  const blobId = want.blob_id;
+  const type = nodeMediaType(JSON.parse(want.json) as WireNode);
+  const limit = want.pinned
+    ? Math.min(ctx.scope.maxTotalBytes - pinnedBytes(db, want.stale_blob_id), ctx.scope.orgMaxFileBytes ?? Infinity)
+    : openedLimit(ctx);
+  db.transaction(() => {
+    db.run('DELETE FROM file_blob_chunk WHERE jmap = ? AND blob_id = ?', [jmap, blobId]);
+    db.run('DELETE FROM file_blob WHERE jmap = ? AND blob_id = ?', [jmap, blobId]);
+    db.run('INSERT INTO file_blob (jmap, blob_id, type, size, kept_at, complete) VALUES (?, ?, ?, 0, ?, 0)', [jmap, blobId, type, ctx.now()]);
+  });
+  const writer = chunkWriter(ctx, jmap, blobId, limit);
+  const forget = () => db.transaction(() => {
+    db.run('DELETE FROM file_blob_chunk WHERE jmap = ? AND blob_id = ?', [jmap, blobId]);
+    db.run('DELETE FROM file_blob WHERE jmap = ? AND blob_id = ?', [jmap, blobId]);
+  });
+  try {
+    if (ctx.transport.downloadStream) await ctx.transport.downloadStream(jmap, blobId, type, want.name, (part) => writer.push(part));
+    else writer.push(await ctx.transport.download(jmap, blobId, type, want.name));
+    completeBlob(ctx, jmap, blobId, writer.finish());
+    return true;
+  } catch (error) {
+    forget();
+    if (error instanceof DownloadStopped) return false;
+    if (error instanceof DownloadTooLarge) {
+      // With the size it now knows, the next pass sees this without starting the download.
+      db.run('UPDATE file_node SET size = ? WHERE jmap = ? AND blob_id = ?', [error.seen, jmap, blobId]);
+      setError(db, jmap, blobId, want.pinned && !overOrgLimit(ctx, error.seen) ? 'overLimit' : 'tooLarge');
+      return false;
+    }
+    // This file broke, or the server won't give it: the next one may be fine. No
+    // network at all or no credentials reach the engine, and the next pass tries again.
+    if (error instanceof TransportError && (error.kind === 'http' || error.interrupted)) {
+      markFailed(ctx, jmap, blobId);
+      return false;
+    }
+    throw error;
+  }
+}
+
+/** Fetches what should be kept and isn't yet, one file at a time, pinned first and failed ones last. */
 export async function fetchKeptFiles(ctx: SyncContext, jmap: string): Promise<void> {
   const { db } = ctx;
-  if (!ctx.scope.attachments) {
-    db.transaction(() => {
-      for (const blob of keptBlobs(db)) dropBlob(db, blob.jmap, blob.blob_id);
-      db.run("UPDATE file_node SET keep_error = 'notAllowed' WHERE jmap = ? AND (pinned = 1 OR opened_at IS NOT NULL)", [jmap]);
-    });
-    return;
-  }
+  // What a closed app or a stopped pass left unfinished.
   db.transaction(() => {
-    db.run("UPDATE file_node SET keep_error = NULL WHERE jmap = ? AND keep_error = 'notAllowed'", [jmap]);
-    // The limit may have gone down since last time.
-    makeRoom(ctx, 0, { pinned: true, openedAt: 0 });
+    for (const row of db.all<{ blob_id: string }>('SELECT blob_id FROM file_blob WHERE jmap = ? AND complete = 0', [jmap])) {
+      db.run('DELETE FROM file_blob_chunk WHERE jmap = ? AND blob_id = ?', [jmap, row.blob_id]);
+      db.run('DELETE FROM file_blob WHERE jmap = ? AND blob_id = ?', [jmap, row.blob_id]);
+    }
   });
+  applyFilesScope(ctx);
+  if (!ctx.scope.attachments) return;
   const todo = db.all<{ blob_id: string }>(
     `SELECT n.blob_id FROM file_node n
      WHERE n.jmap = ? AND n.blob_id IS NOT NULL AND (n.pinned = 1 OR n.opened_at IS NOT NULL)
-       AND NOT EXISTS (SELECT 1 FROM file_blob b WHERE b.jmap = n.jmap AND b.blob_id = n.blob_id)
-     GROUP BY n.blob_id ORDER BY MAX(n.pinned) DESC, MAX(n.opened_at) DESC`,
+       AND NOT EXISTS (SELECT 1 FROM file_blob b WHERE b.jmap = n.jmap AND b.blob_id = n.blob_id AND b.complete = 1)
+     GROUP BY n.blob_id ORDER BY MAX(n.keep_failures) > 0 ASC, MAX(n.pinned) DESC, MAX(n.opened_at) DESC`,
     [jmap],
   ).map((row) => String(row.blob_id));
   let done = 0;
   for (const blobId of todo) {
-    if (ctx.stopped()) return;
+    if (ctx.stopped() || !ctx.scope.attachments) return;
     // The page may have changed its mind during the last download.
     const want = wantedRow(db, jmap, blobId);
-    if (!want) continue;
+    if (!want || stillWaiting(ctx, want)) continue;
     const known = Number(want.size ?? 0);
     if (want.pinned && overOrgLimit(ctx, known)) {
       setError(db, jmap, blobId, 'tooLarge');
       continue;
     }
-    if (!want.pinned && known > Math.min(ctx.scope.maxAttachmentBytes, capBytes(ctx))) {
+    if (!want.pinned && known > openedLimit(ctx)) {
       setError(db, jmap, blobId, 'tooLarge');
       continue;
     }
-    if (want.pinned && pinnedBytes(db) + known > ctx.scope.maxTotalBytes) {
+    if (want.pinned && pinnedBytes(db, want.stale_blob_id) + known > ctx.scope.maxTotalBytes) {
       setError(db, jmap, blobId, 'overLimit');
       continue;
     }
-    let data: Uint8Array;
-    const type = nodeMediaType(JSON.parse(want.json) as WireNode);
-    try {
-      data = await ctx.transport.download(jmap, blobId, type, want.name);
-    } catch (error) {
-      // No network or no credentials reach the engine. One bad file doesn't stop the others.
-      if (error instanceof TransportError && error.kind === 'http') {
-        setError(db, jmap, blobId, 'download');
-        continue;
-      }
-      throw error;
+    if (!makeRoom(ctx, known, { pinned: Boolean(want.pinned), openedAt: Number(want.opened_at ?? 0) }, true)) {
+      // Older than all that's kept, and no room: not worth the download.
+      db.run('UPDATE file_node SET opened_at = NULL WHERE jmap = ? AND blob_id = ? AND pinned = 0', [jmap, blobId]);
+      continue;
+    }
+    if (await fetchOne(ctx, jmap, want)) {
+      done += 1;
+      ctx.emit({ type: 'progress', account: ctx.account, area: 'files', phase: 'contents', done, total: todo.length });
     }
-    db.transaction(() => {
-      const now = wantedRow(db, jmap, blobId);
-      if (!now) return;
-      if (now.pinned ? overOrgLimit(ctx, data.length) || pinnedBytes(db) + data.length > ctx.scope.maxTotalBytes : data.length > Math.min(ctx.scope.maxAttachmentBytes, capBytes(ctx))) {
-        // The size in the metadata was wrong or missing. With the real one
-        // noted, the next pass sees this without downloading it again.
-        db.run('UPDATE file_node SET size = ? WHERE jmap = ? AND blob_id = ?', [data.length, jmap, blobId]);
-        setError(db, jmap, blobId, !now.pinned || overOrgLimit(ctx, data.length) ? 'tooLarge' : 'overLimit');
-        return;
-      }
-      if (!makeRoom(ctx, data.length, { pinned: Boolean(now.pinned), openedAt: Number(now.opened_at ?? 0) })) {
-        db.run('UPDATE file_node SET opened_at = NULL WHERE jmap = ? AND blob_id = ? AND pinned = 0', [jmap, blobId]);
-        return;
-      }
-      storeBlob(ctx, jmap, blobId, type, data);
-      setError(db, jmap, blobId, null);
-    });
-    done += 1;
-    ctx.emit({ type: 'progress', account: ctx.account, area: 'files', phase: 'contents', done, total: todo.length });
   }
   if (done > 0) ctx.emit({ type: 'changed', account: ctx.account, collections: ['FileContent'] });
 }
 
-/**
- * What the scope no longer allows leaves now, without the network. The pins
- * stay, so a loosened scope fetches them back at the next pass. A pin ignores
- * the size limit but not the total: the organisation's cap wins over it.
- */
-export function applyFilesScope(ctx: SyncContext, _change: ScopeChange): void {
-  const { db, scope } = ctx;
-  let changed = false;
-  db.transaction(() => {
-    if (!scope.attachments) {
-      for (const blob of keptBlobs(db)) {
-        dropBlob(db, blob.jmap, blob.blob_id);
-        changed = true;
-      }
-      db.run("UPDATE file_node SET keep_error = 'notAllowed' WHERE pinned = 1 OR opened_at IS NOT NULL");
-      return;
-    }
-    const largest = Math.min(scope.maxAttachmentBytes, capBytes(ctx));
-    for (const blob of keptBlobs(db)) {
-      if (blob.pinned ? !overOrgLimit(ctx, Number(blob.size)) : Number(blob.size) <= largest) continue;
-      if (blob.pinned) {
-        dropBlob(db, blob.jmap, blob.blob_id);
-        setError(db, blob.jmap, blob.blob_id, 'tooLarge');
-      } else {
-        pushOut(db, blob);
-      }
-      changed = true;
-    }
-    const before = keptBlobs(db).length;
-    makeRoom(ctx, 0, { pinned: true, openedAt: 0 });
-    if (keptBlobs(db).length !== before) changed = true;
-    // Past the total with pins alone: the newest kept go first, as they would have been refused.
-    let pinned = pinnedBytes(db);
-    for (const blob of keptBlobs(db).filter((row) => row.pinned).sort((a, b) => Number(b.kept_at) - Number(a.kept_at))) {
-      if (pinned <= scope.maxTotalBytes) break;
-      pinned -= Number(blob.size);
-      dropBlob(db, blob.jmap, blob.blob_id);
-      setError(db, blob.jmap, blob.blob_id, 'overLimit');
-      changed = true;
-    }
-  });
-  if (changed) ctx.emit({ type: 'changed', account: ctx.account, collections: ['FileContent'] });
-}
-
-export function filesUsage(ctx: Pick<StoreContext, 'db'>): StoreUsagePart[] {
-  const row = ctx.db.get<{ bytes: number | null; items: number }>('SELECT SUM(size) AS bytes, COUNT(*) AS items FROM file_blob');
-  return [{ part: 'files', bytes: Number(row?.bytes ?? 0), items: Number(row?.items ?? 0) }];
-}
-
 // ---- Calls from the page ----
 
+const ROW_COLUMNS = 'jmap, id, name, kind, blob_id, stale_blob_id, size, json, pinned, opened_at, keep_error';
+
 function fileRow(db: StorageAdapter, { jmap, id }: NodeRef): NodeRow {
-  const row = db.get<NodeRow>('SELECT jmap, id, name, kind, blob_id, size, json, pinned, opened_at, keep_error FROM file_node WHERE jmap = ? AND id = ?', [jmap, id]);
+  const row = db.get<NodeRow>(`SELECT ${ROW_COLUMNS} FROM file_node WHERE jmap = ? AND id = ?`, [jmap, id]);
   if (!row) throw new MethodError('notFound', `No file ${id}`);
   if (!row.blob_id || row.kind !== 'file') throw new MethodError('invalidArguments', 'Only files can be kept');
   return row;
@@ -291,7 +433,7 @@ function fileRow(db: StorageAdapter, { jmap, id }: NodeRef): NodeRow {
 
 export function pinFile(ctx: StoreContext, ref: NodeRef): { kept: boolean } {
   const row = fileRow(ctx.db, ref);
-  ctx.db.run('UPDATE file_node SET pinned = 1, keep_error = NULL WHERE jmap = ? AND id = ?', [ref.jmap, ref.id]);
+  ctx.db.run('UPDATE file_node SET pinned = 1, pinned_at = ?, keep_error = NULL, keep_failures = 0, keep_failed_at = NULL WHERE jmap = ? AND id = ?', [ctx.now(), ref.jmap, ref.id]);
   const kept = hasBlob(ctx.db, ref.jmap, row.blob_id as string);
   if (!kept) void ctx.requestSync();
   return { kept };
@@ -301,14 +443,17 @@ export function pinFile(ctx: StoreContext, ref: NodeRef): { kept: boolean } {
 export function unpinFile(ctx: StoreContext, ref: NodeRef): void {
   const row = fileRow(ctx.db, ref);
   ctx.db.transaction(() => {
-    ctx.db.run('UPDATE file_node SET pinned = 0 WHERE jmap = ? AND id = ?', [ref.jmap, ref.id]);
-    if (!wantedRow(ctx.db, ref.jmap, row.blob_id as string)) dropBlob(ctx.db, ref.jmap, row.blob_id as string);
+    ctx.db.run('UPDATE file_node SET pinned = 0, pinned_at = NULL WHERE jmap = ? AND id = ?', [ref.jmap, ref.id]);
+    if (!wantedRow(ctx.db, ref.jmap, row.blob_id as string)) {
+      dropBlob(ctx.db, ref.jmap, row.blob_id as string);
+      if (row.stale_blob_id) dropBlob(ctx.db, ref.jmap, row.stale_blob_id);
+    }
   });
 }
 
 export function noteOpened(ctx: StoreContext, ref: NodeRef): { kept: boolean } {
   const row = fileRow(ctx.db, ref);
-  ctx.db.run('UPDATE file_node SET opened_at = ?, keep_error = NULL WHERE jmap = ? AND id = ?', [ctx.now(), ref.jmap, ref.id]);
+  ctx.db.run('UPDATE file_node SET opened_at = ?, keep_error = NULL, keep_failures = 0, keep_failed_at = NULL WHERE jmap = ? AND id = ?', [ctx.now(), ref.jmap, ref.id]);
   const kept = hasBlob(ctx.db, ref.jmap, row.blob_id as string);
   if (!kept) void ctx.requestSync();
   return { kept };
@@ -327,7 +472,10 @@ export interface KeptFileStatus {
   size: number | null;
   pinned: boolean;
   openedAt: number | null;
+  /** Its current content is here. */
   kept: boolean;
+  /** Only the version before is here, until the current one arrives. */
+  stale: boolean;
   /** Why it isn't kept: tooLarge, overLimit, notAllowed or download. */
   error: string | null;
 }
@@ -335,7 +483,7 @@ export interface KeptFileStatus {
 /** Pinned and recently opened files, and how much room their contents take. */
 export function keptStatus(ctx: StoreContext): { files: KeptFileStatus[]; bytes: number; capBytes: number } {
   const rows = ctx.db.all<NodeRow>(
-    `SELECT jmap, id, name, kind, blob_id, size, json, pinned, opened_at, keep_error FROM file_node
+    `SELECT ${ROW_COLUMNS} FROM file_node
      WHERE pinned = 1 OR opened_at IS NOT NULL ORDER BY pinned DESC, opened_at DESC, name`,
   );
   return {
@@ -347,6 +495,7 @@ export function keptStatus(ctx: StoreContext): { files: KeptFileStatus[]; bytes:
       pinned: Boolean(row.pinned),
       openedAt: row.opened_at === null ? null : Number(row.opened_at),
       kept: row.blob_id !== null && hasBlob(ctx.db, row.jmap, row.blob_id),
+      stale: row.stale_blob_id !== null,
       error: row.keep_error,
     })),
     bytes: keptBlobs(ctx.db).reduce((sum, blob) => sum + Number(blob.size), 0),

lib/store/files/methods.ts

@@ -12,10 +12,14 @@ function jmapOf(ctx: MethodContext, args: Record<string, unknown>): string {
   return (args.accountId as string | undefined) ?? ctx.session.primaryAccounts[FILES_CAPABILITY] ?? ctx.defaultJmap;
 }
 
-/** The last state from the server. Throws when the list was never read, so an empty answer always means empty. */
+/**
+ * The last state from the server. Throws until a first full pass finished, so
+ * an answer never passes off part of a tree as the whole. A later resync
+ * (phase `list`) goes on answering from the copy it's refreshing.
+ */
 function syncedState(ctx: MethodContext, jmap: string): string | null {
   const row = ctx.db.get<{ state: string | null; phase: string | null }>("SELECT state, phase FROM sync_state WHERE jmap = ? AND type = 'FileNode'", [jmap]);
-  if (!row || (!row.state && !row.phase)) throw new MethodError('serverUnavailable', 'Files not kept yet');
+  if (!row || (!row.state && row.phase !== 'list')) throw new MethodError('serverUnavailable', 'Files not kept yet');
   return row.state;
 }
 

lib/store/files/records.ts

@@ -29,6 +29,12 @@ export function writeNode(db: StorageAdapter, jmap: string, node: WireNode, gen:
      ON CONFLICT (jmap, id) DO UPDATE SET parent_id = excluded.parent_id, name = excluded.name, kind = excluded.kind,
        size = excluded.size, json = excluded.json, gen = excluded.gen,
        keep_error = CASE WHEN blob_id IS excluded.blob_id THEN keep_error END,
+       keep_failures = CASE WHEN blob_id IS excluded.blob_id THEN keep_failures ELSE 0 END,
+       stale_blob_id = CASE
+         WHEN blob_id IS excluded.blob_id THEN stale_blob_id
+         WHEN EXISTS (SELECT 1 FROM file_blob b WHERE b.jmap = file_node.jmap AND b.blob_id = file_node.blob_id AND b.complete = 1) THEN blob_id
+         ELSE stale_blob_id
+       END,
        blob_id = excluded.blob_id`,
     [
       jmap,

lib/store/files/schema.ts

@@ -17,20 +17,29 @@ export const FILE_MIGRATIONS: readonly Migration[] = [
         json TEXT NOT NULL,
         gen INTEGER NOT NULL DEFAULT 0,
         pinned INTEGER NOT NULL DEFAULT 0,
+        pinned_at INTEGER,
         -- When the page last opened it. Cleared when its content is pushed out.
         opened_at INTEGER,
         -- Why the content isn't kept: tooLarge, overLimit, notAllowed or download.
         keep_error TEXT,
+        -- Failed downloads in a row, and when the last one was.
+        keep_failures INTEGER NOT NULL DEFAULT 0,
+        keep_failed_at INTEGER,
+        -- The kept content of the version before blob_id, until blob_id is kept.
+        stale_blob_id TEXT,
         PRIMARY KEY (jmap, id)
       ) WITHOUT ROWID;
       CREATE INDEX file_node_parent ON file_node (jmap, parent_id);
       CREATE INDEX file_node_blob ON file_node (jmap, blob_id);
+      CREATE INDEX file_node_stale ON file_node (jmap, stale_blob_id);
       CREATE TABLE file_blob (
         jmap TEXT NOT NULL,
         blob_id TEXT NOT NULL,
         type TEXT NOT NULL,
         size INTEGER NOT NULL,
         kept_at INTEGER NOT NULL,
+        -- 0 while its chunks are still coming in. Nothing reads it until it's 1.
+        complete INTEGER NOT NULL DEFAULT 0,
         PRIMARY KEY (jmap, blob_id)
       ) WITHOUT ROWID;
       -- In pieces: a whole file bound at once would grow the WASM heap by its

lib/store/files/sync.ts

@@ -6,30 +6,32 @@
  * 0.16 silently stops at maxObjectsInGet and 1.0 refuses a larger account with
  * requestTooLarge. Past that, ids come from paging FileNode/query. Before
  * 0.16.6 that query leaves out folders, so parents still unknown afterwards are
- * read by id. The cursor is in sync_state, so a reload continues.
+ * read by id. The cursor is in sync_state, so a reload continues. Query pages
+ * go from the last node read (`anchor`): a position moves when an earlier node
+ * is deleted, and a skipped node would be swept at the end.
  */
 
 import type { SyncContext, Syncer } from '../area';
 import { runChangesLoop } from '../engine/changes';
+import { requestLimits } from '../engine/limits';
 import { JmapMethodError, responseFor, type MethodCall } from '../engine/transport';
 import { dropOrphanBlobs, fetchKeptFiles } from './keep';
 import { FILE_NODE_PROPERTIES, FILES_CAPABILITY, FILES_USING, removeNode, writeNode, type WireNode } from './records';
 
-const MAX_GET = 500;
-const GETS_PER_REQUEST = 8;
+const MOST_GETS_PER_REQUEST = 8;
 const QUERY_PAGE = 5000;
 
 function getBatch(ctx: SyncContext): number {
-  const core = ctx.session.capabilities['urn:ietf:params:jmap:core'] as { maxObjectsInGet?: number } | undefined;
-  return Math.min(MAX_GET, core?.maxObjectsInGet || MAX_GET);
+  return requestLimits(ctx.session).maxObjectsInGet;
 }
 
 async function fetchNodes(ctx: SyncContext, jmap: string, ids: string[]): Promise<WireNode[]> {
   const size = getBatch(ctx);
+  const perRequest = Math.min(MOST_GETS_PER_REQUEST, requestLimits(ctx.session).maxCallsInRequest);
   const nodes: WireNode[] = [];
   for (let at = 0; at < ids.length;) {
     const calls: MethodCall[] = [];
-    for (let i = 0; i < GETS_PER_REQUEST && at < ids.length; i += 1, at += size) {
+    for (let i = 0; i < perRequest && at < ids.length; i += 1, at += size) {
       calls.push(['FileNode/get', { accountId: jmap, ids: ids.slice(at, at + size), properties: FILE_NODE_PROPERTIES }, `g${i}`]);
     }
     const response = await ctx.transport.request(calls, FILES_USING);
@@ -38,14 +40,16 @@ async function fetchNodes(ctx: SyncContext, jmap: string, ids: string[]): Promis
   return nodes;
 }
 
-interface ListCursor { step: 'all' | 'query' | 'parents'; position: number; gen: number; captured: string }
+interface ListCursor { step: 'all' | 'query' | 'parents'; position: number; anchor: string | null; gen: number; captured: string }
 
 async function startFullList(ctx: SyncContext, jmap: string): Promise<ListCursor> {
   const response = await ctx.transport.request([['FileNode/get', { accountId: jmap, ids: [], properties: ['id'] }, 's']], FILES_USING);
   const captured = String(responseFor(response, 's')[1].state);
   const gen = Number(ctx.db.get<{ gen: number | null }>('SELECT MAX(gen) AS gen FROM file_node WHERE jmap = ?', [jmap])?.gen ?? 0) + 1;
-  const cursor: ListCursor = { step: 'all', position: 0, gen, captured };
-  ctx.db.transaction(() => ctx.setSyncState(jmap, 'FileNode', { state: null, phase: 'list', cursor: { ...cursor } }));
+  const cursor: ListCursor = { step: 'all', position: 0, anchor: null, gen, captured };
+  // Answers wait for the end of a first pass (phase `first`). A resync has an older copy to give.
+  const phase = ctx.syncState(jmap, 'FileNode').state ? 'list' : 'first';
+  ctx.db.transaction(() => ctx.setSyncState(jmap, 'FileNode', { state: null, phase, cursor: { ...cursor } }));
   return cursor;
 }
 
@@ -76,12 +80,24 @@ async function runFullList(ctx: SyncContext, jmap: string, start: ListCursor): P
 
   while (cursor.step === 'query') {
     if (ctx.stopped()) return false;
-    const response = await ctx.transport.request([['FileNode/query', { accountId: jmap, filter: {}, position: cursor.position, limit: QUERY_PAGE, calculateTotal: true }, 'q']], FILES_USING);
-    const [, query] = responseFor(response, 'q');
-    const ids = query.ids as string[];
+    const from = cursor.anchor === null ? { position: 0 } : { anchor: cursor.anchor, anchorOffset: 1 };
+    const response = await ctx.transport.request([['FileNode/query', { accountId: jmap, filter: {}, ...from, limit: QUERY_PAGE, calculateTotal: true }, 'q']], FILES_USING);
+    let query: { ids: string[]; position?: number; total?: number };
+    try {
+      [, query] = responseFor(response, 'q');
+    } catch (error) {
+      if (!(error instanceof JmapMethodError && error.type === 'anchorNotFound')) throw error;
+      // The node it went from is gone. Going through again costs requests, not nodes.
+      cursor.anchor = null;
+      cursor.position = 0;
+      saveStep(ctx, jmap, [], cursor);
+      continue;
+    }
+    const ids = query.ids;
     const missing = ids.filter((id) => Number(ctx.db.get<{ gen: number }>('SELECT gen FROM file_node WHERE jmap = ? AND id = ?', [jmap, id])?.gen ?? -1) < cursor.gen);
     const nodes = await fetchNodes(ctx, jmap, missing);
-    cursor.position += ids.length;
+    cursor.position = Number(query.position ?? 0) + ids.length;
+    if (ids.length > 0) cursor.anchor = ids[ids.length - 1];
     // The server may clamp the limit, so only an empty page or the total ends this.
     if (ids.length === 0 || (typeof query.total === 'number' && cursor.position >= query.total)) cursor.step = 'parents';
     saveStep(ctx, jmap, nodes, cursor);
@@ -140,7 +156,7 @@ export const filesSyncer: Syncer = {
   capability: FILES_CAPABILITY,
   async run(ctx, jmap) {
     const row = ctx.syncState(jmap, 'FileNode');
-    let cursor = row.phase === 'list' ? row.cursor as unknown as ListCursor : null;
+    let cursor = row.phase === 'list' || row.phase === 'first' ? row.cursor as unknown as ListCursor : null;
     if (!cursor && !row.state) cursor = await startFullList(ctx, jmap);
     if (!cursor && row.state) {
       const outcome = await syncDelta(ctx, jmap, row.state, row.target);
@@ -154,7 +170,8 @@ export const filesSyncer: Syncer = {
       if (after.state && (await syncDelta(ctx, jmap, after.state, after.target)) === 'stopped') return;
     }
     if (ctx.stopped()) return;
-    ctx.db.transaction(() => dropOrphanBlobs(ctx.db));
     await fetchKeptFiles(ctx, jmap);
+    // After the fetching: a file's old content stays until the new one is in.
+    ctx.db.transaction(() => dropOrphanBlobs(ctx.db));
   },
 };