Malachi.
View Source
The shape every background worker in this cluster shares: a tick on a fixed period, an interval that
arrived from the environment and may be nonsense, and the rule that an unknown message is counted and
survived rather than fatal (Malachi.UnexpectedMessage).
Four servers grew that shape independently: Malachi.Cluster.Scrubber,
Malachi.Cluster.RetentionCoordinator, Malachi.Cluster.HealCoordinator and
Malachi.Cluster.AutoRebalancer. Each carried its own schedule/1, its own drop_unexpected/3 and
its own opinion on a bad interval, and the newest of them predates Malachi.UnexpectedMessage and had
no catch-alls at all. This module owns that shape once, so a fifth worker inherits the decisions
instead of re-making them.
Plain functions, not a use macro
Deliberate. A macro that injected handle_call/3 catch-alls would append them after the host's own
clauses, which is the "clauses with the same name and arity should be grouped together" warning, and
this build compiles with --warnings-as-errors. It would also hide from a reader of the host which
callbacks that host actually implements. The host writes four one-line clauses and they all delegate
here, which keeps the behaviour in one place without hiding it.
The state this module owns
new/3 builds the four keys read here, which the host merges into its own state:
:worker- the label unknown messages are reported under (seeMalachi.UnexpectedMessage);:interval- the validated tick period in ms;:unexpected_shapes- the message shapes already logged;:skipping- the cause last recorded byskip/3, as the I18n key it was logged under, ornil; a pass can record a cause without skipping its work (seeskip/3).
What a host writes
@impl true
def init(opts) do
state = Map.merge(PeriodicWorker.new(opts, :scrubber, @default_interval, :scrub_interval_ms), %{...})
PeriodicWorker.schedule(state)
{:ok, state}
end
@impl true
def handle_call(:scrub_now, _from, state), do: ...
def handle_call(message, _from, state), do: PeriodicWorker.unknown_call(state, message)
@impl true
def handle_cast(message, state), do: PeriodicWorker.unknown_cast(state, message)
@impl true
def handle_info(:tick, state), do: PeriodicWorker.tick(state, &run/1)
def handle_info(message, state), do: PeriodicWorker.unknown_info(state, message)The work runs before the next tick is scheduled, as all four did: a pass that takes longer than
the interval then paces itself instead of queueing ticks behind itself. A leader gate, where the
worker has one, stays inside the host's own function, because the synchronous trigger each of them
exposes (scrub_now/1, run_now/1, heal_now/1, reconcile_now/1) deliberately ignores that gate.
Summary
Functions
Runs fun and catches an exit into {:error, reason}.
The state keys this module owns, for the host to merge into its own.
Clears what skip/3 recorded, once a pass got through, so the next skip is logged again.
Schedules the next :tick for this worker's interval.
Records the cause message (an I18n key taking reason) as the last cause, which may or may not have skipped
the pass's work (see :skipping), and logs it when the cause changed.
Runs one pass and schedules the next tick: the whole body of a host's handle_info(:tick, state).
The reply and state for a call this worker does not implement.
The state for a cast this worker does not implement. Nothing casts to any of them, which is the point.
The state for a message this worker did not plan for.
Types
@type t() :: %{ :worker => Malachi.UnexpectedMessage.server(), :interval => pos_integer(), :unexpected_shapes => term(), :skipping => atom() | nil, optional(any()) => any() }
The keys this module owns inside a host's state, alongside whatever else that host keeps.
:unexpected_shapes is Malachi.UnexpectedMessage.seen/0, left loose here on purpose: it is an
opaque set built by that module, and naming it in a map this module hands back to a host that merges
its own keys in would make every one of those merges an opaqueness violation to dialyzer.
Functions
Runs fun and catches an exit into {:error, reason}.
What a background worker needs around every call it makes to another process it does not own. These
workers ask the broker for metadata on every tick, and that is a plain GenServer.call: a broker
that is restarting, wedged or simply slower than the timeout exits the CALLER, which takes the
worker down with whatever that pass had accumulated. The supervisor restarts it, so the damage is
one lost pass rather than a lost node, but a worker that dies on every tick reports nothing and
loses its cycle position each time, and the log fills with a crash that says nothing an operator can
act on beyond what the broker is already saying about itself.
Skipping the pass is also the safe half of the sweep rule the two of them share: a worker that cannot read the state authorizing it to act must not act. Measured on the reshard drill, where a sharded control plane comes back from a full-cluster restart healthy but unable to serve metadata (see #136): six scrubber crashes in one run, one per tick, for as long as that lasted.
@spec new(keyword(), Malachi.UnexpectedMessage.server(), pos_integer(), atom()) :: t()
The state keys this module owns, for the host to merge into its own.
worker is the Malachi.UnexpectedMessage label and default_interval the period used when opts
has no :interval or carries one that cannot be a period.
setting is the name the warning about a refused interval prints, and each host passes its own
because only the host knows it. Deriving it from worker was close enough to read and wrong to act
on: it printed scrubber_interval where the setting is scrub_interval_ms
(MALACHI_SCRUB_INTERVAL_MS), and an operator cannot find that in any configuration.
Malachi.Config.checked/4 asks for the name the operator can act on, which is this one. A worker
whose interval no environment variable reaches (the heal coordinator) names its option instead, so
the line still points at something that exists.
Clears what skip/3 recorded, once a pass got through, so the next skip is logged again.
Schedules the next :tick for this worker's interval.
Records the cause message (an I18n key taking reason) as the last cause, which may or may not have skipped
the pass's work (see :skipping), and logs it when the cause changed.
Once per cause rather than once per pass: a store or a broker that stays unreadable for a minute would
otherwise say the same thing every interval. Per cause rather than once until a pass completes: a
worker that first could not read one thing and then could not read another must say so, or its last
line names a cause that has already recovered. resume/1 clears it once a pass gets through.
Runs one pass and schedules the next tick: the whole body of a host's handle_info(:tick, state).
run takes the state and returns the state, so a host whose pass also returns a result adapts it
here rather than this module guessing which element of a tuple is the state.
The reply and state for a call this worker does not implement.
The state for a cast this worker does not implement. Nothing casts to any of them, which is the point.
The state for a message this worker did not plan for.
A worker meant to run for the lifetime of the node must not die on one: the crash costs whatever the pass had accumulated, and an unexpected message is a fact worth surfacing rather than a reason to take the process down.