Architecture

Malachi is a CP (consistent, partition-tolerant), horizontally scalable log broker, written entirely in Elixir on the BEAM. It follows LinkedIn's NorthGuard log-storage design: clients speak topics, keys, and opaque cursors, never partitions or offsets, so the broker can split, merge, and restripe its storage underneath without breaking them. The control plane (metadata) is replicated by quorum through Raft (the ra library); the data plane (records) uses its own quorum replication; cluster membership uses SWIM.

This document describes how the system is built and the reasons behind the load-bearing choices. For how to use it, see the guides; for the auth design, see the auth ADR.

The log model

Topic   ── a named collection of Ranges covering the whole keyspace
 └ Range  ── a log for a contiguous band of keys (active | sealed)
    └ Segment ── the unit of replication: a sequence of records (seals at 1 GB / 1 h / on failure)
       └ Record ── key + value + headers (bytes), at a logical offset within the segment

A record's key hashes to a position in the topic's keyspace, and that position falls in exactly one range. Records with the same key land in the same range and are ordered relative to each other.

Range split and merge are purely logical metadata operations. Segments are never physically combined or copied; a merge happens only between buddy ranges (buddy-allocator style). Total ordering is preserved through happens-before on splits and merges. Because a range can split or migrate at any time, the client never sees an offset: its position is an opaque cursor it carries and passes back, and the server is free to reshape ranges without invalidating it.

This is the departure from partition-and-offset systems like Kafka, where the client-visible partition is hash(key) mod partitions, so raising the partition count relocates a key and breaks its per-key ordering. Here same-key ordering survives resharding. The key hashes to a fixed range whose single primary serializes appends; a split applies a seal-first fence in two layers, the parent range sealed in the metadata (atomic, inside the split command) and the parent's write head closed in the data plane before either child accepts a write, so a node that has not yet seen the split is refused rather than writing into a range that is already history; and a read of a child chains cross-epoch, draining the parent's sealed slice of the keyspace oldest-first before the child's own records. So a key stays in one child and its history reads in order straight through the split, with no partition count for a client to have frozen into its assumptions.

flowchart LR
  P["parent range (active)"] --> S["seal-first: parent sealed before children write"]
  S --> C["child range (new writes)"]
  C --> R["cross-epoch read: parent's sealed slice oldest-first, then the child"]

Storage layer

Storage is a pluggable behaviour, Malachi.Storage.SegmentStore, so the on-disk format is decoupled from any one implementation. A store opens or recovers a segment, appends batches of records, syncs (fsync) before the write is acked, reads a range of records, seals a segment to make it immutable, and decides when to seal (by size, age, or on failure). See the module for the full callback set.

The shipped implementation is pure Elixir: :file in [:raw, :binary] mode, batching by roughly 10 ms / N records / N bytes, an fsync before every ack, and a sparse index. A native store (a Rust NIF with O_DIRECT, aligned buffers, and an application-level cache) is a possible future optimization; it is justified by the latency tail and page-cache behaviour under many concurrent segments, not by throughput. On the storage critical path the pure BEAM sustains hundreds of MB/s with durable fsync, well above the tens of MB/s per broker the workload calls for, so throughput is not the bottleneck. Reproduce the measurement with mix run benchmark/storage_viability.exs.

flowchart LR
  A["active segment (taking appends)"] -->|"full: by size, age, or failure"| S["sealed segment (immutable)"]
  A -->|"a new one rolls"| A2["new active segment"]

