Merge pull request #8178 from neondatabase/rc/proxy/2024-06-27

Proxy release 2024-06-27

proxy/src/bin/pg_sni_router.rs

@@ -10,7 +10,7 @@ use itertools::Itertools;
 use proxy::config::TlsServerEndPoint;
 use proxy::context::RequestMonitoring;
 use proxy::metrics::{Metrics, ThreadPoolMetrics};
-use proxy::proxy::{copy_bidirectional_client_compute, run_until_cancelled};
+use proxy::proxy::{copy_bidirectional_client_compute, run_until_cancelled, ErrorSource};
 use rustls::pki_types::PrivateKeyDer;
 use tokio::net::TcpListener;
 
@@ -286,7 +286,10 @@ async fn handle_client(
 
     // Starting from here we only proxy the client's traffic.
     info!("performing the proxy pass...");
-    let _ = copy_bidirectional_client_compute(&mut tls_stream, &mut client).await?;
 
-    Ok(())
+    match copy_bidirectional_client_compute(&mut tls_stream, &mut client).await {
+        Ok(_) => Ok(()),
+        Err(ErrorSource::Client(err)) => Err(err).context("client"),
+        Err(ErrorSource::Compute(err)) => Err(err).context("compute"),
+    }
 }

proxy/src/console/messages.rs

@@ -5,7 +5,7 @@ use std::fmt::{self, Display};
 use crate::auth::IpPattern;
 
 use crate::intern::{BranchIdInt, EndpointIdInt, ProjectIdInt};
-use crate::proxy::retry::ShouldRetry;
+use crate::proxy::retry::CouldRetry;
 
 /// Generic error response with human-readable description.
 /// Note that we can't always present it to user as is.
@@ -64,45 +64,47 @@ impl Display for ConsoleError {
     }
 }
 
