Metadata-Version: 2.4
Name: quebec
Version: 0.4.0b1
Classifier: Programming Language :: Rust
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Programming Language :: Python :: Implementation :: PyPy
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Operating System :: Microsoft :: Windows
Classifier: Operating System :: POSIX
Classifier: Operating System :: Unix
Classifier: Operating System :: MacOS
Requires-Dist: sphinx>=7.0 ; extra == 'docs'
Requires-Dist: shibuya ; extra == 'docs'
Requires-Dist: myst-parser ; extra == 'docs'
Requires-Dist: pytest>=7.0.0 ; extra == 'test'
Requires-Dist: pytest-cov ; extra == 'test'
Requires-Dist: anyio>=4.0 ; extra == 'test'
Requires-Dist: trio>=0.26 ; extra == 'test'
Provides-Extra: docs
Provides-Extra: test
License-File: LICENSE
Summary: Quebec is a simple background task queue for processing asynchronous tasks.
Keywords: solid_queue,postgresql,mysql,sqlite,queue
Home-Page: https://github.com/ratazzi/quebec
Author-email: ratazzi <ratazzi.potts@gmail.com>
License-Expression: MIT
Requires-Python: >=3.9
Description-Content-Type: text/markdown; charset=UTF-8; variant=GFM
Project-URL: Documentation, https://github.com/ratazzi/quebec/README.md
Project-URL: Homepage, https://github.com/ratazzi/quebec
Project-URL: Repository, https://github.com/ratazzi/quebec

# Quebec

Quebec is a simple background task queue for processing asynchronous tasks. The name is derived from the NATO phonetic alphabet for "Q", representing "Queue".

