Ankusa.Dedupe (ankusa v0.7.0)

Copy Markdown View Source

Per-source provider event keys.

Providers retry a webhook until they see a 2xx, so a retry can arrive on any node of a fleet. Every provider stamps its deliveries with an event id — a delivery header (GitHub's x-github-delivery, Svix's svix-id) or a JSON field in the body (Stripe's "id"). key/2 extracts that stamp from an envelope, and the queue writer collapses envelopes that share one within the TTL window.

The TTL bounds the memory a provider retry storm can occupy: after it expires, the same event commits again (and consumers dedupe on the idempotency_key the message carries). 259200000 ms (72 h) covers Stripe's 3-day and Shopify's 48-hour retry schedules.

The window is per tenant and source: the same key under two tenants (one tenant_path source serves many) is two events.

Sources

  • {:header, name} — the value of a request header (case-insensitive).
  • {:json, ["a", "b"]} — walk a → b through the decoded JSON body; only map keys are followed. A string value is the key; an integer value becomes its decimal string. Anything else yields nil. A body past dedupe.max_json_bytes (1 MiB by default) is not decoded: no key.

Salesforce

The :salesforce preset is {:json, ["id"]}, and that is a convention, not a Salesforce-documented delivery id. Salesforce stamps no delivery-id header on any outbound mechanism. Outbound Messages are SOAP/XML (their per-notification Notification/Id is not reachable by a JSON path), and Flow/Apex HTTP callouts and Data 360 webhook targets send a body the org builds. The preset therefore reads the top-level "id" of a JSON body, which the callout has to set to a value that is the same on every retry of one event (the record id plus a change stamp, a platform event's EventUuid). A body with no "id" is never deduplicated: pick header: or json: for a body shaped otherwise.

Summary

Types

t()

Functions

key(dedupe, env, decoded \\ :skip, max_json_bytes \\ 1_048_576)

Extract the provider event key from an envelope, or nil when the stamp is missing or the body does not decode.

max_json_bytes(instance)

The dedupe.max_json_bytes of instance: what key/4 is handed. The default (1048576) where no instance runs.

new!(preset)

Build a dedupe spec from config. Accepts

undeduped_sources(config)

The ids, sorted, of the sources config.source_store declares (its :sources option, as Ankusa.SourceStore.Static reads it) that set no dedupe.

warn_undeduped(config)

Log one boot warning per source undeduped_sources/1 names. Dedupe is opt-in, and a source without it is the one setting where a provider's retry is indistinguishable from a new event, so boot says so.

Types

t()

@type t() :: %Ankusa.Dedupe{
  from: {:header, String.t()} | {:json, [String.t()]},
  ttl_ms: pos_integer()
}

Functions

key(dedupe, env, decoded \\ :skip, max_json_bytes \\ 1_048_576)

@spec key(
  t() | nil,
  Ankusa.Envelope.t(),
  {:ok, term()} | {:error, term()} | :skip,
  pos_integer()
) ::
  String.t() | nil

Extract the provider event key from an envelope, or nil when the stamp is missing or the body does not decode.

A key longer than 200 bytes is replaced by its lowercase hex SHA-256, prefixed "sha256:".

decoded is the body's JSON when the caller has decoded it already ({:ok, term} or {:error, reason}: Ankusa.Ordering shares one decode between an ordering key and a dedupe: json); :skip decodes it here.

A json source reads no body of more than max_json_bytes bytes (dedupe.max_json_bytes, 1048576 by default; see max_json_bytes/1): it answers nil before any decode, so a body of megabytes is never decoded on the request process, and the hook is stored with no key. A header source is not affected.

max_json_bytes(instance)

@spec max_json_bytes(atom()) :: pos_integer()

The dedupe.max_json_bytes of instance: what key/4 is handed. The default (1048576) where no instance runs.

new!(preset)

@spec new!(nil | atom() | keyword() | map()) :: t() | nil

Build a dedupe spec from config. Accepts:

  • nil — no dedupe.
  • a preset atom (github, standard_webhooks, svix, shopify, stripe, salesforce).
  • a keyword list / map with exactly one of :preset (atom), :header (string, stored lowercased) or :json (a dot path: every segment non-empty), plus an optional :ttl_ms (positive integer).

Anything else raises ArgumentError with a "dedupe: " message.

undeduped_sources(config)

@spec undeduped_sources(Ankusa.Config.t()) :: [String.t()]

The ids, sorted, of the sources config.source_store declares (its :sources option, as Ankusa.SourceStore.Static reads it) that set no dedupe.

Such a source stores a provider retry as a new hook with a new idempotency_key. A walk over the declared specs, no store read: the gauge ankusa_sources_without_dedupe reports its length. Sources a writable store holds are not counted: they are not known at boot.

warn_undeduped(config)

@spec warn_undeduped(Ankusa.Config.t()) :: :ok

Log one boot warning per source undeduped_sources/1 names. Dedupe is opt-in, and a source without it is the one setting where a provider's retry is indistinguishable from a new event, so boot says so.