tweak cron

common/management/commands/cron.py

@@ -23,11 +23,11 @@ class Command(BaseCommand):
 
     def add_arguments(self, parser):
         parser.add_argument(
-            "--cancel",
+            "--cancel-all",
             action="store_true",
         )
         parser.add_argument(
-            "--schedule",
+            "--reschedule-all",
             action="store_true",
         )
         parser.add_argument(
@@ -35,21 +35,25 @@ class Command(BaseCommand):
             action="store_true",
         )
         parser.add_argument(
-            "--runonce",
+            "--run-once",
+            action="append",
+        )
+        parser.add_argument(
+            "--reschedule-now",
             action="append",
         )
 
     def handle(self, *args, **options):
-        if options["cancel"]:
+        if options["cancel_all"]:
             JobManager.cancel_all()
-        if options["schedule"]:
-            JobManager.cancel_all()  # cancel previously scheduled jobs if any
-            JobManager.schedule_all()
-        if options["runonce"]:
-            for job_id in options["runonce"]:
-                run = JobManager.run(job_id)
-                if not run:
-                    logger.error(f"job not found: {job_id}")
+        if options["reschedule_all"]:
+            JobManager.reschedule_all()
+        if options["reschedule_now"]:
+            for job_id in options["reschedule_now"]:
+                JobManager.get(job_id).reschedule(now=True)
+        if options["run_once"]:
+            for job_id in options["run_once"]:
+                JobManager.get(job_id)().run()
         if options["list"]:
             all_jobs = [j.__name__ for j in JobManager.registry]
             logger.info(f"{len(all_jobs)} available jobs: {' '.join(all_jobs)}")

common/models/cron.py

@@ -25,14 +25,15 @@ class BaseJob:
             pass
 
     @classmethod
-    def schedule(cls):
+    def schedule(cls, now=False):
         job_id = cls.__name__
+        i = timedelta(seconds=1) if now else cls.interval
         if cls.interval <= timedelta(0) or job_id in settings.DISABLE_CRON_JOBS:
-            logger.info(f"Skip scheduling job: {job_id}")
+            logger.info(f"Skip disabled job {job_id}")
             return
-        logger.info(f"Scheduling job: {job_id} in {cls.interval}")
+        logger.info(f"Scheduling job {job_id} in {i}")
         django_rq.get_queue("cron").enqueue_in(
-            cls.interval,
+            i,
             cls._run,
             job_id=job_id,
             result_ttl=-1,
@@ -40,6 +41,11 @@ class BaseJob:
             job_timeout=cls.interval.seconds - 5,
         )
 
+    @classmethod
+    def reschedule(cls, now: bool = False):
+        cls.cancel()
+        cls.schedule(now=now)
+
     @classmethod
     def _run(cls):
         cls.schedule()  # schedule next run
@@ -50,13 +56,25 @@ class BaseJob:
 
 
 class JobManager:
-    registry = set()
+    registry: set[type[BaseJob]] = set()
 
     @classmethod
     def register(cls, target):
         cls.registry.add(target)
         return target
 
+    @classmethod
+    def get(cls, job_id) -> type[BaseJob]:
+        for j in cls.registry:
+            if j.__name__ == job_id:
+                return j
+        raise KeyError(f"Job not found: {job_id}")
+
+    @classmethod
+    def get_scheduled_job_ids(cls):
+        registry = ScheduledJobRegistry(queue=django_rq.get_queue("cron"))
+        return registry.get_job_ids()
+
     @classmethod
     def schedule_all(cls):
         for j in cls.registry:
@@ -67,20 +85,6 @@ class JobManager:
         for j in cls.registry:
             j.cancel()
 
-    @classmethod
-    def run(cls, job_id):
-        for j in cls.registry:
-            if j.__name__ == job_id:
-                logger.info(f"Run job: {job_id}")
-                j().run()
-                return True
-        return False
-
-    @classmethod
-    def get_scheduled_job_ids(cls):
-        registry = ScheduledJobRegistry(queue=django_rq.get_queue("cron"))
-        return registry.get_job_ids()
-
     @classmethod
     def reschedule_all(cls):
         cls.cancel_all()

compose.yml

@@ -278,8 +278,8 @@ services:
 
   dev-neodb-worker:
     <<: *dev-neodb-service
-    command: neodb-manage rqworker-pool --num-workers 4 import export mastodon fetch crawl ap cron
-    # command: neodb-manage rqworker --with-scheduler import export mastodon fetch crawl ap cron
+    # command: neodb-manage rqworker-pool --num-workers 4 import export mastodon fetch crawl ap cron
+    command: neodb-manage rqworker --with-scheduler import export mastodon fetch crawl crawl ap cron
 
   dev-takahe-web:
     <<: *dev-neodb-service