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,138 @@
+import { randomUUID } from "crypto";
+import { Address, mergeAddresses } from "../common/dsn";
+import logger from "../common/log";
+
+export class Cluster {
+    readonly id: string;
+    private _addresses: readonly Address[];
+
+    constructor(addresses: Address[]) {
+        this.id = randomUUID();
+        this._addresses = Cluster.freezeAddresses(addresses);
+    }
+
+    get addresses(): readonly Address[] {
+        return this._addresses;
+    }
+
+    addAddresses(addresses: Address[]): void {
+        const merged = mergeAddresses([...this._addresses], addresses);
+        this._addresses = Cluster.freezeAddresses(merged);
+    }
+
+    private static freezeAddresses(addresses: Address[]): readonly Address[] {
+        return Object.freeze(
+            addresses.map((address) =>
+                Object.freeze(new Address(address.host, address.port))
+            )
+        );
+    }
+}
+
+export class ClusterRegistry {
+    private static _instance?: ClusterRegistry;
+    private endpointToCluster: Map<string, Cluster> = new Map();
+
+    private constructor() { }
+
+    public static instance(): ClusterRegistry {
+        if (!ClusterRegistry._instance) {
+            ClusterRegistry._instance = new ClusterRegistry();
+        }
+        return ClusterRegistry._instance;
+    }
+
+    private endpointKey(address: Address): string {
+        return `${address.host}:${address.port}`;
+    }
+
+    private collectMatchedClusters(addresses: Address[]): Map<string, Cluster> {
+        const matchedClusters = new Map<string, Cluster>();
+        for (const address of addresses) {
+            const cluster = this.endpointToCluster.get(this.endpointKey(address));
+            if (!cluster) {
+                continue;
+            }
+            matchedClusters.set(cluster.id, cluster);
+        }
+        return matchedClusters;
+    }
+
+    public getOrCreateCluster(seeds: Address[]): Cluster | null {
+        const matchedClusters = this.collectMatchedClusters(seeds);
+        if (matchedClusters.size > 1) {
+            logger.warn(
+                "Adapter HA: seed addresses span multiple known clusters, " +
+                "skipping expansion. Ensure all seeds belong to the same cluster."
+            );
+            return null;
+        }
+        if (matchedClusters.size === 1) {
+            return matchedClusters.values().next().value as Cluster;
+        }
+
+        const cluster = new Cluster(seeds);
+        for (const seed of seeds) {
+            this.endpointToCluster.set(this.endpointKey(seed), cluster);
+        }
+        return cluster;
+    }
+
+    public updateCluster(discovered: Address[]): void {
+        if (discovered.length === 0) {
+            return;
+        }
+
+        const matchedClusters = this.collectMatchedClusters(discovered);
+        if (matchedClusters.size === 0) {
+            return;
+        }
+
+        if (matchedClusters.size > 1) {
+            logger.warn(
+                "Adapter HA: discovered endpoints match multiple known clusters, skipping update."
+            );
+            return;
+        }
+
+        const cluster = matchedClusters.values().next().value as Cluster;
+        cluster.addAddresses(discovered);
+        for (const address of discovered) {
+            this.endpointToCluster.set(this.endpointKey(address), cluster);
+        }
+    }
+
+    public expandEndpoints(seeds: Address[]): Address[] {
+        let matchedCluster: Cluster | null = null;
+
+        for (const seed of seeds) {
+            const cluster = this.endpointToCluster.get(this.endpointKey(seed));
+            if (!cluster) {
+                continue;
+            }
+
+            if (matchedCluster === null) {
+                matchedCluster = cluster;
+                continue;
+            }
+
+            if (matchedCluster.id !== cluster.id) {
+                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.addresses]);
+    }
+}

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;
@@ -179,6 +185,16 @@ export class WebSocketConnector {
         }
         this._poolKey = poolKey;
         this._dsn = dsn;
+        if (dsn.isAdapterHA()) {
+            const expanded = ClusterRegistry.instance().expandEndpoints(this._dsn.addresses);
+            if (expanded.length > this._dsn.addresses.length) {
+                logger.info(
+                    "Adapter HA: expanded seed endpoints from registry, " +
+                    `${this._dsn.addresses.length} -> ${expanded.length}`
+                );
+                this._dsn.addresses = expanded;
+            }
+        }
         this._currentAddress = this.selectLeastConnectedAddress();
         this._retryConfig = RetryConfig.fromDsn(dsn);
         this._inflightStore = new InflightRequestStore();
@@ -592,6 +608,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);
+        ClusterRegistry.instance().updateCluster(discovered);
+        const merged = mergeAddresses(this._dsn.addresses, discovered);
+        if (merged.length <= this._dsn.addresses.length) {
+            return;
+        }
+
+        const newCount = merged.length - this._dsn.addresses.length;
+        this._dsn.addresses = merged;
+        logger.info(
+            `Adapter HA: discovered ${newCount} new endpoint(s), total ${merged.length}`
+        );
+    }
+
     private async recoverSessionContext(): Promise<void> {
         if (!this._sessionRecoveryHook) {
             return;

nodejs/src/client/wsConnectorPool.ts

@@ -4,6 +4,7 @@ import { Dsn } from "../common/dsn";
 import { ErrorCode, TDWebSocketClientError } from "../common/wsError";
 import logger from "../common/log";
 import { w3cwebsocket } from "websocket";
+import { ClusterRegistry } from "./clusterRegistry";
 import { WebSocketConnector } from "./wsConnector";
 
 const mutex = new Mutex();
@@ -46,12 +47,18 @@ export class WebSocketConnectionPool {
     }
 
     private getPoolKey(dsn: Dsn): string {
+        const auth = this.buildAuthScope(dsn);
+        const path = dsn.path();
+        if (dsn.isAdapterHA()) {
+            const cluster = ClusterRegistry.instance().getOrCreateCluster(dsn.addresses);
+            if (cluster) {
+                return `${dsn.scheme}://${cluster.id}/${path}#auth=${auth}`;
+            }
+        }
         const addrs = [...dsn.addresses]
             .sort((a, b) => `${a.host}:${a.port}`.localeCompare(`${b.host}:${b.port}`))
             .map((addr) => `${addr.host}:${addr.port}`)
             .join(",");
-        const auth = this.buildAuthScope(dsn);
-        const path = dsn.path();
         return `${dsn.scheme}://${addrs}/${path}#auth=${auth}`;
     }