@spec broker_attributes_for([{term(), node()}], (term() -> map())) :: %{
  required({module(), node()}) => map()
}

Maps membership members to %{ReplicationServer ref => attributes}, keyed by node.

@spec broker_refs([node()]) :: [{module(), node()}]

The ReplicationServer references (one per node) that segment replicas are placed across.

@spec data_plane_opts(atom() | nil, [node()]) :: keyword()

The data-plane options for Malachi.BrokerServer: none when cluster is nil (the BrokerServer owns a single local ReplicationServer, replication_factor 1); otherwise it places segment replicas across every node's named ReplicationServer with the configured replication factor.

@spec desired_placement(atom(), pos_integer(), [node()], pos_integer(), keyword()) ::
  [
    {atom(), non_neg_integer(), [node()]}
  ]

The desired vnode placement for a given set of (live) nodes: the count logical vnodes of control plane base (sharded_vnodes/2, node-independent) placed over nodes by rendezvous hashing (place_vnodes/4). Deterministic and, crucially for rebalancing, minimal-movement, because the vnode ids/tokens are fixed and HRW is stable, adding or removing a node re-places only the vnodes that must move (a vnode changes only if it adopts a joined node, or held a left one), leaving the rest put.

This is the target that the rebalancing plan diffs the current placement against: recomputing it over the live membership (vs the static :log_nodes) is how the ring follows nodes joining/leaving. place_opts is forwarded for rack/zone spread. Pure: the caller supplies nodes (e.g. the live members), so a coordinated recompute stays deterministic across the cluster.

@spec leading_vnodes(
  [{atom(), non_neg_integer(), [node()]}],
  node(),
  (Malachi.Cluster.MetadataServer.server_id() -> boolean())
) :: [atom()]

