⚡️ Update defaults for processor/batching configuration

Author: acco

lib/sequin/runtime/azure_event_hub_pipeline.ex

@@ -16,21 +16,13 @@ defmodule Sequin.Runtime.AzureEventHubPipeline do
   end
 
   @impl SinkPipeline
-  def processors_config(%SinkConsumer{max_waiting: max_waiting}) do
-    [
-      default: [
-        concurrency: max_waiting,
-        max_demand: 100
-      ]
-    ]
-  end
+  def batchers_config(_consumer) do
+    concurrency = min(System.schedulers_online() * 2, 80)
 
-  @impl SinkPipeline
-  def batchers_config(%SinkConsumer{batch_size: batch_size}) do
     [
       default: [
-        concurrency: 1,
-        batch_size: batch_size,
+        concurrency: concurrency,
+        batch_size: 10,
         batch_timeout: 50
       ]
     ]

lib/sequin/runtime/gcp_pubsub_pipeline.ex

@@ -15,22 +15,14 @@ defmodule Sequin.Runtime.GcpPubsubPipeline do
     Map.put(context, :pubsub_client, GcpPubsubSink.pubsub_client(sink))
   end
 
-  @impl SinkPipeline
-  def processors_config(%SinkConsumer{max_waiting: max_waiting}) do
-    [
-      default: [
-        concurrency: max_waiting,
-        max_demand: 100
-      ]
-    ]
-  end
-
   @impl SinkPipeline
   def batchers_config(_consumer) do
+    concurrency = min(System.schedulers_online() * 2, 80)
+
     [
       default: [
-        concurrency: 100,
-        batch_size: 100,
+        concurrency: concurrency,
+        batch_size: 10,
         batch_timeout: 50
       ]
     ]

lib/sequin/runtime/http_push_pipeline.ex

@@ -25,28 +25,6 @@ defmodule Sequin.Runtime.HttpPushPipeline do
     |> Map.put(:features, features)
   end
 
-  @impl SinkPipeline
-  def processors_config(%SinkConsumer{max_waiting: max_waiting}) do
-    [
-      default: [
-        concurrency: max_waiting,
-        max_demand: 10,
-        min_demand: 5
-      ]
-    ]
-  end
-
-  @impl SinkPipeline
-  def batchers_config(%SinkConsumer{batch_size: batch_size}) do
-    [
-      default: [
-        concurrency: 1,
-        batch_size: batch_size,
-        batch_timeout: 50
-      ]
-    ]
-  end
-
   @impl SinkPipeline
   def handle_batch(:default, messages, _batch_info, context) do
     %{

lib/sequin/runtime/nats_pipeline.ex

@@ -2,7 +2,6 @@ defmodule Sequin.Runtime.NatsPipeline do
   @moduledoc false
   @behaviour Sequin.Runtime.SinkPipeline
 
-  alias Sequin.Consumers.SinkConsumer
   alias Sequin.Error
   alias Sequin.Runtime.SinkPipeline
   alias Sequin.Sinks.Nats
@@ -15,22 +14,13 @@ defmodule Sequin.Runtime.NatsPipeline do
   end
 
   @impl SinkPipeline
-  def processors_config(%SinkConsumer{max_waiting: max_waiting}) do
-    [
-      default: [
-        concurrency: max_waiting,
-        max_demand: 10,
-        min_demand: 5
-      ]
-    ]
-  end
+  def batchers_config(_consumer) do
+    concurrency = min(System.schedulers_online() * 2, 80)
 
-  @impl SinkPipeline
-  def batchers_config(%SinkConsumer{batch_size: batch_size}) do
     [
       default: [
-        concurrency: 1,
-        batch_size: batch_size,
+        concurrency: concurrency,
+        batch_size: 10,
         batch_timeout: 50
       ]
     ]

lib/sequin/runtime/rabbitmq_pipeline.ex

@@ -2,7 +2,6 @@ defmodule Sequin.Runtime.RabbitMqPipeline do
   @moduledoc false
   @behaviour Sequin.Runtime.SinkPipeline
 
-  alias Sequin.Consumers.SinkConsumer
   alias Sequin.Error
   alias Sequin.Runtime.SinkPipeline
   alias Sequin.Sinks.RabbitMq
@@ -15,22 +14,13 @@ defmodule Sequin.Runtime.RabbitMqPipeline do
   end
 
   @impl SinkPipeline
-  def processors_config(%SinkConsumer{max_waiting: max_waiting}) do
-    [
-      default: [
-        concurrency: max_waiting,
-        max_demand: 10,
-        min_demand: 5
-      ]
-    ]
-  end
+  def batchers_config(_consumer) do
+    concurrency = min(System.schedulers_online() * 2, 80)
 
-  @impl SinkPipeline
-  def batchers_config(%SinkConsumer{batch_size: batch_size}) do
     [
       default: [
-        concurrency: 1,
-        batch_size: batch_size,
+        concurrency: concurrency,
+        batch_size: 10,
         batch_timeout: 50
       ]
     ]

lib/sequin/runtime/redis_pipeline.ex

@@ -14,28 +14,6 @@ defmodule Sequin.Runtime.RedisPipeline do
     context
   end
 
-  @impl SinkPipeline
-  def processors_config(%SinkConsumer{max_waiting: max_waiting}) do
-    [
-      default: [
-        concurrency: max_waiting,
-        max_demand: 10,
-        min_demand: 5
-      ]
-    ]
-  end
-
-  @impl SinkPipeline
-  def batchers_config(%SinkConsumer{batch_size: batch_size}) do
-    [
-      default: [
-        concurrency: 1,
-        batch_size: batch_size,
-        batch_timeout: 50
-      ]
-    ]
-  end
-
   @impl SinkPipeline
   def handle_batch(:default, messages, _batch_info, context) do
     %{consumer: %SinkConsumer{sink: %RedisSink{} = sink}, test_pid: test_pid} = context

lib/sequin/runtime/sink_pipeline.ex

@@ -188,7 +188,7 @@ defmodule Sequin.Runtime.SinkPipeline do
   defp processors_config(pipeline_mod, consumer) do
     default = [
       default: [
-        concurrency: 1,
+        concurrency: System.schedulers_online(),
         max_demand: 10,
         min_demand: 5
       ]
@@ -202,11 +202,13 @@ defmodule Sequin.Runtime.SinkPipeline do
   end
 
   defp batchers_config(pipeline_mod, consumer) do
+    concurrency = min(System.schedulers_online() * 2, 80)
+
     default = [
       default: [
-        concurrency: 1,
-        batch_size: 1,
-        batch_timeout: 0
+        concurrency: concurrency,
+        batch_size: consumer.batch_size,
+        batch_timeout: 50
       ]
     ]
 

lib/sequin/runtime/sqs_pipeline.ex

@@ -16,22 +16,13 @@ defmodule Sequin.Runtime.SqsPipeline do
   end
 
   @impl SinkPipeline
-  def processors_config(%SinkConsumer{max_waiting: max_waiting}) do
-    [
-      default: [
-        concurrency: max_waiting,
-        max_demand: 10,
-        min_demand: 5
-      ]
-    ]
-  end
+  def batchers_config(_consumer) do
+    concurrency = min(System.schedulers_online() * 2, 80)
 
-  @impl SinkPipeline
-  def batchers_config(%SinkConsumer{batch_size: batch_size}) do
     [
       default: [
-        concurrency: 1,
-        batch_size: batch_size,
+        concurrency: concurrency,
+        batch_size: 10,
         batch_timeout: 50
       ]
     ]