<div class="docs-content" data-render>
# 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.
</div>