diff --git a/app/Http/Controllers/Api/WorkerController.php b/app/Http/Controllers/Api/WorkerController.php index 3db147c71..d6082e44f 100644 --- a/app/Http/Controllers/Api/WorkerController.php +++ b/app/Http/Controllers/Api/WorkerController.php @@ -4508,7 +4508,7 @@ public function pollUpdateValidationTasks(Request $request): JsonResponse } try { - $task = $this->updateValidationTasks->poll( + $poll = $this->updateValidationTasks->pollResult( $namespace, $worker, isset($validated['timeout_seconds']) ? (int) $validated['timeout_seconds'] : null, @@ -4532,10 +4532,7 @@ public function pollUpdateValidationTasks(Request $request): JsonResponse ); } - return WorkerProtocol::json([ - 'task' => $task, - 'poll_status' => $task === null ? 'empty' : 'leased', - ]); + return WorkerProtocol::json($poll); } public function approveUpdateValidationTask(Request $request, string $taskId): JsonResponse diff --git a/app/Support/WorkflowUpdateValidationTaskBroker.php b/app/Support/WorkflowUpdateValidationTaskBroker.php index 7cf575037..e8e0e1c29 100644 --- a/app/Support/WorkflowUpdateValidationTaskBroker.php +++ b/app/Support/WorkflowUpdateValidationTaskBroker.php @@ -139,21 +139,32 @@ public function poll( WorkerRegistration $worker, ?int $timeoutSeconds = null, ): ?array { + return $this->pollResult($namespace, $worker, $timeoutSeconds)['task']; + } + + /** + * @return array{task: array|null, poll_status: string} + */ + public function pollResult( + string $namespace, + WorkerRegistration $worker, + ?int $timeoutSeconds = null, + ): array { if (! $this->workerSupportsValidation($worker)) { - return null; + return ['task' => null, 'poll_status' => 'empty']; } - $task = $this->longPoller->until( - fn (): ?array => $this->claimAvailable($namespace, $worker), - static fn (?array $candidate): bool => is_array($candidate), + $fence = $this->workerFence($worker); + + return $this->longPoller->until( + fn (): array => $this->claimAvailableResult($namespace, $worker, $fence), + static fn (array $candidate): bool => $candidate['poll_status'] !== 'empty', $timeoutSeconds, wakeChannels: $this->signals->updateValidationTaskPollChannels($namespace, $worker->task_queue), reserveWorkerWaitSlot: true, waitSlotPool: 'query-task', waitSlotNamespace: $namespace, ); - - return is_array($task) ? $task : null; } /** @@ -162,18 +173,28 @@ public function poll( * @return array|null */ public function claimAvailable(string $namespace, WorkerRegistration $worker): ?array + { + return $this->claimAvailableResult($namespace, $worker, $this->workerFence($worker))['task']; + } + + /** + * @param array $fence + * @return array{task: array|null, poll_status: string} + */ + private function claimAvailableResult(string $namespace, WorkerRegistration $worker, array $fence): array { app(StoragePressure::class)->requireNewWork(); if (! $this->workerSupportsValidation($worker)) { - return null; + return ['task' => null, 'poll_status' => 'empty']; } - $task = $this->claim($namespace, $worker); + $claim = $this->claim($namespace, $worker, $fence); - return $task instanceof WorkflowUpdateValidationTask - ? $this->taskPayload($task) - : null; + return [ + 'task' => $claim['task'] instanceof WorkflowUpdateValidationTask ? $this->taskPayload($claim['task']) : null, + 'poll_status' => $claim['poll_status'], + ]; } /** @@ -361,9 +382,17 @@ private function findOrCreateTask( } } - private function claim(string $namespace, WorkerRegistration $worker): ?WorkflowUpdateValidationTask + /** + * @param array $fence + * @return array{task: WorkflowUpdateValidationTask|null, poll_status: string} + */ + private function claim(string $namespace, WorkerRegistration $worker, array $fence): array { - $claim = function () use ($namespace, $worker): ?WorkflowUpdateValidationTask { + $claim = function () use ($namespace, $worker, $fence): array { + if (! $this->workerFenceIsCurrent($worker, $fence)) { + return ['task' => null, 'poll_status' => 'stale_worker_registration']; + } + /** @var Collection $tasks */ $tasks = WorkflowUpdateValidationTask::query() ->where('namespace', $namespace) @@ -392,17 +421,42 @@ private function claim(string $namespace, WorkerRegistration $worker): ?Workflow 'lease_expires_at' => now()->addSeconds($this->leaseTimeoutSeconds()), ])->save(); - return $task; + return ['task' => $task, 'poll_status' => 'leased']; } - return null; + return ['task' => null, 'poll_status' => 'empty']; }; return app(WorkerProtocolMutationRetrier::class)->run( - fn (): ?WorkflowUpdateValidationTask => DB::transaction($claim), + fn (): array => DB::transaction($claim), ); } + /** @param array $fence */ + private function workerFenceIsCurrent(WorkerRegistration $worker, array $fence): bool + { + if (! WorkerPollFence::isCurrentForUpdate($fence)) { + return false; + } + + // The existing fence holds the row lock through this claim. A new + // registration can retain every routing and process identity field. + $token = WorkerRegistration::query() + ->where('namespace', $worker->namespace)->where('worker_id', $worker->worker_id)->value('registration_token'); + $incarnation = is_string($token) ? hash('sha256', $token) : null; + + return $fence['registration_incarnation'] === $incarnation; + } + + /** @return array */ + private function workerFence(WorkerRegistration $worker): array + { + return [ + ...WorkerPollFence::snapshot($worker), + 'registration_incarnation' => is_string($worker->registration_token) ? hash('sha256', $worker->registration_token) : null, + ]; + } + /** * @return array */ diff --git a/tests/Feature/UpdateValidationRegistrationFenceTest.php b/tests/Feature/UpdateValidationRegistrationFenceTest.php new file mode 100644 index 000000000..c6b6f9c85 --- /dev/null +++ b/tests/Feature/UpdateValidationRegistrationFenceTest.php @@ -0,0 +1,147 @@ +freezeTime(); + $this->createNamespace('default'); + $this->registerValidator(); + $originalToken = $this->worker()->registration_token; + $response = $this->postJson('/api/workflows', [ + 'workflow_id' => 'fenced-validation', 'workflow_type' => self::WORKFLOW_TYPE, + 'task_queue' => self::TASK_QUEUE, 'input' => ['Ada'], + ], $this->apiHeaders())->assertCreated(); + $run = WorkflowRun::query()->findOrFail($response->json('run_id')); + $task = null; + $queue = $change === 'queue' ? 'replacement-validation' : self::TASK_QUEUE; + $step = function () use ($change, $run, $queue, &$task): void { + if ($change !== 'unchanged') { + $this->registerValidator($queue, $change === 'capabilities'); + } + $codec = CodecRegistry::defaultCodec(); + $arguments = Serializer::serializeWithCodec($codec, [true]); + $task = WorkflowUpdateValidationTask::query()->create([ + 'idempotency_key' => hash('sha256', $run->id.'replacement-validation'), + 'namespace' => 'default', 'workflow_instance_id' => $run->workflow_instance_id, + 'workflow_run_id' => $run->id, 'workflow_type' => self::WORKFLOW_TYPE, + 'task_queue' => $queue, 'update_name' => 'approve', 'request_id' => 'replacement-validation', + 'input_hash' => hash('sha256', $codec."\0".$arguments), + 'payload_codec' => $codec, 'arguments' => $arguments, 'command_context' => [], + 'status' => WorkflowUpdateValidationTask::STATUS_PENDING, + ]); + app(LongPollSignalStore::class)->signalUpdateValidationTaskQueue('default', self::TASK_QUEUE); + }; + $signals = app(LongPollSignalStore::class); + $poller = new class($signals, app(LongPollWaitSlotStore::class), $step) extends LongPoller + { + public function __construct(LongPollSignalStore $signals, LongPollWaitSlotStore $slots, private ?\Closure $step) + { + parent::__construct($signals, $slots); + } + + protected function pause(int $milliseconds): void + { + if ($this->step !== null) { + $step = $this->step; + $this->step = null; + $step(); + } + parent::pause($milliseconds); + } + }; + $this->app->instance(WorkflowUpdateValidationTaskBroker::class, new WorkflowUpdateValidationTaskBroker( + $poller, $signals, app(WorkflowQueryTaskBroker::class), app(ServerWorkflowControlPlane::class), + )); + $oldPoll = $this->postJson('/api/worker/update-validation-tasks/poll', [ + 'worker_id' => 'validator', 'task_queue' => self::TASK_QUEUE, 'timeout_seconds' => 1, + ], $this->workerHeaders())->assertOk(); + $this->assertInstanceOf(WorkflowUpdateValidationTask::class, $task); + + if ($change === 'unchanged') { + $this->assertSame($originalToken, $this->worker()->registration_token); + $oldPoll->assertJsonPath('poll_status', 'leased')->assertJsonPath('task.update_validation_task_id', $task->id); + $this->assertSame(1, $task->fresh()->attempt_count); + + return; + } + + $this->assertNotSame($originalToken, $this->worker()->registration_token); + $oldPoll->assertJsonPath('task', null)->assertJsonPath('poll_status', 'stale_worker_registration'); + $this->assertSame(WorkflowUpdateValidationTask::STATUS_PENDING, $task->refresh()->status); + $this->assertSame(0, $task->attempt_count); + $this->assertNull($task->lease_owner); + $this->assertNull($task->lease_expires_at); + $this->postJson('/api/worker/update-validation-tasks/poll', [ + 'worker_id' => 'validator', 'task_queue' => $queue, 'timeout_seconds' => 0, + ], $this->workerHeaders())->assertOk() + ->assertJsonPath('poll_status', 'leased')->assertJsonPath('task.update_validation_task_id', $task->id); + $this->assertSame(1, $task->fresh()->attempt_count); + } + + public static function registrationChanges(): array + { + return [ + 'same routing and process identity' => ['same'], + 'changed queue' => ['queue'], + 'changed capabilities' => ['capabilities'], + 'unchanged registration' => ['unchanged'], + ]; + } + + private function registerValidator(string $queue = self::TASK_QUEUE, bool $queryCapability = false): void + { + $capabilities = ['workflow_tasks', 'update_validation_tasks']; + if ($queryCapability) { + $capabilities[] = 'query_tasks'; + } + $this->postJson('/api/worker/register', [ + 'worker_id' => 'validator', 'task_queue' => $queue, 'runtime' => 'php', + 'supported_workflow_types' => [self::WORKFLOW_TYPE], 'capabilities' => $capabilities, + 'capability_manifest' => $this->portableWorkerAffinityRefusalManifest(), + 'process_metrics' => ['host' => 'fixture-host', 'process_id' => 42, 'process_started_at' => '2026-10-24T00:00:00Z'], + 'workflow_command_contracts' => [self::WORKFLOW_TYPE => [ + 'queries' => [], 'query_contracts' => [], 'signals' => [], 'signal_contracts' => [], + 'updates' => ['approve'], 'update_validators' => ['approve'], + 'update_contracts' => [[ + 'name' => 'approve', + 'parameters' => [[ + 'name' => 'approved', 'position' => 0, 'required' => true, 'variadic' => false, + 'default_available' => false, 'default' => null, 'type' => 'bool', 'allows_null' => false, + ]], + ]], + ]], + ], $this->workerHeaders())->assertCreated(); + } + + private function worker(): WorkerRegistration + { + return WorkerRegistration::query()->where('namespace', 'default')->where('worker_id', 'validator')->sole(); + } +} diff --git a/tests/Feature/WorkflowUpdateValidationLockPressureTest.php b/tests/Feature/WorkflowUpdateValidationLockPressureTest.php index 8dd748073..7d2d87ae1 100644 --- a/tests/Feature/WorkflowUpdateValidationLockPressureTest.php +++ b/tests/Feature/WorkflowUpdateValidationLockPressureTest.php @@ -142,6 +142,35 @@ public function test_claim_and_reclaim_retry_real_lock_pressure(bool $reclaim): $this->assertSame(0, DB::transactionLevel()); } + public function test_registration_replaced_during_lock_retry_cannot_claim_the_task(): void + { + [, $task] = $this->validationTask(false); + $worker = WorkerRegistration::query()->where('worker_id', 'validator')->firstOrFail(); + $this->holdWriter($task); + $transactions = 0; + DB::connection()->beforeStartingTransaction(function () use (&$transactions): void { + if (++$transactions === 2) { + $this->releaseWriter(); + WorkerRegistration::query()->where('worker_id', 'validator')->update([ + 'registration_token' => 'replacement-validation-fixture', + ]); + } + }); + + $poll = app(WorkflowUpdateValidationTaskBroker::class)->pollResult('default', $worker, 0); + + $this->assertGreaterThanOrEqual(2, $transactions); + $this->assertSame(['task' => null, 'poll_status' => 'stale_worker_registration'], $poll); + $this->assertSame(WorkflowUpdateValidationTask::STATUS_PENDING, $task->fresh()->status); + $this->assertSame(0, $task->attempt_count); + $this->assertNull($task->lease_owner); + $this->assertNull($task->lease_expires_at); + $replacement = WorkerRegistration::query()->where('worker_id', 'validator')->firstOrFail(); + $delivery = app(WorkflowUpdateValidationTaskBroker::class)->poll('default', $replacement, 0); + $this->assertSame($task->id, $delivery['update_validation_task_id']); + $this->assertSame(1, $task->fresh()->attempt_count); + } + public function test_validation_timeout_retries_real_lock_pressure_and_is_terminal(): void { [$run, $task] = $this->validationTask(true);