Workers¶

Run validation and pipeline jobs asynchronously in separate worker processes. Workers poll database-backed queues, claim jobs atomically, execute public PyCharter APIs, and update job/run status. No message broker is required.

Overview¶

Database-backed queues: Validation jobs live in validation_jobs; pipeline jobs live in pipeline_jobs.
Unified worker process: One pycharter worker command polls validation and/or pipeline queues by mode.
Multiple workers: Run several worker processes against the same database; each job is claimed by exactly one worker.
Lease + heartbeat: Long-running jobs extend their claim lease while processing; expired leases are released for retry.
Structured DLQ: Terminal failures are recorded in dead_letter_queue (when enabled) and replayed via /api/v1/dlq.
Durable pipeline runs: Pipeline jobs update pipeline_runs, the same lifecycle table used by the Pipelines UI.

For operability (metrics, health probes, NOTIFY, retention), see Worker hardening runbook.

For layer boundaries and roadmap history, see PyCharter Worker Roadmap ADR.

In the existing worker loops, pipeline and streaming launch claims compare the job's current pending status and due time before granting ownership; competing readers cannot both claim the same attempt. Validation retains its existing atomic claim implementation. Every heartbeat checks the persisted owner and claimed status. SQLite coalesces lease changes within one second (at most half the lease duration); PostgreSQL renews on every legacy heartbeat. SQLite's conditional update still takes a writer lock, so coalescing reduces lease-row/WAL changes rather than removing lock contention. Failed renewal writes retry or raise; they never become cached successes. These loops still use ID-only completion/failure APIs, so the heartbeat checks alone do not fence a stale worker's later result writes.

The separately invoked owned SQL queue methods add caller-known submission recovery, exact locked-row identity checks and final generation/token/live-lease predicates. Current worker and launcher loops do not carry those fences or execute protected owned submissions. Consumer integration remains separate.

When to use the worker¶

Non-blocking validation: API returns immediately with a job_id; clients poll for status.
Heavy or slow checks: Offload validation to background processes so the API stays responsive.
Durable pipeline execution: Run ETL pipelines outside the API request lifecycle.
Scaling: Add more worker processes (Postgres recommended for multi-poller production deployments).

What the worker does not cover¶

Path	Use instead
Streaming validation (`pycharter stream run`)	Streaming worker loop and Kafka sources
Synchronous API validation	Quality Check API endpoints
Background ingest helpers	`_ingest_background` and related API paths

Prerequisites¶

Database configured (PostgreSQL recommended for production; SQLite for local dev) with schema applied (pycharter db upgrade).
validation_jobs and pipeline_jobs tables exist (created by migrations).
For metrics/health/NOTIFY: pip install pycharter[server].

API, finite-job workers, streaming workers and the streaming launcher share database admission. Empty persistent SQLite can initialize on first use; existing incompatible schemas refuse startup with pycharter db upgrade as the repair hint. PostgreSQL must be provisioned explicitly before startup. Admission does not upgrade or stamp an existing database.

Running the worker¶

1. Set the database URL¶

The worker uses the same configuration as the rest of PyCharter:

Environment: PYCHARTER_DATABASE_URL or PYCHARTER_WORKER_DB_URL
Config file: pycharter.cfg [database] / [worker] sections

Verify connectivity:

pycharter db current

2. Start worker processes¶

# Default: mode=all, 5 pollers per enabled queue, 500 ms poll interval
pycharter worker

# Validation queue only
pycharter worker --mode validation

# Pipeline queue only
pycharter worker --mode pipeline

CLI options:

Option	Description	Default
`--mode`	`all`, `validation`, or `pipeline`	`all`
`--concurrency`	Default pollers per enabled queue	`5`
`--validation-concurrency`	Validation queue pollers	inherits `--concurrency`
`--pipeline-concurrency`	Pipeline queue pollers	inherits `--concurrency`
`--poll-interval`	Poll interval in milliseconds	`500`

Examples:

# Mixed workload: more validation pollers, fewer pipeline pollers
pycharter worker --mode all --validation-concurrency 8 --pipeline-concurrency 2

# Production-style operability (Postgres)
export PYCHARTER_WORKER_METRICS_ENABLED=true
export PYCHARTER_WORKER_HEALTH_SERVER_ENABLED=true
export PYCHARTER_WORKER_NOTIFY_LISTENER_ENABLED=true
pycharter worker --mode all

Stop a worker with Ctrl+C. It finishes in-flight jobs (up to PYCHARTER_WORKER_DRAIN_TIMEOUT_S) then exits.

Submitting jobs¶

From the API¶

Validation jobs:

