add support to delay a job and enqueue it N seconds later

docs/jobs-maintenance.md

@@ -7,6 +7,13 @@ To do that, you should add these recurring scheduled jobs to your mrq-config.py:
 ```
 SCHEDULER_TASKS = [
 
+  # This will queue jobs in the 'delayed' status.
+  {
+    "path": "mrq.basetasks.cleaning.QueueDelayedJobs",
+    "params": {},
+    "interval": 60
+  },
+
   # This will requeue jobs in the 'retry' status, until they reach their max_retries.
   {
     "path": "mrq.basetasks.cleaning.RequeueRetryJobs",

docs/jobs.md

@@ -8,8 +8,9 @@ A **Job** is an instance of the execution of a Task. It must link to a specific
 
 MRQ defines a list of statuses for jobs. A job can only be in one of them at a time.
 
-When everything goes fine, a job will go through 3 statuses:
+When everything goes fine, a job will go through 4 statuses:
 
+* ```delayed```: The Job has been created and it is delayed to be queued later.
 * ```queued```: The Job has been created and it is waiting to be dequeued by a Worker.
 * ```started```: A Worker has dequeued the job and started executing it.
 * ```success```: The job was successfully ran.
@@ -51,6 +52,11 @@ Queues a job. If `queue` is not provided, the default queue for that Task as def
 
 Queues multiple jobs at once. Returns a list of IDs of the jobs.
 
+* `queue_job(main_task_path, params, delay=120, queue=None)`
+
+Create a job with `delayed` status, the job will be queued after of at least `delay` seconds. 
+Remember to add the base delayed job as explained in [Jobs maintenance](jobs-maintenance.md) to have `delayed` jobs actually queued.
+
 * `queue_raw_jobs(queue, params_list, batch_size=1000)`
 
 Queues multiple jobs at once on a [raw queue](queues.md#raw-queues). The queued jobs have no IDs on a raw queue so this function has no return.

mrq/basetasks/cleaning.py

@@ -35,6 +35,20 @@ def run(self, params):
         })
 
 
+class QueueDelayedJobs(Task):
+
+    """ Requeue jobs that were marked as delayed. """
+
+    max_concurrency = 1
+
+    def run(self, params):
+        return run_task("mrq.basetasks.utils.JobAction", {
+            "status": "delayed",
+            "dateretry": {"$lte": datetime.datetime.utcnow()},
+            "action": "requeue"
+        })
+
+
 class RequeueStartedJobs(Task):
 
     """ Requeue jobs that were marked as status=started and never finished.

mrq/dashboard/static/js/views/jobs.js

@@ -347,6 +347,9 @@ define(["jquery", "underscore", "views/generic/datatablepage", "models"],functio
                 if (source.time) {
                   display.push("cputime "+String(source.time).substring(0,6)+"s ("+source.switches+" switches)");
                 }
+                if (source.status === 'retry' || source.status === 'delayed') {
+                  display.push("requeue "+moment.utc(source.dateretry).fromNow());
+                }
 
                 return "<small>" + display.join("<br/>") + "</small>";
 

mrq/dashboard/templates/index.html

@@ -391,6 +391,7 @@ <h4 class="modal-title"></h4>
               <select class="form-control input-sm js-datatable-filters-status" id="jobs-form-status">
                 <option <%= filters.status==""?"selected='selected'":"" %> value="">-statuses-</option>
                 <% _.each({
+                    "delayed": "delayed",
                     "queued": "queued",
                     "started": "started",
                     "success": "success",

mrq/job.py

@@ -23,7 +23,7 @@
 from . import context
 
 
-FINAL_STATUSES = {"timeout", "abort", "failed", "success", "interrupt", "retry", "maxretries", "maxconcurrency"}
+FINAL_STATUSES = {"timeout", "abort", "failed", "success", "interrupt", "retry", "maxretries", "maxconcurrency", "delayed"}
 TRANSIENT_STATUSES = {"cancel", "queued", "started"}
 
 
@@ -682,7 +682,7 @@ def set_queues_size(size_by_queues, action="incr"):
                 pipe.expire("queuesize:%s" % queue, context.get_current_config().get("queue_ttl"))
             pipe.execute()
 
-def queue_jobs(main_task_path, params_list, queue=None, batch_size=1000):
+def queue_jobs(main_task_path, params_list, delay=0, queue=None, batch_size=1000):
     """ Queue multiple jobs on a regular queue """
     if len(params_list) == 0:
         return []
@@ -702,14 +702,24 @@ def queue_jobs(main_task_path, params_list, queue=None, batch_size=1000):
 
         context.metric("jobs.status.queued", len(params_group))
 
+        jobs_data = []
+        for params in params_group:
+            job_data = {
+                "path": main_task_path,
+                "params": params,
+                "queue": queue,
+                "datequeued": datetime.datetime.utcnow(),
+                "status": "queued"
+            }
+            if delay and delay > 0:
+                dateretry = datetime.datetime.utcnow() + datetime.timedelta(seconds=delay)
+                job_data['status'] = 'delayed'
+                job_data['dateretry'] = dateretry
+
+            jobs_data.append(job_data)
+
         # Insert the job in MongoDB
-        job_ids = Job.insert([{
-            "path": main_task_path,
-            "params": params,
-            "queue": queue,
-            "datequeued": datetime.datetime.utcnow(),
-            "status": "queued"
-        } for params in params_group], w=1, return_jobs=False)
+        job_ids = Job.insert(jobs_data, w=1, return_jobs=False)
 
         all_ids += job_ids