From d5cd61b09c41b52780ddac538cf6835d892d0dbf Mon Sep 17 00:00:00 2001 From: Jinjing <6427696+AmethystLiang@users.noreply.github.com> Date: Fri, 31 Jul 2026 23:06:58 -0700 Subject: [PATCH] fix(relay): stop the overflow marker from re-killing the link it protects Round-1 review fixes on the P0 freeze work. The control-lane overflow marker could reinstate the exact failure this P0 removes: dispatcher-client-writer closes the client when control-lane admission fails, and admitControl is the only lane that returns an error, so one marker per failing batch accumulated to the 256-frame/1MB bound and dropped the link. Markers are now deduped to one outstanding per (client, root), cleared on settle. Chunking also defeated the renderer's per-payload directory dedupe -- events are now stable-grouped by parent directory so one directory lands in one chunk -- and the halving walk overshot the byte minimum ~1.7x while the fast path paid three JSON encodes; both are fixed by publishing first and sizing from a measured bytes-per-event estimate. Agent-hook shedding now surrenders the blocking interactive prompt LAST rather than first, so a degraded envelope cannot strand a pane at state=waiting with no answerable question card. The dropped-notification log now distinguishes over-capacity from producer queue backpressure and no longer lets the first dropped method silence every other producer for the life of the connection. --- .../agent-hook-envelope-publication.test.ts | 57 +++++++- src/relay/agent-hook-envelope-publication.ts | 9 +- .../dispatcher-capacity-degradation.test.ts | 63 +++++++- src/relay/dispatcher.ts | 33 ++++- src/relay/fs-handler-stream.test.ts | 14 +- .../relay-filesystem-watch-registry.test.ts | 13 ++ ...ay-oversized-notification-survival.test.ts | 41 ++++-- src/relay/relay-watcher-event-emitter.test.ts | 137 ++++++++++++++++-- src/relay/relay-watcher-event-emitter.ts | Bin 2926 -> 5711 bytes 9 files changed, 324 insertions(+), 43 deletions(-) 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 7469821c5e988747f3695bf2a9b5177f868e3c74..b23705975cda71f3477ab1d673b014463d0bf304 100644 GIT binary patch literal 5711 zcmb_g;f~uz4!+-h3Kl53c52JXuLIn8gG;(C0yJ%qJs>|80Xmk(mTFn@&5YLT+xTz~ zaZh+p(vRfGlC$3A4sC#VZHdF-kRLyiQ*5@i3l#QSOZW8NgfzGAt*cY({nf3l0v%}S z>J42a$#i3iYFcd5>DIW^mXn=|Pqr$5F(_2qJvr;^^47jRgk|kElP#o3%@8E~;!s zKDeS<;ip-^wkC*Gy~lheuZdukldiB#U*H_mfpOyt2`W@93-Z#MqQ)MtID&~E* zIO0&n#^mZcjA(7`){}1_T~yd?Vbb*$?)8nc8kx|ziq#l+J>`APqO=@=eU(yCg&J?l zqFPht>TNQBwLbLs|0SCGCZ2CjmEF;MYu51S^G_l$rfVCXt1mAvhoiy3-U))1O_heC zuE=hRAdC8u@IyGh3)zf5Rg2erUPekAlQEv9C7g-i#c=Q=%c%4-9MEKSRqoG$_U6N2 zG^5+P$OHhzlD*2B=;OzzASrFN0xakAIXxYT7)-G2Ol)%Cy~R-*LDh^NqU+`Z&0A%V z8V<#>5zQ}Y$gpdHCyx_`D4YK&{`4(lU*9H4qKc9pAjL`s+fp#_g66tnGU88w>ql2NTL#L`lxxr8Zcnl?`5K^vD;d>k-c-n#mAt~o-V@T3ia`JU_kReG zRnfXiEg4B+GG&QKF}Qt#y-FKs#hY>D&id^mRSnhvY50Q`>gu@WB@#X+A|!?gwaN`UgpZ>`UT2!f4D2y7oh=x*-su@9krK9f@?=0ylyNv5&LZp%ya1+w zM?Gw8o#X5hS)SwBt_XQiAS!LpaQdw>>`L(&=}rk1wU4bCMGim8FV=? zJA0UFhw}-p45ip9*a;-6NoIi~z|mCTBVZL&hR*y~eBxw^tCO%<=sW~l{kd-Kr>}yFqXE!yDi~ z9COvb;G?uU#NAjfe|@sDVc4lvK_**y-5!!uI@vZpj~po)4`fSJN~Ovu%2(nuj?b^()oXMb$OKa7YdM&w02hPt?1y0ei#o2n; z64MMUxmj4p_2FENR7iaR;k;O)VzVCW%s{VPua33!ToRm=(g-ExK&EHS_Al&?^okW6 zyTOwF*?|ioR5YDglUnz49g1ZEYCv;QAv zscX+!70tN<>y-7@z7*}l$IR&X9xaiS+aC#k(3I)e0Q*F<5VJ{i$8 z7K_-aGOScj%wwG6VMa^g;RXydkG+G)Q!L}(OndnG27pX9=5F|mD;K_FNU2@S;-e)~ zTJ&v|mT8w40PaxrK5w^<#2rOvm75Th zu^LBp=;PWJtfk7e-2?Eyetq9IQ-{BNecv`whhK3@rPTg)%=|N#BhvA9ju~O<;P%o`?c;=s zw>Wd0YW2q9q@|JfZFe=JoBMcscENnd7g}h(`Xd4wzj$p!hYxg6joMVUjXmNH_k9u!Y^IpBjF-o{{lUa-l@O-A#VDiCedH+Ve`HBVy~XmcRNET95iyJS^3d z_|&seW#_?`Zk3_>8v%KoQFX>8idwXLbK$lx3I%0i2k90IUoC0)4N~(u>K? z)WB&u>V7UPmq)h~t(6!VFXx&ZMNg;Df_7lZ4^w9BeI^n-f-Nx1DQx(U?8zF7?!0U~ zKQ-KMB7VR`53*eD;ih9Z*8>!w*e3*qm?)g=UC3}+73cjg&%*g+br%K@PadNkCv7Sf^As)UkVgT2yac#%`TTR_Z_8; z?bbj^7`n(NV!Kox_}<$V#k#WZoayQ0*>GcOtumW1WqveHF>eZ~S@-xAXDBp%^X5ma W9nCyhe{3#;z#Gs`^%=dN{`(Ja4!OYq