pawpulse

Device vitals and alerting pipeline. Collars publish body temperature and activity over MQTT, telemetry is bridged into RabbitMQ with a topic exchange, and a worker raises alerts behind a bounded retry ladder that ends in a parking lot rather than an infinite requeue loop.

collars --MQTT--> mosquitto --ingest--> RabbitMQ topic exchange --> worker --> Redis
 QoS 1, LWT,                            routing key carries        prefetch 1,
 retained status                        the device id              retry ladder

Design decisions

Producers publish to an exchange, never to a queue. The routing key is vitals.<deviceId>.reading, so a new consumer interested in one device or one class of reading is a new binding, not a change to the publisher.

The work queue is a quorum queue. Classic mirrored queues are deprecated; quorum queues replicate through Raft and are the current default for anything that must survive a node failure.

Retry is bounded and delayed, and RabbitMQ has no native delayed delivery. It is built from two primitives: a queue with a message TTL, and a dead letter exchange. A failed message is published to a retry queue whose TTL is the delay; when it expires the queue dead-letters it back to the work queue. Three rungs at 5s, 30s and 120s, with the attempt count carried in a header.

After the last rung the message is parked, not retried. vitals.parking is terminal. Rejecting with requeue: true puts a poison message straight back at the head of the queue, where it fails again immediately and pins a consumer forever. src/triage.js inspects the parking lot and replays it once the underlying bug is fixed.

Consumers are idempotent. MQTT QoS 1 and RabbitMQ both deliver at least once, so every reading carries an id which is recorded in Redis with a TTL before the work is done. A redelivery is dropped rather than double-alerting.

Readings older than the last seen timestamp are ignored. With more than one consumer on a queue, ordering is no longer guaranteed, and an out-of-order reading would otherwise clear and re-raise an alert repeatedly. Comparing the device timestamp makes the handler order-insensitive.

prefetch(1) gives fair dispatch. Without it RabbitMQ hands messages out round robin regardless of how busy a consumer is, so one worker can be given a long backlog while another sits idle.

Offline detection uses Last Will and Testament, published retained, so current device state is available immediately to any new subscriber.

Running it

docker compose up -d
npm install
npm run topology    # declare exchanges, work queue, retry ladder, parking lot
npm run ingest      # MQTT -> RabbitMQ
npm run worker      # consume, alert, retry, park

DEVICE_ID=collar-001 npm run collar
DEVICE_ID=collar-002 npm run collar

Management UI at http://localhost:15672 (guest / guest).

Exercise the retry ladder with a reading the handler rejects:

RETRY_DELAYS=2000,3000,4000 npm run worker
docker compose exec mosquitto mosquitto_pub -t "pets/collar-888/vitals" \
  -m '{"id":"bad-1","deviceId":"collar-888","tempC":"hot","activity":1,"ts":1789000000000}'

npm run triage           # inspect the parking lot
node src/triage.js replay # requeue everything parked

retry 1 in 2000ms: tempC must be a number
retry 2 in 3000ms: tempC must be a number
retry 3 in 4000ms: tempC must be a number
parked after 3 attempts: tempC must be a number

Tracing

With an OTLP collector on localhost:4318:

OTEL_ENABLED=1 OTEL_SERVICE_NAME=pawpulse-ingest npm run ingest:traced
OTEL_ENABLED=1 OTEL_SERVICE_NAME=pawpulse-worker npm run worker:traced

Why the :traced scripts exist. This is an ESM project and the OpenTelemetry auto-instrumentations patch CommonJS require; in ESM they hook nothing unless the loader is registered first, which src/hook.mjs does via node --import. Without it you get TCP and DNS spans only - no amqplib spans, and no error explaining the absence.

Stack

Node.js · MQTT (Mosquitto) · RabbitMQ · Redis · OpenTelemetry · Docker Compose