Fix handling of contribs that have non-default directory

Don't try to download already existing shared_preload_libraries files

.github/workflows/benchmarking.yml

@@ -180,7 +180,8 @@ jobs:
       image: 369495373322.dkr.ecr.eu-central-1.amazonaws.com/rust:pinned
       options: --init
 
-    timeout-minutes: 360 # 6h
+    # Increase timeout to 8h, default timeout is 6h
+    timeout-minutes: 480
 
     steps:
     - uses: actions/checkout@v3
@@ -321,8 +322,6 @@ jobs:
       image: 369495373322.dkr.ecr.eu-central-1.amazonaws.com/rust:pinned
       options: --init
 
-    timeout-minutes: 360 # 6h
-
     steps:
     - uses: actions/checkout@v3
 
@@ -414,8 +413,6 @@ jobs:
       image: 369495373322.dkr.ecr.eu-central-1.amazonaws.com/rust:pinned
       options: --init
 
-    timeout-minutes: 360 # 6h
-
     steps:
     - uses: actions/checkout@v3
 
@@ -501,8 +498,6 @@ jobs:
       image: 369495373322.dkr.ecr.eu-central-1.amazonaws.com/rust:pinned
       options: --init
 
-    timeout-minutes: 360 # 6h
-
     steps:
     - uses: actions/checkout@v3
 

.github/workflows/build_and_test.yml

@@ -916,7 +916,7 @@ jobs:
             exit 1
           fi
 
-      - name: Create tag "release-${{ needs.tag.outputs.build-tag }}"
+      - name: Create git tag
         if: github.ref_name == 'release'
         uses: actions/github-script@v6
         with:
@@ -926,7 +926,7 @@ jobs:
             github.rest.git.createRef({
               owner: context.repo.owner,
               repo: context.repo.repo,
-              ref: "refs/tags/release-${{ needs.tag.outputs.build-tag }}",
+              ref: "refs/tags/${{ needs.tag.outputs.build-tag }}",
               sha: context.sha,
             })
 

Dockerfile.compute-node

@@ -481,6 +481,23 @@ RUN wget https://github.com/rdkit/rdkit/archive/refs/tags/Release_2023_03_1.tar.
     make -j $(getconf _NPROCESSORS_ONLN) install && \
     echo 'trusted = true' >> /usr/local/pgsql/share/extension/rdkit.control
 
+#########################################################################################
+#
+# Layer "pg-uuidv7-pg-build"
+# compile pg_uuidv7 extension
+#
+#########################################################################################
+FROM build-deps AS pg-uuidv7-pg-build
+COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/
+
+ENV PATH "/usr/local/pgsql/bin/:$PATH"
+RUN wget https://github.com/fboulnois/pg_uuidv7/archive/refs/tags/v1.0.1.tar.gz -O pg_uuidv7.tar.gz && \
+    echo "0d0759ab01b7fb23851ecffb0bce27822e1868a4a5819bfd276101c716637a7a pg_uuidv7.tar.gz" | sha256sum --check && \
+    mkdir pg_uuidv7-src && cd pg_uuidv7-src && tar xvzf ../pg_uuidv7.tar.gz --strip-components=1 -C . && \
+    make -j $(getconf _NPROCESSORS_ONLN) && \
+    make -j $(getconf _NPROCESSORS_ONLN) install && \
+    echo 'trusted = true' >> /usr/local/pgsql/share/extension/pg_uuidv7.control
+
 #########################################################################################
 #
 # Layer "rust extensions"
@@ -614,6 +631,7 @@ COPY --from=kq-imcx-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg-cron-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg-pgx-ulid-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=rdkit-pg-build /usr/local/pgsql/ /usr/local/pgsql/
+COPY --from=pg-uuidv7-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY pgxn/ pgxn/
 
 RUN make -j $(getconf _NPROCESSORS_ONLN) \

compute_tools/src/bin/compute_ctl.rs

@@ -36,7 +36,7 @@ use std::fs::File;
 use std::panic;
 use std::path::Path;
 use std::process::exit;
-use std::sync::{mpsc, Arc, Condvar, Mutex};
+use std::sync::{mpsc, Arc, Condvar, Mutex, OnceLock};
 use std::{thread, time::Duration};
 
 use anyhow::{Context, Result};
@@ -49,14 +49,13 @@ use compute_api::responses::ComputeStatus;
 
 use compute_tools::compute::{ComputeNode, ComputeState, ParsedSpec};
 use compute_tools::configurator::launch_configurator;
-use compute_tools::extension_server::{download_extension, init_remote_storage, ExtensionType};
+use compute_tools::extension_server::{get_pg_version, init_remote_storage};
 use compute_tools::http::api::launch_http_server;
 use compute_tools::logger::*;
 use compute_tools::monitor::launch_monitor;
 use compute_tools::params::*;
 use compute_tools::spec::*;
 
