Experiment in decoding postgres tuples

compute_tools/src/bin/compute_ctl.rs

@@ -40,7 +40,7 @@ use std::sync::mpsc;
 use std::thread;
 use std::time::Duration;
 
-use anyhow::{Context, Result, bail};
+use anyhow::{Context, Result};
 use clap::Parser;
 use compute_api::responses::ComputeConfig;
 use compute_tools::compute::{
@@ -57,15 +57,31 @@ use tracing::{error, info};
 use url::Url;
 use utils::failpoint_support;
 
-#[derive(Debug, Parser)]
+// Compatibility hack: if the control plane specified any remote-ext-config
+// use the default value for extension storage proxy gateway.
+// Remove this once the control plane is updated to pass the gateway URL
+fn parse_remote_ext_base_url(arg: &str) -> Result<String> {
+    const FALLBACK_PG_EXT_GATEWAY_BASE_URL: &str =
+        "http://pg-ext-s3-gateway.pg-ext-s3-gateway.svc.cluster.local";
+
+    Ok(if arg.starts_with("http") {
+        arg
+    } else {
+        FALLBACK_PG_EXT_GATEWAY_BASE_URL
+    }
+    .to_owned())
+}
+
+#[derive(Parser)]
 #[command(rename_all = "kebab-case")]
 struct Cli {
     #[arg(short = 'b', long, default_value = "postgres", env = "POSTGRES_PATH")]
     pub pgbin: String,
 
     /// The base URL for the remote extension storage proxy gateway.
-    #[arg(short = 'r', long, value_parser = Self::parse_remote_ext_base_url)]
-    pub remote_ext_base_url: Option<Url>,
+    /// Should be in the form of `http(s)://<gateway-hostname>[:<port>]`.
+    #[arg(short = 'r', long, value_parser = parse_remote_ext_base_url, alias = "remote-ext-config")]
+    pub remote_ext_base_url: Option<String>,
 
     /// The port to bind the external listening HTTP server to. Clients running
     /// outside the compute will talk to the compute through this port. Keep
@@ -120,29 +136,6 @@ struct Cli {
         requires = "compute-id"
     )]
     pub control_plane_uri: Option<String>,
-
-    /// Interval in seconds for collecting installed extensions statistics
-    #[arg(long, default_value = "3600")]
-    pub installed_extensions_collection_interval: u64,
-}
-
-impl Cli {
-    /// Parse a URL from an argument. By default, this isn't necessary, but we
-    /// want to do some sanity checking.
-    fn parse_remote_ext_base_url(value: &str) -> Result<Url> {
-        // Remove extra trailing slashes, and add one. We use Url::join() later
-        // when downloading remote extensions. If the base URL is something like
-        // http://example.com/pg-ext-s3-gateway, and join() is called with
-        // something like "xyz", the resulting URL is http://example.com/xyz.
-        let value = value.trim_end_matches('/').to_owned() + "/";
-        let url = Url::parse(&value)?;
-
-        if url.query_pairs().count() != 0 {
-            bail!("parameters detected in remote extensions base URL")
-        }
-
-        Ok(url)
-    }
 }
 
 fn main() -> Result<()> {
@@ -186,7 +179,6 @@ fn main() -> Result<()> {
             cgroup: cli.cgroup,
             #[cfg(target_os = "linux")]
             vm_monitor_addr: cli.vm_monitor_addr,
-            installed_extensions_collection_interval: cli.installed_extensions_collection_interval,
         },
         config,
     )?;
