walredo: log a few more breadcrumbs on redo failure

1. ability to tell whether a replay failure happened in the first or
   later iterations of the request_redo for loop.
2. n_requests, i.e, the  total number of request_redo calls completed
   since the current walredo process start

Possibly these will give us a bit more context about when this is
happening.

This commit is quite ugly and should be reverted or logging
infrastructure in walredo.rs should be generally switched to spans.

Cargo.lock

@@ -570,9 +570,9 @@ dependencies = [
 
 [[package]]
 name = "aws-smithy-types"
-version = "0.56.1"
+version = "0.56.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d90dbc8da2f6be461fa3c1906b20af8f79d14968fe47f2b7d29d086f62a51728"
+checksum = "eed0a94eefd845a2a78677f1b72f02fa75802d38f7f59be675add140279aa8bf"
 dependencies = [
  "base64-simd",
  "itoa",
@@ -2420,16 +2420,6 @@ dependencies = [
  "winapi",
 ]
 
-[[package]]
-name = "nu-ansi-term"
-version = "0.46.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "77a8165726e8236064dbb45459242600304b42a5ea24ee2948e18e023bf7ba84"
-dependencies = [
- "overload",
- "winapi",
-]
-
 [[package]]
 name = "num-bigint"
 version = "0.4.3"
@@ -2662,12 +2652,6 @@ version = "0.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4030760ffd992bef45b0ae3f10ce1aba99e33464c90d14dd7c039884963ddc7a"
 
-[[package]]
-name = "overload"
-version = "0.1.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b15813163c1d831bf4a13c3610c05c0d03b39feb07f7e09fa234dac9b15aaf39"
-
 [[package]]
 name = "pagectl"
 version = "0.1.0"
@@ -3764,11 +3748,9 @@ dependencies = [
  "aws-config",
  "aws-sdk-s3",
  "aws-smithy-http",
- "aws-smithy-types",
  "aws-types",
  "bincode",
  "bytes",
- "camino",
  "chrono",
  "clap",
  "crc32c",
@@ -3776,11 +3758,9 @@ dependencies = [
  "futures-util",
  "hex",
  "histogram",
- "humantime",
  "itertools",
  "pageserver",
  "rand",
- "remote_storage",
  "reqwest",
  "serde",
  "serde_json",
@@ -4962,7 +4942,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "30a651bc37f915e81f087d86e62a18eec5f79550c7faff886f7090b4ea757c77"
 dependencies = [
  "matchers",
- "nu-ansi-term",
  "once_cell",
  "regex",
  "serde",

Dockerfile.compute-node

@@ -224,8 +224,8 @@ RUN wget https://github.com/df7cb/postgresql-unit/archive/refs/tags/7.7.tar.gz -
 FROM build-deps AS vector-pg-build
 COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/
 
-RUN wget https://github.com/pgvector/pgvector/archive/refs/tags/v0.5.0.tar.gz -O pgvector.tar.gz && \
-    echo "d8aa3504b215467ca528525a6de12c3f85f9891b091ce0e5864dd8a9b757f77b pgvector.tar.gz" | sha256sum --check && \
+RUN wget https://github.com/pgvector/pgvector/archive/refs/tags/v0.5.1.tar.gz -O pgvector.tar.gz && \
+    echo "cc7a8e034a96e30a819911ac79d32f6bc47bdd1aa2de4d7d4904e26b83209dc8 pgvector.tar.gz" | sha256sum --check && \
     mkdir pgvector-src && cd pgvector-src && tar xvzf ../pgvector.tar.gz --strip-components=1 -C . && \
     make -j $(getconf _NPROCESSORS_ONLN) PG_CONFIG=/usr/local/pgsql/bin/pg_config && \
     make -j $(getconf _NPROCESSORS_ONLN) install PG_CONFIG=/usr/local/pgsql/bin/pg_config && \

libs/remote_storage/src/s3_bucket.rs

@@ -4,7 +4,7 @@
 //! allowing multiple api users to independently work with the same S3 bucket, if
 //! their bucket prefixes are both specified and different.
 
-use std::sync::Arc;
+use std::{borrow::Cow, sync::Arc};
 
 use anyhow::Context;
 use aws_config::{
@@ -556,6 +556,20 @@ impl RemoteStorage for S3Bucket {
                         .deleted_objects_total
                         .inc_by(chunk.len() as u64);
                     if let Some(errors) = resp.errors {
+                        // Log a bounded number of the errors within the response:
+                        // these requests can carry 1000 keys so logging each one
+                        // would be too verbose, especially as errors may lead us
+                        // to retry repeatedly.
+                        const LOG_UP_TO_N_ERRORS: usize = 10;
+                        for e in errors.iter().take(LOG_UP_TO_N_ERRORS) {
+                            tracing::warn!(
+                                "DeleteObjects key {} failed: {}: {}",
+                                e.key.as_ref().map(Cow::from).unwrap_or("".into()),
+                                e.code.as_ref().map(Cow::from).unwrap_or("".into()),
+                                e.message.as_ref().map(Cow::from).unwrap_or("".into())
+                            );
+                        }
+
                         return Err(anyhow::format_err!(
                             "Failed to delete {} objects",
                             errors.len()

pageserver/benches/bench_walredo.rs

@@ -11,10 +11,7 @@ use std::sync::{Arc, Barrier};
 
 use bytes::{Buf, Bytes};
 use pageserver::{
-    config::PageServerConf,
-    repository::Key,
-    walrecord::NeonWalRecord,
-    walredo::{PostgresRedoManager, WalRedoError},
+    config::PageServerConf, repository::Key, walrecord::NeonWalRecord, walredo::PostgresRedoManager,
 };
 use utils::{id::TenantId, lsn::Lsn};
 
@@ -152,7 +149,7 @@ impl Drop for JoinOnDrop {
     }
 }
 
-fn execute_all<I>(input: I, manager: &PostgresRedoManager) -> Result<(), WalRedoError>
+fn execute_all<I>(input: I, manager: &PostgresRedoManager) -> anyhow::Result<()>
 where
     I: IntoIterator<Item = Request>,
 {
@@ -160,7 +157,7 @@ where
     input.into_iter().try_for_each(|req| {
         let page = req.execute(manager)?;
         assert_eq!(page.remaining(), 8192);
-        Ok::<_, WalRedoError>(())
+        anyhow::Ok(())
     })
 }
 
@@ -473,7 +470,7 @@ struct Request {
 }
 
 impl Request {
-    fn execute(self, manager: &PostgresRedoManager) -> Result<Bytes, WalRedoError> {
+    fn execute(self, manager: &PostgresRedoManager) -> anyhow::Result<Bytes> {
         use pageserver::walredo::WalRedoManager;
 
         let Request {

pageserver/src/deletion_queue.rs

@@ -153,7 +153,7 @@ impl FlushOp {
 
 #[derive(Clone, Debug)]
 pub struct DeletionQueueClient {
-    tx: tokio::sync::mpsc::Sender<ListWriterQueueMessage>,
+    tx: tokio::sync::mpsc::UnboundedSender<ListWriterQueueMessage>,
     executor_tx: tokio::sync::mpsc::Sender<DeleterMessage>,
 
     lsn_table: Arc<std::sync::RwLock<VisibleLsnUpdates>>,
@@ -416,7 +416,7 @@ pub enum DeletionQueueError {
 impl DeletionQueueClient {
     pub(crate) fn broken() -> Self {
         // Channels whose receivers are immediately dropped.
-        let (tx, _rx) = tokio::sync::mpsc::channel(1);
+        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
         let (executor_tx, _executor_rx) = tokio::sync::mpsc::channel(1);
         Self {
             tx,
@@ -428,12 +428,12 @@ impl DeletionQueueClient {
     /// This is cancel-safe.  If you drop the future before it completes, the message
     /// is not pushed, although in the context of the deletion queue it doesn't matter: once
     /// we decide to do a deletion the decision is always final.
-    async fn do_push<T>(
+    fn do_push<T>(
         &self,
-        queue: &tokio::sync::mpsc::Sender<T>,
+        queue: &tokio::sync::mpsc::UnboundedSender<T>,
         msg: T,
     ) -> Result<(), DeletionQueueError> {
-        match queue.send(msg).await {
+        match queue.send(msg) {
             Ok(_) => Ok(()),
             Err(e) => {
                 // This shouldn't happen, we should shut down all tenants before
@@ -445,7 +445,7 @@ impl DeletionQueueClient {
         }
     }
 
-    pub(crate) async fn recover(
+    pub(crate) fn recover(
         &self,
         attached_tenants: HashMap<TenantId, Generation>,
     ) -> Result<(), DeletionQueueError> {
@@ -453,7 +453,6 @@ impl DeletionQueueClient {
             &self.tx,
             ListWriterQueueMessage::Recover(RecoverOp { attached_tenants }),
         )
-        .await
     }
 
     /// When a Timeline wishes to update the remote_consistent_lsn that it exposes to the outside
@@ -526,6 +525,21 @@ impl DeletionQueueClient {
             return self.flush_immediate().await;
         }
 
+        self.push_layers_sync(tenant_id, timeline_id, current_generation, layers)
+    }
+
+    /// When a Tenant has a generation, push_layers is always synchronous because
+    /// the ListValidator channel is an unbounded channel.
+    ///
+    /// This can be merged into push_layers when we remove the Generation-less mode
+    /// support (`<https://github.com/neondatabase/neon/issues/5395>`)
+    pub(crate) fn push_layers_sync(
+        &self,
+        tenant_id: TenantId,
+        timeline_id: TimelineId,
+        current_generation: Generation,
+        layers: Vec<(LayerFileName, Generation)>,
+    ) -> Result<(), DeletionQueueError> {
         metrics::DELETION_QUEUE
             .keys_submitted
             .inc_by(layers.len() as u64);
@@ -539,17 +553,16 @@ impl DeletionQueueClient {
                 objects: Vec::new(),
             }),
         )
-        .await
     }
 
     /// This is cancel-safe.  If you drop the future the flush may still happen in the background.
     async fn do_flush<T>(
         &self,
-        queue: &tokio::sync::mpsc::Sender<T>,
+        queue: &tokio::sync::mpsc::UnboundedSender<T>,
         msg: T,
         rx: tokio::sync::oneshot::Receiver<()>,
     ) -> Result<(), DeletionQueueError> {
-        self.do_push(queue, msg).await?;
+        self.do_push(queue, msg)?;
         if rx.await.is_err() {
             // This shouldn't happen if tenants are shut down before deletion queue.  If we
             // encounter a bug like this, then a flusher will incorrectly believe it has flushed
@@ -570,6 +583,18 @@ impl DeletionQueueClient {
             .await
     }
 
+    /// Issue a flush without waiting for it to complete.  This is useful on advisory flushes where
+    /// the caller wants to avoid the risk of waiting for lots of enqueued work, such as on tenant
+    /// detach where flushing is nice but not necessary.
+    ///
+    /// This function provides no guarantees of work being done.
+    pub fn flush_advisory(&self) {
+        let (flush_op, _) = FlushOp::new();
+
+        // Transmit the flush message, ignoring any result (such as a closed channel during shutdown).
+        drop(self.tx.send(ListWriterQueueMessage::FlushExecute(flush_op)));
+    }
+
     // Wait until all previous deletions are executed
     pub(crate) async fn flush_execute(&self) -> Result<(), DeletionQueueError> {
         debug!("flush_execute: flushing to deletion lists...");
@@ -586,9 +611,7 @@ impl DeletionQueueClient {
         // Flush any immediate-mode deletions (the above backend flush will only flush
         // the executor if deletions had flowed through the backend)
         debug!("flush_execute: flushing execution...");
-        let (flush_op, rx) = FlushOp::new();
-        self.do_flush(&self.executor_tx, DeleterMessage::Flush(flush_op), rx)
-            .await?;
+        self.flush_immediate().await?;
         debug!("flush_execute: finished flushing execution...");
         Ok(())
     }
@@ -643,8 +666,10 @@ impl DeletionQueue {
     where
         C: ControlPlaneGenerationsApi + Send + Sync,
     {
-        // Deep channel: it consumes deletions from all timelines and we do not want to block them
-        let (tx, rx) = tokio::sync::mpsc::channel(16384);
+        // Unbounded channel: enables non-async functions to submit deletions.  The actual length is
+        // constrained by how promptly the ListWriter wakes up and drains it, which should be frequent
+        // enough to avoid this taking pathologically large amount of memory.
+        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
 
         // Shallow channel: it carries DeletionLists which each contain up to thousands of deletions
         let (backend_tx, backend_rx) = tokio::sync::mpsc::channel(16);
@@ -957,7 +982,7 @@ mod test {
         // Basic test that the deletion queue processes the deletions we pass into it
         let ctx = setup("deletion_queue_smoke").expect("Failed test setup");
         let client = ctx.deletion_queue.new_client();
-        client.recover(HashMap::new()).await?;
+        client.recover(HashMap::new())?;
 
         let layer_file_name_1: LayerFileName = "000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__00000000016B59D8-00000000016B5A51".parse().unwrap();
         let tenant_id = ctx.harness.tenant_id;
@@ -1025,7 +1050,7 @@ mod test {
     async fn deletion_queue_validation() -> anyhow::Result<()> {
         let ctx = setup("deletion_queue_validation").expect("Failed test setup");
         let client = ctx.deletion_queue.new_client();
-        client.recover(HashMap::new()).await?;
+        client.recover(HashMap::new())?;
 
         // Generation that the control plane thinks is current
         let latest_generation = Generation::new(0xdeadbeef);
@@ -1082,7 +1107,7 @@ mod test {
         // Basic test that the deletion queue processes the deletions we pass into it
         let mut ctx = setup("deletion_queue_recovery").expect("Failed test setup");
         let client = ctx.deletion_queue.new_client();
-        client.recover(HashMap::new()).await?;
+        client.recover(HashMap::new())?;
 
         let tenant_id = ctx.harness.tenant_id;
 
@@ -1145,9 +1170,7 @@ mod test {
         drop(client);
         ctx.restart().await;
         let client = ctx.deletion_queue.new_client();
-        client
-            .recover(HashMap::from([(tenant_id, now_generation)]))
-            .await?;
+        client.recover(HashMap::from([(tenant_id, now_generation)]))?;
 
         info!("Flush-executing");
         client.flush_execute().await?;
@@ -1173,7 +1196,7 @@ pub(crate) mod mock {
     };
 
     pub struct ConsumerState {
-        rx: tokio::sync::mpsc::Receiver<ListWriterQueueMessage>,
+        rx: tokio::sync::mpsc::UnboundedReceiver<ListWriterQueueMessage>,
         executor_rx: tokio::sync::mpsc::Receiver<DeleterMessage>,
     }
 
@@ -1250,7 +1273,7 @@ pub(crate) mod mock {
     }
 
     pub struct MockDeletionQueue {
-        tx: tokio::sync::mpsc::Sender<ListWriterQueueMessage>,
+        tx: tokio::sync::mpsc::UnboundedSender<ListWriterQueueMessage>,
         executor_tx: tokio::sync::mpsc::Sender<DeleterMessage>,
         executed: Arc<AtomicUsize>,
         remote_storage: Option<GenericRemoteStorage>,
@@ -1260,7 +1283,7 @@ pub(crate) mod mock {
 
     impl MockDeletionQueue {
         pub fn new(remote_storage: Option<GenericRemoteStorage>) -> Self {
-            let (tx, rx) = tokio::sync::mpsc::channel(16384);
+            let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
             let (executor_tx, executor_rx) = tokio::sync::mpsc::channel(16384);
 
             let executed = Arc::new(AtomicUsize::new(0));

pageserver/src/deletion_queue/deleter.rs

@@ -13,6 +13,7 @@ use std::time::Duration;
 use tokio_util::sync::CancellationToken;
 use tracing::info;
 use tracing::warn;
+use utils::backoff;
 
 use crate::metrics;
 
@@ -63,7 +64,19 @@ impl Deleter {
             Err(anyhow::anyhow!("failpoint hit"))
         });
 
-        self.remote_storage.delete_objects(&self.accumulator).await
+        // A backoff::retry is used here for two reasons:
+        // - To provide a backoff rather than busy-polling the API on errors
+        // - To absorb transient 429/503 conditions without hitting our error
+        //   logging path for issues deleting objects.
+        backoff::retry(
+            || async { self.remote_storage.delete_objects(&self.accumulator).await },
+            |_| false,
+            3,
+            10,
+            "executing deletion batch",
+            backoff::Cancel::new(self.cancel.clone(), || anyhow::anyhow!("Shutting down")),
+        )
+        .await
     }
 
     /// Block until everything in accumulator has been executed
@@ -88,7 +101,10 @@ impl Deleter {
                     self.accumulator.clear();
                 }
                 Err(e) => {
-                    warn!("DeleteObjects request failed: {e:#}, will retry");
+                    if self.cancel.is_cancelled() {
+                        return Err(DeletionQueueError::ShuttingDown);
+                    }
+                    warn!("DeleteObjects request failed: {e:#}, will continue trying");
                     metrics::DELETION_QUEUE
                         .remote_errors
                         .with_label_values(&["execute"])

pageserver/src/deletion_queue/list_writer.rs

@@ -85,7 +85,7 @@ pub(super) struct ListWriter {
     conf: &'static PageServerConf,
 
     // Incoming frontend requests to delete some keys
-    rx: tokio::sync::mpsc::Receiver<ListWriterQueueMessage>,
+    rx: tokio::sync::mpsc::UnboundedReceiver<ListWriterQueueMessage>,
 
     // Outbound requests to the backend to execute deletion lists we have composed.
     tx: tokio::sync::mpsc::Sender<ValidatorQueueMessage>,
@@ -111,7 +111,7 @@ impl ListWriter {
 
     pub(super) fn new(
         conf: &'static PageServerConf,
-        rx: tokio::sync::mpsc::Receiver<ListWriterQueueMessage>,
+        rx: tokio::sync::mpsc::UnboundedReceiver<ListWriterQueueMessage>,
         tx: tokio::sync::mpsc::Sender<ValidatorQueueMessage>,
         cancel: CancellationToken,
     ) -> Self {

pageserver/src/http/routes.rs

@@ -77,7 +77,7 @@ impl State {
         disk_usage_eviction_state: Arc<disk_usage_eviction_task::State>,
         deletion_queue_client: DeletionQueueClient,
     ) -> anyhow::Result<Self> {
-        let allowlist_routes = ["/v1/status", "/v1/doc", "/swagger.yml"]
+        let allowlist_routes = ["/v1/status", "/v1/doc", "/swagger.yml", "/metrics"]
             .iter()
             .map(|v| v.parse().unwrap())
             .collect::<Vec<_>>();
@@ -136,9 +136,7 @@ impl From<PageReconstructError> for ApiError {
             PageReconstructError::AncestorStopping(_) => {
                 ApiError::ResourceUnavailable(format!("{pre}").into())
             }
-            PageReconstructError::WalRedo(pre) => {
-                ApiError::InternalServerError(anyhow::Error::new(pre))
-            }
+            PageReconstructError::WalRedo(pre) => ApiError::InternalServerError(pre),
         }
     }
 }
@@ -164,9 +162,6 @@ impl From<TenantStateError> for ApiError {
     fn from(tse: TenantStateError) -> ApiError {
         match tse {
             TenantStateError::NotFound(tid) => ApiError::NotFound(anyhow!("tenant {}", tid).into()),
-            TenantStateError::NotActive(_) => {
-                ApiError::ResourceUnavailable("Tenant not yet active".into())
-            }
             TenantStateError::IsStopping(_) => {
                 ApiError::ResourceUnavailable("Tenant is stopping".into())
             }
@@ -575,9 +570,14 @@ async fn tenant_detach_handler(
 
     let state = get_state(&request);
     let conf = state.conf;
-    mgr::detach_tenant(conf, tenant_id, detach_ignored.unwrap_or(false))
-        .instrument(info_span!("tenant_detach", %tenant_id))
-        .await?;
+    mgr::detach_tenant(
+        conf,
+        tenant_id,
+        detach_ignored.unwrap_or(false),
+        &state.deletion_queue_client,
+    )
+    .instrument(info_span!("tenant_detach", %tenant_id))
+    .await?;
 
     json_response(StatusCode::OK, ())
 }
@@ -1034,7 +1034,7 @@ async fn put_tenant_location_config_handler(
     // The `Detached` state is special, it doesn't upsert a tenant, it removes
     // its local disk content and drops it from memory.
     if let LocationConfigMode::Detached = request_data.config.mode {
-        mgr::detach_tenant(conf, tenant_id, true)
+        mgr::detach_tenant(conf, tenant_id, true, &state.deletion_queue_client)
             .instrument(info_span!("tenant_detach", %tenant_id))
             .await?;
         return json_response(StatusCode::OK, ());

pageserver/src/tenant.rs

@@ -45,6 +45,7 @@ use std::sync::{Mutex, RwLock};
 use std::time::{Duration, Instant};
 
 use self::config::AttachedLocationConfig;
+use self::config::AttachmentMode;
 use self::config::LocationConf;
 use self::config::TenantConf;
 use self::delete::DeleteTenantFlow;
@@ -208,7 +209,7 @@ pub struct Tenant {
 
     /// The remote storage generation, used to protect S3 objects from split-brain.
     /// Does not change over the lifetime of the [`Tenant`] object.
-    ///  
+    ///
     /// This duplicates the generation stored in LocationConf, but that structure is mutable:
     /// this copy enforces the invariant that generatio doesn't change during a Tenant's lifetime.
     generation: Generation,
@@ -2076,6 +2077,15 @@ impl Tenant {
             }
         }
     }
+
+    pub(crate) fn get_attach_mode(&self) -> AttachmentMode {
+        self.tenant_conf
+            .read()
+            .unwrap()
+            .location
+            .attach_mode
+            .clone()
+    }
 }
 
 /// Given a Vec of timelines and their ancestors (timeline_id, ancestor_id),
@@ -2745,6 +2755,11 @@ impl Tenant {
     ) -> Result<Arc<Timeline>, CreateTimelineError> {
         let src_id = src_timeline.timeline_id;
 
+        // First acquire the GC lock so that another task cannot advance the GC
+        // cutoff in 'gc_info', and make 'start_lsn' invalid, while we are
+        // creating the branch.
+        let _gc_cs = self.gc_cs.lock().await;
+
         // If no start LSN is specified, we branch the new timeline from the source timeline's last record LSN
         let start_lsn = start_lsn.unwrap_or_else(|| {
             let lsn = src_timeline.get_last_record_lsn();
@@ -2752,11 +2767,6 @@ impl Tenant {
             lsn
         });
 
-        // First acquire the GC lock so that another task cannot advance the GC
-        // cutoff in 'gc_info', and make 'start_lsn' invalid, while we are
-        // creating the branch.
-        let _gc_cs = self.gc_cs.lock().await;
-
         // Create a placeholder for the new branch. This will error
         // out if the new timeline ID is already in use.
         let timeline_uninit_mark = {
@@ -3438,11 +3448,8 @@ pub mod harness {
 
     use crate::deletion_queue::mock::MockDeletionQueue;
     use crate::{
-        config::PageServerConf,
-        repository::Key,
-        tenant::Tenant,
-        walrecord::NeonWalRecord,
-        walredo::{WalRedoError, WalRedoManager},
+        config::PageServerConf, repository::Key, tenant::Tenant, walrecord::NeonWalRecord,
+        walredo::WalRedoManager,
     };
 
     use super::*;
@@ -3615,7 +3622,7 @@ pub mod harness {
             base_img: Option<(Lsn, Bytes)>,
             records: Vec<(Lsn, NeonWalRecord)>,
             _pg_version: u32,
-        ) -> Result<Bytes, WalRedoError> {
+        ) -> anyhow::Result<Bytes> {
             let s = format!(
                 "redo for {} to get to {}, with {} and {} records",
                 key,

pageserver/src/tenant/delete.rs

@@ -31,7 +31,7 @@ use super::{
 const SHOULD_RESUME_DELETION_FETCH_MARK_ATTEMPTS: u32 = 3;
 
 #[derive(Debug, thiserror::Error)]
-pub enum DeleteTenantError {
+pub(crate) enum DeleteTenantError {
     #[error("GetTenant {0}")]
     Get(#[from] GetTenantError),
 
@@ -376,7 +376,7 @@ impl DeleteTenantFlow {
         Ok(())
     }
 
-    pub async fn should_resume_deletion(
+    pub(crate) async fn should_resume_deletion(
         conf: &'static PageServerConf,
         remote_storage: Option<&GenericRemoteStorage>,
         tenant: &Tenant,

pageserver/src/tenant/mgr.rs

@@ -24,7 +24,7 @@ use crate::control_plane_client::{
 };
 use crate::deletion_queue::DeletionQueueClient;
 use crate::task_mgr::{self, TaskKind};
-use crate::tenant::config::{LocationConf, LocationMode, TenantConfOpt};
+use crate::tenant::config::{AttachmentMode, LocationConf, LocationMode, TenantConfOpt};
 use crate::tenant::delete::DeleteTenantFlow;
 use crate::tenant::{
     create_tenant_files, AttachedTenantConf, CreateTenantFilesMode, Tenant, TenantState,
@@ -50,7 +50,7 @@ use super::TenantSharedResources;
 /// its lifetime, and we can preserve some important safety invariants like `Tenant` always
 /// having a properly acquired generation (Secondary doesn't need a generation)
 #[derive(Clone)]
-pub enum TenantSlot {
+pub(crate) enum TenantSlot {
     Attached(Arc<Tenant>),
     Secondary,
 }
@@ -206,8 +206,7 @@ async fn init_load_generations(
     if resources.remote_storage.is_some() {
         resources
             .deletion_queue_client
-            .recover(generations.clone())
-            .await?;
+            .recover(generations.clone())?;
     }
 
     Ok(Some(generations))
@@ -482,7 +481,7 @@ pub(crate) fn schedule_local_tenant_processing(
 /// management API. For example, it could attach the tenant on a different pageserver.
 /// We would then be in split-brain once this pageserver restarts.
 #[instrument(skip_all)]
-pub async fn shutdown_all_tenants() {
+pub(crate) async fn shutdown_all_tenants() {
     shutdown_all_tenants0(&TENANTS).await
 }
 
@@ -594,7 +593,7 @@ async fn shutdown_all_tenants0(tenants: &tokio::sync::RwLock<TenantsMap>) {
     // caller will log how long we took
 }
 
-pub async fn create_tenant(
+pub(crate) async fn create_tenant(
     conf: &'static PageServerConf,
     tenant_conf: TenantConfOpt,
     tenant_id: TenantId,
@@ -629,14 +628,14 @@ pub async fn create_tenant(
 }
 
 #[derive(Debug, thiserror::Error)]
-pub enum SetNewTenantConfigError {
+pub(crate) enum SetNewTenantConfigError {
     #[error(transparent)]
     GetTenant(#[from] GetTenantError),
     #[error(transparent)]
     Persist(anyhow::Error),
 }
 
-pub async fn set_new_tenant_config(
+pub(crate) async fn set_new_tenant_config(
     conf: &'static PageServerConf,
     new_tenant_conf: TenantConfOpt,
     tenant_id: TenantId,
@@ -695,6 +694,18 @@ pub(crate) async fn upsert_location(
 
     if let Some(tenant) = shutdown_tenant {
         let (_guard, progress) = utils::completion::channel();
+
+        match tenant.get_attach_mode() {
+            AttachmentMode::Single | AttachmentMode::Multi => {
+                // Before we leave our state as the presumed holder of the latest generation,
+                // flush any outstanding deletions to reduce the risk of leaking objects.
+                deletion_queue_client.flush_advisory()
+            }
+            AttachmentMode::Stale => {
+                // If we're stale there's not point trying to flush deletions
+            }
+        };
+
         info!("Shutting down attached tenant");
         match tenant.shutdown(progress, false).await {
             Ok(()) => {}
@@ -765,7 +776,7 @@ pub(crate) async fn upsert_location(
 }
 
 #[derive(Debug, thiserror::Error)]
-pub enum GetTenantError {
+pub(crate) enum GetTenantError {
     #[error("Tenant {0} not found")]
     NotFound(TenantId),
     #[error("Tenant {0} is not active")]
@@ -781,7 +792,7 @@ pub enum GetTenantError {
 /// `active_only = true` allows to query only tenants that are ready for operations, erroring on other kinds of tenants.
 ///
 /// This method is cancel-safe.
-pub async fn get_tenant(
+pub(crate) async fn get_tenant(
     tenant_id: TenantId,
     active_only: bool,
 ) -> Result<Arc<Tenant>, GetTenantError> {
@@ -806,7 +817,7 @@ pub async fn get_tenant(
     }
 }
 
-pub async fn delete_tenant(
+pub(crate) async fn delete_tenant(
     conf: &'static PageServerConf,
     remote_storage: Option<GenericRemoteStorage>,
     tenant_id: TenantId,
@@ -815,7 +826,7 @@ pub async fn delete_tenant(
 }
 
 #[derive(Debug, thiserror::Error)]
-pub enum DeleteTimelineError {
+pub(crate) enum DeleteTimelineError {
     #[error("Tenant {0}")]
     Tenant(#[from] GetTenantError),
 
@@ -823,7 +834,7 @@ pub enum DeleteTimelineError {
     Timeline(#[from] crate::tenant::DeleteTimelineError),
 }
 
-pub async fn delete_timeline(
+pub(crate) async fn delete_timeline(
     tenant_id: TenantId,
     timeline_id: TimelineId,
     _ctx: &RequestContext,
@@ -834,23 +845,29 @@ pub async fn delete_timeline(
 }
 
 #[derive(Debug, thiserror::Error)]
-pub enum TenantStateError {
+pub(crate) enum TenantStateError {
     #[error("Tenant {0} not found")]
     NotFound(TenantId),
     #[error("Tenant {0} is stopping")]
     IsStopping(TenantId),
-    #[error("Tenant {0} is not active")]
-    NotActive(TenantId),
     #[error(transparent)]
     Other(#[from] anyhow::Error),
 }
 
-pub async fn detach_tenant(
+pub(crate) async fn detach_tenant(
     conf: &'static PageServerConf,
     tenant_id: TenantId,
     detach_ignored: bool,
+    deletion_queue_client: &DeletionQueueClient,
 ) -> Result<(), TenantStateError> {
-    let tmp_path = detach_tenant0(conf, &TENANTS, tenant_id, detach_ignored).await?;
+    let tmp_path = detach_tenant0(
+        conf,
+        &TENANTS,
+        tenant_id,
+        detach_ignored,
+        deletion_queue_client,
+    )
+    .await?;
     // Although we are cleaning up the tenant, this task is not meant to be bound by the lifetime of the tenant in memory.
     // After a tenant is detached, there are no more task_mgr tasks for that tenant_id.
     let task_tenant_id = None;
@@ -875,6 +892,7 @@ async fn detach_tenant0(
     tenants: &tokio::sync::RwLock<TenantsMap>,
     tenant_id: TenantId,
     detach_ignored: bool,
+    deletion_queue_client: &DeletionQueueClient,
 ) -> Result<Utf8PathBuf, TenantStateError> {
     let tenant_dir_rename_operation = |tenant_id_to_clean| async move {
         let local_tenant_directory = conf.tenant_path(&tenant_id_to_clean);
@@ -886,6 +904,10 @@ async fn detach_tenant0(
     let removal_result =
         remove_tenant_from_memory(tenants, tenant_id, tenant_dir_rename_operation(tenant_id)).await;
 
+    // Flush pending deletions, so that they have a good chance of passing validation
+    // before this tenant is potentially re-attached elsewhere.
+    deletion_queue_client.flush_advisory();
+
     // Ignored tenants are not present in memory and will bail the removal from memory operation.
     // Before returning the error, check for ignored tenant removal case — we only need to clean its local files then.
     if detach_ignored && matches!(removal_result, Err(TenantStateError::NotFound(_))) {
@@ -902,7 +924,7 @@ async fn detach_tenant0(
     removal_result
 }
 
-pub async fn load_tenant(
+pub(crate) async fn load_tenant(
     conf: &'static PageServerConf,
     tenant_id: TenantId,
     generation: Generation,
@@ -939,7 +961,7 @@ pub async fn load_tenant(
     Ok(())
 }
 
-pub async fn ignore_tenant(
+pub(crate) async fn ignore_tenant(
     conf: &'static PageServerConf,
     tenant_id: TenantId,
 ) -> Result<(), TenantStateError> {
@@ -967,7 +989,7 @@ async fn ignore_tenant0(
 }
 
 #[derive(Debug, thiserror::Error)]
-pub enum TenantMapListError {
+pub(crate) enum TenantMapListError {
     #[error("tenant map is still initiailizing")]
     Initializing,
 }
@@ -975,7 +997,7 @@ pub enum TenantMapListError {
 ///
 /// Get list of tenants, for the mgmt API
 ///
-pub async fn list_tenants() -> Result<Vec<(TenantId, TenantState)>, TenantMapListError> {
+pub(crate) async fn list_tenants() -> Result<Vec<(TenantId, TenantState)>, TenantMapListError> {
     let tenants = TENANTS.read().await;
     let m = match &*tenants {
         TenantsMap::Initializing => return Err(TenantMapListError::Initializing),
@@ -993,7 +1015,7 @@ pub async fn list_tenants() -> Result<Vec<(TenantId, TenantState)>, TenantMapLis
 ///
 /// Downloading all the tenant data is performed in the background, this merely
 /// spawns the background task and returns quickly.
-pub async fn attach_tenant(
+pub(crate) async fn attach_tenant(
     conf: &'static PageServerConf,
     tenant_id: TenantId,
     generation: Generation,
@@ -1030,7 +1052,7 @@ pub async fn attach_tenant(
 }
 
 #[derive(Debug, thiserror::Error)]
-pub enum TenantMapInsertError {
+pub(crate) enum TenantMapInsertError {
     #[error("tenant map is still initializing")]
     StillInitializing,
     #[error("tenant map is shutting down")]
@@ -1193,7 +1215,7 @@ use {
     utils::http::error::ApiError,
 };
 
-pub async fn immediate_gc(
+pub(crate) async fn immediate_gc(
     tenant_id: TenantId,
     timeline_id: TimelineId,
     gc_req: TimelineGcRequest,

pageserver/src/tenant/timeline.rs

@@ -370,7 +370,7 @@ pub enum PageReconstructError {
 
     /// An error happened replaying WAL records
     #[error(transparent)]
-    WalRedo(#[from] crate::walredo::WalRedoError),
+    WalRedo(anyhow::Error),
 }
 
 impl std::fmt::Debug for PageReconstructError {
@@ -4327,7 +4327,7 @@ impl Timeline {
                 let img = match self
                     .walredo_mgr
                     .request_redo(key, request_lsn, data.img, data.records, self.pg_version)
-                    .context("Failed to reconstruct a page image:")
+                    .context("Failed to reconstruct a page image")
                 {
                     Ok(img) => img,
                     Err(e) => return Err(PageReconstructError::from(e)),

pageserver/src/walredo.rs

@@ -18,6 +18,7 @@
 //! any WAL records, so that even if an attacker hijacks the Postgres
 //! process, he cannot escape out of it.
 //!
+use anyhow::Context;
 use byteorder::{ByteOrder, LittleEndian};
 use bytes::{BufMut, Bytes, BytesMut};
 use nix::poll::*;
@@ -89,7 +90,7 @@ pub trait WalRedoManager: Send + Sync {
         base_img: Option<(Lsn, Bytes)>,
         records: Vec<(Lsn, NeonWalRecord)>,
         pg_version: u32,
-    ) -> Result<Bytes, WalRedoError>;
+    ) -> anyhow::Result<Bytes>;
 }
 
 struct ProcessInput {
@@ -100,6 +101,14 @@ struct ProcessInput {
     n_requests: usize,
 }
 
+enum ApplyWalRecordsError {
+    WithRequestNo {
+        error: anyhow::Error,
+        request_no: usize,
+    },
+    NoRequestNo(anyhow::Error),
+}
+
 struct ProcessOutput {
     stdout: ChildStdout,
     pending_responses: VecDeque<Option<Bytes>>,
@@ -140,20 +149,6 @@ fn can_apply_in_neon(rec: &NeonWalRecord) -> bool {
     }
 }
 
-/// An error happened in WAL redo
-#[derive(Debug, thiserror::Error)]
-pub enum WalRedoError {
-    #[error(transparent)]
-    IoError(#[from] std::io::Error),
-
-    #[error("cannot perform WAL redo now")]
-    InvalidState,
-    #[error("cannot perform WAL redo for this request")]
-    InvalidRequest,
-    #[error("cannot perform WAL redo for this record")]
-    InvalidRecord,
-}
-
 ///
 /// Public interface of WAL redo manager
 ///
@@ -171,10 +166,9 @@ impl WalRedoManager for PostgresRedoManager {
         base_img: Option<(Lsn, Bytes)>,
         records: Vec<(Lsn, NeonWalRecord)>,
         pg_version: u32,
-    ) -> Result<Bytes, WalRedoError> {
+    ) -> anyhow::Result<Bytes> {
         if records.is_empty() {
-            error!("invalid WAL redo request with no records");
-            return Err(WalRedoError::InvalidRequest);
+            anyhow::bail!("invalid WAL redo request with no records");
         }
 
         let base_img_lsn = base_img.as_ref().map(|p| p.0).unwrap_or(Lsn::INVALID);
@@ -186,7 +180,13 @@ impl WalRedoManager for PostgresRedoManager {
 
             if rec_neon != batch_neon {
                 let result = if batch_neon {
-                    self.apply_batch_neon(key, lsn, img, &records[batch_start..i])
+                    self.apply_batch_neon(
+                        key,
+                        lsn,
+                        img,
+                        &records[batch_start..i],
+                        (batch_start, records.len()),
+                    )
                 } else {
                     self.apply_batch_postgres(
                         key,
@@ -196,6 +196,7 @@ impl WalRedoManager for PostgresRedoManager {
                         &records[batch_start..i],
                         self.conf.wal_redo_timeout,
                         pg_version,
+                        (batch_start, records.len()),
                     )
                 };
                 img = Some(result?);
@@ -206,7 +207,13 @@ impl WalRedoManager for PostgresRedoManager {
         }
         // last batch
         if batch_neon {
-            self.apply_batch_neon(key, lsn, img, &records[batch_start..])
+            self.apply_batch_neon(
+                key,
+                lsn,
+                img,
+                &records[batch_start..],
+                (batch_start, records.len()),
+            )
         } else {
             self.apply_batch_postgres(
                 key,
@@ -216,6 +223,7 @@ impl WalRedoManager for PostgresRedoManager {
                 &records[batch_start..],
                 self.conf.wal_redo_timeout,
                 pg_version,
+                (batch_start, records.len()),
             )
         }
     }
@@ -260,8 +268,9 @@ impl PostgresRedoManager {
         records: &[(Lsn, NeonWalRecord)],
         wal_redo_timeout: Duration,
         pg_version: u32,
-    ) -> Result<Bytes, WalRedoError> {
-        let (rel, blknum) = key_to_rel_block(key).or(Err(WalRedoError::InvalidRecord))?;
+        diag: impl std::fmt::Debug,
+    ) -> anyhow::Result<Bytes> {
+        let (rel, blknum) = key_to_rel_block(key).context("invalid record")?;
         const MAX_RETRY_ATTEMPTS: u32 = 1;
         let start_time = Instant::now();
         let mut n_attempts = 0u32;
@@ -271,15 +280,15 @@ impl PostgresRedoManager {
 
             // launch the WAL redo process on first use
             if proc.is_none() {
-                self.launch(&mut proc, pg_version)?;
+                self.launch(&mut proc, pg_version)
+                    .context("launch process")?;
             }
             WAL_REDO_WAIT_TIME.observe(lock_time.duration_since(start_time).as_secs_f64());
 
             // Relational WAL records are applied using wal-redo-postgres
             let buf_tag = BufferTag { rel, blknum };
-            let result = self
-                .apply_wal_records(proc, buf_tag, &base_img, records, wal_redo_timeout)
-                .map_err(WalRedoError::IoError);
+            let result =
+                self.apply_wal_records(proc, buf_tag, &base_img, records, wal_redo_timeout);
 
             let end_time = Instant::now();
             let duration = end_time.duration_since(lock_time);
@@ -298,26 +307,35 @@ impl PostgresRedoManager {
             WAL_REDO_BYTES_HISTOGRAM.observe(nbytes as f64);
 
             debug!(
-                "postgres applied {} WAL records ({} bytes) in {} us to reconstruct page image at LSN {}",
+                "postgres applied {} WAL records ({} bytes) in {} us to reconstruct page image at LSN {} diag={:?}",
                 len,
                 nbytes,
                 duration.as_micros(),
-                lsn
+                lsn,
+                diag,
             );
 
             // If something went wrong, don't try to reuse the process. Kill it, and
             // next request will launch a new one.
             if let Err(e) = result.as_ref() {
+                let (e, request_no) = match e {
+                    ApplyWalRecordsError::WithRequestNo { error, request_no } => {
+                        (error, Some(*request_no))
+                    }
+                    ApplyWalRecordsError::NoRequestNo(e) => (e, None),
+                };
                 error!(
-                    n_attempts,
-                    "error applying {} WAL records {}..{} ({} bytes) to base image with LSN {} to reconstruct page image at LSN {}: {}",
+                    "error applying {} WAL records {}..{} ({} bytes) to base image with LSN {} to reconstruct page image at LSN {} n_attempts={} request_no={:?} {:?}: {:?}",
                     records.len(),
                     records.first().map(|p| p.0).unwrap_or(Lsn(0)),
                     records.last().map(|p| p.0).unwrap_or(Lsn(0)),
                     nbytes,
                     base_img_lsn,
                     lsn,
-                    utils::error::report_compact_sources(e),
+                    n_attempts,
+                    request_no,
+                    diag,
+                    e,
                 );
                 // self.stdin only holds stdin & stderr as_raw_fd().
                 // Dropping it as part of take() doesn't close them.
@@ -336,11 +354,14 @@ impl PostgresRedoManager {
                     proc.child.kill_and_wait();
                 }
             } else if n_attempts != 0 {
-                info!(n_attempts, "retried walredo succeeded");
+                info!(n_attempts, ?diag, "retried walredo succeeded");
             }
             n_attempts += 1;
             if n_attempts > MAX_RETRY_ATTEMPTS || result.is_ok() {
-                return result;
+                return result.map_err(|e| match e {
+                    ApplyWalRecordsError::WithRequestNo { error, .. } => error,
+                    ApplyWalRecordsError::NoRequestNo(e) => e,
+                });
             }
         }
     }
@@ -354,7 +375,8 @@ impl PostgresRedoManager {
         lsn: Lsn,
         base_img: Option<Bytes>,
         records: &[(Lsn, NeonWalRecord)],
-    ) -> Result<Bytes, WalRedoError> {
+        diag: impl std::fmt::Debug,
+    ) -> anyhow::Result<Bytes> {
         let start_time = Instant::now();
 
         let mut page = BytesMut::new();
@@ -363,8 +385,7 @@ impl PostgresRedoManager {
             page.extend_from_slice(&fpi[..]);
         } else {
             // All the current WAL record types that we can handle require a base image.
-            error!("invalid neon WAL redo request with no base image");
-            return Err(WalRedoError::InvalidRequest);
+            anyhow::bail!("invalid neon WAL redo request with no base image");
         }
 
         // Apply all the WAL records in the batch
@@ -377,10 +398,11 @@ impl PostgresRedoManager {
         WAL_REDO_TIME.observe(duration.as_secs_f64());
 
         debug!(
-            "neon applied {} WAL records in {} ms to reconstruct page image at LSN {}",
+            "neon applied {} WAL records in {} ms to reconstruct page image at LSN {} diag={:?}",
             records.len(),
             duration.as_micros(),
-            lsn
+            lsn,
+            diag,
         );
 
         Ok(page.freeze())
@@ -392,14 +414,13 @@ impl PostgresRedoManager {
         page: &mut BytesMut,
         _record_lsn: Lsn,
         record: &NeonWalRecord,
-    ) -> Result<(), WalRedoError> {
+    ) -> anyhow::Result<()> {
         match record {
             NeonWalRecord::Postgres {
                 will_init: _,
                 rec: _,
             } => {
-                error!("tried to pass postgres wal record to neon WAL redo");
-                return Err(WalRedoError::InvalidRequest);
+                anyhow::bail!("tried to pass postgres wal record to neon WAL redo");
             }
             NeonWalRecord::ClearVisibilityMapFlags {
                 new_heap_blkno,
@@ -407,7 +428,7 @@ impl PostgresRedoManager {
                 flags,
             } => {
                 // sanity check that this is modifying the correct relation
-                let (rel, blknum) = key_to_rel_block(key).or(Err(WalRedoError::InvalidRecord))?;
+                let (rel, blknum) = key_to_rel_block(key).context("invalid record")?;
                 assert!(
                     rel.forknum == VISIBILITYMAP_FORKNUM,
                     "ClearVisibilityMapFlags record on unexpected rel {}",
@@ -445,7 +466,7 @@ impl PostgresRedoManager {
             // same effects as the corresponding Postgres WAL redo function.
             NeonWalRecord::ClogSetCommitted { xids, timestamp } => {
                 let (slru_kind, segno, blknum) =
-                    key_to_slru_block(key).or(Err(WalRedoError::InvalidRecord))?;
+                    key_to_slru_block(key).context("invalid record")?;
                 assert_eq!(
                     slru_kind,
                     SlruKind::Clog,
@@ -495,7 +516,7 @@ impl PostgresRedoManager {
             }
             NeonWalRecord::ClogSetAborted { xids } => {
                 let (slru_kind, segno, blknum) =
-                    key_to_slru_block(key).or(Err(WalRedoError::InvalidRecord))?;
+                    key_to_slru_block(key).context("invalid record")?;
                 assert_eq!(
                     slru_kind,
                     SlruKind::Clog,
@@ -526,7 +547,7 @@ impl PostgresRedoManager {
             }
             NeonWalRecord::MultixactOffsetCreate { mid, moff } => {
                 let (slru_kind, segno, blknum) =
-                    key_to_slru_block(key).or(Err(WalRedoError::InvalidRecord))?;
+                    key_to_slru_block(key).context("invalid record")?;
                 assert_eq!(
                     slru_kind,
                     SlruKind::MultiXactOffsets,
@@ -559,7 +580,7 @@ impl PostgresRedoManager {
             }
             NeonWalRecord::MultixactMembersCreate { moff, members } => {
                 let (slru_kind, segno, blknum) =
-                    key_to_slru_block(key).or(Err(WalRedoError::InvalidRecord))?;
+                    key_to_slru_block(key).context("invalid record")?;
                 assert_eq!(
                     slru_kind,
                     SlruKind::MultiXactMembers,
@@ -759,7 +780,7 @@ impl PostgresRedoManager {
         base_img: &Option<Bytes>,
         records: &[(Lsn, NeonWalRecord)],
         wal_redo_timeout: Duration,
-    ) -> Result<Bytes, std::io::Error> {
+    ) -> Result<Bytes, ApplyWalRecordsError> {
         // Serialize all the messages to send the WAL redo process first.
         //
         // This could be problematic if there are millions of records to replay,
@@ -782,10 +803,9 @@ impl PostgresRedoManager {
             {
                 build_apply_record_msg(*lsn, postgres_rec, &mut writebuf);
             } else {
-                return Err(Error::new(
-                    ErrorKind::Other,
-                    "tried to pass neon wal record to postgres WAL redo",
-                ));
+                return Err(ApplyWalRecordsError::NoRequestNo(anyhow::anyhow!(
+                    "tried to pass neon wal record to postgres WAL redo"
+                )));
             }
         }
         build_get_page_msg(tag, &mut writebuf);
@@ -807,7 +827,7 @@ impl PostgresRedoManager {
         writebuf: &[u8],
         mut input: MutexGuard<Option<ProcessInput>>,
         wal_redo_timeout: Duration,
-    ) -> Result<Bytes, std::io::Error> {
+    ) -> Result<Bytes, ApplyWalRecordsError> {
         let proc = input.as_mut().unwrap();
         let mut nwrite = 0usize;
         let stdout_fd = proc.stdout_fd;
@@ -828,10 +848,13 @@ impl PostgresRedoManager {
                     Err(nix::errno::Errno::EINTR) => continue,
                     res => break res,
                 }
-            }?;
+            }
+            .map_err(|e| ApplyWalRecordsError::NoRequestNo(anyhow::anyhow!(e)))?;
 
             if n == 0 {
-                return Err(Error::new(ErrorKind::Other, "WAL redo timed out"));
+                return Err(ApplyWalRecordsError::NoRequestNo(anyhow::anyhow!(
+                    "WAL redo timed out"
+                )));
             }
 
             // If we have some messages in stderr, forward them to the log.
@@ -840,7 +863,9 @@ impl PostgresRedoManager {
                 let mut errbuf: [u8; 16384] = [0; 16384];
                 let mut stderr_guard = self.stderr.lock().unwrap();
                 let stderr = stderr_guard.as_mut().unwrap();
-                let len = stderr.read(&mut errbuf)?;
+                let len = stderr
+                    .read(&mut errbuf)
+                    .map_err(|e| ApplyWalRecordsError::NoRequestNo(anyhow::anyhow!(e)))?;
 
                 // The message might not be split correctly into lines here. But this is
                 // good enough, the important thing is to get the message to the log.
@@ -855,22 +880,23 @@ impl PostgresRedoManager {
                     continue;
                 }
             } else if err_revents.contains(PollFlags::POLLHUP) {
-                return Err(Error::new(
-                    ErrorKind::BrokenPipe,
-                    "WAL redo process closed its stderr unexpectedly",
-                ));
+                return Err(ApplyWalRecordsError::NoRequestNo(anyhow::anyhow!(
+                    "WAL redo process closed its stderr unexpectedly"
+                )));
             }
 
             // If 'stdin' is writeable, do write.
             let in_revents = pollfds[0].revents().unwrap();
             if in_revents & (PollFlags::POLLERR | PollFlags::POLLOUT) != PollFlags::empty() {
-                nwrite += proc.stdin.write(&writebuf[nwrite..])?;
+                nwrite += proc
+                    .stdin
+                    .write(&writebuf[nwrite..])
+                    .map_err(|e| ApplyWalRecordsError::NoRequestNo(anyhow::anyhow!(e)))?;
             } else if in_revents.contains(PollFlags::POLLHUP) {
                 // We still have more data to write, but the process closed the pipe.
-                return Err(Error::new(
-                    ErrorKind::BrokenPipe,
-                    "WAL redo process closed its stdin unexpectedly",
-                ));
+                return Err(ApplyWalRecordsError::NoRequestNo(anyhow::anyhow!(
+                    "WAL redo process closed its stdin unexpectedly"
+                )));
             }
         }
         let request_no = proc.n_requests;
@@ -901,10 +927,10 @@ impl PostgresRedoManager {
             //
             // Cross-read this with the comment in apply_batch_postgres if result.is_err().
             // That's where we kill the child process.
-            return Err(Error::new(
-                ErrorKind::BrokenPipe,
-                "WAL redo process closed its stdout unexpectedly",
-            ));
+            return Err(ApplyWalRecordsError::WithRequestNo {
+                request_no,
+                error: anyhow::anyhow!("WAL redo process closed its stdout unexpectedly"),
+            });
         }
         let n_processed_responses = output.n_processed_responses;
         while n_processed_responses + output.pending_responses.len() <= request_no {
@@ -920,10 +946,14 @@ impl PostgresRedoManager {
                         Err(nix::errno::Errno::EINTR) => continue,
                         res => break res,
                     }
-                }?;
+                }
+                .map_err(|e| ApplyWalRecordsError::NoRequestNo(anyhow::anyhow!(e)))?;
 
                 if n == 0 {
-                    return Err(Error::new(ErrorKind::Other, "WAL redo timed out"));
+                    return Err(ApplyWalRecordsError::WithRequestNo {
+                        request_no,
+                        error: anyhow::anyhow!("WAL redo timed out"),
+                    });
                 }
 
                 // If we have some messages in stderr, forward them to the log.
@@ -932,7 +962,12 @@ impl PostgresRedoManager {
                     let mut errbuf: [u8; 16384] = [0; 16384];
                     let mut stderr_guard = self.stderr.lock().unwrap();
                     let stderr = stderr_guard.as_mut().unwrap();
-                    let len = stderr.read(&mut errbuf)?;
+                    let len = stderr.read(&mut errbuf).map_err(|e| {
+                        ApplyWalRecordsError::WithRequestNo {
+                            request_no,
+                            error: anyhow::anyhow!(e),
+                        }
+                    })?;
 
                     // The message might not be split correctly into lines here. But this is
                     // good enough, the important thing is to get the message to the log.
@@ -947,21 +982,26 @@ impl PostgresRedoManager {
                         continue;
                     }
                 } else if err_revents.contains(PollFlags::POLLHUP) {
-                    return Err(Error::new(
-                        ErrorKind::BrokenPipe,
-                        "WAL redo process closed its stderr unexpectedly",
-                    ));
+                    return Err(ApplyWalRecordsError::WithRequestNo {
+                        request_no,
+                        error: anyhow::anyhow!("WAL redo process closed its stderr unexpectedly"),
+                    });
                 }
 
                 // If we have some data in stdout, read it to the result buffer.
                 let out_revents = pollfds[2].revents().unwrap();
                 if out_revents & (PollFlags::POLLERR | PollFlags::POLLIN) != PollFlags::empty() {
-                    nresult += output.stdout.read(&mut resultbuf[nresult..])?;
+                    nresult += output.stdout.read(&mut resultbuf[nresult..]).map_err(|e| {
+                        ApplyWalRecordsError::WithRequestNo {
+                            request_no,
+                            error: anyhow::anyhow!(e),
+                        }
+                    })?;
                 } else if out_revents.contains(PollFlags::POLLHUP) {
-                    return Err(Error::new(
-                        ErrorKind::BrokenPipe,
-                        "WAL redo process closed its stdout unexpectedly",
-                    ));
+                    return Err(ApplyWalRecordsError::WithRequestNo {
+                        request_no,
+                        error: anyhow::anyhow!("WAL redo process closed its stdout unexpectedly"),
+                    });
                 }
             }
             output

proxy/src/bin/proxy.rs

@@ -1,5 +1,6 @@
 use futures::future::Either;
 use proxy::auth;
+use proxy::config::HttpConfig;
 use proxy::console;
 use proxy::http;
 use proxy::metrics;
@@ -79,6 +80,9 @@ struct ProxyCliArgs {
     /// Allow self-signed certificates for compute nodes (for testing)
     #[clap(long, default_value_t = false, value_parser = clap::builder::BoolishValueParser::new(), action = clap::ArgAction::Set)]
     allow_self_signed_compute: bool,
+    /// timeout for http connections
+    #[clap(long, default_value = "15s", value_parser = humantime::parse_duration)]
+    sql_over_http_timeout: tokio::time::Duration,
 }
 
 #[tokio::main]
@@ -220,12 +224,15 @@ fn build_config(args: &ProxyCliArgs) -> anyhow::Result<&'static ProxyConfig> {
             auth::BackendType::Link(Cow::Owned(url))
         }
     };
-
+    let http_config = HttpConfig {
+        sql_over_http_timeout: args.sql_over_http_timeout,
+    };
     let config = Box::leak(Box::new(ProxyConfig {
         tls_config,
         auth_backend,
         metric_collection,
         allow_self_signed_compute: args.allow_self_signed_compute,
+        http_config,
     }));
 
     Ok(config)

proxy/src/config.rs

@@ -13,6 +13,7 @@ pub struct ProxyConfig {
     pub auth_backend: auth::BackendType<'static, ()>,
     pub metric_collection: Option<MetricCollectionConfig>,
     pub allow_self_signed_compute: bool,
+    pub http_config: HttpConfig,
 }
 
 #[derive(Debug)]
@@ -26,6 +27,10 @@ pub struct TlsConfig {
     pub common_names: Option<HashSet<String>>,
 }
 
+pub struct HttpConfig {
+    pub sql_over_http_timeout: tokio::time::Duration,
+}
+
 impl TlsConfig {
     pub fn to_server_config(&self) -> Arc<rustls::ServerConfig> {
         self.config.clone()

proxy/src/http/conn_pool.rs

@@ -20,6 +20,7 @@ use tokio_postgres::AsyncMessage;
 use crate::{
     auth, console,
     metrics::{Ids, MetricCounter, USAGE_METRICS},
+    proxy::{NUM_DB_CONNECTIONS_CLOSED_COUNTER, NUM_DB_CONNECTIONS_OPENED_COUNTER},
 };
 use crate::{compute, config};
 
@@ -418,36 +419,42 @@ async fn connect_to_compute_once(
     };
 
     tokio::spawn(
-        poll_fn(move |cx| {
-            if matches!(rx.has_changed(), Ok(true)) {
-                session = *rx.borrow_and_update();
-                info!(%session, "changed session");
+        async move {
+            NUM_DB_CONNECTIONS_OPENED_COUNTER.with_label_values(&["http"]).inc();
+            scopeguard::defer! {
+                NUM_DB_CONNECTIONS_CLOSED_COUNTER.with_label_values(&["http"]).inc();
             }
+            poll_fn(move |cx| {
+                if matches!(rx.has_changed(), Ok(true)) {
+                    session = *rx.borrow_and_update();
+                    info!(%session, "changed session");
+                }
 
-            loop {
-                let message = ready!(connection.poll_message(cx));
+                loop {
+                    let message = ready!(connection.poll_message(cx));
 
-                match message {
-                    Some(Ok(AsyncMessage::Notice(notice))) => {
-                        info!(%session, "notice: {}", notice);
-                    }
-                    Some(Ok(AsyncMessage::Notification(notif))) => {
-                        warn!(%session, pid = notif.process_id(), channel = notif.channel(), "notification received");
-                    }
-                    Some(Ok(_)) => {
-                        warn!(%session, "unknown message");
-                    }
-                    Some(Err(e)) => {
-                        error!(%session, "connection error: {}", e);
-                        return Poll::Ready(())
-                    }
-                    None => {
-                        info!("connection closed");
-                        return Poll::Ready(())
+                    match message {
+                        Some(Ok(AsyncMessage::Notice(notice))) => {
+                            info!(%session, "notice: {}", notice);
+                        }
+                        Some(Ok(AsyncMessage::Notification(notif))) => {
+                            warn!(%session, pid = notif.process_id(), channel = notif.channel(), "notification received");
+                        }
+                        Some(Ok(_)) => {
+                            warn!(%session, "unknown message");
+                        }
+                        Some(Err(e)) => {
+                            error!(%session, "connection error: {}", e);
+                            return Poll::Ready(())
+                        }
+                        None => {
+                            info!("connection closed");
+                            return Poll::Ready(())
+                        }
                     }
                 }
-            }
-        })
+            }).await
+        }
         .instrument(span)
     );
 

proxy/src/http/sql_over_http.rs

@@ -24,6 +24,9 @@ use url::Url;
 use utils::http::error::ApiError;
 use utils::http::json::json_response;
 
+use crate::config::HttpConfig;
+use crate::proxy::{NUM_CONNECTIONS_ACCEPTED_COUNTER, NUM_CONNECTIONS_CLOSED_COUNTER};
+
 use super::conn_pool::ConnInfo;
 use super::conn_pool::GlobalConnPool;
 
@@ -99,9 +102,9 @@ fn json_array_to_pg_array(value: &Value) -> Result<Option<String>, serde_json::E
         // convert to text with escaping
         Value::Bool(_) => serde_json::to_string(value).map(Some),
         Value::Number(_) => serde_json::to_string(value).map(Some),
-        Value::Object(_) => serde_json::to_string(value).map(Some),
 
         // here string needs to be escaped, as it is part of the array
+        Value::Object(_) => json_array_to_pg_array(&Value::String(serde_json::to_string(value)?)),
         Value::String(_) => serde_json::to_string(value).map(Some),
 
         // recurse into array
@@ -188,28 +191,46 @@ pub async fn handle(
     sni_hostname: Option<String>,
     conn_pool: Arc<GlobalConnPool>,
     session_id: uuid::Uuid,
+    config: &'static HttpConfig,
 ) -> Result<Response<Body>, ApiError> {
-    let result = handle_inner(request, sni_hostname, conn_pool, session_id).await;
-
+    let result = tokio::time::timeout(
+        config.sql_over_http_timeout,
+        handle_inner(request, sni_hostname, conn_pool, session_id),
+    )
+    .await;
     let mut response = match result {
-        Ok(r) => r,
-        Err(e) => {
-            let message = format!("{:?}", e);
-            let code = match e.downcast_ref::<tokio_postgres::Error>() {
-                Some(e) => match e.code() {
-                    Some(e) => serde_json::to_value(e.code()).unwrap(),
+        Ok(r) => match r {
+            Ok(r) => r,
+            Err(e) => {
+                let message = format!("{:?}", e);
+                let code = e.downcast_ref::<tokio_postgres::Error>().and_then(|e| {
+                    e.code()
+                        .map(|s| serde_json::to_value(s.code()).unwrap_or_default())
+                });
+                let code = match code {
+                    Some(c) => c,
                     None => Value::Null,
-                },
-                None => Value::Null,
-            };
-            error!(
-                ?code,
-                "sql-over-http per-client task finished with an error: {e:#}"
+                };
+                error!(
+                    ?code,
+                    "sql-over-http per-client task finished with an error: {e:#}"
+                );
+                // TODO: this shouldn't always be bad request.
+                json_response(
+                    StatusCode::BAD_REQUEST,
+                    json!({ "message": message, "code": code }),
+                )?
+            }
+        },
+        Err(_) => {
+            let message = format!(
+                "HTTP-Connection timed out, execution time exeeded {} seconds",
+                config.sql_over_http_timeout.as_secs()
             );
-            // TODO: this shouldn't always be bad request.
+            error!(message);
             json_response(
-                StatusCode::BAD_REQUEST,
-                json!({ "message": message, "code": code }),
+                StatusCode::GATEWAY_TIMEOUT,
+                json!({ "message": message, "code": StatusCode::GATEWAY_TIMEOUT.as_u16() }),
             )?
         }
     };
@@ -227,6 +248,13 @@ async fn handle_inner(
     conn_pool: Arc<GlobalConnPool>,
     session_id: uuid::Uuid,
 ) -> anyhow::Result<Response<Body>> {
+    NUM_CONNECTIONS_ACCEPTED_COUNTER
+        .with_label_values(&["http"])
+        .inc();
+    scopeguard::defer! {
+        NUM_CONNECTIONS_CLOSED_COUNTER.with_label_values(&["http"]).inc();
+    }
+
     //
     // Determine the destination and connection params
     //
@@ -585,7 +613,7 @@ fn _pg_array_parse(
                     }
                 }
             }
-            '}' => {
+            '}' if !quote => {
                 level -= 1;
                 if level == 0 {
                     push_checked(&mut entry, &mut entries, elem_type)?;
@@ -669,6 +697,14 @@ mod tests {
                 "{{true,false},{NULL,42},{\"foo\",\"bar\\\"-\\\\\"}}".to_owned()
             )]
         );
+        // array of objects
+        let json = r#"[{"foo": 1},{"bar": 2}]"#;
+        let json: Value = serde_json::from_str(json).unwrap();
+        let pg_params = json_to_pg_text(vec![json]).unwrap();
+        assert_eq!(
+            pg_params,
+            vec![Some(r#"{"{\"foo\":1}","{\"bar\":2}"}"#.to_owned())]
+        );
     }
 
     #[test]
@@ -796,4 +832,23 @@ mod tests {
             json!([[[1, 2, 3], [4, 5, 6]]])
         );
     }
+    #[test]
+    fn test_pg_array_parse_json() {
+        fn pt(pg_arr: &str) -> Value {
+            pg_array_parse(pg_arr, &Type::JSONB).unwrap()
+        }
+        assert_eq!(pt(r#"{"{}"}"#), json!([{}]));
+        assert_eq!(
+            pt(r#"{"{\"foo\": 1, \"bar\": 2}"}"#),
+            json!([{"foo": 1, "bar": 2}])
+        );
+        assert_eq!(
+            pt(r#"{"{\"foo\": 1}", "{\"bar\": 2}"}"#),
+            json!([{"foo": 1}, {"bar": 2}])
+        );
+        assert_eq!(
+            pt(r#"{{"{\"foo\": 1}", "{\"bar\": 2}"}}"#),
+            json!([[{"foo": 1}, {"bar": 2}]])
+        );
+    }
 }

proxy/src/http/websocket.rs

@@ -3,7 +3,10 @@ use crate::{
     config::ProxyConfig,
     error::io_error,
     protocol2::{ProxyProtocolAccept, WithClientIp},
-    proxy::{handle_client, ClientMode},
+    proxy::{
+        handle_client, ClientMode, NUM_CLIENT_CONNECTION_CLOSED_COUNTER,
+        NUM_CLIENT_CONNECTION_OPENED_COUNTER,
+    },
 };
 use bytes::{Buf, Bytes};
 use futures::{Sink, Stream, StreamExt};
@@ -202,7 +205,14 @@ async fn ws_handler(
     // TODO: that deserves a refactor as now this function also handles http json client besides websockets.
     // Right now I don't want to blow up sql-over-http patch with file renames and do that as a follow up instead.
     } else if request.uri().path() == "/sql" && request.method() == Method::POST {
-        sql_over_http::handle(request, sni_hostname, conn_pool, session_id).await
+        sql_over_http::handle(
+            request,
+            sni_hostname,
+            conn_pool,
+            session_id,
+            &config.http_config,
+        )
+        .await
     } else if request.uri().path() == "/sql" && request.method() == Method::OPTIONS {
         Response::builder()
             .header("Allow", "OPTIONS, POST")
@@ -275,23 +285,25 @@ pub async fn task_main(
             let conn_pool = conn_pool.clone();
 
             async move {
-                Ok::<_, Infallible>(hyper::service::service_fn(move |req: Request<Body>| {
-                    let sni_name = sni_name.clone();
-                    let conn_pool = conn_pool.clone();
+                Ok::<_, Infallible>(MetricService::new(hyper::service::service_fn(
+                    move |req: Request<Body>| {
+                        let sni_name = sni_name.clone();
+                        let conn_pool = conn_pool.clone();
 
-                    async move {
-                        let cancel_map = Arc::new(CancelMap::default());
-                        let session_id = uuid::Uuid::new_v4();
+                        async move {
+                            let cancel_map = Arc::new(CancelMap::default());
+                            let session_id = uuid::Uuid::new_v4();
 
-                        ws_handler(req, config, conn_pool, cancel_map, session_id, sni_name)
-                            .instrument(info_span!(
-                                "ws-client",
-                                session = %session_id,
-                                %peer_addr,
-                            ))
-                            .await
-                    }
-                }))
+                            ws_handler(req, config, conn_pool, cancel_map, session_id, sni_name)
+                                .instrument(info_span!(
+                                    "ws-client",
+                                    session = %session_id,
+                                    %peer_addr,
+                                ))
+                                .await
+                        }
+                    },
+                )))
             }
         },
     );
@@ -303,3 +315,41 @@ pub async fn task_main(
 
     Ok(())
 }
+
+struct MetricService<S> {
+    inner: S,
+}
+
+impl<S> MetricService<S> {
+    fn new(inner: S) -> MetricService<S> {
+        NUM_CLIENT_CONNECTION_OPENED_COUNTER
+            .with_label_values(&["http"])
+            .inc();
+        MetricService { inner }
+    }
+}
+
+impl<S> Drop for MetricService<S> {
+    fn drop(&mut self) {
+        NUM_CLIENT_CONNECTION_CLOSED_COUNTER
+            .with_label_values(&["http"])
+            .inc();
+    }
+}
+
+impl<S, ReqBody> hyper::service::Service<Request<ReqBody>> for MetricService<S>
+where
+    S: hyper::service::Service<Request<ReqBody>>,
+{
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future = S::Future;
+
+    fn poll_ready(&mut self, cx: &mut std::task::Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, req: Request<ReqBody>) -> Self::Future {
+        self.inner.call(req)
+    }
+}

proxy/src/proxy.rs

@@ -7,6 +7,7 @@ use crate::{
     compute::{self, PostgresConnection},
     config::{ProxyConfig, TlsConfig},
     console::{self, errors::WakeComputeError, messages::MetricsAuxInfo, Api},
+    http::StatusCode,
     metrics::{Ids, USAGE_METRICS},
     protocol2::WithClientIp,
     stream::{PqStream, Stream},
@@ -38,19 +39,55 @@ const RETRY_WAIT_EXPONENT_BASE: f64 = std::f64::consts::SQRT_2;
 const ERR_INSECURE_CONNECTION: &str = "connection is insecure (try using `sslmode=require`)";
 const ERR_PROTO_VIOLATION: &str = "protocol violation";
 
-static NUM_CONNECTIONS_ACCEPTED_COUNTER: Lazy<IntCounterVec> = Lazy::new(|| {
+pub static NUM_DB_CONNECTIONS_OPENED_COUNTER: Lazy<IntCounterVec> = Lazy::new(|| {
     register_int_counter_vec!(
-        "proxy_accepted_connections_total",
-        "Number of TCP client connections accepted.",
+        "proxy_opened_db_connections_total",
+        "Number of opened connections to a database.",
         &["protocol"],
     )
     .unwrap()
 });
 
-static NUM_CONNECTIONS_CLOSED_COUNTER: Lazy<IntCounterVec> = Lazy::new(|| {
+pub static NUM_DB_CONNECTIONS_CLOSED_COUNTER: Lazy<IntCounterVec> = Lazy::new(|| {
+    register_int_counter_vec!(
+        "proxy_closed_db_connections_total",
+        "Number of closed connections to a database.",
+        &["protocol"],
+    )
+    .unwrap()
+});
+
+pub static NUM_CLIENT_CONNECTION_OPENED_COUNTER: Lazy<IntCounterVec> = Lazy::new(|| {
+    register_int_counter_vec!(
+        "proxy_opened_client_connections_total",
+        "Number of opened connections from a client.",
+        &["protocol"],
+    )
+    .unwrap()
+});
+
+pub static NUM_CLIENT_CONNECTION_CLOSED_COUNTER: Lazy<IntCounterVec> = Lazy::new(|| {
+    register_int_counter_vec!(
+        "proxy_closed_client_connections_total",
+        "Number of closed connections from a client.",
+        &["protocol"],
+    )
+    .unwrap()
+});
+
+pub static NUM_CONNECTIONS_ACCEPTED_COUNTER: Lazy<IntCounterVec> = Lazy::new(|| {
+    register_int_counter_vec!(
+        "proxy_accepted_connections_total",
+        "Number of client connections accepted.",
+        &["protocol"],
+    )
+    .unwrap()
+});
+
+pub static NUM_CONNECTIONS_CLOSED_COUNTER: Lazy<IntCounterVec> = Lazy::new(|| {
     register_int_counter_vec!(
         "proxy_closed_connections_total",
-        "Number of TCP client connections closed.",
+        "Number of client connections closed.",
         &["protocol"],
     )
     .unwrap()
@@ -75,6 +112,15 @@ static NUM_CONNECTION_FAILURES: Lazy<IntCounterVec> = Lazy::new(|| {
     .unwrap()
 });
 
+static NUM_WAKEUP_FAILURES: Lazy<IntCounterVec> = Lazy::new(|| {
+    register_int_counter_vec!(
+        "proxy_connection_failures_breakdown",
+        "Number of wake-up failures (per kind).",
+        &["retry", "kind"],
+    )
+    .unwrap()
+});
+
 static NUM_BYTES_PROXIED_COUNTER: Lazy<IntCounterVec> = Lazy::new(|| {
     register_int_counter_vec!(
         "proxy_io_bytes_per_client",
@@ -208,12 +254,16 @@ pub async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
         "handling interactive connection from client"
     );
 
-    // The `closed` counter will increase when this future is destroyed.
+    let proto = mode.protocol_label();
+    NUM_CLIENT_CONNECTION_OPENED_COUNTER
+        .with_label_values(&[proto])
+        .inc();
     NUM_CONNECTIONS_ACCEPTED_COUNTER
-        .with_label_values(&[mode.protocol_label()])
+        .with_label_values(&[proto])
         .inc();
     scopeguard::defer! {
-        NUM_CONNECTIONS_CLOSED_COUNTER.with_label_values(&[mode.protocol_label()]).inc();
+        NUM_CLIENT_CONNECTION_CLOSED_COUNTER.with_label_values(&[proto]).inc();
+        NUM_CONNECTIONS_CLOSED_COUNTER.with_label_values(&[proto]).inc();
     }
 
     let tls = config.tls_config.as_ref();
@@ -248,7 +298,7 @@ pub async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
         mode.allow_self_signed_compute(config),
     );
     cancel_map
-        .with_session(|session| client.connect_to_db(session, mode.allow_cleartext()))
+        .with_session(|session| client.connect_to_db(session, mode))
         .await
 }
 
@@ -397,6 +447,46 @@ impl ConnectMechanism for TcpMechanism<'_> {
     }
 }
 
+const fn bool_to_str(x: bool) -> &'static str {
+    if x {
+        "true"
+    } else {
+        "false"
+    }
+}
+
+fn report_error(e: &WakeComputeError, retry: bool) {
+    use crate::console::errors::ApiError;
+    let retry = bool_to_str(retry);
+    let kind = match e {
+        WakeComputeError::BadComputeAddress(_) => "bad_compute_address",
+        WakeComputeError::ApiError(ApiError::Transport(_)) => "api_transport_error",
+        WakeComputeError::ApiError(ApiError::Console {
+            status: StatusCode::LOCKED,
+            ref text,
+        }) if text.contains("written data quota exceeded")
+            || text.contains("the limit for current plan reached") =>
+        {
+            "quota_exceeded"
+        }
+        WakeComputeError::ApiError(ApiError::Console {
+            status: StatusCode::LOCKED,
+            ..
+        }) => "api_console_locked",
+        WakeComputeError::ApiError(ApiError::Console {
+            status: StatusCode::BAD_REQUEST,
+            ..
+        }) => "api_console_bad_request",
+        WakeComputeError::ApiError(ApiError::Console { status, .. })
+            if status.is_server_error() =>
+        {
+            "api_console_other_server_error"
+        }
+        WakeComputeError::ApiError(ApiError::Console { .. }) => "api_console_other_error",
+    };
+    NUM_WAKEUP_FAILURES.with_label_values(&[retry, kind]).inc();
+}
+
 /// Try to connect to the compute node, retrying if necessary.
 /// This function might update `node_info`, so we take it by `&mut`.
 #[tracing::instrument(skip_all)]
@@ -440,10 +530,12 @@ where
         match handle_try_wake(wake_res, num_retries) {
             Err(e) => {
                 error!(error = ?e, num_retries, retriable = false, "couldn't wake compute node");
+                report_error(&e, false);
                 return Err(e.into());
             }
             // failed to wake up but we can continue to retry
             Ok(ControlFlow::Continue(e)) => {
+                report_error(&e, true);
                 warn!(error = ?e, num_retries, retriable = true, "couldn't wake compute node");
             }
             // successfully woke up a compute node and can break the wakeup loop
@@ -682,7 +774,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin> Client<'_, S> {
     async fn connect_to_db(
         self,
         session: cancellation::Session<'_>,
-        allow_cleartext: bool,
+        mode: ClientMode,
     ) -> anyhow::Result<()> {
         let Self {
             mut stream,
@@ -698,7 +790,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin> Client<'_, S> {
         };
 
         let auth_result = match creds
-            .authenticate(&extra, &mut stream, allow_cleartext)
+            .authenticate(&extra, &mut stream, mode.allow_cleartext())
             .await
         {
             Ok(auth_result) => auth_result,
@@ -724,6 +816,14 @@ impl<S: AsyncRead + AsyncWrite + Unpin> Client<'_, S> {
             .or_else(|e| stream.throw_error(e))
             .await?;
 
+        let proto = mode.protocol_label();
+        NUM_DB_CONNECTIONS_OPENED_COUNTER
+            .with_label_values(&[proto])
+            .inc();
+        scopeguard::defer! {
+            NUM_DB_CONNECTIONS_CLOSED_COUNTER.with_label_values(&[proto]).inc();
+        }
+
         prepare_client_connection(&node, reported_auth_ok, session, &mut stream).await?;
         // Before proxy passing, forward to compute whatever data is left in the
         // PqStream input buffer. Normally there is none, but our serverless npm

s3_scrubber/Cargo.toml

@@ -33,13 +33,9 @@ reqwest = { workspace = true, default-features = false, features = ["rustls-tls"
 aws-config = { workspace = true, default-features = false, features = ["rustls", "credentials-sso"] }
 
 pageserver = { path = "../pageserver" }
-remote_storage = { path = "../libs/remote_storage" }
 
 tracing.workspace = true
-tracing-subscriber = { version = "0.3.17", features = ["ansi"] }
+tracing-subscriber.workspace = true
 clap.workspace = true
 tracing-appender = "0.2"
 histogram = "0.7"
-humantime.workspace = true
-camino.workspace = true
-aws-smithy-types = "0.56.1"

s3_scrubber/src/lib.rs

@@ -4,6 +4,7 @@ pub mod delete_batch_producer;
 pub mod metadata_stream;
 mod s3_deletion;
 pub mod scan_metadata;
+
 use std::env;
 use std::fmt::Display;
 use std::time::Duration;

s3_scrubber/src/main.rs

@@ -16,8 +16,6 @@ use tracing::{info, warn};
 
 use clap::{Parser, Subcommand, ValueEnum};
 
-mod restore_tenant_from_object_versioning;
-
 #[derive(Parser)]
 #[command(author, version, about, long_about = None)]
 #[command(arg_required_else_help(true))]
@@ -61,9 +59,6 @@ enum Command {
         skip_validation: bool,
     },
     ScanMetadata {},
-    RestoreTenantFromObjectVersioningMostRecentIndexPart(
-        restore_tenant_from_object_versioning::Command,
-    ),
 }
 
 async fn tidy(
@@ -252,14 +247,5 @@ async fn main() -> anyhow::Result<()> {
                 }
             }
         },
-        Command::RestoreTenantFromObjectVersioningMostRecentIndexPart(arg) => {
-            match restore_tenant_from_object_versioning::doit(arg).await {
-                Err(e) => {
-                    tracing::error!("Failed: {e:?}");
-                    Err(e)
-                }
-                Ok(()) => Ok(()),
-            }
-        }
     }
 }

s3_scrubber/src/restore_tenant_from_object_versioning.rs

@@ -1,513 +0,0 @@
-//! Restore pageserver state from S3 object versioning.
-//!
-//! This sub-cmmand allows restoring a tenant's pageserver S3 state from S3 object versioning.
-//!
-//! # Instructions
-//!
-//! - Run command
-//!   ```
-//!   SSO_ACCOUNT_ID=... REGION=... \
-//!   BUCKET=neon-{prod,staging}-storage-... \
-//!     cargo run -p s3_scrubber \
-//!         restore-tenant-from-object-versioning-most-recent-index-part \
-//!         TENANT_TO_RESTORE \
-//!         ./restore
-//!         timeline-list TIMELINE_TO_RESTORE TIMELINE_TO_RESTORE ...
-//!   ```
-//! - `./restore` now contains the timeline state referenced by the latest `index_part.json`s of the
-//!   specified timelines in the `timeline-list`` argument
-//! - Use `cargo neon` to start a pageserver
-//!     - `rm -rf .neon`
-//!     - `cargo neon init`
-//!     - `sed -i 's/\(.*control_plane_api.*\)/#\1/' .neon/config`
-//!     - `sed -i 's/\(.*control_plane_api.*\)/#\1/' .neon/pageserver_1/pageserver.toml`
-//!     - configure the pageserver remote storage config to point to the restore directory.
-//!       Use your text editor to edit the TOML file: `.neon/pageserver_1/pageserver.toml`.
-//!       ```
-//!       [remote_storage]
-//!       local_path = "/path/to/restore/pageserver/v1"
-//!       ````
-//!    - `cargo neon start`
-//!    - make sure attaching the tenant works
-//!      - `curl -X POST localhost:9898/v1/tenant/TENANT_TO_RESTORE/attach`
-//!      - check `curl -X GET localhost:9898/v1/tenant/TENANT_TO_RESTORE | jq`
-//!     - for each timeline $timeline_id to restore:
-//!         - `cargo neon mappings map --branch-name restore-$timeline_id --tenant-id TENANT_TO_RESTORE --timeline-id $timeline_id`
-//!         - `cargo neon endpoint create --tenant-id TENANT_TO_RESTORE --branch-name restore-$timeline_id ep-restore-$timeline_id`
-//!         - `cargo neon endpoint start --tenant-id TENANT_TO_RESTORE ep-restore-$timeline_id`
-//!             - it prints a connection string, looking like `postgresql://cloud_admin@127.0.0.1:PORT/DB`
-//!         - dump database contents using postgres tools
-//!             - determine PG version `$restore_pg_version` using
-//!               ```
-//!               curl -s -X GET localhost:9898/v1/tenant/TENANT_TO_RESTORE/timeline/$timeline_id | jq .pg_version
-//!               ```
-//!             - pg_dumpall
-//!               ```
-//!               ./pg_install/$restore_pg_version/bin/pg_dumpall -d THE_CONNECTION_STRING/postgres  > ./restore/pg_dumpall.out
-//!               ```
-//!            - pg_dump a specific database
-//!              ```
-//!              ./pg_install/v15/bin/pg_dump -d 'THE_CONNECTION_STRING/THEDBTODUMP' > ./restore/pg_dump_THEDBTODUMP.out
-//!              ```
-//!         - `cargo neon endpoint stop --tenant-id TENANT_TO_RESTORE restore-$timeline_id`
-//!
-//! - Use the pg_dump files to restore the database into a new Neon project.
-//!
-//! # Limitations & Future Work
-//!
-//! Just restoring Pageserver S3 state restores a consistent state at an LSN that is NOT THE LAST COMMIT LSN.
-//! The reason is that Pageserver uploads layers to S3 with implementation-specific delays that are optimized for day-to-day operation.
-//!
-//! If we still had the Safekeeper WAL, we could restore the Safekeeper S3 state in a similar way.
-//! In that case, we wouldn't need the `pg_dump` step.
-//! We would simply attach the tenant to safekeepers and pageservers.
-//! When attaching to Safekeeper, we would need to tell it that PS remote_consistent_lsn is the restore-point-LSN,
-//! i.e., the LSNs in the restored index_part.json's in Pageserver S3 state.
-//! Pageserver attach would pick up the restored state from S3, and the Safekeeper & Pageserver would
-//! resume normal operation as if the clock had been wound back to restore-point-LSN.
-
-use std::{
-    collections::{HashMap, HashSet},
-    num::{NonZeroU32, NonZeroUsize},
-    sync::Arc,
-};
-
-use anyhow::Context;
-
-use aws_sdk_s3::operation::list_object_versions::ListObjectVersionsOutput;
-use aws_types::region::Region;
-use camino::Utf8PathBuf;
-use pageserver::tenant::{
-    remote_timeline_client::index::LayerFileMetadata, TENANTS_SEGMENT_NAME,
-    TENANT_DELETED_MARKER_FILE_NAME, TIMELINES_SEGMENT_NAME,
-};
-use remote_storage::{GenericRemoteStorage, RemotePath, RemoteStorageConfig, RemoteStorageKind};
-use s3_scrubber::{init_logging, init_s3_client, BucketConfig};
-use tokio::io::AsyncReadExt;
-use tracing::{debug, info, info_span, Instrument};
-use utils::id::{TenantId, TimelineId};
-
-#[derive(Debug, Clone, clap::Subcommand)]
-enum ResurrectTimelines {
-    TimelineList { timeline_ids: Vec<TimelineId> },
-    // AllTimelinesDeletedAfter { timestamp: humantime::Timestamp },
-}
-
-#[derive(clap::Args)]
-pub(crate) struct Command {
-    tenant_id: TenantId,
-    dest_dir: Utf8PathBuf,
-    #[clap(short, long)]
-    dry_run: bool,
-    #[clap(subcommand)]
-    timelines: ResurrectTimelines,
-}
-
-pub(crate) async fn doit(args: Command) -> anyhow::Result<()> {
-    let _logging_guard = {
-        let log_prefix = format!("restore_tenant_from_object_versioning_{}", args.tenant_id);
-        let dry_suffix = if args.dry_run { "__dry" } else { "" };
-        let file_name = {
-            format!(
-                "{}_{}{}.log",
-                log_prefix,
-                chrono::Utc::now().format("%Y_%m_%d__%H_%M_%S"),
-                dry_suffix,
-            )
-        };
-        init_logging(&file_name)
-    };
-
-    let restore_dst = if tokio::fs::try_exists(&args.dest_dir).await? {
-        anyhow::bail!("destination directory already exists: {}", args.dest_dir,);
-    } else {
-        GenericRemoteStorage::from_config(&RemoteStorageConfig {
-            max_concurrent_syncs: NonZeroUsize::new(100).unwrap(),
-            max_sync_errors: NonZeroU32::new(1).unwrap(), // ???? would want so specify 0
-            storage: RemoteStorageKind::LocalFs(args.dest_dir.clone()),
-        })
-        .context("instantiate restore destination")?
-    };
-
-    let bucket_config = BucketConfig::from_env()?;
-
-    let bucket_region = Region::new(bucket_config.region);
-    let delimiter = "/".to_string();
-    let s3_client = Arc::new(init_s3_client(bucket_config.sso_account_id, bucket_region));
-    let tenant_root = [
-        "pageserver",
-        "v1",
-        TENANTS_SEGMENT_NAME,
-        &args.tenant_id.to_string(),
-    ]
-    .join(&delimiter);
-
-    let tenant_delete_marker = [&tenant_root, TENANT_DELETED_MARKER_FILE_NAME].join(&delimiter);
-
-    // - Ensure the prefix is empty when ignoring existence of versions.
-    // - Ensure the tenant delete marker key is part of `DeleteMarkers`. If it isn't, the tenant hasn't finished deletion yet and we should let pageservers complete it first.
-    // - Restore each index_part.json based on the version in DeleteMarkers as well as the layers it references. For the layers, also use the version in DeleteMarkers and ensure it is the latest.
-    // - Remove the deleted_at mark for the specified timelines.
-    //
-    // Notes:
-    //   - The restore will happen in-place because it's hard to change tenant/timeline ids.
-    //   - The restore could be interrupted mid-way.
-    //   - Hence, separate plan-making and plan-execution.
-
-    async {
-        info!("send request");
-        let res = s3_client
-            .list_objects_v2()
-            .bucket(&bucket_config.bucket)
-            .prefix(&tenant_root)
-            .send()
-            .await?;
-
-        info!(response=?res, "got response");
-
-        if res.key_count() > 0 {
-            anyhow::bail!("tenant prefix is not empty in S3");
-        }
-        if res.is_truncated() {
-            unimplemented!("can this even happen")
-        }
-        Ok(())
-    }
-    .instrument(info_span!("ensure prefix empty"))
-    .await
-    .context("ensure prefix is empty")?;
-
-    async {
-        info!("send request");
-        let res = s3_client
-            .list_object_versions()
-            .bucket(&bucket_config.bucket)
-            .prefix(&tenant_delete_marker)
-            .send()
-            .await?;
-        debug!(response=?res, "got response");
-
-        if res.is_truncated() {
-            unimplemented!("can this even happen")
-        }
-
-        let markers = res
-            .delete_markers()
-            .context("expected delete marker in response")?;
-
-        if markers.len() != 1 {
-            anyhow::bail!("expected exactly one delete marker because we create and delete the marker exactly once, got {}", markers.len());
-        }
-
-        if !markers[0].is_latest() {
-            anyhow::bail!("expected delete marker to have IsLatest set: {:?}", markers[0]);
-        }
-
-        Ok(())
-    }
-    .instrument(info_span!(
-        "ensure tenant delete marker exists in DeleteMarkers",
-        tenant_delete_marker,
-    ))
-    .await
-    .context("ensure tenant delete marker exists in DeleteMarkers")?;
-
-    let timelines = match args.timelines {
-        ResurrectTimelines::TimelineList { timeline_ids } => timeline_ids,
-    };
-
-    // Fetch all the information we need to execute the restore.
-    let version_responses_by_timeline = async {
-        let mut out: HashMap<TimelineId, Vec<Arc<ListObjectVersionsOutput>>> = Default::default();
-        for tl in &timelines {
-            async {
-                let timeline_prefix = [tenant_root.as_str(), TIMELINES_SEGMENT_NAME , &tl.to_string()].join(&delimiter);
-                let mut next_key_marker = None;
-                let mut next_version_id_marker = None;
-                loop {
-                    info!("sending request");
-                    let res: ListObjectVersionsOutput = s3_client.list_object_versions()
-                        .bucket(&bucket_config.bucket)
-                        .prefix(&timeline_prefix)
-                        .set_key_marker(next_key_marker.take())
-                        .set_version_id_marker(next_version_id_marker.take())
-                        .send()
-                        .await?;
-
-                    let res = Arc::new(res);
-                    out.entry(*tl).or_default().push(Arc::clone(&res));
-
-                    info!("got response");
-                    match res.versions() {
-                        Some(versions) => {
-                            for version in versions {
-                                info!("version: {:?}", version);
-                            }
-                        }
-                        None => {
-                            info!("no versions");
-                        }
-                    }
-                    match res.delete_markers() {
-                        Some(markers) => {
-                            for marker in markers {
-                                info!("delete marker: {:?}", marker);
-                            }
-                        }
-                        None => {
-                            info!("no delete markers");
-                        }
-                    }
-
-                    if !res.is_truncated() {
-                        break;
-                    }
-                    next_key_marker = res.next_key_marker().map(|s| s.to_string());
-                    next_version_id_marker = res.next_version_id_marker().map(|s| s.to_string());
-                    if let (None, None) = (&next_key_marker, &next_version_id_marker) {
-                        anyhow::bail!("s3 returned is_truncated=true but neither next_key_marker nor next_version_id_marker are set");
-                    }
-                }
-                Ok(())
-            }.instrument(info_span!("timeline", timeline_id=%tl)).await?;
-        }
-        anyhow::Ok(out)
-    }.instrument(info_span!("list all object versions and delete markers")).await?;
-    #[derive(Debug)]
-    struct LatestVersion {
-        key: String,
-        last_modified: aws_smithy_types::DateTime,
-        version_id: String,
-    }
-    let find_latest_version_based_on_delete_marker_last_modified = |tl: &TimelineId, key: &str| {
-        let restore_version_delete_marker = {
-            let mut candidates = Vec::new();
-            for res in &version_responses_by_timeline[tl] {
-                let Some(markers) = res.delete_markers() else {
-                    continue;
-                };
-                for marker in markers {
-                    if !marker.is_latest() {
-                        continue;
-                    }
-                    if marker.key().unwrap() != key {
-                        continue;
-                    }
-                    candidates.push(LatestVersion {
-                        key: marker.key().unwrap().to_owned(),
-                        last_modified: marker.last_modified().unwrap().clone(),
-                        version_id: marker.version_id().unwrap().to_owned(),
-                    });
-                }
-            }
-            info!(?candidates, "marker candidates");
-            if candidates.len() != 1 {
-                anyhow::bail!("expected exactly one IsLatest, got {}", candidates.len());
-            }
-            candidates.pop().unwrap()
-        };
-        info!(?restore_version_delete_marker, "found marker");
-
-        // There's no way to get the latest version from the delete marker.
-        // But, we observe (can't find written guarantee) that the Delete Marker's "Last Modified" is >= the latest version.
-        // So, find latest version based on that.
-        let restore_version = {
-            let mut candidates = Vec::new();
-            for res in &version_responses_by_timeline[tl] {
-                let Some(versions) = res.versions() else {
-                    continue;
-                };
-                for version in versions {
-                    if version.key().unwrap() != restore_version_delete_marker.key {
-                        continue;
-                    }
-                    candidates.push(LatestVersion {
-                        key: version.key().unwrap().to_owned(),
-                        last_modified: version.last_modified().unwrap().clone(),
-                        version_id: version.version_id().unwrap().to_owned(),
-                    });
-                }
-            }
-            candidates.sort_by_key(|v| v.last_modified.clone());
-            info!(?candidates, "version candidates");
-            if candidates.is_empty() {
-                anyhow::bail!(
-                    "expected at least one version matching the delete marker's key, got none"
-                );
-            }
-            {
-                let mut uniq = HashSet::new();
-                for v in &candidates {
-                    if !uniq.insert(v.last_modified.clone()) {
-                        anyhow::bail!("last_modified timestamps are not unique, don't know which version to pick");
-                    }
-                }
-            }
-            candidates.pop().unwrap() // we sorted ascending, so, pop() is the latest
-        };
-        anyhow::Ok(restore_version)
-    };
-
-    let latest_index_part_versions: HashMap<TimelineId, LatestVersion> = {
-        let span = info_span!("find index part version");
-        let _enter = span.enter();
-
-        // The latest index part for a deleted tenant is always a DeletedMarker
-        let mut out = HashMap::new();
-        for tl in &timelines {
-            let span = info_span!("timeline", timeline_id=%tl);
-            let _enter = span.enter();
-            let restore_version = find_latest_version_based_on_delete_marker_last_modified(
-                tl,
-                // TODO: support generation numbers
-                &[
-                    &tenant_root,
-                    TIMELINES_SEGMENT_NAME,
-                    &tl.to_string(),
-                    pageserver::tenant::IndexPart::FILE_NAME,
-                ]
-                .join(&delimiter),
-            )?;
-            out.insert(*tl, restore_version);
-        }
-        out
-    };
-
-    let index_part_contents: HashMap<TimelineId, pageserver::tenant::IndexPart> = async {
-        let mut out = HashMap::new();
-
-        for tl in &timelines {
-            async {
-                let v = &latest_index_part_versions[tl];
-
-                let mut body_buf = Vec::new();
-                info!("send request");
-                let res = s3_client
-                    .get_object()
-                    .bucket(&bucket_config.bucket)
-                    .key(&v.key)
-                    .version_id(&v.version_id)
-                    .send()
-                    .await?;
-                info!(?res, "got response header");
-
-                res.body
-                    .into_async_read()
-                    .read_to_end(&mut body_buf)
-                    .await?;
-
-                let body_buf = String::from_utf8(body_buf)?;
-                info!(body_buf, "received response body");
-
-                let mut index_part: pageserver::tenant::IndexPart =
-                    serde_json::from_str(&body_buf)?;
-                info!(?index_part, "parsed index part");
-
-                let deleted_at = index_part.deleted_at.take();
-                info!(
-                    ?deleted_at,
-                    "removing deleted_at field from index part, previous value logged here"
-                );
-
-                let updated_buf = serde_json::to_vec(&index_part)?;
-                let updated_buf_len = updated_buf.len();
-                info!("uploading modified index part to restore_dst");
-                restore_dst
-                    .upload(
-                        std::io::Cursor::new(updated_buf),
-                        updated_buf_len,
-                        &RemotePath::from_string(&v.key).unwrap(),
-                        None,
-                    )
-                    .await
-                    .context("upload modified index part to restore_dst")?;
-
-                out.insert(*tl, index_part);
-
-                anyhow::Ok(())
-            }
-            .instrument(info_span!("timeline", timeline_id=%tl))
-            .await?;
-        }
-
-        anyhow::Ok(out)
-    }
-    .instrument(info_span!("get index part contents"))
-    .await
-    .context("get index part contents")?;
-
-    async {
-        for (tl, index_part) in &index_part_contents {
-            async {
-                for (layer_file_name, layer_md) in &index_part.layer_metadata {
-                    let layer_md: LayerFileMetadata = layer_md.into();
-                    async {
-                        // TODO: support generations
-                        let layer_file_key = [
-                            &tenant_root,
-                            TIMELINES_SEGMENT_NAME,
-                            &tl.to_string(),
-                            &layer_file_name.file_name(),
-                        ]
-                        .join(&delimiter);
-
-                        // The latest index parts naturally reference the latest layers.
-                        // So, a deleted tenant's latest layers are the ones in DeleteMarkers.
-                        //
-                        // If we want to support restoring from not-latest index part, this will require more work.
-                        // The idea is to
-                        // 1. every index_part.json that we upload contains a strongly monotonically increasing sequence number
-                        // 2. every image layer that we upload is S3-metadata-tagged with the sequence number of the IndexPart
-                        //    in which it first appeared.
-                        // This allows to recover the correct layer object version, even if we have a bug that overwrites layers.
-                        let restore_version =
-                            find_latest_version_based_on_delete_marker_last_modified(
-                                tl,
-                                &layer_file_key,
-                            )?;
-
-                        // TODO: teach RemoteStorage copy operation so we can use s3_client.copy_object()
-                        async {
-                            let res = s3_client
-                                .get_object()
-                                .bucket(&bucket_config.bucket)
-                                .key(&restore_version.key)
-                                .version_id(&restore_version.version_id)
-                                .send()
-                                .await
-                                .context("get object header")?;
-                            // TODO: instead of file_size(), do actual data integrity checking.
-                            restore_dst
-                                .upload(
-                                    res.body.into_async_read(),
-                                    layer_md.file_size().try_into().unwrap(),
-                                    &RemotePath::from_string(&restore_version.key).unwrap(),
-                                    None,
-                                )
-                                .await
-                                .context("download-body-and-upload")?;
-                            anyhow::Ok(())
-                        }
-                        .instrument(info_span!("copy", layer_file_name=%layer_file_name))
-                        .await?;
-
-                        anyhow::Ok(())
-                    }
-                    .instrument(info_span!("layer", layer_file_name=%layer_file_name))
-                    .await?;
-                }
-
-                anyhow::Ok(())
-            }
-            .instrument(info_span!("timeline", timeline_id=%tl))
-            .await?;
-        }
-
-        anyhow::Ok(())
-    }
-    .instrument(info_span!("download layer files into restore_dst"))
-    .await
-    .context("download layers into restore dst")?;
-
-    Ok(())
-}

safekeeper/src/http/routes.rs

@@ -374,8 +374,12 @@ pub fn make_router(conf: SafeKeeperConf) -> RouterBuilder<hyper::Body, ApiError>
     if conf.http_auth.is_some() {
         router = router.middleware(auth_middleware(|request| {
             #[allow(clippy::mutable_key_type)]
-            static ALLOWLIST_ROUTES: Lazy<HashSet<Uri>> =
-                Lazy::new(|| ["/v1/status"].iter().map(|v| v.parse().unwrap()).collect());
+            static ALLOWLIST_ROUTES: Lazy<HashSet<Uri>> = Lazy::new(|| {
+                ["/v1/status", "/metrics"]
+                    .iter()
+                    .map(|v| v.parse().unwrap())
+                    .collect()
+            });
             if ALLOWLIST_ROUTES.contains(request.uri()) {
                 None
             } else {

test_runner/performance/test_wal_backpressure.py

@@ -65,7 +65,7 @@ def start_heavy_write_workload(env: PgCompare, n_tables: int, scale: int, num_it
 
     def start_single_table_workload(table_id: int):
         for _ in range(num_iters):
-            with env.pg.connect().cursor() as cur:
+            with env.pg.connect(options="-cstatement_timeout=300s").cursor() as cur:
                 cur.execute(
                     f"INSERT INTO t{table_id} SELECT FROM generate_series(1,{new_rows_each_update})"
                 )

test_runner/regress/test_proxy.py

@@ -188,7 +188,7 @@ def test_sql_over_http(static_proxy: NeonProxy):
             headers={"Content-Type": "application/sql", "Neon-Connection-String": connstr},
             verify=str(static_proxy.test_output_dir / "proxy.crt"),
         )
-        assert response.status_code == 200
+        assert response.status_code == 200, response.text
         return response.json()
 
     rows = q("select 42 as answer")["rows"]
@@ -206,6 +206,12 @@ def test_sql_over_http(static_proxy: NeonProxy):
     rows = q("select $1::json->'a' as answer", [{"a": {"b": 42}}])["rows"]
     assert rows == [{"answer": {"b": 42}}]
 
+    rows = q("select $1::jsonb[] as answer", [[{}]])["rows"]
+    assert rows == [{"answer": [{}]}]
+
+    rows = q("select $1::jsonb[] as answer", [[{"foo": 1}, {"bar": 2}]])["rows"]
+    assert rows == [{"answer": [{"foo": 1}, {"bar": 2}]}]
+
     rows = q("select * from pg_class limit 1")["rows"]
     assert len(rows) == 1
 

test_runner/regress/test_wal_acceptor_async.py

@@ -6,6 +6,7 @@ from pathlib import Path
 from typing import List, Optional
 
 import asyncpg
+import pytest
 import toml
 from fixtures.log_helper import getLogger
 from fixtures.neon_fixtures import Endpoint, NeonEnv, NeonEnvBuilder, Safekeeper
@@ -597,7 +598,10 @@ async def run_wal_lagging(env: NeonEnv, endpoint: Endpoint, test_output_dir: Pat
     assert res == expected_sum
 
 
-# do inserts while restarting postgres and messing with safekeeper addresses
+# Do inserts while restarting postgres and messing with safekeeper addresses.
+# The test takes more than default 5 minutes on Postgres 16,
+# see https://github.com/neondatabase/neon/issues/5305
+@pytest.mark.timeout(600)
 def test_wal_lagging(neon_env_builder: NeonEnvBuilder, test_output_dir: Path):
     neon_env_builder.num_safekeepers = 3
     env = neon_env_builder.init_start()