From c147db2d07f0e53c2bdcba2b88c19352d20e4758 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 15:18:21 -0600 Subject: [PATCH 1/2] Add lazy workflow list and exact-run history iterators --- CHANGELOG.md | 7 + docs/portal/getting-started/client-setup.md | 27 +++ docs/sdk-reference.md | 21 ++ src/Client.php | 110 +++++++++- src/Exception/PaginationException.php | 14 ++ tests/ClientPaginationTest.php | 221 ++++++++++++++++++++ 6 files changed, 399 insertions(+), 1 deletion(-) create mode 100644 src/Exception/PaginationException.php create mode 100644 tests/ClientPaginationTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index b25f73b2..d0396167 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,13 @@ project follows [Semantic Versioning](https://semver.org/). ### Added +- `Client::iterateWorkflows()` and `iterateWorkflowHistory()` lazily read + complete workflow listings and exact-run histories, retaining one page at a + time. Existing one-page methods stay available. Iteration preserves opaque + cursors, namespace and filters, propagates later-page errors, and refuses + malformed continuations or token cycles instead of silently truncating or + looping (#247 in durable-workflow/.github). + - `WorkflowContext::startChildWorkflow()` starts a child and returns a `ChildWorkflowHandle` once the runtime has committed the start (#185). The handle's `identity()` gives the child call ID, the child workflow ID and the diff --git a/docs/portal/getting-started/client-setup.md b/docs/portal/getting-started/client-setup.md index bf9aab3e..c7c3d099 100644 --- a/docs/portal/getting-started/client-setup.md +++ b/docs/portal/getting-started/client-setup.md @@ -64,6 +64,33 @@ $result = $handle->result(timeoutSeconds: 30); The workflow ID is the stable instance identity. The run ID identifies one execution in that instance's chain. Ordinary handle methods follow the current run after continue-as-new; methods ending in `SelectedRun` keep the original run guard and fail instead of silently crossing to a successor. +## Read complete lists and histories + +`listWorkflows()` and `workflowHistory()` each return one bounded page. The +following iterator methods are unreleased: + +```php +foreach ($controlClient->iterateWorkflows(status: 'running', pageSize: 100) as $execution) { + echo $execution->workflowId, "\n"; +} +foreach ($controlClient->iterateWorkflowHistory($workflowId, $runId, pageSize: 100) as $event) { + echo $event['sequence'], ': ', $event['event_type'], "\n"; +} +``` + +Iterators retain one page and request the next only when iteration reaches it. +Stop with `break` to avoid further requests. The history iterator keeps the +exact workflow/run pair; it does not follow a new run or wait for new events +after the last page. Workflow listing follows the Server's current ordering +and consistency, so concurrent changes are not a snapshot. + +Handle errors around the entire loop. A later HTTP error propagates and closes +the generator; malformed metadata or a token cycle raises +`DurableWorkflow\Exception\PaginationException` with a typed `reason`. +Earlier items may have been yielded before an error. Iteration preserves raw +history event values and uses the same namespace, authentication and per-request +deadline as the one-page methods. + ## Inject a PSR transport when you need one Guzzle is included as the default transport. To reuse your application's PSR-18 client and PSR-17 factories, construct `Psr18Transport` and pass it as `transport:`. Authentication still belongs in an `Authentication` implementation; do not bury rotating role credentials in a generic HTTP middleware stack. diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index e5b38866..988f6367 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -304,6 +304,27 @@ run guard and fail rather than silently targeting a successor. ## Control-plane administration and discovery +`listWorkflows()` and `workflowHistory()` each read one page. The unreleased +iterator API follows the Server's continuation tokens lazily: + +```php +foreach ($client->iterateWorkflows(status: 'running', pageSize: 100) as $execution) { + echo $execution->workflowId, "\n"; +} +foreach ($client->iterateWorkflowHistory($workflowId, $runId, pageSize: 100) as $event) { + echo $event['sequence'], ': ', $event['event_type'], "\n"; +} +``` + +The list yields `WorkflowExecution` values; history yields the original event +maps. Both retain one page and fetch the next only when needed. Breaking out +does not prefetch. History stays on the selected run through continue-as-new; +listing follows the Server's ordering and consistency, not a global snapshot. +Missing or invalid page metadata and repeated/cyclic tokens throw +`PaginationException` with a typed `reason`. An HTTP error propagates unchanged +and closes the generator. Earlier items may already have been yielded, so +consumers must handle a partial export on failure. + `Client::withNamespace()` returns an immutable namespace selection that keeps the configured authentication and transport. Workflow payload encoding remains Apache Avro. Workflow diff --git a/src/Client.php b/src/Client.php index 9d27ddb5..c78caa9d 100644 --- a/src/Client.php +++ b/src/Client.php @@ -67,6 +67,7 @@ use DurableWorkflow\Worker\WorkerSessionOptions; use DurableWorkflow\Worker\ActivityLeaseBinding; use DurableWorkflow\Worker\ActivityLeaseProcess; +use DurableWorkflow\Exception\PaginationException; use InvalidArgumentException; /** Synchronous control-plane and worker-plane client for the standalone server. */ @@ -538,7 +539,7 @@ public function listWorkflows( ]); $response = $this->control('GET', $path); $executions = []; - foreach (($response['workflows'] ?? []) as $value) { + foreach ((is_array($response['workflows'] ?? null) ? $response['workflows'] : []) as $value) { if (is_array($value)) { $executions[] = WorkflowExecution::fromArray($value); } @@ -552,6 +553,27 @@ public function listWorkflows( ); } + /** + * Lazily enumerate workflow visibility results in the Server's order. + * This is not a snapshot across concurrent changes. + * + * @return \Generator + */ + public function iterateWorkflows( + ?string $workflowType = null, + ?string $status = null, + ?string $query = null, + ?int $pageSize = null, + ?string $nextPageToken = null, + ): \Generator { + return $this->iteratePages(function (?string $token) use ($workflowType, $status, $query, $pageSize): array { + $page = $this->listWorkflows($workflowType, $status, $query, $pageSize, $token); + $this->paginationItems($page->raw, 'workflows'); + + return [$page->executions, $page->raw]; + }, $nextPageToken); + } + /** @return list */ public function listWorkflowRuns(string $workflowId): array { @@ -595,6 +617,92 @@ public function workflowHistory( ); } + /** + * Lazily enumerate the raw events of one exact run. This does not follow + * continue-as-new or subscribe after the last page. + * + * @return \Generator> + */ + public function iterateWorkflowHistory( + string $workflowId, + string $runId, + ?int $pageSize = null, + ?string $nextPageToken = null, + ): \Generator { + return $this->iteratePages(function (?string $token) use ($workflowId, $runId, $pageSize): array { + $page = $this->workflowHistory($workflowId, $runId, $pageSize, $token); + + return [$this->paginationItems($page, 'events'), $page]; + }, $nextPageToken); + } + + /** + * @param array $page + * @return list> + */ + private function paginationItems(array $page, string $field): array + { + $items = $page[$field] ?? null; + if (!is_array($items) || !array_is_list($items)) { + throw new PaginationException('The page must contain an ordered '.$field.' list.', 'invalid_page'); + } + foreach ($items as $item) { + if (!is_array($item) || array_is_list($item)) { + throw new PaginationException('A page item must be an object.', 'invalid_item'); + } + } + + return $items; + } + + /** + * @template T + * @param \Closure(?string): array{0: list, 1: array} $fetch + * @return \Generator + */ + private function iteratePages(\Closure $fetch, ?string $token): \Generator + { + $checkpoint = $token; + $distance = 0; + $power = 1; + while (true) { + [$items, $page] = $fetch($token); + if (!array_key_exists('next_page_token', $page)) { + throw new PaginationException('The page omitted its continuation token.', 'missing_continuation'); + } + $next = $page['next_page_token']; + if ($next !== null && !is_string($next)) { + throw new PaginationException('A continuation token must be a string or null.', 'invalid_continuation'); + } + $next = $next === '' ? null : $next; + if ($next !== null) { + // Brent's cycle detection uses two tokens, matching the Rust + // iterator without retaining an unbounded export's token set. + if ($next === $token || $next === $checkpoint) { + throw new PaginationException('The Server repeated a continuation token or formed a token cycle.', 'repeated_page_token'); + } + if ($distance < PHP_INT_MAX) { + ++$distance; + } + if ($distance >= $power) { + $checkpoint = $next; + $power = $power > intdiv(PHP_INT_MAX, 2) ? PHP_INT_MAX : $power * 2; + $distance = 0; + } + } + foreach ($items as $item) { + // yield from would reuse each page's integer keys and could + // overwrite earlier items in iterator_to_array(). + yield $item; + } + unset($items, $page, $item); + if ($next === null) { + return; + } + $token = $next; + } + } + /** @return array */ public function workflowActivities(string $workflowId, string $runId): array { diff --git a/src/Exception/PaginationException.php b/src/Exception/PaginationException.php new file mode 100644 index 00000000..296fd4a4 --- /dev/null +++ b/src/Exception/PaginationException.php @@ -0,0 +1,14 @@ + [self::workflow('first'), self::workflow('second')], 'next_page_token' => 'opaque/+ ='], + ['workflows' => [], 'next_page_token' => 'last'], + ['workflows' => [self::workflow('third')], 'next_page_token' => null, 'workflow_count' => 999], + ]); + $original = new Client('https://server.example', transport: $transport, namespace: 'default', token: 'synthetic'); + $client = $original->withNamespace('orders'); + $iterator = $client->iterateWorkflows(workflowType: 'orders.process', status: 'running', query: 'CustomerId = "42"', pageSize: 2); + self::assertCount(0, $transport->requests); + self::assertInstanceOf(WorkflowExecution::class, $iterator->current()); + self::assertCount(1, $transport->requests); + $all = iterator_to_array($iterator); + self::assertSame([0, 1, 2], array_keys($all)); + self::assertSame(['first', 'second', 'third'], array_column($all, 'workflowId')); + self::assertCount(3, $transport->requests); + self::assertSame('default', $original->namespace); + foreach ($transport->requests as $index => $request) { + parse_str((string) parse_url($request['uri'], PHP_URL_QUERY), $query); + self::assertSame('orders.process', $query['workflow_type']); + self::assertSame('running', $query['status']); + self::assertSame('CustomerId = "42"', $query['query']); + self::assertSame('2', $query['page_size']); + self::assertSame([null, 'opaque/+ =', 'last'][$index], $query['next_page_token'] ?? null); + self::assertSame('orders', $request['headers']['X-Namespace']); + self::assertSame('2', $request['headers']['X-Durable-Workflow-Control-Plane-Version']); + self::assertSame('Bearer synthetic', $request['headers']['Authorization']); + } + } + + public function testHistoryIterationPreservesExactRunRawEventsAndOpaqueCursor(): void + { + $first = ['sequence' => 1, 'payload' => ['integer' => PHP_INT_MAX, 'unicode' => 'λ', 'nil' => null]]; + $second = ['sequence' => 2, 'event_type' => 'WorkflowCompleted']; + $transport = new FakeTransport([ + ['events' => [$first], 'next_page_token' => 'opaque/+ ='], + ['events' => [$second], 'next_page_token' => ''], + ]); + $client = new Client('https://server.example', transport: $transport, namespace: 'orders'); + $iterator = $client->iterateWorkflowHistory('workflow/a', 'run/b', 1, 'resume/+ ='); + self::assertCount(0, $transport->requests); + self::assertSame([$first, $second], iterator_to_array($iterator)); + self::assertCount(2, $transport->requests); + foreach ($transport->requests as $index => $request) { + self::assertSame('/api/workflows/workflow%2Fa/runs/run%2Fb/history', parse_url($request['uri'], PHP_URL_PATH)); + parse_str((string) parse_url($request['uri'], PHP_URL_QUERY), $query); + self::assertSame(['resume/+ =', 'opaque/+ ='][$index], $query['next_page_token']); + self::assertSame('1', $query['page_size']); + self::assertSame('orders', $request['headers']['X-Namespace']); + } + } + + /** @return iterable */ + public static function surfaces(): iterable + { + yield 'workflow list' => [false]; + yield 'run history' => [true]; + } + + #[DataProvider('surfaces')] + public function testAnEmptyTerminalPageStopsWithoutAnotherRequest(bool $history): void + { + $transport = new FakeTransport([[($history ? 'events' : 'workflows') => [], 'next_page_token' => null]]); + $iterator = self::iterator(new Client('https://server.example', transport: $transport), $history); + self::assertSame([], iterator_to_array($iterator)); + self::assertCount(1, $transport->requests); + } + + #[DataProvider('surfaces')] + public function testStoppingEarlyDoesNotPrefetch(bool $history): void + { + $transport = new FakeTransport([self::page($history, 'later')]); + $iterator = self::iterator(new Client('https://server.example', transport: $transport), $history); + foreach ($iterator as $_) { + break; + } + unset($iterator); + self::assertCount(1, $transport->requests); + } + + #[DataProvider('surfaces')] + public function testLaterPageErrorsPropagateOnceAndCloseTheGenerator(bool $history): void + { + $error = new ServerException('Denied', 403, 'not_authorized'); + $transport = new FakeTransport([self::page($history, 'later'), $error]); + $iterator = self::iterator(new Client('https://server.example', transport: $transport), $history); + $iterator->current(); + try { + $iterator->next(); + self::fail('The later-page error was swallowed.'); + } catch (ServerException $caught) { + self::assertSame($error, $caught); + } + self::assertFalse($iterator->valid()); + self::assertCount(2, $transport->requests); + } + + /** @return iterable, string}> */ + public static function invalidHistoryPages(): iterable + { + yield 'missing token' => [['events' => []], 'missing_continuation']; + yield 'integer token' => [['events' => [], 'next_page_token' => 3], 'invalid_continuation']; + yield 'object token' => [['events' => [], 'next_page_token' => ['private' => 'value']], 'invalid_continuation']; + yield 'missing items' => [['next_page_token' => null], 'invalid_page']; + yield 'non-list items' => [['events' => ['wrong' => []], 'next_page_token' => null], 'invalid_page']; + yield 'scalar items' => [['events' => true, 'next_page_token' => null], 'invalid_page']; + yield 'scalar row' => [['events' => ['private-value'], 'next_page_token' => null], 'invalid_item']; + } + + #[DataProvider('invalidHistoryPages')] + public function testMalformedHistoryCannotSilentlyTruncate(array $page, string $reason): void + { + $client = new Client('https://server.example', transport: new FakeTransport([$page])); + try { + iterator_to_array($client->iterateWorkflowHistory('workflow', 'run')); + self::fail('The malformed page was accepted.'); + } catch (PaginationException $error) { + self::assertSame($reason, $error->reason); + self::assertStringNotContainsString('private', $error->getMessage()); + } + } + + public function testMalformedWorkflowRowsAreNotSilentlyDroppedByTheIterator(): void + { + $client = new Client('https://server.example', transport: new FakeTransport([ + ['workflows' => [self::workflow('first'), 'private-value'], 'next_page_token' => null], + ])); + $this->expectException(PaginationException::class); + iterator_to_array($client->iterateWorkflows()); + } + + #[DataProvider('surfaces')] + public function testRepeatedTokenFailsBeforeARepeatedPageIsYielded(bool $history): void + { + $transport = new FakeTransport([self::page($history, 'private-token'), self::page($history, 'private-token')]); + $iterator = self::iterator(new Client('https://server.example', transport: $transport), $history); + $iterator->current(); + try { + $iterator->next(); + self::fail('The repeated token was accepted.'); + } catch (PaginationException $error) { + self::assertSame('repeated_page_token', $error->reason); + self::assertStringNotContainsString('private-token', $error->getMessage()); + } + self::assertFalse($iterator->valid()); + self::assertCount(2, $transport->requests); + } + + #[DataProvider('surfaces')] + public function testLongerTokenCycleTerminatesWithoutRetainingAllTokens(bool $history): void + { + $calls = 0; + $transport = new FakeTransport(handler: static function (string $method, string $uri) use ($history, &$calls): array { + self::assertLessThan(12, ++$calls, 'Token cycle was not detected.'); + parse_str((string) parse_url($uri, PHP_URL_QUERY), $query); + $next = match ($query['next_page_token'] ?? null) { + null => 'a', 'a' => 'b', 'b' => 'c', 'c' => 'a', + }; + + return [($history ? 'events' : 'workflows') => [], 'next_page_token' => $next]; + }); + $this->expectException(PaginationException::class); + iterator_to_array(self::iterator(new Client('https://server.example', transport: $transport), $history)); + } + + public function testFinishedPagesAreReleasedBeforeFetchingTheNextPage(): void + { + $first = null; + $calls = 0; + $transport = new FakeTransport(handler: static function () use (&$calls, &$first): array { + if (++$calls === 1) { + return ['workflows' => [self::workflow('first'), self::workflow('second')], 'next_page_token' => 'later']; + } + self::assertInstanceOf(WeakReference::class, $first); + self::assertNull($first->get(), 'The preceding page is still retained.'); + + return ['workflows' => [], 'next_page_token' => null]; + }); + $iterator = (new Client('https://server.example', transport: $transport))->iterateWorkflows(); + $first = WeakReference::create($iterator->current()); + $iterator->next(); + $iterator->next(); + self::assertFalse($iterator->valid()); + self::assertSame(2, $calls); + } + + /** @return array */ + private static function workflow(string $id): array + { + return ['workflow_id' => $id, 'run_id' => 'run-'.$id, 'workflow_type' => 'orders.process', 'status' => 'running']; + } + + /** @return array */ + private static function page(bool $history, ?string $next): array + { + return [($history ? 'events' : 'workflows') => [$history ? ['sequence' => 1] : self::workflow('first')], 'next_page_token' => $next]; + } + + private static function iterator(Client $client, bool $history): \Generator + { + return $history ? $client->iterateWorkflowHistory('workflow', 'run') : $client->iterateWorkflows(); + } +} From 2d515438ab4943a81ab55350c4550b34786f145c Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 17:03:05 -0600 Subject: [PATCH 2/2] Separate continuation validation from iterator cycle tracking --- src/Client.php | 64 ++++++++++++++++++++++++++++++++------------------ 1 file changed, 41 insertions(+), 23 deletions(-) diff --git a/src/Client.php b/src/Client.php index c78caa9d..12e71453 100644 --- a/src/Client.php +++ b/src/Client.php @@ -667,29 +667,8 @@ private function iteratePages(\Closure $fetch, ?string $token): \Generator $power = 1; while (true) { [$items, $page] = $fetch($token); - if (!array_key_exists('next_page_token', $page)) { - throw new PaginationException('The page omitted its continuation token.', 'missing_continuation'); - } - $next = $page['next_page_token']; - if ($next !== null && !is_string($next)) { - throw new PaginationException('A continuation token must be a string or null.', 'invalid_continuation'); - } - $next = $next === '' ? null : $next; - if ($next !== null) { - // Brent's cycle detection uses two tokens, matching the Rust - // iterator without retaining an unbounded export's token set. - if ($next === $token || $next === $checkpoint) { - throw new PaginationException('The Server repeated a continuation token or formed a token cycle.', 'repeated_page_token'); - } - if ($distance < PHP_INT_MAX) { - ++$distance; - } - if ($distance >= $power) { - $checkpoint = $next; - $power = $power > intdiv(PHP_INT_MAX, 2) ? PHP_INT_MAX : $power * 2; - $distance = 0; - } - } + $next = $this->continuationToken($page); + self::advancePaginationCycle($token, $next, $checkpoint, $distance, $power); foreach ($items as $item) { // yield from would reuse each page's integer keys and could // overwrite earlier items in iterator_to_array(). @@ -703,6 +682,45 @@ private function iteratePages(\Closure $fetch, ?string $token): \Generator } } + /** @param array $page */ + private function continuationToken(array $page): ?string + { + if (!array_key_exists('next_page_token', $page)) { + throw new PaginationException('The page omitted its continuation token.', 'missing_continuation'); + } + $next = $page['next_page_token']; + if ($next !== null && !is_string($next)) { + throw new PaginationException('A continuation token must be a string or null.', 'invalid_continuation'); + } + + return $next === '' ? null : $next; + } + + private static function advancePaginationCycle( + ?string $current, + ?string $next, + ?string &$checkpoint, + int &$distance, + int &$power, + ): void { + if ($next === null) { + return; + } + // Brent's cycle detection retains two tokens, matching Rust, rather + // than retaining every token in an unbounded export. + if ($next === $current || $next === $checkpoint) { + throw new PaginationException('The Server repeated a continuation token or formed a token cycle.', 'repeated_page_token'); + } + if ($distance < PHP_INT_MAX) { + ++$distance; + } + if ($distance >= $power) { + $checkpoint = $next; + $power = $power > intdiv(PHP_INT_MAX, 2) ? PHP_INT_MAX : $power * 2; + $distance = 0; + } + } + /** @return array */ public function workflowActivities(string $workflowId, string $runId): array {