diff --git a/composer.json b/composer.json index fed39923..75c602de 100644 --- a/composer.json +++ b/composer.json @@ -82,7 +82,7 @@ } }, "durable-workflow": { - "product-train": "2.2.8", + "product-train": "2.3.0", "supported-server-versions": "2.5.0", "worker-protocol-version": "1.19", "control-plane-version": "2", diff --git a/docs/portal/operate/troubleshooting.md b/docs/portal/operate/troubleshooting.md index ae0616ca..31344a71 100644 --- a/docs/portal/operate/troubleshooting.md +++ b/docs/portal/operate/troubleshooting.md @@ -41,6 +41,52 @@ Confirm the client's `taskQueue` exactly matches the worker's queue. Then confir If the workflow scheduled an activity, repeat the same check for the activity type. A workflow worker can be healthy while the activity capacity it needs is absent. +## Inspect a durable service call + +Use an operator credential in the namespace you are investigating. The +service catalog read methods return bounded pages and the original durable +call, including its deadline, attempts and terminal cause. + +```php +$page = $client->listServiceCatalogCalls([ + 'scope' => 'caller', + 'status_bucket' => 'failed', + 'per_page' => 50, +]); +foreach ($page['data'] as $row) { + $call = $client->describeServiceCatalogCall($row['id']); +} +``` + +Browse registered endpoints, services and operations with +`listServiceCatalogEndpoints()`, `listServiceCatalogServices()` and +`listServiceCatalogOperations()`. Their corresponding +`describeServiceCatalogEndpoint()`, `describeServiceCatalogService()` and +`describeServiceCatalogOperation()` methods use catalog IDs. Lists accept +`page` and `per_page`. Filter services by `endpoint_id`, operations by +`endpoint_id` or `service_id`, and calls by those IDs or `operation_id`. + +The methods require the backend's advertised `service_catalog_read_contract` +version 1. An absent or incompatible capability raises `ServerException` with +`reason: backend_capability_unavailable` and the required/advertised contract. +A discovery outage keeps its actual connection or backend failure. It does not +become an unsupported-capability message. + +Foreign linked runs remain opaque references. Foreign input, output and failure +payload references are null, with `payloads_in_observer_namespace: false`. +Authorized payload references remain opaque runtime objects. A visible call +does not authorize browsing another namespace. See the +[Server read contract](https://github.com/durable-workflow/server/blob/main/docs/contracts/service-catalog-reads.md) +for scopes, filters, pagination and detail bounds. + +For an authorized owner, call detail also includes `output_available` and the +decoded `output`. This keeps `false`, `0`, an empty string and `null` distinct +from a missing result. The original opaque reference remains available. +Downloads use the configured runtime and operator credential, verify size and +digest, and honor the client's `maxExternalPayloadBytes` limit. Foreign or +unverified output references are not fetched. Decoded values retain the SDK's +typed Avro binary and map adapters. + ## A result wait timed out `WorkflowHandle::result()` polls for a bounded caller duration. When it times out, the durable execution may still be running. Call `describe()` and inspect its status before starting anything again. Reuse the original workflow ID or an application idempotency key; do not create duplicates just because the client lost patience. diff --git a/docs/quickstart-contract.json b/docs/quickstart-contract.json index 3037d763..7b6e3277 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": "2.2.8", - "composer_requirement": "2.2.8", - "onboarding_requirement": "^2.2" + "published_version": "2.3.0", + "composer_requirement": "2.3.0", + "onboarding_requirement": "^2.3" }, "runtime_targets": { "server": { diff --git a/src/Client.php b/src/Client.php index 0953241a..27f55b0e 100644 --- a/src/Client.php +++ b/src/Client.php @@ -173,6 +173,108 @@ public function clusterInfo(bool $includeDiagnostics = false): ClusterInfo return ClusterInfo::fromArray($this->control('GET', $path)); } + /** + * @param array $filters + * @return array + */ + public function listServiceCatalogEndpoints(array $filters = []): array + { + return $this->serviceCatalogRead('/service-catalog/endpoints', $filters); + } + + /** @return array */ + public function describeServiceCatalogEndpoint(string $endpointId): array + { + return $this->serviceCatalogRead('/service-catalog/endpoints', id: $endpointId); + } + + /** + * @param array $filters + * @return array + */ + public function listServiceCatalogServices(array $filters = []): array + { + return $this->serviceCatalogRead('/service-catalog/services', $filters); + } + + /** @return array */ + public function describeServiceCatalogService(string $serviceId): array + { + return $this->serviceCatalogRead('/service-catalog/services', id: $serviceId); + } + + /** + * @param array $filters + * @return array + */ + public function listServiceCatalogOperations(array $filters = []): array + { + return $this->serviceCatalogRead('/service-catalog/operations', $filters); + } + + /** @return array */ + public function describeServiceCatalogOperation(string $operationId): array + { + return $this->serviceCatalogRead('/service-catalog/operations', id: $operationId); + } + + /** + * @param array $filters + * @return array + */ + public function listServiceCatalogCalls(array $filters = []): array + { + return $this->serviceCatalogRead('/service-catalog/calls', $filters); + } + + /** @return array */ + public function describeServiceCatalogCall(string $callId): array + { + $response = $this->serviceCatalogRead('/service-catalog/calls', id: $callId); + if (($response['payloads_in_observer_namespace'] ?? null) === true) { + $response['output_available'] = is_array($response['output_envelope'] ?? null); + $response['output'] = $response['output_available'] + ? $this->decodedResponsePayload($response, 'output') + : null; + } + + return $response; + } + + /** + * @param array $filters + * @return array + */ + private function serviceCatalogRead(string $path, array $filters = [], ?string $id = null): array + { + if ($id !== null) { + if (trim($id) === '') { + throw new InvalidArgumentException('Catalog detail requires a non-empty ID.'); + } + $path .= '/'.$this->segment($id); + } + $schema = 'durable-workflow.v2.service-catalog-read.contract'; + $contract = $this->clusterInfo()->raw['service_catalog_read_contract'] ?? null; + if (!is_array($contract) || ($contract['schema'] ?? null) !== $schema || ($contract['version'] ?? null) !== 1) { + throw new ServerException( + 'The connected backend does not advertise service catalog operator reads with contract version 1.', + 501, + 'backend_capability_unavailable', + [ + 'capability' => 'service_catalog_reads', + 'required_contract' => $schema, + 'required_contract_version' => 1, + 'advertised_contract' => $contract, + ], + ); + } + if ($filters !== []) { + $path .= '?'.http_build_query($filters, '', '&', PHP_QUERY_RFC3986); + } + + return $this->control('GET', $path); + } + /** Discover ordinary runtime capabilities using the worker credential. */ public function workerClusterInfo(): ClusterInfo { diff --git a/src/Transport/RuntimePayloads.php b/src/Transport/RuntimePayloads.php index 3e906262..249e6b31 100644 --- a/src/Transport/RuntimePayloads.php +++ b/src/Transport/RuntimePayloads.php @@ -40,6 +40,13 @@ public function response(array $response, string $requestPath, bool $worker): ar $paths = [...$paths, ...$this->historyPaths(['task', 'history_events']), ...$this->historyPaths(['history_events']), ...$this->exportPaths(['task', 'history_export'])]; } else { + if (preg_match('~\A/service-catalog/calls/[^/]+\z~', $path) + && ($response['payloads_in_observer_namespace'] ?? null) === true + && is_array($response['output_payload_reference'] ?? null)) { + // Keep the opaque reference while hydrating only the authorized + // terminal output of this exact call, never arbitrary metadata. + $response['output_envelope'] = ['codec' => 'avro', 'external_payload' => $response['output_payload_reference']]; + } foreach (['input_envelope', 'output_envelope', 'result_envelope'] as $field) { $paths[] = [$field]; } diff --git a/tests/ServiceCatalogReadClientTest.php b/tests/ServiceCatalogReadClientTest.php new file mode 100644 index 00000000..f2242e6c --- /dev/null +++ b/tests/ServiceCatalogReadClientTest.php @@ -0,0 +1,222 @@ + [['id' => 'call-1']], 'call_deadline_at' => '2026-10-11T19:00:01.123456Z', + 'failure_type' => 'ServiceCallTimedOut', 'input_payload_reference' => ['reference_id' => 'opaque-1']]; + $transport = new FakeTransport([$this->discovery(), $surface]); + $client = new Client('https://server.example', token: 'operator-token', namespace: 'tenant-a', transport: $transport); + $result = $detail ? $client->{$method}('call/1') : $client->{$method}(['page' => 2, 'per_page' => 10]); + + self::assertSame($surface, $result); + self::assertCount(2, $transport->requests); + self::assertSame('https://server.example/api/cluster/info', $transport->requests[0]['uri']); + self::assertSame('https://server.example/api/service-catalog/'.$path.($detail ? '/call%2F1' : '?page=2&per_page=10'), $transport->requests[1]['uri']); + foreach ($transport->requests as $request) { + self::assertSame('GET', $request['method']); + self::assertNull($request['body']); + self::assertSame('tenant-a', $request['headers']['X-Namespace']); + self::assertSame('2', $request['headers']['X-Durable-Workflow-Control-Plane-Version']); + self::assertSame('Bearer operator-token', $request['headers']['Authorization']); + } + } + + public static function readMethods(): array + { + return [ + ['listServiceCatalogEndpoints', 'endpoints', false], ['describeServiceCatalogEndpoint', 'endpoints', true], + ['listServiceCatalogServices', 'services', false], ['describeServiceCatalogService', 'services', true], + ['listServiceCatalogOperations', 'operations', false], ['describeServiceCatalogOperation', 'operations', true], + ['listServiceCatalogCalls', 'calls', false], ['describeServiceCatalogCall', 'calls', true], + ]; + } + + public function testDescribeRejectsBlankIdentitiesBeforeDiscovery(): void + { + $transport = new FakeTransport(); + $client = new Client('https://server.example', transport: $transport); + foreach (self::readMethods() as [$method, , $detail]) { + if (!$detail) { + continue; + } + foreach (['', ' '] as $id) { + try { + $client->{$method}($id); + self::fail('Blank identity must not turn a detail lookup into a collection read.'); + } catch (\InvalidArgumentException $exception) { + self::assertStringContainsString('non-empty ID', $exception->getMessage()); + self::assertCount(0, $transport->requests); + } + } + } + } + + #[DataProvider('unsupportedContracts')] + public function testUnsupportedBackendIsExplainedBeforeAttemptingARead(mixed $contract): void + { + $transport = new FakeTransport([['service_catalog_read_contract' => $contract]]); + $client = new Client('https://server.example', transport: $transport); + try { + $client->listServiceCatalogCalls(); + self::fail('Missing or unsupported read capability must be diagnosed.'); + } catch (ServerException $exception) { + self::assertSame(501, $exception->status); + self::assertSame('backend_capability_unavailable', $exception->reason); + self::assertSame('service_catalog_reads', $exception->details['capability']); + self::assertSame(1, $exception->details['required_contract_version']); + self::assertSame($contract, $exception->details['advertised_contract']); + self::assertCount(1, $transport->requests); + } + } + + public static function unsupportedContracts(): array + { + return [[null], ['legacy'], [[]], [['schema' => 'other', 'version' => 1]], + [['schema' => 'durable-workflow.v2.service-catalog-read.contract', 'version' => 2]], + [['schema' => 'durable-workflow.v2.service-catalog-read.contract', 'version' => '1']]]; + } + + public function testActualReadRefusalsRetainTheServerReasonAndDiscoveryFailuresAreNotUnsupported(): void + { + $transport = new FakeTransport([$this->discovery(), new TransportException('Forbidden', 403, ['reason' => 'forbidden'])]); + $client = new Client('https://server.example', transport: $transport); + try { + $client->describeServiceCatalogCall('call-1'); + self::fail('Worker role must remain refused by the backend.'); + } catch (ServerException $exception) { + self::assertSame(403, $exception->status); + self::assertSame('forbidden', $exception->reason); + } + + $transport = new FakeTransport([new TransportException('Backend unavailable', 503, ['reason' => 'backend_unavailable'])]); + $client = new Client('https://server.example', transport: $transport); + try { + $client->listServiceCatalogCalls(); + self::fail('Discovery failure must retain the backend failure.'); + } catch (ServerException $exception) { + self::assertSame(503, $exception->status); + self::assertSame('backend_unavailable', $exception->reason); + self::assertCount(1, $transport->requests); + } + } + + private function discovery(): array + { + return ['service_catalog_read_contract' => ['schema' => 'durable-workflow.v2.service-catalog-read.contract', 'version' => 1]]; + } + + #[DataProvider('outputs')] + public function testOwnedTerminalOutputIsDecodedWithoutReplacingTheOpaqueReference(mixed $value): void + { + $codec = new AvroPayloadCodec(); + $blob = $codec->envelope($value)['blob']; + $reference = ['schema' => RuntimePayloads::SCHEMA, 'reference_id' => 'ep_01ARZ3NDEKTSV4RRFFQ69G5FAV', + 'codec' => 'avro', 'size_bytes' => strlen($blob), 'sha256' => hash('sha256', $blob)]; + $handler = new MockHandler([ + new Response(200, [], json_encode($this->discovery(), JSON_THROW_ON_ERROR)), + new Response(200, [], json_encode([ + 'id' => 'call-1', 'payloads_in_observer_namespace' => true, 'output_payload_reference' => $reference, + ], JSON_THROW_ON_ERROR)), + new Response(200, [ + 'Content-Type' => 'application/octet-stream', 'X-Durable-Workflow-Payload-Codec' => 'avro', + 'X-Durable-Workflow-Payload-Size' => (string) strlen($blob), + 'X-Durable-Workflow-Payload-SHA256' => hash('sha256', $blob), + ], $blob), + ]); + $transport = new Psr18Transport(new GuzzleClient(['handler' => HandlerStack::create($handler)])); + $result = (new Client('https://runtime.test', transport: $transport))->describeServiceCatalogCall('call-1'); + self::assertTrue($result['output_available']); + self::assertSame($reference, $result['output_payload_reference']); + self::assertSame($codec->encode($value), $codec->encode($result['output'])); + } + + public static function outputs(): array + { + return [[null], [false], [0], [''], [['result', ['nested' => 'value']]]]; + } + + public function testExternalOutputUsesOnlyTheConfiguredRuntimeNamespaceAndOperatorCredential(): void + { + $codec = new AvroPayloadCodec(); + $value = ['result', str_repeat('x', 10000)]; + $blob = $codec->envelope($value)['blob']; + $reference = ['schema' => RuntimePayloads::SCHEMA, 'reference_id' => 'ep_01ARZ3NDEKTSV4RRFFQ69G5FAV', + 'codec' => 'avro', 'size_bytes' => strlen($blob), 'sha256' => hash('sha256', $blob)]; + $envelope = $reference; + $requests = []; + $http = new GuzzleClient(['handler' => function (RequestInterface $request, array $options) use (&$requests, $blob, $envelope) { + $requests[] = $request; + self::assertFalse($options['allow_redirects']); + if (count($requests) === 1) { + return Create::promiseFor(new Response(200, [], json_encode($this->discovery(), JSON_THROW_ON_ERROR))); + } + if (count($requests) === 2) { + return Create::promiseFor(new Response(200, [], json_encode([ + 'id' => 'call-1', 'payloads_in_observer_namespace' => true, 'output_payload_reference' => $envelope, + ], JSON_THROW_ON_ERROR))); + } + + return Create::promiseFor(new Response(200, [ + 'Content-Type' => 'application/octet-stream', 'X-Durable-Workflow-Payload-Codec' => 'avro', + 'X-Durable-Workflow-Payload-Size' => (string) strlen($blob), + 'X-Durable-Workflow-Payload-SHA256' => hash('sha256', $blob), + ], $blob)); + }]); + $client = new Client('https://runtime.test/api/runtime/v1/namespaces/owner-id', namespace: 'owner', + transport: new Psr18Transport($http), controlToken: 'operator', workerToken: 'worker'); + $result = $client->describeServiceCatalogCall('call-1'); + self::assertSame($codec->encode($value), $codec->encode($result['output'])); + self::assertSame($envelope, $result['output_payload_reference']); + self::assertCount(3, $requests); + self::assertSame('https://runtime.test/api/runtime/v1/namespaces/owner-id/api/external-payloads/v1/'.$reference['reference_id'], (string) $requests[2]->getUri()); + self::assertSame('owner', $requests[2]->getHeaderLine('X-Namespace')); + self::assertSame('Bearer operator', $requests[2]->getHeaderLine('Authorization')); + } + + public function testForeignOrUnverifiedOutputsAreNotFetchedAndOwnedDownloadsHonorTheByteLimit(): void + { + $blob = (new AvroPayloadCodec())->envelope(['private'])['blob']; + $reference = ['schema' => RuntimePayloads::SCHEMA, 'reference_id' => 'ep_01ARZ3NDEKTSV4RRFFQ69G5FAV', + 'codec' => 'avro', 'size_bytes' => strlen($blob), 'sha256' => hash('sha256', $blob)]; + foreach ([false, null, true] as $authorized) { + $transport = new FakeTransport([$this->discovery(), [ + 'payloads_in_observer_namespace' => $authorized, + 'output_payload_reference' => $reference, + ]]); + $client = new Client('https://runtime.test', transport: $transport, maxExternalPayloadBytes: 1); + try { + $result = $client->describeServiceCatalogCall('call-1'); + self::assertNotTrue($authorized); + self::assertArrayNotHasKey('output', $result); + } catch (ExternalPayloadException $exception) { + self::assertTrue($authorized); + self::assertSame('external_payload_oversized', $exception->reason); + } + self::assertCount(2, $transport->requests); + } + } +}