Metadata-Version: 2.5
Name: swarmq
Version: 0.6.2
Summary: Async task manager for Python 3.14+ using RabbitMQ and Valkey
License-Expression: MIT
License-File: LICENSE
Requires-Python: >=3.14
Requires-Dist: aio-pika>=9
Requires-Dist: croniter>=2
Requires-Dist: msgpack>=1.0
Requires-Dist: msgspec>=0.18
Requires-Dist: structlog>=24
Requires-Dist: valkey>=6
Provides-Extra: dishka
Requires-Dist: dishka>=1.10.1; extra == 'dishka'
Provides-Extra: docs
Requires-Dist: mkdocs-llmstxt>=0.3; extra == 'docs'
Requires-Dist: mkdocs-material>=9.5; extra == 'docs'
Requires-Dist: mkdocs-static-i18n>=1.2; extra == 'docs'
Requires-Dist: mkdocstrings[python]>=0.26; extra == 'docs'
Requires-Dist: pymdown-extensions>=10.11; extra == 'docs'
Provides-Extra: fast
Requires-Dist: libvalkey>=4; extra == 'fast'
Requires-Dist: orjson; extra == 'fast'
Provides-Extra: metrics
Requires-Dist: prometheus-client; extra == 'metrics'
Provides-Extra: reload
Requires-Dist: watchfiles>=0.21; extra == 'reload'
Provides-Extra: test
Requires-Dist: dishka>=1.10.1; extra == 'test'
Requires-Dist: freezegun; extra == 'test'
Requires-Dist: httpx; extra == 'test'
Requires-Dist: hypothesis; extra == 'test'
Requires-Dist: mutmut; extra == 'test'
Requires-Dist: mypy; extra == 'test'
Requires-Dist: prometheus-client; extra == 'test'
Requires-Dist: psutil; extra == 'test'
Requires-Dist: pytest; extra == 'test'
Requires-Dist: pytest-asyncio; extra == 'test'
Requires-Dist: pytest-cov; extra == 'test'
Requires-Dist: pytest-rerunfailures; extra == 'test'
Requires-Dist: pytest-timeout; extra == 'test'
Requires-Dist: ruff; extra == 'test'
Description-Content-Type: text/markdown

# SwarmQ

Async-first Python task queue on RabbitMQ + Valkey.

> **Status:** 0.6.x on PyPI. Core, workflow primitives, operator
> features (priorities, expiry, burst, supervisor, autoscaling), CPU
> worker mode, metrics, health, hot reload and the machine-readable agent
> spec are all shipped. About 2 700 tests across unit / integration /
> e2e / chaos / property / fuzz / stress suites, plus a separate load-test
> suite that gates every release with conformance, smoke and one-hour
> reference runs. Not yet in production use by anyone but the author.

## Quick Start

```bash
# 1. Start RabbitMQ + Valkey (compose file lives in this repo)
docker compose up -d

# 2. Define a task
cat > myapp/tasks.py <<'EOF'
from swarmq import Task, TaskInfo

class SendEmail(Task):
    task_info = TaskInfo(name="send_email", queues=["default"])

    async def run(self, to: str, subject: str) -> str:
        # ... your code here ...
        return f"sent to {to}"
EOF

# 3. Tell SwarmQ where the broker and backend live
export SWARMQ_BROKER_URL="amqp://guest:guest@localhost:5672/"
export SWARMQ_BACKEND_URL="valkey://localhost:6379/0"

# 4. Run a worker
swarmq worker --module myapp.tasks --queues default --concurrency 10
```

The worker handles SIGTERM / SIGINT for graceful shutdown. Multiple
queues: `--queues default,emails,priority`. Structured logs:
`--log-format json`.

## Hello World — Client side

```python
import asyncio
from swarmq import SwarmQ, RabbitMQBroker, ValkeyBackend
import myapp.tasks  # noqa — auto-registers SendEmail

async def main():
    app = SwarmQ(
        RabbitMQBroker("amqp://guest:guest@localhost:5672/"),
        ValkeyBackend("valkey://localhost:6379/0"),
        queues=["default"],
    )
    async with app:
        task_id = await app.schedule(
            "send_email", to="op@example.com", subject="hi",
        )
        result = await app.get_result(task_id, timeout=10)
        print(result)  # sent to op@example.com

asyncio.run(main())
```

