Fix packet buffer overflow with parallel consumer pool

- Create PacketConsumerPool to run 3 parallel consumers (configurable)
- Increase max_demand from 250 to 750 total (250 per consumer)
- Fix timer management to properly cancel/restart on batch processing
- Configure proper GenStage subscriptions with backpressure control
- Allow unnamed consumers for pool usage

This 3x increase in processing capacity prevents the "Packet buffer full"
warnings by ensuring consumers can keep up with incoming packet rate.

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

Co-Authored-By: Claude <noreply@anthropic.com>

config/config.exs

@@ -67,7 +67,9 @@ config :aprsme,
     # 2 seconds - longer timeout for larger batches
     batch_timeout: 2000,
     # Higher demand to keep pipeline flowing
-    max_demand: 250
+    max_demand: 750,
+    # Number of parallel consumers for better throughput
+    num_consumers: 3
   ]
 
 config :error_tracker,

lib/aprsme/packet_consumer.ex

@@ -11,7 +11,14 @@ defmodule Aprsme.PacketConsumer do
   require Logger
 
   def start_link(opts \\ []) do
-    GenStage.start_link(__MODULE__, opts, name: __MODULE__)
+    # Allow unnamed consumers for pool usage
+    name = opts[:name]
+
+    if name do
+      GenStage.start_link(__MODULE__, opts, name: name)
+    else
+      GenStage.start_link(__MODULE__, opts)
+    end
   end
 
   @impl true
@@ -24,6 +31,9 @@ defmodule Aprsme.PacketConsumer do
     # Start a timer for batch processing
     timer = Process.send_after(self(), :process_batch, batch_timeout)
 
+    # Extract subscription options if provided
+    subscribe_to = opts[:subscribe_to] || [{Aprsme.PacketProducer, max_demand: opts[:max_demand] || 250}]
+
     {:consumer,
      %{
        batch: [],
@@ -31,11 +41,15 @@ defmodule Aprsme.PacketConsumer do
        batch_timeout: batch_timeout,
        max_batch_size: max_batch_size,
        timer: timer
-     }}
+     }, subscribe_to: subscribe_to}
   end
 
   @impl true
-  def handle_events(events, _from, %{batch: batch, batch_size: batch_size, max_batch_size: max_batch_size} = state) do
+  def handle_events(
+        events,
+        _from,
+        %{batch: batch, batch_size: batch_size, max_batch_size: max_batch_size, timer: timer} = state
+      ) do
     new_batch = batch ++ events
     new_batch_length = length(new_batch)
 
@@ -57,12 +71,19 @@ defmodule Aprsme.PacketConsumer do
           )
         end
 
-        {:noreply, [], %{state | batch: []}}
+        # Cancel and restart timer
+        Process.cancel_timer(timer)
+        new_timer = Process.send_after(self(), :process_batch, state.batch_timeout)
+        {:noreply, [], %{state | batch: [], timer: new_timer}}
 
       new_batch_length >= batch_size ->
         # Process the batch immediately
         process_batch(new_batch)
-        {:noreply, [], %{state | batch: []}}
+
+        # Cancel and restart timer
+        Process.cancel_timer(timer)
+        new_timer = Process.send_after(self(), :process_batch, state.batch_timeout)
+        {:noreply, [], %{state | batch: [], timer: new_timer}}
 
       true ->
         # Add to batch and wait for more

lib/aprsme/packet_consumer_pool.ex

@@ -0,0 +1,41 @@
+defmodule Aprsme.PacketConsumerPool do
+  @moduledoc """
+  Manages a pool of packet consumers for parallel processing.
+  Each consumer subscribes to the producer with its own demand.
+  """
+  use Supervisor
+
+  def start_link(opts \\ []) do
+    Supervisor.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  @impl true
+  def init(opts) do
+    config = Application.get_env(:aprsme, :packet_pipeline, [])
+
+    # Number of parallel consumers (default to 3 for better throughput)
+    num_consumers = opts[:num_consumers] || config[:num_consumers] || 3
+
+    # Each consumer gets a portion of the max demand
+    max_demand_per_consumer = div(config[:max_demand] || 250, num_consumers)
+
+    children =
+      for index <- 1..num_consumers do
+        %{
+          id: {Aprsme.PacketConsumer, index},
+          start:
+            {Aprsme.PacketConsumer, :start_link,
+             [
+               [
+                 batch_size: config[:batch_size] || 100,
+                 batch_timeout: config[:batch_timeout] || 1000,
+                 max_demand: max_demand_per_consumer,
+                 subscribe_to: [{Aprsme.PacketProducer, max_demand: max_demand_per_consumer}]
+               ]
+             ]}
+        }
+      end
+
+    Supervisor.init(children, strategy: :one_for_one)
+  end
+end

lib/aprsme/packet_pipeline_supervisor.ex

@@ -12,10 +12,13 @@ defmodule Aprsme.PacketPipelineSupervisor do
   def init(_opts) do
     config = Application.get_env(:aprsme, :packet_pipeline, [])
 
-    children = [
-      {Aprsme.PacketProducer, max_buffer_size: config[:max_buffer_size] || 1000},
-      {Aprsme.PacketConsumer, batch_size: config[:batch_size] || 100, batch_timeout: config[:batch_timeout] || 1000}
-    ]
+    # Configure producer with correct buffer size from config
+    producer_spec = {Aprsme.PacketProducer, max_buffer_size: config[:max_buffer_size] || 1000}
+
+    # Use consumer pool for better throughput
+    consumer_pool_spec = {Aprsme.PacketConsumerPool, num_consumers: config[:num_consumers] || 3}
+
+    children = [producer_spec, consumer_pool_spec]
 
     Supervisor.init(children, strategy: :one_for_one)
   end