@spec boot_topology(atom() | nil, [node()], (-> term())) ::
  {Malachi.Cluster.RingTopology.t() | nil, term()}

The routing topology a clustered node boots with. This is the precedence rule of issue #32: the ring recorded in Malachi.Cluster.RingServer wins over MALACHI_LOG_VNODES, and the environment only ever seeds a cluster the store affirms has never had a ring.

Before this, the ring was gossiped state only, so a full-cluster restart reseeded it from the environment. The environment's even geometry does not match a ring grown by splitting, so a cluster that had been resharded came back routing topics to vnodes that no longer held their metadata.

The ring cluster is formed here, before the read, and its membership comes from the static MALACHI_LOG_NODES, never from the ring itself. That is what keeps the bootstrap acyclic: this Raft group does not live in the vnodes it describes.

before_seed runs only when the environment is about to seed the store, before anything is written: it is where the boot refuses a seed (Malachi.Storage.DataDirGuard.check_seed/3), and what it answered is returned beside the topology (:ok when no seed was due).

Returns {nil, :ok} for an unclustered node (no ring exists) and for a clustered one that is genuinely unsharded. Raises when the store cannot be read within MALACHI_LOG_RING_BOOT_TIMEOUT_MS: refusing to boot is deliberate, because a node that cannot see the ring cannot know which vnode owns which arc, and serving on a guess is the corruption being fixed.

@spec boot_vnodes(Malachi.Cluster.RingTopology.t() | nil) ::
  [{atom(), non_neg_integer(), [node()]}] | nil

The sharded vnode placement ([{vnode_id, token, nodes}]) described by a boot topology, or nil when there is nothing to shard. This is the shape the rest of the sharded wiring already speaks, so a ring restored from the durable store drops into exactly the path an environment-derived one used to take.

@spec bound_topics(String.t()) :: {:ok, [String.t()]} | {:error, term()}

The topics bound to the storage policy name, asked of whoever owns the truth about them, for Malachi.Policies.delete/3.

• Clustered: the vnodes themselves, routed by the topology of record in the durable ring store (Malachi.Cluster.ReplicatedDSRSM.topics_bound_to_stable/3), or the one metadata cluster when the control plane is not sharded. Never this node's cached copy of the metadata, for the reason the orphan sweep gives (#249): it misses a vnode this node has not reached and a bind made through another node since its last refresh.
• The data-plane sharding measurement mode: every shard's own in-memory metadata, the only copy there is (and one a restart empties). A single node is clustered, with one member, and asks it.

@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 cluster_flags() ::
  {:ok, %{known: [atom()], enabled: [atom()]}} | {:error, term()}

What this cluster knows and what it has switched on: %{known: [flag], enabled: [flag]}, read through consensus so an operator never sees a stale answer to a question they are about to act on.

@spec cluster_flags_child([node()]) :: Supervisor.child_spec()

The flag store's reconciler child, which is also its machine-version watcher and the local flag pass.

On every tick it self-joins this node when clustered, checks the store's machine version, and runs Malachi.Cluster.ClusterFlagsCache.refresh/1: refresh the cache, adopt a newly enabled flag, and refuse to serve when an enabled flag names a capability this build lacks.

Deliberately not store_reconciler_child/5: that helper replaces the reconcile with a no-op on a single-node deployment, and the flag pass has to run there too. A one-node deployment still commits to a format-changing feature, and still must not come back on a build that cannot read what it wrote.

@spec console_children(boolean(), :inet.port_number()) :: [
  Supervisor.child_spec() | {module(), term()}
]

The console endpoint on port, with the registry its header deadlines live in (Malachi.Console.HeaderDeadline), or nothing when the console is switched off. It is the last child, so it never delays one the broker needs, and a port it cannot open makes it :ignore rather than failing the boot (see Malachi.Console.Endpoint).

@spec current_ring() :: Malachi.Cluster.HashRing.t() | nil

The cluster's current routing ring, from the gossiped topology.

nil when no ring has been published yet, and also when the topology could not be read: the reshard coordinator, its only caller, refuses to split without a ring either way, so the two collapse here. current_vnodes/0 keeps them apart, because its callers act on the difference.

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

The {vnode_id, token} pairs a rebalancing plan is computed over, from the live ring.

