From cebeab3b15505a314f8dc8d522e2f0f45509a85a Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 14:43:20 +0000 Subject: [PATCH 1/4] Allow redrive through validated activity retry histories --- CHANGELOG.md | 8 + regression-corpus-policy.json | 6 + src/V2/Support/ActivityRetryHistory.php | 85 ++++ src/V2/Support/FailedRunRedrivePlan.php | 42 +- tests/Feature/V2/V2FailedRunRedriveTest.php | 128 ++++- .../redrive-native-active-start-adoption.json | 229 +++++++++ ...etried-activity-after-completed-child.json | 441 ++++++++++++++++++ .../redrive-retried-completed-activity.json | 332 +++++++++++++ .../redrive-retried-failed-activity.json | 210 +++++++++ .../V2/TestRedriveRetryProbeActivity.php | 29 ++ .../V2/TestRedriveRetryProbeWorkflow.php | 29 ++ .../Fixtures/V2/TestRedriveRetryWorkflow.php | 20 + tests/Unit/V2/FailedRunRedrivePlanTest.php | 220 +++++++++ tests/Unit/V2/ReplayRegressionCorpusTest.php | 39 ++ 14 files changed, 1816 insertions(+), 2 deletions(-) create mode 100644 src/V2/Support/ActivityRetryHistory.php create mode 100644 tests/Fixtures/V2/ReplayRegression/redrive-native-active-start-adoption.json create mode 100644 tests/Fixtures/V2/ReplayRegression/redrive-retried-activity-after-completed-child.json create mode 100644 tests/Fixtures/V2/ReplayRegression/redrive-retried-completed-activity.json create mode 100644 tests/Fixtures/V2/ReplayRegression/redrive-retried-failed-activity.json create mode 100644 tests/Fixtures/V2/TestRedriveRetryProbeActivity.php create mode 100644 tests/Fixtures/V2/TestRedriveRetryProbeWorkflow.php create mode 100644 tests/Fixtures/V2/TestRedriveRetryWorkflow.php diff --git a/CHANGELOG.md b/CHANGELOG.md index ef222ece1..8706e54be 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,14 @@ ## Unreleased +- Failed-run redrive accepts a verified linear activity retry chain. A retried + completed prefix is reused and only the failed boundary step runs again. + Substituted attempt/task identities refuse with + `invalid_activity_retry_history`; existing service-worker definition checks + still apply. An exactly bound accepted start/adopt receipt is harmless, while + rejected or mismatched starts remain unsupported. This core correction is + separate from the experimental Saga compensation scope foundation (#1063). + - **Run migrations.** Opt-in deadline-triggered cleanup (#863, experimental, embedded). A start can reserve the last part of its hard execution or run timeout for cleanup with `StartOptions::withTimeoutCleanup($seconds)` (or diff --git a/regression-corpus-policy.json b/regression-corpus-policy.json index 114529c98..2df39af3d 100644 --- a/regression-corpus-policy.json +++ b/regression-corpus-policy.json @@ -74,6 +74,12 @@ "history_sequence", "history_payload" ] + }, + { + "glob": "src/V2/Support/FailedRunRedrivePlan.php" + }, + { + "glob": "src/V2/Support/ActivityRetryHistory.php" } ] } diff --git a/src/V2/Support/ActivityRetryHistory.php b/src/V2/Support/ActivityRetryHistory.php new file mode 100644 index 000000000..dd9b84516 --- /dev/null +++ b/src/V2/Support/ActivityRetryHistory.php @@ -0,0 +1,85 @@ + $events + */ + public static function valid(array $events): bool + { + $hasRetries = false; + foreach ($events as $event) { + $hasRetries = $hasRetries || $event->event_type === HistoryEventType::ActivityRetryScheduled; + } + if (! $hasRetries) { + return true; + } + $execution = null; + $attempt = null; + $task = null; + $number = 0; + $nextTask = null; + $terminal = false; + $seenAttempts = []; + $seenTasks = []; + foreach ($events as $event) { + $payload = $event->payload; + $id = $payload['activity_execution_id'] ?? null; + if (! is_string($id) || $id === '' || ($execution !== null && $id !== $execution) || $terminal) { + return false; + } + $execution = $id; + if ($event->event_type === HistoryEventType::ActivityStarted) { + $currentAttempt = $payload['activity_attempt_id'] ?? null; + $currentTask = $event->workflow_task_id ?? ($payload['workflow_task_id'] ?? null); + if (isset($event->workflow_task_id, $payload['workflow_task_id']) && $event->workflow_task_id !== $payload['workflow_task_id']) { + return false; + } + if (! is_string($currentAttempt) || $currentAttempt === '' || isset($seenAttempts[$currentAttempt]) + || ! is_string($currentTask) || $currentTask === '' || isset($seenTasks[$currentTask]) + || ($payload['attempt_number'] ?? null) !== $number + 1 + || ($number > 0 && $nextTask !== $currentTask)) { + return false; + } + $attempt = $currentAttempt; + $task = $currentTask; + $seenAttempts[$attempt] = true; + $seenTasks[$task] = true; + ++$number; + $nextTask = null; + } elseif ($event->event_type === HistoryEventType::ActivityRetryScheduled) { + $retry = $payload['retry_task_id'] ?? null; + $maximum = $payload['max_attempts'] ?? null; + if ($attempt === null || $nextTask !== null + || ($payload['activity_attempt_id'] ?? null) !== $attempt + || ($payload['retry_after_attempt_id'] ?? null) !== $attempt + || ($payload['retry_after_attempt'] ?? null) !== $number + || ($payload['retry_of_task_id'] ?? null) !== $task + || ! is_string($retry) || $retry === '' || isset($seenTasks[$retry]) + || ($maximum !== null && (! is_int($maximum) || $maximum <= $number))) { + return false; + } + $nextTask = $retry; + } elseif (in_array( + $event->event_type, + [HistoryEventType::ActivityCompleted, HistoryEventType::ActivityFailed], + true + )) { + if ($nextTask !== null || $attempt === null || ($payload['activity_attempt_id'] ?? null) !== $attempt + || ($payload['attempt_number'] ?? null) !== $number) { + return false; + } + $terminal = true; + } + } + return $terminal; + } +} diff --git a/src/V2/Support/FailedRunRedrivePlan.php b/src/V2/Support/FailedRunRedrivePlan.php index ea39c75c8..50ee2a3fc 100644 --- a/src/V2/Support/FailedRunRedrivePlan.php +++ b/src/V2/Support/FailedRunRedrivePlan.php @@ -7,6 +7,9 @@ use Illuminate\Support\Str; use Throwable; use Workflow\Serializers\Serializer; +use Workflow\V2\Enums\CommandOutcome; +use Workflow\V2\Enums\CommandStatus; +use Workflow\V2\Enums\CommandType; use Workflow\V2\Enums\HistoryEventType; use Workflow\V2\Enums\RunStatus; use Workflow\V2\Models\WorkflowHistoryEvent; @@ -86,7 +89,7 @@ public static function forRun(WorkflowRun $run): array } if ($event->event_type === HistoryEventType::StartAccepted) { - if ($started !== null) { + if ($started !== null && ! self::isActiveStartAdoption($run, $event)) { return self::reject('unexpected_start'); } @@ -199,6 +202,7 @@ public static function forRun(WorkflowRun $run): array if (! in_array($event->event_type, [ HistoryEventType::ActivityScheduled, HistoryEventType::ActivityStarted, + HistoryEventType::ActivityRetryScheduled, HistoryEventType::ActivityHeartbeatRecorded, HistoryEventType::ActivityCompleted, HistoryEventType::ActivityFailed, @@ -381,6 +385,10 @@ public static function forRun(WorkflowRun $run): array return self::reject('activity_sequence_gap'); } + if (! ActivityRetryHistory::valid($stepEvents)) { + return self::reject('invalid_activity_retry_history'); + } + $activityType = null; $completion = null; $failureCount = 0; @@ -612,6 +620,38 @@ public static function successorEvents(array $plan, string $sourceRunId): array return $events; } + /** + * Native start/adopt records a command receipt, not another workflow start or reusable step. + * Only accepted adoption proves the same resolved run and command identity. + * A rejected start is not adoption evidence and retains its existing refusal. + */ + private static function isActiveStartAdoption(WorkflowRun $run, WorkflowHistoryEvent $event): bool + { + $payload = $event->payload; + $command = $payload['command'] ?? null; + if (! is_array($command) || ($payload['outcome'] ?? null) !== CommandOutcome::ReturnedExistingActive->value + || ($payload['rejection_reason'] ?? null) !== null + || $event->workflow_run_id !== $run->id) { + return false; + } + foreach (['workflow_instance_id', 'workflow_run_id', 'workflow_class', 'workflow_type'] as $key) { + $expected = $key === 'workflow_run_id' ? $run->id : $run->getAttribute($key); + if (! is_string($expected) || $expected === '' || ($payload[$key] ?? null) !== $expected) { + return false; + } + } + $commandId = $payload['workflow_command_id'] ?? null; + return is_string($commandId) && $commandId !== '' + && $event->workflow_command_id === $commandId + && ($command['id'] ?? null) === $commandId + && ($command['type'] ?? null) === CommandType::Start->value + && ($command['target_scope'] ?? null) === 'instance' + && ($command['resolved_run_id'] ?? null) === $run->id + && ($command['status'] ?? null) === CommandStatus::Accepted->value + && ($command['outcome'] ?? null) === CommandOutcome::ReturnedExistingActive->value + && ($command['rejection_reason'] ?? null) === null; + } + /** * 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 diff --git a/tests/Feature/V2/V2FailedRunRedriveTest.php b/tests/Feature/V2/V2FailedRunRedriveTest.php index 6edb98ef4..f1bed9890 100644 --- a/tests/Feature/V2/V2FailedRunRedriveTest.php +++ b/tests/Feature/V2/V2FailedRunRedriveTest.php @@ -7,6 +7,7 @@ use Illuminate\Support\Facades\Cache; use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Queue; +use PHPUnit\Framework\Attributes\DataProvider; use Symfony\Component\Process\Process; use Tests\Fixtures\V2\TestGreetingWorkflow; use Tests\Fixtures\V2\TestParentFailingChildWorkflow; @@ -17,6 +18,7 @@ use Tests\Fixtures\V2\TestRedriveIncompatibleWorkflow; use Tests\Fixtures\V2\TestRedriveMarkersWorkflow; use Tests\Fixtures\V2\TestRedrivePatchedWorkflow; +use Tests\Fixtures\V2\TestRedriveRetryProbeWorkflow; use Tests\Fixtures\V2\TestRedriveSignalWorkflow; use Tests\Fixtures\V2\TestRedriveTimerWorkflow; use Tests\Fixtures\V2\TestRedriveTwoSignalsWorkflow; @@ -33,7 +35,9 @@ use Workflow\V2\Enums\RunStatus; use Workflow\V2\Enums\TaskStatus; use Workflow\V2\Enums\TaskType; +use Workflow\V2\Jobs\RunActivityTask; use Workflow\V2\Jobs\RunTimerTask; +use Workflow\V2\Jobs\RunWorkflowTask; use Workflow\V2\Models\WorkflowCommand; use Workflow\V2\Models\WorkflowHistoryEvent; use Workflow\V2\Models\WorkflowInstance; @@ -185,6 +189,64 @@ public function testFailedRunReusesCompletedActivityAndRetriesOnlyFailedActivity ); } + /** + * @return array + */ + public static function retriedActivityCases(): array + { + return [ + 'retried failed boundary' => [0], + 'retried completed prefix and failed boundary' => [1], + ]; + } + + #[DataProvider('retriedActivityCases')] + public function testOrdinaryRetriedActivitiesRedriveWithoutRepeatingTheCompletedPrefix(int $firstFailures): void + { + Queue::fake(); + config([ + 'workflows.v2.testing.redrive_retry_failures' => [ + 'first' => $firstFailures, + 'second' => 2, + ], + ]); + $counter = 'test:redrive:retry-probe:' . $firstFailures; + Cache::forget($counter . ':first'); + Cache::forget($counter . ':second'); + $workflow = WorkflowStub::make(TestRedriveRetryProbeWorkflow::class, 'redrive-retry-probe-' . $firstFailures); + $workflow->start($counter, new StartOptions(executionTimeoutSeconds: 600)); + $source = $this->drainRetryProbe($workflow->runId()); + $this->assertSame(RunStatus::Failed, $source->status); + $this->assertSame(1 + $firstFailures, (int) Cache::get($counter . ':first')); + $this->assertSame(2, (int) Cache::get($counter . ':second')); + $history = $source->historyEvents() + ->orderBy('sequence') + ->get() + ->toArray(); + + $control = app(WorkflowControlPlane::class); + $redrive = $control->redrive($source->workflow_instance_id, $source->id, [ + 'request_id' => 'retry-probe', + ]); + $this->assertTrue($redrive['accepted'], (string) $redrive['reason']); + $this->assertSame(2, $redrive['resume_step_sequence']); + $successor = $this->drainRetryProbe($redrive['workflow_run_id']); + $this->assertSame(RunStatus::Completed, $successor->status); + $this->assertSame('value:first:second', $successor->workflowOutput()); + $this->assertSame(1 + $firstFailures, (int) Cache::get($counter . ':first')); + $this->assertSame(3, (int) Cache::get($counter . ':second')); + $this->assertSame( + $source->execution_deadline_at->toISOString(), + $successor->execution_deadline_at->toISOString() + ); + $this->assertSame($history, $source->historyEvents()->orderBy('sequence')->get()->toArray()); + $repeated = $control->redrive($source->workflow_instance_id, $source->id, [ + 'request_id' => 'retry-probe', + ]); + $this->assertSame($successor->id, $repeated['workflow_run_id']); + $this->assertSame(3, (int) Cache::get($counter . ':second')); + } + public function testRedriveCarriesStartMemoAndSearchAttributesToTheSuccessor(): void { WorkflowStub::fake(); @@ -263,7 +325,19 @@ public function testCompletedRunIsNotEligibleForRedrive(): void $this->assertSame(1, WorkflowRun::query()->where('workflow_instance_id', 'redrive-completed-1')->count()); } - public function testServiceRunRequiresMatchingRecordedWorkerDefinitionFingerprint(): void + /** + * @return array + */ + public static function serviceRetryHistories(): array + { + return [ + 'ordinary prefix' => [false], + 'retried completed prefix' => [true], + ]; + } + + #[DataProvider('serviceRetryHistories')] + public function testServiceRunRequiresMatchingRecordedWorkerDefinitionFingerprint(bool $retriedPrefix): void { Queue::fake(); @@ -292,12 +366,45 @@ public function testServiceRunRequiresMatchingRecordedWorkerDefinitionFingerprin 'activity_type' => 'remote.first', 'sequence' => 1, ]); + if ($retriedPrefix) { + WorkflowHistoryEvent::record($source, HistoryEventType::ActivityStarted, [ + 'activity_execution_id' => 'service-activity-1', + 'activity_type' => 'remote.first', + 'activity_attempt_id' => 'service-attempt-1', + 'workflow_task_id' => 'service-activity-task-1', + 'attempt_number' => 1, + 'sequence' => 1, + ]); + WorkflowHistoryEvent::record($source, HistoryEventType::ActivityRetryScheduled, [ + 'activity_execution_id' => 'service-activity-1', + 'activity_type' => 'remote.first', + 'activity_attempt_id' => 'service-attempt-1', + 'retry_after_attempt_id' => 'service-attempt-1', + 'retry_after_attempt' => 1, + 'retry_of_task_id' => 'service-activity-task-1', + 'retry_task_id' => 'service-activity-task-2', + 'max_attempts' => 2, + 'sequence' => 1, + ]); + WorkflowHistoryEvent::record($source, HistoryEventType::ActivityStarted, [ + 'activity_execution_id' => 'service-activity-1', + 'activity_type' => 'remote.first', + 'activity_attempt_id' => 'service-attempt-2', + 'workflow_task_id' => 'service-activity-task-2', + 'attempt_number' => 2, + 'sequence' => 1, + ]); + } WorkflowHistoryEvent::record($source, HistoryEventType::ActivityCompleted, [ 'activity_execution_id' => 'service-activity-1', 'activity_type' => 'remote.first', 'sequence' => 1, 'result' => Serializer::serializeWithCodec('avro', 'first-result'), 'payload_codec' => 'avro', + ...($retriedPrefix ? [ + 'activity_attempt_id' => 'service-attempt-2', + 'attempt_number' => 2, + ] : []), ]); WorkflowHistoryEvent::record($source, HistoryEventType::ActivityScheduled, [ 'activity_execution_id' => 'service-activity-2', @@ -2445,6 +2552,25 @@ private function failedSignalSource(string $instanceId): WorkflowRun return WorkflowRun::query()->findOrFail($workflow->runId()); } + private function drainRetryProbe(string $runId): WorkflowRun + { + for ($turn = 0; $turn < 20; ++$turn) { + $run = WorkflowRun::query()->findOrFail($runId); + if ($run->status->isTerminal()) { + return $run; + } + $task = WorkflowTask::query()->where('workflow_run_id', $runId) + ->where('status', TaskStatus::Ready)->orderBy('available_at')->firstOrFail(); + if ($task->available_at->isFuture()) { + $this->travelTo($task->available_at); + } + $job = $task->task_type === TaskType::Activity + ? new RunActivityTask($task->id) : new RunWorkflowTask($task->id); + $this->app->call([$job, 'handle']); + } + $this->fail('The ordinary retry fixture did not converge within twenty task deliveries.'); + } + /** * Every engine table and read projection a refused redrive must leave * untouched, including the run's message cursor. diff --git a/tests/Fixtures/V2/ReplayRegression/redrive-native-active-start-adoption.json b/tests/Fixtures/V2/ReplayRegression/redrive-native-active-start-adoption.json new file mode 100644 index 000000000..c59213735 --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/redrive-native-active-start-adoption.json @@ -0,0 +1,229 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "redrive-native-active-start-adoption", + "protocol_version": "1.0", + "bindings": [ + "php" + ], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestRedriveRetryWorkflow", + "arguments": [ + "Taylor", + false + ], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": {}, + "recorded_at": "2026-11-12T00:00:00+00:00" + }, + { + "sequence": 2, + "event_type": "StartAccepted", + "recorded_at": "2026-11-12T00:00:01+00:00", + "workflow_run_id": "source-run", + "workflow_command_id": "adoption-command", + "payload": { + "workflow_instance_id": "source-instance", + "workflow_run_id": "source-run", + "workflow_class": "Tests\\Fixtures\\V2\\TestRedriveRetryWorkflow", + "workflow_type": "Tests\\Fixtures\\V2\\TestRedriveRetryWorkflow", + "workflow_command_id": "adoption-command", + "outcome": "returned_existing_active" + } + }, + { + "sequence": 3, + "event_type": "ActivityScheduled", + "payload": { + "sequence": 1, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-1" + }, + "recorded_at": "2026-11-12T00:00:02+00:00" + }, + { + "sequence": 4, + "event_type": "ActivityCompleted", + "payload": { + "sequence": 1, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-1", + "result": "wwHioz3/VYAiNwocSGVsbG8sIFRheWxvciE=", + "payload_codec": "avro" + }, + "recorded_at": "2026-11-12T00:00:03+00:00" + } + ], + "redrive_plan": { + "source_history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": {}, + "recorded_at": "2026-11-12T00:00:00+00:00" + }, + { + "sequence": 2, + "event_type": "StartAccepted", + "recorded_at": "2026-11-12T00:00:01+00:00", + "workflow_run_id": "source-run", + "workflow_command_id": "adoption-command", + "payload": { + "workflow_instance_id": "source-instance", + "workflow_run_id": "source-run", + "workflow_class": "Tests\\Fixtures\\V2\\TestRedriveRetryWorkflow", + "workflow_type": "Tests\\Fixtures\\V2\\TestRedriveRetryWorkflow", + "workflow_command_id": "adoption-command", + "outcome": "returned_existing_active", + "command": { + "id": "adoption-command", + "type": "start", + "target_scope": "instance", + "resolved_run_id": "source-run", + "status": "accepted", + "outcome": "returned_existing_active" + } + } + }, + { + "sequence": 3, + "event_type": "ActivityScheduled", + "payload": { + "sequence": 1, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-1" + }, + "recorded_at": "2026-11-12T00:00:02+00:00" + }, + { + "sequence": 4, + "event_type": "ActivityCompleted", + "payload": { + "sequence": 1, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-1", + "result": "wwHioz3/VYAiNwocSGVsbG8sIFRheWxvciE=", + "payload_codec": "avro" + }, + "recorded_at": "2026-11-12T00:00:03+00:00" + }, + { + "sequence": 5, + "event_type": "ActivityScheduled", + "payload": { + "sequence": 2, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_execution_id": "execution-2" + }, + "recorded_at": "2026-11-12T00:00:04+00:00" + }, + { + "sequence": 6, + "event_type": "ActivityFailed", + "payload": { + "sequence": 2, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_execution_id": "execution-2", + "message": "provider unavailable", + "exception_class": "RuntimeException" + }, + "recorded_at": "2026-11-12T00:00:05+00:00" + }, + { + "sequence": 7, + "event_type": "WorkflowFailed", + "payload": { + "failed_step_kind": "activity", + "failed_step_sequence": 2 + }, + "recorded_at": "2026-11-12T00:00:06+00:00" + } + ], + "expected": { + "eligible": true, + "resume_step_sequence": 2, + "completed_count": 1, + "reused_event_types": [ + "ActivityCompleted" + ] + }, + "refusals": [ + { + "event_index": 1, + "field": "outcome", + "value": "started_new", + "reason": "unexpected_start" + }, + { + "event_index": 1, + "field": "workflow_instance_id", + "value": "other-instance", + "reason": "unexpected_start" + }, + { + "event_index": 1, + "field": "workflow_run_id", + "value": "other-run", + "reason": "unexpected_start" + }, + { + "event_index": 1, + "field": "workflow_class", + "value": "OtherWorkflow", + "reason": "unexpected_start" + }, + { + "event_index": 1, + "field": "workflow_type", + "value": "other-type", + "reason": "unexpected_start" + }, + { + "event_index": 1, + "field": "workflow_command_id", + "value": "other-command", + "reason": "unexpected_start" + }, + { + "event_index": 1, + "field": "rejection_reason", + "value": "refused", + "reason": "unexpected_start" + }, + { + "event_index": 1, + "field": "command", + "value": null, + "reason": "unexpected_start" + } + ], + "source_run": { + "id": "source-run", + "workflow_instance_id": "source-instance", + "workflow_class": "Tests\\Fixtures\\V2\\TestRedriveRetryWorkflow", + "workflow_type": "Tests\\Fixtures\\V2\\TestRedriveRetryWorkflow" + } + }, + "expected": { + "completed": false, + "result": null, + "commands": [ + { + "type": "schedule_activity", + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "payload_codec": "avro" + } + ] + } +} diff --git a/tests/Fixtures/V2/ReplayRegression/redrive-retried-activity-after-completed-child.json b/tests/Fixtures/V2/ReplayRegression/redrive-retried-activity-after-completed-child.json new file mode 100644 index 000000000..7c0d4baae --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/redrive-retried-activity-after-completed-child.json @@ -0,0 +1,441 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "redrive-retried-activity-after-completed-child", + "protocol_version": "1.0", + "bindings": [ + "php" + ], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestRedriveRetryWorkflow", + "arguments": [ + "Taylor", + true + ], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": {}, + "recorded_at": "2026-11-12T00:00:00+00:00" + }, + { + "sequence": 2, + "event_type": "ChildWorkflowScheduled", + "payload": { + "sequence": 1, + "workflow_link_id": "01JCREDRIVECHILDLINK000000", + "child_call_id": "01JCREDRIVECHILDLINK000000", + "child_workflow_instance_id": "01jcredrivechildinstance00", + "child_workflow_run_id": "01jcredrivechildrun0000000", + "child_workflow_class": "Tests\\Fixtures\\V2\\TestRedriveGreetingChildWorkflow", + "child_workflow_type": "test-redrive-greeting-child-workflow", + "parent_close_policy": "abandon", + "cancellation_policy": "abandon" + }, + "recorded_at": "2026-11-12T00:00:01+00:00" + }, + { + "sequence": 3, + "event_type": "ChildRunStarted", + "payload": { + "sequence": 1, + "workflow_link_id": "01JCREDRIVECHILDLINK000000", + "child_call_id": "01JCREDRIVECHILDLINK000000", + "child_workflow_instance_id": "01jcredrivechildinstance00", + "child_workflow_run_id": "01jcredrivechildrun0000000", + "child_workflow_class": "Tests\\Fixtures\\V2\\TestRedriveGreetingChildWorkflow", + "child_workflow_type": "test-redrive-greeting-child-workflow", + "child_run_number": 1, + "parent_close_policy": "abandon", + "cancellation_policy": "abandon" + }, + "recorded_at": "2026-11-12T00:00:02+00:00" + }, + { + "sequence": 4, + "event_type": "ChildRunCompleted", + "payload": { + "sequence": 1, + "workflow_link_id": "01JCREDRIVECHILDLINK000000", + "child_call_id": "01JCREDRIVECHILDLINK000000", + "child_workflow_instance_id": "01jcredrivechildinstance00", + "child_workflow_run_id": "01jcredrivechildrun0000000", + "child_workflow_class": "Tests\\Fixtures\\V2\\TestRedriveGreetingChildWorkflow", + "child_workflow_type": "test-redrive-greeting-child-workflow", + "child_run_number": 1, + "child_status": "completed", + "closed_reason": "completed", + "closed_at": "2026-11-12T00:00:01.000000Z", + "output": "wwHioz3/VYAiNwocSGVsbG8sIFRheWxvciE=", + "result": "wwHioz3/VYAiNwocSGVsbG8sIFRheWxvciE=", + "payload_codec": "avro" + }, + "recorded_at": "2026-11-12T00:00:03+00:00" + }, + { + "sequence": 5, + "event_type": "ActivityScheduled", + "payload": { + "sequence": 2, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-2" + }, + "recorded_at": "2026-11-12T00:00:04+00:00" + }, + { + "sequence": 6, + "event_type": "ActivityStarted", + "payload": { + "sequence": 2, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-2", + "activity_attempt_id": "attempt-2-1", + "attempt_number": 1, + "workflow_task_id": "task-2-1" + }, + "recorded_at": "2026-11-12T00:00:05+00:00" + }, + { + "sequence": 7, + "event_type": "ActivityRetryScheduled", + "payload": { + "sequence": 2, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-2", + "activity_attempt_id": "attempt-2-1", + "retry_after_attempt_id": "attempt-2-1", + "retry_after_attempt": 1, + "retry_of_task_id": "task-2-1", + "retry_task_id": "task-2-2", + "max_attempts": 2, + "retry_reason": "failure", + "retry_backoff_seconds": 0 + }, + "recorded_at": "2026-11-12T00:00:06+00:00" + }, + { + "sequence": 8, + "event_type": "ActivityStarted", + "payload": { + "sequence": 2, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-2", + "activity_attempt_id": "attempt-2-2", + "attempt_number": 2, + "workflow_task_id": "task-2-2" + }, + "recorded_at": "2026-11-12T00:00:07+00:00" + }, + { + "sequence": 9, + "event_type": "ActivityCompleted", + "payload": { + "sequence": 2, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-2", + "activity_attempt_id": "attempt-2-2", + "attempt_number": 2, + "result": "wwHioz3/VYAiNwocSGVsbG8sIFRheWxvciE=", + "payload_codec": "avro" + }, + "recorded_at": "2026-11-12T00:00:08+00:00" + } + ], + "redrive_plan": { + "source_history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": {}, + "recorded_at": "2026-11-12T00:00:00+00:00" + }, + { + "sequence": 2, + "event_type": "ChildWorkflowScheduled", + "payload": { + "sequence": 1, + "workflow_link_id": "01JCREDRIVECHILDLINK000000", + "child_call_id": "01JCREDRIVECHILDLINK000000", + "child_workflow_instance_id": "01jcredrivechildinstance00", + "child_workflow_run_id": "01jcredrivechildrun0000000", + "child_workflow_class": "Tests\\Fixtures\\V2\\TestRedriveGreetingChildWorkflow", + "child_workflow_type": "test-redrive-greeting-child-workflow", + "parent_close_policy": "abandon", + "cancellation_policy": "abandon" + }, + "recorded_at": "2026-11-12T00:00:01+00:00" + }, + { + "sequence": 3, + "event_type": "ChildRunStarted", + "payload": { + "sequence": 1, + "workflow_link_id": "01JCREDRIVECHILDLINK000000", + "child_call_id": "01JCREDRIVECHILDLINK000000", + "child_workflow_instance_id": "01jcredrivechildinstance00", + "child_workflow_run_id": "01jcredrivechildrun0000000", + "child_workflow_class": "Tests\\Fixtures\\V2\\TestRedriveGreetingChildWorkflow", + "child_workflow_type": "test-redrive-greeting-child-workflow", + "child_run_number": 1, + "parent_close_policy": "abandon", + "cancellation_policy": "abandon" + }, + "recorded_at": "2026-11-12T00:00:02+00:00" + }, + { + "sequence": 4, + "event_type": "ChildRunCompleted", + "payload": { + "sequence": 1, + "workflow_link_id": "01JCREDRIVECHILDLINK000000", + "child_call_id": "01JCREDRIVECHILDLINK000000", + "child_workflow_instance_id": "01jcredrivechildinstance00", + "child_workflow_run_id": "01jcredrivechildrun0000000", + "child_workflow_class": "Tests\\Fixtures\\V2\\TestRedriveGreetingChildWorkflow", + "child_workflow_type": "test-redrive-greeting-child-workflow", + "child_run_number": 1, + "child_status": "completed", + "closed_reason": "completed", + "closed_at": "2026-11-12T00:00:01.000000Z", + "output": "wwHioz3/VYAiNwocSGVsbG8sIFRheWxvciE=", + "result": "wwHioz3/VYAiNwocSGVsbG8sIFRheWxvciE=", + "payload_codec": "avro" + }, + "recorded_at": "2026-11-12T00:00:03+00:00" + }, + { + "sequence": 5, + "event_type": "ActivityScheduled", + "payload": { + "sequence": 2, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-2" + }, + "recorded_at": "2026-11-12T00:00:04+00:00" + }, + { + "sequence": 6, + "event_type": "ActivityStarted", + "payload": { + "sequence": 2, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-2", + "activity_attempt_id": "attempt-2-1", + "attempt_number": 1, + "workflow_task_id": "task-2-1" + }, + "recorded_at": "2026-11-12T00:00:05+00:00" + }, + { + "sequence": 7, + "event_type": "ActivityRetryScheduled", + "payload": { + "sequence": 2, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-2", + "activity_attempt_id": "attempt-2-1", + "retry_after_attempt_id": "attempt-2-1", + "retry_after_attempt": 1, + "retry_of_task_id": "task-2-1", + "retry_task_id": "task-2-2", + "max_attempts": 2, + "retry_reason": "failure", + "retry_backoff_seconds": 0 + }, + "recorded_at": "2026-11-12T00:00:06+00:00" + }, + { + "sequence": 8, + "event_type": "ActivityStarted", + "payload": { + "sequence": 2, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-2", + "activity_attempt_id": "attempt-2-2", + "attempt_number": 2, + "workflow_task_id": "task-2-2" + }, + "recorded_at": "2026-11-12T00:00:07+00:00" + }, + { + "sequence": 9, + "event_type": "ActivityCompleted", + "payload": { + "sequence": 2, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-2", + "activity_attempt_id": "attempt-2-2", + "attempt_number": 2, + "result": "wwHioz3/VYAiNwocSGVsbG8sIFRheWxvciE=", + "payload_codec": "avro" + }, + "recorded_at": "2026-11-12T00:00:08+00:00" + }, + { + "sequence": 10, + "event_type": "ActivityScheduled", + "payload": { + "sequence": 3, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_execution_id": "execution-3" + }, + "recorded_at": "2026-11-12T00:00:09+00:00" + }, + { + "sequence": 11, + "event_type": "ActivityStarted", + "payload": { + "sequence": 3, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_execution_id": "execution-3", + "activity_attempt_id": "attempt-3-1", + "attempt_number": 1, + "workflow_task_id": "task-3-1" + }, + "recorded_at": "2026-11-12T00:00:10+00:00" + }, + { + "sequence": 12, + "event_type": "ActivityRetryScheduled", + "payload": { + "sequence": 3, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_execution_id": "execution-3", + "activity_attempt_id": "attempt-3-1", + "retry_after_attempt_id": "attempt-3-1", + "retry_after_attempt": 1, + "retry_of_task_id": "task-3-1", + "retry_task_id": "task-3-2", + "max_attempts": 2, + "retry_reason": "failure", + "retry_backoff_seconds": 0 + }, + "recorded_at": "2026-11-12T00:00:11+00:00" + }, + { + "sequence": 13, + "event_type": "ActivityStarted", + "payload": { + "sequence": 3, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_execution_id": "execution-3", + "activity_attempt_id": "attempt-3-2", + "attempt_number": 2, + "workflow_task_id": "task-3-2" + }, + "recorded_at": "2026-11-12T00:00:12+00:00" + }, + { + "sequence": 14, + "event_type": "ActivityFailed", + "payload": { + "sequence": 3, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_execution_id": "execution-3", + "activity_attempt_id": "attempt-3-2", + "attempt_number": 2, + "message": "provider unavailable", + "exception_class": "RuntimeException" + }, + "recorded_at": "2026-11-12T00:00:13+00:00" + }, + { + "sequence": 15, + "event_type": "WorkflowFailed", + "payload": { + "failed_step_kind": "activity", + "failed_step_sequence": 3 + }, + "recorded_at": "2026-11-12T00:00:14+00:00" + } + ], + "expected": { + "eligible": true, + "resume_step_sequence": 3, + "completed_count": 1, + "reused_event_types": [ + "ChildRunCompleted", + "ActivityCompleted" + ] + }, + "refusals": [ + { + "event_index": 6, + "field": "activity_execution_id", + "value": "substituted-execution", + "reason": "invalid_activity_retry_history" + }, + { + "event_index": 6, + "field": "activity_attempt_id", + "value": "substituted-attempt", + "reason": "invalid_activity_retry_history" + }, + { + "event_index": 6, + "field": "retry_task_id", + "value": "substituted-task", + "reason": "invalid_activity_retry_history" + }, + { + "event_index": 6, + "field": "retry_after_attempt", + "value": 9, + "reason": "invalid_activity_retry_history" + }, + { + "event_index": 11, + "field": "activity_execution_id", + "value": "substituted-execution", + "reason": "invalid_activity_retry_history" + }, + { + "event_index": 11, + "field": "activity_attempt_id", + "value": "substituted-attempt", + "reason": "invalid_activity_retry_history" + }, + { + "event_index": 11, + "field": "retry_task_id", + "value": "substituted-task", + "reason": "invalid_activity_retry_history" + }, + { + "event_index": 11, + "field": "retry_after_attempt", + "value": 9, + "reason": "invalid_activity_retry_history" + } + ] + }, + "expected": { + "completed": false, + "result": null, + "commands": [ + { + "type": "schedule_activity", + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "payload_codec": "avro" + } + ] + } +} diff --git a/tests/Fixtures/V2/ReplayRegression/redrive-retried-completed-activity.json b/tests/Fixtures/V2/ReplayRegression/redrive-retried-completed-activity.json new file mode 100644 index 000000000..a63c15a0a --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/redrive-retried-completed-activity.json @@ -0,0 +1,332 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "redrive-retried-completed-activity", + "protocol_version": "1.0", + "bindings": [ + "php" + ], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestRedriveRetryWorkflow", + "arguments": [ + "Taylor", + false + ], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": {}, + "recorded_at": "2026-11-12T00:00:00+00:00" + }, + { + "sequence": 2, + "event_type": "ActivityScheduled", + "payload": { + "sequence": 1, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-1" + }, + "recorded_at": "2026-11-12T00:00:01+00:00" + }, + { + "sequence": 3, + "event_type": "ActivityStarted", + "payload": { + "sequence": 1, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-1", + "activity_attempt_id": "attempt-1-1", + "attempt_number": 1, + "workflow_task_id": "task-1-1" + }, + "recorded_at": "2026-11-12T00:00:02+00:00" + }, + { + "sequence": 4, + "event_type": "ActivityRetryScheduled", + "payload": { + "sequence": 1, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-1", + "activity_attempt_id": "attempt-1-1", + "retry_after_attempt_id": "attempt-1-1", + "retry_after_attempt": 1, + "retry_of_task_id": "task-1-1", + "retry_task_id": "task-1-2", + "max_attempts": 2, + "retry_reason": "failure", + "retry_backoff_seconds": 0 + }, + "recorded_at": "2026-11-12T00:00:03+00:00" + }, + { + "sequence": 5, + "event_type": "ActivityStarted", + "payload": { + "sequence": 1, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-1", + "activity_attempt_id": "attempt-1-2", + "attempt_number": 2, + "workflow_task_id": "task-1-2" + }, + "recorded_at": "2026-11-12T00:00:04+00:00" + }, + { + "sequence": 6, + "event_type": "ActivityCompleted", + "payload": { + "sequence": 1, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-1", + "activity_attempt_id": "attempt-1-2", + "attempt_number": 2, + "result": "wwHioz3/VYAiNwocSGVsbG8sIFRheWxvciE=", + "payload_codec": "avro" + }, + "recorded_at": "2026-11-12T00:00:05+00:00" + } + ], + "redrive_plan": { + "source_history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": {}, + "recorded_at": "2026-11-12T00:00:00+00:00" + }, + { + "sequence": 2, + "event_type": "ActivityScheduled", + "payload": { + "sequence": 1, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-1" + }, + "recorded_at": "2026-11-12T00:00:01+00:00" + }, + { + "sequence": 3, + "event_type": "ActivityStarted", + "payload": { + "sequence": 1, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-1", + "activity_attempt_id": "attempt-1-1", + "attempt_number": 1, + "workflow_task_id": "task-1-1" + }, + "recorded_at": "2026-11-12T00:00:02+00:00" + }, + { + "sequence": 4, + "event_type": "ActivityRetryScheduled", + "payload": { + "sequence": 1, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-1", + "activity_attempt_id": "attempt-1-1", + "retry_after_attempt_id": "attempt-1-1", + "retry_after_attempt": 1, + "retry_of_task_id": "task-1-1", + "retry_task_id": "task-1-2", + "max_attempts": 2, + "retry_reason": "failure", + "retry_backoff_seconds": 0 + }, + "recorded_at": "2026-11-12T00:00:03+00:00" + }, + { + "sequence": 5, + "event_type": "ActivityStarted", + "payload": { + "sequence": 1, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-1", + "activity_attempt_id": "attempt-1-2", + "attempt_number": 2, + "workflow_task_id": "task-1-2" + }, + "recorded_at": "2026-11-12T00:00:04+00:00" + }, + { + "sequence": 6, + "event_type": "ActivityCompleted", + "payload": { + "sequence": 1, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-1", + "activity_attempt_id": "attempt-1-2", + "attempt_number": 2, + "result": "wwHioz3/VYAiNwocSGVsbG8sIFRheWxvciE=", + "payload_codec": "avro" + }, + "recorded_at": "2026-11-12T00:00:05+00:00" + }, + { + "sequence": 7, + "event_type": "ActivityScheduled", + "payload": { + "sequence": 2, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_execution_id": "execution-2" + }, + "recorded_at": "2026-11-12T00:00:06+00:00" + }, + { + "sequence": 8, + "event_type": "ActivityStarted", + "payload": { + "sequence": 2, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_execution_id": "execution-2", + "activity_attempt_id": "attempt-2-1", + "attempt_number": 1, + "workflow_task_id": "task-2-1" + }, + "recorded_at": "2026-11-12T00:00:07+00:00" + }, + { + "sequence": 9, + "event_type": "ActivityRetryScheduled", + "payload": { + "sequence": 2, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_execution_id": "execution-2", + "activity_attempt_id": "attempt-2-1", + "retry_after_attempt_id": "attempt-2-1", + "retry_after_attempt": 1, + "retry_of_task_id": "task-2-1", + "retry_task_id": "task-2-2", + "max_attempts": 2, + "retry_reason": "failure", + "retry_backoff_seconds": 0 + }, + "recorded_at": "2026-11-12T00:00:08+00:00" + }, + { + "sequence": 10, + "event_type": "ActivityStarted", + "payload": { + "sequence": 2, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_execution_id": "execution-2", + "activity_attempt_id": "attempt-2-2", + "attempt_number": 2, + "workflow_task_id": "task-2-2" + }, + "recorded_at": "2026-11-12T00:00:09+00:00" + }, + { + "sequence": 11, + "event_type": "ActivityFailed", + "payload": { + "sequence": 2, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_execution_id": "execution-2", + "activity_attempt_id": "attempt-2-2", + "attempt_number": 2, + "message": "provider unavailable", + "exception_class": "RuntimeException" + }, + "recorded_at": "2026-11-12T00:00:10+00:00" + }, + { + "sequence": 12, + "event_type": "WorkflowFailed", + "payload": { + "failed_step_kind": "activity", + "failed_step_sequence": 2 + }, + "recorded_at": "2026-11-12T00:00:11+00:00" + } + ], + "expected": { + "eligible": true, + "resume_step_sequence": 2, + "completed_count": 1, + "reused_event_types": [ + "ActivityCompleted" + ] + }, + "refusals": [ + { + "event_index": 3, + "field": "activity_execution_id", + "value": "substituted-execution", + "reason": "invalid_activity_retry_history" + }, + { + "event_index": 3, + "field": "activity_attempt_id", + "value": "substituted-attempt", + "reason": "invalid_activity_retry_history" + }, + { + "event_index": 3, + "field": "retry_task_id", + "value": "substituted-task", + "reason": "invalid_activity_retry_history" + }, + { + "event_index": 3, + "field": "retry_after_attempt", + "value": 9, + "reason": "invalid_activity_retry_history" + }, + { + "event_index": 8, + "field": "activity_execution_id", + "value": "substituted-execution", + "reason": "invalid_activity_retry_history" + }, + { + "event_index": 8, + "field": "activity_attempt_id", + "value": "substituted-attempt", + "reason": "invalid_activity_retry_history" + }, + { + "event_index": 8, + "field": "retry_task_id", + "value": "substituted-task", + "reason": "invalid_activity_retry_history" + }, + { + "event_index": 8, + "field": "retry_after_attempt", + "value": 9, + "reason": "invalid_activity_retry_history" + } + ] + }, + "expected": { + "completed": false, + "result": null, + "commands": [ + { + "type": "schedule_activity", + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "payload_codec": "avro" + } + ] + } +} diff --git a/tests/Fixtures/V2/ReplayRegression/redrive-retried-failed-activity.json b/tests/Fixtures/V2/ReplayRegression/redrive-retried-failed-activity.json new file mode 100644 index 000000000..c448c16be --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/redrive-retried-failed-activity.json @@ -0,0 +1,210 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "redrive-retried-failed-activity", + "protocol_version": "1.0", + "bindings": [ + "php" + ], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestRedriveRetryWorkflow", + "arguments": [ + "Taylor", + false + ], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": {}, + "recorded_at": "2026-11-12T00:00:00+00:00" + }, + { + "sequence": 2, + "event_type": "ActivityScheduled", + "payload": { + "sequence": 1, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-1" + }, + "recorded_at": "2026-11-12T00:00:01+00:00" + }, + { + "sequence": 3, + "event_type": "ActivityCompleted", + "payload": { + "sequence": 1, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-1", + "result": "wwHioz3/VYAiNwocSGVsbG8sIFRheWxvciE=", + "payload_codec": "avro" + }, + "recorded_at": "2026-11-12T00:00:02+00:00" + } + ], + "redrive_plan": { + "source_history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": {}, + "recorded_at": "2026-11-12T00:00:00+00:00" + }, + { + "sequence": 2, + "event_type": "ActivityScheduled", + "payload": { + "sequence": 1, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-1" + }, + "recorded_at": "2026-11-12T00:00:01+00:00" + }, + { + "sequence": 3, + "event_type": "ActivityCompleted", + "payload": { + "sequence": 1, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_execution_id": "execution-1", + "result": "wwHioz3/VYAiNwocSGVsbG8sIFRheWxvciE=", + "payload_codec": "avro" + }, + "recorded_at": "2026-11-12T00:00:02+00:00" + }, + { + "sequence": 4, + "event_type": "ActivityScheduled", + "payload": { + "sequence": 2, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_execution_id": "execution-2" + }, + "recorded_at": "2026-11-12T00:00:03+00:00" + }, + { + "sequence": 5, + "event_type": "ActivityStarted", + "payload": { + "sequence": 2, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_execution_id": "execution-2", + "activity_attempt_id": "attempt-2-1", + "attempt_number": 1, + "workflow_task_id": "task-2-1" + }, + "recorded_at": "2026-11-12T00:00:04+00:00" + }, + { + "sequence": 6, + "event_type": "ActivityRetryScheduled", + "payload": { + "sequence": 2, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_execution_id": "execution-2", + "activity_attempt_id": "attempt-2-1", + "retry_after_attempt_id": "attempt-2-1", + "retry_after_attempt": 1, + "retry_of_task_id": "task-2-1", + "retry_task_id": "task-2-2", + "max_attempts": 2, + "retry_reason": "failure", + "retry_backoff_seconds": 0 + }, + "recorded_at": "2026-11-12T00:00:05+00:00" + }, + { + "sequence": 7, + "event_type": "ActivityStarted", + "payload": { + "sequence": 2, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_execution_id": "execution-2", + "activity_attempt_id": "attempt-2-2", + "attempt_number": 2, + "workflow_task_id": "task-2-2" + }, + "recorded_at": "2026-11-12T00:00:06+00:00" + }, + { + "sequence": 8, + "event_type": "ActivityFailed", + "payload": { + "sequence": 2, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "activity_execution_id": "execution-2", + "activity_attempt_id": "attempt-2-2", + "attempt_number": 2, + "message": "provider unavailable", + "exception_class": "RuntimeException" + }, + "recorded_at": "2026-11-12T00:00:07+00:00" + }, + { + "sequence": 9, + "event_type": "WorkflowFailed", + "payload": { + "failed_step_kind": "activity", + "failed_step_sequence": 2 + }, + "recorded_at": "2026-11-12T00:00:08+00:00" + } + ], + "expected": { + "eligible": true, + "resume_step_sequence": 2, + "completed_count": 1, + "reused_event_types": [ + "ActivityCompleted" + ] + }, + "refusals": [ + { + "event_index": 5, + "field": "activity_execution_id", + "value": "substituted-execution", + "reason": "invalid_activity_retry_history" + }, + { + "event_index": 5, + "field": "activity_attempt_id", + "value": "substituted-attempt", + "reason": "invalid_activity_retry_history" + }, + { + "event_index": 5, + "field": "retry_task_id", + "value": "substituted-task", + "reason": "invalid_activity_retry_history" + }, + { + "event_index": 5, + "field": "retry_after_attempt", + "value": 9, + "reason": "invalid_activity_retry_history" + } + ] + }, + "expected": { + "completed": false, + "result": null, + "commands": [ + { + "type": "schedule_activity", + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "payload_codec": "avro" + } + ] + } +} diff --git a/tests/Fixtures/V2/TestRedriveRetryProbeActivity.php b/tests/Fixtures/V2/TestRedriveRetryProbeActivity.php new file mode 100644 index 000000000..09663f1a9 --- /dev/null +++ b/tests/Fixtures/V2/TestRedriveRetryProbeActivity.php @@ -0,0 +1,29 @@ +heartbeat([ + 'site' => $site, + ]); + if ($calls <= (int) config('workflows.v2.testing.redrive_retry_failures.' . $site, 0)) { + throw new RuntimeException('Retryable fixture failure at ' . $site . '.'); + } + return $value . ':' . $site; + } +} diff --git a/tests/Fixtures/V2/TestRedriveRetryProbeWorkflow.php b/tests/Fixtures/V2/TestRedriveRetryProbeWorkflow.php new file mode 100644 index 000000000..73efb5ba9 --- /dev/null +++ b/tests/Fixtures/V2/TestRedriveRetryProbeWorkflow.php @@ -0,0 +1,29 @@ +runWithHistory($this->retryHistory())); + self::assertTrue($plan['eligible'], (string) $plan['reason']); + self::assertSame(2, $plan['resume_step_sequence']); + } + + public function testARetriedCompletedPrefixIsReusable(): void + { + $plan = FailedRunRedrivePlan::forRun($this->runWithHistory($this->completedRetryHistory())); + + self::assertTrue($plan['eligible'], (string) $plan['reason']); + self::assertSame(2, $plan['resume_step_sequence']); + self::assertCount(1, $plan['completed']); + self::assertSame(1, $plan['completed'][0]['sequence']); + } + + public function testHeartbeatsBetweenRetriedPrefixAttemptsDoNotPreventReuse(): void + { + $history = $this->completedRetryHistory(); + foreach ($history as $index => $event) { + if ($event['event_type'] === HistoryEventType::ActivityStarted + && ($event['payload']['sequence'] ?? null) === 1) { + array_splice($history, $index + 1, 0, [[ + 'event_type' => HistoryEventType::ActivityHeartbeatRecorded, + 'payload' => $event['payload'], + ]]); + break; + } + } + $plan = FailedRunRedrivePlan::forRun($this->runWithHistory($history)); + + self::assertTrue($plan['eligible'], (string) $plan['reason']); + self::assertSame(2, $plan['resume_step_sequence']); + self::assertCount(1, $plan['completed']); + } + + public function testARejectedStartDoesNotBecomeActiveAdoptionEvidence(): void + { + $run = $this->adoptionRun(); + $run->historyEvents[2]->event_type = HistoryEventType::StartRejected; + $plan = FailedRunRedrivePlan::forRun($run); + + self::assertFalse($plan['eligible']); + self::assertSame('unsupported_history', $plan['reason']); + } + + public function testRetryIdentitySubstitutionIsRefused(): void + { + foreach (['activity_execution_id', 'activity_attempt_id', 'retry_after_attempt', 'retry_task_id'] as $field) { + $history = $this->retryHistory(); + $history[6]['payload'][$field] = $field === 'retry_after_attempt' ? 9 : 'substituted'; + $plan = FailedRunRedrivePlan::forRun($this->runWithHistory($history)); + self::assertFalse($plan['eligible']); + self::assertSame('invalid_activity_retry_history', $plan['reason']); + } + } + + public function testNativeActiveAdoptionIsHarmlessOnlyWhenBoundToTheSameRunAndCommand(): void + { + $plan = FailedRunRedrivePlan::forRun($this->adoptionRun()); + self::assertTrue($plan['eligible'], (string) $plan['reason']); + self::assertSame(2, $plan['resume_step_sequence']); + } + + public function testSubstitutedActiveStartAdoptionCannotAuthorizeRedrive(): void + { + foreach ([ + ['outcome', 'started_new'], + ['workflow_instance_id', 'other-instance'], + ['workflow_run_id', 'other-run'], + ['workflow_class', 'OtherWorkflow'], + ['workflow_type', 'other.type'], + ['workflow_command_id', 'other-command'], + ['rejection_reason', 'refused'], + ['command', null], + ['command.id', 'other-command'], + ['command.type', 'signal'], + ['command.status', 'rejected'], + ['command.outcome', 'started_new'], + ['command.target_scope', 'run'], + ['command.resolved_run_id', 'other-run'], + ] as [$field, $replacement]) { + $run = $this->adoptionRun(); + $event = $run->historyEvents[2]; + $payload = $event->payload; + $keys = explode('.', $field); + if (count($keys) === 2) { + $payload[$keys[0]][$keys[1]] = $replacement; + } else { + $payload[$field] = $replacement; + } + $event->payload = $payload; + $plan = FailedRunRedrivePlan::forRun($run); + self::assertFalse($plan['eligible'], $field); + self::assertSame('unexpected_start', $plan['reason'], $field); + } + foreach (['workflow_run_id', 'workflow_command_id'] as $field) { + $run = $this->adoptionRun(); + $run->historyEvents[2]->setAttribute($field, 'substituted'); + self::assertSame('unexpected_start', FailedRunRedrivePlan::forRun($run)['reason']); + } + } + + private function adoptionRun(): WorkflowRun + { + $history = $this->linearFailureHistory(); + array_splice($history, 2, 0, [[ + 'event_type' => HistoryEventType::StartAccepted, + 'payload' => [ + 'workflow_instance_id' => 'instance-a', + 'workflow_run_id' => 'run-a', + 'workflow_class' => 'TestWorkflow', + 'workflow_type' => 'test.workflow', + 'workflow_command_id' => 'adoption-command', + 'outcome' => 'returned_existing_active', + 'command' => [ + 'id' => 'adoption-command', + 'type' => 'start', + 'target_scope' => 'instance', + 'resolved_run_id' => 'run-a', + 'status' => 'accepted', + 'outcome' => 'returned_existing_active', + ], + ], + ]]); + $run = $this->runWithHistory($history); + $run->forceFill([ + 'id' => 'run-a', + 'workflow_instance_id' => 'instance-a', + 'workflow_class' => 'TestWorkflow', + 'workflow_type' => 'test.workflow', + ]); + $run->historyEvents[2]->forceFill([ + 'workflow_run_id' => 'run-a', + 'workflow_command_id' => 'adoption-command', + ]); + return $run; + } + + private function retryHistory(): array + { + $history = $this->linearFailureHistory(); + array_splice($history, 4, 1, [ + [ + 'event_type' => HistoryEventType::ActivityScheduled, + 'payload' => [ + 'sequence' => 2, + 'activity_type' => 'second', + 'activity_execution_id' => 'execution-2', + ], + ], + [ + 'event_type' => HistoryEventType::ActivityStarted, + 'payload' => [ + 'sequence' => 2, + 'activity_type' => 'second', + 'activity_execution_id' => 'execution-2', + 'activity_attempt_id' => 'attempt-1', + 'attempt_number' => 1, + 'workflow_task_id' => 'task-1', + ], + ], + [ + 'event_type' => HistoryEventType::ActivityRetryScheduled, + 'payload' => [ + 'sequence' => 2, + 'activity_type' => 'second', + 'activity_execution_id' => 'execution-2', + 'activity_attempt_id' => 'attempt-1', + 'retry_after_attempt_id' => 'attempt-1', + 'retry_after_attempt' => 1, + 'retry_of_task_id' => 'task-1', + 'retry_task_id' => 'task-2', + 'max_attempts' => 2, + ], + ], + [ + 'event_type' => HistoryEventType::ActivityStarted, + 'payload' => [ + 'sequence' => 2, + 'activity_type' => 'second', + 'activity_execution_id' => 'execution-2', + 'activity_attempt_id' => 'attempt-2', + 'attempt_number' => 2, + 'workflow_task_id' => 'task-2', + ], + ], + [ + 'event_type' => HistoryEventType::ActivityFailed, + 'payload' => [ + 'sequence' => 2, + 'activity_type' => 'second', + 'activity_execution_id' => 'execution-2', + 'activity_attempt_id' => 'attempt-2', + 'attempt_number' => 2, + ], + ], + ]); + return $history; + } + + /** + * @return list}> + */ + private function completedRetryHistory(): array + { + $fixture = json_decode( + file_get_contents( + dirname(__DIR__, 2) . '/Fixtures/V2/ReplayRegression/redrive-retried-completed-activity.json' + ), + true, + flags: JSON_THROW_ON_ERROR, + ); + return array_map(static function (array $event): array { + $event['event_type'] = HistoryEventType::from($event['event_type']); + return $event; + }, $fixture['redrive_plan']['source_history']); + } + /** * @return list}> */ diff --git a/tests/Unit/V2/ReplayRegressionCorpusTest.php b/tests/Unit/V2/ReplayRegressionCorpusTest.php index 8f7d514fa..af16aabca 100644 --- a/tests/Unit/V2/ReplayRegressionCorpusTest.php +++ b/tests/Unit/V2/ReplayRegressionCorpusTest.php @@ -24,12 +24,14 @@ use Throwable; use Workflow\Serializers\Serializer; use Workflow\V2\Enums\HistoryEventType; +use Workflow\V2\Enums\RunStatus; use Workflow\V2\Models\WorkflowHistoryEvent; use Workflow\V2\Models\WorkflowRun; use Workflow\V2\Models\WorkflowServiceCall; use Workflow\V2\Support\ActivitySnapshot; use Workflow\V2\Support\CooperativeCancellationDelivery; use Workflow\V2\Support\ExecutionTracing; +use Workflow\V2\Support\FailedRunRedrivePlan; use Workflow\V2\Support\HistoryEventPayloadContract; use Workflow\V2\Support\PortableLocalActivityPreparation; use Workflow\V2\Support\PreparedLocalRetryPolicy; @@ -81,6 +83,9 @@ protected function tearDown(): void public function testFixtureExecutesThroughColdReplayRunner(array $fixture): void { $this->assertHistoryPayloadContract($fixture); + if (isset($fixture['redrive_plan'])) { + $this->assertRetriedActivityRedrivePlan($fixture); + } if (in_array( $fixture['id'], @@ -409,6 +414,40 @@ public static function replayRegressionFixtures(): array return $fixtures; } + private function assertRetriedActivityRedrivePlan(array $fixture): void + { + $scenario = $fixture['redrive_plan']; + $runWithHistory = static function (array $history) use ($scenario): WorkflowRun { + $run = new WorkflowRun([ + ...($scenario['source_run'] ?? []), + 'status' => RunStatus::Failed, + 'payload_codec' => 'avro', + ]); + $run->setRelation('historyEvents', collect($history)->map( + static fn (array $event): WorkflowHistoryEvent => new WorkflowHistoryEvent($event) + )); + return $run; + }; + $plan = FailedRunRedrivePlan::forRun($runWithHistory($scenario['source_history'])); + $expected = $scenario['expected']; + $this->assertSame($expected['eligible'], $plan['eligible'], $fixture['id'] . ': ' . $plan['reason']); + $this->assertNull($plan['reason']); + $this->assertSame($expected['resume_step_sequence'], $plan['resume_step_sequence']); + $this->assertCount($expected['completed_count'], $plan['completed']); + $this->assertSame($expected['reused_event_types'], array_map( + static fn (array $step): string => $step['event_type']->value, + $plan['reused_steps'] + )); + foreach ($scenario['refusals'] as $case) { + $history = $scenario['source_history']; + $history[$case['event_index']]['payload'][$case['field']] = $case['value']; + $rejected = FailedRunRedrivePlan::forRun($runWithHistory($history)); + $this->assertFalse($rejected['eligible'], $fixture['id'] . ': ' . $case['field']); + $this->assertSame($case['reason'], $rejected['reason']); + $this->assertSame([], $rejected['reused_steps']); + } + } + private function assertPreparedV1RefusesTheVersionedPolicy(array $fixture): void { $policy = $fixture['unqualified_preparation_policy']; From c0f18b322e6f2b00c2559a9027474f7d296c8bd3 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 14:50:14 +0000 Subject: [PATCH 2/4] Exercise redrive retries with supported queue and heartbeat fixtures --- tests/Feature/V2/V2FailedRunRedriveTest.php | 2 ++ tests/Fixtures/V2/TestRedriveRetryProbeActivity.php | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/Feature/V2/V2FailedRunRedriveTest.php b/tests/Feature/V2/V2FailedRunRedriveTest.php index f1bed9890..50e44934b 100644 --- a/tests/Feature/V2/V2FailedRunRedriveTest.php +++ b/tests/Feature/V2/V2FailedRunRedriveTest.php @@ -203,6 +203,8 @@ public static function retriedActivityCases(): array #[DataProvider('retriedActivityCases')] public function testOrdinaryRetriedActivitiesRedriveWithoutRepeatingTheCompletedPrefix(int $firstFailures): void { + config() + ->set('queue.default', 'redis'); Queue::fake(); config([ 'workflows.v2.testing.redrive_retry_failures' => [ diff --git a/tests/Fixtures/V2/TestRedriveRetryProbeActivity.php b/tests/Fixtures/V2/TestRedriveRetryProbeActivity.php index 09663f1a9..c55db17fc 100644 --- a/tests/Fixtures/V2/TestRedriveRetryProbeActivity.php +++ b/tests/Fixtures/V2/TestRedriveRetryProbeActivity.php @@ -19,7 +19,7 @@ public function handle(string $site, string $value, string $counter): string { $calls = Cache::increment($counter . ':' . $site); $this->heartbeat([ - 'site' => $site, + 'message' => $site, ]); if ($calls <= (int) config('workflows.v2.testing.redrive_retry_failures.' . $site, 0)) { throw new RuntimeException('Retryable fixture failure at ' . $site . '.'); From e41807ade65f3dfd0a43ecbbd3cd5124942a42cf Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 15:05:28 +0000 Subject: [PATCH 3/4] Require closed attempt ownership grammar in retry histories --- src/V2/Support/ActivityRetryHistory.php | 93 ++++++++++++++++++++----- 1 file changed, 74 insertions(+), 19 deletions(-) diff --git a/src/V2/Support/ActivityRetryHistory.php b/src/V2/Support/ActivityRetryHistory.php index dd9b84516..c675b994a 100644 --- a/src/V2/Support/ActivityRetryHistory.php +++ b/src/V2/Support/ActivityRetryHistory.php @@ -22,31 +22,40 @@ public static function valid(array $events): bool if (! $hasRetries) { return true; } + $phase = 'schedule'; $execution = null; $attempt = null; $task = null; $number = 0; $nextTask = null; - $terminal = false; $seenAttempts = []; $seenTasks = []; foreach ($events as $event) { $payload = $event->payload; $id = $payload['activity_execution_id'] ?? null; - if (! is_string($id) || $id === '' || ($execution !== null && $id !== $execution) || $terminal) { + if ( + ! is_string($id) + || $id === '' + || ($execution !== null && $id !== $execution) + || $phase === 'terminal' + ) { return false; } $execution = $id; - if ($event->event_type === HistoryEventType::ActivityStarted) { - $currentAttempt = $payload['activity_attempt_id'] ?? null; - $currentTask = $event->workflow_task_id ?? ($payload['workflow_task_id'] ?? null); - if (isset($event->workflow_task_id, $payload['workflow_task_id']) && $event->workflow_task_id !== $payload['workflow_task_id']) { + if ($event->event_type === HistoryEventType::ActivityScheduled) { + if ($phase !== 'schedule') { return false; } - if (! is_string($currentAttempt) || $currentAttempt === '' || isset($seenAttempts[$currentAttempt]) + $phase = 'start'; + } elseif ($event->event_type === HistoryEventType::ActivityStarted) { + $currentAttempt = $payload['activity_attempt_id'] ?? null; + $currentTask = $event->workflow_task_id ?? ($payload['workflow_task_id'] ?? null); + if (! in_array($phase, ['start', 'retry'], true) + || ! is_string($currentAttempt) || $currentAttempt === '' || isset($seenAttempts[$currentAttempt]) || ! is_string($currentTask) || $currentTask === '' || isset($seenTasks[$currentTask]) || ($payload['attempt_number'] ?? null) !== $number + 1 - || ($number > 0 && $nextTask !== $currentTask)) { + || ($number > 0 && $nextTask !== $currentTask) + || ! self::matchesTask($event, $currentTask)) { return false; } $attempt = $currentAttempt; @@ -55,31 +64,77 @@ public static function valid(array $events): bool $seenTasks[$task] = true; ++$number; $nextTask = null; + $phase = 'active'; } elseif ($event->event_type === HistoryEventType::ActivityRetryScheduled) { $retry = $payload['retry_task_id'] ?? null; $maximum = $payload['max_attempts'] ?? null; - if ($attempt === null || $nextTask !== null + if ($phase !== 'active' || ($payload['activity_attempt_id'] ?? null) !== $attempt || ($payload['retry_after_attempt_id'] ?? null) !== $attempt || ($payload['retry_after_attempt'] ?? null) !== $number || ($payload['retry_of_task_id'] ?? null) !== $task || ! is_string($retry) || $retry === '' || isset($seenTasks[$retry]) - || ($maximum !== null && (! is_int($maximum) || $maximum <= $number))) { + || ($maximum !== null && (! is_int($maximum) || $maximum <= $number)) + || ! self::matchesTask($event, $task)) { return false; } $nextTask = $retry; - } elseif (in_array( - $event->event_type, - [HistoryEventType::ActivityCompleted, HistoryEventType::ActivityFailed], - true - )) { - if ($nextTask !== null || $attempt === null || ($payload['activity_attempt_id'] ?? null) !== $attempt - || ($payload['attempt_number'] ?? null) !== $number) { + $phase = 'retry'; + } elseif (in_array($event->event_type, [ + HistoryEventType::ActivityHeartbeatRecorded, + HistoryEventType::ActivityCompleted, + HistoryEventType::ActivityFailed, + ], true)) { + if ($phase !== 'active' || ($payload['activity_attempt_id'] ?? null) !== $attempt + || ($payload['attempt_number'] ?? null) !== $number + || ! self::matchesTask($event, $task)) { return false; } - $terminal = true; + if ($event->event_type !== HistoryEventType::ActivityHeartbeatRecorded) { + $phase = 'terminal'; + } + } else { + return false; + } + if (in_array($phase, ['active', 'retry', 'terminal'], true) + && ! self::matchesAttemptSnapshot($payload, $execution, $attempt, $number, $task)) { + return false; + } + } + return $phase === 'terminal'; + } + + private static function matchesTask(WorkflowHistoryEvent $event, string $task): bool + { + $payload = $event->payload; + return ($event->workflow_task_id === null || $event->workflow_task_id === $task) + && (! array_key_exists('workflow_task_id', $payload) || $payload['workflow_task_id'] === $task); + } + + private static function matchesAttemptSnapshot( + array $payload, + string $execution, + string $attempt, + int $number, + string $task, + ): bool { + if (! array_key_exists('activity_attempt', $payload)) { + return true; + } + $snapshot = $payload['activity_attempt']; + if (! is_array($snapshot)) { + return false; + } + foreach ([ + 'id' => $attempt, + 'activity_execution_id' => $execution, + 'attempt_number' => $number, + 'task_id' => $task, + ] as $field => $expected) { + if (array_key_exists($field, $snapshot) && $snapshot[$field] !== $expected) { + return false; } } - return $terminal; + return true; } } From 42dd4e3349ce79c26d201a813a4c37ee3d23df28 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 15:04:59 +0000 Subject: [PATCH 4/4] Reject malformed heartbeat and schedule retry histories --- tests/Unit/V2/ActivityRetryHistoryTest.php | 184 +++++++++++++++++++++ 1 file changed, 184 insertions(+) create mode 100644 tests/Unit/V2/ActivityRetryHistoryTest.php diff --git a/tests/Unit/V2/ActivityRetryHistoryTest.php b/tests/Unit/V2/ActivityRetryHistoryTest.php new file mode 100644 index 000000000..16206b9cc --- /dev/null +++ b/tests/Unit/V2/ActivityRetryHistoryTest.php @@ -0,0 +1,184 @@ +previousContainer = Container::getInstance(); + $application = new Application(dirname(__DIR__, 3)); + $application->instance('config', new Repository([])); + } + + protected function tearDown(): void + { + Container::setInstance($this->previousContainer); + parent::tearDown(); + } + + public static function terminalKinds(): array + { + return [ + 'completed prefix' => [HistoryEventType::ActivityCompleted], + 'failed boundary' => [HistoryEventType::ActivityFailed], + ]; + } + + #[DataProvider('terminalKinds')] + public function testCanonicalRetryChainBindsEachHeartbeatToItsActiveAttempt(HistoryEventType $terminal): void + { + self::assertTrue(ActivityRetryHistory::valid(self::models(self::history($terminal)))); + } + + public static function malformedChains(): iterable + { + foreach (self::terminalKinds() as $kind => [$terminal]) { + foreach ([ + 'heartbeat wrong attempt', + 'heartbeat wrong number', + 'heartbeat wrong payload task', + 'heartbeat wrong row task', + 'heartbeat wrong snapshot task', + 'heartbeat before start', + 'heartbeat after retry', + 'duplicate schedule', + 'mid-chain schedule', + 'missing schedule', + ] as $mutation) { + yield $kind . ': ' . $mutation => [$terminal, $mutation]; + } + } + } + + #[DataProvider('malformedChains')] + public function testMalformedChainCannotAuthorizeRedrive(HistoryEventType $terminal, string $mutation): void + { + $history = self::history($terminal); + switch ($mutation) { + case 'heartbeat wrong attempt': + $history[2]['payload']['activity_attempt_id'] = 'other-attempt'; + break; + case 'heartbeat wrong number': + $history[2]['payload']['attempt_number'] = 2; + break; + case 'heartbeat wrong payload task': + $history[2]['payload']['workflow_task_id'] = 'other-task'; + break; + case 'heartbeat wrong row task': + $history[2]['workflow_task_id'] = 'other-task'; + break; + case 'heartbeat wrong snapshot task': + $history[2]['payload']['activity_attempt']['task_id'] = 'other-task'; + break; + case 'heartbeat before start': + $heartbeat = array_splice($history, 2, 1); + array_splice($history, 1, 0, $heartbeat); + break; + case 'heartbeat after retry': + $heartbeat = array_splice($history, 2, 1); + array_splice($history, 3, 0, $heartbeat); + break; + case 'duplicate schedule': + array_splice($history, 1, 0, [$history[0]]); + break; + case 'mid-chain schedule': + array_splice($history, 4, 0, [$history[0]]); + break; + case 'missing schedule': + array_shift($history); + break; + } + self::assertFalse(ActivityRetryHistory::valid(self::models($history)), $mutation); + } + + private static function history(HistoryEventType $terminal): array + { + $identity = [ + 'activity_execution_id' => 'execution-1', + 'activity_type' => 'test.activity', + 'sequence' => 1, + ]; + $attempt = static fn (int $number): array => [ + ...$identity, + 'activity_attempt_id' => 'attempt-' . $number, + 'attempt_number' => $number, + ]; + $heartbeat = static fn (int $number): array => [ + ...$attempt($number), + 'activity_attempt' => [ + 'id' => 'attempt-' . $number, + 'attempt_number' => $number, + 'task_id' => 'task-' . $number, + ], + ]; + return [ + [ + 'event_type' => HistoryEventType::ActivityScheduled, + 'payload' => $identity, + ], + [ + 'event_type' => HistoryEventType::ActivityStarted, + 'workflow_task_id' => 'task-1', + 'payload' => [ + ...$attempt(1), + 'workflow_task_id' => 'task-1', + ], + ], + [ + 'event_type' => HistoryEventType::ActivityHeartbeatRecorded, + 'workflow_task_id' => 'task-1', + 'payload' => $heartbeat(1), + ], + [ + 'event_type' => HistoryEventType::ActivityRetryScheduled, + 'workflow_task_id' => 'task-1', + 'payload' => [ + ...$attempt(1), + 'retry_after_attempt_id' => 'attempt-1', + 'retry_after_attempt' => 1, + 'retry_of_task_id' => 'task-1', + 'retry_task_id' => 'task-2', + 'max_attempts' => 2, + ], + ], + [ + 'event_type' => HistoryEventType::ActivityStarted, + 'workflow_task_id' => 'task-2', + 'payload' => [ + ...$attempt(2), + 'workflow_task_id' => 'task-2', + ], + ], + [ + 'event_type' => HistoryEventType::ActivityHeartbeatRecorded, + 'workflow_task_id' => 'task-2', + 'payload' => $heartbeat(2), + ], + [ + 'event_type' => $terminal, + 'workflow_task_id' => 'task-2', + 'payload' => $attempt(2), + ], + ]; + } + + private static function models(array $history): array + { + return array_map(static fn (array $event): WorkflowHistoryEvent => new WorkflowHistoryEvent($event), $history); + } +}