diff --git a/psalm-baseline.xml b/psalm-baseline.xml index 461641215..3a68d3113 100644 --- a/psalm-baseline.xml +++ b/psalm-baseline.xml @@ -1,5 +1,5 @@ - + @@ -346,9 +346,6 @@ - - - @@ -559,10 +556,10 @@ - + - + @@ -1066,11 +1063,6 @@ - - - - - @@ -1084,6 +1076,7 @@ getID()]]> + @@ -1115,17 +1108,10 @@ - - - running->find($runId) ?? throw new \LogicException( - \sprintf(self::ERROR_PROCESS_NOT_FOUND, $runId), - )]]> - - - - - + + getMessage()]]> + diff --git a/src/Exception/Client/ActivityCompletionException.php b/src/Exception/Client/ActivityCompletionException.php index c6d375c49..2f00fbc59 100644 --- a/src/Exception/Client/ActivityCompletionException.php +++ b/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); } /** diff --git a/src/Internal/Declaration/Dispatcher/Dispatcher.php b/src/Internal/Declaration/Dispatcher/Dispatcher.php index 126e41441..f74ef0ca7 100644 --- a/src/Internal/Declaration/Dispatcher/Dispatcher.php +++ b/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 { diff --git a/src/Internal/Exception/UndefinedRequestException.php b/src/Internal/Exception/UndefinedRequestException.php new file mode 100644 index 000000000..ba81b8824 --- /dev/null +++ b/src/Internal/Exception/UndefinedRequestException.php @@ -0,0 +1,22 @@ +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 []; } } diff --git a/src/Internal/Transport/Router/WorkflowProcessAwareRoute.php b/src/Internal/Transport/Router/WorkflowProcessAwareRoute.php index fff39c2e1..66dfcd73b 100644 --- a/src/Internal/Transport/Router/WorkflowProcessAwareRoute.php +++ b/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 + */ 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), ); } diff --git a/src/Internal/Transport/Server.php b/src/Internal/Transport/Server.php index 9926f9fec..03b269f82 100644 --- a/src/Internal/Transport/Server.php +++ b/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; diff --git a/src/Internal/Workflow/Process/Scope.php b/src/Internal/Workflow/Process/Scope.php index bf357dae9..3b3fa0321 100644 --- a/src/Internal/Workflow/Process/Scope.php +++ b/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; diff --git a/tests/Acceptance/Extra/Stability/DynamicSignalWithPromisesTest.php b/tests/Acceptance/Extra/Stability/DynamicSignalWithPromisesTest.php new file mode 100644 index 000000000..f01d9b061 --- /dev/null +++ b/tests/Acceptance/Extra/Stability/DynamicSignalWithPromisesTest.php @@ -0,0 +1,72 @@ +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(); + } +} diff --git a/tests/Acceptance/Extra/Stability/ResetWorkerTest.php b/tests/Acceptance/Extra/Stability/ResetWorkerTest.php new file mode 100644 index 000000000..583de204b --- /dev/null +++ b/tests/Acceptance/Extra/Stability/ResetWorkerTest.php @@ -0,0 +1,141 @@ +withTimeout(1) + ->newUntypedWorkflowStub( + 'Extra_Stability_ResetWorker', + WorkflowOptions::new() + ->withTaskQueue($feature->taskQueue) + ->withWorkflowExecutionTimeout(20), + ); + + # Start the Workflow with a 10-second timer + $client->start($stub, 16); + + # Query the Workflow to kill the Worker + try { + $stub->query('die'); + self::fail('Query must fail with a timeout'); + } catch (WorkflowServiceException $e) { + # Should fail with a timeout + self::assertInstanceOf(TimeoutException::class, $e->getPrevious()); + } + + # Cancel Workflow + $stub->cancel(); + + try { + # Workflow must be canceled + $stub->getResult(timeout: 12); + } catch (WorkflowFailedException $e) { + self::assertInstanceOf(CanceledFailure::class, $e->getPrevious()); + return; + } + + self::fail('Workflow must fail with a canceled failure'); + } + + #[Test] + public function resetWithSignal( + WorkflowClientInterface $client, + Feature $feature, + ): void { + # Create a Workflow stub with an execution timeout 12 seconds + $stub = $client->withTimeout(1) + ->newUntypedWorkflowStub( + 'Extra_Stability_ResetWorker', + WorkflowOptions::new() + ->withTaskQueue($feature->taskQueue) + ->withWorkflowExecutionTimeout(20), + ); + + # Start the Workflow with a 10-second timer + $client->start($stub, 16); + + # Query the Workflow to kill the Worker + try { + $stub->query('die'); + self::fail('Query must fail with a timeout'); + } catch (WorkflowServiceException $e) { + # Should fail with a timeout + self::assertInstanceOf(TimeoutException::class, $e->getPrevious()); + } + + $stub->signal('exit'); + + try { + # Workflow must be canceled + $result = $stub->getResult(timeout: 16); + self::assertSame('Signal', $result); + } catch (\Throwable) { + $this->fail('Workflow must finish successfully and no timeout must be thrown'); + } + + # Check that Side Effect was not lost + $found = false; + foreach ($client->getWorkflowHistory($stub->getExecution()) as $event) { + if ($event->hasMarkerRecordedEventAttributes()) { + $record = $event->getMarkerRecordedEventAttributes(); + self::assertSame('SideEffect', $record->getMarkerName()); + $found = true; + break; + } + } + + self::assertTrue($found, 'Side Effect must be found in the Workflow history'); + } +} + +#[Workflow\WorkflowInterface] +class TestWorkflow +{ + private bool $exit = false; + + #[WorkflowMethod('Extra_Stability_ResetWorker')] + #[ReturnType(Type::TYPE_STRING)] + public function expire(int $seconds = 10): \Generator + { + $isTimer = ! yield Workflow::awaitWithTimeout($seconds, fn(): bool => $this->exit); + + return yield $isTimer ? 'Timer' : 'Signal'; + } + + #[Workflow\QueryMethod('die')] + public function die(int $sleep = 2): void + { + \sleep($sleep); + exit(1); + } + + #[Workflow\SignalMethod('exit')] + public function signal() + { + yield Workflow::uuid7(); + $this->exit = true; + } +} diff --git a/tests/Acceptance/Harness/ChildWorkflow/CancelAbandonTest.php b/tests/Acceptance/Harness/ChildWorkflow/CancelAbandonTest.php new file mode 100644 index 000000000..00fb154f5 --- /dev/null +++ b/tests/Acceptance/Harness/ChildWorkflow/CancelAbandonTest.php @@ -0,0 +1,108 @@ +getWorkflowHistory($stub->getExecution()) as $event) { + if ($event->hasChildWorkflowExecutionStartedEventAttributes()) { + $execution = $event->getChildWorkflowExecutionStartedEventAttributes()->getWorkflowExecution(); + break; + } + } + + if ($execution === null && \microtime(true) < $deadline) { + goto child_id; + } + + self::assertNotNull($execution, 'Child workflow execution not found in history'); + + # Get Child Workflow Stub + $child = $client->newUntypedRunningWorkflowStub( + $execution->getWorkflowId(), + $execution->getRunId(), + 'Harness_ChildWorkflow_CancelAbandon_Child', + ); + + # Cancel the parent workflow + $stub->cancel(); + # Expect the CanceledFailure in the parent workflow + self::assertSame('cancelled', $stub->getResult()); + + # Signal the child workflow to exit + $child->signal('exit'); + # No canceled failure in the child workflow + self::assertSame('test 42', $child->getResult()); + } +} + +#[WorkflowInterface] +class MainWorkflow +{ + #[WorkflowMethod('Harness_ChildWorkflow_CancelAbandon')] + public function run() + { + $child = Workflow::newUntypedChildWorkflowStub( + 'Harness_ChildWorkflow_CancelAbandon_Child', + Workflow\ChildWorkflowOptions::new() + ->withParentClosePolicy(Workflow\ParentClosePolicy::Abandon), + ); + + yield $child->start('test 42'); + + try { + return yield $child->getResult(); + } catch (CanceledFailure) { + return 'cancelled'; + } catch (ChildWorkflowFailure $failure) { + # Check CanceledFailure + return $failure->getPrevious()::class === CanceledFailure::class + ? 'cancelled' + : throw $failure; + } + } +} + +#[WorkflowInterface] +class ChildWorkflow +{ + private bool $exit = false; + + #[WorkflowMethod('Harness_ChildWorkflow_CancelAbandon_Child')] + public function run(string $input) + { + yield Workflow::await(fn(): bool => $this->exit); + return $input; + } + + #[Workflow\SignalMethod('exit')] + public function exit(): void + { + $this->exit = true; + } +}