Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog.d/fixes/10903-loopback-gate-memory-success.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
- **fix(providers):** the loopback readiness gate no longer memorizes a failed probe — the next caller after 30s starts a fresh probe, and a readiness failure is logged once per probe instead of once per caller ([#10903](https://github.com/diegosouzapw/OmniRoute/pull/10903))
35 changes: 30 additions & 5 deletions src/app/api/providers/[id]/sync-models/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -175,17 +175,33 @@ function getModelSyncChannelLabel(connection: unknown) {
// 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.
// errors. On rejection the promise is NOT memorized: the gate re-probes for the
// next caller after the retry window, so a boot-time failure cannot condemn the
// process to the in-process fallback for its whole lifetime.
let __loopbackReadyPromise: Promise<void> | null = null;
let __loopbackLastFailureAt = 0;

/** Anti-storm bound: minimum interval between two probes after a failure. */
const LOOPBACK_RETRY_MIN_INTERVAL_MS = 30_000;

export type EnsureReadyOptions = {
fetch?: typeof fetch;
maxWaitMs?: number;
pollMs?: number;
/** Minimum interval between two probes after a failure (anti-storm). */
minRetryIntervalMs?: number;
};

export async function ensureLoopbackServerReady(opts: EnsureReadyOptions = {}): Promise<void> {
if (__loopbackReadyPromise != null) return __loopbackReadyPromise;
const minRetryIntervalMs = opts.minRetryIntervalMs ?? LOOPBACK_RETRY_MIN_INTERVAL_MS;
if (Date.now() - __loopbackLastFailureAt < minRetryIntervalMs) {
// Anti-storm window: reject immediately without re-probing — callers in the
// same burst all fall back to the in-process route.
throw new Error(
`loopback server not ready (probe failed ${Date.now() - __loopbackLastFailureAt}ms ago; retry after ${minRetryIntervalMs}ms)`
);
}
__loopbackReadyPromise = (async () => {
const f = opts.fetch ?? fetchModelSyncInternal;
const maxWaitMs = opts.maxWaitMs ?? 30_000;
Expand Down Expand Up @@ -213,12 +229,23 @@ export async function ensureLoopbackServerReady(opts: EnsureReadyOptions = {}):
}
throw new Error(`loopback server not ready within ${maxWaitMs}ms: ${String(lastErr)}`);
})();
void __loopbackReadyPromise.catch((err) => {
// Memorize success only: release the gate for the next probe, bounded by
// the anti-storm window. The handler does not reject — callers receive the
// rejection of the original promise.
__loopbackLastFailureAt = Date.now();
__loopbackReadyPromise = null;
console.warn(
`[ModelSync] Loopback server readiness probe failed; falling back to in-process route: ${String(err)}`
);
});
return __loopbackReadyPromise;
}

/** Test helper: reset the cached promise so tests can re-exercise the probe. */
export function __resetLoopbackReadinessForTests(): void {
__loopbackReadyPromise = null;
__loopbackLastFailureAt = 0;
}

// ---------------------------------------------------------------------------
Expand Down Expand Up @@ -284,11 +311,9 @@ export async function selfFetchWithRetry(
if (opts.skipReadinessGate !== true) {
try {
await ensureLoopbackServerReady({ fetch: f });
} catch (err) {
} catch {
// 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)}`
);
// The transition is logged once by the gate itself (per probe, not per caller).
if (opts.inProcessFallback) {
return opts.inProcessFallback();
}
Expand Down
134 changes: 133 additions & 1 deletion tests/unit/api/sync-models-readiness.test.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { test, beforeEach } from "node:test";
import { test, beforeEach, mock } from "node:test";
import assert from "node:assert/strict";
import {
selfFetchWithRetry,
Expand Down Expand Up @@ -248,3 +248,135 @@ test("sanity: without readiness gate, 17 callers retry independently (amplificat
"without gate, callers retry independently, got " + modelFetchCalls + " (expected >17)"
);
});

// ---------------------------------------------------------------------------
// Failure-memory tests: a rejected probe must not condemn the process
// ---------------------------------------------------------------------------

test("ensureLoopbackServerReady: a failed probe is re-attempted after the retry window (no permanent failure memory)", async () => {
__resetLoopbackReadinessForTests();
let probeCalls = 0;
let serverIsUp = false;
const mockFetch = async (_url) => {
probeCalls++;
if (!serverIsUp) throw new Error("ECONNREFUSED");
return new Response("", { status: 200 });
};

// First probe: server down -> rejection after maxWaitMs (no prior failure,
// so the default 30s retry window is inactive)
await assert.rejects(
() =>
ensureLoopbackServerReady({
fetch: mockFetch,
maxWaitMs: 50,
pollMs: 10,
}),
/loopback server not ready/
);
const callsAfterFirstFailure = probeCalls;

// Server comes up shortly after the first failure
setTimeout(() => {
serverIsUp = true;
}, 60);

// Inside the retry window: must reject WITHOUT re-probing (storm bound)
await assert.rejects(
() =>
ensureLoopbackServerReady({
fetch: mockFetch,
maxWaitMs: 50,
pollMs: 10,
minRetryIntervalMs: 500,
}),
/loopback server not ready/
);
assert.equal(
probeCalls,
callsAfterFirstFailure,
"callers inside the retry window must not re-probe"
);

// After the window: a fresh probe runs and succeeds (server is up)
await new Promise((r) => setTimeout(r, 15));
await ensureLoopbackServerReady({
fetch: mockFetch,
maxWaitMs: 500,
pollMs: 10,
minRetryIntervalMs: 5,
});
assert.ok(
probeCalls > callsAfterFirstFailure,
"a later caller must re-probe after the retry window"
);
});

test("ensureLoopbackServerReady: 54 concurrent callers inside the retry window share one rejection (no probe storm)", async () => {
__resetLoopbackReadinessForTests();
let probeCalls = 0;
const mockFetch = async (_url) => {
probeCalls++;
throw new Error("ECONNREFUSED");
};

await assert.rejects(
() =>
ensureLoopbackServerReady({
fetch: mockFetch,
maxWaitMs: 50,
pollMs: 10,
minRetryIntervalMs: 500,
}),
/loopback server not ready/
);
const afterFirst = probeCalls;

const results = await Promise.allSettled(
Array.from({ length: 54 }, () =>
ensureLoopbackServerReady({
fetch: mockFetch,
maxWaitMs: 50,
pollMs: 10,
minRetryIntervalMs: 500,
})
)
);

assert.equal(probeCalls, afterFirst, "54 callers inside the window must not launch 54 probes");
assert.ok(
results.every((r) => r.status === "rejected"),
"all burst callers reject immediately"
);
});

test("ensureLoopbackServerReady: readiness failure is logged once per probe, not once per caller", async () => {
__resetLoopbackReadinessForTests();
const warns: string[] = [];
const warnMock = mock.method(console, "warn", (...args: unknown[]) => {
warns.push(args.map(String).join(" "));
});
try {
const mockFetch = async (_url) => {
throw new Error("ECONNREFUSED");
};
await Promise.allSettled(
Array.from({ length: 17 }, () =>
ensureLoopbackServerReady({
fetch: mockFetch,
maxWaitMs: 50,
pollMs: 10,
minRetryIntervalMs: 500,
})
)
);
} finally {
warnMock.mock.restore();
}
const readinessWarns = warns.filter((w) => w.includes("readiness probe failed"));
assert.equal(
readinessWarns.length,
1,
`expected exactly 1 readiness warn, got ${readinessWarns.length}`
);
});
Loading