We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN 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


        

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `authentik/root/celery.py`
Content:
```
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 FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

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)

{"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"}], "after_files": [{"content": "\"\"\"authentik core celery\"\"\"\nimport os\nfrom contextvars import ContextVar\nfrom logging.config import dictConfig\nfrom pathlib import Path\nfrom tempfile import gettempdir\nfrom typing import Callable\n\nfrom celery import Celery, bootsteps\nfrom celery.apps.worker import Worker\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)\nHEARTBEAT_FILE = Path(gettempdir() + \"/authentik-worker\")\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\nclass 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#   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()\nCELERY_APP.steps[\"worker\"].add(LivenessProbe)\n", "path": "authentik/root/celery.py"}]}