# `Ankusa.Sink.SQS` [🔗](https://github.com/jamescarr/ankusa/blob/ankusa-v0.7.0/packages/ankusa/lib/ankusa/sink/sqs.ex#L1) Sends delivered hooks to an **Amazon SQS** queue, standard or FIFO (or to anything that speaks SQS's JSON protocol, such as the `floci` emulator in `docker-compose.integration.yml`). The message body is `Ankusa.Sink.Message`, byte-identical to what `Ankusa.Sink.Kafka` and `Ankusa.Sink.NATS` publish: inline up to `:inline_max_bytes`, a claim ticket above it. Dispatch's deliveries are **batched** (see "Batching"): up to 10 of them to one queue go out as one SigV4-signed `SendMessageBatch` `POST`. Anything else (direct publishing, `batch: false`) is one `SendMessage` per delivery. Both go through `Ankusa.HttpClient` (Req's connection pool, no hidden retries or redirects). The sink needs nothing beyond core's `req` and `aws_signature`; its batchers are started by `Ankusa.Instance`. Credentials come from `Ankusa.AWS.Credentials`: static opts, then the `AWS_*` environment, web identity (IRSA), then the EC2 instance role (IMDSv2). This sink **never creates a queue**. A queue's type, retention, visibility timeout, redrive policy and encryption are an operator's contract, not something the first hook to arrive should decide. A queue that does not exist is `{:error, {:permanent, {:config, {:sqs, 400, "QueueDoesNotExist", message}}}}`: dead-lettered at once, since only an operator creating the queue fixes it, and a DLQ replay then re-sends the hooks. (Compare `Ankusa.Sink.Kafka` and `Ankusa.Sink.NATS`, which likewise never create their topic or stream.) ## Publishing is confirmed `deliver/3` returns `:ok` only on a `200` from `SendMessage` (or, batched, an entry of a `SendMessageBatch` reply under `Successful`) whose `MD5OfMessageBody` matches the MD5 of the body sent. SQS answers `200` once it has stored the message redundantly, so the sink keeps the default `c:Ankusa.Sink.durable?/1` of `true`. Everything else is an error (see "Error classes" in `Ankusa.Sink`): * `{:permanent, {:message_too_large, size, max}}` — body plus message attributes over `:max_message_bytes`, refused before any request. * `{:permanent, {:sqs, status, code, message}}` — SQS refused the message itself: `InvalidMessageContents` (characters outside SQS's allowed set), `InvalidParameterValue`, `MissingParameter`, `InvalidAttributeName` or `InvalidAttributeValue`. Dead-lettered at once. * `{:permanent, {:config, {:sqs, status, code, message}}}` — SQS refused this sink's configuration or credentials: `AccessDenied`, `QueueDoesNotExist`, `InvalidClientTokenId`, `UnrecognizedClientException`, `SignatureDoesNotMatch`, `InvalidAccessKeyId`, `MissingAuthenticationToken`. Dead-lettered at once and counted in `ankusa_dispatch_config_errors_total{sink}`. * `{:retry_after, ms, {:sqs, status, code, message}}` — SQS throttled the send (`Throttling`, `ThrottlingException`, `RequestThrottled`, `TooManyRequestsException`, `ServiceUnavailable`, `KmsThrottled`, or any HTTP `429` or `503`). `ms` is a full-jitter draw (`Ankusa.Sink.throttle_after_ms/1`). * `{:sqs, status, code, message}` — any other SQS error (a 5xx, an expired token): transient, the retry policy. * `{:status, status, body}` — a non-2xx response that is not an SQS error document (a proxy, a load balancer); a `429` or `503` is a `{:retry_after, ms, {:status, status, body}}`. * `{:md5_mismatch, expected, got}`, `{:unexpected_reply, body}`, `:no_credentials` and transport errors — retried. ## FIFO queues A `:queue_url` ending in `.fifo` is a FIFO queue (AWS requires the suffix on FIFO queue names). Each message then carries: * `MessageGroupId` — `:message_group_id`, default `"tenant_id/source_id"` (`"/source_id"` without a tenant), the scope `Ankusa.Sink.Kafka` uses for its default key: one group's messages are received in the order they were sent. * `MessageDeduplicationId` — the hook's `id`, or `id:replay:` for a replay (the rule `Ankusa.Sink.NATS` uses for `Nats-Msg-Id`), so SQS collapses a retry after a lost response, while a deliberate replay of the same hook is always stored. SQS's deduplication window is **5 minutes**. A retry later than that (a long outage, a slow backoff) can store a second copy, so delivery is at-least-once; consumers dedupe on the message's `idempotency_key`. On a standard queue `MessageGroupId` is sent only when `:message_group_id` is set (fair queues), and `MessageDeduplicationId` never is. ## Batching A delivery from dispatch waits in its queue's `Ankusa.Sink.Batcher` (one per queue and set of credentials) and is sent in a `SendMessageBatch` with up to 9 others, 20 ms after the first arrived or as soon as the batch is full (10 entries, 256 KiB). `deliver/3` returns when the batch's response is known: the hook's row stays claimed until then, so a batcher that dies loses nothing and the attempt is retried. The response is read entry by entry: * `Successful` with the MD5 of the body sent: `:ok`. A different MD5 is `{:md5_mismatch, expected, got}`. * `Failed`: classified on its own, as above (`classify_error/4`), so a batch with one refused message dead-letters exactly that message. A throttled entry is `{:retry_after, ms, reason}`, `ms` full jitter. * a `Failed` or a missing entry never fails its neighbours; a request that fails as a whole (a `5xx`, a timeout, a `429`, `QueueDoesNotExist`) fails every entry the way a single `SendMessage` would, and each retries on its own schedule. Entry `Id`s are positions in the request (`"0"` to `"9"`), not hook ids; the `MessageDeduplicationId` is unchanged. A batch may hold hooks of several tenants and sources; the batcher takes entries round-robin across tenants, and refuses, with `{:retry_after, ms, {:batch_queue_full, reason}}`, a delivery that would pass its queue's ceilings (see `Ankusa.Sink.Batcher`). On a FIFO queue a `MessageGroupId` has one request in flight at a time, and its entries leave in the order they were admitted; SQS enqueues a batch's messages in the order they are sent. With the default group (`tenant/source`) that means a source's messages to one FIFO queue go out as fast as SQS answers, up to 10 to a request. A message that fails and is retried is a later send, as it always was. ## What each message carries * **body** — the `Ankusa.Sink.Message` JSON. * **message attributes** (all `String`) — `ankusa_id`, `ankusa_source_id`, `ankusa_tenant_id`, `ankusa_message_version`, `ankusa_idempotency_key`, `content_type` (`application/json`) — the set `Ankusa.Sink.Kafka` sets as headers — plus `ankusa_dedupe_key` and `ankusa_replay_id` when the hook carries them. `ankusa_tenant_id` is left out when the hook has no tenant, because SQS refuses an empty attribute value. At most 8 of SQS's 10. ## opts * `:queue_url` — required, e.g. `"https://sqs.us-east-1.amazonaws.com/123456789012/hooks.fifo"` * `:region` — required, e.g. `"us-east-1"` * `:endpoint` — default: the queue URL's origin (`scheme://host[:port]`). Set it for floci or a VPC endpoint. * `:message_group_id` — a string, or `(Envelope.t() -> String.t())`; see "FIFO queues" * `:inline_max_bytes` — default 64 KiB (65,536), configurable * `:max_message_bytes` — the largest body plus attributes sent; default `1_048_576`, SQS's limit (1 MiB). Lower it to a queue's own `MaximumMessageSize`. * `:timeout_ms` — default `5_000`, for both connect and response * `:access_key_id`, `:secret_access_key`, `:session_token` — static credentials. Without them the chain in `Ankusa.AWS.Credentials` runs. * `:req_options` — transport options for the HTTP client (a Finch pool, a proxy via `:connect_options`, `plug:` for `Req.Test`); see `Ankusa.HttpClient`. * `:batch` — default `true`; `false` sends every delivery in its own `SendMessage` and counts it against `dispatch.sink_concurrency` as before * `:batch_linger_ms` — default 20 (1 to 60,000), how long a batch waits to fill * `:batch_max_entries` — default and maximum 10 * `:batch_max_bytes` — default 262,144, at most 1,048,576 (SQS's limit for a batch's messages together) * `:batch_queue_max`, `:batch_queue_max_bytes`, `:batch_queue_tenant_max` — the queue's ceilings, default 4,096 entries, 128 MiB and 1,024 entries a tenant * `:batch_max_inflight` — requests in flight per queue; default the instance's `dispatch.sink_concurrency` (8) # `classify_error` ```elixir @spec classify_error(integer(), String.t(), String.t() | nil, pos_integer()) :: {:error, term()} ``` The `{:error, reason}` for an SQS error `code`, whether it answered a whole request (`status` is the HTTP status) or one entry of a batch (`400` for a sender's fault, `500` otherwise). * a refused message (`InvalidMessageContents`, `InvalidParameterValue`, …): `{:permanent, {:sqs, status, code, message}}` * a refused credential or queue (`AccessDenied`, `QueueDoesNotExist`, …): `{:permanent, {:config, {:sqs, status, code, message}}}` * throttling (`ThrottlingException`, `KmsThrottled`, … or a `429`/`503`): `{:retry_after, ms, reason}`, `ms` full jitter for `attempt` (`Ankusa.Sink.throttle_after_ms/1`) * anything else: `{:sqs, status, code, message}`, transient # `validate_opts!` ```elixir @spec validate_opts!(keyword()) :: :ok ``` Raise `ArgumentError` naming the key unless `opts`' `:batch` is a boolean and its `batch_*` options are in range (see "opts"). The image's config loader calls it, so a bad value fails at load instead of at the first delivery. --- *Consult [api-reference.md](api-reference.md) for complete listing*