LCOV - code coverage report
Current view: top level - proxy/src/serverless - backend.rs (source / functions) Coverage Total Hit
Test: 4f58e98c51285c7fa348e0b410c88a10caf68ad2.info Lines: 0.0 % 268 0
Test Date: 2025-01-07 20:58:07 Functions: 0.0 % 29 0

            Line data    Source code
       1              : use std::io;
       2              : use std::sync::Arc;
       3              : use std::time::Duration;
       4              : 
       5              : use async_trait::async_trait;
       6              : use hyper_util::rt::{TokioExecutor, TokioIo, TokioTimer};
       7              : use p256::ecdsa::SigningKey;
       8              : use p256::elliptic_curve::JwkEcKey;
       9              : use rand::rngs::OsRng;
      10              : use tokio::net::{lookup_host, TcpStream};
      11              : use tracing::field::display;
      12              : use tracing::{debug, info};
      13              : 
      14              : use super::conn_pool::poll_client;
      15              : use super::conn_pool_lib::{Client, ConnInfo, EndpointConnPool, GlobalConnPool};
      16              : use super::http_conn_pool::{self, poll_http2_client, HttpConnPool, Send};
      17              : use super::local_conn_pool::{self, LocalConnPool, EXT_NAME, EXT_SCHEMA, EXT_VERSION};
      18              : use crate::auth::backend::local::StaticAuthRules;
      19              : use crate::auth::backend::{ComputeCredentials, ComputeUserInfo};
      20              : use crate::auth::{self, check_peer_addr_is_in_list, AuthError};
      21              : use crate::compute;
      22              : use crate::compute_ctl::{
      23              :     ComputeCtlError, ExtensionInstallRequest, Privilege, SetRoleGrantsRequest,
      24              : };
      25              : use crate::config::{ComputeConfig, ProxyConfig};
      26              : use crate::context::RequestContext;
      27              : use crate::control_plane::client::ApiLockError;
      28              : use crate::control_plane::errors::{GetAuthInfoError, WakeComputeError};
      29              : use crate::control_plane::locks::ApiLocks;
      30              : use crate::control_plane::CachedNodeInfo;
      31              : use crate::error::{ErrorKind, ReportableError, UserFacingError};
      32              : use crate::intern::EndpointIdInt;
      33              : use crate::proxy::connect_compute::ConnectMechanism;
      34              : use crate::proxy::retry::{CouldRetry, ShouldRetryWakeCompute};
      35              : use crate::rate_limiter::EndpointRateLimiter;
      36              : use crate::types::{EndpointId, Host, LOCAL_PROXY_SUFFIX};
      37              : 
      38              : pub(crate) struct PoolingBackend {
      39              :     pub(crate) http_conn_pool: Arc<GlobalConnPool<Send, HttpConnPool<Send>>>,
      40              :     pub(crate) local_pool: Arc<LocalConnPool<postgres_client::Client>>,
      41              :     pub(crate) pool:
      42              :         Arc<GlobalConnPool<postgres_client::Client, EndpointConnPool<postgres_client::Client>>>,
      43              : 
      44              :     pub(crate) config: &'static ProxyConfig,
      45              :     pub(crate) auth_backend: &'static crate::auth::Backend<'static, ()>,
      46              :     pub(crate) endpoint_rate_limiter: Arc<EndpointRateLimiter>,
      47              : }
      48              : 
      49              : impl PoolingBackend {
      50            0 :     pub(crate) async fn authenticate_with_password(
      51            0 :         &self,
      52            0 :         ctx: &RequestContext,
      53            0 :         user_info: &ComputeUserInfo,
      54            0 :         password: &[u8],
      55            0 :     ) -> Result<ComputeCredentials, AuthError> {
      56            0 :         ctx.set_auth_method(crate::context::AuthMethod::Cleartext);
      57            0 : 
      58            0 :         let user_info = user_info.clone();
      59            0 :         let backend = self.auth_backend.as_ref().map(|()| user_info.clone());
      60            0 :         let (allowed_ips, maybe_secret) = backend.get_allowed_ips_and_secret(ctx).await?;
      61            0 :         if self.config.authentication_config.ip_allowlist_check_enabled
      62            0 :             && !check_peer_addr_is_in_list(&ctx.peer_addr(), &allowed_ips)
      63              :         {
      64            0 :             return Err(AuthError::ip_address_not_allowed(ctx.peer_addr()));
      65            0 :         }
      66            0 :         if !self
      67            0 :             .endpoint_rate_limiter
      68            0 :             .check(user_info.endpoint.clone().into(), 1)
      69              :         {
      70            0 :             return Err(AuthError::too_many_connections());
      71            0 :         }
      72            0 :         let cached_secret = match maybe_secret {
      73            0 :             Some(secret) => secret,
      74            0 :             None => backend.get_role_secret(ctx).await?,
      75              :         };
      76              : 
      77            0 :         let secret = match cached_secret.value.clone() {
      78            0 :             Some(secret) => self.config.authentication_config.check_rate_limit(
      79            0 :                 ctx,
      80            0 :                 secret,
      81            0 :                 &user_info.endpoint,
      82            0 :                 true,
      83            0 :             )?,
      84              :             None => {
      85              :                 // If we don't have an authentication secret, for the http flow we can just return an error.
      86            0 :                 info!("authentication info not found");
      87            0 :                 return Err(AuthError::password_failed(&*user_info.user));
      88              :             }
      89              :         };
      90            0 :         let ep = EndpointIdInt::from(&user_info.endpoint);
      91            0 :         let auth_outcome = crate::auth::validate_password_and_exchange(
      92            0 :             &self.config.authentication_config.thread_pool,
      93            0 :             ep,
      94            0 :             password,
      95            0 :             secret,
      96            0 :         )
      97            0 :         .await?;
      98            0 :         let res = match auth_outcome {
      99            0 :             crate::sasl::Outcome::Success(key) => {
     100            0 :                 info!("user successfully authenticated");
     101            0 :                 Ok(key)
     102              :             }
     103            0 :             crate::sasl::Outcome::Failure(reason) => {
     104            0 :                 info!("auth backend failed with an error: {reason}");
     105            0 :                 Err(AuthError::password_failed(&*user_info.user))
     106              :             }
     107              :         };
     108            0 :         res.map(|key| ComputeCredentials {
     109            0 :             info: user_info,
     110            0 :             keys: key,
     111            0 :         })
     112            0 :     }
     113              : 
     114            0 :     pub(crate) async fn authenticate_with_jwt(
     115            0 :         &self,
     116            0 :         ctx: &RequestContext,
     117            0 :         user_info: &ComputeUserInfo,
     118            0 :         jwt: String,
     119            0 :     ) -> Result<ComputeCredentials, AuthError> {
     120            0 :         ctx.set_auth_method(crate::context::AuthMethod::Jwt);
     121            0 : 
     122            0 :         match &self.auth_backend {
     123            0 :             crate::auth::Backend::ControlPlane(console, ()) => {
     124            0 :                 self.config
     125            0 :                     .authentication_config
     126            0 :                     .jwks_cache
     127            0 :                     .check_jwt(
     128            0 :                         ctx,
     129            0 :                         user_info.endpoint.clone(),
     130            0 :                         &user_info.user,
     131            0 :                         &**console,
     132            0 :                         &jwt,
     133            0 :                     )
     134            0 :                     .await?;
     135              : 
     136            0 :                 Ok(ComputeCredentials {
     137            0 :                     info: user_info.clone(),
     138            0 :                     keys: crate::auth::backend::ComputeCredentialKeys::None,
     139            0 :                 })
     140              :             }
     141              :             crate::auth::Backend::Local(_) => {
     142            0 :                 let keys = self
     143            0 :                     .config
     144            0 :                     .authentication_config
     145            0 :                     .jwks_cache
     146            0 :                     .check_jwt(
     147            0 :                         ctx,
     148            0 :                         user_info.endpoint.clone(),
     149            0 :                         &user_info.user,
     150            0 :                         &StaticAuthRules,
     151            0 :                         &jwt,
     152            0 :                     )
     153            0 :                     .await?;
     154              : 
     155            0 :                 Ok(ComputeCredentials {
     156            0 :                     info: user_info.clone(),
     157            0 :                     keys,
     158            0 :                 })
     159              :             }
     160              :         }
     161            0 :     }
     162              : 
     163              :     // Wake up the destination if needed. Code here is a bit involved because
     164              :     // we reuse the code from the usual proxy and we need to prepare few structures
     165              :     // that this code expects.
     166            0 :     #[tracing::instrument(fields(pid = tracing::field::Empty), skip_all)]
     167              :     pub(crate) async fn connect_to_compute(
     168              :         &self,
     169              :         ctx: &RequestContext,
     170              :         conn_info: ConnInfo,
     171              :         keys: ComputeCredentials,
     172              :         force_new: bool,
     173              :     ) -> Result<Client<postgres_client::Client>, HttpConnError> {
     174              :         let maybe_client = if force_new {
     175              :             debug!("pool: pool is disabled");
     176              :             None
     177              :         } else {
     178              :             debug!("pool: looking for an existing connection");
     179              :             self.pool.get(ctx, &conn_info)?
     180              :         };
     181              : 
     182              :         if let Some(client) = maybe_client {
     183              :             return Ok(client);
     184              :         }
     185              :         let conn_id = uuid::Uuid::new_v4();
     186              :         tracing::Span::current().record("conn_id", display(conn_id));
     187              :         info!(%conn_id, "pool: opening a new connection '{conn_info}'");
     188            0 :         let backend = self.auth_backend.as_ref().map(|()| keys);
     189              :         crate::proxy::connect_compute::connect_to_compute(
     190              :             ctx,
     191              :             &TokioMechanism {
     192              :                 conn_id,
     193              :                 conn_info,
     194              :                 pool: self.pool.clone(),
     195              :                 locks: &self.config.connect_compute_locks,
     196              :             },
     197              :             &backend,
     198              :             self.config.wake_compute_retry_config,
     199              :             &self.config.connect_to_compute,
     200              :         )
     201              :         .await
     202              :     }
     203              : 
     204              :     // Wake up the destination if needed
     205            0 :     #[tracing::instrument(fields(pid = tracing::field::Empty), skip_all)]
     206              :     pub(crate) async fn connect_to_local_proxy(
     207              :         &self,
     208              :         ctx: &RequestContext,
     209              :         conn_info: ConnInfo,
     210              :     ) -> Result<http_conn_pool::Client<Send>, HttpConnError> {
     211              :         debug!("pool: looking for an existing connection");
     212              :         if let Ok(Some(client)) = self.http_conn_pool.get(ctx, &conn_info) {
     213              :             return Ok(client);
     214              :         }
     215              : 
     216              :         let conn_id = uuid::Uuid::new_v4();
     217              :         tracing::Span::current().record("conn_id", display(conn_id));
     218              :         debug!(%conn_id, "pool: opening a new connection '{conn_info}'");
     219            0 :         let backend = self.auth_backend.as_ref().map(|()| ComputeCredentials {
     220            0 :             info: ComputeUserInfo {
     221            0 :                 user: conn_info.user_info.user.clone(),
     222            0 :                 endpoint: EndpointId::from(format!(
     223            0 :                     "{}{LOCAL_PROXY_SUFFIX}",
     224            0 :                     conn_info.user_info.endpoint.normalize()
     225            0 :                 )),
     226            0 :                 options: conn_info.user_info.options.clone(),
     227            0 :             },
     228            0 :             keys: crate::auth::backend::ComputeCredentialKeys::None,
     229            0 :         });
     230              :         crate::proxy::connect_compute::connect_to_compute(
     231              :             ctx,
     232              :             &HyperMechanism {
     233              :                 conn_id,
     234              :                 conn_info,
     235              :                 pool: self.http_conn_pool.clone(),
     236              :                 locks: &self.config.connect_compute_locks,
     237              :             },
     238              :             &backend,
     239              :             self.config.wake_compute_retry_config,
     240              :             &self.config.connect_to_compute,
     241              :         )
     242              :         .await
     243              :     }
     244              : 
     245              :     /// Connect to postgres over localhost.
     246              :     ///
     247              :     /// We expect postgres to be started here, so we won't do any retries.
     248              :     ///
     249              :     /// # Panics
     250              :     ///
     251              :     /// Panics if called with a non-local_proxy backend.
     252            0 :     #[tracing::instrument(fields(pid = tracing::field::Empty), skip_all)]
     253              :     pub(crate) async fn connect_to_local_postgres(
     254              :         &self,
     255              :         ctx: &RequestContext,
     256              :         conn_info: ConnInfo,
     257              :     ) -> Result<Client<postgres_client::Client>, HttpConnError> {
     258              :         if let Some(client) = self.local_pool.get(ctx, &conn_info)? {
     259              :             return Ok(client);
     260              :         }
     261              : 
     262              :         let local_backend = match &self.auth_backend {
     263              :             auth::Backend::ControlPlane(_, ()) => {
     264              :                 unreachable!("only local_proxy can connect to local postgres")
     265              :             }
     266              :             auth::Backend::Local(local) => local,
     267              :         };
     268              : 
     269              :         if !self.local_pool.initialized(&conn_info) {
     270              :             // only install and grant usage one at a time.
     271              :             let _permit = local_backend
     272              :                 .initialize
     273              :                 .acquire()
     274              :                 .await
     275              :                 .expect("semaphore should never be closed");
     276              : 
     277              :             // check again for race
     278              :             if !self.local_pool.initialized(&conn_info) {
     279              :                 local_backend
     280              :                     .compute_ctl
     281              :                     .install_extension(&ExtensionInstallRequest {
     282              :                         extension: EXT_NAME,
     283              :                         database: conn_info.dbname.clone(),
     284              :                         version: EXT_VERSION,
     285              :                     })
     286              :                     .await?;
     287              : 
     288              :                 local_backend
     289              :                     .compute_ctl
     290              :                     .grant_role(&SetRoleGrantsRequest {
     291              :                         schema: EXT_SCHEMA,
     292              :                         privileges: vec![Privilege::Usage],
     293              :                         database: conn_info.dbname.clone(),
     294              :                         role: conn_info.user_info.user.clone(),
     295              :                     })
     296              :                     .await?;
     297              : 
     298              :                 self.local_pool.set_initialized(&conn_info);
     299              :             }
     300              :         }
     301              : 
     302              :         let conn_id = uuid::Uuid::new_v4();
     303              :         tracing::Span::current().record("conn_id", display(conn_id));
     304              :         info!(%conn_id, "local_pool: opening a new connection '{conn_info}'");
     305              : 
     306              :         let mut node_info = local_backend.node_info.clone();
     307              : 
     308              :         let (key, jwk) = create_random_jwk();
     309              : 
     310              :         let config = node_info
     311              :             .config
     312              :             .user(&conn_info.user_info.user)
     313              :             .dbname(&conn_info.dbname)
     314              :             .set_param(
     315              :                 "options",
     316              :                 &format!(
     317              :                     "-c pg_session_jwt.jwk={}",
     318              :                     serde_json::to_string(&jwk).expect("serializing jwk to json should not fail")
     319              :                 ),
     320              :             );
     321              : 
     322              :         let pause = ctx.latency_timer_pause(crate::metrics::Waiting::Compute);
     323              :         let (client, connection) = config.connect(postgres_client::NoTls).await?;
     324              :         drop(pause);
     325              : 
     326              :         let pid = client.get_process_id();
     327              :         tracing::Span::current().record("pid", pid);
     328              : 
     329              :         let mut handle = local_conn_pool::poll_client(
     330              :             self.local_pool.clone(),
     331              :             ctx,
     332              :             conn_info,
     333              :             client,
     334              :             connection,
     335              :             key,
     336              :             conn_id,
     337              :             node_info.aux.clone(),
     338              :         );
     339              : 
     340              :         {
     341              :             let (client, mut discard) = handle.inner();
     342              :             debug!("setting up backend session state");
     343              : 
     344              :             // initiates the auth session
     345              :             if let Err(e) = client.execute("select auth.init()", &[]).await {
     346              :                 discard.discard();
     347              :                 return Err(e.into());
     348              :             }
     349              : 
     350              :             info!("backend session state initialized");
     351              :         }
     352              : 
     353              :         Ok(handle)
     354              :     }
     355              : }
     356              : 
     357            0 : fn create_random_jwk() -> (SigningKey, JwkEcKey) {
     358            0 :     let key = SigningKey::random(&mut OsRng);
     359            0 :     let jwk = p256::PublicKey::from(key.verifying_key()).to_jwk();
     360            0 :     (key, jwk)
     361            0 : }
     362              : 
     363              : #[derive(Debug, thiserror::Error)]
     364              : pub(crate) enum HttpConnError {
     365              :     #[error("pooled connection closed at inconsistent state")]
     366              :     ConnectionClosedAbruptly(#[from] tokio::sync::watch::error::SendError<uuid::Uuid>),
     367              :     #[error("could not connection to postgres in compute")]
     368              :     PostgresConnectionError(#[from] postgres_client::Error),
     369              :     #[error("could not connection to local-proxy in compute")]
     370              :     LocalProxyConnectionError(#[from] LocalProxyConnError),
     371              :     #[error("could not parse JWT payload")]
     372              :     JwtPayloadError(serde_json::Error),
     373              : 
     374              :     #[error("could not install extension: {0}")]
     375              :     ComputeCtl(#[from] ComputeCtlError),
     376              :     #[error("could not get auth info")]
     377              :     GetAuthInfo(#[from] GetAuthInfoError),
     378              :     #[error("user not authenticated")]
     379              :     AuthError(#[from] AuthError),
     380              :     #[error("wake_compute returned error")]
     381              :     WakeCompute(#[from] WakeComputeError),
     382              :     #[error("error acquiring resource permit: {0}")]
     383              :     TooManyConnectionAttempts(#[from] ApiLockError),
     384              : }
     385              : 
     386              : #[derive(Debug, thiserror::Error)]
     387              : pub(crate) enum LocalProxyConnError {
     388              :     #[error("error with connection to local-proxy")]
     389              :     Io(#[source] std::io::Error),
     390              :     #[error("could not establish h2 connection")]
     391              :     H2(#[from] hyper::Error),
     392              : }
     393              : 
     394              : impl ReportableError for HttpConnError {
     395            0 :     fn get_error_kind(&self) -> ErrorKind {
     396            0 :         match self {
     397            0 :             HttpConnError::ConnectionClosedAbruptly(_) => ErrorKind::Compute,
     398            0 :             HttpConnError::PostgresConnectionError(p) => p.get_error_kind(),
     399            0 :             HttpConnError::LocalProxyConnectionError(_) => ErrorKind::Compute,
     400            0 :             HttpConnError::ComputeCtl(_) => ErrorKind::Service,
     401            0 :             HttpConnError::JwtPayloadError(_) => ErrorKind::User,
     402            0 :             HttpConnError::GetAuthInfo(a) => a.get_error_kind(),
     403            0 :             HttpConnError::AuthError(a) => a.get_error_kind(),
     404            0 :             HttpConnError::WakeCompute(w) => w.get_error_kind(),
     405            0 :             HttpConnError::TooManyConnectionAttempts(w) => w.get_error_kind(),
     406              :         }
     407            0 :     }
     408              : }
     409              : 
     410              : impl UserFacingError for HttpConnError {
     411            0 :     fn to_string_client(&self) -> String {
     412            0 :         match self {
     413            0 :             HttpConnError::ConnectionClosedAbruptly(_) => self.to_string(),
     414            0 :             HttpConnError::PostgresConnectionError(p) => p.to_string(),
     415            0 :             HttpConnError::LocalProxyConnectionError(p) => p.to_string(),
     416            0 :             HttpConnError::ComputeCtl(_) => "could not set up the JWT authorization database extension".to_string(),
     417            0 :             HttpConnError::JwtPayloadError(p) => p.to_string(),
     418            0 :             HttpConnError::GetAuthInfo(c) => c.to_string_client(),
     419            0 :             HttpConnError::AuthError(c) => c.to_string_client(),
     420            0 :             HttpConnError::WakeCompute(c) => c.to_string_client(),
     421              :             HttpConnError::TooManyConnectionAttempts(_) => {
     422            0 :                 "Failed to acquire permit to connect to the database. Too many database connection attempts are currently ongoing.".to_owned()
     423              :             }
     424              :         }
     425            0 :     }
     426              : }
     427              : 
     428              : impl CouldRetry for HttpConnError {
     429            0 :     fn could_retry(&self) -> bool {
     430            0 :         match self {
     431            0 :             HttpConnError::PostgresConnectionError(e) => e.could_retry(),
     432            0 :             HttpConnError::LocalProxyConnectionError(e) => e.could_retry(),
     433            0 :             HttpConnError::ComputeCtl(_) => false,
     434            0 :             HttpConnError::ConnectionClosedAbruptly(_) => false,
     435            0 :             HttpConnError::JwtPayloadError(_) => false,
     436            0 :             HttpConnError::GetAuthInfo(_) => false,
     437            0 :             HttpConnError::AuthError(_) => false,
     438            0 :             HttpConnError::WakeCompute(_) => false,
     439            0 :             HttpConnError::TooManyConnectionAttempts(_) => false,
     440              :         }
     441            0 :     }
     442              : }
     443              : impl ShouldRetryWakeCompute for HttpConnError {
     444            0 :     fn should_retry_wake_compute(&self) -> bool {
     445            0 :         match self {
     446            0 :             HttpConnError::PostgresConnectionError(e) => e.should_retry_wake_compute(),
     447              :             // we never checked cache validity
     448            0 :             HttpConnError::TooManyConnectionAttempts(_) => false,
     449            0 :             _ => true,
     450              :         }
     451            0 :     }
     452              : }
     453              : 
     454              : impl ReportableError for LocalProxyConnError {
     455            0 :     fn get_error_kind(&self) -> ErrorKind {
     456            0 :         match self {
     457            0 :             LocalProxyConnError::Io(_) => ErrorKind::Compute,
     458            0 :             LocalProxyConnError::H2(_) => ErrorKind::Compute,
     459              :         }
     460            0 :     }
     461              : }
     462              : 
     463              : impl UserFacingError for LocalProxyConnError {
     464            0 :     fn to_string_client(&self) -> String {
     465            0 :         "Could not establish HTTP connection to the database".to_string()
     466            0 :     }
     467              : }
     468              : 
     469              : impl CouldRetry for LocalProxyConnError {
     470            0 :     fn could_retry(&self) -> bool {
     471            0 :         match self {
     472            0 :             LocalProxyConnError::Io(_) => false,
     473            0 :             LocalProxyConnError::H2(_) => false,
     474              :         }
     475            0 :     }
     476              : }
     477              : impl ShouldRetryWakeCompute for LocalProxyConnError {
     478            0 :     fn should_retry_wake_compute(&self) -> bool {
     479            0 :         match self {
     480            0 :             LocalProxyConnError::Io(_) => false,
     481            0 :             LocalProxyConnError::H2(_) => false,
     482              :         }
     483            0 :     }
     484              : }
     485              : 
     486              : struct TokioMechanism {
     487              :     pool: Arc<GlobalConnPool<postgres_client::Client, EndpointConnPool<postgres_client::Client>>>,
     488              :     conn_info: ConnInfo,
     489              :     conn_id: uuid::Uuid,
     490              : 
     491              :     /// connect_to_compute concurrency lock
     492              :     locks: &'static ApiLocks<Host>,
     493              : }
     494              : 
     495              : #[async_trait]
     496              : impl ConnectMechanism for TokioMechanism {
     497              :     type Connection = Client<postgres_client::Client>;
     498              :     type ConnectError = HttpConnError;
     499              :     type Error = HttpConnError;
     500              : 
     501            0 :     async fn connect_once(
     502            0 :         &self,
     503            0 :         ctx: &RequestContext,
     504            0 :         node_info: &CachedNodeInfo,
     505            0 :         compute_config: &ComputeConfig,
     506            0 :     ) -> Result<Self::Connection, Self::ConnectError> {
     507            0 :         let host = node_info.config.get_host();
     508            0 :         let permit = self.locks.get_permit(&host).await?;
     509              : 
     510            0 :         let mut config = (*node_info.config).clone();
     511            0 :         let config = config
     512            0 :             .user(&self.conn_info.user_info.user)
     513            0 :             .dbname(&self.conn_info.dbname)
     514            0 :             .connect_timeout(compute_config.timeout);
     515            0 : 
     516            0 :         let pause = ctx.latency_timer_pause(crate::metrics::Waiting::Compute);
     517            0 :         let res = config.connect(postgres_client::NoTls).await;
     518            0 :         drop(pause);
     519            0 :         let (client, connection) = permit.release_result(res)?;
     520              : 
     521            0 :         tracing::Span::current().record("pid", tracing::field::display(client.get_process_id()));
     522            0 :         Ok(poll_client(
     523            0 :             self.pool.clone(),
     524            0 :             ctx,
     525            0 :             self.conn_info.clone(),
     526            0 :             client,
     527            0 :             connection,
     528            0 :             self.conn_id,
     529            0 :             node_info.aux.clone(),
     530            0 :         ))
     531            0 :     }
     532              : 
     533            0 :     fn update_connect_config(&self, _config: &mut compute::ConnCfg) {}
     534              : }
     535              : 
     536              : struct HyperMechanism {
     537              :     pool: Arc<GlobalConnPool<Send, HttpConnPool<Send>>>,
     538              :     conn_info: ConnInfo,
     539              :     conn_id: uuid::Uuid,
     540              : 
     541              :     /// connect_to_compute concurrency lock
     542              :     locks: &'static ApiLocks<Host>,
     543              : }
     544              : 
     545              : #[async_trait]
     546              : impl ConnectMechanism for HyperMechanism {
     547              :     type Connection = http_conn_pool::Client<Send>;
     548              :     type ConnectError = HttpConnError;
     549              :     type Error = HttpConnError;
     550              : 
     551            0 :     async fn connect_once(
     552            0 :         &self,
     553            0 :         ctx: &RequestContext,
     554            0 :         node_info: &CachedNodeInfo,
     555            0 :         config: &ComputeConfig,
     556            0 :     ) -> Result<Self::Connection, Self::ConnectError> {
     557            0 :         let host = node_info.config.get_host();
     558            0 :         let permit = self.locks.get_permit(&host).await?;
     559              : 
     560            0 :         let pause = ctx.latency_timer_pause(crate::metrics::Waiting::Compute);
     561            0 : 
     562            0 :         let port = node_info.config.get_port();
     563            0 :         let res = connect_http2(&host, port, config.timeout).await;
     564            0 :         drop(pause);
     565            0 :         let (client, connection) = permit.release_result(res)?;
     566              : 
     567            0 :         Ok(poll_http2_client(
     568            0 :             self.pool.clone(),
     569            0 :             ctx,
     570            0 :             &self.conn_info,
     571            0 :             client,
     572            0 :             connection,
     573            0 :             self.conn_id,
     574            0 :             node_info.aux.clone(),
     575            0 :         ))
     576            0 :     }
     577              : 
     578            0 :     fn update_connect_config(&self, _config: &mut compute::ConnCfg) {}
     579              : }
     580              : 
     581            0 : async fn connect_http2(
     582            0 :     host: &str,
     583            0 :     port: u16,
     584            0 :     timeout: Duration,
     585            0 : ) -> Result<(http_conn_pool::Send, http_conn_pool::Connect), LocalProxyConnError> {
     586              :     // assumption: host is an ip address so this should not actually perform any requests.
     587              :     // todo: add that assumption as a guarantee in the control-plane API.
     588            0 :     let mut addrs = lookup_host((host, port))
     589            0 :         .await
     590            0 :         .map_err(LocalProxyConnError::Io)?;
     591              : 
     592            0 :     let mut last_err = None;
     593              : 
     594            0 :     let stream = loop {
     595            0 :         let Some(addr) = addrs.next() else {
     596            0 :             return Err(last_err.unwrap_or_else(|| {
     597            0 :                 LocalProxyConnError::Io(io::Error::new(
     598            0 :                     io::ErrorKind::InvalidInput,
     599            0 :                     "could not resolve any addresses",
     600            0 :                 ))
     601            0 :             }));
     602              :         };
     603              : 
     604            0 :         match tokio::time::timeout(timeout, TcpStream::connect(addr)).await {
     605            0 :             Ok(Ok(stream)) => {
     606            0 :                 stream.set_nodelay(true).map_err(LocalProxyConnError::Io)?;
     607            0 :                 break stream;
     608              :             }
     609            0 :             Ok(Err(e)) => {
     610            0 :                 last_err = Some(LocalProxyConnError::Io(e));
     611            0 :             }
     612            0 :             Err(e) => {
     613            0 :                 last_err = Some(LocalProxyConnError::Io(io::Error::new(
     614            0 :                     io::ErrorKind::TimedOut,
     615            0 :                     e,
     616            0 :                 )));
     617            0 :             }
     618              :         };
     619              :     };
     620              : 
     621            0 :     let (client, connection) = hyper::client::conn::http2::Builder::new(TokioExecutor::new())
     622            0 :         .timer(TokioTimer::new())
     623            0 :         .keep_alive_interval(Duration::from_secs(20))
     624            0 :         .keep_alive_while_idle(true)
     625            0 :         .keep_alive_timeout(Duration::from_secs(5))
     626            0 :         .handshake(TokioIo::new(stream))
     627            0 :         .await?;
     628              : 
     629            0 :     Ok((client, connection))
     630            0 : }
        

Generated by: LCOV version 2.1-beta