diff --git a/CHANGELOG.md b/CHANGELOG.md index 51caffe..ce3c4ff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [2.14.1] - 2026-11-04 + +### Fixed + +- `wait_condition` predicate fingerprints no longer change across Python versions or worker processes. Every earlier release hashed the predicate's bytecode and the `repr` of its constants. So upgrading Python (for example 3.12 to 3.13) failed replay of every in-flight run at its first wait, and a predicate containing a nested function (a generator expression, a nested lambda, or a comprehension on 3.10 and 3.11) got a new fingerprint in every process, so a restart or cache loss failed replay with `condition_fingerprint_changed`. New waits record a v2 fingerprint (`dwpy2:sha256:...`) computed from the predicate's module, qualified name and a canonical form of its source AST, with no bytecode, addresses, paths or line numbers. Without source (for example `eval` or a REPL) no fingerprint is sent, and the wait is identified by its key and call order. A v1 value recorded by an earlier release still matches when this process computes the same v1; otherwise it is accepted as unverifiable instead of failing replay, so drift in those predicates is no longer detected. A reopen sends the fingerprint its first open recorded, so a wait opened before the upgrade keeps its v1 value, and a keyless wait links a v1 value only within its own chain. A keyless wait without source links its false-wake reopens within its chain, as before. **Rollback and mixed fleets:** a run that opened a wait under this release records a v2 fingerprint, which 2.14.0 and earlier refuse on replay. Don't roll a task queue back below this release once such waits exist, and upgrade every worker on a queue together. (#258) + +- Same-line `wait_condition` lambdas whose bodies contain list, dict or set comprehensions share one combined fingerprint on every Python version. Put those predicates on separate lines to detect a change from one candidate to another. (#258) + ## [2.14.0] - 2026-11-04 **Upgrade from 2.13.2 and earlier.** Every earlier release fails replay when a condition wait has an orphaned duplicate open whose timer fires later, which Server 3.0.x can record (#253 below). diff --git a/pyproject.toml b/pyproject.toml index b2e87a1..224c4de 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "durable-workflow" -version = "2.14.0" +version = "2.14.1" description = "Python client and worker SDK for Durable Workflow Cloud and self-hosted Server" readme = "README.md" requires-python = ">=3.10" @@ -79,8 +79,8 @@ durable-workflow-replay-conformance = "durable_workflow.replay_conformance:main" durable-workflow-workflow-updates-conformance = "durable_workflow.workflow_updates_conformance:main" [tool.durable-workflow] -product-train = "2.14.0" -registry-version = "2.14.0" +product-train = "2.14.1" +registry-version = "2.14.1" supported-server-versions = "2.8.0" supported-server-release-baselines = { "2" = "2.8.0", "3" = "3.0.0" } worker-protocol-version = "1.20" diff --git a/src/durable_workflow/_condition_fingerprint.py b/src/durable_workflow/_condition_fingerprint.py new file mode 100644 index 0000000..690d920 --- /dev/null +++ b/src/durable_workflow/_condition_fingerprint.py @@ -0,0 +1,229 @@ +"""Stable identity for ``wait_condition`` predicates (#258). + +The Server records the fingerprint a worker sends with ``open_condition_wait`` +and replay compares it, so the value must be the same on every Python version +and in every process. Version 2 hashes the predicate's module, qualified name +and a canonical form of its source AST. It never hashes bytecode, constant +reprs, memory addresses, file paths or line numbers. + +Version 1 (every release through 2.14.0) hashed bytecode and ``repr`` of code +constants, which change across Python versions and, for nested functions, +across processes. Recorded v1 values cannot be recomputed portably, so replay +treats a v1 value that differs from this process's v1 as unverifiable rather +than as a changed predicate (see :func:`fingerprint_verdict`). +""" + +from __future__ import annotations + +import ast +import hashlib +import inspect +import linecache +import re +from collections.abc import Callable +from typing import Any, Literal + +V2_PREFIX = "dwpy2:sha256:" +_LEGACY = re.compile(r"sha256:[0-9a-f]{64}\Z") +# Fields that carry no predicate meaning or differ between Python versions. +_SKIPPED_FIELDS = frozenset({"ctx", "kind", "type_comment", "type_params", "decorator_list", "returns"}) +_parsed: dict[tuple[str, int], ast.Module | None] = {} +_nodes: dict[Any, str | None] = {} +# Python 3.12 inlines list, dict and set comprehensions (PEP 709), which drops +# their segment from a nested function's __qualname__. +_INLINED_COMPREHENSION = re.compile(r"\.<(?:listcomp|dictcomp|setcomp)>") + +Verdict = Literal["match", "mismatch", "unverifiable"] + + +def legacy_fingerprint(predicate: Callable[[], bool]) -> str: + """The v1 fingerprint, kept only to recognise histories recorded by 2.14.0 and earlier.""" + h = hashlib.sha256() + h.update(b"durable-workflow-python.wait-condition.v1\0") + h.update(f"{getattr(predicate, '__module__', '')}\0".encode()) + h.update(f"{getattr(predicate, '__qualname__', '')}\0".encode()) + original = getattr(predicate, "__typed_workflow_method__", None) + code = getattr(original if original is not None else predicate, "__code__", None) + if code is None: + h.update(repr(predicate).encode()) + else: + h.update(repr(( + code.co_argcount, code.co_posonlyargcount, code.co_kwonlyargcount, code.co_code, + code.co_consts, code.co_names, code.co_varnames, code.co_freevars, + )).encode()) + return f"sha256:{h.hexdigest()}" + + +def is_legacy(value: Any) -> bool: + return isinstance(value, str) and _LEGACY.match(value) is not None + + +def fingerprint(predicate: Callable[[], bool]) -> str | None: + """Return the v2 fingerprint, or None when the predicate's source is unavailable. + + Without a fingerprint the wait is identified by its key and call order, + as in sdk-php. + """ + function = _authored_function(predicate) + if function is None: + return None + code = function.__code__ + if code not in _nodes: + _nodes[code] = _predicate_node(function) + node = _nodes[code] + if node is None: + return None + qualname = _INLINED_COMPREHENSION.sub("", getattr(function, "__qualname__", "")) + h = hashlib.sha256() + h.update(b"durable-workflow-python.wait-condition.v2\0") + h.update(f"{getattr(function, '__module__', '')}\0".encode()) + h.update(f"{qualname}\0".encode()) + h.update(node.encode()) + return f"{V2_PREFIX}{h.hexdigest()}" + + +def fingerprint_verdict(recorded: Any, authored: str | None, predicate: Callable[[], bool]) -> Verdict: + """Compare a recorded fingerprint with the authored predicate. + + - No recorded value: nothing to compare. + - An identity supplied explicitly in another format (not v2, not None) + is compared exactly, as before. + - A v2 value is compared exactly; without source it cannot be checked. + - A v1 value equal to this process's v1 matches. Any other v1 value is + unverifiable, because v1 differs across Python versions and processes. + - Any other string is compared exactly, as before. + """ + if not isinstance(recorded, str): + return "match" + if authored is not None and not authored.startswith(V2_PREFIX): + return "match" if recorded == authored else "mismatch" + if recorded.startswith(V2_PREFIX): + if authored is None: + return "unverifiable" + return "match" if recorded == authored else "mismatch" + if is_legacy(recorded): + return "match" if recorded == legacy_fingerprint(predicate) else "unverifiable" + return "match" if recorded == authored else "mismatch" + + +def _authored_function(predicate: Any) -> Any: + target = getattr(predicate, "__typed_workflow_method__", None) or predicate + if inspect.ismethod(target): + target = target.__func__ + target = getattr(target, "__typed_workflow_method__", None) or target + return target if inspect.isfunction(target) else None + + +def _module_tree(function: Any) -> ast.Module | None: + try: + filename = inspect.getsourcefile(function) or inspect.getfile(function) + except (OSError, TypeError): + return None + lines = linecache.getlines(filename, function.__globals__) + if not lines: + return None + source = "".join(lines) + key = (filename, hash(source)) + if key not in _parsed: + try: + _parsed[key] = ast.parse(source) + except (SyntaxError, ValueError): + _parsed[key] = None + return _parsed[key] + + +def _predicate_node(function: Any) -> str | None: + tree = _module_tree(function) + if tree is None: + return None + code = function.__code__ + first_line = code.co_firstlineno + if function.__name__ == "": + lambdas = [node for node in ast.walk(tree) if isinstance(node, ast.Lambda) and node.lineno == first_line] + lambdas.sort(key=lambda node: node.col_offset) + # PEP 709 moves comprehension instructions into the enclosing lambda. + # Candidates indistinguishable before 3.12 can become distinct after + # inlining, so keep the combined AST on every version in this case. + has_inlined_comprehension = any( + isinstance(child, ast.ListComp | ast.DictComp | ast.SetComp) + for node in lambdas for child in ast.walk(node) + ) + if len(lambdas) > 1 and not has_inlined_comprehension: + # Names and constants are stable across versions; bytecode only + # breaks a tie, and is compared within this interpreter. + matching = [node for node in lambdas if _compiles_to(node, code, compare_bytecode=False)] + if len(matching) > 1: + matching = [node for node in matching if _compiles_to(node, code, compare_bytecode=True)] + if len(matching) == 1: + lambdas = matching + if not lambdas: + return None + # Lambdas that start on the same line and compile identically share + # one fingerprint covering all of them. + return "|".join(_canonical(node) for node in lambdas) + definitions = [ + node for node in ast.walk(tree) + if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef) and node.name == function.__name__ + and min([node.lineno, *(decorator.lineno for decorator in node.decorator_list)]) == first_line + ] + if len(definitions) != 1: + return None + return _canonical(definitions[0]) + + +def _compiles_to(node: ast.Lambda, code: Any, *, compare_bytecode: bool) -> bool: + """Whether *node* compiles to *code* on this interpreter. + + The candidate is compiled inside a function that binds the live code's + free variables, so names resolve the same way. Only the choice of node + uses bytecode, which is valid within one interpreter; the fingerprint + hashes the node's AST, so it stays portable. + """ + free = list(code.co_freevars) + lines = ["def __dw_outer__():"] + if free: + lines.append(f" {' = '.join(free)} = None") + lines.append(f" return ({ast.unparse(node)})") + try: + outer = compile("\n".join(lines), "", "exec") + except SyntaxError: + return False + function = next((const for const in outer.co_consts if hasattr(const, "co_code")), None) + candidate = next( + (const for const in getattr(function, "co_consts", ()) if getattr(const, "co_name", None) == ""), + None, + ) + return candidate is not None and ( + (not compare_bytecode or candidate.co_code == code.co_code) + and candidate.co_argcount == code.co_argcount + and candidate.co_names == code.co_names + and candidate.co_varnames == code.co_varnames + and candidate.co_freevars == code.co_freevars + and _plain_constants(candidate) == _plain_constants(code) + ) + + +def _plain_constants(code: Any) -> list[Any]: + """Constants other than nested code and the qualnames Python 3.10 stores for them.""" + return [ + const for const in code.co_consts + if not hasattr(const, "co_code") and not (isinstance(const, str) and "" in const) + ] + + +def _canonical(value: Any) -> str: + if isinstance(value, ast.AST): + parts = [type(value).__name__] + for name in value._fields: + if name in _SKIPPED_FIELDS: + continue + item = getattr(value, name, None) + if item is None or item == []: + continue + parts.append(f"{name}={_canonical(item)}") + return "(" + " ".join(parts) + ")" + if isinstance(value, list): + return "[" + ",".join(_canonical(item) for item in value) + "]" + if value is None or value is Ellipsis or isinstance(value, str | bytes | int | float | complex | bool): + return f"{type(value).__name__}:{value!r}" + return type(value).__name__ diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 0aaf5e0..baf5f90 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -42,6 +42,8 @@ CommittedCancellationScopeHistory, ScopeBoundary, ) +from ._condition_fingerprint import fingerprint as _stable_condition_fingerprint +from ._condition_fingerprint import fingerprint_verdict, is_legacy from ._cooperative_cancellation import CancellationDelivery, read_cancellation_history from ._failure_reporting import exception_message from ._typed_values import decode_payload_value, to_payload_values, validate_annotation @@ -1720,28 +1722,9 @@ def to_server_command( } -def _condition_predicate_fingerprint(predicate: Callable[[], bool]) -> str: - h = hashlib.sha256() - h.update(b"durable-workflow-python.wait-condition.v1\0") - h.update(f"{getattr(predicate, '__module__', '')}\0".encode()) - h.update(f"{getattr(predicate, '__qualname__', '')}\0".encode()) - - code = getattr(predicate, "__code__", None) - if code is None: - h.update(repr(predicate).encode()) - else: - h.update(repr(( - code.co_argcount, - code.co_posonlyargcount, - code.co_kwonlyargcount, - code.co_code, - code.co_consts, - code.co_names, - code.co_varnames, - code.co_freevars, - )).encode()) - - return f"sha256:{h.hexdigest()}" +def _condition_predicate_fingerprint(predicate: Callable[[], bool]) -> str | None: + """The stable v2 predicate fingerprint, or None without source (#258).""" + return _stable_condition_fingerprint(predicate) Command = ( @@ -6017,10 +6000,7 @@ def _condition_wait_mismatch( ) opened_fingerprint = opened.get("condition_definition_fingerprint") - if ( - isinstance(opened_fingerprint, str) - and cmd.condition_definition_fingerprint != opened_fingerprint - ): + if _fingerprint_verdict(opened, cmd) == "mismatch": return NonDeterministicReplayError( step.workflow_sequence, "condition wait", step.event_types, detail=( @@ -6033,7 +6013,19 @@ def _condition_wait_mismatch( return None - def _same_logical_condition_wait(opened: Mapping[str, Any], cmd: WaitCondition) -> bool: + def _fingerprint_verdict(opened: Mapping[str, Any], cmd: WaitCondition) -> str: + return fingerprint_verdict( + opened.get("condition_definition_fingerprint"), cmd.condition_definition_fingerprint, cmd.predicate, + ) + + def _same_logical_condition_wait( + opened: Mapping[str, Any], + cmd: WaitCondition, + *, + chain_first: Mapping[str, Any] | None = None, + chain_current: Mapping[str, Any] | None = None, + next_index: int | None = None, + ) -> bool: if _condition_wait_mismatch(opened, cmd) is not None: return False @@ -6042,10 +6034,33 @@ def _same_logical_condition_wait(opened: Mapping[str, Any], cmd: WaitCondition) return True opened_fingerprint = opened.get("condition_definition_fingerprint") + if not isinstance(opened_fingerprint, str) or not opened_fingerprint: + # A keyless wait without source has no identity beyond call order. + # Within its own chain (both opens recorded without a fingerprint) + # it is the same wait, as 2.14.0 linked it by its process-local v1. + first_fingerprint = ( + chain_first.get("condition_definition_fingerprint") if chain_first is not None else "absent" + ) + return ( + cmd.condition_definition_fingerprint is None and not cmd.condition_key + and chain_first is not None and not first_fingerprint + ) + verdict = _fingerprint_verdict(opened, cmd) + if verdict == "match": + return True + if verdict != "unverifiable": + return False + # A keyless wait whose fingerprint cannot be checked (a v1 value from + # another Python version or process, #258) links only within its own + # chain: the same recorded v1 value as the chain's first open, or a + # duplicate opened while the current wait was still open (#253). It + # never binds a recorded wait to another call. + first_fingerprint = chain_first.get("condition_definition_fingerprint") if chain_first is not None else None + if is_legacy(opened_fingerprint) and opened_fingerprint == first_fingerprint: + return True return ( - isinstance(opened_fingerprint, str) - and bool(opened_fingerprint) - and opened_fingerprint == cmd.condition_definition_fingerprint + chain_current is not None and next_index is not None + and _opened_before_terminal(next_index, chain_current) ) result_cursor = 0 @@ -6111,6 +6126,10 @@ def _refresh_replay_logging() -> None: pending: list[Command] = [] advanced_cmd: Any = None terminal_condition_reopen_cmd: WaitCondition | None = None + # The first open of the chain a terminal reopen cleanup belongs to (#258). + terminal_condition_chain_first: Mapping[str, Any] | None = None + # Its terminal position proves whether a different-process duplicate belongs. + terminal_condition_chain_current: Mapping[str, Any] | None = None version_decisions: dict[str, int] = {} def _version_error(detail: str, step: _RecordedStep | None = None) -> NonDeterministicReplayError: @@ -6686,7 +6705,10 @@ def _consume_terminal_condition_reopens() -> None: return while wait_yield_count < len(wait_opened): opened = wait_opened[wait_yield_count] - if not _same_logical_condition_wait(opened, terminal_condition_reopen_cmd): + if not _same_logical_condition_wait( + opened, terminal_condition_reopen_cmd, chain_first=terminal_condition_chain_first, + chain_current=terminal_condition_chain_current, next_index=wait_yield_count, + ): break opened_id = opened.get("condition_wait_id") if not isinstance(opened_id, str): @@ -6928,9 +6950,14 @@ def _assert_cancellation_call_matches(command: Any, boundary: CancellationDelive expected = ({"delay_seconds": leaf.delay_seconds} if isinstance(leaf, StartTimer) else { "condition_key": leaf.condition_key, "timeout_seconds": leaf.timeout_seconds, - "condition_definition_fingerprint": leaf.condition_definition_fingerprint, }) - if any(field in payload and payload[field] != actual for field, actual in expected.items()): + fingerprint_changed = isinstance(leaf, WaitCondition) and fingerprint_verdict( + payload.get("condition_definition_fingerprint"), leaf.condition_definition_fingerprint, + leaf.predicate, + ) == "mismatch" + if fingerprint_changed or any( + field in payload and payload[field] != actual for field, actual in expected.items() + ): raise NonDeterministicReplayError( sequence, "original group member", [event_type], detail="scoped group changed its original timer or condition definition", @@ -7054,7 +7081,7 @@ def _opened_before_terminal(next_index: int, opened: Mapping[str, Any] | None) - return terminal is not None and wait_opened_event_indexes[next_index] < terminal def _reopened_with_remaining_timeout(command: WaitCondition, first_index: int) -> WaitCondition: - """Reopen a logical wait with the time left before its first deadline. + """Reopen a logical wait with its first fingerprint and remaining time. The deadline is fixed when the wait first opens, as in embedded Workflow: the first ConditionWaitOpened time plus the authored @@ -7063,14 +7090,20 @@ def _reopened_with_remaining_timeout(command: WaitCondition, first_index: int) - history time, rounded up and clamped to 0..authored (the sdk-php#269 rule). Without recorded times the authored timeout is kept. """ - authored = command.timeout_seconds - if authored is None or first_index >= len(wait_opened): + if first_index >= len(wait_opened): return command + reopened = copy(command) + # A reopen sends the fingerprint its first open recorded, so a wait + # first opened under v1 (2.14.0 and earlier) keeps v1 across the + # upgrade and links v1 to v1, never v1 to v2 (#258). + recorded = wait_opened[first_index].get("condition_definition_fingerprint") + if isinstance(recorded, str) and recorded: + reopened.condition_definition_fingerprint = recorded + authored = command.timeout_seconds opened_at = wait_opened_times[first_index] - if opened_at is None or history_time is None: - return command + if authored is None or opened_at is None or history_time is None: + return reopened remaining = math.ceil(authored - (history_time - opened_at).total_seconds()) - reopened = copy(command) reopened.timeout_seconds = min(authored, max(0, remaining)) return reopened @@ -7730,7 +7763,10 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None has_reopened_same_wait = ( opened is not None and next_wait_index < len(wait_opened) - and _same_logical_condition_wait(wait_opened[next_wait_index], cmd) + and _same_logical_condition_wait( + wait_opened[next_wait_index], cmd, chain_first=wait_opened[first_open_index], + chain_current=opened, next_index=next_wait_index, + ) ) if resolution == "timed_out": if has_reopened_same_wait and _opened_before_terminal(next_wait_index, opened): @@ -7769,6 +7805,8 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None if has_reopened_same_wait and consumed_reopen_for_current_wait else None ) + terminal_condition_chain_first = wait_opened[first_open_index] + terminal_condition_chain_current = opened next_value = True wait_yield_count += 1 break @@ -7778,6 +7816,8 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None if has_reopened_same_wait and consumed_reopen_for_current_wait else None ) + terminal_condition_chain_first = wait_opened[first_open_index] + terminal_condition_chain_current = opened next_value = True wait_yield_count += 1 break diff --git a/tests/condition_fingerprint_predicates.py b/tests/condition_fingerprint_predicates.py new file mode 100644 index 0000000..5e38134 --- /dev/null +++ b/tests/condition_fingerprint_predicates.py @@ -0,0 +1,129 @@ +"""Predicates whose v2 fingerprints are pinned in condition_fingerprints_v2.json (#258). + +The golden values were computed on the official python:3.12-slim image. Every +CI Python version must compute the same values, in this process and in a fresh +one, so a wait recorded on one version or process replays on any other. +Keep this module's source unchanged: editing it changes the fingerprints. +""" + +from __future__ import annotations + +from collections.abc import Callable +from functools import wraps +from typing import Any + +from durable_workflow import workflow + + +def _identity(predicate: Callable[[], bool]) -> Callable[[], bool]: + return predicate + + +def _typed_method_fixture(fn: Callable[..., Any]) -> Callable[..., Any]: + """Exercise the unwrapping marker without the unreleased typed workflow API.""" + @wraps(fn) + def wrapped(*args: Any, **kwargs: Any) -> Any: + return fn(*args, **kwargs) + + wrapped.__typed_workflow_method__ = fn # type: ignore[attr-defined] + return wrapped + + +@workflow.defn(name="tests.fingerprint.typed-ready") +class TypedReady: + def __init__(self) -> None: + self.approvals: list[str] = [] + + @workflow.query("ready") + @_typed_method_fixture + def ready(self, minimum: int = 2) -> bool: + return len(self.approvals) >= minimum + + def run(self, ctx: workflow.WorkflowContext) -> None: + return None + + +class Approvals: + def __init__(self) -> None: + self.items: list[int] = [1, 20] + self.approved = False + self.name = "x" + + def is_done(self) -> bool: + return self.approved + + def has_large(self) -> bool: + return any(n > 10 for n in self.items) + + def _multiline(self) -> Callable[[], bool]: + return _identity( + lambda: ( + self.approved + and len(self.items) > 0 + ) + ) + + def _same_line(self) -> tuple[Callable[[], bool], Callable[[], bool]]: + return (lambda: self.approved), (lambda: not self.approved) + + def _inner(self) -> Callable[[], bool]: + def ready() -> bool: + return self.approved and bool(self.items) + + return ready + + # Python 3.12 inlines list, dict and set comprehensions (PEP 709), so the + # comprehension disappears from these lambdas' __qualname__. + def _in_listcomp(self) -> Callable[[], bool]: + return [lambda: self.approved for _ in range(1)][0] + + def _in_dictcomp(self) -> Callable[[], bool]: + return {key: lambda: bool(self.items) for key in ["only"]}["only"] + + def _in_setcomp(self) -> Callable[[], bool]: + return next(iter({lambda: len(self.items) > 1 for _ in range(1)})) + + def _same_line_in_listcomp(self) -> list[Callable[[], bool]]: + return [(lambda: value > 0) if value else (lambda: not value) for value in [1, 0]] # noqa: B023 + + def _same_line_listcomp_bodies(self) -> tuple[Callable[[], bool], Callable[[], bool]]: + return (lambda: any([n > 0 for n in self.items])), (lambda: any([n < 0 for n in self.items])) + + def _same_line_dictcomp_bodies(self) -> tuple[Callable[[], bool], Callable[[], bool]]: + items = self.items + return (lambda: any({n: n > 0 for n in items}.values())), (lambda: any({n: n < 0 for n in items}.values())) + + def _same_line_setcomp_bodies(self) -> tuple[Callable[[], bool], Callable[[], bool]]: + return (lambda: any({n > 0 for n in self.items})), (lambda: any({n < 0 for n in self.items})) + + def predicates(self) -> dict[str, Any]: + first, second = self._same_line() + positive, negative = self._same_line_in_listcomp() + list_first, list_second = self._same_line_listcomp_bodies() + dict_first, dict_second = self._same_line_dictcomp_bodies() + set_first, set_second = self._same_line_setcomp_bodies() + return { + "lambda": lambda: self.approved, + "genexpr": lambda: any(n > 10 for n in self.items), + "listcomp": lambda: len([n for n in self.items if n]) > 2, + "nested_lambda": lambda: (lambda value: value > 3)(len(self.items)), + "fstring": lambda: f"{self.name}-{len(self.items)}" == "x-1", + "multiline_in_call": self._multiline(), + "same_line_first": first, + "same_line_second": second, + "bound_method": self.is_done, + "bound_genexpr_method": self.has_large, + "inner_def": self._inner(), + "typed_wrapper": TypedReady().ready, + "lambda_in_listcomp": self._in_listcomp(), + "lambda_in_dictcomp": self._in_dictcomp(), + "lambda_in_setcomp": self._in_setcomp(), + "same_line_in_listcomp_first": positive, + "same_line_in_listcomp_second": negative, + "same_line_listcomp_body_first": list_first, + "same_line_listcomp_body_second": list_second, + "same_line_dictcomp_body_first": dict_first, + "same_line_dictcomp_body_second": dict_second, + "same_line_setcomp_body_first": set_first, + "same_line_setcomp_body_second": set_second, + } diff --git a/tests/condition_fingerprint_workflows.py b/tests/condition_fingerprint_workflows.py new file mode 100644 index 0000000..4ce1a1e --- /dev/null +++ b/tests/condition_fingerprint_workflows.py @@ -0,0 +1,50 @@ +"""Workflows for the condition-fingerprint corpus fixtures (#258). + +The v1 fixtures were recorded by durable-workflow 2.14.0 from PyPI on the +official python:3.12-slim image, replaying these classes; the v2 fixture uses +this branch's stable fingerprint. Keep this module's source unchanged: the +recorded fingerprints depend on it. +""" + +from __future__ import annotations + +from durable_workflow import workflow + + +@workflow.defn(name="tests.replay.fingerprint-approval") +class FingerprintApproval: + """A keyless wait whose fingerprint alone identifies it.""" + + def __init__(self) -> None: + self.approved = False + + @workflow.signal("approve") + def approve(self) -> None: + self.approved = True + + def run(self, ctx: workflow.WorkflowContext): # type: ignore[no-untyped-def] + yield ctx.wait_condition(lambda: self.approved) + return "approved" + + +@workflow.defn(name="tests.replay.fingerprint-genexpr") +class FingerprintGenexpr: + """A keyed wait whose v1 fingerprint differs in every process (nested genexpr).""" + + def __init__(self) -> None: + self.items: list[int] = [] + + @workflow.signal("add") + def add(self, value: int) -> None: + self.items.append(value) + + def run(self, ctx: workflow.WorkflowContext): # type: ignore[no-untyped-def] + yield ctx.wait_condition(lambda: any(n > 10 for n in self.items), key="large") + return list(self.items) + + +@workflow.defn(name="tests.replay.fingerprint-terminal-duplicate") +class FingerprintTerminalDuplicate(FingerprintApproval): + @workflow.signal("note") + def note(self) -> None: + return None diff --git a/tests/fixtures/condition_fingerprints_v2.json b/tests/fixtures/condition_fingerprints_v2.json new file mode 100644 index 0000000..ea1c5b6 --- /dev/null +++ b/tests/fixtures/condition_fingerprints_v2.json @@ -0,0 +1,29 @@ +{ + "recorded_on": "python:3.12-slim (CPython 3.12.16)", + "module": "tests.condition_fingerprint_predicates", + "fingerprints": { + "lambda": "dwpy2:sha256:838e5598f19815ba337ba3b3483ba1036f162e1a39c115c1cb1e3259cc4bd0ca", + "genexpr": "dwpy2:sha256:7d23e5fca8a875c14dd2aa01263f8be68b681686cfaf890ff4b494893d2d09c8", + "listcomp": "dwpy2:sha256:905370895eeccf09cc08ca8d5611bfad01b227eaf8bf90402098d093681971a3", + "nested_lambda": "dwpy2:sha256:a75240a391df56bc1a043357649e88182153055bf5b7ac2bb413abbff478ad9c", + "fstring": "dwpy2:sha256:e271b27eb6e6c71e8c2ae40088025003907b231c006da424731d92b40c780635", + "multiline_in_call": "dwpy2:sha256:53a9527b0234805ebffe2dcdd79715fd7743c23d001500b99f4616458e630cb5", + "same_line_first": "dwpy2:sha256:bd14b20c03220f1cddd0c6e07ea364aef66051dbdf7d14a73b2e03724d06c471", + "same_line_second": "dwpy2:sha256:351a2607e1e06e4d0bfa6020f47d577661c259fe1f7f5d43ccbe1e00e59140ad", + "bound_method": "dwpy2:sha256:4b9aabe93ef8c48c2f05481ccb92f743da7c19b76104816c4a3f12ab32d741b9", + "bound_genexpr_method": "dwpy2:sha256:9bbb6600801559c28420aa32085bfdc76ceef33e03c2d9194d2edb507053c934", + "inner_def": "dwpy2:sha256:7be3326bb020ab50b9d1bad496b395551db90324bf6b0b0af9fe888e9dcfdd84", + "typed_wrapper": "dwpy2:sha256:f5f8904a70119a2f2999a4a256bc42549f48aca0c74c47ab42d256db852b919a", + "lambda_in_listcomp": "dwpy2:sha256:f8f58909db214449f4f0e7fbe94937b06bf98c83bb0c995563352f1d760fdb80", + "lambda_in_dictcomp": "dwpy2:sha256:a9ff285d478d5859d137782384a0df64f3f7236a8601ee25278b5e750def4e3d", + "lambda_in_setcomp": "dwpy2:sha256:4d998d02bda534ead4a89dacb82846aaf72c96624aad0c68516c11653f3fe55f", + "same_line_in_listcomp_first": "dwpy2:sha256:3b5d4ea892eb8458abd074e793d6f6fc35118bf1032530a8d16773d68bdcb96d", + "same_line_in_listcomp_second": "dwpy2:sha256:7bbdd2c46be01c6ce8ea970a2fa68f8cf17b8643a8d58b752e8cc6c21e27a641", + "same_line_listcomp_body_first": "dwpy2:sha256:00ce99ef06cf9b60f5831755f511382ea0fc968bf840a5ce8975363612f8e877", + "same_line_listcomp_body_second": "dwpy2:sha256:00ce99ef06cf9b60f5831755f511382ea0fc968bf840a5ce8975363612f8e877", + "same_line_dictcomp_body_first": "dwpy2:sha256:940867da138c61a2bfcf8c892a11cc286f6575519260e6e453d6e1bff57c1208", + "same_line_dictcomp_body_second": "dwpy2:sha256:940867da138c61a2bfcf8c892a11cc286f6575519260e6e453d6e1bff57c1208", + "same_line_setcomp_body_first": "dwpy2:sha256:f7440ac9c37954c7265ed231940426db90d507250819b5f8cb9fb3bd2ceb4e7f", + "same_line_setcomp_body_second": "dwpy2:sha256:f7440ac9c37954c7265ed231940426db90d507250819b5f8cb9fb3bd2ceb4e7f" + } +} diff --git a/tests/fixtures/replay_regressions/condition-fingerprint-v1-genexpr-cross-process.json b/tests/fixtures/replay_regressions/condition-fingerprint-v1-genexpr-cross-process.json new file mode 100644 index 0000000..44a24d3 --- /dev/null +++ b/tests/fixtures/replay_regressions/condition-fingerprint-v1-genexpr-cross-process.json @@ -0,0 +1,106 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "condition-fingerprint-v1-genexpr-cross-process", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "source": { + "runtime": "sdk", + "package": "durable-workflow", + "version": "2.14.0" + }, + "workflow": { + "type": "tests.replay.fingerprint-genexpr", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-04T00:00:00+00:00", + "payload": { + "workflow_type": "tests.replay.fingerprint-genexpr" + }, + "sequence": 1 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T00:00:01+00:00", + "payload": { + "condition_wait_id": "large-1", + "sequence": 1, + "condition_definition_fingerprint": "sha256:dd6cc254ae46267f90fde3ff42b381ae7f6022af4de6fa42a07f1abbe2434b7f", + "condition_key": "large" + }, + "sequence": 2 + }, + { + "event_type": "SignalReceived", + "recorded_at": "2026-11-04T00:00:02+00:00", + "payload": { + "signal_name": "add", + "value": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwCBAoA" + }, + "payload_codec": "avro", + "workflow_sequence": 1 + }, + "sequence": 3 + }, + { + "event_type": "ConditionWaitSatisfied", + "recorded_at": "2026-11-04T00:00:02+00:00", + "payload": { + "condition_wait_id": "large-1", + "sequence": 1, + "condition_key": "large" + }, + "sequence": 4 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T00:00:03+00:00", + "payload": { + "condition_wait_id": "large-2", + "sequence": 2, + "condition_definition_fingerprint": "sha256:b415051d8936710c8a887c6fc30e3bcb8bd20f2b9c79f398dfffa84215050aa3", + "condition_key": "large" + }, + "sequence": 5 + }, + { + "event_type": "SignalReceived", + "recorded_at": "2026-11-04T00:00:04+00:00", + "payload": { + "signal_name": "add", + "value": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwCBCgA" + }, + "payload_codec": "avro", + "workflow_sequence": 2 + }, + "sequence": 6 + }, + { + "event_type": "ConditionWaitSatisfied", + "recorded_at": "2026-11-04T00:00:04+00:00", + "payload": { + "condition_wait_id": "large-2", + "sequence": 2, + "condition_key": "large" + }, + "sequence": 7 + } + ], + "expected": { + "type": "complete_workflow", + "result": [ + 5, + 20 + ] + } +} diff --git a/tests/fixtures/replay_regressions/condition-fingerprint-v1-py312-approval.json b/tests/fixtures/replay_regressions/condition-fingerprint-v1-py312-approval.json new file mode 100644 index 0000000..1aa3f11 --- /dev/null +++ b/tests/fixtures/replay_regressions/condition-fingerprint-v1-py312-approval.json @@ -0,0 +1,66 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "condition-fingerprint-v1-py312-approval", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "source": { + "runtime": "sdk", + "package": "durable-workflow", + "version": "2.14.0" + }, + "workflow": { + "type": "tests.replay.fingerprint-approval", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-04T00:00:00+00:00", + "payload": { + "workflow_type": "tests.replay.fingerprint-approval" + }, + "sequence": 1 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T00:00:01+00:00", + "payload": { + "condition_wait_id": "approval-1", + "sequence": 1, + "condition_definition_fingerprint": "sha256:0f5fd86d297870b40434e114ff4b7b42ce2c0d2df725c1069e8931539091c9d8" + }, + "sequence": 2 + }, + { + "event_type": "SignalReceived", + "recorded_at": "2026-11-04T00:00:02+00:00", + "payload": { + "signal_name": "approve", + "value": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwA" + }, + "payload_codec": "avro", + "workflow_sequence": 1 + }, + "sequence": 3 + }, + { + "event_type": "ConditionWaitSatisfied", + "recorded_at": "2026-11-04T00:00:02+00:00", + "payload": { + "condition_wait_id": "approval-1", + "sequence": 1 + }, + "sequence": 4 + } + ], + "expected": { + "type": "complete_workflow", + "result": "approved" + } +} diff --git a/tests/fixtures/replay_regressions/condition-fingerprint-v1-terminal-cross-process-duplicate.json b/tests/fixtures/replay_regressions/condition-fingerprint-v1-terminal-cross-process-duplicate.json new file mode 100644 index 0000000..2ef3458 --- /dev/null +++ b/tests/fixtures/replay_regressions/condition-fingerprint-v1-terminal-cross-process-duplicate.json @@ -0,0 +1,104 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "condition-fingerprint-v1-terminal-cross-process-duplicate", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.fingerprint-terminal-duplicate", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-04T00:00:00+00:00", + "payload": { + "workflow_type": "tests.replay.fingerprint-terminal-duplicate" + }, + "sequence": 1 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T00:00:01+00:00", + "payload": { + "condition_wait_id": "w1", + "sequence": 1, + "condition_definition_fingerprint": "sha256:1111111111111111111111111111111111111111111111111111111111111111" + }, + "sequence": 2 + }, + { + "event_type": "SignalReceived", + "recorded_at": "2026-11-04T00:00:02+00:00", + "payload": { + "signal_name": "note", + "value": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwA" + }, + "payload_codec": "avro", + "workflow_sequence": 1 + }, + "sequence": 3 + }, + { + "event_type": "ConditionWaitSatisfied", + "recorded_at": "2026-11-04T00:00:02+00:00", + "payload": { + "condition_wait_id": "w1", + "sequence": 1 + }, + "sequence": 4 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T00:00:03+00:00", + "payload": { + "condition_wait_id": "w2", + "sequence": 2, + "condition_definition_fingerprint": "sha256:1111111111111111111111111111111111111111111111111111111111111111" + }, + "sequence": 5 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T00:00:03+00:00", + "payload": { + "condition_wait_id": "w3", + "sequence": 3, + "condition_definition_fingerprint": "sha256:2222222222222222222222222222222222222222222222222222222222222222" + }, + "sequence": 6 + }, + { + "event_type": "SignalReceived", + "recorded_at": "2026-11-04T00:00:04+00:00", + "payload": { + "signal_name": "approve", + "value": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwA" + }, + "payload_codec": "avro", + "workflow_sequence": 2 + }, + "sequence": 7 + }, + { + "event_type": "ConditionWaitSatisfied", + "recorded_at": "2026-11-04T00:00:04+00:00", + "payload": { + "condition_wait_id": "w2", + "sequence": 2 + }, + "sequence": 8 + } + ], + "expected": { + "type": "complete_workflow", + "result": "approved" + } +} diff --git a/tests/fixtures/replay_regressions/condition-fingerprint-v2-approval.json b/tests/fixtures/replay_regressions/condition-fingerprint-v2-approval.json new file mode 100644 index 0000000..cc85aba --- /dev/null +++ b/tests/fixtures/replay_regressions/condition-fingerprint-v2-approval.json @@ -0,0 +1,61 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "condition-fingerprint-v2-approval", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.fingerprint-approval", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-04T00:00:00+00:00", + "payload": { + "workflow_type": "tests.replay.fingerprint-approval" + }, + "sequence": 1 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T00:00:01+00:00", + "payload": { + "condition_wait_id": "approval-1", + "sequence": 1, + "condition_definition_fingerprint": "dwpy2:sha256:21b0cde657ecccaf9e2dc2a362d2e32e0e4d5d822c112576e5e23f4485a03199" + }, + "sequence": 2 + }, + { + "event_type": "SignalReceived", + "recorded_at": "2026-11-04T00:00:02+00:00", + "payload": { + "signal_name": "approve", + "value": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwA" + }, + "payload_codec": "avro", + "workflow_sequence": 1 + }, + "sequence": 3 + }, + { + "event_type": "ConditionWaitSatisfied", + "recorded_at": "2026-11-04T00:00:02+00:00", + "payload": { + "condition_wait_id": "approval-1", + "sequence": 1 + }, + "sequence": 4 + } + ], + "expected": { + "type": "complete_workflow", + "result": "approved" + } +} diff --git a/tests/test_client_timeouts.py b/tests/test_client_timeouts.py index 9d808ac..33a5b2b 100644 --- a/tests/test_client_timeouts.py +++ b/tests/test_client_timeouts.py @@ -60,10 +60,15 @@ async def respond(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> @pytest.mark.asyncio @pytest.mark.parametrize("operation", ["control", "discovery", "worker"]) async def test_configured_timeout_bounds_a_stalled_response(operation: str) -> None: + async def allow_connection_setup(request: httpx.Request) -> None: + # Exercise the configured read budget even when cold TCP setup is slow. + request.extensions["timeout"]["connect"] = 1.0 + async with ( response_server(delay=None) as endpoint, Client(endpoint, timeout=0.05, retry_policy=TransportRetryPolicy(max_attempts=1)) as client, ): + client._http.event_hooks["request"].append(allow_connection_setup) if operation == "control": request = client.health() elif operation == "discovery": diff --git a/tests/test_condition_fingerprint.py b/tests/test_condition_fingerprint.py new file mode 100644 index 0000000..06b2245 --- /dev/null +++ b/tests/test_condition_fingerprint.py @@ -0,0 +1,282 @@ +"""Stable v2 condition fingerprints and v1 compatibility (#258).""" + +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path +from typing import Any + +import pytest + +from durable_workflow import serializer, workflow +from durable_workflow._condition_fingerprint import ( + V2_PREFIX, + fingerprint, + fingerprint_verdict, + is_legacy, + legacy_fingerprint, +) +from durable_workflow.errors import NonDeterministicReplayError +from durable_workflow.workflow import CompleteWorkflow, WaitCondition, WorkflowContext, replay +from tests.condition_fingerprint_predicates import Approvals + +ROOT = Path(__file__).resolve().parent.parent +GOLDEN = json.loads((ROOT / "tests/fixtures/condition_fingerprints_v2.json").read_text(encoding="utf-8")) +FOREIGN_V1 = "sha256:" + "1" * 64 # a v1 value recorded by another Python version or process +OTHER_V1 = "sha256:" + "2" * 64 + + +def test_v2_fingerprints_match_the_values_recorded_on_python_3_12() -> None: + # Every CI Python version (3.10 to 3.15) must reproduce the 3.12 values. + computed = {name: fingerprint(predicate) for name, predicate in Approvals().predicates().items()} + assert computed == GOLDEN["fingerprints"] + assert all(value.startswith(V2_PREFIX) for value in computed.values()) + + +def test_v2_fingerprints_are_the_same_in_fresh_processes() -> None: + # v1 changed in every process for nested functions (genexpr, nested lambda). + script = ( + "import json, sys; sys.path.insert(0, sys.argv[1]);" + "from tests.condition_fingerprint_predicates import Approvals;" + "from durable_workflow._condition_fingerprint import fingerprint;" + "print(json.dumps({k: fingerprint(v) for k, v in Approvals().predicates().items()}))" + ) + for _ in range(2): + output = subprocess.run([sys.executable, "-c", script, str(ROOT)], check=True, capture_output=True, + text=True, cwd=ROOT).stdout + assert json.loads(output) == GOLDEN["fingerprints"] + + +def test_v2_ignores_formatting_but_not_logic() -> None: + reformatted = lambda: ( # noqa: E731 + Approvals().approved # a comment + ) + flipped = lambda: not Approvals().approved # noqa: E731 + original = lambda: Approvals().approved # noqa: E731 + # Same module and qualname: only the predicate's logic decides. + assert fingerprint(reformatted) == fingerprint(original) + assert fingerprint(original) != fingerprint(flipped) + + +def test_a_predicate_without_source_has_no_fingerprint() -> None: + predicate = eval("lambda: False") # noqa: S307 - a predicate with no source file + assert fingerprint(predicate) is None + assert WorkflowContext(run_id="x").wait_condition(predicate, key="k").condition_definition_fingerprint is None + + +@pytest.mark.parametrize(("recorded", "authored", "expected"), [ + (None, "dwpy2:sha256:" + "a" * 64, "match"), + ("dwpy2:sha256:" + "a" * 64, "dwpy2:sha256:" + "a" * 64, "match"), + ("dwpy2:sha256:" + "a" * 64, "dwpy2:sha256:" + "b" * 64, "mismatch"), + ("dwpy2:sha256:" + "a" * 64, None, "unverifiable"), + (FOREIGN_V1, "dwpy2:sha256:" + "a" * 64, "unverifiable"), + (FOREIGN_V1, None, "unverifiable"), + ("sha256:different", "dwpy2:sha256:" + "a" * 64, "mismatch"), + (FOREIGN_V1, "changed", "mismatch"), # an explicitly supplied non-v2 identity is compared exactly + (FOREIGN_V1, FOREIGN_V1, "match"), +]) +def test_fingerprint_verdicts(recorded: Any, authored: str | None, expected: str) -> None: + assert fingerprint_verdict(recorded, authored, lambda: False) == expected + + +def test_a_v1_value_from_this_process_still_matches() -> None: + def predicate() -> bool: + return False + + assert is_legacy(legacy_fingerprint(predicate)) + assert fingerprint_verdict(legacy_fingerprint(predicate), fingerprint(predicate), predicate) == "match" + + +# Mixed chains across the upgrade (General's correction 1). +@workflow.defn(name="tests.fingerprint.keyless-approval") +class KeylessApproval: + def __init__(self) -> None: + self.approved = False + + @workflow.signal("approve") + def approve(self) -> None: + self.approved = True + + @workflow.signal("note") + def note(self) -> None: + return None + + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + yield ctx.wait_condition(lambda: self.approved) + return "approved" + + +AT = "2026-11-04T00:00:0{}+00:00" + + +def _opened(wait_id: str, sequence: int, value: str | None, second: int) -> dict[str, Any]: + payload: dict[str, Any] = {"condition_wait_id": wait_id, "sequence": sequence} + if value is not None: + payload["condition_definition_fingerprint"] = value + return {"event_type": "ConditionWaitOpened", "recorded_at": AT.format(second), "payload": payload} + + +def _signal(name: str, sequence: int, second: int) -> dict[str, Any]: + return {"event_type": "SignalReceived", "recorded_at": AT.format(second), "payload": { + "signal_name": name, "value": serializer.envelope([]), "payload_codec": "avro", + "workflow_sequence": sequence}} + + +def _satisfied(wait_id: str, sequence: int, second: int) -> dict[str, Any]: + return {"event_type": "ConditionWaitSatisfied", "recorded_at": AT.format(second), + "payload": {"condition_wait_id": wait_id, "sequence": sequence}} + + +START = {"event_type": "WorkflowStarted", "recorded_at": AT.format(0), "payload": {}} + + +def test_a_reopen_of_a_v1_first_open_sends_the_recorded_v1_value() -> None: + # Opened under 2.14.0 (v1), then a false wake under 2.14.1. The Server + # records ConditionWaitSatisfied with this task's reopen, so the task sees + # only the open and the signal. + history = [START, _opened("w1", 1, FOREIGN_V1, 1), _signal("note", 1, 2)] + (command,) = replay(KeylessApproval, history, []).commands + assert isinstance(command, WaitCondition) + assert command.condition_definition_fingerprint == FOREIGN_V1 + + +def test_a_v1_chain_links_v1_to_v1_after_the_upgrade() -> None: + history = [START, _opened("w1", 1, FOREIGN_V1, 1), _signal("note", 1, 2), _satisfied("w1", 1, 2), + _opened("w2", 2, FOREIGN_V1, 3), _signal("approve", 2, 4), _satisfied("w2", 2, 4)] + assert list(replay(KeylessApproval, history, []).commands) == [CompleteWorkflow("approved")] + + +def test_a_reopen_without_source_keeps_echoing_the_recorded_value() -> None: + @workflow.defn(name="tests.fingerprint.no-source") + class NoSource: + def __init__(self) -> None: + self.approved = False + + @workflow.signal("note") + def note(self) -> None: + return None + + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + yield ctx.wait_condition(eval("lambda self=self: self.approved")) # noqa: S307 + return "approved" + + history = [START, _opened("w1", 1, FOREIGN_V1, 1), _signal("note", 1, 2)] + (command,) = replay(NoSource, history, []).commands + assert isinstance(command, WaitCondition) and command.condition_definition_fingerprint == FOREIGN_V1 + + +# Keyless unverifiable values never bind a wait to another call (correction 2). +@workflow.defn(name="tests.fingerprint.two-keyless-waits") +class TwoKeylessWaits: + def __init__(self) -> None: + self.approvals = 0 + + @workflow.signal("approve") + def approve(self) -> None: + self.approvals += 1 + + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + yield ctx.wait_condition(lambda: self.approvals >= 1) + yield ctx.wait_condition(lambda: self.approvals >= 2) + return self.approvals + + +def test_a_later_keyless_v1_open_binds_to_the_next_call_not_as_a_reopen() -> None: + history = [START, _opened("w1", 1, FOREIGN_V1, 1), _signal("approve", 1, 2), _satisfied("w1", 1, 2), + _opened("w2", 2, OTHER_V1, 3), _signal("approve", 2, 4), _satisfied("w2", 2, 4)] + assert list(replay(TwoKeylessWaits, history, []).commands) == [CompleteWorkflow(2)] + + +def test_a_changed_v2_predicate_is_refused() -> None: + (command,) = replay(KeylessApproval, [START], []).commands + assert isinstance(command, WaitCondition) and command.condition_definition_fingerprint.startswith(V2_PREFIX) + changed = "dwpy2:sha256:" + "0" * 64 + history = [START, _opened("w1", 1, changed, 1), _signal("approve", 1, 2), _satisfied("w1", 1, 2)] + with pytest.raises(NonDeterministicReplayError, match="condition_fingerprint_changed"): + replay(KeylessApproval, history, []) + + +def test_two_lambdas_on_one_line_get_their_own_fingerprints() -> None: + # A conditional choice between predicates on one line (as in the + # condition-recovery integration test) must still detect drift. Each + # candidate is matched by compiling it on this interpreter; only its AST + # is hashed. + predicates = Approvals().predicates() + assert GOLDEN["fingerprints"]["same_line_first"] != GOLDEN["fingerprints"]["same_line_second"] + assert fingerprint(predicates["same_line_first"]) == GOLDEN["fingerprints"]["same_line_first"] + assert fingerprint(predicates["same_line_second"]) == GOLDEN["fingerprints"]["same_line_second"] + + +def test_inlined_comprehension_segments_are_dropped_from_the_qualname() -> None: + # 3.11 names a lambda in a list comprehension f...; + # 3.12 inlines the comprehension (PEP 709) and names it f... + predicates = Approvals().predicates() + for name in ("lambda_in_listcomp", "lambda_in_dictcomp", "lambda_in_setcomp"): + assert fingerprint(predicates[name]) == GOLDEN["fingerprints"][name] + + +@pytest.mark.parametrize("kind", ["listcomp", "dictcomp", "setcomp"]) +def test_same_line_comprehension_bodies_share_a_portable_combined_fingerprint(kind: str) -> None: + predicates = Approvals().predicates() + first = fingerprint(predicates[f"same_line_{kind}_body_first"]) + second = fingerprint(predicates[f"same_line_{kind}_body_second"]) + assert first == second == GOLDEN["fingerprints"][f"same_line_{kind}_body_first"] + + +# A keyless wait without source (eval, .pyc-only) still links its false-wake +# reopen within its chain, as 2.14.0 did with its process-local v1. +@workflow.defn(name="tests.fingerprint.keyless-no-source") +class KeylessNoSource: + def __init__(self) -> None: + self.approved = False + + @workflow.signal("approve") + def approve(self) -> None: + self.approved = True + + @workflow.signal("note") + def note(self) -> None: + return None + + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + yield ctx.wait_condition(eval("lambda self=self: self.approved")) # noqa: S307 + return "approved" + + +def test_a_keyless_wait_without_source_links_its_false_wake_reopen() -> None: + history = [START, _opened("w1", 1, None, 1), _signal("note", 1, 2), _satisfied("w1", 1, 2), + _opened("w2", 2, None, 3), _signal("approve", 2, 4), _satisfied("w2", 2, 4)] + assert list(replay(KeylessNoSource, history, []).commands) == [CompleteWorkflow("approved")] + + +def test_a_keyless_wait_without_source_reopens_without_a_fingerprint() -> None: + (command,) = replay(KeylessNoSource, [START, _opened("w1", 1, None, 1), _signal("note", 1, 2)], []).commands + assert isinstance(command, WaitCondition) and command.condition_definition_fingerprint is None + + +def test_terminal_cleanup_links_a_trailing_v1_duplicate_within_its_chain() -> None: + # A v1 keyless chain (opened before the upgrade, reopened with the same v1), + # then a trailing duplicate open recorded by a second queued task. + history = [START, _opened("w1", 1, FOREIGN_V1, 1), _signal("note", 1, 2), _satisfied("w1", 1, 2), + _opened("w2", 2, FOREIGN_V1, 3), _signal("approve", 2, 4), _satisfied("w2", 2, 4), + _opened("w3", 3, FOREIGN_V1, 5)] + assert list(replay(KeylessApproval, history, []).commands) == [CompleteWorkflow("approved")] + + +def test_terminal_cleanup_links_a_cross_process_duplicate_opened_before_terminal() -> None: + history = [START, _opened("w1", 1, FOREIGN_V1, 1), _signal("note", 1, 2), _satisfied("w1", 1, 2), + _opened("w2", 2, FOREIGN_V1, 3), _opened("w3", 3, OTHER_V1, 3), + _signal("approve", 2, 4), _satisfied("w2", 2, 4)] + assert list(replay(KeylessApproval, history, []).commands) == [CompleteWorkflow("approved")] + + +def test_terminal_cleanup_refuses_an_unrelated_later_keyless_v1_open() -> None: + history = [START, _opened("w1", 1, FOREIGN_V1, 1), _signal("note", 1, 2), _satisfied("w1", 1, 2), + _opened("w2", 2, FOREIGN_V1, 3), _signal("approve", 2, 4), _satisfied("w2", 2, 4), + _opened("w3", 3, OTHER_V1, 5)] + with pytest.raises(NonDeterministicReplayError) as captured: + replay(KeylessApproval, history, []) + assert captured.value.workflow_sequence == 3 + assert captured.value.expected_shape == "complete workflow" diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index 95e3059..99c790d 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -25,6 +25,7 @@ commands_to_server_commands, query_state, ) +from tests.condition_fingerprint_workflows import FingerprintApproval, FingerprintGenexpr, FingerprintTerminalDuplicate from tests.deregistration_recovery import RecoveredWorkflow as FencedRecoveryWorkflow from tests.test_condition_replay_drift import ConditionIdentityWorkflow from tests.test_generator_lifecycle import CleanupSideEffect, PendingFinally @@ -422,6 +423,9 @@ def run(self, ctx: WorkflowContext, message: str) -> None: RemainingTimeoutApproval, OrphanedDuplicateDecision, TypedActivityResultsWorkflow, + FingerprintApproval, + FingerprintGenexpr, + FingerprintTerminalDuplicate, ] WORKFLOW_TYPES = {str(getattr(workflow, "__workflow_name__", workflow.__name__)): workflow for workflow in WORKFLOWS} WORKFLOW_TYPES["tests.replay.condition-identity"] = ConditionIdentityWorkflow diff --git a/tests/test_wait_condition.py b/tests/test_wait_condition.py index 8aa75e5..00391aa 100644 --- a/tests/test_wait_condition.py +++ b/tests/test_wait_condition.py @@ -250,7 +250,8 @@ def test_wait_condition_returns_dataclass_with_predicate_and_key(self) -> None: assert cmd.condition_key == "k" assert cmd.timeout_seconds == 5 assert cmd.condition_definition_fingerprint is not None - assert cmd.condition_definition_fingerprint.startswith("sha256:") + # v2: stable across Python versions and processes (#258). + assert cmd.condition_definition_fingerprint.startswith("dwpy2:sha256:") assert callable(cmd.predicate) assert cmd.predicate() is False diff --git a/tests/test_worker.py b/tests/test_worker.py index 141df05..b795a3d 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -4911,6 +4911,13 @@ class TestWorkerHeartbeats: async def test_run_drives_periodic_heartbeats_with_slot_state( self, mock_client: AsyncMock ) -> None: + heartbeat_sent = asyncio.Event() + + async def heartbeat(**kwargs: object) -> dict[str, bool]: + heartbeat_sent.set() + return {"acknowledged": True} + + mock_client.heartbeat_worker = AsyncMock(side_effect=heartbeat) worker = Worker( mock_client, task_queue="q1", @@ -4922,11 +4929,13 @@ async def test_run_drives_periodic_heartbeats_with_slot_state( heartbeat_interval=0.05, ) run_task = asyncio.create_task(worker.run()) - await asyncio.sleep(0.2) - await worker.stop() - run_task.cancel() - with contextlib.suppress(asyncio.CancelledError): - await run_task + try: + await asyncio.wait_for(heartbeat_sent.wait(), timeout=2) + finally: + await worker.stop() + run_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await run_task assert mock_client.heartbeat_worker.call_count >= 1 kwargs = mock_client.heartbeat_worker.call_args.kwargs