Skip to content

refactor(relay): collapse the duplicated FrameDecoder into one shared module #12078

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
nwparker merged 1 commit into from
Aug 2, 2026
Merged

refactor(relay): collapse the duplicated FrameDecoder into one shared module #12078

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions src/main/ssh/relay-protocol.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ import {
FRAME_DECODER_MAX_BYTES_PER_TURN,
FRAME_DECODER_MAX_TURN_MS,
FRAME_DECODER_MAX_RETAINED_BYTES
} from './relay-frame-decoder'
} from '../../shared/relay-frame-decoder'

export {
FrameDecoder,
Expand All @@ -23,7 +23,7 @@ export {
FRAME_DECODER_MAX_TURN_MS,
FRAME_DECODER_MAX_RETAINED_BYTES
}
export type { DecodedFrame, FrameDecoderOptions } from './relay-frame-decoder'
export type { DecodedFrame, FrameDecoderOptions } from '../../shared/relay-frame-decoder'

export const RELAY_VERSION = '0.1.0'
export const RELAY_SENTINEL = `ORCA-RELAY v${RELAY_VERSION} READY\n`
Expand Down
268 changes: 18 additions & 250 deletions src/relay/relay-frame-decoder.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,264 +1,32 @@
import {
containFrameDecoderContinuation,
publishFrameDecoderError,
type DecodedFrame,
type FrameDecoderOptions
} from '../shared/relay-frame-decoder-contract'
import { RelayFrameBuffer } from '../shared/relay-frame-buffer'
import { FrameDecoder as SharedFrameDecoder } from '../shared/relay-frame-decoder'
import type { DecodedFrame, FrameDecoderOptions } from '../shared/relay-frame-decoder-contract'

export {
HEADER_LENGTH,
MAX_MESSAGE_SIZE,
FRAME_DECODER_MAX_FRAMES_PER_TURN,
FRAME_DECODER_MAX_BYTES_PER_TURN,
FRAME_DECODER_MAX_TURN_MS,
FRAME_DECODER_MAX_RETAINED_BYTES
} from '../shared/relay-frame-decoder'
export {
FrameDecoderContinuationError,
type DecodedFrame,
type FrameDecoderOptions
} from '../shared/relay-frame-decoder-contract'

export const HEADER_LENGTH = 13
export const MAX_MESSAGE_SIZE = 16 * 1024 * 1024
export const FRAME_DECODER_MAX_FRAMES_PER_TURN = 64
export const FRAME_DECODER_MAX_BYTES_PER_TURN = MAX_MESSAGE_SIZE + HEADER_LENGTH
export const FRAME_DECODER_MAX_TURN_MS = 4,
FRAME_DECODER_MAX_RETAINED_BYTES = MAX_MESSAGE_SIZE + HEADER_LENGTH + 1024 * 1024

export class FrameDecoder {
private readonly buffer = new RelayFrameBuffer()
private oversizedPayloadBytesRemaining = 0
private onFrame: (frame: DecodedFrame) => void
private onError: (err: Error) => void
private maxFramesPerTurn: number
private maxBytesPerTurn: number
private maxTurnMs: number
private now: () => number
private schedule: (callback: () => void) => unknown
private cancelScheduled: (handle: unknown) => void
private pause: (() => void) | null
private resume: (() => void) | null
private continuationHandle: unknown
private continuationHandleAssigned = false
private continuationScheduled = false
private paused = false
private draining = false
private generation = 0

// Why: the relay runs standalone on remote hosts with no renderer to surface
// decode faults, so an omitted handler must still reach stderr.
export class FrameDecoder extends SharedFrameDecoder {
constructor(
onFrame: (frame: DecodedFrame) => void,
onError?: (err: Error) => void,
options: FrameDecoderOptions = {}
) {
this.onFrame = onFrame
this.onError = onError ?? ((error) => process.stderr.write(`[relay] ${error.message}\n`))
this.maxFramesPerTurn = positiveLimit(
options.maxFramesPerTurn,
FRAME_DECODER_MAX_FRAMES_PER_TURN
super(
onFrame,
onError ?? ((error) => process.stderr.write(`[relay] ${error.message}\n`)),
options
)
this.maxBytesPerTurn = positiveLimit(options.maxBytesPerTurn, FRAME_DECODER_MAX_BYTES_PER_TURN)
this.maxTurnMs = positiveLimit(options.maxTurnMs, FRAME_DECODER_MAX_TURN_MS)
this.now = options.now ?? Date.now
this.schedule = options.schedule ?? ((callback) => setImmediate(callback))
this.cancelScheduled =
options.cancelScheduled ?? ((handle) => clearImmediate(handle as NodeJS.Immediate))
this.pause = options.pause ?? null
this.resume = options.resume ?? null
}

feed(chunk: Buffer | Uint8Array): void {
const buf = Buffer.isBuffer(chunk)
? chunk
: Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength)
const retained = this.buffer.length + buf.length
if (retained > FRAME_DECODER_MAX_RETAINED_BYTES) {
this.reset()
publishFrameDecoderError(
this.onError,
new Error(`Frame decoder retained-input limit exceeded: ${retained}`)
)
return
}
if (buf.length > 0) {
this.buffer.append(buf)
}
if (!this.draining && !this.continuationScheduled) {
this.drainTurn()
}
}

reset(): void {
this.generation += 1
this.cancelContinuation()
this.buffer.clear()
this.oversizedPayloadBytesRemaining = 0
this.releasePause()
}

drain(): Buffer {
const out = this.buffer.drain()
this.reset()
return out
}

private drainTurn(): void {
if (this.draining) {
return
}
this.draining = true
const generation = this.generation
const startedAt = this.now()
let frames = 0
let bytes = 0

try {
while (generation === this.generation) {
if (
frames >= this.maxFramesPerTurn ||
bytes >= this.maxBytesPerTurn ||
(frames > 0 && this.now() - startedAt >= this.maxTurnMs)
) {
break
}
const discarded = this.discardOversizedPayload(bytes)
if (discarded > 0) {
bytes += discarded
continue
}
if (this.buffer.length < HEADER_LENGTH) {
break
}
const header = this.buffer.peek(HEADER_LENGTH)
const length = header.readUInt32BE(9)
if (length > MAX_MESSAGE_SIZE) {
this.buffer.discard(HEADER_LENGTH)
this.oversizedPayloadBytesRemaining = length
bytes += HEADER_LENGTH
publishFrameDecoderError(
this.onError,
new Error(`Frame payload too large: ${length} bytes — discarded`)
)
continue
}
const totalLength = HEADER_LENGTH + length
if (this.buffer.length < totalLength) {
break
}
if (frames > 0 && bytes + totalLength > this.maxBytesPerTurn) {
break
}
const framed = this.buffer.take(totalLength)
frames += 1
bytes += totalLength
this.onFrame({
type: framed[0],
id: framed.readUInt32BE(1),
ack: framed.readUInt32BE(5),
payload: framed.subarray(HEADER_LENGTH, totalLength)
})
}
} finally {
this.draining = false
}

if (generation !== this.generation) {
return
}
if (this.hasRunnableWork()) {
this.scheduleContinuation()
} else {
this.releasePause()
}
}

private discardOversizedPayload(bytes: number): number {
if (this.oversizedPayloadBytesRemaining === 0 || this.buffer.length === 0) {
return 0
}
const discarded = Math.min(
this.oversizedPayloadBytesRemaining,
this.buffer.length,
Math.max(1, this.maxBytesPerTurn - bytes)
)
this.buffer.discard(discarded)
this.oversizedPayloadBytesRemaining -= discarded
return discarded
}

private hasRunnableWork(): boolean {
if (this.oversizedPayloadBytesRemaining > 0) {
return this.buffer.length > 0
}
if (this.buffer.length < HEADER_LENGTH) {
return false
}
const length = this.buffer.peek(HEADER_LENGTH).readUInt32BE(9)
return length > MAX_MESSAGE_SIZE || this.buffer.length >= HEADER_LENGTH + length
}

private scheduleContinuation(): void {
if (this.continuationScheduled) {
return
}
const generation = this.generation
this.continuationScheduled = true
try {
this.acquirePause()
} catch (error) {
this.continuationScheduled = false
throw error
}
if (generation !== this.generation) {
this.continuationScheduled = false
return
}
try {
this.continuationHandle = this.schedule(() => {
if (!this.continuationScheduled || generation !== this.generation) {
return
}
this.continuationScheduled = false
this.continuationHandleAssigned = false
this.continuationHandle = undefined
try {
this.drainTurn()
} catch (error) {
containFrameDecoderContinuation(() => this.reset(), this.onError, error)
}
})
this.continuationHandleAssigned = true
} catch (error) {
this.continuationScheduled = false
this.continuationHandle = undefined
this.releasePause()
throw error
}
}

private cancelContinuation(): void {
if (!this.continuationScheduled) {
return
}
this.continuationScheduled = false
if (this.continuationHandleAssigned) {
this.cancelScheduled(this.continuationHandle)
}
this.continuationHandleAssigned = false
this.continuationHandle = undefined
}

private acquirePause(): void {
if (!this.paused) {
this.paused = true
try {
this.pause?.()
} catch (error) {
this.paused = false
throw error
}
}
}

private releasePause(): void {
if (this.paused) {
this.paused = false
this.resume?.()
}
}
}

const positiveLimit = (value: number | undefined, fallback: number): number =>
value !== undefined && Number.isFinite(value) && value > 0 ? value : fallback
6 changes: 3 additions & 3 deletions src/main/ssh/relay-frame-decoder.ts → src/shared/relay-frame-decoder.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,13 +3,13 @@ import {
publishFrameDecoderError,
type DecodedFrame,
type FrameDecoderOptions
} from '../../shared/relay-frame-decoder-contract'
import { RelayFrameBuffer } from '../../shared/relay-frame-buffer'
} from './relay-frame-decoder-contract'
import { RelayFrameBuffer } from './relay-frame-buffer'
export {
FrameDecoderContinuationError,
type DecodedFrame,
type FrameDecoderOptions
} from '../../shared/relay-frame-decoder-contract'
} from './relay-frame-decoder-contract'

export const HEADER_LENGTH = 13
export const MAX_MESSAGE_SIZE = 16 * 1024 * 1024
Expand Down