Skip to content
Merged
1 change: 1 addition & 0 deletions changelog.d/fixes/13017-explicit-inactive-probe.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
- **fix(auth):** an explicit connection pin may probe a quota-disabled row once and re-enable it on success ([#12874](https://github.com/diegosouzapw/OmniRoute/issues/12874)) ([#13017](https://github.com/diegosouzapw/OmniRoute/pull/13017))
2 changes: 1 addition & 1 deletion src/lib/quota/connectionRecovery.ts
Original file line number Diff line number Diff line change
Expand Up @@ -160,7 +160,7 @@ export function isRecoverableCooldownConnection(
* Pure — `nowMs` and `reprobeMs` are injected so callers/tests control the clock.
*/

const EXPIRED_REPROBE_BLOCKLIST = new Set([
export const EXPIRED_REPROBE_BLOCKLIST = new Set([
"account_deactivated",
"invalid_grant",
"unrecoverable_refresh_error",
Expand Down
8 changes: 8 additions & 0 deletions src/sse/handlers/chatHelpers.ts
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import {
markAccountUnavailable,
buildExhaustionOptions,
} from "../services/auth";
import { maybeReactivateAfterExplicitProbe } from "../services/explicitInactiveProbe";
import { connectionHasExtraKeys } from "@omniroute/open-sse/services/apiKeyRotator.ts";
import { createBuiltinAutoCombo } from "@omniroute/open-sse/services/autoCombo/builtinCatalog.ts";
import * as log from "../utils/logger";
Expand Down Expand Up @@ -535,6 +536,13 @@ export async function executeChatWithBreaker({
onRequestSuccess: async () => {
if (isShadowTraffic) return;
await clearAccountError(credentials.connectionId, credentials);
await maybeReactivateAfterExplicitProbe({
connectionId: credentials.connectionId,
reactivatedFromInactive: credentials.reactivatedFromInactive,
isShadowTraffic,
requestedModel: model,
provider,
});
},
onStreamFailure: async (failure: any) => {
if (isShadowTraffic) return;
Expand Down
50 changes: 46 additions & 4 deletions src/sse/services/auth.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ import {
getCachedRawProviderConnections,
getCachedProviderNodes,
getCachedSettings,
getCachedProviderConnectionById,
} from "@/lib/db/readCache";
import {
getProviderConnections,
Expand Down Expand Up @@ -150,6 +151,12 @@ import {
planSessionAffinityConnection,
syncSessionAffinityRuntimeFields,
} from "./sessionAffinityPin";
import {
EXPLICIT_INACTIVE_PROBE_INTERVAL_MS,
lastExplicitProbeTime,
noteExplicitProbe,
selectExplicitInactiveProbe,
} from "./explicitInactiveProbe";
import {
isAnonymousFallbackDisabledBySettings,
isNoAuthProviderBlockedBySettings,
Expand Down Expand Up @@ -1062,7 +1069,10 @@ async function hydrateAccountProxyReferences(
async function materializeConnection(
connection: ProviderConnectionView,
options: CredentialSelectionOptions,
extra: DeferredLeaseSelection & { exclusiveLease?: ExclusiveConnectionLease } = {}
extra: DeferredLeaseSelection & {
exclusiveLease?: ExclusiveConnectionLease;
reactivatedFromInactive?: boolean;
} = {}
) {
const providerSpecificData = await hydrateAccountProxyReferences(connection.providerSpecificData);
const apiKeyHealth = providerSpecificData.apiKeyHealth as Record<string, KeyHealth> | undefined;
Expand Down Expand Up @@ -1261,6 +1271,29 @@ export async function getProviderCredentials(
if (allowedConnections && allowedConnections.length > 0) {
connections = connections.filter((conn) => allowedConnections.includes(conn.id));
}
let explicitProbeKind: "probe" | "suppressed" | "skip" = "skip";
if (forcedConnectionId && !connections.some((c) => c.id === forcedConnectionId)) {
const pinnedRaw = await getCachedProviderConnectionById(forcedConnectionId);
const pinnedRow = pinnedRaw ? toProviderConnection(pinnedRaw) : null;
const nowMs = Date.now();
const decision = selectExplicitInactiveProbe({
forcedConnectionId,
activeConnections: connections,
pinnedRow,
providersToSearch,
allowedConnectionIds: allowedConnections ?? null,
nowMs,
lastProbeAtMs: lastExplicitProbeTime(forcedConnectionId),
intervalMs: EXPLICIT_INACTIVE_PROBE_INTERVAL_MS,
});
explicitProbeKind = decision.kind;
if (decision.kind === "probe" && pinnedRow) {
noteExplicitProbe(forcedConnectionId, nowMs);
connections = [pinnedRow];
}
}
const probeStamp =
explicitProbeKind === "probe" ? { reactivatedFromInactive: true as const } : {};
const forcedConnectionEligible = connections.some((conn) => conn.id === forcedConnectionId);
if (options.lease && forcedConnectionId && !forcedConnectionEligible) return null;
if (options.lease?.mode === "request" && forcedConnectionId) {
Expand Down Expand Up @@ -1487,7 +1520,7 @@ export async function getProviderCredentials(
connectionFilterStatus.set(c.id, "modelNotAdvertised");
return false;
}
if (!allowSuppressedConnections) {
if (!allowSuppressedConnections && explicitProbeKind !== "probe") {
if (!allowRateLimitedConnections && isAccountUnavailable(c.rateLimitedUntil)) {
connectionFilterStatus.set(c.id, "rateLimited");
return false;
Expand Down Expand Up @@ -2120,6 +2153,7 @@ export async function getProviderCredentials(
return materializeConnection(connection, options, {
commitSelectionSideEffects,
selectNextLeaseCandidate,
...probeStamp,
});
}
let claim = mutateExclusiveConnectionLease(
Expand All @@ -2139,7 +2173,12 @@ export async function getProviderCredentials(
exclusiveLease = claim.lease;
await commitSelectionSideEffects?.();
if (options.materializeCredentials === false) {
return { exclusiveLease, connectionId: connection.id, provider: connection.provider };
return {
exclusiveLease,
connectionId: connection.id,
provider: connection.provider,
...probeStamp,
};
}
}

Expand All @@ -2150,7 +2189,10 @@ export async function getProviderCredentials(
);
}

return materializeConnection(connection, options, { exclusiveLease });
return materializeConnection(connection, options, {
exclusiveLease,
...probeStamp,
});
} finally {
selectionLock?.release();
}
Expand Down
126 changes: 126 additions & 0 deletions src/sse/services/explicitInactiveProbe.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,126 @@
import { updateProviderConnection } from "@/lib/db/providers";
import { EXPIRED_REPROBE_BLOCKLIST } from "@/lib/quota/connectionRecovery";

export const EXPLICIT_PROBE_BLOCKLIST = EXPIRED_REPROBE_BLOCKLIST;

export const RECOVERABLE_INACTIVE_TEST_STATUSES = new Set([
"active",
"success",
"credits_exhausted",
"unavailable",
"error",
"",
]);

export function isRecoverableInactiveConnection(
conn: {
isActive?: boolean;
testStatus?: string | null;
lastErrorType?: string | null;
rateLimitedUntil?: string | null;
},
nowMs: number = Date.now()
): boolean {
if (conn.isActive !== false) return false;
const status = (conn.testStatus || "").trim().toLowerCase();
if (status === "banned") return false;
const err = (conn.lastErrorType || "").trim().toLowerCase();
if (EXPLICIT_PROBE_BLOCKLIST.has(err)) return false;
if (status === "expired") return true;
if (status === "unavailable") {
const until = conn.rateLimitedUntil;
if (until) {
const ms = Date.parse(until);
if (Number.isFinite(ms) && ms > nowMs) return false;
}
}
return RECOVERABLE_INACTIVE_TEST_STATUSES.has(status);
}

export function selectExplicitInactiveProbe(params: {
forcedConnectionId: string | null;
activeConnections: { id: string }[];
pinnedRow: {
id: string;
provider?: string | null;
isActive?: boolean;
testStatus?: string | null;
lastErrorType?: string | null;
rateLimitedUntil?: string | null;
} | null;
providersToSearch: string[];
allowedConnectionIds: string[] | null;
nowMs: number;
lastProbeAtMs: number | null;
intervalMs: number;
}): { kind: "probe" } | { kind: "suppressed" } | { kind: "skip" } {
const id = params.forcedConnectionId;
if (!id) return { kind: "skip" };
if (params.activeConnections.some((c) => c.id === id)) return { kind: "skip" };
const row = params.pinnedRow;
if (!row || row.id !== id) return { kind: "skip" };
if (
params.allowedConnectionIds &&
params.allowedConnectionIds.length > 0 &&
!params.allowedConnectionIds.includes(id)
) {
return { kind: "skip" };
}
const prov = (row.provider || "").trim();
if (prov && !params.providersToSearch.includes(prov)) return { kind: "skip" };
if (!isRecoverableInactiveConnection(row, params.nowMs)) return { kind: "skip" };
if (params.lastProbeAtMs != null && params.nowMs - params.lastProbeAtMs < params.intervalMs) {
return { kind: "suppressed" };
}
return { kind: "probe" };
}

export const EXPLICIT_INACTIVE_PROBE_INTERVAL_MS = 60_000;
const MAX_PROBE_MAP = 4096;
const lastExplicitProbeAtMs = new Map<string, number>();

export function noteExplicitProbe(id: string, nowMs: number): void {
lastExplicitProbeAtMs.set(id, nowMs);
if (lastExplicitProbeAtMs.size > MAX_PROBE_MAP) {
const oldest = lastExplicitProbeAtMs.keys().next().value;
if (oldest !== undefined) lastExplicitProbeAtMs.delete(oldest);
}
}

export function lastExplicitProbeTime(id: string): number | null {
return lastExplicitProbeAtMs.get(id) ?? null;
}

export function resetExplicitProbeMapForTests(): void {
lastExplicitProbeAtMs.clear();
}

export async function reactivateRecoveredConnection(connectionId: string): Promise<void> {
await updateProviderConnection(connectionId, { isActive: true });
}

export async function maybeReactivateAfterExplicitProbe(
input: {
connectionId: string;
reactivatedFromInactive?: boolean;
explicitProbeSuppressed?: boolean;
isShadowTraffic?: boolean;
allowSuppressedConnections?: boolean;
requestedModel?: string | null;
provider?: string | null;
},
reactivate: (connectionId: string) => Promise<void> = reactivateRecoveredConnection
): Promise<void> {
if (!input.reactivatedFromInactive) return;
if (input.explicitProbeSuppressed) return;
if (input.isShadowTraffic) return;
if (input.allowSuppressedConnections) return;
if (
input.provider === "openrouter" &&
typeof input.requestedModel === "string" &&
input.requestedModel.includes(":free")
) {
return;
}
await reactivate(input.connectionId);
}
3 changes: 2 additions & 1 deletion stryker.conf.json
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,7 @@
"tests/unit/alibaba-free-tier-exhaustion.test.ts",
"tests/unit/anthropic-thinking-signature-recovery.test.ts",
"tests/unit/agy-family-not-connection-cooldown.test.ts",
"tests/unit/agy-quota-exhaustion-threshold.test.ts",
"tests/unit/agy-quota-exhaustion-threshold.test.ts",
"tests/unit/antigravity-429-quota-cooldown.test.ts",
"tests/unit/antigravity-429-quota-tdd.test.ts",
"tests/unit/antigravity-prefer-stored-project.test.ts",
Expand Down Expand Up @@ -261,6 +261,7 @@
"tests/unit/executor-contract-violation-terminal.test.ts",
"tests/unit/executor-devin-cli-agentic-acp.test.ts",
"tests/unit/executor-web-cookie-sweep.test.ts",
"tests/unit/explicit-inactive-probe-w2.test.ts",
"tests/unit/false-terminal-401-quota.test.ts",
"tests/unit/follow-up-transcript.test.ts",
"tests/unit/format-provider-error-cause.test.ts",
Expand Down
Loading
Loading