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

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

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

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,

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

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

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, _