fix(client): DedupeRequestsPlugin not working well with Event Iterator - #921
Conversation
dosubot
Bot
added
the
size:M
WalkthroughRefactors AsyncIdQueue internals and replicateAsyncIterator to use clearer naming and string IDs; adds AsyncIdQueue.hasBufferedItems and waiterIds tests; updates replicateAsyncIterator error propagation and per-replica logic; tests adjusted to assert the first two next() calls concurrently and with lazy iterator initialization in with-async-iterator. Changes
Sequence Diagram(s)sequenceDiagram
autonumber
participant Caller
participant Wrapper as withAsyncIterator
participant Body as body()
participant Replicator as replicateAsyncIterator
participant Queue as AsyncIdQueue
participant Replica1 as Replica(id:'0')
participant Replica2 as Replica(id:'1')
participant ReplicaN as Replica(id:'2')
Caller->>Wrapper: call wrapper()
Wrapper->>Body: resolve body()
alt body is AsyncIterator
Wrapper->>Replicator: replicateAsyncIterator(body, count)
Replicator->>Queue: open("0"), open("1"), open("2")
Replicator->>Body: source.next() (loop)
Body-->>Replicator: item / throws reason
Replicator->>Queue: push(item) for each id
Note right of Queue: replicas may have buffered items
par concurrent pulls
Replica1->>Queue: pull("0")
Replica2->>Queue: pull("1")
end
alt no error
Queue-->>Replica1: return item
Queue-->>Replica2: return item
else on error(reason)
Replicator->>Queue: close(waiterIds, reason)
Queue-->>ReplicaX: reject(reason) if waiting and not buffered
end
Wrapper->>ReplicaN: shift() -> returns iterator instance
ReplicaN-->>Caller: AsyncIterator result(s)
else non-iterator
Wrapper-->>Caller: body result
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Suggested labels
Poem
Tip 🔌 Remote MCP (Model Context Protocol) integration is now available!Pro plan users can now connect to remote MCP servers from the Integrations page. Connect with popular remote MCPs such as Notion and Linear to add more context to your reviews and chats. 📜 Recent review detailsConfiguration used: CodeRabbit UI Review profile: CHILL Plan: Pro 💡 Knowledge Base configuration:
You can enable these sources in your CodeRabbit configuration. 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (4)
✨ Finishing Touches
🧪 Generate unit tests
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. CodeRabbit Commands (Invoked using PR/Issue comments)Type Other keywords and placeholders
CodeRabbit Configuration File (
|
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Summary of Changes
Hello @unnoq, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!
This pull request addresses a bug where the DedupeRequestsPlugin was not functioning correctly with Event Iterator streams. The problem stemmed from an improper re-initialization of replicated async iterators within the replicateStandardLazyResponse utility. The fix ensures that the underlying replicateAsyncIterator is invoked only once per response, guaranteeing consistent stream behavior for multiple consumers. Additionally, the changes include updates to related tests to properly validate concurrent asynchronous operations, preventing future regressions.
Highlights
- Fixing Async Iterator Replication Logic: The core issue was an incorrect re-initialization of replicated async iterators within the
replicateStandardLazyResponseutility. This has been fixed by ensuring thatreplicateAsyncIteratoris called only once per response body using the nullish coalescing operator (??=), preventing inconsistent behavior when multiple consumers access the same stream. - Improved Concurrent Test Coverage: Tests for both
replicateAsyncIteratorandreplicateStandardLazyResponsehave been updated to usePromise.allfor concurrentnext()andbody()calls. This change ensures that the tests accurately reflect and validate real-world parallel consumption scenarios, preventing potential race conditions in the test suite. - Minor Code Refactoring: Minor refactorings were applied in
iterator.tsto enhance code consistency and readability. This includes cachingid.toString()and simplifyingqueue.closecalls.
Using Gemini Code Assist
The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.
Invoking Gemini
You can request assistance from Gemini at any point in your pull request via creating an issue comment (i.e. comment on the pull request page) using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands.
| Feature | Command | Description |
|---|---|---|
| Code Review | /gemini review |
Performs a code review for the current pull request in its current state. |
| Pull Request Summary | /gemini summary |
Provides a summary of the current pull request in its current state. |
| Comment | @gemini-code-assist | Responds in comments when explicitly tagged, both in issue comments and review comments. |
| Help | /gemini help |
Displays a list of available commands. |
Customization
To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.
Limitations & Feedback
Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here.
You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.
Footnotes
-
Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩
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.
Code Review
This pull request addresses an issue where DedupeRequestsPlugin was not working correctly with event iterators, likely due to a race condition. The core of the fix is in replicateStandardLazyResponse, where a race condition is resolved by using a nullish coalescing assignment (??=). This ensures that replicateAsyncIterator is invoked only once, even when multiple consumers access the body concurrently. The accompanying test changes are excellent, as they now use Promise.all to simulate parallel execution and verify the fix. The other changes are minor refactorings that improve code clarity. Overall, this is a solid fix that correctly addresses the underlying issue.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (6)
packages/standard-server/src/utils.ts (2)
72-85: Avoid order-dependent shift; map replicas deterministically and in O(1).shift() ties iterator selection to call order across replicas and does O(n) array work per call. Capture the loop index and pick by index to:
- Guarantee replica i consistently gets iterators[i] regardless of call timing.
- Remove subtle scheduling dependencies in tests and production.
- Avoid the O(n) shift cost.
Apply this refactor:
- for (let i = 0; i < count; i++) { - replicated.push({ + for (let i = 0; i < count; i++) { + const idx = i + replicated.push({ ...response, body: once(async () => { const body = await (bodyPromise ??= response.body()) if (!isAsyncIteratorObject(body)) { return body } - replicatedAsyncIteratorObjects ??= replicateAsyncIterator(body, count) - return replicatedAsyncIteratorObjects.shift() + replicatedAsyncIteratorObjects ??= replicateAsyncIterator(body, count) + return replicatedAsyncIteratorObjects[idx]! }), }) }
2-2: Import the AsyncIteratorObject type if it isn’t globally available.replicatedAsyncIteratorObjects is typed as AsyncIteratorObject<...>[] here but the type isn’t imported in this file. If your TS config doesn’t expose it globally, add a type-only import:
-import { isAsyncIteratorObject, once, replicateAsyncIterator, toArray, tryDecodeURIComponent } from '@orpc/shared' +import type { AsyncIteratorObject } from '@orpc/shared' +import { isAsyncIteratorObject, once, replicateAsyncIterator, toArray, tryDecodeURIComponent } from '@orpc/shared'If AsyncIteratorObject is globally declared, you can ignore this.
Also applies to: 69-71
packages/shared/src/iterator.ts (2)
156-160: Rename shadowed “id” for readability.In replicated.every((_, id) => ...), id is a numeric index that shadows the outer string id. Rename to index to avoid confusion:
- if (replicated.every((_, id) => !queue.isOpen(id.toString()))) { + if (replicated.every((_, index) => !queue.isOpen(index.toString()))) {
115-121: Micro-optimization: precompute string IDs once.If count is modest this is negligible, but you can avoid repeated toString() calls by precomputing ids:
const replicated: AsyncIteratorClass<T, TReturn, TNext>[] = [] + const ids = Array.from({ length: count }, (_, i) => i.toString()) const start = once(async () => { try { while (true) { const item = await source.next() - for (let i = 0; i < count; i++) { - const id = i.toString() + for (let i = 0; i < count; i++) { + const id = ids[i] if (queue.isOpen(id)) { queue.push(id, item) } } if (item.done) break- for (let i = 0; i < count; i++) { - const id = i.toString() + for (let i = 0; i < count; i++) { + const id = ids[i] queue.open(id) replicated.push(new AsyncIteratorClass( () => { start() return new Promise((resolve, reject) => { - queue.pull(id) + queue.pull(id) .then(resolve) .catch(reject) defer(() => { if (error) { reject(error.value) } }) }) }, async (reason) => { - queue.close({ id }) + queue.close({ id }) if (reason !== 'next') { - if (replicated.every((_, index) => !queue.isOpen(index.toString()))) { + if (replicated.every((_, index) => !queue.isOpen(ids[index]))) { await source?.return?.() } } }, )) }Also applies to: 133-137, 156-160
packages/standard-server/src/utils.test.ts (2)
100-104: Optional: assert lazy creation right after replicateStandardLazyResponse.To harden the test’s intent (no eager replication), assert the spy count immediately after constructing replicated and before invoking body():
const replicated = replicateStandardLazyResponse(response, 3) expect(replicated.length).toBe(3) + expect(replicateAsyncIteratorSpy).toHaveBeenCalledTimes(0)
112-118: Consider testing out-of-order consumption.An additional check where replicated[2].body() is invoked before replicated[0]/[1] would ensure iterator selection remains correct even when consumers arrive out of order.
Do you want me to draft that test case?
Also applies to: 120-124
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (4)
packages/shared/src/iterator.test.ts(2 hunks)packages/shared/src/iterator.ts(3 hunks)packages/standard-server/src/utils.test.ts(1 hunks)packages/standard-server/src/utils.ts(1 hunks)
🧰 Additional context used
🧬 Code graph analysis (1)
packages/standard-server/src/utils.ts (1)
packages/shared/src/iterator.ts (1)
replicateAsyncIterator(100-166)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (3)
- GitHub Check: publish-commit
- GitHub Check: lint
- GitHub Check: test
🔇 Additional comments (6)
packages/standard-server/src/utils.ts (1)
69-85: Lazily initializing replicated iterators fixes the race with Event Iterator.Using bodyPromise ??= response.body() and replicatedAsyncIteratorObjects ??= replicateAsyncIterator(...) inside once(...) ensures:
- response.body() is invoked at most once even under concurrent body() calls.
- replicateAsyncIterator is created exactly once per response body.
This aligns with the PR goal (dedup + SSE) and removes the eager path that previously raced. Good change.packages/shared/src/iterator.ts (2)
115-121: String ID reuse is simpler and less error-prone.Consolidating id = i.toString() and reusing it across queue operations removes repeated conversions and keeps identifiers consistent across open/push/pull/close. Looks good.
133-137: Consistently use the precomputed string id.Using const id = i.toString() once per replica and passing id into queue.pull is the right direction and avoids mismatches. Nice cleanup.
Also applies to: 142-142
packages/standard-server/src/utils.test.ts (1)
114-118: Parallel body() assertions validate the intended concurrency path.Good addition. This catches regressions where iterator replication creates more than once or where selection depends on evaluation order.
packages/shared/src/iterator.test.ts (2)
266-269: Nice: exercising initial parallel next() calls.This better reflects real-world concurrent consumers and guards against serialization bugs in replication.
320-323: Parallel rejection checks cover late-error propagation well.Good to see concurrency verified in the error path too.
dosubot
Bot
added
size:L
There was a problem hiding this comment.
Actionable comments posted: 0
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
packages/shared/src/queue.test.ts (1)
89-103: Bug in test:pullPromise2listens on the wrong queue ID.This test intends to validate custom-reason rejection for both per-ID close and global close. However, both
pullPromise1andpullPromise2pull fromqueueId1, so the second assertion doesn’t actually exercise the global close path.Fix by pulling on
queueId2for the second promise:- const pullPromise1 = queue.pull(queueId1) - const pullPromise2 = queue.pull(queueId1) + const pullPromise1 = queue.pull(queueId1) + const pullPromise2 = queue.pull(queueId2)
🧹 Nitpick comments (5)
packages/shared/src/queue.ts (3)
34-53: Avoid O(n)Array.shift()in hot paths.Both the waiter and item queues use
shift(), which is O(n) due to reindexing. If these queues can grow (e.g., bursts from an event stream), consider using a small ring buffer or tracking a head index to make pops O(1).Optional direction (no API change, keeps semantics):
- Replace
Map<string, T[]>withMap<string, { buf: T[]; head: number }>and advanceheadinstead of shifting.- Compact occasionally when
headcrosses a threshold.Happy to draft a minimal refactor if you want to pursue this.
55-82: Promise waiter bookkeeping is solid; consider minor clarity tweak.The waiter enqueuing logic is correct and deletes the
waitersentry when it drains. As a small readability nit, rename the localwaitingPulls(Line 71) towaitersForIdto match thewaitersmap naming.
84-105: Close semantics are consistent; add an option to discard buffered items with a reason?Current behavior:
- Per-ID close rejects pending pulls for that ID and drops buffered items.
- Global close rejects all pending pulls and clears everything.
This is good. If upper layers need to propagate an error reason to consumers that might otherwise drain buffered items after an upstream failure, you could expose a
closeAll(reason)convenience or aclearBufferedItems(id?, reason?)method so higher layers don't need to iteratewaiterIds. Not required for this PR—just a thought based on usage inreplicateAsyncIterator.packages/shared/src/iterator.ts (2)
137-166: Minor readability tweak in “all replicas closed?” check.The
everycallback parameter is namedid(numeric index), which shadows the outer stringidvariable and can be confusing. Useidxfor the numeric index to make intent explicit:- if (replicated.every((_, id) => !queue.isOpen(id.toString()))) { + if (replicated.every((_, idx) => !queue.isOpen(idx.toString()))) {
156-163: Source cleanup trigger looks right; consider catchingreturn()errors.Calling
source?.return?.()only when all queues are closed is correct. Ifreturn()can throw (some sources do on abort), you might want to catch and ignore it here to avoid surfacing cleanup errors to consumers after they already decided to stop:- if (replicated.every((_, idx) => !queue.isOpen(idx.toString()))) { - await source?.return?.() - } + if (replicated.every((_, idx) => !queue.isOpen(idx.toString()))) { + try { + await source?.return?.() + } catch { /* noop: best-effort cleanup */ } + }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (4)
packages/shared/src/iterator.test.ts(3 hunks)packages/shared/src/iterator.ts(3 hunks)packages/shared/src/queue.test.ts(1 hunks)packages/shared/src/queue.ts(2 hunks)
🚧 Files skipped from review as they are similar to previous changes (1)
- packages/shared/src/iterator.test.ts
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (3)
- GitHub Check: publish-commit
- GitHub Check: lint
- GitHub Check: test
🔇 Additional comments (6)
packages/shared/src/queue.ts (1)
15-21: Nice, lightweight observability for waiters and buffers.The new
waiterIdsgetter andhasBufferedItems(id)are clean and minimal. These make the queue’s state easier to reason about from higher layers (e.g., iterator replication/error propagation) without leaking internals. Looks good.packages/shared/src/queue.test.ts (2)
136-152: waiterIds test is precise and valuable.Great coverage for insertion order and cleanup. Verifies the exact keys and that the set clears after resolution. Nice.
154-174: hasBufferedItems test accurately exercises FIFO buffering across IDs.Good end-to-end assertions before/after pulls and across isolated IDs. This will catch regressions in queue bookkeeping.
packages/shared/src/iterator.ts (3)
115-121: Replicating to string IDs aligns withAsyncIdQueueand removes toString noise.The switch to string IDs (and checking
queue.isOpen(id)before push) is clean and avoids accidental pushes after a replica closes. Good change.
128-134: Error propagation: closing only waiter queues is a good balance.Catching
reason, storing it inerror, then closing justwaiterIdsensures:
- Waiting replicas reject immediately with the upstream reason.
- Replicas with buffered items can still drain before surfacing the error on the next call.
This matches expected iterator semantics for many streaming sources. Nicely done.
145-154: TOCTOU guard aroundhasBufferedItemsvspullis acceptable here.Because each replica has its own per-ID queue and
AsyncIteratorClassserializesnext()calls viasequential, the “check then pull” is safe from cross-replica interference. If you ever allow multiple concurrentnext()calls per replica, you’d want to invert logic (attemptpulland handle immediate rejection) to avoid a tiny window. Not needed today.If you plan to relax serialization in the future, I can prototype a resilient
pullOrError(id, error)helper to encapsulate this pattern.
More templates
@orpc/arktype
@orpc/client
@orpc/contract
@orpc/experimental-durable-event-iterator
@orpc/hey-api
@orpc/interop
@orpc/json-schema
@orpc/nest
@orpc/openapi
@orpc/openapi-client
@orpc/otel
@orpc/react
@orpc/react-query
@orpc/experimental-react-swr
@orpc/server
@orpc/shared
@orpc/solid-query
@orpc/standard-server
@orpc/standard-server-aws-lambda
@orpc/standard-server-fetch
@orpc/standard-server-node
@orpc/standard-server-peer
@orpc/svelte-query
@orpc/tanstack-query
@orpc/trpc
@orpc/valibot
@orpc/vue-colada
@orpc/vue-query
@orpc/zod
commit: |
1 participant
Summary by CodeRabbit
New Features
Bug Fixes
Refactor
Tests