Fix parallel workers blocking when tasks raise exception.

openwebvulndb/common/parallel.py

@@ -51,6 +51,8 @@ async def consume(self, n):
                         self._handle_task_timeout(task)
                 else:
                     await coroutine(*args, **kwargs)
+            except Exception as e:
+                logger.warn("Unexpected exception in {} {}: {}".format(self.name, n, repr(e)))
             finally:
                 self.queue.task_done()
 

openwebvulndb/wordpress/repository.py

@@ -60,7 +60,6 @@ async def fetch(self, url, parser):
         try:
             async with self.session.get(url) as response:
                 data = await response.text()
-                response.close()
                 return parser.parse(data)
         except SoftwareNotFound:
             raise

tests/common_test/parallel_test.py

@@ -17,8 +17,10 @@
 
 from unittest import TestCase
 from fixtures import async_test
+import async_timeout
+from aiohttp.test_utils import make_mocked_coro
 
-from openwebvulndb.common.parallel import BackgroundRunner
+from openwebvulndb.common.parallel import BackgroundRunner, ParallelWorker
 
 
 class ParallelTest(TestCase):
@@ -43,3 +45,17 @@ async def test_configured_runner(self, loop):
     def test_no_loop_uses_default(self):
         runner = BackgroundRunner(None)
         self.assertIs(runner.run, BackgroundRunner.default)
+
+    @async_test()
+    async def test_consume_do_not_block_on_exception(self, loop):
+        async def coro0():
+            raise Exception()
+
+        coro = make_mocked_coro()
+        worker = ParallelWorker(1, loop=loop)
+        await worker.request(coro0)
+        await worker.request(coro)
+        with async_timeout.timeout(timeout=0.01, loop=loop):
+            await worker.wait()
+
+        coro.assert_called_once_with()