From 1d584f51112c2fb5634d43af7186fef1017918d5 Mon Sep 17 00:00:00 2001 From: Dmitriy Derepko Date: Tue, 18 Aug 2026 20:46:37 +0400 Subject: [PATCH 1/9] feat(php): add the nexus sync_success feature Register Nexus service implementations discovered in a feature directory, pass the runner-created endpoint through to the check, and implement the feature to the same contract as the Go and Java versions. --- features/nexus/sync_success/feature.php | 98 +++++++++++++++++++++++++ harness/php/composer.json | 16 +++- harness/php/src/Input/Command.php | 7 +- harness/php/src/Input/Feature.php | 1 + harness/php/src/Runtime/Feature.php | 4 + harness/php/src/Runtime/State.php | 27 ++++++- harness/php/src/RuntimeBuilder.php | 20 +++++ harness/php/worker.php | 5 ++ 8 files changed, 172 insertions(+), 6 deletions(-) create mode 100644 features/nexus/sync_success/feature.php diff --git a/features/nexus/sync_success/feature.php b/features/nexus/sync_success/feature.php new file mode 100644 index 00000000..773a1523 --- /dev/null +++ b/features/nexus/sync_success/feature.php @@ -0,0 +1,98 @@ +withEndpoint($endpoint) + ->withScheduleToCloseTimeout('1 minute'), + ); + + return yield $service->sayHello('world'); + } +} + +class FeatureChecker +{ + #[Check] + public static function check(WorkflowClientInterface $client, Feature $feature): void + { + Assert::notNull($feature->nexusEndpoint, 'Nexus endpoint is not provided by the runner'); + + $stub = $client->newUntypedWorkflowStub( + 'Harness_Nexus_SyncSuccess', + WorkflowOptions::new()->withTaskQueue($feature->taskQueue), + ); + $client->start($stub, $feature->nexusEndpoint); + + Assert::same($stub->getResult('string'), 'Hello, world!'); + + $events = \iterator_to_array($client->getWorkflowHistory($stub->getExecution())->getEvents(), false); + + Assert::true( + self::hasEvent($events, static fn(HistoryEvent $e): bool => $e->hasNexusOperationScheduledEventAttributes()), + 'NexusOperationScheduled event is missing', + ); + Assert::true( + self::hasEvent($events, static fn(HistoryEvent $e): bool => $e->hasNexusOperationCompletedEventAttributes()), + 'NexusOperationCompleted event is missing', + ); + Assert::false( + self::hasEvent($events, static fn(HistoryEvent $e): bool => $e->hasNexusOperationStartedEventAttributes()), + 'Synchronous operation must not produce a NexusOperationStarted event', + ); + } + + /** + * @param list $events + * @param callable(HistoryEvent): bool $predicate + */ + private static function hasEvent(array $events, callable $predicate): bool + { + foreach ($events as $event) { + if ($predicate($event)) { + return true; + } + } + + return false; + } +} diff --git a/harness/php/composer.json b/harness/php/composer.json index f2ddb5a7..a36d74be 100644 --- a/harness/php/composer.json +++ b/harness/php/composer.json @@ -2,14 +2,18 @@ "name": "temporal/harness", "type": "project", "description": "Temporal SDK Harness", - "keywords": ["temporal", "sdk", "harness"], + "keywords": [ + "temporal", + "sdk", + "harness" + ], "license": "MIT", "require": { "php": "^8.2", "buggregator/trap": "^1.9", "spiral/core": "^3.13", "symfony/process": ">=6.4", - "temporal/sdk": "^2.16.0", + "temporal/sdk": "dev-nexus-new as 2.18.0", "webmozart/assert": "^1.11" }, "autoload": { @@ -21,5 +25,11 @@ "rr-get": "rr get" }, "prefer-stable": true, - "minimum-stability": "dev" + "minimum-stability": "dev", + "repositories": [ + { + "type": "vcs", + "url": "https://github.com/temporalio/sdk-php" + } + ] } diff --git a/harness/php/src/Input/Command.php b/harness/php/src/Input/Command.php index 3fab194e..772cc5fb 100644 --- a/harness/php/src/Input/Command.php +++ b/harness/php/src/Input/Command.php @@ -67,11 +67,12 @@ public static function fromCommandLine(array $argv): self continue; } - [$dir, $taskQueue] = \explode(':', $chunk, 2); + [$dir, $taskQueue, $nexusEndpoint] = \array_pad(\explode(':', $chunk, 3), 3, null); $self->features[] = new Feature( dir: $dir, namespace: 'Harness\\Feature\\' . self::namespaceFromPath($dir), taskQueue: $taskQueue, + nexusEndpoint: $nexusEndpoint, ); } @@ -91,7 +92,9 @@ public function toCommandLineArguments(): array $this->tlsServerName === null or $result[] = "tls.server-name=$this->tlsServerName"; $this->tlsCaCert === null or $result[] = "tls.ca-cert=$this->tlsCaCert"; foreach ($this->features as $feature) { - $result[] = "{$feature->dir}:{$feature->taskQueue}"; + $result[] = $feature->nexusEndpoint === null + ? "{$feature->dir}:{$feature->taskQueue}" + : "{$feature->dir}:{$feature->taskQueue}:{$feature->nexusEndpoint}"; } return $result; diff --git a/harness/php/src/Input/Feature.php b/harness/php/src/Input/Feature.php index b3cc2bff..5758734c 100644 --- a/harness/php/src/Input/Feature.php +++ b/harness/php/src/Input/Feature.php @@ -10,6 +10,7 @@ public function __construct( public string $dir, public string $namespace, public string $taskQueue, + public ?string $nexusEndpoint = null, ) { } } diff --git a/harness/php/src/Runtime/Feature.php b/harness/php/src/Runtime/Feature.php index f916ae4b..9f88b9b3 100644 --- a/harness/php/src/Runtime/Feature.php +++ b/harness/php/src/Runtime/Feature.php @@ -20,8 +20,12 @@ final class Feature /** @var list> Lazy callables */ public array $converters = []; + /** @var list Nexus service implementations */ + public array $nexusServices = []; + public function __construct( public readonly string $taskQueue, + public readonly ?string $nexusEndpoint = null, ) { } } diff --git a/harness/php/src/Runtime/State.php b/harness/php/src/Runtime/State.php index 122aabba..5d6dc156 100644 --- a/harness/php/src/Runtime/State.php +++ b/harness/php/src/Runtime/State.php @@ -59,6 +59,20 @@ public function activities(): \Traversable } } + /** + * Iterate over all the Nexus service implementations. + * + * @return \Traversable + */ + public function nexusServices(): \Traversable + { + foreach ($this->features as $feature) { + foreach ($feature->nexusServices as $service) { + yield $feature => $service; + } + } + } + /** * Iterate over all the Payload Converters. * @@ -120,8 +134,19 @@ public function addActivity(\Harness\Input\Feature $inputFeature, string $class) $this->getFeature($inputFeature)->activities[] = $class; } + /** + * @param class-string $class + */ + public function addNexusService(\Harness\Input\Feature $inputFeature, string $class): void + { + $this->getFeature($inputFeature)->nexusServices[] = $class; + } + private function getFeature(\Harness\Input\Feature $feature): Feature { - return $this->features[$feature->namespace] ??= new Feature($feature->taskQueue); + return $this->features[$feature->namespace] ??= new Feature( + $feature->taskQueue, + $feature->nexusEndpoint, + ); } } \ No newline at end of file diff --git a/harness/php/src/RuntimeBuilder.php b/harness/php/src/RuntimeBuilder.php index cc7e3da3..f0ac1f45 100644 --- a/harness/php/src/RuntimeBuilder.php +++ b/harness/php/src/RuntimeBuilder.php @@ -9,6 +9,7 @@ use Harness\Input\Feature; use Harness\Runtime\State; use Temporal\Activity\ActivityInterface; +use Temporal\Nexus\Attribute\Service; use Temporal\DataConverter\PayloadConverterInterface; use Temporal\Workflow\WorkflowInterface; @@ -29,6 +30,10 @@ public static function createState(array $argv, string $workDir): State $class->getAttributes(ActivityInterface::class) === [] or $runtime ->addActivity($feature, $class->getName()); + # Register Nexus Service + self::isNexusService($class) and $runtime + ->addNexusService($feature, $class->getName()); + # Register Converters $class->implementsInterface(PayloadConverterInterface::class) and $runtime ->addConverter($feature, $class->getName()); @@ -43,6 +48,21 @@ public static function createState(array $argv, string $workDir): State return $runtime; } + private static function isNexusService(\ReflectionClass $class): bool + { + if ($class->isInterface() || $class->isAbstract()) { + return false; + } + + foreach ($class->getInterfaces() as $interface) { + if ($interface->getAttributes(Service::class) !== []) { + return true; + } + } + + return $class->getAttributes(Service::class) !== []; + } + public static function init(): void { \ini_set('display_errors', 'stderr'); diff --git a/harness/php/worker.php b/harness/php/worker.php index a416389d..38314ab6 100644 --- a/harness/php/worker.php +++ b/harness/php/worker.php @@ -100,6 +100,11 @@ $getWorker($feature->taskQueue)->registerActivityImplementations($container->make($activity)); } + // Register Nexus Services + foreach ($runtime->nexusServices() as $feature => $service) { + $getWorker($feature->taskQueue)->registerNexusServiceImplementation($container->make($service)); + } + $factory->run(); } catch (\Throwable $e) { \td($e); From 65984a1807ea1bb54c18591dd9ef96f87410f16b Mon Sep 17 00:00:00 2001 From: Dmitriy Derepko Date: Tue, 18 Aug 2026 21:06:41 +0400 Subject: [PATCH 2/9] fix(php): address review of the nexus harness support - pin the SDK to an exact commit instead of the branch tip - start RoadRunner when a feature only registers nexus services - build the worker factory with a workflow client so async operations register - discover nexus services only through a #[Service] interface, never workflows - treat an empty endpoint piece as absent and let the python harness ignore it - set the workflow execution timeout the Go and Java features set --- features/nexus/sync_success/feature.php | 9 +++++---- harness/php/composer.json | 8 ++------ harness/php/runner.php | 5 ++++- harness/php/src/Input/Command.php | 1 + harness/php/src/Runtime/State.php | 2 +- harness/php/src/RuntimeBuilder.php | 8 ++++++-- harness/php/worker.php | 16 ++++++++-------- harness/python/main.py | 2 +- 8 files changed, 28 insertions(+), 23 deletions(-) diff --git a/features/nexus/sync_success/feature.php b/features/nexus/sync_success/feature.php index 773a1523..5c9786ac 100644 --- a/features/nexus/sync_success/feature.php +++ b/features/nexus/sync_success/feature.php @@ -35,10 +35,9 @@ public function sayHello(string $name): string #[WorkflowInterface] class FeatureWorkflow { - #[WorkflowMethod('Harness_Nexus_SyncSuccess')] + #[WorkflowMethod('Workflow')] public function run(string $endpoint) { - /** @var TestService $service */ $service = Workflow::newNexusServiceStub( TestService::class, NexusOperationOptions::new() @@ -58,8 +57,10 @@ public static function check(WorkflowClientInterface $client, Feature $feature): Assert::notNull($feature->nexusEndpoint, 'Nexus endpoint is not provided by the runner'); $stub = $client->newUntypedWorkflowStub( - 'Harness_Nexus_SyncSuccess', - WorkflowOptions::new()->withTaskQueue($feature->taskQueue), + 'Workflow', + WorkflowOptions::new() + ->withTaskQueue($feature->taskQueue) + ->withWorkflowExecutionTimeout('1 minute'), ); $client->start($stub, $feature->nexusEndpoint); diff --git a/harness/php/composer.json b/harness/php/composer.json index a36d74be..f055a658 100644 --- a/harness/php/composer.json +++ b/harness/php/composer.json @@ -2,18 +2,14 @@ "name": "temporal/harness", "type": "project", "description": "Temporal SDK Harness", - "keywords": [ - "temporal", - "sdk", - "harness" - ], + "keywords": ["temporal", "sdk", "harness"], "license": "MIT", "require": { "php": "^8.2", "buggregator/trap": "^1.9", "spiral/core": "^3.13", "symfony/process": ">=6.4", - "temporal/sdk": "dev-nexus-new as 2.18.0", + "temporal/sdk": "dev-nexus-new#85126b125b82c5218c70d9b96a1d8bbdbafd00b7 as 2.18.0", "webmozart/assert": "^1.11" }, "autoload": { diff --git a/harness/php/runner.php b/harness/php/runner.php index d67e2e61..bb1ca667 100644 --- a/harness/php/runner.php +++ b/harness/php/runner.php @@ -36,7 +36,10 @@ $runner = new Runner($runtime); // Run RoadRunner server if workflows or activities are defined -if (\iterator_to_array($runtime->workflows(), false) !== [] || \iterator_to_array($runtime->activities(), false) !== []) { +if (\iterator_to_array($runtime->workflows(), false) !== [] + || \iterator_to_array($runtime->activities(), false) !== [] + || \iterator_to_array($runtime->nexusServices(), false) !== [] +) { $runner->start(); } diff --git a/harness/php/src/Input/Command.php b/harness/php/src/Input/Command.php index 772cc5fb..b76789e9 100644 --- a/harness/php/src/Input/Command.php +++ b/harness/php/src/Input/Command.php @@ -68,6 +68,7 @@ public static function fromCommandLine(array $argv): self } [$dir, $taskQueue, $nexusEndpoint] = \array_pad(\explode(':', $chunk, 3), 3, null); + $nexusEndpoint === '' and $nexusEndpoint = null; $self->features[] = new Feature( dir: $dir, namespace: 'Harness\\Feature\\' . self::namespaceFromPath($dir), diff --git a/harness/php/src/Runtime/State.php b/harness/php/src/Runtime/State.php index 5d6dc156..33dc0cfd 100644 --- a/harness/php/src/Runtime/State.php +++ b/harness/php/src/Runtime/State.php @@ -149,4 +149,4 @@ private function getFeature(\Harness\Input\Feature $feature): Feature $feature->nexusEndpoint, ); } -} \ No newline at end of file +} diff --git a/harness/php/src/RuntimeBuilder.php b/harness/php/src/RuntimeBuilder.php index f0ac1f45..cdd1ef44 100644 --- a/harness/php/src/RuntimeBuilder.php +++ b/harness/php/src/RuntimeBuilder.php @@ -9,8 +9,8 @@ use Harness\Input\Feature; use Harness\Runtime\State; use Temporal\Activity\ActivityInterface; -use Temporal\Nexus\Attribute\Service; use Temporal\DataConverter\PayloadConverterInterface; +use Temporal\Nexus\Attribute\Service; use Temporal\Workflow\WorkflowInterface; final class RuntimeBuilder @@ -54,13 +54,17 @@ private static function isNexusService(\ReflectionClass $class): bool return false; } + if ($class->getAttributes(WorkflowInterface::class) !== [] || $class->getAttributes(ActivityInterface::class) !== []) { + return false; + } + foreach ($class->getInterfaces() as $interface) { if ($interface->getAttributes(Service::class) !== []) { return true; } } - return $class->getAttributes(Service::class) !== []; + return false; } public static function init(): void diff --git a/harness/php/worker.php b/harness/php/worker.php index 38314ab6..a70e3f56 100644 --- a/harness/php/worker.php +++ b/harness/php/worker.php @@ -58,14 +58,6 @@ $converter = new DataConverter(...$converters); $container->bindSingleton(DataConverter::class, $converter); - $factory = WorkerFactory::create(converter: $converter); - $getWorker = static function (string $taskQueue) use (&$workers, $factory): WorkerInterface { - return $workers[$taskQueue] ??= $factory->newWorker( - $taskQueue, - WorkerOptions::new()->withMaxConcurrentActivityExecutionSize(10) - ); - }; - // Create client services $serviceClient = $runtime->command->tlsKey === null && $runtime->command->tlsCert === null ? ServiceClient::create($runtime->address) @@ -79,6 +71,14 @@ $workflowClient = WorkflowClient::create(serviceClient: $serviceClient, options: $options, converter: $converter); $scheduleClient = ScheduleClient::create(serviceClient: $serviceClient, options: $options, converter: $converter); + $factory = WorkerFactory::create(converter: $converter, client: $workflowClient); + $getWorker = static function (string $taskQueue) use (&$workers, $factory): WorkerInterface { + return $workers[$taskQueue] ??= $factory->newWorker( + $taskQueue, + WorkerOptions::new()->withMaxConcurrentActivityExecutionSize(10) + ); + }; + // Bind services $container->bindSingleton(State::class, $runtime); $container->bindSingleton(ServiceClientInterface::class, $serviceClient); diff --git a/harness/python/main.py b/harness/python/main.py index ed4fe4bd..a0649b2e 100644 --- a/harness/python/main.py +++ b/harness/python/main.py @@ -68,7 +68,7 @@ async def run(): failed_features = [] for rel_dir_and_task_queue in cast(List[str], args.features): # Split rel dir and task queue - rel_dir, _, task_queue = rel_dir_and_task_queue.partition(":") + rel_dir, task_queue = rel_dir_and_task_queue.split(":")[:2] if rel_dir not in rel_dirs: raise ValueError(f"Cannot find feature file in {rel_dir}") # Import From 2bc76bdc8db7b94dbc5921a59ec6299a4d6e26dd Mon Sep 17 00:00:00 2001 From: Dmitriy Derepko Date: Tue, 18 Aug 2026 23:42:39 +0400 Subject: [PATCH 3/9] build(php): build RoadRunner with velox instead of downloading a release dload's velox action compiles rr from a pinned template plus the plugin refs in velox.toml, so the harness can test an unreleased rrtemporal by switching one ref. --- harness/php/composer.json | 7 +++++-- harness/php/dload.xml | 6 ++++++ harness/php/velox.toml | 28 ++++++++++++++++++++++++++++ 3 files changed, 39 insertions(+), 2 deletions(-) create mode 100644 harness/php/dload.xml create mode 100644 harness/php/velox.toml diff --git a/harness/php/composer.json b/harness/php/composer.json index f055a658..013fe774 100644 --- a/harness/php/composer.json +++ b/harness/php/composer.json @@ -18,7 +18,7 @@ } }, "scripts": { - "rr-get": "rr get" + "rr-get": "dload build --config ../harness/php/dload.xml" }, "prefer-stable": true, "minimum-stability": "dev", @@ -27,5 +27,8 @@ "type": "vcs", "url": "https://github.com/temporalio/sdk-php" } - ] + ], + "require-dev": { + "internal/dload": "^1.8" + } } diff --git a/harness/php/dload.xml b/harness/php/dload.xml new file mode 100644 index 00000000..9fc5ed02 --- /dev/null +++ b/harness/php/dload.xml @@ -0,0 +1,6 @@ + + + + + + diff --git a/harness/php/velox.toml b/harness/php/velox.toml new file mode 100644 index 00000000..5850ec71 --- /dev/null +++ b/harness/php/velox.toml @@ -0,0 +1,28 @@ +[roadrunner] +ref = "v2025.1.15" + +[log] +level = "info" +mode = "production" + +[github] + [github.token] + token = "${GITHUB_TOKEN}" + + [github.plugins] + [github.plugins.server] + ref = "v5.2.10" + owner = "roadrunner-server" + repository = "server" + + [github.plugins.logger] + ref = "v5.1.9" + owner = "roadrunner-server" + repository = "logger" + + # Switch ref to the branch under test once it exists in this repository: + # a fork cannot be used here, Go resolves the module by its canonical path. + [github.plugins.temporal] + ref = "v5.11.0" + owner = "temporalio" + repository = "roadrunner-temporal" From dca41a01fd21642a03f86e78f7aee43e58e22555 Mon Sep 17 00:00:00 2001 From: Dmitriy Derepko Date: Tue, 18 Aug 2026 23:44:42 +0400 Subject: [PATCH 4/9] refactor(php): drop the unreachable interface check in nexus discovery --- harness/php/src/RuntimeBuilder.php | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/harness/php/src/RuntimeBuilder.php b/harness/php/src/RuntimeBuilder.php index cdd1ef44..9b4b5ea3 100644 --- a/harness/php/src/RuntimeBuilder.php +++ b/harness/php/src/RuntimeBuilder.php @@ -50,7 +50,7 @@ public static function createState(array $argv, string $workDir): State private static function isNexusService(\ReflectionClass $class): bool { - if ($class->isInterface() || $class->isAbstract()) { + if ($class->isAbstract()) { return false; } From 08e1cee54ed40f23e76f2ba4697908719d41ea28 Mon Sep 17 00:00:00 2001 From: Dmitriy Derepko Date: Wed, 19 Aug 2026 09:45:09 +0400 Subject: [PATCH 5/9] build(php): build the temporal plugin from the nexus branch --- harness/php/velox.toml | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/harness/php/velox.toml b/harness/php/velox.toml index 5850ec71..6fb934db 100644 --- a/harness/php/velox.toml +++ b/harness/php/velox.toml @@ -20,9 +20,8 @@ mode = "production" owner = "roadrunner-server" repository = "logger" - # Switch ref to the branch under test once it exists in this repository: - # a fork cannot be used here, Go resolves the module by its canonical path. + # A fork cannot be used here: Go resolves the module by its canonical path. [github.plugins.temporal] - ref = "v5.11.0" + ref = "nexus" owner = "temporalio" repository = "roadrunner-temporal" From 02697b317a5c6f2e09ea8442316e409ca844e5d3 Mon Sep 17 00:00:00 2001 From: Dmitriy Derepko Date: Wed, 19 Aug 2026 09:46:44 +0400 Subject: [PATCH 6/9] revert(python): restore the python harness argument parsing --- harness/python/main.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/harness/python/main.py b/harness/python/main.py index a0649b2e..ed4fe4bd 100644 --- a/harness/python/main.py +++ b/harness/python/main.py @@ -68,7 +68,7 @@ async def run(): failed_features = [] for rel_dir_and_task_queue in cast(List[str], args.features): # Split rel dir and task queue - rel_dir, task_queue = rel_dir_and_task_queue.split(":")[:2] + rel_dir, _, task_queue = rel_dir_and_task_queue.partition(":") if rel_dir not in rel_dirs: raise ValueError(f"Cannot find feature file in {rel_dir}") # Import From 9c41941432ec3c5b0d6d91da0abb7667b6f3baab Mon Sep 17 00:00:00 2001 From: Dmitriy Derepko Date: Wed, 19 Aug 2026 09:48:46 +0400 Subject: [PATCH 7/9] feat(php): register nexus services declared on a class or an interface --- harness/php/src/RuntimeBuilder.php | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/harness/php/src/RuntimeBuilder.php b/harness/php/src/RuntimeBuilder.php index 9b4b5ea3..e5c984a5 100644 --- a/harness/php/src/RuntimeBuilder.php +++ b/harness/php/src/RuntimeBuilder.php @@ -54,8 +54,8 @@ private static function isNexusService(\ReflectionClass $class): bool return false; } - if ($class->getAttributes(WorkflowInterface::class) !== [] || $class->getAttributes(ActivityInterface::class) !== []) { - return false; + if ($class->getAttributes(Service::class) !== []) { + return true; } foreach ($class->getInterfaces() as $interface) { From 40d2ffc4b2c6373b97afa2f3f8fb31886813826f Mon Sep 17 00:00:00 2001 From: Dmitriy Derepko Date: Wed, 19 Aug 2026 21:05:29 +0400 Subject: [PATCH 8/9] style(php): register nexus services with a plain if --- harness/php/src/RuntimeBuilder.php | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/harness/php/src/RuntimeBuilder.php b/harness/php/src/RuntimeBuilder.php index e5c984a5..3a1c2427 100644 --- a/harness/php/src/RuntimeBuilder.php +++ b/harness/php/src/RuntimeBuilder.php @@ -31,8 +31,9 @@ public static function createState(array $argv, string $workDir): State ->addActivity($feature, $class->getName()); # Register Nexus Service - self::isNexusService($class) and $runtime - ->addNexusService($feature, $class->getName()); + if (self::isNexusService($class)) { + $runtime->addNexusService($feature, $class->getName()); + } # Register Converters $class->implementsInterface(PayloadConverterInterface::class) and $runtime From e492fef12c38a99fc87cf01c7dca526e76e08a9a Mon Sep 17 00:00:00 2001 From: Dmitriy Derepko Date: Thu, 20 Aug 2026 10:51:12 +0400 Subject: [PATCH 9/9] feat(php): add async, cancellation, failure and parallel nexus features --- features/nexus/async_cancellation/feature.php | 135 ++++++++++++++++++ features/nexus/async_success/feature.php | 122 ++++++++++++++++ .../nexus/parallel_operations/feature.php | 111 ++++++++++++++ .../nexus/sync_operation_error/feature.php | 127 ++++++++++++++++ 4 files changed, 495 insertions(+) create mode 100644 features/nexus/async_cancellation/feature.php create mode 100644 features/nexus/async_success/feature.php create mode 100644 features/nexus/parallel_operations/feature.php create mode 100644 features/nexus/sync_operation_error/feature.php diff --git a/features/nexus/async_cancellation/feature.php b/features/nexus/async_cancellation/feature.php new file mode 100644 index 00000000..97620a1f --- /dev/null +++ b/features/nexus/async_cancellation/feature.php @@ -0,0 +1,135 @@ +withWorkflowId('async-cancellation-' . $name), + $name, + ); + } +} + +#[WorkflowInterface] +class BlockingWorkflow +{ + #[WorkflowMethod('AsyncCancellationBlockingWorkflow')] + public function run(string $name) + { + yield Workflow::await(static fn(): bool => false); + + return ''; + } +} + +#[WorkflowInterface] +class FeatureWorkflow +{ + #[WorkflowMethod('Workflow')] + public function run(string $endpoint) + { + $stub = Workflow::newUntypedNexusOperationStub( + NexusOperationOptions::new() + ->withEndpoint($endpoint) + ->withService('test-service') + ->withScheduleToCloseTimeout('1 minute'), + ); + + /** @var NexusOperationHandle|null $handle */ + $handle = null; + $scope = Workflow::async(static function () use ($stub, &$handle) { + $handle = yield $stub->start('block-forever', ['world'], 'string'); + yield $handle->getResult(); + }); + + yield Workflow::await(static function () use (&$handle): bool { + return $handle !== null; + }); + yield Workflow::timer(CarbonInterval::seconds(1)); + $scope->cancel(); + + try { + yield $scope; + } catch (CanceledFailure) { + return 'canceled'; + } catch (NexusOperationFailure $e) { + if ($e->getPrevious() instanceof CanceledFailure) { + return 'canceled'; + } + + throw $e; + } + + throw new \RuntimeException('expected the cancelled operation to fail'); + } +} + +class FeatureChecker +{ + #[Check] + public static function check(WorkflowClientInterface $client, Feature $feature): void + { + Assert::notNull($feature->nexusEndpoint, 'Nexus endpoint is not provided by the runner'); + + $stub = $client->newUntypedWorkflowStub( + 'Workflow', + WorkflowOptions::new() + ->withTaskQueue($feature->taskQueue) + ->withWorkflowExecutionTimeout('1 minute'), + ); + $client->start($stub, $feature->nexusEndpoint); + + Assert::same($stub->getResult('string'), 'canceled'); + + $events = \iterator_to_array($client->getWorkflowHistory($stub->getExecution())->getEvents(), false); + + Assert::true( + self::hasEvent( + $events, + static fn(HistoryEvent $e): bool => $e->hasNexusOperationCancelRequestedEventAttributes(), + ), + 'NexusOperationCancelRequested event is missing', + ); + } + + /** + * @param list $events + * @param callable(HistoryEvent): bool $predicate + */ + private static function hasEvent(array $events, callable $predicate): bool + { + foreach ($events as $event) { + if ($predicate($event)) { + return true; + } + } + + return false; + } +} diff --git a/features/nexus/async_success/feature.php b/features/nexus/async_success/feature.php new file mode 100644 index 00000000..332c8423 --- /dev/null +++ b/features/nexus/async_success/feature.php @@ -0,0 +1,122 @@ +withWorkflowId('async-success-' . $name), + $name, + ); + } +} + +#[WorkflowInterface] +class HandlerWorkflow +{ + #[WorkflowMethod('AsyncSuccessHandlerWorkflow')] + public function run(string $name) + { + yield Workflow::timer(CarbonInterval::milliseconds(50)); + + return "Hello, {$name}!"; + } +} + +#[WorkflowInterface] +class FeatureWorkflow +{ + #[WorkflowMethod('Workflow')] + public function run(string $endpoint) + { + $stub = Workflow::newUntypedNexusOperationStub( + NexusOperationOptions::new() + ->withEndpoint($endpoint) + ->withService('test-service') + ->withScheduleToCloseTimeout('1 minute'), + ); + + /** @var NexusOperationHandle $handle */ + $handle = yield $stub->start('say-hello-async', ['world'], 'string'); + + $token = $handle->getOperationToken(); + if ($token === null || $token === '') { + throw new \RuntimeException('expected a non-empty operation token'); + } + + return 'token+' . (yield $handle->getResult()); + } +} + +class FeatureChecker +{ + #[Check] + public static function check(WorkflowClientInterface $client, Feature $feature): void + { + Assert::notNull($feature->nexusEndpoint, 'Nexus endpoint is not provided by the runner'); + + $stub = $client->newUntypedWorkflowStub( + 'Workflow', + WorkflowOptions::new() + ->withTaskQueue($feature->taskQueue) + ->withWorkflowExecutionTimeout('1 minute'), + ); + $client->start($stub, $feature->nexusEndpoint); + + Assert::same($stub->getResult('string'), 'token+Hello, world!'); + + $events = \iterator_to_array($client->getWorkflowHistory($stub->getExecution())->getEvents(), false); + + Assert::true( + self::hasEvent($events, static fn(HistoryEvent $e): bool => $e->hasNexusOperationScheduledEventAttributes()), + 'NexusOperationScheduled event is missing', + ); + Assert::true( + self::hasEvent($events, static fn(HistoryEvent $e): bool => $e->hasNexusOperationStartedEventAttributes()), + 'NexusOperationStarted event is missing', + ); + Assert::true( + self::hasEvent($events, static fn(HistoryEvent $e): bool => $e->hasNexusOperationCompletedEventAttributes()), + 'NexusOperationCompleted event is missing', + ); + } + + /** + * @param list $events + * @param callable(HistoryEvent): bool $predicate + */ + private static function hasEvent(array $events, callable $predicate): bool + { + foreach ($events as $event) { + if ($predicate($event)) { + return true; + } + } + + return false; + } +} diff --git a/features/nexus/parallel_operations/feature.php b/features/nexus/parallel_operations/feature.php new file mode 100644 index 00000000..f5cfadfb --- /dev/null +++ b/features/nexus/parallel_operations/feature.php @@ -0,0 +1,111 @@ +withEndpoint($endpoint) + ->withScheduleToCloseTimeout('1 minute'), + ); + + $promises = []; + foreach (NAMES as $name) { + $promises[] = $service->sayHello($name); + } + + return \implode(' ', yield Promise::all($promises)); + } +} + +class FeatureChecker +{ + #[Check] + public static function check(WorkflowClientInterface $client, Feature $feature): void + { + Assert::notNull($feature->nexusEndpoint, 'Nexus endpoint is not provided by the runner'); + + $stub = $client->newUntypedWorkflowStub( + 'Workflow', + WorkflowOptions::new() + ->withTaskQueue($feature->taskQueue) + ->withWorkflowExecutionTimeout('1 minute'), + ); + $client->start($stub, $feature->nexusEndpoint); + + Assert::same($stub->getResult('string'), 'Hello, one! Hello, two! Hello, three!'); + + $events = \iterator_to_array($client->getWorkflowHistory($stub->getExecution())->getEvents(), false); + + Assert::same( + self::countEvents($events, static fn(HistoryEvent $e): bool => $e->hasNexusOperationScheduledEventAttributes()), + \count(NAMES), + 'Expected one NexusOperationScheduled event per operation', + ); + Assert::same( + self::countEvents($events, static fn(HistoryEvent $e): bool => $e->hasNexusOperationCompletedEventAttributes()), + \count(NAMES), + 'Expected one NexusOperationCompleted event per operation', + ); + Assert::same( + self::countEvents($events, static fn(HistoryEvent $e): bool => $e->hasNexusOperationStartedEventAttributes()), + 0, + 'Synchronous operations must not produce NexusOperationStarted events', + ); + } + + /** + * @param list $events + * @param callable(HistoryEvent): bool $predicate + */ + private static function countEvents(array $events, callable $predicate): int + { + $count = 0; + foreach ($events as $event) { + if ($predicate($event)) { + ++$count; + } + } + + return $count; + } +} diff --git a/features/nexus/sync_operation_error/feature.php b/features/nexus/sync_operation_error/feature.php new file mode 100644 index 00000000..9cf106fc --- /dev/null +++ b/features/nexus/sync_operation_error/feature.php @@ -0,0 +1,127 @@ +withEndpoint($endpoint) + ->withService('test-service') + ->withScheduleToCloseTimeout('1 minute'), + ); + + try { + yield $stub->execute('fail', ['world'], 'string'); + } catch (NexusOperationFailure $e) { + $cause = self::findApplicationFailure($e, ERROR_TYPE); + if ($cause === null) { + throw new \RuntimeException('expected an application error cause of type ' . ERROR_TYPE); + } + + if (!\str_contains($cause->getOriginalMessage(), ERROR_MESSAGE)) { + throw new \RuntimeException( + 'expected the original failure message, got: ' . $cause->getOriginalMessage(), + ); + } + + return $cause->getType() . ': ' . ERROR_MESSAGE; + } + + throw new \RuntimeException('expected the operation to fail'); + } + + private static function findApplicationFailure(\Throwable $error, string $type): ?ApplicationFailure + { + for ($current = $error->getPrevious(); $current !== null; $current = $current->getPrevious()) { + if ($current instanceof ApplicationFailure && $current->getType() === $type) { + return $current; + } + } + + return null; + } +} + +class FeatureChecker +{ + #[Check] + public static function check(WorkflowClientInterface $client, Feature $feature): void + { + Assert::notNull($feature->nexusEndpoint, 'Nexus endpoint is not provided by the runner'); + + $stub = $client->newUntypedWorkflowStub( + 'Workflow', + WorkflowOptions::new() + ->withTaskQueue($feature->taskQueue) + ->withWorkflowExecutionTimeout('1 minute'), + ); + $client->start($stub, $feature->nexusEndpoint); + + Assert::same($stub->getResult('string'), ERROR_TYPE . ': ' . ERROR_MESSAGE); + + $events = \iterator_to_array($client->getWorkflowHistory($stub->getExecution())->getEvents(), false); + + Assert::true( + self::hasEvent($events, static fn(HistoryEvent $e): bool => $e->hasNexusOperationFailedEventAttributes()), + 'NexusOperationFailed event is missing', + ); + Assert::false( + self::hasEvent($events, static fn(HistoryEvent $e): bool => $e->hasNexusOperationCompletedEventAttributes()), + 'Unexpected NexusOperationCompleted event for a failed operation', + ); + } + + /** + * @param list $events + * @param callable(HistoryEvent): bool $predicate + */ + private static function hasEvent(array $events, callable $predicate): bool + { + foreach ($events as $event) { + if ($predicate($event)) { + return true; + } + } + + return false; + } +}