starred/webmail
1
0
Fork 0
mirror of https://github.com/bulwarkmail/webmail.git synced 2026-11-04 16:14:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: send contact edits that remove entries and stop reading own edits as conflicts

Browse Source 
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.
This commit is contained in:
Linus Rath
2026-11-03 06:20:39 +01:00
parent 691deff583
commit 9fbe33004b
8 changed files with 788 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/store/__tests__/contacts.test.ts
+386 -6
View File
@@ -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<string, unknown>, id: string) => call(engine, 'ContactCard/set', { update: { [id]: patch } });
const conflictsOf = async (engine: AccountEngine) => await engine.areaCall('contacts.conflicts', undefined) as Array<Record<string, any>>;
const pendingOf = (engine: AccountEngine) => engine.areaCall('contacts.pending', undefined);
describe('edits that make a card smaller', () => {
it.each<[string, Record<string, unknown>, Record<string, unknown>, Record<string, unknown>]>([
['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<string, unknown> }).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 */
lib/store/__tests__/helpers/fake-contacts.ts
+45 -6
View File
@@ -37,6 +37,15 @@ export function changesFor(coll: Collection, args: Record<string, unknown>): Rec
return { accountId: args.accountId, oldState: since, ...coll.changes(since, maxChanges) };
}
function storedMedia(card: Obj): void {
const media = card.media as Record<string, Record<string, unknown>> | 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<string, unknown>)[key] = { ...rest, blobId: `stored-${uri.length}-${key}` };
}
}
function applyServerPatch(card: Obj, patch: Record<string, unknown>): 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<string, { type: string; description?: string; properties?: string[] }>();
/** 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<string, unknown>) => { 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<string, boolean>)[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<JmapResponse> {
// 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');
lib/store/contacts/methods.ts
+6 -2
View File
@@ -15,10 +15,14 @@ function jmapOf(ctx: MethodContext, args: Record<string, unknown>): 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;
}
lib/store/contacts/patch.ts
+28 -7
View File
@@ -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;
}
lib/store/contacts/pending.ts
+230 -123
View File
@@ -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<Map<string, Card>> {
async function readServerCards(ctx: SyncContext, jmap: string, ids: string[], limits: RequestLimits): Promise<Map<string, Card>> {
const cards = new Map<string, Card>();
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<Map<string, string>> {
const found = new Map<string, string>();
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<Map<string, Card>> {
const ids = new Map<string, string>();
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<string, Card>();
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<string, PendingRow>; update: Map<string, PendingRow>; destroy: Map<string, PendingRow> }
/**
* 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<string, Card>, 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<PendingRow>('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<PendingRow>('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<void> {
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<string, string>();
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<void> {
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<void> {
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<string, Card>();
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'] });
}
lib/store/contacts/schema.ts
+2
View File
@@ -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)
);
lib/store/contacts/sync.ts
+60 -23
View File
@@ -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<void> {
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<void> {
// ---- 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<ListCursor> {
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<boolean> {
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<void> {
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;
lib/store/engine/limits.ts
+31
View File
@@ -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<RequestLimits>;
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'),
};
}