Merge pull request #105 from Jump-App/request-response-uid

Fix response cross-contamination via request-response correlation UIDs

Author: dweill

lib/nodejs/worker.ex

@@ -60,7 +60,7 @@ defmodule NodeJS.Worker do
         ]
       )
 
-    {:ok, [node_service_path(), port]}
+    {:ok, %{service_path: node_service_path(), port: port, uid_counter: 0}}
   end
 
   defp get_env_vars(module_path) do
@@ -70,29 +70,48 @@ defmodule NodeJS.Worker do
     ]
   end
 
-  defp get_response(data, timeout) do
+  defp get_response(data, timeout, port, expected_uid) do
     receive do
-      {_port, {:data, {flag, chunk}}} ->
+      {^port, {:data, {flag, chunk}}} ->
         data = data ++ chunk
 
         case flag do
           :noeol ->
-            get_response(data, timeout)
+            get_response(data, timeout, port, expected_uid)
 
           :eol ->
             case data do
-              @prefix ++ protocol_data -> {:ok, protocol_data}
-              _ -> get_response(~c"", timeout)
+              @prefix ++ protocol_data ->
+                case extract_uid(protocol_data) do
+                  {^expected_uid, response_data} ->
+                    {:ok, response_data}
+
+                  {_stale_uid, _response_data} ->
+                    # Response from a different (likely timed-out) request — discard it
+                    get_response(~c"", timeout, port, expected_uid)
+                end
+
+              _ ->
+                get_response(~c"", timeout, port, expected_uid)
             end
         end
 
-      {_port, {:exit_status, status}} when status != 0 ->
+      {^port, {:exit_status, status}} when status != 0 ->
         {:error, {:exit, status}}
     after
       timeout -> {:error, :timeout}
     end
   end
 
+  # Extracts the UID and response data from protocol data.
+  # Protocol format: "uid:json_response"
+  defp extract_uid(data) do
+    case Enum.split_while(data, &(&1 != ?:)) do
+      {uid_chars, [?: | rest]} -> {List.to_string(uid_chars), rest}
+      _ -> {"", data}
+    end
+  end
+
   defp decode_binary(data, binary) do
     if binary === true do
       :binary.list_to_bin(data)
@@ -102,15 +121,18 @@ defmodule NodeJS.Worker do
   end
 
   @doc false
-  def handle_call({module, args, opts}, _from, [_, port] = state)
+  def handle_call({module, args, opts}, _from, %{port: port, uid_counter: uid_counter} = state)
       when is_tuple(module) do
     timeout = Keyword.get(opts, :timeout)
     binary = Keyword.get(opts, :binary)
     esm = Keyword.get(opts, :esm, false)
-    body = Jason.encode!([Tuple.to_list(module), args, esm])
+    uid = Integer.to_string(uid_counter)
+    body = Jason.encode!([uid, Tuple.to_list(module), args, esm])
     Port.command(port, "#{body}\n")
 
-    case get_response(~c"", timeout) do
+    state = %{state | uid_counter: uid_counter + 1}
+
+    case get_response(~c"", timeout, port, uid) do
       {:ok, response} ->
         decoded_response =
           response
@@ -169,7 +191,7 @@ defmodule NodeJS.Worker do
   end
 
   @doc false
-  def terminate(_reason, [_, port]) do
+  def terminate(_reason, %{port: port}) do
     reset_terminal(port)
     send(port, {self(), :close})
   end

priv/server.js

