Revert "Handle thread safety wrt the messaging API in callbacks"

We are not running Bodhi in Twisted's event loop thread, and we
are not using Twisted's Pika's connections, so we do not need to
use blockingCallFromThread on our publish calls.

This reverts commit e07fc25.

fixes #3236

Signed-off-by: Randy Barlow <randy@electronsweatshop.com>

bodhi/server/consumers/composer.py

@@ -41,8 +41,6 @@
 
 import jinja2
 import fedora_messaging
-from twisted.internet import reactor
-from twisted.internet.threads import blockingCallFromThread
 
 from bodhi.messages.schemas import compose as compose_schemas, update as update_schemas
 from bodhi.server import buildsys, notifications, mail
@@ -186,13 +184,8 @@ def __call__(self, message: fedora_messaging.api.Message):
         message = message.body
         resume = message.get('resume', False)
         agent = message.get('agent')
-        # This callback is run in a thread by Twisted, and must therefore call
-        # the messaging API with the proper Twisted wrapper to ensure thread
-        # safety.
-        blockingCallFromThread(
-            reactor, notifications.publish,
-            compose_schemas.ComposeStartV1.from_dict(dict(agent=agent)),
-            force=True)
+        notifications.publish(compose_schemas.ComposeStartV1.from_dict(dict(agent=agent)),
+                              force=True)
 
         results = []
         threads = []
@@ -366,14 +359,11 @@ def work(self):
 
         log.info('Running ComposerThread(%s)' % self.id)
 
