postgres optimization

config/config.exs

@@ -58,11 +58,16 @@ config :aprsme,
   # Packet retention period in days (default: 365 days = 1 year)
   packet_retention_days: String.to_integer(System.get_env("PACKET_RETENTION_DAYS", "365")),
   # GenStage packet processing configuration
+  # Optimized for PostgreSQL with work_mem=16MB and synchronous_commit=off
   packet_pipeline: [
-    max_buffer_size: 1000,
-    batch_size: 100,
-    batch_timeout: 1000,
-    max_demand: 50
+    # Larger buffer since inserts are async
+    max_buffer_size: 5000,
+    # ~32KB per packet, stays within work_mem
+    batch_size: 500,
+    # 2 seconds - longer timeout for larger batches
+    batch_timeout: 2000,
+    # Higher demand to keep pipeline flowing
+    max_demand: 250
   ]
 
 config :error_tracker,

config/dev.exs

@@ -14,11 +14,27 @@ config :aprsme, Aprsme.Repo,
   database: "aprsme_dev",
   stacktrace: true,
   show_sensitive_data_on_connection_error: true,
-  pool_size: 10,
-  pool_timeout: 5000,
-  timeout: 15_000,
+  # Optimized pool settings for development
+  pool_size: 15,
+  pool_timeout: 10_000,
+  timeout: 30_000,
+  queue_target: 100,
+  queue_interval: 1_000,
   log: :debug,
-  types: Aprsme.PostgresTypes
+  types: Aprsme.PostgresTypes,
+  # Use unnamed prepared statements for better development flexibility
+  prepare: :unnamed,
+  # Socket options for better performance
+  socket_options: [
+    keepalive: true,
+    nodelay: true
+  ],
+  # Development-specific parameters matching PostgreSQL config
+  parameters: [
+    application_name: "aprsme_dev",
+    work_mem: "16MB",
+    statement_timeout: "60s"
+  ]
 
 config :aprsme, AprsmeWeb.Endpoint,
   # Binding to loopback ipv4 address prevents access from other machines.

config/runtime.exs

@@ -49,13 +49,41 @@ if config_env() == :prod do
   config :aprsme, Aprsme.Repo,
     # ssl: true,
     url: database_url,
-    pool_size: String.to_integer(System.get_env("POOL_SIZE") || "20"),
-    pool_timeout: String.to_integer(System.get_env("POOL_TIMEOUT") || "5000"),
-    timeout: String.to_integer(System.get_env("DB_TIMEOUT") || "15000"),
-    socket_options: maybe_ipv6,
+    # Optimized for max_connections=100 with other apps on server
+    pool_size: String.to_integer(System.get_env("POOL_SIZE") || "25"),
+    # Increased timeout for ARM system under load
+    pool_timeout: String.to_integer(System.get_env("POOL_TIMEOUT") || "10000"),
+    # Match PostgreSQL statement timeout capabilities
+    timeout: String.to_integer(System.get_env("DB_TIMEOUT") || "30000"),
+    socket_options:
+      maybe_ipv6 ++
+        [
+          # TCP optimizations for better connection handling
+          keepalive: true,
+          nodelay: true,
+          recbuf: 8192,
+          sndbuf: 8192
+        ],
     types: Aprsme.PostgresTypes,
-    queue_target: 10_000,
-    queue_interval: 20_000
+    # Reduced queue target for faster response
+    queue_target: 100,
+    # Check queue more frequently
+    queue_interval: 1_000,
+    # Optimize for unnamed prepared statements (better for dynamic queries)
+    prepare: :unnamed,
+    # Connection parameters to leverage PostgreSQL settings
+    parameters: [
+      # Application name for monitoring
+      application_name: "aprsme",
+      # Leverage synchronous_commit=off in postgresql.conf
+      synchronous_commit: "off",
+      # Match work_mem setting
+      work_mem: "16MB",
+      # Statement timeout as safety net
+      statement_timeout: "30s",
+      # Prevent idle transactions
+      idle_in_transaction_session_timeout: "60s"
+    ]
 
   config :aprsme, AprsmeWeb.Endpoint,
     url: [host: host, port: 443, scheme: "https"],

lib/aprsme/db_optimizer.ex

