Changelog

Copy Markdown View Source

All notable changes to ankusa_rabbitmq are documented here. Format follows Keep a Changelog; this project follows Semantic Versioning.

Unreleased

0.7.0 - 2026-10-28

Added

  • :max_message_bytes (default 128 MiB, RabbitMQ's max_message_size; DEL-5): the size preflight of Ankusa.Sink.Messaging. Requires an ankusa release with the max_message_bytes sink callback; publish core first.
  • Sink.RabbitMQ: a hook of an ordered: true sink carries the lane's key as the ankusa_ordering_key AMQP header (consumer-side ordering, ADR 0004).

Changed

0.6.0 - 2026-10-14

Added

  • :max_inflight (default 256): publishes awaiting a confirm per connection; past it a publish answers {:error, :busy} without being sent.
  • :idle_timeout_ms (default 600_000, 0 never, at least 16 s): a connection with no delivery for that long is closed (Ankusa.Sink.Reaper, requires core's) and the next delivery opens a new one, so a deleted or re-pointed source no longer leaves its connection open for the life of the node.
  • Every publish carries the ankusa_idempotency_key AMQP header: the same tenant-scoped key as the message's idempotency_key (Ankusa.Envelope.idempotency_key/1), always present. Requires ankusa 0.5 or newer: the key comes from Ankusa.Envelope.idempotency_key/1, which 0.4.x does not have, so the Hex requirement is now ~> 0.5.
  • AMQP message_id is the hook id; the ankusa_dedupe_key and ankusa_replay_id AMQP headers are set when present.

Changed

  • Requires ankusa ~> 0.6 (Ankusa.Sink.Reaper); publish core first.
  • Stopping a connection process closes its AMQP connection: amqp_client owns that connection, so it used to outlive the process. Publishes waiting on it answer {:error, :not_connected}.
  • Confirms are asynchronous: many publishes are outstanding on the channel at once and each caller is answered on its own confirm, so one slow confirm no longer serializes every delivery on the connection.
  • A publish still queued in the connection when its caller's deadline passed answers {:error, :expired} and is never sent.
  • The connection dials after it starts (handle_continue), so a dead broker never blocks the supervisor; it is registered under a digest of the URL, and its format_status/1 redacts the URL, so a password never appears in a process name, :sys.get_state/1 or a crash report.
  • The connection's child spec carries only the options the connection reads, with the URL inside a closure, so :sys.get_status/1 of Ankusa.Sink.RabbitMQ.Supervisor, or a supervisor report, never prints the password.
  • Breaking: Ankusa.Sink.RabbitMQ.Connection.publish/4's fourth argument is a keyword list of properties (message_id:, headers:), not a header list.
  • Breaking: every publish is mandatory. A publish no queue is bound to receive is no longer reported as delivered: deliver/3 answers {:error, {:unroutable, routing_key}} instead of :ok, so the hook is retried and then dead-lettered (a 503 under wal.type: none) rather than settled while the broker kept nothing. There is no option to turn this off.

Fixed

  • amqp_client's processes hold the connection password encrypted: the application sets a random credentials_obfuscation secret at start unless the host already set one, so :sys.get_status/1 or a crash report of an amqp_client process no longer prints {:plaintext, password}.
  • A channel the broker closes (a 404 publishing to a deleted exchange, a 406) is reopened at once on the same connection, re-declaring the exchange. The publish it interrupted answers {:error, {:channel_closed, reason}}. Previously every later publish failed with {:publish_failed, _} until the node restarted.
  • :confirm_timeout_ms is milliseconds. The confirm wait read it as seconds, so an unconfirmed publish blocked the connection for up to 70 s while its caller gave up at 15 s. Each publish now waits for its own confirm only, so one stale publish no longer delays later ones.
  • A channel that fails setup (for example a 406 on exchange.declare) no longer leaks an AMQP connection on every retry; the connection is kept and only the channel is retried.

0.4.0 - 2026-10-02

Added

Changed

  • The Hex requirement on ankusa is ~> 0.4 (was ~> 0.2): describe/2 needs Ankusa.Sink.Description, which core has from 0.4.0.

Removed

  • The sink no longer implements the removed Ankusa.Sink.ordering_key/2 callback: deliveries are unordered. The routing key still decides the queue, so RabbitMQ's per-queue ordering is unchanged. Requires the matching ankusa core.

0.3.0 - 2026-10-01

No code changes. Released as 0.3.0 in step with ankusa 0.3.0 and the other packages in this release.

0.2.1 - 2026-09-28

Added

  • A sha256 field on a claim-checked message: the lowercase hex digest of the payload, for the reader to verify the bytes it redeems against. It used to live inside the claim ref.

Changed

  • The message claim field is urn:ankusa:claim:v1:<tenant>:<claim_id>: a tenant and a canonical (uppercase) ULID, where it used to carry the pack id, byte offset/length, and digest. The message v stays 1. Both come from Ankusa.Sink.Message in ankusa 0.2.1.

0.2.0 - 2026-09-27

Added

  • Ankusa.Sink.ordering_key/2, implemented by this sink as the routing key: RabbitMQ orders per queue, and the routing key decides the queue. Dispatch now serializes deliveries sharing a key and runs different keys concurrently.

Changed

0.1.0 - 2026-09-23

Added

  • Ankusa.Sink.RabbitMQ: publishes delivered hooks to a RabbitMQ exchange (never a queue: consumers own their own queue/binding). Bodies under :inline_max_bytes ride along base64-encoded; larger bodies are checked in through Ankusa.ClaimCheck, with the message carrying a ticket. Publisher-confirmed delivery; supervised, auto-reconnecting connection per (instance, exchange).
  • Messages carry a format version, "v": 1. Additive: consumers that ignore unknown keys need no change.

Changed

  • The message is built by Ankusa.Sink.Message (in ankusa), shared with ankusa_kafka, so both sinks publish byte-identical messages.