feat: enhance reconnect logic and add tests for aborting retries on close

Author: qevolg

nodejs/src/client/wsConnector.ts

@@ -295,7 +295,7 @@ export class WebSocketConnector {
     }
 
     private shouldSkipReconnect(conn: w3cwebsocket): boolean {
-        if (!this._allowReconnect) {
+        if (!this.isReconnectAllowed()) {
             return true;
         }
         return this._suppressedSockets.has(conn);
@@ -439,7 +439,14 @@ export class WebSocketConnector {
         await new Promise((resolve) => setTimeout(resolve, ms));
     }
 
+    private isReconnectAllowed(): boolean {
+        return this._allowReconnect;
+    }
+
     private async triggerReconnect(): Promise<void> {
+        if (!this.isReconnectAllowed()) {
+            return;
+        }
         if (!this._reconnectLock) {
             this._reconnectLock = this._doReconnect();
         }
@@ -455,6 +462,9 @@ export class WebSocketConnector {
     }
 
     private async _doReconnect(): Promise<void> {
+        if (!this.isReconnectAllowed()) {
+            return;
+        }
         this._isReconnecting = true;
         try {
             await this.attemptReconnect();
@@ -485,6 +495,9 @@ export class WebSocketConnector {
 
         for (let i = 0; i < totalAddresses; i++) {
             for (let retry = 0; retry < this._retryConfig.retries; retry++) {
+                if (!this.isReconnectAllowed()) {
+                    return;
+                }
                 try {
                     logger.info(`Reconnecting to ${this.getCurrentAddress()}, attempt ${retry + 1}`);
                     await this.reconnect();
@@ -539,16 +552,19 @@ export class WebSocketConnector {
     }
 
     close() {
+        this._allowReconnect = false;
+        this.failAllInflightRequests(
+            new TDWebSocketClientError(
+                ErrorCode.ERR_CONNECTION_CLOSED,
+                "websocket connection closed"
+            )
+        );
         if (this._conn) {
-            this._allowReconnect = false;
             AddressConnectionTracker.instance().decrement(this.getCurrentAddress());
             this._suppressedSockets.add(this._conn);
             this._conn.close();
         } else {
-            throw new TDWebSocketClientError(
-                ErrorCode.ERR_WEBSOCKET_CONNECTION_FAIL,
-                "WebSocket connection is undefined"
-            );
+            logger.warn("close() called but websocket connection is undefined");
         }
     }
 

nodejs/test/client/wsConnector.failover.test.ts

@@ -344,6 +344,49 @@ describe("WebSocketConnector failover and retry", () => {
         expect(connector._suppressedSockets.has(connector._conn)).toBe(true);
     });
 
+    test("attemptReconnect aborts silently when close is called mid-retry", async () => {
+        const connector = createBareConnector("ws://root:taosdata@host1:6041");
+        connector._retryConfig = new RetryConfig(3, 1, 8);
+        connector.reconnect = jest.fn(async () => {
+            throw new Error("host down");
+        });
+        connector.sleep = jest.fn(async () => {
+            connector.close();
+        });
+
+        await expect(connector.attemptReconnect()).resolves.toBeUndefined();
+        expect(connector.reconnect).toHaveBeenCalledTimes(1);
+    });
+
+    test("close rejects retriable inflight request immediately", async () => {
+        const connector = createBareConnector();
+        connector.triggerReconnect = jest.fn(() => new Promise<void>(() => { }));
+        connector._conn.send = jest.fn(() => {
+            throw new Error("cannot call send() while not connected");
+        });
+
+        const pending = connector.sendMsg(
+            JSON.stringify({
+                action: "insert",
+                args: {
+                    req_id: 401,
+                },
+            })
+        );
+        void pending.catch(() => { });
+
+        await delay(20);
+        connector.close();
+
+        const state = await Promise.race([
+            pending.then(() => "resolved").catch(() => "rejected"),
+            delay(40).then(() => "pending"),
+        ]);
+
+        expect(state).toBe("rejected");
+        expect(hasInflightRequest(connector, 401n)).toBe(false);
+    });
+
     test("reconnect decrements current address before closing old socket", async () => {
         const connector = createBareConnector();
         const decrementSpy = jest