Malachi.Cluster.ReplicatedDSRSM (malachi v0.8.13)

The DS-RSM backed by real Raft: a Malachi.Cluster.HashRing plus one ra cluster per vnode (each running Malachi.Cluster.MetadataMachine). Commands and queries are routed by consistent hashing (topic name) to the owning vnode and submitted to that vnode's Raft cluster, so the cluster's metadata is sharded across vnodes and durably replicated within each one. Leadership of a vnode's cluster is that vnode's coordinator.

This is the production counterpart of the pure Malachi.Cluster.DSRSM (which holds the per-vnode Metadata in memory and is what the property tests exercise). Here each vnode's Metadata lives in a Raft log instead.

The value threaded through calls holds only the ring and a vnode_id => server_id map (both immutable); the metadata itself lives in the ra processes, so command/3/query/3 do not change it: only add_vnode/3 does (and starts the vnode's cluster as a side effect). ra must already be running (e.g. :ra.start_in/1), as with Malachi.Cluster.MetadataServer.

Each vnode's cluster can span several nodes (add_vnode/4), so a vnode survives losing a member: HA per vnode. split_vnode/4 grows the ring at runtime, migrating the displaced topics' metadata between the source and new Raft groups (the dynamically-sharded part); fencing concurrent writes to a migrating topic (zero-window cutover) is a later step.

Summary

Types

t()

vnode_id()

Functions

abort_split(state, new_vnode_id, new_server_id)

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.

add_vnode(state, vnode_id, token, nodes \\ [node()])

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.

command(state, topic_name, command)

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.

complete_split(state, new_vnode_id, token, nodes \\ [node()])

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

delete(state)

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

new(opts \\ [])

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

query(state, topic_name, query_fun)

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.

route_vnode(state, vnode_id, token, server_id)

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.

server_for(state, vnode_id)

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

snapshot(state)

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

split_vnode(state, new_vnode_id, token, nodes \\ [node()])

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.

vnode_for(state, topic_name)

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

vnode_ids(state)

The ids of the vnodes.

Types

t()

@type t() :: %Malachi.Cluster.ReplicatedDSRSM{
  ring: Malachi.Cluster.HashRing.t(),
  vnodes: %{required(vnode_id()) => Malachi.Cluster.MetadataServer.server_id()}
}

vnode_id()

@type vnode_id() :: atom()

Functions

abort_split(state, new_vnode_id, new_server_id)

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

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.

add_vnode(state, vnode_id, token, nodes \\ [node()])

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

command(state, topic_name, command)

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

complete_split(state, new_vnode_id, token, nodes \\ [node()])

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

delete(state)

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

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

new(opts \\ [])

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

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

query(state, topic_name, query_fun)

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

route_vnode(state, vnode_id, token, server_id)

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

server_for(state, vnode_id)

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

snapshot(state)

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

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

split_vnode(state, new_vnode_id, token, nodes \\ [node()])

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

vnode_for(state, topic_name)

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

vnode_ids(state)

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

The ids of the vnodes.