The vnode ids this_node both hosts (its placement includes this_node) and currently leads (its Raft group's leader is the local server {vnode_id, this_node}), given the placement vnodes ([{vnode_id, token, nodes}], as stored in the bootstrap) and a leader? predicate over a local server id (defaults to MetadataServer.leader?/1). This is where 1C-b runs each vnode's retention/healing coordinators, so every vnode is managed by exactly one node: the one leading its Raft group: distributing the control-plane work the NorthGuard-faithful way (vs 1C-a's single membership leader). Pure given leader?; deterministic given the current leadership.

@spec live_rebalance_plan(
  [{atom(), non_neg_integer()}],
  (atom() -> {:ok, [node()]} | {:error, term()}),
  [node()],
  pos_integer(),
  keyword()
) :: [%{vnode_id: atom(), add: [node()], remove: [node()]}]

The rebalancing plan for the live cluster (R3-b-i): diffs the current placement, read from the vnodes' ra memberships via members_of (readable_placement/2), against the placement desired over alive_nodes (the live members), for the readable vnodes only, so an unreachable vnode is neither planned for nor recomputed. rf/place_opts are the replication factor and spread options. Reads the world through the members_of seam (real: :ra.members); pure otherwise. The result feeds Malachi.Cluster.Rebalance.apply_plan/4 (the commit), run under the lease.

@spec live_replication_refs([{term(), node()}]) :: [{module(), node()}]

Maps membership members ({name, node}) to their nodes' ReplicationServer references.

@spec membership_leader(GenServer.server()) :: (-> boolean())

The bootstrap-orchestrator policy used at runtime: a (-> boolean()) true only on the lowest-sorted live member (per SWIM membership on membership_server), so the role fails over when the current leader dies: unlike static_seed/1, which is fixed to the lowest configured node. Conservative: if membership is unavailable it returns false (never risking two orchestrators); a transient double-leader is still fenced by the ra cluster name at bootstrap.

@spec membership_seeds([node()]) :: [{module(), node()}]

The membership seeds this node joins with: the other nodes' membership servers (not self).

@spec metadata_cluster_opts(atom() | nil, [node()]) :: keyword()

The metadata options passed to Malachi.BrokerServer for the given control-plane cluster and nodes: none (single-node in-memory) when cluster is nil, otherwise the ra-backed cluster. Pure: the ra runtime is started separately (see log_children/0).

@spec parse_attributes(String.t() | nil) :: %{optional(String.t()) => String.t()}

Parses this node's broker attributes from a "key=value,key2=value2" string into a %{} of string k/v (nil/"" → %{}). Entries without exactly one = are ignored; keys and values are trimmed.

@spec parse_topology(String.t() | nil) :: %{optional(node()) => String.t()}

Parses a "node1=rack_a,node2=rack_b" cluster topology string into %{node => value} (node atom → attribute string, e.g. rack/zone): the static per-node placement attribute used by place_vnodes/4 for rack-aware vnode placement. nil/"" => %{}. Entries without exactly one = are ignored; node and value are trimmed. Node names come from a trusted operator (deploy config), so String.to_atom is fine here. Being static config identical on every node keeps placement deterministic.

@spec place_vnodes([{atom(), non_neg_integer()}], [node()], pos_integer(), keyword()) ::
  [
    {atom(), non_neg_integer(), [node()]}
  ]

Assigns each vnode ({vnode_id, token}) the nodes its ra cluster lives on: the replication_factor nodes chosen from nodes by rendezvous hashing (the same HRW that places segment replicas), so vnodes spread across the cluster and a node join/leave moves the fewest vnodes. Returns {vnode_id, token, nodes}. The effective replica count is min(replication_factor, length(nodes)).

place_opts selects the placement policy: [spread: {attribute_key, node_attributes}] keeps a vnode's replicas in distinct racks/zones (A1, topology-aware: losing a whole rack does not take a majority of a vnode's replicas); [max_skew: n] instead balances the whole set's load across nodes (A2, Placement.place_balanced/4, a global decision), so no node is overloaded. The two are mutually exclusive (standalone A2). Pure and deterministic: every node computes the same placement. Used by the sharded control plane so vnode leaders land on different nodes.

Called before the application stops (SIGTERM / bin/malachi stop). Quiesces the acceptor, drains in-flight for a bounded window, then closes connections, see Malachi.Shutdown.

@spec readable_placement([{atom(), non_neg_integer()}], (atom() ->
                                                     {:ok, [node()]}
                                                     | {:error, term()})) ::
  [{atom(), non_neg_integer(), [node()]}]

The current placement read from the vnodes' live ra memberships: for each {vnode_id, token} in vnode_configs, members_of.(vnode_id) returns {:ok, nodes} (its current member nodes) or {:error, _}. A vnode whose membership cannot be read is omitted, conservative: we never plan a change for a vnode we cannot currently see. Returns [{vnode_id, token, nodes}]. Pure given the seam.

@spec rebalance_plan([{atom(), non_neg_integer(), [node()]}], [
  {atom(), non_neg_integer(), [node()]}
]) ::
  [%{vnode_id: atom(), add: [node()], remove: [node()]}]

The rebalancing plan: the per-vnode membership changes to move from the current placement to the desired one (from desired_placement/5), staged: it computes what to change without applying anything. For each vnode whose node set differs it yields %{vnode_id:, add:, remove:}, where add are the nodes to join that vnode's ra cluster and remove the ones to drop; vnodes that already match are omitted (so an empty plan means "nothing to do"). R3 executes each change add-before-remove, so a vnode never drops below its replica count mid-move (when the replication factor is unchanged, add and remove are equal-sized).

Assumes current and desired cover the same vnode ids (rebalancing follows membership; changing the vnode count is re-sharding, out of scope). Deterministic: the plan follows desired's order. Pure.

@spec retention_policy() :: Malachi.Cluster.Retention.policy()

The configured retention policy (:max_age_ms / :max_bytes; nil = that rule is off).

@spec sharded_vnodes(atom(), pos_integer()) :: [{atom(), non_neg_integer()}]

The count vnodes ({cluster_name, token}) of a sharded control plane, named from base and spread evenly over the 32-bit ring. Each vnode is its own ra cluster; place_vnodes/4 assigns each one :log_vnode_replication_factor member nodes (HA per vnode). Pure.

Callback implementation for Application.start/2.

@spec static_seed([node()]) :: (-> boolean())

The bootstrap-orchestrator policy for the sharded control plane: a (-> boolean()) that is true only on the deterministic seed node (the lowest-sorted of nodes), so exactly one node starts each vnode's ra cluster and the rest only route to it. Pure. D-c-1d swaps this for a membership-leader policy with fencing (the mature, k8s-style approach).

@spec try_members([node()], (node() -> {:ok, [node()]} | :error)) ::
  {:ok, [node()]} | {:error, :unreachable}

Tries each node in nodes as an entry point, returning the first {:ok, members} from members_fun (node -> {:ok, members} | :error), or {:error, :unreachable} if none answers, how the rebalancing coordinator finds a live member of a vnode it may not host locally. Pure given members_fun.

@spec vnode_metadata_source(atom()) :: (-> Malachi.Metadata.t())

A metadata_source bound to a single vnode: reads this node's view of the vnode's replicated Metadata via a consistent query to its ra cluster ({vnode_id, node()}). Tolerant, an unreachable or not-yet-formed vnode yields empty Metadata (no work) instead of crashing the coordinator. Used by 1C-b so each vnode's retention/heal coordinator sees only that vnode's shard, versus 1C-a's global merge routed through the single membership leader.