feat(middleware): add BaggagePropagator for W3C Baggage

lib/commanded/middleware/baggage_propagator.ex

@@ -0,0 +1,91 @@
+if Code.ensure_loaded?(:otel_propagator_text_map) do
+  defmodule Commanded.Middleware.BaggagePropagator do
+    @moduledoc """
+    A middleware for propagating W3C Baggage to Event Handlers.
+
+    This middleware captures the current baggage from the OpenTelemetry context
+    and stores it in event metadata following the
+    [W3C Baggage](https://www.w3.org/TR/baggage/) specification, allowing event
+    handlers to read user-defined name/value pairs that travel alongside trace
+    context.
+
+    Baggage and W3C Trace Context are independent specifications.
+    `Commanded.Middleware.TraceContextPropagator` propagates `traceparent` and
+    `tracestate`; this middleware propagates `baggage`. Use both together to
+    propagate the full OpenTelemetry context.
+
+    ## W3C Baggage
+
+    The middleware stores baggage using the standard W3C header name:
+
+      * `baggage` - A comma-separated list of `key=value` pairs, with optional
+        per-entry metadata (e.g. `userId=alice,serverNode=DF%2028,isProduction=false`).
+
+    ## Persistence considerations
+
+    Baggage entries are written into event metadata and therefore persisted in
+    the event store for the lifetime of the stream. Do not place PII, secrets,
+    or unbounded values into baggage when this middleware is enabled. Filter or
+    redact at the producer site (before `OpenTelemetry.Baggage.set/1`) if
+    necessary.
+
+    ## Dependencies
+
+    This middleware requires `opentelemetry_api` to be installed and the
+    `baggage` propagator to be present in the global text map propagator
+    composite (the OpenTelemetry SDK's default is `[trace_context, baggage]`).
+
+        {:opentelemetry_api, "~> 1.0"}
+
+    ## Usage
+
+    ```elixir
+    defmodule BankRouter do
+      use Commanded.Commands.Router
+
+      middleware Commanded.Middleware.TraceContextPropagator
+      middleware Commanded.Middleware.BaggagePropagator
+
+      dispatch [OpenAccount, DepositMoney],
+        to: BankAccount,
+        identity: :account_number
+    end
+    ```
+    """
+
+    @behaviour Commanded.Middleware
+
+    alias Commanded.Middleware.Pipeline
+
+    @doc """
+    Injects the W3C `baggage` header into the command pipeline metadata.
+
+    Called before command dispatch to capture the current baggage. If no
+    baggage is set on the current OpenTelemetry context, the pipeline is
+    returned unchanged.
+    """
+    @impl true
+    def before_dispatch(%Pipeline{} = pipeline) do
+      case :otel_propagator_text_map.inject([]) do
+        [] ->
+          pipeline
+
+        headers ->
+          maybe_assign(pipeline, "baggage", List.keyfind(headers, "baggage", 0))
+      end
+    end
+
+    defp maybe_assign(pipeline, _key, nil), do: pipeline
+
+    defp maybe_assign(pipeline, key, {_, value}),
+      do: Pipeline.assign_metadata(pipeline, key, value)
+
+    @doc false
+    @impl true
+    def after_dispatch(pipeline), do: pipeline
+
+    @doc false
+    @impl true
+    def after_failure(pipeline), do: pipeline
+  end
+end

lib/commanded/opentelemetry/helpers.ex

@@ -21,10 +21,12 @@ defmodule Commanded.OpenTelemetry.Helpers do
       |> :otel_propagator_text_map.extract_to(headers)
 
     span_ctx = :otel_tracer.current_span_ctx(ctx)
+    baggage = :otel_baggage.get_all(ctx)
 
-    case span_ctx do
-      :undefined -> {[], :undefined}
-      span_ctx -> {[OpenTelemetry.link(span_ctx)], ctx}
+    case {span_ctx, map_size(baggage)} do
+      {:undefined, 0} -> {[], :undefined}
+      {:undefined, _} -> {[], ctx}
+      {span_ctx, _} -> {[OpenTelemetry.link(span_ctx)], ctx}
     end
   end
 
@@ -36,6 +38,7 @@ defmodule Commanded.OpenTelemetry.Helpers do
     []
     |> maybe_add_header(metadata, "traceparent")
     |> maybe_add_header(metadata, "tracestate")
+    |> maybe_add_header(metadata, "baggage")
   end
 
   defp maybe_add_header(headers, metadata, key) do

test/middleware/baggage_propagator_test.exs

