From 20352e9ac8ad7ac1705001cfccbc28a299a2ec89 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gr=C3=A9gory=20Planchat?= Date: Wed, 26 Aug 2026 17:35:04 +0200 Subject: [PATCH] =?UTF-8?q?fix(temporal):=20tol=C3=A9rer=20une=20t=C3=A2ch?= =?UTF-8?q?e=20p=C3=A9rim=C3=A9e=20sur=20les=20r=C3=A9ponses=20du=20worker?= =?UTF-8?q?=20d'activit=C3=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RespondActivityTask{Completed,Failed,Canceled} pour une tâche dont l'activité a expiré ou dont le workflow est déjà clos renvoie NOT_FOUND (gRPC 5). Le serveur ne suit plus la tâche : c'est bénin, et cela ne doit pas tuer la boucle de poll. Les trois réponses passent par ignoringStaleTask(), à l'image du traitement déjà présent dans WorkflowTaskProcessor::respond(). GrpcUnary reporte désormais le code gRPC dans le code de l'exception : sans lui, distinguer NOT_FOUND du reste passerait par l'analyse du message d'erreur. Reprise de 832da9a (branche backup/pr21-wip), réécrit : cette branche reposait sur une hiérarchie TemporalProtocolException qui n'existe pas sur main. Co-Authored-By: Claude Opus 5 (1M context) --- src/Bridge/Temporal/Grpc/GrpcUnary.php | 7 +- .../Worker/TemporalActivityWorker.php | 28 +++- .../Worker/TemporalActivityWorkerTest.php | 156 ++++++++++++++++++ 3 files changed, 187 insertions(+), 4 deletions(-) create mode 100644 tests/unit/Bridge/Temporal/Worker/TemporalActivityWorkerTest.php diff --git a/src/Bridge/Temporal/Grpc/GrpcUnary.php b/src/Bridge/Temporal/Grpc/GrpcUnary.php index 21c26067..17fa66da 100644 --- a/src/Bridge/Temporal/Grpc/GrpcUnary.php +++ b/src/Bridge/Temporal/Grpc/GrpcUnary.php @@ -17,7 +17,12 @@ public static function wait(UnaryCall $call): object $pair = $call->wait(); [$response, $status] = $pair; if (\Grpc\STATUS_OK !== ($status->code ?? -1)) { - throw new \RuntimeException(\sprintf('Temporal gRPC error [%s]: %s', (string) ($status->code ?? '?'), (string) ($status->details ?? ''))); + // Le code gRPC devient le code de l'exception : NOT_FOUND (5) est bénin sur les + // RespondActivityTask*, et le distinguer par le message serait de l'analyse de chaîne. + throw new \RuntimeException( + \sprintf('Temporal gRPC error [%s]: %s', (string) ($status->code ?? '?'), (string) ($status->details ?? '')), + (int) ($status->code ?? -1), + ); } if (null === $response) { throw new \RuntimeException('Temporal gRPC returned empty response.'); diff --git a/src/Bridge/Temporal/Worker/TemporalActivityWorker.php b/src/Bridge/Temporal/Worker/TemporalActivityWorker.php index f99dce93..94191ee6 100644 --- a/src/Bridge/Temporal/Worker/TemporalActivityWorker.php +++ b/src/Bridge/Temporal/Worker/TemporalActivityWorker.php @@ -36,6 +36,9 @@ */ final class TemporalActivityWorker { + /** gRPC NOT_FOUND: the task token is stale (activity timed out / workflow already closed). */ + private const GRPC_NOT_FOUND = 5; + public function __construct( private readonly WorkflowServiceActivityRpc $activityRpc, private readonly TemporalConnection $connection, @@ -160,7 +163,7 @@ private function respondCompleted(PollActivityTaskQueueResponse $poll, mixed $re $req->setIdentity($this->connection->identity . '-activity'); $req->setResult(JsonPlainPayload::singlePayloads(JsonPlainPayload::encode($result))); - $this->activityRpc->respondActivityTaskCompleted($req); + $this->ignoringStaleTask(fn() => $this->activityRpc->respondActivityTaskCompleted($req)); } private function respondFailed( @@ -187,7 +190,7 @@ private function respondFailed( $req->setIdentity($this->connection->identity . '-activity'); $req->setFailure($failure); - $this->activityRpc->respondActivityTaskFailed($req); + $this->ignoringStaleTask(fn() => $this->activityRpc->respondActivityTaskFailed($req)); } private function respondCanceled(PollActivityTaskQueueResponse $poll): void @@ -197,6 +200,25 @@ private function respondCanceled(PollActivityTaskQueueResponse $poll): void $req->setNamespace($this->connection->namespace->name()); $req->setIdentity($this->connection->identity . '-activity'); - $this->activityRpc->respondActivityTaskCanceled($req); + $this->ignoringStaleTask(fn() => $this->activityRpc->respondActivityTaskCanceled($req)); + } + + /** + * Run a RespondActivityTask* call, tolerating a stale task. + * + * Responding for a task whose workflow/activity already closed or timed out + * yields gRPC NOT_FOUND (5); the server no longer tracks the task, so this + * is benign and must not kill the poll loop. Mirrors the NOT_FOUND handling + * already present in {@see \Gplanchat\Bridge\Temporal\Worker\WorkflowTaskProcessor::respond()}. + */ + private function ignoringStaleTask(\Closure $respond): void + { + try { + $respond(); + } catch (\RuntimeException $e) { + if (self::GRPC_NOT_FOUND !== $e->getCode()) { + throw $e; + } + } } } diff --git a/tests/unit/Bridge/Temporal/Worker/TemporalActivityWorkerTest.php b/tests/unit/Bridge/Temporal/Worker/TemporalActivityWorkerTest.php new file mode 100644 index 00000000..9ddf96b9 --- /dev/null +++ b/tests/unit/Bridge/Temporal/Worker/TemporalActivityWorkerTest.php @@ -0,0 +1,156 @@ +grpcClient = $this->createMock(WorkflowServiceClient::class); + $this->eventStore = new InMemoryEventStore(); + } + + public function testStaleTaskOnRespondIsSwallowed(): void + { + $this->arrangeTerminalActivity('exec-1', 'act-1'); + $this->grpcClient->method('PollActivityTaskQueue') + ->willReturn($this->unaryCall($this->pollFor('exec-1', 'act-1'), \Grpc\STATUS_OK)); + $this->grpcClient->expects($this->once()) + ->method('RespondActivityTaskCompleted') + ->willReturn($this->unaryCall(null, \Grpc\STATUS_NOT_FOUND)); + + // No exception thrown + RespondActivityTaskCompleted called once (mock + // expectation) is the assertion. + $this->makeWorker()->pollOnce(); + } + + public function testNonStaleGrpcErrorOnRespondPropagates(): void + { + $this->arrangeTerminalActivity('exec-2', 'act-2'); + $this->grpcClient->method('PollActivityTaskQueue') + ->willReturn($this->unaryCall($this->pollFor('exec-2', 'act-2'), \Grpc\STATUS_OK)); + $this->grpcClient->method('RespondActivityTaskCompleted') + ->willReturn($this->unaryCall(null, \Grpc\STATUS_UNAVAILABLE)); + + $this->expectException(\RuntimeException::class); + $this->expectExceptionCode(\Grpc\STATUS_UNAVAILABLE); + $this->makeWorker()->pollOnce(); + } + + public function testSuccessfulRespondDoesNotThrow(): void + { + $this->arrangeTerminalActivity('exec-3', 'act-3'); + $this->grpcClient->method('PollActivityTaskQueue') + ->willReturn($this->unaryCall($this->pollFor('exec-3', 'act-3'), \Grpc\STATUS_OK)); + $this->grpcClient->method('RespondActivityTaskCompleted') + ->willReturn($this->unaryCall(new RespondActivityTaskCompletedResponse(), \Grpc\STATUS_OK)); + + $this->makeWorker()->pollOnce(); + + $this->expectNotToPerformAssertions(); + } + + public function testEmptyPollDoesNotRespond(): void + { + $empty = new PollActivityTaskQueueResponse(); + $empty->setTaskToken(''); + $this->grpcClient->method('PollActivityTaskQueue') + ->willReturn($this->unaryCall($empty, \Grpc\STATUS_OK)); + $this->grpcClient->expects($this->never())->method('RespondActivityTaskCompleted'); + + // The never() mock expectation is the assertion. + $this->makeWorker()->pollOnce(); + } + + // ------------------------------------------------------------------------- + + private function makeWorker(): TemporalActivityWorker + { + $connection = new TemporalConnection('localhost:7233', 'test-namespace'); + $processor = new ActivityMessageProcessor( + $this->eventStore, + new NoopActivityTransport(), + new RegistryActivityExecutor(), + new NullWorkflowResumeDispatcher(), + $this->createMock(ActivityHeartbeatSenderInterface::class), + ); + + return new TemporalActivityWorker( + new WorkflowServiceActivityRpc($this->grpcClient), + $connection, + $processor, + $this->eventStore, + $this->createMock(ActivityHeartbeatSenderInterface::class), + ); + } + + private function arrangeTerminalActivity(string $executionId, string $activityId): void + { + $this->eventStore->append(new ActivityCompleted($executionId, $activityId, 'result')); + } + + private function pollFor(string $executionId, string $activityId): PollActivityTaskQueueResponse + { + $payloads = new Payloads(); + $payloads->setPayloads([JsonPlainPayload::encode([ + 'executionId' => $executionId, + 'activityId' => $activityId, + 'activityName' => 'sso_delete_user', + ])]); + + $poll = new PollActivityTaskQueueResponse(); + $poll->setTaskToken('task-token'); + $poll->setInput($payloads); + + return $poll; + } + + private function unaryCall(?object $response, int $code): UnaryCall + { + $status = new \stdClass(); + $status->code = $code; + $status->details = \Grpc\STATUS_NOT_FOUND === $code + ? 'invalid activityID or activity already timed out or invoking workflow is completed' + : 'gRPC failure'; + + $call = $this->createMock(UnaryCall::class); + $call->method('wait')->willReturn([$response, $status]); + + return $call; + } +}