@@ -0,0 +1,226 @@
+defmodule Aprsme.DbOptimizer do
+  @moduledoc """
+  Database optimization utilities that leverage PostgreSQL server configuration
+  for better performance on ARM RK3588 with 16GB RAM.
+
+  PostgreSQL key settings we're optimizing for:
+  - max_connections = 100
+  - shared_buffers = 1GB  
+  - work_mem = 16MB
+  - synchronous_commit = off
+  - effective_io_concurrency = 100 (SSD optimized)
+  """
+
+  alias Aprsme.Repo
+  alias Ecto.Adapters.SQL
+
+  require Logger
+
+  @doc """
+  Execute a large batch insert using PostgreSQL COPY command for maximum performance.
+  This is significantly faster than INSERT for large datasets.
+  """
+  def copy_insert(table_name, columns, rows) when length(rows) > 1000 do
+    # Convert rows to CSV format for COPY
+    csv_data = rows_to_csv(rows, columns)
+
+    # Use COPY command which bypasses much of the overhead of INSERT
+    query = """
+    COPY #{table_name} (#{Enum.join(columns, ", ")})
+    FROM STDIN WITH (FORMAT csv, HEADER false)
+    """
+
+    SQL.query!(Repo, query, [csv_data])
+    length(rows)
+  rescue
+    error ->
+      Logger.error("COPY insert failed: #{inspect(error)}")
+      # Fall back to regular insert
+      regular_batch_insert(table_name, columns, rows)
+  end
+
+  def copy_insert(table_name, columns, rows) do
+    # For smaller batches, use regular insert
+    regular_batch_insert(table_name, columns, rows)
+  end
+
+  @doc """
+  Optimized batch insert that leverages PostgreSQL configuration
+  """
+  def optimized_batch_insert(schema, entries, opts \\ []) do
+    # Calculate optimal batch size based on work_mem
+    optimal_batch_size = calculate_optimal_batch_size(entries)
+
+    # Split into optimal chunks
+    entries
+    |> Enum.chunk_every(optimal_batch_size)
+    |> Enum.map(fn batch ->
+      insert_opts =
+        Keyword.merge(
+          [
+            returning: false,
+            on_conflict: :nothing,
+            timeout: 60_000,
+            placeholders: length(batch) > 100
+          ],
+          opts
+        )
+
+      case Repo.insert_all(schema, batch, insert_opts) do
+        {count, _} -> {:ok, count}
+        error -> {:error, error}
+      end
+    end)
+    |> Enum.reduce({0, 0}, fn
+      {:ok, count}, {success, errors} -> {success + count, errors}
+      {:error, _}, {success, errors} -> {success, errors + 1}
+    end)
+  end
+
+  @doc """
+  Calculate optimal batch size based on PostgreSQL work_mem setting (16MB)
+  and estimated row size
+  """
+  def calculate_optimal_batch_size(entries) when is_list(entries) do
+    # Estimate size of one entry (rough approximation)
+    sample = List.first(entries)
+    estimated_size = estimate_entry_size(sample)
+
+    # work_mem is 16MB, leave some headroom
+    # 14MB in bytes
+    available_memory = 14 * 1024 * 1024
+
+    # Calculate how many entries fit in work_mem
+    max_batch = div(available_memory, estimated_size)
+
+    # Cap at reasonable limits
+    max_batch |> min(2000) |> max(100)
+  end
+
+  @doc """
+  Run ANALYZE on a table after bulk inserts to update statistics
+  This helps PostgreSQL make better query plans
+  """
+  def analyze_table(table_name) do
+    SQL.query!(Repo, "ANALYZE #{table_name}", [])
+    :ok
+  rescue
+    error ->
+      Logger.warning("Failed to analyze table #{table_name}: #{inspect(error)}")
+      :error
+  end
+
+  @doc """
+  Vacuum a table to reclaim space and update visibility map
+  Use this after large delete operations
+  """
+  def vacuum_table(table_name, opts \\ []) do
+    full = Keyword.get(opts, :full, false)
+    analyze = Keyword.get(opts, :analyze, true)
+
+    vacuum_type = if full, do: "VACUUM FULL", else: "VACUUM"
+    analyze_clause = if analyze, do: " ANALYZE", else: ""
+
+    query = "#{vacuum_type}#{analyze_clause} #{table_name}"
+
+    # Vacuum operations can take a long time
+    SQL.query!(Repo, query, [], timeout: :infinity)
+    :ok
+  rescue
+    error ->
+      Logger.error("Failed to vacuum table #{table_name}: #{inspect(error)}")
+      :error
+  end
+
+  @doc """
+  Get current database statistics for monitoring
+  """
+  def get_connection_stats do
+    query = """
+    SELECT 
+      count(*) as total_connections,
+      count(*) FILTER (WHERE state = 'active') as active_connections,
+      count(*) FILTER (WHERE state = 'idle') as idle_connections,
+      count(*) FILTER (WHERE state = 'idle in transaction') as idle_in_transaction,
+      count(*) FILTER (WHERE wait_event_type IS NOT NULL) as waiting_connections
+    FROM pg_stat_activity
+    WHERE datname = current_database()
+    """
+
+    case SQL.query(Repo, query, []) do
+      {:ok, %{rows: [[total, active, idle, idle_tx, waiting]]}} ->
+        %{
+          total: total,
+          active: active,
+          idle: idle,
+          idle_in_transaction: idle_tx,
+          waiting: waiting
+        }
+
+      _ ->
+        %{}
+    end
+  end
+
+  # Private functions
+
+  defp regular_batch_insert(table_name, columns, rows) do
+    # Convert to maps for Ecto.insert_all
+    entries =
+      Enum.map(rows, fn row ->
+        columns |> Enum.zip(row) |> Map.new()
+      end)
+
+    {count, _} =
+      Repo.insert_all(table_name, entries,
+        returning: false,
+        on_conflict: :nothing,
+        timeout: 60_000
+      )
+
+    count
+  end
+
+  defp rows_to_csv(rows, _columns) do
+    Enum.map_join(rows, "\n", fn row ->
+      Enum.map_join(row, ",", &escape_csv_value/1)
+    end)
+  end
+
+  defp escape_csv_value(nil), do: ""
+
+  defp escape_csv_value(value) when is_binary(value) do
+    if String.contains?(value, [",", "\"", "\n"]) do
+      "\"#{String.replace(value, "\"", "\"\"")}\""
+    else
+      value
+    end
+  end
+
+  defp escape_csv_value(value), do: to_string(value)
+
+  # Default 1KB
+  defp estimate_entry_size(nil), do: 1024
+
+  defp estimate_entry_size(entry) when is_map(entry) do
+    # Rough estimation of entry size in bytes
+    entry
+    |> Map.values()
+    |> Enum.map(&estimate_value_size/1)
+    |> Enum.sum()
+    # Add overhead for structure
+    |> Kernel.+(100)
+  end
+
+  defp estimate_entry_size(_), do: 1024
+
+  defp estimate_value_size(nil), do: 4
+  defp estimate_value_size(value) when is_binary(value), do: byte_size(value)
+  defp estimate_value_size(value) when is_integer(value), do: 8
+  defp estimate_value_size(value) when is_float(value), do: 8
+  defp estimate_value_size(value) when is_boolean(value), do: 1
+  defp estimate_value_size(%DateTime{}), do: 8
+  defp estimate_value_size(%Date{}), do: 4
+  # Conservative estimate for complex types
+  defp estimate_value_size(_), do: 50
+end

