feat(miniflare): support node bindings over dev registry#10142
feat(miniflare): support node bindings over dev registry#10142edmundhung merged 7 commits intomainfrom
Conversation
🦋 Changeset detectedLatest commit: 1e3edae The changes in this PR will be included in the next version bump. This PR includes changesets to release 5 packages
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 |
edmundhung
force-pushed
the
edmundhung/dev-registry-on-worker-thread
branch
from
dd8f71c to
c532d1e
Compare
edmundhung
changed the title
edmundhung
force-pushed
the
edmundhung/dev-registry-on-worker-thread
branch
from
c532d1e to
a7e8cd3
Compare
create-cloudflare
@cloudflare/kv-asset-handler
miniflare
@cloudflare/pages-shared
@cloudflare/unenv-preset
@cloudflare/vite-plugin
@cloudflare/vitest-pool-workers
@cloudflare/workers-editor-shared
wrangler
commit: |
edmundhung
force-pushed
the
edmundhung/dev-registry-on-worker-thread
branch
from
a7054d3 to
7c512ab
Compare
petebacondarwin
left a comment
petebacondarwin
left a comment
There was a problem hiding this comment.
Started reviewing but ran out of time...
edmundhung
force-pushed
the
edmundhung/dev-registry-on-worker-thread
branch
from
7e27ac8 to
9854af3
Compare
| #getFallbackServiceAddress( | ||
| service: string, | ||
| entrypoint: string | ||
| ): { | ||
| httpStyle: "host" | "proxy"; | ||
| protocol: "http" | "https"; | ||
| host: string; | ||
| port: number; | ||
| } { | ||
| assert( | ||
| this.#socketPorts !== undefined && this.#runtimeEntryURL !== undefined, | ||
| "Cannot resolve address for fallback service before runtime is initialised" | ||
| ); | ||
|
|
||
| const port = this.#socketPorts.get( | ||
| getProxyFallbackServiceSocketName(service, entrypoint) | ||
| ); | ||
|
|
||
| if (!port) { | ||
| throw new Error( | ||
| `There is no socket opened for "${service}" with the "${entrypoint}" entrypoint` | ||
| ); | ||
| } | ||
|
|
||
| return { | ||
| httpStyle: "proxy", | ||
| protocol: getProtocol(this.#runtimeEntryURL), | ||
| host: this.#runtimeEntryURL.hostname, | ||
| port, | ||
| }; | ||
| } |
There was a problem hiding this comment.
We were looking up the fallback service address dynamically. But this won't work anymore when the proxy server is running on another thread. We are now pre-populating these addresses when the runtime is ready through this.#devRegistry.configureProxyWorker(url.toString(), this.#socketPorts).
| const serviceProxyTarget = extractServiceFetchProxyTarget(req); | ||
|
|
||
| if (serviceProxyTarget) { | ||
| assert(res !== undefined, "No response object provided"); | ||
|
|
||
| const address = | ||
| this.#devRegistry.getExternalServiceAddress( | ||
| serviceProxyTarget.service, | ||
| serviceProxyTarget.entrypoint | ||
| ) ?? | ||
| this.#getFallbackServiceAddress( | ||
| serviceProxyTarget.service, | ||
| serviceProxyTarget.entrypoint | ||
| ); | ||
|
|
||
| this.#handleProxy(req, res, address); | ||
| return; | ||
| } | ||
|
|
||
| const doProxyTarget = extractDoFetchProxyTarget(req); | ||
|
|
||
| if (doProxyTarget) { | ||
| assert(res !== undefined, "No response object provided"); | ||
|
|
||
| const address = this.#devRegistry.getExternalDurableObjectAddress( | ||
| doProxyTarget.scriptName, | ||
| doProxyTarget.className | ||
| ); | ||
|
|
||
| if (!address) { | ||
| res.writeHead(503); | ||
| res.end("Service Unavailable"); | ||
| return; | ||
| } | ||
|
|
||
| this.#handleProxy(req, res, address); | ||
| return; | ||
| } | ||
|
|
There was a problem hiding this comment.
This is moved to the proxy server's handleRequest method.
| #handleLoopbackConnect = async ( | ||
| req: http.IncomingMessage, | ||
| clientSocket: Duplex, | ||
| head: Buffer | ||
| ) => { | ||
| try { | ||
| const connectHost = req.url; | ||
| const [serviceName, entrypoint] = connectHost?.split(":") ?? []; | ||
| const address = | ||
| this.#devRegistry.getExternalServiceAddress(serviceName, entrypoint) ?? | ||
| this.#getFallbackServiceAddress(serviceName, entrypoint); | ||
|
|
||
| const serverSocket = net.connect(address.port, address.host, () => { | ||
| serverSocket.write(`CONNECT ${HOST_CAPNP_CONNECT} HTTP/1.1\r\n\r\n`); | ||
|
|
||
| // Push along any buffered bytes | ||
| if (head && head.length) { | ||
| serverSocket.write(head); | ||
| } | ||
|
|
||
| serverSocket.pipe(clientSocket); | ||
| clientSocket.pipe(serverSocket); | ||
| }); | ||
|
|
||
| // Errors on either side | ||
| serverSocket.on("error", (err) => { | ||
| this.#log.error(err); | ||
| clientSocket.end(); | ||
| }); | ||
| clientSocket.on("error", () => serverSocket.end()); | ||
|
|
||
| // Close the tunnel if the service is updated | ||
| // This make sure workerd will re-connect to the latest address | ||
| this.#devRegistry.subscribe(serviceName, () => { | ||
| this.#log.debug( | ||
| `Closing tunnel as service "${serviceName}" was updated` | ||
| ); | ||
| clientSocket.end(); | ||
| }); | ||
| } catch (ex: any) { | ||
| this.#log.error(ex); | ||
| clientSocket.end(); | ||
| } | ||
| }; |
There was a problem hiding this comment.
This is moved to the proxy server's handleConnect method.
| #handleProxy = ( | ||
| req: http.IncomingMessage, | ||
| res: http.ServerResponse, | ||
| target: { | ||
| protocol: "http" | "https"; | ||
| host: string; | ||
| port: number; | ||
| httpStyle?: "host" | "proxy"; | ||
| path?: string; | ||
| } | ||
| ) => { | ||
| const headers = { ...req.headers }; | ||
| let path = target.path; | ||
|
|
||
| if (!path) { | ||
| switch (target.httpStyle) { | ||
| case "host": { | ||
| const url = new URL(req.url ?? `http://${req.headers.host}`); | ||
| // If the target is a host, use the path from the request URL | ||
| path = url.pathname + url.search + url.hash; | ||
| headers.host = url.host; | ||
| break; | ||
| } | ||
| case "proxy": { | ||
| // If the target is a proxy, use the full request URL | ||
| path = req.url; | ||
| break; | ||
| } | ||
| } | ||
| } | ||
|
|
||
| const options: http.RequestOptions = { | ||
| host: target.host, | ||
| port: target.port, | ||
| method: req.method, | ||
| path, | ||
| headers, | ||
| }; | ||
|
|
||
| // Res is optional only on websocket upgrade requests | ||
| assert(res !== undefined, "No response object provided"); | ||
| const upstream = http.request(options, (upRes) => { | ||
| // Relay status and headers back to the original client | ||
| res.writeHead(upRes.statusCode ?? 500, upRes.headers); | ||
| // Pipe the response body | ||
| upRes.pipe(res); | ||
| }); | ||
|
|
||
| // Pipe the client request body to the upstream | ||
| req.pipe(upstream); | ||
|
|
||
| upstream.on("error", (err) => { | ||
| this.#log.error(err); | ||
| if (!res.headersSent) res.writeHead(502); | ||
| res.end("Bad Gateway"); | ||
| }); | ||
| }; |
There was a problem hiding this comment.
This is moved to the proxy server's handleProxy method.
| { | ||
| // There might be no HOST header when proxying a fetch request made over service binding | ||
| // e.g. env.MY_WORKER.fetch("https://example.com") | ||
| requireHostHeader: false, | ||
| }, |
There was a problem hiding this comment.
This was added only last week to address an issue with dev registry and shouldn't be needed in the loopback server anymore now that it is handled via a new proxy server.
| } | ||
| > | ||
| ): void { | ||
| if (services.size === 0 || !this.registryPath) { |
There was a problem hiding this comment.
We were relying on the subscribers count previously to decide whether there is any external services we were depending on. But we are now passing the full list over, so we can simply check the size of the map
| private handleRequest( | ||
| req: http.IncomingMessage, | ||
| res: http.ServerResponse | ||
| ): void { | ||
| const serviceProxyTarget = extractServiceFetchProxyTarget(req); | ||
| if (serviceProxyTarget) { | ||
| const address = this.getServiceAddress( | ||
| serviceProxyTarget.service, | ||
| serviceProxyTarget.entrypoint | ||
| ); | ||
|
|
||
| this.handleProxy(req, res, address); | ||
| return; | ||
| } | ||
|
|
||
| const doProxyTarget = extractDoFetchProxyTarget(req); | ||
| if (doProxyTarget) { | ||
| const address = this.getDurableObjectAddress( | ||
| doProxyTarget.scriptName, | ||
| doProxyTarget.className | ||
| ); | ||
|
|
||
| if (!address) { | ||
| res.writeHead(503); | ||
| res.end("Service Unavailable"); | ||
| return; | ||
| } | ||
|
|
||
| this.handleProxy(req, res, address); | ||
| return; | ||
| } | ||
|
|
||
| // If no valid target found, return 404 | ||
| res.writeHead(404); | ||
| res.end("Not Found"); | ||
| } |
There was a problem hiding this comment.
This was done in handleLoopback before moving to the proxy server.
| private async handleConnect( | ||
| req: http.IncomingMessage, | ||
| clientSocket: Duplex, | ||
| head: Buffer | ||
| ) { | ||
| try { | ||
| const connectHost = req.url; | ||
| const [serviceName, entrypoint] = connectHost?.split(":") ?? []; | ||
| const address = this.getServiceAddress(serviceName, entrypoint); | ||
| const serverSocket = net.connect(address.port, address.host, () => { | ||
| serverSocket.write(`CONNECT ${HOST_CAPNP_CONNECT} HTTP/1.1\r\n\r\n`); | ||
|
|
||
| // Push along any buffered bytes | ||
| if (head && head.length) { | ||
| serverSocket.write(head); | ||
| } | ||
|
|
||
| serverSocket.pipe(clientSocket); | ||
| clientSocket.pipe(serverSocket); | ||
| }); | ||
|
|
||
| // Errors on either side | ||
| serverSocket.on("error", (error) => { | ||
| log.error(error); | ||
| clientSocket.end(); | ||
| }); | ||
| clientSocket.on("error", () => serverSocket.end()); | ||
| // Close the tunnel if the service is updated | ||
| // This make sure workerd will re-connect to the latest address | ||
| this.subscribe(serviceName, () => { | ||
| log.debug(`Closing tunnel as service "${serviceName}" was updated`); | ||
| clientSocket.end(); | ||
| }); | ||
| } catch (e) { | ||
| log.error(e instanceof Error ? e : new Error(`${e}`)); | ||
| clientSocket.end(); | ||
| } | ||
| } |
There was a problem hiding this comment.
This was done in handleLoopbackConnect before moving to the proxy server.
| private handleProxy( | ||
| req: http.IncomingMessage, | ||
| res: http.ServerResponse, | ||
| target: ProxyAddress | ||
| ) { | ||
| const headers = { ...req.headers }; | ||
| let path = target.path; | ||
|
|
||
| if (!path) { | ||
| switch (target.httpStyle) { | ||
| case "host": { | ||
| const url = new URL(req.url ?? `http://${req.headers.host}`); | ||
| // If the target is a host, use the path from the request URL | ||
| path = url.pathname + url.search + url.hash; | ||
| headers.host = url.host; | ||
| break; | ||
| } | ||
| case "proxy": { | ||
| // If the target is a proxy, use the full request URL | ||
| path = req.url; | ||
| break; | ||
| } | ||
| } | ||
| } | ||
|
|
||
| const options: http.RequestOptions = { | ||
| host: target.host, | ||
| port: target.port, | ||
| method: req.method, | ||
| path, | ||
| headers, | ||
| }; | ||
|
|
||
| const upstream = http.request(options, (upRes) => { | ||
| // Relay status and headers back to the original client | ||
| res.writeHead(upRes.statusCode ?? 500, upRes.headers); | ||
| // Pipe the response body | ||
| upRes.pipe(res); | ||
| }); | ||
|
|
||
| // Pipe the client request body to the upstream | ||
| req.pipe(upstream); | ||
|
|
||
| upstream.on("error", () => { | ||
| if (!res.headersSent) res.writeHead(502); | ||
| res.end("Bad Gateway"); | ||
| }); | ||
| } |
petebacondarwin
left a comment
petebacondarwin
left a comment
There was a problem hiding this comment.
Review from Devin (manually copied over as it is not allowed to comment on PRs):
PR Review: Move address lookup logic to worker thread proxy
Summary
This PR addresses a deadlock issue in Miniflare when using mf.getBindings() with the dev registry enabled. The solution moves address lookup logic from the main loopback server to a new proxy server running in a worker thread, preventing the main thread from blocking while waiting for runtime responses.
✅ Positive Aspects
Clear Problem Definition & Solution
- The deadlock issue is well-documented in the changeset
- The worker thread approach is a solid architectural solution to prevent main thread blocking
- The PR maintains backward compatibility
Comprehensive Test Coverage
The test suite in dev-registry.spec.ts covers:
- Service worker and module worker scenarios
- RPC calls to default and custom entrypoints
- Node bindings integration
- Durable Object proxy functionality
- Graceful failure handling when remote workers are disposed
Clean Code Organization
- New
dev-registry.worker.tsis well-structured with clear separation of concerns - Message-based communication between main thread and worker thread is properly implemented
- Error handling includes proper cleanup and resource management
🔧 Technical Recommendations
1. Error Handling in ProxyServer.handleProxy()
Issue: In dev-registry.worker.ts line 286, the error handler doesn't log the error:
upstream.on("error", () => {
if (!res.headersSent) res.writeHead(502);
res.end("Bad Gateway");
});Recommendation: Add error logging for debugging:
upstream.on("error", (error) => {
log.error(error);
if (!res.headersSent) res.writeHead(502);
res.end("Bad Gateway");
});2. Registry Update Logic Efficiency
Issue: In updateRegistry() method (lines 322-344), the JSON.stringify comparison for detecting changes is inefficient for large objects.
Recommendation: Consider implementing a more efficient change detection mechanism, possibly using object hashing or deep comparison utilities.
3. Resource Cleanup
Issue: The subscribe() method in the worker thread accumulates callbacks but doesn't provide a way to unsubscribe, potentially leading to memory leaks.
Recommendation: Add an unsubscribe mechanism or implement automatic cleanup when services are removed.
4. Type Safety
Issue: The message handling in setupMessageHandlers() uses a switch statement without proper type guards.
Recommendation: Define proper TypeScript interfaces for the message types:
interface UpdateMessage {
type: "update";
workers: Record<string, WorkerDefinition>;
}
interface SetupMessage {
type: "setup";
runtimeEntryURL: string;
fallbackServicePorts: Record<string, Record<string, number>>;
}
type WorkerMessage = UpdateMessage | SetupMessage;5. Configuration Validation
Issue: The getFallbackServiceAddress() method throws errors when configuration is missing, but these errors might not be handled gracefully in all contexts.
Recommendation: Consider returning null/undefined and handling missing configuration more gracefully, especially during initialization phases.
🏗️ Architecture & Design
Worker Thread Communication
The message-passing architecture is well-designed, but consider:
- Adding message acknowledgments for critical operations
- Implementing a heartbeat mechanism to detect worker thread failures
- Adding proper shutdown coordination between main thread and worker
Proxy Address Resolution
The address resolution logic is comprehensive but complex. The fallback mechanism through multiple layers (registry → direct worker → fallback service) is robust but could benefit from:
- Better documentation of the resolution hierarchy
- Metrics/logging to understand which resolution path is taken
- Timeout handling for address resolution
🐛 Minor Issues
Code Comments
- Line 162 comment: "This make sure" should be "This makes sure"
- Consider adding more detailed JSDoc comments for the public methods
Constants
Consider extracting magic numbers like the 30-second heartbeat interval (line 258) and 5-minute cleanup threshold (line 299) to named constants.
Build Configuration
The build script changes in packages/miniflare/scripts/build.mjs look appropriate for including the new worker file in the build output.
🔒 Security Considerations
- The proxy server only listens on localhost (127.0.0.1), which is appropriate for local development
- No authentication is implemented, but this is acceptable for local dev registry use cases
- Consider adding rate limiting if this proxy could be exposed to untrusted code
⚡ Performance Implications
- Moving to a worker thread should improve main thread responsiveness
- The additional IPC overhead is minimal compared to the deadlock prevention benefit
- Registry file watching and updates are efficiently handled
📋 Overall Assessment
This is a well-implemented solution to a real problem. The architectural approach is sound, the code quality is good, and the test coverage is comprehensive. The main areas for improvement are around error handling, resource cleanup, and type safety. The PR successfully addresses the deadlock issue while maintaining the existing functionality.
Recommendation: ✅ Approve with minor improvements suggested above. The core solution is solid and the benefits outweigh the minor issues identified.
Co-authored-by: Pete Bacon Darwin <pete@bacondarwin.com>
3 participants
Fixes n/a.
This PR moves the address lookup logic from the loopback server to a new proxy server running in a worker thread to workaround a deadlock situation with the node bindings locking the main thread when waiting for a response from the runtime while the runtime is also waiting for a response from the loopback server on the address of the other workerd process. 😅