# `Ankusa.Edge.Quarantine`
[🔗](https://github.com/jamescarr/ankusa/blob/ankusa-v0.7.0/packages/ankusa/lib/ankusa/edge/quarantine.ex#L1)

A durable holding pen for envelopes that failed verification under a
`:quarantine` policy. A bad secret rotation should never silently eat real
events, and the pen is how they come back: a `kind: :quarantine` replay job
(`Ankusa.Replay`) re-verifies each held hook against its source's *current*
verifier and commits the ones that now pass; `purge/3` drops the rest.

A flood of forged requests must not fill the disk, or starve another
source's real failures of room, so the pen is bounded three times:

  * one token bucket per source (`config.quarantine`) — `burst` tokens,
    refilled `rate` per second. Over it, `put/3` answers
    `{:rate_limited, retry_after_ms}` (`429 quarantine_rate_limited`).
  * a cap on the pen's total bytes, `max_bytes`. A write that would cross
    it answers `:full` (`503 quarantine_full`). A full pen refuses new
    hooks; it never evicts one it already acked.
  * at most 1 000 writes waiting for the pen's disk write at once (`pending`,
    the `ankusa_quarantine_pending` gauge). Past it `put/3` answers
    `{:error, :overload}` (`503` with `Retry-After: 1`) and spends nothing.

## Who does what

The decisions are made in the calling request process, not in a queue: a
source's bucket is one ETS row updated with a compare-and-swap (GCRA, as
`Ankusa.Edge.RateLimiter`'s buckets are), and the byte and pending counts are
`:ets.update_counter/3`, so a flood on one source never makes another
source's `hit/3` or `put/3` wait for it. Only the disk write is shared: the
caller hands its two rows to the pen process, which commits every write that
has queued up meanwhile in **one** synced `Ankusa.Store.write/3` (a group
commit), so a flood costs one fsync per group instead of one per request and
holds one dirty-IO scheduler however many callers wait. The pen process
otherwise keeps the node-start recovery of the byte count, the bucket sweep,
the retention sweep and `purge/3`.

Each record is two keys in this node's `Ankusa.Store`, committed in one synced
batch: a small summary (id, source, tenant, time, reason, size) that
`recent/2` and `page/5` list, and the whole envelope, which `envelope/2`
reads back. `put/3` answers `:ok` only once both are on disk.

# `filter`

```elixir
@type filter() :: %{
  optional(:source_id) => String.t() | nil,
  optional(:id) => String.t() | nil,
  optional(:since) => non_neg_integer() | nil,
  optional(:until) => non_neg_integer() | nil,
  optional(:tenants) => [String.t()] | nil
}
```

`:source_id`/`:id` match the summary; `:since`/`:until` (ms, inclusive) bound
`received_at`; `:tenants`, a list of tenant ids (`nil`: every tenant), keeps
the entries of those tenants.

# `child_spec`

Returns a specification to start this module under a supervisor.

See `Supervisor`.

# `delete_ops`

```elixir
@spec delete_ops(binary()) :: [Ankusa.Store.op()]
```

The store ops that delete the entry under `summary_key`: both of its keys.

# `envelope`

```elixir
@spec envelope(atom(), {binary(), map()}) ::
  {:ok, Ankusa.Envelope.t()} | :not_found | {:error, :undecodable | term()}
```

