> ## Documentation Index > Fetch the complete documentation index at: https://docs.causeloop.ai/llms.txt > Use this file to discover all available pages before exploring further. # Workers and jobs > control.platform_jobs as a durable, retryable PostgreSQL work queue: job types, the provisioner/product-worker privilege split, heartbeats, retry and recovery, and the encrypted email outbox. Every piece of work the product API can't finish inside one HTTP request — provisioning a tenant's schema, training a model, materializing insight collections, sending an email — is a row in `control.platform_jobs`, claimed and executed by one of two long-running worker processes. There is no separate message broker: the job row *is* the queue message, the retry state, and the audit record, all in the same PostgreSQL database that holds everything else. ## `control.platform_jobs` as a durable work queue The table originates in the `20260722_c001` control-schema migration as a simple job record (`id`, `tenant_id`, `job_type`, `status`, `progress`, `steps_completed`, `error_detail`). The `20260727_c003` migration ("Make control.platform\_jobs a durable retryable work queue") is what turns it into an actual queue by adding: ```sql theme={null} ALTER TABLE control.platform_jobs ADD COLUMN IF NOT EXISTS payload JSONB NOT NULL DEFAULT '{}'::jsonb, ADD COLUMN IF NOT EXISTS idempotency_key TEXT, ADD COLUMN IF NOT EXISTS attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts >= 0), ADD COLUMN IF NOT EXISTS max_attempts INTEGER NOT NULL DEFAULT 5 CHECK (max_attempts >= 1), ADD COLUMN IF NOT EXISTS available_at TIMESTAMPTZ NOT NULL DEFAULT now(), ADD COLUMN IF NOT EXISTS locked_at TIMESTAMPTZ, ADD COLUMN IF NOT EXISTS locked_by TEXT, ADD COLUMN IF NOT EXISTS finished_at TIMESTAMPTZ; CREATE UNIQUE INDEX IF NOT EXISTS platform_jobs_idempotency_idx ON control.platform_jobs (idempotency_key) WHERE idempotency_key IS NOT NULL; CREATE INDEX IF NOT EXISTS platform_jobs_queue_idx ON control.platform_jobs (status, available_at, created_at) WHERE status = 'queued'; CREATE TABLE IF NOT EXISTS control.runtime_heartbeats ( component TEXT PRIMARY KEY, instance_id TEXT NOT NULL, heartbeat_at TIMESTAMPTZ NOT NULL DEFAULT now(), detail JSONB NOT NULL DEFAULT '{}'::jsonb ); ``` `status` is constrained (from `c001`) to `queued | running | succeeded | failed | cancelled`. The queue mechanics live in `services/pipeline/jobs.py`: * **Enqueue** (`enqueue_platform_job`) inserts with `ON CONFLICT (idempotency_key) DO NOTHING` and returns the existing row's id on conflict — so retrying an enqueue call with the same key (e.g., a client retrying a `202` request) never creates a duplicate job. Producers pass an explicit `idempotency_key` such as `f"provision:{tenant_id}"` or `f"train:{tenant_id}:{dataset_id}:{config_hash}"`. Because `enqueue_platform_job` takes a caller-supplied `Session`, a route handler can insert the job row in the *same transaction* as the business write it belongs to — `onboarding_api.py`'s `ingest_commit` inserts the new `Dataset`/`DatasetRecord` rows and enqueues the corresponding `materialize_insights` job in one `tenant_uow()` transaction, so the two either both commit or neither does. * **Claim** (`claim_next_job`) selects one `queued` row whose `available_at <= now()` and `job_type` is in the calling worker's allowed set, ordered by `created_at ASC`, with `FOR UPDATE SKIP LOCKED` — the standard PostgreSQL pattern for a multi-consumer queue: concurrent workers never block each other on the same claim query, and a locked row is invisible to other claimants rather than causing them to wait. The claimant immediately flips `status` to `running`, increments `attempts`, and stamps `locked_at`/`locked_by`, all before returning — a worker never holds an open database transaction while it executes the job's actual work. ## Job types and which worker runs them ```python theme={null} REQUIRED_WORKER_JOB_TYPES: dict[str, frozenset[str]] = { "product-worker": frozenset({"materialize_insights", "send_email", "train_tenant_model"}), "provisioner-worker": frozenset({"provision_tenant"}), } ``` | Job type | Worker | Triggered by | | ---------------------- | -------------------- | ---------------------------------------------------------------------------------------------------------------------- | | `provision_tenant` | `provisioner-worker` | `POST /admin/tenants/{tenant_id}/provision` | | `train_tenant_model` | `product-worker` | `POST /admin/tenants/{tenant_id}/train` | | `materialize_insights` | `product-worker` | `POST /admin/tenants/{tenant_id}/ingest/commit` and `POST /admin/tenants/{tenant_id}/activate` | | `send_email` | `product-worker` | Invitation flow — `create_invitation()` in `services/api/auth/tenant_auth.py`, reached via the onboarding invite route | `request_tenant_password_reset()` (`services/api/auth/tenant_auth.py`) and an equivalent path in `services/api/auth/employee_auth.py` also enqueue `send_email` jobs, but their routes (`/password-reset/*`, `/admin/password-reset/*`) are not in `AUTH_OPERATIONS` and so never reach the deployed product surface today — see [Backend in depth](/architecture/backend). Both processes run the exact same entrypoint — `scripts/run_pipeline_worker.py`, which loads local env files and calls `asyncio.run(run_job_worker())` from `services/pipeline/job_worker.py`. What differs is configuration, not code: `CAUSELOOP_WORKER_COMPONENT` (the heartbeat component name) and `CAUSELOOP_WORKER_JOB_TYPES` (the comma-separated allowlist passed to `claim_next_job`). `run_job_worker()` validates that every configured job type is a member of `SUPPORTED_JOB_TYPES` and refuses to start otherwise. ## The provisioner / product-worker privilege split `infra/docker-compose.product.yml` configures the two workers with deliberately different credentials and network access — this is a real security boundary, not just naming: | | `product-worker` | `provisioner-worker` | | ---------------------------------------- | ------------------------------------------- | ------------------------------------------------------------------------------------------------------ | | Database role | `APP_DATABASE_URL` only (`app_rw`) | **both** `APP_DATABASE_URL` and `MIGRATIONS_DATABASE_URL` (owner) | | `REDIS_URL` | set | **not set** | | Email env (`CAUSELOOP_EMAIL_*`, Mailpit) | set | **not set** | | Object store | set | set, but only ever used for a `head_bucket`/`get_container_properties` readiness check — never a write | | `depends_on` | `mailpit`, `migrate`, `minio-init`, `redis` | `migrate`, `minio-init` only | `provisioner-worker` needs owner/migration credentials because `provision_tenant()` (`services/api/tenancy/provisioner.py`) does real DDL: `CREATE SCHEMA IF NOT EXISTS`, running the `tenant_template` Alembic branch to head inside the new schema, and `GRANT`/`ALTER DEFAULT PRIVILEGES` so the ordinary `app_rw` role can subsequently read and write that schema's tables. None of that is available to — or needed by — `app_rw`. Conversely, `product-worker` never gets `MIGRATIONS_DATABASE_URL`: it only ever needs `app_rw`'s ordinary row-level access, scoped further per tenant by `tenant_uow()`'s transaction-local `search_path` and `app.tenant_id` (see [Tenancy and data model](/architecture/tenancy-and-data-model)). Neither worker container can do more than its job type requires. ## Provisioning: idempotent and resumable `provision_tenant()` runs six named steps in order — `create_schema`, `run_tenant_migrations`, `grant_app_rw`, `seed_role_profiles`, `verify_object_storage`, `activate_onboarding` — and records each one's completion in the job row's `steps_completed` JSONB array as it goes, updating `progress` as `len(done) / len(STEPS) * 100`. On a re-run (worker restart mid-provision, or a retried job), it skips every step already in `steps_completed`. Every individual step is also independently safe to re-run on its own — `CREATE SCHEMA IF NOT EXISTS`, `ON CONFLICT DO NOTHING` on role-profile seeding, Alembic's own per-schema `alembic_version` tracking, `GRANT`/`ALTER DEFAULT PRIVILEGES` being naturally idempotent — so a "completed" step re-executing after a crash mid-write is always safe too. `POST /admin/tenants/{tenant_id}/provision` calls `ensure_provision_job()` synchronously before returning `202`, which creates the `platform_jobs` row (status `queued`, via `enqueue_platform_job()`) if one doesn't already exist — the row transitions to `running` only once `provisioner-worker` claims it. This exists specifically so a client polling `GET .../provision/status` immediately after the `POST` never observes `not_started`: without it, a poller that stops on any terminal-or-absent state would have nothing to distinguish "not yet enqueued" from "already finished," and could give up before the job ever ran. ## Heartbeats and readiness Every worker calls `record_worker_heartbeat()` roughly every 10 seconds (`services/pipeline/job_worker.py`'s main loop), upserting its row in `control.runtime_heartbeats` keyed by `component` with `detail = {"job_types": sorted(allowed_types)}`. This is the same table `GET /health/ready` reads to decide whether `product-worker`/`provisioner-worker` are not just alive but actually configured to run every job type the deployment requires — see [Backend in depth](/architecture/backend) for the readiness check itself. `validate_worker_configuration()` runs once at worker startup, before the first heartbeat: if `send_email` is in the worker's configured job types, it checks `email_delivery_readiness()` and `secret_encryption_readiness()` and raises immediately if either is unready. This means a `product-worker` that cannot actually deliver email never advertises `send_email` capability in the first place — it fails to start rather than silently accepting jobs it can't complete. ## Retry, recovery, and cancellation **Retry.** `fail_or_retry_job()` runs whenever `_execute()` raises. If `attempts >= max_attempts`, the job is marked `failed` (terminal). Otherwise it's requeued with exponential backoff: `available_at = now() + min(300, 2 ** min(attempts, 8))` seconds, capped at 5 minutes. `job_failure_detail()` writes a bounded (4000-character) `error_detail` — except for `send_email` jobs, where it deliberately withholds provider exception detail (`"Email delivery failed; provider detail withheld."`) because an email provider's exception can echo the recipient address or a message fragment, and the encrypted outbox must not reintroduce that data through a plaintext error column. **Recovery.** `recover_stale_jobs()` runs once at worker startup (in addition to whatever periodic recovery a deployment schedules externally): it finds `running` jobs whose lock looks abandoned — `locked_at` older than `CAUSELOOP_JOB_STALE_AFTER_SECONDS` (default 900s / 15 minutes), or no `locked_at` at all with a stale `updated_at` — and either requeues them (if attempts remain) or marks them `failed`, stamping `error_detail = "Recovered after worker lease expired."`. This is what makes a hard worker-process crash (not a graceful shutdown, which would have already released the lock) safe: no job is silently lost, and no job stays claimed forever by a process that no longer exists. **Cancellation.** The `status` column's `CHECK` constraint includes `cancelled`, and `enqueue_platform_job`'s `retry_terminal` option explicitly treats `cancelled` as a resumable terminal state — the same code path that reruns a `failed` job. However, no route currently transitions a job to `cancelled`; the state is reserved in the schema and queue logic for a future cancel endpoint, not reachable through any operation published today. ```mermaid theme={null} stateDiagram-v2 [*] --> queued: enqueue_platform_job (idempotent insert) queued --> running: claim_next_job (FOR UPDATE SKIP LOCKED) running --> succeeded: complete_job running --> queued: fail_or_retry_job (attempts < max_attempts, exponential backoff) running --> failed: fail_or_retry_job (attempts >= max_attempts) running --> queued: recover_stale_jobs (lock expired, attempts remain) running --> failed: recover_stale_jobs (lock expired, attempts exhausted) cancelled --> queued: retry_terminal enqueue (reserved; no route sets cancelled today) failed --> queued: retry_terminal enqueue (attempts reset) succeeded --> [*] failed --> [*] ``` ## The encrypted email outbox Invitation and password-reset emails never carry their recipient, subject, or body as plaintext through `control.platform_jobs.payload`. `build_email_job_payload()` (`services/api/auth/mailbox.py`) JSON-encodes `{to, subject, body}` and immediately calls `encrypt_secret()` (the same credential-encryption boundary described in [Backend in depth](/architecture/backend) — Fernet locally, AWS KMS or Azure Key Vault in production), storing only `{"schema": "causeloop.email.v1", "ciphertext": }` in the job's payload column. `enqueue_email_job()` refuses to enqueue anything that isn't already in that encrypted shape. On the worker side, `deliver_email_job()` decrypts the payload only at the moment of send, inside `_execute()`, and calls `send_email()` against whichever provider is configured (`console` for tests, `local-smtp` against Mailpit locally, `aws-ses` or `azure-communication` in the cloud). `delivery_id` (the job's own UUID) travels through as a stable, at-least-once-safe idempotency signal — used for local console-mailbox dedup and carried as `X-CauseLoop-Delivery-ID` / Azure's `Operation-Id` — because SES and raw SMTP expose no native send-idempotency token of their own. ## Job durability at a glance | | `queued` | `running` | `succeeded` | `failed` | `cancelled` | | ----------------------- | -------------------------------------- | -------------------------- | -------------- | --------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------- | | Set by | `enqueue_platform_job` / `requeue_job` | `claim_next_job` | `complete_job` | `fail_or_retry_job` (attempts exhausted) or `recover_stale_jobs` | reserved; no route sets it today | | `locked_by`/`locked_at` | cleared | set to the claiming worker | cleared | cleared | — | | `default max_attempts` | — | — | — | 5 for `provision_tenant`/`train_tenant_model`/`materialize_insights`; **8** for `send_email` (`enqueue_email_job` overrides the default explicitly) | — | `SUPPORTED_JOB_TYPES` in `services/pipeline/jobs.py` is the single source of truth for valid `job_type` values: `{EMAIL_JOB_TYPE ("send_email"), "materialize_insights", "provision_tenant", "train_tenant_model"}`. Both `enqueue_platform_job` and `claim_next_job` reject any type outside this set — a worker misconfigured with an unknown job type in `CAUSELOOP_WORKER_JOB_TYPES` fails at startup rather than silently claiming nothing. ## Local topology In `infra/docker-compose.product.yml`, both workers build from the same repo `Dockerfile` and run `python scripts/run_pipeline_worker.py` with `restart: unless-stopped`, but their `depends_on` chains reflect the privilege split above: `product-worker` waits on `mailpit` (healthy), `migrate` (completed), `minio-init` (completed), and `redis` (healthy); `provisioner-worker` waits only on `migrate` and `minio-init`. The `api` service itself depends on both workers with `condition: service_started` — the API can boot before either worker is fully warmed up, but `/health/ready` will correctly report `not_ready` until their heartbeats and capabilities land, per [Backend in depth](/architecture/backend). ## Related * [Backend in depth](/architecture/backend) * [System overview](/architecture/overview) * [Tenancy and data model](/architecture/tenancy-and-data-model) * [Client onboarding pipeline](/features/client-onboarding-pipeline)