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/2 (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/2 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); anything else
non-deterministic (timestamps, broker-chosen segment ids) is supplied in the command by
the proposer. 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 named storage policy: per-topic retention overrides and a placement spread attribute. Both keys are optional; a policy applies only the ones it sets, falling back to the global defaults otherwise.
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.
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 always yields the same result on every replica. On failure the state is
returned unchanged with an {:error, reason} reply.
A consumer group's committed offsets for a topic, or %{} if it has never committed.
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 policy named name, or nil if undefined.
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/2.
All segments of a range. O(k) via the range_segments index, see apply/2.
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 policy governing topic (its associated policy), or nil if none/unknown (use globals).
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()]} | {:commit_offset, group(), topic_name(), offsets()} | {:define_policy, policy_name(), policy()} | {:set_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() :: %{ optional(:retention) => %{ optional(:max_age_ms) => non_neg_integer() | nil, optional(:max_bytes) => non_neg_integer() | nil }, optional(:spread_by) => term() | nil }
A named storage policy: per-topic retention overrides and a placement spread attribute. Both keys are optional; a policy applies only the ones it sets, falling back to the global defaults otherwise.
@type policy_name() :: String.t()
@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 }
@type t() :: %Malachi.Metadata{ committed_offsets: %{required(group_topic()) => offsets()}, migrating: %{required(topic_name()) => true}, policies: %{required(policy_name()) => policy()}, 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()} }
A topic's complete metadata, extracted for migration between vnodes.
@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).
Applies a command, returning {new_state, reply}. Deterministic: the same command on the
same state always yields the same result on every replica. On failure the state is
returned unchanged with an {:error, reason} reply.
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 committed_offsets(t(), group(), topic_name()) :: offsets()
A consumer group's committed offsets for a topic, or %{} if it has never committed.
@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_policy(t(), policy_name()) :: policy() | nil
The policy named name, or nil if undefined.
@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/2).
@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/2.
@spec segments_of_range(t(), range_id()) :: [segment_meta()]
All segments of a range. O(k) via the range_segments index, see apply/2.
@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(t(), topic_name()) :: policy() | nil
The policy governing topic (its associated policy), or nil if none/unknown (use globals).