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 | 12x 52196x 52196x 51087x 52196x 12x 1804x 1804x 1052x 1804x 1804x 12x 264449x 264449x 4x 264445x 52289x 52287x 2x 1x 1x 107015x 56x 106959x 52196x 52196x 54763x 1210x 1210x 53553x 210x 210x 53343x 752x 52591x 1052x 51539x 51539x 1805x 1057x 748x 747x 1x 103336x | 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 }
| { 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
}
| { 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 }
}
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}`)
}
}
}
|