Malachi.
View Source
The pure bookkeeping behind Malachi.Retention.SkipReporter: which skips were already reported, and
when each reader was last logged.
A consumer group re-reads from its committed position until it commits, so the page that stepped over
expired data is served, with the same skip, on every poll in between. Counting each of those would
turn one loss into as many as the group polled. observe/4 answers :duplicate for a skip it has
already seen, so a skip is counted once however often it is re-read.
Logging is rate limited per reader (the caller's log_key, a group on a range): the first skip of a
window is logged, the rest of that window only counted, and the next line says how many were held
back.
Both tables are bounded by max entries and forget the oldest first (insertion order), so a flood of
distinct skips costs a fixed amount of memory. What is forgotten is only this bookkeeping: a forgotten
skip read again is counted again, which errs towards an alert rather than towards silence.
Summary
Types
What to do with a skip: nothing (:duplicate), count it (:count), or count it and log it, saying
how many skips of the same reader were counted without a line since the last one ({:log, held}).
Functions
An empty ledger holding at most max entries per table, logging a reader at most once per window_ms.
Records a sighting of the skip identified by skip_key, read by the reader log_key, at now_ms (a
monotonic clock), and says what to do with it.
How many distinct skips the ledger currently remembers.
Types
@type t() :: %Malachi.Retention.SkipLedger{ logged: %{required(term()) => {integer(), non_neg_integer()}}, logged_order: :queue.queue(term()), max: pos_integer(), seen: %{required(term()) => true}, seen_order: :queue.queue(term()), window_ms: non_neg_integer() }
A ledger. Build it with new/2 and read it only through this module.
@type verdict() :: :duplicate | :count | {:log, non_neg_integer()}
What to do with a skip: nothing (:duplicate), count it (:count), or count it and log it, saying
how many skips of the same reader were counted without a line since the last one ({:log, held}).
Functions
@spec new(pos_integer(), non_neg_integer()) :: t()
An empty ledger holding at most max entries per table, logging a reader at most once per window_ms.
Records a sighting of the skip identified by skip_key, read by the reader log_key, at now_ms (a
monotonic clock), and says what to do with it.
@spec size(t()) :: non_neg_integer()
How many distinct skips the ledger currently remembers.