StatifierRouter.Broadway (StatifierRouter v0.11.3)

The Broadway front: a pipeline that hands every message its producer emits to StatifierRouter.route/3.

The host starts it in its own supervision tree, with any Broadway producer it already operates; this package ships no producer and starts no process of its own.

children = [
  MyApp.Repo,
  {StatifierRouter.Broadway,
   name: MyApp.ParcelScansRouter,
   producer: {BroadwayKafka.Producer, kafka_opts},
   router: router_config,
   processors: [default: [concurrency: 8]]}
]

Options

start_link/1 takes a keyword list:

Option	Value	Default
`:name`	the pipeline's name, an atom	required
`:producer`	`{module, opts}`: any Broadway producer and its options	required
`:router`	a `%StatifierRouter.Config{}`	required
`:normalize`	a fun of one `%Broadway.Message{}` returning the event `StatifierRouter.route/3` takes	`normalize/1`
`:processors`	Broadway's `:processors` option, passed through	`[default: []]`

Any other key, or a value the table does not allow, raises ArgumentError, as Broadway.start_link/2 does for its own options. There are no batchers in this release.

Each message

handle_message/3 normalizes the message and calls StatifierRouter.route/3, which returns once every delivery's transaction has ended. {:ok, outcomes} returns the message unchanged, so Broadway acknowledges it after the transactions committed. {:error, reason} returns it failed with reason. A raise inside route/3 is not rescued here either: Broadway fails the message it raised on (ADR-0003, section 1).

Redelivery is the producer's contract

A failed message is not redelivered by this pipeline and not by Broadway, which "does not provide any sort of retries out of the box" and acknowledges a failed message as failed immediately. Whether the event comes back is the producer's contract:

a queue-style producer that leaves an unacknowledged message invisible for a timeout and then hands it over again, Amazon SQS the example Broadway itself names, redelivers it;
BroadwayKafka.Producer, the producer this module's example uses, "will always ack the messages even when they fail" and advances the group's offset past the failed one, so nothing hands it over again and reprocessing is, in its own words, a strategy the host rolls.

A host that needs a failed delivery retried therefore chooses a producer that gives it back, or arranges the replay itself. Nothing in this package retries, and nothing in it holds a failed message.

Partitioning

partition/3 is the pipeline's partition_by. It keeps every message for one address on one processor, so deliveries to one execution reach it one after another instead of queueing on its lock, each holding a pooled connection while it waits. That is an optimisation and never the guarantee: two deliveries to one execution are stepped one at a time by statifier_persistence's per-execution lock, whichever processor they came through (ADR-0003, section 5). A binding whose order is :none is not partitioned by its key (ADR-0003, section 10).

The partitioner runs in the producer's dispatcher, so what it rescues and what it lets through matter to the producer. An ArgumentError from its bindings read is rescued, whether the :bindings_resolver, the :around_delivery wrapper or the router's check of the wrapper's contract raised it: the message is partitioned by its message id, and handle_message/3 is where it fails, if it does. Anything else raised, exited or thrown there, and any raise from normalize, is not rescued and takes the producer down with the messages it holds. partition/3 says which in full.

Summary

Functions

normalize(message)

The default :normalize: the event StatifierRouter.route/3 takes, read from the message. :scope, :message_id and :source are read from the message's metadata, and the message's data is the event's data.

partition(message, router, normalize)

The partition of one message: the hash, by :erlang.phash2/1, of the address {scope, document, key} of the first binding in router's order that is enabled, is for the event's source, has order: :by_key, and whose match holds and key produces a key for the event.

start_link(opts)

Starts the pipeline, linked to the calling process. The module documentation lists the options.

Functions

normalize(message)

@spec normalize(Broadway.Message.t()) :: map()

A message that does not carry all four is not refused here: the event this function builds carries nil for what the metadata did not hold, and StatifierRouter.route/3 refuses it. A message with no :message_id in its metadata is refused with {:error, :no_message_id}, which route/3 checks before anything else; a message whose metadata holds no :scope or no :source, or whose data is not a map, is refused with {:error, {:invalid_event, event}}. Either way the message fails. A producer whose messages carry them elsewhere is paired with a :normalize of the host's own.

iex> message = %Broadway.Message{
...>   data: %{"kind" => "click", "impression_id" => "imp_7f3a"},
...>   metadata: %{scope: "7c1e", message_id: "ad_events/3/1107", source: "ad_events"},
...>   acknowledger: Broadway.NoopAcknowledger.init()
...> }
iex> StatifierRouter.Broadway.normalize(message)
%{
  scope: "7c1e",
  message_id: "ad_events/3/1107",
  source: "ad_events",
  data: %{"kind" => "click", "impression_id" => "imp_7f3a"}
}

partition(message, router, normalize)

@spec partition(
  Broadway.Message.t(),
  StatifierRouter.Config.t(),
  (Broadway.Message.t() -> map())
) ::
  non_neg_integer()

Under a :bindings_resolver, the bindings are the resolver's answer for the event's scope, asked here once per message and again by route/3 in handle_message/3 (ADR-0001, the Amendment of 2026-09-25). An answer that is not a list of bindings raises ArgumentError; here that raise is rescued and the message is partitioned by its message id, so it does not take the producer down. route/3 raises the same ArgumentError in handle_message/3, and Broadway fails the message there. A resolver that raises ArgumentError itself is treated the same way; any other raise from the resolver is not rescued here.

When the configuration gives an :around_delivery, the bindings read runs inside one call of it, here in the producer's dispatcher, handed the event's scope and the door :partition; handle_message/3's route/3 calls it again in the processor, with the door :route (ADR-0003, the Amendment of 2026-10-02). Only the read is wrapped, and no transaction is open around it, so what the wrapper sets in the process is visible to it and a transaction-local setting is not, unless the wrapper opens a transaction itself.

What is rescued here is an ArgumentError, whichever raised it: the wrapper itself, the resolver inside it, or the router's check of the wrapper's contract, which raises after the wrapper returns when it answered anything but the work's answer or called the work other than once. Each is rescued as a malformed resolver answer is: the message is partitioned by its message id, and route/3 in handle_message/3 calls the wrapper again under :route, where nothing rescues it, so a wrapper that raises or breaks its contract there too fails the message. Any other raise from the wrapper, and any exit or throw, is not rescued: it propagates out of the partitioner, in the producer's dispatcher, as a raise from normalize does.

The configuration's :wrap_target adds no call to this pipeline. The partitioner sends nothing, and a send to an execution target from a step handle_message/3 drives runs inside that message's :route call, so it is wrapped once, by that call, and never again under the door :target (ADR-0003, the Amendment of 2026-10-04).

A message no such binding addresses is partitioned by the hash of its message id. So is a message normalize builds no routable event from, one whose resolver answer StatifierRouter.route/3 would refuse, and one whose resolver answer is malformed: a partitioner answers for every message, and handle_message/3 is where such a message fails.

normalize is called here as well as in handle_message/3, and here it runs in the producer's dispatcher, where a raise takes the producer down with the messages it holds. It is not rescued: a :normalize must answer for every message, and the default one does.

start_link(opts)

@spec start_link(keyword()) :: Broadway.on_start()

Starts the pipeline, linked to the calling process. The module documentation lists the options.