diff --git a/CHANGELOG.md b/CHANGELOG.md index 1d84649..b887f45 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,33 @@ # Changelog +## 0.9.0 + +- Synced with Apify OpenAPI spec `v2-2026-10-08T073338Z`. +- Added a `waitForResources` option to `ActorStartOptions`/`TaskStartOptions` (`ActorClient`/ + `TaskClient` `start()`/`call()`): retries a run start the API rejects for lack of account memory or a + concurrent-run slot (`actor-memory-limit-exceeded`/`concurrent-runs-limit-exceeded`) every 10 + seconds. `true` retries until the run starts; an `int` bounds the retrying in seconds and then throws + the last error. In `call()`, time spent retrying does not count toward `$waitSecs`. +- Fixed the offset/limit lazy iterators (`iterate()` on the Actor/build/run/task/schedule/webhook/store + collections, and `DatasetClient::iterateItems()`) to stop only once a page returns nothing, instead of + capping at the `total` the first page reported. A collection that grows while being iterated (e.g. a + dataset a run keeps writing to) is no longer cut off at its starting size; with no explicit `limit`, + iteration now always ends with one extra request that finds nothing left. + **Breaking:** this costs one extra request per unbounded iteration that previously ended exactly on a + full last page. + `ActorEnvVarCollectionClient`/`ActorVersionCollectionClient::iterate()` no longer accept + a `$chunkSize`, and `ActorVersionCollectionClient::list()`/`iterate()` no longer accept a `ListOptions` + argument: both endpoints ignore pagination and always return everything in one response, so paging + them generically could duplicate items after the fix above; `iterate()` on both now makes exactly one + request instead, matching the reference client. +- `RunClient::waitForFinish()`/`BuildClient::waitForFinish()` now give up and throw a + `NotFoundException` after the run/build keeps 404ing for 3 seconds, instead of polling 404s for the + rest of the (by default effectively indefinite) wait budget. A successful poll in between resets this + grace window. +- Added a `$chunkSize` parameter to `KeyValueStoreClient::iterateKeys()`, capping how many keys are + requested per page independently of the total `limit`, matching the offset/limit collection + iterators. + ## 0.8.0 - Added `InvalidRequestException` (400), `UnauthorizedException` (401), `ForbiddenException` (403), diff --git a/docs/actors.md b/docs/actors.md index a9d3edb..91dd2fb 100644 --- a/docs/actors.md +++ b/docs/actors.md @@ -41,6 +41,11 @@ Addressed by ID or `username~name` (a `username/name` is accepted too). `$input` is a JSON-serializable array (the normal case), or a raw `string` of bytes sent exactly as given — pair a raw string with a non-default `$options->contentType` (e.g. a scraper that takes a plain-text URL list instead of JSON). + +`$options->waitForResources` retries a start the API rejects for lack of account memory or a free +concurrent-run slot (`actor-memory-limit-exceeded`/`concurrent-runs-limit-exceeded`), every 10 +seconds: `true` retries until the run starts, an `int` stops retrying after that many seconds and +throws the last error. In `call()`, time spent retrying does not count toward `$waitSecs`. - `build(string $versionNumber, ?ActorBuildOptions $options = null): Build` - `defaultBuild(?int $waitForFinish = null): BuildClient` - `lastRun(?LastRunOptions $options = null): RunClient` @@ -56,9 +61,12 @@ $lastSucceeded = $client->actor('apify/hello-world')->lastRun(new LastRunOptions ## Actor versions — `$client->actor($id)->versions()` / `->version($n)` -- Collection: `list(?ListOptions): PaginationList`, `iterate(?ListOptions $options = null, ?int $chunkSize = null): iterable`, `create(mixed $version): ActorVersion`. +- Collection: `list(): PaginationList`, `iterate(): iterable`, `create(mixed $version): ActorVersion`. - Single: `get(): ?ActorVersion`, `update(mixed $newFields): ActorVersion`, `delete(): void`. +The versions endpoint takes no filtering or pagination parameters — it always returns every version +in one response — so `list()`/`iterate()` take no options and `iterate()` makes a single request. + ```php $version = $client->actor('me~my-actor')->versions()->create([ 'versionNumber' => '0.1', @@ -69,11 +77,11 @@ $version = $client->actor('me~my-actor')->versions()->create([ ## Environment variables — `->version($n)->envVars()` / `->envVar($name)` -- Collection: `list(): PaginationList`, `iterate(?int $chunkSize = null): iterable`, `create(ActorEnvVar $envVar): ActorEnvVar`. +- Collection: `list(): PaginationList`, `iterate(): iterable`, `create(ActorEnvVar $envVar): ActorEnvVar`. - Single: `get(): ?ActorEnvVar`, `update(ActorEnvVar $envVar): ActorEnvVar`, `delete(): void`. -`iterate()` on the environment-variable collection takes only the optional `$chunkSize` (per-page -size); the endpoint has no filters, mirroring the reference client's parameterless iterator. +This endpoint likewise takes no filtering or pagination parameters, so `iterate()` makes a single +request, mirroring the reference client's parameterless iterator. ```php $client->actor('me~my-actor')->version('0.0')->envVars()->create(new ActorEnvVar('API_KEY', 'secret', isSecret: true)); diff --git a/docs/options.md b/docs/options.md index 847121d..a578992 100644 --- a/docs/options.md +++ b/docs/options.md @@ -115,11 +115,12 @@ For `actor()->start()` / `actor()->call()`. | `restartOnError` | `?bool` | Restart the run if it fails. | | `forcePermissionLevel` | `?string` | Override the Actor's permission level (`LIMITED_PERMISSIONS`/`FULL_PERMISSIONS`). | | `webhooks` | `array\|null` | Ad-hoc webhooks to attach to the run (base64-encoded before sending). | +| `waitForResources` | `bool\|int\|null` | Retry the start every 10s while the account lacks resources for the run (`true` = retry until it starts, `int` = stop after that many seconds). | ### `TaskStartOptions` For `task()->start()` / `task()->call()`. Same fields as `ActorStartOptions` except it has no `contentType` or `forcePermissionLevel`: `build`, `memoryMbytes`, `timeoutSecs`, `waitForFinish`, -`maxItems`, `maxTotalChargeUsd`, `restartOnError`, `webhooks`. +`maxItems`, `maxTotalChargeUsd`, `restartOnError`, `webhooks`, `waitForResources`. ### `ValidateInputOptions` For `actor()->validateInput()`. diff --git a/docs/runs.md b/docs/runs.md index a728649..14fdbd7 100644 --- a/docs/runs.md +++ b/docs/runs.md @@ -46,7 +46,9 @@ $items = $client->run('RUN_ID')->dataset()->listItems(); - **`waitForFinish(?int $waitSecs = null)`** — the client-side helper method (on runs and builds). It polls until the run/build reaches a terminal state, transparently issuing repeated server-side waits. `$waitSecs` is the total budget in seconds and is **not** capped; `null` waits indefinitely. - For instance, `waitForFinish(300)` waits up to five minutes. + For instance, `waitForFinish(300)` waits up to five minutes. A 404 (replica lag right after the run + or build is created) is tolerated for up to 3 seconds; if it keeps 404ing past that, the wait gives + up and throws a `NotFoundException` instead of polling for the rest of `$waitSecs`. - **The server-side `waitForFinish` parameter** — `get(?int $waitForFinishSecs = null)` (and `defaultBuild()`) and the `waitForFinish` field on `*Options` (e.g. `ActorStartOptions`). This is a single API-side wait and the server caps it at 60 seconds, so the client clamps larger values. diff --git a/docs/storages.md b/docs/storages.md index 710f4f2..ebe3c97 100644 --- a/docs/storages.md +++ b/docs/storages.md @@ -50,7 +50,7 @@ Single — `$client->keyValueStore($id)`: - `get(): ?KeyValueStore`, `update(mixed $newFields): KeyValueStore`, `delete(): void` - `listKeys(?ListKeysOptions $options = null): KeyValueStoreKeysPage` -- `iterateKeys(?ListKeysOptions $options = null): iterable` — lazily iterate all keys, following cursor pagination (`exclusiveStartKey`/`nextExclusiveStartKey`). The options' `limit` caps the total number of keys yielded across all pages (unset = all); there is no separate page-size argument (the per-page size follows the remaining cap, like the reference client). +- `iterateKeys(?ListKeysOptions $options = null, ?int $chunkSize = null): iterable` — lazily iterate all keys, following cursor pagination (`exclusiveStartKey`/`nextExclusiveStartKey`). The options' `limit` caps the total number of keys yielded across all pages (unset = all); `$chunkSize` is the per-page size (unset = the remaining total cap, or the server default when unbounded). - `recordExists(string $key): bool` — reports whether a record with the given key exists, without downloading its value (a `HEAD` request). - `getRecord(string $key, ?GetRecordOptions $options = null): ?KeyValueStoreRecord` - `setRecord(string $key, string $value, string $contentType, ?SetRecordOptions $options = null): void` diff --git a/docs/tasks.md b/docs/tasks.md index 43e640d..5c50de8 100644 --- a/docs/tasks.md +++ b/docs/tasks.md @@ -30,7 +30,9 @@ foreach ($client->tasks()->iterate(new ListOptions(), 50) as $t) { `publicConfig.inputSchemaFields`/`publicConfig.datasetView` to be set. An Actor can have up to 10 published tasks and an account up to 100; if the conditions aren't met, the request fails and nothing is changed. -- `start(mixed $input = null, ?TaskStartOptions $options = null): ActorRun` +- `start(mixed $input = null, ?TaskStartOptions $options = null): ActorRun` — `$options->waitForResources` + retries a start the API rejects for lack of account resources; see the Actor docs' `start()` for the + full behavior. - `call(mixed $input = null, ?TaskStartOptions $options = null, ?int $waitSecs = null): ActorRun` - `getInput(): mixed` (throws if the task itself does not exist — a 404 here cannot mean anything else), `updateInput(mixed $input): mixed` diff --git a/src/Internal/ResourceContext.php b/src/Internal/ResourceContext.php index 8adbf5a..40f19a3 100644 --- a/src/Internal/ResourceContext.php +++ b/src/Internal/ResourceContext.php @@ -41,6 +41,12 @@ final class ResourceContext */ private const MAX_WAIT_FOR_FINISH_SECS = 999999; + /** + * How long {@see waitForFinish()} keeps polling a job that returns 404, to ride out replica lag, + * before it gives up and throws. A successful poll in between resets this window. + */ + private const MAX_WAIT_WHEN_JOB_NOT_FOUND_MILLIS = 3000; + public QueryParams $baseParams; private string $apiOrigin; @@ -249,6 +255,12 @@ public function listResource(string $subPath, QueryParams $params, callable $hyd * ({@code null} = the server default). The two are independent — {@code $limit} is never reused * as the page size. {@code $startOffset} is the offset of the first page. * + * The next offset and the stop condition follow the number of items a page returned (capped at + * what that page's request asked for) — never the listing's reported {@code total}, which can grow + * while the iteration runs (e.g. a resource collection gaining entries). With no {@code $limit}, + * the loop therefore always ends with one request that finds no more items, matching the + * reference client exactly. + * * @template T * @param callable(int,?int):PaginationList $fetchPage receives (offset, pageLimit) and returns that page * @return Generator @@ -257,34 +269,49 @@ public static function paginateOffset(int $startOffset, ?int $limit, ?int $chunk { // First page: request min(limit, chunkSize) items. A null/0 on either side means "unbounded", // so the other bound wins (mirrors the reference client's minForLimitParam). - $page = $fetchPage($startOffset, self::minLimit($limit, $chunkSize)); - $items = $page->getItems(); - foreach ($items as $item) { + $firstPageLimit = self::minLimit($limit, $chunkSize); + $page = $fetchPage($startOffset, $firstPageLimit); + foreach ($page->getItems() as $item) { yield $item; } - $total = $page->getTotal(); - // Effective total cap: the requested limit (0/null => all, i.e. capped only by what exists). - // The outer bound by $total is redundant here: the next line's min($total - $startOffset, $cap) - // already bounds the result by $total (since $startOffset >= 0). - $cap = ($limit !== null && $limit > 0) ? $limit : $total; - $currentOffset = $startOffset + count($items); - // Items still to yield, bounded both by what remains after the start offset and by the cap. - $remaining = min($total - $startOffset, $cap) - count($items); - - // Guard on the previous page being non-empty so an over-reported total (a page shorter than - // its claimed total) terminates instead of looping forever. - while (count($items) > 0 && $remaining > 0) { - $page = $fetchPage($currentOffset, self::minLimit($remaining, $chunkSize)); - $items = $page->getItems(); - foreach ($items as $item) { + $pageCount = self::advancedBy($page); + $currentOffset = $startOffset + $pageCount; + // Items still to yield, or null for "unbounded" (no $limit given). + $remaining = ($limit !== null && $limit > 0) ? $limit - $pageCount : null; + + // Two distinct ways to stop: a page that returns nothing (the listing is exhausted), or + // $remaining having already reached 0 (the caller's $limit is satisfied) — either ends the loop. + while ($pageCount > 0 && ($remaining === null || $remaining > 0)) { + $pageLimit = self::minLimit($remaining, $chunkSize); + $page = $fetchPage($currentOffset, $pageLimit); + foreach ($page->getItems() as $item) { yield $item; } - $currentOffset += count($items); - $remaining -= count($items); + $pageCount = self::advancedBy($page); + $currentOffset += $pageCount; + if ($remaining !== null) { + $remaining -= $pageCount; + } } } + /** + * How far a page advances the offset: the number of items it returned. Every endpoint still using + * this generic loop honors the {@code limit} it was sent — unlike the dataset-items endpoint (see + * {@see \Apify\Client\Resource\DatasetClient::scannedRows()}), none of them has an {@code unwind}- + * like transform that can make a page return more items than its requested limit, and an endpoint + * that ignores {@code offset}/{@code limit} entirely (env vars, Actor versions) bypasses this loop + * rather than going through it — so the returned count never needs capping to the requested limit. + * + * @template T + * @param PaginationList $page + */ + private static function advancedBy(PaginationList $page): int + { + return count($page->getItems()); + } + /** * Returns the smaller of two optional positive bounds, treating {@code null} or {@code 0} as * "unbounded" (the API treats {@code limit=0} as unset). Mirrors the reference minForLimitParam. @@ -475,8 +502,12 @@ public function clampServerWait(?int $waitForFinishSecs): ?int /** * Polls a GET endpoint with {@code waitForFinish} until the resource reaches a terminal state or * the wait budget elapses. {@code waitSecs === null} means "wait indefinitely", implemented as a - * finite but very large bound so the loop always terminates. A transient 404 (replica lag) is - * treated as "not yet available". + * finite but very large bound so the loop always terminates. + * + * A 404 is treated as "not yet available" (replica lag) rather than failing immediately, but only + * for {@see MAX_WAIT_WHEN_JOB_NOT_FOUND_MILLIS}: if the resource keeps 404ing for that long, the + * wait gives up and rethrows that 404 instead of polling for the rest of {@code $waitSecs} (which + * defaults to effectively indefinite). A successful poll in between resets this grace window. * * @param callable(array):bool $isTerminal * @return array @@ -492,6 +523,8 @@ public function waitForFinish(?int $waitSecs, string $resourceName, callable $is $resource = null; $present = false; + $notFoundSinceMillis = null; + $notFoundException = null; while (true) { $elapsed = self::nowMillis() - $start; @@ -501,24 +534,39 @@ public function waitForFinish(?int $waitSecs, string $resourceName, callable $is $params = new QueryParams(); $params->addInt('waitForFinish', $requestSecs); - $data = $this->getResource('', $params); - if (is_array($data)) { - $resource = $data; - $present = true; - if ($isTerminal($resource)) { - return $resource; + try { + $data = $this->getResourceRequired('', $params); + if (is_array($data)) { + $resource = $data; + $present = true; + $notFoundSinceMillis = null; + if ($isTerminal($resource)) { + return $resource; + } } + } catch (ApifyApiException $e) { + if (!HttpClientCore::isNotFound($e)) { + throw $e; + } + $notFoundSinceMillis ??= self::nowMillis(); + $notFoundException = $e; } if (self::nowMillis() - $start >= $budgetMillis) { break; } + if ($notFoundSinceMillis !== null && self::nowMillis() - $notFoundSinceMillis >= self::MAX_WAIT_WHEN_JOB_NOT_FOUND_MILLIS) { + break; + } usleep((int) (self::WAIT_POLL_INTERVAL_SECS * 1_000_000)); } if ($present && $resource !== null) { return $resource; } + if ($notFoundException !== null) { + throw $notFoundException; + } throw new RuntimeException( sprintf('waiting for %s to finish failed: cannot fetch %s details from the server', $resourceName, $resourceName) ); diff --git a/src/Internal/WaitForResources.php b/src/Internal/WaitForResources.php new file mode 100644 index 0000000..87e86ef --- /dev/null +++ b/src/Internal/WaitForResources.php @@ -0,0 +1,80 @@ +getType(), self::RESOURCE_LIMIT_ERROR_TYPES, true)) { + throw $e; + } + $delaySecs = $cooldownSecs; + if ($deadline !== null) { + $remainingSecs = $deadline - microtime(true); + if ($remainingSecs <= 0) { + throw $e; + } + $delaySecs = min($delaySecs, $remainingSecs); + } + usleep((int) round($delaySecs * 1_000_000)); + } + } + } +} diff --git a/src/Options/ActorStartOptions.php b/src/Options/ActorStartOptions.php index d15fe5c..fa061bd 100644 --- a/src/Options/ActorStartOptions.php +++ b/src/Options/ActorStartOptions.php @@ -40,6 +40,22 @@ public function __construct( */ public readonly ?string $forcePermissionLevel = null, public readonly ?array $webhooks = null, + /** + * Retry the start while the account lacks the resources for the run, that is while the API + * rejects it with an {@see \Apify\Client\Exception\ApifyApiException} of type + * {@code actor-memory-limit-exceeded} or {@code concurrent-runs-limit-exceeded}. Both clear as + * other runs or builds finish. The start is retried every 10 seconds, and any other error is + * thrown right away. + * + * {@code true} retries until the run starts. An {@code int} number of seconds stops retrying + * after that many seconds and throws the last error. {@code null} (the default) makes a single + * attempt. + * + * A run that requests more memory than the whole memory limit of the account is rejected with + * {@code actor-memory-limit-exceeded} as well and never starts, so {@code true} retries it + * forever. + */ + public readonly bool|int|null $waitForResources = null, ) { } diff --git a/src/Options/TaskStartOptions.php b/src/Options/TaskStartOptions.php index 67ce5dd..0888bc9 100644 --- a/src/Options/TaskStartOptions.php +++ b/src/Options/TaskStartOptions.php @@ -33,6 +33,11 @@ public function __construct( /** If {@code true}, restart the run if it fails. */ public readonly ?bool $restartOnError = null, public readonly ?array $webhooks = null, + /** + * Retry the start while the account lacks the resources for the run. See {@see + * \Apify\Client\Options\ActorStartOptions::$waitForResources} for the full behavior. + */ + public readonly bool|int|null $waitForResources = null, ) { } diff --git a/src/Resource/ActorClient.php b/src/Resource/ActorClient.php index 045e6a3..3097d38 100644 --- a/src/Resource/ActorClient.php +++ b/src/Resource/ActorClient.php @@ -9,6 +9,7 @@ use Apify\Client\Internal\QueryParams; use Apify\Client\Internal\ResourceContext; use Apify\Client\Internal\TimeoutTiers; +use Apify\Client\Internal\WaitForResources; use Apify\Client\Model\Actor; use Apify\Client\Model\ActorRun; use Apify\Client\Model\Build; @@ -83,13 +84,16 @@ public function start( $params = new QueryParams(); $options->appendTo($params); $body = ResourceContext::encodeInputBody($input); - return new ActorRun($this->ctx->postWithBody( - 'runs', - $params, - $body, - $options->contentTypeOrDefault(), - $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_MEDIUM) - )); + return WaitForResources::start( + fn () => new ActorRun($this->ctx->postWithBody( + 'runs', + $params, + $body, + $options->contentTypeOrDefault(), + $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_MEDIUM) + )), + $options->waitForResources, + ); } /** @@ -97,7 +101,8 @@ public function start( * * @param mixed $input a JSON-serializable array (or {@code null} for no input), or a raw * string of bytes sent exactly as given — see {@see start()} - * @param int|null $waitSecs bounds the wait; {@code null} waits indefinitely + * @param int|null $waitSecs bounds the wait; {@code null} waits indefinitely. Time spent retrying a + * {@code $options->waitForResources} start does not count toward it. */ public function call(mixed $input = null, ?ActorStartOptions $options = null, ?int $waitSecs = null): ActorRun { diff --git a/src/Resource/ActorEnvVarCollectionClient.php b/src/Resource/ActorEnvVarCollectionClient.php index 517103d..b655040 100644 --- a/src/Resource/ActorEnvVarCollectionClient.php +++ b/src/Resource/ActorEnvVarCollectionClient.php @@ -42,24 +42,18 @@ public function list(int|float|string|null $timeoutSecs = null): PaginationList } /** - * Lazily iterates over the version's environment variables, fetching pages on demand. - * {@code $chunkSize} caps the per-page size ({@code null} = the server default). This endpoint is - * not filtered, so iteration mirrors the reference client's parameterless {@code list()} iterator. + * Lazily iterates over the version's environment variables. + * + * This endpoint ignores {@code offset}/{@code limit} and always returns every environment + * variable in one response, so iteration makes exactly one request — unlike the offset/limit + * collection iterators, there is no per-page {@code chunkSize} to request fewer at a time, and a + * generic paginating loop would re-request (and re-yield) the same full list forever. * * @return Generator */ - public function iterate(?int $chunkSize = null, int|float|string|null $timeoutSecs = null): Generator + public function iterate(int|float|string|null $timeoutSecs = null): Generator { - return ResourceContext::paginateOffset( - 0, - null, - $chunkSize, - function (int $offset, ?int $pageLimit) use ($timeoutSecs) { - $params = new QueryParams(); - $params->addInt('offset', $offset)->addInt('limit', $pageLimit); - return $this->ctx->listResource('', $params, static fn (array $d) => ActorEnvVar::fromArray($d), $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_MEDIUM)); - }, - ); + yield from $this->list($timeoutSecs)->getItems(); } /** Creates a new environment variable. */ diff --git a/src/Resource/ActorVersionCollectionClient.php b/src/Resource/ActorVersionCollectionClient.php index 5a76e10..9896af4 100644 --- a/src/Resource/ActorVersionCollectionClient.php +++ b/src/Resource/ActorVersionCollectionClient.php @@ -10,7 +10,6 @@ use Apify\Client\Internal\TimeoutTiers; use Apify\Client\Model\ActorVersion; use Apify\Client\Model\PaginationList; -use Apify\Client\Options\ListOptions; use Generator; /** A client for an Actor's version collection ({@code GET/POST /v2/actors/{actorId}/versions}). */ @@ -25,33 +24,29 @@ public function __construct(HttpClientCore $http, string $actorUrl) } /** - * Lists the Actor's versions. + * Lists the Actor's versions. The endpoint takes no filtering or pagination parameters: it always + * returns every version in one response. * * @return PaginationList */ - public function list(?ListOptions $options = null, int|float|string|null $timeoutSecs = null): PaginationList + public function list(int|float|string|null $timeoutSecs = null): PaginationList { - $params = new QueryParams(); - ($options ?? new ListOptions())->appendTo($params); - return $this->ctx->listResource('', $params, static fn (array $d) => new ActorVersion($d), $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_MEDIUM)); + return $this->ctx->listResource('', new QueryParams(), static fn (array $d) => new ActorVersion($d), $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_MEDIUM)); } /** - * Lazily iterates over the Actor's versions, fetching pages on demand. The options' {@code limit} - * caps the total number of versions yielded across all pages ({@code null} = all); - * {@code $chunkSize} is the per-page size ({@code null} = the server default). + * Lazily iterates over the Actor's versions. + * + * This endpoint ignores {@code offset}/{@code limit} and always returns every version in one + * response, so iteration makes exactly one request — unlike the offset/limit collection iterators, + * there is no per-page {@code chunkSize} to request fewer at a time, and a generic paginating loop + * would re-request (and re-yield) the same full list forever. * * @return Generator */ - public function iterate(?ListOptions $options = null, ?int $chunkSize = null, int|float|string|null $timeoutSecs = null): Generator + public function iterate(int|float|string|null $timeoutSecs = null): Generator { - $options ??= new ListOptions(); - return ResourceContext::paginateOffset( - $options->offset ?? 0, - $options->limit, - $chunkSize, - fn (int $offset, ?int $pageLimit) => $this->list($options->withPagination($offset, $pageLimit), $timeoutSecs), - ); + yield from $this->list($timeoutSecs)->getItems(); } /** diff --git a/src/Resource/DatasetClient.php b/src/Resource/DatasetClient.php index 96668b5..bbb41f4 100644 --- a/src/Resource/DatasetClient.php +++ b/src/Resource/DatasetClient.php @@ -112,8 +112,18 @@ public function listItems(?DatasetListItemsOptions $options = null, int|float|st * terminates by the scanned count the API reports in the {@code X-Apify-Pagination-Count} header — * never by the returned item count — matching the reference JS client exactly (including its * fallback to the item count on a response that omits the header, e.g. behind a proxy that strips - * it). ({@code skipHidden} only strips hidden fields from each item, not whole items, so it does - * not affect paging.) + * it, and its use of whichever of the two is larger, since the header can lag a fresh push). + * ({@code skipHidden} only strips hidden fields from each item, not whole items, so it does not + * affect paging.) + * + * The scanned count is also capped at what the page's request asked for — see {@see scannedRows()} + * for why {@code unwind} specifically makes that cap necessary here (unlike the offset/limit + * collection iterators in {@see \Apify\Client\Internal\ResourceContext::paginateOffset()}, which + * have no such cap because none of their endpoints has an item-multiplying transform). The + * listing's reported {@code total} is never consulted for the stop condition, there or here: it can + * grow while the iteration runs (e.g. a run still pushing to the dataset being read), so with no + * {@code $limit} the loop always ends with one request that scans nothing, rather than stopping at + * the size the dataset had when it started. * * @return Generator */ @@ -124,30 +134,57 @@ public function iterateItems(?DatasetListItemsOptions $options = null, ?int $chu $limit = $options->limit; $resolved = $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_LONG); - [$page, $scanned] = $this->fetchItemsPage($options->withPagination($startOffset, ResourceContext::minLimit($limit, $chunkSize)), $resolved); + $firstPageLimit = ResourceContext::minLimit($limit, $chunkSize); + [$page, $scanned] = $this->fetchItemsPage($options->withPagination($startOffset, $firstPageLimit), $resolved); foreach ($page->getItems() as $item) { yield $item; } - $total = $page->getTotal(); - $cap = ($limit !== null && $limit > 0) ? $limit : $total; - $pageScanned = $scanned ?? count($page->getItems()); + $pageScanned = self::scannedRows($page, $scanned, $firstPageLimit); $currentOffset = $startOffset + $pageScanned; - $remaining = min($total - $startOffset, $cap) - $pageScanned; + // Items still to yield, or null for "unbounded" (no $limit given). + $remaining = ($limit !== null && $limit > 0) ? $limit - $pageScanned : null; - // Guard on the previous page having scanned something, so an over-reported total terminates - // instead of looping forever. - while ($pageScanned > 0 && $remaining > 0) { - [$page, $scanned] = $this->fetchItemsPage($options->withPagination($currentOffset, ResourceContext::minLimit($remaining, $chunkSize)), $resolved); + // Two distinct ways to stop: a page that scans nothing (the dataset is exhausted), or + // $remaining having already reached 0 (the caller's $limit is satisfied) — either ends the loop. + while ($pageScanned > 0 && ($remaining === null || $remaining > 0)) { + $pageLimit = ResourceContext::minLimit($remaining, $chunkSize); + [$page, $scanned] = $this->fetchItemsPage($options->withPagination($currentOffset, $pageLimit), $resolved); foreach ($page->getItems() as $item) { yield $item; } - $pageScanned = $scanned ?? count($page->getItems()); + $pageScanned = self::scannedRows($page, $scanned, $pageLimit); $currentOffset += $pageScanned; - $remaining -= $pageScanned; + if ($remaining !== null) { + $remaining -= $pageScanned; + } } } + /** + * The number of rows a page advances the offset by: the larger of its scanned-count header (if + * present) and its returned item count, capped at the limit requested for it ({@code null} = + * uncapped). + * + * Unlike {@see \Apify\Client\Internal\ResourceContext}'s plain {@code advancedBy()} (which has no + * such cap — see its docblock), this one is load-bearing whenever {@code unwind} is set: + * {@code unwind} turns each scanned row into one item per unwound array element, so the returned + * item count can legitimately exceed the number of rows the API scanned to produce it (the reverse + * of {@code skipEmpty}/{@code clean}, which only ever drop rows). The server never scans more rows + * for a page than that page's requested limit, so the true row advance can never exceed it either; + * capping the {@code max()} above at the requested limit keeps an unwound page's inflated item count + * (or, if the scanned-count header is also missing, its unwind-inflated fallback) from overstating + * the real advance, which would otherwise skip whatever rows lie between the true (smaller) advance + * and the inflated one. + * + * @param PaginationList $page + */ + private static function scannedRows(PaginationList $page, ?int $scannedHeader, ?int $requestedLimit): int + { + $scanned = max($scannedHeader ?? 0, count($page->getItems())); + return $requestedLimit !== null ? min($scanned, $requestedLimit) : $scanned; + } + /** * Fetches one page of dataset items and parses the pagination headers. Returns the page together * with the {@code X-Apify-Pagination-Count} "scanned rows" count ({@code null} if the response diff --git a/src/Resource/KeyValueStoreClient.php b/src/Resource/KeyValueStoreClient.php index d51f03c..d23cbeb 100644 --- a/src/Resource/KeyValueStoreClient.php +++ b/src/Resource/KeyValueStoreClient.php @@ -101,13 +101,13 @@ public function listKeys(?ListKeysOptions $options = null, int|float|string|null * * The options' {@code limit} caps the total number of keys yielded across all pages ({@code null} * or {@code 0} = all); {@code exclusiveStartKey} starts the listing after a given key; {@code prefix} and - * {@code collection} restrict which keys are listed. Unlike the offset/limit collection iterators, - * there is no separate page-size argument: the per-page size follows the remaining total cap (or - * the server default when unbounded), exactly as the reference client does. + * {@code collection} restrict which keys are listed. {@code $chunkSize} caps how many keys are + * requested per page ({@code null} = the remaining total cap, or the server default when + * unbounded) — the two are independent, matching the offset/limit collection iterators. * * @return Generator */ - public function iterateKeys(?ListKeysOptions $options = null, int|float|string|null $timeoutSecs = null): Generator + public function iterateKeys(?ListKeysOptions $options = null, ?int $chunkSize = null, int|float|string|null $timeoutSecs = null): Generator { $options ??= new ListKeysOptions(); // Total cap across all pages. null or 0 means "iterate the whole store" (the API treats @@ -119,10 +119,11 @@ public function iterateKeys(?ListKeysOptions $options = null, int|float|string|n $iterated = 0; while (true) { - // Ask for only as many keys as remain under the total cap (null = server default). + // Ask for only as many keys as remain under the total cap (null = server default), + // further capped by $chunkSize. $remaining = $limit !== null ? $limit - $iterated : null; $page = $this->listKeys(new ListKeysOptions( - limit: $remaining, + limit: ResourceContext::minLimit($remaining, $chunkSize), exclusiveStartKey: $exclusiveStartKey, prefix: $options->prefix, collection: $options->collection, diff --git a/src/Resource/TaskClient.php b/src/Resource/TaskClient.php index 8a026f4..a9b3b17 100644 --- a/src/Resource/TaskClient.php +++ b/src/Resource/TaskClient.php @@ -10,6 +10,7 @@ use Apify\Client\Internal\QueryParams; use Apify\Client\Internal\ResourceContext; use Apify\Client\Internal\TimeoutTiers; +use Apify\Client\Internal\WaitForResources; use Apify\Client\Model\ActorRun; use Apify\Client\Model\Task; use Apify\Client\Options\LastRunOptions; @@ -97,23 +98,28 @@ public function start( ?TaskStartOptions $options = null, int|float|string|null $timeoutSecs = null, ): ActorRun { + $options ??= new TaskStartOptions(); $params = new QueryParams(); - ($options ?? new TaskStartOptions())->appendTo($params); + $options->appendTo($params); $body = $input === null ? null : Json::encode($input); - return new ActorRun($this->ctx->postWithBody( - 'runs', - $params, - $body, - ResourceContext::CONTENT_TYPE_JSON, - $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_MEDIUM) - )); + return WaitForResources::start( + fn () => new ActorRun($this->ctx->postWithBody( + 'runs', + $params, + $body, + ResourceContext::CONTENT_TYPE_JSON, + $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_MEDIUM) + )), + $options->waitForResources, + ); } /** * Starts the task and waits (client-side polling) for it to finish. * * @param mixed $input optionally overrides the task's stored input - * @param int|null $waitSecs bounds the wait; {@code null} waits indefinitely + * @param int|null $waitSecs bounds the wait; {@code null} waits indefinitely. Time spent retrying a + * {@code $options->waitForResources} start does not count toward it. */ public function call(mixed $input = null, ?TaskStartOptions $options = null, ?int $waitSecs = null): ActorRun { diff --git a/src/Version.php b/src/Version.php index 61db3b3..f86d59d 100644 --- a/src/Version.php +++ b/src/Version.php @@ -17,13 +17,13 @@ final class Version * The semantic version of this client library (see https://semver.org/). * Changes to the public interface other than additive ones are considered breaking changes. */ - public const CLIENT_VERSION = '0.8.0'; + public const CLIENT_VERSION = '0.9.0'; /** * The version of the Apify OpenAPI specification this client was generated and verified * against. Corresponds to the {@code info.version} field of the Apify OpenAPI document. */ - public const API_SPEC_VERSION = 'v2-2026-10-01T153946Z'; + public const API_SPEC_VERSION = 'v2-2026-10-08T073338Z'; private function __construct() { diff --git a/tests/Integration/ActorIntegrationTest.php b/tests/Integration/ActorIntegrationTest.php index 0c5854e..d5436e8 100644 --- a/tests/Integration/ActorIntegrationTest.php +++ b/tests/Integration/ActorIntegrationTest.php @@ -42,7 +42,7 @@ public function testActorCrudFlow(): void $updated = $actor->update(['title' => 'Updated Title']); self::assertSame('Updated Title', $updated->getTitle()); $actor->builds()->list(new ListOptions()); - $actor->versions()->list(new ListOptions()); + $actor->versions()->list(); } finally { $client->actor((string) $created->getId())->delete(); } @@ -62,7 +62,7 @@ public function testActorVersionCrudFlow(): void ]); self::assertSame('0.1', $version->getVersionNumber()); self::assertNotNull($actor->version('0.1')->get()); - $actor->versions()->list(new ListOptions()); + $actor->versions()->list(); $actor->version('0.1')->update([ 'buildTag' => 'beta', 'sourceType' => 'SOURCE_FILES', @@ -110,9 +110,11 @@ public function testIterateActorVersions(): void 'buildTag' => 'latest', 'sourceFiles' => [], ]); + // This endpoint ignores pagination and always returns every version in one response, so + // iterate() makes a single request; the retry wrapper only rides out eventual consistency. self::assertEventuallyIterated(['0.0', '0.1'], static function () use ($actor): array { $seen = []; - foreach ($actor->versions()->iterate(null, 1) as $version) { + foreach ($actor->versions()->iterate() as $version) { $seen[(string) $version->getVersionNumber()] = true; } return $seen; @@ -130,9 +132,11 @@ public function testIterateActorEnvVars(): void $version = $client->actor((string) $created->getId())->version('0.0'); $version->envVars()->create(new ActorEnvVar('ITER_VAR_1', 'v1')); $version->envVars()->create(new ActorEnvVar('ITER_VAR_2', 'v2')); + // This endpoint ignores pagination and always returns every env var in one response, so + // iterate() makes a single request; the retry wrapper only rides out eventual consistency. self::assertEventuallyIterated(['ITER_VAR_1', 'ITER_VAR_2'], static function () use ($version): array { $seen = []; - foreach ($version->envVars()->iterate(1) as $envVar) { + foreach ($version->envVars()->iterate() as $envVar) { $seen[(string) $envVar->getName()] = true; } return $seen; diff --git a/tests/Unit/IterationTest.php b/tests/Unit/IterationTest.php index 4571bc5..0a7e949 100644 --- a/tests/Unit/IterationTest.php +++ b/tests/Unit/IterationTest.php @@ -15,8 +15,10 @@ /** * Hermetic tests for the lazy-iteration helpers, driven entirely by a scripted transport. They - * exercise the paging/termination logic — single short page, over-reported total, total-cap vs. - * page-size split, and cursor threading — without touching the network. + * exercise the paging/termination logic — over-reported total, total-cap vs. page-size split, cursor + * threading, and (matching the reference client) that the stop condition never consults a listing's + * reported {@code total}: with no explicit {@code limit}, the loop always ends only once a page scans + * no rows, so a resource that keeps growing during the iteration is not cut off at its starting size. */ final class IterationTest extends TestCase { @@ -51,21 +53,27 @@ private static function actors(int ...$ids): array return array_map(static fn (int $id): array => ['id' => "a$id", 'name' => "actor$id"], $ids); } - public function testSinglePageStopsAfterOneRequest(): void + public function testSinglePageThenEmptyPageEndsIteration(): void { - // total equals the number of returned items => no second request. - $transport = (new MockTransport())->queueResponse(200, self::page(self::actors(1, 2, 3), 3, 0)); + // With no explicit limit, the iterator never trusts a page's reported total: it always + // confirms the end with one more request, which here comes back empty. + $transport = (new MockTransport()) + ->queueResponse(200, self::page(self::actors(1, 2, 3), 3, 0)) + ->queueResponse(200, self::page([], 3, 3)); $ids = []; foreach ($this->client($transport)->actors()->iterate() as $actor) { $ids[] = $actor->getId(); } self::assertSame(['a1', 'a2', 'a3'], $ids); - self::assertSame(1, $transport->callCount()); + self::assertSame(2, $transport->callCount()); + self::assertStringContainsString('offset=3', (string) $transport->received[1]->getUri()); } public function testOverReportedTotalTerminates(): void { - // The API claims 10 items but only 3 exist; the iterator must stop, not loop forever. + // The API claims 10 items but only 3 exist; the iterator must stop, not loop forever. The + // reported total is irrelevant either way: termination follows only from the empty follow-up + // page. $transport = (new MockTransport()) ->queueResponse(200, self::page(self::actors(1, 2, 3), 10, 0)) ->queueResponse(200, self::page([], 10, 3)); // the follow-up page comes back empty @@ -74,10 +82,26 @@ public function testOverReportedTotalTerminates(): void $ids[] = $actor->getId(); } self::assertSame(['a1', 'a2', 'a3'], $ids); - // One extra fetch is made (remaining > 0) before the empty page stops iteration. self::assertSame(2, $transport->callCount()); } + public function testGrowingCollectionIsNotCutOffAtItsStartingSize(): void + { + // The collection holds 2 items when the first page is read and a 3rd is added before the + // follow-up page is fetched. With no explicit limit the iterator must pick it up instead of + // stopping at the total the first page reported. + $transport = (new MockTransport()) + ->queueResponse(200, self::page(self::actors(1, 2), 2, 0)) + ->queueResponse(200, self::page(self::actors(3), 3, 2)) + ->queueResponse(200, self::page([], 3, 3)); + $ids = []; + foreach ($this->client($transport)->actors()->iterate() as $actor) { + $ids[] = $actor->getId(); + } + self::assertSame(['a1', 'a2', 'a3'], $ids); + self::assertSame(3, $transport->callCount()); + } + public function testLimitIsTotalCapAndChunkSizeIsPageSize(): void { // limit=5 total across all pages; chunkSize=2 per page => pages of 2, 2, 1. @@ -118,7 +142,8 @@ public function testLimitCapStopsBeforeExhaustingPages(): void public function testDatasetIterateItemsPagesViaHeaders(): void { - // The dataset-items endpoint returns a bare array and reports pagination via headers. + // The dataset-items endpoint returns a bare array and reports pagination via headers. With no + // explicit limit, a trailing empty page confirms the end instead of trusting the total header. $transport = (new MockTransport()) ->queueResponse(200, Json::encode([['n' => 1], ['n' => 2]]), [ 'X-Apify-Pagination-Total' => '3', @@ -129,14 +154,20 @@ public function testDatasetIterateItemsPagesViaHeaders(): void 'X-Apify-Pagination-Total' => '3', 'X-Apify-Pagination-Offset' => '2', 'X-Apify-Pagination-Limit' => '2', + ]) + ->queueResponse(200, Json::encode([]), [ + 'X-Apify-Pagination-Total' => '3', + 'X-Apify-Pagination-Offset' => '3', + 'X-Apify-Pagination-Limit' => '2', ]); $values = []; foreach ($this->client($transport)->dataset('ds')->iterateItems(null, 2) as $item) { $values[] = $item['n']; } self::assertSame([1, 2, 3], $values); - self::assertSame(2, $transport->callCount()); + self::assertSame(3, $transport->callCount()); self::assertStringContainsString('offset=2', (string) $transport->received[1]->getUri()); + self::assertStringContainsString('offset=3', (string) $transport->received[2]->getUri()); } public function testDatasetIterateItemsAdvancesByScannedCountNotItemCount(): void @@ -159,14 +190,60 @@ public function testDatasetIterateItemsAdvancesByScannedCountNotItemCount(): voi 'X-Apify-Pagination-Offset' => '3', 'X-Apify-Pagination-Limit' => '3', 'X-Apify-Pagination-Count' => '3', + ]) + // Trailing page confirms there is nothing left to scan. + ->queueResponse(200, Json::encode([]), [ + 'X-Apify-Pagination-Total' => '6', + 'X-Apify-Pagination-Offset' => '6', + 'X-Apify-Pagination-Limit' => '3', + 'X-Apify-Pagination-Count' => '0', ]); $values = []; foreach ($this->client($transport)->dataset('ds')->iterateItems(new DatasetListItemsOptions(skipEmpty: true), 3) as $item) { $values[] = $item['n']; } self::assertSame([0, 1, 3, 4, 5], $values); - self::assertSame(2, $transport->callCount()); + self::assertSame(3, $transport->callCount()); self::assertStringContainsString('offset=3', (string) $transport->received[1]->getUri()); + self::assertStringContainsString('offset=6', (string) $transport->received[2]->getUri()); + } + + public function testDatasetIterateItemsCapsScannedCountWhenUnwindInflatesItemsPastTheRequestedLimit(): void + { + // With unwind active, a page can return MORE items than rows it scanned (the opposite of + // skipEmpty). Here the scanned-count header is also missing (e.g. stripped by a proxy), so the + // iterator falls back to the (unwind-inflated) item count — which must be capped at the + // requested per-page limit (2), since the server never scans more rows per page than that. + // Without the cap, the iterator would advance by the raw item count (6) instead of the real row + // count (2), overshooting to offset=6 and skipping rows 2-3's items entirely. + $transport = (new MockTransport()) + // Page 1: rows 0-1 (2 rows, matching the requested limit), each unwound into 3 items. + ->queueResponse(200, Json::encode([['n' => 0], ['n' => 1], ['n' => 2], ['n' => 3], ['n' => 4], ['n' => 5]]), [ + 'X-Apify-Pagination-Total' => '4', + 'X-Apify-Pagination-Offset' => '0', + 'X-Apify-Pagination-Limit' => '2', + // No X-Apify-Pagination-Count header. + ]) + // Page 2 (must be requested at offset=2, not offset=6): rows 2-3, likewise unwound. + ->queueResponse(200, Json::encode([['n' => 6], ['n' => 7], ['n' => 8], ['n' => 9], ['n' => 10], ['n' => 11]]), [ + 'X-Apify-Pagination-Total' => '4', + 'X-Apify-Pagination-Offset' => '2', + 'X-Apify-Pagination-Limit' => '2', + ]) + // Trailing page confirms the dataset (4 real rows) is exhausted. + ->queueResponse(200, Json::encode([]), [ + 'X-Apify-Pagination-Total' => '4', + 'X-Apify-Pagination-Offset' => '4', + 'X-Apify-Pagination-Limit' => '2', + ]); + $values = []; + foreach ($this->client($transport)->dataset('ds')->iterateItems(new DatasetListItemsOptions(unwind: ['parts']), 2) as $item) { + $values[] = $item['n']; + } + self::assertSame(range(0, 11), $values); + self::assertSame(3, $transport->callCount()); + self::assertStringContainsString('offset=2', (string) $transport->received[1]->getUri()); + self::assertStringContainsString('offset=4', (string) $transport->received[2]->getUri()); } public function testDatasetIterateItemsPreservesFilters(): void @@ -176,6 +253,11 @@ public function testDatasetIterateItemsPreservesFilters(): void 'X-Apify-Pagination-Total' => '1', 'X-Apify-Pagination-Offset' => '0', 'X-Apify-Pagination-Limit' => '1', + ]) + ->queueResponse(200, Json::encode([]), [ + 'X-Apify-Pagination-Total' => '1', + 'X-Apify-Pagination-Offset' => '1', + 'X-Apify-Pagination-Limit' => '1', ]); $it = $this->client($transport)->dataset('ds')->iterateItems(new DatasetListItemsOptions(fields: ['n'], clean: true)); iterator_to_array($it); @@ -280,4 +362,71 @@ public function testIterateKeysLimitZeroIteratesAll(): void self::assertStringNotContainsString('limit=0', (string) $transport->received[0]->getUri()); self::assertStringNotContainsString('limit=', (string) $transport->received[0]->getUri()); } + + public function testIterateKeysChunkSizeCapsPageSizeIndependentlyOfLimit(): void + { + // chunkSize bounds each request independently of the total cap: limit=5, chunkSize=2 pages + // through 2, 2, then the remaining 1. + $transport = (new MockTransport()) + ->queueResponse(200, self::keysPage(true, 'k2', 'k1', 'k2')) + ->queueResponse(200, self::keysPage(true, 'k4', 'k3', 'k4')) + ->queueResponse(200, self::keysPage(true, 'k5', 'k5')); + $keys = []; + foreach ($this->client($transport)->keyValueStore('kvs')->iterateKeys(new ListKeysOptions(limit: 5), 2) as $key) { + $keys[] = $key->getKey(); + } + self::assertSame(['k1', 'k2', 'k3', 'k4', 'k5'], $keys); + self::assertSame(3, $transport->callCount()); + self::assertStringContainsString('limit=2', (string) $transport->received[0]->getUri()); + self::assertStringContainsString('limit=2', (string) $transport->received[1]->getUri()); + // The last page is capped by the remaining total (1), not chunkSize (2). + self::assertStringContainsString('limit=1', (string) $transport->received[2]->getUri()); + } + + public function testIterateKeysChunkSizeWithoutLimitBoundsEveryPage(): void + { + // With no total limit, chunkSize alone still caps each page's requested size. + $transport = (new MockTransport()) + ->queueResponse(200, self::keysPage(true, 'k2', 'k1', 'k2')) + ->queueResponse(200, self::keysPage(false, null, 'k3')); + $keys = []; + foreach ($this->client($transport)->keyValueStore('kvs')->iterateKeys(null, 2) as $key) { + $keys[] = $key->getKey(); + } + self::assertSame(['k1', 'k2', 'k3'], $keys); + self::assertStringContainsString('limit=2', (string) $transport->received[0]->getUri()); + self::assertStringContainsString('limit=2', (string) $transport->received[1]->getUri()); + } + + public function testActorEnvVarsIterateMakesExactlyOneRequest(): void + { + // The env-vars endpoint ignores offset/limit and always returns every item in one response. + // The mock only has one response queued, so a regression that re-pages would fail outright + // instead of looping or duplicating items. + $transport = (new MockTransport())->queueResponse(200, self::page([ + ['name' => 'A', 'value' => '1'], + ['name' => 'B', 'value' => '2'], + ], 2, 0)); + $names = []; + foreach ($this->client($transport)->actor('a1')->version('0.0')->envVars()->iterate() as $envVar) { + $names[] = $envVar->getName(); + } + self::assertSame(['A', 'B'], $names); + self::assertSame(1, $transport->callCount()); + } + + public function testActorVersionsIterateMakesExactlyOneRequest(): void + { + // The versions endpoint likewise ignores pagination and returns everything in one response. + $transport = (new MockTransport())->queueResponse(200, self::page([ + ['versionNumber' => '0.0'], + ['versionNumber' => '0.1'], + ], 2, 0)); + $numbers = []; + foreach ($this->client($transport)->actor('a1')->versions()->iterate() as $version) { + $numbers[] = $version->getVersionNumber(); + } + self::assertSame(['0.0', '0.1'], $numbers); + self::assertSame(1, $transport->callCount()); + } } diff --git a/tests/Unit/WaitForFinishTest.php b/tests/Unit/WaitForFinishTest.php new file mode 100644 index 0000000..3906631 --- /dev/null +++ b/tests/Unit/WaitForFinishTest.php @@ -0,0 +1,87 @@ + ['type' => 'record-not-found', 'message' => 'Run was not found']]); + } + + private static function runBody(string $status): string + { + return Json::encode(['data' => ['id' => 'run1', 'status' => $status]]); + } + + public function testTogglesThroughTransientNotFoundToTerminalStatus(): void + { + // Simulates replica lag right after the run is created: a couple of 404s, then the run + // becomes visible and eventually finishes. Each 404 must not abort the wait outright, and a + // successful poll in between must reset the not-found grace window. + $transport = (new MockTransport()) + ->queueResponse(404, self::notFoundBody()) + ->queueResponse(404, self::notFoundBody()) + ->queueResponse(200, self::runBody('RUNNING')) + ->queueResponse(404, self::notFoundBody()) + ->queueResponse(200, self::runBody('SUCCEEDED')); + + $run = $this->client($transport)->run('run1')->waitForFinish(30); + + self::assertSame('SUCCEEDED', $run->getStatus()); + self::assertSame(5, $transport->callCount()); + } + + public function testGivesUpAfterNotFoundGraceWindowElapses(): void + { + // A run/build that keeps 404ing (e.g. it was never actually created, or was deleted) must not + // be polled for the whole wait budget: the wait gives up after a few seconds and rethrows the + // 404 as a NotFoundException instead. + $transport = new MockTransport(); + for ($i = 0; $i < 30; $i++) { + $transport->queueResponse(404, self::notFoundBody()); + } + + $start = microtime(true); + try { + $this->client($transport)->build('build1')->waitForFinish(60); + self::fail('expected a NotFoundException'); + } catch (NotFoundException $e) { + self::assertSame('record-not-found', $e->getType()); + } + $elapsedSecs = microtime(true) - $start; + + // The grace window is a few seconds; giving up must happen well short of the 60s budget. + self::assertLessThan(10.0, $elapsedSecs); + self::assertGreaterThan(1, $transport->callCount()); + } + + public function testNonNotFoundErrorPropagatesImmediately(): void + { + // 403 is not server-retried (unlike 5xx/429), so this asserts the polling loop itself gives up + // on the first non-404 error rather than treating it like a transient not-found. + $transport = (new MockTransport())->queueResponse(403, Json::encode(['error' => ['type' => 'permission-denied', 'message' => 'no access']])); + + $this->expectException(\Apify\Client\Exception\ForbiddenException::class); + $this->client($transport)->run('run1')->waitForFinish(30); + } +} diff --git a/tests/Unit/WaitForResourcesTest.php b/tests/Unit/WaitForResourcesTest.php new file mode 100644 index 0000000..65bb66c --- /dev/null +++ b/tests/Unit/WaitForResourcesTest.php @@ -0,0 +1,164 @@ + ['type' => $type, 'message' => "rejected: $type"]]); + } + + private static function runBody(string $status = 'READY'): string + { + return Json::encode(['data' => ['id' => 'run1', 'status' => $status]]); + } + + private static function rejectionError(string $type): ApifyApiException + { + return new ApifyApiException(402, $type, "rejected: $type", 1, 'POST', '/v2/actor-runs'); + } + + public function testRetriesEveryCooldownUntilItSucceeds(): void + { + $attempts = 0; + $start = static function () use (&$attempts) { + $attempts++; + if ($attempts < 3) { + throw self::rejectionError($attempts === 1 ? 'actor-memory-limit-exceeded' : 'concurrent-runs-limit-exceeded'); + } + return 'started'; + }; + + $result = WaitForResources::start($start, true, cooldownSecs: 0.001); + + self::assertSame('started', $result); + self::assertSame(3, $attempts); + } + + public function testNumberOfSecondsBoundsTheRetryingAndThrowsTheLastRejection(): void + { + $attempts = 0; + $start = static function () use (&$attempts) { + $attempts++; + throw self::rejectionError('actor-memory-limit-exceeded'); + }; + + // $start rejects unconditionally, so WaitForResources::start() is guaranteed to exhaust the + // 1-second budget and rethrow the last rejection; expectException() (rather than a try/catch + // with a fail()-on-fallthrough branch) asserts that, since PHPUnit itself fails the test if the + // call below returns instead of throwing, keeping the "it must throw" assertion live without a + // branch that PHPStan can (correctly, given this always-throwing $start) prove unreachable. + $this->expectException(ApifyApiException::class); + $this->expectExceptionMessage('rejected: actor-memory-limit-exceeded'); + try { + WaitForResources::start($start, 1, cooldownSecs: 0.001); + } finally { + // Roughly 1 / 0.001 attempts before the 1-second budget runs out; a wide range keeps this + // robust to scheduling jitter while still proving it bounds the retrying rather than + // retrying forever. $waitForResources only accepts a whole number of seconds, so 1 is the + // smallest nonzero budget this can exercise without hitting the (already separately tested) + // "0 makes a single attempt" case. The `finally` runs before the exception propagates out + // of this test method to PHPUnit's expectException() machinery above. + self::assertGreaterThan(10, $attempts); + self::assertLessThan(10000, $attempts); + } + } + + public function testActorStartAndTaskStartWireTheOptionIntoWaitForResources(): void + { + // Succeeding on the very first attempt needs no cooldown at all, so this confirms + // ActorClient/TaskClient actually construct the WaitForResources::start() call (rather than, + // say, silently dropping the option) without the suite paying for even one real sleep; the + // retry/cooldown mechanism itself is exercised above, directly against WaitForResources. + $actorTransport = (new MockTransport())->queueResponse(201, self::runBody()); + $run = $this->client($actorTransport)->actor('a1')->start(null, new ActorStartOptions(waitForResources: true)); + self::assertSame('run1', $run->getId()); + + $taskTransport = (new MockTransport())->queueResponse(201, self::runBody()); + $run = $this->client($taskTransport)->task('t1')->start(null, new TaskStartOptions(waitForResources: true)); + self::assertSame('run1', $run->getId()); + } + + public function testThrowsFirstRejectionWithoutTheOption(): void + { + $transport = (new MockTransport())->queueResponse(402, self::rejection('actor-memory-limit-exceeded')); + + try { + $this->client($transport)->actor('a1')->start(); + self::fail('expected an ApifyApiException'); + } catch (ApifyApiException $e) { + self::assertSame('actor-memory-limit-exceeded', $e->getType()); + } + self::assertSame(1, $transport->callCount()); + } + + public function testFalseBehavesLikeNoOption(): void + { + $transport = (new MockTransport())->queueResponse(402, self::rejection('concurrent-runs-limit-exceeded')); + + $this->expectException(ApifyApiException::class); + $this->client($transport)->actor('a1')->start(null, new ActorStartOptions(waitForResources: false)); + } + + public function testThrowsAnyOtherErrorRightAway(): void + { + $transport = (new MockTransport())->queueResponse(400, self::rejection('invalid-input')); + + try { + $this->client($transport)->actor('a1')->start(null, new ActorStartOptions(waitForResources: true)); + self::fail('expected an ApifyApiException'); + } catch (ApifyApiException $e) { + self::assertSame('invalid-input', $e->getType()); + } + self::assertSame(1, $transport->callCount()); + } + + public function testZeroSecondsMakesASingleAttempt(): void + { + $transport = (new MockTransport())->queueResponse(402, self::rejection('actor-memory-limit-exceeded')); + + $this->expectException(ApifyApiException::class); + $this->client($transport)->actor('a1')->start(null, new ActorStartOptions(waitForResources: 0)); + } + + public function testRejectsANegativeNumberOfSeconds(): void + { + $transport = new MockTransport(); + + try { + $this->client($transport)->actor('a1')->start(null, new ActorStartOptions(waitForResources: -1)); + self::fail('expected an InvalidArgumentException'); + } catch (InvalidArgumentException $e) { + // No request should have been attempted: the option is invalid before any start is tried. + self::assertSame(0, $transport->callCount()); + } + } +}