@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 active_roll(t(), Malachi.Metadata.range_id()) :: roll() | :none

The roll for one range's write head, on demand (a split or a merge retiring the range), or :none when the range has none. Does not mutate: the caller fences and records in one step, so nothing is owed if it never gets that far.

The head comes from the CONTROL PLANE, and falls back to this frontend's cache only when the metadata has nothing to say. Reading it from the cache instead was issue #41: a split is performed on whichever node the operator reached, which is usually NOT the node producing to the range, so a fence conditional on the caller having written there fences nothing on exactly the node where it matters, and reports success. What that leaves behind is not a bounded window: the split seals the RANGE and not its segments, and nothing else ever closes an active segment on a sealed range (failover only seals segments whose primary is dead, retention and healing only touch sealed ones), so every frontend still caching it keeps appending to the parent forever, and a child reads those records ahead of its own.

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

Adopts the primary-assigned end offset of a dispatched batch into the local bookkeeping, when the dispatch still describes the range's write head. 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.

Guarded on the segment id: an answer that arrives after the range's head was closed describes a segment that no longer owns the range's end, and raising the counter past the fenced edge would make the next register_segment fail with :segment_overlap and block the range.

@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 moving the replica that holds a sealed segment's end to the head of its set). 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.

@spec apply_heal_reply(t(), [Malachi.Metadata.command()]) ::
  {t(), :ok | {:error, term()}}

As apply_heal/2, and also answers what the control plane said: :ok when every command took, or the first error it gave back (a refusal, or a submit that failed). Every command is still applied, so the cache ends up exactly as apply_heal/2 leaves it. A caller that acts on the answer, such as the scrubber before it deletes a copy, needs this; the discarded reply of apply_heal/2 cannot tell it anything.

@spec awaiting_fence?(
  t(),
  Malachi.Metadata.range_id(),
  Malachi.Metadata.segment_id(),
  integer(),
  non_neg_integer()
) :: boolean()

Whether this frontend sent the fence of segment_id, the write head of range_id, less than window_ms ago and has not recorded its answer yet.

A {:sealed, _} refusal for that segment is then this frontend's own roll overtaking one of its produces: the primary applied the fence (the refusal proves it) and the answer is on its way. The caller can hold the produce until that answer opens the successor instead of failing it. The window bounds the wait for an answer that was lost.

@spec bind_topic_policy(
  t(),
  Malachi.Metadata.topic_name(),
  Malachi.Metadata.policy_name() | nil
) ::
  {t(), term()}

Binds topic to the storage policy NAME policy_name, or detaches it with nil, through the topic's own vnode ({:bind_topic_policy, topic, name}, machine version 4). Whether a definition backs the name is the caller's check (Malachi.Policies.bind/3): the vnode cannot read the policy store.

@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 drop_stale_active_segments(t()) :: t()

Drops cached active segments the control plane has already sealed, seating each range's next offset at the sealed edge. Called on every metadata refresh, right after put_cache/3.

Only apply_heal/2, on the node running the heal coordinator, used to evict such a segment, so every other frontend kept routing produces at a segment the metadata had sealed and only learned otherwise when the store refused a batch. This makes the convergence level-triggered on every node instead, bounded by the reconcile interval.

An owed roll whose segment is not cached here (an age roll the sweep asked for, request_rolls/2) is dropped too once the control plane has moved past it: the segment is sealed, or the range's head is now a LATER segment (a higher sequence). The sequence grows for as long as the range keeps a segment in the control plane: one whose segments retention has all expired lists nothing to seed a frontend's counter from, so the next head may start again at 0, and a roll owed for an older sequence is then kept until its fence answers (and the control plane says the segment is gone), a later head's sequence passes it, or the node restarts. Its fence may never be answered (a primary that died and was replaced), and since a range owes one roll at a time, a roll left behind would swallow every later roll of the range on this node. A segment the refreshed view does not hold at all, with no later head beside it, keeps its roll: that is a head this view has not caught up with (an older head still listed is just as behind), the very case the age roll exists for.

@spec due_rolls(t()) :: [roll()]

The fences this broker owes: ranges whose active segment crossed :segment_max_bytes, or that the retention sweep found older than their topic's :segment_max_age_ms (request_rolls/2).

The caller fences each one's primary and hands the answer back through record_seal/5. Until it does, the range stays writable and the segment simply overshoots its soft threshold.

@spec fences_to_send(t(), integer(), non_neg_integer()) :: {t(), [roll()]}

