analysis_node.ex

defmodule AnalysisNode do
use GenServer

@backoff_ms 5000
@max_failures 10
@success_threshold_ms 30_000
@default_tick_ms 5000


def start_link(opts) do
GenServer.start_link(__MODULE__, opts, name: :analysis_node)
end

def init(_opts) do
settings_path = Path.join(__DIR__, "settings.txt")
content = read_file(settings_path)
settings = case Jason.decode(content) do
	{:ok, decoded} -> decoded
	{:error, _} -> %{}
end

homepath = Map.get(settings, "homepath", "/app/")
python_script = Path.join(__DIR__, "tools/analysis_service.py")
port_num = Map.get(settings, "analysis_port", 9510)
tick_ms = @default_tick_ms

{:ok, tref} = :timer.send_interval(tick_ms, :tick)

state = %{
	os_port: nil,
	spawn_time: nil,
	fail_count: 0,
	halted: false,
	homepath: homepath,
	python_script: python_script,
	port_num: port_num,
	timer_ref: tref,
	tick_ms: tick_ms
}

{:ok, spawn_port(state)}
end


# ===== PORT LIFECYCLE =====

defp spawn_port(state) do
python = System.find_executable("python3") || "python3"
port = Port.open({:spawn_executable, python}, [
	:binary, :use_stdio, {:line, 65536}, :exit_status,
	{:args, [state.python_script, state.homepath, Integer.to_string(state.port_num)]}
])
%{state | os_port: port, spawn_time: System.monotonic_time(:millisecond)}
end

defp handle_port_exit(state) do
age = System.monotonic_time(:millisecond) - (state.spawn_time || 0)
new_fail = if age >= @success_threshold_ms, do: 0, else: state.fail_count + 1

if new_fail >= @max_failures do
	IO.puts("AnalysisNode: halting respawns after " <> Integer.to_string(new_fail) <> " consecutive quick failures.")
	write_halt_flag(state.homepath, "analysis", state.python_script, new_fail)
	%{state | os_port: nil, fail_count: new_fail, halted: true}
else
	IO.puts("AnalysisNode: python exited after " <> Integer.to_string(age) <> "ms (fail_count=" <> Integer.to_string(new_fail) <> "). Respawning in " <> Integer.to_string(@backoff_ms) <> "ms.")
	Process.send_after(self(), :respawn, @backoff_ms)
	%{state | os_port: nil, fail_count: new_fail}
end
end


# ===== TICK =====

def handle_info(:tick, state) do
if state.os_port != nil and not state.halted and not File.exists?(state.homepath <> "work/.global_pause") do
	# Run the service round-trip off the GenServer so a slow or stalled python
	# response cannot block port lifecycle handling (exit_status / respawn).
	# Fire and forget: the result is not used by the tick.
	port_num = state.port_num
	Task.Supervisor.start_child(Ala.TaskSupervisor, fn ->
		send_to_service(port_num, %{"action" => "process_recent"})
	end)
end
flush_ticks()
{:noreply, state}
end


# ===== PORT MESSAGES =====

def handle_info({port, {:data, {:eol, _line}}}, state) when port == state.os_port do
{:noreply, state}
end

def handle_info({port, {:exit_status, _status}}, state) when port == state.os_port do
{:noreply, handle_port_exit(state)}
end

def handle_info({:EXIT, port, _reason}, state) when port == state.os_port do
{:noreply, handle_port_exit(state)}
end

def handle_info(:respawn, state) do
if state.halted do
	{:noreply, state}
else
	{:noreply, spawn_port(state)}
end
end

def handle_info(_msg, state) do
{:noreply, state}
end


# ===== NODE COMMUNICATION =====

def handle_cast({:core_message, _from, _data}, state) do
{:noreply, state}
end

def handle_cast({:master_message, _data}, state) do
{:noreply, state}
end


# ===== SERVICE COMMUNICATION =====

defp send_to_service(port_num, data) do
case :gen_tcp.connect({127, 0, 0, 1}, port_num, [:binary, active: false, packet: :line], 2000) do
	{:ok, sock} ->
		:gen_tcp.send(sock, Jason.encode!(data) <> "\n")
		result = case :gen_tcp.recv(sock, 0, 5000) do
			{:ok, line} -> Jason.decode(String.trim(line))
			{:error, _} -> {:error, :recv_failed}
		end
		:gen_tcp.close(sock)
		result
	{:error, _} -> {:error, :connect_failed}
end
end


# ===== UTILITY =====

# Surface a permanent halt to the UI/health layer via the flag directory,
# since the GenServer stays alive and the supervisor will not restart it.
defp write_halt_flag(homepath, service, script, fails) do
flags_dir = homepath <> "threads/flags/"
File.mkdir_p(flags_dir)
path = flags_dir <> "flag_halt_" <> service <> ".txt"
File.write(path, "type: system_observation\nsource: " <> service <> "\ntopic: service halted\n\n" <> service <> " halted after " <> Integer.to_string(fails) <> " quick failures. Check python3 " <> script <> " and restart the service.\n")
end

defp read_file(path) do
case File.read(path) do
	{:ok, content} -> content
	{:error, _} -> ""
end
end

defp flush_ticks do
receive do
	:tick -> flush_ticks()
after
	0 -> :ok
end
end

end