diff --git a/src/device-sync/__tests__/engine/internals.test.ts b/src/device-sync/__tests__/engine/internals.test.ts index 07ec9348..974f9c9b 100644 --- a/src/device-sync/__tests__/engine/internals.test.ts +++ b/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 } }] } }]); diff --git a/src/device-sync/engine/batch.ts b/src/device-sync/engine/batch.ts index a7a454ea..334a4419 100644 --- a/src/device-sync/engine/batch.ts +++ b/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; diff --git a/src/device-sync/engine/deps.ts b/src/device-sync/engine/deps.ts index f7bbf4a2..8f5bb227 100644 --- a/src/device-sync/engine/deps.ts +++ b/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,