From 5cfd3df4151a76466297423d0af70b5c3b1a5e63 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 11:57:18 -0600 Subject: [PATCH 1/3] fix: redrive runs whose completed steps include an applied signal wait (#864) A2c (first slice) of #864. A SignalWaitOpened, SignalReceived or SignalApplied event in the completed prefix refused redrive with unsupported_history. - The plan groups a plain signal wait (no timeout, cancellation scope or parallel group) with its receipt and application into one reused step. The successor records all three with reused_from_run_id and reused_recorded_at, and the executor, query replayer and fiber runner resume the wait at the original time with the recorded value. - MessageCursorAdvanced is bookkeeping, not a step. Redrive now transfers the source's message cursor, as continue-as-new does, so a consumed message is not delivered again. - New refusals: unsupported_signal_wait, unsupported_pending_signal (a received signal the source never applied) and incomplete_signal_step. Updates in the prefix are still refused. Adds feature coverage and replay corpus evidence redrive-reused-signal-wait-clock. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 10 ++ docs/api-stability.md | 6 +- .../Support/DefaultWorkflowControlPlane.php | 23 +++ .../Support/FailedRunRedriveEligibility.php | 2 +- src/V2/Support/FailedRunRedrivePlan.php | 170 +++++++++++++++++- .../Support/HistoryEventPayloadContract.php | 6 + src/V2/Support/QueryStateReplayer.php | 2 +- src/V2/Support/WorkflowExecutor.php | 5 +- src/V2/Support/WorkflowFiberRunner.php | 3 +- .../V2EmbeddedReplayRegressionCorpusTest.php | 61 +++++++ tests/Feature/V2/V2FailedRunRedriveTest.php | 151 ++++++++++++++++ .../redrive-reused-signal-wait-clock.json | 92 ++++++++++ .../Fixtures/V2/TestRedriveSignalWorkflow.php | 43 +++++ 13 files changed, 564 insertions(+), 10 deletions(-) create mode 100644 tests/Fixtures/V2/ReplayRegression/redrive-reused-signal-wait-clock.json create mode 100644 tests/Fixtures/V2/TestRedriveSignalWorkflow.php diff --git a/CHANGELOG.md b/CHANGELOG.md index d069d9532..6882dee82 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -47,6 +47,16 @@ 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`. + Updates in the completed prefix are still refused (#864). ## 2.17.1 - 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..7523620e8 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); + + 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,124 @@ 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. + * + * @param list $events + * + * @return array, recorded_at: string|null, signal_events: list, recorded_at: string|null}>}>|string + */ + private static function signalSteps(array $events): 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'] + || ($receipt->payload['signal_name'] ?? null) !== $wait->payload['signal_name'] + ) { + return 'incomplete_signal_step'; + } + + $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 12c60aa97..4f8294d32 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..a01a112e4 100644 --- a/tests/Feature/V2/V2FailedRunRedriveTest.php +++ b/tests/Feature/V2/V2FailedRunRedriveTest.php @@ -11,6 +11,7 @@ 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\TestRedriveVisibilityWorkflow; use Tests\Fixtures\V2\TestRedriveWorkflow; @@ -29,6 +30,7 @@ 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; @@ -627,6 +629,155 @@ 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()); + } + + 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(); 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; + } +} From 73ce17d0b132572fd14a2f75fc10106594c8410a Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 12:29:52 -0600 Subject: [PATCH 2/3] fix: bind reused signal receipts to their source and refuse external payloads (#864) Security pre-review of #1002 found two gaps. - A reused SignalReceived must belong to the source run and instance, carry the same workflow_command_id as its SignalApplied, and name the same wait. A receipt recorded before its wait opened has no wait id, which is the only allowed difference. Mismatches refuse with mismatched_signal_receipt. - An externally stored SignalReceived.arguments or SignalApplied.value refuses with external_signal_payload_not_supported, because the successor gets no signal row or command that owns the reference. Mutation tests cover wait, command, run and instance mismatches and both external payloads; each refuses with no successor, no history write and no cursor change. A buffered receipt with a null wait id is accepted. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 5 +- src/V2/Support/FailedRunRedrivePlan.php | 34 ++++++- tests/Feature/V2/V2FailedRunRedriveTest.php | 100 ++++++++++++++++++++ 3 files changed, 135 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6882dee82..9ec73d9c5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -56,7 +56,10 @@ 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`. - Updates in the completed prefix are still refused (#864). + 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.17.1 - 2026-11-03 diff --git a/src/V2/Support/FailedRunRedrivePlan.php b/src/V2/Support/FailedRunRedrivePlan.php index 7523620e8..6eae7c4c3 100644 --- a/src/V2/Support/FailedRunRedrivePlan.php +++ b/src/V2/Support/FailedRunRedrivePlan.php @@ -195,7 +195,7 @@ public static function forRun(WorkflowRun $run): array return self::reject('unrecorded_failure_boundary'); } - $signalSteps = self::signalSteps($signalEvents); + $signalSteps = self::signalSteps($signalEvents, $run); if (is_string($signalSteps)) { return self::reject($signalSteps); @@ -452,11 +452,17 @@ private static function firedTimerStep(array $events): ?array * 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): array|string + private static function signalSteps(array $events, WorkflowRun $run): array|string { $opened = []; $applied = []; @@ -524,11 +530,33 @@ private static function signalSteps(array $events): array|string || ! $receipt instanceof WorkflowHistoryEvent || $applicationPayload['signal_wait_id'] !== $wait->payload['signal_wait_id'] || $applicationPayload['signal_name'] !== $wait->payload['signal_name'] - || ($receipt->payload['signal_name'] ?? null) !== $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'; + } + + if ( + self::isExternalResult($receiptPayload['arguments'] ?? null) + || self::isExternalResult($applicationPayload['value'] ?? null) + ) { + return 'external_signal_payload_not_supported'; + } + $appliedSignalIds[$signalId] = true; $signalEvents = []; diff --git a/tests/Feature/V2/V2FailedRunRedriveTest.php b/tests/Feature/V2/V2FailedRunRedriveTest.php index a01a112e4..40a2edd5d 100644 --- a/tests/Feature/V2/V2FailedRunRedriveTest.php +++ b/tests/Feature/V2/V2FailedRunRedriveTest.php @@ -753,6 +753,106 @@ public function testRedriveRefusesASignalTheSourceReceivedButNeverApplied(): voi $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 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(); From 959512d61686cf03df8f78fc8a88cf56f5d61f7a Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 12:39:42 -0600 Subject: [PATCH 3/3] fix: let one signal receipt satisfy one wait and require the applied value (#864) Security full review of #1002 found two more fail-opens. - A buffered receipt (null wait id) could satisfy two distinct waits, turning one admitted signal into two workflow inputs. Each receipt now binds to exactly one application; a second refuses with mismatched_signal_receipt. - A SignalApplied without its inline value was accepted, but the embedded executor resumes from that value only, so the successor would see null. A missing value now refuses with incomplete_signal_step. Counterfactuals: a two-signal source whose second application reuses the first buffered receipt, and an application without its value. Both are accepted by the previous plan and now refuse with no successor, history or cursor change. Co-Authored-By: Claude Opus 5.5 --- src/V2/Support/FailedRunRedrivePlan.php | 14 ++- tests/Feature/V2/V2FailedRunRedriveTest.php | 92 +++++++++++++++++++ .../V2/TestRedriveTwoSignalsWorkflow.php | 30 ++++++ 3 files changed, 135 insertions(+), 1 deletion(-) create mode 100644 tests/Fixtures/V2/TestRedriveTwoSignalsWorkflow.php diff --git a/src/V2/Support/FailedRunRedrivePlan.php b/src/V2/Support/FailedRunRedrivePlan.php index 6eae7c4c3..d32fc6353 100644 --- a/src/V2/Support/FailedRunRedrivePlan.php +++ b/src/V2/Support/FailedRunRedrivePlan.php @@ -550,9 +550,21 @@ private static function signalSteps(array $events, WorkflowRun $run): array|stri 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'] ?? null) + || self::isExternalResult($applicationPayload['value']) ) { return 'external_signal_payload_not_supported'; } diff --git a/tests/Feature/V2/V2FailedRunRedriveTest.php b/tests/Feature/V2/V2FailedRunRedriveTest.php index 40a2edd5d..7f08afd0f 100644 --- a/tests/Feature/V2/V2FailedRunRedriveTest.php +++ b/tests/Feature/V2/V2FailedRunRedriveTest.php @@ -13,6 +13,7 @@ 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; @@ -36,6 +37,7 @@ 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; @@ -825,6 +827,61 @@ public function testRedriveRefusesAMismatchedOrExternalSignalStepWithoutMutation $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(); @@ -1222,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/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, + ]; + } +}