# `Ankusa.Intake.Pending`
[🔗](https://github.com/jamescarr/ankusa/blob/ankusa-v0.7.0/packages/ankusa/lib/ankusa/intake/pending.ex#L1)

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, for `Ankusa.Intake.Reconciler.hold/3`
  * `{:k, {tenant, source, dedupe_key}}` — seq of a keyed entry: a retry that
    meets it is never answered `201` (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`.

# `entry`

```elixir
@type entry() :: tuple()
```

# `table`

```elixir
@type table() :: atom()
```

# `batch`

```elixir
@spec batch(table(), term()) :: map() | nil
```

The batch row (ops to re-apply) of a batch, if its local half is still missing.

# `bump_confirmed`

```elixir
@spec bump_confirmed(table(), non_neg_integer() | nil) :: :ok
```

Raises the highest confirmed position (never lowers it).

# `carried`

```elixir
@spec carried(table()) :: [non_neg_integer()]
```

Every carried seq (the reconciler's stuck sweep).

# `carry`

```elixir
@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).

# `confirm`

```elixir
@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.

# `create`

```elixir
@spec create(atom()) :: table()
```

Creates the instance's table. Called from the instance supervisor's `init/1`.

# `due`

```elixir
@spec due(table(), integer(), pos_integer()) :: [non_neg_integer()]
```

Carried entries due at or before `now`, oldest schedule first (at most `limit`).

# `enter`

```elixir
@spec enter(table(), [tuple()], term(), integer()) :: :ok
```

Enters the records of one batch before either half is touched. `records` are
`{seq, id, size, triple | nil, meta}`.

# `exists?`

```elixir
@spec exists?(atom()) :: boolean()
```

Whether the instance has a table (it runs `wal: :dual`).

# `hold`

```elixir
@spec hold(table(), String.t(), term()) :: :held | :completed | :stuck
```

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

# `keyed?`

```elixir
@spec keyed?(table(), {String.t(), String.t(), String.t()}) :: boolean()
```

Whether an unconfirmed record holds this `{tenant, source, dedupe_key}`.

# `local_done`

```elixir
@spec local_done(table(), term()) :: [entry()]
```

The local half of a batch is in the store. Returns the entries that this
completed (their log half was already confirmed).

# `log_done`

```elixir
@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.

# `lookup`

```elixir
@spec lookup(table(), non_neg_integer()) :: entry() | nil
```

# `mark_stuck`

```elixir
@spec mark_stuck(table(), non_neg_integer()) :: boolean()
```

Marks an entry stuck; `true` when it was not already.

# `meta`

```elixir
@spec meta(table(), atom(), term()) :: term()
```

# `offset`

```elixir
@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.

# `partition`

```elixir
@spec partition(atom()) :: non_neg_integer() | nil
```

The partition this node writes, or `nil` (adapter without one, or no dual boot yet).

# `put_meta`

```elixir
@spec put_meta(table(), atom(), term()) :: true
```

# `recovering?`

```elixir
@spec recovering?(table()) :: boolean()
```

`:recovering` while boot recovery runs, `:up` otherwise.

# `reschedule`

```elixir
@spec reschedule(table(), non_neg_integer(), integer(), boolean()) :: :ok
```

Schedules the next attempt of a carried entry.

# `room?`

```elixir
@spec room?(table(), Ankusa.Config.t(), non_neg_integer(), non_neg_integer()) ::
  boolean()
```

Whether `count` more records of `bytes` fit under the pending ceilings.

# `stats`

```elixir
@spec stats(table(), integer()) :: map()
```

Counts for the gauges and `GET /v1/wal`.

# `table`

```elixir
@spec table(atom()) :: table()
```

The table's name (the instance name is operator config, never request input).

# `take_claim`

```elixir
@spec take_claim(table(), non_neg_integer()) :: term()
```

The claim a stuck entry held, cleared so it is released once.

# `watermark`

```elixir
@spec watermark(table(), pos_integer()) :: non_neg_integer()
```

The seq below which every record is confirmed in both places (ADR section 5).

---

*Consult [api-reference.md](api-reference.md) for complete listing*