-use tokio::runtime::Runtime;
 const BUILD_TAG_DEFAULT: &str = "local";
 
 fn main() -> Result<()> {
@@ -72,18 +71,16 @@ fn main() -> Result<()> {
 
     let remote_ext_config = matches.get_one::<String>("remote-ext-config");
     let ext_remote_storage = match remote_ext_config {
-        Some(x) => Some(init_remote_storage(x)?),
+        Some(x) => match init_remote_storage(x) {
+            Ok(y) => Some(y),
+            Err(e) => {
+                dbg!("Error {:?}, setting remote storage to None", e);
+                None
+            }
+        },
         None => None,
     };
 
-    let rt = Runtime::new().unwrap();
-    let copy_remote_storage = ext_remote_storage.clone();
-    rt.block_on(async move {
-        download_extension(&copy_remote_storage, ExtensionType::Shared, pgbin)
-            .await
-            .expect("download extension should work");
-    });
-
     let http_port = *matches
         .get_one::<u16>("http-port")
         .expect("http-port is required");
@@ -184,6 +181,7 @@ fn main() -> Result<()> {
 
     let mut new_state = ComputeState::new();
     let spec_set;
+
     if let Some(spec) = spec {
         let pspec = ParsedSpec::try_from(spec).map_err(|msg| anyhow::anyhow!(msg))?;
         new_state.pspec = Some(pspec);
@@ -195,10 +193,13 @@ fn main() -> Result<()> {
         connstr: Url::parse(connstr).context("cannot parse connstr as a URL")?,
         pgdata: pgdata.to_string(),
         pgbin: pgbin.to_string(),
+        pgversion: get_pg_version(pgbin),
         live_config_allowed,
         state: Mutex::new(new_state),
         state_changed: Condvar::new(),
         ext_remote_storage,
+        available_libraries: OnceLock::new(),
+        available_extensions: OnceLock::new(),
     };
     let compute = Arc::new(compute_node);
 

compute_tools/src/compute.rs

@@ -1,13 +1,15 @@
+use std::collections::HashMap;
 use std::fs;
 use std::os::unix::fs::PermissionsExt;
 use std::path::Path;
 use std::process::{Command, Stdio};
 use std::str::FromStr;
-use std::sync::{Condvar, Mutex};
+use std::sync::{Condvar, Mutex, OnceLock};
 
 use anyhow::{Context, Result};
 use chrono::{DateTime, Utc};
 use postgres::{Client, NoTls};
+use tokio;
 use tokio_postgres;
 use tracing::{info, instrument, warn};
 use utils::id::{TenantId, TimelineId};
@@ -16,11 +18,11 @@ use utils::lsn::Lsn;
 use compute_api::responses::{ComputeMetrics, ComputeStatus};
 use compute_api::spec::{ComputeMode, ComputeSpec};
 
-use remote_storage::GenericRemoteStorage;
+use remote_storage::{GenericRemoteStorage, RemotePath};
 
-use crate::config;
 use crate::pg_helpers::*;
 use crate::spec::*;
+use crate::{config, extension_server};
 
 /// Compute node info shared across several `compute_ctl` threads.
 pub struct ComputeNode {
@@ -28,6 +30,7 @@ pub struct ComputeNode {
     pub connstr: url::Url,
     pub pgdata: String,
     pub pgbin: String,
+    pub pgversion: String,
     /// We should only allow live re- / configuration of the compute node if
     /// it uses 'pull model', i.e. it can go to control-plane and fetch
     /// the latest configuration. Otherwise, there could be a case:
@@ -49,6 +52,8 @@ pub struct ComputeNode {
     pub state_changed: Condvar,
     ///  S3 extensions configuration variables
     pub ext_remote_storage: Option<GenericRemoteStorage>,
+    pub available_libraries: OnceLock<HashMap<String, RemotePath>>,
+    pub available_extensions: OnceLock<HashMap<String, Vec<RemotePath>>>,
 }
 
 #[derive(Clone, Debug)]
@@ -137,6 +142,84 @@ impl TryFrom<ComputeSpec> for ParsedSpec {
     }
 }
 
+/// Create special neon_superuser role, that's a slightly nerfed version of a real superuser
+/// that we give to customers
+fn create_neon_superuser(spec: &ComputeSpec, client: &mut Client) -> Result<()> {
+    let roles = spec
+        .cluster
+        .roles
+        .iter()
+        .map(|r| format!("'{}'", escape_literal(&r.name)))
+        .collect::<Vec<_>>();
+
+    let dbs = spec
+        .cluster
+        .databases
+        .iter()
+        .map(|db| format!("'{}'", escape_literal(&db.name)))
+        .collect::<Vec<_>>();
+
+    let roles_decl = if roles.is_empty() {
+        String::from("roles text[] := NULL;")
+    } else {
+        format!(
+            r#"
+               roles text[] := ARRAY(SELECT rolname
+                                     FROM pg_catalog.pg_roles
+                                     WHERE rolname IN ({}));"#,
+            roles.join(", ")
+        )
+    };
+
+    let database_decl = if dbs.is_empty() {
+        String::from("dbs text[] := NULL;")
+    } else {
+        format!(
+            r#"
+               dbs text[] := ARRAY(SELECT datname
+                                   FROM pg_catalog.pg_database
+                                   WHERE datname IN ({}));"#,
+            dbs.join(", ")
+        )
+    };
+
+    // ALL PRIVILEGES grants CREATE, CONNECT, and TEMPORARY on all databases
+    // (see https://www.postgresql.org/docs/current/ddl-priv.html)
+    let query = format!(
+        r#"
+            DO $$
+                DECLARE
+                    r text;
+                    {}
+                    {}
+                BEGIN
+                    IF NOT EXISTS (
+                        SELECT FROM pg_catalog.pg_roles WHERE rolname = 'neon_superuser')
+                    THEN
+                        CREATE ROLE neon_superuser CREATEDB CREATEROLE NOLOGIN IN ROLE pg_read_all_data, pg_write_all_data;
+                        IF array_length(roles, 1) IS NOT NULL THEN
+                            EXECUTE format('GRANT neon_superuser TO %s',
+                                           array_to_string(ARRAY(SELECT quote_ident(x) FROM unnest(roles) as x), ', '));
+                            FOREACH r IN ARRAY roles LOOP
+                                EXECUTE format('ALTER ROLE %s CREATEROLE CREATEDB', quote_ident(r));
+                            END LOOP;
+                        END IF;
+                        IF array_length(dbs, 1) IS NOT NULL THEN
+                            EXECUTE format('GRANT ALL PRIVILEGES ON DATABASE %s TO neon_superuser',
+                                           array_to_string(ARRAY(SELECT quote_ident(x) FROM unnest(dbs) as x), ', '));
+                        END IF;
+                    END IF;
+                END
+            $$;"#,
+        roles_decl, database_decl,
+    );
+    info!("Neon superuser created:\n{}", &query);
+    client
+        .simple_query(&query)
+        .map_err(|e| anyhow::anyhow!(e).context(query))?;
+    Ok(())
+}
+
 impl ComputeNode {
     pub fn set_status(&self, status: ComputeStatus) {
         let mut state = self.state.lock().unwrap();
@@ -359,6 +442,8 @@ impl ComputeNode {
                     .map_err(|_| anyhow::anyhow!("invalid connstr"))?;
 
                 let mut client = Client::connect(zenith_admin_connstr.as_str(), NoTls)?;
+                // Disable forwarding so that users don't get a cloud_admin role
+                client.simple_query("SET neon.forward_ddl = false")?;
                 client.simple_query("CREATE USER cloud_admin WITH SUPERUSER")?;
                 client.simple_query("GRANT zenith_admin TO cloud_admin")?;
                 drop(client);
@@ -369,14 +454,16 @@ impl ComputeNode {
             Ok(client) => client,
         };
 
-        // Proceed with post-startup configuration. Note, that order of operations is important.
         // Disable DDL forwarding because control plane already knows about these roles/databases.
         client.simple_query("SET neon.forward_ddl = false")?;
+
+        // Proceed with post-startup configuration. Note, that order of operations is important.
         let spec = &compute_state.pspec.as_ref().expect("spec must be set").spec;
+        create_neon_superuser(spec, &mut client)?;
         handle_roles(spec, &mut client)?;
         handle_databases(spec, &mut client)?;
         handle_role_deletions(spec, self.connstr.as_str(), &mut client)?;
-        handle_grants(spec, self.connstr.as_str(), &mut client)?;
+        handle_grants(spec, self.connstr.as_str())?;
         handle_extensions(spec, &mut client)?;
 
         // 'Close' connection
@@ -414,7 +501,7 @@ impl ComputeNode {
             handle_roles(&spec, &mut client)?;
             handle_databases(&spec, &mut client)?;
             handle_role_deletions(&spec, self.connstr.as_str(), &mut client)?;
-            handle_grants(&spec, self.connstr.as_str(), &mut client)?;
+            handle_grants(&spec, self.connstr.as_str())?;
             handle_extensions(&spec, &mut client)?;
         }
 
@@ -443,6 +530,8 @@ impl ComputeNode {
             pspec.timeline_id,
         );
 
+        self.prepare_external_extensions(&compute_state)?;
+
         self.prepare_pgdata(&compute_state, extension_server_port)?;
 
         let start_time = Utc::now();
@@ -579,4 +668,131 @@ LIMIT 100",
             "{{\"pg_stat_statements\": []}}".to_string()
         }
     }
+
+    // If remote extension storage is configured,
+    // download extension control files
+    // and shared preload libraries.
+    #[tokio::main]
+    pub async fn prepare_external_extensions(&self, compute_state: &ComputeState) -> Result<()> {
+        if let Some(ref ext_remote_storage) = self.ext_remote_storage {
+            let pspec = compute_state.pspec.as_ref().expect("spec must be set");
+            // download preload shared libraries before postgres start (if any)
+            let spec = &pspec.spec;
+
+            // 1. parse private extension paths from spec
+            let private_ext_prefixes = match &spec.private_extensions {
+                Some(private_extensions) => private_extensions.clone(),
+                None => Vec::new(),
+            };
+
+            info!("private_ext_prefixes: {:?}", &private_ext_prefixes);
+
+            // 2. parse shared_preload_libraries from spec
+            let mut libs_vec = Vec::new();
+
+            if let Some(libs) = spec.cluster.settings.find("shared_preload_libraries") {
+                libs_vec = libs
+                    .split(&[',', '\'', ' '])
+                    .filter(|s| *s != "neon" && !s.is_empty())
+                    .map(str::to_string)
+                    .collect();
+            }
+
+            info!(
+                "shared_preload_libraries parsed from spec.cluster.settings: {:?}",
+                libs_vec
+            );
+
+            // also parse shared_preload_libraries from provided postgresql.conf
+            // that is used in neon_local and python tests
+            if let Some(conf) = &spec.cluster.postgresql_conf {
+                let conf_lines = conf.split('\n').collect::<Vec<&str>>();
+
+                let mut shared_preload_libraries_line = "";
+                for line in conf_lines {
+                    if line.starts_with("shared_preload_libraries") {
+                        shared_preload_libraries_line = line;
+                    }
+                }
+
+                let mut preload_libs_vec = Vec::new();
+                if let Some(libs) = shared_preload_libraries_line.split("='").nth(1) {
+                    preload_libs_vec = libs
+                        .split(&[',', '\'', ' '])
+                        .filter(|s| *s != "neon" && !s.is_empty())
+                        .map(str::to_string)
+                        .collect();
+                }
+
+                info!(
+                    "shared_preload_libraries parsed from spec.cluster.postgresql_conf: {:?}",
+                    preload_libs_vec
+                );
+
+                libs_vec.extend(preload_libs_vec);
+            }
+
+            // download extension control files & shared_preload_libraries
+
+            let available_extensions = extension_server::get_available_extensions(
+                ext_remote_storage,
+                &self.pgbin,
+                &self.pgversion,
+                &private_ext_prefixes,
+            )
+            .await?;
+            self.available_extensions
+                .set(available_extensions)
+                .expect("available_extensions.set error");
+
+            info!("Libraries to download: {:?}", &libs_vec);
+            let available_libraries = extension_server::get_available_libraries(
+                ext_remote_storage,
+                &self.pgbin,
+                &self.pgversion,
+                &private_ext_prefixes,
+                &libs_vec,
+            )
+            .await?;
+            self.available_libraries
+                .set(available_libraries)
+                .expect("available_libraries.set error");
+        }
+
+        Ok(())
+    }
+
+    pub async fn download_extension_sql_files(&self, filename: String) -> Result<()> {
+        match &self.ext_remote_storage {
+            None => anyhow::bail!("No remote extension storage"),
+            Some(remote_storage) => {
+                extension_server::download_extension_sql_files(
+                    &filename,
+                    remote_storage,
+                    &self.pgbin,
+                    self.available_extensions
+                        .get()
+                        .context("available_extensions broke")?,
+                )
+                .await
+            }
+        }
+    }
+
+    pub async fn download_library_file(&self, filename: String) -> Result<()> {
+        match &self.ext_remote_storage {
+            None => anyhow::bail!("No remote extension storage"),
+            Some(remote_storage) => {
+                extension_server::download_library_file(
+                    &filename,
+                    remote_storage,
+                    &self.pgbin,
+                    self.available_libraries
+                        .get()
+                        .context("available_libraries broke")?,
+                )
+                .await
+            }
+        }
+    }
 }

compute_tools/src/extension_server.rs

@@ -1,6 +1,9 @@
-use anyhow::{self, bail};
+// Download extension files from the extension store
+// and put them in the right place in the postgres directory
+use anyhow::{self, bail, Context, Result};
 use remote_storage::*;
 use serde_json::{self, Value};
+use std::collections::HashMap;
 use std::fs::File;
 use std::io::{BufWriter, Write};
 use std::num::{NonZeroU32, NonZeroUsize};
@@ -9,6 +12,8 @@ use std::str;
 use tokio::io::AsyncReadExt;
 use tracing::info;
 
+const SHARE_EXT_PATH: &str = "share/postgresql/extension";
+
 fn get_pg_config(argument: &str, pgbin: &str) -> String {
     // gives the result of `pg_config [argument]`
     // where argument is a flag like `--version` or `--sharedir`
@@ -23,23 +28,50 @@ fn get_pg_config(argument: &str, pgbin: &str) -> String {
         .to_string()
 }
 
-fn get_pg_version(pgbin: &str) -> String {
+pub fn get_pg_version(pgbin: &str) -> String {
     // pg_config --version returns a (platform specific) human readable string
     // such as "PostgreSQL 15.4". We parse this to v14/v15
     let human_version = get_pg_config("--version", pgbin);
-    if human_version.contains("v15") {
+    if human_version.contains("15") {
         return "v15".to_string();
+    } else if human_version.contains("14") {
+        return "v14".to_string();
     }
-    "v14".to_string()
+    panic!("Unsuported postgres version {human_version}");
 }
 
 async fn download_helper(
     remote_storage: &GenericRemoteStorage,
     remote_from_path: &RemotePath,
+    remote_from_prefix: Option<&Path>,
     download_location: &Path,
 ) -> anyhow::Result<()> {
     // downloads file at remote_from_path to download_location/[file_name]
-    let local_path = download_location.join(remote_from_path.object_name().expect("bad object"));
+
+    // we cannot use remote_from_path.object_name() here
+    // because extension files can be in subdirectories of the extension store.
+    //
+    // To handle this, we use remote_from_prefix to strip the prefix from the path
+    // this gives us the relative path of the file in the extension store,
+    // and we use this relative path to construct the local path.
+    //
+    let local_path = match remote_from_prefix {
+        Some(prefix) => {
+            let p = remote_from_path
+                .get_path()
+                .strip_prefix(prefix)
+                .expect("bad prefix");
+
+            download_location.join(p)
+        }
+        None => download_location.join(remote_from_path.object_name().expect("bad object")),
+    };
+
+    if local_path.exists() {
+        info!("File {:?} already exists. Skipping download", &local_path);
+        return Ok(());
+    }
+
     info!(
         "Downloading {:?} to location {:?}",
         &remote_from_path, &local_path
@@ -50,64 +82,194 @@ async fn download_helper(
         .download_stream
         .read_to_end(&mut write_data_buffer)
         .await?;
+    if remote_from_prefix.is_some() {
+        if let Some(prefix) = local_path.parent() {
+            info!(
+                "Downloading file with prefix. Create directory {:?}",
+                prefix
+            );
+            // if directory already exists, this is a no-op
+            std::fs::create_dir_all(prefix)?;
+        }
+    }
+
     let mut output_file = BufWriter::new(File::create(local_path)?);
     output_file.write_all(&write_data_buffer)?;
     Ok(())
 }
 
-pub enum ExtensionType {
-    Shared,          // we just use the public folder here
-    Tenant(String),  // String is tenant_id
-    Library(String), // String is name of the extension
+// download extension control files
+//
+// if private_ext_prefixes is provided - search also in private extension paths
+//
+pub async fn get_available_extensions(
+    remote_storage: &GenericRemoteStorage,
+    pgbin: &str,
+    pg_version: &str,
+    private_ext_prefixes: &Vec<String>,
+) -> anyhow::Result<HashMap<String, Vec<RemotePath>>> {
+    let local_sharedir = Path::new(&get_pg_config("--sharedir", pgbin)).join("extension");
+
+    let mut paths: Vec<RemotePath> = Vec::new();
+    // public extensions
+    paths.push(RemotePath::new(
+        &Path::new(pg_version).join(SHARE_EXT_PATH),
+    )?);
+    // private extensions
+    for private_prefix in private_ext_prefixes {
+        paths.push(RemotePath::new(
+            &Path::new(pg_version)
+                .join(private_prefix)
+                .join(SHARE_EXT_PATH),
+        )?);
+    }
+
+    let all_available_files = list_files_in_prefixes_for_extensions(remote_storage, &paths).await?;
+
+    info!(
+        "list of available_extension files {:?}",
+        &all_available_files
+    );
+
+    // download all control files
+    for (obj_name, obj_paths) in &all_available_files {
+        for obj_path in obj_paths {
+            if obj_name.ends_with("control") {
+                download_helper(remote_storage, obj_path, None, &local_sharedir).await?;
+            }
+        }
+    }
+
+    Ok(all_available_files)
 }
 
-pub async fn download_extension(
-    remote_storage: &Option<GenericRemoteStorage>,
-    ext_type: ExtensionType,
+// Download requested shared_preload_libraries
+//
+// Note that tenant_id is not optional here, because we only download libraries
+// after we know the tenant spec and the tenant_id.
+//
+// return list of all library files to use it in the future searches
+pub async fn get_available_libraries(
+    remote_storage: &GenericRemoteStorage,
     pgbin: &str,
-) -> anyhow::Result<()> {
-    let remote_storage = match remote_storage {
-        Some(remote_storage) => remote_storage,
-        None => return Ok(()),
-    };
-    let pg_version = get_pg_version(pgbin);
-    match ext_type {
-        ExtensionType::Shared => {
-            // 1. Download control files from s3-bucket/public/*.control to SHAREDIR/extension
-            // We can do this step even before we have spec,
-            // because public extensions are common for all projects.
-            let local_sharedir = Path::new(&get_pg_config("--sharedir", pgbin)).join("extension");
-            let remote_sharedir = Path::new(&pg_version).join("share/postgresql/extension");
-            let remote_sharedir = RemotePath::new(Path::new(&remote_sharedir))?;
-            let from_paths = remote_storage.list_files(Some(&remote_sharedir)).await?;
-            for remote_from_path in from_paths {
-                if remote_from_path.extension() == Some("control") {
-                    download_helper(remote_storage, &remote_from_path, &local_sharedir).await?;
+    pg_version: &str,
+    private_ext_prefixes: &Vec<String>,
+    preload_libraries: &Vec<String>,
+) -> anyhow::Result<HashMap<String, RemotePath>> {
+    let local_libdir: PathBuf = Path::new(&get_pg_config("--pkglibdir", pgbin)).into();
+    // Construct a hashmap of all available libraries
+    // example (key, value) pair: test_lib0.so, v14/lib/test_lib0.so
+
+    let mut paths: Vec<RemotePath> = Vec::new();
+    // public libraries
+    paths.push(
+        RemotePath::new(&Path::new(&pg_version).join("lib/"))
+            .expect("The hard coded path here is valid"),
+    );
+    // private libraries
+    for private_prefix in private_ext_prefixes {
+        paths.push(
+            RemotePath::new(&Path::new(&pg_version).join(private_prefix).join("lib"))
+                .expect("The hard coded path here is valid"),
+        );
+    }
+
+    let all_available_libraries = list_files_in_prefixes(remote_storage, &paths).await?;
+
+    info!("list of library files {:?}", &all_available_libraries);
+
+    // download all requested libraries
+    for lib_name in preload_libraries {
+        // add file extension if it isn't in the filename
+        let lib_name_with_ext = enforce_so_end(lib_name);
+        info!("looking for library {:?}", &lib_name_with_ext);
+
+        match all_available_libraries.get(&*lib_name_with_ext) {
+            Some(remote_path) => {
+                download_helper(remote_storage, remote_path, None, &local_libdir).await?
+            }
+            None => {
+                let file_path = local_libdir.join(&lib_name_with_ext);
+                if file_path.exists() {
+                    info!("File {:?} already exists. Skipping download", &file_path);
+                } else {
+                    bail!("Shared library file {lib_name} is not found in the extension store")
                 }
             }
         }
-        ExtensionType::Tenant(tenant_id) => {
-            // 2. After we have spec, before project start
-            // Download control files from s3-bucket/[tenant-id]/*.control to SHAREDIR/extension
+    }
 
-            let local_sharedir = Path::new(&get_pg_config("--sharedir", pgbin)).join("extension");
-            let remote_path = RemotePath::new(Path::new(&tenant_id.to_string()))?;
-            let from_paths = remote_storage.list_files(Some(&remote_path)).await?;
-            for remote_from_path in from_paths {
-                if remote_from_path.extension() == Some("control") {
-                    download_helper(remote_storage, &remote_from_path, &local_sharedir).await?;
-                }
+    Ok(all_available_libraries)
+}
+
+// download all sqlfiles (and possibly data files) for a given extension name
+//
+pub async fn download_extension_sql_files(
+    ext_name: &str,
+    remote_storage: &GenericRemoteStorage,
+    pgbin: &str,
+    all_available_files: &HashMap<String, Vec<RemotePath>>,
+) -> Result<()> {
+    let local_sharedir = Path::new(&get_pg_config("--sharedir", pgbin)).join("extension");
+    let mut downloaded_something = false;
+
+    if let Some(files) = all_available_files.get(ext_name) {
+        for file in files {
+            if file.extension().context("bad file name")? != "control" {
+                // find files prefix to handle cases when extension files are stored
+                // in a directory with the same name as the extension
+                // example:
+                // share/postgresql/extension/extension_name/extension_name--1.0.sql
+                let index = file
+                    .get_path()
+                    .to_str()
+                    .context("invalid path")?
+                    .find(ext_name)
+                    .context("invalid path")?;
+
+                let prefix_str =
+                    file.get_path().to_str().context("invalid path")?[..index].to_string();
+                let remote_from_prefix = if prefix_str.is_empty() {
+                    None
+                } else {
+                    Some(Path::new(&prefix_str))
+                };
+
+                download_helper(remote_storage, file, remote_from_prefix, &local_sharedir).await?;
+                downloaded_something = true;
             }
         }
-        ExtensionType::Library(library_name) => {
-            // 3. After we have spec, before postgres start
-            // Download preload_shared_libraries from s3-bucket/public/[library-name].control into LIBDIR/
+    }
+    if !downloaded_something {
+        bail!("Files for extension {ext_name} are not found in the extension store");
+    }
+    Ok(())
+}
 
-            let local_libdir: PathBuf = Path::new(&get_pg_config("--libdir", pgbin)).into();
-            let remote_path = format!("{library_name}.control");
-            let remote_from_path = RemotePath::new(Path::new(&remote_path))?;
-            download_helper(remote_storage, &remote_from_path, &local_libdir).await?;
+// appends an .so suffix to libname if it does not already have one
+fn enforce_so_end(libname: &str) -> String {
+    if !libname.ends_with(".so") {
+        format!("{}.so", libname)
+    } else {
+        libname.to_string()
+    }
+}
+
+// download shared library file
+pub async fn download_library_file(
+    lib_name: &str,
+    remote_storage: &GenericRemoteStorage,
+    pgbin: &str,
+    all_available_libraries: &HashMap<String, RemotePath>,
+) -> Result<()> {
+    let local_libdir: PathBuf = Path::new(&get_pg_config("--pkglibdir", pgbin)).into();
+    let lib_name_with_ext = enforce_so_end(lib_name);
+    info!("looking for library {:?}", &lib_name_with_ext);
+    match all_available_libraries.get(&*lib_name_with_ext) {
+        Some(remote_path) => {
+            download_helper(remote_storage, remote_path, None, &local_libdir).await?
         }
+        None => bail!("Shared library file {lib_name} is not found in the extension store"),
     }
     Ok(())
 }
@@ -147,3 +309,71 @@ pub fn init_remote_storage(remote_ext_config: &str) -> anyhow::Result<GenericRem
     };
     GenericRemoteStorage::from_config(&config)
 }
+
+// helper to collect all files in the given prefixes
+// returns hashmap of (file_name, file_remote_path)
+async fn list_files_in_prefixes(
+    remote_storage: &GenericRemoteStorage,
+    paths: &Vec<RemotePath>,
+) -> Result<HashMap<String, RemotePath>> {
+    let mut res = HashMap::new();
+
+    for path in paths {
+        for file in remote_storage.list_files(Some(path)).await? {
+            res.insert(
+                file.object_name().expect("bad object").to_owned(),
+                file.to_owned(),
+            );
+        }
+    }
+
+    Ok(res)
+}
+
+// helper to extract extension name
+// extension files can be in subdirectories of the extension store.
+// examples of layout:
+//
+// share/postgresql/extension/extension_name--1.0.sql
+//
+// or
+//
+// share/postgresql/extension/extension_name/extension_name--1.0.sql
+// share/postgresql/extension/extension_name/extra_data.csv
+//
+// Note: we **assume** that the  extension files is in one of these formats.
+// If it is not, this code will not download it.
+fn get_ext_name(path: &str) -> Result<&str> {
+    let path_suffix: Vec<&str> = path.split(&format!("{SHARE_EXT_PATH}/")).collect();
+
+    let path_suffix = path_suffix.last().expect("bad ext name");
+    // the order of these is important
+    // otherwise we'll return incorrect extension name
+    // for path like share/postgresql/extension/extension_name/extension_name--1.0.sql
+    for index in ["/", "--"] {
+        if let Some(index) = path_suffix.find(index) {
+            return Ok(&path_suffix[..index]);
+        }
+    }
+    Ok(path_suffix)
+}
+
+// helper to collect files of given prefixes for extensions
+// and group them by extension
+// returns a hashmap of (extension_name, Vector of remote paths for all files needed for this extension)
+async fn list_files_in_prefixes_for_extensions(
+    remote_storage: &GenericRemoteStorage,
+    paths: &Vec<RemotePath>,
+) -> Result<HashMap<String, Vec<RemotePath>>> {
+    let mut result = HashMap::new();
+    for path in paths {
+        for file in remote_storage.list_files(Some(path)).await? {
+            let file_ext_name = get_ext_name(file.get_path().to_str().context("invalid path")?)?;
+            let ext_file_list = result
+                .entry(file_ext_name.to_string())
+                .or_insert(Vec::new());
+            ext_file_list.push(file.to_owned());
+        }
+    }
+    Ok(result)
+}

compute_tools/src/http/api.rs

@@ -16,8 +16,6 @@ use tokio::task;
 use tracing::{error, info};
 use tracing_utils::http::OtelName;
 
-use crate::extension_server::{self, ExtensionType};
-
 fn status_response_from_state(state: &ComputeState) -> ComputeStatusResponse {
     ComputeStatusResponse {
         start_time: state.start_time,
@@ -126,25 +124,51 @@ async fn routes(req: Request<Body>, compute: &Arc<ComputeNode>) -> Response<Body
         // download extension files from S3 on demand
         (&Method::POST, route) if route.starts_with("/extension_server/") => {
             info!("serving {:?} POST request", route);
+            info!("req.uri {:?}", req.uri());
 
-            let filename = route.split('/').last().unwrap();
+            let mut is_library = false;
+
+            if let Some(params) = req.uri().query() {
+                info!("serving {:?} POST request with params: {}", route, params);
+
+                if params == "is_library=true" {
+                    is_library = true;
+                } else {
+                    let mut resp = Response::new(Body::from("Wrong request parameters"));
+                    *resp.status_mut() = StatusCode::BAD_REQUEST;
+                    return resp;
+                }
+            }
+
+            let filename = route.split('/').last().unwrap().to_string();
 
             info!(
-                "serving /extension_server POST request, filename: {:?}",
-                filename
+                "serving /extension_server POST request, filename: {:?} is_library: {}",
+                filename, is_library
             );
 
-            match extension_server::download_extension(
-                &compute.ext_remote_storage,
-                ExtensionType::Shared,
-                &compute.pgbin,
-            )
-            .await
-            {
-                Ok(_) => Response::new(Body::from("OK")),
-                Err(e) => {
-                    error!("download_extension failed: {}", e);
-                    Response::new(Body::from(e.to_string()))
+            if is_library {
+                match compute.download_library_file(filename.to_string()).await {
+                    Ok(_) => Response::new(Body::from("OK")),
+                    Err(e) => {
+                        error!("library download failed: {}", e);
+                        let mut resp = Response::new(Body::from(e.to_string()));
+                        *resp.status_mut() = StatusCode::INTERNAL_SERVER_ERROR;
+                        resp
+                    }
+                }
+            } else {
+                match compute
+                    .download_extension_sql_files(filename.to_string())
+                    .await
+                {
+                    Ok(_) => Response::new(Body::from("OK")),
+                    Err(e) => {
+                        error!("extension download failed: {}", e);
+                        let mut resp = Response::new(Body::from(e.to_string()));
+                        *resp.status_mut() = StatusCode::INTERNAL_SERVER_ERROR;
+                        resp
+                    }
                 }
             }
         }

compute_tools/src/pg_helpers.rs

@@ -17,7 +17,7 @@ use compute_api::spec::{Database, GenericOption, GenericOptions, PgIdent, Role};
 const POSTGRES_WAIT_TIMEOUT: Duration = Duration::from_millis(60 * 1000); // milliseconds
 
 /// Escape a string for including it in a SQL literal
-fn escape_literal(s: &str) -> String {
+pub fn escape_literal(s: &str) -> String {
     s.replace('\'', "''").replace('\\', "\\\\")
 }
 

compute_tools/src/spec.rs

@@ -269,17 +269,13 @@ pub fn handle_roles(spec: &ComputeSpec, client: &mut Client) -> Result<()> {
                 xact.execute(query.as_str(), &[])?;
             }
             RoleAction::Create => {
-                let mut query: String = format!("CREATE ROLE {} ", name.pg_quote());
+                let mut query: String = format!(
+                    "CREATE ROLE {} CREATEROLE CREATEDB IN ROLE neon_superuser",
+                    name.pg_quote()
+                );
                 info!("role create query: '{}'", &query);
                 query.push_str(&role.to_pg_options());
                 xact.execute(query.as_str(), &[])?;
-
-                let grant_query = format!(
-                    "GRANT pg_read_all_data, pg_write_all_data TO {}",
-                    name.pg_quote()
-                );
-                xact.execute(grant_query.as_str(), &[])?;
-                info!("role grant query: '{}'", &grant_query);
             }
         }
 
@@ -476,6 +472,11 @@ pub fn handle_databases(spec: &ComputeSpec, client: &mut Client) -> Result<()> {
                 query.push_str(&db.to_pg_options());
                 let _guard = info_span!("executing", query).entered();
                 client.execute(query.as_str(), &[])?;
+                let grant_query: String = format!(
+                    "GRANT ALL PRIVILEGES ON DATABASE {} TO neon_superuser",
+                    name.pg_quote()
+                );
+                client.execute(grant_query.as_str(), &[])?;
             }
         };
 
@@ -495,35 +496,9 @@ pub fn handle_databases(spec: &ComputeSpec, client: &mut Client) -> Result<()> {
 /// Grant CREATE ON DATABASE to the database owner and do some other alters and grants
 /// to allow users creating trusted extensions and re-creating `public` schema, for example.
 #[instrument(skip_all)]
-pub fn handle_grants(spec: &ComputeSpec, connstr: &str, client: &mut Client) -> Result<()> {
+pub fn handle_grants(spec: &ComputeSpec, connstr: &str) -> Result<()> {
     info!("cluster spec grants:");
 
-    // We now have a separate `web_access` role to connect to the database
-    // via the web interface and proxy link auth. And also we grant a
-    // read / write all data privilege to every role. So also grant
-    // create to everyone.
-    // XXX: later we should stop messing with Postgres ACL in such horrible
-    // ways.
-    let roles = spec
-        .cluster
-        .roles
-        .iter()
-        .map(|r| r.name.pg_quote())
-        .collect::<Vec<_>>();
-
-    for db in &spec.cluster.databases {
-        let dbname = &db.name;
-
-        let query: String = format!(
-            "GRANT CREATE ON DATABASE {} TO {}",
-            dbname.pg_quote(),
-            roles.join(", ")
-        );
-        info!("grant query {}", &query);
-
-        client.execute(query.as_str(), &[])?;
-    }
-
     // Do some per-database access adjustments. We'd better do this at db creation time,
     // but CREATE DATABASE isn't transactional. So we cannot create db + do some grants
     // atomically.

control_plane/src/endpoint.rs

@@ -311,7 +311,7 @@ impl Endpoint {
 
                 // TODO: use future host field from safekeeper spec
                 // Pass the list of safekeepers to the replica so that it can connect to any of them,
-                // whichever is availiable.
+                // whichever is available.
                 let sk_ports = self
                     .env
                     .safekeepers
@@ -481,6 +481,7 @@ impl Endpoint {
             pageserver_connstring: Some(pageserver_connstring),
             safekeeper_connstrings,
             storage_auth_token: auth_token.clone(),
+            private_extensions: Some(vec![self.tenant_id.to_string()]), //DEBUG ONLY
         };
         let spec_path = self.endpoint_path().join("spec.json");
         std::fs::write(spec_path, serde_json::to_string_pretty(&spec)?)?;

libs/compute_api/src/spec.rs

@@ -60,6 +60,8 @@ pub struct ComputeSpec {
     /// If set, 'storage_auth_token' is used as the password to authenticate to
     /// the pageserver and safekeepers.
     pub storage_auth_token: Option<String>,
+
+    pub private_extensions: Option<Vec<String>>,
 }
 
 #[serde_as]

libs/metrics/src/lib.rs

@@ -23,6 +23,7 @@ use prometheus::{Registry, Result};
 pub mod launch_timestamp;
 mod wrappers;
 pub use wrappers::{CountedReader, CountedWriter};
+pub mod metric_vec_duration;
 
 pub type UIntGauge = GenericGauge<AtomicU64>;
 pub type UIntGaugeVec = GenericGaugeVec<AtomicU64>;

libs/metrics/src/metric_vec_duration.rs

@@ -0,0 +1,23 @@
+//! Helpers for observing duration on HistogramVec / CounterVec / GaugeVec / MetricVec<T>.
+
+use std::{future::Future, time::Instant};
+
+pub trait DurationResultObserver {
+    fn observe_result<T, E>(&self, res: &Result<T, E>, duration: std::time::Duration);
+}
+
+pub async fn observe_async_block_duration_by_result<
+    T,
+    E,
+    F: Future<Output = Result<T, E>>,
+    O: DurationResultObserver,
+>(
+    observer: &O,
+    block: F,
+) -> Result<T, E> {
+    let start = Instant::now();
+    let result = block.await;
+    let duration = start.elapsed();
+    observer.observe_result(&result, duration);
+    result
+}

libs/remote_storage/src/s3_bucket.rs

@@ -349,11 +349,17 @@ impl RemoteStorage for S3Bucket {
 
     /// See the doc for `RemoteStorage::list_files`
     async fn list_files(&self, folder: Option<&RemotePath>) -> anyhow::Result<Vec<RemotePath>> {
-        // TODO: if bucket prefix is empty, folder is prefixed with a "/" I think. Is this desired?
-        let folder_name = folder
+        let mut folder_name = folder
             .map(|p| self.relative_path_to_s3_object(p))
             .or_else(|| self.prefix_in_bucket.clone());
 
+        // remove leading "/" if one exists
+        if let Some(folder_name_slash) = folder_name.clone() {
+            if folder_name_slash.starts_with(REMOTE_STORAGE_PREFIX_SEPARATOR) {
+                folder_name = Some(folder_name_slash[1..].to_string());
+            }
+        }
+
         // AWS may need to break the response into several parts
         let mut continuation_token = None;
         let mut all_files = vec![];

pageserver/src/bin/pageserver.rs

@@ -495,50 +495,50 @@ fn start_pageserver(
                 Ok(())
             },
         );
+    }
 
-        if let Some(metric_collection_endpoint) = &conf.metric_collection_endpoint {
-            let background_jobs_barrier = background_jobs_barrier;
-            let metrics_ctx = RequestContext::todo_child(
-                TaskKind::MetricsCollection,
-                // This task itself shouldn't download anything.
-                // The actual size calculation does need downloads, and
-                // creates a child context with the right DownloadBehavior.
-                DownloadBehavior::Error,
-            );
-            task_mgr::spawn(
-                MGMT_REQUEST_RUNTIME.handle(),
-                TaskKind::MetricsCollection,
-                None,
-                None,
-                "consumption metrics collection",
-                true,
-                async move {
-                    // first wait until background jobs are cleared to launch.
-                    //
-                    // this is because we only process active tenants and timelines, and the
-                    // Timeline::get_current_logical_size will spawn the logical size calculation,
-                    // which will not be rate-limited.
-                    let cancel = task_mgr::shutdown_token();
+    if let Some(metric_collection_endpoint) = &conf.metric_collection_endpoint {
+        let background_jobs_barrier = background_jobs_barrier;
+        let metrics_ctx = RequestContext::todo_child(
+            TaskKind::MetricsCollection,
+            // This task itself shouldn't download anything.
+            // The actual size calculation does need downloads, and
+            // creates a child context with the right DownloadBehavior.
+            DownloadBehavior::Error,
+        );
+        task_mgr::spawn(
+            crate::BACKGROUND_RUNTIME.handle(),
+            TaskKind::MetricsCollection,
+            None,
+            None,
+            "consumption metrics collection",
+            true,
+            async move {
+                // first wait until background jobs are cleared to launch.
+                //
+                // this is because we only process active tenants and timelines, and the
+                // Timeline::get_current_logical_size will spawn the logical size calculation,
+                // which will not be rate-limited.
+                let cancel = task_mgr::shutdown_token();
 
-                    tokio::select! {
-                        _ = cancel.cancelled() => { return Ok(()); },
-                        _ = background_jobs_barrier.wait() => {}
-                    };
+                tokio::select! {
+                    _ = cancel.cancelled() => { return Ok(()); },
+                    _ = background_jobs_barrier.wait() => {}
+                };
 
-                    pageserver::consumption_metrics::collect_metrics(
-                        metric_collection_endpoint,
-                        conf.metric_collection_interval,
-                        conf.cached_metric_collection_interval,
-                        conf.synthetic_size_calculation_interval,
-                        conf.id,
-                        metrics_ctx,
-                    )
-                    .instrument(info_span!("metrics_collection"))
-                    .await?;
-                    Ok(())
-                },
-            );
-        }
+                pageserver::consumption_metrics::collect_metrics(
+                    metric_collection_endpoint,
+                    conf.metric_collection_interval,
+                    conf.cached_metric_collection_interval,
+                    conf.synthetic_size_calculation_interval,
+                    conf.id,
+                    metrics_ctx,
+                )
+                .instrument(info_span!("metrics_collection"))
+                .await?;
+                Ok(())
+            },
+        );
     }
 
     // Spawn a task to listen for libpq connections. It will spawn further tasks

pageserver/src/config.rs

@@ -96,12 +96,12 @@ pub mod defaults {
 
 #background_task_maximum_delay = '{DEFAULT_BACKGROUND_TASK_MAXIMUM_DELAY}'
 
-# [tenant_config]
+[tenant_config]
 #checkpoint_distance = {DEFAULT_CHECKPOINT_DISTANCE} # in bytes
 #checkpoint_timeout = {DEFAULT_CHECKPOINT_TIMEOUT}
 #compaction_target_size = {DEFAULT_COMPACTION_TARGET_SIZE} # in bytes
 #compaction_period = '{DEFAULT_COMPACTION_PERIOD}'
-#compaction_threshold = '{DEFAULT_COMPACTION_THRESHOLD}'
+#compaction_threshold = {DEFAULT_COMPACTION_THRESHOLD}
 
 #gc_period = '{DEFAULT_GC_PERIOD}'
 #gc_horizon = {DEFAULT_GC_HORIZON}
@@ -111,7 +111,8 @@ pub mod defaults {
 #min_resident_size_override = .. # in bytes
 #evictions_low_residence_duration_metric_threshold = '{DEFAULT_EVICTIONS_LOW_RESIDENCE_DURATION_METRIC_THRESHOLD}'
 #gc_feedback = false
-# [remote_storage]
+
+[remote_storage]
 
 "###
     );

pageserver/src/http/routes.rs

@@ -1128,8 +1128,6 @@ async fn disk_usage_eviction_run(
         freed_bytes: 0,
     };
 
-    use crate::task_mgr::MGMT_REQUEST_RUNTIME;
-
     let (tx, rx) = tokio::sync::oneshot::channel();
 
     let state = get_state(&r);
@@ -1147,7 +1145,7 @@ async fn disk_usage_eviction_run(
     let _g = cancel.drop_guard();
 
     crate::task_mgr::spawn(
-        MGMT_REQUEST_RUNTIME.handle(),
+        crate::task_mgr::BACKGROUND_RUNTIME.handle(),
         TaskKind::DiskUsageEviction,
         None,
         None,

pageserver/src/metrics.rs

@@ -1,3 +1,4 @@
+use metrics::metric_vec_duration::DurationResultObserver;
 use metrics::{
     register_counter_vec, register_histogram, register_histogram_vec, register_int_counter,
     register_int_counter_vec, register_int_gauge, register_int_gauge_vec, register_uint_gauge_vec,
@@ -424,6 +425,27 @@ pub static SMGR_QUERY_TIME: Lazy<HistogramVec> = Lazy::new(|| {
     .expect("failed to define a metric")
 });
 
+pub struct BasebackupQueryTime(HistogramVec);
+pub static BASEBACKUP_QUERY_TIME: Lazy<BasebackupQueryTime> = Lazy::new(|| {
+    BasebackupQueryTime({
+        register_histogram_vec!(
+            "pageserver_basebackup_query_seconds",
+            "Histogram of basebackup queries durations, by result type",
+            &["result"],
+            CRITICAL_OP_BUCKETS.into(),
+        )
+        .expect("failed to define a metric")
+    })
+});
+
+impl DurationResultObserver for BasebackupQueryTime {
+    fn observe_result<T, E>(&self, res: &Result<T, E>, duration: std::time::Duration) {
+        let label_value = if res.is_ok() { "ok" } else { "error" };
+        let metric = self.0.get_metric_with_label_values(&[label_value]).unwrap();
+        metric.observe(duration.as_secs_f64());
+    }
+}
+
 pub static LIVE_CONNECTIONS_COUNT: Lazy<IntGaugeVec> = Lazy::new(|| {
     register_int_gauge_vec!(
         "pageserver_live_connections",
@@ -823,11 +845,6 @@ impl TimelineMetrics {
         let evictions_with_low_residence_duration =
             evictions_with_low_residence_duration_builder.build(&tenant_id, &timeline_id);
 
-        // TODO(chi): remove this once we remove Lazy for all metrics. Otherwise this will not appear in the exporter
-        // and integration test will error.
-        MATERIALIZED_PAGE_CACHE_HIT_DIRECT.get();
-        MATERIALIZED_PAGE_CACHE_HIT.get();
-
         TimelineMetrics {
             tenant_id,
             timeline_id,
@@ -1302,4 +1319,8 @@ pub fn preinitialize_metrics() {
 
     // Same as above for this metric, but, it's a Vec-type metric for which we don't know all the labels.
     BACKGROUND_LOOP_PERIOD_OVERRUN_COUNT.reset();
+
+    // Python tests need these.
+    MATERIALIZED_PAGE_CACHE_HIT_DIRECT.get();
+    MATERIALIZED_PAGE_CACHE_HIT.get();
 }

pageserver/src/page_service.rs

@@ -913,10 +913,24 @@ where
                 None
             };
 
-            // Check that the timeline exists
-            self.handle_basebackup_request(pgb, tenant_id, timeline_id, lsn, None, false, ctx)
-                .await?;
-            pgb.write_message_noflush(&BeMessage::CommandComplete(b"SELECT 1"))?;
+            metrics::metric_vec_duration::observe_async_block_duration_by_result(
+                &*crate::metrics::BASEBACKUP_QUERY_TIME,
+                async move {
+                    self.handle_basebackup_request(
+                        pgb,
+                        tenant_id,
+                        timeline_id,
+                        lsn,
+                        None,
+                        false,
+                        ctx,
+                    )
+                    .await?;
+                    pgb.write_message_noflush(&BeMessage::CommandComplete(b"SELECT 1"))?;
+                    anyhow::Ok(())
+                },
+            )
+            .await?;
         }
         // return pair of prev_lsn and last_lsn
         else if query_string.starts_with("get_last_record_rlsn ") {

pageserver/src/tenant/storage_layer.rs

@@ -12,7 +12,7 @@ use crate::context::RequestContext;
 use crate::repository::{Key, Value};
 use crate::task_mgr::TaskKind;
 use crate::walrecord::NeonWalRecord;
-use anyhow::Result;
+use anyhow::{Context, Result};
 use bytes::Bytes;
 use enum_map::EnumMap;
 use enumset::EnumSet;
@@ -343,7 +343,8 @@ impl LayerAccessStats {
 /// All layers should implement a minimal `std::fmt::Debug` without tenant or
 /// timeline names, because those are known in the context of which the layers
 /// are used in (timeline).
-pub trait Layer: std::fmt::Debug + Send + Sync {
+#[async_trait::async_trait]
+pub trait Layer: std::fmt::Debug + Send + Sync + 'static {
     /// Range of keys that this layer covers
     fn get_key_range(&self) -> Range<Key>;
 
@@ -373,13 +374,42 @@ pub trait Layer: std::fmt::Debug + Send + Sync {
     /// is available. If this returns ValueReconstructResult::Continue, look up
     /// the predecessor layer and call again with the same 'reconstruct_data' to
     /// collect more data.
-    fn get_value_reconstruct_data(
+    fn get_value_reconstruct_data_blocking(
         &self,
         key: Key,
         lsn_range: Range<Lsn>,
-        reconstruct_data: &mut ValueReconstructState,
-        ctx: &RequestContext,
-    ) -> Result<ValueReconstructResult>;
+        reconstruct_data: ValueReconstructState,
+        ctx: RequestContext,
+    ) -> Result<(ValueReconstructState, ValueReconstructResult)>;
+
+    /// CANCEL SAFETY: if the returned future is dropped,
+    /// the wrapped closure still run to completion and the return value discarded.
+    /// For the case of get_value_reconstruct_data, we expect the closure to not
+    /// have any side effects, as it only attempts to read a layer (and stuff like
+    /// page cache isn't considered a real side effect).
+    /// But, ...
+    /// TRACING:
+    /// If the returned future is cancelled, the spawn_blocking span can outlive
+    /// the caller's span.
+    /// So, technically, we should be using `parent: None` and `follows_from: current`
+    /// instead. However, in practice, the advantage of maintaining the span stack
+    /// in logs outweighs the disadvantage of having a dangling span in a case that
+    /// is not expected to happen because in pageserver we generally don't drop pending futures.
+    async fn get_value_reconstruct_data(
+        self: Arc<Self>,
+        key: Key,
+        lsn_range: Range<Lsn>,
+        reconstruct_data: ValueReconstructState,
+        ctx: RequestContext,
+    ) -> Result<(ValueReconstructState, ValueReconstructResult)> {
+        let span = tracing::info_span!("get_value_reconstruct_data_spawn_blocking");
+        tokio::task::spawn_blocking(move || {
+            let _enter = span.enter();
+            self.get_value_reconstruct_data_blocking(key, lsn_range, reconstruct_data, ctx)
+        })
+        .await
+        .context("spawn_blocking")?
+    }
 
     /// A short ID string that uniquely identifies the given layer within a [`LayerMap`].
     fn short_id(&self) -> String;
@@ -499,6 +529,7 @@ impl LayerDescriptor {
     }
 }
 
+#[async_trait::async_trait]
 impl Layer for LayerDescriptor {
     fn get_key_range(&self) -> Range<Key> {
         self.key.clone()
@@ -512,13 +543,13 @@ impl Layer for LayerDescriptor {
         self.is_incremental
     }
 
-    fn get_value_reconstruct_data(
+    fn get_value_reconstruct_data_blocking(
         &self,
         _key: Key,
         _lsn_range: Range<Lsn>,
-        _reconstruct_data: &mut ValueReconstructState,
-        _ctx: &RequestContext,
-    ) -> Result<ValueReconstructResult> {
+        _reconstruct_data: ValueReconstructState,
+        _ctx: RequestContext,
+    ) -> Result<(ValueReconstructState, ValueReconstructResult)> {
         todo!("This method shouldn't be part of the Layer trait")
     }
 

pageserver/src/tenant/storage_layer/delta_layer.rs

@@ -218,6 +218,7 @@ impl std::fmt::Debug for DeltaLayerInner {
     }
 }
 
+#[async_trait::async_trait]
 impl Layer for DeltaLayer {
     /// debugging function to print out the contents of the layer
     fn dump(&self, verbose: bool, ctx: &RequestContext) -> Result<()> {
@@ -294,13 +295,13 @@ impl Layer for DeltaLayer {
         Ok(())
     }
 
-    fn get_value_reconstruct_data(
+    fn get_value_reconstruct_data_blocking(
         &self,
         key: Key,
         lsn_range: Range<Lsn>,
-        reconstruct_state: &mut ValueReconstructState,
-        ctx: &RequestContext,
-    ) -> anyhow::Result<ValueReconstructResult> {
+        mut reconstruct_state: ValueReconstructState,
+        ctx: RequestContext,
+    ) -> anyhow::Result<(ValueReconstructState, ValueReconstructResult)> {
         ensure!(lsn_range.start >= self.desc.lsn_range.start);
         let mut need_image = true;
 
@@ -308,7 +309,7 @@ impl Layer for DeltaLayer {
 
         {
             // Open the file and lock the metadata in memory
-            let inner = self.load(LayerAccessKind::GetValueReconstructData, ctx)?;
+            let inner = self.load(LayerAccessKind::GetValueReconstructData, &ctx)?;
 
             // Scan the page versions backwards, starting from `lsn`.
             let file = &inner.file;
@@ -374,9 +375,9 @@ impl Layer for DeltaLayer {
         // If an older page image is needed to reconstruct the page, let the
         // caller know.
         if need_image {
-            Ok(ValueReconstructResult::Continue)
+            Ok((reconstruct_state, ValueReconstructResult::Continue))
         } else {
-            Ok(ValueReconstructResult::Complete)
+            Ok((reconstruct_state, ValueReconstructResult::Complete))
         }
     }
 

pageserver/src/tenant/storage_layer/image_layer.rs

@@ -149,6 +149,7 @@ impl std::fmt::Debug for ImageLayerInner {
     }
 }
 
+#[async_trait::async_trait]
 impl Layer for ImageLayer {
     /// debugging function to print out the contents of the layer
     fn dump(&self, verbose: bool, ctx: &RequestContext) -> Result<()> {
@@ -181,18 +182,18 @@ impl Layer for ImageLayer {
     }
 
     /// Look up given page in the file
-    fn get_value_reconstruct_data(
+    fn get_value_reconstruct_data_blocking(
         &self,
         key: Key,
         lsn_range: Range<Lsn>,
-        reconstruct_state: &mut ValueReconstructState,
-        ctx: &RequestContext,
-    ) -> anyhow::Result<ValueReconstructResult> {
+        mut reconstruct_state: ValueReconstructState,
+        ctx: RequestContext,
+    ) -> anyhow::Result<(ValueReconstructState, ValueReconstructResult)> {
         assert!(self.desc.key_range.contains(&key));
         assert!(lsn_range.start >= self.lsn);
         assert!(lsn_range.end >= self.lsn);
 
-        let inner = self.load(LayerAccessKind::GetValueReconstructData, ctx)?;
+        let inner = self.load(LayerAccessKind::GetValueReconstructData, &ctx)?;
 
         let file = inner.file.as_ref().unwrap();
         let tree_reader = DiskBtreeReader::new(inner.index_start_blk, inner.index_root_blk, file);
@@ -210,9 +211,9 @@ impl Layer for ImageLayer {
             let value = Bytes::from(blob);
 
             reconstruct_state.img = Some((self.lsn, value));
-            Ok(ValueReconstructResult::Complete)
+            Ok((reconstruct_state, ValueReconstructResult::Complete))
         } else {
-            Ok(ValueReconstructResult::Missing)
+            Ok((reconstruct_state, ValueReconstructResult::Missing))
         }
     }
 

pageserver/src/tenant/storage_layer/inmemory_layer.rs

@@ -110,6 +110,7 @@ impl InMemoryLayer {
     }
 }
 
+#[async_trait::async_trait]
 impl Layer for InMemoryLayer {
     fn get_key_range(&self) -> Range<Key> {
         Key::MIN..Key::MAX
@@ -190,13 +191,13 @@ impl Layer for InMemoryLayer {
     }
 
     /// Look up given value in the layer.
-    fn get_value_reconstruct_data(
+    fn get_value_reconstruct_data_blocking(
         &self,
         key: Key,
         lsn_range: Range<Lsn>,
-        reconstruct_state: &mut ValueReconstructState,
-        _ctx: &RequestContext,
-    ) -> anyhow::Result<ValueReconstructResult> {
+        mut reconstruct_state: ValueReconstructState,
+        _ctx: RequestContext,
+    ) -> anyhow::Result<(ValueReconstructState, ValueReconstructResult)> {
         ensure!(lsn_range.start >= self.start_lsn);
         let mut need_image = true;
 
@@ -213,7 +214,7 @@ impl Layer for InMemoryLayer {
                 match value {
                     Value::Image(img) => {
                         reconstruct_state.img = Some((*entry_lsn, img));
-                        return Ok(ValueReconstructResult::Complete);
+                        return Ok((reconstruct_state, ValueReconstructResult::Complete));
                     }
                     Value::WalRecord(rec) => {
                         let will_init = rec.will_init();
@@ -233,9 +234,9 @@ impl Layer for InMemoryLayer {
         // If an older page image is needed to reconstruct the page, let the
         // caller know.
         if need_image {
-            Ok(ValueReconstructResult::Continue)
+            Ok((reconstruct_state, ValueReconstructResult::Continue))
         } else {
-            Ok(ValueReconstructResult::Complete)
+            Ok((reconstruct_state, ValueReconstructResult::Complete))
         }
     }
 }

pageserver/src/tenant/storage_layer/remote_layer.rs

@@ -6,7 +6,7 @@ use crate::context::RequestContext;
 use crate::repository::Key;
 use crate::tenant::layer_map::BatchedUpdates;
 use crate::tenant::remote_timeline_client::index::LayerFileMetadata;
-use crate::tenant::storage_layer::{Layer, ValueReconstructResult, ValueReconstructState};
+use crate::tenant::storage_layer::{Layer, ValueReconstructState};
 use anyhow::{bail, Result};
 use pageserver_api::models::HistoricLayerInfo;
 use std::ops::Range;
@@ -21,7 +21,7 @@ use utils::{
 use super::filename::{DeltaFileName, ImageFileName};
 use super::{
     DeltaLayer, ImageLayer, LayerAccessStats, LayerAccessStatsReset, LayerIter, LayerKeyIter,
-    LayerResidenceStatus, PersistentLayer, PersistentLayerDesc,
+    LayerResidenceStatus, PersistentLayer, PersistentLayerDesc, ValueReconstructResult,
 };
 
 /// RemoteLayer is a not yet downloaded [`ImageLayer`] or
@@ -63,14 +63,15 @@ impl std::fmt::Debug for RemoteLayer {
     }
 }
 
+#[async_trait::async_trait]
 impl Layer for RemoteLayer {
-    fn get_value_reconstruct_data(
+    fn get_value_reconstruct_data_blocking(
         &self,
         _key: Key,
         _lsn_range: Range<Lsn>,
-        _reconstruct_state: &mut ValueReconstructState,
-        _ctx: &RequestContext,
-    ) -> Result<ValueReconstructResult> {
+        _reconstruct_state: ValueReconstructState,
+        _ctx: RequestContext,
+    ) -> Result<(ValueReconstructState, ValueReconstructResult)> {
         bail!(
             "layer {} needs to be downloaded",
             self.filename().file_name()

pageserver/src/tenant/timeline.rs

@@ -129,7 +129,7 @@ pub struct Timeline {
 
     pub pg_version: u32,
 
-    pub(crate) layers: tokio::sync::RwLock<LayerMap<dyn PersistentLayer>>,
+    pub(crate) layers: Arc<tokio::sync::RwLock<LayerMap<dyn PersistentLayer>>>,
 
     /// Set of key ranges which should be covered by image layers to
     /// allow GC to remove old layers. This set is created by GC and its cutoff LSN is also stored.
@@ -555,13 +555,14 @@ impl Timeline {
             None => None,
         };
 
-        let mut reconstruct_state = ValueReconstructState {
+        let reconstruct_state = ValueReconstructState {
             records: Vec::new(),
             img: cached_page_img,
         };
 
         let timer = self.metrics.get_reconstruct_data_time_histo.start_timer();
-        self.get_reconstruct_data(key, lsn, &mut reconstruct_state, ctx)
+        let reconstruct_state = self
+            .get_reconstruct_data(key, lsn, reconstruct_state, ctx)
             .await?;
         timer.stop_and_record();
 
@@ -1418,7 +1419,7 @@ impl Timeline {
                 timeline_id,
                 tenant_id,
                 pg_version,
-                layers: tokio::sync::RwLock::new(LayerMap::default()),
+                layers: Arc::new(tokio::sync::RwLock::new(LayerMap::default())),
                 wanted_image_layers: Mutex::new(None),
 
                 walredo_mgr,
@@ -2352,9 +2353,9 @@ impl Timeline {
         &self,
         key: Key,
         request_lsn: Lsn,
-        reconstruct_state: &mut ValueReconstructState,
+        mut reconstruct_state: ValueReconstructState,
         ctx: &RequestContext,
-    ) -> Result<(), PageReconstructError> {
+    ) -> Result<ValueReconstructState, PageReconstructError> {
         // Start from the current timeline.
         let mut timeline_owned;
         let mut timeline = self;
@@ -2384,12 +2385,12 @@ impl Timeline {
             // The function should have updated 'state'
             //info!("CALLED for {} at {}: {:?} with {} records, cached {}", key, cont_lsn, result, reconstruct_state.records.len(), cached_lsn);
             match result {
-                ValueReconstructResult::Complete => return Ok(()),
+                ValueReconstructResult::Complete => return Ok(reconstruct_state),
                 ValueReconstructResult::Continue => {
                     // If we reached an earlier cached page image, we're done.
                     if cont_lsn == cached_lsn + 1 {
                         MATERIALIZED_PAGE_CACHE_HIT.inc_by(1);
-                        return Ok(());
+                        return Ok(reconstruct_state);
                     }
                     if prev_lsn <= cont_lsn {
                         // Didn't make any progress in last iteration. Error out to avoid
@@ -2493,13 +2494,19 @@ impl Timeline {
                             // Get all the data needed to reconstruct the page version from this layer.
                             // But if we have an older cached page image, no need to go past that.
                             let lsn_floor = max(cached_lsn + 1, start_lsn);
-                            result = match open_layer.get_value_reconstruct_data(
-                                key,
-                                lsn_floor..cont_lsn,
-                                reconstruct_state,
-                                ctx,
-                            ) {
-                                Ok(result) => result,
+                            result = match Arc::clone(open_layer)
+                                .get_value_reconstruct_data(
+                                    key,
+                                    lsn_floor..cont_lsn,
+                                    reconstruct_state,
+                                    ctx.attached_child(),
+                                )
+                                .await
+                            {
+                                Ok((new_reconstruct_state, result)) => {
+                                    reconstruct_state = new_reconstruct_state;
+                                    result
+                                }
                                 Err(e) => return Err(PageReconstructError::from(e)),
                             };
                             cont_lsn = lsn_floor;
@@ -2520,13 +2527,19 @@ impl Timeline {
                         if cont_lsn > start_lsn {
                             //info!("CHECKING for {} at {} on frozen layer {}", key, cont_lsn, frozen_layer.filename().display());
                             let lsn_floor = max(cached_lsn + 1, start_lsn);
-                            result = match frozen_layer.get_value_reconstruct_data(
-                                key,
-                                lsn_floor..cont_lsn,
-                                reconstruct_state,
-                                ctx,
-                            ) {
-                                Ok(result) => result,
+                            result = match Arc::clone(frozen_layer)
+                                .get_value_reconstruct_data(
+                                    key,
+                                    lsn_floor..cont_lsn,
+                                    reconstruct_state,
+                                    ctx.attached_child(),
+                                )
+                                .await
+                            {
+                                Ok((new_reconstruct_state, result)) => {
+                                    reconstruct_state = new_reconstruct_state;
+                                    result
+                                }
                                 Err(e) => return Err(PageReconstructError::from(e)),
                             };
                             cont_lsn = lsn_floor;
@@ -2555,13 +2568,19 @@ impl Timeline {
                             // Get all the data needed to reconstruct the page version from this layer.
                             // But if we have an older cached page image, no need to go past that.
                             let lsn_floor = max(cached_lsn + 1, lsn_floor);
-                            result = match layer.get_value_reconstruct_data(
-                                key,
-                                lsn_floor..cont_lsn,
-                                reconstruct_state,
-                                ctx,
-                            ) {
-                                Ok(result) => result,
+                            result = match Arc::clone(&layer)
+                                .get_value_reconstruct_data(
+                                    key,
+                                    lsn_floor..cont_lsn,
+                                    reconstruct_state,
+                                    ctx.attached_child(),
+                                )
+                                .await
+                            {
+                                Ok((new_reconstruct_state, result)) => {
+                                    reconstruct_state = new_reconstruct_state;
+                                    result
+                                }
                                 Err(e) => return Err(PageReconstructError::from(e)),
                             };
                             cont_lsn = lsn_floor;
@@ -3370,14 +3389,14 @@ struct CompactLevel0Phase1StatsBuilder {
     version: Option<u64>,
     tenant_id: Option<TenantId>,
     timeline_id: Option<TimelineId>,
-    first_read_lock_acquisition_micros: DurationRecorder,
-    get_level0_deltas_plus_drop_lock_micros: DurationRecorder,
-    level0_deltas_count: Option<usize>,
-    time_spent_between_locks: DurationRecorder,
-    second_read_lock_acquisition_micros: DurationRecorder,
-    second_read_lock_held_micros: DurationRecorder,
-    sort_holes_micros: DurationRecorder,
+    read_lock_acquisition_micros: DurationRecorder,
+    read_lock_held_spawn_blocking_startup_micros: DurationRecorder,
+    read_lock_held_prerequisites_micros: DurationRecorder,
+    read_lock_held_compute_holes_micros: DurationRecorder,
+    read_lock_drop_micros: DurationRecorder,
+    prepare_iterators_micros: DurationRecorder,
     write_layer_files_micros: DurationRecorder,
+    level0_deltas_count: Option<usize>,
     new_deltas_count: Option<usize>,
     new_deltas_size: Option<u64>,
 }
@@ -3390,14 +3409,14 @@ struct CompactLevel0Phase1Stats {
     tenant_id: TenantId,
     #[serde_as(as = "serde_with::DisplayFromStr")]
     timeline_id: TimelineId,
-    first_read_lock_acquisition_micros: RecordedDuration,
-    get_level0_deltas_plus_drop_lock_micros: RecordedDuration,
-    level0_deltas_count: usize,
-    time_spent_between_locks: RecordedDuration,
-    second_read_lock_acquisition_micros: RecordedDuration,
-    second_read_lock_held_micros: RecordedDuration,
-    sort_holes_micros: RecordedDuration,
+    read_lock_acquisition_micros: RecordedDuration,
+    read_lock_held_spawn_blocking_startup_micros: RecordedDuration,
+    read_lock_held_prerequisites_micros: RecordedDuration,
+    read_lock_held_compute_holes_micros: RecordedDuration,
+    read_lock_drop_micros: RecordedDuration,
+    prepare_iterators_micros: RecordedDuration,
     write_layer_files_micros: RecordedDuration,
+    level0_deltas_count: usize,
     new_deltas_count: usize,
     new_deltas_size: u64,
 }
@@ -3406,54 +3425,51 @@ impl TryFrom<CompactLevel0Phase1StatsBuilder> for CompactLevel0Phase1Stats {
     type Error = anyhow::Error;
 
     fn try_from(value: CompactLevel0Phase1StatsBuilder) -> Result<Self, Self::Error> {
-        let CompactLevel0Phase1StatsBuilder {
-            version,
-            tenant_id,
-            timeline_id,
-            first_read_lock_acquisition_micros,
-            get_level0_deltas_plus_drop_lock_micros,
-            level0_deltas_count,
-            time_spent_between_locks,
-            second_read_lock_acquisition_micros,
-            second_read_lock_held_micros,
-            sort_holes_micros,
-            write_layer_files_micros,
-            new_deltas_count,
-            new_deltas_size,
-        } = value;
-        Ok(CompactLevel0Phase1Stats {
-            version: version.ok_or_else(|| anyhow::anyhow!("version not set"))?,
-            tenant_id: tenant_id.ok_or_else(|| anyhow::anyhow!("tenant_id not set"))?,
-            timeline_id: timeline_id.ok_or_else(|| anyhow::anyhow!("timeline_id not set"))?,
-            first_read_lock_acquisition_micros: first_read_lock_acquisition_micros
+        Ok(Self {
+            version: value.version.ok_or_else(|| anyhow!("version not set"))?,
+            tenant_id: value
+                .tenant_id
+                .ok_or_else(|| anyhow!("tenant_id not set"))?,
+            timeline_id: value
+                .timeline_id
+                .ok_or_else(|| anyhow!("timeline_id not set"))?,
+            read_lock_acquisition_micros: value
+                .read_lock_acquisition_micros
                 .into_recorded()
-                .ok_or_else(|| anyhow::anyhow!("first_read_lock_acquisition_micros not set"))?,
-            get_level0_deltas_plus_drop_lock_micros: get_level0_deltas_plus_drop_lock_micros
+                .ok_or_else(|| anyhow!("read_lock_acquisition_micros not set"))?,
+            read_lock_held_spawn_blocking_startup_micros: value
+                .read_lock_held_spawn_blocking_startup_micros
                 .into_recorded()
-                .ok_or_else(|| {
-                    anyhow::anyhow!("get_level0_deltas_plus_drop_lock_micros not set")
-                })?,
-            level0_deltas_count: level0_deltas_count
-                .ok_or_else(|| anyhow::anyhow!("level0_deltas_count not set"))?,
-            time_spent_between_locks: time_spent_between_locks
+                .ok_or_else(|| anyhow!("read_lock_held_spawn_blocking_startup_micros not set"))?,
+            read_lock_held_prerequisites_micros: value
+                .read_lock_held_prerequisites_micros
                 .into_recorded()
-                .ok_or_else(|| anyhow::anyhow!("time_spent_between_locks not set"))?,
-            second_read_lock_acquisition_micros: second_read_lock_acquisition_micros
+                .ok_or_else(|| anyhow!("read_lock_held_prerequisites_micros not set"))?,
+            read_lock_held_compute_holes_micros: value
+                .read_lock_held_compute_holes_micros
                 .into_recorded()
-                .ok_or_else(|| anyhow::anyhow!("second_read_lock_acquisition_micros not set"))?,
-            second_read_lock_held_micros: second_read_lock_held_micros
+                .ok_or_else(|| anyhow!("read_lock_held_compute_holes_micros not set"))?,
+            read_lock_drop_micros: value
+                .read_lock_drop_micros
                 .into_recorded()
-                .ok_or_else(|| anyhow::anyhow!("second_read_lock_held_micros not set"))?,
-            sort_holes_micros: sort_holes_micros
+                .ok_or_else(|| anyhow!("read_lock_drop_micros not set"))?,
+            prepare_iterators_micros: value
+                .prepare_iterators_micros
                 .into_recorded()
-                .ok_or_else(|| anyhow::anyhow!("sort_holes_micros not set"))?,
-            write_layer_files_micros: write_layer_files_micros
+                .ok_or_else(|| anyhow!("prepare_iterators_micros not set"))?,
+            write_layer_files_micros: value
+                .write_layer_files_micros
                 .into_recorded()
-                .ok_or_else(|| anyhow::anyhow!("write_layer_files_micros not set"))?,
-            new_deltas_count: new_deltas_count
-                .ok_or_else(|| anyhow::anyhow!("new_deltas_count not set"))?,
-            new_deltas_size: new_deltas_size
-                .ok_or_else(|| anyhow::anyhow!("new_deltas_size not set"))?,
+                .ok_or_else(|| anyhow!("write_layer_files_micros not set"))?,
+            level0_deltas_count: value
+                .level0_deltas_count
+                .ok_or_else(|| anyhow!("level0_deltas_count not set"))?,
+            new_deltas_count: value
+                .new_deltas_count
+                .ok_or_else(|| anyhow!("new_deltas_count not set"))?,
+            new_deltas_size: value
+                .new_deltas_size
+                .ok_or_else(|| anyhow!("new_deltas_size not set"))?,
         })
     }
 }
@@ -3464,30 +3480,18 @@ impl Timeline {
     /// This method takes the `_layer_removal_cs` guard to highlight it required downloads are
     /// returned as an error. If the `layer_removal_cs` boundary is changed not to be taken in the
     /// start of level0 files compaction, the on-demand download should be revisited as well.
-    async fn compact_level0_phase1(
-        &self,
+    fn compact_level0_phase1(
+        self: Arc<Self>,
         _layer_removal_cs: Arc<tokio::sync::OwnedMutexGuard<()>>,
+        layers: tokio::sync::OwnedRwLockReadGuard<LayerMap<dyn PersistentLayer>>,
+        mut stats: CompactLevel0Phase1StatsBuilder,
         target_file_size: u64,
         ctx: &RequestContext,
     ) -> Result<CompactLevel0Phase1Result, CompactionError> {
-        let mut stats = CompactLevel0Phase1StatsBuilder {
-            version: Some(1),
-            tenant_id: Some(self.tenant_id),
-            timeline_id: Some(self.timeline_id),
-            ..Default::default()
-        };
-
-        let begin = tokio::time::Instant::now();
-        let layers = self.layers.read().await;
-        let now = tokio::time::Instant::now();
-        stats.first_read_lock_acquisition_micros =
-            DurationRecorder::Recorded(RecordedDuration(now - begin), now);
+        stats.read_lock_held_spawn_blocking_startup_micros =
+            stats.read_lock_acquisition_micros.till_now(); // set by caller
         let mut level0_deltas = layers.get_level0_deltas()?;
-        drop(layers);
         stats.level0_deltas_count = Some(level0_deltas.len());
-        stats.get_level0_deltas_plus_drop_lock_micros =
-            stats.first_read_lock_acquisition_micros.till_now();
-
         // Only compact if enough layers have accumulated.
         let threshold = self.get_compaction_threshold();
         if level0_deltas.is_empty() || level0_deltas.len() < threshold {
@@ -3565,6 +3569,53 @@ impl Timeline {
         // we don't accidentally use it later in the function.
         drop(level0_deltas);
 
+        stats.read_lock_held_prerequisites_micros = stats
+            .read_lock_held_spawn_blocking_startup_micros
+            .till_now();
+
+        // Determine N largest holes where N is number of compacted layers.
+        let max_holes = deltas_to_compact.len();
+        let last_record_lsn = self.get_last_record_lsn();
+        let min_hole_range = (target_file_size / page_cache::PAGE_SZ as u64) as i128;
+        let min_hole_coverage_size = 3; // TODO: something more flexible?
+
+        // min-heap (reserve space for one more element added before eviction)
+        let mut heap: BinaryHeap<Hole> = BinaryHeap::with_capacity(max_holes + 1);
+        let mut prev: Option<Key> = None;
+        for (next_key, _next_lsn, _size) in itertools::process_results(
+            deltas_to_compact.iter().map(|l| l.key_iter(ctx)),
+            |iter_iter| iter_iter.kmerge_by(|a, b| a.0 <= b.0),
+        )? {
+            if let Some(prev_key) = prev {
+                // just first fast filter
+                if next_key.to_i128() - prev_key.to_i128() >= min_hole_range {
+                    let key_range = prev_key..next_key;
+                    // Measuring hole by just subtraction of i128 representation of key range boundaries
+                    // has not so much sense, because largest holes will corresponds field1/field2 changes.
+                    // But we are mostly interested to eliminate holes which cause generation of excessive image layers.
+                    // That is why it is better to measure size of hole as number of covering image layers.
+                    let coverage_size = layers.image_coverage(&key_range, last_record_lsn)?.len();
+                    if coverage_size >= min_hole_coverage_size {
+                        heap.push(Hole {
+                            key_range,
+                            coverage_size,
+                        });
+                        if heap.len() > max_holes {
+                            heap.pop(); // remove smallest hole
+                        }
+                    }
+                }
+            }
+            prev = Some(next_key.next());
+        }
+        stats.read_lock_held_compute_holes_micros =
+            stats.read_lock_held_prerequisites_micros.till_now();
+        drop(layers);
+        stats.read_lock_drop_micros = stats.read_lock_held_compute_holes_micros.till_now();
+        let mut holes = heap.into_vec();
+        holes.sort_unstable_by_key(|hole| hole.key_range.start);
+        let mut next_hole = 0; // index of next hole in holes vector
+
         // This iterator walks through all key-value pairs from all the layers
         // we're compacting, in key, LSN order.
         let all_values_iter = itertools::process_results(
@@ -3604,50 +3655,7 @@ impl Timeline {
             },
         )?;
 
-        // Determine N largest holes where N is number of compacted layers.
-        let max_holes = deltas_to_compact.len();
-        let last_record_lsn = self.get_last_record_lsn();
-        stats.time_spent_between_locks = stats.get_level0_deltas_plus_drop_lock_micros.till_now();
-        let layers = self.layers.read().await; // Is'n it better to hold original layers lock till here?
-        stats.second_read_lock_acquisition_micros = stats.time_spent_between_locks.till_now();
-        let min_hole_range = (target_file_size / page_cache::PAGE_SZ as u64) as i128;
-        let min_hole_coverage_size = 3; // TODO: something more flexible?
-
-        // min-heap (reserve space for one more element added before eviction)
-        let mut heap: BinaryHeap<Hole> = BinaryHeap::with_capacity(max_holes + 1);
-        let mut prev: Option<Key> = None;
-        for (next_key, _next_lsn, _size) in itertools::process_results(
-            deltas_to_compact.iter().map(|l| l.key_iter(ctx)),
-            |iter_iter| iter_iter.kmerge_by(|a, b| a.0 <= b.0),
-        )? {
-            if let Some(prev_key) = prev {
-                // just first fast filter
-                if next_key.to_i128() - prev_key.to_i128() >= min_hole_range {
-                    let key_range = prev_key..next_key;
-                    // Measuring hole by just subtraction of i128 representation of key range boundaries
-                    // has not so much sense, because largest holes will corresponds field1/field2 changes.
-                    // But we are mostly interested to eliminate holes which cause generation of excessive image layers.
-                    // That is why it is better to measure size of hole as number of covering image layers.
-                    let coverage_size = layers.image_coverage(&key_range, last_record_lsn)?.len();
-                    if coverage_size >= min_hole_coverage_size {
-                        heap.push(Hole {
-                            key_range,
-                            coverage_size,
-                        });
-                        if heap.len() > max_holes {
-                            heap.pop(); // remove smallest hole
-                        }
-                    }
-                }
-            }
-            prev = Some(next_key.next());
-        }
-        drop(layers);
-        stats.second_read_lock_held_micros = stats.second_read_lock_acquisition_micros.till_now();
-        let mut holes = heap.into_vec();
-        holes.sort_unstable_by_key(|hole| hole.key_range.start);
-        let mut next_hole = 0; // index of next hole in holes vector
-        stats.sort_holes_micros = stats.second_read_lock_held_micros.till_now();
+        stats.prepare_iterators_micros = stats.read_lock_drop_micros.till_now();
 
         // Merge the contents of all the input delta layers into a new set
         // of delta layers, based on the current partitioning.
@@ -3807,7 +3815,7 @@ impl Timeline {
             layer_paths.pop().unwrap();
         }
 
-        stats.write_layer_files_micros = stats.sort_holes_micros.till_now();
+        stats.write_layer_files_micros = stats.prepare_iterators_micros.till_now();
         stats.new_deltas_count = Some(new_layers.len());
         stats.new_deltas_size = Some(new_layers.iter().map(|l| l.desc.file_size).sum());
 
@@ -3846,9 +3854,36 @@ impl Timeline {
         let CompactLevel0Phase1Result {
             new_layers,
             deltas_to_compact,
-        } = self
-            .compact_level0_phase1(layer_removal_cs.clone(), target_file_size, ctx)
-            .await?;
+        } = {
+            let phase1_span = info_span!("compact_level0_phase1");
+            let myself = Arc::clone(self);
+            let ctx = ctx.attached_child(); // technically, the spawn_blocking can outlive this future
+            let mut stats = CompactLevel0Phase1StatsBuilder {
+                version: Some(2),
+                tenant_id: Some(self.tenant_id),
+                timeline_id: Some(self.timeline_id),
+                ..Default::default()
+            };
+
+            let begin = tokio::time::Instant::now();
+            let phase1_layers_locked = Arc::clone(&self.layers).read_owned().await;
+            let now = tokio::time::Instant::now();
+            stats.read_lock_acquisition_micros =
+                DurationRecorder::Recorded(RecordedDuration(now - begin), now);
+            let layer_removal_cs = layer_removal_cs.clone();
+            tokio::task::spawn_blocking(move || {
+                let _entered = phase1_span.enter();
+                myself.compact_level0_phase1(
+                    layer_removal_cs,
+                    phase1_layers_locked,
+                    stats,
+                    target_file_size,
+                    &ctx,
+                )
+            })
+            .await
+            .context("spawn_blocking")??
+        };
 
         if new_layers.is_empty() && deltas_to_compact.is_empty() {
             // nothing to do

pgxn/neon/extension_server.c

@@ -29,9 +29,13 @@ static int extension_server_port = 0;
 
 static download_extension_file_hook_type prev_download_extension_file_hook = NULL;
 
-// curl -X POST http://localhost:8080/extension_server/postgis-3.so
+// to download all SQL files for an extension:
+// curl -X POST http://localhost:8080/extension_server/postgis
+//
+// to download specific library file:
+// curl -X POST http://localhost:8080/extension_server/postgis-3.so?=true
 static bool
-neon_download_extension_file_http(const char *filename)
+neon_download_extension_file_http(const char *filename, bool is_library)
 {
     CURL *curl;
     CURLcode res;
@@ -44,7 +48,14 @@ neon_download_extension_file_http(const char *filename)
         elog(ERROR, "Failed to initialize curl handle");
     }
 
-    compute_ctl_url = psprintf("http://localhost:%d/extension_server/%s", extension_server_port, filename);
+
+    if (is_library)
+    {
+        elog(LOG, "request library");
+    }
+
+    compute_ctl_url = psprintf("http://localhost:%d/extension_server/%s%s",
+                      extension_server_port, filename, is_library?"?is_library=true":"");
 
     elog(LOG, "curl_easy_perform() url: %s", compute_ctl_url);
 
@@ -52,6 +63,7 @@ neon_download_extension_file_http(const char *filename)
     curl_easy_setopt(curl, CURLOPT_URL, compute_ctl_url);
     curl_easy_setopt(curl, CURLOPT_TIMEOUT, 3L /* seconds */);
 
+
     if (curl)
     {
         /* Perform the request, res will get the return code */

rust-toolchain.toml

@@ -1,5 +1,5 @@
 [toolchain]
-channel = "1.68.2"
+channel = "1.70.0"
 profile = "default"
 # The default profile includes rustc, rust-std, cargo, rust-docs, rustfmt and clippy.
 # https://rust-lang.github.io/rustup/concepts/profiles.html

test_runner/fixtures/metrics.py

@@ -62,6 +62,7 @@ PAGESERVER_GLOBAL_METRICS: Tuple[str, ...] = (
     "pageserver_getpage_reconstruct_seconds_bucket",
     "pageserver_getpage_reconstruct_seconds_count",
     "pageserver_getpage_reconstruct_seconds_sum",
+    *[f"pageserver_basebackup_query_seconds_{x}" for x in ["bucket", "count", "sum"]],
 )
 
 PAGESERVER_PER_TENANT_METRICS: Tuple[str, ...] = (

test_runner/fixtures/neon_fixtures.py

@@ -534,6 +534,16 @@ class S3Storage:
             "AWS_SECRET_ACCESS_KEY": self.secret_key,
         }
 
+    def to_string(self) -> str:
+        return json.dumps(
+            {
+                "bucket": self.bucket_name,
+                "region": self.bucket_region,
+                "endpoint": self.endpoint,
+                "prefix": self.prefix_in_bucket,
+            }
+        )
+
 
 RemoteStorage = Union[LocalFsStorage, S3Storage]
 
@@ -600,12 +610,12 @@ class NeonEnvBuilder:
         self.rust_log_override = rust_log_override
         self.port_distributor = port_distributor
         self.remote_storage = remote_storage
-        self.ext_remote_storage: Optional[Any] = None
+        self.ext_remote_storage: Optional[S3Storage] = None
         self.remote_storage_client: Optional[Any] = None
         self.remote_storage_users = remote_storage_users
         self.broker = broker
         self.run_id = run_id
-        self.mock_s3_server = mock_s3_server
+        self.mock_s3_server: MockS3Server = mock_s3_server
         self.pageserver_config_override = pageserver_config_override
         self.num_safekeepers = num_safekeepers
         self.safekeepers_id_start = safekeepers_id_start
@@ -666,7 +676,11 @@ class NeonEnvBuilder:
                 enable_remote_extensions=enable_remote_extensions,
             )
         elif remote_storage_kind == RemoteStorageKind.REAL_S3:
-            self.enable_real_s3_remote_storage(test_name=test_name, force_enable=force_enable)
+            self.enable_real_s3_remote_storage(
+                test_name=test_name,
+                force_enable=force_enable,
+                enable_remote_extensions=enable_remote_extensions,
+            )
         else:
             raise RuntimeError(f"Unknown storage type: {remote_storage_kind}")
 
@@ -709,20 +723,22 @@ class NeonEnvBuilder:
             bucket_region=mock_region,
             access_key=self.mock_s3_server.access_key(),
             secret_key=self.mock_s3_server.secret_key(),
+            prefix_in_bucket="pageserver",
         )
 
         if enable_remote_extensions:
-            ext_bucket_name = f"ext_{bucket_name}"
-            self.remote_storage_client.create_bucket(Bucket=ext_bucket_name)
             self.ext_remote_storage = S3Storage(
-                bucket_name=ext_bucket_name,
+                bucket_name=bucket_name,
                 endpoint=mock_endpoint,
                 bucket_region=mock_region,
                 access_key=self.mock_s3_server.access_key(),
                 secret_key=self.mock_s3_server.secret_key(),
+                prefix_in_bucket="ext",
             )
 
-    def enable_real_s3_remote_storage(self, test_name: str, force_enable: bool = True):
+    def enable_real_s3_remote_storage(
+        self, test_name: str, force_enable: bool = True, enable_remote_extensions: bool = False
+    ):
         """
         Sets up configuration to use real s3 endpoint without mock server
         """
@@ -759,18 +775,16 @@ class NeonEnvBuilder:
             bucket_region=region,
             access_key=access_key,
             secret_key=secret_key,
-            prefix_in_bucket=self.remote_storage_prefix,
+            prefix_in_bucket=f"{self.remote_storage_prefix}/pageserver",
         )
 
-        ext_bucket_name = os.getenv("EXT_REMOTE_STORAGE_S3_BUCKET")
-        if ext_bucket_name is not None:
-            ext_bucket_name = f"ext_{ext_bucket_name}"
+        if enable_remote_extensions:
             self.ext_remote_storage = S3Storage(
-                bucket_name=ext_bucket_name,
+                bucket_name=bucket_name,
                 bucket_region=region,
                 access_key=access_key,
                 secret_key=secret_key,
-                prefix_in_bucket=self.remote_storage_prefix,
+                prefix_in_bucket=f"{self.remote_storage_prefix}/ext",
             )
 
     def cleanup_local_storage(self):

test_runner/fixtures/types.py

@@ -89,6 +89,9 @@ class TenantId(Id):
     def __repr__(self) -> str:
         return f'`TenantId("{self.id.hex()}")'
 
+    def __str__(self) -> str:
+        return self.id.hex()
+
 
 class TimelineId(Id):
     def __repr__(self) -> str:

test_runner/regress/test_compatibility.py

@@ -2,6 +2,7 @@ import copy
 import os
 import shutil
 import subprocess
+import tempfile
 from pathlib import Path
 from typing import Any, Optional
 
@@ -448,7 +449,7 @@ def dump_differs(first: Path, second: Path, output: Path) -> bool:
     """
 
     with output.open("w") as stdout:
-        rv = subprocess.run(
+        res = subprocess.run(
             [
                 "diff",
                 "--unified",  # Make diff output more readable
@@ -460,4 +461,53 @@ def dump_differs(first: Path, second: Path, output: Path) -> bool:
             stdout=stdout,
         )
 
-    return rv.returncode != 0
+    differs = res.returncode != 0
+
+    # TODO: Remove after https://github.com/neondatabase/neon/pull/4425 is merged, and a couple of releases are made
+    if differs:
+        with tempfile.NamedTemporaryFile(mode="w") as tmp:
+            tmp.write(PR4425_ALLOWED_DIFF)
+            tmp.flush()
+
+            allowed = subprocess.run(
+                [
+                    "diff",
+                    "--unified",  # Make diff output more readable
+                    r"--ignore-matching-lines=^---",  # Ignore diff headers
+                    r"--ignore-matching-lines=^\+\+\+",  # Ignore diff headers
+                    "--ignore-matching-lines=^@@",  # Ignore diff blocks location
+                    "--ignore-matching-lines=^ *$",  # Ignore lines with only spaces
+                    "--ignore-matching-lines=^ --.*",  # Ignore the " --" lines for compatibility with PG14
+                    "--ignore-blank-lines",
+                    str(output),
+                    str(tmp.name),
+                ],
+            )
+
+            differs = allowed.returncode != 0
+
+    return differs
+
+
+PR4425_ALLOWED_DIFF = """
+--- /tmp/test_output/test_backward_compatibility[release-pg15]/compatibility_snapshot/dump.sql 2023-06-08 18:12:45.000000000 +0000
++++ /tmp/test_output/test_backward_compatibility[release-pg15]/dump.sql        2023-06-13 07:25:35.211733653 +0000
+@@ -13,12 +13,20 @@
+
+ CREATE ROLE cloud_admin;
+ ALTER ROLE cloud_admin WITH SUPERUSER INHERIT CREATEROLE CREATEDB LOGIN REPLICATION BYPASSRLS;
++CREATE ROLE neon_superuser;
++ALTER ROLE neon_superuser WITH NOSUPERUSER INHERIT CREATEROLE CREATEDB NOLOGIN NOREPLICATION NOBYPASSRLS;
+
+ --
+ -- User Configurations
+ --
+
+
++--
++-- Role memberships
++--
++
++GRANT pg_read_all_data TO neon_superuser GRANTED BY cloud_admin;
++GRANT pg_write_all_data TO neon_superuser GRANTED BY cloud_admin;
+"""

test_runner/regress/test_download_extensions.py

@@ -1,90 +1,252 @@
-import json
 import os
 from contextlib import closing
 from io import BytesIO
 
+import pytest
 from fixtures.log_helper import log
-from fixtures.neon_fixtures import (
-    NeonEnvBuilder,
-    RemoteStorageKind,
+from fixtures.neon_fixtures import NeonEnvBuilder, PgBin, RemoteStorageKind
+from fixtures.pg_version import PgVersion
+from fixtures.types import TenantId
+
+NUM_EXT = 3
+
+
+def control_file_content(owner, i):
+    output = f"""# mock {owner} extension{i}
+comment = 'This is a mock extension'
+default_version = '1.0'
+module_pathname = '$libdir/test_ext{i}'
+relocatable = true"""
+    return output
+
+
+def sql_file_content():
+    output = """
+            CREATE FUNCTION test_ext_add(integer, integer) RETURNS integer
+    AS 'select $1 + $2;'
+    LANGUAGE SQL
+    IMMUTABLE
+    RETURNS NULL ON NULL INPUT;
+        """
+    return output
+
+
+# Prepare some mock extension files and upload them to the bucket
+# returns a list of files that should be cleaned up after the test
+def prepare_mock_ext_storage(
+    pg_version: PgVersion,
+    tenant_id: TenantId,
+    pg_bin: PgBin,
+    ext_remote_storage,
+    remote_storage_client,
+):
+    bucket_prefix = ext_remote_storage.prefix_in_bucket
+    private_prefix = str(tenant_id)
+    PUB_EXT_ROOT = f"v{pg_version}/share/postgresql/extension"
+    PRIVATE_EXT_ROOT = f"v{pg_version}/{private_prefix}/share/postgresql/extension"
+    LOCAL_EXT_ROOT = f"pg_install/{PUB_EXT_ROOT}"
+
+    PUB_LIB_ROOT = f"v{pg_version}/lib"
+    PRIVATE_LIB_ROOT = f"v{pg_version}/{private_prefix}/lib"
+    LOCAL_LIB_ROOT = f"{pg_bin.pg_lib_dir}/postgresql"
+
+    log.info(
+        f"""
+            PUB_EXT_ROOT: {PUB_EXT_ROOT}
+            PRIVATE_EXT_ROOT: {PRIVATE_EXT_ROOT}
+            LOCAL_EXT_ROOT: {LOCAL_EXT_ROOT}
+            PUB_LIB_ROOT: {PUB_LIB_ROOT}
+            PRIVATE_LIB_ROOT: {PRIVATE_LIB_ROOT}
+            LOCAL_LIB_ROOT: {LOCAL_LIB_ROOT}
+            """
+    )
+
+    cleanup_files = []
+
+    # Upload several test_ext{i}.control files to the bucket
+    for i in range(NUM_EXT):
+        public_ext = BytesIO(bytes(control_file_content("public", i), "utf-8"))
+        public_remote_name = f"{bucket_prefix}/{PUB_EXT_ROOT}/test_ext{i}.control"
+        public_local_name = f"{LOCAL_EXT_ROOT}/test_ext{i}.control"
+        private_ext = BytesIO(bytes(control_file_content(str(tenant_id), i), "utf-8"))
+        private_remote_name = f"{bucket_prefix}/{PRIVATE_EXT_ROOT}/private_ext{i}.control"
+        private_local_name = f"{LOCAL_EXT_ROOT}/private_ext{i}.control"
+        cleanup_files += [public_local_name, private_local_name]
+
+        remote_storage_client.upload_fileobj(
+            public_ext, ext_remote_storage.bucket_name, public_remote_name
+        )
+        remote_storage_client.upload_fileobj(
+            private_ext, ext_remote_storage.bucket_name, private_remote_name
+        )
+
+    # Upload SQL file for the extension we're going to create
+    sql_filename = "test_ext0--1.0.sql"
+    test_sql_public_remote_path = f"{bucket_prefix}/{PUB_EXT_ROOT}/{sql_filename}"
+    test_sql_local_path = f"{LOCAL_EXT_ROOT}/{sql_filename}"
+    test_ext_sql_file = BytesIO(bytes(sql_file_content(), "utf-8"))
+    remote_storage_client.upload_fileobj(
+        test_ext_sql_file,
+        ext_remote_storage.bucket_name,
+        test_sql_public_remote_path,
+    )
+    cleanup_files += [test_sql_local_path]
+
+    # upload some fake library files
+    for i in range(2):
+        public_library = BytesIO(bytes("\n111\n", "utf-8"))
+        public_remote_name = f"{bucket_prefix}/{PUB_LIB_ROOT}/test_lib{i}.so"
+        public_local_name = f"{LOCAL_LIB_ROOT}/test_lib{i}.so"
+        private_library = BytesIO(bytes("\n111\n", "utf-8"))
+        private_remote_name = f"{bucket_prefix}/{PRIVATE_LIB_ROOT}/private_lib{i}.so"
+        private_local_name = f"{LOCAL_LIB_ROOT}/private_lib{i}.so"
+
+        log.info(f"uploading library to {public_remote_name}")
+        log.info(f"uploading library to {private_remote_name}")
+
+        remote_storage_client.upload_fileobj(
+            public_library,
+            ext_remote_storage.bucket_name,
+            public_remote_name,
+        )
+        remote_storage_client.upload_fileobj(
+            private_library,
+            ext_remote_storage.bucket_name,
+            private_remote_name,
+        )
+        cleanup_files += [public_local_name, private_local_name]
+
+    return cleanup_files
+
+
+# Generate mock extension files and upload them to the bucket.
+#
+# Then check that compute nodes can download them and use them
+# to CREATE EXTENSION and LOAD 'library.so'
+#
+# NOTE: You must have appropriate AWS credentials to run REAL_S3 test.
+# It may also be necessary to set the following environment variables:
+#   export AWS_ACCESS_KEY_ID='test'
+#   export AWS_SECRET_ACCESS_KEY='test'
+#   export AWS_SECURITY_TOKEN='test'
+#   export AWS_SESSION_TOKEN='test'
+#   export AWS_DEFAULT_REGION='us-east-1'
+
+
+@pytest.mark.parametrize(
+    "remote_storage_kind", [RemoteStorageKind.MOCK_S3, RemoteStorageKind.REAL_S3]
 )
-
-
-def test_file_download(neon_env_builder: NeonEnvBuilder):
-    """
-    Tests we can download a file
-    First we set up the mock s3 bucket by uploading test_ext.control to the bucket
-    Then, we download test_ext.control from the bucket to pg_install/v15/share/postgresql/extension/
-    Finally, we list available extensions and assert that test_ext is present
-    """
+def test_remote_extensions(
+    neon_env_builder: NeonEnvBuilder,
+    remote_storage_kind: RemoteStorageKind,
+    pg_version: PgVersion,
+    pg_bin: PgBin,
+):
     neon_env_builder.enable_remote_storage(
-        remote_storage_kind=RemoteStorageKind.MOCK_S3,
-        test_name="test_file_download",
+        remote_storage_kind=remote_storage_kind,
+        test_name="test_remote_extensions",
         enable_remote_extensions=True,
     )
     neon_env_builder.num_safekeepers = 3
     env = neon_env_builder.init_start()
+    tenant_id, _ = env.neon_cli.create_tenant()
+    env.neon_cli.create_timeline("test_remote_extensions", tenant_id=tenant_id)
 
     assert env.ext_remote_storage is not None
     assert env.remote_storage_client is not None
 
-    TEST_EXT_PATH = "v14/share/postgresql/extension/test_ext.control"
-    BUCKET_PREFIX = "5314225671"  # this is the build number
-
-    # 4. Upload test_ext.control file to the bucket
-    # In the non-mock version this is done by CI/CD
-
-    test_ext_file = BytesIO(
-        b"""# mock extension
-comment = 'This is a mock extension'
-default_version = '1.0'
-module_pathname = '$libdir/test_ext'
-relocatable = true
-    """
-    )
-    env.remote_storage_client.upload_fileobj(
-        test_ext_file,
-        env.ext_remote_storage.bucket_name,
-        os.path.join(BUCKET_PREFIX, TEST_EXT_PATH),
+    # Prepare some mock extension files and upload them to the bucket
+    cleanup_files = prepare_mock_ext_storage(
+        pg_version,
+        tenant_id,
+        pg_bin,
+        env.ext_remote_storage,
+        env.remote_storage_client,
     )
+    # Start a compute node and check that it can download the extensions
+    # and use them to CREATE EXTENSION and LOAD 'library.so'
+    #
+    # This block is wrapped in a try/finally so that the downloaded files
+    # are cleaned up even if the test fails
+    try:
+        endpoint = env.endpoints.create_start(
+            "test_remote_extensions",
+            tenant_id=tenant_id,
+            remote_ext_config=env.ext_remote_storage.to_string(),
+            # config_lines=["log_min_messages=debug3"],
+        )
+        with closing(endpoint.connect()) as conn:
+            with conn.cursor() as cur:
+                # Test query: check that test_ext0 was successfully downloaded
+                cur.execute("SELECT * FROM pg_available_extensions")
+                all_extensions = [x[0] for x in cur.fetchall()]
+                log.info(all_extensions)
+                for i in range(NUM_EXT):
+                    assert f"test_ext{i}" in all_extensions
+                    assert f"private_ext{i}" in all_extensions
 
-    # 5. Download file from the bucket to correct local location
-    # Later this will be replaced by our rust code
-    # resp = env.remote_storage_client.get_object(
-    #     Bucket=env.ext_remote_storage.bucket_name, Key=os.path.join(BUCKET_PREFIX, TEST_EXT_PATH)
-    # )
-    # response = resp["Body"]
-    # fname = f"pg_install/{TEST_EXT_PATH}"
-    # with open(fname, "wb") as f:
-    #     f.write(response.read())
+                cur.execute("CREATE EXTENSION test_ext0")
+                cur.execute("SELECT extname FROM pg_extension")
+                all_extensions = [x[0] for x in cur.fetchall()]
+                log.info(all_extensions)
+                assert "test_ext0" in all_extensions
 
-    tenant, _ = env.neon_cli.create_tenant()
-    env.neon_cli.create_timeline("test_file_download", tenant_id=tenant)
+                # Try to load existing library file
+                try:
+                    cur.execute("LOAD 'test_lib0.so'")
+                except Exception as e:
+                    # expected to fail with
+                    # could not load library ... test_ext.so: file too short
+                    # because test_lib0.so is not real library file
+                    log.info("LOAD test_lib0.so failed (expectedly): %s", e)
+                    assert "file too short" in str(e)
 
-    remote_ext_config = json.dumps(
-        {
-            "bucket": env.ext_remote_storage.bucket_name,
-            "region": "us-east-1",
-            "endpoint": env.ext_remote_storage.endpoint,
-            "prefix": BUCKET_PREFIX,
-        }
-    )
+                # Try to load private library file
+                try:
+                    cur.execute("LOAD 'private_lib0.so'")
+                except Exception as e:
+                    # expected to fail with
+                    # could not load library ... test_ext.so: file too short
+                    # because test_lib0.so is not real library file
+                    log.info("LOAD private_lib0.so failed (expectedly): %s", e)
+                    assert "file too short" in str(e)
 
-    # 6. Start endpoint and ensure that test_ext is present in select * from pg_available_extensions
-    endpoint = env.endpoints.create_start(
-        "test_file_download", tenant_id=tenant, remote_ext_config=remote_ext_config
-    )
-    with closing(endpoint.connect()) as conn:
-        with conn.cursor() as cur:
-            # test query: insert some values and select them
-            cur.execute("CREATE TABLE t(key int primary key, value text)")
-            for i in range(100):
-                cur.execute(f"insert into t values({i}, {2*i})")
-            cur.execute("select * from t")
-            log.info(cur.fetchall())
+                # Try to load existing library file without .so extension
+                try:
+                    cur.execute("LOAD 'test_lib1'")
+                except Exception as e:
+                    # expected to fail with
+                    # could not load library ... test_lib1.so: file too short
+                    # because test_lib1.so is not real library file
+                    log.info("LOAD test_lib1 failed (expectedly): %s", e)
+                    assert "file too short" in str(e)
 
-            # the real test query: check that test_ext is present
-            cur.execute("SELECT * FROM pg_available_extensions")
-            all_extensions = [x[0] for x in cur.fetchall()]
-            log.info(all_extensions)
-            assert "test_ext" in all_extensions
+                # Try to load non-existent library file
+                try:
+                    cur.execute("LOAD 'test_lib_fail.so'")
+                except Exception as e:
+                    # expected to fail because test_lib_fail.so is not found
+                    log.info("LOAD test_lib_fail.so failed (expectedly): %s", e)
+                    assert (
+                        """could not access file "test_lib_fail.so": No such file or directory"""
+                        in str(e)
+                    )
+
+    finally:
+        # this is important because if the files aren't cleaned up then the test can
+        # pass even without successfully downloading the files if a previous run (or
+        # run with different type of remote storage) of the test did download the
+        # files
+        for file in cleanup_files:
+            try:
+                os.remove(file)
+                log.info(f"Deleted {file}")
+            except FileNotFoundError:
+                log.info(f"{file} does not exist, so cannot be deleted")
+
+
+# TODO
+# @pytest.mark.parametrize("remote_storage_kind", available_remote_storages())
+# def test_remote_extensions_shared_preload_libraries(
+#     neon_env_builder: NeonEnvBuilder, remote_storage_kind: RemoteStorageKind, pg_version: PgVersion
+# ):

test_runner/regress/test_hot_standby.py

@@ -1,3 +1,5 @@
+import time
+
 import pytest
 from fixtures.neon_fixtures import NeonEnv
 
@@ -10,9 +12,10 @@ def test_hot_standby(neon_simple_env: NeonEnv):
         branch_name="main",
         endpoint_id="primary",
     ) as primary:
+        time.sleep(1)
         with env.endpoints.new_replica_start(origin=primary, endpoint_id="secondary") as secondary:
             primary_lsn = None
-            cought_up = False
+            caught_up = False
             queries = [
                 "SHOW neon.timeline_id",
                 "SHOW neon.tenant_id",
@@ -56,7 +59,7 @@ def test_hot_standby(neon_simple_env: NeonEnv):
                     res = s_cur.fetchone()
                     assert res is not None
 
-                while not cought_up:
+                while not caught_up:
                     with s_con.cursor() as secondary_cursor:
                         secondary_cursor.execute("SELECT pg_last_wal_replay_lsn()")
                         res = secondary_cursor.fetchone()
@@ -66,7 +69,7 @@ def test_hot_standby(neon_simple_env: NeonEnv):
                         # due to e.g. autovacuum, but that shouldn't impact the content
                         # of the tables, so we check whether we've replayed up to at
                         # least after the commit of the `test` table.
-                        cought_up = secondary_lsn >= primary_lsn
+                        caught_up = secondary_lsn >= primary_lsn
 
                 # Explicit commit to flush any transient transaction-level state.
                 s_con.commit()

test_runner/regress/test_tenant_size.py

@@ -16,6 +16,7 @@ from fixtures.pg_version import PgVersion, xfail_on_postgres
 from fixtures.types import Lsn, TenantId, TimelineId
 
 
+@pytest.mark.xfail
 def test_empty_tenant_size(neon_simple_env: NeonEnv, test_output_dir: Path):
     env = neon_simple_env
     (tenant_id, _) = env.neon_cli.create_tenant()
@@ -44,12 +45,16 @@ def test_empty_tenant_size(neon_simple_env: NeonEnv, test_output_dir: Path):
         # we've disabled the autovacuum and checkpoint
         # so background processes should not change the size.
         # If this test will flake we should probably loosen the check
-        assert size == initial_size, "starting idle compute should not change the tenant size"
+        assert (
+            size == initial_size
+        ), f"starting idle compute should not change the tenant size (Currently {size}, expected {initial_size})"
 
     # the size should be the same, until we increase the size over the
     # gc_horizon
     size, inputs = http_client.tenant_size_and_modelinputs(tenant_id)
-    assert size == initial_size, "tenant_size should not be affected by shutdown of compute"
+    assert (
+        size == initial_size
+    ), f"tenant_size should not be affected by shutdown of compute (Currently {size}, expected {initial_size})"
 
     expected_inputs = {
         "segments": [
@@ -318,6 +323,7 @@ def test_only_heads_within_horizon(neon_simple_env: NeonEnv, test_output_dir: Pa
     size_debug_file.write(size_debug)
 
 
+@pytest.mark.xfail
 def test_single_branch_get_tenant_size_grows(
     neon_env_builder: NeonEnvBuilder, test_output_dir: Path, pg_version: PgVersion
 ):
@@ -333,13 +339,13 @@ def test_single_branch_get_tenant_size_grows(
     # inserts is larger than gc_horizon. for example 0x20000 here hid the fact
     # that there next_gc_cutoff could be smaller than initdb_lsn, which will
     # obviously lead to issues when calculating the size.
-    gc_horizon = 0x38000
+    gc_horizon = 0x3BA00
 
     # it's a bit of a hack, but different versions of postgres have different
     # amount of WAL generated for the same amount of data. so we need to
     # adjust the gc_horizon accordingly.
     if pg_version == PgVersion.V14:
-        gc_horizon = 0x40000
+        gc_horizon = 0x4A000
 
     neon_env_builder.pageserver_config_override = f"tenant_config={{compaction_period='0s', gc_period='0s', pitr_interval='0sec', gc_horizon={gc_horizon}}}"
 
@@ -360,11 +366,11 @@ def test_single_branch_get_tenant_size_grows(
         if current_lsn - initdb_lsn >= gc_horizon:
             assert (
                 size >= prev_size
-            ), "tenant_size may grow or not grow, because we only add gc_horizon amount of WAL to initial snapshot size"
+            ), f"tenant_size may grow or not grow, because we only add gc_horizon amount of WAL to initial snapshot size (Currently at: {current_lsn}, Init at: {initdb_lsn})"
         else:
             assert (
                 size > prev_size
-            ), "tenant_size should grow, because we continue to add WAL to initial snapshot size"
+            ), f"tenant_size should grow, because we continue to add WAL to initial snapshot size (Currently at: {current_lsn}, Init at: {initdb_lsn})"
 
     def get_current_consistent_size(
         env: NeonEnv,