From 311abe7f6a611d079f83c011362939e7781eb41e Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 14:47:04 +0000 Subject: [PATCH 1/3] Add typed control-plane workflow version-usage reads --- CHANGELOG.md | 7 + docs/portal/build/workflows-activities.md | 43 ++++ src/Client.php | 52 +++++ src/Metrics/MetricsRoutes.php | 2 +- src/Model/ClusterInfo.php | 8 + src/Model/WorkflowVersionUsageCounts.php | 39 ++++ src/Model/WorkflowVersionUsagePage.php | 96 ++++++++ src/Model/WorkflowVersionUsageRun.php | 52 +++++ tests/ControlRequestTimeoutTest.php | 2 +- tests/Support/control-timeout-client.php | 1 + tests/Support/control-timeout-peer.php | 2 +- tests/WorkflowVersionUsageTest.php | 264 ++++++++++++++++++++++ 12 files changed, 565 insertions(+), 3 deletions(-) create mode 100644 src/Model/WorkflowVersionUsageCounts.php create mode 100644 src/Model/WorkflowVersionUsagePage.php create mode 100644 src/Model/WorkflowVersionUsageRun.php create mode 100644 tests/WorkflowVersionUsageTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 90f74eef..98667628 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,13 @@ project follows [Semantic Versioning](https://semver.org/). ## [Unreleased] +### Added + +- `Client::workflowVersionUsage()` reads typed live v1 open-run version usage, + with all-state counts, explicit unknown reasons and opaque continuation cursors + (.github#220). It requires the advertised `workflow_version_usage` capability + and retains the control credential, namespace and request timeout policy. + ### Security - Raise the optional Laravel bridge floors to 12.72.0 and 13.41.0 for diff --git a/docs/portal/build/workflows-activities.md b/docs/portal/build/workflows-activities.md index b0e05b64..df183d0e 100644 --- a/docs/portal/build/workflows-activities.md +++ b/docs/portal/build/workflows-activities.md @@ -407,6 +407,49 @@ A later deployment can raise the maximum while keeping the recorded versions in For a boolean rollout, use `patched('require-reservation-review')`. It returns `false` for legacy histories and `true` for new histories. After the legacy branch is no longer supported, replace the conditional call with `deprecatePatch()` for the same ID before eventually removing the marker. Compatible numeric and patch helpers share the same frozen version for a change ID. Every call must support that version, and `patched()` returns whether it equals `1`. Do not reuse a change ID for an unrelated rollout. +### Inspect open-run version usage + +Before raising a minimum or removing a legacy branch, inspect the open runs for +that exact workflow type and change ID. `workflowVersionUsage()` requires the +Server's `workflow_version_usage` capability and throws +`ServerCapabilityUnavailable` when it is absent. Discovery and the metadata read +use the client's control credential, namespace and configured control-request +timeout. They do not use a worker credential or fetch application payloads. + +```php +$page = $client->workflowVersionUsage( + workflowType: 'shipment', + changeId: 'carrier-api', + minVersion: 2, + state: 'legacy', + limit: 50, +); + +// Counts include every state, even though this page lists only legacy runs. +$unknown = $page->counts->unknown; +foreach ($page->runs as $run) { + printf("%s %s %s\n", $run->instanceId, $run->runId, $run->state); +} + +// Repeat the exact namespace/type/change/minimum/state query to continue. +$nextCursor = $page->nextCursor; +``` + +`minVersion` is the proposed minimum supported signed 32-bit version, including +negative legacy sentinels; it is not the largest deployed version. The response +distinguishes `legacy`, `current` and `unknown`. Missing markers, unavailable or +invalid history remain unknown with their original reason. A current build or +definition is not evidence that a run recorded the current branch. + +Each page is a live observation over a fixed upper run-ID bound, not a durable +fleet snapshot. Runs can close or record decisions between pages, so counts can +change. Pass cursors unchanged with the exact same filters; invalid cursors are +errors, not a reason to restart silently. Unknowns remain relevant to rollout +decisions. Fence or drain admissions and restart the traversal before making a +removal decision; this API returns no `safe_to_remove` authorization. The +[shared v1 contract](https://github.com/durable-workflow/workflow/blob/c93cadad8a0104b6e7e5f38e022c3b06c26c86e6/docs/version-usage.md) +defines the classification and live paging limits. + A `waitCondition()` predicate is part of replay identity too. New waits use a `dwphp2:sha256:` fingerprint of the callable's tokens. Whitespace and comments outside literals, file locations and neighboring call-site arguments are diff --git a/src/Client.php b/src/Client.php index 5c8caba6..d8ee1c01 100644 --- a/src/Client.php +++ b/src/Client.php @@ -39,6 +39,7 @@ use DurableWorkflow\Model\ServiceOperationOptions; use DurableWorkflow\Model\WorkflowExecution; use DurableWorkflow\Model\WorkflowPage; +use DurableWorkflow\Model\WorkflowVersionUsagePage; use DurableWorkflow\Model\WorkflowRun; use DurableWorkflow\Model\WorkflowStreamAppendItem; use DurableWorkflow\Model\WorkflowStreamAppendResult; @@ -388,6 +389,57 @@ public function clusterInfo(bool $includeDiagnostics = false): ClusterInfo return ClusterInfo::fromArray($info); } + /** + * Read one page of open-run version decisions for a proposed minimum supported version. + * + * Uses the existing control credential, namespace and control-request timeout policy. + * Cursors are opaque and query-bound: pass the returned nextCursor unchanged with the + * same filters. Counts precede state filtering and each page is a live observation, + * not a durable snapshot or authorization to remove a version branch. + * + * @throws ServerCapabilityUnavailable when workflow_version_usage is not advertised + * @throws \UnexpectedValueException when the v1 response or query identity is invalid + */ + public function workflowVersionUsage( + string $workflowType, + string $changeId, + int $minVersion, + ?string $state = null, + int $limit = 50, + ?string $cursor = null, + ): WorkflowVersionUsagePage { + foreach ([[$workflowType, 255], [$changeId, 1024]] as [$identity, $maximum]) { + if (strlen($identity) > $maximum || preg_match('/[^\s\p{Z}]/u', $identity) !== 1) { + throw new InvalidArgumentException('Version usage needs nonblank UTF-8 workflow type and change ID within their byte limits.'); + } + } + if ($minVersion < -2147483648 || $minVersion > 2147483647 + || !in_array($state, [null, 'legacy', 'current', 'unknown'], true) || $limit < 1 || $limit > 200 + || ($cursor !== null && ($cursor === '' || preg_match('//u', $cursor) !== 1))) { + throw new InvalidArgumentException('Invalid version-usage minimum, state, limit or opaque cursor.'); + } + $info = ClusterInfo::fromArray($this->request('GET', self::CLUSTER_INFO_PATH, false, + payloadPaths: static fn (array $metadata): array => [])); + if (!$info->supportsWorkflowVersionUsage()) { + throw new ServerCapabilityUnavailable(ClusterInfo::WORKFLOW_VERSION_USAGE, + 'This Server does not advertise workflow_version_usage.'); + } + $query = $this->withoutNulls([ + 'workflow_type' => $workflowType, 'change_id' => $changeId, 'min_version' => $minVersion, + 'state' => $state, 'limit' => $limit, 'cursor' => $cursor, + ]); + $response = $this->request('GET', '/workflow-version-usage?'.http_build_query($query, '', '&', PHP_QUERY_RFC3986), + false, payloadPaths: static fn (array $metadata): array => []); + $page = WorkflowVersionUsagePage::fromArray($response); + if ($page->namespace !== $this->namespace || $page->workflowType !== $workflowType + || $page->changeId !== $changeId || $page->minVersion !== $minVersion + || $page->state !== $state || $page->limit !== $limit) { + throw new \UnexpectedValueException('Version-usage response does not match the authenticated namespace and exact query.'); + } + + return $page; + } + /** * @param array $filters * @return array diff --git a/src/Metrics/MetricsRoutes.php b/src/Metrics/MetricsRoutes.php index 906046a6..c2d1f93b 100644 --- a/src/Metrics/MetricsRoutes.php +++ b/src/Metrics/MetricsRoutes.php @@ -136,7 +136,7 @@ private static function sharedTaskRoutes(): array /** @return list Reviewed operations only the PHP client currently sends. */ private static function phpRoutes(): array { - $routes = ['/activities/{activity_id}/result', '/system/health', '/system/operator-dashboard', + $routes = ['/workflow-version-usage', '/activities/{activity_id}/result', '/system/health', '/system/operator-dashboard', '/system/operator-dashboard/bounded', '/system/operator-dashboard/bounded/workflow-types', '/system/operator-metrics', '/task-queues/{task_queue}/activity-dispatch', '/task-queues/{task_queue}/activity-dispatch/pause', '/task-queues/{task_queue}/activity-dispatch/resume']; diff --git a/src/Model/ClusterInfo.php b/src/Model/ClusterInfo.php index 151add3b..5c353b28 100644 --- a/src/Model/ClusterInfo.php +++ b/src/Model/ClusterInfo.php @@ -13,6 +13,9 @@ final class ClusterInfo /** Signal and update `request_id` is a durable command key (native Server 3.9.0 and later). */ public const COMMAND_REQUEST_KEYS = 'command_request_keys'; + /** Read-only v1 open-run version usage; availability is discovered, never inferred from a version. */ + public const WORKFLOW_VERSION_USAGE = 'workflow_version_usage'; + /** * @param array $namespace * @param array $capabilities @@ -69,6 +72,11 @@ public function supportsCommandRequestKeys(): bool return ($this->capabilities[self::COMMAND_REQUEST_KEYS] ?? null) === true; } + public function supportsWorkflowVersionUsage(): bool + { + return ($this->capabilities[self::WORKFLOW_VERSION_USAGE] ?? null) === true; + } + /** @return array */ private static function map(mixed $value): array { diff --git a/src/Model/WorkflowVersionUsageCounts.php b/src/Model/WorkflowVersionUsageCounts.php new file mode 100644 index 00000000..87882b4b --- /dev/null +++ b/src/Model/WorkflowVersionUsageCounts.php @@ -0,0 +1,39 @@ + $value */ + public static function fromArray(array $value): self + { + $keys = array_keys($value); + sort($keys); + if ($keys !== ['current', 'legacy', 'total', 'unknown']) { + throw new UnexpectedValueException('Invalid workflow version-usage counts.'); + } + foreach ($value as $count) { + if (!is_int($count) || $count < 0) { + throw new UnexpectedValueException('Workflow version-usage counts must be non-negative integers.'); + } + } + if ($value['total'] !== $value['legacy'] + $value['current'] + $value['unknown']) { + throw new UnexpectedValueException('Workflow version-usage counts do not reconcile.'); + } + + return new self($value['total'], $value['legacy'], $value['current'], $value['unknown']); + } +} diff --git a/src/Model/WorkflowVersionUsagePage.php b/src/Model/WorkflowVersionUsagePage.php new file mode 100644 index 00000000..86212a1e --- /dev/null +++ b/src/Model/WorkflowVersionUsagePage.php @@ -0,0 +1,96 @@ + $runs + * @param array $raw + */ + private function __construct( + public readonly ?string $namespace, + public readonly string $workflowType, + public readonly string $changeId, + public readonly int $minVersion, + public readonly ?string $state, + public readonly ?string $throughRunId, + public readonly WorkflowVersionUsageCounts $counts, + public readonly array $runs, + public readonly int $limit, + public readonly int $returnedCount, + public readonly bool $hasMore, + public readonly ?string $nextCursor, + public readonly array $raw, + ) { + } + + /** @param array $value */ + public static function fromArray(array $value): self + { + $keys = array_keys($value); + sort($keys); + if ($keys !== ['change_id', 'consistency', 'counts', 'has_more', 'limit', 'min_version', 'namespace', + 'next_cursor', 'returned_count', 'runs', 'schema_version', 'state', 'through_run_id', 'workflow_type'] + || $value['schema_version'] !== 1 || $value['consistency'] !== 'live' + || !is_int($value['min_version']) || $value['min_version'] < -2147483648 || $value['min_version'] > 2147483647 + || !in_array($value['state'], [null, 'legacy', 'current', 'unknown'], true) + || !is_int($value['limit']) || $value['limit'] < 1 || $value['limit'] > 200 + || !is_int($value['returned_count']) || $value['returned_count'] < 0 + || !is_bool($value['has_more']) || !is_array($value['counts']) + || !is_array($value['runs']) || !array_is_list($value['runs'])) { + throw new UnexpectedValueException('Invalid workflow version-usage v1 response.'); + } + foreach (['namespace', 'workflow_type', 'change_id', 'through_run_id', 'next_cursor'] as $field) { + if ($value[$field] === null && in_array($field, ['namespace', 'through_run_id', 'next_cursor'], true)) { + continue; + } + if (!is_string($value[$field]) || $value[$field] === '' || preg_match('//u', $value[$field]) !== 1) { + throw new UnexpectedValueException('Invalid workflow version-usage response identity or cursor.'); + } + } + $counts = WorkflowVersionUsageCounts::fromArray($value['counts']); + if ($value['returned_count'] !== count($value['runs']) || $value['returned_count'] > $value['limit'] + || $value['returned_count'] > $counts->total + || ($value['has_more'] !== ($value['next_cursor'] !== null)) + || ($value['has_more'] && $value['returned_count'] === 0) + || ($value['through_run_id'] === null && $counts->total !== 0)) { + throw new UnexpectedValueException('Workflow version-usage page boundaries do not reconcile.'); + } + $runs = []; + $remaining = ['legacy' => $counts->legacy, 'current' => $counts->current, 'unknown' => $counts->unknown]; + $selectedCount = $value['state'] === null ? $counts->total : $remaining[$value['state']]; + if ($value['has_more'] && $selectedCount <= $value['returned_count']) { + throw new UnexpectedValueException('Workflow version-usage continuation has no remaining matching runs.'); + } + $previous = null; + foreach ($value['runs'] as $row) { + if (!is_array($row)) { + throw new UnexpectedValueException('Invalid workflow version-usage run.'); + } + $run = WorkflowVersionUsageRun::fromArray($row, $value['min_version']); + if (--$remaining[$run->state] < 0) { + throw new UnexpectedValueException('Workflow version-usage page exceeds its all-state counts.'); + } + if (($previous !== null && strcmp($previous, $run->runId) >= 0) + || $value['through_run_id'] === null || strcmp($run->runId, $value['through_run_id']) > 0 + || ($value['state'] !== null && $run->state !== $value['state'])) { + throw new UnexpectedValueException('Workflow version-usage runs violate their order, upper bound or filter.'); + } + $runs[] = $run; + $previous = $run->runId; + } + + return new self($value['namespace'], $value['workflow_type'], $value['change_id'], $value['min_version'], + $value['state'], $value['through_run_id'], $counts, $runs, $value['limit'], $value['returned_count'], + $value['has_more'], $value['next_cursor'], $value); + } +} diff --git a/src/Model/WorkflowVersionUsageRun.php b/src/Model/WorkflowVersionUsageRun.php new file mode 100644 index 00000000..b7380c0b --- /dev/null +++ b/src/Model/WorkflowVersionUsageRun.php @@ -0,0 +1,52 @@ + $value */ + public static function fromArray(array $value, int $minimum): self + { + $keys = array_keys($value); + sort($keys); + if ($keys !== ['instance_id', 'recorded_version', 'run_id', 'state', 'status', 'unknown_reason']) { + throw new UnexpectedValueException('Invalid workflow version-usage run fields.'); + } + foreach (['instance_id', 'run_id'] as $field) { + if (!is_string($value[$field]) || $value[$field] === '' || preg_match('//u', $value[$field]) !== 1) { + throw new UnexpectedValueException('Invalid workflow version-usage run identity.'); + } + } + if (!in_array($value['status'], ['pending', 'running', 'waiting'], true) + || !in_array($value['state'], ['legacy', 'current', 'unknown'], true)) { + throw new UnexpectedValueException('Invalid workflow version-usage run state.'); + } + $version = $value['recorded_version']; + $reason = $value['unknown_reason']; + if ($value['state'] === 'unknown') { + if ($version !== null || !in_array($reason, ['marker_absent', 'history_unavailable', 'history_invalid'], true)) { + throw new UnexpectedValueException('Unknown version usage requires its explicit unknown reason.'); + } + } elseif (!is_int($version) || $version < -2147483648 || $version > 2147483647 || $reason !== null + || (($version < $minimum) !== ($value['state'] === 'legacy'))) { + throw new UnexpectedValueException('Recorded version usage does not match the proposed minimum.'); + } + + return new self($value['instance_id'], $value['run_id'], $value['status'], $value['state'], $version, $reason); + } +} diff --git a/tests/ControlRequestTimeoutTest.php b/tests/ControlRequestTimeoutTest.php index bf86b9b8..ab647e4b 100644 --- a/tests/ControlRequestTimeoutTest.php +++ b/tests/ControlRequestTimeoutTest.php @@ -67,7 +67,7 @@ public static function stalledRequests(): iterable { foreach ([false, true] as $native) { foreach (['stall', 'slow_headers', 'slow_body'] as $mode) { - foreach (['list', 'start', 'signal', 'update', ...self::WORKER_OPERATIONS] as $operation) { + foreach (['list', 'start', 'signal', 'update', 'version-usage', ...self::WORKER_OPERATIONS] as $operation) { yield $mode.' '.$operation.' native='.(int) $native => [$mode, $operation, $native]; } } diff --git a/tests/Support/control-timeout-client.php b/tests/Support/control-timeout-client.php index 6da52e50..03b9d4b6 100644 --- a/tests/Support/control-timeout-client.php +++ b/tests/Support/control-timeout-client.php @@ -36,6 +36,7 @@ public function headers(bool $workerRequest): array try { match ($operation) { 'list' => $client->listWorkflows(), + 'version-usage' => $client->workflowVersionUsage('shipment', 'carrier-api', 2), 'worker' => $client->workerClusterInfo(), 'payload' => (new DurableWorkflow\Transport\Psr18Transport(requestTimeoutSeconds: (float) $seconds)) ->uploadPayload($url.'/api/external-payloads/v1', [], 'fixture', 1), diff --git a/tests/Support/control-timeout-peer.php b/tests/Support/control-timeout-peer.php index 8a01c5db..d38323e0 100644 --- a/tests/Support/control-timeout-peer.php +++ b/tests/Support/control-timeout-peer.php @@ -54,7 +54,7 @@ 'headers' => $headers, 'body' => $body], JSON_THROW_ON_ERROR)."\n", FILE_APPEND | LOCK_EX); $replyMode = str_starts_with($parts[1], '/api/cluster/info') ? 'echo' : $mode; if (str_starts_with($parts[1], '/api/cluster/info')) { - $payload = '{"limits":{"max_payload_bytes":1048576}}'; + $payload = '{"limits":{"max_payload_bytes":1048576},"capabilities":{"workflow_version_usage":true}}'; } elseif ($replyMode === 'echo') { $payload = '{"status":"ok","workflows":[]}'; } else { diff --git a/tests/WorkflowVersionUsageTest.php b/tests/WorkflowVersionUsageTest.php new file mode 100644 index 00000000..0f767b27 --- /dev/null +++ b/tests/WorkflowVersionUsageTest.php @@ -0,0 +1,264 @@ + ['workflow_version_usage' => true]]; + + /** @return array */ + private static function page(): array + { + return [ + 'schema_version' => 1, 'namespace' => 'orders', 'workflow_type' => 'shipment', + 'change_id' => 'carrier-api', 'min_version' => 2, 'state' => null, 'consistency' => 'live', + 'through_run_id' => 'run-3', 'counts' => ['total' => 3, 'legacy' => 1, 'current' => 1, 'unknown' => 1], + 'runs' => [ + ['instance_id' => 'shipment-1', 'run_id' => 'run-1', 'status' => 'waiting', + 'state' => 'legacy', 'recorded_version' => 1, 'unknown_reason' => null], + ['instance_id' => 'shipment-2', 'run_id' => 'run-2', 'status' => 'running', + 'state' => 'current', 'recorded_version' => 2, 'unknown_reason' => null], + ['instance_id' => 'shipment-3', 'run_id' => 'run-3', 'status' => 'pending', + 'state' => 'unknown', 'recorded_version' => null, 'unknown_reason' => 'marker_absent'], + ], + 'limit' => 50, 'returned_count' => 3, 'has_more' => false, 'next_cursor' => null, + ]; + } + + public function testTypedMetadataUsesExactQueryNamespaceAndControlCredential(): void + { + $transport = new FakeTransport([self::CAPABILITY, self::page()]); + $client = new Client('https://runtime.example', namespace: 'orders', controlToken: 'control', workerToken: 'worker', transport: $transport); + $page = $client->workflowVersionUsage('shipment', 'carrier-api', 2); + + self::assertInstanceOf(WorkflowVersionUsagePage::class, $page); + self::assertSame(3, $page->counts->total); + self::assertSame([1, 2, null], array_map(static fn ($run) => $run->recordedVersion, $page->runs)); + self::assertSame('marker_absent', $page->runs[2]->unknownReason); + self::assertNull($page->nextCursor); + self::assertSame(self::page(), $page->raw); + self::assertCount(2, $transport->requests); + foreach ($transport->requests as $request) { + self::assertSame('GET', $request['method']); + self::assertNull($request['body']); + self::assertSame('orders', $request['headers']['X-Namespace']); + self::assertSame('Bearer control', $request['headers']['Authorization']); + self::assertArrayHasKey('X-Durable-Workflow-Control-Plane-Version', $request['headers']); + self::assertArrayNotHasKey('X-Durable-Workflow-Protocol-Version', $request['headers']); + } + self::assertStringEndsWith('/api/workflow-version-usage?workflow_type=shipment&change_id=carrier-api&min_version=2&limit=50', $transport->requests[1]['uri']); + self::assertSame('/workflow-version-usage', MetricsRoutes::template('/workflow-version-usage?change_id=private&cursor=secret')); + } + + public function testOpaqueLongCursorAndExactStringsArePassedWithoutInspection(): void + { + $value = self::page(); + $value['workflow_type'] = " shipment/雪 "; + $value['change_id'] = 'carrier?name=one&two'; + $value['state'] = 'unknown'; + $value['through_run_id'] = 'run-9'; + $value['counts'] = ['total' => 4, 'legacy' => 1, 'current' => 1, 'unknown' => 2]; + $value['runs'] = [$value['runs'][2]]; + $value['returned_count'] = 1; + $value['limit'] = 1; + $value['has_more'] = true; + $value['next_cursor'] = 'opaque-next/+='; + $cursor = str_repeat('x', 9000).'+/=?&opaque'; + $transport = new FakeTransport([self::CAPABILITY, $value]); + $page = (new Client('https://runtime.example', namespace: 'orders', transport: $transport)) + ->workflowVersionUsage($value['workflow_type'], $value['change_id'], 2, 'unknown', 1, $cursor); + parse_str(parse_url($transport->requests[1]['uri'], PHP_URL_QUERY), $query); + self::assertSame($cursor, $query['cursor']); + self::assertSame($value['workflow_type'], $query['workflow_type']); + self::assertSame($value['change_id'], $query['change_id']); + self::assertSame('opaque-next/+=', $page->nextCursor); + self::assertSame(4, $page->counts->total, 'Counts precede the state filter.'); + self::assertSame(2, $page->counts->unknown); + } + + public function testSignedIntegerExtremesRemainExact(): void + { + foreach ([-2147483648, 2147483647] as $minimum) { + $value = self::page(); + $value['min_version'] = $minimum; + $value['runs'] = [$value['runs'][1]]; + $value['runs'][0]['recorded_version'] = $minimum; + $value['returned_count'] = 1; + $transport = new FakeTransport([self::CAPABILITY, $value]); + $page = (new Client('https://runtime.example', namespace: 'orders', transport: $transport)) + ->workflowVersionUsage('shipment', 'carrier-api', $minimum); + self::assertSame($minimum, $page->minVersion); + self::assertSame($minimum, $page->runs[0]->recordedVersion); + } + } + + public function testEachUnknownReasonIsPreservedWithoutGuessingFromDefinition(): void + { + foreach (['marker_absent', 'history_unavailable', 'history_invalid'] as $reason) { + $value = self::page(); + $value['runs'][2]['unknown_reason'] = $reason; + $page = WorkflowVersionUsagePage::fromArray($value); + self::assertSame($reason, $page->runs[2]->unknownReason); + self::assertNull($page->runs[2]->recordedVersion); + } + } + + #[DataProvider('invalidQueries')] + public function testInvalidQueriesFailBeforeDiscovery(array $arguments): void + { + $transport = new FakeTransport(); + try { + (new Client('https://runtime.example', transport: $transport))->workflowVersionUsage(...$arguments); + self::fail('Invalid query must be refused before a request.'); + } catch (InvalidArgumentException) { + self::assertSame([], $transport->requests); + } + } + + public static function invalidQueries(): iterable + { + foreach ([['workflowType' => ' '], ['workflowType' => "\xFF"], ['workflowType' => str_repeat('雪', 86)], + ['changeId' => "\u{2003}"], ['changeId' => str_repeat('x', 1025)], ['minVersion' => -2147483649], + ['minVersion' => 2147483648], ['state' => 'all'], ['limit' => 0], ['limit' => 201], + ['cursor' => ''], ['cursor' => "\xFF"]] as $change) { + yield [array_replace(['workflowType' => 'shipment', 'changeId' => 'carrier-api', 'minVersion' => 2], $change)]; + } + } + + public function testMissingOrFalseCapabilitiesHaveAnExplicitUnsupportedError(): void + { + foreach ([[], ['workflow_version_usage' => false], ['workflow_version_usage' => 'true']] as $capabilities) { + self::assertFalse(ClusterInfo::fromArray(['capabilities' => $capabilities])->supportsWorkflowVersionUsage()); + $transport = new FakeTransport([['capabilities' => $capabilities]]); + try { + (new Client('https://runtime.example', transport: $transport))->workflowVersionUsage('shipment', 'carrier-api', 2); + self::fail('No version-usage read without the capability.'); + } catch (ServerCapabilityUnavailable $error) { + self::assertSame('workflow_version_usage', $error->capability); + self::assertSame(501, $error->status); + self::assertCount(1, $transport->requests); + } + } + } + + public function testWorkerOnlyCredentialsCannotAuthorizeTheControlRead(): void + { + $transport = new FakeTransport(); + try { + (new Client('https://runtime.example', workerToken: 'worker-only', transport: $transport)) + ->workflowVersionUsage('shipment', 'carrier-api', 2); + self::fail('A worker credential cannot be used as a control credential.'); + } catch (InvalidArgumentException $error) { + self::assertStringContainsString('control credential', $error->getMessage()); + self::assertSame([], $transport->requests); + } + } + + public function testAuthenticationValidationStorageAndTimeoutErrorsRemainErrors(): void + { + foreach ([401, 403, 404, 422, 503] as $status) { + foreach ([true, false] as $discovery) { + $error = TransportException::fromResponse($status, ['reason' => 'retained-refusal'], ''); + $transport = new FakeTransport($discovery ? [$error] : [self::CAPABILITY, $error]); + try { + (new Client('https://runtime.example', transport: $transport))->workflowVersionUsage('shipment', 'carrier-api', 2); + self::fail('Server refusal cannot become an empty page.'); + } catch (ServerException $failure) { + self::assertSame($status, $failure->status); + self::assertSame('retained-refusal', $failure->reason); + self::assertCount($discovery ? 1 : 2, $transport->requests); + } + } + } + $transport = new FakeTransport([self::CAPABILITY, new TransportException('timeout')]); + $this->expectException(ServerException::class); + (new Client('https://runtime.example', transport: $transport))->workflowVersionUsage('shipment', 'carrier-api', 2); + } + + #[DataProvider('invalidResponses')] + public function testMalformedOrForeignResponsesCannotBecomeAnEmptySuccess(string $mutation): void + { + $value = self::page(); + match ($mutation) { + 'schema' => $value['schema_version'] = 2, + 'snapshot' => $value['consistency'] = 'snapshot', + 'count' => $value['counts']['total'] = 4, + 'bool-count' => $value['counts']['legacy'] = true, + 'unknown-version' => $value['runs'][2]['recorded_version'] = 2, + 'unknown-reason' => $value['runs'][2]['unknown_reason'] = null, + 'classification' => $value['runs'][0]['recorded_version'] = 3, + 'duplicate' => $value['runs'][1]['run_id'] = 'run-1', + 'upper-bound' => $value['through_run_id'] = 'run-2', + 'returned-count' => $value['returned_count'] = 2, + 'cursor' => $value['next_cursor'] = 'orphaned', + 'closed' => $value['runs'][0]['status'] = 'completed', + 'namespace' => $value['namespace'] = 'foreign', + 'type' => $value['workflow_type'] = 'other', + 'change' => $value['change_id'] = 'other', + 'minimum' => $value['min_version'] = 3, + 'limit' => $value['limit'] = 49, + 'state' => $value['state'] = 'legacy', + 'safe-to-remove' => $value['safe_to_remove'] = true, + 'state-counts' => $value['counts'] = ['total' => 3, 'legacy' => 0, 'current' => 2, 'unknown' => 1], + 'payload' => $value['result_envelope'] = self::unexpectedPayload(), + }; + $transport = new FakeTransport([self::CAPABILITY, $value]); + $this->expectException(UnexpectedValueException::class); + (new Client('https://runtime.example', namespace: 'orders', transport: $transport))->workflowVersionUsage('shipment', 'carrier-api', 2); + } + + public static function invalidResponses(): iterable + { + foreach (['schema', 'snapshot', 'count', 'bool-count', 'unknown-version', 'unknown-reason', 'classification', + 'duplicate', 'upper-bound', 'returned-count', 'cursor', 'closed', 'namespace', 'type', 'change', + 'minimum', 'limit', 'state', 'safe-to-remove', 'state-counts', 'payload'] as $mutation) { + yield $mutation => [$mutation]; + } + } + + public function testEmptyCohortIsDifferentFromUnknownHistory(): void + { + $value = self::page(); + $value['counts'] = ['total' => 0, 'legacy' => 0, 'current' => 0, 'unknown' => 0]; + $value['runs'] = []; + $value['returned_count'] = 0; + $value['through_run_id'] = null; + $page = WorkflowVersionUsagePage::fromArray($value); + self::assertSame([], $page->runs); + self::assertSame(0, $page->counts->total); + } + + public function testDiscoveryDoesNotHydrateApplicationPayloadFields(): void + { + $transport = new FakeTransport([self::CAPABILITY + ['result_envelope' => self::unexpectedPayload()], self::page()]); + $page = (new Client('https://runtime.example', namespace: 'orders', transport: $transport)) + ->workflowVersionUsage('shipment', 'carrier-api', 2); + self::assertSame(3, $page->counts->total); + self::assertCount(2, $transport->requests); + } + + /** @return array */ + private static function unexpectedPayload(): array + { + return ['codec' => 'avro', 'external_payload' => [ + 'schema' => 'durable-workflow.v2.runtime-external-payload-reference.v1', + 'reference_id' => 'ep_00000000000000000000000000', 'codec' => 'avro', + 'size_bytes' => 1, 'sha256' => str_repeat('a', 64), + ]]; + } +} From e0fcb500905008b21efd90fbbc743efec09108e0 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 14:51:10 +0000 Subject: [PATCH 2/3] Separate version-usage page shape and run validation --- src/Model/WorkflowVersionUsagePage.php | 55 ++++++++++++++++++-------- 1 file changed, 39 insertions(+), 16 deletions(-) diff --git a/src/Model/WorkflowVersionUsagePage.php b/src/Model/WorkflowVersionUsagePage.php index 86212a1e..e51e88c4 100644 --- a/src/Model/WorkflowVersionUsagePage.php +++ b/src/Model/WorkflowVersionUsagePage.php @@ -35,6 +35,30 @@ private function __construct( /** @param array $value */ public static function fromArray(array $value): self + { + self::requireShape($value); + self::requireTextFields($value); + $counts = WorkflowVersionUsageCounts::fromArray($value['counts']); + if ($value['returned_count'] !== count($value['runs']) || $value['returned_count'] > $value['limit'] + || $value['returned_count'] > $counts->total + || ($value['has_more'] !== ($value['next_cursor'] !== null)) + || ($value['has_more'] && $value['returned_count'] === 0) + || ($value['through_run_id'] === null && $counts->total !== 0)) { + throw new UnexpectedValueException('Workflow version-usage page boundaries do not reconcile.'); + } + $remaining = ['legacy' => $counts->legacy, 'current' => $counts->current, 'unknown' => $counts->unknown]; + $selectedCount = $value['state'] === null ? $counts->total : $remaining[$value['state']]; + if ($value['has_more'] && $selectedCount <= $value['returned_count']) { + throw new UnexpectedValueException('Workflow version-usage continuation has no remaining matching runs.'); + } + + return new self($value['namespace'], $value['workflow_type'], $value['change_id'], $value['min_version'], + $value['state'], $value['through_run_id'], $counts, self::readRuns($value, $remaining), $value['limit'], $value['returned_count'], + $value['has_more'], $value['next_cursor'], $value); + } + + /** @param array $value */ + private static function requireShape(array $value): void { $keys = array_keys($value); sort($keys); @@ -49,6 +73,11 @@ public static function fromArray(array $value): self || !is_array($value['runs']) || !array_is_list($value['runs'])) { throw new UnexpectedValueException('Invalid workflow version-usage v1 response.'); } + } + + /** @param array $value */ + private static function requireTextFields(array $value): void + { foreach (['namespace', 'workflow_type', 'change_id', 'through_run_id', 'next_cursor'] as $field) { if ($value[$field] === null && in_array($field, ['namespace', 'through_run_id', 'next_cursor'], true)) { continue; @@ -57,20 +86,16 @@ public static function fromArray(array $value): self throw new UnexpectedValueException('Invalid workflow version-usage response identity or cursor.'); } } - $counts = WorkflowVersionUsageCounts::fromArray($value['counts']); - if ($value['returned_count'] !== count($value['runs']) || $value['returned_count'] > $value['limit'] - || $value['returned_count'] > $counts->total - || ($value['has_more'] !== ($value['next_cursor'] !== null)) - || ($value['has_more'] && $value['returned_count'] === 0) - || ($value['through_run_id'] === null && $counts->total !== 0)) { - throw new UnexpectedValueException('Workflow version-usage page boundaries do not reconcile.'); - } + } + + /** + * @param array $value + * @param array $remaining + * @return list + */ + private static function readRuns(array $value, array $remaining): array + { $runs = []; - $remaining = ['legacy' => $counts->legacy, 'current' => $counts->current, 'unknown' => $counts->unknown]; - $selectedCount = $value['state'] === null ? $counts->total : $remaining[$value['state']]; - if ($value['has_more'] && $selectedCount <= $value['returned_count']) { - throw new UnexpectedValueException('Workflow version-usage continuation has no remaining matching runs.'); - } $previous = null; foreach ($value['runs'] as $row) { if (!is_array($row)) { @@ -89,8 +114,6 @@ public static function fromArray(array $value): self $previous = $run->runId; } - return new self($value['namespace'], $value['workflow_type'], $value['change_id'], $value['min_version'], - $value['state'], $value['through_run_id'], $counts, $runs, $value['limit'], $value['returned_count'], - $value['has_more'], $value['next_cursor'], $value); + return $runs; } } From bed228f1416ecd3ee140192652a3a847bced4c49 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 15:00:05 +0000 Subject: [PATCH 3/3] Enforce version-usage page completeness and cursor progress --- src/Client.php | 2 +- src/Model/WorkflowVersionUsagePage.php | 22 ++++++++++-- tests/WorkflowVersionUsageTest.php | 49 ++++++++++++++++++++++++++ 3 files changed, 70 insertions(+), 3 deletions(-) diff --git a/src/Client.php b/src/Client.php index d8ee1c01..5940e244 100644 --- a/src/Client.php +++ b/src/Client.php @@ -430,7 +430,7 @@ public function workflowVersionUsage( ]); $response = $this->request('GET', '/workflow-version-usage?'.http_build_query($query, '', '&', PHP_QUERY_RFC3986), false, payloadPaths: static fn (array $metadata): array => []); - $page = WorkflowVersionUsagePage::fromArray($response); + $page = WorkflowVersionUsagePage::fromArray($response, $cursor); if ($page->namespace !== $this->namespace || $page->workflowType !== $workflowType || $page->changeId !== $changeId || $page->minVersion !== $minVersion || $page->state !== $state || $page->limit !== $limit) { diff --git a/src/Model/WorkflowVersionUsagePage.php b/src/Model/WorkflowVersionUsagePage.php index e51e88c4..6260ec31 100644 --- a/src/Model/WorkflowVersionUsagePage.php +++ b/src/Model/WorkflowVersionUsagePage.php @@ -33,8 +33,11 @@ private function __construct( ) { } - /** @param array $value */ - public static function fromArray(array $value): self + /** + * @param array $value + * @param string|null $requestCursor The opaque cursor supplied for this exact page request. + */ + public static function fromArray(array $value, ?string $requestCursor = null): self { self::requireShape($value); self::requireTextFields($value); @@ -51,12 +54,27 @@ public static function fromArray(array $value): self if ($value['has_more'] && $selectedCount <= $value['returned_count']) { throw new UnexpectedValueException('Workflow version-usage continuation has no remaining matching runs.'); } + self::requireCursorProgress($value, $selectedCount, $requestCursor); return new self($value['namespace'], $value['workflow_type'], $value['change_id'], $value['min_version'], $value['state'], $value['through_run_id'], $counts, self::readRuns($value, $remaining), $value['limit'], $value['returned_count'], $value['has_more'], $value['next_cursor'], $value); } + /** @param array $value */ + private static function requireCursorProgress(array $value, int $selectedCount, ?string $requestCursor): void + { + if ($requestCursor === null && !$value['has_more'] && $value['returned_count'] !== $selectedCount) { + throw new UnexpectedValueException('The first terminal version-usage page must contain the complete selected cohort.'); + } + if ($value['has_more'] && $value['returned_count'] !== $value['limit']) { + throw new UnexpectedValueException('A version-usage continuation page must fill its requested limit.'); + } + if ($requestCursor !== null && $value['next_cursor'] === $requestCursor) { + throw new UnexpectedValueException('A version-usage continuation cursor must advance.'); + } + } + /** @param array $value */ private static function requireShape(array $value): void { diff --git a/tests/WorkflowVersionUsageTest.php b/tests/WorkflowVersionUsageTest.php index 0f767b27..01841d27 100644 --- a/tests/WorkflowVersionUsageTest.php +++ b/tests/WorkflowVersionUsageTest.php @@ -99,6 +99,8 @@ public function testSignedIntegerExtremesRemainExact(): void $value['runs'] = [$value['runs'][1]]; $value['runs'][0]['recorded_version'] = $minimum; $value['returned_count'] = 1; + $value['counts'] = ['total' => 1, 'legacy' => 0, 'current' => 1, 'unknown' => 0]; + $value['through_run_id'] = 'run-2'; $transport = new FakeTransport([self::CAPABILITY, $value]); $page = (new Client('https://runtime.example', namespace: 'orders', transport: $transport)) ->workflowVersionUsage('shipment', 'carrier-api', $minimum); @@ -252,6 +254,53 @@ public function testDiscoveryDoesNotHydrateApplicationPayloadFields(): void self::assertCount(2, $transport->requests); } + #[DataProvider('invalidPageProgress')] + public function testPageProgressRequiresRequestContext(string $mutation): void + { + $value = self::page(); + $state = $mutation === 'filtered-first-terminal' ? 'legacy' : null; + $cursor = $mutation === 'repeated-cursor' ? 'unchanged-opaque-cursor' : null; + $value['state'] = $state; + $value['runs'] = [$value['runs'][0]]; + $value['returned_count'] = 1; + if ($mutation === 'filtered-first-terminal') { + $value['runs'] = []; + $value['returned_count'] = 0; + } + if (in_array($mutation, ['short-has-more', 'repeated-cursor'], true)) { + $value['has_more'] = true; + $value['next_cursor'] = $cursor ?? 'next-opaque-cursor'; + } + if ($mutation === 'repeated-cursor') { + $value['limit'] = 1; + } + $transport = new FakeTransport([self::CAPABILITY, $value]); + $this->expectException(UnexpectedValueException::class); + (new Client('https://runtime.example', namespace: 'orders', transport: $transport)) + ->workflowVersionUsage('shipment', 'carrier-api', 2, state: $state, limit: $value['limit'], cursor: $cursor); + } + + public static function invalidPageProgress(): iterable + { + foreach (['unfiltered-first-terminal', 'filtered-first-terminal', 'short-has-more', 'repeated-cursor'] as $mutation) { + yield $mutation => [$mutation]; + } + } + + public function testLaterTerminalCursorPageCanBeEmptyWithRemainingCohortCounts(): void + { + $value = self::page(); + $value['runs'] = []; + $value['returned_count'] = 0; + $transport = new FakeTransport([self::CAPABILITY, $value]); + $page = (new Client('https://runtime.example', namespace: 'orders', transport: $transport)) + ->workflowVersionUsage('shipment', 'carrier-api', 2, cursor: 'earlier-opaque-cursor'); + self::assertSame(3, $page->counts->total); + self::assertSame([], $page->runs); + self::assertFalse($page->hasMore); + self::assertNull($page->nextCursor); + } + /** @return array */ private static function unexpectedPayload(): array {