socketengine: run direct TLS in the reactor pool — non-blocking handshake + crypto in the worker threads (Sock::Tls, TlsSession), unifying the client I/O model and spreading TLS work across cores; proxied TLS + links keep the thread path

src/main.rs

@@ -102,6 +102,10 @@ fn main() {
         }
     });
 
+    // reactor worker pool: shared by the plaintext acceptor and the direct-TLS
+    // acceptor, so client I/O (framing + TLS crypto) spreads across cores.
+    let reactors = socketengine::spawn_reactors(tx.clone(), max_line, max_sendq, io_threads);
+
     // optional TLS listener (bind_tls + tls_cert + tls_key). A cert/bind problem
     // disables TLS but never takes the plaintext listener down.
     if let (Some(bind_tls), Some(cert), Some(key)) = (&cfg.bind_tls, &cfg.tls_cert, &cfg.tls_key) {
@@ -113,6 +117,7 @@ fn main() {
                     let tls_tx = tx.clone();
                     let tls_counter = counter.clone();
                     let tls_proxy_trust = proxy_trust.clone();
+                    let tls_reactors = reactors.clone();
                     thread::spawn(move || {
                         socketengine::accept_loop(
                             tls_listener,
@@ -122,6 +127,7 @@ fn main() {
                             false,
                             max_line,
                             tls_proxy_trust,
+                            tls_reactors,
                         )
                     });
                 }
@@ -139,7 +145,17 @@ fn main() {
                 let s_tx = tx.clone();
                 let s_counter = counter.clone();
                 thread::spawn(move || {
-                    socketengine::accept_loop(sl, s_tx, None, s_counter, true, max_line, Vec::new())
+                    // links stay on the thread path: no reactor handoff
+                    socketengine::accept_loop(
+                        sl,
+                        s_tx,
+                        None,
+                        s_counter,
+                        true,
+                        max_line,
+                        Vec::new(),
+                        Vec::new(),
+                    )
                 });
             }
             Err(e) => eprintln!("echoircd: cannot bind server port {bind_srv}: {e}"),
@@ -163,17 +179,7 @@ fn main() {
         });
     }
 
-    // client plaintext connections: one mio reactor thread drives them all
+    // client plaintext connections: the acceptor round-robins them across the pool
-    thread::spawn(move || {
+    thread::spawn(move || socketengine::run_acceptor(client_listener, reactors, counter, proxy_trust));
-        socketengine::run_reactor_pool(
-            client_listener,
-            tx,
-            counter,
-            max_line,
-            max_sendq,
-            proxy_trust,
-            io_threads,
-        )
-    });
     let _ = core.join();
 }

src/socketengine.rs

@@ -1,13 +1,16 @@
 //! The socket engine: the I/O edge. Two coexisting models feed the one core:
 //!
-//! - **Client plaintext** connections run on a **pool of mio epoll reactors**
+//! - **Client connections** run on a **pool of mio epoll reactors** — acceptors
-//!   ([`run_reactor_pool`]) — one acceptor round-robins connections across N worker
+//!   ([`run_acceptor`] for plaintext, [`accept_loop`] for TLS) round-robin connections
-//!   threads (one per core by default), each driving tens of thousands of sockets, so
+//!   across N worker threads ([`spawn_reactors`], one per core by default), each
-//!   the daemon scales to hundreds of thousands of users without a thread per
+//!   driving tens of thousands of sockets — plaintext and **direct TLS** alike, the
-//!   connection. The state core stays single-threaded and there is no async runtime;
+//!   handshake and crypto run non-blocking in the worker — so the daemon scales to
-//!   workers only frame lines and feed it Events, so the parallel I/O needs no locks.
+//!   hundreds of thousands of users without a thread per connection. The state core
-//! - **TLS** and **server links** keep a thread per connection (few of them, and
+//!   stays single-threaded and there is no async runtime; workers only frame lines and
-//!   a TLS session can't be split across reader+writer threads).
+//!   feed it Events, so the parallel I/O (including TLS crypto) needs no locks.
+//! - **Proxied TLS** (a PROXY header before the handshake) and **server links** keep a
+//!   thread per connection: few of them, and the pre-handshake header wants the
+//!   simpler blocking path.
 //!
 //! Both hand the core the same [`OutSink`] output handle, so the core never
 //! knows or cares which model a connection uses.
