diff --git a/CHANGELOG.md b/CHANGELOG.md index 0c516fa..35612b2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Lazy `Client.iter_workflows`, `Client.iter_history` and `WorkflowHandle.iter_history` follow opaque page tokens while retaining only the current page and bounded cursor bookkeeping. They preserve run/filter/namespace identity, follow empty continued pages, stop on early close, and surface HTTP, cancellation and invalid/cyclic cursor errors. Existing one-page methods keep their contracts. (durable-workflow/.github#247) + ## [2.15.0] - 2026-11-04 ### Added diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index a2f6041..404d956 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -957,6 +957,38 @@ activities, for example `OrderInput(order_id=UUID(data["order_id"]), ...)`. Adapter output is part of the durable history contract, so changing that shape is a workflow compatibility change. +## Whole run histories and workflow lists (unreleased) + +`Client.get_history(...)` and `Client.list_workflows(...)` each return one +Server page. Use the lazy iterators to visit all pages without collecting +them in memory: + +```python +async for execution in client.iter_workflows(workflow_type="invoice", page_size=100): + print(execution.workflow_id) + +async for event in client.iter_history(workflow_id, run_id, page_size=100): + print(event["event_type"]) +``` + +`WorkflowHandle.iter_history(...)` offers the same traversal when the handle +has an explicit `run_id`. List iterators yield `WorkflowExecution` values; +history iterators yield the event dictionaries from `get_history` in Server +order. Filters, run identity, namespace and page size stay fixed. A supplied +`next_page_token` resumes at that opaque continuation; do not parse or alter it. + +The next page is fetched only after consuming the current page. Stop or close +the iterator to avoid later requests. Empty pages with a continuation are +followed, while a missing or empty token ends traversal. HTTP errors, +cancellation and malformed or cyclic continuations propagate, possibly after +earlier items have been yielded. A changing namespace is refused before the +next request. This traverses the Server's cursor results; it does not create a +snapshot of a concurrently changing workflow list. + +These helpers are unreleased source tracked in +[the shared paging issue](https://github.com/durable-workflow/.github/issues/247). +Published-package and actual multi-page acceptance remain separate. + ## Typed activity values Typed activity arguments and activity `result_type` are available from SDK diff --git a/src/durable_workflow/client.py b/src/durable_workflow/client.py index 1f96f9b..cf009d1 100644 --- a/src/durable_workflow/client.py +++ b/src/durable_workflow/client.py @@ -378,6 +378,33 @@ class _RuntimeExternalPayloadTransport: prepared_completion_context: bool = False +class _PageTokens: + """Detect cyclic opaque continuations with constant-space checkpoints.""" + + def __init__(self, token: str | None) -> None: + self.current = token + self._checkpoint = token + self._distance = 0 + self._power = 1 + + def advance(self, token: Any) -> bool: + if token is None or token == "": + self.current = None + return False + if not isinstance(token, str): + raise RuntimeError("pagination returned an invalid page token") + if token == self.current or token == self._checkpoint: + raise RuntimeError("pagination repeated a page token") + self.current = token + self._distance += 1 + # Brent's cycle detector retains one checkpoint, not every old token. + if self._distance == self._power: + self._checkpoint = token + self._power *= 2 + self._distance = 0 + return True + + @dataclass class WorkflowExecution: """Current server view of one workflow execution.""" @@ -1376,6 +1403,17 @@ async def export_history(self) -> Any: raise ValueError("run_id is required to export workflow history from a handle") return await self._client.export_history(self.workflow_id, self.run_id) + async def iter_history( + self, *, page_size: int | None = None, next_page_token: str | None = None, + ) -> AsyncIterator[dict[str, Any]]: + """Yield this explicit run's history. See :meth:`Client.iter_history`.""" + if self.run_id is None: + raise ValueError("run_id is required to iterate workflow history from a handle") + async for event in self._client.iter_history( + self.workflow_id, self.run_id, page_size=page_size, next_page_token=next_page_token, + ): + yield event + async def list_runs(self) -> WorkflowRunList: """List all runs in this workflow execution chain. See :meth:`Client.list_workflow_runs`.""" return await self._client.list_workflow_runs(self.workflow_id) @@ -3871,6 +3909,39 @@ async def list_workflows( next_page_token=data.get("next_page_token"), ) + async def iter_workflows( + self, + *, + workflow_type: str | None = None, + status: str | None = None, + query: str | None = None, + page_size: int | None = None, + next_page_token: str | None = None, + ) -> AsyncIterator[WorkflowExecution]: + """Yield workflow list results lazily, retaining no previous pages. + + Filters and namespace stay fixed across pages. An empty page with a + continuation is followed; missing/empty continuation ends iteration. + Cancellation, malformed/cyclic continuations and HTTP errors propagate, + possibly after earlier results have been yielded. This is a traversal + of Server pages, not a snapshot of concurrently changing workflows. + """ + tokens = _PageTokens(next_page_token) + namespace = self.namespace + while True: + if self.namespace != namespace: + raise RuntimeError("client namespace changed during pagination") + page = await self.list_workflows( + workflow_type=workflow_type, status=status, query=query, page_size=page_size, + next_page_token=tokens.current, + ) + more = tokens.advance(page.next_page_token) + for execution in page.executions: + yield execution + del page + if not more: + return + async def get_history( self, workflow_id: str, @@ -3893,6 +3964,39 @@ async def get_history( "GET", path, context=workflow_id ) + async def iter_history( + self, + workflow_id: str, + run_id: str, + *, + page_size: int | None = None, + next_page_token: str | None = None, + ) -> AsyncIterator[dict[str, Any]]: + """Yield run-history events in Server order, fetching one page at a time. + + Workflow/run/namespace identity and page size stay fixed. Stopping the + iterator fetches no subsequent page. Empty continued pages are followed; + malformed/cyclic continuations and HTTP errors propagate rather than + appearing as a complete history. Already yielded events are not undone. + """ + tokens = _PageTokens(next_page_token) + namespace = self.namespace + while True: + if self.namespace != namespace: + raise RuntimeError("client namespace changed during pagination") + page = await self.get_history( + workflow_id, run_id, page_size=page_size, next_page_token=tokens.current, + ) + if (not isinstance(page, dict) or not isinstance(page.get("events"), list) + or any(not isinstance(event, dict) for event in page["events"])): + raise RuntimeError("history pagination returned invalid events") + more = tokens.advance(page.get("next_page_token")) + for event in page["events"]: + yield event + del page + if not more: + return + async def export_history(self, workflow_id: str, run_id: str) -> Any: """Export one workflow run history as a replay bundle.""" return await self._request( diff --git a/tests/test_client_iterators.py b/tests/test_client_iterators.py new file mode 100644 index 0000000..60720ab --- /dev/null +++ b/tests/test_client_iterators.py @@ -0,0 +1,207 @@ +from __future__ import annotations + +import asyncio +from collections.abc import AsyncIterator, Callable +from typing import Any + +import httpx +import pytest + +from durable_workflow.client import Client +from durable_workflow.errors import ServerError +from durable_workflow.retry_policy import TransportRetryPolicy + + +def paged_client(handler: Callable[[httpx.Request], Any]) -> Client: + client = Client("https://runtime.example", control_token="synthetic-control", namespace="tenant-a", + retry_policy=TransportRetryPolicy(max_attempts=1)) + client._http = httpx.AsyncClient(base_url=client.base_url, transport=httpx.MockTransport(handler)) + return client + + +def page(kind: str, values: list[int], token: Any = None) -> dict[str, Any]: + if kind == "workflows": + items = [{"workflow_id": str(value), "run_id": "run", "workflow_type": "invoice"} for value in values] + return {"workflows": items, "next_page_token": token} + return {"events": [{"sequence": value, "event_type": "CustomEvent"} for value in values], + "next_page_token": token} + + +def iterate(client: Client, kind: str, *, token: str | None = None) -> AsyncIterator[Any]: + if kind == "workflows": + return client.iter_workflows(workflow_type="invoice", status="running", query="customer = 'fixed'", + page_size=2, next_page_token=token) + return client.iter_history("wf-fixed", "run-fixed", page_size=2, next_page_token=token) + + +def value_of(kind: str, item: Any) -> int: + return int(item.workflow_id) if kind == "workflows" else int(item["sequence"]) + + +@pytest.mark.parametrize("kind", ["workflows", "history"]) +async def test_lazy_pages_preserve_identity_filters_and_opaque_tokens(kind: str) -> None: + requests: list[httpx.Request] = [] + pages = [page(kind, [1, 2], "opaque+/="), page(kind, [], "tail"), page(kind, [3])] + + def respond(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response(200, json=pages[len(requests) - 1]) + + async with paged_client(respond) as client: + iterator = iterate(client, kind) + assert not requests + assert value_of(kind, await anext(iterator)) == 1 + assert len(requests) == 1 + assert value_of(kind, await anext(iterator)) == 2 + assert len(requests) == 1 + assert value_of(kind, await anext(iterator)) == 3 + assert len(requests) == 3 # The empty continued page is followed. + with pytest.raises(StopAsyncIteration): + await anext(iterator) + + assert [r.url.params.get("next_page_token") for r in requests] == [None, "opaque+/=", "tail"] + assert all(r.headers["X-Namespace"] == "tenant-a" for r in requests) + assert all(r.url.params["page_size"] == "2" for r in requests) + assert "opaque%2B%2F%3D" in str(requests[1].url) + if kind == "history": + assert {r.url.path for r in requests} == {"/api/workflows/wf-fixed/runs/run-fixed/history"} + else: + for request in requests: + assert request.url.path == "/api/workflows" + assert request.url.params["workflow_type"] == "invoice" + assert request.url.params["status"] == "running" + assert request.url.params["query"] == "customer = 'fixed'" + + +@pytest.mark.parametrize("kind", ["workflows", "history"]) +async def test_closing_early_does_not_fetch_the_next_page(kind: str) -> None: + requests: list[httpx.Request] = [] + + def respond(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response(200, json=page(kind, [1, 2], "next")) + + async with paged_client(respond) as client: + iterator = iterate(client, kind) + assert value_of(kind, await anext(iterator)) == 1 + await iterator.aclose() # type: ignore[attr-defined] + assert len(requests) == 1 + + +@pytest.mark.parametrize("kind", ["workflows", "history"]) +@pytest.mark.parametrize("final_token", [None, ""]) +async def test_empty_final_page_terminates_without_another_request(kind: str, final_token: str | None) -> None: + async with paged_client(lambda _: httpx.Response(200, json=page(kind, [], final_token))) as client: + assert [item async for item in iterate(client, kind)] == [] + + +@pytest.mark.parametrize("kind", ["workflows", "history"]) +async def test_missing_continuation_is_a_final_page(kind: str) -> None: + body = page(kind, [9]) + del body["next_page_token"] + async with paged_client(lambda _: httpx.Response(200, json=body)) as client: + items = [item async for item in iterate(client, kind)] + assert [value_of(kind, item) for item in items] == [9] + + +@pytest.mark.parametrize("kind", ["workflows", "history"]) +async def test_resume_token_is_sent_without_normalization(kind: str) -> None: + requests: list[httpx.Request] = [] + + def respond(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response(200, json=page(kind, [7])) + + async with paged_client(respond) as client: + items = [item async for item in iterate(client, kind, token=" resume/+? ")] + assert [value_of(kind, item) for item in items] == [7] + assert requests[0].url.params["next_page_token"] == " resume/+? " + + +@pytest.mark.parametrize("kind", ["workflows", "history"]) +@pytest.mark.parametrize("tokens", [["same"], ["a", "b"], ["a", "b", "c"]]) +async def test_cyclic_continuations_fail_instead_of_looping(kind: str, tokens: list[str]) -> None: + calls = 0 + + def respond(_: httpx.Request) -> httpx.Response: + nonlocal calls + token = tokens[calls % len(tokens)] + calls += 1 + assert calls <= 8, "cyclic pagination did not terminate" + return httpx.Response(200, json=page(kind, [], token)) + + async with paged_client(respond) as client: + with pytest.raises(RuntimeError, match="repeated a page token"): + _ = [item async for item in iterate(client, kind)] + + +@pytest.mark.parametrize("kind", ["workflows", "history"]) +@pytest.mark.parametrize("token", [False, 0, 1, [], {}]) +async def test_malformed_continuations_are_not_silent_end_of_data(kind: str, token: Any) -> None: + async with paged_client(lambda _: httpx.Response(200, json=page(kind, [1], token))) as client: + with pytest.raises(RuntimeError, match="invalid page token"): + _ = [item async for item in iterate(client, kind)] + + +@pytest.mark.parametrize("kind", ["workflows", "history"]) +async def test_namespace_change_cannot_reuse_a_cursor_in_another_namespace(kind: str) -> None: + calls = 0 + + def respond(_: httpx.Request) -> httpx.Response: + nonlocal calls + calls += 1 + return httpx.Response(200, json=page(kind, [1], "next")) + + async with paged_client(respond) as client: + iterator = iterate(client, kind) + await anext(iterator) + client.namespace = "another-tenant" + with pytest.raises(RuntimeError, match="namespace changed"): + await anext(iterator) + assert calls == 1 + + +@pytest.mark.parametrize("kind", ["workflows", "history"]) +@pytest.mark.parametrize("cancelled", [False, True], ids=["http-error", "cancellation"]) +async def test_later_page_error_or_cancellation_propagates(kind: str, cancelled: bool) -> None: + calls = 0 + + async def respond(_: httpx.Request) -> httpx.Response: + nonlocal calls + calls += 1 + if calls == 1: + return httpx.Response(200, json=page(kind, [1], "next")) + if cancelled: + raise asyncio.CancelledError() + return httpx.Response(503, json={"message": "temporarily unavailable"}) + + async with paged_client(respond) as client: + iterator = iterate(client, kind) + assert value_of(kind, await anext(iterator)) == 1 + with pytest.raises(asyncio.CancelledError if cancelled else ServerError): + await anext(iterator) + assert calls == 2 + + +@pytest.mark.parametrize("events", [None, {}, [None], ["bad"]]) +async def test_history_requires_an_event_array(events: Any) -> None: + async with paged_client(lambda _: httpx.Response(200, json={"events": events})) as client: + with pytest.raises(RuntimeError, match="invalid events"): + _ = [event async for event in client.iter_history("workflow", "run")] + + +async def test_handle_iteration_requires_an_explicit_run_and_uses_it() -> None: + requests: list[httpx.Request] = [] + + def respond(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response(200, json=page("history", [1])) + + async with paged_client(respond) as client: + missing = client.get_workflow_handle("wf") + with pytest.raises(ValueError, match="run_id is required"): + _ = [event async for event in missing.iter_history()] + assert not requests + handle = client.get_workflow_handle("wf", run_id="original-run") + assert [event["sequence"] async for event in handle.iter_history()] == [1] + assert requests[0].url.path == "/api/workflows/wf/runs/original-run/history"