Stream passed to `Oban.insert_all` is materialized twice?

Environment

Oban Version - 2.18.3 , oban_pro 1.4.13
PostgreSQL Version - 14.13
Elixir & Erlang/OTP Versions - Elixir 1.17.2, Erlang 27.0.1
ecto - 3.12.4
ecto_sql - 3.12.1
postgrex - 0.19.2

Current Behavior

Hi! When optimizing some issues we encountered with passing a big list of jobs to Oban.insert_all, we thought of using a Stream instead. Oban documentation has this example:

(fn -> MyApp.Worker.new(%{}))
|> Stream.repeatedly()
|> Stream.take(100)
|> Oban.insert_all()

Stream.take returns a lazy Stream, so it suggests that Oban should work nicely all streams here.

But when reading the oban_pro code I noticed a pattern that suggests the stream would get executed twice. Simplifying a bit, here is roughly the pattern I saw, with some inspect logging added:

Repo.transaction(fn ->
  s = Repo.stream(User)  |> Stream.map(fn x -> IO.inspect(x.id); x end)

  # Oban.insert_all(s) start

  Enum.reduce(s, 0, fn u, a -> a + 1 end)
  Enum.map(s, fn u -> u.id end)

  # Oban.insert_all(s) end
end)

Expected Behavior

Ideally Oban.insert_all wouldn't run the stream passed to it twice. If that's not possible, I would expect this to be documented as a warning.

And I'm sorry if I mis-read the code! 🙏🏻

oban-bg / oban