From 65362716fdafe46da1ea63412ec46d70e903a1d8 Mon Sep 17 00:00:00 2001 From: Alex Sohn <44201357+alexsohn1126@users.noreply.github.com> Date: Fri, 23 Oct 2026 12:20:08 -0400 Subject: [PATCH 1/7] feat(taskworker): Restore autoreload for the local taskworker (#128564) When you change backend code while running Sentry locally, the web server picks up the change on its own, but the taskworker doesn't. You have to stop and restart the worker by hand, which is easy to forget, and then tasks quietly run the old code. It's also why people end up running a separate worker with `devservices foreground taskworker` instead of the one the devserver starts. The worker used to have an autoreload option, but it was removed in #96486. That option used Django's reloader, which runs your code on a background thread and keeps the main thread for watching files. Once the taskworker started setting up its own handlers for shutdown signals, it broke, because Python only lets the main thread do that. The worker crashed on startup, so the option was taken out. This brings `--autoreload` back with a different setup. A small parent process watches the Python files and does nothing else. The worker runs as its own separate process, so it has its main thread again and can set up its signal handlers like normal. When a file changes, the parent asks the worker to shut down the usual way, waits for it to finish, and starts a fresh one. The devserver now passes the flag to its worker whenever reload is on (the default), and the devservices taskworker programs use it too. Production doesn't pass the flag, so nothing changes there. Saving several times in quick succession needed one more piece. A save could land while the new worker was still starting up, before it had set up its signal handlers, and the stop request then printed a KeyboardInterrupt traceback. Now the parent waits for the new worker to say it's done starting before it restarts it again. Saves made during that wait still cause one more restart afterwards, so nothing is missed. If the worker crashes while starting, say from a typo, the parent notices it's gone and stops waiting, so the next save brings it back. The file watching comes from `watchfiles`, which is already installed because our web server's reload uses it. getsentry needs a matching one-line config change, which should land after this. ai slop diagrams: before: image after: image --- devservices/config.yml | 4 +- src/sentry/runner/commands/devserver.py | 14 +++++- src/sentry/runner/commands/run.py | 57 +++++++++++++++++++++++-- 3 files changed, 68 insertions(+), 7 deletions(-) diff --git a/devservices/config.yml b/devservices/config.yml index 514bb94432d7..8bfa916cfe5f 100644 --- a/devservices/config.yml +++ b/devservices/config.yml @@ -400,9 +400,9 @@ x-programs: devserver: command: sentry devserver taskworker: - command: sentry run taskworker + command: sentry run taskworker --autoreload taskworker-sentry: - command: sentry run taskworker --rpc-host localhost:50052 + command: sentry run taskworker --autoreload --rpc-host localhost:50052 taskworker-scheduler: command: sentry run taskworker-scheduler ingest-events: diff --git a/src/sentry/runner/commands/devserver.py b/src/sentry/runner/commands/devserver.py index b218b58c25e5..0e524fab267d 100644 --- a/src/sentry/runner/commands/devserver.py +++ b/src/sentry/runner/commands/devserver.py @@ -39,6 +39,13 @@ def _get_daemon(name: str) -> tuple[str, list[str]]: return name, _DEFAULT_DAEMONS[name] +def _get_taskworker_daemon(reload: bool) -> tuple[str, list[str]]: + name, command = _get_daemon("taskworker") + if reload: + command = [*command, "--autoreload"] + return name, command + + @click.command() @click.option( "--reload/--no-reload", @@ -303,7 +310,7 @@ def devserver( "Disable TASKWORKER_ALWAYS_EAGER in your settings file to spawn workers." ) - daemons.append(_get_daemon("taskworker")) + daemons.append(_get_taskworker_daemon(reload)) from sentry import eventstream @@ -504,7 +511,10 @@ def devserver( control_services.append("taskworker-scheduler") for service in control_services: - name, cmd = _get_daemon(service) + if service == "taskworker": + name, cmd = _get_taskworker_daemon(reload) + else: + name, cmd = _get_daemon(service) name = f"control.{name}" quiet = bool( name not in (settings.DEVSERVER_LOGS_ALLOWLIST or ()) diff --git a/src/sentry/runner/commands/run.py b/src/sentry/runner/commands/run.py index fef6376bbb29..4020b0782b8a 100644 --- a/src/sentry/runner/commands/run.py +++ b/src/sentry/runner/commands/run.py @@ -1,19 +1,24 @@ from __future__ import annotations import logging +import multiprocessing import os import random import signal import time +from multiprocessing.synchronize import Event from typing import Any import click import taskbroker_client.constants as taskworker_constants +import watchfiles from taskbroker_client.scheduler import RunStorage, ScheduleRunner +import sentry from sentry import options as sentry_options from sentry.bgtasks.api import managed_bgtasks from sentry.runner.decorators import configuration, log_options +from sentry.runner.settings import discover_configs from sentry.utils.kafka import run_processor_with_signals from sentry.viewer_context import ( Actor, @@ -234,15 +239,61 @@ def taskworker_scheduler(redis_cluster: str, **options: Any) -> None: default=None, type=int, ) +@click.option( + "--autoreload", + is_flag=True, + default=False, + help="Restart the worker when Python files change. For local development only.", +) @log_options() @configuration -def taskworker(**options: Any) -> None: +def taskworker(autoreload: bool, **options: Any) -> None: """ Run a taskworker worker """ os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "0" - # TODO(mark) restore autoreload - run_taskworker(**options) + if autoreload: + _run_taskworker_with_autoreload(**options) + else: + run_taskworker(**options) + + +def _run_taskworker_with_autoreload(**options: Any) -> None: + # The worker registers signal handlers, so it must own the main thread of + # its own process. watchfiles restarts that process with SIGINT on change. + configured = multiprocessing.get_context("spawn").Event() + watchfiles.run_process( + os.path.dirname(sentry.__file__), + discover_configs()[0], + target=_run_reloaded_taskworker, + kwargs={"configured": configured, **options}, + callback=lambda _: _wait_for_reloaded_taskworker(configured), + watch_filter=watchfiles.PythonFilter(), + ) + + +def _wait_for_reloaded_taskworker(configured: Event) -> None: + # A SIGINT during startup prints a KeyboardInterrupt traceback, so let the + # worker finish configuring before watchfiles restarts it. + while not configured.wait(timeout=0.1): + if not multiprocessing.active_children(): + break + configured.clear() + + +def _run_reloaded_taskworker(configured: Event, **options: Any) -> None: + # Runs in a fresh spawned process, which has to configure Sentry again. + from sentry.runner import configure + + try: + configure() + finally: + configured.set() + try: + run_taskworker(**options) + except KeyboardInterrupt: + # The restart arrived before the worker registered its SIGINT handler. + pass def run_taskworker( From c59eb618fff6f1dbbf1b591cb125f7b7d2f3d644 Mon Sep 17 00:00:00 2001 From: Alex Sohn Date: Fri, 23 Oct 2026 13:35:46 -0400 Subject: [PATCH 2/7] fix(taskworker): Import watchfiles only when autoreload is on watchfiles imports anyio, which tries to import sniffio. test_devimports blocks sniffio with a plain ImportError, which anyio doesn't catch, so importing watchfiles at CLI startup broke test_startup_imports[sentry]. Import it inside the autoreload path instead. --- src/sentry/runner/commands/run.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/src/sentry/runner/commands/run.py b/src/sentry/runner/commands/run.py index 4020b0782b8a..7c837a4c58d4 100644 --- a/src/sentry/runner/commands/run.py +++ b/src/sentry/runner/commands/run.py @@ -11,14 +11,11 @@ import click import taskbroker_client.constants as taskworker_constants -import watchfiles from taskbroker_client.scheduler import RunStorage, ScheduleRunner -import sentry from sentry import options as sentry_options from sentry.bgtasks.api import managed_bgtasks from sentry.runner.decorators import configuration, log_options -from sentry.runner.settings import discover_configs from sentry.utils.kafka import run_processor_with_signals from sentry.viewer_context import ( Actor, @@ -259,6 +256,12 @@ def taskworker(autoreload: bool, **options: Any) -> None: def _run_taskworker_with_autoreload(**options: Any) -> None: + # watchfiles pulls in anyio, which CLI startup must not import. + import watchfiles + + import sentry + from sentry.runner.settings import discover_configs + # The worker registers signal handlers, so it must own the main thread of # its own process. watchfiles restarts that process with SIGINT on change. configured = multiprocessing.get_context("spawn").Event() From be321e3e8ef754b298163742903d19582747c5f6 Mon Sep 17 00:00:00 2001 From: Alex Sohn Date: Fri, 23 Oct 2026 14:28:15 -0400 Subject: [PATCH 3/7] fix(taskworker): Wait for the worker's SIGINT handler before restarting The reloaded worker reported ready right after configure(), but it only registers its SIGINT handler later in start(). During a burst of saves the restart SIGINT landed in that gap, so the worker skipped its shutdown, hung waiting on its children, and was SIGKILLed by watchfiles, leaving the children orphaned. Report ready once the handler is in place instead. --- src/sentry/runner/commands/run.py | 40 ++++++++++++++++++++----------- 1 file changed, 26 insertions(+), 14 deletions(-) diff --git a/src/sentry/runner/commands/run.py b/src/sentry/runner/commands/run.py index 7c837a4c58d4..f493d32ab000 100644 --- a/src/sentry/runner/commands/run.py +++ b/src/sentry/runner/commands/run.py @@ -5,6 +5,7 @@ import os import random import signal +import threading import time from multiprocessing.synchronize import Event from typing import Any @@ -264,41 +265,52 @@ def _run_taskworker_with_autoreload(**options: Any) -> None: # The worker registers signal handlers, so it must own the main thread of # its own process. watchfiles restarts that process with SIGINT on change. - configured = multiprocessing.get_context("spawn").Event() + ready = multiprocessing.get_context("spawn").Event() watchfiles.run_process( os.path.dirname(sentry.__file__), discover_configs()[0], target=_run_reloaded_taskworker, - kwargs={"configured": configured, **options}, - callback=lambda _: _wait_for_reloaded_taskworker(configured), + kwargs={"ready": ready, **options}, + callback=lambda _: _wait_for_reloaded_taskworker(ready), watch_filter=watchfiles.PythonFilter(), ) -def _wait_for_reloaded_taskworker(configured: Event) -> None: - # A SIGINT during startup prints a KeyboardInterrupt traceback, so let the - # worker finish configuring before watchfiles restarts it. - while not configured.wait(timeout=0.1): +def _wait_for_reloaded_taskworker(ready: Event) -> None: + # A SIGINT before the worker registers its handler skips the shutdown that + # stops its children, which are then orphaned when watchfiles kills it. + # So hold off restarting until the handler is in place. + while not ready.wait(timeout=0.1): if not multiprocessing.active_children(): break - configured.clear() + ready.clear() -def _run_reloaded_taskworker(configured: Event, **options: Any) -> None: +def _run_reloaded_taskworker(ready: Event, **options: Any) -> None: # Runs in a fresh spawned process, which has to configure Sentry again. from sentry.runner import configure - try: - configure() - finally: - configured.set() + threading.Thread( + target=_set_when_sigint_handled, + args=(ready, signal.getsignal(signal.SIGINT)), + daemon=True, + ).start() + configure() try: run_taskworker(**options) except KeyboardInterrupt: - # The restart arrived before the worker registered its SIGINT handler. + # A Ctrl-C arrived before the worker registered its SIGINT handler. pass +def _set_when_sigint_handled(ready: Event, initial_handler: Any) -> None: + # The worker registers its handler deep inside start(), with no hook to + # tell us, so watch for it to replace the initial one. + while signal.getsignal(signal.SIGINT) is initial_handler: + time.sleep(0.05) + ready.set() + + def run_taskworker( push_mode: bool, batch_push_mode: bool, From 8c77d79f3613bf3db99e2d7ad359a30b10432d44 Mon Sep 17 00:00:00 2001 From: Alex Sohn Date: Fri, 23 Oct 2026 14:56:35 -0400 Subject: [PATCH 4/7] fix(taskworker): Exit quietly when autoreload is stopped mid-restart Stopping the autoreloading worker while a new one was still starting printed KeyboardInterrupt tracebacks and made the parent exit 1. Catch the interrupt around configure() as well as the worker run, ignore the second SIGINT watchfiles sends during that exit, and treat an interrupt while waiting on a restart as a normal stop. --- src/sentry/runner/commands/run.py | 26 ++++++++++++++++---------- 1 file changed, 16 insertions(+), 10 deletions(-) diff --git a/src/sentry/runner/commands/run.py b/src/sentry/runner/commands/run.py index f493d32ab000..662298dfa767 100644 --- a/src/sentry/runner/commands/run.py +++ b/src/sentry/runner/commands/run.py @@ -266,14 +266,19 @@ def _run_taskworker_with_autoreload(**options: Any) -> None: # The worker registers signal handlers, so it must own the main thread of # its own process. watchfiles restarts that process with SIGINT on change. ready = multiprocessing.get_context("spawn").Event() - watchfiles.run_process( - os.path.dirname(sentry.__file__), - discover_configs()[0], - target=_run_reloaded_taskworker, - kwargs={"ready": ready, **options}, - callback=lambda _: _wait_for_reloaded_taskworker(ready), - watch_filter=watchfiles.PythonFilter(), - ) + try: + watchfiles.run_process( + os.path.dirname(sentry.__file__), + discover_configs()[0], + target=_run_reloaded_taskworker, + kwargs={"ready": ready, **options}, + callback=lambda _: _wait_for_reloaded_taskworker(ready), + watch_filter=watchfiles.PythonFilter(), + ) + except KeyboardInterrupt: + # Stopped while waiting on a restart. watchfiles has already stopped + # the worker on its way out. + pass def _wait_for_reloaded_taskworker(ready: Event) -> None: @@ -295,12 +300,13 @@ def _run_reloaded_taskworker(ready: Event, **options: Any) -> None: args=(ready, signal.getsignal(signal.SIGINT)), daemon=True, ).start() - configure() try: + configure() run_taskworker(**options) except KeyboardInterrupt: # A Ctrl-C arrived before the worker registered its SIGINT handler. - pass + # Ignore the second SIGINT watchfiles sends while we exit. + signal.signal(signal.SIGINT, signal.SIG_IGN) def _set_when_sigint_handled(ready: Event, initial_handler: Any) -> None: From ae7f843a7cb25723e890d7917bf639f1b910dd2b Mon Sep 17 00:00:00 2001 From: Alex Sohn Date: Fri, 23 Oct 2026 15:54:31 -0400 Subject: [PATCH 5/7] ref(taskworker): Read autoreload from options instead of the signature --- src/sentry/runner/commands/run.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/sentry/runner/commands/run.py b/src/sentry/runner/commands/run.py index 662298dfa767..3509fc1f53fc 100644 --- a/src/sentry/runner/commands/run.py +++ b/src/sentry/runner/commands/run.py @@ -245,12 +245,12 @@ def taskworker_scheduler(redis_cluster: str, **options: Any) -> None: ) @log_options() @configuration -def taskworker(autoreload: bool, **options: Any) -> None: +def taskworker(**options: Any) -> None: """ Run a taskworker worker """ os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "0" - if autoreload: + if options.pop("autoreload"): _run_taskworker_with_autoreload(**options) else: run_taskworker(**options) From e146ae3146b8239995e23618eb4f4a0958ecc457 Mon Sep 17 00:00:00 2001 From: Alex Sohn Date: Fri, 23 Oct 2026 15:55:48 -0400 Subject: [PATCH 6/7] ref(taskworker): Note what happens to long tasks on autoreload --- src/sentry/runner/commands/run.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/sentry/runner/commands/run.py b/src/sentry/runner/commands/run.py index 3509fc1f53fc..5d8cd7c2ae52 100644 --- a/src/sentry/runner/commands/run.py +++ b/src/sentry/runner/commands/run.py @@ -265,6 +265,7 @@ def _run_taskworker_with_autoreload(**options: Any) -> None: # The worker registers signal handlers, so it must own the main thread of # its own process. watchfiles restarts that process with SIGINT on change. + # Known gap: a task still running 5s after SIGINT orphans its child and gets redelivered. ready = multiprocessing.get_context("spawn").Event() try: watchfiles.run_process( From bdebcaac6f994edff4711670ad42f06f3ab9ccad Mon Sep 17 00:00:00 2001 From: Alex Sohn Date: Fri, 23 Oct 2026 15:56:45 -0400 Subject: [PATCH 7/7] ref(taskworker): Note what happens to long tasks on autoreload --- src/sentry/runner/commands/run.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/sentry/runner/commands/run.py b/src/sentry/runner/commands/run.py index 5d8cd7c2ae52..1cd92f0afedd 100644 --- a/src/sentry/runner/commands/run.py +++ b/src/sentry/runner/commands/run.py @@ -263,8 +263,7 @@ def _run_taskworker_with_autoreload(**options: Any) -> None: import sentry from sentry.runner.settings import discover_configs - # The worker registers signal handlers, so it must own the main thread of - # its own process. watchfiles restarts that process with SIGINT on change. + # Run the worker in its own process so its signal handlers get the main thread. # Known gap: a task still running 5s after SIGINT orphans its child and gets redelivered. ready = multiprocessing.get_context("spawn").Event() try: