Fix memory leaks in Driver class

- Remove waitForChannelReady from constructor to prevent Promise chain memory leaks
- Add lazy initialization for discovery client (only created when needed)
- Implement channel connectivity check in ready() method for non-discovery mode
- Add test for memory leak prevention (100,000 Driver instances)
- Use Promise.withResolvers() for cleaner promise handling
- Properly handle AbortSignal cancellation in channel connectivity loop
- Import ConnectivityState enum for type-safe state checks

The main issue was that waitForChannelReady() in constructor created Promise
chains with closures that held references to 'this', preventing garbage
collection. Now channel connectivity is checked lazily only when ready() is
called, and the promise chain is properly managed."

Author: polRk

packages/core/src/driver.test.ts

@@ -56,3 +56,30 @@ test('allows custom channel options override', () => {
 
 	driver.close()
 })
+
+test('creating thousands of drivers with using does not leak memory', async () => {
+	let iterations = 100000
+	let memoryBefore = process.memoryUsage().heapUsed
+
+	for (let i = 0; i < iterations; i++) {
+		using _driver = new Driver('grpc://localhost:2136/local', {
+			'ydb.sdk.enable_discovery': false,
+		})
+
+		if (i % 1000 === 0 && i > 0) {
+			if (global.gc) {
+				global.gc()
+			}
+		}
+	}
+
+	if (global.gc) {
+		global.gc()
+	}
+
+	let memoryAfter = process.memoryUsage().heapUsed
+	let memoryGrowth = memoryAfter - memoryBefore
+	let memoryGrowthMB = memoryGrowth / (1024 * 1024)
+
+	expect(memoryGrowthMB).toBeLessThan(50)
+})

packages/core/src/driver.ts

