Skip to content

Commit 3e3add8

Browse files
rmcdanielrmcdaniel
authored
Merge pull request #886 from durable-workflow/fix/878-watchdog-claimed-attempt
Verify watchdog lease recovery from a genuine original activity claim
2 parents 3941397 + 94589cf commit 3e3add8

1 file changed

Lines changed: 87 additions & 33 deletions

File tree

‎tests/Feature/V2/V2WorkflowTest.php‎

Lines changed: 87 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -96,6 +96,7 @@
9696
use Workflow\V2\Support\ActivityCall;
9797
use Workflow\V2\Support\ActivityCancellation;
9898
use Workflow\V2\Support\ActivityLease;
99+
use Workflow\V2\Support\ActivityTaskClaimer;
99100
use Workflow\V2\Support\DefaultHistoryProjectionRole;
100101
use Workflow\V2\Support\EmbeddedV2HistoryImport;
101102
use Workflow\V2\Support\ExternalPayloads;
@@ -6901,6 +6902,85 @@ public function testTaskWatchdogLeavesRowOnlyRunningActivityFallbackDiagnosticOn
69016902
}
69026903

69036904
public function testTaskWatchdogReclaimsExpiredActivityTaskLease(): void
6905+
{
6906+
Queue::fake();
6907+
Carbon::setTestNow('2026-10-21 08:00:00.123456');
6908+
6909+
try {
6910+
$workflow = WorkflowStub::make(TestRetryWorkflow::class, 'repair-expired-original-activity');
6911+
$workflow->start('Taylor');
6912+
$runId = $workflow->runId();
6913+
$this->assertNotNull($runId);
6914+
$this->runReadyTaskForRun($runId, TaskType::Workflow);
6915+
6916+
$activityTasks = WorkflowTask::query()->where('workflow_run_id', $runId)
6917+
->where('task_type', TaskType::Activity->value)->get();
6918+
$this->assertCount(1, $activityTasks, WorkflowTask::query()->where('workflow_run_id', $runId)
6919+
->get(['status', 'payload', 'last_error'])->toJson());
6920+
/** @var WorkflowTask $task */
6921+
$task = $activityTasks->sole();
6922+
[$claim] = ActivityTaskClaimer::claim($task->id, 'worker-1');
6923+
$this->assertNotNull($claim);
6924+
$originalExpiry = $claim->task->lease_expires_at;
6925+
$this->assertNotNull($originalExpiry);
6926+
Carbon::setTestNow($originalExpiry->copy()->addSecond());
6927+
6928+
// Exclude the initial dispatch so only the watchdog's retry is observed.
6929+
Queue::fake();
6930+
$this->wakeTaskWatchdog();
6931+
6932+
/** @var WorkflowTask $retryTask */
6933+
$retryTask = WorkflowTask::query()->where('workflow_run_id', $runId)
6934+
->where('task_type', TaskType::Activity->value)
6935+
->where('status', TaskStatus::Ready->value)->sole();
6936+
$this->assertNotSame($task->id, $retryTask->id);
6937+
$this->assertSame(TaskStatus::Cancelled, $task->fresh()->status);
6938+
$this->assertSame(1, $task->fresh()->attempt_count);
6939+
$this->assertSame(0, $task->fresh()->repair_count);
6940+
$this->assertSame('expired', $claim->attempt->fresh()->status->value);
6941+
$this->assertNull($claim->attempt->fresh()->lease_expires_at);
6942+
$this->assertNull($task->fresh()->lease_expires_at);
6943+
$this->assertSame(now()->toJSON(), $claim->attempt->fresh()->closed_at->toJSON());
6944+
$this->assertSame('worker-1', $claim->attempt->fresh()->lease_owner);
6945+
$this->assertSame(ActivityStatus::Pending, $claim->execution->fresh()->status);
6946+
$this->assertSame(1, $claim->execution->fresh()->attempt_count);
6947+
$this->assertSame(
6948+
1,
6949+
ActivityAttempt::query()->where('activity_execution_id', $claim->execution->id)->count()
6950+
);
6951+
$this->assertSame(1, $retryTask->attempt_count);
6952+
$this->assertSame($task->id, $retryTask->payload['retry_of_task_id']);
6953+
$this->assertSame($claim->attemptId(), $retryTask->payload['retry_after_attempt_id']);
6954+
$this->assertSame('lease_expired', $retryTask->payload['retry_reason']);
6955+
$this->assertSame(now()->addSeconds(5)->toJSON(), $retryTask->available_at->toJSON());
6956+
6957+
$event = WorkflowHistoryEvent::query()->where('workflow_run_id', $runId)
6958+
->where('event_type', HistoryEventType::ActivityRetryScheduled->value)->sole();
6959+
$this->assertSame($claim->attemptId(), $event->payload['activity_attempt_id']);
6960+
$this->assertSame($retryTask->id, $event->payload['retry_task_id']);
6961+
$this->assertSame('lease_expired', $event->payload['failure_reason']);
6962+
$this->assertSame($originalExpiry->toJSON(), $event->payload['lease_expires_at']);
6963+
Queue::assertPushed(
6964+
RunActivityTask::class,
6965+
static fn (RunActivityTask $job): bool => $job->taskId === $retryTask->id
6966+
);
6967+
Queue::assertNotPushed(
6968+
RunActivityTask::class,
6969+
static fn (RunActivityTask $job): bool => $job->taskId === $task->id
6970+
);
6971+
$this->assertSame($retryTask->id, WorkflowRunSummary::query()->findOrFail($runId)->next_task_id);
6972+
6973+
$this->wakeTaskWatchdog();
6974+
$this->assertSame(1, WorkflowHistoryEvent::query()->where('workflow_run_id', $runId)
6975+
->where('event_type', HistoryEventType::ActivityRetryScheduled->value)->count());
6976+
$this->assertSame(2, WorkflowTask::query()->where('workflow_run_id', $runId)
6977+
->where('task_type', TaskType::Activity->value)->count());
6978+
} finally {
6979+
Carbon::setTestNow();
6980+
}
6981+
}
6982+
6983+
public function testTaskWatchdogDoesNotInventAMissingOriginalActivityAttempt(): void
69046984
{
69056985
Queue::fake();
69066986

@@ -6977,41 +7057,15 @@ public function testTaskWatchdogReclaimsExpiredActivityTaskLease(): void
69777057
'queue' => 'activities',
69787058
]);
69797059

