feat(job): run pool handover jobs on a dedicated queue and worker

Procfile

@@ -1,5 +1,6 @@
 web: PYTHONUNBUFFERED=1 FLASK_ENV=development FLASK_DEBUG=1 FLASK_APP=agent.web:application SENTRY_DSN="https://e9b2a2274f2245daab48c6245fb69431@trace.frappe.cloud/17" ./env/bin/flask run -p 25052
 redis: redis-server redis.conf
-worker_1: PYTHONUNBUFFERED=1 ./repo/wait-for-it.sh redis://127.0.0.1:25025 && ./env/bin/rq worker --url redis://127.0.0.1:25025 high default low --sentry-dsn 'https://e9b2a2274f2245daab48c6245fb69431@trace.frappe.cloud/17'
-worker_2: PYTHONUNBUFFERED=1 ./repo/wait-for-it.sh redis://127.0.0.1:25025 && ./env/bin/rq worker --url redis://127.0.0.1:25025 high default low --sentry-dsn 'https://e9b2a2274f2245daab48c6245fb69431@trace.frappe.cloud/17'
+worker_1: PYTHONUNBUFFERED=1 ./repo/wait-for-it.sh redis://127.0.0.1:25025 && ./env/bin/rq worker --url redis://127.0.0.1:25025 signup high default low --sentry-dsn 'https://e9b2a2274f2245daab48c6245fb69431@trace.frappe.cloud/17'
+worker_2: PYTHONUNBUFFERED=1 ./repo/wait-for-it.sh redis://127.0.0.1:25025 && ./env/bin/rq worker --url redis://127.0.0.1:25025 signup high default low --sentry-dsn 'https://e9b2a2274f2245daab48c6245fb69431@trace.frappe.cloud/17'
 worker_backup: PYTHONUNBUFFERED=1 ./repo/wait-for-it.sh redis://127.0.0.1:25025 && ./env/bin/rq worker --url redis://127.0.0.1:25025 backup high default low --sentry-dsn 'https://e9b2a2274f2245daab48c6245fb69431@trace.frappe.cloud/17'
+worker_signup: PYTHONUNBUFFERED=1 ./repo/wait-for-it.sh redis://127.0.0.1:25025 && ./env/bin/rq worker --url redis://127.0.0.1:25025 signup --sentry-dsn 'https://e9b2a2274f2245daab48c6245fb69431@trace.frappe.cloud/17'

agent/proxy.py