Every record frame carries a CRC32 of its payload, which the store checks whenever it decodes a record. That is only a passive guarantee: a sealed segment is immutable and nothing re-reads it, so corruption at rest stays invisible until a consumer happens to need that exact record, and then it does not surface as an error but as a short read (the read bound comes from the control plane's record count, so the damaged copy simply stops producing records). Malachi.Cluster.Scrubber closes that: on every node, on a slow cadence, it re-verifies the checksums of the sealed segments it stores (Malachi.Log.verify/2, a read-only scan that never repairs or truncates) and rebuilds a damaged copy from a replica that still verifies, moving itself out of the read path first when it is the primary. A copy that will not decode is a lost replica, so the same re-replication clock applies to it. The one rule the repair never breaks: the local copy is not deleted until a peer has confirmed a copy that is both intact and complete against what the control plane recorded at seal time, because a partially readable copy beats no copy, and beats a shorter one. A rebuilt copy is written back record by record, so it comes back without a seal marker, as any replica copy whose own log never rolled does. The repair then seals it again at the length the control plane recorded, so the copy is fenced like the copies it was rebuilt from; if that seal cannot be made, the copy is latched as failed and healing replaces it. A failed catch-up or verify after the delete latches it the same way. If the latch itself does not answer in time, the copy is removed instead. If that removal does not answer either, the copy stays in place, unfenced, until a later scrub pass retries the latch, and a restart loses that retry. The scrub does not copy a missing replica back: the heal pass does, once every replica of the segment is live. See the operations guide for the knobs and cadence.

A copy can also hold more bytes than its seal allows, for example a replica that kept a preallocated tail or a write past the seal. The settling pass brings such a copy down to the length the control plane recorded and fences it there. A copy that holds fewer records than its seal names is handled differently: the pass fences it at the length it holds, which the control plane never recorded, and reports it as not settled. The integrity pass fills in place a copy whose bytes are below the recorded size, when every replica of the segment is live, the segment is not under-replicated, and a copy at or above the recorded size exists to copy from, and no replica's latch state is unread this tick, the copy is not latched as failed (a latched copy is replaced instead, or, with no spare broker, kept behind the healthy replicas), and its durable end is short of the recorded end. A copy whose records are all present but whose bytes are short is reported as integrity suspect and left to the scrubber, which revisits each sealed copy about weekly on a node with 10k sealed segments (see the operations guide). This fence at a length the metadata does not carry is a divergence from NorthGuard, where the seal the control plane records is the only one a copy carries. A copy whose bytes stay above the recorded size, holding exactly its seal's records, is returned in the probe's unsettled list until it is confirmed, on every pass in which its segment is fully live, not under-replicated and the copy is not latched, unless its segment's latch state is unreadable or its own probe times out on that tick, because a sealed size is immutable, so no pass corrects it; the confirmation records that the copy is settled and it is not selected again.

Analogy. A segment is a notebook: you keep writing on the current page until it is full, then you close that notebook for good (seal it, never edited again) and open a fresh one. Because closed notebooks never change, copying them to another node is safe.

Metadata: DS-RSM over ra

Metadata (topics, ranges, segments) lives in a directory of sharded replicated state machines:

• A vnode is a Raft group (an ra cluster) holding one shard of the metadata.
• A coordinator is the vnode's leader; it carries the business logic: sealing or deleting a topic, splitting or merging a range, choosing segment replica sets, and healing under-replicated segments.
• vnodes sit on a hash ring (consistent hashing) keyed by topic name. NorthGuard also places a range by the hash of the range itself; Malachi keeps a topic's ranges and segments on the topic's vnode for now (Malachi.Cluster.DSRSM, and Malachi.Metadata.segment_routing_topic/1, the one function that decides a segment's owner). A vnode's position on the ring is stable even as its Raft replicas join and leave, and a vnode can split, breaking its state into two Raft groups.

flowchart TB
  subgraph ring["Hash ring (metadata sharded by topic)"]
    V1["vnode A (a Raft group)"]
    V2["vnode B (a Raft group)"]
    V3["vnode C (a Raft group)"]
  end
  V1 --> L1["coordinator = leader, does the work"]
  V2 --> L2["coordinator = leader"]
  V3 --> L3["coordinator = leader"]

Analogy. Each vnode is a small committee that votes on its own slice of the metadata. The chair (the leader, called the coordinator) carries out what the committee agrees on. Sharding means many small committees instead of one giant meeting that everything has to wait on.

Because that position is stable while the replicas move, a node decides what to coordinate from two sources, and never from one: the gossiped ring says which vnodes exist, and ra says which of them this node holds a member of and leads. Malachi.Cluster.VnodeCoordinatorManager re-reads both every few seconds, so a vnode this node gains after boot, whether a rebalance added it as a member or a split created it, starts coordinating without the node restarting. A ring it cannot read leaves the coordinators it already runs alone, because stopping them all is worse than waiting for the next read.

Known divergence: a vnode accepts metadata outside its arc

In NorthGuard the vnode that owns a region of the ring is the one responsible for its metadata. Malachi does not yet enforce that on write: a vnode applies whatever command reaches it, without checking the command's topic against its own arc. A broker whose gossiped ring lags behind a vnode split can therefore create a topic, and register its segments, on the vnode that owned the arc before the split. Nothing moves that metadata afterwards, so a reader routing by the current ring does not find it.

The destructive consumer is guarded against it rather than the cause being fixed. Before the orphan sweep (Malachi.Retention.OrphanSweeper) removes a directory whose segment the owner does not list, it asks every vnode on the ring (Malachi.Cluster.ReplicatedDSRSM.known_segments/3); a segment another vnode lists is held, never removed. That read can only prevent a removal, never cause one, and it costs liveness: while any vnode is silent, a pass with a real orphan to decide is skipped. The fix in NorthGuard's shape is for the owning vnode to refuse a command whose topic its arc does not cover; until that exists, this read is what keeps a misrouted topic's acknowledged data on disk.

The metadata state machine is Malachi.Cluster.MetadataMachine (@behaviour :ra_machine). It is a pure function of its input: it never reads the wall clock, configuration, or node(). Anything time- or config-dependent travels inside the command, and the machine reads the meta.system_time its server feeds it. This keeps every replica deterministic, which is what Raft requires.

Membership: SWIM

Broker membership uses SWIM: random probing for failure detection plus infection-style dissemination. It spreads only minimal global state: each broker's host, port, and attributes, plus vnode boundaries, leader, and term for routing. Everything larger stays in the per-vnode Raft groups.

Replication: two planes

Replication is split deliberately, because metadata and records have opposite shapes:

• The control plane (metadata) runs over ra. Metadata is small, changes rarely, and needs linearizability, which is exactly Raft's strength.
• The data plane (records in segments) does not go through ra. Malachi.Cluster.ReplicationServer ships each batch from the primary to the followers as pipelined, windowed replica-appends and acknowledges the write only once a quorum has fsynced it, tolerating up to ⌊(N-1)/2⌋ slow or unreachable followers and returning {:error, :no_quorum} beyond that. Routing high-volume sequential records through a consensus log would pay for a second durable write and gain nothing, because the segment already is the log. Fsyncs can be coalesced (group commit, the NorthGuard triggers of 10ms/20k records/10MB) at two independent levels: broker-level for rf=1 (MALACHI_GROUP_COMMIT) and replication-level for rf>1 (MALACHI_REPLICATION_GROUP_COMMIT, default off because it only pays when many producers share hot ranges on fsync-bound disks; see the clustering guide for the decision rule with examples).

flowchart LR
  subgraph control["Control plane: metadata (over ra)"]
    M["small, changes rarely, everyone must agree (linearizable)"]
  end
  subgraph data["Data plane: records (own quorum)"]
    D["huge, constant, ack once a majority has fsynced it"]
  end

Analogy. The control plane is the town-hall minutes: they change rarely and every council member has to agree before a line is written. The data plane is the warehouse: shipments arrive constantly, and one only needs a majority of the shelves to confirm they stored it before signing off. Different jobs, so different rules.

Client protocol

The wire protocol is length-framed binary, owned by Malachi.Wire: <<len::32, body>> carrying <<api_key::16, correlation_id::32, payload>>. The correlation_id matches each response to its request, so a connection pipelines without a session layer. The protocol covers the log, consumer groups, authentication (password, mTLS, token), and per-topic ACLs across its api_keys.

flowchart LR
  L["len (32 bits): how long the frame is"] --> K["api_key (16): which operation"]
  K --> C["correlation_id (32): the request's number"]
  C --> P["payload: the operation's arguments"]

Analogy. The correlation_id is the number you take at a bakery counter. You can place several orders without waiting, and each tray comes back tagged with your number, so replies match requests even when they arrive out of order.

• Metadata operations are unary (one request, one response): create, delete, topic metadata, segment metadata. Any broker can act as a proxy and route to the vnode leader using the gossiped state.
• Data operations are streaming (produce, consume, replication) with pipelining and windowing for flow control. A zero-copy consume via :file.sendfile is a future optimization, not implemented: today the wire ships records in a compact offset-less encoding distinct from the on-disk frame (see Malachi.Wire), so consumes read, decode, and re-encode through the BEAM, and sendfile would first require aligning the fetch encoding with the on-disk frame. Every published throughput number reflects the current path; produce is unaffected, read-side numbers have headroom.

Placement and policies

A storage policy is a name plus a retention rule plus placement constraints. A constraint is an expression over attributes: opaque key/value pairs that operators attach to brokers. This generalizes rack and data-center awareness without the core needing to understand what a "rack" is; the same mechanism decides segment replica sets and vnode replicas.

The definitions are an administrative object of the cluster, replicated in their own Raft group beside the user, lockout and ACL stores, because an administrator defines them once for the cluster rather than for a vnode. What a topic keeps is the name it points at, in its own metadata, so the binding travels with the topic when a vnode split moves it while the definition stays put. A topic pointing at a name no definition backs is not the same as a topic with no policy: the second uses the cluster defaults, while the first never expires under them at all. Falling back would delete, on every replica and with no way back, exactly the data a more permissive policy was written to keep. So an unresolved name holds disk until an operator fixes the binding, and by default it holds all of it: the cluster limits do not apply and nothing else replaces them. A setting of its own adds an optional age backstop, left unset on purpose because that bound is a retention decision, and a metric of its own names the topic on every sweep. The operations guide covers both.

Placement is deterministic (raft-safe: every replica computes the same result, with no randomized tie-break). It honors a maximum skew across domains, a minimum number of distinct domains, and a hard versus soft distinction for unsatisfiable constraints. Healing prefers surviving replicas, which keeps churn low.

Analogy. Placement is seating guests at a wedding: spread each family across different tables (fault domains) so one mishap does not take them all out, and keep the tables evenly filled (max skew) rather than crowding one. The seating rule is fixed, so every planner drawing the chart lands on the same seats.

Key design decisions

Four choices shape everything above. Each is stated with its reason and the code that implements it.

1. Replicate metadata over ra, but records over our own quorum. Two planes, as described above. Metadata (Malachi.Cluster.MetadataMachine) needs linearizability and changes rarely; records (Malachi.Cluster.ReplicationServer) are high volume and sequential, where a second consensus write would cost more than it is worth.

2. A sidecar .idx file for the sparse index, not persisted ETS or DETS. One index per segment (Malachi.Log.Segment.index_path/1 returns <id>.idx), loaded into memory as an array sorted by offset for an O(log n) floor lookup, with one entry every few kilobytes. ETS and DETS were both rejected because either would couple the on-disk format to BEAM structures; the format must be one a native implementation can reopen without speaking BEAM.

   Analogy. The on-disk format is written in plain handwriting any tool can read, not in a personal shorthand only the BEAM understands, so a future native (Rust) store can reopen the same files.

   The sidecar carries no checksum of its own, and it does not need one, because nothing is allowed to depend on it being right: a read that starts where the index points and does not find what it asked for simply rescans the segment from the beginning, which is exactly what an absent index does. The scrub verifies the sidecar as well (every entry must land on a real frame carrying the offset it claims) and repairs it locally, by rebuilding it from the segment. That is the difference between derived data and the records themselves: a damaged segment has to be refetched from a replica, while a damaged index never left home.

3. Keep and extend the binary protocol rather than sessionize it. The correlation_id already provides the pipelining a session layer would have added, so the protocol stays <<len::32, body>> and grows by adding api_keys. See Malachi.Wire.

4. An opaque cursor from the start, never a plain integer offset. The client never sees an offset; the position travels in the cursor (Malachi.LogApi, @type cursor :: String.t()). That is what lets the broker reshard and split ranges without breaking clients. Because the cursor returns from an untrusted client, decode_cursor/1 deserializes with binary_to_term(_, [:safe]) and validates the shape, so a forged cursor cannot mint a new atom or an arbitrary term.

What we do not replicate

NorthGuard leans on deterministic simulation (a single-threaded cluster and clients with swappable time, network, disk, and RNG, replaying failures exactly) as a reliability pillar. That is essentially unfeasible on the BEAM, whose scheduler is preemptive and multicore and outside our control. This is a real downgrade in guarantees, and it is accepted explicitly. The substitutes, and their delivery status:

• Property-based stateful testing of the log model and the state machine (stream_data): delivered.
• Fault injection with a Jepsen-style acked-durability check: delivered as the chaos certification harness (scripts/docker-chaos-test.sh + scripts/chaos_checker.exs). A 3-node RF=3 cluster takes synthetic traffic while a node is SIGKILLed, partitioned, SIGSTOPped, and rolling-restarted; the harness certifies that no acknowledged write is ever lost, that the cluster reconverges after every event, and that availability recovers. See the operations guide.
• Storage fault injection (the "different types of corruption" scenarios): delivered as scripts/docker-storage-chaos.sh. Follower segment copies are corrupted mid-file, truncated, and deleted outright; the harness certifies the same invariants plus physical reconvergence (every copy holds the same records, byte for byte over its valid bytes), exercising CRC-clamped recovery, write-path catch-up, and the sealed-copy integrity probe in Malachi.Cluster.SelfHealing. In-place corruption that keeps the byte size is now caught by the integrity scrub (see the storage layer above); the harness gains that event with the scrub's certification. A storage failure on any copy, primary or follower (a full volume, a failing device), is sealed on failure the way NorthGuard does it: the node takes that copy out of service, Malachi.Cluster.Failover seals the segment on the intact copies once enough answer to cover every acknowledged write (at two replicas, the one left), and producers move to a new segment. Once sealed, the failed copy is a lost replica on a live node: Malachi.Cluster.SelfHealing backfills a replacement on another broker and the healing pass then deletes the broken copy, while the integrity scrub leaves it alone rather than refetching it onto the disk that failed. The drill's filled-volume event certifies that the node survives it. One deliberate limit remains, tracked as a roadmap item: damage found by reading a primary copy (corruption rather than a failed operation) does not take the copy out of service, so it still needs seal-on-failure of its own, and the corruption events always damage followers.
• Config-deployment certification (the "deployments... and even config deployments" scenarios): delivered as scripts/docker-config-chaos.sh. A harmless config change is rolled node by node under traffic (availability must hold between steps and the new value must take effect), and a fail-fast bad config is pushed to one node and rolled back (the node must crash-loop without ever going healthy, the surviving two must keep serving quorum writes, and the rollback must restore 3/3), on top of the acked-durability and reconvergence invariants.
• Rolling-upgrade certification (the "deploy the build, roll back the build" operations): delivered as scripts/docker-upgrade-chaos.sh. The cluster is rolled node by node from the release before this code to this tree and back under traffic, then forward again with a canary feature switched on and a rollback that must be refused. The canary (a capability, a metadata command at the next machine version, an unknown replication message, a new data format) is a patch applied only to the drill's image, so the gates of the bridge release are certified before any real feature depends on them.
• Concuerror for systematic interleaving exploration: attempted and blocked by the tool, with the spike kept as the reproducible record (scripts/concuerror-setup.sh, scripts/concuerror.sh, test/concuerror/replicate_race.ex). What the spike established, in order: Concuerror builds and runs on our OTP (its own CI matrix stops at OTP 23, so this was not a given); it drives Elixir and OTP code, including a GenServer with the exact park, ack and Process.send_after/cancel_timer race we wanted to check; but it cannot run the scenario that matters, because Malachi.Cluster.ReplicationServer is disk-backed by design (its init creates the data directory, log recovery scans it) and Concuerror's file_server_2 emulation does not implement read_file_info. --non_racing_system file_server_2 does not help: the request type is simply unimplemented. Unblocking it needs either a Concuerror release that supports the request, or a filesystem-free seam through the storage layer (the SegmentStore behaviour is injectable, but directory creation and segment discovery still go straight to the filesystem), which is a production refactor driven purely by a verification tool and not worth it today. The invariant the scenario targets, that a parked batch is answered exactly once, is covered by unit tests for each path separately (ack cancels the timer; a fired timer finds the batch gone), just not by systematic interleaving.

Prior art

Two mature systems informed the distribution design, as references rather than dependencies. Both converge on the same pattern for coordinating shards: a single elected coordinator, fenced through consensus.

• riak_core (Apache 2.0) contributes its ring management: the staged → planned → committed model, where the plan computes a new ring without changing state and the commit validates that nothing diverged, plus placement that guarantees replicas on distinct nodes and distinct locations, uniform balancing, and rebalancing with minimal movement. riak_core is AP (gossip plus vector clocks); Malachi is CP, so it keeps the algorithms and swaps gossip for Raft. Its ring gossip, vector clocks, and preflist-over-vnodes are deliberately not adopted: Malachi shards metadata by topic rather than distributing data keys over the ring.

• Kubernetes contributes its leader election and reconcile patterns: a lease with the duration > renew-deadline > retry-period triangle over a linearizable CAS, proactively giving up leadership when a renewal fails (to avoid split brain), and level-triggered, idempotent controllers where only the leader acts. In Malachi the lease lives in an ra cluster; ra's fencing-by-name already covers the single self-fencing bootstrap, and the lease is what fences the leader's continuous work (retention, healing, rebalancing). A single etcd is Kubernetes's scale ceiling; Malachi shards precisely to scale past one quorum.

Dependencies

In use

The libraries the broker actually depends on (see mix.exs for the pinned versions):

Candidate libraries (not dependencies today)

These are not in mix.exs and nothing uses them. They are the libraries a future increment would reach for, recorded here so the design intent is not lost. Membership today is a home-grown SWIM (Malachi.Cluster.Membership), and the segment store is pure Elixir.