Function runTableUpdate

apps/sim/lib/table/update-runner.ts:61–174

(payload: TableUpdatePayload)

 * failed via `markTableUpdateFailed`. A superseded run returns quietly.
 */
export async function runTableUpdate(payload: TableUpdatePayload): Promise<void> {
  const { jobId, tableId, workspaceId, filter, data, cutoff, maxRows } = payload
  const requestId = generateId().slice(0, 8)
  const budget = maxRows ?? Number.POSITIVE_INFINITY

  try {
    const table = await getTableById(tableId, { includeArchived: true })
    if (!table) throw new Error(`Update target table ${tableId} not found`)

    const filterClause = buildFilterClause(filter, USER_TABLE_ROWS_SQL_NAME, table.schema.columns)
    if (!filterClause) throw new Error('Filter is required for bulk update')

    // Coerce the patch once to the schema's types — the merged validation below and the persisted
    // JSONB merge both use this normalized copy.
    coerceRowValues(data, table.schema)
    const patchJson = JSON.stringify(data)

    // Resume the persisted count: a retried attempt's earlier pages are already committed, so
    // starting at zero would overwrite cumulative progress. Doubles as the initial ownership gate.
    const resumed = await getJobProgress(tableId, jobId)
    if (resumed === null) throw new JobSupersededError()

    let processed = resumed
    let lastReported = resumed
    let afterId: string | undefined

    while (processed < budget) {
      const owns = await updateJobProgress(tableId, processed, jobId)
      if (!owns) throw new JobSupersededError()

      const page = await selectRowDataPage({
        tableId,
        workspaceId,
        cutoff,
        filterClause,
        afterId,
        limit: Math.min(TABLE_LIMITS.DELETE_PAGE_SIZE, budget - processed),
        // Skip rows already carrying the patch so a retried run resumes without re-walking /
        // double-counting the rows an earlier attempt updated (updated rows still exist and may
        // still match the filter, unlike deletes).
        excludeIfPatched: patchJson,
      })
      if (page.length === 0) break
      afterId = page[page.length - 1].id

      // Validate each merged result before writing the page — a row that would overflow the size
      // cap or violate the schema fails the job (earlier pages stay applied; best-effort).
      for (const row of page) {
        const merged = { ...row.data, ...data }
        const sizeValidation = validateRowSize(merged)
        if (!sizeValidation.valid) {
          throw new Error(`Row ${row.id}: ${sizeValidation.errors.join(', ')}`)
        }
        const schemaValidation = coerceRowToSchema(merged, table.schema)
        if (!schemaValidation.valid) {
          throw new Error(`Row ${row.id}: ${schemaValidation.errors.join(', ')}`)
        }
      }