From the live ring rather than the boot list: a vnode born from a split would otherwise never enter any plan, so it would never be rebalanced nor have a lost replica replenished until the node restarted. A ring that cannot be read yields no pairs at all, and so an empty plan, which is the safe answer: nothing moves while this node cannot see the ring.

@spec current_vnodes() :: {:ok, [{atom(), non_neg_integer(), [node()]}]} | :error

The vnode placement the cluster has right now, from the live gossiped ring: {:ok, [{vnode_id, token, nodes}]}, or :error when the ring could not be read.

The live counterpart of boot_vnodes/1, and the seam every consumer that must survive a ring change reads instead of closing over the boot list. {:ok, []} (an unsharded or not-yet-published ring) is distinct from :error, so a caller can tell "there is nothing to do" from "I do not know", and the second one must never be acted on as though it were the first.

The nodes in each entry are the ring's recorded placement, which is a routing decision and not live ra membership: a rebalance moves members without republishing the ring, so a consumer that needs to know whether this node hosts a vnode asks Malachi.Cluster.RaCluster.local_member?/1, not this list. What the ring is authoritative about is which vnodes exist and where they sit.

@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 durable_orphan_authority(Malachi.Cluster.RingServer.server_id()) :: ([
                                                                       String.t()
                                                                     ] ->
                                                                       term())

The sharded control plane's authority for the orphan sweep: segments routed to their owners by the topology of record in the durable ring store at ring_server_id, read linearizably before and after the owners are asked (Malachi.Cluster.ReplicatedDSRSM.known_segments_stable/3).

The store, and not this node's gossiped copy of the topology, because a split writes its pending intent to the store BEFORE it moves any topic and only then gossips it (Malachi.Cluster.VnodeSplit, Malachi.Cluster.TopologyPublisher). A node whose gossip has not caught up would otherwise ask only the old owner of a topic that has already left it, twice, and both reads would agree. Gossip remains what routing uses; the destructive decision reads the record.

@spec enable_cluster_flag(
  String.t(),
  keyword()
) :: :ok | {:error, term()}

Switches the cluster feature flag named name on, from this running node. The entry point mix malachi.flag reaches over RPC.

Refuses, naming them, unless every configured node is alive and advertises the capability the flag names. That population is MALACHI_LOG_NODES, not the membership's alive set and not the flag store's Raft members: the first silently omits a node that has just joined on an old build, and the second omits every node not yet upgraded, since a build without the flag machine never joins that group at all.

:nodes, :reads and :known override that population, the membership view and the flag registry; they exist so the wiring can be driven from a test while this build's own registry is still empty.

@spec ensure_data_dir_identity(atom(), [node()], Path.t(), keyword()) :: term()

The startup gate over the log data directory against the control plane cluster over nodes (Malachi.Storage.DataDirGuard.check/4, whose options opts are): refuses the start when the directory holds segments a control plane formed now would not know, or when a one-member cluster is configured with peers, and logs and goes on when the operator adopted the directory.

@spec ensure_data_format(Path.t(), (non_neg_integer() -> term())) :: term()

The startup gate over the data-directory format marker (Malachi.Storage.FormatMarker): answers :ok when this binary can read dir, writing a marker when there is none, and otherwise refuses the start through halt_fun (default System.halt/1) with exit status 78 and one log line.

Checked at the root of the log data directory, so one marker covers every shard_<n> subdirectory (Malachi.DataPlaneRouter.shards/1).

@spec ensure_ring_identity(atom(), Path.t(), boolean(), keyword()) :: term()

The startup gate's first step, before the boot touches the ring store (Malachi.Storage.DataDirGuard.check_ring/4, whose doc holds the rules it applies).

@spec ensure_seed_identity(atom(), Path.t(), keyword()) :: term()

The startup gate's step before a sharded seed is written (Malachi.Storage.DataDirGuard.check_seed/3): refuses the start over an unsharded control plane that ran here or over segments the vnodes formed now could not know, and logs and goes on when the operator adopted those segments.

@spec incarnation_opts(Path.t()) :: keyword()

The incarnation options this node's membership server starts with: a reservation function over dir, and the callback that extends the ceiling.

A function rather than reserved values, because a child spec is built once and reused on every supervisor restart. Reserving while the spec was built would seed a restarted membership server at the number the first one started from, below whatever it had raised itself to before it crashed, which is exactly the permanent staleness the reservation exists to prevent.

Raises when no reservation can be made, which stops the node. See the comment above for why starting anyway is not a degraded start but a silently wrong one.

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

