Make internal transport Client forkable and destroyable

Author: roxblnfk

src/Internal/Transport/Client.php

@@ -116,6 +116,21 @@ public function reject(CommandInterface $command, \Throwable $reason): void
         $this->fetch($command->getID())->reject($reason);
     }
 
+    public function fork(): ClientInterface
+    {
+        return new DetachedClient($this, function (array $ids): void {
+            foreach ($ids as $id) {
+                unset($this->requests[$id]);
+            }
+        });
+    }
+
+    public function destroy(): void
+    {
+        $this->requests = [];
+        unset($this->queue);
+    }
+
     private function fetch(int $id): Deferred
     {
         $request = $this->get($id);

src/Internal/Transport/ClientInterface.php

@@ -12,11 +12,13 @@
 namespace Temporal\Internal\Transport;
 
 use React\Promise\PromiseInterface;
+use Temporal\Internal\Declaration\Destroyable;
 use Temporal\Worker\Transport\Command\CommandInterface;
 use Temporal\Worker\Transport\Command\RequestInterface;
+use Temporal\Worker\Transport\Command\ServerResponseInterface;
 use Temporal\Workflow\WorkflowContextInterface;
 
-interface ClientInterface
+interface ClientInterface extends Destroyable
 {
     /**
      * Send a request and return a promise.
@@ -39,4 +41,14 @@ public function cancel(CommandInterface $command): void;
      * Reject pending promise.
      */
     public function reject(CommandInterface $command, \Throwable $reason): void;
+
+    /**
+     * Dispatch a response to the request.
+     */
+    public function dispatch(ServerResponseInterface $response): void;
+
+    /**
+     * Create a new client that can work with parent's request queue.
+     */
+    public function fork(): ClientInterface;
 }

src/Internal/Transport/DetachedClient.php

@@ -0,0 +1,87 @@
+<?php
+
+/**
+ * This file is part of Temporal package.
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+declare(strict_types=1);
+
+namespace Temporal\Internal\Transport;
+
+use React\Promise\PromiseInterface;
+use Temporal\Internal\Declaration\Destroyable;
+use Temporal\Worker\Transport\Command\CommandInterface;
+use Temporal\Worker\Transport\Command\RequestInterface;
+use Temporal\Worker\Transport\Command\ServerResponseInterface;
+use Temporal\Workflow\WorkflowContextInterface;
+
+/**
+ * @internal Client is an internal library class, please do not use it in your code.
+ * @psalm-internal Temporal\Internal\Transport
+ */
+final class DetachedClient implements ClientInterface, Destroyable
+{
+    /** @var list<int> */
+    private array $requests = [];
+
+    /**
+     * @param \Closure(list<int>): void $cleanup Handler that removes requests from the parent using their IDs.
+     */
+    public function __construct(
+        private ClientInterface $parent,
+        private \Closure $cleanup,
+    ) {}
+
+    #[\Override]
+    public function request(RequestInterface $request, ?WorkflowContextInterface $context = null): PromiseInterface
+    {
+        $this->requests[] = $request->getID();
+        return $this->parent->request($request, $context);
+    }
+
+    #[\Override]
+    public function send(CommandInterface $request): void
+    {
+        $this->parent->send($request);
+    }
+
+    #[\Override]
+    public function isQueued(CommandInterface $command): bool
+    {
+        return $this->parent->isQueued($command);
+    }
+
+    #[\Override]
+    public function cancel(CommandInterface $command): void
+    {
+        $this->parent->cancel($command);
+    }
+
+    #[\Override]
+    public function reject(CommandInterface $command, \Throwable $reason): void
+    {
+        $this->parent->reject($command, $reason);
+    }
+
+    #[\Override]
+    public function dispatch(ServerResponseInterface $response): void
+    {
+        $this->parent->dispatch($response);
+    }
+
+    #[\Override]
+    public function fork(): ClientInterface
+    {
+        return $this->parent->fork();
+    }
+
+    public function destroy(): void
+    {
+        $this->requests === [] or ($this->cleanup)($this->requests);
+        $this->requests = [];
+        unset($this->parent, $this->cleanup);
+    }
+}

src/Internal/Transport/Router/StartWorkflow.php

@@ -85,7 +85,7 @@ public function handle(ServerRequestInterface $request, array $headers, Deferred
 
         $context = new WorkflowContext(
             $this->services,
-            $this->services->client,
+            $this->services->client->fork(),
             $instance,
             $input,
             $lastCompletionResult,

src/Internal/Workflow/WorkflowContext.php

@@ -666,7 +666,8 @@ public function destroy(): void
     {
         $this->awaits = [];
         $this->workflowInstance->destroy();
-        unset($this->workflowInstance);
+        $this->client->destroy();
+        unset($this->workflowInstance, $this->client);
     }
 
     protected function awaitRequest(callable|Mutex|PromiseInterface ...$conditions): PromiseInterface

tests/Feature/Testing/CapturedClient.php

@@ -52,33 +52,6 @@ public function send(CommandInterface $request): void
         $this->parent->send($request);
     }
 
-    /**
-     * @param RequestInterface $request
-     * @return \Closure
-     */
-    private function onFulfilled(RequestInterface $request): \Closure
-    {
-        return function ($response) use ($request) {
-            unset($this->requests[$request->getID()]);
-
-            return $response;
-        };
-    }
-
-    /**
-     * @param RequestInterface $request
-     * @return \Closure
-     * @psalm-suppress UnusedClosureParam
-     */
-    private function onRejected(RequestInterface $request): \Closure
-    {
-        return function (\Throwable $error) use ($request) {
-            unset($this->requests[$request->getID()]);
-
-            throw $error;
-        };
-    }
-
     /**
      * {@inheritDoc}
      */
@@ -121,4 +94,41 @@ public function reject(CommandInterface $command, \Throwable $reason): void
     {
         $this->parent->reject($command, $reason);
     }
+
+    public function dispatch(CommandInterface $response): void
+    {
+        $this->parent->dispatch($response);
+    }
+
+    public function fork(): ClientInterface
+    {
+        return $this->parent->fork();
+    }
+
+    /**
+     * @param RequestInterface $request
+     * @return \Closure
+     */
+    private function onFulfilled(RequestInterface $request): \Closure
+    {
+        return function ($response) use ($request) {
+            unset($this->requests[$request->getID()]);
+
+            return $response;
+        };
+    }
+
+    /**
+     * @param RequestInterface $request
+     * @return \Closure
+     * @psalm-suppress UnusedClosureParam
+     */
+    private function onRejected(RequestInterface $request): \Closure
+    {
+        return function (\Throwable $error) use ($request): void {
+            unset($this->requests[$request->getID()]);
+
+            throw $error;
+        };
+    }
 }

tests/Functional/WorkflowTestCase.php

@@ -363,12 +363,12 @@ public function testAwaitWithOneTimer_Leaks(): void
         $this->assertSame(0, $after - $before);
     }
 
-    public function testDetachedScope_Leaks(): void
+    public function testDetachedScopeRequests_Leaks(): void
     {
         $worker = WorkerMock::createMock();
 
         // Run the workflow $i times
-        for ($id = 9000, $i = 0; $i < 100; ++$i) {
+        for ($id = 9000, $i = 0; $i < 20; ++$i) {
             $uuid1 = Uuid::v4();
             $uuid2 = Uuid::v4();
             $id1 = ++$id;
@@ -383,19 +383,13 @@ public function testDetachedScope_Leaks(): void
                 LOG;
 
             $worker->run($this, Splitter::createFromString($log)->getQueue());
-            if ($i === 3) {
-                \gc_collect_cycles();
-                $before = \memory_get_usage();
-            }
         }
-        $after = \memory_get_usage();
 
+        // Check there are no hanging requests
         $factory = self::getPrivate($worker, 'factory');
         $client = self::getPrivate($factory, 'client');
         $requests = self::getPrivate($client, 'requests');
         self::assertCount(0, $requests);
-
-        $this->assertSame(0, $after - $before);
     }
 
     /**

tests/Unit/Framework/ClientMock.php

@@ -104,6 +104,13 @@ public function reject(CommandInterface $command, \Throwable $reason): void
         $request->reject($reason);
     }
 
+    public function fork(): self
+    {
+        return new self($this->queue);
+    }
+
+    public function destroy(): void {}
+
     private function fetch(int $id): Deferred
     {
         $request = $this->get($id);

tests/Unit/Router/StartWorkflowTestCase.php

@@ -132,7 +132,7 @@ protected function setUp(): void
         $this->router = new StartWorkflow($this->services);
         $this->workflowContext = new WorkflowContext(
             $this->services,
-            $this->services->client,
+            $this->services->client->fork(),
             $this->createMockForIntersectionOfInterfaces([WorkflowInstanceInterface::class, Destroyable::class]),
             new Input(),
             EncodedValues::empty(),