Add published DBOS cancellation and durable compensation comparison

.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

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).

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

conformance/cancellation/dbos/requirements.txt

@@ -0,0 +1 @@
+dbos[aiosqlite]==3.2.0

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())

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()