Merge pull request #590 : Fix memory leak on upsert SA/MEMO requests

Author: roxblnfk

.gitignore

@@ -20,3 +20,4 @@ Thumbs.db
 rr
 temporal-test-server
 .ai
+.context

psalm-baseline.xml

@@ -966,19 +966,6 @@
     </PossiblyInvalidArgument>
   </file>
   <file src="src/Internal/Transport/Client.php">
-    <InternalClass>
-      <code><![CDATA[self::ERROR_REQUEST_ID_DUPLICATION]]></code>
-      <code><![CDATA[self::ERROR_REQUEST_NOT_FOUND]]></code>
-    </InternalClass>
-    <InternalMethod>
-      <code><![CDATA[fetch]]></code>
-      <code><![CDATA[get]]></code>
-      <code><![CDATA[reject]]></code>
-      <code><![CDATA[request]]></code>
-    </InternalMethod>
-    <InternalProperty>
-      <code><![CDATA[$this->requests]]></code>
-    </InternalProperty>
     <PossiblyInvalidArgument>
       <code><![CDATA[$command->getID()]]></code>
       <code><![CDATA[$command->getID()]]></code>

src/Internal/Transport/Client.php

@@ -25,7 +25,7 @@
 
 /**
  * @internal Client is an internal library class, please do not use it in your code.
- * @psalm-internal Temporal\Client\Internal\Transport
+ * @psalm-internal Temporal\Internal\Transport
  */
 final class Client implements ClientInterface
 {
@@ -49,7 +49,7 @@ public function dispatch(ServerResponseInterface $response): void
     {
         $id = $response->getID();
         if (!isset($this->requests[$id])) {
-            $this->request(new UndefinedResponse(
+            $this->send(new UndefinedResponse(
                 \sprintf('Got the response to undefined request %s', $id),
             ));
             return;
@@ -84,9 +84,9 @@ public function request(RequestInterface $request, ?WorkflowContextInterface $co
 
         $id = $request->getID();
 
-        if (isset($this->requests[$id])) {
-            throw new \OutOfBoundsException(\sprintf(self::ERROR_REQUEST_ID_DUPLICATION, $id));
-        }
+        \array_key_exists($id, $this->requests) and throw new \OutOfBoundsException(
+            \sprintf(self::ERROR_REQUEST_ID_DUPLICATION, $id),
+        );
 
         $deferred = new Deferred();
         $this->requests[$id] = [$deferred, $context];

src/Internal/Workflow/ScopeContext.php

@@ -16,7 +16,6 @@
 use Temporal\Exception\Failure\CanceledFailure;
 use Temporal\Internal\Transport\CompletableResult;
 use Temporal\Internal\Workflow\Process\Scope;
-use Temporal\Promise;
 use Temporal\Worker\Transport\Command\RequestInterface;
 use Temporal\Workflow\CancellationScopeInterface;
 use Temporal\Workflow\ScopedContextInterface;
@@ -75,11 +74,11 @@ public function request(
             'Attempt to send request to cancelled scope',
         );
 
-        $promise = $this->parent->request($request);
         if (!$waitResponse) {
-            return Promise::resolve();
+            return $this->parent->request($request, $cancellable, false);
         }
 
+        $promise = $this->parent->request($request);
         ($this->onRequest)($request, $promise);
 
         return new CompletableResult(

src/Internal/Workflow/WorkflowContext.php

@@ -444,13 +444,21 @@ public function timer($interval): PromiseInterface
         )(new TimerInput($dateInterval));
     }
 
-    public function request(RequestInterface $request, bool $cancellable = true): PromiseInterface
-    {
+    public function request(
+        RequestInterface $request,
+        bool $cancellable = true,
+        bool $waitResponse = true,
+    ): PromiseInterface {
         $this->recordTrace();
 
         // Intercept workflow outbound calls
         return $this->requestInterceptor->with(
-            function (RequestInterface $request): PromiseInterface {
+            function (RequestInterface $request) use ($waitResponse): PromiseInterface {
+                if (!$waitResponse) {
+                    $this->client->send($request);
+                    return Promise::resolve();
+                }
+
                 return $this->client->request($request, $this);
             },
             /** @see WorkflowOutboundRequestInterceptor::handleOutboundRequest() */
@@ -476,7 +484,7 @@ function (UpsertMemoInput $input): PromiseInterface {
                     return resolve();
                 }
 
-                $result = $this->request(new UpsertMemo($input->memo), false);
+                $result = $this->request(new UpsertMemo($input->memo), false, false);
 
                 /** @psalm-suppress UnsupportedPropertyReferenceUsage $memo */
                 $memo = &$this->input->info->memo;
@@ -505,7 +513,7 @@ function (UpsertSearchAttributesInput $input): PromiseInterface {
                     return resolve();
                 }
 
-                $result = $this->request(new UpsertSearchAttributes($input->searchAttributes), false);
+                $result = $this->request(new UpsertSearchAttributes($input->searchAttributes), false, false);
 
                 /** @psalm-suppress UnsupportedPropertyReferenceUsage $sa */
                 $sa = &$this->input->info->searchAttributes;
@@ -533,7 +541,7 @@ function (UpsertTypedSearchAttributesInput $input): PromiseInterface {
                     return resolve();
                 }
 
-                $result = $this->request(new UpsertTypedSearchAttributes($input->updates), false);
+                $result = $this->request(new UpsertTypedSearchAttributes($input->updates), false, false);
 
                 // Merge changes
                 $tsa = $this->input->info->typedSearchAttributes;

src/Workflow/WorkflowContextInterface.php

@@ -128,7 +128,11 @@ public function registerUpdate(string $name, callable $handler, ?callable $valid
      *
      * @internal This is an internal method
      */
-    public function request(RequestInterface $request, bool $cancellable = true): PromiseInterface;
+    public function request(
+        RequestInterface $request,
+        bool $cancellable = true,
+        bool $waitResponse = true,
+    ): PromiseInterface;
 
     /**
      * Updates the behavior of an existing workflow to resolve inconsistency errors during replay.

tests/Acceptance/App/Attribute/Client.php

@@ -16,6 +16,5 @@ public function __construct(
         public float|null $timeout = null,
         public \Closure|array|string|null $pipelineProvider = null,
         public array $payloadConverters = [],
-    ) {
-    }
+    ) {}
 }

tests/Unit/Framework/ClientMock.php

@@ -45,7 +45,7 @@ public function __construct(QueueInterface $queue)
     public function dispatch(ServerResponseInterface $response): void
     {
         if (!isset($this->requests[$response->getID()])) {
-            $this->request(new UndefinedResponse(
+            $this->send(new UndefinedResponse(
                 \sprintf('Got the response to undefined request %s', $response->getID()),
             ));
             return;