Repository: honeybadger-io/honeybadger-elixir
Pull request #578 (2025-02-09 22:47:55)

Insights event batching
This includes an `EventsWorker` GenServer. 

It bundles up sent events into batches and transmits them to the backend if they hit an event threshold (`events_batch_size`) or a time threshold (`events_timeout`). It will drop events if the queue is too full (`events_max_queue_size`).

This implementation differs a little from our Ruby implementation:

* It attempts to retry batches of events if the server does not respond, or responds with unexpected errors. It drops the batch after n attempts (`events_max_batch_retries`)
* It can have multiple concurrently retrying batches, but events will be dropped if the max queue size is it (including events already batched)
* It has a static timeout value if the backend responds with a 429. This is an attempt to make sure that we don't send events too often (currently defaulted to 60s) if we know the server will not accept them.

This is ready for a review, but I would like to add a little more testing around the Client.


Patch:
diff --git a/lib/honeybadger.ex b/lib/honeybadger.ex
index 7d13769..fb42477 100644
--- a/lib/honeybadger.ex
+++ b/lib/honeybadger.ex
@@ -169,7 +169,7 @@ defmodule Honeybadger do
 
   require Logger
 
-  alias Honeybadger.{Client, Notice}
+  alias Honeybadger.{Client, Notice, EventsWorker}
   alias Honeybadger.Breadcrumbs.{Collector, Breadcrumb}
 
   @type notify_options ::
@@ -206,7 +206,9 @@ defmodule Honeybadger do
       Honeybadger.Breadcrumbs.Telemetry.attach()
     end
 
-    Supervisor.start_link([{Client, [config]}], strategy: :one_for_one)
+    children = [{Client, [config]}, EventsWorker]
+
+    Supervisor.start_link(children, strategy: :one_for_one)
   end
 
   @doc """
@@ -345,9 +347,13 @@ defmodule Honeybadger do
   def event(event_data) do
     ts = DateTime.utc_now() |> DateTime.to_string()
 
-    event_data
-    |> Map.put_new(:ts, ts)
-    |> Client.send_event()
+    data = Map.put_new(event_data, :ts, ts)
+
+    if get_env(:events_worker_enabled) do
+      EventsWorker.push(data)
+    else
+      Client.send_event(data)
+    end
   end
 
   @doc """
diff --git a/lib/honeybadger/client.ex b/lib/honeybadger/client.ex
index 1e5c976..4dc092e 100644
--- a/lib/honeybadger/client.ex
+++ b/lib/honeybadger/client.ex
@@ -90,6 +90,23 @@ defmodule Honeybadger.Client do
     end
   end
 
