mz_server_core/
lib.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! Methods common to servers listening for TCP connections.
11
12use std::fmt;
13use std::future::Future;
14use std::io;
15use std::net::SocketAddr;
16use std::path::PathBuf;
17use std::pin::Pin;
18use std::sync::{Arc, Mutex, RwLock, RwLockReadGuard};
19use std::task::{Context, Poll};
20use std::time::Duration;
21
22use anyhow::bail;
23use async_trait::async_trait;
24use clap::builder::ArgPredicate;
25use futures::stream::{BoxStream, Stream, StreamExt};
26use mz_dyncfg::{Config, ConfigSet};
27use mz_ore::channel::trigger;
28use mz_ore::error::ErrorExt;
29use mz_ore::netio::AsyncReady;
30use mz_ore::option::OptionExt;
31use mz_ore::task::JoinSetExt;
32use openssl::ssl::{SslAcceptor, SslContext, SslFiletype, SslMethod};
33use proxy_header::{ParseConfig, ProxiedAddress, ProxyHeader};
34use schemars::JsonSchema;
35use scopeguard::ScopeGuard;
36use serde::{Deserialize, Serialize};
37use socket2::{SockRef, TcpKeepalive};
38use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, Interest, ReadBuf, Ready};
39use tokio::net::{TcpSocket, TcpStream};
40use tokio::sync::oneshot;
41use tokio::task::JoinSet;
42use tokio_metrics::TaskMetrics;
43use tokio_stream::wrappers::{IntervalStream, TcpListenerStream};
44use tracing::{debug, error, warn};
45use uuid::Uuid;
46
47pub mod listeners;
48
49/// TCP keepalive settings. The idle time and interval match CockroachDB [0].
50/// The number of retries matches the Linux default.
51///
52/// [0]: https://github.com/cockroachdb/cockroach/pull/14063
53const KEEPALIVE: TcpKeepalive = TcpKeepalive::new()
54    .with_time(Duration::from_secs(60))
55    .with_interval(Duration::from_secs(60))
56    .with_retries(9);
57
58/// A future that handles a connection.
59pub type ConnectionHandler = Pin<Box<dyn Future<Output = Result<(), anyhow::Error>> + Send>>;
60
61/// A wrapper around a [`TcpStream`] that can identify a connection across
62/// processes.
63pub struct Connection {
64    conn_uuid: Arc<Mutex<Option<Uuid>>>,
65    tcp_stream: TcpStream,
66}
67
68impl Connection {
69    fn new(tcp_stream: TcpStream) -> Connection {
70        Connection {
71            conn_uuid: Arc::new(Mutex::new(None)),
72            tcp_stream,
73        }
74    }
75
76    /// Returns a handle to the connection UUID.
77    pub fn uuid_handle(&self) -> ConnectionUuidHandle {
78        ConnectionUuidHandle(Arc::clone(&self.conn_uuid))
79    }
80
81    /// Attempts to parse a proxy header from the tcp_stream.
82    /// If none is found or it is unable to be parsed None will
83    /// be returned. If a header is found it will be returned and its
84    /// bytes will be removed from the stream.
85    ///
86    /// It is possible an invalid header was sent, if that is the case
87    /// any downstream service will be responsible for returning errors
88    /// to the client.
89    pub async fn take_proxy_header_address(&mut self) -> Option<ProxiedAddress> {
90        // 1024 bytes is a rather large header for tcp proxy header, unless
91        // if the header contains TLV fields or uses a unix socket address
92        // this could easily be hit. We'll use a 1024 byte max buf to allow
93        // limited support for this.
94        let mut buf = [0u8; 1024];
95        let len = match self.tcp_stream.peek(&mut buf).await {
96            Ok(n) if n > 0 => n,
97            _ => {
98                debug!("Failed to read from client socket or no data received");
99                return None;
100            }
101        };
102
103        // Attempt to parse the header, and log failures.
104        let (header, hlen) = match ProxyHeader::parse(
105            &buf[..len],
106            ParseConfig {
107                include_tlvs: false,
108                allow_v1: false,
109                allow_v2: true,
110            },
111        ) {
112            Ok((header, hlen)) => (header, hlen),
113            Err(proxy_header::Error::Invalid) => {
114                debug!("Proxy header is invalid. This is likely due to no header being provided",);
115                return None;
116            }
117            // Data matches the PROXY v2 signature prefix but the header
118            // is incomplete — likely split across TCP segments. Read the
119            // 16-byte fixed v2 header to learn the total size, then read
120            // the remaining address bytes.
121            Err(proxy_header::Error::BufferTooShort) => {
122                return self.read_proxy_v2_header(&mut buf).await;
123            }
124            Err(e) => {
125                debug!("Proxy header parse error '{:?}', ignoring header.", e);
126                return None;
127            }
128        };
129        debug!("Proxied connection with header {:?}", header);
130        let address = header.proxied_address().map(|a| a.to_owned());
131        // Proxy header found, clear the bytes.
132        let _ = self.read_exact(&mut buf[..hlen]).await;
133        address
134    }
135
136    /// Fallback path for [`Self::take_proxy_header_address`] when the initial
137    /// peek returned an incomplete PROXY v2 header. Reads the fixed 16-byte
138    /// v2 prefix to learn the total header size, then reads the rest.
139    async fn read_proxy_v2_header(&mut self, buf: &mut [u8; 1024]) -> Option<ProxiedAddress> {
140        // PROXY v2 fixed prefix: 12-byte signature + ver/cmd + fam/proto + 2-byte length.
141        const V2_PREFIX_LEN: usize = 16;
142        if self.read_exact(&mut buf[..V2_PREFIX_LEN]).await.is_err() {
143            debug!("Failed to read PROXY v2 fixed header");
144            return None;
145        }
146        let addr_len = usize::from(u16::from_be_bytes([buf[14], buf[15]]));
147        let total = V2_PREFIX_LEN + addr_len;
148        if total > buf.len() {
149            debug!("PROXY v2 header too large: {total} bytes");
150            return None;
151        }
152        if self
153            .read_exact(&mut buf[V2_PREFIX_LEN..total])
154            .await
155            .is_err()
156        {
157            debug!("Failed to read PROXY v2 address data");
158            return None;
159        }
160        match ProxyHeader::parse(
161            &buf[..total],
162            ParseConfig {
163                include_tlvs: false,
164                allow_v1: false,
165                allow_v2: true,
166            },
167        ) {
168            Ok((header, _)) => {
169                debug!("Proxied connection with header {:?}", header);
170                header.proxied_address().map(|a| a.to_owned())
171            }
172            Err(e) => {
173                debug!("Proxy header parse error '{:?}', ignoring header.", e);
174                None
175            }
176        }
177    }
178
179    /// Peer address of the inner tcp_stream.
180    pub fn peer_addr(&self) -> Result<std::net::SocketAddr, io::Error> {
181        self.tcp_stream.peer_addr()
182    }
183}
184
185impl AsyncRead for Connection {
186    fn poll_read(
187        mut self: Pin<&mut Self>,
188        cx: &mut Context,
189        buf: &mut ReadBuf,
190    ) -> Poll<io::Result<()>> {
191        Pin::new(&mut self.tcp_stream).poll_read(cx, buf)
192    }
193}
194
195impl AsyncWrite for Connection {
196    fn poll_write(
197        mut self: Pin<&mut Self>,
198        cx: &mut Context,
199        buf: &[u8],
200    ) -> Poll<io::Result<usize>> {
201        Pin::new(&mut self.tcp_stream).poll_write(cx, buf)
202    }
203
204    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
205        Pin::new(&mut self.tcp_stream).poll_flush(cx)
206    }
207
208    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
209        Pin::new(&mut self.tcp_stream).poll_shutdown(cx)
210    }
211}
212
213#[async_trait]
214impl AsyncReady for Connection {
215    async fn ready(&self, interest: Interest) -> io::Result<Ready> {
216        self.tcp_stream.ready(interest).await
217    }
218}
219
220/// A handle that permits getting and setting the UUID for a [`Connection`].
221///
222/// A connection's UUID is a globally unique value that can identify a given
223/// connection across environments and process boundaries. Connection UUIDs are
224/// never reused.
225///
226/// This is distinct from environmentd's concept of a "connection ID", which is
227/// a `u32` that only identifies a connection within a given environment and
228/// only during its lifetime. These connection IDs are frequently reused.
229pub struct ConnectionUuidHandle(Arc<Mutex<Option<Uuid>>>);
230
231impl ConnectionUuidHandle {
232    /// Gets the UUID for the connection, if it exists.
233    pub fn get(&self) -> Option<Uuid> {
234        *self.0.lock().expect("lock poisoned")
235    }
236
237    /// Sets the UUID for this connection.
238    pub fn set(&self, conn_uuid: Uuid) {
239        *self.0.lock().expect("lock poisoned") = Some(conn_uuid);
240    }
241
242    /// Returns a displayable that renders a possibly missing connection UUID.
243    pub fn display(&self) -> impl fmt::Display {
244        self.get().display_or("<unknown>")
245    }
246}
247
248/// A server handles incoming network connections.
249pub trait Server {
250    /// Returns the name of the connection handler for use in e.g. log messages.
251    const NAME: &'static str;
252
253    /// Handles a single connection.
254    fn handle_connection(
255        &self,
256        conn: Connection,
257        tokio_metrics_intervals: impl Iterator<Item = TaskMetrics> + Send + 'static,
258    ) -> ConnectionHandler;
259}
260
261/// A stream of incoming connections.
262pub trait ConnectionStream: Stream<Item = io::Result<TcpStream>> + Unpin + Send {}
263
264impl<T> ConnectionStream for T where T: Stream<Item = io::Result<TcpStream>> + Unpin + Send {}
265
266/// A handle to a listener created by [`listen`].
267#[derive(Debug)]
268pub struct ListenerHandle {
269    pub local_addr: SocketAddr,
270    _trigger: trigger::Trigger,
271}
272
273impl ListenerHandle {
274    /// Returns the local address to which the listener is bound.
275    pub fn local_addr(&self) -> SocketAddr {
276        self.local_addr
277    }
278}
279
280/// The accept-queue length requested for listening sockets.
281///
282/// `environmentd` binds its listeners before it boots and only starts accepting
283/// once boot completes, so every connection made during boot, including
284/// Kubernetes TCP probes, waits in this queue. Once it is full, Linux drops new
285/// connection attempts, which fails TCP probes and can get the pod killed
286/// mid-boot. Linux silently caps this at `net.core.somaxconn`.
287const LISTEN_BACKLOG: u32 = 1024;
288
289/// Listens for incoming TCP connections on the specified address.
290///
291/// Returns a handle to the listener and the stream of incoming connections
292/// produced by the listener. When the handle is dropped, the listener is
293/// closed, and the stream of incoming connections terminates.
294///
295/// Panics if called outside a Tokio runtime.
296pub fn listen(
297    addr: &SocketAddr,
298) -> Result<(ListenerHandle, Pin<Box<dyn ConnectionStream>>), io::Error> {
299    let socket = match addr {
300        SocketAddr::V4(_) => TcpSocket::new_v4()?,
301        SocketAddr::V6(_) => TcpSocket::new_v6()?,
302    };
303    // `TcpListener::bind` sets this on Unix. Without it, a restarted process
304    // cannot rebind while connections from the previous one are in TIME_WAIT.
305    socket.set_reuseaddr(true)?;
306    socket.bind(*addr)?;
307    let listener = socket.listen(LISTEN_BACKLOG)?;
308    let local_addr = listener.local_addr()?;
309    let (trigger, trigger_rx) = trigger::channel();
310    let handle = ListenerHandle {
311        local_addr,
312        _trigger: trigger,
313    };
314    // TODO(benesch): replace `TCPListenerStream`s with `listener.incoming()` if
315    // that is restored when the `Stream` trait stabilizes.
316    let stream = TcpListenerStream::new(listener).take_until(trigger_rx);
317    Ok((handle, Box::pin(stream)))
318}
319
320/// Configuration for [`serve`].
321pub struct ServeConfig<S, C>
322where
323    S: Server,
324    C: ConnectionStream,
325{
326    /// The server for the connections.
327    pub server: S,
328    /// The stream of incoming TCP connections.
329    pub conns: C,
330    /// Optional dynamic configuration for the server.
331    pub dyncfg: Option<ServeDyncfg>,
332}
333
334/// Dynamic configuration for [`ServeConfig`].
335pub struct ServeDyncfg {
336    /// The current bundle of dynamic configuration values.
337    pub config_set: ConfigSet,
338    /// A configuration in `config_set` that specifies how long to wait for
339    /// connections to terminate after receiving a SIGTERM before forcibly
340    /// terminated.
341    ///
342    /// If `None`, then forcible shutdown occurs immediately.
343    pub sigterm_wait_config: &'static Config<Duration>,
344}
345
346/// Serves incoming TCP connections.
347///
348/// Returns handles to the outstanding connections after the configured timeout
349/// has expired or all connections have completed.
350pub async fn serve<S, C>(
351    ServeConfig {
352        server,
353        mut conns,
354        dyncfg,
355    }: ServeConfig<S, C>,
356) -> JoinSet<()>
357where
358    S: Server,
359    C: ConnectionStream,
360{
361    let task_name = format!("handle_{}_connection", S::NAME);
362    let mut set = JoinSet::new();
363    loop {
364        tokio::select! {
365            // next() is cancel safe.
366            conn = conns.next() => {
367                let conn = match conn {
368                    None => break,
369                    Some(Ok(conn)) => conn,
370                    Some(Err(err)) => {
371                        error!("error accepting connection: {}", err);
372                        continue;
373                    }
374                };
375                // Set TCP_NODELAY to disable tinygram prevention (Nagle's
376                // algorithm), which forces a 40ms delay between each query
377                // on linux. According to John Nagle [0], the true problem
378                // is delayed acks, but disabling those is a receive-side
379                // operation (TCP_QUICKACK), and we can't always control the
380                // client. PostgreSQL sets TCP_NODELAY on both sides of its
381                // sockets, so it seems sane to just do the same.
382                //
383                // If set_nodelay fails, it's a programming error, so panic.
384                //
385                // [0]: https://news.ycombinator.com/item?id=10608356
386                conn.set_nodelay(true).expect("set_nodelay failed");
387                // Enable TCP keepalives to avoid any idle connection timeouts that may
388                // be enforced by networking devices between us and the client. Idle SQL
389                // connections are expected--e.g., a `SUBSCRIBE` to a view containing
390                // critical alerts will ideally be producing no data most of the time.
391                if let Err(e) = SockRef::from(&conn).set_tcp_keepalive(&KEEPALIVE) {
392                    error!("failed enabling keepalive: {e}");
393                    continue;
394                }
395                let conn = Connection::new(conn);
396                let conn_uuid = conn.uuid_handle();
397                let metrics_monitor = tokio_metrics::TaskMonitor::new();
398                let tokio_metrics_intervals = metrics_monitor.intervals();
399                let fut = server.handle_connection(conn, tokio_metrics_intervals);
400                set.spawn_named(|| &task_name, metrics_monitor.instrument(async move {
401                    let guard = scopeguard::guard((), |_| {
402                        debug!(
403                            server = S::NAME,
404                            conn_uuid = %conn_uuid.display(),
405                            "dropping connection without explicit termination",
406                        );
407                    });
408
409                    match fut.await {
410                        Ok(()) => {
411                            debug!(
412                                server = S::NAME,
413                                conn_uuid = %conn_uuid.display(),
414                                "successfully handled connection",
415                            );
416                        }
417                        Err(e) => {
418                            warn!(
419                                server = S::NAME,
420                                conn_uuid = %conn_uuid.display(),
421                                "error handling connection: {}",
422                                e.display_with_causes(),
423                            );
424                        }
425                    }
426
427                    let () = ScopeGuard::into_inner(guard);
428                }));
429            }
430            // Actively cull completed tasks from the JoinSet so it does not grow unbounded. This
431            // method is cancel safe.
432            res = set.join_next(), if set.len() > 0 => {
433                if let Some(Err(e)) = res {
434                    warn!(
435                        "error joining connection in {}: {}",
436                        S::NAME,
437                        e.display_with_causes()
438                    );
439                }
440            }
441        }
442    }
443    if let Some(dyncfg) = dyncfg {
444        let wait = dyncfg.sigterm_wait_config.get(&dyncfg.config_set);
445        if set.len() > 0 {
446            warn!(
447                "{} exiting, {} outstanding connections, waiting for {:?}",
448                S::NAME,
449                set.len(),
450                wait
451            );
452        }
453        let timedout = tokio::time::timeout(wait, async {
454            while let Some(res) = set.join_next().await {
455                if let Err(e) = res {
456                    warn!(
457                        "error joining connection in {}: {}",
458                        S::NAME,
459                        e.display_with_causes()
460                    );
461                }
462            }
463        })
464        .await;
465        if timedout.is_err() {
466            warn!(
467                "{}: wait timeout of {:?} exceeded, {} outstanding connections",
468                S::NAME,
469                wait,
470                set.len()
471            );
472        }
473    }
474    set
475}
476
477/// Configures a server's TLS encryption and authentication.
478#[derive(Clone, Debug)]
479pub struct TlsConfig {
480    /// The SSL context used to manage incoming TLS negotiations.
481    pub context: SslContext,
482    /// The TLS mode.
483    pub mode: TlsMode,
484}
485
486/// Specifies how strictly to enforce TLS encryption.
487#[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq, JsonSchema)]
488pub enum TlsMode {
489    /// Allow TLS encryption.
490    Allow,
491    /// Require that clients negotiate TLS encryption.
492    Require,
493}
494
495/// Configures TLS encryption for connections.
496#[derive(Debug, Clone)]
497pub struct TlsCertConfig {
498    /// The path to the TLS certificate.
499    pub cert: PathBuf,
500    /// The path to the TLS key.
501    pub key: PathBuf,
502}
503
504impl TlsCertConfig {
505    /// Returns the SSL context to use in TlsConfigs.
506    pub fn load_context(&self) -> Result<SslContext, anyhow::Error> {
507        // Mozilla publishes three presets: old, intermediate, and modern. They
508        // recommend the intermediate preset for general purpose servers, which
509        // is what we use, as it is compatible with nearly every client released
510        // in the last five years but does not include any known-problematic
511        // ciphers. We once tried to use the modern preset, but it was
512        // incompatible with Fivetran, and presumably other JDBC-based tools.
513        let mut builder = SslAcceptor::mozilla_intermediate_v5(SslMethod::tls())?;
514        builder.set_certificate_chain_file(&self.cert)?;
515        builder.set_private_key_file(&self.key, SslFiletype::PEM)?;
516        Ok(builder.build().into_context())
517    }
518
519    /// Like [Self::load_context] but attempts to reload the files each time `ticker` yields an item.
520    /// Returns an error based on the files currently on disk. When `ticker` receives, the
521    /// certificates are reloaded from the context. The result of the reloading is returned on the
522    /// oneshot if present, and an Ok result means new connections will use the new certificates. An
523    /// Err result will not change the current certificates.
524    pub fn reloading_context(
525        &self,
526        mut ticker: ReloadTrigger,
527    ) -> Result<ReloadingSslContext, anyhow::Error> {
528        let context = Arc::new(RwLock::new(self.load_context()?));
529        let updater_context = Arc::clone(&context);
530        let config = self.clone();
531        mz_ore::task::spawn(|| "TlsCertConfig reloading_context", async move {
532            while let Some(chan) = ticker.next().await {
533                let result = match config.load_context() {
534                    Ok(ctx) => {
535                        *updater_context.write().expect("poisoned") = ctx;
536                        Ok(())
537                    }
538                    Err(err) => {
539                        tracing::error!("failed to reload SSL certificate: {err}");
540                        Err(err)
541                    }
542                };
543                if let Some(chan) = chan {
544                    let _ = chan.send(result);
545                }
546            }
547            tracing::warn!("TlsCertConfig reloading_context updater closed");
548        });
549        Ok(ReloadingSslContext { context })
550    }
551}
552
553/// An SslContext whose inner value can be updated.
554#[derive(Clone, Debug)]
555pub struct ReloadingSslContext {
556    /// The current SSL context.
557    context: Arc<RwLock<SslContext>>,
558}
559
560impl ReloadingSslContext {
561    pub fn get(&self) -> RwLockReadGuard<'_, SslContext> {
562        self.context.read().expect("poisoned")
563    }
564}
565
566/// Configures a server's TLS encryption and authentication with reloading.
567#[derive(Clone, Debug)]
568pub struct ReloadingTlsConfig {
569    /// The SSL context used to manage incoming TLS negotiations.
570    pub context: ReloadingSslContext,
571    /// The TLS mode.
572    pub mode: TlsMode,
573}
574
575pub type ReloadTrigger = BoxStream<'static, Option<oneshot::Sender<Result<(), anyhow::Error>>>>;
576
577/// Returns a ReloadTrigger that triggers once per hour.
578pub fn default_cert_reload_ticker() -> ReloadTrigger {
579    let ticker = IntervalStream::new(tokio::time::interval(Duration::from_secs(60 * 60)));
580    let ticker = ticker.map(|_| None);
581    let ticker = Box::pin(ticker);
582    ticker
583}
584
585/// Returns a ReloadTrigger that never triggers.
586pub fn cert_reload_never_reload() -> ReloadTrigger {
587    let ticker = futures::stream::empty();
588    let ticker = Box::pin(ticker);
589    ticker
590}
591
592/// Command line arguments for TLS.
593#[derive(Debug, Clone, clap::Parser)]
594pub struct TlsCliArgs {
595    /// How stringently to demand TLS authentication and encryption.
596    ///
597    /// If set to "disable", then environmentd rejects HTTP and PostgreSQL
598    /// connections that negotiate TLS.
599    ///
600    /// If set to "require", then environmentd requires that all HTTP and
601    /// PostgreSQL connections negotiate TLS. Unencrypted connections will be
602    /// rejected.
603    #[clap(
604        long, env = "TLS_MODE",
605        value_parser = ["disable", "require"],
606        default_value = "disable",
607        default_value_ifs = [
608            ("frontegg_tenant", ArgPredicate::IsPresent, Some("require")),
609            ("frontegg_resolver_template", ArgPredicate::IsPresent, Some("require")),
610        ],
611        value_name = "MODE",
612    )]
613    tls_mode: String,
614    /// Certificate file for TLS connections.
615    #[clap(
616        long,
617        env = "TLS_CERT",
618        requires = "tls_key",
619        required_if_eq_any([("tls_mode", "require")]),
620        value_name = "PATH"
621    )]
622    tls_cert: Option<PathBuf>,
623    /// Private key file for TLS connections.
624    #[clap(
625        long,
626        env = "TLS_KEY",
627        requires = "tls_cert",
628        required_if_eq_any([("tls_mode", "require")]),
629        value_name = "PATH"
630    )]
631    tls_key: Option<PathBuf>,
632}
633
634impl TlsCliArgs {
635    /// Convert args into configuration.
636    pub fn into_config(self) -> Result<Option<TlsCertConfig>, anyhow::Error> {
637        if self.tls_mode == "disable" {
638            if self.tls_cert.is_some() {
639                bail!("cannot specify --tls-mode=disable and --tls-cert simultaneously");
640            }
641            if self.tls_key.is_some() {
642                bail!("cannot specify --tls-mode=disable and --tls-key simultaneously");
643            }
644            Ok(None)
645        } else {
646            let cert = self.tls_cert.unwrap();
647            let key = self.tls_key.unwrap();
648            Ok(Some(TlsCertConfig { cert, key }))
649        }
650    }
651}
652
653// macOS caps the backlog at `kern.ipc.somaxconn`, 128 by default, so the queue
654// length is only meaningful on Linux.
655#[cfg(all(test, target_os = "linux"))]
656mod tests {
657    use std::net::SocketAddr;
658    use std::time::Duration;
659
660    use tokio::net::TcpStream;
661
662    use super::{LISTEN_BACKLOG, listen};
663
664    #[mz_ore::test(tokio::test)]
665    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `socket`
666    async fn listen_queues_backlog_connections_before_accepting() {
667        let (handle, _conns) = listen(&SocketAddr::from(([127, 0, 0, 1], 0))).unwrap();
668        let addr = handle.local_addr();
669        // Nothing accepts these, so each one stays in the accept queue even
670        // after the client closes it, as a TCP probe does.
671        for i in 0..LISTEN_BACKLOG {
672            let stream = tokio::time::timeout(Duration::from_secs(1), TcpStream::connect(addr))
673                .await
674                .unwrap_or_else(|_| panic!("connection {i} timed out, accept queue is full"))
675                .unwrap();
676            drop(stream);
677        }
678    }
679}
mz_server_core/lib.rs

mz_server_core/
lib.rs
