From 418bda4642de0334c0c30ce992cf680c4a4fa22c Mon Sep 17 00:00:00 2001 From: Linus Rath <139418639+rathlinus@users.noreply.github.com> Date: Tue, 3 Nov 2026 13:41:03 +0100 Subject: [PATCH] fix: stream kept files to the store, keep the old copy until the new one is in, and apply a tighter scope at once --- lib/store/__tests__/files.test.ts | 287 ++++++++++++++++-- lib/store/__tests__/helpers/fake-files.ts | 52 +++- lib/store/__tests__/helpers/fake-jmap.ts | 4 + lib/store/__tests__/transport.test.ts | 129 ++++++++ lib/store/area.ts | 2 + lib/store/engine/account-engine.ts | 17 +- lib/store/engine/transport.ts | 169 +++++++++-- lib/store/files/index.ts | 3 +- lib/store/files/keep.ts | 353 ++++++++++++++++------ lib/store/files/methods.ts | 8 +- lib/store/files/records.ts | 6 + lib/store/files/schema.ts | 9 + lib/store/files/sync.ts | 47 ++- 13 files changed, 926 insertions(+), 160 deletions(-) create mode 100644 lib/store/__tests__/transport.test.ts diff --git a/lib/store/__tests__/files.test.ts b/lib/store/__tests__/files.test.ts index acf81b483..acff96d1c 100644 --- a/lib/store/__tests__/files.test.ts +++ b/lib/store/__tests__/files.test.ts @@ -54,6 +54,9 @@ async function call(engine: AccountEngine, name: string, args: Record ({ jmap, id }); +const bytesOf = async (kept: unknown): Promise => 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 () => { @@ -341,15 +398,208 @@ 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('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); @@ -382,9 +632,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 () => { @@ -398,10 +648,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); }); @@ -426,8 +676,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; @@ -436,7 +686,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(); }); @@ -506,11 +756,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); }); @@ -526,7 +776,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); diff --git a/lib/store/__tests__/helpers/fake-files.ts b/lib/store/__tests__/helpers/fake-files.ts index 0ff22f4ed..19786cc8e 100644 --- a/lib/store/__tests__/helpers/fake-files.ts +++ b/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(); + /** Bytes each blob's body actually delivered, last time. */ + readonly delivered = new Map(); + /** 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 { + 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 { 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((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; + } } } diff --git a/lib/store/__tests__/helpers/fake-jmap.ts b/lib/store/__tests__/helpers/fake-jmap.ts index 1fd274c04..7d84b66e2 100644 --- a/lib/store/__tests__/helpers/fake-jmap.ts +++ b/lib/store/__tests__/helpers/fake-jmap.ts @@ -279,6 +279,10 @@ export class FakeJmap implements JmapTransport { return new Uint8Array(); } + async downloadStream(jmap: string, blobId: string, type: string, name: string, sink: (part: Uint8Array) => void): Promise { + 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 }; } diff --git a/lib/store/__tests__/transport.test.ts b/lib/store/__tests__/transport.test.ts new file mode 100644 index 000000000..780837583 --- /dev/null +++ b/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) => void | Promise; + +/** 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({ + 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): 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({ + 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); + }); +}); diff --git a/lib/store/area.ts b/lib/store/area.ts index 9bd11fd35..c9c9f0688 100644 --- a/lib/store/area.ts +++ b/lib/store/area.ts @@ -108,4 +108,6 @@ export interface StoreArea { methods: Readonly>; syncers: readonly Syncer[]; calls?: Readonly>; + /** At start and when the scope changes: a tighter policy applies at once, not at the next sync. */ + scopeChanged?(ctx: StoreContext): void; } diff --git a/lib/store/engine/account-engine.ts b/lib/store/engine/account-engine.ts index a3d56298e..a8510346d 100644 --- a/lib/store/engine/account-engine.ts +++ b/lib/store/engine/account-engine.ts @@ -34,6 +34,7 @@ export class AccountEngine { private scope: StoreScope; private readonly transport: JmapTransport; private readonly syncers: readonly Syncer[]; + private readonly areas: readonly StoreArea[]; private readonly methods: ReadonlyMap; private readonly calls = new Map(); private readonly emitter: (event: StoreEvent) => void; @@ -51,6 +52,7 @@ export class AccountEngine { this.session = options.session; this.scope = options.scope; this.transport = options.transport; + this.areas = options.areas; this.syncers = options.areas.flatMap((area) => area.syncers); this.methods = buildMethodTable(options.areas); for (const area of options.areas) { @@ -65,6 +67,12 @@ export class AccountEngine { this.invertedKeywordSort = options.invertedKeywordSort ?? false; migrate(this.db, [...CORE_MIGRATIONS, ...options.areas.flatMap((area) => area.migrations)]); metaSet(this.db, 'session', JSON.stringify(options.session)); + this.applyScope(); + } + + private applyScope(): void { + const ctx = this.storeContext(); + for (const area of this.areas) area.scopeChanged?.(ctx); } /** The session kept from the last time, for a start without a network. */ @@ -82,6 +90,7 @@ export class AccountEngine { setScope(scope: StoreScope): void { this.scope = scope; metaSet(this.db, 'scope', JSON.stringify(scope)); + this.applyScope(); } setInvertedKeywordSort(inverted: boolean): void { @@ -188,11 +197,15 @@ export class AccountEngine { } private syncContext(): SyncContext { + const scope = () => this.scope; return { account: this.account, db: this.db, session: this.session, - scope: this.scope, + // A pass can take long, and a tighter policy applies to the rest of it. + get scope() { + return scope(); + }, transport: this.transport, now: this.now, emit: this.emitter, @@ -240,6 +253,8 @@ export class AccountEngine { /** Stops syncing and waits for the current batch. The database stays open for the caller to close. */ async close(): Promise { this.closed = true; + // A large download could otherwise hold up removing the account. + this.transport.abort?.(); await this.running?.catch(() => {}); } } diff --git a/lib/store/engine/transport.ts b/lib/store/engine/transport.ts index cc55ff300..d1edaee6e 100644 --- a/lib/store/engine/transport.ts +++ b/lib/store/engine/transport.ts @@ -34,6 +34,8 @@ export class TransportError extends Error { /** The `type` of an RFC 7807 problem body, when there was one. */ readonly problem?: string, readonly retryAfterMs?: number, + /** The connection worked and this transfer broke or stalled: it says little about the next one. */ + readonly interrupted?: boolean, ) { super(message); this.name = 'TransportError'; @@ -43,7 +45,14 @@ export class TransportError extends Error { export interface JmapTransport { request(calls: MethodCall[], using: string[]): Promise; download(jmap: string, blobId: string, type: string, name: string): Promise; + /** + * The same, piece by piece, so a large file isn't held in memory. An error + * thrown by `sink` ends the download and comes out as it is. + */ + downloadStream(jmap: string, blobId: string, type: string, name: string, sink: (part: Uint8Array) => void): Promise; upload(jmap: string, data: Uint8Array, type: string): Promise<{ blobId: string; type: string; size: number }>; + /** Ends what's running, now and for good. The engine calls this when it closes. */ + abort?(): void; } export interface FetchTransportOptions { @@ -56,23 +65,72 @@ export interface FetchTransportOptions { } const REQUEST_TIMEOUT_MS = 120_000; +/** A download may take as long as it takes, as long as bytes keep coming. */ +const DOWNLOAD_IDLE_MS = 60_000; function fillTemplate(template: string, values: Record): string { return template.replace(/\{(\w+)\}/g, (whole, key: string) => (key in values ? encodeURIComponent(values[key]) : whole)); } +/** One transfer's way to be stopped: by its time limit, or by the transport closing. */ +interface Transfer { + signal: AbortSignal; + /** Bytes came: for a download, the time limit starts over. */ + alive(): void; + /** Whether the time limit ended it. */ + timedOut(): boolean; + done(): void; +} + +function messageOf(error: unknown, fallback: string): string { + return error instanceof Error && error.message ? error.message : fallback; +} + export class FetchTransport implements JmapTransport { + private readonly running = new Set(); + private aborted = false; + constructor(private readonly options: FetchTransportOptions) {} - private async fetchWithAuth(url: string, init: RequestInit): Promise { + abort(): void { + this.aborted = true; + for (const controller of this.running) controller.abort(); + } + + private transfer(ms: number, idle: boolean): Transfer { + const controller = new AbortController(); + this.running.add(controller); + let timedOut = false; + const arm = () => setTimeout(() => { + timedOut = true; + controller.abort(); + }, ms); + let timer = arm(); + if (this.aborted) controller.abort(); + return { + signal: controller.signal, + alive: () => { + if (!idle) return; + clearTimeout(timer); + timer = arm(); + }, + timedOut: () => timedOut, + done: () => { + clearTimeout(timer); + this.running.delete(controller); + }, + }; + } + + private async fetchWithAuth(url: string, init: RequestInit, transfer: Transfer): Promise { const doFetch = this.options.fetch ?? fetch; const send = async (header: string | null): Promise => { if (!header) throw new TransportError('auth', 'No credentials yet'); - const signal = AbortSignal.timeout(REQUEST_TIMEOUT_MS); + if (this.aborted) throw new TransportError('network', 'The store is closing'); try { - return await doFetch(url, { ...init, signal, headers: { ...init.headers, Authorization: header } }); + return await doFetch(url, { ...init, signal: transfer.signal, headers: { ...init.headers, Authorization: header } }); } catch (error) { - throw new TransportError('network', error instanceof Error ? error.message : 'Network failure'); + throw new TransportError('network', messageOf(error, 'Network failure'), undefined, undefined, undefined, transfer.timedOut()); } }; let response = await send(this.options.auth()); @@ -87,40 +145,97 @@ export class FetchTransport implements JmapTransport { return response; } + /** A JSON body. A connection dropping halfway is a network failure, not a bad answer. */ + private async readJson(response: Response): Promise { + try { + return await response.json() as T; + } catch (error) { + if (error instanceof SyntaxError) throw new TransportError('http', 'The server sent something that is not JSON', response.status); + throw new TransportError('network', messageOf(error, 'The response broke off'), undefined, undefined, undefined, true); + } + } + async request(calls: MethodCall[], using: string[]): Promise { - const response = await this.fetchWithAuth(this.options.session().apiUrl, { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ using, methodCalls: calls }), - }); - if (!response.ok) { - let problem: string | undefined; + const transfer = this.transfer(REQUEST_TIMEOUT_MS, false); + try { + const response = await this.fetchWithAuth(this.options.session().apiUrl, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ using, methodCalls: calls }), + }, transfer); + if (!response.ok) { + let problem: string | undefined; + try { + problem = (await response.json() as { type?: string }).type; + } catch { + // Not a problem document. + } + throw new TransportError('http', `JMAP request failed: ${response.status}`, response.status, problem); + } + return await this.readJson(response); + } finally { + transfer.done(); + } + } + + async downloadStream(jmap: string, blobId: string, type: string, name: string, sink: (part: Uint8Array) => void): Promise { + const url = fillTemplate(this.options.session().downloadUrl, { accountId: jmap, blobId, type, name }); + const transfer = this.transfer(DOWNLOAD_IDLE_MS, true); + try { + const response = await this.fetchWithAuth(url, { method: 'GET' }, transfer); + if (!response.ok) throw new TransportError('http', `Download failed: ${response.status}`, response.status); + if (!response.body) return; + const reader = response.body.getReader(); try { - problem = (await response.json() as { type?: string }).type; - } catch { - // Not a problem document. + for (;;) { + let read: Awaited>; + try { + read = await reader.read(); + } catch (error) { + throw new TransportError('network', messageOf(error, 'The download broke off'), undefined, undefined, undefined, !this.aborted); + } + if (read.done) return; + transfer.alive(); + sink(read.value); + } + } finally { + void reader.cancel().catch(() => {}); } - throw new TransportError('http', `JMAP request failed: ${response.status}`, response.status, problem); + } finally { + transfer.done(); } - return await response.json() as JmapResponse; } async download(jmap: string, blobId: string, type: string, name: string): Promise { - const url = fillTemplate(this.options.session().downloadUrl, { accountId: jmap, blobId, type, name }); - const response = await this.fetchWithAuth(url, { method: 'GET' }); - if (!response.ok) throw new TransportError('http', `Download failed: ${response.status}`, response.status); - return new Uint8Array(await response.arrayBuffer()); + const parts: Uint8Array[] = []; + let size = 0; + await this.downloadStream(jmap, blobId, type, name, (part) => { + parts.push(part); + size += part.length; + }); + const data = new Uint8Array(size); + let at = 0; + for (const part of parts) { + data.set(part, at); + at += part.length; + } + return data; } async upload(jmap: string, data: Uint8Array, type: string): Promise<{ blobId: string; type: string; size: number }> { const url = fillTemplate(this.options.session().uploadUrl, { accountId: jmap }); - const response = await this.fetchWithAuth(url, { - method: 'POST', - headers: { 'Content-Type': type || 'application/octet-stream' }, - body: data as BodyInit, - }); - if (!response.ok) throw new TransportError('http', `Upload failed: ${response.status}`, response.status); - return await response.json() as { blobId: string; type: string; size: number }; + const transfer = this.transfer(REQUEST_TIMEOUT_MS, false); + try { + const response = await this.fetchWithAuth(url, { + method: 'POST', + headers: { 'Content-Type': type || 'application/octet-stream' }, + body: data as BodyInit, + }, transfer); + if (!response.ok) throw new TransportError('http', `Upload failed: ${response.status}`, response.status); + return await this.readJson<{ blobId: string; type: string; size: number }>(response); + } finally { + transfer.done(); + } } } diff --git a/lib/store/files/index.ts b/lib/store/files/index.ts index 0957ced89..97f60baef 100644 --- a/lib/store/files/index.ts +++ b/lib/store/files/index.ts @@ -1,5 +1,5 @@ import type { StoreArea } from '../area'; -import { fileBlob, keptStatus, noteOpened, pinFile, readFile, unpinFile, type NodeRef } from './keep'; +import { applyScope, fileBlob, keptStatus, noteOpened, pinFile, readFile, unpinFile, type NodeRef } from './keep'; import { FILE_METHODS } from './methods'; import { FILE_MIGRATIONS } from './schema'; import { filesSyncer } from './sync'; @@ -11,6 +11,7 @@ export const filesArea: StoreArea = { migrations: FILE_MIGRATIONS, methods: FILE_METHODS, syncers: [filesSyncer], + scopeChanged: applyScope, calls: { /** Keep this file's content on the device until it's unpinned. */ 'files.pin': (ctx, params: NodeRef) => pinFile(ctx, params), diff --git a/lib/store/files/keep.ts b/lib/store/files/keep.ts index a11b93455..6ef464f20 100644 --- a/lib/store/files/keep.ts +++ b/lib/store/files/keep.ts @@ -5,11 +5,18 @@ * What gets kept: * - A pinned file stays until it's unpinned or removed on the server. The cap * doesn't push it out and the size limit for one file doesn't apply. The - * account's total limit (scope.maxTotalBytes, which a policy can lower) does. + * account's total limit (scope.maxTotalBytes, which a policy can lower) does: + * past it, 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'; @@ -19,8 +26,17 @@ 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 } @@ -30,6 +46,7 @@ interface NodeRow { name: string; kind: string; blob_id: string | null; + stale_blob_id: string | null; size: number | null; json: string; pinned: number; @@ -43,6 +60,7 @@ interface BlobRow { size: number; kept_at: number; pinned: number; + pinned_at: number | null; opened_at: number | null; } @@ -50,41 +68,52 @@ function capBytes(ctx: Pick): 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, 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): 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], { 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, 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( - `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`, ); } @@ -94,16 +123,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); @@ -117,113 +146,252 @@ 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; } -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. Runs at start, when the scope changes and before fetching. */ +export function applyScope(ctx: StoreContext): void { + const { db } = ctx; + 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'"); + // 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 }); + }); +} + +/** 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( - `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 && pinnedBytes(db, now.stale_blob_id) + size > ctx.scope.maxTotalBytes) { + dropBlob(db, jmap, blobId); + setError(db, jmap, blobId, '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 { + const { db } = ctx; + const blobId = want.blob_id; + const type = nodeMediaType(JSON.parse(want.json) as WireNode); + const limit = want.pinned ? ctx.scope.maxTotalBytes - pinnedBytes(db, want.stale_blob_id) : 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 { + await ctx.transport.downloadStream(jmap, blobId, type, want.name, (part) => writer.push(part)); + 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 ? '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 { 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]); + } }); + applyScope(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 && 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 ? 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 ? 'overLimit' : 'tooLarge'); - 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'] }); } // ---- 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('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(`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; @@ -231,7 +399,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 }; @@ -241,14 +409,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 }; @@ -267,7 +438,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; } @@ -275,7 +449,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( - `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 { @@ -287,6 +461,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), diff --git a/lib/store/files/methods.ts b/lib/store/files/methods.ts index 0945a33eb..49309a8b0 100644 --- a/lib/store/files/methods.ts +++ b/lib/store/files/methods.ts @@ -12,10 +12,14 @@ function jmapOf(ctx: MethodContext, args: Record): 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; } diff --git a/lib/store/files/records.ts b/lib/store/files/records.ts index fd5cbcb51..40d624460 100644 --- a/lib/store/files/records.ts +++ b/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, diff --git a/lib/store/files/schema.ts b/lib/store/files/schema.ts index 9a3f20e35..89a6e8cb3 100644 --- a/lib/store/files/schema.ts +++ b/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 diff --git a/lib/store/files/sync.ts b/lib/store/files/sync.ts index cc9d99918..5b853c229 100644 --- a/lib/store/files/sync.ts +++ b/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 { 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 { 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)); }, };