fix(ModelSync): shared loopback readiness gate + IPv4 force (#2221)

Integrated into release/v3.8.0

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<void> | null = null;
+
+export type EnsureReadyOptions = {
+  fetch?: typeof fetch;
+  maxWaitMs?: number;
+  pollMs?: number;
+};
+
+export async function ensureLoopbackServerReady(opts: EnsureReadyOptions = {}): Promise<void> {
+  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<Response>;
+  /**
+   * 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<Response> {
+  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<void>((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 } }
+      ),
+  });
 }
 
 /**

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;

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)"
+  );
+});

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");
 });