@@ -136,7 +136,7 @@ def _add_auto_scale_sites_to_upstream(
     def add_site_to_upstream_job(self, upstream, site):
         self.add_site_domain_to_upstream(upstream, site)
 
-    @job("Add Domain to Upstream")
+    @job("Add Domain to Upstream", priority="signup")
     def add_domain_to_upstream_job(self, upstream, domain):
         self.add_site_domain_to_upstream(upstream, domain)
 
@@ -289,7 +289,7 @@ def update_site_status(self, upstream, site, status):
         with open(site_file, "w") as f:
             f.write(status)
 
-    @job("Setup Redirects on Hosts")
+    @job("Setup Redirects on Hosts", priority="signup")
     def setup_redirects_job(self, hosts, target):
         with self.proxy_config_modification_lock:
             if target in hosts:

agent/server.py

@@ -86,6 +86,10 @@ def press_url(self):
     def backup_workers(self) -> int:
         return self.config.get("backup_workers", 1)
 
+    @property
+    def signup_workers(self) -> int:
+        return self.config.get("signup_workers", 1)
+
     def docker_login(self, registry):
         url = shlex.quote(registry["url"])
         username = shlex.quote(registry["username"])
@@ -948,6 +952,10 @@ def update_agent_web(self, url=None, branch="master"):
             worker_name = f"agent:worker_backup-{worker}"
             self.execute(f"sudo supervisorctl restart {worker_name}", non_zero_throw=False)
 
+        for worker in range(self.signup_workers):
+            worker_name = f"agent:worker_signup-{worker}"
+            self.execute(f"sudo supervisorctl restart {worker_name}", non_zero_throw=False)
+
         self.execute("sudo supervisorctl restart agent:web")
         run_patches()
 
@@ -975,7 +983,7 @@ def update_agent_cli(  # noqa: C901
 
         # Stop required services
         if restart_rq_workers:
-            for group in ("worker", "worker_backup"):
+            for group in ("worker", "worker_backup", "worker_signup"):
                 for worker_id in supervisor_status.get(group, {}):
                     self.execute(
                         f"sudo supervisorctl stop agent:{group}-{worker_id}",
@@ -1020,6 +1028,8 @@ def update_agent_cli(  # noqa: C901
                 self.execute(f"sudo supervisorctl start agent:worker-{i}")
             for i in range(self.backup_workers):
                 self.execute(f"sudo supervisorctl start agent:worker_backup-{i}", non_zero_throw=False)
+            for i in range(self.signup_workers):
+                self.execute(f"sudo supervisorctl start agent:worker_signup-{i}", non_zero_throw=False)
 
         if restart_web_workers:
             self.execute("sudo supervisorctl start agent:web")
@@ -1281,6 +1291,7 @@ def _generate_supervisor_config(self):
             "gunicorn_workers": self.config.get("gunicorn_workers", 2),
             "workers": self.config["workers"],
             "backup_workers": self.backup_workers,
+            "signup_workers": self.signup_workers,
             "directory": self.directory,
             "user": self.config["user"],
             "sentry_dsn": self.config.get("sentry_dsn"),

agent/site.py

@@ -350,7 +350,7 @@ def update_config(self, value, remove=None):
 
         self.set_config(new_config)
 
-    @job("Add Domain", priority="high")
+    @job("Add Domain", priority="signup")
     def add_domain(self, domain):
         domains = set(self.config.get("domains", []))
         domains.add(domain)
@@ -501,7 +501,7 @@ def complete_setup_wizard(self, data):
         command = f"execute frappe.desk.page.setup_wizard.setup_wizard.setup_complete --kwargs {payload}"
         return self.bench_execute(command)
 
-    @job("Update Site Configuration", priority="high")
+    @job("Update Site Configuration", priority="signup")
     def update_config_job(self, value, remove):
         self.update_config(value, remove)
 

agent/templates/agent/supervisor.conf.jinja2

@@ -18,7 +18,7 @@ user={{ user }}
 directory={{ directory }}
 
 [program:worker]
-command=bash -c "{{ directory }}/repo/wait-for-it.sh redis://127.0.0.1:{{ redis_port }} && exec {{ directory }}/env/bin/rq worker {% if sentry_dsn %}--sentry-dsn '{{ sentry_dsn }}'{% endif %} --url redis://127.0.0.1:{{ redis_port }} high default low"
+command=bash -c "{{ directory }}/repo/wait-for-it.sh redis://127.0.0.1:{{ redis_port }} && exec {{ directory }}/env/bin/rq worker {% if sentry_dsn %}--sentry-dsn '{{ sentry_dsn }}'{% endif %} --url redis://127.0.0.1:{{ redis_port }} signup high default low"
 environment=PYTHONUNBUFFERED=1
 autostart=true
 autorestart=true
@@ -49,6 +49,22 @@ stderr_logfile={{ directory }}/logs/worker_backup.error.log
 user={{ user }}
 directory={{ directory }}
 
+{# Signup jobs get their own worker, so long jobs on the shared queues can't delay a signup. #}
+{# It takes no other queue: a long job picked up while idle would hold it when a signup arrives. #}
+[program:worker_signup]
+command=bash -c "{{ directory }}/repo/wait-for-it.sh redis://127.0.0.1:{{ redis_port }} && exec {{ directory }}/env/bin/rq worker {% if sentry_dsn %}--sentry-dsn '{{ sentry_dsn }}'{% endif %} --url redis://127.0.0.1:{{ redis_port }} signup"
+environment=PYTHONUNBUFFERED=1
+autostart=true
+autorestart=true
+stopwaitsecs=60
+killasgroup=true
+numprocs={{ signup_workers }}
+process_name=%(program_name)s-%(process_num)d
+stdout_logfile={{ directory }}/logs/worker_signup.log
+stderr_logfile={{ directory }}/logs/worker_signup.error.log
+user={{ user }}
+directory={{ directory }}
+
 {% if is_proxy_server or is_standalone %}
 [program:nginx_reload_manager]
 command=bash -c "{{ directory }}/repo/wait-for-it.sh redis://127.0.0.1:{{ redis_port }} && exec {{directory}}/env/bin/python {{ directory }}/repo/agent/nginx_reload_manager.py"
@@ -64,7 +80,7 @@ directory={{ directory }}
 
 [group:agent]
 
-{% set programs = "web, redis, worker, worker_backup" %}
+{% set programs = "web, redis, worker, worker_backup, worker_signup" %}
 
 {% if is_proxy_server or is_standalone %}
     {% set programs = programs + ", nginx_reload_manager" %}

agent/tests/test_server.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import configparser
 import json
 import os
 import shutil
@@ -97,3 +98,49 @@ def test_a_rejected_source_leaves_the_stored_config_untouched(self):
         config = self.server.config
         self.assertEqual(config["ip_accept"], ["183.82.5.84/32"])
         self.assertEqual(config["proxy_ip"], "10.3.1.0/24")
+
+
+class TestServerSupervisorConfig(unittest.TestCase):
+    """Signup jobs get a worker that long jobs on the shared queues can't hold."""
+
+    def setUp(self):
+        self.directory = tempfile.mkdtemp()
+        with open(os.path.join(self.directory, "config.json"), "w") as f:
+            json.dump(
+                {
+                    "name": "f1-mumbai.frappe.cloud",
+                    "benches_directory": os.path.join(self.directory, "benches"),
+                    "nginx_directory": os.path.join(self.directory, "nginx"),
+                    "web_port": 25052,
+                    "redis_port": 25025,
+                    "workers": 2,
+                    "user": "frappe",
+                },
+                f,
+            )
+
+        self.server = Server(self.directory)
+
+    def tearDown(self):
+        shutil.rmtree(self.directory)
+
+    def rendered_supervisor_config(self):
+        self.server._generate_supervisor_config()
+        config = configparser.ConfigParser(interpolation=None)
+        config.read(os.path.join(self.directory, "supervisor.conf"))
+        return config
+
+    def test_signup_worker_listens_only_on_the_signup_queue(self):
+        command = self.rendered_supervisor_config()["program:worker_signup"]["command"]
+
+        self.assertTrue(command.endswith(' --url redis://127.0.0.1:25025 signup"'))
+
+    def test_shared_workers_also_take_signup_jobs(self):
+        command = self.rendered_supervisor_config()["program:worker"]["command"]
+
+        self.assertIn(" signup high default low", command)
+
+    def test_signup_worker_is_in_the_agent_group(self):
+        programs = self.rendered_supervisor_config()["group:agent"]["programs"]
+
+        self.assertIn("worker_signup", [program.strip() for program in programs.split(",")])