@spec command(server_id(), Malachi.Metadata.command()) ::
  {: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()) :: 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).

@spec query(server_id(), (Malachi.Metadata.t() -> result)) ::
  {: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.

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

Whether the cluster addressed by server_id is formed and reachable (a member answers :ra.members). Used by the reconcile loop to decide if a vnode still needs bootstrapping.

@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).