-        blockingCallFromThread(
-            reactor,
-            notifications.publish,
-            compose_schemas.ComposeComposingV1.from_dict(
-                dict(repo=self.id,
-                     updates=[' '.join([b.nvr for b in u.builds]) for u in self.compose.updates],
-                     agent=self.agent,
-                     ctype=self.ctype.value)),
+        notifications.publish(compose_schemas.ComposeComposingV1.from_dict(
+            dict(repo=self.id,
+                 updates=[' '.join([b.nvr for b in u.builds]) for u in self.compose.updates],
+                 agent=self.agent,
+                 ctype=self.ctype.value)),
             force=True,
         )
 
@@ -484,9 +474,7 @@ def eject_from_compose(self, update, reason):
             update.remove_tag(update.release.pending_testing_tag,
                               koji=buildsys.get_session())
         update.request = None
-        blockingCallFromThread(
-            reactor,
-            notifications.publish,
+        notifications.publish(
             update_schemas.UpdateEjectV1.from_dict(
                 dict(
                     repo=self.id,
@@ -532,11 +520,8 @@ def finish(self, success):
             success (bool): True if the compose had been successful, False otherwise.
         """
         log.info('Thread(%s) finished.  Success: %r' % (self.id, success))
-        blockingCallFromThread(
-            reactor,
-            notifications.publish,
-            compose_schemas.ComposeCompleteV1.from_dict(dict(
-                dict(success=success, repo=self.id, agent=self.agent, ctype=self.ctype.value))),
+        notifications.publish(compose_schemas.ComposeCompleteV1.from_dict(dict(
+            dict(success=success, repo=self.id, agent=self.agent, ctype=self.ctype.value))),
             force=True,
         )
 
@@ -704,7 +689,7 @@ def send_notifications(self):
                 UpdateRequest.testing: update_schemas.UpdateCompleteTestingV1
             }
             message = messages[update.request].from_dict(dict(update=update, agent=agent))
-            blockingCallFromThread(reactor, notifications.publish, message, force=True)
+            notifications.publish(message, force=True)
 
     @checkpoint
     def modify_bugs(self):
@@ -1245,11 +1230,8 @@ def _wait_for_repo_signature(self):
         """Wait for a repo signature to appear."""
         # This message indicates to consumers that the repos are fully created and ready to be
         # signed or otherwise processed.
-        blockingCallFromThread(
-            reactor,
-            notifications.publish,
-            compose_schemas.RepoDoneV1.from_dict(
-                dict(repo=self.id, agent=self.agent, path=self.path)),
+        notifications.publish(compose_schemas.RepoDoneV1.from_dict(
+            dict(repo=self.id, agent=self.agent, path=self.path)),
             force=True)
         if config.get('wait_for_repo_sig'):
             self.save_state(ComposeState.signing_repo)
@@ -1286,11 +1268,8 @@ def _wait_for_sync(self):
             Exception: If no folder other than "source" was found in the compose_path.
         """
         log.info('Waiting for updates to hit the master mirror')
-        blockingCallFromThread(
-            reactor,
-            notifications.publish,
-            compose_schemas.ComposeSyncWaitV1.from_dict(
-                dict(repo=self.id, agent=self.agent)),
+        notifications.publish(compose_schemas.ComposeSyncWaitV1.from_dict(
+            dict(repo=self.id, agent=self.agent)),
             force=True)
         compose_path = os.path.join(self.path, 'compose', 'Everything')
         checkarch = None
@@ -1324,11 +1303,8 @@ def _wait_for_sync(self):
                 continue
             if newsum == checksum:
                 log.info("master repomd.xml matches!")
-                blockingCallFromThread(
-                    reactor,
-                    notifications.publish,
-                    compose_schemas.ComposeSyncDoneV1.from_dict(
-                        dict(repo=self.id, agent=self.agent)),
+                notifications.publish(compose_schemas.ComposeSyncDoneV1.from_dict(
+                    dict(repo=self.id, agent=self.agent)),
                     force=True)
                 return
 

bodhi/tests/server/consumers/test_composer.py

@@ -80,11 +80,6 @@
 }
 
 
-def fake_blockingCallFromThread(reactor, function, *args, **kwargs):
-    """Just run the function, no messing with threads."""
-    return function(*args, **kwargs)
-
-
 class TestCheckpoint(unittest.TestCase):
     """Test the checkpoint() decorator."""
     def test_with_return(self):
@@ -126,7 +121,6 @@ def _make_msg(transactional_session_maker, extra_push_args=None):
     return publish.mock_calls[0][1][0]
 
 
-@mock.patch("bodhi.server.consumers.composer.blockingCallFromThread", fake_blockingCallFromThread)
 # We don't need real pungi config files, we just need them to exist. Let's also mock all calls to
 # pungi.
 @mock.patch.dict(
@@ -2153,7 +2147,6 @@ def test_expire_buildroot_overrides_exception(self, expire, exception_log, *args
         exception_log.assert_called_once_with("Problem expiring override")
 
 
-@mock.patch("bodhi.server.consumers.composer.blockingCallFromThread", fake_blockingCallFromThread)
 class ComposerThreadBaseTestCase(base.BaseTestCase):
     """Methods that are useful for testing ComposerThread subclasses."""
 
@@ -2577,8 +2570,6 @@ def test_primary_arches_undefined(self):
 class TestComposerThread_perform_gating(ComposerThreadBaseTestCase):
     """Test the ComposerThread.perform_gating() method."""
 
-    @mock.patch("bodhi.server.consumers.composer.blockingCallFromThread",
-                fake_blockingCallFromThread)
     def test_expires_compose_updates(self):
         """Ensure that the method expires the compose's updates attribute."""
         msg = self._make_msg()
@@ -2669,8 +2660,6 @@ def test_BodhiException(self, mocked_log):
 class TestComposerThread__determine_tag_actions(ComposerThreadBaseTestCase):
     """Test ComposerThread._determine_tag_actions()."""
 
-    @mock.patch("bodhi.server.consumers.composer.blockingCallFromThread",
-                fake_blockingCallFromThread)
     @mock.patch('bodhi.server.models.buildsys.get_session')
     def test_from_tag_not_found(self, get_session):
         """Updates should be ejected if the from tag cannot be determined."""
@@ -2707,8 +2696,6 @@ def test_from_tag_not_found(self, get_session):
 
 class TestComposerThread_eject_from_compose(ComposerThreadBaseTestCase):
     """This test class contains tests for the ComposerThread.eject_from_compose() method."""
-    @mock.patch("bodhi.server.consumers.composer.blockingCallFromThread",
-                fake_blockingCallFromThread)
     def test_testing_request(self):
         """
         Assert correct behavior when the update's request is set to testing.
@@ -2853,8 +2840,6 @@ def test_without_state(self):
 
 class TestPungiComposerThread__wait_for_sync(ComposerThreadBaseTestCase):
     """This test class contains tests for the PungiComposerThread._wait_for_sync() method."""
-    @mock.patch("bodhi.server.consumers.composer.blockingCallFromThread",
-                fake_blockingCallFromThread)
     @mock.patch.dict(
         'bodhi.server.consumers.composer.config',
         {'fedora_testing_master_repomd':
@@ -2897,8 +2882,6 @@ def test_checksum_match_immediately(self, urlopen, save):
         self.assertTrue(urlopen.mock_calls[0] in expected_calls)
         save.assert_called_once_with(ComposeState.syncing_repo)
 
-    @mock.patch("bodhi.server.consumers.composer.blockingCallFromThread",
-                fake_blockingCallFromThread)
     @mock.patch.dict(
         'bodhi.server.consumers.composer.config',
         {'fedora_testing_master_repomd':
@@ -2929,8 +2912,6 @@ def test_no_checkarch(self, urlopen, save):
         self.assertEqual(str(exc.exception), "Not found an arch to _wait_for_sync with")
         save.assert_not_called()
 
-    @mock.patch("bodhi.server.consumers.composer.blockingCallFromThread",
-                fake_blockingCallFromThread)
     @mock.patch.dict(
         'bodhi.server.consumers.composer.config',
         {'fedora_testing_master_repomd':
@@ -2973,8 +2954,6 @@ def test_checksum_match_third_try(self, urlopen, sleep, save):
         sleep.assert_has_calls([mock.call(200), mock.call(200)])
         save.assert_called_with(ComposeState.syncing_repo)
 
-    @mock.patch("bodhi.server.consumers.composer.blockingCallFromThread",
-                fake_blockingCallFromThread)
     @mock.patch.dict(
         'bodhi.server.consumers.composer.config',
         {'fedora_testing_master_repomd':
@@ -3021,8 +3000,6 @@ def test_httperror(self, mocked_log, urlopen, sleep, save):
         sleep.assert_called_once_with(200)
         save.assert_called_once_with(ComposeState.syncing_repo)
 
-    @mock.patch("bodhi.server.consumers.composer.blockingCallFromThread",
-                fake_blockingCallFromThread)
     @mock.patch.dict(
         'bodhi.server.consumers.composer.config',
         {'fedora_testing_master_repomd':
@@ -3070,8 +3047,6 @@ def test_connectionreseterror(self, mocked_log, urlopen, sleep, save):
         sleep.assert_called_once_with(200)
         save.assert_called_once_with(ComposeState.syncing_repo)
 
-    @mock.patch("bodhi.server.consumers.composer.blockingCallFromThread",
-                fake_blockingCallFromThread)
     @mock.patch.dict(
         'bodhi.server.consumers.composer.config',
         {'fedora_testing_master_repomd':
@@ -3118,8 +3093,6 @@ def test_incompleteread(self, mocked_log, urlopen, sleep, save):
         sleep.assert_called_once_with(200)
         save.assert_called_once_with(ComposeState.syncing_repo)
 
-    @mock.patch("bodhi.server.consumers.composer.blockingCallFromThread",
-                fake_blockingCallFromThread)
     @mock.patch.dict(
         'bodhi.server.consumers.composer.config',
         {'fedora_testing_master_repomd': None})
@@ -3153,8 +3126,6 @@ def test_missing_config_key(self, save):
                          'fedora_testing_master_repomd in the config file')
         save.assert_called_once_with(ComposeState.syncing_repo)
 
-    @mock.patch("bodhi.server.consumers.composer.blockingCallFromThread",
-                fake_blockingCallFromThread)
     @mock.patch('bodhi.server.consumers.composer.PungiComposerThread.save_state')
     @mock.patch('bodhi.server.consumers.composer.time.sleep',
                 mock.MagicMock(side_effect=Exception('This should not happen during this test.')))
@@ -3182,8 +3153,6 @@ def test_missing_repomd(self, mocked_log, save):
             'Cannot find local repomd: %s', os.path.join(repodata, 'repomd.xml'))
         save.assert_not_called()
 
-    @mock.patch("bodhi.server.consumers.composer.blockingCallFromThread",
-                fake_blockingCallFromThread)
     @mock.patch.dict(
         'bodhi.server.consumers.composer.config',
         {'fedora_testing_master_repomd':
@@ -3281,8 +3250,6 @@ def test_testing_update(self):
 class TestComposerThread_send_notifications(ComposerThreadBaseTestCase):
     """Test ComposerThread.send_notifications."""
 
-    @mock.patch("bodhi.server.consumers.composer.blockingCallFromThread",
-                fake_blockingCallFromThread)
     def test_getlogin_raising_oserror(self):
         """Assert that "composer" is used as the agent if getlogin() raises OSError."""
         t = ComposerThread(self.semmock, self._make_msg().body['composes'][0],
@@ -3488,8 +3455,6 @@ def test_stage_dir_dne(self, mocked_log):
 class TestPungiComposerThread__wait_for_repo_signature(ComposerThreadBaseTestCase):
     """Test PungiComposerThread._wait_for_repo_signature()."""
 
-    @mock.patch("bodhi.server.consumers.composer.blockingCallFromThread",
-                fake_blockingCallFromThread)
     @mock.patch('bodhi.server.consumers.composer.log')
     def test_dont_wait_for_signatures(self, mocked_log):
         """Test that if wait_for_repo_sig is disabled, nothing happens."""
@@ -3508,8 +3473,6 @@ def test_dont_wait_for_signatures(self, mocked_log):
             mocked_log.info.mock_calls,
             [mock.call('Not waiting for a repo signature')])
 
-    @mock.patch("bodhi.server.consumers.composer.blockingCallFromThread",
-                fake_blockingCallFromThread)
     @mock.patch('os.path.exists', side_effect=[
         # First time, none of the signatures exist
         False, False, False,

docs/user/release_notes.rst

@@ -83,7 +83,6 @@ Dependency changes
 * kitchen is no longer required (:issue:`3094`).
 * hawkey is no longer required.
 * PyYAML is now a required dependency (:issue:`3174`).
-* Twisted is now required (:issue:`3145`).
 * Bodhi now requires Python 3.6 or greater (:issue:`2856`).
 * Bodhi no longer uses or works with ``fedmsg``.
 

requirements.txt

@@ -6,8 +6,7 @@ colander
 cornice>=3.1.0
 dogpile.cache
 pyasn1-modules  # Due to an unfortunate dash in its name, installs break if pyasn1 is installed first
-fedora_messaging>=1.6.0
-Twisted
+fedora_messaging
 feedgen
 jinja2
 markdown