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 | 17x 52232x 52232x 51123x 52232x 3x 52232x 17x 1809x 1809x 1055x 1809x 1809x 17x 264892x 264892x 4x 264888x 52324x 52322x 2x 1x 1x 107375x 56x 107319x 52232x 52232x 55087x 1496x 1496x 53591x 210x 210x 53381x 754x 52627x 1055x 51572x 51572x 1810x 1060x 750x 749x 1x 103379x | import { fail } from 'node:assert/strict'
import type { TransitionResult, TransitionRuntime } from '@ydbjs/fsm'
import type { WriteAck } from './types.js'
// One physical streamWrite lifecycle. All business logic (buffer, seqNo, backoff)
// lives in the writer FSM; the transport only owns the socket and forwards
// classified server messages to the writer as outputs.
//
// The full transition map (table + diagram) lives in packages/topic/ARCHITECTURE.md —
// update it in the same commit when you change this dispatch.
export type TransportState = 'idle' | 'connecting' | 'ready' | 'disconnected' | 'closed'
// The transport FSM keeps no ctx state — every fact rides on the event/output.
export type TransportCtx = {}
// Events are pre-classified by the ingest task, so the transition never touches
// raw protobuf — it just maps stream facts to state and outputs.
export type TransportEvent =
// lifecycle commands the owner dispatches inward
| { type: 'transport.connect' }
| { type: 'transport.close' }
| { type: 'transport.destroy'; reason?: unknown }
// classified stream facts the ingest task forwards
| {
type: 'transport.init'
sessionId: string
lastSeqNo: bigint
partitionId?: bigint
supportedCodecs?: number[]
}
| { type: 'transport.write'; acks: WriteAck[] }
| { type: 'transport.token' }
| { type: 'transport.ended' }
| { type: 'transport.error'; error: unknown }
export type TransportEffect =
| { type: 'transport.effect.open_stream' }
| { type: 'transport.effect.close_stream' }
| { type: 'transport.effect.finalize'; reason: unknown }
// Emitted to the writer FSM, which ingests these as writer events.
export type TransportOutput =
| {
type: 'transport.stream.init_response'
sessionId: string
lastSeqNo: bigint
partitionId?: bigint
supportedCodecs?: number[]
}
| { type: 'transport.stream.write_response'; acks: WriteAck[] }
| { type: 'transport.stream.token_response' }
| { type: 'transport.stream.disconnected'; error?: unknown }
type TransportRuntime = TransitionRuntime<TransportState, TransportEvent, TransportOutput>
let toInitResponse = function toInitResponse(
event: Extract<TransportEvent, { type: 'transport.init' }>
): TransportOutput {
let output: TransportOutput = {
type: 'transport.stream.init_response',
sessionId: event.sessionId,
lastSeqNo: event.lastSeqNo,
}
if (event.partitionId !== undefined) {
output = { ...output, partitionId: event.partitionId }
}
if (event.supportedCodecs !== undefined) {
output = { ...output, supportedCodecs: event.supportedCodecs }
}
return output
}
let disconnect = function disconnect(
error: unknown,
runtime: TransportRuntime
): TransitionResult<TransportState, TransportEffect> {
let output: TransportOutput = { type: 'transport.stream.disconnected' }
if (error !== undefined) {
output = { ...output, error }
}
runtime.emit(output)
return { state: 'disconnected', effects: [{ type: 'transport.effect.close_stream' }] }
}
export let transportTransition = function transportTransition(
_ctx: TransportCtx,
event: TransportEvent,
runtime: TransportRuntime
): TransitionResult<TransportState, TransportEffect> | void {
let state = runtime.state
if (state !== 'closed' && event.type === 'transport.destroy') {
return {
state: 'closed',
effects: [
{ type: 'transport.effect.close_stream' },
{
type: 'transport.effect.finalize',
reason: event.reason ?? new Error('Transport destroyed'),
},
],
}
}
switch (state) {
case 'idle': {
if (event.type === 'transport.connect') {
return { state: 'connecting', effects: [{ type: 'transport.effect.open_stream' }] }
}
if (event.type === 'transport.close') {
return {
state: 'closed',
effects: [
{
type: 'transport.effect.finalize',
reason: new Error('Transport closed'),
},
],
}
}
return
}
case 'connecting':
case 'ready': {
if (event.type === 'transport.connect') {
// Reopen the stream (open_stream disposes the previous one first).
return { state: 'connecting', effects: [{ type: 'transport.effect.open_stream' }] }
}
if (event.type === 'transport.init') {
runtime.emit(toInitResponse(event))
return state === 'ready' ? undefined : { state: 'ready' }
}
if (event.type === 'transport.write') {
runtime.emit({ type: 'transport.stream.write_response', acks: event.acks })
return
}
if (event.type === 'transport.token') {
runtime.emit({ type: 'transport.stream.token_response' })
return
}
if (event.type === 'transport.ended') {
return disconnect(undefined, runtime)
}
if (event.type === 'transport.error') {
return disconnect(event.error, runtime)
}
Eif (event.type === 'transport.close') {
return {
state: 'closed',
effects: [
{ type: 'transport.effect.close_stream' },
{
type: 'transport.effect.finalize',
reason: new Error('Transport closed'),
},
],
}
}
return
}
case 'disconnected': {
if (event.type === 'transport.connect') {
return { state: 'connecting', effects: [{ type: 'transport.effect.open_stream' }] }
}
if (event.type === 'transport.close') {
return {
state: 'closed',
effects: [
{
type: 'transport.effect.finalize',
reason: new Error('Transport closed'),
},
],
}
}
return
}
case 'closed':
return
default: {
let exhaustive: never = state
fail(`Unhandled transport state: ${exhaustive}`)
}
}
}
|