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

Buffers records, assigning each a globally monotonic offset. Records are not durable until sync/1. Returns the updated log and the first/last offsets assigned.

@spec bytes_on_disk(t()) :: non_neg_integer()

How many bytes this log holds on disk, counting committed records and not preallocated space.

Sealed segment files are measured with File.stat/1, because a sealed file is exactly its contents. The ACTIVE segment is asked instead (Malachi.Storage.SegmentStore.logical_bytes/1), because a store may size its file ahead of what it has written, and stat would then report the room rather than the log.

This is the number anything downstream wants: it is what gets recorded as a segment's byte_size when it is sealed, and from there it drives size-based retention and the lost-copy probe that compares replicas byte for byte.

@spec close(t()) :: :ok

Closes the active segment's file handle, if any.

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

Closes the log and removes its directory with every segment file, irreversible. The log owns its directory, so this drops all of its data (used by retention to expire a whole segment). Best-effort on the file removal: a leftover file without control-plane metadata is harmless.

@spec listed_bytes(t()) :: {:ok, non_neg_integer()} | {:error, term()}

bytes_on_disk/1, but a directory that cannot be listed, or a file that cannot be stat'ed, is an error rather than zero bytes. A directory that does not exist holds no sealed files, so only the active segment counts.

@spec open(
  Path.t(),
  keyword()
) :: {:ok, t()} | {:error, term()}

Opens a fresh, empty log in directory.

Options

• :base_offset - first offset of the log (default 0)
• :store - SegmentStore implementation (default ElixirStore)
• any remaining options (:max_bytes, :max_age_ms, :index_interval, :flush_bytes) are passed through to each segment.

@spec pending?(t()) :: boolean()

Whether the log has buffered records not yet flushed.

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

Reads up to max_records committed records from the segment containing offset. Returns :eof past the end of the log, or {:error, :out_of_range} below its start.

@spec rebuild_index(
  Path.t(),
  keyword()
) :: :ok | {:error, term()}

Rebuilds the sparse index of every segment in directory from the segments themselves.

The repair for a damaged index: it is derived data, so no replica is involved, unlike a damaged segment. Only call it after verify/2 reports the records intact, or the rebuilt index will faithfully describe the damage. An absent index file is not an error to begin with (reads just scan from the start of the segment), so a rebuild simply puts the fast path back.

@spec recover(
  Path.t(),
  keyword()
) :: {:ok, t()} | {:error, term()}

Reopens an existing log, recovering all segments. Only the last (active) segment is scanned; sealed segments are trusted via their file names.

A visible seal marker is fsynced through its directory before it is trusted, and a failed fsync is returned rather than trusted. A marker or segment directory that cannot be examined is an error too, never an absent marker. What this does not cover is a directory entry lost before any recovery ran: once that entry is gone, nothing here can see the marker, so the log comes back unsealed.

:sync_fun replaces the directory fsync (Malachi.Storage.Directory.sync/1 by default), the same seam Malachi.Storage.DataDirGuard takes, so a test can make it fail.

@spec roll(t()) :: {:ok, t()} | {:error, term()}

Forces the active segment to seal (no-op if there is no active segment).

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

Seals the whole log: flushes and fsyncs, seals and closes whatever segment file is still open, and records the seal on disk so a reopened log comes back sealed. Idempotent.

The flag is the durable RECORD of the seal, not its enforcement. Malachi.Cluster.ReplicationServer is what refuses a write to a sealed segment, because repair must still be able to write into one: its follow/4 copies records that already carry their assigned offsets, bounded by the sealed end, so it can never hand out a new offset, which is the only thing a fence has to prevent.

@spec seal_marker_path(Path.t()) :: Path.t()

Where a log records that it is closed as a whole, one level above Malachi.Log.Segment.seal_marker_path/1.

A separate marker on purpose: sync/1 seals FILES routinely on :max_bytes/:max_age_ms inside a live log, so "the last file is sealed" says nothing about whether the log is closed.

@spec seal_marker_visible(Path.t()) ::
  {:ok, boolean()} | {:error, {:seal_marker_unreadable, term()}}

Whether a seal marker is visible in directory. An absent marker answers {:ok, false}; a marker that cannot be examined is an error, never an absent one. It makes no durability promise: recover/2 is what fsyncs a visible marker before trusting it.

@spec sealed?(t()) :: boolean()

Whether the log as a whole is sealed (no further append will ever be accepted for it).

@spec segment_offsets(Path.t()) ::
  {:ok, [non_neg_integer()]} | {:error, {:segments_unreadable, term()}}

The base offsets of the segment files in directory, sorted. A directory that does not exist (or whose parent is a file) holds no segments, so {:ok, []}; one that cannot be listed is an error, never an empty list. Lists with File.ls/1 rather than a wildcard, which answers [] for a directory it cannot read. The names match Path.wildcard/1's *.log, which skips dotfiles.

@spec sync(t()) :: {:ok, t()} | {:error, term()}

Flushes and fsyncs the active segment, then rolls it if a seal threshold is hit.

@spec truncate_to(t(), non_neg_integer()) :: {:ok, t()} | {:error, term()}

Truncates the whole log so that end_offset becomes the offset after its last record: every segment file that begins at or above it is removed, and the file that straddles it keeps only the records below it. Idempotent, and a no-op when the log already ends at or below end_offset.

This is how a copy is brought back to the length the control plane recorded for its segment. It is deliberately NOT the inverse of seal/1: sealing a log is a latch and truncating one is not, so a caller that does both must truncate first and seal second, or it leaves a fenced copy still holding records past the fence. Malachi.Cluster.ReplicationServer.seal_at/5 is the only caller and does exactly that.

Answers {:error, :below_base_offset} for an end_offset under the log's first record, which is a caller asking about the wrong log rather than a log that needs emptying. An unreadable segment directory answers {:error, {:segments_unreadable, reason}} instead, because it says nothing about the caller's number. A cut that removes records flushes buffered records first, as seal/1 flushes: they are not durable, so nothing has acknowledged them, but a truncation measured against a file that a pending write is about to extend would answer for the wrong bytes. A no-op or a refusal answers before that flush.

After a cut has changed the disk, a reopen that fails answers {:error, {:reopen_after_cut, reason}}: the caller must drop the log it holds, whose handle is closed and whose end is stale.

@spec verify(
  Path.t(),
  keyword()
) ::
  {:ok,
   %{
     records: non_neg_integer(),
     bytes: non_neg_integer(),
     files: non_neg_integer()
   }}
  | {:error, map() | :enoent}

Verifies every segment file in directory read-only, checking each record's checksum, and aggregates the result. Unlike recover/2 it opens nothing for writing, repairs nothing, and trusts nothing by filename: this is what a background scrub uses to catch corruption at rest in segments that are sealed and therefore never re-scanned by the normal open path.

Returns {:ok, %{records: n, bytes: b, files: k}} when every file is intact, {:error, details} for the first damaged file (with its path, byte position and reason), or {:error, :enoent} when the directory holds no segment files at all (nothing stored here, or retention removed it).

Options

• :store - SegmentStore implementation (default Malachi.Storage.ElixirStore)