diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9739ff306..88eedf03b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -130,6 +130,7 @@ jobs: tests/conformance/decisioning/test_pg_buyer_agent_registry.py \ tests/conformance/decisioning/test_pg_idempotency_backend.py \ tests/conformance/decisioning/test_pg_task_webhook_outbox.py \ + tests/conformance/decisioning/test_pg_reference_workflow_queue.py \ -v conventional-commits: @@ -472,7 +473,7 @@ jobs: - name: Install dependencies run: | python -m pip install --upgrade pip - pip install -e ".[dev]" + pip install -e ".[dev,pg]" # Example-local deps: the v3 reference seller imports # sqlalchemy + asyncpg + httpx-respx but those aren't in the # SDK's [dev] extras. Install them inline rather than adding diff --git a/README.md b/README.md index 6a89e7f39..649e0f03c 100644 --- a/README.md +++ b/README.md @@ -11,7 +11,7 @@ Official Python SDK for the **Ad Context Protocol (AdCP)**. Build and connect to This README serves both sides of an AdCP integration. Jump to what you're doing: - **Connect as a buyer** → [Quick Start: Test Helpers](#quick-start-test-helpers) and [Quick Start: Distributed Operations](#quick-start-distributed-operations). Entry point: `from adcp import ADCPClient, AgentConfig`; start with the `client.simple.*` API. -- **Build a seller / agent** → [Building an AdCP Agent](#building-an-adcp-agent). Entry point: `from adcp.server import ADCPHandler, serve`. +- **Build a seller / agent** → [Building an AdCP Agent](#building-an-adcp-agent). Entry point: `from adcp.server import ADCPHandler, serve`; use the [production seller path](docs/production-seller.md) when adding tenants, durable tasks, and webhooks. - **Understand the type system & imports** → [Type Safety](#type-safety) (import surface, partial modules, cold-start note). - **Test against reference agents** → [Quick Start: Test Helpers](#quick-start-test-helpers) and [Test Helpers](#test-helpers). Entry point: `from adcp.testing import test_agent, creative_agent`. @@ -354,6 +354,7 @@ forward traffic degrades gracefully rather than failing. - **[API Reference](https://adcontextprotocol.github.io/adcp-client-python/)** - Complete API documentation with type signatures and examples - **[Protocol Spec](https://github.com/adcontextprotocol/adcp)** - Ad Context Protocol specification - **[Handler authoring](docs/handler-authoring.md)** - Building an AdCP-compliant agent on `adcp.server` +- **[Production seller path](docs/production-seller.md)** - Choose the server abstraction and wire durable multi-tenant tasks, idempotency, and webhook delivery - **[Migrating from SDK 6 to 7](https://github.com/adcontextprotocol/adcp-client-python/blob/main/MIGRATION_v6_to_v7.md)** - Breaking API, security, concurrency, and webhook changes - **[Migrating from SDK 7 to 8](https://github.com/adcontextprotocol/adcp-client-python/blob/main/MIGRATION_v7_to_v8.md)** - Secure webhook defaults and telemetry changes - **[Migrating from AdCP 3.1 to 3.2 beta](MIGRATION_ADCP_3.1_TO_3.2.md)** - Compact lifecycle adoption and old/new compatibility matrix @@ -1033,8 +1034,10 @@ from adcp.server import ADCPHandler, IdempotencyStore, MemoryBackend, serve from adcp.server.responses import capabilities_response idempotency = IdempotencyStore( - backend=MemoryBackend(), # PgBackend with transactional commit is a follow-up + backend=MemoryBackend(), # Use PgBackend for a durable, multi-worker cache ttl_seconds=86400, # 24h, spec-recommended floor + # In production, also set raise_on_persist_error=True and independently + # deduplicate the business effect using the same buyer key. ) class MySeller(ADCPHandler): @@ -1063,9 +1066,14 @@ serve(MySeller(), name="my-seller") - On cache hit with different hash: raises `IdempotencyConflictError`, which the framework surfaces as `IDEMPOTENCY_CONFLICT` on both MCP (`is_error=true` + text) and A2A (failed task with `adcp_error` DataPart) - On cache miss: runs your handler, then commits the response -**Backends:** `MemoryBackend` ships now (tests, single-process agents). `PgBackend` is scaffolded — it raises `NotImplementedError` with a pointer to the follow-up issue. For production use across multiple workers, implement your own `IdempotencyBackend` subclass against Redis, Postgres, etc. +**Backends:** use `MemoryBackend` for tests and single-process agents. `PgBackend` provides a durable PostgreSQL replay cache for production deployments with multiple workers; it requires a separate advisory-lock pool and the `pg` extra. -**Atomicity caveat:** `MemoryBackend` commits the cache entry AFTER your handler returns, so a crash between `handler success` and `cache commit` causes the retry to re-execute. `PgBackend` (follow-up) will commit the cache row in the same transaction as your business writes. Read the module docstring at `adcp.server.idempotency` before shipping this against a production database. +**Atomicity caveat:** both backends commit the cache entry after your handler returns. `PgBackend` is durable and coordinates concurrent workers, but its cache transaction is not atomic with unrelated business writes. Put a uniqueness constraint on the business effect using the buyer's idempotency key so a crash between the side effect and cache commit cannot duplicate the effect. Read the `PgBackend` docstring before shipping it. + +With `raise_on_persist_error=True`, a failed cache write becomes retryable +`SERVICE_UNAVAILABLE`, but the handler has already completed and its outcome +may be unknown. A retry is safe only when the downstream business effect is +independently deduplicated using the same buyer key. **How caller identity gets populated.** The middleware scopes its cache by `(caller_identity, idempotency_key)` — same key from two buyers must hit different cache slots, and a buyer's retry must replay only against its own prior call. `caller_identity` comes from `ToolContext`, which the transport layer builds per request: diff --git a/docs/production-seller.md b/docs/production-seller.md new file mode 100644 index 000000000..d03aa31b8 --- /dev/null +++ b/docs/production-seller.md @@ -0,0 +1,250 @@ +# Production seller path + +This is the shortest route from a working AdCP seller to a durable, +multi-tenant deployment. The runnable implementation is +[`examples/v3_reference_seller`](../examples/v3_reference_seller/README.md); +this page explains which SDK layer to choose and who owns each lifecycle +transition. + +## Choose the server abstraction + +| Start with | Use it when | Move up when | +|---|---|---| +| `adcp_server()` decorators | You need a small stateless agent and prefer functions | You need custom inheritance or reusable handler behavior | +| `ADCPHandler` | You want direct control of AdCP request and response handlers | You need account resolution, specialism validation, upstream routing, or framework-managed tasks | +| `DecisioningPlatform` | You are building an operational seller with accounts, capabilities, async work, and one upstream | Keep it; add a `PlatformRouter` when routing differs by tenant or platform | +| `PlatformRouter` | One process serves multiple tenants or decisioning backends | This is the top-level composition layer | + +The production wiring reference uses `DecisioningPlatform`. Its modules deliberately +separate transport wiring, tenant identity, business translation, persistence, +and webhook delivery so adopters can replace one boundary at a time. + +## Runtime ownership + +```text +buyer request + │ + ▼ +auth → tenant router → buyer registry → idempotency lock + │ + ▼ +DecisioningPlatform method → upstream ad server + │ + ├─ terminal result ───────────────► return inline and cache result + │ + └─ TaskHandoff / WorkflowHandoff ─► persist submitted task + │ + complete/fail ─────┤ one PostgreSQL transaction + ▼ + task + outbox row + │ + separate worker + ▼ + signed webhook retry +``` + +The web process owns request validation and the atomic task/outbox commit. The +worker owns network delivery and retry. Both processes use the same PostgreSQL +database, 32-byte encryption key, signing key, and advertised retry horizon. +They construct separate pools and `WebhookSender` instances. + +The bundled mock upstream completes its approval path inline, so it does not +pretend that a process-local poll loop is restart-safe. A production adapter +whose upstream approval can outlive the request must return +`WorkflowHandoff`, persist the framework-issued task id in its own durable +queue, and have that queue's consumer call `registry.complete()` or +`registry.fail()`. The PostgreSQL registry makes task state durable; it does +not make arbitrary in-process work durable. The reference includes a leased +PostgreSQL queue and restart-recovery test; adopters still provide the +business-specific approval handler. + +The reference's `IdempotencyStore` wrapping is intentionally paired with its +inline terminal responses. Do not put the same method-level wrapper around a +method that returns a raw `TaskHandoff` or `WorkflowHandoff`: the wrapper runs +before framework task issuance and therefore cannot cache the projected +`{status: "submitted", task_id}` envelope. A durable workflow adapter must not +advertise method-level idempotency for that method unless an external durable +request-to-task mapping can reuse the prior task id. The reference queue's +uniqueness constraint is only on the framework-issued `task_id`, not the +buyer's idempotency key. A web-process +crash after enqueue commits but before the `submitted` response reaches the +buyer can therefore cause a retried request to issue a second task and queue +row. Fully closing that window requires SDK support for looking up or reusing +a workflow task id by buyer idempotency key. + +For a mixed adapter, make the split explicit with +`method_level_idempotency_methods`: include only methods that return terminal +responses from the method-level cache, and implement the workflow method's +deduplication in the durable queue. The default reference includes +`create_media_buy` because its mock path is inline. + +## Task transitions + +| Handler outcome | Work owner | Persistence owner | What the buyer does next | +|---|---|---|---| +| Return a result | Request handler | Idempotency backend caches the terminal response | Consume the inline result | +| Raise `AdcpError` | Request handler | Framework projects the structured error; no task is created | Follow its recovery guidance | +| `ctx.handoff_to_task(fn)` | SDK runs `fn` in the web process | `PgTaskRegistry` records submitted, progress, and terminal state | Poll `tasks/get` or await a webhook | +| `ctx.handoff_to_workflow(enqueue)` | The adopter's queue/worker/HITL system | The enqueue callback stores the task id; that system later calls `registry.complete()` or `registry.fail()` | Poll `tasks/get` or await a webhook | +| Input-required response | The business workflow that needs clarification | That workflow must retain the continuation state and context id | Resume the same context with the requested input | + +Use `TaskHandoff` only for bounded in-process work. A human review, Airflow +DAG, or long-running queue consumer belongs in `WorkflowHandoff`; otherwise a +web-process restart can strand the work even though the task row survived. + +For push-configured tasks, `PgTaskRegistry.complete()` and `.fail()` write the +terminal task and encrypted outbox envelope atomically. The worker leases the +outbox row, sends the exact stored body, and retries it with a stable +idempotency key. Polling and callbacks therefore observe the same terminal +artifact. + +## Durable WorkflowHandoff example + +[`workflow_queue.py`](../examples/v3_reference_seller/src/workflow_queue.py) +is a PostgreSQL-backed adopter queue with expiring leases. The enqueue callback +stores the framework task id before `WorkflowHandoff` returns `submitted`; a +replacement worker reclaims an expired lease after a crash. Handler failures +retry with capped exponential backoff; after the configured attempt limit the +registry task fails and the queue row moves to `dead_lettered`. Jobs without a +matching account-scoped registry task dead-letter immediately. + +```python +queue = task_wiring.workflow_queue + +async def create_media_buy(self, req, ctx): + upstream_order = await create_upstream_order(req) + payload = { + "upstream_order_id": upstream_order["id"], + "downstream_idempotency_key": req.idempotency_key, + } + + async def enqueue(task_ctx): + await queue.enqueue_from_handoff( + task_ctx, + account_id=ctx.account.id, + workflow_type="manual_media_buy_approval", + payload=payload, + ) + + return ctx.handoff_to_workflow(enqueue) + +async def handle_approval(job): + # Any external write here must deduplicate on the stored buyer key. + return await approve_and_build_result(job.payload) + +# In the separately supervised worker entrypoint (includes SIGTERM handling): +await run_with_signals(workflow_handler=handle_approval) +``` + +The queue completes the original `PgTaskRegistry` record only after the +handler returns. A crash after an external side effect but before queue +acknowledgement causes deliberate re-execution after lease expiry, so the +business effect must be independently idempotent. The PostgreSQL conformance +test kills the first logical worker after claim, creates fresh queue/registry +objects, and verifies that the replacement completes the same task id. +Queue payloads are ordinary JSONB: store only minimal continuation state and +never copy push-notification credentials or other secrets into them. + +## Run the reference deployment + +Install the PostgreSQL extra and start the development database: + +```bash +pip install -e '.[dev,pg]' +cd examples/v3_reference_seller +docker compose up -d postgres +``` + +Generate distinct webhook-signing and outbox-encryption keys. Keep both in a +secret manager in production; the environment variables below are for the +local runnable path. + +```bash +adcp-keygen --alg ed25519 --purpose webhook-signing \ + --kid reference-webhook-key \ + --out /tmp/adcp-reference-webhook-signing.pem + +export ADCP_TASK_DATABASE_URL=postgresql://postgres@localhost/adcp +export ADCP_TASK_WEBHOOK_ENCRYPTION_KEY="$(openssl rand -base64 32)" +export ADCP_WEBHOOK_SIGNING_KEY_PATH=/tmp/adcp-reference-webhook-signing.pem +export ADCP_WEBHOOK_SIGNING_KEY_ID=reference-webhook-key +export ADCP_WEBHOOK_SIGNING_ALG=ed25519 +export ADCP_TASK_WEBHOOK_RETRY_HORIZON_SECONDS=86400 +``` + +For a local smoke test, start the mock upstream and web process as background +jobs, then leave the worker in the foreground. The exported configuration is +shared by all three processes: + +```bash +npx -y -p @adcp/client@latest \ + adcp mock-server sales-guaranteed --port 4503 --api-key test-key & + +DATABASE_URL=postgresql+asyncpg://postgres@localhost/adcp python -m seed + +ADCP_ENV=development \ +DATABASE_URL=postgresql+asyncpg://postgres@localhost/adcp \ +MOCK_AD_SERVER_URL=http://127.0.0.1:4503 \ +MOCK_AD_SERVER_API_KEY=test-key \ +python -m src.app & + +python -m src.worker +``` + +That block is a loopback smoke test using public fixture credentials; never +expose it or reuse its token. A real production entrypoint must replace the +seeded bearer map with OAuth or RFC 9421 verification, use managed TLS +PostgreSQL and a non-mock upstream, and inject database credentials and signing +material from a secret manager. Set `ADCP_ENV=production` only for that real +configuration; it makes the durable bundle mandatory at boot. + +In production, supervise those long-lived commands separately and inject the +same secret-manager values into both the web and worker processes. The example +validates the complete durable configuration before binding the HTTP listener. +A partial key, database, encryption, or retry configuration fails with the +missing field names. The retry horizon is projected into capabilities and must +match the outbox value. + +`DurableTaskWiring.startup()` calls `create_schema()` for a convenient local +bootstrap. The workflow example performs the one additive upgrade shown here, +but these runtime DDL calls are not a general schema migration system and do +not detect or safely evolve an arbitrarily mismatched table. +For production, copy the SDK-owned SQL files (`decisioning_tasks.sql` and +`task_webhook_outbox.sql`), the `PgBackend.create_schema()` DDL, and the +reference workflow-queue DDL into reviewed, versioned migrations and apply +them before either process starts. Runtime bootstrap can remain a safety net, +but migrations own schema evolution and rollback. + +The worker installs `SIGTERM` and `SIGINT` handlers, cancels its polling loops, +awaits their cleanup, and then closes the sender and PostgreSQL pools. This is +the shutdown path used by ordinary container and process supervisors. + +## Production checklist + +- Replace bearer fixture authentication with your OAuth or RFC 9421 verifier. +- Use managed PostgreSQL with TLS and credential authentication; never deploy + the example Compose file or seed data. +- Publish the public webhook JWK and keep request-signing and webhook-signing + keys distinct. +- Run at least one separately supervised outbox worker and alert on expired or + quarantined rows. +- Route human or long-running approval work through `WorkflowHandoff` and a + durable queue; reserve `TaskHandoff` for bounded work that may safely fail on + web-process restart. +- Put a uniqueness constraint on business effects keyed by the buyer's + idempotency key. The SDK cache cannot make a separate upstream transaction + atomic. +- Schedule `PgBackend.delete_expired()` (or equivalent SQL/pg_cron cleanup) so + expired idempotency rows do not accumulate. +- Apply URL challenge and SSRF validation before accepting durable callback + destinations. +- Run the in-process tests and the media-buy seller storyboard before deploy. + +`DurableTaskWiring` remains example-owned so its configuration surface can be +validated by adopters first. Once the registry, queue, signing, migration, and +shutdown contracts stabilize together, it is a candidate for an SDK-supported +production builder rather than copyable scaffolding. + +For constructor details and multi-tenant sender resolution, continue with +[`handler-authoring.md`](handler-authoring.md#webhooks). For tenant scoping +invariants, see [`multi-tenant-contract.md`](multi-tenant-contract.md). diff --git a/examples/v3_reference_seller/MIGRATION.md b/examples/v3_reference_seller/MIGRATION.md index d25c7e454..fa988eb2a 100644 --- a/examples/v3_reference_seller/MIGRATION.md +++ b/examples/v3_reference_seller/MIGRATION.md @@ -139,14 +139,14 @@ template. The column on the right calls out the gotchas. | `Tenant` / `BuyerAgent` | local DB | **KEEP** — these are the v3 commercial-identity layer. Strict tenant isolation runs in the framework's `SubdomainTenantMiddleware`. | | `seller_agent_v1.py` entrypoint | `examples/v3_reference_seller/src/app.py` template | Rewrite around `serve(transport='both', ...)`. Drop your hand-rolled MCP/A2A request parsing. | | `get_products(req, ctx)` body | translator method calling upstream | Replace inline catalog logic with HTTP translation. Fall back to your CMS / planner / forecasting service if your upstream's products endpoint isn't enough. | -| `create_media_buy(req, ctx)` body | translator method | Now async with `TaskHandoff` for HITL approval flows. Sync fast path returns `CreateMediaBuySuccessResponse` directly; slow path returns `ctx.handoff_to_task(fn)` and the framework projects the wire `Submitted` envelope. | +| `create_media_buy(req, ctx)` body | translator method | The mock reference returns `CreateMediaBuySuccessResponse` inline. For HITL or long-running approval, persist work in a durable queue with `ctx.handoff_to_workflow(enqueue)`; its consumer completes or fails the framework task. | | `update_media_buy(req, ctx)` body | translator method (or `UNSUPPORTED_FEATURE`) | Wire to your upstream's order-update endpoint (GAM `LineItemService.performLineItemAction`, FreeWheel `updateOrder`). The reference seller raises `UNSUPPORTED_FEATURE` because the JS mock has no update endpoint. Don't ship the shim. | | `sync_creatives(req, ctx)` body | translator method | One upstream `POST /v1/creatives` per creative; AdCP `creative_id` passes through as `client_request_id` for upstream dedup. | | `get_media_buy_delivery(req, ctx)` body | translator method | The upstream's `DeliveryReport` schema may not carry order status — the reference seller double-fetches `get_order` so AdCP `MediaBuyStatus` reflects the actual state (completed / canceled / rejected don't surface as `active`). | | `provide_performance_feedback(req, ctx)` body | translator method | See "CAPI semantic mismatch". | | `list_creative_formats(req, ctx)` body | translator method | Static catalog in the reference seller. Real publishers drive this from their format registry. | | Hand-rolled idempotency tracking | framework `RequestContext` + `idempotency_key` | The framework persists `idempotency_key → response_hash`; replays are constant-time. | -| Hand-rolled task lifecycle | framework `TaskRegistry` + `TaskHandoff` | Adopters call `ctx.handoff_to_task(fn)` and the framework manages submitted → working → completed/failed. Adopter coroutine can `raise AdcpError(...)` to signal terminal failure — the framework projects to wire-shape `failed`. | +| Hand-rolled task lifecycle | framework `TaskRegistry` + handoff markers | Use `TaskHandoff` only for bounded in-process work. Use `WorkflowHandoff` plus a durable queue for HITL or long-running work; its consumer calls `registry.complete()` / `registry.fail()`. | ### Specialism declaration upgrade @@ -507,9 +507,10 @@ class MyAdServerSeller(DecisioningPlatform, SalesPlatform): payload=..., ) if order["status"] == "pending_approval": - # async approval path — return a Submitted envelope - # and poll the upstream in the background - return ctx.handoff_to_task(self._poll_until_approved) + # Durable approval path: enqueue must persist task_ctx.id and all + # continuation state before it returns. A separate consumer polls + # the upstream and calls registry.complete()/registry.fail(). + return ctx.handoff_to_workflow(self._enqueue_approval) # sync fast path return CreateMediaBuySuccessResponse(...) diff --git a/examples/v3_reference_seller/README.md b/examples/v3_reference_seller/README.md index 4dcc94d57..ebdf84210 100644 --- a/examples/v3_reference_seller/README.md +++ b/examples/v3_reference_seller/README.md @@ -22,7 +22,10 @@ salesagent), see [MIGRATION.md](MIGRATION.md). | Account v3 storage (bank-details column) | `src/models.py` | `Account.billing_entity` JSON column | | Audit trail | `src/audit.py` | `adcp.audit_sink.AuditSink` | | MCP + A2A on one binary | `src/app.py` | `serve(transport="both", asgi_middleware=...)` | -| Durable HITL tasks (optional) | swap to `PgTaskRegistry` | `adcp.decisioning.pg.PgTaskRegistry` | +| Durable task state | `src/durable_tasks.py` | `adcp.decisioning.PgTaskRegistry` | +| Restart-safe external workflows | `src/workflow_queue.py` | `WorkflowHandoff` + leased PostgreSQL queue | +| Atomic signed webhook delivery | `src/durable_tasks.py` + `src/worker.py` | `PgTaskWebhookOutbox` | +| Durable request idempotency | `src/durable_tasks.py` | `adcp.server.idempotency.PgBackend` | | Account v3 projection on read | `src/platform.py::list_accounts` | `adcp.types.project_account_for_response` | ## Architecture @@ -83,6 +86,29 @@ DATABASE_URL=postgresql+asyncpg://postgres@localhost/adcp \ The seller binds `0.0.0.0:3001` and serves both transports. +The commands above use the lightweight local path: async tasks are pollable +but intentionally in-memory, and push-configured handoffs are rejected. For +the production-shaped PostgreSQL registry, idempotency cache, signed atomic +outbox, and separately supervised worker, follow the +[production seller path](../../docs/production-seller.md#run-the-reference-deployment). + +The bundled mock upstream resolves approvals quickly, so +`create_media_buy` completes inline. When adapting the example to a human or +long-running approval system, return `ctx.handoff_to_workflow(...)` and let a +durable queue consumer call `registry.complete()` or `registry.fail()`; do not +move a long polling loop into an in-process `TaskHandoff`. Leave method-level +idempotency unadvertised for that method unless an external durable +request-to-task mapping can reuse the prior task id. The reference's method +wrapper is for its inline terminal-response path. +The reference queue deduplicates only the framework-issued `task_id`. It does +not close the crash window between committing enqueue and returning +`submitted`, because a buyer retry can receive a newly issued task id. That +requires SDK support for reusing a workflow task id by buyer idempotency key. +Worker failures use capped exponential backoff and a bounded attempt count; +exhausted or account-mismatched jobs move to `dead_lettered` for operations. +The queue and restart-recovery test are runnable reference infrastructure; +adopters supply the workflow handler that talks to their approval system. + > ⚠️ **Local-dev only.** `docker-compose.yml` uses > `POSTGRES_HOST_AUTH_METHOD=trust` and exposes 5432 on > `0.0.0.0`. Do not run this compose file on a host reachable from @@ -167,11 +193,10 @@ resolved account with `mode="mock"` and guaranteed/non_guaranteed` — real GAM-shaped publishers sell both). Each method calls the upstream over HTTP and translates the response -to AdCP wire shapes. `create_media_buy` returns a `TaskHandoff` for -the upstream's `pending_approval` path — the buyer sees a -`Submitted` envelope; the framework runs a background coroutine that -polls `/v1/tasks/{id}` until the upstream auto-approves, then surfaces -the success via `tasks/get` polling. +to AdCP wire shapes. The bundled mock's `create_media_buy` approval path +polls briefly and returns an inline terminal response. A real approval that +can outlive the request belongs in an adopter-owned durable queue entered via +`WorkflowHandoff`, with completion surfaced through `tasks/get` and webhooks. `update_media_buy` raises `UNSUPPORTED_FEATURE` because the JS mock has no order-update endpoint. Real adopters wire their PATCH / per- @@ -236,13 +261,6 @@ where the framework picks it up. AAO publishes the brand.json registry. - **Brand authorization (Tier 3)** — gated on ADCP spec issue #3690. -- **Postgres `TaskRegistry` / `WebhookDeliverySupervisor`** — - swap `InMemoryTaskRegistry` → `PgTaskRegistry` and - `InMemoryWebhookDeliverySupervisor` → `PgWebhookDeliverySupervisor` - in `src/app.py` for production durability of HITL tasks (the - `create_media_buy` async approval path) and webhook delivery. - Both classes ship in the SDK; this seller's `app.py` uses the - in-memory variants for fast iteration. - **Admin CRUD API** — separate Starlette app for tenant / agent CRUD. Patterns to come; for now use `seed.py` and direct SQL. diff --git a/examples/v3_reference_seller/src/app.py b/examples/v3_reference_seller/src/app.py index 445e28450..85705276f 100644 --- a/examples/v3_reference_seller/src/app.py +++ b/examples/v3_reference_seller/src/app.py @@ -19,6 +19,7 @@ * :class:`TenantScopedBuyerAgentRegistry` for the Tier 2 gate * :class:`DbAuditSink` for compliance trail * :class:`V3ReferenceSeller` (the platform impl) + * optional durable PostgreSQL task registry + atomic webhook outbox 5. ``adcp.decisioning.serve(transport="both", asgi_middleware=[...])`` — single binary serving MCP at ``/mcp`` and A2A at ``/`` with @@ -56,7 +57,7 @@ from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine -from adcp.decisioning import AdcpError, InMemoryMockAdServer, serve +from adcp.decisioning import InMemoryMockAdServer, serve from adcp.decisioning.context import AuthInfo from adcp.decisioning.registry import ApiKeyCredential from adcp.server import ( @@ -71,11 +72,10 @@ enforce_authenticated_tenant, ) from adcp.validation import ValidationHookConfig -from adcp.webhook_sender import WebhookSender -from adcp.webhook_supervisor import InMemoryWebhookDeliverySupervisor from .audit import make_sink as make_audit_sink from .buyer_registry import make_registry as make_buyer_registry +from .durable_tasks import DurableTaskWiring from .platform import V3ReferenceSeller from .tenant_router import SqlSubdomainTenantRouter @@ -260,21 +260,6 @@ def main() -> None: "mock_sales_guaranteed_key_do_not_use_in_prod", ) - # Webhook-signing wiring (#384). Loaded from env so the PEM stays - # off the process command line and out of any os.environ dump that - # would otherwise surface a raw JWK scalar. The key is a separate - # ed25519/es256 keypair from the request-signing key — AdCP requires - # webhook-signing material be distinct so a signature from one - # surface cannot be replayed on the other. - # - # Generate with: - # adcp-keygen --alg ed25519 --purpose webhook-signing \ - # --out /etc/adcp/webhook-signing.pem - # Then publish the printed public JWK at the seller's jwks_uri. - signing_pem_path = os.environ.get("ADCP_WEBHOOK_SIGNING_KEY_PATH") - signing_key_id = os.environ.get("ADCP_WEBHOOK_SIGNING_KEY_ID") - signing_alg = os.environ.get("ADCP_WEBHOOK_SIGNING_ALG", "ed25519") - engine = create_async_engine(db_url, pool_size=10, max_overflow=20) sessionmaker = async_sessionmaker(engine, expire_on_commit=False) @@ -302,42 +287,17 @@ def main() -> None: # Adopters with a real production upstream replace ``mode='mock'`` # with ``mode='live'`` in their ``AccountStore.resolve`` and declare # :attr:`V3ReferenceSeller.upstream_url` to their production URL. - # Wire the webhook supervisor iff signing material is present for - # explicit/manual delivery demonstrations. It is not a beta.5 - # TaskHandoff publisher: push-configured handoffs require an external - # durable outbox that atomically owns terminal state and delivery. - webhook_supervisor: InMemoryWebhookDeliverySupervisor | None = None - if signing_pem_path and signing_key_id: - webhook_sender = WebhookSender.from_pem( - signing_pem_path, - key_id=signing_key_id, - alg=signing_alg, - ) - webhook_supervisor = InMemoryWebhookDeliverySupervisor(sender=webhook_sender) + # In local development the whole durable bundle may be omitted. In + # production DurableTaskWiring fails fast unless PostgreSQL, encryption, + # signing, registry, and outbox configuration are all present. The web + # process commits task terminal state + outbox rows atomically; a separate + # ``python -m src.worker`` process owns delivery and retries. + task_wiring = DurableTaskWiring.from_env() + if task_wiring is not None: logger.info( - "Webhook signing wired: key_id=%s alg=%s pem=%s", - signing_key_id, - signing_alg, - signing_pem_path, - ) - elif signing_pem_path or signing_key_id: - # Partial config is operator error — both env vars must be set - # together, or both omitted. Raise AdcpError (terminal) so an - # adopter wrapping main() in ``except AdcpError`` catches all - # boot misconfigs uniformly, matching the sibling validators. - raise AdcpError( - "INVALID_REQUEST", - message=( - "ADCP_WEBHOOK_SIGNING_KEY_PATH and " - "ADCP_WEBHOOK_SIGNING_KEY_ID must be set together — got " - f"path={signing_pem_path!r}, key_id={signing_key_id!r}" - ), - recovery="terminal", - details={ - "missing": "webhook_signing_env_pair", - "ADCP_WEBHOOK_SIGNING_KEY_PATH_set": bool(signing_pem_path), - "ADCP_WEBHOOK_SIGNING_KEY_ID_set": bool(signing_key_id), - }, + "Durable task registry and webhook outbox wired: alg=%s horizon=%ds", + task_wiring.signing_algorithm, + task_wiring.retry_horizon_seconds, ) platform = V3ReferenceSeller( @@ -345,7 +305,11 @@ def main() -> None: upstream_api_key=upstream_api_key, mock_upstream_url=upstream_url, mock_ad_server=mock_ad_server, - webhook_signing_alg=signing_alg if webhook_supervisor is not None else None, + webhook_signing_alg=(task_wiring.signing_algorithm if task_wiring else None), + webhook_retry_horizon_seconds=( + task_wiring.retry_horizon_seconds if task_wiring else 86_400 + ), + idempotency=task_wiring.idempotency if task_wiring else None, ) logger.info( @@ -363,6 +327,7 @@ def main() -> None: host="0.0.0.0", transport="both", buyer_agent_registry=buyer_registry, + registry=task_wiring.registry if task_wiring else None, # Bearer auth wired so the framework extracts the # ``Authorization: Bearer `` header, resolves the token # to a seeded BuyerAgent via api_key_id lookup, and threads the @@ -397,9 +362,8 @@ def main() -> None: if debug_token is not None else None ), - # Manual/reference delivery transport only. The SDK refuses to use - # this supervisor for beta.5 TaskHandoff push publication. - webhook_supervisor=webhook_supervisor, + on_startup=(task_wiring.startup,) if task_wiring else (), + on_shutdown=((task_wiring.shutdown, engine.dispose) if task_wiring else (engine.dispose,)), # FastMCP's TransportSecurityMiddleware enforces DNS-rebinding # protection: its default ``allowed_hosts`` accepts only # loopback (``127.0.0.1:*``, ``localhost:*``, ``[::1]:*``), so diff --git a/examples/v3_reference_seller/src/durable_tasks.py b/examples/v3_reference_seller/src/durable_tasks.py new file mode 100644 index 000000000..fda404eb2 --- /dev/null +++ b/examples/v3_reference_seller/src/durable_tasks.py @@ -0,0 +1,237 @@ +"""Production task-registry and webhook-outbox wiring. + +The web process and the separately supervised worker process each construct +this bundle from the same environment. They share PostgreSQL and encryption +material, but own separate connection pools and webhook senders. +""" + +from __future__ import annotations + +import asyncio +import base64 +import binascii +import logging +import os +from collections.abc import Mapping +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from adcp.decisioning import PgTaskRegistry, PgTaskWebhookOutbox +from adcp.server.idempotency import IdempotencyStore, PgBackend +from adcp.webhook_sender import WebhookSender + +from .workflow_queue import PgWorkflowQueue + +if TYPE_CHECKING: + from psycopg_pool import AsyncConnectionPool + +logger = logging.getLogger(__name__) + + +DEFAULT_RETRY_HORIZON_SECONDS = 86_400 +MAX_RETRY_HORIZON_SECONDS = 604_800 + +_REQUIRED_ENV = ( + "ADCP_TASK_DATABASE_URL", + "ADCP_TASK_WEBHOOK_ENCRYPTION_KEY", + "ADCP_WEBHOOK_SIGNING_KEY_PATH", + "ADCP_WEBHOOK_SIGNING_KEY_ID", +) + + +def _production_env(environ: Mapping[str, str]) -> bool: + return environ.get("ADCP_ENV", "").strip().lower() in {"prod", "production"} + + +def _decode_encryption_key(encoded: str) -> bytes: + """Decode a base64 secret without ever including it in diagnostics.""" + try: + key = base64.b64decode(encoded, validate=True) + except (binascii.Error, ValueError) as exc: + raise ValueError( + "ADCP_TASK_WEBHOOK_ENCRYPTION_KEY must be valid base64 for exactly 32 bytes" + ) from exc + if len(key) != 32: + raise ValueError("ADCP_TASK_WEBHOOK_ENCRYPTION_KEY must decode to exactly 32 bytes") + return key + + +def _retry_horizon(environ: Mapping[str, str]) -> int: + raw = environ.get( + "ADCP_TASK_WEBHOOK_RETRY_HORIZON_SECONDS", + str(DEFAULT_RETRY_HORIZON_SECONDS), + ) + try: + horizon = int(raw) + except ValueError as exc: + raise ValueError("ADCP_TASK_WEBHOOK_RETRY_HORIZON_SECONDS must be an integer") from exc + if not DEFAULT_RETRY_HORIZON_SECONDS <= horizon <= MAX_RETRY_HORIZON_SECONDS: + raise ValueError( + "ADCP_TASK_WEBHOOK_RETRY_HORIZON_SECONDS must be between " + f"{DEFAULT_RETRY_HORIZON_SECONDS} and {MAX_RETRY_HORIZON_SECONDS}" + ) + return horizon + + +@dataclass(frozen=True) +class DurableTaskWiring: + """One process's validated durable task/webhook resources.""" + + pool: AsyncConnectionPool + lock_pool: AsyncConnectionPool | None + sender: WebhookSender + outbox: PgTaskWebhookOutbox + registry: PgTaskRegistry + workflow_queue: PgWorkflowQueue + idempotency_backend: PgBackend | None + idempotency: IdempotencyStore | None + retry_horizon_seconds: int + signing_algorithm: str + + @classmethod + def from_env( + cls, + environ: Mapping[str, str] | None = None, + *, + required: bool | None = None, + include_idempotency: bool = True, + ) -> DurableTaskWiring | None: + """Validate configuration and build resources without opening sockets. + + Local development may omit the entire durable bundle. Production and + worker processes require all fields. A partial configuration always + fails before the HTTP listener or worker loop starts. + """ + values = os.environ if environ is None else environ + must_configure = _production_env(values) if required is None else required + configured = [name for name in _REQUIRED_ENV if values.get(name)] + if not configured and not must_configure: + return None + + missing = [name for name in _REQUIRED_ENV if not values.get(name)] + if missing: + raise ValueError( + "Durable task/webhook configuration is incomplete; missing: " + ", ".join(missing) + ) + + # Imported only when durable mode is selected so the lightweight local + # example still runs without the optional ``adcp[pg]`` dependency. + from psycopg_pool import AsyncConnectionPool + + retry_horizon = _retry_horizon(values) + signing_algorithm = values.get("ADCP_WEBHOOK_SIGNING_ALG", "ed25519") + encryption_key = _decode_encryption_key(values["ADCP_TASK_WEBHOOK_ENCRYPTION_KEY"]) + sender = WebhookSender.from_pem( + values["ADCP_WEBHOOK_SIGNING_KEY_PATH"], + key_id=values["ADCP_WEBHOOK_SIGNING_KEY_ID"], + alg=signing_algorithm, + ) + pool = AsyncConnectionPool( + values["ADCP_TASK_DATABASE_URL"], + min_size=1, + max_size=10, + open=False, + ) + # Advisory-lock operations reserve a connection for the whole handler + # call, so the SDK requires a distinct pool to prevent self-deadlock. + lock_pool = ( + AsyncConnectionPool( + values["ADCP_TASK_DATABASE_URL"], + min_size=1, + max_size=10, + open=False, + ) + if include_idempotency + else None + ) + outbox = PgTaskWebhookOutbox( + pool=pool, + sender=sender, + encryption_key=encryption_key, + delivery_retry_horizon_seconds=retry_horizon, + ) + registry = PgTaskRegistry(pool=pool, task_webhook_outbox=outbox) + workflow_queue = PgWorkflowQueue(pool=pool, registry=registry) + idempotency_backend = ( + PgBackend(pool=pool, lock_pool=lock_pool) if lock_pool is not None else None + ) + idempotency = ( + IdempotencyStore( + backend=idempotency_backend, + ttl_seconds=retry_horizon, + raise_on_persist_error=True, + ) + if idempotency_backend is not None + else None + ) + return cls( + pool=pool, + lock_pool=lock_pool, + sender=sender, + outbox=outbox, + registry=registry, + workflow_queue=workflow_queue, + idempotency_backend=idempotency_backend, + idempotency=idempotency, + retry_horizon_seconds=retry_horizon, + signing_algorithm=signing_algorithm, + ) + + async def startup(self) -> None: + """Open the process-local pool and idempotently create SDK tables.""" + try: + await self.pool.open() + if self.lock_pool is not None: + await self.lock_pool.open() + await self.registry.create_schema() + await self.outbox.create_schema() + await self.workflow_queue.create_schema() + if self.idempotency_backend is not None: + await self.idempotency_backend.create_schema() + except asyncio.CancelledError: + cleanup = asyncio.create_task(self.shutdown()) + # Once startup owns resources, repeated cancellation must not + # strand them. Shield the cleanup task and preserve cancellation + # for the caller after every close has had a chance to run. + while not cleanup.done(): + try: + await asyncio.shield(cleanup) + except asyncio.CancelledError: + continue + try: + cleanup.result() + except asyncio.CancelledError: + logger.exception("Durable resource cleanup was cancelled") + except Exception: + logger.exception("Durable resource cleanup failed after startup cancellation") + raise + except Exception: + try: + await self.shutdown() + except Exception: + logger.exception("Durable resource cleanup failed after startup error") + raise + + async def shutdown(self) -> None: + """Best-effort close every resource owned by this process.""" + first_error: BaseException | None = None + closers = [self.sender.aclose] + if self.lock_pool is not None: + closers.append(self.lock_pool.close) + closers.append(self.pool.close) + for close in closers: + try: + await close() + except asyncio.CancelledError as exc: + logger.exception("Durable task resource close was cancelled") + if first_error is None: + first_error = exc + except Exception as exc: + logger.exception("Failed to close a durable task resource") + if first_error is None: + first_error = exc + if first_error is not None: + raise first_error + + +__all__ = ["DEFAULT_RETRY_HORIZON_SECONDS", "DurableTaskWiring"] diff --git a/examples/v3_reference_seller/src/platform.py b/examples/v3_reference_seller/src/platform.py index c4d3c6ee1..96f68c1dc 100644 --- a/examples/v3_reference_seller/src/platform.py +++ b/examples/v3_reference_seller/src/platform.py @@ -9,9 +9,9 @@ Required (every sales-* specialism): * :meth:`get_products` — translate ``GET /v1/products`` upstream -* :meth:`create_media_buy` — ``POST /v1/orders``; returns - :class:`Submitted` task envelope; background handoff polls - ``/v1/tasks/{id}`` until approved +* :meth:`create_media_buy` — ``POST /v1/orders``; the bundled mock resolves + approval inline. Production adapters with long-running approval return a + ``WorkflowHandoff`` backed by their durable queue. * :meth:`update_media_buy` — UNSUPPORTED (mock has no order-update endpoint; the framework raises ``UNSUPPORTED_FEATURE``) * :meth:`sync_creatives` — ``POST /v1/creatives`` per creative @@ -66,6 +66,7 @@ import random from dataclasses import replace as _dc_replace from datetime import datetime, timezone +from types import MethodType from typing import TYPE_CHECKING, Any, ClassVar, cast from urllib.parse import urlsplit @@ -90,6 +91,9 @@ from adcp.decisioning.capabilities import ( Account as CapsAccount, ) +from adcp.decisioning.capabilities import ( + Adcp as CapsAdcp, +) from adcp.decisioning.capabilities import ( Features as MediaBuyFeatures, ) @@ -105,6 +109,7 @@ from adcp.decisioning.specialisms import SalesPlatform from adcp.server import current_tenant from adcp.server.helpers import valid_actions_for_status +from adcp.server.idempotency import IdempotencyStore from adcp.server.responses import list_creatives_response from adcp.types import ( BusinessEntity, @@ -128,6 +133,7 @@ UpdateMediaBuyRequest, UpdateMediaBuySuccessResponse, ) +from adcp.types.capabilities import IdempotencySupported from adcp.types.legacy import ( LegacyFormat, LegacyFormatId, @@ -153,6 +159,13 @@ "https://your-platform.example.com", } +_DEFAULT_IDEMPOTENCY_METHODS = ( + "create_media_buy", + "update_media_buy", + "sync_creatives", + "provide_performance_feedback", +) + def _legacy_format_converter(context: LegacyFormatConversionContext) -> dict[str, Any] | None: """Explicitly map the reference fixture catalogs into canonical kinds.""" @@ -787,6 +800,7 @@ class V3ReferenceSeller(DecisioningPlatform, SalesPlatform): CapsSpecialism.sales_non_guaranteed, CapsSpecialism.sales_guaranteed, ], + adcp=CapsAdcp.model_validate({"major_versions": [3], "idempotency": {"supported": False}}), # ``account.supported_billing`` is required by the spec # whenever ``media_buy`` is in ``supported_protocols``. The # reference seller invoices the operator (agency / brand @@ -812,6 +826,9 @@ def __init__( approval_poll_interval_s: float = 1.0, approval_poll_max_iterations: int = 60, webhook_signing_alg: str | None = None, + webhook_retry_horizon_seconds: int = 86_400, + idempotency: IdempotencyStore | None = None, + method_level_idempotency_methods: tuple[str, ...] | None = None, ) -> None: """Construct the reference seller. @@ -845,6 +862,14 @@ def __init__( :class:`~adcp.webhook_sender.WebhookSender` produces RFC 9421 signatures over outbound deliveries. Default ``None`` — no signing advertised, sender wiring optional. + :param webhook_retry_horizon_seconds: Durable retry horizon advertised + to webhook receivers. Must match the configured task outbox. + :param idempotency: Optional durable idempotency store. When supplied, + mutating methods are wrapped and the replay window is advertised. + :param method_level_idempotency_methods: Methods whose inline terminal + responses are cached by ``idempotency``. Workflow adapters must + exclude methods returning handoff markers and deduplicate their + durable queue issuance separately. """ # Override the class-level capabilities iff signing is wired. # ``dataclasses.replace`` preserves every other field from the @@ -858,6 +883,20 @@ def __init__( supported=True, profile="adcp/webhook-signing/v1", algorithms=[webhook_signing_alg], # type: ignore[list-item] + delivery_retry_horizon_seconds=webhook_retry_horizon_seconds, + ), + ) + if idempotency is not None: + assert self.capabilities.adcp is not None + self.capabilities = _dc_replace( + self.capabilities, + adcp=self.capabilities.adcp.model_copy( + update={ + "idempotency": IdempotencySupported( + supported=True, + replay_ttl_seconds=idempotency.ttl_seconds, + ) + } ), ) @@ -899,6 +938,22 @@ def __init__( sessionmaker, mock_upstream_url=mock_upstream_url, ) + if idempotency is not None: + selected_methods = ( + _DEFAULT_IDEMPOTENCY_METHODS + if method_level_idempotency_methods is None + else method_level_idempotency_methods + ) + for method_name in selected_methods: + method = getattr(type(self), method_name, None) + if not callable(method): + raise ValueError(f"Unknown method-level idempotency method: {method_name}") + # Bind the decorated unbound method back to this instance. + # Wrapping an already-bound method and assigning the plain + # closure would shift ``req`` into the decorator's ``self`` + # slot, causing idempotency-key extraction to silently miss. + wrapped = idempotency.wrap(method) + setattr(self, method_name, MethodType(wrapped, self)) def _client(self, ctx: RequestContext) -> UpstreamHttpClient: """Resolve the pooled :class:`UpstreamHttpClient` for this @@ -1131,13 +1186,13 @@ async def refine_get_products( async def create_media_buy(self, req: CreateMediaBuyRequest, ctx: RequestContext): """``POST /v1/orders`` → upstream returns ``pending_approval`` - with an ``approval_task_id``. Hand off to a background coroutine - that polls ``/v1/tasks/{id}`` until approved, then returns the - :class:`CreateMediaBuySuccessResponse`. + with an ``approval_task_id``. The mock reference polls that task + briefly and returns :class:`CreateMediaBuySuccessResponse` inline. - Buyer experience: ``{status: 'submitted', task_id}`` immediately; - framework's task registry surfaces the success on - ``tasks/get`` polling once the upstream approves. + The bundled mock upstream resolves approval quickly, so this reference + awaits the result and returns it inline. A production adapter whose + approval can outlive the request should return ``WorkflowHandoff`` and + let its durable worker complete or fail the framework task. Measurement terms gating: this seller cannot guarantee zero variance on billing measurement (``max_variance_percent == 0`` @@ -1209,20 +1264,13 @@ async def create_media_buy(self, req: CreateMediaBuyRequest, ctx: RequestContext current, req, budget_amount, budget_currency, client, network_code, ctx ) - # Slow path — hand off to background polling. The framework - # allocates a task_id, returns the Submitted envelope, and runs - # the handoff coroutine in the background. When this coroutine - # returns, the framework persists the success as the terminal - # artifact on the registry; buyers see it via ``tasks/get`` or - # via the push-notification webhook. When this coroutine raises - # :class:`AdcpError`, the framework persists ``failed`` with the - # wire-shaped error payload — so terminal-failure projection - # (rejected, timed-out polling) goes through ``raise``, not - # through fabricating a success response. + # Mock slow path — poll briefly in this request. The fixture approves + # within milliseconds, keeping the storyboard's create response + # synchronous. A production adapter must replace this block with a + # durable WorkflowHandoff queue when approval can outlive the request. bound_task_id = approval_task_id - async def _poll_until_approved(task_handoff_ctx: Any) -> CreateMediaBuySuccessResponse: - del task_handoff_ctx + async def _poll_until_approved() -> CreateMediaBuySuccessResponse: for _ in range(self._approval_poll_max_iterations): task = await upstream_helpers.get_task( client, network_code=network_code, task_id=bound_task_id @@ -1284,14 +1332,15 @@ async def _poll_until_approved(task_handoff_ctx: Any) -> CreateMediaBuySuccessRe # lets create_media_buy return the full CreateMediaBuySuccessResponse # with media_buy_id in the response body, which is what AdCP # storyboards and most buyers expect. Production adopters with slow - # real-world approvals swap this for the task-handoff path: + # real-world approvals must move the poll to a durable workflow queue: # - # return ctx.handoff_to_task(_poll_until_approved) + # return ctx.handoff_to_workflow(enqueue_approval) # # which returns a Submitted({task_id, status: 'submitted'}) envelope - # and runs the polling coroutine in the background while buyers - # poll via tasks/get. - return await _poll_until_approved(None) + # after the enqueue callback durably stores the framework task id. + # The external worker later calls registry.complete()/fail(); using + # TaskHandoff for long polling would be stranded by a web restart. + return await _poll_until_approved() def _reject_unworkable_terms(self, req: CreateMediaBuyRequest) -> None: """Reject ``create_media_buy`` requests whose ``measurement_terms`` @@ -1991,7 +2040,7 @@ async def provide_performance_feedback( "event_name": metric_type, "event_time": event_time, "value": performance_index, - "dedup_key": f"{req.media_buy_id}:{metric_type}:{event_time}", + "dedup_key": (f"adcp:{req.idempotency_key}:performance-feedback"), } ], } diff --git a/examples/v3_reference_seller/src/worker.py b/examples/v3_reference_seller/src/worker.py new file mode 100644 index 000000000..75e612d07 --- /dev/null +++ b/examples/v3_reference_seller/src/worker.py @@ -0,0 +1,106 @@ +"""Separately supervised durable task-webhook and workflow worker. + +Run beside the web process with the same ``ADCP_TASK_*`` and +``ADCP_WEBHOOK_SIGNING_*`` environment variables:: + + python -m src.worker + +The stock entrypoint delivers the webhook outbox. Adopters with external +workflow work call :func:`run_with_signals` with their idempotent handler. +""" + +from __future__ import annotations + +import asyncio +import logging +import signal + +from .durable_tasks import DurableTaskWiring +from .workflow_queue import WorkflowHandler + +logger = logging.getLogger(__name__) + + +async def run( + *, + stop_event: asyncio.Event | None = None, + workflow_handler: WorkflowHandler | None = None, +) -> None: + """Run durable workers until a shutdown signal sets ``stop_event``.""" + wiring = DurableTaskWiring.from_env(required=True, include_idempotency=False) + assert wiring is not None # required=True makes None impossible + await wiring.startup() + stop = stop_event or asyncio.Event() + workers = [ + asyncio.create_task( + wiring.outbox.run_worker(), + name="adcp-task-webhook-outbox", + ) + ] + if workflow_handler is not None: + workers.append( + asyncio.create_task( + wiring.workflow_queue.run_worker(workflow_handler), + name="adcp-workflow-queue", + ) + ) + stop_waiter = asyncio.create_task(stop.wait(), name="adcp-worker-shutdown") + logger.info("Durable workers started") + try: + done, _pending = await asyncio.wait( + [*workers, stop_waiter], + return_when=asyncio.FIRST_COMPLETED, + ) + completed_workers = done.intersection(workers) + for task in completed_workers: + exc = task.exception() + if exc is not None: + raise exc + if completed_workers: + raise RuntimeError("A durable worker exited unexpectedly") + logger.info("Shutdown requested; stopping durable workers") + finally: + stop_waiter.cancel() + for task in workers: + task.cancel() + await asyncio.gather(stop_waiter, *workers, return_exceptions=True) + await wiring.shutdown() + + +async def run_with_signals( + *, + workflow_handler: WorkflowHandler | None = None, +) -> None: + """Install SIGTERM/SIGINT handlers and run until either is received.""" + stop_event = asyncio.Event() + loop = asyncio.get_running_loop() + installed: list[signal.Signals] = [] + for signum in (signal.SIGTERM, signal.SIGINT): + try: + loop.add_signal_handler(signum, stop_event.set) + except NotImplementedError: + continue + installed.append(signum) + try: + await run( + stop_event=stop_event, + workflow_handler=workflow_handler, + ) + finally: + for signum in installed: + loop.remove_signal_handler(signum) + + +def main() -> None: + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s: %(message)s", + ) + asyncio.run(run_with_signals()) + + +if __name__ == "__main__": + main() + + +__all__ = ["main", "run", "run_with_signals"] diff --git a/examples/v3_reference_seller/src/workflow_queue.py b/examples/v3_reference_seller/src/workflow_queue.py new file mode 100644 index 000000000..f019cd536 --- /dev/null +++ b/examples/v3_reference_seller/src/workflow_queue.py @@ -0,0 +1,442 @@ +"""PostgreSQL queue for adopter-owned ``WorkflowHandoff`` work. + +This is deliberately example code, not an SDK primitive. It demonstrates the +minimum production invariants: durable enqueue, single-worker leases, +restart recovery after lease expiry, account scoping, and completion through +the SDK's durable task registry. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import re +import uuid +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from psycopg_pool import AsyncConnectionPool + + from adcp.decisioning import TaskRegistry + +logger = logging.getLogger(__name__) + +_SAFE_IDENTIFIER = re.compile(r"^[a-z_][a-z0-9_]{0,62}$") +DEFAULT_WORKFLOW_TABLE = "adcp_reference_workflows" +DEFAULT_MAX_ATTEMPTS = 8 +DEFAULT_MAX_RETRY_SECONDS = 300.0 +DEFAULT_RETRY_BASE_SECONDS = 1.0 + +_RETRY_EXHAUSTED_ERROR: dict[str, str] = { + "code": "INTERNAL_ERROR", + "message": "Workflow processing exhausted its retry budget.", + "recovery": "terminal", +} + + +@dataclass(frozen=True) +class WorkflowJob: + """One leased workflow job returned to a worker.""" + + task_id: str + account_id: str + workflow_type: str + payload: dict[str, Any] + lease_token: str + attempt_count: int + + +WorkflowHandler = Callable[[WorkflowJob], Awaitable[dict[str, Any]]] + + +class PgWorkflowQueue: + """A small durable queue that completes SDK-managed workflow tasks. + + Handlers must make external side effects idempotent. A worker can finish + the side effect and die before acknowledging the queue row; after the + lease expires, a replacement worker intentionally runs the job again. + + ``payload`` is ordinary JSONB, not an encrypted secret store. Enqueue only + the minimum continuation data and never include callback credentials. + """ + + def __init__( + self, + *, + pool: AsyncConnectionPool, + registry: TaskRegistry, + table: str = DEFAULT_WORKFLOW_TABLE, + lease_seconds: int = 60, + max_attempts: int = DEFAULT_MAX_ATTEMPTS, + retry_base_seconds: float = DEFAULT_RETRY_BASE_SECONDS, + max_retry_seconds: float = DEFAULT_MAX_RETRY_SECONDS, + ) -> None: + if not _SAFE_IDENTIFIER.fullmatch(table): + raise ValueError("workflow table must be a safe lowercase PostgreSQL identifier") + if lease_seconds < 2: + raise ValueError("workflow lease_seconds must be at least 2") + if max_attempts < 1: + raise ValueError("workflow max_attempts must be at least 1") + if retry_base_seconds <= 0: + raise ValueError("workflow retry_base_seconds must be positive") + if max_retry_seconds < retry_base_seconds: + raise ValueError("workflow max_retry_seconds must be at least retry_base_seconds") + self._pool = pool + self._registry = registry + self._table = table + suffix = "_state_check" + self._state_constraint = f"{table[: 63 - len(suffix)]}{suffix}" + self._lease_seconds = lease_seconds + self._max_attempts = max_attempts + self._retry_base_seconds = retry_base_seconds + self._max_retry_seconds = max_retry_seconds + + self._sql_claim = ( # noqa: S608 - table is validated above + f"WITH candidate AS (" + f" SELECT task_id FROM {table}" + " WHERE state = 'pending' AND available_at <= now()" + " ORDER BY available_at, created_at" + " FOR UPDATE SKIP LOCKED LIMIT 1" + ") " + f"UPDATE {table} AS jobs SET" + " state = 'in_flight', lease_token = %s," + " lease_expires_at = now() + (%s * interval '1 second')," + " attempt_count = attempt_count + 1, updated_at = now()" + " FROM candidate WHERE jobs.task_id = candidate.task_id" + " RETURNING jobs.task_id, jobs.account_id, jobs.workflow_type," + " jobs.payload, jobs.lease_token, jobs.attempt_count" + ) + + async def create_schema(self) -> None: + """Bootstrap the example table for local development and tests. + + Production deployments should apply the equivalent DDL through their + migration system before starting web or worker processes. + """ + statements = [ + f"""CREATE TABLE IF NOT EXISTS {self._table} ( + task_id TEXT COLLATE "C" PRIMARY KEY, + account_id TEXT COLLATE "C" NOT NULL, + workflow_type TEXT NOT NULL, + payload JSONB NOT NULL, + state TEXT NOT NULL DEFAULT 'pending', + attempt_count INTEGER NOT NULL DEFAULT 0, + available_at TIMESTAMPTZ NOT NULL DEFAULT now(), + lease_token TEXT COLLATE "C", + lease_expires_at TIMESTAMPTZ, + last_error TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + completed_at TIMESTAMPTZ, + dead_lettered_at TIMESTAMPTZ, + CHECK (attempt_count >= 0), + CHECK ( + (state = 'in_flight') = + (lease_token IS NOT NULL AND lease_expires_at IS NOT NULL) + ) + )""", + f"""ALTER TABLE {self._table} + ADD COLUMN IF NOT EXISTS dead_lettered_at TIMESTAMPTZ""", + f"""ALTER TABLE {self._table} + DROP CONSTRAINT IF EXISTS {self._state_constraint}""", + f"""ALTER TABLE {self._table} + ADD CONSTRAINT {self._state_constraint} + CHECK (state IN ('pending', 'in_flight', 'completed', 'dead_lettered'))""", + f"""CREATE INDEX IF NOT EXISTS {self._table}_work_idx + ON {self._table} (available_at, created_at) + WHERE state = 'pending'""", + f"""CREATE INDEX IF NOT EXISTS {self._table}_lease_idx + ON {self._table} (lease_expires_at) + WHERE state = 'in_flight'""", + ] + async with self._pool.connection() as conn: + async with conn.transaction(): + # Web and worker processes may bootstrap concurrently in the + # reference deployment. Serialize this example-owned DDL. + await conn.execute( + "SELECT pg_advisory_xact_lock(hashtext(%s))", + (f"adcp.workflow_queue.schema:{self._table}",), + ) + for statement in statements: + await conn.execute(statement) + + async def enqueue( + self, + *, + task_id: str, + account_id: str, + workflow_type: str, + payload: dict[str, Any], + ) -> None: + """Persist work from a ``WorkflowHandoff`` enqueue callback.""" + if not task_id or not account_id or not workflow_type: + raise ValueError("task_id, account_id, and workflow_type are required") + serialized = json.dumps(payload, separators=(",", ":"), sort_keys=True) + sql = ( # noqa: S608 - table is validated in __init__ + f"INSERT INTO {self._table} " + "(task_id, account_id, workflow_type, payload) " + "VALUES (%s, %s, %s, %s::jsonb) ON CONFLICT (task_id) DO NOTHING " + "RETURNING task_id" + ) + async with self._pool.connection() as conn: + row = await ( + await conn.execute( + sql, + (task_id, account_id, workflow_type, serialized), + ) + ).fetchone() + if row is not None: + return + existing = await ( + await conn.execute( + f"SELECT account_id, workflow_type, payload " # noqa: S608 + f"FROM {self._table} WHERE task_id = %s", + (task_id,), + ) + ).fetchone() + if existing != (account_id, workflow_type, payload): + raise ValueError("workflow task_id already exists with different work") + + async def enqueue_from_handoff( + self, + task_ctx: Any, + *, + account_id: str, + workflow_type: str, + payload: dict[str, Any], + ) -> None: + """Adapter-friendly callback used with ``ctx.handoff_to_workflow``.""" + await self.enqueue( + task_id=task_ctx.id, + account_id=account_id, + workflow_type=workflow_type, + payload=payload, + ) + + async def claim(self) -> WorkflowJob | None: + """Recover expired leases and claim one eligible job.""" + lease_token = uuid.uuid4().hex + async with self._pool.connection() as conn: + await conn.execute( + f"UPDATE {self._table} SET state = 'pending', lease_token = NULL, " # noqa: S608 + "lease_expires_at = NULL, updated_at = now() " + "WHERE state = 'in_flight' AND lease_expires_at <= now()" + ) + row = await ( + await conn.execute( + self._sql_claim, + (lease_token, self._lease_seconds), + ) + ).fetchone() + if row is None: + return None + task_id, account_id, workflow_type, payload, token, attempts = row + payload_dict = payload if isinstance(payload, dict) else json.loads(payload) + return WorkflowJob( + task_id=str(task_id), + account_id=str(account_id), + workflow_type=str(workflow_type), + payload=payload_dict, + lease_token=str(token), + attempt_count=int(attempts), + ) + + def _retry_delay(self, attempt_count: int) -> float: + exponent = min(max(attempt_count - 1, 0), 30) + return min( + self._max_retry_seconds, + self._retry_base_seconds * (2**exponent), + ) + + async def _release(self, job: WorkflowJob, exc: Exception) -> None: + # Persist the class for operations without writing arbitrary exception + # text (which may contain request data or credentials) into PostgreSQL. + message = type(exc).__name__ + retry_delay = self._retry_delay(job.attempt_count) + async with self._pool.connection() as conn: + await conn.execute( + f"UPDATE {self._table} SET state = 'pending', " # noqa: S608 + "available_at = now() + (%s * interval '1 second'), " + "lease_token = NULL, " + "lease_expires_at = NULL, last_error = %s, updated_at = now() " + "WHERE task_id = %s AND state = 'in_flight' AND lease_token = %s", + (retry_delay, message, job.task_id, job.lease_token), + ) + + async def _dead_letter(self, job: WorkflowJob, reason: str) -> None: + async with self._pool.connection() as conn: + cursor = await conn.execute( + f"UPDATE {self._table} SET state = 'dead_lettered', " # noqa: S608 + "lease_token = NULL, lease_expires_at = NULL, last_error = %s, " + "dead_lettered_at = now(), updated_at = now() " + "WHERE task_id = %s AND state = 'in_flight' AND lease_token = %s", + (reason, job.task_id, job.lease_token), + ) + if cursor.rowcount != 1: + raise RuntimeError("workflow lease was lost before dead-lettering") + + async def _handle_failure(self, job: WorkflowJob, exc: Exception) -> None: + if job.attempt_count < self._max_attempts: + await self._release(job, exc) + logger.warning( + "Workflow job %s failed on attempt %d with %s; released for retry", + job.task_id, + job.attempt_count, + type(exc).__name__, + ) + return + + await self._registry.fail(job.task_id, dict(_RETRY_EXHAUSTED_ERROR)) + await self._dead_letter(job, type(exc).__name__) + logger.error( + "Workflow job %s exhausted %d attempts with %s and was dead-lettered", + job.task_id, + job.attempt_count, + type(exc).__name__, + ) + + async def _handle_unverified_failure(self, job: WorkflowJob, exc: Exception) -> None: + """Bound failures before account ownership has been established.""" + if job.attempt_count < self._max_attempts: + await self._release(job, exc) + logger.warning( + "Workflow task lookup failed on attempt %d with %s; released for retry", + job.attempt_count, + type(exc).__name__, + ) + return + + # Never call the task-id-only registry mutation methods until the + # account-scoped get above has positively established ownership. + await self._dead_letter(job, type(exc).__name__) + logger.error( + "Workflow task lookup exhausted %d attempts with %s; dead-lettered", + job.attempt_count, + type(exc).__name__, + ) + + async def _acknowledge(self, job: WorkflowJob) -> None: + async with self._pool.connection() as conn: + cursor = await conn.execute( + f"UPDATE {self._table} SET state = 'completed', " # noqa: S608 + "lease_token = NULL, lease_expires_at = NULL, last_error = NULL, " + "completed_at = now(), updated_at = now() " + "WHERE task_id = %s AND state = 'in_flight' AND lease_token = %s", + (job.task_id, job.lease_token), + ) + if cursor.rowcount != 1: + raise RuntimeError("workflow lease was lost before acknowledgement") + + async def process_one(self, handler: WorkflowHandler) -> bool: + """Run one leased job and complete the matching SDK task.""" + job = await self.claim() + if job is None: + return False + try: + task = await self._registry.get( + job.task_id, + expected_account_id=job.account_id, + ) + except asyncio.CancelledError: + raise + except Exception as exc: + await self._handle_unverified_failure(job, exc) + return True + + if task is None: + # Keep this transition outside the generic failure path. If the + # queue write fails, a later lease retries only dead-lettering and + # can never mutate a task belonging to another account. + await self._dead_letter(job, "task_missing_or_account_mismatch") + logger.error( + "Workflow job %s has no matching account-scoped task; dead-lettered", + job.task_id, + ) + return True + if task.get("state") == "failed" and task.get("error") == _RETRY_EXHAUSTED_ERROR: + # Registry failure and queue dead-lettering are separate commits. + # Finish the second transition after a crash between them without + # running the handler again. + await self._dead_letter(job, "retry_budget_exhausted") + return True + if task.get("state") in {"completed", "failed"}: + # A prior worker may have committed terminal task state and died + # before acknowledging this queue lease. Do not rerun the business + # effect; reconcile the queue row to terminal state. + await self._acknowledge(job) + return True + if job.attempt_count > self._max_attempts: + # A prior attempt exhausted the handler budget but could not + # persist terminal registry state. Retry only finalization; never + # execute the business handler beyond max_attempts. + await self._registry.fail(job.task_id, dict(_RETRY_EXHAUSTED_ERROR)) + await self._dead_letter(job, "retry_budget_exhausted") + return True + + try: + result = await handler(job) + await self._registry.complete(job.task_id, result) + except asyncio.CancelledError: + raise + except Exception as exc: + await self._handle_failure(job, exc) + return True + + # A failed acknowledgement must not enter the handler retry path: the + # registry is already terminal, and lease recovery reconciles the row. + await self._acknowledge(job) + return True + + async def run_worker( + self, + handler: WorkflowHandler, + *, + poll_interval: float = 1.0, + ) -> None: + """Process jobs until cancellation; expired leases recover on restart.""" + if poll_interval <= 0: + raise ValueError("poll_interval must be positive") + while True: + try: + processed = await self.process_one(handler) + except asyncio.CancelledError: + raise + except Exception: + logger.exception("Workflow worker iteration failed; retrying") + processed = False + if not processed: + await asyncio.sleep(poll_interval) + + async def get(self, task_id: str) -> dict[str, Any] | None: + """Read queue state for operations and recovery tests.""" + async with self._pool.connection() as conn: + row = await ( + await conn.execute( + f"SELECT account_id, workflow_type, state, attempt_count, last_error " # noqa: S608 + f"FROM {self._table} WHERE task_id = %s", + (task_id,), + ) + ).fetchone() + if row is None: + return None + return { + "account_id": str(row[0]), + "workflow_type": str(row[1]), + "state": str(row[2]), + "attempt_count": int(row[3]), + "last_error": row[4], + } + + +__all__ = [ + "DEFAULT_MAX_ATTEMPTS", + "DEFAULT_MAX_RETRY_SECONDS", + "DEFAULT_RETRY_BASE_SECONDS", + "DEFAULT_WORKFLOW_TABLE", + "PgWorkflowQueue", + "WorkflowHandler", + "WorkflowJob", +] diff --git a/examples/v3_reference_seller/tests/test_durable_tasks.py b/examples/v3_reference_seller/tests/test_durable_tasks.py new file mode 100644 index 000000000..4ab5772e2 --- /dev/null +++ b/examples/v3_reference_seller/tests/test_durable_tasks.py @@ -0,0 +1,294 @@ +"""Configuration tests for the production task/outbox example.""" + +from __future__ import annotations + +import asyncio +import base64 +import sys +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock + +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import ed25519 + +_HERE = Path(__file__).resolve().parent +sys.path.insert(0, str(_HERE.parent)) + +from src.durable_tasks import DurableTaskWiring # noqa: E402 +from src.workflow_queue import PgWorkflowQueue # noqa: E402 + + +def _write_signing_key(path: Path) -> None: + private_key = ed25519.Ed25519PrivateKey.generate() + path.write_bytes( + private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + ) + + +def _environment(key_path: Path) -> dict[str, str]: + return { + "ADCP_ENV": "production", + "ADCP_TASK_DATABASE_URL": "postgresql://postgres@localhost/adcp", + "ADCP_TASK_WEBHOOK_ENCRYPTION_KEY": base64.b64encode(b"k" * 32).decode(), + "ADCP_WEBHOOK_SIGNING_KEY_PATH": str(key_path), + "ADCP_WEBHOOK_SIGNING_KEY_ID": "reference-webhook-key", + "ADCP_WEBHOOK_SIGNING_ALG": "ed25519", + "ADCP_TASK_WEBHOOK_RETRY_HORIZON_SECONDS": "172800", + } + + +def test_local_development_may_omit_durable_bundle() -> None: + assert DurableTaskWiring.from_env({"ADCP_ENV": "development"}) is None + + +def test_production_fails_before_boot_when_bundle_is_missing() -> None: + with pytest.raises(ValueError, match="ADCP_TASK_DATABASE_URL"): + DurableTaskWiring.from_env({"ADCP_ENV": "production"}) + + +def test_partial_bundle_lists_every_missing_field() -> None: + with pytest.raises(ValueError) as exc_info: + DurableTaskWiring.from_env( + { + "ADCP_ENV": "development", + "ADCP_TASK_DATABASE_URL": "postgresql://postgres@localhost/adcp", + } + ) + message = str(exc_info.value) + assert "ADCP_TASK_WEBHOOK_ENCRYPTION_KEY" in message + assert "ADCP_WEBHOOK_SIGNING_KEY_PATH" in message + assert "ADCP_WEBHOOK_SIGNING_KEY_ID" in message + + +def test_encryption_key_must_decode_to_32_bytes(tmp_path: Path) -> None: + key_path = tmp_path / "webhook-signing.pem" + _write_signing_key(key_path) + environ = _environment(key_path) + environ["ADCP_TASK_WEBHOOK_ENCRYPTION_KEY"] = base64.b64encode(b"short").decode() + + with pytest.raises(ValueError, match="exactly 32 bytes"): + DurableTaskWiring.from_env(environ) + + +def test_retry_horizon_is_validated_before_boot(tmp_path: Path) -> None: + key_path = tmp_path / "webhook-signing.pem" + _write_signing_key(key_path) + environ = _environment(key_path) + environ["ADCP_TASK_WEBHOOK_RETRY_HORIZON_SECONDS"] = "60" + + with pytest.raises(ValueError, match="between 86400 and 604800"): + DurableTaskWiring.from_env(environ) + + +def test_workflow_retry_backoff_is_exponential_and_capped() -> None: + queue = PgWorkflowQueue( + pool=MagicMock(), + registry=MagicMock(), + max_attempts=5, + retry_base_seconds=2, + max_retry_seconds=5, + ) + + assert queue._retry_delay(1) == 2 # noqa: SLF001 + assert queue._retry_delay(2) == 4 # noqa: SLF001 + assert queue._retry_delay(3) == 5 # noqa: SLF001 + + +@pytest.mark.parametrize( + ("kwargs", "message"), + [ + ({"max_attempts": 0}, "max_attempts"), + ({"retry_base_seconds": 0}, "retry_base_seconds"), + ( + {"retry_base_seconds": 2, "max_retry_seconds": 1}, + "max_retry_seconds", + ), + ], +) +def test_workflow_retry_configuration_is_validated( + kwargs: dict[str, int], + message: str, +) -> None: + with pytest.raises(ValueError, match=message): + PgWorkflowQueue(pool=MagicMock(), registry=MagicMock(), **kwargs) + + +def test_complete_bundle_builds_atomic_registry_outbox_pair(tmp_path: Path) -> None: + key_path = tmp_path / "webhook-signing.pem" + _write_signing_key(key_path) + + wiring = DurableTaskWiring.from_env(_environment(key_path)) + + assert wiring is not None + try: + assert wiring.registry.task_webhook_outbox is wiring.outbox + assert wiring.workflow_queue._registry is wiring.registry # noqa: SLF001 + assert wiring.registry.atomic_task_webhook_outbox is True + assert wiring.idempotency_backend is not None + assert wiring.idempotency_backend._lock_pool is wiring.lock_pool # noqa: SLF001 + assert wiring.idempotency is not None + assert wiring.idempotency.raise_on_persist_error is True + assert wiring.retry_horizon_seconds == 172800 + assert wiring.signing_algorithm == "ed25519" + finally: + asyncio.run(wiring.shutdown()) + + +def test_worker_bundle_omits_unused_idempotency_pool(tmp_path: Path) -> None: + key_path = tmp_path / "webhook-signing.pem" + _write_signing_key(key_path) + + wiring = DurableTaskWiring.from_env( + _environment(key_path), + include_idempotency=False, + ) + + assert wiring is not None + try: + assert wiring.lock_pool is None + assert wiring.idempotency_backend is None + assert wiring.idempotency is None + finally: + asyncio.run(wiring.shutdown()) + + +def test_complete_bundle_passes_capability_wiring_preflight(tmp_path: Path) -> None: + from src.platform import V3ReferenceSeller + + from adcp.decisioning import create_adcp_server_from_platform + + key_path = tmp_path / "webhook-signing.pem" + _write_signing_key(key_path) + wiring = DurableTaskWiring.from_env(_environment(key_path)) + assert wiring is not None + assert wiring.idempotency is not None + executor = None + try: + seller = V3ReferenceSeller( + sessionmaker=lambda: None, # type: ignore[arg-type] + upstream_api_key="test-key", + mock_upstream_url=None, + webhook_signing_alg=wiring.signing_algorithm, + webhook_retry_horizon_seconds=wiring.retry_horizon_seconds, + idempotency=wiring.idempotency, + ) + + handler, executor, registry = create_adcp_server_from_platform( + seller, + registry=wiring.registry, + ) + assert handler.get_advertised_tools() + assert registry is wiring.registry + finally: + if executor is not None: + executor.shutdown(wait=True) + asyncio.run(wiring.shutdown()) + + +@pytest.mark.asyncio +async def test_startup_failure_closes_every_resource() -> None: + pool = MagicMock(open=AsyncMock(), close=AsyncMock()) + lock_pool = MagicMock(open=AsyncMock(), close=AsyncMock()) + sender = MagicMock(aclose=AsyncMock()) + registry = MagicMock(create_schema=AsyncMock(side_effect=RuntimeError("DDL failed"))) + outbox = MagicMock(create_schema=AsyncMock()) + workflow_queue = MagicMock(create_schema=AsyncMock()) + backend = MagicMock(create_schema=AsyncMock()) + wiring = DurableTaskWiring( + pool=pool, + lock_pool=lock_pool, + sender=sender, + outbox=outbox, + registry=registry, + workflow_queue=workflow_queue, + idempotency_backend=backend, + idempotency=None, + retry_horizon_seconds=86400, + signing_algorithm="ed25519", + ) + + with pytest.raises(RuntimeError, match="DDL failed"): + await wiring.startup() + + sender.aclose.assert_awaited_once() + lock_pool.close.assert_awaited_once() + pool.close.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_startup_cancellation_closes_every_resource_and_reraises() -> None: + pool = MagicMock(open=AsyncMock(), close=AsyncMock()) + lock_pool = MagicMock(open=AsyncMock(), close=AsyncMock()) + sender_close_started = asyncio.Event() + release_sender_close = asyncio.Event() + + async def close_sender() -> None: + sender_close_started.set() + await release_sender_close.wait() + + sender = MagicMock(aclose=AsyncMock(side_effect=close_sender)) + schema_started = asyncio.Event() + + async def wait_forever() -> None: + schema_started.set() + await asyncio.Event().wait() + + registry = MagicMock(create_schema=AsyncMock(side_effect=wait_forever)) + wiring = DurableTaskWiring( + pool=pool, + lock_pool=lock_pool, + sender=sender, + outbox=MagicMock(create_schema=AsyncMock()), + registry=registry, + workflow_queue=MagicMock(create_schema=AsyncMock()), + idempotency_backend=MagicMock(create_schema=AsyncMock()), + idempotency=None, + retry_horizon_seconds=86400, + signing_algorithm="ed25519", + ) + + startup = asyncio.create_task(wiring.startup()) + await schema_started.wait() + startup.cancel() + await sender_close_started.wait() + startup.cancel() + release_sender_close.set() + + done, pending = await asyncio.wait({startup}) + assert done == {startup} + assert not pending + assert startup.cancelled() + + sender.aclose.assert_awaited_once() + lock_pool.close.assert_awaited_once() + pool.close.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_shutdown_continues_after_close_failure() -> None: + pool = MagicMock(close=AsyncMock()) + lock_pool = MagicMock(close=AsyncMock()) + sender = MagicMock(aclose=AsyncMock(side_effect=RuntimeError("sender close failed"))) + wiring = DurableTaskWiring( + pool=pool, + lock_pool=lock_pool, + sender=sender, + outbox=MagicMock(), + registry=MagicMock(), + workflow_queue=MagicMock(), + idempotency_backend=None, + idempotency=None, + retry_horizon_seconds=86400, + signing_algorithm="ed25519", + ) + + with pytest.raises(RuntimeError, match="sender close failed"): + await wiring.shutdown() + + lock_pool.close.assert_awaited_once() + pool.close.assert_awaited_once() diff --git a/examples/v3_reference_seller/tests/test_smoke.py b/examples/v3_reference_seller/tests/test_smoke.py index 1af9eaf74..d607f607c 100644 --- a/examples/v3_reference_seller/tests/test_smoke.py +++ b/examples/v3_reference_seller/tests/test_smoke.py @@ -205,6 +205,8 @@ def test_platform_default_does_not_advertise_webhook_signing() -> None: from src.platform import V3ReferenceSeller assert V3ReferenceSeller.capabilities.webhook_signing is None + assert V3ReferenceSeller.capabilities.adcp is not None + assert V3ReferenceSeller.capabilities.adcp.idempotency.supported is False def test_platform_advertises_webhook_signing_when_alg_passed() -> None: @@ -219,15 +221,71 @@ def test_platform_advertises_webhook_signing_when_alg_passed() -> None: upstream_api_key="test-key", mock_upstream_url=None, webhook_signing_alg="ed25519", + webhook_retry_horizon_seconds=172800, ) ws = seller.capabilities.webhook_signing assert ws is not None assert ws.supported is True assert ws.profile == "adcp/webhook-signing/v1" + assert ws.delivery_retry_horizon_seconds == 172800 assert ws.algorithms is not None assert [a.value for a in ws.algorithms] == ["ed25519"] +@pytest.mark.asyncio +async def test_platform_runtime_idempotency_wrapper_replays() -> None: + from src.platform import V3ReferenceSeller + + from adcp.decisioning import RequestContext + from adcp.server.idempotency import IdempotencyStore, MemoryBackend, is_wrapped + + class _Seller(V3ReferenceSeller): + calls = 0 + + async def create_media_buy(self, req, ctx): + del req, ctx + self.calls += 1 + return {"status": "completed", "media_buy_id": "mb_1"} + + seller = _Seller( + sessionmaker=lambda: None, # type: ignore[arg-type] + upstream_api_key="test-key", + mock_upstream_url=None, + idempotency=IdempotencyStore(MemoryBackend(), ttl_seconds=86400), + ) + assert is_wrapped(seller.create_media_buy) + context = RequestContext(caller_identity="buyer", tenant_id="tenant") + request = {"idempotency_key": "same-key", "budget": 100} + + first = await seller.create_media_buy(request, context) + replay = await seller.create_media_buy(request, context) + + assert first == {"status": "completed", "media_buy_id": "mb_1"} + assert replay == { + "status": "completed", + "media_buy_id": "mb_1", + "replayed": True, + } + assert seller.calls == 1 + + +def test_platform_idempotency_method_selection_is_explicit() -> None: + from src.platform import V3ReferenceSeller + + from adcp.server.idempotency import IdempotencyStore, MemoryBackend, is_wrapped + + seller = V3ReferenceSeller( + sessionmaker=lambda: None, # type: ignore[arg-type] + upstream_api_key="test-key", + mock_upstream_url=None, + idempotency=IdempotencyStore(MemoryBackend(), ttl_seconds=86400), + method_level_idempotency_methods=("sync_creatives",), + ) + + assert not is_wrapped(seller.create_media_buy) + assert is_wrapped(seller.sync_creatives) + + @pytest.mark.asyncio @pytest.mark.parametrize("authenticated_tenant", ["tenant-a", None]) async def test_bearer_context_rejects_cross_tenant_rebinding( diff --git a/examples/v3_reference_seller/tests/test_smoke_broadening.py b/examples/v3_reference_seller/tests/test_smoke_broadening.py index 025db96e6..b8f44f5d1 100644 --- a/examples/v3_reference_seller/tests/test_smoke_broadening.py +++ b/examples/v3_reference_seller/tests/test_smoke_broadening.py @@ -24,6 +24,7 @@ from __future__ import annotations +import json import sys from datetime import datetime, timezone from pathlib import Path @@ -728,9 +729,9 @@ async def test_create_media_buy_sync_polls_to_success_on_pending_approval( """When the upstream returns ``pending_approval`` + ``approval_task_id``, the platform sync-polls until the approval task completes and returns the full :class:`CreateMediaBuySuccessResponse` with ``media_buy_id``. - AdCP storyboards expect synchronous create — production adopters with - slow real-world approvals swap to ``ctx.handoff_to_task`` (see the - docstring in ``platform.create_media_buy``).""" + AdCP storyboards expect synchronous create. Production adapters with + slow real-world approvals persist them through ``WorkflowHandoff`` and a + durable queue (see ``platform.create_media_buy``).""" from adcp.types import CreateMediaBuyRequest, CreateMediaBuySuccessResponse respx_mock.post("/v1/orders").mock( @@ -1568,9 +1569,12 @@ async def test_provide_performance_feedback_posts_capi_conversion( resp = await platform.provide_performance_feedback(req, ctx) assert route.called assert resp.success is True - body = respx_mock.calls.last.request.read().decode("utf-8") - assert "conversion_rate" in body - assert "0.87" in body + body = json.loads(respx_mock.calls.last.request.read()) + assert body["conversions"][0]["event_name"] == "conversion_rate" + assert body["conversions"][0]["value"] == 0.87 + assert body["conversions"][0]["dedup_key"] == ( + f"adcp:{req.idempotency_key}:performance-feedback" + ) @pytest.mark.asyncio diff --git a/examples/v3_reference_seller/tests/test_worker.py b/examples/v3_reference_seller/tests/test_worker.py new file mode 100644 index 000000000..d2469615a --- /dev/null +++ b/examples/v3_reference_seller/tests/test_worker.py @@ -0,0 +1,122 @@ +"""Lifecycle tests for the separately supervised durable worker.""" + +from __future__ import annotations + +import asyncio +import signal +import sys +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +_HERE = Path(__file__).resolve().parent +sys.path.insert(0, str(_HERE.parent)) + + +@pytest.mark.asyncio +async def test_stop_event_cancels_worker_before_resource_shutdown(monkeypatch) -> None: + import src.worker as worker_module + + started = asyncio.Event() + cancelled = asyncio.Event() + + async def run_outbox() -> None: + started.set() + try: + await asyncio.Event().wait() + finally: + cancelled.set() + + wiring = SimpleNamespace( + startup=AsyncMock(), + shutdown=AsyncMock(), + outbox=SimpleNamespace(run_worker=run_outbox), + workflow_queue=SimpleNamespace(), + ) + monkeypatch.setattr( + worker_module.DurableTaskWiring, + "from_env", + lambda **_kwargs: wiring, + ) + stop_event = asyncio.Event() + + async def request_stop() -> None: + await started.wait() + stop_event.set() + + stopper = asyncio.create_task(request_stop()) + await worker_module.run(stop_event=stop_event) + + assert stopper.done() + assert cancelled.is_set() + wiring.startup.assert_awaited_once() + wiring.shutdown.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_worker_failure_wins_when_shutdown_is_also_ready(monkeypatch) -> None: + import src.worker as worker_module + + stop_event = asyncio.Event() + + async def fail_outbox() -> None: + stop_event.set() + raise RuntimeError("outbox failed") + + wiring = SimpleNamespace( + startup=AsyncMock(), + shutdown=AsyncMock(), + outbox=SimpleNamespace(run_worker=fail_outbox), + workflow_queue=SimpleNamespace(), + ) + monkeypatch.setattr( + worker_module.DurableTaskWiring, + "from_env", + lambda **_kwargs: wiring, + ) + real_wait = asyncio.wait + + async def wait_for_all(awaitables, *, return_when): + assert return_when is asyncio.FIRST_COMPLETED + return await real_wait(awaitables, return_when=asyncio.ALL_COMPLETED) + + monkeypatch.setattr(worker_module.asyncio, "wait", wait_for_all) + + with pytest.raises(RuntimeError, match="outbox failed"): + await worker_module.run(stop_event=stop_event) + + wiring.shutdown.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_signal_runner_installs_and_removes_sigterm_handler(monkeypatch) -> None: + import src.worker as worker_module + + loop = asyncio.get_running_loop() + callbacks = {} + removed = [] + + monkeypatch.setattr( + loop, + "add_signal_handler", + lambda signum, callback: callbacks.__setitem__(signum, callback), + ) + monkeypatch.setattr( + loop, + "remove_signal_handler", + lambda signum: removed.append(signum) or True, + ) + + async def fake_run(*, stop_event, workflow_handler=None) -> None: + assert workflow_handler is None + callbacks[signal.SIGTERM]() + assert stop_event.is_set() + + monkeypatch.setattr(worker_module, "run", fake_run) + + await worker_module.run_with_signals() + + assert set(callbacks) == {signal.SIGTERM, signal.SIGINT} + assert removed == [signal.SIGTERM, signal.SIGINT] diff --git a/src/adcp/server/idempotency/store.py b/src/adcp/server/idempotency/store.py index bb7bd2843..ebbee1f47 100644 --- a/src/adcp/server/idempotency/store.py +++ b/src/adcp/server/idempotency/store.py @@ -111,6 +111,11 @@ class IdempotencyStore: to :func:`canonical_json_sha256`. Exposed for tests and for anyone who wants to experiment with alternative equivalence rules — though note the spec mandates RFC 8785 JCS for interop. + :param raise_on_persist_error: When true, translate a backend write failure + after handler completion to a retryable ``SERVICE_UNAVAILABLE`` error + instead of returning an unrecorded success. Production handlers using + this mode must independently deduplicate their business side effects, + because the handler may already have crossed that boundary. """ def __init__( @@ -120,6 +125,7 @@ def __init__( hash_fn: Callable[[dict[str, Any]], str] = canonical_json_sha256, *, clock: Callable[[], float] = time.time, + raise_on_persist_error: bool = False, ) -> None: if not _MIN_TTL_SECONDS <= ttl_seconds <= _MAX_TTL_SECONDS: raise ValueError( @@ -131,6 +137,7 @@ def __init__( self.ttl_seconds = ttl_seconds self._hash_fn = hash_fn self._clock = clock + self.raise_on_persist_error = raise_on_persist_error self._warned_fallback_hold = False @asynccontextmanager @@ -280,7 +287,7 @@ async def _execute_locked() -> Any: # but it prevents a concurrent duplicate handler execution. try: await self.backend.put(scope_key, idempotency_key, entry) - except Exception: + except Exception as exc: logger.warning( "Idempotency cache put failed for scope=%s key_prefix=%s — " "handler completed but a subsequent retry with this key will " @@ -290,6 +297,22 @@ async def _execute_locked() -> Any: idempotency_key[:8], exc_info=True, ) + if self.raise_on_persist_error: + # Local import avoids making the server middleware + # module depend on the decisioning package at import + # time. The exception is translated consistently by + # both MCP and A2A server boundaries. + from adcp.decisioning.types import AdcpError + + raise AdcpError( + "SERVICE_UNAVAILABLE", + message=( + "Idempotency persistence failed after handler completion; " + "the outcome may be unknown. Retry with the same key only " + "when downstream effects are independently deduplicated." + ), + recovery="transient", + ) from exc return response execution_task = asyncio.create_task(_execute_locked()) diff --git a/tests/conformance/decisioning/test_pg_reference_workflow_queue.py b/tests/conformance/decisioning/test_pg_reference_workflow_queue.py new file mode 100644 index 000000000..62da421d7 --- /dev/null +++ b/tests/conformance/decisioning/test_pg_reference_workflow_queue.py @@ -0,0 +1,706 @@ +"""Restart-recovery test for the reference PostgreSQL WorkflowHandoff queue. + +Set ``ADCP_PG_TEST_URL`` to run this test against PostgreSQL. +""" + +from __future__ import annotations + +import os +import secrets +import sys +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +import pytest +from pydantic import BaseModel + +psycopg_pool = pytest.importorskip("psycopg_pool") + +TEST_URL = os.environ.get("ADCP_PG_TEST_URL") +if not TEST_URL: + pytest.skip( + "ADCP_PG_TEST_URL not set — skipping reference workflow queue test", + allow_module_level=True, + ) + +_EXAMPLE = Path(__file__).resolve().parents[3] / "examples" / "v3_reference_seller" +sys.path.insert(0, str(_EXAMPLE)) + +from adcp.decisioning import Account, PgTaskRegistry, RequestContext # noqa: E402 +from adcp.decisioning.dispatch import _project_workflow_handoff # noqa: E402 +from src.workflow_queue import PgWorkflowQueue # noqa: E402 + + +class _Request(BaseModel): + context: dict[str, str] | None = None + + +@pytest.mark.asyncio +async def test_workflow_handoff_recovers_expired_lease_after_worker_restart() -> None: + suffix = secrets.token_hex(6) + task_table = f"test_dtasks_{suffix}" + workflow_table = f"test_workflows_{suffix}" + account_id = "tenant-a:account-a" + expected_result = {"media_buy_id": "mb_recovered", "status": "active"} + + async with psycopg_pool.AsyncConnectionPool( + TEST_URL, + min_size=1, + max_size=4, + open=False, + ) as web_pool: + await web_pool.open() + registry = PgTaskRegistry(pool=web_pool, _table=task_table) + queue = PgWorkflowQueue( + pool=web_pool, + registry=registry, + table=workflow_table, + lease_seconds=2, + ) + await registry.create_schema() + await queue.create_schema() + executor = ThreadPoolExecutor(max_workers=1) + try: + ctx = RequestContext( + account=Account(id=account_id), + tenant_id="tenant-a", + caller_identity="buyer-a", + ) + + async def enqueue(task_ctx) -> None: + await queue.enqueue_from_handoff( + task_ctx, + account_id=account_id, + workflow_type="manual_media_buy_approval", + payload={"result": expected_result}, + ) + + submitted = await _project_workflow_handoff( + ctx.handoff_to_workflow(enqueue), + ctx, + method_name="create_media_buy", + registry=registry, + executor=executor, + request_params=_Request(context={"trace_id": "restart-test"}), + ) + task_id = submitted["task_id"] + assert submitted == {"task_id": task_id, "status": "submitted"} + + # Worker process 1 claims the row and dies before completion. + async with psycopg_pool.AsyncConnectionPool( + TEST_URL, + min_size=1, + max_size=2, + open=False, + ) as first_worker_pool: + await first_worker_pool.open() + first_worker_queue = PgWorkflowQueue( + pool=first_worker_pool, + registry=registry, + table=workflow_table, + lease_seconds=2, + ) + first_claim = await first_worker_queue.claim() + assert first_claim is not None + assert first_claim.task_id == task_id + assert first_claim.attempt_count == 1 + + # Advance only the database lease, avoiding a wall-clock sleep. + async with web_pool.connection() as conn: + await conn.execute( + f"UPDATE {workflow_table} " # noqa: S608 + "SET lease_expires_at = now() - interval '1 second' " + "WHERE task_id = %s", + (task_id,), + ) + + # Worker process 2 starts with fresh queue and registry objects, + # reclaims the expired lease, and completes the original task. + async with psycopg_pool.AsyncConnectionPool( + TEST_URL, + min_size=1, + max_size=2, + open=False, + ) as replacement_pool: + await replacement_pool.open() + replacement_registry = PgTaskRegistry( + pool=replacement_pool, + _table=task_table, + ) + replacement_queue = PgWorkflowQueue( + pool=replacement_pool, + registry=replacement_registry, + table=workflow_table, + lease_seconds=2, + ) + + async def complete(job): + assert job.attempt_count == 2 + return job.payload["result"] + + assert await replacement_queue.process_one(complete) is True + queue_record = await replacement_queue.get(task_id) + task_record = await replacement_registry.get( + task_id, + expected_account_id=account_id, + ) + + assert queue_record is not None + assert queue_record["state"] == "completed" + assert queue_record["attempt_count"] == 2 + assert task_record is not None + assert task_record["state"] == "completed" + assert task_record["result"] == expected_result + assert task_record["context"] == {"trace_id": "restart-test"} + finally: + executor.shutdown(wait=True) + async with web_pool.connection() as conn: + await conn.execute(f"DROP TABLE IF EXISTS {workflow_table}") # noqa: S608 + await conn.execute(f"DROP TABLE IF EXISTS {task_table}") # noqa: S608 + + +@pytest.mark.asyncio +async def test_workflow_retries_are_bounded_then_task_fails_and_job_dead_letters() -> None: + suffix = secrets.token_hex(6) + task_table = f"test_dtasks_{suffix}" + workflow_table = f"test_workflows_{suffix}" + account_id = "tenant-a:account-a" + + async with psycopg_pool.AsyncConnectionPool( + TEST_URL, + min_size=1, + max_size=4, + open=False, + ) as pool: + await pool.open() + registry = PgTaskRegistry(pool=pool, _table=task_table) + queue = PgWorkflowQueue( + pool=pool, + registry=registry, + table=workflow_table, + max_attempts=2, + retry_base_seconds=1, + max_retry_seconds=2, + ) + await registry.create_schema() + await queue.create_schema() + try: + task_id = await registry.issue( + account_id=account_id, + task_type="create_media_buy", + ) + await queue.enqueue( + task_id=task_id, + account_id=account_id, + workflow_type="manual_media_buy_approval", + payload={"upstream_order_id": "order-1"}, + ) + attempts = 0 + + async def fail(_job): + nonlocal attempts + attempts += 1 + raise RuntimeError("sensitive upstream failure") + + assert await queue.process_one(fail) is True + first_record = await queue.get(task_id) + assert first_record is not None + assert first_record["state"] == "pending" + assert first_record["attempt_count"] == 1 + assert first_record["last_error"] == "RuntimeError" + + # Advance the retry schedule without a wall-clock sleep. + async with pool.connection() as conn: + await conn.execute( + f"UPDATE {workflow_table} SET available_at = now() " # noqa: S608 + "WHERE task_id = %s", + (task_id,), + ) + + assert await queue.process_one(fail) is True + queue_record = await queue.get(task_id) + task_record = await registry.get( + task_id, + expected_account_id=account_id, + ) + + assert attempts == 2 + assert queue_record is not None + assert queue_record["state"] == "dead_lettered" + assert queue_record["attempt_count"] == 2 + assert queue_record["last_error"] == "RuntimeError" + assert task_record is not None + assert task_record["state"] == "failed" + assert task_record["error"] == { + "code": "INTERNAL_ERROR", + "message": "Workflow processing exhausted its retry budget.", + "recovery": "terminal", + } + assert await queue.process_one(fail) is False + finally: + async with pool.connection() as conn: + await conn.execute(f"DROP TABLE IF EXISTS {workflow_table}") # noqa: S608 + await conn.execute(f"DROP TABLE IF EXISTS {task_table}") # noqa: S608 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("issue_task", "job_account_id"), + [ + pytest.param(False, "tenant-a:account-a", id="missing-task"), + pytest.param(True, "tenant-b:account-b", id="account-mismatch"), + ], +) +async def test_missing_or_mismatched_workflow_dead_letters_without_running( + issue_task: bool, + job_account_id: str, +) -> None: + suffix = secrets.token_hex(6) + task_table = f"test_dtasks_{suffix}" + workflow_table = f"test_workflows_{suffix}" + + async with psycopg_pool.AsyncConnectionPool( + TEST_URL, + min_size=1, + max_size=4, + open=False, + ) as pool: + await pool.open() + registry = PgTaskRegistry(pool=pool, _table=task_table) + queue = PgWorkflowQueue(pool=pool, registry=registry, table=workflow_table) + await registry.create_schema() + await queue.create_schema() + try: + task_id = ( + await registry.issue( + account_id="tenant-a:account-a", + task_type="create_media_buy", + ) + if issue_task + else f"task_missing_{suffix}" + ) + await queue.enqueue( + task_id=task_id, + account_id=job_account_id, + workflow_type="manual_media_buy_approval", + payload={"upstream_order_id": "order-1"}, + ) + handler_called = False + + async def handler(_job): + nonlocal handler_called + handler_called = True + return {"status": "active"} + + assert await queue.process_one(handler) is True + queue_record = await queue.get(task_id) + task_record = await registry.get( + task_id, + expected_account_id="tenant-a:account-a", + ) + + assert handler_called is False + assert queue_record is not None + assert queue_record["state"] == "dead_lettered" + assert queue_record["attempt_count"] == 1 + assert queue_record["last_error"] == "task_missing_or_account_mismatch" + if issue_task: + assert task_record is not None + assert task_record["state"] == "submitted" + else: + assert task_record is None + finally: + async with pool.connection() as conn: + await conn.execute(f"DROP TABLE IF EXISTS {workflow_table}") # noqa: S608 + await conn.execute(f"DROP TABLE IF EXISTS {task_table}") # noqa: S608 + + +@pytest.mark.asyncio +async def test_failed_registry_commit_is_recovered_to_dead_letter_after_crash() -> None: + suffix = secrets.token_hex(6) + task_table = f"test_dtasks_{suffix}" + workflow_table = f"test_workflows_{suffix}" + account_id = "tenant-a:account-a" + terminal_error = { + "code": "INTERNAL_ERROR", + "message": "Workflow processing exhausted its retry budget.", + "recovery": "terminal", + } + + async with psycopg_pool.AsyncConnectionPool( + TEST_URL, + min_size=1, + max_size=4, + open=False, + ) as pool: + await pool.open() + registry = PgTaskRegistry(pool=pool, _table=task_table) + queue = PgWorkflowQueue( + pool=pool, + registry=registry, + table=workflow_table, + lease_seconds=2, + max_attempts=1, + ) + await registry.create_schema() + await queue.create_schema() + try: + task_id = await registry.issue( + account_id=account_id, + task_type="create_media_buy", + ) + await queue.enqueue( + task_id=task_id, + account_id=account_id, + workflow_type="manual_media_buy_approval", + payload={"upstream_order_id": "order-1"}, + ) + claimed = await queue.claim() + assert claimed is not None + + # Simulate a crash after the registry transaction commits but + # before the separate queue dead-letter update begins. + await registry.fail(task_id, terminal_error) + async with pool.connection() as conn: + await conn.execute( + f"UPDATE {workflow_table} " # noqa: S608 + "SET lease_expires_at = now() - interval '1 second' " + "WHERE task_id = %s", + (task_id,), + ) + + handler_called = False + + async def handler(_job): + nonlocal handler_called + handler_called = True + return {"status": "active"} + + assert await queue.process_one(handler) is True + queue_record = await queue.get(task_id) + + assert handler_called is False + assert queue_record is not None + assert queue_record["state"] == "dead_lettered" + assert queue_record["last_error"] == "retry_budget_exhausted" + finally: + async with pool.connection() as conn: + await conn.execute(f"DROP TABLE IF EXISTS {workflow_table}") # noqa: S608 + await conn.execute(f"DROP TABLE IF EXISTS {task_table}") # noqa: S608 + + +@pytest.mark.asyncio +async def test_create_schema_upgrades_pre_dead_letter_queue_table() -> None: + suffix = secrets.token_hex(6) + task_table = f"test_dtasks_{suffix}" + workflow_table = f"test_workflows_{suffix}" + + async with psycopg_pool.AsyncConnectionPool( + TEST_URL, + min_size=1, + max_size=4, + open=False, + ) as pool: + await pool.open() + registry = PgTaskRegistry(pool=pool, _table=task_table) + queue = PgWorkflowQueue(pool=pool, registry=registry, table=workflow_table) + await registry.create_schema() + async with pool.connection() as conn: + await conn.execute( + f"""CREATE TABLE {workflow_table} ( -- noqa: S608 + task_id TEXT COLLATE "C" PRIMARY KEY, + account_id TEXT COLLATE "C" NOT NULL, + workflow_type TEXT NOT NULL, + payload JSONB NOT NULL, + state TEXT NOT NULL DEFAULT 'pending', + attempt_count INTEGER NOT NULL DEFAULT 0, + available_at TIMESTAMPTZ NOT NULL DEFAULT now(), + lease_token TEXT COLLATE "C", + lease_expires_at TIMESTAMPTZ, + last_error TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + completed_at TIMESTAMPTZ, + CHECK (state IN ('pending', 'in_flight', 'completed')), + CHECK (attempt_count >= 0), + CHECK ( + (state = 'in_flight') = + (lease_token IS NOT NULL AND lease_expires_at IS NOT NULL) + ) + )""" + ) + try: + await queue.create_schema() + task_id = f"task_missing_{suffix}" + await queue.enqueue( + task_id=task_id, + account_id="tenant-a:account-a", + workflow_type="manual_media_buy_approval", + payload={"upstream_order_id": "order-1"}, + ) + + async def handler(_job): + raise AssertionError("missing task must not reach handler") + + assert await queue.process_one(handler) is True + queue_record = await queue.get(task_id) + assert queue_record is not None + assert queue_record["state"] == "dead_lettered" + finally: + async with pool.connection() as conn: + await conn.execute(f"DROP TABLE IF EXISTS {workflow_table}") # noqa: S608 + await conn.execute(f"DROP TABLE IF EXISTS {task_table}") # noqa: S608 + + +@pytest.mark.asyncio +async def test_registry_finalization_failure_never_reruns_exhausted_handler() -> None: + suffix = secrets.token_hex(6) + task_table = f"test_dtasks_{suffix}" + workflow_table = f"test_workflows_{suffix}" + account_id = "tenant-a:account-a" + + async with psycopg_pool.AsyncConnectionPool( + TEST_URL, + min_size=1, + max_size=4, + open=False, + ) as pool: + await pool.open() + registry = PgTaskRegistry(pool=pool, _table=task_table) + + class FailOnceRegistry: + def __init__(self) -> None: + self.fail_calls = 0 + + async def get(self, task_id, *, expected_account_id=None): + return await registry.get( + task_id, + expected_account_id=expected_account_id, + ) + + async def complete(self, task_id, result): + return await registry.complete(task_id, result) + + async def fail(self, task_id, error): + self.fail_calls += 1 + if self.fail_calls == 1: + raise RuntimeError("registry temporarily unavailable") + return await registry.fail(task_id, error) + + flaky_registry = FailOnceRegistry() + queue = PgWorkflowQueue( + pool=pool, + registry=flaky_registry, # type: ignore[arg-type] + table=workflow_table, + lease_seconds=2, + max_attempts=1, + ) + await registry.create_schema() + await queue.create_schema() + try: + task_id = await registry.issue( + account_id=account_id, + task_type="create_media_buy", + ) + await queue.enqueue( + task_id=task_id, + account_id=account_id, + workflow_type="manual_media_buy_approval", + payload={"upstream_order_id": "order-1"}, + ) + handler_calls = 0 + + async def fail_handler(_job): + nonlocal handler_calls + handler_calls += 1 + raise RuntimeError("upstream failed") + + with pytest.raises(RuntimeError, match="registry temporarily unavailable"): + await queue.process_one(fail_handler) + + async with pool.connection() as conn: + await conn.execute( + f"UPDATE {workflow_table} " # noqa: S608 + "SET lease_expires_at = now() - interval '1 second' " + "WHERE task_id = %s", + (task_id,), + ) + + assert await queue.process_one(fail_handler) is True + queue_record = await queue.get(task_id) + task_record = await registry.get( + task_id, + expected_account_id=account_id, + ) + + assert handler_calls == 1 + assert flaky_registry.fail_calls == 2 + assert queue_record is not None + assert queue_record["state"] == "dead_lettered" + assert task_record is not None + assert task_record["state"] == "failed" + finally: + async with pool.connection() as conn: + await conn.execute(f"DROP TABLE IF EXISTS {workflow_table}") # noqa: S608 + await conn.execute(f"DROP TABLE IF EXISTS {task_table}") # noqa: S608 + + +@pytest.mark.asyncio +async def test_mismatch_dead_letter_failure_never_fails_other_account_task() -> None: + suffix = secrets.token_hex(6) + task_table = f"test_dtasks_{suffix}" + workflow_table = f"test_workflows_{suffix}" + + async with psycopg_pool.AsyncConnectionPool( + TEST_URL, + min_size=1, + max_size=4, + open=False, + ) as pool: + await pool.open() + registry = PgTaskRegistry(pool=pool, _table=task_table) + + class FailOnceDeadLetterQueue(PgWorkflowQueue): + dead_letter_calls = 0 + + async def _dead_letter(self, job, reason): + self.dead_letter_calls += 1 + if self.dead_letter_calls == 1: + raise RuntimeError("queue temporarily unavailable") + return await super()._dead_letter(job, reason) + + queue = FailOnceDeadLetterQueue( + pool=pool, + registry=registry, + table=workflow_table, + lease_seconds=2, + max_attempts=1, + ) + await registry.create_schema() + await queue.create_schema() + try: + task_id = await registry.issue( + account_id="tenant-a:account-a", + task_type="create_media_buy", + ) + await queue.enqueue( + task_id=task_id, + account_id="tenant-b:account-b", + workflow_type="manual_media_buy_approval", + payload={"upstream_order_id": "order-1"}, + ) + handler_called = False + + async def handler(_job): + nonlocal handler_called + handler_called = True + return {"status": "active"} + + with pytest.raises(RuntimeError, match="queue temporarily unavailable"): + await queue.process_one(handler) + + async with pool.connection() as conn: + await conn.execute( + f"UPDATE {workflow_table} " # noqa: S608 + "SET lease_expires_at = now() - interval '1 second' " + "WHERE task_id = %s", + (task_id,), + ) + + assert await queue.process_one(handler) is True + queue_record = await queue.get(task_id) + task_record = await registry.get( + task_id, + expected_account_id="tenant-a:account-a", + ) + + assert handler_called is False + assert queue.dead_letter_calls == 2 + assert queue_record is not None + assert queue_record["state"] == "dead_lettered" + assert task_record is not None + assert task_record["state"] == "submitted" + finally: + async with pool.connection() as conn: + await conn.execute(f"DROP TABLE IF EXISTS {workflow_table}") # noqa: S608 + await conn.execute(f"DROP TABLE IF EXISTS {task_table}") # noqa: S608 + + +@pytest.mark.asyncio +async def test_registry_lookup_outage_never_mutates_unverified_task() -> None: + suffix = secrets.token_hex(6) + task_table = f"test_dtasks_{suffix}" + workflow_table = f"test_workflows_{suffix}" + account_id = "tenant-a:account-a" + + async with psycopg_pool.AsyncConnectionPool( + TEST_URL, + min_size=1, + max_size=4, + open=False, + ) as pool: + await pool.open() + registry = PgTaskRegistry(pool=pool, _table=task_table) + + class LookupOutageRegistry: + async def get(self, task_id, *, expected_account_id=None): + raise RuntimeError("registry lookup unavailable") + + async def complete(self, task_id, result): + raise AssertionError("unverified task must not be completed") + + async def fail(self, task_id, error): + raise AssertionError("unverified task must not be failed") + + queue = PgWorkflowQueue( + pool=pool, + registry=LookupOutageRegistry(), # type: ignore[arg-type] + table=workflow_table, + max_attempts=2, + retry_base_seconds=1, + ) + await registry.create_schema() + await queue.create_schema() + try: + task_id = await registry.issue( + account_id=account_id, + task_type="create_media_buy", + ) + await queue.enqueue( + task_id=task_id, + account_id=account_id, + workflow_type="manual_media_buy_approval", + payload={"upstream_order_id": "order-1"}, + ) + handler_called = False + + async def handler(_job): + nonlocal handler_called + handler_called = True + return {"status": "active"} + + assert await queue.process_one(handler) is True + async with pool.connection() as conn: + await conn.execute( + f"UPDATE {workflow_table} SET available_at = now() " # noqa: S608 + "WHERE task_id = %s", + (task_id,), + ) + assert await queue.process_one(handler) is True + + queue_record = await queue.get(task_id) + task_record = await registry.get( + task_id, + expected_account_id=account_id, + ) + assert handler_called is False + assert queue_record is not None + assert queue_record["state"] == "dead_lettered" + assert queue_record["attempt_count"] == 2 + assert queue_record["last_error"] == "RuntimeError" + assert task_record is not None + assert task_record["state"] == "submitted" + finally: + async with pool.connection() as conn: + await conn.execute(f"DROP TABLE IF EXISTS {workflow_table}") # noqa: S608 + await conn.execute(f"DROP TABLE IF EXISTS {task_table}") # noqa: S608 diff --git a/tests/test_server_idempotency.py b/tests/test_server_idempotency.py index b7791f712..9a20431d8 100644 --- a/tests/test_server_idempotency.py +++ b/tests/test_server_idempotency.py @@ -1320,6 +1320,33 @@ async def put(self, *args: Any, **kwargs: Any) -> None: assert result["media_buy_id"] == "mb_1" # handler ran, result returned assert any("cache put failed" in rec.message for rec in caplog.records) + @pytest.mark.asyncio + async def test_put_failure_can_fail_closed_for_production(self) -> None: + from adcp.decisioning import AdcpError + + class BrokenBackend(MemoryBackend): + async def put(self, *args: Any, **kwargs: Any) -> None: + raise RuntimeError("simulated backend outage") + + store = IdempotencyStore( + backend=BrokenBackend(), + ttl_seconds=86400, + raise_on_persist_error=True, + ) + wrapped = store.wrap(_FakeHandler.create_media_buy) + + with pytest.raises(AdcpError) as exc_info: + await wrapped( + _FakeHandler(), + {"idempotency_key": str(uuid.uuid4()), "b": 1}, + ToolContext(caller_identity="principal-a"), + ) + + assert exc_info.value.code == "SERVICE_UNAVAILABLE" + assert exc_info.value.recovery == "transient" + assert "outcome may be unknown" in str(exc_info.value) + assert "independently deduplicated" in str(exc_info.value) + class TestWireTranslation: """IdempotencyConflictError raised from a wrapped handler must surface on