Use special kind of failure on desync (#639)

Author: roxblnfk

psalm-baseline.xml

@@ -1,5 +1,5 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<files psalm-version="6.12.1@e71404b0465be25cf7f8a631b298c01c5ddd864f">
+<files psalm-version="6.13.0@70cdf647255a1362b426bb0f522a85817b8c791c">
   <file src="src/Activity.php">
     <ImplicitToStringCast>
       <code><![CDATA[$type]]></code>
@@ -346,9 +346,6 @@
     </MissingReturnType>
   </file>
   <file src="src/Exception/Client/ActivityCompletionException.php">
-    <PossiblyInvalidArgument>
-      <code><![CDATA[$code]]></code>
-    </PossiblyInvalidArgument>
     <PossiblyNullReference>
       <code><![CDATA[getID]]></code>
     </PossiblyNullReference>
@@ -559,10 +556,10 @@
   </file>
   <file src="src/Internal/Declaration/Dispatcher/Dispatcher.php">
     <InvalidReturnType>
-      <code><![CDATA[FunctionExecutor]]></code>
+      <code><![CDATA[\Closure(object, array): mixed]]></code>
     </InvalidReturnType>
     <MismatchingDocblockReturnType>
-      <code><![CDATA[FunctionExecutor]]></code>
+      <code><![CDATA[\Closure(object, array): mixed]]></code>
     </MismatchingDocblockReturnType>
     <PropertyNotSetInConstructor>
       <code><![CDATA[$executor]]></code>
@@ -1066,11 +1063,6 @@
       <code><![CDATA[PromiseInterface]]></code>
     </TooManyTemplateParams>
   </file>
-  <file src="src/Internal/Transport/Router/CancelWorkflow.php">
-    <DocblockTypeContradiction>
-      <code><![CDATA[$process === null]]></code>
-    </DocblockTypeContradiction>
-  </file>
   <file src="src/Internal/Transport/Router/DestroyWorkflow.php">
     <UndefinedInterfaceMethod>
       <code><![CDATA[pull]]></code>
@@ -1084,6 +1076,7 @@
   <file src="src/Internal/Transport/Router/InvokeQuery.php">
     <ArgumentTypeCoercion>
       <code><![CDATA[$request->getID()]]></code>
+      <code><![CDATA[$running]]></code>
     </ArgumentTypeCoercion>
     <UnusedVariable>
       <code><![CDATA[$headers]]></code>
@@ -1115,17 +1108,10 @@
       <code><![CDATA[Input]]></code>
     </UnnecessaryVarAnnotation>
   </file>
-  <file src="src/Internal/Transport/Router/WorkflowProcessAwareRoute.php">
-    <LessSpecificReturnStatement>
-      <code><![CDATA[$this->running->find($runId) ?? throw new \LogicException(
-            \sprintf(self::ERROR_PROCESS_NOT_FOUND, $runId),
-        )]]></code>
-    </LessSpecificReturnStatement>
-    <MoreSpecificReturnType>
-      <code><![CDATA[Process]]></code>
-    </MoreSpecificReturnType>
-  </file>
   <file src="src/Internal/Transport/Server.php">
+    <ArgumentTypeCoercion>
+      <code><![CDATA[$e->getMessage()]]></code>
+    </ArgumentTypeCoercion>
     <InvalidArgument>
       <code><![CDATA[$request]]></code>
       <code><![CDATA[$request]]></code>

src/Exception/Client/ActivityCompletionException.php

@@ -23,7 +23,7 @@ class ActivityCompletionException extends TemporalException
 
     final public function __construct(string $message = "", string|int $code = 0, ?\Throwable $previous = null)
     {
-        parent::__construct($message, $code, $previous);
+        parent::__construct($message, (int) $code, $previous);
     }
 
     /**

src/Internal/Declaration/Dispatcher/Dispatcher.php

@@ -13,9 +13,6 @@
 
 use JetBrains\PhpStorm\Pure;
 
-/**
- * @psalm-type FunctionExecutor = \Closure(object, array): mixed
- */
 class Dispatcher implements DispatcherInterface
 {
     /**
@@ -30,7 +27,6 @@ class Dispatcher implements DispatcherInterface
 
     /**
      * @var \Closure(object, array): mixed
-     * @psalm-var FunctionExecutor
      */
     private \Closure $executor;
 
@@ -88,9 +84,7 @@ private function scopeMatches(int $scope): bool
     }
 
     /**
-     *
      * @return \Closure(object, array): mixed
-     * @psalm-return FunctionExecutor
      */
     private function createExecutorFromMethod(\ReflectionMethod $fun): \Closure
     {
@@ -104,9 +98,7 @@ private function createExecutorFromMethod(\ReflectionMethod $fun): \Closure
     }
 
     /**
-     *
      * @return \Closure(object, array): mixed
-     * @psalm-return FunctionExecutor
      */
     private function createExecutorFromFunction(\ReflectionFunction $fun): \Closure
     {
@@ -131,7 +123,7 @@ private function createExecutorFromFunction(\ReflectionFunction $fun): \Closure
     }
 
     /**
-     * @psalm-return FunctionExecutor
+     * @return \Closure(object, array): mixed
      */
     private function boot(\ReflectionFunctionAbstract $fun): void
     {

src/Internal/Exception/UndefinedRequestException.php

@@ -0,0 +1,22 @@
+<?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\Exception;
+
+use Temporal\Internal\Transport\Request\UndefinedResponse;
+
+/**
+ * The exception is converted into {@see UndefinedResponse} and sent to the client.
+ * This kind of failure raises panic in the Temporal Worker on the RoadRunner side.
+ *
+ * @internal
+ */
+final class UndefinedRequestException extends \LogicException {}

src/Internal/Transport/Router/CancelWorkflow.php

@@ -13,14 +13,11 @@
 
 use React\Promise\Deferred;
 use Temporal\DataConverter\EncodedValues;
-use Temporal\Internal\Workflow\Process\Process;
+use Temporal\Internal\Exception\UndefinedRequestException;
 use Temporal\Worker\Transport\Command\ServerRequestInterface;
 
 class CancelWorkflow extends WorkflowProcessAwareRoute
 {
-    /**
-     * @var string
-     */
     private const ERROR_PROCESS_NOT_DEFINED = 'Unable to cancel workflow because workflow process #%s was not found';
 
     public function handle(ServerRequestInterface $request, array $headers, Deferred $resolver): void
@@ -30,17 +27,15 @@ public function handle(ServerRequestInterface $request, array $headers, Deferred
         $resolver->resolve(EncodedValues::fromValues([null]));
     }
 
-    public function cancel(string $runId): array
+    /**
+     * @throws UndefinedRequestException
+     */
+    private function cancel(string $runId): void
     {
-        /** @var Process $process */
-        $process = $this->running->find($runId);
-
-        if ($process === null) {
-            throw new \InvalidArgumentException(\sprintf(self::ERROR_PROCESS_NOT_DEFINED, $runId));
-        }
+        $process = $this->running->find($runId) ?? throw new UndefinedRequestException(
+            \sprintf(self::ERROR_PROCESS_NOT_DEFINED, $runId),
+        );
 
         $process->cancel();
-
-        return [];
     }
 }

src/Internal/Transport/Router/WorkflowProcessAwareRoute.php

@@ -11,7 +11,7 @@
 
 namespace Temporal\Internal\Transport\Router;
 
-use Temporal\Internal\Declaration\WorkflowInstanceInterface;
+use Temporal\Internal\Exception\UndefinedRequestException;
 use Temporal\Internal\Repository\RepositoryInterface;
 use Temporal\Internal\Workflow\Process\Process;
 
@@ -20,23 +20,18 @@ abstract class WorkflowProcessAwareRoute extends Route
     private const ERROR_PROCESS_NOT_FOUND = 'Workflow with the specified run identifier "%s" not found';
 
     public function __construct(
+        /**
+         * @var RepositoryInterface<Process>
+         */
         protected RepositoryInterface $running,
     ) {}
 
-    /**
-     * @param non-empty-string $runId
-     */
-    protected function findInstanceOrFail(string $runId): WorkflowInstanceInterface
-    {
-        return $this->findProcessOrFail($runId)->getWorkflowInstance();
-    }
-
     /**
      * @param non-empty-string $runId
      */
     protected function findProcessOrFail(string $runId): Process
     {
-        return $this->running->find($runId) ?? throw new \LogicException(
+        return $this->running->find($runId) ?? throw new UndefinedRequestException(
             \sprintf(self::ERROR_PROCESS_NOT_FOUND, $runId),
         );
     }

src/Internal/Transport/Server.php

@@ -12,7 +12,9 @@
 namespace Temporal\Internal\Transport;
 
 use React\Promise\PromiseInterface;
+use Temporal\Internal\Exception\UndefinedRequestException;
 use Temporal\Internal\Queue\QueueInterface;
+use Temporal\Internal\Transport\Request\UndefinedResponse;
 use Temporal\Worker\Transport\Command\Client\FailedClientResponse;
 use Temporal\Worker\Transport\Command\Client\SuccessClientResponse;
 use Temporal\Worker\Transport\Command\FailureResponseInterface;
@@ -84,12 +86,19 @@ private function onFulfilled(ServerRequestInterface $request): \Closure
     }
 
     /**
-     * @return \Closure(\Throwable): FailureResponseInterface
+     * @return \Closure(\Throwable): (FailureResponseInterface|never)
      */
     private function onRejected(ServerRequestInterface $request): \Closure
     {
-        return function (\Throwable $result) use ($request) {
-            $response = new FailedClientResponse($request->getID(), $result);
+        return function (\Throwable $e) use ($request) {
+            if ($e::class === UndefinedRequestException::class) {
+                // This is not a FailureResponseInterface, but it's a better place to handle it.
+                $response = new UndefinedResponse($e->getMessage());
+                $this->queue->push($response);
+                throw $e;
+            }
+
+            $response = new FailedClientResponse($request->getID(), $e);
             $this->queue->push($response);
 
             return $response;

src/Internal/Workflow/Process/Scope.php

@@ -215,6 +215,7 @@ public function cancel(?\Throwable $reason = null): void
         if ($this->cancelled) {
             return;
         }
+
         $this->cancelled = true;
 
         foreach ($this->onCancel as $i => $handler) {
@@ -357,19 +358,19 @@ protected function callSignalOrUpdateHandler(callable $handler, ValuesInterface
     protected function onRequest(RequestInterface $request, PromiseInterface $promise): void
     {
         $this->onCancel[++$this->cancelID] = function (?\Throwable $reason = null) use ($request): void {
+            $client = $this->context->getClient();
             if ($reason instanceof DestructMemorizedInstanceException) {
                 // memory flush
-                $this->context->getClient()->reject($request, $reason);
+                $client->reject($request, $reason);
                 return;
             }
 
-            if ($this->context->getClient()->isQueued($request)) {
-                $this->context->getClient()->cancel($request);
+            if ($client->isQueued($request)) {
+                $client->cancel($request);
                 return;
             }
-            // todo ->context or ->scopeContext?
 
-            $this->context->getClient()->request(new Cancel($request->getID()), $this->scopeContext);
+            $client->request(new Cancel($request->getID()), $this->scopeContext);
         };
 
         $cancelID = $this->cancelID;

tests/Acceptance/Extra/Stability/DynamicSignalWithPromisesTest.php

@@ -0,0 +1,72 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Temporal\Tests\Acceptance\Extra\Stability\DynamicSignalWithPromises;
+
+use PHPUnit\Framework\Attributes\Test;
+use React\Promise\Deferred;
+use React\Promise\PromiseInterface;
+use Temporal\Client\WorkflowStubInterface;
+use Temporal\Tests\Acceptance\App\Attribute\Stub;
+use Temporal\Tests\Acceptance\App\TestCase;
+use Temporal\Workflow;
+use Temporal\Workflow\WorkflowMethod;
+
+class DynamicSignalWithPromisesTest extends TestCase
+{
+    #[Test]
+    public function steps(
+        #[Stub('Extra_Stability_DynamicSignalWithPromises')] WorkflowStubInterface $stub,
+    ): void {
+        # Send signals to the workflow to trigger steps
+        $stub->signal('begin', 'foo');
+        $stub->signal('next1', 'bar');
+
+        # Assert that the workflow has processed the signals and updated the value
+        $this->assertSame(2, $stub->query('value')->getValue(0, 'int'));
+
+        # Send another signal to continue the workflow
+        $stub->signal('next2', 'baz');
+
+        # Assert that the workflow has processed the final signal and returned the expected value
+        $this->assertSame(3, $stub->query('value')->getValue(0, 'int'));
+
+        # Assert that the workflow has completed and returned the final result
+        $this->assertSame(3, $stub->getResult());
+    }
+}
+
+#[Workflow\WorkflowInterface]
+class TestWorkflow
+{
+    #[WorkflowMethod(name: 'Extra_Stability_DynamicSignalWithPromises')]
+    public function handler()
+    {
+        $value = 0;
+        Workflow::registerQuery('value', static function () use (&$value) {
+            return $value;
+        });
+
+        yield $this->promiseSignal('begin');
+        $value++;
+
+        yield $this->promiseSignal('next1');
+        $value++;
+
+        yield $this->promiseSignal('next2');
+        $value++;
+
+        return $value;
+    }
+
+    private function promiseSignal(string $name): PromiseInterface
+    {
+        $signal = new Deferred();
+        Workflow::registerSignal($name, static function (mixed $value) use ($signal): void {
+            $signal->resolve($value);
+        });
+
+        return $signal->promise();
+    }
+}