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 | 22x 22x 22x 22x 22x 22x 22x 22x 22x 22x 22x 10734x 22x 399x 4x 22x 10738x 1x 22x 145x 1x 22x 19x 2x 22x 81x 1x 22x 17x 22x 10734x 22x 3x 22x 88x | import { channel as dc, tracingChannel } from 'node:diagnostics_channel'
import type { DriverIdentity } from '@ydbjs/core'
// Structured lifecycle signals for @ydbjs/telemetry and future metrics/traces/logs
// subscribers. Channel names follow AGENTS.md: `ydb:<subsystem>.<concept>.<action>`
// for events and `tracing:ydb:<subsystem>.<concept>.<operation>` for spans. The reader
// owns the `topic` subsystem alongside the writer. Every payload carries `driver` so
// multi-driver subscribers can attribute events; durations/timestamps are ms.
// Stamped onto every payload so multi-driver subscribers can attribute events.
export type ReaderScope = {
driver: DriverIdentity
consumer: string
topics: string[]
}
// One-shot config snapshot, published on `opened` so late-joining subscribers still
// learn the reader's effective configuration.
export type ReaderConfig = {
maxBufferBytes: bigint
updateTokenIntervalMs: number
gracefulShutdownTimeoutMs: number
recoveryWindowMs: number
retryOnSchemeError: boolean
}
let openedCh = dc('ydb:topic.reader.opened')
let bufferChangedCh = dc('ydb:topic.reader.buffer.changed')
let sessionStartedCh = dc('ydb:topic.reader.session.started')
let partitionStartedCh = dc('ydb:topic.reader.partition.started')
let partitionStoppedCh = dc('ydb:topic.reader.partition.stopped')
let committedCh = dc('ydb:topic.reader.committed')
let reconnectingCh = dc('ydb:topic.reader.reconnecting')
let closedCh = dc('ydb:topic.reader.closed')
let erroredCh = dc('ydb:topic.reader.errored')
// One microtask commit batch → one span (server ack + any reconnect in between).
let commitCh = tracingChannel<ReaderScope>('tracing:ydb:topic.reader.commit')
// Every helper below guards with hasSubscribers before publish(). publish() itself
// is a no-op without subscribers, but its argument is built before the call — the
// guard skips that payload allocation (same rule as `if (dbg.enabled)` for logs).
export let publishOpened = function publishOpened(scope: ReaderScope, config: ReaderConfig): void {
Iif (openedCh.hasSubscribers) {
openedCh.publish({ ...scope, config })
}
}
export let publishBufferChanged = function publishBufferChanged(
scope: ReaderScope,
bufferedBytes: bigint
): void {
if (bufferChangedCh.hasSubscribers) {
bufferChangedCh.publish({ ...scope, bufferedBytes })
}
}
// Fired once per (re)established read session.
export let publishSessionStarted = function publishSessionStarted(
scope: ReaderScope,
sessionId: string
): void {
if (sessionStartedCh.hasSubscribers) {
sessionStartedCh.publish({ ...scope, sessionId })
}
}
export let publishPartitionStarted = function publishPartitionStarted(
scope: ReaderScope,
partitionId: bigint,
partitionSessionId: bigint,
committedOffset: bigint
): void {
if (partitionStartedCh.hasSubscribers) {
partitionStartedCh.publish({ ...scope, partitionId, partitionSessionId, committedOffset })
}
}
export let publishPartitionStopped = function publishPartitionStopped(
scope: ReaderScope,
partitionId: bigint,
reason: 'graceful' | 'lost' | 'ended'
): void {
if (partitionStoppedCh.hasSubscribers) {
partitionStoppedCh.publish({ ...scope, partitionId, reason })
}
}
export let publishCommitted = function publishCommitted(
scope: ReaderScope,
partitionId: bigint,
committedOffset: bigint
): void {
if (committedCh.hasSubscribers) {
committedCh.publish({ ...scope, partitionId, committedOffset })
}
}
export let publishReconnecting = function publishReconnecting(
scope: ReaderScope,
attempt: number,
error: unknown
): void {
Iif (reconnectingCh.hasSubscribers) {
reconnectingCh.publish({ ...scope, attempt, error })
}
}
export let publishClosed = function publishClosed(scope: ReaderScope): void {
Iif (closedCh.hasSubscribers) {
closedCh.publish({ ...scope })
}
}
export let publishErrored = function publishErrored(scope: ReaderScope, error: unknown): void {
Iif (erroredCh.hasSubscribers) {
erroredCh.publish({ ...scope, error })
}
}
// Wrap a commit() so it emits a `tracing:ydb:topic.reader.commit` span.
export let traceCommit = function traceCommit(
scope: ReaderScope,
fn: () => Promise<void>
): Promise<void> {
return commitCh.tracePromise(fn, { ...scope })
}
|