Streaming Replication Events to Webhooks #

A downstream service needs to know the instant a CouchDB replication job flips from running to crashing, but CouchDB has no outbound webhook of its own — it only exposes state through the _scheduler API and the _changes feed on _replicator. This page shows how to build the missing piece: a worker that watches replication job state transitions and delivers signed, retryable JSON events to any HTTP receiver, with HMAC-SHA256 signatures, at-least-once delivery, idempotency keys, and a dead-letter file so nothing is silently dropped. It is the delivery-mechanics companion to Async Monitoring & Webhooks; that page explains how to read the event bus, this one explains how to safely ship what you read to somebody else’s endpoint.

Immediate Triage / Prerequisites #

Before wiring an outbound webhook, confirm you can observe replication state transitions at all, and decide which feed to watch. The _scheduler/jobs endpoint gives you a clean per-job state (running, pending, crashing, failed) that is ideal to diff between polls; the _changes feed on _replicator gives you the underlying document lifecycle if you would rather subscribe than poll. Confirm both respond:

# current scheduler state for every replication job (the thing we diff)
curl -s http://localhost:5984/_scheduler/jobs | \
  python3 -c "import sys,json;[print(j['id'],j['info'].get('error'),j.get('history',[{}])[0].get('type')) for j in json.load(sys.stdin)['jobs']]"

# the _replicator event bus, if you prefer a subscription over polling
curl -s "http://localhost:5984/_replicator/_changes?feed=normal&limit=3"

If _scheduler/jobs returns an empty jobs array, no replication is scheduled and there is nothing to stream yet. Prerequisites for the code below: Python 3.8+ with httpx (pip install httpx) for the async worker, admin credentials (the _scheduler and _replicator endpoints require them), and a target webhook URL plus a shared secret held in an environment variable. One rule to internalize before writing a single line: outbound webhook delivery is at-least-once, never exactly-once. Networks drop acknowledgements, so your receiver will occasionally see the same event twice — design for duplicates from the start with an idempotency key rather than pretending they will not happen.

Step-by-Step Implementation #

Follow these steps to detect a transition, sign it, deliver it with retries, and dead-letter what will not land. Each step includes a check so you can verify state before moving on.

1. Snapshot the current job states. Read _scheduler/jobs and build a map of job_id → state. The state is the most recent history entry’s type, which is the value that actually flips on a transition:

   import httpx
   r = httpx.get("http://localhost:5984/_scheduler/jobs", auth=("admin", "password"))
   states = {j["id"]: j["history"][0]["type"] for j in r.json()["jobs"]}
   assert all(isinstance(v, str) for v in states.values())  # e.g. "started", "crashed"

   Persist this map. The next poll diffs against it, so a transition is simply a key whose value changed.

2. Diff against the previous snapshot. A transition exists when a job’s new state differs from its stored state, or when a job appears or disappears. Emit one event per changed job — never one per poll — so a steady running job stays silent:

   transitions = [(jid, prev.get(jid), cur) for jid, cur in states.items() if prev.get(jid) != cur]

   Verify a quiet system produces an empty transitions list; a tight poll loop that emits on every tick is the most common cause of self-inflicted webhook floods.

3. Assign a stable idempotency key. Derive the key deterministically from the job identity, the new state, and the job’s through_seq (or the _changes seq) so a re-delivery of the same transition carries the same key. A random UUID here would defeat receiver-side deduplication:

   import hashlib
   key = hashlib.sha256(f"{job_id}|{new_state}|{through_seq}".encode()).hexdigest()[:32]

4. Sign the canonical body. Serialize the event once with sorted keys and no incidental whitespace, then compute HMAC-SHA256(secret, body). Send the hex digest in an X-Signature header and the key in X-Idempotency-Key; the receiver recomputes the HMAC over the raw bytes it received and compares in constant time.

5. Deliver with bounded retries, then dead-letter. POST the signed bytes. Treat 2xx as delivered and advance your cursor; treat 5xx, connection errors, and timeouts as retryable with capped exponential backoff plus jitter. Treat 4xx (other than 408/429) as a permanent receiver-side reject. When the attempt budget is exhausted, append the event to a dead-letter file instead of blocking the whole stream:

   # a healthy receiver acknowledges fast; measure it before trusting it inline
   curl -s -o /dev/null -w "%{http_code} %{time_total}s\n" -X POST https://receiver.example/hook -d '{}'

