From 193e028e3118adb8e6e15233155c2253bca58064 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 12 Oct 2026 08:11:59 +0000 Subject: [PATCH 1/3] feat: sync Java client with apify-client-js reference updates and spec v2-2026-10-08T073338Z - UserProfile.getReadme() (new Profile.readme spec field) - waitForResources on Actor/Task start+call options: retries a run start rejected for lack of memory or a concurrent-run slot, mirroring apify-client-js#1091 - waitForFinish now gives up on a run/build id that keeps 404-ing after 3s instead of polling it for the whole wait budget, mirroring apify-client-js#1124 - StreamedLog reopens a log stream that ended with zero bytes (a run that hasn't logged anything yet) instead of silently giving up, with a one-shot fallback fetch on stop(), mirroring apify-client-js#1105 --- CHANGELOG.md | 24 ++++ README.md | 6 +- docs/actors.md | 7 +- docs/builds.md | 2 +- docs/misc.md | 5 +- docs/runs.md | 10 ++ docs/tasks.md | 3 +- pom.xml | 2 +- src/main/java/com/apify/client/Version.java | 4 +- .../apify/client/actor/ActorCallOptions.java | 12 ++ .../com/apify/client/actor/ActorClient.java | 18 ++- .../apify/client/actor/ActorStartOptions.java | 40 +++++- .../client/internal/ResourceContext.java | 89 +++++++++--- .../client/internal/RunStartSupport.java | 110 ++++++++++++-- .../com/apify/client/log/StreamedLog.java | 135 +++++++++++++++++- .../apify/client/task/TaskCallOptions.java | 12 ++ .../com/apify/client/task/TaskClient.java | 18 ++- .../apify/client/task/TaskStartOptions.java | 34 ++++- .../com/apify/client/user/UserProfile.java | 6 + .../client/ClientBehaviourRegressionTest.java | 101 ++++++++++++- .../java/com/apify/client/MockTransport.java | 59 +++++--- .../com/apify/client/StreamedLogTest.java | 40 ++++++ 22 files changed, 653 insertions(+), 84 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bcdc8e9..b137456 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,30 @@ All notable changes to the Apify Java client are documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.8.0] - 2026-10-12 + +### Added + +- `UserProfile.getReadme()`, mirroring the OpenAPI spec's new `Profile.readme` field. +- `ActorStartOptions.waitForResources()`/`TaskStartOptions.waitForResources()` (and the matching + `ActorCallOptions`/`TaskCallOptions` overloads): retry starting a run while the account lacks the + memory or a concurrent-run slot for it, matching the reference client. + +### Fixed + +- `RunClient.waitForFinish()`/`BuildClient.waitForFinish()` polled a run/build id that kept + returning 404 for the entire wait budget (up to ~11.5 days when unbounded) instead of giving up + promptly; they now throw the `NotFoundError` once the id has been missing for 3 seconds straight, + matching the reference client. +- `StreamedLog` silently stopped redirecting (without ever reopening) when the live log stream ended + immediately with no bytes - the case for a run that has not logged anything yet (e.g. still queued + for resources). It now reopens the stream, with capped exponential backoff, until it delivers + content; if stopped first, it fetches the log in one plain request instead of redirecting nothing. + +### Changed + +- Updated to Apify OpenAPI specification `v2-2026-10-08T073338Z`. + ## [0.7.0] - 2026-10-02 ### Added diff --git a/README.md b/README.md index 56208d5..63395a6 100644 --- a/README.md +++ b/README.md @@ -23,7 +23,7 @@ Maven (Maven Central is a default repository, so no extra configuration is neede com.apify apify-client - 0.7.0 + 0.8.0 ``` @@ -35,7 +35,7 @@ repositories { } dependencies { - implementation 'com.apify:apify-client:0.7.0' + implementation 'com.apify:apify-client:0.8.0' } ``` @@ -317,7 +317,7 @@ try { The public `com.apify.client.Version` class (`import com.apify.client.Version;`) exposes two constants: -- `Version.CLIENT_VERSION` — the semantic version of this client (`0.7.0`). +- `Version.CLIENT_VERSION` — the semantic version of this client (`0.8.0`). - `Version.API_SPEC_VERSION` — the version of the [Apify OpenAPI specification](https://docs.apify.com/api/openapi.json) (its `info.version` field) that this client's endpoints, parameters and models were last generated and checked against (`v2-2026-10-01T153946Z`). It is a snapshot, not a live compatibility diff --git a/docs/actors.md b/docs/actors.md index 122c02d..43c1f3f 100644 --- a/docs/actors.md +++ b/docs/actors.md @@ -65,9 +65,12 @@ to charge, pay-per-result Actors), `maxTotalChargeUsd(Double)` (maximum total ch pay-per-event Actors), `contentType(String)` (content type of the input body, defaults to `application/json`), `restartOnError(Boolean)` (restart the run if it fails), `forcePermissionLevel(String)` (override the Actor's permission level for this run: -`LIMITED_PERMISSIONS`/`FULL_PERMISSIONS`), and `webhooks(List)` — ad-hoc webhook definitions +`LIMITED_PERMISSIONS`/`FULL_PERMISSIONS`), `webhooks(List)` — ad-hoc webhook definitions (each a JSON-serializable `Map`, as in [Webhooks](webhooks.md)) that the client base64-encodes on the -wire. +wire — and `waitForResources(boolean)`/`waitForResources(Long maxSecs)`: retry the start, every 10 +seconds, while the account lacks the memory or a concurrent-run slot for the run (`true` retries +until it starts — `ActorStartOptions.WAIT_FOR_RESOURCES_INDEFINITELY`; a bound throws the last such +error once exceeded; omitted, the default, does not retry at all). `ActorCallOptions` (for the log-streaming `call` overload) mirrors `ActorStartOptions` field for field, but omits `waitForFinish`: that field asks the API to hold the HTTP response open diff --git a/docs/builds.md b/docs/builds.md index cc71c16..b5704a4 100644 --- a/docs/builds.md +++ b/docs/builds.md @@ -18,7 +18,7 @@ builds) and a single build with `client.build(id)`. | `getWithWait(Long waitForFinishSecs)` | Fetch, optionally waiting server-side for the build to finish (clamped to the request timeout; the API caps server-side waiting at 60s). Completes with `Optional`. | | `abort()` | Abort the build. Completes with `Build`. | | `delete()` | Delete the build. | -| `waitForFinish(Long waitSecs)` | Poll until the build finishes (`null` waits indefinitely). Completes with `Build`. | +| `waitForFinish(Long waitSecs)` | Poll until the build finishes (`null` waits indefinitely); see [the run equivalent](runs.md#runclient) for its 404 handling. Completes with `Build`. | | `getOpenApiDefinition()` | The build's OpenAPI definition. Completes with `Optional`. | | `log()` | A `LogClient` for the build's log. | diff --git a/docs/misc.md b/docs/misc.md index ecda77c..d45be9a 100644 --- a/docs/misc.md +++ b/docs/misc.md @@ -57,8 +57,9 @@ me.ifPresent( JsonNode usage = client.me().monthlyUsage().join(); ``` -`User` fields: `getId()`, `getUsername()`, `getProfile()` (`UserProfile` — `getBio()`, `getName()`, -`getPictureUrl()`, `getGithubUsername()`, `getWebsiteUrl()`, `getTwitterUsername()`, all nullable). +`User` fields: `getId()`, `getUsername()`, `getProfile()` (`UserProfile` — `getBio()`, `getReadme()`, +`getName()`, `getPictureUrl()`, `getGithubUsername()`, `getWebsiteUrl()`, `getTwitterUsername()`, all +nullable). Only present for `me()`: `getEmail()`, `getProxy()` (`UserProxy` — `getPassword()`, `getGroups()` as `List`, each with `getName()`/`getDescription()`/`getAvailableCount()` as `Long`), `getPlan()` (`UserPlan` — `getId()`, `getDescription()`, `getIsEnabled()`, diff --git a/docs/runs.md b/docs/runs.md index 70cfac5..f406ddf 100644 --- a/docs/runs.md +++ b/docs/runs.md @@ -51,6 +51,10 @@ this same possibly-non-terminal `ActorRun` for the same reason: `call` is `start `waitForFinish`, so a `call` with a short `waitSecs` on a long-running Actor can return before the run finishes. +If the id instead keeps returning 404 (e.g. a typo, or it was never created) `waitForFinish` does +not poll it for the whole wait budget: a 404 is tolerated for up to 3 seconds (database-replica lag +right after a run/build is created), past which it throws that `NotFoundError`. + ### Streamed log redirection `getStreamedLog()` completes with a `StreamedLog`: a helper that follows the run's live raw log in a @@ -68,6 +72,12 @@ Actor name and run id (looked up automatically). `StreamedLogOptions` customizes - `fromStart(boolean)` — when `false`, skip log lines produced before redirection started (default `true`). +A run that has not logged anything yet (e.g. still queued for resources, see +[`waitForResources`](actors.md#actorclient)) serves its log as a stream that ends at once; +`StreamedLog` reopens it, with capped exponential backoff, until it actually delivers content, +rather than silently giving up. If `stop()`/`close()` is called before that happens, it fetches the +log in one plain request instead, so whatever was logged by then is still delivered. + ```java RunClient runClient = client.run("RUN_ID"); List collected = new ArrayList<>(); diff --git a/docs/tasks.md b/docs/tasks.md index fbef5e6..dba8e84 100644 --- a/docs/tasks.md +++ b/docs/tasks.md @@ -36,7 +36,8 @@ Task task = client.tasks().create(Map.of( `TaskStartOptions` mirrors `ActorStartOptions` but omits the Actor-only `contentType` and `forcePermissionLevel`: `build`, `memoryMbytes`, `timeoutSecs`, `waitForFinish`, `maxItems`, -`maxTotalChargeUsd`, `restartOnError`, `webhooks`. `TaskCallOptions` mirrors `TaskStartOptions` in +`maxTotalChargeUsd`, `restartOnError`, `webhooks`, `waitForResources` (see +[`ActorStartOptions`](actors.md#actorclient)). `TaskCallOptions` mirrors `TaskStartOptions` in turn, but additionally omits `waitForFinish`: that field asks the API to hold the HTTP response open server-side while the run finishes, which is redundant with (and wastes a request slot next to) `call`'s own client-side `waitSecs` polling. diff --git a/pom.xml b/pom.xml index 9c1829d..e73f266 100644 --- a/pom.xml +++ b/pom.xml @@ -6,7 +6,7 @@ com.apify apify-client - 0.7.0 + 0.8.0 jar Apify Java Client diff --git a/src/main/java/com/apify/client/Version.java b/src/main/java/com/apify/client/Version.java index c637bc4..6e09060 100644 --- a/src/main/java/com/apify/client/Version.java +++ b/src/main/java/com/apify/client/Version.java @@ -13,13 +13,13 @@ public final class Version { * The semantic version of this client library (see SemVer). * Changes to the public interface other than additive ones are considered breaking changes. */ - public static final String CLIENT_VERSION = "0.7.0"; + public static final String CLIENT_VERSION = "0.8.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 static final String API_SPEC_VERSION = "v2-2026-10-01T153946Z"; + public static final String API_SPEC_VERSION = "v2-2026-10-08T073338Z"; private Version() {} } diff --git a/src/main/java/com/apify/client/actor/ActorCallOptions.java b/src/main/java/com/apify/client/actor/ActorCallOptions.java index 91b7839..bc8d095 100644 --- a/src/main/java/com/apify/client/actor/ActorCallOptions.java +++ b/src/main/java/com/apify/client/actor/ActorCallOptions.java @@ -85,6 +85,18 @@ public ActorCallOptions webhooks(List webhooks) { return this; } + /** See {@link ActorStartOptions#waitForResources(boolean)}. */ + public ActorCallOptions waitForResources(boolean waitForResources) { + startOptions.waitForResources(waitForResources); + return this; + } + + /** See {@link ActorStartOptions#waitForResources(Long)}. */ + public ActorCallOptions waitForResources(Long maxSecs) { + startOptions.waitForResources(maxSecs); + return this; + } + /** * Disables the default log streaming, matching the reference client's {@code log: null}. {@code * call} then only starts the run and polls for completion, as if built from {@link diff --git a/src/main/java/com/apify/client/actor/ActorClient.java b/src/main/java/com/apify/client/actor/ActorClient.java index 5f601fb..ed427ff 100644 --- a/src/main/java/com/apify/client/actor/ActorClient.java +++ b/src/main/java/com/apify/client/actor/ActorClient.java @@ -65,7 +65,12 @@ public CompletableFuture delete() { * input} is any JSON-serializable value (or {@code null} for no input). */ public CompletableFuture start(Object input, ActorStartOptions options) { - return RunStartSupport.start(ctx, input, options::apply, options.contentTypeOrDefault()); + return RunStartSupport.start( + ctx, + input, + options::apply, + options.contentTypeOrDefault(), + options.waitForResourcesSecsValue()); } /** @@ -78,7 +83,13 @@ public CompletableFuture start(Object input, ActorStartOptions options */ public CompletableFuture call(Object input, ActorStartOptions options, Long waitSecs) { return RunStartSupport.call( - root, ctx, input, options::apply, options.contentTypeOrDefault(), waitSecs); + root, + ctx, + input, + options::apply, + options.contentTypeOrDefault(), + waitSecs, + options.waitForResourcesSecsValue()); } /** @@ -104,7 +115,8 @@ public CompletableFuture call(Object input, ActorCallOptions options, startOptions.contentTypeOrDefault(), opts.logStreamingEnabledValue(), opts.logOptionsValue(), - waitSecs); + waitSecs, + startOptions.waitForResourcesSecsValue()); } /** Validates the given input against the Actor's default-build input schema. */ diff --git a/src/main/java/com/apify/client/actor/ActorStartOptions.java b/src/main/java/com/apify/client/actor/ActorStartOptions.java index 2724028..9f2680a 100644 --- a/src/main/java/com/apify/client/actor/ActorStartOptions.java +++ b/src/main/java/com/apify/client/actor/ActorStartOptions.java @@ -10,10 +10,18 @@ * Configures starting an Actor run ({@link ActorClient#start}/{@link ActorClient#call}). All fields * are optional. * - *