The vnode ids this_node both hosts and currently leads, given vnodes, the vnodes the ring names ([{vnode_id, token, nodes}]), and two predicates over a local server id: hosts? (defaults to RaCluster.local_member?/1) and leader? (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).

vnodes supplies which vnodes exist, not which ones this node holds. The entries' nodes are the ring's recorded placement and are deliberately not consulted: they are a routing decision that a rebalance moves members out from under, so hosting is asked of ra instead. hosts? runs first because it is a local read, while leader? is a call that a vnode mid-election makes wait; that order keeps a vnode this node does not host from costing anything at all.

Pure given the two predicates; deterministic given the current membership and 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 local_vnode_servers(
  [{atom(), non_neg_integer(), [node()]}],
  node(),
  (Malachi.Cluster.MetadataServer.server_id() -> boolean())
) :: [{module(), {atom(), node()}}]

The metadata vnode members this_node hosts, as {machine, server_id} pairs for Malachi.Cluster.MachineVersion.check/3: every vnode of vnodes this node holds an ra member of, led or not, since a follower can stop applying its log just as a leader can.

Hosting comes from hosts? (defaults to RaCluster.local_member?/1), not from the ring's recorded placement, for the same reason as leading_vnodes/4: a rebalance changes membership without changing the ring. Pure given hosts?.

@spec measurement_children([node()], Path.t()) :: [Supervisor.child_spec()]

The children of the data-plane sharding measurement mode (MALACHI_DATA_SHARDS > 1 with no control plane cluster): N independent in-memory brokers over dir, each with its own name and isolated subdirectory, and a warning that nothing they hold survives a restart.

No orphan sweep: a shard's only authority is its own memory, which a restart empties, and a sweep that trusted it deleted every segment written before the boot (#273). The scrubber follows each broker: it comes after it in the list, so the broker is alive when the scrubber asks for its replication server. Public so a test can hold the mode to that.

@spec membership_attributes() :: map()

This node's gossiped membership attributes: the operator's MALACHI_LOG_ATTRIBUTES with the capability set this build advertises merged in.

The seed the membership server starts with. Malachi.Cluster.Membership.set_attributes/2 replaces the whole map, so attributes set without the capabilities would erase them and every cluster flag would then be refused with no obvious cause; on the runtime path Malachi.Cluster.MembershipServer.set_attributes/2 merges them back itself, and this is the boot path.

@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_reads() :: Malachi.Cluster.Capabilities.reads()

The membership view as node -> {status, attributes}, for the capability check.

An unclustered deployment runs no membership server, so it answers for the local node only: any other node reads as unknown, which refuses, which is the safe direction.

@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 metadata_version_watcher_children(
  atom() | nil,
  [{atom(), non_neg_integer(), [node()]}] | nil
) ::
  [Supervisor.child_spec()]

The machine-version watcher for an unsharded control plane, or none when it is sharded.

The single metadata group is started by Malachi.BrokerServer, not by a reconciler of this application's own, so without this child nothing would call Malachi.Cluster.MachineVersion.check/3 for it. A sharded control plane needs none: the vnode coordinator manager watches every local vnode member (local_vnode_servers/2).

@spec orphan_authority(atom(), [node()], term()) :: ([String.t()] -> term())

What Malachi.Retention.OrphanSweeper asks before it removes anything: which directory names the control plane accounts for, asked of the vnodes that OWN those segments and never of this node's cached copy of the metadata (#249).

• sharded (vnodes given): durable_orphan_authority/1 over this cluster's durable ring store;
• a single ra cluster (cluster without vnodes): that cluster, as the one owner of every topic. A single node is one, with one member.

The data-plane sharding measurement mode has no control plane to ask, and runs no sweep.

@spec orphan_sweeper_children(atom(), [node()], term()) :: [Supervisor.child_spec()]

The node's orphan sweep, beside the scrub and for the same reason: the data directory is a fact about this node, so it is not leader gated. One sweeper over the node's named replication server, whose authority is orphan_authority/3: the vnodes that own the segments, never this node's cached copy of the metadata. A single node is a one-member cluster and runs exactly this; the data-plane sharding measurement mode runs no sweep.

Public so a test can hold the wiring to that, since a sweeper handed a cache-backed authority here would reopen #249 with every other test still green.

