diff --git a/CHANGELOG.md b/CHANGELOG.md index 77c39ca59..e0b4159b1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,20 @@ can be redriven again. A row written by workflow code with no matching reused upsert step is still refused with `unsupported_visibility_metadata` (#864). +- Redrive a failed run whose completed steps include a signal wait + (`await('name')`) that received and applied its signal. The successor + records the wait, receipt and application with `reused_from_run_id` and + `reused_recorded_at`, and replay uses the original time, so the workflow + gets the same value without a new signal. Redrive also carries the source's + message cursor, so a consumed message is not delivered again. A wait with a + timeout, cancellation scope or parallel group is refused with + `unsupported_signal_wait`, a signal the source received but never applied + with `unsupported_pending_signal`, and a wait without its receipt and + application with `incomplete_signal_step`. A receipt that names another + run, instance, command or wait is refused with `mismatched_signal_receipt`, + and an externally stored signal payload with + `external_signal_payload_not_supported`. Updates in the completed prefix + are still refused (#864). ## 2.18.0 - 2026-11-03 diff --git a/docs/api-stability.md b/docs/api-stability.md index 690c17712..14fc9931e 100644 --- a/docs/api-stability.md +++ b/docs/api-stability.md @@ -749,10 +749,10 @@ class, source file path, or stack trace to replay or handle a failure. | `TimerScheduled` | `timer_id`, `sequence`, `delay_seconds`, `fire_at`, `timer_kind`, `condition_wait_id`, `condition_key`, `condition_definition_fingerprint`, `signal_wait_id`, `signal_name` | `WorkflowStepHistory`, `QueryStateReplayer`, `RunTimerView`, `ConditionWaits`, `SignalWaits` | | `TimerFired` | `timer_id`, `sequence`, `delay_seconds`, `fire_at`, `fired_at`, `timer_kind`, `condition_wait_id`, `condition_key`, `condition_definition_fingerprint`, `signal_wait_id`, `signal_name`, `cancellation_scope_id` (candidate 1.20), `reused_from_run_id`, `reused_recorded_at` | `WorkflowStepHistory`, `QueryStateReplayer`, `WorkflowFiberRunner`, `RunTimerView`, `ConditionWaits`, `SignalWaits`, `RedriveReplayClock` | | `TimerCancelled` | `timer_id`, `sequence`, `delay_seconds`, `fire_at`, `timer_kind`, `condition_wait_id`, `condition_key`, `condition_definition_fingerprint`, `signal_wait_id`, `signal_name`, `cancelled_at` | `WorkflowStepHistory`, `RunTimerView`, `ConditionWaits`, `SignalWaits`, `HistoryTimeline` | -| `SignalReceived` | `workflow_command_id`, `signal_id`, `workflow_instance_id`, `workflow_run_id`, `signal_name`, `signal_wait_id`, `arguments`, `payload_codec` | `SignalWaits`, `RunSignalView`, public and worker history payload consumers | -| `SignalApplied` | `workflow_command_id`, `signal_id`, `signal_name`, `signal_wait_id`, `sequence`, `value` | `WorkflowStepHistory`, `SignalWaits`, `RunSignalView`, `QueryStateReplayer` | +| `SignalReceived` | `workflow_command_id`, `signal_id`, `workflow_instance_id`, `workflow_run_id`, `signal_name`, `signal_wait_id`, `arguments`, `payload_codec`, `reused_from_run_id`, `reused_recorded_at` | `SignalWaits`, `RunSignalView`, public and worker history payload consumers, `RedriveReplayClock` | +| `SignalApplied` | `workflow_command_id`, `signal_id`, `signal_name`, `signal_wait_id`, `sequence`, `value`, `reused_from_run_id`, `reused_recorded_at` | `WorkflowStepHistory`, `SignalWaits`, `RunSignalView`, `QueryStateReplayer`, `WorkflowExecutor`, `WorkflowFiberRunner`, `RedriveReplayClock` | | `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) | `WorkflowStepHistory`, `SignalWaits`, `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 | | `UpdateRejected` | `workflow_command_id`, `update_id`, `workflow_instance_id`, `workflow_run_id`, `update_name`, `arguments`, `validation_errors` | `RunUpdateView`, `HistoryTimeline`, command-contract projections | diff --git a/src/V2/Support/DefaultWorkflowControlPlane.php b/src/V2/Support/DefaultWorkflowControlPlane.php index 183e9a28f..f4bef3648 100644 --- a/src/V2/Support/DefaultWorkflowControlPlane.php +++ b/src/V2/Support/DefaultWorkflowControlPlane.php @@ -240,6 +240,9 @@ public function redrive(string $instanceId, string $failedRunId, array $options ], ), null, $command); + // Messages the source consumed stay consumed for the successor. + MessageStreamCursor::transferCursor($source, $continuedRun); + app(MemoUpsertService::class)->copyForRedrive($source, $continuedRun); app(SearchAttributeUpsertService::class)->copyForRedrive($source, $continuedRun); @@ -250,6 +253,26 @@ public function redrive(string $instanceId, string $failedRunId, array $options ? $reused['reused_recorded_at'] : $step['recorded_at']; + if (isset($step['signal_events'])) { + // A completed signal wait: the successor records the + // wait, the receipt and the application without a signal + // row or command to deliver again. + foreach ($step['signal_events'] as $signalEvent) { + $signalPayload = $signalEvent['payload']; + WorkflowHistoryEvent::record($continuedRun, $signalEvent['event_type'], array_merge( + $signalPayload, + [ + 'reused_from_run_id' => $source->id, + 'reused_recorded_at' => is_string($signalPayload['reused_recorded_at'] ?? null) + ? $signalPayload['reused_recorded_at'] + : $signalEvent['recorded_at'], + ], + )); + } + + continue; + } + if ($step['event_type'] === HistoryEventType::TimerFired) { // The timer already fired, so the successor records it // fired with no timer row or task to wait on. diff --git a/src/V2/Support/FailedRunRedriveEligibility.php b/src/V2/Support/FailedRunRedriveEligibility.php index 08b8816d4..6f58fb67b 100644 --- a/src/V2/Support/FailedRunRedriveEligibility.php +++ b/src/V2/Support/FailedRunRedriveEligibility.php @@ -38,7 +38,7 @@ final class FailedRunRedriveEligibility * reason: string|null, * resume_step_sequence: int|null, * completed: list>, - * reused_steps: list, recorded_at: string|null, scheduled_payload?: array}> + * reused_steps: list, recorded_at: string|null, scheduled_payload?: array, signal_events?: list, recorded_at: string|null}>}> * }|null, * started_payload: array * } diff --git a/src/V2/Support/FailedRunRedrivePlan.php b/src/V2/Support/FailedRunRedrivePlan.php index d08dd3161..d32fc6353 100644 --- a/src/V2/Support/FailedRunRedrivePlan.php +++ b/src/V2/Support/FailedRunRedrivePlan.php @@ -26,7 +26,8 @@ final class FailedRunRedrivePlan * event_type: HistoryEventType, * payload: array, * recorded_at: string|null, - * scheduled_payload?: array + * scheduled_payload?: array, + * signal_events?: list, recorded_at: string|null}> * }> * } */ @@ -45,6 +46,7 @@ public static function forRun(WorkflowRun $run): array $activityEvents = []; $markerEvents = []; $timerEvents = []; + $signalEvents = []; $lastActivityEvent = null; foreach ($events as $event) { @@ -113,6 +115,26 @@ public static function forRun(WorkflowRun $run): array continue; } + if ($event->event_type === HistoryEventType::MessageCursorAdvanced) { + // Message stream bookkeeping, not a step. Redrive carries the + // source's cursor position to the successor instead. + continue; + } + + if (in_array($event->event_type, [ + HistoryEventType::SignalWaitOpened, + HistoryEventType::SignalReceived, + HistoryEventType::SignalApplied, + ], true)) { + if ($started === null) { + return self::reject('invalid_step_sequence'); + } + + $signalEvents[] = $event; + + continue; + } + if (in_array($event->event_type, [HistoryEventType::TimerScheduled, HistoryEventType::TimerFired], true)) { $payload = $event->payload; $sequence = is_array($payload) ? ($payload['sequence'] ?? null) : null; @@ -173,7 +195,17 @@ public static function forRun(WorkflowRun $run): array return self::reject('unrecorded_failure_boundary'); } - foreach ([...array_keys($markerEvents), ...array_keys($timerEvents)] as $sequence) { + $signalSteps = self::signalSteps($signalEvents, $run); + + if (is_string($signalSteps)) { + return self::reject($signalSteps); + } + + foreach ([ + ...array_keys($markerEvents), + ...array_keys($timerEvents), + ...array_keys($signalSteps), + ] as $sequence) { if ($sequence >= $boundary) { return self::reject('invalid_failed_step'); } @@ -191,7 +223,13 @@ public static function forRun(WorkflowRun $run): array } } - if (count($activityEvents) + count($markerEvents) + count($timerEvents) !== $boundary) { + foreach (array_keys($signalSteps) as $sequence) { + if (isset($activityEvents[$sequence]) || isset($markerEvents[$sequence]) || isset($timerEvents[$sequence])) { + return self::reject('duplicate_step_sequence'); + } + } + + if (count($activityEvents) + count($markerEvents) + count($timerEvents) + count($signalSteps) !== $boundary) { return self::reject('activity_sequence_gap'); } @@ -215,6 +253,14 @@ public static function forRun(WorkflowRun $run): array continue; } + $signalStep = $signalSteps[$sequence] ?? null; + + if (is_array($signalStep)) { + $reusedSteps[] = $signalStep; + + continue; + } + $timerStep = $timerEvents[$sequence] ?? null; if (is_array($timerStep)) { @@ -400,6 +446,164 @@ private static function firedTimerStep(array $events): ?array ]; } + /** + * Group signal events into completed signal-wait steps keyed by step + * sequence. Only a plain wait (no timeout, cancellation scope or parallel + * group) whose received signal was applied is reusable. A received signal + * the source never applied is refused, because the successor would lose it. + * + * The receipt must belong to the source run and instance, carry the same + * command as the application, and name the same wait. A receipt recorded + * before its wait opened has no wait id yet, which is the only allowed + * difference. External signal payloads are refused, because the successor + * gets no signal row or command that owns the stored reference. + * + * @param list $events + * + * @return array, recorded_at: string|null, signal_events: list, recorded_at: string|null}>}>|string + */ + private static function signalSteps(array $events, WorkflowRun $run): array|string + { + $opened = []; + $applied = []; + $received = []; + + foreach ($events as $event) { + $payload = is_array($event->payload) ? $event->payload : []; + $waitId = $payload['signal_wait_id'] ?? null; + $name = $payload['signal_name'] ?? null; + + if ($event->event_type === HistoryEventType::SignalReceived) { + $signalId = $payload['signal_id'] ?? null; + + if (! is_string($signalId) || $signalId === '' || isset($received[$signalId])) { + return 'unsupported_signal_wait'; + } + + $received[$signalId] = $event; + + continue; + } + + $sequence = $payload['sequence'] ?? null; + + if (! is_int($sequence) || $sequence < 1 || ! is_string($waitId) || $waitId === '' || ! is_string($name)) { + return 'invalid_step_sequence'; + } + + if ( + ($payload['timeout_seconds'] ?? null) !== null + || ($payload['cancellation_scope_id'] ?? null) !== null + || ($payload['parallel_group_id'] ?? null) !== null + ) { + return 'unsupported_signal_wait'; + } + + if ($event->event_type === HistoryEventType::SignalWaitOpened) { + if (isset($opened[$sequence])) { + return 'duplicate_step_sequence'; + } + + $opened[$sequence] = $event; + + continue; + } + + if (isset($applied[$sequence])) { + return 'duplicate_step_sequence'; + } + + $applied[$sequence] = $event; + } + + $steps = []; + $appliedSignalIds = []; + + foreach ($opened as $sequence => $wait) { + $application = $applied[$sequence] ?? null; + $applicationPayload = $application?->payload; + $signalId = is_array($applicationPayload) ? ($applicationPayload['signal_id'] ?? null) : null; + $receipt = is_string($signalId) ? ($received[$signalId] ?? null) : null; + + if ( + ! $application instanceof WorkflowHistoryEvent + || ! $receipt instanceof WorkflowHistoryEvent + || $applicationPayload['signal_wait_id'] !== $wait->payload['signal_wait_id'] + || $applicationPayload['signal_name'] !== $wait->payload['signal_name'] + ) { + return 'incomplete_signal_step'; + } + + $receiptPayload = is_array($receipt->payload) ? $receipt->payload : []; + $receiptWaitId = $receiptPayload['signal_wait_id'] ?? null; + $commandId = $applicationPayload['workflow_command_id'] ?? null; + + if ( + ($receiptPayload['signal_name'] ?? null) !== $wait->payload['signal_name'] + || ($receiptWaitId !== null && $receiptWaitId !== $wait->payload['signal_wait_id']) + || ! 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 + ) { + return 'mismatched_signal_receipt'; + } + + // One admitted signal is one workflow input: a receipt may satisfy + // only one wait. + if (isset($appliedSignalIds[$signalId])) { + return 'mismatched_signal_receipt'; + } + + // Every replay path resumes the wait with the application's value, + // so it must be recorded inline. + if (! array_key_exists('value', $applicationPayload)) { + return 'incomplete_signal_step'; + } + + if ( + self::isExternalResult($receiptPayload['arguments'] ?? null) + || self::isExternalResult($applicationPayload['value']) + ) { + return 'external_signal_payload_not_supported'; + } + + $appliedSignalIds[$signalId] = true; + $signalEvents = []; + + foreach ([$wait, $receipt, $application] as $event) { + $eventPayload = $event->payload; + // The task and command are the source run's recording context. + unset($eventPayload['task'], $eventPayload['command']); + $signalEvents[] = [ + 'event_type' => $event->event_type, + 'payload' => $eventPayload, + 'recorded_at' => $event->recorded_at?->toIso8601String(), + ]; + } + + $steps[$sequence] = [ + 'event_type' => HistoryEventType::SignalApplied, + 'payload' => $signalEvents[2]['payload'], + 'recorded_at' => $signalEvents[2]['recorded_at'], + 'signal_events' => $signalEvents, + ]; + } + + if (count($applied) !== count($opened)) { + return 'incomplete_signal_step'; + } + + foreach (array_keys($received) as $signalId) { + if (! isset($appliedSignalIds[$signalId])) { + return 'unsupported_pending_signal'; + } + } + + return $steps; + } + private static function isExternalResult(mixed $result): bool { return (is_array($result) && isset($result['external_storage'])) diff --git a/src/V2/Support/HistoryEventPayloadContract.php b/src/V2/Support/HistoryEventPayloadContract.php index 67b7c930c..08dedf88b 100644 --- a/src/V2/Support/HistoryEventPayloadContract.php +++ b/src/V2/Support/HistoryEventPayloadContract.php @@ -471,6 +471,8 @@ final class HistoryEventPayloadContract 'signal_wait_id', 'arguments', 'payload_codec', + 'reused_from_run_id', + 'reused_recorded_at', ], 'SignalRejected' => [ 'workflow_command_id', @@ -494,6 +496,8 @@ final class HistoryEventPayloadContract 'parallel_group_size', 'parallel_group_index', 'parallel_group_path', + 'reused_from_run_id', + 'reused_recorded_at', ], 'SignalWaitOpened' => [ 'signal_name', @@ -507,6 +511,8 @@ final class HistoryEventPayloadContract 'parallel_group_size', 'parallel_group_index', 'parallel_group_path', + 'reused_from_run_id', + 'reused_recorded_at', ], 'UpdateAccepted' => [ 'workflow_command_id', diff --git a/src/V2/Support/QueryStateReplayer.php b/src/V2/Support/QueryStateReplayer.php index eca404504..53e0cc2fb 100644 --- a/src/V2/Support/QueryStateReplayer.php +++ b/src/V2/Support/QueryStateReplayer.php @@ -449,7 +449,7 @@ public function replayState(WorkflowRun $run): ReplayState $this->syncWorkflowCursor($workflow, $sequence + 1); $current = $workflowExecution->send( $this->signalValue($signalEvent, $run), - $signalEvent->recorded_at, + RedriveReplayClock::recordedAt($signalEvent), ); ++$sequence; diff --git a/src/V2/Support/WorkflowExecutor.php b/src/V2/Support/WorkflowExecutor.php index 90312d2f8..ba83060f0 100644 --- a/src/V2/Support/WorkflowExecutor.php +++ b/src/V2/Support/WorkflowExecutor.php @@ -1437,7 +1437,10 @@ private function resumeSignal( $sequence = &$frame->sequence; try { $this->syncWorkflowCursor($workflow, $sequence + 1); - $current = $workflowExecution->send($this->signalValue($signalEvent, $run), $signalEvent->recorded_at); + $current = $workflowExecution->send( + $this->signalValue($signalEvent, $run), + RedriveReplayClock::recordedAt($signalEvent), + ); } catch (Throwable $throwable) { $this->failRun($run, $task, $throwable, 'workflow_run', $run->id); return null; diff --git a/src/V2/Support/WorkflowFiberRunner.php b/src/V2/Support/WorkflowFiberRunner.php index 02d9bac16..485bae2ad 100644 --- a/src/V2/Support/WorkflowFiberRunner.php +++ b/src/V2/Support/WorkflowFiberRunner.php @@ -2093,7 +2093,8 @@ private static function indexRecordedSignalOutcomes( $outcomes[$sequence] = [ 'signal_name' => $signalName, 'result' => $result, - 'recorded_at' => self::eventRecordedAt($event, $payload), + 'recorded_at' => self::eventTime($payload['reused_recorded_at'] ?? null) + ?? self::eventRecordedAt($event, $payload), ]; $resolvedKinds[$sequence] = $type === 'SignalApplied' ? 'signal_applied' diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index f59b40371..940462084 100644 --- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php +++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php @@ -378,6 +378,67 @@ public function testRedriveReusedUpsertsCorpusKeepsTheirStepsRowsAndTime(): void ]); } + public function testRedriveReusedSignalWaitCorpusKeepsItsValueAndTime(): void + { + $path = self::FIXTURE_DIR . '/redrive-reused-signal-wait-clock.json'; + if (! is_file($path)) { + $this->markTestSkipped('The redrive reused-signal evidence is absent.'); + } + $fixture = json_decode(file_get_contents($path), true, flags: JSON_THROW_ON_ERROR); + $scenario = $fixture['redrive_reused_signal_wait']; + self::stopWorkers(); + config([ + 'queue.default' => 'sync', + 'queue.connections.sync.driver' => 'sync', + ]); + foreach (['first-calls', 'second-calls'] as $counter) { + Cache::forget("test:redrive:{$counter}"); + } + $this->clearWorkflowState(); + WorkflowStub::fake(); + + $stub = WorkflowStub::make($fixture['workflow']['type'], 'regression-corpus-redrive-signal'); + $stub->start(...$fixture['workflow']['arguments']); + $stub->signal($scenario['signal_name'], $scenario['signal_value']); + $this->assertTrue($stub->refresh()->failed()); + $sourceRunId = $stub->runId(); + $sourceTimes = []; + foreach ($scenario['reused_events'] as $type) { + $sourceTimes[$type] = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $sourceRunId) + ->where('event_type', $type) + ->sole() + ->recorded_at?->toIso8601String(); + } + + $this->travel($scenario['redrive_after_minutes']) + ->minutes(); + $result = app(WorkflowControlPlane::class)->redrive('regression-corpus-redrive-signal', $sourceRunId); + + $this->assertSame($scenario['redrive'] === 'accepted', $result['accepted'], (string) $result['reason']); + $this->assertSame($scenario['resume_step_sequence'], $result['resume_step_sequence']); + $successor = WorkflowRun::query()->findOrFail($result['workflow_run_id']); + $this->assertSame($scenario['successor_status'], $successor->status->value); + $this->assertSame(1, (int) Cache::get('test:redrive:first-calls')); + $reused = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $successor->id) + ->whereIn('event_type', $scenario['reused_events']) + ->orderBy('sequence') + ->get(); + $this->assertSame( + $scenario['reused_events'], + $reused->map(static fn (WorkflowHistoryEvent $event): string => $event->event_type->value) + ->all(), + ); + foreach ($reused as $event) { + $this->assertSame($sourceRunId, $event->payload['reused_from_run_id'] ?? null); + $this->assertSame($sourceTimes[$event->event_type->value], $event->payload['reused_recorded_at'] ?? null); + } + $output = $successor->workflowOutput(); + $this->assertSame($scenario['signal_value'], $output['approved_by']); + $this->assertSame($sourceTimes['SignalApplied'], $output['time_after_signal']); + } + 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 4279b171b..7f08afd0f 100644 --- a/tests/Feature/V2/V2FailedRunRedriveTest.php +++ b/tests/Feature/V2/V2FailedRunRedriveTest.php @@ -11,7 +11,9 @@ use Tests\Fixtures\V2\TestGreetingWorkflow; use Tests\Fixtures\V2\TestRedriveMarkersWorkflow; use Tests\Fixtures\V2\TestRedrivePatchedWorkflow; +use Tests\Fixtures\V2\TestRedriveSignalWorkflow; use Tests\Fixtures\V2\TestRedriveTimerWorkflow; +use Tests\Fixtures\V2\TestRedriveTwoSignalsWorkflow; use Tests\Fixtures\V2\TestRedriveVisibilityWorkflow; use Tests\Fixtures\V2\TestRedriveWorkflow; use Tests\Fixtures\V2\TestThrowAfterGreetingWorkflow; @@ -29,11 +31,13 @@ use Workflow\V2\Models\WorkflowLink; use Workflow\V2\Models\WorkflowRun; use Workflow\V2\Models\WorkflowRunSummary; +use Workflow\V2\Models\WorkflowSignal; use Workflow\V2\Models\WorkflowTask; use Workflow\V2\Models\WorkflowTimer; use Workflow\V2\StartOptions; use Workflow\V2\Support\DefaultWorkflowTaskBridge; use Workflow\V2\Support\EmbeddedV2HistoryImport; +use Workflow\V2\Support\FailedRunRedrivePlan; use Workflow\V2\Support\HistoryExport; use Workflow\V2\Support\RunActivityView; use Workflow\V2\Support\RunDetailView; @@ -627,6 +631,310 @@ public function testARedriveSuccessorWithUpsertsCanBeRedrivenAgain(): void $this->assertSame($memoRecordedAt, $thirdRun->workflowOutput()['time_after_memo']); } + public function testRedriveReusesAnAppliedSignalWaitAtItsRecordedTime(): void + { + WorkflowStub::fake(); + + $workflow = WorkflowStub::make(TestRedriveSignalWorkflow::class, 'redrive-signal-1'); + $workflow->start('Taylor'); + $workflow->signal('approve', 'Grace'); + $this->assertTrue($workflow->refresh()->failed()); + $sourceRunId = $workflow->runId(); + $sourceEvents = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $sourceRunId) + ->whereIn('event_type', [ + HistoryEventType::SignalWaitOpened, + HistoryEventType::SignalReceived, + HistoryEventType::SignalApplied, + ]) + ->orderBy('sequence') + ->get(); + $this->assertCount(3, $sourceEvents); + $sourceApplied = $sourceEvents->firstWhere('event_type', HistoryEventType::SignalApplied); + + $this->travel(5) + ->minutes(); + $result = app(WorkflowControlPlane::class)->redrive('redrive-signal-1', $sourceRunId); + + $this->assertTrue($result['accepted'], (string) $result['reason']); + $this->assertSame(3, $result['resume_step_sequence']); + $successor = WorkflowRun::query()->findOrFail($result['workflow_run_id']); + $this->assertSame(RunStatus::Completed, $successor->status); + $this->assertSame(1, (int) Cache::get('test:redrive:first-calls')); + + $reused = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $successor->id) + ->whereIn('event_type', [ + HistoryEventType::SignalWaitOpened, + HistoryEventType::SignalReceived, + HistoryEventType::SignalApplied, + ]) + ->orderBy('sequence') + ->get(); + $this->assertSame( + ['SignalWaitOpened', 'SignalReceived', 'SignalApplied'], + $reused->map(static fn (WorkflowHistoryEvent $event): string => $event->event_type->value) + ->all(), + ); + foreach ($reused as $event) { + $original = $sourceEvents->firstWhere('event_type', $event->event_type); + $this->assertSame($sourceRunId, $event->payload['reused_from_run_id'] ?? null); + $this->assertSame( + $original->recorded_at?->toIso8601String(), + $event->payload['reused_recorded_at'] ?? null, + ); + } + $this->assertSame(0, WorkflowSignal::query()->where('workflow_run_id', $successor->id)->count()); + // The consumed message stays consumed for the successor. + $this->assertGreaterThan(0, (int) WorkflowRun::query()->findOrFail($sourceRunId)->message_cursor_position); + $this->assertSame( + (int) WorkflowRun::query()->findOrFail($sourceRunId)->message_cursor_position, + (int) $successor->message_cursor_position, + ); + + $expectedTime = $sourceApplied->recorded_at?->toIso8601String(); + $output = $successor->workflowOutput(); + $this->assertSame('Hello, Taylor! Redrive completed.', $output['result']); + $this->assertSame('Grace', $output['approved_by']); + $this->assertSame($expectedTime, $output['time_after_signal']); + $export = json_decode( + json_encode(HistoryExport::forRun($successor->fresh()), JSON_THROW_ON_ERROR), + true, + 512, + JSON_THROW_ON_ERROR, + ); + $this->assertSame($expectedTime, (new WorkflowReplayer())->replayExport($export)->workflow->timeAfterSignal()); + $this->assertSame( + $expectedTime, + (new WorkflowReplayer())->replay($successor->fresh()) + ->workflow->timeAfterSignal(), + ); + } + + public function testRedriveRefusesASignalTheSourceReceivedButNeverApplied(): void + { + WorkflowStub::fake(); + + $workflow = WorkflowStub::make(TestRedriveSignalWorkflow::class, 'redrive-signal-pending'); + $workflow->start('Taylor'); + $workflow->signal('approve', 'Grace'); + $this->assertTrue($workflow->refresh()->failed()); + $source = WorkflowRun::query()->findOrFail($workflow->runId()); + $applied = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $source->id) + ->where('event_type', HistoryEventType::SignalReceived) + ->sole(); + // A second receipt the workflow never consumed. + $receipt = $applied->payload; + unset($receipt['task'], $receipt['command']); + WorkflowHistoryEvent::record($source, HistoryEventType::SignalReceived, array_merge($receipt, [ + 'signal_id' => 'never-applied-signal', + ])); + // Place the receipt before the terminal failure, as a buffered signal. + $failure = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $source->id) + ->where('event_type', HistoryEventType::WorkflowFailed) + ->sole(); + $failureSequence = $failure->sequence; + $pending = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $source->id) + ->where('event_type', HistoryEventType::SignalReceived) + ->orderByDesc('sequence') + ->first(); + $failure->forceFill([ + 'sequence' => $pending->sequence + 1, + ])->save(); + $pending->forceFill([ + 'sequence' => $failureSequence, + ])->save(); + + $result = app(WorkflowControlPlane::class)->redrive('redrive-signal-pending', $source->id); + + $this->assertFalse($result['accepted']); + $this->assertSame('unsupported_pending_signal', $result['reason']); + $this->assertSame(1, WorkflowRun::query()->where('workflow_instance_id', 'redrive-signal-pending')->count()); + } + + /** + * @return iterable + */ + public static function signalStepMutations(): iterable + { + $external = [ + 'codec' => 'avro', + 'external_storage' => [ + 'uri' => 's3://bucket/signal-payload', + 'sha256' => str_repeat('a', 64), + 'size_bytes' => 4096, + 'codec' => 'avro', + ], + ]; + + yield 'receipt for another wait' => [ + HistoryEventType::SignalReceived, 'signal_wait_id', 'another-wait', 'mismatched_signal_receipt', + ]; + yield 'receipt for another command' => [ + HistoryEventType::SignalReceived, 'workflow_command_id', 'another-command', 'mismatched_signal_receipt', + ]; + yield 'receipt for another run' => [ + HistoryEventType::SignalReceived, 'workflow_run_id', 'another-run', 'mismatched_signal_receipt', + ]; + yield 'receipt for another instance' => [ + HistoryEventType::SignalReceived, 'workflow_instance_id', 'another-instance', 'mismatched_signal_receipt', + ]; + yield 'external receipt arguments' => [ + HistoryEventType::SignalReceived, 'arguments', $external, 'external_signal_payload_not_supported', + ]; + yield 'external applied value' => [ + HistoryEventType::SignalApplied, 'value', $external, 'external_signal_payload_not_supported', + ]; + } + + #[\PHPUnit\Framework\Attributes\DataProvider('signalStepMutations')] + public function testRedriveRefusesAMismatchedOrExternalSignalStepWithoutMutation( + HistoryEventType $type, + string $key, + mixed $value, + string $reason, + ): void { + WorkflowStub::fake(); + + $workflow = WorkflowStub::make(TestRedriveSignalWorkflow::class, 'redrive-signal-mutation'); + $workflow->start('Taylor'); + $workflow->signal('approve', 'Grace'); + $this->assertTrue($workflow->refresh()->failed()); + $source = WorkflowRun::query()->findOrFail($workflow->runId()); + $event = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $source->id) + ->where('event_type', $type) + ->sole(); + $event->forceFill([ + 'payload' => array_merge($event->payload, [ + $key => $value, + ]), + ])->save(); + $cursor = (int) $source->fresh() +->message_cursor_position; + $events = WorkflowHistoryEvent::query()->count(); + + $result = app(WorkflowControlPlane::class)->redrive('redrive-signal-mutation', $source->id); + + $this->assertFalse($result['accepted']); + $this->assertSame($reason, $result['reason']); + $this->assertSame(1, WorkflowRun::query()->where('workflow_instance_id', 'redrive-signal-mutation')->count()); + $this->assertSame($cursor, (int) $source->fresh()->message_cursor_position); + $this->assertSame($events, WorkflowHistoryEvent::query()->count()); + $this->assertSame($source->id, $source->instance->fresh()->current_run_id); + } + + public function testRedriveRefusesOneBufferedReceiptAppliedToTwoWaits(): void + { + WorkflowStub::fake(); + + $workflow = WorkflowStub::make(TestRedriveTwoSignalsWorkflow::class, 'redrive-signal-twice'); + $workflow->start('Taylor'); + $workflow->signal('approve', 'Grace'); + $workflow->signal('approve', 'Ada'); + $this->assertTrue($workflow->refresh()->failed()); + $source = WorkflowRun::query()->findOrFail($workflow->runId()); + // The unmodified two-signal source is redrivable. + $this->assertTrue(FailedRunRedrivePlan::forRun($source->fresh())['eligible']); + $receipts = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $source->id) + ->where('event_type', HistoryEventType::SignalReceived) + ->orderBy('sequence') + ->get(); + $applications = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $source->id) + ->where('event_type', HistoryEventType::SignalApplied) + ->orderBy('sequence') + ->get(); + // The first receipt was buffered, and the second application claims it + // too. The second receipt no longer exists, so nothing is pending. + $receipts[0]->forceFill([ + 'payload' => array_merge($receipts[0]->payload, [ + 'signal_wait_id' => null, + ]), + ])->save(); + $applications[1]->forceFill([ + 'payload' => array_merge($applications[1]->payload, [ + 'signal_id' => $receipts[0]->payload['signal_id'], + 'workflow_command_id' => $receipts[0]->payload['workflow_command_id'], + ]), + ])->save(); + $receipts[1]->delete(); + + $this->assertRefusedWithoutMutation('redrive-signal-twice', $source, 'mismatched_signal_receipt'); + } + + public function testRedriveRefusesASignalApplicationWithoutItsValue(): void + { + WorkflowStub::fake(); + + $source = $this->failedSignalSource('redrive-signal-no-value'); + $application = $this->sourceEvent($source, HistoryEventType::SignalApplied); + $payload = $application->payload; + unset($payload['value']); + $application->forceFill([ + 'payload' => $payload, + ])->save(); + + $this->assertRefusedWithoutMutation('redrive-signal-no-value', $source, 'incomplete_signal_step'); + } + + public function testRedriveAcceptsASignalReceivedBeforeItsWaitOpened(): void + { + WorkflowStub::fake(); + + $workflow = WorkflowStub::make(TestRedriveSignalWorkflow::class, 'redrive-signal-buffered'); + $workflow->start('Taylor'); + $workflow->signal('approve', 'Grace'); + $this->assertTrue($workflow->refresh()->failed()); + $receipt = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $workflow->runId()) + ->where('event_type', HistoryEventType::SignalReceived) + ->sole(); + // A signal that arrived before the wait opened has no wait id yet. + $receipt->forceFill([ + 'payload' => array_merge($receipt->payload, [ + 'signal_wait_id' => null, + ]), + ])->save(); + + $result = app(WorkflowControlPlane::class)->redrive('redrive-signal-buffered', $workflow->runId()); + + $this->assertTrue($result['accepted'], (string) $result['reason']); + $this->assertSame( + 'Grace', + WorkflowRun::query()->findOrFail($result['workflow_run_id'])->workflowOutput()['approved_by'] + ); + } + + public function testRedriveRefusesASignalWaitWithATimeout(): void + { + WorkflowStub::fake(); + + $workflow = WorkflowStub::make(TestRedriveSignalWorkflow::class, 'redrive-signal-timeout'); + $workflow->start('Taylor'); + $workflow->signal('approve', 'Grace'); + $this->assertTrue($workflow->refresh()->failed()); + $wait = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $workflow->runId()) + ->where('event_type', HistoryEventType::SignalWaitOpened) + ->sole(); + // Its timeout timer and race are not reused yet. + $wait->forceFill([ + 'payload' => array_merge($wait->payload, [ + 'timeout_seconds' => 60, + ]), + ])->save(); + + $result = app(WorkflowControlPlane::class)->redrive('redrive-signal-timeout', $workflow->runId()); + + $this->assertFalse($result['accepted']); + $this->assertSame('unsupported_signal_wait', $result['reason']); + } + public function testRedriveReusesOneMarkerForARepeatedPatchCall(): void { WorkflowStub::fake(); @@ -971,6 +1279,41 @@ public function testFailedSourceAndRedriveSuccessorSurviveExportImport(): void ], json_decode($coldReplay->getOutput(), true, flags: JSON_THROW_ON_ERROR)); } + private function failedSignalSource(string $instanceId): WorkflowRun + { + $workflow = WorkflowStub::make(TestRedriveSignalWorkflow::class, $instanceId); + $workflow->start('Taylor'); + $workflow->signal('approve', 'Grace'); + $this->assertTrue($workflow->refresh()->failed()); + + return WorkflowRun::query()->findOrFail($workflow->runId()); + } + + private function sourceEvent(WorkflowRun $source, HistoryEventType $type): WorkflowHistoryEvent + { + return WorkflowHistoryEvent::query() + ->where('workflow_run_id', $source->id) + ->where('event_type', $type) + ->orderBy('sequence') + ->firstOrFail(); + } + + private function assertRefusedWithoutMutation(string $instanceId, WorkflowRun $source, string $reason): void + { + $cursor = (int) $source->fresh() +->message_cursor_position; + $events = WorkflowHistoryEvent::query()->count(); + + $result = app(WorkflowControlPlane::class)->redrive($instanceId, $source->id); + + $this->assertFalse($result['accepted']); + $this->assertSame($reason, $result['reason']); + $this->assertSame(1, WorkflowRun::query()->where('workflow_instance_id', $instanceId)->count()); + $this->assertSame($cursor, (int) $source->fresh()->message_cursor_position); + $this->assertSame($events, WorkflowHistoryEvent::query()->count()); + $this->assertSame($source->id, $source->instance->fresh()->current_run_id); + } + private function assertRedriveAction( WorkflowControlPlane $controlPlane, string $instanceId, diff --git a/tests/Fixtures/V2/ReplayRegression/redrive-reused-signal-wait-clock.json b/tests/Fixtures/V2/ReplayRegression/redrive-reused-signal-wait-clock.json new file mode 100644 index 000000000..c0c435d95 --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/redrive-reused-signal-wait-clock.json @@ -0,0 +1,92 @@ +{ + "$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.0", + "bindings": [ + "php" + ], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestRedriveSignalWorkflow", + "arguments": [ + "Taylor" + ], + "payload_codec": "avro" + }, + "redrive_reused_signal_wait": { + "signal_name": "approve", + "signal_value": "Grace", + "redrive_after_minutes": 5, + "redrive": "accepted", + "resume_step_sequence": 3, + "successor_status": "completed", + "reused_events": [ + "SignalWaitOpened", + "SignalReceived", + "SignalApplied" + ] + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": {} + }, + { + "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": { + "completed": false, + "result": null, + "commands": [ + { + "type": "schedule_activity", + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "payload_codec": "avro" + } + ] + } +} diff --git a/tests/Fixtures/V2/TestRedriveSignalWorkflow.php b/tests/Fixtures/V2/TestRedriveSignalWorkflow.php new file mode 100644 index 000000000..6d179fd7a --- /dev/null +++ b/tests/Fixtures/V2/TestRedriveSignalWorkflow.php @@ -0,0 +1,43 @@ +timeAfterSignal = now() + ->toIso8601String(); + $greeting = activity(TestRedriveFirstActivity::class, $name); + + return [ + 'result' => activity(TestRedriveSecondActivity::class, $greeting), + 'approved_by' => $approvedBy, + 'time_after_signal' => $this->timeAfterSignal, + ]; + } + + #[QueryMethod] + public function timeAfterSignal(): ?string + { + return $this->timeAfterSignal; + } +} diff --git a/tests/Fixtures/V2/TestRedriveTwoSignalsWorkflow.php b/tests/Fixtures/V2/TestRedriveTwoSignalsWorkflow.php new file mode 100644 index 000000000..29d5bbd36 --- /dev/null +++ b/tests/Fixtures/V2/TestRedriveTwoSignalsWorkflow.php @@ -0,0 +1,30 @@ +} + */ + public function handle(string $name): array + { + $approvals = [await('approve'), await('approve')]; + $greeting = activity(TestRedriveFirstActivity::class, $name); + + return [ + 'result' => activity(TestRedriveSecondActivity::class, $greeting), + 'approvals' => $approvals, + ]; + } +}