diff --git a/apps/mobile/src/features/tasks/api.ts b/apps/mobile/src/features/tasks/api.ts index e2833e505e..f3541e65a0 100644 --- a/apps/mobile/src/features/tasks/api.ts +++ b/apps/mobile/src/features/tasks/api.ts @@ -113,6 +113,9 @@ function isRetryableError(error: unknown): boolean { return error.status >= 500 && error.status < 600; } if (error instanceof Error) { + if (error.name === "AbortError" || error.name === "TimeoutError") { + return true; + } const message = error.message.toLowerCase(); if (message.includes("network")) return true; if (message.includes("timeout")) return true; @@ -719,9 +722,10 @@ export async function fetchSessionLogs( offset: String(options.offset ?? 0), }); + // The server re-reads the whole log chain per page, so big runs need a generous budget. const response = await authedFetch( `${baseUrl}/api/projects/${projectId}/tasks/${taskId}/runs/${runId}/session_logs/?${params}`, - { signal: createTimeoutSignal(10_000) }, + { signal: createTimeoutSignal(120_000) }, ); if (!response.ok) { diff --git a/apps/mobile/src/features/tasks/lib/cloudTaskStream.ts b/apps/mobile/src/features/tasks/lib/cloudTaskStream.ts index 2541eed7e7..66500b2df9 100644 --- a/apps/mobile/src/features/tasks/lib/cloudTaskStream.ts +++ b/apps/mobile/src/features/tasks/lib/cloudTaskStream.ts @@ -805,14 +805,18 @@ async function fetchTaskRunState( /** * Loads the historical log entries for the run, mirroring the desktop's - * dual-source strategy: - * 1. Try the paginated `session_logs/` API — the live source while a run + * strategy: + * 1. Prefer the presigned resume-chain `log_urls` (S3 NDJSON, oldest + * first) when the server provides them. Downloading straight from S3 + * avoids the paginated API's per-page full-chain re-read, which times + * out on runs with very large histories. + * 2. Try the paginated `session_logs/` API — the live source while a run * is active. For older / archived runs this can come back empty even * though the canonical log exists on S3. - * 2. Fall back to the run's presigned `log_url` (S3 NDJSON), which is the + * 3. Fall back to the run's presigned `log_url` (S3 NDJSON), which is the * canonical archive for completed runs. * - * Returns `null` only when both sources fail outright (so the bootstrap can + * Returns `null` only when the sources fail outright (so the bootstrap can * surface a retryable error). An empty paginated result is treated as "no * data yet" and falls through to S3 — if S3 also has nothing we return the * empty array so the snapshot can still flip the session to `"connected"`. @@ -821,6 +825,13 @@ async function fetchHistoricalEntries( watcher: WatcherState, run: TaskRun, ): Promise { + if (run.log_urls?.length) { + const chainEntries = await fetchChainLogEntries(watcher, run.log_urls); + if (watcher.stopped || watcher.failed) return null; + // An all-empty chain read falls through: a misdirected presigned 404 must not mask real data. + if (chainEntries?.length) return chainEntries; + } + const paginated = await fetchAllSessionLogs(watcher); if (watcher.stopped || watcher.failed) return null; if (paginated && paginated.length > 0) return paginated; @@ -837,13 +848,33 @@ async function fetchHistoricalEntries( return paginated ?? null; } +async function fetchChainLogEntries( + watcher: WatcherState, + logUrls: string[], +): Promise { + const entries: StoredLogEntry[] = []; + for (const [index, logUrl] of logUrls.entries()) { + const chunk = await fetchS3LogEntries(watcher, logUrl); + if (watcher.stopped || watcher.failed) return null; + if (chunk === null) return null; + // An empty ancestor object is missing or expired; fall back rather than truncate history. + if (chunk.length === 0 && index < logUrls.length - 1) return null; + // Per-entry push: spreading a huge chunk into push() overflows the engine's argument limit. + for (const entry of chunk) { + entries.push(entry); + } + } + return entries; +} + async function fetchS3LogEntries( watcher: WatcherState, logUrl: string, ): Promise { try { + // RN fetch buffers the whole body, so the budget must cover a full multi-hundred-MB download. const response = await fetch(logUrl, { - signal: createTimeoutSignal(15_000), + signal: createTimeoutSignal(120_000), }); if (response.status === 404) { // No archived log yet for this run — not an error, just no data. diff --git a/apps/mobile/src/features/tasks/types.ts b/apps/mobile/src/features/tasks/types.ts index 18c31142ea..1574976989 100644 --- a/apps/mobile/src/features/tasks/types.ts +++ b/apps/mobile/src/features/tasks/types.ts @@ -64,6 +64,8 @@ export interface TaskRun { environment?: "local" | "cloud"; status: TaskRunStatus; log_url: string; + // Presigned resume-chain log URLs, oldest first; absent on old servers, empty when presigning fails. + log_urls?: string[]; error_message: string | null; reasoning_effort?: string | null; output: Record | null; diff --git a/packages/core/src/cloud-task/cloud-task.test.ts b/packages/core/src/cloud-task/cloud-task.test.ts index c633a95e99..3bd487ea26 100644 --- a/packages/core/src/cloud-task/cloud-task.test.ts +++ b/packages/core/src/cloud-task/cloud-task.test.ts @@ -483,6 +483,327 @@ describe("CloudTaskService", () => { expect(messages()).toEqual(["before retry", "after retry"]); }); + const consoleLogEntry = (message: string) => ({ + type: "notification", + timestamp: "2026-01-01T00:00:01Z", + notification: { + jsonrpc: "2.0", + method: "_posthog/console", + params: { sessionId: "run-1", level: "info", message }, + }, + }); + + it("bootstraps history from presigned chain log urls without touching session_logs", async () => { + const updates: unknown[] = []; + service.on(CloudTaskEvent.Update, (payload) => updates.push(payload)); + + const sessionLogsCalls: string[] = []; + mockNetFetch.mockImplementation((input: string | Request) => { + const url = typeof input === "string" ? input : input.url; + if (url.includes("/session_logs/")) { + sessionLogsCalls.push(url); + return Promise.resolve( + createJsonResponse([], 200, { "X-Has-More": "false" }), + ); + } + if (url.startsWith("https://storage.example/run-0.jsonl")) { + return Promise.resolve( + new Response( + `${JSON.stringify(consoleLogEntry("ancestor 1"))}\n${JSON.stringify(consoleLogEntry("ancestor 2"))}\n`, + ), + ); + } + if (url.startsWith("https://storage.example/run-1.jsonl")) { + // No trailing newline: the final line of a log object is still a complete entry. + return Promise.resolve( + new Response(JSON.stringify(consoleLogEntry("current"))), + ); + } + return Promise.resolve( + createJsonResponse({ + id: "run-1", + status: "completed", + stage: null, + output: null, + error_message: null, + branch: "main", + updated_at: "2026-01-01T00:00:00Z", + log_urls: [ + "https://storage.example/run-0.jsonl?sig=1", + "https://storage.example/run-1.jsonl?sig=2", + ], + }), + ); + }); + + service.watch({ + taskId: "task-1", + runId: "run-1", + apiHost: "https://app.example.com", + teamId: 2, + }); + + await waitFor(() => + updates.some((u) => (u as { kind?: string }).kind === "snapshot"), + ); + + const snapshot = updates.find( + (u) => (u as { kind?: string }).kind === "snapshot", + ) as { newEntries: unknown[]; totalEntryCount: number }; + expect(snapshot.newEntries).toEqual([ + consoleLogEntry("ancestor 1"), + consoleLogEntry("ancestor 2"), + consoleLogEntry("current"), + ]); + expect(snapshot.totalEntryCount).toBe(3); + expect(sessionLogsCalls).toEqual([]); + }); + + it.each([ + ["a chain download fails", 403], + ["every chain object is missing", 404], + ])( + "falls back to the paginated API when %s", + async (_scenario, storageStatus) => { + const updates: unknown[] = []; + service.on(CloudTaskEvent.Update, (payload) => updates.push(payload)); + + mockNetFetch.mockImplementation((input: string | Request) => { + const url = typeof input === "string" ? input : input.url; + if (url.includes("/session_logs/")) { + return Promise.resolve( + createJsonResponse([consoleLogEntry("from api")], 200, { + "X-Has-More": "false", + }), + ); + } + if (url.startsWith("https://storage.example/")) { + return Promise.resolve( + new Response("unavailable", { status: storageStatus }), + ); + } + return Promise.resolve( + createJsonResponse({ + id: "run-1", + status: "completed", + stage: null, + output: null, + error_message: null, + branch: "main", + updated_at: "2026-01-01T00:00:00Z", + log_urls: ["https://storage.example/run-1.jsonl?sig=1"], + }), + ); + }); + + service.watch({ + taskId: "task-1", + runId: "run-1", + apiHost: "https://app.example.com", + teamId: 2, + }); + + await waitFor(() => + updates.some((u) => (u as { kind?: string }).kind === "snapshot"), + ); + + const snapshot = updates.find( + (u) => (u as { kind?: string }).kind === "snapshot", + ) as { newEntries: unknown[] }; + expect(snapshot.newEntries).toEqual([consoleLogEntry("from api")]); + }, + ); + + it("falls back to the paginated API when an ancestor log object is missing", async () => { + const updates: unknown[] = []; + service.on(CloudTaskEvent.Update, (payload) => updates.push(payload)); + + mockNetFetch.mockImplementation((input: string | Request) => { + const url = typeof input === "string" ? input : input.url; + if (url.includes("/session_logs/")) { + return Promise.resolve( + createJsonResponse([consoleLogEntry("from api")], 200, { + "X-Has-More": "false", + }), + ); + } + if (url.startsWith("https://storage.example/run-0.jsonl")) { + return Promise.resolve(new Response("gone", { status: 404 })); + } + if (url.startsWith("https://storage.example/run-1.jsonl")) { + return Promise.resolve( + new Response(JSON.stringify(consoleLogEntry("current"))), + ); + } + return Promise.resolve( + createJsonResponse({ + id: "run-1", + status: "completed", + stage: null, + output: null, + error_message: null, + branch: "main", + updated_at: "2026-01-01T00:00:00Z", + log_urls: [ + "https://storage.example/run-0.jsonl?sig=1", + "https://storage.example/run-1.jsonl?sig=2", + ], + }), + ); + }); + + service.watch({ + taskId: "task-1", + runId: "run-1", + apiHost: "https://app.example.com", + teamId: 2, + }); + + await waitFor(() => + updates.some((u) => (u as { kind?: string }).kind === "snapshot"), + ); + + // A truncated chain (missing ancestor) must not be presented as full history. + const snapshot = updates.find( + (u) => (u as { kind?: string }).kind === "snapshot", + ) as { newEntries: unknown[] }; + expect(snapshot.newEntries).toEqual([consoleLogEntry("from api")]); + }); + + it("shares one history fetch when a subscriber attaches mid-bootstrap", async () => { + const updates: unknown[] = []; + service.on(CloudTaskEvent.Update, (payload) => updates.push(payload)); + + const sessionLogsOffsets: string[] = []; + let releasePage: () => void = () => {}; + mockNetFetch.mockImplementation((input: string | Request) => { + const url = typeof input === "string" ? input : input.url; + if (url.includes("/session_logs/")) { + sessionLogsOffsets.push(new URL(url).searchParams.get("offset") ?? ""); + return new Promise((resolve) => { + releasePage = () => + resolve( + createJsonResponse([consoleLogEntry("only page")], 200, { + "X-Has-More": "false", + }), + ); + }); + } + return Promise.resolve( + createJsonResponse({ + id: "run-1", + status: "in_progress", + stage: null, + output: null, + error_message: null, + branch: "main", + updated_at: "2026-01-01T00:00:00Z", + }), + ); + }); + mockStreamFetch.mockResolvedValue(createOpenSseResponse("")); + + const watchInput = { + taskId: "task-1", + runId: "run-1", + apiHost: "https://app.example.com", + teamId: 2, + }; + service.watch(watchInput); + await waitFor(() => sessionLogsOffsets.length > 0); + service.watch(watchInput); + releasePage(); + + await waitFor( + () => + updates.filter((u) => (u as { kind?: string }).kind === "snapshot") + .length >= 2, + ); + + // A second concurrent page fetch would corrupt the shared resume progress. + expect(sessionLogsOffsets).toEqual(["0"]); + const snapshots = updates.filter( + (u) => (u as { kind?: string }).kind === "snapshot", + ) as Array<{ newEntries: unknown[] }>; + for (const snapshot of snapshots) { + expect(snapshot.newEntries).toEqual([consoleLogEntry("only page")]); + } + }); + + it("resumes paginated history from the last fetched page after a retry", async () => { + vi.useFakeTimers(); + const updates: unknown[] = []; + service.on(CloudTaskEvent.Update, (payload) => updates.push(payload)); + + const sessionLogsOffsets: string[] = []; + let failPageFetches = true; + mockNetFetch.mockImplementation((input: string | Request) => { + const url = typeof input === "string" ? input : input.url; + if (url.includes("/session_logs/")) { + const offset = new URL(url).searchParams.get("offset") ?? ""; + sessionLogsOffsets.push(offset); + if (offset === "0") { + return Promise.resolve( + createJsonResponse([consoleLogEntry("page one")], 200, { + "X-Has-More": "true", + }), + ); + } + if (failPageFetches) { + return Promise.reject(new Error("socket hang up")); + } + return Promise.resolve( + createJsonResponse([consoleLogEntry("page two")], 200, { + "X-Has-More": "false", + }), + ); + } + return Promise.resolve( + createJsonResponse({ + id: "run-1", + status: "completed", + stage: null, + output: null, + error_message: null, + branch: "main", + updated_at: "2026-01-01T00:00:00Z", + }), + ); + }); + + service.watch({ + taskId: "task-1", + runId: "run-1", + apiHost: "https://app.example.com", + teamId: 2, + }); + + // Page 0 succeeds; the offset-1 page fails all retry attempts and the + // watcher surfaces a retryable error instead of a snapshot. + await waitFor( + () => updates.some((u) => (u as { kind?: string }).kind === "error"), + 30_000, + ); + expect(sessionLogsOffsets).toEqual(["0", "1", "1", "1"]); + + failPageFetches = false; + await service.retry("task-1", "run-1"); + await waitFor( + () => updates.some((u) => (u as { kind?: string }).kind === "snapshot"), + 30_000, + ); + + // The retry resumed from offset 1 instead of refetching page 0. + expect(sessionLogsOffsets).toEqual(["0", "1", "1", "1", "1"]); + const snapshot = updates.find( + (u) => (u as { kind?: string }).kind === "snapshot", + ) as { newEntries: unknown[] }; + expect(snapshot.newEntries).toEqual([ + consoleLogEntry("page one"), + consoleLogEntry("page two"), + ]); + }); + it("reconnects with Last-Event-ID after a stream error", async () => { vi.useFakeTimers(); diff --git a/packages/core/src/cloud-task/cloud-task.ts b/packages/core/src/cloud-task/cloud-task.ts index 296f5573e5..4a9c58932e 100644 --- a/packages/core/src/cloud-task/cloud-task.ts +++ b/packages/core/src/cloud-task/cloud-task.ts @@ -8,9 +8,14 @@ import { type IAnalytics, } from "@posthog/platform/analytics"; import type { StoredLogEntry } from "@posthog/shared"; -import { serializeError, TypedEventEmitter } from "@posthog/shared"; +import { + serializeError, + sleepWithBackoff, + TypedEventEmitter, +} from "@posthog/shared"; import { ANALYTICS_EVENTS } from "@posthog/shared/analytics-events"; import { inject, injectable, preDestroy } from "inversify"; +import { parseSessionLogContent } from "../sessions/sessionLogs"; import type { CloudTaskPermissionRequestUpdate } from "./cloud-task-types"; import { CLOUD_TASK_AUTH, type ICloudTaskAuth } from "./identifiers"; import { @@ -37,6 +42,12 @@ const SSE_HEALTHY_CONNECTION_MS = 60_000; const EVENT_BATCH_FLUSH_MS = 16; const EVENT_BATCH_MAX_SIZE = 50; const SESSION_LOG_PAGE_LIMIT = 5_000; +// The server re-reads the whole log chain per page, so big runs need far more than the 30s auth default. +const SESSION_LOG_PAGE_TIMEOUT_MS = 120_000; +const SESSION_LOG_PAGE_RETRIES = 2; +const SESSION_LOG_PAGE_RETRY_DELAY_MS = 2_000; +// Presigned log downloads are unbounded in size, so cap idle time between chunks, not total duration. +const LOG_DOWNLOAD_IDLE_TIMEOUT_MS = 30_000; // Authoritative end-of-stream sentinel, matched on the SSE event name (event.event, not data.type). // The client stops on it without consulting run status. @@ -82,6 +93,8 @@ interface TaskRunResponse { branch?: string | null; updated_at?: string; completed_at?: string | null; + // Presigned resume-chain log URLs, oldest first; absent on old servers, empty when presigning fails. + log_urls?: string[] | null; } interface TaskRunStateEvent { @@ -155,6 +168,18 @@ interface WatcherState { streamReadToken: string | null; // True once stream_token resolved. False for old servers (404), which fall back to status polling. durableStreamEnabled: boolean; + // Presigned resume-chain log URLs from the last run fetch; null on old servers or presign failure. + logUrls: string[] | null; + // Paginated pages already fetched; survives watcher retries (the log is append-only) so a retry resumes. + sessionLogsProgress: StoredLogEntry[] | null; + // Incremented per bootstrapWatcher call so a superseded bootstrap's awaited work is discarded. + bootstrapGeneration: number; + // Single-flight history fetch: concurrent callers of the same generation share one run. + historyFetch: { + generation: number; + promise: Promise; + } | null; + historyAbortController: AbortController | null; } function watcherKey(taskId: string, runId: string): string { @@ -396,6 +421,7 @@ export class CloudTaskService extends TypedEventEmitter { watcher.sseAbortController?.abort(); watcher.sseAbortController = null; + watcher.historyAbortController?.abort(); if (watcher.batchFlushTimeoutId) { clearTimeout(watcher.batchFlushTimeoutId); @@ -439,6 +465,7 @@ export class CloudTaskService extends TypedEventEmitter { watcher.streamBaseUrl = null; watcher.streamReadToken = null; watcher.durableStreamEnabled = false; + // sessionLogsProgress is deliberately retained: the log is append-only, so the retry resumes. } async sendCommand(input: SendCommandInput): Promise { @@ -633,6 +660,11 @@ export class CloudTaskService extends TypedEventEmitter { streamBaseUrl: null, streamReadToken: null, durableStreamEnabled: false, + logUrls: null, + sessionLogsProgress: null, + bootstrapGeneration: 0, + historyFetch: null, + historyAbortController: null, }; this.watchers.set(key, watcher); @@ -645,6 +677,7 @@ export class CloudTaskService extends TypedEventEmitter { if (!watcher) return; watcher.sseAbortController?.abort(); + watcher.historyAbortController?.abort(); if (watcher.reconnectTimeoutId) { clearTimeout(watcher.reconnectTimeoutId); @@ -668,11 +701,13 @@ export class CloudTaskService extends TypedEventEmitter { watcher.failed = false; watcher.needsPostBootstrapReconnect = false; watcher.needsStopAfterBootstrap = false; + watcher.bootstrapGeneration += 1; + const generation = watcher.bootstrapGeneration; const run = await this.fetchTaskRun(watcher); const currentWatcher = this.watchers.get(key); if (!currentWatcher || currentWatcher !== watcher) return; - if (watcher.failed) return; + if (watcher.failed || watcher.bootstrapGeneration !== generation) return; if (!run) { this.failWatcher(key, { @@ -684,6 +719,7 @@ export class CloudTaskService extends TypedEventEmitter { } this.applyTaskRunState(watcher, run); + watcher.logUrls = run.log_urls?.length ? run.log_urls : null; if ( !isTerminalStatus(run.status) && @@ -697,10 +733,13 @@ export class CloudTaskService extends TypedEventEmitter { } if (isTerminalStatus(run.status)) { - const historicalEntries = await this.fetchAllSessionLogs(watcher); + const historicalEntries = await this.fetchHistoricalEntries( + watcher, + generation, + ); const terminalWatcher = this.watchers.get(key); if (!terminalWatcher || terminalWatcher !== watcher) return; - if (watcher.failed) return; + if (watcher.failed || watcher.bootstrapGeneration !== generation) return; if (!historicalEntries) { this.failWatcher(key, { title: "Failed to load task history", @@ -734,10 +773,13 @@ export class CloudTaskService extends TypedEventEmitter { watcher.bufferedLogBatches = []; void this.connectSse(key, { startLatest: true }); - const historicalEntries = await this.fetchAllSessionLogs(watcher); + const historicalEntries = await this.fetchHistoricalEntries( + watcher, + generation, + ); const bootstrappingWatcher = this.watchers.get(key); if (!bootstrappingWatcher || bootstrappingWatcher !== watcher) return; - if (watcher.failed) return; + if (watcher.failed || watcher.bootstrapGeneration !== generation) return; if (!historicalEntries) { this.failWatcher(key, { title: "Failed to load cloud run history", @@ -1339,7 +1381,10 @@ export class CloudTaskService extends TypedEventEmitter { const watcher = this.watchers.get(key); if (!watcher || watcher.failed) return; - const historicalEntries = await this.fetchAllSessionLogs(watcher); + const historicalEntries = await this.fetchHistoricalEntries( + watcher, + watcher.bootstrapGeneration, + ); const currentWatcher = this.watchers.get(key); if (!currentWatcher || currentWatcher !== watcher || watcher.failed) { return; @@ -1423,6 +1468,7 @@ export class CloudTaskService extends TypedEventEmitter { watcher.sseAbortController?.abort(); watcher.sseAbortController = null; + watcher.historyAbortController?.abort(); this.emit(CloudTaskEvent.Update, { taskId: watcher.taskId, @@ -1662,9 +1708,175 @@ export class CloudTaskService extends TypedEventEmitter { return changed; } + // Presigned chain download when available (the paginated API re-reads the whole chain per page). + private fetchHistoricalEntries( + watcher: WatcherState, + generation: number, + ): Promise { + if (watcher.historyFetch?.generation === generation) { + return watcher.historyFetch.promise; + } + + watcher.historyAbortController?.abort(); + const controller = new AbortController(); + watcher.historyAbortController = controller; + + const promise = this.fetchHistoricalEntriesInner( + watcher, + generation, + controller.signal, + ).finally(() => { + if (watcher.historyFetch?.promise === promise) { + watcher.historyFetch = null; + } + if (watcher.historyAbortController === controller) { + watcher.historyAbortController = null; + } + }); + watcher.historyFetch = { generation, promise }; + return promise; + } + + private async fetchHistoricalEntriesInner( + watcher: WatcherState, + generation: number, + signal: AbortSignal, + ): Promise { + if (watcher.logUrls?.length) { + const chainEntries = await this.fetchChainLogEntries( + watcher, + watcher.logUrls, + generation, + signal, + ); + if ( + watcher.bootstrapGeneration !== generation || + watcher.failed || + signal.aborted + ) { + return null; + } + // An all-empty chain read falls through: a misdirected presigned 404 must not mask real data. + if (chainEntries?.length) { + watcher.sessionLogsProgress = null; + return chainEntries; + } + } + + return this.fetchAllSessionLogs(watcher, generation, signal); + } + + private async fetchChainLogEntries( + watcher: WatcherState, + logUrls: string[], + generation: number, + signal: AbortSignal, + ): Promise { + const entries: StoredLogEntry[] = []; + + for (const [index, logUrl] of logUrls.entries()) { + const chunk = await this.downloadLogEntries(watcher, logUrl, signal); + if (watcher.bootstrapGeneration !== generation) return null; + if (chunk === null) return null; + // An empty ancestor object is missing or expired; fall back rather than truncate history. + if (chunk.length === 0 && index < logUrls.length - 1) return null; + for (const entry of chunk) { + entries.push(entry); + } + } + + return entries; + } + + private async downloadLogEntries( + watcher: WatcherState, + logUrl: string, + signal: AbortSignal, + ): Promise { + const controller = new AbortController(); + const onOuterAbort = () => controller.abort(); + signal.addEventListener("abort", onOuterAbort, { once: true }); + if (signal.aborted) controller.abort(); + let idleTimer = setTimeout( + () => controller.abort(), + LOG_DOWNLOAD_IDLE_TIMEOUT_MS, + ); + const resetIdleTimer = () => { + clearTimeout(idleTimer); + idleTimer = setTimeout( + () => controller.abort(), + LOG_DOWNLOAD_IDLE_TIMEOUT_MS, + ); + }; + const entries: StoredLogEntry[] = []; + const parseInto = (segment: string) => { + if (!segment.trim()) return; + for (const entry of parseSessionLogContent(segment).rawEntries) { + entries.push(entry); + } + }; + + try { + // Plain fetch: S3 rejects presigned requests that also carry an Authorization header. + const response = await fetch(logUrl, { + method: "GET", + signal: controller.signal, + }); + + if (response.status === 404) { + // The run has not written this log object yet. + return entries; + } + + if (!response.ok) { + this.log.warn("Cloud task log object download failed", { + taskId: watcher.taskId, + runId: watcher.runId, + status: response.status, + }); + return null; + } + + if (!response.body) { + parseInto(await response.text()); + return entries; + } + + resetIdleTimer(); + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let pending = ""; + while (true) { + const { done, value } = await reader.read(); + if (done) break; + resetIdleTimer(); + pending += decoder.decode(value, { stream: true }); + const lastNewline = pending.lastIndexOf("\n"); + if (lastNewline === -1) continue; + // Parse completed lines as they arrive so memory stays O(entries), not O(raw log bytes). + parseInto(pending.slice(0, lastNewline)); + pending = pending.slice(lastNewline + 1); + } + pending += decoder.decode(); + parseInto(pending); + return entries; + } catch (error) { + this.log.warn("Cloud task log object download error", { + taskId: watcher.taskId, + runId: watcher.runId, + error, + }); + return null; + } finally { + clearTimeout(idleTimer); + signal.removeEventListener("abort", onOuterAbort); + } + } + private async fetchSessionLogsPage( watcher: WatcherState, offset: number, + signal: AbortSignal, ): Promise { const url = new URL( `${watcher.apiHost}/api/projects/${watcher.teamId}/tasks/${watcher.taskId}/runs/${watcher.runId}/session_logs/`, @@ -1677,6 +1889,10 @@ export class CloudTaskService extends TypedEventEmitter { url.toString(), { method: "GET", + signal: AbortSignal.any([ + signal, + AbortSignal.timeout(SESSION_LOG_PAGE_TIMEOUT_MS), + ]), }, ); @@ -1712,24 +1928,68 @@ export class CloudTaskService extends TypedEventEmitter { } } + private async fetchSessionLogsPageWithRetry( + watcher: WatcherState, + offset: number, + generation: number, + signal: AbortSignal, + ): Promise { + for (let attempt = 0; attempt <= SESSION_LOG_PAGE_RETRIES; attempt++) { + if (attempt > 0) { + await sleepWithBackoff( + attempt - 1, + { initialDelayMs: SESSION_LOG_PAGE_RETRY_DELAY_MS }, + signal, + ); + if ( + watcher.bootstrapGeneration !== generation || + watcher.failed || + signal.aborted + ) { + return null; + } + } + const page = await this.fetchSessionLogsPage(watcher, offset, signal); + if (page) return page; + if ( + watcher.bootstrapGeneration !== generation || + watcher.failed || + signal.aborted + ) { + return null; + } + } + return null; + } + private async fetchAllSessionLogs( watcher: WatcherState, + generation: number, + signal: AbortSignal, ): Promise { - const entries: StoredLogEntry[] = []; - let offset = 0; + const entries = watcher.sessionLogsProgress ?? []; + watcher.sessionLogsProgress = entries; while (true) { - const page = await this.fetchSessionLogsPage(watcher, offset); + const page = await this.fetchSessionLogsPageWithRetry( + watcher, + entries.length, + generation, + signal, + ); + if (watcher.bootstrapGeneration !== generation) return null; if (!page) { + // Progress is kept so the next watcher retry resumes from this offset. return null; } - entries.push(...page.entries); + for (const entry of page.entries) { + entries.push(entry); + } if (!page.hasMore || page.entries.length === 0) { + watcher.sessionLogsProgress = null; return entries; } - - offset += page.entries.length; } }