Record Server-computed task schedule-to-start metrics

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.

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

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):

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

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"]
     }
   ]
 }

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"
 }

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(