Repository: plataformatec/broadway
Pull request #70 (2019-04-12 07:51:19)

Trap exits in processors and consumers


Patch:
diff --git a/lib/broadway.ex b/lib/broadway.ex
index f1c0af9..0747d14 100644
--- a/lib/broadway.ex
+++ b/lib/broadway.ex
@@ -346,7 +346,8 @@ defmodule Broadway do
 
   In case of errors in this callback, the error will be logged and that particular
   message will be immediately acknowledged as failed, not proceeding to the next
-  steps of the pipeline.
+  steps of the pipeline. This callback also traps exits, so failures due to broken
+  links between processes do not automatically cascade.
   """
   @callback handle_message(processor :: atom, message :: Message.t(), context :: term) ::
               Message.t()
@@ -367,7 +368,8 @@ defmodule Broadway do
   explicitly failed will be considered successful and automatically acknowledged.
 
   In case of errors in this callback, the error will be logged and the whole
-  batch will be failed.
+  batch will be failed. This callback also traps exits, so failures due to broken
+  links between processes do not automatically cascade.
   """
   @callback handle_batch(
               batcher :: atom,
diff --git a/lib/broadway/consumer.ex b/lib/broadway/consumer.ex
index 6f1b8ac..8e0f05a 100644
--- a/lib/broadway/consumer.ex
+++ b/lib/broadway/consumer.ex
@@ -15,6 +15,8 @@ defmodule Broadway.Consumer do
 
   @impl true
   def init(args) do
+    Process.flag(:trap_exit, true)
+
     state = %{
       module: args[:module],
       context: args[:context]
diff --git a/lib/broadway/processor.ex b/lib/broadway/processor.ex
index a1f0c1a..1de3abc 100644
--- a/lib/broadway/processor.ex
+++ b/lib/broadway/processor.ex
@@ -13,6 +13,7 @@ defmodule Broadway.Processor do
 
   @impl true
   def init(args) do
+    Process.flag(:trap_exit, true)
     processor_config = args[:processor_config]
 
     state = %{
diff --git a/lib/broadway/server.ex b/lib/broadway/server.ex
index e0cf8cb..ec6cb7d 100644
--- a/lib/broadway/server.ex
+++ b/lib/broadway/server.ex
@@ -98,7 +98,8 @@ defmodule Broadway.Server do
   defp build_producers_specs(config) do
     %{
       name: broadway_name,
-      producers_config: producers_config
+      producers_config: producers_config,
+      shutdown: shutdown
     } = config
 
     [{key, producer_config} | other_producers] = producers_config
@@ -120,7 +121,8 @@ defmodule Broadway.Server do
 
         %{
           start: {Producer, :start_link, [producer_config, opts]},
-          id: name
+          id: name,
+          shutdown: shutdown
         }
       end
 
@@ -135,7 +137,8 @@ defmodule Broadway.Server do
       context: context,
       batchers_config: batchers_config,
       resubscribe_interval: resubscribe_interval,
-      terminator: terminator
+      terminator: terminator,
+      shutdown: shutdown
     } = config
 
     [{key, processor_config} | other_processors] = processors_config
@@ -185,7 +188,8 @@ defmodule Broadway.Server do
 
         %{
           start: {Processor, :start_link, [args, opts]},
-          id: name
+          id: name,
+          shutdown: shutdown
         }
       end
 
@@ -227,7 +231,7 @@ defmodule Broadway.Server do
   end
 
   defp build_batcher_spec(config, batcher_config, processors) do
-    %{terminator: terminator} = config
+    %{terminator: terminator, shutdown: shutdown} = config
     {key, options} = batcher_config
     name = process_name(config.name, "Batcher", key)
 
@@ -244,7 +248,8 @@ defmodule Broadway.Server do
 
     spec = %{
       start: {Batcher, :start_link, [args, opts]},
-      id: name
+      id: name,
+      shutdown: shutdown
     }
 
     {name, spec}
@@ -255,7 +260,8 @@ defmodule Broadway.Server do
       name: broadway_name,
       module: module,
       context: context,
-      terminator: terminator
+      terminator: terminator,
+      shutdown: shutdown
     } = config
 
     {key, options} = batcher_config
@@ -281,7 +287,8 @@ defmodule Broadway.Server do
 
         %{
           start: {Consumer, :start_link, [args, opts]},
-          id: name
+          id: name,
+          shutdown: shutdown
         }
       end
 


Test patch:
diff --git a/test/broadway_test.exs b/test/broadway_test.exs
index bfd5c3c..ab22bef 100644
--- a/test/broadway_test.exs
+++ b/test/broadway_test.exs
@@ -319,6 +319,7 @@ defmodule BroadwayTest do
           :raise -> raise "Error raised"
           :bad_return -> :oops
           :bad_batcher -> %{message | batcher: :unknown}
+          :broken_link -> Task.async(fn -> raise "oops" end) |> Task.await()
           _ -> message
         end
       end
@@ -427,6 +428,18 @@ defmodule BroadwayTest do
 
       refute_received {:EXIT, _, ^processor}
     end
+
+    test "processors do not crash on broken link", %{broadway: broadway, processor: processor} do
+      assert capture_log(fn ->
+               ref = Broadway.test_messages(broadway, [1, :broken_link, 4])
+
+               assert_receive {:ack, ^ref, [], [%{status: {:failed, "due to an unhandled exit"}}]}
+
+               assert_receive {:ack, ^ref, [%{data: 1}, %{data: 4}], []}
+             end) =~ "** (RuntimeError) oops"
+
+      refute_received {:EXIT, _, ^processor}
+    end
   end
 
   describe "pipeline without batchers" do
@@ -956,6 +969,7 @@ defmodule BroadwayTest do
             :fail -> Message.failed(message, "Failed message")
             :raise -> raise "Error raised"
             :bad_return -> :oops
+            :broken_link -> Task.async(fn -> raise "oops" end) |> Task.await()
             _ -> message
           end
         end)
@@ -1047,6 +1061,26 @@ defmodule BroadwayTest do
 
       refute_received {:EXIT, _, ^consumer}
     end
+
+    test "consumers do not crash on broken link",
+         %{broadway: broadway, consumer: consumer} do
+      assert capture_log(fn ->
+               ref = Broadway.test_messages(broadway, [1, 2, :broken_link, 3])
+
+               assert_receive {:ack, ^ref, [],
+                               [
+                                 %{data: 1, status: {:failed, "due to an unhandled exit"}},
+                                 %{data: 2, status: {:failed, "due to an unhandled exit"}},
+                                 %{
+                                   data: :broken_link,
+                                   status: {:failed, "due to an unhandled exit"}
+                                 },
+                                 %{data: 3, status: {:failed, "due to an unhandled exit"}}
+                               ]}
+             end) =~ "** (RuntimeError) oops"
+
+      refute_received {:EXIT, _, ^consumer}
+    end
   end
 
   describe "shutdown" do
@@ -1123,8 +1157,7 @@ defmodule BroadwayTest do
     end
 
     @tag shutdown: 1
-    test "shutting down broadway respects shutdown value",
-         %{broadway: broadway} do
+    test "shutting down broadway respects shutdown value", %{broadway: broadway} do
       Broadway.test_messages(broadway, [:sleep, 1, 2, 3])
       Process.exit(broadway, :shutdown)
       assert_receive {:EXIT, ^broadway, :shutdown}