diff --git a/CHANGELOG.md b/CHANGELOG.md index ef403c0e..91b0c55d 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] +### Fixed + +- Replay now enforces the recorded search attributes (Workflow protocol 1.20 replay identity). A workflow whose `upsert_search_attributes` differs from its `SearchAttributesUpserted` history fails replay with a `NonDeterministicReplayError` whose detail starts `search_attribute_value_mismatch`, as memo replay already does for memos. Values are compared as Core's `UpsertSearchAttributesCall` normalizes them (trimmed strings and list items, a blank string as null) and with their types (`1`, `1.0`, `True` and `"1"` differ). An authored tuple is checked as the JSON list it is sent as. Values outside that portable domain, on either side, are refused rather than coerced: non-string or non-portable keys, objects, sets, bytes, maps, non-string list items, non-finite floats and out-of-range integers. A `SearchAttributesUpserted` event without `attributes` is refused rather than skipped, with a detail starting `search_attribute_update_missing`. One without a workflow sequence is refused with `durable_command_sequence_missing` (or `durable_command_sequence_invalid` for zero, negative, string, float or bool) rather than given one, and a second upsert at the same sequence with `duplicate_search_attribute_update`. An upsert that shares its workflow sequence with any other durable command record is refused with `durable_command_sequence_collision`. Every lifecycle row that carries a command's sequence claims it, including activity heartbeat, retry and cancellation-acknowledged rows, child cancellation rows, service operation scheduling, cancellation scope openings, continue-as-new, and cooperative and scope cancellation deliveries, which claim every sequence in their `sequence_span` (absent or null means 1, as in Core; any other value that is not a positive integer is refused with `durable_command_sequence_invalid`); events that only point at a step, such as a signal receipt, an update or a cancellation request, claim no sequence. When the history records `attribute_types`, it must type exactly the update's non-null keys, and each type must be a known type that can hold the value; anything else, including a malformed type value, fails with `search_attribute_type_mismatch`. The SDK's upsert declares no types and the Server records a key's namespace-registered type ahead of the inferred one, so a string may be recorded as `keyword`, `string` or `datetime` and an integer as `int` or `float`; search-attribute type identity is therefore not pinned for Python. History without `attribute_types` keeps value-only identity. Replay diagnostics name reasons, keys, JSON types, sequences and event types, never a recorded or authored value. (#235) + +### Added + +- Replay of Workflow 2.19 redrive reuse is qualified for `MemoUpserted` and for `SearchAttributesUpserted` value identity. Neither event is a replay clock source: `ctx.now()` stays the persisted start time, which on a redriven run is the failed run's start, and the cancellation budget does not move on an upsert. Search-attribute type identity is not qualified, because the Server records namespace-registered types the SDK cannot see. Signal waits (`SignalWaitOpened`, `SignalReceived`, `SignalApplied`) are not qualified for Python: the SDK has no signal-wait primitive. Replaying such reused events through a signal handler applies the signal once and takes no time from them, but that does not qualify signal waits. (#235) + ## [2.13.1] - 2026-11-03 ### Fixed diff --git a/src/durable_workflow/_cancellation_scope_history.py b/src/durable_workflow/_cancellation_scope_history.py index bb843c10..c5b658d8 100644 --- a/src/durable_workflow/_cancellation_scope_history.py +++ b/src/durable_workflow/_cancellation_scope_history.py @@ -122,10 +122,15 @@ def _cleanup_operation(event: dict[str, Any], prefix: Sequence[dict[str, Any]], "authority_deadline_at": _canonical_time(authority), } sequence = payload.get("sequence") + # Only an explicit null reads as one call, as in Core; any other value + # that is not a positive integer is refused, never narrowed. + span = 1 if recorded.get("sequence_span") is None else recorded["sequence_span"] + if not _positive(span) or not _positive(recorded.get("sequence")): + raise _invalid(operation + " follows a delivery without a canonical sequence span") deadline = scope_timestamp(expected["authority_deadline_at"]) timestamp = scope_timestamp(event.get("timestamp", event.get("recorded_at"))) if (snapshot != expected or scope_id != context.scope_id - or not _positive(sequence) or sequence < recorded["sequence"] + recorded["sequence_span"] + or not _positive(sequence) or sequence < recorded["sequence"] + span or not _positive(event.get("sequence")) or delivery["sequence"] >= event["sequence"] or timestamp < scope_timestamp(delivery.get("timestamp", delivery.get("recorded_at"))) or timestamp >= deadline or not local and ( diff --git a/src/durable_workflow/_cooperative_cancellation.py b/src/durable_workflow/_cooperative_cancellation.py index 96540bbf..5e3fb688 100644 --- a/src/durable_workflow/_cooperative_cancellation.py +++ b/src/durable_workflow/_cooperative_cancellation.py @@ -112,7 +112,9 @@ def from_payload(cls, payload: Mapping[str, Any]) -> CancellationDelivery: request_id = _text(payload.get("workflow_command_id"), "workflow_command_id") sequence = _positive(payload.get("sequence"), "sequence") kind = _text(payload.get("call_kind"), "call_kind") - span = _positive(payload.get("sequence_span", 1), "sequence_span", maximum=1000) + # Absent and null both mean one call, as Core's WorkflowStepHistory reads it. + recorded_span = payload.get("sequence_span") + span = _positive(1 if recorded_span is None else recorded_span, "sequence_span", maximum=1000) operation = payload.get("operation_sequence") operation_span = _positive(payload.get("operation_sequence_span", 1), "operation_sequence_span", maximum=1000) if kind not in _CALL_KINDS or (kind != "parallel" and span != 1): diff --git a/src/durable_workflow/testing.py b/src/durable_workflow/testing.py index 5c64e1ef..7ef9018f 100644 --- a/src/durable_workflow/testing.py +++ b/src/durable_workflow/testing.py @@ -400,8 +400,19 @@ def _apply_commands( } ) elif isinstance(cmd, UpsertSearchAttributes): + # Record the JSON wire form, as the Server does; replay + # checks the workflow's update against it. history.append( - {"event_type": "SearchAttributesUpserted", "payload": {}} + { + "event_type": "SearchAttributesUpserted", + "payload": { + # Every recorded event but a signal is a step. + "sequence": 1 + sum( + event["event_type"] != "SignalReceived" for event in history + ), + "attributes": json.loads(json.dumps(cmd.attributes)), + }, + } ) elif isinstance(cmd, RecordVersionMarker): history.append( diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index e9df4bc4..f82c0716 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -1393,6 +1393,130 @@ def _memo_replay_identity(value: Mapping[str, Any]) -> str: return serializer.encode(_canonical_memo_entries(value), size_warning=None) +# Core's UpsertSearchAttributesCall, which the Server applies before storing +# SearchAttributesUpserted. Replay compares the canonical forms exactly and +# refuses anything outside this portable value domain. +_SEARCH_ATTRIBUTE_KEY_PATTERN = re.compile(r"[A-Za-z0-9_.:-]{1,64}") +_SEARCH_ATTRIBUTE_TRIM = " \t\n\r\0\x0b" +_SEARCH_ATTRIBUTE_MAX_STRING_LENGTH = 2048 +_SEARCH_ATTRIBUTE_MAX_KEYWORD_LENGTH = 255 +_SEARCH_ATTRIBUTE_TYPES = frozenset({"string", "keyword", "keyword_list", "int", "float", "bool", "datetime"}) +_INT64_MIN, _INT64_MAX = -(2**63), 2**63 - 1 + + +def _json_type_name(value: Any) -> str: + """A diagnostic-safe name for a value's JSON type; never the value itself.""" + if value is None: + return "null" + if isinstance(value, bool): + return "boolean" + if isinstance(value, int): + return "integer" + if isinstance(value, float): + return "number" + if isinstance(value, str): + return "string" + if isinstance(value, list | tuple): + return "array" + if isinstance(value, Mapping): + return "object" + return "unsupported value" + + +def _canonical_search_attributes(attributes: Any, *, authored: bool = False) -> dict[str, tuple[str, Any]]: + """Return the type-tagged canonical form of a search-attribute update. + + Raises ValueError for any value Core's UpsertSearchAttributesCall rejects. + The type tag keeps distinct values distinct (1, 1.0, True and "1"). An + authored tuple is sent as the same JSON list, so it is checked as a list. + """ + if not isinstance(attributes, Mapping) or not attributes: + raise ValueError("search attributes must be a non-empty object") + canonical: dict[str, tuple[str, Any]] = {} + for key, value in attributes.items(): + if not isinstance(key, str) or _SEARCH_ATTRIBUTE_KEY_PATTERN.fullmatch(key) is None: + # Name only the type: an invalid key may carry arbitrary data. + raise ValueError(f"a search attribute key ({_json_type_name(key)}) is not 1-64 URL-safe characters") + if authored and type(value) is tuple: + value = list(value) + if value is None: + canonical[key] = ("null", None) + elif isinstance(value, bool): + canonical[key] = ("bool", value) + elif isinstance(value, int): + if not _INT64_MIN <= value <= _INT64_MAX: + raise ValueError(f"search attribute [{key}] integer is outside the 64-bit range") + canonical[key] = ("int", value) + elif isinstance(value, float): + if not math.isfinite(value): + raise ValueError(f"search attribute [{key}] must be a finite number") + canonical[key] = ("float", value) + elif isinstance(value, str): + trimmed = value.strip(_SEARCH_ATTRIBUTE_TRIM) + if len(trimmed) > _SEARCH_ATTRIBUTE_MAX_STRING_LENGTH: + raise ValueError(f"search attribute [{key}] exceeds {_SEARCH_ATTRIBUTE_MAX_STRING_LENGTH} characters") + canonical[key] = ("string", trimmed) if trimmed else ("null", None) + elif type(value) is list or isinstance(value, Mapping) and not value: + # PHP decodes an empty JSON object as an empty list. + items: list[str] = [] + for item in value: + if not isinstance(item, str): + raise ValueError(f"search attribute [{key}] list values must contain only strings") + trimmed = item.strip(_SEARCH_ATTRIBUTE_TRIM) + if len(trimmed) > _SEARCH_ATTRIBUTE_MAX_KEYWORD_LENGTH: + raise ValueError( + f"search attribute [{key}] list values exceed {_SEARCH_ATTRIBUTE_MAX_KEYWORD_LENGTH} characters" + ) + items.append(trimmed) + canonical[key] = ("list", tuple(items)) + else: + raise ValueError(f"search attribute [{key}] must be a string, number, boolean, string list or null") + return canonical + + +def _search_attribute_type_choices(value: Any) -> frozenset[str]: + """Every type a Server can record for one value of this SDK's untyped upsert. + + The SDK declares no types, but native Server 3.0.x resolves each key as + registered, then declared, then inferred from the raw value + (rust/src/runtime/search_attribute_upserts.rs:254-289, recorded at :856), + and the PHP Server fills registered types into the command the same way + (SearchAttributeValueValidator.php:93, WorkerController.php:2734). The SDK + cannot see the namespace registry, so this returns the union of the + inferred type and every registered type the Server accepts for the value. + Any string is treated as possibly registered datetime. The check refuses + only a recorded type that cannot hold the value at all. + """ + if isinstance(value, bool): + return frozenset({"bool"}) + if isinstance(value, int): + return frozenset({"int", "float"}) + if isinstance(value, float): + return frozenset({"float"}) + if isinstance(value, str): + return frozenset({"keyword", "string", "datetime"}) + return frozenset({"keyword_list"}) + + +def _search_attribute_types_mismatch(recorded: Any, authored: Mapping[str, Any]) -> str | None: + if isinstance(recorded, list) and not recorded: + recorded = {} # PHP serializes an empty map as a JSON list. + if not isinstance(recorded, Mapping): + return "recorded attribute_types is not an object" + # A null value has no type. Both Servers type every other raw value, + # including a blank string that is stored as null. + typed = {key for key, value in authored.items() if value is not None} + if not all(isinstance(key, str) for key in recorded) or set(recorded) != typed: + return "recorded attribute_types names different keys from the current update" + for key in sorted(recorded): + kind = recorded[key] + if not isinstance(kind, str) or kind not in _SEARCH_ATTRIBUTE_TYPES: + return f"recorded attribute type for [{key}] is not a search attribute type" + if kind not in _search_attribute_type_choices(authored[key]): + return f"recorded attribute type {kind!r} for [{key}] cannot hold the current value" + return None + + def _canonical_memo_entries(entries: Mapping[str, Any], *, require_entries: bool = True) -> dict[str, Any]: if require_entries and not entries: raise ValueError("workflow memo updates require at least one entry") @@ -3719,6 +3843,16 @@ def _workflow_sequence(payload: Mapping[str, Any]) -> int | None: return None +def _durable_command_sequence_problem(payload: Mapping[str, Any]) -> str | None: + """The sdk-rust reason when a durable command record has no usable sequence.""" + value = payload.get("sequence") + if value is None: + return "durable_command_sequence_missing" + if type(value) is not int or value <= 0: + return "durable_command_sequence_invalid" + return None + + def _parallel_group_integer(payload: Mapping[str, Any], key: str) -> int | None: value = payload.get(key) if isinstance(value, bool): @@ -3873,6 +4007,93 @@ def _reorder_completed_parallel_groups( return ordered_results, ordered_steps +# The events that claim a workflow sequence, by durable command kind: Core's +# WorkflowStepHistory step events and cancellation deliveries, plus each +# command's other lifecycle rows, which carry the same sequence. Events that +# only point at a step's sequence claim nothing and are not listed: signal +# receipts, update records, failure handling, selection records, and +# cooperative or scope cancellation requests and preparations. As in sdk-php, +# a collision is refused when it involves an upsert: an upsert is the only +# command at its sequence. +_DURABLE_COMMAND_KINDS: dict[str, str] = { + **dict.fromkeys(("ActivityScheduled", "ActivityStarted", "ActivityHeartbeatRecorded", "ActivityRetryScheduled", + "ActivityCompleted", "ActivityFailed", "ActivityTimedOut", "ActivityCancelled", + "ActivityCancellationAcknowledged"), "activity"), + **dict.fromkeys(("TimerScheduled", "TimerFired", "TimerCancelled"), "timer"), + **dict.fromkeys(("ChildWorkflowScheduled", "ChildRunStarted", "ChildRunCompleted", "ChildRunFailed", + "ChildRunCancelled", "ChildRunTerminated", "ChildCancellationRequested", + "ChildCancellationResolved"), "child"), + **dict.fromkeys(("ConditionWaitOpened", "ConditionWaitSatisfied", "ConditionWaitTimedOut", + "ConditionWaitCancelled"), "condition"), + **dict.fromkeys(("SignalWaitOpened", "SignalApplied", "SignalWaitReceived", "SignalWaitTimedOut", + "SignalWaitCancelled"), "signal"), + **dict.fromkeys(("ServiceOperationScheduled", "ServiceCallStarted", "ServiceCallCompleted", "ServiceCallFailed", + "ServiceCallCancelled"), "service_call"), + "CancellationScopeOpened": "cancellation_scope", + # A delivered cancellation consumes the call it interrupts, scheduled or + # not, and its sequence_span reserves the whole interrupted range. + **dict.fromkeys(("CooperativeCancellationDelivered", "CancellationScopeDelivered"), "cancellation_delivery"), + "WorkflowContinuedAsNew": "continue_as_new", + "SideEffectRecorded": "side_effect", + "VersionMarkerRecorded": "version_marker", + "MemoUpserted": "memo", + "SearchAttributesUpserted": "search_attributes", +} +_UPSERT_COMMAND_KINDS = frozenset({"memo", "search_attributes"}) + + +def _delivery_sequence_span(payload: Mapping[str, Any]) -> int: + """A cancellation delivery's sequence_span under the producer contract. + + Absent or null means 1, as Core reads it + (WorkflowStepHistory.php:270, `intValue($payload['sequence_span'] ?? null) ?? 1`). + Otherwise it must be a positive JSON integer; zero, negative, bool, + float, string and every other value are malformed and raise ValueError, + so a malformed span never shrinks the claimed range. + """ + span = payload.get("sequence_span") + if span is None: + return 1 + if type(span) is not int or span <= 0: + # Name only the JSON type: the span may carry arbitrary history data. + raise ValueError(f"sequence_span must be a positive integer (got {_json_type_name(span)})") + return span + + +def _assert_delivery_spans(events: Iterable[Mapping[str, Any]]) -> None: + for event in events: + if _DURABLE_COMMAND_KINDS.get(_history_event_type(event) or "") != "cancellation_delivery": + continue + raw = event.get("payload") + payload: Mapping[str, Any] = raw if isinstance(raw, Mapping) else {} + try: + _delivery_sequence_span(payload) + except ValueError as error: + raise NonDeterministicReplayError( + _workflow_sequence(payload) or 0, + "cancellation delivery with a canonical sequence span", + [str(_history_event_type(event))], + detail=f"durable_command_sequence_invalid: {error}.", + ) from error + + +def _durable_command_claimed_range(event_type: str | None, payload: Mapping[str, Any], sequence: int) -> range: + """The workflow sequences one recorded event claims. + + A cancellation delivery claims sequence through sequence + sequence_span + - 1, inclusive, as Core's WorkflowStepHistory::nextSequenceFromEvents + extends it. Its span follows the producer contract + (_delivery_sequence_span), which replay checks before any parsing. + """ + if _DURABLE_COMMAND_KINDS.get(event_type or "") != "cancellation_delivery": + return range(sequence, sequence + 1) + return range(sequence, sequence + _delivery_sequence_span(payload)) + + +def _durable_command_kinds_compatible(kinds: frozenset[str]) -> bool: + return len(kinds) == 1 or not kinds & _UPSERT_COMMAND_KINDS + + def _internal_timeout_timer_kind( payload: Mapping[str, Any], condition_wait_ids_by_sequence: Mapping[int, str] | None = None, @@ -4309,6 +4530,28 @@ def _recorded_detail_mismatch(command: Any, step: _RecordedStep) -> str | None: recorded = step.details.get("entries_identity") if isinstance(recorded, str) and recorded != _memo_replay_identity(command.entries): return "Recorded memo entries differ from the current workflow memo update." + elif isinstance(command, UpsertSearchAttributes): + if "search_attributes" not in step.details: + return "search_attribute_update_missing: SearchAttributesUpserted history is missing its attributes." + try: + authored_attributes = _canonical_search_attributes(command.attributes, authored=True) + except ValueError as exc: + return f"search_attribute_value_mismatch: the current search attribute update is invalid: {exc}" + try: + recorded_attributes = _canonical_search_attributes(step.details["search_attributes"]) + except ValueError as exc: + return f"search_attribute_value_mismatch: SearchAttributesUpserted history has invalid attributes: {exc}" + if recorded_attributes != authored_attributes: + return ( + "search_attribute_value_mismatch: Recorded search attributes differ from the current " + "workflow search attribute update." + ) + # Events recorded before attribute_types existed keep value-only + # compatibility, as in Core's SearchAttributeReplayIdentity. + if "search_attribute_types" in step.details: + problem = _search_attribute_types_mismatch(step.details["search_attribute_types"], command.attributes) + if problem is not None: + return f"search_attribute_type_mismatch: {problem}." return None @@ -4501,6 +4744,9 @@ def _replay_state( ) from exception events = list(history_events) + # A malformed delivery span refuses replay before any history parser can + # read it more permissively. + _assert_delivery_spans(events) _assert_cancellation_scope_replay_supported( events, allow_authoring=allow_cancellation_scope_authoring, allow_delivery=allow_cancellation_scope_delivery, ) @@ -4583,6 +4829,10 @@ def _replay_state( resolved_sequences: set[int] = set() condition_wait_ids_by_sequence: dict[int, str] = {} selected_condition_wait_ids_by_sequence: dict[int, str] = {} + command_kinds_by_sequence: dict[int, frozenset[str]] = {} + search_attribute_sequences: set[int] = set() + upsert_sequences: set[int] = set() + delivery_ranges: list[range] = [] for event in events: event_type = _history_event_type(event) @@ -4590,8 +4840,62 @@ def _replay_state( if not isinstance(payload, Mapping): payload = {} sequence = _workflow_sequence(payload) + if event_type == "SearchAttributesUpserted": + # Never bind an upsert to a guessed step. + sequence_problem = _durable_command_sequence_problem(payload) + if sequence_problem is not None: + raise NonDeterministicReplayError( + len(event_types_by_sequence) + 1, + "search attributes upsert with its workflow sequence", + ["SearchAttributesUpserted"], + detail=f"{sequence_problem}: SearchAttributesUpserted history needs a positive integer " + "workflow sequence.", + ) if sequence is None: continue + if event_type == "SearchAttributesUpserted": + if sequence in search_attribute_sequences: + raise NonDeterministicReplayError( + sequence, + "one SearchAttributesUpserted event", + ["SearchAttributesUpserted", "SearchAttributesUpserted"], + detail="duplicate_search_attribute_update: a workflow sequence records two search " + "attribute updates.", + ) + search_attribute_sequences.add(sequence) + command_kind = _DURABLE_COMMAND_KINDS.get(event_type or "") + if command_kind is not None: + claimed = _durable_command_claimed_range(event_type, payload, sequence) + # A delivery span claims its whole range; an upsert anywhere inside + # it collides, whichever of the two was recorded first. + if command_kind in _UPSERT_COMMAND_KINDS: + inside = next((sequence for delivery in delivery_ranges if sequence in delivery), None) + elif command_kind == "cancellation_delivery": + inside = next((upsert for upsert in sorted(upsert_sequences) if upsert in claimed), None) + else: + inside = None + if inside is not None: + raise NonDeterministicReplayError( + inside, + "one durable command per workflow sequence", + [*event_types_by_sequence.get(inside, []), event_type or ""], + detail=f"durable_command_sequence_collision: {event_type} and an upsert claim workflow " + f"sequence {inside} inside a cancellation delivery span.", + ) + kinds = command_kinds_by_sequence.get(sequence, frozenset()) | {command_kind} + if not _durable_command_kinds_compatible(kinds): + raise NonDeterministicReplayError( + sequence, + "one durable command per workflow sequence", + [*event_types_by_sequence.get(sequence, []), event_type or ""], + detail=f"durable_command_sequence_collision: {event_type} claims a workflow sequence " + f"already recorded by a {', '.join(sorted(kinds - {command_kind}))} command.", + ) + command_kinds_by_sequence[sequence] = kinds + if command_kind in _UPSERT_COMMAND_KINDS: + upsert_sequences.add(sequence) + elif command_kind == "cancellation_delivery": + delivery_ranges.append(claimed) if event_type is not None: event_types_by_sequence.setdefault(sequence, []).append(event_type) try: @@ -5460,6 +5764,32 @@ def _receiver_condition_wait_bindings() -> dict[int, str | None]: memo_sequences_added.add(sequence) _append_resolved_result(None, "memo upsert", ev) elif etype == "SearchAttributesUpserted": + # The recorded values are this step's replay identity; without + # them the step cannot be checked, so replay is refused. + # The first pass refused an upsert without a positive integer sequence. + upsert_sequence = cast(int, _workflow_sequence(payload)) + if "attributes" not in payload: + raise NonDeterministicReplayError( + upsert_sequence, + "search attributes upsert with recorded attributes", + ["SearchAttributesUpserted"], + detail="search_attribute_update_missing: SearchAttributesUpserted history is missing its " + "attributes.", + ) + try: + _canonical_search_attributes(payload["attributes"]) + except ValueError as exc: + raise NonDeterministicReplayError( + upsert_sequence, + "search attributes upsert with canonical attributes", + ["SearchAttributesUpserted"], + detail=f"search_attribute_value_mismatch: SearchAttributesUpserted history has invalid " + f"attributes: {exc}", + ) from exc + upsert_details = details_by_sequence.setdefault(upsert_sequence, {}) + upsert_details["search_attributes"] = payload["attributes"] + if "attribute_types" in payload: + upsert_details["search_attribute_types"] = payload["attribute_types"] _append_resolved_result(None, "search attributes upsert", ev) elif etype == "SignalReceived": signal_name = payload.get("signal_name") diff --git a/tests/fixtures/replay_regressions/duplicate-search-attribute-update.json b/tests/fixtures/replay_regressions/duplicate-search-attribute-update.json new file mode 100644 index 00000000..dfc1da18 --- /dev/null +++ b/tests/fixtures/replay_regressions/duplicate-search-attribute-update.json @@ -0,0 +1,82 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "duplicate-search-attribute-update", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.redrive-visibility", + "input": [ + "Taylor" + ], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": {} + }, + { + "sequence": 2, + "event_type": "MemoUpserted", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": { + "sequence": 1, + "entries": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + }, + "merged": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + } + } + }, + { + "sequence": 3, + "event_type": "SearchAttributesUpserted", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": { + "sequence": 2, + "attributes": { + "Stage": "greeting" + }, + "attribute_types": { + "Stage": "keyword" + }, + "merged": { + "Stage": "greeting" + } + } + }, + { + "sequence": 4, + "event_type": "SearchAttributesUpserted", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": { + "sequence": 2, + "attributes": { + "Stage": "greeting" + }, + "attribute_types": { + "Stage": "keyword" + }, + "merged": { + "Stage": "greeting" + } + } + } + ], + "expected": { + "command_sequence": [] + }, + "expected_replay_error": { + "type": "NonDeterministicReplayError", + "message_contains": "duplicate_search_attribute_update", + "workflow_sequence": 2 + } +} diff --git a/tests/fixtures/replay_regressions/durable-command-sequence-collision.json b/tests/fixtures/replay_regressions/durable-command-sequence-collision.json new file mode 100644 index 00000000..91925fde --- /dev/null +++ b/tests/fixtures/replay_regressions/durable-command-sequence-collision.json @@ -0,0 +1,75 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "durable-command-sequence-collision", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.redrive-visibility", + "input": [ + "Taylor" + ], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": {} + }, + { + "sequence": 2, + "event_type": "MemoUpserted", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": { + "sequence": 1, + "entries": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + }, + "merged": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + } + } + }, + { + "sequence": 3, + "event_type": "SearchAttributesUpserted", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": { + "sequence": 2, + "attributes": { + "Stage": "greeting" + }, + "attribute_types": { + "Stage": "keyword" + }, + "merged": { + "Stage": "greeting" + } + } + }, + { + "sequence": 4, + "event_type": "ActivityScheduled", + "recorded_at": "2026-11-03T00:00:01+00:00", + "payload": { + "sequence": 2, + "activity_type": "tests.redrive.first", + "activity_execution_id": "activity-2" + } + } + ], + "expected": { + "command_sequence": [] + }, + "expected_replay_error": { + "type": "NonDeterministicReplayError", + "message_contains": "durable_command_sequence_collision", + "workflow_sequence": 2 + } +} diff --git a/tests/fixtures/replay_regressions/durable-command-sequence-invalid.json b/tests/fixtures/replay_regressions/durable-command-sequence-invalid.json new file mode 100644 index 00000000..3e87e64f --- /dev/null +++ b/tests/fixtures/replay_regressions/durable-command-sequence-invalid.json @@ -0,0 +1,65 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "durable-command-sequence-invalid", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.redrive-visibility", + "input": [ + "Taylor" + ], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": {} + }, + { + "sequence": 2, + "event_type": "MemoUpserted", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": { + "sequence": 1, + "entries": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + }, + "merged": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + } + } + }, + { + "sequence": 3, + "event_type": "SearchAttributesUpserted", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": { + "sequence": 0, + "attributes": { + "Stage": "greeting" + }, + "attribute_types": { + "Stage": "keyword" + }, + "merged": { + "Stage": "greeting" + } + } + } + ], + "expected": { + "command_sequence": [] + }, + "expected_replay_error": { + "type": "NonDeterministicReplayError", + "message_contains": "durable_command_sequence_invalid", + "workflow_sequence": 2 + } +} diff --git a/tests/fixtures/replay_regressions/durable-command-sequence-missing.json b/tests/fixtures/replay_regressions/durable-command-sequence-missing.json new file mode 100644 index 00000000..830472b9 --- /dev/null +++ b/tests/fixtures/replay_regressions/durable-command-sequence-missing.json @@ -0,0 +1,64 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "durable-command-sequence-missing", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.redrive-visibility", + "input": [ + "Taylor" + ], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": {} + }, + { + "sequence": 2, + "event_type": "MemoUpserted", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": { + "sequence": 1, + "entries": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + }, + "merged": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + } + } + }, + { + "sequence": 3, + "event_type": "SearchAttributesUpserted", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": { + "attributes": { + "Stage": "greeting" + }, + "attribute_types": { + "Stage": "keyword" + }, + "merged": { + "Stage": "greeting" + } + } + } + ], + "expected": { + "command_sequence": [] + }, + "expected_replay_error": { + "type": "NonDeterministicReplayError", + "message_contains": "durable_command_sequence_missing", + "workflow_sequence": 2 + } +} diff --git a/tests/fixtures/replay_regressions/redrive-reused-search-attribute-type-mismatch.json b/tests/fixtures/replay_regressions/redrive-reused-search-attribute-type-mismatch.json new file mode 100644 index 00000000..a5ac9962 --- /dev/null +++ b/tests/fixtures/replay_regressions/redrive-reused-search-attribute-type-mismatch.json @@ -0,0 +1,83 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "redrive-reused-search-attribute-type-mismatch", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.redrive-visibility", + "input": [ + "Taylor" + ], + "payload_codec": "avro", + "workflow_id": "regression-corpus-redrive-visibility", + "run_id": "01JBREDRIVESUCCESSORRUN0000" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "workflow_type": "tests.replay.redrive-visibility", + "workflow_instance_id": "regression-corpus-redrive-visibility", + "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000", + "workflow_command_id": "01JBREDRIVECOMMAND00000000", + "continued_from_run_id": "01JBREDRIVESOURCERUN000000", + "resume_step_sequence": 3, + "recovery_kind": "redrive", + "execution_deadline_at": null, + "run_deadline_at": null, + "replayed_started_at": "2026-11-03T00:00:00+00:00", + "workflow_link_id": "01JBREDRIVELINK00000000000" + } + }, + { + "sequence": 2, + "event_type": "MemoUpserted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "sequence": 1, + "entries": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + }, + "merged": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + }, + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:00+00:00" + } + }, + { + "sequence": 3, + "event_type": "SearchAttributesUpserted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "sequence": 2, + "attributes": { + "Stage": "greeting" + }, + "attribute_types": { + "Stage": "int" + }, + "merged": { + "Stage": "greeting" + }, + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:00+00:00" + } + } + ], + "expected": { + "command_sequence": [] + }, + "expected_replay_error": { + "type": "NonDeterministicReplayError", + "message_contains": "search_attribute_type_mismatch", + "workflow_sequence": 2 + } +} diff --git a/tests/fixtures/replay_regressions/redrive-reused-search-attribute-update-missing.json b/tests/fixtures/replay_regressions/redrive-reused-search-attribute-update-missing.json new file mode 100644 index 00000000..0f538217 --- /dev/null +++ b/tests/fixtures/replay_regressions/redrive-reused-search-attribute-update-missing.json @@ -0,0 +1,80 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "redrive-reused-search-attribute-update-missing", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.redrive-visibility", + "input": [ + "Taylor" + ], + "payload_codec": "avro", + "workflow_id": "regression-corpus-redrive-visibility", + "run_id": "01JBREDRIVESUCCESSORRUN0000" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "workflow_type": "tests.replay.redrive-visibility", + "workflow_instance_id": "regression-corpus-redrive-visibility", + "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000", + "workflow_command_id": "01JBREDRIVECOMMAND00000000", + "continued_from_run_id": "01JBREDRIVESOURCERUN000000", + "resume_step_sequence": 3, + "recovery_kind": "redrive", + "execution_deadline_at": null, + "run_deadline_at": null, + "replayed_started_at": "2026-11-03T00:00:00+00:00", + "workflow_link_id": "01JBREDRIVELINK00000000000" + } + }, + { + "sequence": 2, + "event_type": "MemoUpserted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "sequence": 1, + "entries": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + }, + "merged": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + }, + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:00+00:00" + } + }, + { + "sequence": 3, + "event_type": "SearchAttributesUpserted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "sequence": 2, + "attribute_types": { + "Stage": "keyword" + }, + "merged": { + "Stage": "greeting" + }, + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:00+00:00" + } + } + ], + "expected": { + "command_sequence": [] + }, + "expected_replay_error": { + "type": "NonDeterministicReplayError", + "message_contains": "search_attribute_update_missing", + "workflow_sequence": 2 + } +} diff --git a/tests/fixtures/replay_regressions/redrive-reused-search-attribute-value-mismatch.json b/tests/fixtures/replay_regressions/redrive-reused-search-attribute-value-mismatch.json new file mode 100644 index 00000000..8292c5ab --- /dev/null +++ b/tests/fixtures/replay_regressions/redrive-reused-search-attribute-value-mismatch.json @@ -0,0 +1,83 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "redrive-reused-search-attribute-value-mismatch", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.redrive-visibility-changed-stage", + "input": [ + "Taylor" + ], + "payload_codec": "avro", + "workflow_id": "regression-corpus-redrive-visibility", + "run_id": "01JBREDRIVESUCCESSORRUN0000" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "workflow_type": "tests.replay.redrive-visibility-changed-stage", + "workflow_instance_id": "regression-corpus-redrive-visibility", + "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000", + "workflow_command_id": "01JBREDRIVECOMMAND00000000", + "continued_from_run_id": "01JBREDRIVESOURCERUN000000", + "resume_step_sequence": 3, + "recovery_kind": "redrive", + "execution_deadline_at": null, + "run_deadline_at": null, + "replayed_started_at": "2026-11-03T00:00:00+00:00", + "workflow_link_id": "01JBREDRIVELINK00000000000" + } + }, + { + "sequence": 2, + "event_type": "MemoUpserted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "sequence": 1, + "entries": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + }, + "merged": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + }, + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:00+00:00" + } + }, + { + "sequence": 3, + "event_type": "SearchAttributesUpserted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "sequence": 2, + "attributes": { + "Stage": "greeting" + }, + "attribute_types": { + "Stage": "keyword" + }, + "merged": { + "Stage": "greeting" + }, + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:00+00:00" + } + } + ], + "expected": { + "command_sequence": [] + }, + "expected_replay_error": { + "type": "NonDeterministicReplayError", + "message_contains": "search_attribute_value_mismatch", + "workflow_sequence": 2 + } +} diff --git a/tests/fixtures/replay_regressions/redrive-reused-signal-wait-clock.json b/tests/fixtures/replay_regressions/redrive-reused-signal-wait-clock.json new file mode 100644 index 00000000..6bc105cb --- /dev/null +++ b/tests/fixtures/replay_regressions/redrive-reused-signal-wait-clock.json @@ -0,0 +1,105 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "redrive-reused-signal-wait-clock", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.redrive-signal", + "input": [ + "Taylor" + ], + "payload_codec": "avro", + "workflow_id": "regression-corpus-redrive-signal", + "run_id": "01JBREDRIVESUCCESSORRUN0000" + }, + "redrive_reused_signal_wait": { + "signal_name": "approve", + "signal_value": "Grace", + "redrive_after_minutes": 5, + "redrive": "accepted", + "successor_status": "completed", + "reused_events": [ + "SignalWaitOpened", + "SignalReceived", + "SignalApplied" + ] + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "workflow_type": "tests.replay.redrive-signal", + "workflow_instance_id": "regression-corpus-redrive-signal", + "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000", + "workflow_command_id": "01JBREDRIVECOMMAND00000000", + "continued_from_run_id": "01JBREDRIVESOURCERUN000000", + "resume_step_sequence": 2, + "recovery_kind": "redrive", + "execution_deadline_at": null, + "run_deadline_at": null, + "replayed_started_at": "2026-11-03T00:00:00+00:00", + "workflow_link_id": "01JBREDRIVELINK00000000000" + } + }, + { + "sequence": 2, + "event_type": "SignalWaitOpened", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "signal_name": "approve", + "signal_wait_id": "01M6H8VWAMNSM12WCT4KJ4EQN3", + "sequence": 1, + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:00+00:00" + } + }, + { + "sequence": 3, + "event_type": "SignalReceived", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "workflow_command_id": "01m6h8vwbfwstfzz0nn2gy924h", + "signal_id": "01m6h8vwbfwstfzz0nn2gy924j", + "workflow_instance_id": "regression-corpus-redrive-signal", + "workflow_run_id": "01JBREDRIVESOURCERUN000000", + "signal_name": "approve", + "signal_wait_id": "01M6H8VWAMNSM12WCT4KJ4EQN3", + "arguments": "wwHioz3/VYAiNwwCCgpHcmFjZQA=", + "payload_codec": "avro", + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:00+00:00" + } + }, + { + "sequence": 4, + "event_type": "SignalApplied", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "workflow_command_id": "01m6h8vwbfwstfzz0nn2gy924h", + "signal_id": "01m6h8vwbfwstfzz0nn2gy924j", + "signal_name": "approve", + "signal_wait_id": "01M6H8VWAMNSM12WCT4KJ4EQN3", + "sequence": 1, + "value": "wwHioz3/VYAiNwoKR3JhY2U=", + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:00+00:00" + } + } + ], + "expected": { + "command_sequence": [ + { + "type": "schedule_activity", + "activity_type": "tests.redrive.first", + "arguments": [ + "Taylor" + ] + } + ] + } +} diff --git a/tests/fixtures/replay_regressions/redrive-reused-upserts-cancellation-clock.json b/tests/fixtures/replay_regressions/redrive-reused-upserts-cancellation-clock.json new file mode 100644 index 00000000..a86ddff2 --- /dev/null +++ b/tests/fixtures/replay_regressions/redrive-reused-upserts-cancellation-clock.json @@ -0,0 +1,154 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "redrive-reused-upserts-cancellation-clock", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.redrive-visibility-cleanup", + "input": [ + "Taylor" + ], + "payload_codec": "avro", + "workflow_id": "regression-corpus-redrive-visibility", + "run_id": "01JBREDRIVESUCCESSORRUN0000" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "workflow_type": "tests.replay.redrive-visibility-cleanup", + "workflow_instance_id": "regression-corpus-redrive-visibility", + "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000", + "workflow_command_id": "01JBREDRIVECOMMAND00000000", + "continued_from_run_id": "01JBREDRIVESOURCERUN000000", + "resume_step_sequence": 3, + "recovery_kind": "redrive", + "execution_deadline_at": null, + "run_deadline_at": null, + "replayed_started_at": "2026-11-03T00:00:00+00:00", + "workflow_link_id": "01JBREDRIVELINK00000000000" + } + }, + { + "sequence": 2, + "event_type": "MemoUpserted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "sequence": 1, + "entries": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + }, + "merged": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + }, + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:00+00:00" + } + }, + { + "sequence": 3, + "event_type": "SearchAttributesUpserted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "sequence": 2, + "attributes": { + "Stage": "greeting" + }, + "attribute_types": { + "Stage": "keyword" + }, + "merged": { + "Stage": "greeting" + }, + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:00+00:00" + } + }, + { + "sequence": 4, + "event_type": "CooperativeCancellationRequested", + "recorded_at": "2026-11-03T00:05:02+00:00", + "payload": { + "workflow_command_id": "redrive-cancel-1", + "workflow_instance_id": "regression-corpus-redrive-visibility", + "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000", + "reason": "maintenance", + "cleanup_deadline_at": "2026-11-03T00:05:30.000000Z", + "cancellation": { + "schema": "durable-workflow.cancellation-context/v1", + "request_id": "redrive-cancel-1", + "root_request_id": "redrive-cancel-1", + "root_workflow_instance_id": "regression-corpus-redrive-visibility", + "root_workflow_run_id": "01JBREDRIVESUCCESSORRUN0000", + "parent_request_id": null, + "reason": "maintenance", + "requester": { + "type": "operator", + "id": "operator-1", + "label": "Maintainer" + }, + "source": "control_plane", + "requested_at": "2026-11-03T00:05:02.000000Z", + "cleanup_deadline_at": "2026-11-03T00:05:30.000000Z", + "lineage": [ + { + "request_id": "redrive-cancel-1", + "workflow_instance_id": "regression-corpus-redrive-visibility", + "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000" + } + ] + } + } + }, + { + "sequence": 5, + "event_type": "CooperativeCancellationDelivered", + "recorded_at": "2026-11-03T00:05:04+00:00", + "payload": { + "workflow_command_id": "redrive-cancel-1", + "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000", + "sequence": 3, + "call_kind": "activity", + "cancellation": { + "schema": "durable-workflow.cancellation-context/v1", + "request_id": "redrive-cancel-1", + "root_request_id": "redrive-cancel-1", + "root_workflow_instance_id": "regression-corpus-redrive-visibility", + "root_workflow_run_id": "01JBREDRIVESUCCESSORRUN0000", + "parent_request_id": null, + "reason": "maintenance", + "requester": { + "type": "operator", + "id": "operator-1", + "label": "Maintainer" + }, + "source": "control_plane", + "requested_at": "2026-11-03T00:05:02.000000Z", + "cleanup_deadline_at": "2026-11-03T00:05:30.000000Z", + "lineage": [ + { + "request_id": "redrive-cancel-1", + "workflow_instance_id": "regression-corpus-redrive-visibility", + "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000" + } + ] + } + } + } + ], + "expected": { + "type": "complete_workflow", + "result": { + "after_memo": "2026-11-03T00:00:00+00:00", + "after_search_attributes": "2026-11-03T00:00:00+00:00", + "remaining": 26.0 + } + } +} diff --git a/tests/fixtures/replay_regressions/redrive-reused-upserts-clock.json b/tests/fixtures/replay_regressions/redrive-reused-upserts-clock.json new file mode 100644 index 00000000..70512e98 --- /dev/null +++ b/tests/fixtures/replay_regressions/redrive-reused-upserts-clock.json @@ -0,0 +1,108 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "redrive-reused-upserts-clock", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.redrive-visibility", + "input": [ + "Taylor" + ], + "payload_codec": "avro", + "workflow_id": "regression-corpus-redrive-visibility", + "run_id": "01JBREDRIVESUCCESSORRUN0000" + }, + "redrive_reused_upserts": { + "redrive_after_minutes": 5, + "redrive": "accepted", + "successor_status": "completed", + "reused_steps": [ + { + "sequence": 1, + "event_type": "MemoUpserted" + }, + { + "sequence": 2, + "event_type": "SearchAttributesUpserted" + } + ], + "memo": { + "customer": "Taylor", + "stage": "greeting" + }, + "search_attributes": { + "Stage": "greeting" + } + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "workflow_type": "tests.replay.redrive-visibility", + "workflow_instance_id": "regression-corpus-redrive-visibility", + "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000", + "workflow_command_id": "01JBREDRIVECOMMAND00000000", + "continued_from_run_id": "01JBREDRIVESOURCERUN000000", + "resume_step_sequence": 3, + "recovery_kind": "redrive", + "execution_deadline_at": null, + "run_deadline_at": null, + "replayed_started_at": "2026-11-03T00:00:00+00:00", + "workflow_link_id": "01JBREDRIVELINK00000000000" + } + }, + { + "sequence": 2, + "event_type": "MemoUpserted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "sequence": 1, + "entries": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + }, + "merged": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + }, + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:00+00:00" + } + }, + { + "sequence": 3, + "event_type": "SearchAttributesUpserted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "sequence": 2, + "attributes": { + "Stage": "greeting" + }, + "attribute_types": { + "Stage": "keyword" + }, + "merged": { + "Stage": "greeting" + }, + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:00+00:00" + } + } + ], + "expected": { + "command_sequence": [ + { + "type": "schedule_activity", + "activity_type": "tests.redrive.first", + "arguments": [ + "Taylor" + ] + } + ] + } +} diff --git a/tests/fixtures/replay_regressions/search-attribute-update-missing.json b/tests/fixtures/replay_regressions/search-attribute-update-missing.json new file mode 100644 index 00000000..354fddda --- /dev/null +++ b/tests/fixtures/replay_regressions/search-attribute-update-missing.json @@ -0,0 +1,62 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "search-attribute-update-missing", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.redrive-visibility", + "input": [ + "Taylor" + ], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": {} + }, + { + "sequence": 2, + "event_type": "MemoUpserted", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": { + "sequence": 1, + "entries": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + }, + "merged": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + } + } + }, + { + "sequence": 3, + "event_type": "SearchAttributesUpserted", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": { + "sequence": 2, + "attribute_types": { + "Stage": "keyword" + }, + "merged": { + "Stage": "greeting" + } + } + } + ], + "expected": { + "command_sequence": [] + }, + "expected_replay_error": { + "type": "NonDeterministicReplayError", + "message_contains": "search_attribute_update_missing", + "workflow_sequence": 2 + } +} diff --git a/tests/fixtures/replay_regressions/search-attribute-value-mismatch.json b/tests/fixtures/replay_regressions/search-attribute-value-mismatch.json new file mode 100644 index 00000000..eecdac2d --- /dev/null +++ b/tests/fixtures/replay_regressions/search-attribute-value-mismatch.json @@ -0,0 +1,65 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "search-attribute-value-mismatch", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.redrive-visibility-changed-stage", + "input": [ + "Taylor" + ], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": {} + }, + { + "sequence": 2, + "event_type": "MemoUpserted", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": { + "sequence": 1, + "entries": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + }, + "merged": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + } + } + }, + { + "sequence": 3, + "event_type": "SearchAttributesUpserted", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": { + "sequence": 2, + "attributes": { + "Stage": "greeting" + }, + "attribute_types": { + "Stage": "keyword" + }, + "merged": { + "Stage": "greeting" + } + } + } + ], + "expected": { + "command_sequence": [] + }, + "expected_replay_error": { + "type": "NonDeterministicReplayError", + "message_contains": "search_attribute_value_mismatch", + "workflow_sequence": 2 + } +} diff --git a/tests/test_cancellation_remaining_time.py b/tests/test_cancellation_remaining_time.py index cc32e1e0..24d85410 100644 --- a/tests/test_cancellation_remaining_time.py +++ b/tests/test_cancellation_remaining_time.py @@ -6,7 +6,7 @@ import pytest -from durable_workflow import CancellationContext, serializer +from durable_workflow import CancellationContext, serializer, workflow from durable_workflow.errors import ActivityFailed, DurableOperationCancelled, WorkflowCancelled from durable_workflow.workflow import ( CompleteWorkflow, @@ -282,3 +282,48 @@ def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] events = history() + [{**reused, "timestamp": "2026-10-01T00:00:29Z"}] # Redrive recorded the step again at 00:29; its workflow time stays 00:12. assert replay(Cleanup, events, [], run_id="run-1").commands == [CompleteWorkflow(18.123456)] + + + +@pytest.mark.parametrize("later", ["2026-10-01T00:00:12Z", "2026-10-01T00:00:29Z"]) +def test_upsert_and_signal_events_during_cleanup_do_not_move_the_remaining_budget(later: str) -> None: + # An ordinary run: the cancellation is delivered, then cleanup upserts and + # receives a signal. None of these events is a clock source, so the budget + # stays at the delivery time whatever time they carry. (Reused events + # always precede a redrive successor's cancellation; see test_replay_clock.) + @workflow.defn(name="tests.cancellation.cleanup-upserts-signal") + class Cleanup: + def __init__(self) -> None: + self.approved_by: str | None = None + + @workflow.signal("approve") + def approve(self, approved_by: str) -> None: + self.approved_by = approved_by + + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + try: + yield ctx.start_timer(60) + except WorkflowCancelled as error: + assert error.context is not None + observed = [error.context.remaining()] + with ctx.cancellation_shield(): + yield ctx.upsert_memo({"stage": "cleanup"}) + observed.append(error.context.remaining()) + yield ctx.upsert_search_attributes({"Stage": "cleanup"}) + observed.append(error.context.remaining()) + return [observed, self.approved_by] + + memo = serializer.envelope({"stage": "cleanup"}, codec="avro") + events = history() + [{**event, "timestamp": later} for event in [ + {"event_type": "SignalReceived", "payload": { + "signal_name": "approve", "arguments": serializer.envelope(["Grace"], codec="avro"), + "payload_codec": "avro", + }}, + {"event_type": "MemoUpserted", "payload": {"sequence": 2, "entries": memo, "merged": memo}}, + {"event_type": "SearchAttributesUpserted", "payload": { + "sequence": 3, "attributes": {"Stage": "cleanup"}, "merged": {"Stage": "cleanup"}, + }}, + ]] + assert replay(Cleanup, events, [], run_id="run-1").commands == [ + CompleteWorkflow([[22.123456] * 3, "Grace"]), + ] diff --git a/tests/test_committed_cancellation_scope_replay.py b/tests/test_committed_cancellation_scope_replay.py index aae70dc5..795ab1dd 100644 --- a/tests/test_committed_cancellation_scope_replay.py +++ b/tests/test_committed_cancellation_scope_replay.py @@ -277,3 +277,99 @@ def inner(): # type: ignore[no-untyped-def] with pytest.raises((NonDeterministicReplayError, LocalActivityExecutionAborted)): run(Probe, value) + + +def _scoped_cleanup_probe() -> type: + class Probe: + def run(self, ctx: workflow.WorkflowContext): # type: ignore[no-untyped-def] + def inner(): # type: ignore[no-untyped-def] + try: + yield ctx.start_timer(10) + except WorkflowCancelled as error: + with ctx.cancellation_shield(): + yield ctx.start_timer(1) + return error.context + + def outer(): # type: ignore[no-untyped-def] + return (yield from ctx.cancellation_scope(inner)) + + cancellation = yield from ctx.cancellation_scope(outer) + yield ctx.start_timer(1) + return {"request": cancellation.request_id, "remaining": cancellation.remaining()} + + return Probe + + +def _with_delivery_span(value: dict[str, Any], span: Any) -> dict[str, Any]: + for event in value["history"]: + if event["event_type"] in {"CancellationScopeDeliveryPrepared", "CancellationScopeDelivered"}: + event["payload"]["sequence_span"] = span + return value + + +def _scoped_cleanup_outcomes(span: Any) -> list[Any]: + """Replay the cleanup timer path (whose scope parser applies the cleanup + lower bound, delivery sequence + span) through to completion.""" + probe = _scoped_cleanup_probe() + value = _with_delivery_span(cleanup_timer_history(), span) + outcomes: list[Any] = [run(probe, value).commands] + append_cleanup_event(value, "TimerFired", { + "sequence": 4, "timer_id": "cleanup-timer", "delay_seconds": 1, + "cancellation_scope_id": value["history"][-1]["payload"]["cancellation_scope_id"], + }, "2026-10-04T00:00:09.123456Z") + outcomes.append(workflow.commands_to_server_commands(run(probe, value).commands, "queue")) + append_cleanup_event(value, "TimerScheduled", { + "sequence": 5, "timer_id": "parent-timer", "delay_seconds": 1, "fire_at": "2026-10-04T00:00:10.123456Z", + }, "2026-10-04T00:00:09.123456Z") + append_cleanup_event(value, "TimerFired", { + "sequence": 5, "timer_id": "parent-timer", "delay_seconds": 1, + }, "2026-10-04T00:00:10.123456Z") + outcomes.append(run(probe, value).commands) + return outcomes + + +def test_a_null_scope_delivery_span_replays_the_cleanup_path_exactly_like_one() -> None: + one = _scoped_cleanup_outcomes(1) + assert one[0] == [] and one[1] == [{"type": "start_timer", "delay_seconds": 1}] + assert isinstance(one[2][0], workflow.CompleteWorkflow) + assert _scoped_cleanup_outcomes(None) == one + + +def _cleanup_at(sequence: int, span: Any) -> Any: + value = _with_delivery_span(cleanup_timer_history(), span) + value["history"][-1]["payload"]["sequence"] = sequence + try: + return run(_scoped_cleanup_probe(), value).commands + except (LocalActivityExecutionAborted, NonDeterministicReplayError) as error: + return type(error).__name__, str(error).split(" Keep ")[0] + + +def test_a_null_scope_delivery_span_has_the_same_cleanup_lower_bound_as_one() -> None: + delivered = next(event for event in cleanup_timer_history()["history"] + if event["event_type"] == "CancellationScopeDelivered")["payload"]["sequence"] + # The cleanup timer may not reuse the delivered call's sequence. + below = _cleanup_at(delivered, 1) + assert isinstance(below, tuple) and "changes its original delivery or authority ceiling" in below[1] + assert _cleanup_at(delivered, None) == below + assert _cleanup_at(delivered + 1, None) == _cleanup_at(delivered + 1, 1) == [] + + +@pytest.mark.parametrize("span", [0, False, -1, "1", 1.0, ""]) +def test_a_malformed_scope_delivery_span_is_still_refused(span: Any) -> None: + value = _with_delivery_span(cleanup_timer_history(), span) + with pytest.raises(NonDeterministicReplayError, match="durable_command_sequence_invalid:"): + run(_scoped_cleanup_probe(), value) + + +@pytest.mark.parametrize("span", [0, False, -1, "1", 1.0, ""]) +def test_the_cleanup_lower_bound_refuses_a_malformed_span_on_its_own(span: Any) -> None: + # Called directly, without the replay-level span check, the scope parser + # still refuses rather than reading a falsy span as one call. + from durable_workflow._cancellation_scope_history import _cleanup_operation + + value = _with_delivery_span(cleanup_timer_history(), span) + history = value["history"] + with pytest.raises(ValueError, match="without a canonical sequence span"): + _cleanup_operation(history[-1], history[:-1]) + _with_delivery_span(value, None) + assert _cleanup_operation(history[-1], history[:-1]) is True diff --git a/tests/test_history_event_contract.py b/tests/test_history_event_contract.py index 6311c1b8..608aba4a 100644 --- a/tests/test_history_event_contract.py +++ b/tests/test_history_event_contract.py @@ -674,16 +674,26 @@ } -# Keys Core adds only to runs created by redrive (workflow#864). The SDK reads -# replayed_started_at for the start clock and reused_recorded_at for step time. +# Keys Core adds only to runs created by redrive (workflow#864, #1001, #1002). +# The SDK reads replayed_started_at for the start clock and reused_recorded_at +# for the cancellation step time. Upserts and signal-wait events are not clock +# sources here (test_replay_clock.py proves it), so their reuse keys are inert. REDRIVE_HISTORY_EVENT_PAYLOAD_KEYS = { "WorkflowStarted": ("resume_step_sequence", "recovery_kind", "replayed_started_at"), "ActivityCompleted": ("reused_from_run_id", "reused_activity_execution_id", "reused_recorded_at"), "TimerFired": ("reused_from_run_id", "reused_recorded_at"), "SideEffectRecorded": ("reused_from_run_id", "reused_recorded_at"), "VersionMarkerRecorded": ("reused_from_run_id", "reused_recorded_at"), + "MemoUpserted": ("reused_from_run_id", "reused_recorded_at"), + "SearchAttributesUpserted": ("reused_from_run_id", "reused_recorded_at"), + "SignalWaitOpened": ("reused_from_run_id", "reused_recorded_at"), + "SignalReceived": ("reused_from_run_id", "reused_recorded_at"), + "SignalApplied": ("reused_from_run_id", "reused_recorded_at"), } +# The SDK authors no signal waits, so these reused events are preserved only. +REDRIVE_PRESERVED_ONLY_HISTORY_EVENTS = frozenset({"SignalWaitOpened", "SignalApplied"}) + for _event_type, _keys in REDRIVE_HISTORY_EVENT_PAYLOAD_KEYS.items(): DOCUMENTED_HISTORY_EVENT_PAYLOAD_KEYS[_event_type] = ( *DOCUMENTED_HISTORY_EVENT_PAYLOAD_KEYS[_event_type], @@ -705,6 +715,7 @@ "VersionMarkerRecorded", "ChildRunCompleted", "ChildRunFailed", + "MemoUpserted", "SearchAttributesUpserted", } ) @@ -761,5 +772,7 @@ def test_php_history_fixture_key_expectations_match_documented_contract() -> Non def test_redrive_payload_keys_are_documented_on_the_interpreted_events() -> None: for event_type, keys in REDRIVE_HISTORY_EVENT_PAYLOAD_KEYS.items(): - assert event_type in SDK_INTERPRETED_HISTORY_EVENTS, event_type + expected = (SDK_PRESERVED_ONLY_HISTORY_EVENTS if event_type in REDRIVE_PRESERVED_ONLY_HISTORY_EVENTS + else SDK_INTERPRETED_HISTORY_EVENTS) + assert event_type in expected, event_type assert set(keys) <= set(DOCUMENTED_HISTORY_EVENT_PAYLOAD_KEYS[event_type]), event_type diff --git a/tests/test_replay.py b/tests/test_replay.py index ca80eb84..c237a875 100644 --- a/tests/test_replay.py +++ b/tests/test_replay.py @@ -1211,7 +1211,8 @@ def test_pending_metadata_precedes_every_continue_authoring_form(self, workflow_ replayed = replay( workflow_cls, [ - {"event_type": "SearchAttributesUpserted", "payload": {"sequence": 1}}, + {"event_type": "SearchAttributesUpserted", + "payload": {"sequence": 1, "attributes": {"stage": "continued"}}}, { "event_type": "MemoUpserted", "payload": {"sequence": 2, "entries": entries, "merged": merged}, @@ -1623,7 +1624,8 @@ def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] yield ctx.wait_condition(lambda: self.approved_by is not None, key="approval", timeout=30) side_effect = yield ctx.side_effect(lambda: {"side_effect": "unstable"}) child = yield ctx.start_child_workflow("tests.polyglot.child", []) - yield ctx.upsert_search_attributes({"status": "processing"}) + # The PHP-emitted fixture records {"status": "done"}; replay enforces it. + yield ctx.upsert_search_attributes({"status": "done"}) return { "activity": activity, @@ -2070,7 +2072,8 @@ def test_first_replay_upserts_then_schedules(self) -> None: def test_with_upsert_in_history(self) -> None: history = [ - {"event_type": "SearchAttributesUpserted", "payload": {}}, + {"event_type": "SearchAttributesUpserted", + "payload": {"sequence": 1, "attributes": {"status": "processing"}}}, {"event_type": "ActivityCompleted", "payload": {"result": _avro("result")}}, ] outcome = replay(SearchAttrWorkflow, history, []) diff --git a/tests/test_replay_clock.py b/tests/test_replay_clock.py index f2e2cfcc..7a4ee13e 100644 --- a/tests/test_replay_clock.py +++ b/tests/test_replay_clock.py @@ -1,15 +1,17 @@ from __future__ import annotations +import json from copy import deepcopy from datetime import datetime +from pathlib import Path from typing import Any from uuid import UUID import pytest from durable_workflow import serializer, workflow -from durable_workflow.errors import NonDeterministicReplayError -from durable_workflow.workflow import CompleteWorkflow, WorkflowContext, replay +from durable_workflow.errors import NonDeterministicReplayError, WorkflowCancelled +from durable_workflow.workflow import CompleteWorkflow, WorkflowContext, query_state, replay @workflow.defn(name="tests.replay.persisted-clock") @@ -131,3 +133,672 @@ def test_invalid_replayed_start_time_is_refused_without_fallback(invalid: Any) - history = redriven_history(recovery_kind="redrive", replayed_started_at=invalid) with pytest.raises(NonDeterministicReplayError, match="replayed_started_at"): replay(RedriveClockWorkflow, history, [], run_id="redriven-run", payload_codec="avro") + + +# Core 2.19 lets redrive reuse memo and search-attribute upserts (workflow#1001) +# and an applied signal wait (workflow#1002). Core replays now() after them at +# payload.reused_recorded_at. This SDK's now() is the persisted start time, so +# none of these five events is a clock source: on a redriven run now() is the +# failed run's start (WorkflowStarted.replayed_started_at), whatever time the +# reused events carry. +REDRIVE_FIXTURES = Path(__file__).parent / "fixtures" / "replay_regressions" +REDRIVE_SOURCE_RUN = "01JBREDRIVESOURCERUN000000" +REDRIVE_REPLAYED_START = "2026-11-03T00:00:00+00:00" + + +@workflow.defn(name="tests.replay.redrive-visibility") +class RedriveVisibilityWorkflow: + """Python counterpart of Core's TestRedriveVisibilityWorkflow.""" + + stage = "greeting" + + def __init__(self) -> None: + self.times: dict[str, str] = {} + + def run(self, ctx: WorkflowContext, name: str): # type: ignore[no-untyped-def] + yield ctx.upsert_memo({"stage": "greeting", "customer": name}) + self.times["after_memo"] = ctx.now().isoformat() + yield ctx.upsert_search_attributes({"Stage": self.stage}) + self.times["after_search_attributes"] = ctx.now().isoformat() + greeting = yield ctx.schedule_activity("tests.redrive.first", [name]) + return (yield ctx.schedule_activity("tests.redrive.second", [greeting])) + + @workflow.query("replay_times") + def replay_times(self) -> dict[str, str]: + return self.times + + +@workflow.defn(name="tests.replay.redrive-signal") +class RedriveSignalWorkflow: + """Python counterpart of Core's TestRedriveSignalWorkflow. + + The SDK has no signal-wait primitive; a handler plus a condition waits for + the signal. SignalWaitOpened and SignalApplied are preserved, not replayed. + """ + + applied: list[Any] = [] + + def __init__(self) -> None: + self.approved_by: Any = None + self.time_after_signal: str | None = None + + @workflow.signal("approve") + def approve(self, approved_by: Any) -> None: + type(self).applied.append(approved_by) + self.approved_by = approved_by + + def run(self, ctx: WorkflowContext, name: str): # type: ignore[no-untyped-def] + yield ctx.wait_condition(lambda: self.approved_by is not None, key="approve") + self.time_after_signal = ctx.now().isoformat() + greeting = yield ctx.schedule_activity("tests.redrive.first", [name]) + return (yield ctx.schedule_activity("tests.redrive.second", [greeting])) + + @workflow.query("replay_state") + def replay_state(self) -> dict[str, Any]: + return {"approved_by": self.approved_by, "time_after_signal": self.time_after_signal} + + +@workflow.defn(name="tests.replay.redrive-visibility-changed-stage") +class RedriveChangedStageWorkflow(RedriveVisibilityWorkflow): + """The visibility workflow after a deploy changed its search attribute.""" + + stage = "approval" + + +@workflow.defn(name="tests.replay.redrive-visibility-cleanup") +class RedriveVisibilityCleanupWorkflow(RedriveVisibilityWorkflow): + """Observes the cancellation budget when a cancel reaches the first activity.""" + + def run(self, ctx: WorkflowContext, name: str): # type: ignore[no-untyped-def] + try: + return (yield from super().run(ctx, name)) + except WorkflowCancelled as error: + assert error.context is not None + return {**self.times, "remaining": error.context.remaining()} + + +@workflow.defn(name="tests.replay.redrive-signal-cleanup") +class RedriveSignalCleanupWorkflow(RedriveSignalWorkflow): + """Observes the cancellation budget if a cancel reaches the first activity.""" + + def run(self, ctx: WorkflowContext, name: str): # type: ignore[no-untyped-def] + try: + return (yield from super().run(ctx, name)) + except WorkflowCancelled as error: + assert error.context is not None + return {"approved_by": self.approved_by, "remaining": error.context.remaining()} + + +REUSED_UPSERT_EVENTS = ("MemoUpserted", "SearchAttributesUpserted") +REUSED_SIGNAL_EVENTS = ("SignalWaitOpened", "SignalReceived", "SignalApplied") + + +def redrive_fixture(name: str) -> dict[str, Any]: + fixture = json.loads((REDRIVE_FIXTURES / f"{name}.json").read_text(encoding="utf-8")) + assert isinstance(fixture, dict) + return fixture + + +def redriven(name: str) -> list[dict[str, Any]]: + history = redrive_fixture(name)["history"] + assert isinstance(history, list) + return history + + +def only_event(history: list[dict[str, Any]], event_type: str) -> dict[str, Any]: + matches = [event for event in history if event["event_type"] == event_type] + assert len(matches) == 1, event_type + return matches[0] + + +def replay_redriven(workflow_cls: type, history: list[dict[str, Any]]) -> list[Any]: + outcome = replay(workflow_cls, history, ["Taylor"], run_id="redriven-run", payload_codec="avro") + return list(outcome.commands) + + +def visibility_times(history: list[dict[str, Any]]) -> dict[str, str]: + times = query_state(RedriveVisibilityWorkflow, history, ["Taylor"], "replay_times", + run_id="redriven-run", payload_codec="avro") + assert isinstance(times, dict) + return times + + +def signal_state(history: list[dict[str, Any]]) -> dict[str, Any]: + state = query_state(RedriveSignalWorkflow, history, ["Taylor"], "replay_state", + run_id="redriven-run", payload_codec="avro") + assert isinstance(state, dict) + return state + + +@pytest.mark.parametrize(("fixture", "event_types", "declared"), [ + ("redrive-reused-upserts-clock", REUSED_UPSERT_EVENTS, + lambda fixture: [step["event_type"] for step in fixture["redrive_reused_upserts"]["reused_steps"]]), + ("redrive-reused-signal-wait-clock", REUSED_SIGNAL_EVENTS, + lambda fixture: fixture["redrive_reused_signal_wait"]["reused_events"]), +]) +def test_redrive_fixture_carries_reuse_provenance_on_each_named_event( + fixture: str, event_types: tuple[str, ...], declared: Any, +) -> None: + document = redrive_fixture(fixture) + history = document["history"] + assert declared(document) == list(event_types) + started = only_event(history, "WorkflowStarted")["payload"] + assert started["recovery_kind"] == "redrive" + assert started["replayed_started_at"] == REDRIVE_REPLAYED_START + for event_type in event_types: + event = only_event(history, event_type) + payload = event["payload"] + assert payload["reused_from_run_id"] == REDRIVE_SOURCE_RUN, event_type + assert payload["reused_recorded_at"] == "2026-11-03T00:00:00+00:00", event_type + # Redrive recorded the event again at 00:05; the source run recorded it at 00:00. + assert event["recorded_at"] == "2026-11-03T00:05:00+00:00", event_type + + +def test_redriven_upserts_replay_at_the_failed_run_start_without_reupserting() -> None: + history = redriven("redrive-reused-upserts-clock") + assert visibility_times(history) == { + "after_memo": REDRIVE_REPLAYED_START, + "after_search_attributes": REDRIVE_REPLAYED_START, + } + # Only the step after the reused upserts is new; neither upsert is issued again. + commands = replay_redriven(RedriveVisibilityWorkflow, history) + assert [(type(command).__name__, getattr(command, "activity_type", None)) for command in commands] == [ + ("ScheduleActivity", "tests.redrive.first"), + ] + assert commands[0].arguments == ["Taylor"] + memo = only_event(history, "MemoUpserted")["payload"] + assert serializer.decode_envelope(memo["entries"]) == {"customer": "Taylor", "stage": "greeting"} + assert serializer.decode_envelope(memo["merged"]) == {"customer": "Taylor", "stage": "greeting"} + attributes = only_event(history, "SearchAttributesUpserted")["payload"] + assert attributes["attributes"] == attributes["merged"] == {"Stage": "greeting"} + + +def test_redriven_memo_value_cannot_change_on_replay() -> None: + history = deepcopy(redriven("redrive-reused-upserts-clock")) + + @workflow.defn(name="tests.replay.redrive-visibility-changed") + class ChangedMemo(RedriveVisibilityWorkflow): + def run(self, ctx: WorkflowContext, name: str): # type: ignore[no-untyped-def] + return (yield from super().run(ctx, "Someone else")) + + with pytest.raises(NonDeterministicReplayError, match="memo entries differ"): + replay_redriven(ChangedMemo, history) + + +def test_redriven_signal_wait_replays_at_the_failed_run_start_and_applies_once() -> None: + history = redriven("redrive-reused-signal-wait-clock") + RedriveSignalWorkflow.applied = [] + assert signal_state(history) == {"approved_by": "Grace", "time_after_signal": REDRIVE_REPLAYED_START} + assert RedriveSignalWorkflow.applied == ["Grace"] + RedriveSignalWorkflow.applied = [] + commands = replay_redriven(RedriveSignalWorkflow, history) + assert RedriveSignalWorkflow.applied == ["Grace"] + assert [(type(command).__name__, getattr(command, "activity_type", None)) for command in commands] == [ + ("ScheduleActivity", "tests.redrive.first"), + ] + assert commands[0].arguments == ["Taylor"] + applied = only_event(history, "SignalApplied")["payload"] + assert serializer.decode(applied["value"], codec="avro") == "Grace" + received = only_event(history, "SignalReceived")["payload"] + assert serializer.decode(received["arguments"], codec="avro") == ["Grace"] + + +def _retimed(history: list[dict[str, Any]], event_type: str, mode: str) -> list[dict[str, Any]]: + changed = deepcopy(history) + event = only_event(changed, event_type) + if mode == "recorded_at": + event["recorded_at"] = "2026-11-03T09:09:09+00:00" + elif mode == "reused_recorded_at": + event["payload"]["reused_recorded_at"] = "2026-11-02T22:22:22+00:00" + else: + del event["payload"]["reused_recorded_at"] + del event["payload"]["reused_from_run_id"] + return changed + + +@pytest.mark.parametrize("mode", ["recorded_at", "reused_recorded_at", "without_reuse_keys"]) +@pytest.mark.parametrize("event_type", REUSED_UPSERT_EVENTS) +def test_reused_upsert_time_is_not_a_replay_clock_source(event_type: str, mode: str) -> None: + history = _retimed(redriven("redrive-reused-upserts-clock"), event_type, mode) + assert visibility_times(history) == { + "after_memo": REDRIVE_REPLAYED_START, + "after_search_attributes": REDRIVE_REPLAYED_START, + } + assert replay_redriven(RedriveVisibilityWorkflow, history) == replay_redriven( + RedriveVisibilityWorkflow, redriven("redrive-reused-upserts-clock"), + ) + + +@pytest.mark.parametrize("mode", ["recorded_at", "reused_recorded_at", "without_reuse_keys"]) +@pytest.mark.parametrize("event_type", REUSED_SIGNAL_EVENTS) +def test_reused_signal_wait_time_is_not_a_replay_clock_source(event_type: str, mode: str) -> None: + history = _retimed(redriven("redrive-reused-signal-wait-clock"), event_type, mode) + assert signal_state(history) == {"approved_by": "Grace", "time_after_signal": REDRIVE_REPLAYED_START} + assert replay_redriven(RedriveSignalWorkflow, history) == replay_redriven( + RedriveSignalWorkflow, redriven("redrive-reused-signal-wait-clock"), + ) + + +# Cancellation controls. Redrive writes WorkflowStarted with the reused prefix +# in one write, so a cancellation is requested and delivered only after the +# prefix, at later times. The cancellation clock starts at the delivery and +# only moves on results consumed after it, so a reused event's time cannot be +# observed through remaining() in such a history. These controls prove that +# retiming each reused event changes neither remaining() nor the commands. +def _retime_both(history: list[dict[str, Any]], event_type: str) -> list[dict[str, Any]]: + changed = deepcopy(history) + event = only_event(changed, event_type) + event["recorded_at"] = "2026-11-03T00:05:01+00:00" + event["payload"]["reused_recorded_at"] = "2026-11-02T23:59:30+00:00" + return changed + + +def replay_cleanup(workflow_cls: type, history: list[dict[str, Any]], fixture: dict[str, Any]) -> list[Any]: + outcome = replay(workflow_cls, history, ["Taylor"], run_id=fixture["workflow"]["run_id"], + workflow_id=fixture["workflow"]["workflow_id"], payload_codec="avro") + return list(outcome.commands) + + +def test_reused_upserts_do_not_move_the_cancellation_budget() -> None: + fixture = redrive_fixture("redrive-reused-upserts-cancellation-clock") + history = fixture["history"] + kinds = [event["event_type"] for event in history] + assert kinds.index("CooperativeCancellationRequested") > max(kinds.index(kind) for kind in REUSED_UPSERT_EVENTS) + expected = [CompleteWorkflow({"after_memo": REDRIVE_REPLAYED_START, + "after_search_attributes": REDRIVE_REPLAYED_START, "remaining": 26.0})] + assert replay_cleanup(RedriveVisibilityCleanupWorkflow, history, fixture) == expected + for event_type in REUSED_UPSERT_EVENTS: + retimed = _retime_both(history, event_type) + assert replay_cleanup(RedriveVisibilityCleanupWorkflow, retimed, fixture) == expected, event_type + + +def _signal_successor_with_cancellation() -> tuple[list[dict[str, Any]], dict[str, Any]]: + upserts = redrive_fixture("redrive-reused-upserts-cancellation-clock") + # The same post-prefix request and delivery, bound to the signal workflow. + instance = upserts["workflow"]["workflow_id"] + tail = json.loads(json.dumps(upserts["history"][-2:]).replace(instance, "regression-corpus-redrive-signal")) + upserts["workflow"]["workflow_id"] = "regression-corpus-redrive-signal" + upserts["workflow"]["type"] = RedriveSignalCleanupWorkflow.__workflow_name__ # type: ignore[attr-defined] + for event in tail: + event["sequence"] += 1 + # The reused signal wait is workflow sequence 1, so the next call is 2. + tail[1]["payload"]["sequence"] = 2 + history = redriven("redrive-reused-signal-wait-clock") + tail + history[0]["payload"]["workflow_type"] = upserts["workflow"]["type"] + return history, upserts + + +def test_reused_signal_wait_events_do_not_move_the_cancellation_budget() -> None: + history, fixture = _signal_successor_with_cancellation() + kinds = [event["event_type"] for event in history] + assert kinds.index("CooperativeCancellationRequested") > max(kinds.index(kind) for kind in REUSED_SIGNAL_EVENTS) + RedriveSignalCleanupWorkflow.applied = [] + baseline = replay_cleanup(RedriveSignalCleanupWorkflow, history, fixture) + assert RedriveSignalCleanupWorkflow.applied == ["Grace"] + # Limitation: the SDK authors no signal waits, so its counterpart's wait is + # not a numbered call and its first activity is call 1. The delivery + # recorded at call 2 is therefore not reached and no remaining() is read. + # Such a history comes only from a Core-authored workflow. + assert [type(command).__name__ for command in baseline] == ["ScheduleActivity"] + for event_type in REUSED_SIGNAL_EVENTS: + retimed = _retime_both(history, event_type) + assert replay_cleanup(RedriveSignalCleanupWorkflow, retimed, fixture) == baseline, event_type + + +# A redrive successor must be one Workflow 2.19 can write: a redrive +# WorkflowStarted bound to its source run and successor identity, then the +# reused prefix in the same write, then the successor's own events at or after +# its start. Redrive resumes at the step after the last one it reused. +REDRIVE_START_IDENTITY = ( + "workflow_type", "workflow_instance_id", "workflow_run_id", "workflow_command_id", "workflow_link_id", + "continued_from_run_id", "replayed_started_at", +) +REDRIVE_START_DEADLINES = ("execution_deadline_at", "run_deadline_at") +REDRIVE_START_KEYS = (*REDRIVE_START_IDENTITY, "recovery_kind", "resume_step_sequence", *REDRIVE_START_DEADLINES) + +# The identity Core writes on each event kind these fixtures use. Every field +# is required and non-blank; an event kind not listed here is refused. +REDRIVE_EVENT_IDENTITY: dict[str, tuple[str, ...]] = { + "MemoUpserted": ("sequence",), + "SearchAttributesUpserted": ("sequence",), + "SignalWaitOpened": ("sequence", "signal_name", "signal_wait_id"), + "SignalReceived": ( + "workflow_command_id", "signal_id", "workflow_instance_id", "workflow_run_id", "signal_name", "signal_wait_id", + ), + "SignalApplied": ("workflow_command_id", "signal_id", "signal_name", "signal_wait_id", "sequence"), + "CooperativeCancellationRequested": ("workflow_command_id", "workflow_instance_id", "workflow_run_id"), + "CooperativeCancellationDelivered": ("workflow_command_id", "workflow_run_id", "sequence"), +} +# The events one reused workflow step writes, in order. SignalReceived belongs +# to its wait and has no step of its own. +REDRIVE_STEP_SHAPES = {("MemoUpserted",), ("SearchAttributesUpserted",), ("SignalWaitOpened", "SignalApplied")} + + +def check_redrive_successor(history: list[dict[str, Any]], workflow: dict[str, Any]) -> None: + """Require a successor Workflow 2.19 can write for this fixture's workflow.""" + + def require(condition: bool, message: str) -> None: + if not condition: + raise AssertionError(message) + + def present(payload: dict[str, Any], key: str) -> bool: + value = payload.get(key) + return type(value) is int and value > 0 if key == "sequence" else isinstance(value, str) and bool(value.strip()) + + require(bool(history) and history[0].get("event_type") == "WorkflowStarted", + "history must open with WorkflowStarted") + started = history[0] + require(started.get("sequence") == 1, "history order") + start = started["payload"] + for key in REDRIVE_START_KEYS: + require(key in start, f"WorkflowStarted lacks {key}") + require(start["recovery_kind"] == "redrive", "recovery_kind") + for key in REDRIVE_START_IDENTITY: + require(isinstance(start[key], str) and bool(start[key].strip()), key) + for key in REDRIVE_START_DEADLINES: + deadline = start[key] + if deadline is not None: + require(isinstance(deadline, str) and datetime.fromisoformat(deadline).tzinfo is not None, key) + # The start binds the fixture's workflow type, instance and successor run. + require(start["workflow_type"] == workflow.get("type"), "workflow_type must be the fixture workflow") + require(start["workflow_instance_id"] == workflow.get("workflow_id"), "workflow_instance_id must be the fixture's") + require(start["workflow_run_id"] == workflow.get("run_id"), "workflow_run_id must be the fixture's") + instance, run, source = start["workflow_instance_id"], start["workflow_run_id"], start["continued_from_run_id"] + require(run != source, "the successor is a new run") + started_at = datetime.fromisoformat(started["recorded_at"]) + require(datetime.fromisoformat(start["replayed_started_at"]) <= started_at, "replayed_started_at") + + prefix = True + steps: list[tuple[int, list[str]]] = [] + signals: dict[str, dict[str, Any]] = {} + cancellations: dict[str, dict[str, Any]] = {} + for index, event in enumerate(history[1:], start=2): + require(event.get("sequence") == index, "history order") + kind = event.get("event_type") + require(kind in REDRIVE_EVENT_IDENTITY, f"unexpected event {kind}") + payload = event["payload"] + for key in REDRIVE_EVENT_IDENTITY[kind]: + require(present(payload, key), f"{kind} lacks {key}") + if kind == "SignalReceived": + require("sequence" not in payload, "SignalReceived has no step sequence") + if "workflow_instance_id" in payload: + require(payload["workflow_instance_id"] == instance, "event belongs to another workflow instance") + reused = "reused_from_run_id" in payload or "reused_recorded_at" in payload + if reused: + require(prefix, "reused events must directly follow WorkflowStarted") + for key in ("reused_from_run_id", "reused_recorded_at"): + require(present(payload, key), f"reused event lacks {key}") + require(payload["reused_from_run_id"] == source, "reused_from_run_id must match continued_from_run_id") + # A reused event keeps the run it was first recorded in. + if "workflow_run_id" in payload: + require(payload["workflow_run_id"] == source, "reused event belongs to another run") + require(event["recorded_at"] == started["recorded_at"], "reused events are recorded with the start") + require(datetime.fromisoformat(payload["reused_recorded_at"]) <= started_at, "reused_recorded_at") + if kind != "SignalReceived": + if steps and steps[-1][0] == payload["sequence"]: + steps[-1][1].append(kind) + else: + steps.append((payload["sequence"], [kind])) + else: + prefix = False + if "workflow_run_id" in payload: + require(payload["workflow_run_id"] == run, "successor event belongs to another run") + require(datetime.fromisoformat(event["recorded_at"]) >= started_at, + "successor event predates the successor start") + if kind.startswith("Signal"): + if kind != "SignalWaitOpened": + require(payload["signal_wait_id"] in signals, "signal event has no opened wait") + wait = signals.setdefault(payload["signal_wait_id"], {}) + for key in ("signal_name", "workflow_command_id", "signal_id"): + if key in payload: + require(wait.setdefault(key, payload[key]) == payload[key], f"signal events disagree on {key}") + if kind.startswith("CooperativeCancellation"): + command = cancellations.setdefault(kind, payload["workflow_command_id"]) + require(command == payload["workflow_command_id"], "duplicate cancellation event") + if cancellations: + require(len(set(cancellations.values())) == 1, "cancellation events disagree on workflow_command_id") + # The reused steps are the contiguous prefix 1..resume-1, in order, one + # whole step each. + require(bool(steps), "no reused workflow step") + require([sequence for sequence, _ in steps] == list(range(1, len(steps) + 1)), + "reused steps must be the contiguous prefix from 1") + for sequence, kinds in steps: + require(tuple(kinds) in REDRIVE_STEP_SHAPES, f"reused step {sequence} is not one whole step") + resume = start.get("resume_step_sequence") + require(type(resume) is int and resume == len(steps) + 1, + "resume_step_sequence must follow the last reused step") + + +REDRIVE_SUCCESSOR_FIXTURES = { + "redrive-reused-search-attribute-update-missing": 3, + "redrive-reused-upserts-clock": 3, + "redrive-reused-signal-wait-clock": 2, + "redrive-reused-upserts-cancellation-clock": 3, + "redrive-reused-search-attribute-value-mismatch": 3, + "redrive-reused-search-attribute-type-mismatch": 3, +} + + +def test_every_reused_fixture_is_a_checked_redrive_successor() -> None: + reused = sorted( + path.stem for path in REDRIVE_FIXTURES.glob("*.json") + if "reused_from_run_id" in path.read_text(encoding="utf-8") + ) + assert reused == sorted(REDRIVE_SUCCESSOR_FIXTURES) + + +@pytest.mark.parametrize(("fixture", "resume"), REDRIVE_SUCCESSOR_FIXTURES.items()) +def test_redrive_fixture_is_a_valid_successor(fixture: str, resume: int) -> None: + document = redrive_fixture(fixture) + assert document["history"][0]["payload"]["resume_step_sequence"] == resume + check_redrive_successor(document["history"], document["workflow"]) + + +def test_signal_successor_with_cancellation_is_a_valid_successor() -> None: + history, fixture = _signal_successor_with_cancellation() + check_redrive_successor(history, fixture["workflow"]) + + +def _broken(fixture: str, change: Any) -> list[dict[str, Any]]: + history = deepcopy(redriven(fixture)) + change(history) + return history + + +def _set_start(key: str, value: Any) -> Any: + return lambda history: history[0]["payload"].__setitem__(key, value) + + +def _drop_start(key: str) -> Any: + return lambda history: history[0]["payload"].pop(key) + + +@pytest.mark.parametrize("fixture", ["redrive-reused-upserts-clock", "redrive-reused-signal-wait-clock"]) +@pytest.mark.parametrize(("change", "message"), [ + (lambda resume: resume - 1, "resume_step_sequence"), + (lambda resume: resume + 1, "resume_step_sequence"), + (lambda resume: None, "resume_step_sequence"), + (lambda resume: str(resume), "resume_step_sequence"), +]) +def test_successor_check_rejects_a_wrong_resume_step(fixture: str, change: Any, message: str) -> None: + resume = REDRIVE_SUCCESSOR_FIXTURES[fixture] + with pytest.raises(AssertionError, match=message): + check_redrive_successor(_broken(fixture, _set_start("resume_step_sequence", change(resume))), + redrive_fixture(fixture)["workflow"]) + + +def _last_reused(history: list[dict[str, Any]]) -> dict[str, Any]: + return [event for event in history if "reused_from_run_id" in event["payload"]][-1] + + +@pytest.mark.parametrize(("change", "message"), [ + (_set_start("recovery_kind", "continue_as_new"), "recovery_kind"), + *[(_drop_start(key), key) for key in REDRIVE_START_IDENTITY], + *[(_set_start(key, ""), key) for key in REDRIVE_START_IDENTITY], + (_set_start("continued_from_run_id", "01JBREDRIVESUCCESSORRUN0000"), "new run"), + (lambda history: _last_reused(history)["payload"].__setitem__("reused_from_run_id", "other-run"), + "continued_from_run_id"), + (lambda history: _last_reused(history)["payload"].pop("reused_recorded_at"), "lacks reused_recorded_at"), + (lambda history: _last_reused(history)["payload"].__setitem__("reused_recorded_at", ""), + "lacks reused_recorded_at"), + (lambda history: _last_reused(history)["payload"].__setitem__("reused_from_run_id", None), + "lacks reused_from_run_id"), + (_set_start("workflow_type", "tests.replay.other"), "fixture workflow"), + (_set_start("workflow_instance_id", "other-instance"), "workflow_instance_id must be the fixture"), + (_set_start("workflow_run_id", "other-successor-run"), "workflow_run_id must be the fixture"), + (_set_start("workflow_type", " "), "workflow_type"), + (lambda history: _last_reused(history).__setitem__("recorded_at", "2026-11-03T00:05:01+00:00"), + "recorded with the start"), + (lambda history: [history[1]["payload"].pop(key) for key in ("reused_from_run_id", "reused_recorded_at")], + "directly follow"), + (lambda history: history.append({"sequence": len(history) + 1, "event_type": "ActivityScheduled", + "recorded_at": "2026-11-03T00:05:01+00:00", "payload": {"sequence": 3}}), + "unexpected event"), + (lambda history: history[1].__setitem__("sequence", 9), "history order"), +]) +def test_successor_check_rejects_an_invalid_envelope_or_order(change: Any, message: str) -> None: + with pytest.raises(AssertionError, match=message): + check_redrive_successor(_broken("redrive-reused-upserts-clock", change), + redrive_fixture("redrive-reused-upserts-clock")["workflow"]) + + +def _event(history: list[dict[str, Any]], event_type: str) -> dict[str, Any]: + return next(event for event in history if event["event_type"] == event_type) + + +@pytest.mark.parametrize(("fixture", "change", "message"), [ + ("redrive-reused-signal-wait-clock", + lambda history: _event(history, "SignalReceived")["payload"].__setitem__("workflow_run_id", "other-run"), + "reused event belongs to another run"), + ("redrive-reused-signal-wait-clock", + lambda history: _event(history, "SignalReceived")["payload"].__setitem__("workflow_instance_id", "other"), + "another workflow instance"), + ("redrive-reused-signal-wait-clock", + lambda history: _event(history, "SignalApplied")["payload"].pop("reused_from_run_id"), + "lacks reused_from_run_id"), + ("redrive-reused-signal-wait-clock", + lambda history: _event(history, "SignalWaitOpened")["payload"].pop("reused_recorded_at"), + "lacks reused_recorded_at"), + ("redrive-reused-upserts-cancellation-clock", + lambda history: _event(history, "CooperativeCancellationRequested")["payload"].__setitem__( + "workflow_run_id", REDRIVE_SOURCE_RUN), + "successor event belongs to another run"), + ("redrive-reused-upserts-cancellation-clock", + lambda history: _event(history, "CooperativeCancellationRequested")["payload"].__setitem__( + "workflow_instance_id", "other"), + "another workflow instance"), + ("redrive-reused-upserts-cancellation-clock", + lambda history: _event(history, "CooperativeCancellationRequested").__setitem__( + "recorded_at", "2026-11-03T00:04:59+00:00"), + "predates"), + ("redrive-reused-upserts-cancellation-clock", + lambda history: _event(history, "CooperativeCancellationDelivered")["payload"].__setitem__( + "workflow_command_id", "other-cancel"), + "disagree on workflow_command_id"), + ("redrive-reused-signal-wait-clock", + lambda history: _event(history, "SignalApplied")["payload"].__setitem__("signal_wait_id", "other-wait"), + "no opened wait"), + ("redrive-reused-signal-wait-clock", + lambda history: _event(history, "SignalApplied")["payload"].__setitem__("signal_id", "other-signal"), + "disagree on signal_id"), +]) +def test_successor_check_binds_every_event_to_the_run_and_instance(fixture: str, change: Any, message: str) -> None: + check_redrive_successor(redriven(fixture), redrive_fixture(fixture)["workflow"]) + with pytest.raises(AssertionError, match=message): + check_redrive_successor(_broken(fixture, change), redrive_fixture(fixture)["workflow"]) + + +def _check_fixture(fixture: str, history: list[dict[str, Any]]) -> None: + check_redrive_successor(history, redrive_fixture(fixture)["workflow"]) + + +@pytest.mark.parametrize("fixture", REDRIVE_SUCCESSOR_FIXTURES) +@pytest.mark.parametrize("key", REDRIVE_START_KEYS) +def test_successor_check_requires_every_envelope_field(fixture: str, key: str) -> None: + with pytest.raises(AssertionError, match=f"lacks {key}"): + _check_fixture(fixture, _broken(fixture, _drop_start(key))) + + +@pytest.mark.parametrize("key", REDRIVE_START_DEADLINES) +@pytest.mark.parametrize("value", ["", "not-a-time", "2026-11-03T00:00:00", 5, True, []]) +def test_successor_check_requires_a_null_or_valid_deadline(key: str, value: Any) -> None: + fixture = "redrive-reused-upserts-clock" + with pytest.raises((AssertionError, ValueError, TypeError)): + _check_fixture(fixture, _broken(fixture, _set_start(key, value))) + _check_fixture(fixture, _broken(fixture, _set_start(key, "2026-11-04T00:00:00+00:00"))) + + +def _event_identity_cases() -> list[tuple[str, int, str]]: + cases = [] + for fixture in REDRIVE_SUCCESSOR_FIXTURES: + for index, event in enumerate(redriven(fixture)[1:], start=1): + for key in REDRIVE_EVENT_IDENTITY[event["event_type"]]: + cases.append((fixture, index, key)) + return cases + + +EVENT_IDENTITY_CASES = _event_identity_cases() + + +@pytest.mark.parametrize(("fixture", "index", "key"), EVENT_IDENTITY_CASES) +@pytest.mark.parametrize("mode", ["deleted", "blank"]) +def test_successor_check_requires_every_event_identity_field(fixture: str, index: int, key: str, mode: str) -> None: + def change(history: list[dict[str, Any]]) -> None: + payload = history[index]["payload"] + if mode == "deleted": + del payload[key] + else: + payload[key] = 0 if key == "sequence" else " " + + with pytest.raises(AssertionError, match=f"lacks {key}"): + _check_fixture(fixture, _broken(fixture, change)) + + +@pytest.mark.parametrize("fixture", REDRIVE_SUCCESSOR_FIXTURES) +@pytest.mark.parametrize("field", ["sequence", "event_type"]) +def test_successor_check_requires_every_event_id(fixture: str, field: str) -> None: + for index in range(1, len(redriven(fixture))): + with pytest.raises(AssertionError, match="history order|unexpected event"): + _check_fixture(fixture, _broken(fixture, lambda history, i=index: history[i].pop(field))) + + +def _upsert(history: list[dict[str, Any]], event_type: str) -> dict[str, Any]: + return _event(history, event_type)["payload"] + + +@pytest.mark.parametrize(("fixture", "change", "message"), [ + # Missing step: the reused prefix skips step 2. + ("redrive-reused-upserts-clock", lambda history: ( + _upsert(history, "SearchAttributesUpserted").__setitem__("sequence", 3), + history[0]["payload"].__setitem__("resume_step_sequence", 4)), "contiguous prefix"), + # Prefix does not start at 1. + ("redrive-reused-upserts-clock", lambda history: ( + _upsert(history, "MemoUpserted").__setitem__("sequence", 2), + _upsert(history, "SearchAttributesUpserted").__setitem__("sequence", 3), + history[0]["payload"].__setitem__("resume_step_sequence", 4)), "contiguous prefix"), + # Duplicate step: both upserts claim step 1. + ("redrive-reused-upserts-clock", lambda history: ( + _upsert(history, "SearchAttributesUpserted").__setitem__("sequence", 1), + history[0]["payload"].__setitem__("resume_step_sequence", 2)), "one whole step"), + # Duplicate event: the memo step written twice. + ("redrive-reused-upserts-clock", lambda history: ( + history.insert(2, {**deepcopy(history[1]), "sequence": 3}), + history[3].__setitem__("sequence", 4)), "one whole step"), + # Descending: the steps appear in reverse order. + ("redrive-reused-upserts-clock", lambda history: ( + _upsert(history, "MemoUpserted").__setitem__("sequence", 2), + _upsert(history, "SearchAttributesUpserted").__setitem__("sequence", 1)), "contiguous prefix"), + # SignalReceived given a step of its own. + ("redrive-reused-signal-wait-clock", + lambda history: _event(history, "SignalReceived")["payload"].__setitem__("sequence", 1), "no step sequence"), + ("redrive-reused-signal-wait-clock", lambda history: ( + _event(history, "SignalReceived")["payload"].__setitem__("sequence", 2), + history[0]["payload"].__setitem__("resume_step_sequence", 3)), "no step sequence"), + # A signal wait split across two steps. + ("redrive-reused-signal-wait-clock", lambda history: ( + _event(history, "SignalApplied")["payload"].__setitem__("sequence", 2), + history[0]["payload"].__setitem__("resume_step_sequence", 3)), "one whole step"), +]) +def test_successor_check_requires_contiguous_whole_reused_steps(fixture: str, change: Any, message: str) -> None: + with pytest.raises(AssertionError, match=message): + _check_fixture(fixture, _broken(fixture, change)) diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index 3592c5f2..c455ab69 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -43,7 +43,13 @@ ) from tests.test_pending_operation_replay import PendingOperation from tests.test_poll_failure_backoff import RecoveredWorkflow -from tests.test_replay_clock import PersistedClockWorkflow +from tests.test_replay_clock import ( + PersistedClockWorkflow, + RedriveChangedStageWorkflow, + RedriveSignalWorkflow, + RedriveVisibilityCleanupWorkflow, + RedriveVisibilityWorkflow, +) from tests.test_update_signal_condition_replay import ( PostConditionReceiversWorkflow, UpdateSignalConditionTimerWorkflow, @@ -357,6 +363,10 @@ def run(self, ctx: WorkflowContext, message: str) -> None: IdentityAcrossReplay, IdentityContinueAsNew, PersistedClockWorkflow, + RedriveChangedStageWorkflow, + RedriveSignalWorkflow, + RedriveVisibilityCleanupWorkflow, + RedriveVisibilityWorkflow, NexusCooperativeDeliveryWorkflow, NexusTerminalDeliveryWorkflow, NexusTargetWorkflow, diff --git a/tests/test_search_attribute_replay_identity.py b/tests/test_search_attribute_replay_identity.py new file mode 100644 index 00000000..46ce7ab3 --- /dev/null +++ b/tests/test_search_attribute_replay_identity.py @@ -0,0 +1,675 @@ +"""Replay identity of SearchAttributesUpserted (Workflow protocol 1.20). + +Values follow Core's UpsertSearchAttributesCall exactly and fail closed. +Types follow Core's SearchAttributeReplayIdentity where the SDK can know +them; see test_recorded_type_identity_is_limited_to_what_an_untyped_upsert_fixes. +""" + +from __future__ import annotations + +import math +from copy import deepcopy +from typing import Any + +import pytest + +from durable_workflow import workflow +from durable_workflow.errors import NonDeterministicReplayError +from durable_workflow.workflow import WorkflowContext, replay +from tests.test_replay_clock import ( + RedriveChangedStageWorkflow, + RedriveVisibilityWorkflow, + only_event, + redrive_fixture, + redriven, +) + +PLAIN = "search-attribute-value-mismatch" +REUSED = "redrive-reused-search-attribute-value-mismatch" + + +def authoring(attributes: Any) -> type: + @workflow.defn(name="tests.replay.search-attribute-authored") + class Authored(RedriveVisibilityWorkflow): + def run(self, ctx: WorkflowContext, name: str): # type: ignore[no-untyped-def] + yield ctx.upsert_memo({"stage": "greeting", "customer": name}) + yield ctx.upsert_search_attributes(attributes) + return (yield ctx.schedule_activity("tests.redrive.first", [name])) + + return Authored + + +def recorded(fixture: str, attributes: Any, attribute_types: Any = ...) -> list[dict[str, Any]]: + history = deepcopy(redriven(fixture)) + payload = only_event(history, "SearchAttributesUpserted")["payload"] + payload["attributes"] = attributes + if attribute_types is ...: + payload.pop("attribute_types", None) + else: + payload["attribute_types"] = attribute_types + return history + + +def replays(workflow_cls: type, history: list[dict[str, Any]], **options: Any) -> list[Any]: + return list(replay( + workflow_cls, history, ["Taylor"], run_id="search-attribute-run", payload_codec="avro", **options, + ).commands) + + +def refused(workflow_cls: type, history: list[dict[str, Any]], reason: str) -> None: + with pytest.raises(NonDeterministicReplayError, match=reason) as error: + replays(workflow_cls, history) + assert error.value.workflow_sequence == 2 + + +@pytest.fixture(params=[PLAIN, REUSED]) +def fixture(request: pytest.FixtureRequest) -> str: + return str(request.param) + + +def test_recorded_values_and_types_replay(fixture: str) -> None: + payload = only_event(redriven(fixture), "SearchAttributesUpserted")["payload"] + assert payload["attributes"] == {"Stage": "greeting"} + assert payload["attribute_types"] == {"Stage": "keyword"} + assert [type(command).__name__ for command in replays(RedriveVisibilityWorkflow, redriven(fixture))] == [ + "ScheduleActivity", + ] + + +def test_changed_value_is_refused(fixture: str) -> None: + refused(RedriveChangedStageWorkflow, redriven(fixture), "search_attribute_value_mismatch") + + +@pytest.mark.parametrize("authored", [ + {"Stage": "approval"}, {"Stage": None}, {"Stage": ""}, {"Stage": ["greeting"]}, {"stage": "greeting"}, + {"Stage": "greeting", "Region": "eu"}, +]) +def test_value_identity_is_exact(fixture: str, authored: dict[str, Any]) -> None: + refused(authoring(authored), redriven(fixture), "search_attribute_value_mismatch") + + +@pytest.mark.parametrize(("authored", "stored", "types"), [ + ({"Stage": " greeting\n"}, {"Stage": "greeting"}, {"Stage": "string"}), + ({"Stage": "greeting", "Old": " "}, {"Old": None, "Stage": "greeting"}, {"Old": "string", "Stage": "keyword"}), + ({"Stage": "greeting", "Old": ""}, {"Old": None, "Stage": "greeting"}, {"Old": "keyword", "Stage": "keyword"}), + ({"Tags": [" a ", "b"]}, {"Tags": ["a", "b"]}, {"Tags": "keyword_list"}), + ({"Tags": []}, {"Tags": []}, {"Tags": "keyword_list"}), + ({"Count": 3}, {"Count": 3}, {"Count": "int"}), + ({"Score": 2.5}, {"Score": 2.5}, {"Score": "float"}), + ({"Done": False, "Gone": None}, {"Done": False, "Gone": None}, {"Done": "bool"}), +]) +def test_server_normalized_history_replays( + fixture: str, authored: dict[str, Any], stored: dict[str, Any], types: dict[str, str], +) -> None: + # The Server trims strings and list items, stores a blank string as null, + # and types every non-null raw value, including a blank one. + commands = replays(authoring(authored), recorded(fixture, stored, types)) + assert [type(command).__name__ for command in commands] == ["ScheduleActivity"] + + +@pytest.mark.parametrize(("authored", "stored"), [ + ({"Count": 1}, {"Count": 1.0}), + ({"Count": 1}, {"Count": True}), + ({"Count": 1}, {"Count": "1"}), + ({"Score": 1.0}, {"Score": 1}), + ({"Done": True}, {"Done": 1}), +]) +def test_values_are_compared_with_their_types(fixture: str, authored: dict[str, Any], stored: dict[str, Any]) -> None: + refused(authoring(authored), recorded(fixture, stored), "search_attribute_value_mismatch") + + +class Greeting: + """An unsupported object whose repr equals the recorded string.""" + + def __repr__(self) -> str: + return "greeting" + + +@pytest.mark.parametrize("authored", [ + {"Stage": Greeting()}, # Was accepted through json default=repr. + {"Stage": {"greeting"}}, + {"Stage": ("greeting", 1)}, + {"Stage": {"value": "greeting"}}, + {"Stage": ["greeting", 1]}, + {"Stage": b"greeting"}, + {"Stage": math.nan}, + {"Stage": math.inf}, + {"Stage": 2**63}, + {1: "greeting"}, + {"bad key": "greeting"}, + {"Stage": "x" * 2049}, + {"Stage": ["x" * 256]}, + {}, +]) +def test_unsupported_authored_values_fail_closed(fixture: str, authored: Any) -> None: + refused(authoring(authored), redriven(fixture), "search_attribute_value_mismatch: the current search attribute " + "update is invalid") + + +def test_unsupported_authored_value_with_a_colliding_recorded_form_fails_closed(fixture: str) -> None: + # Each authored value below collapsed onto its recorded form before. + for authored, stored in [ + ({"Stage": Greeting()}, {"Stage": "greeting"}), + ({"Tags": ("a", 1)}, {"Tags": ["a", "1"]}), + ({7: "greeting"}, {"7": "greeting"}), + ]: + refused(authoring(authored), recorded(fixture, stored), "invalid") + + +@pytest.mark.parametrize("stored", [ + ["Stage", "greeting"], {}, "greeting", {"Stage": {"value": "greeting"}}, {"Stage": ["greeting", 1]}, + {"Stage": math.nan}, {"bad key": "greeting"}, +]) +def test_malformed_recorded_values_are_refused(fixture: str, stored: Any) -> None: + refused(RedriveVisibilityWorkflow, recorded(fixture, stored), "search_attribute_value_mismatch") + + +@pytest.mark.parametrize("types", [ + {"Stage": "int"}, {"Stage": "float"}, {"Stage": "bool"}, {"Stage": "keyword_list"}, + {"Stage": "text"}, {"Stage": None}, {}, [], {"Stage": "keyword", "Other": "keyword"}, {"stage": "keyword"}, + "keyword", ["keyword"], +]) +def test_recorded_type_mismatch_is_refused(fixture: str, types: Any) -> None: + history = recorded(fixture, {"Stage": "greeting"}, types) + refused(RedriveVisibilityWorkflow, history, "search_attribute_type_mismatch") + + +@pytest.mark.parametrize(("authored", "types"), [ + ({"Count": 3}, {"Count": "keyword"}), + ({"Count": 3}, {"Count": "bool"}), + ({"Score": 2.5}, {"Score": "int"}), + ({"Done": True}, {"Done": "int"}), + ({"Tags": ["a"]}, {"Tags": "keyword"}), + ({"Gone": None}, {"Gone": "keyword"}), + ({"Old": ""}, {}), + ({"Old": " ", "Stage": "greeting"}, {"Stage": "keyword"}), +]) +def test_recorded_type_must_hold_the_value(fixture: str, authored: dict[str, Any], types: dict[str, str]) -> None: + refused(authoring(authored), recorded(fixture, authored, types), "search_attribute_type_mismatch") + + +@pytest.mark.parametrize(("authored", "types"), [ + ({"Stage": "greeting"}, {"Stage": "string"}), + ({"At": "2026-11-03T00:00:00Z"}, {"At": "datetime"}), + ({"At": "2026-11-03"}, {"At": "datetime"}), + ({"At": "Tue, 03 Nov 2026 00:00:00 +0000"}, {"At": "datetime"}), + ({"Note": "two words"}, {"Note": "keyword"}), + ({"At": "2026-11-03T00:00:00"}, {"At": "datetime"}), + # Conservative: any string may be a registered datetime. The Server + # refuses one that does not parse, so such a history cannot be written. + ({"At": "greeting"}, {"At": "datetime"}), + ({"Old": ""}, {"Old": "datetime"}), + ({"Count": 3}, {"Count": "float"}), +]) +def test_recorded_type_identity_is_limited_to_what_an_untyped_upsert_fixes( + fixture: str, authored: dict[str, Any], types: dict[str, str], +) -> None: + # The SDK's upsert declares no types, but the Server records a registered + # key's namespace type in place of the inferred one: PHP Server + # SearchAttributeValueValidator.php:93 ($registeredType ?? $declaredType) + # filled in at WorkerController.php:2734, and Rust Server + # search_attribute_upserts.rs:254-264. Replay cannot see the registry, so + # these recorded types are legitimate for the same value and must replay. + history = recorded(fixture, authored, types) + assert replays(authoring(authored), history) == replays(authoring(authored), recorded(fixture, authored)) + + +def test_history_without_types_keeps_value_only_identity(fixture: str) -> None: + # Events recorded before attribute_types existed, as in Core. + history = recorded(fixture, {"Stage": "greeting"}) + assert replays(RedriveVisibilityWorkflow, history) == replays(RedriveVisibilityWorkflow, redriven(fixture)) + refused(RedriveChangedStageWorkflow, history, "search_attribute_value_mismatch") + + +@pytest.mark.parametrize("authored", [("a", " b "), ["a", "b"]]) +def test_authored_string_tuple_replays_as_its_wire_list(fixture: str, authored: Any) -> None: + # A tuple of strings is sent as the same JSON list the Server records. + history = recorded(fixture, {"Tags": ["a", "b"]}, {"Tags": "keyword_list"}) + commands = replays(authoring({"Tags": authored}), history) + assert [type(command).__name__ for command in commands] == ["ScheduleActivity"] + + +@pytest.mark.parametrize("authored", [("a", 1), ("a", None), ("a", ("b",)), ("a", "c")]) +def test_authored_tuple_keeps_the_strict_list_rules(fixture: str, authored: Any) -> None: + history = recorded(fixture, {"Tags": ["a", "b"]}, {"Tags": "keyword_list"}) + refused(authoring({"Tags": authored}), history, "search_attribute_value_mismatch") + + +def test_recorded_history_never_treats_a_non_list_as_a_list(fixture: str) -> None: + # Only the authored side is normalized; recorded JSON has no tuples. + from durable_workflow.workflow import _canonical_search_attributes + + with pytest.raises(ValueError): + _canonical_search_attributes({"Tags": ("a", "b")}) + assert _canonical_search_attributes({"Tags": ("a", "b")}, authored=True) == {"Tags": ("list", ("a", "b"))} + + +@pytest.mark.parametrize("types", [ + {"Stage": []}, {"Stage": {}}, {"Stage": ["keyword"]}, {"Stage": {"type": "keyword"}}, {"Stage": 1}, + {"Stage": True}, {"Stage": None}, {"Stage": 1.5}, +]) +def test_malformed_recorded_type_values_are_refused_not_raised(fixture: str, types: Any) -> None: + history = recorded(fixture, {"Stage": "greeting"}, types) + refused(RedriveVisibilityWorkflow, history, "search_attribute_type_mismatch") + + +@pytest.mark.parametrize("workflow_cls", [RedriveVisibilityWorkflow, RedriveChangedStageWorkflow]) +def test_missing_recorded_attributes_are_refused(fixture: str, workflow_cls: type) -> None: + history = deepcopy(redriven(fixture)) + del only_event(history, "SearchAttributesUpserted")["payload"]["attributes"] + refused(workflow_cls, history, "search_attribute_update_missing: SearchAttributesUpserted history is missing " + "its attributes") + + +# Durable command sequences. A recorded upsert must carry its own sequence, +# a sequence records at most one upsert, and two durable command kinds never +# share a sequence (only a wait and its own timeout timer do). +def refused_at( + workflow_cls: type, history: list[dict[str, Any]], reason: str, sequence: int, **options: Any, +) -> None: + with pytest.raises(NonDeterministicReplayError, match=f"{reason}:") as error: + replays(workflow_cls, history, **options) + assert error.value.workflow_sequence == sequence + + +def _with(fixture: str, change: Any) -> list[dict[str, Any]]: + history = deepcopy(redriven(fixture)) + change(history) + return history + + +def _append(history: list[dict[str, Any]], event_type: str, payload: dict[str, Any]) -> None: + history.append({"sequence": len(history) + 1, "event_type": event_type, + "recorded_at": "2026-11-03T00:05:01+00:00", "payload": payload}) + + +@pytest.mark.parametrize(("value", "reason"), [ + (..., "durable_command_sequence_missing"), + (None, "durable_command_sequence_missing"), + *[(value, "durable_command_sequence_invalid") for value in (0, -1, "", "two", "-1", "2", True, 1.0, [1])], +]) +def test_search_attribute_update_without_a_usable_sequence_is_refused(fixture: str, value: Any, reason: str) -> None: + def change(history: list[dict[str, Any]]) -> None: + payload = only_event(history, "SearchAttributesUpserted")["payload"] + if value is ...: + del payload["sequence"] + else: + payload["sequence"] = value + + refused_at(RedriveVisibilityWorkflow, _with(fixture, change), reason, 2) + + +def test_duplicate_search_attribute_update_is_refused(fixture: str) -> None: + def change(history: list[dict[str, Any]]) -> None: + duplicate = deepcopy(only_event(history, "SearchAttributesUpserted")) + _append(history, "SearchAttributesUpserted", duplicate["payload"]) + + refused_at(RedriveVisibilityWorkflow, _with(fixture, change), "duplicate_search_attribute_update", 2) + + +@pytest.mark.parametrize(("event_type", "payload", "sequence"), [ + ("ActivityScheduled", {"sequence": 2, "activity_type": "tests.redrive.first"}, 2), + ("ActivityCompleted", {"sequence": 1, "activity_type": "tests.redrive.first"}, 1), + ("TimerScheduled", {"sequence": 2, "delay_seconds": 5}, 2), + ("SideEffectRecorded", {"sequence": 1}, 1), + ("VersionMarkerRecorded", {"sequence": 2, "change_id": "change", "version": 1}, 2), + ("ConditionWaitOpened", {"sequence": 2, "condition_wait_id": "wait"}, 2), + ("SignalWaitOpened", {"sequence": 1, "signal_name": "approve", "signal_wait_id": "wait"}, 1), + ("MemoUpserted", {"sequence": 2}, 2), +]) +def test_two_durable_commands_at_one_sequence_are_refused( + fixture: str, event_type: str, payload: dict[str, Any], sequence: int, +) -> None: + history = _with(fixture, lambda history: _append(history, event_type, payload)) + refused_at(RedriveVisibilityWorkflow, history, "durable_command_sequence_collision", sequence) + + +def test_a_search_attribute_update_cannot_take_an_earlier_step(fixture: str) -> None: + # The upsert moved onto the memo's step 1: a collision, not an overwrite. + history = _with(fixture, lambda history: only_event(history, "SearchAttributesUpserted")["payload"].__setitem__( + "sequence", 1)) + refused_at(RedriveVisibilityWorkflow, history, "durable_command_sequence_collision", 1) + + +def test_collisions_are_refused_when_they_involve_an_upsert() -> None: + from durable_workflow.workflow import _DURABLE_COMMAND_KINDS, _durable_command_kinds_compatible + + kinds = set(_DURABLE_COMMAND_KINDS.values()) + for upsert in ("memo", "search_attributes"): + assert _durable_command_kinds_compatible(frozenset({upsert})) + for other in kinds - {upsert}: + assert not _durable_command_kinds_compatible(frozenset({upsert, other})), other + # Lifecycle events of one command, and pairs without an upsert, are not + # judged by this rule (sdk-php#246 has the same scope). + for kind in kinds - {"memo", "search_attributes"}: + assert _durable_command_kinds_compatible(frozenset({kind})) + assert _durable_command_kinds_compatible(frozenset({"condition", "timer"})) + lifecycle = {kind: sorted(event for event, owner in _DURABLE_COMMAND_KINDS.items() if owner == kind) + for kind in ("activity", "timer", "condition", "signal", "child", "service_call")} + assert lifecycle["signal"] == [ + "SignalApplied", "SignalWaitCancelled", "SignalWaitOpened", "SignalWaitReceived", "SignalWaitTimedOut", + ] + assert {"ActivityHeartbeatRecorded", "ActivityRetryScheduled"} <= set(lifecycle["activity"]) + assert _DURABLE_COMMAND_KINDS["CancellationScopeOpened"] == "cancellation_scope" + assert "SignalReceived" not in _DURABLE_COMMAND_KINDS + assert _DURABLE_COMMAND_KINDS["CooperativeCancellationDelivered"] == "cancellation_delivery" + assert "CooperativeCancellationRequested" not in _DURABLE_COMMAND_KINDS + + +# Every event that claims a workflow sequence, by command kind. Fixed here so +# the table cannot shrink unnoticed; each kind, with its ordinary opener +# absent, collides with an upsert whichever of the two is recorded first. +EXPECTED_CLAIMS = { + **dict.fromkeys(("ActivityScheduled", "ActivityStarted", "ActivityHeartbeatRecorded", "ActivityRetryScheduled", + "ActivityCompleted", "ActivityFailed", "ActivityTimedOut", "ActivityCancelled", + "ActivityCancellationAcknowledged"), "activity"), + **dict.fromkeys(("TimerScheduled", "TimerFired", "TimerCancelled"), "timer"), + **dict.fromkeys(("ChildWorkflowScheduled", "ChildRunStarted", "ChildRunCompleted", "ChildRunFailed", + "ChildRunCancelled", "ChildRunTerminated", "ChildCancellationRequested", + "ChildCancellationResolved"), "child"), + **dict.fromkeys(("ConditionWaitOpened", "ConditionWaitSatisfied", "ConditionWaitTimedOut", + "ConditionWaitCancelled"), "condition"), + **dict.fromkeys(("SignalWaitOpened", "SignalApplied", "SignalWaitReceived", "SignalWaitTimedOut", + "SignalWaitCancelled"), "signal"), + **dict.fromkeys(("ServiceOperationScheduled", "ServiceCallStarted", "ServiceCallCompleted", "ServiceCallFailed", + "ServiceCallCancelled"), "service_call"), + "CancellationScopeOpened": "cancellation_scope", + **dict.fromkeys(("CooperativeCancellationDelivered", "CancellationScopeDelivered"), "cancellation_delivery"), + "WorkflowContinuedAsNew": "continue_as_new", + "SideEffectRecorded": "side_effect", + "VersionMarkerRecorded": "version_marker", + "MemoUpserted": "memo", + "SearchAttributesUpserted": "search_attributes", +} +NEWLY_CLAIMED = ( + "ActivityHeartbeatRecorded", "ActivityRetryScheduled", "ActivityCancellationAcknowledged", + "ChildCancellationRequested", "ChildCancellationResolved", "ServiceOperationScheduled", "SignalWaitReceived", + "CancellationScopeOpened", "WorkflowContinuedAsNew", +) + + +def test_claim_table_is_exact() -> None: + from durable_workflow.workflow import _DURABLE_COMMAND_KINDS + + assert _DURABLE_COMMAND_KINDS == EXPECTED_CLAIMS + + +# Deliveries need a valid cancellation history to reach the classifier; they +# have their own tests below. +@pytest.mark.parametrize("event_type", sorted(set(EXPECTED_CLAIMS) - { + "SearchAttributesUpserted", "CooperativeCancellationDelivered", "CancellationScopeDelivered", +})) +@pytest.mark.parametrize("order", ["after_upsert", "before_upsert"]) +def test_every_claimed_kind_collides_with_an_upsert_without_its_opener( + fixture: str, event_type: str, order: str, +) -> None: + scope = event_type == "CancellationScopeOpened" + + def change(history: list[dict[str, Any]]) -> None: + upsert = only_event(history, "SearchAttributesUpserted") + payload: dict[str, Any] = {"sequence": 2} + if scope: + # A canonical authored opening, so scope history admits it and + # replay reaches the sequence classification. + payload.update(schema="durable-workflow.cancellation-scope/v1", workflow_run_id="search-attribute-run", + scope_id="scope-one", parent_scope_id="root", shield_parent=False) + claim = {"event_type": event_type, "recorded_at": upsert["recorded_at"], "payload": payload} + at = history.index(upsert) + (1 if order == "after_upsert" else 0) + history.insert(at, claim) + for number, event in enumerate(history, start=1): + event["sequence"] = number + if scope: + event.update(id=f"event-{number}", namespace="tenant") + + options = {"allow_cancellation_scope_authoring": True} if scope else {} + refused_at(RedriveVisibilityWorkflow, _with(fixture, change), "durable_command_sequence_collision", 2, **options) + + +def test_newly_claimed_kinds_are_claimed() -> None: + from durable_workflow.workflow import _DURABLE_COMMAND_KINDS + + assert set(NEWLY_CLAIMED) <= set(_DURABLE_COMMAND_KINDS) + + +@pytest.mark.parametrize("event_type", [ + "SignalReceived", "UpdateApplied", "UpdateCompleted", "FailureHandled", "CooperativeCancellationRequested", + "CancellationScopeRequested", "CancellationScopeDeliveryPrepared", "SelectionResolved", +]) +def test_pointer_records_claim_no_sequence(event_type: str) -> None: + from durable_workflow.workflow import _DURABLE_COMMAND_KINDS + + assert event_type not in _DURABLE_COMMAND_KINDS + + + +# Cancellation deliveries are durable command positions (Core +# WorkflowStepHistory::nextSequenceFromEvents). A delivery claims its sequence +# and, with sequence_span, every sequence through sequence + span - 1. +DELIVERY_FIXTURE = "redrive-reused-upserts-cancellation-clock" + + +def _cooperative(fixture: str, sequence: int, span: int, order: str, drop_memo: bool = False) -> list[dict[str, Any]]: + """The fixture's history with the cancellation of the cleanup fixture, + delivered at ``sequence`` for ``span`` calls, and no scheduled opener.""" + source = redriven(DELIVERY_FIXTURE) + request, delivery = deepcopy(source[-2]), deepcopy(source[-1]) + assert request["event_type"] == "CooperativeCancellationRequested" + delivery["payload"].update(sequence=sequence, call_kind="parallel" if span > 1 else "activity") + if span > 1: + delivery["payload"]["sequence_span"] = span + history = deepcopy(redriven(fixture)) + if drop_memo: + history.remove(only_event(history, "MemoUpserted")) + upsert = only_event(history, "SearchAttributesUpserted") + at = history.index(upsert) + (1 if order == "after_upsert" else 0) + history[at:at] = [request, delivery] + for number, event in enumerate(history, start=1): + event["sequence"] = number + return history + + +def refused_with_cancellation(history: list[dict[str, Any]], sequence: int) -> None: + workflow = redrive_fixture(DELIVERY_FIXTURE)["workflow"] + with pytest.raises(NonDeterministicReplayError, match="durable_command_sequence_collision:") as error: + replay(RedriveVisibilityWorkflow, history, ["Taylor"], run_id=workflow["run_id"], + workflow_id=workflow["workflow_id"], payload_codec="avro") + assert error.value.workflow_sequence == sequence + + +@pytest.mark.parametrize("order", ["after_upsert", "before_upsert"]) +def test_a_single_call_delivery_at_an_upsert_sequence_collides(fixture: str, order: str) -> None: + refused_with_cancellation(_cooperative(fixture, 2, 1, order), 2) + + +@pytest.mark.parametrize("order", ["after_upsert", "before_upsert"]) +@pytest.mark.parametrize(("base", "span"), [(1, 3), (1, 2), (2, 2)]) +def test_a_delivery_span_claims_every_sequence_it_covers(fixture: str, order: str, base: int, span: int) -> None: + # The upsert at 2 is interior to 1..3, the end of 1..2, or the base of 2..3. + # The memo at 1 is removed so the upsert is the only one inside the span. + refused_with_cancellation(_cooperative(fixture, base, span, order, drop_memo=True), 2) + + +def test_a_delivery_span_ending_before_the_upsert_does_not_collide(fixture: str) -> None: + # Span 1..1 does not reach the upsert at 2; the classifier stays quiet and + # replay is refused, if at all, for an unrelated reason. + history = _cooperative(fixture, 1, 1, "after_upsert", drop_memo=True) + workflow = redrive_fixture(DELIVERY_FIXTURE)["workflow"] + try: + replay(RedriveVisibilityWorkflow, history, ["Taylor"], run_id=workflow["run_id"], + workflow_id=workflow["workflow_id"], payload_codec="avro") + except NonDeterministicReplayError as error: + assert "durable_command_sequence_collision" not in str(error) + + +@pytest.mark.parametrize(("span", "claimed"), [ + (..., [3]), # Absent means 1. + (None, [3]), # Null means 1, as in Core (WorkflowStepHistory.php:270). + (1, [3]), + (3, [3, 4, 5]), +]) +@pytest.mark.parametrize("event_type", ["CooperativeCancellationDelivered", "CancellationScopeDelivered"]) +def test_a_valid_delivery_span_claims_its_inclusive_range(event_type: str, span: Any, claimed: list[int]) -> None: + from durable_workflow.workflow import _durable_command_claimed_range + + payload: dict[str, Any] = {"sequence": 3} + if span is not ...: + payload["sequence_span"] = span + assert list(_durable_command_claimed_range(event_type, payload, 3)) == claimed + assert list(_durable_command_claimed_range("ActivityScheduled", payload, 3)) == [3] + + +SECRET_SPANS = ["sk_live_TESTSECRET123", {"token": "TESTSECRET456"}] +MALFORMED_SPANS = [0, -1, "2", "3", 2.0, 3.0, True, False, [3], {"span": 3}, *SECRET_SPANS] + + +@pytest.mark.parametrize("span", MALFORMED_SPANS) +@pytest.mark.parametrize("event_type", ["CooperativeCancellationDelivered", "CancellationScopeDelivered"]) +def test_a_malformed_delivery_span_is_refused_not_narrowed(event_type: str, span: Any) -> None: + from durable_workflow.workflow import _delivery_sequence_span, _durable_command_claimed_range + + payload = {"sequence": 3, "sequence_span": span} + with pytest.raises(ValueError, match="sequence_span must be a positive integer"): + _delivery_sequence_span(payload) + with pytest.raises(ValueError): + _durable_command_claimed_range(event_type, payload, 3) + + +@pytest.mark.parametrize("span", MALFORMED_SPANS) +@pytest.mark.parametrize("order", ["after_upsert", "before_upsert"]) +def test_a_malformed_delivery_span_refuses_replay(fixture: str, order: str, span: Any) -> None: + # The upsert at 2 would be interior to an intended 1..3 span; a malformed + # span refuses replay rather than claiming only the base. + history = _cooperative(fixture, 1, 3, order, drop_memo=True) + next(event for event in history if event["event_type"] == "CooperativeCancellationDelivered")[ + "payload"]["sequence_span"] = span + workflow = redrive_fixture(DELIVERY_FIXTURE)["workflow"] + with pytest.raises(NonDeterministicReplayError, match="durable_command_sequence_invalid:"): + replay(RedriveVisibilityWorkflow, history, ["Taylor"], run_id=workflow["run_id"], + workflow_id=workflow["workflow_id"], payload_codec="avro") + + +@pytest.mark.parametrize("span", MALFORMED_SPANS) +def test_a_malformed_scope_delivery_span_refuses_replay(span: Any) -> None: + history, task = _scope_delivery_with_upsert("after_delivery") + next(event for event in history if event["event_type"] == "CancellationScopeDelivered")[ + "payload"]["sequence_span"] = span + with pytest.raises(NonDeterministicReplayError, match="durable_command_sequence_invalid:"): + replay(RedriveVisibilityWorkflow, history, [], workflow_id=task["workflow_id"], run_id=task["run_id"], + payload_codec="avro", allow_cancellation_scope_authoring=True, allow_cancellation_scope_delivery=True) + + +@pytest.mark.parametrize("span", [..., None]) +def test_an_absent_or_null_delivery_span_claims_one_sequence(fixture: str, span: Any) -> None: + # Absent or null means 1: a delivery at 1 does not reach the upsert at 2, + # and replay is not refused for the span. + history = _cooperative(fixture, 1, 1, "after_upsert", drop_memo=True) + delivered = next(event for event in history if event["event_type"] == "CooperativeCancellationDelivered") + assert "sequence_span" not in delivered["payload"] + if span is None: + delivered["payload"]["sequence_span"] = None + workflow = redrive_fixture(DELIVERY_FIXTURE)["workflow"] + try: + replay(RedriveVisibilityWorkflow, history, ["Taylor"], run_id=workflow["run_id"], + workflow_id=workflow["workflow_id"], payload_codec="avro") + except NonDeterministicReplayError as error: + assert "durable_command_sequence" not in str(error) + + +def _scope_delivery_with_upsert(order: str) -> tuple[list[dict[str, Any]], dict[str, Any]]: + from tests.test_committed_cancellation_scope_history import fixture as scope_fixture + + value = scope_fixture("committed-scope-delivery.json", "unshielded") + history = value["history"] + delivered = next(event for event in history if event["event_type"] == "CancellationScopeDelivered") + upsert = {"id": "upsert-event", "namespace": history[0]["namespace"], "event_type": "SearchAttributesUpserted", + "timestamp": delivered.get("timestamp"), "payload": { + "sequence": delivered["payload"]["sequence"], "attributes": {"Stage": "greeting"}, + "attribute_types": {"Stage": "keyword"}, "merged": {"Stage": "greeting"}}} + at = history.index(delivered) + (1 if order == "after_delivery" else 0) + history.insert(at, upsert) + for number, event in enumerate(history, start=1): + event["sequence"] = number + return history, value["task"] + + +@pytest.mark.parametrize("order", ["after_delivery", "before_delivery"]) +def test_a_scope_delivery_at_an_upsert_sequence_collides(order: str) -> None: + # A valid committed scope delivery with no scheduled timer opener, replayed + # with scope delivery enabled; an upsert at its sequence collides. + history, task = _scope_delivery_with_upsert(order) + sequence = next(event for event in history if event["event_type"] == "CancellationScopeDelivered")[ + "payload"]["sequence"] + with pytest.raises(NonDeterministicReplayError, match="durable_command_sequence_collision:") as error: + replay(RedriveVisibilityWorkflow, history, [], workflow_id=task["workflow_id"], run_id=task["run_id"], + payload_codec="avro", allow_cancellation_scope_authoring=True, allow_cancellation_scope_delivery=True) + assert error.value.workflow_sequence == sequence + + + +# Diagnostics name reasons, keys, types, sequences and event types, never a +# recorded or authored value. +SECRETS = ("sk_live_TESTSECRET123", "TESTSECRET456", "sk_live_TESTSECRET789", "TESTSECRET000", "KEYSECRET") + + +def _assert_no_secret(error: BaseException) -> None: + rendered = [str(error), repr(error), *(repr(argument) for argument in error.args)] + for secret in SECRETS: + assert not any(secret in text for text in rendered), (secret, rendered) + + +@pytest.mark.parametrize("span", SECRET_SPANS) +def test_a_secret_shaped_span_never_reaches_the_diagnostic(fixture: str, span: Any) -> None: + from durable_workflow.workflow import _delivery_sequence_span + + with pytest.raises(ValueError) as helper_error: + _delivery_sequence_span({"sequence": 1, "sequence_span": span}) + _assert_no_secret(helper_error.value) + assert ("got string" if isinstance(span, str) else "got object") in str(helper_error.value) + history = _cooperative(fixture, 1, 3, "after_upsert", drop_memo=True) + next(event for event in history if event["event_type"] == "CooperativeCancellationDelivered")[ + "payload"]["sequence_span"] = span + workflow = redrive_fixture(DELIVERY_FIXTURE)["workflow"] + with pytest.raises(NonDeterministicReplayError, match="durable_command_sequence_invalid:") as error: + replay(RedriveVisibilityWorkflow, history, ["Taylor"], run_id=workflow["run_id"], + workflow_id=workflow["workflow_id"], payload_codec="avro") + _assert_no_secret(error.value) + scope_history, task = _scope_delivery_with_upsert("after_delivery") + next(event for event in scope_history if event["event_type"] == "CancellationScopeDelivered")[ + "payload"]["sequence_span"] = span + with pytest.raises(NonDeterministicReplayError, match="durable_command_sequence_invalid:") as scope_error: + replay(RedriveVisibilityWorkflow, scope_history, [], workflow_id=task["workflow_id"], run_id=task["run_id"], + payload_codec="avro", allow_cancellation_scope_authoring=True, allow_cancellation_scope_delivery=True) + _assert_no_secret(scope_error.value) + + +@pytest.mark.parametrize(("authored", "stored", "types", "reason"), [ + ({"Stage": "sk_live_TESTSECRET789"}, {"Stage": "greeting"}, ..., "search_attribute_value_mismatch"), + ({"Stage": "greeting"}, {"Stage": "sk_live_TESTSECRET789"}, ..., "search_attribute_value_mismatch"), + ({"Stage": ["TESTSECRET000", 1]}, {"Stage": "greeting"}, ..., "search_attribute_value_mismatch"), + ({"Stage": "greeting"}, {"Stage": {"token": "TESTSECRET000"}}, ..., "search_attribute_value_mismatch"), + ({"sk_live_KEYSECRET 1": "x"}, {"Stage": "greeting"}, ..., "search_attribute_value_mismatch"), + ({"Stage": "greeting"}, {"sk_live_KEYSECRET 1": "x"}, ..., "search_attribute_value_mismatch"), + ({"Stage": "greeting"}, {"Stage": "greeting"}, {"Stage": "sk_live_TESTSECRET789"}, + "search_attribute_type_mismatch"), + ({"Stage": "greeting"}, {"Stage": "greeting"}, {"Stage": {"token": "TESTSECRET000"}}, + "search_attribute_type_mismatch"), + ({"Stage": "greeting"}, {"Stage": "greeting"}, "sk_live_TESTSECRET789", "search_attribute_type_mismatch"), +]) +def test_search_attribute_diagnostics_never_echo_values( + fixture: str, authored: dict[str, Any], stored: Any, types: Any, reason: str, +) -> None: + with pytest.raises(NonDeterministicReplayError, match=reason) as error: + replays(authoring(authored), recorded(fixture, stored, types)) + _assert_no_secret(error.value) + + +@pytest.mark.parametrize("change", [ + lambda payload: payload.__setitem__("sequence", "sk_live_TESTSECRET123"), + lambda payload: payload.__setitem__("sequence", {"token": "TESTSECRET456"}), +]) +def test_sequence_diagnostics_never_echo_values(fixture: str, change: Any) -> None: + history = deepcopy(redriven(fixture)) + change(only_event(history, "SearchAttributesUpserted")["payload"]) + with pytest.raises(NonDeterministicReplayError, match="durable_command_sequence_invalid") as error: + replays(RedriveVisibilityWorkflow, history) + _assert_no_secret(error.value)