Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 | 53x 53x 5596x 5592x 4x 3x 1x 53x 53x 53x 53x 2087x 2087x 2087x 2087x 2087x 2087x 2087x 2087x 2087x 2087x 2087x 2087x 2087x 2087x 2087x 2087x 1x 2084x 2084x 2052x 2052x 32x 4x 4x 3509x 4168x 1371x 1371x 1371x 7014x 745x 745x 745x 745x 745x 601x 144x 741x 4x 4x 2061x 1108x 1108x 1108x 1108x 1108x 1x 1x 1108x 1108x 652x 1108x 1108x 1x 1x 1x 2018x 1x 2062x 2062x 2062x 2062x 2062x 1x 2061x 2052x 2052x 2052x 2052x 2052x 2030x 2030x 2030x 2030x 2084x 2084x 2087x 2087x 2087x 2087x 2087x 2086x 2086x 2086x 2086x 2086x 2086x 2086x 2086x 2086x 2086x 3x 2084x 2084x 1x 1x 2083x 48x 48x 16x 16x 2084x 1371x 1371x 2084x 2084x 3x 2084x 42x 42x 42x 35x 35x 1x 33x 33x 33x | import * as assert from 'node:assert/strict'
import { channel as dc } from 'node:diagnostics_channel'
import * as tls from 'node:tls'
import { create } from '@bufbuild/protobuf'
import { anyUnpack } from '@bufbuild/protobuf/wkt'
import { credentials } from '@grpc/grpc-js'
import { abortable, linkSignals } from '@ydbjs/abortable'
import {
DiscoveryServiceDefinition,
EndpointInfoSchema,
ListEndpointsResultSchema,
} from '@ydbjs/api/discovery'
import { StatusIds_StatusCode } from '@ydbjs/api/operation'
import type { CredentialsProvider } from '@ydbjs/auth'
import { AnonymousCredentialsProvider } from '@ydbjs/auth/anonymous'
import { loggers } from '@ydbjs/debug'
import { YDBError } from '@ydbjs/error'
import {
type Channel,
type ChannelOptions,
type Client,
type ClientMiddleware,
type CompatServiceDefinition,
Metadata,
composeClientMiddleware,
createClientFactory,
} from 'nice-grpc'
import pkg from '../package.json' with { type: 'json' }
import { BalancedChannel } from './channel.js'
import { type Connection, GrpcConnection } from './conn.js'
import type { DriverIdentity } from './driver-identity.js'
import {
type DiscoveryResult,
type EndpointPool,
type EndpointsRuntime,
type ListEndpoints,
createEndpointsRuntime,
mapDiscoveryResult,
} from './endpoints/endpoints-runtime.js'
import {
DriverCSDatabaseError,
DriverCSProtocolError,
DriverDegradedThresholdError,
DriverDiscoveryIntervalError,
DriverDiscoveryOptionsError,
DriverDiscoveryTimeoutError,
DriverResponseError,
} from './errors.js'
import type { DriverHooks, EndpointInfo } from './hooks.js'
import { debug, getRegisteredClientMiddlewares } from './middleware.js'
import { detectRuntime } from './runtime.js'
export type { DriverHooks, EndpointInfo }
/**
* Direct-IO target for `createClient` — route a client to an exact node.
*
* For topic direct read/write: the server hands out a `PartitionLocation`
* (`node_id` + `generation`) per partition; pass it here to reach that exact
* node. When `endpoint` is given the node is pinned on create (so it is
* reachable even before the next discovery round) and unpinned on the returned
* client's `[Symbol.dispose]` — use `using`/`await using`.
*/
export type ClientTarget = {
/** Exact node to route to (a discovery nodeId, or a PartitionLocation node_id). */
nodeId: bigint
/**
* Pin this endpoint for `nodeId` — for a server-named node that may not be in
* the current discovery snapshot. Unpinned when the client is disposed.
*/
endpoint?: {
host: string
port: number
location?: string
sslTargetNameOverride?: string
generation?: number
}
/**
* Never substitute another node — every RPC goes to `nodeId` or fails.
* @default false (soft affinity: prefer this node, fall back if unavailable)
*/
hard?: boolean
}
export type DriverOptions = {
/**
* SSL/TLS options for secure connections.
*
* @deprecated Use `secureOptions` instead.
*/
ssl?: tls.SecureContextOptions
secureOptions?: tls.SecureContextOptions | undefined
channelOptions?: ChannelOptions
credentialsProvider?: CredentialsProvider
/**
* Optional driver hooks.
*
* Hooks are synchronous, zero-cost when unused, and fire in the caller's
* AsyncLocalStorage context so OpenTelemetry trace.getActiveSpan() works.
*/
hooks?: DriverHooks
'ydb.sdk.application'?: string
'ydb.sdk.ready_timeout_ms'?: number
'ydb.sdk.token_timeout_ms'?: number
'ydb.sdk.enable_discovery'?: boolean
'ydb.sdk.discovery_timeout_ms'?: number
'ydb.sdk.discovery_interval_ms'?: number
'ydb.sdk.connection_idle_timeout_ms'?: number
'ydb.sdk.connection_idle_interval_ms'?: number
/**
* @deprecated The endpoints engine has no fixed pessimization timer; a node
* recovers on the next successful RPC or discovery round. Ignored.
*/
'ydb.sdk.connection_pessimization_timeout_ms'?: number
/**
* Prefer local-DC endpoints (opt-in, soft — only reorders tiers, never pins).
* @default false
*/
'ydb.sdk.locality_enabled'?: boolean
/**
* Prefer PRIMARY/PROMOTED-pile endpoints in a bridge (2DC) cluster, falling
* back to SYNCHRONIZED when the primary pile has no available node. Opt-in,
* soft, and a no-op outside bridge mode. Takes precedence over
* `ydb.sdk.locality_enabled` in bridge mode.
* @default false
*/
'ydb.sdk.prefer_primary_pile'?: boolean
/**
* Fraction of pessimized nodes (0..1) that forces an early rediscovery round.
* @default 0.5
*/
'ydb.sdk.discovery_degraded_threshold'?: number
}
let dbg = loggers.driver
let databaseFromUrl = function databaseFromUrl(url: URL): string {
if (url.pathname && url.pathname !== '/') {
return url.pathname
}
if (url.searchParams.has('database')) {
return url.searchParams.get('database') || ''
}
return ''
}
export const kRegisterLibrary: unique symbol = Symbol('ydbjs.core.registerLibrary')
let defaultOptions: DriverOptions = {
'ydb.sdk.ready_timeout_ms': 30_000,
'ydb.sdk.token_timeout_ms': 10_000,
'ydb.sdk.enable_discovery': true,
'ydb.sdk.discovery_timeout_ms': 10_000,
'ydb.sdk.discovery_interval_ms': 60_000,
'ydb.sdk.connection_idle_timeout_ms': 300_000,
'ydb.sdk.connection_idle_interval_ms': 60_000,
} as const satisfies DriverOptions
let defaultChannelOptions: ChannelOptions = {
'grpc.primary_user_agent': `ydb-js-sdk/${pkg.version}`,
'grpc.secondary_user_agent': detectRuntime(),
'grpc.keepalive_time_ms': 10_000,
'grpc.keepalive_timeout_ms': 5_000,
'grpc.keepalive_permit_without_calls': 1,
'grpc.max_send_message_length': 64 * 1024 * 1024,
'grpc.max_receive_message_length': 64 * 1024 * 1024,
'grpc.max_reconnect_backoff_ms': 5_000,
'grpc.initial_reconnect_backoff_ms': 50,
}
/* node:coverage ignore if -- polyfill for Node < 20.19; dead on supported runtimes */
if (!Promise.withResolvers) {
Promise.withResolvers = function <T>(): {
promise: Promise<T>
resolve: (value: T | PromiseLike<T>) => void
reject: (reason?: any) => void
} {
let resolve: (value: T | PromiseLike<T>) => void
let reject: (reason?: any) => void
let promise = new Promise<T>((res, rej) => {
resolve = res
reject = rej
})
return { promise, resolve: resolve!, reject: reject! }
}
}
export class Driver implements Disposable, AsyncDisposable {
readonly cs: URL
readonly options: DriverOptions = {}
// The endpoints engine — owns discovery, balancing, pessimization, and all
// ydb:driver.* diagnostics. Undefined when discovery is disabled (the sole
// bootstrap connection is used directly).
#endpoints: EndpointsRuntime | undefined
// Single connection: always the bootstrap endpoint from the connection
// string. Used for the discovery client, and as the sole transport when
// discovery is disabled. GrpcConnection creates the channel eagerly but
// grpc-js starts it IDLE — no TCP/TLS until the first RPC.
#connection: Connection
#middleware: ClientMiddleware
#discoveryClient: Client<typeof DiscoveryServiceDefinition> | undefined
// Ready latch for the discovery-DISABLED path only (resolved immediately at
// construction, rejected on close). The enabled path delegates to the pool.
#ready: PromiseWithResolvers<void> = Promise.withResolvers<void>()
#credentialsProvider: CredentialsProvider = new AnonymousCredentialsProvider()
#libraries: Set<string> = new Set()
#buildInfo: string = `ydb-js-sdk/${pkg.version}`
#initAt: number
#readyAt: number | undefined
#closed = false
#identity!: DriverIdentity
constructor(connectionString: string, userOptions: Readonly<DriverOptions> = defaultOptions) {
dbg.log('Driver(connectionString: %s, options: %o)', connectionString, userOptions)
this.#initAt = performance.now()
// close() rejects #ready to unblock awaiters; silence unhandled
// rejection when no one observes the promise.
this.#ready.promise.catch(() => {})
this.cs = this.#parseConnectionString(connectionString)
this.options = this.#mergeOptions(userOptions)
this.#assertDiscoveryTimings()
this.#identity = this.#buildIdentity()
let channelCredentials = this.#createChannelCredentials()
this.#connection = new GrpcConnection(
this.#initialEndpoint(),
channelCredentials,
this.options.channelOptions
)
if (this.options.credentialsProvider) {
this.#credentialsProvider = this.options.credentialsProvider
}
this.#middleware = this.#buildMiddleware()
if (this.options['ydb.sdk.enable_discovery'] === false) {
dbg.log('discovery disabled, using single endpoint')
this.#markReadyDisabled()
} else {
// The endpoints runtime kicks the first discovery round itself and owns
// the rediscovery loop + all ydb:driver.* diagnostics.
this.#endpoints = createEndpointsRuntime({
identity: this.identity,
listEndpoints: this.#fetchEndpoints,
channelCredentials,
channelOptions: this.options.channelOptions,
hooks: this.options.hooks,
localityEnabled: this.options['ydb.sdk.locality_enabled'],
preferPrimaryPile: this.options['ydb.sdk.prefer_primary_pile'],
degradedThreshold: this.options['ydb.sdk.discovery_degraded_threshold'],
discoveryTimeoutMs: this.options['ydb.sdk.discovery_timeout_ms'],
discoveryIntervalMs: this.options['ydb.sdk.discovery_interval_ms'],
idleIntervalMs: this.options['ydb.sdk.connection_idle_interval_ms'],
// `connection_idle_timeout_ms` now bounds the grace a retired (dropped
// from discovery) channel is kept before reaping — the endpoints engine
// has no separate idle-active teardown.
retiredGraceMs: this.options['ydb.sdk.connection_idle_timeout_ms'],
})
}
}
get token(): Promise<string> {
let signal = AbortSignal.timeout(this.options['ydb.sdk.token_timeout_ms']!)
return this.#credentialsProvider.getToken(false, signal)
}
get database(): string {
return databaseFromUrl(this.cs)
}
get isSecure(): boolean {
return this.cs.protocol === 'https:' || this.cs.protocol === 'grpcs:'
}
get application(): string {
Iif (this.options['ydb.sdk.application']) {
return this.options['ydb.sdk.application']
}
Iif (this.cs.searchParams.has('application')) {
return this.cs.searchParams.get('application') || ''
}
return ''
}
/**
* Stable identity stamped onto every `diagnostics_channel` payload so
* subscribers can attribute events to a specific Driver instance. Returns the
* same frozen object for the driver's lifetime — safe as a Map key.
*/
get identity(): DriverIdentity {
return this.#identity
}
async ready(signal?: AbortSignal): Promise<void> {
dbg.log('waiting for driver to become ready')
let timeout = this.options['ydb.sdk.ready_timeout_ms']!
using linkedSignal = linkSignals(signal, AbortSignal.timeout(timeout))
try {
if (this.#endpoints) {
await this.#endpoints.pool.ready(linkedSignal.signal)
} else {
await abortable(linkedSignal.signal, this.#ready.promise)
}
dbg.log('driver is ready')
} catch (error) {
dbg.log('driver failed to become ready: %O', error)
throw error
}
}
close(): void {
// Synchronous teardown — dispatch destroy; the pool closes channels and
// publishes ydb:driver.closed on a later turn.
this.#teardown((pool) => pool[Symbol.dispose]())
}
/**
* Create a nice-grpc client for the given service.
*
* When discovery is enabled, each RPC is routed through a BalancedChannel that
* selects a connection from the endpoints pool. When disabled, the single
* bootstrap connection is used directly.
*
* @param target Routing hint:
* - omitted → balanced across all healthy nodes.
* - `bigint` → soft affinity to that nodeId (node-bound query sessions).
* - `ClientTarget` → direct-IO: hard/soft routing to an exact node, with
* optional endpoint pinning. The returned client is `Disposable`; dispose
* it (`using`) to unpin.
*/
createClient<Service extends CompatServiceDefinition>(
service: Service,
preferNodeId?: bigint
): Client<Service>
createClient<Service extends CompatServiceDefinition>(
service: Service,
target: ClientTarget
): Client<Service> & Disposable
createClient<Service extends CompatServiceDefinition>(
service: Service,
target?: bigint | ClientTarget
): Client<Service> {
let nodeId = typeof target === 'bigint' ? target : target?.nodeId
let hard = typeof target === 'object' ? (target.hard ?? false) : false
dbg.log('creating client for %s (node %o, hard %o)', service.fullName, nodeId, hard)
// Pin a server-named endpoint so a hard client can reach it before the next
// discovery round; the pin is released on the client's dispose.
let pinned = false
if (typeof target === 'object' && target.endpoint !== undefined && this.#endpoints) {
this.#endpoints.pool.pin(target.nodeId, target.endpoint.host, target.endpoint.port, {
location: target.endpoint.location,
sslTargetNameOverride: target.endpoint.sslTargetNameOverride,
generation: target.endpoint.generation,
})
pinned = true
}
let channel = this.#connection.channel
if (this.#endpoints) {
channel = new BalancedChannel(
this.#endpoints.pool,
this.options.hooks,
nodeId,
hard
) as unknown as Channel
}
let client = createClientFactory().use(this.#middleware).create(service, channel, {
'*': this.options.channelOptions,
})
if (typeof target !== 'object') return client
// Object targets are Disposable — unpin on dispose (no-op if not pinned).
let endpoints = this.#endpoints
return Object.assign(client as object, {
[Symbol.dispose]: () => {
Eif (pinned) endpoints?.pool.invalidate(target.nodeId)
},
}) as Client<Service>
}
[Symbol.dispose](): void {
this.close()
}
async [Symbol.asyncDispose](): Promise<void> {
// Graceful teardown — drain in-flight streams before closing channels.
await this.#teardown((pool) => pool.close())
}
// The single teardown path. `closePool` is the only sync/async fork:
// `close()` destroys synchronously, `asyncDispose` awaits a graceful drain.
#teardown(closePool: (pool: EndpointPool) => void | Promise<void>): void | Promise<void> {
Iif (this.#closed) return
this.#closed = true
dbg.log('closing driver')
let closed = this.#endpoints ? closePool(this.#endpoints.pool) : this.#markClosedDisabled()
if (closed instanceof Promise) {
return closed.finally(() => this.#connection.close())
}
this.#connection.close()
}
#markReadyDisabled(): void {
this.#ready.resolve()
this.#readyAt = performance.now()
let duration = this.#readyAt - this.#initAt
dc('ydb:driver.ready').publish({ driver: this.identity, duration })
dbg.log('driver ready (discovery disabled) in %d ms', duration)
}
#markClosedDisabled(): void {
this.#ready.reject(new Error('driver closed'))
let uptime = this.#readyAt ? performance.now() - this.#readyAt : 0
dc('ydb:driver.closed').publish({ driver: this.identity, uptime })
dbg.log('closing driver (uptime %d ms)', uptime)
}
#buildIdentity(): DriverIdentity {
let port = this.cs.port ? parseInt(this.cs.port, 10) : undefined
return Object.freeze({
database: this.database,
address: this.cs.hostname,
...(port !== undefined && { port }),
})
}
#parseConnectionString(cs: string): URL {
Iif (!cs) {
throw new Error('Invalid connection string. Must be a non-empty string')
}
let url = new URL(cs.replace(/^grpc/, 'http'))
assert.match(url.protocol, /^(grpc|http)(s?):$/, new DriverCSProtocolError())
assert.ok(databaseFromUrl(url), new DriverCSDatabaseError())
return url
}
#mergeOptions(userOptions: Readonly<DriverOptions>): DriverOptions {
let merged: DriverOptions = { ...defaultOptions, ...userOptions }
merged.channelOptions = { ...defaultChannelOptions, ...merged.channelOptions }
return merged
}
#assertDiscoveryTimings(): void {
let timeout = this.options['ydb.sdk.discovery_timeout_ms']!
let interval = this.options['ydb.sdk.discovery_interval_ms']!
assert.ok(timeout > 0, new DriverDiscoveryTimeoutError(timeout))
assert.ok(interval > 0, new DriverDiscoveryIntervalError(interval))
assert.ok(timeout < interval, new DriverDiscoveryOptionsError())
let threshold = this.options['ydb.sdk.discovery_degraded_threshold']
if (threshold !== undefined) {
assert.ok(threshold > 0 && threshold <= 1, new DriverDegradedThresholdError(threshold))
}
}
#initialEndpoint() {
return create(EndpointInfoSchema, {
address: this.cs.hostname,
nodeId: -1,
port: parseInt(this.cs.port || (this.isSecure ? '443' : '80'), 10),
ssl: this.isSecure,
})
}
#createChannelCredentials() {
if ((this.options.secureOptions ??= this.options.ssl)) {
let secureContext = tls.createSecureContext(this.options.secureOptions)
return credentials.createFromSecureContext(secureContext)
}
return this.isSecure ? credentials.createSsl() : credentials.createInsecure()
}
[kRegisterLibrary](name: string, version: string): void {
let entry = `${name}/${version}`
if (this.#libraries.has(entry)) return
this.#libraries.add(entry)
this.#buildInfo = `${this.#buildInfo};${entry}`
}
#buildMiddleware(): ClientMiddleware {
let stamp: ClientMiddleware = (call, options) => {
let metadata = Metadata(options.metadata)
.set('x-ydb-sdk-build-info', this.#buildInfo)
.set('x-ydb-database', this.database)
.set('x-ydb-application-name', this.application)
return call.next(call.request, Object.assign(options, { metadata }))
}
// Order: debug (logging) → stamp (SDK / db / app headers) → any
// externally-registered middleware → auth (x-ydb-auth-ticket). Auth runs
// last so a token refresh from inside another middleware still wins.
//
// The registry snapshot is taken at construction — call
// addClientMiddleware() BEFORE new Driver(...) for it to apply.
let chain = composeClientMiddleware(debug, stamp)
for (let mw of getRegisteredClientMiddlewares()) {
chain = composeClientMiddleware(chain, mw)
}
return composeClientMiddleware(chain, this.#credentialsProvider.middleware)
}
// The listEndpoints seam handed to the endpoints runtime. One plain RPC — the
// FSM owns retry/backoff, and the run_discovery_round effect owns the
// tracing:ydb:driver.discovery span; this closure must not add either.
#fetchEndpoints: ListEndpoints = async (signal: AbortSignal): Promise<DiscoveryResult> => {
let client = (this.#discoveryClient ??= createClientFactory()
.use(this.#middleware)
.create(DiscoveryServiceDefinition, this.#connection.channel))
let response = await client.listEndpoints({ database: this.database }, { signal })
assert.ok(response.operation, new DriverResponseError('Missing operation data.'))
if (response.operation.status !== StatusIds_StatusCode.SUCCESS) {
throw new YDBError(response.operation.status, response.operation.issues)
}
let res = anyUnpack(response.operation.result!, ListEndpointsResultSchema)
assert.ok(res, new DriverResponseError('Missing result in operation data.'))
return mapDiscoveryResult(res)
}
}
|