@spec active_range_ids(GenServer.server(), Malachi.Metadata.topic_name()) :: [
  Malachi.Metadata.range_id()
]

The ids of a topic's active ranges.

@spec adopt_topology(Malachi.Broker.t(), Malachi.Cluster.RingTopology.t()) ::
  Malachi.Broker.t()

Rebuilds a sharded broker's metadata routing for a new ring topology (a vnode split adopted via gossip): the local read cache takes the new ring: existing vnodes keep their cached Metadata, a newly-added vnode starts empty until the next refresh from ra, and the write path is re-routed over the topology's %{vnode_id => nodes} placements (server id = {vnode_id, a_member}). Pure: the metadata catch-up for a new/changed vnode is the separate refresh side effect. Returns the new broker.

@spec apply_heal(GenServer.server(), [Malachi.Metadata.command()]) :: :ok

Applies :set_segment_replicas healing commands to the control plane.

Returns a specification to start this module under a supervisor.

See Supervisor.

@spec commit_offset(
  GenServer.server(),
  Malachi.Metadata.group(),
  Malachi.Metadata.topic_name(),
  Malachi.Metadata.offsets()
) :: term()

Durably commits a consumer group's position for a topic; returns the control-plane reply.

@spec committed_offsets(
  GenServer.server(),
  Malachi.Metadata.group(),
  Malachi.Metadata.topic_name()
) ::
  Malachi.Metadata.offsets()

A consumer group's committed offsets for a topic (empty if it never committed).

@spec consume(
  GenServer.server(),
  Malachi.Metadata.topic_name(),
  map(),
  pos_integer(),
  non_neg_integer(),
  [term()] | nil
) :: {[Malachi.Log.Record.t()], map()} | {:error, :metadata_unavailable}

Consumes a topic's current ranges from positions, returning {records, next_positions}. When wait_ms > 0 and nothing is available yet, the call blocks (long-poll) until a produce to the topic delivers data or wait_ms elapses (then records is []). With wait_ms == 0 it returns immediately. positions/next_positions map each range id to its Broker.consume_cursor.

@spec create_topic(GenServer.server(), Malachi.Metadata.topic_name(), pos_integer()) ::
  term()

Creates a topic (and its root range); returns {:ok, root_range_id} or an error.

@spec delete_segment(GenServer.server(), Malachi.Metadata.segment_id()) :: term()

Removes a sealed segment from the control plane (retention); returns the control-plane reply.

@spec merge_ranges(
  GenServer.server(),
  Malachi.Metadata.range_id(),
  Malachi.Metadata.range_id()
) ::
  {:ok, Malachi.Metadata.range_id()} | {:error, term()}

Merges two buddy ranges; returns {:ok, child_id} or an error.

@spec metadata(GenServer.server()) :: Malachi.Metadata.t()

The current control-plane metadata (e.g. for a healing coordinator to inspect).

@spec metadata_ready?(GenServer.server(), timeout()) :: boolean()

Whether this broker has read every metadata vnode at least once since boot. Until it has, a topic whose vnode is still silent is represented by an empty placeholder in the local cache, and every read of it answers a page with no records: durable data reported as a drained topic. A node in that state can accept connections and pass /health while being unable to answer a single read honestly.

@spec produce(GenServer.server(), Malachi.Metadata.topic_name(), [
  Malachi.Log.Record.t()
]) ::
  {:ok,
   %{
     required(Malachi.Metadata.range_id()) =>
       {non_neg_integer(), non_neg_integer()}
   }}
  | {:error, term()}

Routes, replicates and commits records; returns {:ok, placements} or an error.

@spec read(
  GenServer.server(),
  Malachi.Metadata.range_id(),
  non_neg_integer(),
  pos_integer()
) ::
  {:ok, [Malachi.Log.Record.t()]} | :eof | {:error, term()}

Reads up to max_records committed records from a range, starting at offset.

@spec read_consume(
  GenServer.server(),
  Malachi.Metadata.range_id(),
  Malachi.Broker.consume_cursor(),
  pos_integer()
) ::
  {:ok, [Malachi.Log.Record.t()], Malachi.Broker.consume_cursor()}
  | {:error, term()}

Reads one cross-epoch consume page of a range, tailing the active range (see Malachi.Broker.read_consume/5).

@spec replication_ref(GenServer.server()) :: pid() | nil

