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) —bursttokens, refilledrateper second. Over it,put/3answers{: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, theankusa_quarantine_pendinggauge). Past itput/3answers{:error, :overload}(503withRetry-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.
Summary
Types
: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.
Functions
Returns a specification to start this module under a supervisor.
The store ops that delete the entry under summary_key: both of its keys.
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).
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}.
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).
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.
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.
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.
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).
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.
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.
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.
Check config.quarantine; raises ArgumentError naming the bad key.
Types
@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.
Functions
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec delete_ops(binary()) :: [Ankusa.Store.op()]
The store ops that delete the entry under summary_key: both of its keys.
@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).
@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}.
@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).
@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.
@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.
@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.
@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).
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.
@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.
@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.
@spec validate_config!(Ankusa.Config.t()) :: :ok
Check config.quarantine; raises ArgumentError naming the bad key.