prompt: You will be provided with a partial code base and an issue statement explaining a problem to resolve.
  <issue>
  Celery CPU usage 100% on new install
  **Describe the bug**
  
  I'm using the attached docker-compose and env file (don't worry about the secrets in the env file).
  
  [env.txt](https://github.com/goauthentik/authentik/files/10758594/env.txt)
  [docker-compose.yml.txt](https://github.com/goauthentik/authentik/files/10758616/docker-compose.yml.txt)
  
  
  **To Reproduce**
  
  Run `docker-compose up` with the default configuration. Watch as celery uses 100% of a single CPU core. I've followed the setup guide from scratch twice to make sure I was getting as close as possible to a default install.
  
  **Logs**
  <details>
      <summary>Stacktrace from authentik</summary>
  
  ```
  Traceback (most recent call last):
  django.db.utils.InterfaceError: connection already closed
  ```
  </details>
  
  
  **Version and Deployment (please complete the following information):**
  - authentik version: 2023.2.2
  - Deployment: docker compose
  
  
          
  
  </issue>
  <code>
  [start of authentik/root/celery.py]
  1 """authentik core celery"""
  2 import os
  3 from contextvars import ContextVar
  4 from logging.config import dictConfig
  5 from typing import Callable
  6 
  7 from celery import Celery
  8 from celery.signals import (
  9     after_task_publish,
  10     setup_logging,
  11     task_failure,
  12     task_internal_error,
  13     task_postrun,
  14     task_prerun,
  15     worker_ready,
  16 )
  17 from django.conf import settings
  18 from django.db import ProgrammingError
  19 from structlog.contextvars import STRUCTLOG_KEY_PREFIX
  20 from structlog.stdlib import get_logger
  21 
  22 from authentik.lib.sentry import before_send
  23 from authentik.lib.utils.errors import exception_to_string
  24 
  25 # set the default Django settings module for the 'celery' program.
  26 os.environ.setdefault("DJANGO_SETTINGS_MODULE", "authentik.root.settings")
  27 
  28 LOGGER = get_logger()
  29 CELERY_APP = Celery("authentik")
  30 CTX_TASK_ID = ContextVar(STRUCTLOG_KEY_PREFIX + "task_id", default=Ellipsis)
  31 
  32 
  33 @setup_logging.connect
  34 def config_loggers(*args, **kwargs):
  35     """Apply logging settings from settings.py to celery"""
  36     dictConfig(settings.LOGGING)
  37 
  38 
  39 @after_task_publish.connect
  40 def after_task_publish_hook(sender=None, headers=None, body=None, **kwargs):
  41     """Log task_id after it was published"""
  42     info = headers if "task" in headers else body
  43     LOGGER.info("Task published", task_id=info.get("id", ""), task_name=info.get("task", ""))
  44 
  45 
  46 @task_prerun.connect
  47 def task_prerun_hook(task_id: str, task, *args, **kwargs):
  48     """Log task_id on worker"""
  49     request_id = "task-" + task_id.replace("-", "")
  50     CTX_TASK_ID.set(request_id)
  51     LOGGER.info("Task started", task_id=task_id, task_name=task.__name__)
  52 
  53 
  54 @task_postrun.connect
  55 def task_postrun_hook(task_id, task, *args, retval=None, state=None, **kwargs):
  56     """Log task_id on worker"""
  57     CTX_TASK_ID.set(...)
  58     LOGGER.info("Task finished", task_id=task_id, task_name=task.__name__, state=state)
  59 
  60 
  61 @task_failure.connect
  62 @task_internal_error.connect
  63 def task_error_hook(task_id, exception: Exception, traceback, *args, **kwargs):
  64     """Create system event for failed task"""
  65     from authentik.events.models import Event, EventAction
  66 
  67     LOGGER.warning("Task failure", exc=exception)
  68     CTX_TASK_ID.set(...)
  69     if before_send({}, {"exc_info": (None, exception, None)}) is not None:
  70         Event.new(EventAction.SYSTEM_EXCEPTION, message=exception_to_string(exception)).save()
  71 
  72 
  73 def _get_startup_tasks() -> list[Callable]:
  74     """Get all tasks to be run on startup"""
  75     from authentik.admin.tasks import clear_update_notifications
  76     from authentik.outposts.tasks import outpost_connection_discovery, outpost_controller_all
  77     from authentik.providers.proxy.tasks import proxy_set_defaults
  78 
  79     return [
  80         clear_update_notifications,
  81         outpost_connection_discovery,
  82         outpost_controller_all,
  83         proxy_set_defaults,
  84     ]
  85 
  86 
  87 @worker_ready.connect
  88 def worker_ready_hook(*args, **kwargs):
  89     """Run certain tasks on worker start"""
  90 
  91     LOGGER.info("Dispatching startup tasks...")
  92     for task in _get_startup_tasks():
  93         try:
  94             task.delay()
  95         except ProgrammingError as exc:
  96             LOGGER.warning("Startup task failed", task=task, exc=exc)
  97     from authentik.blueprints.v1.tasks import start_blueprint_watcher
  98 
  99     start_blueprint_watcher()
  100 
  101 
  102 # Using a string here means the worker doesn't have to serialize
  103 # the configuration object to child processes.
  104 # - namespace='CELERY' means all celery-related configuration keys
  105 #   should have a `CELERY_` prefix.
  106 CELERY_APP.config_from_object(settings, namespace="CELERY")
  107 
  108 # Load task modules from all registered Django app configs.
  109 CELERY_APP.autodiscover_tasks()
  110
  [end of authentik/root/celery.py]
  </code>
  
  I need you to solve this issue by generating a single patch file that I can apply directly to this repository using git apply. Please respond with a single patch file in the following format.
  <patch>
  diff --git a/file.py b/file.py
  --- a/file.py
  +++ b/file.py
  @@ -1,27 +1,35 @@
   def euclidean(a, b):
  -    while b:
  -        a, b = b, a % b
  -    return a
  +    if b == 0:
  +        return a
  +    return euclidean(b, a % b)
   
   
   def bresenham(x0, y0, x1, y1):
       points = []
       dx = abs(x1 - x0)
       dy = abs(y1 - y0)
  -    sx = 1 if x0 < x1 else -1
  -    sy = 1 if y0 < y1 else -1
  -    err = dx - dy
  +    x, y = x0, y0
  +    sx = -1 if x0 > x1 else 1
  +    sy = -1 if y0 > y1 else 1
   
  -    while True:
  -        points.append((x0, y0))
  -        if x0 == x1 and y0 == y1:
  -            break
  -        e2 = 2 * err
  -        if e2 > -dy:
  -            err -= dy
  -            x0 += sx
  -        if e2 < dx:
  -            err += dx
  -            y0 += sy
  +    if dx > dy:
  +        err = dx / 2.0
  +        while x != x1:
  +            points.append((x, y))
  +            err -= dy
  +            if err < 0:
  +                y += sy
  +                err += dx
  +            x += sx
  +    else:
  +        err = dy / 2.0
  +        while y != y1:
  +            points.append((x, y))
  +            err -= dx
  +            if err < 0:
  +                x += sx
  +                err += dy
  +            y += sy
  +
  +    points.append((x, y))
       return points
  
  </patch>
golden_diff: diff --git a/authentik/root/celery.py b/authentik/root/celery.py
  --- a/authentik/root/celery.py
  +++ b/authentik/root/celery.py
  @@ -2,9 +2,12 @@
   import os
   from contextvars import ContextVar
   from logging.config import dictConfig
  +from pathlib import Path
  +from tempfile import gettempdir
   from typing import Callable
   
  -from celery import Celery
  +from celery import Celery, bootsteps
  +from celery.apps.worker import Worker
   from celery.signals import (
       after_task_publish,
       setup_logging,
  @@ -28,6 +31,7 @@
   LOGGER = get_logger()
   CELERY_APP = Celery("authentik")
   CTX_TASK_ID = ContextVar(STRUCTLOG_KEY_PREFIX + "task_id", default=Ellipsis)
  +HEARTBEAT_FILE = Path(gettempdir() + "/authentik-worker")
   
   
   @setup_logging.connect
  @@ -99,6 +103,33 @@
       start_blueprint_watcher()
   
   
  +class LivenessProbe(bootsteps.StartStopStep):
  +    """Add a timed task to touch a temporary file for healthchecking reasons"""
  +
  +    requires = {"celery.worker.components:Timer"}
  +
  +    def __init__(self, parent, **kwargs):
  +        super().__init__(parent, **kwargs)
  +        self.requests = []
  +        self.tref = None
  +
  +    def start(self, parent: Worker):
  +        self.tref = parent.timer.call_repeatedly(
  +            10.0,
  +            self.update_heartbeat_file,
  +            (parent,),
  +            priority=10,
  +        )
  +        self.update_heartbeat_file(parent)
  +
  +    def stop(self, parent: Worker):
  +        HEARTBEAT_FILE.unlink(missing_ok=True)
  +
  +    def update_heartbeat_file(self, worker: Worker):
  +        """Touch heartbeat file"""
  +        HEARTBEAT_FILE.touch()
  +
  +
   # Using a string here means the worker doesn't have to serialize
   # the configuration object to child processes.
   # - namespace='CELERY' means all celery-related configuration keys
  @@ -107,3 +138,4 @@
   
   # Load task modules from all registered Django app configs.
   CELERY_APP.autodiscover_tasks()
  +CELERY_APP.steps["worker"].add(LivenessProbe)
  
verification_info: {"golden_diff": "diff --git a/authentik/root/celery.py b/authentik/root/celery.py\n--- a/authentik/root/celery.py\n+++ b/authentik/root/celery.py\n@@ -2,9 +2,12 @@\n import os\n from contextvars import ContextVar\n from logging.config import dictConfig\n+from pathlib import Path\n+from tempfile import gettempdir\n from typing import Callable\n \n-from celery import Celery\n+from celery import Celery, bootsteps\n+from celery.apps.worker import Worker\n from celery.signals import (\n     after_task_publish,\n     setup_logging,\n@@ -28,6 +31,7 @@\n LOGGER = get_logger()\n CELERY_APP = Celery(\"authentik\")\n CTX_TASK_ID = ContextVar(STRUCTLOG_KEY_PREFIX + \"task_id\", default=Ellipsis)\n+HEARTBEAT_FILE = Path(gettempdir() + \"/authentik-worker\")\n \n \n @setup_logging.connect\n@@ -99,6 +103,33 @@\n     start_blueprint_watcher()\n \n \n+class LivenessProbe(bootsteps.StartStopStep):\n+    \"\"\"Add a timed task to touch a temporary file for healthchecking reasons\"\"\"\n+\n+    requires = {\"celery.worker.components:Timer\"}\n+\n+    def __init__(self, parent, **kwargs):\n+        super().__init__(parent, **kwargs)\n+        self.requests = []\n+        self.tref = None\n+\n+    def start(self, parent: Worker):\n+        self.tref = parent.timer.call_repeatedly(\n+            10.0,\n+            self.update_heartbeat_file,\n+            (parent,),\n+            priority=10,\n+        )\n+        self.update_heartbeat_file(parent)\n+\n+    def stop(self, parent: Worker):\n+        HEARTBEAT_FILE.unlink(missing_ok=True)\n+\n+    def update_heartbeat_file(self, worker: Worker):\n+        \"\"\"Touch heartbeat file\"\"\"\n+        HEARTBEAT_FILE.touch()\n+\n+\n # Using a string here means the worker doesn't have to serialize\n # the configuration object to child processes.\n # - namespace='CELERY' means all celery-related configuration keys\n@@ -107,3 +138,4 @@\n \n # Load task modules from all registered Django app configs.\n CELERY_APP.autodiscover_tasks()\n+CELERY_APP.steps[\"worker\"].add(LivenessProbe)\n", "issue": "Celery CPU usage 100% on new install\n**Describe the bug**\r\n\r\nI'm using the attached docker-compose and env file (don't worry about the secrets in the env file).\r\n\r\n[env.txt](https://github.com/goauthentik/authentik/files/10758594/env.txt)\r\n[docker-compose.yml.txt](https://github.com/goauthentik/authentik/files/10758616/docker-compose.yml.txt)\r\n\r\n\r\n**To Reproduce**\r\n\r\nRun `docker-compose up` with the default configuration. Watch as celery uses 100% of a single CPU core. I've followed the setup guide from scratch twice to make sure I was getting as close as possible to a default install.\r\n\r\n**Logs**\r\n<details>\r\n    <summary>Stacktrace from authentik</summary>\r\n\r\n```\r\nTraceback (most recent call last):\r\ndjango.db.utils.InterfaceError: connection already closed\r\n```\r\n</details>\r\n\r\n\r\n**Version and Deployment (please complete the following information):**\r\n- authentik version: 2023.2.2\r\n- Deployment: docker compose\r\n\r\n\r\n        \n", "before_files": [{"content": "\"\"\"authentik core celery\"\"\"\nimport os\nfrom contextvars import ContextVar\nfrom logging.config import dictConfig\nfrom typing import Callable\n\nfrom celery import Celery\nfrom celery.signals import (\n    after_task_publish,\n    setup_logging,\n    task_failure,\n    task_internal_error,\n    task_postrun,\n    task_prerun,\n    worker_ready,\n)\nfrom django.conf import settings\nfrom django.db import ProgrammingError\nfrom structlog.contextvars import STRUCTLOG_KEY_PREFIX\nfrom structlog.stdlib import get_logger\n\nfrom authentik.lib.sentry import before_send\nfrom authentik.lib.utils.errors import exception_to_string\n\n# set the default Django settings module for the 'celery' program.\nos.environ.setdefault(\"DJANGO_SETTINGS_MODULE\", \"authentik.root.settings\")\n\nLOGGER = get_logger()\nCELERY_APP = Celery(\"authentik\")\nCTX_TASK_ID = ContextVar(STRUCTLOG_KEY_PREFIX + \"task_id\", default=Ellipsis)\n\n\n@setup_logging.connect\ndef config_loggers(*args, **kwargs):\n    \"\"\"Apply logging settings from settings.py to celery\"\"\"\n    dictConfig(settings.LOGGING)\n\n\n@after_task_publish.connect\ndef after_task_publish_hook(sender=None, headers=None, body=None, **kwargs):\n    \"\"\"Log task_id after it was published\"\"\"\n    info = headers if \"task\" in headers else body\n    LOGGER.info(\"Task published\", task_id=info.get(\"id\", \"\"), task_name=info.get(\"task\", \"\"))\n\n\n@task_prerun.connect\ndef task_prerun_hook(task_id: str, task, *args, **kwargs):\n    \"\"\"Log task_id on worker\"\"\"\n    request_id = \"task-\" + task_id.replace(\"-\", \"\")\n    CTX_TASK_ID.set(request_id)\n    LOGGER.info(\"Task started\", task_id=task_id, task_name=task.__name__)\n\n\n@task_postrun.connect\ndef task_postrun_hook(task_id, task, *args, retval=None, state=None, **kwargs):\n    \"\"\"Log task_id on worker\"\"\"\n    CTX_TASK_ID.set(...)\n    LOGGER.info(\"Task finished\", task_id=task_id, task_name=task.__name__, state=state)\n\n\n@task_failure.connect\n@task_internal_error.connect\ndef task_error_hook(task_id, exception: Exception, traceback, *args, **kwargs):\n    \"\"\"Create system event for failed task\"\"\"\n    from authentik.events.models import Event, EventAction\n\n    LOGGER.warning(\"Task failure\", exc=exception)\n    CTX_TASK_ID.set(...)\n    if before_send({}, {\"exc_info\": (None, exception, None)}) is not None:\n        Event.new(EventAction.SYSTEM_EXCEPTION, message=exception_to_string(exception)).save()\n\n\ndef _get_startup_tasks() -> list[Callable]:\n    \"\"\"Get all tasks to be run on startup\"\"\"\n    from authentik.admin.tasks import clear_update_notifications\n    from authentik.outposts.tasks import outpost_connection_discovery, outpost_controller_all\n    from authentik.providers.proxy.tasks import proxy_set_defaults\n\n    return [\n        clear_update_notifications,\n        outpost_connection_discovery,\n        outpost_controller_all,\n        proxy_set_defaults,\n    ]\n\n\n@worker_ready.connect\ndef worker_ready_hook(*args, **kwargs):\n    \"\"\"Run certain tasks on worker start\"\"\"\n\n    LOGGER.info(\"Dispatching startup tasks...\")\n    for task in _get_startup_tasks():\n        try:\n            task.delay()\n        except ProgrammingError as exc:\n            LOGGER.warning(\"Startup task failed\", task=task, exc=exc)\n    from authentik.blueprints.v1.tasks import start_blueprint_watcher\n\n    start_blueprint_watcher()\n\n\n# Using a string here means the worker doesn't have to serialize\n# the configuration object to child processes.\n# - namespace='CELERY' means all celery-related configuration keys\n#   should have a `CELERY_` prefix.\nCELERY_APP.config_from_object(settings, namespace=\"CELERY\")\n\n# Load task modules from all registered Django app configs.\nCELERY_APP.autodiscover_tasks()\n", "path": "authentik/root/celery.py"}]}