diff --git a/mrq/scheduler.py b/mrq/scheduler.py index c025e4c9..1b12e2b2 100644 --- a/mrq/scheduler.py +++ b/mrq/scheduler.py @@ -91,7 +91,8 @@ def check(self): interval = datetime.timedelta(seconds=task["interval"]) - last_time = now - interval + if task["datelastqueued"] >= now: + continue if task.get("monthday", current_monthday) != current_monthday: continue @@ -103,19 +104,22 @@ def check(self): if task["datelastqueued"].date() == now.date() or now.time() < task["dailytime"].time(): continue - if task["datelastqueued"] <= last_time: + # if we only have "interval" key + if all(k not in task for k in ["monthday", "weekday", "dailytime"]): + if now - task["datelastqueued"] < interval: + continue - queue_job( - task["path"], - task.get("params") or {}, - queue=task.get("queue") - ) + queue_job( + task["path"], + task.get("params") or {}, + queue=task.get("queue") + ) - self.collection.update({"_id": task["_id"]}, {"$set": { - "datelastqueued": now - }}) + self.collection.update({"_id": task["_id"]}, {"$set": { + "datelastqueued": now + }}) - log.debug("Scheduler: queued %s" % _hash_task(task)) + log.debug("Scheduler: queued %s" % _hash_task(task)) # Make sure we never again execute a scheduler with the same exact second. time.sleep(1) diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index 034b5d62..d1a6f1fa 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -44,7 +44,7 @@ def test_scheduler_simple(worker, p_flags): collection.remove({}) - scheduled_jobs.update_many({}, {"$set": {"datelastqueued": datetime.datetime.utcnow()}}) + scheduled_jobs.update_many({}, {"$set": {"datelastqueued": datetime.datetime.utcnow() + timedelta(seconds=10)}}) # Start with new config worker.start( @@ -114,6 +114,26 @@ def _start(deps=True): assert collection.find({"params.b": "test"}).count() == 1 +def test_scheduler_dailytime_with_datelastqueued(worker): + now = time.time() + worker.start( + flags="--scheduler --config tests/fixtures/config-scheduler3.py", + env={ + # We need to pass this in the environment so that each worker has the + # exact same hash + "MRQ_TEST_SCHEDULER_TIME": str(now - 100 - 3600 * 24) + }) + time.sleep(7) + print(list(worker.mongodb_jobs.tests_inserts.find())) + assert len(list(worker.mongodb_jobs.tests_inserts.find())) == 2 + # pretend first run was yesterday at time(now) + 200s + datelastqueued = datetime.datetime.fromtimestamp(now - 3600*24 + 200) + worker.mongodb_jobs.mrq_scheduled_jobs.update({"params.b": "test"}, {"$set": {'datelastqueued': datelastqueued}}) + time.sleep(3) + # task is ran + assert len(list(worker.mongodb_jobs.tests_inserts.find())) == 3 + + def test_scheduler_weekday_dailytime(worker): # Task is scheduled in 5 seconds worker.start(