Keep exit events behind backpressured stream output

Adversarial review round (3 independent Codex reviewers + baseline pass) found that exit frames were written directly to the stream socket and could overtake output queued in the backpressure queue, so a renderer could tear down the session before the final output tail arrived. Both exit sites now route through writeEventLine on the same ordered path; the priority splice orders the exit after the session's own output, ahead of unrelated backlog, and protects it from bounded-queue trimming. Red-proven regression test at the daemon-server seam. Also documents the soft queue bound (maxQueuedBytes plus at most one frame) found by the same review.

Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
brennanb2025
2026-07-05 09:09:30 -07:00
co-authored by Orca
parent c3310f9af4
commit 4e13b69a10
5 changed files with 102 additions and 10 deletions
+16 -8
View File
@@ -1538,7 +1538,7 @@
"macos"
],
"coveredProviders": [],
"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).",
"coverageNotes": "Deterministic daemon socket write/drain contracts with a fake socket; local macOS evidence. The queue byte bound is soft by one frame: the sole remaining line is never dropped, so the effective cap is maxQueuedBytes plus one frame (<= the NDJSON line cap). 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"
@@ -1546,10 +1546,11 @@
"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": [
"pnpm exec vitest run --config config/vitest.config.ts src/main/daemon/daemon-stream-data-batcher.test.ts"
"pnpm exec vitest run --config config/vitest.config.ts src/main/daemon/daemon-stream-data-batcher.test.ts src/main/daemon/daemon-server.test.ts"
],
"testFiles": [
"src/main/daemon/daemon-stream-data-batcher.test.ts"
"src/main/daemon/daemon-stream-data-batcher.test.ts",
"src/main/daemon/daemon-server.test.ts"
],
"assertionRefs": [
{
@@ -1561,17 +1562,24 @@
"queued stream data stays bounded while a socket remains backpressured",
"global cleanup clears clients that only have backpressured stream writes"
]
},
{
"file": "src/main/daemon/daemon-server.test.ts",
"assertions": [
"exit events do not overtake backpressured stream output; the exit frame is written after the drained tail",
"interactive bypass and exit-flush ordering hold at the daemon-server seam"
]
}
],
"evidenceRuns": [
{
"date": "2026-07-03",
"date": "2026-07-05",
"runner": "local",
"platform": "macos",
"command": "pnpm exec vitest run --config config/vitest.config.ts src/main/daemon/daemon-stream-data-batcher.test.ts",
"command": "pnpm exec vitest run --config config/vitest.config.ts src/main/daemon/daemon-stream-data-batcher.test.ts src/main/daemon/daemon-server.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."
"durationSeconds": 0.6,
"summary": "2 test files passed, 32 tests passed; queue contracts plus the daemon-server exit-ordering regression test added after the adversarial review round."
}
],
"runtimeBudget": {
@@ -1584,7 +1592,7 @@
},
"redGreenEvidence": {
"status": "partial",
"evidence": "Tests assert daemon stream writes pause after socket write(false), later stream data queues behind the pressured socket, flush-immediate output from another session is prioritized ahead of unrelated queued background backlog on drain while preserving per-session order, queued lines resume on drain, global cleanup and close/error clear pending drain listeners, and queued bytes are bounded by preserving priority output plus the newest tail. Needs intentional-break artifact plus active-input latency proof before promotion."
"evidence": "The exit-ordering regression test was proven red against the pre-fix branch (exit frame written directly while output sat in the backpressure queue) and green after routing exit through writeEventLine. The five queue contracts were developed red-first on the reliability stack."
},
"performanceBudget": {
"required": true,
+65
View File
@@ -363,6 +363,71 @@ describe('DaemonServer', () => {
}
})
it('keeps the exit event behind backpressured stream output', async () => {
vi.useFakeTimers()
try {
let subprocess: ReturnType<typeof createMockSubprocess>
server = new DaemonServer({
socketPath,
tokenPath,
spawnSubprocess: () => {
subprocess = createMockSubprocess()
return subprocess
}
})
const daemon = server as unknown as DaemonServerPrivate
const controlSocket = { destroy: vi.fn() } as unknown as Socket
const writes: string[] = []
let drainHandler: (() => void) | null = null
const streamSocket = {
destroyed: false,
destroy: vi.fn(),
write: vi.fn((line: string) => {
writes.push(String(line))
// First accepted write signals backpressure; later writes are accepted.
return writes.length !== 1
}),
once: vi.fn((event: string, cb: () => void) => {
if (event === 'drain') {
drainHandler = cb
}
}),
off: vi.fn()
} as unknown as Socket & { write: ReturnType<typeof vi.fn> }
daemon.clients.set('client-1', {
clientId: 'client-1',
controlSocket,
streamSocket
})
await daemon.routeRequest('client-1', {
id: 'req-1',
type: 'createOrAttach',
payload: { sessionId: 'test-session', cols: 80, rows: 24 }
})
subprocess!._simulateData('first')
vi.advanceTimersByTime(8)
expect(writes).toHaveLength(1)
subprocess!._simulateData('final-tail')
vi.advanceTimersByTime(8)
expect(writes).toHaveLength(1)
subprocess!._simulateExit(42)
expect(writes.some((w) => w.includes('"event":"exit"'))).toBe(false)
drainHandler!()
const tailIndex = writes.findIndex((w) => w.includes('final-tail'))
const exitIndex = writes.findIndex((w) => w.includes('"event":"exit"'))
expect(tailIndex).toBeGreaterThan(-1)
expect(exitIndex).toBeGreaterThan(tailIndex)
} finally {
vi.useRealTimers()
}
})
it('flushes pending batched stream output before the exit event', async () => {
vi.useFakeTimers()
try {
+10 -2
View File
@@ -300,7 +300,12 @@ export class DaemonServer {
this.streamDataBatcher.flush(clientId)
this.lastInputAtBySessionId.delete(p.sessionId)
if (client?.streamSocket) {
client.streamSocket.write(
// Why: exit must ride the same ordered path as stream data so it
// cannot overtake output queued behind a backpressured socket.
this.streamDataBatcher.writeEventLine(
clientId,
client.streamSocket,
p.sessionId,
encodeNdjson({
type: 'event',
event: 'exit',
@@ -422,7 +427,10 @@ export class DaemonServer {
// Why: write/resize are notification-heavy and intentionally do not wait
// for replies. If their target session is gone, this synthetic exit is the
// only signal the renderer gets to clear stale terminal pane bindings.
client.streamSocket.write(
this.streamDataBatcher.writeEventLine(
client.clientId,
client.streamSocket,
sessionId,
encodeNdjson({
type: 'event',
event: 'exit',
@@ -135,6 +135,8 @@ export class BackpressuredStreamWriteQueue {
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.
// The sole remaining line is never dropped, so the effective bound is
// maxQueuedBytes plus at most one frame (frames are <= the NDJSON line cap).
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)
@@ -169,6 +169,15 @@ export class DaemonStreamDataBatcher {
}
}
writeEventLine(clientId: string, streamSocket: Socket, sessionId: string, line: string): void {
// Why: session events (exit) must not overtake this session's queued output
// on a backpressured socket; the priority path orders the event after the
// session's own lines and protects it from bounded-queue trimming.
this.backpressureQueue.write(clientId, streamSocket, [{ sessionId, line, priority: true }], {
prioritizeBackpressuredSession: true
})
}
private writeStreamDataEvent(
clientId: string,
streamSocket: Socket,