Merge pull request #620: Activity pause

Author: roxblnfk

psalm-baseline.xml

@@ -1,5 +1,5 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<files psalm-version="6.12.0@cf420941d061a57050b6c468ef2c778faf40aee2">
+<files psalm-version="6.12.1@e71404b0465be25cf7f8a631b298c01c5ddd864f">
   <file src="src/Activity.php">
     <ImplicitToStringCast>
       <code><![CDATA[$type]]></code>
@@ -352,33 +352,11 @@
   </file>
   <file src="src/Exception/Client/ActivityCompletionException.php">
     <PossiblyInvalidArgument>
-      <code><![CDATA[$e === null ? 0 : $e->getCode()]]></code>
-      <code><![CDATA[$e->getCode()]]></code>
-      <code><![CDATA[$e->getCode()]]></code>
+      <code><![CDATA[$code]]></code>
     </PossiblyInvalidArgument>
     <PossiblyNullReference>
       <code><![CDATA[getID]]></code>
     </PossiblyNullReference>
-    <UnsafeInstantiation>
-      <code><![CDATA[new static(
-            $e->getMessage(),
-            $e->getCode(),
-            $e,
-        )]]></code>
-      <code><![CDATA[new static(
-            self::buildMessage(
-                [
-                    'workflowId' => $info->workflowExecution->getID(),
-                    'runId' => $info->workflowExecution->getRunID(),
-                    'activityId' => $info->id,
-                    'activityType' => $info->type->name,
-                ],
-            ),
-            $e === null ? 0 : $e->getCode(),
-            $e,
-        )]]></code>
-      <code><![CDATA[new static('', $e->getCode(), $e)]]></code>
-    </UnsafeInstantiation>
   </file>
   <file src="src/Exception/Client/MultyOperation/OperationStatus.php">
     <InvalidArgument>

src/Activity.php

@@ -11,6 +11,7 @@
 
 namespace Temporal;
 
+use Temporal\Activity\ActivityCancellationDetails;
 use Temporal\Activity\ActivityContextInterface;
 use Temporal\Activity\ActivityInfo;
 use Temporal\DataConverter\Type;
@@ -83,15 +84,27 @@ public static function hasHeartbeatDetails(): bool
      * This method retrieves the payload that was passed into the last call of the {@see Activity::heartbeat()} method.
      *
      * @param Type|string|\ReflectionType|\ReflectionClass|null $type
-     * @return mixed
      * @throws OutOfContextException in the absence of the activity execution context.
      */
-    public static function getHeartbeatDetails($type = null)
+    public static function getHeartbeatDetails($type = null): mixed
     {
         /** @var ActivityContextInterface $context */
         $context = self::getCurrentContext();
 
-        return $context->getHeartbeatDetails($type);
+        return $context->getLastHeartbeatDetails($type);
+    }
+
+    /**
+     * Cancellation details of the current activity, if any.
+     *
+     * Once set, cancellation details do not change.
+     */
+    public static function getCancellationDetails(): ?ActivityCancellationDetails
+    {
+        /** @var ActivityContextInterface $context */
+        $context = self::getCurrentContext();
+
+        return $context->getCancellationDetails();
     }
 
     /**
@@ -131,7 +144,7 @@ public static function doNotCompleteOnReturn(): void
      * ```
      *
      * @param mixed $details In case of activity timeout details are returned
-     * as a field of the exception thrown.
+     *        as a field of the exception thrown.
      *
      * @throws OutOfContextException in the absence of the activity execution context.
      */

src/Activity/ActivityCancellationDetails.php

@@ -0,0 +1,22 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Temporal\Activity;
+
+/**
+ * Provides the reasons for the activity's cancellation.
+ */
+final class ActivityCancellationDetails
+{
+    /**
+     * @internal
+     */
+    public function __construct(
+        public readonly bool $notFound = false,
+        public readonly bool $cancelRequested = false,
+        public readonly bool $paused = false,
+        public readonly bool $timedOut = false,
+        public readonly bool $workerShutdown = false,
+    ) {}
+}

src/Activity/ActivityContextInterface.php

@@ -14,6 +14,9 @@
 use Temporal\Activity;
 use Temporal\DataConverter\Type;
 use Temporal\DataConverter\ValuesInterface;
