From df1e2355e4fea05df71ab326bfdb466dc805fe41 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 07:44:49 +0000 Subject: [PATCH 1/7] test: reproduce unbounded metric route emissions --- tests/fixtures/sdk_metrics/contract.json | 63 ++++++++++ tests/fixtures/sdk_metrics/provenance.json | 4 + tests/test_metric_routes.py | 135 +++++++++++++++++++++ 3 files changed, 202 insertions(+) create mode 100644 tests/fixtures/sdk_metrics/contract.json create mode 100644 tests/fixtures/sdk_metrics/provenance.json create mode 100644 tests/test_metric_routes.py diff --git a/tests/fixtures/sdk_metrics/contract.json b/tests/fixtures/sdk_metrics/contract.json new file mode 100644 index 0000000..0726392 --- /dev/null +++ b/tests/fixtures/sdk_metrics/contract.json @@ -0,0 +1,63 @@ +{ + "schema": "durable-workflow.sdk-metrics/v1", + "revision": 1, + "reference": { + "repository": "durable-workflow/sdk-python", + "commit": "2da02a7ac8a68a7cd4267e86ef10e0198a309592", + "paths": [ + "src/durable_workflow/metrics.py", + "src/durable_workflow/client.py", + "src/durable_workflow/worker.py" + ] + }, + "metrics": [ + { + "name": "durable_workflow_client_requests", + "type": "counter", + "unit": "1", + "recorder_method": "increment", + "scope": "client_request", + "tags": ["method", "route", "plane", "status_code", "outcome"] + }, + { + "name": "durable_workflow_client_request_duration_seconds", + "type": "histogram", + "unit": "s", + "recorder_method": "record", + "scope": "client_request", + "tags": ["method", "route", "plane", "status_code", "outcome"] + }, + { + "name": "durable_workflow_worker_polls", + "type": "counter", + "unit": "1", + "recorder_method": "increment", + "scope": "worker_poll", + "tags": ["task_kind", "task_queue", "outcome"] + }, + { + "name": "durable_workflow_worker_poll_duration_seconds", + "type": "histogram", + "unit": "s", + "recorder_method": "record", + "scope": "worker_poll", + "tags": ["task_kind", "task_queue", "outcome"] + }, + { + "name": "durable_workflow_worker_tasks", + "type": "counter", + "unit": "1", + "recorder_method": "increment", + "scope": "worker_task", + "tags": ["task_kind", "task_queue", "outcome"] + }, + { + "name": "durable_workflow_worker_task_duration_seconds", + "type": "histogram", + "unit": "s", + "recorder_method": "record", + "scope": "worker_task", + "tags": ["task_kind", "task_queue", "outcome"] + } + ] +} diff --git a/tests/fixtures/sdk_metrics/provenance.json b/tests/fixtures/sdk_metrics/provenance.json new file mode 100644 index 0000000..d24cd9d --- /dev/null +++ b/tests/fixtures/sdk_metrics/provenance.json @@ -0,0 +1,4 @@ +{ + "source": "https://github.com/durable-workflow/.github/blob/f3627e2f52db6f7e4fbc2775e8c53ec8a758d96e/sdk-metrics/contract.json", + "sha256": "c456f4f434c601fa26f0caba07ef0fd29f548ca8f5fc29ba2f6430f01e32c0f0" +} diff --git a/tests/test_metric_routes.py b/tests/test_metric_routes.py new file mode 100644 index 0000000..dabf6ed --- /dev/null +++ b/tests/test_metric_routes.py @@ -0,0 +1,135 @@ +from __future__ import annotations + +from collections.abc import AsyncIterator +from typing import Any + +import httpx +import pytest + +from durable_workflow.client import Client +from durable_workflow.errors import ServerError +from durable_workflow.metrics import CLIENT_REQUEST_DURATION_SECONDS, CLIENT_REQUESTS, InMemoryMetrics +from durable_workflow.retry_policy import TransportRetryPolicy + +ROUTES = [ + ("/workflows/wf-{id}/signal/signal-{id}", "/workflows/{workflow_id}/signal/{name}"), + ("/workflows/wf-{id}/runs/run-{id}/history/export", "/workflows/{workflow_id}/runs/{run_id}/history/export"), + ("/worker/workflow-tasks/task-{id}/complete", "/worker/workflow-tasks/{task_id}/complete"), + ("/worker/activity-tasks/task-{id}/heartbeat", "/worker/activity-tasks/{task_id}/heartbeat"), + ("/worker/query-tasks/task-{id}/status", "/worker/query-tasks/{task_id}/status"), + ("/worker/update-validation-tasks/task-{id}/approve", "/worker/update-validation-tasks/{task_id}/approve"), + ("/worker/update-validation-tasks/task-{id}/reject", "/worker/update-validation-tasks/{task_id}/reject"), + ("/worker/registrations/worker-{id}", "/worker/registrations/{worker_id}"), + ("/worker/sessions/session-{id}/heartbeat", "/worker/sessions/{session_id}/heartbeat"), + ("/worker/sessions/session-{id}", "/worker/sessions/{session_id}"), + ("/activities/activity-{id}", "/activities/{activity_id}"), + ("/workers/worker-{id}", "/workers/{worker_id}"), + ("/schedules/schedule-{id}/history", "/schedules/{schedule_id}/history"), + ("/namespaces/namespace-{id}/external-storage", "/namespaces/{name}/external-storage"), + ("/task-queues/queue-{id}/build-ids/drain", "/task-queues/{task_queue}/build-ids/drain"), + ("/workflows/wf-{id}/message-streams/stream-{id}/messages", "/workflows/{workflow_id}/message-streams/{name}/messages"), + ("/workflows/wf-{id}/runs/run-{id}/streams/stream-{id}/items", "/workflows/{workflow_id}/runs/{run_id}/streams/{name}/items"), + ("/service-endpoints/endpoint-{id}/services/service-{id}/operations/operation-{id}/execute", + "/service-endpoints/{endpoint_name}/services/{service_name}/operations/{operation_name}/execute"), + ("/worker/workflow-tasks/task-{id}/local-activities/attempt-{id}/heartbeat", + "/worker/workflow-tasks/{task_id}/local-activities/{activity_attempt_id}/heartbeat"), + ("/unrecognized/private-{id}", "unknown"), + ("/worker/sessions/session-{id}/unrecognized/private-{id}", "unknown"), + ("/workflows/wf-{id}/unrecognized/private-{id}", "unknown"), +] + + +@pytest.fixture +async def instrumented_client() -> AsyncIterator[tuple[Client, InMemoryMetrics]]: + metrics = InMemoryMetrics() + client = Client( + "https://runtime.example", token="synthetic-credential", metrics=metrics, + retry_policy=TransportRetryPolicy(max_attempts=1), + ) + client._http = httpx.AsyncClient( + base_url=client.base_url, transport=httpx.MockTransport(lambda _: httpx.Response(200, json={})), + ) + client._runtime_external_payload_transport_resolved = True + async with client: + yield client, metrics + + +@pytest.mark.parametrize("path,expected", ROUTES) +async def test_request_emissions_have_bounded_routes( + path: str, expected: str, instrumented_client: tuple[Client, InMemoryMetrics], +) -> None: + client, metrics = instrumented_client + for identifier in ("synthetic-a", "synthetic-b"): + actual = path.replace("{id}", identifier) + "?token=synthetic-query-secret&cursor=" + identifier + await client._request("POST", actual, json={}) + tags = {"method": "POST", "route": expected, "plane": "control", "status_code": "200", "outcome": "ok"} + assert metrics.counter_value(CLIENT_REQUESTS, tags) == 2 + assert len(metrics.observations(CLIENT_REQUEST_DURATION_SECONDS, tags)) == 2 + assert len(metrics.counters) == len(metrics.histograms) == 1 + assert "synthetic-" not in repr(metrics.counters) + + +@pytest.mark.parametrize("kind", ["workflow", "activity", "query", "update-validation"]) +async def test_static_poll_route_is_distinct_from_a_task( + kind: str, instrumented_client: tuple[Client, InMemoryMetrics], +) -> None: + client, metrics = instrumented_client + await client._request("POST", f"/worker/{kind}-tasks/poll", worker=True, json={}) + tags = {"method": "POST", "route": f"/worker/{kind}-tasks/poll", "plane": "worker", + "status_code": "200", "outcome": "ok"} + assert metrics.counter_value(CLIENT_REQUESTS, tags) == 1 + + +@pytest.mark.parametrize("failure", [False, True], ids=["success", "http-failure"]) +async def test_public_worker_apis_do_not_emit_task_or_session_ids( + failure: bool, instrumented_client: tuple[Client, InMemoryMetrics], +) -> None: + client, metrics = instrumented_client + if failure: + await client._http.aclose() + client._http = httpx.AsyncClient(base_url=client.base_url, transport=httpx.MockTransport( + lambda _: httpx.Response(503, json={"reason": "unavailable"}), + )) + for identifier in ("synthetic-a", "synthetic-b"): + calls = [ + client.approve_update_validation_task( + update_validation_task_id=identifier, lease_owner="synthetic-worker", update_validation_attempt=1, + ), + client.reject_update_validation_task( + update_validation_task_id=identifier, lease_owner="synthetic-worker", update_validation_attempt=1, + message="synthetic-payload-secret", reason="invalid", + ), + client.renew_worker_session("synthetic-worker", identifier, 120), + client.close_worker_session("synthetic-worker", identifier), + ] + for call in calls: + if failure: + with pytest.raises(ServerError): + await call + else: + await call + routes = {dict(tags)["route"] for _, tags in metrics.counters} + assert routes == { + "/worker/update-validation-tasks/{task_id}/approve", "/worker/update-validation-tasks/{task_id}/reject", + "/worker/sessions/{session_id}/heartbeat", "/worker/sessions/{session_id}", + } + assert all(count == 2 for count in metrics.counters.values()) + assert "synthetic-" not in repr(metrics.counters) + + +async def test_transport_failure_uses_the_same_bounded_route( + instrumented_client: tuple[Client, InMemoryMetrics], +) -> None: + client, metrics = instrumented_client + + def unavailable(request: httpx.Request) -> Any: + raise httpx.ConnectError("synthetic-transport-secret", request=request) + + await client._http.aclose() + client._http = httpx.AsyncClient(base_url=client.base_url, transport=httpx.MockTransport(unavailable)) + for identifier in ("synthetic-a", "synthetic-b"): + polling = client.renew_worker_session("worker", identifier, 120) + with pytest.raises(httpx.ConnectError): + await polling + assert len(metrics.counters) == 1 + assert "synthetic-" not in repr(metrics.counters) From 2fea3eaecc450f21c4e0b86d638fffeba7d41370 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 07:49:46 +0000 Subject: [PATCH 2/7] fix: bound request metric routes and qualify shared recorder observations --- .github/workflows/ci.yml | 2 +- src/durable_workflow/_metric_routes.py | 57 ++++++ src/durable_workflow/client.py | 33 +--- tests/test_metric_routes.py | 6 +- tests/test_sdk_metrics_contract.py | 232 +++++++++++++++++++++++++ 5 files changed, 295 insertions(+), 35 deletions(-) create mode 100644 src/durable_workflow/_metric_routes.py create mode 100644 tests/test_sdk_metrics_contract.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4655fe2..630af63 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -152,7 +152,7 @@ jobs: uses: durable-workflow/.github/.github/actions/bootstrap-patched-pip@75587fc33ff40c34f32ef777293dad56bb6f48c7 # clean-trust with: clients: pip,python - - run: pip install -e '.[dev]' + - run: pip install -e '.[dev,prometheus]' - name: Run unit tests with stalled-thread diagnostics id: unit_tests shell: bash diff --git a/src/durable_workflow/_metric_routes.py b/src/durable_workflow/_metric_routes.py new file mode 100644 index 0000000..9eb35f7 --- /dev/null +++ b/src/durable_workflow/_metric_routes.py @@ -0,0 +1,57 @@ +"""Bound request labels to reviewed SDK operations, never arbitrary URL values.""" + +from __future__ import annotations + +_ROUTES = ( + "/", "/health", "/cluster/info", "/activities", "/activities/{activity_id}", + "/namespaces", "/namespaces/{name}", "/namespaces/{name}/external-storage", + "/search-attributes", "/search-attributes/{name}", "/storage/test", + "/task-queues", "/task-queues/{task_queue}", "/task-queues/{task_queue}/build-ids", + *(f"/task-queues/{{task_queue}}/build-ids/{action}" for action in ("drain", "promote", "resume")), + *(f"/system/{operation}{suffix}" for operation in ("activity-timeouts", "repair", "retention") + for suffix in ("", "/pass")), + "/workers", "/workers/{worker_id}", + "/worker/register", "/worker/heartbeat", "/worker/registrations/{worker_id}", + "/worker/sessions", "/worker/sessions/{session_id}", "/worker/sessions/{session_id}/heartbeat", + *(f"/worker/{kind}-tasks/poll" for kind in ("workflow", "activity", "query", "update-validation")), + *(f"/worker/{kind}-tasks/{{task_id}}/{action}" for kind, actions in ( + ("workflow", ("complete", "fail", "heartbeat", "history", "deliver-cancellation")), + ("activity", ("complete", "fail", "heartbeat", "status", "renew-lease", "acknowledge-cancellation")), + ("query", ("complete", "fail", "status")), + ("update-validation", ("approve", "reject")), + ) for action in actions), + *(f"/worker/workflow-tasks/{{task_id}}/cancellation-scopes/{action}" + for action in ("open", "prepare", "deliver")), + *(f"/worker/workflow-tasks/{{task_id}}/local-activities/{action}" + for action in ("checkpoint", "checkpoint-group", "prepare", "recover")), + *(f"/worker/workflow-tasks/{{task_id}}/local-activities/{{activity_attempt_id}}/{action}" + for action in ("control", "heartbeat", "outcome", "acknowledge-cancellation")), + "/workflows", "/workflows/{workflow_id}", + *(f"/workflows/{{workflow_id}}/{action}" + for action in ("archive", "cancel", "repair", "terminate", "request-cancellation", "runs")), + *(f"/workflows/{{workflow_id}}/{action}/{{name}}" for action in ("signal", "query", "update")), + "/workflows/{workflow_id}/message-streams/{name}/messages", + "/workflows/{workflow_id}/runs/{run_id}", + *(f"/workflows/{{workflow_id}}/runs/{{run_id}}/{action}" + for action in ("history", "history/export", "redrive", "request-cancellation", "streams")), + "/workflows/{workflow_id}/runs/{run_id}/streams/{name}", + *(f"/workflows/{{workflow_id}}/runs/{{run_id}}/streams/{{name}}/{action}" for action in ("items", "close")), + "/schedules", "/schedules/{schedule_id}", + *(f"/schedules/{{schedule_id}}/{action}" for action in ("backfill", "pause", "resume", "trigger", "history")), + "/bridge-adapters/webhook/{adapter}", + "/service-endpoints/{endpoint_name}/services/{service_name}/operations/{operation_name}/execute", +) +_TEMPLATES = tuple((route, tuple(route.strip("/").split("/"))) for route in _ROUTES) + + +def route_for_metrics(path: str) -> str: + """Return a supported operation template or the single bounded fallback.""" + clean_path = path.split("?", 1)[0].split("#", 1)[0] + parts = tuple(clean_path.strip("/").split("/")) + for route, template in _TEMPLATES: + if len(parts) == len(template) and all( + expected == actual or expected.startswith("{") and bool(actual) + for actual, expected in zip(parts, template, strict=True) + ): + return route + return "unknown" diff --git a/src/durable_workflow/client.py b/src/durable_workflow/client.py index 97287b3..d74c4c9 100644 --- a/src/durable_workflow/client.py +++ b/src/durable_workflow/client.py @@ -47,6 +47,7 @@ scope_timestamp, ) from ._cooperative_cancellation import CancellationDelivery, CancellationRequest +from ._metric_routes import route_for_metrics as _route_for_metrics from ._query_task_authority import QueryTaskAuthority, current_query_authority from .errors import ( REQUEST_ID_CONFLICT, @@ -289,38 +290,6 @@ def _command_request_failure(exc: Exception, *, command: str, request_id: str) - return exc -def _route_for_metrics(path: str) -> str: - clean_path = path.split("?", 1)[0] - parts = [part for part in clean_path.strip("/").split("/") if part] - if not parts: - return "/" - - if parts[0] == "workflows" and len(parts) >= 2: - parts[1] = "{workflow_id}" - if len(parts) >= 4 and parts[2] in {"signal", "query", "update"}: - parts[3] = "{name}" - if len(parts) >= 4 and parts[2] == "runs": - parts[3] = "{run_id}" - elif parts[0] == "schedules" and len(parts) >= 2: - parts[1] = "{schedule_id}" - elif parts[0] in {"namespaces", "search-attributes"} and len(parts) >= 2: - parts[1] = "{name}" - elif parts[0] == "workers" and len(parts) >= 2: - parts[1] = "{worker_id}" - elif parts[:2] == ["worker", "registrations"] and len(parts) >= 3: - parts[2] = "{worker_id}" - elif parts[:2] == ["bridge-adapters", "webhook"] and len(parts) >= 3: - parts[2] = "{adapter}" - elif ( - parts[:2] == ["worker", "workflow-tasks"] - or parts[:2] == ["worker", "activity-tasks"] - or parts[:2] == ["worker", "query-tasks"] - ) and len(parts) >= 3: - parts[2] = "{task_id}" - - return "/" + "/".join(parts) - - def _resolve_namespace_name( name: str | None, namespace_alias: str | None, diff --git a/tests/test_metric_routes.py b/tests/test_metric_routes.py index dabf6ed..ac5b294 100644 --- a/tests/test_metric_routes.py +++ b/tests/test_metric_routes.py @@ -27,8 +27,10 @@ ("/schedules/schedule-{id}/history", "/schedules/{schedule_id}/history"), ("/namespaces/namespace-{id}/external-storage", "/namespaces/{name}/external-storage"), ("/task-queues/queue-{id}/build-ids/drain", "/task-queues/{task_queue}/build-ids/drain"), - ("/workflows/wf-{id}/message-streams/stream-{id}/messages", "/workflows/{workflow_id}/message-streams/{name}/messages"), - ("/workflows/wf-{id}/runs/run-{id}/streams/stream-{id}/items", "/workflows/{workflow_id}/runs/{run_id}/streams/{name}/items"), + ("/workflows/wf-{id}/message-streams/stream-{id}/messages", + "/workflows/{workflow_id}/message-streams/{name}/messages"), + ("/workflows/wf-{id}/runs/run-{id}/streams/stream-{id}/items", + "/workflows/{workflow_id}/runs/{run_id}/streams/{name}/items"), ("/service-endpoints/endpoint-{id}/services/service-{id}/operations/operation-{id}/execute", "/service-endpoints/{endpoint_name}/services/{service_name}/operations/{operation_name}/execute"), ("/worker/workflow-tasks/task-{id}/local-activities/attempt-{id}/heartbeat", diff --git a/tests/test_sdk_metrics_contract.py b/tests/test_sdk_metrics_contract.py new file mode 100644 index 0000000..dbcc272 --- /dev/null +++ b/tests/test_sdk_metrics_contract.py @@ -0,0 +1,232 @@ +from __future__ import annotations + +import hashlib +import json +import math +import re +import time +from collections.abc import Coroutine, Mapping +from dataclasses import dataclass, field +from pathlib import Path +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock + +import httpx +import pytest + +import durable_workflow.client as client_module +import durable_workflow.worker as worker_module +from durable_workflow.client import Client +from durable_workflow.errors import ServerError +from durable_workflow.metrics import NOOP_METRICS, PrometheusMetrics +from durable_workflow.retry_policy import TransportRetryPolicy +from durable_workflow.worker import Worker + +CONTRACT_DIR = Path(__file__).parent / "fixtures/sdk_metrics" +CONTRACT = json.loads((CONTRACT_DIR / "contract.json").read_text()) +SPECS = {metric["name"]: metric for metric in CONTRACT["metrics"]} + + +@dataclass +class Recorder: + calls: list[tuple[str, str, float, dict[str, str]]] = field(default_factory=list) + + def increment(self, name: str, value: float = 1.0, tags: Mapping[str, str] | None = None) -> None: + self.calls.append(("increment", name, value, dict(tags or {}))) + + def record(self, name: str, value: float, tags: Mapping[str, str] | None = None) -> None: + self.calls.append(("record", name, value, dict(tags or {}))) + + +def assert_contract(recorder: Recorder, *, duration: float | None = None) -> set[str]: + assert recorder.calls and len(recorder.calls) % 2 == 0 + scopes: set[str] = set() + for counter, histogram in zip(recorder.calls[::2], recorder.calls[1::2], strict=True): + for method, name, value, tags in (counter, histogram): + spec = SPECS[name] + assert method == spec["recorder_method"] + assert set(tags) == set(spec["tags"]) + assert all(isinstance(value, str) for value in tags.values()) + assert math.isfinite(value) and value >= 0 + counter_spec, duration_spec = SPECS[counter[1]], SPECS[histogram[1]] + assert counter_spec["type"] == "counter" and counter_spec["unit"] == "1" + assert duration_spec["type"] == "histogram" and duration_spec["unit"] == "s" + assert counter_spec["scope"] == duration_spec["scope"] + assert counter[2] == 1 + assert counter[3] == histogram[3] + if duration is not None: + assert histogram[2] == duration + scopes.add(counter_spec["scope"]) + return scopes + + +def test_contract_is_an_exact_pinned_copy() -> None: + provenance = json.loads((CONTRACT_DIR / "provenance.json").read_text()) + assert hashlib.sha256((CONTRACT_DIR / "contract.json").read_bytes()).hexdigest() == provenance["sha256"] + assert provenance["source"].startswith("https://github.com/durable-workflow/.github/blob/") + assert re.fullmatch(r"[0-9a-f]{40}", provenance["source"].split("/blob/", 1)[1].split("/", 1)[0]) + assert CONTRACT["schema"] == "durable-workflow.sdk-metrics/v1" + + +@pytest.mark.parametrize("outcome", ["ok", "http_error", "ConnectError", "retry"]) +async def test_actual_client_request_pair_matches_the_contract( + outcome: str, monkeypatch: pytest.MonkeyPatch, +) -> None: + recorder = Recorder() + now = 1.0 + attempts = 0 + + def respond(request: httpx.Request) -> httpx.Response: + nonlocal now, attempts + attempts += 1 + now += 0.125 + if outcome == "ConnectError": + raise httpx.ConnectError("runtime unavailable", request=request) + return httpx.Response(503 if outcome == "http_error" or outcome == "retry" and attempts == 1 else 200, json={}) + + client = Client("https://runtime.example", token="synthetic-token", metrics=recorder, + retry_policy=TransportRetryPolicy(max_attempts=2, initial_backoff_seconds=0, jitter=False)) + client._http = httpx.AsyncClient(base_url=client.base_url, transport=httpx.MockTransport(respond)) + monkeypatch.setattr(client_module, "time", SimpleNamespace(perf_counter=lambda: now, monotonic=time.monotonic)) + async with client: + if outcome == "http_error": + with pytest.raises(ServerError): + await client.health() + elif outcome == "ConnectError": + with pytest.raises(httpx.ConnectError): + await client.health() + else: + await client.health() + assert_contract(recorder, duration=attempts * 0.125) + assert len(recorder.calls) == 2 # One logical request, including all HTTP retries. + assert attempts == (1 if outcome == "ok" else 2) + assert recorder.calls[0][3]["outcome"] == ("ok" if outcome == "retry" else outcome) + + +@pytest.mark.parametrize("kind", ["workflow", "activity", "query"]) +@pytest.mark.parametrize("outcome", ["empty", "task", "error", "backpressure"]) +async def test_actual_worker_poll_pair_matches_the_contract( + kind: str, outcome: str, monkeypatch: pytest.MonkeyPatch, +) -> None: + recorder = Recorder() + client = AsyncMock(spec=Client) + worker = Worker(client, task_queue="orders", metrics=recorder) + now = 1.0 + + async def poll(**_: object) -> dict[str, str] | None: + nonlocal now + now = 1.25 + if outcome == "error": + raise httpx.ReadError() + if outcome == "backpressure": + raise ServerError(429, { + "task": None, "reason": "long_poll_capacity_exhausted", + "poll_status": "long_poll_capacity_exhausted", "task_kind": f"{kind}_task", + "task_queue": "orders", "retryable": True, "retry_after_seconds": 2, + }) + if outcome == "task": + return {"task_id": "synthetic-task"} + worker._stop.set() + return None + + async def sleep(_: float) -> None: + worker._stop.set() + + def admit_workflow(*_: object) -> None: + worker._release_workflow_capacity() + worker._stop.set() + + def track(coro: Coroutine[Any, Any, Any]) -> None: + coro.close() + if kind == "activity": + worker._act_semaphore.release() + worker._stop.set() + + getattr(client, f"poll_{kind}_task").side_effect = poll + monkeypatch.setattr(worker_module, "time", SimpleNamespace(perf_counter=lambda: now, monotonic=time.monotonic)) + monkeypatch.setattr(worker_module.asyncio, "sleep", sleep) + monkeypatch.setattr(worker, "_admit_workflow_work", admit_workflow) + monkeypatch.setattr(worker, "_track", track) + await getattr(worker, f"_poll_{kind}_tasks")() + assert assert_contract(recorder, duration=0.25) == {"worker_poll"} + assert len(recorder.calls) == 2 + assert recorder.calls[0][3] == {"task_kind": kind, "task_queue": "orders", "outcome": outcome} + + +@pytest.mark.parametrize("kind", ["workflow", "activity", "query", "update_validation"]) +@pytest.mark.parametrize("failed", [False, True], ids=["success", "failure"]) +async def test_actual_task_dispatch_pair_matches_the_contract( + kind: str, failed: bool, monkeypatch: pytest.MonkeyPatch, +) -> None: + recorder = Recorder() + worker = Worker(AsyncMock(spec=Client), task_queue="orders", metrics=recorder) + now = 1.0 + expected = "failed" if failed else "approved" if kind == "update_validation" else "completed" + + async def execute(*_: object, **__: object) -> object: + nonlocal now + now = 1.25 + if kind == "workflow": + return None if failed else [] + return expected + + monkeypatch.setattr(worker_module, "time", SimpleNamespace(perf_counter=lambda: now, monotonic=time.monotonic)) + monkeypatch.setattr(worker, f"_run_{kind}_task", execute) + if kind == "activity": + await worker._act_semaphore.acquire() + await getattr(worker, f"_dispatch_{kind}_task")({"task_id": "synthetic-task"}) + assert assert_contract(recorder, duration=0.25) == {"worker_task"} + assert recorder.calls[0][3] == {"task_kind": kind, "task_queue": "orders", "outcome": expected} + + +@pytest.mark.parametrize("override", [False, True], ids=["inherit", "override"]) +async def test_actual_client_and_worker_respect_recorder_selection(override: bool) -> None: + client_recorder, worker_recorder = Recorder(), Recorder() + client = Client("https://runtime.example", worker_token="synthetic-token", metrics=client_recorder) + worker = Worker(client, task_queue="orders", metrics=worker_recorder if override else None) + + def respond(_: httpx.Request) -> httpx.Response: + worker._stop.set() + return httpx.Response(200, json={"task": None}) + + client._http = httpx.AsyncClient(base_url=client.base_url, transport=httpx.MockTransport(respond)) + client._runtime_external_payload_transport_resolved = True + async with client: + await worker._poll_workflow_tasks() + if override: + assert assert_contract(client_recorder) == {"client_request"} + assert assert_contract(worker_recorder) == {"worker_poll"} + else: + assert assert_contract(client_recorder) == {"client_request", "worker_poll"} + assert not worker_recorder.calls + + +async def test_default_noop_handles_real_client_and_worker_observations() -> None: + client = Client("https://runtime.example", worker_token="synthetic-token") + worker = Worker(client, task_queue="orders") + + def respond(_: httpx.Request) -> httpx.Response: + worker._stop.set() + return httpx.Response(200, json={"task": None}) + + client._http = httpx.AsyncClient(base_url=client.base_url, transport=httpx.MockTransport(respond)) + client._runtime_external_payload_transport_resolved = True + async with client: + await worker._poll_workflow_tasks() + assert client.metrics is NOOP_METRICS and worker.metrics is NOOP_METRICS + + +def test_prometheus_adapter_exports_counter_and_seconds_samples() -> None: + prometheus = pytest.importorskip("prometheus_client") + registry = prometheus.CollectorRegistry() + recorder = PrometheusMetrics(registry=registry) + for spec in CONTRACT["metrics"]: + tags = {name: "bounded" for name in spec["tags"]} + if spec["type"] == "counter": + recorder.increment(spec["name"], tags=tags) + assert registry.get_sample_value(spec["name"] + "_total", tags) == 1 + else: + recorder.record(spec["name"], 0.25, tags) + assert registry.get_sample_value(spec["name"] + "_sum", tags) == 0.25 + assert registry.get_sample_value(spec["name"] + "_count", tags) == 1 From a3c66937e91f6063a355eb4b27a54c9205054651 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 07:51:06 +0000 Subject: [PATCH 3/7] docs: explain bounded SDK metric labels and recorder semantics --- CHANGELOG.md | 7 +++++++ docs/reference/metrics.md | 35 +++++++++++++++++++++++++++++++++++ docs/sdk-reference.md | 6 ++++++ 3 files changed, 48 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 44d48ae..ed4977a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- Bound client metric routes to reviewed operation templates, including + update-validation tasks, worker sessions, streams and Nexus operations. + Unknown paths emit `unknown`; static poll routes retain `/poll`. Variable + identifiers and query strings do not create metric series. + ## [2.9.1] - 2026-10-22 ### Security diff --git a/docs/reference/metrics.md b/docs/reference/metrics.md index db82317..f2be9c0 100644 --- a/docs/reference/metrics.md +++ b/docs/reference/metrics.md @@ -1,3 +1,38 @@ # Metrics +## Recorder contract + +Client request, worker poll and task dispatch observations follow the +[shared SDK metrics contract](https://github.com/durable-workflow/.github/tree/main/sdk-metrics). +It defines six recorder names, counter or histogram instruments, count or +second units, and the exact tag keys. The default recorder drops observations. +A worker inherits its client's recorder unless given an explicit recorder. + +Each instrumented operation increments its counter by one and records one +nonnegative monotonic duration in seconds. Client observations include the +request's transport retries; they do not count each HTTP attempt. Poll and task +observations describe local worker activity, not durable execution counts, +workflow completion, billing units or schedule-to-start latency. + +Client route tags use reviewed path templates, without `/api`, query strings, +or variable IDs. For example, update-validation results use +`/worker/update-validation-tasks/{task_id}/approve`, and session renewal uses +`/worker/sessions/{session_id}/heartbeat`. Static poll operations retain their +`/poll` route. Unknown operations emit the single `unknown` route value. +Names of workflows, runs, activities, workers, sessions, queues, streams and +service operations in paths are replaced with placeholders. Request payloads, +credentials and exception messages are not core metric labels. + +The worker's configured `task_queue` label remains application-controlled. +Use bounded queue names rather than one per run. Exporter resource labels need +their own cardinality review. Missing observations do not establish zero work +or a healthy process; abrupt shutdown can prevent the final observation. + +`PrometheusMetrics` uses normal Prometheus exporter conventions: counter series +end in `_total`, and duration histograms expose bucket, sum and count series. +Those suffixes do not change the recorder names. Install the `prometheus` extra +to use this adapter. Applications own scrape/export configuration and additional +resource labels. Custom recorders should remain fast and handle their exporter +failures without throwing into the client or worker. + ::: durable_workflow.metrics diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index bbb4ddd..e6b6120 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -1000,6 +1000,12 @@ client = Client("http://server:8080", token="dev-token-123", metrics=metrics) Custom recorders implement `increment(name, value=1.0, tags=None)` and `record(name, value, tags=None)`. +The [metrics reference](reference/metrics.md#recorder-contract) explains the +shared names, count/second units, recorder inheritance and exporter conventions. +Client routes use reviewed templates without variable IDs or query strings; +unknown operations use `unknown`. Keep configured worker queue names and any +exporter resource labels bounded. + ## Worker interceptors Use `Worker(interceptors=[...])` when instrumentation needs the task payload, From 530dee13e7f20e27af061f6b06e2ba7bb2c23a79 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 07:52:22 +0000 Subject: [PATCH 4/7] docs: distinguish pending route normalization from installed behavior --- docs/reference/metrics.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/docs/reference/metrics.md b/docs/reference/metrics.md index f2be9c0..755c8fc 100644 --- a/docs/reference/metrics.md +++ b/docs/reference/metrics.md @@ -2,6 +2,10 @@ ## Recorder contract +The expanded route normalization described here is available in unreleased +SDK source. Published packages retain their existing route labels until the +next package release. + Client request, worker poll and task dispatch observations follow the [shared SDK metrics contract](https://github.com/durable-workflow/.github/tree/main/sdk-metrics). It defines six recorder names, counter or histogram instruments, count or From 53774763e819368a42f03693f2c11b9822000989 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 08:00:21 +0000 Subject: [PATCH 5/7] test: pin the merged shared SDK metrics contract --- docs/sdk-reference.md | 2 +- tests/fixtures/sdk_metrics/provenance.json | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index e6b6120..4fe81d1 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -1002,7 +1002,7 @@ Custom recorders implement `increment(name, value=1.0, tags=None)` and `record(n The [metrics reference](reference/metrics.md#recorder-contract) explains the shared names, count/second units, recorder inheritance and exporter conventions. -Client routes use reviewed templates without variable IDs or query strings; +In unreleased source, client routes use reviewed templates without variable IDs or query strings; unknown operations use `unknown`. Keep configured worker queue names and any exporter resource labels bounded. diff --git a/tests/fixtures/sdk_metrics/provenance.json b/tests/fixtures/sdk_metrics/provenance.json index d24cd9d..fc0a9b9 100644 --- a/tests/fixtures/sdk_metrics/provenance.json +++ b/tests/fixtures/sdk_metrics/provenance.json @@ -1,4 +1,4 @@ { - "source": "https://github.com/durable-workflow/.github/blob/f3627e2f52db6f7e4fbc2775e8c53ec8a758d96e/sdk-metrics/contract.json", + "source": "https://github.com/durable-workflow/.github/blob/73e8fc323e4f46322945a007e54a546395904f79/sdk-metrics/contract.json", "sha256": "c456f4f434c601fa26f0caba07ef0fd29f548ca8f5fc29ba2f6430f01e32c0f0" } From 1ad921e3cc0c3a779b1eabb4d70353df6c6e9041 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 08:17:57 +0000 Subject: [PATCH 6/7] test: separate recorder contract assertion diagnostics --- tests/test_sdk_metrics_contract.py | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/tests/test_sdk_metrics_contract.py b/tests/test_sdk_metrics_contract.py index dbcc272..0014a75 100644 --- a/tests/test_sdk_metrics_contract.py +++ b/tests/test_sdk_metrics_contract.py @@ -40,7 +40,8 @@ def record(self, name: str, value: float, tags: Mapping[str, str] | None = None) def assert_contract(recorder: Recorder, *, duration: float | None = None) -> set[str]: - assert recorder.calls and len(recorder.calls) % 2 == 0 + assert recorder.calls + assert len(recorder.calls) % 2 == 0 scopes: set[str] = set() for counter, histogram in zip(recorder.calls[::2], recorder.calls[1::2], strict=True): for method, name, value, tags in (counter, histogram): @@ -48,10 +49,13 @@ def assert_contract(recorder: Recorder, *, duration: float | None = None) -> set assert method == spec["recorder_method"] assert set(tags) == set(spec["tags"]) assert all(isinstance(value, str) for value in tags.values()) - assert math.isfinite(value) and value >= 0 + assert math.isfinite(value) + assert value >= 0 counter_spec, duration_spec = SPECS[counter[1]], SPECS[histogram[1]] - assert counter_spec["type"] == "counter" and counter_spec["unit"] == "1" - assert duration_spec["type"] == "histogram" and duration_spec["unit"] == "s" + assert counter_spec["type"] == "counter" + assert counter_spec["unit"] == "1" + assert duration_spec["type"] == "histogram" + assert duration_spec["unit"] == "s" assert counter_spec["scope"] == duration_spec["scope"] assert counter[2] == 1 assert counter[3] == histogram[3] @@ -214,7 +218,8 @@ def respond(_: httpx.Request) -> httpx.Response: client._runtime_external_payload_transport_resolved = True async with client: await worker._poll_workflow_tasks() - assert client.metrics is NOOP_METRICS and worker.metrics is NOOP_METRICS + assert client.metrics is NOOP_METRICS + assert worker.metrics is NOOP_METRICS def test_prometheus_adapter_exports_counter_and_seconds_samples() -> None: From 9e579edfe8198507d079aad8db38d1299454b1ae Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 08:29:24 +0000 Subject: [PATCH 7/7] fix: preserve valid false-valued metric recorders --- CHANGELOG.md | 2 ++ docs/reference/metrics.md | 2 ++ src/durable_workflow/client.py | 2 +- src/durable_workflow/worker.py | 2 +- tests/test_sdk_metrics_contract.py | 17 +++++++++++++---- 5 files changed, 19 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ed4977a..c39fbc1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 update-validation tasks, worker sessions, streams and Nexus operations. Unknown paths emit `unknown`; static poll routes retain `/poll`. Variable identifiers and query strings do not create metric series. +- Honor explicitly configured and inherited metric recorders even when they + evaluate to false, rather than silently replacing them with the no-op. ## [2.9.1] - 2026-10-22 diff --git a/docs/reference/metrics.md b/docs/reference/metrics.md index 755c8fc..4515e54 100644 --- a/docs/reference/metrics.md +++ b/docs/reference/metrics.md @@ -11,6 +11,8 @@ Client request, worker poll and task dispatch observations follow the It defines six recorder names, counter or histogram instruments, count or second units, and the exact tag keys. The default recorder drops observations. A worker inherits its client's recorder unless given an explicit recorder. +Selection uses `None` to mean unspecified; an empty or false-valued recorder +still receives observations. Pass `NoopMetrics()` explicitly to discard them. Each instrumented operation increments its counter by one and records one nonnegative monotonic duration in seconds. Client observations include the diff --git a/src/durable_workflow/client.py b/src/durable_workflow/client.py index d74c4c9..bf65c1c 100644 --- a/src/durable_workflow/client.py +++ b/src/durable_workflow/client.py @@ -1630,7 +1630,7 @@ def __init__( self._observability_capability = False self.timeout = timeout self.retry_policy = retry_policy or TransportRetryPolicy() - self.metrics = metrics or NOOP_METRICS + self.metrics = metrics if metrics is not None else NOOP_METRICS self.payload_size_warning_config = ( serializer.PayloadSizeWarningConfig( limit_bytes=payload_size_limit_bytes, diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index 7bbc75e..7c13484 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -1168,7 +1168,7 @@ def __init__( self._last_cpu_sample_at: float | None = None self._last_cpu_total_seconds: float = 0.0 configured_metrics = metrics if metrics is not None else getattr(client, "metrics", NOOP_METRICS) - self.metrics: MetricsRecorder = configured_metrics or NOOP_METRICS + self.metrics: MetricsRecorder = configured_metrics if configured_metrics is not None else NOOP_METRICS self.interceptors = tuple(interceptors) self.external_storage = ( external_storage diff --git a/tests/test_sdk_metrics_contract.py b/tests/test_sdk_metrics_contract.py index 0014a75..9a67a68 100644 --- a/tests/test_sdk_metrics_contract.py +++ b/tests/test_sdk_metrics_contract.py @@ -39,6 +39,11 @@ def record(self, name: str, value: float, tags: Mapping[str, str] | None = None) self.calls.append(("record", name, value, dict(tags or {}))) +class FalseyRecorder(Recorder): + def __bool__(self) -> bool: + return False + + def assert_contract(recorder: Recorder, *, duration: float | None = None) -> set[str]: assert recorder.calls assert len(recorder.calls) % 2 == 0 @@ -74,10 +79,11 @@ def test_contract_is_an_exact_pinned_copy() -> None: @pytest.mark.parametrize("outcome", ["ok", "http_error", "ConnectError", "retry"]) +@pytest.mark.parametrize("recorder_class", [Recorder, FalseyRecorder], ids=["normal", "falsey"]) async def test_actual_client_request_pair_matches_the_contract( - outcome: str, monkeypatch: pytest.MonkeyPatch, + outcome: str, recorder_class: type[Recorder], monkeypatch: pytest.MonkeyPatch, ) -> None: - recorder = Recorder() + recorder = recorder_class() now = 1.0 attempts = 0 @@ -185,8 +191,11 @@ async def execute(*_: object, **__: object) -> object: @pytest.mark.parametrize("override", [False, True], ids=["inherit", "override"]) -async def test_actual_client_and_worker_respect_recorder_selection(override: bool) -> None: - client_recorder, worker_recorder = Recorder(), Recorder() +@pytest.mark.parametrize("recorder_class", [Recorder, FalseyRecorder], ids=["normal", "falsey"]) +async def test_actual_client_and_worker_respect_recorder_selection( + override: bool, recorder_class: type[Recorder], +) -> None: + client_recorder, worker_recorder = recorder_class(), recorder_class() client = Client("https://runtime.example", worker_token="synthetic-token", metrics=client_recorder) worker = Worker(client, task_queue="orders", metrics=worker_recorder if override else None)