L4 连接

    /// Create a new [`HttpPeer`] with the given socket address and TLS settings.
    pub fn new<A: ToInetSocketAddrs>(address: A, tls: bool, sni: String) -> Self {
        let mut addrs_iter = address.to_socket_addrs().unwrap(); //TODO: handle error
        let addr = addrs_iter.next().unwrap();
        Self::new_from_sockaddr(SocketAddr::Inet(addr), tls, sni)
    }

struct SlowResolve(&'static str);
impl ToSocketAddrs for SlowResolve {
  type Iter = std::vec::IntoIter<SocketAddr>;
  fn to_socket_addrs(&self) -> std::io::Result<Self::Iter> {
    std::thread::sleep(Duration::from_millis(1000));
    self.0.to_socket_addrs()
  }
}

#[async_trait]
impl ProxyHttp for P {
  type CTX = ();
  fn new_ctx(&self) {}
  async fn upstream_peer(&self, s: &mut Session, _: &mut ()) -> Result<Box<HttpPeer>> {
    let path = s.req_header().uri.path().to_string();
    let t = Instant::now();
    let thread = std::thread::current().name().unwrap_or("?").to_string();
    let peer = match path.as_str() {
      "/slow" => HttpPeer::new(SlowResolve("127.0.0.1:18481"), false, String::new()),
      "/name" => HttpPeer::new("localhost:18481", false, String::new()),
      "/bad" => HttpPeer::new("no-such-host.invalid:80", false, String::new()),
      _ => HttpPeer::new("127.0.0.1:18481", false, String::new()),
    };
    eprintln!("upstream_peer {path} on thread {thread:?}: HttpPeer::new took {:?}, addr {}", t.elapsed(), peer._address);
    Ok(Box::new(peer))
  }
}

upstream_peer /name on thread "Pingora HTTP Proxy Service": HttpPeer::new took 2.378042ms, addr [::1]:18481
[2026-10-29T04:53:15Z ERROR pingora_proxy] Fail to proxy: Upstream ConnectRefused context: Fail to connect to addr: [::1]:18481, scheme: HTTP cause:  context: Fail to connect to [::1]:18481 cause: Connection refused (os error 61), status: 502, tries: 1, retry: false, GET /name, Host: 127.0.0.1:18480

thread 'Pingora HTTP Proxy Service' (5361160) panicked at ~/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/pingora-core-0.9.0/src/upstreams/peer.rs:719:56:
called `Result::unwrap()` on an `Err` value: Custom { kind: Uncategorized, error: "failed to lookup address information: nodename nor servname provided, or not known" }

async fn do_connect_inner<P: Peer + Send + Sync>(
    // ...
) -> Result<Stream> {
    let l4_connect_start = Instant::now();
    let mut stream = l4_connect(peer, bind_to).await?;
    stream.set_establishment_timing(l4_connect_start.elapsed(), offload_wait_duration);
    if peer.tls() {
        // Capture the local address before the handshake consumes the stream, so a TLS failure
        // can still report the address we connected from.
        // ...
        let tls_stream = tls::connect(stream, peer, alpn_override, tls_ctx)
            .await
            .map_err(|e| attach_connect_local_addr(e, local_addr))?;
        Ok(Box::new(tls_stream))
    } else {
        Ok(Box::new(stream))
    }
}

                    let connect_future = tcp_connect(
                        addr,
                        bind_to.as_ref(),
                        |socket| {
                            // ...
                            if peer.tcp_fast_open() {
                                set_tcp_fastopen_connect(raw)?;
                            }
                            if let Some(recv_buf) = peer.tcp_recv_buf() {
                                debug!("Setting recv buf size");
                                set_recv_buf(raw, recv_buf)?;
                            }
                            if let Some(dscp) = peer.dscp() {
                                debug!("Setting dscp");
                                set_dscp(raw, dscp)?;
                            }

                            if let Some(tweak_hook) = peer
                                .get_peer_options()
                                .and_then(|o| o.upstream_tcp_sock_tweak_hook.clone())
                            {
                                tweak_hook(socket)?;
                            }

                            Ok(())
                        },
                        &mut connect_attempt,
                    );

    let socket = into_socket2(socket);
    if let Err(error) = socket.connect(&(*addr).into()) {
        attempt.record_local_addr(&socket);
        if !connect_in_progress(&error) {
            return Err(wrap_os_connect_error(
                error,
                format!("Fail to connect to {}", *addr),
                attempt.local_addr(),
            ));
        }
    } else {
        attempt.record_local_addr(&socket);
    }

    let stream = TcpStream::from_std(socket.into())
        .or_err(SocketError, "failed to register connecting socket")?;
    stream
        .writable()
        .await
        .or_err(ConnectError, "failed to wait for connecting socket")?;

    if let Some(error) = stream
        .take_error()
        .or_err(SocketError, "failed to get connecting socket error")?
    {
        return Err(wrap_os_connect_error(
            error,
            format!("Fail to connect to {}", *addr),
            attempt.local_addr(),
        ));
    }

    let connect_future = handshake(ssl_conf, peer.sni(), stream, complete_hook);

    match peer.connection_timeout() {
        Some(t) => match pingora_timeout::timeout(t, connect_future).await {
            Ok(res) => res,
            Err(_) => Error::e_explain(
                ConnectTimedout,
                format!("connecting to server {}, timeout {:?}", peer, t),
            ),
        },
        None => connect_future.await,
    }

async fn try_connect(label: &str, c: &TransportConnector, peer: &HttpPeer) {
  let t = Instant::now();
  // our own 3s guard, so a hang shows up as "still waiting"
  match tokio::time::timeout(Duration::from_secs(3), c.new_stream(peer)).await {
    Ok(Ok(_)) => println!("{label}: connected after {:?}", t.elapsed()),
    Ok(Err(e)) => println!("{label}: {:?} after {:?}\n    {e}", e.etype(), t.elapsed()),
    Err(_) => println!("{label}: still waiting after {:?}", t.elapsed()),
  }
}

#[tokio::main]
async fn main() {
  let c = TransportConnector::new(None);

  // 1. nothing listens on 18489
  let p = HttpPeer::new("127.0.0.1:18489", false, String::new());
  try_connect("refused", &c, &p).await;

  // 3. TCP accepted, TLS ServerHello never comes
  let l = TcpListener::bind("127.0.0.1:18483").await.unwrap();
  tokio::spawn(async move {
    let mut keep = vec![];
    loop {
      let (s, _) = l.accept().await.unwrap();
      keep.push(s); // hold it open, never answer
    }
  });
  let mut p = HttpPeer::new("127.0.0.1:18483", true, "silent.example".to_string());
  try_connect("tls silent, no timeouts", &c, &p).await;
  p.options.connection_timeout = Some(Duration::from_millis(300));
  try_connect("tls silent, connection_timeout 300ms", &c, &p).await;
  p.options.connection_timeout = None;
  p.options.total_connection_timeout = Some(Duration::from_millis(500));
  try_connect("tls silent, total_connection_timeout 500ms", &c, &p).await;
}

refused: ConnectRefused after 698.834µs
     ConnectRefused context: Fail to connect to addr: 127.0.0.1:18489, scheme: HTTP cause:  context: Fail to connect to 127.0.0.1:18489 cause: Connection refused (os error 61)
tls silent, no timeouts: still waiting after 3.004976208s
tls silent, connection_timeout 300ms: ConnectTimedout after 314.547958ms
     ConnectTimedout context: connecting to server addr: 127.0.0.1:18483, scheme: HTTPS, sni: silent.example, timeout 300ms cause: local address 127.0.0.1:62762
tls silent, total_connection_timeout 500ms: ConnectTimedout after 512.804459ms
     ConnectTimedout context: connecting to server addr: 127.0.0.1:18483, scheme: HTTPS, sni: silent.example, total-connection timeout 500ms

--- /fast alone
/fast 200 0.006213s
--- /slow, then /fast 100ms later
/slow 200 1.013188s
/fast 200 0.885383s
--- two /slow together
/slow 200 2.020329s
/slow 200 2.020173s
--- /name
/name 502 0.004284s
--- /bad
* Empty reply from server
* Closing connection
--- /fast after /bad
/fast 200 0.001504s

upstream_peer /fast on thread "Pingora HTTP Proxy Service": HttpPeer::new took 1.375µs, addr 127.0.0.1:18481
upstream_peer /slow on thread "Pingora HTTP Proxy Service": HttpPeer::new took 1.010022166s, addr 127.0.0.1:18481
upstream_peer /fast on thread "Pingora HTTP Proxy Service": HttpPeer::new took 1.25µs, addr 127.0.0.1:18481
upstream_peer /slow on thread "Pingora HTTP Proxy Service": HttpPeer::new took 1.01003575s, addr 127.0.0.1:18481
upstream_peer /slow on thread "Pingora HTTP Proxy Service": HttpPeer::new took 1.004554125s, addr 127.0.0.1:18481
...