Signal handlers should not be able to propagate exceptions. Closes #2738

celery/tests/utils/test_dispatcher.py

@@ -57,18 +57,22 @@ class DispatcherTests(Case):
 
     def test_exact(self):
         a_signal.connect(receiver_1_arg, sender=self)
-        expected = [(receiver_1_arg, 'test')]
-        result = a_signal.send(sender=self, val='test')
-        self.assertEqual(result, expected)
-        a_signal.disconnect(receiver_1_arg, sender=self)
+        try:
+            expected = [(receiver_1_arg, 'test')]
+            result = a_signal.send(sender=self, val='test')
+            self.assertEqual(result, expected)
+        finally:
+            a_signal.disconnect(receiver_1_arg, sender=self)
         self._testIsClean(a_signal)
 
     def test_ignored_sender(self):
         a_signal.connect(receiver_1_arg)
-        expected = [(receiver_1_arg, 'test')]
-        result = a_signal.send(sender=self, val='test')
-        self.assertEqual(result, expected)
-        a_signal.disconnect(receiver_1_arg)
+        try:
+            expected = [(receiver_1_arg, 'test')]
+            result = a_signal.send(sender=self, val='test')
+            self.assertEqual(result, expected)
+        finally:
+            a_signal.disconnect(receiver_1_arg)
         self._testIsClean(a_signal)
 
     def test_garbage_collected(self):
@@ -83,19 +87,22 @@ class DispatcherTests(Case):
 
     def test_multiple_registration(self):
         a = Callable()
-        a_signal.connect(a)
-        a_signal.connect(a)
-        a_signal.connect(a)
-        a_signal.connect(a)
-        a_signal.connect(a)
-        a_signal.connect(a)
-        result = a_signal.send(sender=self, val='test')
-        self.assertEqual(len(result), 1)
-        self.assertEqual(len(a_signal.receivers), 1)
-        del a
-        del result
-        garbage_collect()
-        self._testIsClean(a_signal)
+        result = None
+        try:
+            a_signal.connect(a)
+            a_signal.connect(a)
+            a_signal.connect(a)
+            a_signal.connect(a)
+            a_signal.connect(a)
+            a_signal.connect(a)
+            result = a_signal.send(sender=self, val='test')
+            self.assertEqual(len(result), 1)
+            self.assertEqual(len(a_signal.receivers), 1)
+        finally:
+            del a
+            del result
+            garbage_collect()
+            self._testIsClean(a_signal)
 
     def test_uid_registration(self):
 
@@ -106,9 +113,11 @@ class DispatcherTests(Case):
             pass
 
         a_signal.connect(uid_based_receiver_1, dispatch_uid='uid')
-        a_signal.connect(uid_based_receiver_2, dispatch_uid='uid')
-        self.assertEqual(len(a_signal.receivers), 1)
-        a_signal.disconnect(dispatch_uid='uid')
+        try:
+            a_signal.connect(uid_based_receiver_2, dispatch_uid='uid')
+            self.assertEqual(len(a_signal.receivers), 1)
+        finally:
+            a_signal.disconnect(dispatch_uid='uid')
         self._testIsClean(a_signal)
 
     def test_robust(self):
@@ -117,22 +126,25 @@ class DispatcherTests(Case):
             raise ValueError('this')
 
         a_signal.connect(fails)
-        result = a_signal.send_robust(sender=self, val='test')
-        err = result[0][1]
-        self.assertTrue(isinstance(err, ValueError))
-        self.assertEqual(err.args, ('this',))
-        a_signal.disconnect(fails)
+        try:
+            a_signal.send(sender=self, val='test')
+        finally:
+            a_signal.disconnect(fails)
         self._testIsClean(a_signal)
 
     def test_disconnection(self):
         receiver_1 = Callable()
         receiver_2 = Callable()
         receiver_3 = Callable()
-        a_signal.connect(receiver_1)
-        a_signal.connect(receiver_2)
-        a_signal.connect(receiver_3)
-        a_signal.disconnect(receiver_1)
-        del receiver_2
-        garbage_collect()
-        a_signal.disconnect(receiver_3)
+        try:
+            try:
+                a_signal.connect(receiver_1)
+                a_signal.connect(receiver_2)
+                a_signal.connect(receiver_3)
+            finally:
+                a_signal.disconnect(receiver_1)
+            del receiver_2
+            garbage_collect()
+        finally:
+            a_signal.disconnect(receiver_3)
         self._testIsClean(a_signal)

celery/utils/dispatch/signal.py

@@ -7,9 +7,12 @@ from . import saferef
 
 from celery.five import range, text_t
 from celery.local import PromiseProxy, Proxy
+from celery.utils.log import get_logger
 
 __all__ = ['Signal']
 
+logger = get_logger(__name__)
+
 WEAKREF_TYPES = (weakref.ReferenceType, saferef.BoundMethodWeakref)
 
 
@@ -165,42 +168,12 @@ class Signal(object):  # pragma: no cover
         if not self.receivers:
             return responses
 
-        for receiver in self._live_receivers(_make_id(sender)):
-            response = receiver(signal=self, sender=sender, **named)
-            responses.append((receiver, response))
-        return responses
-
-    def send_robust(self, sender, **named):
-        """Send signal from sender to all connected receivers catching errors.
-
-        :param sender: The sender of the signal. Can be any python object
-            (normally one registered with a connect if you actually want
-            something to occur).
-
-        :keyword \*\*named: Named arguments which will be passed to receivers.
-            These arguments must be a subset of the argument names defined in
-            :attr:`providing_args`.
-
-        :returns: a list of tuple pairs: `[(receiver, response), … ]`.
-
-        :raises DispatcherKeyError:
-
-        if any receiver raises an error (specifically any subclass of
-        :exc:`Exception`), the error instance is returned as the result
-        for that receiver.
-
-        """
-        responses = []
-        if not self.receivers:
-            return responses
-
-        # Call each receiver with whatever arguments it can accept.
-        # Return a list of tuple pairs [(receiver, response), … ].
         for receiver in self._live_receivers(_make_id(sender)):
             try:
                 response = receiver(signal=self, sender=sender, **named)
-            except Exception as err:
-                responses.append((receiver, err))
+            except Exception as exc:
+                logger.error('Signal handler %r raised: %r',
+                             receiver, exc, exc_info=1)
             else:
                 responses.append((receiver, response))
         return responses