diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5dfe13136..100deb664 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -198,6 +198,7 @@ jobs: tests/conformance/decisioning/test_pg_idempotency_backend.py \ tests/conformance/decisioning/test_pg_idempotency_reservation.py \ tests/test_decisioning_task_polling.py \ + tests/test_decisioning_task_lifecycle.py \ tests/conformance/decisioning/test_pg_task_webhook_outbox.py \ tests/conformance/decisioning/test_pg_reference_workflow_queue.py \ tests/conformance/reporting/ \ diff --git a/docs/task-lifecycle-observers.md b/docs/task-lifecycle-observers.md new file mode 100644 index 000000000..66ed5b9c9 --- /dev/null +++ b/docs/task-lifecycle-observers.md @@ -0,0 +1,37 @@ +# Task lifecycle metrics + +`InMemoryTaskRegistry` and `PgTaskRegistry` offer additive lifecycle hooks. The +minimal `TaskRegistry` protocol and all existing write-method return values stay +unchanged. Register a synchronous callback with `add_lifecycle_observer` and +remove it with `remove_lifecycle_observer` (returns whether removal succeeded). +Registering the same callback twice is a no-op. + +```python +from adcp.decisioning import InMemoryTaskRegistry, TaskTransition + +registry = InMemoryTaskRegistry() + +def record_transition(event: TaskTransition, *, task_id: str, account_id: str, + task_type: str, created_at: float, updated_at: float) -> None: + print(event, task_type, updated_at - created_at) + +registry.add_lifecycle_observer(record_transition) +``` + +Events are `submitted`, the first `working` transition, `completed`, `failed`, +and `discarded`. Repeated progress writes, idempotent terminal retries, unknown +progress/discard requests and failed transactions do not emit events. Terminal +states cannot be changed into another terminal state; memory and PostgreSQL +registries enforce the same rule, including concurrent complete/fail calls. + +Callbacks run after the PostgreSQL transaction and connection contexts exit, or +after the memory registry lock is released. Each event contains a snapshot of +its transition metadata with epoch timestamps; discard uses deletion time for +`updated_at`. Registration is thread-safe, and delivery uses a snapshot of the +registered callbacks. Callback failures are logged and swallowed without +preventing remaining callbacks or changing the write result. + +These hooks are best-effort metrics, not durable events: process exit can lose a +notification, and concurrent writers may notify out of order after commit. Keep +callbacks short and synchronous. Use the durable task webhook outbox for +buyer-facing delivery. No database migration is needed. diff --git a/src/adcp/decisioning/__init__.py b/src/adcp/decisioning/__init__.py index f44143852..2be7e736c 100644 --- a/src/adcp/decisioning/__init__.py +++ b/src/adcp/decisioning/__init__.py @@ -225,8 +225,10 @@ def create_media_buy( InMemoryTaskRegistry, ListableTaskRegistry, TaskHandoffContext, + TaskLifecycleObserver, TaskRegistry, TaskState, + TaskTransition, TaskWebhookAuthentication, ) from adcp.decisioning.tenant_store import create_tenant_store @@ -475,6 +477,8 @@ def __init__(self, *args: object, **kwargs: object) -> None: "TaskHandoff", "TaskHandoffContext", "TaskRegistry", + "TaskLifecycleObserver", + "TaskTransition", "TaskState", "TaskWebhookAuthentication", "WebhookSenderResolver", diff --git a/src/adcp/decisioning/pg/task_registry.py b/src/adcp/decisioning/pg/task_registry.py index 4f0ab50a0..d0a6621ba 100644 --- a/src/adcp/decisioning/pg/task_registry.py +++ b/src/adcp/decisioning/pg/task_registry.py @@ -71,7 +71,7 @@ async def main(): from typing import TYPE_CHECKING, Any, ClassVar, TypeAlias from adcp.decisioning.account_projection import strip_credentials_from_wire_result -from adcp.decisioning.task_registry import TaskWebhookAuthentication +from adcp.decisioning.task_registry import TaskWebhookAuthentication, _TaskLifecycleObservers if TYPE_CHECKING: from psycopg_pool import AsyncConnectionPool @@ -102,7 +102,7 @@ async def main(): WebhookSigningScopeResolver: TypeAlias = Callable[["RequestContext[Any]"], str | Awaitable[str]] -class PgTaskRegistry: +class PgTaskRegistry(_TaskLifecycleObservers): """PostgreSQL-backed :class:`~adcp.decisioning.TaskRegistry` — v6.1. Durable counterpart to :class:`~adcp.decisioning.InMemoryTaskRegistry`. @@ -156,6 +156,7 @@ def __init__( "webhook_signing_scope_resolver is required exactly when the " "PgTaskWebhookOutbox uses sender_resolver" ) + self._init_lifecycle_observers() self._pool = pool self._table = _table self.task_webhook_outbox = task_webhook_outbox @@ -171,24 +172,28 @@ def __init__( f" VALUES (%s, %s, 'submitted', %s, %s::jsonb, %s, %s, %s, %s)" ) self._sql_update_progress = ( # noqa: S608 - f"UPDATE {self._table}" # nosec B608 — table identifier is constructor-validated - f" SET state = CASE state WHEN 'submitted' THEN 'working' ELSE state END," + f"WITH previous AS (SELECT task_id, state FROM {self._table}" # nosec B608 — table identifier is constructor-validated + f" WHERE task_id = %s AND state NOT IN ('completed', 'failed') FOR UPDATE)" + f" UPDATE {self._table} AS task" + f" SET state = CASE task.state WHEN 'submitted' THEN 'working' ELSE task.state END," f" progress = %s::jsonb, updated_at = %s" - f" WHERE task_id = %s AND state NOT IN ('completed', 'failed')" + f" FROM previous WHERE task.task_id = previous.task_id" + f" RETURNING previous.state, task.task_id, task.account_id, task.task_type," + f" task.created_at, task.updated_at" ) self._sql_complete = ( # noqa: S608 f"UPDATE {self._table}" # nosec B608 — table identifier is constructor-validated f" SET state = 'completed', result = %s::jsonb, updated_at = %s" f" WHERE task_id = %s AND state NOT IN ('completed', 'failed')" f" RETURNING task_id, account_id, task_type, webhook_registration," - f" webhook_registration_nonce" + f" webhook_registration_nonce, created_at, updated_at" ) self._sql_fail = ( # noqa: S608 f"UPDATE {self._table}" # nosec B608 — table identifier is constructor-validated f" SET state = 'failed', error = %s::jsonb, updated_at = %s" f" WHERE task_id = %s AND state NOT IN ('completed', 'failed')" f" RETURNING task_id, account_id, task_type, webhook_registration," - f" webhook_registration_nonce" + f" webhook_registration_nonce, created_at, updated_at" ) self._sql_clear_webhook_registration = ( # noqa: S608 f"UPDATE {self._table} SET webhook_registration = NULL," # nosec B608 — table identifier is constructor-validated @@ -206,13 +211,16 @@ def __init__( f" FROM {self._table}" f" WHERE task_id = %s AND (%s::text IS NULL OR account_id = %s)" ) - self._sql_get_state_result = ( + self._sql_get_state_result = ( # noqa: S608 f"SELECT state, result, task_type FROM {self._table}" # nosec B608 — validated identifier " WHERE task_id = %s" ) self._sql_get_task_type = f"SELECT task_type FROM {self._table} WHERE task_id = %s" # noqa: S608 # nosec B608 — table identifier is constructor-validated self._sql_get_state_error = f"SELECT state, error FROM {self._table} WHERE task_id = %s" # noqa: S608 # nosec B608 — table identifier is constructor-validated - self._sql_discard = f"DELETE FROM {self._table} WHERE task_id = %s" # noqa: S608 # nosec B608 — table identifier is constructor-validated + self._sql_discard = ( # noqa: S608 + f"DELETE FROM {self._table} WHERE task_id = %s" # nosec B608 — table identifier is constructor-validated + f" RETURNING task_id, account_id, task_type, created_at, updated_at" + ) self._sql_ddl = ( # noqa: S608 f"CREATE TABLE IF NOT EXISTS {self._table} (" f' task_id TEXT COLLATE "C" NOT NULL PRIMARY KEY,' @@ -328,6 +336,16 @@ async def issue( now, ), ) + self._notify_lifecycle_observers( + "submitted", + { + "task_id": task_id, + "account_id": account_id, + "task_type": task_type, + "created_at": now, + "updated_at": now, + }, + ) return task_id async def resolve_webhook_signing_scope( @@ -387,21 +405,29 @@ async def update_progress( straggler progress event. """ async with self._pool.connection() as conn: - await conn.execute( + cur = await conn.execute( self._sql_update_progress, - (json.dumps(progress), time.time(), task_id), + (task_id, json.dumps(progress), time.time()), ) - # Zero rows updated means unknown task_id or terminal state — silent - # no-op per Protocol contract. The InMemoryTaskRegistry logs a - # WARNING on terminal-state drops; we omit the extra SELECT needed - # to distinguish the two cases since the dispatch wrapper swallows - # the result either way. + row = await cur.fetchone() + if row is not None and row[0] == "submitted": + self._notify_lifecycle_observers("working", self._lifecycle_record(row[1:])) - async def complete( + @staticmethod + def _lifecycle_record(row: tuple[Any, ...]) -> dict[str, Any]: + return dict(zip(("task_id", "account_id", "task_type", "created_at", "updated_at"), row)) + + async def complete(self, task_id: str, result: dict[str, Any]) -> None: + """Complete once; notify metrics only after the transaction commits.""" + row = await self._complete(task_id, result) + if row is not None: + self._notify_lifecycle_observers("completed", self._lifecycle_record(row[:3] + row[5:])) + + async def _complete( self, task_id: str, result: dict[str, Any], - ) -> None: + ) -> tuple[Any, ...] | None: """Mark the task ``completed`` with ``result`` as the terminal artifact. Idempotent on repeated calls with an equal ``result``; raises @@ -427,7 +453,7 @@ async def complete( self._sql_complete, (json.dumps(safe_result), time.time(), task_id), ) - row = await cur.fetchone() + row: tuple[Any, ...] | None = await cur.fetchone() if row is not None: await self._enqueue_terminal_if_registered( conn, @@ -435,7 +461,7 @@ async def complete( status="completed", payload=safe_result, ) - return # updated successfully + return row # connection and transaction exit before the caller notifies # Zero rows in RETURNING — task is unknown or already terminal. cur2 = await conn.execute(self._sql_get_state_result, (task_id,)) @@ -446,15 +472,21 @@ async def complete( safe_result = strip_credentials_from_wire_result(task_type, result) if state == "completed": if existing_result == safe_result: - return # idempotent + return None # idempotent raise ValueError(f"Task {task_id!r} already completed with a different result") raise ValueError(f"Task {task_id!r} already in terminal state {state!r}") - async def fail( + async def fail(self, task_id: str, error: dict[str, Any]) -> None: + """Fail once; notify metrics only after the transaction commits.""" + row = await self._fail(task_id, error) + if row is not None: + self._notify_lifecycle_observers("failed", self._lifecycle_record(row[:3] + row[5:])) + + async def _fail( self, task_id: str, error: dict[str, Any], - ) -> None: + ) -> tuple[Any, ...] | None: """Mark the task ``failed`` with ``error`` as the terminal payload. Idempotent on repeated calls with an equal ``error``; raises @@ -466,7 +498,7 @@ async def fail( self._sql_fail, (json.dumps(error), time.time(), task_id), ) - row = await cur.fetchone() + row: tuple[Any, ...] | None = await cur.fetchone() if row is not None: await self._enqueue_terminal_if_registered( conn, @@ -474,7 +506,7 @@ async def fail( status="failed", payload=error, ) - return # updated successfully + return row # notify only after connection/transaction exit # Zero rows in RETURNING — task is unknown or already terminal. cur2 = await conn.execute(self._sql_get_state_error, (task_id,)) @@ -484,7 +516,7 @@ async def fail( state, existing_error = row if state == "failed": if existing_error == error: - return # idempotent + return None # idempotent raise ValueError(f"Task {task_id!r} already failed with a different error") raise ValueError(f"Task {task_id!r} already in terminal state {state!r}") @@ -574,7 +606,7 @@ async def _enqueue_terminal_if_registered( payload: dict[str, Any], ) -> None: """Enqueue on ``conn`` so task state and webhook commit atomically.""" - task_id, account_id, task_type, encrypted_registration, registration_nonce = row + task_id, account_id, task_type, encrypted_registration, registration_nonce = row[:5] if encrypted_registration is None: return if self.task_webhook_outbox is None: @@ -617,7 +649,12 @@ async def discard(self, task_id: str) -> None: matching the :class:`~adcp.decisioning.InMemoryTaskRegistry` contract. """ async with self._pool.connection() as conn: - await conn.execute(self._sql_discard, (task_id,)) + cur = await conn.execute(self._sql_discard, (task_id,)) + row = await cur.fetchone() + if row is not None: + event_record = self._lifecycle_record(row) + event_record["updated_at"] = time.time() + self._notify_lifecycle_observers("discarded", event_record) #: Backwards-compat alias. Renamed to :class:`PgTaskRegistry` in diff --git a/src/adcp/decisioning/task_registry.py b/src/adcp/decisioning/task_registry.py index 62e6c0540..d3f0a116c 100644 --- a/src/adcp/decisioning/task_registry.py +++ b/src/adcp/decisioning/task_registry.py @@ -37,6 +37,7 @@ import asyncio import logging +import threading import time import uuid from collections.abc import Awaitable, Callable @@ -51,6 +52,71 @@ TaskState = Literal["submitted", "working", "completed", "failed"] +TaskTransition = Literal["submitted", "working", "completed", "failed", "discarded"] + + +class TaskLifecycleObserver(Protocol): + """Synchronous best-effort metrics hook; no durable delivery guarantee.""" + + def __call__( + self, + event: TaskTransition, + *, + task_id: str, + account_id: str, + task_type: str, + created_at: float, + updated_at: float, + ) -> None: ... + + +class _TaskLifecycleObservers: + """Thread-safe observer registration, shared by SDK registry implementations.""" + + def _init_lifecycle_observers(self) -> None: + self._lifecycle_observers: list[TaskLifecycleObserver] = [] + self._lifecycle_observers_lock = threading.Lock() + + def add_lifecycle_observer(self, observer: TaskLifecycleObserver) -> None: + """Register a synchronous hook fired after commit/outside registry locks. + + Re-registering the same callback is a no-op. Notifications use a snapshot; + registration changes during delivery apply to the next notification. + Observer exceptions are logged and swallowed. Hooks are best-effort metrics, + not a durable event stream, and concurrent transitions may notify out of order. + """ + with self._lifecycle_observers_lock: + if observer not in self._lifecycle_observers: + self._lifecycle_observers.append(observer) + + def remove_lifecycle_observer(self, observer: TaskLifecycleObserver) -> bool: + """Remove a hook, returning whether it was registered.""" + with self._lifecycle_observers_lock: + try: + self._lifecycle_observers.remove(observer) + except ValueError: + return False + return True + + def _notify_lifecycle_observers(self, event: TaskTransition, record: dict[str, Any]) -> None: + with self._lifecycle_observers_lock: + observers = tuple(self._lifecycle_observers) + for observer in observers: + try: + observer( + event, + task_id=record["task_id"], + account_id=record["account_id"], + task_type=record["task_type"], + created_at=record["created_at"], + updated_at=record["updated_at"], + ) + except Exception: + logger.exception( + "Task lifecycle observer failed for %s task %s", event, record["task_id"] + ) + + @dataclass(frozen=True) class TaskWebhookAuthentication: """Buyer-selected authentication for a durable task webhook. @@ -408,7 +474,7 @@ async def list( # --------------------------------------------------------------------------- -class InMemoryTaskRegistry: +class InMemoryTaskRegistry(_TaskLifecycleObservers): """Process-local task registry — v6.0 reference implementation. Storage is a plain ``dict[str, TaskRecord]`` guarded by an @@ -431,6 +497,7 @@ class InMemoryTaskRegistry: is_durable: ClassVar[bool] = False def __init__(self) -> None: + self._init_lifecycle_observers() self._records: dict[str, TaskRecord] = {} self._lock = asyncio.Lock() @@ -474,6 +541,8 @@ async def issue( task_type=task_type, request_context=(dict(request_context) if request_context is not None else None), ) + event_record = self._records[task_id].to_dict() + self._notify_lifecycle_observers("submitted", event_record) return task_id async def update_progress( @@ -503,10 +572,14 @@ async def update_progress( record.state, ) return + transitioned = record.state == "submitted" record.progress = dict(progress) if record.state == "submitted": record.state = "working" record.updated_at = time.time() + event_record = record.to_dict() + if transitioned: + self._notify_lifecycle_observers("working", event_record) async def complete( self, @@ -535,9 +608,13 @@ async def complete( if record.result == stripped: return # idempotent raise ValueError(f"Task {task_id!r} already completed with a different result") + if record.state == "failed": + raise ValueError(f"Task {task_id!r} already in terminal state {record.state!r}") record.state = "completed" record.result = dict(stripped) if isinstance(stripped, dict) else stripped record.updated_at = time.time() + event_record = record.to_dict() + self._notify_lifecycle_observers("completed", event_record) async def fail( self, @@ -552,9 +629,13 @@ async def fail( if record.error == error: return # idempotent raise ValueError(f"Task {task_id!r} already failed with a different error") + if record.state == "completed": + raise ValueError(f"Task {task_id!r} already in terminal state {record.state!r}") record.state = "failed" record.error = dict(error) record.updated_at = time.time() + event_record = record.to_dict() + self._notify_lifecycle_observers("failed", event_record) async def get( self, @@ -602,7 +683,12 @@ async def discard(self, task_id: str) -> None: # Idempotent: pop with default. The Protocol contract # tolerates discarding an unknown id (no raise) so the # WorkflowHandoff projection's rollback can be unconditional. - self._records.pop(task_id, None) + record = self._records.pop(task_id, None) + if record is not None: + event_record = record.to_dict() + event_record["updated_at"] = time.time() + if record is not None: + self._notify_lifecycle_observers("discarded", event_record) # --------------------------------------------------------------------------- diff --git a/tests/test_decisioning_task_lifecycle.py b/tests/test_decisioning_task_lifecycle.py new file mode 100644 index 000000000..26944ffdd --- /dev/null +++ b/tests/test_decisioning_task_lifecycle.py @@ -0,0 +1,166 @@ +"""Best-effort metrics hooks count committed transitions, including races.""" + +from __future__ import annotations + +import asyncio +import os +import secrets +from collections.abc import AsyncIterator +from typing import Any + +import pytest + +from adcp.decisioning import InMemoryTaskRegistry, TaskTransition + + +@pytest.fixture(params=["memory", "postgres", "postgres-autocommit"]) +async def registry(request: pytest.FixtureRequest) -> AsyncIterator[Any]: + if request.param == "memory": + yield InMemoryTaskRegistry() + return + url = os.environ.get("ADCP_PG_TEST_URL") + if not url: + pytest.skip("ADCP_PG_TEST_URL required") + from psycopg_pool import AsyncConnectionPool + + from adcp.decisioning.pg import PgTaskRegistry + + table = f"test_lifecycle_{secrets.token_hex(6)}" + async with AsyncConnectionPool( + url, + open=False, + min_size=1, + max_size=4, + kwargs={"autocommit": request.param == "postgres-autocommit"}, + ) as pool: + reg = PgTaskRegistry(pool=pool, _table=table) + await reg.create_schema() + yield reg + async with pool.connection() as conn: + await conn.execute(f"DROP TABLE {table}") + + +async def test_observe_committed_transitions_and_suppress_noops(registry: Any) -> None: + events: list[tuple[str, dict[str, Any]]] = [] + + def observe(event: TaskTransition, **metadata: Any) -> None: + # A synchronous callback can inspect committed state on another connection. + # It cannot observe an uncommitted row or run under the memory registry lock. + if isinstance(registry, InMemoryTaskRegistry): + assert not registry._lock.locked() + record = registry._records.get(metadata["task_id"]) + assert record is None if event == "discarded" else record.state == event + else: + import psycopg + + with psycopg.connect(os.environ["ADCP_PG_TEST_URL"]) as conn: + row = conn.execute( + f"SELECT state FROM {registry._table} WHERE task_id = %s", + (metadata["task_id"],), + ).fetchone() + assert row is None if event == "discarded" else row[0] == event + events.append((event, metadata)) + + registry.add_lifecycle_observer(observe) + registry.add_lifecycle_observer(observe) + task = await registry.issue(account_id="acct_1", task_type="get_products") + await asyncio.gather(*(registry.update_progress(task, {"percentage": n}) for n in range(10))) + await asyncio.gather(*(registry.complete(task, {"products": []}) for _ in range(10))) + await registry.update_progress(task, {"percentage": 100}) + await registry.update_progress("absent", {}) + await registry.discard("absent") + await registry.discard(task) + await registry.discard(task) + assert [event for event, _ in events] == ["submitted", "working", "completed", "discarded"] + for _, metadata in events: + assert metadata["task_id"] == task + assert metadata["account_id"] == "acct_1" + assert metadata["task_type"] == "get_products" + assert metadata["created_at"] <= metadata["updated_at"] + assert registry.remove_lifecycle_observer(observe) is True + assert registry.remove_lifecycle_observer(observe) is False + await registry.issue(account_id="acct_1", task_type="get_products") + assert len(events) == 4 + + +async def test_racing_terminal_transitions_emit_once(registry: Any) -> None: + events = [] + + def observe(event: TaskTransition, **metadata: Any) -> None: + events.append(event) + + registry.add_lifecycle_observer(observe) + task = await registry.issue(account_id="acct", task_type="get_products") + results = await asyncio.gather( + *(registry.complete(task, {"products": []}) for _ in range(10)), + *(registry.fail(task, {"code": "INTERNAL_ERROR", "message": "Failed"}) for _ in range(10)), + return_exceptions=True, + ) + assert events[0] == "submitted" + assert len(events) == 2 and events[1] in {"completed", "failed"} + assert any(isinstance(result, ValueError) for result in results) + record = await registry.get(task) + assert record["state"] == events[1] + if record["state"] == "completed": + with pytest.raises(ValueError): + await registry.fail(task, {"code": "CONFLICT", "message": "Conflict"}) + else: + with pytest.raises(ValueError): + await registry.complete(task, {"products": []}) + assert len(events) == 2 + + +async def test_observer_failure_and_snapshot_removal( + registry: Any, caplog: pytest.LogCaptureFixture +) -> None: + events = [] + + def broken(event: TaskTransition, **metadata: Any) -> None: + raise RuntimeError("Metrics unavailable") + + def removed(event: TaskTransition, **metadata: Any) -> None: + events.append(event) + + def remover(event: TaskTransition, **metadata: Any) -> None: + registry.remove_lifecycle_observer(removed) + + registry.add_lifecycle_observer(broken) + registry.add_lifecycle_observer(remover) + registry.add_lifecycle_observer(removed) + task = await registry.issue(account_id="acct", task_type="get_products") + await registry.fail(task, {"code": "INTERNAL_ERROR", "message": "Failed"}) + await registry.fail(task, {"code": "INTERNAL_ERROR", "message": "Failed"}) + assert events == ["submitted"] + assert "Task lifecycle observer failed" in caplog.text + assert (await registry.get(task))["state"] == "failed" + + +async def test_failed_database_transaction_never_notifies(registry: Any) -> None: + if isinstance(registry, InMemoryTaskRegistry): + with pytest.raises(ValueError): + await registry.complete("missing", {}) + return + import psycopg + + events = [] + registry.add_lifecycle_observer(lambda event, **metadata: events.append(event)) + task = await registry.issue(account_id="acct", task_type="get_products") + function = f"{registry._table}_reject" + async with registry._pool.connection() as conn: + await conn.execute( + f"CREATE FUNCTION {function}() RETURNS trigger LANGUAGE plpgsql AS $$ " + "BEGIN RAISE EXCEPTION 'fail commit'; END $$" + ) + await conn.execute( + f"CREATE CONSTRAINT TRIGGER reject_transition AFTER UPDATE ON {registry._table}" + f" DEFERRABLE INITIALLY DEFERRED FOR EACH ROW EXECUTE FUNCTION {function}()" + ) + try: + with pytest.raises(psycopg.errors.RaiseException): + await registry.complete(task, {"products": []}) + assert events == ["submitted"] + assert (await registry.get(task))["state"] == "submitted" + finally: + async with registry._pool.connection() as conn: + await conn.execute(f"DROP TRIGGER reject_transition ON {registry._table}") + await conn.execute(f"DROP FUNCTION {function}()") diff --git a/tests/type_checks/task_lifecycle_observers.py b/tests/type_checks/task_lifecycle_observers.py new file mode 100644 index 000000000..f628d38c9 --- /dev/null +++ b/tests/type_checks/task_lifecycle_observers.py @@ -0,0 +1,39 @@ +"""Public metrics hook works without changing the minimal registry protocol.""" + +from typing import Any + +from adcp.decisioning import ( + InMemoryTaskRegistry, + TaskLifecycleObserver, + TaskRegistry, + TaskTransition, +) +from adcp.decisioning.pg import PgTaskRegistry + + +class Metrics: + def __call__( + self, + event: TaskTransition, + *, + task_id: str, + account_id: str, + task_type: str, + created_at: float, + updated_at: float, + ) -> None: + print(event, task_id, account_id, task_type, updated_at - created_at) + + +observer: TaskLifecycleObserver = Metrics() + + +async def instrument(memory: InMemoryTaskRegistry, postgres: PgTaskRegistry) -> None: + for registry in (memory, postgres): + registry.add_lifecycle_observer(observer) + minimal: TaskRegistry = registry + task = await minimal.issue(account_id="account", task_type="get_products") + result: dict[str, Any] = {"products": []} + await minimal.complete(task, result) + removed: bool = registry.remove_lifecycle_observer(observer) + assert removed