Complete Working Example #

The script below is a self-contained async worker. It polls _scheduler/jobs, diffs state per job, signs each transition with HMAC-SHA256, delivers it with capped exponential backoff and jitter, and dead-letters exhausted events to a JSON-lines file. Slow receivers cannot backpressure detection because delivery runs concurrently with a bounded semaphore, and the poll interval is deliberately unhurried to avoid hammering the scheduler.

import asyncio
import hashlib
import hmac
import json
import os
import random
import time

import httpx

COUCH_URL = os.environ.get("COUCH_URL", "http://localhost:5984")
COUCH_AUTH = (os.environ.get("COUCH_USER", "admin"), os.environ.get("COUCH_PASSWORD", "password"))
WEBHOOK_URL = os.environ.get("WEBHOOK_URL", "https://receiver.example/hook")
WEBHOOK_SECRET = os.environ.get("WEBHOOK_SECRET", "change-me").encode()  # never hard-code in prod
DEAD_LETTER = os.environ.get("DEAD_LETTER_PATH", "webhook_deadletter.jsonl")
POLL_SECONDS = 15          # unhurried: the scheduler is not a high-frequency bus
MAX_ATTEMPTS = 6
BASE_DELAY = 1.0
MAX_DELAY = 60.0
MAX_INFLIGHT = 4           # cap concurrent deliveries so a slow receiver cannot stall us


def sign(body: bytes) -> str:
    """Hex HMAC-SHA256 of the exact bytes we transmit."""
    return hmac.new(WEBHOOK_SECRET, body, hashlib.sha256).hexdigest()


def idempotency_key(job_id: str, state: str, through_seq) -> str:
    """Stable across re-deliveries of the same transition -> receiver can dedupe."""
    raw = f"{job_id}|{state}|{through_seq}".encode()
    return hashlib.sha256(raw).hexdigest()[:32]


def make_event(job_id: str, old: str, new: str, info: dict) -> dict:
    through_seq = info.get("through_seq") or info.get("changes_pending")
    return {
        "type": "replication.state_changed",
        "job_id": job_id,
        "from_state": old,
        "to_state": new,
        "error": info.get("error"),
        "through_seq": through_seq,
        "emitted_at": time.time(),
        "idempotency_key": idempotency_key(job_id, new, through_seq),
    }


def dead_letter(event: dict, reason: str) -> None:
    """Append, never overwrite: the file is a replayable audit log."""
    with open(DEAD_LETTER, "a", encoding="utf-8") as fh:
        fh.write(json.dumps({"reason": reason, "event": event}) + "\n")


async def deliver(client: httpx.AsyncClient, event: dict) -> None:
    """At-least-once delivery: retry 5xx/timeouts with capped backoff + jitter."""
    body = json.dumps(event, sort_keys=True, separators=(",", ":")).encode()
    headers = {
        "Content-Type": "application/json",
        "X-Signature": "sha256=" + sign(body),
        "X-Idempotency-Key": event["idempotency_key"],
    }
    for attempt in range(1, MAX_ATTEMPTS + 1):
        try:
            resp = await client.post(WEBHOOK_URL, content=body, headers=headers, timeout=10.0)
            if resp.status_code < 300:
                return  # acknowledged
            # 4xx that is not 408/429 is a permanent reject; do not retry forever
            if 400 <= resp.status_code < 500 and resp.status_code not in (408, 429):
                dead_letter(event, f"rejected {resp.status_code}")
                return
        except (httpx.TimeoutException, httpx.TransportError) as exc:
            if attempt == MAX_ATTEMPTS:
                dead_letter(event, f"transport: {exc!r}")
                return
        delay = min(BASE_DELAY * 2 ** (attempt - 1), MAX_DELAY)
        delay += random.uniform(0, delay * 0.25)  # jitter defeats synchronized retries
        await asyncio.sleep(delay)
    dead_letter(event, "attempts exhausted")