+  @doc """
+  Send events in batches
+  """
+  @spec send_events(list) :: :ok | {:error, reason :: atom()}
+  def send_events(events) when is_list(events) do
+    if pid = Process.whereis(__MODULE__) do
+      # 30 second timeout
+      GenServer.call(pid, {:events, events}, 30_000)
+    else
+      Logger.warning(fn ->
+        "[Honeybadger] Unable to post events, the :honeybadger client isn't running"
+      end)
+
+      {:error, :client_not_running}
+    end
+  end
+
   @doc """
   Check whether reporting is enabled for the current environment.
 
@@ -148,7 +165,7 @@ defmodule Honeybadger.Client do
           |> Map.get(:hackney_opts)
           |> Keyword.merge(opts)
 
-        post_notice(url, headers, payload, hackney_opts)
+        post_payload(url, headers, payload, hackney_opts)
 
       {:error, %Jason.EncodeError{message: message}} ->
         Logger.warning(fn -> "[Honeybadger] Notice encoding failed: #{message}" end)
@@ -186,7 +203,7 @@ defmodule Honeybadger.Client do
           |> Keyword.merge(opts)
 
         # post logic for events is the same as notices
-        post_notice(event_url, headers, payload, hackney_opts)
+        post_payload(event_url, headers, payload, hackney_opts)
 
       {:error, %Jason.EncodeError{message: message}} ->
         Logger.warning(fn -> "[Honeybadger] Event encoding failed: #{message}" end)
@@ -205,13 +222,72 @@ defmodule Honeybadger.Client do
     {:noreply, state}
   end
 
+  # Events
+  #
+  @impl GenServer
+  def handle_call({:events, _}, _from, %{enabled: false} = state) do
+    {:reply, :ok, state}
+  end
+
+  def handle_call({:events, _}, _from, %{api_key: nil} = state) do
+    {:reply, {:error, :no_api_key}, state}
+  end
+
+  def handle_call({:events, events}, _from, state) do
+    %{event_url: event_url, headers: headers} = state
+    # Convert each event to JSON and join with newlines
+    encoded_events = Enum.map(events, &Honeybadger.JSON.encode/1)
+
+    # Check if any encoding failed
+    if Enum.any?(encoded_events, &match?({:error, _}, &1)) do
+      {:reply, {:error, :encoding_error}, state}
+    else
+      payload = Enum.map_join(encoded_events, "\n", &elem(&1, 1))
+
+      opts =
+        state
+        |> Map.take([:proxy, :proxy_auth])
+        |> Map.put(:pool, __MODULE__)
+        |> Keyword.new()
+
+      hackney_opts =
+        state
+        |> Map.get(:hackney_opts)
+        |> Keyword.merge(opts)
+
+      headers =
+        List.keyreplace(headers, "Content-Type", 0, {"Content-Type", "application/x-ndjson"})
+
+      response =
+        case HTTPAdapter.request(:post, event_url, payload, headers, hackney_opts) do
+          {:ok, %HTTPResponse{body: body, status: status}} when status in 200..399 ->
+            Logger.debug(fn -> "[Honeybadger] API success: #{inspect(body)}" end)
+            :ok
+
+          {:ok, %HTTPResponse{status: status}} when status == 429 ->
+            Logger.warning("[Honeybadger] API rate limited:")
+            {:error, :throttled}
+
+          {:ok, %HTTPResponse{status: status}} when status in 400..599 ->
+            Logger.warning("[Honeybadger] API failure")
+            {:error, :api_error}
+
+          {:error, reason} ->
+            Logger.warning(fn -> "[Honeybadger] connection error: #{inspect(reason)}" end)
+            {:error, :connection_error}
+        end
+
+      {:reply, response, state}
+    end
+  end
+
   # API Integration
 
   defp build_headers(opts) do
     [{"X-API-Key", get_env(opts, :api_key)}] ++ @headers
   end
 
-  defp post_notice(url, headers, payload, hackney_opts) do
+  defp post_payload(url, headers, payload, hackney_opts) do
     case HTTPAdapter.request(:post, url, payload, headers, hackney_opts) do
       {:ok, %HTTPResponse{body: body, status: status}} when status in 200..399 ->
         Logger.debug(fn -> "[Honeybadger] API success: #{inspect(body)}" end)
diff --git a/lib/honeybadger/events_worker.ex b/lib/honeybadger/events_worker.ex
new file mode 100644
index 0000000..433b15a
--- /dev/null
+++ b/lib/honeybadger/events_worker.ex
@@ -0,0 +1,242 @@
+defmodule Honeybadger.EventsWorker do
+  @moduledoc """
+  A GenServer that batches and sends events with retry and throttling logic.
+
+  It accumulates events in a queue, forms batches when the batch size is reached or
+  when a flush timeout expires, and then sends these batches to a backend module.
+  If a batch fails to send, it will be retried (up to a configurable maximum) or dropped.
+  In case of throttling (e.g. receiving a 429), the flush delay is increased.
+  """
+
+  @dropped_log_interval 60_000
+
+  use GenServer
+  require Logger
+
+  defmodule State do
+    @typedoc """
+    Function that accepts a list of events to be processed.
+    """
+    @type send_events_fn :: ([term()] -> :ok | {:error, :throttled} | {:error, term()})
+
+    @typedoc """
+    State for the event batching GenServer.
+    """
+    @type t :: %__MODULE__{
+            # Configuration
+            send_events_fn: send_events_fn(),
+            batch_size: pos_integer(),
+            max_queue_size: pos_integer(),
+            timeout: pos_integer(),
+            max_batch_retries: non_neg_integer(),
+            throttle_wait: pos_integer(),
+
+            # Internal state
+            timeout_started_at: non_neg_integer(),
+            throttling: boolean(),
+            dropped_events: non_neg_integer(),
+            last_dropped_log: non_neg_integer(),
+            queue: [any()],
+            batches: :queue.queue()
+          }
+
+    @enforce_keys [
+      :send_events_fn,
+      :batch_size,
+      :max_queue_size,
+      :max_batch_retries
+    ]
+
+    defstruct [
+      :send_events_fn,
+      :batch_size,
+      :max_queue_size,
+      :timeout,
+      :max_batch_retries,
+      :last_dropped_log,
+      timeout_started_at: 0,
+      throttle_wait: 60000,
+      throttling: false,
+      dropped_events: 0,
+      queue: [],
+      batches: :queue.new()
+    ]
+  end
+
+  @spec start_link(Keyword.t()) :: GenServer.on_start()
+  def start_link(opts \\ []) do
+    if Honeybadger.get_env(:events_worker_enabled) do
+      {name, opts} = Keyword.pop(opts, :name, __MODULE__)
+      GenServer.start_link(__MODULE__, opts, name: name)
+    else
+      :ignore
+    end
+  end
+
+  @spec push(event :: map(), GenServer.server()) :: :ok
+  def push(event, server \\ __MODULE__) do
+    GenServer.cast(server, {:push, event})
+  end
+
+  @spec state(GenServer.server()) :: State.t()
+  def state(server \\ __MODULE__) do
+    GenServer.call(server, {:state})
+  end
+
+  @impl true
+  def init(opts) do
+    config = %{
+      send_events_fn: Keyword.get(opts, :send_events_fn, &Honeybadger.Client.send_events/1),
+      batch_size: Keyword.get(opts, :batch_size, Honeybadger.get_env(:events_batch_size)),
+      timeout: Keyword.get(opts, :timeout, Honeybadger.get_env(:events_timeout)),
+      throttle_wait:
+        Keyword.get(opts, :throttle_wait, Honeybadger.get_env(:events_throttle_wait)),
+      max_queue_size:
+        Keyword.get(opts, :max_queue_size, Honeybadger.get_env(:events_max_queue_size)),
+      max_batch_retries:
+        Keyword.get(opts, :max_batch_retries, Honeybadger.get_env(:events_max_batch_retries)),
+      last_dropped_log: System.monotonic_time(:millisecond)
+    }
+
+    state = struct!(State, config)
+    {:ok, state}
+  end
+
+  @impl true
+  def handle_call({:state}, _from, %State{} = state) do
+    {:reply, state, state, current_timeout(state)}
+  end
+
+  @impl true
+  def handle_cast({:push, event}, %State{timeout_started_at: 0} = state) do
+    handle_cast({:push, event}, reset_timeout(state))
+  end
+
+  def handle_cast({:push, event}, %State{} = state) do
+    if total_event_count(state) >= state.max_queue_size do
+      {:noreply, %{state | dropped_events: state.dropped_events + 1}, current_timeout(state)}
+    else
+      queue = [event | state.queue]
+
+      if length(queue) >= state.batch_size do
+        flush(%{state | queue: queue})
+      else
+        {:noreply, %{state | queue: queue}, current_timeout(state)}
+      end
+    end
+  end
+
+  @impl true
+  def handle_info(:timeout, state), do: flush(state)
+
+  @impl true
+  def terminate(_reason, %State{} = state) do
+    Logger.debug("[Honeybadger] Terminating with #{total_event_count(state)} events unsent")
+    _ = flush(state)
+    :ok
+  end
+
+  @spec flush(State.t()) :: {:noreply, State.t(), pos_integer()}
+  defp flush(state) do
+    cond do
+      state.queue == [] and :queue.is_empty(state.batches) ->
+        # It's all empty so we stop the timeout and reset the
+        # timeout_started_at which will restart on the next push
+        {:noreply, %{state | timeout_started_at: 0}}
+
+      state.queue == [] ->
+        attempt_send(state)
+
+      true ->
+        batches = :queue.in(%{batch: Enum.reverse(state.queue), attempts: 0}, state.batches)
+        attempt_send(%{state | queue: [], batches: batches})
+    end
+  end
+
+  @spec attempt_send(State.t()) :: {:noreply, State.t(), pos_integer()}
+  # Sends pending batches, handling retries and throttling
+  defp attempt_send(%State{} = state) do
+    {new_batches_list, throttling} =
+      Enum.reduce(:queue.to_list(state.batches), {[], false}, fn
+        # If already throttled, skip sending and retain the batch.
+        b, {acc, true} ->
+          {acc ++ [b], true}
+
+        %{batch: batch, attempts: attempts} = b, {acc, false} ->
+          case state.send_events_fn.(batch) do
+            :ok ->
+              Logger.debug("[Honeybadger] Sent batch of #{length(batch)} events.")
+              {acc, false}
+
+            {:error, reason} ->
+              throttling = reason == :throttled
+              updated_attempts = attempts + 1
+
+              if throttling do
+                Logger.warning(
+                  "[Honeybadger] Rate limited (429) events - (batch attempt #{updated_attempts}) - waiting for #{state.throttle_wait}ms"
+                )
+              else
+                Logger.debug(
+                  "[Honeybadger] Failed to send events batch (attempt #{updated_attempts}): #{inspect(reason)}"
+                )
+              end
+
+              if updated_attempts < state.max_batch_retries do
+                {acc ++ [%{b | attempts: updated_attempts}], throttling}
+              else
+                Logger.debug(
+                  "[Honeybadger] Dropping events batch after #{updated_attempts} attempts."
+                )
+
+                {acc, throttling}
+              end
+          end
+      end)
+
+    current_time = System.monotonic_time(:millisecond)
+
+    # Log dropped events if present and we haven't logged within the last
+    # @dropped_log_interval
+    state =
+      if state.dropped_events > 0 and
+           current_time - state.last_dropped_log >= @dropped_log_interval do
+        Logger.info("[Honeybadger] Dropped #{state.dropped_events} events due to max queue limit")
+        %{state | dropped_events: 0, last_dropped_log: current_time}
+      else
+        state
+      end
+
+    new_state =
+      %{state | batches: :queue.from_list(new_batches_list), throttling: throttling}
+      |> reset_timeout()
+
+    {:noreply, new_state, current_timeout(new_state)}
+  end
+
+  @spec total_event_count(State.t()) :: non_neg_integer()
+  # Counts events in both the queue and pending batches.
+  defp total_event_count(%State{batches: batches, queue: queue}) do
+    events_count = length(queue)
+
+    batch_count = :queue.fold(fn %{batch: b}, acc -> acc + length(b) end, 0, batches)
+
+    events_count + batch_count
+  end
+
+  # Returns the time remaining until the next flush
+  defp current_timeout(%State{
+         throttling: throttling,
+         timeout: timeout,
+         throttle_wait: throttle_wait,
+         timeout_started_at: timeout_started_at
+       }) do
+    elapsed = System.monotonic_time(:millisecond) - timeout_started_at
+    timeout = if throttling, do: throttle_wait, else: timeout
+    max(1, timeout - elapsed)
+  end
+
+  defp reset_timeout(state) do
+    %{state | timeout_started_at: System.monotonic_time(:millisecond)}
+  end
+end
diff --git a/lib/honeybadger/http_adapter.ex b/lib/honeybadger/http_adapter.ex
index 2be19de..63b4d36 100644
--- a/lib/honeybadger/http_adapter.ex
+++ b/lib/honeybadger/http_adapter.ex
@@ -92,8 +92,7 @@ defmodule Honeybadger.HTTPAdapter do
 
     with {:ok, response} <- http_adapter.request(method, url, body, headers, http_adapter_opts),
          {:ok, decoded_response} <- http_adapter.decode_response_body(response, opts) do
-      {request_result(decoded_response),
-       %{decoded_response | http_adapter: http_adapter, request_url: url}}
+      {:ok, %{decoded_response | http_adapter: http_adapter, request_url: url}}
     else
       {:error, %Honeybadger.InvalidResponseError{} = error} ->
         {:error, error}
@@ -108,9 +107,6 @@ defmodule Honeybadger.HTTPAdapter do
     end
   end
 
-  defp request_result(%{status: status}) when status in 200..399, do: :ok
-  defp request_result(%{status: status}) when status in 400..599, do: :error
-
   defp get_adapter(opts) do
     default_http_adapter = Application.get_env(:honeybadger, :http_adapter, installed_adapter())
 
diff --git a/mix.exs b/mix.exs
index d9cf2ea..a4f7a32 100644
--- a/mix.exs
+++ b/mix.exs
@@ -68,6 +68,9 @@ defmodule Honeybadger.Mixfile do
         hackney_opts: [],
         use_logger: true,
         ignored_domains: [:cowboy],
+        exclude_errors: [],
+
+        # Filters
         notice_filter: Honeybadger.NoticeFilter.Default,
         filter: Honeybadger.Filter.Default,
         filter_keys: [:password, :credit_card],
@@ -75,7 +78,14 @@ defmodule Honeybadger.Mixfile do
         filter_disable_url: false,
         filter_disable_params: false,
         filter_disable_session: false,
-        exclude_errors: []
+
+        # Events
+        events_worker_enabled: true,
+        events_max_batch_retries: 3,
+        events_batch_size: 1000,
+        events_max_queue_size: 10000,
+        events_timeout: 30000,
+        events_throttle_wait: 60000
       ],
       mod: {Honeybadger, []}
     ]


Test patch:
diff --git a/test/honeybadger/events_worker_test.exs b/test/honeybadger/events_worker_test.exs
new file mode 100644
index 0000000..2d75dc3
--- /dev/null
+++ b/test/honeybadger/events_worker_test.exs
@@ -0,0 +1,297 @@
+defmodule Honeybadger.EventsWorkerTest do
+  use Honeybadger.Case, async: true
+  require Logger
+
+  alias Honeybadger.EventsWorker
+
+  defp start_worker(config) do
+    name =
+      "test_events_worker_#{System.unique_integer([:positive])}"
+      |> String.to_atom()
+
+    EventsWorker.start_link(config ++ [name: name])
+  end
+
+  setup do
+    {:ok, behavior_agent} = Agent.start_link(fn -> :ok end)
+
+    test_pid = self()
+
+    send_events_fn = fn events ->
+      send(test_pid, {:events_sent, events})
+
+      case Agent.get(behavior_agent, & &1) do
+        :ok -> :ok
+        :throttle -> {:error, :throttled}
+        :error -> {:error, "Other error"}
+      end
+    end
+
+    change_behavior = fn new_behavior ->
+      Agent.update(behavior_agent, fn _ -> new_behavior end)
+    end
+
+    # Common test configuration
+    config = [
+      send_events_fn: send_events_fn,
+      batch_size: 3,
+      max_batch_retries: 2,
+      max_queue_size: 10,
+      timeout: 100
+    ]
+
+    {:ok, config: config, change_behavior: change_behavior}
+  end
+
+  describe "batch size triggering" do
+    test "sends events when batch size is reached", %{config: config} do
+      {:ok, pid} = start_worker(config)
+      events = [%{id: 1}, %{id: 2}, %{id: 3}]
+      Enum.each(events, &EventsWorker.push(&1, pid))
+      GenServer.stop(pid)
+
+      assert_receive {:events_sent, ^events}, 50
+    end
+
+    test "queues events when under batch size", %{config: config} do
+      {:ok, pid} = start_worker(config)
+
+      events = [%{id: 1}, %{id: 2}]
+      Enum.each(events, &EventsWorker.push(&1, pid))
+      refute_receive {:events_sent, _}, 50
+
+      GenServer.stop(pid)
+    end
+  end
+
+  describe "max_queue_size" do
+    test "drops events from queue", %{config: config, change_behavior: change_behavior} do
+      {:ok, pid} =
+        start_worker(Keyword.merge(config, timeout: 5000, max_queue_size: 4))
+
+      change_behavior.(:error)
+      events = [%{id: 1}, %{id: 2}, %{id: 3}, %{id: 4}, %{id: 5}, %{id: 6}]
+      Enum.each(events, &EventsWorker.push(&1, pid))
+
+      state = EventsWorker.state(pid)
+
+      assert state.dropped_events == 2
+      assert state.queue == [%{id: 4}]
+
+      assert :queue.to_list(state.batches) == [
+               %{attempts: 1, batch: [%{id: 1}, %{id: 2}, %{id: 3}]}
+             ]
+
+      GenServer.stop(pid)
+    end
+  end
+
+  describe "timer triggering" do
+    test "flushes events when timer expires", %{config: config} do
+      {:ok, pid} = start_worker(config)
+      events = [%{id: 1}, %{id: 2}]
+      Enum.each(events, &EventsWorker.push(&1, pid))
+      assert_receive {:events_sent, ^events}, config[:timeout] + 50
+      GenServer.stop(pid)
+    end
+
+    test "resets timer when batch is sent", %{config: config} do
+      {:ok, pid} = start_worker(config)
+
+      first_batch = [%{id: 1}, %{id: 2}, %{id: 3}]
+      Enum.each(first_batch, &EventsWorker.push(&1, pid))
+      assert_receive {:events_sent, ^first_batch}, 50
+
+      second_batch = [%{id: 4}, %{id: 5}]
+      Enum.each(second_batch, &EventsWorker.push(&1, pid))
+      refute_receive {:events_sent, _}, 50
+      assert_receive {:events_sent, ^second_batch}, 100
+
+      GenServer.stop(pid)
+    end
+  end
+
+  describe "error handling" do
+    setup %{config: config} do
+      config =
+        Keyword.merge(config,
+          batch_size: 2,
+          timeout: 100,
+          throttle_wait: 300,
+          max_queue_size: 10000
+        )
+
+      {:ok, config: config}
+    end
+
+    test "retries and drops after max attempts", %{
+      config: config,
+      change_behavior: change_behavior
+    } do
+      config = Keyword.merge(config, max_batch_retries: 3)
+
+      {:ok, pid} = start_worker(config)
+
+      # Start with error behavior
+      change_behavior.(:error)
+
+      # Send enough events to trigger a batch
+      events = [%{id: 1}, %{id: 2}]
+      Enum.each(events, &EventsWorker.push(&1, pid))
+
+      # Wait for first attempt
+      assert_receive {:events_sent, ^events}, 50
+
+      # Should retry after timeout
+      assert_receive {:events_sent, ^events}, 150
+
+      # Should retry one more time and then drop
+      assert_receive {:events_sent, ^events}, 250
+
+      # Check final state
+      state = EventsWorker.state(pid)
+      # Batch should be dropped
+      assert :queue.to_list(state.batches) == []
+
+      GenServer.stop(pid)
+    end
+
+    test "queues new events during retry attempts", %{
+      config: config,
+      change_behavior: change_behavior
+    } do
+      {:ok, pid} = start_worker(config)
+
+      change_behavior.(:error)
+
+      # Send first batch
+      first_batch = [%{id: 1}, %{id: 2}]
+      Enum.each(first_batch, &EventsWorker.push(&1, pid))
+
+      # Wait for first attempt
+      assert_receive {:events_sent, ^first_batch}, 50
+
+      # Send new events during retry period
+      second_batch = [%{id: 3}, %{id: 4}]
+      Enum.each(second_batch, &EventsWorker.push(&1, pid))
+
+      # Switch to success before max retries
+      change_behavior.(:ok)
+
+      # Should eventually send both batches
+      assert_receive {:events_sent, ^first_batch}, 150
+      assert_receive {:events_sent, ^second_batch}, 50
+
+      GenServer.stop(pid)
+    end
+
+    test "does not reset flush timer on subsequent pushes", %{config: config} do
+      {:ok, pid} =
+        start_worker(Keyword.merge(config, timeout: 100, batch_size: 1000))
+
+      EventsWorker.push(%{id: 1}, pid)
+      :timer.sleep(60)
+      EventsWorker.push(%{id: 2}, pid)
+      :timer.sleep(60)
+      EventsWorker.push(%{id: 3}, pid)
+
+      assert_receive {:events_sent, [%{id: 1}, %{id: 2}]}
+      assert_receive {:events_sent, [%{id: 3}]}, 100
+
+      GenServer.stop(pid)
+    end
+
+    test "works with pushes after a flush", %{config: config} do
+      {:ok, pid} =
+        start_worker(Keyword.merge(config, timeout: 50, batch_size: 1000))
+
+      EventsWorker.push(%{id: 1}, pid)
+      :timer.sleep(300)
+      EventsWorker.push(%{id: 2}, pid)
+
+      assert_receive {:events_sent, [%{id: 1}]}, 0
+      # Make sure we don't get the second event before the timeout
+      refute_receive {:events_sent, [%{id: 2}]}, 0
+      assert_receive {:events_sent, [%{id: 2}]}, 100
+
+      GenServer.stop(pid)
+    end
+
+    test "handles throttling and resumes after wait period", %{
+      config: config,
+      change_behavior: change_behavior
+    } do
+      {:ok, pid} = start_worker(config)
+
+      # Start with throttle behavior
+      change_behavior.(:throttle)
+
+      # Send first batch
+      first_batch = [%{id: 1}, %{id: 2}]
+      Enum.each(first_batch, &EventsWorker.push(&1, pid))
+
+      # Should get throttled
+      assert_receive {:events_sent, ^first_batch}, 50
+
+      # Send second batch while throttled
+      second_batch = [%{id: 3}, %{id: 4}]
+      Enum.each(second_batch, &EventsWorker.push(&1, pid))
+
+      # Switch to success after throttle period
+      change_behavior.(:ok)
+
+      # Should send both batches after throttle period
+      assert_receive {:events_sent, ^first_batch}, 350
+      assert_receive {:events_sent, ^second_batch}, 50
+
+      GenServer.stop(pid)
+    end
+
+    test "flush delay respects throttle_wait when throttled", %{
+      config: config,
+      change_behavior: change_behavior
+    } do
+      # Set small values for quick testing.
+      config = Keyword.merge(config, batch_size: 2, timeout: 100, throttle_wait: 300)
+      {:ok, pid} = start_worker(config)
+
+      change_behavior.(:throttle)
+
+      # Push a batch to trigger throttling.
+      events = [%{id: 1}, %{id: 2}]
+      Enum.each(events, &EventsWorker.push(&1, pid))
+      assert_receive {:events_sent, ^events}, 50
+
+      # Verify throttling is active.
+      state = EventsWorker.state(pid)
+      assert state.throttling == true
+
+      start_time = System.monotonic_time(:millisecond)
+
+      # Wait less than throttle_wait to ensure no flush occurs.
+      refute_receive {:events_sent, _}, 250
+
+      # Switch backend to success so the retry flush can go through.
+      change_behavior.(:ok)
+
+      # The retry flush should occur after throttle_wait (300ms).
+      assert_receive {:events_sent, ^events}, 150
+
+      elapsed = System.monotonic_time(:millisecond) - start_time
+      assert elapsed >= 300
+
+      GenServer.stop(pid)
+    end
+  end
+
+  describe "termination" do
+    test "sends remaining events on termination", %{config: config} do
+      {:ok, pid} = start_worker(config)
+      events = [%{id: 1}, %{id: 2}]
+      Enum.each(events, &EventsWorker.push(&1, pid))
+      assert_receive {:events_sent, ^events}, 500
+
+      GenServer.stop(pid)
+    end
+  end
+end
diff --git a/test/honeybadger_test.exs b/test/honeybadger_test.exs
index ddc5d68..3173440 100644
--- a/test/honeybadger_test.exs
+++ b/test/honeybadger_test.exs
@@ -3,9 +3,28 @@ defmodule HoneybadgerTest do
 
   doctest Honeybadger
 
+  defmodule MockEventsWorker do
+    use GenServer
+
+    def start_link(_), do: GenServer.start_link(__MODULE__, [], name: __MODULE__)
+    def push(data), do: GenServer.cast(__MODULE__, {:push, data})
+
+    @impl true
+    def init(_), do: {:ok, []}
+
+    @impl true
+    def handle_cast({:push, data}, state) do
+      pid = Application.get_env(:honeybadger, :mock_worker_test_pid)
+      send(pid, {:worker_push, data})
+      {:noreply, state}
+    end
+  end
+
   setup do
     {:ok, _} = Honeybadger.API.start(self())
 
+    Application.put_env(:honeybadger, :mock_worker_test_pid, self())
+
     on_exit(&Honeybadger.API.stop/0)
   end
 
@@ -405,4 +424,82 @@ defmodule HoneybadgerTest do
       Honeybadger.context(3)
     end
   end
+
+  describe "Honeybadger.event/2" do
+    setup do
+      restart_with_config(exclude_envs: [], events_batch_size: 1)
+    end
+
+    test "adds event_type to event data" do
+      Honeybadger.event("test_event", %{key: "value"})
+
+      assert_receive {:api_request, [data]}
+      assert data["event_type"] == "test_event"
+      assert data["key"] == "value"
+    end
+
+    test "works with empty event data" do
+      Honeybadger.event("test_event", %{})
+
+      assert_receive {:api_request, [data]}
+      assert data["event_type"] == "test_event"
+      ts = data["ts"]
+      assert Map.has_key?(data, "ts")
+      # Verify timestamp format matches DateTime.to_string() format
+      assert {:ok, _, _} = DateTime.from_iso8601(ts)
+    end
+  end
+
+  describe "Honeybadger.event/1" do
+    test "adds timestamp if not present" do
+      restart_with_config(
+        exclude_envs: [],
+        events_worker_enabled: false
+      )
+
+      event_data = %{event_type: "test_event", key: "value"}
+
+      Honeybadger.event(event_data)
+
+      assert_receive {:api_request, data}
+      assert data["event_type"] == "test_event"
+      assert data["key"] == "value"
+      assert Map.has_key?(data, "ts")
+    end
+
+    test "sends to Client events_worker is disabled" do
+      restart_with_config(
+        exclude_envs: [],
+        events_worker_enabled: false
+      )
+
+      event_data = %{event_type: "test_event"}
+      Honeybadger.event(event_data)
+      assert_receive {:api_request, _}
+    end
+
+    test "sends to events_worker when enabled" do
+      restart_with_config(
+        exclude_envs: [],
+        events_batch_size: 3,
+        events_worker_enabled: true
+      )
+
+      events = [
+        %{ts: "1", event_type: "test_event"},
+        %{ts: "2", event_type: "test_event"},
+        %{ts: "3", event_type: "test_event"}
+      ]
+
+      Enum.each(events, &Honeybadger.event/1)
+      assert_receive {:api_request, request_events}
+
+      stringified_events =
+        Enum.map(events, fn map ->
+          Map.new(map, fn {k, v} -> {to_string(k), v} end)
+        end)
+
+      assert request_events == stringified_events
+    end
+  end
 end
diff --git a/test/test_helper.exs b/test/test_helper.exs
index 67e2c2b..50a6cce 100644
--- a/test/test_helper.exs
+++ b/test/test_helper.exs
@@ -8,7 +8,8 @@ Application.put_all_env(
   honeybadger: [
     environment_name: :test,
     api_key: "abc123",
-    origin: "http://localhost:4444"
+    origin: "http://localhost:4444",
+    events_worker_enabled: true
   ]
 )
 
@@ -98,13 +99,18 @@ defmodule Honeybadger.API do
 
   def call(%Conn{method: "POST"} = conn, test) do
     {:ok, body, conn} = read_body(conn)
-
-    send(test, {:api_request, Jason.decode!(body)})
-
+    content_type = List.first(get_req_header(conn, "content-type")) || "application/json"
+    send(test, {:api_request, decode_payload(body, content_type)})
     send_resp(conn, 200, "{}")
   end
 
-  def call(conn, _test) do
-    send_resp(conn, 404, "Not Found")
+  def call(conn, _test), do: send_resp(conn, 404, "Not Found")
+
+  defp decode_payload(body, "application/x-ndjson") do
+    body
+    |> String.split("\n", trim: true)
+    |> Enum.map(&Jason.decode!/1)
   end
+
+  defp decode_payload(body, _), do: Jason.decode!(body)
 end