@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 delete(term(), term()) :: :ok

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.

@spec durable_end(term(), term(), non_neg_integer(), timeout()) :: non_neg_integer()

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.

@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 flush(term()) :: :ok

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

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

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.

@spec read(term(), term(), non_neg_integer(), pos_integer()) ::
  {: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.

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

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, or {:error, :empty} for an empty batch.

@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 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).
• any remaining options are forwarded to each segment's Malachi.Log.

@spec stored_bytes(term(), term(), timeout()) :: non_neg_integer()

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