diff --git a/changelog.d/fixes/7794-pipeline-transient-retry.md b/changelog.d/fixes/7794-pipeline-transient-retry.md new file mode 100644 index 0000000000..aa6826ce6c --- /dev/null +++ b/changelog.d/fixes/7794-pipeline-transient-retry.md @@ -0,0 +1 @@ +- fix(combo): retry intermediate pipeline-strategy steps on transient upstream errors (429/502/503/504), gated by `combo.config.maxRetries`/`retryDelayMs`, without retrying terminal auth/request errors (#7794 — thanks @AndrianBalanescu) diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index 4944a3d8e9..519465300c 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -927,6 +927,8 @@ export async function handleComboChat({ handleSingleModel: handleSingleModelWithTimeout, log, comboName: combo.name, + maxRetries: config.maxRetries ?? 0, + retryDelayMs: resolveDelayMs(config.retryDelayMs, 1000), }); } diff --git a/open-sse/services/pipeline.ts b/open-sse/services/pipeline.ts index 3c1c86b06d..03dec5dc94 100644 --- a/open-sse/services/pipeline.ts +++ b/open-sse/services/pipeline.ts @@ -30,6 +30,14 @@ * A step failure fails the whole pipeline EXPLICITLY (never silently swallowed): * a non-OK intermediate response, an unparseable body, or an intermediate step that * yields no text short-circuits with a sanitized error response. + * + * ── Transient retry ────────────────────────────────────────────────────────── + * Intermediate steps that fail with a transient HTTP status (429, 502, 503, 504) + * are retried up to `maxRetries` times with `retryDelayMs` delay between attempts. + * This mirrors the retry behaviour already used by the priority/weighted strategies + * and respects the same `combo.config.maxRetries` / `combo.config.retryDelayMs` + * fields. Non-transient errors (400, 401, 403, 404, …) fail immediately — retrying + * a bad-request or auth error wastes quota and will never succeed. */ import { errorResponse } from "../utils/error.ts"; import type { ComboLogger, HandleSingleModel } from "./combo/types.ts"; @@ -112,8 +120,19 @@ export type HandlePipelineChatOptions = { handleSingleModel: HandleSingleModel; log: ComboLogger; comboName?: string; + /** Max retry attempts on transient errors (429/502/503/504). Default: 0 (no retry). */ + maxRetries?: number; + /** Delay between retries in milliseconds. Default: 1000. */ + retryDelayMs?: number; }; +/** HTTP statuses that are worth retrying (transient / capacity / rate-limit). */ +const TRANSIENT_STATUS = new Set([429, 502, 503, 504]); + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + /** * Handle a pipeline combo: run the steps in order, threading each step's output * into the next step's input, and return only the final step's response. @@ -124,6 +143,8 @@ export async function handlePipelineChat({ handleSingleModel, log, comboName, + maxRetries = 0, + retryDelayMs = 1000, }: HandlePipelineChatOptions): Promise { const chain = (Array.isArray(steps) ? steps : []).filter((s) => s && s.model); if (chain.length === 0) { @@ -153,13 +174,25 @@ export async function handlePipelineChat({ if (!isFinal) stepBody = stripStreaming(stepBody); const t0 = Date.now(); - const res = await handleSingleModel(stepBody, step.model); + let res = await handleSingleModel(stepBody, step.model); if (isFinal) { log.info("PIPELINE", `Final step ${step.model} responded (${Date.now() - t0}ms)`); return res; } + // Transient retry: if the intermediate step failed with a retryable status + // (429/502/503/504), retry the same step up to maxRetries times before + // giving up. Non-transient errors (400/401/403/404) fail immediately. + for (let attempt = 0; attempt < maxRetries && !res.ok && TRANSIENT_STATUS.has(res.status); attempt++) { + log.warn( + "PIPELINE", + `Step ${i + 1} (${step.model}) transient ${res.status}, retrying ${attempt + 1}/${maxRetries} in ${retryDelayMs}ms` + ); + await sleep(retryDelayMs); + res = await handleSingleModel(stepBody, step.model); + } + // An intermediate step must succeed with usable text — otherwise fail the whole // pipeline (never silently swallow; the client gets a clear, sanitized error). if (!res.ok) { diff --git a/tests/unit/combo-pipeline.test.ts b/tests/unit/combo-pipeline.test.ts new file mode 100644 index 0000000000..bf52c34aeb --- /dev/null +++ b/tests/unit/combo-pipeline.test.ts @@ -0,0 +1,198 @@ +/** + * Combo Pipeline Strategy Tests + * + * Tests for open-sse/services/pipeline.ts — the sequential chain combo strategy. + * Focus: transient retry behaviour (429/502/503/504) added to prevent hard-fail + * on rate-limited or temporarily unavailable upstream providers. + */ + +import { describe, it, mock } from "node:test"; +import assert from "node:assert/strict"; + +import { handlePipelineChat, type PipelineStep } from "../../open-sse/services/pipeline.ts"; + +// --------------------------------------------------------------------------- +// Types & helpers +// --------------------------------------------------------------------------- + +type Body = Record; + +interface MockResponse { + ok: boolean; + status: number; + json: () => Promise; + clone: () => MockResponse; +} + +/** Build a successful OpenAI-shaped response with given text. */ +function okResponse(text: string): MockResponse { + const body = { + choices: [{ message: { role: "assistant", content: text } }], + }; + return { + ok: true, + status: 200, + json: async () => body, + // handlePipelineChat calls res.clone().json() on intermediate steps + clone: () => okResponse(text), + }; +} + +/** Build a failed response with given status. */ +function failResponse(status: number): MockResponse { + const body = { error: { message: `HTTP ${status}` } }; + return { + ok: false, + status, + json: async () => body, + clone: () => failResponse(status), + }; +} + +type HandlerFn = (body: Body, model: string) => Promise; + +/** Build a mock handleSingleModel that returns responses in sequence. */ +function makeHandler(responses: MockResponse[], opts?: { loopLast?: boolean }): HandlerFn { + let call = 0; + return async (_body: Body, _model: string): Promise => { + const idx = call++; + if (idx < responses.length) return responses[idx]; + if (opts?.loopLast && responses.length > 0) return responses[responses.length - 1]; + return okResponse("fallback"); + }; +} + +// Minimal stub type — the real type is more complex but we only need (body, model) => Response +// (kept for reference, not used as value) + +const noopLog = { + info: () => {}, + warn: () => {}, + error: () => {}, + debug: () => {}, +}; + +const STEPS: PipelineStep[] = [ + { model: "provider-a/model-a" }, + { model: "provider-b/model-b" }, +]; + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +describe("handlePipelineChat — transient retry", () => { + it("succeeds when all steps return 200", async () => { + const handler = makeHandler([okResponse("step 1 output"), okResponse("step 2 output")]); + const res = await handlePipelineChat({ + body: { messages: [{ role: "user", content: "hi" }] }, + steps: STEPS, + handleSingleModel: handler as unknown as never, + log: noopLog as never, + maxRetries: 2, + }); + assert.equal(res.ok, true); + }); + + it("retries on 429 then succeeds", async () => { + // First call to step 1 → 429, second call to step 1 → 200, step 2 → 200 + const handler = makeHandler([ + failResponse(429), + okResponse("recovered"), + okResponse("final"), + ]); + const res = await handlePipelineChat({ + body: { messages: [{ role: "user", content: "hi" }] }, + steps: STEPS, + handleSingleModel: handler as unknown as never, + log: noopLog as never, + maxRetries: 2, + retryDelayMs: 1, // fast for tests + }); + assert.equal(res.ok, true); + }); + + it("retries on 503 then succeeds", async () => { + const handler = makeHandler([ + failResponse(503), + failResponse(503), + okResponse("recovered after 2"), + okResponse("final"), + ]); + const res = await handlePipelineChat({ + body: { messages: [{ role: "user", content: "hi" }] }, + steps: STEPS, + handleSingleModel: handler as unknown as never, + log: noopLog as never, + maxRetries: 2, + retryDelayMs: 1, + }); + assert.equal(res.ok, true); + }); + + it("fails after exhausting retries on persistent 429", async () => { + // Step 1 always returns 429, even after maxRetries=2 (3 total attempts) + const handler = makeHandler([failResponse(429)], { loopLast: true }); + const res = await handlePipelineChat({ + body: { messages: [{ role: "user", content: "hi" }] }, + steps: STEPS, + handleSingleModel: handler as unknown as never, + log: noopLog as never, + maxRetries: 2, + retryDelayMs: 1, + }); + assert.equal(res.ok, false); + assert.equal(res.status, 429); + }); + + it("fails immediately on 400 (non-transient, no retry)", async () => { + let callCount = 0; + const handler = async (): Promise => { + callCount++; + return failResponse(400); + }; + const res = await handlePipelineChat({ + body: { messages: [{ role: "user", content: "hi" }] }, + steps: STEPS, + handleSingleModel: handler as unknown as never, + log: noopLog as never, + maxRetries: 5, // should NOT retry on 400 + retryDelayMs: 1, + }); + assert.equal(res.ok, false); + assert.equal(res.status, 400); + // Only 1 call — no retry on non-transient error + assert.equal(callCount, 1); + }); + + it("does NOT retry the final step", async () => { + // Step 1 → 200, final step → 502 (should be returned as-is, no retry) + const handler = makeHandler([okResponse("step 1"), failResponse(502)], { loopLast: true }); + const res = await handlePipelineChat({ + body: { messages: [{ role: "user", content: "hi" }] }, + steps: STEPS, + handleSingleModel: handler as unknown as never, + log: noopLog as never, + maxRetries: 3, + retryDelayMs: 1, + }); + // Final step result is returned directly regardless of status + assert.equal(res.status, 502); + assert.equal(res.ok, false); + }); +}); + +describe("handlePipelineChat — backward compat (no retry)", () => { + it("fails on transient error when maxRetries=0 (default)", async () => { + const handler = makeHandler([failResponse(429)], { loopLast: true }); + const res = await handlePipelineChat({ + body: { messages: [{ role: "user", content: "hi" }] }, + steps: STEPS, + handleSingleModel: handler as unknown as never, + log: noopLog as never, + // maxRetries defaults to 0 — no retry + }); + assert.equal(res.ok, false); + assert.equal(res.status, 429); + }); +});