Add connection hardening with retry policy (#169)

Implement automatic retries with exponential backoff and jitter for
transient server errors, improving SDK reliability for production use.

New module: src/durable_workflow/retry_policy.py
- RetryPolicy class with configurable retry behavior
- Retries connection errors, timeouts, 5xx server errors, 429 rate limit
- Does not retry 4xx client errors (except 429)
- Exponential backoff with jitter (default: 0.1s initial, 5s max, 2x multiplier)
- Default: 3 attempts per request

Client changes:
- Added retry_policy parameter to Client.__init__ (optional, defaults to RetryPolicy())
- Updated _request() to wrap httpx calls with retry logic
- Updated health() to use _request() for consistency (gets retries too)

Tests:
- 15 new unit tests for RetryPolicy covering all retry scenarios
- Updated test_sync.py mock (health now uses _request not _http.get)
- All 219 tests pass, mypy --strict clean, ruff clean

Example usage:

Closes a Phase 4 deliverable (connection hardening).

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: durable-workflow-ops

src/durable_workflow/client.py

@@ -13,6 +13,7 @@
     WorkflowTerminated,
     _raise_for_status,
 )
+from .retry_policy import RetryPolicy
 
 PROTOCOL_VERSION = "1.0"
 CONTROL_PLANE_VERSION = "2"
@@ -233,10 +234,12 @@ def __init__(
         token: str | None = None,
         namespace: str = "default",
         timeout: float = 60.0,
+        retry_policy: RetryPolicy | None = None,
     ) -> None:
         self.base_url = base_url.rstrip("/")
         self.token = token
         self.namespace = namespace
+        self.retry_policy = retry_policy or RetryPolicy()
         self._http = httpx.AsyncClient(base_url=self.base_url, timeout=timeout)
 
     async def aclose(self) -> None:
@@ -269,19 +272,29 @@ async def _request(
         timeout: float | None = None,
         context: str = "",
     ) -> Any:
-        resp = await self._http.request(
-            method,
-            f"/api{path}",
-            headers=self._headers(worker=worker),
-            json=json,
-            timeout=timeout,
-        )
-        if resp.status_code >= 400:
+        async def _do_request() -> httpx.Response:
+            resp = await self._http.request(
+                method,
+                f"/api{path}",
+                headers=self._headers(worker=worker),
+                json=json,
+                timeout=timeout,
+            )
+            # Raise HTTPStatusError for 4xx/5xx so retry policy can catch it
+            resp.raise_for_status()
+            return resp
+
+        try:
+            resp = await self.retry_policy.execute(_do_request)
+        except httpx.HTTPStatusError as exc:
+            # Convert to our custom exception types
             try:
-                body = resp.json()
+                body = exc.response.json()
             except ValueError:
-                body = resp.text
-            _raise_for_status(resp.status_code, body, context=context)
+                body = exc.response.text
+            _raise_for_status(exc.response.status_code, body, context=context)
+            raise  # unreachable, but keeps type checker happy
+
         if resp.status_code == 204 or not resp.content:
             return None
         return resp.json()
