xiaoxue_admin
/
orca
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

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

src/relay/relay-frame-decoder.ts and src/main/ssh/relay-frame-decoder.ts
were 264 identical lines apart from one default: the relay logs decode
faults to stderr when no handler is supplied, the SSH side stays silent.
Two copies of framing logic is exactly where a wire-format fix lands in one
and not the other.

The decoder's contract and buffer already live in src/shared, so the class
joins them there. The relay keeps a thin subclass that supplies its stderr
default, preserving behaviour for the call sites that omit onError. The
SSH copy is deleted and relay-protocol.ts points at shared directly.

Verified: pnpm typecheck, 102 tests across the 9 framing/backpressure/
handshake suites, and `pnpm build:relay` for all six platform targets plus
the WSL hook relay — the standalone bundle has no new dependencies.
This commit is contained in:
Neil 2026-08-02 00:58:10 -07:00 committed by GitHub
parent a14ada15e5
commit 673d7ca926
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 23 additions and 255 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/main/ssh/relay-protocol.ts
View File

@ -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,
@ -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`

268
src/relay/relay-frame-decoder.ts
View File

@ -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
src/main/ssh/relay-frame-decoder.ts → src/shared/relay-frame-decoder.ts
View File

@ -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