-impl ShouldRetry for ConsoleError {
+impl CouldRetry for ConsoleError {
     fn could_retry(&self) -> bool {
-        if self.status.is_none() || self.status.as_ref().unwrap().details.retry_info.is_none() {
-            // retry some temporary failures because the compute was in a bad state
-            // (bad request can be returned when the endpoint was in transition)
-            return match &self {
-                ConsoleError {
-                    http_status_code: http::StatusCode::BAD_REQUEST,
-                    ..
-                } => true,
-                // don't retry when quotas are exceeded
-                ConsoleError {
-                    http_status_code: http::StatusCode::UNPROCESSABLE_ENTITY,
-                    ref error,
-                    ..
-                } => !error.contains("compute time quota of non-primary branches is exceeded"),
-                // locked can be returned when the endpoint was in transition
-                // or when quotas are exceeded. don't retry when quotas are exceeded
-                ConsoleError {
-                    http_status_code: http::StatusCode::LOCKED,
-                    ref error,
-                    ..
-                } => {
-                    !error.contains("quota exceeded")
-                        && !error.contains("the limit for current plan reached")
-                }
-                _ => false,
-            };
+        // If the error message does not have a status,
+        // the error is unknown and probably should not retry automatically
+        let Some(status) = &self.status else {
+            return false;
+        };
+
+        // retry if the retry info is set.
+        if status.details.retry_info.is_some() {
+            return true;
         }
 
-        // retry if the response has a retry delay
-        if let Some(retry_info) = self
-            .status
-            .as_ref()
-            .and_then(|s| s.details.retry_info.as_ref())
-        {
-            retry_info.retry_delay_ms > 0
-        } else {
-            false
+        // if no retry info set, attempt to use the error code to guess the retry state.
+        let reason = status
+            .details
+            .error_info
+            .map_or(Reason::Unknown, |e| e.reason);
+        match reason {
+            // not a transitive error
+            Reason::RoleProtected => false,
+            // on retry, it will still not be found
+            Reason::ResourceNotFound
+            | Reason::ProjectNotFound
+            | Reason::EndpointNotFound
+            | Reason::BranchNotFound => false,
+            // we were asked to go away
+            Reason::RateLimitExceeded
+            | Reason::NonDefaultBranchComputeTimeExceeded
+            | Reason::ActiveTimeQuotaExceeded
+            | Reason::ComputeTimeQuotaExceeded
+            | Reason::WrittenDataQuotaExceeded
+            | Reason::DataTransferQuotaExceeded
+            | Reason::LogicalSizeQuotaExceeded => false,
+            // transitive error. control plane is currently busy
+            // but might be ready soon
+            Reason::RunningOperations => true,
+            Reason::ConcurrencyLimitReached => true,
+            Reason::LockAlreadyTaken => true,
+            // unknown error. better not retry it.
+            Reason::Unknown => false,
         }
     }
 }
@@ -121,7 +123,7 @@ pub struct Details {
     pub user_facing_message: Option<UserFacingMessage>,
 }
 
-#[derive(Debug, Deserialize)]
+#[derive(Copy, Clone, Debug, Deserialize)]
 pub struct ErrorInfo {
     pub reason: Reason,
     // Schema could also have `metadata` field, but it's not structured. Skip it for now.
@@ -129,30 +131,59 @@ pub struct ErrorInfo {
 
 #[derive(Clone, Copy, Debug, Deserialize, Default)]
 pub enum Reason {
+    /// RoleProtected indicates that the role is protected and the attempted operation is not permitted on protected roles.
     #[serde(rename = "ROLE_PROTECTED")]
     RoleProtected,
+    /// ResourceNotFound indicates that a resource (project, endpoint, branch, etc.) wasn't found,
+    /// usually due to the provided ID not being correct or because the subject doesn't have enough permissions to
+    /// access the requested resource.
+    /// Prefer a more specific reason if possible, e.g., ProjectNotFound, EndpointNotFound, etc.
     #[serde(rename = "RESOURCE_NOT_FOUND")]
     ResourceNotFound,
+    /// ProjectNotFound indicates that the project wasn't found, usually due to the provided ID not being correct,
+    /// or that the subject doesn't have enough permissions to access the requested project.
     #[serde(rename = "PROJECT_NOT_FOUND")]
     ProjectNotFound,
+    /// EndpointNotFound indicates that the endpoint wasn't found, usually due to the provided ID not being correct,
+    /// or that the subject doesn't have enough permissions to access the requested endpoint.
     #[serde(rename = "ENDPOINT_NOT_FOUND")]
     EndpointNotFound,
+    /// BranchNotFound indicates that the branch wasn't found, usually due to the provided ID not being correct,
+    /// or that the subject doesn't have enough permissions to access the requested branch.
     #[serde(rename = "BRANCH_NOT_FOUND")]
     BranchNotFound,
+    /// RateLimitExceeded indicates that the rate limit for the operation has been exceeded.
     #[serde(rename = "RATE_LIMIT_EXCEEDED")]
     RateLimitExceeded,
+    /// NonDefaultBranchComputeTimeExceeded indicates that the compute time quota of non-default branches has been
+    /// exceeded.
     #[serde(rename = "NON_PRIMARY_BRANCH_COMPUTE_TIME_EXCEEDED")]
-    NonPrimaryBranchComputeTimeExceeded,
+    NonDefaultBranchComputeTimeExceeded,
+    /// ActiveTimeQuotaExceeded indicates that the active time quota was exceeded.
     #[serde(rename = "ACTIVE_TIME_QUOTA_EXCEEDED")]
     ActiveTimeQuotaExceeded,
+    /// ComputeTimeQuotaExceeded indicates that the compute time quota was exceeded.
     #[serde(rename = "COMPUTE_TIME_QUOTA_EXCEEDED")]
     ComputeTimeQuotaExceeded,
+    /// WrittenDataQuotaExceeded indicates that the written data quota was exceeded.
     #[serde(rename = "WRITTEN_DATA_QUOTA_EXCEEDED")]
     WrittenDataQuotaExceeded,
+    /// DataTransferQuotaExceeded indicates that the data transfer quota was exceeded.
     #[serde(rename = "DATA_TRANSFER_QUOTA_EXCEEDED")]
     DataTransferQuotaExceeded,
+    /// LogicalSizeQuotaExceeded indicates that the logical size quota was exceeded.
     #[serde(rename = "LOGICAL_SIZE_QUOTA_EXCEEDED")]
     LogicalSizeQuotaExceeded,
+    /// RunningOperations indicates that the project already has some running operations
+    /// and scheduling of new ones is prohibited.
+    #[serde(rename = "RUNNING_OPERATIONS")]
+    RunningOperations,
+    /// ConcurrencyLimitReached indicates that the concurrency limit for an action was reached.
+    #[serde(rename = "CONCURRENCY_LIMIT_REACHED")]
+    ConcurrencyLimitReached,
+    /// LockAlreadyTaken indicates that the we attempted to take a lock that was already taken.
+    #[serde(rename = "LOCK_ALREADY_TAKEN")]
+    LockAlreadyTaken,
     #[default]
     #[serde(other)]
     Unknown,
@@ -170,7 +201,7 @@ impl Reason {
     }
 }
 
-#[derive(Debug, Deserialize)]
+#[derive(Copy, Clone, Debug, Deserialize)]
 pub struct RetryInfo {
     pub retry_delay_ms: u64,
 }

proxy/src/console/provider.rs

@@ -25,9 +25,9 @@ use tracing::info;
 
 pub mod errors {
     use crate::{
-        console::messages::{self, ConsoleError},
+        console::messages::{self, ConsoleError, Reason},
         error::{io_error, ReportableError, UserFacingError},
-        proxy::retry::ShouldRetry,
+        proxy::retry::CouldRetry,
     };
     use thiserror::Error;
 
@@ -76,21 +76,22 @@ pub mod errors {
                 ApiError::Console(e) => {
                     use crate::error::ErrorKind::*;
                     match e.get_reason() {
-                        crate::console::messages::Reason::RoleProtected => User,
-                        crate::console::messages::Reason::ResourceNotFound => User,
-                        crate::console::messages::Reason::ProjectNotFound => User,
-                        crate::console::messages::Reason::EndpointNotFound => User,
-                        crate::console::messages::Reason::BranchNotFound => User,
-                        crate::console::messages::Reason::RateLimitExceeded => ServiceRateLimit,
-                        crate::console::messages::Reason::NonPrimaryBranchComputeTimeExceeded => {
-                            User
-                        }
-                        crate::console::messages::Reason::ActiveTimeQuotaExceeded => User,
-                        crate::console::messages::Reason::ComputeTimeQuotaExceeded => User,
-                        crate::console::messages::Reason::WrittenDataQuotaExceeded => User,
-                        crate::console::messages::Reason::DataTransferQuotaExceeded => User,
-                        crate::console::messages::Reason::LogicalSizeQuotaExceeded => User,
-                        crate::console::messages::Reason::Unknown => match &e {
+                        Reason::RoleProtected => User,
+                        Reason::ResourceNotFound => User,
+                        Reason::ProjectNotFound => User,
+                        Reason::EndpointNotFound => User,
+                        Reason::BranchNotFound => User,
+                        Reason::RateLimitExceeded => ServiceRateLimit,
+                        Reason::NonDefaultBranchComputeTimeExceeded => User,
+                        Reason::ActiveTimeQuotaExceeded => User,
+                        Reason::ComputeTimeQuotaExceeded => User,
+                        Reason::WrittenDataQuotaExceeded => User,
+                        Reason::DataTransferQuotaExceeded => User,
+                        Reason::LogicalSizeQuotaExceeded => User,
+                        Reason::ConcurrencyLimitReached => ControlPlane,
+                        Reason::LockAlreadyTaken => ControlPlane,
+                        Reason::RunningOperations => ControlPlane,
+                        Reason::Unknown => match &e {
                             ConsoleError {
                                 http_status_code:
                                     http::StatusCode::NOT_FOUND | http::StatusCode::NOT_ACCEPTABLE,
@@ -128,7 +129,7 @@ pub mod errors {
         }
     }
 
-    impl ShouldRetry for ApiError {
+    impl CouldRetry for ApiError {
         fn could_retry(&self) -> bool {
             match self {
                 // retry some transport errors
@@ -239,6 +240,17 @@ pub mod errors {
             }
         }
     }
+
+    impl CouldRetry for WakeComputeError {
+        fn could_retry(&self) -> bool {
+            match self {
+                WakeComputeError::BadComputeAddress(_) => false,
+                WakeComputeError::ApiError(e) => e.could_retry(),
+                WakeComputeError::TooManyConnections => false,
+                WakeComputeError::TooManyConnectionAttempts(_) => false,
+            }
+        }
+    }
 }
 
 /// Auth secret which is managed by the cloud.

proxy/src/context/parquet.rs

@@ -543,7 +543,9 @@ mod tests {
         rx: impl Stream<Item = RequestData>,
     ) -> Vec<(u64, usize, i64)> {
         let remote_storage_config = RemoteStorageConfig {
-            storage: RemoteStorageKind::LocalFs(tmpdir.to_path_buf()),
+            storage: RemoteStorageKind::LocalFs {
+                local_path: tmpdir.to_path_buf(),
+            },
             timeout: std::time::Duration::from_secs(120),
         };
         let storage = GenericRemoteStorage::from_config(&remote_storage_config).unwrap();

proxy/src/proxy.rs

@@ -8,6 +8,7 @@ pub mod passthrough;
 pub mod retry;
 pub mod wake_compute;
 pub use copy_bidirectional::copy_bidirectional_client_compute;
+pub use copy_bidirectional::ErrorSource;
 
 use crate::{
     auth,
@@ -148,8 +149,11 @@ pub async fn task_main(
                     ctx.log_connect();
                     match p.proxy_pass().instrument(span.clone()).await {
                         Ok(()) => {}
-                        Err(e) => {
-                            error!(parent: &span, "per-client task finished with an error: {e:#}");
+                        Err(ErrorSource::Client(e)) => {
+                            error!(parent: &span, "per-client task finished with an IO error from the client: {e:#}");
+                        }
+                        Err(ErrorSource::Compute(e)) => {
+                            error!(parent: &span, "per-client task finished with an IO error from the compute: {e:#}");
                         }
                     }
                 }

proxy/src/proxy/connect_compute.rs

@@ -7,7 +7,7 @@ use crate::{
     error::ReportableError,
     metrics::{ConnectOutcome, ConnectionFailureKind, Metrics, RetriesMetricGroup, RetryType},
     proxy::{
-        retry::{retry_after, ShouldRetry},
+        retry::{retry_after, should_retry, CouldRetry},
         wake_compute::wake_compute,
     },
     Host,
@@ -17,6 +17,8 @@ use pq_proto::StartupMessageParams;
 use tokio::time;
 use tracing::{error, info, warn};
 
+use super::retry::ShouldRetryWakeCompute;
+
 const CONNECT_TIMEOUT: time::Duration = time::Duration::from_secs(2);
 
 /// If we couldn't connect, a cached connection info might be to blame
@@ -104,7 +106,7 @@ pub async fn connect_to_compute<M: ConnectMechanism, B: ComputeConnectBackend>(
     connect_to_compute_retry_config: RetryConfig,
 ) -> Result<M::Connection, M::Error>
 where
-    M::ConnectError: ShouldRetry + std::fmt::Debug,
+    M::ConnectError: CouldRetry + ShouldRetryWakeCompute + std::fmt::Debug,
     M::Error: From<WakeComputeError>,
 {
     let mut num_retries = 0;
@@ -139,10 +141,10 @@ where
 
     error!(error = ?err, "could not connect to compute node");
 
-    let node_info = if !node_info.cached() || !err.should_retry_database_address() {
+    let node_info = if !node_info.cached() || !err.should_retry_wake_compute() {
         // If we just recieved this from cplane and dodn't get it from cache, we shouldn't retry.
         // Do not need to retrieve a new node_info, just return the old one.
-        if !err.should_retry(num_retries, connect_to_compute_retry_config) {
+        if should_retry(&err, num_retries, connect_to_compute_retry_config) {
             Metrics::get().proxy.retries_metric.observe(
                 RetriesMetricGroup {
                     outcome: ConnectOutcome::Failed,
@@ -188,9 +190,8 @@ where
                 return Ok(res);
             }
             Err(e) => {
-                let retriable = e.should_retry(num_retries, connect_to_compute_retry_config);
-                if !retriable {
-                    error!(error = ?e, num_retries, retriable, "couldn't connect to compute node");
+                if !should_retry(&e, num_retries, connect_to_compute_retry_config) {
+                    error!(error = ?e, num_retries, retriable = false, "couldn't connect to compute node");
                     Metrics::get().proxy.retries_metric.observe(
                         RetriesMetricGroup {
                             outcome: ConnectOutcome::Failed,
@@ -200,9 +201,10 @@ where
                     );
                     return Err(e.into());
                 }
-                warn!(error = ?e, num_retries, retriable, "couldn't connect to compute node");
+
+                warn!(error = ?e, num_retries, retriable = true, "couldn't connect to compute node");
             }
-        }
+        };
 
         let wait_duration = retry_after(num_retries, connect_to_compute_retry_config);
         num_retries += 1;

proxy/src/proxy/copy_bidirectional.rs

@@ -13,12 +13,39 @@ enum TransferState {
     Done(u64),
 }
 
+#[derive(Debug)]
+pub enum ErrorDirection {
+    Read(io::Error),
+    Write(io::Error),
+}
+
+impl ErrorSource {
+    fn from_client(err: ErrorDirection) -> ErrorSource {
+        match err {
+            ErrorDirection::Read(client) => Self::Client(client),
+            ErrorDirection::Write(compute) => Self::Compute(compute),
+        }
+    }
+    fn from_compute(err: ErrorDirection) -> ErrorSource {
+        match err {
+            ErrorDirection::Write(client) => Self::Client(client),
+            ErrorDirection::Read(compute) => Self::Compute(compute),
+        }
+    }
+}
+
+#[derive(Debug)]
+pub enum ErrorSource {
+    Client(io::Error),
+    Compute(io::Error),
+}
+
 fn transfer_one_direction<A, B>(
     cx: &mut Context<'_>,
     state: &mut TransferState,
     r: &mut A,
     w: &mut B,
-) -> Poll<io::Result<u64>>
+) -> Poll<Result<u64, ErrorDirection>>
 where
     A: AsyncRead + AsyncWrite + Unpin + ?Sized,
     B: AsyncRead + AsyncWrite + Unpin + ?Sized,
@@ -32,7 +59,7 @@ where
                 *state = TransferState::ShuttingDown(count);
             }
             TransferState::ShuttingDown(count) => {
-                ready!(w.as_mut().poll_shutdown(cx))?;
+                ready!(w.as_mut().poll_shutdown(cx)).map_err(ErrorDirection::Write)?;
                 *state = TransferState::Done(*count);
             }
             TransferState::Done(count) => return Poll::Ready(Ok(*count)),
@@ -44,7 +71,7 @@ where
 pub async fn copy_bidirectional_client_compute<Client, Compute>(
     client: &mut Client,
     compute: &mut Compute,
-) -> Result<(u64, u64), std::io::Error>
+) -> Result<(u64, u64), ErrorSource>
 where
     Client: AsyncRead + AsyncWrite + Unpin + ?Sized,
     Compute: AsyncRead + AsyncWrite + Unpin + ?Sized,
@@ -54,9 +81,11 @@ where
 
     poll_fn(|cx| {
         let mut client_to_compute_result =
-            transfer_one_direction(cx, &mut client_to_compute, client, compute)?;
+            transfer_one_direction(cx, &mut client_to_compute, client, compute)
+                .map_err(ErrorSource::from_client)?;
         let mut compute_to_client_result =
-            transfer_one_direction(cx, &mut compute_to_client, compute, client)?;
+            transfer_one_direction(cx, &mut compute_to_client, compute, client)
+                .map_err(ErrorSource::from_compute)?;
 
         // Early termination checks from compute to client.
         if let TransferState::Done(_) = compute_to_client {
@@ -65,18 +94,20 @@ where
                 // Initiate shutdown
                 client_to_compute = TransferState::ShuttingDown(buf.amt);
                 client_to_compute_result =
-                    transfer_one_direction(cx, &mut client_to_compute, client, compute)?;
+                    transfer_one_direction(cx, &mut client_to_compute, client, compute)
+                        .map_err(ErrorSource::from_client)?;
             }
         }
 
-        // Early termination checks from compute to client.
+        // Early termination checks from client to compute.
         if let TransferState::Done(_) = client_to_compute {
             if let TransferState::Running(buf) = &compute_to_client {
                 info!("Client is done, terminate compute");
                 // Initiate shutdown
                 compute_to_client = TransferState::ShuttingDown(buf.amt);
                 compute_to_client_result =
-                    transfer_one_direction(cx, &mut compute_to_client, client, compute)?;
+                    transfer_one_direction(cx, &mut compute_to_client, compute, client)
+                        .map_err(ErrorSource::from_compute)?;
             }
         }
 
@@ -138,7 +169,7 @@ impl CopyBuffer {
         cx: &mut Context<'_>,
         mut reader: Pin<&mut R>,
         mut writer: Pin<&mut W>,
-    ) -> Poll<io::Result<usize>>
+    ) -> Poll<Result<usize, ErrorDirection>>
     where
         R: AsyncRead + ?Sized,
         W: AsyncWrite + ?Sized,
@@ -149,11 +180,11 @@ impl CopyBuffer {
                 // Top up the buffer towards full if we can read a bit more
                 // data - this should improve the chances of a large write
                 if !me.read_done && me.cap < me.buf.len() {
-                    ready!(me.poll_fill_buf(cx, reader.as_mut()))?;
+                    ready!(me.poll_fill_buf(cx, reader.as_mut())).map_err(ErrorDirection::Read)?;
                 }
                 Poll::Pending
             }
-            res => res,
+            res => res.map_err(ErrorDirection::Write),
         }
     }
 
@@ -162,7 +193,7 @@ impl CopyBuffer {
         cx: &mut Context<'_>,
         mut reader: Pin<&mut R>,
         mut writer: Pin<&mut W>,
-    ) -> Poll<io::Result<u64>>
+    ) -> Poll<Result<u64, ErrorDirection>>
     where
         R: AsyncRead + ?Sized,
         W: AsyncWrite + ?Sized,
@@ -176,12 +207,13 @@ impl CopyBuffer {
 
                 match self.poll_fill_buf(cx, reader.as_mut()) {
                     Poll::Ready(Ok(())) => (),
-                    Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
+                    Poll::Ready(Err(err)) => return Poll::Ready(Err(ErrorDirection::Read(err))),
                     Poll::Pending => {
                         // Try flushing when the reader has no progress to avoid deadlock
                         // when the reader depends on buffered writer.
                         if self.need_flush {
-                            ready!(writer.as_mut().poll_flush(cx))?;
+                            ready!(writer.as_mut().poll_flush(cx))
+                                .map_err(ErrorDirection::Write)?;
                             self.need_flush = false;
                         }
 
@@ -194,10 +226,10 @@ impl CopyBuffer {
             while self.pos < self.cap {
                 let i = ready!(self.poll_write_buf(cx, reader.as_mut(), writer.as_mut()))?;
                 if i == 0 {
-                    return Poll::Ready(Err(io::Error::new(
+                    return Poll::Ready(Err(ErrorDirection::Write(io::Error::new(
                         io::ErrorKind::WriteZero,
                         "write zero byte into writer",
-                    )));
+                    ))));
                 } else {
                     self.pos += i;
                     self.amt += i as u64;
@@ -216,7 +248,7 @@ impl CopyBuffer {
             // If we've written all the data and we've seen EOF, flush out the
             // data and finish the transfer.
             if self.pos == self.cap && self.read_done {
-                ready!(writer.as_mut().poll_flush(cx))?;
+                ready!(writer.as_mut().poll_flush(cx)).map_err(ErrorDirection::Write)?;
                 return Poll::Ready(Ok(self.amt));
             }
         }

proxy/src/proxy/passthrough.rs

@@ -10,13 +10,15 @@ use tokio::io::{AsyncRead, AsyncWrite};
 use tracing::info;
 use utils::measured_stream::MeasuredStream;
 
+use super::copy_bidirectional::ErrorSource;
+
 /// Forward bytes in both directions (client <-> compute).
 #[tracing::instrument(skip_all)]
 pub async fn proxy_pass(
     client: impl AsyncRead + AsyncWrite + Unpin,
     compute: impl AsyncRead + AsyncWrite + Unpin,
     aux: MetricsAuxInfo,
-) -> anyhow::Result<()> {
+) -> Result<(), ErrorSource> {
     let usage = USAGE_METRICS.register(Ids {
         endpoint_id: aux.endpoint_id,
         branch_id: aux.branch_id,
@@ -66,9 +68,11 @@ pub struct ProxyPassthrough<P, S> {
 }
 
 impl<P, S: AsyncRead + AsyncWrite + Unpin> ProxyPassthrough<P, S> {
-    pub async fn proxy_pass(self) -> anyhow::Result<()> {
+    pub async fn proxy_pass(self) -> Result<(), ErrorSource> {
         let res = proxy_pass(self.client, self.compute.stream, self.aux).await;
-        self.compute.cancel_closure.try_cancel_query().await?;
+        if let Err(err) = self.compute.cancel_closure.try_cancel_query().await {
+            tracing::error!(?err, "could not cancel the query in the database");
+        }
         res
     }
 }

proxy/src/proxy/retry.rs

@@ -2,20 +2,22 @@ use crate::{compute, config::RetryConfig};
 use std::{error::Error, io};
 use tokio::time;
 
-pub trait ShouldRetry {
+pub trait CouldRetry {
+    /// Returns true if the error could be retried
     fn could_retry(&self) -> bool;
-    fn should_retry(&self, num_retries: u32, config: RetryConfig) -> bool {
-        match self {
-            _ if num_retries >= config.max_retries => false,
-            err => err.could_retry(),
-        }
-    }
-    fn should_retry_database_address(&self) -> bool {
-        true
-    }
 }
 
-impl ShouldRetry for io::Error {
+pub trait ShouldRetryWakeCompute {
+    /// Returns true if we need to invalidate the cache for this node.
+    /// If false, we can continue retrying with the current node cache.
+    fn should_retry_wake_compute(&self) -> bool;
+}
+
+pub fn should_retry(err: &impl CouldRetry, num_retries: u32, config: RetryConfig) -> bool {
+    num_retries < config.max_retries && err.could_retry()
+}
+
+impl CouldRetry for io::Error {
     fn could_retry(&self) -> bool {
         use std::io::ErrorKind;
         matches!(
@@ -25,7 +27,7 @@ impl ShouldRetry for io::Error {
     }
 }
 
-impl ShouldRetry for tokio_postgres::error::DbError {
+impl CouldRetry for tokio_postgres::error::DbError {
     fn could_retry(&self) -> bool {
         use tokio_postgres::error::SqlState;
         matches!(
@@ -36,7 +38,9 @@ impl ShouldRetry for tokio_postgres::error::DbError {
                 | &SqlState::SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION,
         )
     }
-    fn should_retry_database_address(&self) -> bool {
+}
+impl ShouldRetryWakeCompute for tokio_postgres::error::DbError {
+    fn should_retry_wake_compute(&self) -> bool {
         use tokio_postgres::error::SqlState;
         // Here are errors that happens after the user successfully authenticated to the database.
         // TODO: there are pgbouncer errors that should be retried, but they are not listed here.
@@ -53,7 +57,7 @@ impl ShouldRetry for tokio_postgres::error::DbError {
     }
 }
 
-impl ShouldRetry for tokio_postgres::Error {
+impl CouldRetry for tokio_postgres::Error {
     fn could_retry(&self) -> bool {
         if let Some(io_err) = self.source().and_then(|x| x.downcast_ref()) {
             io::Error::could_retry(io_err)
@@ -63,29 +67,33 @@ impl ShouldRetry for tokio_postgres::Error {
             false
         }
     }
-    fn should_retry_database_address(&self) -> bool {
-        if let Some(io_err) = self.source().and_then(|x| x.downcast_ref()) {
-            io::Error::should_retry_database_address(io_err)
-        } else if let Some(db_err) = self.source().and_then(|x| x.downcast_ref()) {
-            tokio_postgres::error::DbError::should_retry_database_address(db_err)
+}
+impl ShouldRetryWakeCompute for tokio_postgres::Error {
+    fn should_retry_wake_compute(&self) -> bool {
+        if let Some(db_err) = self.source().and_then(|x| x.downcast_ref()) {
+            tokio_postgres::error::DbError::should_retry_wake_compute(db_err)
         } else {
+            // likely an IO error. Possible the compute has shutdown and the
+            // cache is stale.
             true
         }
     }
 }
 
-impl ShouldRetry for compute::ConnectionError {
+impl CouldRetry for compute::ConnectionError {
     fn could_retry(&self) -> bool {
         match self {
             compute::ConnectionError::Postgres(err) => err.could_retry(),
             compute::ConnectionError::CouldNotConnect(err) => err.could_retry(),
+            compute::ConnectionError::WakeComputeError(err) => err.could_retry(),
             _ => false,
         }
     }
-    fn should_retry_database_address(&self) -> bool {
+}
+impl ShouldRetryWakeCompute for compute::ConnectionError {
+    fn should_retry_wake_compute(&self) -> bool {
         match self {
-            compute::ConnectionError::Postgres(err) => err.should_retry_database_address(),
-            compute::ConnectionError::CouldNotConnect(err) => err.should_retry_database_address(),
+            compute::ConnectionError::Postgres(err) => err.should_retry_wake_compute(),
             // the cache entry was not checked for validity
             compute::ConnectionError::TooManyConnectionAttempts(_) => false,
             _ => true,

proxy/src/proxy/tests.rs

@@ -5,21 +5,21 @@ mod mitm;
 use std::time::Duration;
 
 use super::connect_compute::ConnectMechanism;
-use super::retry::ShouldRetry;
+use super::retry::CouldRetry;
 use super::*;
 use crate::auth::backend::{
     ComputeCredentialKeys, ComputeCredentials, ComputeUserInfo, MaybeOwned, TestBackend,
 };
 use crate::config::{CertResolver, RetryConfig};
 use crate::console::caches::NodeInfoCache;
-use crate::console::messages::{ConsoleError, MetricsAuxInfo};
+use crate::console::messages::{ConsoleError, Details, MetricsAuxInfo, Status};
 use crate::console::provider::{CachedAllowedIps, CachedRoleSecret, ConsoleBackend};
 use crate::console::{self, CachedNodeInfo, NodeInfo};
 use crate::error::ErrorKind;
-use crate::proxy::retry::retry_after;
 use crate::{http, sasl, scram, BranchId, EndpointId, ProjectId};
 use anyhow::{bail, Context};
 use async_trait::async_trait;
+use retry::{retry_after, ShouldRetryWakeCompute};
 use rstest::rstest;
 use rustls::pki_types;
 use tokio_postgres::config::SslMode;
@@ -438,11 +438,16 @@ impl std::fmt::Display for TestConnectError {
 
 impl std::error::Error for TestConnectError {}
 
-impl ShouldRetry for TestConnectError {
+impl CouldRetry for TestConnectError {
     fn could_retry(&self) -> bool {
         self.retryable
     }
 }
+impl ShouldRetryWakeCompute for TestConnectError {
+    fn should_retry_wake_compute(&self) -> bool {
+        true
+    }
+}
 
 #[async_trait]
 impl ConnectMechanism for TestConnectMechanism {
@@ -485,7 +490,7 @@ impl TestBackend for TestConnectMechanism {
             ConnectAction::Wake => Ok(helper_create_cached_node_info(self.cache)),
             ConnectAction::WakeFail => {
                 let err = console::errors::ApiError::Console(ConsoleError {
-                    http_status_code: http::StatusCode::FORBIDDEN,
+                    http_status_code: http::StatusCode::BAD_REQUEST,
                     error: "TEST".into(),
                     status: None,
                 });
@@ -496,7 +501,15 @@ impl TestBackend for TestConnectMechanism {
                 let err = console::errors::ApiError::Console(ConsoleError {
                     http_status_code: http::StatusCode::BAD_REQUEST,
                     error: "TEST".into(),
-                    status: None,
+                    status: Some(Status {
+                        code: "error".into(),
+                        message: "error".into(),
+                        details: Details {
+                            error_info: None,
+                            retry_info: Some(console::messages::RetryInfo { retry_delay_ms: 1 }),
+                            user_facing_message: None,
+                        },
+                    }),
                 });
                 assert!(err.could_retry());
                 Err(console::errors::WakeComputeError::ApiError(err))

proxy/src/proxy/wake_compute.rs

@@ -1,18 +1,16 @@
 use crate::config::RetryConfig;
-use crate::console::messages::ConsoleError;
+use crate::console::messages::{ConsoleError, Reason};
 use crate::console::{errors::WakeComputeError, provider::CachedNodeInfo};
 use crate::context::RequestMonitoring;
 use crate::metrics::{
     ConnectOutcome, ConnectionFailuresBreakdownGroup, Metrics, RetriesMetricGroup, RetryType,
     WakeupFailureKind,
 };
-use crate::proxy::retry::retry_after;
+use crate::proxy::retry::{retry_after, should_retry};
 use hyper1::StatusCode;
-use std::ops::ControlFlow;
 use tracing::{error, info, warn};
 
 use super::connect_compute::ComputeConnectBackend;
-use super::retry::ShouldRetry;
 
 pub async fn wake_compute<B: ComputeConnectBackend>(
     num_retries: &mut u32,
@@ -22,9 +20,8 @@ pub async fn wake_compute<B: ComputeConnectBackend>(
 ) -> Result<CachedNodeInfo, WakeComputeError> {
     let retry_type = RetryType::WakeCompute;
     loop {
-        let wake_res = api.wake_compute(ctx).await;
-        match handle_try_wake(wake_res, *num_retries, config) {
-            Err(e) => {
+        match api.wake_compute(ctx).await {
+            Err(e) if !should_retry(&e, *num_retries, config) => {
                 error!(error = ?e, num_retries, retriable = false, "couldn't wake compute node");
                 report_error(&e, false);
                 Metrics::get().proxy.retries_metric.observe(
@@ -36,11 +33,11 @@ pub async fn wake_compute<B: ComputeConnectBackend>(
                 );
                 return Err(e);
             }
-            Ok(ControlFlow::Continue(e)) => {
+            Err(e) => {
                 warn!(error = ?e, num_retries, retriable = true, "couldn't wake compute node");
                 report_error(&e, true);
             }
-            Ok(ControlFlow::Break(n)) => {
+            Ok(n) => {
                 Metrics::get().proxy.retries_metric.observe(
                     RetriesMetricGroup {
                         outcome: ConnectOutcome::Success,
@@ -63,70 +60,28 @@ pub async fn wake_compute<B: ComputeConnectBackend>(
     }
 }
 
-/// Attempts to wake up the compute node.
-/// * Returns Ok(Continue(e)) if there was an error waking but retries are acceptable
-/// * Returns Ok(Break(node)) if the wakeup succeeded
-/// * Returns Err(e) if there was an error
-pub fn handle_try_wake(
-    result: Result<CachedNodeInfo, WakeComputeError>,
-    num_retries: u32,
-    config: RetryConfig,
-) -> Result<ControlFlow<CachedNodeInfo, WakeComputeError>, WakeComputeError> {
-    match result {
-        Err(err) => match &err {
-            WakeComputeError::ApiError(api) if api.should_retry(num_retries, config) => {
-                Ok(ControlFlow::Continue(err))
-            }
-            _ => Err(err),
-        },
-        // Ready to try again.
-        Ok(new) => Ok(ControlFlow::Break(new)),
-    }
-}
-
 fn report_error(e: &WakeComputeError, retry: bool) {
     use crate::console::errors::ApiError;
     let kind = match e {
         WakeComputeError::BadComputeAddress(_) => WakeupFailureKind::BadComputeAddress,
         WakeComputeError::ApiError(ApiError::Transport(_)) => WakeupFailureKind::ApiTransportError,
         WakeComputeError::ApiError(ApiError::Console(e)) => match e.get_reason() {
-            crate::console::messages::Reason::RoleProtected => {
-                WakeupFailureKind::ApiConsoleBadRequest
-            }
-            crate::console::messages::Reason::ResourceNotFound => {
-                WakeupFailureKind::ApiConsoleBadRequest
-            }
-            crate::console::messages::Reason::ProjectNotFound => {
-                WakeupFailureKind::ApiConsoleBadRequest
-            }
-            crate::console::messages::Reason::EndpointNotFound => {
-                WakeupFailureKind::ApiConsoleBadRequest
-            }
-            crate::console::messages::Reason::BranchNotFound => {
-                WakeupFailureKind::ApiConsoleBadRequest
-            }
-            crate::console::messages::Reason::RateLimitExceeded => {
-                WakeupFailureKind::ApiConsoleLocked
-            }
-            crate::console::messages::Reason::NonPrimaryBranchComputeTimeExceeded => {
-                WakeupFailureKind::QuotaExceeded
-            }
-            crate::console::messages::Reason::ActiveTimeQuotaExceeded => {
-                WakeupFailureKind::QuotaExceeded
-            }
-            crate::console::messages::Reason::ComputeTimeQuotaExceeded => {
-                WakeupFailureKind::QuotaExceeded
-            }
-            crate::console::messages::Reason::WrittenDataQuotaExceeded => {
-                WakeupFailureKind::QuotaExceeded
-            }
-            crate::console::messages::Reason::DataTransferQuotaExceeded => {
-                WakeupFailureKind::QuotaExceeded
-            }
-            crate::console::messages::Reason::LogicalSizeQuotaExceeded => {
-                WakeupFailureKind::QuotaExceeded
-            }
-            crate::console::messages::Reason::Unknown => match e {
+            Reason::RoleProtected => WakeupFailureKind::ApiConsoleBadRequest,
+            Reason::ResourceNotFound => WakeupFailureKind::ApiConsoleBadRequest,
+            Reason::ProjectNotFound => WakeupFailureKind::ApiConsoleBadRequest,
+            Reason::EndpointNotFound => WakeupFailureKind::ApiConsoleBadRequest,
+            Reason::BranchNotFound => WakeupFailureKind::ApiConsoleBadRequest,
+            Reason::RateLimitExceeded => WakeupFailureKind::ApiConsoleLocked,
+            Reason::NonDefaultBranchComputeTimeExceeded => WakeupFailureKind::QuotaExceeded,
+            Reason::ActiveTimeQuotaExceeded => WakeupFailureKind::QuotaExceeded,
+            Reason::ComputeTimeQuotaExceeded => WakeupFailureKind::QuotaExceeded,
+            Reason::WrittenDataQuotaExceeded => WakeupFailureKind::QuotaExceeded,
+            Reason::DataTransferQuotaExceeded => WakeupFailureKind::QuotaExceeded,
+            Reason::LogicalSizeQuotaExceeded => WakeupFailureKind::QuotaExceeded,
+            Reason::ConcurrencyLimitReached => WakeupFailureKind::ApiConsoleLocked,
+            Reason::LockAlreadyTaken => WakeupFailureKind::ApiConsoleLocked,
+            Reason::RunningOperations => WakeupFailureKind::ApiConsoleLocked,
+            Reason::Unknown => match e {
                 ConsoleError {
                     http_status_code: StatusCode::LOCKED,
                     ref error,

proxy/src/serverless/backend.rs

@@ -16,7 +16,10 @@ use crate::{
     context::RequestMonitoring,
     error::{ErrorKind, ReportableError, UserFacingError},
     intern::EndpointIdInt,
-    proxy::{connect_compute::ConnectMechanism, retry::ShouldRetry},
+    proxy::{
+        connect_compute::ConnectMechanism,
+        retry::{CouldRetry, ShouldRetryWakeCompute},
+    },
     rate_limiter::EndpointRateLimiter,
     Host,
 };
@@ -179,7 +182,7 @@ impl UserFacingError for HttpConnError {
     }
 }
 
-impl ShouldRetry for HttpConnError {
+impl CouldRetry for HttpConnError {
     fn could_retry(&self) -> bool {
         match self {
             HttpConnError::ConnectionError(e) => e.could_retry(),
@@ -190,9 +193,11 @@ impl ShouldRetry for HttpConnError {
             HttpConnError::TooManyConnectionAttempts(_) => false,
         }
     }
-    fn should_retry_database_address(&self) -> bool {
+}
+impl ShouldRetryWakeCompute for HttpConnError {
+    fn should_retry_wake_compute(&self) -> bool {
         match self {
-            HttpConnError::ConnectionError(e) => e.should_retry_database_address(),
+            HttpConnError::ConnectionError(e) => e.should_retry_wake_compute(),
             // we never checked cache validity
             HttpConnError::TooManyConnectionAttempts(_) => false,
             _ => true,

proxy/src/serverless/websocket.rs

@@ -1,3 +1,4 @@
+use crate::proxy::ErrorSource;
 use crate::{
     cancellation::CancellationHandlerMain,
     config::ProxyConfig,
@@ -7,6 +8,7 @@ use crate::{
     proxy::{handle_client, ClientMode},
     rate_limiter::EndpointRateLimiter,
 };
+use anyhow::Context as _;
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 use framed_websockets::{Frame, OpCode, WebSocketServer};
 use futures::{Sink, Stream};
@@ -165,7 +167,11 @@ pub async fn serve_websocket(
         Ok(Some(p)) => {
             ctx.set_success();
             ctx.log_connect();
-            p.proxy_pass().await
+            match p.proxy_pass().await {
+                Ok(()) => Ok(()),
+                Err(ErrorSource::Client(err)) => Err(err).context("client"),
+                Err(ErrorSource::Compute(err)) => Err(err).context("compute"),
+            }
         }
     }
 }