mirror of
https://github.com/stablyai/orca.git
synced 2026-09-26 16:02:43 +00:00
fix: own the retained serve-channel tail and pin the no-newline invariant
- Copy the retained partial line via ownRetainedString after each drain so a 13+ char tail no longer pins the whole drained response as a V8 SlicedString (measured 23 MB -> 22 KB for 32 pending tails behind 1 Mi lines); regression test with --expose-gc. - Add a test asserting the retained buffer never contains a newline after a drain, which the chunk-only fast path depends on. - Locate the first delimiter with decoded.indexOf offset by the retained length instead of rescanning the whole accumulated buffer; test pins that no indexOf runs over more than the new chunk.
This commit is contained in:
@@ -91,6 +91,73 @@ describe('DesktopScriptServeChannel', () => {
|
||||
expect(handlers.onLine.mock.calls.map(([line]) => line)).toEqual(['first'])
|
||||
})
|
||||
|
||||
it('keeps the retained tail free of newlines after every drain', () => {
|
||||
const { channel, child } = createChannel()
|
||||
const retained = channel as unknown as { buffer: string }
|
||||
for (const chunk of ['a\nb', 'c\r\n\n\nd\ne', '\n', 'f\n\ng', Buffer.from('h\r\ni😀')]) {
|
||||
child.stdout.emit('data', chunk)
|
||||
// The fast path in readStdout scans only the new chunk, which is sound only if this holds.
|
||||
expect(retained.buffer).not.toContain('\n')
|
||||
}
|
||||
expect(retained.buffer).toBe('i😀')
|
||||
})
|
||||
|
||||
it('scans only the new chunk for the first newline of a pending line', () => {
|
||||
const { child, handlers } = createChannel()
|
||||
const pending = 'p'.repeat(1024 * 1024)
|
||||
child.stdout.emit('data', pending)
|
||||
const chunk = 'q\n'
|
||||
const indexOf = vi.spyOn(String.prototype, 'indexOf')
|
||||
let scanned: number[]
|
||||
try {
|
||||
child.stdout.emit('data', chunk)
|
||||
scanned = indexOf.mock.contexts.map((self) => String(self).length)
|
||||
} finally {
|
||||
indexOf.mockRestore()
|
||||
}
|
||||
expect(handlers.onLine).toHaveBeenCalledWith(`${pending}q`)
|
||||
// Locating the delimiter must not rescan the megabytes already known to hold none.
|
||||
expect(scanned.length).toBeGreaterThan(0)
|
||||
expect(Math.max(...scanned)).toBeLessThanOrEqual(chunk.length)
|
||||
})
|
||||
|
||||
it('releases the drained response that a retained tail was sliced from', () => {
|
||||
const gc = (globalThis as { gc?: () => void }).gc
|
||||
if (!gc) {
|
||||
throw new Error('global.gc unavailable - config/vitest.config.ts must pass --expose-gc')
|
||||
}
|
||||
const collectHeap = (): number => {
|
||||
gc()
|
||||
gc()
|
||||
return process.memoryUsage().heapUsed
|
||||
}
|
||||
const tails: string[] = []
|
||||
const feed = (index: number): void => {
|
||||
const child = new FakeChild()
|
||||
const channel = new DesktopScriptServeChannel(child as unknown as RuntimeChildProcess, {
|
||||
onLine: () => {},
|
||||
onGone: () => {},
|
||||
onOverflow: () => {}
|
||||
})
|
||||
const line = String.fromCharCode(65 + (index % 26)).repeat(1024 * 1024)
|
||||
child.stdout.emit('data', `${line}\n{"partial":${index}`)
|
||||
tails.push((channel as unknown as { buffer: string }).buffer)
|
||||
}
|
||||
for (let index = 0; index < 8; index += 1) {
|
||||
feed(index)
|
||||
}
|
||||
tails.length = 0
|
||||
const before = collectHeap()
|
||||
for (let index = 0; index < 32; index += 1) {
|
||||
feed(index)
|
||||
}
|
||||
const used = collectHeap() - before
|
||||
expect(tails).toHaveLength(32)
|
||||
expect(tails[5]).toBe('{"partial":5')
|
||||
// 32 pending tails, each sliced from a 1 Mi-char line; an un-owned tail pins the whole line.
|
||||
expect(used).toBeLessThan(4 * 1024 * 1024)
|
||||
})
|
||||
|
||||
describe('once stopped', () => {
|
||||
/**
|
||||
* The channel's half of the stale-callback guard, pinned here rather than
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { StringDecoder } from 'node:string_decoder'
|
||||
import type { ProcessSpec } from '../../shared/child-process/process-spec'
|
||||
import type { spawnProcess } from '../../shared/child-process/run-process'
|
||||
import { ownRetainedString } from '../../shared/own-retained-string'
|
||||
|
||||
/** The all-pipes child `spawnProcess` returns; avoids a node:child_process import. */
|
||||
export type RuntimeChildProcess = ReturnType<typeof spawnProcess>
|
||||
@@ -113,17 +114,19 @@ export class DesktopScriptServeChannel {
|
||||
return
|
||||
}
|
||||
const decoded = typeof chunk === 'string' ? chunk : this.decoder.write(chunk)
|
||||
const retainedLength = this.buffer.length
|
||||
this.buffer += decoded
|
||||
if (this.buffer.length > MAX_RESPONSE_CHARS) {
|
||||
this.buffer = ''
|
||||
this.handlers.onOverflow()
|
||||
return
|
||||
}
|
||||
// The retained tail has no newline; avoid rescanning and flattening it for every chunk.
|
||||
if (!decoded.includes('\n')) {
|
||||
// The retained tail has no newline, so only the new chunk needs scanning for the first one.
|
||||
const firstNewline = decoded.indexOf('\n')
|
||||
if (firstNewline === -1) {
|
||||
return
|
||||
}
|
||||
for (let newline = this.buffer.indexOf('\n'); newline >= 0;) {
|
||||
for (let newline = retainedLength + firstNewline; newline >= 0;) {
|
||||
// Slice a trailing CR off by index; trimming copies the whole payload.
|
||||
const end = newline > 0 && this.buffer.charCodeAt(newline - 1) === 13 ? newline - 1 : newline
|
||||
const line = this.buffer.slice(0, end)
|
||||
@@ -138,6 +141,8 @@ export class DesktopScriptServeChannel {
|
||||
}
|
||||
newline = this.buffer.indexOf('\n')
|
||||
}
|
||||
// Why own: the tail is a slice that would pin the whole drained buffer until the next newline.
|
||||
this.buffer = ownRetainedString(this.buffer)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user