@@ -25,7 +28,7 @@ use mio::net::{TcpListener as MioListener, TcpStream as MioStream};
 use mio::{Events, Interest, Poll, Token, Waker};
 
 use crate::ircd::Event;
-use crate::tls::TlsBackend;
+use crate::tls::{TlsBackend, TlsSession};
 use crate::Uid;
 
 /// Default recvq: longest single line we'll buffer before dropping it. Overridable
@@ -123,8 +126,38 @@ const LISTENER: Token = Token(0);
 const WAKE: Token = Token(1);
 const FIRST_CONN: usize = 16; // conn tokens start past the reserved ones
 
+/// A reactor connection's socket: a raw plaintext stream, or a non-blocking TLS
+/// session driven by the same reactor. Both expose the underlying mio socket for
+/// poll registration, so the read/write/backpressure machinery is identical.
+enum Sock {
+    Plain(MioStream),
+    Tls(Box<dyn TlsSession>),
+}
+
+impl Sock {
+    /// The underlying socket, for poll (re)register/deregister.
+    fn source(&mut self) -> &mut MioStream {
+        match self {
+            Sock::Plain(s) => s,
+            Sock::Tls(t) => t.source(),
+        }
+    }
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        match self {
+            Sock::Plain(s) => s.read(buf),
+            Sock::Tls(t) => t.read(buf),
+        }
+    }
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        match self {
+            Sock::Plain(s) => s.write(buf),
+            Sock::Tls(t) => t.write(buf),
+        }
+    }
+}
+
 struct Conn {
-    stream: MioStream,
+    sock: Sock,
     uid: Uid,
     addr: SocketAddr, // peer, or the real client once a PROXY header is parsed
     local_port: u16,  // listener port (for the deferred-Connect case)
@@ -138,6 +171,7 @@ struct Conn {
     recvq: usize,      // max buffered unterminated-line bytes before dropping
     hardsendq: usize,  // max queued output bytes before dropping + closing
     softsendq: usize,  // queued output above this pauses reads until it drains
+    handshaking: bool, // TLS: still negotiating; hold reads + the Connect until done
     proxy_pending: bool, // hold the Connect event until a PROXY header is consumed
     pending_out: Option<OutSink>, // the OutSink held for that deferred Connect
 }
@@ -153,7 +187,9 @@ impl Conn {
 /// serviced) while keeping WRITABLE to drain the backlog that paused it.
 fn set_interest(poll: &mut Poll, c: &mut Conn, t: usize) {
     let want_read = !c.paused;
-    let want_write = !c.wbuf.is_empty() || c.paused;
+    // a TLS handshake may need to write (its flight) as well as read, so keep both
+    // until it completes; after that, write only when there's a backlog to drain.
+    let want_write = c.handshaking || !c.wbuf.is_empty() || c.paused;
     if want_read == c.want_read && want_write == c.want_write {
         return;
     }
@@ -165,7 +201,7 @@ fn set_interest(poll: &mut Poll, c: &mut Conn, t: usize) {
         // never both-false (paused ⟹ backlog ⟹ want_write); READABLE is a safe floor
         _ => Interest::READABLE,
     };
-    let _ = poll.registry().reregister(&mut c.stream, Token(t), interest);
+    let _ = poll.registry().reregister(c.sock.source(), Token(t), interest);
 }
 
 /// Largest PROXY header we'll buffer before giving up (v1 ≤ 107, v2 header ≤ ~232).
@@ -178,11 +214,15 @@ struct Accepted {
     addr: SocketAddr,
     local_port: u16,
     via_proxy: bool,
+    tls: Option<Arc<dyn TlsBackend>>, // Some ⇒ the worker negotiates TLS on this socket
 }
 
 /// The acceptor's handle to one reactor worker: its handoff queue and the waker that
-/// nudges the worker to adopt whatever was queued.
+/// nudges the worker to adopt whatever was queued. Cloneable so several acceptors
-struct ReactorHandle {
+/// (plaintext + TLS) can share the same pool, each round-robining independently.
+/// Opaque to callers — `main` only holds a `Vec` of these and passes it along.
+#[derive(Clone)]
+pub struct ReactorHandle {
     handoff: Sender<Accepted>,
     waker: Arc<Waker>,
 }
@@ -200,34 +240,53 @@ fn resolve_io_threads(io_threads: usize) -> usize {
         .clamp(1, 4)
 }
 
-/// Drive the client plaintext listener with a pool of reactor threads. One acceptor
+/// Start the reactor worker pool and return the acceptors' handles to it. Sized by
-/// (this thread) owns the listener and round-robins each new connection to a worker;
+/// `io_threads` (0 = auto: one worker per core, capped). Each worker runs its own poll
-/// each worker runs its own poll and connection map on its own core. The state core
+/// and connection map on its own core; the state core stays single-threaded — workers
-/// stays single-threaded — workers only frame lines and feed it Events — so the
+/// only frame lines and feed it Events — so per-connection I/O (plaintext framing and
-/// per-connection I/O scales across cores with no shared locking.
+/// TLS crypto alike) scales across cores with no shared locking.
-pub fn run_reactor_pool(
+pub fn spawn_reactors(
-    mut listener: MioListener,
     core: Sender<Event>,
-    counter: Arc<AtomicU64>,
     max_line: usize,
     max_sendq: usize,
-    proxy_trust: Vec<String>,
     io_threads: usize,
-) {
+) -> Vec<ReactorHandle> {
     let workers = resolve_io_threads(io_threads);
-    let mut reactors: Vec<ReactorHandle> = Vec::with_capacity(workers);
+    let mut reactors = Vec::with_capacity(workers);
     for _ in 0..workers {
         match spawn_reactor(core.clone(), max_line, max_sendq) {
             Ok(h) => reactors.push(h),
             Err(e) => eprintln!("reactor: cannot start a worker: {e}"),
         }
     }
+    eprintln!("echoircd reactor pool: {} worker thread(s)", reactors.len());
+    reactors
+}
+
+/// Round-robin one accepted connection onto a worker and wake it to adopt the conn.
+fn dispatch(reactors: &[ReactorHandle], rr: &mut usize, a: Accepted) {
     if reactors.is_empty() {
-        eprintln!("reactor: no worker threads started; plaintext clients disabled");
+        return; // no workers: drop it (a.stream closes on drop)
+    }
+    let idx = *rr % reactors.len();
+    *rr = rr.wrapping_add(1);
+    if reactors[idx].handoff.send(a).is_ok() {
+        let _ = reactors[idx].waker.wake();
+    }
+}
+
+/// The plaintext client acceptor: owns the listener and round-robins each new
+/// connection onto a reactor worker.
+pub fn run_acceptor(
+    mut listener: MioListener,
+    reactors: Vec<ReactorHandle>,
+    counter: Arc<AtomicU64>,
+    proxy_trust: Vec<String>,
+) {
+    if reactors.is_empty() {
+        eprintln!("acceptor: no worker threads; plaintext clients disabled");
         return;
     }
-    eprintln!("echoircd plaintext reactor pool: {} thread(s)", reactors.len());
-
     let mut poll = match Poll::new() {
         Ok(p) => p,
         Err(e) => {
@@ -266,18 +325,18 @@ pub fn run_reactor_pool(
                         let via_proxy = proxy_trust.iter().any(|g| {
                             crate::modules::connclass::ip_matches(g, &addr.ip().to_string())
                         });
-                        let idx = rr % reactors.len();
+                        dispatch(
-                        rr = rr.wrapping_add(1);
+                            &reactors,
-                        let accepted = Accepted {
+                            &mut rr,
+                            Accepted {
                                 stream,
                                 uid,
                                 addr,
                                 local_port,
                                 via_proxy,
-                        };
+                                tls: None,
-                        if reactors[idx].handoff.send(accepted).is_ok() {
+                            },
-                            let _ = reactors[idx].waker.wake();
+                        );
-                        }
                     }
                     Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => break,
                     Err(_) => break,
@@ -338,10 +397,23 @@ fn reactor_loop(
                     while let Ok(a) = handoff_rx.try_recv() {
                         let token = next_token;
                         next_token += 1;
-                        let mut stream = a.stream;
+                        // build the socket: a TLS conn negotiates non-blocking in this
+                        // worker; a plaintext one is ready to read immediately.
+                        let (mut sock, handshaking) = match a.tls {
+                            Some(backend) => match backend.start(a.stream) {
+                                Ok(sess) => (Sock::Tls(sess), true),
+                                Err(_) => continue, // couldn't start TLS: drop it
+                            },
+                            None => (Sock::Plain(a.stream), false),
+                        };
+                        let interest = if handshaking {
+                            Interest::READABLE | Interest::WRITABLE
+                        } else {
+                            Interest::READABLE
+                        };
                         if poll
                             .registry()
-                            .register(&mut stream, Token(token), Interest::READABLE)
+                            .register(sock.source(), Token(token), interest)
                             .is_err()
                         {
                             continue;
@@ -354,7 +426,7 @@ fn reactor_loop(
                         conns.insert(
                             token,
                             Conn {
-                                stream,
+                                sock,
                                 uid: a.uid,
                                 addr: a.addr,
                                 local_port: a.local_port,
@@ -362,19 +434,20 @@ fn reactor_loop(
                                 wbuf: Vec::new(),
                                 wpos: 0,
                                 want_read: true,
-                                want_write: false,
+                                want_write: handshaking,
                                 closing: false,
                                 paused: false,
                                 recvq: max_line,
                                 hardsendq: max_sendq,
                                 softsendq: max_sendq,
+                                handshaking,
                                 proxy_pending: a.via_proxy,
                                 pending_out: Some(out),
                             },
                         );
-                        // non-proxy: announce immediately (a proxy conn is announced
+                        // announce now only if nothing defers it: a TLS conn waits for
-                        // from read_conn once its header lands)
+                        // its handshake, a proxy conn for its header.
-                        if !a.via_proxy {
+                        if !a.via_proxy && !handshaking {
                             let out = conns.get_mut(&token).and_then(|c| c.pending_out.take());
                             if let Some(out) = out {
                                 if core
@@ -481,9 +554,70 @@ fn reactor_loop(
     }
 }
 
+/// Drive a pending TLS handshake for `t`. Returns true once the connection is
+/// established — its deferred Connect emitted with the peer's cert fingerprint, so
+/// normal reads/writes may proceed — and false while it still needs I/O or was closed
+/// on a fatal handshake error. A plaintext (or already-established) conn returns true.
+fn try_handshake(
+    poll: &mut Poll,
+    conns: &mut HashMap<usize, Conn>,
+    t: usize,
+    core: &Sender<Event>,
+) -> bool {
+    let mut close = false;
+    let mut connect: Option<(Uid, SocketAddr, u16, Option<String>, OutSink)> = None;
+    if let Some(c) = conns.get_mut(&t) {
+        if !c.handshaking {
+            return true;
+        }
+        if let Sock::Tls(sess) = &mut c.sock {
+            match sess.accept() {
+                Ok(true) => {
+                    c.handshaking = false;
+                    let certfp = sess.peer_cert_fp();
+                    connect = c
+                        .pending_out
+                        .take()
+                        .map(|out| (c.uid, c.addr, c.local_port, certfp, out));
+                    set_interest(poll, c, t); // handshake done: drop the extra WRITABLE
+                }
+                Ok(false) => return false, // still negotiating
+                Err(_) => close = true,
+            }
+        } else {
+            c.handshaking = false; // not TLS (shouldn't happen): treat as established
+        }
+    } else {
+        return false;
+    }
+    if let Some((uid, addr, local_port, certfp, out)) = connect {
+        let _ = core.send(Event::Connect {
+            uid,
+            addr,
+            out,
+            sock: None,
+            secure: true,
+            certfp,
+            local_port,
+            link: false,
+            outbound: false,
+            websocket: false,
+        });
+    }
+    if close {
+        close_conn(poll, conns, t, core);
+        return false;
+    }
+    true
+}
+
 /// Drain readable bytes from `t` (edge-triggered: read until WouldBlock), frame
 /// complete lines and forward them to the core; close on EOF/error.
 fn read_conn(poll: &mut Poll, conns: &mut HashMap<usize, Conn>, t: usize, core: &Sender<Event>) {
+    // a TLS conn must finish negotiating before any application bytes flow
+    if !try_handshake(poll, conns, t, core) {
+        return;
+    }
     let mut chunk = [0u8; 8192];
     let mut lines: Vec<(Uid, String)> = Vec::new();
     // a deferred Connect (PROXY conn) to emit, before any lines from the same read
@@ -491,7 +625,7 @@ fn read_conn(poll: &mut Poll, conns: &mut HashMap<usize, Conn>, t: usize, core:
     let mut close = false;
     if let Some(c) = conns.get_mut(&t) {
         loop {
-            match c.stream.read(&mut chunk) {
+            match c.sock.read(&mut chunk) {
                 Ok(0) => {
                     close = true;
                     break;
@@ -597,11 +731,15 @@ fn read_conn(poll: &mut Poll, conns: &mut HashMap<usize, Conn>, t: usize, core:
 /// backlog dropped back under softsendq, un-pause reads and catch up (edge-triggered:
 /// data that arrived while paused won't re-fire, so read it here).
 fn flush_conn(poll: &mut Poll, conns: &mut HashMap<usize, Conn>, t: usize, core: &Sender<Event>) {
+    // a writable event during a TLS handshake advances it, not the (empty) write queue
+    if !try_handshake(poll, conns, t, core) {
+        return;
+    }
     let mut close = false;
     let mut unpaused = false;
     if let Some(c) = conns.get_mut(&t) {
         while c.wpos < c.wbuf.len() {
-            match c.stream.write(&c.wbuf[c.wpos..]) {
+            match c.sock.write(&c.wbuf[c.wpos..]) {
                 Ok(0) => break,
                 Ok(n) => c.wpos += n,
                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => break,
@@ -635,11 +773,12 @@ fn flush_conn(poll: &mut Poll, conns: &mut HashMap<usize, Conn>, t: usize, core:
 /// Deregister + drop `t`'s socket and tell the core the connection is gone.
 fn close_conn(poll: &mut Poll, conns: &mut HashMap<usize, Conn>, t: usize, core: &Sender<Event>) {
     if let Some(mut c) = conns.remove(&t) {
-        let _ = poll.registry().deregister(&mut c.stream);
+        let _ = poll.registry().deregister(c.sock.source());
         let uid = c.uid;
-        // a still-pending PROXY conn was never announced to the core, so don't tell
+        // a conn whose Connect was never emitted — a still-pending PROXY header or an
-        // it about a disconnect for a uid it never saw
+        // unfinished TLS handshake — must not send the core a Disconnect for a uid it
-        let announced = !c.proxy_pending;
+        // never saw
+        let announced = !c.proxy_pending && !c.handshaking;
         drop(c); // closes the socket
         if announced {
             let _ = core.send(Event::Disconnect { uid });
@@ -649,9 +788,11 @@ fn close_conn(poll: &mut Poll, conns: &mut HashMap<usize, Conn>, t: usize, core:
 
 // === thread model: TLS + server links ========================================
 
-/// Accept forever on a thread-per-connection listener (TLS or S2S). `tls` is the
+/// Accept forever on a listener (TLS or S2S). `tls` is the backend to wrap sockets in
-/// backend to wrap sockets in (None ⇒ plaintext link). `counter` is shared with
+/// (None ⇒ plaintext link). `counter` is shared with the reactor so uids stay unique
-/// the reactor so uids stay unique across every listener.
+/// across every listener. `reactors` is the worker pool: a direct (non-proxy) TLS
+/// client is handed off to it to negotiate non-blocking; a proxied TLS client (PROXY
+/// header before the handshake) and every server link keep the thread path.
 pub fn accept_loop(
     listener: TcpListener,
     core: Sender<Event>,
@@ -660,7 +801,9 @@ pub fn accept_loop(
     link: bool,
     max_line: usize,
     proxy_trust: Vec<String>,
+    reactors: Vec<ReactorHandle>,
 ) {
+    let mut rr: usize = 0;
     for conn in listener.incoming() {
         let Ok(stream) = conn else { continue };
         let Ok(addr) = stream.peer_addr() else {
@@ -701,6 +844,27 @@ pub fn accept_loop(
                 thread::spawn(move || reader_loop(reader, uid, core_tx, max_line));
             }
             Some(backend) => {
+                let via_proxy = proxy_trust
+                    .iter()
+                    .any(|g| crate::modules::connclass::ip_matches(g, &addr.ip().to_string()));
+                // direct TLS clients negotiate in the reactor pool (non-blocking, one
+                // worker per core); a proxied client keeps the thread path so its
+                // plaintext PROXY header is read before the handshake.
+                if !via_proxy && !reactors.is_empty() && stream.set_nonblocking(true).is_ok() {
+                    dispatch(
+                        &reactors,
+                        &mut rr,
+                        Accepted {
+                            stream: MioStream::from_std(stream),
+                            uid,
+                            addr,
+                            local_port,
+                            via_proxy: false,
+                            tls: Some(backend.clone()),
+                        },
+                    );
+                    continue;
+                }
                 let backend = backend.clone();
                 let core_tx = core.clone();
                 let pt = proxy_trust.clone();

src/tls.rs

@@ -9,8 +9,11 @@ use std::io::{self, Read, Write};
 use std::net::{Shutdown, TcpStream};
 use std::time::Duration;
 
+use mio::net::TcpStream as MioStream;
 use openssl::hash::MessageDigest;
-use openssl::ssl::{SslAcceptor, SslFiletype, SslMethod, SslStream, SslVerifyMode};
+use openssl::ssl::{
+    ErrorCode, Ssl, SslAcceptor, SslFiletype, SslMethod, SslMode, SslStream, SslVerifyMode,
+};
 
 /// A live TLS connection: read/write plaintext, tune the read timeout (the
 /// socket engine polls with one to interleave reads and queued writes), and shut
@@ -26,9 +29,31 @@ pub trait TlsConn: Send {
     fn peer_cert_fp(&self) -> Option<String>;
 }
 
-/// A TLS backend: performs the server-side handshake on an accepted socket.
+/// A non-blocking TLS session the reactor drives itself over a mio socket. The
+/// handshake and all reads/writes surface `WouldBlock` (mapped from OpenSSL's
+/// WANT_READ/WANT_WRITE) so the worker can register interest and come back later
+/// instead of blocking a whole thread on one connection.
+pub trait TlsSession: Send {
+    /// Drive the server handshake: `Ok(true)` once complete, `Ok(false)` while it
+    /// still needs I/O, `Err` on a fatal handshake failure.
+    fn accept(&mut self) -> io::Result<bool>;
+    /// Decrypt application data. `Ok(0)` means the peer sent a clean TLS close.
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize>;
+    /// Encrypt+queue application data; returns the plaintext bytes accepted.
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize>;
+    /// The underlying mio socket, for the reactor's poll (re)registration.
+    fn source(&mut self) -> &mut MioStream;
+    /// SHA-256 fingerprint of the peer certificate (CertFP / SASL EXTERNAL), if any.
+    fn peer_cert_fp(&self) -> Option<String>;
+    fn shutdown(&mut self);
+}
+
+/// A TLS backend: wraps an accepted socket in a TLS session — either blocking
+/// ([`accept`], the thread-per-connection path) or non-blocking ([`start`], the
+/// reactor path).
 pub trait TlsBackend: Send + Sync {
     fn accept(&self, sock: TcpStream) -> io::Result<Box<dyn TlsConn>>;
+    fn start(&self, sock: MioStream) -> io::Result<Box<dyn TlsSession>>;
 }
 
 fn err<E: std::fmt::Display>(e: E) -> io::Error {
@@ -52,6 +77,10 @@ impl OpensslBackend {
         // read its fingerprint. We never validate the chain — services match the
         // fingerprint to an account — so the callback always accepts.
         b.set_verify_callback(SslVerifyMode::PEER, |_valid, _ctx| true);
+        // The reactor drives writes non-blocking and may retry SSL_write with a moved
+        // or grown buffer after a WouldBlock; allow that and partial progress so a slow
+        // TLS reader can't wedge a worker.
+        b.set_mode(SslMode::ENABLE_PARTIAL_WRITE | SslMode::ACCEPT_MOVING_WRITE_BUFFER);
         Ok(OpensslBackend {
             acceptor: b.build(),
         })
@@ -63,6 +92,59 @@ impl TlsBackend for OpensslBackend {
         let stream = self.acceptor.accept(sock).map_err(err)?;
         Ok(Box::new(OpensslConn(stream)))
     }
+
+    fn start(&self, sock: MioStream) -> io::Result<Box<dyn TlsSession>> {
+        let ssl = Ssl::new(self.acceptor.context()).map_err(err)?;
+        // handshake isn't driven here: SslStream::new just binds the socket; the
+        // reactor calls accept() as the socket becomes readable/writable.
+        let stream = SslStream::new(ssl, sock).map_err(err)?;
+        Ok(Box::new(OpensslSession(stream)))
+    }
+}
+
+struct OpensslSession(SslStream<MioStream>);
+
+/// Map an OpenSSL ssl error to the reactor's io model: WANT_READ/WANT_WRITE ⇒
+/// `WouldBlock` (retry when ready), everything else ⇒ a real error.
+fn ssl_io_err(e: openssl::ssl::Error) -> io::Error {
+    match e.code() {
+        ErrorCode::WANT_READ | ErrorCode::WANT_WRITE => io::ErrorKind::WouldBlock.into(),
+        _ => e.into_io_error().unwrap_or_else(io::Error::other),
+    }
+}
+
+impl TlsSession for OpensslSession {
+    fn accept(&mut self) -> io::Result<bool> {
+        match self.0.accept() {
+            Ok(()) => Ok(true),
+            Err(e) => match e.code() {
+                ErrorCode::WANT_READ | ErrorCode::WANT_WRITE => Ok(false),
+                _ => Err(e.into_io_error().unwrap_or_else(io::Error::other)),
+            },
+        }
+    }
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        match self.0.ssl_read(buf) {
+            Ok(n) => Ok(n),
+            // a clean TLS close is EOF, like a plaintext socket returning 0
+            Err(e) if e.code() == ErrorCode::ZERO_RETURN => Ok(0),
+            Err(e) => Err(ssl_io_err(e)),
+        }
+    }
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        self.0.ssl_write(buf).map_err(ssl_io_err)
+    }
+    fn source(&mut self) -> &mut MioStream {
+        self.0.get_mut()
+    }
+    fn peer_cert_fp(&self) -> Option<String> {
+        let cert = self.0.ssl().peer_certificate()?;
+        let digest = cert.digest(MessageDigest::sha256()).ok()?;
+        Some(digest.iter().map(|b| format!("{b:02x}")).collect())
+    }
+    fn shutdown(&mut self) {
+        let _ = self.0.get_ref().shutdown(Shutdown::Both);
+    }
 }
 
 struct OpensslConn(SslStream<TcpStream>);