✨ Add batch_timeout_ms to webhooks

Author: acco

assets/svelte/consumers/SinkHttpPushForm.svelte

@@ -170,6 +170,29 @@
         <AccordionTrigger>Advanced configuration</AccordionTrigger>
         <AccordionContent>
           <div class="space-y-4 pt-4">
+            <div class="space-y-2">
+              <Label for="batch-timeout">Batch timeout</Label>
+              <div class="flex items-center space-x-2">
+                <Input
+                  id="batch-timeout"
+                  type="number"
+                  bind:value={form.batchTimeoutMs}
+                  class="w-24"
+                  min="1"
+                />
+                <span class="text-sm text-muted-foreground">ms</span>
+              </div>
+              <p class="text-sm text-muted-foreground">
+                The maximum time to wait for a batch to reach its full size
+                before sending. Defaults to 50ms.
+              </p>
+              {#if errors.batch_timeout_ms}
+                <p class="text-destructive text-sm">
+                  {errors.batch_timeout_ms}
+                </p>
+              {/if}
+            </div>
+
             <div class="space-y-2">
               <Label for="max-ack-pending">Max ack pending</Label>
               <Input

lib/sequin/consumers/sink_consumer.ex

@@ -48,7 +48,8 @@ defmodule Sequin.Consumers.SinkConsumer do
              :health,
              :max_memory_mb,
              :legacy_transform,
-             :timestamp_format
+             :timestamp_format,
+             :batch_timeout_ms
            ]}
   typed_schema "sink_consumers" do
     field :name, :string
@@ -61,6 +62,7 @@ defmodule Sequin.Consumers.SinkConsumer do
     field :status, Ecto.Enum, values: [:active, :disabled, :paused], default: :active
     field :seq, :integer, read_after_writes: true
     field :batch_size, :integer, default: 1
+    field :batch_timeout_ms, :integer, default: nil
     field :annotations, :map, default: %{}
     field :max_memory_mb, :integer, default: 1024
     field :partition_count, :integer, default: 1
@@ -165,7 +167,8 @@ defmodule Sequin.Consumers.SinkConsumer do
       :legacy_transform,
       :transform_id,
       :routing_id,
-      :timestamp_format
+      :timestamp_format,
+      :batch_timeout_ms
     ])
     |> cast_polymorphic_embed(:sink, required: true)
     |> Sequin.Changeset.cast_embed(:source_tables)

lib/sequin/runtime/sink_pipeline.ex

@@ -259,12 +259,13 @@ defmodule Sequin.Runtime.SinkPipeline do
 
   defp batchers_config(pipeline_mod, consumer) do
     concurrency = min(System.schedulers_online() * 2, 80)
+    batch_timeout = consumer.batch_timeout_ms || 50
 
     default = [
       default: [
         concurrency: concurrency,
         batch_size: consumer.batch_size,
-        batch_timeout: 50
+        batch_timeout: batch_timeout
       ]
     ]
 

lib/sequin_web/live/components/consumer_form.ex

@@ -540,6 +540,7 @@ defmodule SequinWeb.Components.ConsumerForm do
           "group_column_attnums" => form["groupColumnAttnums"]
         },
         "batch_size" => form["batchSize"],
+        "batch_timeout_ms" => form["batchTimeoutMs"],
         "initial_backfill" => decode_initial_backfill(form),
         "transform_id" => if(form["transform"] === "none", do: nil, else: form["transform"]),
         "routing_id" => if(form["routingId"] === "none", do: nil, else: form["routingId"]),
@@ -734,7 +735,8 @@ defmodule SequinWeb.Components.ConsumerForm do
       "id" => consumer.id,
       "name" => consumer.name || Name.generate(999),
       "ack_wait_ms" => consumer.ack_wait_ms,
-      "batch_size" => Map.get(consumer, :batch_size),
+      "batch_size" => consumer.batch_size,
+      "batch_timeout_ms" => consumer.batch_timeout_ms,
       "max_memory_mb" => consumer.max_memory_mb,
       "group_column_attnums" => source_table && source_table.group_column_attnums,
       "max_ack_pending" => consumer.max_ack_pending,

priv/repo/migrations/20250430021426_add_batch_timeout_ms_to_sink_consumers.exs

@@ -0,0 +1,11 @@
+defmodule Sequin.Repo.Migrations.AddBatchTimeoutMsToSinkConsumers do
+  use Ecto.Migration
+
+  @config_schema Application.compile_env(:sequin, [Sequin.Repo, :config_schema_prefix])
+
+  def change do
+    alter table(:sink_consumers, prefix: @config_schema) do
+      add :batch_timeout_ms, :integer, default: nil
+    end
+  end
+end