diff --git a/.github/workflows/sdk-activity-upgrades.yml b/.github/workflows/sdk-activity-upgrades.yml new file mode 100644 index 0000000..94a994b --- /dev/null +++ b/.github/workflows/sdk-activity-upgrades.yml @@ -0,0 +1,31 @@ +name: published SDK and application activity upgrades + +on: + push: + branches: [main] + pull_request: + branches: [main] + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: sdk-activity-upgrades-${{ github.event_name }}-${{ github.event_name == 'pull_request' && github.ref || github.sha }} + cancel-in-progress: ${{ github.event_name == 'pull_request' }} + +jobs: + upgrades: + name: Pending activity SDK/application upgrades (nine directions) + runs-on: ubuntu-latest + timeout-minutes: 25 + env: + SDK_ACTIVITY_UPGRADES_COMPOSE_PROJECT_NAME: sample-app-sdk-activity-upgrades-${{ github.run_id }}-${{ github.run_attempt }} + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + - name: Resolve the exact published tuple + run: scripts/resolve-current-artifacts.sh >> "$GITHUB_ENV" + - name: Test original-activity and package evidence + run: python3 -m unittest discover -s polyglot/python_worker/tests -p test_sdk_activity_upgrades.py -v + - name: Replay pending activities across SDK and application replacements + run: scripts/sdk-activity-upgrades.sh diff --git a/polyglot/activities/README.md b/polyglot/activities/README.md index 1672bee..0ddba72 100644 --- a/polyglot/activities/README.md +++ b/polyglot/activities/README.md @@ -1,5 +1,49 @@ # Published remote activity retry and recovery +## Pending activities across SDK and application upgrades + +Run from the prepared development container with the published tuple resolved: + +```bash +while IFS= read -r assignment; do export "$assignment"; done \ + < <(scripts/resolve-current-artifacts.sh) +SDK_ACTIVITY_UPGRADES_COMPOSE_PROJECT_NAME=sample-app-activity-upgrades \ + scripts/sdk-activity-upgrades.sh +``` + +This separate experiment runs eighteen originals: nine pending workflow/activity +directions across PHP, Python and Rust, then nine fresh current-package controls. +It uses the same explicit previous-package baseline as the timer upgrades and +the current qualified tuple. Both image generations are built and their actual +installed package identities checked before the first workflow starts. + +Previous-package workers register the workflow without an activity handler. +Each original must reach public waiting status with one unstarted activity, +version 1/range 1..1, command sequence 2, and its complete retry/timeout policy. +The runner records all nine original histories, kills those workflow workers, +retires only their exact registrations and starts cold current-package workers +with revision-two application helpers and activity handlers. + +Replayed originals must retain version 1 and every event in their original +history prefix. Fresh current runs select version 2/range 1..2 and the revised +policy. Each actual activity executes once on the current SDK after replacement, +returns exact int64/null/boolean/fraction/Unicode/list values and completes the +same original workflow once. The observer checks both durable results, actual +attempt/execution identities and completion inside the original total budget. +Python's registered wrapper and recorded definition stay identical; the selected +business helper changes. Incompatible Python class adoption and build routing +remain separate gates. + +Pending admission has a twenty-second observer bound and each result a +120-second observer bound. Authored activities retain their thirty-second +attempt budget and original 120-second total budget; fresh revision-two controls +use 180 seconds total. No clock, database row or task budget is edited. Built +consumer images, project containers, volumes and networks are removed on exit. The focused +Action and logs report the actual package pairs, original IDs and outcomes; +constructed verifier tests alone do not qualify published-package replay. + +## Retry, deadlines and worker loss + This experiment runs every PHP, Python and Rust workflow/activity direction. Each workflow language calls an activity in each of the three languages, including its own language, for nine directions. diff --git a/polyglot/activities/compose-sdk-upgrades.yml b/polyglot/activities/compose-sdk-upgrades.yml new file mode 100644 index 0000000..b643481 --- /dev/null +++ b/polyglot/activities/compose-sdk-upgrades.yml @@ -0,0 +1,7 @@ +services: + php-same-workflow-worker: + command: ["php", "/app/activity_upgrades.php"] + python-workflow-worker: + command: ["python", "/app/activity_upgrades.py"] + rust-workflow-worker: + entrypoint: ["activity-upgrade-worker"] diff --git a/polyglot/php_worker/Dockerfile b/polyglot/php_worker/Dockerfile index 1b4fc3a..2baed69 100644 --- a/polyglot/php_worker/Dockerfile +++ b/polyglot/php_worker/Dockerfile @@ -30,6 +30,7 @@ RUN test -n "$DURABLE_WORKFLOW_PHP_SDK_VERSION" \ COPY worker.php ./ COPY child_workflows.php ./ COPY timer_upgrades.php ./ +COPY activity_upgrades.php ./ COPY php_created_rust_schedule.php ./ COPY update_client.php ./ COPY cancellation_client.php ./ diff --git a/polyglot/php_worker/activity_upgrades.php b/polyglot/php_worker/activity_upgrades.php new file mode 100644 index 0000000..3fce59a --- /dev/null +++ b/polyglot/php_worker/activity_upgrades.php @@ -0,0 +1,41 @@ +registerWorkflow('polyglot.php.activity-upgrade', + static function (WorkflowContext $context, array $request) use ($revision): array { + $version = $context->getVersion('sample-activity-application', 1, $revision === 'v1' ? 1 : 2); + $result = $context->activity('polyglot.'.$request['activity_runtime'].'.upgrade-activity', + [['case_id' => $request['case_id'], 'version' => $version, 'value' => $request['value']]], + ['queue' => 'polyglot-'.$request['activity_runtime'], + 'retry_policy' => ['max_attempts' => $version + 1, 'backoff_seconds' => $version === 1 ? [1] : [1, 2]], + 'start_to_close_timeout' => 30, 'schedule_to_close_timeout' => $version === 1 ? 120 : 180]); + + return ['workflow_runtime' => 'php', 'application_revision' => $revision, + 'version' => $version, 'activity' => $result]; + }); +// Previous workers deliberately author pending work without an activity handler. +if ($revision === 'v2') { + $worker->registerActivity('polyglot.php.upgrade-activity', + static function (ActivityContext $context, array $request) use ($revision): array { + return ['activity_runtime' => 'php', 'application_revision' => $revision, + 'sdk_version' => 'durable-workflow-php/'.InstalledVersions::getPrettyVersion('durable-workflow/sdk'), + 'attempt_number' => $context->attemptNumber, 'request' => $request]; + }); +} +$worker->run(2); diff --git a/polyglot/python_worker/scripts/sdk_activity_upgrades.py b/polyglot/python_worker/scripts/sdk_activity_upgrades.py new file mode 100644 index 0000000..189ef55 --- /dev/null +++ b/polyglot/python_worker/scripts/sdk_activity_upgrades.py @@ -0,0 +1,250 @@ +"""Replay nine pending remote-activity directions across SDK/application replacement.""" + +import argparse +import asyncio +import json +import os +from datetime import datetime, timedelta + +RUNTIMES = ("php", "python", "rust") +DIRECTIONS = [(workflow, activity) for workflow in RUNTIMES for activity in RUNTIMES] +VALUE = {"bigint": 9223372036854775805, "optional": None, "ready": True, + "fraction": 1.25, "note": "café ✓", "items": [0, False, None, "017"]} +TERMINALS = {"WorkflowCompleted", "WorkflowFailed", "WorkflowCancelled", "WorkflowTimedOut", "WorkflowTerminated"} + + +def require(condition, message): + if not condition: + raise RuntimeError(message) + + +def same(left, right): + return json.dumps(left, sort_keys=True, allow_nan=False) == json.dumps(right, sort_keys=True, allow_nan=False) + + +def events(history, kind): + return [event for event in history["events"] if event["event_type"] == kind] + + +def one(history, kind): + found = events(history, kind) + require(len(found) == 1, f"Expected one {kind}, got {len(found)}.") + return found[0] + + +def timestamp(value): + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + require(parsed.utcoffset() is not None, "Event clock has no timezone.") + return parsed + + +def emit(**record): + print(json.dumps(record, sort_keys=True), flush=True) + + +def check_commands(history, activity, version): + marker = one(history, "VersionMarkerRecorded")["payload"] + expected = {"change_id": "sample-activity-application", "sequence": 1, + "version": version, "min_supported": 1, "max_supported": version} + require(all(same(marker.get(key), value) for key, value in expected.items()), + "Activity replacement changed the recorded version branch or authored bounds.") + scheduled = one(history, "ActivityScheduled")["payload"] + snapshot = scheduled.get("activity", {}) + require(type(scheduled.get("sequence")) is int and scheduled["sequence"] == 2 + and scheduled.get("activity_type") == f"polyglot.{activity}.upgrade-activity" + and isinstance(scheduled.get("activity_execution_id"), str) + and bool(scheduled["activity_execution_id"]) + and snapshot.get("id") == scheduled["activity_execution_id"] + and snapshot.get("queue") == f"polyglot-{activity}", + "Activity command changed its original sequence, execution or target queue.") + policy = snapshot.get("retry_policy", {}) + expected_policy = {"max_attempts": version + 1, "backoff_seconds": [1] if version == 1 else [1, 2], + "start_to_close_timeout": 30, "schedule_to_close_timeout": 120 if version == 1 else 180} + require(all(same(policy.get(key), value) for key, value in expected_policy.items()), + "Activity command lost the selected complete retry/timeout policy.") + kinds = [event["event_type"] for event in history["events"]] + require(kinds.index("VersionMarkerRecorded") < kinds.index("ActivityScheduled"), + "Activity was scheduled before its version decision.") + return scheduled + + +def check_completed(original, execution, history, result, current_workers, targets, restarted_at, decode): + workflow, activity, version = original["workflow_runtime"], original["activity_runtime"], original["version"] + require(execution.workflow_id == original["workflow_id"] and execution.run_id == original["run_id"] + and execution.status == "completed", "Replacement did not complete the same original run.") + prefix = original["history_prefix"] + require(bool(prefix) and same(history["events"][:len(prefix)], prefix), + "Replacement changed the complete original history prefix.") + scheduled = check_commands(history, activity, version) + require(same(scheduled, original["scheduled"]), "Replacement changed the original activity snapshot.") + worker = next(item for item in current_workers if item["runtime"] == workflow) + require(worker.get("sdk_version") == f"durable-workflow-{workflow}/{targets[workflow]}", + "Workflow successor uses another SDK package.") + if version == 1: + require(worker["worker_id"] != original["worker_id"], "Pending run did not use a cold replacement.") + if workflow == "python": + started = one(history, "WorkflowStarted")["payload"] + fingerprint = started.get("workflow_definition_fingerprint") + require(started.get("workflow_definition_fingerprint_source") == "worker" and bool(fingerprint) + and worker["workflow_definition_fingerprints"].get("polyglot.python.activity-upgrade") == fingerprint, + "Replacement rebound the original Python workflow definition.") + started, completed = one(history, "ActivityStarted"), one(history, "ActivityCompleted") + started_payload, completed_payload = started["payload"], completed["payload"] + require(started_payload.get("activity_execution_id") == scheduled["activity_execution_id"] + and completed_payload.get("activity_execution_id") == scheduled["activity_execution_id"] + and isinstance(started_payload.get("activity_attempt_id"), str) + and bool(started_payload["activity_attempt_id"]) + and started_payload["activity_attempt_id"] == completed_payload.get("activity_attempt_id") + and type(started_payload.get("attempt_number")) is int and started_payload["attempt_number"] == 1, + "Activity was duplicated or its original attempt/result identity changed.") + require(not events(history, "ActivityRetryScheduled") and not events(history, "ActivityFailed"), + "Pending activity unexpectedly retried or failed.") + require(timestamp(started["timestamp"]) >= timestamp(restarted_at), + "Activity executed before the current-package workers started.") + total = scheduled["activity"]["retry_policy"]["schedule_to_close_timeout"] + scheduled_at = timestamp(one(history, "ActivityScheduled")["timestamp"]) + require(scheduled_at <= timestamp(started["timestamp"]) <= timestamp(completed["timestamp"]) + <= scheduled_at + timedelta(seconds=total), "Activity escaped its original total deadline.") + expected_activity = {"activity_runtime": activity, "application_revision": "v2", + "sdk_version": f"durable-workflow-{activity}/{targets[activity]}", "attempt_number": 1, + "request": {"case_id": original["workflow_id"], "version": version, "value": VALUE}} + expected = {"workflow_runtime": workflow, "application_revision": "v2", "version": version, + "activity": expected_activity} + require(same(result, expected) and same(decode(completed_payload["result"]), expected_activity), + "Current SDK callbacks changed the recorded branch or exact typed activity/workflow result.") + terminals = [event for event in history["events"] if event["event_type"] in TERMINALS] + require(len(terminals) == 1 and terminals[0]["event_type"] == "WorkflowCompleted" + and same(decode(terminals[0]["payload"]["output"]), result) + and timestamp(terminals[0]["timestamp"]) >= timestamp(completed["timestamp"]), + "Original workflow has another terminal event or result.") + + +def original_records(scenario): + rows = [json.loads(line) for line in os.environ["DURABLE_WORKFLOW_TIMER_PENDING"].splitlines()] + expected = set(DIRECTIONS) + actual = [(row.get("workflow_runtime"), row.get("activity_runtime")) for row in rows] + require(len(rows) == len(expected) and set(actual) == expected + and all(row.get("phase") == "pending" and row.get("scenario") == scenario for row in rows) + and len({row.get("run_id") for row in rows}) == len(rows), + "Pending upgrade set omits, repeats or substitutes an original direction.") + return rows + + +def ready_snapshot(roster, common, selected): + workers = [] + current = selected == common.sdk_versions("DURABLE_WORKFLOW_TIMER_SDK_TARGETS") + for runtime in RUNTIMES: + matches = common.active_workers(roster, runtime) + require(len(matches) <= 1, "Ambiguous active upgrade workers.") + if not matches: + return None + worker = matches[0] + if (f"polyglot.{runtime}.activity-upgrade" not in (worker.supported_workflow_types or []) + or (worker.raw or {}).get("sdk_version") != f"durable-workflow-{runtime}/{selected[runtime]}"): + return None + advertised = set(worker.supported_activity_types or []) + if (f"polyglot.{runtime}.upgrade-activity" in advertised) != current: + return None + workers.append(common.worker_snapshot(worker)) + return workers + + +async def ready_workers(client, common, scenario, selected): + deadline = asyncio.get_running_loop().time() + 20 + while True: + workers = ready_snapshot(await client.list_workers(), common, selected) + if workers is not None: + emit(phase="workers_ready", scenario=scenario, project=common.required("DURABLE_WORKFLOW_TIMER_ID"), workers=workers) + return + if asyncio.get_running_loop().time() >= deadline: + raise TimeoutError("Exact package/role upgrade roster did not become ready.") + await asyncio.sleep(0.25) + + +async def observe(client, workflow_id, workflow): + handle = client.get_workflow_handle(workflow_id, workflow_type=f"polyglot.{workflow}.activity-upgrade") + execution = await handle.describe() + return handle, execution, await client.get_history(workflow_id, execution.run_id) + + +async def start_original(client, common, scenario, selected, workflow, activity): + fresh = scenario.endswith("-fresh") + version = 2 if fresh else 1 + workflow_id = f"{common.required('DURABLE_WORKFLOW_TIMER_ID')}-{scenario}-{workflow}-{activity}" + worker = next(row for row in common.retirement_snapshot() if row["runtime"] == workflow) + await client.start_workflow(workflow_type=f"polyglot.{workflow}.activity-upgrade", workflow_id=workflow_id, + task_queue=f"polyglot-{workflow}", input=[{"case_id": workflow_id, "activity_runtime": activity, "value": VALUE}]) + deadline = asyncio.get_running_loop().time() + 20 + while True: + _, execution, history = await observe(client, workflow_id, workflow) + require(fresh or not events(history, "ActivityStarted"), + f"Activity started before replacement: {workflow_id}, run={execution.run_id}, status={execution.status}.") + if events(history, "ActivityScheduled") and (fresh or execution.status == "waiting"): + break + require(execution.status not in {"failed", "cancelled", "terminated"}, "Original closed before activity scheduling.") + if asyncio.get_running_loop().time() >= deadline: + raise TimeoutError(f"Pending activity did not arrive: {workflow_id}, status={execution.status}.") + await asyncio.sleep(0.25) + scheduled = check_commands(history, activity, version) + if not fresh: + require(execution.status == "waiting" and not events(history, "ActivityStarted"), + "Previous-package activity was not genuinely pending before replacement.") + emit(phase="pending", scenario=scenario, workflow_runtime=workflow, activity_runtime=activity, + workflow_id=workflow_id, run_id=execution.run_id, version=version, worker_id=worker["worker_id"], + sdk_version=selected[workflow], scheduled=scheduled, history_prefix=history["events"]) + + +async def verify_original(client, common, scenario, original, targets, previous): + from durable_workflow import serializer + + fresh = scenario.endswith("-fresh") + workflow = original["workflow_runtime"] + require(type(original.get("version")) is int and original["version"] == (2 if fresh else 1), + "Original observation changed its old/fresh version branch.") + require(original["sdk_version"] == (targets if fresh else previous)[workflow], "Original used the wrong SDK package.") + handle, _, _ = await observe(client, original["workflow_id"], workflow) + result = await handle.result(timeout=120, poll_interval=0.25) + _, execution, history = await observe(client, original["workflow_id"], workflow) + check_completed(original, execution, history, result, common.retirement_snapshot(), targets, + common.required("DURABLE_WORKFLOW_WORKER_RESTART_AT"), lambda value: serializer.decode_envelope(value, codec="avro")) + emit(phase="verified", scenario=scenario, workflow_runtime=workflow, activity_runtime=original["activity_runtime"], + workflow_id=execution.workflow_id, run_id=execution.run_id, result=result, + original_sdk=original["sdk_version"], current_sdk=targets[workflow], + marker=one(history, "VersionMarkerRecorded")["payload"], scheduled=original["scheduled"], + history_events=[row["event_type"] for row in history["events"]]) + + +async def main(phase, scenario): + from durable_workflow import Client + import sdk_timers as common + + selected = common.sdk_versions("DURABLE_WORKFLOW_TIMER_SDK_VERSIONS") + async with Client(common.required("DURABLE_WORKFLOW_SERVER_URL"), namespace="default", + control_token=common.required("DURABLE_WORKFLOW_AUTH_TOKEN")) as client: + if phase == "retire": + await common.retire_workers(client, scenario) + elif phase == "ready": + await ready_workers(client, common, scenario, selected) + elif phase == "start": + await asyncio.gather(*(start_original(client, common, scenario, selected, *direction) for direction in DIRECTIONS)) + elif phase == "verify": + targets = common.sdk_versions("DURABLE_WORKFLOW_TIMER_SDK_TARGETS") + previous = common.sdk_versions("DURABLE_WORKFLOW_TIMER_SDK_PREVIOUS") + require(selected == targets and all(previous[runtime] != targets[runtime] for runtime in RUNTIMES), + "Verification did not select distinct previous/current packages.") + await asyncio.gather(*(verify_original(client, common, scenario, original, targets, previous) + for original in original_records(scenario))) + elif phase == "diagnose": + for workflow, activity in DIRECTIONS: + workflow_id = f"{common.required('DURABLE_WORKFLOW_TIMER_ID')}-{scenario}-{workflow}-{activity}" + _, execution, history = await observe(client, workflow_id, workflow) + emit(phase="diagnostic", workflow_id=workflow_id, run_id=execution.run_id, status=execution.status, + history_events=[row["event_type"] for row in history["events"]]) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("phase", choices=("ready", "retire", "start", "verify", "diagnose")) + parser.add_argument("scenario", choices=("activity-upgrade", "activity-upgrade-fresh")) + args = parser.parse_args() + asyncio.run(main(args.phase, args.scenario)) diff --git a/polyglot/python_worker/tests/test_sdk_activity_upgrades.py b/polyglot/python_worker/tests/test_sdk_activity_upgrades.py new file mode 100644 index 0000000..7f73328 --- /dev/null +++ b/polyglot/python_worker/tests/test_sdk_activity_upgrades.py @@ -0,0 +1,219 @@ +import copy +import importlib.util +import json +import os +from pathlib import Path +import types +import unittest +from unittest.mock import AsyncMock, patch + +path = Path(__file__).parents[1] / "scripts" / "sdk_activity_upgrades.py" +spec = importlib.util.spec_from_file_location("activity_upgrades", path) +upgrades = importlib.util.module_from_spec(spec) +spec.loader.exec_module(upgrades) + +TARGETS = {"php": "2.5.1", "python": "2.6.0", "rust": "3.5.5"} + + +def event(kind, payload, seconds=0): + return {"event_type": kind, "payload": payload, "timestamp": f"2026-10-19T00:00:{seconds:02d}+00:00"} + + +class ActivityUpgradeEvidenceTest(unittest.TestCase): + def case(self, workflow="python", activity="rust", version=1): + self.original = {"workflow_runtime": workflow, "activity_runtime": activity, "version": version, + "workflow_id": f"original-{workflow}-{activity}", "run_id": "original-run", + "worker_id": "previous-worker"} + activity_result = {"activity_runtime": activity, "application_revision": "v2", + "sdk_version": f"durable-workflow-{activity}/{TARGETS[activity]}", "attempt_number": 1, + "request": {"case_id": self.original["workflow_id"], "version": version, "value": copy.deepcopy(upgrades.VALUE)}} + self.result = {"workflow_runtime": workflow, "application_revision": "v2", "version": version, + "activity": activity_result} + policy = {"max_attempts": version + 1, "backoff_seconds": [1] if version == 1 else [1, 2], + "start_to_close_timeout": 30, "schedule_to_close_timeout": 120 if version == 1 else 180} + scheduled = {"activity_execution_id": "original-activity", "sequence": 2, + "activity_type": f"polyglot.{activity}.upgrade-activity", + "activity": {"id": "original-activity", "queue": f"polyglot-{activity}", "retry_policy": policy}} + prefix = [event("WorkflowStarted", {"workflow_definition_fingerprint": "original-definition", + "workflow_definition_fingerprint_source": "worker"}), + event("VersionMarkerRecorded", {"change_id": "sample-activity-application", "sequence": 1, + "version": version, "min_supported": 1, "max_supported": version}), + event("ActivityScheduled", scheduled)] + self.original.update(scheduled=copy.deepcopy(scheduled), history_prefix=copy.deepcopy(prefix)) + attempt = {"activity_execution_id": "original-activity", "activity_attempt_id": "first-attempt", "attempt_number": 1} + self.history = {"events": [*prefix, event("ActivityStarted", attempt, 10), + event("ActivityCompleted", {**attempt, "result": copy.deepcopy(activity_result)}, 11), + event("WorkflowCompleted", {"output": copy.deepcopy(self.result)}, 12)]} + self.execution = types.SimpleNamespace(workflow_id=self.original["workflow_id"], run_id="original-run", status="completed") + self.workers = [{"runtime": language, "worker_id": f"current-{language}", + "sdk_version": f"durable-workflow-{language}/{TARGETS[language]}", + "workflow_definition_fingerprints": {"polyglot.python.activity-upgrade": "original-definition"}} + for language in upgrades.RUNTIMES] + + def verify(self): + upgrades.check_completed(self.original, self.execution, self.history, self.result, + self.workers, TARGETS, "2026-10-19T00:00:09Z", lambda value: value) + + def test_all_nine_directions_accept_old_branch_and_fresh_current_branch(self): + for workflow, activity in upgrades.DIRECTIONS: + for version in (1, 2): + with self.subTest(workflow=workflow, activity=activity, version=version): + self.case(workflow, activity, version) + self.verify() + + def test_rewritten_original_prefix_is_rejected_even_with_same_schedule(self): + self.case() + self.history["events"][0]["payload"]["workflow_definition_fingerprint_source"] = "replacement" + with self.assertRaisesRegex(RuntimeError, "history prefix"): + self.verify() + + def test_changed_marker_bounds_and_rebased_command_sequence_refuse(self): + for event_index, field, value in ((1, "max_supported", 2), (1, "sequence", 7), + (1, "version", True), (2, "sequence", 3)): + with self.subTest(field=field, value=value): + self.case() + self.history["events"][event_index]["payload"][field] = value + # Also replace the captured prefix to test the independent command contract. + self.original["history_prefix"] = copy.deepcopy(self.history["events"][:3]) + with self.assertRaises(RuntimeError): + self.verify() + + def test_every_authored_policy_field_remains_checked(self): + for field, value in (("max_attempts", 3), ("backoff_seconds", [1, 2]), + ("start_to_close_timeout", 60), ("schedule_to_close_timeout", 180)): + with self.subTest(field=field): + self.case() + self.history["events"][2]["payload"]["activity"]["retry_policy"][field] = value + self.original["history_prefix"] = copy.deepcopy(self.history["events"][:3]) + with self.assertRaisesRegex(RuntimeError, "policy"): + self.verify() + + def test_changed_original_run_attempt_or_worker_is_rejected(self): + for field in ("run", "execution", "attempt", "worker", "package", "definition"): + with self.subTest(field=field): + self.case() + if field == "run": + self.execution.run_id = "another-run" + elif field == "execution": + self.history["events"][4]["payload"]["activity_execution_id"] = "another-activity" + elif field == "attempt": + self.history["events"][4]["payload"]["activity_attempt_id"] = "another-attempt" + elif field == "worker": + self.workers[1]["worker_id"] = self.original["worker_id"] + elif field == "package": + self.workers[1]["sdk_version"] = "durable-workflow-python/2.5.15" + else: + self.workers[1]["workflow_definition_fingerprints"]["polyglot.python.activity-upgrade"] = "another-definition" + with self.assertRaises(RuntimeError): + self.verify() + + def test_early_activity_or_completion_beyond_original_budget_refuses(self): + for value in ("2026-10-19T00:00:08Z", "2026-10-19T00:03:00Z"): + self.case() + self.history["events"][3 if value.endswith("08Z") else 4]["timestamp"] = value + with self.assertRaises(RuntimeError): + self.verify() + + def test_scalar_substitutions_do_not_pass_python_equality(self): + for field, value in (("bigint", float(upgrades.VALUE["bigint"])), ("ready", 1), + ("optional", "null"), ("fraction", 1), ("items", [0, 0, None, "017"])): + with self.subTest(field=field): + self.case() + self.result["activity"]["request"]["value"][field] = value + with self.assertRaisesRegex(RuntimeError, "typed"): + self.verify() + + def test_changed_durable_result_duplicate_completion_and_unexpected_retry_refuse(self): + for kind in ("result", "completion", "retry"): + self.case() + if kind == "result": + self.history["events"][4]["payload"]["result"]["sdk_version"] = "old-package" + elif kind == "completion": + self.history["events"].append(copy.deepcopy(self.history["events"][-1])) + else: + self.history["events"].append(event("ActivityRetryScheduled", {}, 9)) + with self.assertRaises(RuntimeError): + self.verify() + + def test_direction_set_requires_nine_distinct_originals(self): + records = [{"phase": "pending", "scenario": "activity-upgrade", "workflow_runtime": workflow, + "activity_runtime": activity, "run_id": f"run-{workflow}-{activity}"} + for workflow, activity in upgrades.DIRECTIONS] + with patch.dict(os.environ, {"DURABLE_WORKFLOW_TIMER_PENDING": "\n".join(map(json.dumps, records))}): + self.assertEqual(len(upgrades.original_records("activity-upgrade")), 9) + for changed in (records[:-1], records[:-1] + [records[0]], [*records, records[0]]): + with patch.dict(os.environ, {"DURABLE_WORKFLOW_TIMER_PENDING": "\n".join(map(json.dumps, changed))}): + with self.assertRaises(RuntimeError): + upgrades.original_records("activity-upgrade") + + def roster_case(self, current): + selected = TARGETS if current else {runtime: "1.0.0" for runtime in upgrades.RUNTIMES} + workers = [types.SimpleNamespace(runtime=runtime, + supported_workflow_types=[f"polyglot.{runtime}.activity-upgrade"], + supported_activity_types=[f"polyglot.{runtime}.upgrade-activity"] if current else [], + raw={"sdk_version": f"durable-workflow-{runtime}/{selected[runtime]}"}) + for runtime in upgrades.RUNTIMES] + common = types.SimpleNamespace( + sdk_versions=lambda key: TARGETS, + active_workers=lambda roster, runtime: [worker for worker in roster if worker.runtime == runtime], + worker_snapshot=lambda worker: {"runtime": worker.runtime}) + return workers, common, selected + + def test_previous_roster_cannot_dispatch_activities_and_current_roster_must_dispatch_them(self): + for current in (False, True): + with self.subTest(current=current): + workers, common, selected = self.roster_case(current) + self.assertEqual(len(upgrades.ready_snapshot(workers, common, selected)), 3) + workers[0].supported_activity_types = [] if current else ["polyglot.php.upgrade-activity"] + self.assertIsNone(upgrades.ready_snapshot(workers, common, selected)) + + def test_wrong_installed_package_or_missing_workflow_refuses_roster_readiness(self): + for wrong in ("package", "workflow", "absent"): + with self.subTest(wrong=wrong): + workers, common, selected = self.roster_case(True) + if wrong == "package": + workers[0].raw["sdk_version"] = "durable-workflow-php/1.0.0" + elif wrong == "workflow": + workers[0].supported_workflow_types = [] + else: + workers.pop() + self.assertIsNone(upgrades.ready_snapshot(workers, common, selected)) + + def test_ambiguous_registered_workers_are_refused(self): + workers, common, selected = self.roster_case(True) + with self.assertRaisesRegex(RuntimeError, "Ambiguous"): + upgrades.ready_snapshot([*workers, workers[0]], common, selected) + + +class PendingCheckpointRaceTest(unittest.IsolatedAsyncioTestCase): + async def test_history_can_observe_schedule_after_the_earlier_describe_read(self): + fixture = ActivityUpgradeEvidenceTest() + fixture.case() + history = {"events": fixture.original["history_prefix"]} + running = types.SimpleNamespace(status="running", run_id="original-run") + waiting = types.SimpleNamespace(status="waiting", run_id="original-run") + common = types.SimpleNamespace(required=lambda key: "owned-project", + retirement_snapshot=lambda: [{"runtime": "python", "worker_id": "previous-worker"}]) + client = types.SimpleNamespace(start_workflow=AsyncMock()) + with patch.object(upgrades, "observe", AsyncMock(side_effect=[(None, running, history), (None, waiting, history)])) as read, \ + patch.object(upgrades.asyncio, "sleep", AsyncMock()), patch.object(upgrades, "emit") as output: + await upgrades.start_original(client, common, "activity-upgrade", {"python": "2.5.15"}, "python", "rust") + self.assertEqual(read.await_count, 2) + self.assertEqual(output.call_args.kwargs["run_id"], "original-run") + self.assertEqual(output.call_args.kwargs["history_prefix"], history["events"]) + + async def test_an_activity_started_before_replacement_still_refuses(self): + fixture = ActivityUpgradeEvidenceTest() + fixture.case() + history = {"events": fixture.history["events"][:4]} + waiting = types.SimpleNamespace(status="waiting", run_id="original-run") + common = types.SimpleNamespace(required=lambda key: "owned-project", + retirement_snapshot=lambda: [{"runtime": "python", "worker_id": "previous-worker"}]) + with patch.object(upgrades, "observe", AsyncMock(return_value=(None, waiting, history))), \ + patch.object(upgrades, "emit"), self.assertRaisesRegex(RuntimeError, "pending|before replacement"): + await upgrades.start_original(types.SimpleNamespace(start_workflow=AsyncMock()), common, + "activity-upgrade", {"python": "2.5.15"}, "python", "rust") + + +if __name__ == "__main__": + unittest.main() diff --git a/polyglot/python_worker/tests/test_sdk_timer_packages.py b/polyglot/python_worker/tests/test_sdk_timer_packages.py index c7c30cd..bde1b22 100644 --- a/polyglot/python_worker/tests/test_sdk_timer_packages.py +++ b/polyglot/python_worker/tests/test_sdk_timer_packages.py @@ -36,7 +36,7 @@ sys.exit(1) if "run" not in args: sys.exit(0) -if "/app/scripts/sdk_timers.py" in args: +if any(script in args for script in ("/app/scripts/sdk_timers.py", "/app/scripts/sdk_activity_upgrades.py")): print(json.dumps({"phase": args[-2], "scenario": args[-1]})) elif "php-same-workflow-worker" in args: print(json.dumps({"sdk_version": versions["php"], "reference": "constructed"})) @@ -129,6 +129,35 @@ def test_combined_change_rejects_a_mislabelled_previous_package_before_admission and record["args"][-2] == "start" for record in records)) self.assertTrue(any("down" in record["args"] for record in records)) + def test_pending_activity_upgrade_keeps_all_originals_before_one_cold_package_replacement(self): + result, records = self.run_control(mode="--sdk-activity-upgrades") + self.assertEqual(result.returncode, 0, result.stderr) + builds = [row for row in records if "build" in row["args"]] + self.assertEqual([row["versions"] for row in builds], [CURRENT, PREVIOUS]) + observer = "/app/scripts/sdk_activity_upgrades.py" + starts = [row for row in records if observer in row["args"] and row["args"][-2] == "start"] + self.assertEqual([row["args"][-1] for row in starts], ["activity-upgrade", "activity-upgrade-fresh"]) + self.assertEqual([row["versions"] for row in starts], [PREVIOUS, CURRENT]) + self.assertEqual([row["application"] for row in starts], ["v1", "v2"]) + phases = [row["args"][-2] for row in records if observer in row["args"]] + self.assertEqual(phases, ["ready", "start", "retire", "ready", "verify", "start", "verify"]) + killed = [index for index, row in enumerate(records) if "kill" in row["args"]] + first = records.index(starts[0]) + current = next(index for index, row in enumerate(records) + if "up" in row["args"] and row["generation"] == "current") + self.assertEqual(len(killed), 1) + self.assertLess(first, killed[0]) + self.assertLess(killed[0], current) + self.assertTrue(all(any(arg.endswith("activities/compose-sdk-upgrades.yml") for arg in row["args"]) + for row in records if row["args"][0] == "compose")) + self.assertTrue(any("down" in row["args"] for row in records)) + + def test_pending_activity_mislabelled_package_refuses_before_original_admission(self): + result, records = self.run_control(mode="--sdk-activity-upgrades", wrong_package=True) + self.assertNotEqual(result.returncode, 0) + self.assertFalse(any("/app/scripts/sdk_activity_upgrades.py" in row["args"] for row in records)) + self.assertTrue(any("down" in row["args"] for row in records)) + class PackageUpgradeEvidenceTest(unittest.IsolatedAsyncioTestCase): def setUp(self): diff --git a/polyglot/python_workflow/Dockerfile b/polyglot/python_workflow/Dockerfile index 99c62c6..2a49e36 100644 --- a/polyglot/python_workflow/Dockerfile +++ b/polyglot/python_workflow/Dockerfile @@ -18,6 +18,7 @@ RUN set -eux; \ COPY workflow.py ./workflow.py COPY child_workflows.py ./child_workflows.py COPY timer_upgrades.py ./timer_upgrades.py +COPY activity_upgrades.py ./activity_upgrades.py COPY replay_fixtures ./replay_fixtures RUN python workflow.py --replay-fixtures diff --git a/polyglot/python_workflow/activity_upgrades.py b/polyglot/python_workflow/activity_upgrades.py new file mode 100644 index 0000000..d9de8ee --- /dev/null +++ b/polyglot/python_workflow/activity_upgrades.py @@ -0,0 +1,64 @@ +"""A stable registered wrapper with replaceable versioned business logic.""" + +import asyncio +import importlib.metadata +import os +import socket + +from durable_workflow import Client, Worker, activity, workflow + +REVISION = None + + +def original_application(context): + return (yield context.get_version("sample-activity-application", 1, 1)) + + +def upgraded_application(context): + return (yield context.get_version("sample-activity-application", 1, 2)) + + +APPLICATION = None + + +@workflow.defn(name="polyglot.python.activity-upgrade") +class ActivityUpgrade: + def run(self, context, request): + version = yield from APPLICATION(context) + result = yield context.schedule_activity( + f"polyglot.{request['activity_runtime']}.upgrade-activity", + [{"case_id": request["case_id"], "version": version, "value": request["value"]}], + queue=f"polyglot-{request['activity_runtime']}", + retry_policy=workflow.ActivityRetryPolicy(max_attempts=version + 1, + backoff_seconds=[1] if version == 1 else [1, 2]), + start_to_close_timeout=30, + schedule_to_close_timeout=120 if version == 1 else 180, + ) + return {"workflow_runtime": "python", "application_revision": REVISION, + "version": version, "activity": result} + + +@activity.defn(name="polyglot.python.upgrade-activity") +def upgraded_activity(request): + return {"activity_runtime": "python", "application_revision": REVISION, + "sdk_version": "durable-workflow-python/" + importlib.metadata.version("durable-workflow"), + "attempt_number": activity.context().info.attempt_number, "request": request} + + +async def main(): + global REVISION, APPLICATION + REVISION = os.environ["POLYGLOT_TIMER_APPLICATION"] + if REVISION not in {"v1", "v2"}: + raise ValueError("Select activity application v1 or v2.") + APPLICATION = original_application if REVISION == "v1" else upgraded_application + async with Client(os.environ["DURABLE_WORKFLOW_SERVER_URL"], + token=os.environ["DURABLE_WORKFLOW_AUTH_TOKEN"], namespace="default") as client: + worker = Worker(client, task_queue="polyglot-python", workflows=[ActivityUpgrade], + activities=[] if REVISION == "v1" else [upgraded_activity], + worker_id="activity-upgrade-python-" + socket.gethostname(), poll_timeout=2, + shutdown_timeout=10) + await worker.run() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/polyglot/rust_worker/Dockerfile b/polyglot/rust_worker/Dockerfile index 8ad9b0b..ff92036 100644 --- a/polyglot/rust_worker/Dockerfile +++ b/polyglot/rust_worker/Dockerfile @@ -32,6 +32,7 @@ RUN apt-get update \ COPY --from=build /build/target/release/durable-workflow-polyglot-rust-worker /usr/local/bin/polyglot-rust-worker COPY --from=build /build/target/release/task_codec_rejection_probe /usr/local/bin/task-codec-rejection-probe COPY --from=build /build/target/release/activity_recovery_worker /usr/local/bin/activity-recovery-worker +COPY --from=build /build/target/release/activity_upgrade_worker /usr/local/bin/activity-upgrade-worker USER 65532:65532 diff --git a/polyglot/rust_worker/src/bin/activity_upgrade_worker.rs b/polyglot/rust_worker/src/bin/activity_upgrade_worker.rs new file mode 100644 index 0000000..67caf11 --- /dev/null +++ b/polyglot/rust_worker/src/bin/activity_upgrade_worker.rs @@ -0,0 +1,45 @@ +use durable_workflow::{json, ActivityOptions, ActivityRetryPolicy, Client, Error, Result, Worker}; +use std::{env, time::Duration}; + +#[tokio::main] +async fn main() -> Result<()> { + if env::var("POLYGLOT_RUST_MODE").ok().as_deref() == Some("sdk-metadata") { + println!("{}", json!({"sdk_identity": durable_workflow::SDK_VERSION})); + return Ok(()); + } + let revision = match env::var("POLYGLOT_TIMER_APPLICATION").ok().as_deref() { + Some("v1") => "v1", + Some("v2") => "v2", + _ => return Err(Error::WorkerLoop("select activity application v1 or v2".into())), + }; + let client = Client::builder(env::var("DURABLE_WORKFLOW_SERVER_URL").expect("server URL")) + .token(Some(env::var("DURABLE_WORKFLOW_AUTH_TOKEN").expect("token"))) + .namespace("default") + .build()?; + let mut worker = Worker::new(client, "polyglot-rust") + .worker_id(format!("activity-upgrade-rust-{}", env::var("HOSTNAME").expect("hostname"))) + .poll_timeout(Duration::from_secs(2)); + worker.register_workflow("polyglot.rust.activity-upgrade", move |context, args| async move { + let request = args[0].clone(); + let runtime = request["activity_runtime"].as_str().expect("activity runtime"); + let version = context.get_version("sample-activity-application", 1, if revision == "v1" { 1 } else { 2 })?; + let backoff = if version == 1 { vec![Duration::from_secs(1)] } + else { vec![Duration::from_secs(1), Duration::from_secs(2)] }; + let options = ActivityOptions::new() + .task_queue(format!("polyglot-{runtime}")) + .retry_policy(ActivityRetryPolicy::new((version + 1) as u32).backoff_intervals(backoff)) + .start_to_close_timeout(Duration::from_secs(30)) + .schedule_to_close_timeout(Duration::from_secs(if version == 1 { 120 } else { 180 })); + let activity = context.activity_with_options(format!("polyglot.{runtime}.upgrade-activity"), options, + json!([{"case_id":request["case_id"],"version":version,"value":request["value"]}])).await?; + Ok(json!({"workflow_runtime":"rust","application_revision":revision,"version":version,"activity":activity})) + }); + if revision == "v2" { + worker.register_activity("polyglot.rust.upgrade-activity", move |context, args| async move { + Ok(json!({"activity_runtime":"rust","application_revision":revision, + "sdk_version":durable_workflow::SDK_VERSION,"attempt_number":context.attempt_number, + "request":args[0]})) + }); + } + worker.run().await +} diff --git a/scripts/sdk-activity-upgrades.sh b/scripts/sdk-activity-upgrades.sh new file mode 100755 index 0000000..f4fb911 --- /dev/null +++ b/scripts/sdk-activity-upgrades.sh @@ -0,0 +1,14 @@ +#!/usr/bin/env bash +set -euo pipefail + +if [[ "${1:-}" == --help ]]; then + printf '%s\n' 'Usage: scripts/sdk-activity-upgrades.sh' \ + 'Runs eighteen SDK/application upgrade originals across all nine PHP/Python/Rust workflow/activity directions.' \ + 'Builds both package generations before admission; previous workers leave activities unclaimed, cold current workers complete the same runs.' \ + 'Requires Docker Compose, Python 3 and scripts/resolve-current-artifacts.sh assignments. All project resources are removed on exit.' + exit 0 +fi +[[ $# == 0 ]] || exit 2 +repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +export SDK_TIMERS_COMPOSE_PROJECT_NAME="${SDK_ACTIVITY_UPGRADES_COMPOSE_PROJECT_NAME:-sample-app-sdk-activity-upgrades-$(date -u +%Y%m%d%H%M%S)}" +exec "$repo_root/scripts/sdk-timers.sh" --sdk-activity-upgrades diff --git a/scripts/sdk-timers.sh b/scripts/sdk-timers.sh index 55f9cd5..7cfd468 100755 --- a/scripts/sdk-timers.sh +++ b/scripts/sdk-timers.sh @@ -2,13 +2,14 @@ set -euo pipefail if [[ "${1:-}" == --help ]]; then - printf '%s\n' 'Usage: scripts/sdk-timers.sh [--unversioned-upgrades|--sdk-upgrades|--sdk-application-upgrades]' \ + printf '%s\n' 'Usage: scripts/sdk-timers.sh [--unversioned-upgrades|--sdk-upgrades|--sdk-application-upgrades|--sdk-activity-upgrades]' \ 'Runs completion, worker SIGKILL/cold replay, Server restart and cooperative cancellation for PHP/Python/Rust timers.' \ 'Runs each interruption/cancellation case for both one timer and three concurrent deadlines in one original per SDK.' \ 'Replaces compatible business helpers across all SDKs and checks pinned Python class refusal, restoration and fresh v2 runs.' \ '--unversioned-upgrades instead runs twelve originals introducing versioning into already pending single/concurrent timers.' \ '--sdk-upgrades instead runs twelve originals across previous/current SDK packages with unchanged application code (requires Python 3).' \ '--sdk-application-upgrades instead changes both SDK packages and versioned application helpers in twelve originals (requires Python 3).' \ + '--sdk-activity-upgrades selects the separate eighteen-original pending-activity experiment (see scripts/sdk-activity-upgrades.sh).' \ 'Requires Docker Compose and exact artifact assignments from scripts/resolve-current-artifacts.sh.' \ 'SDK_TIMERS_COMPOSE_PROJECT_NAME selects an isolated project. All project resources are removed on exit.' exit 0 @@ -16,6 +17,7 @@ fi unversioned_only=0 sdk_packages_only=0 sdk_application_upgrade=0 +sdk_activity_upgrade=0 if [[ "${1:-}" == --unversioned-upgrades && $# == 1 ]]; then unversioned_only=1 shift @@ -26,6 +28,10 @@ elif [[ "${1:-}" == --sdk-application-upgrades && $# == 1 ]]; then sdk_packages_only=1 sdk_application_upgrade=1 shift +elif [[ "${1:-}" == --sdk-activity-upgrades && $# == 1 ]]; then + sdk_packages_only=1 + sdk_activity_upgrade=1 + shift fi [[ $# == 0 ]] || { printf '%s\n' 'Use --help for usage.' >&2; exit 2; } @@ -103,6 +109,9 @@ if (( sdk_packages_only )); then package_json different target_php_pin="${DURABLE_WORKFLOW_PHP_SDK_PIN:-}" fi +if (( sdk_activity_upgrade )); then + compose+=(-f "$repo_root/polyglot/activities/compose-sdk-upgrades.yml") +fi if [[ -n "$(docker ps -aq --filter "label=com.docker.compose.project=$COMPOSE_PROJECT_NAME")" ]]; then printf 'Choose a new isolated project. %s already has containers.\n' "$COMPOSE_PROJECT_NAME" >&2 exit 2 @@ -187,17 +196,20 @@ else "${compose[@]}" build smoke "${workers[@]}" fi "${compose[@]}" pull --policy always bootstrap server timer-queue +if (( sdk_activity_upgrade )); then export POLYGLOT_TIMER_APPLICATION=v1; fi "${compose[@]}" up -d --wait --wait-timeout 180 --no-build server timer-queue "${workers[@]}" docker image inspect "$DURABLE_SERVER_IMAGE" --format '{{json .RepoDigests}}' client() { + local observer=/app/scripts/sdk_timers.py + if (( sdk_activity_upgrade )); then observer=/app/scripts/sdk_activity_upgrades.py; fi "${compose[@]}" run --rm --no-deps --user 1000:1000 \ -e DURABLE_WORKFLOW_TIMER_ID -e DURABLE_WORKFLOW_WORKER_STOPPED_AT -e DURABLE_WORKFLOW_WORKER_RESTART_AT \ -e DURABLE_WORKFLOW_SERVER_STOPPED_AT -e DURABLE_WORKFLOW_SERVER_RESTART_AT \ -e DURABLE_WORKFLOW_TIMER_PENDING \ -e DURABLE_WORKFLOW_TIMER_WORKERS \ -e DURABLE_WORKFLOW_TIMER_SDK_VERSIONS -e DURABLE_WORKFLOW_TIMER_SDK_PREVIOUS -e DURABLE_WORKFLOW_TIMER_SDK_TARGETS \ - smoke python /app/scripts/sdk_timers.py "$@" + smoke python "$observer" "$@" } start() { @@ -306,7 +318,17 @@ package_upgrade() { client verify "$scenario-fresh" } -if (( sdk_packages_only )); then +if (( sdk_activity_upgrade )); then + remember_workers activity-upgrade + start activity-upgrade + "${compose[@]}" kill --signal SIGKILL "${workers[@]}" + export DURABLE_WORKFLOW_WORKER_STOPPED_AT="$(date -u +%Y-%m-%dT%H:%M:%S.%NZ)" + export POLYGLOT_TIMER_APPLICATION=v2 + replace_package_workers current activity-upgrade + client verify activity-upgrade + start activity-upgrade-fresh + client verify activity-upgrade-fresh +elif (( sdk_packages_only )); then package_scenario=sdk-upgrade if (( sdk_application_upgrade )); then package_scenario=sdk-application-upgrade