Ankusa.Edge.Ingest (ankusa v0.7.0)

Orchestrates one webhook on the hot path: build the envelope, verify inline, apply the per-source failure policy, then ack once the hook is durable.

Which system accepts it is one config key. With wal.type: disk (the default) the group-commit batcher commits the hook to this node's WAL and dispatch reads it back later; with wal.type: none the hook is published to the source's sinks inside the request and the ack waits for their confirms (Ankusa.Edge.Publish).

The only place a 2xx-worthy result is produced is after one of those durable accepts returns, or after a durable quarantine write. Everything else is a non-2xx.

Stored headers

Verification and the dedupe key read the request's complete header list. Only then does the stored-header policy (Ankusa.Edge.StoredHeaders: the instance's store_headers, overridden by the source's) remove or hash credential-bearing headers, so the hook that is committed, archived, claim-checked and published to sinks never carries them. The number removed is :dropped_headers on the [:ankusa, :ingest] span's stop metadata.

Lookup precedes the body

The source is looked up (lookup/2) before the request body is read, and ingest/4 only ever runs on a source that exists. A request for a source that does not exist is a refusal (refused/2), counted on its own bounded counter and never on the [:ankusa, :ingest] span, so an unauthenticated caller cannot mint a metric series per URL it invents. ingest/2 is the transport-less entry point (tests, benches, embedding): lookup, then ingest.

Admission

What may be refused without the body — the tenant's max_body_bytes, and the per-address, per-node and per-source limits — is the HTTP edge's, after the lookup and before the read (Ankusa.Edge.Router, Ankusa.Edge.RateLimiter.pre_read/4). What needs the verified hook is here: the tenant's hooks and bytes per second ({:error, {:rate_limited, ms}}), and its share of the batchers, charged by Ankusa.Edge.Batcher.commit/4 ({:error, {:tenant_overload, ms}}). Both are 429s, and neither is the node's 503 overload. With rate_limits.scope: :fleet / dedupe.scope: :fleet the quota and the provider event key are counted fleet-wide (Ankusa.Edge.Fleet): an event key another node has claimed but not yet made durable is {:error, {:dedupe_pending, ms}} (409, retried by the sender), and one it has committed is a {:duplicate, env} carrying that node's hook id.

Tracing

The request's traceparent/tracestate become Ankusa.Envelope.trace (Ankusa.Trace). With an OpenTelemetry SDK running, the whole ingest is an ankusa.ingest span (a child of the provider's), with ankusa.verify and ankusa.commit inside it, and the hook stores the ingest span's context instead. Without one, the request's context is stored verbatim and no span exists.

The request process's Logger metadata carries source_id, tenant_id, hook_id and trace_id while a hook is ingested (and Ankusa.Edge.Router clears them per request), so any line logged for it correlates; the server image's JSON log formatter prints them.

Summary

Types

raw_request()

refusal()

Why a request was answered before it reached ingest/4. The whole set: every refusal goes through refused/2, so a metric label built from it is bounded. The *_rate_limited ones are the pre-read admission limits (Ankusa.Edge.RateLimiter.pre_read/4): per client address, per node, per source.

request()

result()

Functions

check_headers(list)

Refuse header bytes no sink can carry: a name must be visible ASCII (0x21..0x7E), a value visible ASCII, space or tab (RFC 9110 field-value without obs-text). Returns the offending name when it is itself printable.

ingest(instance, req)

Look the source up, refuse undeliverable header bytes, then ingest/4.

ingest(instance, source, tenant_id, req)

Ingest one hook for a source lookup/2 returned. The caller has already run check_headers/1.

lookup(instance, route, binding \\ nil)

Resolve a route to its source and the tenant the hook is stored under.

refused(instance, reason)

Count a request answered without being ingested. Emits [:ankusa, :ingest, :refused] with the :reason; the one place a refusal is counted.

Types

raw_request()

@type raw_request() :: %{
  :method => String.t(),
  :path => String.t(),
  :headers => [{String.t(), String.t()}],
  :body => binary(),
  optional(atom()) => term()
}

refusal()

@type refusal() ::
  :unknown_source
  | :payload_too_large
  | :body_read_failed
  | :invalid_header
  | :store_unavailable
  | :ip_rate_limited
  | :node_rate_limited
  | :source_rate_limited

request()

@type request() :: %{
  :source_id => String.t(),
  :method => String.t(),
  :path => String.t(),
  :headers => [{String.t(), String.t()}],
  :body => binary(),
  optional(:tenant_id) => String.t()
}

result()

@type result() ::
  {:ok, Ankusa.Envelope.t()}
  | {:duplicate, Ankusa.Envelope.t()}
  | {:quarantined, term()}
  | {:rejected, term()}
  | {:error,
     :unknown_source
     | :overload
     | :tenant_limit
     | :tenant_erasing
     | :dedupe_quota
     | :dlq_full
     | :store_unavailable
     | :intake_unavailable
     | :intake_pending
     | :quarantine_full
     | {:invalid_header, String.t() | nil}
     | {:rate_limited, pos_integer()}
     | {:tenant_overload, pos_integer()}
     | {:quarantine_rate_limited, pos_integer()}
     | {:dedupe_pending, pos_integer()}}

Functions

check_headers(list)

@spec check_headers([{String.t(), String.t()}]) ::
  :ok | {:invalid_header, String.t() | nil}

ingest(instance, req)

@spec ingest(atom(), request()) :: result()

Look the source up, refuse undeliverable header bytes, then ingest/4.

ingest(instance, source, tenant_id, req)

@spec ingest(atom(), Ankusa.Source.t(), String.t(), raw_request()) :: result()

Ingest one hook for a source lookup/2 returned. The caller has already run check_headers/1.

lookup(instance, route, binding \\ nil)

@spec lookup(atom(), Ankusa.Route.t(), Ankusa.Routes.Route.t() | nil) ::
  {:ok, Ankusa.Source.t(), tenant_id :: String.t()}
  | {:error, :unknown_source | :store_unavailable}

Resolve a route to its source and the tenant the hook is stored under.

The tenant names a storage partition and a gateway path segment, so it is checked here, before anything is written: a bad one from any RouteResolver is a 404, like an unknown source, not a dispatch failure after the provider was already acked. A miss is counted as a :unknown_source refusal.

A source owned by a tenant (any tenant_id but "default", the value a source gets when none is given) answers only for that tenant: a route that names a different one is the same 404 as a missing source, so the URL's tenant learns nothing about another tenant's sources. A "default" source is shared: it serves whatever tenant the route names (one stripe source behind /webhooks/:tenant/stripe for every customer).

A source store that cannot answer ({:error, :unavailable}) is {:error, :store_unavailable}, a 503 the provider retries, never a 404.

binding is the route the guard matched (Ankusa.Routes.Route; nil with routes off, and for ingest/2), held against what the resolver said. A disagreement is the same 404 as a missing source:

a route bound to a source_id admits only that source;
a route owned by a tenant (any tenant_id but "default") admits only that tenant: a resolver that names another tenant, or a source owned by another, is refused. Where the resolver names no tenant and the source is shared, the hook is stored under the route's tenant: the tenant a route resolves for a shared source;
a route owned by "default" binds no tenant, as before.

refused(instance, reason)

@spec refused(atom(), refusal()) :: :ok

Count a request answered without being ingested. Emits [:ankusa, :ingest, :refused] with the :reason; the one place a refusal is counted.