Bumping a Schema Version Should Not Erase What You Already Read

Flow

Applying the fix in order

1. 1Batch the writes⌄

   One transaction per batch with explicit wait and scaled timeout.

2. 2Serialize writes⌄

   One queue for every database operation; reads run in a few lanes.

3. 3Split coverage from shape⌄

   Keep the range on a version bump; start a refresh cursor.

4. 4Adapt read size⌄

   Longer when the source is cheap, small after errors or partial answers.

5. 5Preview first⌄

   Operator command shows planned ranges and writes only with apply.

6. 6Verify⌄

   Version bump keeps range; one failure does not stall others; re-runs are idempotent.

Building an Incremental Sync That Survives a Schema Change

If you own a job that copies records from an upstream API into your own database on a schedule, this is the checklist and code shape that came out of fixing one that was slow, flaky, and forgetful.

Symptom checklist

You probably have the problems described here if:

• some partitions (locations, tenants, accounts) fail on every run with connection-acquire or "transaction closed" errors while others succeed;
• a backfill that should take minutes is projected to take most of a day;
• adding a column to the stored rows makes the job start over from scratch.

Step 1: batch the writes and set the budget yourself

Replace per-record transactions with per-batch ones. Inside each batch transaction: look up existing rows once, bulk-insert new rows, update only rows whose compared fields actually changed, bulk-touch the rest, then delete and re-insert the batch's child rows.

async function writeBatch(tx: Tx, batch: IncomingRecord[], syncedAt: Date) {
  const existing = await tx.record.findMany({ where: byNaturalKey(batch) })
  const known = new Map(existing.map((row) => [naturalKey(row), row]))

  // New rows: one bulk insert. A row a concurrent writer inserted first is skipped, not duplicated.
  const fresh = batch.filter((item) => !known.has(naturalKey(item)))
  await tx.record.createMany({ data: fresh.map((item) => toRow(item, syncedAt)), skipDuplicates: true })

  // Changed rows: individual updates. Unchanged rows: one bulk touch of the sync timestamp.
  const unchangedIds: string[] = []
  for (const item of batch) {
    const row = known.get(naturalKey(item))
    if (!row) continue
    if (sameFields(row, item)) unchangedIds.push(row.id)
    else await tx.record.update({ where: { id: row.id }, data: { ...toFields(item), syncedAt } })
  }
  await tx.record.updateMany({ where: { id: { in: unchangedIds } }, data: { syncedAt } })

  // Re-read ids so child rows attach to whichever row won any insert race.
  const ids = fresh.length > 0 ? await tx.record.findMany({ where: byNaturalKey(batch) }) : existing
  // Child rows are replaced inside this transaction, so readers never see half a record.
  await replaceChildRows(tx, batch, ids)
}

// Named budgets, defined elsewhere: tune for your workload and your pool.
const BUDGET = {
  acquireWaitMs: ACQUIRE_WAIT_MS,
  baseMs: BASE_TIMEOUT_MS,
  perRecordMs: PER_RECORD_MS,
  perChildMs: PER_CHILD_MS,
  capMs: MAX_TIMEOUT_MS,
}

export const txOptions = (records: number, children: number) => ({
  maxWait: BUDGET.acquireWaitMs,
  timeout: Math.min(BUDGET.capMs, BUDGET.baseMs + records * BUDGET.perRecordMs + children * BUDGET.perChildMs),
})

The constants are named, not given: tune them for your workload and your pool. The pattern is what matters: an explicit connection wait, and a timeout that grows with the work in the batch and is capped. Also chunk large child inserts so a single statement stays under your database's bind-parameter limit.

skipDuplicates covers the race where another writer inserted a row between your lookup and your insert; that is why the ids are re-read before child rows are attached.

Step 2: concurrent reads, serial writes

Run a small, fixed number of partitions' reads at once with a simple lane pool, and route every database write through one serial queue (a promise chain whose tail swallows errors so one failure does not block the rest). Record each partition's failure on that partition and keep going.

Step 3: separate coverage from shape

Store three things per partition: the covered range, the shape version, and a nullable refresh cursor. On a version bump, keep the range and set the cursor to the range's start. After the forward pass, loop in rounds until the run's time budget is spent: in each round, give every unfinished partition one backfill step and one refresh step from the cursor, oldest first, ending before the window the forward pass re-reads anyway.

Step 4: adaptive read size

Start backfill and refresh reads at a moderate span. If the upstream tells you an answer came from a cheap archive, lengthen the span. On an error at a large span, retry once at the small span and stay small for the rest of the run. On a partial answer, stop that direction for this run and record coverage only up to the point the upstream vouched for.

Step 5: an operator verb with preview

Give operators a command that runs the same pass without the per-run time budget, one partition at a time, with progress per read. Make it preview by default: show the forward, backfill and refresh ranges per partition and an estimated read count, and write only with an explicit apply flag.

Verify it

• Test that a version bump keeps the range and starts the cursor.
• Test that a failed write for one partition leaves the others' state advanced.
• Test that a partial answer never advances coverage past the vouched point.
• Test that re-running the same batch leaves row counts unchanged.

Tradeoffs

A bad record now fails its whole batch, so log which batch and let the next run retry. One write queue caps write throughput by design; if that becomes the bottleneck, partition the queue by independent tables rather than removing it.