From 9fbe33004bf71f5f10dcf57e4ca989b6956a4efe Mon Sep 17 00:00:00 2001 From: Linus Rath <139418639+rathlinus@users.noreply.github.com> Date: Tue, 3 Nov 2026 06:20:39 +0100 Subject: [PATCH] fix: send contact edits that remove entries and stop reading own edits as conflicts An edit that removed an entry was dropped at flush, because a server value with more keys counted as already applied. Other fixes in the same path: - the local copy after a flush is the server's read-back, never what was sent, since Stalwart 1.0 stores a data: URI as a blob - a delete or a newer edit after a lost response no longer reads as a change on the server, and an edit made during a refused send stays - a create only becomes a card with its uid if that card holds what was sent, otherwise it is a refused conflict - a refused or failed flush no longer stops contacts syncing - request limits come from the session - answers wait for the first full pass - paging goes by anchor, so a deleted card doesn't make the next one get skipped and swept contacts.test.ts: "continues a stopped full sync" now checks the anchor instead of position 2000, because the cursor no longer pages by position. --- lib/store/__tests__/contacts.test.ts | 392 ++++++++++++++++++- lib/store/__tests__/helpers/fake-contacts.ts | 51 ++- lib/store/contacts/methods.ts | 8 +- lib/store/contacts/patch.ts | 35 +- lib/store/contacts/pending.ts | 353 +++++++++++------ lib/store/contacts/schema.ts | 2 + lib/store/contacts/sync.ts | 83 ++-- lib/store/engine/limits.ts | 31 ++ 8 files changed, 788 insertions(+), 167 deletions(-) create mode 100644 lib/store/engine/limits.ts diff --git a/lib/store/__tests__/contacts.test.ts b/lib/store/__tests__/contacts.test.ts index 2331572e2..c05334fab 100644 --- a/lib/store/__tests__/contacts.test.ts +++ b/lib/store/__tests__/contacts.test.ts @@ -6,19 +6,19 @@ import path from 'node:path'; import { textFilter } from '@/lib/jmap/search-utils'; import { DEFAULT_SCOPE, type StoreEvent } from '../area'; import { contactsArea } from '../contacts'; -import { applyPatch, mergePatches } from '../contacts/patch'; +import { applyPatch, coversJson, mergePatches, sameJson } from '../contacts/patch'; import { AccountEngine } from '../engine/account-engine'; -import type { MethodCall } from '../engine/transport'; +import type { MethodCall, StoreSession } from '../engine/transport'; import type { StorageAdapter } from '../storage'; import { JMAP_ACCOUNT } from './helpers/fake-jmap'; import { CONTACTS_SESSION, FakeContacts, MAIL_ONLY_ACCOUNT, SHARED_ACCOUNT } from './helpers/fake-contacts'; import { openFileStore, openMemoryStore } from './helpers/node-adapter'; -function engineFor(server: FakeContacts, db: StorageAdapter = openMemoryStore(), emit?: (e: StoreEvent) => void) { +function engineFor(server: FakeContacts, db: StorageAdapter = openMemoryStore(), emit?: (e: StoreEvent) => void, session: StoreSession = CONTACTS_SESSION) { return new AccountEngine({ account: 'a@fake.test', db, - session: CONTACTS_SESSION, + session, transport: server, areas: [contactsArea], scope: DEFAULT_SCOPE, @@ -107,9 +107,11 @@ describe('contacts sync', () => { const second = openFileStore(file); await engineFor(server, second).requestSync(); expect(count(second, 'contact_card')).toBe(2600); - const positions = server.calls.filter(([name]) => name === 'ContactCard/query').map(([, args]) => args.position); - expect(positions[0]).toBe(2000); second.close(); + // It goes on from the last card read, not from the start. + const first_ = server.calls.find(([name]) => name === 'ContactCard/query')?.[1]; + expect(first_?.anchor).toBeTruthy(); + expect(first_?.anchorOffset).toBe(1); } finally { rmSync(dir, { recursive: true, force: true }); } @@ -670,4 +672,382 @@ describe('edits while sending', () => { expect(await engine.areaCall('contacts.pending', undefined)).toBe(0); }); }); + +const mine = (engine: AccountEngine, patch: Record, id: string) => call(engine, 'ContactCard/set', { update: { [id]: patch } }); +const conflictsOf = async (engine: AccountEngine) => await engine.areaCall('contacts.conflicts', undefined) as Array>; +const pendingOf = (engine: AccountEngine) => engine.areaCall('contacts.pending', undefined); + +describe('edits that make a card smaller', () => { + it.each<[string, Record, Record, Record]>([ + ['the last email', { emails: { e1: { address: 'a@example.org' } } }, { emails: {} }, { emails: {} }], + ['one of two emails', { emails: { e1: { address: 'a@example.org' }, e2: { address: 'b@example.org' } } }, { emails: { e1: { address: 'a@example.org' } } }, { emails: { e1: { address: 'a@example.org' } } }], + ['an email by pointer', { emails: { e1: { address: 'a@example.org' }, e2: { address: 'b@example.org' } } }, { 'emails/e2': null }, { emails: { e1: { address: 'a@example.org' } } }], + ['a keyword', { keywords: { work: true, home: true } }, { keywords: { work: true } }, { keywords: { work: true } }], + ['an address book', { addressBookIds: { book1: true, book2: true } }, { addressBookIds: { book2: true } }, { addressBookIds: { book2: true } }], + ['a group member', { kind: 'group', members: { 'urn:uuid:a': true, 'urn:uuid:b': true } }, { members: { 'urn:uuid:a': true } }, { members: { 'urn:uuid:a': true } }], + ])('sends the removal of %s', async (_label, start, patch, expected) => { + const server = new FakeContacts(['book1', 'book2']); + const card = server.addCard(start); + const { engine } = await synced(server); + await mine(engine, patch, card.id); + await engine.requestSync(); + expect(server.account().cards.get(card.id)).toMatchObject(expected); + for (const [key, value] of Object.entries(expected)) expect(server.account().cards.get(card.id)?.[key]).toEqual(value); + expect(await conflictsOf(engine)).toEqual([]); + expect(await pendingOf(engine)).toBe(0); + }); + + it('reads a key only the server has as its own, and one it had before as not yet removed', () => { + const wanted = { e1: { address: 'a' } }; + expect(coversJson(wanted, { e1: { address: 'a' }, added: { address: 'x' } }, { e1: { address: 'a' } })).toBe(true); + expect(coversJson(wanted, { e1: { address: 'a' }, e2: { address: 'b' } }, { e1: { address: 'a' }, e2: { address: 'b' } })).toBe(false); + expect(coversJson({ e: { '@type': 'x' } }, { e: { '@type': 'x', extra: 1 } }, undefined)).toBe(true); + // Stalwart 1.0 keeps a data: URI as a blob. + expect(coversJson({ m: { kind: 'photo', uri: 'data:image/png;base64,AA' } }, { m: { kind: 'photo', blobId: 'b1' } }, undefined)).toBe(true); + expect(coversJson({ m: { kind: 'logo', uri: 'data:image/png;base64,AA' } }, { m: { kind: 'photo', blobId: 'b1' } }, undefined)).toBe(false); + expect(sameJson({}, undefined)).toBe(true); + }); +}); + +describe('flush failures', () => { + it('still syncs after the server refuses the set as a whole, and records the edits', async () => { + const server = new FakeContacts(); + const card = server.addCard({ note: 'server' }); + const { engine } = await synced(server); + await mine(engine, { note: 'mine' }, card.id); + await call(engine, 'ContactCard/set', { create: { n: { name: { full: 'New' }, addressBookIds: { book1: true } } } }); + server.failSet = () => ({ jmapError: 'forbidden' }); + server.addCard({ name: { full: 'From the server' } }); + await engine.requestSync(); + expect(await search(engine, 'from the server')).toEqual(['From the server']); + expect((await conflictsOf(engine)).map((c) => c.kind)).toEqual(['refused', 'refused']); + expect(await pendingOf(engine)).toBe(0); + expect((await call(engine, 'ContactCard/get', { ids: [card.id] })).list[0].note).toBe('server'); + }); + + it('keeps edits and still syncs when the failure says nothing about them', async () => { + const server = new FakeContacts(); + const card = server.addCard({ note: 'server' }); + const { engine } = await synced(server); + await mine(engine, { note: 'mine' }, card.id); + server.failSet = () => ({ jmapError: 'serverFail' }); + server.addCard({ name: { full: 'From the server' } }); + await engine.requestSync(); + expect(await search(engine, 'from the server')).toEqual(['From the server']); + expect(await conflictsOf(engine)).toEqual([]); + expect(await pendingOf(engine)).toBe(1); + server.failSet = null; + await engine.requestSync(); + expect(server.account().cards.get(card.id)?.note).toBe('mine'); + }); + + it('takes the rest of a batch through when one card makes the request fail', async () => { + const server = new FakeContacts(); + const good = server.addCard({ note: 'a' }); + const bad = server.addCard({ note: 'b' }); + const { engine } = await synced(server); + await mine(engine, { note: 'good edit' }, good.id); + await mine(engine, { note: 'poison' }, bad.id); + server.failSet = (args) => (JSON.stringify(args).includes('poison') ? { http: 400 } : null); + await engine.requestSync(); + expect(server.account().cards.get(good.id)?.note).toBe('good edit'); + expect(server.account().cards.get(bad.id)?.note).toBe('b'); + const conflicts = await conflictsOf(engine); + expect(conflicts.map((c) => [c.recordId, c.kind])).toEqual([[bad.id, 'refused']]); + expect(await pendingOf(engine)).toBe(0); + }); + + it('takes the request limit from the session', async () => { + const server = new FakeContacts(); + const card = server.addCard(); + const session = { ...CONTACTS_SESSION, capabilities: { ...CONTACTS_SESSION.capabilities, 'urn:ietf:params:jmap:core': { maxSizeRequest: 1000 } } }; + const db = openMemoryStore(); + const engine = engineFor(server, db, undefined, session); + await engine.requestSync(); + const result = await mine(engine, { note: 'x'.repeat(2000) }, card.id); + expect(result.notUpdated[card.id].type).toBe('tooLarge'); + }); +}); + +describe('a uid another card has', () => { + it('does not turn an offline create into that card', async () => { + const server = new FakeContacts(); + const other = server.addCard({ uid: 'urn:uuid:taken', name: { full: 'Their card' } }); + const { db, engine } = await synced(server); + const { created } = await call(engine, 'ContactCard/set', { create: { n: { uid: 'urn:uuid:taken', name: { full: 'My card' }, addressBookIds: { book1: true } } } }); + await engine.requestSync(); + const conflicts = await conflictsOf(engine); + expect(conflicts).toHaveLength(1); + expect(conflicts[0]).toMatchObject({ kind: 'refused', recordId: created.n.id, summary: 'My card', local: { op: 'create', payload: { name: { full: 'My card' } } } }); + expect((await call(engine, 'ContactCard/get', { ids: [created.n.id] })).notFound).toEqual([created.n.id]); + expect(db.all('SELECT id FROM contact_card').map((r) => r.id)).toEqual([other.id]); + expect(server.account().cards.get(other.id)).toMatchObject({ name: { full: 'Their card' } }); + }); + + it('does not take someone else\'s card for its own after a failed attempt either', async () => { + const server = new FakeContacts(); + const { engine } = await synced(server); + const { created } = await call(engine, 'ContactCard/set', { create: { n: { uid: 'urn:uuid:race', name: { full: 'Mine' }, addressBookIds: { book1: true } } } }); + server.offlineBeforeSet = true; + await engine.requestSync(); + const other = server.addCard({ uid: 'urn:uuid:race', name: { full: 'Theirs' } }); + await engine.requestSync(); + expect((await conflictsOf(engine)).map((c) => [c.recordId, c.kind])).toEqual([[created.n.id, 'refused']]); + expect(server.account().cards.size).toBe(1); + expect(server.account().cards.get(other.id)).toMatchObject({ name: { full: 'Theirs' } }); + }); + + it('does not delete someone else\'s card when the create was deleted again', async () => { + const server = new FakeContacts(); + const { engine } = await synced(server); + const { created } = await call(engine, 'ContactCard/set', { create: { n: { uid: 'urn:uuid:race', name: { full: 'Mine' }, addressBookIds: { book1: true } } } }); + server.offlineBeforeSet = true; + await engine.requestSync(); + await call(engine, 'ContactCard/set', { destroy: [created.n.id] }); + const other = server.addCard({ uid: 'urn:uuid:race', name: { full: 'Theirs' } }); + await engine.requestSync(); + expect(server.account().cards.has(other.id)).toBe(true); + expect(await pendingOf(engine)).toBe(0); + }); +}); + +describe('what the server really stored', () => { + const photo = { p: { kind: 'photo', uri: 'data:image/png;base64,iVBORw0KGgo=', mediaType: 'image/png' } }; + + it('does not read its own stored photo as a change after the network went, on 1.0', async () => { + const server = new FakeContacts(); + server.mediaAsBlobs = true; + const { engine } = await synced(server); + const { created } = await call(engine, 'ContactCard/set', { create: { n: { name: { full: 'Foto' }, addressBookIds: { book1: true }, media: photo } } }); + server.offlineAfterSet = true; + await engine.requestSync(); + expect(server.account().cards.size).toBe(1); + server.offline = false; + await call(engine, 'ContactCard/set', { destroy: [created.n.id] }); + await engine.requestSync(); + expect(await conflictsOf(engine)).toEqual([]); + expect(server.account().cards.size).toBe(0); + }); + + it('adopts its own card by uid though the server keeps the photo as a blob', async () => { + const server = new FakeContacts(); + server.mediaAsBlobs = true; + const { engine } = await synced(server); + await call(engine, 'ContactCard/set', { create: { n: { name: { full: 'Foto' }, addressBookIds: { book1: true }, media: photo } } }); + server.loseNextSet = true; + await engine.requestSync(); + await engine.requestSync(); + expect(await conflictsOf(engine)).toEqual([]); + expect(await pendingOf(engine)).toBe(0); + expect(server.account().cards.size).toBe(1); + }); + + it('keeps the card as the server read it back, not as it was sent, before any delta', async () => { + const server = new FakeContacts(); + server.mediaAsBlobs = true; + const { db, engine } = await synced(server); + await call(engine, 'ContactCard/set', { create: { n: { name: { full: 'Foto' }, addressBookIds: { book1: true }, media: photo } } }); + // The network goes right after the read-back, before the delta of this pass. + server.beforeRequest = () => { + const set = server.calls.findIndex(([name]) => name === 'ContactCard/set'); + if (set >= 0 && server.calls.slice(set).some(([name]) => name === 'ContactCard/get')) server.offline = true; + }; + await engine.requestSync(); + server.beforeRequest = null; + server.offline = false; + const row = db.get<{ json: string }>('SELECT json FROM contact_card'); + expect((JSON.parse(row?.json as string) as { media: Record }).media.p).toHaveProperty('blobId'); + const id = [...server.account().cards.keys()][0]; + await mine(engine, { note: 'after' }, id); + await engine.requestSync(); + expect(await conflictsOf(engine)).toEqual([]); + expect(server.account().cards.get(id)?.note).toBe('after'); + }); + + it('does not take a lost photo update for someone else changing it, on 1.0', async () => { + const server = new FakeContacts(); + server.mediaAsBlobs = true; + const card = server.addCard({ note: 'old' }); + const { engine } = await synced(server); + await mine(engine, { media: photo }, card.id); + server.loseNextSet = true; + await engine.requestSync(); + await engine.requestSync(); + expect(await conflictsOf(engine)).toEqual([]); + expect(await pendingOf(engine)).toBe(0); + }); +}); + +describe('after a lost update', () => { + it('still deletes the card', async () => { + const server = new FakeContacts(); + const card = server.addCard({ note: 'old' }); + const { engine } = await synced(server); + await mine(engine, { note: 'mine' }, card.id); + server.loseNextSet = true; + await engine.requestSync(); + await call(engine, 'ContactCard/set', { destroy: [card.id] }); + await engine.requestSync(); + expect(await conflictsOf(engine)).toEqual([]); + expect(server.account().cards.has(card.id)).toBe(false); + }); + + it('does not delete when someone else changed another property too', async () => { + const server = new FakeContacts(); + const card = server.addCard({ note: 'old', name: { full: 'A' } }); + const { engine } = await synced(server); + await mine(engine, { note: 'mine' }, card.id); + server.loseNextSet = true; + await engine.requestSync(); + await call(engine, 'ContactCard/set', { destroy: [card.id] }); + server.editCard(card.id, { name: { full: 'B' } }); + await engine.requestSync(); + expect(server.account().cards.has(card.id)).toBe(true); + expect((await conflictsOf(engine)).map((c) => c.kind)).toEqual(['changed']); + }); + + it('sends a newer edit of the same property', async () => { + const server = new FakeContacts(); + const card = server.addCard({ note: 'old' }); + const { engine } = await synced(server); + await mine(engine, { note: 'first' }, card.id); + server.loseNextSet = true; + await engine.requestSync(); + await mine(engine, { note: 'second' }, card.id); + await engine.requestSync(); + expect(await conflictsOf(engine)).toEqual([]); + expect(server.account().cards.get(card.id)?.note).toBe('second'); + }); +}); + +describe('a refusal while the page kept editing', () => { + it('keeps the newer edit, and only records a conflict if it is refused too', async () => { + const server = new FakeContacts(); + const card = server.addCard({ note: 'old', name: { full: 'A' } }); + const { engine } = await synced(server); + await mine(engine, { note: 'one' }, card.id); + server.refuse.set(card.id, { type: 'forbidden' }); + const original = server.request.bind(server); + let edited = false; + server.request = async (calls: MethodCall[]) => { + if (!edited && calls.some(([name]) => name === 'ContactCard/set')) { + edited = true; + await mine(engine, { 'name/full': 'B' }, card.id); + } + return original(calls); + }; + await engine.requestSync(); + expect(await pendingOf(engine)).toBe(1); + expect(await conflictsOf(engine)).toEqual([]); + server.refuse.clear(); + await engine.requestSync(); + expect(server.account().cards.get(card.id)).toMatchObject({ note: 'one', name: { full: 'B' } }); + expect(await pendingOf(engine)).toBe(0); + }); +}); + +describe('session limits', () => { + it('takes maxObjectsInGet and maxCallsInRequest from the session', async () => { + const server = new FakeContacts(); + server.maxObjectsInGet = 2; + const cards = [server.addCard(), server.addCard(), server.addCard(), server.addCard(), server.addCard()]; + const session = { ...CONTACTS_SESSION, capabilities: { ...CONTACTS_SESSION.capabilities, 'urn:ietf:params:jmap:core': { maxObjectsInGet: 2, maxCallsInRequest: 4 } } }; + const db = openMemoryStore(); + const events: StoreEvent[] = []; + const engine = engineFor(server, db, (e) => events.push(e), session); + await engine.requestSync(); + expect(events.filter((e) => e.type === 'error')).toEqual([]); + expect(count(db, 'contact_card')).toBe(5); + for (const [name, args] of server.calls) { + if (name === 'ContactCard/query') expect(args.limit).toBeLessThanOrEqual(2); + } + for (const card of cards.slice(0, 3)) await mine(engine, { note: 'x' }, card.id); + await engine.requestSync(); + expect(events.filter((e) => e.type === 'error')).toEqual([]); + expect(await pendingOf(engine)).toBe(0); + expect(cards.slice(0, 3).map((c) => server.account().cards.get(c.id)?.note)).toEqual(['x', 'x', 'x']); + }); +}); + +describe('an interrupted first sync', () => { + it('answers serverUnavailable until the first full pass is done', async () => { + const server = new FakeContacts(); + for (let i = 0; i < 2600; i += 1) server.addCard(); + const db = openMemoryStore(); + const engine = engineFor(server, db); + let requests = 0; + server.beforeRequest = () => { + requests += 1; + if (requests === 4) server.offline = true; + }; + await engine.requestSync(); + expect(count(db, 'contact_card')).toBe(2000); + expect(await engine.answer([['ContactCard/query', { accountId: JMAP_ACCOUNT, limit: 50 }, '0']])).toBeNull(); + expect(await engine.answer([['ContactCard/get', { accountId: JMAP_ACCOUNT, ids: null }, '0']])).toBeNull(); + server.offline = false; + server.beforeRequest = null; + await engine.requestSync(); + expect((await call(engine, 'ContactCard/query', { calculateTotal: true })).total).toBe(2600); + }); + + it('goes on answering from the old copy while a later resync runs', async () => { + const server = new FakeContacts(); + for (let i = 0; i < 2600; i += 1) server.addCard(); + const { db, engine } = await synced(server); + db.run("UPDATE sync_state SET state = 'bad state!' WHERE type = 'ContactCard'"); + let requests = 0; + server.beforeRequest = () => { + requests += 1; + if (requests === 3) server.offline = true; + }; + await engine.requestSync(); + expect((await call(engine, 'ContactCard/query', { calculateTotal: true })).total).toBe(2600); + }); +}); + +describe('paging while cards change', () => { + it('does not skip a card when an earlier one is deleted between requests', async () => { + const server = new FakeContacts(); + const cards = Array.from({ length: 2600 }, () => server.addCard()); + const db = openMemoryStore(); + const engine = engineFor(server, db); + let queries = 0; + const original = server.request.bind(server); + server.request = async (calls: MethodCall[]) => { + if (calls.some(([name]) => name === 'ContactCard/query')) { + queries += 1; + if (queries === 2) server.removeCard(cards[0].id); + } + return original(calls); + }; + await engine.requestSync(); + const kept = new Set(db.all<{ id: string }>('SELECT id FROM contact_card').map((r) => r.id)); + expect(cards.slice(1).filter((c) => !kept.has(c.id))).toEqual([]); + expect(kept.has(cards[0].id)).toBe(false); + const anchored = server.calls.filter(([name, args]) => name === 'ContactCard/query' && args.anchor !== undefined); + expect(anchored.length).toBeGreaterThan(0); + }); + + it('starts over when the card it paged from is gone', async () => { + const server = new FakeContacts(); + const cards = Array.from({ length: 2600 }, () => server.addCard()); + const db = openMemoryStore(); + const engine = engineFor(server, db); + let queries = 0; + const original = server.request.bind(server); + server.request = async (calls: MethodCall[]) => { + if (calls.some(([name]) => name === 'ContactCard/query')) { + queries += 1; + if (queries === 2) server.removeCard(cards[1999].id); + } + return original(calls); + }; + const events: StoreEvent[] = []; + await engine.requestSync(); + expect(events).toEqual([]); + const kept = new Set(db.all<{ id: string }>('SELECT id FROM contact_card').map((r) => r.id)); + expect(kept.size).toBe(2599); + expect(kept.has(cards[1999].id)).toBe(false); + }); +}); /* eslint-enable @typescript-eslint/no-explicit-any */ diff --git a/lib/store/__tests__/helpers/fake-contacts.ts b/lib/store/__tests__/helpers/fake-contacts.ts index e6b408755..6cec59cbf 100644 --- a/lib/store/__tests__/helpers/fake-contacts.ts +++ b/lib/store/__tests__/helpers/fake-contacts.ts @@ -37,6 +37,15 @@ export function changesFor(coll: Collection, args: Record): Rec return { accountId: args.accountId, oldState: since, ...coll.changes(since, maxChanges) }; } +function storedMedia(card: Obj): void { + const media = card.media as Record> | undefined; + for (const [key, entry] of Object.entries(media ?? {})) { + if (typeof entry.uri !== 'string' || !entry.uri.startsWith('data:')) continue; + const { uri, ...rest } = entry; + (media as Record)[key] = { ...rest, blobId: `stored-${uri.length}-${key}` }; + } +} + function applyServerPatch(card: Obj, patch: Record): void { for (const [path, value] of Object.entries(patch)) { const steps = path.split('/'); @@ -63,6 +72,16 @@ export class FakeContacts extends FakeJmap { readonly refuse = new Map(); /** A server that lets two cards have one uid, as nothing in the RFC forbids it. */ allowDuplicateUids = false; + /** Stalwart 1.0 stores a data: URI in `media` as a blob: `uri` goes, `blobId` comes (measured). */ + mediaAsBlobs = false; + /** Set to apply the next ContactCard/set, answer it, and then lose the network. */ + offlineAfterSet = false; + /** Set to lose the network before the next ContactCard/set reaches the server. */ + offlineBeforeSet = false; + /** The most ids ContactCard/get and calls a request take, as a session can say. */ + maxObjectsInGet = 500; + /** Answers a ContactCard/set call as a whole, instead of handling it. */ + failSet: ((args: Record) => { jmapError: string } | { http: number } | null) | null = null; constructor(books: string[] = ['book1']) { super(); @@ -131,7 +150,9 @@ export class FakeContacts extends FakeJmap { notCreated[cid] = { type: 'invalidProperties', description: `Contact with UID ${String(raw.uid)} already exists with id ${clash.id}.`, properties: ['uid'] }; } else { const id = `c${this.nextCard++}`; - account.cards.set(id, { '@type': 'Card', version: '1.0', ...raw, id }); + const stored: Obj = { '@type': 'Card', version: '1.0', ...structuredClone(raw), id }; + if (this.mediaAsBlobs) storedMedia(stored); + account.cards.set(id, stored); account.coll.ContactCard.push('created', id); created[cid] = { id }; } @@ -142,7 +163,8 @@ export class FakeContacts extends FakeJmap { if (refused) notUpdated[id] = refused; else if (!card) notUpdated[id] = { type: 'notFound' }; else { - applyServerPatch(card, patch); + applyServerPatch(card, structuredClone(patch)); + if (this.mediaAsBlobs) storedMedia(card); account.coll.ContactCard.push('updated', id); updated[id] = null; } @@ -187,7 +209,12 @@ export class FakeContacts extends FakeJmap { let list = [...account.cards.values()]; if (filter.uid) list = list.filter((c) => c.uid === filter.uid); if (filter.inAddressBook) list = list.filter((c) => (c.addressBookIds as Record)[filter.inAddressBook as string]); - const position = Number(args.position ?? 0); + let position = Number(args.position ?? 0); + if (args.anchor !== undefined) { + const at = list.findIndex((c) => c.id === args.anchor); + if (at < 0) throw Object.assign(new Error('anchorNotFound'), { jmapError: 'anchorNotFound' }); + position = at + Number(args.anchorOffset ?? 0); + } const limit = Number(args.limit ?? 5000) || 5000; return { accountId: args.accountId, @@ -200,8 +227,8 @@ export class FakeContacts extends FakeJmap { } case 'ContactCard/get': { const ids = args.ids as string[] | null | undefined; - if (ids && ids.length > 500) throw Object.assign(new Error('requestTooLarge'), { jmapError: 'requestTooLarge' }); - const chosen = ids ?? [...account.cards.keys()].slice(0, 500); + if (ids && ids.length > this.maxObjectsInGet) throw Object.assign(new Error('requestTooLarge'), { jmapError: 'requestTooLarge' }); + const chosen = ids ?? [...account.cards.keys()].slice(0, this.maxObjectsInGet); return { accountId: args.accountId, state: account.coll.ContactCard.state(), @@ -209,8 +236,12 @@ export class FakeContacts extends FakeJmap { notFound: chosen.filter((id) => !account.cards.has(id)), }; } - case 'ContactCard/set': + case 'ContactCard/set': { + const failure = this.failSet?.(args); + if (failure && 'http' in failure) throw new TransportError('http', `JMAP request failed: ${failure.http}`, failure.http, 'urn:ietf:params:jmap:error:limit'); + if (failure) throw Object.assign(new Error(failure.jmapError), failure); return this.contactSet(account, args); + } default: return super.handle(name, args); } @@ -219,7 +250,15 @@ export class FakeContacts extends FakeJmap { override async request(calls: MethodCall[], using?: string[]): Promise { // The real server fails the whole request for a capability it doesn't list. void using; + if (this.offlineBeforeSet && calls.some(([name]) => name === 'ContactCard/set')) { + this.offlineBeforeSet = false; + throw new TransportError('network', 'Failed to fetch'); + } const response = await super.request(calls); + if (this.offlineAfterSet && calls.some(([name]) => name === 'ContactCard/set')) { + this.offlineAfterSet = false; + this.offline = true; + } if (this.loseNextSet && calls.some(([name]) => name === 'ContactCard/set')) { this.loseNextSet = false; throw new TransportError('network', 'Connection dropped after the server took the request'); diff --git a/lib/store/contacts/methods.ts b/lib/store/contacts/methods.ts index f3c885909..9ef5a391f 100644 --- a/lib/store/contacts/methods.ts +++ b/lib/store/contacts/methods.ts @@ -15,10 +15,14 @@ function jmapOf(ctx: MethodContext, args: Record): string { return (args.accountId as string | undefined) ?? ctx.session.primaryAccounts[CONTACTS_CAPABILITY] ?? ctx.defaultJmap; } -/** The last state from the server. Throws when this type was never synced, 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 list as the whole. A later resync + * (phase `list`) goes on answering from the copy it's refreshing. + */ function syncedState(ctx: MethodContext, jmap: string, type: string): string | null { const row = ctx.db.get<{ state: string | null; phase: string | null }>('SELECT state, phase FROM sync_state WHERE jmap = ? AND type = ?', [jmap, type]); - if (!row || (!row.state && !row.phase)) throw new MethodError('serverUnavailable', 'Contacts not kept yet'); + if (!row || (!row.state && row.phase !== 'list')) throw new MethodError('serverUnavailable', 'Contacts not kept yet'); return row.state; } diff --git a/lib/store/contacts/patch.ts b/lib/store/contacts/patch.ts index 07247b3d9..61a5c1050 100644 --- a/lib/store/contacts/patch.ts +++ b/lib/store/contacts/patch.ts @@ -67,9 +67,13 @@ export function touchedProperties(patch: Patch): string[] { return [...new Set(Object.keys(patch).map((path) => steps(path)[0]))]; } -/** Deep equality that ignores key order. Absent and null count as the same. */ +function isNothing(value: unknown): boolean { + return value == null || (isObject(value) && Object.keys(value).length === 0); +} + +/** Deep equality that ignores key order. Absent, null and an empty map count as the same. */ export function sameJson(a: unknown, b: unknown): boolean { - if (a === b || (a == null && b == null)) return true; + if (a === b || (isNothing(a) && isNothing(b))) return true; if (Array.isArray(a) && Array.isArray(b)) { return a.length === b.length && a.every((item, i) => sameJson(item, b[i])); } @@ -79,10 +83,27 @@ export function sameJson(a: unknown, b: unknown): boolean { return false; } -/** Whether `actual` holds everything in `wanted`. A server may add properties of its own. */ -export function coversJson(wanted: unknown, actual: unknown): boolean { - if (isObject(wanted) && isObject(actual)) { - return Object.keys(wanted).every((key) => coversJson(wanted[key], actual[key])); +/** Stalwart 1.0 keeps a data: URI in `media` as a blob: `uri` goes and `blobId` comes (measured). */ +function inlineKeptAsBlob(wanted: JsonObject, actual: JsonObject): boolean { + return typeof wanted.uri === 'string' && wanted.uri.startsWith('data:') && actual.uri == null && typeof actual.blobId === 'string'; +} + +/** + * Whether `actual` is `wanted` as far as a server can tell. A key only the + * server has is its own, unless `base` shows it was there before the edit: + * then the edit meant to remove it, and it hasn't happened yet. + */ +export function coversJson(wanted: unknown, actual: unknown, base?: unknown): boolean { + if (!isObject(wanted) || !isObject(actual)) return sameJson(wanted, actual); + const before = isObject(base) ? base : {}; + const inline = inlineKeptAsBlob(wanted, actual); + for (const key of Object.keys(wanted)) { + if (inline && key === 'uri') continue; + if (!coversJson(wanted[key], actual[key], before[key])) return false; } - return sameJson(wanted, actual); + for (const key of Object.keys(actual)) { + if (key in wanted || (inline && key === 'blobId')) continue; + if (before[key] != null) return false; + } + return true; } diff --git a/lib/store/contacts/pending.ts b/lib/store/contacts/pending.ts index 4cf2feba2..f6d2bdc78 100644 --- a/lib/store/contacts/pending.ts +++ b/lib/store/contacts/pending.ts @@ -12,11 +12,21 @@ * - Destroy: done if the card is gone, a conflict if it changed since. * - Create: a request that may have reached the server looks for the card's * uid first, so a lost response doesn't make a second card. Stalwart also - * refuses a second card with the same uid, which ends up the same way. + * refuses a second card with the same uid. Either way the card found is only + * ours if it holds what was sent, or an offline card with someone else's uid + * would silently turn into theirs. + * + * The local copy after sending is always the server's own read-back, never what + * was sent: Stalwart 1.0 stores a data: URI as a blob. A base built from what + * was sent would read that as someone else's change. If the network goes before + * the read-back, that counts as a lost response: the row stays, and the next + * pass finds its edit already there. `sent` is what the last request held, so + * the next pass can tell its own edit from someone else's. */ import type { StoreContext, SyncContext } from '../area'; -import { responseFor, type MethodCall } from '../engine/transport'; +import { requestLimits, type RequestLimits } from '../engine/limits'; +import { JmapMethodError, responseFor, TransportError, type MethodCall } from '../engine/transport'; import type { StorageAdapter } from '../storage'; import { applyPatch, coversJson, InvalidPatchError, mergePatches, sameJson, touchedProperties, type JsonObject, type Patch } from './patch'; import { cardDisplayName, CONTACTS_USING, readCard, removeCard, writeCard, type Card } from './records'; @@ -26,11 +36,12 @@ export const STAND_IN_PREFIX = 'bulwark-local-card-'; const LOCAL_BLOB_MARK = '"blobId":"bulwark-local-'; const SET_BATCH = 100; -const SET_BATCH_BYTES = 5 * 1024 * 1024; -/** The server takes 10 MB a request. A bigger card would hold up everything queued behind it, for good. */ -const MAX_RECORD_BYTES = 8 * 1024 * 1024; -const GET_BATCH = 500; -const CALLS_PER_REQUEST = 16; + +/** Of the session's request limit. A bigger card would never go out. */ +function maxRecordBytes(ctx: StoreContext): number { + return Math.floor(requestLimits(ctx.session).maxSizeRequest * 0.8); +} + /** Set errors that say nothing about the edit itself: it goes out again next time. */ const TRANSIENT_ERRORS = new Set(['serverFail', 'serverUnavailable', 'rateLimit']); /** Server-kept timestamps: their changing doesn't mean someone edited the card. */ @@ -45,6 +56,7 @@ export interface PendingRow { payload: string | null; base: string | null; attempted: number; + sent: string | null; queued_at: number; } @@ -105,7 +117,7 @@ export function queueCreate(ctx: StoreContext, jmap: string, card: JsonObject): const id = `${STAND_IN_PREFIX}${crypto.randomUUID()}`; const uid = typeof card.uid === 'string' && card.uid ? card.uid : `urn:uuid:${crypto.randomUUID()}`; const payload = JSON.stringify({ ...card, uid }); - if (payload.length > MAX_RECORD_BYTES) return TOO_LARGE_ERROR; + if (payload.length > maxRecordBytes(ctx)) return TOO_LARGE_ERROR; savePending(ctx.db, ctx, { jmap, id, op: 'create', uid, payload, base: null }); writeCard(ctx.db, jmap, { '@type': 'Card', version: '1.0', ...card, uid, id }); return { id, uid }; @@ -127,11 +139,11 @@ export function queueUpdate(ctx: StoreContext, jmap: string, rawId: string, patc } if (pending?.op === 'create') { const payload = JSON.stringify(applyPatch(JSON.parse(pending.payload as string) as JsonObject, patch)); - if (payload.length > MAX_RECORD_BYTES) return TOO_LARGE_ERROR; + if (payload.length > maxRecordBytes(ctx)) return TOO_LARGE_ERROR; savePending(ctx.db, ctx, { jmap, id, op: 'create', uid: pending.uid, payload, base: null }); } else { const merged = JSON.stringify(pending ? mergePatches(JSON.parse(pending.payload as string) as Patch, patch) : patch); - if (merged.length > MAX_RECORD_BYTES) return TOO_LARGE_ERROR; + if (merged.length > maxRecordBytes(ctx)) return TOO_LARGE_ERROR; savePending(ctx.db, ctx, { jmap, id, op: 'update', payload: merged, base: pending ? pending.base : JSON.stringify(card) }); } writeCard(ctx.db, jmap, visible); @@ -228,30 +240,44 @@ function withoutStamps(card: JsonObject): JsonObject { return copy; } -async function readServerCards(ctx: SyncContext, jmap: string, ids: string[]): Promise> { +async function readServerCards(ctx: SyncContext, jmap: string, ids: string[], limits: RequestLimits): Promise> { const cards = new Map(); - for (let i = 0; i < ids.length; i += GET_BATCH) { - const response = await ctx.transport.request([['ContactCard/get', { accountId: jmap, ids: ids.slice(i, i + GET_BATCH) }, 'g']], CONTACTS_USING); + for (let i = 0; i < ids.length; i += limits.maxObjectsInGet) { + const response = await ctx.transport.request([['ContactCard/get', { accountId: jmap, ids: ids.slice(i, i + limits.maxObjectsInGet) }, 'g']], CONTACTS_USING); for (const card of responseFor(response, 'g')[1].list as Card[]) cards.set(card.id, card); } return cards; } -/** The ids the server has for these uids. */ -async function lookUpUids(ctx: SyncContext, jmap: string, uids: string[]): Promise> { - const found = new Map(); - for (let i = 0; i < uids.length; i += CALLS_PER_REQUEST) { - const group = uids.slice(i, i + CALLS_PER_REQUEST); +/** The server's cards with these uids. */ +async function cardsByUid(ctx: SyncContext, jmap: string, uids: string[], limits: RequestLimits): Promise> { + const ids = new Map(); + for (let i = 0; i < uids.length; i += limits.maxCallsInRequest) { + const group = uids.slice(i, i + limits.maxCallsInRequest); const calls: MethodCall[] = group.map((uid, n) => ['ContactCard/query', { accountId: jmap, filter: { uid }, limit: 1 }, `q${n}`]); const response = await ctx.transport.request(calls, CONTACTS_USING); group.forEach((uid, n) => { - const ids = responseFor(response, `q${n}`)[1].ids as string[]; - if (ids.length > 0) found.set(uid, ids[0]); + const found = responseFor(response, `q${n}`)[1].ids as string[]; + if (found.length > 0) ids.set(uid, found[0]); }); } - return found; + const cards = await readServerCards(ctx, jmap, [...ids.values()], limits); + const byUid = new Map(); + for (const [uid, id] of ids) { + const card = cards.get(id); + if (card) byUid.set(uid, card); + } + return byUid; } +/** Whether a card the server has under our uid is the one we sent. */ +function holdsSent(sentPayload: string, card: Card): boolean { + const sent = JSON.parse(sentPayload) as JsonObject; + return Object.keys(sent).every((key) => coversJson(sent[key], card[key])); +} + +const UID_TAKEN: SetError = { type: 'invalidProperties', description: 'Another contact already has this uid', properties: ['uid'] }; + /** The create went through, or had already. The stand-in row becomes the server's card. */ function settleCreated(ctx: SyncContext, sent: PendingRow, server: Card): void { const { db } = ctx; @@ -282,29 +308,59 @@ function settleCreated(ctx: SyncContext, sent: PendingRow, server: Card): void { interface Decision { create: Map; update: Map; destroy: Map } +/** + * Whether someone else moved one of these properties since the edit was queued. + * The server holding what our last request sent is our own doing, not a change: + * its response may have been lost. + */ +function movedOnServer(row: PendingRow, server: Card, keys: readonly string[]): boolean { + if (!row.base) return false; + const base = JSON.parse(row.base) as JsonObject; + let own: JsonObject | null = null; + if (row.sent) { + try { + own = applyPatch(base, JSON.parse(row.sent) as Patch); + } catch (error) { + if (!(error instanceof InvalidPatchError)) throw error; + } + } + return keys.some((key) => !sameJson(base[key], server[key]) && !(own && coversJson(own[key], server[key], base[key]))); +} + /** What to do with one row now that the server's copy is known. Conflicts are settled here. */ function decide(ctx: SyncContext, row: PendingRow, server: Card | undefined, serverByUid: Map, out: Decision): void { const { db } = ctx; const { jmap, id } = row; const uidHit = row.uid ? serverByUid.get(row.uid) : undefined; if (row.op === 'create') { - if (uidHit) { - settleCreated(ctx, row, { ...(JSON.parse(row.payload as string) as JsonObject), ...uidHit }); - return; + if (!uidHit) { + out.create.set(id, row); + } else if (holdsSent(row.sent ?? (row.payload as string), uidHit)) { + settleCreated(ctx, { ...row, payload: row.sent ?? row.payload }, uidHit); + } else { + recordConflict(ctx, row, 'refused', null, UID_TAKEN); + dropPending(db, jmap, id); + removeCard(db, jmap, id); } - out.create.set(id, row); return; } if (row.op === 'destroy' && id.startsWith(STAND_IN_PREFIX)) { - // A create that may have gone out, deleted since. Without a card on the server, nothing to do. + // A create that may have gone out, deleted since. Only a card holding what was sent is ours to delete. dropPending(db, jmap, id); - if (uidHit) savePending(db, ctx, { jmap, id: uidHit.id, op: 'destroy', payload: null, base: JSON.stringify(uidHit) }); + const sent = row.sent ?? row.payload; + if (uidHit && sent && holdsSent(sent, uidHit)) { + savePending(db, ctx, { jmap, id: uidHit.id, op: 'destroy', payload: null, base: JSON.stringify(uidHit) }); + } return; } if (row.op === 'destroy') { if (!server) { dropPending(db, jmap, id); - } else if (row.base && !sameJson(withoutStamps(JSON.parse(row.base) as JsonObject), withoutStamps(server))) { + return; + } + const stable = withoutStamps(server); + const base = row.base ? withoutStamps(JSON.parse(row.base) as JsonObject) : {}; + if (movedOnServer(row, server, [...new Set([...Object.keys(stable), ...Object.keys(base)])])) { recordConflict(ctx, row, 'changed', server, null); dropPending(db, jmap, id); writeCard(db, jmap, server); @@ -320,7 +376,7 @@ function decide(ctx: SyncContext, row: PendingRow, server: Card | undefined, ser return; } const patch = JSON.parse(row.payload as string) as Patch; - const base = row.base ? JSON.parse(row.base) as JsonObject : null; + const base = row.base ? JSON.parse(row.base) as JsonObject : {}; let applied: Card | null = null; try { applied = applyPatch(server, patch); @@ -328,12 +384,12 @@ function decide(ctx: SyncContext, row: PendingRow, server: Card | undefined, ser if (!(error instanceof InvalidPatchError)) throw error; } const touched = touchedProperties(patch); - if (applied && touched.every((key) => coversJson((applied as Card)[key], server[key]))) { + if (applied && touched.every((key) => coversJson((applied as Card)[key], server[key], base[key]))) { dropPending(db, jmap, id); writeCard(db, jmap, server); return; } - if (!applied || (base && touched.some((key) => !sameJson(base[key], server[key])))) { + if (!applied || movedOnServer(row, server, touched)) { recordConflict(ctx, row, 'changed', server, null); dropPending(db, jmap, id); writeCard(db, jmap, server); @@ -342,125 +398,176 @@ function decide(ctx: SyncContext, row: PendingRow, server: Card | undefined, ser out.update.set(id, row); } -function settleUpdated(ctx: SyncContext, sent: PendingRow, server: Card): void { - const now = pendingFor(ctx.db, sent.jmap, sent.id); +function settleUpdated(ctx: SyncContext, sent: PendingRow, server: Card | undefined): void { + const { db } = ctx; + const now = pendingFor(db, sent.jmap, sent.id); if (!now) return; if (now.op === 'update' && now.payload === sent.payload) { - dropPending(ctx.db, sent.jmap, sent.id); + dropPending(db, sent.jmap, sent.id); + if (server) writeCard(db, sent.jmap, server); return; } - // Edited again while this was on its way: that edit now builds on what was sent. - const base = applyPatch(server, JSON.parse(sent.payload as string) as Patch); - ctx.db.run('UPDATE contact_pending SET base = ? WHERE seq = ?', [JSON.stringify(base), now.seq]); + // Edited again while this was on its way: that edit now builds on what the server holds. + if (!server) return; + db.run('UPDATE contact_pending SET base = ?, sent = NULL, attempted = 0 WHERE seq = ?', [JSON.stringify(server), now.seq]); + if (now.op === 'update') storeServerCard(db, sent.jmap, server); } -function settleRefused(ctx: SyncContext, row: PendingRow, error: SetError, server: Card | undefined): void { +function settleRefused(ctx: SyncContext, sent: PendingRow, error: SetError, server: Card | undefined): void { const { db } = ctx; if (TRANSIENT_ERRORS.has(error.type)) return; - if (row.op === 'destroy' && error.type === 'notFound') { - dropPending(db, row.jmap, row.id); + const now = pendingFor(db, sent.jmap, sent.id); + if (now && (now.op !== sent.op || now.payload !== sent.payload)) { + // Edited again while this was on its way, so the refused edit is still in + // the row. It goes out once more, and a second refusal is the conflict. + db.run('UPDATE contact_pending SET sent = NULL, attempted = 0 WHERE seq = ?', [now.seq]); + return; + } + if (sent.op === 'destroy' && error.type === 'notFound') { + dropPending(db, sent.jmap, sent.id); return; } const kind = error.type === 'notFound' ? 'gone' : 'refused'; - recordConflict(ctx, row, kind, server ?? null, error); - dropPending(db, row.jmap, row.id); - if (server && kind === 'refused') writeCard(db, row.jmap, server); - else removeCard(db, row.jmap, row.id); + recordConflict(ctx, sent, kind, server ?? null, error); + dropPending(db, sent.jmap, sent.id); + const restore = server ?? (sent.base ? JSON.parse(sent.base) as Card : null); + if (restore && kind === 'refused') writeCard(db, sent.jmap, restore); + else removeCard(db, sent.jmap, sent.id); } /** Takes rows up to about the size of one request. */ -function nextBatch(db: StorageAdapter, jmap: string, afterSeq: number): PendingRow[] { - const rows = db.all('SELECT * FROM contact_pending WHERE jmap = ? AND seq > ? ORDER BY seq LIMIT ?', [jmap, afterSeq, SET_BATCH]); +function nextBatch(db: StorageAdapter, jmap: string, afterSeq: number, limits: RequestLimits): PendingRow[] { + const rows = db.all('SELECT * FROM contact_pending WHERE jmap = ? AND seq > ? ORDER BY seq LIMIT ?', [jmap, afterSeq, Math.min(SET_BATCH, limits.maxObjectsInSet)]); + const room = limits.maxSizeRequest / 2; let bytes = 0; const batch: PendingRow[] = []; for (const row of rows) { bytes += row.payload?.length ?? 0; - if (batch.length > 0 && bytes > SET_BATCH_BYTES) break; + if (batch.length > 0 && bytes > room) break; batch.push(row); } return batch; } +/** One exchange with the server for these rows: read, decide, send, read back, settle. */ +async function exchange(ctx: SyncContext, jmap: string, rows: readonly PendingRow[], limits: RequestLimits): Promise { + const { db } = ctx; + const current = await readServerCards(ctx, jmap, rows.filter((row) => row.op !== 'create' && !row.id.startsWith(STAND_IN_PREFIX)).map((row) => row.id), limits); + const serverByUid = await cardsByUid(ctx, jmap, rows.filter((row) => row.uid && row.attempted).map((row) => row.uid as string), limits); + + const out: Decision = { create: new Map(), update: new Map(), destroy: new Map() }; + db.transaction(() => { + for (const { seq, id } of rows) { + // The page may have edited it while the reads were on their way. + const row = pendingFor(db, jmap, id); + if (row && row.seq === seq) decide(ctx, row, current.get(id), serverByUid, out); + } + for (const row of [...out.create.values(), ...out.update.values()]) { + db.run('UPDATE contact_pending SET attempted = 1, sent = payload WHERE seq = ?', [row.seq]); + } + }); + if (out.create.size + out.update.size + out.destroy.size === 0) return; + + const creationIds = new Map([...out.create.values()].map((row) => [`c${row.seq}`, row])); + const response = await ctx.transport.request([['ContactCard/set', { + accountId: jmap, + ...(creationIds.size > 0 ? { create: Object.fromEntries([...creationIds].map(([cid, row]) => [cid, JSON.parse(row.payload as string)])) } : {}), + ...(out.update.size > 0 ? { update: Object.fromEntries([...out.update.values()].map((row) => [row.id, JSON.parse(row.payload as string)])) } : {}), + ...(out.destroy.size > 0 ? { destroy: [...out.destroy.keys()] } : {}), + }, 's']], CONTACTS_USING); + const [, result] = responseFor(response, 's'); + + // What the server now holds. A network failure here is a lost response: the rows stay. + const createdIds = new Map(); + for (const cid of creationIds.keys()) { + const created = result.created?.[cid] as { id?: unknown } | undefined; + if (created?.id) createdIds.set(cid, String(created.id)); + } + const updatedIds = [...out.update.keys()].filter((id) => result.updated && id in result.updated); + const readBack = await readServerCards(ctx, jmap, [...createdIds.values(), ...updatedIds], limits); + // A card already has this uid: maybe this one, sent before. + const uidRefused = [...creationIds].filter(([cid]) => (result.notCreated?.[cid] as SetError | undefined)?.properties?.includes('uid')); + const taken = await cardsByUid(ctx, jmap, uidRefused.map(([, row]) => row.uid as string), limits); + + db.transaction(() => { + for (const [cid, row] of creationIds) { + const realId = createdIds.get(cid); + const refused = result.notCreated?.[cid] as SetError | undefined; + if (realId) { + const card = readBack.get(realId); + if (card) { + settleCreated(ctx, row, card); + } else { + // Created and deleted again already. The delta will say the same. + dropPending(db, jmap, row.id); + removeCard(db, jmap, row.id); + } + } else if (refused?.properties?.includes('uid')) { + const card = taken.get(row.uid as string); + if (card && holdsSent(row.payload as string, card)) settleCreated(ctx, row, card); + else settleRefused(ctx, row, refused, undefined); + } else if (refused) { + settleRefused(ctx, row, refused, undefined); + } + } + for (const [id, row] of out.update) { + const refused = result.notUpdated?.[id] as SetError | undefined; + if (refused) settleRefused(ctx, row, refused, current.get(id)); + else if (updatedIds.includes(id)) settleUpdated(ctx, row, readBack.get(id)); + } + for (const [id, row] of out.destroy) { + const refused = result.notDestroyed?.[id] as SetError | undefined; + if (refused) settleRefused(ctx, row, refused, current.get(id)); + else if ((result.destroyed as string[] | undefined)?.includes(id)) dropPending(db, jmap, id); + } + }); +} + +/** The server said no to the request itself, for a reason that won't go away by itself. */ +function requestRefusal(error: unknown): SetError | null { + if (error instanceof JmapMethodError) return TRANSIENT_ERRORS.has(error.type) ? null : { type: error.type, description: error.message }; + if (error instanceof TransportError && error.kind === 'http' && error.status !== undefined && error.status >= 400 && error.status < 500) { + return { type: 'requestRefused', description: error.message }; + } + return null; +} + +/** One card may be what makes a request fail: the others go one by one, and that one becomes a conflict. */ +async function sendRows(ctx: SyncContext, jmap: string, rows: readonly PendingRow[], limits: RequestLimits): Promise { + try { + await exchange(ctx, jmap, rows, limits); + } catch (error) { + const refusal = requestRefusal(error); + if (!refusal) throw error; + if (rows.length > 1) { + for (const row of rows) { + if (ctx.stopped()) return; + await sendRows(ctx, jmap, [row], limits); + } + return; + } + ctx.db.transaction(() => { + const row = pendingFor(ctx.db, jmap, rows[0].id); + if (row) settleRefused(ctx, row, refusal, undefined); + }); + } +} + /** Sends what's queued for one JMAP account. Network failures reach the engine and the rows stay. */ export async function flushPending(ctx: SyncContext, jmap: string): Promise { - const { db } = ctx; + const limits = requestLimits(ctx.session); let afterSeq = 0; let settled = false; - for (;;) { - if (ctx.stopped()) break; - const batch = nextBatch(db, jmap, afterSeq); - if (batch.length === 0) break; - afterSeq = batch[batch.length - 1].seq; - settled = true; - - const current = await readServerCards(ctx, jmap, batch.filter((row) => row.op !== 'create' && !row.id.startsWith(STAND_IN_PREFIX)).map((row) => row.id)); - const attempted = batch.filter((row) => row.uid && row.attempted).map((row) => row.uid as string); - const uidIds = await lookUpUids(ctx, jmap, attempted); - const uidCards = await readServerCards(ctx, jmap, [...uidIds.values()]); - const serverByUid = new Map(); - for (const [uid, id] of uidIds) { - const card = uidCards.get(id); - if (card) serverByUid.set(uid, card); - } - - const out: Decision = { create: new Map(), update: new Map(), destroy: new Map() }; - db.transaction(() => { - for (const { seq, id } of batch) { - // The page may have edited it while the reads were on their way. - const row = pendingFor(db, jmap, id); - if (row && row.seq === seq) decide(ctx, row, current.get(id), serverByUid, out); - } - for (const row of out.create.values()) db.run('UPDATE contact_pending SET attempted = 1 WHERE seq = ?', [row.seq]); - }); - if (out.create.size + out.update.size + out.destroy.size === 0) continue; - - const creationIds = new Map([...out.create.values()].map((row) => [`c${row.seq}`, row])); - const response = await ctx.transport.request([['ContactCard/set', { - accountId: jmap, - ...(creationIds.size > 0 ? { create: Object.fromEntries([...creationIds].map(([cid, row]) => [cid, JSON.parse(row.payload as string)])) } : {}), - ...(out.update.size > 0 ? { update: Object.fromEntries([...out.update.values()].map((row) => [row.id, JSON.parse(row.payload as string)])) } : {}), - ...(out.destroy.size > 0 ? { destroy: [...out.destroy.keys()] } : {}), - }, 's']], CONTACTS_USING); - const [, result] = responseFor(response, 's'); - - const refusedUid: Array<[PendingRow, SetError]> = []; - db.transaction(() => { - for (const [cid, row] of creationIds) { - const created = result.created?.[cid] as JsonObject | undefined; - const refused = result.notCreated?.[cid] as SetError | undefined; - if (created) { - settleCreated(ctx, row, { ...(JSON.parse(row.payload as string) as JsonObject), ...created, id: String(created.id) }); - } else if (refused?.properties?.includes('uid')) { - refusedUid.push([row, refused]); - } else if (refused) { - settleRefused(ctx, row, refused, undefined); - } - } - for (const [id, row] of out.update) { - const refused = result.notUpdated?.[id] as SetError | undefined; - if (refused) settleRefused(ctx, row, refused, current.get(id)); - else if (result.updated && id in result.updated) settleUpdated(ctx, row, current.get(id) as Card); - } - for (const [id, row] of out.destroy) { - const refused = result.notDestroyed?.[id] as SetError | undefined; - if (refused) settleRefused(ctx, row, refused, current.get(id)); - else if ((result.destroyed as string[] | undefined)?.includes(id)) dropPending(db, jmap, id); - } - }); - - // The server has a card with this uid already: this one, sent before. - if (refusedUid.length > 0) { - const ids = await lookUpUids(ctx, jmap, refusedUid.map(([row]) => row.uid as string)); - const cards = await readServerCards(ctx, jmap, [...ids.values()]); - db.transaction(() => { - for (const [row, refused] of refusedUid) { - const id = ids.get(row.uid as string); - const card = id ? cards.get(id) : undefined; - if (card) settleCreated(ctx, row, { ...(JSON.parse(row.payload as string) as JsonObject), ...card }); - else settleRefused(ctx, row, refused, undefined); - } - }); + try { + for (;;) { + if (ctx.stopped()) break; + const batch = nextBatch(ctx.db, jmap, afterSeq, limits); + if (batch.length === 0) break; + afterSeq = batch[batch.length - 1].seq; + settled = true; + await sendRows(ctx, jmap, batch, limits); } + } finally { + if (settled) ctx.emit({ type: 'changed', account: ctx.account, collections: ['ContactCard'] }); } - if (settled) ctx.emit({ type: 'changed', account: ctx.account, collections: ['ContactCard'] }); } diff --git a/lib/store/contacts/schema.ts b/lib/store/contacts/schema.ts index 66b30832c..9aceaf22b 100644 --- a/lib/store/contacts/schema.ts +++ b/lib/store/contacts/schema.ts @@ -48,6 +48,8 @@ export const CONTACT_MIGRATIONS: readonly Migration[] = [ base TEXT, -- A request may have reached the server, so a create looks for its uid first. attempted INTEGER NOT NULL DEFAULT 0, + -- The payload that request held: the server may have this, not the newer payload. + sent TEXT, queued_at INTEGER NOT NULL, UNIQUE (jmap, id) ); diff --git a/lib/store/contacts/sync.ts b/lib/store/contacts/sync.ts index f734263d6..b06ddc835 100644 --- a/lib/store/contacts/sync.ts +++ b/lib/store/contacts/sync.ts @@ -2,18 +2,24 @@ * Contact sync for one JMAP account: first what's queued goes out, then the * address books, then the cards. A full pass over the cards captures the * state first and pages, with its cursor in sync_state, so a reload continues - * where it stopped. Cards the pass didn't see are gone from the server. + * where it stopped. Cards the pass didn't see are gone from the server, so a + * page must not skip any: it goes from the last card read (`anchor`), since a + * position moves when an earlier card is deleted. Both Stalwarts take that. */ import type { SyncContext, Syncer } from '../area'; import { runChangesLoop } from '../engine/changes'; import { deleteObject, putObject } from '../engine/core-schema'; -import { responseFor, type MethodCall } from '../engine/transport'; +import { requestLimits } from '../engine/limits'; +import { JmapMethodError, responseFor, TransportError, type MethodCall } from '../engine/transport'; import { flushPending, serverDestroyed, storeServerCard } from './pending'; import { ADDRESS_BOOK_PROPERTIES, CONTACTS_CAPABILITY, CONTACTS_USING, type Card } from './records'; -const PAGE = 500; -const PAIRS_PER_REQUEST = 4; +const MOST_PAIRS_PER_REQUEST = 4; + +function pageSize(ctx: SyncContext): number { + return requestLimits(ctx.session).maxObjectsInGet; +} // ---- Address books ---- @@ -42,8 +48,9 @@ async function syncBooks(ctx: SyncContext, jmap: string): Promise { apply: async (page) => { const ids = [...page.created, ...page.updated]; const list: Array<{ id: string }> = []; - for (let i = 0; i < ids.length; i += PAGE) { - const response = await ctx.transport.request([['AddressBook/get', { accountId: jmap, ids: ids.slice(i, i + PAGE), properties: ADDRESS_BOOK_PROPERTIES }, 'b']], CONTACTS_USING); + const size = pageSize(ctx); + for (let i = 0; i < ids.length; i += size) { + const response = await ctx.transport.request([['AddressBook/get', { accountId: jmap, ids: ids.slice(i, i + size), properties: ADDRESS_BOOK_PROPERTIES }, 'b']], CONTACTS_USING); list.push(...responseFor(response, 'b')[1].list); } ctx.db.transaction(() => { @@ -59,14 +66,16 @@ async function syncBooks(ctx: SyncContext, jmap: string): Promise { // ---- Cards ---- -interface ListCursor { position: number; gen: number; captured: string } +interface ListCursor { position: number; anchor: string | null; gen: number; captured: string } async function startFullList(ctx: SyncContext, jmap: string): Promise { const response = await ctx.transport.request([['ContactCard/get', { accountId: jmap, ids: [] }, 's']], CONTACTS_USING); const captured = String(responseFor(response, 's')[1].state); const gen = Number(ctx.db.get<{ gen: number | null }>('SELECT MAX(gen) AS gen FROM contact_card WHERE jmap = ?', [jmap])?.gen ?? 0) + 1; - const cursor: ListCursor = { position: 0, gen, captured }; - ctx.db.transaction(() => ctx.setSyncState(jmap, 'ContactCard', { state: null, phase: 'list', cursor: { ...cursor } })); + const cursor: ListCursor = { 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, 'ContactCard').state ? 'list' : 'first'; + ctx.db.transaction(() => ctx.setSyncState(jmap, 'ContactCard', { state: null, phase, cursor: { ...cursor } })); return cursor; } @@ -82,27 +91,47 @@ function sweep(ctx: SyncContext, jmap: string, gen: number): void { async function runFullList(ctx: SyncContext, jmap: string, start: ListCursor): Promise { const cursor = { ...start }; + const limits = requestLimits(ctx.session); + const page = limits.maxObjectsInGet; + const pairs = Math.max(1, Math.min(MOST_PAIRS_PER_REQUEST, Math.floor(limits.maxCallsInRequest / 2))); for (;;) { if (ctx.stopped()) return false; const calls: MethodCall[] = []; - for (let i = 0; i < PAIRS_PER_REQUEST; i += 1) { - calls.push(['ContactCard/query', { accountId: jmap, position: cursor.position + i * PAGE, limit: PAGE, calculateTotal: i === 0 }, `q${i}`]); + for (let i = 0; i < pairs; i += 1) { + const from = cursor.anchor === null ? { position: i * page } : { anchor: cursor.anchor, anchorOffset: 1 + i * page }; + calls.push(['ContactCard/query', { accountId: jmap, ...from, limit: page, calculateTotal: i === 0 }, `q${i}`]); calls.push(['ContactCard/get', { accountId: jmap, '#ids': { resultOf: `q${i}`, name: 'ContactCard/query', path: '/ids' } }, `g${i}`]); } const response = await ctx.transport.request(calls, CONTACTS_USING); let total: number | null = null; let got = 0; + let reached = 0; let short = false; + let last: string | null = null; const cards: Card[] = []; - for (let i = 0; i < PAIRS_PER_REQUEST && !short; i += 1) { - const [, query] = responseFor(response, `q${i}`); - if (i === 0 && typeof query.total === 'number') total = query.total; - cards.push(...responseFor(response, `g${i}`)[1].list as Card[]); - got += (query.ids as string[]).length; - short = (query.ids as string[]).length < PAGE; + try { + for (let i = 0; i < pairs && !short; i += 1) { + const [, query] = responseFor(response, `q${i}`); + const ids = query.ids as string[]; + if (i === 0 && typeof query.total === 'number') total = query.total; + cards.push(...responseFor(response, `g${i}`)[1].list as Card[]); + got += ids.length; + if (ids.length > 0) last = ids[ids.length - 1]; + reached = Number(query.position ?? 0) + ids.length; + short = ids.length < page; + } + } catch (error) { + if (!(error instanceof JmapMethodError && error.type === 'anchorNotFound')) throw error; + // The card it went from is gone. Going through again costs requests, not cards. + cursor.anchor = null; + cursor.position = 0; + ctx.db.transaction(() => ctx.setSyncState(jmap, 'ContactCard', { cursor: { ...cursor } })); + continue; } - cursor.position += got; - const done = got === 0 || (total === null ? short : cursor.position >= total); + if (last !== null) cursor.anchor = last; + cursor.position = reached; + // The server may clamp a page, so a short one only ends this if there's no total to go by. + const done = got === 0 || (total === null ? short : reached >= total); ctx.db.transaction(() => { for (const card of cards) storeServerCard(ctx.db, jmap, card, cursor.gen); if (done) { @@ -132,8 +161,9 @@ async function syncDelta(ctx: SyncContext, jmap: string, since: string, target: apply: async (page) => { const ids = [...page.created, ...page.updated]; const fetched: Card[] = []; - for (let i = 0; i < ids.length; i += PAGE) { - const response = await ctx.transport.request([['ContactCard/get', { accountId: jmap, ids: ids.slice(i, i + PAGE) }, 'g']], CONTACTS_USING); + const size = pageSize(ctx); + for (let i = 0; i < ids.length; i += size) { + const response = await ctx.transport.request([['ContactCard/get', { accountId: jmap, ids: ids.slice(i, i + size) }, 'g']], CONTACTS_USING); fetched.push(...responseFor(response, 'g')[1].list as Card[]); } ctx.db.transaction(() => { @@ -151,7 +181,7 @@ async function syncDelta(ctx: SyncContext, jmap: string, since: string, target: async function syncCards(ctx: SyncContext, jmap: string): Promise { const row = ctx.syncState(jmap, 'ContactCard'); - 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); @@ -170,7 +200,14 @@ export const contactsSyncer: Syncer = { name: 'contacts', capability: CONTACTS_CAPABILITY, async run(ctx, jmap) { - await flushPending(ctx, jmap); + try { + await flushPending(ctx, jmap); + } catch (error) { + // No network, no credentials or a busy server is the engine's to handle. + // Anything else leaves the edits queued and mustn't stop contacts syncing. + if (error instanceof TransportError && error.kind !== 'http') throw error; + ctx.emit({ type: 'error', account: ctx.account, area: 'contacts', message: error instanceof Error ? error.message : String(error) }); + } if (ctx.stopped()) return; await syncBooks(ctx, jmap); if (ctx.stopped()) return; diff --git a/lib/store/engine/limits.ts b/lib/store/engine/limits.ts new file mode 100644 index 000000000..e79f8da8e --- /dev/null +++ b/lib/store/engine/limits.ts @@ -0,0 +1,31 @@ +/** What one request may hold: the session's limits, never past the ones the engine was built for. */ + +import type { StoreSession } from './transport'; + +export interface RequestLimits { + maxObjectsInGet: number; + maxObjectsInSet: number; + maxCallsInRequest: number; + maxSizeRequest: number; +} + +const DEFAULTS: RequestLimits = { + maxObjectsInGet: 500, + maxObjectsInSet: 500, + maxCallsInRequest: 16, + maxSizeRequest: 10 * 1024 * 1024, +}; + +export function requestLimits(session: StoreSession): RequestLimits { + const core = (session.capabilities['urn:ietf:params:jmap:core'] ?? {}) as Partial; + const limit = (key: keyof RequestLimits) => { + const value = core[key]; + return typeof value === 'number' && value > 0 ? Math.min(value, DEFAULTS[key]) : DEFAULTS[key]; + }; + return { + maxObjectsInGet: limit('maxObjectsInGet'), + maxObjectsInSet: limit('maxObjectsInSet'), + maxCallsInRequest: limit('maxCallsInRequest'), + maxSizeRequest: limit('maxSizeRequest'), + }; +}