diff --git a/chat-sdk/src/chat.ts b/chat-sdk/src/chat.ts index a5d08f4734..892e4e6b09 100644 --- a/chat-sdk/src/chat.ts +++ b/chat-sdk/src/chat.ts @@ -682,22 +682,29 @@ class ChatImpl implements Chat { * The flow job plus every step job it ran, the failure and preprocessor steps * included (a failure handler's answer is persisted under its own job), and the * jobs an agent step's tool calls ran as (a tool row is persisted under its own - * job too). Unknown when the read fails. + * job too). A failed read is retried like the rows are; unknown when it keeps + * failing or the credential may not read jobs. Unknown accepts every row after the + * question: refusing them would leave a token without job access with no turn ever + * answered, each one finished a second time from its result. */ async #turnJobIds(turn: Turn): Promise | undefined> { - try { - const job = await this.#api.getFlowJob(turn.jobId!, turn.controller.signal) - const ids = new Set([turn.jobId!]) - const status = job.flow_status - for (const m of [...(status?.modules ?? []), status?.failure_module, status?.preprocessor_module]) { - if (m?.job) ids.add(m.job) - for (const j of m?.flow_jobs ?? []) ids.add(j) - for (const a of m?.agent_actions ?? []) if (a.job_id) ids.add(a.job_id) + for (let attempt = 1; ; attempt++) { + try { + const job = await this.#api.getFlowJob(turn.jobId!, turn.controller.signal) + const ids = new Set([turn.jobId!]) + const status = job.flow_status + for (const m of [...(status?.modules ?? []), status?.failure_module, status?.preprocessor_module]) { + if (m?.job) ids.add(m.job) + for (const j of m?.flow_jobs ?? []) ids.add(j) + for (const a of m?.agent_actions ?? []) if (a.job_id) ids.add(a.job_id) + } + return ids + } catch (e) { + if (isAbortError(e)) throw e + const refused = e instanceof WindmillApiError && e.status >= 400 && e.status < 500 + if (refused || attempt === RECONCILE_ATTEMPTS) return undefined + await sleep(RECONCILE_DELAY_MS, turn.controller.signal) } - return ids - } catch (e) { - if (isAbortError(e)) throw e - return undefined } } diff --git a/chat-sdk/src/follow.ts b/chat-sdk/src/follow.ts index 2908f645df..14511f1e2f 100644 --- a/chat-sdk/src/follow.ts +++ b/chat-sdk/src/follow.ts @@ -41,8 +41,10 @@ export async function* followJob( let reopen = false try { for await (const update of api.streamJob(jobId, { streamOffset: offset, signal: options.signal })) { - failures = 0 + // A ping proves the connection opened, not that it carries the job: only an update + // clears the count, or a connection that pings and drops would never reach polling. if (update.type === 'ping') continue + failures = 0 if (update.type === 'timeout') { reopen = true break diff --git a/chat-sdk/test/chat.test.ts b/chat-sdk/test/chat.test.ts index 7b7bb8d961..47ebef9d76 100644 --- a/chat-sdk/test/chat.test.ts +++ b/chat-sdk/test/chat.test.ts @@ -1005,13 +1005,20 @@ describe('createChat with server history', () => { test("the previous run's answer landing after the next message is not that turn's answer", async () => { let reads = 0 + let jobReads = 0 const { fetch } = fetchMock( run, (c) => c.url.pathname === streamPath ? sse([{ type: 'update', completed: true, only_result: { windmill_chat_answer: 'second answer' } }]) : undefined, - (c) => (c.url.pathname.endsWith('/jobs_u/get/job-1') ? json({ flow_status: { modules: [{ job: 'step-2' }] } }) : undefined), + // A transient failure of the job read is retried, not taken as "any row counts". + (c) => + c.url.pathname.endsWith('/jobs_u/get/job-1') + ? ++jobReads === 1 + ? text('bad gateway', 502) + : json({ flow_status: { modules: [{ job: 'step-2' }] } }) + : undefined, (c) => { if (!c.url.pathname.endsWith('/messages')) return undefined if (!c.url.searchParams.has('after_seq')) return json([messageRow(71, 'user', 'first')]) @@ -1058,9 +1065,20 @@ describe('createChat with server history', () => { }, 15000) test('a stream that keeps failing hands the turn to polling the job', async () => { + // Each connection opens, sends the server's ping, then drops. + const pingThenDrop = () => + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(`data: ${JSON.stringify({ type: 'ping' })}\n\n`)) + setTimeout(() => controller.error(new TypeError('network connection was lost')), 20) + } + }), + { status: 200, headers: { 'content-type': 'text/event-stream' } } + ) const { fetch } = fetchMock( run, - (c) => (c.url.pathname === streamPath ? text('bad gateway', 502) : undefined), + (c) => (c.url.pathname === streamPath ? pingThenDrop() : undefined), (c) => c.url.pathname.endsWith('/get_result_maybe/job-1') ? json({ completed: true, success: true, result: { windmill_chat_answer: 'polled' } })