@spec append(term(), term(), [term()], non_neg_integer(), [Malachi.Log.Record.t()]) ::
  {:ok, non_neg_integer()}
  | {:error, :not_primary | :empty | :empty_replica_set | term()}

Group-commit append: buffers records for segment_id on the primary WITHOUT fsyncing, returning {:ok, last_offset} as soon as they are in the buffer. Durability comes from a later flush/1, which coalesces the fsyncs of many appends into one. Single-broker (rf=1) only: it does no follower fan-out, so the caller must use replicate/5 when the replica set has followers. Same offset and return contract as replicate/5, so the two are interchangeable as the broker's write function.

Returns a specification to start this module under a supervisor.

See Supervisor.

@spec confirm_settled(term(), term(), timeout()) :: :ok | {:error, term()}

Records, durably, that this server's copy of segment_id is settled. The caller has already checked that the copy holds exactly the records its seal names and measures above the recorded byte_size. This server does not read the seal, so it writes the marker it is asked for and the check stays with the caller. The integrity probe then stops selecting the copy, so one whose only divergence is a size the metadata can never correct is settled once, not on every pass.

Answers :ok once the marker is on disk and its directory entry is synced. An error reply means the marker is not known to be durable. A failed write removes the marker only when this call created it, and leaves the copy selected again, which is the level-triggered retry. An earlier marker stays in place. A marker that survives a failed sync stays until the segment's directory is removed, and the probe keeps skipping its copy, which is safe because the copy holds exactly its seal's records.

@spec delete(term(), term()) :: :ok | {:error, :unreachable}

Deletes segment_id's stored data from this server (used by retention once the control plane has dropped the segment). Idempotent. Deleting an unknown or already-removed segment is :ok, and it also clears any on-disk files left after a restart when the log was not reopened.

A server that is down or on an unreachable node answers {:error, :unreachable} rather than exiting the caller, like read/4: retention's periodic sweep must survive a replica that is not there. It says so rather than answering :ok because the difference is a directory that stayed on that replica's disk with no control-plane metadata left to name it, which is what Malachi.Retention.OrphanSweeper exists to reclaim and what an operator watching the sweep needs to see. The caller cannot retry it either: once the control plane has dropped the segment, no later sweep returns that id.

@spec delete_directory(term(), String.t(), timeout()) :: :ok | {:error, term()}

Deletes one directory under this server's data directory by NAME, closing the log it holds if this server has it open.

What Malachi.Retention.OrphanSweeper calls, and the reason it exists: the mapping from a segment id to its directory is one-way (Malachi.Storage.Layout), so a directory the control plane no longer names cannot be turned back into a segment id to pass to delete/2. Routing the removal through this server anyway is what keeps one owner of the files: a directory that looks orphaned may still have an open handle here, and removing it behind this server's back would leave that log writing into an unlinked inode. The caller decides WHETHER a directory should go; this decides HOW.

name must be a single path segment. Anything with a separator or a parent reference is refused rather than joined, because a name that escapes the data directory is not something to act on even when it came from a listing of that directory.

Answers :ok, {:error, :invalid_name}, or {:error, :unreachable} when this server is not there.

@spec durable_end(term(), term(), non_neg_integer(), timeout()) ::
  non_neg_integer() | {:error, term()}

The durable end offset this server holds for segment_id, recovering the log from disk when it is not open yet. Unlike end_offset/3 (which answers :empty for a segment that exists on disk but has not been touched since this server booted), this gives the true resume point after a restart, which is what a repair needs as its copy start. base_offset seats a missing or empty log at the segment's base. A copy that failed here answers {:error, {:storage, reason}}.

@spec durable_stats(term(), term(), non_neg_integer(), timeout()) ::
  {:ok, non_neg_integer(), non_neg_integer()} | {:error, term()}

What this server durably holds for segment_id, as {end_offset, byte_size}, WITHOUT fencing it.

The measuring half of seal/4, kept separate because the two answer different questions and only one of them has consequences. A failover pass has to learn what its replicas hold before it knows whether enough answered to seal (Malachi.Cluster.Failover.seal_quorum/1), and fencing to find that out would close replicas of a segment the pass then declines to seal: a fence has no inverse, so those replicas keep refusing writes after their primary returns, and every later write needs its quorum without them. So the pass measures first with this, and fences with seal/4 only once it knows it is going to seal.