Submit: POST /api/v1/quality/jobs
Status: GET /api/v1/quality/jobs/{job_id}
Cancel: POST /api/v1/quality/jobs/{job_id}/cancel (when supported)
Health/stats: GET /api/v1/quality/jobs/health, GET /api/v1/quality/jobs/stats

Pipeline jobs:

Submit: POST /api/v1/pipeline/jobs
List: GET /api/v1/pipeline/jobs
Stats: GET /api/v1/pipeline/jobs/stats
Cancel: POST /api/v1/pipeline/jobs/{run_id}/cancel

DLQ (terminal failures):

List/stats: GET /api/v1/dlq, GET /api/v1/dlq/stats
Replay: POST /api/v1/dlq/{id}/replay
Discard: POST /api/v1/dlq/{id}/discard

See the REST API or Swagger UI for request/response shapes.

From Python¶

from pycharter.worker import submit_job_sync

job_id = submit_job_sync(
    contract_name="user_schema",
    contract_version="1.0",
    data_source="/path/to/data.json",
    options={"record_violations": True},
)

Async variant:

from pycharter.worker import submit_job, get_job

job_id = await submit_job(
    contract_name="user_schema",
    contract_version="1.0",
    data_source="s3://bucket/data/users.parquet",
)
status = await get_job(job_id)

Optional arguments: options, job_id, idempotency_key, max_attempts, db_url.

Architecture¶

flowchart LR
    API[API / Client] --> VQ[(validation_jobs)]
    API --> PQ[(pipeline_jobs)]
    PQ --> PR[(pipeline_runs)]
    VQ --> W[Worker]
    PQ --> W
    W --> VQ
    W --> PQ
    W --> PR
    W --> DLQ[(dead_letter_queue)]

Producer: API or Python code inserts a pending row.
Worker: Pollers claim jobs atomically, setting claimed_by, lease_expires_at, and incrementing attempt.
Heartbeat: While processing, the worker extends lease_expires_at on the configured interval.
Execution: Validation runs QualityCheck; pipeline runs Pipeline.from_dict(...).run(...).
Result: Success marks completed. Retryable failure returns to pending with backoff. Terminal failure sets failed, writes DLQ when enabled, and stores dlq_record_id.

On Postgres, job submission emits pg_notify on channel pycharter_worker_jobs so opt-in LISTEN consumers wake pollers early.

Status semantics¶

Persisted status	Meaning
`pending`	Queued, waiting for a worker
`claimed`	Running in a worker process
`completed`	Finished successfully
`failed`	Terminal failure (may have `dlq_record_id` for replay)
`cancelled`	Cancelled by API/operator

Legacy dead_letter rows may exist from older deployments; new terminal failures use failed + structured DLQ.

Late completions do not overwrite cancelled jobs.

Configuration¶

All worker settings are documented in Configuration — Worker and DLQ and pycharter.cfg.example [worker].

Quick reference:

Source	Purpose
`PYCHARTER_DATABASE_URL`	Default database URL (worker and API)
`PYCHARTER_WORKER_DB_URL`	Override database URL for worker only
`PYCHARTER_WORKER_MODE`	`all`, `validation`, or `pipeline`
`PYCHARTER_WORKER_LEASE_SECONDS` / `PYCHARTER_WORKER_HEARTBEAT_SECONDS`	Claim lease and heartbeat
`PYCHARTER_WORKER_DLQ_*`	Structured dead-letter persistence
`PYCHARTER_WORKER_METRICS_ENABLED`	Prometheus metrics (opt-in)
`PYCHARTER_WORKER_HEALTH_SERVER_*`	Health/metrics HTTP server (opt-in)
`PYCHARTER_WORKER_NOTIFY_LISTENER_ENABLED`	Postgres NOTIFY wake-up (opt-in)
`PYCHARTER_WORKER_RETENTION_*`	Terminal row retention sweeps (opt-in)

Troubleshooting¶

Worker won't start¶

"No database URL": Set PYCHARTER_DATABASE_URL or PYCHARTER_WORKER_DB_URL.
Connection errors: Ensure the database is reachable and migrations are applied: pycharter db current.
SQLite + multiple pollers: The worker logs an error when running multiple pollers against SQLite; use Postgres for production.

Jobs stay pending¶

Confirm at least one worker is running (pycharter worker --mode all).
Check worker logs for validation/pipeline errors.
Verify the same database URL is used by submitters and workers.

Retries and failures¶

Default max_attempts is 5. After exhaustion, the job is failed and a DLQ record is written when DLQ is enabled.
Inspect the job row, call get_job(job_id), or use the DLQ API for replay.