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() -> :ok | {:error, term()}), removes one expired segment from the control plane and deletes its stored data on the replicas, and answers what the control plane answered (labeled by Malachi.Cluster.Retention.reply_label/1);

• :policy - a Malachi.Cluster.Retention.policy() (:max_age_ms / :max_bytes; nil = off);

• :policies - (-> {:ok, %{name => Malachi.Cluster.Policy.t()}} | {:error, term()}), the cluster's policy definitions, resolved ONCE per sweep (default Malachi.Cluster.PolicyStore.fetch_all/0). A topic's own metadata says which name it points at; the definitions are an administrative object of the cluster. An error skips the sweep entirely, for the reason Malachi.Cluster.PolicyStore gives: expiring under the global limits because the policy could not be read deletes exactly what the policy existed to keep;

• :unresolved_policy_max_age_ms - the backstop for a topic pointing at a name that does not resolve (default none, meaning nothing expires for it). Off by default because a bound nobody stated is not one this code gets to invent, and the counter below is what makes the case visible;
• :roll_segments - ([Malachi.Metadata.segment_meta()] -> any()), asks for the given active segments to be sealed (default none). Each sweep hands it what Malachi.Cluster.SegmentRoll.due/5 finds: the active segments older than their topic's :segment_max_age_ms. A request, answered asynchronously by the fence path (Malachi.BrokerServer.request_rolls/2), so the sweep neither waits for it nor counts it; the seal lands before a later sweep, which is the one that can expire the segment;
• :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/5 which sealed segments to drop, resolves each to its metadata (for its replica set), and calls expire_segment on it; then it asks SegmentRoll.due/5 which active segments are old enough to roll, from the same metadata and policies, and hands them to roll_segments. run_now/1 runs one sweep synchronously, ignoring :leader? (it is a manual trigger).

Every segment it tries emits [:malachi, :retention, :expire] and every sweep that runs emits [:malachi, :retention, :sweep] (see Malachi.Telemetry), so an operator can tell whether the sweep runs, how much it deleted and whether the control plane refused any of it.