Function tcp_acceptor

src/main.rs:298–415

(globals: Arc<Globals>, tcp_listener: TcpListener)

}

async fn tcp_acceptor(globals: Arc<Globals>, tcp_listener: TcpListener) -> Result<(), Error> {
    let runtime_handle = globals.runtime_handle.clone();
    let timeout = globals.tcp_timeout;
    let concurrent_connections = globals.tcp_concurrent_connections.clone();
    let active_connections = globals.tcp_active_connections.clone();
    loop {
        let (mut client_connection, client_addr) = match tcp_listener.accept().await {
            Ok(x) => x,
            Err(e) => {
                if e.kind() == std::io::ErrorKind::WouldBlock {
                    continue;
                }
                error!("TCP accept error: {}", e);

                // Rate limit repeated errors to avoid spinning
                let is_resource_error = matches!(
                    e.kind(),
                    std::io::ErrorKind::ConnectionRefused
                        | std::io::ErrorKind::ConnectionReset
                        | std::io::ErrorKind::ConnectionAborted
                        | std::io::ErrorKind::AddrInUse
                        | std::io::ErrorKind::AddrNotAvailable
                );

                if is_resource_error {
                    // For resource-related errors, try to free up connections
                    let mut connections = active_connections.lock();
                    let freed = connections.len().min(5);
                    for _ in 0..freed {
                        if let Some(tx_oldest) = connections.pop_back() {
                            let _ = tx_oldest.send(());
                        }
                    }
                    info!("Freed {} connections to recover from resource error", freed);
                }

                // Add delay to prevent CPU spinning on persistent errors
                tokio::time::sleep(Duration::from_secs(1)).await;
                continue;
            }
        };

        if let Some(ref rate_limiter) = globals.rate_limiter {
            if !rate_limiter.is_allowed(client_addr.ip()) {
                debug!("Rate limit exceeded for {}", client_addr.ip());
                #[cfg(feature = "metrics")]
                globals.varz.client_queries_rate_limited.inc();
                continue;
            }
        }

        let (tx, rx) = oneshot::channel::<()>();
        let tx_channel_index = {
            let mut active_connections = active_connections.lock();
            if active_connections.is_full() {
                let tx_oldest = active_connections.pop_back().unwrap();
                let _ = tx_oldest.send(());
            }