Ankusa.Sink.Reaper (ankusa v0.7.0)

Copy Markdown View Source

Closes the connections a sink adapter opened and then stopped using.

The broker adapters (ankusa_rabbitmq, ankusa_kafka, ankusa_nats, ankusa_redis) start one connection process per broker on the first delivery, under their own DynamicSupervisor. Without this, a source that is deleted, or re-pointed at another broker, leaves its connection open for the life of the node. Each adapter's Ankusa.Sink.Connections tree runs one reaper next to its supervisor, and Ankusa.Sink.Connections.ensure/5 calls touch/4 before every publish; a connection not touched for its idle_ms is stopped through that supervisor (DynamicSupervisor.terminate_child/2, so a :permanent child is removed, not restarted), and the next delivery that needs it starts a new one.

touch/4 is one :ets.insert/2 into a public table, so the publish path never waits on this process. A sweep runs every :tick_ms (default 30 s); a row whose connection already died is dropped without a word.

The race it accepts

A delivery that looked its connection up just before the sweep stopped it fails that attempt (the call exits, which Ankusa.Sink turns into {:error, _}), and the retry policy's next attempt starts a fresh connection. A connection touched while the sweep looks at it is kept: the sweep deletes only the exact row it judged idle. Because the touch comes before the publish, a publish in flight is never older than idle_ms as long as idle_ms is longer than the publish can take; idle_ms/2 enforces that floor for each adapter.

start_link/1 opts: :name (the reaper's registered name and its table's), :supervisor (the DynamicSupervisor whose children it stops), :tick_ms.

Summary

Functions

An adapter's idle timeout from its sink opts: :idle_timeout_ms (default 10 minutes), 0 for never, else at least floor_ms — the longest one publish can take, plus a margin — so a sweep never stops a connection under a publish. Raises ArgumentError on anything but a non-negative integer.

Record that the connection pid, known as key, is about to be used, and that it may be stopped once unused for idle_ms. idle_ms of 0 means never: the row is dropped. key is printed when the connection is closed, so it must not carry a credential.

Functions

idle_ms(opts, floor_ms)

@spec idle_ms(keyword(), pos_integer()) :: non_neg_integer()

An adapter's idle timeout from its sink opts: :idle_timeout_ms (default 10 minutes), 0 for never, else at least floor_ms — the longest one publish can take, plus a margin — so a sweep never stops a connection under a publish. Raises ArgumentError on anything but a non-negative integer.

start_link(opts)

touch(name, key, pid, idle_ms)

@spec touch(atom(), term(), pid(), non_neg_integer()) :: :ok

Record that the connection pid, known as key, is about to be used, and that it may be stopped once unused for idle_ms. idle_ms of 0 means never: the row is dropped. key is printed when the connection is closed, so it must not carry a credential.