diff --git a/.gitignore b/.gitignore index c36cea390..5bed7945a 100644 --- a/.gitignore +++ b/.gitignore @@ -8,4 +8,5 @@ coverage.xml .phpunit.result.cache .php_cs.cache .php-cs-fixer.cache -.vs \ No newline at end of file +.vs +/.regression-corpus-consumer-*/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 4d3e5a435..cad42acbf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,36 @@ ## Unreleased +- Redrive a failed run whose workflow applied an update before the failure. + The successor records the update's acceptance, application and completion + with `reused_from_run_id` and `reused_recorded_at`, ahead of the step it + ran before, and replay runs the update handler again from its recorded + arguments so the workflow state matches. The update is not delivered again. + A rejected update, and one the failure closed before it ran, are not reused. + Only three update shapes are accepted, in history order: a rejection, an + acceptance closed by the failure, or an acceptance, application and + successful completion with identical arguments. Every event in each shape + must name the same update, command and handler and belong to the source run. + An acceptance closed by the failure must record a terminal reason and a + failure, with no result and no step. An update the source + accepted but never resolved is refused with `unsupported_pending_update`, + any other shape or binding with `mismatched_update_history` or + `incomplete_update_step`, an externally stored update payload with + `external_update_payload_not_supported`, and an inline payload that can't be + decoded with `unavailable_update_payload`. A run a service worker started + that applied an update is refused with `service_redrive_history_not_qualified` + (#864). +- Redrive a successor that itself reused signal waits or updates. A reused + payload keeps the run that first recorded it, and `reused_from_run_id` names + the immediate source, so a chain of redrives reuses each step once. + `reused_from_run_id` and `reused_recorded_at` come as a pair on every reused + update and signal event. A reused event keeps its original time, and a + history where one key is missing, blank or names another source is refused. + Any `external_storage` key on a signal payload counts as external, even a + null one. Before, + redriving such a successor was refused with `mismatched_signal_receipt` + (#864). + ## 2.19.0 - 2026-11-03 - Rollback: releases before 2.19.0 ignore `reused_recorded_at` on the signal, diff --git a/docs/api-stability.md b/docs/api-stability.md index 538ec4f07..3c820f7c5 100644 --- a/docs/api-stability.md +++ b/docs/api-stability.md @@ -754,10 +754,10 @@ class, source file path, or stack trace to replay or handle a failure. | `SignalRejected` | `workflow_command_id`, `signal_id`, `signal_name`, `signal_wait_id`, `command_sequence`, `rejection_reason`, `terminal_reason` | `PendingSignalCloser`, `RunSignalView`, `HistoryTimeline` | | `SignalWaitOpened` | `signal_name`, `signal_wait_id`, `sequence`, `timeout_seconds`, `cancellation_scope_id` (candidate 1.20), `reused_from_run_id`, `reused_recorded_at` | `WorkflowStepHistory`, `SignalWaits`, `RunSignalView`, `HistoryTimeline` | | `SignalWaitCancelled` (candidate 1.20) | `signal_name`, `signal_wait_id`, `sequence`, `timeout_seconds`, `timer_id`, `cancelled_at`, `cancellation_scope` | `ScopedWaitCancellation`, `SignalWaits`, `HistoryTimeline`, scope delivery and worker history consumers | -| `UpdateAccepted` | `workflow_command_id`, `update_id`, `workflow_instance_id`, `workflow_run_id`, `update_name`, `arguments`, `ordering_state`, `queued_behind_command_id`, `queued_behind_command_sequence`, `queued_behind_command_type` | `RunUpdateView`, worker history payload consumers | +| `UpdateAccepted` | `workflow_command_id`, `update_id`, `workflow_instance_id`, `workflow_run_id`, `update_name`, `arguments`, `ordering_state`, `queued_behind_command_id`, `queued_behind_command_sequence`, `queued_behind_command_type`, `reused_from_run_id`, `reused_recorded_at` | `RunUpdateView`, worker history payload consumers | | `UpdateRejected` | `workflow_command_id`, `update_id`, `workflow_instance_id`, `workflow_run_id`, `update_name`, `arguments`, `validation_errors` | `RunUpdateView`, `HistoryTimeline`, command-contract projections | -| `UpdateApplied` | `workflow_command_id`, `update_id`, `workflow_instance_id`, `workflow_run_id`, `update_name`, `arguments`, `sequence` | `QueryStateReplayer`, `RunUpdateView`, worker history payload consumers | -| `UpdateCompleted` | `workflow_command_id`, `update_id`, `workflow_instance_id`, `workflow_run_id`, `update_name`, `sequence`, `result`, `failure_id`, `failure_category`, `non_retryable`, `exception_type`, `exception_class`, `message`, `code`, `exception`, `terminal_reason` | `RunUpdateView`, worker history payload consumers | +| `UpdateApplied` | `workflow_command_id`, `update_id`, `workflow_instance_id`, `workflow_run_id`, `update_name`, `arguments`, `sequence`, `reused_from_run_id`, `reused_recorded_at` | `QueryStateReplayer`, `RunUpdateView`, worker history payload consumers | +| `UpdateCompleted` | `workflow_command_id`, `update_id`, `workflow_instance_id`, `workflow_run_id`, `update_name`, `sequence`, `result`, `failure_id`, `failure_category`, `non_retryable`, `exception_type`, `exception_class`, `message`, `code`, `exception`, `terminal_reason`, `reused_from_run_id`, `reused_recorded_at` | `RunUpdateView`, worker history payload consumers | | `ConditionWaitOpened` | `condition_wait_id`, `condition_key`, `condition_definition_fingerprint`, `sequence`, `timeout_seconds`, `cancellation_scope_id` (candidate 1.20) | `WorkflowStepHistory`, `ConditionWaits`, worker history payload consumers | | `ConditionWaitSatisfied` | `condition_wait_id`, `condition_key`, `condition_definition_fingerprint`, `sequence`, `timer_id`, `timeout_seconds`, `workflow_signal_id`, `signal_name`, `signal_wait_id` | `WorkflowStepHistory`, `ConditionWaits`, `QueryStateReplayer`, `HistoryTimeline` | | `ConditionWaitTimedOut` | `condition_wait_id`, `condition_key`, `condition_definition_fingerprint`, `sequence`, `timer_id`, `timeout_seconds` | `WorkflowStepHistory`, `ConditionWaits`, `QueryStateReplayer`, `HistoryTimeline` | diff --git a/scripts/ci/test-regression-corpus-policy.py b/scripts/ci/test-regression-corpus-policy.py index 9aff68341..132a5a2c3 100644 --- a/scripts/ci/test-regression-corpus-policy.py +++ b/scripts/ci/test-regression-corpus-policy.py @@ -2254,5 +2254,63 @@ def test_content_heuristics_can_be_corrected_without_removing_core_guards(self) self.assertIn("replay implementation changed but its corpus did not grow", result.stderr) + +def load_validator() -> Any: + import importlib.util + + # Loading the validator must not leave bytecode in the checkout either. + sys.dont_write_bytecode = True + spec = importlib.util.spec_from_file_location("validate_regression_corpus", VALIDATOR) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +class ConsumerScratchTreeTest(unittest.TestCase): + """The official consumer tree must never be committable source.""" + + def test_a_consumer_tree_under_the_repository_root_is_ignored(self) -> None: + validator = load_validator() + probe = "src/ConsumerScratchProbe.php" + root = str(REPOSITORY_ROOT) + + # The real consumer run: it materializes the tree under the root, then + # lists untracked and ignored files from inside it. + result = validator._consumer_result( + REPOSITORY_ROOT, + {probe: b" None: + prefix = load_validator().CONSUMER_TREE_PREFIX + result = run( + "git", + "check-ignore", + "--no-index", + "--quiet", + f"{prefix}abc123/src/V2/Support/Probe.php", + cwd=REPOSITORY_ROOT, + ) + self.assertEqual(0, result.returncode, "consumer scratch tree is not ignored by .gitignore") + + if __name__ == "__main__": unittest.main() diff --git a/scripts/ci/validate-regression-corpus.py b/scripts/ci/validate-regression-corpus.py index b1066490b..b1a5ba97b 100644 --- a/scripts/ci/validate-regression-corpus.py +++ b/scripts/ci/validate-regression-corpus.py @@ -308,6 +308,10 @@ PHP_CODEC_ALIASES: dict[str, str] = {} PHP_REPLAY_PAYLOAD_DECODER = Path(__file__).with_name("decode-replay-payload.php") RUNTIME_DEPENDENCY_PATHS = ("vendor",) +# The consumer tree lives under the repository root so it can hard-link the +# runtime dependencies. .gitignore excludes it, so a run that is killed before +# cleanup cannot leave committable source behind. +CONSUMER_TREE_PREFIX = ".regression-corpus-consumer-" ZERO_COMMIT = re.compile(r"^0+$") LEGACY_MALFORMED_WIRE_REPAIRS = { "%%%": "JSUl", @@ -1820,7 +1824,7 @@ def _consumer_result( ) -> subprocess.CompletedProcess[str]: started = time.monotonic() with tempfile.TemporaryDirectory( - prefix=".regression-corpus-consumer-", + prefix=CONSUMER_TREE_PREFIX, dir=source_root, ) as temporary: consumer_root = Path(temporary) diff --git a/src/V2/Support/FailedRunRedriveEligibility.php b/src/V2/Support/FailedRunRedriveEligibility.php index 9ff5d57c3..737a2ec16 100644 --- a/src/V2/Support/FailedRunRedriveEligibility.php +++ b/src/V2/Support/FailedRunRedriveEligibility.php @@ -108,12 +108,12 @@ public static function evaluate( } // Service-mode SDK workers are qualified to replay reused activities, - // side effects, version markers and timers. Reused signal, memo and - // search-attribute steps are not qualified yet, so a run a service - // worker started refuses them before anything is written. + // side effects, version markers and timers. Reused signal, memo, + // search-attribute and update steps are not qualified yet, so a run a + // service worker started refuses them before anything is written. if ( ($startedPayload['workflow_definition_fingerprint_source'] ?? null) === 'worker' - && self::reusesServiceUnqualifiedSteps($plan['reused_steps']) + && (self::reusesServiceUnqualifiedSteps($plan['reused_steps']) || ($plan['update_bundles'] ?? []) !== []) ) { return self::refuse('service_redrive_history_not_qualified'); } diff --git a/src/V2/Support/FailedRunRedrivePlan.php b/src/V2/Support/FailedRunRedrivePlan.php index 953912490..748be6d3e 100644 --- a/src/V2/Support/FailedRunRedrivePlan.php +++ b/src/V2/Support/FailedRunRedrivePlan.php @@ -49,6 +49,7 @@ public static function forRun(WorkflowRun $run): array $markerEvents = []; $timerEvents = []; $signalEvents = []; + $updateEvents = []; $lastActivityEvent = null; foreach ($events as $event) { @@ -117,6 +118,21 @@ public static function forRun(WorkflowRun $run): array continue; } + if (in_array($event->event_type, [ + HistoryEventType::UpdateAccepted, + HistoryEventType::UpdateRejected, + HistoryEventType::UpdateApplied, + HistoryEventType::UpdateCompleted, + ], true)) { + if ($started === null) { + return self::reject('invalid_step_sequence'); + } + + $updateEvents[] = $event; + + continue; + } + if ($event->event_type === HistoryEventType::MessageCursorAdvanced) { // Message stream bookkeeping, not a step. Redrive carries the // source's cursor position to the successor instead. @@ -212,12 +228,19 @@ public static function forRun(WorkflowRun $run): array } } - $signalSteps = self::signalSteps($signalEvents, $run); + $provenance = self::provenance($run, $started); + $signalSteps = self::signalSteps($signalEvents, $run, $provenance); if (is_string($signalSteps)) { return self::reject($signalSteps); } + $updateBundles = self::updateBundles($updateEvents, $run, $boundary, $provenance); + + if (is_string($updateBundles)) { + return self::reject($updateBundles); + } + foreach ([ ...array_keys($markerEvents), ...array_keys($timerEvents), @@ -392,6 +415,7 @@ public static function forRun(WorkflowRun $run): array 'failed_step_kind' => $failedStepKind, 'completed' => $completed, 'reused_steps' => $reusedSteps, + 'update_bundles' => $updateBundles, ]; } @@ -406,8 +430,30 @@ public static function forRun(WorkflowRun $run): array public static function successorEvents(array $plan, string $sourceRunId): array { $events = []; + $updateBundles = $plan['update_bundles'] ?? []; + // An applied update runs before the step it is recorded at, so it is + // written ahead of that step. + $emitUpdatesThrough = static function (?int $sequence) use (&$updateBundles, &$events, $sourceRunId): void { + while ($updateBundles !== [] && ($sequence === null || $updateBundles[0]['sequence'] <= $sequence)) { + foreach (array_shift($updateBundles)['events'] as $updateEvent) { + $updatePayload = $updateEvent['payload']; + $events[] = [ + 'event_type' => $updateEvent['event_type'], + 'payload' => array_merge($updatePayload, [ + 'reused_from_run_id' => $sourceRunId, + 'reused_recorded_at' => self::originalRecordedAt( + $updatePayload, + $updateEvent['recorded_at'] + ), + ]), + ]; + } + } + }; foreach ($plan['reused_steps'] as $step) { + $stepSequence = $step['payload']['sequence'] ?? null; + $emitUpdatesThrough(is_int($stepSequence) ? $stepSequence : null); $reused = $step['payload']; // A step reused more than once keeps the time it first ran. $reusedRecordedAt = is_string($reused['reused_recorded_at'] ?? null) @@ -424,9 +470,10 @@ public static function successorEvents(array $plan, string $sourceRunId): array 'event_type' => $signalEvent['event_type'], 'payload' => array_merge($signalPayload, [ 'reused_from_run_id' => $sourceRunId, - 'reused_recorded_at' => is_string($signalPayload['reused_recorded_at'] ?? null) - ? $signalPayload['reused_recorded_at'] - : $signalEvent['recorded_at'], + 'reused_recorded_at' => self::originalRecordedAt( + $signalPayload, + $signalEvent['recorded_at'] + ), ]), ]; } @@ -484,9 +531,324 @@ public static function successorEvents(array $plan, string $sourceRunId): array ]; } + $emitUpdatesThrough(null); + return $events; } + /** + * Who may have recorded a reused-step payload in this run's history: the + * run itself, or, for a step it reused, its immediate redrive source while + * the payload keeps the earlier run of the same instance that recorded it. + * + * @return array{immediate_source: string|null, earlier_runs: array} + */ + private static function provenance(WorkflowRun $run, ?WorkflowHistoryEvent $started): array + { + $startedPayload = is_array($started?->payload) ? $started->payload : []; + $immediateSource = ($startedPayload['recovery_kind'] ?? null) === 'redrive' + ? ($startedPayload['continued_from_run_id'] ?? null) + : null; + $earlierRuns = []; + + if (is_string($immediateSource) && $immediateSource !== '') { + foreach (WorkflowRun::query() + ->where('workflow_instance_id', $run->workflow_instance_id) + ->where('run_number', '<', $run->run_number) + ->pluck('id') as $id) { + $earlierRuns[(string) $id] = true; + } + } else { + $immediateSource = null; + } + + return [ + 'immediate_source' => $immediateSource, + 'earlier_runs' => $earlierRuns, + ]; + } + + /** + * @param array $payload + * @param array{immediate_source: string|null, earlier_runs: array} $provenance + */ + private static function boundToRun(array $payload, WorkflowRun $run, array $provenance): bool + { + if (($payload['workflow_instance_id'] ?? null) !== $run->workflow_instance_id + || ! self::hasReuseProvenance($payload, $provenance)) { + return false; + } + + if (! array_key_exists('reused_from_run_id', $payload)) { + return ($payload['workflow_run_id'] ?? null) === $run->id; + } + + $original = $payload['workflow_run_id'] ?? null; + + return is_string($original) && isset($provenance['earlier_runs'][$original]); + } + + /** + * The reuse keys come as a pair. Both absent: the event was recorded in + * this run. Both present: this run reused it from its immediate redrive + * source, keeping the time the event was first recorded. + * + * @param array $payload + * @param array{immediate_source: string|null, earlier_runs: array} $provenance + */ + private static function hasReuseProvenance(array $payload, array $provenance): bool + { + $hasSource = array_key_exists('reused_from_run_id', $payload); + + if ($hasSource !== array_key_exists('reused_recorded_at', $payload)) { + return false; + } + + if (! $hasSource) { + return true; + } + + $recordedAt = $payload['reused_recorded_at']; + + return $provenance['immediate_source'] !== null + && $payload['reused_from_run_id'] === $provenance['immediate_source'] + && is_string($recordedAt) + && trim($recordedAt) !== '' + && strtotime($recordedAt) !== false; + } + + /** + * When the event was first recorded. Only called for events whose reuse + * pair was validated, so a reused event always carries its original time. + * + * @param array $payload + */ + private static function originalRecordedAt(array $payload, ?string $recordedAt): ?string + { + return array_key_exists('reused_from_run_id', $payload) + ? $payload['reused_recorded_at'] + : $recordedAt; + } + + /** + * Every event of one update names the same update, command and handler, + * and was recorded by the same run with the same reuse provenance. + * + * @param list $events + * @param array{immediate_source: string|null, earlier_runs: array} $provenance + */ + private static function sameUpdate(array $events, WorkflowRun $run, array $provenance): bool + { + $first = $events[0]->payload; + + foreach ($events as $event) { + $payload = $event->payload; + + if (! is_array($payload)) { + return false; + } + + foreach (['update_id', 'workflow_command_id', 'update_name'] as $key) { + if (! is_string($payload[$key] ?? null) || $payload[$key] === '' || $payload[$key] !== $first[$key]) { + return false; + } + } + + if (! self::boundToRun($payload, $run, $provenance) + || ($payload['workflow_run_id'] ?? null) !== ($first['workflow_run_id'] ?? null) + || ($payload['reused_from_run_id'] ?? null) !== ($first['reused_from_run_id'] ?? null) + || (! array_key_exists('reused_from_run_id', $payload) && $event->recorded_at === null)) { + return false; + } + } + + return true; + } + + /** + * Group update events into the update's exact lifecycle. Only three shapes + * are accepted, in history order: a rejection alone; an acceptance closed + * by the run's failure before it ran; or an acceptance, application and + * successful completion, which the successor reuses at the step it ran + * before. Anything else is refused, because replay would not reproduce it. + * + * @param list $events + * @param array{immediate_source: string|null, earlier_runs: array} $provenance + * + * @return list, recorded_at: string|null}>}>|string + */ + private static function updateBundles( + array $events, + WorkflowRun $run, + int $boundary, + array $provenance + ): array|string { + $byUpdate = []; + + foreach ($events as $event) { + $updateId = is_array($event->payload) ? ($event->payload['update_id'] ?? null) : null; + + if (! is_string($updateId) || $updateId === '') { + return 'mismatched_update_history'; + } + + $byUpdate[$updateId][] = $event; + } + + $bundles = []; + + foreach ($byUpdate as $updateEvents) { + usort( + $updateEvents, + static fn (WorkflowHistoryEvent $left, WorkflowHistoryEvent $right): int => $left->sequence <=> $right->sequence + ); + $shape = array_map( + static fn (WorkflowHistoryEvent $event): string => $event->event_type->value, + $updateEvents + ); + + if ($shape === [HistoryEventType::UpdateRejected->value]) { + // Rejected before acceptance, so replay never sees it. + if (! self::sameUpdate($updateEvents, $run, $provenance)) { + return 'mismatched_update_history'; + } + + continue; + } + + if ($shape === [HistoryEventType::UpdateAccepted->value, HistoryEventType::UpdateCompleted->value]) { + // Closed with the run before its handler ran: a terminal + // failure with no result and no step. + $closed = $updateEvents[1]->payload; + + if ( + ! self::sameUpdate($updateEvents, $run, $provenance) + || ! is_string($closed['terminal_reason'] ?? null) + || trim($closed['terminal_reason']) === '' + || ! is_string($closed['failure_id'] ?? null) + || trim($closed['failure_id']) === '' + || array_key_exists('result', $closed) + || array_key_exists('sequence', $closed) + ) { + return 'mismatched_update_history'; + } + + continue; + } + + if ($shape === [HistoryEventType::UpdateAccepted->value]) { + return 'unsupported_pending_update'; + } + + if ($shape !== [ + HistoryEventType::UpdateAccepted->value, + HistoryEventType::UpdateApplied->value, + HistoryEventType::UpdateCompleted->value, + ]) { + return 'mismatched_update_history'; + } + + [$accepted, $applied, $completed] = $updateEvents; + $first = $accepted->payload; + + if (! self::sameUpdate($updateEvents, $run, $provenance)) { + return 'mismatched_update_history'; + } + + // Replay runs the handler from the applied arguments, so they must + // be exactly the accepted ones. + if ( + ! array_key_exists('arguments', $first) + || ! array_key_exists('arguments', $applied->payload) + || $first['arguments'] !== $applied->payload['arguments'] + ) { + return 'mismatched_update_history'; + } + + if ( + ! array_key_exists('result', $completed->payload) + || array_key_exists('failure_id', $completed->payload) + || array_key_exists('terminal_reason', $completed->payload) + ) { + return 'incomplete_update_step'; + } + + $sequence = $applied->payload['sequence'] ?? null; + + if (! is_int($sequence) || $sequence < 1 || $sequence > $boundary + || ($completed->payload['sequence'] ?? null) !== $sequence) { + return 'invalid_step_sequence'; + } + + foreach ([$first['arguments'], $completed->payload['result']] as $value) { + if (self::holdsExternalPayload($value)) { + return 'external_update_payload_not_supported'; + } + + if (! self::decodesInline($value, $run)) { + return 'unavailable_update_payload'; + } + } + + $bundleEvents = []; + foreach ([$accepted, $applied, $completed] as $event) { + $eventPayload = $event->payload; + // The task and command are the source run's recording context. + unset($eventPayload['task'], $eventPayload['command']); + $bundleEvents[] = [ + 'event_type' => $event->event_type, + 'payload' => $eventPayload, + 'recorded_at' => $event->recorded_at?->toIso8601String(), + ]; + } + + $bundles[] = [ + 'sequence' => $sequence, + 'order' => (int) $applied->sequence, + 'events' => $bundleEvents, + ]; + } + + usort($bundles, static fn (array $left, array $right): int => [$left['sequence'], $left['order']] + <=> [$right['sequence'], $right['order']]); + + return array_map(static fn (array $bundle): array => [ + 'sequence' => $bundle['sequence'], + 'events' => $bundle['events'], + ], $bundles); + } + + /** + * Any external-storage shape, including a key whose value is null. + */ + private static function holdsExternalPayload(mixed $value): bool + { + return (is_array($value) && array_key_exists('external_storage', $value)) + || (is_string($value) && ExternalPayloads::isStoredReference($value)); + } + + /** + * An inline payload the successor's replay can decode with the run's codec. + */ + private static function decodesInline(mixed $value, WorkflowRun $run): bool + { + $codec = is_string($run->payload_codec) && $run->payload_codec !== '' ? $run->payload_codec : 'avro'; + + try { + $blob = ExternalPayloads::payloadBlob($value, $codec, $run->namespace); + + if (! is_string($blob)) { + return false; + } + + Serializer::unserializeWithCodec($codec, $blob); + + return true; + } catch (Throwable) { + return false; + } + } + /** * A durable timer authored with timer(). Condition and signal timeouts, * cancellation cleanup and timers in a scope or parallel group belong to @@ -574,7 +936,7 @@ private static function firedTimerStep(array $events): ?array * * @return array, recorded_at: string|null, signal_events: list, recorded_at: string|null}>}>|string */ - private static function signalSteps(array $events, WorkflowRun $run): array|string + private static function signalSteps(array $events, WorkflowRun $run, array $provenance): array|string { $opened = []; $applied = []; @@ -656,12 +1018,25 @@ private static function signalSteps(array $events, WorkflowRun $run): array|stri || ! is_string($commandId) || $commandId === '' || ($receiptPayload['workflow_command_id'] ?? null) !== $commandId - || ($receiptPayload['workflow_run_id'] ?? null) !== $run->id - || ($receiptPayload['workflow_instance_id'] ?? null) !== $run->workflow_instance_id + || ! self::boundToRun($receiptPayload, $run, $provenance) ) { return 'mismatched_signal_receipt'; } + // The wait and application carry no run id, so the reuse pair is + // their provenance. All three events come from the same run. + foreach ([$wait, $application, $receipt] as $event) { + $eventPayload = is_array($event->payload) ? $event->payload : []; + + if ( + ! self::hasReuseProvenance($eventPayload, $provenance) + || ($eventPayload['reused_from_run_id'] ?? null) !== ($receiptPayload['reused_from_run_id'] ?? null) + || (! array_key_exists('reused_from_run_id', $eventPayload) && $event->recorded_at === null) + ) { + return 'mismatched_signal_receipt'; + } + } + // One admitted signal is one workflow input: a receipt may satisfy // only one wait. if (isset($appliedSignalIds[$signalId])) { @@ -675,8 +1050,8 @@ private static function signalSteps(array $events, WorkflowRun $run): array|stri } if ( - self::isExternalResult($receiptPayload['arguments'] ?? null) - || self::isExternalResult($applicationPayload['value']) + self::holdsExternalPayload($receiptPayload['arguments'] ?? null) + || self::holdsExternalPayload($applicationPayload['value']) ) { return 'external_signal_payload_not_supported'; } diff --git a/src/V2/Support/HistoryEventPayloadContract.php b/src/V2/Support/HistoryEventPayloadContract.php index 09c86d1ee..563b1857d 100644 --- a/src/V2/Support/HistoryEventPayloadContract.php +++ b/src/V2/Support/HistoryEventPayloadContract.php @@ -526,6 +526,8 @@ final class HistoryEventPayloadContract 'queued_behind_command_id', 'queued_behind_command_sequence', 'queued_behind_command_type', + 'reused_from_run_id', + 'reused_recorded_at', ], 'UpdateRejected' => [ 'workflow_command_id', @@ -544,6 +546,8 @@ final class HistoryEventPayloadContract 'update_name', 'arguments', 'sequence', + 'reused_from_run_id', + 'reused_recorded_at', ], 'UpdateCompleted' => [ 'workflow_command_id', @@ -562,6 +566,8 @@ final class HistoryEventPayloadContract 'code', 'exception', 'terminal_reason', + 'reused_from_run_id', + 'reused_recorded_at', ], 'ConditionWaitOpened' => [ 'condition_wait_id', diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index 78639eb69..4d7bd206d 100644 --- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php +++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php @@ -594,6 +594,50 @@ public function testRedriveCorrectedDefinitionCorpusReplaysThePrefixWithNewCode( ); } + public function testRedriveReusedUpdateCorpusReappliesItsStateChange(): void + { + $path = self::FIXTURE_DIR . '/redrive-reused-update.json'; + if (! is_file($path)) { + $this->markTestSkipped('The redrive reused-update evidence is absent.'); + } + $fixture = json_decode(file_get_contents($path), true, flags: JSON_THROW_ON_ERROR); + $scenario = $fixture['redrive_reused_update']; + self::stopWorkers(); + config([ + 'queue.default' => 'sync', + 'queue.connections.sync.driver' => 'sync', + ]); + Cache::forget('test:redrive:first-calls'); + Cache::forget('test:redrive:second-calls'); + $this->clearWorkflowState(); + WorkflowStub::fake(); + + $stub = WorkflowStub::make($fixture['workflow']['type'], 'regression-corpus-redrive-update'); + $stub->start(...$fixture['workflow']['arguments']); + $this->assertSame( + $scenario['update_result'], + $stub->update($scenario['update'], ...$scenario['update_arguments']) + ); + $stub->signal($scenario['signal']); + $this->assertTrue($stub->refresh()->failed()); + $sourceRunId = $stub->runId(); + + $result = app(WorkflowControlPlane::class)->redrive('regression-corpus-redrive-update', $sourceRunId); + + $this->assertSame($scenario['redrive'] === 'accepted', $result['accepted'], (string) $result['reason']); + $successor = WorkflowRun::query()->findOrFail($result['workflow_run_id']); + $this->assertSame($scenario['successor_status'], $successor->status->value); + $this->assertSame($scenario['output'], $successor->workflowOutput()); + $this->assertSame(1, (int) Cache::get('test:redrive:first-calls')); + foreach ($scenario['reused_events'] as $type) { + $reused = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $successor->id) + ->where('event_type', $type) + ->sole(); + $this->assertSame($sourceRunId, $reused->payload['reused_from_run_id'] ?? null); + } + } + public function testRedriveReusedTimerCorpusDoesNotWaitAgain(): void { $path = self::FIXTURE_DIR . '/redrive-reused-timer-clock.json'; diff --git a/tests/Feature/V2/V2FailedRunRedriveTest.php b/tests/Feature/V2/V2FailedRunRedriveTest.php index 585fadd47..98cb2cc76 100644 --- a/tests/Feature/V2/V2FailedRunRedriveTest.php +++ b/tests/Feature/V2/V2FailedRunRedriveTest.php @@ -18,6 +18,7 @@ use Tests\Fixtures\V2\TestRedriveSignalWorkflow; use Tests\Fixtures\V2\TestRedriveTimerWorkflow; use Tests\Fixtures\V2\TestRedriveTwoSignalsWorkflow; +use Tests\Fixtures\V2\TestRedriveUpdateWorkflow; use Tests\Fixtures\V2\TestRedriveVersionedBuggyWorkflow; use Tests\Fixtures\V2\TestRedriveVersionedIncompatibleWorkflow; use Tests\Fixtures\V2\TestRedriveVisibilityWorkflow; @@ -40,6 +41,7 @@ use Workflow\V2\Models\WorkflowSignal; use Workflow\V2\Models\WorkflowTask; use Workflow\V2\Models\WorkflowTimer; +use Workflow\V2\Models\WorkflowUpdate; use Workflow\V2\StartOptions; use Workflow\V2\Support\DefaultWorkflowTaskBridge; use Workflow\V2\Support\EmbeddedV2HistoryImport; @@ -796,6 +798,25 @@ public static function signalStepMutations(): iterable yield 'external applied value' => [ HistoryEventType::SignalApplied, 'value', $external, 'external_signal_payload_not_supported', ]; + // Any external-storage key is external, even a null one. + yield 'null external storage key on receipt arguments' => [ + HistoryEventType::SignalReceived, + 'arguments', + [ + 'codec' => 'avro', + 'external_storage' => null, + ], + 'external_signal_payload_not_supported', + ]; + yield 'null external storage key on applied value' => [ + HistoryEventType::SignalApplied, + 'value', + [ + 'codec' => 'avro', + 'external_storage' => null, + ], + 'external_signal_payload_not_supported', + ]; } #[\PHPUnit\Framework\Attributes\DataProvider('signalStepMutations')] @@ -1146,6 +1167,447 @@ public function testALaterRedriveOnTheSameDefinitionRecordsNoInheritedChange(): ])['run']['definition_changed_from'] ?? null); } + public function testRedriveReappliesAnUpdateTheSourceApplied(): void + { + WorkflowStub::fake(); + + $workflow = WorkflowStub::make(TestRedriveUpdateWorkflow::class, 'redrive-update-1'); + $workflow->start('Taylor'); + $this->assertSame('noted urgent', $workflow->update('set-note', 'urgent')); + $workflow->signal('go'); + $this->assertTrue($workflow->refresh()->failed()); + $source = WorkflowRun::query()->findOrFail($workflow->runId()); + $sourceApplied = $this->sourceEvent($source, HistoryEventType::UpdateApplied); + + $result = app(WorkflowControlPlane::class)->redrive('redrive-update-1', $source->id); + + $this->assertTrue($result['accepted'], (string) $result['reason']); + $successor = WorkflowRun::query()->findOrFail($result['workflow_run_id']); + $this->assertSame(RunStatus::Completed, $successor->status); + $this->assertSame(1, (int) Cache::get('test:redrive:first-calls')); + // The update's state change is replayed, not delivered again. + $this->assertSame([ + 'result' => 'Hello, Taylor! Redrive completed.', + 'note' => 'urgent', + ], $successor->workflowOutput()); + $this->assertSame(0, WorkflowUpdate::query()->where('workflow_run_id', $successor->id)->count()); + + $types = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $successor->id) + ->orderBy('sequence') + ->pluck('event_type') + ->map(static fn (HistoryEventType $type): string => $type->value) + ->all(); + $applied = array_search('UpdateApplied', $types, true); + $this->assertNotFalse($applied); + $this->assertSame('UpdateAccepted', $types[$applied - 1]); + $this->assertSame('UpdateCompleted', $types[$applied + 1]); + // It is written ahead of the step it ran before. + $this->assertLessThan(array_search('SignalWaitOpened', $types, true), $applied); + // Replay applies it from the same recorded position as the source. + $this->assertSame( + $sourceApplied->payload['sequence'], + $this->sourceEvent($successor, HistoryEventType::UpdateApplied)->payload['sequence'], + ); + foreach ([ + HistoryEventType::UpdateAccepted, + HistoryEventType::UpdateApplied, + HistoryEventType::UpdateCompleted, + ] as $type) { + $original = $this->sourceEvent($source, $type); + $reused = $this->sourceEvent($successor, $type); + $this->assertSame($source->id, $reused->payload['reused_from_run_id'] ?? null); + $this->assertSame( + $original->recorded_at?->toIso8601String(), + $reused->payload['reused_recorded_at'] ?? null + ); + } + } + + public function testAnUpdateAndSignalWaitRedriveAcrossAChain(): void + { + WorkflowStub::fake(); + config([ + 'workflows.v2.testing.redrive_failures_before_success' => 2, + ]); + $controlPlane = app(WorkflowControlPlane::class); + + $workflow = WorkflowStub::make(TestRedriveUpdateWorkflow::class, 'redrive-update-chain'); + $workflow->start('Taylor'); + $workflow->update('set-note', 'urgent'); + $workflow->signal('go'); + $this->assertTrue($workflow->refresh()->failed()); + $first = WorkflowRun::query()->findOrFail($workflow->runId()); + + $second = $controlPlane->redrive('redrive-update-chain', $first->id); + $this->assertTrue($second['accepted'], (string) $second['reason']); + $secondRun = WorkflowRun::query()->findOrFail($second['workflow_run_id']); + $this->assertSame(RunStatus::Failed, $secondRun->status); + + $third = $controlPlane->redrive('redrive-update-chain', $secondRun->id); + + $this->assertTrue($third['accepted'], (string) $third['reason']); + $thirdRun = WorkflowRun::query()->findOrFail($third['workflow_run_id']); + $this->assertSame(RunStatus::Completed, $thirdRun->status); + $this->assertSame('urgent', $thirdRun->workflowOutput()['note']); + $this->assertSame(1, (int) Cache::get('test:redrive:first-calls')); + foreach ([ + HistoryEventType::UpdateAccepted, + HistoryEventType::UpdateApplied, + HistoryEventType::UpdateCompleted, + HistoryEventType::SignalReceived, + HistoryEventType::SignalApplied, + ] as $type) { + $events = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $thirdRun->id) + ->where('event_type', $type) + ->get(); + $this->assertCount(1, $events, $type->value); + $original = $this->sourceEvent($first, $type); + // The immediate source, the original run and its first time. + $this->assertSame($secondRun->id, $events[0]->payload['reused_from_run_id'], $type->value); + $this->assertSame($first->id, $events[0]->payload['workflow_run_id'] ?? $first->id, $type->value); + $this->assertSame( + $original->recorded_at?->toIso8601String(), + $events[0]->payload['reused_recorded_at'], + $type->value, + ); + } + $this->assertSame( + 1, + WorkflowUpdate::query()->where('workflow_instance_id', 'redrive-update-chain')->count(), + ); + $this->assertSame( + 0, + WorkflowCommand::query() + ->whereIn('workflow_run_id', [$secondRun->id, $thirdRun->id]) + ->whereIn('command_type', ['update', 'signal']) + ->count(), + ); + } + + /** + * @return iterable + */ + public static function malformedUpdateHistories(): iterable + { + yield 'acceptance then rejection' => ['accepted_then_rejected', 'mismatched_update_history']; + yield 'lone terminal completion' => ['lone_terminal_completion', 'mismatched_update_history']; + yield 'application before acceptance' => ['reversed_order', 'mismatched_update_history']; + yield 'no command or update name' => ['null_identity', 'mismatched_update_history']; + yield 'applied arguments differ' => ['changed_arguments', 'mismatched_update_history']; + yield 'failure key on a successful completion' => ['null_failure_key', 'incomplete_update_step']; + yield 'null external storage key' => ['null_external_storage', 'external_update_payload_not_supported']; + yield 'undecodable inline result' => ['undecodable_result', 'unavailable_update_payload']; + } + + #[\PHPUnit\Framework\Attributes\DataProvider('malformedUpdateHistories')] + public function testRedriveRefusesMalformedUpdateHistoryWithoutMutation(string $mutation, string $reason): void + { + WorkflowStub::fake(); + + $workflow = WorkflowStub::make(TestRedriveUpdateWorkflow::class, 'redrive-update-malformed'); + $workflow->start('Taylor'); + $workflow->update('set-note', 'urgent'); + $workflow->signal('go'); + $this->assertTrue($workflow->refresh()->failed()); + $source = WorkflowRun::query()->findOrFail($workflow->runId()); + $accepted = $this->sourceEvent($source, HistoryEventType::UpdateAccepted); + $applied = $this->sourceEvent($source, HistoryEventType::UpdateApplied); + $completed = $this->sourceEvent($source, HistoryEventType::UpdateCompleted); + $merge = static function (WorkflowHistoryEvent $event, array $values): void { + $event->forceFill([ + 'payload' => array_merge($event->payload, $values), + ])->save(); + }; + + match ($mutation) { + 'accepted_then_rejected' => (static function () use ($applied, $completed): void { + $completed->delete(); + $applied->forceFill([ + 'event_type' => HistoryEventType::UpdateRejected, + ])->save(); + })(), + 'lone_terminal_completion' => (static function () use ($accepted, $applied, $completed, $merge): void { + $accepted->delete(); + $applied->delete(); + $merge($completed, [ + 'terminal_reason' => 'failed', + ]); + })(), + 'reversed_order' => (static function () use ($accepted, $applied): void { + [$acceptedSequence, $appliedSequence] = [$accepted->sequence, $applied->sequence]; + $accepted->forceFill([ + 'sequence' => 100000, + ])->save(); + $applied->forceFill([ + 'sequence' => $acceptedSequence, + ])->save(); + $accepted->forceFill([ + 'sequence' => $appliedSequence, + ])->save(); + })(), + 'null_identity' => array_map(static fn (WorkflowHistoryEvent $event) => $merge($event, [ + 'workflow_command_id' => null, + 'update_name' => null, + ]), [$accepted, $applied, $completed]), + 'changed_arguments' => $merge($applied, [ + 'arguments' => Serializer::serializeWithCodec('avro', ['calm']), + ]), + 'null_failure_key' => $merge($completed, [ + 'failure_id' => null, + ]), + 'null_external_storage' => array_map(static fn (WorkflowHistoryEvent $event) => $merge($event, [ + 'arguments' => [ + 'codec' => 'avro', + 'external_storage' => null, + ], + ]), [$accepted, $applied]), + 'undecodable_result' => $merge($completed, [ + 'result' => 'not-an-avro-payload', + ]), + }; + + $before = $this->engineSnapshot(); + $currentRun = $source->instance->fresh() +->current_run_id; + + $result = app(WorkflowControlPlane::class)->redrive('redrive-update-malformed', $source->id); + + $this->assertFalse($result['accepted']); + $this->assertSame($reason, $result['reason']); + $this->assertSame($before, $this->engineSnapshot()); + $this->assertSame($currentRun, $source->instance->fresh()->current_run_id); + } + + /** + * @return iterable + */ + public static function closedUpdateShapes(): iterable + { + yield 'closed with the run' => ['terminal', 'none', null]; + yield 'closed with a null terminal reason' => ['terminal', 'null_terminal_reason', 'mismatched_update_history']; + yield 'closed with a blank terminal reason' => [ + 'terminal', + 'blank_terminal_reason', + 'mismatched_update_history', + ]; + yield 'closed with a result' => ['terminal', 'result', 'mismatched_update_history']; + yield 'closed at a step' => ['terminal', 'step_sequence', 'mismatched_update_history']; + yield 'closed without a failure' => ['terminal', 'missing_failure_id', 'mismatched_update_history']; + yield 'closed by another run' => ['terminal', 'foreign_run', 'mismatched_update_history']; + yield 'closed for another command' => ['terminal', 'foreign_command', 'mismatched_update_history']; + yield 'closed for another handler' => ['terminal', 'foreign_name', 'mismatched_update_history']; + yield 'closed with a lone reuse time' => ['terminal', 'lone_recorded_at', 'mismatched_update_history']; + yield 'rejected' => ['rejected', 'none', null]; + yield 'rejected by another run' => ['rejected', 'foreign_run', 'mismatched_update_history']; + yield 'rejected in another instance' => ['rejected', 'foreign_instance', 'mismatched_update_history']; + yield 'rejected without a command' => ['rejected', 'null_command', 'mismatched_update_history']; + yield 'rejected without a handler name' => ['rejected', 'blank_name', 'mismatched_update_history']; + yield 'rejected with a lone reuse time' => ['rejected', 'lone_recorded_at', 'mismatched_update_history']; + yield 'rejected with a lone reuse source' => ['rejected', 'lone_source', 'mismatched_update_history']; + } + + /** + * A rejected update, or one the run's failure closed before it ran, is not + * replayed. Redrive skips it only when it is exactly that shape and belongs + * to the source run. + */ + #[\PHPUnit\Framework\Attributes\DataProvider('closedUpdateShapes')] + public function testRedriveSkipsOnlyAnExactlyClosedOrRejectedUpdate( + string $shape, + string $mutation, + ?string $reason, + ): void { + WorkflowStub::fake(); + + $workflow = WorkflowStub::make(TestRedriveUpdateWorkflow::class, 'redrive-update-closed'); + $workflow->start('Taylor'); + $workflow->update('set-note', 'urgent'); + $workflow->signal('go'); + $this->assertTrue($workflow->refresh()->failed()); + $source = WorkflowRun::query()->findOrFail($workflow->runId()); + $accepted = $this->sourceEvent($source, HistoryEventType::UpdateAccepted); + $applied = $this->sourceEvent($source, HistoryEventType::UpdateApplied); + $completed = $this->sourceEvent($source, HistoryEventType::UpdateCompleted); + $applied->delete(); + + if ($shape === 'terminal') { + // The shape PendingUpdateCloser records: a failure, no result, no step. + $closed = $completed->payload; + unset($closed['result'], $closed['sequence']); + $completed->forceFill([ + 'payload' => array_merge($closed, [ + 'failure_id' => 'update-closed-failure', + 'terminal_reason' => 'failed', + ]), + ])->save(); + $target = $completed; + } else { + $completed->delete(); + $accepted->forceFill([ + 'event_type' => HistoryEventType::UpdateRejected, + ])->save(); + $target = $accepted; + } + + $payload = $target->fresh() + ->payload; + match ($mutation) { + 'none' => null, + 'null_terminal_reason' => $payload['terminal_reason'] = null, + 'blank_terminal_reason' => $payload['terminal_reason'] = ' ', + 'result' => $payload['result'] = Serializer::serializeWithCodec('avro', 'noted urgent'), + 'step_sequence' => $payload['sequence'] = 2, + 'missing_failure_id' => (static function () use (&$payload): void { + unset($payload['failure_id']); + })(), + 'foreign_run' => $payload['workflow_run_id'] = 'another-run', + 'foreign_instance' => $payload['workflow_instance_id'] = 'another-instance', + 'foreign_command' => $payload['workflow_command_id'] = 'another-command', + 'null_command' => $payload['workflow_command_id'] = null, + 'foreign_name' => $payload['update_name'] = 'another-update', + 'blank_name' => $payload['update_name'] = '', + 'lone_recorded_at' => $payload['reused_recorded_at'] = now()->toIso8601String(), + 'lone_source' => $payload['reused_from_run_id'] = $source->id, + }; + $target->forceFill([ + 'payload' => $payload, + ])->save(); + + $plan = FailedRunRedrivePlan::forRun($source->fresh()); + + if ($reason === null) { + $this->assertTrue($plan['eligible'], (string) $plan['reason']); + $this->assertSame([], $plan['update_bundles']); + + return; + } + + $this->assertRefusedWithoutMutation('redrive-update-closed', $source, $reason); + } + + /** + * @return iterable + */ + public static function chainedProvenanceMutations(): iterable + { + $types = [ + HistoryEventType::UpdateAccepted, + HistoryEventType::UpdateApplied, + HistoryEventType::UpdateCompleted, + HistoryEventType::SignalWaitOpened, + HistoryEventType::SignalReceived, + HistoryEventType::SignalApplied, + ]; + + foreach ($types as $type) { + $reason = str_starts_with($type->value, 'Update') + ? 'mismatched_update_history' + : 'mismatched_signal_receipt'; + + // Recorded by the source itself: the reuse keys are both absent. + yield $type->value . ' original with a lone reuse time' => ['source', $type, 'lone_recorded_at', $reason]; + yield $type->value . ' original with a lone reuse source' => ['source', $type, 'lone_source', $reason]; + + // Reused by a redrive: the keys are both present, name the + // immediate source and keep the first time. + foreach ([ + 'missing_time', + 'null_time', + 'blank_time', + 'missing_source', + 'null_source', + 'foreign_source', + ] as $mutation) { + yield $type->value . ' reused with ' . str_replace('_', ' ', $mutation) => [ + 'successor', + $type, + $mutation, + $reason, + ]; + } + } + } + + #[\PHPUnit\Framework\Attributes\DataProvider('chainedProvenanceMutations')] + public function testRedriveRefusesABrokenReuseProvenancePairWithoutMutation( + string $run, + HistoryEventType $type, + string $mutation, + string $reason, + ): void { + WorkflowStub::fake(); + config([ + 'workflows.v2.testing.redrive_failures_before_success' => 2, + ]); + + $workflow = WorkflowStub::make(TestRedriveUpdateWorkflow::class, 'redrive-update-pair'); + $workflow->start('Taylor'); + $workflow->update('set-note', 'urgent'); + $workflow->signal('go'); + $this->assertTrue($workflow->refresh()->failed()); + $source = WorkflowRun::query()->findOrFail($workflow->runId()); + + if ($run === 'successor') { + $redriven = app(WorkflowControlPlane::class)->redrive('redrive-update-pair', $source->id); + $this->assertTrue($redriven['accepted'], (string) $redriven['reason']); + $source = WorkflowRun::query()->findOrFail($redriven['workflow_run_id']); + $this->assertSame(RunStatus::Failed, $source->status); + // The unmodified redriven run is redrivable again. + $this->assertTrue(FailedRunRedrivePlan::forRun($source->fresh())['eligible']); + } + + $event = $this->sourceEvent($source, $type); + $payload = $event->payload; + match ($mutation) { + 'lone_recorded_at' => $payload['reused_recorded_at'] = now()->toIso8601String(), + 'lone_source' => $payload['reused_from_run_id'] = $source->id, + 'missing_time' => (static function () use (&$payload): void { + unset($payload['reused_recorded_at']); + })(), + 'null_time' => $payload['reused_recorded_at'] = null, + 'blank_time' => $payload['reused_recorded_at'] = '', + 'missing_source' => (static function () use (&$payload): void { + unset($payload['reused_from_run_id']); + })(), + 'null_source' => $payload['reused_from_run_id'] = null, + 'foreign_source' => $payload['reused_from_run_id'] = 'another-run', + }; + $event->forceFill([ + 'payload' => $payload, + ])->save(); + $before = $this->engineSnapshot(); + $currentRun = $source->instance->fresh() + ->current_run_id; + + $result = app(WorkflowControlPlane::class)->redrive('redrive-update-pair', $source->id); + + $this->assertFalse($result['accepted']); + $this->assertSame($reason, $result['reason']); + $this->assertSame($before, $this->engineSnapshot()); + $this->assertSame($currentRun, $source->instance->fresh()->current_run_id); + } + + public function testRedriveRefusesAnUpdateTheSourceNeverResolved(): void + { + WorkflowStub::fake(); + + $workflow = WorkflowStub::make(TestRedriveUpdateWorkflow::class, 'redrive-update-pending'); + $workflow->start('Taylor'); + $workflow->update('set-note', 'urgent'); + $workflow->signal('go'); + $this->assertTrue($workflow->refresh()->failed()); + $source = WorkflowRun::query()->findOrFail($workflow->runId()); + // Only the acceptance remains: the source never applied or closed it. + WorkflowHistoryEvent::query() + ->where('workflow_run_id', $source->id) + ->whereIn('event_type', [HistoryEventType::UpdateApplied->value, HistoryEventType::UpdateCompleted->value]) + ->delete(); + + $this->assertRefusedWithoutMutation('redrive-update-pending', $source, 'unsupported_pending_update'); + } + public function testRedriveReusesOneMarkerForARepeatedPatchCall(): void { WorkflowStub::fake(); @@ -1500,6 +1962,40 @@ private function failedSignalSource(string $instanceId): WorkflowRun return WorkflowRun::query()->findOrFail($workflow->runId()); } + /** + * Every engine table and read projection a refused redrive must leave + * untouched, including the run's message cursor. + * + * @return array>> + */ + private function engineSnapshot(): array + { + $snapshot = []; + foreach ([ + \Workflow\V2\Models\WorkflowInstance::class, + WorkflowRun::class, + WorkflowCommand::class, + WorkflowHistoryEvent::class, + \Workflow\V2\Models\WorkflowTask::class, + WorkflowLink::class, + \Workflow\V2\Models\WorkflowMemo::class, + \Workflow\V2\Models\WorkflowSearchAttribute::class, + WorkflowSignal::class, + WorkflowUpdate::class, + \Workflow\V2\Models\ActivityExecution::class, + \Workflow\V2\Models\ActivityAttempt::class, + \Workflow\V2\Models\WorkflowFailure::class, + \Workflow\V2\Models\WorkflowRunSummary::class, + \Workflow\V2\Models\WorkflowTimelineEntry::class, + ] as $model) { + $snapshot[$model] = $model::query()->orderBy((new $model())->getKeyName())->get()->map( + static fn (\Illuminate\Database\Eloquent\Model $row): array => $row->getRawOriginal(), + )->all(); + } + + return $snapshot; + } + private function sourceEvent(WorkflowRun $source, HistoryEventType $type): WorkflowHistoryEvent { return WorkflowHistoryEvent::query() diff --git a/tests/Fixtures/V2/ReplayRegression/redrive-reused-update.json b/tests/Fixtures/V2/ReplayRegression/redrive-reused-update.json new file mode 100644 index 000000000..25e0080bb --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/redrive-reused-update.json @@ -0,0 +1,44 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "redrive-reused-update", + "protocol_version": "1.0", + "bindings": ["php"], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestRedriveUpdateWorkflow", + "arguments": ["Taylor"], + "payload_codec": "avro" + }, + "redrive_reused_update": { + "update": "set-note", + "update_arguments": ["urgent"], + "update_result": "noted urgent", + "signal": "go", + "redrive": "accepted", + "successor_status": "completed", + "output": { + "result": "Hello, Taylor! Redrive completed.", + "note": "urgent" + }, + "reused_events": ["UpdateAccepted", "UpdateApplied", "UpdateCompleted"] + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": {} + } + ], + "expected": { + "completed": false, + "result": null, + "commands": [ + { + "type": "schedule_activity", + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "payload_codec": "avro" + } + ] + } +} diff --git a/tests/Fixtures/V2/TestRedriveUpdateWorkflow.php b/tests/Fixtures/V2/TestRedriveUpdateWorkflow.php new file mode 100644 index 000000000..784acb911 --- /dev/null +++ b/tests/Fixtures/V2/TestRedriveUpdateWorkflow.php @@ -0,0 +1,41 @@ + activity(TestRedriveSecondActivity::class, $greeting), + 'note' => $this->note, + ]; + } + + #[UpdateMethod('set-note')] + public function setNote(string $note): string + { + $this->note = $note; + + return "noted {$note}"; + } +} diff --git a/tests/Unit/V2/DefaultWorkflowControlPlaneRedriveRejectionTest.php b/tests/Unit/V2/DefaultWorkflowControlPlaneRedriveRejectionTest.php index 51fbd3d21..59bec342e 100644 --- a/tests/Unit/V2/DefaultWorkflowControlPlaneRedriveRejectionTest.php +++ b/tests/Unit/V2/DefaultWorkflowControlPlaneRedriveRejectionTest.php @@ -221,6 +221,7 @@ public static function serviceUnqualifiedSteps(): iterable yield 'memo upsert' => ['memo']; yield 'search attribute upsert' => ['search_attributes']; yield 'signal wait' => ['signal']; + yield 'applied update' => ['update']; } #[DataProvider('serviceUnqualifiedSteps')] @@ -258,8 +259,11 @@ public function testAServiceRunRefusesReusedStepsItsSdksAreNotQualifiedToReplay( ], ]), 'signal' => $this->recordAppliedSignal($source), + 'update' => null, }; - $this->recordFailure($source, includeCompletedPrefix: false); + $this->recordFailure($source, includeCompletedPrefix: $step === 'update', beforeFailure: $step === 'update' + ? fn () => $this->recordAppliedUpdate($source) + : null); $source = $source->fresh(); $before = $this->records(); $projections = $this->projections(); @@ -671,8 +675,11 @@ private function failedSource(array $options = [], string $instanceId = 'redrive return $source->fresh(); } - private function recordFailure(WorkflowRun $source, bool $includeCompletedPrefix = true): void - { + private function recordFailure( + WorkflowRun $source, + bool $includeCompletedPrefix = true, + ?\Closure $beforeFailure = null, + ): void { if ($includeCompletedPrefix) { $firstActivityId = (string) Str::ulid(); WorkflowHistoryEvent::record($source, HistoryEventType::ActivityScheduled, [ @@ -688,6 +695,9 @@ private function recordFailure(WorkflowRun $source, bool $includeCompletedPrefix 'payload_codec' => 'avro', ]); } + if ($beforeFailure !== null) { + $beforeFailure(); + } $activityId = (string) Str::ulid(); WorkflowHistoryEvent::record($source, HistoryEventType::ActivityScheduled, [ 'activity_execution_id' => $activityId, @@ -720,6 +730,29 @@ private function recordFailure(WorkflowRun $source, bool $includeCompletedPrefix $this->assertSame(RunStatus::Failed, $source->fresh()->status); } + private function recordAppliedUpdate(WorkflowRun $source): void + { + $identity = [ + 'workflow_command_id' => (string) Str::ulid(), + 'update_id' => (string) Str::ulid(), + 'workflow_instance_id' => $source->workflow_instance_id, + 'workflow_run_id' => $source->id, + 'update_name' => 'set-note', + ]; + $arguments = Serializer::serializeWithCodec('avro', ['urgent']); + WorkflowHistoryEvent::record($source, HistoryEventType::UpdateAccepted, $identity + [ + 'arguments' => $arguments, + ]); + WorkflowHistoryEvent::record($source, HistoryEventType::UpdateApplied, $identity + [ + 'arguments' => $arguments, + 'sequence' => 1, + ]); + WorkflowHistoryEvent::record($source, HistoryEventType::UpdateCompleted, $identity + [ + 'sequence' => 1, + 'result' => Serializer::serializeWithCodec('avro', 'noted urgent'), + ]); + } + private function recordAppliedSignal(WorkflowRun $source): WorkflowHistoryEvent { $waitId = (string) Str::ulid();