diff --git a/src/main/providers/filesystem-provider-contract.ts b/src/main/providers/filesystem-provider-contract.ts index 076b0ed8ee2..fb266bf1490 100644 --- a/src/main/providers/filesystem-provider-contract.ts +++ b/src/main/providers/filesystem-provider-contract.ts @@ -26,8 +26,10 @@ export type FileReadLimits = { export type FileRangeReadResult = { /** Raw bytes for `[position, position + bytesRead)`. `bytesRead < length` - * always means end of file: the host loops internally, and an over-cap - * request is rejected rather than clamped. */ + * always means end of file: the host loops until the window is filled, and + * an over-cap request is rejected rather than clamped, so a short read is + * never a partial syscall or a silently narrowed window. A `position` at or + * past EOF yields `bytesRead === 0`. */ bytes: Buffer bytesRead: number } @@ -49,7 +51,11 @@ export type IFilesystemProvider = { /** Positional read. Optional because an older remote host cannot serve one. * Strict by design: it throws `FileRangeReadUnsupportedError` rather than * silently degrading, so a caller cannot accidentally pay a whole-file - * transfer per chunk while tailing. */ + * transfer per chunk while tailing. + * + * `position` must be a non-negative safe integer and `length` a positive one + * no larger than `MAX_FILE_RANGE_READ_BYTES`; anything else throws + * `FileRangeReadRequestError` without reaching the host. */ readFileRange?( filePath: string, position: number, diff --git a/src/main/providers/ssh-filesystem-provider-capabilities.test.ts b/src/main/providers/ssh-filesystem-provider-capabilities.test.ts index 426c89f3054..2a1f574449f 100644 --- a/src/main/providers/ssh-filesystem-provider-capabilities.test.ts +++ b/src/main/providers/ssh-filesystem-provider-capabilities.test.ts @@ -1,5 +1,8 @@ import { describe, expect, it, vi } from 'vitest' -import { probeSshQuickOpenSearchCapability } from './ssh-filesystem-provider-capabilities' +import { + probeSshQuickOpenSearchCapability, + probeSshRangedReadCapability +} from './ssh-filesystem-provider-capabilities' import { JsonRpcErrorCode } from '../ssh/relay-protocol' describe('SSH Quick Open capability probe', () => { @@ -34,3 +37,42 @@ describe('SSH Quick Open capability probe', () => { ) }) }) + +describe('SSH filesystem capability document', () => { + // fs.getCapabilities answers every feature at once. Probing per feature would + // spend an extra round trip per connection for a document already in hand. + it('is fetched once for every feature probe on a connection', async () => { + const mux = { + request: vi.fn().mockResolvedValue({ quickOpenSearchVersion: 1, rangedReadVersion: 1 }) + } + await expect(probeSshQuickOpenSearchCapability(mux as never)).resolves.toBe(true) + await expect(probeSshRangedReadCapability(mux as never)).resolves.toBe(true) + expect(mux.request).toHaveBeenCalledTimes(1) + }) + + it('reads each feature independently off the shared document', async () => { + const mux = { request: vi.fn().mockResolvedValue({ quickOpenSearchVersion: 1 }) } + await expect(probeSshQuickOpenSearchCapability(mux as never)).resolves.toBe(true) + await expect(probeSshRangedReadCapability(mux as never)).resolves.toBe(false) + }) + + // A transport failure is not a verdict about the host, so it must not be the + // cached answer for the rest of the connection's life. + it('retries after a failed fetch instead of caching the failure', async () => { + const mux = { + request: vi + .fn() + .mockRejectedValueOnce(new Error('connection closed')) + .mockResolvedValue({ rangedReadVersion: 1 }) + } + await expect(probeSshRangedReadCapability(mux as never)).rejects.toThrow('connection closed') + await expect(probeSshRangedReadCapability(mux as never)).resolves.toBe(true) + expect(mux.request).toHaveBeenCalledTimes(2) + }) + + it('treats a non-object response as no capabilities', async () => { + const mux = { request: vi.fn().mockResolvedValue('yes') } + await expect(probeSshRangedReadCapability(mux as never)).resolves.toBe(false) + await expect(probeSshQuickOpenSearchCapability(mux as never)).resolves.toBe(false) + }) +}) diff --git a/src/main/providers/ssh-filesystem-provider-capabilities.ts b/src/main/providers/ssh-filesystem-provider-capabilities.ts index 95839a9090d..86451e5228b 100644 --- a/src/main/providers/ssh-filesystem-provider-capabilities.ts +++ b/src/main/providers/ssh-filesystem-provider-capabilities.ts @@ -2,75 +2,66 @@ import type { SshChannelMultiplexer } from '../ssh/ssh-channel-multiplexer' import { isMethodNotFoundError } from '../ssh/ssh-filesystem-stream-reader' import { waitForSshCapabilityProbe } from './ssh-capability-probe-waiter' -const quickOpenSearchSupport = new WeakMap>() +/** `null` means the host has no capability document at all: it predates + * `fs.getCapabilities`, or answered with something that is not an object. + * Either way every feature reads as unsupported. */ +type RelayFsCapabilities = Record | null + +const capabilitiesByMux = new WeakMap>() + +/** One `fs.getCapabilities` per multiplexer, shared by every feature probe -- + * the response is a single document, so fetching it once per feature would + * spend a round trip for nothing. Keyed by the multiplexer, which is replaced + * on reconnect, so a host upgraded behind a reconnect is re-probed. + * + * A failed fetch is evicted so the next probe retries; a resolved one is not, + * because a relay does not gain methods without a new connection. */ +function readSshFsCapabilities( + mux: SshChannelMultiplexer, + signal?: AbortSignal +): Promise { + const cached = capabilitiesByMux.get(mux) + const probe = + cached ?? + mux + .request('fs.getCapabilities', undefined, { timeoutMs: 5_000 }) + .then((result) => + typeof result === 'object' && result !== null ? (result as Record) : null + ) + .catch((error) => { + if (isMethodNotFoundError(error)) { + return null + } + throw error + }) + if (!cached) { + capabilitiesByMux.set(mux, probe) + } + return waitForSshCapabilityProbe(probe, signal).then( + (capabilities) => capabilities, + (error) => { + if (!signal?.aborted && capabilitiesByMux.get(mux) === probe) { + capabilitiesByMux.delete(mux) + } + throw error + } + ) +} export function probeSshQuickOpenSearchCapability( mux: SshChannelMultiplexer, signal?: AbortSignal ): Promise { - const cached = quickOpenSearchSupport.get(mux) - const probe = - cached ?? - mux - .request('fs.getCapabilities', undefined, { timeoutMs: 5_000 }) - .then((result) => { - const version = (result as { quickOpenSearchVersion?: unknown } | null) - ?.quickOpenSearchVersion - return version === 1 - }) - .catch((error) => { - if (isMethodNotFoundError(error)) { - return false - } - throw error - }) - if (!cached) { - quickOpenSearchSupport.set(mux, probe) - } - return waitForSshCapabilityProbe(probe, signal).then( - (supported) => supported, - (error) => { - if (!signal?.aborted && quickOpenSearchSupport.get(mux) === probe) { - quickOpenSearchSupport.delete(mux) - } - throw error - } + return readSshFsCapabilities(mux, signal).then( + (capabilities) => capabilities?.quickOpenSearchVersion === 1 ) } -const rangedReadSupport = new WeakMap>() - -/** Mirrors the quick-open probe: one cached `fs.getCapabilities` per mux rather - * than eating a guaranteed -32601 on every ranged read against an old relay. */ export function probeSshRangedReadCapability( mux: SshChannelMultiplexer, signal?: AbortSignal ): Promise { - const cached = rangedReadSupport.get(mux) - const probe = - cached ?? - mux - .request('fs.getCapabilities', undefined, { timeoutMs: 5_000 }) - .then((result) => { - const version = (result as { rangedReadVersion?: unknown } | null)?.rangedReadVersion - return version === 1 - }) - .catch((error) => { - if (isMethodNotFoundError(error)) { - return false - } - throw error - }) - if (!cached) { - rangedReadSupport.set(mux, probe) - } - return waitForSshCapabilityProbe(probe, signal).then( - (supported) => supported, - (error) => { - if (!signal?.aborted && rangedReadSupport.get(mux) === probe) { - rangedReadSupport.delete(mux) - } - throw error - } + return readSshFsCapabilities(mux, signal).then( + (capabilities) => capabilities?.rangedReadVersion === 1 ) } diff --git a/src/main/providers/ssh-filesystem-provider-range.test.ts b/src/main/providers/ssh-filesystem-provider-range.test.ts index 621c23efdec..a947d53845e 100644 --- a/src/main/providers/ssh-filesystem-provider-range.test.ts +++ b/src/main/providers/ssh-filesystem-provider-range.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it, vi } from 'vitest' import { SshFilesystemProvider } from './ssh-filesystem-provider' import { FileRangeReadUnsupportedError } from './filesystem-provider-contract' +import { FileRangeReadRequestError, MAX_FILE_RANGE_READ_BYTES } from '../../shared/file-range-read' import type { SshChannelMultiplexer } from '../ssh/ssh-channel-multiplexer' function methodNotFound(): Error { @@ -31,6 +32,26 @@ describe('SshFilesystemProvider.readFileRange', () => { ) }) + it('reports a short window as read without complaint', async () => { + const payload = Buffer.from('tail') + const request = vi.fn().mockResolvedValue({ + base64: payload.toString('base64'), + bytesRead: payload.length + }) + const result = await providerWith(request).readFileRange('/x.jsonl', 0, 64) + expect(result.bytesRead).toBe(4) + expect(result.bytes).toEqual(payload) + }) + + it('forwards an abort signal to the multiplexer', async () => { + const controller = new AbortController() + const request = vi.fn().mockResolvedValue({ base64: '', bytesRead: 0 }) + await providerWith(request).readFileRange('/x.jsonl', 0, 4, { signal: controller.signal }) + expect(request).toHaveBeenCalledWith('fs.readFileRange', expect.anything(), { + signal: controller.signal + }) + }) + // Throwing beats a whole-file fallback here: a tailing caller issues several // reads per snapshot, so falling back per call is quadratic on a growing file. it('throws a typed unsupported error against a relay without the method', async () => { @@ -45,6 +66,24 @@ describe('SshFilesystemProvider.readFileRange', () => { await expect(providerWith(request).readFileRange('/x.jsonl', 0, 4)).rejects.toThrow(/EACCES/) }) + // The client runs the host's own validator, so a request the host would + // refuse fails as a typed error instead of an opaque -32000 a round trip later. + describe('request validation', () => { + it.each([ + ['a negative position', -1, 4], + ['a fractional position', 0.5, 4], + ['a zero length', 0, 0], + ['a negative length', 0, -1], + ['an over-cap length', 0, MAX_FILE_RANGE_READ_BYTES + 1] + ])('rejects %s without a round trip', async (_label, position, length) => { + const request = vi.fn() + await expect( + providerWith(request).readFileRange('/x.jsonl', position, length) + ).rejects.toBeInstanceOf(FileRangeReadRequestError) + expect(request).not.toHaveBeenCalled() + }) + }) + // The remote is untrusted for framing: a count disagreeing with the payload // would shift every downstream offset while looking like a success. it('rejects a byte count that disagrees with the payload', async () => { @@ -99,4 +138,13 @@ describe('SshFilesystemProvider.supportsFileRangeRead', () => { await provider.supportsFileRangeRead() expect(request).toHaveBeenCalledTimes(1) }) + + // The cache is keyed by the multiplexer, which is replaced on reconnect, so a + // second connection must not inherit the first host's answer. + it('does not share the answer across connections', async () => { + const legacy = vi.fn().mockResolvedValue({ quickOpenSearchVersion: 1 }) + const current = vi.fn().mockResolvedValue({ rangedReadVersion: 1 }) + await expect(providerWith(legacy).supportsFileRangeRead()).resolves.toBe(false) + await expect(providerWith(current).supportsFileRangeRead()).resolves.toBe(true) + }) }) diff --git a/src/main/providers/ssh-filesystem-provider.ts b/src/main/providers/ssh-filesystem-provider.ts index 436bfb645d8..c21d42f65b8 100644 --- a/src/main/providers/ssh-filesystem-provider.ts +++ b/src/main/providers/ssh-filesystem-provider.ts @@ -16,6 +16,7 @@ import { } from './ssh-filesystem-provider-watch' import type { IFilesystemProvider, + FileRangeReadResult, FileReadLimits, FileStat, FileReadResult, @@ -46,9 +47,6 @@ export class SshFilesystemProvider implements IFilesystemProvider { private disposed = false private loggedStreamFallback = false readonly downloadFolder?: IFilesystemProvider['downloadFolder'] - // Optional capabilities are wired in the constructor, matching downloadFolder. - readonly readFileRange: NonNullable - readonly supportsFileRangeRead: NonNullable constructor( connectionId: string, @@ -72,11 +70,6 @@ export class SshFilesystemProvider implements IFilesystemProvider { }) } - this.readFileRange = (filePath, position, length, options) => - readSshFileRange(this.mux, filePath, position, length, options?.signal) - this.supportsFileRangeRead = (options) => - probeSshRangedReadCapability(this.mux, options?.signal) - this.unsubscribeNotifications = mux.onNotification((method, params) => routeSshFilesystemWatchNotification(this.watchListeners, method, params) ) @@ -127,6 +120,19 @@ export class SshFilesystemProvider implements IFilesystemProvider { } } + readFileRange( + filePath: string, + position: number, + length: number, + options?: { signal?: AbortSignal } + ): Promise { + return readSshFileRange(this.mux, filePath, position, length, options?.signal) + } + + supportsFileRangeRead(options?: { signal?: AbortSignal }): Promise { + return probeSshRangedReadCapability(this.mux, options?.signal) + } + readTerminalArtifact( filePath: string, options: TerminalArtifactAccessOptions diff --git a/src/main/providers/ssh-filesystem-range-read.ts b/src/main/providers/ssh-filesystem-range-read.ts index 64413dac708..77ec0872363 100644 --- a/src/main/providers/ssh-filesystem-range-read.ts +++ b/src/main/providers/ssh-filesystem-range-read.ts @@ -1,5 +1,6 @@ import type { SshChannelMultiplexer } from '../ssh/ssh-channel-multiplexer' import { isMethodNotFoundError } from '../ssh/ssh-filesystem-stream-reader' +import { validateFileRangeRequest } from '../../shared/file-range-read' import { FileRangeReadUnsupportedError, type FileRangeReadResult @@ -8,12 +9,19 @@ import { export async function readSshFileRange( mux: SshChannelMultiplexer, filePath: string, - position: number, - length: number, + rawPosition: number, + rawLength: number, signal?: AbortSignal ): Promise { + // Same validator the host runs, so an out-of-contract request fails locally + // as a typed error instead of costing a round trip and coming back as an + // opaque -32000. + const { position, length } = validateFileRangeRequest(rawPosition, rawLength) let result: unknown try { + // No timeoutMs: the multiplexer's 30s default is right for a bounded + // MAX_FILE_RANGE_READ_BYTES window on a slow link. The capability probe + // pins a shorter one because it gates a UI decision, not a data read. result = await mux.request( 'fs.readFileRange', { filePath, position, length }, diff --git a/src/relay/fs-handler-file-range-dispatch.test.ts b/src/relay/fs-handler-file-range-dispatch.test.ts new file mode 100644 index 00000000000..b408b6360da --- /dev/null +++ b/src/relay/fs-handler-file-range-dispatch.test.ts @@ -0,0 +1,160 @@ +import { mkdtemp, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { FsHandler } from './fs-handler' +import { RelayContext } from './context' +import { RelayDispatcher } from './dispatcher' +import { encodeJsonRpcFrame, RelayErrorCode } from './protocol' +import { FileRangeReadRequestError, MAX_FILE_RANGE_READ_BYTES } from '../shared/file-range-read' + +/** Minimal dispatcher: this suite only needs the registered request handlers. + * The full harness lives in fs-handler.test.ts, which is at its line budget. */ +function createHandlerUnderTest() { + const requests = new Map) => Promise>() + const dispatcher = { + onRequest: (method: string, handler: (params: Record) => Promise) => { + requests.set(method, handler) + }, + onNotification: vi.fn(), + notify: vi.fn(), + notifyClient: vi.fn(), + onClientDetached: vi.fn(() => () => {}) + } + const handler = new FsHandler(dispatcher as unknown as RelayDispatcher, new RelayContext()) + const call = (method: string, params: Record): Promise => { + const registered = requests.get(method) + if (!registered) { + throw new Error(`No handler for ${method}`) + } + return registered(params) + } + return { handler, call } +} + +let root: string +let filePath: string +let underTest: ReturnType + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-relay-range-dispatch-')) + filePath = join(root, 'data.jsonl') + await writeFile(filePath, '0123456789') + underTest = createHandlerUnderTest() +}) + +afterEach(async () => { + underTest.handler.dispose() + await rm(root, { recursive: true, force: true }) +}) + +describe('fs.readFileRange dispatch', () => { + it('serves the requested window', async () => { + const result = (await underTest.call('fs.readFileRange', { + filePath, + position: 2, + length: 3 + })) as { base64: string; bytesRead: number } + expect(Buffer.from(result.base64, 'base64').toString('utf8')).toBe('234') + expect(result.bytesRead).toBe(3) + }) + + // Params reach the handler as an unschema-d bag, so a missing or wrong-typed + // path must be a stated refusal rather than a TypeError from expandTilde. + it.each([ + ['a missing filePath', {}], + ['a non-string filePath', { filePath: 42 }], + ['an empty filePath', { filePath: '' }] + ])('rejects %s', async (_label, params) => { + await expect( + underTest.call('fs.readFileRange', { position: 0, length: 4, ...params }) + ).rejects.toBeInstanceOf(FileRangeReadRequestError) + }) + + it('rejects an out-of-contract offset before touching the file', async () => { + await expect( + underTest.call('fs.readFileRange', { filePath, position: -1, length: 4 }) + ).rejects.toBeInstanceOf(FileRangeReadRequestError) + }) +}) + +// A direct call to readRelayFileRange never meets the writer, which is where an +// over-wide window actually dies: a response frame past the control lane's +// budget is demoted to `legacy-response` and refused as ResponseOverCapacity. +// This drives a full-cap read through the real dispatcher instead. +describe('fs.readFileRange over the real dispatcher', () => { + function decodePayload(frame: Buffer): Record { + const length = frame.readUInt32BE(9) + return JSON.parse(frame.subarray(13, 13 + length).toString('utf-8')) + } + + it('delivers a full-cap window as a result, not a capacity error', async () => { + const contents = Buffer.allocUnsafe(MAX_FILE_RANGE_READ_BYTES) + for (let i = 0; i < contents.length; i++) { + contents[i] = (i * 37) % 256 + } + const capPath = join(root, 'full-cap.bin') + await writeFile(capPath, contents) + + const frames: Buffer[] = [] + let closes = 0 + const dispatcher = new RelayDispatcher( + (data: Buffer) => { + frames.push(Buffer.from(data)) + return true + }, + { + writableHighWaterMark: () => 64 * 1024, + writableLength: () => 0, + close: () => { + closes++ + } + } + ) + const handler = new FsHandler(dispatcher, new RelayContext()) + try { + dispatcher.feed( + encodeJsonRpcFrame( + { + jsonrpc: '2.0', + id: 91, + method: 'fs.readFileRange', + params: { filePath: capPath, position: 0, length: MAX_FILE_RANGE_READ_BYTES } + }, + 1, + 0 + ) + ) + for (let i = 0; i < 200 && frames.length === 0; i++) { + await new Promise((resolve) => setImmediate(resolve)) + } + expect(frames).toHaveLength(1) + const response = decodePayload(frames[0]) as { + id: number + error?: { code: number } + result?: { base64: string; bytesRead: number } + } + expect(response.id).toBe(91) + expect(response.error?.code).not.toBe(RelayErrorCode.ResponseOverCapacity) + expect(response.error).toBeUndefined() + expect(response.result?.bytesRead).toBe(MAX_FILE_RANGE_READ_BYTES) + expect(Buffer.from(response.result?.base64 ?? '', 'base64').equals(contents)).toBe(true) + expect(closes).toBe(0) + } finally { + handler.dispose() + dispatcher.dispose() + } + }) +}) + +describe('fs.getCapabilities', () => { + // Rule 1 of docs/reference/remote-wire-compatibility.md: the ranged-read flag + // is additive. Dropping the pre-existing key would strand an older desktop's + // quick-open probe on a host that still serves it. + it('advertises ranged reads without dropping the existing capability', async () => { + await expect(underTest.call('fs.getCapabilities', {})).resolves.toEqual({ + quickOpenSearchVersion: 1, + rangedReadVersion: 1 + }) + }) +}) diff --git a/src/relay/fs-handler-file-range.test.ts b/src/relay/fs-handler-file-range.test.ts index 3af01a673fe..2ebe0bcab59 100644 --- a/src/relay/fs-handler-file-range.test.ts +++ b/src/relay/fs-handler-file-range.test.ts @@ -2,11 +2,10 @@ import { mkdtemp, rm, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import { afterEach, describe, expect, it } from 'vitest' -import { - MAX_RELAY_RANGE_READ_BYTES, - RelayRangeTooLargeError, - readRelayFileRange -} from './fs-handler-file-range' +import { readRelayFileRange } from './fs-handler-file-range' +import { FileRangeReadRequestError, MAX_FILE_RANGE_READ_BYTES } from '../shared/file-range-read' +import { HEADER_LENGTH, prepareJsonRpcPayload } from './protocol' +import { DISPATCHER_CONTROL_QUEUE_MAX_BYTES } from './dispatcher-writer-admission' let roots: string[] = [] @@ -57,6 +56,28 @@ describe('readRelayFileRange', () => { expect(decode(result.base64)).toHaveLength(0) }) + it('returns zero bytes when the position is exactly at the end', async () => { + const path = await fileWith('abc') + const result = await readRelayFileRange(path, 3, 10) + expect(result.bytesRead).toBe(0) + }) + + // The window is one fixed allocation filled across however many syscalls the + // kernel takes, so every partial read has to land at the right offset in it. + it('assembles a many-page window without misplacing any byte', async () => { + const size = MAX_FILE_RANGE_READ_BYTES + const contents = Buffer.allocUnsafe(size) + for (let i = 0; i < size; i++) { + contents[i] = i % 251 + } + const path = await fileWith(contents) + const result = await readRelayFileRange(path, 1024, size - 1024) + expect(result.bytesRead).toBe(size - 1024) + // .equals(), not toEqual(): a byte-wise diff of a 256 KiB buffer costs + // minutes when it passes. + expect(decode(result.base64).equals(contents.subarray(1024))).toBe(true) + }) + // Base64 rather than utf-8: a range boundary can split a multi-byte sequence, // and a utf-8 round trip would substitute U+FFFD and shift every later offset. it('preserves bytes that split a multi-byte character at both edges', async () => { @@ -73,12 +94,22 @@ describe('readRelayFileRange', () => { expect(decode(result.base64)).toEqual(bytes) }) - // Rejecting beats clamping: a clamped read is indistinguishable from EOF. - it('rejects an over-cap request instead of silently clamping', async () => { - const path = await fileWith('abc') - await expect( - readRelayFileRange(path, 0, MAX_RELAY_RANGE_READ_BYTES + 1) - ).rejects.toBeInstanceOf(RelayRangeTooLargeError) + // The cap is a TRANSPORT budget, not just a memory one. A response frame over + // DISPATCHER_CONTROL_QUEUE_MAX_BYTES is demoted to the `legacy-response` lane, + // which the writer refuses once the producer queue is busy -- so an over-wide + // cap fails with ResponseOverCapacity depending on unrelated load. A full-cap + // window of incompressible bytes must stay inside the control lane. + it('encodes a full-cap window into a frame the control lane always admits', async () => { + const contents = Buffer.allocUnsafe(MAX_FILE_RANGE_READ_BYTES) + for (let i = 0; i < contents.length; i++) { + contents[i] = (i * 37) % 256 + } + const path = await fileWith(contents) + const result = await readRelayFileRange(path, 0, MAX_FILE_RANGE_READ_BYTES) + expect(result.bytesRead).toBe(MAX_FILE_RANGE_READ_BYTES) + const frameBytes = + HEADER_LENGTH + prepareJsonRpcPayload({ jsonrpc: '2.0', id: 4294967295, result }).byteLength + expect(frameBytes).toBeLessThanOrEqual(DISPATCHER_CONTROL_QUEUE_MAX_BYTES) }) it('rejects a missing file', async () => { @@ -86,4 +117,51 @@ describe('readRelayFileRange', () => { roots.push(root) await expect(readRelayFileRange(join(root, 'nope.bin'), 0, 4)).rejects.toThrow() }) + + // Every rejection below happens before the file is opened: these offsets go + // straight into a read syscall and the relay has no request schema. + describe('request validation', () => { + // Rejecting beats clamping: a clamped read is indistinguishable from EOF. + it('rejects an over-cap length instead of silently clamping', async () => { + const path = await fileWith('abc') + await expect( + readRelayFileRange(path, 0, MAX_FILE_RANGE_READ_BYTES + 1) + ).rejects.toBeInstanceOf(FileRangeReadRequestError) + }) + + it('accepts a length exactly at the cap', async () => { + const path = await fileWith('abc') + const result = await readRelayFileRange(path, 0, MAX_FILE_RANGE_READ_BYTES) + expect(result.bytesRead).toBe(3) + }) + + it.each([ + ['a negative position', -1, 4], + ['a fractional position', 1.5, 4], + ['a NaN position', Number.NaN, 4], + ['an infinite position', Number.POSITIVE_INFINITY, 4], + ['a position past the safe-integer range', Number.MAX_SAFE_INTEGER + 2, 4], + ['a zero length', 0, 0], + ['a negative length', 0, -4], + ['a fractional length', 0, 4.5], + ['a NaN length', 0, Number.NaN] + ])('rejects %s', async (_label, position, length) => { + const path = await fileWith('abc') + await expect(readRelayFileRange(path, position, length)).rejects.toBeInstanceOf( + FileRangeReadRequestError + ) + }) + + it.each([ + ['a missing position', undefined, 4], + ['a string position', '0', 4], + ['a missing length', 0, undefined], + ['a string length', 0, '4'] + ])('rejects %s from an unschema-d param bag', async (_label, position, length) => { + const path = await fileWith('abc') + await expect(readRelayFileRange(path, position, length)).rejects.toBeInstanceOf( + FileRangeReadRequestError + ) + }) + }) }) diff --git a/src/relay/fs-handler-file-range.ts b/src/relay/fs-handler-file-range.ts index 56384c3dc91..cba1ed9f594 100644 --- a/src/relay/fs-handler-file-range.ts +++ b/src/relay/fs-handler-file-range.ts @@ -1,43 +1,29 @@ import { open } from 'node:fs/promises' +import { validateFileRangeRequest } from '../shared/file-range-read' +import { readFullStreamChunk } from './fs-handler-file-read' -/** Largest range one JSON-RPC response may carry. Requests above this are - * REJECTED rather than clamped: a silently shortened read is indistinguishable - * from EOF or a truncation, and callers page by advancing `position`. */ -export const MAX_RELAY_RANGE_READ_BYTES = 4 * 1024 * 1024 - -export class RelayRangeTooLargeError extends Error { - constructor(requested: number) { - super( - `Requested range of ${requested} bytes exceeds the ${MAX_RELAY_RANGE_READ_BYTES}-byte limit` - ) - this.name = 'RelayRangeTooLargeError' - } -} - -/** Positional read for tailing an append-only file. Loops until `length` is - * satisfied or the file genuinely ends, so a short result always means EOF and - * never a partial syscall. Base64 because the payload is arbitrary bytes and - * may split a UTF-8 sequence at either edge. */ +/** Positional read for tailing an append-only file. Fills the window until + * `length` is satisfied or the file genuinely ends, so a short result always + * means EOF and never a partial syscall. Base64 because the payload is + * arbitrary bytes and may split a UTF-8 sequence at either edge. + * + * Validates here rather than at the dispatcher: this is the function that puts + * a caller-supplied offset into a read syscall, so it is the boundary that has + * to hold even if a future call site forgets. */ export async function readRelayFileRange( filePath: string, - position: number, - length: number + rawPosition: unknown, + rawLength: unknown ): Promise<{ base64: string; bytesRead: number }> { - if (length > MAX_RELAY_RANGE_READ_BYTES) { - throw new RelayRangeTooLargeError(length) - } + const { position, length } = validateFileRangeRequest(rawPosition, rawLength) const handle = await open(filePath, 'r') try { - const buffer = Buffer.alloc(length) - let total = 0 - while (total < length) { - const { bytesRead } = await handle.read(buffer, total, length - total, position + total) - if (bytesRead <= 0) { - break - } - total += bytesRead - } - return { base64: buffer.subarray(0, total).toString('base64'), bytesRead: total } + // allocUnsafe over alloc: only `subarray(0, bytesRead)` is ever read back, so + // uninitialised bytes cannot escape, and a tailing poll with a generous + // `length` should not pay a memset over the whole window per call. + const buffer = Buffer.allocUnsafe(length) + const bytesRead = await readFullStreamChunk(handle, buffer, length, position) + return { base64: buffer.subarray(0, bytesRead).toString('base64'), bytesRead } } finally { await handle.close() } diff --git a/src/relay/fs-handler-file-read.ts b/src/relay/fs-handler-file-read.ts index 4a6f663484d..e8b7993fbcd 100644 --- a/src/relay/fs-handler-file-read.ts +++ b/src/relay/fs-handler-file-read.ts @@ -334,7 +334,9 @@ async function pumpChunks( // Why: fs.read() may return fewer bytes than requested before EOF. Fill each // protocol chunk so strict clients reject corruption, not valid short reads. -async function readFullStreamChunk( +// Shared with fs.readFileRange, where the same rule makes a short result mean +// EOF and nothing else. +export async function readFullStreamChunk( handle: StreamChunkReader, buffer: Buffer, length: number, diff --git a/src/relay/fs-handler.ts b/src/relay/fs-handler.ts index cd13a0735db..480f0da52c7 100644 --- a/src/relay/fs-handler.ts +++ b/src/relay/fs-handler.ts @@ -27,6 +27,7 @@ import { import { buildExcludePathPrefixes } from '../shared/quick-open-filter' import { readRelayFileContent, readRelayFileStreamMetadata } from './fs-handler-file-read' import { readRelayFileRange } from './fs-handler-file-range' +import { FileRangeReadRequestError } from '../shared/file-range-read' import { readVerifiedTerminalArtifact, writeVerifiedTerminalArtifact @@ -111,18 +112,13 @@ export class FsHandler { return readRelayFileContent(filePath) } - // Why hand-validated: relay params arrive as raw casts with no schema, and - // both of these land directly in an fd read position. + // Why hand-checked: relay params arrive as raw casts with no schema. The + // offsets are validated inside readRelayFileRange, next to the read syscall. private async readFileRange(params: Record) { - const filePath = expandTilde(params.filePath as string) - const { position, length } = params - if (typeof position !== 'number' || !Number.isSafeInteger(position) || position < 0) { - throw new Error('fs.readFileRange requires a non-negative integer position') + if (typeof params.filePath !== 'string' || params.filePath.length === 0) { + throw new FileRangeReadRequestError('fs.readFileRange requires a filePath') } - if (typeof length !== 'number' || !Number.isSafeInteger(length) || length <= 0) { - throw new Error('fs.readFileRange requires a positive integer length') - } - return readRelayFileRange(filePath, position, length) + return readRelayFileRange(expandTilde(params.filePath), params.position, params.length) } private async readTerminalArtifact(params: Record) { diff --git a/src/shared/file-range-read.ts b/src/shared/file-range-read.ts new file mode 100644 index 00000000000..70cf278525c --- /dev/null +++ b/src/shared/file-range-read.ts @@ -0,0 +1,49 @@ +/** Largest window one `fs.readFileRange` response may carry: `STREAM_CHUNK_SIZE`, + * the house per-frame budget for file bytes. + * + * The binding constraint is the relay writer's admission budget, NOT the 16 MiB + * frame cap. A response frame over `DISPATCHER_CONTROL_QUEUE_MAX_BYTES` (1 MiB) + * is demoted to the `legacy-response` lane, which is refused once the producer + * queue passes `DEFAULT_PRODUCER_QUEUE_MAX_BYTES` (2 MiB) -- so a wide window + * fails with an opaque `ResponseOverCapacity` that depends on what else is + * queued at the time. 256 KiB of raw bytes is ~350 KB of base64, which stays in + * the control lane unconditionally. Bigger transfers belong on the ack-paced + * bulk lane (`fs.readFileStream`), not on a wider cap here. + * + * Requests above it are REJECTED, never clamped: a clamped read is + * indistinguishable from EOF, and callers page by advancing `position`. */ +export const MAX_FILE_RANGE_READ_BYTES = 256 * 1024 + +/** A range request the host will refuse. Separate from a read failure so a + * caller can tell "I asked for the wrong thing" from "the file is unreadable". */ +export class FileRangeReadRequestError extends Error { + constructor(message: string) { + super(message) + this.name = 'FileRangeReadRequestError' + } +} + +/** The single source of truth for what `fs.readFileRange` accepts. Both sides + * call it: the client so an invalid request never costs a round trip, the host + * because `position`/`length` land straight in an fd read and the relay has no + * request schema. Two independent copies would drift into a client that sends + * what the host rejects. */ +export function validateFileRangeRequest( + position: unknown, + length: unknown +): { position: number; length: number } { + if (typeof position !== 'number' || !Number.isSafeInteger(position) || position < 0) { + throw new FileRangeReadRequestError( + 'fs.readFileRange requires a non-negative safe-integer position' + ) + } + if (typeof length !== 'number' || !Number.isSafeInteger(length) || length <= 0) { + throw new FileRangeReadRequestError('fs.readFileRange requires a positive integer length') + } + if (length > MAX_FILE_RANGE_READ_BYTES) { + throw new FileRangeReadRequestError( + `fs.readFileRange length ${length} exceeds the ${MAX_FILE_RANGE_READ_BYTES}-byte limit` + ) + } + return { position, length } +}