Propagate internal errors to the master node

This should help users diagnose internal errors in workers like
exceptions from hooks or in pytest itself.

changelog/608.feature.rst

@@ -0,0 +1 @@
+Internal errors in workers are now propagated to the master node.

src/xdist/dsession.py

@@ -174,6 +174,24 @@ def worker_workerfinished(self, node):
             assert not crashitem, (crashitem, node)
         self._active_nodes.remove(node)
 
+    def worker_internal_error(self, node, formatted_error):
+        """
+        pytest_internalerror() was called on the worker.
+
+        pytest_internalerror() arguments are an excinfo and an excrepr, which can't
+        be serialized, so we go with a poor man's solution of raising an exception
+        here ourselves using the formatted message.
+        """
+        self._active_nodes.remove(node)
+        try:
+            assert False, formatted_error
+        except AssertionError:
+            from _pytest._code import ExceptionInfo
+
+            excinfo = ExceptionInfo.from_current()
+            excrepr = excinfo.getrepr()
+            self.config.hook.pytest_internalerror(excrepr=excrepr, excinfo=excinfo)
+
     def worker_errordown(self, node, error):
         """Emitted by the WorkerController when a node dies."""
         self.config.hook.pytest_testnodedown(node=node, error=error)

src/xdist/remote.py

@@ -33,8 +33,10 @@ def sendevent(self, name, **kwargs):
         self.channel.send((name, kwargs))
 
     def pytest_internalerror(self, excrepr):
-        for line in str(excrepr).split("\n"):
+        formatted_error = str(excrepr)
+        for line in formatted_error.split("\n"):
             self.log("IERROR>", line)
+        interactor.sendevent("internal_error", formatted_error=formatted_error)
 
     def pytest_sessionstart(self, session):
         self.session = session

src/xdist/workermanage.py

@@ -324,6 +324,8 @@ def process_from_remote(self, eventcall):  # noqa too complex
                 self.log("ignoring {}({})".format(eventname, kwargs))
             elif eventname == "workerready":
                 self.notify_inproc(eventname, node=self, **kwargs)
+            elif eventname == "internal_error":
+                self.notify_inproc(eventname, node=self, **kwargs)
             elif eventname == "workerfinished":
                 self._down = True
                 self.workeroutput = kwargs["workeroutput"]

testing/acceptance_test.py

@@ -1138,6 +1138,18 @@ def test_aaa1(crasher):
     assert "INTERNALERROR" not in result.stderr.str()
 
 
+def test_internal_errors_propagate_to_master(testdir):
+    testdir.makeconftest(
+        """
+        def pytest_collection_modifyitems():
+            raise RuntimeError("Some runtime error")
+        """
+    )
+    testdir.makepyfile("def test(): pass")
+    result = testdir.runpytest("-n1")
+    result.stdout.fnmatch_lines(["*RuntimeError: Some runtime error*"])
+
+
 class TestLoadScope:
     def test_by_module(self, testdir):
         test_file = """