Malachi.
View Source
Pure retention policy for sealed segments: the decision layer, no storage or timers.
Given the control-plane Malachi.Metadata, the current time, and a policy, it returns the sealed
segment ids to expire. Two independent rules, unioned:
- age. A sealed segment older than
:max_age_ms(by itssealed_at) expires; - size. Per range, if the sealed segments total more than
:max_bytes, the oldest expire until the range is back under the limit.
Size is scoped per range (the natural unit; a range's segments have contiguous offsets), age
is per segment. The active segment is never returned: it is still being written. A nil
bound disables that rule. Malachi.Cluster.RetentionCoordinator executes the returned ids.
The bound applied to each range is its topic's policy retention merged over the global policy
(the policy overrides only the keys it sets), or the global policy passed in when the topic points at
no policy, so retention is per-topic (C3c-2b). The topic's metadata says which policy NAME it points
at (Malachi.Metadata.topic_policy_name/2); the definitions come from the cluster's policy store and
are passed in already resolved, which keeps this module pure. A name the caller could not resolve
expires nothing rather than falling back to the global limits: see expired/5.
Summary
Types
A topic's effective retention, as effective/4 reports it.
Where an effective bound came from.
A retention policy. A nil (or absent) bound disables that rule.
What an expire answered, as a closed set so it can label a metric: the control plane's own replies to a
delete, and :other for anything else (a Raft timeout, an unexpected term).
How a topic's policy name resolved: no name, a defined name, or a name nothing defines.
Functions
Whether a control-plane answer authorizes deleting the segment's stored bytes on its replicas.
A topic's effective retention: each bound's value and where it came from, and how the topic's policy
NAME resolved against policies.
The sealed segment ids to expire at now_ms (epoch ms). global_policy is the fallback; each
range uses its topic's policy retention merged over it (see the module doc).
The reply_label of an expire's answer. :ok deleted the segment; :no_such_segment means it was
already gone (a retried sweep), which is neither an expiry nor a failure; :migrating (the topic is
moving between vnodes) and :segment_active (the segment is still the write head) are refusals.
The topics whose bound policy name policies does not resolve, sorted.
Types
@type effective() :: %{ resolution: resolution(), retention: %{required(atom()) => {non_neg_integer() | nil, origin()}} }
A topic's effective retention, as effective/4 reports it.
@type origin() :: :policy | :global | :unresolved_backstop
Where an effective bound came from.
@type policy() :: %{ optional(:max_age_ms) => non_neg_integer() | nil, optional(:max_bytes) => non_neg_integer() | nil, optional(:segment_max_age_ms) => pos_integer() | nil }
A retention policy. A nil (or absent) bound disables that rule.
@type reply_label() :: :ok | :no_such_segment | :migrating | :segment_active | :other
What an expire answered, as a closed set so it can label a metric: the control plane's own replies to a
delete, and :other for anything else (a Raft timeout, an unexpected term).
@type resolution() :: :none | :resolved | :unresolved
How a topic's policy name resolved: no name, a defined name, or a name nothing defines.
Functions
Whether a control-plane answer authorizes deleting the segment's stored bytes on its replicas.
Only two do, and both mean the same thing: the control plane no longer lists the segment, so no read
can be served from it and nothing will ever repair it. :ok is this sweep having removed it;
:no_such_segment is an earlier sweep having removed it and died before deleting the files, which is
the only retry the files ever get, since a segment gone from the control plane never comes back from
expired/3.
Every other answer keeps the files. A refusal (:migrating, :segment_active) leaves the segment
listed with its replica set intact, and deleting the bytes under it would leave every replica short
of its recorded byte_size: Malachi.Cluster.SelfHealing then finds no intact copy to repair from,
and a consumer positioned in that segment reads :eof until some later sweep finally gets :ok.
:other covers a Raft timeout, where the command may still have committed and may not: skipping the
delete is the recoverable side of that ambiguity, because the next sweep gets :no_such_segment and
deletes then, while deleting now cannot be undone.
@spec effective( Malachi.Metadata.policy_name() | nil, %{required(Malachi.Metadata.policy_name()) => Malachi.Cluster.Policy.t()}, policy(), non_neg_integer() | nil ) :: effective()
A topic's effective retention: each bound's value and where it came from, and how the topic's policy
NAME resolved against policies.
The one place that decides what a topic's retention is. The sweep applies it (expired/5) and the
admin read-back reports it (Malachi.Policies.topic_policy/1), so what an operator is told is what the
data sees. Three cases, and the third is the one worth spelling out.
A topic that points at no policy (name is nil) uses the global limits, as it always has. A topic
whose policy resolves takes the bounds that policy sets, nil and 0 included, and inherits the
others from the global limits.
A topic that points at a NAME policies does not hold is neither. Falling back to the global limits
there is what the administrator did not ask for: the name exists because they wanted something other
than the default, and the most common reason a more permissive policy is asked for is to keep data
longer. Expiring under the global bound would then delete exactly the data the policy existed to hold.
So nothing expires, unless the operator has set a backstop bound for this case
(:retention_unresolved_policy_max_age_ms), which is the only bound anyone has stated for it.
Examples
iex> Malachi.Cluster.Retention.effective(nil, %{}, %{max_age_ms: 1_000}, nil)
%{
resolution: :none,
retention: %{max_age_ms: {1_000, :global}, max_bytes: {nil, :global}, segment_max_age_ms: {nil, :global}}
}
iex> policies = %{"keep" => %{retention: %{max_bytes: 0}}}
iex> Malachi.Cluster.Retention.effective("keep", policies, %{max_age_ms: 1_000}, nil)
%{
resolution: :resolved,
retention: %{max_age_ms: {1_000, :global}, max_bytes: {0, :policy}, segment_max_age_ms: {nil, :global}}
}@spec expired( Malachi.Metadata.t(), non_neg_integer(), policy(), %{required(Malachi.Metadata.policy_name()) => Malachi.Cluster.Policy.t()}, non_neg_integer() | nil ) :: [Malachi.Metadata.segment_id()]
The sealed segment ids to expire at now_ms (epoch ms). global_policy is the fallback; each
range uses its topic's policy retention merged over it (see the module doc).
policies maps a policy NAME to its definition. The metadata says which name a topic points at, and
the definitions are an administrative object of the cluster
(Malachi.Cluster.PolicyStore.fetch_all/0), so the caller resolves them once per sweep and hands
them in:
this stays pure, and a sweep costs one read of the policy store rather than one per range.
@spec reply_label(term()) :: reply_label()
The reply_label of an expire's answer. :ok deleted the segment; :no_such_segment means it was
already gone (a retried sweep), which is neither an expiry nor a failure; :migrating (the topic is
moving between vnodes) and :segment_active (the segment is still the write head) are refusals.
@spec unresolved_policies(Malachi.Metadata.t(), %{ required(Malachi.Metadata.policy_name()) => Malachi.Cluster.Policy.t() }) :: [Malachi.Metadata.topic_name()]
The topics whose bound policy name policies does not resolve, sorted.
What effective/4 refuses to expire under the global limits, named so a caller can count
it. A name that stays unresolved is a misconfiguration (a policy deleted, or a typo in the binding)
and the disk it holds is invisible otherwise.
Sorted explicitly, not by accident: metadata.topics is a map, and above 32 keys it iterates in hash
order rather than key order, so the same set of topics would come back in a different order on a
cluster large enough to matter and in a stable one on every cluster small enough to test.