Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 7 additions & 21 deletions psalm-baseline.xml
Original file line number Diff line number Diff line change
@@ -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>
Expand Down Expand Up @@ -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>
Expand Down Expand Up @@ -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>
Expand Down Expand Up @@ -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>
Expand All @@ -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>
Expand Down Expand Up @@ -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>
Expand Down
2 changes: 1 addition & 1 deletion src/Exception/Client/ActivityCompletionException.php
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}

/**
Expand Down
10 changes: 1 addition & 9 deletions src/Internal/Declaration/Dispatcher/Dispatcher.php
Original file line number Diff line number Diff line change
Expand Up @@ -13,9 +13,6 @@

use JetBrains\PhpStorm\Pure;

/**
* @psalm-type FunctionExecutor = \Closure(object, array): mixed
*/
class Dispatcher implements DispatcherInterface
{
/**
Expand All @@ -30,7 +27,6 @@ class Dispatcher implements DispatcherInterface

/**
* @var \Closure(object, array): mixed
* @psalm-var FunctionExecutor
*/
private \Closure $executor;

Expand Down Expand Up @@ -88,9 +84,7 @@ private function scopeMatches(int $scope): bool
}

/**
*
* @return \Closure(object, array): mixed
* @psalm-return FunctionExecutor
*/
private function createExecutorFromMethod(\ReflectionMethod $fun): \Closure
{
Expand All @@ -104,9 +98,7 @@ private function createExecutorFromMethod(\ReflectionMethod $fun): \Closure
}

/**
*
* @return \Closure(object, array): mixed
* @psalm-return FunctionExecutor
*/
private function createExecutorFromFunction(\ReflectionFunction $fun): \Closure
{
Expand All @@ -131,7 +123,7 @@ private function createExecutorFromFunction(\ReflectionFunction $fun): \Closure
}

/**
* @psalm-return FunctionExecutor
* @return \Closure(object, array): mixed
*/
private function boot(\ReflectionFunctionAbstract $fun): void
{
Expand Down
22 changes: 22 additions & 0 deletions src/Internal/Exception/UndefinedRequestException.php
Original file line number Diff line number Diff line change
@@ -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 {}
21 changes: 8 additions & 13 deletions src/Internal/Transport/Router/CancelWorkflow.php
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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 [];
}
}
15 changes: 5 additions & 10 deletions src/Internal/Transport/Router/WorkflowProcessAwareRoute.php
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -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),
);
}
Expand Down
15 changes: 12 additions & 3 deletions src/Internal/Transport/Server.php
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down
11 changes: 6 additions & 5 deletions src/Internal/Workflow/Process/Scope.php
Original file line number Diff line number Diff line change
Expand Up @@ -215,6 +215,7 @@ public function cancel(?\Throwable $reason = null): void
if ($this->cancelled) {
return;
}

$this->cancelled = true;

foreach ($this->onCancel as $i => $handler) {
Expand Down Expand Up @@ -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;
Expand Down
Original file line number Diff line number Diff line change
@@ -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();
}
}
Loading
Loading