Batch Task State Machine¶

This page is the authoritative reference for how batch tasks move through the system — from initial API request to final result. It covers the Task-level status transitions, the reconciler loop, per-job lifecycles, and the exact shape of the provider_state JSONB column for every supported provider.

Audience: contributors adding new batch services or debugging stuck/failed tasks.

Two-Phase Design¶

Batch tasks are intentionally split into two phases to decouple HTTP request latency from long-running provider operations.

Phase	Where	What happens
Submit	`app/tasks/{type}/batch/task_handler.py`	Validate inputs, call the provider API to start the job(s), write `provider_state` to the DB, mark task `IN_PROGRESS`. Returns immediately.
Poll & Finalize	`core/background_tasks/reconciler/`	Every 120 s, query all `IN_PROGRESS` tasks, ask each provider if the job finished, update `provider_state`, and on completion write the output file and flip the task to a terminal status.

This means the Pub/Sub handler is not responsible for finalizing the task. It only submits and exits. The reconciler owns every status transition after IN_PROGRESS.

Task-Level State Machine¶

flowchart TD
    A["API Request"] --> B["create_task_checkpoint\nstatus = PENDING"]
    B --> C["Pub/Sub delivers message\nto task handler"]
    C -->|"submission succeeds"| D["mark_task_in_progress\nPENDING → IN_PROGRESS\nprovider_state written, version=0"]
    C -->|"submission fails"| E["mark_task_failed\n→ FAILED"]
    D --> F["Reconciler loop\nevery 120 s"]
    F -->|"provider_state is NULL"| E
    F -->|"created_at is NULL"| E
    F -->|"provider_state fails schema\nvalidation (malformed)"| E
    F -->|"task age > 26 h"| E
    F -->|"all jobs terminal,\n≥1 succeeded"| G["finalize_batch_checkpoint\n→ COMPLETED or PARTIAL_COMPLETE"]
    F -->|"all jobs FAILED"| E
    F -->|"jobs still running"| F

