diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4617e1f0..09f6abda 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -293,7 +293,7 @@ jobs: composer require durable-workflow/sdk:@dev "${framework_packages[@]}" \ --no-interaction --prefer-dist --with-all-dependencies fi - composer require --dev phpstan/phpstan:^2.0 --no-interaction --prefer-dist + composer require --dev phpstan/phpstan:^2.0 'monolog/monolog:^2.9|^3.0' --no-interaction --prefer-dist - name: Install SDK into a fresh Laravel application for cached role isolation if: matrix.framework == 'Laravel' && matrix.version == '13' && matrix.php == '8.4' run: | @@ -329,12 +329,32 @@ jobs: bridge_path=src/Bridge/Symfony compatibility_scripts=("tests/compat/${COMPAT_SCRIPT}") fi - compatibility_scripts+=(tests/compat/history-cache.php) + compatibility_scripts+=(tests/compat/history-cache.php tests/compat/execution-telemetry.php) "$consumer/vendor/bin/phpstan" analyse -c phpstan-framework.neon \ --autoload-file="$consumer/vendor/autoload.php" \ src/Bridge/FrameworkRuntimeException.php \ src/Bridge/ServiceConfiguration.php src/Bridge/Event \ "$bridge_path" "${compatibility_scripts[@]}" + - name: Exercise optional tracing in framework workers + if: >- + (matrix.framework == 'Laravel' && matrix.version == '9' && matrix.php == '8.1') || + (matrix.framework == 'Laravel' && matrix.version == '13' && matrix.php == '8.4') || + (matrix.framework == 'Symfony' && matrix.version == '6.4') + env: + COMPAT_SCRIPT: ${{ matrix.script }} + FRAMEWORK_VERSION: ${{ matrix.version }} + DURABLE_WORKFLOW_TEST_TRACING: '1' + run: | + consumer="$(cat "$RUNNER_TEMP/framework-consumer")" + monolog='monolog/monolog:^3.0' + if [ "$COMPAT_SCRIPT" = laravel.php ] && [ "$FRAMEWORK_VERSION" = 9 ]; then + monolog='monolog/monolog:^2.9' + fi + composer --working-dir="$consumer" require --dev open-telemetry/api:1.10.1 "$monolog" \ + --no-interaction --prefer-dist --no-progress + DURABLE_WORKFLOW_CLIENT_TOKEN=client-secret \ + DURABLE_WORKFLOW_WORKER_TOKEN=worker-secret \ + php "tests/compat/${COMPAT_SCRIPT}" "$consumer/vendor/autoload.php" laravel-transition-compat: name: Laravel ${{ matrix.source }} to service mode diff --git a/CHANGELOG.md b/CHANGELOG.md index 6f40b7a4..2cb48e0d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,21 @@ project follows [Semantic Versioning](https://semver.org/). ## [Unreleased] +## [3.1.0] - 2026-10-21 + +### Added + +- Optional execution spans and Monolog correlation for workflow/query tasks + and real activity attempts. Applications own providers, exporters, sampling + and bounded shutdown. Default installations do not require OpenTelemetry. +- Bounded W3C caller propagation when Server discovery advertises the optional + observability capability. Run origins parent tasks; real dispatch, command + and outcome contexts become links after existing authority checks. Baggage + defaults empty and is excluded from links. +- Fiber scope restoration and fresh child-provider configuration for + supervised callbacks. Replay does not emit activity attempts for recorded + results. Existing request IDs and business arguments remain unchanged. + ## [3.0.0] - 2026-10-20 ### Changed diff --git a/composer.json b/composer.json index 43f3f774..680e7847 100644 --- a/composer.json +++ b/composer.json @@ -28,6 +28,9 @@ "psr/log": "^1.1|^2.0|^3.0" }, "require-dev": { + "monolog/monolog": "^2.9|^3.0", + "open-telemetry/api": "1.10.1", + "open-telemetry/sdk": "^1.15", "phpdocumentor/phpdocumentor": "^3.7", "phpstan/phpstan": "^2.0", "phpunit/phpunit": "^10.5|^11.0" @@ -35,9 +38,13 @@ "suggest": { "ext-pcntl": "Required by framework console workers for graceful SIGINT and SIGTERM handling.", "laravel/framework": "Enables the auto-discovered Laravel 9 through 13 service-mode bridge.", + "open-telemetry/api": "Optional worker tracing with the application's provider; supports API 1.10.1 and later in the 1.x line.", "symfony/framework-bundle": "Enables the Symfony 6.4, 7, and 8 service-mode bundle.", "symfony/console": "Required with symfony/framework-bundle to run the Symfony 6.4, 7, and 8 worker command." }, + "conflict": { + "open-telemetry/api": "<1.10.1 || >=2.0" + }, "autoload": { "psr-4": { "DurableWorkflow\\": "src/" @@ -82,7 +89,7 @@ } }, "durable-workflow": { - "product-train": "3.0.0", + "product-train": "3.1.0", "supported-server-versions": "2.5.0", "worker-protocol-version": "1.20", "control-plane-version": "2", diff --git a/docs/quickstart-contract.json b/docs/quickstart-contract.json index b42ac914..12985d9e 100644 --- a/docs/quickstart-contract.json +++ b/docs/quickstart-contract.json @@ -3,9 +3,9 @@ "schema_version": 2, "package": { "name": "durable-workflow/sdk", - "published_version": "3.0.0", - "composer_requirement": "3.0.0", - "onboarding_requirement": "^3.0" + "published_version": "3.1.0", + "composer_requirement": "3.1.0", + "onboarding_requirement": "^3.1" }, "runtime_targets": { "server": { diff --git a/docs/tracing.md b/docs/tracing.md new file mode 100644 index 00000000..128b83d2 --- /dev/null +++ b/docs/tracing.md @@ -0,0 +1,43 @@ +# Optional worker execution context + +This integration adds scoped worker spans and log correlation. It also supports the optional service trace-context contract when the connected Server explicitly advertises it. Servers without that capability use independent task traces. Use the configured Server/SDK tuple qualified in [Workflow #773](https://github.com/durable-workflow/workflow/issues/773). Installing OpenTelemetry alone does not enable cross-process propagation. + +Install the optional API and your preferred OpenTelemetry SDK/exporter. API 1.10.1 is the minimum supported API within the 1.x line. Configure application providers, propagators, sampling, buffering, exporter timeouts and shutdown before constructing the worker integration. The SDK does not install a provider, force a sampling decision or flush a provider per task. + +```php +use DurableWorkflow\Telemetry\ExecutionTelemetry; +use DurableWorkflow\Telemetry\LogContext; +use DurableWorkflow\Worker; + +// $client, $provider and $logger are application-owned. +$logger->pushProcessor(new LogContext()); // Monolog 2 or 3 +$worker = new Worker($client, 'orders', telemetry: new ExecutionTelemetry($provider)); +``` + +Omit `telemetry` for the default execution path. `new ExecutionTelemetry(enabled: false)` also disables instrumentation. When the optional API is absent, the integration is a no-op. Applications can use `ExecutionTelemetry::current()` within a supported execution scope to obtain the current correlation fields without adding wrappers to individual HTTP or database calls. + +For a Laravel console worker, explicitly bind `ExecutionTelemetry::class` in the application container after configuring OpenTelemetry. The bridge injects that binding into its worker factory. Add `DurableWorkflow\Bridge\Laravel\LogContext::class` to the relevant logging channel's `tap` list. For a Symfony console worker, define an `ExecutionTelemetry` service and register the `DurableWorkflow\Telemetry\LogContext` service with the `monolog.processor` tag. Both bridges leave instrumentation disabled when the service is absent. Handler constructors resolved at worker startup are outside a task scope; per-invocation handler resolution runs within the execution scope. + +`durable.workflow.task`, `durable.activity.task` and `durable.query.task` span the claimed task. Actual callbacks produce `durable.activity.attempt` or `durable.local_activity.attempt` children. Replaying a completed activity does not create another attempt span. SDK-created workflow Fibers bind their hosting task context inside the Fiber and restore it on unwinding. Unrelated application Fibers receive no durable correlation fields automatically. + +The instrumentation scope is `durable-workflow/sdk`, with the installed package version. Fields include `workflow_instance_id`, `workflow_run_id`, `workflow_type`, `namespace`, `task_id`, `task_queue`, `activity_id`, `activity_type`, `activity_attempt_id` and `activity_attempt_number` when available. `activity_id` identifies the logical Server activity execution across retries. Older local activity reports without a logical execution ID omit it. Trace and span IDs follow the currently active application child span. An application's log-record `request_id` is preserved. + +Field values are bounded and control characters are rejected. Arguments, results, exception messages, stack arguments, credentials and HTTP bodies are excluded from span attributes. High-cardinality IDs are not installed as metric labels. Telemetry setup, status and exporter failures preserve the original callback result or exception. + +## Optional service propagation + +Configure the same integration on a client with `$client = $client->withTelemetry($telemetry)`. The returned client is a clone. Use the existing explicit `$client->clusterInfo()` discovery during application setup. Workers use their existing cluster discovery. The SDK caches only an exact supported `worker_protocol.server_capabilities.observability_context` object. Missing, malformed or unknown capabilities disable propagation. Business commands do not add a discovery request or retry because telemetry is unavailable. + +When available, start, signal, update and cancellation requests capture the caller's context before HTTP instrumentation begins. Child-workflow and continue-as-new commands capture their actual workflow execution context. The SDK adds only the structural `observability_context` field and preserves application arguments and request keys. + +Workers read optional metadata after the existing original claim and authority gates. A valid run origin becomes the task's parent. A task-bound dispatch context and validated command or outcome observations become links, without replacing the run parent. Child terminal events may omit their producer task ID. The SDK never fabricates one. Existing Servers continue to use independent task spans. + +Official W3C propagators validate trace headers. Baggage is dropped unless its key appears in the `baggageAllowlist` constructor option, and links never carry baggage. Context and link envelopes are bounded to 4096 bytes, with at most 32 links. Invalid metadata is ignored; span diagnostics contain only `discarded` or `filtered`. The optional capability and published-tuple qualification determine Server propagation support. + +## Cooperative callback processes + +Cooperative activity callbacks run in a Unix child process. The owner retains the task span. By default, tracing in the child is disabled to avoid reusing inherited exporter connections. Durable log identities remain available when the API is installed. + +To trace the child callback, supply `childProviderFactory` when constructing `ExecutionTelemetry`. The factory runs after fork and must create a fresh provider with a synchronous span processor and bounded exporter I/O. Child processes stop without running inherited PHP shutdown hooks; a batch processor that only drains at shutdown loses its buffered spans. A missing or failing factory uses a no-op child provider. Abrupt process termination can lose telemetry and does not provide exactly-once export. + +The application owns the export queue limits, timeouts, retry policy and final shutdown of its long-running worker provider. Keep export within the worker's existing execution and lease budgets. Configure a synchronous child exporter with a finite request timeout before enabling child spans. diff --git a/scripts/ci/run-replay-regression-fixture.php b/scripts/ci/run-replay-regression-fixture.php index 6bd062fd..ebb7c387 100644 --- a/scripts/ci/run-replay-regression-fixture.php +++ b/scripts/ci/run-replay-regression-fixture.php @@ -532,6 +532,11 @@ private static function executeLayout( } }, ); + if ($workflowType === 'golden.telemetry-context') { + $worker = new Worker(new Client('https://replay.invalid', transport: $transport, codec: $codec), + 'regression-corpus', workerId: 'regression-consumer', clock: static fn (): float => 0, + telemetry: new \DurableWorkflow\Telemetry\ExecutionTelemetry()); + } if ($workflowType === 'golden.attributed-stateful') { $worker->register(ReplayRegressionAttributedStatefulWorkflow::class); } elseif ($workflowType === '0') { @@ -639,6 +644,14 @@ private static function assertExpected(string $identity, array $fixture, array $ private static function workflow(string $workflowType): callable { return match ($workflowType) { + 'golden.telemetry-context' => static function (WorkflowContext $context): array { + $result = $context->activity('golden.greet', ['Ada']); + $fields = \DurableWorkflow\Telemetry\ExecutionTelemetry::current(); + + return ['correlated' => ($fields['workflow_instance_id'] ?? null) === $context->workflowId + && ($fields['workflow_run_id'] ?? null) === $context->runId, + 'namespace' => $fields['namespace'] ?? null, 'result' => $result]; + }, 'golden.entry-arguments-required' => static fn (WorkflowContext $context, int $amount, int $required): int => $amount + $required, 'golden.finally-disposal' => static function (WorkflowContext $context): void { try { diff --git a/src/Bridge/Laravel/DurableWorkflowServiceProvider.php b/src/Bridge/Laravel/DurableWorkflowServiceProvider.php index 8149a84d..a8d30f67 100644 --- a/src/Bridge/Laravel/DurableWorkflowServiceProvider.php +++ b/src/Bridge/Laravel/DurableWorkflowServiceProvider.php @@ -78,6 +78,8 @@ public function register(): void $logger, $events, ProcessCredentialResolver::workerCredentialRole(), + $app->bound(\DurableWorkflow\Telemetry\ExecutionTelemetry::class) + ? $app->make(\DurableWorkflow\Telemetry\ExecutionTelemetry::class) : null, ); }); } diff --git a/src/Bridge/Laravel/LogContext.php b/src/Bridge/Laravel/LogContext.php new file mode 100644 index 00000000..860b2b40 --- /dev/null +++ b/src/Bridge/Laravel/LogContext.php @@ -0,0 +1,19 @@ +getLogger(); + if (method_exists($target, 'pushProcessor')) { + $target->pushProcessor(new \DurableWorkflow\Telemetry\LogContext()); + } + } +} diff --git a/src/Bridge/Laravel/WorkerFactory.php b/src/Bridge/Laravel/WorkerFactory.php index 6b558c7f..80d07663 100644 --- a/src/Bridge/Laravel/WorkerFactory.php +++ b/src/Bridge/Laravel/WorkerFactory.php @@ -7,6 +7,7 @@ use DurableWorkflow\Bridge\Event\WorkerDiagnosticEvent; use DurableWorkflow\Bridge\ServiceConfiguration; use DurableWorkflow\Client; +use DurableWorkflow\Telemetry\ExecutionTelemetry; use DurableWorkflow\Worker; use DurableWorkflow\Worker\WorkflowFactory; use Illuminate\Contracts\Container\Container; @@ -24,6 +25,7 @@ public function __construct( private readonly LoggerInterface $logger, private readonly ?Dispatcher $events = null, private readonly string $credentialRole = 'worker', + private readonly ?ExecutionTelemetry $telemetry = null, ) { if (!in_array($this->credentialRole, ['worker', 'shared'], true)) { throw new InvalidArgumentException('The Laravel worker credential role must be worker or shared.'); @@ -63,6 +65,7 @@ public function make(?string $taskQueue = null): Worker }, stickyCacheCapacity: $this->configuration->stickyCacheCapacity, stickyCacheTtlSeconds: $this->configuration->stickyCacheTtlSeconds, + telemetry: $this->telemetry, ); $handlers = []; diff --git a/src/Bridge/Symfony/DependencyInjection/DurableWorkflowExtension.php b/src/Bridge/Symfony/DependencyInjection/DurableWorkflowExtension.php index e9867ed6..66adcdd7 100644 --- a/src/Bridge/Symfony/DependencyInjection/DurableWorkflowExtension.php +++ b/src/Bridge/Symfony/DependencyInjection/DurableWorkflowExtension.php @@ -69,6 +69,7 @@ public function load(array $configs, ContainerBuilder $container): void new ServiceLocatorArgument(new TaggedIteratorArgument(self::HANDLER_TAG, needsIndexes: true)), new Reference(LoggerInterface::class, ContainerBuilder::NULL_ON_INVALID_REFERENCE), new Reference(EventDispatcherInterface::class, ContainerBuilder::NULL_ON_INVALID_REFERENCE), + new Reference(\DurableWorkflow\Telemetry\ExecutionTelemetry::class, ContainerBuilder::NULL_ON_INVALID_REFERENCE), ]) ->setPublic(true); $container->register(WorkerCommand::class, WorkerCommand::class) diff --git a/src/Bridge/Symfony/WorkerFactory.php b/src/Bridge/Symfony/WorkerFactory.php index 482576d6..4ffa7043 100644 --- a/src/Bridge/Symfony/WorkerFactory.php +++ b/src/Bridge/Symfony/WorkerFactory.php @@ -7,6 +7,7 @@ use DurableWorkflow\Bridge\Event\WorkerDiagnosticEvent; use DurableWorkflow\Bridge\ServiceConfiguration; use DurableWorkflow\Client; +use DurableWorkflow\Telemetry\ExecutionTelemetry; use DurableWorkflow\Worker; use DurableWorkflow\Worker\WorkflowFactory; use Psr\Log\LoggerInterface; @@ -26,6 +27,7 @@ public function __construct( private readonly ServiceProviderInterface|iterable $handlers, private readonly ?LoggerInterface $logger = null, private readonly ?EventDispatcherInterface $events = null, + private readonly ?ExecutionTelemetry $telemetry = null, ) { } @@ -41,6 +43,7 @@ public function make(?string $taskQueue = null): Worker }, stickyCacheCapacity: $this->configuration->stickyCacheCapacity, stickyCacheTtlSeconds: $this->configuration->stickyCacheTtlSeconds, + telemetry: $this->telemetry, ); $handlers = []; diff --git a/src/Client.php b/src/Client.php index 97278fc9..d972b07e 100644 --- a/src/Client.php +++ b/src/Client.php @@ -46,6 +46,8 @@ use DurableWorkflow\Transport\RuntimePayloads; use DurableWorkflow\Transport\RuntimePayloadUploads; use DurableWorkflow\Transport\RequestBudget; +use DurableWorkflow\Telemetry\ExecutionTelemetry; +use DurableWorkflow\Telemetry\TraceCarrier; use DurableWorkflow\Worker\PollResponse; use DurableWorkflow\Worker\CapabilityManifest; use DurableWorkflow\Worker\CancellationRequest; @@ -69,6 +71,8 @@ final class Client implements WorkflowClientInterface private readonly PayloadCodec $codec; private readonly RuntimePayloadUploads $payloadUploads; private bool $boundedWorkerRequests = false; + private ?ExecutionTelemetry $executionTelemetry = null; + private bool $observabilityContextSupported = false; /** @var (\Closure(string, array|null, array, RequestBudget|null): (?\Closure))|null */ private ?\Closure $workerPayloadObserver = null; @@ -127,6 +131,21 @@ public function payloadCodec(): PayloadCodec return $this->codec; } + /** Configure tracing before explicit cluster discovery and business I/O. */ + public function withTelemetry(ExecutionTelemetry $telemetry): self + { + $copy = clone $this; + $copy->executionTelemetry = $telemetry; + + return $copy; + } + + /** @internal Uses actual cached discovery; never performs optional I/O in a business command. */ + public function observabilityContextAvailable(): bool + { + return $this->observabilityContextSupported; + } + /** * @internal Keep Worker payload-transfer coordination on its own Client. * @param \Closure(string, array|null, array, RequestBudget|null): (?\Closure) $observer @@ -192,9 +211,13 @@ public function health(): array public function clusterInfo(bool $includeDiagnostics = false): ClusterInfo { + $this->observabilityContextSupported = false; $path = $includeDiagnostics ? '/cluster/info?include=diagnostics' : '/cluster/info'; - return ClusterInfo::fromArray($this->control('GET', $path)); + $info = $this->control('GET', $path); + $this->observabilityContextSupported = TraceCarrier::supported($info); + + return ClusterInfo::fromArray($info); } /** @@ -302,7 +325,11 @@ private function serviceCatalogRead(string $path, array $filters = [], ?string $ /** Discover ordinary runtime capabilities using the worker credential. */ public function workerClusterInfo(): ClusterInfo { - return ClusterInfo::fromArray($this->worker('GET', '/cluster/info')); + $this->observabilityContextSupported = false; + $info = $this->worker('GET', '/cluster/info'); + $this->observabilityContextSupported = TraceCarrier::supported($info); + + return ClusterInfo::fromArray($info); } /** @@ -340,7 +367,7 @@ public function startWorkflow( 'fairness_weight' => $fairnessWeight, 'build_id' => $buildId, ]); - $response = $this->control('POST', '/workflows', $body); + $response = $this->control('POST', '/workflows', $this->withObservabilityContext($body)); return new WorkflowHandle( $this, @@ -718,6 +745,13 @@ public function workflowObservation( } /** + * Send a signal and return the Server receipt. + * + * `$requestId` is the caller command key (1 to 255 Unicode characters). Omit it + * to generate `sdk-php-signal-` once per call. The receipt always + * carries `request_id`; retrying with the same key and intent returns the + * retained receipt (`retained: true`) instead of a second signal. + * * @param list $arguments * @return array */ @@ -726,10 +760,16 @@ public function signalWorkflow( string $signalName, array $arguments = [], ?string $runId = null, + ?string $requestId = null, ): array { + $requestId = $this->commandRequestId('signal', $requestId); $path = $this->workflowOperationPath($workflowId, $runId, 'signal/'.$this->segment($signalName)); + $response = $this->control('POST', $path, $this->withObservabilityContext([ + 'input' => $this->codec->envelope($arguments), + 'request_id' => $requestId, + ]), 'signal', $requestId); - return $this->control('POST', $path, ['input' => $this->codec->envelope($arguments)], 'signal'); + return $response + ['request_id' => $requestId]; } /** @@ -765,7 +805,13 @@ public function queryWorkflow( return $this->resultFromResponse($response); } - /** @param list $arguments */ + /** + * Send an update and return its decoded result, or the receipt while the + * update is still pending. Use updateWorkflowReceipt() to read the receipt + * with its `request_id` after a completed update. + * + * @param list $arguments + */ public function updateWorkflow( string $workflowId, string $updateName, @@ -775,15 +821,41 @@ public function updateWorkflow( ?string $requestId = null, ?string $runId = null, ): mixed { + return $this->resultFromResponse($this->updateWorkflowReceipt( + $workflowId, $updateName, $arguments, $waitFor, $waitTimeoutSeconds, $requestId, $runId, + )); + } + + /** + * Send an update and return the Server receipt. + * + * `$requestId` is the caller command key (1 to 255 Unicode characters). Omit it + * to generate `sdk-php-update-` once per call. The receipt always + * carries `request_id`; retrying with the same key and intent returns the + * retained receipt (`retained: true`) instead of a second update. + * + * @param list $arguments + * @return array + */ + public function updateWorkflowReceipt( + string $workflowId, + string $updateName, + array $arguments = [], + string $waitFor = 'completed', + ?int $waitTimeoutSeconds = null, + ?string $requestId = null, + ?string $runId = null, + ): array { + $requestId = $this->commandRequestId('update', $requestId); $path = $this->workflowOperationPath($workflowId, $runId, 'update/'.$this->segment($updateName)); - $response = $this->control('POST', $path, $this->withoutNulls([ + $response = $this->control('POST', $path, $this->withObservabilityContext($this->withoutNulls([ 'input' => $this->codec->envelope($arguments), 'wait_for' => $waitFor, 'wait_timeout_seconds' => $waitTimeoutSeconds, 'request_id' => $requestId, - ]), 'update'); + ])), 'update', $requestId); - return $this->resultFromResponse($response); + return $response + ['request_id' => $requestId]; } /** @return array */ @@ -792,10 +864,26 @@ public function cancelWorkflow(string $workflowId, ?string $reason = null, ?stri return $this->control( 'POST', $this->workflowOperationPath($workflowId, $runId, 'cancel'), - $this->withoutNulls(['reason' => $reason]), + $this->withObservabilityContext($this->withoutNulls(['reason' => $reason])), ); } + /** @param array $body + * @return array + */ + private function withObservabilityContext(array $body): array + { + if (!$this->observabilityContextSupported || $this->executionTelemetry === null) { + return $body; + } + $carrier = $this->executionTelemetry->capture(); + if ($carrier !== null) { + $body['observability_context'] = $carrier; + } + + return $body; + } + /** * Request bounded workflow-authored cleanup on a capable runtime. * Repeated requests retain the Server's original identity and deadline. @@ -1950,6 +2038,15 @@ public function completeWorkflowTask( ); } + if ($this->executionTelemetry !== null && $this->observabilityContextSupported) { + foreach ($commands as &$command) { + if (in_array($command['type'] ?? null, ['start_child_workflow', 'continue_as_new'], true)) { + $command = $this->withObservabilityContext($command); + } + } + unset($command); + } + return $this->worker('POST', '/worker/workflow-tasks/'.$this->segment($taskId).'/complete', array_filter([ 'lease_owner' => $leaseOwner, 'workflow_task_attempt' => $attempt, @@ -2186,9 +2283,14 @@ public function failQueryTask( * @param array|null $body * @return array */ - private function control(string $method, string $path, ?array $body = null, ?string $operation = null): array - { - return $this->request($method, $path, false, $body, $operation); + private function control( + string $method, + string $path, + ?array $body = null, + ?string $operation = null, + ?string $requestId = null, + ): array { + return $this->request($method, $path, false, $body, $operation, requestId: $requestId); } /** @@ -2384,6 +2486,7 @@ private function request( ?string $operation = null, ?RequestBudget $budget = null, ?\Closure $payloadPaths = null, + ?string $requestId = null, ): array { $headers = $this->requestHeaders($worker); try { @@ -2432,10 +2535,10 @@ private function request( throw new QueryFailed($message, $status, $reason, $details, $exception); } if ($operation === 'update') { - throw new UpdateFailed($message, $status, $reason, $details, $exception); + throw new UpdateFailed($message, $status, $reason, $details, $exception, $requestId); } if ($operation === 'signal') { - throw new SignalFailed($message, $status, $reason, $details, $exception); + throw new SignalFailed($message, $status, $reason, $details, $exception, $requestId); } throw new ServerException($message, $status, $reason, $details, $exception); @@ -2629,6 +2732,20 @@ private function requestId(string $prefix): string return $prefix.'-'.bin2hex(random_bytes(16)); } + /** The caller command key for a signal or update: validated when supplied, otherwise generated once per call. */ + private function commandRequestId(string $command, ?string $requestId): string + { + if ($requestId === null) { + return $this->requestId('sdk-php-'.$command); + } + $characters = preg_match_all('/./us', $requestId); + if ($characters === false || $characters < 1 || $characters > 255) { + throw new InvalidArgumentException(ucfirst($command).' request IDs must be from 1 to 255 Unicode characters.'); + } + + return $requestId; + } + /** * Keep a keyed registration map from becoming a JSON list for names 0, 1, ... * diff --git a/src/Exception/CarriesCommandRequestId.php b/src/Exception/CarriesCommandRequestId.php new file mode 100644 index 00000000..dae59cc4 --- /dev/null +++ b/src/Exception/CarriesCommandRequestId.php @@ -0,0 +1,48 @@ +requestId; + } + + /** The Server refusal reason, or null when no protocol envelope arrived. */ + public function reason(): ?string + { + return $this->reason; + } + + /** + * The Server response fields, or an empty array when no envelope arrived. + * + * A `request_id_conflict` refusal carries `request_id`, `retained_command_id`, + * `retained_command_type`, `retained_outcome` and the mismatched `conflict` parts. + * + * @return array + */ + public function details(): array + { + return $this->details !== null && !array_is_list($this->details) ? $this->details : []; + } + + /** Whether the Server retained a command with a different intent under the same key. */ + public function isRequestIdConflict(): bool + { + return $this->status === 409 && $this->reason === 'request_id_conflict'; + } +} diff --git a/src/Exception/SignalFailed.php b/src/Exception/SignalFailed.php index d0ec502b..d12b8616 100644 --- a/src/Exception/SignalFailed.php +++ b/src/Exception/SignalFailed.php @@ -4,6 +4,23 @@ namespace DurableWorkflow\Exception; +use Throwable; + +/** A signal the Server refused, or whose request did not complete. */ final class SignalFailed extends ServerException { + use CarriesCommandRequestId; + + /** @param array|list|null $details */ + public function __construct( + string $message, + int $status, + ?string $reason = null, + ?array $details = null, + ?Throwable $previous = null, + ?string $requestId = null, + ) { + parent::__construct($message, $status, $reason, $details, $previous); + $this->requestId = $requestId; + } } diff --git a/src/Exception/UpdateFailed.php b/src/Exception/UpdateFailed.php index 7629f4a4..34ac72fc 100644 --- a/src/Exception/UpdateFailed.php +++ b/src/Exception/UpdateFailed.php @@ -4,6 +4,23 @@ namespace DurableWorkflow\Exception; +use Throwable; + +/** An update the Server refused, or whose request did not complete. */ final class UpdateFailed extends ServerException { + use CarriesCommandRequestId; + + /** @param array|list|null $details */ + public function __construct( + string $message, + int $status, + ?string $reason = null, + ?array $details = null, + ?Throwable $previous = null, + ?string $requestId = null, + ) { + parent::__construct($message, $status, $reason, $details, $previous); + $this->requestId = $requestId; + } } diff --git a/src/Telemetry/ExecutionTelemetry.php b/src/Telemetry/ExecutionTelemetry.php new file mode 100644 index 00000000..49223696 --- /dev/null +++ b/src/Telemetry/ExecutionTelemetry.php @@ -0,0 +1,420 @@ +>|null */ + private static ?ContextKeyInterface $key = null; + /** @var WeakMap|null */ + private static ?WeakMap $fibers = null; + private readonly int $ownerPid; + private readonly ?TracerProviderInterface $provider; + private readonly ?ContextInterface $applicationContext; + private ?int $childProviderPid = null; + private ?TracerProviderInterface $childProvider = null; + + /** + * A cooperative callback must create its own provider after fork. Use a + * synchronous processor and bounded exporter: child shutdown hooks do not run. + * Without a factory, child tracing is disabled rather than reusing sockets. + * + * @param (Closure(): TracerProviderInterface)|null $childProviderFactory + * @param list $baggageAllowlist + */ + public function __construct( + ?TracerProviderInterface $tracerProvider = null, + private readonly bool $enabled = true, + private readonly ?Closure $childProviderFactory = null, + private readonly array $baggageAllowlist = [], + ) { + $this->ownerPid = getmypid(); + $provider = null; + $applicationContext = null; + try { + if ($enabled && class_exists(Globals::class)) { + $provider = $tracerProvider ?? Globals::tracerProvider(); + // Capture providers when the worker is configured, without capturing its caller's span or baggage. + $applicationContext = Configurator::create()->withTracerProvider($provider) + ->withMeterProvider(Globals::meterProvider())->withLoggerProvider(Globals::loggerProvider()) + ->withPropagator(Globals::propagator())->withResponsePropagator(Globals::responsePropagator()) + ->storeInContext(Context::getRoot()); + } + } catch (Throwable) { + // Optional provider discovery must not prevent constructing a worker. + } + $this->provider = $provider; + $this->applicationContext = $applicationContext; + } + + /** Capture the actual caller before HTTP transport, using the app's propagators. + * @return array|null + */ + public function capture(): ?array + { + return $this->enabled && $this->provider !== null + ? TraceCarrier::capture($this->baggageAllowlist) : null; + } + + /** + * Uses only existing poll identities. Durable wire carriers are not adopted here. + * + * @param array $task + */ + public function task(string $kind, array $task, string $namespace, string $queue, callable $operation): mixed + { + $fields = self::taskFields($task, $namespace, $queue); + if ($kind !== 'activity') { + // Workflow resume metadata may describe an earlier activity, not this callback. + unset($fields['activity_id'], $fields['activity_type'], $fields['activity_attempt_id'], $fields['activity_attempt_number']); + } + + return $this->run('durable.'.$kind.'.task', $fields, $operation); + } + + /** @internal Call only after the existing claim/receipt gates have passed. + * @param array $task + * @return Closure(): void + */ + public function enterTask(string $kind, array $task, string $namespace, string $queue): Closure + { + $fields = self::taskFields($task, $namespace, $queue); + if ($kind !== 'activity') { + unset($fields['activity_id'], $fields['activity_type'], $fields['activity_attempt_id'], $fields['activity_attempt_number']); + } + $opened = $this->open('durable.'.$kind.'.task', $fields, false, $task); + $closed = false; + + return static function () use ($opened, &$closed): void { + if ($opened !== null && !$closed) { + $closed = true; + [$scope, $span, $shield] = $opened; + self::close($scope, $span); + self::close($shield, null); + } + }; + } + + /** @param array $fields */ + public function activity(array $fields, callable $operation, bool $local = false): mixed + { + return $this->run($local ? 'durable.local_activity.attempt' : 'durable.activity.attempt', + $fields + self::current(), $operation, inherit: true); + } + + /** @return array */ + public static function current(): array + { + if (self::$key === null || !class_exists(Context::class)) { + return []; + } + $fiber = Fiber::getCurrent(); + if ($fiber !== null && !isset(self::$fibers[$fiber])) { + return []; + } + try { + $fields = Context::getCurrent()->get(self::$key); + if (!is_array($fields)) { + return []; + } + $span = Span::getCurrent()->getContext(); + if ($span->isValid()) { + $fields['trace_id'] = $span->getTraceId(); + $fields['span_id'] = $span->getSpanId(); + } + + return $fields; + } catch (Throwable) { + return []; + } + } + + /** Bind only SDK-created workflow Fibers; unrelated Fibers inherit no durable identities. */ + public static function bind(callable $operation): callable + { + if (self::current() === []) { + return $operation; + } + try { + $context = Context::getCurrent(); + } catch (Throwable) { + return $operation; + } + + return static function (mixed ...$arguments) use ($operation, $context): mixed { + $scope = null; + try { + $scope = $context->activate(); + self::enterFiber(); + } catch (Throwable) { + // Instrumentation setup cannot prevent the callback from running. + } + try { + return $operation(...$arguments); + } finally { + self::close($scope, null); + } + }; + } + + /** Record a failure already classified by the worker, without exception messages or stack arguments. */ + public static function failure(Throwable $failure): void + { + if (self::current() === []) { + return; + } + try { + Span::getCurrent()->setStatus(StatusCode::STATUS_ERROR); + Span::getCurrent()->setAttribute('error.type', $failure::class); + } catch (Throwable) { + // The original failure and its durable handling remain authoritative. + } + } + + /** @param array $task + * @return array + */ + public static function taskFields(array $task, string $namespace, string $queue): array + { + return [ + 'workflow_instance_id' => $task['workflow_id'] ?? null, + 'workflow_run_id' => $task['run_id'] ?? null, + 'workflow_type' => $task['workflow_type'] ?? null, + 'namespace' => $namespace, + 'task_id' => $task['task_id'] ?? $task['query_task_id'] ?? null, + 'activity_id' => $task['activity_execution_id'] ?? null, + 'activity_type' => $task['activity_type'] ?? null, + 'activity_attempt_id' => $task['activity_attempt_id'] ?? $task['attempt_id'] ?? null, + 'activity_attempt_number' => $task['attempt_number'] ?? null, + 'request_id' => $task['request_id'] ?? null, + 'task_queue' => $queue, + ]; + } + + /** @param array $attributes */ + private function run(string $name, array $attributes, callable $operation, bool $inherit = false): mixed + { + $opened = $this->open($name, $attributes, $inherit); + if ($opened === null) { + return $operation(); + } + [$scope, $span, $childShield] = $opened; + try { + return $operation(); + } catch (Throwable $failure) { + self::failure($failure); + throw $failure; + } finally { + self::close($scope, $span); + self::close($childShield, null); + } + } + + /** @param array $attributes + * @param array|null $wireTask + * @return array{?ScopeInterface, ?SpanInterface, ?ScopeInterface}|null + */ + private function open(string $name, array $attributes, bool $inherit, ?array $wireTask = null): ?array + { + if (!$this->enabled || $this->provider === null || $this->applicationContext === null) { + return null; + } + $span = null; + $scope = null; + $childShield = null; + try { + $child = getmypid() !== $this->ownerPid; + $provider = $this->provider; + $parent = $inherit ? Context::getCurrent() : $this->applicationContext; + if ($child) { + // Shield even setup failures from inherited exporter connections. + $safe = Configurator::create()->withTracerProvider(new NoopTracerProvider()) + ->withMeterProvider(new \OpenTelemetry\API\Metrics\Noop\NoopMeterProvider()) + ->withLoggerProvider(\OpenTelemetry\API\Logs\NoopLoggerProvider::getInstance()) + ->storeInContext($this->applicationContext); + $childShield = $safe->activate(); + self::enterFiber(); + $provider = $this->childTracerProvider(); + if ($provider instanceof NoopTracerProvider) { + $parent = $safe; + } + $parent = Configurator::create()->withTracerProvider($provider) + ->withMeterProvider(new \OpenTelemetry\API\Metrics\Noop\NoopMeterProvider()) + ->withLoggerProvider(\OpenTelemetry\API\Logs\NoopLoggerProvider::getInstance()) + ->storeInContext($parent); + } + $fields = self::fields($attributes); + $version = InstalledVersions::isInstalled(self::PACKAGE_NAME) + ? InstalledVersions::getPrettyVersion(self::PACKAGE_NAME) : null; + $builder = $provider->getTracer(self::PACKAGE_NAME, $version)->spanBuilder($name) + ->setSpanKind($inherit ? SpanKind::KIND_INTERNAL : SpanKind::KIND_CONSUMER) + ->setAttributes($fields); + if ($wireTask !== null) { + $parent = $this->observeTask($builder, $parent, $wireTask, $attributes); + } + $span = $builder->setParent($parent)->startSpan(); + self::$key ??= Context::createKey('durable-workflow.sdk.correlation'); + $scope = $span->storeInContext($parent)->with(self::$key, $fields)->activate(); + self::enterFiber(); + } catch (Throwable) { + self::close($scope, $span); + // Keep a child's no-op shield active through callback fallback. + return $childShield === null ? null : [null, null, $childShield]; + } + + return [$scope, $span, $childShield]; + } + + /** @param array $wireTask + * @param array $attributes + */ + private function observeTask(SpanBuilderInterface $builder, ContextInterface $parent, array $wireTask, array $attributes): ContextInterface + { + $origin = TraceCarrier::normalize($wireTask['observability_context'] ?? null, $this->baggageAllowlist, $diagnostic); + if ($origin !== null) { + $parent = TraceContextPropagator::getInstance()->extract($origin['carrier'], context: $parent); + $parent = BaggagePropagator::getInstance()->extract($origin['carrier'], context: $parent); + } + if ($diagnostic !== null) { + $builder->setAttribute('durable.observability_context', $diagnostic); + } + $taskId = $wireTask['task_id'] ?? $wireTask['query_task_id'] ?? null; + $taskId = is_string($taskId) ? $taskId : ''; + $this->observeDispatch($builder, $wireTask, $taskId); + $this->observeLinks($builder, $wireTask, $attributes, $taskId); + return $parent; + } + + /** @param array $wireTask */ + private function observeDispatch(SpanBuilderInterface $builder, array $wireTask, string $taskId): void + { + $dispatch = TraceCarrier::dispatch($wireTask['dispatch_observability_context'] ?? null, $taskId, $diagnostic); + if ($dispatch !== null) { + $linked = TraceContextPropagator::getInstance()->extract($dispatch['carrier'], context: Context::getRoot()); + $builder->addLink(Span::fromContext($linked)->getContext(), ['durable.link.type' => 'dispatch', 'task_id' => $taskId]); + } + if ($diagnostic !== null) { + $builder->setAttribute('durable.dispatch_context', $diagnostic); + } + } + + /** @param array $wireTask + * @param array $attributes + */ + private function observeLinks(SpanBuilderInterface $builder, array $wireTask, array $attributes, string $taskId): void + { + $runId = $wireTask['run_id'] ?? null; + $namespace = $attributes['namespace'] ?? null; + $links = TraceLinks::normalize($wireTask['observability_links'] ?? null, $taskId, + is_string($runId) ? $runId : '', is_string($namespace) ? $namespace : '', $diagnostic); + foreach ($links as $entry) { + $linked = TraceContextPropagator::getInstance()->extract($entry['context']['carrier'], context: Context::getRoot()); + $linkFields = array_diff_key($entry, ['kind' => true, 'context' => true]); + // A nullable producer task stays absent; it is never fabricated. + $linkFields = array_filter($linkFields, static fn (mixed $value): bool => $value !== null); + $builder->addLink(Span::fromContext($linked)->getContext(), ['durable.link.type' => $entry['kind']] + $linkFields); + } + if ($diagnostic !== null) { + $builder->setAttribute('durable.observability_links', $diagnostic); + } + } + + private function childTracerProvider(): TracerProviderInterface + { + if ($this->childProviderPid !== getmypid()) { + $this->childProviderPid = getmypid(); + $this->childProvider = new NoopTracerProvider(); + if ($this->childProviderFactory !== null) { + try { + $this->childProvider = ($this->childProviderFactory)(); + } catch (Throwable) { + // One failed setup stays no-op for this process. + } + } + } + + return $this->childProvider ?? new NoopTracerProvider(); + } + + /** @param array $attributes + * @return array + */ + private static function fields(array $attributes): array + { + $fields = []; + foreach (self::FIELDS as $name) { + $value = $attributes[$name] ?? null; + if ($name === 'activity_attempt_number') { + if (is_int($value) && $value > 0) { + $fields[$name] = $value; + } + } elseif (is_string($value) && $value !== '' && strlen($value) <= 256 + && !preg_match('/[\x00-\x1f\x7f]/', $value)) { + $fields[$name] = $value; + } + } + + return $fields; + } + + private static function enterFiber(): void + { + $fiber = Fiber::getCurrent(); + if ($fiber !== null) { + self::$fibers ??= new WeakMap(); + self::$fibers[$fiber] = (self::$fibers[$fiber] ?? 0) + 1; + } + } + + private static function close(?ScopeInterface $scope, ?SpanInterface $span): void + { + try { + $scope?->detach(); + } catch (Throwable) { + // Telemetry cleanup cannot replace a business result or exception. + } + $fiber = Fiber::getCurrent(); + if ($scope !== null && $fiber !== null && isset(self::$fibers[$fiber]) + && --self::$fibers[$fiber] === 0) { + unset(self::$fibers[$fiber]); + } + try { + // The application owns provider buffering and shutdown. Never flush per task. + $span?->end(); + } catch (Throwable) { + // Export failures must not affect execution or acknowledgement. + } + } +} diff --git a/src/Telemetry/LogContext.php b/src/Telemetry/LogContext.php new file mode 100644 index 00000000..60c1281f --- /dev/null +++ b/src/Telemetry/LogContext.php @@ -0,0 +1,41 @@ +|LogRecord $record + * @return array|LogRecord + */ + public function __invoke(array|LogRecord $record): array|LogRecord + { + try { + $fields = ExecutionTelemetry::current(); + if ($fields === []) { + return $record; + } + $context = $record instanceof LogRecord ? $record->context : ($record['context'] ?? []); + if (!is_array($context)) { + return $record; + } + if (array_key_exists('request_id', $context)) { + unset($fields['request_id']); + } + $context = $fields + $context; + if ($record instanceof LogRecord) { + return $record->with(context: $context); + } + $record['context'] = $context; + } catch (Throwable) { + // Preserve the application's logging path if optional correlation fails. + } + + return $record; + } +} diff --git a/src/Telemetry/TraceCarrier.php b/src/Telemetry/TraceCarrier.php new file mode 100644 index 00000000..7f0e8922 --- /dev/null +++ b/src/Telemetry/TraceCarrier.php @@ -0,0 +1,191 @@ + 256, 'tracestate' => 512, 'baggage' => 2048]; + private const CAPABILITY = [ + 'supported' => true, 'schema' => self::SCHEMA, 'version' => 1, + 'dispatch_schema' => 'durable-workflow.queue-trace', 'links_schema' => 'durable-workflow.trace-links', + 'max_context_bytes' => 4096, 'max_links' => 32, 'max_links_bytes' => 4096, + ]; + + /** @param array $clusterInfo */ + public static function supported(array $clusterInfo): bool + { + $protocol = $clusterInfo['worker_protocol'] ?? null; + $capabilities = is_array($protocol) ? ($protocol['server_capabilities'] ?? null) : null; + $value = is_array($capabilities) ? ($capabilities['observability_context'] ?? null) : null; + if (!is_array($value)) { + return false; + } + foreach (self::CAPABILITY as $key => $expected) { + if (($value[$key] ?? null) !== $expected) { + return false; + } + } + + return true; + } + + public static function identity(mixed $value): bool + { + return is_string($value) && $value !== '' && strlen($value) <= 512 + && preg_match('//u', $value) === 1 && preg_match('/\p{Cc}/u', $value) === 0; + } + + /** @param list $allowlist + * @return array|null + */ + public static function capture(array $allowlist = [], ?string &$diagnostic = null): ?array + { + $diagnostic = null; + if (!class_exists(Globals::class)) { + return null; + } + try { + $headers = []; + Globals::propagator()->inject($headers); + $headers = array_intersect_key($headers, self::LIMITS); + if ($headers === []) { + return null; + } + + return self::normalize(['schema' => self::SCHEMA, 'version' => 1, 'carrier' => $headers], $allowlist, $diagnostic); + } catch (Throwable) { + $diagnostic = 'discarded'; + + return null; + } + } + + /** @param list $allowlist + * @return array|null + */ + public static function normalize(mixed $value, array $allowlist = [], ?string &$diagnostic = null): ?array + { + $diagnostic = $value === null ? null : 'discarded'; + if (!is_array($value) || count($value) !== 3 || ($value['schema'] ?? null) !== self::SCHEMA || ($value['version'] ?? null) !== 1 + || array_diff(array_keys($value), ['schema', 'version', 'carrier']) !== [] + || !is_array($value['carrier'] ?? null) || !class_exists(TraceContextPropagator::class)) { + return null; + } + $headers = $value['carrier']; + if (count($headers) > 3 || !isset($headers['traceparent']) || array_diff(array_keys($headers), array_keys(self::LIMITS)) !== []) { + return null; + } + if (!self::validHeaderValues($headers)) { + return null; + } + try { + if (strlen(json_encode($value, JSON_THROW_ON_ERROR)) > 4096) { + return null; + } + $normalized = self::normalizeHeaders($headers, $allowlist); + if ($normalized === null) { + return null; + } + $result = ['schema' => self::SCHEMA, 'version' => 1, 'carrier' => $normalized]; + if (strlen($normalized['baggage'] ?? '') > 2048 || strlen(json_encode($result, JSON_THROW_ON_ERROR)) > 4096) { + return null; + } + $diagnostic = $normalized === $headers ? null : 'filtered'; + + return $result; + } catch (Throwable) { + return null; + } + } + + /** @param array $headers + * @phpstan-assert-if-true array $headers + */ + private static function validHeaderValues(array $headers): bool + { + foreach ($headers as $key => $header) { + if (!is_string($header) || strlen($header) > self::LIMITS[$key]) { + return false; + } + } + return true; + } + + /** @param array $headers + * @param list $allowlist + * @return array|null + */ + private static function normalizeHeaders(array $headers, array $allowlist): ?array + { + $context = TraceContextPropagator::getInstance()->extract($headers, context: Context::getRoot()); + if (!Span::fromContext($context)->getContext()->isValid()) { + return null; + } + $normalized = []; + TraceContextPropagator::getInstance()->inject($normalized, context: $context); + if ($allowlist !== [] && isset($headers['baggage'])) { + self::allowlistedBaggage($headers, $allowlist, $normalized); + } + return $normalized; + } + + /** @param array $headers + * @param list $allowlist + * @param array $normalized + */ + private static function allowlistedBaggage(array $headers, array $allowlist, array &$normalized): void + { + $baggage = BaggagePropagator::getInstance()->extract($headers, context: Context::getRoot()); + $builder = Baggage::getBuilder(); + $count = 0; + foreach (Baggage::fromContext($baggage)->getAll() as $key => $entry) { + if (!in_array($key, $allowlist, true)) { + continue; + } + if (++$count > 32) { + throw new \LengthException('Baggage entry limit exceeded.'); + } + $builder->set($key, $entry->getValue(), $entry->getMetadata()); + } + BaggagePropagator::getInstance()->inject($normalized, context: $builder->build()->storeInContext(Context::getRoot())); + } + + /** @return array|null */ + public static function dispatch(mixed $value, string $taskId, ?string &$diagnostic = null): ?array + { + $diagnostic = $value === null ? null : 'discarded'; + if (!self::identity($taskId) || !is_array($value) || count($value) !== 4 || ($value['schema'] ?? null) !== 'durable-workflow.queue-trace' + || ($value['version'] ?? null) !== 1 || ($value['task_id'] ?? null) !== $taskId + || array_diff(array_keys($value), ['schema', 'version', 'task_id', 'context']) !== [] + || !is_array($value['context'] ?? null) || !is_array($value['context']['carrier'] ?? null) || count($value['context']['carrier']) > 2 + || array_diff(array_keys($value['context']['carrier']), ['traceparent', 'tracestate']) !== []) { + return null; + } + $context = self::normalize($value['context'], diagnostic: $contextDiagnostic); + if ($context === null) { + return null; + } + try { + if (strlen(json_encode($value, JSON_THROW_ON_ERROR)) > 4096) { + return null; + } + $diagnostic = $contextDiagnostic; + + return $context; + } catch (Throwable) { + return null; + } + } +} diff --git a/src/Telemetry/TraceLinks.php b/src/Telemetry/TraceLinks.php new file mode 100644 index 00000000..e6833e88 --- /dev/null +++ b/src/Telemetry/TraceLinks.php @@ -0,0 +1,101 @@ +> */ + public static function normalize(mixed $value, string $taskId, string $runId, string $namespace, ?string &$diagnostic = null): array + { + $diagnostic = $value === null ? null : 'discarded'; + if (!TraceCarrier::identity($taskId) || !TraceCarrier::identity($runId) || !TraceCarrier::identity($namespace) + || !is_array($value) || count($value) !== 6 + || ($value['schema'] ?? null) !== 'durable-workflow.trace-links' || ($value['version'] ?? null) !== 1 + || ($value['task_id'] ?? null) !== $taskId || ($value['run_id'] ?? null) !== $runId + || ($value['namespace'] ?? null) !== $namespace + || array_diff(array_keys($value), ['schema', 'version', 'task_id', 'run_id', 'namespace', 'links']) !== [] + || !is_array($value['links'] ?? null) || !array_is_list($value['links']) || count($value['links']) > 32) { + return []; + } + $result = []; + $filtered = false; + foreach ($value['links'] as $entry) { + $entry = self::entry($entry, $contextDiagnostic); + if ($entry === null) { + return []; + } + $result[] = $entry; + $filtered = $filtered || $contextDiagnostic !== null; + } + try { + if (strlen(json_encode($value, JSON_THROW_ON_ERROR)) > 4096) { + return []; + } + $diagnostic = $filtered ? 'filtered' : null; + + return $result; + } catch (Throwable) { + return []; + } + } + + /** @return array|null */ + private static function entry(mixed $value, ?string &$diagnostic): ?array + { + $diagnostic = 'discarded'; + if (!is_array($value) || count($value) > 10) { + return null; + } + $shape = match ($value['kind'] ?? null) { + 'command' => [['command_id'], []], + 'task_outcome' => [['task_id', 'event_id', 'sequence', 'event_type'], + ['activity_execution_id', 'activity_attempt_id', 'timer_id']], + 'child_outcome' => [['workflow_link_id', 'child_run_id', 'event_id', 'sequence', 'event_type'], ['child_task_id']], + default => null, + }; + if ($shape === null) { + return null; + } + [$required, $optional] = $shape; + if (array_diff($required, array_keys($value)) !== [] + || array_diff(array_keys($value), [...$required, ...$optional, 'kind', 'context']) !== []) { + return null; + } + if (!self::validIdentities($value, $required, $optional)) { + return null; + } + if (!is_array($value['context'] ?? null) || !is_array($value['context']['carrier'] ?? null) + || count($value['context']['carrier']) > 2 + || array_diff(array_keys($value['context']['carrier']), ['traceparent', 'tracestate']) !== []) { + return null; + } + $context = TraceCarrier::normalize($value['context'], diagnostic: $diagnostic); + + return $context === null ? null : array_replace($value, ['context' => $context]); + } + /** @param array $value + * @param list $required + * @param list $optional + */ + private static function validIdentities(array $value, array $required, array $optional): bool + { + foreach ([...$required, ...array_intersect($optional, array_keys($value))] as $key) { + if ($key === 'sequence') { + if (!is_int($value[$key]) || $value[$key] < 1) { + return false; + } + } elseif ($key !== 'child_task_id' || $value[$key] !== null) { + if (!TraceCarrier::identity($value[$key])) { + return false; + } + } + } + return true; + } + +} diff --git a/src/Testing/TestWorkflowHandle.php b/src/Testing/TestWorkflowHandle.php index cebb4076..451ae751 100644 --- a/src/Testing/TestWorkflowHandle.php +++ b/src/Testing/TestWorkflowHandle.php @@ -18,7 +18,7 @@ public function __construct( } /** @param list $arguments */ - public function signal(string $name, array $arguments = []): void + public function signal(string $name, array $arguments = [], ?string $requestId = null): void { $this->client->signal($this->workflowId, $name, $arguments); } diff --git a/src/Worker.php b/src/Worker.php index 583e6e06..905c6a51 100644 --- a/src/Worker.php +++ b/src/Worker.php @@ -47,6 +47,7 @@ use DurableWorkflow\Worker\WorkerSessionOptions; use DurableWorkflow\Worker\WorkflowCommand; use DurableWorkflow\Transport\RequestBudget; +use DurableWorkflow\Telemetry\ExecutionTelemetry; use Psr\Container\ContainerInterface; use Psr\Log\LoggerInterface; use Psr\Log\NullLogger; @@ -134,6 +135,7 @@ public function __construct( /** Unfrozen source opt-in for scalar/all-group scope delivery and committed scope replay. */ private readonly bool $enableCancellationScopes = false, private readonly bool $requireQueryTaskAuthority = false, + private readonly ?ExecutionTelemetry $telemetry = null, ) { if ($enableCancellationScopes && !$enableCooperativeCancellation) { throw new \InvalidArgumentException('Cancellation scopes require the cooperative worker opt-in.'); @@ -147,7 +149,8 @@ public function __construct( if ($enableCooperativeCancellation && !CooperativeActivityExecutor::available()) { throw new \InvalidArgumentException('Cooperative workers require Unix CLI with pcntl and posix.'); } - $workerClient = $enableCooperativeCancellation ? $client->withBoundedWorkerRequests() : $client; + $workerClient = $telemetry === null ? $client : $client->withTelemetry($telemetry); + $workerClient = $enableCooperativeCancellation ? $workerClient->withBoundedWorkerRequests() : $workerClient; $this->client = $workerClient->withWorkerPayloadObserver($this->workerPayloadProgress(...)); $this->workerId = $workerId ?? 'php-worker-'.bin2hex(random_bytes(8)); $this->heartbeatIntervalSeconds = $this->validHeartbeatInterval($heartbeatIntervalSeconds) @@ -179,6 +182,7 @@ public static function create( int $stickyCacheCapacity = 100, int $stickyCacheTtlSeconds = 300, bool $requireQueryTaskAuthority = false, + ?ExecutionTelemetry $telemetry = null, ): self { return new self( $client, @@ -189,6 +193,7 @@ public static function create( stickyCacheCapacity: $stickyCacheCapacity, stickyCacheTtlSeconds: $stickyCacheTtlSeconds, requireQueryTaskAuthority: $requireQueryTaskAuthority, + telemetry: $telemetry, ); } @@ -1156,6 +1161,18 @@ private function taskFromPoll(array $response): ?array /** @param array $task */ private function executePolledTask(string $taskKind, array $task): void + { + if ($this->telemetry === null || $this->client->observabilityContextAvailable()) { + $this->executeTask($taskKind, $task); + + return; + } + $this->telemetry->task($taskKind, $task, $this->client->namespace, $this->taskQueue, + fn () => $this->executeTask($taskKind, $task)); + } + + /** @param array $task */ + private function executeTask(string $taskKind, array $task): void { try { $this->assertSupportedTaskPayloadCodec($task); @@ -1272,6 +1289,7 @@ private function rejectPolledTask(string $taskKind, array $task, CodecException /** @param array $task */ private function executeWorkflowTask(array $task): void { + $traceScope = null; $this->claimCancellation = null; $this->claimDeliveredCancellationId = null; $taskId = (string) ($task['task_id'] ?? ''); @@ -1287,6 +1305,9 @@ private function executeWorkflowTask(array $task): void if (!$this->renewWorkflowTaskLease($taskId, $leaseOwner, $attempt)) { return; } + if ($this->client->observabilityContextAvailable()) { + $traceScope = $this->telemetry?->enterTask('workflow', $task, $this->client->namespace, $this->taskQueue); + } $workflowType = (string) ($task['workflow_type'] ?? ''); $updateId = isset($task['workflow_update_id']) ? (string) $task['workflow_update_id'] : null; if ($updateId !== null && $updateId !== '') { @@ -1387,6 +1408,7 @@ function (Throwable $failure) use ($taskId, $leaseOwner, $attempt): void { $attempt, ); } finally { + $traceScope?->__invoke(); $this->claimCancellation = null; $this->claimDeliveredCancellationId = null; } @@ -2189,12 +2211,13 @@ function (int $relay, int $callback) use ($taskId, $attempt, $activityType): voi }, function (RequestBudget $budget): void { $this->heartbeatIfDue($budget); }, ); - $callback = function (\Closure $heartbeat) use ($handler, $attempt, $activityType, $arguments): mixed { + $callback = function (\Closure $heartbeat) use ($handler, $attempt, $activityType, $arguments, $task): mixed { if ($handler === null) { throw new InvalidLocalActivityReport('No local activity handler is registered for '.$activityType.'.'); } - return $handler(new ActivityContext($this->client, $attempt->taskId, $attempt->attemptId, - $attempt->leaseOwner, $activityType, $attempt->attemptNumber, localHeartbeat: $heartbeat), ...$arguments); + return $this->invokeActivity($handler, new ActivityContext($this->client, $attempt->taskId, $attempt->attemptId, + $attempt->leaseOwner, $activityType, $attempt->attemptNumber, localHeartbeat: $heartbeat), $arguments, + ['activity_id' => $attempt->executionId] + ExecutionTelemetry::taskFields($task, $this->client->namespace, $this->taskQueue), true); }; return ['runner' => $runner, 'callback' => $callback]; @@ -2389,7 +2412,8 @@ function (\Closure $heartbeat) use ($handler, $taskId, $attemptId, $leaseOwner, $callbackContext = new ActivityContext($this->client, $taskId, $attemptId, $leaseOwner, $activityType, (int) ($task['attempt_number'] ?? 1), localHeartbeat: $heartbeat); - return $handler($callbackContext, ...$arguments); + return $this->invokeActivity($handler, $callbackContext, $arguments, + ExecutionTelemetry::taskFields($task, $this->client->namespace, $this->taskQueue), authorizedTask: $task); }, function (array $details) use ($taskId, $attemptId, $leaseOwner): array { try { @@ -2418,7 +2442,8 @@ static function () use (&$callbackStopped): void { $callbackStopped = true; }, ); $check(true); } else { - $result = $handler($context, ...$arguments); + $result = $this->invokeActivity($handler, $context, $arguments, + ExecutionTelemetry::taskFields($task, $this->client->namespace, $this->taskQueue), authorizedTask: $task); } $this->retryStorageAdmission( 'activity_complete', @@ -2625,6 +2650,7 @@ private function assertRemoteActivityClaimActive(array $task, ?RequestBudget $bu /** @param array $task */ private function executeQueryTask(array $task): void { + $traceScope = null; $taskId = (string) ($task['query_task_id'] ?? $task['task_id'] ?? ''); $attempt = (int) ($task['query_task_attempt'] ?? 1); $leaseOwner = (string) ($task['lease_owner'] ?? $this->workerId); @@ -2636,6 +2662,9 @@ private function executeQueryTask(array $task): void throw new WorkflowClaimRevoked('query_claim_changed', 'Loaded query identity differs from its original claim.'); } $authority?->confirm(); + if ($this->client->observabilityContextAvailable()) { + $traceScope = $this->telemetry?->enterTask('query', $task, $this->client->namespace, $this->taskQueue); + } $workflowType = (string) ($task['workflow_type'] ?? ''); $queryName = (string) ($task['query_name'] ?? ''); $handler = $this->queries[$workflowType][$queryName] ?? null; @@ -2679,6 +2708,7 @@ function (Throwable $failure) use ($taskId, $leaseOwner, $attempt, $authority): }, ); } finally { + $traceScope?->__invoke(); $this->queryTaskAuthority = null; } } @@ -3075,11 +3105,11 @@ private function executeLocalActivity( ), ]; }; - $callback = fn (\Closure $heartbeat): mixed => $handler(new ActivityContext( + $callback = fn (\Closure $heartbeat): mixed => $this->invokeActivity($handler, new ActivityContext( $this->client, (string) ($task['task_id'] ?? ''), $attemptId, (string) ($task['lease_owner'] ?? $this->workerId), $activityType, $attemptNumber, localHeartbeat: $heartbeat, - ), ...$arguments); + ), $arguments, ['activity_id' => null] + ExecutionTelemetry::taskFields($task, $this->client->namespace, $this->taskQueue), true); if ($this->enableCooperativeCancellation) { $nextRenewal = 0.0; $check = function (bool $force) use ($task, &$nextRenewal, $attemptStartedAt, @@ -3791,6 +3821,7 @@ private function assertHandlerContext(callable $handler, string $contextClass, s private function handlerFailure(string $kind, string $identity, Throwable $exception): void { + ExecutionTelemetry::failure($exception); $context = [ 'worker_id' => $this->workerId, 'handler_kind' => $kind, @@ -3804,6 +3835,29 @@ private function handlerFailure(string $kind, string $identity, Throwable $excep $this->diagnostic('worker.handler_failed', $context, 'error'); } + /** @param callable(ActivityContext, mixed ...$arguments): mixed $handler + * @param list $arguments + * @param array $fields + * @param array|null $authorizedTask + */ + private function invokeActivity(callable $handler, ActivityContext $context, array $arguments, array $fields, bool $local = false, ?array $authorizedTask = null): mixed + { + if ($this->telemetry === null) { + return $handler($context, ...$arguments); + } + $fields['activity_type'] = $context->activityType; + $fields['activity_attempt_id'] = $context->activityAttemptId; + $fields['activity_attempt_number'] = $context->attemptNumber; + + $scope = !$local && $authorizedTask !== null && $this->client->observabilityContextAvailable() + ? $this->telemetry->enterTask('activity', $authorizedTask, $this->client->namespace, $this->taskQueue) : null; + try { + return $this->telemetry->activity($fields, fn () => $handler($context, ...$arguments), $local); + } finally { + $scope?->__invoke(); + } + } + /** @return array */ private function workflowFailureCommand( Throwable $exception, diff --git a/src/Worker/Replayer.php b/src/Worker/Replayer.php index 7dc709e4..2f6b625c 100644 --- a/src/Worker/Replayer.php +++ b/src/Worker/Replayer.php @@ -12,6 +12,7 @@ use DurableWorkflow\Exception\NonDeterministicWorkflow; use DurableWorkflow\Exception\WorkflowCancelled; use DurableWorkflow\Exception\WorkflowFiberDiscarded; +use DurableWorkflow\Telemetry\ExecutionTelemetry; use Closure; use Fiber; use LogicException; @@ -125,7 +126,7 @@ public function replay( $allowScopeAuthoring = $allowCancellationScopeAuthoring && ($cancellation->request === null || $replayCommittedCancellationScopes); $context = null; - $execution = new Fiber(function () use ($handler, $history, $input, $task, $cancellation, $scopes, $localActivityExecutor, $prepareLocalActivities, $prepareLocalActivityGroups, $localActivityCancellationPolicies, $allowScopeAuthoring, $allowScopedPreparedLocalActivities, &$context): mixed { + $execution = new Fiber(ExecutionTelemetry::bind(function () use ($handler, $history, $input, $task, $cancellation, $scopes, $localActivityExecutor, $prepareLocalActivities, $prepareLocalActivityGroups, $localActivityCancellationPolicies, $allowScopeAuthoring, $allowScopedPreparedLocalActivities, &$context): mixed { $current = Fiber::getCurrent(); if ($current === null) { throw new LogicException('Workflow execution did not start inside its Fiber.'); @@ -155,7 +156,7 @@ public function replay( // No executor observes a return from a force-closed Fiber. return null; } - }); + })); $stepCursor = 0; $commands = []; diff --git a/src/WorkflowHandle.php b/src/WorkflowHandle.php index e28f4e99..f1373946 100644 --- a/src/WorkflowHandle.php +++ b/src/WorkflowHandle.php @@ -36,9 +36,9 @@ public function runs(): array } /** @param list $arguments */ - public function signal(string $name, array $arguments = []): void + public function signal(string $name, array $arguments = [], ?string $requestId = null): void { - $this->client->signalWorkflow($this->workflowId, $name, $arguments); + $this->client->signalWorkflow($this->workflowId, $name, $arguments, null, $requestId); } /** @@ -51,9 +51,9 @@ public function appendMessage(string $streamName, string $messageId, array $argu } /** @param list $arguments */ - public function signalSelectedRun(string $name, array $arguments = []): void + public function signalSelectedRun(string $name, array $arguments = [], ?string $requestId = null): void { - $this->client->signalWorkflow($this->workflowId, $name, $arguments, $this->requireSelectedRun()); + $this->client->signalWorkflow($this->workflowId, $name, $arguments, $this->requireSelectedRun(), $requestId); } /** @param list $arguments */ diff --git a/tests/ClientTelemetryTest.php b/tests/ClientTelemetryTest.php new file mode 100644 index 00000000..e8dbdfa2 --- /dev/null +++ b/tests/ClientTelemetryTest.php @@ -0,0 +1,96 @@ +getTracer('application')->spanBuilder('caller')->startSpan(); + $scope = Configurator::create()->withTracerProvider($provider)->withPropagator(TraceContextPropagator::getInstance()) + ->storeInContext($caller->storeInContext(Context::getRoot()))->activate(); + $supported = true; + $failDiscovery = false; + $transport = new FakeTransport(handler: static function (string $method, string $uri) use ($provider, &$supported, &$failDiscovery): array { + $http = $provider->getTracer('http-client')->spanBuilder('HTTP '.$method)->startSpan(); + $httpScope = $http->activate(); + try { + if (str_ends_with($uri, '/cluster/info')) { + if ($failDiscovery) { + throw new RuntimeException('discovery unavailable'); + } + return ['worker_protocol' => ['server_capabilities' => ['observability_context' => $supported ? [ + 'supported' => true, 'schema' => 'durable-workflow.trace-context', 'version' => 1, + 'dispatch_schema' => 'durable-workflow.queue-trace', 'links_schema' => 'durable-workflow.trace-links', + 'max_context_bytes' => 4096, 'max_links' => 32, 'max_links_bytes' => 4096, + ] : ['supported' => true, 'version' => 99]]]]; + } + return ['workflow_id' => 'order', 'run_id' => 'run', 'workflow_type' => 'orders', 'status' => 'pending']; + } finally { + $httpScope->detach(); + $http->end(); + } + }); + $client = (new Client('https://server.example', transport: $transport))->withTelemetry(new ExecutionTelemetry($provider)); + try { + $input = [['observability_context' => ['business' => 'preserved']]]; + $client->startWorkflow('orders', 'order', 'orders', $input); + self::assertCount(1, $transport->requests); + self::assertArrayNotHasKey('observability_context', $transport->requests[0]['body']); + $client->clusterInfo(); + self::assertTrue($client->observabilityContextAvailable()); + $client->startWorkflow('orders', 'order', 'orders', $input); + $client->signalWorkflow('order', 'ready', $input, requestId: 'signal-original'); + $client->updateWorkflowReceipt('order', 'status', $input, requestId: 'update-original'); + $client->cancelWorkflow('order', 'requested'); + $commands = array_slice($transport->requests, 2); + self::assertCount(4, $commands); + foreach ($commands as $command) { + $context = $command['body']['observability_context']; + self::assertSame('durable-workflow.trace-context', $context['schema']); + self::assertSame($caller->getContext()->getSpanId(), substr($context['carrier']['traceparent'], 36, 16)); + self::assertArrayNotHasKey('baggage', $context['carrier']); + } + self::assertSame($input, $client->payloadCodec()->decodeEnvelope($commands[0]['body']['input'])); + self::assertSame('signal-original', $commands[1]['body']['request_id']); + self::assertSame('update-original', $commands[2]['body']['request_id']); + $supported = false; + $client->clusterInfo(); + self::assertFalse($client->observabilityContextAvailable()); + $client->startWorkflow('orders', 'order', 'orders'); + self::assertArrayNotHasKey('observability_context', $transport->requests[array_key_last($transport->requests)]['body']); + $supported = true; + $client->clusterInfo(); + $failDiscovery = true; + try { + $client->clusterInfo(); + self::fail('Explicit failed discovery returned success.'); + } catch (RuntimeException $failure) { + self::assertSame('discovery unavailable', $failure->getMessage()); + } + self::assertFalse($client->observabilityContextAvailable()); + $before = count($transport->requests); + $client->startWorkflow('orders', 'order', 'orders'); + self::assertCount($before + 1, $transport->requests); + self::assertArrayNotHasKey('observability_context', $transport->requests[array_key_last($transport->requests)]['body']); + } finally { + $scope->detach(); + $caller->end(); + } + } +} diff --git a/tests/CommandRequestIdTest.php b/tests/CommandRequestIdTest.php new file mode 100644 index 00000000..908ddc1e --- /dev/null +++ b/tests/CommandRequestIdTest.php @@ -0,0 +1,303 @@ + 'request_id_conflict', + 'message' => 'Request ID approve-1 retained a command with a different intent.', + 'request_id' => 'approve-1', + 'retained_command_id' => 'cmd-9', + 'retained_command_type' => 'signal', + 'retained_outcome' => 'accepted', + 'conflict' => ['arguments'], + ]; + + public function testSignalWithoutAKeySendsAGeneratedKeyAndReturnsIt(): void + { + $transport = new FakeTransport([['accepted' => true, 'command_id' => 'cmd-1']]); + $client = new Client('https://server.example', transport: $transport); + + $receipt = $client->signalWorkflow('order-1', 'approve', ['Ada']); + + $sent = $transport->requests[0]['body']['request_id']; + self::assertMatchesRegularExpression(self::SIGNAL_KEY_PATTERN, $sent); + self::assertSame(['accepted' => true, 'command_id' => 'cmd-1', 'request_id' => $sent], $receipt); + self::assertSame('https://server.example/api/workflows/order-1/signal/approve', $transport->requests[0]['uri']); + self::assertSame(['Ada'], $client->payloadCodec()->decodeEnvelope($transport->requests[0]['body']['input'])); + } + + public function testUpdateWithoutAKeySendsAGeneratedKeyAndTheReceiptCarriesIt(): void + { + $transport = new FakeTransport([ + ['update_id' => 'u-1', 'status' => 'completed', 'result_envelope' => (new AvroPayloadCodec())->envelope('renamed')], + ['update_id' => 'u-2', 'status' => 'accepted'], + ]); + $client = new Client('https://server.example', transport: $transport); + + self::assertSame('renamed', $client->updateWorkflow('order-1', 'rename', ['Grace'])); + $first = $transport->requests[0]['body']['request_id']; + self::assertMatchesRegularExpression(self::UPDATE_KEY_PATTERN, $first); + self::assertSame('completed', $transport->requests[0]['body']['wait_for']); + + $receipt = $client->updateWorkflowReceipt('order-1', 'rename', ['Grace'], 'accepted'); + $second = $transport->requests[1]['body']['request_id']; + self::assertMatchesRegularExpression(self::UPDATE_KEY_PATTERN, $second); + self::assertNotSame($first, $second); + self::assertSame(['update_id' => 'u-2', 'status' => 'accepted', 'request_id' => $second], $receipt); + } + + public function testAnExplicitKeyIsSentVerbatimForSignalsAndUpdates(): void + { + $key = 'Order/1 Approve:Ünïcode'; + $transport = new FakeTransport([[], ['update_id' => 'u-1', 'status' => 'accepted']]); + $client = new Client('https://server.example', transport: $transport); + + $signal = $client->signalWorkflow('order-1', 'approve', ['Ada'], 'run-1', $key); + $update = $client->updateWorkflow('order-1', 'rename', ['Grace'], 'accepted', 5, $key, 'run-1'); + + self::assertSame($key, $transport->requests[0]['body']['request_id']); + self::assertSame(['request_id' => $key], $signal); + self::assertSame('https://server.example/api/workflows/order-1/runs/run-1/signal/approve', $transport->requests[0]['uri']); + self::assertSame($key, $transport->requests[1]['body']['request_id']); + self::assertSame(5, $transport->requests[1]['body']['wait_timeout_seconds']); + self::assertSame(['update_id' => 'u-1', 'status' => 'accepted', 'request_id' => $key], $update); + self::assertSame('https://server.example/api/workflows/order-1/runs/run-1/update/rename', $transport->requests[1]['uri']); + } + + public function testTheSameKeyAppearsInTheRequestBodyAndTheTransportFailure(): void + { + $transport = new FakeTransport([ + new TransportException('Signal reply lost.', transientConnectionFailure: true), + new TransportException('Update reply lost.', transientConnectionFailure: true), + ]); + $client = new Client('https://server.example', transport: $transport); + + try { + $client->signalWorkflow('order-1', 'approve', ['Ada']); + self::fail('A lost signal reply must surface as SignalFailed.'); + } catch (SignalFailed $exception) { + self::assertCount(1, $transport->requests); + self::assertSame($transport->requests[0]['body']['request_id'], $exception->requestId()); + self::assertMatchesRegularExpression(self::SIGNAL_KEY_PATTERN, (string) $exception->requestId()); + self::assertNull($exception->reason()); + self::assertSame([], $exception->details()); + self::assertFalse($exception->isRequestIdConflict()); + self::assertTrue($exception->isTransientConnectionFailure()); + } + + try { + $client->updateWorkflow('order-1', 'rename', ['Grace']); + self::fail('A lost update reply must surface as UpdateFailed.'); + } catch (UpdateFailed $exception) { + self::assertCount(2, $transport->requests); + self::assertSame($transport->requests[1]['body']['request_id'], $exception->requestId()); + self::assertMatchesRegularExpression(self::UPDATE_KEY_PATTERN, (string) $exception->requestId()); + self::assertNull($exception->reason()); + self::assertSame([], $exception->details()); + self::assertFalse($exception->isRequestIdConflict()); + self::assertTrue($exception->isTransientConnectionFailure()); + } + } + + public function testARequestIdConflictIsATypedFailureThatNamesTheRetainedCommand(): void + { + $updateConflict = ['request_id' => 'rename-1', 'retained_command_type' => 'update', 'conflict' => ['name', 'target']] + + self::CONFLICT; + $transport = new FakeTransport([ + new TransportException('Request ID approve-1 retained a different signal.', 409, self::CONFLICT), + new TransportException('Request ID rename-1 retained a different update.', 409, $updateConflict), + ]); + $client = new Client('https://server.example', transport: $transport); + + try { + $client->signalWorkflow('order-1', 'approve', ['Grace'], null, 'approve-1'); + self::fail('A conflicting signal key must surface as SignalFailed.'); + } catch (SignalFailed $exception) { + self::assertSame(409, $exception->status); + self::assertSame('request_id_conflict', $exception->reason()); + self::assertTrue($exception->isRequestIdConflict()); + self::assertSame('approve-1', $exception->requestId()); + self::assertSame(self::CONFLICT, $exception->details()); + self::assertSame('cmd-9', $exception->details()['retained_command_id']); + self::assertSame('Request ID approve-1 retained a different signal.', $exception->getMessage()); + } + + try { + $client->updateWorkflow('order-1', 'rename', ['Grace'], requestId: 'rename-1'); + self::fail('A conflicting update key must surface as UpdateFailed.'); + } catch (UpdateFailed $exception) { + self::assertSame(409, $exception->status); + self::assertSame('request_id_conflict', $exception->reason()); + self::assertTrue($exception->isRequestIdConflict()); + self::assertSame('rename-1', $exception->requestId()); + self::assertSame('cmd-9', $exception->details()['retained_command_id']); + self::assertSame('update', $exception->details()['retained_command_type']); + self::assertSame(['name', 'target'], $exception->details()['conflict']); + } + } + + public function testARetainedReceiptIsReturnedUnchangedWithItsKey(): void + { + $signalReceipt = [ + 'command_id' => 'cmd-1', + 'status' => 'received', + 'request_id' => 'approve-1', + 'retained' => true, + 'intent_verified' => true, + ]; + $updateReceipt = [ + 'update_id' => 'u-1', + 'status' => 'completed', + 'result_envelope' => (new AvroPayloadCodec())->envelope('renamed'), + 'request_id' => 'rename-1', + 'retained' => true, + 'intent_verified' => true, + ]; + $transport = new FakeTransport([$signalReceipt, $updateReceipt, $updateReceipt]); + $client = new Client('https://server.example', transport: $transport); + + self::assertSame($signalReceipt, $client->signalWorkflow('order-1', 'approve', ['Ada'], null, 'approve-1')); + self::assertSame($updateReceipt, $client->updateWorkflowReceipt('order-1', 'rename', ['Grace'], requestId: 'rename-1')); + self::assertSame('renamed', $client->updateWorkflow('order-1', 'rename', ['Grace'], requestId: 'rename-1')); + foreach ($transport->requests as $index => $request) { + self::assertSame($index === 0 ? 'approve-1' : 'rename-1', $request['body']['request_id']); + } + } + + public function testHandleMethodsPassTheKeyThrough(): void + { + $transport = new FakeTransport([[], [], [], [], []]); + $client = new Client('https://server.example', transport: $transport); + $handle = $client->workflowHandle('order/1', 'run/2'); + + $handle->signal('approve', ['Ada'], 'sig-1'); + $handle->signalSelectedRun('approve', ['Ada'], 'sig-2'); + $handle->update('rename', ['Grace'], 'accepted', 5, 'upd-1'); + $handle->updateSelectedRun('rename', ['Grace'], requestId: 'upd-2'); + $handle->signal('approve'); + + $bodies = array_column($transport->requests, 'body'); + self::assertSame(['sig-1', 'sig-2', 'upd-1', 'upd-2'], array_slice(array_column($bodies, 'request_id'), 0, 4)); + self::assertMatchesRegularExpression(self::SIGNAL_KEY_PATTERN, $bodies[4]['request_id']); + self::assertSame('accepted', $bodies[2]['wait_for']); + self::assertSame(5, $bodies[2]['wait_timeout_seconds']); + self::assertSame([ + 'https://server.example/api/workflows/order%2F1/signal/approve', + 'https://server.example/api/workflows/order%2F1/runs/run%2F2/signal/approve', + 'https://server.example/api/workflows/order%2F1/update/rename', + 'https://server.example/api/workflows/order%2F1/runs/run%2F2/update/rename', + 'https://server.example/api/workflows/order%2F1/signal/approve', + ], array_column($transport->requests, 'uri')); + } + + public function testTheKeyTravelsInTheBodyAndNeverInARequestIdHeader(): void + { + $transport = new FakeTransport([[], []]); + $client = new Client('https://server.example', transport: $transport, token: 'token', namespace: 'tenant-a'); + + $client->signalWorkflow('order-1', 'approve', ['Ada'], null, 'approve-1'); + $client->updateWorkflow('order-1', 'rename', ['Grace'], requestId: 'rename-1'); + + self::assertCount(2, $transport->requests); + foreach ($transport->requests as $request) { + self::assertNotContains('x-request-id', array_map(strtolower(...), array_keys($request['headers']))); + self::assertSame('tenant-a', $request['headers']['X-Namespace']); + self::assertArrayHasKey('request_id', $request['body']); + } + } + + public function testAnInvalidKeyIsRejectedBeforeAnyRequestIsSent(): void + { + $transport = new FakeTransport([[]]); + $client = new Client('https://server.example', transport: $transport); + + foreach (['', str_repeat('k', 256), str_repeat('é', 256), "\xFF"] as $invalid) { + foreach (['signal', 'update'] as $command) { + try { + $command === 'signal' + ? $client->signalWorkflow('order-1', 'approve', [], null, $invalid) + : $client->updateWorkflow('order-1', 'rename', [], requestId: $invalid); + self::fail("A {$command} must reject an invalid request ID."); + } catch (InvalidArgumentException $exception) { + self::assertSame(ucfirst($command).' request IDs must be from 1 to 255 Unicode characters.', $exception->getMessage()); + } + } + } + self::assertSame([], $transport->requests); + } + + public function testTheMaximumAsciiKeyIsAccepted(): void + { + $transport = new FakeTransport([[]]); + $client = new Client('https://server.example', transport: $transport); + $limit = str_repeat('k', 255); + $client->signalWorkflow('order-1', 'approve', [], null, $limit); + self::assertSame($limit, $transport->requests[0]['body']['request_id']); + } + + public function testThePublishedServerUnicodeKeyBoundaryIsPreserved(): void + { + $key = str_repeat('é', 255); + $transport = new FakeTransport([[], []]); + $client = new Client('https://server.example', transport: $transport); + $client->signalWorkflow('order-1', 'approve', [], requestId: $key); + $client->updateWorkflowReceipt('order-1', 'approve', [], requestId: $key); + self::assertSame([$key, $key], array_column(array_column($transport->requests, 'body'), 'request_id')); + } + + #[RunInSeparateProcess] + #[PreserveGlobalState(false)] + public function testPublishedCustomHandleImplementationsStillLoadAndDeliverSignals(): void + { + $handle = new class implements WorkflowHandleInterface { + /** @var list}> */ + public array $signals = []; + + public function signal(string $name, array $arguments = []): void + { + $this->signals[] = [$name, $arguments]; + } + + public function appendMessage(string $streamName, string $messageId, array $arguments = []): array + { + return []; + } + + public function query(string $name, array $arguments = []): mixed + { + return null; + } + + public function update(string $name, array $arguments = [], string $waitFor = 'completed', ?int $waitTimeoutSeconds = null, ?string $requestId = null): mixed + { + return null; + } + + public function result(float $timeoutSeconds = 30.0, float $pollIntervalSeconds = 0.5): mixed + { + return null; + } + }; + $handle->signal('approve', ['Київ']); + self::assertSame([['approve', ['Київ']]], $handle->signals); + } +} diff --git a/tests/ExecutionTelemetryTest.php b/tests/ExecutionTelemetryTest.php new file mode 100644 index 00000000..efb4c4dc --- /dev/null +++ b/tests/ExecutionTelemetryTest.php @@ -0,0 +1,455 @@ +instrumentation(); + $caller = $provider->getTracer('application')->spanBuilder('request')->startSpan(); + $callerScope = Configurator::create()->withTracerProvider($provider) + ->storeInContext($caller->storeInContext(Context::getRoot()))->activate(); + $handler = new TestHandler(); + $logger = new Logger('application', [$handler], [new LogContext()]); + try { + $result = $telemetry->task('activity', self::task(), 'tenant-a', 'orders', function () use ($telemetry, $provider, $logger): object { + $task = ExecutionTelemetry::current(); + self::assertSame('run-1', $task['workflow_run_id']); + self::assertSame('logical-activity-1', $task['activity_id']); + return $telemetry->activity(['activity_attempt_id' => 'attempt-2', 'activity_attempt_number' => 2], + function () use ($provider, $logger, $task): object { + $attempt = ExecutionTelemetry::current(); + self::assertSame('attempt-2', $attempt['activity_attempt_id']); + self::assertNotSame($task['span_id'], $attempt['span_id']); + $http = $provider->getTracer('http-client')->spanBuilder('http.request')->startSpan(); + $scope = $http->activate(); + try { + $logger->info('application message', ['request_id' => 'application-request', 'custom' => 'kept']); + self::assertSame($http->getContext()->getSpanId(), ExecutionTelemetry::current()['span_id']); + } finally { + $scope->detach(); + $http->end(); + } + self::assertSame($attempt, ExecutionTelemetry::current()); + + return new \stdClass(); + }); + }); + self::assertInstanceOf(\stdClass::class, $result); + self::assertSame([], ExecutionTelemetry::current()); + self::assertSame($caller->getContext()->getSpanId(), Span::getCurrent()->getContext()->getSpanId()); + $records = $handler->getRecords(); + $record = $records[0]; + $context = $record instanceof \Monolog\LogRecord ? $record->context : $record['context']; + self::assertSame('application-request', $context['request_id']); + self::assertSame('run-1', $context['workflow_run_id']); + self::assertSame('logical-activity-1', $context['activity_id']); + self::assertSame('attempt-2', $context['activity_attempt_id']); + self::assertSame(2, $context['activity_attempt_number']); + self::assertSame('kept', $context['custom']); + $spans = $exporter->getSpans(); + self::assertSame('http.request', $spans[0]->getName()); + self::assertSame($spans[1]->getContext()->getSpanId(), $spans[0]->getParentContext()->getSpanId()); + self::assertSame($spans[2]->getContext()->getSpanId(), $spans[1]->getParentContext()->getSpanId()); + self::assertFalse($spans[2]->getParentContext()->isValid()); + self::assertNotSame($caller->getContext()->getTraceId(), $spans[2]->getContext()->getTraceId()); + } finally { + $callerScope->detach(); + $caller->end(); + } + } + + public function testBoundWorkflowFibersKeepSeparateContextsAcrossSuspensionAndUnwinding(): void + { + [$telemetry] = $this->instrumentation(); + $fibers = []; + foreach (['run-a', 'run-b'] as $runId) { + $fibers[] = new Fiber(function () use ($telemetry, $runId): void { + $telemetry->task('workflow', ['run_id' => $runId], 'tenant-a', 'orders', static function () use ($runId): void { + self::assertSame($runId, ExecutionTelemetry::current()['workflow_run_id']); + $unbound = new Fiber(static fn (): array => ExecutionTelemetry::current()); + $unbound->start(); + self::assertSame([], $unbound->getReturn()); + Fiber::suspend(); + self::assertSame($runId, ExecutionTelemetry::current()['workflow_run_id']); + }); + }); + } + $fibers[0]->start(); + self::assertSame([], ExecutionTelemetry::current()); + $fibers[1]->start(); + self::assertSame([], ExecutionTelemetry::current()); + $fibers[0]->resume(); + $failure = new RuntimeException('private callback message'); + try { + $fibers[1]->throw($failure); + self::fail('Fiber failure was swallowed.'); + } catch (RuntimeException $caught) { + self::assertSame($failure, $caught); + } + self::assertSame([], ExecutionTelemetry::current()); + } + + public function testReplayerBindsContextBeforeResolvingAndCallingWorkflowHandler(): void + { + [$telemetry, , $exporter] = $this->instrumentation(); + $client = new Client('https://server.example'); + $result = $telemetry->task('workflow', self::task(), 'tenant-a', 'orders', + static fn () => (new Replayer($client->payloadCodec()))->replay( + static function (WorkflowContext $context): string { + self::assertSame('run-1', ExecutionTelemetry::current()['workflow_run_id']); + self::assertArrayNotHasKey('activity_id', ExecutionTelemetry::current()); + + return 'workflow-result'; + }, [], [], 'orders', self::task())); + self::assertSame('workflow-result', $client->payloadCodec()->decodeEnvelope($result->commands[0]['result'])); + self::assertSame([], ExecutionTelemetry::current()); + self::assertCount(1, $exporter->getSpans()); + self::assertSame('durable.workflow.task', $exporter->getSpans()[0]->getName()); + } + + public function testFreshWorkflowFactoriesResolveInsideEachActualTaskContext(): void + { + [$telemetry, , $exporter] = $this->instrumentation(); + $polls = 0; + $transport = new FakeTransport(handler: static function (string $method, string $uri, array $headers, ?array $body) use (&$polls): ?array { + if (str_ends_with($uri, '/workflow-tasks/poll')) { + ++$polls; + + return ['task' => ['task_id' => 'workflow-task-'.$polls, 'workflow_id' => 'workflow-'.$polls, + 'run_id' => 'run-'.$polls, 'workflow_type' => 'telemetry.fresh-workflow', + 'workflow_task_attempt' => 1, 'payload_codec' => 'avro', 'history_events' => []], + 'poll_status' => 'leased']; + } + if (str_ends_with($uri, '/activity-tasks/poll') || str_ends_with($uri, '/query-tasks/poll')) { + return ['task' => null, 'poll_status' => 'empty']; + } + if (str_ends_with($uri, '/complete')) { + return ['recorded' => true]; + } + if (str_ends_with($uri, '/heartbeat')) { + return ['renewed' => true, 'task_id' => 'workflow-task-'.$polls, + 'lease_owner' => $body['lease_owner'], 'workflow_task_attempt' => $body['workflow_task_attempt']]; + } + if (str_ends_with($uri, '/fail')) { + throw new \LogicException('Unexpected workflow failure: '.($body['failure']['message'] ?? 'missing failure message')); + } + throw new \LogicException('Unexpected request '.$method.' '.$uri); + }); + $constructions = []; + $shared = new \stdClass(); + $worker = Worker::create(new Client('https://server.example', transport: $transport), 'orders', telemetry: $telemetry) + ->register(new WorkflowFactory(static function () use (&$constructions, $shared): TelemetryFreshWorkflow { + $constructions[] = ExecutionTelemetry::current(); + + return new TelemetryFreshWorkflow($shared); + })); + self::assertTrue($worker->tick(0)); + self::assertTrue($worker->tick(0)); + $scoped = array_values(array_filter($constructions, static fn (array $fields): bool => isset($fields['workflow_run_id']))); + self::assertSame(['run-1', 'run-2'], array_column($scoped, 'workflow_run_id')); + $completions = array_values(array_filter($transport->requests, + static fn (array $request): bool => str_ends_with($request['uri'], '/complete'))); + self::assertCount(2, $completions); + $codec = (new Client('https://server.example'))->payloadCodec(); + foreach ($completions as $completion) { + self::assertSame(1, $codec->decodeEnvelope($completion['body']['commands'][0]['result'])); + } + self::assertCount(2, $exporter->getSpans()); + self::assertSame([], ExecutionTelemetry::current()); + } + + public function testPrivacyAndOriginalExceptionIdentity(): void + { + [$telemetry, , $exporter] = $this->instrumentation(); + $failure = new RuntimeException('secret credential and customer payload'); + try { + $telemetry->task('activity', self::task() + ['arguments' => ['secret'], 'authorization' => 'secret'], + 'tenant-a', 'orders', static function () use ($failure): never { throw $failure; }); + self::fail('Business failure was swallowed.'); + } catch (RuntimeException $caught) { + self::assertSame($failure, $caught); + } + $span = $exporter->getSpans()[0]; + self::assertSame(StatusCode::STATUS_ERROR, $span->getStatus()->getCode()); + $attributes = $span->getAttributes()->toArray(); + self::assertSame(RuntimeException::class, $attributes['error.type']); + self::assertStringNotContainsString('secret', json_encode($attributes, JSON_THROW_ON_ERROR)); + self::assertSame([], $span->getEvents()); + self::assertSame([], ExecutionTelemetry::current()); + } + + public function testDisabledAndFailingProvidersExecuteExactlyOnceWithoutChangingTheResult(): void + { + $provider = $this->createMock(TracerProviderInterface::class); + $provider->expects(self::once())->method('getTracer')->willThrowException(new RuntimeException('provider unavailable')); + $calls = 0; + $result = new \stdClass(); + foreach ([new ExecutionTelemetry($provider, enabled: false), new ExecutionTelemetry($provider)] as $telemetry) { + self::assertSame($result, $telemetry->task('activity', self::task(), 'tenant-a', 'orders', + static function () use (&$calls, $result): object { ++$calls; return $result; })); + self::assertSame([], ExecutionTelemetry::current()); + } + self::assertSame(2, $calls); + } + + #[DataProvider('workerSurfaces')] + public function testSequentialWorkerAttemptsKeepLogicalIdentityAndWireOutcomes(bool $factory): void + { + [$telemetry, , $exporter] = $this->instrumentation(); + $polls = 0; + $transport = new FakeTransport(handler: static function (string $method, string $uri) use (&$polls): ?array { + if (str_ends_with($uri, '/workflow-tasks/poll') || str_ends_with($uri, '/query-tasks/poll')) { + return ['task' => null, 'poll_status' => 'empty']; + } + if (str_ends_with($uri, '/activity-tasks/poll')) { + ++$polls; + + return ['task' => array_replace(self::task(), ['task_id' => 'task-'.$polls, + 'activity_attempt_id' => 'attempt-'.$polls, 'attempt_number' => $polls]), 'poll_status' => 'leased']; + } + if (str_ends_with($uri, '/fail') || str_ends_with($uri, '/complete')) { + return ['recorded' => true]; + } + throw new \LogicException('Unexpected request '.$method.' '.$uri); + }); + $seen = []; + $client = new Client('https://server.example', namespace: 'tenant-a', transport: $transport); + $worker = $factory ? Worker::create($client, 'orders', telemetry: $telemetry) + : new Worker($client, 'orders', workerId: 'worker-1', clock: static fn (): float => 0, telemetry: $telemetry); + $worker->registerActivity('orders.charge', static function (ActivityContext $context) use (&$seen): string { + $seen[] = ExecutionTelemetry::current(); + if ($context->attemptNumber === 1) { + throw new RuntimeException('genuine body failure'); + } + + return 'complete'; + }); + self::assertTrue($worker->tick(0)); + self::assertSame([], ExecutionTelemetry::current()); + self::assertTrue($worker->tick(0)); + self::assertSame([], ExecutionTelemetry::current()); + self::assertCount(2, $seen); + self::assertSame('logical-activity-1', $seen[0]['activity_id']); + self::assertSame($seen[0]['activity_id'], $seen[1]['activity_id']); + self::assertSame('attempt-1', $seen[0]['activity_attempt_id']); + self::assertSame('attempt-2', $seen[1]['activity_attempt_id']); + self::assertNotSame($seen[0]['span_id'], $seen[1]['span_id']); + self::assertNotSame($seen[0]['trace_id'], $seen[1]['trace_id']); + $outcomes = array_values(array_filter($transport->requests, + static fn (array $r): bool => str_ends_with($r['uri'], '/fail') || str_ends_with($r['uri'], '/complete'))); + self::assertCount(2, $outcomes); + self::assertStringEndsWith('/task-1/fail', $outcomes[0]['uri']); + self::assertStringEndsWith('/task-2/complete', $outcomes[1]['uri']); + self::assertSame('attempt-1', $outcomes[0]['body']['activity_attempt_id']); + self::assertSame('attempt-2', $outcomes[1]['body']['activity_attempt_id']); + self::assertCount(4, $exporter->getSpans()); + } + + public static function workerSurfaces(): iterable + { + yield 'constructor' => [false]; + yield 'class-oriented factory' => [true]; + } + + public function testOptionalApiAndMonologCanBothBeAbsent(): void + { + $pipes = []; + $process = proc_open([PHP_BINARY, __DIR__.'/Fixtures/telemetry-without-api.php'], + [1 => ['pipe', 'w'], 2 => ['pipe', 'w']], $pipes); + self::assertIsResource($process); + $stdout = stream_get_contents($pipes[1]); + $stderr = stream_get_contents($pipes[2]); + fclose($pipes[1]); + fclose($pipes[2]); + self::assertSame(0, proc_close($process), $stderr); + self::assertSame("optional-api-absent-pass\n", $stdout); + self::assertSame('', $stderr); + } + + public function testArrayLogRecordsPreserveApplicationRequestAndOutsideLogs(): void + { + [$telemetry] = $this->instrumentation(); + $processor = new LogContext(); + $record = ['message' => 'message', 'context' => ['request_id' => 'application-request']]; + self::assertSame($record, $processor($record)); + $actual = $telemetry->task('activity', self::task(), 'tenant-a', 'orders', static fn () => $processor($record)); + self::assertSame('application-request', $actual['context']['request_id']); + self::assertSame('run-1', $actual['context']['workflow_run_id']); + self::assertSame('logical-activity-1', $actual['context']['activity_id']); + self::assertSame($record, $processor($record)); + } + + public function testCooperativeCallbackUsesFreshProviderAndExportsBeforeProcessExit(): void + { + self::assertTrue(CooperativeActivityExecutor::available()); + $path = tempnam(sys_get_temp_dir(), 'wf773-child-'); + self::assertIsString($path); + [$owner, $provider, $ownerExporter] = $this->instrumentation(); + $telemetry = new ExecutionTelemetry($provider, childProviderFactory: static function () use ($path): TracerProvider { + $exporter = new class($path) extends InMemoryExporter { + public function __construct(private readonly string $path) { parent::__construct(); } + protected function doExport(iterable $spans): bool + { + foreach ($spans as $span) { + file_put_contents($this->path, json_encode(['pid' => getmypid(), 'name' => $span->getName(), + 'span_id' => $span->getContext()->getSpanId(), + 'parent_span_id' => $span->getParentContext()->getSpanId()], JSON_THROW_ON_ERROR)); + } + return true; + } + }; + return new TracerProvider(new SimpleSpanProcessor($exporter)); + }); + try { + $codec = (new Client('https://server.example'))->payloadCodec(); + $fields = $owner->task('activity', self::task(), 'tenant-a', 'orders', + static fn () => (new CooperativeActivityExecutor($codec))->execute( + static fn (\Closure $heartbeat) => $telemetry->activity( + ExecutionTelemetry::taskFields(self::task(), 'tenant-a', 'orders'), + static fn () => ExecutionTelemetry::current()), + static fn (array $details): array => [], static function (bool $force): void {})); + $exported = json_decode(file_get_contents($path), true, flags: JSON_THROW_ON_ERROR); + self::assertNotSame(getmypid(), $exported['pid']); + self::assertSame('durable.activity.attempt', $exported['name']); + self::assertSame($fields['span_id'], $exported['span_id']); + self::assertSame('logical-activity-1', $fields['activity_id']); + self::assertCount(1, $ownerExporter->getSpans()); + self::assertSame($ownerExporter->getSpans()[0]->getContext()->getSpanId(), $exported['parent_span_id']); + self::assertSame([], ExecutionTelemetry::current()); + } finally { + unlink($path); + } + } + + public function testMissingOrFailingChildFactoryCannotReuseTheOwnerProvider(): void + { + self::assertTrue(CooperativeActivityExecutor::available()); + [$owner, $provider, $exporter] = $this->instrumentation(); + $codec = (new Client('https://server.example'))->payloadCodec(); + foreach ([null, static function (): never { throw new RuntimeException('child setup failed'); }] as $factory) { + $telemetry = new ExecutionTelemetry($provider, childProviderFactory: $factory); + $actual = $owner->task('activity', self::task(), 'tenant-a', 'orders', + static fn () => (new CooperativeActivityExecutor($codec))->execute( + static fn (\Closure $heartbeat) => $telemetry->activity( + ExecutionTelemetry::taskFields(self::task(), 'tenant-a', 'orders'), static function (): array { + $span = Globals::tracerProvider()->getTracer('application')->spanBuilder('external-call')->startSpan(); + return ['recording' => $span->isRecording(), 'fields' => ExecutionTelemetry::current()]; + }), static fn (array $details): array => [], static function (bool $force): void {})); + self::assertFalse($actual['recording']); + self::assertSame('run-1', $actual['fields']['workflow_run_id']); + self::assertArrayNotHasKey('trace_id', $actual['fields']); + self::assertSame([], ExecutionTelemetry::current()); + } + self::assertCount(2, $exporter->getSpans()); + } + + public function testExporterEndFailureCannotReplaceCallbackResultOrException(): void + { + $span = $this->createMock(SpanInterface::class); + $span->method('getContext')->willReturn(SpanContext::getInvalid()); + $span->method('storeInContext')->willReturn(Context::getRoot()); + $span->expects(self::exactly(2))->method('end')->willThrowException(new RuntimeException('export failed')); + $builder = $this->createMock(SpanBuilderInterface::class); + foreach (['setSpanKind', 'setParent', 'setAttributes'] as $method) { + $builder->method($method)->willReturnSelf(); + } + $builder->method('startSpan')->willReturn($span); + $tracer = $this->createMock(TracerInterface::class); + $tracer->method('spanBuilder')->willReturn($builder); + $provider = $this->createMock(TracerProviderInterface::class); + $provider->method('getTracer')->willReturn($tracer); + $telemetry = new ExecutionTelemetry($provider); + $result = new \stdClass(); + self::assertSame($result, $telemetry->task('activity', self::task(), 'tenant-a', 'orders', static fn () => $result)); + $failure = new RuntimeException('original business failure'); + try { + $telemetry->task('activity', self::task(), 'tenant-a', 'orders', static function () use ($failure): never { throw $failure; }); + self::fail('Original failure was swallowed.'); + } catch (RuntimeException $caught) { + self::assertSame($failure, $caught); + } + self::assertSame([], ExecutionTelemetry::current()); + } + + public function testUntrustedIdentityFieldsAreBoundedAndResumeMetadataIsNotAnActiveActivity(): void + { + [$telemetry, , $exporter] = $this->instrumentation(); + $fields = $telemetry->task('workflow', array_replace(self::task(), [ + 'workflow_type' => "orders\nsecret", 'request_id' => str_repeat('r', 257), + 'run_id' => ['malformed'], 'attempt_number' => '2', + ]), 'tenant-a', 'orders', static fn () => ExecutionTelemetry::current()); + self::assertSame('workflow-1', $fields['workflow_instance_id']); + foreach (['workflow_type', 'request_id', 'workflow_run_id', 'activity_id', 'activity_type', + 'activity_attempt_id', 'activity_attempt_number'] as $name) { + self::assertArrayNotHasKey($name, $fields); + self::assertArrayNotHasKey($name, $exporter->getSpans()[0]->getAttributes()->toArray()); + } + } + + /** @return array */ + private static function task(): array + { + return ['task_id' => 'task-1', 'workflow_id' => 'workflow-1', 'run_id' => 'run-1', + 'workflow_type' => 'orders.workflow', 'activity_execution_id' => 'logical-activity-1', + 'activity_attempt_id' => 'attempt-1', 'attempt_number' => 1, 'activity_type' => 'orders.charge', + 'lease_owner' => 'worker-1', 'payload_codec' => 'avro', 'request_id' => 'transport-request']; + } + + /** @return array{ExecutionTelemetry, TracerProvider, InMemoryExporter} */ + private function instrumentation(): array + { + $exporter = new InMemoryExporter(); + $provider = new TracerProvider(new SimpleSpanProcessor($exporter)); + + return [new ExecutionTelemetry($provider), $provider, $exporter]; + } +} + +final class TelemetryFreshWorkflow +{ + private int $calls = 0; + + public function __construct(public readonly object $shared) + { + } + + #[\DurableWorkflow\Attribute\Workflow('telemetry.fresh-workflow')] + public function run(WorkflowContext $context): int + { + return ++$this->calls; + } +} diff --git a/tests/Fixtures/telemetry-without-api.php b/tests/Fixtures/telemetry-without-api.php new file mode 100644 index 00000000..2e830272 --- /dev/null +++ b/tests/Fixtures/telemetry-without-api.php @@ -0,0 +1,24 @@ +task('activity', [], 'default', 'orders', + static function () use (&$calls): string { ++$calls; return 'business-result'; }); +$record = ['message' => 'application message', 'context' => ['request_id' => 'application-request']]; +if ($result !== 'business-result' || $calls !== 1 + || \DurableWorkflow\Telemetry\ExecutionTelemetry::current() !== [] + || (new \DurableWorkflow\Telemetry\LogContext())($record) !== $record) { + throw new \RuntimeException('Optional API absence changed execution or logging.'); +} +echo "optional-api-absent-pass\n"; diff --git a/tests/Support/ReplayRegressionFixture.php b/tests/Support/ReplayRegressionFixture.php index 79d635cb..d72c29cf 100644 --- a/tests/Support/ReplayRegressionFixture.php +++ b/tests/Support/ReplayRegressionFixture.php @@ -122,7 +122,7 @@ public static function execute(array $fixture): array } : null; try { - $result = (new Replayer($codec))->replay( + $execute = static fn () => (new Replayer($codec))->replay( self::workflow($workflowType), $history, $input, @@ -134,6 +134,10 @@ public static function execute(array $fixture): array allowCancellationScopeAuthoring: $scopedCleanup, replayCommittedCancellationScopes: $scopedCleanup, ); + $result = $workflowType === 'golden.telemetry-context' + ? (new \DurableWorkflow\Telemetry\ExecutionTelemetry())->task('workflow', $task, + 'default', 'regression-corpus', $execute) + : $execute(); gc_collect_cycles(); $commands = array_map( static fn (array $command): array => self::decodeEnvelopes($command, $codec), @@ -325,6 +329,14 @@ private static function executeWorkflowTasks( private static function workflow(string $workflowType): callable { return match ($workflowType) { + 'golden.telemetry-context' => static function (WorkflowContext $context): array { + $result = $context->activity('golden.greet', ['Ada']); + $fields = \DurableWorkflow\Telemetry\ExecutionTelemetry::current(); + + return ['correlated' => ($fields['workflow_instance_id'] ?? null) === $context->workflowId + && ($fields['workflow_run_id'] ?? null) === $context->runId, + 'namespace' => $fields['namespace'] ?? null, 'result' => $result]; + }, 'golden.entry-arguments-required' => static fn (WorkflowContext $context, int $amount, int $required): int => $amount + $required, 'golden.nexus-caller' => static function (WorkflowContext $context): mixed { return $context->serviceOperation('billing', 'invoices', 'create', ['invoice_id' => 42], diff --git a/tests/TraceCarrierTest.php b/tests/TraceCarrierTest.php new file mode 100644 index 00000000..a6b0b81c --- /dev/null +++ b/tests/TraceCarrierTest.php @@ -0,0 +1,113 @@ + true, 'schema' => 'durable-workflow.trace-context', 'version' => 1, + 'dispatch_schema' => 'durable-workflow.queue-trace', 'links_schema' => 'durable-workflow.trace-links', + 'max_context_bytes' => 4096, 'max_links' => 32, 'max_links_bytes' => 4096]; + $info = ['worker_protocol' => ['server_capabilities' => ['observability_context' => $capability]]]; + self::assertTrue(TraceCarrier::supported($info)); + foreach (array_keys($capability) as $key) { + foreach ([null, false, '1', 1.0, [], 'unknown'] as $invalid) { + $candidate = $info; + $candidate['worker_protocol']['server_capabilities']['observability_context'][$key] = $invalid; + self::assertFalse(TraceCarrier::supported($candidate)); + } + } + self::assertFalse(TraceCarrier::supported(['version' => '999.0.0'])); + self::assertFalse(TraceCarrier::supported(['worker_protocol' => ['server_capabilities' => ['observability_context' => true]]])); + } + + public function testNormalizationUsesW3cValidationAndNeverFallsBackToTheCaller(): void + { + $caller = TraceContextPropagator::getInstance()->extract(['traceparent' => self::PARENT], context: Context::getRoot()); + $scope = $caller->activate(); + try { + foreach ([ + '00-00000000000000000000000000000000-1234567890123456-01', + '00-12345678901234567890123456789012-0000000000000000-01', + 'ff-12345678901234567890123456789012-1234567890123456-01', + 'private-header', str_repeat('x', 257), + ] as $invalid) { + self::assertNull(TraceCarrier::normalize(self::envelope(['traceparent' => $invalid]), diagnostic: $diagnostic)); + self::assertSame('discarded', $diagnostic); + self::assertSame(substr(self::PARENT, 3, 32), Span::getCurrent()->getContext()->getTraceId()); + } + self::assertSame(self::envelope(), TraceCarrier::normalize(self::envelope(), diagnostic: $diagnostic)); + self::assertNull($diagnostic); + } finally { + $scope->detach(); + } + } + + public function testBaggageDefaultsEmptyAndAllowlistRetainsAtMost32Entries(): void + { + $value = self::envelope(['baggage' => 'tenant=public,credential=private']); + self::assertSame(self::envelope(), TraceCarrier::normalize($value, diagnostic: $diagnostic)); + self::assertSame('filtered', $diagnostic); + self::assertSame(self::envelope(['baggage' => 'tenant=public']), TraceCarrier::normalize($value, ['tenant'], $diagnostic)); + self::assertSame('filtered', $diagnostic); + self::assertSame('tenant=public,credential=private', $value['carrier']['baggage']); + $keys = array_map(static fn (int $number): string => 'k'.$number, range(1, 33)); + $members = implode(',', array_map(static fn (string $key): string => $key.'=v', $keys)); + self::assertNull(TraceCarrier::normalize(self::envelope(['baggage' => $members]), $keys, $diagnostic)); + self::assertSame('discarded', $diagnostic); + self::assertNotNull(TraceCarrier::normalize(self::envelope(['baggage' => substr($members, 0, (int) strrpos($members, ','))]), $keys)); + } + + public function testByteBoundsAndUnknownFieldsAreDiscardedWithoutChangingInput(): void + { + foreach (['traceparent' => 257, 'tracestate' => 513, 'baggage' => 2049] as $key => $length) { + $value = self::envelope([$key => str_repeat('x', $length)]); + self::assertNull(TraceCarrier::normalize($value, diagnostic: $diagnostic)); + self::assertSame('discarded', $diagnostic); + self::assertSame(str_repeat('x', $length), $value['carrier'][$key]); + } + foreach ([null, [], true, 'carrier', self::envelope() + ['unknown' => 'private']] as $invalid) { + self::assertNull(TraceCarrier::normalize($invalid, diagnostic: $diagnostic)); + self::assertSame($invalid === null ? null : 'discarded', $diagnostic); + } + self::assertNull(TraceCarrier::normalize(self::envelope(['authorization' => 'private']), diagnostic: $diagnostic)); + self::assertSame('discarded', $diagnostic); + } + + public function testDispatchRequiresExactTaskBindingAndForbidsBaggage(): void + { + $value = ['schema' => 'durable-workflow.queue-trace', 'version' => 1, 'task_id' => 'task', 'context' => self::envelope()]; + self::assertSame(self::envelope(), TraceCarrier::dispatch($value, 'task', $diagnostic)); + self::assertNull($diagnostic); + self::assertNull(TraceCarrier::dispatch($value, 'other', $diagnostic)); + self::assertSame('discarded', $diagnostic); + $value['context'] = self::envelope(['baggage' => 'tenant=public']); + self::assertNull(TraceCarrier::dispatch($value, 'task', $diagnostic)); + self::assertSame('discarded', $diagnostic); + foreach (['', str_repeat('x', 513), "task\n", "task\u{0085}", "\xff"] as $invalid) { + self::assertFalse(TraceCarrier::identity($invalid)); + } + self::assertTrue(TraceCarrier::identity(str_repeat('x', 512))); + self::assertTrue(TraceCarrier::identity('任务')); + } + + /** @param array $headers + * @return array{schema: string, version: int, carrier: array} + */ + private static function envelope(array $headers = []): array + { + return ['schema' => 'durable-workflow.trace-context', 'version' => 1, + 'carrier' => array_replace(['traceparent' => self::PARENT], $headers)]; + } +} diff --git a/tests/TraceLinksTest.php b/tests/TraceLinksTest.php new file mode 100644 index 00000000..c7f49632 --- /dev/null +++ b/tests/TraceLinksTest.php @@ -0,0 +1,86 @@ + */ + private static function envelope(): array + { + $context = ['schema' => 'durable-workflow.trace-context', 'version' => 1, + 'carrier' => ['traceparent' => '00-12345678901234567890123456789012-1234567890123456-01']]; + + return ['schema' => 'durable-workflow.trace-links', 'version' => 1, 'task_id' => 'task', 'run_id' => 'run', + 'namespace' => 'tenant', 'links' => [ + ['kind' => 'task_outcome', 'task_id' => 'producer', 'event_id' => 'event', 'sequence' => 3, + 'event_type' => 'ActivityCompleted', 'activity_execution_id' => 'activity', + 'activity_attempt_id' => 'attempt', 'context' => $context], + ['kind' => 'child_outcome', 'workflow_link_id' => 'link', 'child_run_id' => 'child', + 'child_task_id' => null, 'event_id' => 'terminal', 'sequence' => 7, + 'event_type' => 'WorkflowCancelled', 'context' => $context], + ]]; + } +} diff --git a/tests/WorkerTelemetryWireTest.php b/tests/WorkerTelemetryWireTest.php new file mode 100644 index 00000000..1a23d134 --- /dev/null +++ b/tests/WorkerTelemetryWireTest.php @@ -0,0 +1,161 @@ + 'tenant=public,credential=private']); + $task['dispatch_observability_context'] = ['schema' => 'durable-workflow.queue-trace', 'version' => 1, + 'task_id' => 'task', 'context' => self::context(self::DISPATCH)]; + $task['observability_links'] = ['schema' => 'durable-workflow.trace-links', 'version' => 1, + 'task_id' => 'task', 'run_id' => 'run', 'namespace' => 'tenant-a', 'links' => [ + ['kind' => 'command', 'command_id' => 'accepted-command', 'context' => self::context(self::DISPATCH)], + ['kind' => 'child_outcome', 'workflow_link_id' => 'child-link', 'child_run_id' => 'child-run', + 'child_task_id' => null, 'event_id' => 'child-terminal', 'sequence' => 7, + 'event_type' => 'WorkflowCancelled', 'context' => self::context(self::DISPATCH)], + ]]; + [$client, $transport] = self::client($task); + $client->workerClusterInfo(); + $worker = (new Worker($client, 'orders', workerId: 'worker', telemetry: $telemetry)) + ->registerWorkflow('orders', static function (WorkflowContext $context): string { + self::assertSame('run', ExecutionTelemetry::current()['workflow_run_id']); + self::assertSame('public', Baggage::getCurrent()->getValue('tenant')); + self::assertNull(Baggage::getCurrent()->getValue('credential')); + + return 'original-result'; + }); + self::assertTrue($worker->tick(0)); + $spans = $exporter->getSpans(); + self::assertCount(1, $spans); + self::assertSame(substr(self::ORIGIN, 3, 32), $spans[0]->getContext()->getTraceId()); + self::assertSame(substr(self::ORIGIN, 36, 16), $spans[0]->getParentContext()->getSpanId()); + self::assertCount(3, $spans[0]->getLinks()); + foreach ($spans[0]->getLinks() as $link) { + self::assertSame(substr(self::DISPATCH, 3, 32), $link->getSpanContext()->getTraceId()); + } + $completions = array_values(array_filter($transport->requests, + static fn (array $request): bool => str_ends_with($request['uri'], '/complete'))); + self::assertCount(1, $completions); + self::assertSame('original-result', $client->payloadCodec()->decodeEnvelope($completions[0]['body']['commands'][0]['result'])); + self::assertSame([], ExecutionTelemetry::current()); + } + + public function testRefusedClaimNeverOpensARecoveredScopeOrInvokesTheWorkflow(): void + { + $exporter = new InMemoryExporter(); + $provider = new TracerProvider(new SimpleSpanProcessor($exporter)); + $task = self::task(); + $task['observability_context'] = self::context(self::ORIGIN); + [$client] = self::client($task, renewed: false); + $client->workerClusterInfo(); + $calls = 0; + $worker = (new Worker($client, 'orders', workerId: 'worker', telemetry: new ExecutionTelemetry($provider))) + ->registerWorkflow('orders', static function (WorkflowContext $context) use (&$calls): string { + ++$calls; + + return 'must-not-run'; + }); + try { + $worker->tick(0); + self::fail('Original lease refusal was swallowed.'); + } catch (ServerException $failure) { + self::assertSame('invalid_workflow_task_lease_response', $failure->reason); + } + self::assertSame(0, $calls); + self::assertSame([], $exporter->getSpans()); + self::assertSame([], ExecutionTelemetry::current()); + } + + public function testMalformedObservationDoesNotReplaceTheBusinessOutcome(): void + { + $exporter = new InMemoryExporter(); + $provider = new TracerProvider(new SimpleSpanProcessor($exporter)); + $task = self::task(); + $task['observability_context'] = ['schema' => 'unknown', 'version' => 9, 'carrier' => ['traceparent' => 'private']]; + $task['dispatch_observability_context'] = ['schema' => 'durable-workflow.queue-trace', 'version' => 1, + 'task_id' => 'wrong-task', 'context' => self::context(self::DISPATCH)]; + [$client] = self::client($task); + $client->workerClusterInfo(); + $worker = (new Worker($client, 'orders', workerId: 'worker', telemetry: new ExecutionTelemetry($provider))) + ->registerWorkflow('orders', static fn (WorkflowContext $context): string => 'original-result'); + self::assertTrue($worker->tick(0)); + $span = $exporter->getSpans()[0]; + self::assertFalse($span->getParentContext()->isValid()); + self::assertSame([], $span->getLinks()); + self::assertSame('discarded', $span->getAttributes()->get('durable.observability_context')); + self::assertSame('discarded', $span->getAttributes()->get('durable.dispatch_context')); + self::assertStringNotContainsString('private', json_encode($span->getAttributes()->toArray(), JSON_THROW_ON_ERROR)); + self::assertSame([], ExecutionTelemetry::current()); + } + + /** @return array */ + private static function task(): array + { + return ['task_id' => 'task', 'workflow_id' => 'workflow', 'run_id' => 'run', 'workflow_type' => 'orders', + 'workflow_task_attempt' => 1, 'lease_owner' => 'worker', 'payload_codec' => 'avro', 'history_events' => []]; + } + + /** @param array $headers + * @return array + */ + private static function context(string $parent, array $headers = []): array + { + return ['schema' => 'durable-workflow.trace-context', 'version' => 1, + 'carrier' => ['traceparent' => $parent] + $headers]; + } + + /** @param array $task + * @return array{Client, FakeTransport} + */ + private static function client(array $task, bool $renewed = true): array + { + $transport = new FakeTransport(handler: static function (string $method, string $uri, array $headers, ?array $body) use ($task, $renewed): array { + if (str_ends_with($uri, '/cluster/info')) { + return ['worker_protocol' => ['server_capabilities' => ['observability_context' => [ + 'supported' => true, 'schema' => 'durable-workflow.trace-context', 'version' => 1, + 'dispatch_schema' => 'durable-workflow.queue-trace', 'links_schema' => 'durable-workflow.trace-links', + 'max_context_bytes' => 4096, 'max_links' => 32, 'max_links_bytes' => 4096, + ]]]]; + } + if (str_ends_with($uri, '/workflow-tasks/poll')) { + return ['task' => $task, 'poll_status' => 'leased']; + } + if (str_ends_with($uri, '/activity-tasks/poll') || str_ends_with($uri, '/query-tasks/poll')) { + return ['task' => null, 'poll_status' => 'empty']; + } + if (str_ends_with($uri, '/heartbeat')) { + return ['renewed' => $renewed, 'task_id' => $task['task_id'], 'lease_owner' => $body['lease_owner'], + 'workflow_task_attempt' => $body['workflow_task_attempt'], 'retryable' => false]; + } + if (str_ends_with($uri, '/complete') || str_ends_with($uri, '/fail')) { + return ['recorded' => true]; + } + throw new \LogicException('Unexpected request '.$method.' '.$uri); + }); + + return [new Client('https://server.example', namespace: 'tenant-a', transport: $transport), $transport]; + } +} diff --git a/tests/compat/execution-telemetry.php b/tests/compat/execution-telemetry.php new file mode 100644 index 00000000..460e5990 --- /dev/null +++ b/tests/compat/execution-telemetry.php @@ -0,0 +1,126 @@ +> */ + private static array $seen = []; + public static ?\Psr\Log\LoggerInterface $logger = null; + + public static function reset(): void { self::$seen = []; } + + /** @return list> */ + public static function observed(): array { return self::$seen; } + + #[Activity('telemetry.framework')] + public function activity(ActivityContext $context): string + { + self::$seen[] = ExecutionTelemetry::current(); + self::$logger?->info('framework callback', ['request_id' => 'application-request']); + + return 'framework-result'; + } +} + +/** @param Closure(Client, ?ExecutionTelemetry): Worker $create + * @param (Closure(\Monolog\Logger): void)|null $configureLogger + */ +function verifyFrameworkExecutionTelemetry(Closure $create, ?Closure $configureLogger = null): void +{ + if (getenv('DURABLE_WORKFLOW_TEST_TRACING') === '1' + && (!class_exists(\OpenTelemetry\API\Globals::class) || !class_exists(\Monolog\Logger::class))) { + throw new RuntimeException('The optional tracing qualification requires the API and Monolog.'); + } + foreach ([null, new ExecutionTelemetry()] as $telemetry) { + FrameworkTelemetryHandler::reset(); + $logs = null; + if (class_exists(\Monolog\Logger::class)) { + $logs = new \Monolog\Handler\TestHandler(); + $logger = new \Monolog\Logger('framework', [$logs]); + if ($configureLogger !== null) { + $configureLogger($logger); + } else { + $logger->pushProcessor(new \DurableWorkflow\Telemetry\LogContext()); + } + FrameworkTelemetryHandler::$logger = $logger; + } + $transport = new class implements Transport { + /** @var list> */ + private array $outcomes = []; + /** @return list> */ + public function outcomes(): array { return $this->outcomes; } + /** @param array $headers + * @param array|null $body + * @return array + */ + public function send(string $method, string $uri, array $headers, ?array $body = null): array + { + if (str_ends_with($uri, '/workflow-tasks/poll') || str_ends_with($uri, '/query-tasks/poll')) { + return ['task' => null, 'poll_status' => 'empty']; + } + if (str_ends_with($uri, '/activity-tasks/poll')) { + return ['task' => ['task_id' => 'framework-task', 'workflow_id' => 'framework-workflow', + 'run_id' => 'framework-run', 'activity_execution_id' => 'framework-activity', + 'activity_attempt_id' => 'framework-attempt', 'attempt_number' => 1, + 'activity_type' => 'telemetry.framework', 'lease_owner' => 'framework-worker', + 'payload_codec' => 'avro'], 'poll_status' => 'leased']; + } + if (str_ends_with($uri, '/framework-task/complete')) { + if ($body === null) { + throw new RuntimeException('Framework completion omitted its body.'); + } + $this->outcomes[] = $body; + + return ['recorded' => true]; + } + throw new RuntimeException('Unexpected framework telemetry request '.$method.' '.$uri); + } + }; + $client = new Client('https://framework.invalid', namespace: 'framework-namespace', transport: $transport); + $worker = $create($client, $telemetry); + $worked = $worker->tick(0); + $seen = FrameworkTelemetryHandler::observed(); + $outcomes = $transport->outcomes(); + if (!$worked || count($seen) !== 1 || count($outcomes) !== 1 + || ($outcomes[0]['activity_attempt_id'] ?? null) !== 'framework-attempt') { + throw new RuntimeException('Framework telemetry changed the callback or durable outcome.'); + } + $fields = $seen[0]; + $expected = $telemetry !== null && class_exists(\OpenTelemetry\API\Globals::class); + if ($expected) { + foreach (['workflow_run_id' => 'framework-run', 'activity_id' => 'framework-activity', + 'activity_attempt_id' => 'framework-attempt', 'namespace' => 'framework-namespace'] as $key => $value) { + if (($fields[$key] ?? null) !== $value) { + throw new RuntimeException('Framework worker omitted its scoped '.$key.'.'); + } + } + } elseif ($fields !== []) { + throw new RuntimeException('Framework worker enabled unrequested correlation.'); + } + if ($logs !== null) { + $records = $logs->getRecords(); + if (count($records) !== 1) { + throw new RuntimeException('Framework callback did not reach the application logger.'); + } + $record = $records[0]; + $context = ((array) $record)['context']; + if (($context['request_id'] ?? null) !== 'application-request' + || ($expected && ($context['activity_attempt_id'] ?? null) !== 'framework-attempt') + || (!$expected && array_key_exists('workflow_run_id', $context))) { + throw new RuntimeException('Framework logging lost its request ID or execution scope.'); + } + } + FrameworkTelemetryHandler::$logger = null; + if (ExecutionTelemetry::current() !== []) { + throw new RuntimeException('Framework worker leaked callback context.'); + } + } +} diff --git a/tests/compat/laravel.php b/tests/compat/laravel.php index 2dec6330..8464c1ff 100644 --- a/tests/compat/laravel.php +++ b/tests/compat/laravel.php @@ -4,6 +4,7 @@ require $argv[1]; require __DIR__.'/history-cache.php'; +require __DIR__.'/execution-telemetry.php'; use DurableWorkflow\Attribute\Activity; use DurableWorkflow\Attribute\Workflow; @@ -271,6 +272,14 @@ function laravelAuthenticationRejectsRole(Authentication $authentication, bool $ throw new RuntimeException('Laravel provider did not preserve role-specific client credentials.'); } $worker = $factory->make(); +verifyFrameworkExecutionTelemetry(static function (Client $client, ?\DurableWorkflow\Telemetry\ExecutionTelemetry $telemetry) use ($app): \DurableWorkflow\Worker { + $configuration = new \DurableWorkflow\Bridge\ServiceConfiguration('https://framework.invalid', 'framework-namespace', + 'framework-queue', [FrameworkTelemetryHandler::class]); + + return (new WorkerFactory($app, $configuration, $client, new LaravelBridgeLogger(), telemetry: $telemetry))->make(); +}, static function (\Monolog\Logger $logger): void { + (new \DurableWorkflow\Bridge\Laravel\LogContext())(new \Illuminate\Log\Logger($logger)); +}); $cacheStoppedContext = null; $cacheEvents = new Dispatcher($app); $cacheEvents->listen(WorkerDiagnosticEvent::class, static function (WorkerDiagnosticEvent $event) use (&$cacheStoppedContext): void { diff --git a/tests/compat/symfony.php b/tests/compat/symfony.php index 18d74321..773b2d8c 100644 --- a/tests/compat/symfony.php +++ b/tests/compat/symfony.php @@ -4,6 +4,7 @@ require $argv[1]; require __DIR__.'/history-cache.php'; +require __DIR__.'/execution-telemetry.php'; use DurableWorkflow\Attribute\Activity; use DurableWorkflow\Attribute\Query; @@ -205,6 +206,12 @@ public function fake(): WorkflowClientFake throw new RuntimeException('Symfony extension did not preserve role-specific client credentials.'); } $worker = $factory->make(); +verifyFrameworkExecutionTelemetry(static function (Client $client, ?\DurableWorkflow\Telemetry\ExecutionTelemetry $telemetry): \DurableWorkflow\Worker { + $configuration = new \DurableWorkflow\Bridge\ServiceConfiguration('https://framework.invalid', 'framework-namespace', + 'framework-queue', [FrameworkTelemetryHandler::class]); + + return (new WorkerFactory($configuration, $client, [new FrameworkTelemetryHandler()], telemetry: $telemetry))->make(); +}); $cacheStoppedContext = null; $cacheEvents = new EventDispatcher(); $cacheEvents->addListener('worker.stopped', static function (WorkerDiagnosticEvent $event) use (&$cacheStoppedContext): void { diff --git a/tests/fixtures/replay-regressions/telemetry-context-after-activity-replay.json b/tests/fixtures/replay-regressions/telemetry-context-after-activity-replay.json new file mode 100644 index 00000000..5852a157 --- /dev/null +++ b/tests/fixtures/replay-regressions/telemetry-context-after-activity-replay.json @@ -0,0 +1,13 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "php-telemetry-context-after-activity-replay", + "protocol_version": "1.20", + "bindings": ["php"], + "workflow": {"type": "golden.telemetry-context", "input": []}, + "history": [ + {"event_type": "ActivityScheduled", "payload": {"sequence": 1, "activity_type": "golden.greet"}}, + {"event_type": "ActivityCompleted", "payload": {"sequence": 1, "activity_type": "golden.greet", "result": {"codec": "avro", "blob": "wwHioz3/VYAiNwoUaGVsbG8sIEFkYQ=="}}} + ], + "expected": {"command_sequence": [{"type": "complete_workflow", "result": {"correlated": true, "namespace": "default", "result": "hello, Ada"}}]} +}