Data Migration and Consistency Questions

# pseudocode (Python-like)
WORKER_ID = uuid()
BATCH_SIZE = 100
MAX_RETRIES = 5

# compute_full_document(row) -> string
# db.execute(query, params) executes transactional SQL

while True:
    # 1) claim a batch of candidate rows atomically
    # candidate: rows where full_search_document IS NULL OR needs recompute (stale)
    rows = db.fetch(
      "SELECT id, row_version, source_columns FROM table WHERE (full_search_document IS NULL OR backfill_version < ?) AND backfill_owner IS NULL LIMIT ? FOR UPDATE SKIP LOCKED",
      [TARGET_BACKFILL_VERSION, BATCH_SIZE]
    )

    if not rows:
        break

    for r in rows:
        retries = 0
        while retries < MAX_RETRIES:
            try:
                new_doc = compute_full_document(r.source_columns)

                # atomic conditional update: only apply if row_version unchanged and backfill_version < target
                updated = db.execute(
                  "UPDATE table SET full_search_document = ?, backfill_version = ?, backfill_owner = NULL, updated_at = now() WHERE id = ? AND row_version = ? AND backfill_version < ?",
                  [new_doc, TARGET_BACKFILL_VERSION, r.id, r.row_version, TARGET_BACKFILL_VERSION]
                )

                if updated == 1:
                    # success: row converged
                    break
                else:
                    # someone else changed row (app write or other worker). reload row and decide:
                    r = db.fetch_one("SELECT id,row_version,source_columns,backfill_version FROM table WHERE id = ?", [r.id])
                    if r.backfill_version >= TARGET_BACKFILL_VERSION and r.full_search_document is not NULL:
                        # already converged
                        break
                    # otherwise recompute with new data and retry
                retries += 1
            except TransientDBError:
                sleep(exp_backoff(retries))
                retries += 1
        if retries >= MAX_RETRIES:
            log.warning("failed to backfill row", r.id)

# termination: ensure no remaining candidates
# run periodically until zero rows match (full_search_document IS NULL OR backfill_version < TARGET_BACKFILL_VERSION)