Fence dedicated update-validation polls to worker registrations

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

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<string, mixed>|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<string, mixed>|null
      */
     public function claimAvailable(string $namespace, WorkerRegistration $worker): ?array
+    {
+        return $this->claimAvailableResult($namespace, $worker, $this->workerFence($worker))['task'];
+    }
+
+    /**
+     * @param  array<string, mixed>  $fence
+     * @return array{task: array<string, mixed>|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<string, mixed>  $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<int, WorkflowUpdateValidationTask> $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<string, mixed> $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<string, mixed> */
+    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<string, mixed>
      */

tests/Feature/UpdateValidationRegistrationFenceTest.php

@@ -0,0 +1,147 @@
+<?php
+
+namespace Tests\Feature;
+
+use App\Models\WorkerRegistration;
+use App\Models\WorkflowUpdateValidationTask;
+use App\Support\LongPoller;
+use App\Support\LongPollSignalStore;
+use App\Support\LongPollWaitSlotStore;
+use App\Support\ServerWorkflowControlPlane;
+use App\Support\WorkflowQueryTaskBroker;
+use App\Support\WorkflowUpdateValidationTaskBroker;
+use Illuminate\Foundation\Testing\RefreshDatabase;
+use PHPUnit\Framework\Attributes\DataProvider;
+use Tests\Feature\Concerns\ServerTestHelpers;
+use Tests\TestCase;
+use Workflow\Serializers\CodecRegistry;
+use Workflow\Serializers\Serializer;
+use Workflow\V2\Models\WorkflowRun;
+
+class UpdateValidationRegistrationFenceTest extends TestCase
+{
+    use RefreshDatabase;
+    use ServerTestHelpers;
+
+    private const WORKFLOW_TYPE = 'tests.fenced-validation';
+
+    private const TASK_QUEUE = 'fenced-validation';
+
+    #[DataProvider('registrationChanges')]
+    public function test_waiting_poll_fences_a_replacement_but_keeps_current_registration(string $change): void
+    {
+        $this->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();
+    }
+}

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);