From b18872f8a5cc135c4e14fe2aa9758d7e9b2f4015 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 06:51:41 -0600 Subject: [PATCH] Record capability-gated task schedule-to-start metrics --- CHANGELOG.md | 2 + docs/reference/metrics.md | 40 +++++- src/durable_workflow/metrics.py | 1 + src/durable_workflow/worker.py | 21 +++ tests/fixtures/sdk_metrics/contract.json | 10 +- tests/fixtures/sdk_metrics/provenance.json | 4 +- tests/test_sdk_metrics_contract.py | 146 ++++++++++++++++++++- 7 files changed, 216 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4bea38eb..b708f011 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Worker schedule-to-start histogram `durable_workflow_worker_task_schedule_to_start_seconds` for workflow and activity dispatches, tagged only with `task_kind` and `task_queue`. It requires Server's `claimed_task_ready_age` capability and a valid integer `ready_age_ms`, records Server-computed milliseconds as seconds, preserves real zero and omits missing/invalid samples. The shared metrics contract is pinned at revision 2. Published-Server and installed-package acceptance remains pending in #188. + - Condition wait opens carry a deterministic `python:condition-wait:{ordinal}` occurrence ID for each authored call, including selection members. False-wake reopens preserve the first open's ID and remaining timeout; legacy chains without an ID keep that absence even when later physical opens have an ID. Separate calls and grace waits receive new ordinals. These identities enable unresolved-open deduplication on supporting runtimes. (#259) - Opt-in typed workflow values (#184). `@workflow.defn(..., typed=True)` decodes the arguments of `run` (after `ctx`) and of every signal, query, update and update-validator handler to their annotations. `ctx.start_child_workflow(..., result_type=T)` decodes a child's result, and `WorkflowHandle.result(result_type=T)` / `Client.get_result(..., result_type=T)` decode a workflow's result on the client. Dataclass and `Enum` values in client workflow, signal, update, query and schedule arguments, child and continue-as-new arguments, and workflow and update results are sent as their neutral values. A typed `TypedValueError` names the path but never the value. Untyped workflows receive plain values as before, so in-flight histories replay unchanged. diff --git a/docs/reference/metrics.md b/docs/reference/metrics.md index 54f9faa5..d6e33ca0 100644 --- a/docs/reference/metrics.md +++ b/docs/reference/metrics.md @@ -7,15 +7,16 @@ or later. 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 +It defines seven 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 -request's transport retries; they do not count each HTTP attempt. Poll and task +The existing client, poll and task processing instruments increment their +counter by one and record 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. @@ -40,6 +41,37 @@ 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. +### Schedule-to-start latency (unreleased) + +This unreleased feature adds the histogram below. Package and published-Server +acceptance remains pending in +[#188](https://github.com/durable-workflow/sdk-python/issues/188). + +| Recorder name | Instrument / unit | Tag keys | Measurement | +| --- | --- | --- | --- | +| `durable_workflow_worker_task_schedule_to_start_seconds` | Histogram / seconds | `task_kind`, `task_queue` | This attempt's known eligibility instant to its committed Server lease instant. | + +At actual workflow or activity dispatch, the worker requires discovery's +`worker_protocol.server_capabilities.claimed_task_ready_age` to be Boolean +`true` and the claimed task's `ready_age_ms` to be an integer in +`0..9007199254740991`, excluding booleans. It divides that value by 1000 and +records once, before running the task. This histogram has no paired counter +or `outcome` tag. Query and update-validation tasks are outside its scope. +HTTP retries, empty or rejected polls do not record this sample. + +The Server-computed age excludes delay/backoff before eligibility, transport +time and local handler processing. Supporting Servers must retain a known +eligibility origin for newly created, retried, released and expired attempts. +A replayed receipt of the same lease preserves its original age; a new lease +uses its own eligibility origin. The worker never subtracts its clock from a +Server timestamp or substitutes an aggregate oldest-ready age. + +Missing or unsupported fields produce no sample. A malformed value produces +no sample and a fixed warning without task IDs or payload data. A real zero +is recorded; missing data is never replaced with zero. Legacy or genuinely +unknown/invalid origins may be omitted by Server. Server clock assumptions +still apply even though worker clock skew cannot enter this measurement. + ## Request outcomes and errors Named poll warnings and the local cancellation outcome below require Python diff --git a/src/durable_workflow/metrics.py b/src/durable_workflow/metrics.py index eb4d95fd..151bdbb9 100644 --- a/src/durable_workflow/metrics.py +++ b/src/durable_workflow/metrics.py @@ -26,6 +26,7 @@ WORKER_POLL_DURATION_SECONDS = "durable_workflow_worker_poll_duration_seconds" WORKER_TASKS = "durable_workflow_worker_tasks" WORKER_TASK_DURATION_SECONDS = "durable_workflow_worker_task_duration_seconds" +WORKER_TASK_SCHEDULE_TO_START_SECONDS = "durable_workflow_worker_task_schedule_to_start_seconds" class MetricsRecorder(Protocol): diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index dae62da0..cc4e3edb 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -104,6 +104,7 @@ WORKER_POLL_DURATION_SECONDS, WORKER_POLLS, WORKER_TASK_DURATION_SECONDS, + WORKER_TASK_SCHEDULE_TO_START_SECONDS, WORKER_TASKS, MetricsRecorder, ) @@ -1292,6 +1293,24 @@ def _record_task_metrics(self, task_kind: str, outcome: str, duration: float) -> self.metrics.increment(WORKER_TASKS, tags=tags) self.metrics.record(WORKER_TASK_DURATION_SECONDS, duration, tags=tags) + def _record_task_schedule_to_start(self, task_kind: str, task: dict[str, Any]) -> None: + # Use discovery cached by registration, as activity lease renewal does. + info = getattr(self.client, "_cluster_info", None) + protocol = info.get("worker_protocol") if isinstance(info, Mapping) else None + capabilities = protocol.get("server_capabilities") if isinstance(protocol, Mapping) else None + if not isinstance(capabilities, Mapping) or capabilities.get("claimed_task_ready_age") is not True: + return + if "ready_age_ms" not in task: + return + ready_age_ms = task["ready_age_ms"] + if type(ready_age_ms) is not int or not 0 <= ready_age_ms <= 9007199254740991: + log.warning("invalid claimed task ready_age_ms; schedule-to-start sample omitted") + return + self.metrics.record( + WORKER_TASK_SCHEDULE_TO_START_SECONDS, ready_age_ms / 1000, + tags={"task_kind": task_kind, "task_queue": self.task_queue}, + ) + def _payload_size_warning_config(self) -> serializer.PayloadSizeWarningConfig | None: config = getattr(self.client, "payload_size_warning_config", serializer.DEFAULT_PAYLOAD_SIZE_WARNING) if config is None or isinstance(config, serializer.PayloadSizeWarningConfig): @@ -3871,6 +3890,7 @@ async def _poll_workflow_tasks(self) -> None: async def _dispatch_workflow_task( self, task: dict[str, Any] ) -> list[dict[str, Any]] | None: + self._record_task_schedule_to_start("workflow", task) task_start = time.perf_counter() outcome = "error" commands: list[dict[str, Any]] | None = None @@ -3979,6 +3999,7 @@ async def _poll_activity_tasks(self) -> None: self._track(self._dispatch_activity_task(task)) async def _dispatch_activity_task(self, task: dict[str, Any]) -> None: + self._record_task_schedule_to_start("activity", task) task_start = time.perf_counter() outcome = "error" self._activity_inflight += 1 diff --git a/tests/fixtures/sdk_metrics/contract.json b/tests/fixtures/sdk_metrics/contract.json index 07263924..f6b675ef 100644 --- a/tests/fixtures/sdk_metrics/contract.json +++ b/tests/fixtures/sdk_metrics/contract.json @@ -1,6 +1,6 @@ { "schema": "durable-workflow.sdk-metrics/v1", - "revision": 1, + "revision": 2, "reference": { "repository": "durable-workflow/sdk-python", "commit": "2da02a7ac8a68a7cd4267e86ef10e0198a309592", @@ -58,6 +58,14 @@ "recorder_method": "record", "scope": "worker_task", "tags": ["task_kind", "task_queue", "outcome"] + }, + { + "name": "durable_workflow_worker_task_schedule_to_start_seconds", + "type": "histogram", + "unit": "s", + "recorder_method": "record", + "scope": "worker_task_dispatch", + "tags": ["task_kind", "task_queue"] } ] } diff --git a/tests/fixtures/sdk_metrics/provenance.json b/tests/fixtures/sdk_metrics/provenance.json index fc0a9b91..600291b0 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/73e8fc323e4f46322945a007e54a546395904f79/sdk-metrics/contract.json", - "sha256": "c456f4f434c601fa26f0caba07ef0fd29f548ca8f5fc29ba2f6430f01e32c0f0" + "source": "https://github.com/durable-workflow/.github/blob/23f9bc3c3798cb6aab6ccf5f9cd210854b820c65/sdk-metrics/contract.json", + "sha256": "081d98ec6f5f6e3812f81ddbe8d2589d75e7c6e457c580dc6631d7857594516a" } diff --git a/tests/test_sdk_metrics_contract.py b/tests/test_sdk_metrics_contract.py index 9a67a683..a0065cb2 100644 --- a/tests/test_sdk_metrics_contract.py +++ b/tests/test_sdk_metrics_contract.py @@ -2,6 +2,7 @@ import hashlib import json +import logging import math import re import time @@ -22,10 +23,12 @@ from durable_workflow.metrics import NOOP_METRICS, PrometheusMetrics from durable_workflow.retry_policy import TransportRetryPolicy from durable_workflow.worker import Worker +from tests.test_worker import compatible_cluster_info 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"]} +SCHEDULE_TO_START = "durable_workflow_worker_task_schedule_to_start_seconds" @dataclass @@ -122,6 +125,7 @@ async def test_actual_worker_poll_pair_matches_the_contract( recorder = Recorder() client = AsyncMock(spec=Client) worker = Worker(client, task_queue="orders", metrics=recorder) + client._cluster_info = {"worker_protocol": {"server_capabilities": {"claimed_task_ready_age": True}}} now = 1.0 async def poll(**_: object) -> dict[str, str] | None: @@ -136,7 +140,7 @@ async def poll(**_: object) -> dict[str, str] | None: "task_queue": "orders", "retryable": True, "retry_after_seconds": 2, }) if outcome == "task": - return {"task_id": "synthetic-task"} + return {"task_id": "synthetic-task", "ready_age_ms": 1250} worker._stop.set() return None @@ -190,6 +194,146 @@ async def execute(*_: object, **__: object) -> object: assert recorder.calls[0][3] == {"task_kind": kind, "task_queue": "orders", "outcome": expected} +@pytest.mark.parametrize("kind", ["workflow", "activity"]) +@pytest.mark.parametrize("ready_age_ms", [0, 1, 1250, 9007199254740991]) +@pytest.mark.parametrize("recorder_class", [Recorder, FalseyRecorder], ids=["normal", "falsey"]) +async def test_dispatch_records_server_ready_age_before_execution( + kind: str, ready_age_ms: int, recorder_class: type[Recorder], monkeypatch: pytest.MonkeyPatch, +) -> None: + recorder = recorder_class() + client = Client("https://runtime.example", metrics=recorder) + client._cluster_info = {"worker_protocol": {"server_capabilities": {"claimed_task_ready_age": True}}} + worker = Worker(client, task_queue="orders") + now = 1.0 + expected_sample = ("record", SCHEDULE_TO_START, ready_age_ms / 1000, + {"task_kind": kind, "task_queue": "orders"}) + + async def execute(_: object) -> object: + nonlocal now + assert recorder.calls == [expected_sample] + now = 20.0 # Processing time must not enter the Server-computed age. + return [] if kind == "workflow" else "completed" + + monkeypatch.setattr(worker_module, "time", SimpleNamespace(perf_counter=lambda: now)) + monkeypatch.setattr(worker, f"_run_{kind}_task", execute) + if kind == "activity": + await worker._act_semaphore.acquire() + await getattr(worker, f"_dispatch_{kind}_task")({"ready_age_ms": ready_age_ms}) + assert recorder.calls[0] == expected_sample + spec = SPECS[SCHEDULE_TO_START] + assert (spec["recorder_method"], spec["type"], spec["unit"], spec["scope"]) == ( + "record", "histogram", "s", "worker_task_dispatch", + ) + assert set(expected_sample[3]) == set(spec["tags"]) + assert assert_contract(Recorder(recorder.calls[1:]), duration=19.0) == {"worker_task"} + + +@pytest.mark.parametrize("kind", ["workflow", "activity"]) +@pytest.mark.parametrize("value", [None, True, False, -1, 9007199254740992, 1.0, float("nan"), + float("inf"), "private-value", {"private-key": "private-value"}, []]) +async def test_malformed_ready_age_drops_only_the_sample_with_a_bounded_warning( + kind: str, value: object, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, +) -> None: + recorder = Recorder() + client = Client("https://runtime.example", metrics=recorder) + client._cluster_info = {"worker_protocol": {"server_capabilities": {"claimed_task_ready_age": True}}} + worker = Worker(client, task_queue="private-queue") + execute = AsyncMock(return_value=[] if kind == "workflow" else "completed") + monkeypatch.setattr(worker, f"_run_{kind}_task", execute) + if kind == "activity": + await worker._act_semaphore.acquire() + with caplog.at_level(logging.WARNING, logger="durable_workflow.worker"): + await getattr(worker, f"_dispatch_{kind}_task")({"ready_age_ms": value, "task_id": "private-id"}) + execute.assert_awaited_once() + assert assert_contract(recorder) == {"worker_task"} + assert [r.getMessage() for r in caplog.records] == [ + "invalid claimed task ready_age_ms; schedule-to-start sample omitted", + ] + + +@pytest.mark.parametrize("kind", ["workflow", "activity"]) +@pytest.mark.parametrize("value", [1250, "unsupported-private-value"]) +@pytest.mark.parametrize("info", [None, {}, {"worker_protocol": None}, {"worker_protocol": []}, + *({"worker_protocol": {"server_capabilities": caps}} for caps in ( + None, [], {}, {"claimed_task_ready_age": False}, {"claimed_task_ready_age": 1}, + {"claimed_task_ready_age": "true"}, {"claimed_task_ready_age": {"supported": True}}, + )), +]) +async def test_ready_age_requires_explicit_boolean_discovery( + kind: str, value: object, info: Any, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, +) -> None: + recorder = Recorder() + client = Client("https://runtime.example", metrics=recorder) + client._cluster_info = info + worker = Worker(client, task_queue="orders") + monkeypatch.setattr(worker, f"_run_{kind}_task", AsyncMock(return_value=[] if kind == "workflow" else "completed")) + if kind == "activity": + await worker._act_semaphore.acquire() + await getattr(worker, f"_dispatch_{kind}_task")({"ready_age_ms": value}) + assert assert_contract(recorder) == {"worker_task"} + assert not caplog.records + + +@pytest.mark.parametrize("kind", ["workflow", "activity", "query", "update_validation"]) +async def test_missing_age_and_other_dispatch_kinds_emit_no_schedule_sample( + kind: str, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, +) -> None: + recorder = Recorder() + client = Client("https://runtime.example", metrics=recorder) + client._cluster_info = {"worker_protocol": {"server_capabilities": {"claimed_task_ready_age": True}}} + worker = Worker(client, task_queue="orders") + result: object = [] if kind == "workflow" else "approved" if kind == "update_validation" else "completed" + monkeypatch.setattr(worker, f"_run_{kind}_task", AsyncMock(return_value=result)) + if kind == "activity": + await worker._act_semaphore.acquire() + task = {} if kind in {"workflow", "activity"} else {"ready_age_ms": 1250} + await getattr(worker, f"_dispatch_{kind}_task")(task) + assert assert_contract(recorder) == {"worker_task"} + assert not caplog.records + + +@pytest.mark.parametrize("kind", ["workflow", "activity"]) +async def test_registration_discovery_and_retried_claim_record_once_at_dispatch( + kind: str, monkeypatch: pytest.MonkeyPatch, +) -> None: + recorder = Recorder() + info = compatible_cluster_info() + protocol = info["worker_protocol"] + assert isinstance(protocol, dict) + protocol["server_capabilities"] = {"claimed_task_ready_age": True} + attempts = 0 + + def respond(request: httpx.Request) -> httpx.Response: + nonlocal attempts + if request.url.path == "/api/cluster/info": + return httpx.Response(200, json=info) + if request.url.path == "/api/worker/register": + return httpx.Response(200, json={"worker_id": "worker", "registered": True}) + assert request.url.path == f"/api/worker/{kind}-tasks/poll" + attempts += 1 + return httpx.Response(503 if attempts == 1 else 200, json={"task": {"ready_age_ms": 1250}}) + + client = Client("https://runtime.example", worker_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)) + client._runtime_external_payload_transport_resolved = True + worker = Worker(client, task_queue="orders", worker_id="worker") + monkeypatch.setattr(worker, f"_run_{kind}_task", AsyncMock(return_value=[] if kind == "workflow" else "completed")) + async with client: + await worker._register() + task = await getattr(client, f"poll_{kind}_task")(worker_id="worker", task_queue="orders") + assert attempts == 2 + assert all(call[1] != SCHEDULE_TO_START for call in recorder.calls) + assert task is not None + if kind == "activity": + await worker._act_semaphore.acquire() + await getattr(worker, f"_dispatch_{kind}_task")(task) + assert [call for call in recorder.calls if call[1] == SCHEDULE_TO_START] == [ + ("record", SCHEDULE_TO_START, 1.25, {"task_kind": kind, "task_queue": "orders"}), + ] + assert_contract(Recorder([call for call in recorder.calls if call[1] != SCHEDULE_TO_START])) + + @pytest.mark.parametrize("override", [False, True], ids=["inherit", "override"]) @pytest.mark.parametrize("recorder_class", [Recorder, FalseyRecorder], ids=["normal", "falsey"]) async def test_actual_client_and_worker_respect_recorder_selection(