From ae2ba35852d4e5a55486a1c0e6a779105564fd6d Mon Sep 17 00:00:00 2001 From: Dizzle <112548150+maxmad64bis@users.noreply.github.com> Date: Sat, 26 Sep 2026 00:44:37 +0200 Subject: [PATCH] fix(proxies): rank pool members sharing a refused egress address last (#14657) Adds PROXY_POOL_SHARED_EGRESS_ORDER (default off): for egress-bucketed providers, a pool member sharing a recently refused member's observed egress address ranks just below healthy members (order only, never excluded). Maintainer rework: reconciled with release/v3.8.51 (kept #14755's maybeEmitPoolExhausted and the #14688 default-on refusal memory; flag-off functional case pins PROXY_SKIP_RECENTLY_FAILED=false; registry/catalog totals 80 flags / Network 22 after #14659). Focused proxy pool/rotation/refusal/flag suites 305/305; typecheck:core and open-sse typecheck clean; file-size green; env-doc-sync red only on the inherited tip TAVILY_BASE_URL (#14442). Thank you @maxmad64bis! --- .env.example | 6 + .../fixes/14657-shared-egress-pool-order.md | 1 + docs/reference/ENVIRONMENT.md | 1 + docs/reference/FEATURE_FLAGS.md | 5 +- src/i18n/messages/en.json | 1 + src/i18n/messages/vi.json | 1 + src/lib/db/proxies/rotation.ts | 431 ++++++++++++++---- src/lib/db/proxyLogs.ts | 28 ++ .../constants/featureFlagDefinitions.ts | 12 + src/shared/utils/featureFlags.ts | 17 + tests/unit/feature-flags-settings.test.ts | 4 +- .../proxy-pool-shared-egress-order.test.ts | 307 +++++++++++++ .../unit/server-owned-tool-loop-flag.test.ts | 5 +- 13 files changed, 733 insertions(+), 86 deletions(-) create mode 100644 changelog.d/fixes/14657-shared-egress-pool-order.md create mode 100644 tests/unit/proxy-pool-shared-egress-order.test.ts diff --git a/.env.example b/.env.example index 3d5f3e1140..79b9b33724 100644 --- a/.env.example +++ b/.env.example @@ -836,6 +836,12 @@ NEXT_PUBLIC_ENABLE_SOCKS5_PROXY=true # Lowering the cap also shortens streak forgetting: entries purge 2 x max after expiry. # PROXY_QUOTA_429_MAX_MS=3600000 +# Opt-in feature flag (default off; a dashboard DB override wins over this value): for +# providers whose quota is bucketed by egress address, rank a pool member sharing a recently +# refused member's observed egress address just below healthy members. Order only, never +# excluded. Needs PROXY_SKIP_RECENTLY_FAILED, which produces the refusal signal it reads. +# PROXY_POOL_SHARED_EGRESS_ORDER=false + # Rebound window capping repeat set-aside webhook alerts per pool member (read on each emit; no restart needed). # Bounded to 1000..3600000; out-of-range or unreadable falls back to 300000 with a silent default. # Used by: src/lib/proxyEvents/proxyTransitionBridge.ts. diff --git a/changelog.d/fixes/14657-shared-egress-pool-order.md b/changelog.d/fixes/14657-shared-egress-pool-order.md new file mode 100644 index 0000000000..ace842a175 --- /dev/null +++ b/changelog.d/fixes/14657-shared-egress-pool-order.md @@ -0,0 +1 @@ +- **fix(proxies):** a pool member sharing a recently refused egress address is served last for providers whose quota is bucketed by egress address — order only, nobody is ever excluded, and other providers keep the plain rotation (covers refused members only; recently served members are a follow-up). ([#14657](https://github.com/diegosouzapw/OmniRoute/pull/14657)) — thanks @maxmad64bis diff --git a/docs/reference/ENVIRONMENT.md b/docs/reference/ENVIRONMENT.md index b5ccbed63a..7b3c876bd3 100644 --- a/docs/reference/ENVIRONMENT.md +++ b/docs/reference/ENVIRONMENT.md @@ -381,6 +381,7 @@ Route upstream LLM provider calls through an HTTP or SOCKS5 proxy for egress con | `SELECTOR_CONTROL_ALLOWLIST` | _(unset)_ | `src/lib/proxySubscription/selectorGuard.ts` | Allow-list for local-core selector-control targets beyond loopback (comma/space separated hosts or IPs, `*` opens LAN/private but never metadata/link-local). Loopback over http/https is allowed by default. A dashboard DB value wins over this variable when set and non-empty. | | `PROXY_QUOTA_429_BASE_MS` | `120000` | `open-sse/utils/proxyRefusalMemory.ts` | Base period (ms) of the 429 set-aside, doubled on each repeat. Read once at startup. Bounded to `1000..3600000`; out-of-range or unreadable falls back to the default with a warning. | | `PROXY_QUOTA_429_MAX_MS` | `3600000` | `open-sse/utils/proxyRefusalMemory.ts` | Cap (ms) of the 429 set-aside. Read once at startup. Bounded to `1000..3600000`; out-of-range or unreadable falls back to the default with a warning. When the cap is below the base, it falls back to the default. Lowering the cap also shortens streak forgetting: entries purge `2 x max` after expiry. | +| `PROXY_POOL_SHARED_EGRESS_ORDER` | `false` | `src/shared/utils/featureFlags.ts` | Opt-in feature flag (see [FEATURE_FLAGS.md](./FEATURE_FLAGS.md); a dashboard DB override wins). For providers whose quota is bucketed by egress address, rank a pool member sharing a recently refused member's observed egress address just below healthy members. Order only, never excluded. Needs `PROXY_SKIP_RECENTLY_FAILED`, which produces the refusal signal it reads. | | `PROXY_WEBHOOK_REBOUND_MS` | `300000` | `src/lib/proxyEvents/proxyTransitionBridge.ts` | Rebound window (ms) capping repeat `proxy.set_aside` webhook alerts per pool member so a burst of refusals sends one event. Read on each emit. Bounded to `1000..3600000`; out-of-range or unreadable falls back to the default silently. | | `HTTP_PROXY` | _(unset)_ | Node.js standard | HTTP proxy for upstream calls. | | `HTTPS_PROXY` | _(unset)_ | Node.js standard | HTTPS proxy for upstream calls. | diff --git a/docs/reference/FEATURE_FLAGS.md b/docs/reference/FEATURE_FLAGS.md index 80a21a7de5..bf37f3c06f 100644 --- a/docs/reference/FEATURE_FLAGS.md +++ b/docs/reference/FEATURE_FLAGS.md @@ -46,7 +46,7 @@ A boolean flag is considered **enabled** when its effective value is `"true"`, ## Flag Catalog -79 flags across 6 categories. **Default** is the definition default — the value +80 flags across 6 categories. **Default** is the definition default — the value used when neither a DB override nor an environment variable is present. ### Security (10) @@ -64,7 +64,7 @@ used when neither a DB override nor an environment variable is present. | `AUTH_LOG_INCLUDE_ACCOUNT_ID` | boolean | `false` | Include account prefix in AUTH log lines (e.g. "Using account: abc12345..."). Disabled by default so account identifiers are redacted from shared/multi-tenant process logs. Independent from Debug Mode; flipping Debug Mode does not reveal this. | | `OMNIROUTE_OIDC_DISABLE_PASSWORD_LOGIN` | boolean | `false` | When OIDC is enabled, disable password login so users can only authenticate via OIDC Single Sign-On. When disabled (default), both password login and OIDC are available. | -### Network (21) +### Network (22) | Key | Type | Default | Restart | Description | | ----------------------------------------------- | ------- | ------- | ------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | @@ -76,6 +76,7 @@ used when neither a DB override nor an environment variable is present. | `NETWORK_ROTATION_SHARED_EGRESS_GUARD` | boolean | `true` | | On a network exception (timeout, connection refused/reset) for a multi-account rotation executor, when the failing account has no dedicated proxy, apply a short cooldown and skip other proxy-less accounts for the rest of the request instead of retrying each one. On by default (safe: no egress IP change, only reduces latency/cooldown risk on shared-egress accounts). Disable to restore immediate propagation on the first proxy-less throw. | | `ROTATION_ATTRIBUTION` | boolean | `false` | | Opencode rotation records which account served or was skipped (masked ids only, never full account ids) and links proxy log entries to their request, so the operator can tell skipped accounts apart from unused ones. Off by default. | | `PROXY_SKIP_RECENTLY_FAILED` | boolean | `true` | | Proxy pools and the per-account rotation of opencode stop re-serving a proxy that just failed (refused TCP probe, or a 429 received through it) for a per-process period that doubles on each repeat, up to a cap. No proxy status is written; with every candidate set aside the choice is unchanged. On by default; `false` restores plain selection. | +| `PROXY_POOL_SHARED_EGRESS_ORDER` | boolean | `false` | | For providers whose quota is bucketed by egress address, rank a pool member sharing a recently refused member's observed egress address just below healthy members. Order only, never excluded. Needs PROXY_SKIP_RECENTLY_FAILED, which produces the refusal signal it reads. Off by default. | | `PROXY_POOL_EGRESS_OBSERVATION` | boolean | `false` | | Show, under a proxy pool in the dashboard, how many observed egress IPs served its members over the last 24 h and how many connections used them. Read-only, computed from the proxy log, never used for routing. Off by default. | | `OPENCODE_RESPONSES_STALL_ROTATION` | boolean | `false` | | For the OpenCode executor, watch the first body byte of a streamed Responses reply (window: `RESPONSES_FIRST_BYTE_TIMEOUT_MS`, default `15000`). A 2xx Responses stream that stays silent past the window is treated as stalled: the account is cooled down and the request rotates to the next account once; a second stall fails fast. Off by default: stalled streams keep today's wait until the stream readiness timeout. | | `OPENCODE_USER_BLOCKED_ROTATION` | boolean | `false` | | OpenCode executor: on a 403/451 carrying a `user_blocked` refusal (not geo, not a Cloudflare fingerprint rejection), cool the refused account down and rotate to the next account at most once per request; a second refusal is returned as-is, without a success mark. Off by default: routing around an upstream user block can look like evasion and spread the flag across the fleet. | diff --git a/src/i18n/messages/en.json b/src/i18n/messages/en.json index e02c662595..5aa91eec63 100644 --- a/src/i18n/messages/en.json +++ b/src/i18n/messages/en.json @@ -994,6 +994,7 @@ "featureFlagChatVirtualLanesEnabledDescription": "Enable per-tenant adaptive virtual admission lanes for provider dispatch (#9654): one tenant's burst no longer 503s another. The OMNIROUTE_CHAT_VIRTUAL_LANES env var wins over this dashboard override; changes take effect at server restart.", "featureFlagFreeBadgeRequiresProviderFreeTierDescription": "Dashboard provider pages: show the Free badge only on signals the provider honors — drops the display-name heuristic, non-boolean free fields and :free suffixes on registered providers without a documented free tier. Off keeps the historical badge rule.", "featureFlagProxySkipRecentlyFailedDescription": "Proxy pools and the per-account rotation of opencode stop re-serving a proxy that just failed (refused TCP probe, or a 429 received through it) for a per-process period that doubles on each repeat, up to a cap. No proxy status is written; with every candidate set aside the choice is unchanged. On by default; `false` restores plain selection.", + "featureFlagProxyPoolSharedEgressOrderDescription": "For providers whose quota is bucketed by egress address, rank a pool member sharing a recently refused member's observed egress address just below healthy members. Order only, never excluded. Needs PROXY_SKIP_RECENTLY_FAILED, which produces the refusal signal it reads. Off by default: selection order is exactly the plain rotation.", "featureFlagProxyPoolEgressObservationDescription": "Show, under a proxy pool in the dashboard, how many observed egress IPs served its members over the last 24 h, how many connections used them and the most seen behind one IP. Read-only, computed from the proxy log, never used for routing. Off by default: the pool editor is unchanged and the observation route answers null.", "featureFlagProxyHealthBlockedResetsStreakDescription": "In the proxy health sweep, let a probe the target refused (401/403/429: the proxy relayed, the destination refused this egress IP) reset the consecutive-failure streak of the proxy, like a served probe. Off by default: a refusal stays neutral and keeps the streak. A 5xx stays inconclusive either way, and a refusal never removes, disables or re-activates a proxy.", "featureFlagBatchAndFileAutoCleanupEnabledDescription": "Let the automatic cleanup sweep delete terminal (completed/failed/cancelled/expired) Batch API jobs older than OMNIROUTE_BATCH_RETENTION_DAYS, along with their per-line checkpoints, and clear the BLOB content of uploaded files past their own expires_at. Off by default: every existing install keeps this data exactly as before until an operator opts in. The operator-triggered DELETE /api/v1/batches/delete-completed route is unaffected either way -- it is a separate, unconditional public API contract.", diff --git a/src/i18n/messages/vi.json b/src/i18n/messages/vi.json index 7727322961..6e7090670d 100644 --- a/src/i18n/messages/vi.json +++ b/src/i18n/messages/vi.json @@ -994,6 +994,7 @@ "featureFlagChatVirtualLanesEnabledDescription": "Bật làn tiếp nhận ảo thích ứng cho từng đối tượng thuê (tenant) để phân phối nhà cung cấp (#9654): một đợt bùng phát của tenant này không còn trả 503 cho tenant khác. Biến môi trường OMNIROUTE_CHAT_VIRTUAL_LANES được ưu tiên hơn cài đặt bảng điều khiển này; các thay đổi có hiệu lực khi khởi động lại máy chủ.", "featureFlagFreeBadgeRequiresProviderFreeTierDescription": "Các trang nhà cung cấp trên bảng điều khiển: chỉ hiển thị huy hiệu Miễn phí đối với các tín hiệu được nhà cung cấp công nhận — loại bỏ phương pháp suy đoán dựa trên tên hiển thị, các trường miễn phí không phải kiểu boolean và hậu tố :free trên các nhà cung cấp đã đăng ký nhưng không có gói miễn phí được ghi rõ trong tài liệu. Khi tắt, quy tắc huy hiệu trước đây vẫn được áp dụng.", "featureFlagProxySkipRecentlyFailedDescription": "Các pool proxy và cơ chế xoay vòng theo tài khoản của opencode ngừng dùng lại một proxy vừa thất bại (thăm dò TCP bị từ chối, hoặc nhận 429 qua proxy đó) trong một khoảng thời gian theo tiến trình, tăng gấp đôi sau mỗi lần lặp lại, tối đa đến một giới hạn. Không ghi trạng thái proxy nào; khi mọi ứng viên đều bị tạm gác, lựa chọn vẫn như cũ. Bật theo mặc định; `false` khôi phục lựa chọn thông thường.", + "featureFlagProxyPoolSharedEgressOrderDescription": "Đối với các nhà cung cấp áp dụng hạn ngạch theo địa chỉ egress, xếp hạng thành viên pool có cùng địa chỉ egress quan sát được với thành viên vừa bị từ chối xuống dưới các thành viên khỏe mạnh. Chỉ sắp xếp thứ tự, không bao giờ loại trừ. Cần PROXY_SKIP_RECENTLY_FAILED, nguồn tín hiệu từ chối mà nó đọc. Mặc định tắt: thứ tự chọn đúng như vòng xoay thông thường.", "featureFlagProxyPoolEgressObservationDescription": "Hiển thị, bên dưới một nhóm proxy trong bảng điều khiển, số IP đầu ra quan sát được đã phục vụ các thành viên của nhóm trong 24 giờ qua, số kết nối đã sử dụng các IP đó và số thành viên nhiều nhất được ghi nhận phía sau một IP. Chỉ đọc, được tính từ nhật ký proxy và không bao giờ được dùng để định tuyến. Tắt theo mặc định: trình chỉnh sửa nhóm không thay đổi và tuyến quan sát trả về null.", "featureFlagProxyHealthBlockedResetsStreakDescription": "Trong lượt kiểm tra sức khỏe proxy, cho phép một lần thăm dò bị đích từ chối (401/403/429: proxy đã chuyển tiếp, đích từ chối IP đầu ra này) đặt lại chuỗi lỗi liên tiếp của proxy, giống như một lần thăm dò được phục vụ. Tắt theo mặc định: lần từ chối vẫn trung lập và giữ nguyên chuỗi. Lỗi 5xx vẫn không kết luận trong mọi trường hợp, và lần từ chối không bao giờ xóa, vô hiệu hóa hay kích hoạt lại proxy.", "featureFlagBatchAndFileAutoCleanupEnabledDescription": "Cho phép lượt dọn dẹp tự động xóa các batch job ở trạng thái cuối (completed/failed/cancelled/expired) của Batch API cũ hơn OMNIROUTE_BATCH_RETENTION_DAYS, cùng với các checkpoint theo từng dòng của chúng, và xóa nội dung BLOB của các tệp đã tải lên quá hạn expires_at riêng. Mặc định tắt: mọi bản cài đặt hiện có giữ nguyên dữ liệu này cho đến khi người vận hành bật tùy chọn này. Tuyến DELETE /api/v1/batches/delete-completed do người vận hành kích hoạt không bị ảnh hưởng theo bất kỳ cách nào -- đây là một hợp đồng API công khai riêng biệt và vô điều kiện.", diff --git a/src/lib/db/proxies/rotation.ts b/src/lib/db/proxies/rotation.ts index 8ad59e9fc7..1f154c75b5 100644 --- a/src/lib/db/proxies/rotation.ts +++ b/src/lib/db/proxies/rotation.ts @@ -6,6 +6,7 @@ // dependency back on ../proxies.ts — mutators that also need to bump the registry // generation counter (addProxyToScopePool, removeProxyFromScopePool, // setScopeRotationStrategy) stay in ../proxies.ts and import the pure helpers here. +import { isIP } from "node:net"; import { randomInt } from "crypto"; import { getDbInstance } from "../core"; import { pickByLatency } from "../proxyLatency"; @@ -14,9 +15,14 @@ import { isProxyAvoided, proxyEgressKey, } from "@omniroute/open-sse/utils/proxyRefusalMemory.ts"; +import { isEgressBucketedLockScope } from "@omniroute/open-sse/config/providerErrorRules.ts"; import { maybeEmitPoolExhausted } from "@/lib/proxyEvents/proxyTransitionBridge"; -import { isProxySkipRecentlyFailedEnabled } from "@/shared/utils/featureFlags"; +import { + isProxySkipRecentlyFailedEnabled, + isProxyPoolSharedEgressOrderEnabled, +} from "@/shared/utils/featureFlags"; import { getCachedProxyHealth } from "@/lib/proxyHealth"; +import { getRecentEgressIpsForProxy } from "../proxyLogs"; import type { JsonRecord, ProxyScope, ProxyRotationStrategy } from "./types"; import { PROXY_ROTATION_STRATEGIES, DEFAULT_PROXY_ROTATION_STRATEGY } from "./types"; import { @@ -177,6 +183,7 @@ function firstEligibleFrom(start: number, eligible: number[], size: number): num export interface PoolRankSignals { isAvoided: (key: string | null) => boolean; probeHealth: (url: string) => boolean | null; + sharesHotEgress?: (candidate: unknown) => boolean; } const DEFAULT_POOL_RANK_SIGNALS: PoolRankSignals = { @@ -189,6 +196,34 @@ const RELAY_TYPES = new Set(["vercel", "deno", "cloudflare"]); // Same scheme defaults as proxyConfigToUrl() so the rebuilt URL hits the probe cache key. const DEFAULT_PORTS: Record = { http: "8080", https: "443", socks5: "1080" }; +function isRankableProxyType(type: string): boolean { + return !RELAY_TYPES.has(type) && type in DEFAULT_PORTS; +} + +function probePortForType(record: Record, type: string): string { + const parsed = Number(record.port); + if (record.port && Number.isInteger(parsed) && parsed >= 1 && parsed <= 65535) { + return String(parsed); + } + return DEFAULT_PORTS[type]; +} + +function probeAuthPart(record: Record): string { + const username = typeof record.username === "string" ? record.username : ""; + const password = typeof record.password === "string" ? record.password : ""; + if (!username && !password) return ""; + return `${encodeURIComponent(username)}:${encodeURIComponent(password)}@`; +} + +function probeFamilyMarker(record: Record): string { + const family = typeof record.family === "string" ? record.family : ""; + return family === "ipv4" || family === "ipv6" ? `?family=${family}` : ""; +} + +function bracketIpv6Host(host: string): string { + return host.includes(":") && !host.startsWith("[") ? `[${host}]` : host; +} + // Rebuild the probe URL for a pool row the way the dispatcher builds it // (`proxyConfigToUrl`, read-only replica): scheme + encoded auth + host + port, // with the `?family=` marker when set. Null when the row cannot egress. @@ -198,38 +233,191 @@ function candidateProbeUrl(row: unknown): string | null { const host = typeof record.host === "string" ? record.host : ""; if (!host) return null; const type = String(record.type || "http").toLowerCase(); - if (RELAY_TYPES.has(type) || !(type in DEFAULT_PORTS)) return null; - const parsed = Number(record.port); - const port = - record.port && Number.isInteger(parsed) && parsed >= 1 && parsed <= 65535 - ? String(parsed) - : DEFAULT_PORTS[type]; - const bracketed = host.includes(":") && !host.startsWith("[") ? `[${host}]` : host; - const username = typeof record.username === "string" ? record.username : ""; - const password = typeof record.password === "string" ? record.password : ""; - const auth = - username || password ? `${encodeURIComponent(username)}:${encodeURIComponent(password)}@` : ""; - const family = typeof record.family === "string" ? record.family : ""; - const marker = family === "ipv4" || family === "ipv6" ? `?family=${family}` : ""; - return `${type}://${auth}${bracketed}:${port}${marker}`; + if (!isRankableProxyType(type)) return null; + const port = probePortForType(record, type); + const auth = probeAuthPart(record); + const marker = probeFamilyMarker(record); + return `${type}://${auth}${bracketIpv6Host(host)}:${port}${marker}`; +} + +// Expand an IPv6 literal to its eight 16-bit groups, or null when malformed. +// `::` supplies the missing zero groups; anything else must list all eight. +function expandIpv6ToGroups(text: string): number[] | null { + const halves = text.split("::"); + if (halves.length > 2) return null; + const head = halves[0] ? halves[0].split(":") : []; + const tail = halves[1] ? halves[1].split(":") : []; + if (halves.length === 1 && head.length !== 8) return null; + if (halves.length === 2 && head.length + tail.length > 7) return null; + if ([...head, ...tail].some((part) => !/^[0-9a-f]{1,4}$/.test(part))) return null; + const groups = [ + ...head.map((part) => parseInt(part, 16)), + ...new Array(halves.length === 2 ? 8 - head.length - tail.length : 0).fill(0), + ...tail.map((part) => parseInt(part, 16)), + ]; + if (groups.length !== 8 || groups.some((group) => !Number.isInteger(group))) return null; + return groups; +} + +/** + * Normalize an observed egress address for pool ranking: a whole IPv4 address + * (`203.0.113.7`), an IPv4-mapped IPv6 form demapped to IPv4 + * (`::ffff:203.0.113.7`), or an IPv6 /64 prefix (`v6:2001:db8:1:0`) — members + * behind one shared /64 consume one quota. Anything else (zone ids + * `fe80::1%eth0`, unparseable input) gives null so the caller falls back to + * today's order instead of guessing. + */ +export function normalizeEgressAddressForRanking(ip: unknown): string | null { + if (typeof ip !== "string") return null; + const text = ip.trim().toLowerCase(); + if (!text || text.includes("%")) return null; + const mapped = text.startsWith("::ffff:") ? text.slice("::ffff:".length) : text; + if (isIP(mapped) === 4) return mapped; + if (isIP(text) !== 6) return null; + const groups = expandIpv6ToGroups(text); + if (groups === null) return null; + return `v6:${groups + .slice(0, 4) + .map((group) => group.toString(16)) + .join(":")}`; +} + +// Short-lived cache of the last observed egress address per pool entry point +// (same 30 s / 200-entry regime as the pool observation panel): keeps repeated +// picks from re-reading the journal row by row. +const HOT_EGRESS_CACHE_TTL_MS = 30_000; +const HOT_EGRESS_CACHE_MAX_ENTRIES = 200; + +const hotEgressCache = new Map(); + +function readHotEgressCache(key: string, nowMs: number): { hit: boolean; value: string | null } { + const hit = hotEgressCache.get(key); + if (!hit || nowMs - hit.at >= HOT_EGRESS_CACHE_TTL_MS) return { hit: false, value: null }; + return { hit: true, value: hit.value }; +} + +function writeHotEgressCache(key: string, value: string | null, nowMs: number): void { + if (!hotEgressCache.has(key) && hotEgressCache.size >= HOT_EGRESS_CACHE_MAX_ENTRIES) { + const oldest = hotEgressCache.keys().next().value; + if (oldest !== undefined) hotEgressCache.delete(oldest); + } + hotEgressCache.set(key, { at: nowMs, value }); +} + +/** Test-only: forget the cached egress observations. */ +export function __resetHotEgressCacheForTesting(): void { + hotEgressCache.clear(); +} + +// Key a (host, port) couple exactly the way the journal reader does: trimmed, +// unbracketed, lower-cased host plus the validated port. Anything unusable maps +// to null so the reader below falls back to today's behavior. +function hotEgressCacheKey(row: unknown): string | null { + if (!row || typeof row !== "object" || Array.isArray(row)) return null; + const record = row as Record; + const host = typeof record.host === "string" ? record.host.trim() : ""; + if (!host) return null; + const unbracketed = + host.startsWith("[") && host.endsWith("]") && host.length > 2 ? host.slice(1, -1).trim() : host; + if (!unbracketed) return null; + const port = Number(record.port); + if (!Number.isInteger(port) || port < 1 || port > 65535) return null; + return `${unbracketed.toLowerCase()}:${port}`; +} + +// Observation window for the shared-egress ranking signal: an entry that +// showed two or more distinct egress addresses within the window is opaque +// (a gateway or core switching behind one port) and reads as unknown. +const EGRESS_RANKING_WINDOW_MS = 30 * 60_000; + +// Last normalized egress address observed through a pool entry point, or null. +// Best-effort by design: a cold probe cache yields no observation and the caller +// keeps today's order. An entry that showed two or more distinct egress +// addresses within the window is opaque (a gateway or core switching behind +// one port): treated as unknown, never ranked. +function readNormalizedEgressForMember(row: unknown, nowMs: number): string | null { + const record = row as Record; + const host = typeof record.host === "string" ? record.host : ""; + const port = Number(record.port); + if (!host || !Number.isInteger(port) || port < 1 || port > 65535) return null; + const key = hotEgressCacheKey(row); + if (key === null) return null; + const cached = readHotEgressCache(key, nowMs); + if (cached.hit) return cached.value; + let value: string | null = null; + try { + const sinceIso = new Date(nowMs - EGRESS_RANKING_WINDOW_MS).toISOString(); + const observed = getRecentEgressIpsForProxy(host, port, sinceIso, 3); + const distinct = new Set(); + for (const ip of observed) { + const normalized = normalizeEgressAddressForRanking(ip); + if (normalized !== null) distinct.add(normalized); + } + // One value: that value. Zero or two-or-more: null — an entry that + // showed two or more distinct egress addresses within the window is + // opaque (a gateway or core switching behind one port): treated as + // unknown, never ranked. + value = distinct.size === 1 ? [...distinct][0] : null; + } catch { + value = null; + } + writeHotEgressCache(key, value, nowMs); + return value; +} + +/** + * Predicate over pool candidates sharing a recently refused egress address, + * for a provider whose quota is bucketed by egress address. Seeded only from + * members the refusal memory currently sets aside (a refused member's observed + * address is the hot one); an empty hot set, a provider outside the list, or + * any read failure yields a predicate that is false for every candidate, so + * the pool keeps today's order (order only, never exclude). + */ +export function buildHotEgressPredicate( + provider: string | null | undefined, + candidates: unknown[], + signals?: Partial, + nowMs: number = Date.now() +): (candidate: unknown) => boolean { + const none = () => false; + if (!isProxySkipRecentlyFailedEnabled()) return none; + if (!isProxyPoolSharedEgressOrderEnabled()) return none; + if (!isEgressBucketedLockScope(provider)) return none; + if (!hasProxyRefusals()) return none; + const isAvoided = signals?.isAvoided ?? DEFAULT_POOL_RANK_SIGNALS.isAvoided; + const hot = new Set(); + for (const row of candidates) { + if (!isAvoided(proxyEgressKey(row))) continue; + const egress = readNormalizedEgressForMember(row, nowMs); + if (egress !== null) hot.add(egress); + } + if (hot.size === 0) return none; + return (candidate: unknown) => { + const egress = readNormalizedEgressForMember(candidate, nowMs); + return egress !== null && hot.has(egress); + }; } /** * Order pool candidates by crossed short-memory health signals without removing - * anyone: a member just set aside ranks last, then a member whose last cached - * probe verdict was negative. Unknown (no signal, unreconstructible URL) keeps - * the current position order. Stable: health ties keep their relative order, so - * an all-clear or all-set-aside pool returns its input order unchanged. + * anyone: a member just set aside ranks last, then a member sharing a recently + * refused egress address (per-address-quota providers only), then a member + * whose last cached probe verdict was negative. Unknown (no signal, + * unreconstructible URL) keeps the current position order. Stable: health ties + * keep their relative order, so an all-clear or all-set-aside pool returns its + * input order unchanged. */ export function rankPoolCandidates(candidates: T[], signals?: Partial): T[] { if (candidates.length < 2) return [...candidates]; - const { isAvoided, probeHealth } = { + const { isAvoided, probeHealth, sharesHotEgress } = { ...DEFAULT_POOL_RANK_SIGNALS, isAvoided: signals?.isAvoided ?? DEFAULT_POOL_RANK_SIGNALS.isAvoided, probeHealth: signals?.probeHealth ?? DEFAULT_POOL_RANK_SIGNALS.probeHealth, + sharesHotEgress: signals?.sharesHotEgress ?? DEFAULT_POOL_RANK_SIGNALS.sharesHotEgress, }; const scored = candidates.map((candidate, index) => { - if (isAvoided(proxyEgressKey(candidate))) return { candidate, index, score: 2 }; + if (isAvoided(proxyEgressKey(candidate))) return { candidate, index, score: 3 }; + if (sharesHotEgress?.(candidate) === true) return { candidate, index, score: 2 }; const url = candidateProbeUrl(candidate); if (url !== null && probeHealth(url) === false) return { candidate, index, score: 1 }; return { candidate, index, score: 0 }; @@ -240,6 +428,103 @@ export function rankPoolCandidates(candidates: T[], signals?: Partial entry.candidate); } +type RotationPickState = { + strategy: ProxyRotationStrategy; + cursor: number; + stickyWindowMinutes: number; + rotatedAt: string | null; +}; + +// A held sticky member is served untouched (replaced for this pick only when set +// aside): no extra write, no ranking. Returns undefined when not held. +function pickHeldStickyMember(state: RotationPickState, candidates: T[]): T | undefined { + if (state.strategy !== "sticky" || isStickyExpired(state)) return undefined; + const idx = ((state.cursor % candidates.length) + candidates.length) % candidates.length; + const eligible = eligibleMemberIndexes(candidates); + return candidates[eligible ? firstEligibleFrom(idx, eligible, candidates.length) : idx]; +} + +// Order by crossed short-memory health signals (PROXY_SKIP_RECENTLY_FAILED, default on): +// stops re-serving at the head a proxy that just failed, without removing anyone. +// For providers whose quota is bucketed by egress address, a member sharing the +// refused member's observed address ranks just below healthy members (order only, +// never excluded). +// NOTE: ranking changes which member the persisted cursor lands on. After a +// set-aside, the next pick serves the healthiest member at-or-after the cursor +// (not the cursor member itself when it was set aside) — the cursor then +// advances past the member served, preserving rotation without re-serving the +// failed head first. +function rankCandidatesForPick(candidates: T[], provider?: string | null): T[] { + if (!isProxySkipRecentlyFailedEnabled()) return [...candidates]; + if (provider == null) return rankPoolCandidates(candidates); + return rankPoolCandidates(candidates, { + sharesHotEgress: buildHotEgressPredicate(provider, candidates), + }); +} + +function pickRandomCandidate(ranked: T[], eligible: number[] | null): T { + // crypto.randomInt (unbiased, uniform in [0, length)) instead of Math.random — + // CodeQL js/insecure-randomness flags Math.random flowing into the selected proxy's + // credentials (a "security context"). Load-balancing selection is not a secret, but + // crypto.randomInt silences the alert at the source and is unbiased (#6365 follow-up). + if (eligible) return ranked[eligible[randomInt(eligible.length)]]; + return ranked[randomInt(ranked.length)]; +} + +function pickLatencyCandidate( + db: ReturnType, + ranked: T[], + eligible: number[] | null +): T { + return pickByLatency(db, eligible ? eligible.map((index) => ranked[index]) : ranked); +} + +function pickStickyCandidate( + db: ReturnType, + state: RotationPickState, + normalizedScope: string, + rotationScopeId: string, + ranked: T[], + eligible: number[] | null +): T { + const expired = isStickyExpired(state); + let cursor = state.cursor; + if (expired) { + cursor = state.cursor + 1; + db.prepare( + "UPDATE proxy_scope_rotation SET cursor = ?, rotated_at = ?, updated_at = ? WHERE scope = ? AND scope_id IS ?" + ).run( + cursor, + new Date().toISOString(), + new Date().toISOString(), + normalizedScope, + rotationScopeId + ); + } + const idx = ((cursor % ranked.length) + ranked.length) % ranked.length; + // A held member set aside is replaced for this pick only: no extra write. + return ranked[eligible ? firstEligibleFrom(idx, eligible, ranked.length) : idx]; +} + +// round-robin (default): pick at the current cursor, then advance it monotonically, +// past any member skipped so the next pick starts after the one actually served. +function pickRoundRobinCandidate( + db: ReturnType, + state: RotationPickState, + normalizedScope: string, + rotationScopeId: string, + ranked: T[], + eligible: number[] | null +): T { + const idx = ((state.cursor % ranked.length) + ranked.length) % ranked.length; + const served = eligible ? firstEligibleFrom(idx, eligible, ranked.length) : idx; + const skipped = (served - idx + ranked.length) % ranked.length; + db.prepare( + "UPDATE proxy_scope_rotation SET cursor = ?, updated_at = ? WHERE scope = ? AND scope_id IS ?" + ).run(state.cursor + skipped + 1, new Date().toISOString(), normalizedScope, rotationScopeId); + return ranked[served]; +} + /** * Pick one member from an already-alive candidate list according to the scope's * rotation strategy. Assumes `candidates` is non-empty and ordered by position. @@ -252,7 +537,8 @@ function pickFromCandidates( db: ReturnType, normalizedScope: string, rotationScopeId: string, - candidates: T[] + candidates: T[], + provider?: string | null ): T { // Pool-exhausted check first: a single-member pool set aside is exhausted // too, and this runs before the length-1 early return below. Flag-gated @@ -267,73 +553,22 @@ function pickFromCandidates( const state = getOrCreateRotationRow(db, normalizedScope, rotationScopeId); - if (state.strategy === "sticky") { - const expired = isStickyExpired(state); - if (!expired) { - const idx = ((state.cursor % candidates.length) + candidates.length) % candidates.length; - const eligible = eligibleMemberIndexes(candidates); - return candidates[eligible ? firstEligibleFrom(idx, eligible, candidates.length) : idx]; - } - } + const held = pickHeldStickyMember(state, candidates); + if (held !== undefined) return held; - // Order by crossed short-memory health signals (PROXY_SKIP_RECENTLY_FAILED, default on): - // stops re-serving at the head a proxy that just failed, without removing anyone. // Sticky past its window and every other strategy rank normally; a held sticky // member returns above, untouched. The eligible-skip below still applies on the // ranked list, so a set-aside member stays skipped while another is eligible and // the cursor advances past the member actually served. - // NOTE: ranking changes which member the persisted cursor lands on. After a - // set-aside, the next pick serves the healthiest member at-or-after the cursor - // (not the cursor member itself when it was set aside) — the cursor then - // advances past the member served, preserving rotation without re-serving the - // failed head first. - const ranked = isProxySkipRecentlyFailedEnabled() - ? rankPoolCandidates(candidates) - : [...candidates]; + const ranked = rankCandidatesForPick(candidates, provider); const eligible = eligibleMemberIndexes(ranked); - if (state.strategy === "random") { - // crypto.randomInt (unbiased, uniform in [0, length)) instead of Math.random — - // CodeQL js/insecure-randomness flags Math.random flowing into the selected proxy's - // credentials (a "security context"). Load-balancing selection is not a secret, but - // crypto.randomInt silences the alert at the source and is unbiased (#6365 follow-up). - if (eligible) return ranked[eligible[randomInt(eligible.length)]]; - return ranked[randomInt(ranked.length)]; - } - - if (state.strategy === "latency") { - return pickByLatency(db, eligible ? eligible.map((index) => ranked[index]) : ranked); - } - + if (state.strategy === "random") return pickRandomCandidate(ranked, eligible); + if (state.strategy === "latency") return pickLatencyCandidate(db, ranked, eligible); if (state.strategy === "sticky") { - const expired = isStickyExpired(state); - let cursor = state.cursor; - if (expired) { - cursor = state.cursor + 1; - db.prepare( - "UPDATE proxy_scope_rotation SET cursor = ?, rotated_at = ?, updated_at = ? WHERE scope = ? AND scope_id IS ?" - ).run( - cursor, - new Date().toISOString(), - new Date().toISOString(), - normalizedScope, - rotationScopeId - ); - } - const idx = ((cursor % ranked.length) + ranked.length) % ranked.length; - // A held member set aside is replaced for this pick only: no extra write. - return ranked[eligible ? firstEligibleFrom(idx, eligible, ranked.length) : idx]; + return pickStickyCandidate(db, state, normalizedScope, rotationScopeId, ranked, eligible); } - - // round-robin (default): pick at the current cursor, then advance it monotonically, - // past any member skipped so the next pick starts after the one actually served. - const idx = ((state.cursor % ranked.length) + ranked.length) % ranked.length; - const served = eligible ? firstEligibleFrom(idx, eligible, ranked.length) : idx; - const skipped = (served - idx + ranked.length) % ranked.length; - db.prepare( - "UPDATE proxy_scope_rotation SET cursor = ?, updated_at = ? WHERE scope = ? AND scope_id IS ?" - ).run(state.cursor + skipped + 1, new Date().toISOString(), normalizedScope, rotationScopeId); - return ranked[served]; + return pickRoundRobinCandidate(db, state, normalizedScope, rotationScopeId, ranked, eligible); } // Fetch the alive, position-ordered candidate rows for a (scope, scope_id) pool. @@ -394,10 +629,44 @@ function resolveScopePoolInternal( options.matchAnyScopeId === true ); if (rows.length === 0) return null; - const picked = pickFromCandidates(db, scope, options.rotationScopeId, rows); + const picked = pickFromCandidates( + db, + scope, + options.rotationScopeId, + rows, + resolvePoolProvider(db, scope, levelId, options) + ); return toRegistryProxyResolution(picked, scope, levelId); } +// Which provider a pool pick serves, for the egress-aware ranking signal only. +// Provider scope already knows it; an account pool resolves its connection's +// provider with one bounded row (only reached when the signal guards already +// passed); combo and global pools serve several providers, so they abstain. +function resolvePoolProvider( + db: ReturnType, + scope: ProxyScope, + levelId: string | null, + options: { rotationScopeId: string; matchAnyScopeId?: boolean; scopeIdFilter?: string | null } +): string | null { + if (scope === "provider") return levelId; + if (scope !== "account" || !options.scopeIdFilter) return null; + if ( + !isProxySkipRecentlyFailedEnabled() || + !isProxyPoolSharedEgressOrderEnabled() || + !hasProxyRefusals() + ) + return null; + try { + const row = db + .prepare("SELECT provider FROM provider_connections WHERE id = ?") + .get(options.scopeIdFilter) as { provider?: string } | undefined; + return typeof row?.provider === "string" && row.provider ? row.provider : null; + } catch { + return null; + } +} + export async function resolveProxyForConnectionFromRegistry(connectionId: string) { try { const db = getDbInstance(); diff --git a/src/lib/db/proxyLogs.ts b/src/lib/db/proxyLogs.ts index f94778a558..5144eb1b57 100644 --- a/src/lib/db/proxyLogs.ts +++ b/src/lib/db/proxyLogs.ts @@ -161,6 +161,34 @@ export function getRecentEgressIpForProxy( return { egressIp: row.egress_ip, at: row.timestamp }; } +/** + * Distinct non-null egress IPs observed through a proxy endpoint since + * `sinceIso` (up to `limit`). Same host-key normalization and port validation + * as `getRecentEgressIpForProxy`: anything unusable yields `[]`, never a throw. + */ +export function getRecentEgressIpsForProxy( + host: string, + port: number, + sinceIso: string, + limit = 3 +): string[] { + const h = normalizeProxyHostKey(host); + if (!h) return []; + if (!Number.isInteger(port) || port < 1 || port > 65535) return []; + if (typeof sinceIso !== "string" || !sinceIso) return []; + const capped = Number.isInteger(limit) && limit > 0 ? Math.min(limit, 10) : 3; + const db = getDbInstance(); + const rows = db + .prepare( + `SELECT DISTINCT egress_ip FROM proxy_logs + WHERE proxy_host = ? AND proxy_port = ? + AND egress_ip IS NOT NULL AND timestamp >= ? + LIMIT ?` + ) + .all(h, port, sinceIso, capped) as Array<{ egress_ip: string }>; + return rows.map((r) => r.egress_ip).filter((ip) => typeof ip === "string" && ip); +} + export type PoolEgressObservationCounts = { connections: number; distinctExits: number; diff --git a/src/shared/constants/featureFlagDefinitions.ts b/src/shared/constants/featureFlagDefinitions.ts index f1941e7558..1bd198f0a3 100644 --- a/src/shared/constants/featureFlagDefinitions.ts +++ b/src/shared/constants/featureFlagDefinitions.ts @@ -215,6 +215,18 @@ export const FEATURE_FLAG_DEFINITIONS: FeatureFlagDefinition[] = [ requiresRestart: false, warningLevel: "caution", }, + { + key: "PROXY_POOL_SHARED_EGRESS_ORDER", + label: "Shared Egress Pool Order", + description: + "For providers whose quota is bucketed by egress address, rank a pool member sharing a recently refused member's observed egress address just below healthy members. Order only, never excluded. Needs PROXY_SKIP_RECENTLY_FAILED, which produces the refusal signal it reads. Off by default: selection order is exactly the plain rotation.", + descriptionI18nKey: "featureFlagProxyPoolSharedEgressOrderDescription", + category: "network", + defaultValue: "false", + type: "boolean", + requiresRestart: false, + warningLevel: "caution", + }, { key: "ROTATION_ATTRIBUTION", label: "Rotation Attribution Logging", diff --git a/src/shared/utils/featureFlags.ts b/src/shared/utils/featureFlags.ts index 6a2f21794f..1310d52c6c 100644 --- a/src/shared/utils/featureFlags.ts +++ b/src/shared/utils/featureFlags.ts @@ -211,6 +211,23 @@ export function isProxySkipRecentlyFailedEnabled(): boolean { } } +/** + * Shared-egress pool ordering (opt-in, default off). Needs + * PROXY_SKIP_RECENTLY_FAILED, which produces the refusal signal it reads. + * Fail-closed: an unreadable flag store keeps the plain selection. + */ +export function isProxyPoolSharedEgressOrderEnabled(): boolean { + try { + return isFeatureFlagEnabled("PROXY_POOL_SHARED_EGRESS_ORDER"); + } catch (error) { + console.error( + "[featureFlags] Failed to resolve PROXY_POOL_SHARED_EGRESS_ORDER, defaulting to disabled:", + error instanceof Error ? error.message : error + ); + return false; + } +} + /** * Rotation attribution (skipped-account log lines, per-account rotation state, * masked serving-account id and request correlation on proxy log entries). diff --git a/tests/unit/feature-flags-settings.test.ts b/tests/unit/feature-flags-settings.test.ts index e69894954c..10d6d59ecc 100644 --- a/tests/unit/feature-flags-settings.test.ts +++ b/tests/unit/feature-flags-settings.test.ts @@ -48,7 +48,9 @@ const { // takes it to 78. // OPENCODE_POOL_RESELECT (re-select a pool member per attempt after a // per-address 429, default off) takes the registry to 79. -const EXPECTED_FEATURE_FLAG_COUNT = 79; +// PROXY_POOL_SHARED_EGRESS_ORDER (shared-egress pool ordering, default off) +// takes it to 80. +const EXPECTED_FEATURE_FLAG_COUNT = 80; // ────────────────────────────────────────────────────── // Test group 1 — Flag definitions registry diff --git a/tests/unit/proxy-pool-shared-egress-order.test.ts b/tests/unit/proxy-pool-shared-egress-order.test.ts new file mode 100644 index 0000000000..a589ea3318 --- /dev/null +++ b/tests/unit/proxy-pool-shared-egress-order.test.ts @@ -0,0 +1,307 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +// Shared-egress pool ordering (opt-in via PROXY_POOL_SHARED_EGRESS_ORDER, +// default off; also needs PROXY_SKIP_RECENTLY_FAILED for the refusal signal): +// a pool member whose recently observed egress address matches a refused +// member's address ranks just below healthy members for providers whose quota +// is bucketed by egress address — without ever excluding anyone. A member with +// two distinct observed addresses in the 30 min window is opaque and never +// ranked. With either flag off (the default) or a provider outside that list +// the order stays exactly the plain rotation. + +const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-shared-egress-")); +process.env.DATA_DIR = TEST_DATA_DIR; +process.env.API_KEY_SECRET = "test-secret"; + +const core = await import("../../src/lib/db/core.ts"); +const proxiesDb = await import("../../src/lib/db/proxies.ts"); +const rotation = await import("../../src/lib/db/proxies/rotation.ts"); +const memory = await import("../../open-sse/utils/proxyRefusalMemory.ts"); +const proxyLogger = await import("../../src/lib/proxyLogger.ts"); +const { noteProxyOutcome } = await import("../../src/sse/handlers/proxyOutcomeMemory.ts"); + +function resetStorage() { + memory.__resetProxyRefusalMemoryForTesting(); + process.env.PROXY_SKIP_RECENTLY_FAILED = "true"; + process.env.PROXY_POOL_SHARED_EGRESS_ORDER = "true"; + proxyLogger.clearProxyLogs(); + core.resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 }); + fs.mkdirSync(TEST_DATA_DIR, { recursive: true }); +} + +test.beforeEach(() => { + resetStorage(); +}); + +test.after(() => { + delete process.env.PROXY_SKIP_RECENTLY_FAILED; + delete process.env.PROXY_POOL_SHARED_EGRESS_ORDER; + memory.__resetProxyRefusalMemoryForTesting(); + if (typeof rotation.__resetHotEgressCacheForTesting === "function") { + rotation.__resetHotEgressCacheForTesting(); + } + core.resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 }); +}); + +type Member = { type: string; host: string; port: number }; +let seq = 0; + +function member(): Member { + seq++; + return { type: "http", host: `10.9.0.${seq}`, port: 9200 + seq }; +} + +function keyOf(m: Member) { + return memory.proxyEgressKey(m); +} + +function refuse(m: Member) { + memory.noteProxyRefusal(keyOf(m), "ip_quota_429"); +} + +function logEgress(m: Member, egressIp: string | null) { + proxyLogger.logProxyEvent({ + status: "success", + provider: "opencode", + targetUrl: "https://api.opencode.ai/chat", + proxy: { type: "http", host: m.host, port: m.port }, + egressIp, + }); +} + +function logEgressOld(m: Member, egressIp: string | null, ageMs: number) { + // Backdate by writing directly: the logger stamps Date.now(), so shift the + // row afterwards. Bounded helper for the window test only. + logEgress(m, egressIp); + flush(); + const db = core.getDbInstance(); + const old = new Date(Date.now() - ageMs).toISOString(); + db.prepare( + `UPDATE proxy_logs SET timestamp = ? WHERE proxy_host = ? AND proxy_port = ? AND egress_ip = ?` + ).run(old, m.host, m.port, egressIp); +} + +function flush() { + proxyLogger.flushProxyLogsSync(); +} + +function hotFor(members: Member[], provider: string | null = "opencode") { + return rotation.buildHotEgressPredicate(provider, members); +} + +async function dbPool(size: number): Promise { + const members: Member[] = []; + for (let i = 0; i < size; i++) { + const m = member(); + const proxy = await proxiesDb.createProxy({ name: `member ${seq}`, ...m }); + await proxiesDb.addProxyToScopePool("provider", "opencode", proxy.id); + members.push(m); + } + return members; +} + +async function pick() { + const resolved = await proxiesDb.resolveProxyForScopeFromRegistry("provider", "opencode"); + return (resolved as { proxy: { host: string } }).proxy.host; +} + +test("a member sharing the refused egress address is demoted below healthy members", () => { + const a = member(); + const b = member(); + const c = member(); + refuse(a); + logEgress(a, "203.0.113.7"); + logEgress(b, "203.0.113.7"); + logEgress(c, "203.0.113.44"); + flush(); + const ranked = rotation.rankPoolCandidates([a, b, c], { sharesHotEgress: hotFor([a, b, c]) }); + assert.deepEqual( + ranked.map((m) => m.host), + [c.host, b.host, a.host] + ); +}); + +test("same address on two members: the healthy one serves first, nobody is removed", () => { + const a = member(); + const b = member(); + refuse(a); + logEgress(a, "203.0.113.7"); + logEgress(b, "203.0.113.7"); + flush(); + const ranked = rotation.rankPoolCandidates([a, b], { sharesHotEgress: hotFor([a, b]) }); + assert.equal(ranked.length, 2); + assert.deepEqual( + ranked.map((m) => m.host), + [b.host, a.host] + ); +}); + +test("a provider outside the per-address-quota list is untouched", () => { + const a = member(); + const b = member(); + refuse(a); + logEgress(a, "203.0.113.7"); + logEgress(b, "203.0.113.7"); + flush(); + const pred = rotation.buildHotEgressPredicate("openai", [a, b]); + assert.equal(pred(a), false); + assert.equal(pred(b), false); + const ranked = rotation.rankPoolCandidates([a, b], { sharesHotEgress: pred }); + assert.deepEqual( + ranked.map((m) => m.host), + [b.host, a.host] + ); +}); + +test("a missing observation keeps the current order", () => { + const a = member(); + const b = member(); + const c = member(); + refuse(a); + const pred = hotFor([a, b, c]); + assert.equal(pred(b), false); + const ranked = rotation.rankPoolCandidates([a, b, c], { sharesHotEgress: pred }); + assert.deepEqual( + ranked.map((m) => m.host), + [b.host, c.host, a.host] + ); +}); + +test("a fully shared pool keeps its order and size", () => { + const a = member(); + const b = member(); + refuse(a); + refuse(b); + logEgress(a, "203.0.113.7"); + logEgress(b, "203.0.113.7"); + flush(); + const ranked = rotation.rankPoolCandidates([a, b], { sharesHotEgress: hotFor([a, b]) }); + assert.equal(ranked.length, 2); + assert.deepEqual( + ranked.map((m) => m.host), + [a.host, b.host] + ); +}); + +test("an IPv6 member sharing the refused /64 is demoted, another /64 is not", () => { + const a = member(); + const b = member(); + const c = member(); + refuse(a); + logEgress(a, "2001:db8:1::1"); + logEgress(b, "2001:db8:1::2"); + logEgress(c, "2001:db8:2::1"); + flush(); + const ranked = rotation.rankPoolCandidates([a, b, c], { sharesHotEgress: hotFor([a, b, c]) }); + assert.deepEqual( + ranked.map((m) => m.host), + [c.host, b.host, a.host] + ); +}); + +test("normalizeEgressAddressForRanking: v4 whole, v6 /64, mapped, invalid", () => { + const normalize = rotation.normalizeEgressAddressForRanking; + assert.equal(normalize("203.0.113.7"), "203.0.113.7"); + assert.equal(normalize("2001:DB8:1::1"), "v6:2001:db8:1:0"); + assert.equal(normalize("2001:db8:1:0:0:0:0:2"), "v6:2001:db8:1:0"); + assert.equal(normalize("::ffff:203.0.113.7"), "203.0.113.7"); + assert.equal(normalize("fe80::1%eth0"), null); + assert.equal(normalize("not-an-ip"), null); + assert.equal(normalize(null), null); + assert.equal(normalize(""), null); +}); + +test("functional: the second request avoids the refused egress address", async () => { + const [a, b] = await dbPool(2); + logEgress(a, "203.0.113.7"); + logEgress(b, "203.0.113.7"); + flush(); + assert.equal(await pick(), a.host); + noteProxyOutcome("opencode", { + proxy: { type: "http", host: a.host, port: a.port }, + upstreamStatus: 429, + }); + assert.equal(await pick(), b.host); +}); + +test("functional: with the flag off the pool serves the head even when egress is shared", async () => { + const [a, b] = await dbPool(2); + refuse(a); + logEgress(a, "203.0.113.7"); + logEgress(b, "203.0.113.7"); + flush(); + process.env.PROXY_SKIP_RECENTLY_FAILED = "false"; + delete process.env.PROXY_POOL_SHARED_EGRESS_ORDER; + try { + assert.equal(await pick(), a.host); + } finally { + process.env.PROXY_SKIP_RECENTLY_FAILED = "true"; + process.env.PROXY_POOL_SHARED_EGRESS_ORDER = "true"; + } +}); + +test("a member with two distinct egress addresses in the window is opaque: not demoted", () => { + const a = member(); + const b = member(); + refuse(a); + logEgress(a, "203.0.113.7"); + logEgress(b, "203.0.113.7"); + logEgress(b, "203.0.113.99"); + flush(); + const pred = hotFor([a, b]); + // b is opaque (two addresses): not demoted. a is refused (avoided), so it + // still ranks last by the refusal signal; b serves first. + assert.equal(pred(b), false); + const ranked = rotation.rankPoolCandidates([a, b], { sharesHotEgress: pred }); + assert.deepEqual( + ranked.map((m) => m.host), + [b.host, a.host] + ); +}); + +test("an observation older than 30 minutes is ignored: not demoted", () => { + const a = member(); + const b = member(); + refuse(a); + logEgressOld(a, "203.0.113.7", 31 * 60_000); + logEgress(b, "203.0.113.7"); + flush(); + const pred = hotFor([a, b]); + // a's observation is outside the window: no hot address seeded from it, + // so b is not demoted. a still ranks last by the refusal signal itself. + assert.equal(pred(b), false); + const ranked = rotation.rankPoolCandidates([a, b], { sharesHotEgress: pred }); + assert.deepEqual( + ranked.map((m) => m.host), + [b.host, a.host] + ); +}); + +test("skip on plus new flag off never demotes for shared egress", () => { + const a = member(); + const b = member(); + refuse(a); + logEgress(a, "203.0.113.7"); + logEgress(b, "203.0.113.7"); + flush(); + delete process.env.PROXY_POOL_SHARED_EGRESS_ORDER; + try { + // The new flag is off: no shared-egress demotion. a still ranks last by + // the skip-recently-failed refusal signal itself; b serves first. + const pred = hotFor([a, b]); + assert.equal(pred(b), false); + const ranked = rotation.rankPoolCandidates([a, b], { sharesHotEgress: pred }); + assert.deepEqual( + ranked.map((m) => m.host), + [b.host, a.host] + ); + } finally { + process.env.PROXY_POOL_SHARED_EGRESS_ORDER = "true"; + } +}); diff --git a/tests/unit/server-owned-tool-loop-flag.test.ts b/tests/unit/server-owned-tool-loop-flag.test.ts index 0be03f789c..d5ed2ce02b 100644 --- a/tests/unit/server-owned-tool-loop-flag.test.ts +++ b/tests/unit/server-owned-tool-loop-flag.test.ts @@ -71,8 +71,9 @@ describe("feature-flags-settings count update", () => { // 893fef9c added OPENCODE_PARK_AND_RESUME (74 -> 75); FLUSH_EMPTY_RETRY_ENABLED // (flush empty-turn retry, default off) bumps it to 76; ROTATION_ATTRIBUTION (#14223, // default off) to 77; STREAM_READINESS_STALL_RETRY (#14669, default off) to 78; - // OPENCODE_POOL_RESELECT (default off) to 79. - assert.equal(FEATURE_FLAG_DEFINITIONS.length, 79); + // OPENCODE_POOL_RESELECT (default off) to 79; PROXY_POOL_SHARED_EGRESS_ORDER + // (#14657, default off) to 80. + assert.equal(FEATURE_FLAG_DEFINITIONS.length, 80); }); });