Validation and the created_at/age checks run in that order — a malformed provider_state fails immediately rather than looping silently for up to TASK_TIMEOUT_HOURS (issue #218).

Status values¶

Status	Who sets it	Meaning
`PENDING`	`create_task_checkpoint`	Task created; waiting for Pub/Sub delivery
`IN_PROGRESS`	`mark_task_in_progress`	Jobs submitted to provider; reconciler polling
`COMPLETED`	`finalize_batch_checkpoint`	All jobs succeeded
`PARTIAL_COMPLETE`	`finalize_batch_checkpoint`	At least one job succeeded; some failed
`FAILED`	`mark_task_failed`	All jobs failed, submission error, timeout, or NULL state

Idempotency¶

mark_task_in_progress uses a compound WHERE id = ? AND status = 'PENDING' update. If Pub/Sub re-delivers the message, the second handler invocation sees rows_updated = 0 and skips without submitting duplicate jobs.

Concurrent `provider_state` writes (optimistic locking)¶

Task.version is an integer counter, starting at 0, incremented on every successful provider_state write. update_provider_state(task_id, provider_state, expected_version) matches on WHERE id = ? AND status = 'IN_PROGRESS' AND version = expected_version, and bumps version in the same statement. This closes a JSONB lost-update race (issue #200): if two reconcile passes read the same row and both try to write, only the first commit succeeds; the second gets back None (not the new version) and bails for this cycle rather than clobbering the winner's write — the next poll re-derives fresh state from the provider API, so the discarded write is superseded within one poll interval. Pessimistic locking (SELECT ... FOR UPDATE) was deliberately avoided here because the reconciler calls external provider APIs between reading and writing provider_state; holding a row lock across that I/O window would serialize otherwise-independent tasks against each other for the duration of a network call.

Terminal transitions are version-guarded too. mark_task_failed(task_id, error, expected_version=...) and finalize_batch_checkpoint(..., expected_version=...) extend the same optimistic-locking pattern to the COMPLETED/PARTIAL_COMPLETE/FAILED transition itself — without this, two racing reconcile passes could each independently decide a task is done and both flip its status and both send a notification. expected_version is optional on mark_task_failed (default None, unguarded) since it's also called from many non-reconciler submission-failure paths with no version context; the reconciler always passes it. finalize_batch_checkpoint requires it, since its only callers are the two reconcilers. A lost guard rolls back the entire transaction (including any Dataset/FileRecord rows just inserted) and the caller skips notifying — whichever pass actually won already has, or will next cycle.

Reconciler Loop¶

File: core/background_tasks/reconciler/loop.py

batch_polling_loop
  └─ asyncio.wait_for(reconcile_in_progress_tasks(), timeout=RECONCILER_CYCLE_TIMEOUT_SECONDS)
       ├─ get_in_progress_tasks(limit=RECONCILER_MAX_TASKS_PER_CYCLE)   -- oldest first
       └─ asyncio.gather over tasks, bounded by
          Semaphore(RECONCILER_MAX_CONCURRENT_TASKS)
            └─ per task (_reconcile_one_task):
                 ├─ guard: provider_state NULL → mark_task_failed
                 ├─ guard: created_at NULL → mark_task_failed
                 ├─ guard: provider_state fails schema validation → mark_task_failed
                 ├─ guard: age > TASK_TIMEOUT_HOURS → mark_task_failed (+ shielded notify)
                 └─ dispatch on provider_state["task_type"]
                      ├─ "transcription" → reconcile_transcription_task
                      └─ "completion"    → reconcile_completion_task

A cycle that exceeds its timeout is aborted (logged, not raised) rather than blocking the next cycle indefinitely (issue #215). Dispatch across tasks within a cycle is concurrent but bounded (issue #204) — replacing a fully-sequential loop whose wall-clock grew linearly with backlog size, while staying under the DB pool's spare capacity. Once a task's terminal DB write (finalize_batch_checkpoint/mark_task_failed) has committed, its notification is sent from inside an asyncio.shield()-protected task so an outer cycle-timeout cancellation can't cut it off after the fact — the notification is a fire-and-forget scheduled task either way, but shielding guarantees the code path that schedules it actually runs.

After every cycle (including an empty one), loop.py records a naive-UTC "last cycle completed" timestamp, retrievable via get_last_reconciler_cycle_time() — written for a future reconciler liveness health check (issue #216, owned by the observability workstream, not built here).

Knobs¶

Setting	Default	Effect
`RECONCILER_POLL_INTERVAL_SECONDS`	`120`	Sleep between reconcile cycles
`TASK_TIMEOUT_HOURS`	`26`	Age at which a stuck `IN_PROGRESS` task is force-failed
`RECONCILER_CYCLE_TIMEOUT_SECONDS`	`300`	Hard ceiling on one reconcile cycle's wall-clock time
`RECONCILER_MAX_TASKS_PER_CYCLE`	`200`	Max `IN_PROGRESS` tasks fetched per cycle (oldest first)
`RECONCILER_MAX_CONCURRENT_TASKS`	`10`	Max tasks dispatched concurrently within one cycle
`DB_RETRY_MAX_TRIES` / `DB_RETRY_MAX_TIME_SECONDS`	`3` / `10.0`	Retry budget for transient `OperationalError`s on the reconciler's DB hot path

RECONCILER_MAX_CONCURRENT_TASKS counts toward the same pool-capacity startup check as the single/batch task semaphores (core/settings.py) — raising it requires headroom in DB_POOL_SIZE/DB_MAX_OVERFLOW.

Transient DB error retries¶

get_in_progress_tasks, mark_task_in_progress, update_provider_state, mark_task_failed, mark_task_completed, and finalize_batch_checkpoint are wrapped with a backoff-based retry (core/db/retry.py) on sqlalchemy.exc.OperationalError only — not constraint/programming errors, and not pool-checkout timeouts (retrying those immediately would compound pool exhaustion rather than relieve it). Whole-function retry is safe for each of these because they already follow the atomic-compound-WHERE / single-transaction idempotency pattern described above, so re-running the whole function on a fresh transaction after a transient failure is safe.

Per-Job Lifecycle — Transcription¶

File: core/background_tasks/reconciler/transcription.py

Each entry in provider_state["jobs"] moves through these statuses independently:

flowchart LR
    SUBMITTED --> Polling["polling\nprovider-specific status"]
    Polling -->|"provider reports done"| COMPLETED
    Polling -->|"provider reports failure"| FAILED
    Polling -->|"transient ERROR, attempt < 3"| Polling
    Polling -->|"transient ERROR, attempt = 3"| FAILED

Transient errors: If the provider poll call returns status = "ERROR", the job's transient_failures counter is incremented. The job is retried on the next reconciler cycle. After 3 transient errors the job is forced to FAILED.

Finalization: Once every job is in a terminal state:

All jobs FAILED → mark_task_failed (task = FAILED)
At least one job COMPLETED → upload merged result, call finalize_batch_checkpoint
All succeeded → COMPLETED
Mixed → PARTIAL_COMPLETE

Result merging by provider:

Provider	Merge strategy
GOOGLE	Per-model dict: `{model: {"transcriptions": result}}`
AZURE, SARVAM, AWS	Flat list: `{"transcriptions": [...]}`

Per-Job Lifecycle — Completion¶

File: core/background_tasks/reconciler/completion.py

Completion delegates status checking to batch_completion.check_status, which returns is_complete plus updated per-model outcomes. The reconciler:

Calls provider_state_to_outcomes to deserialize the stored state.
Calls check_status(outcomes, models) — one pass, no retry loop.
Writes outcomes_to_provider_state(updated_outcomes) back to the DB.
If is_complete, fetches results and finalizes.

Terminal status aggregation:

Condition	Final status
All models `COMPLETED`	`COMPLETED`
At least one `COMPLETED` or `PARTIAL_COMPLETE`	`PARTIAL_COMPLETE`
No model succeeded	`FAILED`

`provider_state` JSONB Schema¶

provider_state is written by the task handler when the task moves to IN_PROGRESS and updated in-place by the reconciler on each poll cycle.

Typed validation gate (issue #217): common/types/provider_state.py defines CompletionProviderState/TranscriptionProviderState Pydantic models (extra="allow") that validate the required shape shown below — task_type, and for transcription also provider and a jobs list with the common per-job fields. These are a validation boundary only, not a storage format: the reconciler still reads/writes plain dicts. Validation runs at read-time (in loop.py, before the age/timeout check — a malformed row fails fast instead of silently retrying for up to TASK_TIMEOUT_HOURS) and at write-time (in completion.py/transcription.py, immediately before update_provider_state, where a validation failure indicates a bug in that cycle's own logic and is left to propagate and retry next cycle rather than failing the task). TranscriptionJobState.status is intentionally str, not a Literal, since the 5 provider poll modules don't share a unified status vocabulary.

Transcription — AZURE¶

{
  "task_type": "transcription",
  "provider": "AZURE",
  "dataset_ids": ["<dataset-uuid>"],
  "invalid_urls": ["<url-that-failed-validation>"],
  "jobs": [
    {
      "external_job_id": "<azure-transcription-job-url>",
      "status": "SUBMITTED",
      "result": [{ ... }],
      "error": "<set on failure>"
    }
  ]
}

Transcription — SARVAM¶

{
  "task_type": "transcription",
  "provider": "SARVAM",
  "dataset_ids": ["<dataset-uuid>"],
  "invalid_urls": [],
  "gcs_map": { "<job-id>": "<gcs-uri>" },
  "jobs": [
    {
      "external_job_id": "<sarvam-job-id>",
      "status": "SUBMITTED",
      "result": [{ ... }],
      "error": "<set on failure>",
      "transient_failures": 0
    }
  ]
}

gcs_map is passed to poll_sarvam_once so it can resolve the source GCS URI for a given job ID.

Transcription — AWS¶

{
  "task_type": "transcription",
  "provider": "AWS",
  "dataset_ids": ["<dataset-uuid>"],
  "invalid_urls": [],
  "jobs": [
    {
      "external_job_id": "<aws-transcription-job-name>",
      "gcs_uri": "<source-gcs-uri>",
      "filename": "<original-filename>",
      "status": "SUBMITTED",
      "result": [{ ... }],
      "error": "<set on failure>",
      "transient_failures": 0
    }
  ]
}

Transcription — GOOGLE¶

{
  "task_type": "transcription",
  "provider": "GOOGLE",
  "dataset_ids": ["<dataset-uuid>"],
  "invalid_urls": [],
  "jobs": [
    {
      "model": "long",
      "external_job_id": "<google-lro-operation-name>",
      "status": "SUBMITTED",
      "result": [{ ... }],
      "error": "<set on failure>",
      "transient_failures": 0
    }
  ]
}

One job object per model (e.g. long, short, telephony).

Completion¶

{
  "task_type": "completion",
  "input_dataset_ids": ["<dataset-uuid>"],
  "jobs": [
    {
      "model": "gemini-3.5-flash-lite",
      "status": "SUBMITTED",
      "external_job_id": "<provider-batch-job-id>",
      "output_uri": "<gcs-uri-to-results>",
      "error": "<set on failure>",
      "transient_failures": 0
    }
  ]
}

One job object per model. output_uri is populated by check_status once the provider reports the batch is done.

Adding a New Batch Service to the Reconciler¶

When onboarding a new batch service (see Service Onboarding for the full checklist), the reconciler needs two additions:

A poll function — implement _poll_<service>_job_once(provider, job, provider_state) -> dict that returns {"status": "COMPLETED"|"FAILED"|"ERROR"|<running-status>, "result": ..., "error": ...}. Add it in an existing or new module under core/background_tasks/reconciler/.
A route in reconcile_in_progress_tasks — add a branch in loop.py:

elif task_type == "translation":
    await reconcile_translation_task(task)

The task handler must write provider_state["task_type"] = "<your-type>" so the reconciler can route it correctly.