Fix APRS-IS connection, packet pipeline, and cleanup reliability

- Handle login send failure in AprsIsConnection instead of crashing
- Close leaked sockets on login failure in Aprsme.Is (init + reconnect)
- Keep rescheduling keepalive timer when disconnected to prevent permanent loss
- Use supervised BroadcastTaskSupervisor instead of unsupervised Task.start
  for packet broadcasting
- Remove racy Process.alive? check in StreamingPacketsPubSub; send/2 to
  dead PIDs is a no-op, :DOWN monitors handle cleanup
- Add dateline wrapping to SpatialPubSub.point_in_bounds? matching
  StreamingPacketsPubSub behavior
- Add bad_packets table cleanup to PacketCleanupWorker
- Add TODO.md with remaining improvements

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

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

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

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)

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

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

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.
 

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