@spec active_range_ids(t(), Malachi.Metadata.topic_name()) :: [
  Malachi.Metadata.range_id()
]

The ids of a topic's active ranges (those that currently tile the keyspace).

@spec adopt_offsets(t(), Malachi.Metadata.range_id(), non_neg_integer()) :: t()

Adopts the primary-assigned end offset of a dispatched batch into the local bookkeeping. The range's primary serializes appends and assigns the REAL offsets (the NorthGuard invariant), so when several broker frontends produce to the same range their interleaving makes a frontend's precomputed offsets diverge from what the primary assigned; the frontend then adopts the primary's truth instead of failing. The counter only moves forward (max), so this frontend's own in-flight batches keep their reservations; a later collision just adopts again.

@spec apply_heal(t(), [Malachi.Metadata.command()]) :: t()

Applies control-plane :set_segment_replicas commands (from Malachi.Cluster.SelfHealing healing sealed segments, or Malachi.Cluster.Failover promoting an active segment's primary). Each command updates the metadata; when it targets a range's active segment, the broker's active-segment cache is updated too, so the next produce routes to the new replica set/primary.

@spec commit_offset(
  t(),
  Malachi.Metadata.group(),
  Malachi.Metadata.topic_name(),
  Malachi.Metadata.offsets()
) :: {t(), term()}

Durably records a consumer group's committed position (offsets, per range) for topic, through the control plane (Raft-backed when configured). Returns {broker, reply}.

@spec committed_offsets(t(), Malachi.Metadata.group(), Malachi.Metadata.topic_name()) ::
  Malachi.Metadata.offsets()

A consumer group's committed offsets for topic (empty if it never committed).

@spec create_topic(t(), Malachi.Metadata.topic_name(), pos_integer()) :: {t(), term()}

Creates a topic (and its root range) in the control plane. Returns the updated broker and {:ok, root_range_id} or a Metadata error.

@spec delete_segment(t(), Malachi.Metadata.segment_id()) :: {t(), term()}

Removes a sealed segment_id from the control plane (retention), through the control plane (Raft-backed when configured). Returns {broker, reply} (:ok, or a Metadata error such as :segment_active/:no_such_segment).

@spec domain_violations(t()) :: %{required(String.t()) => non_neg_integer()}

Per-topic count of segments whose replica set spans fewer than min_domains distinct spread_by domains: the failure-domain diversity violations (Malachi.Cluster.Placement.domain_violations/4), keyed by topic. Empty when spread_by or min_domains is unset (nothing to check). This surfaces the HA degradation a :soft policy allows (under-diversified placements are kept, not rejected), for metrics/alerting.

@spec domain_violations(t(), Malachi.Metadata.t()) :: %{
  required(String.t()) => non_neg_integer()
}

Like domain_violations/1 but over an already-computed metadata view (avoids a second merge).

@spec merge_ranges(t(), Malachi.Metadata.range_id(), Malachi.Metadata.range_id()) ::
  {t(), term()}

Merges two buddy ranges: the control plane seals both and creates a child; both parents' active segments are sealed. Returns {broker, {:ok, child_id}} or a Metadata error.

@spec metadata(t()) :: Malachi.Metadata.t()

The current metadata as one flat view: the union of the sharded vnodes (see Malachi.Cluster.DSRSM.merged_metadata/1), for whole-cluster consumers like retention and healing.

@spec open(keyword()) :: {:ok, t()}

Opens an empty broker.

Options

• :brokers - the broker set segment replicas are placed on (default [node()]). Must be non-empty.
• :replication_factor - replicas per segment, clamped to the broker count (default 1).
• :segment_max_bytes - the active segment seals once it reaches this many encoded bytes (default 64 MiB).
• :command_fun - how metadata mutations are routed and applied (default &DSRSM.command/3, an in-memory single vnode); pass a Raft-backed function to make the metadata authoritative.
• :dsrsm - the initial sharded metadata view (default a single-vnode DSRSM.single/1), e.g. seeded from a replicated cluster.

@spec produce(
  t(),
  Malachi.Metadata.topic_name(),
  [Malachi.Log.Record.t()],
  replicate_fun()
) ::
  {t(), {:ok, placements()} | {:error, term()}}

Routes each record to the active range that owns its key and replicates the batch to that range's active segment via replicate_fun. Returns {broker, {:ok, placements}}, or {broker, {:error, reason}} (:no_such_topic; {:unroutable, key}; or a replicate_fun error). On a replicate_fun failure the returned broker reflects the groups that already committed (their data is durable); the failing group is not committed.

@spec produce_plan(t(), Malachi.Metadata.topic_name(), [Malachi.Log.Record.t()]) ::
  {t(), {:ok, placements(), [dispatch()]} | {:error, term()}}

Like produce/4, but PLANS the replication instead of executing it: routes each record to its range, opens segments as needed, commits the offsets optimistically, and returns the replication dispatches for the caller to execute (typically asynchronously, so a broker frontend never blocks its loop on replication). Returns {broker, {:ok, placements, dispatches}} or {broker, {:error, reason}}; on error NOTHING was dispatched, so the original broker is returned untouched (stronger atomicity than the executing variant, which may have committed earlier groups).

Committing before durability is what makes the frontend non-blocking, and it is safe because clients never see offsets (positions travel in opaque cursors): a dispatch that later fails burns its offsets, the client gets the error and retries, and the local counters stay in lockstep with the primary's log, which appended the batch even when its quorum did not close.

@spec put_cache(t(), Malachi.Cluster.DSRSM.t(), [Malachi.Cluster.DSRSM.vnode_id()]) ::
  t()

Replaces the broker's local metadata cache, e.g. re-seeded from the authoritative ra clusters by a periodic refresh, which fills in vnodes not yet ready at boot and picks up writes made through other nodes. The ra log is the source of truth, so a refresh only ever moves the cache forward.

unreachable names the vnodes the refresh could not read; those keep whatever view this broker already held, because the refresh represents them with an empty Metadata and installing it would delete live topics from the cache. See Malachi.Cluster.DSRSM.retain_vnodes/3.

@spec read(
  t(),
  Malachi.Metadata.range_id(),
  non_neg_integer(),
  pos_integer(),
  read_fun()
) ::
  {:ok, [Malachi.Log.Record.t()]} | :eof | {:error, term()}

Reads up to max_records records from range_id starting at offset, from the owning segment's primary via read_fun. Returns :eof past the range's end (or if nothing was produced to it).

@spec read_consume(
  t(),
  Malachi.Metadata.range_id(),
  consume_cursor(),
  pos_integer(),
  read_fun()
) ::
  {:ok, [Malachi.Log.Record.t()], consume_cursor()} | {:error, term()}

Reads up to max_records of range_id's cross-epoch stream for live consumption: first the records its sealed ancestors hold for this range's keyspace slice (oldest first, in happens-before order), then the range's own records, and it tails the active range. Unlike stream_history/5, the self source never terminates: when the range is caught up it returns an empty page whose cursor stays on the self source, so records produced later are delivered on a later call. This is what lets a consumer drain a range's full history across splits/merges (the pre-split records live in the now-sealed parent's segments) without ever seeing partition/offset.

Returns {:ok, records, next_cursor} (call again with next_cursor), or {:error, :no_such_range} if the range is unknown.

@spec read_history(t(), Malachi.Metadata.range_id(), read_fun()) ::
  {:ok, [Malachi.Log.Record.t()]} | {:error, term()}

Convenience that pages stream_history/5 to the end and returns every record as one ordered list. Loads the whole history into memory, for bounded/administrative use.

@spec seed_range_state(
  t(),
  Malachi.Metadata.range_id(),
  non_neg_integer(),
  non_neg_integer()
) :: t()

Seeds a range's recovered bookkeeping after a restart: the next offset to hand out and the floor for the segment sequence counter. The in-memory offsets/segment_seq maps start empty on boot, and without seeding every read of pre-restart data would clamp to :eof at offset 0 even though the records are durable on disk and the metadata survived (the failure the chaos harness caught). Both merges are monotone (max), so re-seeding never rewinds live state.

@spec set_broker_attributes(t(), %{required(Malachi.Metadata.broker()) => map()}) ::
  t()

Replaces the per-broker attributes used for rack/DC-aware placement (refreshed from membership). New segments spread over spread_by using these; a broker absent here has no attributes.

@spec set_brokers(t(), [Malachi.Metadata.broker()]) :: t()

Replaces the broker set new segments are placed on (e.g. refreshed from live membership), so freshly opened segments land on currently-alive brokers. Must be non-empty.

@spec split_range(t(), Malachi.Metadata.range_id()) :: {t(), term()}

Splits a range: the control plane seals the parent and creates two children; the parent's active segment is sealed. Returns {broker, {:ok, left_id, right_id}} or a Metadata error.

@spec stream_history(
  t(),
  Malachi.Metadata.range_id(),
  history_cursor(),
  pos_integer(),
  read_fun()
) ::
  {:ok, [Malachi.Log.Record.t()], history_cursor()} | {:error, term()}

Streams one bounded page of a range's cross-epoch history: records its sealed ancestors hold for this range's keyspace slice (oldest first, in happens-before order), then the range's own records. The lineage comes from the control plane (Metadata parents); ancestor records are filtered to the range's slice via Keyspace, and segments are read through read_fun.

Returns {:ok, records, next_cursor}; call again with next_cursor until it is :done. A page may be empty while next_cursor is not :done. {:error, :no_such_range} if the range is unknown.