fix(shared): make Queue.pull() O(1) to avoid quadratic backlog drains - #103
Conversation
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LDugk98gQoDyPK2NCVvDuD
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LDugk98gQoDyPK2NCVvDuD
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LDugk98gQoDyPK2NCVvDuD
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LDugk98gQoDyPK2NCVvDuD
@standard-server/aws-lambda
@standard-server/core
@standard-server/fastify
@standard-server/fetch
@standard-server/node
@standard-server/peer
@standard-server/shared
commit: |
Merging this PR will not alter performance
Comparing Footnotes
|
Codecov Report✅ All modified and coverable lines are covered by tests. 📢 Thoughts on this report? Let us know! |
There was a problem hiding this comment.
✅ No new issues found.
Reviewed changes
- Head-pointer buffer in
Queue—packages/shared/src/queue.tsdropsArray.shift()inpull()for aheadindex over(T | undefined)[]; the pulled slot is overwritten withundefinedto release the reference. - Lazy compaction — once
head >= 1024 && head * 2 >= items.length, a fully drained buffer is reset viaitems.length = 0, otherwiseitems.splice(0, head), andheadreturns to0. abort()resetshead— keeps the pointer consistent with the cleared buffer.- Compaction-ordering test —
packages/shared/src/queue.test.tspushes 2048 items (every thirdundefined) and pulls them all, hitting both the partialsplice(pull #1024) and full-reset (pull #2048) branches.
I traced the 0 <= head <= items.length invariant through push/pull/close/abort and it holds; a pending pull can only exist when head === items.length, so compaction never races one. Private representation change only — no public API impact. Tests, tsc -b, and eslint all pass.
ℹ️ PR description overstates the memory benefit
The description says the change "prevents memory leaks from held references", but Array.prototype.shift() already removed and released each pulled element — there was no reference leak. The measurable win is CPU: pull() becomes amortized O(1) instead of O(n), removing the quadratic cost of draining a large backlog. Worth stating the rationale accurately since the old path is now in history.
Technical details
# PR rationale attributes a reference leak that didn't exist
## Affected sites
- PR body / commit `99d7286` ("fix(shared): make Queue.pull() O(1) ...") — describes the prior `shift()` path as leaking held references.
- `packages/shared/src/queue.ts:37-53` — the new head-pointer pull + compaction path.
## Required outcome
- Rationale should attribute the win to CPU (amortized O(1) `pull()`, no quadratic backlog drain), not to a reference leak.
- Optionally note the memory tradeoff: consumed slots are retained as `undefined` until compaction, so peak array length is bounded by roughly `threshold + backlog` and can be marginally higher than the old path for tiny queues.
## Open questions for the human (optional)
- Was "memory leak" meant to describe V8 backing-store capacity retention rather than element references? If so, the description could say so precisely.DeepSeek Flash (default — pick a model for stronger reviews) | 𝕏
…ts (#2104) Resuming an iterator subscription with `lastEventId` silently dropped the oldest missed events whenever more than `maxBufferedEvents` (default 100) were missed. Adapters replay the whole backlog in one burst before the consumer can read, so the buffer overflowed even for a fast consumer, and the client's `lastEventId` then skipped past the gap for good. Resumed subscribers now receive every missed event, while slow consumers still drop the oldest ones. ### Fixes - Memory, Redis, Upstash and Bun publishers deliver the full resume backlog to `subscribe(event, { lastEventId })` iterators. Before, resuming 150 missed events with the default limit lost the first 50. - A consumer that falls more than `maxBufferedEvents` further behind after resuming still drops the oldest events, so memory stays bounded by the backlog plus the limit. - Subscriptions without `lastEventId` behave exactly as before. ### Performance - Draining a large buffer no longer copies the array on every read: a 50K-event resume drains in about 57 ms instead of about 2 s. ### Notes for reviewers - Adapters must deliver the resume backlog before `subscribeListener` resolves. This is now documented on the method, and the Memory and BaseRedis adapters already do it. Cloudflare's Durable Object replays over the WebSocket after subscribing, so its resumed events still count toward the limit; left for a follow-up. - The private `Queue` mirrors the O(1) `Queue.pull` fix in middleapi/standard-server#103, which is unreleased. It can be replaced once standard-server ships a queue that exposes its size and can drop the oldest item. ### Testing - New regression tests in the base publisher, memory and Redis suites fail on `main` and pass here. - Publisher and Bun Redis suites pass against a local Redis 8.0 (Upstash skipped without credentials).

Summary
Optimized the Queue implementation to reduce memory overhead when pulling items by using a head pointer instead of repeatedly shifting the array. This prevents memory leaks from held references and amortizes the cost of array compaction.
Key Changes
Array.shift()with a head pointer approach to avoid O(n) operations on every pullundefinedafter being pulled to allow garbage collectionabort()to reset the head pointer when clearing the queueundefinedvaluesImplementation Details
COMPACT_THRESHOLDconstant (1024) to control when compaction becomes eligiblehead >= COMPACT_THRESHOLD && head * 2 >= items.length, ensuring the amortized cost remains acceptablehead === items.length), simply reset array length instead of splicing(T | undefined)[]to reflect that pulled items are set to undefinedhttps://claude.ai/code/session_01LDugk98gQoDyPK2NCVvDuD