Flushes before answering, for the same reason seal/4 does: a log's next offset counts buffered records while the store serves only committed ones, so an unflushed answer describes records a read cannot return. base_offset seats a missing or empty log at the segment's base.

@spec end_offset(term(), term(), timeout()) :: non_neg_integer() | :empty

This server's next offset for segment_id, or :empty if it stores none of it yet. timeout bounds the call: pollers (the broker's periodic range-state refresh) pass a short one so an unreachable replica cannot block their loop for the default five seconds.

@spec fail_copy(term(), term(), term(), timeout()) :: :ok | {:error, :unreachable}

Latches segment_id's copy on this server as failed, for reason, exactly as a storage failure on the copy does: reads and writes are refused, failed_segments/3 reports it, and healing replaces the copy. The scrubber uses it when a copy it rebuilt cannot be fenced: a copy that is not sealed must not stay in service.

@spec failed_segments(term(), [term()], timeout()) ::
  {:ok, MapSet.t()} | {:error, term()}

Which of segment_ids have FAILED on this server (see "Storage failures"), as a MapSet.

Read-only and free of disk access: it is a lookup in the failure latch, so a heal pass can ask every replica about every active segment it holds on each pass. A failed copy is one this server will never write again, which is what Malachi.Cluster.Failover needs to know to seal the segment elsewhere.

A dead or unreachable server answers {:error, :unreachable} rather than exiting the caller.

@spec fenced_segments(term(), [{term(), non_neg_integer()}], timeout()) ::
  {:ok, %{optional(term()) => {non_neg_integer(), non_neg_integer()}}}
  | {:error, term()}

Which of segments this server has already FENCED, as %{segment_id => {end_offset, byte_size}}. Never fences anything: it reports a latch somebody else already closed.

segments is a list of {segment_id, base_offset} pairs, batched into one call on purpose. The caller (Malachi.Cluster.HealCoordinator) asks about EVERY active segment on every pass, which is a different shape of question from durable_stats/4's: that one is asked about a handful of failover candidates and pays an open plus a flush per segment, and paying that for every active segment of a cluster would put descriptors and I/O on a poll where there is none today.

A segment that is not fenced is simply absent from the answer and costs a map lookup, or a stat of the seal marker when this server has not opened it since booting (the same test replicate/5 itself applies). Only the fenced ones, which is the rare case this exists to find, pay an open, and that open fsyncs the segment's directory before it believes the marker (Malachi.Log.recover/2), so a fenced answer is never built on a marker that a power failure could still take away.

A dead or unreachable server answers {:error, :unreachable} rather than exiting the caller, as durable_stats/4 and seal/4 do: the caller is a coordinator loop that must survive a replica it cannot reach.

@spec flush(term()) :: :ok | {:error, [{term(), term()}]}

Fsyncs every segment on this server that has buffered (un-synced) records, making all prior append/5s durable in one pass. This is the flush half of group commit.

Returns :ok only when every record appended since the previous flush is durable. Otherwise it returns {:error, [{segment_id, reason}]} naming each segment whose buffered records were lost to a storage failure, whether that failure happened in this flush or earlier, between the append and it: either way the append was already answered {:ok, last}, and a caller that acks on this reply must not ack those records.

@spec follow(term(), term(), non_neg_integer(), [Malachi.Log.Record.t()]) ::
  {:ok, non_neg_integer()}
  | {:error, :out_of_sync | :unreachable | {:storage, term()}}

Appends a replicated batch of segment_id to this server (the follower side). expected_first is the offset the batch must start at: it must equal this server's current end for the segment (or the segment's base when it is opened here for the first time). Returns {:ok, last_offset} or {:error, :out_of_sync} if this server is behind.

This is the directed append, used by Malachi.Cluster.Catchup to copy a span into a target replica. The primary's own fan-out does not come through here: it pushes :replica_append casts from its loop, which is what keeps the pipeline per-pair FIFO. A batch that arrives here never triggers a catch-up, since the caller is already driving one.

