Move SSL handshake from main thread to worker thread

Port fix from gunicorn PR #3440. Previously, conn.init() (which does
the SSL handshake via ssl_wrap_socket) ran on the main event loop thread
in enqueue_req() with no error handling. A failed TLS handshake from any
client (port scanners, browsers rejecting self-signed certs, connection
resets) would crash the entire worker process and all its in-flight
connections.

Now conn.init() runs inside handle() in the thread pool, where
ssl.SSLError, OSError, and Exception are already caught. A failed
handshake only affects that one connection.

Also:
- Add double-init guard to TConn.init() for keepalive safety
- Set initialized=True only after success, so partial init failures
  propagate the real error instead of a confusing AssertionError
- Set setblocking(True) explicitly in handle() for keepalive
  connections where finish_request switched to non-blocking

Author: davegaeddert

plain/plain/server/workers/thread.py

@@ -64,8 +64,8 @@ def __init__(
         self.sock.setblocking(False)
 
     def init(self) -> None:
-        self.initialized = True
-        self.sock.setblocking(True)
+        if self.initialized:
+            return
 
         if self.parser is None:
             # wrap the socket if needed
@@ -75,6 +75,8 @@ def init(self) -> None:
             # initialize the parser
             self.parser = http.RequestParser(self.cfg, self.sock, self.client)
 
+        self.initialized = True
+
     def set_timeout(self) -> None:
         # set the timeout
         self.timeout = time.time() + KEEPALIVE
@@ -130,8 +132,9 @@ def _wrap_future(self, fs: futures.Future[tuple[bool, TConn]], conn: TConn) -> N
         fs.add_done_callback(self.finish_request)
 
     def enqueue_req(self, conn: TConn) -> None:
-        conn.init()
-        # submit the connection to a worker
+        # conn.init() is called inside handle(), not here, so that SSL
+        # handshake errors are caught in the worker thread instead of
+        # crashing the main loop. (Ported from gunicorn PR #3440.)
         fs = self.tpool.submit(self.handle, conn)
         self._wrap_future(fs, conn)
 
@@ -303,7 +306,10 @@ def handle(self, conn: TConn) -> tuple[bool, TConn]:
         keepalive = False
         req = None
         try:
-            # conn.parser is guaranteed to be initialized by enqueue_req -> conn.init()
+            # Ensure blocking mode before init/parsing. Critical for keepalive
+            # connections where finish_request sets non-blocking for the poller.
+            conn.sock.setblocking(True)
+            conn.init()
             assert conn.parser is not None
             req = next(conn.parser)
             if not req: