From 32810f6654b4d97028dec3858ef9d020342d0474 Mon Sep 17 00:00:00 2001 From: Zihan Dai Date: Wed, 4 Nov 2026 00:39:54 +1100 Subject: [PATCH] Reject getSession() waiters when the pool is closed close() cleared the wait queue without settling the queued promises. A caller waiting in getSession() when the pool was closed stayed pending until waitTimeout (60 s by default) and then failed with "Timeout waiting for available session", which points at pool exhaustion rather than at the close. If nothing else kept the process alive, the promise never settled at all, because the wait timer is unref'd. Reject the waiters with " is closed" as soon as close() starts. Do it again at the end instead of clearing the queue, for callers that started waiting while the sessions were being closed and were not handed a session released in the meantime. Signed-off-by: Zihan Dai --- src/client/BaseSessionPool.ts | 28 ++++++++++-- tests/unit/BaseSessionPoolLifecycle.test.ts | 47 +++++++++++++++++++++ 2 files changed, 71 insertions(+), 4 deletions(-) diff --git a/src/client/BaseSessionPool.ts b/src/client/BaseSessionPool.ts index cfd0ff0..523ae55 100644 --- a/src/client/BaseSessionPool.ts +++ b/src/client/BaseSessionPool.ts @@ -50,7 +50,8 @@ export abstract class BaseSessionPool { protected config: PoolConfig; protected endPoints: EndPoint[]; protected pool: PooledSession[] = []; - protected waitQueue: Denque<(session: Session) => boolean> = new Denque(); + protected waitQueue: Denque<(result: Session | Error) => boolean> = + new Denque(); protected idleSessions: Denque = new Denque(); protected activeSessions: Set = new Set(); protected currentEndPointIndex = 0; @@ -295,15 +296,20 @@ export abstract class BaseSessionPool { // reference (not `resolve`, which is never in the queue), and the // `settled` guard makes timeout and fulfillment mutually exclusive so a // session is never handed to a waiter whose promise already rejected. - const waiter = (session: Session): boolean => { + // close() passes an Error instead of a session to fail the waiter. + const waiter = (result: Session | Error): boolean => { if (settled) { return false; } settled = true; clearTimeout(timeoutId); + if (result instanceof Error) { + reject(result); + return true; + } const duration = Date.now() - startTime; logger.debug(`[PERF] getSession (waited): ${duration}ms`); - resolve(session); + resolve(result); return true; }; @@ -508,6 +514,11 @@ export abstract class BaseSessionPool { this.cleanupInterval = null; } + // Fail the callers waiting in getSession() now. Otherwise each of them + // waits for waitTimeout and then reports pool exhaustion, or never settles + // at all if nothing else keeps the process alive (the timeout is unref'd). + this.rejectWaiters(); + // Close all sessions await Promise.all( this.pool.map(async (ps) => { @@ -522,13 +533,22 @@ export abstract class BaseSessionPool { this.pool = []; this.idleSessions.clear(); this.activeSessions.clear(); - this.waitQueue.clear(); + // getSession() calls that started waiting while the sessions were closing. + // A session released during close() can still have been handed to one of + // them, as before. + this.rejectWaiters(); this.endPointToSession.clear(); this.redirectCache.clear(); unregisterClosable(this); logger.info(`${this.getPoolName()} closed`); } + private rejectWaiters(): void { + while (this.waitQueue.length > 0) { + this.waitQueue.shift()?.(new Error(`${this.getPoolName()} is closed`)); + } + } + getPoolSize(): number { return this.pool.length; } diff --git a/tests/unit/BaseSessionPoolLifecycle.test.ts b/tests/unit/BaseSessionPoolLifecycle.test.ts index 64bf2e6..b9b3cda 100644 --- a/tests/unit/BaseSessionPoolLifecycle.test.ts +++ b/tests/unit/BaseSessionPoolLifecycle.test.ts @@ -77,6 +77,18 @@ function newPool(overrides: Record): TestPool { } as any); } +/** Resolves to "still pending" if the promise has not settled within `ms`. */ +function settledWithin( + promise: Promise, + ms: number, +): Promise { + let timer: NodeJS.Timeout; + const pending = new Promise((resolve) => { + timer = setTimeout(() => resolve("still pending"), ms); + }); + return Promise.race([promise, pending]).finally(() => clearTimeout(timer)); +} + describe("BaseSessionPool lifecycle", () => { it("does not leak a released session to a timed-out waiter (no starvation)", async () => { const pool = newPool({ maxPoolSize: 1, waitTimeout: 50 }); @@ -202,4 +214,39 @@ describe("BaseSessionPool lifecycle", () => { await pool.close(); }); + + it("rejects a caller waiting in getSession() as soon as close() starts", async () => { + const pool = newPool({ maxPoolSize: 1, waitTimeout: 10000 }); + const s1: any = await pool.getSession(); // pool is now full (1/1) + const waiting = settledWithin(pool.getSession(), 1000); + let openClose!: () => void; + s1.closeGate = new Promise((resolve) => { + openClose = resolve; + }); + + const closing = pool.close(); // parks on s1.close() + + // On the buggy code the waiter stays pending until waitTimeout and then + // reports "Timeout waiting for available session". + await expect(waiting).rejects.toThrow("SessionPool is closed"); + openClose(); + await closing; + }); + + it("rejects a caller that starts waiting while close() is running", async () => { + const pool = newPool({ maxPoolSize: 1, waitTimeout: 10000 }); + const s1: any = await pool.getSession(); + let openClose!: () => void; + s1.closeGate = new Promise((resolve) => { + openClose = resolve; + }); + + const closing = pool.close(); + // s1 is still in the pool while it closes, so this call has to wait. + const waiting = settledWithin(pool.getSession(), 1000); + openClose(); + await closing; + + await expect(waiting).rejects.toThrow("SessionPool is closed"); + }); });