From 2fc4706d3d4cd0fc7f5b9b52d96ff4d44acb7aba Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 21 Apr 2026 09:29:05 -0500 Subject: [PATCH] fix: stop silent packet loss, timer leak, and seq-scans - packet_consumer: when a batch exceeds max_batch_size, carry the excess over to the next cycle instead of dropping it. Previously any overage was split off into a drop list that was only logged, losing packets the producer had already acknowledged. - map_live: cancel the hover_end_timer in terminate/2 so a user disconnecting while a marker is still highlighted doesn't leave an orphaned timer behind. - packet_consumer: rescue/log exceptions inside the async broadcast task so PubSub serialization bugs or outages are observable instead of silently killing the Task. - prepared_queries: swap ST_Y/ST_X BETWEEN predicates for the && / ST_MakeEnvelope pattern so bounds queries hit the GIST index on packets.location instead of sequentially scanning. --- lib/aprsme/packet_consumer.ex | 70 +++++++++++++++++--------- lib/aprsme/packets/prepared_queries.ex | 8 +-- lib/aprsme_web/live/map_live/index.ex | 7 +++ test/aprsme/packet_consumer_test.exs | 30 ++++++----- 4 files changed, 76 insertions(+), 39 deletions(-) diff --git a/lib/aprsme/packet_consumer.ex b/lib/aprsme/packet_consumer.ex index 526a091..40284d6 100644 --- a/lib/aprsme/packet_consumer.ex +++ b/lib/aprsme/packet_consumer.ex @@ -84,16 +84,38 @@ defmodule Aprsme.PacketConsumer do handle_partial_batch(batch, new_batch_length, state) end - # Handle oversized batch with pattern matching + # Handle oversized batch with pattern matching. + # Process the oldest `max_size` packets now and keep the rest for the next + # batch cycle — dropping them here would silently lose packets that the + # producer has already acknowledged. defp handle_oversized_batch(batch, max_size, state) do - # Reverse once for processing + # Reverse once to process in arrival order reversed_batch = Enum.reverse(batch) - {process_batch, drop_batch} = Enum.split(reversed_batch, max_size) + {process_now, remainder} = Enum.split(reversed_batch, max_size) - process_batch(process_batch) - log_dropped_packets(drop_batch, process_batch) + process_batch(process_now) + + # Keep the remainder queued. Internal batch is stored reversed + # (newest-at-head) so re-reverse the leftover arrival-order slice. + carryover_batch = Enum.reverse(remainder) + carryover_length = length(remainder) + + if carryover_length > 0 do + Logger.info("Carrying over #{carryover_length} packets past batch-size limit to next cycle", + batch_info: + LogSanitizer.log_data( + carryover_count: carryover_length, + processed_count: length(process_now) + ) + ) + end + + new_state = + state + |> reset_batch_timer() + |> Map.put(:batch, carryover_batch) + |> Map.put(:batch_length, carryover_length) - new_state = reset_batch_timer(state) {:noreply, [], new_state} end @@ -125,21 +147,6 @@ defmodule Aprsme.PacketConsumer do %{state | batch: [], batch_length: 0, timer: new_timer} end - # Pattern matching for logging - @spec log_dropped_packets(list(), list()) :: :ok - defp log_dropped_packets([], _), do: :ok - - defp log_dropped_packets(dropped, processed) do - Logger.warning("Dropped #{length(dropped)} packets due to batch size limit", - batch_info: - LogSanitizer.log_data( - dropped_count: length(dropped), - processed_count: length(processed), - reason: "batch_size_limit_exceeded" - ) - ) - end - @impl true # Pattern matching for empty batch def handle_info(:process_batch, %{batch: []} = state) do @@ -329,11 +336,28 @@ defmodule Aprsme.PacketConsumer do end) end - # Broadcast packets asynchronously using supervised task pool + # Broadcast packets asynchronously using supervised task pool. + # Failures here were previously silent — the Task died, the broadcast never + # reached subscribers, and there was no log trail. Rescue and log so + # operational issues (PubSub down, serialization bugs) are observable. defp broadcast_packets_async(bcasts) do fun = fn -> cluster_enabled = Application.get_env(:aprsme, :cluster_enabled, false) - Enum.each(bcasts, &broadcast_single_packet(&1, cluster_enabled)) + + Enum.each(bcasts, fn bcast -> + try do + broadcast_single_packet(bcast, cluster_enabled) + rescue + error -> + Logger.error("Packet broadcast failed: #{inspect(error)}", + broadcast_error: + LogSanitizer.log_data( + error: Exception.message(error), + routing_callsign: Map.get(bcast, :routing_callsign) + ) + ) + end + end) end if test_env?() do diff --git a/lib/aprsme/packets/prepared_queries.ex b/lib/aprsme/packets/prepared_queries.ex index c236f00..7a02504 100644 --- a/lib/aprsme/packets/prepared_queries.ex +++ b/lib/aprsme/packets/prepared_queries.ex @@ -88,8 +88,9 @@ defmodule Aprsme.Packets.PreparedQueries do from(p in Packet, where: p.has_position == true, where: p.received_at >= ^time_ago, - where: fragment("ST_Y(?) BETWEEN ? AND ?", p.location, ^south, ^north), - where: fragment("ST_X(?) BETWEEN ? AND ?", p.location, ^west, ^east), + # Use the && operator with ST_MakeEnvelope so PostGIS can hit the GIST + # index on `location`. ST_Y / ST_X in the WHERE clause force a seq scan. + where: fragment("? && ST_MakeEnvelope(?, ?, ?, ?, 4326)", p.location, ^west, ^south, ^east, ^north), order_by: [desc: p.received_at], limit: 500, select: %{p | lat: fragment("ST_Y(?)", p.location), lon: fragment("ST_X(?)", p.location)} @@ -208,8 +209,7 @@ defmodule Aprsme.Packets.PreparedQueries do from(p in Packet, where: p.has_position == true, where: p.received_at >= ^time_ago, - where: fragment("ST_Y(?) BETWEEN ? AND ?", p.location, ^south, ^north), - where: fragment("ST_X(?) BETWEEN ? AND ?", p.location, ^west, ^east), + where: fragment("? && ST_MakeEnvelope(?, ?, ?, ?, 4326)", p.location, ^west, ^south, ^east, ^north), select: count(p.id) ) diff --git a/lib/aprsme_web/live/map_live/index.ex b/lib/aprsme_web/live/map_live/index.ex index a8eca8d..aafb9ca 100644 --- a/lib/aprsme_web/live/map_live/index.ex +++ b/lib/aprsme_web/live/map_live/index.ex @@ -1882,6 +1882,13 @@ defmodule AprsmeWeb.MapLive.Index do Process.cancel_timer(socket.assigns.bounds_update_timer) end + # Clean up pending hover-end timer — set when the mouse leaves a marker + # and only cleared when it re-enters; if the LV terminates between the two, + # the timer would otherwise outlive the socket. + if socket.assigns[:hover_end_timer] do + Process.cancel_timer(socket.assigns.hover_end_timer) + end + # Clean up any pending batch tasks if socket.assigns[:pending_batch_tasks] do Enum.each(socket.assigns.pending_batch_tasks, &Process.cancel_timer/1) diff --git a/test/aprsme/packet_consumer_test.exs b/test/aprsme/packet_consumer_test.exs index d32895f..6fcbd9f 100644 --- a/test/aprsme/packet_consumer_test.exs +++ b/test/aprsme/packet_consumer_test.exs @@ -315,16 +315,16 @@ defmodule Aprsme.PacketConsumerTest do assert packet.sender == "BROADCAST1" end - test "respects max_batch_size and drops excess packets" do + test "respects max_batch_size and carries excess packets over to the next cycle" do import ExUnit.CaptureLog - # Temporarily enable warning level for this test - Logger.configure(level: :warning) + # Temporarily enable info level so we can see the carryover log line + Logger.configure(level: :info) # Create more packets than max_batch_size events = for i <- 1..150 do %{ - sender: "K#{i}DROP", + sender: "K#{i}CARRY", lat: 35.0, lon: -75.0, data_type: "position" @@ -336,25 +336,31 @@ defmodule Aprsme.PacketConsumerTest do batch_length: 0, batch_size: 50, batch_timeout: 1000, - # Only 100 will be processed + # Only 100 will be processed this cycle; 50 carry over max_batch_size: 100, timer: nil } - # Capture logs to verify warning - log = - capture_log(fn -> - {:noreply, [], _new_state} = PacketConsumer.handle_events(events, nil, state) + # Capture logs to verify carryover notice + {new_state, log} = + with_log(fn -> + {:noreply, [], new_state} = PacketConsumer.handle_events(events, nil, state) Process.sleep(20) + new_state end) - # Should log warning about dropped packets - assert log =~ "Dropped 50 packets due to batch size limit" + # Should log that 50 packets carried over to the next cycle + assert log =~ "Carrying over 50 packets past batch-size limit to next cycle" + + # The remaining 50 packets should still be in the consumer's batch, + # waiting for the next tick — they were not silently dropped. + assert new_state.batch_length == 50 + assert length(new_state.batch) == 50 # Reset log level Logger.configure(level: :error) - # Only max_batch_size packets should be processed + # Only max_batch_size packets should have been inserted this cycle packet_count = Repo.aggregate(Packet, :count) assert packet_count <= 100 end