graham/aprs.me
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Remove broken dual packet state system (PacketManager/PacketStore)

Browse source 
System A (PacketManager + PacketStore ETS) had a fatal ID mismatch:
extract_packet_ids returned Ecto UUIDs but PacketStore stored under
MD5 hashes — cleanup could never find stored packets. Unified all
packet tracking on visible_packets map in LiveView assigns.

Deletes packet_manager.ex and packet_store.ex, removes from
supervision tree and PacketDistributor.
This commit is contained in:
Graham McIntire 2026-02-20 07:47:30 -06:00
parent b1358a8fad
commit 4153f66b86
No known key found for this signature in database
6 changed files with 6 additions and 393 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

37
TODO.md
View file

@ -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

3
lib/aprsme/application.ex
View file

@ -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)

4
lib/aprsme/cluster/packet_distributor.ex
View file

@ -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

11
lib/aprsme_web/live/map_live/index.ex
View file

@ -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)

180
lib/aprsme_web/live/map_live/packet_manager.ex
View file

@ -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

164
lib/aprsme_web/live/map_live/packet_store.ex
View file

@ -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