Fixed #31515 -- Made ASGIHandler dispatch lifecycle signals with thread sensitive.

2 files changed, 40 insertions, 3 deletions

diff --git a/django/core/handlers/asgi.py b/django/core/handlers/asgi.py
index 82d2e1ab9d..7fbabe4510 100644
--- a/django/core/handlers/asgi.py
+++ b/django/core/handlers/asgi.py
@@ -151,7 +151,7 @@ class ASGIHandler(base.BaseHandler):
             return
         # Request is complete and can be served.
         set_script_prefix(self.get_script_prefix(scope))
-        await sync_to_async(signals.request_started.send)(sender=self.__class__, scope=scope)
+        await sync_to_async(signals.request_started.send, thread_sensitive=True)(sender=self.__class__, scope=scope)
         # Get the request and check for basic issues.
         request, error_response = self.create_request(scope, body_file)
         if request is None:
@@ -259,7 +259,7 @@ class ASGIHandler(base.BaseHandler):
                     'body': chunk,
                     'more_body': not last,
                 })
-        await sync_to_async(response.close)()
+        await sync_to_async(response.close, thread_sensitive=True)()
 
     @classmethod
     def chunk_bytes(cls, data):
diff --git a/tests/asgi/tests.py b/tests/asgi/tests.py
index c123f027fb..41b39ec9ea 100644
--- a/tests/asgi/tests.py
+++ b/tests/asgi/tests.py
@@ -1,11 +1,13 @@
 import asyncio
 import sys
+import threading
 from unittest import skipIf
 
+from asgiref.sync import SyncToAsync
 from asgiref.testing import ApplicationCommunicator
 
 from django.core.asgi import get_asgi_application
-from django.core.signals import request_started
+from django.core.signals import request_finished, request_started
 from django.db import close_old_connections
 from django.test import AsyncRequestFactory, SimpleTestCase, override_settings
 
@@ -151,3 +153,38 @@ class ASGITest(SimpleTestCase):
         response_body = await communicator.receive_output()
         self.assertEqual(response_body['type'], 'http.response.body')
         self.assertEqual(response_body['body'], b'')
+
+    async def test_request_lifecycle_signals_dispatched_with_thread_sensitive(self):
+        class SignalHandler:
+            """Track threads handler is dispatched on."""
+            threads = []
+
+            def __call__(self, **kwargs):
+                self.threads.append(threading.current_thread())
+
+        signal_handler = SignalHandler()
+        request_started.connect(signal_handler)
+        request_finished.connect(signal_handler)
+
+        # Perform a basic request.
+        application = get_asgi_application()
+        scope = self.async_request_factory._base_scope(path='/')
+        communicator = ApplicationCommunicator(application, scope)
+        await communicator.send_input({'type': 'http.request'})
+        response_start = await communicator.receive_output()
+        self.assertEqual(response_start['type'], 'http.response.start')
+        self.assertEqual(response_start['status'], 200)
+        response_body = await communicator.receive_output()
+        self.assertEqual(response_body['type'], 'http.response.body')
+        self.assertEqual(response_body['body'], b'Hello World!')
+        # Give response.close() time to finish.
+        await communicator.wait()
+
+        # At this point, AsyncToSync does not have a current executor. Thus
+        # SyncToAsync falls-back to .single_thread_executor.
+        target_thread = next(iter(SyncToAsync.single_thread_executor._threads))
+        request_started_thread, request_finished_thread = signal_handler.threads
+        self.assertEqual(request_started_thread, target_thread)
+        self.assertEqual(request_finished_thread, target_thread)
+        request_started.disconnect(signal_handler)
+        request_finished.disconnect(signal_handler)