The held envelope for a `page/5` hit. An entry written before the pen kept
whole envelopes (by 0.4, or imported from a 0.3 data dir) holds only headers
and body; it comes back as a `POST /` from the summary's source with no
tenant (the caller supplies the source's own).

# `held`

```elixir
@spec held(atom()) ::
  {:ok,
   %{
     bytes: non_neg_integer(),
     entries: non_neg_integer(),
     pending: non_neg_integer()
   }}
  | {:error, :store_unavailable}
```

What the pen holds: `bytes` (the count the `max_bytes` cap is enforced
against), `pending` (writes waiting for the pen's disk write) and `entries`.
The entry count is a scan of the pen's summary keys, run in the caller, so it
never queues a write behind it. A pen that is not running is
`{:error, :store_unavailable}`.

# `hit`

```elixir
@spec hit(atom(), String.t(), String.t() | nil) ::
  :ok | {:rate_limited, pos_integer()} | {:error, :store_unavailable}
```

Spend one token of `source_id`'s bucket without writing anything: the charge
for a hook a source accepts *flagged* (`on_verify_failure: :accept_flag`),
which is committed like any hook but must not be free to forge. Same bucket,
same arithmetic as `put/3`; `tenant_id` only labels the refusal's telemetry.
`{:rate_limited, retry_after_ms}` when it is empty; `{:error, :store_unavailable}`
when the pen's tables are gone (it is restarting).

# `page`

```elixir
@spec page(atom(), {binary(), binary()}, filter(), pos_integer(), pos_integer()) ::
  {:ok, [{binary(), map()}], binary() | nil, boolean(), non_neg_integer()}
  | {:error, term()}
```

A paged scan over the pen's summaries in `{lower, upper}` (half-open), for
the replay engine. Keeps entries matching `filter` (`:source_id`, `:id`; the
range carries time). Stops after `limit` hits or `max_scan` keys examined; a
summary that does not decode is scanned and skipped. Returns the hits
ascending as `{summary_key, summary}`, the last key seen (`nil` when the
range held nothing), whether the range was exhausted, and the keys scanned.

# `prune`

```elixir
@spec prune(atom()) ::
  {:ok,
   %{deleted: non_neg_integer(), bytes: non_neg_integer(), more: boolean()}}
  | {:error, term()}
```

Delete the entries held longer than `quarantine.retention_days`, oldest first,
in synced batches of at most 1000, each with its audit row
(`quarantine.retention`). An entry a live `kind: :quarantine` replay job can
still reach (received at or after the earliest `since` of one) stays; so does
everything while the replayer cannot say. Runs hourly, in a process of its own;
callable on its own. `more` says a pass's budget, not the end of the pen,
stopped it. Emits `[:ankusa, :quarantine, :pruned]` (`count`, `bytes`) per batch.

# `purge`

```elixir
@spec purge(atom(), filter(), pos_integer(), keyword()) ::
  {:ok, %{deleted: non_neg_integer(), bytes: non_neg_integer()}}
  | {:error, :store_unavailable}
```

Delete up to `limit` entries matching `filter`, oldest first, in one synced
batch together with the purge's audit row (`quarantine.purge`, with the
filter, the count and the bytes as its `detail`). Returns how many went and
the bytes they held. A pen that is not running (its domain restarting) is
`{:error, :store_unavailable}`. Options: `audit:`, the
`Ankusa.Audit.Context` of the change.

# `put`

```elixir
@spec put(atom(), Ankusa.Envelope.t(), term()) ::
  :ok
  | {:rate_limited, pos_integer()}
  | :full
  | {:error, :store_unavailable | :tenant_erasing | :overload}
```

Durably record a quarantined envelope. Returns `:ok` once it is on disk;
`{:rate_limited, retry_after_ms}` when the source's bucket is empty; `:full`
when the entry would take the pen past `max_bytes` (no token is spent);
`{:error, :overload}` when 1000 writes already wait for the pen (nothing is
spent); `{:error, :store_unavailable}` when the store could not take the
write (no token is spent) or the pen did not answer; `{:error,
:tenant_erasing}` when an erase of the envelope's tenant is running
(`Ankusa.Tenant`).

# `range`

```elixir
@spec range(integer() | nil, integer() | nil) :: {binary(), binary()}
```

The summary-key range holding entries received in `since..until` (ms,
inclusive; `nil` is unbounded). A bound outside what the key's 64 bits hold
is clamped, never wrapped: a negative `until` is an empty range, not the
whole pen.

# `recent`

```elixir
@spec recent(atom(), non_neg_integer(), filter()) :: {:ok, [map()]} | {:error, term()}
```

The most recent quarantined entries matching `filter` (`:source_id`, `:id`,
`:tenants`), newest first, at most `limit`. Reads the store directly, so it
never queues behind a write.

# `released`

```elixir
@spec released(atom(), non_neg_integer()) :: :ok
```

Tell the pen `bytes` left it outside `purge/3` — a release commit deleted
their entries. A no-op when the pen is not running on this node.

# `start_link`

# `validate_config!`

```elixir
@spec validate_config!(Ankusa.Config.t()) :: :ok
```

Check `config.quarantine`; raises `ArgumentError` naming the bad key.

---

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