From 709780ff24495318305c1297287577e229c53d80 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Mon, 28 Jul 2025 08:35:12 -0500 Subject: [PATCH] Fix potential crashes and unsafe operations throughout the app MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Fixed File.read crash in devices_seeder.ex with proper error handling - Added timeout to TCP connect in aprs_is_connection.ex (10s timeout) - Fixed unsafe socket send operations with error handling - Added Process.alive? check before sending to PIDs in streaming_packets_pubsub.ex - Fixed stream materialization memory issue in packet_consumer.ex - Added error handling for datetime parsing errors - Prevented crashes from dead processes and closed sockets These fixes ensure the application is more resilient to: - Missing files - Network timeouts - Dead processes - Closed sockets - Invalid data parsing 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude --- lib/aprsme/aprs_is_connection.ex | 19 +++++++++++++++---- lib/aprsme/devices_seeder.ex | 20 ++++++++++++++++++-- lib/aprsme/packet_consumer.ex | 9 ++++----- lib/aprsme/streaming_packets_pubsub.ex | 10 +++++++++- 4 files changed, 46 insertions(+), 12 deletions(-) diff --git a/lib/aprsme/aprs_is_connection.ex b/lib/aprsme/aprs_is_connection.ex index a35ffdc..71b6f8e 100644 --- a/lib/aprsme/aprs_is_connection.ex +++ b/lib/aprsme/aprs_is_connection.ex @@ -114,8 +114,16 @@ defmodule Aprsme.AprsIsConnection do @impl true def handle_call({:send, packet}, _from, %{socket: socket} = state) when is_port(socket) do - :ok = :gen_tcp.send(socket, packet <> "\r\n") - {:reply, :ok, state} + # Safely send to socket with error handling + case :gen_tcp.send(socket, packet <> "\r\n") do + :ok -> + {:reply, :ok, state} + + {:error, reason} = error -> + Logger.error("Failed to send packet: #{inspect(reason)}") + # Socket is likely closed, reset state + {:reply, error, %{state | socket: nil}} + end end def handle_call({:send, _packet}, _from, state) do @@ -141,9 +149,12 @@ defmodule Aprsme.AprsIsConnection do passcode = Application.get_env(:aprsme, :aprsme_is_passcode, "00000") filter = Application.get_env(:aprsme, :aprsme_is_filter, "") - opts = [:binary, active: true, packet: :line, keepalive: true] + # Add timeout to prevent indefinite hanging + opts = [:binary, active: true, packet: :line, keepalive: true, send_timeout: 5000] + # 10 seconds + connect_timeout = 10_000 - case :gen_tcp.connect(host, port, opts) do + case :gen_tcp.connect(host, port, opts, connect_timeout) do {:ok, socket} -> login = "user #{callsign} pass #{passcode} vers aprs.me 0.1 #{filter}\r\n" :ok = :gen_tcp.send(socket, login) diff --git a/lib/aprsme/devices_seeder.ex b/lib/aprsme/devices_seeder.ex index 1397d11..01ba131 100644 --- a/lib/aprsme/devices_seeder.ex +++ b/lib/aprsme/devices_seeder.ex @@ -5,8 +5,22 @@ defmodule Aprsme.DevicesSeeder do alias Aprsme.Repo def seed_from_json(path \\ "test/support/test_devices.json") do - {:ok, body} = File.read(path) - {:ok, json} = Jason.decode(body) + case File.read(path) do + {:ok, body} -> + case Jason.decode(body) do + {:ok, json} -> + seed_from_decoded_json(json) + + {:error, reason} -> + {:error, "Failed to decode JSON: #{inspect(reason)}"} + end + + {:error, reason} -> + {:error, "Failed to read file #{path}: #{inspect(reason)}"} + end + end + + defp seed_from_decoded_json(json) do tocalls = Map.get(json, "tocalls", %{}) mice = Map.get(json, "mice", %{}) micelegacy = Map.get(json, "micelegacy", %{}) @@ -17,6 +31,8 @@ defmodule Aprsme.DevicesSeeder do Enum.each([tocalls, mice, micelegacy], fn group -> seed_device_group(group, now) end) + + {:ok, :seeded} end defp seed_device_group(group, now) do diff --git a/lib/aprsme/packet_consumer.ex b/lib/aprsme/packet_consumer.ex index ffb7c42..d191111 100644 --- a/lib/aprsme/packet_consumer.ex +++ b/lib/aprsme/packet_consumer.ex @@ -127,14 +127,12 @@ defmodule Aprsme.PacketConsumer do chunk_size = Application.get_env(:aprsme, :packet_pipeline)[:batch_size] || 100 # Use Stream for memory-efficient processing - results = + # Process and reduce in one pass to avoid materializing the entire list + {success_count, error_count} = packets |> Stream.chunk_every(chunk_size) |> Stream.map(&process_chunk/1) - |> Enum.to_list() - - {success_count, error_count} = - Enum.reduce(results, {0, 0}, fn {success, error}, {total_success, total_error} -> + |> Enum.reduce({0, 0}, fn {success, error}, {total_success, total_error} -> {total_success + success, total_error + error} end) @@ -511,6 +509,7 @@ defmodule Aprsme.PacketConsumer do defp truncate_datetimes_to_second(%DateTime{} = dt), do: DateTime.truncate(dt, :second) defp truncate_datetimes_to_second({:ok, %DateTime{} = dt}), do: DateTime.truncate(dt, :second) + defp truncate_datetimes_to_second({:error, _reason}), do: nil defp truncate_datetimes_to_second(term) when is_map(term) and not is_struct(term) do Map.new(term, fn {k, v} -> {k, truncate_datetimes_to_second(v)} end) diff --git a/lib/aprsme/streaming_packets_pubsub.ex b/lib/aprsme/streaming_packets_pubsub.ex index f05e5d5..9b283b7 100644 --- a/lib/aprsme/streaming_packets_pubsub.ex +++ b/lib/aprsme/streaming_packets_pubsub.ex @@ -118,7 +118,15 @@ defmodule Aprsme.StreamingPacketsPubSub do Aprsme.BroadcastTaskSupervisor.async_execute(fn -> subscribers |> Stream.filter(fn {_pid, bounds} -> packet_in_bounds?(lat, lon, bounds) end) - |> Enum.each(fn {pid, _bounds} -> send(pid, {:streaming_packet, packet}) end) + |> Enum.each(fn {pid, _bounds} -> + # Only send if process is alive + if Process.alive?(pid) do + send(pid, {:streaming_packet, packet}) + else + # Clean up dead subscriber + :ets.delete(@table_name, pid) + end + end) end) end