broker

Operational code for the Cannabis Observer change-bus broker - the Redis Streams instance the cluster's three services publish to and consume from.

This repo owns the broker's tuning, its monitoring, and the cluster stream inventory. It owns no application logic and no data model. Nothing here is imported by any service; the services reach the broker over the network, by URL.

Path	What it is
deploy/redis-server.dropin.conf	The tracked statement of the broker's tuning - AOF persistence, noeviction, an explicit maxmemory cap. Not the mechanism on the node today: those settings live in /etc/redis/redis.conf and the drop-in slot holds tailnet ordering. Reconciled in broker#1 Phase 5 - see deploy/README.md
deploy/broker-bus-health.service / .timer	The periodic WARN-only health probe, every 10 minutes
src/broker/bus_health.py	The probe: memory headroom, per-stream XLEN against retention caps, last-entry age on groupless streams, XPENDING, DLQ depth, disk
docs/STREAMS.md	The cluster stream inventory - who produces, who consumes, which health primitive applies, and who drains each DLQ

Provenance

Every file here moved out of CannObserv/archiver under archiver#193 D6, tracked by broker#1 Phase 1.

Archiver operated the broker from the same VM it ran on (archiver#109). Once the broker moved to a neutral node, two of these artifacts began measuring the wrong machine: the drop-in's parity test asserted a path under /etc/systemd/system/ on archiver's host, and the probe's disk check - which exists for AOF headroom - reported archiver's disk. Splitting the repo is what makes them true again.

What stayed in archiver

The split is not clean, and the seam is worth knowing:

• collect_outbox_findings queries information.changes_outbox. Nothing broker-side about it; archiver keeps it and keeps a reduced bus-health timer to run it.
• collect_group_lag feeds archiver's dashboard bus panel. XPENDING against a remote broker is an ordinary client call.
• scripts/check_redis_floor.sh is a client-side assertion that the broker a service is about to talk to is Redis >= 7.0. It stays with each client.

The OOM seam, in both directions

deploy/redis-server.dropin.conf sets maxmemory-policy noeviction with an explicit cap. That converts memory pressure into bounded, instance-wide OOM command not allowed errors instead of a kernel OOM-kill of the whole broker. It is only safe because archiver's outbox publisher classifies that error as transient and retries through it, rather than dead-lettering valid events - _TRANSIENT_PUBLISH_ERRORS in CannObserv/archiver:src/core/changes/publisher.py.

The cap and that classification are one decision, and they now live in two repositories with no test spanning them. Each side names the other in a comment. Do not change either alone.

Whether Watcher and Replicator have an equivalent durable retry is their own property, and this repo does not assert it - see the Producer durability under OOM column in docs/STREAMS.md.

Development

Python >= 3.12, uv, pytest, ruff.

co-core resolves from a local wheelhouse (./.wheelhouse, gitignored), not PyPI. Populate it before uv sync/uv run or resolution fails:

set -a; . /etc/broker/.env; set +a   # GOOGLE_APPLICATION_CREDENTIALS=co-pypi-reader key
uv run --no-project --with 'google-cloud-storage>=2,<4' python scripts/sync_wheelhouse.py
uv sync --group dev
uv run pytest
uv run ruff check .

CI authenticates keyless via Workload Identity Federation instead - the read-scoped github-ci provider impersonating the objectViewer-only co-pypi-reader service account.