Add shutdown_timeout configuration option

.tool-versions

@@ -1,2 +1,2 @@
-erlang 27.2
+erlang 27.3
 elixir 1.18.0-otp-27

lib/data_buffer.ex

@@ -64,6 +64,7 @@ defmodule DataBuffer do
   - `:flush_jitter` - Random jitter added to flush_interval (default: 2000)
   - `:flush_timeout` - Timeout in ms for flush operations (default: 60000)
   - `:flush_meta` - Metadata passed to handle_flush callback (optional)
+  - `:shutdown_timeout` - Timeout in ms or `:infinity` for shutdown flush operations (default: 5000)
 
   ## Telemetry Events
 
@@ -297,6 +298,7 @@ defmodule DataBuffer do
   - `:flush_interval` - Time interval for automatic flush
   - `:flush_jitter` - Random jitter for flush interval
   - `:flush_timeout` - Timeout for flush operations
+  - `:shutdown_timeout` - Timeout for shutdown flush operations
 
   ## Examples
 

lib/data_buffer/partition.ex

@@ -19,6 +19,7 @@ defmodule DataBuffer.Partition do
       :flush_interval,
       :flush_jitter,
       :flush_timeout,
+      :shutdown_timeout,
       :flush_opts,
       :flush_complete_ref,
       :flush_schedule_ref,
@@ -42,7 +43,8 @@ defmodule DataBuffer.Partition do
                  flush_interval: [is: :integer, default: 10_000, required: true],
                  flush_jitter: [is: :integer, default: 2_000, required: true],
                  flush_meta: [is: :any, required: false],
-                 flush_timeout: [is: :timeout, default: 60_000, required: true]
+                 flush_timeout: [is: :timeout, default: 60_000, required: true],
+                 shutdown_timeout: [is: :timeout, default: 5_000, required: true]
                )
 
   ################################
@@ -53,7 +55,8 @@ defmodule DataBuffer.Partition do
   def child_spec(opts) do
     %{
       id: Keyword.fetch!(opts, :name),
-      start: {__MODULE__, :start_link, [opts]}
+      start: {__MODULE__, :start_link, [opts]},
+      shutdown: Keyword.get(opts, :shutdown_timeout, 5_000)
     }
   end
 
@@ -192,6 +195,7 @@ defmodule DataBuffer.Partition do
       flush_interval: Keyword.get(opts, :flush_interval),
       flush_jitter: Keyword.get(opts, :flush_jitter),
       flush_timeout: Keyword.get(opts, :flush_timeout),
+      shutdown_timeout: Keyword.get(opts, :shutdown_timeout),
       flush_opts: [
         meta: Keyword.get(opts, :flush_meta)
       ],
@@ -369,6 +373,7 @@ defmodule DataBuffer.Partition do
       flush_interval: state.flush_interval,
       flush_jitter: state.flush_jitter,
       flush_timeout: state.flush_timeout,
+      shutdown_timeout: state.shutdown_timeout,
       pid: self()
     }
   end

mix.exs

@@ -1,7 +1,7 @@
 defmodule DataBuffer.MixProject do
   use Mix.Project
 
-  @version "1.0.0"
+  @version "1.1.0"
 
   def project do
     [

test/data_buffer_test.exs

@@ -203,6 +203,43 @@ defmodule DataBufferTest do
            end) =~ "DataBuffer: flush timeout error"
   end
 
+  test "flushes data on shutdown" do
+    start_buffer(partitions: 1)
+    DataBuffer.insert(TestBuffer, "foo")
+    stop_supervised!(TestBuffer)
+    assert_receive {:data, ["foo"], _}
+  end
+
+  test "configures default and custom shutdown_timeout on partition" do
+    start_buffer(partitions: 1)
+    [partition] = DataBuffer.info(TestBuffer)
+    assert partition.shutdown_timeout == 5_000
+
+    spec =
+      DataBuffer.Partition.child_spec(
+        name: :test_part,
+        buffer: TestBuffer,
+        shutdown_timeout: 12_345
+      )
+
+    assert spec.shutdown == 12_345
+
+    default_spec = DataBuffer.Partition.child_spec(name: :test_part, buffer: TestBuffer)
+    assert default_spec.shutdown == 5_000
+
+    stop_supervised!(TestBuffer)
+
+    start_buffer(partitions: 1, shutdown_timeout: 10_000)
+    [partition] = DataBuffer.info(TestBuffer)
+    assert partition.shutdown_timeout == 10_000
+    stop_supervised!(TestBuffer)
+
+    start_buffer(partitions: 1, shutdown_timeout: :infinity)
+    [partition] = DataBuffer.info(TestBuffer)
+    assert partition.shutdown_timeout == :infinity
+    stop_supervised!(TestBuffer)
+  end
+
   defp receive_all(partitions \\ partitions()) do
     for _ <- 1..partitions, reduce: [] do
       data ->