diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 389730d..ceb5948 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -23,6 +23,10 @@ on: description: Run the reviewed published SDK/Server activity identity tuple type: boolean default: false + published_timer_redrive: + description: Run the reviewed published SDK/native timer-redrive tuple + type: boolean + default: false permissions: contents: read @@ -32,6 +36,31 @@ concurrency: cancel-in-progress: ${{ github.event_name == 'pull_request' }} jobs: + published-timer-redrive: + if: github.event_name == 'workflow_dispatch' && inputs.published_timer_redrive + needs: [action-policy, lint] + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6 + with: + persist-credentials: false + - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6 + with: + python-version: "3.12.16" + - name: Bootstrap patched pip trust + uses: durable-workflow/.github/.github/actions/bootstrap-patched-pip@75587fc33ff40c34f32ef777293dad56bb6f48c7 # clean-trust + with: + clients: pip,python + - name: Run the published cold Worker timer-redrive case + run: python scripts/ci/published-activity-acceptance.py --scenario timer-redrive + - name: Retain published tuple and scenario evidence + if: always() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 + with: + name: published-timer-redrive-${{ github.sha }} + path: published-activity-results/ + retention-days: 7 published-activity-acceptance: if: github.event_name == 'workflow_dispatch' && inputs.published_activity_acceptance needs: [action-policy, lint] diff --git a/scripts/ci/activity_acceptance_consumer.py b/scripts/ci/activity_acceptance_consumer.py index c0dcb11..f51b756 100644 --- a/scripts/ci/activity_acceptance_consumer.py +++ b/scripts/ci/activity_acceptance_consumer.py @@ -1,4 +1,4 @@ -"""Install a closed wheel set, then await admission to the ordinary retry case.""" +"""Install a closed wheel set, then await admission to one fixed reviewed case.""" from __future__ import annotations @@ -23,6 +23,11 @@ LIMIT = 64 * 1024 * 1024 JUNIT = DATA / "identity.xml" CAPTURED_VERIFIER = "" +CAPTURED_ENTRY = "" +CASE_FILES = { + "activity-identity": {"test_activity_identity.py"}, + "timer-redrive": {"test_native_timer_redrive.py", "redrive_timer_clock_workflow.py"}, +} def receive(maximum): @@ -53,9 +58,11 @@ def run(arguments, environment, input_data=None): def validate_setup(packet): - if set(packet) != {"files", "wheels", "wheel_name", "wheel_sha256", "sdk_version"}: + if set(packet) != {"files", "wheels", "wheel_name", "wheel_sha256", "sdk_version", "scenario"}: raise ValueError("consumer setup shape differs") - expected_files = {"requirements.txt", "verify_activity_install.py", "test_activity_identity.py", "conftest.py"} + if packet["scenario"] not in CASE_FILES: + raise ValueError("consumer scenario differs") + expected_files = {"requirements.txt", "verify_activity_install.py", "conftest.py"} | CASE_FILES[packet["scenario"]] if set(packet["files"]) != expected_files or not 1 <= len(packet["wheels"]) <= 30: raise ValueError("consumer setup membership differs") if packet["wheel_name"] not in packet["wheels"]: @@ -132,28 +139,27 @@ def main(): verifier = retained_verifier() environment, python = prepare_environment() before = inventory(verifier, python, packet) + entry = captured_entry() + installed = entry["snapshot"](ENVIRONMENT) + site = ENVIRONMENT / "lib" / ("python" + str(sys.version_info.major) + "." + str(sys.version_info.minor)) + distributions = sorted( + (item.metadata["Name"], item.version) + for item in importlib.metadata.distributions(path=[str(site / "site-packages")]) + ) ready = { "sdk_inventory": before, "python": sys.version, "environment": str(ENVIRONMENT), - "distributions": run( - [ - str(python), - "-I", - "-B", - "-c", - "import importlib.metadata,json; print(json.dumps(sorted((d.metadata['Name']," - "d.version) for d in importlib.metadata.distributions())))", - ], - environment, - ), + "distributions": json.dumps(distributions), + "environment_inventory": installed, + "environment_inventory_sha256": entry["digest"](installed), } print("ACTIVITY_CONSUMER_READY=" + json.dumps(ready, sort_keys=True), flush=True) admission = receive(8192) if set(admission) != {"runtime_ipv4", "token"}: raise ValueError("consumer admission shape differs") admission["server_url"] = await_runtime(admission["runtime_ipv4"]) - output = execute_case(verifier, python, packet, before, environment, admission) + output = execute_case(verifier, python, packet, before, environment, admission, installed) print("ACTIVITY_CONSUMER_RESULT=" + json.dumps(output, sort_keys=True), flush=True) # Remain attached while the controller rechecks the same two IDs and IP. if receive(8192) != {"release": True}: @@ -217,7 +223,31 @@ def prepare_environment(): return environment, python -def execute_case(verifier, python, packet, before, environment, admission): +def captured_entry(): + if not CAPTURED_ENTRY: + raise ValueError("controller did not bind the Git-captured entry") + namespace = {"__name__": "captured_entry"} + exec(compile(CAPTURED_ENTRY, "", "exec"), namespace) + return namespace + + +def entry_source(mode, case_name, case_sha, installed, worker_entry=""): + source = CAPTURED_ENTRY + values = { + "MODE": mode, "INVENTORY_SHA256": captured_entry()["digest"](installed), + "CASE_SHA256": case_sha, "CASE_NAME": case_name, "WORKER_ENTRY": worker_entry, + } + for name, value in values.items(): + placeholder = name + ' = ""' + if source.count(placeholder) != 1: + raise ValueError("captured entry binding placeholder differs") + source = source.replace(placeholder, name + " = " + repr(value), 1) + return source + + +def execute_case(verifier, python, packet, before, environment, admission, installed): + redrive = packet["scenario"] == "timer-redrive" + test_name = "test_native_timer_redrive.py" if redrive else "test_activity_identity.py" environment.update( { "DURABLE_WORKFLOW_SERVER_URL": admission["server_url"], @@ -226,34 +256,30 @@ def execute_case(verifier, python, packet, before, environment, admission): "PYTEST_DISABLE_PLUGIN_AUTOLOAD": "1", } ) + if redrive: + environment["DURABLE_WORKFLOW_REDRIVE_CELL"] = "1" + worker_entry = entry_source("worker", test_name, packet["files"][test_name]["sha256"], installed) + parent_entry = entry_source("pytest", test_name, packet["files"][test_name]["sha256"], installed, worker_entry) result = subprocess.run( [ - str(python), + sys.executable, "-I", + "-S", "-B", - "-m", - "pytest", "-c", - "/dev/null", - "-p", - "pytest_asyncio.plugin", - "-p", - "no:cacheprovider", - "--asyncio-mode=auto", - "--import-mode=importlib", - "-v", - "--capture=tee-sys", - str(DATA / "test_activity_identity.py"), - "--junitxml=" + str(JUNIT), + parent_entry, ], env=environment, + input=json.dumps({"inventory": installed}), capture_output=True, text=True, - timeout=90, + timeout=180 if redrive else 90, ) after = inventory(verifier, python, packet) if after != before: raise ValueError("installed SDK inventory changed") + if captured_entry()["snapshot"](ENVIRONMENT) != installed: + raise ValueError("private install changed during the case") return { "complete": result.returncode == 0, "sdk_inventory": after, diff --git a/scripts/ci/activity_acceptance_entry.py b/scripts/ci/activity_acceptance_entry.py new file mode 100644 index 0000000..35d5bbb --- /dev/null +++ b/scripts/ci/activity_acceptance_entry.py @@ -0,0 +1,160 @@ +"""Captured entry: verify source and the private install before enabling imports.""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import os +import stat +import sys +import types +from pathlib import Path + +MODE = "" +INVENTORY_SHA256 = "" +CASE_SHA256 = "" +CASE_NAME = "" +WORKER_ENTRY = "" +ENVIRONMENT = Path("/run/acceptance-env") +DATA = Path("/run/acceptance-data") +MAXIMUM_FILE = 64 * 1024 * 1024 + + +def digest(value): + return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest() + + +def metadata(path): + value = path.lstat() + return [value.st_dev, value.st_ino, value.st_uid, value.st_gid, value.st_mode, value.st_size, value.st_mtime_ns] + + +def captured_file(path, maximum=MAXIMUM_FILE): + descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW) + try: + before = os.fstat(descriptor) + if not stat.S_ISREG(before.st_mode) or before.st_size > maximum: + raise ValueError("captured file shape or size differs") + chunks, size = [], 0 + while block := os.read(descriptor, min(1024 * 1024, maximum + 1 - size)): + chunks.append(block) + size += len(block) + if size > maximum: + raise ValueError("captured file exceeds its bound") + after = os.fstat(descriptor) + if (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns) != ( + after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns, + ): + raise ValueError("captured file changed during its read") + return b"".join(chunks) + finally: + os.close(descriptor) + + +def snapshot(root): + before = metadata(root) + if not stat.S_ISDIR(before[4]): + raise ValueError("private install root is not a directory") + entries = {} + for path in sorted(root.rglob("*")): + info = metadata(path) + item = {"metadata": info} + if stat.S_ISREG(info[4]): + item["sha256"] = hashlib.sha256(captured_file(path)).hexdigest() + elif stat.S_ISLNK(info[4]): + item["target"] = os.readlink(path) + elif not stat.S_ISDIR(info[4]): + raise ValueError("private install contains an unsupported entry") + entries[path.relative_to(root).as_posix()] = item + if before != metadata(root): + raise ValueError("private install root changed during capture") + return {"root": before, "entries": entries} + + +def guarded_source(path, expected): + source = captured_file(path, 256 * 1024) + if hashlib.sha256(source).hexdigest() != expected: + raise ValueError("captured case source differs") + return source + + +def source_module(source): + module = types.ModuleType("captured_published_case") + module.__file__ = str(DATA / CASE_NAME) + sys.modules[module.__name__] = module + exec(compile(source, module.__file__, "exec", optimize=0), module.__dict__) + return module + + +def run_pytest(source, inventory): + # pytest never opens the writable test or conftest as executable code. + import pytest + + class CapturedModule(pytest.Module): + def _getobj(self): + module = source_module(source) + module._COLD_BOOTSTRAP = WORKER_ENTRY + module._COLD_INVENTORY = inventory + module._COLD_PYTHON = sys.executable + return module + + class CapturedPlugin: + @pytest.hookimpl(tryfirst=True) + def pytest_pycollect_makemodule(self, module_path, parent): + if module_path != DATA / CASE_NAME: + raise ValueError("pytest requested an uncaptured module") + return CapturedModule.from_parent(parent, path=module_path) + + @pytest.fixture(scope="session") + def server_url(self): + return os.environ["DURABLE_WORKFLOW_SERVER_URL"] + + @pytest.fixture(scope="session") + def server_token(self): + return os.environ["DURABLE_WORKFLOW_AUTH_TOKEN"] + + return pytest.main([ + "-c", "/dev/null", "--noconftest", "-p", "pytest_asyncio.plugin", "-p", "no:cacheprovider", + "-o", "asyncio_default_fixture_loop_scope=function", + "--asyncio-mode=auto", "--import-mode=importlib", "-v", "--capture=tee-sys", str(DATA / CASE_NAME), + "--junitxml=" + str(DATA / "identity.xml"), + ], plugins=[CapturedPlugin()]) + + +def main(): + if not (sys.flags.isolated and sys.flags.no_site and sys.dont_write_bytecode) or sys.flags.optimize: + raise ValueError("captured entry requires isolated no-site Python with active assertions") + payload = sys.stdin.buffer.read(4 * 1024 * 1024 + 1) + if len(payload) > 4 * 1024 * 1024: + raise ValueError("captured entry input exceeds its bound") + packet = json.loads(payload) + if set(packet) != ({"inventory", "worker"} if MODE == "worker" else {"inventory"}): + raise ValueError("captured entry input shape differs") + inventory = packet["inventory"] + if digest(inventory) != INVENTORY_SHA256: + raise ValueError("private install commitment differs") + source = guarded_source(DATA / CASE_NAME, CASE_SHA256) + if snapshot(ENVIRONMENT) != inventory: + raise ValueError("private install inventory changed") + # No site.addsitedir or site.main: .pth and sitecustomize are never run. + site = ENVIRONMENT / "lib" / ("python" + str(sys.version_info.major) + "." + str(sys.version_info.minor)) + site /= "site-packages" + if not site.is_dir() or site.is_symlink(): + raise ValueError("verified site directory differs") + sys.path.insert(0, str(site)) + sys.prefix = sys.exec_prefix = str(ENVIRONMENT) + if MODE == "worker": + asyncio.run(source_module(source).cold_worker(packet["worker"])) + result = 0 + elif MODE == "pytest": + result = run_pytest(source, inventory) + else: + raise ValueError("captured entry mode differs") + if snapshot(ENVIRONMENT) != inventory: + raise ValueError("private install changed during execution") + raise SystemExit(result) + + +if __name__ == "__main__": + main() diff --git a/scripts/ci/published-activity-acceptance.md b/scripts/ci/published-activity-acceptance.md index fb275ac..645114d 100644 --- a/scripts/ci/published-activity-acceptance.md +++ b/scripts/ci/published-activity-acceptance.md @@ -19,6 +19,11 @@ SDK wheel member, complete distribution metadata and installed RECORD, and uses isolated Python with explicit pytest plugins. Installer-created metadata and console scripts have exact expected bytes. The checker stays in the trusted bootstrap's memory through the case. The checkout is never installed as the SDK. +The complete private installed-file inventory is captured before SDK admission. +Parent pytest uses the read-only base interpreter with `-I -S -B`; its literal +entry verifies captured test bytes and the installed inventory before enabling +the exact private site directory. A captured collector and fixed fixtures avoid +executing mutable test/conftest paths, and no `.pth` startup is processed. The published native runtime and official Python consumer share one owned internal network. Neither publishes a port or receives a host mount. The diff --git a/scripts/ci/published-activity-acceptance.py b/scripts/ci/published-activity-acceptance.py index 0931ac5..01b9fe6 100644 --- a/scripts/ci/published-activity-acceptance.py +++ b/scripts/ci/published-activity-acceptance.py @@ -1,7 +1,8 @@ -"""Run one exact published SDK retry-identity case inside an owned internal network.""" +"""Run one fixed published SDK acceptance case inside an owned internal network.""" from __future__ import annotations +import argparse import base64 import hashlib import ipaddress @@ -38,6 +39,15 @@ LABEL_FILTER = "label=" ID_FORMAT = "{{.ID}}" DIAGNOSTIC_LIMIT = 64 * 1024 +SCENARIO = "activity-identity" +SCENARIOS = { + "activity-identity": ( + "test_activity_identity.py", "test_retry_preserves_received_execution_id_and_changes_attempt_id", + ), + "timer-redrive": ( + "test_native_timer_redrive.py", "test_cold_workers_preserve_fractional_timer_clock_across_repeated_redrive", + ), +} # No inherited context, registry credentials, proxy, SSH or remote Docker endpoint. # This kernel directory is private to each CLI process and contains no config.json. DOCKER_ENVIRONMENT = { @@ -111,7 +121,7 @@ def verify_case_result(result): raise ValueError("published retry case reported failure") root = ElementTree.fromstring(result["junit"]) cases = list(root.iter("testcase")) - expected = "test_retry_preserves_received_execution_id_and_changes_attempt_id" + expected = SCENARIOS[SCENARIO][1] if len(cases) != 1 or cases[0].get("name") != expected: raise ValueError("published retry case inventory differs") if any(cases[0].find(tag) is not None for tag in ("failure", "error", "skipped")): @@ -131,10 +141,12 @@ def packet_file(data): def bind_bootstrap(captured): source = captured["bootstrap.py"].decode() - placeholder = 'CAPTURED_VERIFIER = ""' - if source.count(placeholder) != 1: - raise ValueError("consumer verifier binding placeholder differs") - return source.replace(placeholder, "CAPTURED_VERIFIER = " + repr(captured[VERIFIER_FILENAME].decode()), 1) + for name, filename in (("CAPTURED_VERIFIER", VERIFIER_FILENAME), ("CAPTURED_ENTRY", "entry.py")): + placeholder = name + ' = ""' + if source.count(placeholder) != 1: + raise ValueError("consumer source binding placeholder differs") + source = source.replace(placeholder, name + " = " + repr(captured[filename].decode()), 1) + return source def capture_source(report): @@ -144,11 +156,16 @@ def capture_source(report): paths = { REQUIREMENTS: "scripts/ci/published-activity-requirements.txt", VERIFIER_FILENAME: "scripts/ci/verify_activity_install.py", - "test_activity_identity.py": "tests/integration/test_activity_identity.py", + SCENARIOS[SCENARIO][0]: "tests/integration/" + SCENARIOS[SCENARIO][0], "conftest.py": "tests/integration/conftest.py", "bootstrap.py": "scripts/ci/activity_acceptance_consumer.py", - "tuple.json": "tests/fixtures/published-activity-acceptance.json", + "entry.py": "scripts/ci/activity_acceptance_entry.py", + "tuple.json": "tests/fixtures/" + ( + "published-timer-redrive.json" if SCENARIO == "timer-redrive" else "published-activity-acceptance.json" + ), } + if SCENARIO == "timer-redrive": + paths["redrive_timer_clock_workflow.py"] = "tests/integration/redrive_timer_clock_workflow.py" captured = { name: call("git", "show", commit + ":" + path, cwd=ROOT).stdout.encode() for name, path in paths.items() } @@ -196,10 +213,14 @@ def download_wheels(selected, captured, files): wheels[path.name] = packet_file(path.read_bytes()) if wheels[wheel_name]["sha256"] != approval["files"][wheel_name]: raise ValueError("registry wheel differs from the independently reviewed release") + inputs = [REQUIREMENTS, VERIFIER_FILENAME, SCENARIOS[SCENARIO][0], "conftest.py"] + if SCENARIO == "timer-redrive": + inputs.append("redrive_timer_clock_workflow.py") return { + "scenario": SCENARIO, "files": { name: packet_file(captured[name]) - for name in (REQUIREMENTS, VERIFIER_FILENAME, "test_activity_identity.py", "conftest.py") + for name in inputs }, "wheels": wheels, "wheel_name": wheel_name, @@ -468,8 +489,9 @@ def run_case(owned, selected, images, packet, bootstrap, report): if ready["sdk_inventory"] != result["sdk_inventory"]: raise ValueError("installed SDK bytes changed during the case") owned.files.write("sdk-inventory.json", json.dumps(result["sdk_inventory"], indent=2)) - owned.files.write("identity.log", result["case_output"]) - owned.files.write("identity.xml", result["junit"]) + name = "timer-redrive" if SCENARIO == "timer-redrive" else "identity" + owned.files.write(name + ".log", result["case_output"]) + owned.files.write(name + ".xml", result["junit"]) report["consumer_ready"] = {key: value for key, value in ready.items() if key != "sdk_inventory"} session.send({"release": True}) code = session.finish() @@ -482,7 +504,7 @@ def run_case(owned, selected, images, packet, bootstrap, report): def main(): files = output_files.Output(OUTPUT) - report = {"complete": False, "started_at": datetime.now(timezone.utc).isoformat()} + report = {"complete": False, "scenario": SCENARIO, "started_at": datetime.now(timezone.utc).isoformat()} owned = OwnedDocker(uuid4().hex, files, report) signal.signal(signal.SIGALRM, deadline) signal.alarm(360) @@ -513,4 +535,7 @@ def main(): if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--scenario", choices=tuple(SCENARIOS), default=SCENARIO) + SCENARIO = parser.parse_args().scenario main() diff --git a/scripts/ci/published-timer-redrive.md b/scripts/ci/published-timer-redrive.md new file mode 100644 index 0000000..62d2860 --- /dev/null +++ b/scripts/ci/published-timer-redrive.md @@ -0,0 +1,71 @@ +The timer-redrive cell uses the exact published tuple in +`tests/fixtures/published-timer-redrive.json` and the same closed install, +internal network, container authority, SDK inventory and independent cleanup +profile as [activity identity](published-activity-acceptance.md). The identity +tuple and its default command remain unchanged. + +After the changed case, tuple and runner have source review, terminal normal +checks and normal integration, run the ordinary GitHub Actions cell: + +```bash +gh workflow run ci.yml --ref main -f published_timer_redrive=true +``` + +One unchanged workflow definition first completes an activity, waits for a +timer, and fails its second activity. The source TimerFired must have a real +nonzero fractional event timestamp. The test never edits published history. +It obtains original start authority from the persisted run, then performs first +and repeated redrive through the public client API. Each of the three Workers +runs in a fresh Python process with the same definition and fingerprint. +The first successor fails the second activity again; the next succeeds. + +The retained observations prove callback identity and clock, closed emitted +command/task inventories, new shared timer-pair identities, immediate-predecessor +lineage, original start anchor, completed-result reuse and unchanged source +history. Recorded nonempty Worker fingerprints must match every original and +successor `WorkflowStarted` fingerprint with its `worker` source. Successors must +already have the reused timer prefix before a Worker starts, issue no timer +command, expose no timer wait, and execute only the failed activity. Parsed UTC +instants retain all supported microseconds; unrepresentable precision refuses +rather than being silently rounded. The original definition does not read an +ordinary entry clock, so this does not advertise that separate SDK capability. +Unpruned public history exports bind the selected workflow/run and include every +task/wait status. Original stored timer tasks/waits must each count one; +successors must each count zero before and after their cold Worker. All emitted +completion task IDs and activity callback task IDs must match the stored export +inventory. Full timer payloads must match their predecessor modulo the explicitly +authorized new pair ID, immediate predecessor and preserved clock fields. +The paged public history supplies the original TimerFired `timestamp`. +First reuse compares the authoritative start and event clocks as full-precision +UTC instants. Native fallback serialization may change the public API's `Z` to +`+00:00` and omit insignificant fractional digits. Later reuse preserves the +immediate predecessor's inherited strings exactly as well as comparing their +parsed UTC instants. + +An absent matching registration must refuse redrive without changing the chain +or source history. Admission then holds a matching source-cohort registration +that never polls or executes work; its metadata is copied from the actual source +Worker and its cleanup uses the returned registration generation token. Only the +fresh child may execute the accepted successor. Physical `workflow_run_timers` +counts remain a separate actual published-producer observation. Reconstructed +export timers and an immutable qualification link are not per-fixture physical +counts; this API case makes no such claim. + +The trusted consumer captures the complete private installed-file inventory +before admitting SDK execution. Parent pytest and every cold child use the +read-only base interpreter with `-I -S -B`. Their literal entry verifies the +independently fixed case digest and installed-inventory commitment before adding +the exact private site directory, without processing `.pth` or sitecustomize. +Pytest uses captured source bytes and fixed fixtures instead of reopening the +mutable test or conftest as executable code. Each Worker has a 45-second SDK +budget and a 55-second child wall limit. The selected pytest case has a +180-second bound within the existing 240-second consumer exchange and +six-minute controller deadline; cleanup keeps its independent per-call bounds. +The older identity case retains its 90-second pytest bound. No local SDK or +native runtime run is part of source qualification. + +The seven-day artifact retains the complete fixed wheel set, selected platform +and source, unchanged SDK inventory, before/after runtime authority, all three +cold-process fingerprints and command/callback observations, actual histories, +JUnit and full-ID/label absence. An installed cell is qualified only after its +retained evidence receives independent review. diff --git a/tests/fixtures/published-timer-redrive.json b/tests/fixtures/published-timer-redrive.json new file mode 100644 index 0000000..421f036 --- /dev/null +++ b/tests/fixtures/published-timer-redrive.json @@ -0,0 +1,11 @@ +{ + "schema": "sdk-python.published-timer-redrive/v1", + "sdk_version": "2.17.2", + "server_image": "ghcr.io/durable-workflow/server@sha256:39d218fc5f251a3e7c4ff1dcda06d9cc53f66c034f0c6d0f67e8371fd0cc97d6", + "server_source": "da38c0ede9b15046fcab2bb0f25412c85a3e3404", + "server_amd64_manifest": "sha256:8bc8400c4ff203c1e4d5c7f82028cfa1d429ab00a47a6e698453d8345ae4bcfc", + "server_qualification": "https://github.com/durable-workflow/server/issues/980#issuecomment-6867470082", + "python_image": "docker.io/library/python@sha256:935c3c5b12357a7e10caa9c1e6ca5692344d97cf229832ef584be2cec9fc03df", + "python_amd64_manifest": "sha256:0f086103b026048cc50cb5da25ec435c023c5241a83fc44daea593377abc61c4", + "python_version": "3.12.16" +} diff --git a/tests/integration/test_native_timer_redrive.py b/tests/integration/test_native_timer_redrive.py new file mode 100644 index 0000000..9dc0695 --- /dev/null +++ b/tests/integration/test_native_timer_redrive.py @@ -0,0 +1,404 @@ +"""Published-package cold Worker proof; run only through its reviewed manual cell.""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import linecache +import os +import re +import subprocess +import sys +import types +from collections import Counter +from copy import deepcopy +from datetime import datetime, timezone +from pathlib import Path +from typing import Any +from uuid import uuid4 + +import pytest + +from durable_workflow import ActivityRetryPolicy, Client, Worker, activity +from durable_workflow.errors import ServerError +from durable_workflow.workflow import CompleteWorkflow, ScheduleActivity, StartTimer + +DEFINITION_SHA = "af24e69d5d4ed57e3cb56915a5e2febe535438e8bb2b33ecfd998098c1557b2b" +WORKFLOW_TYPE = "tests.python-redrive-timer-clock" +_COLD_BOOTSTRAP = "" +_COLD_INVENTORY: dict[str, Any] = {} +_COLD_PYTHON = "" + + +def instant(value: Any) -> datetime: + assert isinstance(value, str) + assert value + assert not re.search(r"\.\d{7}", value), "clock precision exceeds Python's exact microsecond representation" + result = datetime.fromisoformat(value.replace("Z", "+00:00")) + assert result.tzinfo is not None + return result.astimezone(timezone.utc) + + +def one(events: list[dict[str, Any]], kind: str, *, step: int | None = None) -> dict[str, Any]: + selected = [ + event for event in events + if event["event_type"] == kind and (step is None or event["payload"].get("sequence") == step) + ] + assert len(selected) == 1, (kind, step, len(selected)) + return selected[0] + + +def source_timer_boundary(events: list[dict[str, Any]]) -> tuple[str, datetime]: + text = one(events, "TimerFired", step=2)["timestamp"] + clock = instant(text) + assert clock.microsecond != 0, "real TimerFired must expose nonzero source precision" + return text, clock + + +def verify_reuse( + events: list[dict[str, Any]], predecessor: list[dict[str, Any]], *, + predecessor_id: str, source_clock: datetime, source_start: datetime, used_timer_ids: set[str], + source_clock_text: str, source_start_text: str, +) -> str: + started = one(events, "WorkflowStarted")["payload"] + assert started["continued_from_run_id"] == predecessor_id + assert started["recovery_kind"] == "redrive" + assert started["resume_step_sequence"] == 3 + assert instant(started["replayed_started_at"]) == source_start + inherited_start = one(predecessor, "WorkflowStarted")["payload"].get("replayed_started_at") + if inherited_start is not None: + assert isinstance(inherited_start, str) + assert started["replayed_started_at"] == inherited_start + assert instant(source_start_text) == source_start + scheduled = one(events, "TimerScheduled", step=2)["payload"] + fired = one(events, "TimerFired", step=2)["payload"] + timer_id = scheduled["timer_id"] + assert isinstance(timer_id, str) + assert timer_id + assert timer_id not in used_timer_ids + assert fired["timer_id"] == timer_id + assert fired["reused_from_run_id"] == predecessor_id + assert instant(fired["reused_recorded_at"]) == source_clock + expected_scheduled = deepcopy(one(predecessor, "TimerScheduled", step=2)["payload"]) + expected_fired = deepcopy(one(predecessor, "TimerFired", step=2)["payload"]) + inherited_clock = expected_fired.get("reused_recorded_at") + if inherited_clock is not None: + assert isinstance(inherited_clock, str) + assert fired["reused_recorded_at"] == inherited_clock + assert instant(source_clock_text) == source_clock + for payload in (expected_scheduled, expected_fired): + payload.pop("task", None) + payload.pop("command", None) + payload["timer_id"] = timer_id + expected_fired.update(reused_from_run_id=predecessor_id, reused_recorded_at=fired["reused_recorded_at"]) + assert scheduled == expected_scheduled + assert fired == expected_fired + original = one(predecessor, "ActivityCompleted", step=1)["payload"] + reused = one(events, "ActivityCompleted", step=1)["payload"] + assert reused["result"] == original["result"] + assert reused["reused_from_run_id"] == predecessor_id + assert reused["reused_activity_execution_id"] == original["activity_execution_id"] + assert not [event for event in events if event["event_type"] == "ActivityScheduled"] + return timer_id + + +def definition(source: str) -> type: + assert hashlib.sha256(source.encode()).hexdigest() == DEFINITION_SHA + name, filename = "accepted_redrive_timer", "" + module = types.ModuleType(name) + module.__file__ = filename + sys.modules[name] = module + linecache.cache[filename] = (len(source), None, source.splitlines(keepends=True), filename) + exec(compile(source, filename, "exec"), module.__dict__) + return module.RedriveTimerClockWorkflow + + +async def cold_worker(packet: dict[str, Any]) -> None: + callbacks: list[dict[str, Any]] = [] + completions: list[dict[str, Any]] = [] + registration: dict[str, Any] = {} + + class ObservedClient(Client): + async def register_worker(self, **kwargs: Any) -> Any: + acknowledgement = await super().register_worker(**kwargs) + registration.update(deepcopy({key: value for key, value in kwargs.items() if key != "process_metrics"})) + return acknowledgement + + async def complete_workflow_task(self, **kwargs: Any) -> Any: + completions.append({"task_id": kwargs["task_id"], "commands": deepcopy(kwargs["commands"])}) + return await super().complete_workflow_task(**kwargs) + + @activity.defn(name="tests.redrive.first") + async def first(name: str) -> str: + info = activity.context().info + callbacks.append({ + "activity": "first", "pid": os.getpid(), "run_id": info.run_id, + "task_id": info.task_id, "attempt_id": info.activity_attempt_id, "execution_id": info.activity_execution_id, + }) + return f"Hello, {name}!" + + @activity.defn(name="tests.redrive.second") + async def second(greeting: str, clock: str) -> str: + info = activity.context().info + callbacks.append({ + "activity": "second", "clock": clock, "pid": os.getpid(), "run_id": info.run_id, + "task_id": info.task_id, "attempt_id": info.activity_attempt_id, + "execution_id": info.activity_execution_id, + }) + if packet["fail_second"]: + raise RuntimeError("fixed failed-activity redrive boundary") + return "resumed" + + async with ObservedClient(packet["server_url"], token=packet["token"], namespace="default") as client: + worker = Worker( + client, task_queue=packet["queue"], worker_id="python-clock-" + uuid4().hex, + workflows=[definition(packet["definition"])], activities=[first, second], sticky_cache_capacity=0, + ) + fingerprints = dict(worker.workflow_definition_fingerprints) + try: + outcome = await worker.run_until(workflow_id=packet["workflow_id"], timeout=45) + finally: + await worker.stop() + print("COLD_WORKER_RESULT=" + json.dumps({ + "pid": os.getpid(), "worker_id": worker.worker_id, "fingerprints": fingerprints, + "definition_sha256": DEFINITION_SHA, "status": outcome.status, + "callbacks": callbacks, "completions": completions, "registration": registration, + "registration_namespace": client.namespace, + }, sort_keys=True), flush=True) + + +def run_cold_worker(packet: dict[str, Any]) -> dict[str, Any]: + assert _COLD_BOOTSTRAP + assert _COLD_INVENTORY + assert _COLD_PYTHON + environment = {"PATH": os.defpath, "PYTHONDONTWRITEBYTECODE": "1"} + result = subprocess.run( + [_COLD_PYTHON, "-I", "-S", "-B", "-c", _COLD_BOOTSTRAP], + input=json.dumps({"inventory": _COLD_INVENTORY, "worker": packet}), + env=environment, capture_output=True, text=True, timeout=55, check=False, + ) + assert result.returncode == 0, (result.returncode, result.stdout[-16384:], result.stderr[-16384:]) + lines = [line.removeprefix("COLD_WORKER_RESULT=") for line in result.stdout.splitlines() + if line.startswith("COLD_WORKER_RESULT=")] + assert len(lines) == 1 + return json.loads(lines[0]) + + +def verify_export(document: dict[str, Any], workflow_id: str, run_id: str, *, terminal: bool, source: bool) -> None: + assert document["schema"] == "durable-workflow.v2.history-export" + assert document["schema_version"] == 2 + selected = document["workflow"] + assert selected["instance_id"] == workflow_id + assert selected["run_id"] == run_id + assert selected["workflow_type"] == WORKFLOW_TYPE + assert selected["details_state"] == "retained" + assert selected["details_pruned_at"] is None + assert document["history_complete"] is terminal + assert document["selected_run"]["waits_projection_source"] == "workflow_run_waits" + assert document["redaction"]["applied"] is False + for section, kind in (("tasks", "type"), ("waits", "kind")): + rows = document[section] + assert isinstance(rows, list) + assert all(isinstance(row, dict) and isinstance(row[kind], str) and row[kind] for row in rows) + assert all(isinstance(row["status"], str) and row["status"] for row in rows) + assert sum(row[kind] == "timer" for row in rows) == (1 if source else 0) + + +def verify_recorded_fingerprints(observations: list[dict[str, Any]], histories: list[list[dict[str, Any]]]) -> None: + assert len(observations) == len(histories) == 3 + expected = observations[0]["fingerprints"][WORKFLOW_TYPE] + assert isinstance(expected, str) + assert re.fullmatch(r"sha256:[0-9a-f]{64}", expected) + for observed, events in zip(observations, histories, strict=True): + assert observed["fingerprints"] == {WORKFLOW_TYPE: expected} + started = one(events, "WorkflowStarted")["payload"] + assert started["workflow_definition_fingerprint"] == expected + assert started["workflow_definition_fingerprint_source"] == "worker" + + +def verify_worker( + observed: dict[str, Any], *, redrive: bool, clock: datetime, queue: str, exported: dict[str, Any], +) -> None: + assert observed["definition_sha256"] == DEFINITION_SHA + assert set(observed["fingerprints"]) == {WORKFLOW_TYPE} + assert re.fullmatch(r"sha256:[0-9a-f]{64}", observed["fingerprints"][WORKFLOW_TYPE]) + assert observed["worker_id"] + assert isinstance(observed["pid"], int) + assert observed["pid"] > 0 + assert all(set(completion) == {"task_id", "commands"} for completion in observed["completions"]) + task_ids = [completion["task_id"] for completion in observed["completions"]] + assert all(isinstance(identity, str) and identity for identity in task_ids) + assert len(set(task_ids)) == len(task_ids) == (2 if redrive else 4) + assert set(task_ids) == {task["id"] for task in exported["tasks"] if task["type"] == "workflow"} + counts = {"workflow": 2, "activity": 1} if redrive else {"workflow": 4, "activity": 2, "timer": 1} + assert Counter(task["type"] for task in exported["tasks"]) == counts + assert [item["activity"] for item in observed["callbacks"]] == (["second"] if redrive else ["first", "second"]) + for callback in observed["callbacks"]: + assert all(isinstance(callback[name], str) and callback[name] for name in ( + "task_id", "attempt_id", "execution_id", "run_id", + )) + assert callback["run_id"] == exported["workflow"]["run_id"] + assert {callback["task_id"] for callback in observed["callbacks"]} == { + task["id"] for task in exported["tasks"] if task["type"] == "activity" + } + assert all(len(completion["commands"]) == 1 for completion in observed["completions"]) + commands = [command for completion in observed["completions"] for command in completion["commands"]] + clock_text = clock.isoformat(timespec="microseconds") + scheduled = ScheduleActivity( + "tests.redrive.second", ["Hello, Taylor!", clock_text], retry_policy=ActivityRetryPolicy(max_attempts=1), + ).to_server_command(queue) + prefix = [scheduled] if redrive else [ + ScheduleActivity("tests.redrive.first", ["Taylor"]).to_server_command(queue), + StartTimer(1).to_server_command(queue), scheduled, + ] + assert commands[:-1] == prefix + terminal = commands[-1] + if observed["status"] == "completed": + assert terminal == CompleteWorkflow({ + "greeting": "Hello, Taylor!", "timer_clock": clock_text, "resumed": "resumed", + }).to_server_command(queue) + else: + required = {"type", "message", "failed_step_sequence", "failed_activity_execution_id"} + allowed = required | {"exception_type", "exception_class", "exception", "non_retryable"} + assert required <= set(terminal) <= allowed + assert terminal["type"] == "fail_workflow" + assert terminal["failed_step_sequence"] == 3 + assert isinstance(terminal["message"], str) + assert terminal["message"] + assert terminal["failed_activity_execution_id"] == observed["callbacks"][-1]["execution_id"] + assert instant(observed["callbacks"][-1]["clock"]) == clock + + +async def admit_redrive( + client: Client, queue: str, run_id: str, observed: dict[str, Any], events: list[dict[str, Any]], +): + started = one(events, "WorkflowStarted")["payload"] + fingerprint = observed["fingerprints"][WORKFLOW_TYPE] + assert re.fullmatch(r"sha256:[0-9a-f]{64}", fingerprint) + assert started["workflow_definition_fingerprint"] == fingerprint + assert started["workflow_definition_fingerprint_source"] == "worker" + assert client.namespace == observed["registration_namespace"] == "default" + before = [run.run_id for run in (await client.list_workflow_runs(queue)).runs] + try: + await client.redrive_workflow(queue, run_id) + except ServerError as error: + assert error.status == 409 + assert error.reason() == "workflow_definition_unavailable_or_changed" + assert isinstance(error.body, dict) + assert error.body["outcome"] == "rejected" + assert error.body["workflow_id"] == queue + assert error.body["run_id"] == run_id + assert error.body["accepted"] is False + assert error.body["workflow_run_id"] is None + assert error.body["task_id"] is None + refusal = {"status": error.status, "body": error.body} + else: + raise AssertionError("redrive unexpectedly admitted without a matching registration") + assert [run.run_id for run in (await client.list_workflow_runs(queue)).runs] == before + assert [event async for event in client.iter_history(queue, run_id)] == events + request = deepcopy(observed["registration"]) + assert request["task_queue"] == queue + assert request["supported_workflow_types"] == [WORKFLOW_TYPE] + assert request["workflow_definition_fingerprints"] == {WORKFLOW_TYPE: fingerprint} + holder_id = "python-clock-admission-" + uuid4().hex + request["worker_id"] = holder_id + acknowledgement = await client.register_worker(**request) + token = acknowledgement.get("registration_token") + assert isinstance(token, str) + assert re.fullmatch(r"[0-9a-f]{32}", token) + try: + assert acknowledgement["worker_id"] == holder_id + assert acknowledgement["registered"] is True + assert acknowledgement["namespace"] == "default" + assert acknowledgement["task_queue"] == queue + # The holder never creates a Worker or polls: only the later cold child + # may claim and execute the successor's task. + receipt = await client.redrive_workflow(queue, run_id) + finally: + removed = await client.deregister_worker_registration( + holder_id, registration_token=token, expected_namespace="default", timeout=10, + ) + assert removed["outcome"] == "deregistered" + assert removed["recovered_workflow_task_count"] == 0 + return receipt, { + "absent_registration": refusal, "matching_registration": request, + "registration_token_sha256": hashlib.sha256(token.encode()).hexdigest(), + "cleanup": {key: value for key, value in removed.items() if key != "registration_token"}, + } + + +@pytest.mark.skipif( + os.environ.get("DURABLE_WORKFLOW_REDRIVE_CELL") != "1", reason="reviewed manual published cell only", +) +async def test_cold_workers_preserve_fractional_timer_clock_across_repeated_redrive( + server_url: str, server_token: str, +) -> None: + source = Path(__file__).with_name("redrive_timer_clock_workflow.py").read_text() + assert hashlib.sha256(source.encode()).hexdigest() == DEFINITION_SHA + queue = "python-timer-clock-" + uuid4().hex + common = {"server_url": server_url, "token": server_token, "queue": queue, + "workflow_id": queue, "definition": source} + observations = [] + histories = [] + exports = [] + admissions = [] + run_ids = [] + async with Client(server_url, token=server_token, namespace="default") as client: + handle = await client.start_workflow( + workflow_type=WORKFLOW_TYPE, workflow_id=queue, task_queue=queue, input=["Taylor", 1], + ) + assert handle.run_id + run_ids.append(handle.run_id) + observations.append(await asyncio.to_thread(run_cold_worker, {**common, "fail_second": True})) + assert observations[0]["status"] == "failed" + histories.append([event async for event in client.iter_history(queue, handle.run_id)]) + frozen_source = deepcopy(histories[0]) + original = await client.describe_workflow_run(queue, handle.run_id) + source_start = instant(original.started_at) + source_clock_text, source_clock = source_timer_boundary(histories[0]) + exports.append(await client.export_history(queue, handle.run_id)) + verify_export(exports[0], queue, handle.run_id, terminal=True, source=True) + used_timer_ids = {one(histories[0], "TimerFired", step=2)["payload"]["timer_id"]} + for repeat in (1, 2): + receipt, admission = await admit_redrive(client, queue, run_ids[-1], observations[-1], histories[-1]) + admissions.append(admission) + assert receipt.outcome == "redriven" + assert receipt.run_id + assert receipt.run_id not in run_ids + run_id = receipt.run_id + prefix = [event async for event in client.iter_history(queue, run_id)] + timer_id = verify_reuse( + prefix, histories[-1], predecessor_id=run_ids[-1], source_clock=source_clock, + source_start=source_start, used_timer_ids=used_timer_ids, + source_start_text=original.started_at, source_clock_text=source_clock_text, + ) + used_timer_ids.add(timer_id) + before = await client.describe_workflow_run(queue, run_id) + assert before.wait_kind != "timer" + admitted_export = await client.export_history(queue, run_id) + verify_export(admitted_export, queue, run_id, terminal=False, source=False) + observed = await asyncio.to_thread(run_cold_worker, {**common, "fail_second": repeat == 1}) + assert observed["status"] == ("failed" if repeat == 1 else "completed") + completed_export = await client.export_history(queue, run_id) + verify_export(completed_export, queue, run_id, terminal=True, source=False) + verify_worker(observed, redrive=True, clock=source_clock, queue=queue, exported=completed_export) + assert observed["callbacks"][0]["run_id"] == run_id + observations.append(observed) + run_ids.append(run_id) + histories.append([event async for event in client.iter_history(queue, run_id)]) + exports.append({"before": admitted_export, "after": completed_export}) + assert histories[-1][:len(prefix)] == prefix + assert [event async for event in client.iter_history(queue, handle.run_id)] == frozen_source + result = await client.get_workflow_handle(queue, run_id=run_ids[-1]).result(timeout=10) + verify_worker(observations[0], redrive=False, clock=source_clock, queue=queue, exported=exports[0]) + assert len({item["pid"] for item in observations}) == len({item["worker_id"] for item in observations}) == 3 + verify_recorded_fingerprints(observations, histories) + assert result == {"greeting": "Hello, Taylor!", "timer_clock": source_clock.isoformat(timespec="microseconds"), + "resumed": "resumed"} + print(json.dumps({ + "scope": "published cold-Worker timer redrive", "runs": run_ids, "source_started_at": original.started_at, + "source_timer_clock": source_clock_text, "source_timer_clock_utc": source_clock.isoformat(), + "definition_sha256": DEFINITION_SHA, + "workers": observations, "histories": histories, "exports": exports, "admissions": admissions, + }, sort_keys=True)) diff --git a/tests/test_activity_acceptance_consumer.py b/tests/test_activity_acceptance_consumer.py index e9fcf66..0bcb27d 100644 --- a/tests/test_activity_acceptance_consumer.py +++ b/tests/test_activity_acceptance_consumer.py @@ -135,6 +135,7 @@ def test_closed_consumer_refuses_sdk_wheel_omission_or_substitution(): "wheel_name": "sdk.whl", "wheel_sha256": "b" * 64, "sdk_version": "test", + "scenario": "activity-identity", } with pytest.raises(ValueError, match="differs from approval"): consumer.validate_setup(packet) @@ -143,6 +144,24 @@ def test_closed_consumer_refuses_sdk_wheel_omission_or_substitution(): consumer.validate_setup(packet) +@pytest.mark.parametrize("scenario", ["activity-identity", "timer-redrive", "foreign"]) +def test_consumer_scenario_has_closed_captured_file_membership(scenario): + inputs = {"requirements.txt", "verify_activity_install.py", "conftest.py"} + inputs |= consumer.CASE_FILES.get(scenario, set()) + packet = { + "scenario": scenario, "files": dict.fromkeys(inputs, {}), "wheels": {"sdk.whl": {"sha256": "a" * 64}}, + "wheel_name": "sdk.whl", "wheel_sha256": "a" * 64, "sdk_version": "test", + } + if scenario == "foreign": + with pytest.raises(ValueError, match="scenario differs"): + consumer.validate_setup(packet) + return + consumer.validate_setup(packet) + packet["files"]["unreviewed.py"] = {} + with pytest.raises(ValueError, match="membership differs"): + consumer.validate_setup(packet) + + def test_sdk_writable_verifier_replacement_cannot_change_the_retained_checker(monkeypatch, tmp_path): original = "def verify_sdk(*args, **kwargs): return {'checker': 'Git-captured'}\n" monkeypatch.setattr(consumer, "CAPTURED_VERIFIER", original) diff --git a/tests/test_activity_acceptance_entry.py b/tests/test_activity_acceptance_entry.py new file mode 100644 index 0000000..1ff9eeb --- /dev/null +++ b/tests/test_activity_acceptance_entry.py @@ -0,0 +1,121 @@ +"""Inert first-open controls; no SDK, API, Worker or Docker invocation.""" + +from __future__ import annotations + +import hashlib +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest +from scripts.ci import activity_acceptance_consumer as consumer +from scripts.ci import activity_acceptance_entry as entry + + +@pytest.fixture +def sandbox(tmp_path, monkeypatch): + environment, data = tmp_path / "environment", tmp_path / "data" + site = environment / "lib" / ("python" + str(sys.version_info.major) + "." + str(sys.version_info.minor)) + site = site / "site-packages" + site.mkdir(parents=True) + data.mkdir() + case = data / "test_captured.py" + case.write_text("async def cold_worker(p): print('fixed inert result', p['marker'])\n") + source = Path(entry.__file__).read_text() + source = source.replace('Path("/run/acceptance-env")', "Path(" + repr(str(environment)) + ")") + source = source.replace('Path("/run/acceptance-data")', "Path(" + repr(str(data)) + ")") + monkeypatch.setattr(consumer, "CAPTURED_ENTRY", source) + monkeypatch.setattr(entry, "ENVIRONMENT", environment) + monkeypatch.setattr(entry, "DATA", data) + monkeypatch.setattr(entry, "CASE_NAME", case.name) + return environment, data, site, case + + +def launch(case, installed, *, no_site=True): + source = consumer.entry_source( + "worker", case.name, hashlib.sha256(case.read_bytes()).hexdigest(), installed, + ) + arguments = [sys.executable, "-I", *(["-S"] if no_site else []), "-B", "-c", source] + return subprocess.run( + arguments, input=json.dumps({"inventory": installed, "worker": {"marker": "owned"}}), + env={"PATH": os.defpath}, capture_output=True, text=True, timeout=10, + ) + + +def test_no_site_entry_can_execute_only_the_verified_inert_source(sandbox): + environment, _data, _site, case = sandbox + installed = entry.snapshot(environment) + result = launch(case, installed) + assert result.returncode == 0 + assert result.stdout == "fixed inert result owned\n" + + +def test_entry_refuses_normal_site_even_when_other_inputs_match(sandbox): + environment, _data, _site, case = sandbox + result = launch(case, entry.snapshot(environment), no_site=False) + assert result.returncode != 0 + assert "no-site Python" in result.stderr + assert not result.stdout + + +def test_added_startup_hook_refuses_before_any_case_code(sandbox): + environment, _data, site, case = sandbox + installed = entry.snapshot(environment) + (site / "sitecustomize.py").write_text("raise RuntimeError('startup replacement executed')\n") + result = launch(case, installed) + assert result.returncode != 0 + assert "install inventory changed" in result.stderr + assert "startup replacement executed" not in result.stderr + assert not result.stdout + + +def test_predecessor_runs_pth_before_guard_successor_refuses_source_before_effect(sandbox, tmp_path): + environment, _data, site, case = sandbox + subprocess.run( + [sys.executable, "-I", "-B", "-m", "venv", "--without-pip", "--copies", str(environment)], + check=True, capture_output=True, text=True, timeout=30, + ) + marker = tmp_path / "pre-open-marker" + (site / "preopen.pth").write_text( + "import pathlib; pathlib.Path(" + repr(str(marker)) + ").write_text('pre-open')\n", + ) + installed = entry.snapshot(environment) + source = consumer.entry_source("worker", case.name, hashlib.sha256(case.read_bytes()).hexdigest(), installed) + old = subprocess.run( + [str(environment / "bin/python"), "-I", "-B", "-c", "raise RuntimeError('late guard')"], + env={"PATH": os.defpath}, capture_output=True, text=True, timeout=10, + ) + assert old.returncode != 0 + assert marker.read_text() == "pre-open" + marker.unlink() + case.write_text("raise RuntimeError('foreign case executed')\n") + corrected = subprocess.run( + [sys.executable, "-I", "-S", "-B", "-c", source], + input=json.dumps({"inventory": installed, "worker": {"marker": "owned"}}), + env={"PATH": os.defpath}, capture_output=True, text=True, timeout=10, + ) + assert corrected.returncode != 0 + assert "captured case source differs" in corrected.stderr + assert "foreign case executed" not in corrected.stderr + assert not marker.exists() + assert not corrected.stdout + + +def test_post_capture_path_replacement_keeps_the_verified_bytes(sandbox): + _environment, _data, _site, case = sandbox + case.write_text("chosen = 'captured'\n") + source = entry.guarded_source(case, hashlib.sha256(case.read_bytes()).hexdigest()) + case.write_text("raise RuntimeError('replacement must not execute')\n") + assert entry.source_module(source).chosen == "captured" + + +def test_pytest_collector_uses_captured_bytes_and_no_mutable_conftest(sandbox, monkeypatch): + _environment, data, _site, case = sandbox + source = b"def test_captured(): assert 2 + 2 == 4\n" + case.write_text("raise RuntimeError('mutable test must not execute')\n") + (data / "conftest.py").write_text("raise RuntimeError('mutable conftest must not execute')\n") + monkeypatch.setenv("PYTEST_DISABLE_PLUGIN_AUTOLOAD", "1") + assert entry.run_pytest(source, {}) == 0 + assert (data / "identity.xml").is_file() diff --git a/tests/test_native_timer_redrive_cell.py b/tests/test_native_timer_redrive_cell.py new file mode 100644 index 0000000..5c524d3 --- /dev/null +++ b/tests/test_native_timer_redrive_cell.py @@ -0,0 +1,345 @@ +"""Offline contract controls for the separate installed cold-Worker cell.""" + +from __future__ import annotations + +import hashlib +import json +from copy import deepcopy +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from tests.integration import test_native_timer_redrive as cell +from tests.test_native_timer_redrive_clock import successor + + +def prefix(): + previous, current = successor(), successor() + source_time = cell.instant("2026-11-03T00:00:06.654321+00:00") + source_start = cell.instant("2026-11-03T00:00:00+00:00") + for document in (previous, current): + for event in document["history"]: + event["timestamp"] = event.pop("recorded_at") + for event in previous["history"]: + if event["event_type"] in {"TimerScheduled", "TimerFired"}: + event["payload"]["timer_id"] = "old-timer" + event["payload"].pop("reused_recorded_at", None) + event["payload"].pop("reused_from_run_id", None) + cell.one(previous["history"], "WorkflowStarted")["payload"].pop("replayed_started_at", None) + cell.one(previous["history"], "TimerFired")["timestamp"] = "2026-11-03T00:00:06.654321Z" + events = current["history"] + cell.one(events, "WorkflowStarted")["payload"]["continued_from_run_id"] = "previous" + fired = cell.one(events, "TimerFired")["payload"] + fired["reused_from_run_id"] = "previous" + fired["reused_recorded_at"] = source_time.isoformat() + completed = cell.one(events, "ActivityCompleted")["payload"] + completed["reused_from_run_id"] = "previous" + completed["reused_activity_execution_id"] = cell.one(previous["history"], "ActivityCompleted")["payload"][ + "activity_execution_id" + ] + return events, previous["history"], { + "predecessor_id": "previous", "source_clock": source_time, "source_start": source_start, + "used_timer_ids": {"old-timer"}, + "source_clock_text": "2026-11-03T00:00:06.654321Z", "source_start_text": "2026-11-03T00:00:00.000000Z", + } + + +def test_exact_fraction_and_immediate_predecessor_preserve_unchanged_inputs(): + events, previous, arguments = prefix() + original = deepcopy((events, previous, arguments)) + assert cell.verify_reuse(events, previous, **arguments) + assert original == (events, previous, arguments) + + +@pytest.mark.parametrize("reformat", [None, "clock", "start"]) +def test_repeated_reuse_preserves_inherited_strings_as_well_as_instants(reformat): + events, previous, arguments = prefix() + original_clock = arguments["source_clock"].isoformat() + original_start = arguments["source_start"].isoformat() + cell.one(previous, "TimerFired")["payload"]["reused_recorded_at"] = original_clock + cell.one(previous, "WorkflowStarted")["payload"]["replayed_started_at"] = original_start + if reformat == "clock": + cell.one(events, "TimerFired")["payload"]["reused_recorded_at"] = original_clock.replace("+00:00", "Z") + elif reformat == "start": + cell.one(events, "WorkflowStarted")["payload"]["replayed_started_at"] = original_start.replace("+00:00", "Z") + if reformat: + assert cell.instant(cell.one(events, "TimerFired")["payload"]["reused_recorded_at"]) == arguments[ + "source_clock" + ] + assert cell.instant(cell.one(events, "WorkflowStarted")["payload"]["replayed_started_at"]) == arguments[ + "source_start" + ] + with pytest.raises(AssertionError): + cell.verify_reuse(events, previous, **arguments) + else: + cell.verify_reuse(events, previous, **arguments) + + +@pytest.mark.parametrize("fraction,rendered", [ + ("000000", ""), ("654000", ".654"), ("654321", ".654321"), +]) +def test_first_reuse_accepts_native_offset_and_precision_rendering(fraction, rendered): + events, previous, arguments = prefix() + arguments["source_clock_text"] = f"2026-11-03T00:00:06.{fraction}Z" + arguments["source_clock"] = cell.instant(arguments["source_clock_text"]) + arguments["source_start_text"] = f"2026-11-03T00:00:00.{fraction}Z" + arguments["source_start"] = cell.instant(arguments["source_start_text"]) + cell.one(events, "TimerFired")["payload"]["reused_recorded_at"] = f"2026-11-03T00:00:06{rendered}+00:00" + cell.one(events, "WorkflowStarted")["payload"]["replayed_started_at"] = f"2026-11-03T00:00:00{rendered}+00:00" + cell.verify_reuse(events, previous, **arguments) + + +@pytest.mark.parametrize("field", ["clock", "start"]) +def test_first_reuse_refuses_changed_instant(field): + events, previous, arguments = prefix() + kind, name, source_name = ( + ("TimerFired", "reused_recorded_at", "source_clock") if field == "clock" + else ("WorkflowStarted", "replayed_started_at", "source_start") + ) + payload = cell.one(events, kind)["payload"] + payload[name] = arguments[source_name].replace(microsecond=1).isoformat() + with pytest.raises(AssertionError): + cell.verify_reuse(events, previous, **arguments) + + +def test_source_boundary_uses_timestamp_only_public_history_shape(): + events, previous, arguments = prefix() + assert all("timestamp" in event and "recorded_at" not in event for event in previous) + text, clock = cell.source_timer_boundary(previous) + assert (text, clock) == (arguments["source_clock_text"], arguments["source_clock"]) + arguments.update(source_clock_text=text, source_clock=clock) + cell.verify_reuse(events, previous, **arguments) + + +def test_export_recorded_at_cannot_replace_public_timestamp(): + _, previous, _ = prefix() + fired = cell.one(previous, "TimerFired") + fired["recorded_at"] = fired.pop("timestamp") + with pytest.raises(KeyError, match="timestamp"): + cell.source_timer_boundary(previous) + + +@pytest.mark.parametrize("fault", [ + "floor", "root_lineage", "new_start", "pair_id", "old_id", "timer_task", "scheduled_activity", "changed_result", + "extra_fired", "extra_scheduled", "changed_delay", +]) +def test_false_reuse_evidence_refuses(fault): + events, previous, arguments = prefix() + fired = cell.one(events, "TimerFired")["payload"] + if fault == "floor": + fired["reused_recorded_at"] = "2026-11-03T00:00:06+00:00" + elif fault == "root_lineage": + fired["reused_from_run_id"] = "root-instead-of-immediate-predecessor" + elif fault == "new_start": + cell.one(events, "WorkflowStarted")["payload"]["replayed_started_at"] = "2026-11-03T00:05:00Z" + elif fault == "pair_id": + fired["timer_id"] = "different-pair" + elif fault == "old_id": + arguments["used_timer_ids"].add(fired["timer_id"]) + elif fault == "timer_task": + fired["task"] = {"id": "new-timer-work"} + elif fault == "scheduled_activity": + events.append({"event_type": "ActivityScheduled", "payload": {"sequence": 1}}) + elif fault == "extra_fired": + fired["unreviewed"] = True + elif fault == "extra_scheduled": + cell.one(events, "TimerScheduled")["payload"]["unreviewed"] = True + elif fault == "changed_delay": + cell.one(events, "TimerScheduled")["payload"]["delay_seconds"] = 999 + else: + cell.one(events, "ActivityCompleted")["payload"]["result"] = "changed" + with pytest.raises(AssertionError): + cell.verify_reuse(events, previous, **arguments) + + +@pytest.mark.parametrize("clock", ["2026-11-03T00:00:06.654321001Z", "2026-11-03T00:00:06", ""]) +def test_unrepresentable_or_unzoned_clock_cannot_hide_precision_loss(clock): + with pytest.raises(AssertionError): + cell.instant(clock) + + +def test_changed_definition_or_absent_trusted_entry_refuses_before_child(monkeypatch): + with pytest.raises(AssertionError): + cell.definition("changed definition") + monkeypatch.setattr(cell.subprocess, "run", lambda *a, **k: pytest.fail("must refuse before any child")) + with pytest.raises(AssertionError): + cell.run_cold_worker({}) + + +def test_accepted_workflow_definition_bytes_remain_frozen(): + path = Path(__file__).parent / "integration/redrive_timer_clock_workflow.py" + assert hashlib.sha256(path.read_bytes()).hexdigest() == cell.DEFINITION_SHA + + +@pytest.mark.parametrize("fault", [ + None, "timer_command", "extra_command", "extra_completion", "missing_task", "foreign_task", + "first_callback", "no_task", "floor", "empty_fingerprint", +]) +def test_redrive_worker_evidence_requires_only_the_failed_activity_and_exact_clock(fault): + clock = cell.instant("2026-11-03T00:00:06.654321Z") + observed = { + "definition_sha256": cell.DEFINITION_SHA, "fingerprints": {cell.WORKFLOW_TYPE: "sha256:" + "a" * 64}, + "worker_id": "fixed-worker", "pid": 123, + "status": "failed", + "completions": [ + {"task_id": "one", "commands": [cell.ScheduleActivity( + "tests.redrive.second", ["Hello, Taylor!", clock.isoformat(timespec="microseconds")], + retry_policy=cell.ActivityRetryPolicy(max_attempts=1), + ).to_server_command("queue")]}, + {"task_id": "two", "commands": [{"type": "fail_workflow", "message": "fixed failure", + "failed_step_sequence": 3, "failed_activity_execution_id": "execution"}]}, + ], + "callbacks": [{"activity": "second", "clock": clock.isoformat(), "execution_id": "execution", + "task_id": "activity", "attempt_id": "attempt", "run_id": "run"}], + } + exported = { + "workflow": {"run_id": "run"}, + "tasks": [{"id": "one", "type": "workflow"}, {"id": "two", "type": "workflow"}, + {"id": "activity", "type": "activity"}], + } + if fault == "timer_command": + observed["completions"][0]["commands"].append({"type": "start_timer"}) + elif fault == "first_callback": + observed["callbacks"].insert(0, {"activity": "first"}) + elif fault == "no_task": + observed["completions"] = [] + elif fault == "extra_command": + observed["completions"][0]["commands"].append({"type": "upsert_search_attributes"}) + elif fault == "extra_completion": + observed["completions"].append(deepcopy(observed["completions"][0])) + elif fault == "missing_task": + observed["completions"][0]["task_id"] = "" + elif fault == "foreign_task": + observed["completions"][0]["task_id"] = "foreign" + elif fault == "empty_fingerprint": + observed["fingerprints"][cell.WORKFLOW_TYPE] = "" + elif fault == "floor": + observed["callbacks"][0]["clock"] = "2026-11-03T00:00:06Z" + if fault: + with pytest.raises(AssertionError): + cell.verify_worker(observed, redrive=True, clock=clock, queue="queue", exported=exported) + else: + cell.verify_worker(observed, redrive=True, clock=clock, queue="queue", exported=exported) + + +@pytest.mark.parametrize("fault", [None, "empty", "changed", "source"]) +def test_recorded_worker_fingerprint_is_exact_across_original_and_successors(fault): + fingerprint = "sha256:" + "a" * 64 + observations = [{"fingerprints": {cell.WORKFLOW_TYPE: fingerprint}} for _ in range(3)] + histories = [[{"event_type": "WorkflowStarted", "payload": { + "workflow_definition_fingerprint": fingerprint, "workflow_definition_fingerprint_source": "worker", + }}] for _ in range(3)] + if fault == "empty": + observations[0]["fingerprints"][cell.WORKFLOW_TYPE] = "" + elif fault == "changed": + histories[1][0]["payload"]["workflow_definition_fingerprint"] = "sha256:" + "b" * 64 + elif fault == "source": + histories[2][0]["payload"]["workflow_definition_fingerprint_source"] = "unrelated" + if fault: + with pytest.raises(AssertionError): + cell.verify_recorded_fingerprints(observations, histories) + else: + cell.verify_recorded_fingerprints(observations, histories) + + +@pytest.mark.parametrize("fault", [None, "done_timer", "cancelled_wait", "pruned", "run", "redacted", "incomplete"]) +def test_export_requires_bound_retained_all_status_inventory(fault): + document = { + "schema": "durable-workflow.v2.history-export", "schema_version": 2, "history_complete": True, + "workflow": {"instance_id": "workflow", "run_id": "run", "workflow_type": cell.WORKFLOW_TYPE, + "details_state": "retained", "details_pruned_at": None}, + "selected_run": {"waits_projection_source": "workflow_run_waits"}, "redaction": {"applied": False}, + "tasks": [{"id": "workflow-task", "type": "workflow", "status": "completed"}], "waits": [], + "timers": [{"id": "history-reconstructed-pair"}], + } + if fault == "done_timer": + document["tasks"].append({"id": "timer", "type": "timer", "status": "completed"}) + elif fault == "cancelled_wait": + document["waits"].append({"kind": "timer", "status": "cancelled"}) + elif fault == "pruned": + document["workflow"]["details_state"] = "pruned" + elif fault == "run": + document["workflow"]["run_id"] = "foreign" + elif fault == "redacted": + document["redaction"]["applied"] = True + elif fault == "incomplete": + document["history_complete"] = False + if fault: + with pytest.raises(AssertionError): + cell.verify_export(document, "workflow", "run", terminal=True, source=False) + else: + cell.verify_export(document, "workflow", "run", terminal=True, source=False) + + +@pytest.mark.parametrize("fault", [None, "absent_admitted", "wrong_refusal", "positive_failure"]) +async def test_matching_nonconsuming_registration_is_required_and_fenced_cleanup_runs(fault): + fingerprint = "sha256:" + "a" * 64 + events = [{"event_type": "WorkflowStarted", "payload": { + "workflow_definition_fingerprint": fingerprint, "workflow_definition_fingerprint_source": "worker", + }}] + observed = { + "fingerprints": {cell.WORKFLOW_TYPE: fingerprint}, "registration_namespace": "default", + "registration": {"worker_id": "original", "task_queue": "queue", + "supported_workflow_types": [cell.WORKFLOW_TYPE], + "workflow_definition_fingerprints": {cell.WORKFLOW_TYPE: fingerprint}, + "capability_manifest": {}}, + } + + class FakeClient: + namespace = "default" + + def __init__(self): + self.calls = [] + self.registered = False + + async def list_workflow_runs(self, queue): + return SimpleNamespace(runs=[SimpleNamespace(run_id="source")]) + + async def iter_history(self, queue, run_id): + for event in events: + yield event + + async def register_worker(self, **request): + self.calls.append(("register", request)) + self.registered = True + self.worker_id = request["worker_id"] + return {"worker_id": self.worker_id, "registered": True, "registration_token": "b" * 32, + "namespace": "default", "task_queue": "queue"} + + async def redrive_workflow(self, queue, run_id): + self.calls.append(("redrive", self.registered)) + if not self.registered and fault != "absent_admitted": + raise cell.ServerError(409, { + "reason": "other" if fault == "wrong_refusal" else "workflow_definition_unavailable_or_changed", + "outcome": "rejected", "workflow_id": queue, "run_id": run_id, + "workflow_run_id": None, "accepted": False, "task_id": None, + }) + if self.registered and fault == "positive_failure": + raise cell.ServerError(503, {"reason": "fixed admission failure"}) + return SimpleNamespace(outcome="redriven", run_id="next") + + async def deregister_worker_registration(self, worker_id, **arguments): + assert worker_id == self.worker_id + assert arguments == { + "registration_token": "b" * 32, "expected_namespace": "default", "timeout": 10, + } + self.calls.append(("deregister", worker_id)) + self.registered = False + return {"worker_id": worker_id, "registration_token": "b" * 32, + "outcome": "deregistered", "recovered_workflow_task_count": 0} + + client = FakeClient() + if fault: + with pytest.raises(cell.ServerError if fault == "positive_failure" else AssertionError): + await cell.admit_redrive(client, "queue", "source", observed, events) + assert not client.registered + else: + receipt, evidence = await cell.admit_redrive(client, "queue", "source", observed, events) + assert receipt.run_id == "next" + assert not client.registered + assert [name for name, _value in client.calls] == ["redrive", "register", "redrive", "deregister"] + assert "b" * 32 not in json.dumps(evidence) + assert evidence["matching_registration"]["workflow_definition_fingerprints"] == { + cell.WORKFLOW_TYPE: fingerprint, + } diff --git a/tests/test_published_activity_acceptance.py b/tests/test_published_activity_acceptance.py index 31078ab..1dbe4a8 100644 --- a/tests/test_published_activity_acceptance.py +++ b/tests/test_published_activity_acceptance.py @@ -374,6 +374,17 @@ def test_complete_case_requires_the_actual_single_retry_test_to_pass(): runner.verify_case_result(case_result()) +def test_timer_scenario_cannot_borrow_a_passing_identity_receipt(monkeypatch): + monkeypatch.setattr(runner, "SCENARIO", "timer-redrive") + result = case_result() + with pytest.raises(ValueError, match="inventory differs"): + runner.verify_case_result(result) + result["junit"] = result["junit"].replace( + "test_retry_preserves_received_execution_id_and_changes_attempt_id", runner.SCENARIOS["timer-redrive"][1], + ) + runner.verify_case_result(result) + + @pytest.mark.parametrize("change", ["skip", "failure", "error", "wrong_case", "empty", "reported_failure"]) def test_exit_zero_cannot_qualify_a_skipped_missing_or_different_case(change): result = case_result()