From f1d16d656b899b8aaaabd9fd3b61e9e8a35d4606 Mon Sep 17 00:00:00 2001
From: Graham McIntire <graham@mcintire.me>
Date: Tue, 15 Jul 2025 08:25:02 -0500
Subject: [PATCH] Fix packet buffer overflow by reverting dynamic batch sizing
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- Increased packet buffer size from 1,000 to 10,000 to handle traffic spikes
- Removed InsertOptimizer dynamic batch sizing that was slowing processing
- Reverted to fixed batch size of 200 for consistent performance
- Simplified insert options to use static configuration
- Added telemetry metrics for buffer overflow and utilization monitoring
- Fixed unused variable warning in packet_consumer.ex

The root cause was the recent performance optimizations that introduced
dynamic batch sizing, which actually degraded performance and caused
the producer buffer to overflow when the consumer couldn't keep up.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
---
 config/config.exs             | 12 ++++++++----
 lib/aprsme/packet_consumer.ex | 26 +++++++++++---------------
 lib/aprsme/packet_producer.ex | 14 ++++++++++++++
 3 files changed, 33 insertions(+), 19 deletions(-)

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