This project is inspired by [Solid Queue](https://github.com/rails/solid_queue).

> [!NOTE]
> **Project status: Production-tested beta.** Quebec is actively maintained and has
> been running in the maintainers' own production projects since April 2026, with no
> known stability issues. Its core job-processing APIs are suitable for production
> deployments. The project remains pre-1.0, so pin the version you deploy and review
> changes before upgrading. APIs and configuration explicitly marked **experimental**
> may still change between minor releases.

## Why Quebec?

- **Simplified Architecture**: No dependencies on Redis or message queues
- **Database-Powered**: Leverages RDBMS capabilities for complex task queries and management
- **Rust Implementation**: High performance and safety with Python compatibility
- **Framework Agnostic**: Works with asyncio, Trio, threading, SQLAlchemy, Django, FastAPI, etc.

## Features

- Scheduled tasks
- Recurring tasks
- Concurrency control
- Batches (Solid Queue-compatible)
- Per-queue concurrency limits
- Rate limiting
- Exclusive (stop-the-world) jobs
- Multi-process (fork) mode
- Memory-based worker recycling
- cgroup v2 memory limits per worker (Linux)
- Web dashboard
- Automatic retries
- Signal handling & graceful restart
- Lifecycle hooks

### Control Plane

Built-in web dashboard for monitoring jobs, queues, and workers in real-time.

![Control Plane](docs/images/control-plane.png)

## Database Support

- SQLite
- PostgreSQL
- MySQL

### Upgrading an existing PostgreSQL deployment

On PostgreSQL only, Quebec no longer creates the `(key, value)` and
`(expires_at)` indexes on the semaphores table. They prevent HOT updates: every
concurrency wait/signal rewrites both `value` and `expires_at`, and indexing a
column that every `UPDATE` touches forces a new row version plus an index write
each time. Dead tuples then pile up faster than autovacuum can reclaim them, and
a table holding a handful of live rows can end up costing thousands of buffer
hits per statement.

Databases created before this version still carry both indexes.

`create_tables()` does not drop indexes, so run this once against an existing
database (substitute your table prefix):

```sql
DROP INDEX IF EXISTS idx_solid_queue_semaphores_key_value;
DROP INDEX IF EXISTS idx_solid_queue_semaphores_expires_at;

ALTER TABLE solid_queue_semaphores SET (
  fillfactor = 70,
  autovacuum_vacuum_scale_factor = 0,
  autovacuum_vacuum_threshold = 1000
);

-- Reclaims space already lost to bloat and applies the new fillfactor.
-- Takes an ACCESS EXCLUSIVE lock, so concurrency operations block for its
-- duration -- normally well under a second on a healthy table.
VACUUM FULL solid_queue_semaphores;
```

The unique index on `key` stays: it serves the only hot-path lookup
(`WHERE key = $1`), and PostgreSQL applies the `value` predicate as a filter
after it. Calling `create_tables()` afterwards re-applies the storage
parameters but never re-creates the dropped indexes.

The `expires_at` index only served `delete_expired`, which the dispatcher runs
once per `concurrency_maintenance_interval` (default 600s). That scan is a
sequential one now -- roughly 11 ms on a 50k-row table, against an index that
would otherwise be maintained on every write.

SQLite and MySQL keep both indexes and need no migration. InnoDB's undo-log
MVCC and SQLite's rollback journal do not accumulate heap bloat this way, so
there the indexes are a plain win -- `delete_expired` in particular gets to use
`(expires_at)` instead of scanning.

## Quick Start

### Module Runner (Recommended)

Define jobs in a package:

```python
# jobs/email_job.py
import quebec

class EmailJob(quebec.BaseClass):
    queue_as = "default"

    def perform(self, to, subject):
        self.logger.info(f"Sending email to {to}: {subject}")
```

Export them in `__init__.py`:

```python
# jobs/__init__.py
from .email_job import EmailJob
```

Run with `python -m quebec`:

```bash
DATABASE_URL=sqlite:///demo.db?mode=rwc python -m quebec jobs
```

All configuration via `QUEBEC_*` environment variables — no boilerplate entry script needed.

### Script Mode

For more control, use Quebec directly in a script:

```python
import logging
from pathlib import Path
from quebec.logger import setup_logging

setup_logging(level=logging.DEBUG)

import quebec

db_path = Path('demo.db')
qc = quebec.Quebec(f'sqlite://{db_path}?mode=rwc')


@qc.register_job
class FakeJob(quebec.BaseClass):
    def perform(self, *args, **kwargs):
        self.logger.info(f"Processing job {self.id}: args={args}, kwargs={kwargs}")


if __name__ == "__main__":
    # Enqueue a job (qc is inferred from @qc.register_job)
    FakeJob.perform_later(123, foo='bar')

    # Start Quebec (handles signal, spawns workers, runs main loop)
    qc.run(
        create_tables=not db_path.exists(),
        control_plane='127.0.0.1:5006',  # Optional: web dashboard
    )
```

Or run the quickstart script directly:

```bash
curl -O https://raw.githubusercontent.com/ratazzi/quebec/refs/heads/master/quickstart.py
uv run quickstart.py
```

### Auto-Discovering Jobs

If your jobs are organized in a package (e.g. `app.jobs.*`), call
`Quebec.discover_jobs()` instead of decorating each class with
`@qc.register_job` or calling `qc.register_job_class(...)` one by one:

```python
# app/jobs/cleanup.py
class CleanupJob(quebec.BaseClass):
    def perform(self, *args, **kwargs): ...

# main.py
qc = quebec.Quebec(dsn)
qc.discover_jobs("app.jobs", "worker.tasks")   # recursively scans each
qc.run()
```

`discover_jobs` takes one or more dotted package paths as positional
arguments (varargs) — no need to wrap a single package in a list.

`discover_jobs(*packages, recursive=True, on_error="raise")`:

- Registers every `BaseClass` subclass whose `__module__` falls under one of
  the given packages. Classes imported from elsewhere (e.g. `from
  some.lib import JobMixin`) are ignored.
- Raises `ValueError` if two discovered classes share the same
  `__qualname__`, since Quebec's worker registry is keyed by qualname and
  the later registration would otherwise silently replace the earlier one.
- `on_error="raise"` (default) propagates submodule `ImportError`. Pass
  `on_error="warn"` to emit a `RuntimeWarning` and keep scanning —
  useful when a package contains optional-integration modules that may
  fail to import in some environments. The top-level package is always
  imported strictly.

### Multiple Quebec Instances

Quebec is designed for one instance per process. Registering a job class
(via `@qc.register_job`, `qc.register_job_class`, or `qc.discover_jobs`)
binds it to that Quebec instance, so `MyJob.perform_later(...)` shorthand
routes to the binding. If a process holds more than one Quebec instance
and registers the same job class to each, the most recent registration
wins — pass the target instance explicitly to disambiguate:

```python
MyJob.perform_later(qc2, arg1)                  # route to qc2
MyJob.set(queue='critical').perform_later(qc2, arg1)
```

### `qc.run()` Options

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `create_tables` | `bool` | `False` | Create database tables (requires DDL permissions) |
| `control_plane` | `str` | `None` | Web dashboard address, e.g. `'127.0.0.1:5006'` |
| `spawn` | `list[str]` | `None` | Components to spawn: `['worker', 'dispatcher', 'scheduler']`. `None` = all |

Recommended: configure worker thread count in `queue.yml` via `workers.threads`.
If you need a one-off override, `Quebec(..., worker_threads=3)` is also supported.

### Multi-Process Mode (fork supervisor)

By default `qc.run()` runs all components as threads in a single process. To scale across CPU cores, set `QUEBEC_SUPERVISOR=1` to fork a pool of child processes instead:

```bash
QUEBEC_SUPERVISOR=1 python -m quebec your.jobs
```

```yaml
# queue.yml (under your environment, e.g. production:)
workers:
  - queues: "*"
    threads: 5
    processes: 4        # fork 4 worker processes
dispatchers:
  - polling_interval: 1
    processes: 1        # fork 1 dispatcher process
```

The supervisor forks `workers[].processes` worker children and `dispatchers[].processes` dispatcher children, each taking its config from the matching yml entry, and reforks any child that dies (matching Solid Queue's process model). Fork mode is opt-in via the env var so an existing config with `processes` set doesn't silently switch process model on upgrade; `spawn` is ignored in this mode. Outside supervisor mode the `processes` keys are ignored and Quebec uses the single-process threaded runtime.

On Linux the supervisor can also give each child a memory limit the kernel enforces — see [cgroup Memory Limits](#cgroup-memory-limits-linux-supervisor-mode).

### Force Queue Override (multi-branch development)

Set `QUEBEC_FORCE_OVERRIDE_QUEUE` to pin every enqueue and consumption to one queue — handy when several development branches share a single database:

```bash
QUEBEC_FORCE_OVERRIDE_QUEUE=branch_x python -m quebec your.jobs
```

Every enqueue path rewrites `queue_name` to this value (ignoring whatever the class, call site, or scheduler specified), and the worker only consumes that queue — so jobs enqueued by one branch are never picked up by another. URL-hostile characters and `*` in the name are sanitized to `-` (a literal `*` would otherwise be reinterpreted as a wildcard by the consuming worker).

### Transactional Enqueue

> [!IMPORTANT]
> **Enqueuing is _not_ part of your database transaction — even on the same database.** Quebec's enqueue runs through the Rust engine on its own connection pool, completely separate from your Python connection (SQLAlchemy / Django / psycopg). There is no way to atomically commit a business write and a job enqueue together.

This is the deliberate cost of keeping the engine fully decoupled from your ORM and connection — the upside is that Quebec drags no Python database dependencies into your app, but it means the enqueue cannot join your transaction. Two failure windows follow:

- The business transaction commits but the enqueue fails → **the job is lost**.
- The enqueue commits but the business transaction rolls back → **the job runs against missing or stale data**.

Recommendations:

- **Enqueue after your business transaction commits.** This removes the worse direction — a job running for a write that was rolled back.
- **Make jobs idempotent** and tolerant of data that may not be visible yet; lean on retries.
- If you genuinely need atomicity, use a **transactional outbox**: write an outbox row inside your own transaction (business + outbox commit atomically), then relay it into a real job (at-least-once delivery).

### Delayed Jobs

```python
from datetime import timedelta

# Run after 1 hour
FakeJob.set(wait=3600).perform_later(arg1)

# Run at specific time
FakeJob.set(wait_until=tomorrow_9am).perform_later(arg1)

# Override queue and priority
FakeJob.set(queue='critical', priority=1).perform_later(arg1)
```

### Automatic Retries

```python
from datetime import timedelta

class PaymentJob(quebec.BaseClass):
    retry_on = [
        quebec.RetryStrategy(
            (ConnectionError, TimeoutError),
            wait=timedelta(seconds=30),
            attempts=3,
        ),
        quebec.RetryStrategy(
            (ValueError,),
            wait=timedelta(seconds=5),
            attempts=1,
            # Called once retries are exhausted; receives (job, error).
            handler=lambda job, error: notify_admin(error),
        ),
    ]

    def perform(self, order_id):
        process_payment(order_id)
```

Multiple `RetryStrategy` entries can target different exception types with independent wait/attempts. The optional `handler` fires only when a strategy's attempts are exhausted (mirroring ActiveJob's `retry_on ... do |job, error|` block) and is called with `(job, error)`. `discard_on` and `rescue_from` handlers use the same `(job, error)` signature.

### Concurrency Control

Limit how many jobs with the same key can run simultaneously:

```python
class ReportJob(quebec.BaseClass):
    concurrency_limit = 3          # max 3 concurrent executions per key
    concurrency_duration = 120     # semaphore TTL in seconds

    def concurrency_key(self, account_id, **kwargs):
        return str(account_id)     # final key: "ReportJob/123"

    def perform(self, account_id):
        generate_report(account_id)
```

The actual concurrency key is `"ClassName/key"` (e.g. `"ReportJob/123"`), so different job classes never conflict. When the limit is reached, new jobs are blocked until a slot becomes available. The `concurrency_duration` acts as a safety TTL — the semaphore is released automatically if a worker crashes.

### Batches

Group jobs so you can track the set as a whole and run callbacks when it finishes. Batches follow Solid Queue's tables and semantics (`solid_queue_batches` / `solid_queue_batch_executions`, added in Solid Queue 1.5), so a batch started by Rails can be finished by Quebec and vice versa.

```python
class ImportRow(quebec.BaseClass):
    def perform(self, row):
        ...

class ImportDone(quebec.BaseClass):
    def perform(self):
        b = self.batch          # the batch that enqueued this callback
        print(f"{b.completed_jobs}/{b.total_jobs} imported, {b.failed_jobs} failed")

with qc.batch(description="import 42",
              on_success=ImportDone,                                # job class ...
              on_failure=AlertJob.set(queue="alerts").build("import"),  # ... or a built descriptor
              on_finish=ImportDone,
              user_id=42) as batch:                                # extra kwargs -> batch.metadata
    for row in rows:
        ImportRow.perform_later(qc, row)
    qc.perform_all_later([ImportRow.build(r) for r in more_rows])  # bulk enqueue joins too

batch.id, batch.status            # "enqueued"
batch.total_jobs, batch.pending_jobs, batch.completed_jobs, batch.failed_jobs, batch.progress_percentage
batch.reload()                    # refresh from the database
qc.find_batch(batch.id)           # or qc.find_batch_by_active_job_batch_id(uuid)
```

Every job enqueued inside the `with` block joins the batch; leaving the block starts it (an empty batch finishes right away). To add jobs later, including from one of the batch's own jobs, use `with batch.enqueue(): ...`. A member job can read `self.batch_id` / `self.batch`. Nested `with qc.batch()` blocks are independent batches.

Nested batch contexts on the same Quebec instance share an enqueue transaction, without savepoints. Jobs become visible to workers and completion checks start only after the outermost context commits. An exception escaping the **outermost block body** rolls back the transaction. If an inner block raises and the outer block catches it, the inner block's enqueues are kept and commit with the outer block. That failed inner block does not register a start: its batch remains pending until a later successful `batch.enqueue()` context or the dispatcher's stalled-batch sweep starts it. Errors from post-commit start/completion checks are collected and raised after all pending checks have been attempted; they do not undo already-committed enqueues.

SQLite's single connection is held for the duration of a batch transaction, so other threads' database operations wait for it. A regular thread pool does not inherit the batch ContextVars: do not wait inside the block for other threads to enqueue or query through the same Quebec instance, as that can deadlock. Prepare work outside the block and use `perform_all_later` inside it, or give independent producers their own batch contexts.

Enqueue hooks on registered batch callbacks run when the batch finishes, inside the callback transaction; `AbortEnqueue` skips that callback without preventing batch completion. Maintenance-only processes can finish batches without registering their callback classes: they enqueue the stored callback payload and log a warning. Python enqueue hooks and fresh class-level concurrency resolution require registration in the process finishing the batch; otherwise only serialized options and any legacy serialized concurrency fields are available.

- `on_success` runs when every job finished without failing, `on_failure` when at least one job exhausted its retries, and `on_finish` in either case. Callback jobs are enqueued when the batch finishes; their queue, priority and `wait` come from the `.set(...)` used when the batch was created.
- Counters count logical jobs: a job that retries and then succeeds is one `total_jobs`. Jobs discarded by `discard_on` or a concurrency `Discard` conflict count as completed. Manually retrying a failed job (`qc.retry_failed`) does not rejoin its batch.
- Adding to a finished batch raises `quebec.BatchAlreadyFinished`. Building a descriptor inside a batch does not keep it open: enqueue it before the batch finishes.

Completion is detected as jobs finish. The few cases that can't trigger it (a crash between a job's terminal write and its batch release, a bulk delete that cascaded a tracking row away, a callback enqueue that failed, a process that died before starting its batch) are repaired by the dispatcher's maintenance timer (`batch_maintenance: true` by default, sharing `concurrency_maintenance_interval`; disable it via queue.yml, `dispatcher_batch_maintenance=False` or `QUEBEC_DISPATCHER_BATCH_MAINTENANCE=false`). Without a dispatcher, call `qc.sweep_stalled_batches()` yourself. Succeeded batches older than `clear_finished_jobs_after` are cleared by the worker's periodic cleanup or `qc.clear_finished_batches()`; failed batches are kept, like failed jobs.

The control plane lists batches under **Batches** (with progress and a status filter), shows each batch's jobs and callbacks, and links a job's page to its batch.

`create_tables()` creates the batch tables and adds `jobs.batch_id` to an existing database. Against a Rails-managed database that predates the Solid Queue batches migration, jobs enqueue and run without batch bookkeeping and `qc.batch()` raises `RuntimeError` until the migration is applied.

### Rate Limiting (experimental)

Cap how many jobs run within a sliding time window, scoped per key:

```python
from datetime import timedelta

class ApiCallJob(quebec.BaseClass):
    rate_limit_max = 5                          # at most 5 runs...
    rate_limit_duration = timedelta(seconds=2)  # ...per rolling 2-second window
    rate_limit_on_throttle = quebec.RateLimitConflict.Reschedule  # default

    def rate_limit_key(self, region="us", **kwargs):
        return region                           # bucket key: "ApiCallJob/us"

    def perform(self, region="us"):
        call_external_api(region)
```

Like concurrency control, the bucket is `"ClassName/key"`, and `rate_limit_key` defaults to the class name when not overridden. `rate_limit_duration` must be a `datetime.timedelta` of at least one second. When the window is exhausted, `rate_limit_on_throttle` decides what happens: `Reschedule` (the default) pushes the job to a later run, while `Discard` drops it.

### Exclusive Jobs

Let an occasional memory-heavy job own the whole worker process while it runs:

```python
class RebuildSearchIndexJob(quebec.BaseClass):
    exclusive = True

    def perform(self):
        rebuild_index()                         # runs alone on this worker
```

When an `exclusive` job is claimed, the worker stops claiming new jobs, waits for any in-flight siblings to finish, then runs the exclusive job by itself before resuming normal claiming. The scope is the **current worker process** — it does not coordinate across separate worker processes; pair it with `concurrency_limit = 1` and a `concurrency_key` if you also need cluster-wide single-instance execution.

### Graceful Restart (quiet-then-exit)

Drain in-flight work and exit on a quiet signal, for zero-downtime rolling restarts:

```python
qc = quebec.Quebec(database_url="...", quiet_then_exit=True)
qc.run()
```

Sending `SIGUSR1` (or `SIGTSTP`) puts the worker into quiet mode: it stops claiming new jobs but keeps running until every in-flight job finishes, then exits cleanly — with no time limit (unlike the `SIGTERM` path, which is bounded by `shutdown_timeout`). The usual flow is: signal the old instance quiet, start a new instance, and the old one exits once drained. Opt-in (default off), and standalone-only — under the fork supervisor a self-exited child would just be reforked, so use a supervisor-level rolling restart there instead. Also settable via `QUEBEC_QUIET_THEN_EXIT=1`.

### Memory-Based Worker Recycling

Long-lived Python workers tend to hold onto RSS the interpreter never returns to the OS. Quebec can recycle a bloated worker by draining it and exiting with a dedicated code, leaving the actual restart to your process supervisor. It is configured by environment variables — there is no in-process restart:

```bash
QUEBEC_WORKER_MAX_RSS_MB=512                  # soft limit; unset = disabled
QUEBEC_WORKER_MEMORY_RECYCLE_CONFIRMATIONS=3  # consecutive over-limit samples before recycling (default)
QUEBEC_WORKER_MEMORY_CHECK_INTERVAL=5s        # how often RSS is sampled (default)
```

In supervisor mode each worker entry can set its own threshold, which overrides the environment variable for that entry:

```yaml
workers:
  - queues: "*"
    processes: 4
    memory_recycle_at: 512MiB   # or `max` / `0` to switch recycling off here
```

When a worker's RSS stays above the limit for that many consecutive samples, it enters quiet mode, stops claiming, drains its in-flight jobs (no time limit), and exits with code **75** — the planned-recycle code. The supervisor then relaunches a fresh process. Under the built-in fork supervisor (`QUEBEC_SUPERVISOR=1`) this refork is automatic; under systemd, `Restart=on-failure` relaunches the worker after the non-zero recycle exit:

```ini
# /etc/systemd/system/quebec-worker.service
[Service]
ExecStart=/usr/bin/python -m quebec your.jobs
Environment=QUEBEC_DATABASE_URL=postgresql://localhost/myapp
Environment=QUEBEC_WORKER_MAX_RSS_MB=512
Restart=on-failure

[Install]
WantedBy=multi-user.target
```

Exit code 75 is non-zero, so `Restart=on-failure` treats the planned recycle as a failure and relaunches the worker. If you'd rather not have planned recycles show up as failures (in `systemctl status` or the start-limit counter), add `SuccessExitStatus=75` together with `RestartForceExitStatus=75` — the former keeps 75 out of the failure tally, the latter still forces the restart.

Recycling is cooperative: it samples RSS and acts between jobs, so a single job that allocates faster than the sampling interval still takes the process past the limit. On Linux, pair it with a cgroup limit (below) to have the kernel stop that case outright.

### Per-Job Memory Metrics (Linux)

Quebec observes two different Linux signals during `perform()`:

- `minor_faults` / `major_faults` are native-thread activity counters from
  `getrusage(RUSAGE_THREAD)`. They are useful when investigating allocation and
  I/O behaviour, but are not converted to bytes and are not RSS.
- Process RSS is read at job start and end and sampled every 100 ms in between.
  This produces `process_rss_start`, `process_rss_peak`, `process_rss_end`, and
  `process_rss_peak_delta`. Shorter-lived peaks may fall between samples.

RSS belongs to the process, not a thread. Quebec marks a window
`process_rss_single_job=true` only in a supervisor-managed worker where either
`threads: 1` or the job is `exclusive`. Only those single-job windows enter the
per-class RSS aggregate. Other windows remain useful as process context but are
not presented as memory attributable to one job. Allocations in subprocesses are
not included in the worker's RSS. Even a single-job window is a sampled process
envelope: allocator reuse and worker-runtime activity can still affect it.

These are observability metrics, not enforcement. When one job must not exhaust
the host, give each worker process its own `memory.high` / `memory.max` — see
[cgroup Memory Limits](#cgroup-memory-limits-linux-supervisor-mode) below.

The observations appear on every `job.completed` log line and on
`execution.metric`. For offline analysis, record one CSV row per finished job:

```bash
kill -USR2 <worker pid>   # start recording; send again to stop
```

or from code: `qc.start_job_metrics(path=None)`, `qc.stop_job_metrics()`, `qc.toggle_job_metrics()`, `qc.job_metrics_path`. Under the fork supervisor the signal is forwarded to every worker child, and each child writes its own file. Columns:

```
ts_ms,pid,tid,jid,class,queue,status,duration_ms,minor_faults,major_faults,process_rss_start_kb,process_rss_peak_kb,process_rss_end_kb,process_rss_peak_delta_kb,process_rss_single_job,active_jobs
```

`active_jobs` shows how many jobs the process owned when the row was recorded.
Aggregate attributable samples with whatever reads CSV, e.g.

```sql
select class, count(*), max(process_rss_peak_delta_kb),
       quantile_cont(process_rss_peak_delta_kb, 0.95)
from 'quebec-job-metrics-*.csv'
where process_rss_single_job = true
group by class order by 3 desc;
```

Environment variables:

```bash
QUEBEC_JOB_METRICS_DIR=/var/log/quebec   # output dir for SIGUSR2 recordings (default: OS temp dir)
QUEBEC_JOB_METRICS_MAX_ROWS=100000       # recording stops itself after this many rows
QUEBEC_JOB_METRICS_MAX_SECONDS=3600      # ...or after this long
```

Each Quebec instance also keeps per-class aggregates since startup: count,
failures, duration, thread faults, and average / p50 / p95 / max of single-job
`process_rss_peak_delta_kb` samples with the jid of the largest job.
`qc.job_metrics_summary(reset=False)` returns them as a dict;
`reset=True` takes and clears the current snapshot atomically.
`qc.log_job_metrics_summary()` writes one `job_metrics.summary` log line per
class, and stopping a recording with `SIGUSR2` logs them too. Percentiles come
from a log2 histogram, so they are bucket upper bounds rather than exact values.

Rows are handed to a writer thread through a bounded queue and flushed every 5
seconds. If the writer falls behind, rows are dropped rather than blocking jobs.
The time limit also stops idle recordings. When a row/time limit automatically
ends a recording, its writer drains accepted rows and flushes in the background.
Normal shutdown and `qc.close()` wait for active and already-stopping recordings
to finish flushing; forced termination can still lose buffered rows.

**USDT probes.** The Linux extension module carries `quebec:job_start` and
`quebec:job_end`. They are a single `nop` until a tracer attaches. The end probe
exports the minor-fault delta and the sampled RSS peak delta; the RSS argument is
`-1` unless `process_rss_single_job` is true. `job_start` exports jid, class,
and queue as pointer/length pairs. `job_end` exports jid, class, success,
duration nanoseconds, minor faults, and the attributable RSS peak delta.
The strings are not NUL-terminated: in bpftrace read them as
`buf(argN, argN+1)` printed with `%r`, or `str(argN, argN+1 + 1)` (that
argument is a buffer size, so `str(argN, argN+1)` drops the last character).
### cgroup Memory Limits (Linux, supervisor mode)

RSS recycling reacts after the fact. A cgroup limit lets the kernel enforce the ceiling as it is hit: the job dies instead of the host, and because each child sits in a cgroup of its own, the kill is attributable — the job that was running is marked failed with the memory cause, rather than left to the heartbeat pruner as an anonymous crash.

```yaml
# queue.yml (under your environment, e.g. production:)
workers_pool_memory_max: 7GiB   # budget for all workers together
workers:
  - queues: "*"
    threads: 5
    processes: 4
    memory_recycle_at: 500MiB   # RSS soft limit (cooperative, see above)
    memory_max: 2GiB            # cgroup hard limit (kernel-enforced)
    memory_high: 1500MiB        # throttle + reclaim, no kill
    memory_swap_max: 0          # default once a memory limit is set
    memory_oom_group: true      # default once a memory limit is set
dispatchers:
  - processes: 1
    memory_max: 256MiB
```

The supervisor builds two tiers under its delegated cgroup:

```
root
├── control/
│   ├── supervisor/     the supervisor itself
│   ├── dispatcher-0/   limits from dispatchers[]
│   └── scheduler-0/    no limits (queue.yml has no scheduler section)
└── workers/            workers_pool_memory_max
    ├── worker-0/       limits from workers[]
    └── worker-1/
```

Worker limits may overcommit against the pool: the sum of `memory_max` can exceed `workers_pool_memory_max`, which then caps them collectively. That is the point of the pool — a spike that a single worker's own limit would not catch is still contained, and the resulting OOM picks a worker rather than the dispatcher or the supervisor. `control` deliberately holds no process of its own so the memory controller can be enabled for the control slots at all (cgroup v2 refuses to enable controllers for the children of a cgroup that has member processes).

`memory_oom_group=true` means a worker is killed as a unit, taking any subprocess a job forked with it. At the pool level it is off, so a pool-level OOM removes one worker rather than all of them. Every child gets a leaf whether or not it has limits, since that is what makes its counters attributable to it rather than to whichever process ran in the slot before it.

Sizes accept a plain byte count, binary suffixes (`512MiB`), decimal suffixes (`1GB`), or `max`. An explicit `max` is not the same as omitting the key: it asks the kernel for nothing, and — unlike a real limit — neither derives a `memory_max` nor pulls in the `memory_swap_max=0` / `memory_oom_group=true` companions. When `memory_max` is omitted it is derived as `memory_recycle_at × 1.5`, because `memory.current` includes page cache and so has to sit well above the RSS line the soft recycle watches; set it explicitly to override, and note that derivation only happens once a cgroup has actually been found, so an existing deployment that only sets `memory_recycle_at` still boots on a host without cgroups.

**Failure policy.** With no limits configured and no writable cgroup, Quebec logs one warning and runs exactly as it did before. Once any limit is configured — per-slot or the pool budget — a cgroup that cannot deliver it is a startup error instead: an unusable subtree, a failed `prepare`, a leaf that cannot be created, a child that cannot be migrated. Silently dropping the limit would leave you believing in a protection you do not have, and a worker that never reaches the `workers` subtree does not merely run unconstrained — it keeps running in the supervisor's own cgroup, outside the budget and beside the control processes. After the initial fleet is up, the same failure on a later refork takes down only that slot.

**Delegation.** Quebec needs a cgroup subtree it may write to: run as root, run under systemd with `Delegate=yes`, or point `QUEBEC_CGROUP_ROOT` at a subtree someone else delegated. Containers usually mount `/sys/fs/cgroup` read-only, which is enough for the metrics below but not for limits.

```ini
# /etc/systemd/system/quebec.service
[Service]
Type=notify
Delegate=yes
OOMScoreAdjust=-1000
WatchdogSec=30
ExecStart=/usr/bin/env QUEBEC_SUPERVISOR=1 python -m quebec your.jobs
Restart=on-failure
```

`OOMScoreAdjust=` is how the supervisor gets protected — Quebec never writes its own `oom_score_adj`, since lowering it requires `CAP_SYS_RESOURCE` and a self-write would only ever work for deployments that need it least. Whatever the unit grants is inherited by the whole tree, and only workers give it up: they run your code and must stay killable, or a pool OOM finds no valid target. The dispatcher and scheduler keep it, which is the point — a worker's memory spike must not take the dispatcher with it.

**What an OOM looks like.** After reaping a child the supervisor reads that leaf's `memory.events`. The cgroup's own `oom_kill` counter outranks any guess from the exit signal, because SIGKILL alone cannot separate a kernel OOM from a shutdown escalation or an operator's `kill -9`. The job is then failed with the counters as evidence:

```
Worker process killed by the OOM killer (pid=175017, oom_kill=2,
memory.max=134217728, memory.peak=134217728).
Likely exceeded this worker's memory.max.
```

The last sentence appears only when a `max` event or a peak that reached the limit actually implicates this worker's own limit — `memory.events` counts kills by any OOM killer, the global one included, so `oom_kill > 0` on its own does not prove it outgrew its own ceiling. Repeated OOMs count against the same crash-loop guard as ordinary crashes, so a `memory_max` too small to boot the interpreter disables the slot instead of fork-looping.

**Metrics work without delegation.** Reading a cgroup is independent of managing one: a process can always read its own counters even where `/sys/fs/cgroup` is mounted read-only, which is the normal case under Docker and Kubernetes. Workers publish `memory.current`, `memory.peak`, the configured max and high, the `oom_kill` / `high` / `max` event counts, and `cpu.stat` usage and throttle counts in their heartbeat metadata; the control plane's workers page shows usage against the limit and flags a throttled worker. Every reader degrades to nothing rather than failing — `memory.peak` only exists on kernels 5.19 and newer, `memory.events` keys come and go, and a non-cgroup host simply reports none of it.

Limits can also be retuned without a restart. The change is written to the live cgroup and remembered for the next fork, so it survives a refork:

```python
# In the supervisor process — a signal handler, or a lifecycle hook.
# `current_supervisor()` is None everywhere else, children included.
from quebec.supervisor import current_supervisor

current_supervisor().adjust_slot_limit("worker", 0, memory_max="3GiB")
```

Lowering `memory_max` below what the worker is already using starts reclaim immediately, and OOM-kills it if the kernel cannot shrink it that far.

Environment variables: `QUEBEC_CGROUP=0` turns the whole mechanism off, `QUEBEC_CGROUP_ROOT` names the delegated subtree, and `QUEBEC_WORKERS_POOL_MEMORY_MAX` sets the pool budget on hosts with no config file (queue.yml wins over it, the same way `memory_recycle_at` wins over `QUEBEC_WORKER_MAX_RSS_MB`).

Only cgroup **v2** is supported — v1 lacks `memory.oom.group` and `cgroup.kill`, which the attribution and cleanup paths rely on — and only memory is limited: `cpu.max` and `pids.max` are not written, though CPU usage and throttle counts are reported.

### Per-Queue Concurrency (experimental)

Cap how many jobs run concurrently across the cluster for specific queues, independent of per-class `concurrency_key`:

```python
qc = quebec.Quebec(
    database_url="...",
    experimental_queue_concurrency={"reports": 2, "exports": 1},
)
qc.run()
```

Each listed queue acquires a `queue:<name>` semaphore at claim time; queues not present are unlimited. Useful for isolating a misbehaving queue during remediation. Naming and semantics are experimental and may change.

### Global Priority Across Queues (experimental, off by default)

With `queues: "*"` and nothing to skip, a worker polls with a single unfiltered query, so `priority` orders jobs across every queue. That stops being possible as soon as a queue must be skipped — paused, or with a full `experimental_queue_concurrency` slot — because excluding queues with `NOT IN` cannot use an index. Quebec then does what Solid Queue does: one query per live queue, which makes queue order override `priority` until the queue is resumed.

Enabling this flag keeps one global order in that situation by polling with an `IN` list instead:

```sql
SELECT * FROM solid_queue_ready_executions
WHERE queue_name IN ($live_queues)
ORDER BY priority, job_id
LIMIT $batch
FOR UPDATE SKIP LOCKED;
```

```python
qc = quebec.Quebec(
    "postgresql://localhost/myapp",
    experimental_global_priority=True,   # or QUEBEC_EXPERIMENTAL_GLOBAL_PRIORITY=true
)
```

**Measure before enabling — this is not universally faster.** It moves the scan from the excluded side to the live side, and which one wins depends on where your backlog sits:

| Backlog distribution | Default (per-queue) | `experimental_global_priority` |
|---|---|---|
| Skipped queues hold most rows | fine — skipped rows never scanned | likely much better: one query, shallow live set |
| Live queues hold most rows | fine — each query is an index range | likely far worse: reads and sorts the live rows |
| Many live queues, all deep | fine | worst case |

PostgreSQL picks between two plans for the `IN` form, and the choice depends on statistics: walk `(queue_name, priority, job_id)` per listed queue and sort the union, or walk `(priority, job_id)` and filter. Check which one you get on real data:

```sql
EXPLAIN (ANALYZE, BUFFERS)
SELECT * FROM solid_queue_ready_executions
WHERE queue_name IN ('live_a', 'live_b')
ORDER BY priority, job_id
LIMIT 10
FOR UPDATE SKIP LOCKED;
```

Look at whether the queue_name index is used, how many rows feed the Sort, `Rows Removed by Filter`, and shared buffers. Note that `LockRows` sits *above* `Sort`, so `SKIP LOCKED` does not shrink the sort input.

Two things it does not change:

- It only ever applies to `*`. Explicit queue lists (`["real_time", "background"]`) and wildcard prefixes (`"beta*"`) keep Solid Queue's contract that queue order takes precedence over `priority`.
- Ordering and locking stay in one statement, so there is no window where the order is decided from rows another worker has already taken.

### Pausing Recurring Tasks (opt-in)

Recurring tasks can be paused and resumed at runtime — from Python or from the control plane's Recurring Jobs page — without editing `recurring.yml` or restarting the scheduler:

```python
qc = quebec.Quebec(db_url, recurring_pause=True)   # or QUEBEC_RECURRING_PAUSE=true

qc.pause_recurring("nightly_report")      # True; False if it was already paused
qc.recurring_paused("nightly_report")     # True
qc.paused_recurring_tasks()               # ["nightly_report"]
qc.resume_recurring("nightly_report")     # True; False if it was not paused
```

While paused, the scheduler skips each occurrence (nothing is enqueued, no `recurring_executions` row is written) and moves on to the next one. Resuming does not replay the skipped runs; the next occurrence after the resume fires as scheduled. `run_recurring_now()` still works on a paused task. Unknown keys raise `LookupError` — static tasks appear in the table once a scheduler has started.

Solid Queue has no such state, so this is the one place Quebec extends its schema: enabling `recurring_pause` adds a nullable `paused_at` column to the recurring tasks table. It is added automatically by `create_tables()` and when a scheduler starts, and each process checks for it on its own. It is off by default so an unmodified Solid Queue database keeps working exactly as before; with it off, the column is not added even if it exists elsewhere, and the pause API raises `RuntimeError`.

Sharing the database with a Rails app:

- Solid Queue reads and writes the table as usual and leaves `paused_at` alone, with one exception: its **scheduler** upserts every attribute of the static tasks when it boots, which resets `paused_at`. Solid Queue's scheduler also ignores the pause, so pausing only takes effect when Quebec runs the scheduler.
- If the connecting role is not allowed to `ALTER TABLE`, Quebec logs a warning and pausing is unavailable in that process until the column exists. Add it yourself in that case:

  ```ruby
  add_column :solid_queue_recurring_tasks, :paused_at, :datetime
  ```

  No restart is needed: a process whose attempt failed keeps looking for the column (a catalog lookup, at most every 5 seconds) and starts honouring pauses as soon as it appears. Explicit calls — `create_tables()`, the pause API, the control-plane buttons — retry the `ALTER` right away.

### TLS Configuration (PostgreSQL)

Quebec links `sqlx` against `rustls` + `webpki-roots`. Public CAs (AWS RDS,
Neon, Google Cloud SQL, Supabase, etc.) are trusted out of the box — no OS
trust store is consulted.

Pass libpq-style SSL options as `Quebec(...)` kwargs, as DSN query params, or
via `QUEBEC_SSL*` environment variables:

```python
qc = quebec.Quebec(
    "postgresql://user:pass@host:5432/db",
    sslmode="verify-full",             # or QUEBEC_SSLMODE
    sslrootcert="/etc/ssl/certs/ca.pem",  # internal CAs only
)
```

Priority is **kwargs > env > DSN query**. Passing any `ssl*` kwarg/env against
a non-postgres URL raises `ValueError`.

| `sslmode`     | Transport              | Certificate verification | Hostname verification |
|---------------|------------------------|--------------------------|-----------------------|
| `disable`     | plaintext              | —                        | —                     |
| `prefer`      | TLS if offered, else plaintext | —                | —                     |
| `require`     | TLS (fails if unsupported) | — (accepts any cert)  | —                     |
| `verify-ca`   | TLS                    | CA-signed                | —                     |
| `verify-full` | TLS                    | CA-signed                | hostname matches CN/SAN |

For public CAs, `verify-full` works zero-config. Use `sslrootcert` for
internal/self-signed CAs. `sslcert` + `sslkey` enable client certificate
(mTLS) auth.

> `sslmode=allow` is **rejected** with a `ValueError`. Upstream `sqlx-postgres`
> 0.8 treats `allow` identically to `disable` (plaintext, marked `FIXME` in
> the driver); to avoid a silent downgrade, Quebec refuses it. Use `prefer`
> for opportunistic TLS, or `require`/`verify-*` to enforce it.

> Note: some managed Postgres services (e.g. Neon) terminate TLS at a proxy
> layer. In those cases `pg_stat_ssl.ssl` may report `false` because the
> backend sees plaintext from the proxy — not the client.

## Lifecycle Hooks

Quebec provides several lifecycle hooks that you can use to execute code at different stages of the application lifecycle:

- `@qc.on_start`: Called when Quebec starts
- `@qc.on_stop`: Called when Quebec stops
- `@qc.on_worker_start`: Called when a worker starts
- `@qc.on_worker_stop`: Called when a worker stops
- `@qc.on_shutdown`: Called during graceful shutdown

These hooks are useful for:
- Initializing resources
- Cleaning up resources
- Logging application state
- Monitoring worker lifecycle
- Graceful shutdown handling

