@spec abort_split(t(), vnode_id(), Malachi.Cluster.MetadataServer.server_id()) ::
  :ok | {:error, :incomplete}

Aborts a split that a crashed coordinator left in flight, rolling it back to the pre-split state: moves every topic that reached the new vnode back to its owner under state's (unchanged) ring and lifts any migration fence left on a source: the same derived, best-effort rollback an in-call failure runs. state is the pre-split topology (a pending split never advanced the ring); new_server_id addresses the new vnode's (possibly unreachable) cluster.

Returns :ok only when the rollback is complete: the new vnode is confirmed empty (every topic moved back), so its orphan ra cluster is deleted (letting a later retry recreate it). Returns {:error, :incomplete} when the new vnode still holds topics or is unreachable: the cluster is left intact (deleting it would lose those topics) for the caller to retry: the new vnode's data is safe there, just not yet moved back. Idempotent: safe to re-run.

@spec add_vnode(t(), vnode_id(), Malachi.Cluster.HashRing.token(), [node()]) ::
  {:ok, t()} | {:error, term()}

Adds a vnode at token and starts its Raft cluster (named vnode_id) across nodes (default the local node). With several nodes the vnode is replicated and survives losing a member, HA per vnode. The stored server id addresses a real member (see MetadataServer.start/2), so a vnode placed on a subset of nodes is reachable even from a node that hosts no replica of it. Propagates ring placement errors and ra start errors.

@spec command(t(), Malachi.Metadata.topic_name(), Malachi.Metadata.command()) ::
  term()

Routes a Malachi.Metadata command to the vnode owning topic_name and submits it through that vnode's Raft log. Returns the machine reply (e.g. {:ok, root_id} or {:error, :already_exists}), {:error, :no_vnode} if the ring is empty, or {:error, {:raft, reason}} on a transport failure.

@spec complete_split(t(), vnode_id(), Malachi.Cluster.HashRing.token(), [node()]) ::
  {:ok, t()} | {:error, term()}

Resumes and completes a split whose coordinator crashed mid-way: the complete-forward counterpart of abort_split/3 (the NorthGuard "carrying it out to the end"). Re-drives the same migration as split_vnode/4, but idempotently and without rolling back on failure: ensure_started/2 reuses the new vnode's cluster if it is already up (a crash may have started it), and the migration re-drives only what is left: a topic already moved off its source is skipped, and re-fencing / re-inserting are no-ops (see Malachi.Metadata.insert_topic/2). state is the pre-split topology (a pending split never advanced the ring). On success returns the grown state; on failure returns the error leaving the partial state in place for the next resume to finish (keep-trying, so a transient outage does not undo progress).

@spec delete(t()) :: :ok

Stops and deletes every vnode's Raft cluster (removing on-disk state).

@spec known_segments(
  Malachi.Cluster.RingTopology.t() | nil,
  [Malachi.Metadata.segment_id()],
  pos_integer()
) ::
  {:ok, found()} | {:error, :no_topology | {:vnodes_unreachable, [vnode_id()]}}

Which of segment_ids the vnodes that OWN them list, with every other vnode asked only about the ids their owner did not list.

The question a caller about to act on an ABSENCE has, answered where the answer lives. Each id is routed by Malachi.Metadata.segment_routing_topic/1, the function every segment command is routed by, so the sweep asks exactly the vnode the segment was written to; and each owning vnode is read linearizably (MetadataServer.segments/2), so a segment registered a moment ago is already there: registration commits before any replica creates its directory. Asking the owner rather than a copy of everyone's metadata is NorthGuard's shape: the only global state is which vnodes exist, and a vnode's leader answers for the metadata it owns (the meetup transcript, 502-508 and 609-613). Only the owners can make an id known; the wider question below can only keep an id, and it departs from that shape for a reason docs/ARCHITECTURE.md records ("a vnode accepts metadata outside its arc").

Malachi co-locates a topic's ranges and segments on the topic's vnode, while NorthGuard routes a range by the hash of the range itself (520-522); that difference predates this function and lives in Malachi.Metadata.segment_routing_topic/1 alone, so the sweep follows it if it ever changes.

While a split is pending, a topic's metadata can already sit on the new vnode while the ring still routes it to the old one, so each id is also routed under the ring the split will install, and asked of that owner too. An id counts as known when either owner lists it. One that the split is moving and that NEITHER owner lists is returned as migrating, not as unknown: the two owners are read at two moments, and a topic copied to the new vnode and then extracted from the old one between those two reads is listed by neither, although it never stopped existing.

An id its owner does not list, and that no split is moving, is then asked of every vnode on the ring before it can count as unknown. A vnode only stores what was routed to it, but a broker routing by a ring gossip had not yet updated can write a new topic to a vnode the recorded ring no longer sends it to, and nothing moves it afterwards. Its segments are real, and one that another vnode lists is returned as misplaced, never as unknown. The wider question costs only the ids that were about to be declared absent, which on a healthy cluster are the real orphans alone.

An id that does not carry a topic cannot be routed and is returned as unroutable, never as unknown: the caller must treat it as explained. A vnode that does not answer within timeout on any node of its placement fails the whole call, because a vnode that did not answer knows nothing the caller can act on. Returns {:error, :no_topology} when there is no ring to route by.

@spec known_segments_stable(
  (-> {:ok, Malachi.Cluster.RingTopology.t() | nil} | {:error, term()}),
  [Malachi.Metadata.segment_id()],
  pos_integer()
) :: {:ok, found()} | {:error, term()}

known_segments/3 over the topology read_topology returns, read again once the owners answered.

