From 62cef61b52e826020cb97174eb8dc47034d3fe81 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Mon, 5 Oct 2026 21:21:31 +0000 Subject: [PATCH 1/3] Add published Restate cancellation comparison fixture --- conformance/cancellation/restate/README.md | 43 +++++ conformance/cancellation/restate/compose.yml | 46 +++++ .../cancellation/restate/requirements.txt | 3 + conformance/cancellation/restate/scenario.py | 157 ++++++++++++++++++ conformance/cancellation/restate/service.py | 101 +++++++++++ 5 files changed, 350 insertions(+) create mode 100644 conformance/cancellation/restate/README.md create mode 100644 conformance/cancellation/restate/compose.yml create mode 100644 conformance/cancellation/restate/requirements.txt create mode 100644 conformance/cancellation/restate/scenario.py create mode 100644 conformance/cancellation/restate/service.py diff --git a/conformance/cancellation/restate/README.md b/conformance/cancellation/restate/README.md new file mode 100644 index 0000000..f47a4ea --- /dev/null +++ b/conformance/cancellation/restate/README.md @@ -0,0 +1,43 @@ +# Published Restate cancellation experiment + +This fixture supports [cancellation issue 136](https://github.com/durable-workflow/.github/issues/136). It exercises Restate Server 1.7.13 and Python SDK 1.0.5 through the documented cooperative cancellation API. It measures behavior, not runtime throughput. + +The parent and child are durable Restate Workflows. The child calls a leaf service. Each handler catches `TerminalError`, performs durable compensation and rethrows, following [Restate's cancellation guidance](https://docs.restate.dev/services/invocation/managing-invocations). + +## Cases + +Three repetitions each: + +- Cancel a durable wait and inspect all three invocation outcomes. +- Cancel an asynchronous `ctx.run_typed` callback, with no application heartbeat. Observe its actual exit and a possible late side effect. +- Cancel a blocking `ctx.run_typed` callback. Wait for actual callback exit even if the durable invocation already completed. +- SIGKILL the SDK endpoint after root cleanup starts. Start a fresh process and inspect journal replay and cleanup completion. +- Repeat cancellation while root cleanup awaits a durable timer. Inspect whether cleanup completes. + +The callback duration is 12 seconds. Recovery and duplicate cases insert a three-second durable wait between two cleanup actions. Application markers and durable `sys_invocation` records are retained separately. A cancellation result cannot stand in for callback exit or completed compensation. + +## Run + +Use Docker Compose and a host checkout owned by UID/GID 1000. No host ports are published. The runtime is isolated and its database is disposable. + +```sh +cd conformance/cancellation/restate +docker compose -p cancellation-restate up -d --wait +docker compose -p cancellation-restate exec -T sdk sh -ec ' + python -m venv /tmp/venv + /tmp/venv/bin/pip install --no-cache-dir --report /experiment/packages.json -r requirements.txt + /tmp/venv/bin/python scenario.py + /tmp/venv/bin/python scenario.py duplicate +' +mkdir -p evidence +docker compose -p cancellation-restate cp sdk:/experiment/results.jsonl evidence/results.jsonl +docker compose -p cancellation-restate cp sdk:/experiment/events.jsonl evidence/events.jsonl +docker compose -p cancellation-restate cp sdk:/experiment/packages.json evidence/packages.json +docker compose -p cancellation-restate cp sdk:/experiment/service.log evidence/service.log +docker compose -p cancellation-restate logs runtime > evidence/runtime.log +docker compose -p cancellation-restate down -v --remove-orphans +``` + +Keep measured failures and successes. Setup and import failures do not establish cancellation behavior. Read each case's application observations and durable outcomes before drawing a conclusion. The fixture records observed behavior without requiring another product to satisfy Durable Workflow's contract. + +The pinned images and installed-package report bind each run to published artifacts. This comparison does not qualify Durable Workflow's unpublished cancellation candidate or replace its required published mixed-language cascade. diff --git a/conformance/cancellation/restate/compose.yml b/conformance/cancellation/restate/compose.yml new file mode 100644 index 0000000..e2209d7 --- /dev/null +++ b/conformance/cancellation/restate/compose.yml @@ -0,0 +1,46 @@ +services: + runtime: + image: docker.restate.dev/restatedev/restate@sha256:65abc8016d408b8b424f5e492a5da37c24573207e217a3d43242dd8b3fd8dc61 + user: "1000:1000" + init: true + command: ["--no-logo"] + environment: + RESTATE_BASE_DIR: /restate-data + RESTATE_ADVERTISED_HOST: restate + RESTATE_NODE_NAME: cancellation-comparison + tmpfs: + - /restate-data:uid=1000,gid=1000,mode=0755 + networks: + default: + aliases: [restate] + mem_limit: 1024m + memswap_limit: 1024m + cpus: 2 + healthcheck: + test: ["CMD", "curl", "--fail", "--silent", "http://localhost:9070/health"] + interval: 2s + timeout: 5s + retries: 30 + sdk: + image: python@sha256:02108f5d322dd89f1c9e552442c25acb0543dfdbc455693a5599624f20d9155d + user: "1000:1000" + init: true + working_dir: /experiment + command: ["sleep", "3600"] + environment: + PYTHONUNBUFFERED: "1" + volumes: + - ./service.py:/experiment/service.py:ro + - ./scenario.py:/experiment/scenario.py:ro + - ./requirements.txt:/experiment/requirements.txt:ro + tmpfs: + - /experiment:uid=1000,gid=1000,mode=0755 + networks: + default: + aliases: [sdk] + mem_limit: 512m + memswap_limit: 512m + cpus: 1 + depends_on: + runtime: + condition: service_healthy diff --git a/conformance/cancellation/restate/requirements.txt b/conformance/cancellation/restate/requirements.txt new file mode 100644 index 0000000..092c0e0 --- /dev/null +++ b/conformance/cancellation/restate/requirements.txt @@ -0,0 +1,3 @@ +restate-sdk==1.0.5 +Hypercorn==0.18.0 +httpx==0.28.1 diff --git a/conformance/cancellation/restate/scenario.py b/conformance/cancellation/restate/scenario.py new file mode 100644 index 0000000..85af761 --- /dev/null +++ b/conformance/cancellation/restate/scenario.py @@ -0,0 +1,157 @@ +import json +import os +import signal +import subprocess +import sys +import time +import uuid +from pathlib import Path + +import httpx + +ROOT = Path('/experiment') +client = httpx.Client(timeout=5) +process = None +service_log = (ROOT / 'service.log').open('a') + + +def start(): + global process + process = subprocess.Popen([sys.executable, 'service.py'], stdout=service_log, + stderr=subprocess.STDOUT) + until = time.monotonic() + 15 + while time.monotonic() < until: + try: + response = client.get('http://sdk:9080/health') + if response.status_code in (200, 404): + return + except httpx.HTTPError: + pass + time.sleep(0.1) + raise RuntimeError('SDK endpoint did not become ready') + + +def events(token): + path = ROOT / 'events.jsonl' + if not path.exists(): + return [] + return [row for line in path.read_text().splitlines() + if (row := json.loads(line))['token'] == token] + + +def wait_for(token, stage, timeout): + until = time.monotonic() + timeout + while time.monotonic() < until: + found = [row for row in events(token) if row['stage'] == stage] + if found: + return found[-1] + time.sleep(0.05) + raise RuntimeError('Missing ' + stage + ' for ' + token) + + +def query(sql): + response = client.post('http://restate:9070/query', json={'query': sql}, + headers={'Accept': 'application/json'}) + response.raise_for_status() + return response.json() + + +def case(mode, repetition, kill_cleanup=False, duration=12, duplicate_request=False): + token = 'comparison-' + uuid.uuid4().hex + arg = dict(token=token, mode=mode, duration=duration, kill_cleanup=kill_cleanup, + duplicate_during_cleanup=duplicate_request) + response = client.post(f'http://restate:8080/CancellationRoot/{token}/run/send', json=arg) + response.raise_for_status() + submitted = response.json() + invocation = submitted['invocationId'] + ready = 'leaf.wait.started' if mode == 'durable_wait' else 'leaf.callback.started' + wait_for(token, ready, 15) + requested = time.monotonic() + accepted = client.patch(f'http://restate:9070/invocations/{invocation}/cancel') + accepted.raise_for_status() + duplicate = None + if duplicate_request: + wait_for(token, 'root.cleanup.entered', 20) + duplicate = client.patch(f'http://restate:9070/invocations/{invocation}/cancel') + duplicate.raise_for_status() + killed_pid = None + replacement_pid = None + if kill_cleanup: + wait_for(token, 'root.cleanup.entered', 20) + killed_pid = process.pid + os.kill(killed_pid, signal.SIGKILL) + process.wait(timeout=5) + start() + replacement_pid = process.pid + until = time.monotonic() + max(35, duration + 10) + while time.monotonic() < until: + observed = events(token) + invocation_ids = sorted({event['invocation_id'] for event in observed if 'invocation_id' in event}) + call_graph = query("SELECT * FROM sys_invocation WHERE id IN (" + + ','.join("'" + value + "'" for value in invocation_ids) + ")") + rows = call_graph['rows'] + if len(rows) == 3 and all(item['status'] == 'completed' for item in rows): + break + time.sleep(0.1) + observed = events(token) + completed = [event for event in observed if event['stage'] == 'root.cleanup.completed'] + callback_exit = None + settled = time.monotonic() - requested + if mode in ('async_run', 'sync_run'): + callback_exit = wait_for(token, 'leaf.callback.exited', duration + 10) + observed = events(token) + row = dict(mode=mode, repetition=repetition, token=token, request_monotonic=requested, + submitted=submitted, cancel_status=accepted.status_code, + cancel_body=accepted.text, duplicate_status=duplicate.status_code if duplicate is not None else None, + duplicate_body=duplicate.text if duplicate is not None else None, + cleanup_elapsed=completed[-1]['monotonic'] - requested if completed else None, + cascade_settled_elapsed=settled, + cascade_completed=len(rows) == 3 and all(item['status'] == 'completed' for item in rows), + callback_stop_observed=callback_exit is not None, + callback_exit_elapsed=callback_exit['monotonic'] - requested if callback_exit else None, + late_effect_observed=any(event['stage'] == 'leaf.callback.late_effect' for event in observed), + duplicate_during_cleanup=duplicate_request, + killed_pid=killed_pid, replacement_pid=replacement_pid, events=observed) + # Query the durable invocation projection, not just application log entries. + time.sleep(0.2) + row['durable_invocations'] = query("SELECT * FROM sys_invocation WHERE id = '" + invocation + "'") + invocation_ids = sorted({event['invocation_id'] for event in row['events'] if 'invocation_id' in event}) + row['call_graph'] = query("SELECT * FROM sys_invocation WHERE id IN (" + + ','.join("'" + value + "'" for value in invocation_ids) + ")") + with (ROOT / 'results.jsonl').open('a') as output: + output.write(json.dumps(row) + '\n') + print(json.dumps({key: row[key] for key in ( + 'mode', 'repetition', 'cascade_completed', 'cleanup_elapsed', + 'callback_exit_elapsed', 'late_effect_observed', 'duplicate_during_cleanup', + 'killed_pid', 'replacement_pid' + )}), flush=True) + + +try: + start() + response = client.post('http://restate:9070/deployments', json={'uri': 'http://sdk:9080'}) + response.raise_for_status() + (ROOT / 'deployment.json').write_text(json.dumps(response.json(), indent=2)) + if len(sys.argv) > 1 and sys.argv[1] == 'duplicate': + for repetition in range(1, 4): + case('durable_wait', repetition, duplicate_request=True, duration=12) + elif len(sys.argv) > 1 and sys.argv[1] == 'callback': + for mode in ('async_run', 'sync_run'): + for repetition in range(1, 4): + case(mode, repetition) + else: + for mode in ('durable_wait', 'async_run', 'sync_run'): + for repetition in range(1, 4): + case(mode, repetition) + for repetition in range(1, 4): + case('durable_wait', repetition, kill_cleanup=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) + service_log.close() + client.close() diff --git a/conformance/cancellation/restate/service.py b/conformance/cancellation/restate/service.py new file mode 100644 index 0000000..9ccdd2b --- /dev/null +++ b/conformance/cancellation/restate/service.py @@ -0,0 +1,101 @@ +import asyncio +import json +import logging +import os +import time +from datetime import timedelta +from pathlib import Path + +import restate + +logging.basicConfig(level=logging.INFO) +ROOT = Path('/experiment') +root = restate.Workflow('CancellationRoot') +child = restate.Workflow('CancellationChild') +leaf = restate.Service('CancellationLeaf') + + +def observe(token, stage, **extra): + row = dict(token=token, stage=stage, monotonic=time.monotonic(), + wall=time.time(), pid=os.getpid(), **extra) + with (ROOT / 'events.jsonl').open('a') as output: + output.write(json.dumps(row) + '\n') + output.flush() + + +async def async_callback(arg): + observe(arg['token'], 'leaf.callback.started') + try: + await asyncio.sleep(arg['duration']) + observe(arg['token'], 'leaf.callback.late_effect') + return 'late-result' + finally: + observe(arg['token'], 'leaf.callback.exited') + + +def sync_callback(arg): + observe(arg['token'], 'leaf.callback.started') + try: + time.sleep(arg['duration']) + observe(arg['token'], 'leaf.callback.late_effect') + return 'late-result' + finally: + observe(arg['token'], 'leaf.callback.exited') + + +async def compensate(ctx, arg, stage, error): + # Follow the documented TerminalError -> durable compensation -> rethrow. + await ctx.run_typed(stage + '.cleanup.entered', observe, + token=arg['token'], stage=stage + '.cleanup.entered', error=str(error)) + if stage == 'root' and (arg.get('kill_cleanup') or arg.get('duplicate_during_cleanup')): + await ctx.sleep(timedelta(seconds=3)) + await ctx.run_typed(stage + '.cleanup.completed', observe, + token=arg['token'], stage=stage + '.cleanup.completed') + + +@leaf.handler() +async def work(ctx: restate.Context, arg: dict): + observe(arg['token'], 'leaf.handler.entered', invocation_id=ctx.request().id) + try: + if arg['mode'] == 'durable_wait': + await ctx.run_typed('leaf.ready', observe, token=arg['token'], stage='leaf.wait.started') + await ctx.sleep(timedelta(seconds=60)) + elif arg['mode'] == 'async_run': + await ctx.run_typed('leaf.callback', async_callback, arg=arg) + else: + await ctx.run_typed('leaf.callback', sync_callback, arg=arg) + observe(arg['token'], 'leaf.handler.completed') + return 'completed' + except restate.TerminalError as error: + await compensate(ctx, arg, 'leaf', error) + raise + + +@child.main() +async def child_run(ctx: restate.WorkflowContext, arg: dict): + observe(arg['token'], 'child.handler.entered', invocation_id=ctx.request().id) + try: + return await ctx.service_call(work, arg) + except restate.TerminalError as error: + await compensate(ctx, arg, 'child', error) + raise + + +@root.main() +async def run(ctx: restate.WorkflowContext, arg: dict): + observe(arg['token'], 'root.handler.entered', invocation_id=ctx.request().id) + try: + return await ctx.workflow_call(child_run, arg['token'] + '-child', arg) + except restate.TerminalError as error: + await compensate(ctx, arg, 'root', error) + raise + + +app = restate.app(services=[root, child, leaf]) + +if __name__ == '__main__': + from hypercorn.asyncio import serve + from hypercorn.config import Config + config = Config() + config.bind = ['0.0.0.0:9080'] + asyncio.run(serve(app, config)) From 34a3435388dcb29119459fa28e4c4b01fef8ae38 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Mon, 5 Oct 2026 21:23:08 +0000 Subject: [PATCH 2/3] Run published cancellation cases in a thin hosted job --- .../workflows/cancellation-competitors.yml | 56 +++++++++++++++++++ 1 file changed, 56 insertions(+) create mode 100644 .github/workflows/cancellation-competitors.yml diff --git a/.github/workflows/cancellation-competitors.yml b/.github/workflows/cancellation-competitors.yml new file mode 100644 index 0000000..e8226b2 --- /dev/null +++ b/.github/workflows/cancellation-competitors.yml @@ -0,0 +1,56 @@ +name: Published cancellation behavior + +on: + pull_request: + paths: + - conformance/cancellation/** + - .github/workflows/cancellation-competitors.yml + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: cancellation-competitors-${{ github.ref }} + cancel-in-progress: true + +jobs: + restate: + name: Restate published behavior + runs-on: ubuntu-latest + timeout-minutes: 10 + defaults: + run: + working-directory: conformance/cancellation/restate + steps: + - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6 + with: + persist-credentials: false + - name: Run documented cancellation cases + run: | + set -euo pipefail + mkdir -p evidence + git rev-parse HEAD > evidence/source.txt + docker compose -p cancellation-restate up -d --wait + docker compose -p cancellation-restate exec -T sdk sh -ec ' + python -m venv /tmp/venv + /tmp/venv/bin/pip install --no-cache-dir --report /experiment/packages.json -r requirements.txt + /tmp/venv/bin/python scenario.py + /tmp/venv/bin/python scenario.py duplicate + ' | tee evidence/run.log + - name: Collect observations and remove the fixture + if: always() + run: | + mkdir -p evidence + for file in results.jsonl events.jsonl packages.json service.log deployment.json; do + docker compose -p cancellation-restate cp "sdk:/experiment/$file" "evidence/$file" || true + done + docker compose -p cancellation-restate logs runtime > evidence/runtime.log + docker compose -p cancellation-restate down -v --remove-orphans + - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 + if: always() + with: + name: restate-cancellation-behavior + path: conformance/cancellation/restate/evidence + if-no-files-found: error + retention-days: 90 From 2a384595387a5dd027f58d8a814d5e1bf127736d Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Mon, 5 Oct 2026 21:25:20 +0000 Subject: [PATCH 3/3] Keep experiment observations on the mounted filesystem --- .github/workflows/cancellation-competitors.yml | 6 ++---- .gitignore | 1 + conformance/cancellation/restate/README.md | 8 ++------ conformance/cancellation/restate/compose.yml | 1 + conformance/cancellation/restate/scenario.py | 2 +- conformance/cancellation/restate/service.py | 2 +- 6 files changed, 8 insertions(+), 12 deletions(-) diff --git a/.github/workflows/cancellation-competitors.yml b/.github/workflows/cancellation-competitors.yml index e8226b2..f9214f8 100644 --- a/.github/workflows/cancellation-competitors.yml +++ b/.github/workflows/cancellation-competitors.yml @@ -30,11 +30,12 @@ jobs: run: | set -euo pipefail mkdir -p evidence + chmod 0777 evidence git rev-parse HEAD > evidence/source.txt docker compose -p cancellation-restate up -d --wait docker compose -p cancellation-restate exec -T sdk sh -ec ' python -m venv /tmp/venv - /tmp/venv/bin/pip install --no-cache-dir --report /experiment/packages.json -r requirements.txt + /tmp/venv/bin/pip install --no-cache-dir --report /evidence/packages.json -r requirements.txt /tmp/venv/bin/python scenario.py /tmp/venv/bin/python scenario.py duplicate ' | tee evidence/run.log @@ -42,9 +43,6 @@ jobs: if: always() run: | mkdir -p evidence - for file in results.jsonl events.jsonl packages.json service.log deployment.json; do - docker compose -p cancellation-restate cp "sdk:/experiment/$file" "evidence/$file" || true - done docker compose -p cancellation-restate logs runtime > evidence/runtime.log docker compose -p cancellation-restate down -v --remove-orphans - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 diff --git a/.gitignore b/.gitignore index 9db3855..1c4b5bc 100644 --- a/.gitignore +++ b/.gitignore @@ -2,6 +2,7 @@ __pycache__/ *.py[cod] node_modules/ +/conformance/cancellation/*/evidence/ /authoritative-verification.json /candidate.json diff --git a/conformance/cancellation/restate/README.md b/conformance/cancellation/restate/README.md index f47a4ea..f2d5558 100644 --- a/conformance/cancellation/restate/README.md +++ b/conformance/cancellation/restate/README.md @@ -22,18 +22,14 @@ Use Docker Compose and a host checkout owned by UID/GID 1000. No host ports are ```sh cd conformance/cancellation/restate +mkdir -p evidence docker compose -p cancellation-restate up -d --wait docker compose -p cancellation-restate exec -T sdk sh -ec ' python -m venv /tmp/venv - /tmp/venv/bin/pip install --no-cache-dir --report /experiment/packages.json -r requirements.txt + /tmp/venv/bin/pip install --no-cache-dir --report /evidence/packages.json -r requirements.txt /tmp/venv/bin/python scenario.py /tmp/venv/bin/python scenario.py duplicate ' -mkdir -p evidence -docker compose -p cancellation-restate cp sdk:/experiment/results.jsonl evidence/results.jsonl -docker compose -p cancellation-restate cp sdk:/experiment/events.jsonl evidence/events.jsonl -docker compose -p cancellation-restate cp sdk:/experiment/packages.json evidence/packages.json -docker compose -p cancellation-restate cp sdk:/experiment/service.log evidence/service.log docker compose -p cancellation-restate logs runtime > evidence/runtime.log docker compose -p cancellation-restate down -v --remove-orphans ``` diff --git a/conformance/cancellation/restate/compose.yml b/conformance/cancellation/restate/compose.yml index e2209d7..468a3c3 100644 --- a/conformance/cancellation/restate/compose.yml +++ b/conformance/cancellation/restate/compose.yml @@ -33,6 +33,7 @@ services: - ./service.py:/experiment/service.py:ro - ./scenario.py:/experiment/scenario.py:ro - ./requirements.txt:/experiment/requirements.txt:ro + - ./evidence:/evidence tmpfs: - /experiment:uid=1000,gid=1000,mode=0755 networks: diff --git a/conformance/cancellation/restate/scenario.py b/conformance/cancellation/restate/scenario.py index 85af761..1f9ced4 100644 --- a/conformance/cancellation/restate/scenario.py +++ b/conformance/cancellation/restate/scenario.py @@ -9,7 +9,7 @@ import httpx -ROOT = Path('/experiment') +ROOT = Path('/evidence') client = httpx.Client(timeout=5) process = None service_log = (ROOT / 'service.log').open('a') diff --git a/conformance/cancellation/restate/service.py b/conformance/cancellation/restate/service.py index 9ccdd2b..92c4b56 100644 --- a/conformance/cancellation/restate/service.py +++ b/conformance/cancellation/restate/service.py @@ -9,7 +9,7 @@ import restate logging.basicConfig(level=logging.INFO) -ROOT = Path('/experiment') +ROOT = Path('/evidence') root = restate.Workflow('CancellationRoot') child = restate.Workflow('CancellationChild') leaf = restate.Service('CancellationLeaf')