6980-
$this->wakeTaskWatchdog();
7060+
$before = [$task->fresh()->getAttributes(), $execution->fresh()->getAttributes()];
7061+
$historyCount = WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)->count();
69817062

6982-
$task->refresh();
6983-
$execution->refresh();
6984-
6985-
/** @var ActivityAttempt $attempt */
6986-
$attempt = ActivityAttempt::query()
6987-
->where('activity_execution_id', $execution->id)
6988-
->firstOrFail();
6989-
6990-
$this->assertSame(TaskStatus::Ready, $task->status);
6991-
$this->assertNull($task->leased_at);
6992-
$this->assertNull($task->lease_owner);
6993-
$this->assertNull($task->lease_expires_at);
6994-
$this->assertSame(1, $task->attempt_count);
6995-
$this->assertSame(1, $task->repair_count);
6996-
$this->assertNotNull($execution->current_attempt_id);
6997-
$this->assertSame(1, $execution->attempt_count);
6998-
$this->assertSame($execution->current_attempt_id, $attempt->id);
6999-
$this->assertSame(1, $attempt->attempt_number);
7000-
$this->assertSame('expired', $attempt->status->value);
7001-
$this->assertSame($task->id, $attempt->workflow_task_id);
7002-
$this->assertSame('worker-1', $attempt->lease_owner);
7003-
$this->assertNotNull($attempt->closed_at);
7004-
7005-
Queue::assertPushed(
7006-
RunActivityTask::class,
7007-
static fn (RunActivityTask $job): bool => $job->taskId === $task->id
7008-
);
7009-
7010-
$summary = WorkflowRunSummary::query()->findOrFail($run->id);
7063+
$this->wakeTaskWatchdog();
70117064

7012-
$this->assertSame('activity', $summary->wait_kind);
7013-
$this->assertSame('activity_task_ready', $summary->liveness_state);
7014-
$this->assertSame($task->id, $summary->next_task_id);
7065+
$this->assertSame($before, [$task->fresh()->getAttributes(), $execution->fresh()->getAttributes()]);
7066+
$this->assertSame(0, ActivityAttempt::query()->where('activity_execution_id', $execution->id)->count());
7067+
$this->assertSame($historyCount, WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)->count());
7068+
Queue::assertNotPushed(RunActivityTask::class);
70157069
}
70167070

70177071
public function testLateActivityOutcomeIsIgnoredAfterNewerAttemptTakesOwnership(): void

0 commit comments

Comments
 (0)