@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 resume_local_vnodes([{atom(), non_neg_integer(), [node()]}], node(), keyword()) ::
  [
    {atom(), :ok | {:error, term()}}
  ]

Restarts this node's stopped member of every metadata vnode the group still counts it in, and returns what happened to each one it tried: {vnode_id, :ok} for a member brought back, {vnode_id, {:error, reason}} for one that could not be.

ra does not restart a node's registered servers when the node comes back. Every other store has its own reconciler doing it for the node's own member on every tick; the metadata vnodes had none, so a restarted node never rejoined any vnode (the leader-only bootstrap in Malachi.BrokerServer skips a vnode whose other members still answer, and resumes only its own member). A rolling restart then took the vnodes below quorum at its second node, and a full restart at once (#136). Restoring a replica is the duty of the node that hosts it, as in NorthGuard, where a vnode stays in place while its Raft group's replicas come and go; which node leads it, and so runs its coordinators, follows from there.

The rule, per vnode of vnodes:

• a member that is running is left alone, and nothing is asked;
• a vnode this node never registered a member of is left alone too, without a remote call, which is most of the ring on a large cluster;
• otherwise the group is asked whether it still counts this node, through every other node of the cluster and of the ring's placement. Not the placement alone: it changes only on a split, never on a rebalance, so a vnode moved to other nodes still lists the old ones. A rebalance deletes the member it removes once the leave commits (Malachi.Cluster.Rebalance.ra_remove_member/4), but a member whose leave outlasted the call is only stopped, and one removed before rebalances deleted, or whose delete could not reach its node, is still registered too: for each of them the group as it is now says it no longer counts it;
• when no node answers for the group, the member is resumed anyway: that is a group whose every member is down, and resuming each node's own member is the only way it regains a quorum;
• resuming only ever restarts a registered member (RaCluster.resume/1), never forms one.

The asks run on the vnode coordinator manager's loop, before it reads leadership, so they run at once rather than one after another: a group with no leader holds each ask for its whole timeout, and asked in turn, every stopped member times every node would hold the loop that many seconds. At once, a pass costs one bounded ask while the stopped members times the nodes stay under 256, and one more per 256 above.

Seams, all defaulting to live ra and this node's configuration: :peers (the nodes to ask through, default the configured cluster), :running? and :registered? (server_id -> boolean), :members (server_id -> {:ok, [node]} | :error, bounded well below a reconcile tick) and :resume (server_id -> :ok | :not_hosted | {:error, reason}).

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

The configured retention policy: :max_age_ms / :max_bytes (nil = that rule is off) and :segment_max_age_ms, how old an active segment may get before the sweep rolls it (7 days unless set).

@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 store_reconciler_child(atom(), module(), atom(), [node()], (-> any())) ::
  Supervisor.child_spec()

The reconciler child of a replicated store, which is also its machine-version watcher.

On a clustered node it self-joins this node through reconcile on every tick (see Malachi.Cluster.LeaseReconciler). On a single-node deployment there is nothing to join, so the child stays with a no-op reconcile and only watches: a one-member group still reaches a machine version, and a member that restarts on code below it stops applying its log exactly as a clustered one does (Malachi.Cluster.MachineVersion). Losing the watch there would leave the only member of the control plane silent.

@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_coordinator_manager_opts() :: keyword()

The options the vnode coordinator manager runs with in production.

Split out of its child spec so a test can drive the real seams rather than stand-ins: the bug this wiring exists to prevent lives in the seams, not in the manager, so a test that supplies its own cannot see it.

The placement is resolved on every reconcile (current_vnodes/0), never captured here. Capturing it is what left a vnode gained through a rebalance or a split without coordinators and without a machine version watch until the node restarted.

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

@spec warn_group_commit_needs_rf1(boolean(), pos_integer()) :: :warned | :ok

Warns, once at boot, when broker group commit is asked for (MALACHI_GROUP_COMMIT=true) with a configured replication factor above 1, and answers :warned; answers :ok otherwise.

Broker group commit appends to a segment's primary alone, so it is only correct when every segment has one replica, which only a configured factor of 1 guarantees whatever the broker set does later. A single node runs a one-member cluster with the default factor of 3, so without this line the knob would just stop working: the fix is MALACHI_LOG_REPLICATION_FACTOR=1, or MALACHI_REPLICATION_GROUP_COMMIT=true, which batches the fsync on every replica as NorthGuard does.