Conversation
🦋 Changeset detectedLatest commit: 3fc73a8 The changes in this PR will be included in the next version bump. This PR includes changesets to release 1 package
Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
@cloudflare/autoconfig
@cloudflare/build-output-utils
@cloudflare/codemods
@cloudflare/config
@cloudflare/containers-shared
create-cloudflare
@cloudflare/deploy-helpers
@cloudflare/kv-asset-handler
miniflare
@cloudflare/pages-functions
@cloudflare/pages-shared
@cloudflare/unenv-preset
@cloudflare/vite-plugin
@cloudflare/vitest-plugin
@cloudflare/workers-auth
@cloudflare/workers-editor-shared
@cloudflare/workers-utils
wrangler
commit: |
![ask-bonk[bot]](https://avatars.githubusercontent.com/in/2454580?s=60&v=4){kind=small}
Comment on lines
+183
to
+244
| // Sockets left open for another listener, mapped to their write baseline. | ||
| // `server.close()` would wait on them forever if nobody answers. | ||
| const unresolved = new Map<Duplex, number>(); | ||
| let closing = false; | ||
|
|
||
| const nodeServer = server as Server; | ||
| const emit = nodeServer.emit.bind(nodeServer); | ||
|
|
||
| // `once()` and `prependOnceListener()` owners remove themselves as they are | ||
| // invoked, so whether anyone else could own an upgrade is only knowable | ||
| // before the event is dispatched. | ||
| nodeServer.emit = function (event: string, ...args: unknown[]) { | ||
| if (event === "upgrade") { | ||
| const [request, socket] = args as [IncomingMessage, Duplex]; | ||
| snapshots.set(request, { | ||
| hasOtherListeners: nodeServer.listenerCount("upgrade") > 1, | ||
| bytesWritten: bytesWritten(socket), | ||
| }); | ||
| } | ||
| return emit(event, ...args); | ||
| } as Server["emit"]; | ||
|
|
||
| const close = nodeServer.close.bind(nodeServer); | ||
| nodeServer.close = function (callback?: (error?: Error) => void) { | ||
| closing = true; | ||
| for (const [socket, baseline] of unresolved) { | ||
| if (bytesWritten(socket) === baseline) { | ||
| socket.destroy(); | ||
| } | ||
| } | ||
| unresolved.clear(); | ||
| return close(callback); | ||
| } as Server["close"]; | ||
|
|
||
| nodeServer.on("listening", () => { | ||
| closing = false; | ||
| }); | ||
|
|
||
| return function upgrade(request: IncomingMessage, socket: Duplex): Upgrade { | ||
| const snapshot = snapshots.get(request) ?? { | ||
| hasOtherListeners: nodeServer.listenerCount("upgrade") > 1, | ||
| bytesWritten: bytesWritten(socket), | ||
| }; | ||
| // `bytesWritten` is cumulative for the connection, so a reused keep-alive | ||
| // socket needs the baseline taken when this upgrade started. | ||
| const isClaimed = () => bytesWritten(socket) > snapshot.bytesWritten; | ||
|
|
||
| return { | ||
| isAnswerable() { | ||
| return !isClaimed() && !socket.destroyed && socket.writable; | ||
| }, | ||
| release() { | ||
| if (isClaimed() || socket.destroyed) { | ||
| return; | ||
| } | ||
| if (!snapshot.hasOtherListeners || closing) { | ||
| socket.destroy(); | ||
| return; | ||
| } | ||
| unresolved.set(socket, snapshot.bytesWritten); | ||
| socket.once("close", () => unresolved.delete(socket)); | ||
| }, |
Contributor
There was a problem hiding this comment.
Claimed sockets are returned before entering unresolved, while the shutdown wrapper only destroys unclaimed entries. As a result, an upgrade that another listener has already completed (for example DevTools) remains open and makes httpServer.close() wait indefinitely, blocking Vite shutdown/restart. Track every socket left to another possible owner, and destroy all tracked sockets during shutdown.
Suggested change
| // Sockets left open for another listener, mapped to their write baseline. | |
| // `server.close()` would wait on them forever if nobody answers. | |
| const unresolved = new Map<Duplex, number>(); | |
| let closing = false; | |
| const nodeServer = server as Server; | |
| const emit = nodeServer.emit.bind(nodeServer); | |
| // `once()` and `prependOnceListener()` owners remove themselves as they are | |
| // invoked, so whether anyone else could own an upgrade is only knowable | |
| // before the event is dispatched. | |
| nodeServer.emit = function (event: string, ...args: unknown[]) { | |
| if (event === "upgrade") { | |
| const [request, socket] = args as [IncomingMessage, Duplex]; | |
| snapshots.set(request, { | |
| hasOtherListeners: nodeServer.listenerCount("upgrade") > 1, | |
| bytesWritten: bytesWritten(socket), | |
| }); | |
| } | |
| return emit(event, ...args); | |
| } as Server["emit"]; | |
| const close = nodeServer.close.bind(nodeServer); | |
| nodeServer.close = function (callback?: (error?: Error) => void) { | |
| closing = true; | |
| for (const [socket, baseline] of unresolved) { | |
| if (bytesWritten(socket) === baseline) { | |
| socket.destroy(); | |
| } | |
| } | |
| unresolved.clear(); | |
| return close(callback); | |
| } as Server["close"]; | |
| nodeServer.on("listening", () => { | |
| closing = false; | |
| }); | |
| return function upgrade(request: IncomingMessage, socket: Duplex): Upgrade { | |
| const snapshot = snapshots.get(request) ?? { | |
| hasOtherListeners: nodeServer.listenerCount("upgrade") > 1, | |
| bytesWritten: bytesWritten(socket), | |
| }; | |
| // `bytesWritten` is cumulative for the connection, so a reused keep-alive | |
| // socket needs the baseline taken when this upgrade started. | |
| const isClaimed = () => bytesWritten(socket) > snapshot.bytesWritten; | |
| return { | |
| isAnswerable() { | |
| return !isClaimed() && !socket.destroyed && socket.writable; | |
| }, | |
| release() { | |
| if (isClaimed() || socket.destroyed) { | |
| return; | |
| } | |
| if (!snapshot.hasOtherListeners || closing) { | |
| socket.destroy(); | |
| return; | |
| } | |
| unresolved.set(socket, snapshot.bytesWritten); | |
| socket.once("close", () => unresolved.delete(socket)); | |
| }, | |
| // Sockets left open for another listener. `server.close()` would wait on | |
| // them forever if nobody answers, or if their owner keeps them alive. | |
| const unresolved = new Set<Duplex>(); | |
| let closing = false; | |
| const nodeServer = server as Server; | |
| const emit = nodeServer.emit.bind(nodeServer); | |
| // `once()` and `prependOnceListener()` owners remove themselves as they are | |
| // invoked, so whether anyone else could own an upgrade is only knowable | |
| // before the event is dispatched. | |
| nodeServer.emit = function (event: string, ...args: unknown[]) { | |
| if (event === "upgrade") { | |
| const [request, socket] = args as [IncomingMessage, Duplex]; | |
| snapshots.set(request, { | |
| hasOtherListeners: nodeServer.listenerCount("upgrade") > 1, | |
| bytesWritten: bytesWritten(socket), | |
| }); | |
| } | |
| return emit(event, ...args); | |
| } as Server["emit"]; | |
| const close = nodeServer.close.bind(nodeServer); | |
| nodeServer.close = function (callback?: (error?: Error) => void) { | |
| closing = true; | |
| for (const socket of unresolved) { | |
| socket.destroy(); | |
| } | |
| unresolved.clear(); | |
| return close(callback); | |
| } as Server["close"]; | |
| nodeServer.on("listening", () => { | |
| closing = false; | |
| }); | |
| return function upgrade(request: IncomingMessage, socket: Duplex): Upgrade { | |
| const snapshot = snapshots.get(request) ?? { | |
| hasOtherListeners: nodeServer.listenerCount("upgrade") > 1, | |
| bytesWritten: bytesWritten(socket), | |
| }; | |
| // `bytesWritten` is cumulative for the connection, so a reused keep-alive | |
| // socket needs the baseline taken when this upgrade started. | |
| const isClaimed = () => bytesWritten(socket) > snapshot.bytesWritten; | |
| return { | |
| isAnswerable() { | |
| return !isClaimed() && !socket.destroyed && socket.writable; | |
| }, | |
| release() { | |
| if (socket.destroyed) { | |
| return; | |
| } | |
| if (closing || (!isClaimed() && !snapshot.hasOtherListeners)) { | |
| socket.destroy(); | |
| return; | |
| } | |
| if (snapshot.hasOtherListeners) { | |
| unresolved.add(socket); | |
| socket.once("close", () => unresolved.delete(socket)); | |
| } | |
| }, | |
| }; | |
| }; |
Contributor
|
I'm Bonk, and I've done a quick review of your PR. Prevents the Cloudflare WebSocket handler from closing upgrades claimed by other Vite listeners.
|
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
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
2 participants
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.
This PR fixes #15654 by making the Cloudflare Vite plugin defer to another
upgradelistener that has claimed, or could still claim, a WebSocket connection.Why
upgradelistener on Vite's HTTP server. The Cloudflare listener therefore shares requests with Vite HMR and third-party listeners such as Vite DevTools.dispatchFetch()was pending, this closed its connection with code 1006.Architectural Changes
Code Changes
packages/vite-plugin-cloudflare/src/websockets.tsnow tracks each upgrade's write baseline and listener ownership, releases unowned Worker WebSockets, and handles server shutdown and restart..changeset/lucky-pans-listen.mdrecords a patch release for@cloudflare/vite-plugin.