mirror of
https://github.com/stablyai/orca.git
synced 2026-10-19 00:03:20 +00:00
Add daemon stream backpressure contracts and register the reliability gate
Daemon stream batcher now pauses on socket write(false), queues until drain, prioritizes interactive output over unrelated backpressured backlog, bounds queued bytes, and clears backpressure-only clients on global cleanup — with deterministic fake-socket contracts. The backpressured write queue and NDJSON frame helpers are extracted into daemon-stream-backpressure-queue.ts and daemon-stream-ndjson-frames.ts to keep modules focused. Adds the byte-exact 512KB mobile initial-snapshot budget test. Re-promotes terminal-performance.daemon-stream-backpressure to partial and extends the mobile stream budget gate evidence. Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
brennanb2025
Orca
@@ -1516,7 +1516,7 @@
|
||||
"id": "terminal-performance.daemon-stream-backpressure",
|
||||
"title": "Daemon terminal streams respect socket backpressure under output floods",
|
||||
"maturity": "experimental",
|
||||
"protection": "none",
|
||||
"protection": "partial",
|
||||
"owner": "terminal-performance",
|
||||
"layer": "daemon-provider-contract",
|
||||
"surfaces": [
|
||||
@@ -1534,19 +1534,46 @@
|
||||
"providers": [
|
||||
"daemon"
|
||||
],
|
||||
"coveredPlatforms": [],
|
||||
"coveredPlatforms": [
|
||||
"macos"
|
||||
],
|
||||
"coveredProviders": [],
|
||||
"coverageNotes": "Registered gap on main. The daemon batcher write(false)/drain contracts exist only on the pending reliability stack. It registers here with its owning split PR.",
|
||||
"coverageNotes": "Deterministic daemon socket write/drain contracts with a fake socket; local macOS evidence. Live hidden-output floods, active key latency artifacts, and the renderer ACK slice remain gaps (pending perf slice B).",
|
||||
"motivatingLinks": [
|
||||
"https://github.com/stablyai/orca/pull/6836",
|
||||
"https://github.com/stablyai/orca/pull/6858"
|
||||
],
|
||||
"invariant": "Daemon terminal output floods must pause when socket writes return false, resume on drain, keep buffered bytes bounded, and not starve focused input.",
|
||||
"oracle": "The current executable slice injects a slow socket and asserts write(false)/drain ordering, bounded queued daemon stream bytes, close/error listener cleanup, newest-tail preservation under sustained pressure, and flush-immediate cross-session priority ahead of unrelated background backlog on drain. Active input latency remains part of the broader perf gate.",
|
||||
"commands": [],
|
||||
"testFiles": [],
|
||||
"assertionRefs": [],
|
||||
"evidenceRuns": [],
|
||||
"commands": [
|
||||
"pnpm exec vitest run --config config/vitest.config.ts src/main/daemon/daemon-stream-data-batcher.test.ts"
|
||||
],
|
||||
"testFiles": [
|
||||
"src/main/daemon/daemon-stream-data-batcher.test.ts"
|
||||
],
|
||||
"assertionRefs": [
|
||||
{
|
||||
"file": "src/main/daemon/daemon-stream-data-batcher.test.ts",
|
||||
"assertions": [
|
||||
"stream writes pause when the socket applies backpressure and resume on drain",
|
||||
"later stream data queues behind a backpressured socket until drain",
|
||||
"interactive session output is prioritized over unrelated backpressured backlog on drain",
|
||||
"queued stream data stays bounded while a socket remains backpressured",
|
||||
"global cleanup clears clients that only have backpressured stream writes"
|
||||
]
|
||||
}
|
||||
],
|
||||
"evidenceRuns": [
|
||||
{
|
||||
"date": "2026-07-03",
|
||||
"runner": "local",
|
||||
"platform": "macos",
|
||||
"command": "pnpm exec vitest run --config config/vitest.config.ts src/main/daemon/daemon-stream-data-batcher.test.ts",
|
||||
"result": "passed",
|
||||
"durationSeconds": 0.2,
|
||||
"summary": "1 test file passed, 10 tests passed; write/drain pause, queue-until-drain, interactive drain priority, bounded queued bytes, and backpressure-only cleanup contracts."
|
||||
}
|
||||
],
|
||||
"runtimeBudget": {
|
||||
"p95Seconds": 60,
|
||||
"scope": "daemon provider-contract gate"
|
||||
@@ -1569,9 +1596,8 @@
|
||||
"Pair with the broader output-backpressure budget before blocking."
|
||||
],
|
||||
"knownGaps": [
|
||||
"No executable coverage on main yet; the slice lives on the pending fix-terminal-reliability stack.",
|
||||
"Current command is daemon batcher contract coverage, not live daemon/Electron perf.",
|
||||
"Active input latency is not proven in live daemon/Electron perf."
|
||||
"No live flood or latency metric artifact yet; deterministic byte-accounting only.",
|
||||
"The renderer/main ACK and pending-cap extensions arrive with the second perf slice (pty.ts / pty-connection.ts hunks)."
|
||||
],
|
||||
"demotionRule": "Cannot promote if backpressure is inferred only from renderer output success."
|
||||
},
|
||||
@@ -1834,7 +1860,7 @@
|
||||
"macos"
|
||||
],
|
||||
"coveredProviders": [],
|
||||
"coverageNotes": "Local macOS evidence over the runtime-RPC stream budgets on main@1282f5c2d. The pending stack adds byte-exact 512KB/2MB/256KB/48KB budget assertions; legacy JSON subscribe parity remains undecided.",
|
||||
"coverageNotes": "Local macOS evidence over the runtime-RPC stream budgets on main@1282f5c2d. The 512KB mobile initial-snapshot budget is now byte-exact; remaining byte-exact 2MB/256KB/48KB assertions arrive with the pending stack; legacy JSON subscribe parity remains undecided.",
|
||||
"motivatingLinks": [
|
||||
"https://github.com/stablyai/orca/pull/6951",
|
||||
"https://github.com/stablyai/orca/pull/6955",
|
||||
@@ -1856,7 +1882,8 @@
|
||||
"assertions": [
|
||||
"legacy binary output queued during initial snapshot serialization stays bounded",
|
||||
"stale mobile resize re-stream completions are dropped for legacy binary streams",
|
||||
"aborted stream signals do not register stale listeners"
|
||||
"aborted stream signals do not register stale listeners",
|
||||
"mobile initial snapshots downgrade until they fit the 512KB byte budget"
|
||||
]
|
||||
},
|
||||
{
|
||||
@@ -1884,6 +1911,15 @@
|
||||
"result": "passed",
|
||||
"durationSeconds": 3.8,
|
||||
"summary": "3 test file(s) passed, 28 tests passed on main@1282f5c2d in a clean checkout."
|
||||
},
|
||||
{
|
||||
"date": "2026-07-03",
|
||||
"runner": "local",
|
||||
"platform": "macos",
|
||||
"command": "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/rpc/terminal-subscribe-buffer.test.ts src/main/runtime/rpc/terminal-output-batching.test.ts src/main/runtime/rpc/terminal-multiplex.test.ts",
|
||||
"result": "passed",
|
||||
"durationSeconds": 3.3,
|
||||
"summary": "3 test files passed, 29 tests passed, including the new byte-exact 512KB mobile initial-snapshot budget test."
|
||||
}
|
||||
],
|
||||
"runtimeBudget": {
|
||||
|
||||
@@ -0,0 +1,156 @@
|
||||
import type { Socket } from 'node:net'
|
||||
|
||||
export type BackpressuredStreamLine = {
|
||||
sessionId: string
|
||||
line: string
|
||||
priority?: boolean
|
||||
}
|
||||
|
||||
type BackpressuredStreamWrites = {
|
||||
socket: Socket
|
||||
lines: BackpressuredStreamLine[]
|
||||
queuedBytes: number
|
||||
onDrain: () => void
|
||||
onClose: () => void
|
||||
}
|
||||
|
||||
export type WriteStreamLinesOptions = {
|
||||
prioritizeBackpressuredSession?: boolean
|
||||
}
|
||||
|
||||
export class BackpressuredStreamWriteQueue {
|
||||
private byClient = new Map<string, BackpressuredStreamWrites>()
|
||||
private maxQueuedBytes: number
|
||||
|
||||
constructor(maxQueuedBytes: number) {
|
||||
this.maxQueuedBytes = Math.max(1, maxQueuedBytes)
|
||||
}
|
||||
|
||||
clientIds(): Iterable<string> {
|
||||
return this.byClient.keys()
|
||||
}
|
||||
|
||||
write(
|
||||
clientId: string,
|
||||
streamSocket: Socket,
|
||||
lines: BackpressuredStreamLine[],
|
||||
opts: WriteStreamLinesOptions = {}
|
||||
): void {
|
||||
const existing = this.byClient.get(clientId)
|
||||
if (existing) {
|
||||
if (existing.socket === streamSocket && !streamSocket.destroyed) {
|
||||
this.appendLines(existing, lines, opts)
|
||||
return
|
||||
}
|
||||
this.clear(clientId)
|
||||
}
|
||||
|
||||
for (let index = 0; index < lines.length; index += 1) {
|
||||
// Why: only Node's explicit `false` return signals backpressure; duck-typed
|
||||
// sockets in tests or wrappers may return undefined and must keep the
|
||||
// legacy direct-write path.
|
||||
if (streamSocket.write(lines[index].line) === false) {
|
||||
this.deferUntilDrain(clientId, streamSocket, lines.slice(index + 1))
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
clear(clientId: string): void {
|
||||
const pending = this.byClient.get(clientId)
|
||||
if (!pending) {
|
||||
return
|
||||
}
|
||||
this.byClient.delete(clientId)
|
||||
pending.socket.off('drain', pending.onDrain)
|
||||
pending.socket.off('close', pending.onClose)
|
||||
pending.socket.off('error', pending.onClose)
|
||||
}
|
||||
|
||||
private deferUntilDrain(
|
||||
clientId: string,
|
||||
streamSocket: Socket,
|
||||
lines: BackpressuredStreamLine[]
|
||||
): void {
|
||||
const onDrain = (): void => {
|
||||
const pending = this.byClient.get(clientId)
|
||||
if (!pending || pending.socket !== streamSocket) {
|
||||
return
|
||||
}
|
||||
this.byClient.delete(clientId)
|
||||
streamSocket.off('close', pending.onClose)
|
||||
streamSocket.off('error', pending.onClose)
|
||||
if (!streamSocket.destroyed) {
|
||||
this.write(clientId, streamSocket, pending.lines)
|
||||
}
|
||||
}
|
||||
const onClose = (): void => {
|
||||
this.clear(clientId)
|
||||
}
|
||||
|
||||
const pending: BackpressuredStreamWrites = {
|
||||
socket: streamSocket,
|
||||
lines: [],
|
||||
queuedBytes: 0,
|
||||
onDrain,
|
||||
onClose
|
||||
}
|
||||
this.appendLines(pending, lines)
|
||||
this.byClient.set(clientId, pending)
|
||||
streamSocket.once('drain', onDrain)
|
||||
streamSocket.once('close', onClose)
|
||||
streamSocket.once('error', onClose)
|
||||
}
|
||||
|
||||
private appendLines(
|
||||
pending: BackpressuredStreamWrites,
|
||||
lines: BackpressuredStreamLine[],
|
||||
opts: WriteStreamLinesOptions = {}
|
||||
): void {
|
||||
if (opts.prioritizeBackpressuredSession && lines.length > 0) {
|
||||
// Why: input-triggered daemon output should not sit behind an unrelated
|
||||
// hidden-output flood once the socket drains, but each session's bytes
|
||||
// must still keep their own order.
|
||||
const sessionId = lines[0].sessionId
|
||||
let lastSameSessionIndex = -1
|
||||
for (let index = pending.lines.length - 1; index >= 0; index -= 1) {
|
||||
if (pending.lines[index].sessionId === sessionId) {
|
||||
lastSameSessionIndex = index
|
||||
break
|
||||
}
|
||||
}
|
||||
if (lastSameSessionIndex >= 0) {
|
||||
pending.lines.splice(lastSameSessionIndex + 1, 0, ...lines)
|
||||
} else {
|
||||
const firstNonPriorityIndex = pending.lines.findIndex((line) => !line.priority)
|
||||
pending.lines.splice(
|
||||
firstNonPriorityIndex === -1 ? pending.lines.length : firstNonPriorityIndex,
|
||||
0,
|
||||
...lines
|
||||
)
|
||||
}
|
||||
} else {
|
||||
pending.lines.push(...lines)
|
||||
}
|
||||
pending.queuedBytes += this.measureLinesBytes(lines)
|
||||
// Why: a wedged stream socket can otherwise retain unbounded hidden-output
|
||||
// floods below the renderer ACK budget. Preserve the newest bounded tail.
|
||||
while (pending.queuedBytes > this.maxQueuedBytes && pending.lines.length > 1) {
|
||||
const dropIndex = pending.lines.findIndex((line) => !line.priority)
|
||||
const [dropped] = pending.lines.splice(dropIndex === -1 ? 0 : dropIndex, 1)
|
||||
pending.queuedBytes -= dropped ? this.measureLineBytes(dropped) : 0
|
||||
}
|
||||
}
|
||||
|
||||
private measureLinesBytes(lines: BackpressuredStreamLine[]): number {
|
||||
let bytes = 0
|
||||
for (const line of lines) {
|
||||
bytes += this.measureLineBytes(line)
|
||||
}
|
||||
return bytes
|
||||
}
|
||||
|
||||
private measureLineBytes(line: BackpressuredStreamLine): number {
|
||||
return Buffer.byteLength(line.line, 'utf8')
|
||||
}
|
||||
}
|
||||
@@ -1,13 +1,14 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import type { Socket } from 'node:net'
|
||||
import { EventEmitter } from 'node:events'
|
||||
import { DaemonStreamDataBatcher } from './daemon-stream-data-batcher'
|
||||
import { createNdjsonParser } from './ndjson'
|
||||
|
||||
function createBatcher(options?: ConstructorParameters<typeof DaemonStreamDataBatcher>[1]) {
|
||||
const streamSocket = {
|
||||
const streamSocket = Object.assign(new EventEmitter(), {
|
||||
destroyed: false,
|
||||
write: vi.fn()
|
||||
} as unknown as Socket & { write: ReturnType<typeof vi.fn> }
|
||||
write: vi.fn(() => true)
|
||||
}) as unknown as Socket & { write: ReturnType<typeof vi.fn> }
|
||||
const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket }), options)
|
||||
return { batcher, streamSocket }
|
||||
}
|
||||
@@ -129,4 +130,143 @@ describe('DaemonStreamDataBatcher', () => {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('pauses stream writes when the socket applies backpressure', () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
const maxLineBytes = 256
|
||||
const { batcher, streamSocket } = createBatcher({ maxLineBytes })
|
||||
streamSocket.write.mockReturnValueOnce(false).mockReturnValue(true)
|
||||
|
||||
batcher.enqueue('client-1', 'session-1', 'x'.repeat(maxLineBytes * 3))
|
||||
vi.advanceTimersByTime(8)
|
||||
|
||||
expect(streamSocket.write).toHaveBeenCalledTimes(1)
|
||||
|
||||
streamSocket.emit('drain')
|
||||
|
||||
expect(streamSocket.write.mock.calls.length).toBeGreaterThan(1)
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('queues later stream data behind a backpressured socket until drain', () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
const maxLineBytes = 256
|
||||
const { batcher, streamSocket } = createBatcher({ maxLineBytes })
|
||||
streamSocket.write.mockReturnValueOnce(false).mockReturnValue(true)
|
||||
|
||||
batcher.enqueue('client-1', 'session-1', 'x'.repeat(maxLineBytes * 3))
|
||||
vi.advanceTimersByTime(8)
|
||||
batcher.enqueue('client-1', 'session-2', 'interactive', {
|
||||
flushImmediately: true,
|
||||
flushMaxChars: 1024
|
||||
})
|
||||
|
||||
expect(streamSocket.write).toHaveBeenCalledTimes(1)
|
||||
|
||||
streamSocket.emit('drain')
|
||||
|
||||
expect(
|
||||
streamSocket.write.mock.calls.some(([line]) =>
|
||||
String(line).includes('"sessionId":"session-2"')
|
||||
)
|
||||
).toBe(true)
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('globally clears clients that only have backpressured stream writes', () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
const maxLineBytes = 256
|
||||
const { batcher, streamSocket } = createBatcher({ maxLineBytes })
|
||||
streamSocket.write.mockReturnValueOnce(false).mockReturnValue(true)
|
||||
|
||||
batcher.enqueue('client-1', 'session-1', 'x'.repeat(maxLineBytes * 3))
|
||||
vi.advanceTimersByTime(8)
|
||||
|
||||
expect(streamSocket.write).toHaveBeenCalledTimes(1)
|
||||
expect(streamSocket.listenerCount('drain')).toBe(1)
|
||||
|
||||
batcher.clear()
|
||||
expect(streamSocket.listenerCount('drain')).toBe(0)
|
||||
expect(streamSocket.listenerCount('close')).toBe(0)
|
||||
expect(streamSocket.listenerCount('error')).toBe(0)
|
||||
|
||||
streamSocket.write.mockClear()
|
||||
streamSocket.emit('drain')
|
||||
|
||||
expect(streamSocket.write).not.toHaveBeenCalled()
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('prioritizes interactive session output over unrelated backpressured backlog on drain', () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
const maxLineBytes = 256
|
||||
const { batcher, streamSocket } = createBatcher({ maxLineBytes })
|
||||
streamSocket.write.mockReturnValueOnce(false).mockReturnValue(true)
|
||||
|
||||
batcher.enqueue('client-1', 'session-background', 'x'.repeat(maxLineBytes * 3))
|
||||
vi.advanceTimersByTime(8)
|
||||
batcher.enqueue('client-1', 'session-interactive', 'prompt-redraw', {
|
||||
flushImmediately: true,
|
||||
flushMaxChars: 1024
|
||||
})
|
||||
|
||||
expect(streamSocket.write).toHaveBeenCalledTimes(1)
|
||||
streamSocket.write.mockClear()
|
||||
|
||||
streamSocket.emit('drain')
|
||||
|
||||
expect(String(streamSocket.write.mock.calls[0]?.[0])).toContain(
|
||||
'"sessionId":"session-interactive"'
|
||||
)
|
||||
expect(String(streamSocket.write.mock.calls[0]?.[0])).toContain('"data":"prompt-redraw"')
|
||||
expect(
|
||||
streamSocket.write.mock.calls.some(([line]) =>
|
||||
String(line).includes('"sessionId":"session-background"')
|
||||
)
|
||||
).toBe(true)
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('bounds queued stream data while a socket remains backpressured', () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
const { batcher, streamSocket } = createBatcher({
|
||||
maxLineBytes: 256,
|
||||
maxBackpressuredBytes: 1
|
||||
})
|
||||
streamSocket.write.mockReturnValueOnce(false).mockReturnValue(true)
|
||||
|
||||
batcher.enqueue('client-1', 'session-old', 'old-output'.repeat(200))
|
||||
vi.advanceTimersByTime(8)
|
||||
|
||||
for (let index = 0; index < 5; index += 1) {
|
||||
batcher.enqueue('client-1', `session-new-${index}`, `new-output-${index}`, {
|
||||
flushImmediately: true,
|
||||
flushMaxChars: 1024
|
||||
})
|
||||
}
|
||||
|
||||
expect(streamSocket.write).toHaveBeenCalledTimes(1)
|
||||
streamSocket.write.mockClear()
|
||||
|
||||
streamSocket.emit('drain')
|
||||
|
||||
expect(streamSocket.write).toHaveBeenCalledTimes(1)
|
||||
expect(String(streamSocket.write.mock.calls[0]?.[0])).toContain('"sessionId":"session-new-4"')
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
import type { Socket } from 'node:net'
|
||||
import { encodeNdjson, NDJSON_MAX_LINE_BYTES } from './ndjson'
|
||||
import { NDJSON_MAX_LINE_BYTES } from './ndjson'
|
||||
import { BackpressuredStreamWriteQueue } from './daemon-stream-backpressure-queue'
|
||||
import { encodeStreamDataEvent, splitStreamDataForNdjson } from './daemon-stream-ndjson-frames'
|
||||
|
||||
type StreamDataClient = {
|
||||
streamSocket: Socket | null
|
||||
@@ -14,6 +16,7 @@ type PendingStreamDataBatch = {
|
||||
// Why: match main-process PTY IPC batching to avoid adding latency while
|
||||
// removing daemon socket writes and JSON framing during bursty output.
|
||||
const STREAM_DATA_BATCH_INTERVAL_MS = 8
|
||||
const DEFAULT_MAX_BACKPRESSURED_STREAM_BYTES = 8 * 1024 * 1024
|
||||
|
||||
type EnqueueOptions = {
|
||||
flushImmediately?: boolean
|
||||
@@ -22,88 +25,12 @@ type EnqueueOptions = {
|
||||
|
||||
type DaemonStreamDataBatcherOptions = {
|
||||
maxLineBytes?: number
|
||||
}
|
||||
|
||||
function encodeStreamDataEvent(sessionId: string, data: string): string {
|
||||
return encodeNdjson({
|
||||
type: 'event',
|
||||
event: 'data',
|
||||
sessionId,
|
||||
payload: { data }
|
||||
})
|
||||
}
|
||||
|
||||
function streamDataEventLineBytes(sessionId: string, data: string): number {
|
||||
return Buffer.byteLength(encodeStreamDataEvent(sessionId, data), 'utf8')
|
||||
}
|
||||
|
||||
function isHighSurrogate(value: number): boolean {
|
||||
return value >= 0xd800 && value <= 0xdbff
|
||||
}
|
||||
|
||||
function isLowSurrogate(value: number): boolean {
|
||||
return value >= 0xdc00 && value <= 0xdfff
|
||||
}
|
||||
|
||||
function clampToSafeSplitIndex(value: string, start: number, end: number): number {
|
||||
if (end <= start || end >= value.length) {
|
||||
return end
|
||||
}
|
||||
const prev = value.charCodeAt(end - 1)
|
||||
const next = value.charCodeAt(end)
|
||||
return isHighSurrogate(prev) && isLowSurrogate(next) ? end - 1 : end
|
||||
}
|
||||
|
||||
function nextSafeSplitIndex(value: string, start: number): number {
|
||||
const next = Math.min(value.length, start + 1)
|
||||
if (
|
||||
next < value.length &&
|
||||
isHighSurrogate(value.charCodeAt(start)) &&
|
||||
isLowSurrogate(value.charCodeAt(next))
|
||||
) {
|
||||
return next + 1
|
||||
}
|
||||
return next
|
||||
}
|
||||
|
||||
function splitStreamDataForNdjson(sessionId: string, data: string, maxLineBytes: number): string[] {
|
||||
if (streamDataEventLineBytes(sessionId, data) <= maxLineBytes) {
|
||||
return [data]
|
||||
}
|
||||
|
||||
const chunks: string[] = []
|
||||
let start = 0
|
||||
while (start < data.length) {
|
||||
let low = start + 1
|
||||
let high = data.length
|
||||
let best = start
|
||||
|
||||
while (low <= high) {
|
||||
const rawMid = Math.floor((low + high) / 2)
|
||||
const mid = clampToSafeSplitIndex(data, start, rawMid)
|
||||
if (mid <= start) {
|
||||
low = rawMid + 1
|
||||
continue
|
||||
}
|
||||
|
||||
if (streamDataEventLineBytes(sessionId, data.slice(start, mid)) <= maxLineBytes) {
|
||||
best = mid
|
||||
low = rawMid + 1
|
||||
} else {
|
||||
high = rawMid - 1
|
||||
}
|
||||
}
|
||||
|
||||
const end = best > start ? best : nextSafeSplitIndex(data, start)
|
||||
chunks.push(data.slice(start, end))
|
||||
start = end
|
||||
}
|
||||
|
||||
return chunks
|
||||
maxBackpressuredBytes?: number
|
||||
}
|
||||
|
||||
export class DaemonStreamDataBatcher {
|
||||
private pendingByClient = new Map<string, PendingStreamDataBatch>()
|
||||
private backpressureQueue: BackpressuredStreamWriteQueue
|
||||
private getClient: (clientId: string) => StreamDataClient | undefined
|
||||
private maxLineBytes: number
|
||||
|
||||
@@ -113,6 +40,9 @@ export class DaemonStreamDataBatcher {
|
||||
) {
|
||||
this.getClient = getClient
|
||||
this.maxLineBytes = Math.max(1, options.maxLineBytes ?? NDJSON_MAX_LINE_BYTES)
|
||||
this.backpressureQueue = new BackpressuredStreamWriteQueue(
|
||||
options.maxBackpressuredBytes ?? DEFAULT_MAX_BACKPRESSURED_STREAM_BYTES
|
||||
)
|
||||
}
|
||||
|
||||
enqueue(clientId: string, sessionId: string, data: string, options: EnqueueOptions = {}): void {
|
||||
@@ -166,7 +96,7 @@ export class DaemonStreamDataBatcher {
|
||||
}
|
||||
|
||||
for (const entry of batch.queue) {
|
||||
this.writeStreamDataEvent(client.streamSocket, entry.sessionId, entry.data)
|
||||
this.writeStreamDataEvent(clientId, client.streamSocket, entry.sessionId, entry.data)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -217,30 +147,43 @@ export class DaemonStreamDataBatcher {
|
||||
}
|
||||
|
||||
for (const entry of flushed) {
|
||||
this.writeStreamDataEvent(client.streamSocket, entry.sessionId, entry.data)
|
||||
this.writeStreamDataEvent(clientId, client.streamSocket, entry.sessionId, entry.data, {
|
||||
prioritizeBackpressuredSession: true
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
clear(clientId?: string): void {
|
||||
const batches =
|
||||
const clientIds =
|
||||
clientId === undefined
|
||||
? Array.from(this.pendingByClient.entries())
|
||||
: [[clientId, this.pendingByClient.get(clientId)] as const]
|
||||
? new Set([...this.pendingByClient.keys(), ...this.backpressureQueue.clientIds()])
|
||||
: new Set([clientId])
|
||||
|
||||
for (const [id, batch] of batches) {
|
||||
for (const id of clientIds) {
|
||||
const batch = this.pendingByClient.get(id)
|
||||
if (batch?.timer) {
|
||||
clearTimeout(batch.timer)
|
||||
}
|
||||
this.pendingByClient.delete(id)
|
||||
this.backpressureQueue.clear(id)
|
||||
}
|
||||
}
|
||||
|
||||
private writeStreamDataEvent(streamSocket: Socket, sessionId: string, data: string): void {
|
||||
private writeStreamDataEvent(
|
||||
clientId: string,
|
||||
streamSocket: Socket,
|
||||
sessionId: string,
|
||||
data: string,
|
||||
opts: { prioritizeBackpressuredSession?: boolean } = {}
|
||||
): void {
|
||||
// Why: createNdjsonParser rejects oversized lines. Terminal output can
|
||||
// burst faster than the batch interval, so writer-side chunking prevents
|
||||
// the daemon from dropping its own stream events at the receiver.
|
||||
for (const chunk of splitStreamDataForNdjson(sessionId, data, this.maxLineBytes)) {
|
||||
streamSocket.write(encodeStreamDataEvent(sessionId, chunk))
|
||||
}
|
||||
const lines = splitStreamDataForNdjson(sessionId, data, this.maxLineBytes).map((chunk) => ({
|
||||
sessionId,
|
||||
line: encodeStreamDataEvent(sessionId, chunk),
|
||||
...(opts.prioritizeBackpressuredSession ? { priority: true } : {})
|
||||
}))
|
||||
this.backpressureQueue.write(clientId, streamSocket, lines, opts)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,85 @@
|
||||
import { encodeNdjson } from './ndjson'
|
||||
|
||||
// NDJSON framing for daemon stream-data events: encoding, byte measurement, and
|
||||
// surrogate-safe splitting so oversized chunks never break a UTF-16 pair mid-frame.
|
||||
export function encodeStreamDataEvent(sessionId: string, data: string): string {
|
||||
return encodeNdjson({
|
||||
type: 'event',
|
||||
event: 'data',
|
||||
sessionId,
|
||||
payload: { data }
|
||||
})
|
||||
}
|
||||
|
||||
export function streamDataEventLineBytes(sessionId: string, data: string): number {
|
||||
return Buffer.byteLength(encodeStreamDataEvent(sessionId, data), 'utf8')
|
||||
}
|
||||
|
||||
function isHighSurrogate(value: number): boolean {
|
||||
return value >= 0xd800 && value <= 0xdbff
|
||||
}
|
||||
|
||||
function isLowSurrogate(value: number): boolean {
|
||||
return value >= 0xdc00 && value <= 0xdfff
|
||||
}
|
||||
|
||||
function clampToSafeSplitIndex(value: string, start: number, end: number): number {
|
||||
if (end <= start || end >= value.length) {
|
||||
return end
|
||||
}
|
||||
const prev = value.charCodeAt(end - 1)
|
||||
const next = value.charCodeAt(end)
|
||||
return isHighSurrogate(prev) && isLowSurrogate(next) ? end - 1 : end
|
||||
}
|
||||
|
||||
function nextSafeSplitIndex(value: string, start: number): number {
|
||||
const next = Math.min(value.length, start + 1)
|
||||
if (
|
||||
next < value.length &&
|
||||
isHighSurrogate(value.charCodeAt(start)) &&
|
||||
isLowSurrogate(value.charCodeAt(next))
|
||||
) {
|
||||
return next + 1
|
||||
}
|
||||
return next
|
||||
}
|
||||
|
||||
export function splitStreamDataForNdjson(
|
||||
sessionId: string,
|
||||
data: string,
|
||||
maxLineBytes: number
|
||||
): string[] {
|
||||
if (streamDataEventLineBytes(sessionId, data) <= maxLineBytes) {
|
||||
return [data]
|
||||
}
|
||||
|
||||
const chunks: string[] = []
|
||||
let start = 0
|
||||
while (start < data.length) {
|
||||
let low = start + 1
|
||||
let high = data.length
|
||||
let best = start
|
||||
|
||||
while (low <= high) {
|
||||
const rawMid = Math.floor((low + high) / 2)
|
||||
const mid = clampToSafeSplitIndex(data, start, rawMid)
|
||||
if (mid <= start) {
|
||||
low = rawMid + 1
|
||||
continue
|
||||
}
|
||||
|
||||
if (streamDataEventLineBytes(sessionId, data.slice(start, mid)) <= maxLineBytes) {
|
||||
best = mid
|
||||
low = rawMid + 1
|
||||
} else {
|
||||
high = rawMid - 1
|
||||
}
|
||||
}
|
||||
|
||||
const end = best > start ? best : nextSafeSplitIndex(data, start)
|
||||
chunks.push(data.slice(start, end))
|
||||
start = end
|
||||
}
|
||||
|
||||
return chunks
|
||||
}
|
||||
@@ -246,6 +246,9 @@ describe('terminal output batching', () => {
|
||||
.map((frame) => decodeTerminalStreamFrame(frame))
|
||||
.filter((frame) => frame?.opcode === TerminalStreamOpcode.Output)
|
||||
expect(outputFrames.length).toBeGreaterThan(1)
|
||||
for (const frame of outputFrames) {
|
||||
expect(frame?.payload.byteLength ?? 0).toBeLessThanOrEqual(48 * 1024)
|
||||
}
|
||||
expect(firstOutputEncodeCount).toBe(1)
|
||||
expect(
|
||||
outputFrames.map((frame) => (frame ? decodeTerminalStreamText(frame.payload) : '')).join('')
|
||||
|
||||
@@ -430,6 +430,88 @@ describe('terminal subscribe buffering', () => {
|
||||
}
|
||||
})
|
||||
|
||||
it('downgrades mobile initial snapshots until they fit the 512KB byte budget', async () => {
|
||||
const messages: string[] = []
|
||||
const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
|
||||
const cleanups = new Map<string, () => void>()
|
||||
const oversizedSnapshot = 'x'.repeat(600 * 1024)
|
||||
const budgetedSnapshot = 'tail'.repeat(40 * 1024)
|
||||
const runtime = stubRuntime({
|
||||
resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }),
|
||||
readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }),
|
||||
serializeTerminalBuffer: vi.fn(async (_ptyId: string, opts?: { scrollbackRows?: number }) => {
|
||||
if ((opts?.scrollbackRows ?? 0) > 250) {
|
||||
return { data: oversizedSnapshot, cols: 100, rows: 30 }
|
||||
}
|
||||
return { data: budgetedSnapshot, cols: 100, rows: 30, seq: 42 }
|
||||
}),
|
||||
getTerminalSize: vi.fn().mockReturnValue({ cols: 80, rows: 24 }),
|
||||
getMobileDisplayMode: vi.fn().mockReturnValue('mobile-fit'),
|
||||
getLayout: vi.fn().mockReturnValue({ seq: 1 }),
|
||||
isTerminalAlternateScreen: vi.fn().mockReturnValue(false),
|
||||
handleMobileSubscribe: vi.fn().mockResolvedValue(undefined),
|
||||
handleMobileUnsubscribe: vi.fn(),
|
||||
subscribeToTerminalData: vi.fn().mockReturnValue(vi.fn()),
|
||||
subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()),
|
||||
subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()),
|
||||
registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => {
|
||||
cleanups.set(id, cleanup)
|
||||
}),
|
||||
cleanupSubscription: vi.fn((id: string) => {
|
||||
const cleanup = cleanups.get(id)
|
||||
cleanups.delete(id)
|
||||
cleanup?.()
|
||||
}),
|
||||
waitForTerminal: vi.fn(() => new Promise<RuntimeTerminalWait>(() => {})),
|
||||
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
|
||||
updateMobileViewport: vi.fn().mockResolvedValue(false)
|
||||
})
|
||||
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
|
||||
|
||||
const dispatchPromise = dispatcher.dispatchStreaming(
|
||||
makeRequest('terminal.subscribe', {
|
||||
terminal: 'terminal-1',
|
||||
client: { id: 'phone-1', type: 'mobile' },
|
||||
capabilities: { terminalBinaryStream: 1 }
|
||||
}),
|
||||
(msg) => messages.push(msg),
|
||||
{
|
||||
connectionId: 'conn-mobile-budget',
|
||||
sendBinary: (bytes) => binaryFrames.push(bytes)
|
||||
}
|
||||
)
|
||||
|
||||
await vi.waitFor(() =>
|
||||
expect(messages.some((msg) => JSON.parse(msg).result?.type === 'subscribed')).toBe(true)
|
||||
)
|
||||
|
||||
expect(runtime.serializeTerminalBuffer).toHaveBeenNthCalledWith(1, 'pty-1', {
|
||||
scrollbackRows: 1000
|
||||
})
|
||||
expect(runtime.serializeTerminalBuffer).toHaveBeenNthCalledWith(2, 'pty-1', {
|
||||
scrollbackRows: 500
|
||||
})
|
||||
expect(runtime.serializeTerminalBuffer).toHaveBeenNthCalledWith(3, 'pty-1', {
|
||||
scrollbackRows: 250
|
||||
})
|
||||
const decodedFrames = binaryFrames.map((frame) => decodeTerminalStreamFrame(frame))
|
||||
const snapshotStart = decodedFrames.find(
|
||||
(frame) => frame?.opcode === TerminalStreamOpcode.SnapshotStart
|
||||
)
|
||||
expect(snapshotStart && decodeTerminalStreamJson(snapshotStart.payload)).toMatchObject({
|
||||
truncatedByByteBudget: true
|
||||
})
|
||||
const snapshotData = decodedFrames
|
||||
.filter((frame) => frame?.opcode === TerminalStreamOpcode.SnapshotChunk)
|
||||
.map((frame) => (frame ? decodeTerminalStreamText(frame.payload) : ''))
|
||||
.join('')
|
||||
expect(snapshotData).toBe(budgetedSnapshot)
|
||||
expect(Buffer.byteLength(snapshotData, 'utf8')).toBeLessThanOrEqual(512 * 1024)
|
||||
|
||||
runtime.cleanupSubscription('terminal-1:phone-1')
|
||||
await dispatchPromise
|
||||
})
|
||||
|
||||
it('drops stale mobile resize re-stream completions for legacy binary streams', async () => {
|
||||
const messages: string[] = []
|
||||
const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
|
||||
|
||||
Reference in New Issue
Block a user