chore: sync with upstream commanded/commanded

Co-authored-by: Benjamin Moss <drteeth@gmail.com>
Co-authored-by: Tyler Pachal <tylerpachalandfriends@gmail.com>
Co-authored-by: Adrien Pré <adrien.pre@gmail.com>
Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>

Author: 4 people

guides/explanations/events.md

@@ -223,6 +223,71 @@ You request the consistency guarantee, either `:strong` or `:eventual`, when dis
 
 An event handler is a `GenServer` process that subscribes to the configured event store. For each event persisted to the store the `handle/2` callback is called, passing the domain event and its metadata.
 
+## Batching
+
+Event handlers can process events in batches for improved throughput. Specify `batch_size` and implement `handle_batch/1`:
+
+```elixir
+defmodule AccountProjector do
+  use Commanded.Event.Handler,
+    application: BankingApp,
+    name: "AccountProjector",
+    batch_size: 100
+
+  def handle_batch(events) do
+    # events is a list of {event_data, metadata} tuples
+    events
+    |> Enum.map(fn {event, metadata} -> transform(event, metadata) end)
+    |> Repo.insert_all()
+
+    :ok
+  end
+end
+```
+
+All events in the batch are acknowledged when `:ok` is returned. Batching and concurrency cannot be used together.
+
+### Batch failure and idempotency
+
+If `handle_batch/1` returns `{:error, reason}` or crashes, **none** of the events in the batch are acknowledged. The EventStore will redeliver the entire batch.
+
+In case you wanted to make your handler idempotent per event, you could track the last processed `event_number` to skip already-seen events.
+
+### Batch timeout
+
+Add `batch_timeout` to flush batches on size OR time, whichever comes first:
+
+```elixir
+use Commanded.Event.Handler,
+  batch_size: 50,
+  batch_timeout: 100  # milliseconds
+```
+
+The batch flushes when either 50 events accumulate or 100ms elapse. Default is `:infinity` (no timeout).
+
+**Recommended starting values:** `batch_size: 50, batch_timeout: 100`
+
+### Monitoring with Telemetry
+
+Track `flush_reason` to tune batch parameters:
+
+```elixir
+:telemetry.attach(
+  "batch-monitor",
+  [:commanded, :event, :batch, :stop],
+  fn _event, measurements, metadata, _config ->
+    # metadata.flush_reason is :size | :timeout | :immediate
+    # metadata.event_count is the batch size
+    # measurements.duration is processing time
+
+    if metadata.flush_reason == :timeout and metadata.event_count < 10 do
+      Logger.warning("Small timeout flush - consider tuning batch_size")
+    end
+  end,
+  nil
+)
+```
+
 ## Reset an EventHandler
 
 An event handler can be reset (using a mix task), it will restart the event store subscription from the configured

lib/commanded/aggregates/aggregate_state_builder.ex

@@ -62,6 +62,16 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do
     """
   })
 
+  @moduledoc """
+  Builds an aggregate's state by loading its snapshot and/or events from the
+  event store.
+
+  ## Telemetry
+
+  #{telemetry_docs()}
+
+  """
+
   @read_event_batch_size 1_000
 
   @doc """

lib/commanded/commands/router.ex

@@ -204,6 +204,22 @@ defmodule Commanded.Commands.Router do
 
       :ok = BankApp.dispatch(command, metadata: %{"ip_address" => "127.0.0.1"})
 
+  ## Causation id
+
+  You can associate a `:causation_id` with the dispatched command, which will
+  also be persisted as the `causation_id` of every event the command produces.
+  This lets you reconstruct a chain-of-causation across multiple events
+  without persisting commands separately.
+
+      :ok = BankApp.dispatch(command, causation_id: some_event_id)
+
+  When `:causation_id` is not provided, the command's own `command_uuid` is
+  used as the resulting events' `causation_id`.
+
+  Process managers and event handlers that dispatch follow-up commands
+  already propagate the handled event's id this way, so the full causation
+  chain is observable on the events alone.
+
   """
 
   alias Commanded.Aggregates.DefaultLifespan

lib/commanded/event/handler.ex

@@ -61,7 +61,8 @@ defmodule Commanded.Event.Handler do
       handler_state: map(),
       first_event_id: binary(),
       last_event_id: binary(),
-      event_count: integer()}
+      event_count: integer(),
+      flush_reason: :size | :timeout | :immediate}
     """
   })
 
@@ -79,6 +80,7 @@ defmodule Commanded.Event.Handler do
       last_event_id: binary(),
       event_count: integer(),
       recorded_event: RecordedEvent.t() | nil,
