From 76d897f62f5028373bf2440f15e49ca5d5ee9c81 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Wed, 9 Sep 2026 18:47:12 +0000 Subject: [PATCH 1/3] Preserve activity timeout deadlines across lease repair --- src/V2/Support/ActivityTaskClaimer.php | 6 ++ src/V2/Support/ActivityTimeoutEnforcer.php | 13 ++- tests/Unit/V2/ActivityLeaseTimeoutTest.php | 117 +++++++++++++++++++++ 3 files changed, 135 insertions(+), 1 deletion(-) create mode 100644 tests/Unit/V2/ActivityLeaseTimeoutTest.php diff --git a/src/V2/Support/ActivityTaskClaimer.php b/src/V2/Support/ActivityTaskClaimer.php index ed0ddaf3..2e8e22e4 100644 --- a/src/V2/Support/ActivityTaskClaimer.php +++ b/src/V2/Support/ActivityTaskClaimer.php @@ -116,6 +116,12 @@ public static function claimDetailed( } $now = now(); + if (ActivityTimeoutEnforcer::hasExpiredDeadline($execution, $now)) { + // The normal timeout sweep owns retry/failure recording. Do not + // replace the expired deadline while waiting for that sweep. + return self::claimFailure('task_not_claimable'); + } + $attemptId = (string) Str::ulid(); $attemptCount = ((int) $task->attempt_count) + 1; diff --git a/src/V2/Support/ActivityTimeoutEnforcer.php b/src/V2/Support/ActivityTimeoutEnforcer.php index 48b12014..ac24834c 100644 --- a/src/V2/Support/ActivityTimeoutEnforcer.php +++ b/src/V2/Support/ActivityTimeoutEnforcer.php @@ -180,6 +180,15 @@ public static function enforce(string $executionId): array } } + /** + * @internal Shared with task claiming so a new attempt cannot erase an elapsed deadline. + */ + public static function hasExpiredDeadline(ActivityExecution $execution, CarbonInterface $now): bool + { + return in_array($execution->status, [ActivityStatus::Pending, ActivityStatus::Running], true) + && self::resolveTimeoutKind($execution, $now) !== null; + } + private static function deadlineBoundary(CarbonInterface $deadline): string { return $deadline->format((new ActivityExecution())->getDateFormat()); @@ -566,7 +575,9 @@ private static function currentAttemptStateReason( return 'current_attempt_changed'; } - if ($attempt->status !== ActivityAttemptStatus::Running) { + // Lease repair closes the current attempt before the timeout sweep. + // It does not cancel the execution's deadlines or change its identity. + if (! in_array($attempt->status, [ActivityAttemptStatus::Running, ActivityAttemptStatus::Expired], true)) { return 'current_attempt_not_running'; } diff --git a/tests/Unit/V2/ActivityLeaseTimeoutTest.php b/tests/Unit/V2/ActivityLeaseTimeoutTest.php new file mode 100644 index 00000000..b7c5af0e --- /dev/null +++ b/tests/Unit/V2/ActivityLeaseTimeoutTest.php @@ -0,0 +1,117 @@ + + */ + public static function repairModes(): iterable + { + yield 'ordinary repair pass' => [false]; + yield 'already repaired by an older worker' => [true]; + } + + #[DataProvider('repairModes')] + public function testExpiredLeaseCannotBypassOverallTimeout(bool $alreadyRepaired): void + { + Queue::fake(); + Carbon::setTestNow('2026-09-09 00:00:00'); + try { + $task = $this->scheduleActivity(); + [$claim] = ActivityTaskClaimer::claim($task->id); + $this->assertNotNull($claim); + + // The worker disappeared after its claim committed, before recording an outcome. + Carbon::setTestNow(now()->addMinutes(12)); + if ($alreadyRepaired) { + TaskRepair::recoverExistingTask($task->fresh(), $claim->run->fresh()); + $this->assertNull(ActivityTaskClaimer::claim($task->id)[0]); + } + + $report = TaskWatchdog::runPass(); + $this->assertSame(1, $report['activity_timeouts_enforced']); + $this->assertSame([], $report['activity_timeout_failures']); + $this->assertSame(ActivityStatus::Failed, $claim->execution->fresh()->status); + $this->assertSame(TaskStatus::Cancelled, $task->fresh()->status); + $this->assertSame(1, ActivityAttempt::count()); + $event = WorkflowHistoryEvent::where('event_type', HistoryEventType::ActivityTimedOut)->sole(); + $this->assertSame('schedule_to_close', $event->payload['timeout_kind']); + + $late = ActivityOutcomeRecorder::record( + $task->id, + $claim->attemptId(), + $claim->attemptNumber(), + 'late', + null, + 1, + 0 + ); + $this->assertFalse($late['recorded']); + $this->assertSame('stale_attempt', $late['reason']); + TaskWatchdog::runPass(); + $this->assertSame( + 1, + WorkflowHistoryEvent::where('event_type', HistoryEventType::ActivityTimedOut)->count() + ); + $this->assertSame( + 0, + WorkflowHistoryEvent::where('event_type', HistoryEventType::ActivityCompleted)->count() + ); + } finally { + Carbon::setTestNow(); + } + } + + public function testPendingActivityCannotBeClaimedAfterItsOverallDeadline(): void + { + Queue::fake(); + Carbon::setTestNow('2026-09-09 00:00:00'); + try { + $task = $this->scheduleActivity(); + Carbon::setTestNow(now()->addMinute()); + $this->assertNull(ActivityTaskClaimer::claim($task->id)[0]); + $this->assertSame(0, ActivityAttempt::count()); + $this->assertSame(1, TaskWatchdog::runPass()['activity_timeouts_enforced']); + } finally { + Carbon::setTestNow(); + } + } + + private function scheduleActivity(): WorkflowTask + { + config([ + 'queue.default' => 'database', + ]); + WorkflowStub::make(TestActivityTimeoutCleanupWorkflow::class, 'lease-timeout')->start(); + $workflowTask = WorkflowTask::where('task_type', TaskType::Workflow)->where( + 'status', + TaskStatus::Ready + )->sole(); + $this->app->call([new RunWorkflowTask($workflowTask->id), 'handle']); + + return WorkflowTask::where('task_type', TaskType::Activity)->where('status', TaskStatus::Ready)->sole(); + } +} From 2b8ad8cfdf420a2c0ffe29466269e309961e4d51 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Wed, 9 Sep 2026 18:53:50 +0000 Subject: [PATCH 2/3] Run lease deadline regressions in the MySQL feature selection --- tests/Feature/V2/V2ActivityTimeoutTest.php | 98 +++++++++++++++++ tests/Unit/V2/ActivityLeaseTimeoutTest.php | 117 --------------------- 2 files changed, 98 insertions(+), 117 deletions(-) delete mode 100644 tests/Unit/V2/ActivityLeaseTimeoutTest.php diff --git a/tests/Feature/V2/V2ActivityTimeoutTest.php b/tests/Feature/V2/V2ActivityTimeoutTest.php index 26106698..a7acac7d 100644 --- a/tests/Feature/V2/V2ActivityTimeoutTest.php +++ b/tests/Feature/V2/V2ActivityTimeoutTest.php @@ -6,6 +6,9 @@ use Illuminate\Support\Carbon; use Illuminate\Support\Facades\DB; +use Illuminate\Support\Facades\Queue; +use PHPUnit\Framework\Attributes\DataProvider; +use Tests\Fixtures\V2\TestActivityTimeoutCleanupWorkflow; use Tests\Fixtures\V2\TestGreetingActivity; use Tests\Fixtures\V2\TestGreetingWorkflow; use Tests\TestCase; @@ -20,6 +23,7 @@ use Workflow\V2\Enums\RunStatus; use Workflow\V2\Enums\TaskStatus; use Workflow\V2\Enums\TaskType; +use Workflow\V2\Jobs\RunWorkflowTask; use Workflow\V2\Models\ActivityAttempt; use Workflow\V2\Models\ActivityExecution; use Workflow\V2\Models\WorkflowFailure; @@ -28,15 +32,94 @@ use Workflow\V2\Models\WorkflowRun; use Workflow\V2\Models\WorkflowRunSummary; use Workflow\V2\Models\WorkflowTask; +use Workflow\V2\Support\ActivityOutcomeRecorder; +use Workflow\V2\Support\ActivityTaskClaimer; use Workflow\V2\Support\ActivityTimeoutEnforcer; use Workflow\V2\Support\DefaultHistoryProjectionRole; use Workflow\V2\Support\FailureSnapshots; use Workflow\V2\Support\RunActivityView; use Workflow\V2\Support\RuntimeObjectFactory; +use Workflow\V2\Support\TaskRepair; use Workflow\V2\TaskWatchdog; +use Workflow\V2\WorkflowStub; final class V2ActivityTimeoutTest extends TestCase { + /** + * @return iterable + */ + public static function repairModes(): iterable + { + yield 'ordinary repair pass' => [false]; + yield 'already repaired by an older worker' => [true]; + } + + #[DataProvider('repairModes')] + public function testExpiredLeaseCannotBypassOverallTimeout(bool $alreadyRepaired): void + { + Queue::fake(); + Carbon::setTestNow('2026-09-09 00:00:00'); + try { + $task = $this->scheduleLeaseTimeoutActivity(); + [$claim] = ActivityTaskClaimer::claim($task->id); + $this->assertNotNull($claim); + + // The worker disappeared after its claim committed, before recording an outcome. + Carbon::setTestNow(now()->addMinutes(12)); + if ($alreadyRepaired) { + TaskRepair::recoverExistingTask($task->fresh(), $claim->run->fresh()); + $this->assertNull(ActivityTaskClaimer::claim($task->id)[0]); + } + + $report = TaskWatchdog::runPass(); + $this->assertSame(1, $report['activity_timeouts_enforced']); + $this->assertSame([], $report['activity_timeout_failures']); + $this->assertSame(ActivityStatus::Failed, $claim->execution->fresh()->status); + $this->assertSame(TaskStatus::Cancelled, $task->fresh()->status); + $this->assertSame(1, ActivityAttempt::count()); + $event = WorkflowHistoryEvent::where('event_type', HistoryEventType::ActivityTimedOut)->sole(); + $this->assertSame('schedule_to_close', $event->payload['timeout_kind']); + + $late = ActivityOutcomeRecorder::record( + $task->id, + $claim->attemptId(), + $claim->attemptNumber(), + 'late', + null, + 1, + 0 + ); + $this->assertFalse($late['recorded']); + $this->assertSame('stale_attempt', $late['reason']); + TaskWatchdog::runPass(); + $this->assertSame( + 1, + WorkflowHistoryEvent::where('event_type', HistoryEventType::ActivityTimedOut)->count() + ); + $this->assertSame( + 0, + WorkflowHistoryEvent::where('event_type', HistoryEventType::ActivityCompleted)->count() + ); + } finally { + Carbon::setTestNow(); + } + } + + public function testPendingActivityCannotBeClaimedAfterItsOverallDeadline(): void + { + Queue::fake(); + Carbon::setTestNow('2026-09-09 00:00:00'); + try { + $task = $this->scheduleLeaseTimeoutActivity(); + Carbon::setTestNow(now()->addMinute()); + $this->assertNull(ActivityTaskClaimer::claim($task->id)[0]); + $this->assertSame(0, ActivityAttempt::count()); + $this->assertSame(1, TaskWatchdog::runPass()['activity_timeouts_enforced']); + } finally { + Carbon::setTestNow(); + } + } + public function testScheduleToStartDeadlineStoredOnScheduling(): void { $startedAt = Carbon::parse('2026-01-15 10:00:00'); @@ -1177,6 +1260,21 @@ public function testScheduleToStartRetryClearsDeadlineWhenNoTimeoutConfigured(): Carbon::setTestNow(); } + private function scheduleLeaseTimeoutActivity(): WorkflowTask + { + config([ + 'queue.default' => 'database', + ]); + WorkflowStub::make(TestActivityTimeoutCleanupWorkflow::class, 'lease-timeout')->start(); + $workflowTask = WorkflowTask::where('task_type', TaskType::Workflow)->where( + 'status', + TaskStatus::Ready + )->sole(); + $this->app->call([new RunWorkflowTask($workflowTask->id), 'handle']); + + return WorkflowTask::where('task_type', TaskType::Activity)->where('status', TaskStatus::Ready)->sole(); + } + /** * @return array{0: WorkflowRun, 1: ActivityExecution, 2: WorkflowTask} */ diff --git a/tests/Unit/V2/ActivityLeaseTimeoutTest.php b/tests/Unit/V2/ActivityLeaseTimeoutTest.php deleted file mode 100644 index b7c5af0e..00000000 --- a/tests/Unit/V2/ActivityLeaseTimeoutTest.php +++ /dev/null @@ -1,117 +0,0 @@ - - */ - public static function repairModes(): iterable - { - yield 'ordinary repair pass' => [false]; - yield 'already repaired by an older worker' => [true]; - } - - #[DataProvider('repairModes')] - public function testExpiredLeaseCannotBypassOverallTimeout(bool $alreadyRepaired): void - { - Queue::fake(); - Carbon::setTestNow('2026-09-09 00:00:00'); - try { - $task = $this->scheduleActivity(); - [$claim] = ActivityTaskClaimer::claim($task->id); - $this->assertNotNull($claim); - - // The worker disappeared after its claim committed, before recording an outcome. - Carbon::setTestNow(now()->addMinutes(12)); - if ($alreadyRepaired) { - TaskRepair::recoverExistingTask($task->fresh(), $claim->run->fresh()); - $this->assertNull(ActivityTaskClaimer::claim($task->id)[0]); - } - - $report = TaskWatchdog::runPass(); - $this->assertSame(1, $report['activity_timeouts_enforced']); - $this->assertSame([], $report['activity_timeout_failures']); - $this->assertSame(ActivityStatus::Failed, $claim->execution->fresh()->status); - $this->assertSame(TaskStatus::Cancelled, $task->fresh()->status); - $this->assertSame(1, ActivityAttempt::count()); - $event = WorkflowHistoryEvent::where('event_type', HistoryEventType::ActivityTimedOut)->sole(); - $this->assertSame('schedule_to_close', $event->payload['timeout_kind']); - - $late = ActivityOutcomeRecorder::record( - $task->id, - $claim->attemptId(), - $claim->attemptNumber(), - 'late', - null, - 1, - 0 - ); - $this->assertFalse($late['recorded']); - $this->assertSame('stale_attempt', $late['reason']); - TaskWatchdog::runPass(); - $this->assertSame( - 1, - WorkflowHistoryEvent::where('event_type', HistoryEventType::ActivityTimedOut)->count() - ); - $this->assertSame( - 0, - WorkflowHistoryEvent::where('event_type', HistoryEventType::ActivityCompleted)->count() - ); - } finally { - Carbon::setTestNow(); - } - } - - public function testPendingActivityCannotBeClaimedAfterItsOverallDeadline(): void - { - Queue::fake(); - Carbon::setTestNow('2026-09-09 00:00:00'); - try { - $task = $this->scheduleActivity(); - Carbon::setTestNow(now()->addMinute()); - $this->assertNull(ActivityTaskClaimer::claim($task->id)[0]); - $this->assertSame(0, ActivityAttempt::count()); - $this->assertSame(1, TaskWatchdog::runPass()['activity_timeouts_enforced']); - } finally { - Carbon::setTestNow(); - } - } - - private function scheduleActivity(): WorkflowTask - { - config([ - 'queue.default' => 'database', - ]); - WorkflowStub::make(TestActivityTimeoutCleanupWorkflow::class, 'lease-timeout')->start(); - $workflowTask = WorkflowTask::where('task_type', TaskType::Workflow)->where( - 'status', - TaskStatus::Ready - )->sole(); - $this->app->call([new RunWorkflowTask($workflowTask->id), 'handle']); - - return WorkflowTask::where('task_type', TaskType::Activity)->where('status', TaskStatus::Ready)->sole(); - } -} From 2367d36872b34bf67d840c44469c19b4301f86b4 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Wed, 9 Sep 2026 18:57:12 +0000 Subject: [PATCH 3/3] Prepare activity lease timeout patch release 2.0.11 --- CHANGELOG.md | 8 ++++++++ composer.json | 2 +- 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f8de2f60..20db390b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,14 @@ ## Unreleased +## 2.0.11 - 2026-09-09 + +- Preserve activity deadlines after an expired worker lease is repaired. + Pending and repaired activities cannot start another attempt after an elapsed + deadline, and the normal timeout sweep can settle the expired current attempt. + Schedule-to-close timeouts remain terminal across worker restarts without + accepting a late completion or running an extra activity effect. + ## 2.0.10 - 2026-09-09 - Recognize Laravel's native `mariadb` database driver in backend readiness and diff --git a/composer.json b/composer.json index 8f7d9121..f8e7ac6e 100644 --- a/composer.json +++ b/composer.json @@ -79,7 +79,7 @@ "dev-main": "2.0.x-dev" }, "durable-workflow": { - "product-train": "2.0.10", + "product-train": "2.0.11", "laravel-embedded-upgrade-contract": "resources/laravel-embedded-upgrade-contract.json", "laravel-dependency-security-policy": "resources/laravel-dependency-security-policy.json" },