fix: rotate proxy on every failure, return proper HTTP status
- Add rotateNow() to SessionProxyPool — force-rotate session to a different proxy immediately (excludes current index to ensure real rotation). Uses round-robin scan from oldIndex+1 so all proxies get used, not just bouncing between two. - fetchWithSessionRetry: call rotateNow() on every failure instead of markFailed() which only rotated after threshold. Return last HTTP response (e.g. 429) instead of classifying as 502 when we have one. - fetchWithRetry: rotate pool on every failure for consistency. Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
+28
-16
@@ -168,18 +168,18 @@ export async function fetchWithRetry(
|
|||||||
return { response };
|
return { response };
|
||||||
}
|
}
|
||||||
|
|
||||||
// Non-2xx — mark proxy as failed and retry
|
// Non-2xx — mark proxy as failed and rotate for next attempt
|
||||||
lastError = new Error(`Upstream returned ${response.status}`);
|
lastError = new Error(`Upstream returned ${response.status}`);
|
||||||
logProxy("fetchWithRetry", `non-2xx attempt=${attempt + 1} status=${response.status}`, { context });
|
logProxy("fetchWithRetry", `non-2xx attempt=${attempt + 1} status=${response.status}`, { context });
|
||||||
if (usedProxy && proxyPool && proxyPool.size > 0 && init.proxy) {
|
if (usedProxy && proxyPool && proxyPool.size > 0 && init.proxy) {
|
||||||
proxyPool.markFailed();
|
proxyPool.markFailed();
|
||||||
usedProxy = false;
|
proxyPool.rotate();
|
||||||
}
|
}
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
lastError = err;
|
lastError = err;
|
||||||
if (usedProxy && proxyPool && proxyPool.size > 0 && init.proxy) {
|
if (usedProxy && proxyPool && proxyPool.size > 0 && init.proxy) {
|
||||||
proxyPool.markFailed();
|
proxyPool.markFailed();
|
||||||
usedProxy = false;
|
proxyPool.rotate();
|
||||||
}
|
}
|
||||||
|
|
||||||
// Log the actual error so operators can diagnose
|
// Log the actual error so operators can diagnose
|
||||||
@@ -190,16 +190,17 @@ export async function fetchWithRetry(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// All attempts exhausted — classify the last error
|
// All attempts exhausted
|
||||||
if (!response) {
|
logProxy("fetchWithRetry", "exhausted all retries", { context, hadResponse: !!response });
|
||||||
const err = lastError ?? new Error("All connection attempts failed");
|
|
||||||
logProxy("fetchWithRetry", "exhausted all retries", { context });
|
// If we got at least one HTTP response, return it as-is so the caller
|
||||||
return { errorClassification: classifyFetchErrorSafe(err) };
|
// can relay the proper status code (e.g. 429).
|
||||||
|
if (response) {
|
||||||
|
return { response };
|
||||||
}
|
}
|
||||||
|
|
||||||
// Non-2xx but we have a response — pass it along (caller handles it)
|
const err = lastError ?? new Error("All connection attempts failed");
|
||||||
logProxy("fetchWithRetry", `returning non-2xx response status=${response.status}`, { context });
|
return { errorClassification: classifyFetchErrorSafe(err) };
|
||||||
return { response };
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -270,6 +271,7 @@ export async function fetchWithSessionRetry(
|
|||||||
}
|
}
|
||||||
|
|
||||||
let lastError: unknown;
|
let lastError: unknown;
|
||||||
|
let lastResponse: Response | undefined;
|
||||||
|
|
||||||
for (let attempt = 0; attempt < maxRetries; attempt++) {
|
for (let attempt = 0; attempt < maxRetries; attempt++) {
|
||||||
// First attempt: acquire() assigns a proxy to this session (least-loaded
|
// First attempt: acquire() assigns a proxy to this session (least-loaded
|
||||||
@@ -303,9 +305,10 @@ export async function fetchWithSessionRetry(
|
|||||||
return { response };
|
return { response };
|
||||||
}
|
}
|
||||||
|
|
||||||
// Non-2xx — mark session failed and retry
|
// Non-2xx — rotate proxy immediately for the next attempt
|
||||||
lastError = new Error(`Upstream returned ${response.status}`);
|
lastError = new Error(`Upstream returned ${response.status}`);
|
||||||
const rotated = sessionPool.markFailed(sessionId);
|
lastResponse = response;
|
||||||
|
const rotated = sessionPool.rotateNow(sessionId);
|
||||||
logProxy("fetchWithSessionRetry", `non-2xx attempt=${attempt + 1} status=${response.status} rotated=${rotated}`, {
|
logProxy("fetchWithSessionRetry", `non-2xx attempt=${attempt + 1} status=${response.status} rotated=${rotated}`, {
|
||||||
context,
|
context,
|
||||||
sessionId: sessionId.slice(0, 8),
|
sessionId: sessionId.slice(0, 8),
|
||||||
@@ -319,7 +322,7 @@ export async function fetchWithSessionRetry(
|
|||||||
);
|
);
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
lastError = err;
|
lastError = err;
|
||||||
const rotated = sessionPool.markFailed(sessionId);
|
const rotated = sessionPool.rotateNow(sessionId);
|
||||||
const errMsg = err instanceof Error ? err.message : String(err);
|
const errMsg = err instanceof Error ? err.message : String(err);
|
||||||
logProxy("fetchWithSessionRetry", `failed attempt=${attempt + 1} err=${errMsg} rotated=${rotated}`, {
|
logProxy("fetchWithSessionRetry", `failed attempt=${attempt + 1} err=${errMsg} rotated=${rotated}`, {
|
||||||
context,
|
context,
|
||||||
@@ -335,12 +338,21 @@ export async function fetchWithSessionRetry(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// All attempts exhausted — release session and classify last error
|
// All attempts exhausted
|
||||||
logProxy("fetchWithSessionRetry", "exhausted all retries — releasing session", {
|
logProxy("fetchWithSessionRetry", "exhausted all retries", {
|
||||||
context,
|
context,
|
||||||
sessionId: sessionId.slice(0, 8),
|
sessionId: sessionId.slice(0, 8),
|
||||||
|
hadResponse: !!lastResponse,
|
||||||
});
|
});
|
||||||
sessionPool.release(sessionId);
|
sessionPool.release(sessionId);
|
||||||
|
|
||||||
|
// If we got at least one HTTP response (e.g. 429), return it so the caller
|
||||||
|
// can relay the proper status code. Only classify as error on network
|
||||||
|
// failures where there's no response at all.
|
||||||
|
if (lastResponse) {
|
||||||
|
return { response: lastResponse };
|
||||||
|
}
|
||||||
|
|
||||||
const err = lastError ?? new Error("All session proxy attempts failed");
|
const err = lastError ?? new Error("All session proxy attempts failed");
|
||||||
return { errorClassification: classifyFetchErrorSafe(err) };
|
return { errorClassification: classifyFetchErrorSafe(err) };
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -361,6 +361,60 @@ export class SessionProxyPool {
|
|||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Force-rotate this session to the least-loaded proxy immediately,
|
||||||
|
* regardless of failure count. Resets the session's failure counter.
|
||||||
|
*
|
||||||
|
* @returns true if the session was moved to a different proxy.
|
||||||
|
*/
|
||||||
|
rotateNow(sessionId: string): boolean {
|
||||||
|
const info = this.sessions.get(sessionId);
|
||||||
|
if (!info) return false;
|
||||||
|
|
||||||
|
const oldIndex = info.proxyIndex;
|
||||||
|
const oldEntry = this.poolEntryAtIndex(oldIndex);
|
||||||
|
|
||||||
|
// Remove from old proxy usage first
|
||||||
|
const usedBy = this.proxyUsage.get(oldIndex);
|
||||||
|
if (usedBy) {
|
||||||
|
usedBy.delete(sessionId);
|
||||||
|
if (usedBy.size === 0) this.proxyUsage.delete(oldIndex);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Find the least-loaded proxy that is NOT the current one.
|
||||||
|
// Start scanning from (oldIndex + 1) so we don't immediately
|
||||||
|
// bounce back to index 0 when all usage counts are equal.
|
||||||
|
let bestIndex = -1;
|
||||||
|
let bestCount = Infinity;
|
||||||
|
for (let step = 1; step <= this.pool.size; step++) {
|
||||||
|
const i = (oldIndex + step) % this.pool.size;
|
||||||
|
const count = this.proxyUsage.get(i)?.size ?? 0;
|
||||||
|
if (count < bestCount) {
|
||||||
|
bestCount = count;
|
||||||
|
bestIndex = i;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (bestIndex === -1) {
|
||||||
|
// Single-proxy pool — reinstate and give up
|
||||||
|
if (usedBy) usedBy.add(sessionId);
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
this.sessions.set(sessionId, { proxyIndex: bestIndex, failures: 0 });
|
||||||
|
|
||||||
|
let newUsedBy = this.proxyUsage.get(bestIndex);
|
||||||
|
if (!newUsedBy) {
|
||||||
|
newUsedBy = new Set();
|
||||||
|
this.proxyUsage.set(bestIndex, newUsedBy);
|
||||||
|
}
|
||||||
|
newUsedBy.add(sessionId);
|
||||||
|
|
||||||
|
const newEntry = this.poolEntryAtIndex(bestIndex);
|
||||||
|
logPool(`rotateNow session=${sessionId.slice(0, 8)} ${oldEntry?.host} -> ${newEntry?.host}`);
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
/** Reset failure count for this session's proxy. */
|
/** Reset failure count for this session's proxy. */
|
||||||
markSuccess(sessionId: string): void {
|
markSuccess(sessionId: string): void {
|
||||||
const info = this.sessions.get(sessionId);
|
const info = this.sessions.get(sessionId);
|
||||||
|
|||||||
Reference in New Issue
Block a user