The Malachi.Cluster.ReplicationServer this broker owns, or nil when it was given an external broker set (the clustered shape, where the replication server is supervised and named separately).

A single-node broker starts its own, unnamed, so this is the only way to name it as a replica: the integrity scrub needs it to tell which stored segments are its own. Ask per use rather than caching it, since a broker restart replaces the process.

@spec split_range(GenServer.server(), Malachi.Metadata.range_id()) ::
  {:ok, Malachi.Metadata.range_id(), Malachi.Metadata.range_id()}
  | {:error, term()}

Splits a range; returns {:ok, left_id, right_id} or an error.

@spec start_link(
  Path.t(),
  keyword()
) :: GenServer.on_start()

Starts a server whose segment storage is rooted at directory.

Options

• :brokers - references of the Malachi.Cluster.ReplicationServers that segments are placed on. When given, this server uses them and does not start its own; when omitted, it starts a single local replication server rooted at directory (single-node default).
• :live_brokers - a (-> [broker]) (e.g. from membership); when given, the placement broker set is refreshed from it every :brokers_refresh_interval ms, so new segments land on currently-alive brokers. An empty result is ignored (the last non-empty set is kept).
• :brokers_refresh_interval - refresh period in ms (default 1000).
• :metadata_cluster - a Raft cluster name (atom). When given, the metadata is made authoritative via that ra cluster (mutations go through the log; reads come from a local cache); ra must already be running. When omitted, metadata is in-memory (single node).
• :metadata_nodes - the nodes the metadata Raft cluster spans (default [node()]); several nodes make the control plane HA (the metadata survives losing a member).
• :replication_factor - replicas per segment (default 1; clamped to the broker count).
• :group_commit - when true, produce buffers its batch and defers the client reply to the next flush so concurrent producers coalesce into one fsync (default from app env; only active at rf=1). See the moduledoc.
• :group_commit_interval_ms - group-commit flush period in ms (default 5, or app env).
• :group_commit_flush_max_records - flush eagerly once this many records are parked, bounding the per-flush fsync (default 8000, or app env).
• :group_commit_max_inflight - past this many parked records, shed produces with :overloaded instead of dropping the connection (default 200000, or app env).
• :segment_max_bytes - byte threshold at which the active segment seals and rolls.
• remaining options are forwarded to a started Malachi.Cluster.ReplicationServer (segment log options such as :max_bytes, :flush_bytes, :index_interval); ignored with :brokers.
• standard GenServer options (:name, etc.) are honored.

@spec stop(GenServer.server()) :: :ok

Stops the server (and its replication storage).

@spec stream_ack(
  GenServer.server(),
  Malachi.Metadata.topic_name(),
  Malachi.Metadata.group(),
  Malachi.Metadata.offsets(),
  non_neg_integer(),
  [term()] | nil,
  GenServer.server() | nil
) :: :ok

Acks count streamed records of topic/group at positions (a decoded cursor): returns that much window credit (unblocking further pushes) and durably commits the group's position. Returns :ok.

@spec stream_history(
  GenServer.server(),
  Malachi.Metadata.range_id(),
  Malachi.Broker.history_cursor(),
  pos_integer()
) ::
  {:ok, [Malachi.Log.Record.t()], Malachi.Broker.history_cursor()}
  | {:error, term()}

Streams one bounded page of a range's cross-epoch history (see Malachi.Broker.stream_history/5).

@spec subscribe(
  GenServer.server(),
  Malachi.Metadata.topic_name(),
  Malachi.Metadata.group(),
  pos_integer(),
  pos_integer(),
  keyword()
) :: :ok

Subscribes the calling process as a streaming consumer of topic for consumer group: records are pushed to it as {:log_records, topic, records, next_positions} as they are produced, resuming from the group's committed position and bounded by a credit window (at most window records in flight, at most max per push). Ack with stream_ack/5 to return credit and commit progress. :ok.

@spec sync(GenServer.server()) :: :ok

No-op: writes are already durable on return. Kept for API compatibility.

The per-topic overview (Malachi.Metadata.overview/1) annotated with each topic's failure-domain violation count (Malachi.Broker.domain_violations/2), computed from a single merged-metadata view.

@spec unsubscribe(GenServer.server(), Malachi.Metadata.topic_name()) :: :ok

Removes the calling process's streaming subscription to topic. Returns :ok.