👍 Add accumulate() function for automatic RPC batching with parallel execution support #297
Conversation
WalkthroughAdds an accumulate batching helper for Denops RPCs, a cancellation error type, tests, public exports, and an import-map alias for std/async. Provides AccumulateHelper with guarded batched call methods and an exported accumulate(denops, executor) API. Changes
Sequence Diagram(s)sequenceDiagram
participant User
participant Accumulate as accumulate()
participant Helper as AccumulateHelper
participant Scheduler as nextTick
participant DenopsBatch as denops.batch()
User->>Accumulate: accumulate(denops, executor)
Accumulate->>Helper: new Helper(denops)
Accumulate->>User: run executor(helper)
loop executor schedules batched calls
User->>Helper: helper.call / .batch / .cmd / .eval / .dispatch
Helper->>Helper: enqueue Call, return Promise
Helper->>Scheduler: schedule `#resolvePendingCalls`
end
Scheduler->>Helper: `#resolvePendingCalls`
Helper->>DenopsBatch: denops.batch(enqueued calls)
DenopsBatch-->>Helper: results[] or BatchError
alt success
Helper->>Helper: map results, resolve Promises
Helper-->>User: Promises resolved
else batch error / cancellation
Helper->>Helper: wrap/convert errors (BatchError | AccumulateCancelledError)
Helper-->>User: Promises rejected
end
User->>Accumulate: executor returns
Accumulate->>Helper: close() / cleanup
Accumulate-->>User: final awaited result
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes
Poem
Pre-merge checks and finishing touches✅ Passed checks (3 passed)
✨ Finishing touches
🧪 Generate unit tests (beta)
📜 Recent review detailsConfiguration used: CodeRabbit UI Review profile: CHILL Plan: Pro 📒 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). (6)
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. Comment |
Add accumulate() function that automatically batches multiple RPC calls that occur at the same timing during microtask processing. This enables parallel RPC execution with automatic batching and proper error handling.
Milly
force-pushed
the
batch-accumulate
branch
from
6d1c3c9 to
c4da175
Compare
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #297 +/- ##
==========================================
+ Coverage 84.82% 85.73% +0.91%
==========================================
Files 64 66 +2
Lines 2879 3064 +185
Branches 281 306 +25
==========================================
+ Hits 2442 2627 +185
Misses 435 435
Partials 2 2 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
![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 (3)
batch/accumulate.ts (3)
1-1: Prefer queueMicrotask for portability (Deno-first) over node:process nextTick.Using node:process couples to Node polyfills and has different ordering vs Promise microtasks. queueMicrotask keeps semantics simple and avoids Node dependency.
Apply this diff:
-import { nextTick } from "node:process"; +// Prefer runtime-agnostic microtask scheduling +const schedule = queueMicrotask; ... - nextTick(() => { + schedule(() => { if (end === this.#calls.length) { this.#resolvePendingCalls(); } });Also applies to: 139-144
19-22: Consider std/async deferred() instead of Promise.withResolvers() for broader compatibility.Some environments lag Promise.withResolvers(). std/async’s deferred is stable across Deno versions.
- readonly #disposer = Promise.withResolvers<void>(); + // Using std/async deferred for compatibility + readonly #disposer = deferred<void>(); ... - #resolvedWaiter = Promise.withResolvers<void>(); + #resolvedWaiter = deferred<void>(); ... - const { resolve } = this.#resolvedWaiter; - this.#resolvedWaiter = Promise.withResolvers(); + const { resolve } = this.#resolvedWaiter; + this.#resolvedWaiter = deferred<void>();Add import:
+import { deferred } from "@std/async/deferred";Also applies to: 160-162
84-109: Batch error plumbing is correct; minor message polish (optional).Current repr uses the first fn and total count; optional: include the index at which failure occurred for quicker triage.
- const [[fn]] = calls; - const repr = `[['${fn}', ...], ... total ${calls.length} calls]`; + const [[fn]] = calls; + const repr = `[['${fn}', ...], ... total ${calls.length} calls] (failed at index ${errorIndex})`;
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (5)
batch/accumulate.ts(1 hunks)batch/accumulate_test.ts(1 hunks)batch/error.ts(1 hunks)batch/mod.ts(2 hunks)deno.jsonc(1 hunks)
🧰 Additional context used
🧬 Code graph analysis (3)
batch/error.ts (1)
batch/accumulate.ts (2)
calls(135-154)calls(169-191)
batch/accumulate_test.ts (3)
batch/accumulate.ts (5)
calls(135-154)calls(169-191)accumulate(254-264)batch(84-109)TypeError(127-133)batch/collect.ts (1)
collect(138-151)batch/error.ts (1)
AccumulateCancelledError(22-43)
batch/accumulate.ts (1)
batch/error.ts (1)
AccumulateCancelledError(22-43)
🔇 Additional comments (8)
deno.jsonc (1)
47-47: Add std/async alias — LGTM.Alignment with tests (delay) and future use is fine. No action needed.
batch/error.ts (1)
1-43: Error type design looks solid.Clear payload (calls) and stable name via static initializer. Good fit for cancellation signaling.
batch/mod.ts (1)
6-8: Public surface expansion and docs — LGTM.Re-exports and example are coherent with accumulate() usage.
Also applies to: 46-49
batch/accumulate.ts (2)
61-83: Error propagation from call() reads clean.Good separation of error/cancel/unknown cases; cancellation message includes call repr and carries cause/calls.
169-191: Edge-cases handled well in #resolveCalls().
- BatchError path composes results + error + cancels correctly.
- Non-BatchError path maps to “unknown” errors with cause.
batch/accumulate_test.ts (3)
320-346: Nested accumulate/batch/collect scenarios — great coverage.These steps validate interop and batching boundaries thoroughly.
Also applies to: 347-369, 370-394
425-437: Outside-scope rejection and concurrent disposal behavior — excellent.Covers disposer race, ensures no stray denops.batch calls. Nicely prevents unhandled rejections.
Also applies to: 438-466, 467-495, 496-543
1014-1021: Review comment is incorrect — code already rejects as intended.The review misunderstands how
resolvesNextfrom@std/testing/mockworks. WhenresolvesNextreceives an Error object in the iterable, it automatically throws/rejects that Error when the stubbed function is awaited.At lines 1014-1021 and 691-697,
resolvesNext<unknown[]>([underlyingError])withunderlyingError = new Error("Network error")is correct and will reject as expected. The test assertions (e.g.,assertRejects(() => actual, Error, "Network error")at line 1032) confirm the intended rejection behavior occurs.The pattern difference at line 730—using
() => Promise.reject(underlyingError)instead—exists because that test uses a non-Error value (42), whichresolvesNextwould not automatically reject.No changes are needed.
Likely an incorrect or invalid review comment.
lambdalisue
left a comment
lambdalisue
left a comment
There was a problem hiding this comment.
How should we distinguish this from the existing ones?
Is it something that replaces them, or is it meant to be used in a different context?
If it’s something that replaces the existing ones, we don’t need to delete them, but I’d like to mark them as deprecated.
|
Each serves a specific purpose:
They complement each other nicely - developers can choose based on their specific needs. I don't think we need to deprecate the existing functions. What do you think? |
OK. Then, we should have clear documentation for that difference. |
Wrote to module document (JSDoc in |
Summary
accumulate()function to thebatchmodule that automatically batches multiple RPC callsPromise.all()while maintaining automatic batchingAccumulateCancelledErrorfor proper error handling in parallel execution scenariosBackground
The existing
batch()andcollect()functions require sequential execution and manual batching. The newaccumulate()function enables more natural async/await patterns with automatic batching during microtask processing.This function was originally developed in a separate repository and has been published on jsr.io with proven usage in various Denops plugins. Given its
maturity and utility, we're now integrating it into the standard library.
Example Usage
In this example, the following 3 RPCs are called:
getlinematchstrcalls in one RPClencalls in one RPCTesting
Breaking Changes
None. This is a purely additive change.
Related
Additional Context
gather()was suggested by AI but rejected due to:versions
accumulate()in some plugin librariesSummary by CodeRabbit
New Features
Tests
Chores