diff --git a/config/config.exs b/config/config.exs index 4ad2210..1f58adb 100644 --- a/config/config.exs +++ b/config/config.exs @@ -59,10 +59,14 @@ config :aprsme, packet_retention_days: String.to_integer(System.get_env("PACKET_RETENTION_DAYS", "365")), # GenStage packet processing configuration packet_pipeline: [ - max_buffer_size: 1000, - batch_size: 100, - batch_timeout: 1000, - max_demand: 50 + # Increased from 1000 to handle traffic spikes + max_buffer_size: 10_000, + # Increased to process more packets per batch + batch_size: 200, + # Reduced timeout for faster processing + batch_timeout: 500, + # Increased demand for better throughput + max_demand: 100 ] config :error_tracker, diff --git a/lib/aprsme/packet_consumer.ex b/lib/aprsme/packet_consumer.ex index 808acb2..bbab17f 100644 --- a/lib/aprsme/packet_consumer.ex +++ b/lib/aprsme/packet_consumer.ex @@ -6,7 +6,6 @@ defmodule Aprsme.PacketConsumer do use GenStage alias Aprsme.LogSanitizer - alias Aprsme.Performance.InsertOptimizer alias Aprsme.Repo require Logger @@ -21,7 +20,7 @@ defmodule Aprsme.PacketConsumer do initial_batch_size = Aprsme.SystemMonitor.get_recommended_batch_size() batch_timeout = opts[:batch_timeout] || 500 # Maximum batch size to prevent unbounded memory growth - max_batch_size = opts[:max_batch_size] || 1000 + max_batch_size = opts[:max_batch_size] || 2000 # Start a timer for batch processing timer = Process.send_after(self(), :process_batch, batch_timeout) @@ -138,8 +137,8 @@ defmodule Aprsme.PacketConsumer do {memory_before, _} = :erlang.statistics(:runtime) start_time = System.monotonic_time(:millisecond) - # Use optimized batch size for INSERT performance - batch_size = InsertOptimizer.get_optimal_batch_size() + # Use fixed batch size for consistent performance + batch_size = 200 Logger.debug("Processing batch of #{length(packets)} packets with insert chunk size: #{batch_size}") @@ -215,15 +214,19 @@ defmodule Aprsme.PacketConsumer do if Enum.empty?(valid_packets) do {0, invalid_count} else - # Get optimized insert options - insert_options = InsertOptimizer.get_insert_options() + # Use simple insert options for reliability + insert_options = [ + returning: false, + on_conflict: :nothing, + timeout: 15_000 + ] - # Insert valid packets in batch with optimization + # Insert valid packets in batch result = Repo.insert_all(Aprsme.Packet, valid_packets, insert_options) # Record performance metrics for optimization end_time = System.monotonic_time(:millisecond) - duration = end_time - start_time + _duration = end_time - start_time case result do {:error, error} -> @@ -231,13 +234,6 @@ defmodule Aprsme.PacketConsumer do {0, length(packets)} {inserted_count, _} -> - # Record metrics for optimization - InsertOptimizer.record_insert_metrics( - length(valid_packets), - duration, - inserted_count - ) - {inserted_count, invalid_count} end end diff --git a/lib/aprsme/packet_producer.ex b/lib/aprsme/packet_producer.ex index 2d12ad2..1f00b2f 100644 --- a/lib/aprsme/packet_producer.ex +++ b/lib/aprsme/packet_producer.ex @@ -47,6 +47,13 @@ defmodule Aprsme.PacketProducer do } ) + # Emit telemetry for monitoring + :telemetry.execute( + [:aprsme, :packet_producer, :buffer_overflow], + %{dropped_count: 1, buffer_size: buffer_size}, + %{max_size: max_size} + ) + {:noreply, [], %{state | buffer: Enum.take(new_buffer, max_size)}} else # Log when buffer is getting full @@ -60,6 +67,13 @@ defmodule Aprsme.PacketProducer do ) end + # Emit telemetry for buffer utilization + :telemetry.execute( + [:aprsme, :packet_producer, :buffer_utilization], + %{buffer_size: buffer_size, utilization: buffer_size / max_size}, + %{max_size: max_size} + ) + {:noreply, [], %{state | buffer: new_buffer}} end end