fix: settle await on first settled condition incl. rejection (#399)

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

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;
 }

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
     {

tests/Unit/WorkflowContext/AwaitPromiseSettlementTestCase.php

@@ -0,0 +1,233 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Temporal\Tests\Unit\WorkflowContext;
+
+use Temporal\Tests\Unit\Framework\WorkerFactoryMock;
+use Temporal\Tests\Unit\Framework\WorkerMock;
+use Temporal\Tests\Unit\AbstractUnit;
+use Temporal\Worker\FeatureFlags;
+use Temporal\Worker\WorkerFactoryInterface;
+use Temporal\Worker\WorkerInterface;
+use Temporal\Workflow;
+use Temporal\Workflow\WorkflowMethod;
+
+use function React\Promise\reject;
+use function React\Promise\resolve;
+
+final class AwaitPromiseSettlementTestCase extends AbstractUnit
+{
+    private WorkerFactoryInterface $factory;
+    /** @var WorkerMock|WorkerInterface */
+    private $worker;
+    private bool $flagBackup;
+
+    protected function setUp(): void
+    {
+        $this->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);
+                }
+            }
+        );
+    }
+}