@@ -12,6 +12,7 @@ import { loggers } from '@ydbjs/debug'
 import { YDBError } from '@ydbjs/error'
 import { type RetryConfig, defaultRetryConfig, retry } from '@ydbjs/retry'
 import {
+	type Channel,
 	type ChannelOptions,
 	type Client,
 	ClientError,
@@ -21,13 +22,13 @@ import {
 	Status,
 	composeClientMiddleware,
 	createClientFactory,
-	waitForChannelReady,
 } from 'nice-grpc'
 import pkg from '../package.json' with { type: 'json' }
 import { type Connection, LazyConnection } from './conn.js'
 import { debug } from './middleware.js'
 import { ConnectionPool } from './pool.js'
 import { detectRuntime } from './runtime.js'
+import { ConnectivityState } from '@grpc/grpc-js/build/src/connectivity-state.js'
 
 let dbg = loggers.driver
 
@@ -170,24 +171,11 @@ export class Driver implements Disposable {
 
 		this.#pool = new ConnectionPool(channelCredentials, this.options.channelOptions)
 
-		this.#discoveryClient = createClientFactory()
-			.use(this.#middleware)
-			.create(DiscoveryServiceDefinition, this.#connection.channel)
-
 		if (this.options['ydb.sdk.enable_discovery'] === false) {
 			dbg.log('discovery disabled, using single endpoint')
-			waitForChannelReady(
-				this.#connection.channel,
-				new Date(Date.now() + (this.options['ydb.sdk.ready_timeout_ms'] || 10000))
-			)
-				.then(() => {
-					dbg.log('single endpoint ready')
-					return this.#ready.resolve()
-				})
-				.catch((error) => {
-					dbg.log('single endpoint failed to become ready: %O', error)
-					this.#ready.reject(error)
-				})
+			// Channel will be lazily created on first use
+			// Readiness check is skipped to avoid memory leaks from Promise chains
+			this.#ready.resolve()
 		}
 
 		if (this.options['ydb.sdk.enable_discovery'] === true) {
@@ -242,6 +230,16 @@ export class Driver implements Disposable {
 		return this.cs.protocol === 'https:' || this.cs.protocol === 'grpcs:'
 	}
 
+	get #getDiscoveryClient(): Client<typeof DiscoveryServiceDefinition> {
+		if (this.#discoveryClient === null) {
+			dbg.log('creating discovery client')
+			this.#discoveryClient = createClientFactory()
+				.use(this.#middleware)
+				.create(DiscoveryServiceDefinition, this.#connection.channel)
+		}
+		return this.#discoveryClient
+	}
+
 	async #discovery(signal: AbortSignal): Promise<void> {
 		dbg.log('starting discovery for database: %s', this.database)
 
@@ -255,7 +253,7 @@ export class Driver implements Disposable {
 
 		let result = await retry(retryConfig, async (signal) => {
 			dbg.log('attempting to list endpoints for database: %s', this.database)
-			let response = await this.#discoveryClient.listEndpoints({ database: this.database }, { signal })
+			let response = await this.#getDiscoveryClient.listEndpoints({ database: this.database }, { signal })
 			if (!response.operation) {
 				throw new ClientError(
 					DiscoveryServiceDefinition.listEndpoints.path,
@@ -289,19 +287,63 @@ export class Driver implements Disposable {
 
 	async ready(signal?: AbortSignal): Promise<void> {
 		dbg.log('waiting for driver to become ready')
-		signal = signal
-			? AbortSignal.any([signal, AbortSignal.timeout(this.options['ydb.sdk.ready_timeout_ms']!)])
-			: AbortSignal.timeout(this.options['ydb.sdk.ready_timeout_ms']!)
+
+		let timeoutMs = this.options['ydb.sdk.ready_timeout_ms']!
+		let effectiveSignal = signal
+			? AbortSignal.any([signal, AbortSignal.timeout(timeoutMs)])
+			: AbortSignal.timeout(timeoutMs)
 
 		try {
-			await abortable(signal, this.#ready.promise)
+			await abortable(effectiveSignal, this.#ready.promise)
+
+			if (this.options['ydb.sdk.enable_discovery'] === false) {
+				dbg.log('checking channel connectivity for single endpoint mode')
+				await this.#checkChannelConnectivity(this.#connection.channel, timeoutMs, effectiveSignal)
+			}
+
 			dbg.log('driver is ready')
 		} catch (error) {
 			dbg.log('driver failed to become ready: %O', error)
 			throw error
 		}
 	}
 
+	async #checkChannelConnectivity(channel: Channel, timeoutMs: number, signal: AbortSignal): Promise<void> {
+		let deadline = new Date(Date.now() + timeoutMs)
+
+		while (true) {
+			if (signal.aborted) {
+				throw signal.reason || new Error('Aborted while waiting for channel connectivity')
+			}
+
+			let state = channel.getConnectivityState(true) // true = try to connect
+			dbg.log('channel connectivity state: %d', state)
+
+			if (state === ConnectivityState.READY) {
+				dbg.log('channel is ready')
+				return
+			}
+
+			if (state === ConnectivityState.SHUTDOWN) {
+				throw new Error('Channel is shutdown')
+			}
+
+			let { promise, resolve, reject } = Promise.withResolvers<void>()
+			channel.watchConnectivityState(state, deadline, (err?: Error) => {
+				if (err) {
+					dbg.log('channel connectivity state change timeout: %O', err)
+					reject(err)
+				} else {
+					dbg.log('channel connectivity state changed')
+					resolve()
+				}
+			})
+
+			// oxlint-disable-next-line no-await-in-loop
+			await abortable(signal, promise)
+		}
+	}
+
 	close(): void {
 		dbg.log('closing driver')
 		if (this.#rediscoverTimer) {

vitest.config.ts

@@ -2,25 +2,27 @@ import { defineConfig } from 'vitest/config'
 
 export default defineConfig({
 	test: {
+		execArgv: ['--expose-gc'],
+		vmMemoryLimit: '300Mb',
 		projects: [
 			{
 				test: {
 					name: {
 						label: 'uni',
-						color: 'yellow'
+						color: 'yellow',
 					},
 					include: ['packages/*/src/**/*.test.ts'],
 					environment: 'node',
 					benchmark: {
-						include: ['packages/*/src/**/*.bench.ts']
-					}
+						include: ['packages/*/src/**/*.bench.ts'],
+					},
 				},
 			},
 			{
 				test: {
 					name: {
 						label: 'int',
-						color: 'blue'
+						color: 'blue',
 					},
 					include: ['packages/*/tests/**/*.test.ts'],
 					environment: 'node',
@@ -34,7 +36,7 @@ export default defineConfig({
 				test: {
 					name: {
 						label: 'e2e',
-						color: 'magenta'
+						color: 'magenta',
 					},
 					include: ['e2e/**/*.test.ts'],
 					environment: 'node',
@@ -43,8 +45,8 @@ export default defineConfig({
 					hookTimeout: 30000,
 					maxConcurrency: 1,
 					benchmark: {
-						include: ['e2e/**/*.bench.ts']
-					}
+						include: ['e2e/**/*.bench.ts'],
+					},
 				},
 			},
 		],
@@ -59,7 +61,7 @@ export default defineConfig({
 				'**/vitest.*',
 				'**/*.test.ts',
 				'**/*.bench.ts',
-			]
-		}
+			],
+		},
 	},
 })

vitest.setup.ydb.ts

@@ -32,10 +32,8 @@ export async function setup(project: TestProject) {
 		return
 	}
 
-	let ports = ['2135', '2136', '8765', '9092'].map((port) => `--publish ${port}`).join(' ')
-
 	// prettier-ignore
-	let container = await $`docker run --rm --detach --hostname localhost --platform linux/amd64 ${ports} ydbplatform/local-ydb:25.2`.text()
+	let container = await $`docker run --rm --detach --hostname localhost --platform linux/amd64 --publish 2135:2135 --publish 2136:2136 --publish 8765:8765 --publish 9092:9092 ydbplatform/local-ydb:25.2.1`.text()
 	containerID = container.trim()
 
 	let signal = AbortSignal.timeout(30 * 1000)