Module adcp.webhook_supervisor_pg
PostgreSQL-backed :class:WebhookDeliverySupervisor for a durable retry queue.
Gives multi-instance AdCP sellers a shared circuit-breaker state and a retry queue that survives process crashes. It does not retain completed delivery proofs for the AdCP 3.2 retry horizon and does not atomically commit task terminal state with its queue, so it must not back advertised TaskHandoff push publication.
Rubric: REQUIRED: run_worker()
Unlike :class:~adcp.webhook_supervisor.InMemoryWebhookDeliverySupervisor,
:meth:PgWebhookDeliverySupervisor.send_mcp() does not deliver
immediately — it enqueues the job and returns None. You MUST start
:meth:run_worker somewhere in your application; without it enqueued jobs
accumulate and are never sent:
::
from psycopg_pool import AsyncConnectionPool
from adcp.webhook_supervisor_pg import PgWebhookDeliverySupervisor
pool = AsyncConnectionPool("postgresql://...", min_size=4, max_size=20)
supervisor = PgWebhookDeliverySupervisor(pool=pool, sender=my_sender)
await supervisor.create_schema() # idempotent; call once per boot
asyncio.create_task(supervisor.run_worker()) # REQUIRED before serve()
# Explicit/manual delivery only; SDK-managed TaskHandoff push remains off.
serve(my_handler, webhook_supervisor=supervisor, auto_emit_task_webhooks=False)
Rubric: Observability
send_mcp always returns None. For per-attempt outcomes configure a
:class:~adcp.webhook_supervisor.DeliveryLogSink or query the
adcp_webhook_delivery_log table directly.
Rubric: Cross-process guarantees
- Circuit breaker state — shared via
adcp_webhook_circuit_state(one row per breaker_key). Half-open probe atomicity is enforced viaUPDATE … RETURNINGso two concurrent workers can't both transition toclosedfrom a single success. - Sequence numbers — the
BIGSERIALqueue-row id is the sequence number (monotonically increasing, crash-durable across restarts). - Worker leasing —
SELECT … FOR UPDATE SKIP LOCKED LIMIT 1prevents simultaneous delivery by multiple healthy workers. A crash after receiver acceptance but before local commit remains ambiguous and may retry.
Rubric: DDL for migration tools
::
-- adcp_webhook_circuit_state
CREATE TABLE adcp_webhook_circuit_state (
breaker_key TEXT COLLATE "C" NOT NULL PRIMARY KEY,
state TEXT NOT NULL DEFAULT 'closed',
failure_count INT NOT NULL DEFAULT 0,
success_count INT NOT NULL DEFAULT 0,
opened_at TIMESTAMPTZ,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
-- adcp_webhook_delivery_queue
CREATE TABLE adcp_webhook_delivery_queue (
id BIGSERIAL PRIMARY KEY,
breaker_key TEXT NOT NULL,
url TEXT NOT NULL,
task_id TEXT NOT NULL,
task_type TEXT,
status_str TEXT NOT NULL DEFAULT 'pending',
result_json TEXT,
token TEXT,
sequence_key TEXT,
attempt_count INT NOT NULL DEFAULT 0,
max_attempts INT NOT NULL DEFAULT 3,
scheduled_at TIMESTAMPTZ NOT NULL DEFAULT now(),
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
idempotency_key TEXT,
sent_body BYTEA,
notification_type TEXT
);
CREATE INDEX adcp_webhook_delivery_queue_work_idx
ON adcp_webhook_delivery_queue (status_str, scheduled_at)
WHERE status_str IN ('pending', 'retry');
-- adcp_webhook_delivery_log
CREATE TABLE adcp_webhook_delivery_log (
id BIGSERIAL PRIMARY KEY,
queue_id BIGINT,
breaker_key TEXT NOT NULL,
url TEXT NOT NULL,
task_id TEXT NOT NULL,
sequence_key TEXT,
sequence_number BIGINT,
attempt_number INT NOT NULL,
max_attempts INT NOT NULL,
outcome TEXT NOT NULL,
http_status_code INT,
error_message TEXT,
response_time_ms INT NOT NULL DEFAULT 0,
occurred_at TIMESTAMPTZ NOT NULL,
will_retry BOOLEAN NOT NULL DEFAULT false,
next_retry_at TIMESTAMPTZ,
task_type TEXT,
payload_size_bytes INT,
notification_type TEXT
);
CREATE INDEX adcp_webhook_delivery_log_queue_id_idx
ON adcp_webhook_delivery_log (queue_id);
CREATE INDEX adcp_webhook_delivery_log_task_id_idx
ON adcp_webhook_delivery_log (task_id);
Rubric: Stranded-job note
Jobs remain in status_str = 'pending' (via rollback) if a worker
process crashes mid-send. No separate recovery sweep is needed — the next
worker poll picks them up automatically. Jobs that raised a Python exception
on the final attempt are deleted from the queue and logged with
outcome = 'failure'.
Classes
class PgWebhookDeliverySupervisor (pool: Any,
sender: WebhookSender | None,
*,
retry: RetryPolicy | None = None,
circuit: CircuitBreakerPolicy | None = None,
log_sink: DeliveryLogSink | None = None,
circuit_table: str = 'adcp_webhook_circuit_state',
queue_table: str = 'adcp_webhook_delivery_queue',
log_table: str = 'adcp_webhook_delivery_log')-
Expand source code
class PgWebhookDeliverySupervisor: """Postgres-backed :class:`~adcp.webhook_supervisor.WebhookDeliverySupervisor`. Parameters ---------- pool: A ``psycopg_pool.AsyncConnectionPool`` owned by the caller. Each operation acquires a short-lived connection (or holds one for the duration of a delivery). We don't open, own, or close the pool. sender: The underlying :class:`~adcp.webhook_sender.WebhookSender` used for the actual HTTP-Signatures POST. Must be non-None. retry: Retry policy (default: 3 attempts, exponential backoff with jitter). circuit: Circuit-breaker tuning (default: 5 failures open, 60s recovery, 2 successes close). log_sink: Optional :class:`~adcp.webhook_supervisor.DeliveryLogSink` called after each attempt. Failures are swallowed. Use for custom observability pipelines in addition to the built-in log table. circuit_table / queue_table / log_table: Override table names (ASCII only, 1–63 chars). Useful for multi- tenant schemas. Defaults are ``adcp_webhook_circuit_state``, ``adcp_webhook_delivery_queue``, and ``adcp_webhook_delivery_log``. Concurrency ----------- Safe to share across asyncio tasks in a single process. Multiple processes/pods each running :meth:`run_worker` are explicitly supported via ``FOR UPDATE SKIP LOCKED``. """ # The queue survives process loss, but successful/finally-failed rows are # removed and therefore do not yet retain an immutable delivery binding # for the complete AdCP 3.2 advertised horizon. delivery_state_is_durable = False delivery_retry_horizon_seconds: int | None = None def __init__( self, pool: Any, # psycopg_pool.AsyncConnectionPool; Any avoids import at runtime sender: WebhookSender | None, *, retry: RetryPolicy | None = None, circuit: CircuitBreakerPolicy | None = None, log_sink: DeliveryLogSink | None = None, circuit_table: str = DEFAULT_CIRCUIT_TABLE, queue_table: str = DEFAULT_QUEUE_TABLE, log_table: str = DEFAULT_LOG_TABLE, ) -> None: if not PG_AVAILABLE: raise ImportError(_INSTALL_HINT) if sender is None: raise ValueError( "PgWebhookDeliverySupervisor requires a non-None WebhookSender. " "Construct one via WebhookSender.from_jwk(...) or " "WebhookSender.from_pem(...) and pass it as the first positional argument." ) self._pool = pool self._sender = sender self._retry = retry or RetryPolicy() self._circuit_policy = circuit or CircuitBreakerPolicy() self._log_sink = log_sink ct = _safe_identifier(circuit_table) qt = _safe_identifier(queue_table) lt = _safe_identifier(log_table) self._circuit_t = ct self._queue_t = qt self._log_t = lt # Pre-format SQL at construction time to avoid per-call f-string overhead # and to bake in the validated table names once. self._sql_circuit_get = ( f"SELECT state, opened_at FROM {ct} WHERE breaker_key = %s" # noqa: S608 ) self._sql_circuit_set_half_open = ( f"UPDATE {ct} SET state = 'half_open', success_count = 0, " # noqa: S608 f"updated_at = now() WHERE breaker_key = %s AND state = 'open'" ) # Atomic upsert: increment failure_count; open circuit when threshold crossed. # {ct}.column_name in DO UPDATE SET refers to the *existing* row's value # (before the update), which is what Postgres requires for self-referential # ON CONFLICT expressions. # Coerce thresholds via ``int()`` before f-string interpolation — # an adopter passing an ``IntEnum`` / ``bool`` / custom ``__index__`` # subclass for ``failure_threshold`` would format verbatim into # SQL otherwise. Coercion produces a plain int whose ``str()`` # is byte-safe for static SQL emission. failure_t = int(self._circuit_policy.failure_threshold) self._sql_circuit_failure = ( f"INSERT INTO {ct} " # noqa: S608 f"(breaker_key, state, failure_count, success_count, updated_at) " f"VALUES (%s, 'closed', 1, 0, now()) " f"ON CONFLICT (breaker_key) DO UPDATE SET " f" failure_count = {ct}.failure_count + 1, " f" success_count = 0, " f" state = CASE " f" WHEN {ct}.failure_count + 1 >= {failure_t} OR {ct}.state = 'half_open' " f" THEN 'open' ELSE {ct}.state END, " f" opened_at = CASE " f" WHEN ({ct}.failure_count + 1 >= {failure_t} OR {ct}.state = 'half_open') " f" AND {ct}.state != 'open' THEN now() " f" ELSE {ct}.opened_at END, " f" updated_at = now()" ) # Atomic upsert: increment success_count; close circuit when threshold crossed. # RETURNING lets us read the post-update state without a second query, which # eliminates the half-open race: two concurrent workers see different final # counts and only the worker whose UPDATE produces count >= threshold # transitions to 'closed'. success_t = int(self._circuit_policy.success_threshold) self._sql_circuit_success = ( f"INSERT INTO {ct} " # noqa: S608 f"(breaker_key, state, failure_count, success_count, updated_at) " f"VALUES (%s, 'closed', 0, 0, now()) " f"ON CONFLICT (breaker_key) DO UPDATE SET " f" failure_count = 0, " f" success_count = CASE " f" WHEN {ct}.state = 'half_open' THEN {ct}.success_count + 1 ELSE 0 END, " f" state = CASE " f" WHEN {ct}.state = 'half_open' AND {ct}.success_count + 1 >= {success_t} " f" THEN 'closed' ELSE {ct}.state END, " f" updated_at = now() " f"RETURNING state, success_count" ) self._sql_enqueue = ( f"INSERT INTO {qt} " # noqa: S608 f"(breaker_key, url, task_id, task_type, status_str, result_json, " f"token, sequence_key, max_attempts, notification_type, operation_id) " f"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id" ) self._sql_poll = ( f"SELECT id, breaker_key, url, task_id, task_type, status_str, " # noqa: S608 f"result_json, token, sequence_key, attempt_count, max_attempts, " f"idempotency_key, sent_body, notification_type, operation_id " f"FROM {qt} " f"WHERE status_str IN ('pending', 'retry') AND scheduled_at <= now() " f"ORDER BY scheduled_at LIMIT 1 FOR UPDATE SKIP LOCKED" ) self._sql_delete_job = f"DELETE FROM {qt} WHERE id = %s" # noqa: S608 self._sql_reschedule = ( f"UPDATE {qt} SET " # noqa: S608 f" status_str = 'retry', " f" attempt_count = %s, " f" scheduled_at = %s, " f" sent_body = %s, " f" idempotency_key = %s " f"WHERE id = %s" ) self._sql_log_insert = ( f"INSERT INTO {lt} " # noqa: S608 f"(queue_id, breaker_key, url, task_id, sequence_key, sequence_number, " f"attempt_number, max_attempts, outcome, http_status_code, error_message, " f"response_time_ms, occurred_at, will_retry, next_retry_at, task_type, " f"payload_size_bytes, notification_type) " f"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)" ) self._worker_started = False self._worker_warned = False # ------------------------------------------------------------------ schema async def create_schema(self) -> None: """Bootstrap all three tables and their indexes. Idempotent. Safe to call on every app boot. Each DDL statement is executed separately — psycopg does not split on ``;``. :: pool = AsyncConnectionPool("postgresql://...") supervisor = PgWebhookDeliverySupervisor(pool=pool, sender=sender) await supervisor.create_schema() """ ct, qt, lt = self._circuit_t, self._queue_t, self._log_t statements = [ f"""CREATE TABLE IF NOT EXISTS {ct} ( breaker_key TEXT COLLATE "C" NOT NULL PRIMARY KEY, state TEXT NOT NULL DEFAULT 'closed', failure_count INT NOT NULL DEFAULT 0, success_count INT NOT NULL DEFAULT 0, opened_at TIMESTAMPTZ, updated_at TIMESTAMPTZ NOT NULL DEFAULT now() )""", f"""CREATE TABLE IF NOT EXISTS {qt} ( id BIGSERIAL PRIMARY KEY, breaker_key TEXT NOT NULL, url TEXT NOT NULL, task_id TEXT NOT NULL, task_type TEXT, status_str TEXT NOT NULL DEFAULT 'pending', result_json TEXT, token TEXT, sequence_key TEXT, attempt_count INT NOT NULL DEFAULT 0, max_attempts INT NOT NULL DEFAULT 3, scheduled_at TIMESTAMPTZ NOT NULL DEFAULT now(), created_at TIMESTAMPTZ NOT NULL DEFAULT now(), idempotency_key TEXT, sent_body BYTEA, notification_type TEXT, operation_id TEXT )""", # Backfill the operation_id column on tables created before the # async-completion-webhook operation_id echo landed. ADD COLUMN # IF NOT EXISTS is a no-op on fresh tables (the column is already # in the CREATE above) and an in-place add on pre-existing ones — # so a CREATE-only deployment and an upgrading one converge. f"ALTER TABLE {qt} ADD COLUMN IF NOT EXISTS operation_id TEXT", # Partial index on work-eligible rows; avoids scanning completed/in-flight rows. f"""CREATE INDEX IF NOT EXISTS {qt}_work_idx ON {qt} (status_str, scheduled_at) WHERE status_str IN ('pending', 'retry')""", f"""CREATE TABLE IF NOT EXISTS {lt} ( id BIGSERIAL PRIMARY KEY, queue_id BIGINT, breaker_key TEXT NOT NULL, url TEXT NOT NULL, task_id TEXT NOT NULL, sequence_key TEXT, sequence_number BIGINT, attempt_number INT NOT NULL, max_attempts INT NOT NULL, outcome TEXT NOT NULL, http_status_code INT, error_message TEXT, response_time_ms INT NOT NULL DEFAULT 0, occurred_at TIMESTAMPTZ NOT NULL, will_retry BOOLEAN NOT NULL DEFAULT false, next_retry_at TIMESTAMPTZ, task_type TEXT, payload_size_bytes INT, notification_type TEXT )""", f"CREATE INDEX IF NOT EXISTS {lt}_queue_id_idx ON {lt} (queue_id)", f"CREATE INDEX IF NOT EXISTS {lt}_task_id_idx ON {lt} (task_id)", ] async with self._pool.connection() as conn: for stmt in statements: await conn.execute(stmt) # ----------------------------------------------------------------- send async def send_mcp( self, *, url: str, task_id: str, status: GeneratedTaskStatus | str, task_type: TaskType | str, result: Any = None, operation_id: str, token: str | None = None, sequence_key: str | None = None, breaker_key: str | None = None, notification_type: str | None = None, ) -> WebhookDeliveryResult | None: """Enqueue one MCP-style webhook delivery; **always returns None**. The actual HTTP send happens in :meth:`run_worker`. This method writes the job to ``adcp_webhook_delivery_queue`` and returns immediately. ``None`` is returned whether the circuit is open (delivery skipped) or the job was successfully enqueued. For delivery outcomes use a :class:`~adcp.webhook_supervisor.DeliveryLogSink` or query the ``adcp_webhook_delivery_log`` table directly. :param breaker_key: Override the circuit-breaker lookup key (default: ``url``). Multi-tenant sellers whose buyers share a SaaS receiver URL MUST pass a tenant-scoped key (e.g. ``f"{tenant_id}:{url}"``). :param operation_id: Buyer-supplied correlation id echoed verbatim into the webhook payload per spec. Persisted on the queue row and replayed to :meth:`WebhookSender.send_mcp` when the worker delivers the job. :param notification_type: Passed through to the delivery log for delivery-report webhooks (``scheduled`` / ``final`` / etc.). """ if not self._worker_started and not self._worker_warned: self._worker_warned = True logger.warning( "[adcp.webhook_supervisor_pg] send_mcp() called before run_worker() " "has been started. Deliveries will be queued but not sent until a worker " "is running. Call asyncio.create_task(supervisor.run_worker()) at startup." ) # The queue column is TEXT; psycopg would otherwise bind a TaskType # enum via repr (`"TaskType.create_media_buy"`), corrupting the wire # value. Normalize once here so every persistence + log site sees # the on-wire string. task_type_str: str = task_type.value if isinstance(task_type, TaskType) else task_type bkey = breaker_key or url # Check circuit state; reject immediately if OPEN within the timeout window. async with self._pool.connection() as conn: cur = await conn.execute(self._sql_circuit_get, (bkey,)) row = await cur.fetchone() if row is not None: state: str = row[0] opened_at: datetime | None = row[1] if state == "open" and opened_at is not None: if opened_at.tzinfo is None: opened_at = opened_at.replace(tzinfo=UTC) elapsed = (datetime.now(UTC) - opened_at).total_seconds() if elapsed < self._circuit_policy.open_timeout_seconds: occurred_at = datetime.now(UTC) attempt = DeliveryAttempt( url=url, sequence_key=sequence_key, sequence_number=None, attempt_number=0, max_attempts=self._retry.max_attempts, outcome="circuit_open", http_status_code=None, error_message=f"circuit breaker OPEN for {bkey} — skipped delivery", response_time_ms=0, occurred_at=occurred_at, will_retry=False, next_retry_at=None, task_type=task_type_str, task_id=task_id, payload_size_bytes=None, notification_type=notification_type, ) await self._log_circuit_open(bkey, attempt) await self._call_sink(attempt) logger.warning( "[adcp.webhook_supervisor_pg] circuit OPEN for %s — skipped %s", bkey, task_type_str, ) return None # Open timeout elapsed; transition to half_open so next worker probes. async with self._pool.connection() as conn: await conn.execute(self._sql_circuit_set_half_open, (bkey,)) status_str = status if isinstance(status, str) else str(status) result_json = json.dumps(result) if result is not None else None async with self._pool.connection() as conn: cur = await conn.execute( self._sql_enqueue, ( bkey, url, task_id, task_type_str, status_str, result_json, token, sequence_key, self._retry.max_attempts, notification_type, operation_id, ), ) enqueue_row = await cur.fetchone() queue_id = enqueue_row[0] if enqueue_row else None logger.debug( "[adcp.webhook_supervisor_pg] enqueued %s → %s (queue_id=%s)", task_id, url, queue_id, ) return None # ----------------------------------------------------------------- worker async def run_worker(self, *, poll_interval: float = 0.5) -> None: """Poll the delivery queue with ``FOR UPDATE SKIP LOCKED``; runs forever. **REQUIRED** — start at app startup:: asyncio.create_task(supervisor.run_worker()) Multiple processes or coroutines can run concurrently; ``SKIP LOCKED`` ensures each job is processed by exactly one worker. The DB connection is held for the duration of the HTTP send so a crashed worker automatically releases the job back to the queue via transaction rollback (no separate recovery sweep needed). :param poll_interval: Seconds to sleep when the queue is empty. """ self._worker_started = True logger.info( "[adcp.webhook_supervisor_pg] worker started (poll_interval=%.2fs)", poll_interval, ) while True: try: delivered = await self._poll_and_process() if not delivered: await asyncio.sleep(poll_interval) except asyncio.CancelledError: logger.info("[adcp.webhook_supervisor_pg] worker cancelled — shut down") raise except Exception: logger.exception( "[adcp.webhook_supervisor_pg] worker error; will retry after %.2fs", poll_interval, ) await asyncio.sleep(poll_interval) # ---------------------------------------------------------- internal helpers async def _poll_and_process(self) -> bool: """Lease one job from the queue, send it, update state. Returns True if processed.""" # The connection stays open for the duration of the HTTP send so the # FOR UPDATE lock is held throughout. On crash (or CancelledError), # the transaction rolls back and the job returns to 'pending'. async with self._pool.connection() as conn: cur = await conn.execute(self._sql_poll) row = await cur.fetchone() if row is None: return False ( queue_id, bkey, url, task_id, task_type, status_str, result_json, token, sequence_key, attempt_count, max_attempts, idempotency_key, sent_body, notification_type, operation_id, ) = row attempt_number = attempt_count + 1 will_retry = attempt_number < max_attempts # Deferred import avoids circular module dependency at package init. from adcp.webhook_sender import WebhookDeliveryResult # noqa: PLC0415 t0 = time.monotonic() occurred_at = datetime.now(UTC) delivery_result: WebhookDeliveryResult | None = None exc_caught: BaseException | None = None try: if sent_body: # Spec-compliant retry: replay the exact bytes so the receiver # can dedup via the same idempotency_key (per mcp-webhook-payload.json: # "Publishers MUST … reuse the same key on every retry"). prev = WebhookDeliveryResult( status_code=0, idempotency_key=idempotency_key or "", url=url, response_headers={}, response_body=b"", sent_body=sent_body, ) delivery_result = await self._sender.resend(prev) else: result_obj = json.loads(result_json) if result_json else None delivery_result = await self._sender.send_mcp( url=url, task_id=task_id, status=status_str, task_type=task_type, result=result_obj, operation_id=operation_id, token=token, ) except Exception as exc: exc_caught = exc response_time_ms = int((time.monotonic() - t0) * 1000) success = delivery_result is not None and delivery_result.ok if success: assert delivery_result is not None # narrowing for mypy await conn.execute(self._sql_delete_job, (queue_id,)) await conn.execute(self._sql_circuit_success, (bkey,)) attempt = DeliveryAttempt( url=url, sequence_key=sequence_key, sequence_number=queue_id, attempt_number=attempt_number, max_attempts=max_attempts, outcome="success", http_status_code=delivery_result.status_code, error_message=None, response_time_ms=response_time_ms, occurred_at=occurred_at, will_retry=False, next_retry_at=None, task_type=task_type, task_id=task_id, payload_size_bytes=( len(delivery_result.sent_body) if delivery_result.sent_body else None ), notification_type=notification_type, ) else: await conn.execute(self._sql_circuit_failure, (bkey,)) next_delay = ( self._retry.delay_for_attempt(attempt_number + 1) if will_retry else None ) next_retry_at = ( occurred_at + timedelta(seconds=next_delay) if next_delay is not None else None ) if will_retry: stored_body = delivery_result.sent_body if delivery_result is not None else None stored_ikey = ( delivery_result.idempotency_key if delivery_result is not None else None ) await conn.execute( self._sql_reschedule, ( attempt_number, next_retry_at or occurred_at, stored_body, stored_ikey, queue_id, ), ) else: await conn.execute(self._sql_delete_job, (queue_id,)) if delivery_result is not None: err_msg: str | None = ( f"HTTP {delivery_result.status_code}: " f"{delivery_result.response_body[:200]!r}" ) http_status: int | None = delivery_result.status_code psize: int | None = ( len(delivery_result.sent_body) if delivery_result.sent_body else None ) elif exc_caught is not None: err_msg = f"{type(exc_caught).__name__}: {exc_caught}" http_status = None psize = None else: err_msg = None http_status = None psize = None attempt = DeliveryAttempt( url=url, sequence_key=sequence_key, sequence_number=queue_id, attempt_number=attempt_number, max_attempts=max_attempts, outcome="failure", http_status_code=http_status, error_message=err_msg, response_time_ms=response_time_ms, occurred_at=occurred_at, will_retry=will_retry, next_retry_at=next_retry_at, task_type=task_type, task_id=task_id, payload_size_bytes=psize, notification_type=notification_type, ) await self._log_attempt_via_conn(conn, queue_id, bkey, attempt) # connection __aexit__ commits here; lock released # Sink is called outside the transaction so a slow/broken sink # cannot interfere with the queue update. await self._call_sink(attempt) if exc_caught is not None and not will_retry: raise exc_caught # propagate to run_worker's exception logger return True async def _log_attempt_via_conn( self, conn: Any, queue_id: int | None, bkey: str, attempt: DeliveryAttempt, ) -> None: """Write a delivery attempt row using an already-open connection.""" try: await conn.execute( self._sql_log_insert, ( queue_id, bkey, attempt.url, attempt.task_id, attempt.sequence_key, attempt.sequence_number, attempt.attempt_number, attempt.max_attempts, attempt.outcome, attempt.http_status_code, attempt.error_message, attempt.response_time_ms, attempt.occurred_at, attempt.will_retry, attempt.next_retry_at, attempt.task_type, attempt.payload_size_bytes, attempt.notification_type, ), ) except Exception: logger.warning( "[adcp.webhook_supervisor_pg] failed to log attempt for %s — delivery unaffected", attempt.url, exc_info=True, ) async def _log_circuit_open(self, bkey: str, attempt: DeliveryAttempt) -> None: """Write a circuit_open attempt row using a fresh connection.""" try: async with self._pool.connection() as conn: await self._log_attempt_via_conn(conn, None, bkey, attempt) except Exception: logger.warning( "[adcp.webhook_supervisor_pg] failed to log circuit_open for %s", attempt.url, exc_info=True, ) async def _call_sink(self, attempt: DeliveryAttempt) -> None: if self._log_sink is None: return try: await asyncio.wait_for( self._log_sink.record(attempt), timeout=self._retry.sink_timeout_seconds, ) except asyncio.TimeoutError: logger.warning( "[adcp.webhook_supervisor_pg] DeliveryLogSink timed out for %s", attempt.url, ) except Exception: logger.warning( "[adcp.webhook_supervisor_pg] DeliveryLogSink raised for %s", attempt.url, exc_info=True, )Postgres-backed :class:
~adcp.webhook_supervisor.WebhookDeliverySupervisor.Parameters
pool: A
psycopg_pool.AsyncConnectionPoolowned by the caller. Each operation acquires a short-lived connection (or holds one for the duration of a delivery). We don't open, own, or close the pool. sender: The underlying :class:~adcp.webhook_sender.WebhookSenderused for the actual HTTP-Signatures POST. Must be non-None. retry: Retry policy (default: 3 attempts, exponential backoff with jitter). circuit: Circuit-breaker tuning (default: 5 failures open, 60s recovery, 2 successes close). log_sink: Optional :class:~adcp.webhook_supervisor.DeliveryLogSinkcalled after each attempt. Failures are swallowed. Use for custom observability pipelines in addition to the built-in log table. circuit_table / queue_table / log_table: Override table names (ASCII only, 1–63 chars). Useful for multi- tenant schemas. Defaults areadcp_webhook_circuit_state,adcp_webhook_delivery_queue, andadcp_webhook_delivery_log.Concurrency
Safe to share across asyncio tasks in a single process. Multiple processes/pods each running :meth:
run_workerare explicitly supported viaFOR UPDATE SKIP LOCKED.Class variables
var delivery_retry_horizon_seconds : int | Nonevar delivery_state_is_durable
Methods
async def create_schema(self) ‑> None-
Expand source code
async def create_schema(self) -> None: """Bootstrap all three tables and their indexes. Idempotent. Safe to call on every app boot. Each DDL statement is executed separately — psycopg does not split on ``;``. :: pool = AsyncConnectionPool("postgresql://...") supervisor = PgWebhookDeliverySupervisor(pool=pool, sender=sender) await supervisor.create_schema() """ ct, qt, lt = self._circuit_t, self._queue_t, self._log_t statements = [ f"""CREATE TABLE IF NOT EXISTS {ct} ( breaker_key TEXT COLLATE "C" NOT NULL PRIMARY KEY, state TEXT NOT NULL DEFAULT 'closed', failure_count INT NOT NULL DEFAULT 0, success_count INT NOT NULL DEFAULT 0, opened_at TIMESTAMPTZ, updated_at TIMESTAMPTZ NOT NULL DEFAULT now() )""", f"""CREATE TABLE IF NOT EXISTS {qt} ( id BIGSERIAL PRIMARY KEY, breaker_key TEXT NOT NULL, url TEXT NOT NULL, task_id TEXT NOT NULL, task_type TEXT, status_str TEXT NOT NULL DEFAULT 'pending', result_json TEXT, token TEXT, sequence_key TEXT, attempt_count INT NOT NULL DEFAULT 0, max_attempts INT NOT NULL DEFAULT 3, scheduled_at TIMESTAMPTZ NOT NULL DEFAULT now(), created_at TIMESTAMPTZ NOT NULL DEFAULT now(), idempotency_key TEXT, sent_body BYTEA, notification_type TEXT, operation_id TEXT )""", # Backfill the operation_id column on tables created before the # async-completion-webhook operation_id echo landed. ADD COLUMN # IF NOT EXISTS is a no-op on fresh tables (the column is already # in the CREATE above) and an in-place add on pre-existing ones — # so a CREATE-only deployment and an upgrading one converge. f"ALTER TABLE {qt} ADD COLUMN IF NOT EXISTS operation_id TEXT", # Partial index on work-eligible rows; avoids scanning completed/in-flight rows. f"""CREATE INDEX IF NOT EXISTS {qt}_work_idx ON {qt} (status_str, scheduled_at) WHERE status_str IN ('pending', 'retry')""", f"""CREATE TABLE IF NOT EXISTS {lt} ( id BIGSERIAL PRIMARY KEY, queue_id BIGINT, breaker_key TEXT NOT NULL, url TEXT NOT NULL, task_id TEXT NOT NULL, sequence_key TEXT, sequence_number BIGINT, attempt_number INT NOT NULL, max_attempts INT NOT NULL, outcome TEXT NOT NULL, http_status_code INT, error_message TEXT, response_time_ms INT NOT NULL DEFAULT 0, occurred_at TIMESTAMPTZ NOT NULL, will_retry BOOLEAN NOT NULL DEFAULT false, next_retry_at TIMESTAMPTZ, task_type TEXT, payload_size_bytes INT, notification_type TEXT )""", f"CREATE INDEX IF NOT EXISTS {lt}_queue_id_idx ON {lt} (queue_id)", f"CREATE INDEX IF NOT EXISTS {lt}_task_id_idx ON {lt} (task_id)", ] async with self._pool.connection() as conn: for stmt in statements: await conn.execute(stmt)Bootstrap all three tables and their indexes. Idempotent.
Safe to call on every app boot. Each DDL statement is executed separately — psycopg does not split on
;.::
pool = AsyncConnectionPool("postgresql://...") supervisor = PgWebhookDeliverySupervisor(pool=pool, sender=sender) await supervisor.create_schema() async def run_worker(self, *, poll_interval: float = 0.5) ‑> None-
Expand source code
async def run_worker(self, *, poll_interval: float = 0.5) -> None: """Poll the delivery queue with ``FOR UPDATE SKIP LOCKED``; runs forever. **REQUIRED** — start at app startup:: asyncio.create_task(supervisor.run_worker()) Multiple processes or coroutines can run concurrently; ``SKIP LOCKED`` ensures each job is processed by exactly one worker. The DB connection is held for the duration of the HTTP send so a crashed worker automatically releases the job back to the queue via transaction rollback (no separate recovery sweep needed). :param poll_interval: Seconds to sleep when the queue is empty. """ self._worker_started = True logger.info( "[adcp.webhook_supervisor_pg] worker started (poll_interval=%.2fs)", poll_interval, ) while True: try: delivered = await self._poll_and_process() if not delivered: await asyncio.sleep(poll_interval) except asyncio.CancelledError: logger.info("[adcp.webhook_supervisor_pg] worker cancelled — shut down") raise except Exception: logger.exception( "[adcp.webhook_supervisor_pg] worker error; will retry after %.2fs", poll_interval, ) await asyncio.sleep(poll_interval)Poll the delivery queue with
FOR UPDATE SKIP LOCKED; runs forever.REQUIRED — start at app startup::
asyncio.create_task(supervisor.run_worker())Multiple processes or coroutines can run concurrently;
SKIP LOCKEDensures each job is processed by exactly one worker. The DB connection is held for the duration of the HTTP send so a crashed worker automatically releases the job back to the queue via transaction rollback (no separate recovery sweep needed).:param poll_interval: Seconds to sleep when the queue is empty.
async def send_mcp(self,
*,
url: str,
task_id: str,
status: GeneratedTaskStatus | str,
task_type: TaskType | str,
result: Any = None,
operation_id: str,
token: str | None = None,
sequence_key: str | None = None,
breaker_key: str | None = None,
notification_type: str | None = None) ‑> WebhookDeliveryResult | None-
Expand source code
async def send_mcp( self, *, url: str, task_id: str, status: GeneratedTaskStatus | str, task_type: TaskType | str, result: Any = None, operation_id: str, token: str | None = None, sequence_key: str | None = None, breaker_key: str | None = None, notification_type: str | None = None, ) -> WebhookDeliveryResult | None: """Enqueue one MCP-style webhook delivery; **always returns None**. The actual HTTP send happens in :meth:`run_worker`. This method writes the job to ``adcp_webhook_delivery_queue`` and returns immediately. ``None`` is returned whether the circuit is open (delivery skipped) or the job was successfully enqueued. For delivery outcomes use a :class:`~adcp.webhook_supervisor.DeliveryLogSink` or query the ``adcp_webhook_delivery_log`` table directly. :param breaker_key: Override the circuit-breaker lookup key (default: ``url``). Multi-tenant sellers whose buyers share a SaaS receiver URL MUST pass a tenant-scoped key (e.g. ``f"{tenant_id}:{url}"``). :param operation_id: Buyer-supplied correlation id echoed verbatim into the webhook payload per spec. Persisted on the queue row and replayed to :meth:`WebhookSender.send_mcp` when the worker delivers the job. :param notification_type: Passed through to the delivery log for delivery-report webhooks (``scheduled`` / ``final`` / etc.). """ if not self._worker_started and not self._worker_warned: self._worker_warned = True logger.warning( "[adcp.webhook_supervisor_pg] send_mcp() called before run_worker() " "has been started. Deliveries will be queued but not sent until a worker " "is running. Call asyncio.create_task(supervisor.run_worker()) at startup." ) # The queue column is TEXT; psycopg would otherwise bind a TaskType # enum via repr (`"TaskType.create_media_buy"`), corrupting the wire # value. Normalize once here so every persistence + log site sees # the on-wire string. task_type_str: str = task_type.value if isinstance(task_type, TaskType) else task_type bkey = breaker_key or url # Check circuit state; reject immediately if OPEN within the timeout window. async with self._pool.connection() as conn: cur = await conn.execute(self._sql_circuit_get, (bkey,)) row = await cur.fetchone() if row is not None: state: str = row[0] opened_at: datetime | None = row[1] if state == "open" and opened_at is not None: if opened_at.tzinfo is None: opened_at = opened_at.replace(tzinfo=UTC) elapsed = (datetime.now(UTC) - opened_at).total_seconds() if elapsed < self._circuit_policy.open_timeout_seconds: occurred_at = datetime.now(UTC) attempt = DeliveryAttempt( url=url, sequence_key=sequence_key, sequence_number=None, attempt_number=0, max_attempts=self._retry.max_attempts, outcome="circuit_open", http_status_code=None, error_message=f"circuit breaker OPEN for {bkey} — skipped delivery", response_time_ms=0, occurred_at=occurred_at, will_retry=False, next_retry_at=None, task_type=task_type_str, task_id=task_id, payload_size_bytes=None, notification_type=notification_type, ) await self._log_circuit_open(bkey, attempt) await self._call_sink(attempt) logger.warning( "[adcp.webhook_supervisor_pg] circuit OPEN for %s — skipped %s", bkey, task_type_str, ) return None # Open timeout elapsed; transition to half_open so next worker probes. async with self._pool.connection() as conn: await conn.execute(self._sql_circuit_set_half_open, (bkey,)) status_str = status if isinstance(status, str) else str(status) result_json = json.dumps(result) if result is not None else None async with self._pool.connection() as conn: cur = await conn.execute( self._sql_enqueue, ( bkey, url, task_id, task_type_str, status_str, result_json, token, sequence_key, self._retry.max_attempts, notification_type, operation_id, ), ) enqueue_row = await cur.fetchone() queue_id = enqueue_row[0] if enqueue_row else None logger.debug( "[adcp.webhook_supervisor_pg] enqueued %s → %s (queue_id=%s)", task_id, url, queue_id, ) return NoneEnqueue one MCP-style webhook delivery; always returns None.
The actual HTTP send happens in :meth:
run_worker. This method writes the job toadcp_webhook_delivery_queueand returns immediately.Noneis returned whether the circuit is open (delivery skipped) or the job was successfully enqueued.For delivery outcomes use a :class:
~adcp.webhook_supervisor.DeliveryLogSinkor query theadcp_webhook_delivery_logtable directly.:param breaker_key: Override the circuit-breaker lookup key (default:
url). Multi-tenant sellers whose buyers share a SaaS receiver URL MUST pass a tenant-scoped key (e.g.f"{tenant_id}:{url}"). :param operation_id: Buyer-supplied correlation id echoed verbatim into the webhook payload per spec. Persisted on the queue row and replayed to :meth:WebhookSender.send_mcpwhen the worker delivers the job. :param notification_type: Passed through to the delivery log for delivery-report webhooks (scheduled/final/ etc.).