diff --git a/src/device-sync/__tests__/calendar/big-events.test.ts b/src/device-sync/__tests__/calendar/big-events.test.ts index d367ab0f..0b1942ba 100644 --- a/src/device-sync/__tests__/calendar/big-events.test.ts +++ b/src/device-sync/__tests__/calendar/big-events.test.ts @@ -4,17 +4,23 @@ import { describe, expect, it } from 'vitest'; import { Attendees, Events } from '../../android-columns'; import { calendarPlanner } from '../../calendar/planner'; +import { contactsPlanner } from '../../contacts/planner'; +import { estimateBatchBytes } from '../../engine/batch'; import type { OpGroup } from '../../planner'; +import { CALENDAR_AUTHORITY, type ProviderPort } from '../../types'; import type { CalendarEventWire } from '../../wire'; +import { createHarness, rowWrites, type Harness as EngineHarness } from '../engine/harness'; import { Harness, ME, serverApply } from './harness'; const TRANSACTION = 1_000_000; const estimate = (group: OpGroup) => 2 * JSON.stringify(group.ops).length + 300 * group.ops.length; +/** A plan's groups: the group and the ones applied after it. */ +const chain = (group: OpGroup): OpGroup[] => [group, ...(group.next ?? [])]; /** A daily meeting with `attendees` guests whose `overrides` occurrences each carry the whole participants map. */ -function meeting(attendees: number, overrides: number): CalendarEventWire { +function meeting(attendees: number, overrides: number, me = ME, calendarId = 'b'): CalendarEventWire { const participants: Record = { - me: { '@type': 'Participant', calendarAddress: `mailto:${ME}`, roles: { owner: true, attendee: true }, participationStatus: 'accepted' }, + me: { '@type': 'Participant', calendarAddress: `mailto:${me}`, roles: { owner: true, attendee: true }, participationStatus: 'accepted' }, }; for (let i = 0; i < attendees; i++) { participants[`p${i}`] = { '@type': 'Participant', calendarAddress: `mailto:person${i}@example.com`, name: `Person ${i}`, roles: { attendee: true }, participationStatus: 'needs-action', expectReply: true }; @@ -28,19 +34,54 @@ function meeting(attendees: number, overrides: number): CalendarEventWire { id: 'big', '@type': 'Event', uid: 'big@example.org', - calendarIds: { b: true }, + calendarIds: { [calendarId]: true }, title: 'Standup', description: 'Daily standup', start: '2026-10-05T09:00:00', timeZone: 'Europe/Berlin', duration: 'PT15M', recurrenceRule: { frequency: 'daily', count: 300 }, - organizerCalendarAddress: `mailto:${ME}`, + organizerCalendarAddress: `mailto:${me}`, participants, recurrenceOverrides, } as unknown as CalendarEventWire; } +/** The engine with the real planners, its provider refusing a batch bigger than one Binder transaction (as Android does). */ +function engineWithBinderLimit(): EngineHarness { + const h = createHarness(); + h.deps.planners = { contacts: contactsPlanner, calendar: calendarPlanner }; + const base = h.deps.provider; + h.deps.provider = (name, authority): ProviderPort => { + const port = base(name, authority); + return { + accountName: port.accountName, + authority: port.authority, + query: (q) => port.query(q), + readSyncState: () => port.readSyncState(), + readPhoto: (id, px) => port.readPhoto(id, px), + applyBatch: async (ops) => + estimateBatchBytes(ops) > TRANSACTION ? { ok: false, reason: 'tooLarge', message: 'TransactionTooLargeException' } : port.applyBatch(ops), + }; + }; + return h; +} + +/** The big meeting's rows on the engine's device: one master, one exception row per override, every attendee once. */ +function expectWholeMeeting(h: EngineHarness, id: string, attendees: number, overrides: number, context = ''): void { + const rows = h.events(); + const masters = rows.filter((e) => e[Events._SYNC_ID] === `a/${id}`); + expect(masters, context).toHaveLength(1); + const exceptions = rows.filter((e) => e[Events.ORIGINAL_SYNC_ID] === `a/${id}`); + expect(exceptions, context).toHaveLength(overrides); + expect(new Set(exceptions.map((x) => x[Events._SYNC_ID])).size, context).toBe(overrides); + expect(exceptions.every((x) => Number(x[Events.ORIGINAL_ID]) === Number(masters[0]._id)), context).toBe(true); + const attendeesOf = (eventId: unknown) => h.device.rows('attendees').filter((a) => Number(a.event_id) === Number(eventId)).length; + expect([masters[0], ...exceptions].every((row) => attendeesOf(row._id) === attendees + 1), context).toBe(true); + expect(Object.keys(JSON.parse(String(masters[0][Events.SYNC_DATA1])).recurrenceOverrides), context).toHaveLength(overrides); + expect(rows.filter((e) => e[Events.ORIGINAL_SYNC_ID] === null), context).toHaveLength(1); +} + async function synced(event: CalendarEventWire) { const h = await new Harness().setup(); await h.download(event); @@ -87,4 +128,113 @@ describe('calendar planner: big meetings', () => { expect(await h.tryApply(accepted.ops)).toMatchObject({ ok: false, reason: 'assert' }); expect(h.fake.rows('attendees', `${Attendees.EVENT_ID} = ? AND ${Attendees.ATTENDEE_EMAIL} = ?`, [x.eventId, 'person1@example.com'])[0][Attendees.ATTENDEE_STATUS]).toBe(1); }); + + it.each([ + [25, 52], + [100, 12], + ])('writes the first download of %i attendees on %i overrides in groups that each fit one transaction', async (attendees, overrides) => { + const h = await new Harness().setup(); + const event = meeting(attendees, overrides); + + const plan = calendarPlanner.planDownload(event, null, h.ctx); + + expect(plan.effect).toBe('insert'); + expect(chain(plan.ops).length).toBeGreaterThan(1); + for (const group of chain(plan.ops)) expect(estimate(group)).toBeLessThan(TRANSACTION); + await h.apply(plan.ops); + const local = (await h.local('big'))!; + expect(local.shadow).toEqual(event); + expect(local.exceptions.map((x) => x.recurrenceId).sort()).toEqual(Object.keys(event.recurrenceOverrides!).sort()); + expect(local.exceptions.every((x) => x.attendees.length === attendees + 1 && !x.dirty)).toBe(true); + // Its echo writes nothing. + expect(calendarPlanner.planDownload(event, local, h.ctx).effect).toBe('none'); + }); + + it('finishes a first download cut off between its groups, without writing a row twice', async () => { + const event = meeting(25, 52); + const groups = chain(calendarPlanner.planDownload(event, null, (await new Harness().setup()).ctx).ops).length; + for (let cut = 1; cut < groups; cut++) { + const h = await new Harness().setup(); + for (const group of chain(calendarPlanner.planDownload(event, null, h.ctx).ops).slice(0, cut)) await h.apply({ ref: group.ref, ops: group.ops }); + + // Planned again from what the first groups wrote. + const again = calendarPlanner.planDownload(event, await h.local('big'), h.ctx); + for (const group of chain(again.ops)) expect(estimate(group)).toBeLessThan(TRANSACTION); + await h.apply(again.ops); + + const local = (await h.local('big'))!; + expect(local.shadow, `cut after ${cut}`).toEqual(event); + expect(new Set(local.exceptions.map((x) => x.recurrenceId)).size, `cut after ${cut}`).toBe(52); + expect(local.exceptions, `cut after ${cut}`).toHaveLength(52); + expect(local.exceptions.every((x) => x.attendees.length === 26), `cut after ${cut}`).toBe(true); + expect(calendarPlanner.planDownload(event, local, h.ctx).effect, `cut after ${cut}`).toBe('none'); + } + // Every cut point writes a 2 MB meeting again. + }, 60_000); + + it('writes a change of every occurrence in groups, the series with its new shadow last', async () => { + const h = await new Harness().setup(); + const event = meeting(25, 52); + await h.download(event); + // Another client adds a guest to the meeting and to every occurrence. + const guest = { '@type': 'Participant', calendarAddress: 'mailto:guest@example.com', name: 'Guest', roles: { attendee: true }, participationStatus: 'needs-action' }; + const changed = JSON.parse(JSON.stringify(event)) as CalendarEventWire; + (changed.participants as Record).guest = guest; + for (const override of Object.values(changed.recurrenceOverrides!)) (override as { participants: Record }).participants.guest = guest; + + const plan = calendarPlanner.planDownload(changed, await h.local('big'), h.ctx); + + const groups = chain(plan.ops); + expect(groups.length).toBeGreaterThan(1); + for (const group of groups) expect(estimate(group)).toBeLessThan(TRANSACTION); + const writesShadow = (group: OpGroup) => group.ops.some((op) => op.op === 'update' && Events.SYNC_DATA1 in op.values); + expect(groups.map(writesShadow)).toEqual(groups.map((_, i) => i === groups.length - 1)); + // Cut off after the first group: planned again from what it wrote, the rest follows. + await h.apply({ ref: groups[0].ref, ops: groups[0].ops }); + await h.apply(calendarPlanner.planDownload(changed, await h.local('big'), h.ctx).ops); + const local = (await h.local('big'))!; + expect(local.shadow).toEqual(changed); + expect(local.exceptions.every((x) => x.attendees.length === 27)).toBe(true); + expect(calendarPlanner.planDownload(changed, local, h.ctx).effect).toBe('none'); + }); +}); + +describe('device sync engine: big meetings', () => { + it.each([ + [25, 52], + [100, 12], + ])('downloads a meeting of %i attendees on %i overrides in several transactions', async (attendees, overrides) => { + const h = engineWithBinderLimit(); + const { id: _drop, ...event } = meeting(attendees, overrides, 'alice@example.com', h.calendar) as unknown as Record; + const id = h.server.addEvent('a', event); + + const report = await h.run(CALENDAR_AUTHORITY); + + expect(report).toMatchObject({ outcome: 'ok', itemErrors: [], stats: { downloaded: { created: 1 } } }); + expectWholeMeeting(h, id, attendees, overrides); + const before = h.batches.log.length; + expect((await h.run(CALENDAR_AUTHORITY)).outcome).toBe('ok'); + expect(rowWrites(h.batches.log.slice(before))).toEqual([]); + }); + + it('finishes a big meeting cut off between its groups by a crash, without duplicating rows', async () => { + const setup = () => { + const h = engineWithBinderLimit(); + const { id: _drop, ...event } = meeting(25, 52, 'alice@example.com', h.calendar) as unknown as Record; + return { h, id: h.server.addEvent('a', event) }; + }; + const probe = setup(); + await probe.h.run(CALENDAR_AUTHORITY); + const batches = probe.h.batches.applied; + + for (let n = 1; n < batches; n++) { + const { h, id } = setup(); + h.batches.crashAfter = n; + await h.run(CALENDAR_AUTHORITY); + h.batches.crashAfter = null; + expect((await h.run(CALENDAR_AUTHORITY)).outcome, `crash after batch ${n}`).toBe('ok'); + expectWholeMeeting(h, id, 25, 52, `crash after batch ${n}`); + } + // Two runs of a 2 MB meeting per crash point. + }, 60_000); }); diff --git a/src/device-sync/__tests__/calendar/harness.ts b/src/device-sync/__tests__/calendar/harness.ts index e8f5e7af..1ec0873c 100644 --- a/src/device-sync/__tests__/calendar/harness.ts +++ b/src/device-sync/__tests__/calendar/harness.ts @@ -86,9 +86,12 @@ export class Harness { return this; } + /** Applies a group and the groups after it (`next`), each in a batch of its own, as the engine does. */ async apply(group: OpGroup): Promise { - const result = await this.port.applyBatch(group.ops); - if (!result.ok) throw new Error(`batch ${group.ref} failed: ${result.reason} ${result.message}`); + for (const g of [group, ...(group.next ?? [])]) { + const result = await this.port.applyBatch(g.ops); + if (!result.ok) throw new Error(`batch ${group.ref} failed: ${result.reason} ${result.message}`); + } } async tryApply(group: OpGroup) { diff --git a/src/device-sync/__tests__/engine/internals.test.ts b/src/device-sync/__tests__/engine/internals.test.ts index 974f9c9b..33323444 100644 --- a/src/device-sync/__tests__/engine/internals.test.ts +++ b/src/device-sync/__tests__/engine/internals.test.ts @@ -201,6 +201,57 @@ describe('provider batches', () => { await writer(p).write([{ group: { ref: 'g', ops: [{ op: 'assert', table: 'raw_contacts', id: 1, values: { dirty: 0 } }] } }]); expect(p.sent).toEqual([]); }); + + describe('an item too big for one group (a chain)', () => { + const named = (n: string) => insert('data', { n }); + const namesOf = (batches: ProviderOp[][]) => batches.map((b) => b.map((op) => (op.op === 'insert' ? op.values.n : op.op))); + const chained = (extra: Partial = {}): Work => ({ + group: { ref: 'big', ops: [named('1')], next: [{ ref: 'big', ops: [named('2')] }, { ref: 'big', ops: [named('3')] }] }, + ...extra, + }); + + it('applies its groups in order, each in a batch of its own, between the other works, and counts it after the last', async () => { + const p = port(ok); + const applied: string[] = []; + const store = new StateStore(emptySyncState(), { registryId: 'r', origin: 'o' }); + const big = chained({ + applied: (results) => void applied.push(`big:${results.length}`), + state: (next) => { + accountOf(next, 'a').stale = ['big']; + }, + }); + const other = (n: string): Work => ({ group: { ref: n, ops: [named(n)] }, applied: () => void applied.push(n) }); + + await new BatchWriter(p, { maxOps: 400, maxBytes: 300_000, maxReplans: 3 }, store).write( + [other('a'), big, other('b')], + store.tail(() => undefined), + ); + + expect(namesOf(p.sent)).toEqual([['a'], ['1'], ['2'], ['3', 'syncState'], ['b', 'syncState']]); + expect(applied).toEqual(['a', 'big:3', 'b']); + expect(store.committed.accounts.a.stale).toEqual(['big']); + }); + + it('plans the item again from what its first groups wrote when a later one fails, never sending that one again', async () => { + let failures = 1; + const p = port((ops) => (namesOf([ops])[0].includes('2') && failures-- > 0 ? { ok: false, reason: 'assert', message: 'changed' } : ok(ops))); + const rest: Work = { group: { ref: 'big', ops: [named('rest')] } }; + + await writer(p).write([chained({ replan: async () => rest })]); + + expect(namesOf(p.sent)).toEqual([['1'], ['2'], ['rest']]); + }); + + it('gives the item up when one of its groups is refused for good', async () => { + const p = port((ops) => (namesOf([ops])[0].includes('2') ? { ok: false, reason: 'tooLarge', message: 'TransactionTooLargeException' } : ok(ops))); + const failed = vi.fn(); + + await writer(p).write([chained({ failed, replan: async () => null })]); + + expect(namesOf(p.sent)).toEqual([['1'], ['2']]); + expect(failed).toHaveBeenCalledWith('tooLarge', 'TransactionTooLargeException'); + }); + }); }); describe('SyncState', () => { diff --git a/src/device-sync/calendar/merge.ts b/src/device-sync/calendar/merge.ts index eadd6b71..f03a9834 100644 --- a/src/device-sync/calendar/merge.ts +++ b/src/device-sync/calendar/merge.ts @@ -19,7 +19,7 @@ * edit of them) and the number of exception rows. */ import { Events } from '../android-columns'; -import type { CalendarContext, EventBaseline, LocalEvent, LocalEventRow, LocalException } from '../planner'; +import type { CalendarContext, EventBaseline, LocalEvent, LocalEventRow, LocalException, OpGroup } from '../planner'; import type { Row } from '../types'; import type { CalendarEventWire } from '../wire'; import { collectionKey, exceptionRef, objectRef } from '../common/ids'; @@ -31,6 +31,8 @@ import { assertExceptionCount, assertRow, deleteRow, + eventInsert, + eventWrite, insertRow, updateRow, type RowState, @@ -240,7 +242,8 @@ export type WriteMode = | 'overwrite'; export interface EventWriteResult { - group: GroupBuilder; + /** One group, or a chain for an event too big for one (see `eventWrite`). */ + ops: OpGroup; conflicts: number; stillDirty: boolean; } @@ -277,7 +280,6 @@ export function writeEvent( options: EventWriteOptions, ): EventWriteResult { const { ctx } = options; - const group = new GroupBuilder(options.ref); const overwrite = options.mode === 'overwrite'; const masterRef = objectRef(ctx.jmapAccountId, event.id); const uid = typeof event.uid === 'string' && event.uid ? event.uid : null; @@ -288,9 +290,14 @@ export function writeEvent( // Asserts first: the rows as read. Attendees and reminders only for dirty rows (see `assertRow`): a big // meeting's occurrences would otherwise assert every attendee and outgrow one provider transaction. - assertRow(group, local, false, local.dirty || local.deleted); - for (const x of local.exceptions) assertRow(group, x, true, x.dirty || x.deleted); - assertExceptionCount(group, local.eventId, linkedExceptionCount(local)); + const opsOf = (build: (group: GroupBuilder) => void) => { + const group = new GroupBuilder(options.ref); + build(group); + return group.build().ops; + }; + const masterAsserts = opsOf((g) => assertRow(g, local, false, local.dirty || local.deleted)); + const rowAsserts = new Map(local.exceptions.map((x) => [x.eventId, opsOf((g) => assertRow(g, x, true, x.dirty || x.deleted))])); + const countAssert = opsOf((g) => assertExceptionCount(g, local.eventId, linkedExceptionCount(local))); let conflicts = 0; let stillDirty = false; @@ -412,12 +419,17 @@ export function writeEvent( updates.push({ row, state: exceptionState(key, merged, left, row, !!r) }); } - for (const row of deletes) deleteRow(group, row.eventId); - updateRow(group, local, masterState); - for (const { row, state } of updates) updateRow(group, row, state); - for (const state of inserts) insertRow(group, state, { id: local.eventId }); + const ops = eventWrite(options.ref, { + master: masterAsserts, + rows: rowAsserts, + count: countAssert, + deletes: deletes.map((row) => ({ row: row.eventId, asserts: [], writes: opsOf((g) => deleteRow(g, row.eventId)) })), + masterWrite: opsOf((g) => updateRow(g, local, masterState)), + updates: updates.map(({ row, state }) => ({ row: row.eventId, asserts: [], writes: opsOf((g) => updateRow(g, row, state)) })), + inserts: inserts.map((state) => ({ asserts: [], writes: opsOf((g) => insertRow(g, state, { id: local.eventId })) })), + }); - return { group, conflicts, stillDirty: !overwrite && (stillDirty || local.deleted) }; + return { ops, conflicts, stillDirty: !overwrite && (stillDirty || local.deleted) }; } /** The row an exception image becomes, written clean. */ @@ -438,12 +450,8 @@ export function imageState(x: ExceptionImage, calendarRowId: number, masterRef: }; } -/** A new event's rows: the master and one row per exception, attendees and reminders included. */ -export function insertEvent(event: CalendarEventWire, image: EventImage, ctx: CalendarContext, ref: string): GroupBuilder { - const group = new GroupBuilder(ref); - const masterRef = objectRef(ctx.jmapAccountId, event.id); - const uid = typeof event.uid === 'string' && event.uid ? event.uid : null; - const master = insertRow(group, { +function masterInsertState(event: CalendarEventWire, image: EventImage, masterRef: string, uid: string | null): RowState { + return { isException: false, cells: image.master.cells, attendees: image.master.attendees, @@ -453,10 +461,39 @@ export function insertEvent(event: CalendarEventWire, image: EventImage, ctx: Ca uid, shadow: event, hasParticipants: image.master.hasParticipants, - }); + }; +} + +/** A new event's rows: the master and one row per exception, attendees and reminders included. */ +export function insertEvent(event: CalendarEventWire, image: EventImage, ctx: CalendarContext, ref: string): GroupBuilder { + const group = new GroupBuilder(ref); + const masterRef = objectRef(ctx.jmapAccountId, event.id); + const uid = typeof event.uid === 'string' && event.uid ? event.uid : null; + const master = insertRow(group, masterInsertState(event, image, masterRef, uid)); for (const key of [...image.exceptions.keys()].sort()) { const x = image.exceptions.get(key)!; insertRow(group, imageState(x, Number(image.master.cells[Events.CALENDAR_ID]), masterRef, uid), { ref: master }); } return group; } + +/** A new event's rows as one group, or as a chain for an event too big for one (see `eventInsert`). */ +export function insertEventOps(event: CalendarEventWire, image: EventImage, ctx: CalendarContext, ref: string): OpGroup { + const masterRef = objectRef(ctx.jmapAccountId, event.id); + const uid = typeof event.uid === 'string' && event.uid ? event.uid : null; + const calendarRowId = Number(image.master.cells[Events.CALENDAR_ID]); + const opsOf = (build: (group: GroupBuilder) => void) => { + const group = new GroupBuilder(ref); + build(group); + return group.build().ops; + }; + return eventInsert(ref, insertEvent(event, image, ctx, ref).build(), () => ({ + masterRef, + master: opsOf((g) => insertRow(g, { ...masterInsertState(event, image, masterRef, uid), shadow: undefined })), + exceptions: [...image.exceptions.keys()].sort().map((key) => ({ + asserts: [], + writes: opsOf((g) => insertRow(g, imageState(image.exceptions.get(key)!, calendarRowId, masterRef, uid))), + })), + shadow: event, + })); +} diff --git a/src/device-sync/calendar/planner.ts b/src/device-sync/calendar/planner.ts index 5db04219..4ee3b26b 100644 --- a/src/device-sync/calendar/planner.ts +++ b/src/device-sync/calendar/planner.ts @@ -31,7 +31,7 @@ import { claimOps, claimUid, claimedUid, computeCreate, createTarget, deleteActi import { decodeCalendar, decodeEvents, isExceptionRow, isNewMaster } from './decode'; import { isExcluded, overridesOf } from './exceptions'; import { eventImage } from './image'; -import { insertEvent, isRemovedInstance, linkedExceptionCount, pickCalendar, writeEvent, type CalendarTarget } from './merge'; +import { insertEvent, insertEventOps, isRemovedInstance, linkedExceptionCount, pickCalendar, writeEvent, type CalendarTarget } from './merge'; import { pairKind, pairOps, pairPatch } from './pairs'; import { GroupBuilder, assertExceptionCount, assertRow, deleteRow, writesOf } from './rows'; import { eventZone } from './timing'; @@ -74,7 +74,7 @@ function planDownload(event: CalendarEventWire, local: LocalEvent | null, ctx: C if (!image) return none(!!local && (local.dirty || local.deleted)); if (!local) { - const ops = insertEvent(event, image, ctx, ref).build(); + const ops = insertEventOps(event, image, ctx, ref); return { ops, conflicts: 0, stillDirty: false, effect: 'insert', writes: writesOf(ops) }; } // A split clone only looks like this object: it is a new event and uploads as one. The source row is the object's. @@ -118,7 +118,7 @@ function planDownload(event: CalendarEventWire, local: LocalEvent | null, ctx: C ...(local.shadow ? {} : { pending: null }), clearPoison: !itemDirty, }); - const ops = result.group.build(); + const ops = result.ops; const writes = writesOf(ops); return { ops: writes ? ops : { ref, ops: [] }, @@ -222,7 +222,7 @@ function revertGroup(local: LocalEvent, ctx: CalendarContext): OpGroup | null { if (!target) return null; const image = eventImage(shadow, target.calendarRowId, target.calendarId, ctx); if (!image) return null; - return writeEvent(shadow, image, target, local, { mode: 'overwrite', ref: refOf(local), ctx, base: shadow }).group.build(); + return writeEvent(shadow, image, target, local, { mode: 'overwrite', ref: refOf(local), ctx, base: shadow }).ops; } /** Clears DIRTY where nothing uploadable changed, taking the rows as they are as the new baselines. */ @@ -368,7 +368,7 @@ function planAccepted(local: LocalEvent, server: CalendarEventWire, ctx: Calenda base: local.shadow ?? server, pending: null, clearPoison: true, - }).group.build(); + }).ops; return { ops, keepDirtyOps }; } @@ -411,7 +411,7 @@ function planZoneChange(local: LocalEvent, previousZone: string, ctx: CalendarCo if (!target) return null; const image = eventImage(shadow, target.calendarRowId, target.calendarId, ctx); if (!image) return null; - const ops = writeEvent(shadow, image, target, local, { mode: 'overwrite', ref: refOf(local), ctx, base: shadow }).group.build(); + const ops = writeEvent(shadow, image, target, local, { mode: 'overwrite', ref: refOf(local), ctx, base: shadow }).ops; return writesOf(ops) ? ops : null; } diff --git a/src/device-sync/calendar/rows.ts b/src/device-sync/calendar/rows.ts index 8c877a6c..79111f57 100644 --- a/src/device-sync/calendar/rows.ts +++ b/src/device-sync/calendar/rows.ts @@ -14,6 +14,8 @@ import type { EventBaseline, LocalEventRow, OpGroup, PendingCreate, PoisonMarker import type { ProviderOp, Row, WriteRow } from '../types'; import type { CalendarEventWire } from '../wire'; import { deepEqual } from '../common/json'; +import { MAX_OPS_PER_YIELD, estimateBatchBytes } from '../engine/batch'; +import { DEFAULT_TUNING } from '../engine/deps'; import { ATTENDEE_CELLS, MASTER_CELLS, EXCEPTION_CELLS, REMINDER_CELLS, encodeBaseline, sameCell } from './columns'; import { rowReminder } from './reminders'; @@ -245,7 +247,144 @@ function writeReminders(group: GroupBuilder, current: LocalEventRow, wanted: Row for (const add of inserts) group.write({ op: 'insert', table: 'reminders', values: { ...add, [Reminders.EVENT_ID]: current.eventId } }); } -/** Counts the provider writes of a group (asserts are not writes). */ +/** Counts the provider writes of a group and the groups after it (asserts are not writes). */ export function writesOf(group: OpGroup): number { - return group.ops.filter((op) => op.op === 'insert' || op.op === 'update' || op.op === 'delete').length; + return [group, ...(group.next ?? [])].reduce( + (n, g) => n + g.ops.filter((op) => op.op === 'insert' || op.op === 'update' || op.op === 'delete').length, + 0, + ); +} + +// ─── Items too big for one group ──────────────────────── + +/** + * What one group may hold: the engine's batch budget as `estimateBatchBytes` + * weighs it, and one op less than a provider yield window (room for the + * engine's insert guard). A bigger event is written as a chain of groups + * (`OpGroup.next`); a master's shadow is one column and may take a group of + * its own beyond the budget (under one transaction for any event that syncs). + */ +const GROUP_BYTES = DEFAULT_TUNING.maxBatchBytes; +const GROUP_OPS = MAX_OPS_PER_YIELD - 1; + +const weight = (ops: readonly ProviderOp[]) => (ops.length ? estimateBatchBytes(ops) : 0); + +/** One group from asserts and write blocks whose `refs` index into their own block, asserts first. */ +function assemble(ref: string, asserts: readonly ProviderOp[], blocks: ReadonlyArray): OpGroup { + const ops: ProviderOp[] = [...asserts]; + for (const block of blocks) { + const offset = ops.length; + for (const op of block) { + if (op.op !== 'insert' || !op.refs) ops.push(op); + else ops.push({ ...op, refs: Object.fromEntries(Object.entries(op.refs).map(([column, i]) => [column, i + offset])) }); + } + } + return { ref, ops }; +} + +/** Writes of one row (`refs` index into them), and the asserts that guard that row (none for a new row). */ +export interface RowWrite { + asserts: ProviderOp[]; + writes: ProviderOp[]; +} + +/** + * Cuts row writes into groups of the budget, in order, a row never split: + * each group starts with `guard`, the first also with `first`. + */ +function cutRows(ref: string, rows: readonly RowWrite[], guard: readonly ProviderOp[], first: readonly ProviderOp[] = []): OpGroup[] { + const groups: OpGroup[] = []; + let asserts: ProviderOp[] = [...guard, ...first]; + let blocks: ProviderOp[][] = []; + let ops = asserts.length; + let bytes = weight(asserts); + for (const row of rows) { + const n = row.asserts.length + row.writes.length; + const b = weight(row.asserts) + weight(row.writes); + if (blocks.length && (ops + n > GROUP_OPS || bytes + b > GROUP_BYTES)) { + groups.push(assemble(ref, asserts, blocks)); + asserts = [...guard]; + blocks = []; + ops = asserts.length; + bytes = weight(asserts); + } + asserts.push(...row.asserts); + blocks.push(row.writes); + ops += n; + bytes += b; + } + if (blocks.length) groups.push(assemble(ref, asserts, blocks)); + return groups; +} + +function chained(groups: readonly OpGroup[]): OpGroup { + const [first, ...next] = groups; + return next.length ? { ...first, next } : first; +} + +/** + * The writes into an existing event as one group: asserts first (`master`, + * the exception rows' `rows` in order, `count`), then the writes (`deletes`, + * the master's `master`, `updates`, `inserts`). When that is more than a group + * may hold, a chain: the exception rows' writes in groups of the budget, each + * asserting the master row and the rows it writes, the first also the number + * of exception rows; then the master's write, with its new shadow, last, + * asserting the other rows. Cut off in between, the item is planned again + * from the old shadow, so no server change to a row not written yet is taken + * as done. + */ +export function eventWrite( + ref: string, + parts: { + master: ProviderOp[]; + rows: Map; + count: ProviderOp[]; + deletes: Array; + masterWrite: ProviderOp[]; + updates: Array; + inserts: RowWrite[]; + }, +): OpGroup { + const rowAsserts = [...parts.rows.values()].flat(); + const single = assemble(ref, [...parts.master, ...rowAsserts, ...parts.count], [ + ...parts.deletes.map((d) => d.writes), + parts.masterWrite, + ...parts.updates.map((u) => u.writes), + ...parts.inserts.map((i) => i.writes), + ]); + if (single.ops.length <= GROUP_OPS && estimateBatchBytes(single.ops) <= GROUP_BYTES) return single; + const writing = [...parts.deletes, ...parts.updates].filter((w) => w.writes.length); + if (!writing.length && !parts.inserts.length) return single; + const written = new Set(writing.map((w) => w.row)); + const rows: RowWrite[] = [ + ...writing.map((w) => ({ asserts: parts.rows.get(w.row) ?? [], writes: w.writes })), + ...parts.inserts, + ]; + // The master row's own assert guards every group; its attendees and reminders go with its write. + const groups = cutRows(ref, rows, parts.master.slice(0, 1), parts.count); + const rest = [...parts.rows].filter(([row]) => !written.has(row)).flatMap(([, asserts]) => asserts); + groups.push(assemble(ref, [...parts.master, ...rest], [parts.masterWrite])); + return chained(groups); +} + +/** + * A new event's rows as one group, or, when they are more than a group may + * hold, as a chain: `master` (the master's insert with its identity but + * without its shadow), its exception rows in groups of the budget, each + * asserting the master by `masterRef` and linked to it by ORIGINAL_SYNC_ID + * (the provider fills ORIGINAL_ID from it and CALENDAR_ID), then `shadow`. + * Cut off in between, the next download finds the master by its identity and + * writes what is missing; a master without a shadow merges like an adopted one. + */ +export function eventInsert( + ref: string, + single: OpGroup, + partsOf: () => { masterRef: string; master: ProviderOp[]; exceptions: RowWrite[]; shadow: CalendarEventWire }, +): OpGroup { + if (single.ops.length <= GROUP_OPS && estimateBatchBytes(single.ops) <= GROUP_BYTES) return single; + const parts = partsOf(); + const byIdentity = { table: 'events' as const, where: `${Events._SYNC_ID} = ?`, args: [parts.masterRef], expectCount: 1 }; + const groups = [assemble(ref, [], [parts.master]), ...cutRows(ref, parts.exceptions, [{ op: 'assert', ...byIdentity }])]; + groups.push({ ref, ops: [{ op: 'update', ...byIdentity, values: { [Events.SYNC_DATA1]: JSON.stringify(parts.shadow) } }] }); + return chained(groups); } diff --git a/src/device-sync/engine/batch.ts b/src/device-sync/engine/batch.ts index 334a4419..686d7aea 100644 --- a/src/device-sync/engine/batch.ts +++ b/src/device-sync/engine/batch.ts @@ -23,6 +23,12 @@ * 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. + * + * An item too big for one transaction comes as a chain (`OpGroup.next`): its + * groups go in order, each in a batch of its own, between the batches of the + * works before and after it; its `state` and `applied` go with the last. A + * group of the chain that fails leaves the earlier ones written: after an + * assert the item is read and planned again from them, else it is given up. */ import type { OpGroup } from '../planner'; import type { BatchFailure, OpResult, ProviderOp, ProviderPort } from '../types'; @@ -67,6 +73,17 @@ export function estimateBatchBytes(ops: readonly ProviderOp[]): number { return 2 * JSON.stringify(ops).length + 300 * ops.length; } +/** A group and the groups of its item that follow it (`next`), each without `next`. */ +export function chainOf(group: OpGroup): OpGroup[] { + const { next, ...first } = group; + return [first, ...(next ?? [])]; +} + +/** Whether a group, or one of the groups after it, writes anything. */ +export function chainHasWrites(group: OpGroup): boolean { + return chainOf(group).some((g) => hasWrites(g.ops)); +} + function carriesBlob(ops: readonly ProviderOp[]): boolean { return ops.some( (op) => @@ -100,19 +117,30 @@ export function buildBatch(groups: readonly OpGroup[], tailOp?: ProviderOp): Pro return out; } -/** Prepends ops to a group (e.g. an engine-side assert), shifting the group's refs. */ +/** Prepends ops to a group (e.g. an engine-side assert), shifting the group's refs; the groups after it stay. */ export function prependOps(group: OpGroup, ops: ProviderOp[]): OpGroup { - return { ref: group.ref, ops: [...ops, ...placeGroup(group.ops, ops.length).map(stripYield)] }; + return { ...group, ops: [...ops, ...placeGroup(group.ops, ops.length).map(stripYield)] }; } -/** One group running `first` then `second` atomically. */ +/** One group running `first` then `second` atomically (then the groups after `second`). */ export function concatGroups(ref: string, first: OpGroup, second: OpGroup): OpGroup { return { ref, ops: [...first.ops.map(stripYield), ...placeGroup(second.ops, first.ops.length).map(stripYield)], + ...(second.next ? { next: second.next } : {}), }; } +/** Appends ops to the last group of a chain (e.g. clearing a poison marker once the item is written). */ +export function appendToLast(group: OpGroup, ops: ProviderOp[]): OpGroup { + if (!ops.length) return group; + if (!group.next?.length) return { ...group, ops: [...group.ops, ...ops] }; + const next = [...group.next]; + const last = next[next.length - 1]; + next[next.length - 1] = { ...last, ops: [...last.ops, ...ops] }; + return { ...group, next }; +} + function stripYield(op: ProviderOp): ProviderOp { const { yieldAllowed: _y, ...rest } = op as ProviderOp & { yieldAllowed?: boolean }; return rest as ProviderOp; @@ -134,7 +162,22 @@ export class BatchWriter { * every group of `works` was applied or given up. */ async write(works: readonly Work[], tail?: Tail): Promise { - const todo = works.filter((w) => hasWrites(w.group.ops)); + let run: Work[] = []; + for (const work of works) { + if (!chainHasWrites(work.group)) continue; + if (!work.group.next?.length) { + run.push(work); + continue; + } + await this.writeRun(run); + run = []; + await this.applyChain(work, 0); + } + await this.writeRun(run, tail); + } + + /** Works of one group each, packed into batches. */ + private async writeRun(todo: readonly Work[], tail?: Tail): Promise { const batches = this.pack(todo); if (batches.length === 0) { if (tail) await this.applyTail(tail); @@ -243,7 +286,37 @@ export class BatchWriter { if (tail) await this.applyTail(tail); } + /** + * An item's chain of groups, each in a batch of its own and so committed + * whole or not at all; the item's state and `applied` go with the last. + */ + private async applyChain(work: Work, attempt: number): Promise { + const groups = chainOf(work.group).filter((g) => hasWrites(g.ops)); + const results: OpResult[] = []; + for (let i = 0; i < groups.length; i++) { + const last = i === groups.length - 1; + const stated = last ? this.stage([work]) : { staged: [], tail: undefined }; + let result; + try { + result = await this.send(buildBatch([groups[i]], stated.tail?.op())); + } finally { + this.unstage(stated.staged); + } + if (!result.ok) { + // The groups before it are written: planned again from them, never sent again as they are. + await this.retryAlone(work, result.reason, result.message, attempt); + return; + } + results.push(...result.results.slice(0, groups[i].ops.length)); + if (last) { + await work.applied?.(results); + stated.tail?.applied(); + } + } + } + private async applyAlone(work: Work, attempt: number): Promise { + if (work.group.next?.length) return this.applyChain(work, attempt); if (!hasWrites(work.group.ops)) return; const stated = this.stage([work]); let result; diff --git a/src/device-sync/engine/calendar-sync.ts b/src/device-sync/engine/calendar-sync.ts index 12dcb233..6fecd598 100644 --- a/src/device-sync/engine/calendar-sync.ts +++ b/src/device-sync/engine/calendar-sync.ts @@ -10,13 +10,13 @@ import type { CalendarContext, CalendarEventWire, CalendarLike, CalendarPlanner, import { CALENDAR_AUTHORITY, type ReminderOwner, type Row } from '../types'; import { CALENDAR_EVENT_PROPERTIES, CALENDAR_PROPERTIES } from '../wire'; import { calendarAddresses, type CalendarAddresses, type CapabilityAccount } from '../jmap/session'; -import type { Work } from './batch'; +import { chainHasWrites, type Work } from './batch'; import type { RunEnv } from './context'; import type { SubscriptionCalendar } from './deps'; import { ItemSync, type Pending } from './item-sync'; import { accountOfRef, idInAccount, refOf, type Held, type Kind, type ServerObject } from './kinds'; import { poisonFingerprint } from './poison'; -import { flag, hasWrites, num, str } from './provider'; +import { flag, num, str } from './provider'; import { isCollectionSelected } from './selection'; import { accountOf, type SyncState } from './sync-state'; @@ -501,7 +501,7 @@ export class CalendarSync extends ItemSync { const held: Held = { kind: this.eventKind, local: moved }; try { const work = this.acceptWork(acct, { held, meta: this.eventKind.meta(moved), fingerprint: '', actions: [] }, server, false); - if (hasWrites(work.group.ops)) return [work]; + if (chainHasWrites(work.group)) return [work]; // The pair's ops already wrote what the server holds. this.env.report.stats.uploaded.updated++; return []; diff --git a/src/device-sync/engine/item-sync.ts b/src/device-sync/engine/item-sync.ts index c5856617..ce7947f5 100644 --- a/src/device-sync/engine/item-sync.ts +++ b/src/device-sync/engine/item-sync.ts @@ -16,7 +16,7 @@ import { type SetResponse, } from '../jmap/caller'; import { isMethodError } from '../jmap/errors'; -import { concatGroups, prependOps, type Work } from './batch'; +import { appendToLast, chainHasWrites, concatGroups, prependOps, type Work } from './batch'; import type { RunEnv } from './context'; import { RunAbort, StateMismatch } from './errors'; import { @@ -687,7 +687,7 @@ export abstract class ItemSync { ): Work | null { const ref = refOf(acct, object.id); const plan = kind.planDownload(object, local ?? null, acct); - if (plan.effect === 'none' || !hasWrites(plan.ops.ops)) { + if (plan.effect === 'none' || !chainHasWrites(plan.ops)) { if (local) this.onEcho(kind, acct, object); if (!local || kind.meta(local).dirty) return null; const heal = kind.planBaselineHeal(local); @@ -1405,7 +1405,8 @@ export abstract class ItemSync { const { kind, local } = p.held; const plan = kind.planAccepted(local, server, acct); const extra = p.meta.poison ? [poisonOp(kind, p.meta.rowId, null)] : []; - const withExtra = (group: OpGroup): OpGroup => (extra.length ? { ref: group.ref, ops: [...group.ops, ...extra] } : group); + // With the item's last group: a big one comes in several (OpGroup.next). + const withExtra = (group: OpGroup): OpGroup => appendToLast(group, extra); const counted = () => { if (created) this.env.report.stats.uploaded.created++; else this.env.report.stats.uploaded.updated++; diff --git a/src/device-sync/planner.ts b/src/device-sync/planner.ts index 97de9ef4..65f8a659 100644 --- a/src/device-sync/planner.ts +++ b/src/device-sync/planner.ts @@ -25,6 +25,15 @@ export interface OpGroup { /** For logs and reports: SOURCE_ID/_SYNC_ID or `row:<_id>`. */ ref: string; ops: ProviderOp[]; + /** + * Further groups of the same item, for one too big for a provider + * transaction (a big meeting): the engine applies `[this, ...next]` in + * order, each atomically in a batch of its own, and counts the item once + * the last one is in. When one fails, the item is read and planned again + * from what the earlier ones wrote, so a planner orders them for that: a + * new row's identity first, a master's shadow last. They have no `next`. + */ + next?: OpGroup[]; } /** Why an item is left alone, with the fingerprint that lifts the back-off when the item changes. */