diff --git a/src/app/api/providers/[id]/sync-models/route.ts b/src/app/api/providers/[id]/sync-models/route.ts index a8b807985b..3ff89b1ce9 100644 --- a/src/app/api/providers/[id]/sync-models/route.ts +++ b/src/app/api/providers/[id]/sync-models/route.ts @@ -154,44 +154,202 @@ function getModelSyncChannelLabel(connection: unknown) { ); } +// --------------------------------------------------------------------------- +// Shared loopback readiness gate — eliminates 17× retry amplification at boot +// --------------------------------------------------------------------------- + +// Module-level shared promise: gates all selfFetchWithRetry callers behind a +// single readiness probe. The 17 connections that fire ModelSync at boot all +// await the same promise; the underlying HTTP probe runs exactly once per +// process. Resolves on first HTTP response (any status — even 4xx confirms the +// server is up); rejects only if maxWaitMs elapses with consistent network +// errors. +let __loopbackReadyPromise: Promise | null = null; + +export type EnsureReadyOptions = { + fetch?: typeof fetch; + maxWaitMs?: number; + pollMs?: number; +}; + +export async function ensureLoopbackServerReady(opts: EnsureReadyOptions = {}): Promise { + if (__loopbackReadyPromise) return __loopbackReadyPromise; + __loopbackReadyPromise = (async () => { + const f = opts.fetch ?? fetch; + const maxWaitMs = opts.maxWaitMs ?? 30_000; + const pollMs = opts.pollMs ?? 250; + const deadline = Date.now() + maxWaitMs; + let lastErr: unknown; + while (Date.now() < deadline) { + try { + // Hit a stable endpoint; any HTTP status (200/404/etc) confirms + // readiness — we only care that the dispatcher succeeds (no + // ECONNREFUSED). Using a synthetic connection id so no real DB lookup + // is needed; the 404 is sufficient proof the server is dispatching. + const probePort = process.env.OMNIROUTE_PORT || process.env.PORT || "20128"; + const res = await f(`http://127.0.0.1:${probePort}/api/providers/__readiness_probe__/models`, { + signal: AbortSignal.timeout(2_000), + }); + if (res.status >= 200 && res.status < 600) return; + } catch (err) { + lastErr = err; + } + await new Promise((r) => setTimeout(r, pollMs)); + } + throw new Error(`loopback server not ready within ${maxWaitMs}ms: ${String(lastErr)}`); + })(); + return __loopbackReadyPromise; +} + +/** Test helper: reset the cached promise so tests can re-exercise the probe. */ +export function __resetLoopbackReadinessForTests(): void { + __loopbackReadyPromise = null; +} + +// --------------------------------------------------------------------------- +// selfFetchWithRetry — exported for unit testing +// --------------------------------------------------------------------------- + +export type SelfFetchWithRetryOptions = { + /** Injectable fetch implementation; defaults to global fetch. */ + fetch?: typeof fetch; + /** Maximum number of HTTP attempts before falling back. Default: 3. */ + maxRetries?: number; + /** + * Base backoff in ms. Each attempt waits backoffMs * (attempt + 1) before + * the next try (linear growth: 200 ms, 400 ms, 600 ms, ... at default 200 ms). + * Default: 200. + */ + backoffMs?: number; + /** Connection ID used only for the warning log message. Optional. */ + connectionId?: string; + /** + * Called as last-resort fallback after all retries are exhausted. + * Must return a Response. If omitted, returns a synthetic 503. + */ + inProcessFallback?: () => Promise; + /** + * Skip the shared readiness gate. Use in tests that exercise the retry + * loop in isolation without needing a live loopback server. + * Default: false. + */ + skipReadinessGate?: boolean; +}; + +/** + * Wraps a single loopback self-fetch URL with linear-backoff retry logic. + * + * Motivation: at container boot, ModelSync fires up to 17 concurrent + * self-fetches against http://127.0.0.1:PORT before the in-process HTTP + * listener is fully accepting connections. A shared readiness gate + * (ensureLoopbackServerReady) now serialises the boot race — all 17 callers + * await the same promise, so only one probe sequence runs. Retries here are + * a last-resort for transient failures AFTER the server is confirmed up. + * + * Retry contract: + * - Network errors (fetch failed, ECONNREFUSED): retry up to `maxRetries`. + * - Any HTTP response (2xx/4xx/5xx): returned as-is — the server is up, so the + * caller (route handler) handles status interpretation. Retries are only + * for network-level failures, not for HTTP errors. + */ +export async function selfFetchWithRetry( + url: string, + opts: SelfFetchWithRetryOptions = {} +): Promise { + const f = opts.fetch ?? fetch; + // Reduced from 5 to 3: the readiness gate now handles the boot race. + // Retries here are only for transient failures after server is confirmed up. + const maxRetries = opts.maxRetries ?? 3; + const backoffMs = opts.backoffMs ?? 200; + const connLabel = opts.connectionId ? opts.connectionId.slice(0, 8) : url.slice(-8); + + // Wait for the loopback server to be ready before firing. All concurrent + // callers share the same readiness promise — exactly ONE probe runs per + // process, eliminating the 17× retry amplification observed at boot. + if (opts.skipReadinessGate !== true) { + try { + await ensureLoopbackServerReady({ fetch: f }); + } catch (err) { + // Readiness probe timed out — fall straight through to in-process fallback. + console.warn( + `[ModelSync] Loopback server readiness probe failed; falling back to in-process route immediately (${connLabel}): ${String(err)}` + ); + if (opts.inProcessFallback) { + return opts.inProcessFallback(); + } + return new Response(JSON.stringify({ error: "self-fetch unavailable" }), { status: 503 }); + } + } + + let lastErr: unknown; + for (let attempt = 0; attempt < maxRetries; attempt++) { + try { + const res = await f(url, { method: "GET" }); + // Any HTTP response (2xx, 4xx, 5xx) means the server is up — return as-is. + // We only retry on network-level failures (ECONNREFUSED, "fetch failed") + // which indicate the loopback listener is not yet accepting connections. + return res; + } catch (err) { + lastErr = err; + } + if (attempt < maxRetries - 1) { + await new Promise((r) => setTimeout(r, backoffMs * (attempt + 1))); + } + } + + // All retries exhausted (network-level failures only) — fall back to in-process route + console.warn( + `[ModelSync] Internal /models self-fetch failed for ${connLabel} after ${maxRetries} attempt(s); falling back to in-process route (last err: ${String(lastErr)})` + ); + + if (opts.inProcessFallback) { + return opts.inProcessFallback(); + } + return new Response(JSON.stringify({ error: "self-fetch unavailable" }), { status: 503 }); +} + +// --------------------------------------------------------------------------- +// fetchProviderModelsForSync — private orchestrator (uses selfFetchWithRetry) +// --------------------------------------------------------------------------- + async function fetchProviderModelsForSync(request: Request, connectionId: string) { // Construct a safe localhost URL from the incoming request's origin. // The route only accepts authenticated or internal-scheduler requests, // and the path is hardcoded — no user-controlled URL components reach fetch. + // Always use 127.0.0.1 (IPv4) — never "localhost" which may resolve to ::1 + // (IPv6) in containers, causing TypeError: fetch failed even when the HTTP + // server is bound only to 0.0.0.0 (IPv4 only). const SAFE_HOSTS = new Set(["localhost", "127.0.0.1", "0.0.0.0", "::1"]); const incomingUrl = new URL(request.url); - const safeOrigin = SAFE_HOSTS.has(incomingUrl.hostname) - ? incomingUrl.origin - : `http://127.0.0.1:${process.env.PORT || "20128"}`; + const loopbackPort = + SAFE_HOSTS.has(incomingUrl.hostname) && incomingUrl.port + ? incomingUrl.port + : process.env.PORT || "20128"; + const safeOrigin = `http://127.0.0.1:${loopbackPort}`; const modelsPath = `/api/providers/${encodeURIComponent(connectionId)}/models?refresh=true`; const headers = { cookie: request.headers.get("cookie") || "", ...buildModelSyncInternalHeaders(), }; - try { - return await fetch(new URL(modelsPath, safeOrigin).href, { - method: "GET", - cache: "no-store", - headers, - }); - } catch (error) { - const message = error instanceof Error ? error.message : String(error); - console.warn( - `[ModelSync] Internal /models self-fetch failed for ${connectionId.slice( - 0, - 8 - )}; falling back to in-process route: ${message}` - ); + const targetUrl = new URL(modelsPath, safeOrigin).href; - return getProviderModels( - new Request(new URL(modelsPath, "http://localhost").href, { - method: "GET", - headers, - }), - { params: { id: connectionId } } - ); - } + // Wrap fetch so it forwards the required headers on every retry attempt. + const fetchWithHeaders: typeof fetch = (input, init) => + fetch(input as string, { ...init, headers }); + + return selfFetchWithRetry(targetUrl, { + fetch: fetchWithHeaders, + connectionId, + inProcessFallback: () => + getProviderModels( + new Request(new URL(modelsPath, "http://localhost").href, { + method: "GET", + headers, + }), + { params: { id: connectionId } } + ), + }); } /** diff --git a/src/shared/services/modelSyncScheduler.ts b/src/shared/services/modelSyncScheduler.ts index f6d8d8e948..fe9478cdb6 100644 --- a/src/shared/services/modelSyncScheduler.ts +++ b/src/shared/services/modelSyncScheduler.ts @@ -22,7 +22,7 @@ const INTERNAL_BASE_URL = process.env.BASE_URL || process.env.NEXT_PUBLIC_BASE_URL || process.env.NEXT_PUBLIC_APP_URL || - `http://localhost:${dashboardPort}`; + `http://127.0.0.1:${dashboardPort}`; const globalState = globalThis as typeof globalThis & { __omnirouteModelSyncInternalAuthToken?: string; diff --git a/tests/unit/api/sync-models-readiness.test.ts b/tests/unit/api/sync-models-readiness.test.ts new file mode 100644 index 0000000000..941a970db1 --- /dev/null +++ b/tests/unit/api/sync-models-readiness.test.ts @@ -0,0 +1,250 @@ +import { test, beforeEach } from "node:test"; +import assert from "node:assert/strict"; +import { + selfFetchWithRetry, + ensureLoopbackServerReady, + __resetLoopbackReadinessForTests, +} from "../../../src/app/api/providers/[id]/sync-models/route.ts"; + +// --------------------------------------------------------------------------- +// Test 1: retry succeeds on attempt 3 +// --------------------------------------------------------------------------- +test("self-fetch retries with backoff and succeeds on attempt 3", async () => { + let attempts = 0; + const fetchMock: typeof fetch = async () => { + attempts++; + if (attempts < 3) { + throw new Error("fetch failed"); + } + return new Response(JSON.stringify({ models: [{ id: "model-1" }] }), { status: 200 }); + }; + + let inProcCalls = 0; + const inProcMock = async () => { + inProcCalls++; + return new Response(JSON.stringify({ models: [] }), { status: 200 }); + }; + + const result = await selfFetchWithRetry("http://127.0.0.1:20128/api/providers/conn-1/models", { + fetch: fetchMock, + maxRetries: 5, + backoffMs: 5, + inProcessFallback: inProcMock, + skipReadinessGate: true, + }); + + assert.equal(attempts, 3, "should have retried twice before succeeding on attempt 3"); + assert.equal(inProcCalls, 0, "should not have called in-process route"); + assert.equal(result.ok, true, "response should be ok"); +}); + +// --------------------------------------------------------------------------- +// Test 2: falls back to in-process after maxRetries failures +// --------------------------------------------------------------------------- +test("self-fetch falls back to in-process route after maxRetries failures", async () => { + let attempts = 0; + const fetchMock: typeof fetch = async () => { + attempts++; + throw new Error("fetch failed"); + }; + + let inProcCalls = 0; + const inProcMock = async () => { + inProcCalls++; + return new Response(JSON.stringify({ models: [{ id: "in-proc-model" }] }), { status: 200 }); + }; + + const result = await selfFetchWithRetry("http://127.0.0.1:20128/api/providers/conn-2/models", { + fetch: fetchMock, + maxRetries: 3, + backoffMs: 5, + connectionId: "conn-2", + inProcessFallback: inProcMock, + skipReadinessGate: true, + }); + + assert.equal(attempts, 3, "should retry exactly maxRetries times"); + assert.equal(inProcCalls, 1, "should fall back to in-process exactly once"); + const body = await result.json(); + assert.equal(body.models[0].id, "in-proc-model"); +}); + +// --------------------------------------------------------------------------- +// Test 3: HTTP error responses are returned as-is (no retry on HTTP errors) +// +// Retry contract: only network-level failures (ECONNREFUSED, "fetch failed") +// are retried. HTTP responses (even 4xx/5xx) mean the server IS up and +// returned an error that should be propagated as-is to the caller. +// --------------------------------------------------------------------------- +test("self-fetch returns HTTP error responses immediately without retrying", async () => { + // 5xx HTTP response: server is up but returned error + { + let attempts = 0; + const fetchMock: typeof fetch = async () => { + attempts++; + return new Response("server error", { status: 503 }); + }; + const inProcMock = async () => new Response(JSON.stringify({ models: [] }), { status: 200 }); + + const res = await selfFetchWithRetry("http://127.0.0.1:20128/api/providers/conn-3/models", { + fetch: fetchMock, + maxRetries: 5, + backoffMs: 5, + inProcessFallback: inProcMock, + skipReadinessGate: true, + }); + + assert.equal(attempts, 1, "5xx HTTP response should NOT retry (got " + attempts + ")"); + assert.equal(res.status, 503, "should propagate the 503 response as-is"); + } + + // 4xx HTTP response: also returned immediately without retry + { + let attempts = 0; + const fetchMock: typeof fetch = async () => { + attempts++; + return new Response("not found", { status: 404 }); + }; + const inProcMock = async () => new Response(JSON.stringify({ models: [] }), { status: 200 }); + + const res = await selfFetchWithRetry("http://127.0.0.1:20128/api/providers/conn-4/models", { + fetch: fetchMock, + maxRetries: 5, + backoffMs: 5, + inProcessFallback: inProcMock, + skipReadinessGate: true, + }); + + assert.equal(attempts, 1, "4xx HTTP response should NOT retry (got " + attempts + ")"); + assert.equal(res.status, 404, "should propagate the 404 response as-is"); + } +}); + +// --------------------------------------------------------------------------- +// Readiness gate tests +// --------------------------------------------------------------------------- + +test("ensureLoopbackServerReady: 17 concurrent callers trigger exactly ONE probe sequence", async () => { + __resetLoopbackReadinessForTests(); + let probeCalls = 0; + let serverIsUp = false; + // Server becomes ready after 30ms + setTimeout(() => { + serverIsUp = true; + }, 30); + const mockFetch = async (_url) => { + probeCalls++; + if (!serverIsUp) throw new Error("fetch failed"); + return new Response("", { status: 200 }); + }; + + await Promise.all( + Array.from({ length: 17 }, () => + ensureLoopbackServerReady({ fetch: mockFetch, pollMs: 5, maxWaitMs: 1000 }) + ) + ); + + // Probe may have polled multiple times before server came up -- that is fine. + // What matters: 17 concurrent callers share ONE probe sequence. + // Expected: roughly (30ms / 5ms) = ~6 attempts, not 17 x 6 = 102. + assert.ok(probeCalls <= 15, "single shared probe expected <=15 attempts, got " + probeCalls); +}); + +test("ensureLoopbackServerReady: rejects after maxWaitMs with consistent network errors", async () => { + __resetLoopbackReadinessForTests(); + const mockFetch = async (_url) => { + throw new Error("ECONNREFUSED"); + }; + await assert.rejects( + () => + ensureLoopbackServerReady({ + fetch: mockFetch, + maxWaitMs: 50, + pollMs: 10, + }), + /loopback server not ready/ + ); +}); + +test("ensureLoopbackServerReady: resolves on 4xx (any HTTP status confirms server is up)", async () => { + __resetLoopbackReadinessForTests(); + let calls = 0; + const mockFetch = async (_url) => { + calls++; + return new Response("not found", { status: 404 }); + }; + // Should not throw: 404 means the server is dispatching + await ensureLoopbackServerReady({ fetch: mockFetch, maxWaitMs: 500, pollMs: 10 }); + assert.equal(calls, 1, "resolved after exactly 1 probe (server immediately responded 404)"); +}); + +test("selfFetchWithRetry with gate: 17 concurrent callers produce one probe + one fetch each", async () => { + __resetLoopbackReadinessForTests(); + let probeCalls = 0; + let modelFetchCalls = 0; + let serverIsUp = false; + setTimeout(() => { + serverIsUp = true; + }, 30); + + const mockFetch = async (url) => { + const isReadinessProbe = url.includes("__readiness_probe__"); + if (isReadinessProbe) { + probeCalls++; + if (!serverIsUp) throw new Error("fetch failed"); + return new Response("", { status: 404 }); + } + modelFetchCalls++; + if (!serverIsUp) throw new Error("fetch failed"); + return new Response(JSON.stringify({ models: [{ id: "model-x" }] }), { status: 200 }); + }; + + await Promise.all( + Array.from({ length: 17 }, (_, i) => + selfFetchWithRetry("http://127.0.0.1:20128/api/providers/conn-" + i + "/models", { + fetch: mockFetch, + maxRetries: 3, + backoffMs: 5, + }) + ) + ); + + // After readiness gate succeeds, each connection makes EXACTLY one model fetch + assert.equal(modelFetchCalls, 17, "each connection fetches its own models exactly once"); + // Probe may have polled several times but NOT 17 x poll-count + assert.ok(probeCalls <= 15, "probe should be shared, got " + probeCalls + " attempts"); +}); + +// Sanity check: disabling the gate shows amplification (verifies the gate is doing work) +test("sanity: without readiness gate, 17 callers retry independently (amplification confirmed)", async () => { + __resetLoopbackReadinessForTests(); + let modelFetchCalls = 0; + let serverIsUp = false; + setTimeout(() => { + serverIsUp = true; + }, 30); + + const mockFetch = async (_url) => { + modelFetchCalls++; + if (!serverIsUp) throw new Error("fetch failed"); + return new Response(JSON.stringify({ models: [{ id: "model-x" }] }), { status: 200 }); + }; + + await Promise.all( + Array.from({ length: 17 }, (_, i) => + selfFetchWithRetry("http://127.0.0.1:20128/api/providers/conn-" + i + "/models", { + fetch: mockFetch, + maxRetries: 5, + backoffMs: 5, + skipReadinessGate: true, + }) + ) + ); + + // Without gate: each of the 17 callers retries independently during boot race. + // Expect well above 17 total fetch attempts. + assert.ok( + modelFetchCalls > 17, + "without gate, callers retry independently, got " + modelFetchCalls + " (expected >17)" + ); +}); diff --git a/tests/unit/model-sync-route.test.ts b/tests/unit/model-sync-route.test.ts index 6d5f8aff1d..c173ab68e4 100644 --- a/tests/unit/model-sync-route.test.ts +++ b/tests/unit/model-sync-route.test.ts @@ -919,9 +919,19 @@ test("model sync route falls back to in-process discovery when internal self-fet }, }); + // Reset shared readiness gate so this test exercises the probe path cleanly. + modelSyncRoute.__resetLoopbackReadinessForTests(); + const fetchCalls: string[] = []; globalThis.fetch = async (url) => { const urlString = String(url); + + // Loopback readiness probe: respond 404 so the gate opens immediately. + // (Any HTTP response confirms the server is up — see ensureLoopbackServerReady.) + if (urlString.includes("__readiness_probe__")) { + return new Response(null, { status: 404 }); + } + fetchCalls.push(urlString); if (urlString === `http://localhost/api/providers/${connection.id}/models?refresh=true`) { @@ -965,8 +975,18 @@ test("model sync route falls back to in-process discovery when internal self-fet availableModels.map((model) => ({ id: model.id, source: model.source })), [{ id: "aio-model", source: "imported" }] ); - assert.deepEqual(fetchCalls, [ - `http://localhost/api/providers/${connection.id}/models?refresh=true`, - "https://api.bltcy.ai/v1/models", - ]); + // selfFetchWithRetry default maxRetries=3: all 3 attempts throw, then in-process + // fallback fires (which triggers the upstream bltcy.ai fetch). So fetchCalls + // contains 3 self-fetch URLs followed by 1 upstream URL. + // Route forces IPv4 origin (http://127.0.0.1:PORT) — never "localhost" — to avoid + // ::1 (IPv6) resolution issues in containers. PORT defaults to 20128 when env unset. + const expectedPort = process.env.OMNIROUTE_PORT || process.env.PORT || "20128"; + const selfFetchUrl = `http://127.0.0.1:${expectedPort}/api/providers/${connection.id}/models?refresh=true`; + assert.equal( + fetchCalls.slice(0, 3).every((u) => u === selfFetchUrl), + true, + "first 3 calls should be self-fetch retries" + ); + assert.equal(fetchCalls[3], "https://api.bltcy.ai/v1/models", "4th call should be upstream"); + assert.equal(fetchCalls.length, 4, "should have exactly 3 retries + 1 upstream call"); });