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..484a4e9 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,35 @@ 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 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);
+ }
+
+ static PollOutcome notFound(ApifyApiException error) {
+ return new PollOutcome<>(null, error);
+ }
+
+ boolean isAbsent() {
+ return value == 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 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,
@@ -543,7 +578,8 @@ private CompletableFuture pollOnce(
long start,
long budgetMillis,
long serverWaitCap,
- T lastSeen) {
+ T lastSeen,
+ Long absentSinceMillis) {
long elapsed = System.currentTimeMillis() - start;
long remainingSecs = (budgetMillis - elapsed) / 1000L;
long requestSecs =
@@ -552,25 +588,43 @@ 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.isAbsent() && isTerminal.test(outcome.value())) {
+ return CompletableFuture.completedFuture(outcome.value());
}
- if (System.currentTimeMillis() - start >= budgetMillis) {
+ T seen = outcome.isAbsent() ? lastSeen : outcome.value();
+ Long absentSince =
+ outcome.isAbsent() ? (absentSinceMillis != null ? absentSinceMillis : now) : null;
+ boolean missingTooLong =
+ absentSince != null && now - absentSince >= 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 +637,8 @@ private CompletableFuture pollOnce(
start,
budgetMillis,
serverWaitCap,
- seen));
+ seen,
+ absentSince));
});
}
diff --git a/src/main/java/com/apify/client/internal/RunStartSupport.java b/src/main/java/com/apify/client/internal/RunStartSupport.java
index bc8b8a4..c02ce02 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,38 @@
* {@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);
+
+ /**
+ * 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() {}
/**
@@ -34,19 +61,80 @@ 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.min(Math.max(0, waitForResourcesSecs), MAX_WAIT_FOR_RESOURCES_SECS) * 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 +142,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 +161,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 +171,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..c2497d5 100644
--- a/src/main/java/com/apify/client/log/StreamedLog.java
+++ b/src/main/java/com/apify/client/log/StreamedLog.java
@@ -1,10 +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;
@@ -55,6 +59,23 @@ 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;
+
+ /**
+ * 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;
@@ -167,21 +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);
+ // 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();
@@ -189,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;
}
@@ -202,15 +240,47 @@ 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. 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) {
+ 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);
@@ -235,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
@@ -247,6 +323,89 @@ 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);
+ }
+ // 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;
+ }
+ }
+ }
+
+ /**
+ * 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..3233912 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;
@@ -693,16 +694,127 @@ 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
- // 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..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;
@@ -397,6 +398,72 @@ 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 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