Skip to main content
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:
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

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. 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: 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). 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 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.

​
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 — Fernet locally, AWS KMS or Azure Key Vault in production), storing only {"schema": "causeloop.email.v1", "ciphertext": <base64>} 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

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.