+use Temporal\Exception\Client\ActivityCanceledException;
+use Temporal\Exception\Client\ActivityCompletionException;
+use Temporal\Exception\Client\ActivityPausedException;
 
 interface ActivityContextInterface
 {
@@ -44,9 +47,8 @@ public function hasHeartbeatDetails(): bool;
      * @see Activity::getHeartbeatDetails()
      *
      * @param Type|string $type
-     * @return mixed
      */
-    public function getHeartbeatDetails($type = null);
+    public function getLastHeartbeatDetails($type = null): mixed;
 
     /**
      * Marks the activity as incomplete for asynchronous completion.
@@ -58,11 +60,23 @@ public function doNotCompleteOnReturn(): void;
     /**
      * Use to notify workflow that activity execution is alive.
      *
+     * @throws ActivityCompletionException
+     * @throws ActivityCanceledException
+     * @throws ActivityPausedException
+     *
      * @see Activity::heartbeat()
      *
-     * @param mixed $details
      */
-    public function heartbeat($details): void;
+    public function heartbeat(mixed $details): void;
+
+    /**
+     * Cancellation details of the current activity, if any.
+     *
+     * Once set, cancellation details do not change.
+     *
+     * @see Activity::getCancellationDetails()
+     */
+    public function getCancellationDetails(): ?ActivityCancellationDetails;
 
     /**
      * Get the currently running activity instance.

src/Exception/Client/ActivityCompletionException.php

@@ -21,10 +21,15 @@ class ActivityCompletionException extends TemporalException
     private ?string $activityType = null;
     private ?string $activityId = null;
 
+    final public function __construct(string $message = "", string|int $code = 0, ?\Throwable $previous = null)
+    {
+        parent::__construct($message, $code, $previous);
+    }
+
     /**
-     * @return static
+     * @internal
      */
-    public static function fromPrevious(\Throwable $e): self
+    public static function fromPrevious(\Throwable $e): static
     {
         return new static(
             $e->getMessage(),
@@ -34,9 +39,9 @@ public static function fromPrevious(\Throwable $e): self
     }
 
     /**
-     * @return static
+     * @internal
      */
-    public static function fromPreviousWithActivityId(string $activityId, \Throwable $e): self
+    public static function fromPreviousWithActivityId(string $activityId, \Throwable $e): static
     {
         $e = new static('', $e->getCode(), $e);
         $e->activityId = $activityId;
@@ -45,9 +50,9 @@ public static function fromPreviousWithActivityId(string $activityId, \Throwable
     }
 
     /**
-     * @return static
+     * @internal
      */
-    public static function fromActivityInfo(ActivityInfo $info, ?\Throwable $e = null): self
+    public static function fromActivityInfo(ActivityInfo $info, ?\Throwable $e = null): static
     {
         $e = new static(
             self::buildMessage(

src/Exception/Client/ActivityPausedException.php

@@ -0,0 +1,20 @@
+<?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\Exception\Client;
+
+/**
+ * Indicates that the activity was paused by the user.
+ *
+ * Catching this exception directly is discouraged and catching
+ * the parent class {@link ActivityCompletionException} is recommended instead.
+ */
+final class ActivityPausedException extends ActivityCompletionException {}

src/Internal/Activity/ActivityContext.php

@@ -11,6 +11,7 @@
 
 namespace Temporal\Internal\Activity;
 
+use Temporal\Activity\ActivityCancellationDetails;
 use Temporal\Activity\ActivityContextInterface;
 use Temporal\Activity\ActivityInfo;
 use Temporal\DataConverter\DataConverterInterface;
@@ -19,6 +20,7 @@
 use Temporal\DataConverter\ValuesInterface;
 use Temporal\Exception\Client\ActivityCanceledException;
 use Temporal\Exception\Client\ActivityCompletionException;
+use Temporal\Exception\Client\ActivityPausedException;
 use Temporal\Exception\Client\ServiceClientException;
 use Temporal\Interceptor\HeaderInterface;
 use Temporal\Internal\Interceptor\HeaderCarrier;
@@ -31,26 +33,17 @@ final class ActivityContext implements ActivityContextInterface, HeaderCarrier
     private ActivityInfo $info;
 
     private bool $doNotCompleteOnReturn = false;
-    private RPCConnectionInterface $rpc;
-    private DataConverterInterface $converter;
-    private ?ValuesInterface $heartbeatDetails;
-    private ValuesInterface $input;
-    private HeaderInterface $header;
     private ?\WeakReference $instance = null;
+    private ?ActivityCancellationDetails $cancellationDetails = null;
 
     public function __construct(
-        RPCConnectionInterface $rpc,
-        DataConverterInterface $converter,
-        ValuesInterface $input,
-        HeaderInterface $header,
-        ?ValuesInterface $lastHeartbeatDetails = null,
+        private readonly RPCConnectionInterface $rpc,
+        private readonly DataConverterInterface $converter,
+        private ValuesInterface $input,
+        private HeaderInterface $header,
+        private readonly ?ValuesInterface $lastHeartbeatDetails = null,
     ) {
         $this->info = new ActivityInfo();
-        $this->rpc = $rpc;
-        $this->converter = $converter;
-        $this->heartbeatDetails = $lastHeartbeatDetails;
-        $this->input = $input;
-        $this->header = $header;
     }
 
     public function getInfo(): ActivityInfo
@@ -91,20 +84,19 @@ public function getDataConverter(): DataConverterInterface
 
     public function hasHeartbeatDetails(): bool
     {
-        return $this->heartbeatDetails !== null;
+        return $this->lastHeartbeatDetails !== null;
     }
 
     /**
      * @param Type|string $type
-     * @return mixed
      */
-    public function getHeartbeatDetails($type = null)
+    public function getLastHeartbeatDetails($type = null): mixed
     {
         if (!$this->hasHeartbeatDetails()) {
             return null;
         }
 
-        return $this->heartbeatDetails->getValue(0, $type);
+        return $this->lastHeartbeatDetails->getValue(0, $type);
     }
 
     public function doNotCompleteOnReturn(): void
@@ -117,13 +109,7 @@ public function isDoNotCompleteOnReturn(): bool
         return $this->doNotCompleteOnReturn;
     }
 
-    /**
-     * @param mixed $details
-     *
-     * @throws ActivityCompletionException
-     * @throws ActivityCanceledException
-     */
-    public function heartbeat($details): void
+    public function heartbeat(mixed $details): void
     {
         // we use native host process RPC here to avoid excessive GRPC connections and to handle throttling
         // on Golang end
@@ -141,14 +127,29 @@ public function heartbeat($details): void
                 ],
             );
 
-            if (!empty($response['canceled'])) {
-                throw ActivityCanceledException::fromActivityInfo($this->info);
+            $cancelled = (bool) ($response['canceled'] ?? false);
+            $paused = (bool) ($response['paused'] ?? false);
+
+            if ($cancelled || $paused) {
+                $this->cancellationDetails ??= new ActivityCancellationDetails(
+                    cancelRequested: $cancelled,
+                    paused: $paused,
+                );
+
+                throw $cancelled
+                    ? ActivityCanceledException::fromActivityInfo($this->info)
+                    : ActivityPausedException::fromActivityInfo($this->info);
             }
         } catch (ServiceClientException $e) {
             throw ActivityCompletionException::fromActivityInfo($this->info, $e);
         }
     }
 
+    public function getCancellationDetails(): ?ActivityCancellationDetails
+    {
+        return $this->cancellationDetails;
+    }
+
     public function getInstance(): object
     {
         \assert($this->instance !== null, 'Activity instance is not available');

src/Internal/Declaration/Dispatcher/Dispatcher.php

@@ -88,9 +88,9 @@ private function scopeMatches(int $scope): bool
     }
 
     /**
-     * @psalm-return FunctionExecutor
      *
      * @return \Closure(object, array): mixed
+     * @psalm-return FunctionExecutor
      */
     private function createExecutorFromMethod(\ReflectionMethod $fun): \Closure
     {
@@ -104,9 +104,9 @@ private function createExecutorFromMethod(\ReflectionMethod $fun): \Closure
     }
 
     /**
-     * @psalm-return FunctionExecutor
      *
      * @return \Closure(object, array): mixed
+     * @psalm-return FunctionExecutor
      */
     private function createExecutorFromFunction(\ReflectionFunction $fun): \Closure
     {

src/Promise.php

@@ -153,8 +153,8 @@ static function (iterable $array) use ($count, $cancellationQueue, $resolve, $re
      * The `$callback` function receives each item as an argument, where the item is
      * a fully resolved value of a promise or a value in `$promises`.
      *
-     * @psalm-param PromiseMapCallback $map
      * @param iterable<int, PromiseInterface|mixed> $promises
+     * @psalm-param PromiseMapCallback $map
      */
     public static function map(iterable $promises, callable $map): PromiseInterface
     {
@@ -203,10 +203,10 @@ static function (mixed $mapped) use ($i, &$values, &$toResolve, $resolve): void
      * promises and/or values. The `$reduce` callback may return either a value or a promise,
      * and `$initial` may be a promise or a value for the starting value.
      *
-     * @psalm-param PromiseReduceCallback $reduce
      * @param iterable<int, PromiseInterface|mixed> $promises
      * @param callable(mixed $current, mixed $carry, int $current, positive-int $items): mixed $reduce
      * @param mixed $initial
+     * @psalm-param PromiseReduceCallback $reduce
      */
     public static function reduce(iterable $promises, callable $reduce, $initial = null): PromiseInterface
     {

testing/src/Environment.php

@@ -97,7 +97,6 @@ public function startTemporalServer(
                 '--dynamic-config-value', 'frontend.enableUpdateWorkflowExecution=true',
                 '--dynamic-config-value', 'frontend.enableUpdateWorkflowExecutionAsyncAccepted=true',
                 '--dynamic-config-value', 'frontend.enableExecuteMultiOperation=true',
-                '--dynamic-config-value', 'system.enableEagerWorkflowStart=true',
                 '--log-level', 'error',
                 '--headless',
                 ...$parameters,