Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion composer.json
Original file line number Diff line number Diff line change
Expand Up @@ -82,7 +82,7 @@
}
},
"durable-workflow": {
"product-train": "2.3.0-rc.2",
"product-train": "2.4.0-rc.1",
"supported-server-versions": "2.5.0",
"worker-protocol-version": "1.19",
"control-plane-version": "2",
Expand Down
8 changes: 8 additions & 0 deletions docs/portal/operate/troubleshooting.md
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,14 @@ 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.
Expand Down
6 changes: 3 additions & 3 deletions docs/quickstart-contract.json
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,9 @@
"schema_version": 2,
"package": {
"name": "durable-workflow/sdk",
"published_version": "2.3.0-rc.2",
"composer_requirement": "2.3.0-rc.2",
"onboarding_requirement": "^2.3@RC"
"published_version": "2.4.0-rc.1",
"composer_requirement": "2.4.0-rc.1",
"onboarding_requirement": "^2.4@RC"
},
"runtime_targets": {
"server": {
Expand Down
10 changes: 9 additions & 1 deletion src/Client.php
Original file line number Diff line number Diff line change
Expand Up @@ -230,7 +230,15 @@ public function listServiceCatalogCalls(array $filters = []): array
/** @return array<string, mixed> */
public function describeServiceCatalogCall(string $callId): array
{
return $this->serviceCatalogRead('/service-catalog/calls', id: $callId);
$response = $this->serviceCatalogRead('/service-catalog/calls', id: $callId);
if (($response['payloads_in_observer_namespace'] ?? null) === true) {
$response['output_available'] = is_array($response['result_envelope'] ?? null);
$response['output'] = $response['output_available']
? $this->decodedResponsePayload($response, 'result')
: null;
}

return $response;
}

/**
Expand Down
12 changes: 12 additions & 0 deletions src/Transport/RuntimePayloads.php
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,19 @@ 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 {
$catalogCall = preg_match('~\A/service-catalog/calls/[^/]+\z~', $path) === 1;
$catalogOutputAllowed = ($response['payloads_in_observer_namespace'] ?? null) === true;
if ($catalogCall && $catalogOutputAllowed
&& !is_array($response['result_envelope'] ?? null)
&& 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['result_envelope'] = ['codec' => 'avro', 'external_payload' => $response['output_payload_reference']];
}
foreach (['input_envelope', 'output_envelope', 'result_envelope'] as $field) {
if ($catalogCall && !$catalogOutputAllowed) {
continue;
}
$paths[] = [$field];
}
if (preg_match('~\A/activities/[^/]+\z~', $path)) {
Expand Down
118 changes: 118 additions & 0 deletions tests/ServiceCatalogReadClientTest.php
Original file line number Diff line number Diff line change
Expand Up @@ -5,11 +5,21 @@
namespace DurableWorkflow\Tests;

use DurableWorkflow\Client;
use DurableWorkflow\Codec\AvroPayloadCodec;
use DurableWorkflow\Exception\ExternalPayloadException;
use DurableWorkflow\Exception\ServerException;
use DurableWorkflow\Exception\TransportException;
use DurableWorkflow\Tests\Support\FakeTransport;
use DurableWorkflow\Transport\Psr18Transport;
use DurableWorkflow\Transport\RuntimePayloads;
use GuzzleHttp\Client as GuzzleClient;
use GuzzleHttp\Handler\MockHandler;
use GuzzleHttp\HandlerStack;
use GuzzleHttp\Promise\Create;
use GuzzleHttp\Psr7\Response;
use PHPUnit\Framework\Attributes\DataProvider;
use PHPUnit\Framework\TestCase;
use Psr\Http\Message\RequestInterface;

final class ServiceCatalogReadClientTest extends TestCase
{
Expand Down Expand Up @@ -118,4 +128,112 @@ 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']]]];
}

#[DataProvider('outputs')]
public function testCommittedInlineResultEnvelopeIsDecoded(mixed $value): void
{
$codec = new AvroPayloadCodec();
$envelope = $codec->envelope($value);
$transport = new FakeTransport([$this->discovery(), [
'payloads_in_observer_namespace' => true, 'output_payload_reference' => null,
'result_envelope' => $envelope,
]]);
$result = (new Client('https://runtime.test', transport: $transport))->describeServiceCatalogCall('call-1');
self::assertTrue($result['output_available']);
self::assertSame($codec->encode($value), $codec->encode($result['output']));
self::assertCount(2, $transport->requests);
}

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,
'result_envelope' => ['codec' => 'avro', 'external_payload' => $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,
'result_envelope' => ['codec' => 'avro', 'external_payload' => $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);
}
}
}
Loading