diff --git a/CHANGELOG.md b/CHANGELOG.md index f44df21..7cf4c37 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,29 @@ # Changelog +## 0.5.0 + +- Bumped `ApifyClientVersion.ApiSpecVersion` to the Apify OpenAPI spec `v2-2026-10-08T073338Z` and + the project version to `0.5.0`. +- Added a `WaitForResources` option to `ActorStartOptions`/`TaskStartOptions`: `ActorClient.StartAsync`/ + `CallAsync` and `TaskClient.StartAsync`/`CallAsync` retry a start rejected with + `actor-memory-limit-exceeded`/`concurrent-runs-limit-exceeded` every 10 seconds instead of throwing + immediately. `WaitForResources.UntilAvailable` retries until the run starts; `WaitForResources.For(seconds)` + bounds the retrying, after which the last error is thrown. Matches the reference client's + `waitForResources` start option. +- Fixed the auto-paging collection iterators (`IterateAsync` on every collection client) to stop only once + a page returns no items, instead of once `offset` reached an earlier page's `total` — a stale bound that + cut the iteration short when the listed collection grew while it ran (e.g. a dataset collection while + another process created more datasets). +- Fixed `RunClient.WaitForFinishAsync`/`BuildClient.WaitForFinishAsync` to throw `NotFoundException` after + 3 seconds of a run/build id answering 404 continuously, instead of polling it as replica lag for the + entire (possibly multi-day) `waitSecs` budget. +- Fixed `StreamedLog` (`RunClient.GetStreamedLog`) to reopen the log stream (with growing backoff up to 5s) + when it ends empty before the run has logged anything, instead of treating that empty stream as the + run's whole log. Stopping before any bytes have streamed now falls back to one plain log request instead + of redirecting nothing. +- `WaitForResources.For` now rejects `NaN`/infinite budgets (use `WaitForResources.UntilAvailable` instead + of `For(double.PositiveInfinity)`). + ## 0.4.0 - Bumped `ApifyClientVersion.ApiSpecVersion` to the Apify OpenAPI spec `v2-2026-10-01T153946Z` and diff --git a/docs/actors.md b/docs/actors.md index 63f1675..5d5f695 100644 --- a/docs/actors.md +++ b/docs/actors.md @@ -42,7 +42,8 @@ foreach (var actor in page.Items) - `StartAsync(object? input = null, ActorStartOptions? options = null)` → `ActorRun` (returns immediately). - `CallAsync(object? input = null, ActorStartOptions? options = null, int? waitSecs = null, Action? log = null)` → `ActorRun` (starts then waits; `waitSecs` bounds the wait, `null` waits indefinitely; `log`, if set, - redirects the run's live log to that sink for the duration of the wait). + redirects the run's live log to that sink for the duration of the wait). Time spent retrying + `ActorStartOptions.WaitForResources` does not count toward `waitSecs`. - `ValidateInputAsync(object? input = null, ValidateInputOptions? options = null)` → `bool`. `input` on `StartAsync`/`CallAsync`/`ValidateInputAsync` (and `RunClient.MetamorphAsync`, see @@ -68,6 +69,7 @@ without re-encoding it as JSON. | `MaxTotalChargeUsd` | `double?` | Maximum total USD the run is allowed to charge. | | `ContentType` | `string?` | Content type of the `input` body (defaults to `application/json`). | | `RestartOnError` | `bool?` | Automatically restart the run's container if it exits with an error. | +| `WaitForResources` | `WaitForResources?` | Retry the start while the account has no free memory or concurrent-run slot for it, instead of throwing immediately. `WaitForResources.UntilAvailable` retries until the run starts; `WaitForResources.For(seconds)` bounds the retrying, after which the last error is thrown. `null` (the default) disables retrying. | | `ForcePermissionLevel` | `string?` | Override the Actor's permission level (`LIMITED_PERMISSIONS`/`FULL_PERMISSIONS`). | | `Webhooks` | `object?` | Ad-hoc webhooks (any JSON-serializable list) to attach to this run. | diff --git a/docs/builds.md b/docs/builds.md index e29eb70..cfb39c7 100644 --- a/docs/builds.md +++ b/docs/builds.md @@ -15,7 +15,8 @@ Access the account-wide build collection with `client.Builds()`, an Actor's buil - `AbortAsync()` → `Build`. - `DeleteAsync()`. - `WaitForFinishAsync(int? waitSecs = null)` → `Build` — client-side polling until terminal (`null` - waits indefinitely). + waits indefinitely). A build id that keeps answering 404 for 3 seconds straight (replica lag resolves + faster than that) throws `NotFoundException` instead of polling for the rest of `waitSecs`. - `GetOpenApiDefinitionAsync()` → `JsonObject?`. - `Log()` → `LogClient` (its `GetAsync`/`StreamAsync` methods are documented under [Logs in misc.md](misc.md#logs--clientlogbuildorrunid); on this build-nested client they throw on a 404 rather than resolve `null`, since the build itself could be what's missing). diff --git a/docs/runs.md b/docs/runs.md index 793c828..75be828 100644 --- a/docs/runs.md +++ b/docs/runs.md @@ -24,7 +24,9 @@ collections). - `RebootAsync()` → `ActorRun`. - `ResurrectAsync(RunResurrectOptions? options = null)` → `ActorRun`. - `ChargeAsync(RunChargeOptions options)` → `Task` — record pay-per-event charges (idempotent). -- `WaitForFinishAsync(int? waitSecs = null)` → `ActorRun`. +- `WaitForFinishAsync(int? waitSecs = null)` → `ActorRun` — client-side polling until terminal (`null` + waits indefinitely). A run id that keeps answering 404 for 3 seconds straight (replica lag resolves + faster than that) throws `NotFoundException` instead of polling for the rest of `waitSecs`. - `Dataset()`, `KeyValueStore()`, `RequestQueue()` — the run's default storages. Their own `GetAsync()`/`DeleteAsync()` throw on a 404 instead of resolving `null`/no-op, since the run itself (not just the storage) could be what's missing; a record/request looked up by its own key or id diff --git a/docs/tasks.md b/docs/tasks.md index 8a554b9..1677ce5 100644 --- a/docs/tasks.md +++ b/docs/tasks.md @@ -31,8 +31,10 @@ a specific task with `client.Task(id)`. `TaskStartOptions` overrides the task's stored run settings for a single start: `Build`, `MemoryMbytes`, `TimeoutSecs`, `WaitForFinish` (server-side wait on the start call, max 60), -`MaxItems`, `MaxTotalChargeUsd`, `RestartOnError`, and `Webhooks` (ad-hoc webhooks for this run). See -[Actors](actors.md) for the meaning of each field. +`MaxItems`, `MaxTotalChargeUsd`, `RestartOnError`, `WaitForResources` (retry the start while the +account has no free memory or concurrent-run slot for it; time spent retrying does not count toward +`CallAsync`'s `waitSecs`), and `Webhooks` (ad-hoc webhooks for this run). See [Actors](actors.md) for +the meaning of each field. The model is named `ActorTask` (not `Task`) to avoid colliding with `System.Threading.Tasks.Task`. diff --git a/src/Apify.Client/Apify.Client.csproj b/src/Apify.Client/Apify.Client.csproj index 432ccf0..157148d 100644 --- a/src/Apify.Client/Apify.Client.csproj +++ b/src/Apify.Client/Apify.Client.csproj @@ -6,7 +6,7 @@ Apify.Client - 0.4.0 + 0.5.0 Apify Apify Apify API client for .NET diff --git a/src/Apify.Client/ApifyClientVersion.cs b/src/Apify.Client/ApifyClientVersion.cs index 35c9810..dd0bf82 100644 --- a/src/Apify.Client/ApifyClientVersion.cs +++ b/src/Apify.Client/ApifyClientVersion.cs @@ -14,11 +14,11 @@ public static class ApifyClientVersion /// The semantic version of this client library (see https://semver.org/). Changes to the public /// interface other than additive ones are considered breaking changes. /// - public const string ClientVersion = "0.4.0"; + public const string ClientVersion = "0.5.0"; /// /// The version of the Apify OpenAPI specification this client was generated and verified against. /// Corresponds to the info.version field of the Apify OpenAPI document. /// - public const string ApiSpecVersion = "v2-2026-10-01T153946Z"; + public const string ApiSpecVersion = "v2-2026-10-08T073338Z"; } diff --git a/src/Apify.Client/Internal/ResourceContext.cs b/src/Apify.Client/Internal/ResourceContext.cs index c784ef3..61f99d3 100644 --- a/src/Apify.Client/Internal/ResourceContext.cs +++ b/src/Apify.Client/Internal/ResourceContext.cs @@ -38,6 +38,15 @@ internal sealed class ResourceContext /// private const int MaxWaitForFinishSecs = 999999; + /// + /// How long keeps polling a resource that keeps answering 404 (to ride + /// out database replica lag just after creation) before giving up and throwing, instead of running out + /// the full wait budget. Matches the reference client's MAX_WAIT_WHEN_JOB_NOT_EXIST_MILLIS — a + /// job that never existed (e.g. a typo'd run id) would otherwise hang for up to + /// (~11.5 days) when waitSecs is left unset. + /// + private const int MaxWaitWhenJobNotExistMillis = 3000; + private readonly string _apiOrigin; private string _publicOrigin; private TimeSpan? _requestTimeout; @@ -211,6 +220,14 @@ public async Task> ListResourceAsync(string subPath, QueryP /// if is set, yields at most that many items in total. /// carries the caller's filters; its offset/limit are overwritten per page. /// + /// + /// Paging stops once a page returns no items (or is reached), never by + /// comparing the offset to a page's reported total: the listed resource can grow while the + /// iteration runs (e.g. a dataset collection a running Actor keeps adding to), and a reported total + /// only ever reflects the moment its own page was fetched. Stopping at an early page's total would cut + /// the iteration off before items added later, so — like the reference client — this always issues one + /// more request at the end to confirm there is nothing left, unless is hit first. + /// public async IAsyncEnumerable IterateListAsync( string subPath, QueryParams baseQuery, @@ -238,7 +255,7 @@ public async IAsyncEnumerable IterateListAsync( } offset += (int)page.Count; - if (page.Count == 0 || offset >= page.Total || (limit is not null && yielded >= limit.Value)) + if (page.Count == 0 || (limit is not null && yielded >= limit.Value)) { yield break; } @@ -408,8 +425,10 @@ private int ServerWaitCapSecs() /// /// Polls a GET endpoint with waitForFinish until the resource reaches a terminal state or the /// wait budget elapses. == null means "wait indefinitely", - /// implemented as a finite but very large bound so the loop always terminates. A transient 404 (replica - /// lag) is treated as "not yet available". + /// implemented as a finite but very large bound so the loop always terminates. A 404 is treated as "not + /// yet available" (database replica lag just after creation) for up to + /// of continuous 404s, after which it is reported as a + /// genuine not-found rather than retried for the rest of the (possibly multi-day) wait budget. /// public async Task WaitForFinishAsync(int? waitSecs, string resourceName, Func isTerminal, CancellationToken ct) { @@ -421,6 +440,8 @@ public async Task WaitForFinishAsync(int? waitSecs, string resourceN var serverWaitCap = ServerWaitCapSecs(); JsonObject? resource = null; + long? notFoundSinceMillis = null; + ApifyApiException? notFoundError = null; while (true) { @@ -431,18 +452,35 @@ public async Task WaitForFinishAsync(int? waitSecs, string resourceN var p = new QueryParams(); p.AddInt("waitForFinish", requestSecs); - // Long tier (the overall per-request budget): the server may hold the connection open for up to - // requestSecs (derived from that same budget via ServerWaitCapSecs), so the HTTP timeout for this - // call must cover the full budget, not a shorter tier. - var data = await GetResourceAsync(string.Empty, p, RequestTier.Long, ct).ConfigureAwait(false); - if (data is JsonObject obj) + var isNotFoundNow = false; + try { - resource = obj; - if (isTerminal(obj)) + // Long tier (the overall per-request budget): the server may hold the connection open for + // up to requestSecs (derived from that same budget via ServerWaitCapSecs), so the HTTP + // timeout for this call must cover the full budget, not a shorter tier. + var data = await GetResourceRequiredAsync(string.Empty, p, RequestTier.Long, ct).ConfigureAwait(false); + if (data is JsonObject obj) { - return obj; + resource = obj; + notFoundSinceMillis = null; + notFoundError = null; + if (isTerminal(obj)) + { + return obj; + } } } + catch (ApifyApiException e) when (HttpClientCore.IsNotFound(e)) + { + isNotFoundNow = true; + notFoundSinceMillis ??= stopwatch.ElapsedMilliseconds; + notFoundError = e; + } + + if (isNotFoundNow && stopwatch.ElapsedMilliseconds - notFoundSinceMillis!.Value >= MaxWaitWhenJobNotExistMillis) + { + throw notFoundError!; + } if (stopwatch.ElapsedMilliseconds >= budgetMillis) { @@ -457,6 +495,11 @@ public async Task WaitForFinishAsync(int? waitSecs, string resourceN return resource; } + if (notFoundError is not null) + { + throw notFoundError; + } + throw new InvalidOperationException( $"waiting for {resourceName} to finish failed: cannot fetch {resourceName} details from the server"); } diff --git a/src/Apify.Client/Internal/ResourceLimitRetry.cs b/src/Apify.Client/Internal/ResourceLimitRetry.cs new file mode 100644 index 0000000..585239c --- /dev/null +++ b/src/Apify.Client/Internal/ResourceLimitRetry.cs @@ -0,0 +1,83 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using Apify.Client.Exceptions; +using Apify.Client.Options; + +namespace Apify.Client.Internal; + +/// +/// Implements /: +/// retries a run start while the API rejects it for lack of resources. Matches the reference client's +/// startWaitingForResources(). +/// +internal static class ResourceLimitRetry +{ + /// + /// Error types the API rejects a run start with while the account has no free memory or concurrent-run + /// slot for it. Both clear as other runs or builds finish. + /// + private const string MemoryLimitErrorType = "actor-memory-limit-exceeded"; + + private const string ConcurrentRunsLimitErrorType = "concurrent-runs-limit-exceeded"; + + /// Cooldown between two attempts to start a run that was rejected for lack of resources. + private const int CooldownMillis = 10_000; + + /// + /// Calls , retrying it every while it fails + /// with a resource-limit error. null disables retrying + /// entirely (the first error is thrown as-is); otherwise + /// null retries until the call succeeds, and a value bounds the retrying, after which the last + /// error is thrown. Any other error is thrown right away. + /// + /// Makes one start attempt. + /// The retry configuration, or null to disable retrying. + /// A token to cancel the wait between attempts. + /// + /// The cooldown between attempts; defaults to . Overridable only so unit + /// tests can exercise multi-attempt retries without a real 10-second wait per attempt. + /// + public static async Task StartAsync( + Func> start, + WaitForResources? waitForResources, + CancellationToken cancellationToken, + int cooldownMillis = CooldownMillis) + { + if (waitForResources is null) + { + return await start().ConfigureAwait(false); + } + + var budgetMillis = waitForResources.Value.MaxWaitSecs is { } secs ? (long)(secs * 1000) : (long?)null; + var stopwatch = budgetMillis is not null ? System.Diagnostics.Stopwatch.StartNew() : null; + + while (true) + { + try + { + return await start().ConfigureAwait(false); + } + catch (ApifyApiException e) when (IsResourceLimitError(e)) + { + if (budgetMillis is not null) + { + var remainingMillis = budgetMillis.Value - stopwatch!.ElapsedMilliseconds; + if (remainingMillis <= 0) + { + throw; + } + + await Task.Delay((int)Math.Min(cooldownMillis, remainingMillis), cancellationToken).ConfigureAwait(false); + } + else + { + await Task.Delay(cooldownMillis, cancellationToken).ConfigureAwait(false); + } + } + } + } + + private static bool IsResourceLimitError(ApifyApiException e) => + e.Type is MemoryLimitErrorType or ConcurrentRunsLimitErrorType; +} diff --git a/src/Apify.Client/Options/ActorStartOptions.cs b/src/Apify.Client/Options/ActorStartOptions.cs index 3665dbe..badc4d7 100644 --- a/src/Apify.Client/Options/ActorStartOptions.cs +++ b/src/Apify.Client/Options/ActorStartOptions.cs @@ -31,6 +31,13 @@ public sealed class ActorStartOptions /// If true, restart the run if it fails. public bool? RestartOnError { get; init; } + /// + /// Retries the start while the account lacks the memory or a concurrent-run slot for the run (an + /// actor-memory-limit-exceeded/concurrent-runs-limit-exceeded error), instead of throwing + /// immediately. null (the default) disables retrying. + /// + public WaitForResources? WaitForResources { get; init; } + /// /// Override the Actor's permission level for this run (LIMITED_PERMISSIONS/FULL_PERMISSIONS). /// diff --git a/src/Apify.Client/Options/TaskStartOptions.cs b/src/Apify.Client/Options/TaskStartOptions.cs index 0a6fd03..f11ee5b 100644 --- a/src/Apify.Client/Options/TaskStartOptions.cs +++ b/src/Apify.Client/Options/TaskStartOptions.cs @@ -32,6 +32,13 @@ public sealed class TaskStartOptions /// If true, restart the run if it fails. public bool? RestartOnError { get; init; } + /// + /// Retries the start while the account lacks the memory or a concurrent-run slot for the run (an + /// actor-memory-limit-exceeded/concurrent-runs-limit-exceeded error), instead of throwing + /// immediately. null (the default) disables retrying. + /// + public WaitForResources? WaitForResources { get; init; } + /// Ad-hoc webhooks to attach (a JSON-serializable list serialized to base64-encoded JSON). public object? Webhooks { get; init; } diff --git a/src/Apify.Client/Options/WaitForResources.cs b/src/Apify.Client/Options/WaitForResources.cs new file mode 100644 index 0000000..be3b33a --- /dev/null +++ b/src/Apify.Client/Options/WaitForResources.cs @@ -0,0 +1,62 @@ +using System; + +namespace Apify.Client.Options; + +/// +/// Configures retrying a run start while the account has no free memory or concurrent-run slot for it (the +/// API's actor-memory-limit-exceeded/concurrent-runs-limit-exceeded errors, which clear as +/// other runs or builds finish). Passed to and +/// ; matches the reference client's waitForResources +/// start option (there a boolean | number union, modelled here as this small value type so the +/// "retry forever" and "retry for N seconds" cases stay distinct and self-documenting). +/// +public readonly struct WaitForResources : IEquatable +{ + private WaitForResources(double? maxWaitSecs) + { + MaxWaitSecs = maxWaitSecs; + } + + /// + /// The maximum time to keep retrying, in seconds, or null when retrying until the run starts + /// (see ). + /// + public double? MaxWaitSecs { get; } + + /// Retries indefinitely until the run starts. + public static WaitForResources UntilAvailable { get; } = new(null); + + /// + /// Retries for up to seconds, after which the last + /// actor-memory-limit-exceeded/concurrent-runs-limit-exceeded error is thrown. + /// + /// + /// The retry budget in seconds; must be finite and not negative (use , + /// not , to retry forever). + /// + public static WaitForResources For(double maxWaitSecs) + { + if (maxWaitSecs < 0 || !double.IsFinite(maxWaitSecs)) + { + throw new ArgumentOutOfRangeException( + nameof(maxWaitSecs), maxWaitSecs, "must be a non-negative, finite number of seconds; use UntilAvailable to retry forever"); + } + + return new WaitForResources(maxWaitSecs); + } + + /// + public bool Equals(WaitForResources other) => Nullable.Equals(MaxWaitSecs, other.MaxWaitSecs); + + /// + public override bool Equals(object? obj) => obj is WaitForResources other && Equals(other); + + /// + public override int GetHashCode() => MaxWaitSecs.GetHashCode(); + + /// Equality by . + public static bool operator ==(WaitForResources left, WaitForResources right) => left.Equals(right); + + /// Inequality by . + public static bool operator !=(WaitForResources left, WaitForResources right) => !left.Equals(right); +} diff --git a/src/Apify.Client/Resources/ActorClient.cs b/src/Apify.Client/Resources/ActorClient.cs index ebaab05..5a8e61a 100644 --- a/src/Apify.Client/Resources/ActorClient.cs +++ b/src/Apify.Client/Resources/ActorClient.cs @@ -61,9 +61,17 @@ public async Task UpdateAsync(object newFields, CancellationToken cancell /// /// Optional run-start options. /// A token to cancel the request. - public async Task StartAsync(object? input = null, ActorStartOptions? options = null, CancellationToken cancellationToken = default) + public Task StartAsync(object? input = null, ActorStartOptions? options = null, CancellationToken cancellationToken = default) { options ??= new ActorStartOptions(); + return ResourceLimitRetry.StartAsync( + () => StartOnceAsync(input, options, cancellationToken), + options.WaitForResources, + cancellationToken); + } + + private async Task StartOnceAsync(object? input, ActorStartOptions options, CancellationToken cancellationToken) + { var q = new QueryParams(); options.AppendTo(q); ResourceContext.EncodeInputBody(input, out var body, out var bodyBytes); @@ -75,7 +83,10 @@ public async Task StartAsync(object? input = null, ActorStartOptions? /// Starts the Actor and waits (client-side polling) for it to finish. /// Any JSON-serializable value (or null for no input). - /// Optional run-start options. + /// + /// Optional run-start options. Time spent retrying + /// does not count toward , which only bounds the wait once the run starts. + /// /// Bounds the wait; null waits indefinitely. /// /// If provided, the run's live log is redirected to this sink (one complete message per call) for the diff --git a/src/Apify.Client/Resources/TaskClient.cs b/src/Apify.Client/Resources/TaskClient.cs index 74553c2..220d349 100644 --- a/src/Apify.Client/Resources/TaskClient.cs +++ b/src/Apify.Client/Resources/TaskClient.cs @@ -90,10 +90,19 @@ public Task UnpublishAsync(CancellationToken cancellationToken = defa /// /// Optional run-start options. /// A token to cancel the request. - public async Task StartAsync(object? input = null, TaskStartOptions? options = null, CancellationToken cancellationToken = default) + public Task StartAsync(object? input = null, TaskStartOptions? options = null, CancellationToken cancellationToken = default) + { + options ??= new TaskStartOptions(); + return ResourceLimitRetry.StartAsync( + () => StartOnceAsync(input, options, cancellationToken), + options.WaitForResources, + cancellationToken); + } + + private async Task StartOnceAsync(object? input, TaskStartOptions options, CancellationToken cancellationToken) { var q = new QueryParams(); - (options ?? new TaskStartOptions()).AppendTo(q); + options.AppendTo(q); var body = input is null ? null : Json.Encode(input); // Long, not the reference's base "medium": options.WaitForFinish can ask the server to hold the // connection open for up to 60s and this call does not clamp it, so the HTTP timeout must cover it. @@ -102,7 +111,10 @@ public async Task StartAsync(object? input = null, TaskStartOptions? o /// Starts the task and waits (client-side polling) for it to finish. /// Optionally overrides the task's stored input. - /// Optional run-start options. + /// + /// Optional run-start options. Time spent retrying does + /// not count toward , which only bounds the wait once the run starts. + /// /// Bounds the wait; null waits indefinitely. /// /// If provided, the run's live log is redirected to this sink (one complete message per call) for the diff --git a/src/Apify.Client/StreamedLog.cs b/src/Apify.Client/StreamedLog.cs index 1ab6f60..45c8a2b 100644 --- a/src/Apify.Client/StreamedLog.cs +++ b/src/Apify.Client/StreamedLog.cs @@ -33,6 +33,16 @@ public sealed class StreamedLog : IAsyncDisposable /// Size of the read buffer used while draining the log stream. private const int ReadChunkChars = 4096; + /// + /// Pause before the first reopen of a log stream that ended before the run had logged anything. Each + /// further reopen doubles the pause up to , bounding the request + /// rate while a run waits a long time to start (e.g. for free memory). + /// + private const int EmptyStreamInitialRetryMillis = 500; + + /// Upper bound on the pause between reopens of an empty log stream. + private const int EmptyStreamMaxRetryMillis = 5_000; + private readonly LogClient _logClient; private readonly Action _toLog; private readonly DateTimeOffset? _relevancyTimeLimit; @@ -104,26 +114,62 @@ public async Task StopAsync() private async Task StreamLogAsync(CancellationToken cancellationToken) { - // _logClient is always the run/build-nested (ambiguous) client (see the internal constructor's only - // caller, RunClient.GetStreamedLog), which never resolves a 404 to null — it throws instead. - using var stream = (await _logClient.StreamAsync(new LogOptions { Raw = true }, cancellationToken).ConfigureAwait(false))!; - using var reader = new StreamReader(stream, Encoding.UTF8); - var buffer = new StringBuilder(); - var chunk = new char[ReadChunkChars]; - int read; - while ((read = await reader.ReadAsync(chunk.AsMemory(0, chunk.Length), cancellationToken).ConfigureAwait(false)) > 0) + var sawAnyBytes = false; + var retryMillis = EmptyStreamInitialRetryMillis; + + try { - buffer.Append(chunk, 0, read); - FlushMessages(buffer, flushRemainder: false); - if (cancellationToken.IsCancellationRequested) + while (!sawAnyBytes) { - break; + // _logClient is always the run/build-nested (ambiguous) client (see the internal + // constructor's only caller, RunClient.GetStreamedLog), which never resolves a 404 to + // null — it throws instead. + using var stream = (await _logClient.StreamAsync(new LogOptions { Raw = true }, cancellationToken).ConfigureAwait(false))!; + using var reader = new StreamReader(stream, Encoding.UTF8); + + var chunk = new char[ReadChunkChars]; + int read; + while ((read = await reader.ReadAsync(chunk.AsMemory(0, chunk.Length), cancellationToken).ConfigureAwait(false)) > 0) + { + sawAnyBytes = true; + buffer.Append(chunk, 0, read); + FlushMessages(buffer, flushRemainder: false); + } + + if (sawAnyBytes) + { + break; + } + + // The API serves the log of a run that has not logged anything yet as an empty stream that + // closes at once, so reopen it with growing backoff until the first bytes arrive. Without + // this, a run that was still queued (no memory free yet) when streaming started would never + // have any of its log redirected, even once it did start producing output. + await Task.Delay(TimeSpan.FromMilliseconds(retryMillis), cancellationToken).ConfigureAwait(false); + retryMillis = Math.Min(retryMillis * 2, EmptyStreamMaxRetryMillis); } } - - // Emit whatever is left when the stream ends or is stopped (possibly a message without a trailing newline). - FlushMessages(buffer, flushRemainder: true); + catch (OperationCanceledException) when (!sawAnyBytes) + { + // Stopped before the run had logged anything (e.g. it was still queued): fetch whatever log + // exists in one plain request instead of silently redirecting nothing. Best-effort — a failure + // here must not turn a clean stop into a thrown exception. + try + { + var content = await _logClient.GetAsync(new LogOptions { Raw = true }, CancellationToken.None).ConfigureAwait(false); + buffer.Append(content); + } + catch (Exception) + { + // Leave the buffer as-is; the run's log (if any) simply isn't redirected for this stop. + } + } + finally + { + // Emit whatever is left: a message without a trailing newline, or the fallback fetched above. + FlushMessages(buffer, flushRemainder: true); + } } /// diff --git a/tests/Apify.Client.Tests/Unit/AutoPagingTests.cs b/tests/Apify.Client.Tests/Unit/AutoPagingTests.cs index 698ed80..0176199 100644 --- a/tests/Apify.Client.Tests/Unit/AutoPagingTests.cs +++ b/tests/Apify.Client.Tests/Unit/AutoPagingTests.cs @@ -39,9 +39,14 @@ public sealed class AutoPagingTests [Fact] public async Task CollectionIterateWalksAllPagesByOffset() { + // The iterator pages until a page scans no rows, not until `offset` reaches an earlier page's + // `total` (that total can go stale if the collection grows mid-iteration — see + // CollectionIterateDoesNotStopAtAnEarlierPagesStaleTotal), so it always sends one more request + // after the last non-empty page to confirm nothing is left. var transport = new MockTransport() .QueueResponse(200, "{\"data\":{\"total\":3,\"items\":[{\"id\":\"a\"},{\"id\":\"b\"}]}}") - .QueueResponse(200, "{\"data\":{\"total\":3,\"items\":[{\"id\":\"c\"}]}}"); + .QueueResponse(200, "{\"data\":{\"total\":3,\"items\":[{\"id\":\"c\"}]}}") + .QueueResponse(200, "{\"data\":{\"total\":3,\"items\":[]}}"); var ids = new List(); await foreach (var actor in Client(transport).Actors().IterateAsync()) @@ -50,9 +55,34 @@ public async Task CollectionIterateWalksAllPagesByOffset() } Assert.Equal(new[] { "a", "b", "c" }, ids); - Assert.Equal(2, transport.CallCount); + Assert.Equal(3, transport.CallCount); Assert.Contains("offset=0", transport.Received[0].Uri, StringComparison.Ordinal); Assert.Contains("offset=2", transport.Received[1].Uri, StringComparison.Ordinal); + Assert.Contains("offset=3", transport.Received[2].Uri, StringComparison.Ordinal); + } + + [Fact] + public async Task CollectionIterateDoesNotStopAtAnEarlierPagesStaleTotal() + { + // Regression test (mirrors a reference-client bug fix): the API recomputes `total` fresh on every + // request, so it can grow between pages when something else pushes to the collection while the + // iteration runs. The iterator must keep paging past an earlier page's `total` and only stop once a + // page actually scans no rows. + var transport = new MockTransport() + .QueueResponse(200, "{\"data\":{\"total\":2,\"items\":[{\"id\":\"a\"},{\"id\":\"b\"}]}}") + // By the time this page is fetched, a third item was added: total is now 3, not the 2 the first + // page reported. + .QueueResponse(200, "{\"data\":{\"total\":3,\"items\":[{\"id\":\"c\"}]}}") + .QueueResponse(200, "{\"data\":{\"total\":3,\"items\":[]}}"); + + var ids = new List(); + await foreach (var actor in Client(transport).Actors().IterateAsync()) + { + ids.Add(actor.Id); + } + + Assert.Equal(new[] { "a", "b", "c" }, ids); + Assert.Equal(3, transport.CallCount); } [Fact] diff --git a/tests/Apify.Client.Tests/Unit/StreamedLogTests.cs b/tests/Apify.Client.Tests/Unit/StreamedLogTests.cs index 8a12699..ff2318d 100644 --- a/tests/Apify.Client.Tests/Unit/StreamedLogTests.cs +++ b/tests/Apify.Client.Tests/Unit/StreamedLogTests.cs @@ -104,4 +104,73 @@ public async Task StartTwiceThrows() Assert.Throws(() => streamedLog.Start()); await streamedLog.StopAsync(); } + + [Fact] + public async Task ReopensStreamWhenTheFirstAttemptEndsEmptyBeforeTheRunHasLoggedAnything() + { + // Regression test (mirrors a reference-client bug fix): the API serves the log of a run that has + // not produced any output yet (e.g. still queued for free memory) as an empty stream that closes + // at once. Treating that as the whole log would silently drop everything the run logs afterwards, + // so StreamedLog must reopen the stream instead. + var transport = new MockTransport() + .QueueResponse(200, string.Empty) + .QueueResponse(200, LogBody); + + var messages = new List(); + var streamedLog = Client(transport).Run("run1").GetStreamedLog( + m => + { + lock (messages) + { + messages.Add(m); + } + }); + + streamedLog.Start(); + for (var i = 0; i < 200; i++) + { + await Task.Delay(25); + lock (messages) + { + if (messages.Count >= 3) + { + break; + } + } + } + + await streamedLog.StopAsync(); + + Assert.Equal(3, messages.Count); + Assert.Equal(2, transport.CallCount); + } + + [Fact] + public async Task StoppingBeforeAnyBytesArriveFallsBackToAOneShotGet() + { + // Regression test: if stop() lands while still waiting out the empty-stream backoff above (the run + // never produced output before the caller gave up waiting), StreamedLog must still redirect + // whatever the log holds via one plain GET, instead of silently emitting nothing. + var transport = new MockTransport() + .QueueResponse(200, string.Empty) // first stream attempt: empty, closes at once + .QueueResponse(200, LogBody); // fallback plain GET once stopped + + var messages = new List(); + var streamedLog = Client(transport).Run("run1").GetStreamedLog( + m => + { + lock (messages) + { + messages.Add(m); + } + }); + + streamedLog.Start(); + // Give the first (empty) stream attempt time to complete and enter its backoff wait before stopping. + await Task.Delay(100); + await streamedLog.StopAsync(); + + Assert.Equal(3, messages.Count); + Assert.Equal(2, transport.CallCount); + } } diff --git a/tests/Apify.Client.Tests/Unit/WaitForFinishTests.cs b/tests/Apify.Client.Tests/Unit/WaitForFinishTests.cs new file mode 100644 index 0000000..6c7c439 --- /dev/null +++ b/tests/Apify.Client.Tests/Unit/WaitForFinishTests.cs @@ -0,0 +1,79 @@ +using System; +using System.Diagnostics; +using System.Threading.Tasks; +using Apify.Client.Exceptions; +using Xunit; + +namespace Apify.Client.Tests.Unit; + +/// +/// Offline tests for RunClient.WaitForFinishAsync's handling of a run/build that keeps answering 404. +/// +[Trait("Category", "Unit")] +public sealed class WaitForFinishTests +{ + private static ApifyClient Client(MockTransport transport) => new(new ApifyClientOptions + { + Token = "t", + MinDelayBetweenRetriesMillis = 1, + TimeoutSecs = 30, + HttpTransport = transport, + }); + + [Fact] + public async Task WaitForFinishThrowsNotFoundOnPersistent404InsteadOfRunningOutTheFullWaitBudget() + { + // Regression test (mirrors a reference-client bug fix): a run id that never existed used to make + // WaitForFinishAsync treat every 404 as transient database replica lag and poll for the entire wait + // budget (up to ~11.5 days when waitSecs is left unset). It must instead give up a few seconds into + // a *persistent* 404, well before a much larger waitSecs elapses. + var transport = new MockTransport(); + for (var i = 0; i < 60; i++) + { + transport.QueueResponse(404, "{\"error\":{\"type\":\"record-not-found\",\"message\":\"Run was not found\"}}"); + } + + var stopwatch = Stopwatch.StartNew(); + await Assert.ThrowsAsync( + () => Client(transport).Run("does-not-exist").WaitForFinishAsync(waitSecs: 100)); + stopwatch.Stop(); + + // Gives up a few seconds into the persistent 404, nowhere near the 100s budget passed above. + Assert.True(stopwatch.Elapsed < TimeSpan.FromSeconds(20), $"took {stopwatch.Elapsed}"); + } + + [Fact] + public async Task WaitForFinishThrowsNotFoundWhenTheWaitBudgetElapsesBeforeTheNotFoundGraceWindow() + { + // Regression test: when `waitSecs` is small enough that the overall wait budget runs out before the + // 3-second not-found grace window does, the loop exits via the budget check rather than the grace- + // window check. It must still surface the persistent 404 as NotFoundException, not the generic + // "cannot fetch details from the server" fallback that only applies when nothing was ever fetched + // and no error was ever recorded (which cannot happen once a 404 has been seen). + var transport = new MockTransport(); + for (var i = 0; i < 10; i++) + { + transport.QueueResponse(404, "{\"error\":{\"type\":\"record-not-found\",\"message\":\"Run was not found\"}}"); + } + + var error = await Assert.ThrowsAsync( + () => Client(transport).Run("does-not-exist").WaitForFinishAsync(waitSecs: 1)); + + Assert.Equal("record-not-found", error.Type); + } + + [Fact] + public async Task WaitForFinishToleratesABriefNotFoundThenSucceeds() + { + // A short run of 404s (shorter than the not-found grace window) is replica lag, not a missing run: + // once the resource starts resolving, the run completes normally. + var transport = new MockTransport() + .QueueResponse(404, "{\"error\":{\"type\":\"record-not-found\",\"message\":\"Run was not found\"}}") + .QueueResponse(200, "{\"data\":{\"id\":\"run1\",\"status\":\"SUCCEEDED\"}}"); + + var run = await Client(transport).Run("run1").WaitForFinishAsync(waitSecs: 30); + + Assert.Equal("SUCCEEDED", run.Status); + Assert.Equal(2, transport.CallCount); + } +} diff --git a/tests/Apify.Client.Tests/Unit/WaitForResourcesTests.cs b/tests/Apify.Client.Tests/Unit/WaitForResourcesTests.cs new file mode 100644 index 0000000..4206869 --- /dev/null +++ b/tests/Apify.Client.Tests/Unit/WaitForResourcesTests.cs @@ -0,0 +1,186 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using Apify.Client.Exceptions; +using Apify.Client.Internal; +using Apify.Client.Options; +using Xunit; + +namespace Apify.Client.Tests.Unit; + +/// +/// Offline tests for ActorStartOptions.WaitForResources/TaskStartOptions.WaitForResources: +/// retrying a run start while the account has no free memory or concurrent-run slot for it. +/// +[Trait("Category", "Unit")] +public sealed class WaitForResourcesTests +{ + private const string MemoryLimitBody = + "{\"error\":{\"type\":\"actor-memory-limit-exceeded\",\"message\":\"not enough memory\"}}"; + + private const string ConcurrentRunsLimitBody = + "{\"error\":{\"type\":\"concurrent-runs-limit-exceeded\",\"message\":\"too many concurrent runs\"}}"; + + private const string InvalidInputBody = + "{\"error\":{\"type\":\"invalid-input\",\"message\":\"bad input\"}}"; + + private static ApifyClient Client(MockTransport transport) => new(new ApifyClientOptions + { + Token = "t", + MinDelayBetweenRetriesMillis = 1, + TimeoutSecs = 5, + HttpTransport = transport, + }); + + // ---- ResourceLimitRetry (the shared retry primitive) ----------------------------------------------- + + [Fact] + public async Task DisabledByDefaultThrowsTheFirstResourceLimitErrorImmediately() + { + var attempts = 0; + Task Start() + { + attempts++; + throw MakeError("actor-memory-limit-exceeded"); + } + + await Assert.ThrowsAsync( + () => ResourceLimitRetry.StartAsync(Start, waitForResources: null, CancellationToken.None)); + + Assert.Equal(1, attempts); + } + + [Fact] + public async Task UntilAvailableRetriesPastResourceLimitErrorsThenReturnsTheResult() + { + var attempts = 0; + Task Start() + { + attempts++; + if (attempts < 3) + { + throw MakeError(attempts == 1 ? "actor-memory-limit-exceeded" : "concurrent-runs-limit-exceeded"); + } + + return Task.FromResult("started"); + } + + // A tiny cooldown keeps this test fast; production call sites use the real 10s cooldown. + var result = await ResourceLimitRetry.StartAsync(Start, Options.WaitForResources.UntilAvailable, CancellationToken.None, cooldownMillis: 1); + + Assert.Equal("started", result); + Assert.Equal(3, attempts); + } + + [Fact] + public async Task BoundedWaitThrowsTheLastErrorOnceTheBudgetIsExhausted() + { + var attempts = 0; + Task Start() + { + attempts++; + throw MakeError("actor-memory-limit-exceeded"); + } + + // A zero-second budget exhausts on the very first failure without any real wait. + var error = await Assert.ThrowsAsync( + () => ResourceLimitRetry.StartAsync(Start, Options.WaitForResources.For(0), CancellationToken.None)); + + Assert.Equal("actor-memory-limit-exceeded", error.Type); + Assert.Equal(1, attempts); + } + + [Fact] + public async Task NonResourceLimitErrorsAreNeverRetried() + { + var attempts = 0; + Task Start() + { + attempts++; + throw MakeError("invalid-input"); + } + + await Assert.ThrowsAsync( + () => ResourceLimitRetry.StartAsync(Start, Options.WaitForResources.UntilAvailable, CancellationToken.None, cooldownMillis: 1)); + + Assert.Equal(1, attempts); + } + + private static ApifyApiException MakeError(string type) => + new(402, type, "test error", attempt: 1, httpMethod: "POST", path: "/v2/acts/act/runs"); + + // ---- WaitForResources value type --------------------------------------------------------------------- + + [Theory] + [InlineData(-1)] + [InlineData(double.NaN)] + [InlineData(double.PositiveInfinity)] + [InlineData(double.NegativeInfinity)] + public void ForRejectsNegativeNaNAndInfiniteBudgets(double invalidSecs) + { + // PositiveInfinity would silently behave like UntilAvailable; reject it so the two "retry + // forever" spellings stay distinct and deliberate. + Assert.Throws(() => Options.WaitForResources.For(invalidSecs)); + } + + [Fact] + public void ForAcceptsZeroAndFiniteBudgets() + { + Assert.Equal(0, Options.WaitForResources.For(0).MaxWaitSecs); + Assert.Equal(30, Options.WaitForResources.For(30).MaxWaitSecs); + } + + // ---- ActorClient / TaskClient wiring ---------------------------------------------------------------- + + [Fact] + public async Task ActorStartWithoutWaitForResourcesThrowsOnFirstResourceLimitError() + { + var transport = new MockTransport().QueueResponse(402, MemoryLimitBody); + + var error = await Assert.ThrowsAsync( + () => Client(transport).Actor("act").StartAsync()); + + Assert.Equal("actor-memory-limit-exceeded", error.Type); + Assert.Equal(1, transport.CallCount); + } + + [Fact] + public async Task ActorStartNeverSendsWaitForResourcesAsAQueryParameter() + { + // waitForResources is a client-side retry policy, not an API parameter; the retry loop's cooldown is + // a real 10s per attempt in production (see ResourceLimitRetry), which the generic retry tests above + // already cover with an injectable cooldown, so this only checks the request shape of a single + // (immediately failing) attempt. + var transport = new MockTransport().QueueResponse(402, MemoryLimitBody); + + var error = await Assert.ThrowsAsync( + () => Client(transport).Actor("act").StartAsync(null, new ActorStartOptions { WaitForResources = Options.WaitForResources.For(0) })); + + Assert.Equal("actor-memory-limit-exceeded", error.Type); + Assert.DoesNotContain("waitForResources", transport.LastRequest.Uri, StringComparison.Ordinal); + } + + [Fact] + public async Task TaskStartWithoutWaitForResourcesThrowsOnFirstResourceLimitError() + { + var transport = new MockTransport().QueueResponse(402, ConcurrentRunsLimitBody); + + var error = await Assert.ThrowsAsync( + () => Client(transport).Task("task1").StartAsync()); + + Assert.Equal("concurrent-runs-limit-exceeded", error.Type); + Assert.Equal(1, transport.CallCount); + } + + [Fact] + public async Task TaskStartNeverSendsWaitForResourcesAsAQueryParameter() + { + var transport = new MockTransport().QueueResponse(402, MemoryLimitBody); + + var error = await Assert.ThrowsAsync( + () => Client(transport).Task("task1").StartAsync(null, new TaskStartOptions { WaitForResources = Options.WaitForResources.For(0) })); + + Assert.Equal("actor-memory-limit-exceeded", error.Type); + Assert.DoesNotContain("waitForResources", transport.LastRequest.Uri, StringComparison.Ordinal); + } +}