{:error, :unreachable} when the target cannot be called (a server that is not running, its node restarting, or a call that times out), as read/4 answers: the caller is a healing pass, and the exit would take it down with the copy.

@spec read(term(), term(), non_neg_integer(), pos_integer(), timeout()) ::
  {:ok, [Malachi.Log.Record.t()]} | :eof | {:error, term()}

Reads up to max_records records of segment_id stored on this server, from offset.

A server that is down or on an unreachable node answers {:error, :unreachable} rather than exiting the caller. The caller is the broker loop serving a consumer, and letting it exit would take the whole node's reads down with the one segment whose primary went away.

timeout (5 seconds by default) bounds the wait. A call that runs past it answers {:error, :unreachable} too, the same answer a server that is down gives, so a caller that passes a short timeout must not read that answer as proof the replica is gone: a slow but live server gives it as well.

@spec replicate(term(), term(), [term()], non_neg_integer(), [Malachi.Log.Record.t()]) ::
  {:ok, non_neg_integer()}
  | {:error,
     :no_quorum
     | :not_primary
     | :empty
     | :empty_replica_set
     | :unreachable
     | {:sealed, non_neg_integer()}
     | {:storage, term()}}

Replicates records for segment_id across replica_set, called on the primary (the first broker of the set). base_offset is the segment's first offset; it is used only when the segment's log is opened for the first time, so a segment's offsets continue its range (start_offset, start_offset + 1, ...) rather than restarting at zero.

Returns {:ok, last_offset} once a quorum has the batch durably, {:error, :no_quorum} if too few replicas acked, {:error, :not_primary} if this server is not the set's primary, {:error, :empty} for an empty batch, {:error, {:sealed, end_offset}} for a fenced segment, or {:error, {:storage, reason}} when this server's copy failed (see "Storage failures").

@spec replicate_async(
  term(),
  term(),
  [term()],
  non_neg_integer(),
  [Malachi.Log.Record.t()],
  pid(),
  term()
) :: :ok

Fire-and-forget variant of replicate/5 for a frontend that must never block its loop on replication (the NorthGuard end-to-end pipelined produce): same semantics and quorum rules, but the result is DELIVERED as a message {:replicate_result, tag, {:ok, last} | {:error, reason}} to notify_pid instead of a call reply. The caller owns retry/timeout policy for a lost cast (an unreachable primary never answers), typically with its own safety timer.

@spec seal(term(), term(), non_neg_integer(), timeout()) ::
  {:ok, non_neg_integer(), non_neg_integer()} | {:error, term()}

Seals segment_id on this server and reports what it durably holds: {:ok, end_offset, byte_size}.

The write fence. After it returns, replicate/5, replicate_async/7 and append/5 are all refused here with {:error, {:sealed, end_offset}}, and again after a restart, so the segment's log can never grow past the offset reported. That is what lets a caller RECORD the returned end as the sealed length instead of measuring one beside it: the length becomes a consequence of closing the segment rather than a number racing it.

The primary's :replica_append fan-out is refused as well, but it answers {:error, :sealed} WITHOUT the end offset. The primary discards an ack's error reason and only stops counting that replica toward the quorum, so an offset nobody reads would be one more message shape to keep in step across a rolling upgrade, for no reader.

Idempotent and cheap on an already-fenced segment (no fsync, the same pair). base_offset seats a missing or empty log at the segment's base, so fencing a segment this server never stored succeeds at {:ok, base_offset, 0} and refuses a later stray append rather than silently accepting it. Reads (read/4) and repair (follow/4) still work on a fenced segment.

A dead or unreachable server answers {:error, :unreachable} rather than exiting the caller, as read/4 and delete/2 already do: the caller is the broker loop, and letting it exit would take a node's writes down with one segment.

@spec seal_async(term(), term(), non_neg_integer(), pid(), term()) :: :ok

The same fence as seal/4, without waiting for it: the answer is DELIVERED to notify_pid as {:seal_result, tag, {:ok, end_offset, byte_size} | {:error, reason}}.

For a caller that must not block its loop on a network call, which is the produce roll in Malachi.BrokerServer. A cast to a dead or unreachable server is simply never answered, so the caller owns the retry: sending the fence again is safe, since it is idempotent and answers the same numbers.

