Malachi.
View Source
What a long-lived server does with a message it has no clause for: it logs the message's shape, counts it, and keeps running.
The servers on the cluster path (Malachi.Cluster.ReplicationServer, Malachi.Cluster.MembershipServer,
Malachi.BrokerServer, Malachi.Cluster.Scrubber, Malachi.Cluster.RetentionCoordinator,
Malachi.Cluster.HealCoordinator and Malachi.Retention.SkipReporter) each end their
handle_cast/2, handle_info/2 and handle_call/3
with a catch-all that calls drop/4. Without it, a message shape a server does not know raises
FunctionClauseError, and during a rolling upgrade a newer node sends exactly that: the first upgraded
primary that pushed a new replication message would take down the whole data plane of every older
follower, and keep taking it down after every restart.
What each kind gets
- a cast or an info message is dropped;
- a call is answered
{:error, :unknown_call}(seeunknown_call_reply/0), so a newer caller gets an answer it can fall back on instead of waiting out its timeout. That atom is part of the cross-version contract: callers match on it, so it does not change.
The logged line carries the message's SHAPE and never its data: strings, charlists and numbers are elided, so what is left is the tags, the atoms and the nesting that identify the sender.
Every drop emits [:malachi, :process, :unexpected_message] (see Malachi.Telemetry), which the
metrics reporter folds into malachi_unexpected_messages_total{server, kind}. Only the first
occurrence of each {kind, shape} per server process is logged, because a newer primary can push
thousands of batches a second at an older follower, and a warning per message would put the logger in
synchronous mode and slow down the very process this protects. The count is never deduplicated.
What this does not do
- A drop does not make a mixed-version cluster work. An older follower that drops a new replication message sends no ack, so the primary waits for its replication timeout; an older member that drops a new SWIM ping sends no ack either, so the newer member suspects it and can declare it dead. Not sending a new shape until every member understands it is the capability gate's job, not this module's.
- Only the outer shape is covered. A message whose tag and arity a server knows, carrying an
inner term it does not (a new record struct inside
:replica_append, a new update tuple inside a SWIM:ping), still matches the old clause and can still crash inside it. A change to what a message carries therefore needs a new tag, not a new inner shape under an old one. - It hides typos. A clause added later with a mistyped pattern now drops its message instead of crashing. The counter is the signal: it must stay at zero on a cluster where every node runs the same build, and the test suite fails on any drop it did not ask for.
Summary
Types
How the message arrived.
The {kind, shape} pairs a server process has already logged (kept in its own state).
The label of a server that drops unexpected messages (anything else is counted as other).
A message's shape: {tag, arity} for a tuple led by an atom, {:tuple, arity} for any other tuple,
the atom itself for an atom, and the type otherwise. It identifies the sender without carrying data.
Functions
Counts message, which server received as kind and has no clause for, logs its shape if this
process has not logged that {kind, shape} before, and returns the updated seen set to keep in the
server's state.
The kinds exported as malachi_unexpected_messages_total series.
The server labels exported as malachi_unexpected_messages_total series.
The shape of message (see shape/0).
The reply a server gives to a call it has no clause for.
Types
@type kind() :: :cast | :info | :call
How the message arrived.
@type seen() :: MapSet.t()
The {kind, shape} pairs a server process has already logged (kept in its own state).
@type server() ::
:replication
| :membership
| :broker
| :scrubber
| :retention
| :heal
| :skip_reporter
| :rebalance
| :orphan_sweeper
| :vnode_coordinator
The label of a server that drops unexpected messages (anything else is counted as other).
@type shape() :: {atom(), non_neg_integer()} | atom()
A message's shape: {tag, arity} for a tuple led by an atom, {:tuple, arity} for any other tuple,
the atom itself for an atom, and the type otherwise. It identifies the sender without carrying data.
Functions
Counts message, which server received as kind and has no clause for, logs its shape if this
process has not logged that {kind, shape} before, and returns the updated seen set to keep in the
server's state.
@spec kinds() :: [kind()]
The kinds exported as malachi_unexpected_messages_total series.
@spec servers() :: [server()]
The server labels exported as malachi_unexpected_messages_total series.
The shape of message (see shape/0).
@spec unknown_call_reply() :: {:error, :unknown_call}
The reply a server gives to a call it has no clause for.