Repository: virtool/virtool
Issue: web_protocol ERROR: Error handling request

A job shows "waiting" forever, when clicking the red cross to delete it, the following message shows up in the terminal and then anything in virtool becomes unviewable (blank white page when click) and unable to logout:

2017-11-07 16:53:56  web_protocol ERROR    Error handling request
Traceback (most recent call last):
  File "/home/travis/virtualenv/python3.6.3/lib/python3.6/site-packages/aiohttp/web_protocol.py", line 422, in start
  File "/home/travis/virtualenv/python3.6.3/lib/python3.6/site-packages/aiohttp/web.py", line 306, in _handle
  File "/home/travis/build/virtool/virtool/virtool/error_pages.py", line 14, in middleware_handler
  File "/home/travis/build/virtool/virtool/virtool/app_auth.py", line 96, in middleware_handler
  File "/home/travis/build/virtool/virtool/virtool/handlers/jobs.py", line 95, in cancel
  File "/home/travis/build/virtool/virtool/virtool/job_manager.py", line 116, in cancel
KeyError: 'ycwovqus'


Changed files:
tests/handlers/test_jobs.py, virtool/job_manager.py, virtool/handlers/jobs.py, tests/test_job.py, virtool/job.py

diff --git a/tests/handlers/test_jobs.py b/tests/handlers/test_jobs.py
index 30da75078..3b639cffa 100644
--- a/tests/handlers/test_jobs.py
+++ b/tests/handlers/test_jobs.py
@@ -1,12 +1,20 @@
-class TestGet:
+import pytest
 
-    async def test(self, spawn_client, test_job):
-        client = await spawn_client()
 
+@pytest.mark.parametrize("status", [200, 404])
+async def test_get(status, spawn_client, test_job, resp_is):
+    client = await spawn_client()
+
+    if status == 200:
         await client.db.jobs.insert_one(test_job)
 
-        resp = await client.get("/api/jobs/4c530449")
+    resp = await client.get("/api/jobs/4c530449")
+
+    assert resp.status == status
 
+    if status == 404:
+        assert await resp_is.not_found(resp)
+    else:
         assert await resp.json() == {
             "task": "rebuild_index",
             "mem": 16,
@@ -55,9 +63,6 @@ class TestGet:
             ]
         }
 
-    async def test_not_found(self, spawn_client, resp_is):
-        client = await spawn_client()
 
-        resp = await client.get("/api/jobs/4c530449")
-
-        assert await resp_is.not_found(resp)
+async def test_cancel(spawn_client):
+    pass
diff --git a/tests/test_job.py b/tests/test_job.py
index 120a3c378..7b31fe711 100644
--- a/tests/test_job.py
+++ b/tests/test_job.py
@@ -1,37 +1,86 @@
 import datetime
+import pytest
 
 import virtool.job
 
 
-class TestDispatchProcessor:
+def test_processor(test_db, test_job):
+    """
+    Test that the dispatch processor properly formats a raw job document into a dispatchable format.
 
-    def test(self, test_db, test_job):
-        """
-        Test that the dispatch processor properly formats a raw job document into a dispatchable format.
-         
-        """
-        test_db.jobs.insert_one(test_job)
+    """
+    test_db.jobs.insert_one(test_job)
 
-        document = test_db.jobs.find_one()
+    document = test_db.jobs.find_one()
 
