Prepare compatible Python 2.5.14 query authority patch

docs/query-task-authority.md

@@ -0,0 +1,40 @@
+# Query loading and execution deadlines
+
+Query loading and replay consume the original query budget. They do not give a
+worker extra time to invoke a handler. When the query lease expires, the worker
+stops loading and suppresses the handler. Server still decides whether a result
+can be accepted.
+
+Servers advertising `query_task_authority=true` expose a non-renewing query
+status API for protocol 1.20. The SDK checks the original claim while loading
+and again after replay, immediately before invoking the handler. Missing,
+malformed, denied or unavailable observations suppress execution.
+
+To require that capability before registering a worker:
+
+```python
+worker = Worker(
+    client,
+    task_queue="orders",
+    workflows=[OrderWorkflow],
+    require_query_task_authority=True,
+)
+```
+
+The default retains compatibility with older Servers and explicit protocol
+1.19. This path checks the supplied lease and any original caller deadline
+against the local clock, then charges elapsed time monotonically. It cannot
+observe an earlier caller timeout that Server did not expose. Use synchronized
+clocks, or require the newer authority capability when that limitation is
+unacceptable. A missing or malformed lease cannot authorize execution.
+
+An already-running synchronous Python handler cannot be safely interrupted.
+Keep query handlers short and free of side effects. Expiry can cancel async
+execution and payload I/O, but an observation never reserves future commit
+authority. A result refused by Server does not restart the query or its budget.
+
+The non-renewing authority path is qualified against Server 2.8.0, including
+expiry during external payload loading and async handler execution. Earlier
+Server 2.6.3 remains compatible in default 1.20 and explicit 1.19 mode, with the
+legacy limits described above. Qualification and release follow-through are
+recorded in [#143](https://github.com/durable-workflow/sdk-python/issues/143).

mkdocs.yml

@@ -77,6 +77,7 @@ nav:
   - Start here: index.md
   - SDK reference: sdk-reference.md
   - Sticky execution: guides/sticky-execution.md
+  - Query deadlines: query-task-authority.md
   - SDK guide: https://durable-workflow.com/docs/2.0/polyglot/python/
   - PyPI: https://pypi.org/project/durable-workflow/
   - API reference:

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "durable-workflow"
-version = "2.5.13"
+version = "2.5.14"
 description = "Python client and worker SDK for Durable Workflow Cloud and self-hosted Server"
 readme = "README.md"
 requires-python = ">=3.10"
@@ -71,9 +71,9 @@ durable-workflow-replay-conformance = "durable_workflow.replay_conformance:main"
 durable-workflow-workflow-updates-conformance = "durable_workflow.workflow_updates_conformance:main"
 
 [tool.durable-workflow]
-product-train = "2.5.13"
-registry-version = "2.5.13"
-supported-server-versions = "2.5.16"
+product-train = "2.5.14"
+registry-version = "2.5.14"
+supported-server-versions = "2.8.0"
 worker-protocol-version = "1.20"
 control-plane-version = "2"
 durable-selection = true

src/durable_workflow/__init__.py

@@ -88,6 +88,7 @@
     NonDeterministicReplayError,
     NonRetryableError,
     QueryFailed,
+    QueryTaskAuthorityLost,
     RuntimeCapabilityUnsupported,
     RuntimeDiscoveryUnavailable,
     SagaCompensationFailed,
@@ -375,6 +376,7 @@
     "InvalidWorkflowDefinition",
     "NoopMetrics",
     "QueryFailed",
+    "QueryTaskAuthorityLost",
     "RuntimeCapabilityUnsupported",
     "RuntimeDiscoveryUnavailable",
     "RuntimeExternalPayloadReference",

src/durable_workflow/_query_task_authority.py

@@ -0,0 +1,153 @@
+"""A query delivery's original, non-renewing execution budget."""
+
+from __future__ import annotations
+
+import asyncio
+import re
+import time
+from collections.abc import Awaitable, Callable, Mapping
+from contextvars import ContextVar
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Any, TypeVar
+
+from .errors import QueryTaskAuthorityLost
+
+_T = TypeVar("_T")
+_UTC6 = re.compile(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{6}Z\Z")
+_OBSERVATION_INTERVAL_SECONDS = 5.0
+
+
+def _timestamp(value: Any, *, canonical: bool) -> datetime:
+    if not isinstance(value, str) or (canonical and _UTC6.fullmatch(value) is None):
+        raise ValueError("missing or malformed query timing")
+    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    if parsed.tzinfo is None:
+        raise ValueError("query timing must include a timezone")
+    return parsed
+
+
+@dataclass
+class QueryTaskAuthority:
+    client: object
+    namespace: str
+    task_id: str
+    attempt: int
+    owner: str
+    observed: bool
+    timing: dict[str, str]
+    until: float
+    identity: dict[str, Any]
+    observer: Callable[[], Awaitable[Mapping[str, Any]]] | None = field(default=None, repr=False)
+
+    @classmethod
+    def delivery(cls, client: object, namespace: str, task: Mapping[str, Any], *,
+                 owner: str, task_queue: str, observed: bool) -> QueryTaskAuthority:
+        task_id, attempt = task.get("query_task_id"), task.get("query_task_attempt")
+        if (not isinstance(owner, str) or not owner or not isinstance(task_queue, str) or not task_queue
+                or not isinstance(task_id, str) or not task_id or type(attempt) is not int or attempt < 1
+                or task.get("lease_owner") != owner or task.get("task_queue") != task_queue
+                or task.get("namespace", namespace) != namespace):
+            raise QueryTaskAuthorityLost(str(task_id or ""), "invalid_query_claim")
+        try:
+            lease = _timestamp(task.get("lease_expires_at"), canonical=observed)
+            timing = {"lease_expires_at": task["lease_expires_at"]}
+            bounds = [lease]
+            if observed or "query_admitted_at" in task or "query_deadline_at" in task:
+                admitted = _timestamp(task.get("query_admitted_at"), canonical=True)
+                deadline = _timestamp(task.get("query_deadline_at"), canonical=True)
+                if deadline < admitted or (observed and lease > deadline):
+                    raise ValueError("inconsistent original query timing")
+                timing.update({"query_admitted_at": task["query_admitted_at"],
+                               "query_deadline_at": task["query_deadline_at"]})
+                bounds.append(deadline)
+            # Legacy compatibility has no non-renewing observation. Its local
+            # clock check cannot establish an undisclosed caller deadline.
+            reference = (_timestamp(task.get("server_time"), canonical=True) if observed
+                         else datetime.now(timezone.utc))
+            until = time.monotonic() + (min(bounds) - reference).total_seconds()
+        except (ValueError, TypeError, OverflowError) as error:
+            raise QueryTaskAuthorityLost(task_id, "invalid_query_timing") from error
+        result = cls(client, namespace, task_id, attempt, owner, observed, timing, until,
+                     {key: task.get(key) for key in ("workflow_id", "run_id", "task_queue")})
+        result.remaining()
+        return result
+
+    def matches(self, client: object, namespace: str, task: Mapping[str, Any], owner: str) -> bool:
+        return (self.client is client and self.namespace == namespace and self.owner == owner
+                and task.get("query_task_id") == self.task_id
+                and type(task.get("query_task_attempt")) is int and task.get("query_task_attempt") == self.attempt
+                and task.get("lease_owner") == owner
+                and all(task.get(key) == value for key, value in self.identity.items())
+                and all(task.get(key) == value for key, value in self.timing.items()))
+
+    def remaining(self) -> float:
+        remaining = self.until - time.monotonic()
+        if remaining <= 0:
+            raise QueryTaskAuthorityLost(self.task_id, "query_execution_budget_expired")
+        return remaining
+
+    async def confirm(self) -> None:
+        remaining = self.remaining()
+        if not self.observed:
+            return
+        started = time.monotonic()
+        try:
+            if self.observer is None:
+                raise ValueError("query authority observer is unavailable")
+            reply = await asyncio.wait_for(self.observer(), timeout=min(5.0, remaining))
+            if (not isinstance(reply, Mapping) or reply.get("query_task_id") != self.task_id
+                    or type(reply.get("query_task_attempt")) is not int
+                    or reply.get("query_task_attempt") != self.attempt or reply.get("lease_owner") != self.owner
+                    or any(reply.get(key) != value for key, value in self.timing.items())):
+                raise ValueError("query authority observation changed its original claim")
+            if (reply.get("can_continue") is not True or reply.get("query_task_status") != "leased"
+                    or reply.get("reason") is not None):
+                raise QueryTaskAuthorityLost(self.task_id, str(reply.get("reason") or "query_claim_unavailable"))
+            server_time = _timestamp(reply.get("server_time"), canonical=True)
+            bound = min(_timestamp(value, canonical=True) for key, value in self.timing.items()
+                        if key != "query_admitted_at")
+            # Charging the entire observation round trip is conservative. A
+            # later observation may shorten this bound but can never reset it.
+            self.until = min(self.until, started + (bound - server_time).total_seconds())
+        except QueryTaskAuthorityLost:
+            raise
+        except Exception as error:
+            raise QueryTaskAuthorityLost(self.task_id, "query_authority_unavailable") from error
+        self.remaining()
+
+    async def run(self, operation: Callable[[], Awaitable[_T]], *, observe: bool = True,
+                  confirm_after: bool = True) -> _T:
+        await self.confirm()
+        execution = asyncio.ensure_future(operation())
+
+        async def watch() -> None:
+            while True:
+                await asyncio.sleep(min(_OBSERVATION_INTERVAL_SECONDS, self.remaining()))
+                await self.confirm()
+
+        observation = asyncio.create_task(watch()) if observe and self.observed else None
+        pending: set[asyncio.Future[Any]] = {execution}
+        if observation is not None:
+            pending.add(observation)
+        try:
+            done, _ = await asyncio.wait(pending, timeout=self.remaining(), return_when=asyncio.FIRST_COMPLETED)
+            if not done:
+                raise QueryTaskAuthorityLost(self.task_id, "query_execution_budget_expired")
+            if observation is not None and observation in done:
+                await observation
+                raise QueryTaskAuthorityLost(self.task_id, "query_authority_unavailable")
+            result = await execution
+            if confirm_after:
+                await self.confirm()
+            return result
+        finally:
+            for future in pending:
+                future.cancel()
+            await asyncio.gather(*pending, return_exceptions=True)
+
+
+# Async task creation copies this context. Each dispatched query keeps the
+# loading budget without adding private fields to the public wire dictionary
+# or retaining an unbounded task-ID registry on Client/Worker.
+current_query_authority: ContextVar[QueryTaskAuthority | None] = ContextVar("query_task_authority", default=None)

src/durable_workflow/client.py

@@ -45,6 +45,7 @@
     scope_timestamp,
 )
 from ._cooperative_cancellation import CancellationDelivery, CancellationRequest
+from ._query_task_authority import QueryTaskAuthority, current_query_authority
 from .errors import (
     ExternalPayloadIntegrityMismatch,
     ExternalPayloadOversized,
@@ -1809,6 +1810,9 @@ async def _request(
         if discovery and path != "/cluster/info":
             raise ValueError("cluster discovery credentials can only authorize /cluster/info")
 
+        query_poll = worker and path == "/worker/query-tasks/poll"
+        if query_poll:
+            current_query_authority.set(None)
         request_json = json
         if (
             not discovery
@@ -1872,21 +1876,29 @@ async def _do_request() -> httpx.Response:
                 outcome = "ok"
                 return None
             result = resp.json()
+            query_authority = None
+            if query_poll and isinstance(result, dict) and isinstance(result.get("task"), dict):
+                query_authority = await self._query_delivery_authority(result["task"], request_json)
+                current_query_authority.set(query_authority)
             if not discovery and _contains_runtime_payload_envelope(result):
-                transport = await self._runtime_external_payload_transport()
+                transport = (await query_authority.run(self._runtime_external_payload_transport)
+                             if query_authority is not None else await self._runtime_external_payload_transport())
                 if transport is None:
                     raise ExternalPayloadUnsupported(
                         "runtime returned an external payload reference without advertising "
                         "the authenticated namespace transport"
                     )
-                result = await self._resolve_workflow_claim_payloads(
-                    result,
-                    worker=worker,
-                    transport=transport,
-                    path=path,
-                    request_json=request_json,
-                )
+                if query_authority is not None:
+                    result = await query_authority.run(lambda: self._resolve_runtime_payloads(
+                        result, worker=worker, transport=transport, resolved={},
+                    ))
+                else:
+                    result = await self._resolve_workflow_claim_payloads(
+                        result, worker=worker, transport=transport, path=path, request_json=request_json,
+                    )
             result = self._hydrate_result_envelopes(result)
+            if query_authority is not None:
+                await query_authority.confirm()
             outcome = "ok"
             return result
         except Exception as exc:
@@ -2332,6 +2344,7 @@ async def _resolve_runtime_payloads(
         *,
         worker: bool,
         transport: _RuntimeExternalPayloadTransport,
+        resolved: dict[tuple[str, str, int, str], str] | None = None,
     ) -> Any:
         if isinstance(value, dict):
             if "external_payload" in value:
@@ -2347,6 +2360,9 @@ async def _resolve_runtime_payloads(
                         "runtime external payload envelope codec does not match its reference",
                         reference_id=reference.reference_id,
                     )
+                identity = (reference.reference_id, reference.codec, reference.size_bytes, reference.sha256)
+                if resolved is not None and identity in resolved:
+                    return {"codec": reference.codec, "blob": resolved[identity]}
                 data = await self._fetch_runtime_payload(
                     reference,
                     worker=worker,
@@ -2359,12 +2375,15 @@ async def _resolve_runtime_payloads(
                         "runtime external payload bytes are not a UTF-8 payload blob",
                         reference_id=reference.reference_id,
                     ) from exc
+                if resolved is not None:
+                    resolved[identity] = blob
                 return {"codec": reference.codec, "blob": blob}
             return {
                 key: await self._resolve_runtime_payloads(
                     item,
                     worker=worker,
                     transport=transport,
+                    resolved=resolved,
                 )
                 for key, item in value.items()
             }
@@ -2374,6 +2393,7 @@ async def _resolve_runtime_payloads(
                     item,
                     worker=worker,
                     transport=transport,
+                    resolved=resolved,
                 )
                 for item in value
             ]
@@ -5625,6 +5645,48 @@ async def workflow_task_history(
             "POST", f"/worker/workflow-tasks/{task_id}/history", worker=True, json=body
         )
 
+    async def _query_delivery_authority(self, task: dict[str, Any], request: Any) -> QueryTaskAuthority:
+        # Worker registration establishes discovery before it polls. Advanced
+        # callers without discovery retain only the exposed local lease checks.
+        info = self._cluster_info or {}
+        protocol = info.get("worker_protocol")
+        capabilities = protocol.get("server_capabilities") if isinstance(protocol, dict) else None
+        observed = (isinstance(capabilities, dict) and capabilities.get("query_task_authority") is True
+                    and _supports_cooperative_cancellation_protocol(_protocol_version_from_env(
+                        "DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION", PROTOCOL_VERSION)))
+        request = request if isinstance(request, dict) else {}
+        authority = QueryTaskAuthority.delivery(self, self.namespace, task, owner=request.get("worker_id", ""),
+                                               task_queue=request.get("task_queue", ""), observed=observed)
+        if observed:
+            authority.observer = lambda: self.query_task_status(query_task_id=authority.task_id,
+                lease_owner=authority.owner, query_task_attempt=authority.attempt)
+        await authority.confirm()
+        return authority
+
+    async def query_task_status(self, *, query_task_id: str, lease_owner: str,
+                                query_task_attempt: int) -> Any:
+        """Observe a protocol 1.20 query claim without renewing it.
+
+        Requires ``query_task_authority=true`` discovery. A positive observation
+        does not reserve authority for later execution or completion.
+        """
+        if not _supports_cooperative_cancellation_protocol(_protocol_version_from_env(
+            "DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION", PROTOCOL_VERSION,
+        )):
+            raise ValueError("query task observation requires worker protocol 1.20")
+        capability_path = "worker_protocol.server_capabilities.query_task_authority"
+        info = await self._runtime_discovery(operation="Client.query_task_status", required_path=capability_path)
+        protocol = info.get("worker_protocol")
+        capabilities = protocol.get("server_capabilities") if isinstance(protocol, dict) else None
+        supported = capabilities.get("query_task_authority") if isinstance(capabilities, dict) else None
+        if supported is not True:
+            error_type = RuntimeCapabilityUnsupported if supported is False else RuntimeDiscoveryUnavailable
+            raise error_type("Client.query_task_status", capability_path,
+                             "Query task observation requires Server query_task_authority=true discovery.")
+        query = urlencode({"lease_owner": lease_owner, "query_task_attempt": query_task_attempt})
+        return await self._request("GET", f"/worker/query-tasks/{quote(query_task_id, safe='')}/status?{query}",
+                                   worker=True, timeout=5.0)
+
     async def poll_query_task(
         self,
         *,
@@ -5640,6 +5702,10 @@ async def poll_query_task(
         must route a control-plane query to a non-PHP workflow runtime.
         ``timeout`` controls the server long-poll window; the HTTP request
         uses a small grace margin.
+
+        Worker registration establishes capability discovery. Custom workers
+        should call ``get_cluster_info()`` first to enable advertised query
+        status checks; without discovery only supplied local lease bounds apply.
         """
         body: dict[str, Any] = {
             "worker_id": worker_id,