Repository: livebook-dev/kino
Pull request #263 (2023-03-08 12:59:22)

Ignore failures in listen and animate
It's important for apps that a failure in event callback is isolated and doesn't crash the stream.


Patch:
diff --git a/lib/kino.ex b/lib/kino.ex
index 5c3ae62..2c3c648 100644
--- a/lib/kino.ex
+++ b/lib/kino.ex
@@ -142,6 +142,8 @@ defmodule Kino do
 
   import Kernel, except: [inspect: 1]
 
+  require Logger
+
   @type nothing :: :"do not show this result in output"
 
   @doc """
@@ -281,13 +283,16 @@ defmodule Kino do
     frame = Kino.Frame.new()
 
     Kino.Frame.periodically(frame, interval_ms, {0, state}, fn {i, state} ->
-      case fun.(i, state) do
-        {:cont, term, state} ->
+      case safe_apply(fun, [i, state], "Kino.animate") do
+        {:ok, {:cont, term, state}} ->
           Kino.Frame.render(frame, term)
           {:cont, {i + 1, state}}
 
-        :halt ->
+        {:ok, :halt} ->
           :halt
+
+        {:error, _, _} ->
+          {:cont, {i + 1, state}}
       end
     end)
 
@@ -300,13 +305,16 @@ defmodule Kino do
     frame = Kino.Frame.new() |> Kino.render()
 
     listen(stream, state, fn item, state ->
-      case fun.(item, state) do
-        {:cont, term, state} ->
+      case safe_apply(fun, [item, state], "Kino.animate") do
+        {:ok, {:cont, term, state}} ->
           Kino.Frame.render(frame, term)
           {:cont, state}
 
-        :halt ->
+        {:ok, :halt} ->
           :halt
+
+        {:error, _, _} ->
+          {:cont, state}
       end
     end)
 
@@ -355,7 +363,7 @@ defmodule Kino do
   end
 
   def listen(stream, fun) when is_function(fun, 1) do
-    async(fn -> Enum.each(stream, fun) end)
+    async(fn -> Enum.each(stream, &safe_apply(fun, [&1], "Kino.listen")) end)
   end
 
   @doc ~S"""
@@ -393,16 +401,38 @@ defmodule Kino do
   def listen(stream, state, fun) when is_function(fun, 2) do
     async(fn ->
       Enum.reduce_while(stream, state, fn item, state ->
-        case fun.(item, state) do
-          {:cont, state} -> {:cont, state}
-          :halt -> {:halt, state}
+        case safe_apply(fun, [item, state], "Kino.listen") do
+          {:ok, {:cont, state}} -> {:cont, state}
+          {:ok, :halt} -> {:halt, state}
+          {:error, _, _} -> {:cont, state}
         end
       end)
     end)
   end
 
+  defp safe_apply(fun, args, context) do
+    try do
+      {:ok, apply(fun, args)}
+    catch
+      kind, error ->
+        Logger.error(
+          "#{context} with #{Kernel.inspect(fun)} failed with reason:\n\n" <>
+            Exception.format(kind, error, __STACKTRACE__)
+        )
+
+        {:error, kind, error}
+    end
+  end
+
   defp async(fun) do
-    Kino.start_child({Task, fun})
+    {:ok, pid} = Kino.start_child({Task, fun})
+    # In case the stream breaks we want to propagate the failure
+    try do
+      Process.link(pid)
+    catch
+      :error, :noproc -> :ok
+    end
+
     :ok
   end
 


Test patch:
diff --git a/test/kino_test.exs b/test/kino_test.exs
index 83d758d..79e5962 100644
--- a/test/kino_test.exs
+++ b/test/kino_test.exs
@@ -1,6 +1,8 @@
 defmodule KinoTest do
   use Kino.LivebookCase, async: true
 
+  import ExUnit.CaptureLog
+
   describe "inspect/2" do
     test "sends a text output to the group leader" do
       gl =
@@ -29,6 +31,24 @@ defmodule KinoTest do
       assert_output({:frame, [{:text, "\e[34m0\e[0m"}], %{ref: ^ref, type: :replace}})
       assert_output({:frame, [{:text, "\e[34m1\e[0m"}], %{ref: ^ref, type: :replace}})
     end
+
+    test "ignores failures" do
+      log =
+        capture_log(fn ->
+          Stream.iterate(0, &(&1 + 1))
+          |> Stream.take(2)
+          |> Kino.animate(fn i ->
+            1 = i
+            i
+          end)
+
+          assert_output({:frame, [], %{ref: ref, type: :default}})
+          assert_output({:frame, [{:text, "\e[34m1\e[0m"}], %{ref: ^ref, type: :replace}})
+        end)
+
+      assert log =~ "Kino.animate"
+      assert log =~ "(MatchError) no match of right hand side value: 0"
+    end
   end
 
   describe "animate/3" do
@@ -36,12 +56,31 @@ defmodule KinoTest do
       Stream.iterate(0, &(&1 + 1))
       |> Stream.take(2)
       |> Kino.animate(0, fn i, state ->
-        {:cont, i + state, state + 1}
+        {:cont, i + state, i + state}
       end)
 
       assert_output({:frame, [], %{ref: ref, type: :default}})
       assert_output({:frame, [{:text, "\e[34m0\e[0m"}], %{ref: ^ref, type: :replace}})
-      assert_output({:frame, [{:text, "\e[34m2\e[0m"}], %{ref: ^ref, type: :replace}})
+      assert_output({:frame, [{:text, "\e[34m1\e[0m"}], %{ref: ^ref, type: :replace}})
+    end
+
+    test "ignores failures" do
+      log =
+        capture_log(fn ->
+          Stream.iterate(0, &(&1 + 1))
+          |> Stream.take(4)
+          |> Kino.animate(0, fn i, state ->
+            true = i in [1, 3]
+            {:cont, i + state, i + state}
+          end)
+
+          assert_output({:frame, [], %{ref: ref, type: :default}})
+          assert_output({:frame, [{:text, "\e[34m1\e[0m"}], %{ref: ^ref, type: :replace}})
+          assert_output({:frame, [{:text, "\e[34m4\e[0m"}], %{ref: ^ref, type: :replace}})
+        end)
+
+      assert log =~ "Kino.animate"
+      assert log =~ "** (MatchError) no match of right hand side value: false"
     end
   end
 
@@ -78,6 +117,26 @@ defmodule KinoTest do
       assert_receive ^info
       assert_receive ^info
     end
+
+    test "ignores failures" do
+      log =
+        capture_log(fn ->
+          myself = self()
+
+          Stream.iterate(0, &(&1 + 1))
+          |> Stream.take(2)
+          |> Kino.listen(fn i ->
+            1 = i
+
+            send(myself, {:item, i})
+          end)
+
+          assert_receive {:item, 1}
+        end)
+
+      assert log =~ "Kino.listen"
+      assert log =~ "** (MatchError) no match of right hand side value: 0"
+    end
   end
 
   describe "listen/3" do
@@ -88,11 +147,11 @@ defmodule KinoTest do
       |> Stream.take(2)
       |> Kino.listen(0, fn i, state ->
         send(myself, {:item, i + state})
-        {:cont, state + 1}
+        {:cont, i + state}
       end)
 
       assert_receive {:item, 0}
-      assert_receive {:item, 2}
+      assert_receive {:item, 1}
     end
 
     test "with control events" do
@@ -116,6 +175,27 @@ defmodule KinoTest do
       assert_receive {:counter, 1}
       assert_receive {:counter, 2}
     end
+
+    test "ignores failures" do
+      log =
+        capture_log(fn ->
+          myself = self()
+
+          Stream.iterate(0, &(&1 + 1))
+          |> Stream.take(4)
+          |> Kino.listen(0, fn i, state ->
+            true = i in [1, 3]
+            send(myself, {:item, i + state})
+            {:cont, i + state}
+          end)
+
+          assert_receive {:item, 1}
+          assert_receive {:item, 4}
+        end)
+
+      assert log =~ "Kino.listen"
+      assert log =~ "** (MatchError) no match of right hand side value: false"
+    end
   end
 
   defp await_process(pid) do