feat: allow UseExisting conflicts policy

Author: xepozz

src/Client/WorkflowOptions.php

@@ -26,6 +26,7 @@
 use Temporal\Common\Versioning\VersioningOverride;
 use Temporal\Common\WorkflowIdConflictPolicy;
 use Temporal\DataConverter\DataConverterInterface;
+use Temporal\Internal\Client\OnConflictOptions;
 use Temporal\Internal\Marshaller\Meta\Marshal;
 use Temporal\Internal\Marshaller\Type\ArrayType;
 use Temporal\Internal\Marshaller\Type\CronType;
@@ -193,6 +194,9 @@ final class WorkflowOptions extends Options
     #[Marshal(name: 'VersioningOverride')]
     public ?VersioningOverride $versioningOverride = null;
 
+    /** @internal */
+    public ?OnConflictOptions $onConflictOptions = null;
+
     /**
      * @throws \Exception
      */
@@ -612,4 +616,16 @@ public function withPriority(Priority $priority): self
         $self->priority = $priority;
         return $self;
     }
+
+    /**
+     * @internal
+     * @return $this
+     */
+    #[Pure]
+    public function withOnConflictOptionsInternal(?OnConflictOptions $options): self
+    {
+        $self = clone $this;
+        $self->onConflictOptions = $options;
+        return $self;
+    }
 }

src/Internal/Client/OnConflictOptions.php

@@ -0,0 +1,28 @@
+<?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\Client;
+
+/**
+ * Workflow-id conflict policy options: what to attach to the existing run when
+ * a start request conflicts. Serialized to {@see \Temporal\Api\Workflow\V1\OnConflictOptions}.
+ *
+ * @internal
+ * @psalm-immutable
+ */
+final class OnConflictOptions
+{
+    public function __construct(
+        public readonly bool $attachRequestId = true,
+        public readonly bool $attachCompletionCallbacks = true,
+        public readonly bool $attachLinks = true,
+    ) {}
+}

src/Internal/Client/WorkflowStarter.php

@@ -12,6 +12,7 @@
 namespace Temporal\Internal\Client;
 
 use Temporal\Api\Common\V1\WorkflowType;
+use Temporal\Api\Workflow\V1\OnConflictOptions as OnConflictOptionsProto;
 use Temporal\Api\Deployment\V1\WorkerDeploymentVersion;
 use Temporal\Api\Errordetails\V1\MultiOperationExecutionFailure;
 use Temporal\Api\Errordetails\V1\WorkflowExecutionAlreadyStartedFailure;
@@ -32,6 +33,7 @@
 use Temporal\Client\Update\UpdateHandle;
 use Temporal\Client\Update\UpdateOptions;
 use Temporal\Client\WorkflowOptions;
+use Temporal\Common\WorkflowIdConflictPolicy;
 use Temporal\Common\Uuid;
 use Temporal\Common\Versioning\VersioningBehavior;
 use Temporal\DataConverter\DataConverterInterface;
@@ -47,6 +49,7 @@
 use Temporal\Interceptor\WorkflowClient\UpdateWithStartInput;
 use Temporal\Interceptor\WorkflowClient\UpdateWithStartOutput;
 use Temporal\Interceptor\WorkflowClientCallsInterceptor;
+use Temporal\Internal\Client\OnConflictOptions;
 use Temporal\Internal\Interceptor\Pipeline;
 use Temporal\Internal\Support\DateInterval;
 use Temporal\Workflow\WorkflowExecution;
@@ -276,6 +279,15 @@ function (UpdateWithStartInput $input): UpdateWithStartOutput {
         );
     }
 
