diff --git a/README.md b/README.md index 3812aff..48f1e90 100644 --- a/README.md +++ b/README.md @@ -99,6 +99,7 @@ for the complete cross-SDK contract. - [Python SDK guide](https://durable-workflow.com/docs/2.0/polyglot/python/) - [Complete SDK reference](docs/sdk-reference.md) - [Runnable examples](examples/) +- [Checkpointed batch source example](examples/CHECKPOINTED_BATCH.md) (unreleased accessor; synthetic provider) - [Separate workflow and activity worker processes](docs/guides/separate-workers.md) - [Symmetric SDK playground](https://github.com/durable-workflow/sample-app#symmetric-sdk-playground) diff --git a/examples/CHECKPOINTED_BATCH.md b/examples/CHECKPOINTED_BATCH.md new file mode 100644 index 0000000..88f7e1b --- /dev/null +++ b/examples/CHECKPOINTED_BATCH.md @@ -0,0 +1,64 @@ +# Resume a bounded batch + +This source example uses the unreleased acknowledged-checkpoint API from +[SDK333](https://github.com/durable-workflow/sdk-python/pull/333). It requires a +Server advertising `activity_heartbeat_resume` v1. Its source tests do not prove +published-package compatibility, real process loss or Server persistence. + +On Linux, from a source checkout with its development environment, point it at +an isolated development Server. Supply a **new** directory for each invocation. +The synthetic provider receipts remain there for inspection; existing +directories are refused. + +```bash +export SERVER_URL=http://127.0.0.1:8080 +export WORKFLOW_TOKEN=your-development-token +python -m examples.checkpointed_batch ./receipts-before before-commit +python -m examples.checkpointed_batch ./receipts-commit after-commit +python -m examples.checkpointed_batch ./receipts-checkpoint after-checkpoint +``` + +The importable `BatchActivity.execute` handler carries only serializable provider +configuration, so it also passes the SDK's cooperative callback preflight. +The activity processes three items, with a maximum of sixteen bounded items. +Its two attempts, one-second retry delay, ten-second heartbeat timeout, +thirty-second start-to-close timeout and sixty-second schedule-to-close timeout +stay fixed. Recovery does not extend those budgets. The example's worker wait +is bounded separately. + +| First-attempt failure | Replacement action | +| --- | --- | +| Before provider commit | Commit item zero, then acknowledge progress | +| After commit, before acknowledgement | Reconcile the stable provider key, then acknowledge progress | +| After acknowledgement | Resume at the received next-item cursor | +| Acknowledgement response lost | The next claim may contain the checkpoint or none; reconcile any repeated provider operation | + +`last_heartbeat_details(result_type=Progress)` reads the immutable received +snapshot. Unsupported claims refuse instead of inventing empty state. A +supported empty checkpoint starts from zero. The typed cursor binds the full +business input and must be within the item range. + +Provider receipts are separate durable state. A key uses the business ID and +item index, never an attempt ID. Each receipt also binds the entire batch +fingerprint, so appending, truncating or reordering input under a repeated +business ID refuses even without a checkpoint. Revisited keys require the exact +receipt. Partial or conflicting receipts are preserved and require +reconciliation. File and directory entries are synchronized before commit or +reconciliation returns. This is a private synthetic provider, not an actual +payment or other external service. + +The ordering is provider commit/reconcile, then `await ctx.heartbeat(...)`, then +continue. A failed heartbeat stops the attempt; an uncertain response does not +mean the checkpoint failed to commit. The SDK snapshot does not advance in +place. A checkpoint does not atomically commit an external business effect. +Use the real provider's idempotency, operation lookup and fencing contract, +retain the business ID across restarts of that operation, and choose a new ID +for a new business operation. The example does not promise exactly-once effects. + +The source tests use synthetic files and acknowledgement callbacks without a +Server. They cover all three failure boundaries, both lost-response outcomes, +changed input without a checkpoint, malformed cursors and partial receipts: + +```bash +pytest -q tests/test_checkpointed_batch_example.py +``` diff --git a/examples/checkpointed_batch.py b/examples/checkpointed_batch.py new file mode 100644 index 0000000..9978279 --- /dev/null +++ b/examples/checkpointed_batch.py @@ -0,0 +1,168 @@ +"""Bounded source example using the unreleased checkpoint API. See CHECKPOINTED_BATCH.md.""" +from __future__ import annotations + +import argparse +import asyncio +import hashlib +import json +import os +from collections.abc import Awaitable, Callable, Generator +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any +from uuid import uuid4 + +from durable_workflow import ActivityRetryPolicy, Client, Worker, activity, workflow +from durable_workflow.errors import ActivityHeartbeatResumeUnavailable, NonRetryableError + +FAILURES = ("before-commit", "after-commit", "after-checkpoint") +ACTIVITY = "examples.checkpointed-batch" + + +@dataclass(frozen=True) +class Batch: + business_id: str + items: list[str] + failure: str + + def snapshot(self) -> tuple[str, tuple[str, ...], str]: + if not isinstance(self.items, list) or not 1 <= len(self.items) <= 16 or self.failure not in FAILURES: + raise NonRetryableError("use a business ID and one to sixteen bounded items") + try: + valid = all(isinstance(value, str) and 1 <= len(value.encode()) <= 128 + for value in [self.business_id, *self.items]) + except UnicodeEncodeError: + valid = False + if not valid: + raise NonRetryableError("use a business ID and one to sixteen bounded items") + items = tuple(self.items) + encoded = json.dumps([self.business_id, items], ensure_ascii=False, separators=(",", ":")).encode() + return self.business_id, items, hashlib.sha256(encoded).hexdigest() + + +@dataclass(frozen=True) +class Progress: + batch_sha256: str + next_item: int + + +@dataclass(frozen=True) +class ReceiptStore: + """Synthetic durable provider state, independent of Server checkpoints.""" + + root: Path + + @classmethod + def create(cls, root: Path) -> ReceiptStore: + root.mkdir(mode=0o700) + return cls(root) + + def sync_directory(self) -> None: + descriptor = os.open(self.root, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) + + def commit_or_reconcile(self, business_id: str, fingerprint: str, index: int, item: str) -> bool: + key = json.dumps([business_id, index], ensure_ascii=False, separators=(",", ":")).encode() + path = self.root / (hashlib.sha256(key).hexdigest() + ".json") + receipt = json.dumps({"key": [business_id, index], "batch_sha256": fingerprint, "item": item}, + ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode() + try: + with path.open("xb") as stream: + stream.write(receipt) + stream.flush() + os.fsync(stream.fileno()) + created = True + except FileExistsError: + with path.open("rb") as stream: + if stream.read(4097) != receipt: + raise NonRetryableError( + "provider receipt conflicts or is partial; reconcile before retrying", + ) from None + os.fsync(stream.fileno()) + created = False + self.sync_directory() + return created + + +async def process_batch( + store: ReceiptStore, batch: Batch, received: Progress | None, attempt: int, + acknowledge: Callable[[Progress], Awaitable[None]], +) -> dict[str, Any]: + business_id, items, fingerprint = batch.snapshot() + if type(attempt) is not int or attempt < 1: + raise NonRetryableError("the activity attempt must be positive") + if received is not None and ( + received.batch_sha256 != fingerprint or type(received.next_item) is not int + or not 0 <= received.next_item <= len(items) + ): + raise NonRetryableError("checkpoint belongs to different input or is out of range") + start = received.next_item if received is not None else 0 + def fail(point: str, index: int) -> None: + if attempt == 1 and index == 0 and batch.failure == point: + raise RuntimeError("planned first-attempt failure at " + point) + + for index in range(start, len(items)): + fail("before-commit", index) + await asyncio.to_thread(store.commit_or_reconcile, business_id, fingerprint, index, items[index]) + fail("after-commit", index) + await acknowledge(Progress(fingerprint, index + 1)) + fail("after-checkpoint", index) + return {"business_id": business_id, "completed": len(items)} + + +@workflow.defn(name=ACTIVITY) +class BatchWorkflow: + def run(self, ctx: workflow.WorkflowContext, batch: dict[str, Any]) -> Generator[Any, Any, Any]: + return (yield ctx.schedule_activity( + ACTIVITY, [batch], retry_policy=ActivityRetryPolicy(max_attempts=2, backoff_seconds=[1]), + heartbeat_timeout=10, start_to_close_timeout=30, schedule_to_close_timeout=60, + )) + + +@dataclass(frozen=True) +class BatchActivity: + """Importable handler carrying only the provider's serializable configuration.""" + + store: ReceiptStore + + @activity.defn(name=ACTIVITY) + async def execute(self, body: dict[str, Any]) -> dict[str, Any]: + ctx = activity.context() + try: + received = ctx.last_heartbeat_details(result_type=Progress) + except ActivityHeartbeatResumeUnavailable: + raise NonRetryableError("this example requires activity_heartbeat_resume v1") from None + + async def acknowledge(progress: Progress) -> None: + await ctx.heartbeat(asdict(progress)) + + return await process_batch(self.store, Batch(**body), received, ctx.info.attempt_number, acknowledge) + + +async def run(root: Path, failure: str) -> None: + store = await asyncio.to_thread(ReceiptStore.create, root) + queue = "checkpointed-batch-" + uuid4().hex + async with Client(os.environ.get("SERVER_URL", "http://127.0.0.1:8080"), + token=os.environ.get("WORKFLOW_TOKEN"), namespace="default", timeout=10) as client: + worker = Worker(client, task_queue=queue, workflows=[BatchWorkflow], activities=[BatchActivity(store).execute]) + handle = await client.start_workflow( + workflow_type=ACTIVITY, task_queue=queue, workflow_id=queue, + input=[asdict(Batch("demo-" + uuid4().hex, ["first", "second", "third"], failure))], + ) + await worker.run_until(workflow_id=handle.workflow_id, timeout=70) + print(json.dumps(await handle.result(timeout=3))) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("receipts", type=Path, help="a new private synthetic-provider directory") + parser.add_argument("failure", choices=FAILURES) + args = parser.parse_args() + asyncio.run(run(args.receipts, args.failure)) + + +if __name__ == "__main__": + main() diff --git a/tests/test_checkpointed_batch_example.py b/tests/test_checkpointed_batch_example.py new file mode 100644 index 0000000..bc5c57b --- /dev/null +++ b/tests/test_checkpointed_batch_example.py @@ -0,0 +1,174 @@ +"""Synthetic provider/acknowledgement boundaries; no Server or worker invocation.""" +from __future__ import annotations + +import pickle +from dataclasses import asdict, replace +from pathlib import Path +from typing import Any +from unittest.mock import AsyncMock + +import pytest +from examples.checkpointed_batch import ACTIVITY, FAILURES, Batch, BatchActivity, Progress, ReceiptStore, process_batch + +from durable_workflow import ActivityInfo, Client, Worker +from durable_workflow._activity_process import CallbackInvocation +from durable_workflow.errors import NonRetryableError + + +def batch(failure: str = "after-commit") -> Batch: + return Batch("business-batch", ["first", "second", "third"], failure) + + +def inventory(store: ReceiptStore) -> list[tuple[str, bytes]]: + return sorted((path.name, path.read_bytes()) for path in store.root.iterdir()) + + +@pytest.mark.parametrize("point", FAILURES) +async def test_commit_boundaries_resume_only_acknowledged_items(tmp_path: Path, point: str) -> None: + store = ReceiptStore.create(tmp_path / "provider") + request = batch(point) + persisted = None + + async def acknowledge(progress: Progress) -> None: + nonlocal persisted + persisted = progress + + with pytest.raises(RuntimeError, match="planned first-attempt failure"): + await process_batch(store, request, None, 1, acknowledge) + assert (persisted.next_item if persisted else None) == (1 if point == "after-checkpoint" else None) + assert len(inventory(store)) == (0 if point == "before-commit" else 1) + resumed = [] + + async def acknowledge_replacement(progress: Progress) -> None: + resumed.append(progress.next_item) + + assert await process_batch(store, request, persisted, 2, acknowledge_replacement) == { + "business_id": "business-batch", "completed": 3, + } + assert resumed == ([2, 3] if point == "after-checkpoint" else [1, 2, 3]) + before = inventory(store) + business_id, items, fingerprint = request.snapshot() + for index, item in enumerate(items): + assert store.commit_or_reconcile(business_id, fingerprint, index, item) is False + assert inventory(store) == before + + +@pytest.mark.parametrize("committed", [False, True]) +async def test_lost_ack_response_handles_both_commit_outcomes(tmp_path: Path, committed: bool) -> None: + store = ReceiptStore.create(tmp_path / "provider") + request = batch("after-checkpoint") + persisted = None + + async def lost_response(progress: Progress) -> None: + nonlocal persisted + if committed: + persisted = progress + raise TimeoutError("synthetic acknowledgement response lost") + + with pytest.raises(TimeoutError): + await process_batch(store, request, None, 1, lost_response) + assert len(inventory(store)) == 1 + next_items = [] + + async def acknowledge(progress: Progress) -> None: + next_items.append(progress.next_item) + + assert (await process_batch(store, request, persisted, 2, acknowledge))["completed"] == 3 + assert next_items == ([2, 3] if committed else [1, 2, 3]) + assert len(inventory(store)) == 3 + + +@pytest.mark.parametrize("items", [ + ["first", "second", "third", "fourth"], ["first", "second"], ["second", "first", "third"], +]) +async def test_changed_input_without_checkpoint_preserves_provider_state(tmp_path: Path, items: list[str]) -> None: + store = ReceiptStore.create(tmp_path / "provider") + + async def no_op(_progress: Progress) -> None: + pass + + request = batch() + await process_batch(store, request, None, 2, no_op) + before = inventory(store) + + async def forbidden(_progress: Progress) -> None: + raise AssertionError("conflicting business input was acknowledged") + + with pytest.raises(NonRetryableError, match="provider receipt conflicts"): + await process_batch(store, replace(request, items=items), None, 2, forbidden) + assert inventory(store) == before + + +@pytest.mark.parametrize("cursor", [Progress("foreign", 1), Progress("same", -1), + Progress("same", 4), Progress("same", True)]) +async def test_bad_checkpoint_refuses_before_provider_access(tmp_path: Path, cursor: Progress) -> None: + store = ReceiptStore.create(tmp_path / "provider") + request = batch() + if cursor.batch_sha256 == "same": + cursor = replace(cursor, batch_sha256=request.snapshot()[2]) + + async def forbidden(_progress: Progress) -> None: + raise AssertionError("invalid checkpoint was acknowledged") + + with pytest.raises(NonRetryableError, match="checkpoint belongs"): + await process_batch(store, request, cursor, 2, forbidden) + assert inventory(store) == [] + + +@pytest.mark.parametrize("items", [[], ["item"] * 17, [""], ["x" * 129], ["\ud800"], [False], "not-a-list"]) +async def test_invalid_batch_refuses_before_provider_access(tmp_path: Path, items: Any) -> None: + store = ReceiptStore.create(tmp_path / "provider") + + async def forbidden(_progress: Progress) -> None: + raise AssertionError("invalid batch was acknowledged") + + with pytest.raises(NonRetryableError): + await process_batch(store, Batch("business", items, "after-commit"), None, 2, forbidden) + assert inventory(store) == [] + + +def test_existing_directory_and_partial_receipt_are_preserved(tmp_path: Path) -> None: + store = ReceiptStore.create(tmp_path / "provider") + with pytest.raises(FileExistsError): + ReceiptStore.create(store.root) + business_id, items, fingerprint = batch().snapshot() + assert store.commit_or_reconcile(business_id, fingerprint, 0, items[0]) + path = next(store.root.iterdir()) + path.write_bytes(b"partial") + with pytest.raises(NonRetryableError, match="partial"): + store.commit_or_reconcile(business_id, fingerprint, 0, items[0]) + assert path.read_bytes() == b"partial" + + +def test_actual_handler_survives_cooperative_preflight_and_callback_serialization(tmp_path: Path) -> None: + store = ReceiptStore.create(tmp_path / "provider") + handler = BatchActivity(store).execute + worker = Worker(AsyncMock(spec=Client), task_queue="queue", activities=[handler]) + worker._cooperative_cancellation_supported = True + worker._validate_cooperative_activity_handlers() + assert worker.activities[ACTIVITY] == handler + invocation = CallbackInvocation( + handler=worker.activities[ACTIVITY], args=(asdict(batch()),), + info=ActivityInfo("task", ACTIVITY, "attempt", 2, "queue", "worker"), task={}, + ) + restored = pickle.loads(invocation.serialize()) + assert isinstance(restored.handler.__self__, BatchActivity) + assert restored.handler.__self__.store == store + assert restored.args == (asdict(batch()),) + assert inventory(store) == [] + + +def test_nested_handler_is_refused_by_the_same_preflight_and_serializer() -> None: + async def legacy_nested_handler(_body: dict[str, Any]) -> None: + pass + + worker = Worker(AsyncMock(spec=Client), task_queue="queue", activities=[legacy_nested_handler]) + worker._cooperative_cancellation_supported = True + with pytest.raises(RuntimeError, match="spawn-compatible"): + worker._validate_cooperative_activity_handlers() + invocation = CallbackInvocation( + handler=legacy_nested_handler, args=(), + info=ActivityInfo("task", ACTIVITY, "attempt", 2, "queue", "worker"), task={}, + ) + with pytest.raises(ValueError, match="spawn-compatible"): + invocation.serialize()