From 471a2a2c3f79f3530dc35801dc00c9c61b6ca9e5 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Oct 2026 16:59:57 +0000 Subject: [PATCH 1/3] feat: sync PHP client with Apify OpenAPI spec v2-2026-10-19T133704Z Realigns with the reference apify-client-js updates since the client's last sync (apify/apify-client-js#1139, #1140, #1141, #1145, #1152) and the OpenAPI schema fixes in apify/apify-docs#3087 (Actor/version/schedule/env-var request bodies, which this client already passes through untyped, so no code changes were needed for those specifically). - Add DatasetClient::streamItems(), a streamed (non-buffered) counterpart to downloadItems() for large exports. - Add a $contentEncoding parameter to KeyValueStoreClient::setRecord(), so an already-compressed value is uploaded as is instead of being compressed again. - Add ApifyClientException, a common base for every error the client throws for a failed API interaction; ApifyApiException (and its status subclasses) now extend it. - Add InvalidResponseBodyException (extends ApifyClientException), thrown instead of a raw JsonException when a successful response body fails to parse as JSON. - Bump Version::API_SPEC_VERSION and Version::CLIENT_VERSION. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01G3FfQf9r45KVCa5VAYAjkd --- CHANGELOG.md | 14 ++++ README.md | 16 ++++ docs/misc.md | 4 +- docs/options.md | 5 ++ docs/storages.md | 15 +++- src/Exception/ApifyApiException.php | 7 +- src/Exception/ApifyClientException.php | 22 ++++++ .../InvalidResponseBodyException.php | 34 +++++++++ src/Internal/Json.php | 15 +++- src/Internal/ResourceContext.php | 42 ++++++++++- src/Resource/DatasetClient.php | 27 +++++++ src/Resource/KeyValueStoreClient.php | 22 +++++- src/Resource/LogClient.php | 25 ++----- src/Version.php | 4 +- tests/Examples/DocSnippetsTest.php | 1 + tests/Integration/ActorIntegrationTest.php | 21 ++++++ tests/Integration/DatasetIntegrationTest.php | 22 ++++++ .../KeyValueStoreIntegrationTest.php | 36 +++++++++ tests/Integration/ScheduleIntegrationTest.php | 20 +++++ tests/Unit/ApifyClientExceptionTest.php | 72 ++++++++++++++++++ tests/Unit/DatasetStreamItemsTest.php | 57 ++++++++++++++ .../Unit/KeyValueStoreContentEncodingTest.php | 75 +++++++++++++++++++ tests/Unit/LogClientTest.php | 14 ++++ 23 files changed, 537 insertions(+), 33 deletions(-) create mode 100644 src/Exception/ApifyClientException.php create mode 100644 src/Exception/InvalidResponseBodyException.php create mode 100644 tests/Unit/ApifyClientExceptionTest.php create mode 100644 tests/Unit/DatasetStreamItemsTest.php create mode 100644 tests/Unit/KeyValueStoreContentEncodingTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index b887f45..b8dd130 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,19 @@ # Changelog +## 0.10.0 + +- Synced with Apify OpenAPI spec `v2-2026-10-19T133704Z` (realigned Actor, version, env var and + schedule request schemas; no client code changes required, since those bodies were already passed + through untyped). +- Added `DatasetClient::streamItems()`, returning a stream over a dataset items export instead of + buffering it fully in memory, matching the reference client's streamed export. +- Added a `$contentEncoding` parameter to `KeyValueStoreClient::setRecord()`, so an already-compressed + value (gzip, br, ...) is uploaded as is instead of being compressed a second time. +- Added `ApifyClientException`, a common base class for every error the client throws for a failed API + interaction. `ApifyApiException` (and its status subclasses) now extend it. +- Added `InvalidResponseBodyException` (extends `ApifyClientException`), thrown instead of a raw + `JsonException` when a successful response body fails to parse as JSON. + ## 0.9.0 - Synced with Apify OpenAPI spec `v2-2026-10-08T073338Z`. diff --git a/README.md b/README.md index 685f40c..52fd73f 100644 --- a/README.md +++ b/README.md @@ -206,6 +206,22 @@ exhausted does the underlying error surface, as an `Apify\Client\Exception\Trans `TransportException` means the request never produced a usable response (network failure or timeout) after all retries. Requests are retried on network errors, HTTP 429 and 5xx. +Every error the client throws for a failed API interaction — `ApifyApiException` and its status +subclasses above, plus `Apify\Client\Exception\InvalidResponseBodyException` (thrown when a +successful response's body does not parse as JSON, e.g. a connection dropped mid-response) — extends +the common `Apify\Client\Exception\ApifyClientException`, so one `catch` can handle both: + +```php +try { + $client->actor('my-actor')->update(['title' => 'x']); +} catch (ApifyClientException $e) { + // Either the API returned an error response, or its body could not be parsed. +} +``` + +`ArgumentValidationError`-style invalid-argument errors (built-in `\InvalidArgumentException`s) and +`TransportException` stay outside this hierarchy, matching the reference client. + ## Versioning - `Apify\Client\Version::CLIENT_VERSION` — the semantic version of this library. diff --git a/docs/misc.md b/docs/misc.md index 1e86136..1636766 100644 --- a/docs/misc.md +++ b/docs/misc.md @@ -35,7 +35,9 @@ $publicProfile = $client->user('some-username')->get(); ## Logs — `$client->log($buildOrRunId)` -- `get(?LogOptions $options = null): ?string` — the full log as text. +- `get(?LogOptions $options = null): ?string` — the full log as text. A PHP string is already a byte + sequence and is never re-decoded, so this already returns the log's exact bytes — there is no + separate "raw bytes" mode to opt into. - `stream(?LogOptions $options = null): StreamInterface` — a live stream of the log. ```php diff --git a/docs/options.md b/docs/options.md index a578992..d34bfc7 100644 --- a/docs/options.md +++ b/docs/options.md @@ -237,6 +237,11 @@ For `keyValueStore()->setRecord()`. | `timeoutSecs` | `?int` | Per-request timeout for the upload (capped at the client's overall request timeout). | | `doNotRetryTimeouts` | `bool` | If `true`, do not retry the upload on request timeout (default `false`). | +`setRecord()` also takes a trailing `?string $contentEncoding = null` parameter (not a field of +`SetRecordOptions`, which only holds the write-retry options above): set it to the encoding already +applied to `$value` (e.g. `'gzip'`) so the client's own request compression backs off and the bytes +are uploaded exactly as given. + ## Request queues ### `RequestQueueClientOptions` diff --git a/docs/storages.md b/docs/storages.md index ebe3c97..df462c3 100644 --- a/docs/storages.md +++ b/docs/storages.md @@ -21,6 +21,7 @@ Single — `$client->dataset($id)`: - `listItems(?DatasetListItemsOptions $options = null): PaginationList` — one page of items decoded to PHP values. - `iterateItems(?DatasetListItemsOptions $options = null, ?int $chunkSize = null): iterable` — lazily iterate all items, paging on demand. The options' `limit` caps the total number of items yielded across all pages (unset = all); `$chunkSize` is the per-page size. The iterator advances and terminates by the number of rows the API scanned (the `X-Apify-Pagination-Count` response header), not by the number of items a page returns, matching the reference JS client exactly — see the method's doc comment for why that matters with `skipEmpty`/`clean`/`unwind`. - `downloadItems(DownloadItemsFormat $format, ?DatasetDownloadOptions $options = null): string` — raw export bytes. +- `streamItems(DownloadItemsFormat $format, ?DatasetDownloadOptions $options = null): StreamInterface` — the same export as `downloadItems()`, as a PSR-7 stream instead of a fully-buffered string, so a large export can be read incrementally (e.g. piped to a file) without holding it all in memory. Bypasses the retrying transport, so a failed request is not retried. - `pushItems(mixed $items): void` - `getStatistics(): ?array` — throws if the dataset itself does not exist (a 404 here cannot mean anything else). @@ -36,6 +37,14 @@ $csv = $client->dataset($dataset->getId())->downloadItems(DownloadItemsFormat::C foreach ($client->dataset($dataset->getId())->iterateItems(new DatasetListItemsOptions(), 1000) as $item) { echo ($item['url'] ?? '') . PHP_EOL; } + +// Stream a large export straight to a file instead of buffering it in memory. +$stream = $client->dataset($dataset->getId())->streamItems(DownloadItemsFormat::CSV); +$out = fopen('output.csv', 'wb'); +while (!$stream->eof()) { + fwrite($out, $stream->read(8192)); +} +fclose($out); ``` ## Key-value stores @@ -53,7 +62,7 @@ Single — `$client->keyValueStore($id)`: - `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` +- `setRecord(string $key, string $value, string $contentType, ?SetRecordOptions $options = null, ?string $contentEncoding = null): void` — `$contentEncoding`, if given, is sent as the `Content-Encoding` header, for a `$value` that is already compressed (gzip, br, ...); the client's own best-effort request compression always backs off when `Content-Encoding` is already set, so the bytes reach the API exactly as given. - `setRecordJson(string $key, mixed $value): void` — convenience over `setRecord()` that JSON-encodes `$value` and stores it with a JSON content type. - `deleteRecord(string $key): void` — permanently removes the record with the given key. - `getRecordPublicUrl(string $key): string` — builds a shareable URL for downloading a single record; for a private store it appends an access signature so the URL works without an API token. @@ -71,6 +80,10 @@ echo ($decoded['answer'] ?? '') . PHP_EOL; foreach ($client->keyValueStore($store->getId())->iterateKeys() as $key) { echo $key->getKey() . PHP_EOL; } + +// Upload an already-gzipped value without the client compressing it a second time. +$gzipped = gzencode(json_encode(['answer' => 42])); +$client->keyValueStore($store->getId())->setRecord('OUTPUT.json', $gzipped, 'application/json', contentEncoding: 'gzip'); ``` ## Request queues diff --git a/src/Exception/ApifyApiException.php b/src/Exception/ApifyApiException.php index 2ac5aa3..1dbb09a 100644 --- a/src/Exception/ApifyApiException.php +++ b/src/Exception/ApifyApiException.php @@ -4,17 +4,16 @@ namespace Apify\Client\Exception; -use RuntimeException; - /** * Thrown for HTTP requests that reach the Apify API but receive a non-success status code. * * It mirrors the {@code ApifyApiError} of the reference JavaScript client and exposes the parsed * error {@see getType() type}, the human-readable {@see getMessage() message}, the HTTP * {@see getStatusCode() status code}, the number of the final {@see getAttempt() attempt}, and the - * request {@see getHttpMethod() method}/{@see getPath() path}. + * request {@see getHttpMethod() method}/{@see getPath() path}. Extends {@see ApifyClientException}, + * the common base every client error shares. */ -class ApifyApiException extends RuntimeException +class ApifyApiException extends ApifyClientException { private int $statusCode; private ?string $type; diff --git a/src/Exception/ApifyClientException.php b/src/Exception/ApifyClientException.php new file mode 100644 index 0000000..ff59f4d --- /dev/null +++ b/src/Exception/ApifyClientException.php @@ -0,0 +1,22 @@ +getMessage()), 0, $cause); + $this->rawBody = $rawBody; + } + + /** The response body exactly as received, for diagnosing the parse failure. */ + public function getRawBody(): string + { + return $this->rawBody; + } +} diff --git a/src/Internal/Json.php b/src/Internal/Json.php index cc993a8..ee4a5cf 100644 --- a/src/Internal/Json.php +++ b/src/Internal/Json.php @@ -4,6 +4,7 @@ namespace Apify\Client\Internal; +use Apify\Client\Exception\InvalidResponseBodyException; use JsonException; /** @@ -29,6 +30,12 @@ public static function encode(mixed $value): string /** * Decodes a JSON string into PHP values (objects become associative arrays). * + * Every caller of this method (directly or via {@see decodeData()}) decodes a response body the + * client just received over the wire, never a value the caller constructed itself (that path is + * {@see encode()}), so a malformed body is wrapped as an {@see InvalidResponseBodyException} — + * mirroring the reference client's {@code InvalidResponseBodyError} — instead of leaking the raw + * {@see JsonException}. + * * @return mixed the decoded value */ public static function decode(string $body): mixed @@ -36,7 +43,11 @@ public static function decode(string $body): mixed if ($body === '') { return null; } - return json_decode($body, true, self::MAX_JSON_DEPTH, JSON_THROW_ON_ERROR); + try { + return json_decode($body, true, self::MAX_JSON_DEPTH, JSON_THROW_ON_ERROR); + } catch (JsonException $e) { + throw new InvalidResponseBodyException($body, $e); + } } /** @@ -59,7 +70,7 @@ public static function tryDecode(string $body): mixed { try { return self::decode($body); - } catch (JsonException) { + } catch (InvalidResponseBodyException) { return null; } } diff --git a/src/Internal/ResourceContext.php b/src/Internal/ResourceContext.php index 40f19a3..175e81a 100644 --- a/src/Internal/ResourceContext.php +++ b/src/Internal/ResourceContext.php @@ -9,6 +9,7 @@ use Generator; use InvalidArgumentException; use Psr\Http\Message\ResponseInterface; +use Psr\Http\Message\StreamInterface; use RuntimeException; /** @@ -459,17 +460,52 @@ public function headExists(string $subPath, QueryParams $params, ?float $timeout } } - /** PUT with raw bytes and a content type, with an explicit per-request timeout and retry control. */ + /** + * Opens a live, streaming connection to {@code $subPath} and returns its body as a stream, for + * incremental consumption instead of buffering the whole response in memory. Bypasses the + * buffered/retrying transport ({@see HttpClientCore::stream()}), so because the response is + * consumed incrementally, it is not retried: a non-2xx status is read eagerly and thrown as the + * matching {@see ApifyApiException} (there is nothing to stream in that case anyway). + * + * Shared by {@see \Apify\Client\Resource\LogClient::stream()} and + * {@see \Apify\Client\Resource\DatasetClient::streamItems()}, the two endpoints that expose a + * streamed read alongside their buffered one. + */ + public function streamRaw(string $subPath, QueryParams $params): StreamInterface + { + $url = $this->mergedParams($params)->applyToUrl($this->subUrl($subPath)); + $response = $this->http->stream($url); + if ($response->getStatusCode() >= HttpClientCore::MAX_SUCCESS_STATUS) { + $body = (string) $response->getBody(); + throw HttpClientCore::buildApiError( + $response->getStatusCode(), + $body, + 1, + 'GET', + HttpClientCore::extractPath($url) + ); + } + return $response->getBody(); + } + + /** + * PUT with raw bytes and a content type, with an explicit per-request timeout and retry control. + * + * @param array $extraHeaders additional request headers, e.g. a caller-supplied + * {@code Content-Encoding} ({@see + * \Apify\Client\Resource\KeyValueStoreClient::setRecord()}) + */ public function putRaw( string $subPath, QueryParams $params, string $body, string $contentType, ?float $timeoutSecs = null, - bool $doNotRetryTimeouts = false + bool $doNotRetryTimeouts = false, + array $extraHeaders = [] ): void { $url = $this->mergedParams($params)->applyToUrl($this->subUrl($subPath)); - $this->http->call('PUT', $url, $body, $contentType, $timeoutSecs ?? $this->requestTimeoutSecs, $doNotRetryTimeouts); + $this->http->call('PUT', $url, $body, $contentType, $timeoutSecs ?? $this->requestTimeoutSecs, $doNotRetryTimeouts, $extraHeaders); } // ---- Wait-for-finish ------------------------------------------------------ diff --git a/src/Resource/DatasetClient.php b/src/Resource/DatasetClient.php index bbb41f4..fb46e65 100644 --- a/src/Resource/DatasetClient.php +++ b/src/Resource/DatasetClient.php @@ -17,6 +17,7 @@ use Apify\Client\Options\DownloadItemsFormat; use Generator; use Psr\Http\Message\ResponseInterface; +use Psr\Http\Message\StreamInterface; /** A client for a specific dataset (and run-nested variants). */ final class DatasetClient @@ -239,6 +240,32 @@ public function downloadItems( return (string) $response->getBody(); } + /** + * Opens a live, streaming connection to the dataset's items export in the given format and + * returns a stream over the serialized bytes, so a large export can be read (and, e.g., piped to a + * file) in chunks without holding the whole thing in memory. + * + * The reference client exposes this as a {@code stream} option of {@code downloadItems()} (having + * briefly shipped it as its own {@code streamItems()} method, folded back in one release later — + * apify/apify-client-js#1139, apify/apify-client-js#1152); this client keeps the separate method + * name since PHP has no equivalent of a single method whose return type switches on a boolean + * option, and a dedicated {@see streamItems()} name matches this client's existing {@see + * \Apify\Client\Resource\LogClient::get()}/{@see \Apify\Client\Resource\LogClient::stream()} split + * for the same buffered/streamed choice. + * + * Like {@see \Apify\Client\Resource\LogClient::stream()}, this bypasses the buffered/retrying + * transport, so because the response is consumed incrementally, it is not retried. + */ + public function streamItems( + DownloadItemsFormat $format, + ?DatasetDownloadOptions $options = null, + ): StreamInterface { + $params = new QueryParams(); + $params->addString('format', $format->value); + ($options ?? new DatasetDownloadOptions())->appendTo($params); + return $this->ctx->streamRaw('items', $params); + } + /** * Pushes one or more items to the dataset. * diff --git a/src/Resource/KeyValueStoreClient.php b/src/Resource/KeyValueStoreClient.php index d23cbeb..cf5d833 100644 --- a/src/Resource/KeyValueStoreClient.php +++ b/src/Resource/KeyValueStoreClient.php @@ -183,9 +183,24 @@ public function getRecord(string $key, ?GetRecordOptions $options = null, int|fl /** * Stores a record with raw bytes and the given content type, honoring the given write options * ({@code timeoutSecs}, {@code doNotRetryTimeouts}). + * + * {@code $contentEncoding}, if given, is sent as the request's {@code Content-Encoding} header, for + * a {@code $value} that is already compressed (gzip, br, deflate, ...): the client's own best-effort + * request compression ({@see \Apify\Client\Internal\Compression}) always backs off when a {@code + * Content-Encoding} header is already present (see {@see + * \Apify\Client\Internal\HttpClientCore::call()}), so the bytes are stored exactly as given, never + * compressed a second time. Unlike the reference client's {@code setRecord()}, this has no + * "non-identity encoding on a non-binary value" validation case to replicate: a PHP string is + * already the raw-bytes case there, since JSON values go through the separate {@see setRecordJson()} + * instead of a union type on one method. */ - public function setRecord(string $key, string $value, string $contentType, ?SetRecordOptions $options = null): void - { + public function setRecord( + string $key, + string $value, + string $contentType, + ?SetRecordOptions $options = null, + ?string $contentEncoding = null, + ): void { $options ??= new SetRecordOptions(); $timeoutSecs = $this->ctx->resolveTimeout($options->timeoutSecs, TimeoutTiers::TIER_LONG); $this->ctx->putRaw( @@ -194,7 +209,8 @@ public function setRecord(string $key, string $value, string $contentType, ?SetR $value, $contentType, $timeoutSecs, - $options->doNotRetryTimeouts + $options->doNotRetryTimeouts, + $contentEncoding !== null ? ['Content-Encoding' => $contentEncoding] : [] ); } diff --git a/src/Resource/LogClient.php b/src/Resource/LogClient.php index cf7bc65..9d45698 100644 --- a/src/Resource/LogClient.php +++ b/src/Resource/LogClient.php @@ -17,14 +17,14 @@ */ final class LogClient { - private function __construct(private HttpClientCore $http, private ResourceContext $ctx) + private function __construct(private ResourceContext $ctx) { } /** @internal */ public static function forId(HttpClientCore $http, string $baseUrl, string $id): self { - return new self($http, ResourceContext::single($http, $baseUrl, 'logs', $id)); + return new self(ResourceContext::single($http, $baseUrl, 'logs', $id)); } /** @@ -38,12 +38,16 @@ public static function nested(HttpClientCore $http, string $base, ?QueryParams $ if ($inheritedParams !== null) { $ctx->baseParams = $inheritedParams->copy(); } - return new self($http, $ctx); + return new self($ctx); } /** * Fetches the log as text, or {@code null} if the log does not exist. * + * Unlike the reference client, there is no separate {@code buffer} option: a PHP string is already + * a byte sequence, never re-decoded, so this already returns the exact bytes the API sent — the + * same thing the reference client's {@code buffer: true} opts into for its own string/Buffer split. + * * When this client was obtained without an ID (e.g. {@see \Apify\Client\Resource\RunClient::log()} or * {@see \Apify\Client\Resource\BuildClient::log()}), a 404 is rethrown instead, since it could mean * either the parent resource or the log is gone. @@ -68,19 +72,6 @@ public function stream(?LogOptions $options = null): StreamInterface $params = new QueryParams(); $params->addBool('stream', true); ($options ?? new LogOptions())->appendTo($params); - $url = $this->ctx->mergedParams($params)->applyToUrl($this->ctx->subUrl('')); - - $response = $this->http->stream($url); - if ($response->getStatusCode() >= HttpClientCore::MAX_SUCCESS_STATUS) { - $body = (string) $response->getBody(); - throw HttpClientCore::buildApiError( - $response->getStatusCode(), - $body, - 1, - 'GET', - HttpClientCore::extractPath($url) - ); - } - return $response->getBody(); + return $this->ctx->streamRaw('', $params); } } diff --git a/src/Version.php b/src/Version.php index f86d59d..b25a77a 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.9.0'; + public const CLIENT_VERSION = '0.10.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-08T073338Z'; + public const API_SPEC_VERSION = 'v2-2026-10-19T133704Z'; private function __construct() { diff --git a/tests/Examples/DocSnippetsTest.php b/tests/Examples/DocSnippetsTest.php index 01c4190..dec1a2c 100644 --- a/tests/Examples/DocSnippetsTest.php +++ b/tests/Examples/DocSnippetsTest.php @@ -75,6 +75,7 @@ private function imports(): string 'Apify\Client\Model\RequestQueueRequest', 'Apify\Client\Model\ActorEnvVar', 'Apify\Client\Exception\ApifyApiException', + 'Apify\Client\Exception\ApifyClientException', 'Apify\Client\Exception\NotFoundException', 'Apify\Client\Http\GuzzleHttpClient', 'Apify\Client\Http\Psr18HttpClient', diff --git a/tests/Integration/ActorIntegrationTest.php b/tests/Integration/ActorIntegrationTest.php index d5436e8..8aa7c8a 100644 --- a/tests/Integration/ActorIntegrationTest.php +++ b/tests/Integration/ActorIntegrationTest.php @@ -146,6 +146,27 @@ public function testIterateActorEnvVars(): void } } + public function testActorPermissionLevelRoundTrip(): void + { + // `actorPermissionLevel` (apify-docs #3087: CreateActorRequest/UpdateActorRequest) is not + // exposed via a typed Actor getter, only the generic ApifyResource::get() fallback, since this + // client passes create/update bodies through untyped (see client requirements) — this proves + // the field actually round-trips through the real API rather than only through this client's + // own types. + $client = $this->requireClient(); + $def = self::minimalActor(self::uniqueName('perm-level')); + $def['actorPermissionLevel'] = 'LIMITED_PERMISSIONS'; + $created = $client->actors()->create($def); + try { + self::assertSame('LIMITED_PERMISSIONS', $created->get('actorPermissionLevel')); + + $updated = $client->actor((string) $created->getId())->update(['actorPermissionLevel' => 'FULL_PERMISSIONS']); + self::assertSame('FULL_PERMISSIONS', $updated->get('actorPermissionLevel')); + } finally { + $client->actor((string) $created->getId())->delete(); + } + } + public function testValidateInput(): void { $client = $this->requireClient(); diff --git a/tests/Integration/DatasetIntegrationTest.php b/tests/Integration/DatasetIntegrationTest.php index 4e9acf3..2f6015f 100644 --- a/tests/Integration/DatasetIntegrationTest.php +++ b/tests/Integration/DatasetIntegrationTest.php @@ -87,6 +87,28 @@ public function testIterateDatasetItems(): void } } + public function testStreamItemsMatchesDownloadItems(): void + { + $client = $this->requireClient(); + $ds = $client->datasets()->getOrCreate(self::uniqueName('ds-stream')); + try { + $dataset = $client->dataset((string) $ds->getId()); + $dataset->pushItems([['n' => 1], ['n' => 2]]); + + $buffered = $dataset->downloadItems(DownloadItemsFormat::CSV); + $stream = $dataset->streamItems(DownloadItemsFormat::CSV); + $streamed = ''; + while (!$stream->eof()) { + $streamed .= $stream->read(8192); + } + + self::assertSame($buffered, $streamed); + self::assertStringContainsString('n', $buffered); + } finally { + $client->dataset((string) $ds->getId())->delete(); + } + } + public function testDatasetCrudFlow(): void { $client = $this->requireClient(); diff --git a/tests/Integration/KeyValueStoreIntegrationTest.php b/tests/Integration/KeyValueStoreIntegrationTest.php index fa4f8de..da8730d 100644 --- a/tests/Integration/KeyValueStoreIntegrationTest.php +++ b/tests/Integration/KeyValueStoreIntegrationTest.php @@ -118,6 +118,42 @@ public function testKeyValueStoreCrudFlow(): void } } + public function testSetRecordWithContentEncodingStoresPreCompressedValueAsIs(): void + { + $client = $this->requireClient(); + $store = $client->keyValueStores()->getOrCreate(self::uniqueName('kvs-gzip')); + try { + $kvs = $client->keyValueStore((string) $store->getId()); + // High-entropy padding (hex of random bytes), so it survives gzip compression large + // enough to stay above the client's own compression threshold (see assertion below) — + // repetitive padding would compress away almost entirely. + $original = json_encode(['hello' => 'world', 'n' => 42, 'padding' => bin2hex(random_bytes(2048))]); + self::assertIsString($original); + $gzipped = gzencode($original); + self::assertIsString($gzipped); + // A gzipped JSON payload is itself well over the 1 KiB auto-compression threshold, so + // this also proves the client's own compression genuinely backed off: if it had gzipped + // the already-gzipped bytes a second time, the API would receive double-compressed gzip + // under a single Content-Encoding: gzip header, decode it once, and store the *inner* + // gzip bytes as the record's "JSON" — which could never decode back to the original value + // below. + self::assertGreaterThan(1024, strlen($gzipped)); + + $kvs->setRecord('precompressed.json', $gzipped, 'application/json', contentEncoding: 'gzip'); + + // The live API decompresses a Content-Encoding-tagged upload on ingest and serves the + // record's plain bytes back out (it does not echo Content-Encoding on GET, nor keep the + // compressed bytes verbatim), so the round-trip is verified through the parsed value + // rather than a byte-for-byte comparison against $gzipped. + $record = $kvs->getRecord('precompressed.json'); + self::assertNotNull($record); + self::assertSame($original, $record->getValue()); + self::assertStringContainsString('application/json', (string) $record->getContentType()); + } finally { + $client->keyValueStore((string) $store->getId())->delete(); + } + } + public function testRecordPublicUrlIsFetchable(): void { $client = $this->requireClient(); diff --git a/tests/Integration/ScheduleIntegrationTest.php b/tests/Integration/ScheduleIntegrationTest.php index ea9b40e..ea2310c 100644 --- a/tests/Integration/ScheduleIntegrationTest.php +++ b/tests/Integration/ScheduleIntegrationTest.php @@ -83,4 +83,24 @@ public function testScheduleCrudFlow(): void $client->schedule((string) $sch->getId())->delete(); } } + + public function testScheduleNotificationsRoundTrip(): void + { + // `notifications` (apify-docs #3087: ScheduleNotifications) is not exposed via a typed + // Schedule getter, only the generic ApifyResource::get() fallback, since this client passes + // create/update bodies through untyped (see client requirements) — this proves the field + // actually round-trips through the real API rather than only through this client's own types. + $client = $this->requireClient(); + $def = self::scheduleDef(self::uniqueName('sch-notif')); + $def['notifications'] = ['email' => true]; + $sch = $client->schedules()->create($def); + try { + self::assertSame(['email' => true], $sch->get('notifications')); + + $updated = $client->schedule((string) $sch->getId())->update(['notifications' => ['email' => false]]); + self::assertSame(['email' => false], $updated->get('notifications')); + } finally { + $client->schedule((string) $sch->getId())->delete(); + } + } } diff --git a/tests/Unit/ApifyClientExceptionTest.php b/tests/Unit/ApifyClientExceptionTest.php new file mode 100644 index 0000000..225a914 --- /dev/null +++ b/tests/Unit/ApifyClientExceptionTest.php @@ -0,0 +1,72 @@ +queueResponse(404, '{"error":{"type":"record-not-found","message":"no"}}'); + + try { + // A chained accessor without an ID always rethrows a 404, giving us an exception to inspect. + $this->client($transport)->run('r1')->dataset()->delete(); + self::fail('expected ApifyApiException'); + } catch (ApifyApiException $e) { + self::assertInstanceOf(ApifyClientException::class, $e); + self::assertInstanceOf(NotFoundException::class, $e); + } + } + + public function testInvalidResponseBodyExceptionIsAnApifyClientException(): void + { + $transport = (new MockTransport())->queueResponse(200, 'this is not valid JSON {'); + + try { + $this->client($transport)->actor('a1')->get(); + self::fail('expected InvalidResponseBodyException'); + } catch (InvalidResponseBodyException $e) { + self::assertInstanceOf(ApifyClientException::class, $e); + self::assertSame('this is not valid JSON {', $e->getRawBody()); + self::assertStringContainsString('could not be parsed', $e->getMessage()); + } + } + + public function testTransportExceptionIsNotAnApifyClientException(): void + { + // Network/timeout failures stay outside the hierarchy, matching the reference client, where + // transport errors are not instances of ApifyClientError either. + // maxRetries: 0 so the single queued failure surfaces immediately instead of being retried. + $transport = (new MockTransport())->queueError(); + $client = new ApifyClient(token: 't', maxRetries: 0, timeoutSecs: 5, httpClient: $transport); + + try { + $client->actor('a1')->get(); + self::fail('expected TransportException'); + } catch (TransportException $e) { + self::assertNotInstanceOf(ApifyClientException::class, $e); + } + } +} diff --git a/tests/Unit/DatasetStreamItemsTest.php b/tests/Unit/DatasetStreamItemsTest.php new file mode 100644 index 0000000..19c6174 --- /dev/null +++ b/tests/Unit/DatasetStreamItemsTest.php @@ -0,0 +1,57 @@ +queueResponse(200, "url,n\nhttps://a.com,1\n"); + $stream = $this->client($transport)->dataset('ds1')->streamItems( + DownloadItemsFormat::CSV, + new DatasetDownloadOptions(bom: true) + ); + + self::assertSame("url,n\nhttps://a.com,1\n", (string) $stream); + $uri = (string) $transport->lastRequest()->getUri(); + self::assertStringContainsString('/datasets/ds1/items', $uri); + self::assertStringContainsString('format=csv', $uri); + self::assertStringContainsString('bom=1', $uri); + } + + public function testStreamItemsUsesStreamingTransportNotRetrying(): void + { + // A server error on a streamed request must surface immediately (streaming bypasses retries), + // unlike the buffered downloadItems()/get() calls, which retry 5xx responses. + $transport = (new MockTransport()) + ->queueResponse(500, Json::encode(['error' => ['type' => 'internal-error', 'message' => 'boom']])); + + try { + $this->client($transport)->dataset('ds1')->streamItems(DownloadItemsFormat::JSON); + self::fail('expected ApifyApiException'); + } catch (ApifyApiException $e) { + self::assertSame(500, $e->getStatusCode()); + self::assertSame(1, $transport->callCount()); + } + } +} diff --git a/tests/Unit/KeyValueStoreContentEncodingTest.php b/tests/Unit/KeyValueStoreContentEncodingTest.php new file mode 100644 index 0000000..59b57e7 --- /dev/null +++ b/tests/Unit/KeyValueStoreContentEncodingTest.php @@ -0,0 +1,75 @@ +queueResponse(200, ''); + $gzipped = gzencode('a value that would normally be compressed by the client itself'); + self::assertIsString($gzipped); + + $this->client($transport)->keyValueStore('s1')->setRecord( + 'report.json', + $gzipped, + 'application/json', + contentEncoding: 'gzip', + ); + + $request = $transport->lastRequest(); + self::assertSame('gzip', $request->getHeaderLine('Content-Encoding')); + // The body reaches the wire byte-for-byte: the client's own compression must have backed off + // (it already does for any caller-supplied Content-Encoding — see HttpClientCore::maybeCompressBody()), + // so the bytes are not gzipped a second time. + self::assertSame($gzipped, (string) $request->getBody()); + } + + public function testOmittingContentEncodingStillAllowsAutomaticCompression(): void + { + if (!function_exists('gzencode') && !function_exists('brotli_compress')) { + self::markTestSkipped('no compression codec available in this PHP build'); + } + $transport = (new MockTransport())->queueResponse(200, ''); + $large = str_repeat('x', 4096); + + $this->client($transport)->keyValueStore('s1')->setRecord('k', $large, 'text/plain'); + + $request = $transport->lastRequest(); + self::assertContains($request->getHeaderLine('Content-Encoding'), ['br', 'gzip']); + self::assertLessThan(strlen($large), strlen((string) $request->getBody())); + } + + public function testContentEncodingIsForwardedAlongsideWriteOptions(): void + { + $transport = (new MockTransport())->queueResponse(200, ''); + + $this->client($transport)->keyValueStore('s1')->setRecord( + 'k', + 'raw-bytes', + 'application/octet-stream', + new SetRecordOptions(doNotRetryTimeouts: true), + 'identity', + ); + + self::assertSame('identity', $transport->lastRequest()->getHeaderLine('Content-Encoding')); + self::assertSame('raw-bytes', (string) $transport->lastRequest()->getBody()); + } +} diff --git a/tests/Unit/LogClientTest.php b/tests/Unit/LogClientTest.php index d4ab553..17f94b1 100644 --- a/tests/Unit/LogClientTest.php +++ b/tests/Unit/LogClientTest.php @@ -44,6 +44,20 @@ public function testRunNestedLogGet(): void self::assertStringContainsString('raw=1', $uri); } + public function testGetReturnsExactBytesForNonUtf8Content(): void + { + // The reference client's get({ buffer: true }) exists so binary/non-UTF-8 log bytes are not + // mangled by decoding to a string. A PHP string is already a byte sequence and get() never + // decodes it, so it already behaves like that mode with no separate option needed — this + // proves bytes that are not valid UTF-8 still round-trip exactly. + $invalidUtf8 = "line with invalid utf-8: \xFF\xFE bytes\n"; + $transport = (new MockTransport())->queueResponse(200, $invalidUtf8); + $log = $this->client($transport)->log('run1')->get(); + + self::assertSame($invalidUtf8, $log); + self::assertSame(strlen($invalidUtf8), strlen((string) $log)); + } + public function testStreamedLogUsesStreamQueryAndReturnsReadableStream(): void { $transport = (new MockTransport())->queueResponse(200, 'streamed log body'); From 4cb12def3bc806242eb2290cc936e0e0f1799e75 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Oct 2026 17:20:22 +0000 Subject: [PATCH 2/3] fix: retry a malformed JSON response body like a retryable status HttpClientCore::call() now accepts an optional $validateBody hook, run against the buffered response body inside the existing retry loop before a successful status is returned; ResourceContext's JSON-decoding primitives (and the few direct resource-client call sites that decode JSON themselves) pass their decode step through it. A body that fails to parse as JSON - usually a connection dropped mid-response - is now retried with the same backoff as a 5xx/429, instead of failing on the very first attempt, matching the reference client's own retry-loop placement of this check. Also: fix InvalidResponseBodyException's message to emit a real newline (was a literal backslash-n from a single-quoted format string), update its docblock to stop asserting the now-fixed non-retry behavior, and trim changelog-trivia from DatasetClient::streamItems()'s docblock. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01G3FfQf9r45KVCa5VAYAjkd --- CHANGELOG.md | 4 +- README.md | 9 ++-- .../InvalidResponseBodyException.php | 10 +++-- src/Internal/HttpClientCore.php | 27 ++++++++++-- src/Internal/ResourceContext.php | 41 +++++++++++++------ src/Resource/BuildClient.php | 2 +- src/Resource/DatasetClient.php | 16 ++++---- src/Resource/RequestQueueClient.php | 5 ++- src/Resource/TaskClient.php | 5 ++- tests/Unit/ApifyClientExceptionTest.php | 5 ++- tests/Unit/HttpClientTest.php | 31 ++++++++++++++ 11 files changed, 115 insertions(+), 40 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b8dd130..5935808 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,7 +12,9 @@ - Added `ApifyClientException`, a common base class for every error the client throws for a failed API interaction. `ApifyApiException` (and its status subclasses) now extend it. - Added `InvalidResponseBodyException` (extends `ApifyClientException`), thrown instead of a raw - `JsonException` when a successful response body fails to parse as JSON. + `JsonException` when a successful response body fails to parse as JSON. For an endpoint whose body + is decoded as JSON, this is now retried with the same backoff as a retryable status code, matching + the reference client. ## 0.9.0 diff --git a/README.md b/README.md index 52fd73f..94a4277 100644 --- a/README.md +++ b/README.md @@ -206,10 +206,13 @@ exhausted does the underlying error surface, as an `Apify\Client\Exception\Trans `TransportException` means the request never produced a usable response (network failure or timeout) after all retries. Requests are retried on network errors, HTTP 429 and 5xx. +A successful (2xx) response whose body does not parse as JSON (e.g. a connection dropped +mid-response) is retried the same way, for an endpoint whose body this client decodes as JSON; only +once those retries are exhausted does it surface, as `Apify\Client\Exception\InvalidResponseBodyException`. + Every error the client throws for a failed API interaction — `ApifyApiException` and its status -subclasses above, plus `Apify\Client\Exception\InvalidResponseBodyException` (thrown when a -successful response's body does not parse as JSON, e.g. a connection dropped mid-response) — extends -the common `Apify\Client\Exception\ApifyClientException`, so one `catch` can handle both: +subclasses above, plus `InvalidResponseBodyException` — extends the common +`Apify\Client\Exception\ApifyClientException`, so one `catch` can handle both: ```php try { diff --git a/src/Exception/InvalidResponseBodyException.php b/src/Exception/InvalidResponseBodyException.php index 8565285..e7377da 100644 --- a/src/Exception/InvalidResponseBodyException.php +++ b/src/Exception/InvalidResponseBodyException.php @@ -11,9 +11,11 @@ * short by a connection dropped mid-response. Mirrors the reference client's * {@code InvalidResponseBodyError}. * - * Unlike a non-2xx status or a transport failure, this is raised by the decoding step that runs after - * {@see \Apify\Client\Internal\HttpClientCore::call()} already returned a successful response, so it is - * not covered by that method's retry loop and always surfaces on the first occurrence. + * For an endpoint whose body this client decodes as JSON, {@see \Apify\Client\Internal\HttpClientCore::call()} + * runs that decoding step itself before returning, inside its own retry loop, so this is retried exactly + * like a retryable status code and surfaces only once the retries are exhausted — matching the + * reference client, which parses a (non-streamed) response body inside its own retry loop for the same + * reason (a dropped connection is usually fixed by retrying). */ final class InvalidResponseBodyException extends ApifyClientException { @@ -22,7 +24,7 @@ final class InvalidResponseBodyException extends ApifyClientException public function __construct(string $rawBody, JsonException $cause) { - parent::__construct(sprintf('Response body could not be parsed as JSON.\nCause: %s', $cause->getMessage()), 0, $cause); + parent::__construct(sprintf("Response body could not be parsed as JSON.\nCause: %s", $cause->getMessage()), 0, $cause); $this->rawBody = $rawBody; } diff --git a/src/Internal/HttpClientCore.php b/src/Internal/HttpClientCore.php index b66ddef..74699d6 100644 --- a/src/Internal/HttpClientCore.php +++ b/src/Internal/HttpClientCore.php @@ -8,6 +8,7 @@ use Apify\Client\Exception\ConflictException; use Apify\Client\Exception\ForbiddenException; use Apify\Client\Exception\InvalidRequestException; +use Apify\Client\Exception\InvalidResponseBodyException; use Apify\Client\Exception\NotFoundException; use Apify\Client\Exception\RateLimitException; use Apify\Client\Exception\ServerException; @@ -83,6 +84,13 @@ public function timeoutTiers(): TimeoutTiers * Sends a request with auth, User-Agent and the retry policy applied. * * @param array $extraHeaders + * @param (callable(string): mixed)|null $validateBody For an endpoint whose successful response body this + * client decodes as JSON, the decoding step itself (e.g. {@see \Apify\Client\Internal\Json::decode()}), + * run here against the buffered body before this method returns, so a body that fails to parse — + * usually a connection dropped mid-response — is retried exactly like a retryable status code, instead + * of surfacing on the first occurrence. Matches the reference client, which parses a (non-streamed) + * response body inside its own retry loop for the same reason. Omit it for an endpoint whose body this + * client does not decode (raw bytes/text, or a response it never reads), which has nothing to validate. */ public function call( string $method, @@ -91,7 +99,8 @@ public function call( string $contentType = '', ?float $timeoutSecs = null, bool $doNotRetryTimeouts = false, - array $extraHeaders = [] + array $extraHeaders = [], + ?callable $validateBody = null ): ResponseInterface { // Compress the body once, up front, so every retry reuses the already-compressed payload. [$body, $extraHeaders] = $this->maybeCompressBody($body, $contentType, $extraHeaders); @@ -114,10 +123,20 @@ public function call( ); $status = $response->getStatusCode(); if ($status < self::MAX_SUCCESS_STATUS) { - return $response; + if ($validateBody === null) { + return $response; + } + try { + $validateBody((string) $response->getBody()); + return $response; + } catch (InvalidResponseBodyException $e) { + $lastError = $e; + $retryable = true; + } + } else { + $lastError = self::buildApiError($status, (string) $response->getBody(), $attempt, $method, $path); + $retryable = self::isStatusRetryable($status); } - $lastError = self::buildApiError($status, (string) $response->getBody(), $attempt, $method, $path); - $retryable = self::isStatusRetryable($status); } catch (TransportException $e) { $lastError = $e; // Network/timeout failures are retryable, unless the caller opted out of retrying timeouts. diff --git a/src/Internal/ResourceContext.php b/src/Internal/ResourceContext.php index 175e81a..676a02d 100644 --- a/src/Internal/ResourceContext.php +++ b/src/Internal/ResourceContext.php @@ -166,7 +166,7 @@ public function getResource(string $subPath, QueryParams $params, ?float $timeou public function getResourceRequired(string $subPath, QueryParams $params, ?float $timeoutSecs = null): mixed { $url = $this->mergedParams($params)->applyToUrl($this->subUrl($subPath)); - $response = $this->http->call('GET', $url, timeoutSecs: $this->effectiveTimeout($timeoutSecs)); + $response = $this->http->call('GET', $url, timeoutSecs: $this->effectiveTimeout($timeoutSecs), validateBody: Json::decode(...)); return Json::decodeData((string) $response->getBody()); } @@ -178,7 +178,7 @@ public function getResourceRequired(string $subPath, QueryParams $params, ?float public function updateResource(string $subPath, mixed $body, ?float $timeoutSecs = null): array { $url = $this->mergedParams(new QueryParams())->applyToUrl($this->subUrl($subPath)); - $response = $this->http->call('PUT', $url, Json::encode($body), self::CONTENT_TYPE_JSON, timeoutSecs: $this->effectiveTimeout($timeoutSecs)); + $response = $this->http->call('PUT', $url, Json::encode($body), self::CONTENT_TYPE_JSON, timeoutSecs: $this->effectiveTimeout($timeoutSecs), validateBody: Json::decode(...)); return self::asArray(Json::decodeData((string) $response->getBody())); } @@ -347,7 +347,7 @@ public static function minLimit(?int $a, ?int $b): ?int public function createResource(QueryParams $params, mixed $body, ?float $timeoutSecs = null): array { $url = $this->mergedParams($params)->applyToUrl($this->subUrl('')); - $response = $this->http->call('POST', $url, Json::encode($body), self::CONTENT_TYPE_JSON, timeoutSecs: $this->effectiveTimeout($timeoutSecs)); + $response = $this->http->call('POST', $url, Json::encode($body), self::CONTENT_TYPE_JSON, timeoutSecs: $this->effectiveTimeout($timeoutSecs), validateBody: Json::decode(...)); return self::asArray(Json::decodeData((string) $response->getBody())); } @@ -368,8 +368,8 @@ public function getOrCreateNamed(?string $name, ?array $schema = null, ?float $t $url = $params->applyToUrl($this->subUrl('')); $effectiveTimeout = $this->effectiveTimeout($timeoutSecs); $response = $schema !== null - ? $this->http->call('POST', $url, Json::encode(['schema' => $schema]), self::CONTENT_TYPE_JSON, timeoutSecs: $effectiveTimeout) - : $this->http->call('POST', $url, timeoutSecs: $effectiveTimeout); + ? $this->http->call('POST', $url, Json::encode(['schema' => $schema]), self::CONTENT_TYPE_JSON, timeoutSecs: $effectiveTimeout, validateBody: Json::decode(...)) + : $this->http->call('POST', $url, timeoutSecs: $effectiveTimeout, validateBody: Json::decode(...)); return self::asArray(Json::decodeData((string) $response->getBody())); } @@ -381,7 +381,7 @@ public function getOrCreateNamed(?string $name, ?array $schema = null, ?float $t public function postWithBody(string $subPath, QueryParams $params, ?string $body, string $contentType, ?float $timeoutSecs = null): array { $url = $this->mergedParams($params)->applyToUrl($this->subUrl($subPath)); - $response = $this->http->call('POST', $url, $body, $contentType, timeoutSecs: $this->effectiveTimeout($timeoutSecs)); + $response = $this->http->call('POST', $url, $body, $contentType, timeoutSecs: $this->effectiveTimeout($timeoutSecs), validateBody: Json::decode(...)); return self::asArray(Json::decodeData((string) $response->getBody())); } @@ -394,7 +394,7 @@ public function postWithBody(string $subPath, QueryParams $params, ?string $body public function postWithBodyNoEnvelope(string $subPath, QueryParams $params, ?string $body, string $contentType, ?float $timeoutSecs = null): mixed { $url = $this->mergedParams($params)->applyToUrl($this->subUrl($subPath)); - $response = $this->http->call('POST', $url, $body, $contentType, timeoutSecs: $this->effectiveTimeout($timeoutSecs)); + $response = $this->http->call('POST', $url, $body, $contentType, timeoutSecs: $this->effectiveTimeout($timeoutSecs), validateBody: Json::decode(...)); return Json::decode((string) $response->getBody()); } @@ -406,7 +406,7 @@ public function postWithBodyNoEnvelope(string $subPath, QueryParams $params, ?st public function deleteWithBody(string $subPath, QueryParams $params, mixed $body, ?float $timeoutSecs = null): array { $url = $this->mergedParams($params)->applyToUrl($this->subUrl($subPath)); - $response = $this->http->call('DELETE', $url, Json::encode($body), self::CONTENT_TYPE_JSON, timeoutSecs: $this->effectiveTimeout($timeoutSecs)); + $response = $this->http->call('DELETE', $url, Json::encode($body), self::CONTENT_TYPE_JSON, timeoutSecs: $this->effectiveTimeout($timeoutSecs), validateBody: Json::decode(...)); return self::asArray(Json::decodeData((string) $response->getBody())); } @@ -414,11 +414,13 @@ public function deleteWithBody(string $subPath, QueryParams $params, mixed $body * GET returning the raw response (no data envelope). Returns {@code null} on not-found, unless * this context addresses an ID-less chained resource ({@see $hasId}) and {@code $subPath} is the * resource itself ({@code ''}), in which case the 404 is rethrown (see {@see deleteResource()}). + * + * @param (callable(string): mixed)|null $validateBody See {@see getRawRequired()}. */ - public function getRaw(string $subPath, QueryParams $params, ?float $timeoutSecs = null): ?ResponseInterface + public function getRaw(string $subPath, QueryParams $params, ?float $timeoutSecs = null, ?callable $validateBody = null): ?ResponseInterface { try { - return $this->getRawRequired($subPath, $params, $timeoutSecs); + return $this->getRawRequired($subPath, $params, $timeoutSecs, $validateBody); } catch (ApifyApiException $e) { if (HttpClientCore::isNotFound($e) && $this->swallowsNotFound($subPath)) { return null; @@ -427,11 +429,24 @@ public function getRaw(string $subPath, QueryParams $params, ?float $timeoutSecs } } - /** GET returning the raw response (no data envelope). Propagates every error, including a 404. */ - public function getRawRequired(string $subPath, QueryParams $params, ?float $timeoutSecs = null): ResponseInterface + /** + * GET returning the raw response (no data envelope). Propagates every error, including a 404. + * + * This primitive is shared by endpoints whose body the caller decodes as JSON afterward (e.g. + * {@see \Apify\Client\Resource\TaskClient::getInput()}) and ones that return raw bytes/text it + * never decodes (e.g. {@see \Apify\Client\Resource\ScheduleClient::getLog()}), so — unlike the + * JSON-returning primitives above, which always decode and can validate unconditionally — it + * cannot validate the body itself. A JSON-decoding caller passes its decode step (e.g. {@see + * \Apify\Client\Internal\Json::decode()}) as {@code $validateBody} so a malformed body is retried + * instead of failing on the first occurrence (see {@see HttpClientCore::call()}); a raw caller + * omits it. + * + * @param (callable(string): mixed)|null $validateBody + */ + public function getRawRequired(string $subPath, QueryParams $params, ?float $timeoutSecs = null, ?callable $validateBody = null): ResponseInterface { $url = $this->mergedParams($params)->applyToUrl($this->subUrl($subPath)); - return $this->http->call('GET', $url, timeoutSecs: $this->effectiveTimeout($timeoutSecs)); + return $this->http->call('GET', $url, timeoutSecs: $this->effectiveTimeout($timeoutSecs), validateBody: $validateBody); } /** diff --git a/src/Resource/BuildClient.php b/src/Resource/BuildClient.php index d381382..49a4765 100644 --- a/src/Resource/BuildClient.php +++ b/src/Resource/BuildClient.php @@ -78,7 +78,7 @@ public function waitForFinish(?int $waitSecs = null): Build */ public function getOpenApiDefinition(int|float|string|null $timeoutSecs = null): ?array { - $response = $this->ctx->getRawRequired('openapi.json', new QueryParams(), $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_MEDIUM)); + $response = $this->ctx->getRawRequired('openapi.json', new QueryParams(), $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_MEDIUM), Json::decode(...)); $decoded = Json::decode((string) $response->getBody()); return is_array($decoded) ? $decoded : null; } diff --git a/src/Resource/DatasetClient.php b/src/Resource/DatasetClient.php index fb46e65..1723a86 100644 --- a/src/Resource/DatasetClient.php +++ b/src/Resource/DatasetClient.php @@ -198,7 +198,7 @@ private function fetchItemsPage(DatasetListItemsOptions $options, ?float $timeou $params = new QueryParams(); $options->appendTo($params); $url = $this->ctx->mergedParams($params)->applyToUrl($this->ctx->subUrl('items')); - $response = $this->http->call('GET', $url, timeoutSecs: $timeoutSecs); + $response = $this->http->call('GET', $url, timeoutSecs: $timeoutSecs, validateBody: Json::decode(...)); $items = Json::decode((string) $response->getBody()); $items = is_array($items) ? array_values($items) : []; @@ -245,13 +245,11 @@ public function downloadItems( * returns a stream over the serialized bytes, so a large export can be read (and, e.g., piped to a * file) in chunks without holding the whole thing in memory. * - * The reference client exposes this as a {@code stream} option of {@code downloadItems()} (having - * briefly shipped it as its own {@code streamItems()} method, folded back in one release later — - * apify/apify-client-js#1139, apify/apify-client-js#1152); this client keeps the separate method - * name since PHP has no equivalent of a single method whose return type switches on a boolean - * option, and a dedicated {@see streamItems()} name matches this client's existing {@see - * \Apify\Client\Resource\LogClient::get()}/{@see \Apify\Client\Resource\LogClient::stream()} split - * for the same buffered/streamed choice. + * The reference client exposes this as a {@code stream} option of {@code downloadItems()}; this + * client keeps a separate method instead, since PHP has no equivalent of a single method whose + * return type switches on a boolean option, and a dedicated {@see streamItems()} name matches this + * client's existing {@see \Apify\Client\Resource\LogClient::get()}/{@see + * \Apify\Client\Resource\LogClient::stream()} split for the same buffered/streamed choice. * * Like {@see \Apify\Client\Resource\LogClient::stream()}, this bypasses the buffered/retrying * transport, so because the response is consumed incrementally, it is not retried. @@ -295,7 +293,7 @@ public function pushItems(mixed $items, int|float|string|null $timeoutSecs = nul */ public function getStatistics(int|float|string|null $timeoutSecs = null): ?array { - $response = $this->ctx->getRawRequired('statistics', new QueryParams(), $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_SHORT)); + $response = $this->ctx->getRawRequired('statistics', new QueryParams(), $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_SHORT), Json::decode(...)); $decoded = Json::decodeData((string) $response->getBody()); return is_array($decoded) ? $decoded : null; } diff --git a/src/Resource/RequestQueueClient.php b/src/Resource/RequestQueueClient.php index 482573f..a9e69e0 100644 --- a/src/Resource/RequestQueueClient.php +++ b/src/Resource/RequestQueueClient.php @@ -191,7 +191,8 @@ public function updateRequest(RequestQueueRequest $request, bool $forefront = fa $url, Json::encode($request->toArray()), ResourceContext::CONTENT_TYPE_JSON, - timeoutSecs: $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_MEDIUM) + timeoutSecs: $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_MEDIUM), + validateBody: Json::decode(...) ); $data = Json::decodeData((string) $response->getBody()); return new RequestQueueOperationInfo(is_array($data) ? $data : []); @@ -516,7 +517,7 @@ public function prolongRequestLock( $this->applyClientKey($params); $url = $this->ctx->mergedParams($params) ->applyToUrl($this->ctx->subUrl('requests/' . ResourceContext::encodePathSegment($id) . '/lock')); - $response = $this->http->call('PUT', $url, null, '', timeoutSecs: $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_MEDIUM)); + $response = $this->http->call('PUT', $url, null, '', timeoutSecs: $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_MEDIUM), validateBody: Json::decode(...)); return RequestLockInfo::fromData(Json::decodeData((string) $response->getBody())); } diff --git a/src/Resource/TaskClient.php b/src/Resource/TaskClient.php index a9b3b17..ea09ac0 100644 --- a/src/Resource/TaskClient.php +++ b/src/Resource/TaskClient.php @@ -139,7 +139,7 @@ public function call(mixed $input = null, ?TaskStartOptions $options = null, ?in */ public function getInput(int|float|string|null $timeoutSecs = null): mixed { - $response = $this->ctx->getRawRequired('input', new QueryParams(), $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_SHORT)); + $response = $this->ctx->getRawRequired('input', new QueryParams(), $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_SHORT), Json::decode(...)); return Json::decode((string) $response->getBody()); } @@ -156,7 +156,8 @@ public function updateInput(mixed $input, int|float|string|null $timeoutSecs = n $this->ctx->subUrl('input'), Json::encode($input), ResourceContext::CONTENT_TYPE_JSON, - $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_SHORT) + $this->ctx->resolveTimeout($timeoutSecs, TimeoutTiers::TIER_SHORT), + validateBody: Json::decode(...) ); return Json::decode((string) $response->getBody()); } diff --git a/tests/Unit/ApifyClientExceptionTest.php b/tests/Unit/ApifyClientExceptionTest.php index 225a914..294cc26 100644 --- a/tests/Unit/ApifyClientExceptionTest.php +++ b/tests/Unit/ApifyClientExceptionTest.php @@ -42,10 +42,13 @@ public function testApifyApiExceptionIsAnApifyClientException(): void public function testInvalidResponseBodyExceptionIsAnApifyClientException(): void { + // maxRetries: 0 so the single queued malformed body surfaces immediately instead of being + // retried (see HttpClientTest for coverage of the retry behavior itself). $transport = (new MockTransport())->queueResponse(200, 'this is not valid JSON {'); + $client = new ApifyClient(token: 't', maxRetries: 0, timeoutSecs: 5, httpClient: $transport); try { - $this->client($transport)->actor('a1')->get(); + $client->actor('a1')->get(); self::fail('expected InvalidResponseBodyException'); } catch (InvalidResponseBodyException $e) { self::assertInstanceOf(ApifyClientException::class, $e); diff --git a/tests/Unit/HttpClientTest.php b/tests/Unit/HttpClientTest.php index c53f60e..e68372a 100644 --- a/tests/Unit/HttpClientTest.php +++ b/tests/Unit/HttpClientTest.php @@ -6,6 +6,7 @@ use Apify\Client\ApifyClient; use Apify\Client\Exception\ApifyApiException; +use Apify\Client\Exception\InvalidResponseBodyException; use Apify\Client\Internal\Json; use Apify\Client\Options\DatasetListItemsOptions; use PHPUnit\Framework\TestCase; @@ -76,6 +77,36 @@ public function testValidationErrorIsNotRetriedAndThrows(): void } } + public function testMalformedJsonBodyIsRetriedThenSucceeds(): void + { + // A successful (2xx) response whose body fails to parse as JSON is retried exactly like a + // retryable status code — matching the reference client, which parses the body inside its own + // retry loop for the same reason (a connection dropped mid-response is usually fixed by a retry). + $transport = (new MockTransport()) + ->queueResponse(200, 'not valid JSON {') + ->queueResponse(200, Json::encode(['data' => ['id' => 'recovered']])); + + $actor = $this->client($transport)->actor('x')->get(); + self::assertSame('recovered', $actor?->getId()); + self::assertSame(2, $transport->callCount()); + } + + public function testMalformedJsonBodyThrowsOnceRetriesAreExhausted(): void + { + $transport = (new MockTransport()) + ->queueResponse(200, 'not valid JSON {') + ->queueResponse(200, 'still not valid JSON {'); + $client = new ApifyClient(token: 'test-token', maxRetries: 1, minDelayBetweenRetriesMillis: 1, timeoutSecs: 5, httpClient: $transport); + + try { + $client->actor('x')->get(); + self::fail('expected InvalidResponseBodyException'); + } catch (InvalidResponseBodyException $e) { + self::assertSame('still not valid JSON {', $e->getRawBody()); + self::assertSame(2, $transport->callCount()); + } + } + public function testTransportErrorsAreRetried(): void { $transport = (new MockTransport()) From f4a86a32e1d1d8fb6ab8f8d2d5062b8398063740 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Oct 2026 17:52:23 +0000 Subject: [PATCH 3/3] fix: buffer the response body once in HttpClientCore::call() A validated (JSON-decoded) response body was read twice: call() ran $validateBody against (string) $response->getBody() and discarded the result, then returned the same $response; every JSON call site in ResourceContext then re-read the body via a second (string) $response->getBody(). A seekable stream (the bundled Guzzle client, MockTransport) rewinds transparently on each cast, masking this, but PSR-18 does not guarantee a seekable/re-readable body, so a transport that returns one exhausts it on the first read and the second read silently returns '' - the resource then looks empty/absent with no exception. Buffer the body into a string exactly once, validate it, and hand the caller a response rewrapped around a fresh stream built from that buffered string via the injected StreamFactoryInterface, which backs it with a seekable resource either way. Added a regression test using a GuzzleHttp\Psr7\NoSeekStream response body (confirmed it fails against the prior code) and a MockTransport::queueRawResponse() helper to queue it. Also collapsed the retry-on-malformed-JSON rationale, previously restated in three docblocks (HttpClientCore::call's $validateBody @param, Json::decode, and InvalidResponseBodyException's class doc), down to one canonical copy on HttpClientCore::call(), with the other two trimmed to a one-line cross-reference. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01G3FfQf9r45KVCa5VAYAjkd --- .../InvalidResponseBodyException.php | 10 +++------ src/Internal/HttpClientCore.php | 12 +++++++++-- src/Internal/Json.php | 6 +++--- tests/Unit/HttpClientTest.php | 21 +++++++++++++++++++ tests/Unit/MockTransport.php | 10 +++++++++ 5 files changed, 47 insertions(+), 12 deletions(-) diff --git a/src/Exception/InvalidResponseBodyException.php b/src/Exception/InvalidResponseBodyException.php index e7377da..0c55756 100644 --- a/src/Exception/InvalidResponseBodyException.php +++ b/src/Exception/InvalidResponseBodyException.php @@ -9,13 +9,9 @@ /** * Thrown when a successful (2xx) response body does not parse as valid JSON, typically a document cut * short by a connection dropped mid-response. Mirrors the reference client's - * {@code InvalidResponseBodyError}. - * - * For an endpoint whose body this client decodes as JSON, {@see \Apify\Client\Internal\HttpClientCore::call()} - * runs that decoding step itself before returning, inside its own retry loop, so this is retried exactly - * like a retryable status code and surfaces only once the retries are exhausted — matching the - * reference client, which parses a (non-streamed) response body inside its own retry loop for the same - * reason (a dropped connection is usually fixed by retrying). + * {@code InvalidResponseBodyError}. See {@see \Apify\Client\Internal\HttpClientCore::call()}'s + * {@code $validateBody} param for why this is retried like a retryable status code instead of thrown on + * the first occurrence. */ final class InvalidResponseBodyException extends ApifyClientException { diff --git a/src/Internal/HttpClientCore.php b/src/Internal/HttpClientCore.php index 74699d6..2a65e2e 100644 --- a/src/Internal/HttpClientCore.php +++ b/src/Internal/HttpClientCore.php @@ -126,9 +126,17 @@ public function call( if ($validateBody === null) { return $response; } + // Buffer the body into a string exactly once: a PSR-18 transport's body is not + // guaranteed seekable/re-readable (@see HttpClientInterface), so a second + // `(string) $response->getBody()` by the caller (every JSON call site does this to + // get the body it already validated here) could otherwise read past EOF and see + // '' instead of the real payload. The response handed back therefore carries a + // fresh, rewound stream built from that buffered string, which every PSR-7 stream + // factory implementation backs with a seekable resource. + $raw = (string) $response->getBody(); try { - $validateBody((string) $response->getBody()); - return $response; + $validateBody($raw); + return $response->withBody($this->streamFactory->createStream($raw)); } catch (InvalidResponseBodyException $e) { $lastError = $e; $retryable = true; diff --git a/src/Internal/Json.php b/src/Internal/Json.php index ee4a5cf..236196f 100644 --- a/src/Internal/Json.php +++ b/src/Internal/Json.php @@ -32,9 +32,9 @@ public static function encode(mixed $value): string * * Every caller of this method (directly or via {@see decodeData()}) decodes a response body the * client just received over the wire, never a value the caller constructed itself (that path is - * {@see encode()}), so a malformed body is wrapped as an {@see InvalidResponseBodyException} — - * mirroring the reference client's {@code InvalidResponseBodyError} — instead of leaking the raw - * {@see JsonException}. + * {@see encode()}), so a malformed body is wrapped as an {@see InvalidResponseBodyException} instead + * of leaking the raw {@see JsonException}. See {@see \Apify\Client\Internal\HttpClientCore::call()}'s + * {@code $validateBody} param for why this is retried rather than thrown on the first occurrence. * * @return mixed the decoded value */ diff --git a/tests/Unit/HttpClientTest.php b/tests/Unit/HttpClientTest.php index e68372a..6eac724 100644 --- a/tests/Unit/HttpClientTest.php +++ b/tests/Unit/HttpClientTest.php @@ -9,6 +9,9 @@ use Apify\Client\Exception\InvalidResponseBodyException; use Apify\Client\Internal\Json; use Apify\Client\Options\DatasetListItemsOptions; +use GuzzleHttp\Psr7\NoSeekStream; +use GuzzleHttp\Psr7\Response; +use GuzzleHttp\Psr7\Utils; use PHPUnit\Framework\TestCase; final class HttpClientTest extends TestCase @@ -107,6 +110,24 @@ public function testMalformedJsonBodyThrowsOnceRetriesAreExhausted(): void } } + public function testNonSeekableResponseBodyIsOnlyReadOnce(): void + { + // A PSR-18 transport's response body is not guaranteed seekable/re-readable (the contract + // HttpClientInterface/Psr18HttpClient exposes); a NoSeekStream reproduces that here. call() + // must validate/return the body from a single buffered read, since a second + // `(string) $response->getBody()` on a stream like this one yields '' once already exhausted + // (GuzzleHttp\Psr7\StreamDecoratorTrait::__toString() skips the rewind-on-seek it does for a + // seekable stream), which would otherwise make the resource silently look empty. + $body = new NoSeekStream(Utils::streamFor(Json::encode(['data' => ['id' => 'abc', 'name' => 'my-actor']]))); + $transport = (new MockTransport())->queueRawResponse(new Response(200, [], $body)); + + $actor = $this->client($transport)->actor('abc')->get(); + + self::assertNotNull($actor); + self::assertSame('abc', $actor->getId()); + self::assertSame('my-actor', $actor->getName()); + } + public function testTransportErrorsAreRetried(): void { $transport = (new MockTransport()) diff --git a/tests/Unit/MockTransport.php b/tests/Unit/MockTransport.php index 359977e..4f7c65b 100644 --- a/tests/Unit/MockTransport.php +++ b/tests/Unit/MockTransport.php @@ -42,6 +42,16 @@ public function queueError(bool $timeout = false): self return $this; } + /** + * Queues a fully-constructed response as-is, e.g. one built with a non-seekable body, instead of + * the plain string body {@see queueResponse()} always wraps in a seekable stream. + */ + public function queueRawResponse(ResponseInterface $response): self + { + $this->queue[] = $response; + return $this; + } + public function lastRequest(): RequestInterface { if ($this->received === []) {