+    private static function onConflictOptionsToProto(OnConflictOptions $options): OnConflictOptionsProto
+    {
+        $proto = new OnConflictOptionsProto();
+        $proto->setAttachRequestId($options->attachRequestId);
+        $proto->setAttachCompletionCallbacks($options->attachCompletionCallbacks);
+        $proto->setAttachLinks($options->attachLinks);
+        return $proto;
+    }
+
     /**
      * @param StartWorkflowExecutionRequest|SignalWithStartWorkflowExecutionRequest $request
      *        use {@see configureExecutionRequest()} to prepare request
@@ -296,6 +308,10 @@ private function executeRequest(
             \assert($f instanceof WorkflowExecutionAlreadyStartedFailure);
             $execution = new WorkflowExecution($request->getWorkflowId(), $f->getRunId());
 
+            if ($request->getWorkflowIdConflictPolicy() === WorkflowIdConflictPolicy::UseExisting->value) {
+                return $execution;
+            }
+
             throw new WorkflowExecutionAlreadyStartedException(
                 $execution,
                 $request->getWorkflowType()->getName(),
@@ -395,6 +411,10 @@ private function configureExecutionRequest(
 
         if ($req instanceof StartWorkflowExecutionRequest) {
             $req->setRequestEagerExecution($options->eagerStart);
+
+            if ($options->onConflictOptions !== null) {
+                $req->setOnConflictOptions(self::onConflictOptionsToProto($options->onConflictOptions));
+            }
         }
 
         if (!$input->arguments->isEmpty()) {

tests/Acceptance/Extra/Update/UpdateWithStartTest.php

@@ -10,6 +10,7 @@
 use Temporal\Client\Update\UpdateOptions;
 use Temporal\Client\WorkflowClientInterface;
 use Temporal\Client\WorkflowOptions;
+use Temporal\Common\WorkflowIdConflictPolicy;
 use Temporal\Exception\Client\WorkflowExecutionAlreadyStartedException;
 use Temporal\Exception\Client\WorkflowFailedException;
 use Temporal\Exception\Client\WorkflowServiceException;
@@ -138,6 +139,96 @@ public function failOnReuseExistingWorkflowId(
             $stub1->signal('exit');
         }
     }
+
+    #[Test]
+    public function useExistingReturnsRunningExecution(
+        WorkflowClientInterface $client,
+        Feature $feature,
+    ): void {
+        $id = Uuid::uuid7()->__toString();
+
+        $first = $client->newUntypedWorkflowStub(
+            'Extra_Update_UseExisting',
+            WorkflowOptions::new()->withTaskQueue($feature->taskQueue)->withWorkflowId($id),
+        );
+        $client->start($first);
+        $firstRunId = $first->getExecution()->getRunID();
+
+        $second = $client->newUntypedWorkflowStub(
+            'Extra_Update_UseExisting',
+            WorkflowOptions::new()
+                ->withTaskQueue($feature->taskQueue)
+                ->withWorkflowId($id)
+                ->withWorkflowIdConflictPolicy(WorkflowIdConflictPolicy::UseExisting),
+        );
+
+        try {
+            $client->start($second);
+
+            $this->assertSame($id, $second->getExecution()->getID());
+            $this->assertSame(
+                $firstRunId,
+                $second->getExecution()->getRunID(),
+                'UseExisting must resolve to the already-running execution instead of throwing',
+            );
+        } finally {
+            $first->signal('exit');
+        }
+    }
+
+    #[Test]
+    public function failPolicyThrowsOnRunningWorkflowId(
+        WorkflowClientInterface $client,
+        Feature $feature,
+    ): void {
+        $id = Uuid::uuid7()->__toString();
+
+        $first = $client->newUntypedWorkflowStub(
+            'Extra_Update_UseExisting',
+            WorkflowOptions::new()->withTaskQueue($feature->taskQueue)->withWorkflowId($id),
+        );
+        $client->start($first);
+
+        $second = $client->newUntypedWorkflowStub(
+            'Extra_Update_UseExisting',
+            WorkflowOptions::new()
+                ->withTaskQueue($feature->taskQueue)
+                ->withWorkflowId($id)
+                ->withWorkflowIdConflictPolicy(WorkflowIdConflictPolicy::Fail),
+        );
+
+        try {
+            $this->expectException(WorkflowExecutionAlreadyStartedException::class);
+            $client->start($second);
+        } finally {
+            $first->signal('exit');
+        }
+    }
+
+    #[Test]
+    public function useExistingStartsFreshWhenNoneRunning(
+        WorkflowClientInterface $client,
+        Feature $feature,
+    ): void {
+        $id = Uuid::uuid7()->__toString();
+
+        $stub = $client->newUntypedWorkflowStub(
+            'Extra_Update_UseExisting',
+            WorkflowOptions::new()
+                ->withTaskQueue($feature->taskQueue)
+                ->withWorkflowId($id)
+                ->withWorkflowIdConflictPolicy(WorkflowIdConflictPolicy::UseExisting),
+        );
+
+        try {
+            $client->start($stub);
+
+            $this->assertSame($id, $stub->getExecution()->getID());
+            $this->assertNotSame('', $stub->getExecution()->getRunID());
+        } finally {
+            $stub->signal('exit');
+        }
+    }
 }
 
 #[WorkflowInterface]
@@ -195,3 +286,22 @@ public function __construct(
         public int $length = 0,
     ) {}
 }
+
+#[WorkflowInterface]
+class UseExistingWorkflow
+{
+    private bool $exit = false;
+
+    #[WorkflowMethod(name: 'Extra_Update_UseExisting')]
+    public function handle(): \Generator
+    {
+        yield Workflow::await(fn(): bool => $this->exit);
+        return 'done';
+    }
+
+    #[Workflow\SignalMethod]
+    public function exit(): void
+    {
+        $this->exit = true;
+    }
+}

tests/Unit/Internal/Client/OnConflictOptionsTestCase.php

@@ -0,0 +1,42 @@
+<?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\Tests\Unit\Internal\Client;
+
+use PHPUnit\Framework\Attributes\CoversClass;
+use PHPUnit\Framework\TestCase;
+use Temporal\Internal\Client\OnConflictOptions;
+
+#[CoversClass(OnConflictOptions::class)]
+final class OnConflictOptionsTestCase extends TestCase
+{
+    public function testDefaultsAreAllTrue(): void
+    {
+        $options = new OnConflictOptions();
+
+        self::assertTrue($options->attachRequestId);
+        self::assertTrue($options->attachCompletionCallbacks);
+        self::assertTrue($options->attachLinks);
+    }
+
+    public function testAcceptsExplicitFlags(): void
+    {
+        $options = new OnConflictOptions(
+            attachRequestId: false,
+            attachCompletionCallbacks: false,
+            attachLinks: false,
+        );
+
+        self::assertFalse($options->attachRequestId);
+        self::assertFalse($options->attachCompletionCallbacks);
+        self::assertFalse($options->attachLinks);
+    }
+}