fix: size provider batches by what the Binder transaction carries #34

The batch budget counted JSON characters, but a Parcel carries strings
as UTF-16 plus a few hundred bytes per operation, so a full batch was
about 700 KB of a 1 MB Binder buffer the whole process shares. Batches
are now weighed at two bytes per character plus 300 bytes per op and
kept under 500 KB of that. The state op no longer joins a batch whose
last item already fills a provider yield window of 499 operations,
which ContactsProvider refused.

src/device-sync/__tests__/engine/internals.test.ts

@@ -3,7 +3,7 @@
 // rebuild and how failures map to outcomes.
 
 import { describe, expect, it, vi } from 'vitest';
-import { BatchWriter, buildBatch, prependOps, type Work } from '../../engine/batch';
+import { BatchWriter, buildBatch, estimateBatchBytes, MAX_OPS_PER_YIELD, prependOps, type Work } from '../../engine/batch';
 import { classifyFailure, RunAbort, StateMismatch, StopRun } from '../../engine/errors';
 import { acquireLock } from '../../engine/mutex';
 import { isBackedOff, nextMarker } from '../../engine/poison';
@@ -169,6 +169,33 @@ describe('provider batches', () => {
     expect(p.sent.slice(1)).toEqual([planned.group.ops.map((op) => ({ ...op, yieldAllowed: true })), buildBatch([other.group])]);
   });
 
+  it('weighs a batch as the Parcel carries it: two bytes per character of its JSON and 300 per op', () => {
+    const ops: ProviderOp[] = [insert('data', { data1: 'x'.repeat(1000) }), insert('raw_contacts')];
+
+    expect(estimateBatchBytes(ops)).toBe(2 * JSON.stringify(ops).length + 2 * 300);
+    expect(estimateBatchBytes(ops)).toBeGreaterThan(2 * 1000);
+  });
+
+  it('packs batches by that weight', async () => {
+    const p = port(ok);
+    const group = (i: number): Work => ({ group: { ref: `g${i}`, ops: [insert('data', { data1: 'x'.repeat(1000) })] } });
+
+    await writer(p, { maxBytes: 2 * estimateBatchBytes(group(0).group.ops) }).write([group(0), group(1), group(2)]);
+
+    expect(p.sent.map((b) => b.length)).toEqual([2, 1]);
+  });
+
+  it('sends the state op alone after a group that fills a yield window, and with a group that leaves room', async () => {
+    const tail = () => ({ op: () => ({ op: 'syncState', value: 's' }) as ProviderOp, applied: () => undefined });
+    const group = (n: number): Work => ({ group: { ref: `g${n}`, ops: Array.from({ length: n }, () => insert('data')) } });
+    const p = port(ok);
+
+    await writer(p, { maxOps: 1000 }).write([group(MAX_OPS_PER_YIELD)], tail());
+    await writer(p, { maxOps: 1000 }).write([group(MAX_OPS_PER_YIELD - 1)], tail());
+
+    expect(p.sent.map((b) => b.length)).toEqual([MAX_OPS_PER_YIELD, 1, MAX_OPS_PER_YIELD]);
+  });
+
   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 } }] } }]);

src/device-sync/engine/batch.ts

@@ -4,9 +4,11 @@
  *
  * - a group is never split; `refs` are rebased to the batch and the group's
  *   first op is its yield point, so the provider may commit between groups
- *   but never inside one;
- * - at most `maxOps` ops and about `maxBytes` of JSON per batch; groups that
- *   carry photo bytes travel alone (Binder budget);
+ *   but never inside one (a group of 499 ops or more is the planner's to
+ *   avoid: ContactsProvider refuses it);
+ * - at most `maxOps` ops and about `maxBytes` per batch, as
+ *   `estimateBatchBytes` counts them; groups that carry photo bytes travel
+ *   alone (Binder budget);
  * - the SyncState op, when there is one, is the last op of the last batch
  *   and never a yield point: the stored state can't get ahead of its rows.
  *
@@ -47,12 +49,22 @@ export interface Work {
 
 export interface BatchWriterOptions {
   maxOps: number;
+  /** Per batch, as `estimateBatchBytes` counts them. */
   maxBytes: number;
   maxReplans: number;
 }
 
-function sizeOf(ops: readonly ProviderOp[]): number {
-  return JSON.stringify(ops).length;
+/** Ops a provider takes up to and including one yield point (ContactsProvider throws at the 500th). */
+export const MAX_OPS_PER_YIELD = 499;
+
+/**
+ * What ops weigh in the Binder transaction of an `applyBatch`, for the batch
+ * budget and for planners that split big items: a Parcel carries strings as
+ * UTF-16 (two bytes per character of their JSON) plus about 300 bytes per op
+ * (URI, flags, value types, back-reference classes).
+ */
+export function estimateBatchBytes(ops: readonly ProviderOp[]): number {
+  return 2 * JSON.stringify(ops).length + 300 * ops.length;
 }
 
 function carriesBlob(ops: readonly ProviderOp[]): boolean {
@@ -131,8 +143,11 @@ export class BatchWriter {
     for (let i = 0; i < batches.length; i++) {
       const last = i === batches.length - 1;
       const batch = batches[i];
-      // A photo batch is already at the Binder budget: its state op goes alone.
-      const tailHere = last && tail && !(batch.length === 1 && carriesBlob(batch[0].group.ops)) ? tail : undefined;
+      // A photo batch is already at the Binder budget, and a group that fills a yield window leaves no room:
+      // the state op goes alone.
+      const photo = batch.length === 1 && carriesBlob(batch[0].group.ops);
+      const full = batch[batch.length - 1].group.ops.length >= MAX_OPS_PER_YIELD;
+      const tailHere = last && tail && !photo && !full ? tail : undefined;
       await this.applyBatch(batch, tailHere);
       if (last && tail && !tailHere) await this.applyTail(tail);
     }
@@ -156,7 +171,7 @@ export class BatchWriter {
         continue;
       }
       const n = work.group.ops.length;
-      const b = sizeOf(work.group.ops);
+      const b = estimateBatchBytes(work.group.ops);
       if (current.length && (ops + n > this.options.maxOps || bytes + b > this.options.maxBytes)) flush();
       current.push(work);
       ops += n;

src/device-sync/engine/deps.ts

@@ -58,7 +58,7 @@ export interface Tuning {
   chunkSize: number;
   /** Provider ops per applyBatch, with a yield point at every item group. */
   maxBatchOps: number;
-  /** Serialized bytes per applyBatch (the Binder budget is 1 MB for the whole transaction). */
+  /** Bytes per applyBatch as `estimateBatchBytes` counts them (the Binder budget is 1 MB for the whole transaction). */
   maxBatchBytes: number;
   /** Re-reads and re-plans of an item whose group failed an assert. */
   maxReplans: number;
@@ -94,7 +94,7 @@ export interface Tuning {
 export const DEFAULT_TUNING: Tuning = {
   chunkSize: 50,
   maxBatchOps: 400,
-  maxBatchBytes: 300_000,
+  maxBatchBytes: 500_000,
   maxReplans: 3,
   maxCreatesPerSet: 50,
   maxSetBytes: 1_000_000,