The owed rolls whose fence the caller should send now, marked as sent at now_ms: each roll whose fence has not been sent yet, and each whose last send is at least retry_after_ms old (a fence that went unanswered, from a primary that is slow, mute or gone). Pure; the caller sends the fences and hands each answer to record_fence/5.

The mark is what keeps a busy range from fanning out a fence per produce while one is in flight, and the window is what keeps a lost fence from leaving the roll owed forever.

A roll is never sealed without its fence. It used to be, from this frontend's counter, and a frontend that had not yet seen that seal kept appending to the segment through a primary that knew nothing of it: acknowledged records landed above the recorded edge, stored and unreachable.

@spec forget_fence(t(), roll()) :: t()

Stops awaiting the fence of roll after it FAILED. The roll stays owed (fences_to_send/3 resends it), but a produce refused meanwhile is no longer taken for this frontend's own roll overtaking it.

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

Seats this frontend behind a seal somebody else performed (a {:sealed, end_offset} refusal): drops the cached active segment for segment_id, clears any roll naming it, and moves the range's next offset to end_offset, so the retry opens or adopts the successor exactly where this segment closed instead of racing :segment_overlap.

Guarded on the id, like the seal command's own cache eviction: a refusal naming an older segment must not evict the one currently open.

@spec journal(t()) :: t()

Turns the command journal on, so take_journal/1 can report what this broker applied locally.

Only a broker that re-seeds its cache from somewhere else needs it: with in-memory metadata the local cache is the only copy and nothing ever replaces it, so journaling would only grow a list nobody reads.

@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. Returns {broker, {:ok, child_id}} or a Metadata error.

Metadata only, for the same reason as split_range/2: the caller fences and records both parents' active segments first.

@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 range_awaiting_fence(
  t(),
  Malachi.Metadata.topic_name(),
  integer(),
  non_neg_integer()
) ::
  Malachi.Metadata.range_id() | nil

A range of topic whose fence this frontend is still waiting on (see awaiting_fence?/5), or nil: for a caller that learns only that a produce to topic was refused, not which of its segments refused it. The lowest such range, so the answer is deterministic.

@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(), [Malachi.Broker.Skip.t()]}
  | {: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, skips} (call again with next_cursor), {:error, :no_such_range} if the range is unknown, or {:error, :metadata_unavailable} while the range waits for a recovery that learns its end (seed_unrecovered_range/3).

skips lists every stretch of history this page moved the cursor past because it was no longer stored (see Malachi.Broker.Skip), oldest first, and is empty on an ordinary page. The page still steps over missing data exactly as before, so a consumer is never wedged on a hole; the list is what lets the layers above say so. A skip is reported only when the cursor this call returns has moved past it: a page that stepped over a hole and then read nothing leaves the cursor where it was, and the page that finally delivers reports it.

@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 record_fence(
  t(),
  roll(),
  non_neg_integer(),
  non_neg_integer(),
  non_neg_integer()
) ::
  {t(), :ok | {:error, term()}}

Records the answer to a produce roll's fence: end_offset and byte_size are what closing the segment answered, so they become its sealed length (record_seal/5).

Applied while the control plane still calls the segment ACTIVE, and applied then even if this frontend no longer owes the roll. That second half matters: a produce this frontend sent after the fence is refused with {:sealed, end_offset}, which clears the roll (forget_sealed/4) before the fence's own answer arrives, and dropping the answer then would leave the store fenced under a segment the metadata still calls active, refusing every write to the range until a heal pass reconciled it. It is safe for the same reason it is needed: while the segment is active its range has no successor, so nothing can have been written above the end this seats the counter at.