@spec seal_at(term(), term(), non_neg_integer(), non_neg_integer(), timeout()) ::
  {:ok, non_neg_integer(), non_neg_integer(), non_neg_integer()}
  | {:error, term()}

Brings segment_id down to end_offset and fences it there: {:ok, end_offset, byte_size, dropped}.

The same fence as seal/4, for a copy whose disk disagrees with the control plane. A replica can come back holding records past the length its segment was sealed at (an old primary that wrote locally, lost quorum and rejoined after the seal landed on the others), and nothing else brings it back: the fence only stops a copy GROWING after the seal reached this server, and the sealed-copy integrity probe reports a copy that falls short, never one that runs past.

For a SEALED segment only, and the length is the caller's, never this server's. A sealed length is immutable once the control plane records it (Malachi.Metadata), and the offsets above it already belong to the successor segment, so the recorded number is the one that decides and the bytes above it were never acknowledged to anyone. This server holds no control-plane state of its own and reads none; Malachi.Cluster.SealedOverrun is what knows the number and passes it in.

Truncating comes first and fencing second, always. The reverse order leaves a copy fenced ABOVE the recorded length, which is the same defect wearing a marker, and the marker is what later passes use to skip a copy they have already settled.

An end_offset at or above what this server holds trims nothing and fences as seal/4 does: a copy that is SHORT of the recorded length is a different defect, repaired from a peer by Malachi.Cluster.SelfHealing, and repair still works on a fenced segment (follow/4).

An end_offset below the base offset the caller passed answers {:error, :below_base_offset} without opening the log and without failing the copy. An end_offset below the log's real base is refused by the log, after the log is opened and flushed; that refusal does not fail the copy, though an error opening or flushing the log can. Every storage error here takes the copy out of service, except a segment directory or seal marker that a read cannot list or stat. Those answer their reason and leave the copy in service, as opening does: bare on a log that is already open, as {:storage, reason} on one that has to be opened. A write or an fsync that fails on the directory latches like any other storage error, and so does a directory that cannot be opened to fsync. A coordinator that passed the wrong number must not cost a healthy replica.

dropped is how many records the truncation removed, and it is why this answers a four-tuple where seal/4 answers three. Settling a copy is overwhelmingly a no-op that only writes a marker, and one that actually drops a record is the defect this exists for happening in production. A caller that counted both the same way would bury the second in the first on the very first pass, which is the one pass that touches every copy in the cluster.

@spec settled_confirmed?(term(), term(), timeout()) :: boolean()

Whether confirm_settled/3 has been recorded for this server's copy of segment_id. A single check that the segment's SETTLED file is a regular file: the probe asks it only for copies it already found above the recorded size. A latched segment answers false, since its marker is not read while it refuses every request.

@spec start_link(keyword()) :: GenServer.on_start()

Starts a replication server.

Options

• :name (optional) - the broker reference this server is registered under and known by in replica sets. When omitted, the server is unregistered and its reference is its pid.
• :directory (required) - where replicated segment logs are stored.
• :follow_timeout - ms a parked replicate waits for its quorum before :no_quorum (default 5000).
• :replication_window - max unacked replica-append batches in flight per segment (default 32).
• :group_commit - coalesce fsyncs under replication (NorthGuard: fsync on every replica by time/count/size triggers, before the produce ack). Default false (fsync per batch).
• :group_commit_interval_ms - the time trigger for that coalescing (default 10).
• :min_heap_size - the server process's minimum heap, in words. Defaults to the :malachi application's :replication_min_heap_size, and to 0 (the VM's own default) when that is unset.
• any remaining options are forwarded to each segment's Malachi.Log.

@spec stored_bytes(term(), term(), timeout()) :: non_neg_integer() | {:error, term()}

The on-disk byte size this server stores for segment_id: the sum of its segment files' sizes, read without opening the log (no descriptors, no state change), so it is cheap enough to poll. 0 when nothing is stored. A segment that failed in storage answers the storage error instead. The first stage of the sealed-copy integrity probe (Malachi.Cluster.SelfHealing): a sealed segment whose stored bytes fall short of the metadata's sealed byte_size has lost data on this replica. Only meaningful for sealed segments; an active segment's file legitimately trails its in-memory log by the unflushed buffer.