1//! The HTTPS adapter behind `jev-client`'s ports: one HTTP/2 connection
2//! per backend over rustls, driven by the executor, and a `Runtime` on
3//! the executor's timers. It sends and classifies; retrying, budgets and
4//! statuses are the client's (contract §2).
5
6use std::cell::RefCell;
7use std::rc::Rc;
8use std::task::Waker;
9use std::future::Future;
10use std::net::{IpAddr, SocketAddr};
11use std::pin::Pin;
12use std::sync::Arc;
13use std::sync::atomic::{AtomicU64, Ordering};
14use std::time::{Duration, Instant, SystemTime};
15
16use bytes::Bytes;
17use futures_rustls::TlsConnector;
18use http_body_util::{BodyExt, Full, LengthLimitError, Limited};
19use hyper::client::conn::http2::SendRequest;
20use jev_client::{HttpRequest, HttpResponse, Runtime, Transport, TransportError, TransportErrorKind};
21use rustls::pki_types::pem::PemObject;
22use rustls::pki_types::{CertificateDer, ServerName};
23use rustls::{ClientConfig, RootCertStore};
24
25use crate::Error;
26use crate::dns::{self, Servers};
27use crate::executor::{self, random_unit};
28use crate::net::{HyperIo, PeerSettings, TcpIo};
29use crate::ratelimit::RateLimit;
30
31/// Cloudflare closes idle connections at 400 s; reconnect well before.
32const IDLE_LIMIT: Duration = Duration::from_secs(300);
33/// The response cap (pg_typesafe's 8 MB).
34const MAX_BODY: usize = 8 << 20;
35/// Time for each address but the last, so one blackholed address (a
36/// broken IPv6 route, a dead IP) cannot spend a whole attempt.
37const CONNECT_PER_ADDRESS: Duration = Duration::from_secs(2);
38/// Streams assumed before a peer has said: RFC 9113 §6.5.2 recommends
39/// no fewer than 100, and api.typesafe.ai advertises 100 (contract §2).
40const ASSUMED_STREAMS: u32 = 100;
41/// No limit seen yet on this connection.
42const UNKNOWN: u64 = u64::MAX;
43
44/// Where requests go: the URL, how its host is reached, which CA file
45/// (if any) is trusted besides the Mozilla roots, and how the connection
46/// is kept alive. A connection is reused only for an equal endpoint, so
47/// changing any of these opens a new one.
48#[derive(Clone, PartialEq)]
49pub struct Endpoint {
50    uri: http::Uri,
51    host: Host,
52    port: u16,
53    server_name: ServerName<'static>,
54    ca_file: Option<String>,
55    keep_alive: KeepAlive,
56}
57
58/// A ping after `interval` without a frame from the server, while a scan
59/// runs the executor; the connection is closed if one goes unanswered for
60/// `timeout` (contract §2: 20 s and 10 s).
61#[derive(Clone, Copy, PartialEq, Debug)]
62pub struct KeepAlive {
63    pub interval: Duration,
64    pub timeout: Duration,
65}
66
67impl Default for KeepAlive {
68    fn default() -> Self {
69        KeepAlive { interval: Duration::from_secs(20), timeout: Duration::from_secs(10) }
70    }
71}
72
73#[derive(Clone, PartialEq)]
74enum Host {
75    Ip(IpAddr),
76    Name { name: String, servers: Servers },
77}
78
79impl Endpoint {
80    pub fn parse(url: &str, ca_file: Option<&str>, dns_servers: Option<&str>) -> Result<Self, Error> {
81        let config = |e: String| Error::Config(e);
82        let uri: http::Uri = url.parse().map_err(|e| config(format!("jev.endpoint {url}: {e}")))?;
83        if uri.scheme() != Some(&http::uri::Scheme::HTTPS) {
84            return Err(config(format!("jev.endpoint must be https: {url}")));
85        }
86        let host = uri.host().ok_or_else(|| config(format!("jev.endpoint has no host: {url}")))?;
87        let (host, server_name) = match host.trim_start_matches('[').trim_end_matches(']').parse::<IpAddr>() {
88            Ok(ip) => (Host::Ip(ip), ServerName::IpAddress(ip.into())),
89            Err(_) => (
90                Host::Name { name: host.to_owned(), servers: Servers::parse(dns_servers)? },
91                ServerName::try_from(host.to_owned())
92                    .map_err(|e| config(format!("jev.endpoint host {host}: {e}")))?,
93            ),
94        };
95        Ok(Endpoint {
96            port: uri.port_u16().unwrap_or(443),
97            host,
98            server_name,
99            ca_file: ca_file.map(str::to_owned),
100            keep_alive: KeepAlive::default(),
101            uri,
102        })
103    }
104
105    pub fn with_keep_alive(mut self, keep_alive: KeepAlive) -> Self {
106        self.keep_alive = keep_alive;
107        self
108    }
109
110    /// Every address to try, in the resolver's order (IPv4 first).
111    async fn addrs(&self) -> Result<Vec<SocketAddr>, TransportError> {
112        let ips = match &self.host {
113            Host::Ip(ip) => vec![*ip],
114            Host::Name { name, servers } => dns::resolve(name, servers).await?,
115        };
116        Ok(ips.into_iter().map(|ip| SocketAddr::new(ip, self.port)).collect())
117    }
118}
119
120/// The `Transport` port for one endpoint.
121pub struct HttpsTransport {
122    endpoint: Endpoint,
123    limit: Option<RateLimit>,
124}
125
126impl HttpsTransport {
127    pub fn new(endpoint: Endpoint) -> Self {
128        HttpsTransport { endpoint, limit: None }
129    }
130
131    /// Every attempt first waits for the cluster's rate limit, and its
132    /// outcome feeds the limit back (429s halve it, answers restore it
133    /// and correct the token estimate).
134    pub fn rate_limited(mut self, limit: RateLimit) -> Self {
135        self.limit = Some(limit);
136        self
137    }
138}
139
140impl Transport for HttpsTransport {
141    async fn admit(&self, request: &HttpRequest) {
142        if let Some(limit) = &self.limit {
143            limit.admit(&request.body).await;
144        }
145    }
146
147    async fn send(&self, request: HttpRequest) -> Result<HttpResponse, TransportError> {
148        let limit = self.limit;
149        let sent = request.body.clone();
150        let result = self.exchange(request).await;
151        if let Some(limit) = limit {
152            match &result {
153                Ok(response) => limit.settle(&sent, response.status, &response.body),
154                Err(e) if e.kind == TransportErrorKind::NotSent => limit.unsent(&sent),
155                Err(_) => {}
156            }
157        }
158        result
159    }
160
161    fn distrust_connection(&self) {
162        forget_connection();
163    }
164}
165
166impl HttpsTransport {
167    async fn exchange(&self, request: HttpRequest) -> Result<HttpResponse, TransportError> {
168        let mut sender = connection(&self.endpoint).await?;
169        let mut builder = http::Request::post(self.endpoint.uri.clone());
170        *builder.headers_mut().expect("a fresh builder has headers") = request.headers;
171        let request = builder
172            .body(Full::new(request.body))
173            .map_err(|e| TransportError::new(TransportErrorKind::Config, format!("building the request: {e}")))?;
174
175        // From here until the body is read, or this future is dropped.
176        let _stream = crate::stats::InFlight::open();
177        let response = match sender.try_send_request(request).await {
178            Ok(response) => response,
179            Err(mut e) => {
180                forget_connection();
181                // hyper hands the request back when it never left.
182                let kind = if e.take_message().is_some() || never_processed(e.error()) {
183                    TransportErrorKind::NotSent
184                } else {
185                    TransportErrorKind::Interrupted
186                };
187                return Err(TransportError::new(kind, format!("sending to {}: {}", self.endpoint.uri, e.error())));
188            }
189        };
190        let (parts, body) = response.into_parts();
191        let body = match Limited::new(body, MAX_BODY).collect().await {
192            Ok(collected) => collected.to_bytes(),
193            Err(e) if e.is::<LengthLimitError>() => {
194                return Err(TransportError::new(
195                    TransportErrorKind::TooLarge,
196                    format!("Jev's response exceeded {} MiB", MAX_BODY >> 20),
197                ));
198            }
199            Err(e) => {
200                forget_connection();
201                return Err(TransportError::new(TransportErrorKind::Interrupted, format!("reading Jev's response: {e}")));
202            }
203        };
204        touch_connection();
205        Ok(HttpResponse { status: parts.status.as_u16(), headers: parts.headers, body })
206    }
207}
208
209/// True when h2 says the server never processed the stream: refused
210/// (REFUSED_STREAM), or cut by a GOAWAY whose last stream id is below
211/// ours. Such a request can be resent without being billed twice.
212fn never_processed(error: &hyper::Error) -> bool {
213    let mut source: Option<&(dyn std::error::Error + 'static)> = std::error::Error::source(error);
214    while let Some(e) = source {
215        if let Some(h2) = e.downcast_ref::<h2::Error>() {
216            return h2.reason() == Some(h2::Reason::REFUSED_STREAM) || (h2.is_go_away() && h2.is_remote());
217        }
218        source = e.source();
219    }
220    false
221}
222
223/// The `Runtime` port on the executor: its timers and Postgres's PRNG.
224pub struct Backend;
225
226impl Runtime for Backend {
227    fn now(&self) -> Instant {
228        Instant::now()
229    }
230
231    fn wall_clock(&self) -> SystemTime {
232        SystemTime::now()
233    }
234
235    fn sleep(&self, duration: Duration) -> impl Future<Output = ()> {
236        executor::sleep(duration)
237    }
238
239    fn timeout<F: Future>(&self, duration: Duration, future: F) -> impl Future<Output = Option<F::Output>> {
240        executor::timeout(duration, future)
241    }
242
243    fn jitter(&self) -> f64 {
244        random_unit()
245    }
246}
247
248struct Connection {
249    endpoint: Endpoint,
250    sender: SendRequest<Full<Bytes>>,
251    last_used: Instant,
252}
253
254thread_local! {
255    static CONNECTION: RefCell<Option<Connection>> = const { RefCell::new(None) };
256    static DIALING: RefCell<Option<Rc<Dial>>> = const { RefCell::new(None) };
257    /// Keyed by the CA file's path and modification time, so a rotated
258    /// file is picked up by the next connection (review #16).
259    static TLS: RefCell<Option<(CaSource, Arc<ClientConfig>)>> = const { RefCell::new(None) };
260    /// The newest connection's `SETTINGS_MAX_CONCURRENT_STREAMS`, written
261    /// by its reader as frames arrive. It outlives the connection, so a
262    /// reconnect to the same endpoint starts from the last value seen.
263    static PEER_STREAMS: RefCell<Option<(Endpoint, Arc<AtomicU64>)>> = const { RefCell::new(None) };
264}
265
266/// How many streams the peer at `endpoint` accepts at once: the last
267/// `SETTINGS_MAX_CONCURRENT_STREAMS` it sent, or [`ASSUMED_STREAMS`]
268/// before any. hyper queues streams past the limit without saying so, so
269/// `jev.concurrency` is clamped to this (contract §2).
270pub fn stream_limit(endpoint: &Endpoint) -> usize {
271    PEER_STREAMS.with(|p| match p.borrow().as_ref() {
272        Some((seen, max)) if seen == endpoint => match max.load(Ordering::Relaxed) {
273            UNKNOWN => ASSUMED_STREAMS,
274            n => n as u32,
275        },
276        _ => ASSUMED_STREAMS,
277    }) as usize
278}
279
280/// A handle on the backend's connection. The connection stays cached
281/// while requests run on it, so a cancelled request costs only its
282/// stream; it is replaced when closed, idle past the limit, or distrusted.
283async fn connection(endpoint: &Endpoint) -> Result<SendRequest<Full<Bytes>>, TransportError> {
284    // Between statements nothing polls hyper's connection driver, so a
285    // GOAWAY or a close the server sent meanwhile is still unread and
286    // `is_closed` would say the connection is fine. Two passes: one
287    // wakes the driver on the readable socket, the next lets it run.
288    if CONNECTION.with(|c| c.borrow().is_some()) {
289        executor::yield_now().await;
290        executor::yield_now().await;
291    }
292    let cached = CONNECTION.with(|c| {
293        let mut c = c.borrow_mut();
294        match c.as_mut() {
295            Some(conn)
296                if conn.endpoint == *endpoint
297                    && conn.last_used.elapsed() < IDLE_LIMIT
298                    && !conn.sender.is_closed() =>
299            {
300                conn.last_used = Instant::now();
301                Some(conn.sender.clone())
302            }
303            _ => {
304                *c = None;
305                None
306            }
307        }
308    });
309    if let Some(sender) = cached {
310        return Ok(sender);
311    }
312    // One dial at a time: a scan starts a window of requests at once, and
313    // each must not open its own connection.
314    let joined = DIALING.with(|d| d.borrow().as_ref().filter(|dial| dial.endpoint == *endpoint).cloned());
315    if let Some(dial) = joined {
316        return std::future::poll_fn(|cx| match dial.result.borrow().as_ref() {
317            Some(result) => std::task::Poll::Ready(result.clone()),
318            None => {
319                dial.waiters.borrow_mut().push(cx.waker().clone());
320                std::task::Poll::Pending
321            }
322        })
323        .await;
324    }
325    let dial = Rc::new(Dial { endpoint: endpoint.clone(), result: RefCell::new(None), waiters: RefCell::new(Vec::new()) });
326    DIALING.with(|d| *d.borrow_mut() = Some(dial.clone()));
327    let mut publish = Publish(Some(dial));
328    let result = connect(endpoint).await;
329    if let Ok(sender) = &result {
330        crate::stats::connected(&endpoint.uri);
331        CONNECTION.with(|c| {
332            *c.borrow_mut() =
333                Some(Connection { endpoint: endpoint.clone(), sender: sender.clone(), last_used: Instant::now() })
334        });
335    }
336    publish.finish(result.clone());
337    result
338}
339
340/// A dial in progress, which later senders wait on.
341struct Dial {
342    endpoint: Endpoint,
343    result: RefCell<Option<Result<SendRequest<Full<Bytes>>, TransportError>>>,
344    waiters: RefCell<Vec<Waker>>,
345}
346
347/// Publishes the dial's outcome to its waiters, exactly once. Dropped
348/// unfinished (the dialing request was cancelled), it tells them nothing
349/// was sent, so each resends and one of them dials again.
350struct Publish(Option<Rc<Dial>>);
351
352impl Publish {
353    fn finish(&mut self, result: Result<SendRequest<Full<Bytes>>, TransportError>) {
354        let Some(dial) = self.0.take() else { return };
355        DIALING.with(|d| {
356            let mut d = d.borrow_mut();
357            if d.as_ref().is_some_and(|current| Rc::ptr_eq(current, &dial)) {
358                *d = None;
359            }
360        });
361        *dial.result.borrow_mut() = Some(result);
362        for waker in dial.waiters.borrow_mut().drain(..) {
363            waker.wake();
364        }
365    }
366}
367
368impl Drop for Publish {
369    fn drop(&mut self) {
370        // try_with: this may run during thread-local teardown.
371        if self.0.is_some() && DIALING.try_with(|_| ()).is_ok() {
372            self.finish(Err(TransportError::new(TransportErrorKind::NotSent, "the connection was being opened by a cancelled request")));
373        }
374    }
375}
376
377fn forget_connection() {
378    CONNECTION.with(|c| *c.borrow_mut() = None);
379}
380
381fn touch_connection() {
382    CONNECTION.with(|c| {
383        if let Some(conn) = c.borrow_mut().as_mut() {
384            conn.last_used = Instant::now();
385        }
386    });
387}
388
389async fn connect(endpoint: &Endpoint) -> Result<SendRequest<Full<Bytes>>, TransportError> {
390    use TransportErrorKind::{Config, Connect, Tls};
391    let config = tls_config(endpoint.ca_file.as_deref()).map_err(|e| TransportError::new(Config, e.to_string()))?;
392
393    let addrs = endpoint.addrs().await?;
394    let mut failures = Vec::new();
395    let mut tcp = None;
396    for (i, addr) in addrs.iter().enumerate() {
397        let attempt = TcpIo::connect(*addr);
398        let result = if i + 1 < addrs.len() {
399            executor::timeout(CONNECT_PER_ADDRESS, attempt).await.unwrap_or_else(|| {
400                Err(std::io::Error::new(std::io::ErrorKind::TimedOut, "no answer within 2 s"))
401            })
402        } else {
403            attempt.await
404        };
405        match result {
406            Ok(io) => {
407                tcp = Some(io);
408                break;
409            }
410            Err(e) => failures.push(format!("{addr}: {e}")),
411        }
412    }
413    let tcp = tcp.ok_or_else(|| {
414        TransportError::new(Connect, format!("connecting to {}: {}", endpoint.uri, failures.join("; ")))
415    })?;
416
417    let tls = TlsConnector::from(config).connect(endpoint.server_name.clone(), tcp).await.map_err(|e| {
418        // rustls refusing (a certificate, a protocol alert) will refuse
419        // again; an I/O failure mid-handshake is the network.
420        let refused = e.get_ref().is_some_and(|inner| inner.is::<rustls::Error>());
421        TransportError::new(if refused { Tls } else { Connect }, format!("TLS with {}: {e}", endpoint.uri))
422    })?;
423    if tls.get_ref().1.alpn_protocol() != Some(b"h2") {
424        return Err(TransportError::new(Tls, format!("{} did not negotiate HTTP/2", endpoint.uri)));
425    }
426    let max_streams = PEER_STREAMS.with(|p| {
427        let seed = p.borrow().as_ref().filter(|(seen, _)| seen == endpoint).map_or(UNKNOWN, |(_, m)| m.load(Ordering::Relaxed));
428        let max = Arc::new(AtomicU64::new(seed));
429        *p.borrow_mut() = Some((endpoint.clone(), max.clone()));
430        max
431    });
432    let tls = PeerSettings::new(tls, move |n| max_streams.store(u64::from(n), Ordering::Relaxed));
433    // hyper's fixed receive windows (5 MB per connection, 2 MB per
434    // stream) are below 8 MB answers times the scan's window; BDP-based
435    // adaptive windows grow to what the path carries (contract §2).
436    let (sender, connection) = hyper::client::conn::http2::Builder::new(LocalExec)
437        .adaptive_window(true)
438        // Pings find a dead path before the attempt timeout would, and
439        // keep a long scan's quiet connection open (contract §2). They
440        // run only while a scan polls the executor.
441        .timer(BackendTimer)
442        .keep_alive_interval(endpoint.keep_alive.interval)
443        .keep_alive_timeout(endpoint.keep_alive.timeout)
444        .keep_alive_while_idle(true)
445        .handshake(HyperIo(tls))
446        .await
447        .map_err(|e| TransportError::new(Connect, format!("HTTP/2 with {}: {e}", endpoint.uri)))?;
448    executor::spawn_local(async move {
449        let _ = connection.await;
450    });
451    Ok(sender)
452}
453
454/// Which extra CA file, as last seen on disk.
455#[derive(Clone, PartialEq)]
456struct CaSource {
457    path: Option<String>,
458    modified: Option<SystemTime>,
459}
460
461/// Built on first use, not in `_PG_init`, and rebuilt when `jev.ca_file`
462/// names another file or the file changes.
463fn tls_config(ca_file: Option<&str>) -> Result<Arc<ClientConfig>, Error> {
464    let source = CaSource {
465        path: ca_file.map(str::to_owned),
466        modified: ca_file.and_then(|p| std::fs::metadata(p).and_then(|m| m.modified()).ok()),
467    };
468    if let Some(config) =
469        TLS.with(|t| t.borrow().as_ref().filter(|(s, _)| *s == source).map(|(_, c)| c.clone()))
470    {
471        return Ok(config);
472    }
473    let mut roots = RootCertStore { roots: webpki_roots::TLS_SERVER_ROOTS.to_vec() };
474    if let Some(path) = ca_file {
475        for cert in CertificateDer::pem_file_iter(path).map_err(|e| Error::Config(format!("jev.ca_file {path}: {e}")))? {
476            let cert = cert.map_err(|e| Error::Config(format!("jev.ca_file {path}: {e}")))?;
477            roots.add(cert).map_err(|e| Error::Config(format!("jev.ca_file {path}: {e}")))?;
478        }
479    }
480    let mut config = ClientConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider()))
481        .with_safe_default_protocol_versions()
482        .map_err(|e| Error::Config(e.to_string()))?
483        .with_root_certificates(roots)
484        .with_no_client_auth();
485    config.alpn_protocols = vec![b"h2".to_vec()];
486    let config = Arc::new(config);
487    TLS.with(|t| *t.borrow_mut() = Some((source, config.clone())));
488    Ok(config)
489}
490
491/// hyper's timers (keep-alive pings) on the executor's.
492#[derive(Clone, Copy)]
493struct BackendTimer;
494
495impl hyper::rt::Timer for BackendTimer {
496    fn sleep(&self, duration: Duration) -> Pin<Box<dyn hyper::rt::Sleep>> {
497        Box::pin(executor::sleep(duration))
498    }
499
500    fn sleep_until(&self, deadline: Instant) -> Pin<Box<dyn hyper::rt::Sleep>> {
501        Box::pin(executor::sleep_until(deadline))
502    }
503}
504
505// Send + Sync: a deadline and a timer id; it is polled only on the
506// backend thread, where its timer lives.
507impl hyper::rt::Sleep for executor::Sleep {}
508
509/// hyper spawns its connection driver here.
510#[derive(Clone, Copy)]
511struct LocalExec;
512
513impl<F: Future<Output = ()> + 'static> hyper::rt::Executor<F> for LocalExec {
514    fn execute(&self, future: F) {
515        executor::spawn_local(future);
516    }
517}