Stale accumulator passed to collectable halt on exception during for ... into: ...

[lukaszsamson]

Elixir and Erlang/OTP versions

Erlang/OTP 28 [erts-16.2] [source] [64-bit] [smp:12:12] [ds:12:12:10] [async-threads:1] [jit]

Elixir 1.20.0-rc.4 (603602e) (compiled with Erlang/OTP 28)

Operating system

any

Current behavior

Consider this code:

defmodule LeakCollectable do
  defstruct []
end

defimpl Collectable, for: LeakCollectable do
  def into(_original) do
    collector = fn
      acc, {:cont, elem} ->
        new_acc = acc ++ [elem]
        Process.put(:trace, [{:cont, acc, elem, new_acc} | Process.get(:trace, [])])
        new_acc
      acc, :done ->
        Process.put(:done_acc, acc)
        acc
      acc, :halt ->
        Process.put(:halt_acc, acc)
        acc
    end
    {[:initial], collector}
  end
end

Process.put(:trace, [])
try do
  for x <- [1, 2, 3], into: %LeakCollectable{} do
    if x == 3, do: raise("boom")
    x
  end
rescue
  _ -> :ok
end

Result:

IO.inspect(Enum.reverse(Process.get(:trace)))
[
  {:cont, [:initial], 1, [:initial, 1]},
  {:cont, [:initial, 1], 2, [:initial, 1, 2]}
]
IO.inspect(Process.get(:halt_acc))
[:initial]

Note that final invocation was passed the initial accumulator possibly leading to resource leak or unaccounted for side effects

The bug is in

elixir/lib/elixir/src/elixir_erl_for.erl

Line 168 in b7a832e

build_into(Ann, Clauses, Expr, Into, Uniq, S) ->

The build_reduce passes Acc as the initial value, but internally creates a new variable for the evolving accumulator. Then in stacktrace_clause the original Acc is used

Expected behavior

Collectable halt invoked with final accumulator value

[lukaszsamson]

I noticed the same bug affects Enum

elixir/lib/elixir/lib/enum.ex

Line 1568 in 03b45e5

fun.(initial, :halt)

elixir/lib/elixir/lib/enum.ex

Line 1639 in 03b45e5

fun.(initial, :halt)

and Stream

https://github.com/elixir-lang/elixir/blob/main/lib/elixir/lib/stream.ex#L605

defmodule C do
  defstruct [:pid]
end

defimpl Collectable, for: C do
  def into(%C{pid: pid}) do
    {[], fn
      acc, {:cont, x} -> [x | acc]
      acc, :done -> send(pid, {:done, Enum.reverse(acc)}); :ok
      acc, :halt -> send(pid, {:halt, Enum.reverse(acc)}); :ok
    end}
  end
end

parent = self()
try do
  Enum.into([1, 2], struct(C, pid: parent), fn x -> if x == 2, do: raise("boom"); x end)
rescue
  _ -> :ok
end

receive do
  msg -> IO.inspect(msg)
end

{:halt, []}

stream = Stream.into([1, 2], struct(C, pid: parent), fn x -> if x == 2, do: raise("boom"); x end)
try do
  Enum.to_list(stream)
rescue
  _ -> :ok
end

receive do
  msg -> IO.inspect(msg)
end

{:halt, []}

[josevalim]

Yes, there are situations where it is not possible to recover the latest accumulator. We could push each implementation to use side-effects but it is sort of redundant. The only reason you would need side-effects is because you need to close a resource (which is by definition a side-effect), so it could track the side-effects directly. I will update the documentation instead.

[lukaszsamson]

@josevalim I updated #15266 - added fixes for enum and stream cases