feat: supports adapter high availability

.github/workflows/build.yml

@@ -39,11 +39,31 @@ jobs:
           cd TDengine
           mkdir debug
           cd debug
-          cmake .. -DBUILD_HTTP=false -DBUILD_JDBC=false -DBUILD_TOOLS=false -DBUILD_TEST=off -DBUILD_DEPENDENCY_TESTS=false
+          cmake .. -DBUILD_JDBC=false -DBUILD_TOOLS=false -DBUILD_TEST=off -DBUILD_DEPENDENCY_TESTS=false -DBUILD_CONTRIB=ON
           make -j 4
           sudo make install
           which taosd
-          which taosadapter
+
+      - name: Checkout taosadapter
+        uses: actions/checkout@v4
+        with:
+          repository: "taosdata/taosadapter"
+          path: "taosadapter"
+          ref: "main"
+
+      - name: Set up Go
+        uses: actions/setup-go@v6
+        with:
+          go-version: "1.25.10"
+          cache-dependency-path: taosadapter/go.sum
+
+      - name: Build and install taosadapter
+        run: |
+          cd taosadapter
+          go mod download
+          go build -o taosadapter
+          sudo install -m 755 taosadapter /bin/taosadapter
+          taosadapter --version
 
       - name: Start taosd
         run: nohup sudo taosd &

nodejs/package.json

@@ -42,7 +42,7 @@
     "json-bigint": "^1.0.0",
     "moment-timezone": "^0.5.45",
     "typescript": "^5.3.3",
-    "uuid": "^9.0.1",
+    "uuid": "^11.1.1",
     "websocket": "^1.0.34",
     "winston": "^3.17.0",
     "winston-daily-rotate-file": "^5.0.0"

nodejs/src/client/clusterRegistry.ts

@@ -0,0 +1,67 @@
+import { Address, mergeAddresses } from "../common/dsn";
+import logger from "../common/log";
+
+export class ClusterRegistry {
+    private static _instance?: ClusterRegistry;
+    private endpointToCluster: Map<string, readonly Address[]> = new Map();
+
+    private constructor() { }
+
+    public static instance(): ClusterRegistry {
+        if (!ClusterRegistry._instance) {
+            ClusterRegistry._instance = new ClusterRegistry();
+        }
+        return ClusterRegistry._instance;
+    }
+
+    public static _resetForTest(): void {
+        if (ClusterRegistry._instance) {
+            ClusterRegistry._instance.endpointToCluster.clear();
+        }
+        ClusterRegistry._instance = undefined;
+    }
+
+    public registerCluster(addresses: Address[]): void {
+        const snapshot: Address[] = addresses.map(
+            (address) => Object.freeze(new Address(address.host, address.port))
+        );
+        Object.freeze(snapshot);
+        for (const address of snapshot) {
+            this.endpointToCluster.set(`${address.host}:${address.port}`, snapshot);
+        }
+    }
+
+    public expandEndpoints(seeds: Address[]): Address[] {
+        let matchedCluster: readonly Address[] | null = null;
+
+        for (const seed of seeds) {
+            const cluster = this.endpointToCluster.get(`${seed.host}:${seed.port}`);
+            if (!cluster) {
+                continue;
+            }
+
+            if (matchedCluster === null) {
+                matchedCluster = cluster;
+                continue;
+            }
+
+            if (matchedCluster !== cluster) {
+                logger.warn(
+                    "Adapter HA: seed addresses span multiple known clusters, " +
+                    "skipping expansion. Ensure all seeds belong to the same cluster."
+                );
+                return seeds.map((seedAddress) =>
+                    new Address(seedAddress.host, seedAddress.port)
+                );
+            }
+        }
+
+        if (!matchedCluster) {
+            return seeds.map((seedAddress) =>
+                new Address(seedAddress.host, seedAddress.port)
+            );
+        }
+
+        return mergeAddresses(seeds, [...matchedCluster]);
+    }
+}

nodejs/src/client/wsClient.ts

@@ -48,7 +48,10 @@ export class WsClient {
         }
     }
 
