# `Ankusa.Sink.Reaper`
[🔗](https://github.com/jamescarr/ankusa/blob/ankusa-v0.7.0/packages/ankusa/lib/ankusa/sink/reaper.ex#L1)

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`.

# `idle_ms`

```elixir
@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`

# `touch`

```elixir
@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.

---

*Consult [api-reference.md](api-reference.md) for complete listing*