async def watch() -> None:
    prev: dict[str, str] = {}
    sem = asyncio.Semaphore(MAX_INFLIGHT)
    async with httpx.AsyncClient(auth=COUCH_AUTH) as client:
        while True:
            try:
                jobs = (await client.get(f"{COUCH_URL}/_scheduler/jobs", timeout=30)).json()["jobs"]
            except (httpx.TransportError, KeyError):
                await asyncio.sleep(POLL_SECONDS)
                continue
            cur = {j["id"]: j["history"][0]["type"] for j in jobs}
            info_by_id = {j["id"]: j["info"] or {} for j in jobs}
            tasks = []
            for job_id, state in cur.items():
                if prev.get(job_id) != state:  # one event per transition, not per poll
                    event = make_event(job_id, prev.get(job_id), state, info_by_id[job_id])

                    async def _bounded(ev=event):
                        async with sem:
                            await deliver(client, ev)

                    tasks.append(asyncio.create_task(_bounded()))
            if tasks:
                await asyncio.gather(*tasks)
            prev = cur
            await asyncio.sleep(POLL_SECONDS)


if __name__ == "__main__":
    asyncio.run(watch())

Run it and it stays silent while jobs hold steady, fires exactly one signed POST when a job changes state, retries a flaky receiver, and drops anything undeliverable into webhook_deadletter.jsonl for later replay — without ever blocking detection on a slow endpoint.

Gotchas & Edge Cases #

• Duplicate deliveries are guaranteed, not exceptional. Because a lost 2xx acknowledgement forces a retry that the receiver already processed, the same event arrives twice. The stable X-Idempotency-Key is the contract that lets the receiver treat the second copy as a no-op. Never key on wall-clock time or a random UUID.
• A slow receiver must not backpressure detection. If you await each delivery inline in the poll loop, one unreachable endpoint freezes state tracking for every job. Run deliveries concurrently behind a bounded semaphore, as the example does, so detection keeps pace regardless of receiver latency.
• Never sign the parsed object — sign the exact bytes. Recompute the HMAC over the raw request body on the receiver, before any JSON parsing. Re-serializing on the receiver can reorder keys or change whitespace and silently break every signature. Fix the serialization (sort_keys, tight separators) once, on the sender.
• Keep the secret out of the image and the logs. Load WEBHOOK_SECRET from the environment or a secrets store, never a committed default, and never log the body at a level that reaches shipped logs. Rotate it the same disciplined way you rotate replication credentials, covered in Rotating Replication Credentials Without Downtime.
• Avoid the tight-poll trap. Polling _scheduler/jobs every few hundred milliseconds turns a monitoring tool into a load generator and still emits at most one event per real transition. A 10–30s interval catches every state change that matters; for sub-second latency, subscribe to the _replicator _changes feed with feed=continuous instead of shrinking the poll interval.

Verification & Observability #

Confirm delivery works end to end at two levels: the transition is detected, and the receiver acknowledges. Force a transition by pausing a job, then watch the scheduler and the dead-letter file:

# force a state change to exercise the pipeline (job goes to a non-running state)
curl -s -X POST http://localhost:5984/_scheduler/jobs -H 'Content-Type: application/json' \
  -d '{}'  # or edit the _replicator doc's owner/target to trigger a re-schedule

# the dead-letter file should stay empty on a healthy receiver
wc -l webhook_deadletter.jsonl 2>/dev/null || echo "no dead letters yet"

# scheduler still reporting the job so the next transition is observable
curl -s http://localhost:5984/_scheduler/jobs | \
  python3 -c "import sys,json;[print(j['id'],j['history'][0]['type']) for j in json.load(sys.stdin)['jobs']]"

A healthy pipeline shows an empty dead-letter file and a receiver whose logs record each idempotency key exactly once even across retries. Emit two metrics from the worker: deliveries per state and dead-letters per minute. A rising dead-letter rate means the receiver is failing, not that replication is failing — page on it separately. To alert specifically on jobs falling behind rather than changing state, pair this with Alerting on Replication Lag with _scheduler/jobs; to detect a job that is running yet silently stalled, add checkpoint-progress detection from Monitoring CouchDB Replication Checkpoints via the API. For dashboards rather than push events, expose the same signals through Prometheus Metrics & Exporter Integration for CouchDB Replication.

FAQ #

Related #

• Async Monitoring & Webhooks
• Alerting on Replication Lag with _scheduler/jobs
• Monitoring CouchDB Replication Checkpoints via the API

Part of: Async Monitoring & Webhooks