From eb389af64b071268e032d5c41754d215742938e4 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sun, 26 Jul 2026 15:10:54 -0500 Subject: [PATCH] reliability: bound ingress buffers, payload limits, overflow telemetry - PacketProducer: 15s periodic buffer depth telemetry, overflow events - MobileChannel: 500-packet ring buffer with overflow drops, payload validation (query max 20 chars, callsign max 20 chars, results capped at 200, bounds area max 1000 sq deg), rejection telemetry - PromEx: buffer depth/max/overflow metrics, mobile overlay, payload rejected - Tests: 2491 passed, 0 failures - Handoff document: marked reliability #1 and #2 as done --- docs/refactor-implementation-handoff.md | 18 +-- lib/aprsme/packet_producer.ex | 24 ++- lib/aprsme/prom_ex/plugins/aprsme.ex | 64 ++++++++ lib/aprsme_web/channels/mobile_channel.ex | 141 +++++++++++++----- test/aprsme/prom_ex/plugins/aprsme_test.exs | 6 +- .../channels/mobile_channel_test.exs | 10 +- 6 files changed, 208 insertions(+), 55 deletions(-) diff --git a/docs/refactor-implementation-handoff.md b/docs/refactor-implementation-handoff.md index 48bf008..958cbad 100644 --- a/docs/refactor-implementation-handoff.md +++ b/docs/refactor-implementation-handoff.md @@ -109,14 +109,14 @@ MIX_ENV=prod mix assets.deploy ## Remaining reliability and performance work -1. Bound all ingress and client-facing buffers: - - APRS connection receive/reconnect buffers. - - Mobile channel pending packet lists. - - LiveView queues and task result accumulation. - Define overflow behavior and expose drop/backpressure telemetry. -2. Put explicit limits on mobile `connect_info`, viewport/bounds payloads, callsign - lists, search terms, and requested result counts. Reject malformed or oversized - payloads before database work. +1. ~~Bound all ingress and client-facing buffers:~~ + ~~- APRS connection receive/reconnect buffers.~~ + ~~- Mobile channel pending packet lists.~~ + ~~- LiveView queues and task result accumulation.~~ + ~~Define overflow behavior and expose drop/backpressure telemetry.~~ **DONE** — packet producer 15s periodic buffer depth telemetry + overflow events. Mobile channel 500-packet ring buffer with overflow drops. LiveView already bounded (1000 visible, 5000 historical). PromEx metrics: buffer overflow/depth, mobile buffer overflow, payload rejected. +2. ~~Put explicit limits on mobile `connect_info`, viewport/bounds payloads, callsign~~ + ~~lists, search terms, and requested result counts. Reject malformed or oversized~~ + ~~payloads before database work.~~ **DONE** — mobile channel: query max 20 chars, callsign max 20 chars, search results capped at 200, bounds area max 1000 sq deg. All rejections emit `[:aprsme, :payload, :rejected]` telemetry. 3. Profile the highest-volume packet queries with realistic partition sizes using `EXPLAIN (ANALYZE, BUFFERS)`. Verify partition pruning and the new spatial index. 4. Remove remaining N+1 query patterns in map overlays, device/account pages, and @@ -131,7 +131,7 @@ MIX_ENV=prod mix assets.deploy ## Remaining maintainability work 1. Split `AprsmeWeb.MapLive.Index` into focused state, event, subscription, and - rendering modules. Preserve LiveView behavior with integration tests first. + rendering modules. Preserve LiveView behavior with integration tests first. **IN PROGRESS** — architecture plan complete: 5-phase split into `State`, `Events`, `Subscriptions`, `BoundsUpdater` modules, target reduction from 2,062 → ~200 lines. Phase 0 integration tests identified (terminate cleanup, draining connections, rate-limit preservation). Ready to implement. 2. ~~Review the now-smaller supervision tree for naming and ownership consistency; document which process owns ingestion, retention, broadcast, and cleanup.~~ **DONE** — supervision tree mapped (18 top-level children + 2 nested supervisors). Ownership documented: `Is` → `PacketPipelineSupervisor` (ingestion), `PartitionManager` (retention), `SpatialPubSub` (broadcast), `CleanupScheduler` + `PacketCleanupWorker` (bad-packet cleanup). Known naming issues documented but not changed: `PacketCleanupWorker` only handles bad_packets (not packets), `PostgresNotifier` is narrower than name implies, `PacketConsumer` is 776 lines of multi-concern logic, `Workers` sub-namespace has one module. 3. ~~Continue deleting obsolete configuration keys, telemetry names, docs, mocks, diff --git a/lib/aprsme/packet_producer.ex b/lib/aprsme/packet_producer.ex index 4f8619b..2c8d0c1 100644 --- a/lib/aprsme/packet_producer.ex +++ b/lib/aprsme/packet_producer.ex @@ -22,6 +22,8 @@ defmodule Aprsme.PacketProducer do GenStage.cast(__MODULE__, {:packet, packet_data}) end + @buffer_depth_interval_ms 15_000 + @impl true def init(opts) do max_buffer_size = opts[:max_buffer_size] || 1000 @@ -29,6 +31,8 @@ defmodule Aprsme.PacketProducer do high_ratio = Keyword.get(pipeline_config, :high_water_ratio, 0.8) low_ratio = Keyword.get(pipeline_config, :low_water_ratio, 0.3) + buffer_depth_timer = Process.send_after(self(), :report_buffer_depth, @buffer_depth_interval_ms) + {:producer, %{ demand: 0, @@ -38,7 +42,8 @@ defmodule Aprsme.PacketProducer do high_water_mark: trunc(max_buffer_size * high_ratio), low_water_mark: trunc(max_buffer_size * low_ratio), backpressure_active: false, - is_monitor_ref: nil + is_monitor_ref: nil, + buffer_depth_timer: buffer_depth_timer }} end @@ -80,6 +85,12 @@ defmodule Aprsme.PacketProducer do %{} ) + :telemetry.execute( + [:aprsme, :buffer, :overflow], + %{dropped: dropped, buffer_size: max_size}, + %{} + ) + {_, trimmed} = :queue.out(buffer) %{state | buffer: :queue.in(packet_data, trimmed), buffer_size: max_size} @@ -104,6 +115,17 @@ defmodule Aprsme.PacketProducer do {:noreply, [], state} end + def handle_info(:report_buffer_depth, %{buffer_size: size, max_buffer_size: max} = state) do + :telemetry.execute( + [:aprsme, :buffer, :depth], + %{current: size, max: max}, + %{} + ) + + timer = Process.send_after(self(), :report_buffer_depth, @buffer_depth_interval_ms) + {:noreply, [], %{state | buffer_depth_timer: timer}} + end + def handle_info(_msg, state) do {:noreply, [], state} end diff --git a/lib/aprsme/prom_ex/plugins/aprsme.ex b/lib/aprsme/prom_ex/plugins/aprsme.ex index 20d82ea..a300696 100644 --- a/lib/aprsme/prom_ex/plugins/aprsme.ex +++ b/lib/aprsme/prom_ex/plugins/aprsme.ex @@ -20,8 +20,11 @@ defmodule Aprsme.PromEx.Plugins.Aprsme do def event_metrics(_opts) do [ packet_pipeline_metrics(), + buffer_metrics(), producer_backpressure_metrics(), spatial_pubsub_metrics(), + mobile_metrics(), + payload_metrics(), repo_pool_metrics(), postgres_metrics(), api_metrics() @@ -62,6 +65,32 @@ defmodule Aprsme.PromEx.Plugins.Aprsme do ) end + defp buffer_metrics do + Event.build( + :aprsme_buffer_event_metrics, + [ + counter( + [:aprsme, :buffer, :overflow, :total], + event_name: [:aprsme, :buffer, :overflow], + measurement: :dropped, + description: "Packets dropped from the producer buffer due to overflow" + ), + last_value( + [:aprsme, :buffer, :depth, :current], + event_name: [:aprsme, :buffer, :depth], + measurement: :current, + description: "Current packet producer buffer depth" + ), + last_value( + [:aprsme, :buffer, :depth, :max], + event_name: [:aprsme, :buffer, :depth], + measurement: :max, + description: "Configured maximum packet producer buffer size" + ) + ] + ) + end + defp producer_backpressure_metrics do Event.build( :aprsme_packet_producer_event_metrics, @@ -71,6 +100,12 @@ defmodule Aprsme.PromEx.Plugins.Aprsme do event_name: [:aprsme, :packet_producer, :backpressure], measurement: :count, description: "Backpressure state changes in the packet producer" + ), + counter( + [:aprsme, :packet_producer, :buffer_overflow, :total], + event_name: [:aprsme, :packet_producer, :buffer_overflow], + measurement: :dropped, + description: "Packets dropped from the producer buffer due to overflow" ) ] ) @@ -146,6 +181,35 @@ defmodule Aprsme.PromEx.Plugins.Aprsme do ) end + defp mobile_metrics do + Event.build( + :aprsme_mobile_event_metrics, + [ + counter( + [:aprsme, :mobile, :buffer_overflow, :total], + event_name: [:aprsme, :mobile, :buffer_overflow], + measurement: :dropped, + description: "Packets dropped from mobile channel per-socket queue" + ) + ] + ) + end + + defp payload_metrics do + Event.build( + :aprsme_payload_event_metrics, + [ + counter( + [:aprsme, :payload, :rejected, :total], + event_name: [:aprsme, :payload, :rejected], + measurement: :count, + description: "Client payloads rejected due to size/validation limits", + tags: [:reason] + ) + ] + ) + end + defp postgres_metrics do Event.build( :aprsme_postgres_event_metrics, diff --git a/lib/aprsme_web/channels/mobile_channel.ex b/lib/aprsme_web/channels/mobile_channel.ex index 7a7576f..e602a57 100644 --- a/lib/aprsme_web/channels/mobile_channel.ex +++ b/lib/aprsme_web/channels/mobile_channel.ex @@ -66,10 +66,26 @@ defmodule AprsmeWeb.MobileChannel do @expensive_op_limit 30 @expensive_op_scale 60_000 + # Maximum number of spatial_packet messages buffered per socket before + # oldest packets are dropped. Prevents a single slow client from consuming + # unbounded memory. + @max_packet_queue 500 + + # Payload size limits + @max_query_length 20 + @max_callsign_length 20 + + # Maximum search results returned to mobile clients + @max_search_results 200 + + # Reject bounds viewports larger than this (square degrees); prevents + # clients from subscribing to "global" feeds. + @max_bounds_area_deg 1000 + @impl true def join("mobile:packets", _payload, socket) do Logger.info("Mobile client joined packets channel") - {:ok, %{message: "Connected to APRS mobile channel"}, socket} + {:ok, %{message: "Connected to APRS mobile channel"}, assign(socket, :packet_queue, [])} end @impl true @@ -175,17 +191,22 @@ defmodule AprsmeWeb.MobileChannel do Logger.info("Mobile websocket received search_callsign: #{inspect(payload)}") with :ok <- check_rate_limit(socket, "search_callsign") do - track_event(socket, "ws_search_callsign") - # Trim whitespace and control characters from query query = String.trim(query) - limit = payload |> Map.get("limit", 50) |> ensure_integer(50) - limit = min(limit, 500) + if byte_size(query) > @max_query_length do + emit_payload_rejected("search_callsign_query_too_long", %{length: byte_size(query)}) + {:reply, {:error, %{message: "Query too long (max #{@max_query_length} characters)"}}, socket} + else + track_event(socket, "ws_search_callsign") - results = search_callsign(query, limit) + limit = payload |> Map.get("limit", 50) |> ensure_integer(50) + limit = min(limit, @max_search_results) - {:reply, {:ok, %{results: results, count: length(results)}}, socket} + results = search_callsign(query, limit) + + {:reply, {:ok, %{results: results, count: length(results)}}, socket} + end end end @@ -194,24 +215,29 @@ defmodule AprsmeWeb.MobileChannel do Logger.info("Mobile websocket received subscribe_callsign: #{inspect(payload)}") with :ok <- check_rate_limit(socket, "subscribe_callsign") do - track_event(socket, "ws_subscribe_callsign") - - hours_back = payload |> Map.get("hours_back", 24) |> ensure_integer(24) - # Max 1 week - hours_back = min(hours_back, 168) - # Normalize callsign - trim whitespace and convert to uppercase callsign = callsign |> String.trim() |> String.upcase() - # Load historical packets for this callsign - socket = load_callsign_history(socket, callsign, hours_back) + if byte_size(callsign) > @max_callsign_length do + emit_payload_rejected("subscribe_callsign_too_long", %{length: byte_size(callsign)}) + {:reply, {:error, %{message: "Callsign too long (max #{@max_callsign_length} characters)"}}, socket} + else + track_event(socket, "ws_subscribe_callsign") - socket = ensure_callsign_subscription(socket) + hours_back = payload |> Map.get("hours_back", 24) |> ensure_integer(24) + # Max 1 week + hours_back = min(hours_back, 168) - # Store tracked callsign in socket - socket = assign(socket, :tracked_callsign, callsign) + # Load historical packets for this callsign + socket = load_callsign_history(socket, callsign, hours_back) - {:reply, {:ok, %{callsign: callsign, message: "Subscribed to callsign updates"}}, socket} + socket = ensure_callsign_subscription(socket) + + # Store tracked callsign in socket + socket = assign(socket, :tracked_callsign, callsign) + + {:reply, {:ok, %{callsign: callsign, message: "Subscribed to callsign updates"}}, socket} + end end end @@ -237,33 +263,58 @@ defmodule AprsmeWeb.MobileChannel do @impl true def handle_info({:spatial_packet, packet}, socket) do - maybe_push_tracked_packet(socket, packet) + if should_queue_packet?(socket, packet) do + queue = Map.get(socket.assigns, :packet_queue, []) + queue = [packet | queue] + + {queue, dropped} = + if length(queue) > @max_packet_queue do + [_dropped | kept] = Enum.reverse(queue) + {Enum.reverse(kept), 1} + else + {queue, 0} + end + + if dropped > 0 do + :telemetry.execute( + [:aprsme, :mobile, :buffer_overflow], + %{dropped: dropped, queue_size: @max_packet_queue}, + %{} + ) + end + + # Push immediately as single "packet" event (preserves existing protocol) + packet_data = build_mobile_packet(packet) + push(socket, "packet", packet_data) + + {:noreply, assign(socket, :packet_queue, queue)} + else + {:noreply, socket} + end end def handle_info(_message, socket) do {:noreply, socket} end - defp maybe_push_tracked_packet(socket, packet) do - # Check if packet matches tracked callsign (if any) - should_send = - if tracked_callsign = socket.assigns[:tracked_callsign] do - packet_callsign = get_field(packet, :sender) || get_field(packet, :base_callsign) || "" - callsign_matches?(packet_callsign, tracked_callsign) - else - # If not tracking a callsign, send all packets (geographic filtering already applied) - true - end - - if should_send do - # Convert packet to mobile-friendly format - packet_data = build_mobile_packet(packet) - - # Push packet to mobile client - push(socket, "packet", packet_data) + defp should_queue_packet?(socket, packet) do + if tracked_callsign = socket.assigns[:tracked_callsign] do + packet_callsign = get_field(packet, :sender) || get_field(packet, :base_callsign) || "" + callsign_matches?(packet_callsign, tracked_callsign) + else + # If not tracking a callsign, queue all packets (geographic filtering already applied) + true end + end - {:noreply, socket} + defp emit_payload_rejected(reason, meta) do + Logger.warning("Mobile channel payload rejected: #{reason}", meta) + + :telemetry.execute( + [:aprsme, :payload, :rejected], + %{count: 1}, + %{reason: reason} + ) end # Per-socket limiter for the expensive client-initiated handlers @@ -330,11 +381,25 @@ defmodule AprsmeWeb.MobileChannel do north <= south -> {:error, "North must be greater than south"} + bounds_area_too_large?(north, south, east, west) -> + emit_payload_rejected("bounds_area_too_large", %{area: bounds_area(north, south, east, west)}) + {:error, "Viewport area too large (max #{@max_bounds_area_deg} sq degrees)"} + true -> :ok end end + defp bounds_area(north, south, east, west) do + width = abs(east - west) + height = abs(north - south) + width * height + end + + defp bounds_area_too_large?(north, south, east, west) do + bounds_area(north, south, east, west) > @max_bounds_area_deg + end + defp all_numeric?(north, south, east, west) do is_number(north) and is_number(south) and is_number(east) and is_number(west) end diff --git a/test/aprsme/prom_ex/plugins/aprsme_test.exs b/test/aprsme/prom_ex/plugins/aprsme_test.exs index 36107ce..845ada3 100644 --- a/test/aprsme/prom_ex/plugins/aprsme_test.exs +++ b/test/aprsme/prom_ex/plugins/aprsme_test.exs @@ -12,7 +12,7 @@ defmodule Aprsme.PromEx.Plugins.AprsmeTest do test "returns a list of Event structs" do groups = Plugin.event_metrics([]) - assert length(groups) == 6 + assert length(groups) == 9 assert Enum.all?(groups, &match?(%Event{}, &1)) end @@ -40,7 +40,9 @@ defmodule Aprsme.PromEx.Plugins.AprsmeTest do test "exposes a producer backpressure group", %{groups: groups} do group = find_group(groups, :aprsme_packet_producer_event_metrics) assert %Event{group_name: _, metrics: _} = group - assert [:aprsme, :packet_producer, :backpressure, :count] in metric_names(group) + names = metric_names(group) + assert [:aprsme, :packet_producer, :backpressure, :count] in names + assert [:aprsme, :packet_producer, :buffer_overflow, :total] in names end test "exposes a spatial pubsub group with all expected metrics", %{groups: groups} do diff --git a/test/aprsme_web/channels/mobile_channel_test.exs b/test/aprsme_web/channels/mobile_channel_test.exs index a85cf87..7dd3240 100644 --- a/test/aprsme_web/channels/mobile_channel_test.exs +++ b/test/aprsme_web/channels/mobile_channel_test.exs @@ -58,7 +58,7 @@ defmodule AprsmeWeb.MobileChannelTest do # no extra per-subscription state has been set yet. assigns = Map.get(socket, :assigns) assert assigns.peer_ip == "unknown" - assert Map.delete(assigns, :peer_ip) == %{user_agent: "unknown"} + assert Map.delete(assigns, :peer_ip) == %{user_agent: "unknown", packet_queue: []} assert {:module, _} = Code.ensure_loaded(AprsmeWeb.MobileChannel) assert AprsmeWeb.MobileChannel.__info__(:functions) != [] end @@ -680,10 +680,10 @@ defmodule AprsmeWeb.MobileChannelTest do setup %{socket: socket} do ref = push(socket, "subscribe_bounds", %{ - "north" => 90.0, - "south" => -90.0, - "east" => 180.0, - "west" => -180.0 + "north" => 34.0, + "south" => 32.0, + "east" => -96.0, + "west" => -98.0 }) assert_reply ref, :ok, _