Background verification of data at rest (a "scrub"), and repair of what it finds.

A sealed segment is immutable, and nothing re-reads it: Malachi.Log.recover/2 trusts sealed files by name, so a checksum is only ever confirmed when a consumer happens to read that exact frame. Bit rot there is therefore silent, and worse than an error: the read bound comes from the control plane's record count rather than from the file, so a read past the damaged frame returns no records, which the broker reads as "this source is drained". The consumer then stalls at that offset or skips the rest of the range, with nothing logged anywhere.

This worker closes that by walking the node's own sealed segments on a slow cadence, verifying every frame's checksum (Malachi.Log.verify/2), and repairing a damaged copy from an intact replica. NorthGuard's rule that you are on the clock to re-replicate a lost replica applies here too: a copy that will not decode is a lost replica, whatever the metadata says about it.

Why it runs on every node, not just the leader

Unlike the healing and retention coordinators, this one is deliberately not leader-gated: each node is the only one that can read its own disk, and corruption is a per-copy fact, not a cluster-wide one. Every node scrubs what it stores.

Repair, and its one safety rule

Damage is repaired by fetching the segment again from a replica that still has it intact, which means deleting the local copy first. The rule is that the local copy is never destroyed before a peer has confirmed it holds the whole segment, intact: swapping a partially readable copy for a shorter one, or for no copy at all, would be worse than the damage. Both halves of that sentence carry weight, and the second is the one easy to lose: a peer whose checksums all pass can still be missing whole frames at the end, and repairing from it deletes the local copy and then catches up to an offset the source cannot reach. So a peer qualifies only when its scan matches what the control plane recorded at seal time, which is the same test cross_check/4 applies to this node's own copy. Peers are verified through their own Scrubber (never the replication server), so the verification scan stays off the replication hot loop. The confirmation that precedes a repair reads the peer's replication server directly, in bounded batches under a time budget, and the peer is read through to the end before any delete. Each batch holds that server for its duration, so a confirmation competes with the peer's writes and fsyncs for up to the budget; the per-read bound keeps any one batch short.

During a rolling upgrade, a repair whose damaged copy is the head is refused while the effective machine version is below the one that introduces the compare-and-swap demote. That repair reports {:demote, ...} as unrepairable and deletes nothing. The refusal lifts when the effective version reaches that version, which needs every member on the newer code and no operator pin below it (MALACHI_RA_MACHINE_VERSION): a pin keeps the effective version down, so the refusal lasts until the pin is removed.

A copy that failed in storage

A copy this node's replication server has latched as failed (see "Storage failures" in Malachi.Cluster.ReplicationServer) is skipped, reported under :skipped, and never repaired here. The repair deletes the copy and refetches it onto the disk that just failed: the delete clears the latch, a disk still failing latches it again during the refetch, and the next pass does it all over, reporting the same damage every time. Malachi.Cluster.SelfHealing replaces such a copy on another broker instead. For the same reason a peer asked about a latched copy answers {:error, :failed_copy}, so it is never chosen as a repair source: this node would delete its own copy and then be refused by the peer.

A latch that cannot be read (the replication server is slow, restarting or gone) is UNKNOWN, never absent, because the repair is the one step here that destroys something. The pass still verifies the copies and reports their damage, but repairs nothing from a peer (:latch_unknown under :unrepairable), and a peer in that state answers {:error, :latch_unknown} and is not a source.

When the damaged copy is the segment's primary, the repair first submits a :set_segment_replicas that moves this node to the end of the replica set. Reads follow the head of that set, so they move to an intact replica immediately instead of hitting the hole (or the gap left while the copy is being refetched).

Every pass returns its outcome; logging is only the default :on_result callback, so tests assert on data rather than on log lines.

Options

• :metadata_source - (-> Malachi.Metadata.t()), the current metadata (required);
• :local_ref - this node's replication server reference, or a (-> ref) resolved per pass (required). A single-node broker owns an unnamed replication server, so its reference is a pid that a broker restart replaces: passing a function keeps the scrub from silently matching nothing after such a restart;
• :directory - the data directory whose segments are scrubbed (required);
• :apply_command - (Metadata.command() -> any), applies the demote command (required);
• :peer_scrubber - (node() -> GenServer.server()), how to reach a peer's scrubber. Defaults to this process's own registered name on that node, since every node runs the same worker under the same name;
• :peer_scrubber_for - (replica -> GenServer.server()), the scrubber of one replica. It defaults to :peer_scrubber on the replica's node, and exists so a test can tell two replicas on one node apart;
• :interval - ms between ticks (default 60s). A value that is not a positive integer is refused with a warning and the default used instead, since it arrives from the environment (Malachi.Cluster.PeriodicWorker);
• :segments_per_tick - segments verified per tick (default 1). With the default 64MB segment size a full cycle takes segments * interval / segments_per_tick, so a node holding 10k sealed segments revisits each one about weekly;
• :on_result - (result -> any) for each pass (default: log damage and repairs);
• :confirm_deadline_ms - the budget, in ms, of one source confirmation (default 10s). A confirmation that runs out of it defers the repair and resumes on a later tick;
• :call_timeout - how long the calls that decide the seal and the latch of a rebuilt copy wait for their answer, in ms (default 1s). The demotion, the metadata read and the latch lookup keep the fixed one-second bound, and the peer verification keeps its own longer timeout;
• :confirm_clock - (-> integer), the clock the confirmation budget is measured by (default: monotonic ms);
• :fail_copy - (ref, segment_id, reason -> any), the latch call of a rebuilt copy (default: ReplicationServer.fail_copy/3). It is a seam for tests, which make a latch fail on demand;
• :delete - (ref, segment_id -> any), the delete that takes a rebuilt copy away when its latch does not answer (default: ReplicationServer.delete/2). A seam for tests, as :fail_copy is;