Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

## [Unreleased]

### Fixed

- `Worker.run_until()` now records the schedule-to-start histogram for activity tasks, as `run()` already did (#188).

## [2.15.1] - 2026-11-11

### Fixed
Expand Down
3 changes: 3 additions & 0 deletions src/durable_workflow/worker.py
Original file line number Diff line number Diff line change
Expand Up @@ -4743,6 +4743,9 @@ async def _retry_run_until_poll_refusal(
return True

async def _run_until_activity_task(self, task: dict[str, Any]) -> None:
# run_until() dispatches activities here instead of through
# _dispatch_activity_task, so record the Server-computed age too (#188).
self._record_task_schedule_to_start("activity", task)
task_start = time.perf_counter()
outcome = "error"
self._activity_inflight += 1
Expand Down
18 changes: 18 additions & 0 deletions tests/test_sdk_metrics_contract.py
Original file line number Diff line number Diff line change
Expand Up @@ -228,6 +228,24 @@ async def execute(_: object) -> object:
assert assert_contract(Recorder(recorder.calls[1:]), duration=19.0) == {"worker_task"}


async def test_run_until_activity_records_server_ready_age(monkeypatch: pytest.MonkeyPatch) -> None:
# run_until() runs activities through _run_until_activity_task, not
# _dispatch_activity_task; both must record schedule-to-start (#188).
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")

async def execute(_: object) -> object:
return "completed"

monkeypatch.setattr(worker, "_run_activity_task", execute)
await worker._run_until_activity_task({"ready_age_ms": 1250})
assert recorder.calls[0] == (
"record", SCHEDULE_TO_START, 1.25, {"task_kind": "activity", "task_queue": "orders"},
)


@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"}, []])
Expand Down
Loading