# Async job services The async job framework (`services/jobs/`) runs heavy, blocking work off the request path and hands the caller a result URL when it finishes. It builds on the [service framework](/docs/services-framework.html): a `JobService` is a `BaseService` whose `run_once` drives a queue of database-backed jobs. [ZipService](/docs/services-zip.html) is the first consumer; the current kinds are `zip` (ZipService), `fork` (ForkService), `seo` (SeoService), `deepsearch` (DeepsearchService), and `seo_meta` (SeoMetaService). > Audience: administrators and maintainers. ## When to use it Use a job service whenever a request would otherwise block on slow work: archiving, exporting, rendering, batch processing, or anything that should survive a page reload and be polled. The request enqueues a row and returns immediately; a background worker does the work; the client polls for completion. The shared `JobPoller.run(statusUrl, { onDone, onFailed, onTimeout })` helper (`static/js/JobPoller.js`) drives that poll on the client. ## The jobs table All job kinds share one `jobs` table, discriminated by a `kind` column. The row is the queue. Columns: `uid` (a sortable uuid7, also the job id), `kind`, `status` (`pending`, `running`, `done`, `failed`), `owner_kind`/`owner_id` (attribution only), `preferred_name`, `payload` and `result` (JSON strings), `error`, `retry_count`, the timestamps `created_at`/`started_at`/`completed_at`/`updated_at`, `duration_ms`, `last_accessed_at`/`expires_at` (retention), and the common stat columns `bytes_in`/`bytes_out`/`item_count`. Kind-specific fields live inside the `result` JSON. Indexes cover `(kind, status)`, `(owner_kind, owner_id)`, and `expires_at`. ## Enqueue from any worker `services/jobs/queue.py` is pure database access and is safe to call from any worker's request handler: - `enqueue(kind, payload, owner_kind, owner_id, preferred_name)` inserts a `pending` row and returns its uid. - `get_job(uid)` returns the hydrated job (payload and result decoded). - `touch_job(uid, extend_seconds)` bumps `last_accessed_at` and extends `expires_at`; the download path calls this so an actively-used artifact is not pruned. - `list_jobs(kind, status, owner)` backs metrics and the CLI. ## Single-owner processing Processing happens only in the worker that owns the service lock, since only that worker calls `supervise()` and runs `run_once()`. With exactly one processor across the deployment, no cross-worker locking or atomic claim is needed. Status reads and downloads work from any worker because they read the shared database and the shared data directory (`config.DATA_DIR`, outside the package and not under `/static`; downloads are served by a `FileResponse` route, not the static mount). `JobService.run_once` does four things each tick: 1. **Reap** finished in-flight tasks, writing `result`, timing, and final status. 2. **Recover orphans**: any `running` row not in this process's in-flight map (necessarily left by a previous owner) is reset to `pending` with a bounded `retry_count`, or marked `failed` past the limit. 3. **Refill**: claim the oldest `pending` jobs up to the concurrency limit (uuid7 sorts in creation order, giving FIFO), mark them `running`, and spawn an `asyncio` task per job. 4. **Sweep**: delete jobs whose `expires_at` has passed, calling the subclass `cleanup` hook to remove the artifact first. In-flight tasks span ticks: stored in an instance map and reaped when done, so the loop returns quickly each tick. A per-tick timeout guard recovers a `running` row whose task vanished. ## Writing a new kind Subclass `JobService`, set `kind`, and implement two methods: ```python class MyJobService(JobService): kind = "mykind" title = "My jobs" async def process(self, job): # do the work; return a dict stored as result JSON. # include bytes_in / bytes_out / item_count for the shared stats. return {"download_url": f"/.../{job['uid']}/download", "local_path": "...", "item_count": n} def cleanup(self, job): # remove the artifact named in job["result"]. ... ``` Register it in `main.py` startup with `service_manager.register(MyJobService())`. It inherits the retention, concurrency, and timeout config fields (keyed `{name}_retention_seconds`, `{name}_max_concurrent`, `{name}_job_timeout_seconds`), the aggregate metrics card, crash recovery, and automatic pruning. Add enqueue endpoints that own their authorization, a status route, and a download route, then document the kind under the Services section. ## Retention is built in There is no separate reaper service. Each job service prunes its own expired artifacts every tick through its `cleanup` hook, so every kind gets deletion for free. Retention is admin-configurable per service and defaults to seven days; downloading an artifact extends its expiry.