@spec domain_violations(
  Malachi.Metadata.t(),
  term(),
  %{required(Malachi.Metadata.broker()) => map()},
  pos_integer()
) :: [Malachi.Metadata.segment_id()]

The ids of segments whose replica set spans fewer than min_domains distinct values of attribute_key (per attributes): i.e. segments whose placement does not meet the failure-domain diversity target and so are not HA to that degree. Independent of liveness (it audits the placed set, not whether replicas are up); under_replicated/3 covers lost replicas. Returns a sorted list, for alerting/observability (heal/3 re-replicates for durability but stays best-effort about domains, so a cluster that lost a whole domain can converge to a diversity violation this surfaces). Empty when every segment meets the target.

@spec heal(Malachi.Metadata.t(), [Malachi.Metadata.broker()], pos_integer()) :: [
  Malachi.Metadata.command()
]

A self-healing plan: a list of {:set_segment_replicas, segment_id, replica_set} commands that restore every under-replicated segment to a full replica set, to be applied through Malachi.Metadata. Re-placing over the live broker set preserves the surviving replicas (HRW retention) and only fills the vacated slots, so one pass reaches a fully-replicated fixpoint. Returns [] when nothing needs healing.

@spec place(
  Malachi.Metadata.segment_id(),
  [Malachi.Metadata.broker()],
  pos_integer(),
  keyword()
) ::
  {:ok, [Malachi.Metadata.broker()]}
  | {:error,
     :no_brokers
     | :invalid_replication_factor
     | {:insufficient_domains, non_neg_integer(), pos_integer()}}

Chooses the replica set for segment_id from available_brokers via rendezvous hashing, returning the min(replication_factor, length(available_brokers)) highest-scoring brokers (all distinct). Duplicate brokers in the input are ignored.

Options

• :spread - {attribute_key, attributes} to spread replicas across the distinct values of an attribute (e.g. {"rack", %{broker => %{"rack" => "a"}}}): the best-ranked broker of each value is taken first, then the next of each, until replication_factor. With rf <= number of values every replica lands in a different value; otherwise it is best-effort round-robin. This is deterministic (rendezvous ranking + stable grouping) and is what makes placement rack/DC aware. Omitted → the plain top-rf ranking.
• :min_domains - the minimum number of distinct failure domains the replica set must span (the distinct :spread attribute values it covers; without :spread, distinct brokers). Best-effort spread already maximises domain coverage, but with few domains or missing attributes the set can still concentrate; :min_domains makes that a checkable guarantee. Absent → no domain requirement.
• :policy - :soft (default) keeps the current best-effort behavior; :hard rejects a placement that cannot reach :min_domains with {:error, {:insufficient_domains, covered, required}} so a caller can fail fast instead of silently placing an under-diversified (not HA) replica set.

Returns {:error, :no_brokers} if there are none, {:error, :invalid_replication_factor} if replication_factor < 1, or {:error, {:insufficient_domains, covered, required}} under :hard.

@spec place_balanced(
  [term()],
  [Malachi.Metadata.broker()],
  pos_integer(),
  pos_integer()
) :: [
  {term(), [Malachi.Metadata.broker()]}
]

Places a whole set of items across available_brokers with best-effort load balancing (A2: global balancing). Each item still ranks brokers by rendezvous hashing, but a broker at capacity is skipped in favour of the next-ranked one, so load spreads instead of piling on the highest-ranked few. The soft capacity is ceil(total_replicas / brokers) + max_skew - 1: with max_skew 1 it targets the even share (rounded up); a larger max_skew leaves slack, staying closer to plain HRW (which reshuffles fewer replicas when membership changes). Returns [{item, replica_set}] in the input order; [] brokers → each item gets [].

The cap is a preference, not a hard bound: replication factor is never sacrificed for balance, so when an item cannot otherwise reach min(rf, brokers) distinct replicas it takes the least-loaded broker even if that exceeds the cap (this can happen with rf > 1, where a per-item greedy cannot always pack perfectly). In practice HRW is uniform, so the cap holds and load lands even; the hard guarantees are that every item gets min(rf, brokers) distinct replicas and the result is deterministic (ranking, input order, and running counts are identical on every node). This is a standalone balancer: it does not combine with :spread (rack-aware placement); use one or the other for now.

@spec under_replicated(
  Malachi.Metadata.t(),
  [Malachi.Metadata.broker()],
  pos_integer()
) :: [
  Malachi.Metadata.segment_id()
]

The ids of segments that are under-replicated for the given live broker set: those with fewer live replicas than the achievable target min(replication_factor, length(available_brokers)).

A replica counts as live only if it is in available_brokers, so a segment whose broker has left is flagged. The target is clamped to what is achievable, so a cluster that is simply too small to reach replication_factor is not reported as perpetually under-replicated. Returns a sorted list (deterministic).