Reclaims the replica directories retention could not delete, on a slow cadence and behind explicit guards.

The leak it closes

Malachi.Retention.Expirer deletes a segment from the control plane and then deletes its bytes on each replica. A replica that did not answer keeps its directory, and no later sweep can ask for it again: a segment gone from the control plane never comes back from Malachi.Cluster.Retention.expired/3. Without this worker the disk a node outage cost is never returned, whatever retention is configured to do. Other paths leak the same way, a catch-up that failed after creating the directory or a copy healing moved to another broker, and this reclaims those too, because it works from what is on disk rather than from what went wrong.

Why it runs per node, and asks the owning vnode

The data directory is a fact about the node: Malachi.Storage.Layout puts the segments of every vnode in one root, so a per-vnode worker would see its neighbours' directories as unexplained. It is not leader gated for the same reason Malachi.Cluster.Scrubber is not: only a node can read its own disk.

What it does NOT read is this broker's cached copy of the metadata. That cache is refreshed by a reconcile that can skip ticks, drop a stale result or fail, and a view that aged without anyone noticing is exactly a view missing segments whose replicas are already on this disk (#249). So before a directory can be removed, the vnode that OWNS its segment is asked, linearizably, whether it lists it (Malachi.Retention.Orphans.explain/2, Malachi.Cluster.ReplicatedDSRSM.known_segments/3). That is NorthGuard's shape: the only global state is which vnodes exist, and a vnode's leader answers for the metadata it owns (the meetup transcript, 502-508 and 609-613). The owner is the vnode every segment command is routed to (Malachi.Metadata.segment_routing_topic/1). A segment's registration commits on its owner before any replica creates its directory, so the answer has no registration lag.

The guards

Deleting data is the one thing here that cannot be undone, so the pass gives up rather than guesses:

• The owners answer, in time. The question runs in a task under one deadline (:authority_deadline_ms), and a pass whose owners did not all answer, whose topology changed while it asked, that had no ring to route by, or that ran out of time is skipped whole. A partial answer is not an answer. A directory its owner does not list is also asked of every other vnode before it can go, so while one vnode anywhere is silent, a pass with a real orphan to decide is skipped too. On a sharded control plane the topology is the one of record in the durable ring store, not this node's gossiped copy (Malachi.Application.durable_orphan_authority/1 says why).
• An undecided directory is kept. One whose segment id carries no topic has no owner to ask; one whose topic a pending split is moving may be listed by neither owner at the instant each is read; and one a vnode that does not own it lists was written there by a broker routing by a stale ring. All are reported as held, without a sighting, never removed on that answer, and named in the log when that set changes.
• Age, repeated sightings and a cap per pass, which Malachi.Retention.Orphans owns and explains. Only the candidates old enough to matter are asked about, oldest first and at most :max_tracked of them, which bounds this node's work and how many owners one pass asks (each owner still answers with its whole segment map). The overflow waits for a later pass, and the pass says so in the log.
• Removal through the replication server. A directory that looks orphaned may still have an open log here, so Malachi.Cluster.ReplicationServer.delete_directory/3 closes it first. Nothing here touches the filesystem directly.

Out of scope, stated so it is not mistaken for covered: a vnode split whose rollback failed can leave a topic on a vnode no topology lists. Nothing can ask that vnode, this sweep included, and the broker's cache never read it either.

Modes

:delete reclaims, :report does everything except the removal (for a first run on a cluster whose operator wants to see the list first), and :off does not even list. A value that cannot be a mode is refused at boot by Malachi.Config.retention_orphan_sweep/1 rather than defaulted, because the default is the mode that removes and a typo would have selected it in silence. The bounds come from the environment too, and each is checked rather than trusted (Malachi.Config.checked/4).

Options

• :authority - ([name] -> {:ok, %{known: MapSet.t(), undecided: MapSet.t()}} | {:error, term()}), which of the candidate directory names the control plane accounts for (required). Built on Malachi.Retention.Orphans.explain/2 by Malachi.Application for each kind of control plane;

• :local_ref - this node's replication server reference, or a (-> ref) resolved per pass, as in Malachi.Cluster.Scrubber (required);
• :directory - the data directory to sweep (required);
• :mode - :delete (default), :report or :off. A value outside those three falls back to :report, not to the default, since the default removes;
• :interval - ms between passes (default 300_000: a leak accumulates slowly and a pass reads the whole directory, so this is deliberately slower than the other workers);
• :min_age_ms - default 600_000;
• :sightings - default 2;
• :max_per_pass - default 50;
• :max_tracked - default 10_000;
• :authority_deadline_ms - how long one pass waits for the :authority, every read included (default 15_000). Past it, the pass is skipped as {:unreachable, :deadline};
• :clock - (-> non_neg_integer()) epoch ms, for the directory ages;
• :on_result - (result -> any) for each pass (default: logs what was removed and what failed; the directories held as undecided are logged by the sweeper itself, when that set changes).

Every pass returns its outcome, so tests assert on data rather than on log lines.