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 | 15x 15x 15x 15x 15x 15x 15x 15x 51121x 1x 15x 118x 1x 15x 773x 2x 15x 14x 1x 15x 51121x 15x 11x 1x 15x 825x 15x 773x 773x 773x 773x 1049308x 1049296x 12x 6x 6x 773x | import { channel as dc, tracingChannel } from 'node:diagnostics_channel'
import type { DriverIdentity } from '@ydbjs/core'
import type { AckStatus } from './types.js'
// Structured lifecycle signals for @ydbjs/telemetry and future metrics/traces/logs
// subscribers. Channel names follow AGENTS.md: event channels are
// `ydb:<subsystem>.<concept>.<action>`, tracing channels are
// `tracing:ydb:<subsystem>.<concept>.<operation>`. The writer owns the `topic`
// subsystem. Every payload carries `driver` so multi-driver subscribers can
// attribute it; durations/timestamps are ms.
// Stamped onto every payload so multi-driver subscribers can attribute events.
export type WriterScope = {
driver: DriverIdentity
topic: string
producer: string
}
// One-shot config snapshot, published on `opened` so metrics/traces subscribers
// that join later still learn the writer's effective configuration.
export type WriterConfig = {
codec: number
maxInflightCount: number
maxBufferBytes: bigint
flushIntervalMs: number
updateTokenIntervalMs: number
gracefulShutdownTimeoutMs: number
recoveryWindowMs: number
retryOnSchemeError: boolean
partitionId?: bigint
messageGroupId?: string
}
// Per-write-response breakdown, so a metrics subscriber can split written vs
// server-deduplicated (`skipped`) vs `writtenInTx` and track acked byte throughput.
export type AckBreakdown = {
written: number
skipped: number
writtenInTx: number
bytes: bigint
}
let openedCh = dc('ydb:topic.writer.opened')
let sessionStartedCh = dc('ydb:topic.writer.session.started')
let acknowledgedCh = dc('ydb:topic.writer.acknowledged')
let reconnectingCh = dc('ydb:topic.writer.reconnecting')
let closedCh = dc('ydb:topic.writer.closed')
let erroredCh = dc('ydb:topic.writer.errored')
// One flush() call → one span, so latency of the "wait until durable" barrier is
// traceable end to end (it spans batching + server acks + any reconnect in between).
let flushCh = tracingChannel<WriterScope>('tracing:ydb:topic.writer.flush')
// Fired once when the writer is created, carrying its effective configuration.
// 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: WriterScope, config: WriterConfig): void {
if (openedCh.hasSubscribers) {
openedCh.publish({ ...scope, config })
}
}
// Fired once per (re)established write session, carrying the recovered high-water mark.
export let publishSessionStarted = function publishSessionStarted(
scope: WriterScope,
sessionId: string,
lastSeqNo: bigint
): void {
if (sessionStartedCh.hasSubscribers) {
sessionStartedCh.publish({ ...scope, sessionId, lastSeqNo })
}
}
// Fired per server write-response with the per-status breakdown and acked bytes.
export let publishAcknowledged = function publishAcknowledged(
scope: WriterScope,
acks: AckBreakdown
): void {
if (acknowledgedCh.hasSubscribers) {
acknowledgedCh.publish({ ...scope, ...acks })
}
}
// Fired when the stream drops and the writer enters transparent reconnect.
export let publishReconnecting = function publishReconnecting(
scope: WriterScope,
attempt: number,
error: unknown
): void {
if (reconnectingCh.hasSubscribers) {
reconnectingCh.publish({ ...scope, attempt, error })
}
}
// Fired once when the writer shuts down (graceful close or destroy).
export let publishClosed = function publishClosed(scope: WriterScope): void {
Iif (closedCh.hasSubscribers) {
closedCh.publish({ ...scope })
}
}
// Fired once when the writer fails terminally and stops — distinct from the
// transient `reconnecting` event, which keeps the writer alive.
export let publishErrored = function publishErrored(scope: WriterScope, error: unknown): void {
if (erroredCh.hasSubscribers) {
erroredCh.publish({ ...scope, error })
}
}
// Wrap a flush() so it emits a `tracing:ydb:topic.writer.flush` span.
export let traceFlush = function traceFlush(
scope: WriterScope,
fn: () => Promise<bigint>
): Promise<bigint> {
return flushCh.tracePromise(fn, { ...scope })
}
// Build the per-status breakdown for a write-response acknowledgment.
export let ackBreakdown = function ackBreakdown(
acknowledgments: Map<bigint, AckStatus>,
bytes: bigint
): AckBreakdown {
let written = 0
let skipped = 0
let writtenInTx = 0
for (let status of acknowledgments.values()) {
if (status === 'written') {
written += 1
} else if (status === 'skipped') {
skipped += 1
} else {
writtenInTx += 1
}
}
return { written, skipped, writtenInTx, bytes }
}
|