The pending set of wal: :dual (ADR 0002, section 5): one entry per record not
yet confirmed in both the local store and the intake log, and the small
state the dual commit shares between the writer, the reconciler, the breaker
and readiness.
One public ETS ordered_set per instance, created by Ankusa.Instance (so it
outlives the processes that use it). Its keys, by tag:
{:p, seq}— the entry (see@pos_*); in flight until its batch's publish returns, carried afterwards if a half failed{:id, id}— seq of a carried entry, forAnkusa.Intake.Reconciler.hold/3{:k, {tenant, source, dedupe_key}}— seq of a keyed entry: a retry that meets it is never answered201(ADR section 3, rule 2){:b, batch_id}— the ops of a batch whose local half is missing, to re-apply{:c, next_at, seq}— the reconciler's schedule{:h, position, seq}— carried entries whose record is in the log but not the local store: the offset never passes the lowest (section 5){:meta, name}— counters and flags (:records,:bytes,:carried,:stuck,:confirmed_max,:state,:partition,:flushed_at)
Entries are small tuples updated in place (:ets.update_element/3), so the
writer, the reconciler and hold/3 race safely: a completed entry is removed
with one atomic :ets.take/2.
Summary
Functions
The batch row (ops to re-apply) of a batch, if its local half is still missing.
Raises the highest confirmed position (never lowers it).
Every carried seq (the reconciler's stuck sweep).
A half of the batch failed: the records stay, carried. local and log are
:ok | :missing for the whole batch; records are {seq, bin, meta} (kept
while the log half is missing), batch is %{id, ops, probe_seq, seqs, at}
(kept while the local half is missing).
Both halves of the batch's records are confirmed: remove them and advance the highest confirmed position.
Creates the instance's table. Called from the instance supervisor's init/1.
Carried entries due at or before now, oldest schedule first (at most limit).
Enters the records of one batch before either half is touched. records are
{seq, id, size, triple | nil, meta}.
Whether the instance has a table (it runs wal: :dual).
Lets the reconciler settle claim once both halves of hook id are
confirmed. :completed when it already finished (the caller promotes),
:stuck when it has been past reconcile_deadline_ms (the caller releases).
Whether an unconfirmed record holds this {tenant, source, dedupe_key}.
The local half of a batch is in the store. Returns the entries that this completed (their log half was already confirmed).
The log half of one entry is confirmed at position; the entries it completed.
Marks an entry stuck; true when it was not already.
The highest log position whose record is also in the local store: the highest
confirmed position, held below the lowest record that is only in the log.
nil when none is known.
The partition this node writes, or nil (adapter without one, or no dual boot yet).
:recovering while boot recovery runs, :up otherwise.
Schedules the next attempt of a carried entry.
Whether count more records of bytes fit under the pending ceilings.
Counts for the gauges and GET /v1/wal.
The table's name (the instance name is operator config, never request input).
The claim a stuck entry held, cleared so it is released once.
The seq below which every record is confirmed in both places (ADR section 5).
Types
Functions
The batch row (ops to re-apply) of a batch, if its local half is still missing.
@spec bump_confirmed(table(), non_neg_integer() | nil) :: :ok
Raises the highest confirmed position (never lowers it).
@spec carried(table()) :: [non_neg_integer()]
Every carried seq (the reconciler's stuck sweep).
@spec carry( table(), %{local: atom(), log: atom(), position: integer() | nil}, [tuple()], map(), integer() ) :: :ok
A half of the batch failed: the records stay, carried. local and log are
:ok | :missing for the whole batch; records are {seq, bin, meta} (kept
while the log half is missing), batch is %{id, ops, probe_seq, seqs, at}
(kept while the local half is missing).
@spec confirm(table(), [non_neg_integer()], non_neg_integer() | nil) :: :ok
Both halves of the batch's records are confirmed: remove them and advance the highest confirmed position.
Creates the instance's table. Called from the instance supervisor's init/1.
@spec due(table(), integer(), pos_integer()) :: [non_neg_integer()]
Carried entries due at or before now, oldest schedule first (at most limit).
Enters the records of one batch before either half is touched. records are
{seq, id, size, triple | nil, meta}.
Whether the instance has a table (it runs wal: :dual).
Lets the reconciler settle claim once both halves of hook id are
confirmed. :completed when it already finished (the caller promotes),
:stuck when it has been past reconcile_deadline_ms (the caller releases).
Whether an unconfirmed record holds this {tenant, source, dedupe_key}.
The local half of a batch is in the store. Returns the entries that this completed (their log half was already confirmed).
@spec log_done(table(), non_neg_integer(), non_neg_integer()) :: [entry()]
The log half of one entry is confirmed at position; the entries it completed.
@spec lookup(table(), non_neg_integer()) :: entry() | nil
@spec mark_stuck(table(), non_neg_integer()) :: boolean()
Marks an entry stuck; true when it was not already.
@spec offset(table()) :: non_neg_integer() | nil
The highest log position whose record is also in the local store: the highest
confirmed position, held below the lowest record that is only in the log.
nil when none is known.
@spec partition(atom()) :: non_neg_integer() | nil
The partition this node writes, or nil (adapter without one, or no dual boot yet).
:recovering while boot recovery runs, :up otherwise.
@spec reschedule(table(), non_neg_integer(), integer(), boolean()) :: :ok
Schedules the next attempt of a carried entry.
@spec room?(table(), Ankusa.Config.t(), non_neg_integer(), non_neg_integer()) :: boolean()
Whether count more records of bytes fit under the pending ceilings.
Counts for the gauges and GET /v1/wal.
The table's name (the instance name is operator config, never request input).
@spec take_claim(table(), non_neg_integer()) :: term()
The claim a stuck entry held, cleared so it is released once.
@spec watermark(table(), pos_integer()) :: non_neg_integer()
The seq below which every record is confirmed in both places (ADR section 5).