An append-only log of opaque entries, with snapshots that replace its
prefix, stored in Durable Streams (Slap.Streams) below a base path. A
consumer rebuilds its state from the current snapshot and the entries
after it, follows the log to stay current, and publishes snapshots of
what it has applied, which trims the entries they cover.
{:reset, %{snapshot: nil, offset: 0}} = SnapshotLog.next(base, nil)
{:ok, _} = SnapshotLog.append(base, "entry")
{:ok, %{entries: ["entry"], offset: offset}} = SnapshotLog.next(base, 0)
:ok = SnapshotLog.snapshot(base, offset, state_bytes)A consumer's loop starts with next(base, nil) and handles {:reset, …}
the same way on start and later: another process's snapshot may trim
the entries the consumer has not read yet.
Snapshots
A snapshot at offset must contain the effect of every entry before
offset; anything more it contains must be harmless to apply again when
the entries from offset on are replayed. A consumer whose entries cannot
be applied twice snapshots only the entries it has read from the log; one
whose entries can (Yjs updates, for example) may snapshot its live state.
The log cannot check this: snapshots and entries are opaque bytes.
Snapshots are immutable. Of several processes publishing at once, only
the snapshot at the highest offset becomes current ({:error, :superseded} for the others), and each step of a publication can be
interrupted: the log stays readable, and a later publication, or a
retry at the same offset, cleans up.
Deletion
delete/1 deletes the log for good: its entries and snapshots go, and
every later call on its base returns {:error, :deleted}, including a
publication or an append that started before the delete. A base cannot
be used again; its .index, an empty, closed .updates and a seal on
its placement group (Slap.Streams.seal/2) stay, to fence out those
calls. The log reads as deleted from the delete's first step, so an
interrupted delete acknowledges nothing more; a retry finishes it.
Storage
The log's streams are <base>/.updates (the entries, one message each),
<base>/.index (JSON entries {"snapshotOffset", "createdAt", "retained"}, the last current) and <base>/.snapshots/<offset>_snapshot:
the reference y-durable-streams server's layout. They share base as
their placement key, so they are on one shard. base must be a stream
path with no segment that starts with ..
Pass cluster: to use an application-owned cluster defined with
use Slap.Streams.Cluster, otp_app: :my_app; run one Streams cluster per VM.
Errors
Invalid base paths, entries, snapshots and offsets return
{:error, {:bad_request, reason}}. Invalid control options and unknown
option names raise ArgumentError. The path/2 path helper raises for an
invalid base path.
Calls return Slap.Streams errors, and {:error, :deleted} on a
deleted log (next/3 also when the log is deleted while it waits). On
an append, :unavailable and
:timeout mean the outcome is unknown: the entry may still be stored.
Retry an ambiguous append with the same :producer sequence to avoid a
second entry. next/3 retries transient read errors within its :timeout
budget.
:timeout on append/3, snapshot/4, snapshots/2,
read_snapshot/3, tail/2, and delete/2 bounds each underlying
Streams call (default 30 s). For next/3 it bounds the whole retry
loop (default 60 s).
Summary
Functions
Appends an entry (a non-empty binary), creating the log if needed.
Returns {:ok, offset}, the offset after the entry, once it is durable.
On {:error, :unavailable} or {:error, :timeout} the entry may still
be stored; it is not retried.
Deletes the log for good (see "Deletion"). Deleting a deleted log, or one never created, is fine; a retry finishes an interrupted delete.
Reads the log. With offset nil, or an offset its entries were trimmed
past, returns {:reset, %{snapshot: bytes | nil, offset: offset}}: the
current snapshot (nil if none), and the offset to read the entries after
it from. Otherwise returns {:ok, %{entries: entries, offset: next, up_to_date: boolean}}: a page of entries from offset, the offset after
them, and whether they reach the durable tail.
The path of the log's :updates or :index stream, or of a {:snapshot, offset}.
The snapshot at offset.
Publishes a snapshot that replaces the entries before offset (see
"Snapshots"), makes it current, then trims those entries. Returns :ok,
also when it is current already (finishing the cleanup an interrupted
publication left), {:error, :superseded} when a snapshot at a higher
offset became current first, or {:error, :offset_beyond_tail} when
offset is after the tail.
The stored snapshots, oldest first: the current one, and the older ones
the history rules keep. Each is %{offset: offset, created_at: ms}.
The durable tail: the offset after the last entry.
Types
@type base() :: String.t()
@type error() :: Slap.Streams.error() | :deleted | :index_seq_conflict | :offset_beyond_tail | :superseded
@type offset() :: non_neg_integer()
Functions
@spec append(base(), binary(), keyword()) :: {:ok, offset()} | {:duplicate, offset()} | {:error, error()}
Appends an entry (a non-empty binary), creating the log if needed.
Returns {:ok, offset}, the offset after the entry, once it is durable.
On {:error, :unavailable} or {:error, :timeout} the entry may still
be stored; it is not retried.
:producer accepts {id, epoch, seq} as in Slap.Streams.append/3.
Retrying with the same producer sequence does not append again and returns
{:duplicate, tail}. tail is the current tail, which may be past the
original entry. :timeout bounds each Streams call; :cluster selects
the Streams cluster.
Deletes the log for good (see "Deletion"). Deleting a deleted log, or one never created, is fine; a retry finishes an interrupted delete.
@spec next(base(), offset() | nil, keyword()) :: {:ok, page()} | {:reset, reset()} | {:error, error()}
Reads the log. With offset nil, or an offset its entries were trimmed
past, returns {:reset, %{snapshot: bytes | nil, offset: offset}}: the
current snapshot (nil if none), and the offset to read the entries after
it from. Otherwise returns {:ok, %{entries: entries, offset: next, up_to_date: boolean}}: a page of entries from offset, the offset after
them, and whether they reach the durable tail.
At the tail with nothing to read, waits up to :wait ms (default 30,000;
0 returns at once) for an entry. Options also: :max_bytes (about how
many bytes a page holds, default 1 MiB; at least one entry), :timeout
(total budget including waits and retries, default 60,000 ms), and
:cluster (the Streams cluster). next(base, nil) creates the log if it
does not exist; a timed-out call may still have created it.
The path of the log's :updates or :index stream, or of a {:snapshot, offset}.
The snapshot at offset.
Publishes a snapshot that replaces the entries before offset (see
"Snapshots"), makes it current, then trims those entries. Returns :ok,
also when it is current already (finishing the cleanup an interrupted
publication left), {:error, :superseded} when a snapshot at a higher
offset became current first, or {:error, :offset_beyond_tail} when
offset is after the tail.
Options: :history, a list of {every_ms, keep_ms} rules that keep the
latest snapshot of each every_ms period for keep_ms (default []:
only the current one), :cluster, and :timeout.
@spec snapshots(base(), keyword()) :: {:ok, [snapshot_info()]} | {:error, error()}
The stored snapshots, oldest first: the current one, and the older ones
the history rules keep. Each is %{offset: offset, created_at: ms}.
The durable tail: the offset after the last entry.