When the connection queue is full, respond with a 503 error

The 503 responses are run in a thread

Author: pythonspeed

cheroot/server.py

@@ -101,6 +101,38 @@
 )
 
 
+if sys.version_info[:2] >= (3, 13):
+    from queue import (
+        Queue as QueueWithShutdown,
+        ShutDown as QueueShutDown,
+    )
+else:
+
+    class QueueShutDown(Exception):
+        """Queue has been shut down."""
+
+    class QueueWithShutdown(queue.Queue):
+        """Equivalent to Python 3.13+ Queue."""
+
+        # Unique identifier used to indicate the queue is shut down().
+        _SHUT_DOWN_TOKEN = object()
+
+        def shutdown(self, immediate=False):
+            if immediate:
+                while True:
+                    try:
+                        self.get_nowait()
+                    except queue.Empty:
+                        break
+            self.put(self._SHUT_DOWN_TOKEN)
+
+        def get(self, *args, **kwargs):
+            result = super(QueueWithShutdown, self).get(*args, **kwargs)
+            if result is self._SHUT_DOWN_TOKEN:
+                raise QueueShutDown
+            return result
+
+
 IS_WINDOWS = platform.system() == 'Windows'
 """Flag indicating whether the app is running under Windows."""
 
@@ -1658,6 +1690,8 @@ def __init__(
         self.reuse_port = reuse_port
         self.clear_stats()
 
+        self._unservicable_conns = QueueWithShutdown()
+
     def clear_stats(self):
         """Reset server stat counters.."""
         self._start_time = None
@@ -1866,8 +1900,39 @@ def prepare(self):  # noqa: C901  # FIXME
         self.ready = True
         self._start_time = time.time()
 
+    def _serve_unservicable(self):
+        """Serve connections we can't handle a 503."""
+        while self.ready:
+            try:
+                conn = self._unservicable_conns.get()
+            except QueueShutDown:
+                return
+            request = HTTPRequest(self, conn)
+            try:
+                request.simple_response('503 Service Unavailable')
+            except (socket.error, errors.FatalSSLAlert):
+                # We're sending the 503 error to be polite, it it fails that's
+                # fine.
+                continue
+            except Exception as ex:
+                # We can't just raise an exception because that will kill this
+                # thread, and prevent 503 errors from being sent to future
+                # connections.
+                self.server.error_log(
+                    repr(ex),
+                    level=logging.ERROR,
+                    traceback=True,
+                )
+            conn.linger = True
+            conn.close()
+
     def serve(self):
         """Serve requests, after invoking :func:`prepare()`."""
+        # This thread will handle unservicable connections, as added to
+        # self._unservicable_conns queue. It will run forever, until
+        # self.stop() tells it to shut down.
+        threading.Thread(target=self._serve_unservicable).start()
+
         while self.ready and not self.interrupt:
             try:
                 self._connections.run(self.expiration_interval)
@@ -2162,8 +2227,7 @@ def process_conn(self, conn):
         try:
             self.requests.put(conn)
         except queue.Full:
-            # Just drop the conn. TODO: write 503 back?
-            conn.close()
+            self._unservicable_conns.put(conn)
 
     @property
     def interrupt(self):
@@ -2201,6 +2265,11 @@ def stop(self):  # noqa: C901  # FIXME
             return  # already stopped
 
         self.ready = False
+
+        # This tells the thread that handles unservicable connections to shut
+        # down:
+        self._unservicable_conns.shutdown(immediate=True)
+
         if self._start_time is not None:
             self._run_time += time.time() - self._start_time
         self._start_time = None

cheroot/test/test_server.py

@@ -8,6 +8,7 @@
 import types
 import urllib.parse  # noqa: WPS301
 import uuid
+from http import HTTPStatus
 
 import pytest
 
@@ -570,3 +571,36 @@ def test_threadpool_multistart_validation(monkeypatch):
         match='Threadpools can only be started once.',
     ):
         tp.start()
+
+
+def test_overload_results_in_suitable_http_error(request):
+    """A server that can't keep up with requests returns a 503 HTTP error."""
+    localhost = '127.0.0.1'
+    httpserver = HTTPServer(
+        bind_addr=(localhost, EPHEMERAL_PORT),
+        gateway=Gateway,
+    )
+    # Can only handle on request in parallel:
+    httpserver.requests = ThreadPool(
+        min=1,
+        max=1,
+        accepted_queue_size=1,
+        accepted_queue_timeout=0,
+        server=httpserver,
+    )
+
+    httpserver.prepare()
+    serve_thread = threading.Thread(target=httpserver.serve)
+    serve_thread.start()
+    request.addfinalizer(httpserver.stop)
+    # Stop the thread pool to ensure the queue fills up:
+    httpserver.requests.stop()
+
+    _host, port = httpserver.bind_addr
+
+    # Use up the very limited thread pool queue we've set up, so future
+    # requests fail:
+    httpserver.requests._queue.put(None)
+
+    response = requests.get(f'http://{localhost}:{port}', timeout=20)
+    assert response.status_code == HTTPStatus.SERVICE_UNAVAILABLE