Skip to content

Commit 697f4a9

Browse files
durable-workflow-opsdurable-workflow-ops
committed
Polyglot PHP signal/query path times out after worker answers state query (#335)
1 parent 30ed20a commit 697f4a9

3 files changed

Lines changed: 146 additions & 18 deletions

File tree

‎app/Polyglot/ServerClient.php‎

Lines changed: 64 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,13 @@ final class ServerClient
3030

3131
private readonly string $protocolVersion;
3232

33+
private readonly int $processStartedAt;
34+
35+
/**
36+
* @var array<string, string>
37+
*/
38+
private array $pollRequestIds = [];
39+
3340
public function __construct(
3441
private readonly HttpFactory $http,
3542
private readonly string $baseUrl,
@@ -38,6 +45,7 @@ public function __construct(
3845
?string $protocolVersion = null,
3946
) {
4047
$this->protocolVersion = $protocolVersion ?? WorkerProtocolVersion::VERSION;
48+
$this->processStartedAt = time();
4149
}
4250

4351
/**
@@ -63,6 +71,7 @@ public function registerWorker(
6371
'supported_workflow_types' => $supportedWorkflowTypes,
6472
'supported_activity_types' => $supportedActivityTypes,
6573
'capabilities' => $capabilities,
74+
'process_metrics' => $this->processMetrics(),
6675
]);
6776
}
6877

@@ -74,6 +83,7 @@ public function heartbeatWorker(string $workerId): ?array
7483
try {
7584
return $this->workerPost('/api/worker/heartbeat', [
7685
'worker_id' => $workerId,
86+
'process_metrics' => $this->processMetrics(),
7787
], requestTimeoutSeconds: self::HEARTBEAT_TIMEOUT_SECONDS);
7888
} catch (ConnectionException $exception) {
7989
if ($this->isHttpTimeout($exception)) {
@@ -199,12 +209,14 @@ public function completeQueryTask(
199209
int $queryTaskAttempt,
200210
mixed $result,
201211
): void {
202-
$this->workerPost("/api/worker/query-tasks/{$queryTaskId}/complete", [
212+
$response = $this->workerPost("/api/worker/query-tasks/{$queryTaskId}/complete", [
203213
'lease_owner' => $leaseOwner,
204214
'query_task_attempt' => $queryTaskAttempt,
205215
'result' => $result,
206216
'result_envelope' => Avro::envelope($result),
207217
]);
218+
219+
$this->ensureWorkerOutcome($response, 'completed', 'complete query task', $queryTaskId);
208220
}
209221

210222
public function failQueryTask(
@@ -224,11 +236,13 @@ public function failQueryTask(
224236
$failure['type'] = $failureType;
225237
}
226238

227-
$this->workerPost("/api/worker/query-tasks/{$queryTaskId}/fail", [
239+
$response = $this->workerPost("/api/worker/query-tasks/{$queryTaskId}/fail", [
228240
'lease_owner' => $leaseOwner,
229241
'query_task_attempt' => $queryTaskAttempt,
230242
'failure' => $failure,
231243
]);
244+
245+
$this->ensureWorkerOutcome($response, 'failed', 'fail query task', $queryTaskId);
232246
}
233247

234248
/**
@@ -237,20 +251,30 @@ public function failQueryTask(
237251
private function pollTask(string $path, string $workerId, string $taskQueue, int $timeoutSeconds): ?array
238252
{
239253
$pollTimeoutSeconds = WorkerProtocolVersion::clampLongPollTimeout($timeoutSeconds);
254+
$pollRequestKey = $this->pollRequestKey($path, $workerId, $taskQueue);
255+
$pollRequestId = $this->pollRequestIds[$pollRequestKey]
256+
?? 'polyglot-poll-'.bin2hex(random_bytes(16));
257+
$this->pollRequestIds[$pollRequestKey] = $pollRequestId;
258+
240259
try {
241260
$response = $this->workerPost($path, [
242261
'worker_id' => $workerId,
243262
'task_queue' => $taskQueue,
263+
'poll_request_id' => $pollRequestId,
244264
'timeout_seconds' => $pollTimeoutSeconds,
245265
], requestTimeoutSeconds: $this->requestTimeoutForPoll($pollTimeoutSeconds));
246266
} catch (ConnectionException $exception) {
247267
if ($this->isHttpTimeout($exception)) {
248268
return null;
249269
}
250270

271+
unset($this->pollRequestIds[$pollRequestKey]);
272+
251273
throw $exception;
252274
}
253275

276+
unset($this->pollRequestIds[$pollRequestKey]);
277+
254278
if ($response === null) {
255279
return null;
256280
}
@@ -273,6 +297,44 @@ private function requestTimeoutForPoll(int $pollTimeoutSeconds): int
273297
return $pollTimeoutSeconds + 15;
274298
}
275299

300+
/**
301+
* @return array<string, int|string>
302+
*/
303+
private function processMetrics(): array
304+
{
305+
return [
306+
'process_id' => getmypid() ?: 0,
307+
'host' => gethostname() ?: 'unknown',
308+
'process_started_at' => (string) $this->processStartedAt,
309+
];
310+
}
311+
312+
private function pollRequestKey(string $path, string $workerId, string $taskQueue): string
313+
{
314+
return implode('|', [$path, $workerId, $taskQueue]);
315+
}
316+
317+
/**
318+
* @param array<string, mixed>|null $response
319+
*/
320+
private function ensureWorkerOutcome(
321+
?array $response,
322+
string $expectedOutcome,
323+
string $operation,
324+
string $taskId,
325+
): void {
326+
if (($response['outcome'] ?? null) === $expectedOutcome) {
327+
return;
328+
}
329+
330+
throw new RuntimeException(sprintf(
331+
'Durable Workflow server rejected %s for %s: %s',
332+
$operation,
333+
$taskId,
334+
json_encode($response, JSON_UNESCAPED_SLASHES) ?: 'empty response',
335+
));
336+
}
337+
276338
/**
277339
* @param array<string, mixed> $body
278340
* @return array<string, mixed>|null

‎tests/Unit/PolyglotServerClientTest.php‎

Lines changed: 59 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -34,11 +34,10 @@ public function test_poll_workflow_task_sends_clamped_protocol_timeout(): void
3434
$task = $client->pollWorkflowTask('php-worker', 'polyglot-php-to-python', 120);
3535

3636
$this->assertSame('task-1', $task['task_id'] ?? null);
37-
$this->assertSame([
38-
'worker_id' => 'php-worker',
39-
'task_queue' => 'polyglot-php-to-python',
40-
'timeout_seconds' => 60,
41-
], $requestBody);
37+
$this->assertSame('php-worker', $requestBody['worker_id'] ?? null);
38+
$this->assertSame('polyglot-php-to-python', $requestBody['task_queue'] ?? null);
39+
$this->assertMatchesRegularExpression('/^polyglot-poll-[0-9a-f]{32}$/', $requestBody['poll_request_id'] ?? '');
40+
$this->assertSame(60, $requestBody['timeout_seconds'] ?? null);
4241
}
4342

4443
public function test_poll_query_task_can_send_immediate_protocol_probe(): void
@@ -58,11 +57,35 @@ public function test_poll_query_task_can_send_immediate_protocol_probe(): void
5857

5958
$this->assertNull($client->pollQueryTask('php-worker', 'polyglot-php-to-python', 0));
6059
$this->assertSame('http://server:8080/api/worker/query-tasks/poll', $requestPath);
61-
$this->assertSame([
62-
'worker_id' => 'php-worker',
63-
'task_queue' => 'polyglot-php-to-python',
64-
'timeout_seconds' => 0,
65-
], $requestBody);
60+
$this->assertSame('php-worker', $requestBody['worker_id'] ?? null);
61+
$this->assertSame('polyglot-php-to-python', $requestBody['task_queue'] ?? null);
62+
$this->assertMatchesRegularExpression('/^polyglot-poll-[0-9a-f]{32}$/', $requestBody['poll_request_id'] ?? '');
63+
$this->assertSame(0, $requestBody['timeout_seconds'] ?? null);
64+
}
65+
66+
public function test_query_poll_reuses_poll_request_id_after_http_timeout(): void
67+
{
68+
$http = new HttpFactory();
69+
$attempt = 0;
70+
$requestBodies = [];
71+
72+
$http->fake(function (Request $request) use ($http, &$attempt, &$requestBodies) {
73+
$attempt++;
74+
$requestBodies[] = $request->data();
75+
76+
if ($attempt === 1) {
77+
throw new ConnectionException('cURL error 28: Operation timed out');
78+
}
79+
80+
return $http->response(['task' => null]);
81+
});
82+
83+
$client = new ServerClient($http, 'http://server:8080', 'test-token', 'default');
84+
85+
$this->assertNull($client->pollQueryTask('php-query-worker', 'polyglot-php-to-python', 5));
86+
$this->assertNull($client->pollQueryTask('php-query-worker', 'polyglot-php-to-python', 5));
87+
$this->assertCount(2, $requestBodies);
88+
$this->assertSame($requestBodies[0]['poll_request_id'] ?? null, $requestBodies[1]['poll_request_id'] ?? null);
6689
}
6790

6891
public function test_poll_request_timeout_tracks_the_requested_poll_window(): void
@@ -127,11 +150,10 @@ public function test_poll_activity_task_uses_activity_endpoint_and_clamped_timeo
127150

128151
$this->assertSame('http://server:8080/api/worker/activity-tasks/poll', $requestPath);
129152
$this->assertSame('activity-task-1', $task['task_id'] ?? null);
130-
$this->assertSame([
131-
'worker_id' => 'php-activity-worker',
132-
'task_queue' => 'polyglot-python-to-php',
133-
'timeout_seconds' => 60,
134-
], $requestBody);
153+
$this->assertSame('php-activity-worker', $requestBody['worker_id'] ?? null);
154+
$this->assertSame('polyglot-python-to-php', $requestBody['task_queue'] ?? null);
155+
$this->assertMatchesRegularExpression('/^polyglot-poll-[0-9a-f]{32}$/', $requestBody['poll_request_id'] ?? '');
156+
$this->assertSame(60, $requestBody['timeout_seconds'] ?? null);
135157
}
136158

137159
public function test_complete_activity_task_sends_avro_result_envelope(): void
@@ -194,4 +216,26 @@ public function test_fail_activity_task_sends_failure_payload(): void
194216
],
195217
], $requestBody);
196218
}
219+
220+
public function test_complete_query_task_requires_completed_worker_outcome(): void
221+
{
222+
$http = new HttpFactory();
223+
224+
$http->fake(fn (Request $request) => $http->response([
225+
'outcome' => 'rejected',
226+
'reason' => 'query_task_timed_out',
227+
]));
228+
229+
$client = new ServerClient($http, 'http://server:8080', 'test-token', 'default');
230+
231+
$this->expectException(RuntimeException::class);
232+
$this->expectExceptionMessage('rejected complete query task');
233+
234+
$client->completeQueryTask(
235+
'query-task-1',
236+
'php-query-worker',
237+
1,
238+
['stage' => 'waiting'],
239+
);
240+
}
197241
}

‎tests/Unit/PolyglotWorkerReplayTest.php‎

Lines changed: 23 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -87,6 +87,9 @@ public function test_server_client_can_heartbeat_registered_worker(): void
8787
$requests[0]['url'],
8888
);
8989
$this->assertSame('php-query-worker', $requests[0]['body']['worker_id'] ?? null);
90+
$this->assertSame(getmypid(), $requests[0]['body']['process_metrics']['process_id'] ?? null);
91+
$this->assertIsString($requests[0]['body']['process_metrics']['host'] ?? null);
92+
$this->assertIsString($requests[0]['body']['process_metrics']['process_started_at'] ?? null);
9093
}
9194

9295
public function test_workflow_worker_polls_query_tasks_before_heartbeat_after_registration(): void
@@ -128,11 +131,16 @@ public function test_workflow_worker_polls_query_tasks_before_heartbeat_after_re
128131
[WorkerProtocolVersion::CAPABILITY_QUERY_TASKS],
129132
$requests[0]['body']['capabilities'] ?? null,
130133
);
134+
$this->assertSame(getmypid(), $requests[0]['body']['process_metrics']['process_id'] ?? null);
131135
$this->assertSame(
132136
'http://server:8080/api/worker/query-tasks/poll',
133137
$requests[1]['url'] ?? null,
134138
);
135139
$this->assertSame(0, $requests[1]['body']['timeout_seconds'] ?? null);
140+
$this->assertMatchesRegularExpression(
141+
'/^polyglot-poll-[0-9a-f]{32}$/',
142+
$requests[1]['body']['poll_request_id'] ?? '',
143+
);
136144
$this->assertSame(
137145
'http://server:8080/api/worker/heartbeat',
138146
$requests[2]['url'] ?? null,
@@ -178,6 +186,10 @@ public function test_query_capable_workflow_worker_slices_workflow_polls_between
178186
$this->assertNotNull($workflowPollIndex);
179187
$this->assertLessThan($workflowPollIndex, $queryPollIndex);
180188
$this->assertSame(0, $requests[$queryPollIndex]['body']['timeout_seconds'] ?? null);
189+
$this->assertMatchesRegularExpression(
190+
'/^polyglot-poll-[0-9a-f]{32}$/',
191+
$requests[$queryPollIndex]['body']['poll_request_id'] ?? '',
192+
);
181193
$this->assertSame(1, $requests[$workflowPollIndex]['body']['timeout_seconds'] ?? null);
182194
}
183195

@@ -226,6 +238,11 @@ public function test_workflow_worker_keeps_polling_queries_after_heartbeat_timeo
226238
return $http->response(['task' => null]);
227239
}
228240

241+
if (str_contains($url, '/api/worker/query-tasks/')
242+
&& str_ends_with($url, '/complete')) {
243+
return $http->response(['outcome' => 'completed']);
244+
}
245+
229246
return $http->response(['task' => null, 'ok' => true]);
230247
});
231248

@@ -318,6 +335,11 @@ public function test_query_worker_answers_repeated_php_signal_state_queries_with
318335
return $http->response(['task' => null]);
319336
}
320337

338+
if (str_contains($url, '/api/worker/query-tasks/')
339+
&& str_ends_with($url, '/complete')) {
340+
return $http->response(['outcome' => 'completed']);
341+
}
342+
321343
return $http->response(['ok' => true]);
322344
});
323345

@@ -957,7 +979,7 @@ private function processQueryTask(array $task): array
957979
'body' => $request->data(),
958980
];
959981

960-
return $http->response(['ok' => true]);
982+
return $http->response(['outcome' => 'completed']);
961983
});
962984

963985
$client = new ServerClient($http, 'http://server:8080', 'test-token', 'default');

0 commit comments

Comments
 (0)