@@ -0,0 +1,107 @@
+defmodule Commanded.Middleware.BaggagePropagatorTest do
+  use ExUnit.Case, async: false
+
+  alias Commanded.Middleware.BaggagePropagator
+  alias Commanded.Middleware.Commands.Fail
+  alias Commanded.Middleware.Pipeline
+
+  setup do
+    on_exit(fn -> :otel_baggage.clear() end)
+    :otel_baggage.clear()
+    :ok
+  end
+
+  describe "before_dispatch/1" do
+    test "does not modify metadata when no baggage is set" do
+      pipeline = %Pipeline{command: %Fail{}, metadata: %{}}
+
+      result = BaggagePropagator.before_dispatch(pipeline)
+
+      assert result.metadata == %{}
+    end
+
+    test "captures baggage when present on the current context" do
+      :otel_baggage.set(%{"userId" => "alice", "tenant" => "acme"})
+
+      pipeline = %Pipeline{command: %Fail{}, metadata: %{}}
+
+      result = BaggagePropagator.before_dispatch(pipeline)
+
+      assert %{"baggage" => header} = result.metadata
+      assert header =~ "userId=alice"
+      assert header =~ "tenant=acme"
+    end
+
+    test "preserves existing metadata" do
+      :otel_baggage.set(%{"userId" => "alice"})
+
+      pipeline = %Pipeline{
+        command: %Fail{},
+        metadata: %{"correlation_id" => "abc-123"}
+      }
+
+      result = BaggagePropagator.before_dispatch(pipeline)
+
+      assert result.metadata["correlation_id"] == "abc-123"
+      assert result.metadata["baggage"] =~ "userId=alice"
+    end
+
+    test "does not modify pipeline command" do
+      :otel_baggage.set(%{"userId" => "alice"})
+
+      command = %Fail{}
+      pipeline = %Pipeline{command: command, metadata: %{}}
+
+      result = BaggagePropagator.before_dispatch(pipeline)
+
+      assert result.command == command
+    end
+
+    test "does not modify pipeline assigns" do
+      :otel_baggage.set(%{"userId" => "alice"})
+
+      pipeline = %Pipeline{
+        command: %Fail{},
+        metadata: %{},
+        assigns: %{existing: "value"}
+      }
+
+      result = BaggagePropagator.before_dispatch(pipeline)
+
+      assert result.assigns == %{existing: "value"}
+    end
+
+    test "round-trips through W3C baggage decoding" do
+      :otel_baggage.set(%{"userId" => "alice", "tenant" => "acme"})
+
+      pipeline = %Pipeline{command: %Fail{}, metadata: %{}}
+
+      result = BaggagePropagator.before_dispatch(pipeline)
+
+      ctx =
+        :otel_ctx.new()
+        |> :otel_propagator_text_map.extract_to([{"baggage", result.metadata["baggage"]}])
+
+      decoded = :otel_baggage.get_all(ctx)
+
+      assert decoded["userId"] == {"alice", []}
+      assert decoded["tenant"] == {"acme", []}
+    end
+  end
+
+  describe "after_dispatch/1" do
+    test "returns pipeline unchanged" do
+      pipeline = %Pipeline{command: %Fail{}, metadata: %{"test" => "value"}}
+
+      assert BaggagePropagator.after_dispatch(pipeline) == pipeline
+    end
+  end
+
+  describe "after_failure/1" do
+    test "returns pipeline unchanged" do
+      pipeline = %Pipeline{command: %Fail{}, metadata: %{"test" => "value"}}
+
+      assert BaggagePropagator.after_failure(pipeline) == pipeline
+    end
+  end
+end

test/opentelemetry/helpers_test.exs

@@ -64,6 +64,32 @@ defmodule Commanded.OpenTelemetry.HelpersTest do
       after_ctx = :otel_ctx.get_current()
       assert before_ctx == after_ctx
     end
+
+    test "returns ctx with baggage when only baggage is present" do
+      {links, ctx} = Helpers.extract_propagated_ctx(%{"baggage" => "userId=alice"})
+
+      assert links == []
+      refute ctx == :undefined
+      assert :otel_baggage.get_all(ctx) == %{"userId" => {"alice", []}}
+    end
+
+    test "returns ctx with both span and baggage when both are present" do
+      {_trace_id, _span_id, traceparent} = make_traceparent()
+
+      metadata = %{
+        "traceparent" => traceparent,
+        "baggage" => "userId=alice,tenant=acme"
+      }
+
+      {links, ctx} = Helpers.extract_propagated_ctx(metadata)
+
+      refute ctx == :undefined
+      assert [%{}] = links
+
+      baggage = :otel_baggage.get_all(ctx)
+      assert baggage["userId"] == {"alice", []}
+      assert baggage["tenant"] == {"acme", []}
+    end
   end
 
   describe "clear_ctx/0" do