Run Soniq workers¶

A Soniq worker claims jobs from PostgreSQL, runs your handlers and records the result. Run workers with the soniq worker command from a process manager such as systemd, Kubernetes or supervisord. The manager restarts and scales them. The Worker class and soniq.run_worker() are documented here for advanced use cases (tests, embedding, custom orchestration).

soniq worker¶

soniq worker myapp.jobs:soniq  # import the module and use that Soniq instance
soniq worker myapp.jobs:soniq --concurrency 8 --queues emails,billing
soniq worker myapp.jobs:soniq --run-once  # process available jobs once and exit
SONIQ_JOBS_MODULES=myapp.jobs soniq worker  # no app path: name the job modules

The command reads SONIQ_DATABASE_URL and the other SONIQ_* settings from the environment. Without an app path, the command exits with an error unless SONIQ_JOBS_MODULES or --jobs-modules names the job modules.

See the CLI reference for all commands.

run_worker()¶

The in-process entry point. Use this for tests or when embedding Soniq inside a larger Python application that owns the lifecycle.

soniq = Soniq(database_url="postgresql://localhost/myapp")

await soniq.run_worker(
    concurrency=4,
    run_once=False,
    queues=None,
    shutdown_event=None,
)

Parameters¶

On PostgreSQL, run_worker() raises SoniqError with code SONIQ_POOL_TOO_SMALL when concurrency + pool_headroom is greater than pool_max_size. Raise pool_max_size or lower concurrency.

What happens during run_worker¶

1. The instance auto-initializes if needed (connects to the database, runs lazy setup).
2. A Worker instance is created with the instance's backend, job registry, settings, and hooks.
3. In continuous mode, the worker registers itself in the database, starts a heartbeat loop, subscribes to LISTEN/NOTIFY for instant job pickup, and launches concurrency processing tasks.
4. In run_once mode, the worker processes available jobs sequentially until the queue is empty, then returns.

Worker configuration via environment variables¶

These environment variables control worker behavior when using the CLI or default settings:

Worker class (advanced)¶

Most users never instantiate Worker directly. It is documented here for contributors and users who need fine-grained control.

from soniq.core.worker import Worker
from soniq.core.registry import JobRegistry
from soniq.settings import SoniqSettings

worker = Worker(
    backend=backend,       # A StorageBackend instance
    registry=registry,     # A JobRegistry with registered jobs
    settings=settings,     # SoniqSettings (optional, uses global defaults)
    hooks=hooks,           # Dict of hook lists (optional)
    middleware=None,       # List of middleware (optional)
)

Constructor parameters¶

run()¶

async def run(
    self,
    concurrency: int = 4,
    run_once: bool = False,
    queues: list[str] | None = None,
    shutdown_event: asyncio.Event | None = None,
) -> bool

Returns True if any jobs were processed. shutdown_event is ignored when run_once is True.

run_once()¶

async def run_once(
    self,
    queues: list[str] | None = None,
    max_jobs: int | None = None,
) -> bool

Process available jobs and return. Pass max_jobs to cap how many jobs are processed in one call.

Graceful shutdown¶

In continuous mode, the worker installs signal handlers for SIGINT, SIGTERM and SIGHUP. On receiving either signal:

1. The shutdown event is set and the worker stops taking new jobs.
2. Running jobs get up to shutdown_timeout seconds (default 30) to finish.
3. Async jobs still running after that are cancelled and put back in the queue with their attempt count kept. Sync handlers are not cancelled. They get an extra sync_handler_grace_seconds (default: the job_timeout value), and after that the worker keeps waiting for the thread.
4. The worker deregisters itself from the database.
5. The LISTEN/NOTIFY connection is released.

A second signal does not force an exit. To stop a stuck worker, send SIGKILL. Jobs run at least once, so make handlers idempotent.

LISTEN/NOTIFY¶

When the backend supports push notifications (PostgreSQL), the worker subscribes to the soniq_new_job channel. When a job is enqueued, the worker wakes up immediately instead of waiting for the next poll cycle. This keeps latency low without hammering the database with frequent polls.

If LISTEN/NOTIFY setup fails (for example behind PgBouncer in transaction mode), the worker falls back to polling at poll_interval.