The answers are only worth something if they came from the owners the ring still routes to. A topology that moved while the owners were being asked (a split advanced, a pending one appeared) is refused as {:topology_changed, before, after} rather than trusted, and the caller asks again on its next pass. read_topology answers {:ok, topology_or_nil} or {:error, reason}, which is returned.

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

Builds an empty replicated DS-RSM. Options are forwarded to HashRing.new/1.

@spec query(t(), Malachi.Metadata.topic_name(), (Malachi.Metadata.t() -> result)) ::
  {:ok, result} | {:error, term()}
when result: term()

Routes a linearizable query to the vnode owning topic_name. query_fun receives that vnode's Metadata state. {:error, :no_vnode} if the ring is empty.

@spec route_vnode(
  t(),
  vnode_id(),
  Malachi.Cluster.HashRing.token(),
  Malachi.Cluster.MetadataServer.server_id()
) :: {:ok, t()} | {:error, term()}

Places vnode_id at token on the ring pointing at server_id, without starting its ra cluster: the routing-only counterpart of add_vnode/4 for a node that is not the bootstrap orchestrator: the orchestrator started the cluster (across the placement nodes), and this node only routes to it. server_id must address a real member of the vnode's placement. Propagates ring placement errors.

@spec server_for(t(), vnode_id()) :: Malachi.Cluster.MetadataServer.server_id()

The ra server id of vnode_id: for routing a write to that vnode's cluster.

@spec snapshot(
  t(),
  keyword()
) :: {:ok, Malachi.Cluster.DSRSM.t(), [vnode_id()]}

Reads every vnode's replicated Metadata into a local Malachi.Cluster.DSRSM cache sharing this ring: the read-side mirror a broker threads (reads served locally; writes routed back through the vnodes' ra clusters via server_for/2).

A vnode whose cluster is not ready yet (still electing, or the orchestrator has not bootstrapped it) contributes an empty Metadata and its id is returned in the second list, so a caller can tell "this vnode holds no topics" from "this vnode did not answer". Collapsing those two is what made a restarted broker serve an empty page for a topic with durable records on disk: the unreachable vnode contributed nothing, the cache replaced the real topics with that nothing, and every read of them succeeded with zero records. Re-snapshotting later fills the vnode in (the ra log is authoritative, so a refresh only ever moves the cache forward).

The vnodes are read concurrently, and :timeout (in ms, default ra's own 5s) bounds each read. Both matter to a caller on a latency-sensitive path: read sequentially with the default, a snapshot of n silent vnodes costs 5s x n, which is what used to be paid inside the broker's own loop while it was supposed to be serving clients (#178). A read that times out is just an unreachable vnode, the case this function already reports, so bounding it adds no outcome a caller must learn.

@spec split_vnode(t(), vnode_id(), Malachi.Cluster.HashRing.token(), [node()]) ::
  {:ok, t()} | {:error, term()}

Splits the ring by adding a vnode at token (a new ra cluster on nodes) and migrating every topic that now routes to it out of its current vnode. Vnode split over real Raft (the NorthGuard model: spawn a new group and break off that half of the state). Each displaced topic is fenced on the source first (:begin_migration, so a concurrent write is rejected and cannot race the copy, seal-first), then copy-first: insert_topic into the new vnode, then extract_topic from the source (which lifts the fence), so no single failure loses a topic (a crash after the insert leaves a harmless duplicate the new ring routes past). Returns the grown state on full success; propagates a ring/start error, or {:error, {:fence | :migrate, topic, reason}} on failure: a partial split leaves its remaining fences up (writes to those topics stay blocked) for the caller/coordinator to reconcile. A topic created mid-split that routes to the new vnode is not caught here (create is not fenced); today's caller quiesces the split.

@spec topics_bound_to(
  Malachi.Cluster.RingTopology.t() | nil,
  Malachi.Metadata.policy_name(),
  pos_integer()
) ::
  {:ok, [Malachi.Metadata.topic_name()]}
  | {:error,
     :no_topology | :split_in_progress | {:vnodes_unreachable, [vnode_id()]}}

The topics bound to the storage policy name, sorted, asked of every vnode topology places.

What Malachi.Policies.delete/3 must know before it removes a policy: a topic bound to a name nothing defines holds its data, so a delete that missed a binding is a retention decision nobody made. The answer comes from the vnodes themselves, read linearizably, and never from a node's cached copy of their metadata, which misses a vnode it has not reached yet and a bind made through another node since its last refresh. The same shape as known_segments/3, the other question asked before a destructive act, with two differences that both follow from a binding living on its topic's record:

• every vnode is asked, not only the owner of some id, because the question is about every topic, and a topic written to a vnode that does not own it by the ring is still bound;
• a pending split is refused as {:error, :split_in_progress}, because a topic between its two owners can be listed by neither (extracted from the source, not yet inserted in the destination) and there is no id to recognise it by, as known_segments/3 does with moving?.

A vnode that does not answer within timeout on any node of its placement fails the whole call, as {:error, {:vnodes_unreachable, ids}}: what it would have said is exactly what the caller cannot do without.

@spec topics_bound_to_stable(
  (-> {:ok, Malachi.Cluster.RingTopology.t() | nil} | {:error, term()}),
  Malachi.Metadata.policy_name(),
  pos_integer()
) :: {:ok, [Malachi.Metadata.topic_name()]} | {:error, term()}

topics_bound_to/3 over the topology read_topology returns, read again once the vnodes answered, refused as {:topology_changed, before, after} when it moved in between, for the reason known_segments_stable/3 gives.

@spec vnode_for(t(), Malachi.Metadata.topic_name()) ::
  {:ok, vnode_id()} | {:error, :empty}

The vnode id owning topic_name, or {:error, :empty} if there are no vnodes.

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

The ids of the vnodes.