Once the segment is sealed (a failover, or another frontend's roll, closed it first), its length is not this answer's to say, and the roll is only cleared.

A segment this frontend's view does not hold at all is the control plane's to judge, not the view's: an age roll (request_rolls/2) is fenced by the sweeping leader, whose cache may never have seen a head another frontend opened, and dropping the answer there would leave the store fenced under a segment the control plane still calls active. So the seal is sent anyway. It is safe for the reason a retry is: the control plane answers an already sealed segment with its own length (record_seal/5 converges on it), and one that is gone (expired by retention) with :no_such_segment, which only clears the roll.

@spec record_seal(
  t(),
  roll(),
  non_neg_integer(),
  non_neg_integer(),
  non_neg_integer()
) ::
  {t(), :ok | {:error, term()}}

Records a fenced seal. end_offset and byte_size are what the fence ANSWERED, so the sealed length is a consequence of closing the segment rather than a measurement racing it.

All or nothing: on :ok the metadata command landed, the cached active segment is dropped, the roll is cleared, and the range's next offset is SET to exactly end_offset; on any other reply the broker is returned untouched and the roll stays owed, so the next pass re-fences (idempotent, the same numbers) and re-issues the same command.

Set, not max. A batch the fence refused, or one that never reached the primary at all, burned reserved offsets, so the local counter can sit ABOVE the fence's end; opening the next segment there would be rejected by the tiling rule and the range would wedge. The rewind is invisible to clients, because positions travel in opaque cursors and every burned offset belonged to a produce that returned an error. It is safe precisely because the fence is a latch: after it, nothing can land in this segment.

A {:already_sealed, existing} conflict (a roll fence racing a failover seal) converges on the winner rather than wedging: this broker adopts existing as the length and reports :ok.

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

Re-applies commands to the local cache only, in order, ignoring whatever each one answers.

They are already in the Raft log, so this is not a second write: it is how a cache re-seeded from a read taken before them ends up agreeing with them again. A command the read already carries answers an error (:already_exists, :segment_exists, {:already_sealed, _}) and leaves the state alone, which is why the replies are dropped rather than surfaced.

@spec request_rolls(t(), [Malachi.Metadata.segment_meta()]) :: t()

Owes a roll for each of segments, active segments the retention sweep found older than their topic's :segment_max_age_ms (Malachi.Cluster.SegmentRoll.due/5). The caller sends the fences (fences_to_send/3), exactly as for a roll requested by size.

Built from the segment's control-plane metadata rather than from this frontend's cache, because the sweep runs on the leader and the leader is usually not where the range is produced to: a quiet topic's head is in no frontend's cache at all after a restart. That is the split path's reasoning too (active_roll/2). A segment that is no longer active, or has no replica to fence, is skipped; one whose seal this view has not seen yet is fenced anyway, which answers the end it already has and is then only cleared (record_fence/5).

The range's segment sequence is raised past the rolled segment's, as adopt_active_segment/2 does: this frontend may never have opened or adopted it, and the next produce here would otherwise try to register a successor under the very id the roll is closing.

@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.

A seeded range is recovered: it leaves the set seed_unrecovered_range/3 put it in, if it was there.

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

Seeds a range whose end offset could not be learned (its primary did not answer the recovery): the segment sequence floor, which the metadata alone decides, and an inert zero for the next offset, which the monotone merge never lets rewind live state. The range is then UNRECOVERED until a later seed_range_state/4 succeeds, and read_consume/5 refuses it with {:error, :metadata_unavailable} rather than serve a page that stops at the zero: that page would be empty and successful, and a consumer cannot tell it from having caught up.

Only a range this frontend has never learned an end for becomes unrecovered. Every reconcile recovers again, and a range already serving reads keeps its end when one of those later probes times out: the end it holds is real and only moves forward, so refusing its reads then would be an outage with nothing to protect.

Produces are not held back: the primary assigns the real offsets, and adopt_offsets/4 takes them and ends the wait, since what the primary reports is the end. The frontend adopts every answer for this, including one that matches what it planned.

@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. Returns {broker, {:ok, left_id, right_id}} or a Metadata error.

Metadata only. The caller must FENCE the parent's active segment and record its seal (through active_roll/2 and record_seal/5) BEFORE calling, because a segment's sealed length has to be what closing it answered: sealing it here, from this frontend's counter, is the guess this design removes. Fencing first is also what closes the write half of the split gap, since a node that has not seen the split can no longer get a record into the parent once its store is fenced.

@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.

@spec take_journal(t()) :: {[Malachi.Metadata.command()], t()}

The control plane commands applied locally since the last call, oldest first, and a broker whose journal is empty again. [] when journaling is off.

Taken immediately before put_cache/3 and handed to replay_journal/2 immediately after: the value a re-seed installs was read before these commands, so installing it alone would undo them.

@spec topic_policy_name(t(), Malachi.Metadata.topic_name()) ::
  {:ok, Malachi.Metadata.policy_name() | nil} | {:error, :no_such_topic}

The policy NAME topic is bound to (nil for none), or {:error, :no_such_topic}.

@spec topics_bound_to(t(), Malachi.Metadata.policy_name()) :: [
  Malachi.Metadata.topic_name()
]

The topics bound to the policy name, sorted, across every vnode this broker sees.

@spec unrecovered?(t(), Malachi.Metadata.range_id()) :: boolean()

Whether range_id is waiting for a recovery that learns its end (see seed_unrecovered_range/3).