Turns the skips a consume read reports (Malachi.Broker.Skip: data a consumer was moved past because it was no longer stored) into telemetry and a rate-limited log line, attributed to the consumer group that read them.

One runs beside each data-plane broker (name_for/1 derives its name from the broker's). The broker itself never learns what a group is on the fetch path: it hands the skips back with the page, and the consumer layer (Malachi.LogApi, or the push loop, which already serves a group) sends them here by report/4, a cast. The deduplication and the log's rate limit live in this process's Malachi.Retention.SkipLedger, off the broker's loop, which serializes every produce and consume.

Reporting is best effort by design: a cast to a reporter that is restarting is dropped, which loses a count, never a record, and never fails the read that found the skip. A message this server has no clause for is dropped the way every long-lived server here drops one, through Malachi.UnexpectedMessage: counted by server and kind, its shape logged once per process, and a call answered rather than left to time out.

Options: :name, :clock ((-> integer()) monotonic milliseconds, default System.monotonic_time/1), :max (ledger entries, default :retention_skip_ledger_max, 10_000) and :window_ms (how often one reader may be logged, default :retention_skip_log_window_ms, 10 minutes).