From c74c4035071cda35039074bd84b6b0669297b0bb Mon Sep 17 00:00:00 2001 From: regdocs <55305804+regdocs@users.noreply.github.com> Date: Fri, 30 Oct 2026 16:51:12 +0530 Subject: [PATCH] feat(job): run pool handover jobs on a dedicated queue and worker A trial signup that gets a standby site waits for four short jobs: Add Domain and Update Site Configuration on the app server, and Add Domain to Upstream and Setup Redirects on Hosts on the proxy. Each runs in a second or two, but they shared the queues with New Site, Update Site Migrate and Install App, which take minutes. When long jobs held every worker, the signup waited minutes for a free one. Move these jobs to a new `signup` queue with a dedicated `worker_signup` supervisor program that listens only on that queue, so a long job can never hold it. The shared workers list `signup` first, so they also take signup jobs when free. The worker count comes from `signup_workers` in the agent config, defaulting to 1. Co-Authored-By: Claude Opus 5.5 --- Procfile | 5 ++- agent/proxy.py | 4 +- agent/server.py | 13 +++++- agent/site.py | 4 +- agent/templates/agent/supervisor.conf.jinja2 | 20 ++++++++- agent/tests/test_server.py | 47 ++++++++++++++++++++ 6 files changed, 84 insertions(+), 9 deletions(-) diff --git a/Procfile b/Procfile index d79de9e89..620e6817a 100755 --- a/Procfile +++ b/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' diff --git a/agent/proxy.py b/agent/proxy.py index b004ce431..7da465d0b 100644 --- a/agent/proxy.py +++ b/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: diff --git a/agent/server.py b/agent/server.py index f931babc8..0f9de5433 100644 --- a/agent/server.py +++ b/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"), diff --git a/agent/site.py b/agent/site.py index 16e559cc6..2064106bd 100644 --- a/agent/site.py +++ b/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) diff --git a/agent/templates/agent/supervisor.conf.jinja2 b/agent/templates/agent/supervisor.conf.jinja2 index 4095f05c0..2f280bff1 100644 --- a/agent/templates/agent/supervisor.conf.jinja2 +++ b/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" %} diff --git a/agent/tests/test_server.py b/agent/tests/test_server.py index d2f7800e3..a9aade277 100644 --- a/agent/tests/test_server.py +++ b/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(",")])