Raise in Stream.cycle when enumerable reduce call yields no elements

lib/elixir/lib/stream.ex

@@ -1437,7 +1437,7 @@ defmodule Stream do
         do_cycle(cycle, [], cycle, fun.(element, acc), fun)
 
       {_, []} ->
-        do_cycle(cycle, [], cycle, {:cont, acc}, fun)
+        do_cycle(check_cycle_subsequent_element(cycle), [], cycle, {:cont, acc}, fun)
     end
   end
 
@@ -1446,10 +1446,26 @@ defmodule Stream do
   end
 
   defp check_cycle_first_element(reduce) do
+    check_cycle_non_empty(
+      reduce,
+      ArgumentError,
+      "cannot cycle over an empty enumerable"
+    )
+  end
+
+  defp check_cycle_subsequent_element(reduce) do
+    check_cycle_non_empty(
+      reduce,
+      RuntimeError,
+      "cycled enumerable became empty after a previous iteration produced elements"
+    )
+  end
+
+  defp check_cycle_non_empty(reduce, exception, message) do
     fn acc ->
       case reduce.(acc) do
         {state, []} when state in [:done, :halted] and elem(acc, 0) != :halt ->
-          raise ArgumentError, "cannot cycle over an empty enumerable"
+          raise exception, message
 
         other ->
           other

lib/elixir/test/elixir/stream_test.exs

@@ -263,6 +263,10 @@ defmodule StreamTest do
       Stream.cycle(%{}) |> Enum.to_list()
     end
 
+    assert_raise ArgumentError, "cannot cycle over an empty enumerable", fn ->
+      Stream.cycle(%HaltAcc{acc: []}) |> Enum.to_list()
+    end
+
     assert Stream.cycle([1, 2, 3]) |> Stream.take(5) |> Enum.to_list() == [1, 2, 3, 1, 2]
     assert Enum.take(stream, 5) == [1, 2, 3, 1, 2]
   end
@@ -281,6 +285,28 @@ defmodule StreamTest do
              [1, 1, 1, 1, 1]
   end
 
+  test "cycle/1 raises when a subsequent reduce yields no elements" do
+    Process.put(:cycle_counter, 0)
+
+    stream =
+      Stream.resource(
+        fn ->
+          n = Process.get(:cycle_counter)
+          Process.put(:cycle_counter, n + 1)
+          n
+        end,
+        fn
+          0 -> {[:a], :done}
+          _ -> {:halt, :ok}
+        end,
+        fn _ -> :ok end
+      )
+
+    assert_raise RuntimeError,
+                 "cycled enumerable became empty after a previous iteration produced elements",
+                 fn -> Stream.cycle(stream) |> Enum.take(3) end
+  end
+
   test "dedup/1 is lazy" do
     assert lazy?(Stream.dedup([1, 2, 3]))
   end