diff --git a/CLAUDE.md b/CLAUDE.md index 979ae11..1e51037 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -7,49 +7,8 @@ Key architecture facts: - Middleware stack: rate limiter, body limiter, structured logger, SSRF protection - WebSocket relay: bidirectional relay via `x-relay-target` header with `ws://` or `wss://` - Error classification: DNS errors -> 502, timeouts -> 504, SSRF blocks -> 403, rate limits -> 429 -- IPv6 support: dual-stack listen + outbound source rotation via `Bun.spawn` + `curl --interface` - The old Next.js `src/app/route.ts` still exists as a legacy file but is no longer the active entry point -## IPv6 Configuration - -### Environment Variables - -| Variable | Default | Description | -|----------|---------|-------------| -| `HOST` | `::` | Bind address. Use `::` for dual-stack (IPv4+IPv6) | -| `IPV6_SOURCES` | _(empty)_ | Comma-separated IPv6 source addresses for outbound rotation | - -### Setup - -1. Add IPv6 addresses to your interface: -```bash -ip -6 addr add 2001:df4:c140:1f::d6/128 dev eth0 -ip -6 addr add 2001:df4:c140:1f:ffff:ffff:ffff:ffff/128 dev eth0 -``` - -2. Configure the proxy with IPv6 source rotation: -```bash -IPV6_SOURCES=2001:df4:c140:1f::d6,2001:df4:c140:1f:ffff:ffff:ffff:ffff bun run src/index.ts -``` - -### How It Works - -- **Listen**: Server binds to `::` (all IPv6 interfaces) with `ipv6Only: false` (dual-stack) -- **Outbound**: When `IPV6_SOURCES` is configured, each outbound request rotates through the source addresses using `curl --interface ` -- **Failover**: Failed source addresses are automatically disabled after 3 consecutive failures - -### Source: `src/lib/ipv6-pool.ts` - -```typescript -import { IPv6SourcePool } from "./lib/ipv6-pool"; - -const pool = new IPv6SourcePool(); -pool.loadFromEnv(); // reads IPV6_SOURCES - -const source = pool.getNext(); // round-robin -pool.markSuccess(source); // reset failure count -pool.markFailed(source); // increment failure count -``` Default to using Bun instead of Node.js. diff --git a/src/index.ts b/src/index.ts index 3435993..8621a54 100644 --- a/src/index.ts +++ b/src/index.ts @@ -5,7 +5,6 @@ * Adds: * - Bun.serve() bindings * - WebSocket relay support - * - IPv6 source rotation * - Proxy pool file loading * - Graceful shutdown */ @@ -25,7 +24,6 @@ import { } from "./lib/router"; import type { RouterEnv } from "./lib/router"; -import { IPv6SourcePool } from "./lib/ipv6-pool"; import { closeAllActiveReaders, isDevMode } from "./lib/fetch-utils"; import type { Server, ServerWebSocket } from "bun"; @@ -34,7 +32,7 @@ import type { Server, ServerWebSocket } from "bun"; const RELAY_VERSION = "1.0.0"; const PORT = Number.parseInt(process.env.PORT ?? "3000", 10); -const HOST = process.env.HOST ?? "::"; +const HOST = process.env.HOST ?? "0.0.0.0"; // --- Proxy pool (optional) ---------------------------------------------------- @@ -43,14 +41,6 @@ proxyPool.tryLoad( process.env.PROXY_FILE || process.env.PROXY_LIST || "./proxy.txt", ); -// --- IPv6 source pool (optional) ---------------------------------------------- - -const ipv6Pool = new IPv6SourcePool(); -ipv6Pool.loadFromEnv(); -if (ipv6Pool.configured) { - console.log(`[relay] IPv6 source pool loaded: ${ipv6Pool.size} addresses`); -} - // --- SSRF DNS rebinding protection -------------------------------------------- if (process.env.SSRF_DNS_CHECK === "true") { @@ -116,13 +106,11 @@ function handleWebSocketUpgrade( const server: Server = Bun.serve({ hostname: HOST, - ipv6Only: false, port: PORT, development: isDevMode() ? { hmr: true, console: true } : undefined, async fetch(req: Request): Promise { const url = new URL(req.url); - const ipv6Source = ipv6Pool.getNext() ?? undefined; // Static routes if (url.pathname === "/health") return handleHealth(); @@ -157,10 +145,7 @@ const server: Server = Bun.serve({ req, routerEnv, clientIP, - { - isWebSocketSupported: true, - ipv6Source, - }, + { isWebSocketSupported: true }, ); if (result !== undefined) return result; diff --git a/src/lib/ai-proxy.ts b/src/lib/ai-proxy.ts index 17a5ab8..e20f857 100644 --- a/src/lib/ai-proxy.ts +++ b/src/lib/ai-proxy.ts @@ -407,14 +407,6 @@ export async function handleChatCompletion( proxyPool?: ProxyPool, sessionPool?: SessionProxyPool, sessionId?: string, - ipv6Source?: string, -): Promise; -export async function handleChatCompletion( - body: unknown, - proxyPool?: ProxyPool, - sessionPool?: SessionProxyPool, - sessionId?: string, - ipv6Source?: string, ): Promise { // -- Input validation ------------------------------------------------------- const validationError = validateChatRequest(body); @@ -457,8 +449,8 @@ export async function handleChatCompletion( // -- Execute with session-aware or standard retry -------------------------- let result: FetchWithRetryResult = sessionPool && sessionId - ? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `openai:${req.model}`, undefined, ipv6Source) - : await fetchWithRetry(url, init, proxyPool, `openai:${req.model}`, ipv6Source); + ? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `openai:${req.model}`, undefined) + : await fetchWithRetry(url, init, proxyPool, `openai:${req.model}`); // -- Mimo Free: auth failure → invalidate JWT and retry once --------------- if ( @@ -474,8 +466,8 @@ export async function handleChatCompletion( }; result = sessionPool && sessionId - ? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `openai:${req.model}`, undefined, ipv6Source) - : await fetchWithRetry(url, init, proxyPool, `openai:${req.model}`, ipv6Source); + ? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `openai:${req.model}`, undefined) + : await fetchWithRetry(url, init, proxyPool, `openai:${req.model}`); } if (result.errorClassification) { diff --git a/src/lib/anthropic-proxy.ts b/src/lib/anthropic-proxy.ts index a830652..95cdb13 100644 --- a/src/lib/anthropic-proxy.ts +++ b/src/lib/anthropic-proxy.ts @@ -1034,22 +1034,6 @@ export async function handleAnthropicMessages( proxyPool?: ProxyPool, sessionPool?: SessionProxyPool, sessionId?: string, - ipv6Source?: string, -): Promise; -export async function handleAnthropicMessages( - body: unknown, - proxyPool?: ProxyPool, - sessionPool?: SessionProxyPool, - sessionId?: string, - ipv6Source?: string, - anthropicVersion?: string, -): Promise; -export async function handleAnthropicMessages( - body: unknown, - proxyPool?: ProxyPool, - sessionPool?: SessionProxyPool, - sessionId?: string, - ipv6Source?: string, anthropicVersion?: string, ): Promise { // -- Input validation ------------------------------------------------------- @@ -1091,8 +1075,8 @@ export async function handleAnthropicMessages( // -- Execute with session-aware or standard retry -------------------------- const result: FetchWithRetryResult = sessionPool && sessionId - ? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `anthropic:${req.model}`, undefined, ipv6Source) - : await fetchWithRetry(url, init, proxyPool, `anthropic:${req.model}`, ipv6Source); + ? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `anthropic:${req.model}`, undefined) + : await fetchWithRetry(url, init, proxyPool, `anthropic:${req.model}`); if (result.errorClassification) { if (sessionPool && sessionId) sessionPool.release(sessionId); diff --git a/src/lib/fetch-utils.ts b/src/lib/fetch-utils.ts index 2bb2ea2..7aeb666 100644 --- a/src/lib/fetch-utils.ts +++ b/src/lib/fetch-utils.ts @@ -6,13 +6,6 @@ */ import type { ProxyPool, SessionProxyPool } from "./proxy-pool"; -import type { IPv6SourcePool } from "./ipv6-pool"; -import { unlink } from "node:fs/promises"; - -// ─── Constants ───────────────────────────────────────────────────────── - -/** Default relay timeout (30 seconds). Used for curl IPv6 source requests. */ -const DEFAULT_TIMEOUT_MS = 30_000; // ─── Active stream tracking (for graceful shutdown) ─────────────────────── @@ -129,7 +122,7 @@ export class SSELineBuffer { if (this.overflow) return []; this.buffer += chunk; - + if (this.buffer.length > this.MAX_BUFFER_SIZE) { console.warn(`[SSELineBuffer] Buffer exceeded ${this.MAX_BUFFER_SIZE} bytes — discarding remaining stream data`); this.overflow = true; @@ -219,142 +212,11 @@ export function sanitizeErrorMessage(raw: string): string { return "Upstream error"; } -// ─── Fetch via curl (for IPv6 source binding) ──────────────────────────── - -/** - * Execute an HTTP request via curl with a specific source IPv6 address. - * Uses `Bun.spawn` to run curl with `--interface` to bind to the given address. - * - * This is used when outbound IPv6 source rotation is needed, since - * Bun's built-in fetch() does not support specifying a local address. - * - * Status code is extracted from a temp file written by curl's `-w` flag - * (written after body completes). The body is collected into a single - * buffer and returned as a ReadableStream for zero-copy handoff. - * - * @param url - Target URL - * @param init - Request init (method, headers, body) - * @param ipv6Source - IPv6 source address to bind to - * @param timeoutMs - Request timeout in milliseconds - */ -export async function fetchViaCurl( - url: string, - init: RequestInit, - ipv6Source: string, - timeoutMs: number, -): Promise { - const method = (init.method ?? "GET").toUpperCase(); - - const statusFile = `/tmp/curl-status-${crypto.randomUUID().slice(0, 8)}`; - const args = [ - "curl", - "-6", - "--interface", ipv6Source, - "-X", method, - "-s", // silent mode - "--compressed", // auto-decompress gzip/brotli - "-o", "-", // output body to stdout - "-w", statusFile, // write status code to file (plain text, appended after body) - "--max-time", String(Math.ceil(timeoutMs / 1000)), - "--connect-timeout", "10", - ]; - - // Add headers - if (init.headers) { - const headers = init.headers instanceof Headers - ? Object.fromEntries(init.headers.entries()) - : init.headers; - for (const [key, value] of Object.entries(headers)) { - if (key.toLowerCase() === "host") continue; - args.push("-H", `${key}: ${value}`); - } - } - - // Add body for non-GET/HEAD methods - if (init.body && method !== "GET" && method !== "HEAD") { - if (typeof init.body === "string") { - args.push("-d", init.body); - } else if (init.body instanceof ArrayBuffer) { - args.push("--data-binary", "@-"); - } - } - - args.push(url); - - logProxy("fetchViaCurl", `${args.slice(0, 6).join(" ")}...`, { ipv6Source, url }); - - const proc = Bun.spawn(args, { - stdout: "pipe", - stderr: "pipe", - stdin: "pipe", - }); - - // Kill process after timeout + buffer - const killTimer = setTimeout(() => { - try { proc.kill("SIGKILL"); } catch { /* already dead */ } - }, timeoutMs + 5000); - - // Collect stdout into minimal chunks then concatenate into a single buffer. - // This is needed because curl's -w (status code) is written AFTER the body - // completes, so we must wait for proc.exited before reading the status file. - // We minimize peak memory by streaming chunks into a pre-allocated buffer. - const stdoutChunks: Uint8Array[] = []; - const stdoutReader = proc.stdout.getReader(); - try { - while (true) { - const { done, value } = await stdoutReader.read(); - if (done) break; - stdoutChunks.push(value); - } - } catch { /* stream cancelled */ } - - // Wait for process to exit, then read status code - await proc.exited; - clearTimeout(killTimer); - - let statusCode = 502; - try { - const statusText = await Bun.file(statusFile).text(); - statusCode = parseInt(statusText.trim(), 10) || 502; - } catch { - // Status file not written — connection likely failed - } - - // Clean up temp file (async, non-blocking) - unlink(statusFile).catch(() => {}); - - logProxy("fetchViaCurl", `response status=${statusCode}`, { ipv6Source, url }); - - // Concatenate chunks into a single buffer for the Response body stream. - // Uses a single allocation to reduce GC pressure from many small chunks. - const totalLength = stdoutChunks.reduce((sum, c) => sum + c.byteLength, 0); - const combined = new Uint8Array(totalLength); - let offset = 0; - for (const chunk of stdoutChunks) { - combined.set(chunk, offset); - offset += chunk.byteLength; - } - - // Zero-copy: hand the buffer directly to ReadableStream - const bodyStream = new ReadableStream({ - start(controller) { - controller.enqueue(combined); - controller.close(); - }, - }); - - return new Response(bodyStream, { - status: statusCode, - statusText: statusCode === 200 ? "OK" : "Error", - headers: { "Content-Type": "text/plain" }, - }); -} - // ─── Fetch with retry (direct → proxy fallback) ───────────────────────── export interface FetchWithRetryResult { - response?: Response; - errorClassification?: { code: string; status: number; message: string }; + response?: Response; + errorClassification?: { code: string; status: number; message: string }; } /** @@ -364,17 +226,12 @@ export interface FetchWithRetryResult { * Falls back to direct when no pool is available. * Logs every failure to `console.warn` so the operator can diagnose without * the error body leaking to the downstream client. - * - * @param ipv6Source - Optional IPv6 source address for outbound binding. - * When provided, uses curl instead of fetch() to bind - * to the specified source address. */ export async function fetchWithRetry( - url: string, - init: RequestInit & { proxy?: string }, - proxyPool?: ProxyPool, - context?: string, - ipv6Source?: string, + url: string, + init: RequestInit & { proxy?: string }, + proxyPool?: ProxyPool, + context?: string, ): Promise { let response: Response | undefined; let lastError: unknown; @@ -419,22 +276,9 @@ export async function fetchWithRetry( const proxyShort = init.proxy ? init.proxy.replace(/https?:\/\//, "").replace(/@.*/, "@***") : "direct"; logProxy("fetchWithRetry", `attempt=${attempt + 1}/${maxAttempts} proxy=${proxyShort}`, { context }); - try { - // Use curl for IPv6 source binding (only for direct connections) - if (ipv6Source && !init.proxy) { - response = await fetchViaCurl(url, init, ipv6Source, DEFAULT_TIMEOUT_MS); - // If curl failed with connection error (502, empty body), fallback to regular fetch - if (response.status === 502) { - const cloned = response.clone(); - const bodyText = await cloned.text(); - if (!bodyText) { - logProxy("fetchWithRetry", `IPv6 connection failed, falling back to regular fetch`, { ipv6Source, url }); - response = await fetch(url, init); - } - } - } else { - response = await fetch(url, init); - } + try { + response = await fetch(url, init); + if (response.ok) { if (usedProxy && proxyPool && proxyPool.size > 0) { proxyPool.markSuccess(); @@ -542,8 +386,6 @@ function classifyFetchErrorSafe(error: unknown): { * SSE streams: the initial request is retried normally. Once the response body * starts streaming, mid-stream errors are **not** retried; the session is * released and the error is returned to the caller. - * - * @param ipv6Source - Optional IPv6 source address for outbound binding. */ export async function fetchWithSessionRetry( url: string, @@ -552,18 +394,12 @@ export async function fetchWithSessionRetry( sessionId: string, context?: string, maxRetries?: number, - ipv6Source?: string, ): Promise { // Fallback when no session pool is available if (!sessionPool) { logProxy("fetchWithSessionRetry", "no session pool — direct fetch", { context, sessionId: sessionId.slice(0, 8) }); try { - let response: Response; - if (ipv6Source) { - response = await fetchViaCurl(url, init, ipv6Source, DEFAULT_TIMEOUT_MS); - } else { - response = await fetch(url, init); - } + const response = await fetch(url, init); return { response }; } catch (err) { return { errorClassification: classifyFetchErrorSafe(err) }; @@ -620,13 +456,7 @@ export async function fetchWithSessionRetry( }); try { - // Use curl for IPv6 source binding (only for direct connections) - let response: Response; - if (ipv6Source && !init.proxy) { - response = await fetchViaCurl(url, init, ipv6Source, DEFAULT_TIMEOUT_MS); - } else { - response = await fetch(url, init); - } + const response = await fetch(url, init); if (response.ok) { sessionPool.markSuccess(sessionId); @@ -650,7 +480,7 @@ export async function fetchWithSessionRetry( context, sessionId: sessionId.slice(0, 8), }); - const rotated = sessionPool.rotateNow(sessionId, model); + sessionPool.rotateNow(sessionId, model); continue; } diff --git a/src/lib/ipv6-pool.test.ts b/src/lib/ipv6-pool.test.ts deleted file mode 100644 index 7f495c1..0000000 --- a/src/lib/ipv6-pool.test.ts +++ /dev/null @@ -1,118 +0,0 @@ -import { test, expect, describe } from "bun:test"; -import { IPv6SourcePool } from "./ipv6-pool"; - -describe("IPv6SourcePool", () => { - test("should load addresses from comma-separated string", () => { - const pool = new IPv6SourcePool(); - pool.loadFromString("2001:db8::1,2001:db8::2,2001:db8::3"); - - expect(pool.size).toBe(3); - expect(pool.configured).toBe(true); - }); - - test("should return null when empty", () => { - const pool = new IPv6SourcePool(); - expect(pool.size).toBe(0); - expect(pool.configured).toBe(false); - expect(pool.getNext()).toBeNull(); - }); - - test("should rotate through addresses round-robin", () => { - const pool = new IPv6SourcePool(); - pool.loadFromString("2001:db8::1,2001:db8::2,2001:db8::3"); - - expect(pool.getNext()).toBe("2001:db8::1"); - expect(pool.getNext()).toBe("2001:db8::2"); - expect(pool.getNext()).toBe("2001:db8::3"); - expect(pool.getNext()).toBe("2001:db8::1"); // wraps around - }); - - test("should skip failed sources", () => { - const pool = new IPv6SourcePool(); - pool.loadFromString("2001:db8::1,2001:db8::2,2001:db8::3"); - pool.setFailureThreshold(1); - - pool.markFailed("2001:db8::2"); - - const results = [pool.getNext(), pool.getNext(), pool.getNext(), pool.getNext()]; - // Should skip 2001:db8::2 (disabled after 1 failure with threshold=1) - expect(results).not.toContain("2001:db8::2"); - }); - - test("should reset after all sources fail", () => { - const pool = new IPv6SourcePool(); - pool.loadFromString("2001:db8::1,2001:db8::2"); - - // Mark all as failed - pool.markFailed("2001:db8::1"); - pool.markFailed("2001:db8::1"); - pool.markFailed("2001:db8::1"); - pool.markFailed("2001:db8::2"); - pool.markFailed("2001:db8::2"); - pool.markFailed("2001:db8::2"); - - // Should auto-reset and return first - const addr = pool.getNext(); - expect(addr).toBe("2001:db8::1"); - }); - - test("should reset failure count on success", () => { - const pool = new IPv6SourcePool(); - pool.loadFromString("2001:db8::1,2001:db8::2"); - - pool.markFailed("2001:db8::1"); - pool.markFailed("2001:db8::1"); - pool.markSuccess("2001:db8::1"); - - // Should not be disabled after success - const addr = pool.getNext(); - expect(addr).toBe("2001:db8::1"); - }); - - test("should get address by index", () => { - const pool = new IPv6SourcePool(); - pool.loadFromString("2001:db8::1,2001:db8::2"); - - expect(pool.getAtIndex(0)).toBe("2001:db8::1"); - expect(pool.getAtIndex(1)).toBe("2001:db8::2"); - expect(pool.getAtIndex(2)).toBeNull(); - }); - - test("should handle whitespace in input", () => { - const pool = new IPv6SourcePool(); - pool.loadFromString(" 2001:db8::1 , 2001:db8::2 "); - - expect(pool.size).toBe(2); - expect(pool.getNext()).toBe("2001:db8::1"); - }); - - test("should handle empty input", () => { - const pool = new IPv6SourcePool(); - pool.loadFromString(""); - expect(pool.size).toBe(0); - }); - - test("should reset all sources", () => { - const pool = new IPv6SourcePool(); - pool.loadFromString("2001:db8::1,2001:db8::2"); - - pool.markFailed("2001:db8::1"); - pool.markFailed("2001:db8::1"); - pool.markFailed("2001:db8::1"); - pool.reset(); - - expect(pool.getNext()).toBe("2001:db8::1"); - }); - - test("should change failure threshold", () => { - const pool = new IPv6SourcePool(); - pool.loadFromString("2001:db8::1"); - pool.setFailureThreshold(1); - - pool.markFailed("2001:db8::1"); - // With threshold 1, should be disabled after 1 failure - // getNext should auto-reset since all failed - const addr = pool.getNext(); - expect(addr).toBe("2001:db8::1"); - }); -}); diff --git a/src/lib/ipv6-pool.ts b/src/lib/ipv6-pool.ts deleted file mode 100644 index 42bde3c..0000000 --- a/src/lib/ipv6-pool.ts +++ /dev/null @@ -1,186 +0,0 @@ -/** - * IPv6 source address pool for outbound rotation. - * - * Loads multiple IPv6 source addresses and rotates through them - * round-robin. Each outbound request can pick the next source IP - * via getNext(), ensuring requests are distributed across addresses. - * - * --- Environment Variables ---------------------------------------------------- - * IPV6_SOURCES — Comma-separated list of IPv6 source addresses - * e.g. "2001:df4:c140:1f::d6,2001:df4:c140:1f:ffff:ffff:ffff:ffff" - */ - -const POOL_PREFIX = "[ipv6-pool]"; - -function logPool(msg: string, extra?: Record): void { - const ts = new Date().toISOString().slice(11, 23); - const parts = [`${POOL_PREFIX} ${ts}`, msg]; - if (extra) { - for (const [k, v] of Object.entries(extra)) { - parts.push(`${k}=${v ?? "null"}`); - } - } - console.log(parts.join(" ")); -} - -// --- Types ------------------------------------------------------------------- - -export interface IPv6SourceEntry { - address: string; - /** Number of consecutive failures for this source */ - failures: number; - /** Whether this source is temporarily disabled */ - disabled: boolean; -} - -// --- IPv6SourcePool ---------------------------------------------------------- - -export class IPv6SourcePool { - private sources: IPv6SourceEntry[] = []; - private currentIndex = 0; - private failureThreshold = 3; - - /** - * Load IPv6 source addresses from a comma-separated string. - * - * @param csv - Comma-separated IPv6 addresses (e.g. "addr1,addr2,addr3") - */ - loadFromString(csv: string): void { - if (!csv) return; - const addrs = csv - .split(",") - .map((a) => a.trim()) - .filter(Boolean); - - this.sources = addrs.map((addr) => ({ - address: addr, - failures: 0, - disabled: false, - })); - this.currentIndex = 0; - - logPool(`loaded ${this.sources.length} IPv6 source addresses`); - for (const src of this.sources) { - logPool(` source: ${src.address}`); - } - } - - /** - * Load from environment variable IPV6_SOURCES. - */ - loadFromEnv(): void { - const env = process.env.IPV6_SOURCES; - if (env) { - this.loadFromString(env); - } - } - - // -- Access ------------------------------------------------------------------ - - /** Total number of source addresses. */ - get size(): number { - return this.sources.length; - } - - /** Whether any source addresses are configured. */ - get configured(): boolean { - return this.sources.length > 0; - } - - /** - * Get the next IPv6 source address (round-robin). - * Skips sources that have exceeded the failure threshold. - * - * @returns The next IPv6 address, or null if pool is empty or all failed. - */ - getNext(): string | null { - if (this.sources.length === 0) return null; - - const startIndex = this.currentIndex; - let checked = 0; - - do { - const entry = this.sources[this.currentIndex]; - if (entry && !entry.disabled) { - const addr = entry.address; - // Advance index for next call - this.currentIndex = (this.currentIndex + 1) % this.sources.length; - logPool(`getNext -> ${addr} (index=${this.currentIndex})`); - return addr; - } - this.currentIndex = (this.currentIndex + 1) % this.sources.length; - checked++; - } while (this.currentIndex !== startIndex && checked <= this.sources.length); - - // All sources failed — reset and return first - logPool("all IPv6 sources failed, resetting"); - for (const src of this.sources) { - src.failures = 0; - src.disabled = false; - } - this.currentIndex = 0; - return this.sources[0]?.address ?? null; - } - - /** - * Get a specific source address by index. - * Returns null if out of bounds. - */ - getAtIndex(index: number): string | null { - return this.sources[index]?.address ?? null; - } - - // -- Failure tracking -------------------------------------------------------- - - /** - * Mark a source address as failed. - * If failures exceed threshold, the source is disabled until reset. - * - * @param address - The IPv6 address that failed - */ - markFailed(address: string): void { - const entry = this.sources.find((s) => s.address === address); - if (!entry) return; - - entry.failures++; - logPool(`markFailed ${address} (${entry.failures}/${this.failureThreshold})`); - - if (entry.failures >= this.failureThreshold) { - entry.disabled = true; - console.warn( - `[ipv6-pool] Source ${address} failed ${entry.failures}/${this.failureThreshold} times — disabled`, - ); - } - } - - /** - * Mark a source address as successful (resets failure count). - * - * @param address - The IPv6 address that succeeded - */ - markSuccess(address: string): void { - const entry = this.sources.find((s) => s.address === address); - if (!entry) return; - - if (entry.failures > 0) { - logPool(`markSuccess ${address} (was ${entry.failures} failures)`); - } - entry.failures = 0; - entry.disabled = false; - } - - /** Set the failure threshold (default 3). */ - setFailureThreshold(n: number): void { - this.failureThreshold = n; - } - - /** Reset all failure states. */ - reset(): void { - for (const src of this.sources) { - src.failures = 0; - src.disabled = false; - } - this.currentIndex = 0; - logPool("reset all sources"); - } -} diff --git a/src/lib/router.ts b/src/lib/router.ts index 9b4908d..7439987 100644 --- a/src/lib/router.ts +++ b/src/lib/router.ts @@ -558,7 +558,7 @@ async function handleRelay( req: Request, env: RouterEnv, clientIP: string, - extra?: { ipv6Source?: string; skipProxyPool?: boolean }, + extra?: { skipProxyPool?: boolean }, ): Promise { const startTime = performance.now(); const method = req.method; @@ -603,7 +603,7 @@ async function handleRelay( const result = await fetchWithRetry( targetUrlString, fetchOptions, extra?.skipProxyPool ? undefined : proxyPool!, - "relay", extra?.ipv6Source, + "relay", ); if (result.errorClassification) { @@ -678,7 +678,6 @@ export interface RouterOptions { isWebSocketSupported?: boolean; getTestApiHtml?: () => string | Promise; skipProxyPool?: boolean; - ipv6Source?: string; } async function handleRequest( @@ -707,7 +706,7 @@ async function handleRequest( if (authErr) return authErr; try { const body = await req.json(); - return await handleChatCompletion(body, proxyPool ?? undefined, sessionPool ?? undefined, crypto.randomUUID(), options.ipv6Source); + return await handleChatCompletion(body, proxyPool ?? undefined, sessionPool ?? undefined, crypto.randomUUID()); } catch (err) { return createJsonErrorResponse(err); } @@ -722,7 +721,7 @@ async function handleRequest( try { const body = await req.json(); const anthropicVersion = req.headers.get("anthropic-version") ?? undefined; - return await handleAnthropicMessages(body, proxyPool ?? undefined, sessionPool ?? undefined, crypto.randomUUID(), options.ipv6Source, anthropicVersion); + return await handleAnthropicMessages(body, proxyPool ?? undefined, sessionPool ?? undefined, crypto.randomUUID(), anthropicVersion); } catch (err) { return createAnthropicJsonErrorResponse(err); } @@ -755,7 +754,6 @@ async function handleRequest( // Generic HTTP relay return handleRelay(req, env, clientIP, { - ipv6Source: options.ipv6Source, skipProxyPool: options.skipProxyPool, }); }