+      flush_reason: :size | :timeout | :immediate,
       optional(:error) => any()}
     """
   })
@@ -97,6 +99,7 @@ defmodule Commanded.Event.Handler do
       first_event_id: binary(),
       last_event_id: binary(),
       event_count: integer(),
+      flush_reason: :size | :timeout | :immediate,
       kind: :throw | :error | :exit,
       reason: any(),
       optional(:stacktrace) => list()}
@@ -646,8 +649,14 @@ defmodule Commanded.Event.Handler do
         - :subscribe_to - which stream to subscribe to can be either `:all` to
           subscribe to all events or a named stream (default: `:all`).
 
-        - :batch_size - the size of batches to deliver to `handle_batch` in batched
-          mode.
+        - :batch_size - controls the EventStore subscription's in-flight buffer
+          size. When `batch_timeout` is also set, this is the maximum number of
+          events the handler buffers before flushing. Enables `handle_batch/1`.
+
+        - :batch_timeout - maximum milliseconds to wait for events to accumulate
+          in the handler buffer before flushing. Defaults to `:infinity` (no
+          buffering; events processed immediately as delivered). Requires
+          `:batch_size`.
 
       The default options supported by `GenServer.start_link/3` are supported,
       including the `:hibernate_after` option which allows the process to go
@@ -734,7 +743,8 @@ defmodule Commanded.Event.Handler do
     :subscribe_to,
     :subscription_opts,
     :state,
-    :batch_size
+    :batch_size,
+    :batch_timeout
   ]
 
   @doc false
@@ -753,6 +763,12 @@ defmodule Commanded.Event.Handler do
             "both `:concurrency` and `:batch_size` are specified, this is not yet supported. Please choose one or the other."
     end
 
+    if Keyword.has_key?(config, :batch_timeout) and not Keyword.has_key?(config, :batch_size) do
+      raise ArgumentError,
+            inspect(module) <>
+              " :batch_timeout requires :batch_size. Remove the timeout or configure batching."
+    end
+
     {application, config} = Keyword.pop(config, :application)
 
     unless application do
@@ -766,20 +782,37 @@ defmodule Commanded.Event.Handler do
     end
 
     {batch_size, config} = Keyword.pop(config, :batch_size)
+    {batch_timeout, config} = Keyword.pop(config, :batch_timeout, :infinity)
+
+    unless is_nil(batch_size) or (is_integer(batch_size) and batch_size > 0) do
+      raise ArgumentError,
+            inspect(module) <>
+              " :batch_size must be nil or positive integer, got: " <> inspect(batch_size)
+    end
+
+    unless batch_timeout == :infinity or (is_integer(batch_timeout) and batch_timeout > 0) do
+      raise ArgumentError,
+            inspect(module) <>
+              " :batch_timeout must be :infinity or positive integer, got: " <>
+              inspect(batch_timeout)
+    end
 
     config =
       case batch_size do
         nil ->
-          # Delegate to `handle_event/2` when `batch_size` is not specified
-          Keyword.put(config, :handler_callback, :event)
+          config
+          |> Keyword.put(:handler_callback, :event)
+          |> Keyword.put(:batch_size, nil)
+          |> Keyword.put(:batch_timeout, :infinity)
 
-        size when is_integer(size) ->
+        size when is_integer(size) and size > 0 ->
           config
           |> Keyword.update(:subscription_opts, [buffer_size: size], fn opts ->
             Keyword.put(opts, :buffer_size, size)
           end)
-          # Delegate to `handle_batch/2` when `batch_size` is specified
           |> Keyword.put(:handler_callback, :batch)
+          |> Keyword.put(:batch_size, size)
+          |> Keyword.put(:batch_timeout, batch_timeout)
       end
 
     {application, name, config}
@@ -823,7 +856,11 @@ defmodule Commanded.Event.Handler do
     :handler_state,
     :last_seen_event,
     :subscription,
-    :subscribe_timer
+    :subscribe_timer,
+    :batch_size,
+    :batch_timer_ref,
+    batch_timeout: :infinity,
+    batch_buffer: []
   ]
 
   @doc false
@@ -842,7 +879,11 @@ defmodule Commanded.Event.Handler do
       handler_callback: Keyword.fetch!(handler_opts, :handler_callback),
       handler_state: Keyword.get(handler_opts, :state),
       consistency: consistency,
-      subscription: subscription
+      subscription: subscription,
+      batch_size: Keyword.get(handler_opts, :batch_size),
+      batch_timeout: Keyword.get(handler_opts, :batch_timeout, :infinity),
+      batch_timer_ref: nil,
+      batch_buffer: []
     }
 
     with {:ok, pid} <- Registration.start_link(application, name, __MODULE__, handler, start_opts) do
@@ -940,18 +981,22 @@ defmodule Commanded.Event.Handler do
   @doc false
   @impl GenServer
   def handle_info({:events, events}, state) do
-    %Handler{handler_callback: callback} = state
-
-    processor =
-      case callback do
-        :event -> fn events, state -> Enum.reduce(events, state, &handle_event/2) end
-        :batch -> &handle_batch/2
-      end
+    %Handler{handler_callback: callback, batch_timeout: batch_timeout} = state
 
     Logger.debug(describe(state) <> " received events: #{inspect(events)}")
 
     try do
-      state = processor.(events, state)
+      state =
+        case {callback, batch_timeout} do
+          {:event, _} ->
+            Enum.reduce(events, state, &handle_event/2)
+
+          {:batch, :infinity} ->
+            handle_batch(events, state)
+
+          {:batch, _timeout} ->
+            buffer_and_maybe_flush(events, state)
+        end
 
       {:noreply, state}
     catch
@@ -961,6 +1006,24 @@ defmodule Commanded.Event.Handler do
     end
   end
 
+  @doc false
+  @impl GenServer
+  def handle_info(:flush_batch_timeout, state) do
+    %Handler{batch_buffer: buffer} = state
+
+    Logger.debug(
+      describe(state) <> " flushing batch due to timeout: #{length(buffer || [])} event(s)"
+    )
+
+    try do
+      state = flush_batch_buffer(state, :timeout)
+      {:noreply, state}
+    catch
+      {:error, reason} ->
+        {:stop, reason, state}
+    end
+  end
+
   @doc false
   @impl GenServer
   def handle_info(
@@ -1131,6 +1194,70 @@ defmodule Commanded.Event.Handler do
     end
   end
 
+  defp buffer_and_maybe_flush(events, %Handler{} = state) do
+    %Handler{batch_buffer: buffer, batch_size: batch_size} = state
+
+    new_buffer = (buffer || []) ++ events
+    state = %Handler{state | batch_buffer: new_buffer} |> maybe_start_batch_timer()
+
+    if length(new_buffer) >= batch_size do
+      state
+      |> cancel_batch_timer()
+      |> flush_batch_buffer(:size)
+    else
+      state
+    end
+  end
+
+  defp flush_batch_buffer(%Handler{batch_buffer: buffer} = state, _flush_reason)
+       when buffer in [nil, []],
+       do: state
+
+  defp flush_batch_buffer(%Handler{batch_buffer: buffer} = state, flush_reason) do
+    Logger.debug(
+      describe(state) <> " flushing batch of #{length(buffer)} event(s) due to #{flush_reason}"
+    )
+
+    state = %Handler{state | batch_buffer: [], batch_timer_ref: nil}
+    handle_batch(buffer, %{flush_reason: flush_reason}, state)
+  end
+
+  defp maybe_start_batch_timer(%Handler{batch_buffer: buffer} = state) when buffer in [nil, []],
+    do: state
+
+  defp maybe_start_batch_timer(%Handler{batch_timeout: :infinity} = state), do: state
+
+  defp maybe_start_batch_timer(
+         %Handler{batch_timer_ref: nil, batch_timeout: batch_timeout} = state
+       )
+       when is_integer(batch_timeout) do
+    timer_ref = Process.send_after(self(), :flush_batch_timeout, batch_timeout)
+    %Handler{state | batch_timer_ref: timer_ref}
+  end
+
+  defp maybe_start_batch_timer(state), do: state
+
+  defp cancel_batch_timer(%Handler{batch_timer_ref: nil} = state), do: state
+
+  defp cancel_batch_timer(%Handler{batch_timer_ref: ref} = state) do
+    case Process.cancel_timer(ref) do
+      false ->
+        drain_flush_batch_timeout_message()
+        %Handler{state | batch_timer_ref: nil}
+
+      _remaining ->
+        %Handler{state | batch_timer_ref: nil}
+    end
+  end
+
+  defp drain_flush_batch_timeout_message do
+    receive do
+      :flush_batch_timeout -> :ok
+    after
+      0 -> :ok
+    end
+  end
+
   defp handle_batch(events, context \\ %{}, handler)
 
   defp handle_batch(events, context, %Handler{last_seen_event: last_seen_event} = state)
@@ -1538,7 +1665,8 @@ defmodule Commanded.Event.Handler do
       recorded_event: nil,
       first_event_id: first_event.event_id,
       last_event_id: last_event.event_id,
-      event_count: length(recorded_events)
+      event_count: length(recorded_events),
+      flush_reason: Map.get(context, :flush_reason, :immediate)
     }
   end