diff --git a/src/Internal/Workflow/WorkflowContext.php b/src/Internal/Workflow/WorkflowContext.php index b32c0c9b2..f32ed58e9 100644 --- a/src/Internal/Workflow/WorkflowContext.php +++ b/src/Internal/Workflow/WorkflowContext.php @@ -15,6 +15,7 @@ use Psr\Log\LoggerInterface; use Ramsey\Uuid\UuidInterface; use React\Promise\Deferred; +use React\Promise\Exception\LengthException; use React\Promise\PromiseInterface; use Temporal\Activity\ActivityOptions; use Temporal\Activity\ActivityOptionsInterface; @@ -68,6 +69,7 @@ use Temporal\Internal\Transport\Request\UpsertTypedSearchAttributes; use Temporal\Internal\Workflow\Process\HandlerState; use Temporal\Promise; +use Temporal\Worker\FeatureFlags; use Temporal\Worker\Transport\Command\RequestInterface; use Temporal\Workflow\ActivityStubInterface; use Temporal\Workflow\ChildWorkflowOptions; @@ -634,15 +636,29 @@ function (AwaitWithTimeoutInput $input): PromiseInterface { $timer = $this->request($request); \assert($timer instanceof CompletableResultInterface); - return $this->awaitRequest($timer, ...$input->conditions) - ->then(function () use ($timer, $requestId): bool { - $isCompleted = $timer->isComplete(); - if (!$isCompleted) { - // If internal timer was not completed then cancel it - $this->request(new Cancel($requestId)); - } - return !$isCompleted; - }); + $cancelPendingTimer = function () use ($timer, $requestId): void { + if (!$timer->isComplete()) { + $this->request(new Cancel($requestId)); + } + }; + + $onTimeout = static function () use ($timer, $cancelPendingTimer): bool { + $cancelPendingTimer(); + return !$timer->isComplete(); + }; + + if (FeatureFlags::$settleAwaitOnFirstSettledCondition) { + return $this->awaitRequest($timer, ...$input->conditions) + ->then( + $onTimeout, + static function (\Throwable $failure) use ($cancelPendingTimer): never { + $cancelPendingTimer(); + throw $failure; + }, + ); + } + + return $this->awaitRequest($timer, ...$input->conditions)->then($onTimeout); }, /** @see WorkflowOutboundCallsInterceptor::awaitWithTimeout() */ 'awaitWithTimeout', @@ -774,15 +790,31 @@ protected function awaitRequest(callable|Mutex|PromiseInterface ...$conditions): } } + if ($result === []) { + return reject(new LengthException('At least one condition is required to await.')); + } + if (\count($result) === 1) { return $result[0]; } + $onResolved = function (mixed $result) use ($conditionGroupId): mixed { + $this->resolveConditionGroup($conditionGroupId); + return $result; + }; + + if (FeatureFlags::$settleAwaitOnFirstSettledCondition) { + return Promise::race($result)->then( + $onResolved, + function (\Throwable $reason) use ($conditionGroupId): never { + $this->rejectConditionGroup($conditionGroupId); + throw $reason; + }, + ); + } + return Promise::any($result)->then( - function (mixed $result) use ($conditionGroupId): mixed { - $this->resolveConditionGroup($conditionGroupId); - return $result; - }, + $onResolved, function (\Throwable $reason) use ($conditionGroupId): void { $this->rejectConditionGroup($conditionGroupId); // Throw the first reason diff --git a/src/Worker/FeatureFlags.php b/src/Worker/FeatureFlags.php index 66557df0b..f4e15eabd 100644 --- a/src/Worker/FeatureFlags.php +++ b/src/Worker/FeatureFlags.php @@ -78,4 +78,15 @@ final class FeatureFlags * @link https://github.com/temporalio/sdk-php/issues/769 */ public static bool $propagateCancellationToNewScopes = false; + + /** + * Unblock multi-condition {@see Workflow::await()} / {@see Workflow::awaitWithTimeout()} on the + * first settled condition, propagating a rejected promise instead of ignoring it until the + * timeout. FALSE (default) keeps the old behavior so existing histories stay replay-compatible. + * + * @experimental + * @since SDK 2.18.0 + * @link https://github.com/temporalio/sdk-php/issues/399 + */ + public static bool $settleAwaitOnFirstSettledCondition = false; } diff --git a/src/Workflow.php b/src/Workflow.php index 0938d890a..6a303c793 100644 --- a/src/Workflow.php +++ b/src/Workflow.php @@ -304,6 +304,10 @@ public static function asyncDetached(callable $task): CancellationScopeInterface * $this->continued = true; * } * ``` + * + * To wait for the first *fulfilled* condition and ignore rejected promise + * conditions, combine them explicitly: + * `yield Workflow::await(\Temporal\Promise::any([$a, $b]))`. */ public static function await(callable|Mutex|PromiseInterface ...$conditions): PromiseInterface { diff --git a/tests/Unit/WorkflowContext/AwaitPromiseSettlementTestCase.php b/tests/Unit/WorkflowContext/AwaitPromiseSettlementTestCase.php new file mode 100644 index 000000000..872008f39 --- /dev/null +++ b/tests/Unit/WorkflowContext/AwaitPromiseSettlementTestCase.php @@ -0,0 +1,233 @@ +flagBackup = FeatureFlags::$settleAwaitOnFirstSettledCondition; + $this->factory = WorkerFactoryMock::create(); + $this->worker = $this->factory->newWorker(); + + parent::setUp(); + } + + protected function tearDown(): void + { + FeatureFlags::$settleAwaitOnFirstSettledCondition = $this->flagBackup; + + parent::tearDown(); + } + + public function testClosureFalseTimesOut(): void + { + $this->addToAssertionCount(1); + $this->worker->registerWorkflowObject( + new + #[Workflow\WorkflowInterface] + class { + #[WorkflowMethod(name: 'AwaitPromiseWorkflow')] + public function handler(): iterable + { + $result = yield Workflow::awaitWithTimeout(5, static fn(): bool => false); + + return $result === false ? 'TIMEOUT' : 'MET'; + } + } + ); + + $this->worker->runWorkflow('AwaitPromiseWorkflow'); + $this->worker->assertWorkflowReturns('TIMEOUT'); + $this->factory->run($this->worker); + } + + public function testClosureTrueUnblocks(): void + { + $this->addToAssertionCount(1); + $this->worker->registerWorkflowObject( + new + #[Workflow\WorkflowInterface] + class { + #[WorkflowMethod(name: 'AwaitPromiseWorkflow')] + public function handler(): iterable + { + $result = yield Workflow::awaitWithTimeout(5, static fn(): bool => true); + + return $result === true ? 'MET' : 'TIMEOUT'; + } + } + ); + + $this->worker->runWorkflow('AwaitPromiseWorkflow'); + $this->worker->assertWorkflowReturns('MET'); + $this->factory->run($this->worker); + } + + public function testFulfilledPromiseUnblocks(): void + { + $this->addToAssertionCount(1); + $this->worker->registerWorkflowObject( + new + #[Workflow\WorkflowInterface] + class { + #[WorkflowMethod(name: 'AwaitPromiseWorkflow')] + public function handler(): iterable + { + $result = yield Workflow::awaitWithTimeout(5, resolve(true)); + + return $result === true ? 'MET' : 'TIMEOUT'; + } + } + ); + + $this->worker->runWorkflow('AwaitPromiseWorkflow'); + $this->worker->assertWorkflowReturns('MET'); + $this->factory->run($this->worker); + } + + public function testSingleRejectedPromisePropagates(): void + { + $this->addToAssertionCount(1); + $this->worker->registerWorkflowObject( + new + #[Workflow\WorkflowInterface] + class { + #[WorkflowMethod(name: 'AwaitPromiseWorkflow')] + public function handler(): iterable + { + try { + yield Workflow::await(reject(new \RuntimeException('boom'))); + } catch (\Throwable) { + return 'THREW'; + } + + return 'NO_THROW'; + } + } + ); + + $this->worker->runWorkflow('AwaitPromiseWorkflow'); + $this->worker->assertWorkflowReturns('THREW'); + $this->factory->run($this->worker); + } + + public function testEmptyAwaitFailsFast(): void + { + $this->addToAssertionCount(1); + $this->worker->registerWorkflowObject( + new + #[Workflow\WorkflowInterface] + class { + #[WorkflowMethod(name: 'AwaitPromiseWorkflow')] + public function handler(): iterable + { + try { + yield Workflow::await(); + } catch (\Throwable) { + return 'THREW'; + } + + return 'NO_THROW'; + } + } + ); + + $this->worker->runWorkflow('AwaitPromiseWorkflow'); + $this->worker->assertWorkflowReturns('THREW'); + $this->factory->run($this->worker); + } + + public function testRejectedConditionPropagatesWhenFlagEnabled(): void + { + FeatureFlags::$settleAwaitOnFirstSettledCondition = true; + $this->addToAssertionCount(1); + $this->registerRejectingAwaitWithTimeoutWorkflow(); + + $this->worker->runWorkflow('AwaitPromiseWorkflow'); + $this->worker->assertWorkflowReturns('THREW'); + $this->factory->run($this->worker); + } + + public function testRejectedConditionIsIgnoredWhenFlagDisabled(): void + { + FeatureFlags::$settleAwaitOnFirstSettledCondition = false; + $this->addToAssertionCount(1); + $this->registerRejectingAwaitWithTimeoutWorkflow(); + + $this->worker->runWorkflow('AwaitPromiseWorkflow'); + $this->worker->assertWorkflowReturns('RESULT:false'); + $this->factory->run($this->worker); + } + + public function testMultiConditionRejectPropagatesWhenFlagEnabled(): void + { + FeatureFlags::$settleAwaitOnFirstSettledCondition = true; + $this->addToAssertionCount(1); + $this->worker->registerWorkflowObject( + new + #[Workflow\WorkflowInterface] + class { + #[WorkflowMethod(name: 'AwaitPromiseWorkflow')] + public function handler(): iterable + { + try { + yield Workflow::awaitWithTimeout( + 5, + reject(new \RuntimeException('boom')), + static fn(): bool => false, + ); + } catch (\Throwable) { + return 'THREW'; + } + + return 'NO_THROW'; + } + } + ); + + $this->worker->runWorkflow('AwaitPromiseWorkflow'); + $this->worker->assertWorkflowReturns('THREW'); + $this->factory->run($this->worker); + } + + private function registerRejectingAwaitWithTimeoutWorkflow(): void + { + $this->worker->registerWorkflowObject( + new + #[Workflow\WorkflowInterface] + class { + #[WorkflowMethod(name: 'AwaitPromiseWorkflow')] + public function handler(): iterable + { + try { + $result = yield Workflow::awaitWithTimeout(5, reject(new \RuntimeException('boom'))); + } catch (\Throwable) { + return 'THREW'; + } + + return 'RESULT:' . \var_export($result, true); + } + } + ); + } +}