From 99d72869d0e48213d0f04c13f33af7196f4795aa Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 25 Sep 2026 13:07:40 +0000 Subject: [PATCH 1/4] fix(shared): make Queue.pull() O(1) to avoid quadratic backlog drains Queue.pull() used Array#shift(), which is O(n) once V8 stops using its fast path for large arrays (~16k+ elements). Draining a large buffered backlog, e.g. a peer streaming many octet-stream chunks before the handler reads the body, was therefore quadratic and blocked the event loop (200k items: ~3.3s before, ~21ms after). Track a head index instead, clear pulled slots so items can be GC'd, reset the array once fully drained, and compact with splice once at least half of a large array is consumed (amortized O(1) per pull). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LDugk98gQoDyPK2NCVvDuD --- packages/shared/src/queue.test.ts | 43 +++++++++++++++++++++++++++++++ packages/shared/src/queue.ts | 25 +++++++++++++++--- 2 files changed, 65 insertions(+), 3 deletions(-) diff --git a/packages/shared/src/queue.test.ts b/packages/shared/src/queue.test.ts index 1a5f550..ec89937 100644 --- a/packages/shared/src/queue.test.ts +++ b/packages/shared/src/queue.test.ts @@ -22,6 +22,49 @@ describe('queue', () => { expect(await queue.pull()).toBe('a') }) + it('keeps order across internal compaction of a large backlog', async () => { + const queue = new Queue() + const pulled: (number | undefined)[] = [] + let next = 0 + + // Keep the buffer non-empty while pulling so it never fully drains between batches. + for (let round = 0; round < 5; round++) { + for (let i = 0; i < 3000; i++) { + queue.push(next % 7 === 0 ? undefined : next) + next++ + } + + for (let i = 0; i < 2500; i++) { + pulled.push(await queue.pull()) + } + } + + queue.close() + + while (pulled.length < next) { + pulled.push(await queue.pull()) + } + + await expect(queue.pull()).rejects.toThrow(AbortError) + expect(pulled).toEqual(Array.from({ length: next }, (_, i) => i % 7 === 0 ? undefined : i)) + }) + + it('abort after a partial drain discards the remaining items', async () => { + const queue = new Queue() + + for (let i = 0; i < 5000; i++) { + queue.push(i) + } + + for (let i = 0; i < 3000; i++) { + expect(await queue.pull()).toBe(i) + } + + queue.abort() + + await expect(queue.pull()).rejects.toThrow('Queue was aborted.') + }) + it('resolves a pending pull on push', async () => { const queue = new Queue() diff --git a/packages/shared/src/queue.ts b/packages/shared/src/queue.ts index b4db816..3f4bec7 100644 --- a/packages/shared/src/queue.ts +++ b/packages/shared/src/queue.ts @@ -1,7 +1,12 @@ import { AbortError } from './error' export class Queue { - private readonly items: T[] = [] + /** + * Buffered items live at `items[head..]`. Advancing `head` instead of calling `Array#shift()` + * keeps pulls O(1); `shift()` becomes O(n) on large arrays, making a backlog drain quadratic. + */ + private readonly items: (T | undefined)[] = [] + private head = 0 private readonly pendingPulls: (readonly [resolve: (item: T) => void, reject: (err: unknown) => void])[] = [] private closed: undefined | { reason: unknown } @@ -30,8 +35,21 @@ export class Queue { * @throws when the queue is closed or aborted. Note that buffered items can still be pulled after close until the buffer is drained. */ async pull(): Promise { - if (this.items.length > 0) { - return this.items.shift() as T + if (this.head < this.items.length) { + const item = this.items[this.head] as T + this.items[this.head++] = undefined // release the reference so pulled items can be GC'd + + if (this.head === this.items.length) { + this.items.length = 0 + this.head = 0 + } + else if (this.head >= 1024 && this.head * 2 >= this.items.length) { + // Compact once at least half the array is consumed, so the O(n) splice is amortized O(1) per pull. + this.items.splice(0, this.head) + this.head = 0 + } + + return item } if (this.closed) { @@ -66,6 +84,7 @@ export class Queue { abort(reason?: unknown): void { reason ??= new AbortError('Queue was aborted.') this.items.length = 0 + this.head = 0 this.close(reason) } } From 5ebc7de467006a72d1ed280bf27b426ea145aa3f Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 25 Sep 2026 13:29:20 +0000 Subject: [PATCH 2/4] test(shared): trim Queue compaction test to the paths it needs A single fill-then-drain of 3000 items already hits both compactions and the fully-drained reset. Drop the abort-after-partial-drain test, which cannot fail differently from the existing abort test. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LDugk98gQoDyPK2NCVvDuD --- packages/shared/src/queue.test.ts | 35 ++----------------------------- 1 file changed, 2 insertions(+), 33 deletions(-) diff --git a/packages/shared/src/queue.test.ts b/packages/shared/src/queue.test.ts index ec89937..034341a 100644 --- a/packages/shared/src/queue.test.ts +++ b/packages/shared/src/queue.test.ts @@ -22,47 +22,16 @@ describe('queue', () => { expect(await queue.pull()).toBe('a') }) - it('keeps order across internal compaction of a large backlog', async () => { - const queue = new Queue() - const pulled: (number | undefined)[] = [] - let next = 0 - - // Keep the buffer non-empty while pulling so it never fully drains between batches. - for (let round = 0; round < 5; round++) { - for (let i = 0; i < 3000; i++) { - queue.push(next % 7 === 0 ? undefined : next) - next++ - } - - for (let i = 0; i < 2500; i++) { - pulled.push(await queue.pull()) - } - } - - queue.close() - - while (pulled.length < next) { - pulled.push(await queue.pull()) - } - - await expect(queue.pull()).rejects.toThrow(AbortError) - expect(pulled).toEqual(Array.from({ length: next }, (_, i) => i % 7 === 0 ? undefined : i)) - }) - - it('abort after a partial drain discards the remaining items', async () => { + it('keeps order across internal compaction', async () => { const queue = new Queue() - for (let i = 0; i < 5000; i++) { + for (let i = 0; i < 3000; i++) { queue.push(i) } for (let i = 0; i < 3000; i++) { expect(await queue.pull()).toBe(i) } - - queue.abort() - - await expect(queue.pull()).rejects.toThrow('Queue was aborted.') }) it('resolves a pending pull on push', async () => { From 73009c7599374410a51e04ed2dc2948a9f6888ad Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 25 Sep 2026 13:43:51 +0000 Subject: [PATCH 3/4] perf(shared): only compact Queue buffer past a threshold Resetting the buffer with `length = 0` on every full drain made V8 drop and reallocate the backing store on each pull when the consumer keeps up (the common 0 -> 1 -> 0 case). Fold the reset into the thresholded compaction instead, so it runs at most once per 1024 pulls: alternating push/pull goes from ~161ms to ~85ms per 1M ops, with backlog drains unchanged. Also name the threshold constant, trim the field comment, and make the compaction test cover both the splice and the full-drain reset. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LDugk98gQoDyPK2NCVvDuD --- packages/shared/src/queue.test.ts | 8 ++++++-- packages/shared/src/queue.ts | 23 ++++++++++++----------- 2 files changed, 18 insertions(+), 13 deletions(-) diff --git a/packages/shared/src/queue.test.ts b/packages/shared/src/queue.test.ts index 034341a..419c19a 100644 --- a/packages/shared/src/queue.test.ts +++ b/packages/shared/src/queue.test.ts @@ -25,13 +25,17 @@ describe('queue', () => { it('keeps order across internal compaction', async () => { const queue = new Queue() - for (let i = 0; i < 3000; i++) { + // Splices the pulled half at 1024, then fully drains the remaining 1024. + for (let i = 0; i < 2048; i++) { queue.push(i) } - for (let i = 0; i < 3000; i++) { + for (let i = 0; i < 2048; i++) { expect(await queue.pull()).toBe(i) } + + queue.push(2048) + expect(await queue.pull()).toBe(2048) }) it('resolves a pending pull on push', async () => { diff --git a/packages/shared/src/queue.ts b/packages/shared/src/queue.ts index 3f4bec7..59a187e 100644 --- a/packages/shared/src/queue.ts +++ b/packages/shared/src/queue.ts @@ -1,10 +1,9 @@ import { AbortError } from './error' +const COMPACT_THRESHOLD = 1024 + export class Queue { - /** - * Buffered items live at `items[head..]`. Advancing `head` instead of calling `Array#shift()` - * keeps pulls O(1); `shift()` becomes O(n) on large arrays, making a backlog drain quadratic. - */ + /** Items before `head` have already been pulled. */ private readonly items: (T | undefined)[] = [] private head = 0 private readonly pendingPulls: (readonly [resolve: (item: T) => void, reject: (err: unknown) => void])[] = [] @@ -39,13 +38,15 @@ export class Queue { const item = this.items[this.head] as T this.items[this.head++] = undefined // release the reference so pulled items can be GC'd - if (this.head === this.items.length) { - this.items.length = 0 - this.head = 0 - } - else if (this.head >= 1024 && this.head * 2 >= this.items.length) { - // Compact once at least half the array is consumed, so the O(n) splice is amortized O(1) per pull. - this.items.splice(0, this.head) + // Compact once at least half the array is consumed, so the O(n) splice is amortized O(1) per pull. + if (this.head >= COMPACT_THRESHOLD && this.head * 2 >= this.items.length) { + if (this.head === this.items.length) { + this.items.length = 0 + } + else { + this.items.splice(0, this.head) + } + this.head = 0 } From 011d3b4f47a5c2bc9057d3f437ef4299ef66e9a4 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 25 Sep 2026 14:00:37 +0000 Subject: [PATCH 4/4] test(shared): cover undefined items across Queue compaction The queue decides whether an item is buffered by index, not by value, so undefined items must survive both the splice and the full-drain reset. Mix undefined into the compaction test; the trailing push stays a defined value so a stale cleared slot cannot pass as a real item. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LDugk98gQoDyPK2NCVvDuD --- packages/shared/src/queue.test.ts | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/packages/shared/src/queue.test.ts b/packages/shared/src/queue.test.ts index 419c19a..70177ce 100644 --- a/packages/shared/src/queue.test.ts +++ b/packages/shared/src/queue.test.ts @@ -22,16 +22,17 @@ describe('queue', () => { expect(await queue.pull()).toBe('a') }) - it('keeps order across internal compaction', async () => { - const queue = new Queue() + it('keeps order across internal compaction, including undefined items', async () => { + const queue = new Queue() + const value = (i: number) => i % 3 === 0 ? undefined : i // Splices the pulled half at 1024, then fully drains the remaining 1024. for (let i = 0; i < 2048; i++) { - queue.push(i) + queue.push(value(i)) } for (let i = 0; i < 2048; i++) { - expect(await queue.pull()).toBe(i) + expect(await queue.pull()).toBe(value(i)) } queue.push(2048)