Periodically expires sealed segments that exceed the retention policy, using the pure Malachi.Cluster.Retention decision and executing it through injected seams, so it is testable in-process and wired to the real broker/replication later without change:

• :metadata_source - (-> Malachi.Metadata.t()), the current control-plane metadata;
• :expire_segment - (Malachi.Metadata.segment_meta() -> any), removes one expired segment from the control plane and deletes its stored data on the replicas;
• :policy - a Malachi.Cluster.Retention.policy() (:max_age_ms / :max_bytes; nil = off);
• :clock - (-> non_neg_integer()) epoch ms (default System.system_time/1);
• :interval - the sweep period in ms (default 60_000);
• :leader? - (-> boolean()), whether this node should sweep (default always). Only the cluster's membership leader sweeps, so N nodes do not redo the same work (1C); a non-leader still ticks but skips the sweep.

Each sweep asks Retention.expired/3 which sealed segments to drop, resolves each to its metadata (for its replica set), and calls expire_segment on it. run_now/1 runs one sweep synchronously, ignoring :leader? (it is a manual trigger).