## Reusing an existing connection pool (producer)

If your app already holds its own RabbitMQ connection and/or Valkey
client, hand them to SwarmQ instead of a URL — SwarmQ reuses them rather
than opening a second, parallel connection to the same server. You can
pass either a high-level client/connection **or** a low-level pool, and
mix sources per backend:

```python
import aio_pika
import valkey.asyncio
from swarmq import SwarmQ, RabbitMQBroker, ValkeyBackend

# objects your app already owns
connection = await aio_pika.connect_robust("amqp://guest:guest@localhost:5672/")
valkey_client = valkey.asyncio.Valkey.from_url("valkey://localhost:6379/0")

app = SwarmQ(
    RabbitMQBroker(connection=connection),   # or connection_pool=<aio_pika.pool.Pool>
    ValkeyBackend(client=valkey_client),     # or pool=<valkey ConnectionPool>
    # no broker_url / backend_url needed when a connection object is injected
)
async with app:
    await app.schedule("send_email", to="op@example.com", subject="hi")
# closing `app` does NOT close `connection` or `valkey_client` — your app keeps them
```

Things to know:

- **Ownership:** SwarmQ never closes a connection/client/pool you inject —
  it belongs to your app. SwarmQ does close the publish channel it opens
  on a borrowed RabbitMQ connection (that channel is SwarmQ's).
- **Producer-only:** injection is for the task-*sending* path. Calling
  `start_worker()` on an injected broker/backend raises
  `ConfigurationError` — workers are spawned as subprocesses and
  reconstruct their broker/backend from `SWARMQ_BROKER_URL` /
  `SWARMQ_BACKEND_URL`, which a borrowed connection can't cross.
- **Robustness is yours:** an injected Valkey client/pool does **not**
  inherit SwarmQ's retry / health-check policy, and an injected aio-pika
  connection should be a robust one (`connect_robust`) for SwarmQ's
  channel-reconnect handling to behave as designed.
- **Same namespace assumed:** the injected object must point at the same
  server and logical namespace (RabbitMQ vhost, Valkey DB) the rest of
  your config expects — SwarmQ does not validate this, and a mismatch
  silently sends/reads tasks in the wrong namespace.

## What's in the box

### Core

- **At-least-once delivery** with Quorum-Queue durability and
  `x-delivery-limit=20` against pathological redelivery loops.
- **Retries** with exponential / linear / fixed backoff +
  configurable `max_retries`, `NoRetry`, `Retry(delay=...)`.
- **DLQ** routing for retry-exhaustion + `swarmq dlq list/inspect/
  retry/purge` CLI for operator recovery.
- **Middleware** stack with 6 hooks (pre/post enqueue, pre/post
  execute, on_error, on_retry).
- **Signals** — 23 lifecycle events with `@app.on_signal(...)`
  decorator, from `TASK_ENQUEUED` over `TASK_DLQ` / `TASK_REJECTED` to
  worker, scheduler and workflow-step events.
- **Rate limiting** — Cloudflare 2-period sliding window via Lua
  (`TaskInfo(rate_limit="100/m")`).
- **Locking** — mutex + semaphore via Valkey atomic primitives, with
  format-string keys (`lock="user:{user_id}"`).
- **Unique tasks** — dedup by canonical-encoded args, with
  `unique_until="start"` or `"completion"`.
- **Cancellation** — pre-pickup + during-execution via
  `Client.cancel(task_id)`, dedicated `TASK_CANCELLED` signal.
- **Scheduler** — cron, delayed (`eta=...`), recurring with
  leader-election failover.
- **Progress tracking** — `Task.update_progress(current, total, msg)`
  + `Client.get_progress` / `Client.watch_progress` AsyncIterator.
- **CLI** — `swarmq worker`, `swarmq schedule`, `swarmq dlq *`,
  `swarmq inspect`.

### Workflow primitives

- **`chain` / `group` / `chord`** with `Signature` building blocks
  (`Task.s()`, `sig("name")`, immutable `Task.si()`) and operator
  syntax (`A.s() | B.s()`, `group(...) | merge.s()`). Flat-DAG
  composition with construction-time limits (`max_workflow_depth=10`,
  `max_group_size=1000`).
- **Fire-and-forget `apply()`** returns a reattachable handle:
  `await app.apply(chain(...))` → `handle.get(timeout=...)`, `cancel()`,
  `children`, stable `workflow_id` reattach via `app.workflow(id)`.
- **Worker-driven orchestration** — workers advance the workflow on each
  task completion (chain result-injection, atomic+idempotent chord
  fan-in via Lua). Fail-fast with liveness: a failed step marks the
  workflow failed and wakes `get()` instead of hanging.

```python
result = await (await app.apply(
    chord(group(Download.s(u) for u in urls), Merge.s())
)).get(timeout=30)
```

### Operations

- **Supervisor by default** — every `swarmq worker` runs under a
  process manager, also with `--processes 1`: it restarts crashed
  workers into the same slot (5/hr rolling-window budget), replaces
  workers whose heartbeat stops, reloads on SIGHUP via `os.execvp`
  (replaying the `sys.argv` snapshot captured at start — the container
  CMD or systemd ExecStart stays the trust boundary for argv), and
  takes its children down with it even on SIGKILL. Configuration errors
  end the run with exit code `78` on every level; connection errors stay
  at `1`.
- **Multi-process workers** (`--processes N`) with a stable slot
  identity: `worker_id` is `<host>:<slot>` in signals, metrics and logs,
  and `--metrics-port` / `--health-port` are base ports — slot `s`
  binds `base + s`, so scrape targets survive restarts.
- **Autoscaling** (`--autoscale=MIN,MAX`) — queue-depth-driven
  scale up/down with 30s cooldown, 60s idle window, and a 30s
  minimum worker age that suppresses spawn→immediate-kill churn
  when a crashed worker is restarted during an idle period.
- **CPU worker mode** (`--pool cpu`) — task bodies run in compute
  children forked from a pre-warmed forkserver; the loop process never
  computes, `hard_timeout` and cancel kill the child's process group,
  DI children are recycled by RSS or task count. POSIX only.
- **Priorities** (`--priorities`, `TaskInfo.priority`), **expiry**
  (`TaskInfo.max_age` → `TASK_EXPIRED` + DLQ) and **burst mode**
  (`--burst --burst-idle-timeout N` drains and exits 0).
- **`--configure MODULE:CALLABLE`** — register middleware and signal
  handlers on the CLI worker without writing your own entry point.
- **Prometheus metrics** (`pip install swarmq[metrics]`) — 10 metrics
  exposed on `/metrics`, drop-in no-op when the dep is absent.
- **Health endpoints** — `/health` (liveness) + `/ready` (broker +
  backend reachable, forkserver in CPU mode, structured 503 body).
- **Hot reload** (`--reload`, `pip install swarmq[reload]`) — file-
  watch trigger that fires the existing SIGHUP drain-and-execvpe
  pipeline. Developer-only (WARNING log on every start); see
  `website/guides/hot-reload.md`.

### Reliability hardening (E2E + Chaos suites)

- **Channel-recreate** survives RabbitMQ broker restarts mid-publish.
- **Pub/Sub resubscribe** survives Valkey restarts mid-stream.
- **Toxiproxy chaos suite** verifies behavior under network latency,
  bandwidth limits, slicer (packet loss), full disconnect, and
  partial partitions.
- **Property-based tests** on dedup-hash, serialization, backoff,
  rate-limiter; **fuzz-tests** on message parsers and external
  publishers.

## Configuration

Worker behaviour is configured via env vars and CLI flags (CLI flag
wins where both are set).

| Variable / flag | Default | Purpose |
|---|---|---|
| `SWARMQ_BROKER_URL` | required | AMQP URL for RabbitMQ |
| `SWARMQ_BACKEND_URL` | required | Valkey/Redis URL for results |
| `--queues a,b,c` | `default` | Queues to consume |
| `--concurrency N` | `10` | Max parallel tasks per worker |
| `--log-level` | `INFO` | DEBUG / INFO / WARNING / ERROR |
| `--log-format` | `human` | `human` or `json` |

Other flags: `--processes`, `--autoscale`, `--pool`, `--metrics-port`,
`--health-port`, `--burst`, `--priorities`, `--configure`, `--check`.
Every flag has an env-var twin (`SWARMQ_QUEUES`, `SWARMQ_CONCURRENCY`,
`SWARMQ_POOL`, …); the full list with defaults is in the
configuration reference (`website/reference/configuration.md`) and the
CLI reference (`website/reference/cli.md`).

## Status and roadmap

Everything listed above is shipped and covered by tests; the
changelog (`website/reference/changelog.md`) records each release. A
machine-readable agent spec — generated JSON for errors, config, CLI
and message schemas plus an `llms.txt` index under
`website/reference/agent/spec/` — is kept in sync with the code by tests
(`python -m swarmq.agent_spec`).

Still open:

- **Bulk-throughput optimisation** — a `schedule_many` fast path
  (design note in `docs/brainstorms/implementation-order-requirements.md`).
- **24-hour acceptance run** of the load-test suite before the first
  production deployment, and the resulting SLO thresholds for p95
  latency and throughput.

## Project layout

```
src/swarmq/
  app.py             SwarmQ application
  worker.py          Consumer loop
  client.py          Schedule API + result retrieval
  task.py            Task base + TaskInfo
  middleware.py      Middleware base + chain
  signals.py         Signal enum + dispatcher
  limits.py          RateLimiter + parser
  locking.py         Mutex / semaphore
  cancellation.py    Cancel-flag protocol
  progress.py        ProgressInfo (R3.7)
  scheduler.py       Embedded cron + delayed scheduler
  metrics.py         Prometheus exporter (R5.1)
  health.py          /health + /ready (R5.2)
  cli.py             argparse entrypoint
  lua/               Lua scripts loaded into Valkey
  broker/rabbitmq.py
  backend/valkey.py
tests/
  unit/              ~895 fast, no infra
  integration/       broker + backend, real Docker
  e2e/               full client + worker round-trip
  chaos/             toxiproxy + container restarts (opt-in, nightly)
  property/          hypothesis property-based (opt-in)
  fuzz/              byte + JSON fuzz (opt-in)
  stress/            high-parallelism (opt-in)
  soak/              1h leak detection (opt-in, weekly)
docs/
  brainstorms/       requirements (per feature)
  plans/             implementation plans (per feature)
  solutions/         postmortem-style learnings
  testing/           test strategy + feature-parity matrix
```

## Development

```bash
# Fast dev loop — unit only
uv run --extra test pytest tests/unit -q

# E2E suite (needs docker compose up -d)
SWARMQ_TEST_RABBITMQ_PORT=5673 SWARMQ_TEST_VALKEY_PORT=6380 \
  uv run --extra test pytest tests/e2e -q

# Chaos suite (starts its own toxiproxy stack per test class)
uv run --extra test pytest tests/chaos -q -m chaos

# E2E feature-combinations suite (31 combos × 3 failure tiers)
# Tier-1 only — no Toxiproxy needed, runs as part of the regular e2e
# suite above. Tier-2/3 (chaos-marked) needs the Toxiproxy stack:
docker compose -f docker-compose.yml -f docker-compose.chaos.yml up -d
uv run --extra test pytest tests/e2e/combinations -v -m chaos

# Property + fuzz
uv run --extra test pytest tests/property tests/fuzz -q -m "property or fuzz"
```

Three-tier CI in `.github/workflows/`: pre-merge (≤5 min, every PR),
nightly (chaos + stress + fuzz), weekly (mutation + 1h soak).

For coding conventions and TDD discipline see `AGENTS.md`.

## License

[MIT](LICENSE).
