diff --git a/TODO.md b/TODO.md index e8be6aa..cd0f52d 100644 --- a/TODO.md +++ b/TODO.md @@ -2,23 +2,7 @@ ## Critical Bugs -### 1. LiveView: Missing `{:noreply, socket}` in drain handler — crashes LiveView -**File:** `lib/aprsme_web/live/map_live/index.ex:885-896` - -The `true` branch returns a bare socket instead of `{:noreply, socket}`, which will crash the LiveView with a `MatchError` when the drain condition is hit. - -```elixir -# BUG: true branch returns bare socket -def handle_info({:drain_connections, to_drain}, socket) do - if :rand.uniform(100) <= to_drain * 10 do - socket # <-- needs {:noreply, socket |> put_flash(...) |> push_event(...)} - |> put_flash(:info, "Server load balancing in progress. Reconnecting...") - |> push_event("reconnect", %{delay: :rand.uniform(5000)}) - else - {:noreply, socket} - end -end -``` +### ~~1. LiveView: Missing `{:noreply, socket}` in drain handler — crashes LiveView~~ FIXED ### ~~2. ConnectionMonitor: Deadlock via self-RPC every 30 seconds~~ FIXED **File:** `lib/aprsme/connection_monitor.ex:126-141` @@ -27,28 +11,12 @@ end **Fixed:** Now computes local stats directly from state instead of RPC to self. Remote nodes still use RPC. -### 3. LiveView: `get_callsign_key` generates random keys for Ecto structs — breaks all dedup -**File:** `lib/aprsme_web/live/shared/packet_utils.ex:13-14` +### ~~3. LiveView: `get_callsign_key` generates random keys for Ecto structs — breaks all dedup~~ FIXED +Now matches atom `:id`, string `"id"`, and falls back to `:sender`/`"sender"` before random key. -```elixir -def get_callsign_key(%{"id" => id}), do: to_string(id) -def get_callsign_key(_packet), do: [:positive] |> System.unique_integer() |> to_string() -``` +### ~~4. Cluster: `PacketDistributor` never calls `SpatialPubSub.broadcast_packet`~~ FIXED -Only matches string key `"id"`. Ecto structs with atom key `:id` fall through and get a random integer key every time. This means: -- Dedup never works (same packet gets different keys each call) -- `visible_packets` map accumulates duplicates -- Cleanup comparison (lines 1756-1766) never finds matches — all packets marked as expired - -### 4. Cluster: `PacketDistributor` never calls `SpatialPubSub.broadcast_packet` -**File:** `lib/aprsme/cluster/packet_distributor.ex:30-37` - -`handle_distributed_packet` calls `StreamingPacketsPubSub.broadcast_packet` but never calls `SpatialPubSub.broadcast_packet`. Any SpatialPubSub subscriber receives zero packets when clustering is enabled. - -### 5. LeaderElection: `pid_alive?/2` treats `{:badrpc, _}` as truthy -**File:** `lib/aprsme/cluster/leader_election.ex:257-263` - -`:rpc.call/4` returns `{:badrpc, reason}` on failure. Since tuples are truthy in Elixir, the function returns `true` when the RPC fails, preventing cleanup of stale leader registrations. +### ~~5. LeaderElection: `pid_alive?/2` treats `{:badrpc, _}` as truthy~~ FIXED ### ~~5b. LeaderElection: Loser never detects lost registration after `:global` conflict resolution~~ FIXED **File:** `lib/aprsme/cluster/leader_election.ex` @@ -57,10 +25,8 @@ When two pods start in isolation (before the Erlang cluster forms), both registe **Fixed:** Added `verify_leadership/1` which checks `:global.whereis_name` on every `check_leadership` tick. If the registration no longer points to `self()`, the node steps down and notifies `ConnectionManager` to stop the APRS-IS connection. -### 6. SpatialPubSub: `ensure_float/1` crashes on integer strings -**File:** `lib/aprsme/spatial_pubsub.ex:239` - -`String.to_float("42")` raises `ArgumentError`. If bounds arrive as integer strings, this crashes the singleton SpatialPubSub GenServer. +### ~~6. SpatialPubSub: `ensure_float/1` crashes on integer strings~~ FIXED +Now uses `Float.parse/1` which handles both `"42"` and `"42.5"`. --- @@ -87,10 +53,8 @@ These are never synchronized: Every non-weather marker on the map triggers an individual `Repo.exists?` query. With hundreds of visible markers, this is hundreds of DB round-trips per map load. Should batch-query weather callsigns upfront. -### 9. Duplicate packet transformation — every packet processed twice -**File:** `lib/aprsme/is/is.ex:596-604` and `lib/aprsme/packet_consumer.ex:382-430` - -`Is.dispatch/1` runs `struct_to_map`, `extract_additional_data`, and `normalize_data_type`. Then `PacketConsumer.prepare_packet_for_insert/1` runs them all again. Every packet does double work. +### ~~9. Duplicate packet transformation — every packet processed twice~~ FIXED +Removed `extract_additional_data` and `normalize_data_type` from `Is.dispatch/1` — `PacketConsumer` already handles these. ### 10. `batch_length/1` recomputes O(n) list length on every event arrival **File:** `lib/aprsme/packet_consumer.ex:64-68, 130` @@ -175,8 +139,8 @@ Each packet has 100+ columns. 2000 full structs per connected client. Never clea ## Dead Code -### 25. `handle_info({:ssl, ...})` in Is module — connection uses `:gen_tcp`, not SSL -**File:** `lib/aprsme/is/is.ex:318-352` +### ~~25. `handle_info({:ssl, ...})` in Is module — connection uses `:gen_tcp`, not SSL~~ FIXED +Removed dead SSL handler. Consolidated TCP handler into `handle_socket_data/2` with nil-safe timer cancel. ### 26. `PacketPipelineSetup` module — not in supervision tree, references unnamed consumers **File:** `lib/aprsme/packet_pipeline_setup.ex` @@ -211,8 +175,8 @@ Both `if cluster_enabled` and `else` return `{Phoenix.PubSub, name: Aprsme.PubSu Heat map never updates from real-time packets, historical loading at low zoom, or zoom threshold crossing. -### 35. `placeholders` option passed to `Repo.insert_all` — not a valid option -**Files:** `lib/aprsme/packet_consumer.ex:304`, `lib/aprsme/db_optimizer.ex:64` +### ~~35. `placeholders` option passed to `Repo.insert_all` — not a valid option~~ FIXED +Removed from both `packet_consumer.ex` and `db_optimizer.ex`. --- @@ -223,15 +187,11 @@ Heat map never updates from real-time packets, historical loading at low zoom, o `calculate_current_state` returns `:half_open` but never writes it back to state. Multiple callers can all enter half-open simultaneously. -### 37. `Process.cancel_timer(nil)` crash risk in Is module -**File:** `lib/aprsme/is/is.ex:321, 356` +### ~~37. `Process.cancel_timer(nil)` crash risk in Is module~~ FIXED +Consolidated into `handle_socket_data/2` with `if state.timer` guard. -If `state.timer` is nil, `Process.cancel_timer(nil)` raises `FunctionClauseError`. Some paths protect with `if state.timer`, but the main data handlers don't. - -### 38. `PacketPipelineSupervisor` uses `:one_for_one` — producer restart orphans consumers -**File:** `lib/aprsme/packet_pipeline_supervisor.ex:23` - -If the producer crashes, consumers remain subscribed to the dead PID. Should use `:rest_for_one`. +### ~~38. `PacketPipelineSupervisor` uses `:one_for_one` — producer restart orphans consumers~~ FIXED +Changed to `:rest_for_one` so consumer pool restarts when producer restarts. ### 39. `handle_update_time_display` doesn't trigger re-renders **File:** `lib/aprsme_web/live/map_live/index.ex:1774-1781` @@ -243,10 +203,8 @@ Returns `{:noreply, socket}` with unchanged socket — LiveView diff is empty. T Different LiveViews can overwrite each other's stored packets. TTL cleanup runs globally. -### 41. ConnectionMonitor memory calculation is inverted -**File:** `lib/aprsme/connection_monitor.ex:219-232` - -`total / system` produces a ratio >1.0 (typically 1.5-3.0), not a percentage. +### ~~41. ConnectionMonitor memory calculation is inverted~~ FIXED +Now computes `processes / total` — process memory as fraction of total VM memory. ### 42. `Callsign.valid?/1` accepts any non-empty string — docstring examples are wrong **File:** `lib/aprsme/callsign.ex:40-45` diff --git a/lib/aprsme/cluster/leader_election.ex b/lib/aprsme/cluster/leader_election.ex index 728a71e..44d990f 100644 --- a/lib/aprsme/cluster/leader_election.ex +++ b/lib/aprsme/cluster/leader_election.ex @@ -275,10 +275,9 @@ defmodule Aprsme.Cluster.LeaderElection do end defp pid_alive?(pid, pid_node) do - if :rpc.call(pid_node, Process, :alive?, [pid]) do - true - else - false + case :rpc.call(pid_node, Process, :alive?, [pid]) do + {:badrpc, _} -> false + result -> result == true end end diff --git a/lib/aprsme/cluster/packet_distributor.ex b/lib/aprsme/cluster/packet_distributor.ex index 6cc5044..c828689 100644 --- a/lib/aprsme/cluster/packet_distributor.ex +++ b/lib/aprsme/cluster/packet_distributor.ex @@ -28,8 +28,9 @@ defmodule Aprsme.Cluster.PacketDistributor do end def handle_distributed_packet({:distributed_packet, packet}) do - # Broadcast to local LiveView clients + # Broadcast to local LiveView clients via both PubSub systems Aprsme.StreamingPacketsPubSub.broadcast_packet(packet) + Aprsme.SpatialPubSub.broadcast_packet(packet) # Update packet store for LiveView PacketStore.store_packet(packet) diff --git a/lib/aprsme/connection_monitor.ex b/lib/aprsme/connection_monitor.ex index a837d1e..487e095 100644 --- a/lib/aprsme/connection_monitor.ex +++ b/lib/aprsme/connection_monitor.ex @@ -223,13 +223,13 @@ defmodule Aprsme.ConnectionMonitor do end defp get_memory_usage do - # Get memory usage as a percentage mem_data = :erlang.memory() total = Keyword.get(mem_data, :total, 0) - system = Keyword.get(mem_data, :system, 0) + # processes + system = total; report process memory as fraction of total + processes = Keyword.get(mem_data, :processes, 0) - if system > 0 do - total / system + if total > 0 do + processes / total else 0.0 end diff --git a/lib/aprsme/db_optimizer.ex b/lib/aprsme/db_optimizer.ex index 47b91c9..dfdb1d8 100644 --- a/lib/aprsme/db_optimizer.ex +++ b/lib/aprsme/db_optimizer.ex @@ -60,8 +60,7 @@ defmodule Aprsme.DbOptimizer do [ returning: false, on_conflict: :nothing, - timeout: 60_000, - placeholders: length(batch) > 100 + timeout: 60_000 ], opts ) diff --git a/lib/aprsme/is/is.ex b/lib/aprsme/is/is.ex index 2a5374c..b0b0868 100644 --- a/lib/aprsme/is/is.ex +++ b/lib/aprsme/is/is.ex @@ -315,75 +315,8 @@ defmodule Aprsme.Is do end @impl true - @spec handle_info({:ssl, port(), binary()} | any(), state()) :: {:noreply, state()} - def handle_info({:ssl, _socket, data}, state) do - # Cancel the previous timer - Process.cancel_timer(state.timer) - - # Update packet statistics - current_time = System.system_time(:second) - packet_stats = update_packet_stats(state.packet_stats, current_time) - - # Append new packet data to buffer - buffer = state.buffer <> data - - # Process complete lines (ending with \r\n or \n) - {complete_lines, remaining_buffer} = extract_complete_lines(buffer) - - # Dispatch each complete line - Enum.each(complete_lines, fn line -> - trimmed = String.trim(line) - - if trimmed != "" do - dispatch(trimmed) - end - end) - - # Start a new timer - timer = Process.send_after(self(), :aprsme_no_message_timeout, @aprs_timeout) - - state = - state - |> Map.put(:timer, timer) - |> Map.put(:packet_stats, packet_stats) - |> Map.put(:buffer, remaining_buffer) - - {:noreply, state} - end - def handle_info({:tcp, _socket, data}, state) do - # Cancel the previous timer - Process.cancel_timer(state.timer) - - # Update packet statistics - current_time = System.system_time(:second) - packet_stats = update_packet_stats(state.packet_stats, current_time) - - # Append new packet data to buffer - buffer = state.buffer <> data - - # Process complete lines (ending with \r\n or \n) - {complete_lines, remaining_buffer} = extract_complete_lines(buffer) - - # Dispatch each complete line - Enum.each(complete_lines, fn line -> - trimmed = String.trim(line) - - if trimmed != "" do - dispatch(trimmed) - end - end) - - # Start a new timer - timer = Process.send_after(self(), :aprsme_no_message_timeout, @aprs_timeout) - - state = - state - |> Map.put(:timer, timer) - |> Map.put(:packet_stats, packet_stats) - |> Map.put(:buffer, remaining_buffer) - - {:noreply, state} + handle_socket_data(data, state) end def handle_info({:backpressure, :activate}, %{socket: nil} = state) do @@ -507,6 +440,25 @@ defmodule Aprsme.Is do end end + defp handle_socket_data(data, state) do + if state.timer, do: Process.cancel_timer(state.timer) + + current_time = System.system_time(:second) + packet_stats = update_packet_stats(state.packet_stats, current_time) + + buffer = state.buffer <> data + {complete_lines, remaining_buffer} = extract_complete_lines(buffer) + + Enum.each(complete_lines, fn line -> + trimmed = String.trim(line) + if trimmed != "", do: dispatch(trimmed) + end) + + timer = Process.send_after(self(), :aprsme_no_message_timeout, @aprs_timeout) + + {:noreply, %{state | timer: timer, packet_stats: packet_stats, buffer: remaining_buffer}} + end + # Extract complete lines from buffer, returning {complete_lines, remaining_buffer} @spec extract_complete_lines(String.t()) :: {[String.t()], String.t()} defp extract_complete_lines(buffer) do @@ -590,20 +542,16 @@ defmodule Aprsme.Is do # Store the packet in the database for future replay # Use GenStage pipeline for efficient batch processing try do - # Always set received_at timestamp to ensure consistency + # Set received_at and convert struct to map — PacketConsumer.prepare_packet_for_insert + # handles extract_additional_data and normalize_data_type, so don't duplicate here. current_time = DateTime.truncate(DateTime.utc_now(), :microsecond) - packet_data = Map.put(parsed_message, :received_at, current_time) - # Convert to map before storing to avoid struct conversion issues - attrs = struct_to_map(packet_data) + attrs = + parsed_message + |> Map.put(:received_at, current_time) + |> Map.put(:raw, message) + |> struct_to_map() - # Extract additional data from the parsed packet including raw packet - attrs = Aprsme.Packet.extract_additional_data(attrs, message) - - # Normalize data_type to string if it's an atom - attrs = Aprsme.EncodingUtils.normalize_data_type(attrs) - - # Submit to GenStage pipeline for batch processing Aprsme.PacketProducer.submit_packet(attrs) rescue error -> diff --git a/lib/aprsme/packet_consumer.ex b/lib/aprsme/packet_consumer.ex index 3ec458c..a38768a 100644 --- a/lib/aprsme/packet_consumer.ex +++ b/lib/aprsme/packet_consumer.ex @@ -299,9 +299,7 @@ defmodule Aprsme.PacketConsumer do # Skip conflicts to avoid blocking on_conflict: :nothing, # Increased timeout for large batches - timeout: 60_000, - # Use placeholders for better performance with large batches - placeholders: length(valid_packets) > 100 + timeout: 60_000 ] try do diff --git a/lib/aprsme/packet_pipeline_supervisor.ex b/lib/aprsme/packet_pipeline_supervisor.ex index 826670a..c2aaa03 100644 --- a/lib/aprsme/packet_pipeline_supervisor.ex +++ b/lib/aprsme/packet_pipeline_supervisor.ex @@ -20,6 +20,6 @@ defmodule Aprsme.PacketPipelineSupervisor do children = [producer_spec, consumer_pool_spec] - Supervisor.init(children, strategy: :one_for_one) + Supervisor.init(children, strategy: :rest_for_one) end end diff --git a/lib/aprsme/spatial_pubsub.ex b/lib/aprsme/spatial_pubsub.ex index 204e1a2..6a9f400 100644 --- a/lib/aprsme/spatial_pubsub.ex +++ b/lib/aprsme/spatial_pubsub.ex @@ -236,7 +236,13 @@ defmodule Aprsme.SpatialPubSub do } end - defp ensure_float(val) when is_binary(val), do: String.to_float(val) + defp ensure_float(val) when is_binary(val) do + case Float.parse(val) do + {f, _} -> f + :error -> 0.0 + end + end + defp ensure_float(val) when is_integer(val), do: val * 1.0 defp ensure_float(val) when is_float(val), do: val defp ensure_float(_), do: 0.0 diff --git a/lib/aprsme_web/live/map_live/index.ex b/lib/aprsme_web/live/map_live/index.ex index f70327c..05b538f 100644 --- a/lib/aprsme_web/live/map_live/index.ex +++ b/lib/aprsme_web/live/map_live/index.ex @@ -883,13 +883,11 @@ defmodule AprsmeWeb.MapLive.Index do end def handle_info({:drain_connections, to_drain}, socket) do - # Check if this connection should be drained - # Use a random selection to determine if this connection should disconnect if :rand.uniform(100) <= to_drain * 10 do - # Gracefully disconnect this client - socket - |> put_flash(:info, "Server load balancing in progress. Reconnecting...") - |> push_event("reconnect", %{delay: :rand.uniform(5000)}) + {:noreply, + socket + |> put_flash(:info, "Server load balancing in progress. Reconnecting...") + |> push_event("reconnect", %{delay: :rand.uniform(5000)})} else {:noreply, socket} end diff --git a/lib/aprsme_web/live/shared/packet_utils.ex b/lib/aprsme_web/live/shared/packet_utils.ex index 2cf0d9e..6ea2505 100644 --- a/lib/aprsme_web/live/shared/packet_utils.ex +++ b/lib/aprsme_web/live/shared/packet_utils.ex @@ -10,7 +10,10 @@ defmodule AprsmeWeb.Live.Shared.PacketUtils do Get unique callsign key from packet. """ @spec get_callsign_key(map()) :: binary() - def get_callsign_key(%{"id" => id}), do: to_string(id) + def get_callsign_key(%{id: id}) when not is_nil(id), do: to_string(id) + def get_callsign_key(%{"id" => id}) when not is_nil(id), do: to_string(id) + def get_callsign_key(%{sender: sender}) when is_binary(sender), do: sender + def get_callsign_key(%{"sender" => sender}) when is_binary(sender), do: sender def get_callsign_key(_packet), do: [:positive] |> System.unique_integer() |> to_string() @doc """ diff --git a/test/aprsme/is_test.exs b/test/aprsme/is_test.exs index 0658138..1e77835 100644 --- a/test/aprsme/is_test.exs +++ b/test/aprsme/is_test.exs @@ -314,20 +314,7 @@ defmodule Aprsme.IsTest do end end - describe "handle_info({:ssl, ...}, ...)" do - test "processes SSL data the same as TCP" do - state = build_state() - - data = "# ssl comment\r\n" - - capture_log(fn -> - {:noreply, new_state} = Aprsme.Is.handle_info({:ssl, :fake_port, data}, state) - - assert new_state.buffer == "" - assert new_state.packet_stats.total_packets == 1 - end) - end - end + # SSL handler was removed — connection uses :gen_tcp, not SSL (TODO #25) describe "handle_info({:tcp_closed, ...}, ...)" do test "schedules reconnect and clears socket" do diff --git a/test/aprsme/packet_pipeline_supervisor_test.exs b/test/aprsme/packet_pipeline_supervisor_test.exs index 0ccc3fa..7c3ce7d 100644 --- a/test/aprsme/packet_pipeline_supervisor_test.exs +++ b/test/aprsme/packet_pipeline_supervisor_test.exs @@ -19,10 +19,10 @@ defmodule Aprsme.PacketPipelineSupervisorTest do :ok end - test "returns :one_for_one strategy" do + test "returns :rest_for_one strategy" do {:ok, {sup_flags, _children}} = PacketPipelineSupervisor.init([]) - assert %{strategy: :one_for_one} = sup_flags + assert %{strategy: :rest_for_one} = sup_flags end test "returns two child specs" do