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