fix(temporal): tolérer une tâche périmée sur les réponses du worker d'activité

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

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;
+            }
+        }
     }
 }

tests/unit/Bridge/Temporal/Worker/TemporalActivityWorkerTest.php

@@ -0,0 +1,156 @@
+<?php
+
+declare(strict_types=1);
+
+namespace unit\Gplanchat\Bridge\Temporal\Worker;
+
+use Gplanchat\Bridge\Temporal\Codec\JsonPlainPayload;
+use Gplanchat\Bridge\Temporal\Grpc\WorkflowServiceActivityRpc;
+use Gplanchat\Bridge\Temporal\TemporalConnection;
+use Gplanchat\Bridge\Temporal\Worker\TemporalActivityWorker;
+use Gplanchat\Durable\Event\ActivityCompleted;
+use Gplanchat\Durable\Port\ActivityHeartbeatSenderInterface;
+use Gplanchat\Durable\Port\NullWorkflowResumeDispatcher;
+use Gplanchat\Durable\RegistryActivityExecutor;
+use Gplanchat\Durable\Store\InMemoryEventStore;
+use Gplanchat\Durable\Transport\NoopActivityTransport;
+use Gplanchat\Durable\Worker\ActivityMessageProcessor;
+use Grpc\UnaryCall;
+use PHPUnit\Framework\Attributes\RequiresPhpExtension;
+use PHPUnit\Framework\TestCase;
+use Temporal\Api\Common\V1\Payloads;
+use Temporal\Api\Workflowservice\V1\PollActivityTaskQueueResponse;
+use Temporal\Api\Workflowservice\V1\RespondActivityTaskCompletedResponse;
+use Temporal\Api\Workflowservice\V1\WorkflowServiceClient;
+
+/**
+ * The activity worker must tolerate a stale task: responding for a task whose
+ * workflow/activity already closed or timed out yields gRPC NOT_FOUND (5),
+ * which is benign and must not crash the poll loop. Any other gRPC error still
+ * propagates.
+ *
+ * Strategy: seed the event store with a terminal ActivityCompleted so pollOnce()
+ * takes the "already terminal" shortcut straight to RespondActivityTaskCompleted,
+ * and control that RPC's gRPC status via a mocked WorkflowServiceClient.
+ */
+#[RequiresPhpExtension('grpc')]
+final class TemporalActivityWorkerTest extends TestCase
+{
+    private WorkflowServiceClient $grpcClient;
+    private InMemoryEventStore $eventStore;
+
+    protected function setUp(): void
+    {
+        $this->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;
+    }
+}