import test from "node:test"; import assert from "node:assert/strict"; import fs from "node:fs"; import fsp from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import http from "node:http"; import net from "node:net"; import { spawn } from "node:child_process"; import { fileURLToPath } from "node:url"; const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-resilience-http-e2e-")); const DASHBOARD_PORT = await getFreePort(); const REPO_ROOT = fileURLToPath(new URL("../..", import.meta.url)); process.env.DATA_DIR = TEST_DATA_DIR; process.env.DISABLE_SQLITE_AUTO_BACKUP = "true"; process.env.API_KEY_SECRET = process.env.API_KEY_SECRET || "resilience-http-e2e-secret-123456"; process.env.REQUIRE_API_KEY = "false"; const core = await import("../../src/lib/db/core.ts"); const providersDb = await import("../../src/lib/db/providers.ts"); const combosDb = await import("../../src/lib/db/combos.ts"); const settingsDb = await import("../../src/lib/db/settings.ts"); function getFreePort() { return new Promise((resolve, reject) => { const server = net.createServer(); server.once("error", reject); server.listen(0, "127.0.0.1", () => { const address = server.address(); if (!address || typeof address === "string") { server.close(); reject(new Error("Failed to allocate a free port")); return; } const { port } = address; server.close((closeError) => { if (closeError) reject(closeError); else resolve(port); }); }); }); } function sleep(ms: number) { return new Promise((resolve) => setTimeout(resolve, ms)); } function buildCompletion(content: string) { return { status: 200, body: { id: `chatcmpl_${Math.random().toString(16).slice(2, 8)}`, object: "chat.completion", choices: [{ index: 0, message: { role: "assistant", content }, finish_reason: "stop" }], usage: { prompt_tokens: 4, completion_tokens: 2, total_tokens: 6 }, }, }; } function buildError(status: number, message: string, headers: Record = {}) { return { status, headers, body: { error: { message } }, }; } type PlannedResponse = { status: number; body: Record; headers?: Record; delayMs?: number; }; type TokenBehavior = { defaultResponse: PlannedResponse; queue: PlannedResponse[]; hits: number; startedAt: number[]; bodies: Array>; }; function createFakeOpenAiRelay() { const behaviors = new Map(); let server: http.Server | null = null; let baseUrl = ""; const handleRequest = async (req: http.IncomingMessage, res: http.ServerResponse) => { const chunks: Buffer[] = []; req.on("data", (chunk) => chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk))); req.on("end", async () => { const authHeader = String(req.headers.authorization || ""); const token = authHeader.replace(/^Bearer\s+/i, "").trim(); const rawBody = Buffer.concat(chunks).toString("utf8"); const parsedBody = rawBody ? JSON.parse(rawBody) : {}; if (req.method === "GET" && req.url?.startsWith("/v1/models")) { res.writeHead(200, { "Content-Type": "application/json" }); res.end(JSON.stringify({ data: [{ id: "test-model", object: "model" }] })); return; } if (req.method !== "POST" || !req.url?.startsWith("/v1/chat/completions")) { res.writeHead(404, { "Content-Type": "application/json" }); res.end(JSON.stringify({ error: { message: `Unhandled path: ${req.method} ${req.url}` } })); return; } const behavior = behaviors.get(token); if (!behavior) { res.writeHead(401, { "Content-Type": "application/json" }); res.end(JSON.stringify({ error: { message: `Unknown token: ${token || "missing"}` } })); return; } behavior.hits += 1; behavior.startedAt.push(Date.now()); behavior.bodies.push(parsedBody as Record); const planned = behavior.queue.shift() || behavior.defaultResponse; if (planned.delayMs && planned.delayMs > 0) { await sleep(planned.delayMs); } const headers = { "Content-Type": "application/json", ...(planned.headers || {}) }; res.writeHead(planned.status, headers); res.end(JSON.stringify(planned.body)); }); }; return { async start() { const port = await getFreePort(); await new Promise((resolve, reject) => { server = http.createServer((req, res) => { void handleRequest(req, res); }); server.once("error", reject); server.listen(port, "127.0.0.1", () => resolve()); }); baseUrl = `http://127.0.0.1:${port}/v1`; return baseUrl; }, getBaseUrl() { if (!baseUrl) throw new Error("Fake relay has not started yet"); return baseUrl; }, configureToken( token: string, config: { defaultResponse: PlannedResponse; queue?: PlannedResponse[] } ) { behaviors.set(token, { defaultResponse: config.defaultResponse, queue: [...(config.queue || [])], hits: 0, startedAt: [], bodies: [], }); }, getState(token: string) { const state = behaviors.get(token); if (!state) throw new Error(`Unknown token state for ${token}`); return state; }, resetState(token: string, queue?: PlannedResponse[]) { const state = behaviors.get(token); if (!state) throw new Error(`Unknown token state for ${token}`); state.hits = 0; state.startedAt = []; state.bodies = []; state.queue = [...(queue || [])]; }, async stop() { if (!server) return; await new Promise((resolve) => server?.close(() => resolve())); server = null; }, }; } function createServerProcess(dataDir: string, port: number) { const stdoutLines: string[] = []; const stderrLines: string[] = []; let exitInfo: { code: number | null; signal: NodeJS.Signals | null } | null = null; const child = spawn(process.execPath, ["scripts/dev/run-next-playwright.mjs", "dev"], { cwd: REPO_ROOT, env: { ...process.env, DATA_DIR: dataDir, PORT: String(port), DASHBOARD_PORT: String(port), API_PORT: String(port), HOST: "127.0.0.1", REQUIRE_API_KEY: "false", API_KEY_SECRET: process.env.API_KEY_SECRET || "resilience-http-e2e-secret-123456", DISABLE_SQLITE_AUTO_BACKUP: "true", INITIAL_PASSWORD: "", NEXT_TELEMETRY_DISABLED: "1", OMNIROUTE_DISABLE_BACKGROUND_SERVICES: "true", OMNIROUTE_DISABLE_TOKEN_HEALTHCHECK: "true", OMNIROUTE_DISABLE_LOCAL_HEALTHCHECK: "true", OMNIROUTE_HIDE_HEALTHCHECK_LOGS: "true", OMNIROUTE_E2E_BOOTSTRAP_MODE: "open", }, stdio: ["ignore", "pipe", "pipe"], }); child.once("exit", (code, signal) => { exitInfo = { code, signal }; }); child.stdout.on("data", (chunk) => { const lines = String(chunk).split(/\r?\n/).filter(Boolean); stdoutLines.push(...lines); if (stdoutLines.length > 200) stdoutLines.splice(0, stdoutLines.length - 200); }); child.stderr.on("data", (chunk) => { const lines = String(chunk).split(/\r?\n/).filter(Boolean); stderrLines.push(...lines); if (stderrLines.length > 200) stderrLines.splice(0, stderrLines.length - 200); }); return { child, stdoutLines, stderrLines, baseUrl: `http://127.0.0.1:${port}`, get exitInfo() { return exitInfo; }, }; } async function waitForServer( baseUrl: string, logs: { stdoutLines: string[]; stderrLines: string[]; exitInfo?: { code: number | null; signal: NodeJS.Signals | null } | null; } ) { const startedAt = Date.now(); let lastError = ""; while (Date.now() - startedAt < 120_000) { if (logs.exitInfo) { throw new Error( [ `OmniRoute exited before it became ready (code=${logs.exitInfo.code}, signal=${logs.exitInfo.signal})`, "--- stdout ---", ...logs.stdoutLines.slice(-40), "--- stderr ---", ...logs.stderrLines.slice(-40), ].join("\n") ); } try { const response = await fetch(`${baseUrl}/api/monitoring/health`, { signal: AbortSignal.timeout(5_000), }); if (response.ok) return; lastError = `HTTP ${response.status}`; } catch (error: any) { lastError = error instanceof Error ? error.message : String(error); } await sleep(500); } throw new Error( [ `Timed out waiting for OmniRoute to start: ${lastError}`, "--- stdout ---", ...logs.stdoutLines.slice(-40), "--- stderr ---", ...logs.stderrLines.slice(-40), ].join("\n") ); } async function stopProcess(child: ReturnType) { if (child.killed) return; child.kill("SIGTERM"); const exited = await Promise.race([ new Promise((resolve) => child.once("exit", () => resolve(true))), sleep(5_000).then(() => false), ]); if (!exited && !child.killed) { child.kill("SIGKILL"); await new Promise((resolve) => child.once("exit", () => resolve())); } } async function seedCompatibleProvider(prefix: string, apiKey: string, baseUrl: string) { const providerId = `openai-compatible-chat-e2e-${prefix}`; await providersDb.createProviderNode({ id: providerId, type: "openai-compatible", name: `E2E ${prefix}`, prefix, apiType: "chat", baseUrl, }); await providersDb.createProviderConnection({ provider: providerId, authType: "apikey", name: `conn-${prefix}`, apiKey, isActive: true, testStatus: "active", providerSpecificData: { baseUrl, apiType: "chat", }, }); return { providerId, model: `${prefix}/test-model`, apiKey }; } function buildResilienceConfig(overrides: Record = {}) { const base = { requestQueue: { autoEnableApiKeyProviders: true, requestsPerMinute: 120, minTimeBetweenRequestsMs: 0, concurrentRequests: 4, maxWaitMs: 2_000, }, connectionCooldown: { oauth: { baseCooldownMs: 500, useUpstreamRetryHints: true, maxBackoffSteps: 3, }, apikey: { baseCooldownMs: 5_000, useUpstreamRetryHints: false, maxBackoffSteps: 0, }, }, providerBreaker: { oauth: { failureThreshold: 3, degradationThreshold: 2, resetTimeoutMs: 2_000, }, apikey: { failureThreshold: 2, degradationThreshold: 1, resetTimeoutMs: 1_500, }, }, waitForCooldown: { enabled: false, maxRetries: 0, maxRetryWaitSec: 0, }, }; return { ...base, ...overrides, requestQueue: { ...base.requestQueue, ...((overrides.requestQueue as Record) || {}), }, connectionCooldown: { oauth: { ...base.connectionCooldown.oauth, ...(((overrides.connectionCooldown as Record)?.oauth as Record< string, unknown >) || {}), }, apikey: { ...base.connectionCooldown.apikey, ...(((overrides.connectionCooldown as Record)?.apikey as Record< string, unknown >) || {}), }, }, providerBreaker: { oauth: { ...base.providerBreaker.oauth, ...(((overrides.providerBreaker as Record)?.oauth as Record< string, unknown >) || {}), }, apikey: { ...base.providerBreaker.apikey, ...(((overrides.providerBreaker as Record)?.apikey as Record< string, unknown >) || {}), }, }, waitForCooldown: { ...base.waitForCooldown, ...((overrides.waitForCooldown as Record) || {}), }, }; } async function patchResilience(baseUrl: string, config: Record) { const response = await fetch(`${baseUrl}/api/resilience`, { method: "PATCH", headers: { "Content-Type": "application/json" }, body: JSON.stringify(config), signal: AbortSignal.timeout(10_000), }); const payload = (await response.json()) as any; assert.equal(response.status, 200, JSON.stringify(payload)); return payload; } async function getJson(url: string) { const response = await fetch(url, { signal: AbortSignal.timeout(10_000) }); const json = (await response.json()) as any; return { response, json }; } async function postChat(baseUrl: string, model: string, content: string) { const response = await fetch(`${baseUrl}/api/v1/chat/completions`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ model, stream: false, messages: [{ role: "user", content }], }), signal: AbortSignal.timeout(20_000), }); const text = await response.text(); const json = text ? JSON.parse(text) : {}; return { response, json }; } const relay = createFakeOpenAiRelay(); let app: | { child: ReturnType; stdoutLines: string[]; stderrLines: string[]; baseUrl: string; } | undefined; const TOKENS = { p1: "sk-e2e-p1", p2: "sk-e2e-p2", p3: "sk-e2e-p3", p4: "sk-e2e-p4", p5: "sk-e2e-p5", p6: "sk-e2e-p6", p7: "sk-e2e-p7", p8: "sk-e2e-p8", }; test.before(async () => { const fakeBaseUrl = await relay.start(); relay.configureToken(TOKENS.p1, { defaultResponse: buildCompletion("primary healthy again"), queue: [buildError(503, "primary transient failure")], }); relay.configureToken(TOKENS.p2, { defaultResponse: buildCompletion("secondary stable"), }); relay.configureToken(TOKENS.p3, { defaultResponse: buildCompletion("wait-for-cooldown via upstream hint"), queue: [buildError(429, "rate limited, retry after 1 second", { "Retry-After": "1" })], }); relay.configureToken(TOKENS.p4, { defaultResponse: buildCompletion("ignored upstream retry hint"), queue: [buildError(429, "rate limited, retry after 5 seconds", { "Retry-After": "5" })], }); relay.configureToken(TOKENS.p5, { defaultResponse: buildCompletion("breaker target recovered"), queue: [buildError(503, "breaker failure #1"), buildError(503, "breaker failure #2")], }); relay.configureToken(TOKENS.p6, { defaultResponse: buildCompletion("round robin A"), }); relay.configureToken(TOKENS.p7, { defaultResponse: buildCompletion("round robin B"), }); relay.configureToken(TOKENS.p8, { defaultResponse: { ...buildCompletion("queued connection request"), delayMs: 250, }, }); await seedCompatibleProvider("p1", TOKENS.p1, fakeBaseUrl); await seedCompatibleProvider("p2", TOKENS.p2, fakeBaseUrl); await seedCompatibleProvider("p3", TOKENS.p3, fakeBaseUrl); await seedCompatibleProvider("p4", TOKENS.p4, fakeBaseUrl); await seedCompatibleProvider("p5", TOKENS.p5, fakeBaseUrl); await seedCompatibleProvider("p6", TOKENS.p6, fakeBaseUrl); await seedCompatibleProvider("p7", TOKENS.p7, fakeBaseUrl); await seedCompatibleProvider("p8", TOKENS.p8, fakeBaseUrl); await combosDb.createCombo({ name: "res-priority-fallback", strategy: "priority", config: { maxRetries: 0, retryDelayMs: 0 }, models: ["p1/test-model", "p2/test-model"], }); await combosDb.createCombo({ name: "res-breaker-combo", strategy: "priority", config: { maxRetries: 0, retryDelayMs: 0 }, models: ["p5/test-model", "p2/test-model"], }); await combosDb.createCombo({ name: "res-rr", strategy: "round-robin", config: { maxRetries: 0, retryDelayMs: 0, concurrencyPerModel: 1, queueTimeoutMs: 800 }, models: ["p6/test-model", "p7/test-model"], }); await settingsDb.updateSettings({ resilienceSettings: buildResilienceConfig(), requestRetry: 0, maxRetryIntervalSec: 0, stickyRoundRobinLimit: 1, requireLogin: false, setupComplete: true, }); core.closeDbInstance(); app = createServerProcess(TEST_DATA_DIR, DASHBOARD_PORT); await waitForServer(app.baseUrl, app); await patchResilience(app.baseUrl, buildResilienceConfig()); const warmup = await postChat(app.baseUrl, "p2/test-model", "warm up chat route"); assert.equal(warmup.response.status, 200, JSON.stringify(warmup.json)); relay.resetState(TOKENS.p2); }); test.after(async () => { if (app) { await stopProcess(app.child); } await relay.stop(); core.closeDbInstance(); await fsp.rm(TEST_DATA_DIR, { recursive: true, force: true }); }); test("resilience API only exposes configuration, not runtime breaker state", async () => { assert.ok(app); const { response, json } = await getJson(`${app.baseUrl}/api/resilience`); assert.equal(response.status, 200); assert.deepEqual(Object.keys(json).sort(), [ "comboCooldownWait", "connectionCooldown", "legacy", "providerBreaker", "providerCooldown", "quotaShareConcurrencyLimit", "requestQueue", "waitForCooldown", ]); assert.equal("providerBreakers" in json, false); assert.equal("runtime" in json, false); }); test("request queue serializes concurrent requests on the same connection", async () => { assert.ok(app); await patchResilience( app.baseUrl, buildResilienceConfig({ requestQueue: { concurrentRequests: 1, maxWaitMs: 1_500, }, }) ); relay.resetState(TOKENS.p8); const startedAt = Date.now(); const [first, second] = await Promise.all([ postChat(app.baseUrl, "p8/test-model", "queue-one"), postChat(app.baseUrl, "p8/test-model", "queue-two"), ]); const elapsed = Date.now() - startedAt; const state = relay.getState(TOKENS.p8); assert.equal(first.response.status, 200, JSON.stringify(first.json)); assert.equal(second.response.status, 200, JSON.stringify(second.json)); assert.equal(state.hits, 2); assert.ok(elapsed >= 450, `expected queued elapsed >= 450ms, got ${elapsed}ms`); assert.ok( state.startedAt[1] - state.startedAt[0] >= 180, `expected second request to be delayed by queue, got ${state.startedAt[1] - state.startedAt[0]}ms` ); }); test("priority combo falls back on 503 and skips the cooled-down primary on the next request", async () => { assert.ok(app); await patchResilience(app.baseUrl, buildResilienceConfig()); relay.resetState(TOKENS.p1, [buildError(503, "primary transient failure")]); relay.resetState(TOKENS.p2); const first = await postChat(app.baseUrl, "res-priority-fallback", "priority fallback request"); assert.equal(first.response.status, 200, JSON.stringify(first.json)); assert.equal(first.json.choices[0].message.content, "secondary stable"); assert.equal(relay.getState(TOKENS.p1).hits, 1); assert.equal(relay.getState(TOKENS.p2).hits, 1); // Brief pause to ensure the P1 connection cooldown write has been committed // and is visible to the second request's credential lookup. await sleep(200); const second = await postChat(app.baseUrl, "res-priority-fallback", "priority fallback again"); assert.equal(second.response.status, 200, JSON.stringify(second.json)); assert.equal(second.json.choices[0].message.content, "secondary stable"); assert.equal(relay.getState(TOKENS.p1).hits, 1); assert.equal(relay.getState(TOKENS.p2).hits, 2); }); test.skip("wait-for-cooldown honors upstream Retry-After when enabled", async () => { assert.ok(app); await patchResilience( app.baseUrl, buildResilienceConfig({ connectionCooldown: { apikey: { useUpstreamRetryHints: true, baseCooldownMs: 200, }, }, waitForCooldown: { enabled: true, maxRetries: 1, maxRetryWaitSec: 2, }, }) ); relay.resetState(TOKENS.p3); const warmup = await postChat(app.baseUrl, "p3/test-model", "warm provider-specific route"); assert.equal(warmup.response.status, 200, JSON.stringify(warmup.json)); relay.resetState(TOKENS.p3, [ buildError(429, "rate limited, retry after 1 second", { "Retry-After": "1" }), ]); const startedAt = Date.now(); const result = await postChat(app.baseUrl, "p3/test-model", "wait for cooldown via upstream"); const elapsed = Date.now() - startedAt; assert.equal(result.response.status, 200, JSON.stringify(result.json)); assert.equal(result.json.choices[0].message.content, "wait-for-cooldown via upstream hint"); const hits = relay.getState(TOKENS.p3).hits; assert.ok(hits >= 2, `expected at least one retry after cooldown, got ${hits} hits`); assert.ok(elapsed >= 800, `expected upstream wait >= 800ms, got ${elapsed}ms`); }); test.skip("connection cooldown can ignore upstream Retry-After and use the configured local cooldown", async () => { assert.ok(app); await patchResilience( app.baseUrl, buildResilienceConfig({ connectionCooldown: { apikey: { useUpstreamRetryHints: false, baseCooldownMs: 200, }, }, waitForCooldown: { enabled: true, maxRetries: 1, maxRetryWaitSec: 2, }, }) ); relay.resetState(TOKENS.p4); const warmup = await postChat(app.baseUrl, "p4/test-model", "warm provider-specific route"); assert.equal(warmup.response.status, 200, JSON.stringify(warmup.json)); relay.resetState(TOKENS.p4, [ buildError(429, "rate limited, retry after 30 seconds", { "Retry-After": "30" }), ]); const startedAt = Date.now(); const result = await postChat(app.baseUrl, "p4/test-model", "ignore upstream retry-after"); const elapsed = Date.now() - startedAt; assert.equal(result.response.status, 200, JSON.stringify(result.json)); assert.equal(result.json.choices[0].message.content, "ignored upstream retry hint"); const hits = relay.getState(TOKENS.p4).hits; assert.ok(hits >= 2, `expected at least one retry after cooldown, got ${hits} hits`); assert.ok( elapsed < 5_000, `expected ignored upstream hint to avoid a 30s wait, got ${elapsed}ms` ); }); test.skip("provider circuit breaker opens after repeated final failures and Health reports it", async () => { assert.ok(app); await patchResilience( app.baseUrl, buildResilienceConfig({ waitForCooldown: { enabled: false, maxRetries: 0, maxRetryWaitSec: 0, }, providerBreaker: { apikey: { failureThreshold: 2, degradationThreshold: 1, resetTimeoutMs: 1_500, }, }, }) ); relay.resetState(TOKENS.p5, [ buildError(503, "breaker failure #1"), buildError(503, "breaker failure #2"), ]); const first = await postChat(app.baseUrl, "p5/test-model", "breaker first failure"); const second = await postChat(app.baseUrl, "p5/test-model", "breaker second failure"); const third = await postChat(app.baseUrl, "p5/test-model", "breaker should now be open"); assert.equal(first.response.status, 503); assert.equal(second.response.status, 503); assert.equal(third.response.status, 503); assert.match(String(second.json.error?.message || ""), /reset after/i); assert.match(String(third.json.error?.message || ""), /circuit breaker is open/i); assert.equal(relay.getState(TOKENS.p5).hits, 1); const health = await getJson(`${app.baseUrl}/api/monitoring/health`); assert.equal(health.response.status, 200); const breakerEntry = (health.json.providerBreakers || []).find( (entry: Record) => entry.provider === "openai-compatible-chat-e2e-p5" ); assert.ok(breakerEntry, "expected provider breaker entry for p5"); assert.equal(breakerEntry.state, "OPEN"); assert.ok(Number(breakerEntry.failureCount) >= 2); assert.ok(Number(breakerEntry.retryAfterMs) > 0); assert.ok( (health.json.providerBreakers || []).every( (entry: Record) => !String(entry.provider || "").includes(":") ) ); }); test("combo respects the global provider breaker and falls through without a combo-local breaker", async () => { assert.ok(app); relay.resetState(TOKENS.p2); const result = await postChat( app.baseUrl, "res-breaker-combo", "combo should skip broken provider" ); assert.equal(result.response.status, 200, JSON.stringify(result.json)); assert.equal(result.json.choices[0].message.content, "secondary stable"); assert.equal(relay.getState(TOKENS.p5).hits, 1); assert.equal(relay.getState(TOKENS.p2).hits, 1); }); test("round-robin combo still alternates healthy providers after combo breaker removal", async () => { assert.ok(app); await patchResilience(app.baseUrl, buildResilienceConfig()); relay.resetState(TOKENS.p6); relay.resetState(TOKENS.p7); const first = await postChat(app.baseUrl, "res-rr", "round robin one"); const second = await postChat(app.baseUrl, "res-rr", "round robin two"); assert.equal(first.response.status, 200, JSON.stringify(first.json)); assert.equal(second.response.status, 200, JSON.stringify(second.json)); assert.equal(first.json.choices[0].message.content, "round robin A"); assert.equal(second.json.choices[0].message.content, "round robin B"); assert.equal(relay.getState(TOKENS.p6).hits, 1); assert.equal(relay.getState(TOKENS.p7).hits, 1); });