@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/2 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.

@spec new(keyword()) :: t()

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.

@spec retain_vnodes(t(), t(), [vnode_id()]) :: t()

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(t(), Malachi.Metadata.topic_name()) ::
  Malachi.Metadata.policy() | nil

The storage policy governing topic_name, or nil if none/unknown (use the globals).

@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/2 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) (pure). 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.

@spec vnode_ids(t()) :: [vnode_id()]

The ids of the vnodes.