From 39a673b7d6c59f68e90b400da24449f2d1780c8f Mon Sep 17 00:00:00 2001 From: Paijo <14921983+oyi77@users.noreply.github.com> Date: Tue, 2 Jun 2026 05:46:58 +0700 Subject: [PATCH] fix(pollinations/duckduckgo): wire session pool for fingerprint rotation (#3049) * fix(pollinations): wire session pool + add idle pruning PollinationsExecutor.getPool() returned null because poolConfig was never set. Now sets DEFAULT_POOL_CONFIG in constructor so anonymous requests get fingerprint rotation and 429 cooldown management. Also: - Use acquireBlocking() instead of acquire() to wait for available session - Add startAutoPrune() for periodic idle session cleanup (5min idle timeout) - Improve execute() error handling with proper finally block * fix(duckduckgo-web): wire session pool for fingerprint rotation DuckDuckGoWebExecutor had poolConfig set but never called getPool() in execute(). Added session acquisition via acquireBlocking() and merges fingerprint headers into fetch calls for rate limit evasion. * fix: address PR #3049 review comments - duckduckgo-web: fix session leak (add finally release), fix race condition (report status before release), remove duplicate sessionHeaders - pollinations: fix race condition (move release to finally, report before) * fix: address all PR #3049 review comments - duckduckgo-web: add pool reporting on 429/500 early returns (was missing) - duckduckgo-web: add sessionHeaders to retry fetch on 401/403 - sessionPool: add .unref() to setInterval to prevent keeping process alive --------- Co-authored-by: oyi77 --- open-sse/executors/duckduckgo-web.ts | 34 ++++++++++++++++- open-sse/executors/pollinations.ts | 39 +++++++++++--------- open-sse/services/sessionPool/sessionPool.ts | 21 +++++++++-- 3 files changed, 72 insertions(+), 22 deletions(-) diff --git a/open-sse/executors/duckduckgo-web.ts b/open-sse/executors/duckduckgo-web.ts index 71a24034ed..0cea817dec 100644 --- a/open-sse/executors/duckduckgo-web.ts +++ b/open-sse/executors/duckduckgo-web.ts @@ -73,6 +73,16 @@ export class DuckDuckGoWebExecutor extends BaseExecutor { ); } + // Acquire session from pool for fingerprint rotation + const pool = this.getPool(); + let session; + try { + session = pool ? await pool.acquireBlocking(10_000) : null; + } catch { + session = null; + } + const sessionHeaders = session ? session.buildHeaders() : {}; + try { const controller = new AbortController(); const timeout = setTimeout(() => controller.abort(), FETCH_TIMEOUT_MS); @@ -101,6 +111,7 @@ export class DuckDuckGoWebExecutor extends BaseExecutor { method: "POST", headers: { ...FAKE_HEADERS, + ...sessionHeaders, "Content-Type": "application/json", "x-vqd-hash-1": vqdToken, }, @@ -115,6 +126,7 @@ export class DuckDuckGoWebExecutor extends BaseExecutor { clearTimeout(timeout); if (chatResponse.status === 429) { + if (pool && session) pool.reportCooldown(session); return new Response( JSON.stringify({ error: { message: "DuckDuckGo rate limited" } }), { status: 429, headers: { "Content-Type": "application/json" } } @@ -148,14 +160,32 @@ export class DuckDuckGoWebExecutor extends BaseExecutor { } if (chatResponse.status >= 500) { + if (pool && session) pool.reportDead(session); return new Response( JSON.stringify({ error: { message: "Upstream error" } }), { status: 502, headers: { "Content-Type": "application/json" } } ); } - return this.processResponse(chatResponse, stream !== false); + const result = this.processResponse(chatResponse, stream !== false); + + // Report pool status based on response + if (pool && session) { + if (chatResponse.status === 429) { + pool.reportCooldown(session); + } else if (chatResponse.status >= 500) { + pool.reportDead(session); + } else { + pool.reportSuccess(session); + } + } + + return result; } catch (error) { + if (pool && session) { + pool.reportCooldown(session); + } + if (error instanceof DOMException && error.name === "AbortError") { return new Response( JSON.stringify({ error: { message: "Request cancelled" } }), @@ -167,6 +197,8 @@ export class DuckDuckGoWebExecutor extends BaseExecutor { JSON.stringify({ error: { message: error instanceof Error ? error.message : "Unknown error" } }), { status: 500, headers: { "Content-Type": "application/json" } } ); + } finally { + session?.release(); } } diff --git a/open-sse/executors/pollinations.ts b/open-sse/executors/pollinations.ts index 3f5330d93a..5414823dfc 100644 --- a/open-sse/executors/pollinations.ts +++ b/open-sse/executors/pollinations.ts @@ -1,11 +1,12 @@ import { BaseExecutor } from "./base.ts"; import { PROVIDERS } from "../config/constants.ts"; -import { SessionPool, PoolRegistry } from "../services/sessionPool/index.ts"; +import { DEFAULT_POOL_CONFIG } from "../services/sessionPool/types.ts"; import type { ExecuteInput } from "./base.ts"; export class PollinationsExecutor extends BaseExecutor { constructor() { super("pollinations", PROVIDERS["pollinations"] || { format: "openai" }); + this.poolConfig = DEFAULT_POOL_CONFIG; } buildUrl(_model: string, _stream: boolean, urlIndex = 0, _credentials = null): string { @@ -50,7 +51,15 @@ export class PollinationsExecutor extends BaseExecutor { } const pool = this.getPool(); - const session = pool ? pool.acquire() : null; + + // Use acquireBlocking for anonymous requests to wait for available session + let session; + try { + session = pool ? await pool.acquireBlocking(10_000) : null; + } catch { + // Pool exhausted — fall through to direct request without fingerprint + session = null; + } if (session) { const fpHeaders = session.buildHeaders(); @@ -60,19 +69,10 @@ export class PollinationsExecutor extends BaseExecutor { }; } - let result; try { - result = await super.execute(input); - } catch (err) { - if (session && pool) { - pool.reportCooldown(session); - session.release(); - } - throw err; - } + const result = await super.execute(input); - if (session && pool) { - try { + if (session && pool) { const status = result.response.status; if (status === 429) { pool.reportCooldown(session); @@ -81,12 +81,17 @@ export class PollinationsExecutor extends BaseExecutor { } else { pool.reportSuccess(session); } - } finally { - session.release(); } - } - return result; + return result; + } catch (err) { + if (session && pool) { + pool.reportCooldown(session); + } + throw err; + } finally { + session?.release(); + } } } diff --git a/open-sse/services/sessionPool/sessionPool.ts b/open-sse/services/sessionPool/sessionPool.ts index 184c792cea..356928babd 100644 --- a/open-sse/services/sessionPool/sessionPool.ts +++ b/open-sse/services/sessionPool/sessionPool.ts @@ -300,14 +300,27 @@ export class SessionPool { } } - /** Remove dead sessions (call periodically for reclamation */ - pruneDeadSessions(): void { + /** Remove dead sessions and idle sessions older than maxIdleMs */ + pruneDeadSessions(maxIdleMs = 300_000): void { + const now = Date.now(); const before = this.sessions.length; - this.sessions = this.sessions.filter((s) => s.status !== "dead"); + this.sessions = this.sessions.filter((s) => { + if (s.status === "dead") return false; + // Prune idle sessions older than maxIdleMs (default 5min) + if (s.inflight === 0 && s.lastUsedAt > 0 && now - s.lastUsedAt > maxIdleMs) return false; + return true; + }); - // If we pruned sessions, report + // If we pruned sessions, ensure minimum if (this.sessions.length < before && this.sessions.length < this.config.minSessions) { this.ensureMinSessions(); } } + + /** Start periodic pruning (every 60s) */ + startAutoPrune(intervalMs = 60_000): ReturnType { + const timer = setInterval(() => this.pruneDeadSessions(), intervalMs); + timer.unref(); + return timer; + } }