fix: write big meetings in several provider transactions #34

A recurring meeting with many attendees on many occurrences (25 on 52,
100 on 12) is about 2 MB of provider operations, and one Android
provider transaction carries 1 MB. Every event was written as one
group that the engine never splits, so such a meeting was refused every
run and never reached a phone that synced it for the first time.

An item too big for one group now comes as a chain of groups that the
engine applies one by one, each in a transaction of its own. A new
event is written master first, with its identity, then its occurrences
in groups that fit, then its shadow; a write into an existing event
writes the occurrences first and the series with its new shadow last.
When a run stops between two groups, the next one finds the event by
its identity and writes only what is missing, planned from the old
shadow, so no row is written twice and no server change is lost.

Checked with planner tests for both meeting sizes and every cut point,
and runs through the engine with a provider that refuses transactions
over 1 MB, a crash after every batch included.

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<string, unknown> = {
-    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<string, unknown>).guest = guest;
+    for (const override of Object.values(changed.recurrenceOverrides!)) (override as { participants: Record<string, unknown> }).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<string, unknown>;
+    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<string, unknown>;
+      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);
 });

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<void> {
-    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) {

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> = {}): 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', () => {

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,
+  }));
+}

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;
 }
 

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<readonly ProviderOp[]>): 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<number, ProviderOp[]>;
+    count: ProviderOp[];
+    deletes: Array<RowWrite & { row: number }>;
+    masterWrite: ProviderOp[];
+    updates: Array<RowWrite & { row: number }>;
+    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);
 }

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<void> {
-    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<void> {
     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<void> {
+    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<void> {
+    if (work.group.next?.length) return this.applyChain(work, attempt);
     if (!hasWrites(work.group.ops)) return;
     const stated = this.stage([work]);
     let result;

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 [];

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++;

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. */