Malachi.
View Source
The Dynamically-Sharded Replicated State Machine: a Malachi.Cluster.HashRing plus one
Malachi.Metadata state machine per vnode. Metadata commands and queries are routed by
consistent hashing to the vnode that owns them, so the cluster's metadata is sharded
across vnodes (and different topics land on different vnodes, avoiding hotspots).
Sharding granularity
A topic's metadata: the topic plus all its ranges and segments - is co-located on a single vnode, routed by topic name. Every command and query therefore carries the topic name. This keeps each operation single-vnode and keeps range ids unique within the vnode that owns them.
NorthGuard additionally shards a topic's ranges/segments by range id across vnodes (so a
single hot topic spreads out). That requires cross-vnode operations and a
globally-unique range-id scheme, so it, and vnode split (rebalancing, which migrates
metadata between vnodes): are deferred to a later increment. See docs/ARCHITECTURE.md.
Caller contract: (topic_name, range_id/segment_id) must match, and is enforced
Because the topic name is used only for routing, the target range_id/segment_id of a mutating
command must actually belong to topic_name: several co-located topics share a vnode (range ids are
only unique within one), so a command naming topic A but targeting a range of co-located topic B would
otherwise act on B's range. command/3 now rejects that pair with {:error, :range_topic_mismatch},
comparing topic_name against the topic the command's range/segment id embeds
(Malachi.Metadata.routed_to_foreign_topic?/2, over the {topic, seq} range id or the {range_id, seq}
segment id, and both ids of a merge). Only range/segment commands are checked: a command that names its
topic directly cannot be aimed at a co-located topic's range. The broker only ever derives topic_name
from the command, so this is defense in depth against a future or direct caller, not a bug the broker
can hit today.
When range-id sharding lands, range/segment operations route by range id and the range's own vnode is authoritative, so the topic param drops out for those ops and the mismatch cannot even be expressed. The structural check here stays correct in the meantime.
Like Metadata, this is a pure, deterministic value: command/3 returns
{new_dsrsm, reply}. Backing each vnode with its own ra group leaves this routing layer
unchanged; see Malachi.Cluster.ReplicatedDSRSM, which does exactly that.
Summary
Functions
The active ranges of a topic.
Adds a vnode at token, migrating any displaced topics to it: the general "grow the
ring" entry point. Delegates to split_vnode/3, so it is safe whether or not topics
already exist (migration is a no-op on an empty ring). Propagates HashRing placement
errors (:token_out_of_range, :token_taken, :already_present).
Routes a Malachi.Metadata command to the vnode owning topic_name and applies it there,
returning {new_dsrsm, reply}. reply is whatever Metadata.apply/3 returns, or
{:error, :no_vnode} if the ring is empty.
A consumer group's committed offsets for topic_name (empty if it never committed).
A range of topic_name, or nil. range_id must belong to topic_name (see caller contract).
A segment of topic_name, or nil. segment_id must belong to topic_name (see caller contract).
The topic metadata, or nil.
The union of every vnode's Metadata: a single flat view for whole-cluster consumers (retention,
healing) that iterate all segments. Topics/ranges/segments/offsets are disjoint across vnodes (each
topic lives on one vnode), so the union is unambiguous; with one vnode it is that vnode's metadata.
Builds an empty DS-RSM with no vnodes. Options are forwarded to HashRing.new/1
(e.g. :ring_bits).
All ranges of a topic.
A fresh cache with vnode_ids' entries taken from previous instead: what a reader does with the
vnodes a snapshot could not reach. Their entry in fresh is an empty placeholder, and installing
that placeholder would erase topics whose records are durable on disk, turning every read of them
into a successful empty page. Keeping the last view this reader held is stale at worst, and the
reader already tolerates staleness between refreshes; it is nil metadata that has no honest
reading. A vnode absent from previous (never yet reached) keeps the placeholder, since there is no
older view to keep.
A DS-RSM over ring whose vnodes already hold metadata_by_vnode (%{vnode_id => Metadata}):
used to seed a local read cache that mirrors an authoritative Malachi.Cluster.ReplicatedDSRSM
sharing the same ring: reads are served from this cache, writes routed back through the vnodes' ra
clusters. metadata_by_vnode must key exactly the ring's vnodes.
All segments of a range under topic_name. range_id must belong to topic_name (see caller contract).
A DS-RSM with a single vnode holding metadata: the trivial, unsharded shape. Seeds a one-vnode
control plane (the current single-cluster runtime and tests) before real multi-vnode sharding: with
one vnode every topic routes to it, so behavior matches a plain Malachi.Metadata.
Adds a new vnode at token and migrates to it every topic that now routes there:
the "dynamically sharded" part of DS-RSM (rebalancing). Adding a vnode only steals an arc
from one existing vnode, so only that vnode's affected topics move; their full metadata
(topic + ranges + segments) is relocated. Range ids stay valid because they are globally
unique ({topic, seq}); migration is likewise safe for segments only if segment ids are
globally unique (the broker-assigned contract, see Malachi.Metadata). Propagates
HashRing placement errors.
The NAME of the storage policy topic_name points at, or nil if it points at none or the topic is
unknown. The definition behind the name is a cluster object (Malachi.Cluster.PolicyStore).
The topics bound to the policy name on every vnode, sorted. Reads each vnode's topics where they are
rather than building merged_metadata/1, which copies every range and segment to answer a question
about one field of the topic records.
Routes topic_name to its vnode and updates that vnode's Metadata with update_fun, the
general single-vnode mutation combinator. update_fun receives the vnode's Metadata and returns
{new_metadata, reply} (the Malachi.Metadata.apply/3 shape); the new metadata replaces the
vnode's and reply is returned as-is. {dsrsm, {:error, :no_vnode}} if the ring is empty.
The vnode id that owns topic_name, or {:error, :empty} if the ring has no vnodes.
The ids of the vnodes.
Types
@type t() :: %Malachi.Cluster.DSRSM{ ring: Malachi.Cluster.HashRing.t(), vnodes: %{required(vnode_id()) => Malachi.Metadata.t()} }
@type vnode_id() :: Malachi.Cluster.HashRing.vnode_id()
Functions
@spec active_ranges_of_topic(t(), Malachi.Metadata.topic_name()) :: [ Malachi.Metadata.range_meta() ]
The active ranges of a topic.
@spec add_vnode(t(), vnode_id(), Malachi.Cluster.HashRing.token()) :: {:ok, t()} | {:error, atom()}
Adds a vnode at token, migrating any displaced topics to it: the general "grow the
ring" entry point. Delegates to split_vnode/3, so it is safe whether or not topics
already exist (migration is a no-op on an empty ring). Propagates HashRing placement
errors (:token_out_of_range, :token_taken, :already_present).
@spec command(t(), Malachi.Metadata.topic_name(), Malachi.Metadata.command()) :: {t(), term()}
Routes a Malachi.Metadata command to the vnode owning topic_name and applies it there,
returning {new_dsrsm, reply}. reply is whatever Metadata.apply/3 returns, or
{:error, :no_vnode} if the ring is empty.
For range/segment commands, the targeted id must belong to topic_name, see the
caller contract in the module docs (a mismatch is not rejected and may act on a
co-located topic's metadata).
@spec committed_offsets(t(), Malachi.Metadata.group(), Malachi.Metadata.topic_name()) :: Malachi.Metadata.offsets()
A consumer group's committed offsets for topic_name (empty if it never committed).
@spec get_range(t(), Malachi.Metadata.topic_name(), Malachi.Metadata.range_id()) :: Malachi.Metadata.range_meta() | nil
A range of topic_name, or nil. range_id must belong to topic_name (see caller contract).
@spec get_segment(t(), Malachi.Metadata.topic_name(), Malachi.Metadata.segment_id()) :: Malachi.Metadata.segment_meta() | nil
A segment of topic_name, or nil. segment_id must belong to topic_name (see caller contract).
@spec get_topic(t(), Malachi.Metadata.topic_name()) :: Malachi.Metadata.topic_meta() | nil
The topic metadata, or nil.
@spec merged_metadata(t()) :: Malachi.Metadata.t()
The union of every vnode's Metadata: a single flat view for whole-cluster consumers (retention,
healing) that iterate all segments. Topics/ranges/segments/offsets are disjoint across vnodes (each
topic lives on one vnode), so the union is unambiguous; with one vnode it is that vnode's metadata.
Builds an empty DS-RSM with no vnodes. Options are forwarded to HashRing.new/1
(e.g. :ring_bits).
@spec ranges_of_topic(t(), Malachi.Metadata.topic_name()) :: [ Malachi.Metadata.range_meta() ]
All ranges of a topic.
A fresh cache with vnode_ids' entries taken from previous instead: what a reader does with the
vnodes a snapshot could not reach. Their entry in fresh is an empty placeholder, and installing
that placeholder would erase topics whose records are durable on disk, turning every read of them
into a successful empty page. Keeping the last view this reader held is stale at worst, and the
reader already tolerates staleness between refreshes; it is nil metadata that has no honest
reading. A vnode absent from previous (never yet reached) keeps the placeholder, since there is no
older view to keep.
@spec seed(Malachi.Cluster.HashRing.t(), %{ required(vnode_id()) => Malachi.Metadata.t() }) :: t()
A DS-RSM over ring whose vnodes already hold metadata_by_vnode (%{vnode_id => Metadata}):
used to seed a local read cache that mirrors an authoritative Malachi.Cluster.ReplicatedDSRSM
sharing the same ring: reads are served from this cache, writes routed back through the vnodes' ra
clusters. metadata_by_vnode must key exactly the ring's vnodes.
@spec segments_of_range( t(), Malachi.Metadata.topic_name(), Malachi.Metadata.range_id() ) :: [ Malachi.Metadata.segment_meta() ]
All segments of a range under topic_name. range_id must belong to topic_name (see caller contract).
@spec single(Malachi.Metadata.t()) :: t()
A DS-RSM with a single vnode holding metadata: the trivial, unsharded shape. Seeds a one-vnode
control plane (the current single-cluster runtime and tests) before real multi-vnode sharding: with
one vnode every topic routes to it, so behavior matches a plain Malachi.Metadata.
@spec split_vnode(t(), vnode_id(), Malachi.Cluster.HashRing.token()) :: {:ok, t()} | {:error, atom()}
Adds a new vnode at token and migrates to it every topic that now routes there:
the "dynamically sharded" part of DS-RSM (rebalancing). Adding a vnode only steals an arc
from one existing vnode, so only that vnode's affected topics move; their full metadata
(topic + ranges + segments) is relocated. Range ids stay valid because they are globally
unique ({topic, seq}); migration is likewise safe for segments only if segment ids are
globally unique (the broker-assigned contract, see Malachi.Metadata). Propagates
HashRing placement errors.
@spec topic_policy_name(t(), Malachi.Metadata.topic_name()) :: Malachi.Metadata.policy_name() | nil
The NAME of the storage policy topic_name points at, or nil if it points at none or the topic is
unknown. The definition behind the name is a cluster object (Malachi.Cluster.PolicyStore).
@spec topics_bound_to(t(), Malachi.Metadata.policy_name()) :: [ Malachi.Metadata.topic_name() ]
The topics bound to the policy name on every vnode, sorted. Reads each vnode's topics where they are
rather than building merged_metadata/1, which copies every range and segment to answer a question
about one field of the topic records.
@spec update_vnode(t(), Malachi.Metadata.topic_name(), (Malachi.Metadata.t() -> {Malachi.Metadata.t(), reply})) :: {t(), reply | {:error, :no_vnode}} when reply: term()
Routes topic_name to its vnode and updates that vnode's Metadata with update_fun, the
general single-vnode mutation combinator. update_fun receives the vnode's Metadata and returns
{new_metadata, reply} (the Malachi.Metadata.apply/3 shape); the new metadata replaces the
vnode's and reply is returned as-is. {dsrsm, {:error, :no_vnode}} if the ring is empty.
command/3 is this with &Malachi.Metadata.apply(&1, command, now_ms) (pure but for the clock). A Raft-backed control plane
injects an authoritative apply here instead (see Malachi.BrokerServer), so purity/determinism
hold only when update_fun is itself pure.
@spec vnode_for(t(), Malachi.Metadata.topic_name()) :: {:ok, vnode_id()} | {:error, :empty}
The vnode id that owns topic_name, or {:error, :empty} if the ring has no vnodes.
The ids of the vnodes.