From 186d14c357026e4c74ee8cf86e8cad58032078bf Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 11:35:43 -0600 Subject: [PATCH] fix: redrive runs whose workflow code upserted memos or search attributes (#864) A2d of #864. Any MemoUpserted or SearchAttributesUpserted event in the completed prefix refused redrive with unsupported_history. - The redrive plan treats both upsert events as reusable steps, like side effects and version markers. The successor records them with reused_from_run_id and reused_recorded_at, and the executor, query replayer and fiber runner replay them at the original time. - Memo and search attribute rows carry to the successor unchanged, keeping upserted_at_sequence and inherited_from_parent, so the successor's rows match its reused history and a successor that fails again can be redriven again. - A row written by workflow code with no matching reused upsert step is still refused with unsupported_visibility_metadata. Adds feature coverage (single redrive and a redrive chain) and replay corpus evidence redrive-reused-upserts-clock. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 13 +- docs/api-stability.md | 4 +- .../Support/DefaultWorkflowControlPlane.php | 4 +- .../Support/FailedRunRedriveEligibility.php | 50 ++++-- src/V2/Support/FailedRunRedrivePlan.php | 2 + .../Support/HistoryEventPayloadContract.php | 11 +- src/V2/Support/MemoUpsertService.php | 31 ++++ src/V2/Support/QueryStateReplayer.php | 4 +- .../Support/SearchAttributeUpsertService.php | 34 ++++ src/V2/Support/WorkflowExecutor.php | 12 +- src/V2/Support/WorkflowFiberRunner.php | 6 +- .../V2EmbeddedReplayRegressionCorpusTest.php | 62 ++++++++ tests/Feature/V2/V2FailedRunRedriveTest.php | 145 ++++++++++++++++++ .../redrive-reused-upserts-clock.json | 95 ++++++++++++ .../V2/TestRedriveVisibilityWorkflow.php | 58 +++++++ 15 files changed, 508 insertions(+), 23 deletions(-) create mode 100644 tests/Fixtures/V2/ReplayRegression/redrive-reused-upserts-clock.json create mode 100644 tests/Fixtures/V2/TestRedriveVisibilityWorkflow.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 5ecbc6b74..1fbb5b1c8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,8 +5,7 @@ - Redrive a failed run that has start-time memos or search attributes. Before, any memo or search attribute row refused redrive with `unsupported_visibility_metadata`. The successor now inherits the start - values, as continue-as-new does. A value upserted by workflow code is still - refused until redrive replays that part of the history (#864). + values, as continue-as-new does (#864). - Run detail and describe report whether redrive would accept a run. `can_redrive` and `redrive_blocked_reason` come from the same decision `redrive()` applies, so the action an operator sees matches what redrive @@ -29,6 +28,16 @@ timers in a cancellation scope or parallel group are refused with `unsupported_timer`. A timer step without both its scheduled and fired events is refused with `incomplete_timer_step` (#864). +- Redrive a failed run whose workflow code upserted memos or search + attributes. Before, any `MemoUpserted` or `SearchAttributesUpserted` event + refused redrive with `unsupported_history`. The successor reuses those + events in step order, carrying `reused_from_run_id` and + `reused_recorded_at`, and replay uses the original time. Memo and search + attribute rows now carry to the successor unchanged, keeping the step that + wrote them and whether they were inherited, so a successor that fails again + can be redriven again. A row written by workflow + code with no matching reused upsert step is still refused with + `unsupported_visibility_metadata` (#864). ## 2.17.1 - 2026-11-03 diff --git a/docs/api-stability.md b/docs/api-stability.md index 6c3066dd1..441117bef 100644 --- a/docs/api-stability.md +++ b/docs/api-stability.md @@ -748,8 +748,8 @@ class, source file path, or stack trace to replay or handle a failure. | `ServiceCallCompleted` | `sequence`, `service_call_id`, `workflow_instance_id`, `workflow_run_id`, `caller_workflow_instance_id`, `caller_workflow_run_id`, `caller_sdk_language`, `endpoint_name`, `service_name`, `operation_name`, `service_sdk_language`, `request_payload`, `response_payload`, `payload_codec`, `operation_mode`, `wait_for`, `status`, `outcome`, `resolved_binding_kind`, `resolved_target_reference`, `linked_workflow_instance_id`, `linked_workflow_run_id`, `linked_workflow_update_id`, `service_call`, `response_or_failure_surface` | `WorkflowStepHistory`, `WorkflowExecutor`, `WorkflowFiberRunner`, `QueryStateReplayer`, Nexus observability | | `ServiceCallFailed` | `sequence`, `service_call_id`, `workflow_instance_id`, `workflow_run_id`, `caller_workflow_instance_id`, `caller_workflow_run_id`, `caller_sdk_language`, `endpoint_name`, `service_name`, `operation_name`, `service_sdk_language`, `request_payload`, `response_payload`, `payload_codec`, `operation_mode`, `wait_for`, `status`, `outcome`, `resolved_binding_kind`, `resolved_target_reference`, `linked_workflow_instance_id`, `linked_workflow_run_id`, `linked_workflow_update_id`, `service_call`, `response_or_failure_surface`, `exception_type`, `exception_class`, `message`, `code`, `exception` | `WorkflowStepHistory`, `WorkflowExecutor`, `WorkflowFiberRunner`, `QueryStateReplayer`, typed service failure propagation | | `ServiceCallCancelled` | `sequence`, `service_call_id`, `workflow_instance_id`, `workflow_run_id`, `caller_workflow_instance_id`, `caller_workflow_run_id`, `caller_sdk_language`, `endpoint_name`, `service_name`, `operation_name`, `service_sdk_language`, `request_payload`, `response_payload`, `payload_codec`, `operation_mode`, `wait_for`, `status`, `outcome`, `resolved_binding_kind`, `resolved_target_reference`, `linked_workflow_instance_id`, `linked_workflow_run_id`, `linked_workflow_update_id`, `service_call`, `response_or_failure_surface`, `exception_type`, `exception_class`, `message`, `code`, `exception` | `WorkflowStepHistory`, `WorkflowExecutor`, `WorkflowFiberRunner`, `QueryStateReplayer`, typed service failure propagation | -| `SearchAttributesUpserted` | `sequence`, `attributes`, `merged` | `WorkflowStepHistory`, `HistoryTimeline`, visibility/search projections, history export | -| `MemoUpserted` | `sequence`, `entries`, `merged` | `WorkflowStepHistory`, `HistoryTimeline`, run detail projections, history export | +| `SearchAttributesUpserted` | `sequence`, `attributes`, `attribute_types`, `merged`, `reused_from_run_id`, `reused_recorded_at` | `WorkflowStepHistory`, `WorkflowExecutor`, `QueryStateReplayer`, `WorkflowFiberRunner`, `HistoryTimeline`, visibility/search projections, history export, `RedriveReplayClock` | +| `MemoUpserted` | `sequence`, `entries`, `merged`, `reused_from_run_id`, `reused_recorded_at` | `WorkflowStepHistory`, `WorkflowExecutor`, `QueryStateReplayer`, `WorkflowFiberRunner`, `HistoryTimeline`, run detail projections, history export, `RedriveReplayClock` | | `RepairRequested` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `command_type`, `outcome`, `liveness_state`, `wait_kind`, `task_id`, `task_type` | `HistoryTimeline`, `RunCommandContract`, repair diagnostics, operator detail projections | | `CancelRequested` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `command_type`, `reason` | `HistoryTimeline`, `RunCommandContract`, cancellation projections | | `WorkflowCancelled` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `failure_id`, `failure_category`, `closed_reason`, `exception_class`, `message`, `reason`, `cancellation_cleanup` | `HistoryTimeline`, `FailureSnapshots`, `ChildRunHistory`, cancellation projections | diff --git a/src/V2/Support/DefaultWorkflowControlPlane.php b/src/V2/Support/DefaultWorkflowControlPlane.php index 958d633fb..e02fbf049 100644 --- a/src/V2/Support/DefaultWorkflowControlPlane.php +++ b/src/V2/Support/DefaultWorkflowControlPlane.php @@ -239,8 +239,8 @@ public function redrive(string $instanceId, string $failedRunId, array $options ], ), null, $command); - app(MemoUpsertService::class)->inheritFromParent($source, $continuedRun, 1); - app(SearchAttributeUpsertService::class)->inheritFromParent($source, $continuedRun, 1); + app(MemoUpsertService::class)->copyForRedrive($source, $continuedRun); + app(SearchAttributeUpsertService::class)->copyForRedrive($source, $continuedRun); foreach ($plan['reused_steps'] as $step) { $reused = $step['payload']; diff --git a/src/V2/Support/FailedRunRedriveEligibility.php b/src/V2/Support/FailedRunRedriveEligibility.php index a7dbb1a50..08b8816d4 100644 --- a/src/V2/Support/FailedRunRedriveEligibility.php +++ b/src/V2/Support/FailedRunRedriveEligibility.php @@ -59,16 +59,22 @@ public static function evaluate( return self::refuse((string) $plan['reason']); } - // Start-time memos and search attributes carry to the successor. - // Values upserted by workflow code belong to the replayed prefix, - // which redrive does not copy yet. + // Memo and search attribute rows carry to the successor as they are. + // A value written by workflow code must come from an upsert step the + // successor reuses, or its replay would not explain the row. if ( - $source->memos() - ->where('upserted_at_sequence', '>', 0) - ->exists() - || $source->searchAttributes() - ->where('upserted_at_sequence', '>', 0) - ->exists() + ! self::rowsMatchReusedUpserts( + $source->memos() + ->get(), + $plan['reused_steps'], + HistoryEventType::MemoUpserted, + ) + || ! self::rowsMatchReusedUpserts( + $source->searchAttributes() + ->get(), + $plan['reused_steps'], + HistoryEventType::SearchAttributesUpserted, + ) ) { return self::refuse('unsupported_visibility_metadata'); } @@ -134,6 +140,32 @@ public static function blockedReason(WorkflowRun $run, ?WorkflowRun $currentRun) return self::evaluate($run, $currentRun, fingerprintPresentedAtRedrive: true)['reason']; } + /** + * @param iterable<\Workflow\V2\Models\WorkflowMemo|\Workflow\V2\Models\WorkflowSearchAttribute> $rows + * @param list, recorded_at: string|null}> $reusedSteps + */ + private static function rowsMatchReusedUpserts(iterable $rows, array $reusedSteps, HistoryEventType $type): bool + { + $upsertSequences = []; + foreach ($reusedSteps as $step) { + $sequence = $step['payload']['sequence'] ?? null; + if ($step['event_type'] === $type && is_int($sequence)) { + $upsertSequences[$sequence] = true; + } + } + + foreach ($rows as $row) { + $sequence = (int) $row->upserted_at_sequence; + // Start-time values (step 0) and values inherited from an earlier + // run in the chain are not written by this run's workflow code. + if ($sequence > 0 && ! $row->inherited_from_parent && ! isset($upsertSequences[$sequence])) { + return false; + } + } + + return true; + } + private static function definitionMatches( WorkflowRun $source, mixed $sourceFingerprint, diff --git a/src/V2/Support/FailedRunRedrivePlan.php b/src/V2/Support/FailedRunRedrivePlan.php index 066fadddc..d08dd3161 100644 --- a/src/V2/Support/FailedRunRedrivePlan.php +++ b/src/V2/Support/FailedRunRedrivePlan.php @@ -87,6 +87,8 @@ public static function forRun(WorkflowRun $run): array if (in_array($event->event_type, [ HistoryEventType::SideEffectRecorded, HistoryEventType::VersionMarkerRecorded, + HistoryEventType::MemoUpserted, + HistoryEventType::SearchAttributesUpserted, ], true)) { $payload = $event->payload; $sequence = is_array($payload) ? ($payload['sequence'] ?? null) : null; diff --git a/src/V2/Support/HistoryEventPayloadContract.php b/src/V2/Support/HistoryEventPayloadContract.php index 4d6ddde73..67b7c930c 100644 --- a/src/V2/Support/HistoryEventPayloadContract.php +++ b/src/V2/Support/HistoryEventPayloadContract.php @@ -908,8 +908,15 @@ final class HistoryEventPayloadContract 'code', 'exception', ], - 'SearchAttributesUpserted' => ['sequence', 'attributes', 'attribute_types', 'merged'], - 'MemoUpserted' => ['sequence', 'entries', 'merged'], + 'SearchAttributesUpserted' => [ + 'sequence', + 'attributes', + 'attribute_types', + 'merged', + 'reused_from_run_id', + 'reused_recorded_at', + ], + 'MemoUpserted' => ['sequence', 'entries', 'merged', 'reused_from_run_id', 'reused_recorded_at'], 'RepairRequested' => [ 'workflow_command_id', 'workflow_instance_id', diff --git a/src/V2/Support/MemoUpsertService.php b/src/V2/Support/MemoUpsertService.php index dafce9ada..06c190cbe 100644 --- a/src/V2/Support/MemoUpsertService.php +++ b/src/V2/Support/MemoUpsertService.php @@ -102,6 +102,37 @@ public function inheritFromParent( }); } + /** + * Copy memos to a redrive successor as they are. The successor reuses the + * source's history, so each row keeps the step that wrote it and whether + * it was inherited. + */ + public function copyForRedrive(WorkflowRun $source, WorkflowRun $successor): void + { + $sourceMemos = WorkflowMemo::where('workflow_run_id', $source->id)->get(); + + ConfiguredV2Models::connection()->transaction(static function () use ($sourceMemos, $successor): void { + foreach ($sourceMemos as $sourceMemo) { + $memo = new WorkflowMemo([ + 'workflow_run_id' => $successor->id, + 'workflow_instance_id' => $successor->workflow_instance_id, + 'key' => $sourceMemo->key, + 'upserted_at_sequence' => $sourceMemo->upserted_at_sequence, + 'inherited_from_parent' => $sourceMemo->inherited_from_parent, + ]); + + $memo->setValue($sourceMemo->getValue()); + + $memo->save(); + } + + if ($sourceMemos->isNotEmpty()) { + WorkflowMemo::validateTotalSize($successor->id); + WorkflowMemo::validateCount($successor->id); + } + }); + } + /** * Get memos as a key-value array. * diff --git a/src/V2/Support/QueryStateReplayer.php b/src/V2/Support/QueryStateReplayer.php index 44e31852e..eca404504 100644 --- a/src/V2/Support/QueryStateReplayer.php +++ b/src/V2/Support/QueryStateReplayer.php @@ -401,7 +401,7 @@ public function replayState(WorkflowRun $run): ReplayState ); $this->syncWorkflowCursor($workflow, $sequence + 1); - $current = $workflowExecution->send(null, $upsertEvent->recorded_at); + $current = $workflowExecution->send(null, RedriveReplayClock::recordedAt($upsertEvent)); ++$sequence; @@ -428,7 +428,7 @@ public function replayState(WorkflowRun $run): ReplayState ); $this->syncWorkflowCursor($workflow, $sequence + 1); - $current = $workflowExecution->send(null, $upsertEvent->recorded_at); + $current = $workflowExecution->send(null, RedriveReplayClock::recordedAt($upsertEvent)); ++$sequence; diff --git a/src/V2/Support/SearchAttributeUpsertService.php b/src/V2/Support/SearchAttributeUpsertService.php index 2c620fcff..cba1b3c65 100644 --- a/src/V2/Support/SearchAttributeUpsertService.php +++ b/src/V2/Support/SearchAttributeUpsertService.php @@ -236,6 +236,40 @@ public function inheritFromParent( }); } + /** + * Copy search attributes to a redrive successor as they are. The successor + * reuses the source's history, so each row keeps the step that wrote it + * and whether it was inherited. + */ + public function copyForRedrive(WorkflowRun $source, WorkflowRun $successor): void + { + ConfiguredV2Models::connection()->transaction(static function () use ($source, $successor): void { + $sourceAttributes = WorkflowSearchAttribute::where('workflow_run_id', $source->id) + ->get(); + + foreach ($sourceAttributes as $sourceAttribute) { + (new WorkflowSearchAttribute([ + 'workflow_run_id' => $successor->id, + 'workflow_instance_id' => $successor->workflow_instance_id, + 'key' => $sourceAttribute->key, + 'type' => $sourceAttribute->type, + 'value_string' => $sourceAttribute->value_string, + 'value_keyword' => $sourceAttribute->value_keyword, + 'value_keyword_list' => $sourceAttribute->value_keyword_list, + 'value_int' => $sourceAttribute->value_int, + 'value_float' => $sourceAttribute->value_float, + 'value_bool' => $sourceAttribute->value_bool, + 'value_datetime' => $sourceAttribute->value_datetime, + 'upserted_at_sequence' => $sourceAttribute->upserted_at_sequence, + 'inherited_from_parent' => $sourceAttribute->inherited_from_parent, + ]))->save(); + } + + WorkflowSearchAttribute::validateCount($successor->id); + WorkflowSearchAttribute::validateTotalSize($successor->id); + }); + } + /** * Get all search attributes for a run as key-value array. * diff --git a/src/V2/Support/WorkflowExecutor.php b/src/V2/Support/WorkflowExecutor.php index b3be0d9e7..90312d2f8 100644 --- a/src/V2/Support/WorkflowExecutor.php +++ b/src/V2/Support/WorkflowExecutor.php @@ -1162,7 +1162,11 @@ private function executeSearchAttributes( SearchAttributeReplayIdentity::assertCompatible($sequence, $upsertEvent->payload, $current); } $this->syncWorkflowCursor($workflow, $sequence + 1); - $current = $workflowExecution->send(null, $upsertEvent?->recorded_at, advanceCancellationTime: false); + $current = $workflowExecution->send( + null, + RedriveReplayClock::recordedAt($upsertEvent), + advanceCancellationTime: false, + ); } catch (Throwable $throwable) { $this->failRun($run, $task, $throwable, 'workflow_run', $run->id); return null; @@ -1201,7 +1205,11 @@ private function executeMemo( ); } $this->syncWorkflowCursor($workflow, $sequence + 1); - $current = $workflowExecution->send(null, $memoEvent?->recorded_at, advanceCancellationTime: false); + $current = $workflowExecution->send( + null, + RedriveReplayClock::recordedAt($memoEvent), + advanceCancellationTime: false, + ); } 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 8a66814df..02d9bac16 100644 --- a/src/V2/Support/WorkflowFiberRunner.php +++ b/src/V2/Support/WorkflowFiberRunner.php @@ -2428,7 +2428,8 @@ private static function indexRecordedSearchAttributeUpserts(array $historyEvents $upserts[$sequence] = [ 'payload' => $payload, - 'recorded_at' => self::eventRecordedAt($event, $payload), + 'recorded_at' => self::eventTime($payload['reused_recorded_at'] ?? null) + ?? self::eventRecordedAt($event, $payload), ]; } @@ -2457,7 +2458,8 @@ private static function indexRecordedMemoUpserts(array $historyEvents): array $upserts[$sequence] = [ 'entries' => $payload['entries'] ?? null, - 'recorded_at' => self::eventRecordedAt($event, $payload), + 'recorded_at' => self::eventTime($payload['reused_recorded_at'] ?? null) + ?? self::eventRecordedAt($event, $payload), ]; } diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index aaed0f209..12c60aa97 100644 --- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php +++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php @@ -316,6 +316,68 @@ public function testRedriveReusedMarkersCorpusKeepsTheirStepsAndTime(): void ]); } + public function testRedriveReusedUpsertsCorpusKeepsTheirStepsRowsAndTime(): void + { + $path = self::FIXTURE_DIR . '/redrive-reused-upserts-clock.json'; + if (! is_file($path)) { + $this->markTestSkipped('The redrive reused-upsert evidence is absent.'); + } + $fixture = json_decode(file_get_contents($path), true, flags: JSON_THROW_ON_ERROR); + $scenario = $fixture['redrive_reused_upserts']; + 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-upserts'); + $stub->start(...$fixture['workflow']['arguments']); + $this->assertTrue($stub->refresh()->failed()); + $sourceRunId = $stub->runId(); + $sourceTimes = []; + foreach ($scenario['reused_steps'] as $step) { + $sourceTimes[$step['sequence']] = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $sourceRunId) + ->where('event_type', $step['event_type']) + ->firstOrFail() + ->recorded_at?->toIso8601String(); + } + + $this->travel($scenario['redrive_after_minutes']) + ->minutes(); + $result = app(WorkflowControlPlane::class)->redrive('regression-corpus-redrive-upserts', $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')); + foreach ($scenario['reused_steps'] as $step) { + $reused = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $successor->id) + ->where('event_type', $step['event_type']) + ->sole(); + $this->assertSame($step['sequence'], $reused->payload['sequence']); + $this->assertSame($sourceRunId, $reused->payload['reused_from_run_id'] ?? null); + $this->assertSame($sourceTimes[$step['sequence']], $reused->payload['reused_recorded_at'] ?? null); + } + $this->assertEquals($scenario['memo'], $successor->typedMemos()); + $this->assertEquals($scenario['search_attributes'], $successor->typedSearchAttributes()); + $output = $successor->workflowOutput(); + $this->assertSame([ + 'time_after_memo' => $sourceTimes[1], + 'time_after_search_attributes' => $sourceTimes[2], + ], [ + 'time_after_memo' => $output['time_after_memo'], + 'time_after_search_attributes' => $output['time_after_search_attributes'], + ]); + } + 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 6c934ee05..4279b171b 100644 --- a/tests/Feature/V2/V2FailedRunRedriveTest.php +++ b/tests/Feature/V2/V2FailedRunRedriveTest.php @@ -12,6 +12,7 @@ use Tests\Fixtures\V2\TestRedriveMarkersWorkflow; use Tests\Fixtures\V2\TestRedrivePatchedWorkflow; use Tests\Fixtures\V2\TestRedriveTimerWorkflow; +use Tests\Fixtures\V2\TestRedriveVisibilityWorkflow; use Tests\Fixtures\V2\TestRedriveWorkflow; use Tests\Fixtures\V2\TestThrowAfterGreetingWorkflow; use Tests\TestCase; @@ -482,6 +483,150 @@ public function testRedriveReusesSideEffectsAndVersionMarkersAtTheirRecordedTime $this->assertSame($output, $fiber->result); } + public function testRedriveReusesWorkflowCodeMemoAndSearchAttributeUpserts(): void + { + WorkflowStub::fake(); + + $workflow = WorkflowStub::make(TestRedriveVisibilityWorkflow::class, 'redrive-upserts-1'); + $workflow->start('Taylor'); + $this->assertTrue($workflow->refresh()->failed()); + $sourceRunId = $workflow->runId(); + $source = WorkflowRun::query()->findOrFail($sourceRunId); + $sourceEvent = static fn (HistoryEventType $type): WorkflowHistoryEvent => WorkflowHistoryEvent::query() + ->where('workflow_run_id', $sourceRunId) + ->where('event_type', $type) + ->firstOrFail(); + $sourceMemoEvent = $sourceEvent(HistoryEventType::MemoUpserted); + $sourceAttributesEvent = $sourceEvent(HistoryEventType::SearchAttributesUpserted); + $rows = static fn (WorkflowRun $run): array => [ + 'memos' => $run->memos() + ->orderBy('key') + ->get() + ->map(static fn ($memo): array => [ + $memo->key, + $memo->getValue(), + $memo->upserted_at_sequence, + (bool) $memo->inherited_from_parent, + ]) + ->all(), + 'search_attributes' => $run->searchAttributes() + ->orderBy('key') + ->get() + ->map(static fn ($attribute): array => [ + $attribute->key, + $attribute->type, + $attribute->upserted_at_sequence, + (bool) $attribute->inherited_from_parent, + ]) + ->all(), + ]; + $sourceRows = $rows($source); + $this->assertSame([ + ['customer', 'Taylor', 1, false], + ['stage', 'greeting', 1, false], + ], $sourceRows['memos']); + $this->assertSame([['Stage', 'keyword', 2, false]], $sourceRows['search_attributes']); + + $this->travel(5) + ->minutes(); + $result = app(WorkflowControlPlane::class)->redrive('redrive-upserts-1', $sourceRunId); + + $this->assertTrue($result['accepted'], (string) $result['reason']); + $this->assertSame(4, $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')); + $this->assertSame($sourceRows, $rows($successor->fresh())); + $this->assertSame($source->typedMemos(), $successor->typedMemos()); + $this->assertSame($source->typedSearchAttributes(), $successor->typedSearchAttributes()); + + foreach ([ + [$sourceMemoEvent, HistoryEventType::MemoUpserted], + [$sourceAttributesEvent, HistoryEventType::SearchAttributesUpserted], + ] as [$original, $type]) { + $reused = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $successor->id) + ->where('event_type', $type) + ->sole(); + $payload = $reused->payload; + $this->assertSame($sourceRunId, $payload['reused_from_run_id'] ?? null); + $this->assertSame($original->recorded_at?->toIso8601String(), $payload['reused_recorded_at'] ?? null); + unset($payload['reused_from_run_id'], $payload['reused_recorded_at']); + // The source's task and command are its recording context, not the step. + $expected = $original->payload; + unset($expected['task'], $expected['command']); + $this->assertSame($expected, $payload); + } + + $expectedTimes = [ + 'time_after_memo' => $sourceMemoEvent->recorded_at?->toIso8601String(), + 'time_after_search_attributes' => $sourceAttributesEvent->recorded_at?->toIso8601String(), + ]; + $output = $successor->workflowOutput(); + $this->assertSame('Hello, Taylor! Redrive completed.', $output['result']); + $this->assertSame($expectedTimes, [ + 'time_after_memo' => $output['time_after_memo'], + 'time_after_search_attributes' => $output['time_after_search_attributes'], + ]); + $export = json_decode( + json_encode(HistoryExport::forRun($successor->fresh()), JSON_THROW_ON_ERROR), + true, + 512, + JSON_THROW_ON_ERROR, + ); + $this->assertSame($expectedTimes, (new WorkflowReplayer())->replayExport($export)->workflow->replayTimes()); + $this->assertSame( + $expectedTimes, + (new WorkflowReplayer())->replay($successor->fresh()) + ->workflow->replayTimes(), + ); + } + + public function testARedriveSuccessorWithUpsertsCanBeRedrivenAgain(): void + { + WorkflowStub::fake(); + config([ + 'workflows.v2.testing.redrive_failures_before_success' => 2, + ]); + + $workflow = WorkflowStub::make(TestRedriveVisibilityWorkflow::class, 'redrive-upserts-chain'); + $workflow->start('Taylor'); + $this->assertTrue($workflow->refresh()->failed()); + $firstRunId = $workflow->runId(); + $memoRecordedAt = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $firstRunId) + ->where('event_type', HistoryEventType::MemoUpserted) + ->sole() + ->recorded_at?->toIso8601String(); + + $second = app(WorkflowControlPlane::class)->redrive('redrive-upserts-chain', $firstRunId); + $this->assertTrue($second['accepted'], (string) $second['reason']); + $secondRun = WorkflowRun::query()->findOrFail($second['workflow_run_id']); + $this->assertSame(RunStatus::Failed, $secondRun->status); + + $this->travel(5) + ->minutes(); + $third = app(WorkflowControlPlane::class)->redrive('redrive-upserts-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(1, (int) Cache::get('test:redrive:first-calls')); + $this->assertSame(WorkflowRun::query()->findOrFail($firstRunId)->typedMemos(), $thirdRun->typedMemos()); + $this->assertSame( + WorkflowRun::query()->findOrFail($firstRunId)->typedSearchAttributes(), + $thirdRun->typedSearchAttributes(), + ); + $reusedMemo = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $thirdRun->id) + ->where('event_type', HistoryEventType::MemoUpserted) + ->sole(); + // A step reused twice keeps the run it came from and its first time. + $this->assertSame($secondRun->id, $reusedMemo->payload['reused_from_run_id']); + $this->assertSame($memoRecordedAt, $reusedMemo->payload['reused_recorded_at']); + $this->assertSame($memoRecordedAt, $thirdRun->workflowOutput()['time_after_memo']); + } + public function testRedriveReusesOneMarkerForARepeatedPatchCall(): void { WorkflowStub::fake(); diff --git a/tests/Fixtures/V2/ReplayRegression/redrive-reused-upserts-clock.json b/tests/Fixtures/V2/ReplayRegression/redrive-reused-upserts-clock.json new file mode 100644 index 000000000..dfee3d9ad --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/redrive-reused-upserts-clock.json @@ -0,0 +1,95 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "redrive-reused-upserts-clock", + "protocol_version": "1.0", + "bindings": [ + "php" + ], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestRedriveVisibilityWorkflow", + "arguments": [ + "Taylor" + ], + "payload_codec": "avro" + }, + "redrive_reused_upserts": { + "redrive_after_minutes": 5, + "redrive": "accepted", + "resume_step_sequence": 4, + "successor_status": "completed", + "reused_steps": [ + { + "sequence": 1, + "event_type": "MemoUpserted" + }, + { + "sequence": 2, + "event_type": "SearchAttributesUpserted" + } + ], + "memo": { + "customer": "Taylor", + "stage": "greeting" + }, + "search_attributes": { + "Stage": "greeting" + } + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": {} + }, + { + "sequence": 2, + "event_type": "MemoUpserted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "sequence": 1, + "entries": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + }, + "merged": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + }, + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:00+00:00" + } + }, + { + "sequence": 3, + "event_type": "SearchAttributesUpserted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "sequence": 2, + "attributes": { + "Stage": "greeting" + }, + "attribute_types": { + "Stage": "keyword" + }, + "merged": { + "Stage": "greeting" + }, + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:00+00:00" + } + } + ], + "expected": { + "completed": false, + "result": null, + "commands": [ + { + "type": "schedule_activity", + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "payload_codec": "avro" + } + ] + } +} diff --git a/tests/Fixtures/V2/TestRedriveVisibilityWorkflow.php b/tests/Fixtures/V2/TestRedriveVisibilityWorkflow.php new file mode 100644 index 000000000..044676fc6 --- /dev/null +++ b/tests/Fixtures/V2/TestRedriveVisibilityWorkflow.php @@ -0,0 +1,58 @@ + 'greeting', + 'customer' => $name, + ]); + $this->timeAfterMemo = now() + ->toIso8601String(); + upsertSearchAttributes([ + 'Stage' => 'greeting', + ]); + $this->timeAfterSearchAttributes = now() + ->toIso8601String(); + $greeting = activity(TestRedriveFirstActivity::class, $name); + + return [ + 'result' => activity(TestRedriveSecondActivity::class, $greeting), + 'time_after_memo' => $this->timeAfterMemo, + 'time_after_search_attributes' => $this->timeAfterSearchAttributes, + ]; + } + + /** + * @return array{time_after_memo: string|null, time_after_search_attributes: string|null} + */ + #[QueryMethod] + public function replayTimes(): array + { + return [ + 'time_after_memo' => $this->timeAfterMemo, + 'time_after_search_attributes' => $this->timeAfterSearchAttributes, + ]; + } +}