Address comments

Author: minggangw

lib/runtime/transports/http.js

@@ -13,8 +13,8 @@ const debug = require('debug')('rclnodejs:runtime:http');
 const { Connection } = require('../connection.js');
 const { TransportAdapter } = require('../transport_adapter.js');
 
-// Map dispatcher error codes to HTTP status codes. Anything not listed
-// here falls back to 500 — see the README for the full table.
+// Map dispatcher error codes to HTTP status codes. Codes not listed
+// here fall back to 500.
 const _STATUS_BY_CODE = Object.freeze({
   not_exposed: 404,
   not_implemented: 501,
@@ -64,9 +64,23 @@ class HttpRequestConnection extends Connection {
     this._payload = payload;
     this._replied = false;
     this._closed = false;
-    // Synthesised frame id so the dispatcher can correlate even though
-    // the HTTP path doesn't carry one on the wire.
+    // Placeholder frame id. The dispatcher reads `frame.id` from incoming
+    // frames and echoes it on the reply; over HTTP we discard the reply id
+    // and write the body directly, so the value just needs to be present.
     this._id = '__http__';
+
+    // NOTE: we deliberately do *not* wire `req.on('close')` /
+    // `res.on('close')` to `_emitCloseOnce()` here. Doing so would
+    // catch client-side disconnects, but the close event also fires
+    // on normal end-of-response, racing the still-in-flight rcl
+    // reply callback and tearing down the dispatcher's lazy Client
+    // mid-flight (rmw throws "client will not receive response" and
+    // crashes the process). For now, dispatcher cleanup runs from
+    // `send()` only — a long-running call that the client gave up
+    // on holds onto its Publisher/Client until `send()` completes.
+    // The leak is bounded (one entry per capability per request)
+    // and acceptable until we have a way to drive cleanup that
+    // doesn't race the rcl reply path.
   }
 
   /**
@@ -160,15 +174,16 @@ class HttpRequestConnection extends Connection {
 }
 
 /**
- * Layer-2 HTTP adapter for the Web Runtime.
+ * HTTP adapter for the Web Runtime.
  *
  * Exposes `call` and `publish` capabilities over plain HTTP:
  *
  *     POST /capability/call/<name>
  *     POST /capability/publish/<name>
  *
- * Subscriptions stay on the WebSocket transport (see WEB_RUNTIME_ROADMAP.md
- * "Why no SSE?"). Anything that isn't `POST /capability/call/<name>` or
+ * Subscriptions stay on the WebSocket transport (one HTTP connection
+ * per inbound message would burn the browser's per-origin budget).
+ * Anything that isn't `POST /capability/call/<name>` or
  * `POST /capability/publish/<name>` returns a 404 / 405 — the adapter
  * does not serve unrelated routes.
  *
@@ -195,7 +210,7 @@ class HttpTransport extends TransportAdapter {
     super();
     this.port = options.port != null ? options.port : 9001;
     this.host = options.host != null ? options.host : '::';
-    this.basePath = options.basePath || '/capability';
+    this.basePath = _normaliseBasePath(options.basePath);
     this.verifyRequest = options.verifyRequest;
     this._server = null;
     this._onConnection = null;
@@ -230,7 +245,15 @@ class HttpTransport extends TransportAdapter {
     if (!this._server) return;
     const server = this._server;
     this._server = null;
-    return new Promise((resolve) => server.close(() => resolve()));
+    return new Promise((resolve) => {
+      // Force-close keep-alive sockets so server.close() doesn't hang
+      // waiting for clients (notably Node's fetch/undici, which keeps
+      // sockets warm by default). Available since Node 18.2.
+      if (typeof server.closeAllConnections === 'function') {
+        server.closeAllConnections();
+      }
+      server.close(() => resolve());
+    });
   }
 
   // ---------- internals ----------
@@ -276,7 +299,7 @@ class HttpTransport extends TransportAdapter {
       });
     }
 
-    const tail = pathname.slice(this.basePath.length + 1); // strip "/capability/"
+    const tail = pathname.slice(this.basePath.length + 1); // strip basePath + '/'
     // tail = "<kind>/<rest...>"; first segment is the kind, rest is the
     // ROS name (which itself can contain slashes).
     const slash = tail.indexOf('/');
@@ -355,21 +378,24 @@ function _readJsonBody(req, maxBytes, cb) {
   };
 
   const ctype = (req.headers['content-type'] || '').toLowerCase();
+  // Only the media-type segment matters; ignore parameters like
+  // `; charset=utf-8` and reject sneaky values such as
+  // `text/plain;application/json` that include the right substring
+  // but mean something else.
+  const mediaType = ctype.split(';')[0].trim();
   if (
     req.method === 'POST' &&
-    !ctype.includes('application/json') &&
-    !ctype.startsWith('application/json')
+    ctype !== '' &&
+    mediaType !== 'application/json'
   ) {
     // Be lenient: empty bodies + missing content-type are treated as
     // "publish with no fields", same as `{}`. Only reject explicit
     // non-JSON content types.
-    if (ctype && ctype !== '') {
-      const e = new Error(
-        `unsupported content-type: ${ctype} (expected application/json)`
-      );
-      e.code = 'invalid_content_type';
-      return finish(e);
-    }
+    const e = new Error(
+      `unsupported content-type: ${ctype} (expected application/json)`
+    );
+    e.code = 'invalid_content_type';
+    return finish(e);
   }
   let total = 0;
   const chunks = [];
@@ -403,4 +429,22 @@ function _readJsonBody(req, maxBytes, cb) {
   });
 }
 
+function _normaliseBasePath(value) {
+  if (value === undefined || value === null || value === '') {
+    return '/capability';
+  }
+  if (typeof value !== 'string') {
+    throw new TypeError(
+      `HttpTransport: basePath must be a string, got ${typeof value}`
+    );
+  }
+  // Ensure single leading slash, no trailing slash.
+  let p = value.replace(/\/+$/, '');
+  if (!p.startsWith('/')) p = '/' + p;
+  if (p === '') {
+    throw new TypeError('HttpTransport: basePath must not be empty');
+  }
+  return p;
+}
+
 module.exports = { HttpTransport, HttpRequestConnection };

test/test-web-http.js

@@ -326,6 +326,85 @@ describe('rclnodejs/web — HTTP transport (call + publish)', function () {
         /ws must start with ws:\/\/ or wss:\/\//
       );
     });
+
+    it('call() before connect() rejects asynchronously, never throws sync', async function () {
+      // Regression: previously `(this._http || this._ws).call(...)` could
+      // throw a synchronous TypeError when constructed with ws:// and
+      // connect() hadn't run yet. Now call() is async and rejects.
+      const r = new RosClient('ws://127.0.0.1:1');
+      const p = r.call('/never_used', {});
+      assert.ok(typeof p.then === 'function', 'call() must return a Promise');
+      await assert.rejects(p);
+      await r.close();
+    });
+
+    it('stop() does not hang when keep-alive sockets are open', async function () {
+      // Open a keep-alive request and don't read the body, then stop().
+      // closeAllConnections() must let server.close() resolve promptly.
+      const t0 = Date.now();
+      const stopPromise = (async () => {
+        await fetch(httpBase + '/capability/call/http_add', {
+          method: 'POST',
+          headers: { 'content-type': 'application/json' },
+          body: '{}',
+        }).catch(() => undefined);
+      })();
+      await stopPromise;
+      // Build a one-off runtime so we can stop it cleanly here.
+      const tmp = createRuntime({
+        node,
+        transport: new HttpTransport({ port: 0, host: '127.0.0.1' }),
+      });
+      tmp.expose({ call: { '/x': 'std_srvs/srv/Empty' } });
+      await tmp.start();
+      const port = tmp.transports[0].port;
+      // Fire-and-forget a request; ignore its outcome. The keep-alive
+      // socket would normally hold server.close() open for ~5s.
+      fetch(`http://127.0.0.1:${port}/capability/call/x`, {
+        method: 'POST',
+        headers: { 'content-type': 'application/json' },
+        body: '{}',
+      }).catch(() => undefined);
+      await new Promise((r) => setTimeout(r, 50));
+      const t1 = Date.now();
+      await tmp.stop();
+      assert.ok(
+        Date.now() - t1 < 2000,
+        `stop() took ${Date.now() - t1}ms; expected fast shutdown`
+      );
+      assert.ok(Date.now() - t0 < 5000, 'whole test should finish under 5s');
+    });
+
+    it('normalises basePath: leading/trailing slash, missing leading', async function () {
+      const cases = [
+        { input: 'cap', expected: '/cap' },
+        { input: '/cap/', expected: '/cap' },
+        { input: '/cap///', expected: '/cap' },
+      ];
+      for (const { input, expected } of cases) {
+        const t = new HttpTransport({ basePath: input });
+        assert.strictEqual(
+          t.basePath,
+          expected,
+          `for ${JSON.stringify(input)}`
+        );
+      }
+      assert.throws(
+        () => new HttpTransport({ basePath: 42 }),
+        /basePath must be a string/
+      );
+    });
+
+    it('rejects sneaky content-type that includes application/json', async function () {
+      const res = await fetch(httpBase + '/capability/call/http_add', {
+        method: 'POST',
+        headers: { 'content-type': 'text/plain;application/json' },
+        body: '{}',
+      });
+      assert.strictEqual(res.status, 400);
+      const body = await res.json();
+      assert.strictEqual(body.code, 'invalid_content_type');
+    });
   });
 });
 

web/client.js

@@ -369,13 +369,13 @@ export class RosClient {
     this.url = httpUrl || wsUrl;
     this._http = httpUrl ? new _HttpLink(httpUrl) : null;
     this._wsUrl = wsUrl;
-    this._ws = null; // built lazily by _ensureWs() unless eager
-    // Eagerly open WS only when the user *explicitly* asked for it
-    // (passed `ws://...` or `{ws}`). When we *derived* a sibling WS
-    // URL from an HTTP base, it's lazy — most HTTP-only callers
-    // never subscribe.
+    // Eagerly construct (but don't yet open) the WS link when the user
+    // explicitly asked for it. When the WS URL was *derived* from an
+    // HTTP base, leave construction lazy — most HTTP-only callers
+    // never subscribe and never need the WS sibling at all.
+    this._ws = wsExplicit && wsUrl ? new _WsLink(wsUrl) : null;
     this._wsEager = !!wsExplicit;
-    this._wsConnect = null; // memoised connect promise
+    this._wsConnect = null; // memoised connect promise (in-flight or settled)
   }
 
   /** Open the link(s). */
@@ -392,8 +392,20 @@ export class RosClient {
   /** Close the underlying link(s). */
   async close() {
     const tasks = [];
-    if (this._ws) tasks.push(this._ws.close());
     if (this._http) tasks.push(this._http.close());
+    if (this._wsConnect) {
+      // WS is connecting or connected — wait for the open to settle,
+      // then close. Swallow the open error: nothing to close in that case.
+      tasks.push(
+        this._wsConnect.then(
+          (link) => link.close(),
+          () => undefined
+        )
+      );
+    } else if (this._ws) {
+      // Constructed eagerly but connect() never ran — no-op close.
+      tasks.push(this._ws.close());
+    }
     await Promise.all(tasks);
   }
 
@@ -410,36 +422,38 @@ export class RosClient {
         { code: 'transport_unavailable' }
       );
     }
-    if (this._ws) return this._ws;
-    if (!this._wsConnect) {
-      const link = new _WsLink(this._wsUrl);
-      this._wsConnect = link.connect().then(
-        () => {
-          this._ws = link;
-          return link;
-        },
-        (err) => {
-          this._wsConnect = null; // allow a retry on next subscribe()
-          throw Object.assign(
-            new Error(
-              `failed to open WebSocket sibling at ${this._wsUrl}: ${err && err.message ? err.message : String(err)}`
-            ),
-            { code: 'transport_unavailable', cause: err }
-          );
-        }
-      );
-    }
+    if (this._wsConnect) return this._wsConnect; // open is in-flight or done.
+    // Reuse the eagerly-constructed link if present, otherwise build it
+    // now (HTTP-derived sibling case).
+    if (!this._ws) this._ws = new _WsLink(this._wsUrl);
+    const link = this._ws;
+    this._wsConnect = link.connect().then(
+      () => link,
+      (err) => {
+        this._wsConnect = null; // allow a retry on next subscribe()
+        throw Object.assign(
+          new Error(
+            `failed to open WebSocket sibling at ${this._wsUrl}: ${err && err.message ? err.message : String(err)}`
+          ),
+          { code: 'transport_unavailable', cause: err }
+        );
+      }
+    );
     return this._wsConnect;
   }
 
   // -- verb API --
 
-  call(capability, payload) {
-    return (this._http || this._ws).call(capability, payload);
+  async call(capability, payload) {
+    if (this._http) return this._http.call(capability, payload);
+    const ws = await this._ensureWs();
+    return ws.call(capability, payload);
   }
 
-  publish(capability, payload) {
-    return (this._http || this._ws).publish(capability, payload);
+  async publish(capability, payload) {
+    if (this._http) return this._http.publish(capability, payload);
+    const ws = await this._ensureWs();
+    return ws.publish(capability, payload);
   }
 
   async subscribe(capability, callback) {