From db0f74ddc68fb1c2c0a468ada5c8d7cb586cbd5c Mon Sep 17 00:00:00 2001 From: Neil <4138956+nwparker@users.noreply.github.com> Date: Sun, 2 Aug 2026 00:19:46 -0700 Subject: [PATCH] refactor(relay): collapse the duplicated FrameDecoder into one shared module MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- src/main/ssh/relay-protocol.ts | 4 +- src/relay/relay-frame-decoder.ts | 268 ++---------------- .../ssh => shared}/relay-frame-decoder.ts | 6 +- 3 files changed, 23 insertions(+), 255 deletions(-) rename src/{main/ssh => shared}/relay-frame-decoder.ts (97%) diff --git a/src/main/ssh/relay-protocol.ts b/src/main/ssh/relay-protocol.ts index 4c5e258b01a..da61921ca6a 100644 --- a/src/main/ssh/relay-protocol.ts +++ b/src/main/ssh/relay-protocol.ts @@ -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` diff --git a/src/relay/relay-frame-decoder.ts b/src/relay/relay-frame-decoder.ts index a2c898f017e..f0466bfb87b 100644 --- a/src/relay/relay-frame-decoder.ts +++ b/src/relay/relay-frame-decoder.ts @@ -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 diff --git a/src/main/ssh/relay-frame-decoder.ts b/src/shared/relay-frame-decoder.ts similarity index 97% rename from src/main/ssh/relay-frame-decoder.ts rename to src/shared/relay-frame-decoder.ts index 6ee4d2f2b14..22a1c348185 100644 --- a/src/main/ssh/relay-frame-decoder.ts +++ b/src/shared/relay-frame-decoder.ts @@ -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