@@ -24,15 +24,15 @@ async function importModuleRespectingNodePath(modulePath) {
   for(const nodePath of NODE_PATHS) {
     // Try to resolve the module in the current path
     const modulePathToTry = path.join(nodePath, modulePath)
-    if (fileExists(modulePathToTry)) {
+    if (await fileExists(modulePathToTry)) {
       // imports are cached. To bust that cache, add unique query string to module name
       // eg NodeJS.call({"esm-module.mjs?q=#{System.unique_integer()}", :fn})
-      // it will leak memory, so I'm not doing it by default! 
+      // it will leak memory, so I'm not doing it by default!
       // see more: https://ar.al/2021/02/22/cache-busting-in-node.js-dynamic-esm-imports/#cache-invalidation-in-esm-with-dynamic-imports
       return await import(modulePathToTry)
     }
   }
-  
+
   throw new Error(`Could not find module '${modulePath}'. Hint: File extensions are required in ESM. Tried ${NODE_PATHS.join(", ")}`)
 }
 
@@ -45,22 +45,28 @@ function getAncestor(parent, [key, ...keys]) {
 }
 
 async function getResponse(string) {
+  let uid = ""
   try {
-    const [[modulePath, ...keys], args, useImport] = JSON.parse(string)
+    const parsed = JSON.parse(string)
+    uid = parsed[0]
+    const [[modulePath, ...keys], args, useImport] = parsed.slice(1)
     const importFn = useImport ? importModuleRespectingNodePath : requireModule
-    const mod = await importFn(modulePath) 
+    const mod = await importFn(modulePath)
     const fn = await getAncestor(mod, keys)
     if (!fn) throw new Error(`Could not find function '${keys.join(".")}' in module '${modulePath}'`)
     const returnValue = fn(...args)
     const result = returnValue instanceof Promise ? await returnValue : returnValue
-    return JSON.stringify([true, result])
-  } catch ({ message, stack }) {
-    return JSON.stringify([false, `${message}\n${stack}`])
+    return { uid, data: JSON.stringify([true, result]) }
+  } catch (err) {
+    const message = err?.message ?? String.valueOf(err)
+    const stack = err?.stack ?? ""
+    return { uid, data: JSON.stringify([false, `${message}\n${stack}`]) }
   }
 }
 
 async function onLine(string) {
-  const buffer = Buffer.from(`${await getResponse(string)}\n`)
+  const { uid, data } = await getResponse(string)
+  const buffer = Buffer.from(`${uid}:${data}\n`)
 
   // The function we called might have written something to stdout without starting a new line.
   // So we add one here and write the response after the prefix

test/nodejs_test.exs

@@ -174,6 +174,39 @@ defmodule NodeJS.Test do
     end
   end
 
+  describe "request-response correlation" do
+    test "stale responses from timed-out calls never leak into subsequent calls" do
+      # Use a dedicated single-worker pool so all calls go to the same worker.
+      # This guarantees the stale response lands in the same GenServer mailbox
+      # that the follow-up call is waiting on.
+      path = __ENV__.file |> Path.dirname() |> Path.join("js")
+
+      start_supervised!(
+        Supervisor.child_spec(
+          {NodeJS.Supervisor, path: path, name: NodeJS.RaceTest, pool_size: 1},
+          id: NodeJS.RaceTest
+        )
+      )
+
+      # Send a call that takes 300ms but times out after 50ms.
+      # After timeout, the worker is free but the Node.js response is still pending.
+      assert {:error, "Call timed out."} =
+               NodeJS.call("slow-async-echo", [9999, 300],
+                 timeout: 50,
+                 name: NodeJS.RaceTest
+               )
+
+      # Immediately send a follow-up call that takes 500ms.
+      # Its `receive` window overlaps with the stale response arriving at ~300ms.
+      # Without UID correlation, `receive` would pick up the stale `9999` response.
+      assert {:ok, 42} =
+               NodeJS.call("slow-async-echo", [42, 500],
+                 timeout: 5_000,
+                 name: NodeJS.RaceTest
+               )
+    end
+  end
+
   describe "overriding call timeout" do
     test "works, and you can tell because the slow function will time out" do
       assert {:error, "Call timed out."} = NodeJS.call("slow-async-echo", [1111], timeout: 0)
@@ -243,12 +276,12 @@ defmodule NodeJS.Test do
 
     test "fails if extension is not specified" do
       assert {:error, msg} = NodeJS.call({"esm-module", :hello}, ["me"], esm: true)
-      assert js_error_message(msg) =~ "Cannot find module"
+      assert msg =~ "find module"
     end
 
     test "fails if file not found" do
       assert {:error, msg} = NodeJS.call({"nonexisting.js", :hello}, [], esm: true)
-      assert js_error_message(msg) =~ "Cannot find module"
+      assert msg =~ "find module"
     end
 
     test "fails if file has errors" do