lib/aprsme/packet_consumer.ex

@@ -100,9 +100,13 @@ defmodule Aprsme.PacketConsumer do
     {memory_before, _} = :erlang.statistics(:runtime)
     start_time = System.monotonic_time(:millisecond)
 
+    # Chunk size optimized for PostgreSQL work_mem=16MB
+    # With ~32KB per packet, we can fit ~500 packets in work_mem
+    chunk_size = Application.get_env(:aprsme, :packet_pipeline)[:batch_size] || 500
+
     results =
       packets
-      |> Enum.chunk_every(50)
+      |> Enum.chunk_every(chunk_size)
       |> Enum.map(&process_chunk/1)
 
     {success_count, error_count} =
@@ -172,7 +176,19 @@ defmodule Aprsme.PacketConsumer do
     {valid_packets, invalid_packets} = Enum.split_with(packet_attrs, &valid_packet?/1)
 
     # Insert valid packets in batch
-    case Repo.insert_all(Aprsme.Packet, valid_packets, returning: [:id]) do
+    # Optimized for PostgreSQL with synchronous_commit=off
+    insert_opts = [
+      # Don't return IDs for better performance
+      returning: false,
+      # Skip conflicts to avoid blocking
+      on_conflict: :nothing,
+      # Increased timeout for large batches
+      timeout: 60_000,
+      # Use placeholders for better performance with large batches
+      placeholders: length(valid_packets) > 100
+    ]
+
+    case Repo.insert_all(Aprsme.Packet, valid_packets, insert_opts) do
       {:error, error} ->
         Logger.error("Batch insert failed: #{inspect(error)}")
         {0, length(packets)}