From f225f57044aa95425be0ed5ec40d0b4fcfe80db0 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 21 Apr 2026 10:14:59 -0500 Subject: [PATCH] perf(packets): keyset-based pagination for recent-packet queries get_recent_packets/1 and get_recent_packets_for_map/1 now accept a :cursor option (%{received_at: DateTime, id: uuid}) instead of :offset for deep paging. On the 100M-row partitioned packets table, offset scanning was linear in page depth; keyset is constant. Ordering tightened to (received_at DESC, id DESC) so the cursor is stable across ties. idx_packets_received_desc still covers the lookup; a dedicated (received_at DESC, id DESC) index isn't worth its size given microsecond timestamp granularity. historical_loader migrated from batch_offset*batch_size scheduling to a sequential cursor chain. The old loader fanned out N Process.send_after batches up-front which could arrive out-of-order; the cursor version waits for batch N before scheduling N+1, and also short-circuits the remaining schedule when an under-full batch signals exhaustion. mobile_channel load_historical_packets dropped its meaningless `offset: 0`; callsign history already lacked offset. No LiveView or WebSocket payload shapes changed. offset: 0 (or omitted) still behaves as first page; non-zero offsets are silently ignored. --- lib/aprsme/packets.ex | 101 +++++++-------- lib/aprsme_web/channels/mobile_channel.ex | 1 - .../live/map_live/historical_loader.ex | 118 ++++++++++++------ 3 files changed, 128 insertions(+), 92 deletions(-) diff --git a/lib/aprsme/packets.ex b/lib/aprsme/packets.ex index be5f5ca..985b1df 100644 --- a/lib/aprsme/packets.ex +++ b/lib/aprsme/packets.ex @@ -484,21 +484,49 @@ defmodule Aprsme.Packets do @doc """ Gets recent packets for initial map load. This uses an efficient query pattern for the most common use case. + + Ordering is `received_at DESC, id DESC`. Pagination is keyset/cursor-based: + + * `:cursor` - Optional `%{received_at: DateTime.t(), id: Ecto.UUID.t()}` marking + the last row from the previous page. When present, only rows strictly older + than the cursor are returned. When absent, the first page is returned. + * `:limit` - Maximum number of packets to return (default 200). + + Offset-based pagination is no longer supported; callers doing deep paging must + thread `{received_at, id}` from the last returned packet back in as `:cursor`. + Passing `offset: 0` (or omitting it) is still accepted and behaves as "first page". """ @impl true @spec get_recent_packets(map()) :: [struct()] def get_recent_packets(opts \\ %{}) do - require Logger + opts + |> build_recent_packets_query() + |> QueryBuilder.with_coordinates() + |> Repo.all() + end - # Use hours_back from opts if provided, otherwise default to 24 hours + @doc """ + Gets recent packets for map display with only the columns needed for rendering. + Returns maps (not full Packet structs) to avoid loading all 73 columns. + + Pagination is cursor/keyset-based — see `get_recent_packets/1`. + """ + @spec get_recent_packets_for_map(map()) :: [map()] + def get_recent_packets_for_map(opts \\ %{}) do + opts + |> build_recent_packets_query() + |> QueryBuilder.select_map_fields() + |> Repo.all() + end + + # Shared query construction for get_recent_packets/1 and get_recent_packets_for_map/1. + # Applies time filter, position/callsign/bounds/region filters, cursor keyset, + # and ordering/limit. The two public functions differ only in their select shape. + defp build_recent_packets_query(opts) do hours_back = Map.get(opts, :hours_back, 24) time_ago = DateTime.add(DateTime.utc_now(), -hours_back * 3600, :second) limit = Map.get(opts, :limit, 200) - offset = Map.get(opts, :offset, 0) - # Build base query with time filter - # Only filter by position if not tracking a specific callsign - # Also normalize callsign to handle edge cases callsign = opts[:callsign] normalized_callsign = if is_binary(callsign), do: String.trim(callsign), else: "" @@ -516,62 +544,29 @@ defmodule Aprsme.Packets do ) end - # Apply spatial and other filters BEFORE limiting - # This ensures we get the most recent packets within the specified bounds - bounds = Map.get(opts, :bounds) - - query = - base_query - |> QueryBuilder.maybe_filter_region(opts) - |> maybe_filter_by_callsign(opts) - |> maybe_filter_by_bounds(bounds) - |> order_by(desc: :received_at) - |> limit(^limit) - |> offset(^offset) - |> QueryBuilder.with_coordinates() - - Repo.all(query) - end - - @doc """ - Gets recent packets for map display with only the columns needed for rendering. - Returns maps (not full Packet structs) to avoid loading all 73 columns. - """ - @spec get_recent_packets_for_map(map()) :: [map()] - def get_recent_packets_for_map(opts \\ %{}) do - hours_back = Map.get(opts, :hours_back, 24) - time_ago = DateTime.add(DateTime.utc_now(), -hours_back * 3600, :second) - limit = Map.get(opts, :limit, 200) - offset = Map.get(opts, :offset, 0) - - callsign = opts[:callsign] - normalized_callsign = if is_binary(callsign), do: String.trim(callsign), else: "" - - base_query = - if normalized_callsign == "" do - from(p in Packet, - where: p.has_position == true, - where: p.received_at >= ^time_ago - ) - else - from(p in Packet, - where: p.received_at >= ^time_ago - ) - end - bounds = Map.get(opts, :bounds) base_query |> QueryBuilder.maybe_filter_region(opts) |> maybe_filter_by_callsign(opts) |> maybe_filter_by_bounds(bounds) - |> order_by(desc: :received_at) + |> maybe_apply_cursor(Map.get(opts, :cursor)) + |> order_by([p], desc: p.received_at, desc: p.id) |> limit(^limit) - |> offset(^offset) - |> QueryBuilder.select_map_fields() - |> Repo.all() end + # Keyset pagination: only rows strictly older than the cursor in the + # (received_at DESC, id DESC) ordering. + defp maybe_apply_cursor(query, %{received_at: %DateTime{} = cursor_received_at, id: cursor_id}) + when is_binary(cursor_id) do + from p in query, + where: + p.received_at < ^cursor_received_at or + (p.received_at == ^cursor_received_at and p.id < ^cursor_id) + end + + defp maybe_apply_cursor(query, _), do: query + @doc """ Gets the closest stations to a given point. Uses PostGIS spatial indexes for efficient querying. diff --git a/lib/aprsme_web/channels/mobile_channel.ex b/lib/aprsme_web/channels/mobile_channel.ex index 58ddd8f..d6b026a 100644 --- a/lib/aprsme_web/channels/mobile_channel.ex +++ b/lib/aprsme_web/channels/mobile_channel.ex @@ -439,7 +439,6 @@ defmodule AprsmeWeb.MobileChannel do params = %{ bounds: bounds_list, limit: limit, - offset: 0, hours_back: hours_back } diff --git a/lib/aprsme_web/live/map_live/historical_loader.ex b/lib/aprsme_web/live/map_live/historical_loader.ex index 61ca9d9..b5eec29 100644 --- a/lib/aprsme_web/live/map_live/historical_loader.ex +++ b/lib/aprsme_web/live/map_live/historical_loader.ex @@ -69,29 +69,23 @@ defmodule AprsmeWeb.MapLive.HistoricalLoader do |> assign(:total_batches, 1) |> assign(:historical_loading, true) |> assign(:loading_generation, new_generation) + |> assign(:historical_cursor, nil) |> load_historical_batch(0, new_generation) else - # Low zoom - use progressive loading to prevent overwhelming + # Low zoom - use progressive loading to prevent overwhelming. + # Batches are keyset-paginated via :historical_cursor, so they MUST run + # sequentially — each batch's cursor comes from the tail of the previous + # batch. Subsequent batches are scheduled from process_loaded_packets/4 + # (and the empty-batch branch) after the previous batch completes. total_batches = calculate_batch_count_for_zoom(zoom) - # Start with first batch - socket = - socket - |> assign(:loading_batch, 0) - |> assign(:total_batches, total_batches) - |> assign(:historical_loading, true) - |> assign(:loading_generation, new_generation) - |> load_historical_batch(0, new_generation) - - # Schedule remaining batches with generation check - batch_refs = - Enum.map(1..(total_batches - 1), fn batch_index -> - Process.send_after(self(), {:load_historical_batch, batch_index, new_generation}, batch_index * 50) - end) - - socket = assign(socket, :pending_batch_tasks, batch_refs) - socket + |> assign(:loading_batch, 0) + |> assign(:total_batches, total_batches) + |> assign(:historical_loading, true) + |> assign(:loading_generation, new_generation) + |> assign(:historical_cursor, nil) + |> load_historical_batch(0, new_generation) end end @@ -149,22 +143,26 @@ defmodule AprsmeWeb.MapLive.HistoricalLoader do # Calculate zoom-based batch size - higher zoom = smaller batches for faster response zoom = socket.assigns.map_zoom || 5 batch_size = calculate_batch_size_for_zoom(zoom) - offset = batch_offset * batch_size + + # Track cumulative packets loaded so far (batch_offset * batch_size is the + # pre-keyset approximation; we stop when we've fetched at least + # max_packets_for_zoom rows.) + packets_so_far = batch_offset * batch_size # Get total packet limit based on zoom level max_packets_for_zoom = get_packet_limit_for_zoom(zoom) # Check if we've reached the zoom-based limit - if offset >= max_packets_for_zoom do + if packets_so_far >= max_packets_for_zoom do finish_historical_loading(socket) else # Adjust batch size if it would exceed the limit - adjusted_batch_size = min(batch_size, max_packets_for_zoom - offset) + adjusted_batch_size = min(batch_size, max_packets_for_zoom - packets_so_far) query_params = %{ bounds: bounds_list, limit: adjusted_batch_size, - offset: offset, + cursor: socket.assigns[:historical_cursor], zoom: zoom, callsign: socket.assigns.tracked_callsign, hours_back: socket.assigns.historical_hours || "1" @@ -197,7 +195,15 @@ defmodule AprsmeWeb.MapLive.HistoricalLoader do # If tracking a callsign and this is the first batch, also load RF path stations socket = maybe_load_rf_path_stations(socket, batch_offset, historical_packets) - process_loaded_packets(socket, historical_packets, packet_data_list, batch_offset) + # Advance the cursor to the last packet returned (packets are ordered + # received_at DESC, id DESC, so the tail is the next page's boundary). + socket = advance_historical_cursor(socket, historical_packets) + + # Signal "there may be more" when the batch was fully saturated; an + # under-full batch means we've drained the result set for this window. + has_more = length(historical_packets) >= adjusted_batch_size + + process_loaded_packets(socket, historical_packets, packet_data_list, batch_offset, has_more) else # No packets found - handle this as a completed batch to prevent infinite loading require Logger @@ -206,21 +212,40 @@ defmodule AprsmeWeb.MapLive.HistoricalLoader do "No historical packets found for batch #{batch_offset}, hours_back: #{Map.get(socket.assigns, :historical_hours, "1")}, callsign: #{Map.get(socket.assigns, :tracked_callsign, "")}" ) - total_batches = socket.assigns.total_batches || 1 - is_final_batch = batch_offset >= total_batches - 1 + # Empty batch → we've drained the window; stop regardless of batch count. + socket = + socket + |> assign(:loading_batch, batch_offset + 1) + |> update_loading_status(true) - # Update progress for user feedback - socket = assign(socket, :loading_batch, batch_offset + 1) - - # Mark loading as complete if this was the final batch or if we're doing single-batch loading - socket = update_loading_status(socket, is_final_batch) - - # Process any pending bounds update if loading is complete - handle_pending_bounds(socket, is_final_batch) + # Process any pending bounds update + handle_pending_bounds(socket, true) end end end + # Extract the next keyset cursor from the last packet in the list. + defp advance_historical_cursor(socket, packets) do + case List.last(packets) do + nil -> + socket + + last -> + received_at = get_cursor_field(last, :received_at) + id = get_cursor_field(last, :id) + + if is_struct(received_at, DateTime) and is_binary(id) do + assign(socket, :historical_cursor, %{received_at: received_at, id: id}) + else + socket + end + end + end + + defp get_cursor_field(packet, key) when is_map(packet) do + Map.get(packet, key) || Map.get(packet, to_string(key)) + end + defp query_historical_packets(query_params, batch_offset) do require Logger @@ -240,7 +265,7 @@ defmodule AprsmeWeb.MapLive.HistoricalLoader do packets_module.get_recent_packets(%{ bounds: query_params.bounds, limit: query_params.limit, - offset: query_params.offset + cursor: query_params.cursor }) end rescue @@ -257,7 +282,7 @@ defmodule AprsmeWeb.MapLive.HistoricalLoader do params = %{ bounds: query_params.bounds, limit: query_params.limit, - offset: query_params.offset, + cursor: query_params.cursor, zoom: query_params.zoom, hours_back: historical_hours } @@ -283,12 +308,15 @@ defmodule AprsmeWeb.MapLive.HistoricalLoader do defp maybe_include_latest_packet(recent_packets, _callsign, _batch_offset), do: recent_packets - defp process_loaded_packets(socket, _historical_packets, [], _batch_offset), do: socket + defp process_loaded_packets(socket, _historical_packets, [], _batch_offset, _has_more), do: socket - defp process_loaded_packets(socket, historical_packets, packet_data_list, batch_offset) do + defp process_loaded_packets(socket, historical_packets, packet_data_list, batch_offset, has_more) do # Check zoom level to decide between heat map and markers total_batches = socket.assigns.total_batches || 4 - is_final_batch = batch_offset >= total_batches - 1 + + # A batch is final when: we've hit the configured batch count, or the + # result set has been fully drained (has_more = false). + is_final_batch = batch_offset >= total_batches - 1 or not has_more socket = handle_zoom_based_display(socket, historical_packets, packet_data_list, is_final_batch, batch_offset) @@ -298,10 +326,24 @@ defmodule AprsmeWeb.MapLive.HistoricalLoader do # Mark loading as complete if this was the final batch socket = update_loading_status(socket, is_final_batch) + # Schedule the next batch if there's more to load. Batches run sequentially + # because each inherits the :historical_cursor advanced by the previous one. + socket = maybe_schedule_next_batch(socket, batch_offset, is_final_batch) + # Process any pending bounds update if loading is complete handle_pending_bounds(socket, is_final_batch) end + defp maybe_schedule_next_batch(socket, _batch_offset, true), do: socket + + defp maybe_schedule_next_batch(socket, batch_offset, false) do + next_batch = batch_offset + 1 + generation = socket.assigns.loading_generation + timer_ref = Process.send_after(self(), {:load_historical_batch, next_batch, generation}, 50) + + assign(socket, :pending_batch_tasks, [timer_ref | socket.assigns.pending_batch_tasks]) + end + # Handle low zoom (heat map) defp handle_zoom_based_display( %{assigns: %{map_zoom: zoom}} = socket,