chore: hapus fitur IPv6 source rotation
- Hapus src/lib/ipv6-pool.ts dan test - Hapus fetchViaCurl() dari fetch-utils.ts - Hapus parameter ipv6Source dari router, ai-proxy, anthropic-proxy - Hapus ipv6Only: false dari Bun.serve(), default bind ke 0.0.0.0 - Hapus dokumentasi IPv6 dari CLAUDE.md dan README.md - SSRF protection untuk IPv6 private/loopback tetap ada Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.8
parent
9a379d09b5
commit
f78f1f5bbf
@@ -7,49 +7,8 @@ Key architecture facts:
|
|||||||
- Middleware stack: rate limiter, body limiter, structured logger, SSRF protection
|
- Middleware stack: rate limiter, body limiter, structured logger, SSRF protection
|
||||||
- WebSocket relay: bidirectional relay via `x-relay-target` header with `ws://` or `wss://`
|
- 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
|
- 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
|
- 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 <ipv6>`
|
|
||||||
- **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.
|
Default to using Bun instead of Node.js.
|
||||||
|
|
||||||
|
|||||||
+2
-17
@@ -5,7 +5,6 @@
|
|||||||
* Adds:
|
* Adds:
|
||||||
* - Bun.serve() bindings
|
* - Bun.serve() bindings
|
||||||
* - WebSocket relay support
|
* - WebSocket relay support
|
||||||
* - IPv6 source rotation
|
|
||||||
* - Proxy pool file loading
|
* - Proxy pool file loading
|
||||||
* - Graceful shutdown
|
* - Graceful shutdown
|
||||||
*/
|
*/
|
||||||
@@ -25,7 +24,6 @@ import {
|
|||||||
} from "./lib/router";
|
} from "./lib/router";
|
||||||
import type { RouterEnv } from "./lib/router";
|
import type { RouterEnv } from "./lib/router";
|
||||||
|
|
||||||
import { IPv6SourcePool } from "./lib/ipv6-pool";
|
|
||||||
import { closeAllActiveReaders, isDevMode } from "./lib/fetch-utils";
|
import { closeAllActiveReaders, isDevMode } from "./lib/fetch-utils";
|
||||||
|
|
||||||
import type { Server, ServerWebSocket } from "bun";
|
import type { Server, ServerWebSocket } from "bun";
|
||||||
@@ -34,7 +32,7 @@ import type { Server, ServerWebSocket } from "bun";
|
|||||||
|
|
||||||
const RELAY_VERSION = "1.0.0";
|
const RELAY_VERSION = "1.0.0";
|
||||||
const PORT = Number.parseInt(process.env.PORT ?? "3000", 10);
|
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) ----------------------------------------------------
|
// --- Proxy pool (optional) ----------------------------------------------------
|
||||||
|
|
||||||
@@ -43,14 +41,6 @@ proxyPool.tryLoad(
|
|||||||
process.env.PROXY_FILE || process.env.PROXY_LIST || "./proxy.txt",
|
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 --------------------------------------------
|
// --- SSRF DNS rebinding protection --------------------------------------------
|
||||||
|
|
||||||
if (process.env.SSRF_DNS_CHECK === "true") {
|
if (process.env.SSRF_DNS_CHECK === "true") {
|
||||||
@@ -116,13 +106,11 @@ function handleWebSocketUpgrade(
|
|||||||
|
|
||||||
const server: Server<WSRelayData> = Bun.serve<WSRelayData>({
|
const server: Server<WSRelayData> = Bun.serve<WSRelayData>({
|
||||||
hostname: HOST,
|
hostname: HOST,
|
||||||
ipv6Only: false,
|
|
||||||
port: PORT,
|
port: PORT,
|
||||||
development: isDevMode() ? { hmr: true, console: true } : undefined,
|
development: isDevMode() ? { hmr: true, console: true } : undefined,
|
||||||
|
|
||||||
async fetch(req: Request): Promise<Response | undefined> {
|
async fetch(req: Request): Promise<Response | undefined> {
|
||||||
const url = new URL(req.url);
|
const url = new URL(req.url);
|
||||||
const ipv6Source = ipv6Pool.getNext() ?? undefined;
|
|
||||||
|
|
||||||
// Static routes
|
// Static routes
|
||||||
if (url.pathname === "/health") return handleHealth();
|
if (url.pathname === "/health") return handleHealth();
|
||||||
@@ -157,10 +145,7 @@ const server: Server<WSRelayData> = Bun.serve<WSRelayData>({
|
|||||||
req,
|
req,
|
||||||
routerEnv,
|
routerEnv,
|
||||||
clientIP,
|
clientIP,
|
||||||
{
|
{ isWebSocketSupported: true },
|
||||||
isWebSocketSupported: true,
|
|
||||||
ipv6Source,
|
|
||||||
},
|
|
||||||
);
|
);
|
||||||
if (result !== undefined) return result;
|
if (result !== undefined) return result;
|
||||||
|
|
||||||
|
|||||||
+4
-12
@@ -407,14 +407,6 @@ export async function handleChatCompletion(
|
|||||||
proxyPool?: ProxyPool,
|
proxyPool?: ProxyPool,
|
||||||
sessionPool?: SessionProxyPool,
|
sessionPool?: SessionProxyPool,
|
||||||
sessionId?: string,
|
sessionId?: string,
|
||||||
ipv6Source?: string,
|
|
||||||
): Promise<Response>;
|
|
||||||
export async function handleChatCompletion(
|
|
||||||
body: unknown,
|
|
||||||
proxyPool?: ProxyPool,
|
|
||||||
sessionPool?: SessionProxyPool,
|
|
||||||
sessionId?: string,
|
|
||||||
ipv6Source?: string,
|
|
||||||
): Promise<Response> {
|
): Promise<Response> {
|
||||||
// -- Input validation -------------------------------------------------------
|
// -- Input validation -------------------------------------------------------
|
||||||
const validationError = validateChatRequest(body);
|
const validationError = validateChatRequest(body);
|
||||||
@@ -457,8 +449,8 @@ export async function handleChatCompletion(
|
|||||||
// -- Execute with session-aware or standard retry --------------------------
|
// -- Execute with session-aware or standard retry --------------------------
|
||||||
let result: FetchWithRetryResult =
|
let result: FetchWithRetryResult =
|
||||||
sessionPool && sessionId
|
sessionPool && sessionId
|
||||||
? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `openai:${req.model}`, undefined, ipv6Source)
|
? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `openai:${req.model}`, undefined)
|
||||||
: await fetchWithRetry(url, init, proxyPool, `openai:${req.model}`, ipv6Source);
|
: await fetchWithRetry(url, init, proxyPool, `openai:${req.model}`);
|
||||||
|
|
||||||
// -- Mimo Free: auth failure → invalidate JWT and retry once ---------------
|
// -- Mimo Free: auth failure → invalidate JWT and retry once ---------------
|
||||||
if (
|
if (
|
||||||
@@ -474,8 +466,8 @@ export async function handleChatCompletion(
|
|||||||
};
|
};
|
||||||
result =
|
result =
|
||||||
sessionPool && sessionId
|
sessionPool && sessionId
|
||||||
? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `openai:${req.model}`, undefined, ipv6Source)
|
? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `openai:${req.model}`, undefined)
|
||||||
: await fetchWithRetry(url, init, proxyPool, `openai:${req.model}`, ipv6Source);
|
: await fetchWithRetry(url, init, proxyPool, `openai:${req.model}`);
|
||||||
}
|
}
|
||||||
|
|
||||||
if (result.errorClassification) {
|
if (result.errorClassification) {
|
||||||
|
|||||||
@@ -1034,22 +1034,6 @@ export async function handleAnthropicMessages(
|
|||||||
proxyPool?: ProxyPool,
|
proxyPool?: ProxyPool,
|
||||||
sessionPool?: SessionProxyPool,
|
sessionPool?: SessionProxyPool,
|
||||||
sessionId?: string,
|
sessionId?: string,
|
||||||
ipv6Source?: string,
|
|
||||||
): Promise<Response>;
|
|
||||||
export async function handleAnthropicMessages(
|
|
||||||
body: unknown,
|
|
||||||
proxyPool?: ProxyPool,
|
|
||||||
sessionPool?: SessionProxyPool,
|
|
||||||
sessionId?: string,
|
|
||||||
ipv6Source?: string,
|
|
||||||
anthropicVersion?: string,
|
|
||||||
): Promise<Response>;
|
|
||||||
export async function handleAnthropicMessages(
|
|
||||||
body: unknown,
|
|
||||||
proxyPool?: ProxyPool,
|
|
||||||
sessionPool?: SessionProxyPool,
|
|
||||||
sessionId?: string,
|
|
||||||
ipv6Source?: string,
|
|
||||||
anthropicVersion?: string,
|
anthropicVersion?: string,
|
||||||
): Promise<Response> {
|
): Promise<Response> {
|
||||||
// -- Input validation -------------------------------------------------------
|
// -- Input validation -------------------------------------------------------
|
||||||
@@ -1091,8 +1075,8 @@ export async function handleAnthropicMessages(
|
|||||||
// -- Execute with session-aware or standard retry --------------------------
|
// -- Execute with session-aware or standard retry --------------------------
|
||||||
const result: FetchWithRetryResult =
|
const result: FetchWithRetryResult =
|
||||||
sessionPool && sessionId
|
sessionPool && sessionId
|
||||||
? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `anthropic:${req.model}`, undefined, ipv6Source)
|
? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `anthropic:${req.model}`, undefined)
|
||||||
: await fetchWithRetry(url, init, proxyPool, `anthropic:${req.model}`, ipv6Source);
|
: await fetchWithRetry(url, init, proxyPool, `anthropic:${req.model}`);
|
||||||
|
|
||||||
if (result.errorClassification) {
|
if (result.errorClassification) {
|
||||||
if (sessionPool && sessionId) sessionPool.release(sessionId);
|
if (sessionPool && sessionId) sessionPool.release(sessionId);
|
||||||
|
|||||||
+13
-183
@@ -6,13 +6,6 @@
|
|||||||
*/
|
*/
|
||||||
|
|
||||||
import type { ProxyPool, SessionProxyPool } from "./proxy-pool";
|
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) ───────────────────────
|
// ─── Active stream tracking (for graceful shutdown) ───────────────────────
|
||||||
|
|
||||||
@@ -129,7 +122,7 @@ export class SSELineBuffer {
|
|||||||
if (this.overflow) return [];
|
if (this.overflow) return [];
|
||||||
|
|
||||||
this.buffer += chunk;
|
this.buffer += chunk;
|
||||||
|
|
||||||
if (this.buffer.length > this.MAX_BUFFER_SIZE) {
|
if (this.buffer.length > this.MAX_BUFFER_SIZE) {
|
||||||
console.warn(`[SSELineBuffer] Buffer exceeded ${this.MAX_BUFFER_SIZE} bytes — discarding remaining stream data`);
|
console.warn(`[SSELineBuffer] Buffer exceeded ${this.MAX_BUFFER_SIZE} bytes — discarding remaining stream data`);
|
||||||
this.overflow = true;
|
this.overflow = true;
|
||||||
@@ -219,142 +212,11 @@ export function sanitizeErrorMessage(raw: string): string {
|
|||||||
return "Upstream error";
|
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<Response> {
|
|
||||||
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) ─────────────────────────
|
// ─── Fetch with retry (direct → proxy fallback) ─────────────────────────
|
||||||
|
|
||||||
export interface FetchWithRetryResult {
|
export interface FetchWithRetryResult {
|
||||||
response?: Response;
|
response?: Response;
|
||||||
errorClassification?: { code: string; status: number; message: string };
|
errorClassification?: { code: string; status: number; message: string };
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -364,17 +226,12 @@ export interface FetchWithRetryResult {
|
|||||||
* Falls back to direct when no pool is available.
|
* Falls back to direct when no pool is available.
|
||||||
* Logs every failure to `console.warn` so the operator can diagnose without
|
* Logs every failure to `console.warn` so the operator can diagnose without
|
||||||
* the error body leaking to the downstream client.
|
* 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(
|
export async function fetchWithRetry(
|
||||||
url: string,
|
url: string,
|
||||||
init: RequestInit & { proxy?: string },
|
init: RequestInit & { proxy?: string },
|
||||||
proxyPool?: ProxyPool,
|
proxyPool?: ProxyPool,
|
||||||
context?: string,
|
context?: string,
|
||||||
ipv6Source?: string,
|
|
||||||
): Promise<FetchWithRetryResult> {
|
): Promise<FetchWithRetryResult> {
|
||||||
let response: Response | undefined;
|
let response: Response | undefined;
|
||||||
let lastError: unknown;
|
let lastError: unknown;
|
||||||
@@ -419,22 +276,9 @@ export async function fetchWithRetry(
|
|||||||
const proxyShort = init.proxy ? init.proxy.replace(/https?:\/\//, "").replace(/@.*/, "@***") : "direct";
|
const proxyShort = init.proxy ? init.proxy.replace(/https?:\/\//, "").replace(/@.*/, "@***") : "direct";
|
||||||
logProxy("fetchWithRetry", `attempt=${attempt + 1}/${maxAttempts} proxy=${proxyShort}`, { context });
|
logProxy("fetchWithRetry", `attempt=${attempt + 1}/${maxAttempts} proxy=${proxyShort}`, { context });
|
||||||
|
|
||||||
try {
|
try {
|
||||||
// Use curl for IPv6 source binding (only for direct connections)
|
response = await fetch(url, init);
|
||||||
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);
|
|
||||||
}
|
|
||||||
if (response.ok) {
|
if (response.ok) {
|
||||||
if (usedProxy && proxyPool && proxyPool.size > 0) {
|
if (usedProxy && proxyPool && proxyPool.size > 0) {
|
||||||
proxyPool.markSuccess();
|
proxyPool.markSuccess();
|
||||||
@@ -542,8 +386,6 @@ function classifyFetchErrorSafe(error: unknown): {
|
|||||||
* SSE streams: the initial request is retried normally. Once the response body
|
* SSE streams: the initial request is retried normally. Once the response body
|
||||||
* starts streaming, mid-stream errors are **not** retried; the session is
|
* starts streaming, mid-stream errors are **not** retried; the session is
|
||||||
* released and the error is returned to the caller.
|
* released and the error is returned to the caller.
|
||||||
*
|
|
||||||
* @param ipv6Source - Optional IPv6 source address for outbound binding.
|
|
||||||
*/
|
*/
|
||||||
export async function fetchWithSessionRetry(
|
export async function fetchWithSessionRetry(
|
||||||
url: string,
|
url: string,
|
||||||
@@ -552,18 +394,12 @@ export async function fetchWithSessionRetry(
|
|||||||
sessionId: string,
|
sessionId: string,
|
||||||
context?: string,
|
context?: string,
|
||||||
maxRetries?: number,
|
maxRetries?: number,
|
||||||
ipv6Source?: string,
|
|
||||||
): Promise<FetchWithRetryResult> {
|
): Promise<FetchWithRetryResult> {
|
||||||
// Fallback when no session pool is available
|
// Fallback when no session pool is available
|
||||||
if (!sessionPool) {
|
if (!sessionPool) {
|
||||||
logProxy("fetchWithSessionRetry", "no session pool — direct fetch", { context, sessionId: sessionId.slice(0, 8) });
|
logProxy("fetchWithSessionRetry", "no session pool — direct fetch", { context, sessionId: sessionId.slice(0, 8) });
|
||||||
try {
|
try {
|
||||||
let response: Response;
|
const response = await fetch(url, init);
|
||||||
if (ipv6Source) {
|
|
||||||
response = await fetchViaCurl(url, init, ipv6Source, DEFAULT_TIMEOUT_MS);
|
|
||||||
} else {
|
|
||||||
response = await fetch(url, init);
|
|
||||||
}
|
|
||||||
return { response };
|
return { response };
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
return { errorClassification: classifyFetchErrorSafe(err) };
|
return { errorClassification: classifyFetchErrorSafe(err) };
|
||||||
@@ -620,13 +456,7 @@ export async function fetchWithSessionRetry(
|
|||||||
});
|
});
|
||||||
|
|
||||||
try {
|
try {
|
||||||
// Use curl for IPv6 source binding (only for direct connections)
|
const response = await fetch(url, init);
|
||||||
let response: Response;
|
|
||||||
if (ipv6Source && !init.proxy) {
|
|
||||||
response = await fetchViaCurl(url, init, ipv6Source, DEFAULT_TIMEOUT_MS);
|
|
||||||
} else {
|
|
||||||
response = await fetch(url, init);
|
|
||||||
}
|
|
||||||
|
|
||||||
if (response.ok) {
|
if (response.ok) {
|
||||||
sessionPool.markSuccess(sessionId);
|
sessionPool.markSuccess(sessionId);
|
||||||
@@ -650,7 +480,7 @@ export async function fetchWithSessionRetry(
|
|||||||
context,
|
context,
|
||||||
sessionId: sessionId.slice(0, 8),
|
sessionId: sessionId.slice(0, 8),
|
||||||
});
|
});
|
||||||
const rotated = sessionPool.rotateNow(sessionId, model);
|
sessionPool.rotateNow(sessionId, model);
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -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");
|
|
||||||
});
|
|
||||||
});
|
|
||||||
@@ -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<string, unknown>): 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");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
+4
-6
@@ -558,7 +558,7 @@ async function handleRelay(
|
|||||||
req: Request,
|
req: Request,
|
||||||
env: RouterEnv,
|
env: RouterEnv,
|
||||||
clientIP: string,
|
clientIP: string,
|
||||||
extra?: { ipv6Source?: string; skipProxyPool?: boolean },
|
extra?: { skipProxyPool?: boolean },
|
||||||
): Promise<Response> {
|
): Promise<Response> {
|
||||||
const startTime = performance.now();
|
const startTime = performance.now();
|
||||||
const method = req.method;
|
const method = req.method;
|
||||||
@@ -603,7 +603,7 @@ async function handleRelay(
|
|||||||
const result = await fetchWithRetry(
|
const result = await fetchWithRetry(
|
||||||
targetUrlString, fetchOptions,
|
targetUrlString, fetchOptions,
|
||||||
extra?.skipProxyPool ? undefined : proxyPool!,
|
extra?.skipProxyPool ? undefined : proxyPool!,
|
||||||
"relay", extra?.ipv6Source,
|
"relay",
|
||||||
);
|
);
|
||||||
|
|
||||||
if (result.errorClassification) {
|
if (result.errorClassification) {
|
||||||
@@ -678,7 +678,6 @@ export interface RouterOptions {
|
|||||||
isWebSocketSupported?: boolean;
|
isWebSocketSupported?: boolean;
|
||||||
getTestApiHtml?: () => string | Promise<string>;
|
getTestApiHtml?: () => string | Promise<string>;
|
||||||
skipProxyPool?: boolean;
|
skipProxyPool?: boolean;
|
||||||
ipv6Source?: string;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
async function handleRequest(
|
async function handleRequest(
|
||||||
@@ -707,7 +706,7 @@ async function handleRequest(
|
|||||||
if (authErr) return authErr;
|
if (authErr) return authErr;
|
||||||
try {
|
try {
|
||||||
const body = await req.json();
|
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) {
|
} catch (err) {
|
||||||
return createJsonErrorResponse(err);
|
return createJsonErrorResponse(err);
|
||||||
}
|
}
|
||||||
@@ -722,7 +721,7 @@ async function handleRequest(
|
|||||||
try {
|
try {
|
||||||
const body = await req.json();
|
const body = await req.json();
|
||||||
const anthropicVersion = req.headers.get("anthropic-version") ?? undefined;
|
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) {
|
} catch (err) {
|
||||||
return createAnthropicJsonErrorResponse(err);
|
return createAnthropicJsonErrorResponse(err);
|
||||||
}
|
}
|
||||||
@@ -755,7 +754,6 @@ async function handleRequest(
|
|||||||
|
|
||||||
// Generic HTTP relay
|
// Generic HTTP relay
|
||||||
return handleRelay(req, env, clientIP, {
|
return handleRelay(req, env, clientIP, {
|
||||||
ipv6Source: options.ipv6Source,
|
|
||||||
skipProxyPool: options.skipProxyPool,
|
skipProxyPool: options.skipProxyPool,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user