mirror of
https://github.com/stablyai/orca.git
synced 2026-09-21 16:02:20 +00:00
fix(daemon): pause producers when stream backlogs grow (#20947)
* fix(daemon): pause producers when stream backlogs grow * fix(daemon): reset stream backpressure on socket replacement * docs(daemon): point retention audit at current reproducer * test(daemon): validate stream retention audit outcomes * fix(daemon): bound the stream producer stall and leave a visible gap Stream backpressure pauses a session's PTY with no deadline: the only un-pause comes from the consumer draining, so a half-open peer that stops reading without closing freezes the shell for the rest of the session. Arm a 60s watchdog on the false->true stream-pause transition (not on the re-assertions refresh() makes for neighbouring sessions). On fire, mark the session stall-released: it becomes keep-tail droppable, its backlog is thinned behind a dataGap, and the producer runs again. The existing dataGap path makes the renderer restore that pane from the daemon's snapshot, so the user sees the terminal jump to current rather than sit frozen. The mark clears once the session's last byte leaves the daemon, restoring ordinary pausing. Nothing here reports a process exit - loss of contact with a consumer is not evidence about the child. Also enable TCP keepalive on the stream socket so a genuinely dead peer closes and onStreamDisconnected clears the pause. * test(daemon): put each casting SAFETY: directive on one line `oxlint-disable-next-line` covers only the line directly after it, so a rationale wrapped onto a second comment line suppressed nothing and the casts failed the changed-code quality gate. Drop the remaining JSON.parse cast for an annotated binding. --------- Co-authored-by: m4air <m4air@Mac.localdomain> Co-authored-by: Neil <neil@stably.ai>
This commit is contained in:
OrcaWin
m4air
Neil
@@ -0,0 +1,33 @@
|
||||
# Daemon stream retention reproduction
|
||||
|
||||
Run from the worktree root after installing dependencies:
|
||||
|
||||
```sh
|
||||
ORCA_BACKGROUND_LAUNCH=1 node docs/audits/daemon-stream-retention/reproduce.mjs
|
||||
```
|
||||
|
||||
The script bundles the current `DaemonStreamDataBatcher` with esbuild, then uses real loopback TCP sockets. It pauses the reader, feeds at most 8 MiB, honors the producer-pause callback, resumes the reader, and closes every socket in `finally`. It starts no PTY, agent, or application window. Output includes the bundle hash, queued payload sizes, RSS, and drain outcome.
|
||||
|
||||
Visible 64 KiB and 1 KiB producers should pause within the backlog budget and resume after both queues drain. A hidden 1 KiB producer should keep running while the existing keep-tail policy bounds its held output. RSS includes allocator/GC timing; queue measurements are the direct evidence.
|
||||
|
||||
`after.json` records a historical fixed run under Node v26.6.0: the visible bulk producer paused after 2,686,976 characters with 131,300 socket bytes and 2,031,616 held characters; the visible small-write producer paused with 3,401,200 socket bytes. The hidden producer processed all 8 MiB with 131,300 socket bytes and 667,648 held characters. Every case resumed and drained both queues to zero.
|
||||
|
||||
## Preserved measurements before the fix
|
||||
|
||||
`before-1mib-chunks.json` and `before-64kib-chunks.json` were captured before adding producer backpressure, with a real paused reader and 96 MiB of input. The latter includes the original compiled bundle hash and a hidden-session comparison.
|
||||
|
||||
| Input | Socket buffered bytes | Held characters |
|
||||
| ------ | --------------------: | --------------: |
|
||||
| 32 MiB | 131,321 | 32,636,928 |
|
||||
| 64 MiB | 32,809,583 | 33,554,432 |
|
||||
| 96 MiB | 66,406,511 | 33,554,432 |
|
||||
|
||||
The last 32 MiB retained another 33,596,928 queued payload bytes. The held queue plateaued at 32 MiB while the socket queue grew at 1.00127 bytes per produced ASCII byte. Resuming the reader drained both queues in 74 ms. Sampled RSS reached 1.38 GB with 64 KiB chunks; this includes string coalescing/slicing and GC effects.
|
||||
|
||||
The original measurement used Node v26.6.0 on macOS; the current reproduction records its runtime and should be run on the supported Node 24 toolchain. The historical build command was:
|
||||
|
||||
```sh
|
||||
ORCA_BACKGROUND_LAUNCH=1 node docs/audits/daemon-stream-retention/reproduce.mjs
|
||||
```
|
||||
|
||||
This proves a current stalled-consumer retaining path. It does not attribute issue #19831's whole-system memory total to this path; that report lacks per-process measurements.
|
||||
@@ -0,0 +1,46 @@
|
||||
{
|
||||
"node": "v26.6.0",
|
||||
"platform": "darwin",
|
||||
"bundleSha256": "7c63ca547aa62570878664f9e0504a8747f4daec2daf7c3cffcfce65ebede6bd",
|
||||
"cases": [
|
||||
{
|
||||
"chunkSize": 65536,
|
||||
"hidden": false,
|
||||
"stalled": {
|
||||
"producedChars": 2686976,
|
||||
"producerPaused": true,
|
||||
"socketBufferedBytes": 131300,
|
||||
"batcherQueuedChars": 2031616,
|
||||
"rssBytes": 62291968
|
||||
},
|
||||
"receivedBytes": 2690244,
|
||||
"afterDrain": { "producerPaused": false, "socketBufferedBytes": 0, "batcherQueuedChars": 0 }
|
||||
},
|
||||
{
|
||||
"chunkSize": 1024,
|
||||
"hidden": false,
|
||||
"stalled": {
|
||||
"producedChars": 3674112,
|
||||
"producerPaused": true,
|
||||
"socketBufferedBytes": 3401200,
|
||||
"batcherQueuedChars": 0,
|
||||
"rssBytes": 92307456
|
||||
},
|
||||
"receivedBytes": 3946800,
|
||||
"afterDrain": { "producerPaused": false, "socketBufferedBytes": 0, "batcherQueuedChars": 0 }
|
||||
},
|
||||
{
|
||||
"chunkSize": 1024,
|
||||
"hidden": true,
|
||||
"stalled": {
|
||||
"producedChars": 8388608,
|
||||
"producerPaused": false,
|
||||
"socketBufferedBytes": 131300,
|
||||
"batcherQueuedChars": 667648,
|
||||
"rssBytes": 105005056
|
||||
},
|
||||
"receivedBytes": 1607892,
|
||||
"afterDrain": { "producerPaused": false, "socketBufferedBytes": 0, "batcherQueuedChars": 0 }
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
{
|
||||
"source": "src/main/daemon/daemon-stream-data-batcher.ts",
|
||||
"node": "v26.6.0",
|
||||
"platform": "darwin",
|
||||
"stalledReader": true,
|
||||
"samples": [
|
||||
{
|
||||
"fedMiB": 0,
|
||||
"socketBufferedBytes": 0,
|
||||
"batcherQueuedChars": 0,
|
||||
"retainedAsciiPayloadBytes": 0,
|
||||
"rssBytes": 51249152
|
||||
},
|
||||
{
|
||||
"fedMiB": 32,
|
||||
"socketBufferedBytes": 131333,
|
||||
"batcherQueuedChars": 32636928,
|
||||
"retainedAsciiPayloadBytes": 32768261,
|
||||
"rssBytes": 57917440
|
||||
},
|
||||
{
|
||||
"fedMiB": 64,
|
||||
"socketBufferedBytes": 32811587,
|
||||
"batcherQueuedChars": 33554432,
|
||||
"retainedAsciiPayloadBytes": 66366019,
|
||||
"rssBytes": 239484928
|
||||
},
|
||||
{
|
||||
"fedMiB": 96,
|
||||
"socketBufferedBytes": 66410563,
|
||||
"batcherQueuedChars": 33554432,
|
||||
"retainedAsciiPayloadBytes": 99964995,
|
||||
"rssBytes": 311885824
|
||||
}
|
||||
],
|
||||
"retainedBytesPerProducedByteAfterValve": 1.0013275146484375
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
{
|
||||
"source": "src/main/daemon/daemon-stream-data-batcher.ts",
|
||||
"bundleSha256": "7555dbf7f6020022aca2a0d75f5d9eae2e2c3cf1586e0c1e66fbc357f3a3143e",
|
||||
"node": "v26.6.0",
|
||||
"platform": "darwin",
|
||||
"cases": [
|
||||
{
|
||||
"droppable": false,
|
||||
"samples": [
|
||||
{
|
||||
"fedMiB": 0,
|
||||
"socketBufferedBytes": 0,
|
||||
"batcherQueuedChars": 0,
|
||||
"rssBytes": 53968896
|
||||
},
|
||||
{
|
||||
"fedMiB": 32,
|
||||
"socketBufferedBytes": 131321,
|
||||
"batcherQueuedChars": 32636928,
|
||||
"rssBytes": 59834368
|
||||
},
|
||||
{
|
||||
"fedMiB": 64,
|
||||
"socketBufferedBytes": 32809583,
|
||||
"batcherQueuedChars": 33554432,
|
||||
"rssBytes": 1241956352
|
||||
},
|
||||
{
|
||||
"fedMiB": 96,
|
||||
"socketBufferedBytes": 66406511,
|
||||
"batcherQueuedChars": 33554432,
|
||||
"rssBytes": 1381416960
|
||||
}
|
||||
],
|
||||
"finalRetainedQueuedPayloadBytes": 99960943,
|
||||
"queuedBytesPerProducedByteAfter64MiB": 1.0012664794921875,
|
||||
"afterDrain": {
|
||||
"fedMiB": 96,
|
||||
"socketBufferedBytes": 0,
|
||||
"batcherQueuedChars": 0,
|
||||
"rssBytes": 327680000
|
||||
},
|
||||
"drainMs": 74.25875000000042,
|
||||
"receivedBytes": 100791448
|
||||
},
|
||||
{
|
||||
"droppable": true,
|
||||
"samples": [
|
||||
{
|
||||
"fedMiB": 0,
|
||||
"socketBufferedBytes": 0,
|
||||
"batcherQueuedChars": 0,
|
||||
"rssBytes": 327974912
|
||||
},
|
||||
{
|
||||
"fedMiB": 32,
|
||||
"socketBufferedBytes": 137492,
|
||||
"batcherQueuedChars": 786432,
|
||||
"rssBytes": 390053888
|
||||
},
|
||||
{
|
||||
"fedMiB": 64,
|
||||
"socketBufferedBytes": 144389,
|
||||
"batcherQueuedChars": 720896,
|
||||
"rssBytes": 454852608
|
||||
},
|
||||
{
|
||||
"fedMiB": 96,
|
||||
"socketBufferedBytes": 151286,
|
||||
"batcherQueuedChars": 655360,
|
||||
"rssBytes": 519733248
|
||||
}
|
||||
],
|
||||
"finalRetainedQueuedPayloadBytes": 806646,
|
||||
"queuedBytesPerProducedByteAfter64MiB": -0.0017475783824920654,
|
||||
"afterDrain": {
|
||||
"fedMiB": 96,
|
||||
"socketBufferedBytes": 0,
|
||||
"batcherQueuedChars": 0,
|
||||
"rssBytes": 520437760
|
||||
},
|
||||
"drainMs": 5.139125000000149,
|
||||
"receivedBytes": 1595433
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -0,0 +1,120 @@
|
||||
import { once } from 'node:events'
|
||||
import { createServer, Socket } from 'node:net'
|
||||
import { setImmediate as nextTurn, setTimeout as delay } from 'node:timers/promises'
|
||||
import { createHash } from 'node:crypto'
|
||||
import { fileURLToPath } from 'node:url'
|
||||
import { resolve } from 'node:path'
|
||||
import { build } from 'esbuild'
|
||||
|
||||
if (process.env.ORCA_BACKGROUND_LAUNCH !== '1') {
|
||||
throw new Error('Run with ORCA_BACKGROUND_LAUNCH=1')
|
||||
}
|
||||
const root = fileURLToPath(new URL('../../../', import.meta.url))
|
||||
const built = await build({
|
||||
entryPoints: [resolve(root, 'src/main/daemon/daemon-stream-data-batcher.ts')],
|
||||
bundle: true,
|
||||
platform: 'node',
|
||||
format: 'esm',
|
||||
write: false
|
||||
})
|
||||
const bundle = built.outputFiles[0].text
|
||||
const { DaemonStreamDataBatcher } = await import(
|
||||
`data:text/javascript;base64,${Buffer.from(bundle).toString('base64')}`
|
||||
)
|
||||
const MiB = 1024 * 1024
|
||||
|
||||
async function run(chunkSize, hidden) {
|
||||
const server = createServer()
|
||||
const writer = new Socket()
|
||||
let reader
|
||||
let paused = false
|
||||
let produced = 0
|
||||
let receivedBytes = 0
|
||||
const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: writer }), {
|
||||
onProducerBackpressureChanged: (_sessionId, value) => {
|
||||
paused = value
|
||||
},
|
||||
isSessionDroppable: () => hidden
|
||||
})
|
||||
writer.on('drain', () => batcher.flush('client'))
|
||||
try {
|
||||
const accepted = once(server, 'connection')
|
||||
server.listen(0, '127.0.0.1')
|
||||
await once(server, 'listening')
|
||||
const address = server.address()
|
||||
if (!address || typeof address === 'string') {
|
||||
throw new Error('Expected TCP address')
|
||||
}
|
||||
writer.connect(address.port, '127.0.0.1')
|
||||
await once(writer, 'connect')
|
||||
;[reader] = await accepted
|
||||
reader.on('data', (data) => {
|
||||
receivedBytes += data.length
|
||||
})
|
||||
reader.pause()
|
||||
const chunk = 'x'.repeat(chunkSize)
|
||||
while (!paused && produced < 8 * MiB) {
|
||||
batcher.enqueue('client', 'session', chunk, { flushImmediately: true, flushMaxChars: 1024 })
|
||||
batcher.flush('client')
|
||||
produced += chunkSize
|
||||
if (produced % MiB === 0) {
|
||||
await nextTurn()
|
||||
}
|
||||
}
|
||||
const stalled = {
|
||||
producedChars: produced,
|
||||
producerPaused: paused,
|
||||
socketBufferedBytes: writer.writableLength,
|
||||
batcherQueuedChars: batcher.queuedCharsForClient('client'),
|
||||
rssBytes: process.memoryUsage().rss
|
||||
}
|
||||
reader.resume()
|
||||
const start = performance.now()
|
||||
while (paused || writer.writableLength || batcher.queuedCharsForClient('client')) {
|
||||
if (performance.now() - start > 5000) {
|
||||
throw new Error('Failed to drain')
|
||||
}
|
||||
await delay(5)
|
||||
}
|
||||
const ended = once(reader, 'end')
|
||||
writer.end()
|
||||
await ended
|
||||
return {
|
||||
chunkSize,
|
||||
hidden,
|
||||
stalled,
|
||||
receivedBytes,
|
||||
afterDrain: {
|
||||
producerPaused: paused,
|
||||
socketBufferedBytes: writer.writableLength,
|
||||
batcherQueuedChars: batcher.queuedCharsForClient('client')
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
batcher.clear()
|
||||
writer.destroy()
|
||||
reader?.destroy()
|
||||
await new Promise((done) => server.close(done))
|
||||
}
|
||||
}
|
||||
|
||||
const cases = [await run(64 * 1024, false), await run(1024, false), await run(1024, true)]
|
||||
if (!cases[0].stalled.producerPaused || !cases[1].stalled.producerPaused) {
|
||||
throw new Error('Visible producer did not pause under the stalled-reader budget')
|
||||
}
|
||||
if (cases[2].stalled.producerPaused || cases[2].stalled.producedChars !== 8 * MiB) {
|
||||
throw new Error('Droppable producer paused or failed to process the full reproduction input')
|
||||
}
|
||||
|
||||
console.log(
|
||||
JSON.stringify(
|
||||
{
|
||||
node: process.version,
|
||||
platform: process.platform,
|
||||
bundleSha256: createHash('sha256').update(bundle).digest('hex'),
|
||||
cases
|
||||
},
|
||||
null,
|
||||
2
|
||||
)
|
||||
)
|
||||
@@ -5,6 +5,10 @@ import type { DaemonStreamDataBatcher } from './daemon-stream-data-batcher'
|
||||
import { createNdjsonParser, encodeNdjson } from './ndjson'
|
||||
import type { DaemonRequest, HelloMessage } from './types'
|
||||
|
||||
// Idle time before the first probe. How long the close then takes is the OS's probe schedule, not
|
||||
// ours, which is why the producer-stall watchdog never waits on it.
|
||||
const STREAM_SOCKET_KEEPALIVE_DELAY_MS = 30_000
|
||||
|
||||
export type ConnectedDaemonClient = {
|
||||
clientId: string
|
||||
controlSocket: Socket
|
||||
@@ -229,6 +233,11 @@ export class DaemonClientConnections {
|
||||
private installStreamSocket(socket: Socket, client: ConnectedDaemonClient): void {
|
||||
const previous = client.streamSocket
|
||||
socket.removeAllListeners('data')
|
||||
// A half-open peer (slept laptop, dropped NAT state) stops draining without closing, which would
|
||||
// otherwise hold a session's producer pause open with no event to release it. Kernel probes give
|
||||
// that peer a close. TCP only: a no-op over a local pipe, and over SSH it is the relay's own link
|
||||
// that dies — the producer-stall watchdog, not this, is what bounds those.
|
||||
socket.setKeepAlive(true, STREAM_SOCKET_KEEPALIVE_DELAY_MS)
|
||||
client.streamSocket = socket
|
||||
socket.on('drain', () => this.options.streamDataBatcher.flush(client.clientId))
|
||||
const cleanup = (): void => {
|
||||
@@ -244,6 +253,8 @@ export class DaemonClientConnections {
|
||||
socket.on('error', cleanup)
|
||||
if (previous && previous !== socket) {
|
||||
previous.destroy()
|
||||
this.options.streamDataBatcher.replaceStream(client.clientId)
|
||||
this.options.streamDataBatcher.flush(client.clientId)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,103 @@
|
||||
import { EventEmitter } from 'node:events'
|
||||
import type { Socket } from 'node:net'
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { DaemonClientConnections } from './daemon-client-connections'
|
||||
import type { DaemonFileLog } from './daemon-file-log'
|
||||
import type { DaemonStreamDataBatcher } from './daemon-stream-data-batcher'
|
||||
import { encodeNdjson } from './ndjson'
|
||||
|
||||
const PROTOCOL_VERSION = 42
|
||||
const TOKEN = 'token'
|
||||
|
||||
class FakeSocket extends EventEmitter {
|
||||
destroyed = false
|
||||
readonly setKeepAlive = vi.fn()
|
||||
|
||||
write(): boolean {
|
||||
return true
|
||||
}
|
||||
|
||||
end(): void {
|
||||
this.destroy()
|
||||
}
|
||||
|
||||
destroy(): void {
|
||||
if (this.destroyed) {
|
||||
return
|
||||
}
|
||||
this.destroyed = true
|
||||
this.emit('close')
|
||||
}
|
||||
|
||||
hello(role: 'control' | 'stream'): void {
|
||||
this.emit(
|
||||
'data',
|
||||
Buffer.from(
|
||||
encodeNdjson({
|
||||
type: 'hello',
|
||||
version: PROTOCOL_VERSION,
|
||||
token: TOKEN,
|
||||
role,
|
||||
clientId: 'client'
|
||||
})
|
||||
)
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
function connect() {
|
||||
const onStreamDisconnected = vi.fn()
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the connection path only reaches flush/clear/replaceStream on the batcher.
|
||||
const streamDataBatcher = {
|
||||
flush: vi.fn(),
|
||||
clear: vi.fn(),
|
||||
replaceStream: vi.fn()
|
||||
} as unknown as DaemonStreamDataBatcher
|
||||
const connections = new DaemonClientConnections({
|
||||
token: TOKEN,
|
||||
protocolVersion: PROTOCOL_VERSION,
|
||||
identity: {
|
||||
launchNonce: null,
|
||||
startedAtMs: null,
|
||||
entryPath: null,
|
||||
appVersion: null,
|
||||
spawnerExecPath: null
|
||||
},
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: only log() is called.
|
||||
log: { log: vi.fn() } as unknown as DaemonFileLog,
|
||||
streamDataBatcher,
|
||||
isAcceptingWork: () => true,
|
||||
onTransportChanged: vi.fn(),
|
||||
onConnectionAccepted: vi.fn(),
|
||||
onAuthenticatedPair: vi.fn(),
|
||||
onLastAuthenticatedClientDisconnected: vi.fn(),
|
||||
onControlRequest: vi.fn(),
|
||||
onControlReplaced: vi.fn(),
|
||||
onClientDisconnected: vi.fn(),
|
||||
onStreamDisconnected
|
||||
})
|
||||
const control = new FakeSocket()
|
||||
const stream = new FakeSocket()
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: accept() uses only the EventEmitter surface plus write/end/destroy, all implemented by FakeSocket.
|
||||
connections.accept(control as unknown as Socket)
|
||||
control.hello('control')
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: as above.
|
||||
connections.accept(stream as unknown as Socket)
|
||||
stream.hello('stream')
|
||||
return { connections, control, onStreamDisconnected, stream }
|
||||
}
|
||||
|
||||
describe('daemon stream socket keepalive', () => {
|
||||
it('probes the stream peer so a half-open link eventually closes', () => {
|
||||
const { stream } = connect()
|
||||
expect(stream.setKeepAlive).toHaveBeenCalledWith(true, 30_000)
|
||||
})
|
||||
|
||||
it('reports the disconnect that a failed keepalive probe produces', () => {
|
||||
const { onStreamDisconnected, stream } = connect()
|
||||
expect(onStreamDisconnected).not.toHaveBeenCalled()
|
||||
// What a keepalive timeout looks like at this layer: the socket errors, then closes.
|
||||
stream.emit('error', new Error('ETIMEDOUT'))
|
||||
expect(onStreamDisconnected).toHaveBeenCalledWith('client')
|
||||
})
|
||||
})
|
||||
@@ -74,6 +74,14 @@ export class DaemonServer {
|
||||
this.streamDataBatcher = new DaemonStreamDataBatcher(
|
||||
(clientId) => this.connections.get(clientId),
|
||||
{
|
||||
onProducerBackpressureChanged: (sessionId, paused, onStallTimeout) =>
|
||||
paused
|
||||
? this.host.pauseProducer(sessionId, 'stream', onStallTimeout)
|
||||
: this.host.resumeProducer(sessionId, 'stream'),
|
||||
isSessionAttachedToClient: (clientId, sessionId) => {
|
||||
const owner = this.attachments.clientIdForSession(sessionId)
|
||||
return owner === undefined || owner === clientId
|
||||
},
|
||||
isSessionDroppable: (sessionId) =>
|
||||
BACKGROUND_STREAM_DROP_ENABLED && this.transientFactRelay.isBackgrounded(sessionId),
|
||||
salvageDroppedData: (dropped) => {
|
||||
@@ -128,6 +136,7 @@ export class DaemonServer {
|
||||
onControlReplaced: (clientId) => {
|
||||
this.preparations.cancelForClient(clientId)
|
||||
this.historySeedTransfers.clearOwner(clientId)
|
||||
this.streamDataBatcher.clear(clientId)
|
||||
},
|
||||
onClientDisconnected: (clientId) => {
|
||||
this.preparations.cancelForClient(clientId)
|
||||
|
||||
@@ -0,0 +1,145 @@
|
||||
import { once } from 'node:events'
|
||||
import { createServer, Socket } from 'node:net'
|
||||
import { setImmediate as nextTurn, setTimeout as delay } from 'node:timers/promises'
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { DaemonStreamDataBatcher } from './daemon-stream-data-batcher'
|
||||
import { createNdjsonParser } from './ndjson'
|
||||
import { SessionProducerPause } from './session-producer-pause'
|
||||
|
||||
const MiB = 1024 * 1024
|
||||
|
||||
describe('daemon stream backpressure with a stalled Socket reader', () => {
|
||||
it.each([
|
||||
{ chunkSize: 64 * 1024, hidden: false },
|
||||
{ chunkSize: 1024, hidden: false },
|
||||
{ chunkSize: 64 * 1024, hidden: true },
|
||||
{ chunkSize: 1024, hidden: true }
|
||||
])(
|
||||
'bounds $chunkSize-character writes (hidden=$hidden) and resumes',
|
||||
async ({ chunkSize, hidden }) => {
|
||||
const server = createServer()
|
||||
const writer = new Socket()
|
||||
let reader: Socket | undefined
|
||||
let paused = false
|
||||
let produced = 0
|
||||
let dropped = 0
|
||||
const received = new Map<string, number>()
|
||||
const producer = new SessionProducerPause({
|
||||
pause: () => {
|
||||
paused = true
|
||||
},
|
||||
resume: () => {
|
||||
paused = false
|
||||
}
|
||||
})
|
||||
const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: writer }), {
|
||||
isSessionDroppable: (sessionId) => hidden && sessionId === 'flood',
|
||||
onProducerBackpressureChanged: (sessionId, value) => {
|
||||
expect(sessionId).toBe('flood')
|
||||
producer.setStreamBackpressured(value)
|
||||
}
|
||||
})
|
||||
writer.on('drain', () => batcher.flush('client'))
|
||||
try {
|
||||
const accepted = once(server, 'connection')
|
||||
server.listen(0, '127.0.0.1')
|
||||
await once(server, 'listening')
|
||||
const address = server.address()
|
||||
if (!address || typeof address === 'string') {
|
||||
throw new Error('Expected TCP address')
|
||||
}
|
||||
writer.connect(address.port, '127.0.0.1')
|
||||
await once(writer, 'connect')
|
||||
const [acceptedSocket] = await accepted
|
||||
if (!(acceptedSocket instanceof Socket)) {
|
||||
throw new Error('Expected accepted Socket')
|
||||
}
|
||||
reader = acceptedSocket
|
||||
const parser = createNdjsonParser(
|
||||
(event) => {
|
||||
if (
|
||||
!event ||
|
||||
typeof event !== 'object' ||
|
||||
!('sessionId' in event) ||
|
||||
typeof event.sessionId !== 'string' ||
|
||||
!('payload' in event)
|
||||
) {
|
||||
return
|
||||
}
|
||||
const payload = event.payload
|
||||
if (
|
||||
payload &&
|
||||
typeof payload === 'object' &&
|
||||
'droppedChars' in payload &&
|
||||
typeof payload.droppedChars === 'number'
|
||||
) {
|
||||
dropped += payload.droppedChars
|
||||
}
|
||||
if (
|
||||
!payload ||
|
||||
typeof payload !== 'object' ||
|
||||
!('data' in payload) ||
|
||||
typeof payload.data !== 'string'
|
||||
) {
|
||||
return
|
||||
}
|
||||
received.set(
|
||||
event.sessionId,
|
||||
(received.get(event.sessionId) ?? 0) + payload.data.length
|
||||
)
|
||||
},
|
||||
() => {
|
||||
throw new Error('Invalid stream frame')
|
||||
}
|
||||
)
|
||||
reader.on('data', (data) => parser.feed(data.toString('utf8')))
|
||||
reader.pause()
|
||||
|
||||
const chunk = 'x'.repeat(chunkSize)
|
||||
while (!paused && produced < 8 * MiB) {
|
||||
batcher.enqueue('client', 'flood', chunk, {
|
||||
flushImmediately: chunkSize <= 1024,
|
||||
flushMaxChars: 1024
|
||||
})
|
||||
batcher.flush('client')
|
||||
produced += chunkSize
|
||||
if (produced % MiB === 0) {
|
||||
await nextTurn()
|
||||
}
|
||||
}
|
||||
expect(paused).toBe(!hidden)
|
||||
expect(writer.writableLength + 2 * batcher.queuedCharsForClient('client')).toBeLessThan(
|
||||
5 * MiB
|
||||
)
|
||||
|
||||
if (!hidden) {
|
||||
producer.pause()
|
||||
producer.resumeClient()
|
||||
expect(paused).toBe(true)
|
||||
}
|
||||
batcher.enqueue('client', 'typing', 'echo', { flushImmediately: true, flushMaxChars: 1024 })
|
||||
reader.resume()
|
||||
const startedAt = performance.now()
|
||||
while (paused || writer.writableLength || batcher.queuedCharsForClient('client')) {
|
||||
if (performance.now() - startedAt > 5_000) {
|
||||
throw new Error('Stream failed to drain')
|
||||
}
|
||||
await delay(5)
|
||||
}
|
||||
const ended = once(reader, 'end')
|
||||
writer.end()
|
||||
await ended
|
||||
expect((received.get('flood') ?? 0) + dropped).toBe(produced)
|
||||
expect(dropped > 0).toBe(hidden)
|
||||
expect(received.get('typing')).toBe(4)
|
||||
expect(paused).toBe(false)
|
||||
} finally {
|
||||
batcher.clear()
|
||||
producer.release({ resume: false })
|
||||
writer.destroy()
|
||||
reader?.destroy()
|
||||
await new Promise<void>((resolve) => server.close(() => resolve()))
|
||||
}
|
||||
}
|
||||
)
|
||||
})
|
||||
@@ -0,0 +1,201 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { DaemonStreamBackpressure } from './daemon-stream-backpressure'
|
||||
import { DaemonStreamHeldRefill } from './daemon-stream-held-refill'
|
||||
|
||||
const MiB = 1024 * 1024
|
||||
|
||||
function createBackpressure(isDroppable: (sessionId: string) => boolean = () => false) {
|
||||
const paused = new Set<string>()
|
||||
const setPaused = vi.fn((sessionId: string, value: boolean) => {
|
||||
if (value) {
|
||||
paused.add(sessionId)
|
||||
} else {
|
||||
paused.delete(sessionId)
|
||||
}
|
||||
})
|
||||
return { pressure: new DaemonStreamBackpressure(setPaused, isDroppable), paused, setPaused }
|
||||
}
|
||||
|
||||
describe('DaemonStreamBackpressure', () => {
|
||||
it('aggregates held output across clients and spares low-output sessions', () => {
|
||||
const { pressure, paused } = createBackpressure()
|
||||
pressure.setQueued(
|
||||
'a',
|
||||
new Map([
|
||||
['flood-a', MiB],
|
||||
['typing', 4]
|
||||
])
|
||||
)
|
||||
expect(paused.size).toBe(0)
|
||||
pressure.setQueued('b', new Map([['flood-b', MiB]]))
|
||||
expect([...paused].sort()).toEqual(['flood-a', 'flood-b'])
|
||||
pressure.clear()
|
||||
expect(paused.size).toBe(0)
|
||||
})
|
||||
|
||||
it('keeps a shared producer paused until every slow client releases its backlog', () => {
|
||||
const { pressure, paused } = createBackpressure()
|
||||
pressure.setQueued('a', new Map([['shared', MiB]]))
|
||||
pressure.setQueued('b', new Map([['shared', MiB]]))
|
||||
expect(paused.has('shared')).toBe(true)
|
||||
pressure.clear('a')
|
||||
expect(paused.has('shared')).toBe(true)
|
||||
pressure.clear('b')
|
||||
expect(paused.has('shared')).toBe(false)
|
||||
})
|
||||
|
||||
it('releases a drained session while another client remains stalled', () => {
|
||||
const { pressure, paused } = createBackpressure()
|
||||
pressure.setQueued('a', new Map([['flood-a', 2 * MiB]]))
|
||||
pressure.setQueued('b', new Map([['flood-b', MiB]]))
|
||||
expect(paused.has('flood-b')).toBe(true)
|
||||
pressure.setQueued('b', new Map())
|
||||
expect([...paused]).toEqual(['flood-a'])
|
||||
})
|
||||
|
||||
it('leaves background shedding in control when a session becomes hidden', () => {
|
||||
let hidden = false
|
||||
const { pressure, paused } = createBackpressure(() => hidden)
|
||||
pressure.setQueued('a', new Map([['flood', 2 * MiB]]))
|
||||
expect(paused.has('flood')).toBe(true)
|
||||
hidden = true
|
||||
pressure.refresh()
|
||||
expect(paused.size).toBe(0)
|
||||
pressure.setQueued('a', new Map([['flood', 3 * MiB]]))
|
||||
expect(paused.size).toBe(0)
|
||||
})
|
||||
|
||||
it('uses hysteresis and resumes after the aggregate falls below the low watermark', () => {
|
||||
const { pressure, paused } = createBackpressure()
|
||||
pressure.setQueued('a', new Map([['flood', 2 * MiB]]))
|
||||
pressure.setQueued('a', new Map([['flood', MiB]]))
|
||||
expect(paused.has('flood')).toBe(true)
|
||||
pressure.setQueued('a', new Map([['flood', MiB / 2]]))
|
||||
expect(paused.size).toBe(0)
|
||||
})
|
||||
|
||||
it('does not let a superseded subscription pause the new attachment owner', () => {
|
||||
let owner = 'old'
|
||||
const setPaused = vi.fn()
|
||||
const pressure = new DaemonStreamBackpressure(
|
||||
setPaused,
|
||||
() => false,
|
||||
(clientId) => clientId === owner
|
||||
)
|
||||
pressure.setQueued('old', new Map([['session', 2 * MiB]]))
|
||||
expect(setPaused).toHaveBeenLastCalledWith('session', true, expect.any(Function))
|
||||
owner = 'new'
|
||||
pressure.refresh()
|
||||
expect(setPaused).toHaveBeenLastCalledWith('session', false)
|
||||
setPaused.mockClear()
|
||||
pressure.setQueued('new', new Map([['session', 1024]]))
|
||||
expect(setPaused).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('bounds undroppable control output even when the producer is backgrounded', () => {
|
||||
const { pressure, paused } = createBackpressure(() => true)
|
||||
pressure.setQueued('a', new Map([['hidden', 2 * MiB]]), new Map([['hidden', 4 * MiB]]))
|
||||
expect(paused.has('hidden')).toBe(true)
|
||||
pressure.setQueued('a', new Map([['hidden', 2 * MiB]]))
|
||||
expect(paused.size).toBe(0)
|
||||
})
|
||||
|
||||
it('ignores old write completions after a client and session id are reused', () => {
|
||||
const completions: (() => void)[] = []
|
||||
const socket = {
|
||||
write(...args: unknown[]): boolean {
|
||||
const complete = args[1]
|
||||
if (typeof complete === 'function') {
|
||||
completions.push(() => complete())
|
||||
}
|
||||
return false
|
||||
}
|
||||
}
|
||||
const { pressure, paused } = createBackpressure()
|
||||
const line = 'x'.repeat(4 * MiB)
|
||||
pressure.write('client', 'session', socket, line)
|
||||
expect(paused.has('session')).toBe(true)
|
||||
pressure.clear('client')
|
||||
expect(paused.size).toBe(0)
|
||||
pressure.write('client', 'session', socket, line)
|
||||
completions.shift()?.()
|
||||
expect(paused.has('session')).toBe(true)
|
||||
completions.shift()?.()
|
||||
expect(paused.size).toBe(0)
|
||||
})
|
||||
|
||||
it('hands the producer a stall release for the pause it is arming', () => {
|
||||
const stallTimeouts = new Map<string, () => void>()
|
||||
const paused = new Set<string>()
|
||||
const pressure = new DaemonStreamBackpressure(
|
||||
(sessionId, value, onStallTimeout) => {
|
||||
if (value) {
|
||||
paused.add(sessionId)
|
||||
if (onStallTimeout) {
|
||||
stallTimeouts.set(sessionId, onStallTimeout)
|
||||
}
|
||||
} else {
|
||||
paused.delete(sessionId)
|
||||
}
|
||||
},
|
||||
() => false
|
||||
)
|
||||
pressure.setQueued('a', new Map([['flood', 2 * MiB]]))
|
||||
expect(paused.has('flood')).toBe(true)
|
||||
// The un-pause the watchdog reaches back through.
|
||||
stallTimeouts.get('flood')?.()
|
||||
expect(paused.has('flood')).toBe(false)
|
||||
expect(pressure.isStallReleased('flood')).toBe(true)
|
||||
})
|
||||
|
||||
it('will not re-pause a stall-released session whose backlog is still undelivered', () => {
|
||||
const stallReleased = new Set<string>()
|
||||
const { pressure, paused } = createBackpressure((sessionId) => stallReleased.has(sessionId))
|
||||
pressure.setQueued('a', new Map([['flood', 2 * MiB]]))
|
||||
expect(paused.has('flood')).toBe(true)
|
||||
stallReleased.add('flood')
|
||||
pressure.releaseStalledSession('flood')
|
||||
expect(paused.size).toBe(0)
|
||||
// Undroppable frames the keep-tail cannot shed must not drag the producer back under.
|
||||
pressure.setQueued('a', new Map([['flood', 2 * MiB]]), new Map([['flood', 4 * MiB]]))
|
||||
expect(paused.size).toBe(0)
|
||||
})
|
||||
|
||||
it('reconciles droppability and restores ordinary pausing once the backlog is gone', () => {
|
||||
const onDroppabilityChanged = vi.fn()
|
||||
const stallReleased = new Set<string>()
|
||||
const setPaused = vi.fn()
|
||||
const pressure = new DaemonStreamBackpressure(
|
||||
setPaused,
|
||||
(sessionId) => stallReleased.has(sessionId),
|
||||
() => true,
|
||||
onDroppabilityChanged
|
||||
)
|
||||
pressure.setQueued('a', new Map([['flood', 2 * MiB]]))
|
||||
stallReleased.add('flood')
|
||||
pressure.releaseStalledSession('flood')
|
||||
expect(onDroppabilityChanged).toHaveBeenCalledWith('flood')
|
||||
expect(pressure.isStallReleased('flood')).toBe(true)
|
||||
|
||||
stallReleased.delete('flood')
|
||||
pressure.setQueued('a', new Map())
|
||||
expect(pressure.isStallReleased('flood')).toBe(false)
|
||||
setPaused.mockClear()
|
||||
pressure.setQueued('a', new Map([['flood', 2 * MiB]]))
|
||||
expect(setPaused).toHaveBeenCalledWith('flood', true, expect.any(Function))
|
||||
})
|
||||
|
||||
it('does not let an old refill callback flush a replacement connection', () => {
|
||||
const flush = vi.fn()
|
||||
const completions: (() => void)[] = []
|
||||
const write = (_line: string, complete: () => void) => completions.push(complete)
|
||||
const refill = new DaemonStreamHeldRefill(flush)
|
||||
refill.arm('client', 'session', write)
|
||||
refill.clear('client')
|
||||
refill.arm('client', 'session', write)
|
||||
completions.shift()?.()
|
||||
expect(flush).not.toHaveBeenCalled()
|
||||
completions.shift()?.()
|
||||
expect(flush).toHaveBeenCalledOnce()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,184 @@
|
||||
import type { Socket } from 'node:net'
|
||||
import { STREAM_ENTRY_OVERHEAD_BYTES } from './daemon-stream-entry-accounting'
|
||||
|
||||
const HIGH_WATER_BYTES = 4 * 1024 * 1024
|
||||
const LOW_WATER_BYTES = 1024 * 1024
|
||||
const SESSION_HIGH_WATER_BYTES = 64 * 1024
|
||||
const SESSION_LOW_WATER_BYTES = 32 * 1024
|
||||
|
||||
type ClientBacklog = {
|
||||
queuedChars: Map<string, number>
|
||||
queuedMetadataBytes: Map<string, number>
|
||||
pendingWriteBytes: Map<string, number>
|
||||
}
|
||||
|
||||
/** Counts held strings and socket writes together, including small writes that bypass the hold. */
|
||||
export class DaemonStreamBackpressure {
|
||||
private readonly clients = new Map<string, ClientBacklog>()
|
||||
private readonly pausedSessions = new Set<string>()
|
||||
// Sessions the producer-stall watchdog gave up on. Keep-tail dropping bounds them now, so re-pausing
|
||||
// would only re-freeze the shell behind the same unreachable consumer.
|
||||
private readonly stallReleasedSessions = new Set<string>()
|
||||
private pressured = false
|
||||
|
||||
constructor(
|
||||
private readonly setPaused: (
|
||||
sessionId: string,
|
||||
paused: boolean,
|
||||
onStallTimeout?: () => void
|
||||
) => void,
|
||||
private readonly isDroppable: (sessionId: string) => boolean,
|
||||
private readonly ownsSession: (clientId: string, sessionId: string) => boolean = () => true,
|
||||
// Already-queued data is pinned to the droppable membership cached when it was enqueued, so a
|
||||
// stall release has to ask the batcher to reconcile it.
|
||||
private readonly onDroppabilityChanged: (sessionId: string) => void = () => {}
|
||||
) {}
|
||||
|
||||
setQueued(
|
||||
clientId: string,
|
||||
queuedChars: ReadonlyMap<string, number>,
|
||||
queuedMetadataBytes: ReadonlyMap<string, number> = new Map()
|
||||
): void {
|
||||
const client = this.getOrCreateClient(clientId)
|
||||
client.queuedChars = new Map(queuedChars)
|
||||
client.queuedMetadataBytes = new Map(queuedMetadataBytes)
|
||||
this.pruneClient(clientId, client)
|
||||
this.refresh()
|
||||
}
|
||||
|
||||
write(
|
||||
clientId: string,
|
||||
sessionId: string,
|
||||
socket: Pick<Socket, 'write'>,
|
||||
line: string,
|
||||
onComplete?: () => void
|
||||
): void {
|
||||
const client = this.getOrCreateClient(clientId)
|
||||
const bytes = Buffer.byteLength(line) + STREAM_ENTRY_OVERHEAD_BYTES
|
||||
client.pendingWriteBytes.set(sessionId, (client.pendingWriteBytes.get(sessionId) ?? 0) + bytes)
|
||||
this.refresh()
|
||||
socket.write(line, () => {
|
||||
// A disconnected client's callbacks must not debit a replacement connection's writes.
|
||||
if (this.clients.get(clientId) === client) {
|
||||
const remaining = (client.pendingWriteBytes.get(sessionId) ?? 0) - bytes
|
||||
if (remaining > 0) {
|
||||
client.pendingWriteBytes.set(sessionId, remaining)
|
||||
} else {
|
||||
client.pendingWriteBytes.delete(sessionId)
|
||||
}
|
||||
this.pruneClient(clientId, client)
|
||||
this.refresh()
|
||||
}
|
||||
onComplete?.()
|
||||
})
|
||||
}
|
||||
|
||||
refresh(): void {
|
||||
const bytesBySession = new Map<string, number>()
|
||||
// Every session with anything accounted anywhere, droppable or not: a stall ends when the session's
|
||||
// last byte leaves the daemon, which the pausing totals alone cannot see.
|
||||
const accountedSessions = new Set<string>()
|
||||
let total = 0
|
||||
for (const [clientId, client] of this.clients) {
|
||||
for (const [sessionId, chars] of client.queuedChars) {
|
||||
accountedSessions.add(sessionId)
|
||||
// Background data has its own keep-tail budget; only its undroppable frames need pausing.
|
||||
if (this.isDroppable(sessionId)) {
|
||||
continue
|
||||
}
|
||||
const bytes = 2 * chars
|
||||
total += bytes
|
||||
if (this.ownsSession(clientId, sessionId)) {
|
||||
bytesBySession.set(sessionId, (bytesBySession.get(sessionId) ?? 0) + bytes)
|
||||
}
|
||||
}
|
||||
for (const counts of [client.queuedMetadataBytes, client.pendingWriteBytes]) {
|
||||
for (const [sessionId, bytes] of counts) {
|
||||
accountedSessions.add(sessionId)
|
||||
total += bytes
|
||||
if (this.ownsSession(clientId, sessionId)) {
|
||||
bytesBySession.set(sessionId, (bytesBySession.get(sessionId) ?? 0) + bytes)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if (total >= HIGH_WATER_BYTES) {
|
||||
this.pressured = true
|
||||
} else if (total <= LOW_WATER_BYTES) {
|
||||
this.pressured = false
|
||||
}
|
||||
|
||||
for (const sessionId of this.pausedSessions) {
|
||||
if (!this.pressured || (bytesBySession.get(sessionId) ?? 0) <= SESSION_LOW_WATER_BYTES) {
|
||||
this.pausedSessions.delete(sessionId)
|
||||
this.setPaused(sessionId, false)
|
||||
}
|
||||
}
|
||||
if (this.pressured) {
|
||||
for (const [sessionId, bytes] of bytesBySession) {
|
||||
if (bytes >= SESSION_HIGH_WATER_BYTES && !this.stallReleasedSessions.has(sessionId)) {
|
||||
this.pausedSessions.add(sessionId)
|
||||
// Detach/termination may have released the session's pause since the last update.
|
||||
this.setPaused(sessionId, true, () => this.releaseStalledSession(sessionId))
|
||||
}
|
||||
}
|
||||
}
|
||||
for (const sessionId of this.stallReleasedSessions) {
|
||||
// Nothing of this session's is queued or in flight any more, so the consumer caught up (or went
|
||||
// away) and ordinary producer pausing applies again.
|
||||
if (!accountedSessions.has(sessionId)) {
|
||||
this.stallReleasedSessions.delete(sessionId)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** The producer-stall watchdog gave up on this session's consumer. Hand its backlog to the keep-tail
|
||||
* machinery and let the PTY run: the user gets a dataGap (the renderer restores the pane from the
|
||||
* daemon's snapshot) rather than a shell frozen behind an unreachable client. Says nothing about the
|
||||
* child process — loss of contact is not evidence of an exit, and nothing here reports one. */
|
||||
releaseStalledSession(sessionId: string): void {
|
||||
this.stallReleasedSessions.add(sessionId)
|
||||
this.onDroppabilityChanged(sessionId)
|
||||
if (this.pausedSessions.delete(sessionId)) {
|
||||
this.setPaused(sessionId, false)
|
||||
}
|
||||
this.refresh()
|
||||
}
|
||||
|
||||
isStallReleased(sessionId: string): boolean {
|
||||
return this.stallReleasedSessions.has(sessionId)
|
||||
}
|
||||
|
||||
clear(clientId?: string): void {
|
||||
if (clientId === undefined) {
|
||||
this.clients.clear()
|
||||
this.stallReleasedSessions.clear()
|
||||
} else {
|
||||
this.clients.delete(clientId)
|
||||
}
|
||||
this.refresh()
|
||||
}
|
||||
|
||||
private getOrCreateClient(clientId: string): ClientBacklog {
|
||||
let client = this.clients.get(clientId)
|
||||
if (!client) {
|
||||
client = {
|
||||
queuedChars: new Map(),
|
||||
queuedMetadataBytes: new Map(),
|
||||
pendingWriteBytes: new Map()
|
||||
}
|
||||
this.clients.set(clientId, client)
|
||||
}
|
||||
return client
|
||||
}
|
||||
|
||||
private pruneClient(clientId: string, client: ClientBacklog): void {
|
||||
if (
|
||||
client.queuedChars.size === 0 &&
|
||||
client.queuedMetadataBytes.size === 0 &&
|
||||
client.pendingWriteBytes.size === 0
|
||||
) {
|
||||
this.clients.delete(clientId)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
export type DaemonStreamDataBatcherOptions = {
|
||||
maxLineBytes?: number
|
||||
/** onStallTimeout (pause only) bounds a pause whose consumer never drains and never closes. */
|
||||
onProducerBackpressureChanged?: (
|
||||
sessionId: string,
|
||||
paused: boolean,
|
||||
onStallTimeout?: () => void
|
||||
) => void
|
||||
isSessionAttachedToClient?: (clientId: string, sessionId: string) => boolean
|
||||
/** True for sessions whose queued output may be keep-tail dropped (main-marked background sessions). */
|
||||
isSessionDroppable?: (sessionId: string) => boolean
|
||||
/** Carve reply-eliciting query bytes (DSR/DA/DECRQM/OSC probes) out of dropped data — the hidden program blocks on the reply, so they must still be delivered even when their flood is not. */
|
||||
salvageDroppedData?: (dropped: string) => string
|
||||
}
|
||||
@@ -38,6 +38,30 @@ function nonSentinelWrites(streamSocket: { write: ReturnType<typeof vi.fn> }): P
|
||||
}
|
||||
|
||||
describe('DaemonStreamDataBatcher', () => {
|
||||
it.each(['', 'x'])('accounts held transformed entries with %s payload data', (data) => {
|
||||
vi.useFakeTimers()
|
||||
let paused = false
|
||||
const { batcher, streamSocket } = createBatcher({
|
||||
onProducerBackpressureChanged: (_sessionId, value) => {
|
||||
paused = value
|
||||
}
|
||||
})
|
||||
try {
|
||||
streamSocket.writableLength = 128 * 1024
|
||||
batcher.enqueue('client-1', 'session-1', 'bulk'.repeat(16 * 1024))
|
||||
for (let seq = 1; seq <= 20_000 && !paused; seq++) {
|
||||
batcher.enqueue('client-1', 'session-1', data, { transformed: true, rawLength: 1, seq })
|
||||
}
|
||||
expect(paused).toBe(true)
|
||||
expect(batcher.queuedCharsForClient('client-1')).toBeLessThan(100 * 1024)
|
||||
batcher.clear()
|
||||
expect(paused).toBe(false)
|
||||
} finally {
|
||||
batcher.clear()
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('coalesces background output before writing daemon stream events', () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
|
||||
@@ -1,18 +1,25 @@
|
||||
import type { Socket } from 'node:net'
|
||||
import { encodeNdjson, NDJSON_MAX_LINE_BYTES } from './ndjson'
|
||||
import { recordDaemonStreamBacklogEvent } from './daemon-stream-backlog-probe'
|
||||
import { DaemonStreamBackpressure } from './daemon-stream-backpressure'
|
||||
import {
|
||||
clampToSafeSplitIndex,
|
||||
encodeStreamDataEvent,
|
||||
writeStreamDataEvents
|
||||
} from './daemon-stream-data-split'
|
||||
accountDaemonStreamEntry,
|
||||
releaseDaemonStreamEntry
|
||||
} from './daemon-stream-entry-accounting'
|
||||
import { DaemonStreamHeldRefill } from './daemon-stream-held-refill'
|
||||
import { clampToSafeSplitIndex, writeStreamDataEvents } from './daemon-stream-data-split'
|
||||
import type { PendingStreamDataBatch } from './daemon-stream-keep-tail-drop'
|
||||
import type { DaemonEvent } from './types'
|
||||
import { appendDaemonStreamData, type DaemonStreamEnqueueOptions } from './daemon-stream-data-entry'
|
||||
import {
|
||||
appendDaemonStreamData,
|
||||
flushDaemonStreamSession,
|
||||
type DaemonStreamEnqueueOptions
|
||||
} from './daemon-stream-data-entry'
|
||||
import {
|
||||
evaluateDroppableEnqueue,
|
||||
refreshDroppableSessionMembership
|
||||
} from './daemon-stream-droppable-membership'
|
||||
import type { DaemonStreamDataBatcherOptions } from './daemon-stream-data-batcher-options'
|
||||
|
||||
type StreamDataClient = {
|
||||
streamSocket: Socket | null
|
||||
@@ -27,26 +34,17 @@ const SHALLOW_SOCKET_WRITE_GATE_BYTES =
|
||||
process.env.ORCA_DAEMON_SHALLOW_SOCKET_GATE === '0' ? Number.POSITIVE_INFINITY : 128 * 1024
|
||||
// Sliced writes: a coalesced entry can grow to megabytes; writing it whole would re-deepen the socket past the gate in one call.
|
||||
const BULK_WRITE_SLICE_CHARS = 64 * 1024
|
||||
// Safety valve: past this, write through — bounded daemon memory beats bounded echo latency in the extreme. Must sit FAR above the pacer's pause watermark + overshoot (~5MB) or an engaged valve buries interactive echo behind the whole backlog.
|
||||
// Last resort for handles without pause support; actual memory bounds come from producer backpressure.
|
||||
const HELD_WRITE_THROUGH_TOTAL_CHARS = 32 * 1024 * 1024
|
||||
// Small-session bypass: a few-KB session (echo, redraws, query replies) is never the flood, so it must not wait FIFO behind others' megabytes; backstops the 100ms interactive fast-path, which misses under event-loop load.
|
||||
const SMALL_SESSION_HOLD_BYPASS_CHARS = 4 * 1024
|
||||
|
||||
type DaemonStreamDataBatcherOptions = {
|
||||
maxLineBytes?: number
|
||||
/** Fires after each stream-socket write — the only place backlog grows, so the backlog pacer checks its watermark here. */
|
||||
onAfterSocketWrite?: () => void
|
||||
/** True for sessions whose queued output may be keep-tail dropped (main-marked background sessions). */
|
||||
isSessionDroppable?: (sessionId: string) => boolean
|
||||
/** Carve reply-eliciting query bytes (DSR/DA/DECRQM/OSC probes) out of dropped data — the hidden program blocks on the reply, so they must still be delivered even when their flood is not. */
|
||||
salvageDroppedData?: (dropped: string) => string
|
||||
}
|
||||
|
||||
export class DaemonStreamDataBatcher {
|
||||
private pendingByClient = new Map<string, PendingStreamDataBatch>()
|
||||
private getClient: (clientId: string) => StreamDataClient | undefined
|
||||
private maxLineBytes: number
|
||||
private onAfterSocketWrite: (() => void) | undefined
|
||||
private readonly backpressure: DaemonStreamBackpressure | undefined
|
||||
private readonly heldRefill = new DaemonStreamHeldRefill((clientId) => this.flush(clientId))
|
||||
private isSessionDroppable: (sessionId: string) => boolean
|
||||
private salvageDroppedData: (dropped: string) => string
|
||||
|
||||
@@ -56,9 +54,20 @@ export class DaemonStreamDataBatcher {
|
||||
) {
|
||||
this.getClient = getClient
|
||||
this.maxLineBytes = Math.max(1, options.maxLineBytes ?? NDJSON_MAX_LINE_BYTES)
|
||||
this.onAfterSocketWrite = options.onAfterSocketWrite
|
||||
this.isSessionDroppable = options.isSessionDroppable ?? (() => false)
|
||||
const isBackgroundDroppable = options.isSessionDroppable ?? (() => false)
|
||||
// A stall-released session is droppable for as long as its backlog survives, whether or not main
|
||||
// has backgrounded it: keep-tail thinning is what lets its producer run past an unreachable client.
|
||||
this.isSessionDroppable = (sessionId) =>
|
||||
this.backpressure?.isStallReleased(sessionId) === true || isBackgroundDroppable(sessionId)
|
||||
this.salvageDroppedData = options.salvageDroppedData ?? (() => '')
|
||||
this.backpressure = options.onProducerBackpressureChanged
|
||||
? new DaemonStreamBackpressure(
|
||||
options.onProducerBackpressureChanged,
|
||||
this.isSessionDroppable,
|
||||
options.isSessionAttachedToClient,
|
||||
(sessionId) => this.refreshSessionDroppability(sessionId)
|
||||
)
|
||||
: undefined
|
||||
}
|
||||
|
||||
enqueue(
|
||||
@@ -83,10 +92,13 @@ export class DaemonStreamDataBatcher {
|
||||
this.isSessionDroppable,
|
||||
this.salvageDroppedData
|
||||
)
|
||||
this.updateBackpressure(clientId, batch)
|
||||
|
||||
if (
|
||||
options.flushImmediately === true &&
|
||||
this.queuedCharsForSession(batch, sessionId, options.flushMaxChars) <=
|
||||
(!batch.droppableQueuedSessionIds.has(sessionId) ||
|
||||
client.streamSocket.writableLength < SHALLOW_SOCKET_WRITE_GATE_BYTES) &&
|
||||
(batch.queuedCharsBySession.get(sessionId) ?? 0) <=
|
||||
(options.flushMaxChars ?? Number.POSITIVE_INFINITY)
|
||||
) {
|
||||
this.flushSession(clientId, sessionId)
|
||||
@@ -104,7 +116,8 @@ export class DaemonStreamDataBatcher {
|
||||
return
|
||||
}
|
||||
const batch = this.getOrCreateBatch(clientId)
|
||||
batch.queue.push({ sessionId, data: '', control })
|
||||
batch.queue.push(accountDaemonStreamEntry(batch, { sessionId, data: '', control }))
|
||||
this.updateBackpressure(clientId, batch)
|
||||
if (!batch.timer) {
|
||||
batch.timer = setTimeout(() => this.flush(clientId), STREAM_DATA_BATCH_INTERVAL_MS)
|
||||
}
|
||||
@@ -113,6 +126,7 @@ export class DaemonStreamDataBatcher {
|
||||
refreshSessionDroppability(sessionId: string): void {
|
||||
const droppable = this.isSessionDroppable(sessionId)
|
||||
refreshDroppableSessionMembership(this.pendingByClient.values(), sessionId, droppable)
|
||||
this.backpressure?.refresh()
|
||||
}
|
||||
|
||||
private getOrCreateBatch(clientId: string): PendingStreamDataBatch {
|
||||
@@ -123,6 +137,7 @@ export class DaemonStreamDataBatcher {
|
||||
queue: [],
|
||||
queuedChars: 0,
|
||||
queuedCharsBySession: new Map(),
|
||||
queuedMetadataBytesBySession: new Map(),
|
||||
droppableQueuedSessionIds: new Set()
|
||||
}
|
||||
this.pendingByClient.set(clientId, batch)
|
||||
@@ -148,7 +163,7 @@ export class DaemonStreamDataBatcher {
|
||||
const client = this.getClient(clientId)
|
||||
if (!client?.streamSocket || client.streamSocket.destroyed) {
|
||||
// A vanished stream socket drops the batch — the model owns the bytes and reconnect restores from a snapshot.
|
||||
this.pendingByClient.delete(clientId)
|
||||
this.clear(clientId)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -158,22 +173,30 @@ export class DaemonStreamDataBatcher {
|
||||
const retained: PendingStreamDataBatch['queue'] = []
|
||||
while (batch.queue.length > 0) {
|
||||
const entry = batch.queue[0]
|
||||
const socketDeep = (socket.writableLength ?? 0) >= SHALLOW_SOCKET_WRITE_GATE_BYTES
|
||||
if (entry.control) {
|
||||
// Control entries only respect the held-session order latch; at ~100B, writing them onto a deep socket is as harmless as the small-session bypass.
|
||||
if (heldSessions.has(entry.sessionId)) {
|
||||
// Holding gaps lets repeated background drops coalesce without filling the socket with markers.
|
||||
if (
|
||||
heldSessions.has(entry.sessionId) ||
|
||||
(socketDeep && entry.control.event === 'dataGap')
|
||||
) {
|
||||
heldSessions.add(entry.sessionId)
|
||||
retained.push(entry)
|
||||
batch.queue.shift()
|
||||
continue
|
||||
}
|
||||
batch.queue.shift()
|
||||
socket.write(encodeNdjson(entry.control))
|
||||
this.onAfterSocketWrite?.()
|
||||
releaseDaemonStreamEntry(batch, entry)
|
||||
this.write(clientId, entry.sessionId, socket, encodeNdjson(entry.control))
|
||||
continue
|
||||
}
|
||||
const socketDeep = (socket.writableLength ?? 0) >= SHALLOW_SOCKET_WRITE_GATE_BYTES
|
||||
if (socketDeep && batch.queuedChars <= HELD_WRITE_THROUGH_TOTAL_CHARS) {
|
||||
const sessionHeld = batch.queuedCharsBySession.get(entry.sessionId) ?? 0
|
||||
if (heldSessions.has(entry.sessionId) || sessionHeld > SMALL_SESSION_HOLD_BYPASS_CHARS) {
|
||||
if (
|
||||
heldSessions.has(entry.sessionId) ||
|
||||
batch.droppableQueuedSessionIds.has(entry.sessionId) ||
|
||||
sessionHeld > SMALL_SESSION_HOLD_BYPASS_CHARS
|
||||
) {
|
||||
// Hold this flooding session's entry; small talkers keep flowing. No timer: a deep socket implies a prior false write(), so 'drain' (routed back to flush) is guaranteed to resume held bulk.
|
||||
heldSessions.add(entry.sessionId)
|
||||
retained.push(entry)
|
||||
@@ -200,6 +223,7 @@ export class DaemonStreamDataBatcher {
|
||||
: slice.length
|
||||
if (end >= entry.data.length) {
|
||||
batch.queue.shift()
|
||||
releaseDaemonStreamEntry(batch, entry)
|
||||
} else {
|
||||
entry.data = entry.data.slice(end)
|
||||
const remainingSequenceChars = entrySequenceChars - sliceSequenceChars
|
||||
@@ -216,7 +240,7 @@ export class DaemonStreamDataBatcher {
|
||||
batch.queuedCharsBySession.set(entry.sessionId, sessionHeldAfter)
|
||||
}
|
||||
writeStreamDataEvents(
|
||||
socket,
|
||||
{ write: (line) => this.write(clientId, entry.sessionId, socket, line) },
|
||||
entry.sessionId,
|
||||
slice,
|
||||
this.maxLineBytes,
|
||||
@@ -224,46 +248,42 @@ export class DaemonStreamDataBatcher {
|
||||
entry.seq,
|
||||
entry.transformed
|
||||
)
|
||||
this.onAfterSocketWrite?.()
|
||||
}
|
||||
this.updateBackpressure(clientId, batch)
|
||||
if (retained.length > 0) {
|
||||
batch.queue = retained
|
||||
// 'drain' only fires when the buffer fully empties (one gate-depth/turn = seconds for multi-MB backlogs); arm a no-op data event whose flush callback re-flushes while bytes are still in flight.
|
||||
this.armHeldQueueRefill(socket, clientId, retained[0].sessionId)
|
||||
if (!socket.destroyed) {
|
||||
const sessionId = retained[0].sessionId
|
||||
this.heldRefill.arm(clientId, sessionId, (line, complete) =>
|
||||
this.write(clientId, sessionId, socket, line, complete)
|
||||
)
|
||||
}
|
||||
return
|
||||
}
|
||||
this.pendingByClient.delete(clientId)
|
||||
}
|
||||
|
||||
private refillArmedClients = new Set<string>()
|
||||
|
||||
private armHeldQueueRefill(socket: Socket, clientId: string, sessionId: string): void {
|
||||
if (this.refillArmedClients.has(clientId) || socket.destroyed) {
|
||||
return
|
||||
private write(
|
||||
clientId: string,
|
||||
sessionId: string,
|
||||
socket: Socket,
|
||||
line: string,
|
||||
onComplete?: () => void
|
||||
): void {
|
||||
if (this.backpressure) {
|
||||
this.backpressure.write(clientId, sessionId, socket, line, onComplete)
|
||||
} else {
|
||||
socket.write(line, onComplete)
|
||||
}
|
||||
this.refillArmedClients.add(clientId)
|
||||
// Must be a real protocol no-op line, not an empty write: an empty write's callback fires immediately, defeating the in-flight re-flush.
|
||||
socket.write(encodeStreamDataEvent(sessionId, ''), () => {
|
||||
this.refillArmedClients.delete(clientId)
|
||||
this.flush(clientId)
|
||||
})
|
||||
}
|
||||
|
||||
private queuedCharsForSession(
|
||||
batch: PendingStreamDataBatch,
|
||||
sessionId: string,
|
||||
stopAfter = Number.POSITIVE_INFINITY
|
||||
): number {
|
||||
let chars = 0
|
||||
for (const entry of batch.queue) {
|
||||
if (entry.sessionId === sessionId) {
|
||||
chars += entry.data.length
|
||||
if (chars > stopAfter) {
|
||||
return chars
|
||||
}
|
||||
}
|
||||
}
|
||||
return chars
|
||||
private updateBackpressure(clientId: string, batch: PendingStreamDataBatch): void {
|
||||
this.backpressure?.setQueued(
|
||||
clientId,
|
||||
batch.queuedCharsBySession,
|
||||
batch.queuedMetadataBytesBySession
|
||||
)
|
||||
}
|
||||
|
||||
private flushSession(clientId: string, sessionId: string): void {
|
||||
@@ -272,55 +292,31 @@ export class DaemonStreamDataBatcher {
|
||||
return
|
||||
}
|
||||
|
||||
const flushed: PendingStreamDataBatch['queue'] = []
|
||||
const retained: PendingStreamDataBatch['queue'] = []
|
||||
let flushedChars = 0
|
||||
for (const entry of batch.queue) {
|
||||
if (entry.sessionId === sessionId) {
|
||||
flushed.push(entry)
|
||||
flushedChars += entry.data.length
|
||||
} else {
|
||||
retained.push(entry)
|
||||
}
|
||||
}
|
||||
if (flushed.length === 0) {
|
||||
return
|
||||
}
|
||||
|
||||
batch.queue = retained
|
||||
batch.queuedChars -= flushedChars
|
||||
batch.queuedCharsBySession.delete(sessionId)
|
||||
batch.droppableQueuedSessionIds.delete(sessionId)
|
||||
if (batch.queue.length === 0) {
|
||||
if (batch.timer) {
|
||||
clearTimeout(batch.timer)
|
||||
batch.timer = null
|
||||
}
|
||||
this.pendingByClient.delete(clientId)
|
||||
}
|
||||
|
||||
const client = this.getClient(clientId)
|
||||
if (!client?.streamSocket || client.streamSocket.destroyed) {
|
||||
this.clear(clientId)
|
||||
return
|
||||
}
|
||||
|
||||
for (const entry of flushed) {
|
||||
if (entry.control) {
|
||||
client.streamSocket.write(encodeNdjson(entry.control))
|
||||
this.onAfterSocketWrite?.()
|
||||
} else {
|
||||
writeStreamDataEvents(
|
||||
client.streamSocket,
|
||||
entry.sessionId,
|
||||
entry.data,
|
||||
this.maxLineBytes,
|
||||
entry.sequenceChars ?? entry.data.length,
|
||||
entry.seq,
|
||||
entry.transformed
|
||||
)
|
||||
this.onAfterSocketWrite?.()
|
||||
}
|
||||
const socket = client.streamSocket
|
||||
flushDaemonStreamSession(batch, sessionId, this.maxLineBytes, (line) =>
|
||||
this.write(clientId, sessionId, socket, line)
|
||||
)
|
||||
if (batch.queue.length === 0) {
|
||||
this.pendingByClient.delete(clientId)
|
||||
}
|
||||
this.updateBackpressure(clientId, batch)
|
||||
}
|
||||
|
||||
/** Reset socket-generation state without discarding queued payloads for a replacement stream. */
|
||||
replaceStream(clientId: string): void {
|
||||
const batch = this.pendingByClient.get(clientId)
|
||||
if (batch?.timer) {
|
||||
clearTimeout(batch.timer)
|
||||
batch.timer = null
|
||||
}
|
||||
this.heldRefill.clear(clientId)
|
||||
this.backpressure?.clear(clientId)
|
||||
}
|
||||
|
||||
clear(clientId?: string): void {
|
||||
@@ -335,5 +331,7 @@ export class DaemonStreamDataBatcher {
|
||||
}
|
||||
this.pendingByClient.delete(id)
|
||||
}
|
||||
this.heldRefill.clear(clientId)
|
||||
this.backpressure?.clear(clientId)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,4 +1,7 @@
|
||||
import type { PendingStreamDataBatch } from './daemon-stream-keep-tail-drop'
|
||||
import { writeStreamDataEvents } from './daemon-stream-data-split'
|
||||
import { encodeNdjson } from './ndjson'
|
||||
import { accountDaemonStreamEntry } from './daemon-stream-entry-accounting'
|
||||
|
||||
export type DaemonStreamEnqueueOptions = {
|
||||
flushImmediately?: boolean
|
||||
@@ -28,18 +31,68 @@ export function appendDaemonStreamData(
|
||||
last.sequenceChars = combinedRawLength === last.data.length ? undefined : combinedRawLength
|
||||
last.seq = options.seq
|
||||
} else {
|
||||
batch.queue.push({
|
||||
sessionId,
|
||||
data,
|
||||
...(options.rawLength === undefined || options.rawLength === data.length
|
||||
? {}
|
||||
: { sequenceChars: options.rawLength }),
|
||||
...(options.transformed ? { transformed: true } : {}),
|
||||
...(options.seq === undefined ? {} : { seq: options.seq })
|
||||
})
|
||||
batch.queue.push(
|
||||
accountDaemonStreamEntry(batch, {
|
||||
sessionId,
|
||||
data,
|
||||
...(options.rawLength === undefined || options.rawLength === data.length
|
||||
? {}
|
||||
: { sequenceChars: options.rawLength }),
|
||||
...(options.transformed ? { transformed: true } : {}),
|
||||
...(options.seq === undefined ? {} : { seq: options.seq })
|
||||
})
|
||||
)
|
||||
}
|
||||
batch.queuedChars += data.length
|
||||
const queuedAfter = (batch.queuedCharsBySession.get(sessionId) ?? 0) + data.length
|
||||
batch.queuedCharsBySession.set(sessionId, queuedAfter)
|
||||
return queuedAfter
|
||||
}
|
||||
|
||||
function takeDaemonStreamSession(
|
||||
batch: PendingStreamDataBatch,
|
||||
sessionId: string
|
||||
): PendingStreamDataBatch['queue'] {
|
||||
const flushed: PendingStreamDataBatch['queue'] = []
|
||||
const retained: PendingStreamDataBatch['queue'] = []
|
||||
for (const entry of batch.queue) {
|
||||
if (entry.sessionId === sessionId) {
|
||||
flushed.push(entry)
|
||||
batch.queuedChars -= entry.data.length
|
||||
} else {
|
||||
retained.push(entry)
|
||||
}
|
||||
}
|
||||
batch.queue = retained
|
||||
batch.queuedCharsBySession.delete(sessionId)
|
||||
batch.queuedMetadataBytesBySession.delete(sessionId)
|
||||
batch.droppableQueuedSessionIds.delete(sessionId)
|
||||
if (batch.queue.length === 0 && batch.timer) {
|
||||
clearTimeout(batch.timer)
|
||||
batch.timer = null
|
||||
}
|
||||
return flushed
|
||||
}
|
||||
|
||||
export function flushDaemonStreamSession(
|
||||
batch: PendingStreamDataBatch,
|
||||
sessionId: string,
|
||||
maxLineBytes: number,
|
||||
write: (line: string) => void
|
||||
): void {
|
||||
for (const entry of takeDaemonStreamSession(batch, sessionId)) {
|
||||
if (entry.control) {
|
||||
write(encodeNdjson(entry.control))
|
||||
} else {
|
||||
writeStreamDataEvents(
|
||||
{ write },
|
||||
entry.sessionId,
|
||||
entry.data,
|
||||
maxLineBytes,
|
||||
entry.sequenceChars ?? entry.data.length,
|
||||
entry.seq,
|
||||
entry.transformed
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,7 +4,6 @@
|
||||
* clamp shared by the batcher's bulk write slicing and keep-tail dropping.
|
||||
*/
|
||||
import { encodeNdjson } from './ndjson'
|
||||
import type { Socket } from 'node:net'
|
||||
|
||||
export function encodeStreamDataEvent(
|
||||
sessionId: string,
|
||||
@@ -113,7 +112,7 @@ function splitOversizedStreamDataForNdjson(
|
||||
}
|
||||
|
||||
export function writeStreamDataEvents(
|
||||
streamSocket: Pick<Socket, 'write'>,
|
||||
streamSocket: { write(data: string): void },
|
||||
sessionId: string,
|
||||
data: string,
|
||||
maxLineBytes: number,
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
import { encodeNdjson } from './ndjson'
|
||||
import type { PendingStreamDataBatch, StreamQueueEntry } from './daemon-stream-keep-tail-drop'
|
||||
|
||||
// Budget callbacks, stream requests and queue objects even when their data payload is empty.
|
||||
export const STREAM_ENTRY_OVERHEAD_BYTES = 256
|
||||
|
||||
export function accountDaemonStreamEntry(
|
||||
batch: PendingStreamDataBatch,
|
||||
entry: StreamQueueEntry
|
||||
): StreamQueueEntry {
|
||||
entry.retainedBytes =
|
||||
STREAM_ENTRY_OVERHEAD_BYTES +
|
||||
(entry.control ? Buffer.byteLength(encodeNdjson(entry.control)) : 0)
|
||||
batch.queuedMetadataBytesBySession.set(
|
||||
entry.sessionId,
|
||||
(batch.queuedMetadataBytesBySession.get(entry.sessionId) ?? 0) + entry.retainedBytes
|
||||
)
|
||||
return entry
|
||||
}
|
||||
|
||||
export function releaseDaemonStreamEntry(
|
||||
batch: PendingStreamDataBatch,
|
||||
entry: StreamQueueEntry
|
||||
): void {
|
||||
const remaining =
|
||||
(batch.queuedMetadataBytesBySession.get(entry.sessionId) ?? 0) - (entry.retainedBytes ?? 0)
|
||||
if (remaining > 0) {
|
||||
batch.queuedMetadataBytesBySession.set(entry.sessionId, remaining)
|
||||
} else {
|
||||
batch.queuedMetadataBytesBySession.delete(entry.sessionId)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
import { encodeStreamDataEvent } from './daemon-stream-data-split'
|
||||
|
||||
export class DaemonStreamHeldRefill {
|
||||
private readonly armed = new Map<string, symbol>()
|
||||
|
||||
constructor(private readonly flush: (clientId: string) => void) {}
|
||||
|
||||
arm(
|
||||
clientId: string,
|
||||
sessionId: string,
|
||||
write: (line: string, complete: () => void) => void
|
||||
): void {
|
||||
if (this.armed.has(clientId)) {
|
||||
return
|
||||
}
|
||||
const refill = Symbol()
|
||||
this.armed.set(clientId, refill)
|
||||
// A real no-op frame waits for preceding writes; an empty write can complete immediately.
|
||||
write(encodeStreamDataEvent(sessionId, ''), () => {
|
||||
if (this.armed.get(clientId) !== refill) {
|
||||
return
|
||||
}
|
||||
this.armed.delete(clientId)
|
||||
this.flush(clientId)
|
||||
})
|
||||
}
|
||||
|
||||
clear(clientId?: string): void {
|
||||
if (clientId === undefined) {
|
||||
this.armed.clear()
|
||||
} else {
|
||||
this.armed.delete(clientId)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -10,6 +10,10 @@
|
||||
import { clampToSafeSplitIndex } from './daemon-stream-data-split'
|
||||
import { recordDaemonStreamBacklogEvent } from './daemon-stream-backlog-probe'
|
||||
import type { DaemonEvent, DataGapEvent } from './types'
|
||||
import {
|
||||
accountDaemonStreamEntry,
|
||||
releaseDaemonStreamEntry
|
||||
} from './daemon-stream-entry-accounting'
|
||||
|
||||
// A control entry carries a whole pre-shaped stream event (background marker,
|
||||
// data gap, transient fact) that must ride at its exact position in the
|
||||
@@ -24,6 +28,7 @@ export type StreamQueueEntry = {
|
||||
seq?: number
|
||||
transformed?: boolean
|
||||
control?: DaemonEvent
|
||||
retainedBytes?: number
|
||||
}
|
||||
|
||||
export type PendingStreamDataBatch = {
|
||||
@@ -33,6 +38,7 @@ export type PendingStreamDataBatch = {
|
||||
// Per-session held totals so the flush hold can spare small talkers
|
||||
// (echo/replies) from waiting behind other sessions' floods.
|
||||
queuedCharsBySession: Map<string, number>
|
||||
queuedMetadataBytesBySession: Map<string, number>
|
||||
// Membership is reconciled when queued data first appears and on rare
|
||||
// background lifecycle changes, keeping steady-state enqueue constant-time.
|
||||
droppableQueuedSessionIds: Set<string>
|
||||
@@ -121,6 +127,7 @@ export function dropOldestQueuedForSession(
|
||||
if (insertGapAt === -1) {
|
||||
insertGapAt = i
|
||||
}
|
||||
releaseDaemonStreamEntry(batch, entry)
|
||||
batch.queue.splice(i, 1)
|
||||
i--
|
||||
} else {
|
||||
@@ -159,16 +166,20 @@ export function dropOldestQueuedForSession(
|
||||
sessionIdSuffix: sessionId.slice(-10),
|
||||
droppedChars: dropped
|
||||
})
|
||||
batch.queue.splice(Math.max(0, insertGapAt), 0, {
|
||||
sessionId,
|
||||
data: '',
|
||||
control: {
|
||||
type: 'event',
|
||||
event: 'dataGap',
|
||||
batch.queue.splice(
|
||||
Math.max(0, insertGapAt),
|
||||
0,
|
||||
accountDaemonStreamEntry(batch, {
|
||||
sessionId,
|
||||
payload: { droppedChars: dropped, sequenceChars: droppedSequenceChars }
|
||||
}
|
||||
})
|
||||
data: '',
|
||||
control: {
|
||||
type: 'event',
|
||||
event: 'dataGap',
|
||||
sessionId,
|
||||
payload: { droppedChars: dropped, sequenceChars: droppedSequenceChars }
|
||||
}
|
||||
})
|
||||
)
|
||||
insertGapAt = Math.max(0, insertGapAt) + 1
|
||||
}
|
||||
if (salvaged.length > 0) {
|
||||
@@ -177,7 +188,11 @@ export function dropOldestQueuedForSession(
|
||||
const at = existingGap
|
||||
? batch.queue.findIndex((e) => e.control === existingGap) + 1
|
||||
: insertGapAt
|
||||
batch.queue.splice(at, 0, { sessionId, data: salvaged, sequenceChars: 0 })
|
||||
batch.queue.splice(
|
||||
at,
|
||||
0,
|
||||
accountDaemonStreamEntry(batch, { sessionId, data: salvaged, sequenceChars: 0 })
|
||||
)
|
||||
batch.queuedChars += salvaged.length
|
||||
batch.queuedCharsBySession.set(
|
||||
sessionId,
|
||||
|
||||
@@ -0,0 +1,166 @@
|
||||
import type { Socket } from 'node:net'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { DaemonStreamDataBatcher } from './daemon-stream-data-batcher'
|
||||
import {
|
||||
SessionProducerPause,
|
||||
STREAM_BACKPRESSURE_STALL_WATCHDOG_MS
|
||||
} from './session-producer-pause'
|
||||
|
||||
const MiB = 1024 * 1024
|
||||
const CHUNK = 'x'.repeat(64 * 1024)
|
||||
|
||||
/** A stream socket whose peer accepts nothing until drain() is called — the half-open link that
|
||||
* neither drains nor closes. */
|
||||
function createStallableSocket() {
|
||||
const written: string[] = []
|
||||
const completions: (() => void)[] = []
|
||||
let buffered = 0
|
||||
const socket = {
|
||||
destroyed: false,
|
||||
get writableLength(): number {
|
||||
return buffered
|
||||
},
|
||||
write(line: string, complete?: () => void): boolean {
|
||||
written.push(line)
|
||||
buffered += Buffer.byteLength(line)
|
||||
if (complete) {
|
||||
completions.push(complete)
|
||||
}
|
||||
return false
|
||||
}
|
||||
}
|
||||
const drain = (): void => {
|
||||
buffered = 0
|
||||
for (const complete of completions.splice(0)) {
|
||||
complete()
|
||||
}
|
||||
}
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the batcher only uses write/writableLength/destroyed, all implemented above.
|
||||
return { socket: socket as unknown as Socket, written, drain }
|
||||
}
|
||||
|
||||
function droppedChars(written: readonly string[]): number {
|
||||
return written.reduce((total, line) => {
|
||||
const message: { event?: string; payload?: { droppedChars?: number } } = JSON.parse(line)
|
||||
return message.event === 'dataGap' ? total + (message.payload?.droppedChars ?? 0) : total
|
||||
}, 0)
|
||||
}
|
||||
|
||||
function createWiring() {
|
||||
const { socket, written, drain } = createStallableSocket()
|
||||
const subprocess = { pause: vi.fn(), resume: vi.fn() }
|
||||
const producer = new SessionProducerPause(subprocess)
|
||||
const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: socket }), {
|
||||
// Mirrors DaemonServer's wiring onto TerminalHost.
|
||||
onProducerBackpressureChanged: (_sessionId, paused, onStallTimeout) => {
|
||||
if (paused) {
|
||||
producer.pause('stream', true, onStallTimeout)
|
||||
} else {
|
||||
producer.resumeClient('stream')
|
||||
}
|
||||
}
|
||||
})
|
||||
const produceUntilPaused = (): void => {
|
||||
let produced = 0
|
||||
while (subprocess.pause.mock.calls.length === 0 && produced < 8 * MiB) {
|
||||
batcher.enqueue('client', 'flood', CHUNK)
|
||||
batcher.flush('client')
|
||||
produced += CHUNK.length
|
||||
}
|
||||
expect(subprocess.pause).toHaveBeenCalledOnce()
|
||||
}
|
||||
const drainFully = (advanceMsPerPass = 0): void => {
|
||||
for (let pass = 0; pass < 200 && batcher.queuedCharsForClient('client') > 0; pass++) {
|
||||
vi.advanceTimersByTime(advanceMsPerPass)
|
||||
drain()
|
||||
batcher.flush('client')
|
||||
}
|
||||
// Settle the last writes: the session stays accounted until the kernel takes its in-flight bytes.
|
||||
drain()
|
||||
expect(batcher.queuedCharsForClient('client')).toBe(0)
|
||||
}
|
||||
return {
|
||||
batcher,
|
||||
drain,
|
||||
drainFully,
|
||||
produceUntilPaused,
|
||||
producer,
|
||||
subprocess,
|
||||
written
|
||||
}
|
||||
}
|
||||
|
||||
describe('producer stall watchdog end to end', () => {
|
||||
afterEach(() => {
|
||||
vi.useRealTimers()
|
||||
})
|
||||
|
||||
it('turns a wedged consumer into a data gap instead of a frozen shell', () => {
|
||||
vi.useFakeTimers()
|
||||
const { batcher, drain, produceUntilPaused, subprocess, written } = createWiring()
|
||||
produceUntilPaused()
|
||||
expect(droppedChars(written)).toBe(0)
|
||||
|
||||
vi.advanceTimersByTime(STREAM_BACKPRESSURE_STALL_WATCHDOG_MS)
|
||||
expect(subprocess.resume).toHaveBeenCalledOnce()
|
||||
|
||||
// The shell runs again; its output is now keep-tail thinned rather than queued without bound.
|
||||
for (let chunk = 0; chunk < 16; chunk++) {
|
||||
batcher.enqueue('client', 'flood', CHUNK)
|
||||
batcher.flush('client')
|
||||
}
|
||||
expect(subprocess.pause).toHaveBeenCalledOnce()
|
||||
expect(2 * batcher.queuedCharsForClient('client')).toBeLessThan(4 * MiB)
|
||||
|
||||
// The gap rides the stream in byte order, so the client sees it as soon as the link recovers.
|
||||
drain()
|
||||
batcher.flush('client')
|
||||
expect(droppedChars(written)).toBeGreaterThan(0)
|
||||
})
|
||||
|
||||
it('restores ordinary pausing once the consumer catches up', () => {
|
||||
vi.useFakeTimers()
|
||||
const { batcher, drainFully, produceUntilPaused, subprocess, written } = createWiring()
|
||||
produceUntilPaused()
|
||||
vi.advanceTimersByTime(STREAM_BACKPRESSURE_STALL_WATCHDOG_MS)
|
||||
drainFully()
|
||||
|
||||
const gapBefore = droppedChars(written)
|
||||
let produced = 0
|
||||
while (subprocess.pause.mock.calls.length < 2 && produced < 8 * MiB) {
|
||||
batcher.enqueue('client', 'flood', CHUNK)
|
||||
batcher.flush('client')
|
||||
produced += CHUNK.length
|
||||
}
|
||||
expect(subprocess.pause).toHaveBeenCalledTimes(2)
|
||||
expect(droppedChars(written)).toBe(gapBefore)
|
||||
})
|
||||
|
||||
it('does not fire against a consumer that is draining', () => {
|
||||
vi.useFakeTimers()
|
||||
const { drainFully, produceUntilPaused, subprocess, written } = createWiring()
|
||||
produceUntilPaused()
|
||||
|
||||
// A real client drains multi-MB backlogs in well under a second; give it 10ms a pass.
|
||||
drainFully(10)
|
||||
expect(subprocess.resume).toHaveBeenCalledOnce()
|
||||
|
||||
vi.advanceTimersByTime(STREAM_BACKPRESSURE_STALL_WATCHDOG_MS * 2)
|
||||
expect(droppedChars(written)).toBe(0)
|
||||
expect(subprocess.pause).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('clears the watchdog when the stream socket closes', () => {
|
||||
vi.useFakeTimers()
|
||||
const { batcher, produceUntilPaused, subprocess, written } = createWiring()
|
||||
produceUntilPaused()
|
||||
|
||||
// What onStreamDisconnected does once keepalive probes fail and the peer's socket closes.
|
||||
batcher.clear('client')
|
||||
expect(subprocess.resume).toHaveBeenCalledOnce()
|
||||
|
||||
vi.advanceTimersByTime(STREAM_BACKPRESSURE_STALL_WATCHDOG_MS * 2)
|
||||
expect(droppedChars(written)).toBe(0)
|
||||
expect(subprocess.pause).toHaveBeenCalledOnce()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,102 @@
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import {
|
||||
PRODUCER_PAUSE_FAILSAFE_MS,
|
||||
SessionProducerPause,
|
||||
STREAM_BACKPRESSURE_STALL_WATCHDOG_MS
|
||||
} from './session-producer-pause'
|
||||
|
||||
function createProducer() {
|
||||
const subprocess = { pause: vi.fn(), resume: vi.fn() }
|
||||
const onStreamStall = vi.fn()
|
||||
return {
|
||||
subprocess,
|
||||
onStreamStall,
|
||||
producer: new SessionProducerPause(subprocess)
|
||||
}
|
||||
}
|
||||
|
||||
describe('SessionProducerPause stream stall watchdog', () => {
|
||||
afterEach(() => {
|
||||
vi.useRealTimers()
|
||||
})
|
||||
|
||||
it('resumes the producer and sheds the backlog when the consumer never drains', () => {
|
||||
vi.useFakeTimers()
|
||||
const { producer, subprocess, onStreamStall } = createProducer()
|
||||
producer.pause('stream', true, onStreamStall)
|
||||
expect(subprocess.pause).toHaveBeenCalledOnce()
|
||||
|
||||
vi.advanceTimersByTime(STREAM_BACKPRESSURE_STALL_WATCHDOG_MS - 1)
|
||||
expect(onStreamStall).not.toHaveBeenCalled()
|
||||
expect(subprocess.resume).not.toHaveBeenCalled()
|
||||
|
||||
vi.advanceTimersByTime(1)
|
||||
// Shed first: resuming into an unbounded queue would just rebuild the stall.
|
||||
expect(onStreamStall).toHaveBeenCalledOnce()
|
||||
expect(subprocess.resume).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('never fires against a session whose consumer is draining', () => {
|
||||
vi.useFakeTimers()
|
||||
const { producer, subprocess, onStreamStall } = createProducer()
|
||||
for (let cycle = 0; cycle < 5; cycle++) {
|
||||
producer.pause('stream', true, onStreamStall)
|
||||
vi.advanceTimersByTime(STREAM_BACKPRESSURE_STALL_WATCHDOG_MS / 2)
|
||||
producer.resumeClient('stream')
|
||||
vi.advanceTimersByTime(STREAM_BACKPRESSURE_STALL_WATCHDOG_MS)
|
||||
}
|
||||
expect(onStreamStall).not.toHaveBeenCalled()
|
||||
expect(subprocess.pause).toHaveBeenCalledTimes(5)
|
||||
expect(subprocess.resume).toHaveBeenCalledTimes(5)
|
||||
})
|
||||
|
||||
it('does not let a re-asserted pause defer the watchdog', () => {
|
||||
vi.useFakeTimers()
|
||||
const { producer, onStreamStall } = createProducer()
|
||||
producer.pause('stream', true, onStreamStall)
|
||||
// refresh() re-asserts a standing pause on every enqueue for any session sharing the client.
|
||||
for (let tick = 0; tick < 10; tick++) {
|
||||
vi.advanceTimersByTime(STREAM_BACKPRESSURE_STALL_WATCHDOG_MS / 10)
|
||||
producer.pause('stream', true, onStreamStall)
|
||||
}
|
||||
expect(onStreamStall).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it.each([
|
||||
['a stream resume', (producer: SessionProducerPause) => producer.resumeClient('stream')],
|
||||
['a release', (producer: SessionProducerPause) => producer.release({ resume: true })],
|
||||
[
|
||||
'a detach without resume',
|
||||
(producer: SessionProducerPause) => producer.release({ resume: false })
|
||||
]
|
||||
])('clears the watchdog on %s', (_label, unpause) => {
|
||||
vi.useFakeTimers()
|
||||
const { producer, onStreamStall } = createProducer()
|
||||
producer.pause('stream', true, onStreamStall)
|
||||
unpause(producer)
|
||||
vi.advanceTimersByTime(STREAM_BACKPRESSURE_STALL_WATCHDOG_MS * 2)
|
||||
expect(onStreamStall).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('leaves an unattached session unarmed — nothing is consuming it', () => {
|
||||
vi.useFakeTimers()
|
||||
const { producer, subprocess, onStreamStall } = createProducer()
|
||||
producer.pause('stream', false, onStreamStall)
|
||||
vi.advanceTimersByTime(STREAM_BACKPRESSURE_STALL_WATCHDOG_MS * 2)
|
||||
expect(onStreamStall).not.toHaveBeenCalled()
|
||||
expect(subprocess.pause).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('keeps an outstanding client pause in force after the stream watchdog fires', () => {
|
||||
vi.useFakeTimers()
|
||||
const { producer, subprocess, onStreamStall } = createProducer()
|
||||
producer.pause('stream', true, onStreamStall)
|
||||
producer.pause()
|
||||
vi.advanceTimersByTime(PRODUCER_PAUSE_FAILSAFE_MS)
|
||||
// The client failsafe cannot resume while the stream still holds the producer.
|
||||
expect(subprocess.resume).not.toHaveBeenCalled()
|
||||
vi.advanceTimersByTime(STREAM_BACKPRESSURE_STALL_WATCHDOG_MS)
|
||||
expect(onStreamStall).toHaveBeenCalledOnce()
|
||||
expect(subprocess.resume).toHaveBeenCalledOnce()
|
||||
})
|
||||
})
|
||||
@@ -4,38 +4,122 @@ import type { SubprocessHandle } from './session-subprocess-handle'
|
||||
// resume must never wedge a shell, so auto-resume after this window — a still-flooded main re-pauses.
|
||||
export const PRODUCER_PAUSE_FAILSAFE_MS = 5_000
|
||||
|
||||
// Why: a stream pause has no lost-resume problem (the batcher un-pauses from its own drain accounting),
|
||||
// but it does have a stalled-consumer problem — a half-open peer (slept laptop, dropped NAT state)
|
||||
// neither drains nor closes, so nothing ever calls back and the shell blocks on write() forever. Bound
|
||||
// the stall; the owner's callback then sheds the backlog so the session runs again with a visible gap.
|
||||
// 60s: far above any healthy drain (the shallow-socket gate turns over in milliseconds), so a slow but
|
||||
// live consumer is never mistaken for a wedged one.
|
||||
export const STREAM_BACKPRESSURE_STALL_WATCHDOG_MS = 60_000
|
||||
|
||||
/** Producer-side flow control for one session's PTY fd, with the lost-resume failsafe. */
|
||||
export class SessionProducerPause {
|
||||
private paused = false
|
||||
private streamBackpressured = false
|
||||
private failsafeTimer: ReturnType<typeof setTimeout> | null = null
|
||||
private streamStallTimer: ReturnType<typeof setTimeout> | null = null
|
||||
private onStreamStall: (() => void) | null = null
|
||||
|
||||
constructor(private readonly subprocess: Pick<SubprocessHandle, 'pause' | 'resume'>) {}
|
||||
|
||||
/** Stop reading the PTY fd so a flooding child blocks on write. Arms the failsafe; re-pausing re-arms it. */
|
||||
pause(): void {
|
||||
pause(source?: 'stream', canPauseStream = true, onStreamStall?: () => void): void {
|
||||
if (source === 'stream') {
|
||||
if (canPauseStream) {
|
||||
this.onStreamStall = onStreamStall ?? null
|
||||
this.setStreamBackpressured(true)
|
||||
}
|
||||
return
|
||||
}
|
||||
const wasPaused = this.paused || this.streamBackpressured
|
||||
this.paused = true
|
||||
this.subprocess.pause?.()
|
||||
if (!wasPaused) {
|
||||
this.subprocess.pause?.()
|
||||
}
|
||||
if (this.failsafeTimer) {
|
||||
clearTimeout(this.failsafeTimer)
|
||||
}
|
||||
this.failsafeTimer = setTimeout(() => {
|
||||
this.failsafeTimer = null
|
||||
this.paused = false
|
||||
this.subprocess.resume?.()
|
||||
if (!this.streamBackpressured) {
|
||||
this.subprocess.resume?.()
|
||||
}
|
||||
}, PRODUCER_PAUSE_FAILSAFE_MS)
|
||||
}
|
||||
|
||||
setStreamBackpressured(paused: boolean): void {
|
||||
const wasPaused = this.paused || this.streamBackpressured
|
||||
const wasStreamBackpressured = this.streamBackpressured
|
||||
this.streamBackpressured = paused
|
||||
if (paused) {
|
||||
// Only on the transition: refresh() re-asserts an already-standing pause on every enqueue for any
|
||||
// session sharing the client, and re-arming there would let a busy neighbour defer the watchdog forever.
|
||||
if (!wasStreamBackpressured) {
|
||||
this.armStreamStallWatchdog()
|
||||
}
|
||||
} else {
|
||||
this.clearStreamStallWatchdog()
|
||||
}
|
||||
const nowPaused = this.paused || this.streamBackpressured
|
||||
if (nowPaused && !wasPaused) {
|
||||
this.subprocess.pause?.()
|
||||
} else if (wasPaused && !nowPaused) {
|
||||
this.subprocess.resume?.()
|
||||
}
|
||||
}
|
||||
|
||||
resumeClient(source?: 'stream'): void {
|
||||
if (source === 'stream') {
|
||||
this.setStreamBackpressured(false)
|
||||
return
|
||||
}
|
||||
if (this.failsafeTimer) {
|
||||
clearTimeout(this.failsafeTimer)
|
||||
this.failsafeTimer = null
|
||||
}
|
||||
const wasPaused = this.paused
|
||||
this.paused = false
|
||||
if (wasPaused && !this.streamBackpressured) {
|
||||
this.subprocess.resume?.()
|
||||
}
|
||||
}
|
||||
|
||||
release(opts: { resume: boolean }): void {
|
||||
if (this.failsafeTimer) {
|
||||
clearTimeout(this.failsafeTimer)
|
||||
this.failsafeTimer = null
|
||||
}
|
||||
if (!this.paused) {
|
||||
this.clearStreamStallWatchdog()
|
||||
if (!this.paused && !this.streamBackpressured) {
|
||||
return
|
||||
}
|
||||
this.paused = false
|
||||
this.streamBackpressured = false
|
||||
if (opts.resume) {
|
||||
this.subprocess.resume?.()
|
||||
}
|
||||
}
|
||||
|
||||
private armStreamStallWatchdog(): void {
|
||||
const onStreamStall = this.onStreamStall
|
||||
if (!onStreamStall) {
|
||||
return
|
||||
}
|
||||
this.streamStallTimer = setTimeout(() => {
|
||||
this.streamStallTimer = null
|
||||
// Shed the backlog BEFORE resuming, so the producer never refills an unbounded queue. Loss of
|
||||
// contact with the consumer says nothing about the child process — nothing here reports an exit.
|
||||
onStreamStall()
|
||||
this.setStreamBackpressured(false)
|
||||
}, STREAM_BACKPRESSURE_STALL_WATCHDOG_MS)
|
||||
}
|
||||
|
||||
private clearStreamStallWatchdog(): void {
|
||||
if (this.streamStallTimer) {
|
||||
clearTimeout(this.streamStallTimer)
|
||||
this.streamStallTimer = null
|
||||
}
|
||||
this.onStreamStall = null
|
||||
}
|
||||
}
|
||||
|
||||
@@ -175,6 +175,48 @@ describe('Session terminal control', () => {
|
||||
})
|
||||
|
||||
describe('producer flow control', () => {
|
||||
it('client resume and its failsafe cannot release daemon stream backpressure', () => {
|
||||
createSession()
|
||||
session.attachClient({ onData: () => {}, onExit: () => {} })
|
||||
session.pauseProducer('stream')
|
||||
session.pauseProducer()
|
||||
session.resumeProducer()
|
||||
expect(subprocess.resumeCalls).toBe(0)
|
||||
session.pauseProducer()
|
||||
vi.advanceTimersByTime(PRODUCER_PAUSE_FAILSAFE_MS * 2)
|
||||
expect(subprocess.resumeCalls).toBe(0)
|
||||
session.resumeProducer('stream')
|
||||
expect(subprocess.resumeCalls).toBe(1)
|
||||
})
|
||||
|
||||
it('draining the stream cannot release an outstanding client pause', () => {
|
||||
createSession()
|
||||
session.attachClient({ onData: () => {}, onExit: () => {} })
|
||||
session.pauseProducer()
|
||||
session.pauseProducer('stream')
|
||||
session.resumeProducer('stream')
|
||||
expect(subprocess.resumeCalls).toBe(0)
|
||||
session.resumeProducer()
|
||||
expect(subprocess.resumeCalls).toBe(1)
|
||||
})
|
||||
|
||||
it('detach and termination release daemon stream backpressure', () => {
|
||||
createSession()
|
||||
const client = { onData: () => {}, onExit: () => {} }
|
||||
const token = session.attachClient(client)
|
||||
session.pauseProducer('stream')
|
||||
session.detachClient(token)
|
||||
expect(subprocess.resumeCalls).toBe(1)
|
||||
session.pauseProducer('stream')
|
||||
expect(subprocess.pauseCalls).toBe(1)
|
||||
session.attachClient(client)
|
||||
session.pauseProducer('stream')
|
||||
session.kill()
|
||||
expect(subprocess.resumeCalls).toBe(2)
|
||||
session.pauseProducer('stream')
|
||||
expect(subprocess.pauseCalls).toBe(2)
|
||||
})
|
||||
|
||||
it('auto-resumes when the owner loses the resume signal', () => {
|
||||
createSession()
|
||||
session.pauseProducer()
|
||||
|
||||
@@ -168,16 +168,17 @@ export class Session {
|
||||
}
|
||||
|
||||
/** Producer-side flow control: stop reading the PTY fd so a flooding child blocks on write.
|
||||
* Arms the lost-resume failsafe; re-pausing re-arms it. */
|
||||
pauseProducer(): void {
|
||||
* Arms the lost-resume failsafe; re-pausing re-arms it. onStreamStall bounds a stream pause whose
|
||||
* consumer never drains and never closes. */
|
||||
pauseProducer(source?: 'stream', onStreamStall?: () => void): void {
|
||||
if (this._state === 'exited' || this._disposed) {
|
||||
return
|
||||
}
|
||||
this.producerPause.pause()
|
||||
this.producerPause.pause(source, this.hasAttachedClients && !this.isTerminating, onStreamStall)
|
||||
}
|
||||
|
||||
resumeProducer(): void {
|
||||
this.producerPause.release({ resume: true })
|
||||
resumeProducer(source?: 'stream'): void {
|
||||
this.producerPause.resumeClient(source)
|
||||
}
|
||||
|
||||
kill(): void {
|
||||
|
||||
@@ -168,16 +168,16 @@ export class TerminalHost {
|
||||
}
|
||||
|
||||
// Why null-not-throw (unlike write/resize): pause/resume are best-effort hints against a session that may have exited.
|
||||
pauseProducer(sessionId: string): void {
|
||||
pauseProducer(sessionId: string, source?: 'stream', onStreamStall?: () => void): void {
|
||||
const session = this.sessions.get(sessionId)
|
||||
if (!session || !session.isAlive) {
|
||||
return
|
||||
}
|
||||
session.pauseProducer()
|
||||
session.pauseProducer(source, onStreamStall)
|
||||
}
|
||||
|
||||
resumeProducer(sessionId: string): void {
|
||||
this.sessions.get(sessionId)?.resumeProducer()
|
||||
resumeProducer(sessionId: string, source?: 'stream'): void {
|
||||
this.sessions.get(sessionId)?.resumeProducer(source)
|
||||
}
|
||||
|
||||
kill(sessionId: string, opts: { immediate?: boolean } = {}): Promise<void> {
|
||||
|
||||
Reference in New Issue
Block a user