diff --git a/src/Internal/Declaration/Prototype/WorkflowCollection.php b/src/Internal/Declaration/Prototype/WorkflowCollection.php index 7c31d603d..0d2c4c69a 100644 --- a/src/Internal/Declaration/Prototype/WorkflowCollection.php +++ b/src/Internal/Declaration/Prototype/WorkflowCollection.php @@ -12,8 +12,37 @@ namespace Temporal\Internal\Declaration\Prototype; use Temporal\Internal\Repository\ArrayRepository; +use Temporal\Internal\Repository\Identifiable; /** * @template-extends ArrayRepository */ -final class WorkflowCollection extends ArrayRepository {} +final class WorkflowCollection extends ArrayRepository +{ + /** + * A dynamic (catch-all) workflow is the single fallback used when no + * statically registered workflow matches the requested type. As in the + * other SDKs (Go panics, Python raises), at most one may be registered per + * worker — a second would make dispatch ambiguous. + */ + public function add(Identifiable $entry, bool $overwrite = false): void + { + if ($entry instanceof WorkflowPrototype && $entry->isDynamic()) { + foreach ($this as $existing) { + if ($existing instanceof WorkflowPrototype + && $existing->isDynamic() + && $existing->getID() !== $entry->getID() + ) { + throw new \LogicException(\sprintf( + 'Cannot register dynamic workflow "%s": a dynamic (catch-all) workflow "%s" is ' + . 'already registered. At most one dynamic workflow is allowed per worker.', + $entry->getID(), + $existing->getID(), + )); + } + } + } + + parent::add($entry, $overwrite); + } +} diff --git a/src/Internal/Declaration/Prototype/WorkflowPrototype.php b/src/Internal/Declaration/Prototype/WorkflowPrototype.php index cf47d7fae..0934dcee8 100644 --- a/src/Internal/Declaration/Prototype/WorkflowPrototype.php +++ b/src/Internal/Declaration/Prototype/WorkflowPrototype.php @@ -44,6 +44,7 @@ final class WorkflowPrototype extends Prototype private ?MethodRetry $methodRetry = null; private ?ReturnType $returnType = null; private bool $hasInitializer = false; + private bool $dynamic = false; private VersioningBehavior $versioningBehavior; /** @@ -72,6 +73,20 @@ public function setHasInitializer(bool $hasInitializer): void $this->hasInitializer = $hasInitializer; } + /** + * Whether this is the dynamic (catch-all) workflow, invoked when no + * statically registered workflow matches the requested type name. + */ + public function isDynamic(): bool + { + return $this->dynamic; + } + + public function setDynamic(bool $dynamic): void + { + $this->dynamic = $dynamic; + } + public function getCronSchedule(): ?CronSchedule { return $this->cronSchedule; diff --git a/src/Internal/Declaration/Reader/WorkflowReader.php b/src/Internal/Declaration/Reader/WorkflowReader.php index d97877a22..55dc116eb 100644 --- a/src/Internal/Declaration/Reader/WorkflowReader.php +++ b/src/Internal/Declaration/Reader/WorkflowReader.php @@ -432,6 +432,9 @@ private function findProto( $name = $info->name ?? $interface->getShortName(); - return new WorkflowPrototype($name, $handler, $class); + $prototype = new WorkflowPrototype($name, $handler, $class); + $prototype->setDynamic($info->dynamic); + + return $prototype; } } diff --git a/src/Internal/Transport/Router/GetWorkerInfo.php b/src/Internal/Transport/Router/GetWorkerInfo.php index 596c214d4..ac435962e 100644 --- a/src/Internal/Transport/Router/GetWorkerInfo.php +++ b/src/Internal/Transport/Router/GetWorkerInfo.php @@ -51,6 +51,9 @@ private function workerToArray(WorkerInterface $worker): array 'queries' => \array_keys($workflow->getQueryHandlers()), 'signals' => \array_keys($workflow->getSignalHandlers()), 'versioning_behavior' => $workflow->getVersioningBehavior()->value, + // Advertise the dynamic (catch-all) workflow so the RoadRunner + // temporal plugin registers a Go dynamic-workflow proxy for it. + 'dynamic' => $workflow->isDynamic(), ]; $activityMap = static fn(ActivityPrototype $activity): array => [ diff --git a/src/Internal/Transport/Router/StartWorkflow.php b/src/Internal/Transport/Router/StartWorkflow.php index 14e12083e..3d0cc0d70 100644 --- a/src/Internal/Transport/Router/StartWorkflow.php +++ b/src/Internal/Transport/Router/StartWorkflow.php @@ -91,7 +91,21 @@ public function handle(ServerRequestInterface $request, array $headers, Deferred private function findWorkflowOrFail(WorkflowInfo $info): WorkflowPrototype { - return $this->services->workflows->find($info->type->name) ?? throw new \OutOfRangeException( + $found = $this->services->workflows->find($info->type->name); + + if ($found instanceof WorkflowPrototype) { + return $found; + } + + // Fall back to the dynamic (catch-all) workflow, if one is registered. + // The handler reads the real type name via Workflow::getInfo()->type. + foreach ($this->services->workflows as $prototype) { + if ($prototype instanceof WorkflowPrototype && $prototype->isDynamic()) { + return $prototype; + } + } + + throw new \OutOfRangeException( \sprintf(self::ERROR_NOT_FOUND, $info->type->name), ); } diff --git a/src/Internal/Workflow/Process/Process.php b/src/Internal/Workflow/Process/Process.php index d1a0a234a..16c7585b0 100644 --- a/src/Internal/Workflow/Process/Process.php +++ b/src/Internal/Workflow/Process/Process.php @@ -186,6 +186,7 @@ public function initAndStart( bool $deferred, ): void { $handler = $instance->getHandler(); + $dynamic = $instance->getPrototype()->isDynamic(); $instance = $context->getWorkflowInstance(); $arguments = null; $values = []; @@ -220,9 +221,13 @@ public function initAndStart( $this->services->interceptorProvider ->getPipeline(WorkflowInboundCallsInterceptor::class) ->with( - function (WorkflowInput $input) use ($context, $arguments, $handler, $deferred): void { + function (WorkflowInput $input) use ($context, $arguments, $handler, $deferred, $dynamic): void { // Prepare typed input if values have been changed - if ($arguments === null || $input->arguments !== $context->getInput()) { + if ($dynamic) { + // Dynamic workflows receive the untouched argument collection so + // they can interpret types that are unknown at registration time. + $arguments = EncodedValues::fromValues([$input->arguments]); + } elseif ($arguments === null || $input->arguments !== $context->getInput()) { $arguments = EncodedValues::fromValues($handler->resolveArguments($input->arguments)); } diff --git a/src/Workflow/WorkflowMethod.php b/src/Workflow/WorkflowMethod.php index 4195f84b4..12e07d585 100644 --- a/src/Workflow/WorkflowMethod.php +++ b/src/Workflow/WorkflowMethod.php @@ -35,11 +35,23 @@ final class WorkflowMethod #[Immutable] public ?string $name = null; + /** + * Marks this as a dynamic (catch-all) workflow: it is invoked when the + * worker receives a workflow whose type name is not statically registered. + * At most one dynamic workflow may be registered per worker. The handler + * reads the actual type name via {@see \Temporal\Workflow::getInfo()} and + * receives the raw arguments (declare a {@see \Temporal\DataConverter\ValuesInterface} + * parameter to access them). + */ + #[Immutable] + public bool $dynamic = false; + /** * @param non-empty-string|null $name */ - public function __construct(?string $name = null) + public function __construct(?string $name = null, bool $dynamic = false) { $this->name = $name; + $this->dynamic = $dynamic; } } diff --git a/tests/Unit/Declaration/Fixture/WorkflowWithAnotherDynamic.php b/tests/Unit/Declaration/Fixture/WorkflowWithAnotherDynamic.php new file mode 100644 index 000000000..fc5c390af --- /dev/null +++ b/tests/Unit/Declaration/Fixture/WorkflowWithAnotherDynamic.php @@ -0,0 +1,26 @@ +assertNull($prototype->getHandler()); } + /** + * @param WorkflowReader $reader + * @throws \ReflectionException + */ + #[TestDox("Reading a dynamic (catch-all) workflow sets the dynamic flag")] + #[DataProvider('workflowReaderDataProvider')] + public function testDynamicWorkflow(WorkflowReader $reader): void + { + $this->assertTrue($reader->fromClass(WorkflowWithDynamic::class)->isDynamic()); + $this->assertFalse($reader->fromClass(SimpleWorkflow::class)->isDynamic()); + } + + /** + * @param WorkflowReader $reader + * @throws \ReflectionException + */ + #[TestDox("At most one dynamic workflow may be registered per worker")] + #[DataProvider('workflowReaderDataProvider')] + public function testMultipleDynamicWorkflowsAreRejected(WorkflowReader $reader): void + { + $collection = new WorkflowCollection(); + $collection->add($reader->fromClass(WorkflowWithDynamic::class)); + + $this->expectException(\LogicException::class); + $collection->add($reader->fromClass(WorkflowWithAnotherDynamic::class)); + } + /** * @param WorkflowReader $reader * @throws \ReflectionException diff --git a/tests/Unit/Framework/WorkerTestCase.php b/tests/Unit/Framework/WorkerTestCase.php index 63e2e51f6..353dabeba 100644 --- a/tests/Unit/Framework/WorkerTestCase.php +++ b/tests/Unit/Framework/WorkerTestCase.php @@ -4,6 +4,7 @@ namespace Temporal\Tests\Unit\Framework; +use Temporal\DataConverter\ValuesInterface; use Temporal\Tests\Unit\AbstractUnit; use Temporal\Worker\WorkerFactoryInterface; use Temporal\Worker\WorkerInterface; @@ -18,17 +19,10 @@ final class WorkerTestCase extends AbstractUnit { private WorkerFactoryInterface $factory; + /** @var WorkerMock|WorkerInterface */ private $worker; - protected function setUp(): void - { - $this->factory = WorkerFactoryMock::create(); - $this->worker = $this->factory->newWorker(); - - parent::setUp(); - } - public function testRunWorker(): void { $this->worker->registerWorkflowObject( @@ -38,15 +32,104 @@ class { #[WorkflowMethod(name: 'SimpleWorkflow')] public function handler(): iterable { - $result = yield Workflow::awaitWithTimeout(5, fn() => false); + $result = yield Workflow::awaitWithTimeout(5, static fn() => false); assertFalse($result); return $result; } - } + }, ); $this->worker->runWorkflow('SimpleWorkflow'); $this->worker->expectTimer(5); $this->factory->run($this->worker); } + + public function testDynamicWorkflowReceivesActualTypeAndRawArguments(): void + { + $this->worker->registerWorkflowObject( + new + #[Workflow\WorkflowInterface] + class { + #[WorkflowMethod(name: 'DynamicWorkflow', dynamic: true)] + public function handler(ValuesInterface $arguments): array + { + return [ + Workflow::getInfo()->type->name, + $arguments->getValues(), + ]; + } + }, + ); + + $this->worker->runWorkflow('RuntimeDefinedWorkflow', 'alpha', 42); + $this->worker->assertWorkflowReturns(['RuntimeDefinedWorkflow', ['alpha', 42]]); + + $this->factory->run($this->worker); + self::assertCount(1, $this->worker->getWorkflows()); + } + + public function testNamedWorkflowTakesPrecedenceOverDynamicWorkflow(): void + { + $this->worker->registerWorkflowObject( + new + #[Workflow\WorkflowInterface] + class { + #[WorkflowMethod(name: 'DynamicWorkflow', dynamic: true)] + public function handler(ValuesInterface $arguments): string + { + return 'dynamic'; + } + }, + ); + $this->worker->registerWorkflowObject( + new + #[Workflow\WorkflowInterface] + class { + #[WorkflowMethod(name: 'NamedWorkflow')] + public function handler(): string + { + return 'named'; + } + }, + ); + + $this->worker->runWorkflow('NamedWorkflow'); + $this->worker->assertWorkflowReturns('named'); + + $this->factory->run($this->worker); + self::assertCount(2, $this->worker->getWorkflows()); + } + + public function testEachWorkerCanRegisterOneDynamicWorkflow(): void + { + $secondWorker = $this->factory->newWorker('other-task-queue'); + + $this->worker->registerWorkflowObject( + new + #[Workflow\WorkflowInterface] + class { + #[WorkflowMethod(name: 'DynamicWorkflow', dynamic: true)] + public function handler(): void {} + }, + ); + $secondWorker->registerWorkflowTypes(PerWorkerDynamicWorkflow::class); + + self::assertCount(1, $this->worker->getWorkflows()); + self::assertCount(1, $secondWorker->getWorkflows()); + } + + protected function setUp(): void + { + $this->factory = WorkerFactoryMock::create(); + $this->worker = $this->factory->newWorker(); + + parent::setUp(); + } +} + +#[Workflow\WorkflowInterface] +final class PerWorkerDynamicWorkflow +{ + #[WorkflowMethod(name: 'DynamicWorkflow', dynamic: true)] + public function handler(): void {} } diff --git a/tests/Unit/Router/GetWorkerInfoTestCase.php b/tests/Unit/Router/GetWorkerInfoTestCase.php new file mode 100644 index 000000000..159c21064 --- /dev/null +++ b/tests/Unit/Router/GetWorkerInfoTestCase.php @@ -0,0 +1,63 @@ +createMock(WorkerInterface::class); + $options = WorkerOptions::new(); + + $worker->method('getID')->willReturn('test-queue'); + $worker->method('getOptions')->willReturn($options); + $worker->method('getWorkflows')->willReturn([ + $reader->fromClass(WorkflowWithDynamic::class), + $reader->fromClass(SimpleWorkflow::class), + ]); + $worker->method('getActivities')->willReturn([]); + + $marshaller = $this->createMock(MarshallerInterface::class); + $marshaller->expects($this->once()) + ->method('marshal') + ->with($options) + ->willReturn([]); + + $router = new GetWorkerInfo( + new ArrayRepository([$worker]), + $marshaller, + ServiceCredentials::create(), + new PluginRegistry(), + ); + $resolver = new Deferred(); + $response = null; + $resolver->promise()->then(static function (EncodedValues $values) use (&$response): void { + $response = $values->getValues(); + }); + + $router->handle(new Request(), [], $resolver); + + self::assertNotNull($response); + self::assertSame(true, $response[0]['Workflows'][0]['dynamic']); + self::assertSame(false, $response[0]['Workflows'][1]['dynamic']); + } +} diff --git a/tests/Unit/Router/StartWorkflowTestCase.php b/tests/Unit/Router/StartWorkflowTestCase.php index a42f54c8c..0a49520a6 100644 --- a/tests/Unit/Router/StartWorkflowTestCase.php +++ b/tests/Unit/Router/StartWorkflowTestCase.php @@ -86,6 +86,24 @@ public function testStartingAlreadyRunningWorkflow(): void $this->router->handle($request, [], new Deferred()); } + public function testUnknownWorkflowWithoutDynamicWorkflowThrows(): void + { + $request = new Request($runId = Uuid::v4(), 'UnknownWorkflow', EncodedValues::fromValues([])); + + $workflowInfo = new WorkflowInfo(); + $workflowInfo->type->name = 'UnknownWorkflow'; + $workflowInfo->execution = new WorkflowExecution('123', $runId); + + $this->marshaller->expects($this->once()) + ->method('unmarshal') + ->willReturn(new Input($workflowInfo)); + + $this->expectException(\OutOfRangeException::class); + $this->expectExceptionMessage('Workflow with the specified name "UnknownWorkflow" was not registered'); + + $this->router->handle($request, [], new Deferred()); + } + protected function setUp(): void { $workflow = new \stdClass();