|
| 1 | +# SPDX-License-Identifier: PMPL-1.0-or-later |
| 2 | +# Copyright (c) 2026 Jonathan D.A. Jewell (hyperpolymath) <j.d.a.jewell@open.ac.uk> |
| 3 | +defmodule BojRest.JsWorker do |
| 4 | + @moduledoc """ |
| 5 | + GenServer wrapping one persistent Deno pool-worker process. |
| 6 | +
|
| 7 | + Communication: newline-delimited JSON over the process's stdin/stdout. |
| 8 | + Requests are processed sequentially on the Deno side (one `await` at a time), |
| 9 | + but multiple callers can queue requests concurrently — replies are matched by |
| 10 | + the `id` field in each response. |
| 11 | +
|
| 12 | + When the underlying Deno process exits unexpectedly all pending callers |
| 13 | + receive `{:error, %{classification: :worker_crashed, ...}}` and the GenServer |
| 14 | + stops so the pool Supervisor can restart it with a fresh Deno process. |
| 15 | + """ |
| 16 | + use GenServer |
| 17 | + require Logger |
| 18 | + |
| 19 | + # Budget slightly above the JsInvoker timeout so the caller sees a clean |
| 20 | + # timeout error rather than a GenServer call timeout exception. |
| 21 | + @timeout_ms 30_000 |
| 22 | + @call_timeout_ms @timeout_ms + 5_000 |
| 23 | + |
| 24 | + defstruct [:port, :pending, :buffer, :id_counter] |
| 25 | + |
| 26 | + # ── public API ─────────────────────────────────────────────────────────────── |
| 27 | + |
| 28 | + def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: Keyword.get(opts, :name)) |
| 29 | + |
| 30 | + @doc """ |
| 31 | + Invoke `tool_name` on the cartridge at `mod_js_path` via the persistent worker. |
| 32 | +
|
| 33 | + `extra_env` is forwarded inside the JSON request and applied/cleared by the |
| 34 | + pool worker per invocation — safe because the worker processes requests one |
| 35 | + at a time. |
| 36 | + """ |
| 37 | + @spec invoke(GenServer.server(), String.t(), String.t(), map(), map()) :: |
| 38 | + {:ok, map()} | {:error, map()} |
| 39 | + def invoke(server, mod_js_path, tool_name, args, extra_env \\ %{}) do |
| 40 | + GenServer.call(server, {:invoke, mod_js_path, tool_name, args, extra_env}, @call_timeout_ms) |
| 41 | + end |
| 42 | + |
| 43 | + # ── GenServer callbacks ──────────────────────────────────────────────────── |
| 44 | + |
| 45 | + @impl true |
| 46 | + def init(opts) do |
| 47 | + deno = Keyword.fetch!(opts, :deno_path) |
| 48 | + runner = Keyword.fetch!(opts, :runner_path) |
| 49 | + |
| 50 | + port = |
| 51 | + Port.open({:spawn_executable, deno}, [ |
| 52 | + :binary, |
| 53 | + :exit_status, |
| 54 | + {:args, ["run", "--allow-net", "--allow-env", "--allow-read", runner]} |
| 55 | + ]) |
| 56 | + |
| 57 | + Logger.debug("JsWorker started", deno: deno, runner: runner) |
| 58 | + {:ok, %__MODULE__{port: port, pending: %{}, buffer: "", id_counter: 0}} |
| 59 | + end |
| 60 | + |
| 61 | + @impl true |
| 62 | + def handle_call({:invoke, mod_js_path, tool_name, args, extra_env}, from, state) do |
| 63 | + id = Integer.to_string(state.id_counter) |
| 64 | + |
| 65 | + request = |
| 66 | + Jason.encode!(%{ |
| 67 | + id: id, |
| 68 | + mod: mod_js_path, |
| 69 | + tool: tool_name, |
| 70 | + args: args, |
| 71 | + env: extra_env |
| 72 | + }) |
| 73 | + |
| 74 | + Port.command(state.port, request <> "\n") |
| 75 | + timer = Process.send_after(self(), {:timeout, id}, @timeout_ms) |
| 76 | + pending = Map.put(state.pending, id, {from, timer}) |
| 77 | + {:noreply, %{state | pending: pending, id_counter: state.id_counter + 1}} |
| 78 | + end |
| 79 | + |
| 80 | + @impl true |
| 81 | + def handle_info({port, {:data, data}}, %{port: port} = state) do |
| 82 | + full = state.buffer <> data |
| 83 | + {lines, rest} = split_lines(full) |
| 84 | + new_state = Enum.reduce(lines, state, &dispatch_response/2) |
| 85 | + {:noreply, %{new_state | buffer: rest}} |
| 86 | + end |
| 87 | + |
| 88 | + def handle_info({port, {:exit_status, code}}, %{port: port} = state) do |
| 89 | + Logger.error("Deno pool worker exited unexpectedly", exit_code: code) |
| 90 | + |
| 91 | + Enum.each(state.pending, fn {_id, {from, timer}} -> |
| 92 | + Process.cancel_timer(timer) |
| 93 | + GenServer.reply(from, {:error, %{classification: :worker_crashed, body: "Deno worker exited with code #{code}"}}) |
| 94 | + end) |
| 95 | + |
| 96 | + {:stop, :normal, %{state | pending: %{}}} |
| 97 | + end |
| 98 | + |
| 99 | + def handle_info({:timeout, id}, state) do |
| 100 | + case Map.pop(state.pending, id) do |
| 101 | + {{from, _timer}, pending} -> |
| 102 | + GenServer.reply(from, {:error, %{classification: :timeout, body: "JS invocation timed out after #{@timeout_ms}ms"}}) |
| 103 | + {:noreply, %{state | pending: pending}} |
| 104 | + |
| 105 | + {nil, _} -> |
| 106 | + {:noreply, state} |
| 107 | + end |
| 108 | + end |
| 109 | + |
| 110 | + # ── private helpers ──────────────────────────────────────────────────────── |
| 111 | + |
| 112 | + # Split accumulated bytes at newlines; return {complete_lines, remainder}. |
| 113 | + defp split_lines(data) do |
| 114 | + parts = String.split(data, "\n") |
| 115 | + {Enum.slice(parts, 0, length(parts) - 1) |> Enum.reject(&(&1 == "")), |
| 116 | + List.last(parts) || ""} |
| 117 | + end |
| 118 | + |
| 119 | + defp dispatch_response(line, state) do |
| 120 | + case Jason.decode(line) do |
| 121 | + {:ok, %{"id" => id, "status" => status, "data" => data}} -> |
| 122 | + case Map.pop(state.pending, id) do |
| 123 | + {{from, timer}, pending} -> |
| 124 | + Process.cancel_timer(timer) |
| 125 | + result = if status in 200..299, do: {:ok, data}, else: {:error, %{classification: :js_error, body: data, status: status}} |
| 126 | + GenServer.reply(from, result) |
| 127 | + %{state | pending: pending} |
| 128 | + |
| 129 | + {nil, _} -> |
| 130 | + Logger.warning("JsWorker: unexpected response id", id: id) |
| 131 | + state |
| 132 | + end |
| 133 | + |
| 134 | + _ -> |
| 135 | + Logger.warning("JsWorker: could not decode response line", line: line) |
| 136 | + state |
| 137 | + end |
| 138 | + end |
| 139 | +end |
0 commit comments