diff --git a/TODO.md b/TODO.md index d3b2ab2..0a4a746 100644 --- a/TODO.md +++ b/TODO.md @@ -1,41 +1,12 @@ # Code Review Findings -## Dual Packet State System (Architectural Bug) - -### 7. LiveView maintains two unsynchronized packet tracking systems -**Files:** `lib/aprsme_web/live/map_live/index.ex`, `packet_processor.ex`, `display_manager.ex` - -**System A:** `packet_state` wrapping `PacketManager`/`PacketStore` (ETS-backed). -**System B:** `all_packets`, `visible_packets`, `historical_packets` as plain maps in assigns. - -These are never synchronized: -- Real-time packets update System B but never touch System A -- Cleanup runs on System A but never touches System B -- Bounds filtering reads System A but display reads System B -- `PacketManager.extract_packet_ids` uses different ID generation than `PacketStore.generate_packet_id` — stored packets can never be found through the manager API - ---- - -## Memory Issues - -### 23. `all_packets` map in LiveView assigns holds 2000 full Ecto structs -**File:** `lib/aprsme_web/live/map_live/packet_processor.ex:27-37` - -Each packet has 100+ columns. 2000 full structs per connected client. Never cleaned by `handle_cleanup_old_packets` (which only touches `packet_state`). Appears to be dead state — written to but never read. - ---- - -## Correctness Issues - -### 40. `PacketStore` uses global named ETS table shared across all LiveViews -**File:** `lib/aprsme_web/live/map_live/packet_store.ex:12, 114` - -Different LiveViews can overwrite each other's stored packets. TTL cleanup runs globally. - ---- +All 50 items addressed. ## Completed +### 7. Dual packet state system — deleted broken PacketManager/PacketStore, unified on visible_packets +### 23. `all_packets` dead state — removed write-only map from LiveView assigns +### 40. `PacketStore` global ETS table — deleted along with PacketManager ### 1. LiveView: Missing `{:noreply, socket}` in drain handler — crashes LiveView ### 2. ConnectionMonitor: Deadlock via self-RPC every 30 seconds ### 3. LiveView: `get_callsign_key` generates random keys for Ecto structs — breaks all dedup diff --git a/lib/aprsme/application.ex b/lib/aprsme/application.ex index 42f0ff1..2a98932 100644 --- a/lib/aprsme/application.ex +++ b/lib/aprsme/application.ex @@ -35,9 +35,6 @@ defmodule Aprsme.Application do Aprsme.SpatialPubSub, # Start global streaming packets PubSub Aprsme.StreamingPacketsPubSub, - # Start packet store for efficient LiveView memory usage - AprsmeWeb.MapLive.PacketStore, - # Start the Endpoint (http/https) AprsmeWeb.Endpoint, # Start a worker by calling: Aprsme.Worker.start_link(arg) diff --git a/lib/aprsme/cluster/packet_distributor.ex b/lib/aprsme/cluster/packet_distributor.ex index c828689..fff4ffe 100644 --- a/lib/aprsme/cluster/packet_distributor.ex +++ b/lib/aprsme/cluster/packet_distributor.ex @@ -5,7 +5,6 @@ defmodule Aprsme.Cluster.PacketDistributor do only the leader maintains the APRS-IS connection. """ alias Aprsme.Cluster.LeaderElection - alias AprsmeWeb.MapLive.PacketStore @pubsub_topic "cluster:packets" @@ -32,9 +31,6 @@ defmodule Aprsme.Cluster.PacketDistributor do Aprsme.StreamingPacketsPubSub.broadcast_packet(packet) Aprsme.SpatialPubSub.broadcast_packet(packet) - # Update packet store for LiveView - PacketStore.store_packet(packet) - :ok end end diff --git a/lib/aprsme_web/live/map_live/index.ex b/lib/aprsme_web/live/map_live/index.ex index 5813cf4..0316029 100644 --- a/lib/aprsme_web/live/map_live/index.ex +++ b/lib/aprsme_web/live/map_live/index.ex @@ -24,7 +24,6 @@ defmodule AprsmeWeb.MapLive.Index do alias AprsmeWeb.MapLive.HistoricalLoader alias AprsmeWeb.MapLive.Navigation alias AprsmeWeb.MapLive.PacketBatcher - alias AprsmeWeb.MapLive.PacketManager alias AprsmeWeb.MapLive.PacketProcessor alias AprsmeWeb.MapLive.RfPath alias AprsmeWeb.MapLive.UrlParams @@ -211,7 +210,6 @@ defmodule AprsmeWeb.MapLive.Index do map_center: final_map_center, map_zoom: final_map_zoom, should_skip_initial_url_update: should_skip_initial_url_update, - packet_state: PacketManager.init_packet_state(), overlay_callsign: "", tracked_callsign: tracked_callsign, tracked_callsign_latest_packet: tracked_callsign_latest_packet, @@ -262,7 +260,6 @@ defmodule AprsmeWeb.MapLive.Index do visible_packets: %{}, historical_packets: %{}, page_title: "APRS Map", - packet_state: PacketManager.init_packet_state(), station_popup_open: false, map_bounds: %{ north: 49.0, @@ -2013,14 +2010,10 @@ defmodule AprsmeWeb.MapLive.Index do # Always filter markers by bounds socket = push_event(socket, "filter_markers_by_bounds", %{bounds: map_bounds}) - # Update packet state to keep only packets within bounds - filtered_packets = Map.values(new_visible_packets) - {updated_packet_state, _} = PacketManager.add_visible_packets(socket.assigns.packet_state, filtered_packets) - - # Update map bounds FIRST so progressive loading uses the correct bounds + # Update map bounds and visible_packets with bounds-filtered result socket = socket - |> assign(map_bounds: map_bounds, packet_state: updated_packet_state) + |> assign(map_bounds: map_bounds, visible_packets: new_visible_packets) |> assign(needs_initial_historical_load: false) # Load historical packets for the new bounds (now socket.assigns.map_bounds is correct) diff --git a/lib/aprsme_web/live/map_live/packet_manager.ex b/lib/aprsme_web/live/map_live/packet_manager.ex deleted file mode 100644 index 11e2fad..0000000 --- a/lib/aprsme_web/live/map_live/packet_manager.ex +++ /dev/null @@ -1,180 +0,0 @@ -defmodule AprsmeWeb.MapLive.PacketManager do - @moduledoc """ - Optimized packet management for LiveView with reduced memory usage. - - This module provides efficient packet storage and retrieval optimized for - LiveView performance by: - - Storing minimal data in LiveView assigns - - Using sliding window for packet management - - Providing just-in-time data fetching - - Batching operations for better performance - """ - - alias AprsmeWeb.MapLive.PacketStore - - # Maximum packets to keep in memory per category - @max_visible_packets 1000 - @max_historical_packets 2000 - # Clean up when over limit by this amount - @packet_cleanup_threshold 50 - - @doc """ - Initialize packet management state for a socket. - Returns minimal state optimized for memory usage. - """ - def init_packet_state do - %{ - visible_packet_ids: [], - historical_packet_ids: [], - packet_count: %{visible: 0, historical: 0}, - last_cleanup: System.monotonic_time(:millisecond) - } - end - - @doc """ - Add packets to visible storage with memory management. - Returns updated state and packet IDs for broadcasting. - """ - def add_visible_packets(%{visible_packet_ids: visible_ids, packet_count: packet_count} = packet_state, packets) - when is_list(packets) do - packet_ids = PacketStore.store_packets(packets) - new_visible_ids = packet_ids ++ visible_ids - - {managed_ids, cleanup_needed} = apply_memory_limits(new_visible_ids, @max_visible_packets) - - updated_state = %{ - packet_state - | visible_packet_ids: managed_ids, - packet_count: Map.put(packet_count, :visible, length(managed_ids)) - } - - final_state = maybe_cleanup(updated_state, cleanup_needed, :visible) - {final_state, packet_ids} - end - - @doc """ - Add packets to historical storage with sliding window management. - """ - def add_historical_packets( - %{historical_packet_ids: historical_ids, packet_count: packet_count} = packet_state, - packets - ) - when is_list(packets) do - packet_ids = PacketStore.store_packets(packets) - new_historical_ids = packet_ids ++ historical_ids - - {managed_ids, cleanup_needed} = apply_memory_limits(new_historical_ids, @max_historical_packets) - - updated_state = %{ - packet_state - | historical_packet_ids: managed_ids, - packet_count: Map.put(packet_count, :historical, length(managed_ids)) - } - - final_state = maybe_cleanup(updated_state, cleanup_needed, :historical) - {final_state, packet_ids} - end - - @doc """ - Get visible packets for rendering. Only fetches data when needed. - """ - def get_visible_packets(packet_state, limit \\ @max_visible_packets) do - packet_ids = Enum.take(packet_state.visible_packet_ids, limit) - packet_ids |> PacketStore.get_packets() |> Map.values() - end - - @doc """ - Get historical packets for rendering. - """ - def get_historical_packets(packet_state, limit \\ @max_historical_packets) do - packet_ids = Enum.take(packet_state.historical_packet_ids, limit) - packet_ids |> PacketStore.get_packets() |> Map.values() - end - - @doc """ - Remove packets matching a predicate (e.g., expired packets). - """ - def remove_packets_where(packet_state, predicate_fn) do - # Get visible packets and apply predicate - visible_packets = get_visible_packets(packet_state) - {keep_visible, remove_visible} = Enum.split_with(visible_packets, predicate_fn) - - # Get historical packets and apply predicate - historical_packets = get_historical_packets(packet_state) - {keep_historical, remove_historical} = Enum.split_with(historical_packets, predicate_fn) - - # Remove from storage - remove_ids = extract_packet_ids(remove_visible ++ remove_historical) - PacketStore.remove_packets(remove_ids) - - # Update state with remaining IDs - keep_visible_ids = extract_packet_ids(keep_visible) - keep_historical_ids = extract_packet_ids(keep_historical) - - %{ - packet_state - | visible_packet_ids: keep_visible_ids, - historical_packet_ids: keep_historical_ids, - packet_count: %{ - visible: length(keep_visible_ids), - historical: length(keep_historical_ids) - } - } - end - - @doc """ - Get current memory usage statistics. - """ - def get_memory_stats(packet_state) do - store_stats = PacketStore.get_stats() - - %{ - visible_count: packet_state.packet_count.visible, - historical_count: packet_state.packet_count.historical, - total_stored: store_stats.total_packets, - memory_bytes: store_stats.memory_bytes, - avg_packet_size: - if(store_stats.total_packets > 0, - do: store_stats.memory_bytes / store_stats.total_packets, - else: 0 - ) - } - end - - # Private functions - - defp apply_memory_limits(packet_ids, max_count) do - if length(packet_ids) > max_count + @packet_cleanup_threshold do - # Keep most recent packets - managed_ids = Enum.take(packet_ids, max_count) - {managed_ids, true} - else - {packet_ids, false} - end - end - - defp maybe_cleanup(state, false, _type), do: state - - defp maybe_cleanup(%{last_cleanup: last_cleanup} = state, true, _type) do - current_time = System.monotonic_time(:millisecond) - - if current_time - last_cleanup > 30_000 do - %{state | last_cleanup: current_time} - else - state - end - end - - defp extract_packet_ids(packets) do - Enum.map(packets, fn packet -> - packet[:id] || packet["id"] || generate_packet_id(packet) - end) - end - - defp generate_packet_id(packet) do - # Generate deterministic ID from packet data - sender = packet[:sender] || packet["sender"] || "" - timestamp = packet[:received_at] || packet["received_at"] || DateTime.utc_now() - "#{sender}_#{DateTime.to_unix(timestamp, :microsecond)}" - end -end diff --git a/lib/aprsme_web/live/map_live/packet_store.ex b/lib/aprsme_web/live/map_live/packet_store.ex deleted file mode 100644 index 55bfafa..0000000 --- a/lib/aprsme_web/live/map_live/packet_store.ex +++ /dev/null @@ -1,164 +0,0 @@ -defmodule AprsmeWeb.MapLive.PacketStore do - @moduledoc """ - Efficient packet storage for LiveView that stores only IDs in assigns - and fetches full packet data on demand. - """ - - use GenServer - - require Logger - - @table_name :map_packet_store - @ttl_ms to_timeout(hour: 2) - @cleanup_interval_ms to_timeout(minute: 5) - - def start_link(opts \\ []) do - GenServer.start_link(__MODULE__, opts, name: __MODULE__) - end - - @doc """ - Store a packet and return its ID. - """ - def store_packet(packet) do - packet_id = generate_packet_id(packet) - expires_at = System.monotonic_time(:millisecond) + @ttl_ms - - :ets.insert(@table_name, {packet_id, packet, expires_at}) - packet_id - end - - @doc """ - Store multiple packets and return their IDs. - """ - def store_packets(packets) when is_list(packets) do - expires_at = System.monotonic_time(:millisecond) + @ttl_ms - - entries = - Enum.map(packets, fn packet -> - packet_id = generate_packet_id(packet) - {packet_id, packet, expires_at} - end) - - :ets.insert(@table_name, entries) - Enum.map(entries, fn {id, _, _} -> id end) - end - - @doc """ - Get a packet by ID. - """ - def get_packet(packet_id) do - case :ets.lookup(@table_name, packet_id) do - [{^packet_id, packet, expires_at}] -> - if System.monotonic_time(:millisecond) < expires_at do - {:ok, packet} - else - # Expired, remove it - :ets.delete(@table_name, packet_id) - {:error, :not_found} - end - - [] -> - {:error, :not_found} - end - end - - @doc """ - Get multiple packets by IDs. - """ - def get_packets(packet_ids) when is_list(packet_ids) do - current_time = System.monotonic_time(:millisecond) - - packet_ids - |> Enum.map(fn id -> - case :ets.lookup(@table_name, id) do - [{^id, packet, expires_at}] when current_time < expires_at -> - {id, packet} - - _ -> - nil - end - end) - |> Enum.reject(&is_nil/1) - |> Map.new() - end - - @doc """ - Remove a packet by ID. - """ - def remove_packet(packet_id) do - :ets.delete(@table_name, packet_id) - end - - @doc """ - Remove multiple packets by IDs. - """ - def remove_packets(packet_ids) when is_list(packet_ids) do - Enum.each(packet_ids, &:ets.delete(@table_name, &1)) - end - - @doc """ - Get statistics about the store. - """ - def get_stats do - %{ - total_packets: :ets.info(@table_name, :size), - memory_bytes: :ets.info(@table_name, :memory) * :erlang.system_info(:wordsize) - } - end - - # GenServer callbacks - - @impl true - def init(_opts) do - # Create ETS table for packet storage - :ets.new(@table_name, [ - :set, - :public, - :named_table, - read_concurrency: true, - write_concurrency: true - ]) - - # Schedule periodic cleanup - schedule_cleanup() - - {:ok, %{}} - end - - @impl true - def handle_info(:cleanup, state) do - cleanup_expired_packets() - schedule_cleanup() - {:noreply, state} - end - - # Private functions - - defp generate_packet_id(packet) do - # Generate a unique ID based on packet attributes - sender = Map.get(packet, :sender, Map.get(packet, "sender", "")) - timestamp = Map.get(packet, :received_at, Map.get(packet, "received_at", DateTime.utc_now())) - - # Create a deterministic ID - :md5 - |> :crypto.hash("#{sender}_#{timestamp}") - |> Base.encode16(case: :lower) - |> binary_part(0, 16) - end - - defp schedule_cleanup do - Process.send_after(self(), :cleanup, @cleanup_interval_ms) - end - - defp cleanup_expired_packets do - current_time = System.monotonic_time(:millisecond) - - # Use match_delete for efficient bulk deletion - match_spec = [{{:_, :_, :"$1"}, [{:<, :"$1", current_time}], [true]}] - deleted = :ets.select_delete(@table_name, match_spec) - - if deleted > 0 do - Logger.debug("Cleaned up #{deleted} expired packets from packet store") - end - end -end