mirror of
https://github.com/bulwarkmail/webmail.git
synced 2026-11-04 08:15:41 +00:00
fix: stream kept files to the store, keep the old copy until the new one is in, and apply a tighter scope at once
This commit is contained in:
Linus Rath
@@ -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 () => {
|
||||
@@ -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);
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<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 };
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -108,4 +108,6 @@ export interface StoreArea {
|
||||
methods: Readonly<Record<string, MethodHandler>>;
|
||||
syncers: readonly Syncer[];
|
||||
calls?: Readonly<Record<string, AreaCall>>;
|
||||
/** At start and when the scope changes: a tighter policy applies at once, not at the next sync. */
|
||||
scopeChanged?(ctx: StoreContext): void;
|
||||
}
|
||||
|
||||
@@ -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<string, MethodHandler>;
|
||||
private readonly calls = new Map<string, AreaCall>();
|
||||
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<void> {
|
||||
this.closed = true;
|
||||
// A large download could otherwise hold up removing the account.
|
||||
this.transport.abort?.();
|
||||
await this.running?.catch(() => {});
|
||||
}
|
||||
}
|
||||
|
||||
+142
-27
@@ -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<JmapResponse>;
|
||||
download(jmap: string, blobId: string, type: string, name: string): Promise<Uint8Array>;
|
||||
/**
|
||||
* 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<void>;
|
||||
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, string>): 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<AbortController>();
|
||||
private aborted = false;
|
||||
|
||||
constructor(private readonly options: FetchTransportOptions) {}
|
||||
|
||||
private async fetchWithAuth(url: string, init: RequestInit): Promise<Response> {
|
||||
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<Response> {
|
||||
const doFetch = this.options.fetch ?? fetch;
|
||||
const send = async (header: string | null): Promise<Response> => {
|
||||
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<T>(response: Response): Promise<T> {
|
||||
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<JmapResponse> {
|
||||
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<JmapResponse>(response);
|
||||
} finally {
|
||||
transfer.done();
|
||||
}
|
||||
}
|
||||
|
||||
async downloadStream(jmap: string, blobId: string, type: string, name: string, sink: (part: Uint8Array) => void): Promise<void> {
|
||||
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<ReturnType<typeof reader.read>>;
|
||||
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<Uint8Array> {
|
||||
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();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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),
|
||||
|
||||
+264
-89
@@ -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<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`,
|
||||
);
|
||||
}
|
||||
@@ -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<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 && 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<boolean> {
|
||||
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<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]);
|
||||
}
|
||||
});
|
||||
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<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;
|
||||
@@ -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<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 {
|
||||
@@ -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),
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
+32
-15
@@ -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));
|
||||
},
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user