From 58b74527b221cf7286e1acabe58f43dd1a5e1f59 Mon Sep 17 00:00:00 2001 From: Linus Rath <139418639+rathlinus@users.noreply.github.com> Date: Sun, 27 Sep 2026 07:42:08 +0200 Subject: [PATCH] fix: keep group cards on the server when a sync stops while dropping them #34 The contacts SyncState lists the group rows on the device, so a group an app hard-deleted (Fossify does) is found by its absence and deleted on the server. The list changed only with a later state op, not with the batch that inserted or deleted the rows. When a run stopped in between (cancelled, out of time, a crash, or turning sync off while a deselected book was being dropped), a group the engine had removed was still listed, taken for a hard delete by the next run or by the teardown's upload, and its card destroyed on the server; a group it had inserted was never listed, so a later hard delete never went up. A batch that inserts or deletes a group row now ends with a state op holding that change, retries included, so the list and the rows never disagree. A group a download finds unchanged is listed too, for rows whose batch lost its state op at a provider yield point. And an absent group whose card is no longer a group or is in no synced book is only forgotten, never destroyed. Checked with a crash at every checkpoint of a deselection, a teardown and a deadline stop during one, a server-side move in a page of several chunks, and a group written before an interruption and then deleted in Fossify. --- .../__tests__/engine/group-listing.test.ts | 195 ++++++++++++++++++ .../__tests__/engine/internals.test.ts | 32 ++- src/device-sync/engine/batch.ts | 44 +++- src/device-sync/engine/contacts-sync.ts | 30 ++- src/device-sync/engine/item-sync.ts | 41 +++- src/device-sync/engine/run.ts | 10 +- src/device-sync/engine/sync-state.ts | 20 +- 7 files changed, 350 insertions(+), 22 deletions(-) create mode 100644 src/device-sync/__tests__/engine/group-listing.test.ts diff --git a/src/device-sync/__tests__/engine/group-listing.test.ts b/src/device-sync/__tests__/engine/group-listing.test.ts new file mode 100644 index 00000000..16e47c49 --- /dev/null +++ b/src/device-sync/__tests__/engine/group-listing.test.ts @@ -0,0 +1,195 @@ +// SyncState.groups lists the group cards present on the device, so a group an +// app hard-deleted (Fossify) is found by its absence and deleted on the server. +// The list must change in the same batch as the rows: a group row the engine +// removed but still listed reads as a hard delete and destroys the card, and a +// row it inserted but never listed hides a later hard delete. + +import { describe, expect, it } from 'vitest'; +import { calendarPlanner } from '../../calendar/planner'; +import { contactsPlanner } from '../../contacts/planner'; +import { serializeSyncState } from '../../engine/sync-state'; +import { addServerCards, ANDROID_ACCOUNT, CONTACTS_AUTHORITY, createHarness, RUN_BUDGET, type Harness } from './harness'; + +function real(): Harness { + const h = createHarness(); + h.deps.planners = { contacts: contactsPlanner, calendar: calendarPlanner }; + return h; +} + +/** The personal book with Ada, and a Work book with Boss, Colleague and the group "Team" (Boss). */ +async function withWorkGroup() { + const h = real(); + const work = h.server.addAddressBook('a', { name: 'Work' }); + addServerCards(h, ['Ada']); + const [boss] = addServerCards(h, ['Boss', 'Colleague'], 'a', work); + const bossUid = h.server.get('ContactCard', 'a', boss)!.uid as string; + const group = h.server.addCard('a', { + uid: 'g-team', + kind: 'group', + name: { full: 'Team' }, + addressBookIds: { [work]: true }, + members: { [bossUid]: true }, + }); + expect((await h.run()).outcome).toBe('ok'); + expect(h.state()?.accounts.a.groups).toEqual([`a/${group}`]); + h.checkpoints.count = 0; + return { h, group, work }; +} + +/** Replaces the stored SyncState's `groups` of account `a` (a list that drifted from the rows). */ +async function storeGroups(h: Harness, groups: string[]): Promise { + const state = h.state()!; + state.accounts.a.groups = groups; + await h.device.port(ANDROID_ACCOUNT, CONTACTS_AUTHORITY).applyBatch([{ op: 'syncState', value: serializeSyncState(state) }]); +} + +describe('device sync engine: the groups a SyncState lists', () => { + it('keeps a group card on the server after a deselection stopped at any checkpoint', async () => { + const probe = await withWorkGroup(); + probe.h.prefs.contactsSelection[`a/${probe.work}`] = false; + await probe.h.run(); + const total = probe.h.checkpoints.count; + expect(probe.h.server.get('ContactCard', 'a', probe.group)).toBeDefined(); + + for (let n = 1; n <= total; n++) { + const { h, group, work } = await withWorkGroup(); + h.prefs.contactsSelection[`a/${work}`] = false; + h.checkpoints.crashAt = n; + await h.run(); + h.checkpoints.crashAt = null; + expect((await h.run()).outcome, `run after a crash at checkpoint ${n}`).toBe('ok'); + expect(h.server.get('ContactCard', 'a', group), `group card after a crash at checkpoint ${n}`).toBeDefined(); + expect(h.device.rows('groups')).toEqual([]); + expect(h.state()?.accounts.a.groups).toEqual([]); + } + }); + + it('keeps a group card on the server when its book is selected again after a stopped deselection', async () => { + const { h, group, work } = await withWorkGroup(); + h.prefs.contactsSelection[`a/${work}`] = false; + // Checkpoint 3 comes right after the deselected book's group rows went. + h.checkpoints.crashAt = 3; + await h.run(); + h.checkpoints.crashAt = null; + expect(h.device.rows('groups')).toEqual([]); + + h.prefs.contactsSelection[`a/${work}`] = true; + const report = await h.run(); + + expect(report).toMatchObject({ outcome: 'ok', stats: { uploaded: { deleted: 0 } } }); + expect(h.server.get('ContactCard', 'a', group)).toBeDefined(); + }); + + it('keeps a group card on the server when sync is turned off while a deselection drops it', async () => { + const { h, group, work } = await withWorkGroup(); + h.prefs.contactsSelection[`a/${work}`] = false; + let teardown: Promise<{ pending: number }> | null = null; + h.checkpoints.onCheckpoint = (n) => { + // Right before the run drops the deselected book's rows. + if (n === 2 && !teardown) teardown = h.teardown(CONTACTS_AUTHORITY); + }; + + const run = await h.run(); + h.checkpoints.onCheckpoint = undefined; + + expect(run.outcome).toBe('cancelled'); + expect(await teardown!).toEqual({ pending: 0 }); + expect(h.contacts()).toHaveLength(0); + expect(h.server.get('ContactCard', 'a', group)).toBeDefined(); + }); + + it('keeps a group card on the server when the deselecting run runs out of time after dropping the group', async () => { + const { h, group, work } = await withWorkGroup(); + h.prefs.contactsSelection[`a/${work}`] = false; + h.checkpoints.onCheckpoint = (n) => { + if (n === 2) h.clock.now += RUN_BUDGET; + }; + + expect((await h.run()).outcome).toBe('cancelled'); + h.checkpoints.onCheckpoint = undefined; + expect((await h.run()).outcome).toBe('ok'); + + expect(h.server.get('ContactCard', 'a', group)).toBeDefined(); + }); + + it('keeps a group card the server moved to a book that does not sync when the run stops within the page', async () => { + const h = real(); + const archive = h.server.addAddressBook('a', { name: 'Archive' }); + h.prefs.contactsSelection[`a/${archive}`] = false; + const ids = addServerCards(h, Array.from({ length: 60 }, (_, i) => `P${String(i).padStart(2, '0')}`)); + const group = h.server.addCard('a', { uid: 'g-team', kind: 'group', name: { full: 'Team' }, addressBookIds: { [h.book]: true }, members: {} }); + await h.run(); + expect(h.device.rows('groups')).toHaveLength(1); + // Another client moves the group to Archive and renames 60 contacts: one /changes page, several chunks. + h.server.serverUpdate('ContactCard', 'a', group, { addressBookIds: { [archive]: true } }); + ids.forEach((id, i) => h.server.serverUpdate('ContactCard', 'a', id, { 'name/full': `Q${i}` })); + // Checkpoints: 'changes', 'download' (the group's chunk), 'download' (the first contacts chunk) … + h.checkpoints.count = 0; + h.checkpoints.crashAt = 3; + + await h.run(); + h.checkpoints.crashAt = null; + expect(h.device.rows('groups')).toHaveLength(0); + expect((await h.run()).outcome).toBe('ok'); + + expect(h.server.get('ContactCard', 'a', group)).toBeDefined(); + expect(h.state()?.accounts.a.groups).toEqual([]); + }); + + it('lists a group written before an interruption, so a hard delete by an app still reaches the server', async () => { + const h = real(); + addServerCards(h, Array.from({ length: 60 }, (_, i) => `P${String(i).padStart(2, '0')}`)); + const group = h.server.addCard('a', { uid: 'g1', kind: 'group', name: { full: 'Friends' }, members: {} }); + // First sync: 'reconcile', 'reconcile:ids', 'reconcile:objects' (the group's chunk), then out of time. + h.checkpoints.onCheckpoint = (n) => { + if (n === 3) h.clock.now += RUN_BUDGET; + }; + expect((await h.run()).outcome).toBe('cancelled'); + h.checkpoints.onCheckpoint = undefined; + expect(h.device.rows('groups')).toHaveLength(1); + expect(h.state()?.accounts.a.groups).toEqual([`a/${group}`]); + expect((await h.run()).outcome).toBe('ok'); + expect(h.contacts()).toHaveLength(60); + + // Fossify hard-deletes the group through a sync-adapter URI. + h.device.user.fossifyDeleteGroup(Number(h.device.rows('groups')[0]._id)); + expect((await h.run()).outcome).toBe('ok'); + + expect(h.server.get('ContactCard', 'a', group)).toBeUndefined(); + }); + + it('lists a group again when it comes back as an echo', async () => { + const h = real(); + const group = h.server.addCard('a', { uid: 'g1', kind: 'group', name: { full: 'Friends' }, members: {} }); + await h.run(); + // A list that lost the group (a write committed without its state op), then a download of the unchanged card. + await storeGroups(h, []); + addServerCards(h, ['Ada']); + h.server.truncateChangeLog('a', 'contacts'); + expect((await h.run()).outcome).toBe('ok'); + expect(h.state()?.accounts.a.groups).toEqual([`a/${group}`]); + + h.device.user.fossifyDeleteGroup(Number(h.device.rows('groups')[0]._id)); + await h.run(); + + expect(h.server.get('ContactCard', 'a', group)).toBeUndefined(); + }); + + it('never destroys a listed group card that is in no synced book or no longer a group, and forgets it', async () => { + const h = real(); + const archive = h.server.addAddressBook('a', { name: 'Archive' }); + h.prefs.contactsSelection[`a/${archive}`] = false; + const filed = h.server.addCard('a', { uid: 'g1', kind: 'group', name: { full: 'Filed' }, addressBookIds: { [archive]: true }, members: {} }); + const person = h.server.addCard('a', { uid: 'p1', name: { full: 'Was a group' }, addressBookIds: { [h.book]: true } }); + await h.run(); + // The list names both as groups whose rows are gone (the engine removed them, and the batch storing that was lost). + await storeGroups(h, [`a/${filed}`, `a/${person}`]); + + const report = await h.run(); + + expect(report).toMatchObject({ outcome: 'ok', stats: { uploaded: { deleted: 0 } } }); + expect(h.server.get('ContactCard', 'a', filed)).toBeDefined(); + expect(h.server.get('ContactCard', 'a', person)).toBeDefined(); + expect(h.state()?.accounts.a.groups).toEqual([]); + }); +}); diff --git a/src/device-sync/__tests__/engine/internals.test.ts b/src/device-sync/__tests__/engine/internals.test.ts index 8b58b263..9609e1b4 100644 --- a/src/device-sync/__tests__/engine/internals.test.ts +++ b/src/device-sync/__tests__/engine/internals.test.ts @@ -8,7 +8,7 @@ import { classifyFailure, RunAbort, StateMismatch, StopRun } from '../../engine/ import { acquireLock } from '../../engine/mutex'; import { isBackedOff, nextMarker } from '../../engine/poison'; import { collectionDefaultOn, isCollectionSelected } from '../../engine/selection'; -import { emptySyncState, parseSyncState, serializeSyncState, StateStore } from '../../engine/sync-state'; +import { accountOf, emptySyncState, parseSyncState, serializeSyncState, StateStore, type SyncState } from '../../engine/sync-state'; import { classifySetError, existingIdFromUidError, JmapCaller } from '../../jmap/caller'; import { openWithAuthRebuild } from '../../jmap/connection'; import { JMAPMethodError } from '../../jmap/errors'; @@ -127,6 +127,36 @@ describe('provider batches', () => { await expect(writer(scope).write([{ group: { ref: 'g', ops: [insert('raw_contacts')] } }])).rejects.toThrow(/not our account/); }); + it("stores a group's state change in the batch that applies it, its retry included, and never without it", async () => { + const store = new StateStore(emptySyncState(), { registryId: 'r', origin: 'o' }); + let failures = 1; + const p = port((ops) => (ops.some((op) => op.op === 'assert') && failures-- > 0 ? { ok: false, reason: 'assert', message: 'stale' } : ok(ops))); + const listed = (ref: string) => (next: SyncState) => { + accountOf(next, 'a').groups = [...(accountOf(next, 'a').groups ?? []), ref]; + }; + const groupOps = (id: number): ProviderOp[] => [ + { op: 'assert', table: 'groups', id, values: { version: 1 } }, + { op: 'update', table: 'groups', id, values: { sourceid: `a/g${id}` } }, + ]; + const plain: Work = { group: { ref: 'c', ops: [insert('raw_contacts'), insert('data')] } }; + const retried: Work = { group: { ref: 'a/g1', ops: groupOps(1) }, state: listed('a/g1') }; + retried.replan = async () => ({ group: retried.group, state: retried.state }); + const w = new BatchWriter(p, { maxOps: 2, maxBytes: 300_000, maxReplans: 3 }, store); + + await w.write([plain, retried]); + + // The plain group alone, without a state op; the group with its state op, refused; its retry, with the state op. + expect(p.sent.map((b) => b.map((op) => op.op))).toEqual([['insert', 'insert'], ['assert', 'update', 'syncState'], ['assert', 'update', 'syncState']]); + expect(store.committed.accounts.a.groups).toEqual(['a/g1']); + + failures = Infinity; + const given: Work = { group: { ref: 'a/g2', ops: groupOps(2) }, state: listed('a/g2'), replan: async () => null }; + await w.write([given], store.tail(() => undefined)); + + expect(p.sent.at(-1)!.map((op) => op.op)).toEqual(['syncState']); + expect(store.committed.accounts.a.groups).toEqual(['a/g1']); + }); + it('skips groups that only assert', async () => { const p = port(ok); await writer(p).write([{ group: { ref: 'g', ops: [{ op: 'assert', table: 'raw_contacts', id: 1, values: { dirty: 0 } }] } }]); diff --git a/src/device-sync/engine/batch.ts b/src/device-sync/engine/batch.ts index 2b4095ee..2375b86f 100644 --- a/src/device-sync/engine/batch.ts +++ b/src/device-sync/engine/batch.ts @@ -15,12 +15,16 @@ * a committed group fail harmlessly); a group that fails an assert is * re-read and re-planned up to `maxReplans` times, then given up. `tooLarge` * splits the batch; `permission` ends the run. + * + * A group whose rows the SyncState describes (a group row: `groups`) carries + * that `state` change: every batch that applies such a group, a retry + * included, ends with a state op holding it. */ import type { OpGroup } from '../planner'; import type { BatchFailure, OpResult, ProviderOp, ProviderPort } from '../types'; import { RunAbort } from './errors'; import { hasWrites } from './provider'; -import type { Tail } from './sync-state'; +import type { StateChange, StateStore, Tail } from './sync-state'; export interface Work { group: OpGroup; @@ -30,6 +34,8 @@ export interface Work { replan?(): Promise; /** Given up: the item's rows stay as they are. */ failed?(reason: BatchFailure, message: string): void | Promise; + /** What the SyncState lists changes with these rows: stored by the batch that applies them. */ + state?: StateChange; } export interface BatchWriterOptions { @@ -100,6 +106,8 @@ export class BatchWriter { constructor( private readonly port: ProviderPort, private readonly options: BatchWriterOptions, + /** Where the `state` of works goes; without it, it is ignored. */ + private readonly store?: StateStore, ) {} /** @@ -162,9 +170,26 @@ export class BatchWriter { return result; } + /** The works' state changes go into the state op of the batch being sent: `tail`, or one of its own. */ + private stage(works: readonly Work[], tail?: Tail): { staged: Work[]; tail?: Tail } { + const store = this.store; + const staged = store ? works.filter((w) => w.state) : []; + for (const work of staged) store!.stage(work, work.state!); + return { staged, tail: tail ?? (staged.length ? store!.tail(() => undefined) : undefined) }; + } + + private unstage(staged: readonly Work[]): void { + for (const work of staged) this.store!.unstage(work); + } + private async applyBatch(works: Work[], tail?: Tail): Promise { - const tailOp = tail?.op(); - const result = await this.send(buildBatch(works.map((w) => w.group), tailOp)); + const stated = this.stage(works, tail); + let result; + try { + result = await this.send(buildBatch(works.map((w) => w.group), stated.tail?.op())); + } finally { + this.unstage(stated.staged); + } if (result.ok) { let offset = 0; for (const work of works) { @@ -172,7 +197,7 @@ export class BatchWriter { await work.applied?.(result.results.slice(offset, offset + n)); offset += n; } - tail?.applied(); + stated.tail?.applied(); return; } if (result.reason === 'tooLarge' && works.length > 1) { @@ -191,9 +216,16 @@ export class BatchWriter { private async applyAlone(work: Work, attempt: number): Promise { if (!hasWrites(work.group.ops)) return; - const result = await this.send(buildBatch([work.group])); + const stated = this.stage([work]); + let result; + try { + result = await this.send(buildBatch([work.group], stated.tail?.op())); + } finally { + this.unstage(stated.staged); + } if (result.ok) { - await work.applied?.(result.results); + await work.applied?.(result.results.slice(0, work.group.ops.length)); + stated.tail?.applied(); return; } await this.retryAlone(work, result.reason, result.message, attempt); diff --git a/src/device-sync/engine/contacts-sync.ts b/src/device-sync/engine/contacts-sync.ts index 40257ee4..9a35eb32 100644 --- a/src/device-sync/engine/contacts-sync.ts +++ b/src/device-sync/engine/contacts-sync.ts @@ -24,7 +24,7 @@ import { accountOfRef, idInAccount, refOf, type Held, type Kind, type ServerObje import { poisonFingerprint } from './poison'; import { flag, num, str } from './provider'; import { isCollectionSelected } from './selection'; -import { accountOf, type SyncState } from './sync-state'; +import { accountOf, type StateChange, type SyncState } from './sync-state'; /** * The context the engine hands the contacts planner. `groupsOf` is @@ -496,10 +496,29 @@ export class ContactsSync extends ItemSync { else this.indexCard(acct, object as ContactCardWire); } + /** A group whose row is there already is listed too: the batch that listed it may never have been stored. */ + protected onEcho(kind: Kind, acct: string, object: ServerObject): void { + const ref = refOf(acct, object.id); + if (kind === this.groupKind && !(this.env.store.account(acct).groups ?? []).includes(ref)) this.groupChange(acct, ref, true); + } + protected onRemoved(kind: Kind, acct: string, id: string): void { if (kind === this.groupKind) this.groupChange(acct, refOf(acct, id), false); } + /** `groups` names a group row from the batch that inserts it until the batch that deletes it. */ + protected listing(kind: Kind, acct: string, id: string, present: boolean): StateChange | undefined { + if (kind !== this.groupKind) return undefined; + const ref = refOf(acct, id); + return (next) => { + const account = accountOf(next, acct); + const groups = new Set(account.groups ?? []); + if (present) groups.add(ref); + else groups.delete(ref); + account.groups = [...groups]; + }; + } + protected onAccepted(kind: Kind, acct: string, server: ServerObject): void { if (kind === this.groupKind) this.groupChange(acct, refOf(acct, server.id), true); else this.indexCard(acct, server as ContactCardWire); @@ -628,21 +647,24 @@ export class ContactsSync extends ItemSync { /** * A group hard-deleted on the device: destroyed on the server, or, when * it is also in address books that do not sync, removed from the synced - * ones only. + * ones only. A card that is no longer a group, or in no book that syncs, + * is not what the device showed: it is only forgotten. */ private async deleteAbsentGroups(acct: string): Promise { const refs = (this.absent.get(acct) ?? []).filter((ref) => this.groupChanges.get(acct)?.get(ref) !== false); if (!refs.length) return; await this.env.checkpoints.check('upload:groups'); const ids = refs.map((ref) => idInAccount(ref, acct)).filter((id): id is string => !!id); - const { list, notFound } = await this.env.jmap.get('ContactCard', acct, ids, ['id', 'addressBookIds']); + const { list, notFound } = await this.env.jmap.get('ContactCard', acct, ids, ['id', 'addressBookIds', 'kind']); for (const id of notFound) this.groupChange(acct, refOf(acct, id), false); const update: Record> = {}; const destroy: string[] = []; for (const card of list) { const books = onIds(card.addressBookIds); const synced = books.filter((id) => this.syncedKeys.has(collectionKey(acct, id))); - if (synced.length && synced.length < books.length) { + if (card.kind !== 'group' || !synced.length) { + this.groupChange(acct, refOf(acct, card.id), false); + } else if (synced.length < books.length) { update[card.id] = Object.fromEntries(synced.map((id) => [`addressBookIds/${id}`, null])); } else { destroy.push(card.id); diff --git a/src/device-sync/engine/item-sync.ts b/src/device-sync/engine/item-sync.ts index e93ec84a..84e9c254 100644 --- a/src/device-sync/engine/item-sync.ts +++ b/src/device-sync/engine/item-sync.ts @@ -32,10 +32,18 @@ import { } from './kinds'; import { isBackedOff, nextMarker } from './poison'; import { hasWrites } from './provider'; -import { accountOf, type SyncState, type Tail } from './sync-state'; +import { accountOf, type StateChange, type SyncState, type Tail } from './sync-state'; type Action = UploadAction; +function both(first: StateChange | undefined, second: StateChange | undefined): StateChange | undefined { + if (!first || !second) return first ?? second; + return (next) => { + first(next); + second(next); + }; +} + /** An item's upload, on its way to the server. */ export interface Pending { held: Held; @@ -122,8 +130,18 @@ export abstract class ItemSync { /** Authority changes folded into a state op of the account; the callback runs once that op is stored. */ protected decorateState(_acct: string, _next: SyncState): (() => void) | void {} protected onDownloaded(_kind: Kind, _acct: string, _object: ServerObject): void {} + /** A download found the object's rows as they should be already. */ + protected onEcho(_kind: Kind, _acct: string, _object: ServerObject): void {} protected onRemoved(_kind: Kind, _acct: string, _id: string): void {} protected onAccepted(_kind: Kind, _acct: string, _server: ServerObject): void {} + /** + * What the SyncState lists changes when a row of `kind` is inserted + * (`present`) or deleted (contacts: `groups`): stored by the batch that + * writes the row, so the list never disagrees with the rows. + */ + protected listing(_kind: Kind, _acct: string, _id: string, _present: boolean): StateChange | undefined { + return undefined; + } protected onDiscard(_next: SyncState, _accounts: Set): void {} /** Whether a pending create of `target` may adopt this object (contacts: the target book holds it). */ protected adoptionTargetMatches(acct: string, _object: ServerObject, target: string): boolean { @@ -588,6 +606,7 @@ export abstract class ItemSync { const ref = refOf(acct, object.id); const plan = kind.planDownload(object, local ?? null, acct); if (plan.effect === 'none' || !hasWrites(plan.ops.ops)) { + if (local) this.onEcho(kind, acct, object); if (!local || kind.meta(local).dirty) return null; const heal = kind.planBaselineHeal(local); return heal && hasWrites(heal.ops) ? { group: heal } : null; @@ -595,6 +614,7 @@ export abstract class ItemSync { const group = local ? plan.ops : prependOps(plan.ops, [insertGuard(kind, ref)]); return { group, + state: this.listing(kind, acct, object.id, true), applied: () => { if (plan.effect === 'insert') this.env.report.stats.downloaded.created++; else this.env.report.stats.downloaded.updated++; @@ -629,6 +649,7 @@ export abstract class ItemSync { const plan = kind.planDownload(object, null, acct); return { group: concatGroups(ref, found.kind.planLocalDelete(found.local), plan.ops), + state: both(this.listing(found.kind, acct, object.id, false), this.listing(kind, acct, object.id, true)), applied: () => { this.env.report.stats.downloaded.updated++; this.markResolved(acct, object.id); @@ -657,6 +678,7 @@ export abstract class ItemSync { const conflict = reason === 'destroyed' && meta.dirty && !meta.deleted; return { group: found.kind.planLocalDelete(found.local), + state: this.listing(found.kind, acct, id, false), applied: () => { this.env.report.stats.downloaded.deleted++; if (conflict) this.env.report.conflicts++; @@ -952,6 +974,7 @@ export abstract class ItemSync { const id = idInAccount(meta.sourceId, acct); return { group: held.kind.planLocalDelete(held.local), + state: id ? this.listing(held.kind, acct, id, false) : undefined, applied: () => { if (uploaded) this.env.report.stats.uploaded.deleted++; if (id) this.onRemoved(held.kind, acct, id); @@ -1280,10 +1303,12 @@ export abstract class ItemSync { this.markStale(acct, server.id); this.env.report.itemError({ ref: refOf(acct, server.id), side: 'upload', type: reason, description: message }); }; + const state = this.listing(kind, acct, server.id, true); return { group: withExtra(plan.ops), + state, applied: counted, - replan: async () => ({ group: withExtra(plan.keepDirtyOps), applied: counted, failed }), + replan: async () => ({ group: withExtra(plan.keepDirtyOps), state, applied: counted, failed }), failed, }; } @@ -1310,11 +1335,19 @@ export abstract class ItemSync { const object = id ? (await this.env.jmap.get(this.itemType, acct, [id], this.itemProperties)).list[0] : undefined; const objectKind = object ? this.kindOf(object) : null; if (!object || !objectKind || !this.inSelection(acct, object)) { - return [{ group: remove, applied: () => (id ? this.onRemoved(kind, acct, id) : undefined) }]; + return [{ + group: remove, + state: id ? this.listing(kind, acct, id, false) : undefined, + applied: () => (id ? this.onRemoved(kind, acct, id) : undefined), + }]; } await this.beforePlanning(acct, [object]); const insert = objectKind.planDownload(object, null, acct); - return [{ group: concatGroups(ref, remove, insert.ops), applied: () => this.onDownloaded(objectKind, acct, object) }]; + return [{ + group: concatGroups(ref, remove, insert.ops), + state: both(this.listing(kind, acct, object.id, false), this.listing(objectKind, acct, object.id, true)), + applied: () => this.onDownloaded(objectKind, acct, object), + }]; } /** diff --git a/src/device-sync/engine/run.ts b/src/device-sync/engine/run.ts index d037c332..39c06178 100644 --- a/src/device-sync/engine/run.ts +++ b/src/device-sync/engine/run.ts @@ -124,11 +124,11 @@ export async function createEnv(options: EnvOptions): Promise<{ env: RunEnv; syn authority, extras: options.extras, reader: options.reader, - writer: new BatchWriter(options.reader.port, { - maxOps: tuning.maxBatchOps, - maxBytes: tuning.maxBatchBytes, - maxReplans: tuning.maxReplans, - }), + writer: new BatchWriter( + options.reader.port, + { maxOps: tuning.maxBatchOps, maxBytes: tuning.maxBatchBytes, maxReplans: tuning.maxReplans }, + store, + ), report: options.report, checkpoints, store, diff --git a/src/device-sync/engine/sync-state.ts b/src/device-sync/engine/sync-state.ts index f9df3106..46e1e73a 100644 --- a/src/device-sync/engine/sync-state.ts +++ b/src/device-sync/engine/sync-state.ts @@ -132,14 +132,20 @@ export interface Tail { applied(): void; } +/** A change of the SyncState that has to be stored together with the rows it describes. */ +export type StateChange = (next: SyncState) => void; + /** * The committed SyncState of a run. Changes are proposed as a tail: the * batch writer calls `op()` right before sending the last batch (so the * change sees what earlier batches decided, e.g. items put on `stale`), and - * the state is only adopted after that batch applied. + * the state is only adopted after that batch applied. Changes that describe + * the rows of one batch (contacts: the groups present) are staged by the + * batch writer while it sends that batch, so its state op carries them. */ export class StateStore { private state: SyncState; + private readonly staged = new Map(); constructor( initial: SyncState, @@ -157,7 +163,16 @@ export class StateStore { return this.state.accounts[jmapAccountId] ?? emptyAccountState(); } - /** A tail applying `mutate` to a copy of the committed state (plus the owner). */ + /** Every state op built until `unstage(key)` applies `change` too, after its own changes. */ + stage(key: object, change: StateChange): void { + this.staged.set(key, change); + } + + unstage(key: object): void { + this.staged.delete(key); + } + + /** A tail applying `mutate` (then the staged changes) to a copy of the committed state (plus the owner). */ tail(mutate: (next: SyncState) => void): Tail { let next: SyncState | null = null; return { @@ -165,6 +180,7 @@ export class StateStore { next = clone(this.state); if (this.owner) next.owner = { ...this.owner }; mutate(next); + for (const change of this.staged.values()) change(next); return { op: 'syncState', value: serializeSyncState(next) }; }, applied: () => {