diff --git a/.github/grype-sbom.yaml b/.github/grype-sbom.yaml new file mode 100644 index 00000000..e8b94527 --- /dev/null +++ b/.github/grype-sbom.yaml @@ -0,0 +1,5 @@ +ignore: + - vulnerability: GHSA-w5hq-g745-h8pq + package: + name: uuid + type: npm diff --git a/.github/workflows/audit.yml b/.github/workflows/audit.yml index b2426195..c66fcae7 100644 --- a/.github/workflows/audit.yml +++ b/.github/workflows/audit.yml @@ -20,7 +20,7 @@ jobs: - name: Audit check working-directory: nodejs - run: npx audit-ci --moderate --allowlist GHSA-f886-m6hf-6m8v + run: npx audit-ci --moderate --allowlist GHSA-f886-m6hf-6m8v --allowlist GHSA-w5hq-g745-h8pq - name: License check working-directory: nodejs diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index c2dec3ce..7b4fea87 100644 --- a/.github/workflows/build.yml +++ b/.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 & diff --git a/.github/workflows/sbom.yml b/.github/workflows/sbom.yml index 5d03e40c..715bf51a 100644 --- a/.github/workflows/sbom.yml +++ b/.github/workflows/sbom.yml @@ -30,4 +30,5 @@ jobs: with: sbom: "${{ env.SBOM_FILENAME }}" cache-db: true + config: ".github/grype-sbom.yaml" output-format: "table" diff --git a/nodejs/package-lock.json b/nodejs/package-lock.json index 9987b204..1bc1cd18 100644 --- a/nodejs/package-lock.json +++ b/nodejs/package-lock.json @@ -1,12 +1,12 @@ { "name": "@tdengine/websocket", - "version": "3.4.0", + "version": "3.5.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@tdengine/websocket", - "version": "3.4.0", + "version": "3.5.0", "license": "MIT", "dependencies": { "async-mutex": "^0.5.0", diff --git a/nodejs/package.json b/nodejs/package.json index e3758555..63434b00 100644 --- a/nodejs/package.json +++ b/nodejs/package.json @@ -1,6 +1,6 @@ { "name": "@tdengine/websocket", - "version": "3.4.0", + "version": "3.5.0", "description": "The websocket Node.js connector for TDengine. TDengine versions 3.3.2.0 and above are recommended to use this connector.", "source": "index.ts", "main": "lib/index.js", diff --git a/nodejs/src/client/clusterRegistry.ts b/nodejs/src/client/clusterRegistry.ts new file mode 100644 index 00000000..0549c3cb --- /dev/null +++ b/nodejs/src/client/clusterRegistry.ts @@ -0,0 +1,138 @@ +import { v4 as uuidv4 } from "uuid"; +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 = uuidv4(); + 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 = 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 { + const matchedClusters = new Map(); + 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]); + } +} diff --git a/nodejs/src/client/wsClient.ts b/nodejs/src/client/wsClient.ts index dde09ba1..7aa260e4 100644 --- a/nodejs/src/client/wsClient.ts +++ b/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 { - 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 { logger.debug("[wsClient.sendMsg]===>" + message); return this.getWsConnector().sendMsg(message); diff --git a/nodejs/src/client/wsConnector.ts b/nodejs/src/client/wsConnector.ts index a1b8b40e..961ab97c 100644 --- a/nodejs/src/client/wsConnector.ts +++ b/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 { if (!this._sessionRecoveryHook) { return; diff --git a/nodejs/src/client/wsConnectorPool.ts b/nodejs/src/client/wsConnectorPool.ts index 64dd54df..61e3d991 100644 --- a/nodejs/src/client/wsConnectorPool.ts +++ b/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}`; } diff --git a/nodejs/src/common/dsn.ts b/nodejs/src/common/dsn.ts index 31f49230..da06663c 100644 --- a/nodejs/src/common/dsn.ts +++ b/nodejs/src/common/dsn.ts @@ -1,4 +1,5 @@ import { ErrorCode, TDWebSocketClientError } from "./wsError"; +import logger from "./log"; export type WebSocketEndpoint = "sql" | "tmq"; export const WS_SQL_ENDPOINT: WebSocketEndpoint = "sql"; @@ -77,6 +78,10 @@ export class Dsn { path(): string { return this.endpoint === WS_TMQ_ENDPOINT ? WS_TMQ_PATH : WS_SQL_PATH; } + + isAdapterHA(): boolean { + return this.params.get("adapter_ha") === "true"; + } } /** @@ -200,6 +205,106 @@ export function getDefaultPortForHost(host: string): number { return isCloudServiceHost(host) ? CLOUD_DEFAULT_PORT : DEFAULT_PORT; } +function parsePortValue(portStr: string): number | null { + if (portStr.length === 0) { + return null; + } + if (!/^\d+$/.test(portStr)) { + return null; + } + const port = Number.parseInt(portStr, 10); + if (Number.isNaN(port) || port < 1 || port > 65535) { + return null; + } + return port; +} + +function warnInvalidEndpoint(raw: string): void { + logger.warn(`Adapter HA: ignoring invalid endpoint: ${raw}`); +} + +export function parseDiscoveredEndpoints(instances: string[]): Address[] { + const endpoints: Address[] = []; + for (const raw of instances) { + const trimmed = raw.trim(); + if (trimmed.length === 0) { + continue; + } + + if (trimmed.startsWith("[")) { + const closeBracket = trimmed.indexOf("]"); + if (closeBracket === -1) { + warnInvalidEndpoint(raw); + continue; + } + + const ipv6Host = trimmed.slice(1, closeBracket); + if (ipv6Host.length === 0) { + warnInvalidEndpoint(raw); + continue; + } + + const remainder = trimmed.slice(closeBracket + 1); + if (remainder.length > 0 && !remainder.startsWith(":")) { + warnInvalidEndpoint(raw); + continue; + } + + const portStr = remainder.startsWith(":") ? remainder.slice(1) : ""; + const parsedPort = parsePortValue(portStr); + if (parsedPort === null) { + warnInvalidEndpoint(raw); + continue; + } + endpoints.push( + new Address(`[${ipv6Host}]`, parsedPort) + ); + continue; + } + + const lastColon = trimmed.lastIndexOf(":"); + if (lastColon === -1) { + warnInvalidEndpoint(raw); + continue; + } + + const host = trimmed.slice(0, lastColon); + if (host.length === 0 || host.includes(":")) { + warnInvalidEndpoint(raw); + continue; + } + + const portStr = trimmed.slice(lastColon + 1); + const parsedPort = parsePortValue(portStr); + if (parsedPort === null) { + warnInvalidEndpoint(raw); + continue; + } + + endpoints.push(new Address(host, parsedPort)); + } + return endpoints; +} + +export function mergeAddresses(existing: Address[], discovered: Address[]): Address[] { + const seen = new Set(); + const merged: Address[] = []; + const appendUnique = (source: Address[]) => { + for (const address of source) { + const key = `${address.host}:${address.port}`; + if (seen.has(key)) { + continue; + } + seen.add(key); + merged.push(new Address(address.host, address.port)); + } + }; + + appendUnique(existing); + appendUnique(discovered); + return merged; +} + /** * Parse comma-separated host list. Supports IPv6 in brackets. * Examples: "host1:6041,host2:6042", "[::1]:6041,host2:6042" diff --git a/nodejs/src/tmq/wsTmq.ts b/nodejs/src/tmq/wsTmq.ts index 327820c8..48bf812c 100644 --- a/nodejs/src/tmq/wsTmq.ts +++ b/nodejs/src/tmq/wsTmq.ts @@ -90,7 +90,11 @@ export class WsConsumer { }); } - private buildSubscribeMessage(topics: Array, reqId?: number) { + private buildSubscribeMessage( + topics: Array, + reqId?: number, + listInstances?: boolean + ) { const user = this._config.user === null ? null : safeDecodeURIComponent(this._config.user); @@ -114,6 +118,7 @@ export class WsConsumer { ...(this._config.userApp && { app: this._config.userApp }), ...(this._config.userIp && { ip: this._config.userIp }), connector: ConnectorInfo, + ...(listInstances !== undefined && { list_instances: listInstances }), }, }; } @@ -139,8 +144,12 @@ export class WsConsumer { ); } - let queryMsg = this.buildSubscribeMessage(topics, reqId); - await this._wsClient.exec(JSON.stringify(queryMsg)); + const listInstances = this._wsClient.isAdapterHA() ? true : undefined; + let queryMsg = this.buildSubscribeMessage(topics, reqId, listInstances); + const result = await this._wsClient.exec(JSON.stringify(queryMsg), false); + if (result?.msg?.list_instances) { + this._wsClient.mergeDiscoveredEndpoints(result.msg.list_instances); + } this._topics = [...topics]; } diff --git a/nodejs/test/client/clusterRegistry.test.ts b/nodejs/test/client/clusterRegistry.test.ts new file mode 100644 index 00000000..d4a6af68 --- /dev/null +++ b/nodejs/test/client/clusterRegistry.test.ts @@ -0,0 +1,218 @@ +import { Cluster, ClusterRegistry } from "@src/client/clusterRegistry"; +import { Address } from "@src/common/dsn"; +import logger from "@src/common/log"; + +function resetClusterRegistrySingleton(): void { + (ClusterRegistry as any)._instance = undefined; +} + +describe("Cluster", () => { + test("stores frozen snapshot that is isolated from input mutation", () => { + const input = [new Address("host1", 6041), new Address("host2", 6042)]; + const cluster = new Cluster(input); + + input[0].host = "changed"; + input.push(new Address("host3", 6043)); + + expect(Object.isFrozen(cluster.addresses)).toBe(true); + expect(Object.isFrozen(cluster.addresses[0])).toBe(true); + expect(cluster.addresses).toHaveLength(2); + expect(cluster.addresses[0]).toEqual({ host: "host1", port: 6041 }); + expect(cluster.addresses[0]).not.toBe(input[0]); + }); + + test("addAddresses merges unique addresses and keeps cluster id stable", () => { + const cluster = new Cluster([new Address("host1", 6041)]); + const initialId = cluster.id; + + cluster.addAddresses([ + new Address("host1", 6041), + new Address("host2", 6042), + ]); + + expect(cluster.id).toBe(initialId); + expect(cluster.addresses).toEqual([ + { host: "host1", port: 6041 }, + { host: "host2", port: 6042 }, + ]); + }); +}); + +describe("ClusterRegistry", () => { + beforeEach(() => { + resetClusterRegistrySingleton(); + }); + + afterEach(() => { + jest.restoreAllMocks(); + resetClusterRegistrySingleton(); + }); + + test("getOrCreateCluster returns the same cluster for known seed address", () => { + const registry = ClusterRegistry.instance(); + const first = registry.getOrCreateCluster([new Address("host1", 6041)]); + const second = registry.getOrCreateCluster([new Address("host1", 6041)]); + expect(first).not.toBeNull(); + expect(second).not.toBeNull(); + expect(second?.id).toBe(first?.id); + }); + + test("getOrCreateCluster returns known cluster without mapping unmatched seeds", () => { + const registry = ClusterRegistry.instance(); + const cluster = registry.getOrCreateCluster([new Address("host1", 6041)]); + const resolved = registry.getOrCreateCluster([ + new Address("host1", 6041), + new Address("host2", 6042), + ]); + expect(cluster).not.toBeNull(); + expect(resolved).not.toBeNull(); + expect(resolved?.id).toBe(cluster?.id); + expect((registry as any).endpointToCluster.get("host2:6042")).toBeUndefined(); + }); + + test("getOrCreateCluster creates a new cluster for unknown seeds", () => { + const registry = ClusterRegistry.instance(); + const cluster = registry.getOrCreateCluster([ + new Address("seed1", 6041), + new Address("seed2", 6042), + ]); + expect(cluster).not.toBeNull(); + expect(cluster?.addresses).toEqual([ + { host: "seed1", port: 6041 }, + { host: "seed2", port: 6042 }, + ]); + expect((registry as any).endpointToCluster.get("seed1:6041")?.id).toBe(cluster?.id); + expect((registry as any).endpointToCluster.get("seed2:6042")?.id).toBe(cluster?.id); + }); + + test("getOrCreateCluster returns null when seeds span multiple known clusters", () => { + const registry = ClusterRegistry.instance(); + const warnSpy = jest.spyOn(logger, "warn").mockImplementation(() => logger); + + registry.getOrCreateCluster([new Address("a1", 6041)]); + registry.getOrCreateCluster([new Address("b1", 6041)]); + + const cluster = registry.getOrCreateCluster([ + new Address("a1", 6041), + new Address("b1", 6041), + ]); + + expect(cluster).toBeNull(); + expect(warnSpy).toHaveBeenCalledWith( + "Adapter HA: seed addresses span multiple known clusters, skipping expansion. Ensure all seeds belong to the same cluster." + ); + }); + + test("updateCluster merges discovered endpoints into one matched cluster", () => { + const registry = ClusterRegistry.instance(); + const existing = registry.getOrCreateCluster([new Address("host1", 6041)]); + expect(existing).not.toBeNull(); + + registry.updateCluster([ + new Address("host1", 6041), + new Address("host2", 6042), + new Address("host3", 6043), + ]); + + const expanded = registry.expandEndpoints([new Address("host1", 6041)]); + expect(expanded).toEqual([ + { host: "host1", port: 6041 }, + { host: "host2", port: 6042 }, + { host: "host3", port: 6043 }, + ]); + expect((registry as any).endpointToCluster.get("host2:6042")?.id).toBe(existing?.id); + expect((registry as any).endpointToCluster.get("host3:6043")?.id).toBe(existing?.id); + }); + + test("updateCluster ignores discovered endpoints when no cluster matches", () => { + const registry = ClusterRegistry.instance(); + registry.updateCluster([ + new Address("host9", 6049), + new Address("host10", 6050), + ]); + expect((registry as any).endpointToCluster.size).toBe(0); + }); + + test("updateCluster aborts when discovered endpoints hit multiple clusters", () => { + const registry = ClusterRegistry.instance(); + const warnSpy = jest.spyOn(logger, "warn").mockImplementation(() => logger); + + registry.getOrCreateCluster([new Address("a1", 6041)]); + registry.getOrCreateCluster([new Address("b1", 6042)]); + + registry.updateCluster([ + new Address("a1", 6041), + new Address("b1", 6042), + new Address("c1", 6043), + ]); + + expect((registry as any).endpointToCluster.get("c1:6043")).toBeUndefined(); + expect(warnSpy).toHaveBeenCalledWith( + "Adapter HA: discovered endpoints match multiple known clusters, skipping update." + ); + }); + + test("expandEndpoints returns merged endpoints when seed matches known cluster", () => { + const registry = ClusterRegistry.instance(); + registry.getOrCreateCluster([new Address("host1", 6041)]); + registry.updateCluster([ + new Address("host1", 6041), + new Address("host2", 6042), + new Address("host3", 6043), + ]); + + const seeds = [new Address("host1", 6041)]; + const expanded = registry.expandEndpoints(seeds); + + expect(expanded).toEqual([ + { host: "host1", port: 6041 }, + { host: "host2", port: 6042 }, + { host: "host3", port: 6043 }, + ]); + expect(expanded).not.toBe(seeds); + expect(expanded[0]).not.toBe(seeds[0]); + }); + + test("expandEndpoints returns seed deep copy when no known cluster matches", () => { + const registry = ClusterRegistry.instance(); + const seeds = [new Address("seed1", 6041), new Address("seed2", 6042)]; + + const expanded = registry.expandEndpoints(seeds); + + expect(expanded).toEqual([ + { host: "seed1", port: 6041 }, + { host: "seed2", port: 6042 }, + ]); + expect(expanded).not.toBe(seeds); + expect(expanded[0]).not.toBe(seeds[0]); + + expanded[0].host = "changed"; + expect(seeds[0].host).toBe("seed1"); + }); + + test("expandEndpoints uses cluster id comparison instead of reference equality", () => { + const registry = ClusterRegistry.instance(); + const warnSpy = jest.spyOn(logger, "warn").mockImplementation(() => logger); + + const clusterA = new Cluster([ + new Address("a1", 6041), + new Address("a3", 6043), + ]); + const clusterB = new Cluster([new Address("b1", 6042)]); + (clusterB as any).id = clusterA.id; + + (registry as any).endpointToCluster.set("a1:6041", clusterA); + (registry as any).endpointToCluster.set("a3:6043", clusterA); + (registry as any).endpointToCluster.set("b1:6042", clusterB); + + const seeds = [new Address("a1", 6041), new Address("b1", 6042)]; + const expanded = registry.expandEndpoints(seeds); + + expect(expanded).toEqual([ + { host: "a1", port: 6041 }, + { host: "b1", port: 6042 }, + { host: "a3", port: 6043 }, + ]); + expect(warnSpy).not.toHaveBeenCalled(); + }); +}); diff --git a/nodejs/test/client/wsClient.recovery.test.ts b/nodejs/test/client/wsClient.recovery.test.ts index 49427b73..ec76cd82 100644 --- a/nodejs/test/client/wsClient.recovery.test.ts +++ b/nodejs/test/client/wsClient.recovery.test.ts @@ -272,4 +272,96 @@ describe("WsClient recovery hook", () => { await secondBorrower.close(); }); + + test("connect sends list_instances=true when adapter_ha is enabled", async () => { + const dsn = parse("ws://root:taosdata@localhost:6041?adapter_ha=true"); + const connector = createMockConnector(); + connector.readyState.mockReturnValue(0); + jest + .spyOn(WebSocketConnectionPool.instance(), "getConnection") + .mockResolvedValue(connector as any); + + const client = new WsClient(dsn, 5000); + await client.connect("db_ha"); + + expect(connector.sendMsg).toHaveBeenCalledTimes(1); + const connMsg = JSON.parse((connector.sendMsg.mock.calls[0] as any[])[0]); + expect(connMsg.action).toBe("conn"); + expect(connMsg.args.db).toBe("db_ha"); + expect(connMsg.args.list_instances).toBe(true); + }); + + test("connect does not include list_instances when adapter_ha is disabled", async () => { + const dsn = parse("ws://root:taosdata@localhost:6041"); + const connector = createMockConnector(); + connector.readyState.mockReturnValue(0); + jest + .spyOn(WebSocketConnectionPool.instance(), "getConnection") + .mockResolvedValue(connector as any); + + const client = new WsClient(dsn, 5000); + await client.connect("db_plain"); + + const connMsg = JSON.parse((connector.sendMsg.mock.calls[0] as any[])[0]); + expect(connMsg.args.list_instances).toBeUndefined(); + }); + + test("connect merges discovered endpoints from conn response", async () => { + const dsn = parse("ws://root:taosdata@localhost:6041?adapter_ha=true"); + const connector = createMockConnector(); + connector.readyState.mockReturnValue(0); + (connector as any).mergeDiscoveredEndpoints = jest.fn(); + connector.sendMsg.mockResolvedValue({ + msg: { + code: 0, + message: "", + list_instances: ["host2:6042", "host3:6043"], + } + } as any); + jest + .spyOn(WebSocketConnectionPool.instance(), "getConnection") + .mockResolvedValue(connector as any); + + const client = new WsClient(dsn, 5000); + await client.connect("db_ha"); + + expect((connector as any).mergeDiscoveredEndpoints).toHaveBeenCalledWith([ + "host2:6042", + "host3:6043", + ]); + }); + + test("sql recovery does not include list_instances during reconnect restore", async () => { + const dsn = parse("ws://root:taosdata@localhost:6041?adapter_ha=true"); + const connector = createMockConnector(false); + jest + .spyOn(WebSocketConnectionPool.instance(), "getConnection") + .mockResolvedValue(connector as any); + + const client = new WsClient(dsn, 5000); + await client.connect("db_recovery"); + + expect(connector.sendMsgDirect).toHaveBeenCalledTimes(1); + const connMsg = JSON.parse((connector.sendMsgDirect.mock.calls[0] as [string])[0]); + expect(connMsg.action).toBe("conn"); + expect(connMsg.args.db).toBe("db_recovery"); + expect(connMsg.args.list_instances).toBeUndefined(); + }); + + test("isAdapterHA proxies to dsn adapter_ha config", () => { + expect(new WsClient(parse("ws://root:taosdata@localhost:6041?adapter_ha=true"), 5000).isAdapterHA()) + .toBe(true); + expect(new WsClient(parse("ws://root:taosdata@localhost:6041"), 5000).isAdapterHA()) + .toBe(false); + }); + + test("mergeDiscoveredEndpoints proxies to current connector", () => { + const client = new WsClient(parse("ws://root:taosdata@localhost:6041"), 5000); + const connector = { mergeDiscoveredEndpoints: jest.fn() }; + (client as any)._wsConnector = connector; + + client.mergeDiscoveredEndpoints(["host2:6042"]); + + expect(connector.mergeDiscoveredEndpoints).toHaveBeenCalledWith(["host2:6042"]); + }); }); diff --git a/nodejs/test/client/wsConnector.adapterHa.test.ts b/nodejs/test/client/wsConnector.adapterHa.test.ts new file mode 100644 index 00000000..e01e1782 --- /dev/null +++ b/nodejs/test/client/wsConnector.adapterHa.test.ts @@ -0,0 +1,147 @@ +import { ClusterRegistry } from "@src/client/clusterRegistry"; +import { RetryConfig, WebSocketConnector } from "@src/client/wsConnector"; +import { AddressConnectionTracker } from "@src/common/addressConnectionTracker"; +import { Address, parse } from "@src/common/dsn"; + +function createInflightStore(): any { + return { + insert: jest.fn(), + remove: jest.fn(), + getRequests: jest.fn(() => []), + clear: jest.fn(), + }; +} + +function createBareAdapterHaConnector(seedDsn: string): any { + const connector = Object.create(WebSocketConnector.prototype) as any; + connector._poolKey = "ws://pool/key"; + connector._dsn = parse(seedDsn); + connector._currentAddress = connector._dsn.addresses[0]; + connector._retryConfig = new RetryConfig(1, 1, 8); + connector._reconnectLock = null; + connector._isReconnecting = false; + connector._allowReconnect = true; + connector._connectionReady = Promise.resolve(); + connector._suppressedSockets = new WeakSet(); + connector._sessionRecoveryHook = null; + connector._inflightStore = createInflightStore(); + connector._conn = { + readyState: 1, + send: jest.fn(), + close: jest.fn(), + }; + return connector; +} + +function resetClusterRegistrySingleton(): void { + (ClusterRegistry as any)._instance = undefined; +} + +describe("WebSocketConnector adapter ha", () => { + beforeEach(() => { + resetClusterRegistrySingleton(); + }); + + afterEach(() => { + jest.restoreAllMocks(); + resetClusterRegistrySingleton(); + }); + + test("constructor expands dsn addresses from registry when adapter_ha=true", () => { + const createConnectionSpy = jest + .spyOn(WebSocketConnector.prototype as any, "createConnection") + .mockImplementation(() => { }); + + const registry = ClusterRegistry.instance(); + registry.getOrCreateCluster([new Address("host1", 6041)]); + registry.updateCluster([ + new Address("host1", 6041), + new Address("host2", 6042), + ]); + + const connector = new WebSocketConnector( + parse("ws://root:taosdata@host1:6041?adapter_ha=true"), + "pool-key", + null + ) as any; + + expect(connector._dsn.addresses).toEqual([ + { host: "host1", port: 6041 }, + { host: "host2", port: 6042 }, + ]); + expect(createConnectionSpy).toHaveBeenCalledTimes(1); + }); + + test("mergeDiscoveredEndpoints extends dsn addresses and updates cluster registry", () => { + const connector = createBareAdapterHaConnector( + "ws://root:taosdata@host1:6041?adapter_ha=true" + ); + const updateSpy = jest.spyOn(ClusterRegistry.instance(), "updateCluster"); + + connector.mergeDiscoveredEndpoints([ + "host2:6042", + "host3:6043", + ]); + + expect(updateSpy).toHaveBeenCalledWith([ + { host: "host2", port: 6042 }, + { host: "host3", port: 6043 }, + ]); + expect(connector._dsn.addresses).toEqual([ + { host: "host1", port: 6041 }, + { host: "host2", port: 6042 }, + { host: "host3", port: 6043 }, + ]); + }); + + test("mergeDiscoveredEndpoints keeps pool key unchanged", () => { + const connector = createBareAdapterHaConnector( + "ws://root:taosdata@host1:6041?adapter_ha=true" + ); + const initialPoolKey = connector.getPoolKey(); + + connector.mergeDiscoveredEndpoints(["host2:6042"]); + + expect(connector.getPoolKey()).toBe(initialPoolKey); + }); + + test("selectLeastConnectedAddress selects from dsn addresses", () => { + const connector = createBareAdapterHaConnector( + "ws://root:taosdata@host1:6041?adapter_ha=true" + ); + connector._dsn.addresses = [ + new Address("host1", 6041), + new Address("host2", 6042), + ]; + + const selectSpy = jest + .spyOn(AddressConnectionTracker.instance(), "selectLeastConnected") + .mockReturnValue(1); + + const selected = connector.selectLeastConnectedAddress(); + + expect(selectSpy).toHaveBeenCalledWith(connector._dsn.addresses); + expect(selected).toEqual({ host: "host2", port: 6042 }); + }); + + test("attemptReconnect iterates all dsn addresses", async () => { + const connector = createBareAdapterHaConnector( + "ws://root:taosdata@host1:6041?adapter_ha=true" + ); + connector._retryConfig = new RetryConfig(1, 1, 8); + connector._dsn.addresses = [ + new Address("host1", 6041), + new Address("host2", 6042), + ]; + connector.sleep = jest.fn(async () => { }); + connector.reconnect = jest.fn(async () => { + throw new Error("down"); + }); + connector.selectLeastConnectedAddress = jest.fn(() => connector._dsn.addresses[1]); + + await expect(connector.attemptReconnect()).rejects.toThrow( + "Failed to reconnect to any available address" + ); + expect(connector.reconnect).toHaveBeenCalledTimes(2); + }); +}); diff --git a/nodejs/test/client/wsConnectorPool.key.test.ts b/nodejs/test/client/wsConnectorPool.key.test.ts index 91f8da8b..fb086124 100644 --- a/nodejs/test/client/wsConnectorPool.key.test.ts +++ b/nodejs/test/client/wsConnectorPool.key.test.ts @@ -1,7 +1,8 @@ +import { ClusterRegistry } from "@src/client/clusterRegistry"; import { WebSocketConnectionPool } from "@src/client/wsConnectorPool"; import { RetryConfig } from "@src/client/wsConnector"; import { WSConfig } from "@src/common/config"; -import { parse, WS_TMQ_ENDPOINT } from "@src/common/dsn"; +import { Address, parse, WS_TMQ_ENDPOINT } from "@src/common/dsn"; import { WsSql } from "@src/sql/wsSql"; import { testPassword, testUsername } from "@test-helpers/utils"; import { w3cwebsocket } from "websocket"; @@ -14,13 +15,20 @@ function resetPoolSingleton() { } } +function resetClusterRegistrySingleton(): void { + (ClusterRegistry as any)._instance = undefined; +} + describe("WebSocketConnectionPool key generation", () => { beforeEach(() => { resetPoolSingleton(); + resetClusterRegistrySingleton(); }); afterEach(() => { + jest.restoreAllMocks(); resetPoolSingleton(); + resetClusterRegistrySingleton(); }); test("normalizes address order when generating pool key", () => { @@ -123,6 +131,53 @@ describe("WebSocketConnectionPool key generation", () => { const tmqKey = (pool as any).getPoolKey(tmqDsn); expect(sqlKey).not.toBe(tmqKey); }); + + test("uses cluster uuid for adapter_ha pool key", () => { + const pool = WebSocketConnectionPool.instance(); + const dsnA = parse("ws://root:taosdata@ha1:6041/mydb?adapter_ha=true"); + const dsnB = parse("ws://root:taosdata@ha1:6041,ha2:6042/mydb?adapter_ha=true"); + + const keyA = (pool as any).getPoolKey(dsnA); + const keyB = (pool as any).getPoolKey(dsnB); + + expect(keyA).toBe(keyB); + expect(keyA).toMatch(/^ws:\/\/[0-9a-f-]{36}\/ws#auth=/); + }); + + test("avoids address sorting on adapter_ha fast path when cluster is resolved", () => { + const pool = WebSocketConnectionPool.instance(); + const dsn = parse("ws://root:taosdata@ha-fast:6041/mydb?adapter_ha=true"); + ClusterRegistry.instance().getOrCreateCluster(dsn.addresses); + + const sortSpy = jest + .spyOn(Array.prototype, "sort") + .mockImplementation(() => { + throw new Error("sort should not be called on HA fast path"); + }); + + const poolKey = (pool as any).getPoolKey(dsn); + expect(poolKey).toMatch(/^ws:\/\/[0-9a-f-]{36}\/ws#auth=/); + expect(sortSpy).not.toHaveBeenCalled(); + }); + + test("falls back to address-based key when adapter_ha seeds span multiple clusters", () => { + const pool = WebSocketConnectionPool.instance(); + const registry = ClusterRegistry.instance(); + registry.getOrCreateCluster([new Address("ha-a", 6041)]); + registry.getOrCreateCluster([new Address("ha-b", 6042)]); + + const dsn = parse("ws://root:taosdata@ha-a:6041,ha-b:6042/mydb?adapter_ha=true"); + const key = (pool as any).getPoolKey(dsn); + + expect(key).toContain("ha-a:6041,ha-b:6042"); + }); + + test("keeps address-based pool key for non adapter_ha dsn", () => { + const pool = WebSocketConnectionPool.instance(); + const dsn = parse("ws://root:taosdata@host2:6042,host1:6041/mydb"); + const key = (pool as any).getPoolKey(dsn); + expect(key).toContain("host1:6041,host2:6042"); + }); }); describe("Security: pool key must include auth identity", () => { diff --git a/nodejs/test/common/dsn.test.ts b/nodejs/test/common/dsn.test.ts index f01a78e0..1bf788a7 100644 --- a/nodejs/test/common/dsn.test.ts +++ b/nodejs/test/common/dsn.test.ts @@ -1,4 +1,11 @@ -import { parse, WS_SQL_ENDPOINT } from "@src/common/dsn"; +import { + Address, + mergeAddresses, + parse, + parseDiscoveredEndpoints, + WS_SQL_ENDPOINT +} from "@src/common/dsn"; +import logger from "@src/common/log"; describe("dsn", () => { describe("parse", () => { @@ -483,4 +490,133 @@ describe("dsn", () => { expect(masked.params["TD.CONNECT.TOKEN"]).toBe("[REDACTED]"); }); }); + + describe("adapter ha helpers", () => { + test("isAdapterHA returns true only when adapter_ha=true", () => { + expect(parse("ws://root:taosdata@localhost:6041?adapter_ha=true").isAdapterHA()).toBe(true); + expect(parse("ws://root:taosdata@localhost:6041?adapter_ha=false").isAdapterHA()).toBe(false); + expect(parse("ws://root:taosdata@localhost:6041").isAdapterHA()).toBe(false); + }); + + test("parseDiscoveredEndpoints parses hostname and ipv6 endpoint", () => { + const result = parseDiscoveredEndpoints([ + "node2:6042", + "[::1]:6043", + ]); + expect(result).toEqual([ + { host: "node2", port: 6042 }, + { host: "[::1]", port: 6043 }, + ]); + }); + + test("parseDiscoveredEndpoints skips endpoints without explicit port and warns", () => { + const warnSpy = jest.spyOn(logger, "warn").mockImplementation(() => logger); + const result = parseDiscoveredEndpoints([ + "node3", + "tenant.cloud.tdengine.com", + "[2001:db8::10]", + ]); + expect(result).toEqual([]); + expect(warnSpy).toHaveBeenCalledTimes(3); + expect(warnSpy).toHaveBeenNthCalledWith( + 1, + "Adapter HA: ignoring invalid endpoint: node3" + ); + expect(warnSpy).toHaveBeenNthCalledWith( + 2, + "Adapter HA: ignoring invalid endpoint: tenant.cloud.tdengine.com" + ); + expect(warnSpy).toHaveBeenNthCalledWith( + 3, + "Adapter HA: ignoring invalid endpoint: [2001:db8::10]" + ); + warnSpy.mockRestore(); + }); + + test("parseDiscoveredEndpoints skips invalid endpoints and warns", () => { + const warnSpy = jest.spyOn(logger, "warn").mockImplementation(() => logger); + const result = parseDiscoveredEndpoints([ + "", + " ", + "::1:6041", + "bad_port:abc", + ]); + expect(result).toEqual([]); + expect(warnSpy).toHaveBeenCalledTimes(2); + expect(warnSpy).toHaveBeenNthCalledWith( + 1, + "Adapter HA: ignoring invalid endpoint: ::1:6041" + ); + expect(warnSpy).toHaveBeenNthCalledWith( + 2, + "Adapter HA: ignoring invalid endpoint: bad_port:abc" + ); + warnSpy.mockRestore(); + }); + + test("parseDiscoveredEndpoints skips out-of-range ports and warns", () => { + const warnSpy = jest.spyOn(logger, "warn").mockImplementation(() => logger); + const result = parseDiscoveredEndpoints([ + "host_over_max:70000", + "host_zero:0", + ]); + + expect(result).toEqual([]); + expect(warnSpy).toHaveBeenCalledTimes(2); + expect(warnSpy).toHaveBeenNthCalledWith( + 1, + "Adapter HA: ignoring invalid endpoint: host_over_max:70000" + ); + expect(warnSpy).toHaveBeenNthCalledWith( + 2, + "Adapter HA: ignoring invalid endpoint: host_zero:0" + ); + warnSpy.mockRestore(); + }); + + test("mergeAddresses deduplicates and appends newly discovered endpoints", () => { + const existing = [ + new Address("host1", 6041), + new Address("host2", 6042), + ]; + const discovered = [ + new Address("host2", 6042), + new Address("host3", 6043), + ]; + const merged = mergeAddresses(existing, discovered); + expect(merged).toEqual([ + { host: "host1", port: 6041 }, + { host: "host2", port: 6042 }, + { host: "host3", port: 6043 }, + ]); + }); + + test("mergeAddresses returns deep-copied addresses and does not mutate inputs", () => { + const existing = [new Address("seed1", 6041)]; + const discovered = [new Address("seed2", 6042)]; + const merged = mergeAddresses(existing, discovered); + + expect(merged).toEqual([ + { host: "seed1", port: 6041 }, + { host: "seed2", port: 6042 }, + ]); + expect(merged).not.toBe(existing); + expect(merged[0]).not.toBe(existing[0]); + expect(merged[1]).not.toBe(discovered[0]); + + merged[0].host = "changed"; + expect(existing[0].host).toBe("seed1"); + expect(discovered[0].host).toBe("seed2"); + }); + + test("mergeAddresses handles empty inputs", () => { + expect(mergeAddresses([], [])).toEqual([]); + expect(mergeAddresses([new Address("seed", 6041)], [])).toEqual([ + { host: "seed", port: 6041 }, + ]); + expect(mergeAddresses([], [new Address("discovered", 6042)])).toEqual([ + { host: "discovered", port: 6042 }, + ]); + }); + }); }); diff --git a/nodejs/test/sql/sql.failover.test.ts b/nodejs/test/sql/sql.failover.test.ts index 6bb93460..69911f85 100644 --- a/nodejs/test/sql/sql.failover.test.ts +++ b/nodejs/test/sql/sql.failover.test.ts @@ -1,7 +1,8 @@ +import { ClusterRegistry } from "@src/client/clusterRegistry"; import { WebSocketConnectionPool } from "@src/client/wsConnectorPool"; import { WSConfig } from "@src/common/config"; import { WsSql } from "@src/sql/wsSql"; -import { testPassword, testUsername } from "@test-helpers/utils"; +import { testNon3360, testPassword, testUsername } from "@test-helpers/utils"; import { WsProxy, WsProxyEvent } from "@test-helpers/wsProxy"; function parseBinaryAction(rawData: Buffer | string): bigint | null { @@ -11,6 +12,10 @@ function parseBinaryAction(rawData: Buffer | string): bigint | null { return rawData.readBigInt64LE(16); } +function resetClusterRegistrySingleton(): void { + (ClusterRegistry as any)._instance = undefined; +} + describe("sql failover", () => { jest.setTimeout(120 * 1000); @@ -395,3 +400,52 @@ describe("sql failover", () => { 300 * 1000 ); }); + +describe("sql adapter_ha pooled connector", () => { + afterEach(() => { + WebSocketConnectionPool.instance().destroyed(); + resetClusterRegistrySingleton(); + }); + + testNon3360("adapter_ha connect reuses pooled connector across seed variants", async () => { + const dsnA = + `ws://${testUsername()}:${testPassword()}` + + "@localhost:6041?adapter_ha=true"; + const dsnB = + `ws://${testUsername()}:${testPassword()}` + + "@localhost:6041,127.0.0.1:6041?adapter_ha=true"; + + let connA: WsSql | null = null; + let connB: WsSql | null = null; + + WebSocketConnectionPool.instance().destroyed(); + resetClusterRegistrySingleton(); + + try { + connA = await WsSql.open(new WSConfig(dsnA)); + const firstResult = await connA.exec("select server_version()"); + expect(firstResult).toBeTruthy(); + + const connectorA = ((connA as any)._wsClient as any)._wsConnector; + expect(connectorA).toBeDefined(); + expect(connectorA.getPoolKey()).toMatch(/^ws:\/\/[0-9a-f-]{36}\/ws#auth=/); + + await connA.close(); + connA = null; + + connB = await WsSql.open(new WSConfig(dsnB)); + const connectorB = ((connB as any)._wsClient as any)._wsConnector; + expect(connectorB).toBe(connectorA); + + const secondResult = await connB.exec("select server_version()"); + expect(secondResult).toBeTruthy(); + } finally { + if (connB) { + await connB.close(); + } + if (connA) { + await connA.close(); + } + } + }); +}); diff --git a/nodejs/test/sql/sql.test.ts b/nodejs/test/sql/sql.test.ts index f09848c3..2d6ed6f1 100644 --- a/nodejs/test/sql/sql.test.ts +++ b/nodejs/test/sql/sql.test.ts @@ -7,6 +7,7 @@ import { setLevel } from "@src/common/log"; let dsn = "ws://localhost:6041"; let password1 = "Ab1!@#$%,.:?<>;~"; let password2 = "Bc%^&*()-_+=[]{}"; + setLevel("debug"); beforeAll(async () => { let conf: WSConfig = new WSConfig(dsn); diff --git a/nodejs/test/tmq/tmq.failover.test.ts b/nodejs/test/tmq/tmq.failover.test.ts index 77bf53dd..7fd0b85c 100644 --- a/nodejs/test/tmq/tmq.failover.test.ts +++ b/nodejs/test/tmq/tmq.failover.test.ts @@ -1,9 +1,10 @@ +import { ClusterRegistry } from "@src/client/clusterRegistry"; import { WebSocketConnectionPool } from "@src/client/wsConnectorPool"; import { WSConfig } from "@src/common/config"; import { WsSql } from "@src/sql/wsSql"; import { TMQConstants } from "@src/tmq/constant"; import { WsConsumer } from "@src/tmq/wsTmq"; -import { testPassword, testUsername } from "@test-helpers/utils"; +import { testNon3360, testPassword, testUsername } from "@test-helpers/utils"; import { WsProxy } from "@test-helpers/wsProxy"; function parseJsonAction(rawData: Buffer | string): string | null { @@ -18,6 +19,10 @@ function parseJsonAction(rawData: Buffer | string): string | null { } } +function resetClusterRegistrySingleton(): void { + (ClusterRegistry as any)._instance = undefined; +} + describe("tmq failover", () => { jest.setTimeout(120 * 1000); @@ -435,3 +440,133 @@ describe("tmq failover", () => { } }, 300 * 1000); }); + +describe("tmq adapter_ha pooled connector", () => { + afterEach(() => { + WebSocketConnectionPool.instance().destroyed(); + resetClusterRegistrySingleton(); + }); + + testNon3360("adapter_ha consumes data and reuses pooled connector across seed variants", async () => { + const now = Date.now(); + const dbName = `test_adapter_ha_${now}`; + const tableName = "t0"; + const topicName = `topic_adapter_ha_${now}`; + const dsnA = "ws://localhost:6041?adapter_ha=true"; + const dsnB = "ws://localhost:6041,127.0.0.1:6041?adapter_ha=true"; + const localDsn = `ws://${testUsername()}:${testPassword()}@127.0.0.1:6041`; + const buildConfig = ( + groupId: string, + clientId: string, + url: string + ): Map => { + const cfg = new Map(); + cfg.set(TMQConstants.GROUP_ID, groupId); + cfg.set(TMQConstants.CONNECT_USER, testUsername()); + cfg.set(TMQConstants.CONNECT_PASS, testPassword()); + cfg.set(TMQConstants.AUTO_OFFSET_RESET, "earliest"); + cfg.set(TMQConstants.CLIENT_ID, clientId); + cfg.set(TMQConstants.WS_URL, url); + cfg.set(TMQConstants.ENABLE_AUTO_COMMIT, false); + cfg.set(TMQConstants.AUTO_COMMIT_INTERVAL_MS, 1000); + cfg.set("session.timeout.ms", "10000"); + cfg.set("max.poll.interval.ms", "30000"); + cfg.set("msg.with.table.name", "true"); + return cfg; + }; + + let setupSql: WsSql | null = null; + let cleanupSql: WsSql | null = null; + let consumerA: WsConsumer | null = null; + let consumerB: WsConsumer | null = null; + + WebSocketConnectionPool.instance().destroyed(); + resetClusterRegistrySingleton(); + + setupSql = await WsSql.open(new WSConfig(localDsn)); + try { + await setupSql.exec(`drop topic if exists ${topicName}`); + await setupSql.exec(`drop database if exists ${dbName}`); + await setupSql.exec(`create database ${dbName}`); + await setupSql.exec(`create table ${dbName}.${tableName}(ts timestamp, c1 int)`); + + const baseTs = 1700030000000; + const values = Array.from({ length: 10 }, (_, i) => `(${baseTs + i}, ${i})`); + await setupSql.exec( + `insert into ${dbName}.${tableName} values ${values.join(" ")}` + ); + await setupSql.exec(`create topic ${topicName} as select * from ${dbName}.${tableName}`); + } finally { + await setupSql.close(); + setupSql = null; + } + + try { + const cfgA = buildConfig( + `g_adapter_ha_${now}_a`, + `c_adapter_ha_${now}_a`, + dsnA + ); + consumerA = await WsConsumer.newConsumer(cfgA); + await consumerA.subscribe([topicName]); + + const connectorA = ((consumerA as any)._wsClient as any)._wsConnector; + expect(connectorA).toBeDefined(); + expect(connectorA.getPoolKey()).toMatch(/^ws:\/\/[0-9a-f-]{36}\/rest\/tmq#auth=/); + + let count = 0; + for (let i = 0; i < 20 && count < 10; i++) { + const res = await consumerA.poll(500); + for (const [, value] of res) { + const data = value.getData(); + if (!data || data.length === 0) { + continue; + } + count += data.length; + } + } + expect(count).toBeGreaterThanOrEqual(10); + + await consumerA.unsubscribe(); + await consumerA.close(); + consumerA = null; + + const cfgB = buildConfig( + `g_adapter_ha_${now}_b`, + `c_adapter_ha_${now}_b`, + dsnB + ); + consumerB = await WsConsumer.newConsumer(cfgB); + + const connectorB = ((consumerB as any)._wsClient as any)._wsConnector; + expect(connectorB).toBe(connectorA); + + await consumerB.subscribe([topicName]); + const res = await consumerB.poll(500); + expect(res).toBeTruthy(); + + await consumerB.unsubscribe(); + await consumerB.close(); + consumerB = null; + } finally { + if (consumerB) { + await consumerB.close(); + } + if (consumerA) { + await consumerA.close(); + } + + cleanupSql = await WsSql.open(new WSConfig(localDsn)); + try { + await cleanupSql.exec(`drop topic if exists ${topicName}`); + await cleanupSql.exec(`drop database if exists ${dbName}`); + } finally { + await cleanupSql.close(); + cleanupSql = null; + } + + WebSocketConnectionPool.instance().destroyed(); + resetClusterRegistrySingleton(); + } + }); +}); diff --git a/nodejs/test/tmq/wsTmq.adapterHa.test.ts b/nodejs/test/tmq/wsTmq.adapterHa.test.ts new file mode 100644 index 00000000..a0ee47bf --- /dev/null +++ b/nodejs/test/tmq/wsTmq.adapterHa.test.ts @@ -0,0 +1,98 @@ +import { TMQConstants } from "@src/tmq/constant"; +import { WsConsumer } from "@src/tmq/wsTmq"; + +function createConfigMap(): Map { + return new Map([ + [TMQConstants.WS_URL, "ws://root:taosdata@localhost:6041"], + [TMQConstants.GROUP_ID, "g1"], + ]); +} + +describe("WsConsumer adapter ha", () => { + afterEach(() => { + jest.restoreAllMocks(); + }); + + test("buildSubscribeMessage includes list_instances when requested", () => { + const consumer = new (WsConsumer as any)(createConfigMap()); + const msg = consumer.buildSubscribeMessage(["t1"], 1, true); + + expect(msg.action).toBe("subscribe"); + expect(msg.args.list_instances).toBe(true); + }); + + test("buildSubscribeMessage omits list_instances by default", () => { + const consumer = new (WsConsumer as any)(createConfigMap()); + const msg = consumer.buildSubscribeMessage(["t1"], 2); + + expect(msg.action).toBe("subscribe"); + expect(msg.args.list_instances).toBeUndefined(); + }); + + test("subscribe requests endpoint discovery once when adapter_ha is enabled", async () => { + const consumer = new (WsConsumer as any)(createConfigMap()); + const wsClient = { + isAdapterHA: jest.fn(() => true), + exec: jest.fn(async () => ({ + msg: { + code: 0, + list_instances: ["host2:6042", "host3:6043"], + }, + })), + mergeDiscoveredEndpoints: jest.fn(), + sendMsgDirect: jest.fn(), + setSessionRecoveryHook: jest.fn(), + }; + consumer._wsClient = wsClient; + + await consumer.subscribe(["topic_a"], 101); + + expect(wsClient.exec).toHaveBeenCalledTimes(1); + const [payload, bSqlQuery] = wsClient.exec.mock.calls[0] as any[]; + const msg = JSON.parse(payload as string); + expect(msg.args.list_instances).toBe(true); + expect(bSqlQuery).toBe(false); + expect(wsClient.mergeDiscoveredEndpoints).toHaveBeenCalledWith([ + "host2:6042", + "host3:6043", + ]); + }); + + test("subscribe omits endpoint discovery when adapter_ha is disabled", async () => { + const consumer = new (WsConsumer as any)(createConfigMap()); + const wsClient = { + isAdapterHA: jest.fn(() => false), + exec: jest.fn(async () => ({ msg: { code: 0 } })), + mergeDiscoveredEndpoints: jest.fn(), + sendMsgDirect: jest.fn(), + setSessionRecoveryHook: jest.fn(), + }; + consumer._wsClient = wsClient; + + await consumer.subscribe(["topic_b"], 102); + + const [payload, bSqlQuery] = wsClient.exec.mock.calls[0] as any[]; + const msg = JSON.parse(payload as string); + expect(msg.args.list_instances).toBeUndefined(); + expect(bSqlQuery).toBe(false); + expect(wsClient.mergeDiscoveredEndpoints).not.toHaveBeenCalled(); + }); + + test("recoverSessionContext re-subscribes without list_instances", async () => { + const consumer = new (WsConsumer as any)(createConfigMap()); + const wsClient = { + sendMsgDirect: jest.fn(async () => ({ msg: { code: 0 } })), + setSessionRecoveryHook: jest.fn(), + }; + consumer._wsClient = wsClient; + consumer._topics = ["topic_recover"]; + + await consumer.recoverSessionContext(); + + const [payload, bSqlQuery] = wsClient.sendMsgDirect.mock.calls[0] as any[]; + const msg = JSON.parse(payload as string); + expect(msg.action).toBe("subscribe"); + expect(msg.args.list_instances).toBeUndefined(); + expect(bSqlQuery).toBe(false); + }); +}); diff --git a/nodejs/tsconfig.json b/nodejs/tsconfig.json index 801d8bdc..95e14c9e 100644 --- a/nodejs/tsconfig.json +++ b/nodejs/tsconfig.json @@ -26,15 +26,14 @@ /* Modules */ "module": "commonjs", /* Specify what module code is generated. */ "rootDir": "./", /* Specify the root folder within your source files. */ - "baseUrl": "./", /* Specify the base directory to resolve non-relative module names. */ "paths": { /* Specify a set of entries that re-map imports to additional lookup locations. */ - "@src/*": ["src/*"], - "@test-helpers/*": ["test/helpers/*"] + "@src/*": ["./src/*"], + "@test-helpers/*": ["./test/helpers/*"] }, // "moduleResolution": "node", /* Specify how TypeScript looks up a file from a given module specifier. */ // "rootDirs": [], /* Allow multiple folders to be treated as one when resolving modules. */ // "typeRoots": [], /* Specify multiple folders that act like `./node_modules/@types`. */ - // "types": [], /* Specify type package names to be included without being referenced in a source file. */ + "types": ["node", "jest"], /* Explicitly include runtime and test globals for editor/tsserver consistency. */ // "allowUmdGlobalAccess": true, /* Allow accessing UMD globals from modules. */ "resolveJsonModule": true, /* Enable importing .json files */ // "noResolve": true, /* Disallow `import`s, `require`s or ``s from expanding the number of files TypeScript should add to a project. */