@@ -293,9 +306,8 @@ def get_workflow_handle(
 
     # ── Health ─────────────────────────────────────────────────────────
     async def health(self) -> dict[str, Any]:
-        resp = await self._http.get("/api/health")
-        resp.raise_for_status()
-        result: dict[str, Any] = resp.json()
+        result = await self._request("GET", "/health")
+        assert isinstance(result, dict)
         return result
 
     # ── Workflows ──────────────────────────────────────────────────────

src/durable_workflow/retry_policy.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+import asyncio
+import random
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from typing import TypeVar
+
+import httpx
+
+T = TypeVar("T")
+
+
+@dataclass
+class RetryPolicy:
+    """
+    Retry policy for transient server errors.
+
+    Retries requests that fail with transient errors (connection errors,
+    timeouts, 5xx server errors, 429 rate limit). Does not retry client
+    errors (4xx except 429).
+
+    Uses exponential backoff with jitter to avoid thundering herd.
+    """
+
+    max_attempts: int = 3
+    initial_backoff_seconds: float = 0.1
+    max_backoff_seconds: float = 5.0
+    backoff_multiplier: float = 2.0
+    jitter: bool = True
+
+    def should_retry(self, exc: Exception, attempt: int) -> bool:
+        """Check if the error is retryable and we haven't exceeded max attempts."""
+        if attempt >= self.max_attempts:
+            return False
+
+        # Retry connection errors and timeouts
+        if isinstance(exc, (httpx.ConnectError, httpx.TimeoutException, httpx.NetworkError)):
+            return True
+
+        # Retry 5xx server errors and 429 rate limit
+        if isinstance(exc, httpx.HTTPStatusError):
+            return exc.response.status_code >= 500 or exc.response.status_code == 429
+
+        return False
+
+    def backoff_seconds(self, attempt: int) -> float:
+        """Calculate backoff duration for the given attempt number (0-indexed)."""
+        backoff = min(
+            self.initial_backoff_seconds * (self.backoff_multiplier**attempt),
+            self.max_backoff_seconds,
+        )
+        if self.jitter:
+            # Add ±25% jitter
+            backoff *= random.uniform(0.75, 1.25)
+        return backoff
+
+    async def execute(self, fn: Callable[[], Awaitable[T]]) -> T:
+        """
+        Execute the given async function with retries.
+
+        Raises the last exception if all retries are exhausted.
+        """
+        attempt = 0
+        last_exc: Exception | None = None
+
+        while attempt < self.max_attempts:
+            try:
+                result = await fn()
+                return result
+            except Exception as exc:
+                last_exc = exc
+                if not self.should_retry(exc, attempt):
+                    raise
+
+                if attempt + 1 < self.max_attempts:
+                    backoff = self.backoff_seconds(attempt)
+                    await asyncio.sleep(backoff)
+
+                attempt += 1
+
+        # All retries exhausted
+        if last_exc:
+            raise last_exc
+        raise RuntimeError("retry loop exhausted with no exception")

tests/test_retry_policy.py

@@ -0,0 +1,151 @@
+from __future__ import annotations
+
+import httpx
+import pytest
+
+from durable_workflow.retry_policy import RetryPolicy
+
+
+class TestRetryPolicy:
+    def test_should_retry_connection_error(self) -> None:
+        policy = RetryPolicy(max_attempts=3)
+        exc = httpx.ConnectError("connection failed")
+        assert policy.should_retry(exc, attempt=0) is True
+        assert policy.should_retry(exc, attempt=1) is True
+        assert policy.should_retry(exc, attempt=2) is True
+        assert policy.should_retry(exc, attempt=3) is False  # max_attempts reached
+
+    def test_should_retry_timeout(self) -> None:
+        policy = RetryPolicy(max_attempts=3)
+        exc = httpx.TimeoutException("timeout")
+        assert policy.should_retry(exc, attempt=0) is True
+
+    def test_should_retry_network_error(self) -> None:
+        policy = RetryPolicy(max_attempts=3)
+        exc = httpx.NetworkError("network error")
+        assert policy.should_retry(exc, attempt=0) is True
+
+    def test_should_retry_5xx_server_error(self) -> None:
+        policy = RetryPolicy(max_attempts=3)
+        response = httpx.Response(status_code=500, request=httpx.Request("GET", "http://test"))
+        exc = httpx.HTTPStatusError("server error", request=response.request, response=response)
+        assert policy.should_retry(exc, attempt=0) is True
+
+    def test_should_retry_429_rate_limit(self) -> None:
+        policy = RetryPolicy(max_attempts=3)
+        response = httpx.Response(status_code=429, request=httpx.Request("GET", "http://test"))
+        exc = httpx.HTTPStatusError("rate limited", request=response.request, response=response)
+        assert policy.should_retry(exc, attempt=0) is True
+
+    def test_should_not_retry_4xx_client_error(self) -> None:
+        policy = RetryPolicy(max_attempts=3)
+        response = httpx.Response(status_code=404, request=httpx.Request("GET", "http://test"))
+        exc = httpx.HTTPStatusError("not found", request=response.request, response=response)
+        assert policy.should_retry(exc, attempt=0) is False
+
+    def test_should_not_retry_400_bad_request(self) -> None:
+        policy = RetryPolicy(max_attempts=3)
+        response = httpx.Response(status_code=400, request=httpx.Request("GET", "http://test"))
+        exc = httpx.HTTPStatusError("bad request", request=response.request, response=response)
+        assert policy.should_retry(exc, attempt=0) is False
+
+    def test_should_not_retry_other_exceptions(self) -> None:
+        policy = RetryPolicy(max_attempts=3)
+        exc = ValueError("not a network error")
+        assert policy.should_retry(exc, attempt=0) is False
+
+    def test_backoff_calculation(self) -> None:
+        policy = RetryPolicy(
+            initial_backoff_seconds=0.1,
+            max_backoff_seconds=5.0,
+            backoff_multiplier=2.0,
+            jitter=False,
+        )
+        assert policy.backoff_seconds(0) == 0.1
+        assert policy.backoff_seconds(1) == 0.2
+        assert policy.backoff_seconds(2) == 0.4
+        assert policy.backoff_seconds(10) == 5.0  # capped at max
+
+    def test_backoff_with_jitter(self) -> None:
+        policy = RetryPolicy(initial_backoff_seconds=1.0, jitter=True)
+        # Jitter should give us ±25%
+        backoff = policy.backoff_seconds(0)
+        assert 0.75 <= backoff <= 1.25
+
+    @pytest.mark.asyncio
+    async def test_execute_success_on_first_try(self) -> None:
+        policy = RetryPolicy(max_attempts=3)
+        call_count = 0
+
+        async def fn() -> str:
+            nonlocal call_count
+            call_count += 1
+            return "success"
+
+        result = await policy.execute(fn)
+        assert result == "success"
+        assert call_count == 1
+
+    @pytest.mark.asyncio
+    async def test_execute_success_after_retry(self) -> None:
+        policy = RetryPolicy(max_attempts=3, initial_backoff_seconds=0.01, jitter=False)
+        call_count = 0
+
+        async def fn() -> str:
+            nonlocal call_count
+            call_count += 1
+            if call_count < 3:
+                raise httpx.ConnectError("connection failed")
+            return "success"
+
+        result = await policy.execute(fn)
+        assert result == "success"
+        assert call_count == 3
+
+    @pytest.mark.asyncio
+    async def test_execute_exhausted_retries(self) -> None:
+        policy = RetryPolicy(max_attempts=3, initial_backoff_seconds=0.01, jitter=False)
+        call_count = 0
+
+        async def fn() -> str:
+            nonlocal call_count
+            call_count += 1
+            raise httpx.ConnectError("connection failed")
+
+        with pytest.raises(httpx.ConnectError):
+            await policy.execute(fn)
+
+        assert call_count == 3
+
+    @pytest.mark.asyncio
+    async def test_execute_non_retryable_error(self) -> None:
+        policy = RetryPolicy(max_attempts=3)
+        call_count = 0
+
+        async def fn() -> str:
+            nonlocal call_count
+            call_count += 1
+            response = httpx.Response(status_code=400, request=httpx.Request("GET", "http://test"))
+            raise httpx.HTTPStatusError("bad request", request=response.request, response=response)
+
+        with pytest.raises(httpx.HTTPStatusError):
+            await policy.execute(fn)
+
+        assert call_count == 1  # Should not retry 400 errors
+
+    @pytest.mark.asyncio
+    async def test_execute_retries_500_but_not_400(self) -> None:
+        policy = RetryPolicy(max_attempts=3, initial_backoff_seconds=0.01, jitter=False)
+        call_count = 0
+
+        async def fn() -> str:
+            nonlocal call_count
+            call_count += 1
+            if call_count == 1:
+                response = httpx.Response(status_code=500, request=httpx.Request("GET", "http://test"))
+                raise httpx.HTTPStatusError("server error", request=response.request, response=response)
+            return "success"
+
+        result = await policy.execute(fn)
+        assert result == "success"
+        assert call_count == 2  # First attempt 500, second attempt success

tests/test_sync.py

@@ -23,7 +23,7 @@ class TestSyncClientHealth:
     def test_health(self) -> None:
         client = Client("http://localhost:8080")
         resp = _mock_response(200, {"status": "ok"})
-        with patch.object(client._async._http, "get", new_callable=AsyncMock, return_value=resp):
+        with patch.object(client._async._http, "request", new_callable=AsyncMock, return_value=resp):
             result = client.health()
             assert result["status"] == "ok"