-    private buildConnMessage(database?: string | undefined | null) {
+    private buildConnMessage(
+        database?: string | undefined | null,
+        listInstances?: boolean
+    ) {
         return {
             action: "conn",
             args: {
@@ -61,6 +64,7 @@ export class WsClient {
                 ...(this._userApp && { app: this._userApp }),
                 ...(this._userIp && { ip: this._userIp }),
                 ...(this._bearerToken && { bearer_token: this._bearerToken }),
+                ...(listInstances !== undefined && { list_instances: listInstances }),
             },
         };
     }
@@ -140,7 +144,8 @@ export class WsClient {
     }
 
     async connect(database?: string | undefined | null): Promise<void> {
-        const connMsg = this.buildConnMessage(database);
+        const listInstances = this._dsn.isAdapterHA() ? true : undefined;
+        const connMsg = this.buildConnMessage(database, listInstances);
         const normalizedDatabase = this.normalizeConnectedDatabase(database ?? null);
         if (logger.isDebugEnabled()) {
             logger.debug("[wsClient.connect.connMsg]===>" + JSONBig.stringify(connMsg, (key, value) =>
@@ -165,6 +170,9 @@ export class WsClient {
             if (result.msg.code == 0) {
                 this._connectedDatabase = normalizedDatabase;
                 this._wsConnector.markSessionReady();
+                if (result.msg.list_instances) {
+                    this._wsConnector.mergeDiscoveredEndpoints(result.msg.list_instances);
+                }
                 return;
             }
             await this.close();
@@ -320,6 +328,14 @@ export class WsClient {
         return this.getWsConnector().getReconnectRetries();
     }
 
+    isAdapterHA(): boolean {
+        return this._dsn.isAdapterHA();
+    }
+
+    mergeDiscoveredEndpoints(instances: string[]): void {
+        this.getWsConnector().mergeDiscoveredEndpoints(instances);
+    }
+
     async sendMsg(message: string): Promise<any> {
         logger.debug("[wsClient.sendMsg]===>" + message);
         return this.getWsConnector().sendMsg(message);

nodejs/src/client/wsConnector.ts

@@ -1,5 +1,10 @@
 import { ICloseEvent, w3cwebsocket } from "websocket";
-import { Address, Dsn } from "../common/dsn";
+import {
+    Address,
+    Dsn,
+    mergeAddresses,
+    parseDiscoveredEndpoints
+} from "../common/dsn";
 import { AddressConnectionTracker } from "../common/addressConnectionTracker";
 import {
     ErrorCode,
@@ -12,6 +17,7 @@ import {
     maskSensitiveForLog,
     maskUrlForLog
 } from "../common/utils";
+import { ClusterRegistry } from "./clusterRegistry";
 
 interface InflightRequest {
     reqId: bigint;
@@ -154,6 +160,7 @@ export class WebSocketConnector {
     private _conn!: w3cwebsocket;
     private readonly _poolKey: string;
     private readonly _dsn: Dsn;
+    private _failoverAddresses: Address[];
     private _currentAddress: Address;
     private _retryConfig: RetryConfig;
     private _inflightStore: InflightRequestStore;
@@ -179,6 +186,19 @@ export class WebSocketConnector {
         }
         this._poolKey = poolKey;
         this._dsn = dsn;
+        this._failoverAddresses = dsn.addresses.map(
+            (address) => new Address(address.host, address.port)
+        );
+        if (dsn.isAdapterHA()) {
+            const expanded = ClusterRegistry.instance().expandEndpoints(this._failoverAddresses);
+            if (expanded.length > this._failoverAddresses.length) {
+                logger.info(
+                    "Adapter HA: expanded seed endpoints from registry, " +
+                    `${this._failoverAddresses.length} -> ${expanded.length}`
+                );
+                this._failoverAddresses = expanded;
+            }
+        }
         this._currentAddress = this.selectLeastConnectedAddress();
         this._retryConfig = RetryConfig.fromDsn(dsn);
         this._inflightStore = new InflightRequestStore();
@@ -431,11 +451,11 @@ export class WebSocketConnector {
     }
 
     private selectLeastConnectedAddress(excludedAddresses: Set<string> = new Set()): Address {
-        const candidates = this._dsn.addresses.filter((address) => {
+        const candidates = this._failoverAddresses.filter((address) => {
             const addressKey = `${address.host}:${address.port}`;
             return !excludedAddresses.has(addressKey);
         });
-        const selectableAddresses = candidates.length > 0 ? candidates : this._dsn.addresses;
+        const selectableAddresses = candidates.length > 0 ? candidates : this._failoverAddresses;
         const selectedIndex = AddressConnectionTracker.instance()
             .selectLeastConnected(selectableAddresses);
         return selectableAddresses[selectedIndex];
@@ -496,7 +516,7 @@ export class WebSocketConnector {
     }
 
     private async attemptReconnect(): Promise<void> {
-        const totalAddresses = this._dsn.addresses.length;
+        const totalAddresses = this._failoverAddresses.length;
         const failedAddresses = new Set<string>();
 
         for (let i = 0; i < totalAddresses; i++) {
@@ -592,6 +612,24 @@ export class WebSocketConnector {
         this._sessionReady = true;
     }
 
+    public mergeDiscoveredEndpoints(instances: string[]): void {
+        if (!this._dsn.isAdapterHA() || !instances || instances.length === 0) {
+            return;
+        }
+        const discovered = parseDiscoveredEndpoints(instances);
+        const merged = mergeAddresses(this._failoverAddresses, discovered);
+        if (merged.length <= this._failoverAddresses.length) {
+            return;
+        }
+
+        const newCount = merged.length - this._failoverAddresses.length;
+        this._failoverAddresses = merged;
+        ClusterRegistry.instance().registerCluster(merged);
+        logger.info(
+            `Adapter HA: discovered ${newCount} new endpoint(s), total ${merged.length}`
+        );
+    }
+
     private async recoverSessionContext(): Promise<void> {
         if (!this._sessionRecoveryHook) {
             return;