diff --git a/src/relay/agent-hook-envelope-publication.test.ts b/src/relay/agent-hook-envelope-publication.test.ts index 4794b76cbe0..83b6594ad26 100644 --- a/src/relay/agent-hook-envelope-publication.test.ts +++ b/src/relay/agent-hook-envelope-publication.test.ts @@ -131,40 +131,65 @@ describe('publishAgentHookEnvelope', () => { } }) - it('sheds interactivePrompt next when dropping the assistant message is not enough', () => { + it('sheds subagents next when dropping the assistant message is not enough', () => { const primary = makeBoundedClient(16384) const dispatcher = new RelayDispatcher(primary.write, primary.options) try { publishAgentHookEnvelope( dispatcher, - makeEnvelope({ lastAssistantMessage: 8_000, interactivePrompt: 13_000, subagents: 2 }) + makeEnvelope({ lastAssistantMessage: 8_000, interactivePrompt: 6_000, subagents: 40 }) ) const published = decodeEnvelopes(primary) expect(published).toHaveLength(1) expect(published[0].payload.lastAssistantMessage).toBeUndefined() - expect(published[0].payload.interactivePrompt).toBeUndefined() - expect(published[0].payload.subagents).toHaveLength(2) + expect(published[0].payload.subagents).toBeUndefined() + expect(published[0].payload.interactivePrompt).toBe('q'.repeat(6_000)) expect(primary.closes).toBe(0) } finally { dispatcher.dispose() } }) - it('sheds subagents last and still delivers the status', () => { + it('keeps the blocking question card when a waiting envelope has to shed', () => { + const primary = makeBoundedClient(16384) + const dispatcher = new RelayDispatcher(primary.write, primary.options) + try { + // A blocked pane is only answerable from web/mobile if interactivePrompt survives the ladder. + const envelope = makeEnvelope({ + lastAssistantMessage: 8_000, + interactivePrompt: 4_000, + subagents: 40 + }) + envelope.payload.state = 'waiting' + publishAgentHookEnvelope(dispatcher, envelope) + + const published = decodeEnvelopes(primary) + expect(published).toHaveLength(1) + expect(published[0].payload.state).toBe('waiting') + expect(published[0].payload.interactivePrompt).toBe('q'.repeat(4_000)) + expect(published[0].payload.subagents).toBeUndefined() + expect(published[0].payload.lastAssistantMessage).toBeUndefined() + expect(primary.closes).toBe(0) + } finally { + dispatcher.dispose() + } + }) + + it('sheds interactivePrompt last and still delivers the status', () => { const primary = makeBoundedClient(16384) const dispatcher = new RelayDispatcher(primary.write, primary.options) try { publishAgentHookEnvelope( dispatcher, - makeEnvelope({ lastAssistantMessage: 6_000, interactivePrompt: 6_000, subagents: 40 }) + makeEnvelope({ lastAssistantMessage: 6_000, interactivePrompt: 13_000, subagents: 2 }) ) const published = decodeEnvelopes(primary) expect(published).toHaveLength(1) expect(published[0].payload.lastAssistantMessage).toBeUndefined() - expect(published[0].payload.interactivePrompt).toBeUndefined() expect(published[0].payload.subagents).toBeUndefined() + expect(published[0].payload.interactivePrompt).toBeUndefined() expect(published[0].payload.state).toBe('working') expect(primary.closes).toBe(0) } finally { @@ -172,6 +197,24 @@ describe('publishAgentHookEnvelope', () => { } }) + it('measures the frame budget once up front and once per real shed', () => { + const primary = makeBoundedClient(16384) + const dispatcher = new RelayDispatcher(primary.write, primary.options) + const probe = vi.spyOn(dispatcher, 'producerEnvelopeBudget') + try { + publishAgentHookEnvelope(dispatcher, makeEnvelope({ lastAssistantMessage: 128 })) + expect(probe).toHaveBeenCalledTimes(1) + + probe.mockClear() + // Two of the three ladder steps are absent here; skipping them must not cost an extra encode. + publishAgentHookEnvelope(dispatcher, makeEnvelope({ interactivePrompt: 20_000 })) + expect(probe).toHaveBeenCalledTimes(2) + } finally { + probe.mockRestore() + dispatcher.dispose() + } + }) + it('skips absent fields instead of consuming a shed step on them', () => { const primary = makeBoundedClient(16384) const dispatcher = new RelayDispatcher(primary.write, primary.options) diff --git a/src/relay/agent-hook-envelope-publication.ts b/src/relay/agent-hook-envelope-publication.ts index 0ed68f6e5d3..c5ffbb6fea5 100644 --- a/src/relay/agent-hook-envelope-publication.ts +++ b/src/relay/agent-hook-envelope-publication.ts @@ -6,7 +6,9 @@ import type { RelayDispatcher } from './dispatcher' // Why: shed the biggest, most reconstructible fields first — state/paneKey must survive or the pane // stays stuck on its last spinner, and an over-capacity frame is now dropped rather than sent. -const SHED_ORDER = ['lastAssistantMessage', 'interactivePrompt', 'subagents'] as const +// Why: interactivePrompt is last on purpose — a blocking question is load-bearing (without its card +// the pane sits on `waiting` with no way to answer from web or mobile) while the roster is cosmetic. +const SHED_ORDER = ['lastAssistantMessage', 'subagents', 'interactivePrompt'] as const function fitsProducerFrame(dispatcher: RelayDispatcher, envelope: AgentHookRelayEnvelope): boolean { return ( @@ -24,8 +26,10 @@ export function publishAgentHookEnvelope( envelope: AgentHookRelayEnvelope ): void { let candidate = envelope + // Why: re-encoding is the expensive part, so only re-measure after a field was actually dropped. + let fits = fitsProducerFrame(dispatcher, candidate) for (const field of SHED_ORDER) { - if (fitsProducerFrame(dispatcher, candidate)) { + if (fits) { break } if (candidate.payload[field] === undefined) { @@ -35,6 +39,7 @@ export function publishAgentHookEnvelope( // would permanently strip the cached copy too. candidate = { ...candidate, payload: { ...candidate.payload } } delete candidate.payload[field] + fits = fitsProducerFrame(dispatcher, candidate) } dispatcher.notify(AGENT_HOOK_NOTIFICATION_METHOD, candidate as unknown as Record) } diff --git a/src/relay/dispatcher-capacity-degradation.test.ts b/src/relay/dispatcher-capacity-degradation.test.ts index 598044f2e96..c41dcd4d860 100644 --- a/src/relay/dispatcher-capacity-degradation.test.ts +++ b/src/relay/dispatcher-capacity-degradation.test.ts @@ -9,6 +9,17 @@ type BoundedClient = { write: (data: Buffer) => boolean } +// Why: a sink that never accepts a write and never drains retains producer bytes, so the queue saturates +// while every individual frame stays comfortably inside the per-frame capacity. +function makeSaturatingClient(highWaterMark: number): BoundedClient { + const client = makeBoundedClient(highWaterMark) + client.write = (data: Buffer) => { + client.frames.push(Buffer.from(data)) + return false + } + return client +} + function makeBoundedClient(highWaterMark: number): BoundedClient { const client: BoundedClient = { frames: [], @@ -121,7 +132,7 @@ describe('RelayDispatcher bounded-capacity degradation', () => { expect(primary.frames).toHaveLength(0) expect(stderr).toHaveBeenCalledTimes(1) expect(stderr.mock.calls[0][0]).toMatch( - /^\[relay\] Dropped fs\.changed \(\d+B > producer capacity 12288B\)\n$/ + /^\[relay\] Dropped fs\.changed \(\d+B > producer frame capacity 12288B\)\n$/ ) bounded.notify('pty.exit', { id: 'pty-1' }) @@ -153,6 +164,56 @@ describe('RelayDispatcher bounded-capacity degradation', () => { } }) + it('distinguishes an over-capacity drop from a producer-queue-full drop on one client', () => { + const primary = makeSaturatingClient(4 * 1024 * 1024) + const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true) + const bounded = new RelayDispatcher(primary.write, primary.options) + try { + // The first 1.5MB frame is retained by the stalled sink; the second exceeds the 2MB producer queue. + bounded.notify('fs.changed', { events: ['x'.repeat(1_500_000)] }) + expect(primary.frames).toHaveLength(1) + bounded.notify('fs.changed', { events: ['x'.repeat(1_500_000)] }) + bounded.notify('fs.changed', { events: ['x'.repeat(5_000_000)] }) + + expect(primary.closes).toBe(0) + expect(stderr).toHaveBeenCalledTimes(2) + expect(stderr.mock.calls[0][0]).toMatch( + /^\[relay\] Dropped fs\.changed \(\d+B, producer queue full; frame capacity 4128768B\)\n$/ + ) + expect(stderr.mock.calls[1][0]).toMatch( + /^\[relay\] Dropped fs\.changed \(\d+B > producer frame capacity 4128768B\)\n$/ + ) + + // Both classifications stay one-per-generation once logged. + bounded.notify('fs.changed', { events: ['x'.repeat(1_500_000)] }) + bounded.notify('fs.changed', { events: ['x'.repeat(5_000_000)] }) + expect(stderr).toHaveBeenCalledTimes(2) + } finally { + stderr.mockRestore() + bounded.dispose() + } + }) + + it('logs each dropped method once instead of letting the first method silence the rest', () => { + const primary = makeBoundedClient(16384) + const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true) + const bounded = new RelayDispatcher(primary.write, primary.options) + try { + const flood = { events: ['x'.repeat(20_000)] } + bounded.notify('fs.changed', flood) + bounded.notify('pty.data', flood) + bounded.notify('fs.changed', flood) + bounded.notify('pty.data', flood) + + expect(stderr).toHaveBeenCalledTimes(2) + expect(stderr.mock.calls[0][0]).toContain('Dropped fs.changed') + expect(stderr.mock.calls[1][0]).toContain('Dropped pty.data') + } finally { + stderr.mockRestore() + bounded.dispose() + } + }) + it('publishProducerNotification publishes on the producer lane and never closes', () => { const primary = makeBoundedClient(65536) const secondary = makeBoundedClient(16384) diff --git a/src/relay/dispatcher.ts b/src/relay/dispatcher.ts index c79ad84b630..77a4e6867a7 100644 --- a/src/relay/dispatcher.ts +++ b/src/relay/dispatcher.ts @@ -69,10 +69,19 @@ type RelayClient = { highestReceivedSeq: number generation: number closed: boolean - droppedNotificationLogGeneration: number | null + droppedNotificationLog: DroppedProducerNotificationLog | null sessionIdentity: RelayClientSessionIdentity } +// Why: the log key set is rebuilt per generation, but a producer minting synthetic method names would still +// grow it inside one generation — cap it well above the fixed relay method vocabulary. +const DROPPED_NOTIFICATION_LOG_KEY_LIMIT = 64 + +type DroppedProducerNotificationLog = { + generation: number + loggedKeys: Set +} + type PendingRelayRequest = { resolve: (result: unknown) => void reject: (error: Error) => void @@ -467,18 +476,30 @@ export class RelayDispatcher { ) } - // Why: one line per connection generation — a flooding producer retries every batch and would spam stderr. + // Why: one line per generation, method and drop reason — a flooding producer retries every batch and would + // spam stderr, but a transient queue-full drop must not consume the slot a real over-capacity drop needs. private logDroppedProducerNotification( client: RelayClient, method: string, msg: JsonRpcNotification ): void { - if (client.droppedNotificationLogGeneration === client.generation) { + const bytes = this.estimateFrameBytes(msg) + const capacity = client.writer.producerFrameCapacity + const overCapacity = bytes > capacity + const key = `${method}:${overCapacity ? 'over-capacity' : 'queue-full'}` + let log = client.droppedNotificationLog + if (!log || log.generation !== client.generation) { + log = { generation: client.generation, loggedKeys: new Set() } + client.droppedNotificationLog = log + } + if (log.loggedKeys.has(key) || log.loggedKeys.size >= DROPPED_NOTIFICATION_LOG_KEY_LIMIT) { return } - client.droppedNotificationLogGeneration = client.generation + log.loggedKeys.add(key) process.stderr.write( - `[relay] Dropped ${method} (${this.estimateFrameBytes(msg)}B > producer capacity ${client.writer.producerFrameCapacity}B)\n` + overCapacity + ? `[relay] Dropped ${method} (${bytes}B > producer frame capacity ${capacity}B)\n` + : `[relay] Dropped ${method} (${bytes}B, producer queue full; frame capacity ${capacity}B)\n` ) } @@ -674,7 +695,7 @@ export class RelayDispatcher { highestReceivedSeq: 0, generation: 0, closed: false, - droppedNotificationLogGeneration: null, + droppedNotificationLog: null, sessionIdentity: sessionIdentity ?? { principal: `unproved:${id}`, authenticated: false, diff --git a/src/relay/fs-handler-stream.test.ts b/src/relay/fs-handler-stream.test.ts index e12f2acd8ae..4e1a3330c3d 100644 --- a/src/relay/fs-handler-stream.test.ts +++ b/src/relay/fs-handler-stream.test.ts @@ -194,9 +194,12 @@ describe('FsHandler readFileStream', () => { expect(frames[lastChunkIndex].clientId).toBe(7) }) - it('delivers fs.streamEnd on the control lane and releases the stream when the producer lane is saturated', async () => { - const filePath = path.join(tmpDir, 'saturated.png') + // Real saturation is not constructible: the fixed-bulk lane admits a chunk only while producerBytes is 0, + // so a real backlog parks the chunk pump long before the terminal frame — hence the lane-tagging mock. + it('orders fs.streamEnd after the final chunk on the control lane and releases the stream', async () => { + const filePath = path.join(tmpDir, 'stream-end-release.png') writeFileSync(filePath, Buffer.alloc(300 * 1024, 0x42)) + // Producer-lane frames now divert to the drop log, so an empty log proves no stream frame took that lane. dispatcher.saturateProducerLane() const meta = (await dispatcher.callRequest( @@ -209,6 +212,13 @@ describe('FsHandler readFileStream', () => { expect(dispatcher._droppedProducerFrames).toHaveLength(0) expect(collectStream(dispatcher).end).toEqual({ streamId: meta.streamId }) + const frames = dispatcher._notifications + const endIndex = frames.findIndex((n) => n.method === 'fs.streamEnd') + const lastChunkIndex = frames.map((n) => n.method).lastIndexOf('fs.streamChunk') + expect(lastChunkIndex).toBeGreaterThanOrEqual(0) + expect(endIndex).toBe(lastChunkIndex + 1) + expect(frames[endIndex].lane).toBe('control') + const registry = (handler as unknown as { streamRegistry: { size(): number } }).streamRegistry await waitFor(() => registry.size() === 0) }) diff --git a/src/relay/relay-filesystem-watch-registry.test.ts b/src/relay/relay-filesystem-watch-registry.test.ts index 8cbfe4018a7..0c6109a55ea 100644 --- a/src/relay/relay-filesystem-watch-registry.test.ts +++ b/src/relay/relay-filesystem-watch-registry.test.ts @@ -72,6 +72,19 @@ function createDispatcher() { fsChanged, notify: vi.fn(), notifyClient: vi.fn(record), + tryNotifyClient: vi.fn( + ( + clientId: number, + method: string, + params?: Record, + onSettled?: (result: { ok: true } | { ok: false; error: Error }) => void + ) => { + record(clientId, method, params) + // A healthy sink settles as the frame is written, which releases the emitter's outstanding-marker slot. + onSettled?.({ ok: true }) + return true + } + ), activeClientIds: vi.fn(() => [1]), producerEnvelopeBudget: vi.fn(() => Number.MAX_SAFE_INTEGER), publishProducerNotification: vi.fn( diff --git a/src/relay/relay-oversized-notification-survival.test.ts b/src/relay/relay-oversized-notification-survival.test.ts index 3ded2894c8b..7b1189bea41 100644 --- a/src/relay/relay-oversized-notification-survival.test.ts +++ b/src/relay/relay-oversized-notification-survival.test.ts @@ -58,17 +58,6 @@ function buildMaximalHookEnvelope(): Record { } as unknown as Record } -/** Pure measurement shape only: the real emitter never emits `watchId`. */ -function buildWatcherBatch(count: number): Record { - return { - watchId: 'watch-1', - events: Array.from({ length: count }, (_, i) => ({ - type: 'update', - path: `${WATCHER_ROOT}/node_modules/.vite/deps/chunk-${String(i).padStart(6, '0')}.js` - })) - } -} - function watcherProcessEvents(count: number): WatcherProcessEvent[] { return Array.from({ length: count }, (_, i) => ({ type: 'update', @@ -76,6 +65,16 @@ function watcherProcessEvents(count: number): WatcherProcessEvent[] { })) as unknown as WatcherProcessEvent[] } +/** The exact `fs.changed` params emitRelayWatcherEvents publishes, so the pinned sizes measure reality. */ +function buildWatcherBatch(count: number): Record { + return { + events: watcherProcessEvents(count).map((event) => ({ + kind: event.type, + absolutePath: event.path + })) + } +} + function frameBytes(method: string, params: Record): number { return encodeJsonRpcFrame({ jsonrpc: '2.0', method, params }, 1, 0).length } @@ -102,6 +101,20 @@ function createDispatcher(hwm: number, onClose: () => void, sink: Buffer[]): Rel } describe('relay oversized notification survival', () => { + // Without this the pinned byte counts below could measure a payload the emitter never sends. + it('measures the params emitRelayWatcherEvents actually publishes', () => { + const sink: Buffer[] = [] + const dispatcher = createDispatcher(NODE22_HWM, () => {}, sink) + + try { + emitRelayWatcherEvents(dispatcher, WATCHER_ROOT, false, watcherProcessEvents(1)) + expect(sink).toHaveLength(1) + expect(frameParams(sink[0])).toEqual(buildWatcherBatch(1)) + } finally { + dispatcher.dispose() + } + }) + it('measures both producers against both Node tiers', () => { const hookBytes = frameBytes(AGENT_HOOK_NOTIFICATION_METHOD, buildMaximalHookEnvelope()) const fullBatchBytes = frameBytes('fs.changed', buildWatcherBatch(MAX_BATCHED_WATCHER_EVENTS)) @@ -124,9 +137,9 @@ describe('relay oversized notification survival', () => { expect(capacityFor(NODE22_HWM)).toBe(49152) expect(capacityFor(NODE21_HWM)).toBe(12288) expect(hookBytes).toBe(29894) - expect(fullBatchBytes).toBe(425094) - expect(trip(capacityFor(NODE22_HWM))).toBe(577) - expect(trip(capacityFor(NODE21_HWM))).toBe(143) + expect(fullBatchBytes).toBe(465074) + expect(trip(capacityFor(NODE22_HWM))).toBe(527) + expect(trip(capacityFor(NODE21_HWM))).toBe(131) // agent.hook: fits a modern remote, over-cap on an older one. expect(hookBytes).toBeLessThan(capacityFor(NODE22_HWM)) diff --git a/src/relay/relay-watcher-event-emitter.test.ts b/src/relay/relay-watcher-event-emitter.test.ts index 340b50d7fdc..3bf42296a52 100644 --- a/src/relay/relay-watcher-event-emitter.test.ts +++ b/src/relay/relay-watcher-event-emitter.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it } from 'vitest' +import type { WatcherProcessEvent } from '../main/ipc/parcel-watcher-process-protocol' import { RelayDispatcher } from './dispatcher' import type { RelayClientSinkOptions, RelayClientWrite } from './dispatcher-writer-sink' import { HEADER_LENGTH, parseJsonRpcMessage } from './protocol' @@ -78,6 +79,25 @@ function watcherBatch(count: number): { type: 'create'; path: string; isDirector })) } +function parentDir(absolutePath: string): string { + return absolutePath.slice(0, absolutePath.lastIndexOf('/')) +} + +// Fills the 2 MB producer queue so every subsequent watcher batch is rejected on the producer lane. +function saturateProducerQueue(dispatcher: RelayDispatcher): void { + let admitted = 0 + for (const chunk of [40_000, 1_000, 1]) { + while ( + admitted < 5_000 && + dispatcher.tryNotifyPtyData({ paneId: 'pane', data: 'x'.repeat(chunk) }) + ) { + admitted += 1 + } + } + expect(admitted).toBeGreaterThan(0) + expect(admitted).toBeLessThan(5_000) +} + describe('relay watcher writer admission', () => { it('keeps watcher batches on the producer lane while overflow markers ride the control lane', () => { const sink = createRecordingSink() @@ -158,17 +178,7 @@ describe('relay watcher batch chunking', () => { try { // Fill the 2 MB producer queue down to a residue smaller than a one-event fs.changed frame. - let admitted = 0 - for (const chunk of [40_000, 1_000, 1]) { - while ( - admitted < 5_000 && - dispatcher.tryNotifyPtyData({ paneId: 'pane', data: 'x'.repeat(chunk) }) - ) { - admitted += 1 - } - } - expect(admitted).toBeGreaterThan(0) - expect(admitted).toBeLessThan(5_000) + saturateProducerQueue(dispatcher) emitRelayWatcherEvents(dispatcher, '/workspace', false, [ { type: 'create', path: '/workspace/first', isDirectory: false } @@ -186,6 +196,111 @@ describe('relay watcher batch chunking', () => { } }) + it('keeps at most one outstanding overflow marker per root without closing the client', () => { + const sink = createRecordingSink(65536) + sink.blockNextWrite() + const dispatcher = new RelayDispatcher(sink.write, sink.options) + + try { + saturateProducerQueue(dispatcher) + + // More rejected batches than the 256-frame control queue can hold: one marker each would close the link. + for (let index = 0; index < 300; index += 1) { + emitRelayWatcherEvents(dispatcher, '/workspace', false, [ + { type: 'create', path: `/workspace/file-${index}`, isDirectory: false } + ]) + } + // A second root must still get its own marker, or that tree silently desyncs. + emitRelayWatcherEvents(dispatcher, '/other', false, [ + { type: 'create', path: '/other/file', isDirectory: false } + ]) + expect(sink.closes()).toBe(0) + + sink.drainWaiters.shift()?.() + const markers = sink.frames + .filter((frame) => frameMethod(frame) === 'fs.changed') + .flatMap(frameEvents) + expect(markers).toEqual([ + { kind: 'overflow', absolutePath: '/workspace' }, + { kind: 'overflow', absolutePath: '/other' } + ]) + expect(sink.closes()).toBe(0) + + // The slot clears once the marker drains, so a later batch can still resync the same root. + emitRelayWatcherEvents(dispatcher, '/workspace', false, [ + { type: 'create', path: '/workspace/later', isDirectory: false } + ]) + expect( + sink.frames.filter((frame) => frameMethod(frame) === 'fs.changed').flatMap(frameEvents) + ).toHaveLength(3) + } finally { + dispatcher.dispose() + } + }) + + it('keeps one directory in one chunk when the batch interleaves directories', () => { + const sink = createRecordingSink(16384) + const dispatcher = new RelayDispatcher(sink.write, sink.options) + const directoryCount = 8 + const replacedPath = '/workspace/project/src/module-7/component-0.tsx' + const events: WatcherProcessEvent[] = [ + ...Array.from({ length: 200 }, (_unused, index) => ({ + type: 'create' as const, + path: `/workspace/project/src/module-${index % directoryCount}/component-${Math.floor(index / directoryCount)}.tsx`, + isDirectory: false + })), + { type: 'delete' as const, path: replacedPath } + ] + + try { + emitRelayWatcherEvents(dispatcher, '/workspace', false, events) + + expect(sink.frames.length).toBeGreaterThan(1) + const frameDirs = sink.frames.map( + (frame) => new Set(frameEvents(frame).map((event) => parentDir(event.absolutePath))) + ) + const dirs = Array.from(new Set(events.map((event) => parentDir(event.path)))) + expect(dirs).toHaveLength(directoryCount) + // Each directory in one frame; the renderer dedupes refreshes per payload, so extra frames = extra readDir RPCs. + const refreshes = frameDirs.reduce((total, frameDir) => total + frameDir.size, 0) + expect(refreshes).toBeLessThanOrEqual(directoryCount + sink.frames.length - 1) + + const delivered = sink.frames.flatMap(frameEvents) + for (const dir of dirs) { + const positions = delivered + .map((event, index) => ({ event, index })) + .filter((entry) => parentDir(entry.event.absolutePath) === dir) + .map((entry) => entry.index) + expect(positions.at(-1)! - positions[0]).toBe(positions.length - 1) + } + // Stable grouping keeps per-path order: the create still precedes the delete of the same path. + expect( + delivered.filter((event) => event.absolutePath === replacedPath).map((event) => event.kind) + ).toEqual(['create', 'delete']) + expect(delivered).toHaveLength(events.length) + } finally { + dispatcher.dispose() + } + }) + + it('sizes chunks near the byte minimum instead of overshooting it', () => { + const sink = createRecordingSink(16384) + const capacity = 12288 + const dispatcher = new RelayDispatcher(sink.write, sink.options) + + try { + emitRelayWatcherEvents(dispatcher, '/workspace', false, watcherBatch(5000)) + + const totalBytes = sink.frames.reduce((total, frame) => total + frame.length, 0) + expect(sink.frames.length).toBeLessThanOrEqual(Math.ceil(totalBytes / capacity) + 1) + for (const frame of sink.frames.slice(0, -1)) { + expect(frame.length).toBeGreaterThanOrEqual(capacity * 0.9) + } + } finally { + dispatcher.dispose() + } + }) + it('chunks per client so a small sink never degrades a healthy one', () => { const healthy = createRecordingSink(65536) const small = createRecordingSink(16384) diff --git a/src/relay/relay-watcher-event-emitter.ts b/src/relay/relay-watcher-event-emitter.ts index 7469821c5e9..b23705975cd 100644 Binary files a/src/relay/relay-watcher-event-emitter.ts and b/src/relay/relay-watcher-event-emitter.ts differ