diff --git a/CHANGELOG.md b/CHANGELOG.md index 41a4b9d63..14a5a26d9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ # Changelog +## 2.13.1 - 2026-10-24 + +- Cancel the live task when a retried activity receives cooperative cancellation. + Preserve completed tasks and snapshot the cancelled attempt's matching task so + its worker can acknowledge the stop and repeat that acknowledgement safely. +- Apply the same task selection when waiting for cancellation completion. + ## 2.13.0 - 2026-10-24 - Recognize bounded `ActivityOutcomeRejected` diagnostics in history pages, diff --git a/composer.json b/composer.json index 6286efe51..0edb9efc8 100644 --- a/composer.json +++ b/composer.json @@ -87,7 +87,7 @@ "dev-main": "2.0.x-dev" }, "durable-workflow": { - "product-train": "2.13.0", + "product-train": "2.13.1", "laravel-embedded-upgrade-contract": "resources/laravel-embedded-upgrade-contract.json", "laravel-dependency-security-policy": "resources/laravel-dependency-security-policy.json" }, diff --git a/src/V2/Support/ActivityCancellationWait.php b/src/V2/Support/ActivityCancellationWait.php index 36e42fdf6..6768609eb 100644 --- a/src/V2/Support/ActivityCancellationWait.php +++ b/src/V2/Support/ActivityCancellationWait.php @@ -46,7 +46,8 @@ public static function prepare(WorkflowRun $run, WorkflowTask $workflowTask, Act if (in_array($execution->status, [ActivityStatus::Pending, ActivityStatus::Running], true)) { $activityTask = $run->tasks->first(static fn (WorkflowTask $task): bool => $task->task_type === TaskType::Activity - && ($task->payload['activity_execution_id'] ?? null) === $execution->id); + && ($task->payload['activity_execution_id'] ?? null) === $execution->id + && in_array($task->status, [TaskStatus::Ready, TaskStatus::Leased], true)); $event = ActivityCancellation::record($run, $execution, $activityTask, $context->requestId); if ($event !== null) { $run->historyEvents->push($event); diff --git a/src/V2/Support/WorkflowExecutor.php b/src/V2/Support/WorkflowExecutor.php index b65285f66..5a5ab5882 100644 --- a/src/V2/Support/WorkflowExecutor.php +++ b/src/V2/Support/WorkflowExecutor.php @@ -5831,7 +5831,8 @@ private function cancelOpenWaitAtSequence( /** @var WorkflowTask|null $activityTask */ $activityTask = $run->tasks->first( static fn (WorkflowTask $candidate): bool => $candidate->task_type === TaskType::Activity - && ($candidate->payload['activity_execution_id'] ?? null) === $execution->id, + && ($candidate->payload['activity_execution_id'] ?? null) === $execution->id + && in_array($candidate->status, [TaskStatus::Ready, TaskStatus::Leased], true), ); ActivityCancellation::record( $run, diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index 93749b3da..b2e788402 100644 --- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php +++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php @@ -24,8 +24,10 @@ use Throwable; use Workflow\Serializers\Serializer; use Workflow\V2\Activity; +use Workflow\V2\Contracts\ActivityTaskBridge; use Workflow\V2\Contracts\HistoryProjectionRole; use Workflow\V2\Contracts\WorkflowTaskBridge; +use Workflow\V2\Enums\CancellationPolicy; use Workflow\V2\Enums\HistoryEventType; use Workflow\V2\Enums\RunStatus; use Workflow\V2\Enums\TaskStatus; @@ -52,6 +54,7 @@ use Workflow\V2\Models\WorkflowTask; use Workflow\V2\Models\WorkflowTimelineEntry; use Workflow\V2\StartOptions; +use Workflow\V2\Support\ActivityCancellationAcknowledgement; use Workflow\V2\Support\CancellationScopeDelivery; use Workflow\V2\Support\CancellationScopeHistory; use Workflow\V2\Support\CancellationScopeRequests; @@ -64,6 +67,7 @@ use Workflow\V2\Support\ReplayDiff; use Workflow\V2\Support\RunActivityView; use Workflow\V2\Support\RunTimelineProjector; +use Workflow\V2\Support\WorkflowCommandNormalizer; use Workflow\V2\Support\WorkflowFiberRunner; use Workflow\V2\Support\WorkflowReplayer; use Workflow\V2\Support\WorkflowStep; @@ -81,6 +85,124 @@ final class V2EmbeddedReplayRegressionCorpusTest extends TestCase private int $workflowNumber = 0; + public function testRetriedActivityCancellationCorpusKeepsTheCompletedTaskAndAcknowledgesTheLiveTask(): void + { + $path = self::FIXTURE_DIR . '/retried-activity-cancellation-live-task.json'; + if (! is_file($path)) { + $this->markTestSkipped('The retained retried-activity cancellation evidence is absent.'); + } + $fixture = json_decode(file_get_contents($path), true, flags: JSON_THROW_ON_ERROR); + self::stopWorkers(); + Queue::fake(); + config([ + 'queue.default' => 'database', + 'workflows.v2.compatibility.current' => 'build-a', + 'workflows.v2.compatibility.supported' => ['build-a'], + ]); + $this->executeColdReplayFixture($fixture); + $scenario = $fixture['retry_cancellation']; + $scheduled = collect($fixture['history'])->firstWhere('event_type', 'ActivityScheduled')['payload']; + $recordedCancellation = collect($fixture['history'])->firstWhere('event_type', 'ActivityCancelled')['payload']; + $bridge = $this->app->make(DefaultWorkflowTaskBridge::class); + $activities = $this->app->make(ActivityTaskBridge::class); + $lease = static function (WorkflowTask $task): WorkflowTask { + $task->forceFill([ + 'status' => TaskStatus::Leased, + 'lease_owner' => 'portable-worker', + 'lease_expires_at' => now() + ->addMinutes(5), + ])->save(); + return $task; + }; + foreach ($scenario['policies'] as $policyValue) { + $this->clearWorkflowState(); + $policy = CancellationPolicy::from($policyValue); + $stub = WorkflowStub::make( + $fixture['workflow']['type'], + 'corpus-retried-cancellation-' . ++$this->workflowNumber + ); + $stub->start(...$fixture['workflow']['arguments']); + $run = WorkflowRun::query()->findOrFail($stub->runId()); + $initial = $lease($run->tasks()->where('task_type', TaskType::Workflow)->sole()); + $commands = WorkflowCommandNormalizer::normalize([[ + 'type' => 'schedule_activity', + 'activity_type' => $scheduled['activity_type'], + 'arguments' => Serializer::serialize($scenario['activity_arguments']), + 'cancellation_policy' => $policy->value, + 'retry_policy' => $scenario['retry_policy'], + ]], $fixture['protocol_version']); + $this->assertTrue($bridge->complete($initial->id, $commands)['completed']); + $execution = $run->activityExecutions() + ->sole(); + $firstTask = $run->tasks() + ->where('task_type', TaskType::Activity)->sole(); + $this->assertTrue($activities->claimStatus($firstTask->id, 'first-owner')['claimed']); + $firstAttempt = $execution->attempts() + ->sole(); + $retry = $activities->fail($firstAttempt->id, [ + 'class' => \RuntimeException::class, + 'message' => 'temporary failure', + ]); + $this->assertTrue($retry['recorded']); + $liveTask = WorkflowTask::query()->findOrFail($retry['next_task_id']); + $this->assertTrue($activities->claimStatus($liveTask->id, 'replacement-owner')['claimed']); + $liveAttempt = $execution->attempts() + ->where('attempt_number', $recordedCancellation['attempt_number'])->sole(); + $this->assertSame($scenario['completed_task_status'], $firstTask->refresh()->status->value); + $completedBefore = $firstTask->getRawOriginal(); + $this->assertTrue( + $stub->requestCancellation('maintenance', $scenario['cleanup_timeout_seconds'])->accepted() + ); + $run->refresh(); + $deadline = $run->cancellation_deadline_at->toISOString(); + $ready = static fn (): WorkflowTask => $run->tasks() + ->where('task_type', TaskType::Workflow)->where('status', TaskStatus::Ready)->sole(); + $deliver = static fn (WorkflowTask $task): array => $bridge->deliverCancellation( + $task->id, + $run->cancellation_request_command_id, + $scheduled['sequence'], + 'activity' + ); + $delivery = $deliver($lease($ready())); + $this->assertSame($policy === CancellationPolicy::TryCancel, $delivery['delivered']); + if ($policy === CancellationPolicy::WaitCancellationCompleted) { + $this->assertTrue($delivery['claim_released']); + $this->assertSame('cancellation_waiting_for_activity', $delivery['reason']); + } + $this->assertSame($completedBefore, $firstTask->refresh()->getRawOriginal()); + $this->assertSame($scenario['live_task_status'], $liveTask->refresh()->status->value); + $this->assertNull($liveTask->lease_expires_at); + $cancelled = $run->historyEvents() + ->where('event_type', HistoryEventType::ActivityCancelled)->sole(); + $this->assertSame($liveTask->id, $cancelled->payload['task']['id']); + $this->assertSame($liveAttempt->id, $cancelled->payload['activity_attempt']['id']); + $accepted = ActivityCancellationAcknowledgement::recordStopped( + $liveAttempt->id, + 'replacement-owner', + $run->cancellation_request_command_id + ); + $this->assertTrue($accepted['acknowledged']); + $this->assertFalse($accepted['duplicate']); + $duplicate = ActivityCancellationAcknowledgement::recordStopped( + $liveAttempt->id, + 'replacement-owner', + $run->cancellation_request_command_id + ); + $this->assertTrue($duplicate['acknowledged']); + $this->assertTrue($duplicate['duplicate']); + $this->assertSame($accepted['history_event_id'], $duplicate['history_event_id']); + $this->assertSame( + 1, + $run->historyEvents() + ->where('event_type', HistoryEventType::ActivityCancellationAcknowledged)->count() + ); + if ($policy === CancellationPolicy::WaitCancellationCompleted) { + $this->assertTrue($deliver($lease($ready()))['delivered']); + } + $this->assertSame($deadline, $run->refresh()->cancellation_deadline_at->toISOString()); + } + } + #[RunInSeparateProcess] #[PreserveGlobalState(false)] public function testCommittedTimerContinuationUsesOneWorkflowConstruction(): void @@ -187,6 +309,11 @@ public function testFixturesExecuteThroughDeclaredReplayConsumers(): void foreach ($this->fixtures() as $fixture) { $this->executeColdReplayFixture($fixture); + if ($fixture['id'] === 'retried-activity-cancellation-live-task') { + // Its live retry/delivery proof is exercised by the dedicated corpus test. + continue; + } + if ($fixture['id'] === 'embedded-committed-timer-continuation') { // Its positive executor path runs in the isolated queue test above. continue; diff --git a/tests/Fixtures/V2/ReplayRegression/retried-activity-cancellation-live-task.json b/tests/Fixtures/V2/ReplayRegression/retried-activity-cancellation-live-task.json new file mode 100644 index 000000000..83647c134 --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/retried-activity-cancellation-live-task.json @@ -0,0 +1,135 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "retried-activity-cancellation-live-task", + "protocol_version": "1.20", + "bindings": [ + "php" + ], + "consumers": [ + "workflow-fiber-runner", + "workflow-executor" + ], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestCooperativeActivityCleanupWorkflow", + "arguments": [], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": {}, + "recorded_at": "2026-10-24T01:00:00Z" + }, + { + "sequence": 2, + "event_type": "ActivityScheduled", + "payload": { + "sequence": 1, + "activity_execution_id": "corpus-activity", + "activity_class": "Tests\\Fixtures\\V2\\TestFinallyWorkActivity", + "activity_type": "Tests\\Fixtures\\V2\\TestFinallyWorkActivity" + }, + "recorded_at": "2026-10-24T01:00:01Z" + }, + { + "sequence": 3, + "event_type": "ActivityStarted", + "payload": { + "sequence": 1, + "activity_execution_id": "corpus-activity", + "activity_class": "Tests\\Fixtures\\V2\\TestFinallyWorkActivity", + "activity_type": "Tests\\Fixtures\\V2\\TestFinallyWorkActivity", + "activity_attempt_id": "corpus-attempt-1", + "attempt_number": 1 + }, + "recorded_at": "2026-10-24T01:00:02Z" + }, + { + "sequence": 4, + "event_type": "ActivityRetryScheduled", + "payload": { + "sequence": 1, + "activity_execution_id": "corpus-activity", + "activity_class": "Tests\\Fixtures\\V2\\TestFinallyWorkActivity", + "activity_type": "Tests\\Fixtures\\V2\\TestFinallyWorkActivity", + "activity_attempt_id": "corpus-attempt-1" + }, + "recorded_at": "2026-10-24T01:00:03Z" + }, + { + "sequence": 5, + "event_type": "ActivityStarted", + "payload": { + "sequence": 1, + "activity_execution_id": "corpus-activity", + "activity_class": "Tests\\Fixtures\\V2\\TestFinallyWorkActivity", + "activity_type": "Tests\\Fixtures\\V2\\TestFinallyWorkActivity", + "activity_attempt_id": "corpus-attempt-2", + "attempt_number": 2 + }, + "recorded_at": "2026-10-24T01:00:04Z" + }, + { + "sequence": 6, + "event_type": "CooperativeCancellationRequested", + "payload": { + "workflow_command_id": "corpus-cancel", + "command_type": "request_cancellation" + }, + "recorded_at": "2026-10-24T01:00:05Z" + }, + { + "sequence": 7, + "event_type": "ActivityCancelled", + "payload": { + "sequence": 1, + "activity_execution_id": "corpus-activity", + "activity_class": "Tests\\Fixtures\\V2\\TestFinallyWorkActivity", + "activity_type": "Tests\\Fixtures\\V2\\TestFinallyWorkActivity", + "activity_attempt_id": "corpus-attempt-2", + "attempt_number": 2 + }, + "recorded_at": "2026-10-24T01:00:06Z" + }, + { + "sequence": 8, + "event_type": "CooperativeCancellationDelivered", + "payload": { + "workflow_command_id": "corpus-cancel", + "sequence": 1, + "call_kind": "activity" + }, + "recorded_at": "2026-10-24T01:00:07Z" + } + ], + "expected": { + "completed": false, + "result": null, + "commands": [ + { + "type": "schedule_activity", + "activity_type": "Tests\\Fixtures\\V2\\TestGreetingActivity" + } + ] + }, + "retry_cancellation": { + "policies": [ + "try_cancel", + "wait_cancellation_completed" + ], + "activity_arguments": [ + false + ], + "retry_policy": { + "max_attempts": 2, + "backoff_seconds": [ + 0 + ] + }, + "cleanup_timeout_seconds": 60, + "completed_task_status": "completed", + "live_task_status": "cancelled" + } +}