fix(ipc): serialize concurrent execute() calls through a request queue#6
Open
belowzeroff wants to merge 1 commit into
Open
fix(ipc): serialize concurrent execute() calls through a request queue#6belowzeroff wants to merge 1 commit into
belowzeroff wants to merge 1 commit into
Conversation
The wire protocol carries no request id: a RESP is correlated to a SYNC request purely by being the next thing to arrive on the socket after it was sent, so at most one SYNC request may be outstanding at a time. But RayforceIpcClient.execute() had a single pendingResolve/pendingReject slot with no such guard, so a second concurrent call (a double Enter in the REPL, a pagination click landing while refreshEnv() still has a request in flight) silently overwrote the first call's callbacks: whichever response arrived first got handed to the wrong caller, and the caller whose callbacks got overwritten just hung until its own timeout. Reworked execute() around a FIFO request queue: each call enqueues and calls pumpQueue(), which sends the next request only once the current one has settled (response, timeout, or write error, via finishCurrent()). finishCurrent() checks that the request it's asked to settle is still the current one, so a stale timeout firing after rejectAll() already cleared it (disconnect mid-flight) is a no-op rather than a double-settle. disconnect() and the socket 'close' handler now drain the whole queue through rejectAll(), not just the one in-flight request, so queued callers fail immediately instead of hanging until their own timeout on a dead socket. scripts/test-ipc.js: added a concurrency test that fires 10 execute() calls back-to-back without awaiting and asserts each resolves to its own result. Confirmed this reproduces the bug against the pre-fix client (request #9's promise settles with request #0's result "0n", requests #0-8 hang to timeout) and passes against the fix. Full suite: 33/33. Co-authored-by: Jack Bell <6161232+belljack@users.noreply.github.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem
The wire protocol carries no request id (
ray_ipc_header_tisprefix/version/flags/endian/msgtype/size— seeserde.hupstream): aRESPis correlated to aSYNCrequest purely by being the next thing to arrive on the socket after it was sent. That means at most oneSYNCrequest may be outstanding at a time.RayforceIpcClient.execute()didn't enforce that. It kept a singlependingResolve/pendingRejectslot and wrote straight to the socket on every call, so a second concurrent call — a double Enter in the REPL input, a pagination click landing whilerefreshEnv()still has a request in flight — silently overwrote the first call's callbacks. Whichever response arrived first got handed to whichever caller happened to be last to callexecute(), and every other caller just hung until its own timeout.Fix
execute()now enqueues onto a FIFOrequestQueueand callspumpQueue(), which sends the next request only once the current one has fully settled (response, timeout, or write error — all funneled throughfinishCurrent()).finishCurrent()checks that the request it's asked to settle is still the current one, so a stale timeout firing afterrejectAll()already cleared it (e.g. a disconnect mid-flight) is a no-op instead of a double-settle.disconnect()and the socket'close'handler now drain the entire queue viarejectAll(), not just the one in-flight request, so anything still queued fails immediately instead of hanging on a dead socket until its own timeout.executeAsync()is untouched — async messages never get aRESP, so they don't need to go through the queue and can still be fired at any time.Testing
scripts/test-ipc.jsgained a concurrency test: fire 10execute()calls back-to-back without awaiting between them, then assert each one resolves to its own result. I confirmed this reproduces the bug against the pre-fix client — request #9's promise (the last call, whose closure ends up in the single slot) settles with request #0's result ((* 0 0)→0n), and requests #0–8 hang until their 3s test timeout — and passes cleanly against the fix. Full suite: 33/33.