Drives reactive self-healing: periodically (and on demand) it re-replicates under-replicated sealed segments to the currently live brokers, closing the loop broker dies → membership marks it gone → its segments are healed.

It is decoupled from where membership and metadata actually live, via injected seams, so it can be tested in-process and wired to the real Malachi.Cluster.MembershipServer and control plane later without change:

• :live_brokers - (-> [broker]), the currently alive broker set (e.g. from Malachi.Cluster.MembershipServer.alive_members/1);
• :metadata_source - (-> Malachi.Metadata.t()), the current metadata;
• :apply_command - (Malachi.Metadata.command() -> any), applies a :set_segment_replicas command to the control plane;
• :replication_factor - the target replica count;
• :interval - the healing period in ms (default 5000). No environment variable sets it, which is why the warning about a value that cannot be a period names the option rather than a setting an operator could look for;
• :leader? - (-> boolean()), whether this node should heal this pass (default always). Only the cluster's membership leader heals, so N nodes do not redo the same work (1C); a non-leader still ticks but skips the pass. heal_now/1 is a manual trigger and always runs;
• :heal_opts - forwarded to Malachi.Cluster.SelfHealing.heal_sealed/4 (e.g. :batch_size).

Each pass reconciles against the live set: it runs Malachi.Cluster.SelfHealing.heal_sealed/4 (re-replicating under-replicated sealed segments, backfilling via Malachi.Cluster.Catchup) and Malachi.Cluster.Failover.plan/5 (sealing active segments whose primary died, that have a copy that failed in storage, or whose live replicas fall below the acknowledgement quorum, so writing rolls to a fresh segment), and hands the resulting commands to the control plane, seals first, stopping at the first one the broker does not answer. A refusal does not stop the rest: the first refusal of a pass is logged unless the last recorded skip already has its kind, and the commands after it still go out. A later refused pass logs again only when the kind of skip differs from the last one recorded, and a pass with neither a refusal nor an unanswered call resets that record. The commands after an unanswered one wait for the next pass, which plans them again, except a failover seal whose reason is gone by then (see apply_commands/2's comment, and #269). Membership is a view, so before fencing for failover the pass asks, read-only, each broker the view calls gone that a candidate needs, and leaves alone a segment that is no candidate once those that answered are counted: a suspicion is not a failure. The exception is a segment an earlier pass already fenced a follower of for a seal that did not land, which is kept so the seal is retried. heal_now/1 runs one pass synchronously and returns the combined result, for tests and manual triggers; its applied is what the pass planned, as in Malachi.Cluster.SelfHealing, not what reached the control plane.

Failover needs to know what each surviving replica holds, which no pure function can answer, so this pass does the probing, in two steps whose order carries the safety. Failover.candidates/3 names the segments; each live replica is MEASURED (:probe), which leaves it writable; and only once those answers reach the seal quorum (Failover.seal_quorum/1) is each answering replica FENCED (:fence), which is what makes its answer final. The fence answers go to Failover.plan/5, which applies the same rule again to them, so a fence that fails on enough replicas still declines rather than sealing on too few.

Fencing before knowing whether enough answered would close replicas of a segment the pass then declines to seal, and nothing unseals a store: see Malachi.Cluster.Failover's moduledoc. A replica that does not answer in time simply does not count, which is what leaves a segment below the seal quorum unsealed and its range blocked; that case is logged every pass, because a blocked range that says nothing is the failure mode worth avoiding.

• :probe - ((replica, segment_id, base_offset) -> {end_offset, byte_size} | :error), how a replica is MEASURED (default Malachi.Cluster.ReplicationServer.durable_stats/4 with a short timeout, so an unreachable replica cannot stall the pass);

• :fence - the same shape, how a replica is CLOSED once the seal quorum has answered (default Malachi.Cluster.ReplicationServer.seal/4). Separate from :probe so a test can watch a pass measure without fencing, which is the property that must hold below the seal quorum;
• :seal_state - ((replica, [{segment_id, base_offset}]) -> %{segment_id => {end_offset, byte_size}}), which of a replica's segments are ALREADY fenced (default Malachi.Cluster.ReplicationServer.fenced_segments/3, answering %{} on any error). See the orphaned-fence pass below; the failover pass also asks it, batched per follower, about the few candidates a suspected broker's answer would otherwise drop;

• :failed_state - ((replica, [segment_id]) -> MapSet.t(segment_id) | :unknown), which of a replica's segments have a copy that FAILED there in storage (default Malachi.Cluster.ReplicationServer.failed_segments/3, answering :unknown on any error, which leaves those segments out of the integrity pass for that tick; while a live replica stays silent, its segments wait there, which is a deliberate trade for never probing a copy that may be latched. The membership pass is not held back by this, but it never backfills from a replica whose latch was not read: a segment with no other source waits for a later tick). An active one becomes a failover candidate (see "A copy that failed" in Malachi.Cluster.Failover); a sealed one is a lost replica Malachi.Cluster.SelfHealing replaces on another broker;

• :discard_copy - ((replica, segment_id) -> any), how a failed copy that was replaced is removed from the broker it failed on (default Malachi.Cluster.ReplicationServer.delete/2, which also clears the latch there). Called only once the control plane shows the copy gone from the replica set: a copy deleted while still listed would answer a read with nothing rather than an error, and a read that finds nothing takes the range as drained;
• :probe_timeout - ms for the probing defaults :probe, :fence, :seal_state, :failed_state and :settled_confirmed (default 1000).

The orphaned-fence pass

Alongside failover, each pass reconciles the state where a segment's store is FENCED while the control plane still calls it active, which stops its range from taking any write and which nothing else converges (Malachi.Cluster.OrphanedFence documents how it arises and why it is terminal). OrphanedFence.candidates/2 names the segments, :seal_state asks each primary which of them are fenced, and OrphanedFence.plan/3 turns the answers into seal commands applied like every other.

Three seams rather than two, because this one differs from :probe on both axes that matter. It is asked about EVERY active segment on every pass rather than a handful of failover candidates, so it is batched per primary and its default answers from a marker check rather than opening and flushing each log. And like :probe it must never fence: this pass visits the whole workload, so a probe that fenced here would close a replica of every range it asked about, and nothing unseals a store.

The settling pass

The other half of the same split, on SEALED segments: the control plane records a length and a copy on disk does not agree with it (Malachi.Cluster.SealedOverrun documents how a copy comes to hold more than the seal says, and why the recorded length is the one that wins). The copies to settle come from the integrity probe Malachi.Cluster.SelfHealing already runs over sealed segments whose replicas are all live, not under-replicated, and whose latch state was readable this tick, as its :unsettled list, so this pass asks the probe for the bytes and, for each copy above the recorded size, whether its marker is on disk: SealedOverrun.plan/4 turns that list into seal_at calls, starting just past the copy the last pass reached, the :settle_copy seam makes them, and the :confirm_settled seam then records, for each copy that holds exactly its seal's records without trimming this pass, the marker that :settled_confirmed reads on the next pass. A copy trimmed this pass is confirmed on a later pass only if its bytes are still above the recorded size.

• :settle_copy - ((replica, segment_id, base_offset, end_offset) -> {:ok, dropped, byte_size, held_end} | :unreachable | :error), how one copy is trimmed to the recorded length and fenced there, or fenced at its shorter length when it holds fewer records (default Malachi.Cluster.ReplicationServer.seal_at/5). :unreachable stops the pass, so the copies after it wait for the next tick. Unlike :fence it takes the length, because the whole point is that the number is the control plane's and not the replica's;
• :settle_timeout - ms for that call (default 5000, not :probe_timeout). It bounds the trim and fence, which open the log, check the records kept and fsync. A probe's one second would time out on a large segment and have the pass retry work it had already done;

• :confirm_settled - ((replica, segment_id) -> :ok | {:error, reason}), writes the marker that records a copy as settled (default Malachi.Cluster.ReplicationServer.confirm_settled/3). {:error, :unreachable} ends the confirmations of the pass, which resume on the next tick;

• :settled_confirmed - ((replica, segment_id) -> boolean), whether that marker is on disk (default Malachi.Cluster.ReplicationServer.settled_confirmed?/3, with :probe_timeout);
• :confirm_timeout - ms for the marker write and directory sync of :confirm_settled (default 2000). The confirmation runs inside the tick, and each one is bounded by this timeout. The pass's total is not: a call already started finishes past its budget, as :settle_budget describes. A call that times out ends the pass's confirmations, and the copy is confirmed on a later pass;
• :settle_budget - ms a settle pass may spend starting its trims (default 2000), and ms its confirmations may spend starting theirs, on a clock of their own. A call already started finishes, so each phase can run one call past its budget; what it does not reach waits for the next tick;
• :settle_batch_size - copies attempted per pass (default 64), taken in turn: each pass starts just past the last copy the previous one attempted. The FIRST pass after this ships may find many sealed copies over their recorded size at once, and an unbounded pass would put that whole scan and fsync load on the disk the produce path is using. Separate from :heal_opts' :batch_size, which bounds how much a backfill copies, because the two answer different questions and tying them would make one of them unexplainable.

Fencing here is safe, unlike fencing while probing an active segment: the control plane has already sealed the segment, so the pass has no decision left to decline and there is no copy it can close that it might have wanted open. SealedOverrun's moduledoc carries the full argument.