{@link #apply}/{@link #contentTypeOrDefault} are package-private on purpose — see {@link - * com.apify.client.internal.RunStartSupport}'s class javadoc for why. + *

{@link #apply}/{@link #contentTypeOrDefault}/{@link #waitForResourcesSecsValue} are + * package-private on purpose — see {@link com.apify.client.internal.RunStartSupport}'s class + * javadoc for why. */ public final class ActorStartOptions { + + /** + * Sentinel for {@link #waitForResources(Long)} (and what {@link #waitForResources(boolean)} + * passes for {@code true}): retry starting the run until it succeeds, with no time bound. + */ + public static final long WAIT_FOR_RESOURCES_INDEFINITELY = -1L; + private String build; private Long memoryMbytes; private Long timeoutSecs; @@ -24,6 +32,7 @@ public final class ActorStartOptions { private Boolean restartOnError; private String forcePermissionLevel; private List webhooks; + private Long waitForResourcesSecs; /** The tag or number of the build to run (e.g. {@code "latest"}, {@code "0.1.2"}). */ public ActorStartOptions build(String build) { @@ -91,12 +100,39 @@ public ActorStartOptions webhooks(List webhooks) { return this; } + /** + * Retries the start while the account lacks the memory or a concurrent-run slot for the run (the + * API rejects it with {@code actor-memory-limit-exceeded} or {@code + * concurrent-runs-limit-exceeded}, both of which clear as other runs or builds finish), waiting + * 10 seconds between attempts. {@code true} retries until the run starts ({@link + * #WAIT_FOR_RESOURCES_INDEFINITELY}); {@code false} (the default) does not retry — a + * resource-limit error is thrown immediately, like any other error. + */ + public ActorStartOptions waitForResources(boolean waitForResources) { + this.waitForResourcesSecs = waitForResources ? WAIT_FOR_RESOURCES_INDEFINITELY : null; + return this; + } + + /** + * As {@link #waitForResources(boolean)}, bounding the retrying to at most {@code maxSecs} seconds + * - once exceeded, the last resource-limit error is thrown. {@code null} disables retrying (the + * default). + */ + public ActorStartOptions waitForResources(Long maxSecs) { + this.waitForResourcesSecs = maxSecs; + return this; + } + String contentTypeOrDefault() { return (contentType != null && !contentType.isEmpty()) ? contentType : ResourceContext.CONTENT_TYPE_JSON; } + Long waitForResourcesSecsValue() { + return waitForResourcesSecs; + } + void apply(QueryParams q) { q.addString("build", build) .addLong("memory", memoryMbytes) diff --git a/src/main/java/com/apify/client/internal/ResourceContext.java b/src/main/java/com/apify/client/internal/ResourceContext.java index b49d4ad..06b2196 100644 --- a/src/main/java/com/apify/client/internal/ResourceContext.java +++ b/src/main/java/com/apify/client/internal/ResourceContext.java @@ -52,6 +52,15 @@ public final class ResourceContext { private static final int NOT_FOUND = 404; + /** + * Upper bound on how long {@link #waitForFinish} keeps polling a resource that keeps answering + * 404, before giving up and throwing that {@link ApifyApiException}. A resource that was just + * created can 404 briefly (database-replica lag), which this bound rides out; anything longer + * means the id most likely never existed, so polling for the rest of a long (or indefinite) wait + * budget would only delay reporting that. Mirrors the reference client. + */ + private static final long MAX_WAIT_FOR_MISSING_RESOURCE_MILLIS = 3000; + /** API error type for a resource that does not exist. */ private static final String ERROR_TYPE_RECORD_NOT_FOUND = "record-not-found"; @@ -511,7 +520,11 @@ public Long clampServerWait(Long waitForFinishSecs) { * *

The budget is a pure time bound, evaluated independently of whether the resource is * currently present: a just-started run/build can transiently return 404 (database-replica lag), - * which is treated as "not yet available". + * which is treated as "not yet available" - but only for up to {@link + * #MAX_WAIT_FOR_MISSING_RESOURCE_MILLIS}. A resource that keeps 404-ing past that almost + * certainly never existed (e.g. a typo'd id), so continuing to poll it for the rest of a long or + * indefinite wait budget would only delay reporting that; past the bound, the 404 is rethrown + * instead. * *

Each poll is chained via {@link CompletableFuture#thenCompose}, and the inter-poll delay is * a scheduled timer (see {@link Async}), so waiting never blocks a thread. @@ -528,13 +541,30 @@ public CompletableFuture waitForFinish( long budgetMillis = effectiveWaitSecs * 1000L; long start = System.currentTimeMillis(); long serverWaitCap = serverWaitCapSecs(); - return pollOnce(dataType, isTerminal, resourceName, start, budgetMillis, serverWaitCap, null); + return pollOnce( + dataType, isTerminal, resourceName, start, budgetMillis, serverWaitCap, null, null); + } + + /** The outcome of a single {@code waitForFinish} poll: either a fetched value, or a 404. */ + private record PollOutcome(T value, ApifyApiException notFoundError) { + static PollOutcome of(T value) { + return new PollOutcome<>(value, null); + } + + static PollOutcome notFound(ApifyApiException error) { + return new PollOutcome<>(null, error); + } + + boolean isNotFound() { + return notFoundError != null; + } } /** * One poll of the {@code waitForFinish} loop. {@code lastSeen} carries the most recently observed - * resource across recursive calls (there is no shared mutable state between polls, so each - * recursive step gets everything it needs as a parameter). + * resource and {@code notFoundSinceMillis} the instant 404s started, both across recursive calls + * (there is no shared mutable state between polls, so each recursive step gets everything it + * needs as a parameter). */ private CompletableFuture pollOnce( JavaType dataType, @@ -543,7 +573,8 @@ private CompletableFuture pollOnce( long start, long budgetMillis, long serverWaitCap, - T lastSeen) { + T lastSeen, + Long notFoundSinceMillis) { long elapsed = System.currentTimeMillis() - start; long remainingSecs = (budgetMillis - elapsed) / 1000L; long requestSecs = @@ -552,25 +583,46 @@ private CompletableFuture pollOnce( QueryParams params = new QueryParams(); params.addLong("waitForFinish", requestSecs); - return this.getResource("", params, dataType) + return this.getResourceRequired("", params, dataType) + .handle( + (value, error) -> { + if (error == null) { + return PollOutcome.of(value); + } + Throwable cause = HttpClientCore.unwrapCompletion(error); + if (cause instanceof ApifyApiException apiError && isNotFound(apiError)) { + return PollOutcome.notFound(apiError); + } + throw asRuntimeException(cause); + }) .thenCompose( - res -> { - T seen = res.orElse(lastSeen); - if (res.isPresent() && isTerminal.test(res.get())) { - return CompletableFuture.completedFuture(res.get()); + outcome -> { + long now = System.currentTimeMillis(); + if (!outcome.isNotFound() && isTerminal.test(outcome.value())) { + return CompletableFuture.completedFuture(outcome.value()); } - if (System.currentTimeMillis() - start >= budgetMillis) { + T seen = outcome.isNotFound() ? lastSeen : outcome.value(); + Long notFoundSince = + outcome.isNotFound() + ? (notFoundSinceMillis != null ? notFoundSinceMillis : now) + : null; + boolean missingTooLong = + notFoundSince != null + && now - notFoundSince >= MAX_WAIT_FOR_MISSING_RESOURCE_MILLIS; + if (missingTooLong || now - start >= budgetMillis) { if (seen != null) { return CompletableFuture.completedFuture(seen); } CompletableFuture failed = new CompletableFuture<>(); failed.completeExceptionally( - new IllegalStateException( - "waiting for " - + resourceName - + " to finish failed: cannot fetch " - + resourceName - + " details from the server")); + outcome.notFoundError() != null + ? outcome.notFoundError() + : new IllegalStateException( + "waiting for " + + resourceName + + " to finish failed: cannot fetch " + + resourceName + + " details from the server")); return failed; } return Async.delay(WAIT_POLL_INTERVAL) @@ -583,7 +635,8 @@ private CompletableFuture pollOnce( start, budgetMillis, serverWaitCap, - seen)); + seen, + notFoundSince)); }); } diff --git a/src/main/java/com/apify/client/internal/RunStartSupport.java b/src/main/java/com/apify/client/internal/RunStartSupport.java index bc8b8a4..434eee4 100644 --- a/src/main/java/com/apify/client/internal/RunStartSupport.java +++ b/src/main/java/com/apify/client/internal/RunStartSupport.java @@ -1,12 +1,18 @@ package com.apify.client.internal; import com.apify.client.ApifyClient; +import com.apify.client.actor.ActorStartOptions; +import com.apify.client.http.ApifyApiException; import com.apify.client.log.StreamedLog; import com.apify.client.log.StreamedLogOptions; import com.apify.client.run.ActorRun; import com.apify.client.run.RunClient; +import java.time.Duration; +import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.function.Consumer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * Shared implementation of "start a run" and "call: start, then wait, optionally streaming the log" @@ -15,17 +21,30 @@ *

{@code ActorStartOptions}/{@code TaskStartOptions}/{@code ActorCallOptions}/{@code * TaskCallOptions} (in {@code com.apify.client.actor}/{@code com.apify.client.task}) are themselves * public classes — only the handful of accessor methods this class needs from them ({@code apply}, - * {@code contentTypeOrDefault}, {@code toStartOptions}, {@code logStreamingEnabledValue}, {@code - * logOptionsValue}) are package-private, with no public interface or shared supertype exposing - * them. {@code ActorClient}/{@code TaskClient} (same package as their respective options types) - * read those values directly and pass them here as plain values and method references (e.g. {@code - * options::apply} bound as a {@code Consumer}). That keeps this one implementation - * backing both Actor and task calls (instead of each duplicating the same ~90 lines) without - * forcing any of the options builders' internal plumbing — including the internal {@link - * QueryParams} type — into their public API. + * {@code contentTypeOrDefault}, {@code waitForResourcesSecsValue}, {@code toStartOptions}, {@code + * logStreamingEnabledValue}, {@code logOptionsValue}) are package-private, with no public interface + * or shared supertype exposing them. {@code ActorClient}/{@code TaskClient} (same package as their + * respective options types) read those values directly and pass them here as plain values and + * method references (e.g. {@code options::apply} bound as a {@code Consumer}). That + * keeps this one implementation backing both Actor and task calls (instead of each duplicating the + * same ~90 lines) without forcing any of the options builders' internal plumbing — including the + * internal {@link QueryParams} type — into their public API. */ public final class RunStartSupport { + private static final Logger LOG = LoggerFactory.getLogger(RunStartSupport.class); + + /** + * API error types the run-start endpoint returns while the account lacks the resources for the + * run (free memory or a concurrent-run slot). Both clear as other runs or builds finish, which is + * what makes retrying worthwhile. + */ + private static final Set RESOURCE_LIMIT_ERROR_TYPES = + Set.of("actor-memory-limit-exceeded", "concurrent-runs-limit-exceeded"); + + /** Delay between two attempts to start a run rejected for lack of resources. */ + private static final Duration WAIT_FOR_RESOURCES_RETRY_INTERVAL = Duration.ofSeconds(10); + private RunStartSupport() {} /** @@ -34,19 +53,79 @@ private RunStartSupport() {} * for no input. {@code applyParams} applies every start-option query parameter (build, memory, * timeout, ...) the caller's options type configures; {@code contentType} is the input body's * content type. + * + * @param waitForResourcesSecs mirrors {@link ActorStartOptions#waitForResources(Long)}: {@code + * null} does not retry a start rejected for lack of resources (the default); {@link + * ActorStartOptions#WAIT_FOR_RESOURCES_INDEFINITELY} retries until it succeeds; any other + * value bounds the retrying to that many seconds, after which the last such error is thrown */ public static CompletableFuture start( - ResourceContext ctx, Object input, Consumer applyParams, String contentType) { + ResourceContext ctx, + Object input, + Consumer applyParams, + String contentType, + Long waitForResourcesSecs) { QueryParams params = new QueryParams(); applyParams.accept(params); byte[] body = input == null ? null : Json.toBytes(input); - return ctx.postWithBody("runs", params, body, contentType, ActorRun.class); + if (waitForResourcesSecs == null) { + return ctx.postWithBody("runs", params, body, contentType, ActorRun.class); + } + long deadlineMillis = + waitForResourcesSecs == ActorStartOptions.WAIT_FOR_RESOURCES_INDEFINITELY + ? Long.MAX_VALUE + : System.currentTimeMillis() + Math.max(0, waitForResourcesSecs) * 1000L; + return startRetryingForResources(ctx, params, body, contentType, deadlineMillis); + } + + /** + * One attempt of the {@code waitForResources} retry loop; recurses (via a non-blocking delay) + * while the start keeps failing with a {@link #RESOURCE_LIMIT_ERROR_TYPES resource-limit error} + * and the deadline has not passed. Any other error, or a resource-limit error once the deadline + * has passed, is rethrown as-is. + */ + private static CompletableFuture startRetryingForResources( + ResourceContext ctx, + QueryParams params, + byte[] body, + String contentType, + long deadlineMillis) { + return ctx.postWithBody("runs", params, body, contentType, ActorRun.class) + .handle( + (run, error) -> { + if (error == null) { + return CompletableFuture.completedFuture(run); + } + Throwable cause = HttpClientCore.unwrapCompletion(error); + if (!(cause instanceof ApifyApiException apiError) + || !RESOURCE_LIMIT_ERROR_TYPES.contains(apiError.getType())) { + return CompletableFuture.failedFuture(cause); + } + long remainingMillis = deadlineMillis - System.currentTimeMillis(); + if (remainingMillis <= 0) { + return CompletableFuture.failedFuture(cause); + } + Duration delay = + Duration.ofMillis( + Math.min(WAIT_FOR_RESOURCES_RETRY_INTERVAL.toMillis(), remainingMillis)); + LOG.info( + "Not enough resources to start the run, retrying in {}s: {}", + delay.toSeconds(), + apiError.getMessage()); + return Async.delay(delay) + .thenCompose( + unused -> + startRetryingForResources( + ctx, params, body, contentType, deadlineMillis)); + }) + .thenCompose(future -> future); } /** * Starts a run and waits (non-blocking, polling) for it to finish, without log streaming. * * @param waitSecs bounds the wait; {@code null} waits indefinitely + * @param waitForResourcesSecs see {@link #start} */ public static CompletableFuture call( ApifyClient root, @@ -54,8 +133,9 @@ public static CompletableFuture call( Object input, Consumer applyParams, String contentType, - Long waitSecs) { - return start(ctx, input, applyParams, contentType) + Long waitSecs, + Long waitForResourcesSecs) { + return start(ctx, input, applyParams, contentType, waitForResourcesSecs) .thenCompose(run -> root.run(run.getId()).waitForFinish(waitSecs)); } @@ -72,6 +152,7 @@ public static CompletableFuture call( * @param logStreamingEnabled whether to stream the log for the duration of the wait * @param logOptions a custom log-streaming destination, or {@code null} for the default * @param waitSecs bounds the wait; {@code null} waits indefinitely + * @param waitForResourcesSecs see {@link #start} */ public static CompletableFuture callWithLogStreaming( ApifyClient root, @@ -81,8 +162,9 @@ public static CompletableFuture callWithLogStreaming( String contentType, boolean logStreamingEnabled, StreamedLogOptions logOptions, - Long waitSecs) { - return start(ctx, input, applyParams, contentType) + Long waitSecs, + Long waitForResourcesSecs) { + return start(ctx, input, applyParams, contentType, waitForResourcesSecs) .thenCompose( run -> { RunClient runClient = root.run(run.getId()); diff --git a/src/main/java/com/apify/client/log/StreamedLog.java b/src/main/java/com/apify/client/log/StreamedLog.java index 9188352..ec30cbb 100644 --- a/src/main/java/com/apify/client/log/StreamedLog.java +++ b/src/main/java/com/apify/client/log/StreamedLog.java @@ -2,8 +2,10 @@ import com.apify.client.http.ApifyApiException; import com.apify.client.internal.HttpClientCore; +import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; +import java.io.SequenceInputStream; import java.nio.charset.StandardCharsets; import java.time.Instant; import java.time.format.DateTimeParseException; @@ -55,6 +57,17 @@ public final class StreamedLog implements AutoCloseable { /** Read buffer size for pulling bytes off the live log stream. */ private static final int READ_BUFFER_BYTES = 8192; + /** + * Pause before the first reopen of a log stream that ended before the run had logged anything + * (the API serves that case as a stream that ends at once). Doubles on each further reopen, up to + * {@link #EMPTY_STREAM_MAX_RETRY_MILLIS}, which bounds the request rate while a run takes a long + * time to start producing output (e.g. while queued for resources). + */ + private static final long EMPTY_STREAM_RETRY_MILLIS = 500; + + /** Upper bound on the pause between reopens of an empty log stream. */ + private static final long EMPTY_STREAM_MAX_RETRY_MILLIS = 5_000; + private final LogClient logClient; private final Consumer destination; @@ -175,6 +188,18 @@ private void stopStreaming() { stopLogging = true; // Close the live stream so a blocked read returns promptly instead of waiting for more bytes. closeQuietly(activeStream); + // Wakes the reader if it is currently backing off before reopening a log stream that has not + // delivered anything yet (see awaitNonEmptyStream) - that wait is a plain Thread.sleep, not + // tied + // to the stream, so closing activeStream above does not interrupt it on its own. The interrupt + // is + // a pure wake signal here (stopLogging, checked right after, is the actual stop condition), not + // a + // cancellation request the reader needs to honor, so awaitNonEmptyStream clears rather than + // re-raises it - a lingering interrupt status would otherwise risk breaking the reader's next + // blocking I/O call. Callers have already checked a thread is running (see each's own + // contract). + streamingThread.interrupt(); // A thread must never join itself. The destination consumer runs on the streaming thread, so a // user calling stop()/close() from inside that consumer (e.g. "stop redirecting once I see line // X") would otherwise join() the current thread on itself and block forever - and because @@ -202,15 +227,41 @@ private void streamLog() { // (not a shared field) so that if a start() ever races a still-draining reader, the two readers // never mutate the same non-thread-safe buffer. StringBuilder pending = new StringBuilder(); - // The stream was opened in start() and stored in activeStream; own its lifecycle here via - // try-with-resources so the reader closes it when it exits (idempotent with stop()'s close, - // which may close it first to unblock a pending read). - try (InputStream stream = activeStream) { + InputStream stream; + try { + // The API serves the log of a run that has not logged anything yet as a stream that ends at + // once; reopen it until it actually delivers a byte, so redirection does not silently give up + // on a run that is still queued or starting. + stream = awaitNonEmptyStream(); + } catch (IOException e) { + if (!stopLogging) { + LOG.warn("Log redirection stopped due to error", e); + } + return; + } + if (stream == null) { + // Stopped before any stream ever delivered a byte: fetch whatever the log holds right now in + // one plain request instead of redirecting nothing at all. + try { + String logContent = logClient.get(new LogOptions().raw(true)).join().orElse(""); + if (!forwardingFailed) { + emitMessages(pending, logContent, true); + } + } catch (RuntimeException e) { + LOG.warn("Log redirection stopped due to error", e); + } + return; + } + // The stream was opened (possibly reopened by awaitNonEmptyStream) and stored in activeStream; + // own its lifecycle here via try-with-resources so the reader closes it when it exits + // (idempotent + // with stop()'s close, which may close it first to unblock a pending read). + try (InputStream liveStream = stream) { byte[] readBuffer = new byte[READ_BUFFER_BYTES]; int read; // Process each chunk before honouring a stop request, so a stop issued right after start // still redirects whatever the run has already produced (matching the reference client). - while ((read = stream.read(readBuffer)) != -1) { + while ((read = liveStream.read(readBuffer)) != -1) { byte[] combined = new byte[lineRemainder.length + read]; System.arraycopy(lineRemainder, 0, combined, 0, lineRemainder.length); System.arraycopy(readBuffer, 0, combined, lineRemainder.length, read); @@ -247,6 +298,80 @@ private void streamLog() { } } + /** + * Waits for the log stream to actually start delivering bytes, reopening it (pausing {@link + * #EMPTY_STREAM_RETRY_MILLIS}, doubled on each further attempt) while it keeps ending with none. + * Returns a stream with any bytes already consumed while probing it stitched back onto the front, + * as soon as one delivers at least one byte; or {@code null} if {@link #stop()}/{@link #close()} + * is requested before any stream ever delivers a byte. + */ + private InputStream awaitNonEmptyStream() throws IOException { + InputStream stream = activeStream; + long backoffMillis = EMPTY_STREAM_RETRY_MILLIS; + while (true) { + // Always attempt the read before consulting stopLogging - like the main loop below, which + // processes a chunk before honouring a stop request - so a stop issued right after start() + // still drains whatever is already buffered (e.g. a finite stream in tests) instead of + // discarding it. + byte[] probe = new byte[READ_BUFFER_BYTES]; + int read; + try { + read = stream.read(probe); + } catch (IOException e) { + // An explicit stop can close `stream` out from under this read; that is expected, not a + // genuine failure (mirrors the main loop's own IOException handling below). + if (stopLogging) { + return null; + } + throw e; + } + if (read != -1) { + return new SequenceInputStream(new ByteArrayInputStream(probe, 0, read), stream); + } + closeQuietly(stream); + if (stopLogging) { + return null; + } + sleepAsWakeableBackoff(backoffMillis); + backoffMillis = Math.min(backoffMillis * 2, EMPTY_STREAM_MAX_RETRY_MILLIS); + if (stopLogging) { + return null; + } + try { + stream = logClient.stream(new LogOptions().raw(true)).join(); + } catch (RuntimeException e) { + if (stopLogging) { + return null; + } + Throwable cause = HttpClientCore.unwrapCompletion(e); + if (cause instanceof IOException ioException) { + throw ioException; + } + throw new IOException("failed to reopen log stream", cause); + } + // Published under the monitor stop()/close() also use for their own activeStream read, so a + // concurrent stop cannot race this assignment: it either closes the old stream (and this + // method's own stopLogging check above then closes the new one too) or the new one. + synchronized (this) { + activeStream = stream; + } + } + } + + /** + * Sleeps for {@code millis}, waking early if interrupted and clearing (rather than re-raising) + * the interrupt status - {@link #stopStreaming()} uses interruption purely to wake this wait + * promptly; {@link #stopLogging}, checked by the caller right after, is the actual stop signal, + * so a lingering interrupt status would only risk breaking a later, unrelated blocking call. + */ + private static void sleepAsWakeableBackoff(long millis) { + try { + Thread.sleep(millis); + } catch (InterruptedException ignored) { + // Intentionally not restoring the interrupt status - see method javadoc. + } + } + /** * Appends {@code text} to the pending buffer, then forwards every complete message it contains. * When {@code flush} is {@code false} the last message is held back (it may still be growing); diff --git a/src/main/java/com/apify/client/task/TaskCallOptions.java b/src/main/java/com/apify/client/task/TaskCallOptions.java index 670ffb5..e7097ab 100644 --- a/src/main/java/com/apify/client/task/TaskCallOptions.java +++ b/src/main/java/com/apify/client/task/TaskCallOptions.java @@ -70,6 +70,18 @@ public TaskCallOptions webhooks(List webhooks) { return this; } + /** See {@link TaskStartOptions#waitForResources(boolean)}. */ + public TaskCallOptions waitForResources(boolean waitForResources) { + startOptions.waitForResources(waitForResources); + return this; + } + + /** See {@link TaskStartOptions#waitForResources(Long)}. */ + public TaskCallOptions waitForResources(Long maxSecs) { + startOptions.waitForResources(maxSecs); + return this; + } + /** * Disables the default log streaming, matching the reference client's {@code log: null}. {@code * call} then only starts the run and polls for completion, as if built from {@link diff --git a/src/main/java/com/apify/client/task/TaskClient.java b/src/main/java/com/apify/client/task/TaskClient.java index 8e05bdf..2b74c4f 100644 --- a/src/main/java/com/apify/client/task/TaskClient.java +++ b/src/main/java/com/apify/client/task/TaskClient.java @@ -78,7 +78,12 @@ public CompletableFuture unpublish() { * input} optionally overrides the task's stored input ({@code null} to use the stored input). */ public CompletableFuture start(Object input, TaskStartOptions options) { - return RunStartSupport.start(ctx, input, options::apply, options.contentTypeOrDefault()); + return RunStartSupport.start( + ctx, + input, + options::apply, + options.contentTypeOrDefault(), + options.waitForResourcesSecsValue()); } /** @@ -90,7 +95,13 @@ public CompletableFuture start(Object input, TaskStartOptions options) */ public CompletableFuture call(Object input, TaskStartOptions options, Long waitSecs) { return RunStartSupport.call( - root, ctx, input, options::apply, options.contentTypeOrDefault(), waitSecs); + root, + ctx, + input, + options::apply, + options.contentTypeOrDefault(), + waitSecs, + options.waitForResourcesSecsValue()); } /** @@ -115,7 +126,8 @@ public CompletableFuture call(Object input, TaskCallOptions options, L startOptions.contentTypeOrDefault(), opts.logStreamingEnabledValue(), opts.logOptionsValue(), - waitSecs); + waitSecs, + startOptions.waitForResourcesSecsValue()); } /** diff --git a/src/main/java/com/apify/client/task/TaskStartOptions.java b/src/main/java/com/apify/client/task/TaskStartOptions.java index 6d06253..25d6bd4 100644 --- a/src/main/java/com/apify/client/task/TaskStartOptions.java +++ b/src/main/java/com/apify/client/task/TaskStartOptions.java @@ -12,8 +12,9 @@ * accept (the Actor-only {@code contentType} and {@code forcePermissionLevel}), matching the * reference client. * - *

{@link #apply}/{@link #contentTypeOrDefault} are package-private on purpose — see {@link - * com.apify.client.internal.RunStartSupport}'s class javadoc for why. + *

{@link #apply}/{@link #contentTypeOrDefault}/{@link #waitForResourcesSecsValue} are + * package-private on purpose — see {@link com.apify.client.internal.RunStartSupport}'s class + * javadoc for why. */ public final class TaskStartOptions { private String build; @@ -24,6 +25,7 @@ public final class TaskStartOptions { private Double maxTotalChargeUsd; private Boolean restartOnError; private List webhooks; + private Long waitForResourcesSecs; /** The tag or number of the build to run (e.g. {@code "latest"}, {@code "0.1.2"}). */ public TaskStartOptions build(String build) { @@ -73,6 +75,34 @@ public TaskStartOptions webhooks(List webhooks) { return this; } + /** + * Retries the start while the account lacks the memory or a concurrent-run slot for the run (the + * API rejects it with {@code actor-memory-limit-exceeded} or {@code + * concurrent-runs-limit-exceeded}, both of which clear as other runs or builds finish), waiting + * 10 seconds between attempts. {@code true} retries until the run starts ({@link + * ActorStartOptions#WAIT_FOR_RESOURCES_INDEFINITELY}); {@code false} (the default) does not retry + * — a resource-limit error is thrown immediately, like any other error. + */ + public TaskStartOptions waitForResources(boolean waitForResources) { + this.waitForResourcesSecs = + waitForResources ? ActorStartOptions.WAIT_FOR_RESOURCES_INDEFINITELY : null; + return this; + } + + /** + * As {@link #waitForResources(boolean)}, bounding the retrying to at most {@code maxSecs} seconds + * - once exceeded, the last resource-limit error is thrown. {@code null} disables retrying (the + * default). + */ + public TaskStartOptions waitForResources(Long maxSecs) { + this.waitForResourcesSecs = maxSecs; + return this; + } + + Long waitForResourcesSecsValue() { + return waitForResourcesSecs; + } + void apply(QueryParams q) { q.addString("build", build) .addLong("memory", memoryMbytes) diff --git a/src/main/java/com/apify/client/user/UserProfile.java b/src/main/java/com/apify/client/user/UserProfile.java index aa78094..857eb0d 100644 --- a/src/main/java/com/apify/client/user/UserProfile.java +++ b/src/main/java/com/apify/client/user/UserProfile.java @@ -5,6 +5,7 @@ /** The public profile of a {@link User}. */ public final class UserProfile extends ApifyResource { private String bio; + private String readme; private String name; private String pictureUrl; private String githubUsername; @@ -16,6 +17,11 @@ public String getBio() { return bio; } + /** Markdown README shown on the user's public profile page, if set. */ + public String getReadme() { + return readme; + } + /** The user's display name, if set. */ public String getName() { return name; diff --git a/src/test/java/com/apify/client/ClientBehaviourRegressionTest.java b/src/test/java/com/apify/client/ClientBehaviourRegressionTest.java index 0ffaef7..ffbd51c 100644 --- a/src/test/java/com/apify/client/ClientBehaviourRegressionTest.java +++ b/src/test/java/com/apify/client/ClientBehaviourRegressionTest.java @@ -7,6 +7,7 @@ import com.apify.client.actor.Actor; import com.apify.client.actor.ActorListOptions; +import com.apify.client.actor.ActorStartOptions; import com.apify.client.actor.ActorVersion; import com.apify.client.dataset.DatasetDownloadOptions; import com.apify.client.dataset.DatasetListItemsOptions; @@ -696,13 +697,107 @@ void waitForFinishReturnsWhenResourceAppearsAfter404() { @Test void waitForFinishThrowsWhenResourceNeverAppears() { // If the resource is never fetchable within the budget, the wait must fail loudly rather than - // return a phantom result. + // return a phantom result - with the actual 404 the server gave, matching the reference client, + // not a generic wrapped error. MockTransport backend = MockTransport.ofConstant( 404, "{\"error\":{\"type\":\"record-not-found\",\"message\":\"missing\"}}"); + NotFoundError ex = + assertThrows( + NotFoundError.class, + () -> TestAsync.await(client(backend).run("r1").waitForFinish(0L))); + assertEquals(404, ex.getStatusCode()); + } + + @Test + void waitForFinishGivesUpOnSustained404BeforeTheFullWaitBudget() { + // A run/build id that never existed 404s forever. waitForFinish must not poll it for the whole + // (here effectively indefinite, waitSecs omitted) wait budget - it gives up once the resource + // has + // been missing continuously for a few seconds, bounding how long a bad id can hang the caller. + MockTransport backend = + MockTransport.ofConstant( + 404, "{\"error\":{\"type\":\"record-not-found\",\"message\":\"missing\"}}"); + long startedAt = System.currentTimeMillis(); assertThrows( - IllegalStateException.class, - () -> TestAsync.await(client(backend).run("r1").waitForFinish(0L))); + NotFoundError.class, () -> TestAsync.await(client(backend).run("r1").waitForFinish(null))); + long elapsedMillis = System.currentTimeMillis() - startedAt; + assertTrue( + elapsedMillis < 10_000, + "a sustained 404 must give up within a few seconds, not run for the full wait budget, took " + + elapsedMillis + + "ms"); + } + + @Test + void startDoesNotRetryAResourceLimitErrorByDefault() { + // waitForResources is opt-in; without it, a start rejected for lack of resources is thrown + // immediately like any other error, with no retry. + MockTransport backend = + MockTransport.ofConstant( + 400, + "{\"error\":{\"type\":\"actor-memory-limit-exceeded\",\"message\":\"not enough memory\"}}"); + assertThrows( + ApifyApiException.class, + () -> TestAsync.await(client(backend).actor("a1").start(null, new ActorStartOptions()))); + assertEquals(1, backend.calls, "no retry without waitForResources"); + } + + @Test + void startDoesNotRetryANonResourceLimitErrorEvenWithWaitForResources() { + // Only the two resource-limit error types are retried; any other error (even with + // waitForResources enabled) is thrown on the first attempt. + MockTransport backend = + MockTransport.ofConstant( + 400, "{\"error\":{\"type\":\"invalid-input\",\"message\":\"bad input\"}}"); + assertThrows( + ApifyApiException.class, + () -> + TestAsync.await( + client(backend) + .actor("a1") + .start(null, new ActorStartOptions().waitForResources(true)))); + assertEquals(1, backend.calls, "a non-resource-limit error must not be retried"); + } + + @Test + void startRetriesOnResourceLimitErrorUntilItStarts() { + // The first attempt is rejected for lack of resources; the retry (bounded here to keep the test + // fast) succeeds once resources free up, and the caller sees the created run, not the error. + MockTransport backend = + new MockTransport( + List.of( + MockTransport.ok( + 400, + "{\"error\":{\"type\":\"concurrent-runs-limit-exceeded\",\"message\":\"too many" + + " runs\"}}"), + MockTransport.ok(200, "{\"data\":{\"id\":\"run1\",\"status\":\"READY\"}}"))); + ActorRun run = + TestAsync.await( + client(backend).actor("a1").start(null, new ActorStartOptions().waitForResources(3L))); + assertEquals("run1", run.getId()); + assertEquals(2, backend.calls, "one rejected attempt, then the retry that succeeds"); + } + + @Test + void startGivesUpOnResourceLimitErrorOnceTheBoundIsExceeded() { + // Resources never free up within the caller's bound, so the last resource-limit error is + // eventually thrown rather than retrying forever. + MockTransport backend = + MockTransport.ofConstant( + 400, + "{\"error\":{\"type\":\"actor-memory-limit-exceeded\",\"message\":\"not enough" + + " memory\"}}"); + ApifyApiException ex = + assertThrows( + ApifyApiException.class, + () -> + TestAsync.await( + client(backend) + .actor("a1") + .start(null, new ActorStartOptions().waitForResources(1L)))); + assertEquals("actor-memory-limit-exceeded", ex.getType()); + assertTrue(backend.calls >= 2, "must have retried at least once before giving up"); } @Test diff --git a/src/test/java/com/apify/client/MockTransport.java b/src/test/java/com/apify/client/MockTransport.java index 162779b..e013f8a 100644 --- a/src/test/java/com/apify/client/MockTransport.java +++ b/src/test/java/com/apify/client/MockTransport.java @@ -57,16 +57,26 @@ static final class Scripted { final List bodies = new ArrayList<>(); final List urls = new ArrayList<>(); + /** One scripted response for {@link #sendStreamingAsync}: a status + ready-made body stream. */ + private static final class StreamEntry { + final int status; + final InputStream body; + + StreamEntry(int status, InputStream body) { + this.status = status; + this.body = body; + } + } + /** - * Optional scripted response for {@link #sendStreamingResponse}; defaults to the first {@code - * send} entry. + * Queue of scripted {@link #sendStreamingAsync} responses, consumed one per call (the last entry + * repeats once exhausted, like {@link #responses}). Lets a test script a sequence - e.g. an empty + * stream followed by one with content, to exercise {@code StreamedLog}'s reopen-on-empty-stream + * behavior - instead of only ever the same single response. */ - private Scripted streamResponse; - - /** Optional custom stream body (e.g. a blocking stream that simulates a live, open log). */ - private InputStream streamBody; + private final List streamQueue = new ArrayList<>(); - private int streamBodyStatus; + private int streamCallIndex = 0; MockTransport(List responses) { this.responses = responses; @@ -122,22 +132,24 @@ public synchronized CompletableFuture> sendAsync(HttpReques new FakeResponse(request.uri(), r.status, r.body, r.headers)); } - /** Scripts the next {@link #sendStreamingResponse} call to return the given status and body. */ - void scriptStream(int status, String body) { - this.streamResponse = new Scripted(status, body, null); - this.streamBody = null; + /** + * Appends a scripted {@link #sendStreamingAsync} response (status + in-memory body). A test that + * calls this once (the common case) gets that same response for every streaming call, since the + * single queued entry repeats; calling it several times scripts a sequence consumed in order. + */ + synchronized void scriptStream(int status, String body) { + streamQueue.add( + new StreamEntry( + status, new java.io.ByteArrayInputStream(body.getBytes(StandardCharsets.UTF_8)))); } /** - * Scripts the next {@link #sendStreamingResponse} call to return the given status and an - * arbitrary, possibly blocking, {@link InputStream}. Unlike {@link #scriptStream(int, String)}, - * the body is not a finite in-memory buffer, so {@code read()} does not naturally reach + * As {@link #scriptStream(int, String)}, with an arbitrary, possibly blocking, {@link + * InputStream} instead of a finite in-memory buffer - so {@code read()} does not naturally reach * end-of-stream; this models a live log that only ends when the stream is closed. */ - void scriptStream(int status, InputStream body) { - this.streamBodyStatus = status; - this.streamBody = body; - this.streamResponse = null; + synchronized void scriptStream(int status, InputStream body) { + streamQueue.add(new StreamEntry(status, body)); } @Override @@ -146,13 +158,16 @@ public synchronized CompletableFuture> sendStreamingAs calls++; lastUrl = request.uri().toString(); lastMethod = request.method(); - if (streamBody != null) { + if (streamQueue.isEmpty()) { + Scripted r = responses.get(0); return CompletableFuture.completedFuture( - new FakeStreamResponse(request.uri(), streamBodyStatus, streamBody)); + new FakeStreamResponse( + request.uri(), r.status, new java.io.ByteArrayInputStream(r.body))); } - Scripted r = streamResponse != null ? streamResponse : responses.get(0); + int idx = Math.min(streamCallIndex++, streamQueue.size() - 1); + StreamEntry entry = streamQueue.get(idx); return CompletableFuture.completedFuture( - new FakeStreamResponse(request.uri(), r.status, new java.io.ByteArrayInputStream(r.body))); + new FakeStreamResponse(request.uri(), entry.status, entry.body)); } private static byte[] readBody(HttpRequest request) { diff --git a/src/test/java/com/apify/client/StreamedLogTest.java b/src/test/java/com/apify/client/StreamedLogTest.java index f41cbc8..e03aaa4 100644 --- a/src/test/java/com/apify/client/StreamedLogTest.java +++ b/src/test/java/com/apify/client/StreamedLogTest.java @@ -397,6 +397,46 @@ void closeFromInsideConsumerDoesNotDeadlock() { "self-close from inside the consumer must not deadlock and must flush the retained message"); } + @Test + void reopensEmptyStreamUntilItDeliversContent() { + // The API serves the log of a run that has not logged anything yet as a stream that ends at + // once. The first open is scripted empty (as if the run were still queued/starting); the + // redirector must reopen rather than giving up, and the second open's content must still be + // delivered. + MockTransport backend = MockTransport.ofConstant(200, ""); + backend.scriptStream(200, ""); + backend.scriptStream(200, "2999-01-01T00:00:00.000Z hello\n"); + List collected = new CopyOnWriteArrayList<>(); + StreamedLog streamedLog = + client(backend) + .run("run123") + .getStreamedLog(new StreamedLogOptions().toLog(collected::add)) + .join(); + streamedLog.start().join(); + pollUntil(REDIRECT_POLL_ATTEMPTS, REDIRECT_POLL_MILLIS, () -> !collected.isEmpty()); + streamedLog.stop(); + assertEquals(List.of("2999-01-01T00:00:00.000Z hello"), collected); + } + + @Test + void stopBeforeAnyBytesStreamedFallsBackToOneShotFetch() { + // If stop() is requested before any stream ever delivers a byte, redirection must not simply + // end + // with nothing: it fetches the log in one plain request instead, so whatever was logged by then + // is still delivered. + MockTransport backend = MockTransport.ofConstant(200, "fallback content"); + backend.scriptStream(200, ""); // every streamed open stays empty - the run never logs anything + List collected = new CopyOnWriteArrayList<>(); + StreamedLog streamedLog = + client(backend) + .run("run123") + .getStreamedLog(new StreamedLogOptions().toLog(collected::add)) + .join(); + streamedLog.start().join(); + streamedLog.stop(); + assertEquals(List.of("fallback content"), collected); + } + @Test void defaultPrefixLookupFailureStillCreatesHelper() { // The no-arg getStreamedLog() builds a cosmetic per-run prefix by GETting the run (and its From f0a8714a8bfdf5f4b17895d60fb5df7710f6c9bf Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 12 Oct 2026 08:38:58 +0000 Subject: [PATCH 2/3] fix: address review feedback on the resource-retry/404-bound/log-reopen update - README: fix a stray old Version.API_SPEC_VERSION mention left over from the version bump - StreamedLog.stop()/close(): restore the reference client's ~1s grace window before force-closing the live stream, so a message already in flight (e.g. the run's last line) is not cut off - the earlier immediate-close only unblocked an already-blocked read, it did not give a pending one a chance to arrive - StreamedLog: fix a stop()/close() vs. reader-thread deadlock risk (both are `synchronized` methods that may block in `join()` while holding the monitor a `synchronized` activeStream publish from the reader thread would then need) by publishing the reopened stream as a plain volatile write instead, and explicitly close it too if a stop raced in while it was in flight - ResourceContext.waitForFinish: tolerate a 200 response with a null `data` body as "not yet available" again (it stopped doing so when the 404-bound fix swapped in a stricter accessor), instead of a NullPointerException - RunStartSupport: clamp waitForResourcesSecs before computing its deadline, so a value near Long.MAX_VALUE cannot overflow it --- README.md | 2 +- .../client/internal/ResourceContext.java | 34 ++++---- .../client/internal/RunStartSupport.java | 11 ++- .../com/apify/client/log/StreamedLog.java | 84 +++++++++++++------ .../client/ClientBehaviourRegressionTest.java | 17 ++++ .../com/apify/client/StreamedLogTest.java | 27 ++++++ 6 files changed, 132 insertions(+), 43 deletions(-) diff --git a/README.md b/README.md index 63395a6..a8265fb 100644 --- a/README.md +++ b/README.md @@ -320,7 +320,7 @@ constants: - `Version.CLIENT_VERSION` — the semantic version of this client (`0.8.0`). - `Version.API_SPEC_VERSION` — the version of the [Apify OpenAPI specification](https://docs.apify.com/api/openapi.json) (its `info.version` field) that this client's endpoints, parameters and models were last generated - and checked against (`v2-2026-10-01T153946Z`). It is a snapshot, not a live compatibility + and checked against (`v2-2026-10-08T073338Z`). It is a snapshot, not a live compatibility guarantee: the client keeps working against newer, backward-compatible spec revisions, but a feature added to the API after this snapshot has no corresponding method here yet. diff --git a/src/main/java/com/apify/client/internal/ResourceContext.java b/src/main/java/com/apify/client/internal/ResourceContext.java index 06b2196..f6a8838 100644 --- a/src/main/java/com/apify/client/internal/ResourceContext.java +++ b/src/main/java/com/apify/client/internal/ResourceContext.java @@ -545,7 +545,12 @@ public CompletableFuture waitForFinish( dataType, isTerminal, resourceName, start, budgetMillis, serverWaitCap, null, null); } - /** The outcome of a single {@code waitForFinish} poll: either a fetched value, or a 404. */ + /** + * The outcome of a single {@code waitForFinish} poll: either a usable fetched value, or "absent" + * - a 404 ({@code notFoundError} set), or a 200 whose {@code data} was {@code null} (tolerated + * the same as "not yet available", not a programming error, matching {@link #getResource}'s own + * null-tolerant handling of that case). + */ private record PollOutcome(T value, ApifyApiException notFoundError) { static PollOutcome of(T value) { return new PollOutcome<>(value, null); @@ -555,16 +560,16 @@ static PollOutcome notFound(ApifyApiException error) { return new PollOutcome<>(null, error); } - boolean isNotFound() { - return notFoundError != null; + boolean isAbsent() { + return value == null; } } /** * One poll of the {@code waitForFinish} loop. {@code lastSeen} carries the most recently observed - * resource and {@code notFoundSinceMillis} the instant 404s started, both across recursive calls - * (there is no shared mutable state between polls, so each recursive step gets everything it - * needs as a parameter). + * resource and {@code absentSinceMillis} the instant it was last seen present, both across + * recursive calls (there is no shared mutable state between polls, so each recursive step gets + * everything it needs as a parameter). */ private CompletableFuture pollOnce( JavaType dataType, @@ -574,7 +579,7 @@ private CompletableFuture pollOnce( long budgetMillis, long serverWaitCap, T lastSeen, - Long notFoundSinceMillis) { + Long absentSinceMillis) { long elapsed = System.currentTimeMillis() - start; long remainingSecs = (budgetMillis - elapsed) / 1000L; long requestSecs = @@ -598,17 +603,14 @@ private CompletableFuture pollOnce( .thenCompose( outcome -> { long now = System.currentTimeMillis(); - if (!outcome.isNotFound() && isTerminal.test(outcome.value())) { + if (!outcome.isAbsent() && isTerminal.test(outcome.value())) { return CompletableFuture.completedFuture(outcome.value()); } - T seen = outcome.isNotFound() ? lastSeen : outcome.value(); - Long notFoundSince = - outcome.isNotFound() - ? (notFoundSinceMillis != null ? notFoundSinceMillis : now) - : null; + T seen = outcome.isAbsent() ? lastSeen : outcome.value(); + Long absentSince = + outcome.isAbsent() ? (absentSinceMillis != null ? absentSinceMillis : now) : null; boolean missingTooLong = - notFoundSince != null - && now - notFoundSince >= MAX_WAIT_FOR_MISSING_RESOURCE_MILLIS; + absentSince != null && now - absentSince >= MAX_WAIT_FOR_MISSING_RESOURCE_MILLIS; if (missingTooLong || now - start >= budgetMillis) { if (seen != null) { return CompletableFuture.completedFuture(seen); @@ -636,7 +638,7 @@ private CompletableFuture pollOnce( budgetMillis, serverWaitCap, seen, - notFoundSince)); + absentSince)); }); } diff --git a/src/main/java/com/apify/client/internal/RunStartSupport.java b/src/main/java/com/apify/client/internal/RunStartSupport.java index 434eee4..c02ce02 100644 --- a/src/main/java/com/apify/client/internal/RunStartSupport.java +++ b/src/main/java/com/apify/client/internal/RunStartSupport.java @@ -45,6 +45,14 @@ public final class RunStartSupport { /** Delay between two attempts to start a run rejected for lack of resources. */ private static final Duration WAIT_FOR_RESOURCES_RETRY_INTERVAL = Duration.ofSeconds(10); + /** + * Upper bound (seconds) a caller-supplied {@code waitForResourcesSecs} is clamped to before + * computing a deadline, so a value near {@link Long#MAX_VALUE} cannot overflow the millisecond + * deadline computation below - the same reason {@code ResourceContext.waitForFinish} clamps its + * own wait budget before converting it to milliseconds. + */ + private static final long MAX_WAIT_FOR_RESOURCES_SECS = 999_999; + private RunStartSupport() {} /** @@ -74,7 +82,8 @@ public static CompletableFuture start( long deadlineMillis = waitForResourcesSecs == ActorStartOptions.WAIT_FOR_RESOURCES_INDEFINITELY ? Long.MAX_VALUE - : System.currentTimeMillis() + Math.max(0, waitForResourcesSecs) * 1000L; + : System.currentTimeMillis() + + Math.min(Math.max(0, waitForResourcesSecs), MAX_WAIT_FOR_RESOURCES_SECS) * 1000L; return startRetryingForResources(ctx, params, body, contentType, deadlineMillis); } diff --git a/src/main/java/com/apify/client/log/StreamedLog.java b/src/main/java/com/apify/client/log/StreamedLog.java index ec30cbb..c2497d5 100644 --- a/src/main/java/com/apify/client/log/StreamedLog.java +++ b/src/main/java/com/apify/client/log/StreamedLog.java @@ -1,12 +1,14 @@ package com.apify.client.log; import com.apify.client.http.ApifyApiException; +import com.apify.client.internal.Async; import com.apify.client.internal.HttpClientCore; import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; import java.io.SequenceInputStream; import java.nio.charset.StandardCharsets; +import java.time.Duration; import java.time.Instant; import java.time.format.DateTimeParseException; import java.util.ArrayList; @@ -68,6 +70,12 @@ public final class StreamedLog implements AutoCloseable { /** Upper bound on the pause between reopens of an empty log stream. */ private static final long EMPTY_STREAM_MAX_RETRY_MILLIS = 5_000; + /** + * How long {@link #stop()}/{@link #close()} wait for a message already in flight before + * force-closing the stream, matching the reference client's {@code STOP_GRACE_MILLIS}. + */ + private static final long STOP_GRACE_MILLIS = 1_000; + private final LogClient logClient; private final Consumer destination; @@ -180,33 +188,33 @@ public synchronized void close() { } /** - * Signals the reader to stop, unblocks it by closing the live stream, and joins it (unless called - * from the reader thread itself, which cannot join itself). Callers must hold the monitor and - * must have already checked that a thread is running. + * Signals the reader to stop and joins it (unless called from the reader thread itself, which + * cannot join itself). Callers must hold the monitor and must have already checked that a thread + * is running. */ private void stopStreaming() { stopLogging = true; - // Close the live stream so a blocked read returns promptly instead of waiting for more bytes. - closeQuietly(activeStream); - // Wakes the reader if it is currently backing off before reopening a log stream that has not - // delivered anything yet (see awaitNonEmptyStream) - that wait is a plain Thread.sleep, not - // tied - // to the stream, so closing activeStream above does not interrupt it on its own. The interrupt - // is - // a pure wake signal here (stopLogging, checked right after, is the actual stop condition), not - // a - // cancellation request the reader needs to honor, so awaitNonEmptyStream clears rather than - // re-raises it - a lingering interrupt status would otherwise risk breaking the reader's next - // blocking I/O call. Callers have already checked a thread is running (see each's own - // contract). + // Grace window before force-closing the live stream, so a message already in flight (e.g. the + // run's last line, sent right as it finishes) still has a chance to arrive and be processed + // instead of being cut off - matching the reference client's STOP_GRACE_MILLIS. A blocked read + // either completes normally within this window or is unblocked by the forced close below. + InputStream streamToForceClose = activeStream; + CompletableFuture graceTimer = + Async.delay(Duration.ofMillis(STOP_GRACE_MILLIS)) + .thenRun(() -> closeQuietly(streamToForceClose)); + // Wakes the reader if it is currently backing off before reopening an empty log stream (a plain + // Thread.sleep - see awaitNonEmptyStream), which neither the stream close above nor the grace + // window otherwise interrupts. It is a wake signal only, not a cancellation the reader must + // honor: sleepAsWakeableBackoff clears it, matching its own javadoc. streamingThread.interrupt(); // A thread must never join itself. The destination consumer runs on the streaming thread, so a // user calling stop()/close() from inside that consumer (e.g. "stop redirecting once I see line // X") would otherwise join() the current thread on itself and block forever - and because // stop()/close() are synchronized, that hung thread keeps the monitor, deadlocking every later - // start/stop/close with no exception. Setting stopLogging + closing the stream already unwinds - // the reader loop, so in that self-stop case we skip only the join; the field is still cleared - // below so lifecycle state stays consistent. + // start/stop/close with no exception. Setting stopLogging already unwinds the reader loop + // cooperatively (no read is blocked from inside the consumer's own call stack), so in that + // self-stop case we skip only the join; the field is still cleared below so lifecycle state + // stays consistent. if (Thread.currentThread() != streamingThread) { try { streamingThread.join(); @@ -214,6 +222,11 @@ private void stopStreaming() { Thread.currentThread().interrupt(); } } + // The reader has exited (or, in the self-stop case, will unwind on its own without needing the + // forced close) - cancelling here is what makes the grace window end early instead of always + // costing the full STOP_GRACE_MILLIS, and prevents a late timer firing from closing a stream a + // subsequent start() may already have replaced. + graceTimer.cancel(false); streamingThread = null; } @@ -241,7 +254,13 @@ private void streamLog() { } if (stream == null) { // Stopped before any stream ever delivered a byte: fetch whatever the log holds right now in - // one plain request instead of redirecting nothing at all. + // one plain request instead of redirecting nothing at all. Clears a pending interrupt status + // first - stopStreaming()'s wake-up interrupt (see its javadoc) may not yet have been + // consumed + // by a Thread.sleep if it landed between two awaitNonEmptyStream attempts, and it must not + // leak + // into the destination consumer invoked below or into the blocking HTTP call just below that. + Thread.interrupted(); try { String logContent = logClient.get(new LogOptions().raw(true)).join().orElse(""); if (!forwardingFailed) { @@ -286,6 +305,12 @@ private void streamLog() { LOG.warn("Log redirection stopped due to error", e); } } finally { + // Clears a pending interrupt status before the flush below can invoke the destination + // consumer: stopStreaming()'s wake-up interrupt (see its javadoc) is only guaranteed consumed + // when it lands during awaitNonEmptyStream's Thread.sleep; landing instead while this loop + // was + // running would otherwise leak it into user code. + Thread.interrupted(); // Flush whatever is left when the stream ends OR when a stop closes the stream mid-read: the // last complete message is held back in `pending` by emitMessages(..., false), and a stop // unblocks the read via an IOException that skips the loop body. Running the final flush in @@ -349,11 +374,20 @@ private InputStream awaitNonEmptyStream() throws IOException { } throw new IOException("failed to reopen log stream", cause); } - // Published under the monitor stop()/close() also use for their own activeStream read, so a - // concurrent stop cannot race this assignment: it either closes the old stream (and this - // method's own stopLogging check above then closes the new one too) or the new one. - synchronized (this) { - activeStream = stream; + // A plain volatile write (no lock needed for a single field): stop()/close() read this same + // field to decide what to force-close, and a `synchronized` publish here could instead + // deadlock against a concurrent stop()/close() - both are `synchronized` methods that may be + // blocked in streamingThread.join() while holding that monitor, which this reader thread + // could + // never then acquire. + activeStream = stream; + // A stop() requested while the reopen above was in flight force-closed the *previous* stream + // (the one this method already closed on seeing it empty), not this newly opened one - so it + // would otherwise be handed to the caller with nothing left able to unblock a read that then + // blocks. Closing it here too keeps every reopened stream subject to the same stop signal. + if (stopLogging) { + closeQuietly(stream); + return null; } } } diff --git a/src/test/java/com/apify/client/ClientBehaviourRegressionTest.java b/src/test/java/com/apify/client/ClientBehaviourRegressionTest.java index ffbd51c..3233912 100644 --- a/src/test/java/com/apify/client/ClientBehaviourRegressionTest.java +++ b/src/test/java/com/apify/client/ClientBehaviourRegressionTest.java @@ -694,6 +694,23 @@ void waitForFinishReturnsWhenResourceAppearsAfter404() { assertEquals(2, backend.calls, "one 404 poll then the terminal poll"); } + @Test + void waitForFinishToleratesNullDataBodyAsNotYetAvailable() { + // A 200 response with a null `data` body (as opposed to a 404) must be tolerated the same way - + // treated as "not yet available" and retried - not thrown as an NPE from testing a null value + // for + // terminality. + MockTransport backend = + new MockTransport( + List.of( + MockTransport.ok(200, "{\"data\":null}"), + MockTransport.ok(200, "{\"data\":{\"id\":\"r1\",\"status\":\"SUCCEEDED\"}}"))); + ActorRun run = client(backend).run("r1").waitForFinish(5L).join(); + assertEquals("r1", run.getId()); + assertEquals("SUCCEEDED", run.getStatus()); + assertEquals(2, backend.calls, "one null-data poll then the terminal poll"); + } + @Test void waitForFinishThrowsWhenResourceNeverAppears() { // If the resource is never fetchable within the budget, the wait must fail loudly rather than diff --git a/src/test/java/com/apify/client/StreamedLogTest.java b/src/test/java/com/apify/client/StreamedLogTest.java index e03aaa4..7423bfe 100644 --- a/src/test/java/com/apify/client/StreamedLogTest.java +++ b/src/test/java/com/apify/client/StreamedLogTest.java @@ -6,6 +6,7 @@ import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; import static org.junit.jupiter.api.Assertions.assertTrue; import com.apify.client.log.StreamedLog; @@ -437,6 +438,32 @@ void stopBeforeAnyBytesStreamedFallsBackToOneShotFetch() { assertEquals(List.of("fallback content"), collected); } + @Test + void stopDuringAnInFlightReopenStillCompletesPromptly() { + // Regression test: stop() must still unblock redirection (not hang) when it races a reopen that + // is already in flight - every scripted open comes back empty, so without draining into the + // "stopped, fetch once" path the reader would otherwise keep reopening for the whole backoff + // schedule (or, pre-fix, could block forever on a newly installed stream nothing then closes). + MockTransport backend = MockTransport.ofConstant(200, ""); + backend.scriptStream(200, ""); + backend.scriptStream(200, ""); + backend.scriptStream(200, ""); + StreamedLog streamedLog = + client(backend) + .run("run123") + .getStreamedLog(new StreamedLogOptions().toLog(m -> {})) + .join(); + streamedLog.start().join(); + long startedAt = System.currentTimeMillis(); + assertTimeoutPreemptively( + Duration.ofSeconds(10), + streamedLog::stop, + "stop() must not hang while a stream reopen is in flight"); + assertTrue( + System.currentTimeMillis() - startedAt < 10_000, + "stop() should return well within the full empty-stream backoff/retry schedule"); + } + @Test void defaultPrefixLookupFailureStillCreatesHelper() { // The no-arg getStreamedLog() builds a cosmetic per-run prefix by GETting the run (and its From 5b82b6c5169ab47113d0a22e082eb6e47b145ea2 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 12 Oct 2026 08:53:04 +0000 Subject: [PATCH 3/3] fix: correct pollOnce's absentSinceMillis javadoc It described the field as "the instant it was last seen present", but it is set to the instant the resource was first observed absent. --- .../java/com/apify/client/internal/ResourceContext.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/main/java/com/apify/client/internal/ResourceContext.java b/src/main/java/com/apify/client/internal/ResourceContext.java index f6a8838..484a4e9 100644 --- a/src/main/java/com/apify/client/internal/ResourceContext.java +++ b/src/main/java/com/apify/client/internal/ResourceContext.java @@ -567,9 +567,9 @@ boolean isAbsent() { /** * One poll of the {@code waitForFinish} loop. {@code lastSeen} carries the most recently observed - * resource and {@code absentSinceMillis} the instant it was last seen present, both across - * recursive calls (there is no shared mutable state between polls, so each recursive step gets - * everything it needs as a parameter). + * resource and {@code absentSinceMillis} the instant it was first observed absent (404, or a null + * {@code data} body), both across recursive calls (there is no shared mutable state between + * polls, so each recursive step gets everything it needs as a parameter). */ private CompletableFuture pollOnce( JavaType dataType,