Add drain on shutdown option

[avenderossa]

No description provided.

[louisvisser]

Suggested change

	test "flushes a backlog that is not a multiple of the drain chunk size", %{queue: queue} do
	state = %Producer{queue: queue, check_interval: 10, drain_on_shutdown: true}
	backlog = Enum.to_list(1..2_501)
	:ok = Queue.push(queue, backlog)
	assert {:noreply, ^backlog, %Producer{demand: 0, timer: nil}} =
	Producer.prepare_for_draining(state)
	assert Queue.pop(queue, 1) == []
	end
	test "returns an empty list when enabled but the queue is already empty", %{queue: queue} do
	state = %Producer{queue: queue, check_interval: 10, drain_on_shutdown: true}
	assert {:noreply, [], %Producer{demand: 0, timer: nil}} =
	Producer.prepare_for_draining(state)
	end

[louisvisser]

In code, draining-default is infinity.

Also they can change the GenStage default at any time. Below comments revert logic so it leaves this value unset so the real GenStage default will be used, rather than apply a hardcoded 10000:

Suggested change

	* `:buffer_size` - GenStage producer buffer (default 100_000 when draining, else GenStage default)
	* `:buffer_size` - GenStage producer buffer (default `:infinity` when
	draining; otherwise unset, so GenStage's own default applies)

[louisvisser]

Suggested change

	{:producer, state, buffer_size: buffer_size(opts, drain?)}
	producer(state, opts, drain?)

[louisvisser]

Suggested change

	defp buffer_size(opts, true), do: Keyword.get(opts, :buffer_size, :infinity)
	defp buffer_size(opts, false), do: Keyword.get(opts, :buffer_size, 10_000)
	defp producer(state, opts, true) do
	{:producer, state, buffer_size: Keyword.get(opts, :buffer_size, :infinity)}
	end
	defp producer(state, opts, false) do
	case Keyword.fetch(opts, :buffer_size) do
	{:ok, size} -> {:producer, state, buffer_size: size}
	:error -> {:producer, state}
	end
	end

[louisvisser]

Nifty way of doing a much stronger assertion in a scenario like this (if the timer fails to cancel, we'd see the event within around 20ms - so refute_receive for 50ms is a strong test)

Suggested change

	timer = Process.send_after(self(), :dispatch_events, 60_000)
	state = %Producer{queue: queue, check_interval: 10, timer: timer, drain_on_shutdown: true}
	assert {:noreply, [], %Producer{timer: nil}} = Producer.prepare_for_draining(state)
	assert Process.cancel_timer(timer) == false
	timer = Process.send_after(self(), :dispatch_events, 20)
	state = %Producer{queue: queue, check_interval: 10, timer: timer, drain_on_shutdown: true}
	assert {:noreply, [], %Producer{timer: nil}} = Producer.prepare_for_draining(state)
	refute_receive :dispatch_events, 50

[louisvisser]

Worth considering:

Undersized :buffer_size silently drops drained events

On graceful shutdown prepare_for_draining/1 emits the whole queue as events → GenStage buffers them, capped by :buffer_size.

• Draining default :infinity → safe.
• Caller overrides :buffer_size to a finite N < queue depth at shutdown → GenStage drops the overflow, logs an error only → silent data loss — the exact failure the feature exists to prevent.

Doc says "size the buffer to exceed worst-case depth" but never warns that undersizing loses data. Fix = add that warning to the moduledoc.