@spec command(server_id(), Malachi.Metadata.command(), timeout()) ::
  {:ok, term()} | {:error, term()}

Submits a Malachi.Metadata command through the Raft log; returns the machine reply.

@spec delete(server_id() | cluster_name()) :: :ok | {:error, term()}

Stops and deletes the vnode's Raft cluster (removing its on-disk state). Prefer passing a server_id addressing a real member (as start/2 returns and the cluster callers already hold), so a vnode placed on a subset of nodes is deleted through a node that actually hosts it: :ra finds the leader from there and propagates the deletion to every member. A bare cluster_name is accepted as {cluster_name, node()} for the single-node case (tests, local setups). Returns {:error, reason} when the deletion cannot be committed, instead of reporting :ok regardless.

@spec ensure_started(cluster_name(), [node()]) ::
  {:ok, server_id()} | {:error, term()}

Like start/2, but idempotent: if the cluster is already running (reachable via :ra.members), returns its server_id without restarting; otherwise starts it. This is what resuming a split needs: a coordinator that crashed after starting the new vnode's cluster must be able to re-drive the split without start/2 failing on an already-formed cluster. Returns {:error, reason} only when the cluster is neither running nor startable (e.g. its placement nodes are unreachable), so the caller can retry.

@spec leader?(server_id(), timeout()) :: boolean()

Whether server_id is currently the leader of its Raft cluster. :ra.members (answered by any reachable member) reports the leader as a server_id; this returns true iff it equals server_id itself, so pass the local server id ({cluster_name, node()}) to ask "does this node lead this vnode?". Unreachable/unformed clusters answer false (never assume leadership). Used by 1C-b to run a vnode's coordinators only on the node that leads its Raft group (the NorthGuard-faithful placement).

timeout_ms bounds the leader call. A caller that asks about many vnodes on one pass (the vnode coordinator manager) passes a timeout well below its own period: a vnode mid-election answers false either way, and waiting ra's default would let one such vnode hold up the whole pass.

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

Reads the replicated Metadata state with a linearizable (consistent) read and returns query_fun applied to it. query_fun receives the Metadata state (e.g. &Malachi.Metadata.get_topic(&1, name)), and is applied here, in the calling process, not inside the ra server.

That split is forced by ra, which only accepts an {M, F, A} for a consistent query and applies it as apply(M, F, A ++ [State]), appending the state as the last argument. Our projections take the state first, so no direct translation exists that does not either flip their arguments or grow a layer of arity-shuffling wrappers. Asking for the state itself and projecting locally costs nothing in practice, since every caller outside the test suite asks for the whole state anyway, and it keeps a raising query_fun from taking the replicated server down with it.

A failed read never reaches query_fun: an error or a timeout is returned as-is rather than projected over a stand-in state.

timeout bounds the wait and defaults to ra's 5s. The reconcile path passes a short one, because a vnode that does not answer must cost milliseconds there, not five seconds.

@spec ready?(server_id(), timeout()) :: boolean()

Whether the cluster addressed by server_id is formed and reachable (a member answers :ra.members) within timeout. Used by the reconcile to decide if a vnode still needs bootstrapping, which is why it takes a bound: an unreachable placement node costs the whole timeout.

@spec segments(server_id(), timeout()) ::
  {:ok, %{required(Malachi.Metadata.segment_id()) => term()}} | {:error, term()}

The vnode's segments, read linearizably and without copying the rest of its state.

Projected inside the leader with {:maps, :with, [[:segments]]}, a standard library function every node has whatever Malachi version it runs: Malachi.Cluster.RaCluster.project/3 explains why a Malachi function must never be the one the leader applies. What a caller does with the map it gets back (Malachi.Retention.Orphans.known_among/2) runs in the caller.

@spec start(cluster_name(), [node()]) :: {:ok, server_id()} | {:error, term()}

Starts a Raft cluster named cluster_name running the metadata machine across nodes (default the local node), and returns a server_id for a real member: the local node when it is one, otherwise the first of nodes. (The starter need not be a member: a sharded control plane places a vnode on a subset of nodes, so the node bootstrapping it may not host a replica; ra still routes commands/queries from that member to the leader.) With several nodes the metadata is replicated and survives the loss of a member. ra must be running on every node (:ra.start_in/1).

@spec topics(server_id(), timeout()) ::
  {:ok, %{required(Malachi.Metadata.topic_name()) => term()}} | {:error, term()}

The vnode's topic records, read linearizably and without copying the rest of its state.

Projected inside the leader with {:maps, :with, [[:topics]]}, for the reason segments/2 gives: a Malachi function must never be the one the leader applies. Which of them are bound to a policy is decided in the caller (Malachi.Cluster.ReplicatedDSRSM.topics_bound_to/3).