@@ -271,8 +263,7 @@ fn handle_exit_signal(sig: i32) {
 
 #[cfg(test)]
 mod test {
-    use clap::{CommandFactory, Parser};
-    use url::Url;
+    use clap::CommandFactory;
 
     use super::Cli;
 
@@ -282,41 +273,16 @@ mod test {
     }
 
     #[test]
-    fn verify_remote_ext_base_url() {
-        let cli = Cli::parse_from([
-            "compute_ctl",
-            "--pgdata=test",
-            "--connstr=test",
-            "--compute-id=test",
-            "--remote-ext-base-url",
-            "https://example.com/subpath",
-        ]);
-        assert_eq!(
-            cli.remote_ext_base_url.unwrap(),
-            Url::parse("https://example.com/subpath/").unwrap()
-        );
+    fn parse_pg_ext_gateway_base_url() {
+        let arg = "http://pg-ext-s3-gateway2";
+        let result = super::parse_remote_ext_base_url(arg).unwrap();
+        assert_eq!(result, arg);
 
-        let cli = Cli::parse_from([
-            "compute_ctl",
-            "--pgdata=test",
-            "--connstr=test",
-            "--compute-id=test",
-            "--remote-ext-base-url",
-            "https://example.com//",
-        ]);
+        let arg = "pg-ext-s3-gateway";
+        let result = super::parse_remote_ext_base_url(arg).unwrap();
         assert_eq!(
-            cli.remote_ext_base_url.unwrap(),
-            Url::parse("https://example.com").unwrap()
+            result,
+            "http://pg-ext-s3-gateway.pg-ext-s3-gateway.svc.cluster.local"
         );
-
-        Cli::try_parse_from([
-            "compute_ctl",
-            "--pgdata=test",
-            "--connstr=test",
-            "--compute-id=test",
-            "--remote-ext-base-url",
-            "https://example.com?hello=world",
-        ])
-        .expect_err("URL parameters are not allowed");
     }
 }

compute_tools/src/bin/fast_import.rs

@@ -339,8 +339,6 @@ async fn run_dump_restore(
     destination_connstring: String,
 ) -> Result<(), anyhow::Error> {
     let dumpdir = workdir.join("dumpdir");
-    let num_jobs = num_cpus::get().to_string();
-    info!("using {num_jobs} jobs for dump/restore");
 
     let common_args = [
         // schema mapping (prob suffices to specify them on one side)
@@ -356,7 +354,7 @@ async fn run_dump_restore(
         "directory".to_string(),
         // concurrency
         "--jobs".to_string(),
-        num_jobs,
+        num_cpus::get().to_string(),
         // progress updates
         "--verbose".to_string(),
     ];

compute_tools/src/compute.rs

@@ -31,7 +31,6 @@ use std::time::{Duration, Instant};
 use std::{env, fs};
 use tokio::spawn;
 use tracing::{Instrument, debug, error, info, instrument, warn};
-use url::Url;
 use utils::id::{TenantId, TimelineId};
 use utils::lsn::Lsn;
 use utils::measured_stream::MeasuredReader;
@@ -97,10 +96,7 @@ pub struct ComputeNodeParams {
     pub internal_http_port: u16,
 
     /// the address of extension storage proxy gateway
-    pub remote_ext_base_url: Option<Url>,
-
-    /// Interval for installed extensions collection
-    pub installed_extensions_collection_interval: u64,
+    pub remote_ext_base_url: Option<String>,
 }
 
 /// Compute node info shared across several `compute_ctl` threads.
@@ -746,7 +742,17 @@ impl ComputeNode {
 
             let conf = self.get_tokio_conn_conf(None);
             tokio::task::spawn(async {
-                let _ = installed_extensions(conf).await;
+                let res = get_installed_extensions(conf).await;
+                match res {
+                    Ok(extensions) => {
+                        info!(
+                            "[NEON_EXT_STAT] {}",
+                            serde_json::to_string(&extensions)
+                                .expect("failed to serialize extensions list")
+                        );
+                    }
+                    Err(err) => error!("could not get installed extensions: {err:?}"),
+                }
             });
         }
 
@@ -776,9 +782,6 @@ impl ComputeNode {
         // Log metrics so that we can search for slow operations in logs
         info!(?metrics, postmaster_pid = %postmaster_pid, "compute start finished");
 
-        // Spawn the extension stats background task
-        self.spawn_extension_stats_task();
-
         if pspec.spec.prewarm_lfc_on_startup {
             self.prewarm_lfc();
         }
@@ -2189,41 +2192,6 @@ LIMIT 100",
             info!("Pageserver config changed");
         }
     }
-
-    pub fn spawn_extension_stats_task(&self) {
-        let conf = self.tokio_conn_conf.clone();
-        let installed_extensions_collection_interval =
-            self.params.installed_extensions_collection_interval;
-        tokio::spawn(async move {
-            // An initial sleep is added to ensure that two collections don't happen at the same time.
-            // The first collection happens during compute startup.
-            tokio::time::sleep(tokio::time::Duration::from_secs(
-                installed_extensions_collection_interval,
-            ))
-            .await;
-            let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(
-                installed_extensions_collection_interval,
-            ));
-            loop {
-                interval.tick().await;
-                let _ = installed_extensions(conf.clone()).await;
-            }
-        });
-    }
-}
-
-pub async fn installed_extensions(conf: tokio_postgres::Config) -> Result<()> {
-    let res = get_installed_extensions(conf).await;
-    match res {
-        Ok(extensions) => {
-            info!(
-                "[NEON_EXT_STAT] {}",
-                serde_json::to_string(&extensions).expect("failed to serialize extensions list")
-            );
-        }
-        Err(err) => error!("could not get installed extensions: {err:?}"),
-    }
-    Ok(())
 }
 
 pub fn forward_termination_signal() {

compute_tools/src/extension_server.rs

@@ -83,7 +83,6 @@ use reqwest::StatusCode;
 use tar::Archive;
 use tracing::info;
 use tracing::log::warn;
-use url::Url;
 use zstd::stream::read::Decoder;
 
 use crate::metrics::{REMOTE_EXT_REQUESTS_TOTAL, UNKNOWN_HTTP_STATUS};
@@ -159,7 +158,7 @@ fn parse_pg_version(human_version: &str) -> PostgresMajorVersion {
 pub async fn download_extension(
     ext_name: &str,
     ext_path: &RemotePath,
-    remote_ext_base_url: &Url,
+    remote_ext_base_url: &str,
     pgbin: &str,
 ) -> Result<u64> {
     info!("Download extension {:?} from {:?}", ext_name, ext_path);
@@ -271,14 +270,10 @@ pub fn create_control_files(remote_extensions: &RemoteExtSpec, pgbin: &str) {
 }
 
 // Do request to extension storage proxy, e.g.,
-// curl http://pg-ext-s3-gateway.pg-ext-s3-gateway.svc.cluster.local/latest/v15/extensions/anon.tar.zst
+// curl http://pg-ext-s3-gateway/latest/v15/extensions/anon.tar.zst
 // using HTTP GET and return the response body as bytes.
-async fn download_extension_tar(remote_ext_base_url: &Url, ext_path: &str) -> Result<Bytes> {
-    let uri = remote_ext_base_url.join(ext_path).with_context(|| {
-        format!(
-            "failed to create the remote extension URI for {ext_path} using {remote_ext_base_url}"
-        )
-    })?;
+async fn download_extension_tar(remote_ext_base_url: &str, ext_path: &str) -> Result<Bytes> {
+    let uri = format!("{}/{}", remote_ext_base_url, ext_path);
     let filename = Path::new(ext_path)
         .file_name()
         .unwrap_or_else(|| std::ffi::OsStr::new("unknown"))
@@ -288,7 +283,7 @@ async fn download_extension_tar(remote_ext_base_url: &Url, ext_path: &str) -> Re
 
     info!("Downloading extension file '{}' from uri {}", filename, uri);
 
-    match do_extension_server_request(uri).await {
+    match do_extension_server_request(&uri).await {
         Ok(resp) => {
             info!("Successfully downloaded remote extension data {}", ext_path);
             REMOTE_EXT_REQUESTS_TOTAL
@@ -307,7 +302,7 @@ async fn download_extension_tar(remote_ext_base_url: &Url, ext_path: &str) -> Re
 
 // Do a single remote extensions server request.
 // Return result or (error message + stringified status code) in case of any failures.
-async fn do_extension_server_request(uri: Url) -> Result<Bytes, (String, String)> {
+async fn do_extension_server_request(uri: &str) -> Result<Bytes, (String, String)> {
     let resp = reqwest::get(uri).await.map_err(|e| {
         (
             format!(

docker-compose/compute_wrapper/shell/compute.sh

@@ -20,7 +20,7 @@ first_path="$(ldconfig --verbose 2>/dev/null \
     | grep --invert-match ^$'\t' \
     | cut --delimiter=: --fields=1 \
     | head --lines=1)"
-test "$first_path" == '/usr/local/lib'
+test "$first_path" == '/usr/local/lib' || true # Remove the || true in a follow-up PR. Needed for backwards compat.
 
 echo "Waiting pageserver become ready."
 while ! nc -z pageserver 6400; do

libs/compute_api/src/spec.rs

@@ -250,44 +250,34 @@ impl RemoteExtSpec {
         }
 
         match self.extension_data.get(real_ext_name) {
-            Some(_ext_data) => Ok((
-                real_ext_name.to_string(),
-                Self::build_remote_path(build_tag, pg_major_version, real_ext_name)?,
-            )),
+            Some(_ext_data) => {
+                // We have decided to use the Go naming convention due to Kubernetes.
+
+                let arch = match std::env::consts::ARCH {
+                    "x86_64" => "amd64",
+                    "aarch64" => "arm64",
+                    arch => arch,
+                };
+
+                // Construct the path to the extension archive
+                // BUILD_TAG/PG_MAJOR_VERSION/extensions/EXTENSION_NAME.tar.zst
+                //
+                // Keep it in sync with path generation in
+                // https://github.com/neondatabase/build-custom-extensions/tree/main
+                let archive_path_str = format!(
+                    "{build_tag}/{arch}/{pg_major_version}/extensions/{real_ext_name}.tar.zst"
+                );
+                Ok((
+                    real_ext_name.to_string(),
+                    RemotePath::from_string(&archive_path_str)?,
+                ))
+            }
             None => Err(anyhow::anyhow!(
                 "real_ext_name {} is not found",
                 real_ext_name
             )),
         }
     }
-
-    /// Get the architecture-specific portion of the remote extension path. We
-    /// use the Go naming convention due to Kubernetes.
-    fn get_arch() -> &'static str {
-        match std::env::consts::ARCH {
-            "x86_64" => "amd64",
-            "aarch64" => "arm64",
-            arch => arch,
-        }
-    }
-
-    /// Build a [`RemotePath`] for an extension.
-    fn build_remote_path(
-        build_tag: &str,
-        pg_major_version: &str,
-        ext_name: &str,
-    ) -> anyhow::Result<RemotePath> {
-        let arch = Self::get_arch();
-
-        // Construct the path to the extension archive
-        // BUILD_TAG/PG_MAJOR_VERSION/extensions/EXTENSION_NAME.tar.zst
-        //
-        // Keep it in sync with path generation in
-        // https://github.com/neondatabase/build-custom-extensions/tree/main
-        RemotePath::from_string(&format!(
-            "{build_tag}/{arch}/{pg_major_version}/extensions/{ext_name}.tar.zst"
-        ))
-    }
 }
 
 #[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Deserialize, Serialize)]
@@ -528,37 +518,6 @@ mod tests {
             .expect("Library should be found");
     }
 
-    #[test]
-    fn remote_extension_path() {
-        let rspec: RemoteExtSpec = serde_json::from_value(serde_json::json!({
-            "public_extensions": ["ext"],
-            "custom_extensions": [],
-            "library_index": {
-                "extlib": "ext",
-            },
-            "extension_data": {
-                "ext": {
-                    "control_data": {
-                        "ext.control": ""
-                    },
-                    "archive_path": ""
-                }
-            },
-        }))
-        .unwrap();
-
-        let (_ext_name, ext_path) = rspec
-            .get_ext("ext", false, "latest", "v17")
-            .expect("Extension should be found");
-        // Starting with a forward slash would have consequences for the
-        // Url::join() that occurs when downloading a remote extension.
-        assert!(!ext_path.to_string().starts_with("/"));
-        assert_eq!(
-            ext_path,
-            RemoteExtSpec::build_remote_path("latest", "v17", "ext").unwrap()
-        );
-    }
-
     #[test]
     fn parse_spec_file() {
         let file = File::open("tests/cluster_spec.json").unwrap();

libs/pageserver_api/src/models.rs

@@ -354,9 +354,6 @@ pub struct ShardImportProgressV1 {
     pub completed: usize,
     /// Hash of the plan
     pub import_plan_hash: u64,
-    /// Soft limit for the job size
-    /// This needs to remain constant throughout the import
-    pub job_soft_size_limit: usize,
 }
 
 impl ShardImportStatus {

libs/postgres_ffi/bindgen_deps.h

@@ -12,3 +12,5 @@
 #include "storage/bufpage.h"
 #include "storage/off.h"
 #include "access/multixact.h"
+#include "postgres.h"
+#include "access/htup.h"

libs/postgres_ffi/build.rs

@@ -29,6 +29,11 @@ impl ParseCallbacks for PostgresFfiCallbacks {
             "CheckPoint",
             "FullTransactionId",
             "ControlFileData",
+            "PageXLogRecPtr",
+            "HeapTupleHeaderData",
+            "ItemPointerData",
+            "BlockIdData",
+            "OffsetNumber",
         ];
 
         if serde_list.contains(&derive_info.name) {
@@ -125,6 +130,11 @@ fn main() -> anyhow::Result<()> {
             .allowlist_var("PG_CONTROL_FILE_SIZE")
             .allowlist_var("PG_CONTROLFILEDATA_OFFSETOF_CRC")
             .allowlist_type("PageHeaderData")
+            .allowlist_type("HeapTupleHeaderData")
+            .allowlist_type("ItemPointerData")
+            .allowlist_type("TransactionId")
+            .allowlist_type("LocationIndex")
+            .allowlist_type("Datum")
             .allowlist_type("DBState")
             .allowlist_type("RelMapFile")
             .allowlist_type("RepOriginId")

libs/postgres_ffi/src/lib.rs

@@ -223,9 +223,10 @@ pub mod walrecord;
 
 // Export some widely used datatypes that are unlikely to change across Postgres versions
 pub use v14::bindings::{
-    BlockNumber, CheckPoint, ControlFileData, MultiXactId, OffsetNumber, Oid, PageHeaderData,
-    RepOriginId, TimeLineID, TimestampTz, TransactionId, XLogRecPtr, XLogRecord, XLogSegNo, uint32,
-    uint64,
+    BlockNumber, CheckPoint, ControlFileData, HeapTupleHeaderData, ItemIdData, ItemPointerData,
+    LocationIndex, MultiXactId, OffsetNumber, Oid, PageHeaderData, PageXLogRecPtr, RepOriginId,
+    TimeLineID, TimestampTz, TransactionId, XLogRecPtr, XLogRecord, XLogSegNo, uint8, uint16,
+    uint32, uint64,
 };
 // Likewise for these, although the assumption that these don't change is a little more iffy.
 pub use v14::bindings::{MultiXactOffset, MultiXactStatus};

libs/posthog_client_lite/src/lib.rs

@@ -37,7 +37,7 @@ pub struct LocalEvaluationFlag {
 #[derive(Deserialize)]
 pub struct LocalEvaluationFlagFilters {
     groups: Vec<LocalEvaluationFlagFilterGroup>,
-    multivariate: Option<LocalEvaluationFlagMultivariate>,
+    multivariate: LocalEvaluationFlagMultivariate,
 }
 
 #[derive(Deserialize)]
@@ -254,7 +254,7 @@ impl FeatureStore {
         }
     }
 
-    /// Evaluate a multivariate feature flag. Returns an error if the flag is not available or if there are errors
+    /// Evaluate a multivariate feature flag. Returns `None` if the flag is not available or if there are errors
     /// during the evaluation.
     ///
     /// The parsing logic is as follows:
@@ -272,10 +272,6 @@ impl FeatureStore {
     /// Example: we have a multivariate flag with 3 groups of the configured global rollout percentage: A (10%), B (20%), C (70%).
     /// There is a single group with a condition that has a rollout percentage of 10% and it does not have a variant override.
     /// Then, we will have 1% of the users evaluated to A, 2% to B, and 7% to C.
-    ///
-    /// Error handling: the caller should inspect the error and decide the behavior when a feature flag
-    /// cannot be evaluated (i.e., default to false if it cannot be resolved). The error should *not* be
-    /// propagated beyond where the feature flag gets resolved.
     pub fn evaluate_multivariate(
         &self,
         flag_key: &str,
@@ -294,35 +290,6 @@ impl FeatureStore {
         )
     }
 
-    /// Evaluate a boolean feature flag. Returns  an error if the flag is not available or if there are errors
-    /// during the evaluation.
-    ///
-    /// The parsing logic is as follows:
-    ///
-    /// * Generate a consistent hash for the tenant-feature.
-    /// * Match each filter group.
-    ///   - If a group is matched, it will first determine whether the user is in the range of the rollout
-    ///     percentage.
-    ///   - If the hash falls within the group's rollout percentage, return true.
-    /// * Otherwise, continue with the next group until all groups are evaluated and no group is within the
-    ///   rollout percentage.
-    /// * If there are no matching groups, return an error.
-    ///
-    /// Returns `Ok(())` if the feature flag evaluates to true. In the future, it will return a payload.
-    ///
-    /// Error handling: the caller should inspect the error and decide the behavior when a feature flag
-    /// cannot be evaluated (i.e., default to false if it cannot be resolved). The error should *not* be
-    /// propagated beyond where the feature flag gets resolved.
-    pub fn evaluate_boolean(
-        &self,
-        flag_key: &str,
-        user_id: &str,
-        properties: &HashMap<String, PostHogFlagFilterPropertyValue>,
-    ) -> Result<(), PostHogEvaluationError> {
-        let hash_on_global_rollout_percentage = Self::consistent_hash(user_id, flag_key, "boolean");
-        self.evaluate_boolean_inner(flag_key, hash_on_global_rollout_percentage, properties)
-    }
-
     /// Evaluate a multivariate feature flag. Note that we directly take the mapped user ID
     /// (a consistent hash ranging from 0 to 1) so that it is easier to use it in the tests
     /// and avoid duplicate computations.
@@ -349,11 +316,6 @@ impl FeatureStore {
                     flag_key
                 )));
             }
-            let Some(ref multivariate) = flag_config.filters.multivariate else {
-                return Err(PostHogEvaluationError::Internal(format!(
-                    "No multivariate available, should use evaluate_boolean?: {flag_key}"
-                )));
-            };
             // TODO: sort the groups so that variant overrides always get evaluated first and it follows the PostHog
             // Python SDK behavior; for now we do not configure conditions without variant overrides in Neon so it
             // does not matter.
@@ -362,7 +324,7 @@ impl FeatureStore {
                     GroupEvaluationResult::MatchedAndOverride(variant) => return Ok(variant),
                     GroupEvaluationResult::MatchedAndEvaluate => {
                         let mut percentage = 0;
-                        for variant in &multivariate.variants {
+                        for variant in &flag_config.filters.multivariate.variants {
                             percentage += variant.rollout_percentage;
                             if self
                                 .evaluate_percentage(hash_on_global_rollout_percentage, percentage)
@@ -390,64 +352,6 @@ impl FeatureStore {
             )))
         }
     }
-
-    /// Evaluate a multivariate feature flag. Note that we directly take the mapped user ID
-    /// (a consistent hash ranging from 0 to 1) so that it is easier to use it in the tests
-    /// and avoid duplicate computations.
-    ///
-    /// Use a different consistent hash for evaluating the group rollout percentage.
-    /// The behavior: if the condition is set to rolling out to 10% of the users, and
-    /// we set the variant A to 20% in the global config, then 2% of the total users will
-    /// be evaluated to variant A.
-    ///
-    /// Note that the hash to determine group rollout percentage is shared across all groups. So if we have two
-    /// exactly-the-same conditions with 10% and 20% rollout percentage respectively, a total of 20% of the users
-    /// will be evaluated (versus 30% if group evaluation is done independently).
-    pub(crate) fn evaluate_boolean_inner(
-        &self,
-        flag_key: &str,
-        hash_on_global_rollout_percentage: f64,
-        properties: &HashMap<String, PostHogFlagFilterPropertyValue>,
-    ) -> Result<(), PostHogEvaluationError> {
-        if let Some(flag_config) = self.flags.get(flag_key) {
-            if !flag_config.active {
-                return Err(PostHogEvaluationError::NotAvailable(format!(
-                    "The feature flag is not active: {}",
-                    flag_key
-                )));
-            }
-            if flag_config.filters.multivariate.is_some() {
-                return Err(PostHogEvaluationError::Internal(format!(
-                    "This looks like a multivariate flag, should use evaluate_multivariate?: {flag_key}"
-                )));
-            };
-            // TODO: sort the groups so that variant overrides always get evaluated first and it follows the PostHog
-            // Python SDK behavior; for now we do not configure conditions without variant overrides in Neon so it
-            // does not matter.
-            for group in &flag_config.filters.groups {
-                match self.evaluate_group(group, hash_on_global_rollout_percentage, properties)? {
-                    GroupEvaluationResult::MatchedAndOverride(_) => {
-                        return Err(PostHogEvaluationError::Internal(format!(
-                            "Boolean flag cannot have overrides: {}",
-                            flag_key
-                        )));
-                    }
-                    GroupEvaluationResult::MatchedAndEvaluate => {
-                        return Ok(());
-                    }
-                    GroupEvaluationResult::Unmatched => continue,
-                }
-            }
-            // If no group is matched, the feature is not available, and up to the caller to decide what to do.
-            Err(PostHogEvaluationError::NoConditionGroupMatched)
-        } else {
-            // The feature flag is not available yet
-            Err(PostHogEvaluationError::NotAvailable(format!(
-                "Not found in the local evaluation spec: {}",
-                flag_key
-            )))
-        }
-    }
 }
 
 pub struct PostHogClientConfig {
@@ -565,162 +469,95 @@ mod tests {
 
     fn data() -> &'static str {
         r#"{
-  "flags": [
-    {
-      "id": 141807,
-      "team_id": 152860,
-      "name": "",
-      "key": "image-compaction-boundary",
-      "filters": {
-        "groups": [
-          {
-            "variant": null,
-            "properties": [
-              {
-                "key": "plan_type",
-                "type": "person",
-                "value": [
-                  "free"
-                ],
-                "operator": "exact"
-              }
+            "flags": [
+                {
+                    "id": 132794,
+                    "team_id": 152860,
+                    "name": "",
+                    "key": "gc-compaction",
+                    "filters": {
+                        "groups": [
+                            {
+                                "variant": "enabled-stage-2",
+                                "properties": [
+                                    {
+                                        "key": "plan_type",
+                                        "type": "person",
+                                        "value": [
+                                            "free"
+                                        ],
+                                        "operator": "exact"
+                                    },
+                                    {
+                                        "key": "pageserver_remote_size",
+                                        "type": "person",
+                                        "value": "10000000",
+                                        "operator": "lt"
+                                    }
+                                ],
+                                "rollout_percentage": 50
+                            },
+                            {
+                                "properties": [
+                                    {
+                                        "key": "plan_type",
+                                        "type": "person",
+                                        "value": [
+                                            "free"
+                                        ],
+                                        "operator": "exact"
+                                    },
+                                    {
+                                        "key": "pageserver_remote_size",
+                                        "type": "person",
+                                        "value": "10000000",
+                                        "operator": "lt"
+                                    }
+                                ],
+                                "rollout_percentage": 80
+                            }
+                        ],
+                        "payloads": {},
+                        "multivariate": {
+                            "variants": [
+                                {
+                                    "key": "disabled",
+                                    "name": "",
+                                    "rollout_percentage": 90
+                                },
+                                {
+                                    "key": "enabled-stage-1",
+                                    "name": "",
+                                    "rollout_percentage": 10
+                                },
+                                {
+                                    "key": "enabled-stage-2",
+                                    "name": "",
+                                    "rollout_percentage": 0
+                                },
+                                {
+                                    "key": "enabled-stage-3",
+                                    "name": "",
+                                    "rollout_percentage": 0
+                                },
+                                {
+                                    "key": "enabled",
+                                    "name": "",
+                                    "rollout_percentage": 0
+                                }
+                            ]
+                        }
+                    },
+                    "deleted": false,
+                    "active": true,
+                    "ensure_experience_continuity": false,
+                    "has_encrypted_payloads": false,
+                    "version": 6
+                }
             ],
-            "rollout_percentage": 40
-          },
-          {
-            "variant": null,
-            "properties": [],
-            "rollout_percentage": 10
-          }
-        ],
-        "payloads": {},
-        "multivariate": null
-      },
-      "deleted": false,
-      "active": true,
-      "ensure_experience_continuity": false,
-      "has_encrypted_payloads": false,
-      "version": 1
-    },
-    {
-      "id": 135586,
-      "team_id": 152860,
-      "name": "",
-      "key": "boolean-flag",
-      "filters": {
-        "groups": [
-          {
-            "variant": null,
-            "properties": [
-              {
-                "key": "plan_type",
-                "type": "person",
-                "value": [
-                  "free"
-                ],
-                "operator": "exact"
-              }
-            ],
-            "rollout_percentage": 47
-          }
-        ],
-        "payloads": {},
-        "multivariate": null
-      },
-      "deleted": false,
-      "active": true,
-      "ensure_experience_continuity": false,
-      "has_encrypted_payloads": false,
-      "version": 1
-    },
-    {
-      "id": 132794,
-      "team_id": 152860,
-      "name": "",
-      "key": "gc-compaction",
-      "filters": {
-        "groups": [
-          {
-            "variant": "enabled-stage-2",
-            "properties": [
-              {
-                "key": "plan_type",
-                "type": "person",
-                "value": [
-                  "free"
-                ],
-                "operator": "exact"
-              },
-              {
-                "key": "pageserver_remote_size",
-                "type": "person",
-                "value": "10000000",
-                "operator": "lt"
-              }
-            ],
-             "rollout_percentage": 50
-          },
-          {
-            "properties": [
-              {
-                "key": "plan_type",
-                "type": "person",
-                "value": [
-                  "free"
-                ],
-                "operator": "exact"
-              },
-              {
-                "key": "pageserver_remote_size",
-                "type": "person",
-                "value": "10000000",
-                "operator": "lt"
-              }
-            ],
-            "rollout_percentage": 80
-          }
-        ],
-        "payloads": {},
-        "multivariate": {
-          "variants": [
-            {
-              "key": "disabled",
-              "name": "",
-              "rollout_percentage": 90
-            },
-            {
-              "key": "enabled-stage-1",
-              "name": "",
-              "rollout_percentage": 10
-            },
-            {
-              "key": "enabled-stage-2",
-              "name": "",
-              "rollout_percentage": 0
-            },
-            {
-              "key": "enabled-stage-3",
-              "name": "",
-              "rollout_percentage": 0
-            },
-            {
-              "key": "enabled",
-              "name": "",
-              "rollout_percentage": 0
-            }
-          ]
-        }
-      },
-      "deleted": false,
-      "active": true,
-      "ensure_experience_continuity": false,
-      "has_encrypted_payloads": false,
-      "version": 7
-    }
-  ],
-  "group_type_mapping": {},
-  "cohorts": {}
-}"#
+            "group_type_mapping": {},
+            "cohorts": {}
+        }"#
     }
 
     #[test]
@@ -796,125 +633,4 @@ mod tests {
             Err(PostHogEvaluationError::NoConditionGroupMatched)
         ),);
     }
-
-    #[test]
-    fn evaluate_boolean_1() {
-        // The `boolean-flag` feature flag only has one group that matches on the free user.
-
-        let mut store = FeatureStore::new();
-        let response: LocalEvaluationResponse = serde_json::from_str(data()).unwrap();
-        store.set_flags(response.flags);
-
-        // This lacks the required properties and cannot be evaluated.
-        let variant = store.evaluate_boolean_inner("boolean-flag", 1.00, &HashMap::new());
-        assert!(matches!(
-            variant,
-            Err(PostHogEvaluationError::NotAvailable(_))
-        ),);
-
-        let properties_unmatched = HashMap::from([
-            (
-                "plan_type".to_string(),
-                PostHogFlagFilterPropertyValue::String("paid".to_string()),
-            ),
-            (
-                "pageserver_remote_size".to_string(),
-                PostHogFlagFilterPropertyValue::Number(1000.0),
-            ),
-        ]);
-
-        // This does not match any group so there will be an error.
-        let variant = store.evaluate_boolean_inner("boolean-flag", 1.00, &properties_unmatched);
-        assert!(matches!(
-            variant,
-            Err(PostHogEvaluationError::NoConditionGroupMatched)
-        ),);
-
-        let properties = HashMap::from([
-            (
-                "plan_type".to_string(),
-                PostHogFlagFilterPropertyValue::String("free".to_string()),
-            ),
-            (
-                "pageserver_remote_size".to_string(),
-                PostHogFlagFilterPropertyValue::Number(1000.0),
-            ),
-        ]);
-
-        // It matches the first group as 0.10 <= 0.50 and the properties are matched. Then it gets evaluated to the variant override.
-        let variant = store.evaluate_boolean_inner("boolean-flag", 0.10, &properties);
-        assert!(variant.is_ok());
-
-        // It matches the group conditions but not the group rollout percentage.
-        let variant = store.evaluate_boolean_inner("boolean-flag", 1.00, &properties);
-        assert!(matches!(
-            variant,
-            Err(PostHogEvaluationError::NoConditionGroupMatched)
-        ),);
-    }
-
-    #[test]
-    fn evaluate_boolean_2() {
-        // The `image-compaction-boundary` feature flag has one group that matches on the free user and a group that matches on all users.
-
-        let mut store = FeatureStore::new();
-        let response: LocalEvaluationResponse = serde_json::from_str(data()).unwrap();
-        store.set_flags(response.flags);
-
-        // This lacks the required properties and cannot be evaluated.
-        let variant =
-            store.evaluate_boolean_inner("image-compaction-boundary", 1.00, &HashMap::new());
-        assert!(matches!(
-            variant,
-            Err(PostHogEvaluationError::NotAvailable(_))
-        ),);
-
-        let properties_unmatched = HashMap::from([
-            (
-                "plan_type".to_string(),
-                PostHogFlagFilterPropertyValue::String("paid".to_string()),
-            ),
-            (
-                "pageserver_remote_size".to_string(),
-                PostHogFlagFilterPropertyValue::Number(1000.0),
-            ),
-        ]);
-
-        // This does not match the filtered group but the all user group.
-        let variant =
-            store.evaluate_boolean_inner("image-compaction-boundary", 1.00, &properties_unmatched);
-        assert!(matches!(
-            variant,
-            Err(PostHogEvaluationError::NoConditionGroupMatched)
-        ),);
-        let variant =
-            store.evaluate_boolean_inner("image-compaction-boundary", 0.05, &properties_unmatched);
-        assert!(variant.is_ok());
-
-        let properties = HashMap::from([
-            (
-                "plan_type".to_string(),
-                PostHogFlagFilterPropertyValue::String("free".to_string()),
-            ),
-            (
-                "pageserver_remote_size".to_string(),
-                PostHogFlagFilterPropertyValue::Number(1000.0),
-            ),
-        ]);
-
-        // It matches the first group as 0.30 <= 0.40 and the properties are matched. Then it gets evaluated to the variant override.
-        let variant = store.evaluate_boolean_inner("image-compaction-boundary", 0.30, &properties);
-        assert!(variant.is_ok());
-
-        // It matches the group conditions but not the group rollout percentage.
-        let variant = store.evaluate_boolean_inner("image-compaction-boundary", 1.00, &properties);
-        assert!(matches!(
-            variant,
-            Err(PostHogEvaluationError::NoConditionGroupMatched)
-        ),);
-
-        // It matches the second "all" group conditions.
-        let variant = store.evaluate_boolean_inner("image-compaction-boundary", 0.09, &properties);
-        assert!(variant.is_ok());
-    }
 }

pageserver/page_api/proto/page_service.proto

@@ -54,9 +54,9 @@ service PageService {
   // RPCs use regular unary requests, since they are not as frequent and
   // performance-critical, and this simplifies implementation.
   //
-  // NB: a gRPC status response (e.g. errors) will terminate the stream. The
-  // stream may be shared by multiple Postgres backends, so we avoid this by
-  // sending them as GetPageResponse.status_code instead.
+  // NB: a status response (e.g. errors) will terminate the stream. The stream
+  // may be shared by e.g. multiple Postgres backends, so we should avoid this.
+  // Most errors are therefore sent as GetPageResponse.status instead.
   rpc GetPages (stream GetPageRequest) returns (stream GetPageResponse);
 
   // Returns the size of a relation, as # of blocks.
@@ -159,8 +159,8 @@ message GetPageRequest {
 // A GetPageRequest class. Primarily intended for observability, but may also be
 // used for prioritization in the future.
 enum GetPageClass {
-  // Unknown class. For backwards compatibility: used when an older client version sends a class
-  // that a newer server version has removed.
+  // Unknown class. For forwards compatibility: used when the client sends a
+  // class that the server doesn't know about.
   GET_PAGE_CLASS_UNKNOWN = 0;
   // A normal request. This is the default.
   GET_PAGE_CLASS_NORMAL = 1;
@@ -180,37 +180,31 @@ message GetPageResponse {
   // The original request's ID.
   uint64 request_id = 1;
   // The response status code.
-  GetPageStatusCode status_code = 2;
+  GetPageStatus status = 2;
   // A string describing the status, if any.
   string reason = 3;
-  // The 8KB page images, in the same order as the request. Empty if status_code != OK.
+  // The 8KB page images, in the same order as the request. Empty if status != OK.
   repeated bytes page_image = 4;
 }
 
-// A GetPageResponse status code.
-//
-// These are effectively equivalent to gRPC statuses. However, we use a bidirectional stream
-// (potentially shared by many backends), and a gRPC status response would terminate the stream so
-// we send GetPageResponse messages with these codes instead.
-enum GetPageStatusCode {
-  // Unknown status. For forwards compatibility: used when an older client version receives a new
-  // status code from a newer server version.
-  GET_PAGE_STATUS_CODE_UNKNOWN = 0;
+// A GetPageResponse status code. Since we use a bidirectional stream, we don't
+// want to send errors as gRPC statuses, since this would terminate the stream.
+enum GetPageStatus {
+  // Unknown status. For forwards compatibility: used when the server sends a
+  // status code that the client doesn't know about.
+  GET_PAGE_STATUS_UNKNOWN = 0;
   // The request was successful.
-  GET_PAGE_STATUS_CODE_OK = 1;
+  GET_PAGE_STATUS_OK = 1;
   // The page did not exist. The tenant/timeline/shard has already been
   // validated during stream setup.
-  GET_PAGE_STATUS_CODE_NOT_FOUND = 2;
+  GET_PAGE_STATUS_NOT_FOUND = 2;
   // The request was invalid.
-  GET_PAGE_STATUS_CODE_INVALID_REQUEST = 3;
-  // The request failed due to an internal server error.
-  GET_PAGE_STATUS_CODE_INTERNAL_ERROR = 4;
+  GET_PAGE_STATUS_INVALID = 3;
   // The tenant is rate limited. Slow down and retry later.
-  GET_PAGE_STATUS_CODE_SLOW_DOWN = 5;
-  // NB: shutdown errors are emitted as a gRPC Unavailable status.
-  //
-  // TODO: consider adding a GET_PAGE_STATUS_CODE_LAYER_DOWNLOAD in the case of a layer download.
-  // This could free up the server task to process other requests while the download is in progress.
+  GET_PAGE_STATUS_SLOW_DOWN = 4;
+  // TODO: consider adding a GET_PAGE_STATUS_LAYER_DOWNLOAD in the case of a
+  // layer download. This could free up the server task to process other
+  // requests while the layer download is in progress.
 }
 
 // Fetches the size of a relation at a given LSN, as # of blocks. Only valid on

pageserver/page_api/src/model.rs

@@ -35,12 +35,6 @@ impl ProtocolError {
     }
 }
 
-impl From<ProtocolError> for tonic::Status {
-    fn from(err: ProtocolError) -> Self {
-        tonic::Status::invalid_argument(format!("{err}"))
-    }
-}
-
 /// The LSN a request should read at.
 #[derive(Clone, Copy, Debug)]
 pub struct ReadLsn {
@@ -334,7 +328,7 @@ pub type RequestID = u64;
 /// A GetPage request class.
 #[derive(Clone, Copy, Debug)]
 pub enum GetPageClass {
-    /// Unknown class. For backwards compatibility: used when an older client version sends a class
+    /// Unknown status. For backwards compatibility: used when an older client version sends a class
     /// that a newer server version has removed.
     Unknown,
     /// A normal request. This is the default.
@@ -392,7 +386,7 @@ pub struct GetPageResponse {
     /// The original request's ID.
     pub request_id: RequestID,
     /// The response status code.
-    pub status_code: GetPageStatusCode,
+    pub status: GetPageStatus,
     /// A string describing the status, if any.
     pub reason: Option<String>,
     /// The 8KB page images, in the same order as the request. Empty if status != OK.
@@ -403,7 +397,7 @@ impl From<proto::GetPageResponse> for GetPageResponse {
     fn from(pb: proto::GetPageResponse) -> Self {
         Self {
             request_id: pb.request_id,
-            status_code: pb.status_code.into(),
+            status: pb.status.into(),
             reason: Some(pb.reason).filter(|r| !r.is_empty()),
             page_images: pb.page_image.into(),
         }
@@ -414,20 +408,16 @@ impl From<GetPageResponse> for proto::GetPageResponse {
     fn from(response: GetPageResponse) -> Self {
         Self {
             request_id: response.request_id,
-            status_code: response.status_code.into(),
+            status: response.status.into(),
             reason: response.reason.unwrap_or_default(),
             page_image: response.page_images.into_vec(),
         }
     }
 }
 
-/// A GetPage response status code.
-///
-/// These are effectively equivalent to gRPC statuses. However, we use a bidirectional stream
-/// (potentially shared by many backends), and a gRPC status response would terminate the stream so
-/// we send GetPageResponse messages with these codes instead.
+/// A GetPage response status.
 #[derive(Clone, Copy, Debug)]
-pub enum GetPageStatusCode {
+pub enum GetPageStatus {
     /// Unknown status. For forwards compatibility: used when an older client version receives a new
     /// status code from a newer server version.
     Unknown,
@@ -437,50 +427,46 @@ pub enum GetPageStatusCode {
     /// setup.
     NotFound,
     /// The request was invalid.
-    InvalidRequest,
-    /// The request failed due to an internal server error.
-    InternalError,
+    Invalid,
     /// The tenant is rate limited. Slow down and retry later.
     SlowDown,
 }
 
-impl From<proto::GetPageStatusCode> for GetPageStatusCode {
-    fn from(pb: proto::GetPageStatusCode) -> Self {
+impl From<proto::GetPageStatus> for GetPageStatus {
+    fn from(pb: proto::GetPageStatus) -> Self {
         match pb {
-            proto::GetPageStatusCode::Unknown => Self::Unknown,
-            proto::GetPageStatusCode::Ok => Self::Ok,
-            proto::GetPageStatusCode::NotFound => Self::NotFound,
-            proto::GetPageStatusCode::InvalidRequest => Self::InvalidRequest,
-            proto::GetPageStatusCode::InternalError => Self::InternalError,
-            proto::GetPageStatusCode::SlowDown => Self::SlowDown,
+            proto::GetPageStatus::Unknown => Self::Unknown,
+            proto::GetPageStatus::Ok => Self::Ok,
+            proto::GetPageStatus::NotFound => Self::NotFound,
+            proto::GetPageStatus::Invalid => Self::Invalid,
+            proto::GetPageStatus::SlowDown => Self::SlowDown,
         }
     }
 }
 
-impl From<i32> for GetPageStatusCode {
-    fn from(status_code: i32) -> Self {
-        proto::GetPageStatusCode::try_from(status_code)
-            .unwrap_or(proto::GetPageStatusCode::Unknown)
+impl From<i32> for GetPageStatus {
+    fn from(status: i32) -> Self {
+        proto::GetPageStatus::try_from(status)
+            .unwrap_or(proto::GetPageStatus::Unknown)
             .into()
     }
 }
 
-impl From<GetPageStatusCode> for proto::GetPageStatusCode {
-    fn from(status_code: GetPageStatusCode) -> Self {
-        match status_code {
-            GetPageStatusCode::Unknown => Self::Unknown,
-            GetPageStatusCode::Ok => Self::Ok,
-            GetPageStatusCode::NotFound => Self::NotFound,
-            GetPageStatusCode::InvalidRequest => Self::InvalidRequest,
-            GetPageStatusCode::InternalError => Self::InternalError,
-            GetPageStatusCode::SlowDown => Self::SlowDown,
+impl From<GetPageStatus> for proto::GetPageStatus {
+    fn from(status: GetPageStatus) -> Self {
+        match status {
+            GetPageStatus::Unknown => Self::Unknown,
+            GetPageStatus::Ok => Self::Ok,
+            GetPageStatus::NotFound => Self::NotFound,
+            GetPageStatus::Invalid => Self::Invalid,
+            GetPageStatus::SlowDown => Self::SlowDown,
         }
     }
 }
 
-impl From<GetPageStatusCode> for i32 {
-    fn from(status_code: GetPageStatusCode) -> Self {
-        proto::GetPageStatusCode::from(status_code).into()
+impl From<GetPageStatus> for i32 {
+    fn from(status: GetPageStatus) -> Self {
+        proto::GetPageStatus::from(status).into()
     }
 }
 

pageserver/src/disk_usage_eviction_task.rs

@@ -837,30 +837,7 @@ async fn collect_eviction_candidates(
                 continue;
             }
             let info = tl.get_local_layers_for_disk_usage_eviction().await;
-            debug!(
-                tenant_id=%tl.tenant_shard_id.tenant_id,
-                shard_id=%tl.tenant_shard_id.shard_slug(),
-                timeline_id=%tl.timeline_id,
-                "timeline resident layers count: {}", info.resident_layers.len()
-            );
-
-            tenant_candidates.extend(info.resident_layers.into_iter());
-            max_layer_size = max_layer_size.max(info.max_layer_size.unwrap_or(0));
-
-            if cancel.is_cancelled() {
-                return Ok(EvictionCandidates::Cancelled);
-            }
-        }
-
-        // Also consider layers of timelines being imported for eviction
-        for tl in tenant.list_importing_timelines() {
-            let info = tl.timeline.get_local_layers_for_disk_usage_eviction().await;
-            debug!(
-                tenant_id=%tl.timeline.tenant_shard_id.tenant_id,
-                shard_id=%tl.timeline.tenant_shard_id.shard_slug(),
-                timeline_id=%tl.timeline.timeline_id,
-                "timeline resident layers count: {}", info.resident_layers.len()
-            );
+            debug!(tenant_id=%tl.tenant_shard_id.tenant_id, shard_id=%tl.tenant_shard_id.shard_slug(), timeline_id=%tl.timeline_id, "timeline resident layers count: {}", info.resident_layers.len());
 
             tenant_candidates.extend(info.resident_layers.into_iter());
             max_layer_size = max_layer_size.max(info.max_layer_size.unwrap_or(0));

pageserver/src/feature_resolver.rs

@@ -45,10 +45,6 @@ impl FeatureResolver {
     }
 
     /// Evaluate a multivariate feature flag. Currently, we do not support any properties.
-    ///
-    /// Error handling: the caller should inspect the error and decide the behavior when a feature flag
-    /// cannot be evaluated (i.e., default to false if it cannot be resolved). The error should *not* be
-    /// propagated beyond where the feature flag gets resolved.
     pub fn evaluate_multivariate(
         &self,
         flag_key: &str,
@@ -66,29 +62,4 @@ impl FeatureResolver {
             ))
         }
     }
-
-    /// Evaluate a boolean feature flag. Currently, we do not support any properties.
-    ///
-    /// Returns `Ok(())` if the flag is evaluated to true, otherwise returns an error.
-    ///
-    /// Error handling: the caller should inspect the error and decide the behavior when a feature flag
-    /// cannot be evaluated (i.e., default to false if it cannot be resolved). The error should *not* be
-    /// propagated beyond where the feature flag gets resolved.
-    pub fn evaluate_boolean(
-        &self,
-        flag_key: &str,
-        tenant_id: TenantId,
-    ) -> Result<(), PostHogEvaluationError> {
-        if let Some(inner) = &self.inner {
-            inner.feature_store().evaluate_boolean(
-                flag_key,
-                &tenant_id.to_string(),
-                &HashMap::new(),
-            )
-        } else {
-            Err(PostHogEvaluationError::NotAvailable(
-                "PostHog integration is not enabled".to_string(),
-            ))
-        }
-    }
 }

pageserver/src/http/openapi_spec.yml

@@ -353,33 +353,6 @@ paths:
         "200":
           description: OK
 
-  /v1/tenant/{tenant_shard_id}/timeline/{timeline_id}/mark_invisible:
-    parameters:
-      - name: tenant_shard_id
-        in: path
-        required: true
-        schema:
-          type: string
-      - name: timeline_id
-        in: path
-        required: true
-        schema:
-          type: string
-          format: hex
-    put:
-      requestBody:
-        content:
-          application/json:
-            schema:
-              type: object
-              properties:
-                is_visible:
-                  type: boolean
-                  default: false
-      responses:
-        "200":
-          description: OK
-
   /v1/tenant/{tenant_shard_id}/location_config:
     parameters:
       - name: tenant_shard_id

pageserver/src/page_service.rs

@@ -769,9 +769,6 @@ struct BatchedGetPageRequest {
     timer: SmgrOpTimer,
     lsn_range: LsnRange,
     ctx: RequestContext,
-    // If the request is perf enabled, this contains a context
-    // with a perf span tracking the time spent waiting for the executor.
-    batch_wait_ctx: Option<RequestContext>,
 }
 
 #[cfg(feature = "testing")]
@@ -784,7 +781,6 @@ struct BatchedTestRequest {
 /// so that we don't keep the [`Timeline::gate`] open while the batch
 /// is being built up inside the [`spsc_fold`] (pagestream pipelining).
 #[derive(IntoStaticStr)]
-#[allow(clippy::large_enum_variant)]
 enum BatchedFeMessage {
     Exists {
         span: Span,
@@ -1302,22 +1298,6 @@ impl PageServerHandler {
                     }
                 };
 
-                let batch_wait_ctx = if ctx.has_perf_span() {
-                    Some(
-                        RequestContextBuilder::from(&ctx)
-                            .perf_span(|crnt_perf_span| {
-                                info_span!(
-                                    target: PERF_TRACE_TARGET,
-                                    parent: crnt_perf_span,
-                                    "WAIT_EXECUTOR",
-                                )
-                            })
-                            .attached_child(),
-                    )
-                } else {
-                    None
-                };
-
                 BatchedFeMessage::GetPage {
                     span,
                     shard: shard.downgrade(),
@@ -1329,7 +1309,6 @@ impl PageServerHandler {
                             request_lsn: req.hdr.request_lsn
                         },
                         ctx,
-                        batch_wait_ctx,
                     }],
                     // The executor grabs the batch when it becomes idle.
                     // Hence, [`GetPageBatchBreakReason::ExecutorSteal`] is the
@@ -1485,7 +1464,7 @@ impl PageServerHandler {
             let mut flush_timers = Vec::with_capacity(handler_results.len());
             for handler_result in &mut handler_results {
                 let flush_timer = match handler_result {
-                    Ok((_response, timer, _ctx)) => Some(
+                    Ok((_, timer)) => Some(
                         timer
                             .observe_execution_end(flushing_start_time)
                             .expect("we are the first caller"),
@@ -1505,7 +1484,7 @@ impl PageServerHandler {
         // Some handler errors cause exit from pagestream protocol.
         // Other handler errors are sent back as an error message and we stay in pagestream protocol.
         for (handler_result, flushing_timer) in handler_results.into_iter().zip(flush_timers) {
-            let (response_msg, ctx) = match handler_result {
+            let response_msg = match handler_result {
                 Err(e) => match &e.err {
                     PageStreamError::Shutdown => {
                         // If we fail to fulfil a request during shutdown, which may be _because_ of
@@ -1530,30 +1509,15 @@ impl PageServerHandler {
                             error!("error reading relation or page version: {full:#}")
                         });
 
-                        (
-                            PagestreamBeMessage::Error(PagestreamErrorResponse {
-                                req: e.req,
-                                message: e.err.to_string(),
-                            }),
-                            None,
-                        )
+                        PagestreamBeMessage::Error(PagestreamErrorResponse {
+                            req: e.req,
+                            message: e.err.to_string(),
+                        })
                     }
                 },
-                Ok((response_msg, _op_timer_already_observed, ctx)) => (response_msg, Some(ctx)),
+                Ok((response_msg, _op_timer_already_observed)) => response_msg,
             };
 
-            let ctx = ctx.map(|req_ctx| {
-                RequestContextBuilder::from(&req_ctx)
-                    .perf_span(|crnt_perf_span| {
-                        info_span!(
-                            target: PERF_TRACE_TARGET,
-                            parent: crnt_perf_span,
-                            "FLUSH_RESPONSE",
-                        )
-                    })
-                    .attached_child()
-            });
-
             //
             // marshal & transmit response message
             //
@@ -1576,17 +1540,6 @@ impl PageServerHandler {
                 )),
                 None => futures::future::Either::Right(flush_fut),
             };
-
-            let flush_fut = if let Some(req_ctx) = ctx.as_ref() {
-                futures::future::Either::Left(
-                    flush_fut.maybe_perf_instrument(req_ctx, |current_perf_span| {
-                        current_perf_span.clone()
-                    }),
-                )
-            } else {
-                futures::future::Either::Right(flush_fut)
-            };
-
             // do it while respecting cancellation
             let _: () = async move {
                 tokio::select! {
@@ -1616,7 +1569,7 @@ impl PageServerHandler {
         ctx: &RequestContext,
     ) -> Result<
         (
-            Vec<Result<(PagestreamBeMessage, SmgrOpTimer, RequestContext), BatchedPageStreamError>>,
+            Vec<Result<(PagestreamBeMessage, SmgrOpTimer), BatchedPageStreamError>>,
             Span,
         ),
         QueryError,
@@ -1643,7 +1596,7 @@ impl PageServerHandler {
                         self.handle_get_rel_exists_request(&shard, &req, &ctx)
                             .instrument(span.clone())
                             .await
-                            .map(|msg| (msg, timer, ctx))
+                            .map(|msg| (msg, timer))
                             .map_err(|err| BatchedPageStreamError { err, req: req.hdr }),
                     ],
                     span,
@@ -1662,7 +1615,7 @@ impl PageServerHandler {
                         self.handle_get_nblocks_request(&shard, &req, &ctx)
                             .instrument(span.clone())
                             .await
-                            .map(|msg| (msg, timer, ctx))
+                            .map(|msg| (msg, timer))
                             .map_err(|err| BatchedPageStreamError { err, req: req.hdr }),
                     ],
                     span,
@@ -1709,7 +1662,7 @@ impl PageServerHandler {
                         self.handle_db_size_request(&shard, &req, &ctx)
                             .instrument(span.clone())
                             .await
-                            .map(|msg| (msg, timer, ctx))
+                            .map(|msg| (msg, timer))
                             .map_err(|err| BatchedPageStreamError { err, req: req.hdr }),
                     ],
                     span,
@@ -1728,7 +1681,7 @@ impl PageServerHandler {
                         self.handle_get_slru_segment_request(&shard, &req, &ctx)
                             .instrument(span.clone())
                             .await
-                            .map(|msg| (msg, timer, ctx))
+                            .map(|msg| (msg, timer))
                             .map_err(|err| BatchedPageStreamError { err, req: req.hdr }),
                     ],
                     span,
@@ -2080,25 +2033,12 @@ impl PageServerHandler {
                             return Ok(());
                         }
                     };
-                    let mut batch = match batch {
+                    let batch = match batch {
                         Ok(batch) => batch,
                         Err(e) => {
                             return Err(e);
                         }
                     };
-
-                    if let BatchedFeMessage::GetPage {
-                        pages,
-                        span: _,
-                        shard: _,
-                        batch_break_reason: _,
-                    } = &mut batch
-                    {
-                        for req in pages {
-                            req.batch_wait_ctx.take();
-                        }
-                    }
-
                     self.pagestream_handle_batched_message(
                         pgb_writer,
                         batch,
@@ -2411,8 +2351,7 @@ impl PageServerHandler {
         io_concurrency: IoConcurrency,
         batch_break_reason: GetPageBatchBreakReason,
         ctx: &RequestContext,
-    ) -> Vec<Result<(PagestreamBeMessage, SmgrOpTimer, RequestContext), BatchedPageStreamError>>
-    {
+    ) -> Vec<Result<(PagestreamBeMessage, SmgrOpTimer), BatchedPageStreamError>> {
         debug_assert_current_span_has_tenant_and_timeline_id();
 
         timeline
@@ -2519,7 +2458,6 @@ impl PageServerHandler {
                                 page,
                             }),
                             req.timer,
-                            req.ctx,
                         )
                     })
                     .map_err(|e| BatchedPageStreamError {
@@ -2564,8 +2502,7 @@ impl PageServerHandler {
         timeline: &Timeline,
         requests: Vec<BatchedTestRequest>,
         _ctx: &RequestContext,
-    ) -> Vec<Result<(PagestreamBeMessage, SmgrOpTimer, RequestContext), BatchedPageStreamError>>
-    {
+    ) -> Vec<Result<(PagestreamBeMessage, SmgrOpTimer), BatchedPageStreamError>> {
         // real requests would do something with the timeline
         let mut results = Vec::with_capacity(requests.len());
         for _req in requests.iter() {
@@ -2592,10 +2529,6 @@ impl PageServerHandler {
                                 req: req.req.clone(),
                             }),
                             req.timer,
-                            RequestContext::new(
-                                TaskKind::PageRequestHandler,
-                                DownloadBehavior::Warn,
-                            ),
                         )
                     })
                     .map_err(|e| BatchedPageStreamError {

pageserver/src/tenant.rs

@@ -300,7 +300,7 @@ pub struct TenantShard {
     ///   as in progress.
     /// * Imported timelines are removed when the storage controller calls the post timeline
     ///   import activation endpoint.
-    timelines_importing: std::sync::Mutex<HashMap<TimelineId, Arc<ImportingTimeline>>>,
+    timelines_importing: std::sync::Mutex<HashMap<TimelineId, ImportingTimeline>>,
 
     /// The last tenant manifest known to be in remote storage. None if the manifest has not yet
     /// been either downloaded or uploaded. Always Some after tenant attach.
@@ -672,7 +672,6 @@ pub enum MaybeOffloaded {
 pub enum TimelineOrOffloaded {
     Timeline(Arc<Timeline>),
     Offloaded(Arc<OffloadedTimeline>),
-    Importing(Arc<ImportingTimeline>),
 }
 
 impl TimelineOrOffloaded {
@@ -684,9 +683,6 @@ impl TimelineOrOffloaded {
             TimelineOrOffloaded::Offloaded(offloaded) => {
                 TimelineOrOffloadedArcRef::Offloaded(offloaded)
             }
-            TimelineOrOffloaded::Importing(importing) => {
-                TimelineOrOffloadedArcRef::Importing(importing)
-            }
         }
     }
     pub fn tenant_shard_id(&self) -> TenantShardId {
@@ -699,16 +695,12 @@ impl TimelineOrOffloaded {
         match self {
             TimelineOrOffloaded::Timeline(timeline) => &timeline.delete_progress,
             TimelineOrOffloaded::Offloaded(offloaded) => &offloaded.delete_progress,
-            TimelineOrOffloaded::Importing(importing) => &importing.delete_progress,
         }
     }
     fn maybe_remote_client(&self) -> Option<Arc<RemoteTimelineClient>> {
         match self {
             TimelineOrOffloaded::Timeline(timeline) => Some(timeline.remote_client.clone()),
             TimelineOrOffloaded::Offloaded(_offloaded) => None,
-            TimelineOrOffloaded::Importing(importing) => {
-                Some(importing.timeline.remote_client.clone())
-            }
         }
     }
 }
@@ -716,7 +708,6 @@ impl TimelineOrOffloaded {
 pub enum TimelineOrOffloadedArcRef<'a> {
     Timeline(&'a Arc<Timeline>),
     Offloaded(&'a Arc<OffloadedTimeline>),
-    Importing(&'a Arc<ImportingTimeline>),
 }
 
 impl TimelineOrOffloadedArcRef<'_> {
@@ -724,14 +715,12 @@ impl TimelineOrOffloadedArcRef<'_> {
         match self {
             TimelineOrOffloadedArcRef::Timeline(timeline) => timeline.tenant_shard_id,
             TimelineOrOffloadedArcRef::Offloaded(offloaded) => offloaded.tenant_shard_id,
-            TimelineOrOffloadedArcRef::Importing(importing) => importing.timeline.tenant_shard_id,
         }
     }
     pub fn timeline_id(&self) -> TimelineId {
         match self {
             TimelineOrOffloadedArcRef::Timeline(timeline) => timeline.timeline_id,
             TimelineOrOffloadedArcRef::Offloaded(offloaded) => offloaded.timeline_id,
-            TimelineOrOffloadedArcRef::Importing(importing) => importing.timeline.timeline_id,
         }
     }
 }
@@ -748,12 +737,6 @@ impl<'a> From<&'a Arc<OffloadedTimeline>> for TimelineOrOffloadedArcRef<'a> {
     }
 }
 
-impl<'a> From<&'a Arc<ImportingTimeline>> for TimelineOrOffloadedArcRef<'a> {
-    fn from(timeline: &'a Arc<ImportingTimeline>) -> Self {
-        Self::Importing(timeline)
-    }
-}
-
 #[derive(Debug, thiserror::Error, PartialEq, Eq)]
 pub enum GetTimelineError {
     #[error("Timeline is shutting down")]
@@ -1806,25 +1789,20 @@ impl TenantShard {
                     },
                 ) => {
                     let timeline_id = timeline.timeline_id;
-                    let import_task_gate = Gate::default();
-                    let import_task_guard = import_task_gate.enter().unwrap();
                     let import_task_handle =
                         tokio::task::spawn(self.clone().create_timeline_import_pgdata_task(
                             timeline.clone(),
                             import_pgdata,
                             guard,
-                            import_task_guard,
                             ctx.detached_child(TaskKind::ImportPgdata, DownloadBehavior::Warn),
                         ));
 
                     let prev = self.timelines_importing.lock().unwrap().insert(
                         timeline_id,
-                        Arc::new(ImportingTimeline {
+                        ImportingTimeline {
                             timeline: timeline.clone(),
                             import_task_handle,
-                            import_task_gate,
-                            delete_progress: TimelineDeleteProgress::default(),
-                        }),
+                        },
                     );
 
                     assert!(prev.is_none());
@@ -2442,17 +2420,6 @@ impl TenantShard {
             .collect()
     }
 
-    /// Lists timelines the tenant contains.
-    /// It's up to callers to omit certain timelines that are not considered ready for use.
-    pub fn list_importing_timelines(&self) -> Vec<Arc<ImportingTimeline>> {
-        self.timelines_importing
-            .lock()
-            .unwrap()
-            .values()
-            .map(Arc::clone)
-            .collect()
-    }
-
     /// Lists timelines the tenant manages, including offloaded ones.
     ///
     /// It's up to callers to omit certain timelines that are not considered ready for use.
@@ -2886,25 +2853,19 @@ impl TenantShard {
 
         let (timeline, timeline_create_guard) = uninit_timeline.finish_creation_myself();
 
-        let import_task_gate = Gate::default();
-        let import_task_guard = import_task_gate.enter().unwrap();
-
         let import_task_handle = tokio::spawn(self.clone().create_timeline_import_pgdata_task(
             timeline.clone(),
             index_part,
             timeline_create_guard,
-            import_task_guard,
             timeline_ctx.detached_child(TaskKind::ImportPgdata, DownloadBehavior::Warn),
         ));
 
         let prev = self.timelines_importing.lock().unwrap().insert(
             timeline.timeline_id,
-            Arc::new(ImportingTimeline {
+            ImportingTimeline {
                 timeline: timeline.clone(),
                 import_task_handle,
-                import_task_gate,
-                delete_progress: TimelineDeleteProgress::default(),
-            }),
+            },
         );
 
         // Idempotency is enforced higher up the stack
@@ -2963,7 +2924,6 @@ impl TenantShard {
         timeline: Arc<Timeline>,
         index_part: import_pgdata::index_part_format::Root,
         timeline_create_guard: TimelineCreateGuard,
-        _import_task_guard: GateGuard,
         ctx: RequestContext,
     ) {
         debug_assert_current_span_has_tenant_and_timeline_id();
@@ -3875,9 +3835,6 @@ impl TenantShard {
                         .build_timeline_client(offloaded.timeline_id, self.remote_storage.clone());
                     Arc::new(remote_client)
                 }
-                TimelineOrOffloadedArcRef::Importing(_) => {
-                    unreachable!("Importing timelines are not included in the iterator")
-                }
             };
 
             // Shut down the timeline's remote client: this means that the indices we write
@@ -5087,14 +5044,6 @@ impl TenantShard {
                 info!("timeline already exists but is offloaded");
                 Err(CreateTimelineError::Conflict)
             }
-            Err(TimelineExclusionError::AlreadyExists {
-                existing: TimelineOrOffloaded::Importing(_existing),
-                ..
-            }) => {
-                // If there's a timeline already importing, then we would hit
-                // the [`TimelineExclusionError::AlreadyCreating`] branch above.
-                unreachable!("Importing timelines hold the creation guard")
-            }
             Err(TimelineExclusionError::AlreadyExists {
                 existing: TimelineOrOffloaded::Timeline(existing),
                 arg,
@@ -5366,7 +5315,6 @@ impl TenantShard {
             l0_compaction_trigger: self.l0_compaction_trigger.clone(),
             l0_flush_global_state: self.l0_flush_global_state.clone(),
             basebackup_prepare_sender: self.basebackup_prepare_sender.clone(),
-            feature_resolver: self.feature_resolver.clone(),
         }
     }
 
@@ -8411,24 +8359,10 @@ mod tests {
             }
 
             tline.freeze_and_flush().await?;
-            // Force layers to L1
-            tline
-                .compact(
-                    &cancel,
-                    {
-                        let mut flags = EnumSet::new();
-                        flags.insert(CompactFlags::ForceL0Compaction);
-                        flags
-                    },
-                    &ctx,
-                )
-                .await?;
 
             if iter % 5 == 0 {
-                let scan_lsn = Lsn(lsn.0 + 1);
-                info!("scanning at {}", scan_lsn);
                 let (_, before_delta_file_accessed) =
-                    scan_with_statistics(&tline, &keyspace, scan_lsn, &ctx, io_concurrency.clone())
+                    scan_with_statistics(&tline, &keyspace, lsn, &ctx, io_concurrency.clone())
                         .await?;
                 tline
                     .compact(
@@ -8437,14 +8371,13 @@ mod tests {
                             let mut flags = EnumSet::new();
                             flags.insert(CompactFlags::ForceImageLayerCreation);
                             flags.insert(CompactFlags::ForceRepartition);
-                            flags.insert(CompactFlags::ForceL0Compaction);
                             flags
                         },
                         &ctx,
                     )
                     .await?;
                 let (_, after_delta_file_accessed) =
-                    scan_with_statistics(&tline, &keyspace, scan_lsn, &ctx, io_concurrency.clone())
+                    scan_with_statistics(&tline, &keyspace, lsn, &ctx, io_concurrency.clone())
                         .await?;
                 assert!(
                     after_delta_file_accessed < before_delta_file_accessed,
@@ -8885,8 +8818,6 @@ mod tests {
 
         let cancel = CancellationToken::new();
 
-        // Image layer creation happens on the disk_consistent_lsn so we need to force set it now.
-        tline.force_set_disk_consistent_lsn(Lsn(0x40));
         tline
             .compact(
                 &cancel,
@@ -8900,7 +8831,8 @@ mod tests {
             )
             .await
             .unwrap();
-        // Image layers are created at repartition LSN
+
+        // Image layers are created at last_record_lsn
         let images = tline
             .inspect_image_layers(Lsn(0x40), &ctx, io_concurrency.clone())
             .await

pageserver/src/tenant/remote_timeline_client.rs

@@ -1348,21 +1348,6 @@ impl RemoteTimelineClient {
         Ok(())
     }
 
-    pub(crate) fn schedule_unlinking_of_layers_from_index_part<I>(
-        self: &Arc<Self>,
-        names: I,
-    ) -> Result<(), NotInitialized>
-    where
-        I: IntoIterator<Item = LayerName>,
-    {
-        let mut guard = self.upload_queue.lock().unwrap();
-        let upload_queue = guard.initialized_mut()?;
-
-        self.schedule_unlinking_of_layers_from_index_part0(upload_queue, names);
-
-        Ok(())
-    }
-
     /// Update the remote index file, removing the to-be-deleted files from the index,
     /// allowing scheduling of actual deletions later.
     fn schedule_unlinking_of_layers_from_index_part0<I>(

pageserver/src/tenant/timeline.rs

@@ -103,7 +103,6 @@ use crate::context::{
     DownloadBehavior, PerfInstrumentFutureExt, RequestContext, RequestContextBuilder,
 };
 use crate::disk_usage_eviction_task::{DiskUsageEvictionInfo, EvictionCandidate, finite_f32};
-use crate::feature_resolver::FeatureResolver;
 use crate::keyspace::{KeyPartitioning, KeySpace};
 use crate::l0_flush::{self, L0FlushGlobalState};
 use crate::metrics::{
@@ -199,7 +198,6 @@ pub struct TimelineResources {
     pub l0_compaction_trigger: Arc<Notify>,
     pub l0_flush_global_state: l0_flush::L0FlushGlobalState,
     pub basebackup_prepare_sender: BasebackupPrepareSender,
-    pub feature_resolver: FeatureResolver,
 }
 
 pub struct Timeline {
@@ -446,8 +444,6 @@ pub struct Timeline {
 
     /// A channel to send async requests to prepare a basebackup for the basebackup cache.
     basebackup_prepare_sender: BasebackupPrepareSender,
-
-    feature_resolver: FeatureResolver,
 }
 
 pub(crate) enum PreviousHeatmap {
@@ -3076,8 +3072,6 @@ impl Timeline {
                 wait_lsn_log_slow: tokio::sync::Semaphore::new(1),
 
                 basebackup_prepare_sender: resources.basebackup_prepare_sender,
-
-                feature_resolver: resources.feature_resolver,
             };
 
             result.repartition_threshold =
@@ -4912,7 +4906,6 @@ impl Timeline {
                     LastImageLayerCreationStatus::Initial,
                     false, // don't yield for L0, we're flushing L0
                 )
-                .instrument(info_span!("create_image_layers", mode = %ImageLayerCreationMode::Initial, partition_mode = "initial", lsn = %self.initdb_lsn))
                 .await?;
             debug_assert!(
                 matches!(is_complete, LastImageLayerCreationStatus::Complete),
@@ -5469,8 +5462,7 @@ impl Timeline {
 
     /// Returns the image layers generated and an enum indicating whether the process is fully completed.
     /// true = we have generate all image layers, false = we preempt the process for L0 compaction.
-    ///
-    /// `partition_mode` is only for logging purpose and is not used anywhere in this function.
+    #[tracing::instrument(skip_all, fields(%lsn, %mode))]
     async fn create_image_layers(
         self: &Arc<Timeline>,
         partitioning: &KeyPartitioning,
@@ -7528,11 +7520,15 @@ fn is_send() {
 
 #[cfg(test)]
 mod tests {
+    use std::io::Cursor;
+    use std::path::PathBuf;
     use std::sync::Arc;
 
     use pageserver_api::key::Key;
     use pageserver_api::value::Value;
+    use postgres_ffi::v17::bindings::PageHeaderData;
     use std::iter::Iterator;
+    use tokio_epoll_uring::BoundedBuf;
     use tracing::Instrument;
     use utils::id::TimelineId;
     use utils::lsn::Lsn;
@@ -7844,4 +7840,126 @@ mod tests {
 
         layers.get_from_desc(&desc)
     }
+
+    fn fixture_path(relative: &str) -> PathBuf {
+        PathBuf::from(env!("CARGO_MANIFEST_DIR")).join(relative)
+    }
+
+    use postgres_ffi::ItemIdData;
+    use postgres_ffi::LocationIndex;
+    use postgres_ffi::PageXLogRecPtr;
+    use postgres_ffi::TransactionId;
+    /// We already have a PageHeaderData via bindgen, but it's not serializable and
+    /// contains a non-serializable __IncompleteArrayField.
+    /// FIXME: we should be able to exclude pd_linp from the bindgen, but blocklist_item()
+    /// wasn't working for some reason.
+    use serde::{Deserialize, Serialize};
+
+    #[repr(C)]
+    #[derive(Debug, Default, Serialize, Deserialize)]
+    pub struct SerPageHeaderData {
+        pub pd_lsn: PageXLogRecPtr,
+        pub pd_checksum: postgres_ffi::uint16,
+        pub pd_flags: postgres_ffi::uint16,
+        pub pd_lower: LocationIndex,
+        pub pd_upper: LocationIndex,
+        pub pd_special: LocationIndex,
+        pub pd_pagesize_version: postgres_ffi::uint16,
+        pub pd_prune_xid: TransactionId,
+    }
+
+    #[repr(C)]
+    #[repr(align(4))]
+    #[derive(Debug, Copy, Clone, Deserialize)]
+    pub struct SerItemIdData {
+        pub raw: [u8; 4],
+    }
+
+    /// Postgres HeapTupleHeaderData type is actually some kind of placeholder,
+    /// so we have a hand-rolled struct to describe the actual serialized header of a tuple
+    #[repr(C)]
+    #[derive(Debug, Default, Serialize, Deserialize)]
+    struct SerMinimalTupleData {
+        t_xmin: postgres_ffi::uint32, /* actual length of minimal tuple */
+        t_xmax: postgres_ffi::uint32, /* actual length of minimal tuple */
+        t_cid_or_t_xvac: postgres_ffi::uint32,
+        t_ctid: postgres_ffi::ItemPointerData,
+        t_infomask2: postgres_ffi::uint16, /* number of attributes + various flags */
+        t_infomask: postgres_ffi::uint16,  /* various flag bits, see below */
+
+        t_hoff: postgres_ffi::uint8, /* sizeof header incl. bitmap, padding */
+
+                                     /* ^ - 23 bytes - ^ */
+    }
+
+    #[test]
+    fn decode_tuples() {
+        use utils::bin_ser::LeSer;
+
+        //assert_eq!(std::mem::size_of::<SerMinimalTupleData>(), 23);
+
+        // A simple example with three valid tuples, each tuple is (id, string, string, timestamp)
+        let input_raw = std::fs::read(fixture_path("test_data/example_page.bin")).unwrap();
+        // 00000000  00 00 00 00 a0 3d 53 01  00 00 00 00 24 00 70 1f  |.....=S.....$.p.|
+        // 00000010  00 20 04 20 00 00 00 00  d0 9f 60 00 a0 9f 60 00  |. . ......`...`.|
+        // 00000020  70 9f 60 00 00 00 00 00  00 00 00 00 00 00 00 00  |p.`.............|
+        // 00000030  00 00 00 00 00 00 00 00  00 00 00 00 00 00 00 00  |................|
+        // *
+        // 00001f70  e7 02 00 00 00 00 00 00  00 00 00 00 00 00 00 00  |................|
+        // 00001f80  03 00 04 00 02 08 18 00  03 00 00 00 0b 6b 65 79  |.............key|
+        // 00001f90  31 0f 76 61 6c 75 65 31  9f 2a e0 69 79 d8 02 00  |1.value1.*.iy...|
+        // 00001fa0  e6 02 00 00 00 00 00 00  00 00 00 00 00 00 00 00  |................|
+        // 00001fb0  02 00 04 00 02 08 18 00  02 00 00 00 0b 6b 65 79  |.............key|
+        // 00001fc0  33 0f 76 61 6c 75 65 33  54 b9 a3 69 79 d8 02 00  |3.value3T..iy...|
+        // 00001fd0  e5 02 00 00 00 00 00 00  00 00 00 00 00 00 00 00  |................|
+        // 00001fe0  01 00 04 00 02 08 18 00  01 00 00 00 0b 6b 65 79  |.............key|
+        // 00001ff0  32 0f 76 61 6c 75 65 32  dd 0e 6b 68 79 d8 02 00  |2.value2..khy...|
+        // 00002000
+
+        // The attributes for this example:
+        // attrelid |   attname   | atttypid | attlen | attnum | attcacheoff | atttypmod | attndims | attbyval | attalign | attstorage | attcompression | attnotnull | atthasdef | atthasmissing | attidentity | attgenerated | attisdropped | attislocal | attinhcount | attcollation | attstattarget | attacl | attoptions | attfdwoptions | attmissingval
+        // ----------+-------------+----------+--------+--------+-------------+-----------+----------+----------+----------+------------+----------------+------------+-----------+---------------+-------------+--------------+--------------+------------+-------------+--------------+---------------+--------+------------+---------------+---------------
+        //     16385 | id          |       23 |      4 |      1 |          -1 |        -1 |        0 | t        | i        | p          |                | t          | t         | f             |             |              | f            | t          |           0 |            0 |               |        |            |               |
+        //     16385 | key         |     1043 |     -1 |      2 |          -1 |       259 |        0 | f        | i        | x          |                | f          | f         | f             |             |              | f            | t          |           0 |          100 |               |        |            |               |
+        //     16385 | value       |       25 |     -1 |      3 |          -1 |        -1 |        0 | f        | i        | x          |                | f          | f         | f             |             |              | f            | t          |           0 |          100 |               |        |            |               |
+        //     16385 | modified_at |     1184 |      8 |      4 |          -1 |        -1 |        0 | t        | d        | p          |                | f          | t         | f             |             |              | f            | t          |           0 |            0 |               |        |            |               |
+
+        // TODO: test cases for schema changes:
+        // 1. baseline: data that exactly matches attributes
+        // 2. add one column, write a row, read all rows
+        // 3. drop one column, write a row, read all rows.
+        // 4. vacuum full, write a row, read all rows.
+
+        //        let mut cursor = Cursor::new(input_raw);
+        let header =
+            SerPageHeaderData::des(&input_raw[0..std::mem::size_of::<SerPageHeaderData>()])
+                .unwrap();
+
+        let item_count = (header.pd_lower as usize - std::mem::size_of::<SerPageHeaderData>())
+            / std::mem::size_of::<ItemIdData>();
+
+        assert_eq!(item_count, 3);
+
+        for item_counter in 0..item_count {
+            let item_base = std::mem::size_of::<SerPageHeaderData>()
+                + item_counter * std::mem::size_of::<ItemIdData>();
+            let item = SerItemIdData::des(
+                &input_raw[item_base..item_base + std::mem::size_of::<ItemIdData>()],
+            )
+            .unwrap();
+
+            let item = unsafe { std::mem::transmute::<[u8; 4], ItemIdData>(item.raw) };
+
+            println!("item: {} {}", item.lp_off(), item.lp_len());
+
+            let tuple_header_bytes =
+                &input_raw[item.lp_off() as usize..item.lp_off() as usize + 23];
+            let tuple_header = SerMinimalTupleData::des(tuple_header_bytes).unwrap();
+
+            println!("tuple_header: insert offset {}", tuple_header.t_hoff);
+
+            let tuple_raw = &input_raw[item.lp_off() as usize + tuple_header.t_hoff as usize
+                ..item.lp_off() as usize + item.lp_len() as usize];
+        }
+    }
 }

pageserver/src/tenant/timeline/compaction.rs

@@ -206,8 +206,8 @@ pub struct GcCompactionQueue {
 }
 
 static CONCURRENT_GC_COMPACTION_TASKS: Lazy<Arc<Semaphore>> = Lazy::new(|| {
-    // Only allow one timeline on one pageserver to run gc compaction at a time.
-    Arc::new(Semaphore::new(1))
+    // Only allow two timelines on one pageserver to run gc compaction at a time.
+    Arc::new(Semaphore::new(2))
 });
 
 impl GcCompactionQueue {
@@ -1278,55 +1278,11 @@ impl Timeline {
         }
 
         let gc_cutoff = *self.applied_gc_cutoff_lsn.read();
-        let l0_l1_boundary_lsn = {
-            // We do the repartition on the L0-L1 boundary. All data below the boundary
-            // are compacted by L0 with low read amplification, thus making the `repartition`
-            // function run fast.
-            let guard = self.layers.read().await;
-            guard
-                .all_persistent_layers()
-                .iter()
-                .map(|x| {
-                    // Use the end LSN of delta layers OR the start LSN of image layers.
-                    if x.is_delta {
-                        x.lsn_range.end
-                    } else {
-                        x.lsn_range.start
-                    }
-                })
-                .max()
-        };
-
-        let (partition_mode, partition_lsn) = if cfg!(test)
-            || cfg!(feature = "testing")
-            || self
-                .feature_resolver
-                .evaluate_boolean("image-compaction-boundary", self.tenant_shard_id.tenant_id)
-                .is_ok()
-        {
-            let last_repartition_lsn = self.partitioning.read().1;
-            let lsn = match l0_l1_boundary_lsn {
-                Some(boundary) => gc_cutoff
-                    .max(boundary)
-                    .max(last_repartition_lsn)
-                    .max(self.initdb_lsn)
-                    .max(self.ancestor_lsn),
-                None => self.get_last_record_lsn(),
-            };
-            if lsn <= self.initdb_lsn || lsn <= self.ancestor_lsn {
-                // Do not attempt to create image layers below the initdb or ancestor LSN -- no data below it
-                ("l0_l1_boundary", self.get_last_record_lsn())
-            } else {
-                ("l0_l1_boundary", lsn)
-            }
-        } else {
-            ("latest_record", self.get_last_record_lsn())
-        };
 
         // 2. Repartition and create image layers if necessary
         match self
             .repartition(
-                partition_lsn,
+                self.get_last_record_lsn(),
                 self.get_compaction_target_size(),
                 options.flags,
                 ctx,
@@ -1345,19 +1301,18 @@ impl Timeline {
                     .extend(sparse_partitioning.into_dense().parts);
 
                 // 3. Create new image layers for partitions that have been modified "enough".
-                let mode = if options
-                    .flags
-                    .contains(CompactFlags::ForceImageLayerCreation)
-                {
-                    ImageLayerCreationMode::Force
-                } else {
-                    ImageLayerCreationMode::Try
-                };
                 let (image_layers, outcome) = self
                     .create_image_layers(
                         &partitioning,
                         lsn,
-                        mode,
+                        if options
+                            .flags
+                            .contains(CompactFlags::ForceImageLayerCreation)
+                        {
+                            ImageLayerCreationMode::Force
+                        } else {
+                            ImageLayerCreationMode::Try
+                        },
                         &image_ctx,
                         self.last_image_layer_creation_status
                             .load()
@@ -1365,7 +1320,6 @@ impl Timeline {
                             .clone(),
                         options.flags.contains(CompactFlags::YieldForL0),
                     )
-                    .instrument(info_span!("create_image_layers", mode = %mode, partition_mode = %partition_mode, lsn = %lsn))
                     .await
                     .inspect_err(|err| {
                         if let CreateImageLayersError::GetVectoredError(
@@ -1390,8 +1344,7 @@ impl Timeline {
             }
 
             Ok(_) => {
-                // This happens very frequently so we don't want to log it.
-                debug!("skipping repartitioning due to image compaction LSN being below GC cutoff");
+                info!("skipping repartitioning due to image compaction LSN being below GC cutoff");
             }
 
             // Suppress errors when cancelled.

pageserver/src/tenant/timeline/delete.rs

@@ -121,7 +121,6 @@ async fn remove_maybe_offloaded_timeline_from_tenant(
     // This observes the locking order between timelines and timelines_offloaded
     let mut timelines = tenant.timelines.lock().unwrap();
     let mut timelines_offloaded = tenant.timelines_offloaded.lock().unwrap();
-    let mut timelines_importing = tenant.timelines_importing.lock().unwrap();
     let offloaded_children_exist = timelines_offloaded
         .iter()
         .any(|(_, entry)| entry.ancestor_timeline_id == Some(timeline.timeline_id()));
@@ -151,12 +150,8 @@ async fn remove_maybe_offloaded_timeline_from_tenant(
                 .expect("timeline that we were deleting was concurrently removed from 'timelines_offloaded' map");
             offloaded_timeline.delete_from_ancestor_with_timelines(&timelines);
         }
-        TimelineOrOffloaded::Importing(importing) => {
-            timelines_importing.remove(&importing.timeline.timeline_id);
-        }
     }
 
-    drop(timelines_importing);
     drop(timelines_offloaded);
     drop(timelines);
 
@@ -208,17 +203,8 @@ impl DeleteTimelineFlow {
         guard.mark_in_progress()?;
 
         // Now that the Timeline is in Stopping state, request all the related tasks to shut down.
-        // TODO(vlad): shut down imported timeline here
-        match &timeline {
-            TimelineOrOffloaded::Timeline(timeline) => {
-                timeline.shutdown(super::ShutdownMode::Hard).await;
-            }
-            TimelineOrOffloaded::Importing(importing) => {
-                importing.shutdown().await;
-            }
-            TimelineOrOffloaded::Offloaded(_offloaded) => {
-                // Nothing to shut down in this case
-            }
+        if let TimelineOrOffloaded::Timeline(timeline) = &timeline {
+            timeline.shutdown(super::ShutdownMode::Hard).await;
         }
 
         tenant.gc_block.before_delete(&timeline.timeline_id());
@@ -403,18 +389,10 @@ impl DeleteTimelineFlow {
             Err(anyhow::anyhow!("failpoint: timeline-delete-before-rm"))?
         });
 
-        match timeline {
-            TimelineOrOffloaded::Timeline(timeline) => {
-                delete_local_timeline_directory(conf, tenant.tenant_shard_id, timeline).await;
-            }
-            TimelineOrOffloaded::Importing(importing) => {
-                delete_local_timeline_directory(conf, tenant.tenant_shard_id, &importing.timeline)
-                    .await;
-            }
-            TimelineOrOffloaded::Offloaded(_offloaded) => {
-                // Offloaded timelines have no local state
-                // TODO: once we persist offloaded information, delete the timeline from there, too
-            }
+        // Offloaded timelines have no local state
+        // TODO: once we persist offloaded information, delete the timeline from there, too
+        if let TimelineOrOffloaded::Timeline(timeline) = timeline {
+            delete_local_timeline_directory(conf, tenant.tenant_shard_id, timeline).await;
         }
 
         fail::fail_point!("timeline-delete-after-rm", |_| {
@@ -473,16 +451,12 @@ pub(super) fn make_timeline_delete_guard(
     // For more context see this discussion: `https://github.com/neondatabase/neon/pull/4552#discussion_r1253437346`
     let timelines = tenant.timelines.lock().unwrap();
     let timelines_offloaded = tenant.timelines_offloaded.lock().unwrap();
-    let timelines_importing = tenant.timelines_importing.lock().unwrap();
 
     let timeline = match timelines.get(&timeline_id) {
         Some(t) => TimelineOrOffloaded::Timeline(Arc::clone(t)),
         None => match timelines_offloaded.get(&timeline_id) {
             Some(t) => TimelineOrOffloaded::Offloaded(Arc::clone(t)),
-            None => match timelines_importing.get(&timeline_id) {
-                Some(t) => TimelineOrOffloaded::Importing(Arc::clone(t)),
-                None => return Err(DeleteTimelineError::NotFound),
-            },
+            None => return Err(DeleteTimelineError::NotFound),
         },
     };
 

pageserver/src/tenant/timeline/import_pgdata.rs

@@ -8,10 +8,8 @@ use tokio::task::JoinHandle;
 use tokio_util::sync::CancellationToken;
 use tracing::info;
 use utils::lsn::Lsn;
-use utils::pausable_failpoint;
-use utils::sync::gate::Gate;
 
-use super::{Timeline, TimelineDeleteProgress};
+use super::Timeline;
 use crate::context::RequestContext;
 use crate::controller_upcall_client::{StorageControllerUpcallApi, StorageControllerUpcallClient};
 use crate::tenant::metadata::TimelineMetadata;
@@ -21,23 +19,15 @@ mod importbucket_client;
 mod importbucket_format;
 pub(crate) mod index_part_format;
 
-pub struct ImportingTimeline {
+pub(crate) struct ImportingTimeline {
     pub import_task_handle: JoinHandle<()>,
-    pub import_task_gate: Gate,
     pub timeline: Arc<Timeline>,
-    pub delete_progress: TimelineDeleteProgress,
-}
-
-impl std::fmt::Debug for ImportingTimeline {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "ImportingTimeline<{}>", self.timeline.timeline_id)
-    }
 }
 
 impl ImportingTimeline {
-    pub async fn shutdown(&self) {
+    pub(crate) async fn shutdown(self) {
         self.import_task_handle.abort();
-        self.import_task_gate.close().await;
+        let _ = self.import_task_handle.await;
 
         self.timeline.remote_client.shutdown().await;
     }
@@ -111,8 +101,6 @@ pub async fn doit(
                 .schedule_index_upload_for_file_changes()?;
             timeline.remote_client.wait_completion().await?;
 
-            pausable_failpoint!("import-timeline-pre-success-notify-pausable");
-
             // Communicate that shard is done.
             // Ensure at-least-once delivery of the upcall to storage controller
             // before we mark the task as done and never come here again.

pageserver/src/tenant/timeline/import_pgdata/flow.rs

@@ -30,7 +30,6 @@
 
 use std::collections::HashSet;
 use std::hash::{Hash, Hasher};
-use std::num::NonZeroUsize;
 use std::ops::Range;
 use std::sync::Arc;
 
@@ -101,24 +100,8 @@ async fn run_v1(
         tasks: Vec::default(),
     };
 
-    // Use the job size limit encoded in the progress if we are resuming an import.
-    // This ensures that imports have stable plans even if the pageserver config changes.
-    let import_config = {
-        match &import_progress {
-            Some(progress) => {
-                let base = &timeline.conf.timeline_import_config;
-                TimelineImportConfig {
-                    import_job_soft_size_limit: NonZeroUsize::new(progress.job_soft_size_limit)
-                        .unwrap(),
-                    import_job_concurrency: base.import_job_concurrency,
-                    import_job_checkpoint_threshold: base.import_job_checkpoint_threshold,
-                }
-            }
-            None => timeline.conf.timeline_import_config.clone(),
-        }
-    };
-
-    let plan = planner.plan(&import_config).await?;
+    let import_config = &timeline.conf.timeline_import_config;
+    let plan = planner.plan(import_config).await?;
 
     // Hash the plan and compare with the hash of the plan we got back from the storage controller.
     // If the two match, it means that the planning stage had the same output.
@@ -143,7 +126,7 @@ async fn run_v1(
     pausable_failpoint!("import-timeline-pre-execute-pausable");
 
     let start_from_job_idx = import_progress.map(|progress| progress.completed);
-    plan.execute(timeline, start_from_job_idx, plan_hash, &import_config, ctx)
+    plan.execute(timeline, start_from_job_idx, plan_hash, import_config, ctx)
         .await
 }
 
@@ -470,7 +453,6 @@ impl Plan {
                                     jobs: jobs_in_plan,
                                     completed: last_completed_job_idx,
                                     import_plan_hash,
-                                    job_soft_size_limit: import_config.import_job_soft_size_limit.into(),
                                 };
 
                                 timeline.remote_client.schedule_index_upload_for_file_changes()?;
@@ -982,15 +964,6 @@ impl ChunkProcessingJob {
             .cloned();
         match existing_layer {
             Some(existing) => {
-                // Unlink the remote layer from the index without scheduling its deletion.
-                // When `existing_layer` drops [`LayerInner::drop`] will schedule its deletion from
-                // remote storage, but that assumes that the layer was unlinked from the index first.
-                timeline
-                    .remote_client
-                    .schedule_unlinking_of_layers_from_index_part(std::iter::once(
-                        existing.layer_desc().layer_name(),
-                    ))?;
-
                 guard.open_mut()?.rewrite_layers(
                     &[(existing.clone(), resident_layer.clone())],
                     &[],

safekeeper/src/timelines_global_map.rs

@@ -44,7 +44,6 @@ struct GlobalTimelinesState {
     // on-demand timeline creation from recreating deleted timelines.  This is only soft-enforced, as
     // this map is dropped on restart.
     tombstones: HashMap<TenantTimelineId, Instant>,
-    tenant_tombstones: HashMap<TenantId, Instant>,
 
     conf: Arc<SafeKeeperConf>,
     broker_active_set: Arc<TimelinesSet>,
@@ -82,25 +81,10 @@ impl GlobalTimelinesState {
         }
     }
 
-    fn has_tombstone(&self, ttid: &TenantTimelineId) -> bool {
-        self.tombstones.contains_key(ttid) || self.tenant_tombstones.contains_key(&ttid.tenant_id)
-    }
-
-    /// Removes all blocking tombstones for the given timeline ID.
-    /// Returns `true` if there have been actual changes.
-    fn remove_tombstone(&mut self, ttid: &TenantTimelineId) -> bool {
-        self.tombstones.remove(ttid).is_some()
-            || self.tenant_tombstones.remove(&ttid.tenant_id).is_some()
-    }
-
     fn delete(&mut self, ttid: TenantTimelineId) {
         self.timelines.remove(&ttid);
         self.tombstones.insert(ttid, Instant::now());
     }
-
-    fn add_tenant_tombstone(&mut self, tenant_id: TenantId) {
-        self.tenant_tombstones.insert(tenant_id, Instant::now());
-    }
 }
 
 /// A struct used to manage access to the global timelines map.
@@ -115,7 +99,6 @@ impl GlobalTimelines {
             state: Mutex::new(GlobalTimelinesState {
                 timelines: HashMap::new(),
                 tombstones: HashMap::new(),
-                tenant_tombstones: HashMap::new(),
                 conf,
                 broker_active_set: Arc::new(TimelinesSet::default()),
                 global_rate_limiter: RateLimiter::new(1, 1),
@@ -262,7 +245,7 @@ impl GlobalTimelines {
                 return Ok(timeline);
             }
 
-            if state.has_tombstone(&ttid) {
+            if state.tombstones.contains_key(&ttid) {
                 anyhow::bail!("Timeline {ttid} is deleted, refusing to recreate");
             }
 
@@ -312,14 +295,13 @@ impl GlobalTimelines {
                 _ => {}
             }
             if check_tombstone {
-                if state.has_tombstone(&ttid) {
+                if state.tombstones.contains_key(&ttid) {
                     anyhow::bail!("timeline {ttid} is deleted, refusing to recreate");
                 }
             } else {
                 // We may be have been asked to load a timeline that was previously deleted (e.g. from `pull_timeline.rs`).  We trust
                 // that the human doing this manual intervention knows what they are doing, and remove its tombstone.
-                // It's also possible that we enter this when the tenant has been deleted, even if the timeline itself has never existed.
-                if state.remove_tombstone(&ttid) {
+                if state.tombstones.remove(&ttid).is_some() {
                     warn!("un-deleted timeline {ttid}");
                 }
             }
@@ -500,7 +482,6 @@ impl GlobalTimelines {
         let tli_res = {
             let state = self.state.lock().unwrap();
 
-            // Do NOT check tenant tombstones here: those were set earlier
             if state.tombstones.contains_key(ttid) {
                 // Presence of a tombstone guarantees that a previous deletion has completed and there is no work to do.
                 info!("Timeline {ttid} was already deleted");
@@ -576,10 +557,6 @@ impl GlobalTimelines {
         action: DeleteOrExclude,
     ) -> Result<HashMap<TenantTimelineId, TimelineDeleteResult>> {
         info!("deleting all timelines for tenant {}", tenant_id);
-
-        // Adding a tombstone before getting the timelines to prevent new timeline additions
-        self.state.lock().unwrap().add_tenant_tombstone(*tenant_id);
-
         let to_delete = self.get_all_for_tenant(*tenant_id);
 
         let mut err = None;
@@ -623,9 +600,6 @@ impl GlobalTimelines {
         state
             .tombstones
             .retain(|_, v| now.duration_since(*v) < *tombstone_ttl);
-        state
-            .tenant_tombstones
-            .retain(|_, v| now.duration_since(*v) < *tombstone_ttl);
     }
 }
 

storage_controller/src/http.rs

@@ -482,10 +482,6 @@ async fn handle_tenant_timeline_delete(
         ForwardOutcome::NotForwarded(_req) => {}
     };
 
-    service
-        .maybe_delete_timeline_import(tenant_id, timeline_id)
-        .await?;
-
     // For timeline deletions, which both implement an "initially return 202, then 404 once
     // we're done" semantic, we wrap with a retry loop to expose a simpler API upstream.
     async fn deletion_wrapper<R, F>(service: Arc<Service>, f: F) -> Result<Response<Body>, ApiError>

storage_controller/src/metrics.rs

@@ -139,14 +139,6 @@ pub(crate) struct StorageControllerMetricGroup {
     /// HTTP request status counters for handled requests
     pub(crate) storage_controller_reconcile_long_running:
         measured::CounterVec<ReconcileLongRunningLabelGroupSet>,
-
-    /// Indicator of safekeeper reconciler queue depth, broken down by safekeeper, excluding ongoing reconciles.
-    pub(crate) storage_controller_safkeeper_reconciles_queued:
-        measured::GaugeVec<SafekeeperReconcilerLabelGroupSet>,
-
-    /// Indicator of completed safekeeper reconciles, broken down by safekeeper.
-    pub(crate) storage_controller_safkeeper_reconciles_complete:
-        measured::CounterVec<SafekeeperReconcilerLabelGroupSet>,
 }
 
 impl StorageControllerMetrics {
@@ -265,17 +257,6 @@ pub(crate) enum Method {
     Other,
 }
 
-#[derive(measured::LabelGroup, Clone)]
-#[label(set = SafekeeperReconcilerLabelGroupSet)]
-pub(crate) struct SafekeeperReconcilerLabelGroup<'a> {
-    #[label(dynamic_with = lasso::ThreadedRodeo, default)]
-    pub(crate) sk_az: &'a str,
-    #[label(dynamic_with = lasso::ThreadedRodeo, default)]
-    pub(crate) sk_node_id: &'a str,
-    #[label(dynamic_with = lasso::ThreadedRodeo, default)]
-    pub(crate) sk_hostname: &'a str,
-}
-
 impl From<hyper::Method> for Method {
     fn from(value: hyper::Method) -> Self {
         if value == hyper::Method::GET {

storage_controller/src/service.rs

@@ -99,8 +99,8 @@ use crate::tenant_shard::{
     ScheduleOptimization, ScheduleOptimizationAction, TenantShard,
 };
 use crate::timeline_import::{
-    FinalizingImport, ImportResult, ShardImportStatuses, TimelineImport,
-    TimelineImportFinalizeError, TimelineImportState, UpcallClient,
+    ImportResult, ShardImportStatuses, TimelineImport, TimelineImportFinalizeError,
+    TimelineImportState, UpcallClient,
 };
 
 const WAITER_FILL_DRAIN_POLL_TIMEOUT: Duration = Duration::from_millis(500);
@@ -232,9 +232,6 @@ struct ServiceState {
 
     /// Queue of tenants who are waiting for concurrency limits to permit them to reconcile
     delayed_reconcile_rx: tokio::sync::mpsc::Receiver<TenantShardId>,
-
-    /// Tracks ongoing timeline import finalization tasks
-    imports_finalizing: BTreeMap<(TenantId, TimelineId), FinalizingImport>,
 }
 
 /// Transform an error from a pageserver into an error to return to callers of a storage
@@ -311,7 +308,6 @@ impl ServiceState {
             scheduler,
             ongoing_operation: None,
             delayed_reconcile_rx,
-            imports_finalizing: Default::default(),
         }
     }
 
@@ -4101,58 +4097,13 @@ impl Service {
     ///
     /// If this method gets pre-empted by shut down, it will be called again at start-up (on-going
     /// imports are stored in the database).
-    ///
-    /// # Cancel-Safety
-    /// Not cancel safe.
-    /// If the caller stops polling, the import will not be removed from
-    /// [`ServiceState::imports_finalizing`].
     #[instrument(skip_all, fields(
         tenant_id=%import.tenant_id,
         timeline_id=%import.timeline_id,
     ))]
-
     async fn finalize_timeline_import(
         self: &Arc<Self>,
         import: TimelineImport,
-    ) -> Result<(), TimelineImportFinalizeError> {
-        let tenant_timeline = (import.tenant_id, import.timeline_id);
-
-        let (_finalize_import_guard, cancel) = {
-            let mut locked = self.inner.write().unwrap();
-            let gate = Gate::default();
-            let cancel = CancellationToken::default();
-
-            let guard = gate.enter().unwrap();
-
-            locked.imports_finalizing.insert(
-                tenant_timeline,
-                FinalizingImport {
-                    gate,
-                    cancel: cancel.clone(),
-                },
-            );
-
-            (guard, cancel)
-        };
-
-        let res = tokio::select! {
-            res = self.finalize_timeline_import_impl(import) => {
-                res
-            },
-            _ = cancel.cancelled() => {
-                Err(TimelineImportFinalizeError::Cancelled)
-            }
-        };
-
-        let mut locked = self.inner.write().unwrap();
-        locked.imports_finalizing.remove(&tenant_timeline);
-
-        res
-    }
-
-    async fn finalize_timeline_import_impl(
-        self: &Arc<Self>,
-        import: TimelineImport,
     ) -> Result<(), TimelineImportFinalizeError> {
         tracing::info!("Finalizing timeline import");
 
@@ -4352,46 +4303,6 @@ impl Service {
         .await;
     }
 
-    /// Delete a timeline import if it exists
-    ///
-    /// Firstly, delete the entry from the database. Any updates
-    /// from pageservers after the update will fail with a 404, so the
-    /// import cannot progress into finalizing state if it's not there already.
-    /// Secondly, cancel the finalization if one is in progress.
-    pub(crate) async fn maybe_delete_timeline_import(
-        self: &Arc<Self>,
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-    ) -> Result<(), DatabaseError> {
-        let tenant_has_ongoing_import = {
-            let locked = self.inner.read().unwrap();
-            locked
-                .tenants
-                .range(TenantShardId::tenant_range(tenant_id))
-                .any(|(_tid, shard)| shard.importing == TimelineImportState::Importing)
-        };
-
-        if !tenant_has_ongoing_import {
-            return Ok(());
-        }
-
-        self.persistence
-            .delete_timeline_import(tenant_id, timeline_id)
-            .await?;
-
-        let maybe_finalizing = {
-            let mut locked = self.inner.write().unwrap();
-            locked.imports_finalizing.remove(&(tenant_id, timeline_id))
-        };
-
-        if let Some(finalizing) = maybe_finalizing {
-            finalizing.cancel.cancel();
-            finalizing.gate.close().await;
-        }
-
-        Ok(())
-    }
-
     pub(crate) async fn tenant_timeline_archival_config(
         &self,
         tenant_id: TenantId,
@@ -8627,9 +8538,8 @@ impl Service {
         Some(ShardCount(new_shard_count))
     }
 
-    /// Fetches the top tenant shards from every available node, in descending order of
-    /// max logical size. Offline nodes are skipped, and any errors from available nodes
-    /// will be logged and ignored.
+    /// Fetches the top tenant shards from every node, in descending order of
+    /// max logical size. Any node errors will be logged and ignored.
     async fn get_top_tenant_shards(
         &self,
         request: &TopTenantShardsRequest,
@@ -8640,7 +8550,6 @@ impl Service {
             .unwrap()
             .nodes
             .values()
-            .filter(|node| node.is_available())
             .cloned()
             .collect_vec();
 

storage_controller/src/service/safekeeper_reconciler.rs

@@ -20,9 +20,7 @@ use utils::{
 };
 
 use crate::{
-    metrics::{METRICS_REGISTRY, SafekeeperReconcilerLabelGroup},
-    persistence::SafekeeperTimelineOpKind,
-    safekeeper::Safekeeper,
+    persistence::SafekeeperTimelineOpKind, safekeeper::Safekeeper,
     safekeeper_client::SafekeeperClient,
 };
 
@@ -220,26 +218,7 @@ impl ReconcilerHandle {
     fn schedule_reconcile(&self, req: ScheduleRequest) {
         let (cancel, token_id) = self.new_token_slot(req.tenant_id, req.timeline_id);
         let hostname = req.safekeeper.skp.host.clone();
-        let sk_az = req.safekeeper.skp.availability_zone_id.clone();
-        let sk_node_id = req.safekeeper.get_id().to_string();
-
-        // We don't have direct access to the queue depth here, so increase it blindly by 1.
-        // We know that putting into the queue increases the queue depth. The receiver will
-        // update with the correct value once it processes the next item. To avoid races where we
-        // reduce before we increase, leaving the gauge with a 1 value for a long time, we
-        // increase it before putting into the queue.
-        let queued_gauge = &METRICS_REGISTRY
-            .metrics_group
-            .storage_controller_safkeeper_reconciles_queued;
-        let label_group = SafekeeperReconcilerLabelGroup {
-            sk_az: &sk_az,
-            sk_node_id: &sk_node_id,
-            sk_hostname: &hostname,
-        };
-        queued_gauge.inc(label_group.clone());
-
         if let Err(err) = self.tx.send((req, cancel, token_id)) {
-            queued_gauge.set(label_group, 0);
             tracing::info!("scheduling request onto {hostname} returned error: {err}");
         }
     }
@@ -304,18 +283,6 @@ impl SafekeeperReconciler {
                 continue;
             }
 
-            let queued_gauge = &METRICS_REGISTRY
-                .metrics_group
-                .storage_controller_safkeeper_reconciles_queued;
-            queued_gauge.set(
-                SafekeeperReconcilerLabelGroup {
-                    sk_az: &req.safekeeper.skp.availability_zone_id,
-                    sk_node_id: &req.safekeeper.get_id().to_string(),
-                    sk_hostname: &req.safekeeper.skp.host,
-                },
-                self.rx.len() as i64,
-            );
-
             tokio::task::spawn(async move {
                 let kind = req.kind;
                 let tenant_id = req.tenant_id;
@@ -544,16 +511,6 @@ impl SafekeeperReconcilerInner {
                             req.generation,
                         )
                         .await;
-
-                    let complete_counter = &METRICS_REGISTRY
-                        .metrics_group
-                        .storage_controller_safkeeper_reconciles_complete;
-                    complete_counter.inc(SafekeeperReconcilerLabelGroup {
-                        sk_az: &req.safekeeper.skp.availability_zone_id,
-                        sk_node_id: &req.safekeeper.get_id().to_string(),
-                        sk_hostname: &req.safekeeper.skp.host,
-                    });
-
                     if let Err(err) = res {
                         tracing::info!(
                             "couldn't remove reconciliation request onto {} from persistence: {err:?}",

storage_controller/src/timeline_import.rs

@@ -7,7 +7,6 @@ use serde::{Deserialize, Serialize};
 
 use pageserver_api::models::{ShardImportProgress, ShardImportStatus};
 use tokio_util::sync::CancellationToken;
-use utils::sync::gate::Gate;
 use utils::{
     id::{TenantId, TimelineId},
     shard::ShardIndex,
@@ -56,8 +55,6 @@ pub(crate) enum TimelineImportUpdateFollowUp {
 pub(crate) enum TimelineImportFinalizeError {
     #[error("Shut down interrupted import finalize")]
     ShuttingDown,
-    #[error("Import finalization was cancelled")]
-    Cancelled,
     #[error("Mismatched shard detected during import finalize: {0}")]
     MismatchedShards(ShardIndex),
 }
@@ -167,11 +164,6 @@ impl TimelineImport {
     }
 }
 
-pub(crate) struct FinalizingImport {
-    pub(crate) gate: Gate,
-    pub(crate) cancel: CancellationToken,
-}
-
 pub(crate) type ImportResult = Result<(), String>;
 
 pub(crate) struct UpcallClient {

test_runner/fixtures/fast_import.py

@@ -1,4 +1,3 @@
-import json
 import os
 import shutil
 import subprocess
@@ -12,7 +11,6 @@ from _pytest.config import Config
 
 from fixtures.log_helper import log
 from fixtures.neon_cli import AbstractNeonCli
-from fixtures.neon_fixtures import Endpoint, VanillaPostgres
 from fixtures.pg_version import PgVersion
 from fixtures.remote_storage import MockS3Server
 
@@ -163,57 +161,3 @@ def fast_import(
             f.write(fi.cmd.stderr)
 
         log.info("Written logs to %s", test_output_dir)
-
-
-def mock_import_bucket(vanilla_pg: VanillaPostgres, path: Path):
-    """
-    Mock the import S3 bucket into a local directory for a provided vanilla PG instance.
-    """
-    assert not vanilla_pg.is_running()
-
-    path.mkdir()
-    # what cplane writes before scheduling fast_import
-    specpath = path / "spec.json"
-    specpath.write_text(json.dumps({"branch_id": "somebranch", "project_id": "someproject"}))
-    # what fast_import writes
-    vanilla_pg.pgdatadir.rename(path / "pgdata")
-    statusdir = path / "status"
-    statusdir.mkdir()
-    (statusdir / "pgdata").write_text(json.dumps({"done": True}))
-    (statusdir / "fast_import").write_text(json.dumps({"command": "pgdata", "done": True}))
-
-
-def populate_vanilla_pg(vanilla_pg: VanillaPostgres, target_relblock_size: int) -> int:
-    assert vanilla_pg.is_running()
-
-    vanilla_pg.safe_psql("create user cloud_admin with password 'postgres' superuser")
-    # fillfactor so we don't need to produce that much data
-    # 900 byte per row is > 10% => 1 row per page
-    vanilla_pg.safe_psql("""create table t (data char(900)) with (fillfactor = 10)""")
-
-    nrows = 0
-    while True:
-        relblock_size = vanilla_pg.safe_psql_scalar("select pg_relation_size('t')")
-        log.info(
-            f"relblock size: {relblock_size / 8192} pages (target: {target_relblock_size // 8192}) pages"
-        )
-        if relblock_size >= target_relblock_size:
-            break
-        addrows = int((target_relblock_size - relblock_size) // 8192)
-        assert addrows >= 1, "forward progress"
-        vanilla_pg.safe_psql(
-            f"insert into t select generate_series({nrows + 1}, {nrows + addrows})"
-        )
-        nrows += addrows
-
-    return nrows
-
-
-def validate_import_from_vanilla_pg(endpoint: Endpoint, nrows: int):
-    assert endpoint.safe_psql_many(
-        [
-            "set effective_io_concurrency=32;",
-            "SET statement_timeout='300s';",
-            "select count(*), sum(data::bigint)::bigint from t",
-        ]
-    ) == [[], [], [(nrows, nrows * (nrows + 1) // 2)]]

test_runner/fixtures/neon_fixtures.py

@@ -2337,22 +2337,6 @@ class NeonStorageController(MetricsGetter, LogUtils):
             headers=self.headers(TokenScope.ADMIN),
         )
 
-    def import_status(
-        self, tenant_shard_id: TenantShardId, timeline_id: TimelineId, generation: int
-    ):
-        payload = {
-            "tenant_shard_id": str(tenant_shard_id),
-            "timeline_id": str(timeline_id),
-            "generation": generation,
-        }
-
-        self.request(
-            "GET",
-            f"{self.api}/upcall/v1/timeline_import_status",
-            headers=self.headers(TokenScope.GENERATIONS_API),
-            json=payload,
-        )
-
     def reconcile_all(self):
         r = self.request(
             "POST",
@@ -2829,11 +2813,6 @@ class NeonPageserver(PgProtocol, LogUtils):
         if self.running:
             self.http_client().configure_failpoints([(name, action)])
 
-    def clear_persistent_failpoint(self, name: str):
-        del self._persistent_failpoints[name]
-        if self.running:
-            self.http_client().configure_failpoints([(name, "off")])
-
     def timeline_dir(
         self,
         tenant_shard_id: TenantId | TenantShardId,

test_runner/fixtures/pageserver/http.py

@@ -675,7 +675,7 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
 
     def timeline_delete(
         self, tenant_id: TenantId | TenantShardId, timeline_id: TimelineId, **kwargs
-    ) -> int:
+    ):
         """
         Note that deletion is not instant, it is scheduled and performed mostly in the background.
         So if you need to wait for it to complete use `timeline_delete_wait_completed`.
@@ -688,8 +688,6 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
         res_json = res.json()
         assert res_json is None
 
-        return res.status_code
-
     def timeline_gc(
         self,
         tenant_id: TenantId | TenantShardId,

test_runner/regress/test_disk_usage_eviction.py

@@ -1,41 +1,31 @@
 from __future__ import annotations
 
 import enum
-import json
 import time
 from collections import Counter
 from dataclasses import dataclass
 from enum import StrEnum
-from threading import Event
 from typing import TYPE_CHECKING
 
 import pytest
 from fixtures.common_types import Lsn, TenantId, TimelineId
-from fixtures.fast_import import mock_import_bucket, populate_vanilla_pg
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import (
     NeonEnv,
     NeonEnvBuilder,
     NeonPageserver,
     PgBin,
-    VanillaPostgres,
     wait_for_last_flush_lsn,
 )
-from fixtures.pageserver.http import (
-    ImportPgdataIdemptencyKey,
-)
 from fixtures.pageserver.utils import wait_for_upload_queue_empty
 from fixtures.remote_storage import RemoteStorageKind
-from fixtures.utils import human_bytes, run_only_on_default_postgres, wait_until
-from werkzeug.wrappers.response import Response
+from fixtures.utils import human_bytes, wait_until
 
 if TYPE_CHECKING:
     from collections.abc import Iterable
     from typing import Any
 
     from fixtures.pageserver.http import PageserverHttpClient
-    from pytest_httpserver import HTTPServer
-    from werkzeug.wrappers.request import Request
 
 
 GLOBAL_LRU_LOG_LINE = "tenant_min_resident_size-respecting LRU would not relieve pressure, evicting more following global LRU policy"
@@ -174,7 +164,6 @@ class EvictionEnv:
         min_avail_bytes,
         mock_behavior,
         eviction_order: EvictionOrder,
-        wait_logical_size: bool = True,
     ):
         """
         Starts pageserver up with mocked statvfs setup. The startup is
@@ -212,12 +201,11 @@ class EvictionEnv:
         pageserver.start()
 
         # we now do initial logical size calculation on startup, which on debug builds can fight with disk usage based eviction
-        if wait_logical_size:
-            for tenant_id, timeline_id in self.timelines:
-                tenant_ps = self.neon_env.get_tenant_pageserver(tenant_id)
-                # Pageserver may be none if we are currently not attached anywhere, e.g. during secondary eviction test
-                if tenant_ps is not None:
-                    tenant_ps.http_client().timeline_wait_logical_size(tenant_id, timeline_id)
+        for tenant_id, timeline_id in self.timelines:
+            tenant_ps = self.neon_env.get_tenant_pageserver(tenant_id)
+            # Pageserver may be none if we are currently not attached anywhere, e.g. during secondary eviction test
+            if tenant_ps is not None:
+                tenant_ps.http_client().timeline_wait_logical_size(tenant_id, timeline_id)
 
         def statvfs_called():
             pageserver.assert_log_contains(".*running mocked statvfs.*")
@@ -894,121 +882,3 @@ def test_secondary_mode_eviction(eviction_env_ha: EvictionEnv):
     assert total_size - post_eviction_total_size >= evict_bytes, (
         "we requested at least evict_bytes worth of free space"
     )
-
-
-@run_only_on_default_postgres(reason="PG version is irrelevant here")
-def test_import_timeline_disk_pressure_eviction(
-    neon_env_builder: NeonEnvBuilder,
-    vanilla_pg: VanillaPostgres,
-    make_httpserver: HTTPServer,
-    pg_bin: PgBin,
-):
-    """
-    TODO
-    """
-    # Set up mock control plane HTTP server to listen for import completions
-    import_completion_signaled = Event()
-
-    def handler(request: Request) -> Response:
-        log.info(f"control plane /import_complete request: {request.json}")
-        import_completion_signaled.set()
-        return Response(json.dumps({}), status=200)
-
-    cplane_mgmt_api_server = make_httpserver
-    cplane_mgmt_api_server.expect_request(
-        "/storage/api/v1/import_complete", method="PUT"
-    ).respond_with_handler(handler)
-
-    # Plug the cplane mock in
-    neon_env_builder.control_plane_hooks_api = (
-        f"http://{cplane_mgmt_api_server.host}:{cplane_mgmt_api_server.port}/storage/api/v1/"
-    )
-
-    # The import will specifiy a local filesystem path mocking remote storage
-    neon_env_builder.enable_pageserver_remote_storage(RemoteStorageKind.LOCAL_FS)
-
-    vanilla_pg.start()
-    target_relblock_size = 1024 * 1024 * 128
-    populate_vanilla_pg(vanilla_pg, target_relblock_size)
-    vanilla_pg.stop()
-
-    env = neon_env_builder.init_configs()
-    env.start()
-
-    importbucket_path = neon_env_builder.repo_dir / "test_import_completion_bucket"
-    mock_import_bucket(vanilla_pg, importbucket_path)
-
-    tenant_id = TenantId.generate()
-    timeline_id = TimelineId.generate()
-    idempotency = ImportPgdataIdemptencyKey.random()
-
-    eviction_env = EvictionEnv(
-        timelines=[(tenant_id, timeline_id)],
-        neon_env=env,
-        pageserver_http=env.pageserver.http_client(),
-        layer_size=5 * 1024 * 1024,  # Doesn't apply here
-        pg_bin=pg_bin,  # Not used here
-        pgbench_init_lsns={},  # Not used here
-    )
-
-    # Pause before delivering the final notification to storcon.
-    # This keeps the import in progress.
-    failpoint_name = "import-timeline-pre-success-notify-pausable"
-    env.pageserver.add_persistent_failpoint(failpoint_name, "pause")
-
-    env.storage_controller.tenant_create(tenant_id)
-    env.storage_controller.timeline_create(
-        tenant_id,
-        {
-            "new_timeline_id": str(timeline_id),
-            "import_pgdata": {
-                "idempotency_key": str(idempotency),
-                "location": {"LocalFs": {"path": str(importbucket_path.absolute())}},
-            },
-        },
-    )
-
-    def hit_failpoint():
-        log.info("Checking log for pattern...")
-        try:
-            assert env.pageserver.log_contains(f".*at failpoint {failpoint_name}.*")
-        except Exception:
-            log.exception("Failed to find pattern in log")
-            raise
-
-    wait_until(hit_failpoint)
-    assert not import_completion_signaled.is_set()
-
-    env.pageserver.stop()
-
-    total_size, _, _ = eviction_env.timelines_du(env.pageserver)
-    blocksize = 512
-    total_blocks = (total_size + (blocksize - 1)) // blocksize
-
-    eviction_env.pageserver_start_with_disk_usage_eviction(
-        env.pageserver,
-        period="1s",
-        max_usage_pct=33,
-        min_avail_bytes=0,
-        mock_behavior={
-            "type": "Success",
-            "blocksize": blocksize,
-            "total_blocks": total_blocks,
-            # Only count layer files towards used bytes in the mock_statvfs.
-            # This avoids accounting for metadata files & tenant conf in the tests.
-            "name_filter": ".*__.*",
-        },
-        eviction_order=EvictionOrder.RELATIVE_ORDER_SPARE,
-        wait_logical_size=False,
-    )
-
-    wait_until(lambda: env.pageserver.assert_log_contains(".*disk usage pressure relieved"))
-
-    env.pageserver.clear_persistent_failpoint(failpoint_name)
-
-    def cplane_notified():
-        assert import_completion_signaled.is_set()
-
-    wait_until(cplane_notified)
-
-    env.pageserver.allowed_errors.append(r".* running disk usage based eviction due to pressure.*")

test_runner/regress/test_download_extensions.py

@@ -159,8 +159,7 @@ def test_remote_extensions(
 
     # Setup a mock nginx S3 gateway which will return our test extension.
     (host, port) = httpserver_listen_address
-    remote_ext_base_url = f"http://{host}:{port}/pg-ext-s3-gateway"
-    log.info(f"remote extensions base URL: {remote_ext_base_url}")
+    extensions_endpoint = f"http://{host}:{port}/pg-ext-s3-gateway"
 
     extension.build(pg_config, test_output_dir)
     tarball = extension.package(test_output_dir)
@@ -222,7 +221,7 @@ def test_remote_extensions(
 
     endpoint.create_remote_extension_spec(spec)
 
-    endpoint.start(remote_ext_base_url=remote_ext_base_url)
+    endpoint.start(remote_ext_base_url=extensions_endpoint)
 
     with endpoint.connect() as conn:
         with conn.cursor() as cur:
@@ -250,7 +249,7 @@ def test_remote_extensions(
     # Remove the extension files to force a redownload of the extension.
     extension.remove(test_output_dir, pg_version)
 
-    endpoint.start(remote_ext_base_url=remote_ext_base_url)
+    endpoint.start(remote_ext_base_url=extensions_endpoint)
 
     # Test that ALTER EXTENSION UPDATE statements also fetch remote extensions.
     with endpoint.connect() as conn:

test_runner/regress/test_import_pgdata.py

@@ -12,19 +12,13 @@ import psycopg2
 import psycopg2.errors
 import pytest
 from fixtures.common_types import Lsn, TenantId, TenantShardId, TimelineId
-from fixtures.fast_import import (
-    FastImport,
-    mock_import_bucket,
-    populate_vanilla_pg,
-    validate_import_from_vanilla_pg,
-)
+from fixtures.fast_import import FastImport
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import (
     NeonEnvBuilder,
     PageserverImportConfig,
     PgBin,
     PgProtocol,
-    StorageControllerApiException,
     StorageControllerMigrationConfig,
     VanillaPostgres,
 )
@@ -65,6 +59,24 @@ smoke_params = [
 ]
 
 
+def mock_import_bucket(vanilla_pg: VanillaPostgres, path: Path):
+    """
+    Mock the import S3 bucket into a local directory for a provided vanilla PG instance.
+    """
+    assert not vanilla_pg.is_running()
+
+    path.mkdir()
+    # what cplane writes before scheduling fast_import
+    specpath = path / "spec.json"
+    specpath.write_text(json.dumps({"branch_id": "somebranch", "project_id": "someproject"}))
+    # what fast_import writes
+    vanilla_pg.pgdatadir.rename(path / "pgdata")
+    statusdir = path / "status"
+    statusdir.mkdir()
+    (statusdir / "pgdata").write_text(json.dumps({"done": True}))
+    (statusdir / "fast_import").write_text(json.dumps({"command": "pgdata", "done": True}))
+
+
 @skip_in_debug_build("MULTIPLE_RELATION_SEGMENTS has non trivial amount of data")
 @pytest.mark.parametrize("shard_count,stripe_size,rel_block_size", smoke_params)
 def test_pgdata_import_smoke(
@@ -119,6 +131,10 @@ def test_pgdata_import_smoke(
     # Put data in vanilla pg
     #
 
+    vanilla_pg.start()
+    vanilla_pg.safe_psql("create user cloud_admin with password 'postgres' superuser")
+
+    log.info("create relblock data")
     if rel_block_size == RelBlockSize.ONE_STRIPE_SIZE:
         target_relblock_size = stripe_size * 8192
     elif rel_block_size == RelBlockSize.TWO_STRPES_PER_SHARD:
@@ -129,8 +145,45 @@ def test_pgdata_import_smoke(
     else:
         raise ValueError
 
-    vanilla_pg.start()
-    rows_inserted = populate_vanilla_pg(vanilla_pg, target_relblock_size)
+    # fillfactor so we don't need to produce that much data
+    # 900 byte per row is > 10% => 1 row per page
+    vanilla_pg.safe_psql("""create table t (data char(900)) with (fillfactor = 10)""")
+
+    nrows = 0
+    while True:
+        relblock_size = vanilla_pg.safe_psql_scalar("select pg_relation_size('t')")
+        log.info(
+            f"relblock size: {relblock_size / 8192} pages (target: {target_relblock_size // 8192}) pages"
+        )
+        if relblock_size >= target_relblock_size:
+            break
+        addrows = int((target_relblock_size - relblock_size) // 8192)
+        assert addrows >= 1, "forward progress"
+        vanilla_pg.safe_psql(
+            f"insert into t select generate_series({nrows + 1}, {nrows + addrows})"
+        )
+        nrows += addrows
+    expect_nrows = nrows
+    expect_sum = (
+        (nrows) * (nrows + 1) // 2
+    )  # https://stackoverflow.com/questions/43901484/sum-of-the-integers-from-1-to-n
+
+    def validate_vanilla_equivalence(ep):
+        # TODO: would be nicer to just compare pgdump
+
+        # Enable IO concurrency for batching on large sequential scan, to avoid making
+        # this test unnecessarily onerous on CPU. Especially on debug mode, it's still
+        # pretty onerous though, so increase statement_timeout to avoid timeouts.
+        assert ep.safe_psql_many(
+            [
+                "set effective_io_concurrency=32;",
+                "SET statement_timeout='300s';",
+                "select count(*), sum(data::bigint)::bigint from t",
+            ]
+        ) == [[], [], [(expect_nrows, expect_sum)]]
+
+    validate_vanilla_equivalence(vanilla_pg)
+
     vanilla_pg.stop()
 
     #
@@ -221,14 +274,14 @@ def test_pgdata_import_smoke(
         config_lines=ep_config,
     )
 
-    validate_import_from_vanilla_pg(ro_endpoint, rows_inserted)
+    validate_vanilla_equivalence(ro_endpoint)
 
     # ensure the import survives restarts
     ro_endpoint.stop()
     env.pageserver.stop(immediate=True)
     env.pageserver.start()
     ro_endpoint.start()
-    validate_import_from_vanilla_pg(ro_endpoint, rows_inserted)
+    validate_vanilla_equivalence(ro_endpoint)
 
     #
     # validate the layer files in each shard only have the shard-specific data
@@ -268,7 +321,7 @@ def test_pgdata_import_smoke(
     child_workload = workload.branch(timeline_id=child_timeline_id, branch_name="br-tip")
     child_workload.validate()
 
-    validate_import_from_vanilla_pg(child_workload.endpoint(), rows_inserted)
+    validate_vanilla_equivalence(child_workload.endpoint())
 
     # ... at the initdb lsn
     _ = env.create_branch(
@@ -283,7 +336,7 @@ def test_pgdata_import_smoke(
         tenant_id=tenant_id,
         config_lines=ep_config,
     )
-    validate_import_from_vanilla_pg(br_initdb_endpoint, rows_inserted)
+    validate_vanilla_equivalence(br_initdb_endpoint)
     with pytest.raises(psycopg2.errors.UndefinedTable):
         br_initdb_endpoint.safe_psql(f"select * from {workload.table}")
 
@@ -370,12 +423,8 @@ def test_import_completion_on_restart(
 
 
 @run_only_on_default_postgres(reason="PG version is irrelevant here")
-@pytest.mark.parametrize("action", ["restart", "delete"])
-def test_import_respects_timeline_lifecycle(
-    neon_env_builder: NeonEnvBuilder,
-    vanilla_pg: VanillaPostgres,
-    make_httpserver: HTTPServer,
-    action: str,
+def test_import_respects_tenant_shutdown(
+    neon_env_builder: NeonEnvBuilder, vanilla_pg: VanillaPostgres, make_httpserver: HTTPServer
 ):
     """
     Validate that importing timelines respect the usual timeline life cycle:
@@ -443,33 +492,16 @@ def test_import_respects_timeline_lifecycle(
     wait_until(hit_failpoint)
     assert not import_completion_signaled.is_set()
 
-    if action == "restart":
-        # Restart the pageserver while an import job is in progress.
-        # This clears the failpoint and we expect that the import starts up afresh
-        # after the restart and eventually completes.
-        env.pageserver.stop()
-        env.pageserver.start()
+    # Restart the pageserver while an import job is in progress.
+    # This clears the failpoint and we expect that the import starts up afresh
+    # after the restart and eventually completes.
+    env.pageserver.stop()
+    env.pageserver.start()
 
-        def cplane_notified():
-            assert import_completion_signaled.is_set()
+    def cplane_notified():
+        assert import_completion_signaled.is_set()
 
-        wait_until(cplane_notified)
-    elif action == "delete":
-        status = env.storage_controller.pageserver_api().timeline_delete(tenant_id, timeline_id)
-        assert status == 200
-
-        timeline_path = env.pageserver.timeline_dir(tenant_id, timeline_id)
-        assert not timeline_path.exists(), "Timeline dir exists after deletion"
-
-        shard_zero = TenantShardId(tenant_id, 0, 0)
-        location = env.storage_controller.inspect(shard_zero)
-        assert location is not None
-        generation = location[0]
-
-        with pytest.raises(StorageControllerApiException, match="not found"):
-            env.storage_controller.import_status(shard_zero, timeline_id, generation)
-    else:
-        raise RuntimeError(f"{action} param not recognized")
+    wait_until(cplane_notified)
 
 
 @skip_in_debug_build("Validation query takes too long in debug builds")
@@ -524,8 +556,23 @@ def test_import_chaos(
     neon_env_builder.enable_pageserver_remote_storage(RemoteStorageKind.LOCAL_FS)
 
     vanilla_pg.start()
+    vanilla_pg.safe_psql("create user cloud_admin with password 'postgres' superuser")
+    vanilla_pg.safe_psql("""create table t (data char(900)) with (fillfactor = 10)""")
 
-    inserted_rows = populate_vanilla_pg(vanilla_pg, TARGET_RELBOCK_SIZE)
+    nrows = 0
+    while True:
+        relblock_size = vanilla_pg.safe_psql_scalar("select pg_relation_size('t')")
+        log.info(
+            f"relblock size: {relblock_size / 8192} pages (target: {TARGET_RELBOCK_SIZE // 8192}) pages"
+        )
+        if relblock_size >= TARGET_RELBOCK_SIZE:
+            break
+        addrows = int((TARGET_RELBOCK_SIZE - relblock_size) // 8192)
+        assert addrows >= 1, "forward progress"
+        vanilla_pg.safe_psql(
+            f"insert into t select generate_series({nrows + 1}, {nrows + addrows})"
+        )
+        nrows += addrows
 
     vanilla_pg.stop()
 
@@ -693,7 +740,13 @@ def test_import_chaos(
     endpoint = env.endpoints.create_start(branch_name=import_branch_name, tenant_id=tenant_id)
 
     # Validate the imported data is legit
-    validate_import_from_vanilla_pg(endpoint, inserted_rows)
+    assert endpoint.safe_psql_many(
+        [
+            "set effective_io_concurrency=32;",
+            "SET statement_timeout='300s';",
+            "select count(*), sum(data::bigint)::bigint from t",
+        ]
+    ) == [[], [], [(nrows, nrows * (nrows + 1) // 2)]]
 
     endpoint.stop()
 

test_runner/regress/test_layers_from_future.py

@@ -20,9 +20,6 @@ from fixtures.remote_storage import LocalFsStorage, RemoteStorageKind
 from fixtures.utils import query_scalar, wait_until
 
 
-@pytest.mark.skip(
-    reason="We won't create future layers any more after https://github.com/neondatabase/neon/pull/10548"
-)
 @pytest.mark.parametrize(
     "attach_mode",
     ["default_generation", "same_generation"],

test_runner/regress/test_storage_controller.py

@@ -4158,12 +4158,17 @@ def test_storcon_create_delete_sk_down(
         env.storage_controller.stop()
         env.storage_controller.start()
 
-    with env.endpoints.create("main", tenant_id=tenant_id) as ep:
+    config_lines = [
+        "neon.safekeeper_proto_version = 3",
+    ]
+    with env.endpoints.create("main", tenant_id=tenant_id, config_lines=config_lines) as ep:
         # endpoint should start.
         ep.start(safekeeper_generation=1, safekeepers=[1, 2, 3])
         ep.safe_psql("CREATE TABLE IF NOT EXISTS t(key int, value text)")
 
-    with env.endpoints.create("child_of_main", tenant_id=tenant_id) as ep:
+    with env.endpoints.create(
+        "child_of_main", tenant_id=tenant_id, config_lines=config_lines
+    ) as ep:
         # endpoint should start.
         ep.start(safekeeper_generation=1, safekeepers=[1, 2, 3])
         ep.safe_psql("CREATE TABLE IF NOT EXISTS t(key int, value text)")
@@ -4192,10 +4197,10 @@ def test_storcon_create_delete_sk_down(
     # ensure the safekeeper deleted the timeline
     def timeline_deleted_on_active_sks():
         env.safekeepers[0].assert_log_contains(
-            f"((deleting timeline|Timeline) {tenant_id}/{child_timeline_id} (from disk|was already deleted)|DELETE.*tenant/{tenant_id} .*status: 200 OK)"
+            f"deleting timeline {tenant_id}/{child_timeline_id} from disk"
         )
         env.safekeepers[2].assert_log_contains(
-            f"((deleting timeline|Timeline) {tenant_id}/{child_timeline_id} (from disk|was already deleted)|DELETE.*tenant/{tenant_id} .*status: 200 OK)"
+            f"deleting timeline {tenant_id}/{child_timeline_id} from disk"
         )
 
     wait_until(timeline_deleted_on_active_sks)
@@ -4210,7 +4215,7 @@ def test_storcon_create_delete_sk_down(
     # ensure that there is log msgs for the third safekeeper too
     def timeline_deleted_on_sk():
         env.safekeepers[1].assert_log_contains(
-            f"((deleting timeline|Timeline) {tenant_id}/{child_timeline_id} (from disk|was already deleted)|DELETE.*tenant/{tenant_id} .*status: 200 OK)"
+            f"deleting timeline {tenant_id}/{child_timeline_id} from disk"
         )
 
     wait_until(timeline_deleted_on_sk)
@@ -4244,12 +4249,17 @@ def test_storcon_few_sk(
 
     env.safekeepers[0].assert_log_contains(f"creating new timeline {tenant_id}/{timeline_id}")
 
-    with env.endpoints.create("main", tenant_id=tenant_id) as ep:
+    config_lines = [
+        "neon.safekeeper_proto_version = 3",
+    ]
+    with env.endpoints.create("main", tenant_id=tenant_id, config_lines=config_lines) as ep:
         # endpoint should start.
         ep.start(safekeeper_generation=1, safekeepers=safekeeper_list)
         ep.safe_psql("CREATE TABLE IF NOT EXISTS t(key int, value text)")
 
-    with env.endpoints.create("child_of_main", tenant_id=tenant_id) as ep:
+    with env.endpoints.create(
+        "child_of_main", tenant_id=tenant_id, config_lines=config_lines
+    ) as ep:
         # endpoint should start.
         ep.start(safekeeper_generation=1, safekeepers=safekeeper_list)
         ep.safe_psql("CREATE TABLE IF NOT EXISTS t(key int, value text)")

test_runner/regress/test_wal_acceptor.py

@@ -2012,7 +2012,10 @@ def test_explicit_timeline_creation(neon_env_builder: NeonEnvBuilder):
     tenant_id = env.initial_tenant
     timeline_id = env.initial_timeline
 
-    ep = env.endpoints.create("main")
+    config_lines = [
+        "neon.safekeeper_proto_version = 3",
+    ]
+    ep = env.endpoints.create("main", config_lines=config_lines)
 
     # expected to fail because timeline is not created on safekeepers
     with pytest.raises(Exception, match=r".*timed out.*"):
@@ -2040,7 +2043,10 @@ def test_explicit_timeline_creation_storcon(neon_env_builder: NeonEnvBuilder):
     }
     env = neon_env_builder.init_start()
 
-    ep = env.endpoints.create("main")
+    config_lines = [
+        "neon.safekeeper_proto_version = 3",
+    ]
+    ep = env.endpoints.create("main", config_lines=config_lines)
 
     # endpoint should start.
     ep.start(safekeeper_generation=1, safekeepers=[1, 2, 3])

test_runner/regress/test_wal_acceptor_async.py

@@ -637,7 +637,10 @@ async def quorum_sanity_single(
     # create timeline on `members_sks`
     Safekeeper.create_timeline(tenant_id, timeline_id, env.pageservers[0], mconf, members_sks)
 
-    ep = env.endpoints.create(branch_name)
+    config_lines = [
+        "neon.safekeeper_proto_version = 3",
+    ]
+    ep = env.endpoints.create(branch_name, config_lines=config_lines)
     ep.start(safekeeper_generation=1, safekeepers=compute_sks_ids)
     ep.safe_psql("create table t(key int, value text)")