Malachi.
View Source
The deterministic state machine behind a NorthGuard vnode/coordinator: the durable, replicated source of truth for metadata about topics, ranges and segments.
Unlike the data-plane storage (Malachi.Broker/Log, which hold open file handles),
this holds only metadata: it is pure data and a pure transition function. All
mutations go through apply/3 (command -> {new_state, reply}), exactly the contract a
Raft machine's apply needs, so the ra integration replicates this state without
changing it. This is what makes topic structure durable.
Determinism
apply/3 must be deterministic so every replica reaches the same state from the same
command log: no wall-clock time and no random/process-unique values are generated inside
it. New range ids come from a counter in the state (next_range_id); a seal's time and the
broker-chosen segment ids are supplied in the command by the proposer, and a registered segment's
opened_at is the at_ms argument, which inside ra is the timestamp the leader wrote into the log
entry (see apply/3). Bad input returns an error tuple rather than raising (a raise would crash a
replica).
Reads are plain functions over the state (get_topic/2, ranges_of_topic/2, …), not
commands.
Summary
Types
A consumer group's name.
Identifies a consumer group's position for a topic.
A consumer's position across the ranges of a topic.
A policy's name. The shape it names lives in Malachi.Cluster.Policy.
A consumer's per-range stream position. Opaque to the metadata (it only stores/returns it); set
by the log layer as a Malachi.Broker.consume_cursor, :start or {source_index, source_offset}.
A topic's complete metadata, extracted for migration between vnodes. export_format is the machine
version whose state shape the export carries (see export_format/0); an export logged before the
key existed has none and counts as 0.
Functions
The active ranges of a topic (the ones that currently tile the keyspace).
Applies a command, returning {new_state, reply}. Deterministic: the same command on the
same state, at the same at_ms, always yields the same result on every replica. On failure the
state is returned unchanged with an {:error, reason} reply.
The topic a command belongs to, for routing it to the owning vnode, or nil if it names none.
Every command shape, mapped to the machine version that introduced it (see Malachi.Cluster.MachineVersion).
A consumer group's committed offsets for a topic, or %{} if it has never committed.
The state shape version a topic export carries (see topic_export/0).
A read-only snapshot of a topic's full metadata: topic + all its ranges/segments + its consumer
groups' committed offsets (keyed by group, the topic implied): as a topic_export, or nil if the
topic is absent. This is what a copy-first vnode split reads and insert_topic/2s into the new
vnode before extract_topic/2 removes it from the source, so a failed migration never loses a topic.
Removes a topic and all its ranges/segments, and its consumer groups' committed offsets:
from state, returning {state_without_topic, export} (or {state, nil} if the topic is
absent). The export can be re-inserted on another vnode with insert_topic/2: this is how
a vnode split migrates a topic's metadata to a new vnode. Offsets ride along (keyed by group,
the topic implied) so a consumer group keeps its committed position across a split.
The range metadata, or nil.
The segment metadata, or nil.
The topic metadata, or nil.
Inserts a topic export (from extract_topic/2) into state. Range ids are globally
unique ({topic, seq}), so merged ranges never collide with another topic's.
An empty metadata state.
A read-only, JSON-serializable summary of the whole log stack for the ops dashboard: one entry per
topic (sorted by name) with state, keyspace, policy, range/segment counts, total bytes, and the
consumer groups that have committed a position. Pure. This is the light payload the dashboard streams
every tick; the per-range/segment drill-down is fetched on demand via topic_detail/2.
All ranges of a topic (any state). O(k) via the topic_ranges index, see apply/3.
The topics a range/segment command's ids belong to (structurally), or [] for any other command.
Whether a command carries a range/segment id belonging to a topic other than topic_name, the routing
layer's mismatch check. True when any recognized id's topic differs; false for a command that names its
topic directly or whose ids all match. Shared by Malachi.Cluster.DSRSM and its replicated counterpart.
The seal command for segment ending at end_offset with byte_size bytes on disk, at at.
The topic a segment's metadata is routed by, or nil for an id that carries none.
All segments of a range. O(k) via the range_segments index, see apply/3.
The JSON-serializable drill-down for a single name: its ranges (sorted by seq), each with its
segments (sorted by start_offset). nil if the topic does not exist. Pure: the on-demand counterpart
to overview/1, so the per-second stream stays light and segment detail is fetched only when a topic
is expanded. range_id/segment_id tuples are flattened to display fields (seq, start_offset).
The NAME of the policy topic points at, or nil if it points at none or the topic is unknown.
The topics bound to the policy name, sorted.
Types
@type broker() :: term()
@type command() :: {:create_topic, topic_name(), pos_integer()} | {:seal_topic, topic_name()} | {:delete_topic, topic_name()} | {:split_range, range_id()} | {:merge_ranges, range_id(), range_id()} | {:register_segment, range_id(), segment_id(), [broker()], non_neg_integer()} | {:seal_segment, segment_id(), non_neg_integer(), non_neg_integer(), non_neg_integer()} | {:delete_segment, segment_id()} | {:set_segment_replicas, segment_id(), [broker()]} | {:set_segment_replicas, segment_id(), [broker()], [broker()]} | {:commit_offset, group(), topic_name(), offsets()} | {:bind_topic_policy, topic_name(), policy_name() | nil} | {:extract_topic, topic_name()} | {:insert_topic, topic_export()} | {:begin_migration, topic_name()} | {:end_migration, topic_name()}
@type group() :: String.t()
A consumer group's name.
@type group_topic() :: {group(), topic_name()}
Identifies a consumer group's position for a topic.
A consumer's position across the ranges of a topic.
@type policy_name() :: Malachi.Cluster.Policy.name()
A policy's name. The shape it names lives in Malachi.Cluster.Policy.
@type position() :: :start | {non_neg_integer(), non_neg_integer()}
A consumer's per-range stream position. Opaque to the metadata (it only stores/returns it); set
by the log layer as a Malachi.Broker.consume_cursor, :start or {source_index, source_offset}.
@type range_id() :: {topic_name(), non_neg_integer()}
@type range_meta() :: %{ id: range_id(), topic: topic_name(), key_start: non_neg_integer(), key_end: non_neg_integer(), keyspace_size: pos_integer(), state: :active | :sealed, parents: [range_id()] }
@type segment_id() :: term()
@type segment_meta() :: %{ id: segment_id(), range_id: range_id(), replica_set: [broker()], state: :active | :sealed, start_offset: non_neg_integer(), length: non_neg_integer() | nil, byte_size: non_neg_integer() | nil, sealed_at: non_neg_integer() | nil, opened_at: non_neg_integer() | nil }
@type t() :: %Malachi.Metadata{ committed_offsets: %{required(group_topic()) => offsets()}, migrating: %{required(topic_name()) => true}, range_segments: %{required(range_id()) => MapSet.t(segment_id())}, ranges: %{required(range_id()) => range_meta()}, segments: %{required(segment_id()) => segment_meta()}, topic_ranges: %{required(topic_name()) => MapSet.t(range_id())}, topics: %{required(topic_name()) => topic_meta()} }
@type topic_export() :: %{ :topic => topic_meta(), :ranges => %{required(range_id()) => range_meta()}, :segments => %{required(segment_id()) => segment_meta()}, :offsets => %{required(group()) => offsets()}, optional(:export_format) => non_neg_integer() }
A topic's complete metadata, extracted for migration between vnodes. export_format is the machine
version whose state shape the export carries (see export_format/0); an export logged before the
key existed has none and counts as 0.
@type topic_meta() :: %{ name: topic_name(), keyspace_size: pos_integer(), state: :active | :sealed, next_range_seq: non_neg_integer(), policy: policy_name() | nil }
@type topic_name() :: String.t()
Functions
@spec active_ranges_of_topic(t(), topic_name()) :: [range_meta()]
The active ranges of a topic (the ones that currently tile the keyspace).
@spec apply(t(), command(), non_neg_integer() | nil) :: {t(), term()}
Applies a command, returning {new_state, reply}. Deterministic: the same command on the
same state, at the same at_ms, always yields the same result on every replica. On failure the
state is returned unchanged with an {:error, reason} reply.
at_ms is the epoch ms the command was accepted at, and only register_segment reads it: it
becomes the new segment's opened_at. Inside ra it is the timestamp the leader wrote into the log
entry (Malachi.Cluster.MetadataMachine), so every replica and every replay stamps the same value
without the command carrying it. A cache applying a command it already submitted passes its own
clock, and the next re-seed replaces that guess with the replicated value. nil (the two argument
form, for callers with no clock) leaves opened_at unset, which Malachi.Cluster.SegmentRoll
treats as already due.
register_segment requires the segment_id to be globally unique across the cluster
(the broker-assigned contract). Within a vnode this is checked (:segment_exists), but
uniqueness across vnodes is the caller's responsibility: it is what keeps a topic's
segments safe when it migrates to another vnode (see insert_topic/2).
@spec command_target_topic(command()) :: topic_name() | nil
The topic a command belongs to, for routing it to the owning vnode, or nil if it names none.
Structural, no state: a topic command names its topic; a range_id is {topic, seq} and a
segment_id is {range_id, seq}, so the topic is elem/2 away. The broker uses this to dispatch.
Every command shape, mapped to the machine version that introduced it (see Malachi.Cluster.MachineVersion).
@spec committed_offsets(t(), group(), topic_name()) :: offsets()
A consumer group's committed offsets for a topic, or %{} if it has never committed.
@spec export_format() :: non_neg_integer()
The state shape version a topic export carries (see topic_export/0).
@spec export_topic(t(), topic_name()) :: topic_export() | nil
A read-only snapshot of a topic's full metadata: topic + all its ranges/segments + its consumer
groups' committed offsets (keyed by group, the topic implied): as a topic_export, or nil if the
topic is absent. This is what a copy-first vnode split reads and insert_topic/2s into the new
vnode before extract_topic/2 removes it from the source, so a failed migration never loses a topic.
@spec extract_topic(t(), topic_name()) :: {t(), topic_export() | nil}
Removes a topic and all its ranges/segments, and its consumer groups' committed offsets:
from state, returning {state_without_topic, export} (or {state, nil} if the topic is
absent). The export can be re-inserted on another vnode with insert_topic/2: this is how
a vnode split migrates a topic's metadata to a new vnode. Offsets ride along (keyed by group,
the topic implied) so a consumer group keeps its committed position across a split.
@spec get_range(t(), range_id()) :: range_meta() | nil
The range metadata, or nil.
@spec get_segment(t(), segment_id()) :: segment_meta() | nil
The segment metadata, or nil.
@spec get_topic(t(), topic_name()) :: topic_meta() | nil
The topic metadata, or nil.
@spec insert_topic(t(), topic_export()) :: t()
Inserts a topic export (from extract_topic/2) into state. Range ids are globally
unique ({topic, seq}), so merged ranges never collide with another topic's.
Segment ids, however, are caller-supplied and independent of range ids: this merges them
by id, so a segment id that already exists in state is overwritten. Migration is
therefore safe only if segment ids are globally unique across the cluster (the
broker-assigned contract, see register_segment in apply/3).
@spec new() :: t()
An empty metadata state.
A read-only, JSON-serializable summary of the whole log stack for the ops dashboard: one entry per
topic (sorted by name) with state, keyspace, policy, range/segment counts, total bytes, and the
consumer groups that have committed a position. Pure. This is the light payload the dashboard streams
every tick; the per-range/segment drill-down is fetched on demand via topic_detail/2.
@spec ranges_of_topic(t(), topic_name()) :: [range_meta()]
All ranges of a topic (any state). O(k) via the topic_ranges index, see apply/3.
@spec routed_range_topics(command()) :: [topic_name()]
The topics a range/segment command's ids belong to (structurally), or [] for any other command.
This is the narrower half of command_target_topic/1, and the one the routing guard uses: only a
command carrying a range_id/segment_id can be pointed at a co-located topic's range, so only these
can mismatch the topic they were routed by (:range_topic_mismatch, in Malachi.Cluster.DSRSM). A
command that names its topic directly (create/seal/delete/commit) targets exactly that topic and is not
guarded. merge_ranges carries two range ids and register_segment a range id and a segment id,
so each yields two topics and both must match where it was routed. An id of an unrecognized shape
drops out (no topic to check) rather than raising.
Distinct from the private command_topic/2, which resolves a range's stored topic from the state
for the migration fence: that one needs the range to exist; this reads the id itself, catching a
mismatch before any lookup.
@spec routed_to_foreign_topic?(command(), topic_name()) :: boolean()
Whether a command carries a range/segment id belonging to a topic other than topic_name, the routing
layer's mismatch check. True when any recognized id's topic differs; false for a command that names its
topic directly or whose ids all match. Shared by Malachi.Cluster.DSRSM and its replicated counterpart.
@spec seal_command( %{ :id => segment_id(), :start_offset => non_neg_integer(), optional(atom()) => any() }, non_neg_integer(), non_neg_integer(), non_neg_integer() ) :: command()
The seal command for segment ending at end_offset with byte_size bytes on disk, at at.
One constructor rather than two, because both seal paths (the produce roll in Malachi.Broker and
primary failover in Malachi.Cluster.Failover) turn an END OFFSET into a LENGTH the same way, and a
second copy of that subtraction is a copy that can drift.
@spec segment_routing_topic(term()) :: topic_name() | nil
The topic a segment's metadata is routed by, or nil for an id that carries none.
The one place that decides which vnode owns a segment. The seal, delete and replica commands are
routed through it (command_target_topic/1). register_segment is routed by the range it joins, and
the routing guard refuses one whose segment id this function sends to another topic
(routed_range_topics/1), so a registration lands where this function points. The orphan sweep asks
the owner through it too (Malachi.Cluster.ReplicatedDSRSM.known_segments/3). Today a topic's ranges and segments are
co-located on the topic's vnode, so this is the topic the id embeds. NorthGuard routes a range by the
hash of the range itself (see Malachi.Cluster.DSRSM); if that sharding lands here, it lands in this
function, and the sweep follows.
@spec segments_of_range(t(), range_id()) :: [segment_meta()]
All segments of a range. O(k) via the range_segments index, see apply/3.
@spec topic_detail(t(), topic_name()) :: map() | nil
The JSON-serializable drill-down for a single name: its ranges (sorted by seq), each with its
segments (sorted by start_offset). nil if the topic does not exist. Pure: the on-demand counterpart
to overview/1, so the per-second stream stays light and segment detail is fetched only when a topic
is expanded. range_id/segment_id tuples are flattened to display fields (seq, start_offset).
@spec topic_policy_name(t(), topic_name()) :: policy_name() | nil
The NAME of the policy topic points at, or nil if it points at none or the topic is unknown.
The binding stays here, and only the binding: which policy a topic uses is a fact about that topic,
so it travels with the topic when a vnode split moves it. Malachi.Cluster.PolicyStore resolves the
name to the definition.
@spec topics_bound_to(t(), policy_name()) :: [topic_name()]
The topics bound to the policy name, sorted.
What an administrator deleting a policy needs to see first: every topic here would stop expiring once
the name no longer resolves. Sorted explicitly, because state.topics iterates in hash order above 32
keys.