mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-08-21 22:52:19 +03:00
fix(providers): stop the loopback readiness gate from memorizing failure (#10903)
Validado no worktree combinado: typecheck:core, changelog-integrity, complexity, cognitive-complexity, file-size, lint todos verdes. Fix real bem documentado (loopback readiness gate memorizava falha permanentemente + log-spam por caller). CI vermelho é o base-red já rastreado em #9985. Obrigado!
This commit is contained in:
1
changelog.d/fixes/10903-loopback-gate-memory-success.md
Normal file
1
changelog.d/fixes/10903-loopback-gate-memory-success.md
Normal file
@@ -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))
|
||||
@@ -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;
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
@@ -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}`
|
||||
);
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user