diff --git a/TODO.md b/TODO.md new file mode 100644 index 0000000..4566c4c --- /dev/null +++ b/TODO.md @@ -0,0 +1,34 @@ +# TODO + +## APRS-IS Connection + +- [ ] Consolidate `AprsIsConnection` and `Aprsme.Is` into a single implementation — two separate APRS-IS clients with different reconnection strategies and error handling is a maintenance burden +- [ ] Validate APRS-IS server login response — neither implementation checks if the server accepted the login (wrong credentials or invalid filter syntax goes undetected) +- [ ] Add APRS-IS login response validation — server sends `# logresp` line indicating accept/reject +- [ ] Cap circuit breaker half-open recovery — transition from `:open` to `:half_open` is passive (only happens on call), not automatic + +## Packet Intake Pipeline + +- [ ] Add retry logic for batch insert failures in `PacketConsumer.process_chunk/1` — when `Repo.insert_all` fails, packets are lost from real-time broadcast with no retry +- [ ] Improve `PacketProducer` buffer drop logging — `length(new_buffer)` is O(n), no count of dropped packets, no telemetry emitted +- [ ] Add backpressure mechanism — no global rate limiting if APRS-IS sends burst traffic; only defense is fixed-size buffer with silent drops +- [ ] Cluster packet distribution race — `PacketDistributor.distribute_packet/1` only broadcasts if currently leader; leadership change between receipt and broadcast drops packets + +## Front-End Display + +- [ ] Audit PopupComponent for XSS — verify HEEx auto-escaping covers `@comment` and weather data fields from APRS packets (`popup_component.ex`) +- [ ] Simplify coordinate extraction in `packets_live/index.html.heex:65-134` — deeply nested conditional logic with multiple fallback chains +- [ ] Fix memory leak in InfoMap hook — `setTimeout` at `info_map.js:110` reference never stored or canceled in `destroyed()` +- [ ] Fix Leaflet bundle loading race — `app.js:67-97` has two paths modifying `window.mapBundleLoaded` without singleton pattern +- [ ] Add loading indicator for real-time bounds updates — only `@historical_loading` triggers spinner, not bounds filtering +- [ ] Fix stale generation check bypass in `historical_loader.ex:100-108` — nil generation skips stale check +- [ ] Consolidate coordinate/bounds validation — `valid_coordinates?`, `within_bounds?`, `get_coordinates` duplicated across `coordinate_utils.ex`, `bounds_utils.ex`, `map_helpers.ex` +- [ ] Extract hard-coded zoom threshold (8) for heat map to a constant — duplicated in `display_manager.ex:17,33` + +## Packet Purging + +- [ ] Consider shorter default retention for APRS data — 365 days is very long for ephemeral APRS packets; most use cases need hours-to-days +- [ ] Run cleanup more frequently when backlog exists — 6-hour cycle with 5-minute time limit means large backlogs take days to clear +- [ ] Add cleanup telemetry — worker logs but doesn't emit telemetry events for monitoring dashboards +- [ ] Add partial index for cleanup queries — `WHERE received_at < cutoff` scans full B-tree; a partial index on old packets would be smaller and faster +- [ ] ETS PacketStore TTL mismatch — 2-hour TTL means expired packets get re-fetched from DB (still within 365-day retention), causing repeated queries diff --git a/lib/aprsme/aprs_is_connection.ex b/lib/aprsme/aprs_is_connection.ex index 34e0fd1..938b143 100644 --- a/lib/aprsme/aprs_is_connection.ex +++ b/lib/aprsme/aprs_is_connection.ex @@ -174,8 +174,15 @@ defmodule Aprsme.AprsIsConnection 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) - {:ok, socket} + + case :gen_tcp.send(socket, login) do + :ok -> + {:ok, socket} + + {:error, reason} -> + :gen_tcp.close(socket) + {:error, {:login_send_failed, reason}} + end error -> error diff --git a/lib/aprsme/is/is.ex b/lib/aprsme/is/is.ex index ed7dfff..77da059 100644 --- a/lib/aprsme/is/is.ex +++ b/lib/aprsme/is/is.ex @@ -99,6 +99,7 @@ defmodule Aprsme.Is do error -> Logger.warning("Failed to login to APRS-IS: #{inspect(error)}, will retry") + :gen_tcp.close(socket) schedule_reconnect(5000) {:ok, state} end @@ -289,8 +290,9 @@ defmodule Aprsme.Is do case state.socket do nil -> Logger.debug("Skipping keepalive - not connected to APRS-IS") - # Don't schedule another keepalive if we're not connected - {:noreply, state} + # Keep rescheduling so the timer stays alive for when we reconnect + keepalive_timer = create_keepalive_timer(@keepalive_interval) + {:noreply, %{state | keepalive_timer: keepalive_timer}} socket -> # Send a comment line as keepalive (APRS-IS standard) @@ -420,6 +422,7 @@ defmodule Aprsme.Is do error -> Logger.warning("Failed to login to APRS-IS: #{inspect(error)}, will retry in 10 seconds") + :gen_tcp.close(socket) schedule_reconnect(10_000) {:noreply, state} end diff --git a/lib/aprsme/packet_consumer.ex b/lib/aprsme/packet_consumer.ex index df53002..2521ea1 100644 --- a/lib/aprsme/packet_consumer.ex +++ b/lib/aprsme/packet_consumer.ex @@ -325,43 +325,35 @@ defmodule Aprsme.PacketConsumer do end end - # Broadcast packets asynchronously to avoid blocking + # Broadcast packets asynchronously using supervised task pool defp broadcast_packets_async(packets) do - Task.start(fn -> - try do - cluster_enabled = Application.get_env(:aprsme, :cluster_enabled, false) - - Enum.each(packets, fn packet_attrs -> - # Convert back to a format suitable for broadcasting - packet = %{ - sender: packet_attrs[:sender], - latitude: packet_attrs[:lat], - longitude: packet_attrs[:lon], - received_at: packet_attrs[:received_at], - data_type: packet_attrs[:data_type], - altitude: packet_attrs[:altitude], - speed: packet_attrs[:speed], - course: packet_attrs[:course], - comment: packet_attrs[:comment] - } - - if cluster_enabled do - # Use cluster distributor to broadcast to all nodes - PacketDistributor.distribute_packet(packet) - else - # Normal single-node broadcasting - Aprsme.StreamingPacketsPubSub.broadcast_packet(packet) - # Also broadcast to SpatialPubSub for viewport-based filtering - Aprsme.SpatialPubSub.broadcast_packet(packet) - end - end) - rescue - error -> - Logger.error("Failed to broadcast packets asynchronously: #{inspect(error)}") - end + Aprsme.BroadcastTaskSupervisor.async_execute(fn -> + cluster_enabled = Application.get_env(:aprsme, :cluster_enabled, false) + Enum.each(packets, &broadcast_single_packet(&1, cluster_enabled)) end) end + defp broadcast_single_packet(packet_attrs, cluster_enabled) do + packet = %{ + sender: packet_attrs[:sender], + latitude: packet_attrs[:lat], + longitude: packet_attrs[:lon], + received_at: packet_attrs[:received_at], + data_type: packet_attrs[:data_type], + altitude: packet_attrs[:altitude], + speed: packet_attrs[:speed], + course: packet_attrs[:course], + comment: packet_attrs[:comment] + } + + if cluster_enabled do + PacketDistributor.distribute_packet(packet) + else + Aprsme.StreamingPacketsPubSub.broadcast_packet(packet) + Aprsme.SpatialPubSub.broadcast_packet(packet) + end + end + defp prepare_packet_for_insert(packet_data) do # Always set received_at timestamp to ensure consistency current_time = DateTime.truncate(DateTime.utc_now(), :microsecond) diff --git a/lib/aprsme/spatial_pubsub.ex b/lib/aprsme/spatial_pubsub.ex index 3932470..204e1a2 100644 --- a/lib/aprsme/spatial_pubsub.ex +++ b/lib/aprsme/spatial_pubsub.ex @@ -320,7 +320,19 @@ defmodule Aprsme.SpatialPubSub do end defp point_in_bounds?(lat, lon, %{north: n, south: s, east: e, west: w}) do - lat >= s and lat <= n and lon >= w and lon <= e + lat_in_bounds = lat >= s and lat <= n + + # Handle longitude wrap-around at international date line + lon_in_bounds = + if w > e do + # Bounds cross the date line + lon >= w or lon <= e + else + # Normal bounds + lon >= w and lon <= e + end + + lat_in_bounds and lon_in_bounds end defp extract_location(packet) do diff --git a/lib/aprsme/streaming_packets_pubsub.ex b/lib/aprsme/streaming_packets_pubsub.ex index b0b663c..5265dd3 100644 --- a/lib/aprsme/streaming_packets_pubsub.ex +++ b/lib/aprsme/streaming_packets_pubsub.ex @@ -133,16 +133,6 @@ defmodule Aprsme.StreamingPacketsPubSub do {:noreply, state} end - @impl true - def handle_info({:cleanup_dead_subscribers, pids}, state) do - # Clean up dead subscribers from the GenServer process - Enum.each(pids, fn pid -> - :ets.delete(@table_name, pid) - end) - - {:noreply, state} - end - # Private functions defp valid_bounds?(%{north: n, south: s, east: e, west: w}) do @@ -184,25 +174,13 @@ defmodule Aprsme.StreamingPacketsPubSub do lat_in_bounds and lon_in_bounds end - defp send_to_matching_subscribers(subscribers, lat, lon, packet, server_pid) do - dead_pids = - subscribers - |> Stream.filter(fn {_pid, bounds} -> packet_in_bounds?(lat, lon, bounds) end) - |> Enum.reduce([], fn {pid, _bounds}, acc -> - send_to_subscriber_if_alive(pid, packet, acc) - end) - - # Send dead pids back to GenServer for cleanup - send(server_pid, {:cleanup_dead_subscribers, dead_pids}) - end - - defp send_to_subscriber_if_alive(pid, packet, acc) do - if Process.alive?(pid) do + defp send_to_matching_subscribers(subscribers, lat, lon, packet, _server_pid) do + # send/2 to a dead PID is a no-op in Erlang (does not crash). + # Process.monitor :DOWN messages handle cleanup, so no alive? check needed. + subscribers + |> Stream.filter(fn {_pid, bounds} -> packet_in_bounds?(lat, lon, bounds) end) + |> Enum.each(fn {pid, _bounds} -> send(pid, {:streaming_packet, packet}) - acc - else - # Collect dead pid for cleanup - [pid | acc] - end + end) end end diff --git a/lib/aprsme/workers/packet_cleanup_worker.ex b/lib/aprsme/workers/packet_cleanup_worker.ex index 85224ed..56df24b 100644 --- a/lib/aprsme/workers/packet_cleanup_worker.ex +++ b/lib/aprsme/workers/packet_cleanup_worker.ex @@ -20,6 +20,7 @@ defmodule Aprsme.Workers.PacketCleanupWorker do # Import modules needed for database operations import Ecto.Query + alias Aprsme.BadPacket alias Aprsme.Packet alias Aprsme.Repo @@ -61,10 +62,15 @@ defmodule Aprsme.Workers.PacketCleanupWorker do # Perform the cleanup of old packets (older than 1 year by default) using batch processing {deleted_count, _} = cleanup_old_packets_batched() + # Clean up old bad packets using the same retention period + bad_packet_count = cleanup_old_bad_packets() + # Log results retention_days = Application.get_env(:aprsme, :packet_retention_days, 365) - Logger.info("APRS packet cleanup complete: removed #{deleted_count} packets older than #{retention_days} days") + Logger.info( + "APRS packet cleanup complete: removed #{deleted_count} packets and #{bad_packet_count} bad packets older than #{retention_days} days" + ) # Return success :ok @@ -87,6 +93,20 @@ defmodule Aprsme.Workers.PacketCleanupWorker do cleanup_packets_older_than_batched(retention_days) end + defp cleanup_old_bad_packets do + retention_days = Application.get_env(:aprsme, :packet_retention_days, 365) + cutoff_time = DateTime.add(DateTime.utc_now(), -retention_days * 86_400, :second) + + {deleted_count, _} = + Repo.delete_all(from(b in BadPacket, where: b.attempted_at < ^cutoff_time)) + + deleted_count + rescue + error -> + Logger.error("Failed to clean up bad packets: #{inspect(error)}") + 0 + end + @doc """ Perform cleanup of packets older than a specific number of days using batch processing. diff --git a/test/aprsme/is_test.exs b/test/aprsme/is_test.exs index e018ab8..a86ae62 100644 --- a/test/aprsme/is_test.exs +++ b/test/aprsme/is_test.exs @@ -180,10 +180,13 @@ defmodule Aprsme.IsTest do end describe "handle_info(:send_keepalive, ...)" do - test "skips keepalive when socket is nil" do + test "reschedules keepalive when socket is nil" do state = build_state(%{socket: nil}) - assert {:noreply, ^state} = Aprsme.Is.handle_info(:send_keepalive, state) + assert {:noreply, new_state} = Aprsme.Is.handle_info(:send_keepalive, state) + assert new_state.socket == nil + # Timer should be rescheduled even when disconnected + assert new_state.keepalive_timer != state.keepalive_timer end end