-        assert virtool.job.dispatch_processor(document) == {
-            "id": "4c530449",
-            "created_at": datetime.datetime(2017, 10, 6, 20, 0),
-            "args": {
-                "algorithm": "nuvs",
-                "analysis_id": "e410429b",
-                "index_id": "465428b0",
-                "name": None,
-                "sample_id": "1e01a382",
-                "username": "igboyes"
+    assert virtool.job.processor(document) == {
+        "id": "4c530449",
+        "created_at": datetime.datetime(2017, 10, 6, 20, 0),
+        "args": {
+            "algorithm": "nuvs",
+            "analysis_id": "e410429b",
+            "index_id": "465428b0",
+            "name": None,
+            "sample_id": "1e01a382",
+            "username": "igboyes"
+        },
+        "mem": 16,
+        "proc": 10,
+        "progress": 1.0,
+        "stage": "import_results",
+        "state": "complete",
+        "task": "rebuild_index",
+        "user": {
+            "id": "igboyes"
+        }
+    }
+
+
+@pytest.mark.parametrize("empty", [True, False], ids=["empty", "not-empty"])
+async def test_get_waiting_and_running(empty, test_motor):
+    documents = list()
+
+    if not empty:
+        documents.append({
+            "_id": "foo",
+            "status": [
+                {"state": "waiting"},
+                {"state": "running"},
+            ]
+        })
+
+    documents.append({
+        "_id": "bar",
+        "status": [
+            {"state": "waiting"},
+            {"state": "running"},
+            {"state": "complete"}
+        ]
+    })
+
+    if not empty:
+        documents += [
+            {
+                "_id": "baz",
+                "status": [
+                    {"state": "waiting"}
+                ]
             },
-            "mem": 16,
-            "proc": 10,
-            "progress": 1.0,
-            "stage": "import_results",
-            "state": "complete",
-            "task": "rebuild_index",
-            "user": {
-                "id": "igboyes"
+            {
+                "_id": "bat",
+                "status": [
+                    {"state": "waiting"},
+                    {"state": "running"}
+                ]
             }
-        }
+        ]
+
+    await test_motor.jobs.insert_many(documents)
+
+    expected = list()
+
+    if not empty:
+        expected = ["foo", "baz", "bat"]
+
+    assert await virtool.job.get_waiting_and_running_ids(test_motor) == expected
+
diff --git a/virtool/handlers/jobs.py b/virtool/handlers/jobs.py
index 7bc4d9417..98e488002 100644
--- a/virtool/handlers/jobs.py
+++ b/virtool/handlers/jobs.py
@@ -1,11 +1,10 @@
 import os
-import math
 from cerberus import Validator
 
-import virtool.utils
 import virtool.job
-from virtool.handlers.utils import json_response, bad_request, not_found, invalid_query, \\
-    compose_regex_query, no_content
+import virtool.utils
+from virtool.handlers.utils import compose_regex_query, bad_request, invalid_query, json_response, no_content,\\
+    not_found, paginate
 
 
 async def find(req):
@@ -15,7 +14,6 @@ async def find(req):
     """
     db = req.app["db"]
 
-    # Validator for URL query.
     v = Validator({
         "term": {"type": "string", "default": "", "coerce": str},
         "page": {"type": "integer", "coerce": int, "default": 1, "min": 1},
@@ -27,37 +25,23 @@ async def find(req):
 
     query = v.document
 
-    page = query["page"]
-    per_page = query["per_page"]
+    term = query["term"]
 
     db_query = dict()
 
-    if query["term"]:
-        db_query.update(compose_regex_query(query["term"], ["task", "user_id"]))
+    if term:
+        db_query.update(compose_regex_query(term, ["task", "user.id"]))
 
-    total_count = await db.jobs.count()
-
-    cursor = db.jobs.find(
+    data = await paginate(
+        db.jobs,
         db_query,
-        virtool.job.LIST_PROJECTION,
-        sort=[("name", 1)]
+        req.query,
+        "name",
+        projection=virtool.job.LIST_PROJECTION,
+        processor=virtool.job.processor
     )
 
-    found_count = await cursor.count()
-
-    if page > 1:
-        cursor.skip((page - 1) * per_page)
-
-    documents = [virtool.job.dispatch_processor(d) for d in await cursor.to_list(per_page)]
-
-    return json_response({
-        "documents": documents,
-        "total_count": total_count,
-        "found_count": found_count,
-        "page": page,
-        "per_page": per_page,
-        "page_count": int(math.ceil(found_count / per_page))
-    })
+    return json_response(data)
 
 
 async def get(req):
diff --git a/virtool/job.py b/virtool/job.py
index 8244325db..737a5d7e8 100644
--- a/virtool/job.py
+++ b/virtool/job.py
@@ -19,7 +19,7 @@ LIST_PROJECTION = [
 ]
 
 
-def dispatch_processor(document):
+def processor(document):
     """
     Removes the ``status`` and ``args`` fields from the job document.
     Adds a ``username`` field, an ``added`` date taken from the first status entry in the job document, and
@@ -46,6 +46,29 @@ def dispatch_processor(document):
     return document
 
 
+async def get_waiting_and_running_ids(db):
+    agg = await db.jobs.aggregate([
+        {"$project": {
+            "status": {
+                "$arrayElemAt": ["$status", -1]
+            }
+        }},
+
+        {"$match": {
+            "$or": [
+                {"status.state": "waiting"},
+                {"status.state": "running"},
+            ]
+        }},
+
+        {"$project": {
+            "_id": True
+        }}
+    ]).to_list(None)
+
+    return [a["_id"] for a in agg]
+
+
 class Job:
 
     def __init__(self, loop, executor, db, settings, dispatch, job_id, task_name, task_args, proc, mem):
@@ -197,7 +220,7 @@ class Job:
             }
         }, return_document=pymongo.ReturnDocument.AFTER, projection=LIST_PROJECTION)
 
-        await self.dispatch("jobs", "update", dispatch_processor(document))
+        await self.dispatch("jobs", "update", processor(document))
 
     async def add_log(self, line, indent=0):
         timestamp = virtool.utils.timestamp().isoformat()
diff --git a/virtool/job_manager.py b/virtool/job_manager.py
index 4e21af4f7..a90224063 100644
--- a/virtool/job_manager.py
+++ b/virtool/job_manager.py
@@ -31,6 +31,8 @@ class Manager:
         #: A dict to store all the tracked job objects in.
         self._jobs_dict = dict()
 
+        self._run_task = None
+
         #: The main loop
         self._stop = False
         self.started = False
@@ -40,9 +42,11 @@ class Manager:
 
     def start(self):
         self.started = True
-        asyncio.ensure_future(self.run(), loop=self.loop)
+        self._run_task = asyncio.ensure_future(self.run(), loop=self.loop)
 
     async def run(self):
+        iteration = 0
+
         while True:
             to_delete = list()
 
@@ -61,6 +65,19 @@ class Manager:
             for job_id in to_delete:
                 del self._jobs_dict[job_id]
 
+            iteration += 1
+
+            if iteration == 100:
+                iteration = 0
+
+                ids = await virtool.job.get_waiting_and_running_ids(self.db)
+
+                await self.db.jobs.delete_many({
+                    "_id": {
+                        "$in": [job_id for job_id in ids if job_id not in self._jobs_dict]
+                    }
+                })
+
             await asyncio.sleep(0.1, loop=self.loop)
 
             if self._stop and len(self._jobs_dict) == 0:
@@ -108,7 +125,7 @@ class Manager:
 
         document = await self.db.jobs.find_one(job_id, virtool.job.LIST_PROJECTION)
 
-        await self.dispatch("jobs", "update", virtool.job.dispatch_processor(document))
+        await self.dispatch("jobs", "update", virtool.job.processor(document))
 
         return virtool.utils.base_processor(document)