Performance pass: stored-JSON pokes, bootstrap reuse, filter indexes, cheaper client apply

ARCHITECTURE.md

@@ -99,10 +99,20 @@ Message shapes are the zod schemas in `packages/protocol/src/messages.ts`
 promise). What the shapes don't say:
 
 - The **three-part poke** (`pokeStart`/`pokePart`/`pokeEnd`) is Zero's shape
-  (`zero-protocol/src/poke.ts:32-73`): large payloads stream in chunks without
-  server-side buffering, and it doubles as the chunked bootstrap. The
+  (`zero-protocol/src/poke.ts:32-73`): large payloads go out in frame-sized
+  chunks, and it doubles as the chunked bootstrap. The
   `remaining`/`pageInfo` countdown is LiveStore's progress signal
   (`sync-backend.ts:143-155`).
+- **Frames are assembled from stored JSON** (`server/src/frames.ts`): SQLite
+  concatenates each row's op text around the `data` column, and a part is a
+  join of those strings — a row is never parsed and re-serialized on its way
+  out, which was most of a large hello's cost. A poke's frames are all built,
+  then sent, in one synchronous turn (invariant 3). The DO keeps the last
+  bootstrap patch keyed by `(backendId, currentVersion)`: hellos that arrive
+  together (a class opening one workspace, every tab re-bootstrapping after a
+  schema bump) share one build, and any write moves the version, so a stale
+  snapshot is never served. The build is dropped as soon as the version
+  moves, since no later hello could use it.
 - **Every mutation carries a `seed`** (protocol 2): the client-minted value
   behind `ctx.seed`/`ctx.nextId`, echoed to the authoritative run so both
   mint the same ids (ARCHITECTURE.md#optimistic-intents). It is not logged
@@ -203,9 +213,16 @@ collection. `SqlRowStore` also pushes the filter into SQL as
 `json_extract(data, '$.field') = ?` — a *prefilter*, since JSON equality is
 looser than `===` (a JSON `true` extracts as `1`, a JSON `null` and a missing
 key both extract as NULL), so only matching rows are parsed but every parsed
-row is re-checked. The expression form is deliberate: a future declared index
-is a partial expression index over the same text, no generated column and no
-backfill. Non-identifier keys skip the SQL clause and match in JS alone.
+row is re-checked. Each `(table, field)` a filter names gets a partial
+expression index on first use — `json_extract(data, '$.field') WHERE tbl =
+'<table>'`, over the same text the query uses, so no generated column and no
+backfill. The query inlines the table as a literal because SQLite only uses a
+partial index when the query's own WHERE repeats the index's term (the name is
+already restricted to `TABLE_NAME_RE`). Indexes are derived state: every
+filtered list issues `CREATE INDEX IF NOT EXISTS` rather than remembering
+which exist, so one lost to a rolled-back mutation or to admin reset is back
+on the next filtered list.
+Non-identifier keys skip the SQL clause and match in JS alone.
 
 **Post-commit hook.** `onMutationCommitted` on the engine config is the one
 seam for effects outside the workspace's rows (notifications, projections
@@ -265,7 +282,11 @@ We evaluated `@tanstack/db-sqlite-persistence-core` and
 
 The seam is `SyncClient`'s `store: SyncStore` (`packages/client/src/store.ts`);
 `IndexedDBSyncStore` is the browser implementation, `MemorySyncStore` the test
-double and reference.
+double and reference. Each IndexedDB `put` structured-clones its value on the
+calling thread, so a poke's row writes are issued in batches, each queued from
+the previous batch's last request callback: the transaction stays active
+inside request callbacks, so atomicity holds, and a bootstrap no longer
+persists in one long main-thread task.
 
 **Multi-tab needs no leader election.** Rows + cursor are shared per workspace;
 outbox records are partitioned by clientId (each tab replays only its own;
@@ -415,8 +436,8 @@ Lifted from partyserver/tldraw/LiveStore, considered settled:
 - **Broadcast iterates `getWebSockets()` live**; a failed send closes that
   socket with 1011. Slow clients are not backpressured — pokes are deltas and
   a reconnect catches up by cursor, so dropping a laggard is always safe.
-- **Frame budget 900 KB**; the chunker packs by item count and encoded bytes
-  (LiveStore `splitArrayBySize`, `transport-chunking.ts:38-85`).
+- **Frame budget 900 KB**; pokes pack by encoded UTF-8 bytes (LiveStore
+  `splitArrayBySize`, `transport-chunking.ts:38-85`).
 
 ## Schema evolution
 

CLAUDE.md

@@ -27,9 +27,7 @@ Invariants that must never break (see ARCHITECTURE.md#invariants):
 No release workflow, changesets, or tags: bump `version` in the package's package.json, commit, then from the repo root:
 
 ```bash
-npm login                                                  # only if `npm whoami` fails; the token in ~/.npmrc expires
-pnpm build
-pnpm -r --filter './packages/*' publish --access public --no-git-checks
+pnpm release
 ```
 
-Recursive publish skips packages whose version is already on the registry, so it is safe to run for the whole workspace after bumping one package. `--no-git-checks` is required because the untracked `packages/yjs/reference/` clones dirty the tree. `prepack` only syncs docs, so run `pnpm build` first; `publishConfig` swaps exports to `dist/` at pack time. Then `pnpm --filter @cf-sync/demo deploy` if the demo should pick the release up. Consumers on pnpm hold new versions for the release-age window unless they exclude `@cf-sync/*` (corates does).
+It builds, then publishes every package. Recursive publish skips packages whose version is already on the registry, so it is safe to run for the whole workspace after bumping one package. `--no-git-checks` is required because the untracked `packages/yjs/reference/` clones dirty the tree. `prepack` only syncs docs, which is why the script builds first; `publishConfig` swaps exports to `dist/` at pack time. If publish fails on auth, the token in ~/.npmrc has expired: `npm login` and rerun. Then `pnpm --filter @cf-sync/demo deploy` if the demo should pick the release up. Consumers on pnpm hold new versions for the release-age window unless they exclude `@cf-sync/*` (corates does).

docs/guide/defining-your-app.md

@@ -54,7 +54,7 @@ for (const { id } of tx.list('checklists', { where: { studyId, assignedTo: userI
 }
 ```
 
-The filter runs in SQL on the server before rows are parsed and before rows are cloned on the client; it still walks the table, so a mutator that filters once per item in a large batch should build its own lookup up front.
+On the server the filter runs in SQL, and each table and field you filter on gets an index the first time a list uses it, so a filtered read does not walk the table. The client's optimistic run still walks its collection, so a mutator that filters once per item in a large batch should build its own lookup up front.
 
 The definitions can also be written directly inside `defineApp({ mutators: { ... } })` — inference is identical either way. `defineMutators` is only *required* when declaring an [`authContext`](/guide/auth#reading-the-verdict-in-mutators), its third argument.
 

package.json

@@ -6,6 +6,7 @@
   "scripts": {
     "build": "pnpm --filter './packages/*' -r build",
     "check:packages": "pnpm build && bash scripts/check-packages.sh",
+    "release": "pnpm build && pnpm -r --filter './packages/*' publish --access public --no-git-checks",
     "test": "pnpm -r test",
     "typecheck": "pnpm -r typecheck",
     "docs:dev": "pnpm --filter docs dev",

packages/client/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cf-sync/client",
-  "version": "0.3.1",
+  "version": "0.3.2",
   "description": "Browser client for cf-sync: SyncClient (optimistic mutations, offline outbox, IndexedDB mirror, presence) and a TanStack DB collection adapter, with React hooks",
   "license": "MIT",
   "repository": {

packages/client/src/idb-store.ts

@@ -40,6 +40,38 @@ function request<T>(req: IDBRequest<T>): Promise<T> {
   })
 }
 
+/**
+ * Row writes per batch. Each put structured-clones its value synchronously
+ * on the calling thread, so a bootstrap's worth of puts issued at once is
+ * one long main-thread task; batches let the page breathe between them.
+ */
+const WRITE_BATCH = 1_000
+
+/**
+ * Issues the row ops in batches, each queued from the previous batch's last
+ * request callback — the transaction stays active inside request callbacks,
+ * so every op still commits in the one transaction.
+ */
+function writeRows(rowStore: IDBObjectStore, ops: PokePersist['ops']): Promise<void> {
+  return new Promise((resolve, reject) => {
+    let i = 0
+    const next = (): void => {
+      let last: IDBRequest | null = null
+      for (const end = Math.min(i + WRITE_BATCH, ops.length); i < end; i++) {
+        const op = ops[i]!
+        last = op.op === 'put' ? rowStore.put(op.value, [op.tbl, op.id]) : rowStore.delete([op.tbl, op.id])
+      }
+      if (!last || i >= ops.length) {
+        resolve()
+        return
+      }
+      last.onsuccess = next
+      last.onerror = () => reject(last!.error ?? new Error('IndexedDB request failed'))
+    }
+    next()
+  })
+}
+
 function txnDone(txn: IDBTransaction): Promise<void> {
   return new Promise((resolve, reject) => {
     txn.oncomplete = () => resolve()
@@ -155,10 +187,7 @@ export class IndexedDBSyncStore implements SyncStore {
       if (!subsumed) {
         const rowStore = txn.objectStore(ROWS)
         if (update.clear) rowStore.clear()
-        for (const op of update.ops) {
-          if (op.op === 'put') rowStore.put(op.value, [op.tbl, op.id])
-          else rowStore.delete([op.tbl, op.id])
-        }
+        await writeRows(rowStore, update.ops)
         metaStore.put({ schemaVersion: update.schemaVersion, cursor: update.cursor } satisfies MetaRecord, 'state')
       }
       txn.objectStore(OUTBOX).put(

packages/client/test/bench/bootstrap.bench.test.ts

@@ -0,0 +1,139 @@
+/**
+ * Client-side cost of a large workspace (~60k rows). Opt-in, never part of
+ * `pnpm test`:
+ *
+ *   CF_SYNC_BENCH=1 pnpm vitest run test/bench
+ *
+ * Frames are built the way the server chunks them, then fed through a fake
+ * socket into a SyncClient with real TanStack DB collections.
+ */
+import { MAX_PART_PATCH_BYTES, chunkBySize, jsonByteSize, serverMsgSchema } from '@cf-sync/protocol/internal'
+import { it } from 'vitest'
+import { SyncClient } from '../../src/client'
+import { createCollections } from '../../src/collection'
+import { MemorySyncStore } from '../../src/store'
+import { FakeSocket, flushMicrotasks } from '../fake-socket'
+import { testApp } from '../test-schema'
+
+const ROWS = 60_000
+const CHECKLISTS = 840
+
+function uuid(n: number): string {
+  return `00000000-0000-4000-8000-${n.toString(16).padStart(12, '0')}`
+}
+
+type Op = { op: 'clear' } | { op: 'put'; tbl: string; id: string; value: Record<string, unknown> }
+
+function snapshotOps(): Op[] {
+  const ops: Op[] = [{ op: 'clear' }]
+  for (let c = 0; c < CHECKLISTS; c++) {
+    ops.push({ op: 'put', tbl: 'notes', id: uuid(c), value: { id: uuid(c), studyId: uuid(c >> 2), type: 'ROB2' } })
+  }
+  for (let i = 0; i < ROWS; i++) {
+    const checklistId = uuid(i % CHECKLISTS)
+    const key = `domain${i % 7}.q${i % 13}`
+    const id = `${checklistId}:${key}:${i}`
+    ops.push({
+      op: 'put',
+      tbl: 'todos',
+      id,
+      value: { id, studyId: uuid((i % CHECKLISTS) >> 2), checklistId, key, value: i % 3 ? 'Y' : 'PN' },
+    })
+  }
+  // The server sends a snapshot ORDER BY tbl, id.
+  const [clear, ...puts] = ops
+  puts.sort((a, b) => {
+    const x = a as Extract<Op, { op: 'put' }>
+    const y = b as Extract<Op, { op: 'put' }>
+    return x.tbl < y.tbl ? -1 : x.tbl > y.tbl ? 1 : x.id < y.id ? -1 : x.id > y.id ? 1 : 0
+  })
+  return [clear!, ...puts]
+}
+
+function snapshotFrames(clientId: string): string[] {
+  const ops = snapshotOps()
+  const pokeId = 'bench-poke'
+  const frames = [JSON.stringify({ type: 'pokeStart', pokeId, baseCursor: null })]
+  const chunks = chunkBySize(ops, { maxBytes: MAX_PART_PATCH_BYTES, sizeOf: jsonByteSize })
+  let sent = 0
+  chunks.forEach((patch, i) => {
+    sent += patch.length
+    const part: Record<string, unknown> = { type: 'pokePart', pokeId, patch, remaining: ops.length - sent }
+    if (i === 0) part.lastMutationIdChanges = { [clientId]: 0 }
+    frames.push(JSON.stringify(part))
+  })
+  frames.push(
+    JSON.stringify({ type: 'pokeEnd', pokeId, cursor: { backendId: 'b1', version: 1 }, pageInfo: { more: false } }),
+  )
+  return frames
+}
+
+function time(fn: () => void): number {
+  const t0 = performance.now()
+  fn()
+  return Math.round(performance.now() - t0)
+}
+
+async function timeAsync(fn: () => Promise<void>): Promise<number> {
+  const t0 = performance.now()
+  await fn()
+  return Math.round(performance.now() - t0)
+}
+
+function makeClient(store?: MemorySyncStore) {
+  let socket: FakeSocket | null = null
+  const client = new SyncClient({
+    url: 'ws://bench',
+    workspaceId: 'bench',
+    clientId: 'bench-client',
+    autoStart: false,
+    app: testApp,
+    ...(store ? { store } : {}),
+    createSocket: () => (socket = new FakeSocket()),
+  })
+  const collections = createCollections(client, { startSync: true })
+  return { client, collections, socket: () => socket! }
+}
+
+it('client bootstrap timings', async () => {
+  const frames = snapshotFrames('bench-client')
+  const bytes = frames.reduce((n, f) => n + f.length, 0)
+  const results: Record<string, string> = { 'snapshot size': `${Math.round(bytes / 1024)} KB in ${frames.length} frames` }
+
+  // Warm the JIT on the parse path before timing it.
+  for (const frame of frames.slice(0, 3)) serverMsgSchema.safeParse(JSON.parse(frame))
+
+  results['JSON.parse all frames'] = `${time(() => frames.forEach((f) => JSON.parse(f)))} ms`
+  results['JSON.parse + serverMsgSchema.safeParse'] = `${time(() => frames.forEach((f) => serverMsgSchema.safeParse(JSON.parse(f))))} ms`
+
+  const runs: number[] = []
+  for (let r = 0; r < 3; r++) {
+    const store = new MemorySyncStore()
+    const { client, collections, socket } = makeClient(store)
+    client.start()
+    await flushMicrotasks(20)
+    socket().open()
+    runs.push(
+      await timeAsync(async () => {
+        for (const frame of frames) socket().receiveRaw(frame)
+        await flushMicrotasks(20)
+      }),
+    )
+    if (collections.todos.size !== ROWS) throw new Error(`expected ${ROWS} todos, got ${collections.todos.size}`)
+    if (r === 2) {
+      const hydrate = await timeAsync(async () => {
+        const next = makeClient(store)
+        next.client.start()
+        await next.client.whenHydrated
+        if (next.collections.todos.size !== ROWS) throw new Error(`hydrated ${next.collections.todos.size} todos`)
+        await next.client.destroy()
+      })
+      results['hydrate 60k rows from a store into collections'] = `${hydrate} ms`
+    }
+    await client.destroy()
+  }
+  runs.sort((a, b) => a - b)
+  results['receive + apply bootstrap poke (median of 3)'] = `${runs[1]} ms`
+
+  console.log(`\n[cf-sync client bench] ${ROWS} rows\n${Object.entries(results).map(([k, v]) => `  ${k.padEnd(48)} ${v}`).join('\n')}\n`)
+})

packages/client/test/idb-store.test.ts

@@ -30,6 +30,21 @@ describe('IndexedDBSyncStore', () => {
     expect(await store.load()).toBeNull()
   })
 
+  it('persists a poke larger than one write batch, deletes included, in one commit', async () => {
+    const { store } = makeStore()
+    const puts = Array.from({ length: 2_500 }, (_, i) => ({ op: 'put' as const, tbl: 'todos', id: `t${i}`, value: { i } }))
+    await store.applyPoke(pokeUpdate({ clear: true, ops: puts, outbox: [{ id: 1, name: 'x', args: null, seed: 's' }] }))
+    // Deletes land on both sides of the 1,000-op batch boundaries.
+    const dels = [5, 999, 1_000, 1_001, 2_499].map((i) => ({ op: 'del' as const, tbl: 'todos', id: `t${i}` }))
+    await store.applyPoke(pokeUpdate({ ops: [...dels, ...puts.slice(0, 1_200)], cursor: { backendId: 'b1', version: 2 } }))
+    const state = await store.load()
+    expect(state?.cursor).toEqual({ backendId: 'b1', version: 2 })
+    const ids = new Set(state!.rows.map((r) => r.id))
+    expect(ids.size).toBe(2_499)
+    for (const id of ['t5', 't999', 't1000', 't1001']) expect(ids.has(id)).toBe(true) // re-put after the delete
+    expect(ids.has('t2499')).toBe(false)
+  })
+
   it('round-trips rows, cursor, lmid, and outbox through applyPoke', async () => {
     const { store } = makeStore()
     await store.applyPoke(

packages/client/vitest.config.ts

@@ -2,6 +2,9 @@ import { defineConfig } from 'vitest/config'
 
 export default defineConfig({
   test: {
-    include: ['test/**/*.test.ts'],
+    // Timings over a large workspace, opt-in: CF_SYNC_BENCH=1 pnpm vitest run test/bench
+    include: process.env.CF_SYNC_BENCH ? ['test/bench/**/*.test.ts'] : ['test/**/*.test.ts'],
+    exclude: process.env.CF_SYNC_BENCH ? [] : ['test/bench/**', 'node_modules/**'],
+    testTimeout: process.env.CF_SYNC_BENCH ? 600_000 : 5_000,
   },
 })

packages/protocol/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cf-sync/protocol",
-  "version": "0.2.0",
+  "version": "0.2.1",
   "description": "Shared app-definition kit and wire protocol for cf-sync: defineApp, defineSchema, defineMutators, zod-validated frames \u2014 importable from both worker and browser",
   "license": "MIT",
   "repository": {

packages/protocol/src/messages.ts

@@ -181,12 +181,20 @@ export type ClientMsg = z.infer<typeof clientMsgSchema>
 // server -> client
 // ---------------------------------------------------------------------------
 
+// A row value is checked for shape only. z.record would copy every value,
+// which on a bootstrap is a second full copy of the workspace on the client's
+// main thread; the value came out of JSON.parse and the server validated it.
+const rowValueSchema = z.custom<Record<string, unknown>>(
+  (value) => typeof value === 'object' && value !== null && !Array.isArray(value),
+  'expected an object',
+)
+
 export const patchOpSchema = z.discriminatedUnion('op', [
   z.object({
     op: z.literal('put'),
     tbl: z.string().min(1),
     id: z.string().min(1),
-    value: z.record(z.string(), z.unknown()),
+    value: rowValueSchema,
   }),
   z.object({ op: z.literal('del'), tbl: z.string().min(1), id: z.string().min(1) }),
   z.object({ op: z.literal('clear') }),

packages/protocol/test/messages.test.ts

@@ -0,0 +1,17 @@
+import { describe, expect, it } from 'vitest'
+import { patchOpSchema } from '../src/internal'
+
+describe('patchOpSchema', () => {
+  it('passes a put value through by reference, with no copy', () => {
+    const value = { title: 'x', nested: { n: 1 } }
+    const parsed = patchOpSchema.parse({ op: 'put', tbl: 'todos', id: 't1', value })
+    expect(parsed).toEqual({ op: 'put', tbl: 'todos', id: 't1', value })
+    expect((parsed as { value: unknown }).value).toBe(value)
+  })
+
+  it('rejects put values that are not objects', () => {
+    for (const value of [null, [], 'x', 1, true, undefined]) {
+      expect(patchOpSchema.safeParse({ op: 'put', tbl: 'todos', id: 't1', value }).success).toBe(false)
+    }
+  })
+})

packages/server/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cf-sync/server",
-  "version": "0.3.0",
+  "version": "0.3.1",
   "description": "Server-authoritative sync engine on Cloudflare Durable Objects: createWorkspaceDO, worker routers, admin surface, and an in-memory test engine",
   "license": "MIT",
   "repository": {