[cross-repo from server#244] Conformance: run signals/queries against current published artifacts (#601)

durable-workflow-ops · durable-workflow-ops · commit 3929609963ed · 2026-05-18T00:43:03.000Z
diff --git a/docs/api-stability.md b/docs/api-stability.md
@@ -202,10 +202,11 @@ registration, heartbeat, workflow-task polling/history/complete/fail,
 activity-task polling/heartbeat/complete/fail, and query-task
 polling/complete/fail all use `POST /api/worker/...` with the
 worker-protocol headers. Standalone `poll*` methods return leased tasks from
-the server's `task` envelope; the client caches the returned lease fields so
-follow-up history, heartbeat, complete, and fail calls can send the required
-`lease_owner`, `workflow_task_attempt`, `activity_attempt_id`, and
-`query_task_attempt` values.
+the server's `task` envelope; query polling uses a stable `poll_request_id`
+to recover a leased task after a local HTTP timeout. The client caches the
+returned lease fields so follow-up history, heartbeat, complete, and fail
+calls can send the required `lease_owner`, `workflow_task_attempt`,
+`activity_attempt_id`, and `query_task_attempt` values.
 `WorkflowQueryTaskExecutor` is the stable PHP worker shim for the
 `query_tasks` capability. It accepts the server-routed query task envelope,
 replays the supplied history export in query mode, validates query targets
diff --git a/docs/architecture/query-and-live-debug.md b/docs/architecture/query-and-live-debug.md
@@ -180,12 +180,13 @@ lease query work through:
 - `POST /api/worker/query-tasks/{query_task_id}/complete`
 - `POST /api/worker/query-tasks/{query_task_id}/fail`
 
-The poll request names `worker_id` and `task_queue`. A successful
-poll leases at most one query task and returns `poll_status =
-'leased'`; an empty long-poll returns `poll_status = 'empty'` with
-no task. Query task long-poll timeout semantics are the same
-clamped `WorkerProtocolVersion::longPollSemantics()` used by
-workflow and activity task polling.
+The poll request names `worker_id` and `task_queue`, and may include
+`poll_request_id` so a worker can recover the same poll result after
+a local HTTP timeout. A successful poll leases at most one query task
+and returns `poll_status = 'leased'`; an empty long-poll returns
+`poll_status = 'empty'` with no task. Query task long-poll timeout
+semantics are the same clamped `WorkerProtocolVersion::longPollSemantics()`
+used by workflow and activity task polling.
 
 Each leased query task carries `query_task_id`,
 `query_task_attempt`, `lease_owner`, `workflow_id`, `run_id`,
diff --git a/src/V2/Support/WorkerProtocolVersion.php b/src/V2/Support/WorkerProtocolVersion.php
@@ -29,7 +29,7 @@ final class WorkerProtocolVersion
      * pagination semantics). Bump the minor for additive changes (new
      * optional fields, new non-terminal command types).
      */
-    public const VERSION = '1.6';
+    public const VERSION = '1.7';
 
     /**
      * Worker registration capability for server-routed workflow query
@@ -335,7 +335,7 @@ public static function queryTaskSemantics(): array
                 'poll' => [
                     'method' => 'POST',
                     'path' => '/api/worker/query-tasks/poll',
-                    'request_fields' => ['worker_id', 'task_queue'],
+                    'request_fields' => ['worker_id', 'task_queue', 'poll_request_id'],
                     'response_fields' => ['task', 'poll_status'],
                 ],
                 'complete' => [
@@ -357,6 +357,7 @@ public static function queryTaskSemantics(): array
             'poll' => [
                 'leases_on_return' => true,
                 'long_poll' => self::longPollSemantics(),
+                'poll_request_idempotency' => true,
                 'empty_response_poll_status' => 'empty',
                 'requires_registered_worker' => true,
                 'requires_worker_capability' => self::CAPABILITY_QUERY_TASKS,
diff --git a/src/V2/Worker/WorkerProtocolClient.php b/src/V2/Worker/WorkerProtocolClient.php
@@ -363,38 +363,45 @@ public function pollQueryTasks(
         ?string $queue = null,
         int $timeoutSeconds = WorkerProtocolVersion::DEFAULT_LONG_POLL_TIMEOUT,
         ?string $workerId = null,
+        ?string $pollRequestId = null,
     ): array {
         if ($this->embeddedBridgeMode) {
             return [];
         }
 
+        $pollRequestId = $this->stringValue($pollRequestId) ?? 'query-poll-'.bin2hex(random_bytes(16));
         $body = [
             'worker_id' => $this->resolveStandaloneWorkerId($workerId),
             'task_queue' => $this->resolveStandaloneTaskQueue($queue),
+            'poll_request_id' => $pollRequestId,
         ];
 
-        try {
-            $response = $this->workerPost(
-                $this->workerApiPath.'/query-tasks/poll',
-                $body,
-                $this->longPollRequestTimeoutSeconds($timeoutSeconds),
-            );
-        } catch (ConnectionException $exception) {
-            if ($this->isHttpTimeout($exception)) {
+        for ($attempt = 0; $attempt < 2; $attempt++) {
+            try {
+                $response = $this->workerPost(
+                    $this->workerApiPath.'/query-tasks/poll',
+                    $body,
+                    $this->longPollRequestTimeoutSeconds($timeoutSeconds),
+                );
+            } catch (ConnectionException $exception) {
+                if ($this->isHttpTimeout($exception)) {
+                    continue;
+                }
+
+                throw $exception;
+            }
+
+            $task = $response['task'] ?? null;
+            if (! is_array($task)) {
                 return [];
             }
 
-            throw $exception;
-        }
+            $this->rememberQueryTaskLease($task);
 
-        $task = $response['task'] ?? null;
-        if (! is_array($task)) {
-            return [];
+            return [$task];
         }
 
-        $this->rememberQueryTaskLease($task);
-
-        return [$task];
+        return [];
     }
 
     /**
diff --git a/tests/Unit/V2/WorkerProtocolClientTest.php b/tests/Unit/V2/WorkerProtocolClientTest.php
@@ -97,7 +97,11 @@ public function testStandaloneQueryPollAndCompleteUseCachedLease(): void
         });
 
         $client = new WorkerProtocolClient($http, 'http://server:8080', 'test-token', 'default');
-        $tasks = $client->pollQueryTasks(queue: 'polyglot', workerId: 'php-worker');
+        $tasks = $client->pollQueryTasks(
+            queue: 'polyglot',
+            workerId: 'php-worker',
+            pollRequestId: 'query-poll-request-1',
+        );
         $complete = $client->completeQueryTask('query-task-1', 'waiting', [
             'codec' => 'avro',
             'blob' => 'encoded-result',
@@ -110,7 +114,11 @@ public function testStandaloneQueryPollAndCompleteUseCachedLease(): void
             [
                 'method' => 'POST',
                 'url' => 'http://server:8080/api/worker/query-tasks/poll',
-                'body' => ['worker_id' => 'php-worker', 'task_queue' => 'polyglot'],
+                'body' => [
+                    'worker_id' => 'php-worker',
+                    'task_queue' => 'polyglot',
+                    'poll_request_id' => 'query-poll-request-1',
+                ],
             ],
             [
                 'method' => 'POST',
@@ -154,7 +162,11 @@ public function testStandaloneQueryFailureUsesCachedLease(): void
         });
 
         $client = new WorkerProtocolClient($http, 'http://server:8080', 'test-token', 'default');
-        $client->pollQueryTasks(queue: 'polyglot', workerId: 'php-worker');
+        $client->pollQueryTasks(
+            queue: 'polyglot',
+            workerId: 'php-worker',
+            pollRequestId: 'query-poll-request-2',
+        );
         $failure = $client->failQueryTask(
             'query-task-1',
             'No query handler.',
@@ -170,7 +182,11 @@ public function testStandaloneQueryFailureUsesCachedLease(): void
             [
                 'method' => 'POST',
                 'url' => 'http://server:8080/api/worker/query-tasks/poll',
-                'body' => ['worker_id' => 'php-worker', 'task_queue' => 'polyglot'],
+                'body' => [
+                    'worker_id' => 'php-worker',
+                    'task_queue' => 'polyglot',
+                    'poll_request_id' => 'query-poll-request-2',
+                ],
             ],
             [
                 'method' => 'POST',
@@ -191,6 +207,40 @@ public function testStandaloneQueryFailureUsesCachedLease(): void
         ], $requests);
     }
 
+    public function testStandaloneQueryPollRetriesTimeoutWithSamePollRequestId(): void
+    {
+        $http = new HttpFactory();
+        $requests = [];
+
+        $http->fake(function (Request $request) use ($http, &$requests) {
+            $requests[] = $request->data();
+
+            if (count($requests) === 1) {
+                throw new ConnectionException('cURL error 28: Operation timed out');
+            }
+
+            return $http->response([
+                'task' => [
+                    'query_task_id' => 'query-task-retry',
+                    'query_task_attempt' => 1,
+                    'lease_owner' => 'php-worker',
+                ],
+            ]);
+        });
+
+        $client = new WorkerProtocolClient($http, 'http://server:8080', 'test-token', 'default');
+        $tasks = $client->pollQueryTasks(queue: 'polyglot', workerId: 'php-worker');
+
+        $this->assertCount(1, $tasks);
+        $this->assertSame('query-task-retry', $tasks[0]['query_task_id']);
+        $this->assertCount(2, $requests);
+        $this->assertSame('php-worker', $requests[0]['worker_id']);
+        $this->assertSame('polyglot', $requests[0]['task_queue']);
+        $this->assertIsString($requests[0]['poll_request_id']);
+        $this->assertNotSame('', $requests[0]['poll_request_id']);
+        $this->assertSame($requests[0]['poll_request_id'], $requests[1]['poll_request_id']);
+    }
+
     public function testPollWorkflowTasksUsesStandaloneWorkerApiByDefault(): void
     {
         $http = new HttpFactory();
diff --git a/tests/Unit/V2/WorkerProtocolVersionTest.php b/tests/Unit/V2/WorkerProtocolVersionTest.php
@@ -17,7 +17,7 @@ public function testVersionIsNonEmptyString(): void
 
     public function testVersionTracksQueryTaskWorkerProtocolShape(): void
     {
-        $this->assertSame('1.6', WorkerProtocolVersion::VERSION);
+        $this->assertSame('1.7', WorkerProtocolVersion::VERSION);
     }
 
     public function testWorkflowTaskVerbsIncludesAllBridgeMethods(): void
@@ -129,6 +129,7 @@ public function testDescribeIncludesQueryTaskSemantics(): void
         $this->assertSame(WorkerProtocolVersion::queryTaskVerbs(), $queryTasks['verbs']);
         $this->assertSame('/api/worker/query-tasks', $queryTasks['path_prefix']);
         $this->assertSame('/api/worker/query-tasks/poll', $queryTasks['endpoints']['poll']['path']);
+        $this->assertContains('poll_request_id', $queryTasks['endpoints']['poll']['request_fields']);
         $this->assertSame(
             '/api/worker/query-tasks/{query_task_id}/complete',
             $queryTasks['endpoints']['complete']['path'],
@@ -139,6 +140,7 @@ public function testDescribeIncludesQueryTaskSemantics(): void
         );
         $this->assertTrue($queryTasks['poll']['leases_on_return']);
         $this->assertSame(WorkerProtocolVersion::longPollSemantics(), $queryTasks['poll']['long_poll']);
+        $this->assertTrue($queryTasks['poll']['poll_request_idempotency']);
         $this->assertSame('empty', $queryTasks['poll']['empty_response_poll_status']);
         $this->assertTrue($queryTasks['poll']['requires_registered_worker']);
         $this->assertSame(
