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
7 changes: 7 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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::upsertSearchAttributes()` accepts an optional complete map
of canonical type declarations (#244). Explicitly typed calls bind replay to
both values and recorded type identity. Existing one-argument calls retain
Expand Down
27 changes: 27 additions & 0 deletions docs/portal/getting-started/client-setup.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
21 changes: 21 additions & 0 deletions docs/sdk-reference.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
128 changes: 127 additions & 1 deletion src/Client.php
Original file line number Diff line number Diff line change
Expand Up @@ -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. */
Expand Down Expand Up @@ -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);
}
Expand All @@ -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<int, WorkflowExecution>
*/
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<WorkflowRun> */
public function listWorkflowRuns(string $workflowId): array
{
Expand Down Expand Up @@ -595,6 +617,110 @@ 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<int, array<string, mixed>>
*/
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<string, mixed> $page
* @return list<array<string, mixed>>
*/
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<T>, 1: array<string, mixed>} $fetch
* @return \Generator<int, T>
*/
private function iteratePages(\Closure $fetch, ?string $token): \Generator
{
$checkpoint = $token;
$distance = 0;
$power = 1;
while (true) {
[$items, $page] = $fetch($token);
$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().
yield $item;
}
unset($items, $page, $item);
if ($next === null) {
return;
}
$token = $next;
}
}

/** @param array<string, mixed> $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<string, mixed> */
public function workflowActivities(string $workflowId, string $runId): array
{
Expand Down
14 changes: 14 additions & 0 deletions src/Exception/PaginationException.php
Original file line number Diff line number Diff line change
@@ -0,0 +1,14 @@
<?php

declare(strict_types=1);

namespace DurableWorkflow\Exception;

/** A malformed page cannot safely continue an enumeration. */
final class PaginationException extends DurableWorkflowException
{
public function __construct(string $message, public readonly string $reason)
{
parent::__construct($message);
}
}
Loading
Loading