From 6d588133f1b27285bef4f6f2e76b405db443bcb6 Mon Sep 17 00:00:00 2001 From: Charles Vien Date: Wed, 15 Jul 2026 17:59:17 -0700 Subject: [PATCH 1/3] load cloud run history from presigned log urls --- apps/mobile/src/features/tasks/api.ts | 4 +- .../src/features/tasks/lib/cloudTaskStream.ts | 36 ++- apps/mobile/src/features/tasks/types.ts | 3 + .../core/src/cloud-task/cloud-task.test.ts | 197 ++++++++++++++++ packages/core/src/cloud-task/cloud-task.ts | 215 ++++++++++++++++-- 5 files changed, 435 insertions(+), 20 deletions(-) diff --git a/apps/mobile/src/features/tasks/api.ts b/apps/mobile/src/features/tasks/api.ts index e2833e505e..08638dbe19 100644 --- a/apps/mobile/src/features/tasks/api.ts +++ b/apps/mobile/src/features/tasks/api.ts @@ -719,9 +719,11 @@ export async function fetchSessionLogs( offset: String(options.offset ?? 0), }); + // Big runs can take the server a long time per page (it re-reads the whole + // log chain each request), so this needs far more than the default 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..1b501f30ea 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,12 @@ 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; + if (chainEntries) return chainEntries; + } + const paginated = await fetchAllSessionLogs(watcher); if (watcher.stopped || watcher.failed) return null; if (paginated && paginated.length > 0) return paginated; @@ -837,13 +847,29 @@ async function fetchHistoricalEntries( return paginated ?? null; } +async function fetchChainLogEntries( + watcher: WatcherState, + logUrls: string[], +): Promise { + const entries: StoredLogEntry[] = []; + for (const logUrl of logUrls) { + const chunk = await fetchS3LogEntries(watcher, logUrl); + if (watcher.stopped || watcher.failed) return null; + if (chunk === null) return null; + entries.push(...chunk); + } + return entries; +} + async function fetchS3LogEntries( watcher: WatcherState, logUrl: string, ): Promise { try { + // Chain logs of long-running tasks can be hundreds of MB; RN fetch buffers + // the whole body, so the budget covers the full download, not just TTFB. 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..3f65642e32 100644 --- a/apps/mobile/src/features/tasks/types.ts +++ b/apps/mobile/src/features/tasks/types.ts @@ -64,6 +64,9 @@ export interface TaskRun { environment?: "local" | "cloud"; status: TaskRunStatus; log_url: string; + /** Presigned S3 URLs for every log in the run's resume chain, oldest first. + * Absent on old servers; empty when the server can't presign. */ + 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..5e143284e8 100644 --- a/packages/core/src/cloud-task/cloud-task.test.ts +++ b/packages/core/src/cloud-task/cloud-task.test.ts @@ -483,6 +483,203 @@ 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("falls back to the paginated API when a chain log download fails", 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/")) { + return Promise.resolve(new Response("expired", { status: 403 })); + } + 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("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..5c880519be 100644 --- a/packages/core/src/cloud-task/cloud-task.ts +++ b/packages/core/src/cloud-task/cloud-task.ts @@ -8,7 +8,11 @@ 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 type { CloudTaskPermissionRequestUpdate } from "./cloud-task-types"; @@ -37,6 +41,15 @@ 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; +// Session-logs pages of a big run can take the server well past the 30s default +// authenticatedFetch timeout (it re-reads the whole log chain per page), so give +// each page an explicit, generous budget and retry transient failures. +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 +// rather than 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 +95,9 @@ interface TaskRunResponse { branch?: string | null; updated_at?: string; completed_at?: string | null; + /** Presigned S3 URLs for every log in the run's resume chain, oldest first. + * Absent on old servers; empty when the server can't presign. */ + log_urls?: string[] | null; } interface TaskRunStateEvent { @@ -155,6 +171,13 @@ 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 history progress that survives watcher retries: the log is append-only, so pages + * already fetched stay valid and a retry resumes from `offset` instead of restarting at 0. */ + sessionLogsProgress: { offset: number; entries: StoredLogEntry[] } | null; + // Incremented per bootstrapWatcher call so a superseded bootstrap's awaited work is discarded. + bootstrapGeneration: number; } function watcherKey(taskId: string, runId: string): string { @@ -439,6 +462,8 @@ export class CloudTaskService extends TypedEventEmitter { watcher.streamBaseUrl = null; watcher.streamReadToken = null; watcher.durableStreamEnabled = false; + // sessionLogsProgress is deliberately retained: the log is append-only, so history + // pages fetched before the failure stay valid and the retry resumes where it left off. } async sendCommand(input: SendCommandInput): Promise { @@ -633,6 +658,9 @@ export class CloudTaskService extends TypedEventEmitter { streamBaseUrl: null, streamReadToken: null, durableStreamEnabled: false, + logUrls: null, + sessionLogsProgress: null, + bootstrapGeneration: 0, }; this.watchers.set(key, watcher); @@ -668,11 +696,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 +714,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 +728,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 +768,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 +1376,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; @@ -1662,6 +1702,122 @@ export class CloudTaskService extends TypedEventEmitter { return changed; } + // Loads a run's full history: straight from the presigned resume-chain log objects when the + // server provides them, else via the paginated session_logs API. The direct download avoids + // the API's per-page full-chain re-read, which times out on runs with very large histories. + private async fetchHistoricalEntries( + watcher: WatcherState, + generation: number, + ): Promise { + if (watcher.logUrls?.length) { + const chainEntries = await this.fetchChainLogEntries( + watcher, + watcher.logUrls, + generation, + ); + if (watcher.bootstrapGeneration !== generation) return null; + if (chainEntries) { + watcher.sessionLogsProgress = null; + return chainEntries; + } + } + + return this.fetchAllSessionLogs(watcher, generation); + } + + private async fetchChainLogEntries( + watcher: WatcherState, + logUrls: string[], + generation: number, + ): Promise { + const entries: StoredLogEntry[] = []; + + for (const logUrl of logUrls) { + const content = await this.downloadLogObject(watcher, logUrl); + if (watcher.bootstrapGeneration !== generation) return null; + if (content === null) return null; + + for (const line of content.split("\n")) { + const trimmed = line.trim(); + if (!trimmed) continue; + try { + entries.push(JSON.parse(trimmed) as StoredLogEntry); + } catch { + // Malformed lines are skipped, matching the server-side JSONL parser. + } + } + } + + return entries; + } + + private async downloadLogObject( + watcher: WatcherState, + logUrl: string, + ): Promise { + const controller = new AbortController(); + let idleTimer = setTimeout( + () => controller.abort(), + LOG_DOWNLOAD_IDLE_TIMEOUT_MS, + ); + const resetIdleTimer = () => { + clearTimeout(idleTimer); + idleTimer = setTimeout( + () => controller.abort(), + LOG_DOWNLOAD_IDLE_TIMEOUT_MS, + ); + }; + + try { + // Presigned URL: plain fetch. S3 rejects requests that carry both query + // signing and an Authorization header, so this must not go through auth. + const response = await fetch(logUrl, { + method: "GET", + signal: controller.signal, + }); + + if (response.status === 404) { + // The run has not written its log object yet. + return ""; + } + + 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) { + return await response.text(); + } + + resetIdleTimer(); + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let content = ""; + while (true) { + const { done, value } = await reader.read(); + if (done) break; + resetIdleTimer(); + content += decoder.decode(value, { stream: true }); + } + content += decoder.decode(); + return content; + } catch (error) { + this.log.warn("Cloud task log object download error", { + taskId: watcher.taskId, + runId: watcher.runId, + error, + }); + return null; + } finally { + clearTimeout(idleTimer); + } + } + private async fetchSessionLogsPage( watcher: WatcherState, offset: number, @@ -1677,6 +1833,7 @@ export class CloudTaskService extends TypedEventEmitter { url.toString(), { method: "GET", + signal: AbortSignal.timeout(SESSION_LOG_PAGE_TIMEOUT_MS), }, ); @@ -1712,24 +1869,54 @@ export class CloudTaskService extends TypedEventEmitter { } } + private async fetchSessionLogsPageWithRetry( + watcher: WatcherState, + offset: number, + generation: number, + ): 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, + }); + if (watcher.bootstrapGeneration !== generation || watcher.failed) { + return null; + } + } + const page = await this.fetchSessionLogsPage(watcher, offset); + if (page) return page; + if (watcher.bootstrapGeneration !== generation || watcher.failed) { + return null; + } + } + return null; + } + private async fetchAllSessionLogs( watcher: WatcherState, + generation: number, ): Promise { - const entries: StoredLogEntry[] = []; - let offset = 0; + const progress = watcher.sessionLogsProgress ?? { offset: 0, entries: [] }; + watcher.sessionLogsProgress = progress; while (true) { - const page = await this.fetchSessionLogsPage(watcher, offset); + const page = await this.fetchSessionLogsPageWithRetry( + watcher, + progress.offset, + generation, + ); + 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); + progress.entries.push(...page.entries); + progress.offset += page.entries.length; if (!page.hasMore || page.entries.length === 0) { - return entries; + watcher.sessionLogsProgress = null; + return progress.entries; } - - offset += page.entries.length; } } From f1f1e072a9a9e12e792726d0a98f9d3bdcf5ba98 Mon Sep 17 00:00:00 2001 From: Charles Vien Date: Wed, 15 Jul 2026 18:45:13 -0700 Subject: [PATCH 2/3] harden history fetch after review --- apps/mobile/src/features/tasks/api.ts | 6 +- .../src/features/tasks/lib/cloudTaskStream.ts | 11 +- apps/mobile/src/features/tasks/types.ts | 3 +- .../core/src/cloud-task/cloud-task.test.ts | 104 ++++++++-- packages/core/src/cloud-task/cloud-task.ts | 179 ++++++++++++------ 5 files changed, 223 insertions(+), 80 deletions(-) diff --git a/apps/mobile/src/features/tasks/api.ts b/apps/mobile/src/features/tasks/api.ts index 08638dbe19..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,8 +722,7 @@ export async function fetchSessionLogs( offset: String(options.offset ?? 0), }); - // Big runs can take the server a long time per page (it re-reads the whole - // log chain each request), so this needs far more than the default budget. + // 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(120_000) }, diff --git a/apps/mobile/src/features/tasks/lib/cloudTaskStream.ts b/apps/mobile/src/features/tasks/lib/cloudTaskStream.ts index 1b501f30ea..3b9f8fe141 100644 --- a/apps/mobile/src/features/tasks/lib/cloudTaskStream.ts +++ b/apps/mobile/src/features/tasks/lib/cloudTaskStream.ts @@ -828,7 +828,8 @@ async function fetchHistoricalEntries( if (run.log_urls?.length) { const chainEntries = await fetchChainLogEntries(watcher, run.log_urls); if (watcher.stopped || watcher.failed) return null; - if (chainEntries) return chainEntries; + // 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); @@ -856,7 +857,10 @@ async function fetchChainLogEntries( const chunk = await fetchS3LogEntries(watcher, logUrl); if (watcher.stopped || watcher.failed) return null; if (chunk === null) return null; - entries.push(...chunk); + // 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; } @@ -866,8 +870,7 @@ async function fetchS3LogEntries( logUrl: string, ): Promise { try { - // Chain logs of long-running tasks can be hundreds of MB; RN fetch buffers - // the whole body, so the budget covers the full download, not just TTFB. + // RN fetch buffers the whole body, so the budget must cover a full multi-hundred-MB download. const response = await fetch(logUrl, { signal: createTimeoutSignal(120_000), }); diff --git a/apps/mobile/src/features/tasks/types.ts b/apps/mobile/src/features/tasks/types.ts index 3f65642e32..1574976989 100644 --- a/apps/mobile/src/features/tasks/types.ts +++ b/apps/mobile/src/features/tasks/types.ts @@ -64,8 +64,7 @@ export interface TaskRun { environment?: "local" | "cloud"; status: TaskRunStatus; log_url: string; - /** Presigned S3 URLs for every log in the run's resume chain, oldest first. - * Absent on old servers; empty when the server can't presign. */ + // 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; diff --git a/packages/core/src/cloud-task/cloud-task.test.ts b/packages/core/src/cloud-task/cloud-task.test.ts index 5e143284e8..8ecf170eff 100644 --- a/packages/core/src/cloud-task/cloud-task.test.ts +++ b/packages/core/src/cloud-task/cloud-task.test.ts @@ -559,51 +559,119 @@ describe("CloudTaskService", () => { expect(sessionLogsCalls).toEqual([]); }); - it("falls back to the paginated API when a chain log download fails", async () => { + 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("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/")) { - return Promise.resolve( - createJsonResponse([consoleLogEntry("from api")], 200, { - "X-Has-More": "false", - }), - ); - } - if (url.startsWith("https://storage.example/")) { - return Promise.resolve(new Response("expired", { status: 403 })); + 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: "completed", + status: "in_progress", 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"], }), ); }); + mockStreamFetch.mockResolvedValue(createOpenSseResponse("")); - service.watch({ + 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.some((u) => (u as { kind?: string }).kind === "snapshot"), + await waitFor( + () => + updates.filter((u) => (u as { kind?: string }).kind === "snapshot") + .length >= 2, ); - const snapshot = updates.find( + // 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 { newEntries: unknown[] }; - expect(snapshot.newEntries).toEqual([consoleLogEntry("from api")]); + ) 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 () => { diff --git a/packages/core/src/cloud-task/cloud-task.ts b/packages/core/src/cloud-task/cloud-task.ts index 5c880519be..224caaf20c 100644 --- a/packages/core/src/cloud-task/cloud-task.ts +++ b/packages/core/src/cloud-task/cloud-task.ts @@ -15,6 +15,7 @@ import { } 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 { @@ -41,14 +42,11 @@ 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; -// Session-logs pages of a big run can take the server well past the 30s default -// authenticatedFetch timeout (it re-reads the whole log chain per page), so give -// each page an explicit, generous budget and retry transient failures. +// 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 -// rather than total duration. +// 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). @@ -95,8 +93,7 @@ interface TaskRunResponse { branch?: string | null; updated_at?: string; completed_at?: string | null; - /** Presigned S3 URLs for every log in the run's resume chain, oldest first. - * Absent on old servers; empty when the server can't presign. */ + // Presigned resume-chain log URLs, oldest first; absent on old servers, empty when presigning fails. log_urls?: string[] | null; } @@ -173,11 +170,16 @@ interface WatcherState { durableStreamEnabled: boolean; // Presigned resume-chain log URLs from the last run fetch; null on old servers or presign failure. logUrls: string[] | null; - /** Paginated history progress that survives watcher retries: the log is append-only, so pages - * already fetched stay valid and a retry resumes from `offset` instead of restarting at 0. */ - sessionLogsProgress: { offset: number; entries: StoredLogEntry[] } | 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 { @@ -419,6 +421,7 @@ export class CloudTaskService extends TypedEventEmitter { watcher.sseAbortController?.abort(); watcher.sseAbortController = null; + watcher.historyAbortController?.abort(); if (watcher.batchFlushTimeoutId) { clearTimeout(watcher.batchFlushTimeoutId); @@ -462,8 +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 history - // pages fetched before the failure stay valid and the retry resumes where it left off. + // sessionLogsProgress is deliberately retained: the log is append-only, so the retry resumes. } async sendCommand(input: SendCommandInput): Promise { @@ -661,6 +663,8 @@ export class CloudTaskService extends TypedEventEmitter { logUrls: null, sessionLogsProgress: null, bootstrapGeneration: 0, + historyFetch: null, + historyAbortController: null, }; this.watchers.set(key, watcher); @@ -673,6 +677,7 @@ export class CloudTaskService extends TypedEventEmitter { if (!watcher) return; watcher.sseAbortController?.abort(); + watcher.historyAbortController?.abort(); if (watcher.reconnectTimeoutId) { clearTimeout(watcher.reconnectTimeoutId); @@ -1463,6 +1468,7 @@ export class CloudTaskService extends TypedEventEmitter { watcher.sseAbortController?.abort(); watcher.sseAbortController = null; + watcher.historyAbortController?.abort(); this.emit(CloudTaskEvent.Update, { taskId: watcher.taskId, @@ -1702,60 +1708,93 @@ export class CloudTaskService extends TypedEventEmitter { return changed; } - // Loads a run's full history: straight from the presigned resume-chain log objects when the - // server provides them, else via the paginated session_logs API. The direct download avoids - // the API's per-page full-chain re-read, which times out on runs with very large histories. - private async fetchHistoricalEntries( + // 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) return null; - if (chainEntries) { + 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); + return this.fetchAllSessionLogs(watcher, generation, signal); } private async fetchChainLogEntries( watcher: WatcherState, logUrls: string[], generation: number, + signal: AbortSignal, ): Promise { const entries: StoredLogEntry[] = []; for (const logUrl of logUrls) { - const content = await this.downloadLogObject(watcher, logUrl); + const chunk = await this.downloadLogEntries(watcher, logUrl, signal); if (watcher.bootstrapGeneration !== generation) return null; - if (content === null) return null; - - for (const line of content.split("\n")) { - const trimmed = line.trim(); - if (!trimmed) continue; - try { - entries.push(JSON.parse(trimmed) as StoredLogEntry); - } catch { - // Malformed lines are skipped, matching the server-side JSONL parser. - } + if (chunk === null) return null; + for (const entry of chunk) { + entries.push(entry); } } return entries; } - private async downloadLogObject( + private async downloadLogEntries( watcher: WatcherState, logUrl: string, - ): Promise { + 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, @@ -1767,18 +1806,24 @@ export class CloudTaskService extends TypedEventEmitter { 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 { - // Presigned URL: plain fetch. S3 rejects requests that carry both query - // signing and an Authorization header, so this must not go through auth. + // 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 its log object yet. - return ""; + // The run has not written this log object yet. + return entries; } if (!response.ok) { @@ -1791,21 +1836,28 @@ export class CloudTaskService extends TypedEventEmitter { } if (!response.body) { - return await response.text(); + parseInto(await response.text()); + return entries; } resetIdleTimer(); const reader = response.body.getReader(); const decoder = new TextDecoder(); - let content = ""; + let pending = ""; while (true) { const { done, value } = await reader.read(); if (done) break; resetIdleTimer(); - content += decoder.decode(value, { stream: true }); + 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); } - content += decoder.decode(); - return content; + pending += decoder.decode(); + parseInto(pending); + return entries; } catch (error) { this.log.warn("Cloud task log object download error", { taskId: watcher.taskId, @@ -1815,12 +1867,14 @@ export class CloudTaskService extends TypedEventEmitter { 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/`, @@ -1833,7 +1887,10 @@ export class CloudTaskService extends TypedEventEmitter { url.toString(), { method: "GET", - signal: AbortSignal.timeout(SESSION_LOG_PAGE_TIMEOUT_MS), + signal: AbortSignal.any([ + signal, + AbortSignal.timeout(SESSION_LOG_PAGE_TIMEOUT_MS), + ]), }, ); @@ -1873,19 +1930,30 @@ export class CloudTaskService extends TypedEventEmitter { 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, - }); - if (watcher.bootstrapGeneration !== generation || watcher.failed) { + 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); + const page = await this.fetchSessionLogsPage(watcher, offset, signal); if (page) return page; - if (watcher.bootstrapGeneration !== generation || watcher.failed) { + if ( + watcher.bootstrapGeneration !== generation || + watcher.failed || + signal.aborted + ) { return null; } } @@ -1895,15 +1963,17 @@ export class CloudTaskService extends TypedEventEmitter { private async fetchAllSessionLogs( watcher: WatcherState, generation: number, + signal: AbortSignal, ): Promise { - const progress = watcher.sessionLogsProgress ?? { offset: 0, entries: [] }; - watcher.sessionLogsProgress = progress; + const entries = watcher.sessionLogsProgress ?? []; + watcher.sessionLogsProgress = entries; while (true) { const page = await this.fetchSessionLogsPageWithRetry( watcher, - progress.offset, + entries.length, generation, + signal, ); if (watcher.bootstrapGeneration !== generation) return null; if (!page) { @@ -1911,11 +1981,12 @@ export class CloudTaskService extends TypedEventEmitter { return null; } - progress.entries.push(...page.entries); - progress.offset += page.entries.length; + for (const entry of page.entries) { + entries.push(entry); + } if (!page.hasMore || page.entries.length === 0) { watcher.sessionLogsProgress = null; - return progress.entries; + return entries; } } } From b4148ade7f810d5cf041ed70daf328561fabf6f4 Mon Sep 17 00:00:00 2001 From: Charles Vien Date: Wed, 15 Jul 2026 18:48:10 -0700 Subject: [PATCH 3/3] fall back when an ancestor log object is missing --- .../src/features/tasks/lib/cloudTaskStream.ts | 4 +- .../core/src/cloud-task/cloud-task.test.ts | 56 +++++++++++++++++++ packages/core/src/cloud-task/cloud-task.ts | 4 +- 3 files changed, 62 insertions(+), 2 deletions(-) diff --git a/apps/mobile/src/features/tasks/lib/cloudTaskStream.ts b/apps/mobile/src/features/tasks/lib/cloudTaskStream.ts index 3b9f8fe141..66500b2df9 100644 --- a/apps/mobile/src/features/tasks/lib/cloudTaskStream.ts +++ b/apps/mobile/src/features/tasks/lib/cloudTaskStream.ts @@ -853,10 +853,12 @@ async function fetchChainLogEntries( logUrls: string[], ): Promise { const entries: StoredLogEntry[] = []; - for (const logUrl of logUrls) { + 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); diff --git a/packages/core/src/cloud-task/cloud-task.test.ts b/packages/core/src/cloud-task/cloud-task.test.ts index 8ecf170eff..3bd487ea26 100644 --- a/packages/core/src/cloud-task/cloud-task.test.ts +++ b/packages/core/src/cloud-task/cloud-task.test.ts @@ -614,6 +614,62 @@ describe("CloudTaskService", () => { }, ); + 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)); diff --git a/packages/core/src/cloud-task/cloud-task.ts b/packages/core/src/cloud-task/cloud-task.ts index 224caaf20c..4a9c58932e 100644 --- a/packages/core/src/cloud-task/cloud-task.ts +++ b/packages/core/src/cloud-task/cloud-task.ts @@ -1774,10 +1774,12 @@ export class CloudTaskService extends TypedEventEmitter { ): Promise { const entries: StoredLogEntry[] = []; - for (const logUrl of logUrls) { + 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); }