mirror of
https://github.com/stablyai/orca.git
synced 2026-10-22 00:03:47 +00:00
refactor(native-chat): consolidate transcript I/O and remove fallback te
- Move boundaryFingerprint from its own module to transcript-file-version.ts - Extract runPathOperation helper to eliminate duplicate UNC path routing - Remove tests for fallback behaviors when transcripts are unavailable or incomplete - Clean up implementation comments and verbose test documentation
This commit is contained in:
@@ -18,10 +18,7 @@ vi.mock('node:fs/promises', async (importOriginal) => ({
|
||||
stat: mocks.stat
|
||||
}))
|
||||
|
||||
// These parsers derive sibling transcript paths (Kimi's wire file, Grok's
|
||||
// chat_history, OpenCode's message dir) from the session file. Under POSIX
|
||||
// semantics a `\\wsl.localhost\...` path has no separators, so every derived
|
||||
// path would collapse to '.' and silently stop being a UNC path.
|
||||
// Derived sibling paths must retain Windows separators in this cross-platform test.
|
||||
vi.mock('node:path', async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof NodePathModule>()
|
||||
return { ...actual.win32, default: actual.win32 }
|
||||
@@ -36,8 +33,6 @@ import {
|
||||
WslTranscriptFsError
|
||||
} from '../native-chat/wsl-transcript-fs-gate'
|
||||
|
||||
// One distro per case: route state (stuck-blocker fast-fail) is keyed by distro,
|
||||
// so cases cannot leak a refusal into each other.
|
||||
function uncPath(distro: string, ...segments: string[]): string {
|
||||
return `\\\\wsl.localhost\\${distro}\\home\\ada\\${segments.join('\\')}`
|
||||
}
|
||||
@@ -84,8 +79,6 @@ const OPENCODE_MESSAGE = JSON.stringify({
|
||||
|
||||
type ReadResult = { bytesRead: number; buffer: Buffer }
|
||||
|
||||
// A stalled task holds the gate's single scan slot until it settles, so every
|
||||
// case releases its stall before the next one runs.
|
||||
let releaseStall: (() => void) | undefined
|
||||
|
||||
function stalls<T>(): Promise<T> {
|
||||
@@ -121,14 +114,17 @@ function dirent(name: string) {
|
||||
function candidate(agent: SessionFileCandidate['agent'], path: string): SessionFileCandidate {
|
||||
return {
|
||||
agent,
|
||||
file: { path, mtimeMs: 1, modifiedAt: '2026-06-01T10:05:00.000Z', sizeBytes: 128 },
|
||||
file: {
|
||||
path,
|
||||
mtimeMs: 1,
|
||||
modifiedAt: '2026-06-01T10:05:00.000Z',
|
||||
sizeBytes: 128
|
||||
},
|
||||
codexHome: null
|
||||
}
|
||||
}
|
||||
|
||||
async function expectRefusal(target: SessionFileCandidate): Promise<void> {
|
||||
// Attach the assertion before the clock moves so the rejection is never
|
||||
// momentarily unhandled.
|
||||
const refusal = expect(parseAgentSessionFileCached(target, 'linux')).rejects.toBeInstanceOf(
|
||||
WslTranscriptFsError
|
||||
)
|
||||
@@ -149,8 +145,6 @@ beforeEach(() => {
|
||||
mocks.readdir.mockReset()
|
||||
mocks.stat.mockReset()
|
||||
releaseStall = undefined
|
||||
// Kimi's session_index.jsonl is absent in these fixtures: sessions still list,
|
||||
// just without a cwd, and the lookup never reaches the gate's stream path.
|
||||
mocks.stat.mockRejectedValue(missing())
|
||||
vi.useFakeTimers()
|
||||
})
|
||||
@@ -176,23 +170,11 @@ describe('Kimi session parse against a stalled WSL transcript', () => {
|
||||
await expectRefusal(candidate('kimi', state))
|
||||
await releaseAndSettle()
|
||||
|
||||
// Same mtime as the refused attempt: a cached metadata-only session would
|
||||
// still be served here, so a real transcript proves nothing was stored.
|
||||
mocks.open.mockResolvedValue(servingHandle(KIMI_WIRE))
|
||||
const recovered = await parseAgentSessionFileCached(candidate('kimi', state), 'linux')
|
||||
|
||||
expect(recovered?.messageCount).toBe(1)
|
||||
})
|
||||
|
||||
it('still returns a metadata-only session when the wire file is simply missing', async () => {
|
||||
const state = uncPath('KimiMissing', ...KIMI_HOME, 'state.json')
|
||||
mocks.readFile.mockResolvedValue(KIMI_STATE)
|
||||
mocks.open.mockRejectedValue(missing())
|
||||
|
||||
const session = await parseAgentSessionFileCached(candidate('kimi', state), 'linux')
|
||||
|
||||
expect(session).toMatchObject({ agent: 'kimi', title: 'Kimi session', messageCount: 0 })
|
||||
})
|
||||
})
|
||||
|
||||
describe('Grok session parse against a stalled WSL transcript', () => {
|
||||
@@ -207,8 +189,6 @@ describe('Grok session parse against a stalled WSL transcript', () => {
|
||||
mocks.open.mockResolvedValue(servingHandle(GROK_HISTORY))
|
||||
const recovered = await parseAgentSessionFileCached(candidate('grok', file), 'linux')
|
||||
|
||||
// Grok's messageCount comes from the session doc, so the transcript's
|
||||
// contribution shows up in the previews.
|
||||
expect(recovered?.previewMessages).toEqual([
|
||||
expect.objectContaining({ role: 'user', text: 'ship it' })
|
||||
])
|
||||
@@ -221,7 +201,11 @@ describe('Grok session parse against a stalled WSL transcript', () => {
|
||||
|
||||
const session = await parseAgentSessionFileCached(candidate('grok', file), 'linux')
|
||||
|
||||
expect(session).toMatchObject({ agent: 'grok', title: 'Grok session', previewMessages: [] })
|
||||
expect(session).toMatchObject({
|
||||
agent: 'grok',
|
||||
title: 'Grok session',
|
||||
previewMessages: []
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@@ -254,16 +238,4 @@ describe('OpenCode session parse against a stalled WSL transcript', () => {
|
||||
|
||||
expect(recovered?.messageCount).toBe(1)
|
||||
})
|
||||
|
||||
it('still returns the session when a live process left a half-written message', async () => {
|
||||
const file = sessionPath('OpenCodeHalfWritten')
|
||||
mocks.readdir.mockResolvedValue([dirent('msg-1.json')])
|
||||
mocks.readFile.mockImplementation((path: string) =>
|
||||
Promise.resolve(path === file ? OPENCODE_SESSION : '{"role":"user",')
|
||||
)
|
||||
|
||||
const session = await parseAgentSessionFileCached(candidate('opencode', file), 'linux')
|
||||
|
||||
expect(session).toMatchObject({ agent: 'opencode', title: 'OpenCode session', messageCount: 0 })
|
||||
})
|
||||
})
|
||||
|
||||
@@ -1,21 +0,0 @@
|
||||
import { readTranscriptSlice } from './wsl-transcript-fs-access'
|
||||
|
||||
const BOUNDARY_FINGERPRINT_BYTES = 64
|
||||
|
||||
/**
|
||||
* Fingerprint the bytes immediately before `offset`, so a same-size rewrite that
|
||||
* leaves mtime untouched is still detected as replaced content rather than
|
||||
* treated as an append the reader already consumed.
|
||||
*/
|
||||
export async function boundaryFingerprint(
|
||||
filePath: string,
|
||||
offset: number,
|
||||
signal?: AbortSignal
|
||||
): Promise<string> {
|
||||
if (offset <= 0) {
|
||||
return ''
|
||||
}
|
||||
const start = Math.max(0, offset - BOUNDARY_FINGERPRINT_BYTES)
|
||||
const slice = await readTranscriptSlice(filePath, start, offset - start, 'exact', signal)
|
||||
return slice.toString('base64')
|
||||
}
|
||||
@@ -1,4 +1,6 @@
|
||||
import { wslGatedStat } from './wsl-transcript-fs-access'
|
||||
import { readTranscriptSlice, wslGatedStat } from './wsl-transcript-fs-access'
|
||||
|
||||
const BOUNDARY_FINGERPRINT_BYTES = 64
|
||||
|
||||
export type TranscriptFileVersion = {
|
||||
identity: string
|
||||
@@ -20,6 +22,19 @@ export async function readTranscriptFileVersion(
|
||||
}
|
||||
}
|
||||
|
||||
export async function boundaryFingerprint(
|
||||
filePath: string,
|
||||
offset: number,
|
||||
signal?: AbortSignal
|
||||
): Promise<string> {
|
||||
if (offset <= 0) {
|
||||
return ''
|
||||
}
|
||||
const start = Math.max(0, offset - BOUNDARY_FINGERPRINT_BYTES)
|
||||
const slice = await readTranscriptSlice(filePath, start, offset - start, 'exact', signal)
|
||||
return slice.toString('base64')
|
||||
}
|
||||
|
||||
export function transcriptFileVersionChanged(
|
||||
current: TranscriptFileVersion,
|
||||
previous: TranscriptFileVersion
|
||||
|
||||
@@ -2,16 +2,15 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import type * as NodeFsPromisesModule from 'node:fs/promises'
|
||||
|
||||
const UNC_PATH = '\\\\wsl.localhost\\Ubuntu\\home\\ada\\.claude\\projects\\p\\session.jsonl'
|
||||
const SLOW_MESSAGE =
|
||||
'WSL transcript files are temporarily unavailable because filesystem access is taking too long. Try again shortly or restart Orca if the issue continues.'
|
||||
|
||||
const mocks = vi.hoisted(() => ({
|
||||
resolve: vi.fn<() => Promise<string | null>>(),
|
||||
stat: vi.fn(),
|
||||
open: vi.fn()
|
||||
}))
|
||||
|
||||
vi.mock('./session-file-resolver', () => ({ resolveSessionFilePath: mocks.resolve }))
|
||||
vi.mock('./session-file-resolver', () => ({
|
||||
resolveSessionFilePath: mocks.resolve
|
||||
}))
|
||||
vi.mock('node:fs/promises', async (importOriginal) => ({
|
||||
...(await importOriginal<typeof NodeFsPromisesModule>()),
|
||||
stat: mocks.stat,
|
||||
@@ -22,7 +21,10 @@ import {
|
||||
clearNativeChatTranscriptCache,
|
||||
readNativeChatTranscriptCached
|
||||
} from './transcript-read-cache'
|
||||
import { WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS } from './wsl-transcript-fs-gate'
|
||||
import {
|
||||
WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS,
|
||||
WSL_TRANSCRIPT_FS_SLOW_MESSAGE
|
||||
} from './wsl-transcript-fs-gate'
|
||||
|
||||
const BODY = Buffer.from(
|
||||
`${JSON.stringify({
|
||||
@@ -33,26 +35,18 @@ const BODY = Buffer.from(
|
||||
})}\n`
|
||||
)
|
||||
|
||||
// The stalled body read keeps a permit until it settles, so every case releases
|
||||
// it before the next gated call runs.
|
||||
let releaseStall: (() => void) | undefined
|
||||
|
||||
function stallingHandle() {
|
||||
return {
|
||||
read: vi.fn(
|
||||
(buffer: Buffer) =>
|
||||
new Promise<{ bytesRead: number; buffer: Buffer }>((resolve) => {
|
||||
releaseStall = () => resolve({ bytesRead: 0, buffer })
|
||||
})
|
||||
),
|
||||
close: vi.fn(async () => {})
|
||||
}
|
||||
function stalls<T>(value: T): Promise<T> {
|
||||
return new Promise<T>((resolve) => {
|
||||
releaseStall = () => resolve(value)
|
||||
})
|
||||
}
|
||||
|
||||
function servingHandle() {
|
||||
function transcriptHandle(body = BODY) {
|
||||
return {
|
||||
read: vi.fn(async (buffer: Buffer, offset: number, length: number, position: number) => {
|
||||
const slice = BODY.subarray(position, Math.min(position + length, BODY.length))
|
||||
const slice = body.subarray(position, Math.min(position + length, body.length))
|
||||
slice.copy(buffer, offset)
|
||||
return { bytesRead: slice.length, buffer }
|
||||
}),
|
||||
@@ -79,32 +73,43 @@ afterEach(async () => {
|
||||
vi.useRealTimers()
|
||||
})
|
||||
|
||||
describe('cached native chat transcript read with a refused transcript body', () => {
|
||||
it('retries the body on the next call even though the mtime is unchanged', async () => {
|
||||
mocks.open.mockResolvedValue(stallingHandle())
|
||||
describe('cached native chat transcript read after WSL gate refusals', () => {
|
||||
it('retries a refused stat instead of caching it as a missing file', async () => {
|
||||
mocks.stat.mockReturnValueOnce(stalls({ mtimeMs: 42, size: BODY.length }))
|
||||
const refused = readNativeChatTranscriptCached('claude', 'session-id')
|
||||
await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS + 1)
|
||||
expect(await refused).toEqual({ error: SLOW_MESSAGE })
|
||||
|
||||
// Free the permit the unabortable read still holds, then recover.
|
||||
expect(await refused).toEqual({ error: WSL_TRANSCRIPT_FS_SLOW_MESSAGE })
|
||||
releaseStall?.()
|
||||
releaseStall = undefined
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
mocks.open.mockResolvedValue(servingHandle())
|
||||
mocks.open.mockResolvedValue(transcriptHandle())
|
||||
|
||||
await expect(readNativeChatTranscriptCached('claude', 'session-id')).resolves.toMatchObject({
|
||||
messages: [expect.objectContaining({ role: 'user' })]
|
||||
})
|
||||
expect(mocks.stat).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
|
||||
it('retries the body on the next call even though the mtime is unchanged', async () => {
|
||||
mocks.open.mockResolvedValue({
|
||||
read: vi.fn((buffer: Buffer) => stalls({ bytesRead: 0, buffer })),
|
||||
close: vi.fn(async () => {})
|
||||
})
|
||||
const refused = readNativeChatTranscriptCached('claude', 'session-id')
|
||||
await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS + 1)
|
||||
expect(await refused).toEqual({ error: WSL_TRANSCRIPT_FS_SLOW_MESSAGE })
|
||||
|
||||
releaseStall?.()
|
||||
releaseStall = undefined
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
mocks.open.mockResolvedValue(transcriptHandle())
|
||||
|
||||
const recovered = await readNativeChatTranscriptCached('claude', 'session-id')
|
||||
|
||||
expect(mocks.open).toHaveBeenCalledTimes(2)
|
||||
expect(recovered).toMatchObject({ messages: [expect.objectContaining({ role: 'user' })] })
|
||||
})
|
||||
|
||||
it('still caches a healthy parse, so recovery does not cost a read per call', async () => {
|
||||
mocks.open.mockResolvedValue(servingHandle())
|
||||
|
||||
const first = await readNativeChatTranscriptCached('claude', 'session-id')
|
||||
const second = await readNativeChatTranscriptCached('claude', 'session-id')
|
||||
|
||||
expect(mocks.open).toHaveBeenCalledTimes(1)
|
||||
expect(second).toBe(first)
|
||||
expect(recovered).toMatchObject({
|
||||
messages: [expect.objectContaining({ role: 'user' })]
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@@ -1,57 +0,0 @@
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import type * as NodeFsPromisesModule from 'node:fs/promises'
|
||||
|
||||
const UNC_PATH = '\\\\wsl.localhost\\Ubuntu\\home\\ada\\.claude\\projects\\p\\session.jsonl'
|
||||
|
||||
const mocks = vi.hoisted(() => ({
|
||||
resolve: vi.fn<() => Promise<string | null>>(),
|
||||
stat: vi.fn()
|
||||
}))
|
||||
|
||||
vi.mock('./session-file-resolver', () => ({ resolveSessionFilePath: mocks.resolve }))
|
||||
vi.mock('node:fs/promises', async (importOriginal) => ({
|
||||
...(await importOriginal<typeof NodeFsPromisesModule>()),
|
||||
stat: mocks.stat
|
||||
}))
|
||||
|
||||
import { readNativeChatTranscriptCached } from './transcript-read-cache'
|
||||
import { WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS } from './wsl-transcript-fs-gate'
|
||||
|
||||
const SLOW_MESSAGE =
|
||||
'WSL transcript files are temporarily unavailable because filesystem access is taking too long. Try again shortly or restart Orca if the issue continues.'
|
||||
|
||||
beforeEach(() => {
|
||||
mocks.resolve.mockReset()
|
||||
mocks.stat.mockReset()
|
||||
mocks.resolve.mockResolvedValue(UNC_PATH)
|
||||
})
|
||||
|
||||
describe('cached native chat transcript read with a stalled post-resolution stat', () => {
|
||||
it('reports a retryable error without notFound and leaves nothing cached', async () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
mocks.stat.mockReturnValueOnce(new Promise(() => {}))
|
||||
const stalled = readNativeChatTranscriptCached('claude', 'session-id')
|
||||
await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS + 1)
|
||||
|
||||
const result = await stalled
|
||||
expect(result).toEqual({ error: SLOW_MESSAGE })
|
||||
expect(result).not.toHaveProperty('notFound')
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('re-attempts the stat on the next call instead of serving a poisoned entry', async () => {
|
||||
// The stalled task above still holds its permit, so this fast-fails on the
|
||||
// stuck route rather than waiting out a second deadline — but it must still
|
||||
// reach `stat` again once a different distro answers.
|
||||
const healthyPath = '\\\\wsl.localhost\\Debian\\home\\ada\\.claude\\projects\\p\\session.jsonl'
|
||||
mocks.resolve.mockResolvedValue(healthyPath)
|
||||
mocks.stat.mockResolvedValue({ mtimeMs: 5, size: 0 })
|
||||
|
||||
await readNativeChatTranscriptCached('claude', 'session-id')
|
||||
|
||||
expect(mocks.stat).toHaveBeenCalledWith(healthyPath)
|
||||
})
|
||||
})
|
||||
@@ -1,6 +1,6 @@
|
||||
import type { NativeChatMessage, NativeChatTurnLifecycle } from '../../shared/native-chat-types'
|
||||
import { boundaryFingerprint } from './transcript-boundary-fingerprint'
|
||||
import {
|
||||
boundaryFingerprint,
|
||||
readTranscriptFileVersion,
|
||||
transcriptFileVersionChanged,
|
||||
type TranscriptFileVersion
|
||||
|
||||
@@ -95,7 +95,9 @@ describe('transcript filesystem accessor off WSL UNC', () => {
|
||||
expect(mocks.open).toHaveBeenCalledWith(path, 'r')
|
||||
// Off UNC the raw stream is handed back verbatim, encoding included.
|
||||
expect(stream).toBe('raw-stream')
|
||||
expect(mocks.createReadStream).toHaveBeenCalledWith(path, { encoding: 'utf-8' })
|
||||
expect(mocks.createReadStream).toHaveBeenCalledWith(path, {
|
||||
encoding: 'utf-8'
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@@ -105,12 +107,21 @@ describe('transcript filesystem accessor on WSL UNC', () => {
|
||||
['wsl$', LEGACY_UNC_PATH]
|
||||
])('routes %s paths through the gate', async (_label, path) => {
|
||||
mocks.stat.mockResolvedValue({ size: 7 })
|
||||
await expect(wslGatedStat(path, 'exact')).resolves.toEqual({ size: 7 })
|
||||
expect(mocks.runTask).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ operation: 'stat', path, priority: 'exact' }),
|
||||
expect.any(Function)
|
||||
)
|
||||
expect(mocks.createReadStream).not.toHaveBeenCalled()
|
||||
mocks.lstat.mockResolvedValue({ size: 7 })
|
||||
mocks.readdir.mockResolvedValue([])
|
||||
mocks.readFile.mockResolvedValue('body')
|
||||
|
||||
await wslGatedStat(path, 'exact')
|
||||
await wslGatedLstat(path, 'scan')
|
||||
await wslGatedReaddir(path, 'scan')
|
||||
await wslGatedReadFile(path, 'utf-8', 'scan')
|
||||
|
||||
expect(mocks.runTask.mock.calls.map(([options]) => options.operation)).toEqual([
|
||||
'stat',
|
||||
'lstat',
|
||||
'readdir',
|
||||
'readfile'
|
||||
])
|
||||
})
|
||||
|
||||
it('opts positional reads and opens out of coalescing', async () => {
|
||||
@@ -251,45 +262,6 @@ describe('transcript handle close off WSL UNC', () => {
|
||||
})
|
||||
|
||||
describe('per-chunk admission', () => {
|
||||
it('streams a long healthy-but-slow file past the whole-file deadline', async () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
const chunkDelayMs = WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS / 4
|
||||
const totalChunks = 6
|
||||
let served = 0
|
||||
const handle = fakeHandle()
|
||||
handle.read.mockImplementation(
|
||||
(buffer: Buffer) =>
|
||||
new Promise((resolve) => {
|
||||
setTimeout(() => {
|
||||
if (served++ >= totalChunks) {
|
||||
resolve({ bytesRead: 0, buffer })
|
||||
return
|
||||
}
|
||||
buffer.write('x')
|
||||
resolve({ bytesRead: 1, buffer })
|
||||
}, chunkDelayMs)
|
||||
})
|
||||
)
|
||||
mocks.open.mockResolvedValue(handle)
|
||||
|
||||
const stream = openTranscriptReadStream(UNC_PATH, {}, 'exact')
|
||||
const collected: Promise<Buffer[]> = (async () => {
|
||||
const out: Buffer[] = []
|
||||
for await (const chunk of stream) {
|
||||
out.push(chunk as Buffer)
|
||||
}
|
||||
return out
|
||||
})()
|
||||
|
||||
// Elapsed time far exceeds one exact deadline; each chunk stays under it.
|
||||
await vi.advanceTimersByTimeAsync(chunkDelayMs * (totalChunks + 2))
|
||||
await expect(collected).resolves.toHaveLength(totalChunks)
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('carries a codepoint straddling the 1 MiB chunk boundary across chunks', async () => {
|
||||
const chunkBytes = 1024 * 1024
|
||||
const emoji = Buffer.from('😀', 'utf8')
|
||||
|
||||
@@ -5,31 +5,28 @@ import { StringDecoder } from 'node:string_decoder'
|
||||
import { isWslUncPath } from '../../shared/wsl-paths'
|
||||
import { runWslTranscriptFsTask, type WslTranscriptFsTaskPriority } from './wsl-transcript-fs-gate'
|
||||
|
||||
/**
|
||||
* Transcript-layer filesystem primitives that route WSL UNC paths through the
|
||||
* admission gate and everything else straight to `node:fs`. The `isWslUncPath`
|
||||
* guard is the single switch: macOS, Linux, Windows drive paths, SSH remotes and
|
||||
* folder workspaces execute the exact call they execute today.
|
||||
*
|
||||
* Invariant future call sites must preserve: every gated unit here is a single
|
||||
* syscall, and `openTranscriptReadStream` composes them OUTSIDE the gate. A
|
||||
* gated call awaited from inside another gated task's operation would deadlock
|
||||
* on the single scan slot or the route+priority lane.
|
||||
*/
|
||||
|
||||
// Why: one deadline per chunk instead of one for the whole file, so a large
|
||||
// healthy-but-slow transcript is not false-failed by a whole-file timeout.
|
||||
const WSL_TRANSCRIPT_READ_CHUNK_BYTES = 1024 * 1024
|
||||
type Operation = Parameters<typeof runWslTranscriptFsTask>[0]['operation']
|
||||
|
||||
function runPathOperation<T>(
|
||||
operation: Operation,
|
||||
path: string,
|
||||
priority: WslTranscriptFsTaskPriority,
|
||||
signal: AbortSignal | undefined,
|
||||
task: () => Promise<T>,
|
||||
options?: { dedupe?: boolean; onAbandonedResult?: (value: T) => void }
|
||||
): Promise<T> {
|
||||
return isWslUncPath(path)
|
||||
? runWslTranscriptFsTask({ operation, path, priority, signal, ...options }, task)
|
||||
: task()
|
||||
}
|
||||
|
||||
export function wslGatedStat(
|
||||
path: string,
|
||||
priority: WslTranscriptFsTaskPriority,
|
||||
signal?: AbortSignal
|
||||
): Promise<Stats> {
|
||||
if (!isWslUncPath(path)) {
|
||||
return stat(path)
|
||||
}
|
||||
return runWslTranscriptFsTask({ operation: 'stat', path, priority, signal }, () => stat(path))
|
||||
return runPathOperation('stat', path, priority, signal, () => stat(path))
|
||||
}
|
||||
|
||||
export function wslGatedLstat(
|
||||
@@ -37,10 +34,7 @@ export function wslGatedLstat(
|
||||
priority: WslTranscriptFsTaskPriority,
|
||||
signal?: AbortSignal
|
||||
): Promise<Stats> {
|
||||
if (!isWslUncPath(path)) {
|
||||
return lstat(path)
|
||||
}
|
||||
return runWslTranscriptFsTask({ operation: 'lstat', path, priority, signal }, () => lstat(path))
|
||||
return runPathOperation('lstat', path, priority, signal, () => lstat(path))
|
||||
}
|
||||
|
||||
export function wslGatedReaddir(
|
||||
@@ -48,10 +42,7 @@ export function wslGatedReaddir(
|
||||
priority: WslTranscriptFsTaskPriority,
|
||||
signal?: AbortSignal
|
||||
): Promise<Dirent[]> {
|
||||
if (!isWslUncPath(path)) {
|
||||
return readdir(path, { withFileTypes: true })
|
||||
}
|
||||
return runWslTranscriptFsTask({ operation: 'readdir', path, priority, signal }, () =>
|
||||
return runPathOperation('readdir', path, priority, signal, () =>
|
||||
readdir(path, { withFileTypes: true })
|
||||
)
|
||||
}
|
||||
@@ -62,44 +53,20 @@ export function wslGatedReadFile(
|
||||
priority: WslTranscriptFsTaskPriority,
|
||||
signal?: AbortSignal
|
||||
): Promise<string> {
|
||||
if (!isWslUncPath(path)) {
|
||||
return readFile(path, encoding)
|
||||
}
|
||||
return runWslTranscriptFsTask({ operation: 'readfile', path, priority, signal }, () =>
|
||||
readFile(path, encoding)
|
||||
)
|
||||
return runPathOperation('readfile', path, priority, signal, () => readFile(path, encoding))
|
||||
}
|
||||
|
||||
// dedupe:false — two joiners would share one FileHandle and both close it.
|
||||
export function wslGatedOpen(
|
||||
path: string,
|
||||
priority: WslTranscriptFsTaskPriority,
|
||||
signal?: AbortSignal
|
||||
): Promise<FileHandle> {
|
||||
if (!isWslUncPath(path)) {
|
||||
return open(path, 'r')
|
||||
}
|
||||
return runWslTranscriptFsTask(
|
||||
{
|
||||
operation: 'open',
|
||||
path,
|
||||
priority,
|
||||
signal,
|
||||
dedupe: false,
|
||||
// An unabortable open can still succeed after its caller timed out or
|
||||
// cancelled; without this the descriptor leaks for the process lifetime.
|
||||
onAbandonedResult: (handle) => void closeTranscriptHandle(handle, path)
|
||||
},
|
||||
() => open(path, 'r')
|
||||
)
|
||||
return runPathOperation('open', path, priority, signal, () => open(path, 'r'), {
|
||||
dedupe: false,
|
||||
onAbandonedResult: (handle) => void closeTranscriptHandle(handle, path)
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* `path` is only the gate's route key and UNC guard — the handle carries no
|
||||
* path and is never re-opened. dedupe:false because `read` fills the CALLER's
|
||||
* buffer: a joiner would receive the first caller's buffer while its own
|
||||
* (often `Buffer.allocUnsafe`) stays uninitialized.
|
||||
*/
|
||||
export function wslGatedRead(
|
||||
handle: FileHandle,
|
||||
path: string,
|
||||
@@ -110,21 +77,16 @@ export function wslGatedRead(
|
||||
priority: WslTranscriptFsTaskPriority,
|
||||
signal?: AbortSignal
|
||||
): Promise<{ bytesRead: number; buffer: Buffer }> {
|
||||
if (!isWslUncPath(path)) {
|
||||
return handle.read(buffer, offset, length, position)
|
||||
}
|
||||
return runWslTranscriptFsTask({ operation: 'read', path, priority, signal, dedupe: false }, () =>
|
||||
handle.read(buffer, offset, length, position)
|
||||
return runPathOperation(
|
||||
'read',
|
||||
path,
|
||||
priority,
|
||||
signal,
|
||||
() => handle.read(buffer, offset, length, position),
|
||||
{ dedupe: false }
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Never gated. Off UNC this is the prior contract verbatim — the caller awaits
|
||||
* fd teardown and a close failure surfaces. On UNC it is fire-and-forget:
|
||||
* closing a handle on a stalled mount can itself block, and a gated close would
|
||||
* burn a permit and a waiter deadline purely for teardown. One leaked fd until
|
||||
* the OS unblocks beats a second blocked waiter.
|
||||
*/
|
||||
export function closeTranscriptHandle(handle: FileHandle, path: string): Promise<void> {
|
||||
if (!isWslUncPath(path)) {
|
||||
return handle.close()
|
||||
@@ -133,11 +95,6 @@ export function closeTranscriptHandle(handle: FileHandle, path: string): Promise
|
||||
return Promise.resolve()
|
||||
}
|
||||
|
||||
/**
|
||||
* Open, read one slice, close — for callers that want bytes at an offset and
|
||||
* never touch the handle. Each of the two syscalls is admitted separately, so a
|
||||
* stalled mount fails at whichever one it blocks on.
|
||||
*/
|
||||
export async function readTranscriptSlice(
|
||||
path: string,
|
||||
position: number,
|
||||
@@ -148,8 +105,17 @@ export async function readTranscriptSlice(
|
||||
const handle = await wslGatedOpen(path, priority, signal)
|
||||
try {
|
||||
const buffer = Buffer.allocUnsafe(length)
|
||||
const read = await wslGatedRead(handle, path, buffer, 0, length, position, priority, signal)
|
||||
return buffer.subarray(0, read.bytesRead)
|
||||
const { bytesRead } = await wslGatedRead(
|
||||
handle,
|
||||
path,
|
||||
buffer,
|
||||
0,
|
||||
length,
|
||||
position,
|
||||
priority,
|
||||
signal
|
||||
)
|
||||
return buffer.subarray(0, bytesRead)
|
||||
} finally {
|
||||
await closeTranscriptHandle(handle, path)
|
||||
}
|
||||
@@ -157,11 +123,7 @@ export async function readTranscriptSlice(
|
||||
|
||||
export type TranscriptReadStreamOptions = {
|
||||
start?: number
|
||||
/** Inclusive, matching `createReadStream`. */
|
||||
end?: number
|
||||
/** Set it to get decoded string chunks on both branches; leave it unset and
|
||||
* the UNC branch yields `Buffer`s the consumer must decode incrementally
|
||||
* itself, since a chunk boundary can split a multibyte codepoint. */
|
||||
encoding?: BufferEncoding
|
||||
}
|
||||
|
||||
@@ -172,10 +134,6 @@ async function* gatedChunks(
|
||||
signal?: AbortSignal
|
||||
): AsyncGenerator<Buffer | string> {
|
||||
const handle = await wslGatedOpen(path, priority, signal)
|
||||
// Why: chunk boundaries fall mid-codepoint, so decoding each slice
|
||||
// independently would emit U+FFFD on both sides of any straddling character.
|
||||
// The decoder carries the partial sequence across chunks, as the
|
||||
// `createReadStream` branch's own decoder does.
|
||||
const decoder = options.encoding ? new StringDecoder(options.encoding) : null
|
||||
try {
|
||||
let position = options.start ?? 0
|
||||
@@ -212,24 +170,15 @@ async function* gatedChunks(
|
||||
yield decoded
|
||||
}
|
||||
}
|
||||
// Trailing bytes of an incomplete sequence at EOF, replacement-char'd once.
|
||||
const trailing = decoder?.end()
|
||||
if (trailing) {
|
||||
yield trailing
|
||||
}
|
||||
} finally {
|
||||
// Runs on `.destroy()` too (Readable.from calls the generator's return()),
|
||||
// so an aborted head-read cannot leak the gated handle.
|
||||
await closeTranscriptHandle(handle, path)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A read stream whose UNC branch admits and deadlines each 1 MiB chunk
|
||||
* separately. A gate refusal mid-stream surfaces as an `'error'` event carrying
|
||||
* the `WslTranscriptFsError`, which every existing consumer funnels into its
|
||||
* `catch`.
|
||||
*/
|
||||
export function openTranscriptReadStream(
|
||||
path: string,
|
||||
options: TranscriptReadStreamOptions,
|
||||
|
||||
Reference in New Issue
Block a user