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 | 23x 167943x 2x 167941x 167941x 167943x 5x 23x 167938x 7x 23x 95x 95x 17768x 17768x 3x 17765x 3x 17762x 17762x 17762x 17762x 17762x 90x 17672x 89x 90x 89x 23x 142x 120x 22x 35x 35x 35x 35x 35x 35x 19x 16x 12x 23x 12x 1x 11x 11x 13x 11x | import { timestampDate } from '@bufbuild/protobuf/wkt'
import { Codec } from '@ydbjs/api/topic'
import { type CodecMap, getCodec } from '../codec.js'
import { TopicMessage } from '../message.js'
import type { TopicPartitionSession } from '../partition-session.js'
import { type OffsetRange, type ReaderMessage, mergeRanges, partitionKey } from './reader-state.js'
// Pure facade helpers — no instance state, no I/O. Extracted from TopicReader so the
// wire→user mapping, commit-range building and tx offset accounting are directly
// unit-testable without a fake driver.
export type TxReadOffsets = {
session: TopicPartitionSession
firstOffset: bigint
lastOffset: bigint
}
export type TxReadOffsetUpdate = {
partitionSession: TopicPartitionSession
offsetRange: { firstOffset: bigint; lastOffset: bigint }
}
export let decodePayload = function decodePayload(
codecs: CodecMap,
session: TopicPartitionSession,
message: ReaderMessage
): Uint8Array {
// UNSPECIFIED means "no codec recorded" — the payload is raw bytes (the current
// server normalizes missing codecs to RAW before delivery; older ones could
// leave it unset). Decompressing would corrupt it; erroring would kill the
// reader over absent metadata.
if (message.codec === Codec.UNSPECIFIED) {
return message.data
}
try {
let codec = codecs.get(message.codec) ?? getCodec(message.codec as Codec)
return codec.decompress(message.data)
} catch (error) {
// Terminal by design: the protocol has no way to refuse a single partition,
// and skipping silently would be data loss. Make the error actionable.
throw new Error(
`Cannot decode message at offset ${message.offset} of partition ${session.partitionId} (${session.topicPath}): codec ${message.codec} — register it in codecMap`,
{ cause: error }
)
}
}
export let toTopicMessage = function toTopicMessage(
codecs: CodecMap,
session: TopicPartitionSession,
message: ReaderMessage
): TopicMessage {
return new TopicMessage({
partitionSession: session,
producer: message.producer,
payload: decodePayload(codecs, session, message),
codec: message.codec as Codec,
seqNo: message.seqNo,
offset: message.offset,
uncompressedSize: message.uncompressedSize,
commitRangeStart: message.commitRangeStart,
...(message.createdAt && { createdAt: timestampDate(message.createdAt).getTime() }),
...(message.writtenAt && { writtenAt: timestampDate(message.writtenAt).getTime() }),
...(message.metadataItems.length > 0 && {
metadataItems: Object.fromEntries(
message.metadataItems.map((item) => [item.key, item.value])
),
}),
})
}
// Group commit input by the stable partitionKey — partition ids alone collide across
// the topics of a multi-topic reader. Each message acknowledges its own stitched
// range: the offset itself plus the server-side hole immediately preceding it — never
// other delivered messages (the server commits gap-free ACKED intervals only). The
// per-partition ranges come back merged, ready for the wire.
export let buildCommitRanges = function buildCommitRanges(
messages: TopicMessage[],
isOwned: (session: TopicPartitionSession) => boolean
): Map<string, OffsetRange[]> {
let byPartition = new Map<string, OffsetRange[]>()
for (let message of messages) {
let session = message.partitionSession.deref()
if (!session || session.isStopped) {
throw new Error('Cannot commit a message from a stopped or expired partition session')
}
// Ownership: a message from another reader would be committed against this
// reader's consumer and anchor — silently corrupting both consumers' progress.
if (!isOwned(session)) {
throw new Error(
`Cannot commit a message from a foreign reader's partition session (partition ${session.partitionId} of ${session.topicPath})`
)
}
Iif (message.offset === undefined) {
throw new Error('Cannot commit a message without an offset')
}
let key = partitionKey(session.topicPath, session.partitionId)
let ranges = byPartition.get(key)
let range = { start: message.commitRangeStart, end: message.offset + 1n }
if (ranges === undefined) {
byPartition.set(key, [range])
} else {
ranges.push(range)
}
}
for (let [key, ranges] of byPartition) {
byPartition.set(key, mergeRanges(ranges))
}
return byPartition
}
// Record delivered offsets for the tx commit hook — at yield time, never at
// buffering: a tx commit must cover exactly what the consumer saw. No-op for
// non-transactional readers (no map).
export let growTxOffsets = function growTxOffsets(
offsets: Map<string, TxReadOffsets> | undefined,
messages: TopicMessage[]
): void {
if (!offsets) {
return
}
for (let message of messages) {
let session = message.partitionSession.deref()
Iif (!session) {
continue
}
let offset = message.offset ?? message.commitRangeStart
let key = partitionKey(session.topicPath, session.partitionId)
let existing = offsets.get(key)
if (existing === undefined) {
offsets.set(key, {
session,
firstOffset: message.commitRangeStart,
lastOffset: offset,
})
} else if (offset > existing.lastOffset) {
// Grow-only: a mid-tx reconnect redelivers from the committed offset, and
// a rewound range would commit fewer offsets than the transaction consumed.
existing.lastOffset = offset
}
}
}
// Snapshot of the tx read offsets, mapped to the sessions the tx commit hook needs.
// The session is embedded in each record, so the snapshot stays valid after close()
// (a closed tx reader still binds its offsets at tx commit).
export let txOffsetUpdates = function txOffsetUpdates(
offsets: Map<string, TxReadOffsets> | undefined
): TxReadOffsetUpdate[] {
if (!offsets) {
return []
}
let updates: TxReadOffsetUpdate[] = []
for (let record of offsets.values()) {
updates.push({
partitionSession: record.session,
offsetRange: { firstOffset: record.firstOffset, lastOffset: record.lastOffset },
})
}
return updates
}
|