From 574e27b49a27e5969e75d5bf801d991af58b1af7 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Mon, 5 Oct 2026 22:15:54 +0000 Subject: [PATCH] Add published DBOS cancellation and durable compensation comparison --- .../workflows/cancellation-competitors.yml | 34 ++++ conformance/cancellation/dbos/README.md | 59 ++++++ conformance/cancellation/dbos/compose.yml | 18 ++ .../cancellation/dbos/requirements.txt | 1 + conformance/cancellation/dbos/scenario.py | 168 ++++++++++++++++++ conformance/cancellation/dbos/worker.py | 104 +++++++++++ 6 files changed, 384 insertions(+) create mode 100644 conformance/cancellation/dbos/README.md create mode 100644 conformance/cancellation/dbos/compose.yml create mode 100644 conformance/cancellation/dbos/requirements.txt create mode 100644 conformance/cancellation/dbos/scenario.py create mode 100644 conformance/cancellation/dbos/worker.py diff --git a/.github/workflows/cancellation-competitors.yml b/.github/workflows/cancellation-competitors.yml index fa23229..2841129 100644 --- a/.github/workflows/cancellation-competitors.yml +++ b/.github/workflows/cancellation-competitors.yml @@ -89,3 +89,37 @@ jobs: path: conformance/cancellation/temporal/evidence if-no-files-found: error retention-days: 90 + dbos: + name: DBOS published behavior + runs-on: ubuntu-latest + timeout-minutes: 10 + defaults: + run: + working-directory: conformance/cancellation/dbos + steps: + - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6 + with: + persist-credentials: false + - name: Run supported cancellation and compensation cases + run: | + set -euo pipefail + mkdir -p evidence + chmod 0777 evidence + git rev-parse HEAD > evidence/source.txt + sha256sum *.py requirements.txt compose.yml > evidence/source-files.txt + docker compose -p cancellation-dbos up -d + docker compose -p cancellation-dbos exec -T sdk sh -ec ' + python -m venv /tmp/venv + /tmp/venv/bin/pip install --no-cache-dir --report /evidence/packages.json -r requirements.txt + /tmp/venv/bin/python scenario.py + ' | tee evidence/run.log + - name: Remove the fixture + if: always() + run: docker compose -p cancellation-dbos down -v --remove-orphans + - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 + if: always() + with: + name: dbos-cancellation-behavior + path: conformance/cancellation/dbos/evidence + if-no-files-found: error + retention-days: 90 diff --git a/conformance/cancellation/dbos/README.md b/conformance/cancellation/dbos/README.md new file mode 100644 index 0000000..a5547c0 --- /dev/null +++ b/conformance/cancellation/dbos/README.md @@ -0,0 +1,59 @@ +# Published DBOS cancellation behavior + +This fixture uses the published Python SDK pinned in `requirements.txt` and +the digest-pinned Python image in `compose.yml`. SQLite stores synthetic +workflows in the container, and ordinary evidence files retain the observations. +No network service, application heartbeats or host ports are required. + +The request explicitly sets `cancel_children=True`. Async work enables the +supported `preemptible=True` step option. Blocking work uses a synchronous +step through `asyncio.to_thread` so it does not block the event loop or the +cancellation observer. + +## Run + +```sh +mkdir -p evidence +chmod 0777 evidence +docker compose -p cancellation-dbos up -d +docker compose -p cancellation-dbos exec -T sdk sh -ec ' + python -m venv /tmp/venv + /tmp/venv/bin/pip install --no-cache-dir --report /evidence/packages.json -r requirements.txt + /tmp/venv/bin/python scenario.py +' +docker compose -p cancellation-dbos down -v --remove-orphans +``` + +Three repetitions of seven cases record the physical callback, stored workflow +status and durable step outcomes separately: + +| Case | Observation | +| --- | --- | +| Preemptible async step with plain shielded cleanup | Work stops without heartbeats and its three-second cleanup completes while parent/child status is already CANCELLED | +| Blocking synchronous step | Runs its twelve-second sleep and performs a late effect, with no completed step checkpoint | +| Durable workflow steps in cancellation cleanup | Parent and child enter `finally`, but their cancelled workflow authority refuses the cleanup steps | +| Separately queued durable compensation | Completes with SUCCESS after the original parent/child are CANCELLED | +| Worker SIGKILL during plain step cleanup | A fresh worker does not automatically resume that cleanup in the four-second observation window, and the original runs remain CANCELLED | +| Worker SIGKILL during separately durable compensation | A fresh worker resumes its committed timer and completes compensation | +| Another cancellation during separately durable compensation | Original parent/child remain CANCELLED and compensation completes | + +The separately durable compensation has a stable workflow ID derived from the +original parent. The controller starts it outside the cancelled ancestry after +observing the callback's exit. Its entered step and `DBOS.sleep` are committed +before SIGKILL. This is an application pattern using supported APIs. The +application explicitly starts compensation and manages its relationship to +the cancellation request. + +`results.jsonl` records workflow statuses and public step inspection, process IDs, +request timestamps, callback exit and cleanup markers. Each run checks the +observed outcomes, and compensation finishes within thirty seconds of the +original request. That observed duration is separate from a native cancellation +budget. Physical stop is separate from an immediately recorded CANCELLED status. +The scenario retains raw results before an assertion fails. + +The elapsed times locate callback exit and cleanup completion in each case. +The fixture demonstrates DBOS's useful async preemption and durable recovery pattern as well +as their operational boundaries. See the official +[workflow management guide](https://docs.dbos.dev/python/tutorials/workflow-management), +[preemptible step reference](https://docs.dbos.dev/python/reference/decorators), +and [recursive cancellation API](https://docs.dbos.dev/python/reference/contexts#cancel_workflow). diff --git a/conformance/cancellation/dbos/compose.yml b/conformance/cancellation/dbos/compose.yml new file mode 100644 index 0000000..56fede6 --- /dev/null +++ b/conformance/cancellation/dbos/compose.yml @@ -0,0 +1,18 @@ +services: + sdk: + image: python@sha256:3dd7cc108ec1493442514f5c2a871af6af0ec31d768ff6e378a93340c3b3db5f + user: "1000:1000" + init: true + working_dir: /experiment + command: ["sleep", "3600"] + environment: + PYTHONUNBUFFERED: "1" + PYTHONDONTWRITEBYTECODE: "1" + volumes: + - ./worker.py:/experiment/worker.py:ro + - ./scenario.py:/experiment/scenario.py:ro + - ./requirements.txt:/experiment/requirements.txt:ro + - ./evidence:/evidence + mem_limit: 768m + memswap_limit: 768m + cpus: 2 diff --git a/conformance/cancellation/dbos/requirements.txt b/conformance/cancellation/dbos/requirements.txt new file mode 100644 index 0000000..fdd193f --- /dev/null +++ b/conformance/cancellation/dbos/requirements.txt @@ -0,0 +1 @@ +dbos[aiosqlite]==3.2.0 diff --git a/conformance/cancellation/dbos/scenario.py b/conformance/cancellation/dbos/scenario.py new file mode 100644 index 0000000..52ebcf3 --- /dev/null +++ b/conformance/cancellation/dbos/scenario.py @@ -0,0 +1,168 @@ +import asyncio +import importlib.metadata +import json +import os +import signal +import subprocess +import sys +import time +import uuid +from pathlib import Path + +from dbos import DBOSClient + +ROOT = Path('/evidence') +process = None +log = (ROOT / 'worker.log').open('a') + + +def snapshot(value): + return value if isinstance(value, dict) else vars(value) + + +def start_worker(): + global process + (ROOT / 'worker-ready').unlink(missing_ok=True) + process = subprocess.Popen([sys.executable, 'worker.py'], stdout=log, stderr=subprocess.STDOUT) + + +def events(token): + path = ROOT / 'events.jsonl' + return [row for line in path.read_text().splitlines() + if (row := json.loads(line))['token'] == token] if path.exists() else [] + + +async def wait_for(token, stage, timeout=20): + until = time.monotonic() + timeout + while time.monotonic() < until: + found = [row for row in events(token) if row['stage'] == stage] + if found: + return found[-1] + if process.poll() is not None: + raise RuntimeError('DBOS worker exited unexpectedly') + await asyncio.sleep(0.02) + raise RuntimeError('Missing ' + stage) + + +async def ready(): + until = time.monotonic() + 20 + while time.monotonic() < until: + if (ROOT / 'worker-ready').exists(): + return + if process.poll() is not None: + raise RuntimeError('Worker failed before ready') + await asyncio.sleep(0.05) + raise RuntimeError('Worker did not become ready') + + +async def case(client, mode, repetition, kill_cleanup=False, duplicate=False): + token = 'comparison-' + uuid.uuid4().hex + arg = dict(token=token, mode=mode) + handle = await client.enqueue_async(dict(workflow_name='parent', queue_name='comparison', + workflow_id=token, app_version='cancellation-comparison-v1'), arg) + await wait_for(token, 'callback.started') + requested = time.monotonic() + await client.cancel_workflow_async(token, cancel_children=True) + status_at_reply = (await handle.get_status()).status + compensation_handle = None + killed_pid = replacement_pid = None + if mode == 'detached': + await wait_for(token, 'callback.exited') + compensation_handle = await client.enqueue_async(dict(workflow_name='compensation', + queue_name='comparison', workflow_id=token + '-compensation', + app_version='cancellation-comparison-v1'), arg) + await wait_for(token, 'compensation.entered') + # Wait for the entered step to be committed before killing its worker. + until = time.monotonic() + 5 + while len(await client.list_workflow_steps_async(token + '-compensation')) < 2: + if time.monotonic() >= until: + raise RuntimeError('Compensation entry did not commit') + await asyncio.sleep(0.02) + if kill_cleanup: + if mode == 'step_cleanup': + await wait_for(token, 'step.cleanup.entered') + killed_pid = process.pid + os.kill(process.pid, signal.SIGKILL) + process.wait(timeout=5) + start_worker() + await ready() + if duplicate: + await client.cancel_workflow_async(token, cancel_children=True) + if mode == 'detached': + await asyncio.wait_for(compensation_handle.get_result(), 20) + await wait_for(token, 'compensation.completed') + elif not kill_cleanup: + await wait_for(token, 'callback.exited') + if mode == 'workflow_cleanup': + await wait_for(token, 'root.cleanup.refused') + await wait_for(token, 'child.cleanup.refused') + else: + # Cancelled runs are terminal, unlike a separately pending compensation. + await asyncio.sleep(4) + if kill_cleanup: + replacement_pid = process.pid + child_handle = await client.retrieve_workflow_async(token + '-child') + rows = events(token) + parent_status = await handle.get_status() + child_status = await child_handle.get_status() + row = dict(token=token, mode=mode, repetition=repetition, requested=requested, + status_at_reply=status_at_reply, parent_status=snapshot(parent_status), + child_status=snapshot(child_status), killed_pid=killed_pid, + replacement_pid=replacement_pid, duplicate=duplicate, events=rows, + parent_steps=[snapshot(step) for step in await client.list_workflow_steps_async(token)], + child_steps=[snapshot(step) for step in await client.list_workflow_steps_async(token + '-child')], + compensation_status=snapshot(await compensation_handle.get_status()) if compensation_handle else None, + compensation_steps=[snapshot(step) for step in await client.list_workflow_steps_async(token + '-compensation')] if compensation_handle else []) + with (ROOT / 'results.jsonl').open('a') as output: + output.write(json.dumps(row, default=str) + '\n') + print(json.dumps(dict(mode=mode, repetition=repetition, status_at_reply=status_at_reply, + parent=parent_status.status, child=child_status.status, + stages=[event['stage'] for event in rows], killed_pid=killed_pid, + replacement_pid=replacement_pid, duplicate=duplicate)), flush=True) + assert parent_status.status == child_status.status == 'CANCELLED', row + assert row['child_steps'] == [], row + if mode != 'sync': + assert not any(event['stage'] == 'callback.late_effect' for event in rows), row + if kill_cleanup: + assert killed_pid != replacement_pid, row + if mode == 'detached': + assert row['compensation_status']['status'] == 'SUCCESS', row + assert sum(event['stage'] == 'compensation.entered' for event in rows) == 1, row + assert sum(event['stage'] == 'compensation.completed' for event in rows) == 1, row + assert rows[-1]['monotonic'] - requested < 30, row + elif mode == 'sync': + assert any(event['stage'] == 'callback.late_effect' for event in rows), row + elif mode == 'step_cleanup' and kill_cleanup: + assert not any(event['stage'] == 'step.cleanup.completed' for event in rows), row + elif mode == 'workflow_cleanup': + assert not any(event['stage'].endswith('.cleanup.completed') for event in rows), row + else: + assert any(event['stage'] == 'step.cleanup.completed' for event in rows), row + + +async def main(): + start_worker() + try: + await ready() + client = DBOSClient(system_database_url='sqlite:////tmp/state.sqlite') + print(json.dumps(dict(package=importlib.metadata.version('dbos'))), flush=True) + for mode in ('step_cleanup', 'sync', 'workflow_cleanup', 'detached'): + for repetition in range(1, 4): + await case(client, mode, repetition) + for mode in ('step_cleanup', 'detached'): + for repetition in range(1, 4): + await case(client, mode, repetition, kill_cleanup=True) + for repetition in range(1, 4): + await case(client, 'detached', repetition, duplicate=True) + finally: + if process is not None and process.poll() is None: + process.terminate() + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=5) + log.close() + + +asyncio.run(main()) diff --git a/conformance/cancellation/dbos/worker.py b/conformance/cancellation/dbos/worker.py new file mode 100644 index 0000000..8dea4ef --- /dev/null +++ b/conformance/cancellation/dbos/worker.py @@ -0,0 +1,104 @@ +import asyncio +import json +import os +import time +from pathlib import Path + +from dbos import DBOS, SetWorkflowID, error as dbos_error + +ROOT = Path('/evidence') + + +def observe(token, stage, **extra): + with (ROOT / 'events.jsonl').open('a') as output: + output.write(json.dumps(dict(token=token, stage=stage, pid=os.getpid(), + monotonic=time.monotonic(), **extra)) + '\n') + + +@DBOS.step() +async def mark(token, stage): + observe(token, stage) + + +@DBOS.step(preemptible=True) +async def async_work(arg): + observe(arg['token'], 'callback.started') + try: + until = time.monotonic() + 12 + while time.monotonic() < until: + await asyncio.sleep(0.02) + observe(arg['token'], 'callback.late_effect') + return 'late-result' + finally: + if arg['mode'] == 'step_cleanup': + observe(arg['token'], 'step.cleanup.entered') + await asyncio.shield(asyncio.sleep(3)) + observe(arg['token'], 'step.cleanup.completed') + observe(arg['token'], 'callback.exited') + + +@DBOS.step() +def sync_work(arg): + observe(arg['token'], 'callback.started') + try: + time.sleep(12) + observe(arg['token'], 'callback.late_effect') + return 'late-result' + finally: + observe(arg['token'], 'callback.exited') + + +async def workflow_cleanup(arg, stage): + observe(arg['token'], stage + '.cleanup.attempted') + try: + await asyncio.shield(mark(arg['token'], stage + '.cleanup.entered')) + await asyncio.shield(DBOS.sleep_async(3)) + await asyncio.shield(mark(arg['token'], stage + '.cleanup.completed')) + except (dbos_error.DBOSWorkflowCancelledError, Exception) as failure: + observe(arg['token'], stage + '.cleanup.refused', exception=type(failure).__name__) + + +@DBOS.workflow() +async def child(arg): + try: + if arg['mode'] == 'sync': + return await asyncio.to_thread(sync_work, arg) + return await async_work(arg) + finally: + if arg['mode'] == 'workflow_cleanup': + await workflow_cleanup(arg, 'child') + + +@DBOS.workflow() +async def parent(arg): + try: + with SetWorkflowID(arg['token'] + '-child'): + handle = await DBOS.start_workflow_async(child, arg) + return await handle.get_result() + finally: + if arg['mode'] == 'workflow_cleanup': + await workflow_cleanup(arg, 'root') + + +@DBOS.workflow() +async def compensation(arg): + await mark(arg['token'], 'compensation.entered') + await DBOS.sleep_async(3) + await mark(arg['token'], 'compensation.completed') + + +async def main(): + (ROOT / 'worker-ready').write_text(str(os.getpid())) + await asyncio.Event().wait() + + +DBOS(config=dict(name='cancellation-comparison', + application_version='cancellation-comparison-v1', executor_id='local', + system_database_url='sqlite:////tmp/state.sqlite', + run_admin_server=False, enable_otlp=False, log_level='WARNING')) +DBOS.launch() +DBOS.register_queue('comparison') +try: + asyncio.run(main()) +finally: + DBOS.destroy()