wip

WIP
prefix endpoint ids with test name
2026-08-12 09:19:39 +00:00 · 2024-09-12 01:36:11 +03:00 · 2024-09-12 01:20:59 +03:00 · 2024-09-12 01:20:59 +03:00 · 2024-09-12 01:20:59 +03:00 · 2024-09-12 01:20:59 +03:00
101 changed files with 2174 additions and 2247 deletions
@@ -1189,9 +1189,9 @@ dependencies = [

 [[package]]
 name = "comfy-table"
-version = "6.1.4"
+version = "7.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6e7b787b0dc42e8111badfdbe4c3059158ccb2db8780352fa1b01e8ccf45cc4d"
+checksum = "b34115915337defe99b2aff5c2ce6771e5fbc4079f4b506301f5cf394c8452f7"
 dependencies = [
 "crossterm",
 "strum",
@@ -1246,7 +1246,7 @@ dependencies = [
 "tokio-postgres",
 "tokio-stream",
 "tokio-util",
- "toml_edit 0.19.10",
+ "toml_edit",
 "tracing",
 "tracing-opentelemetry",
 "tracing-subscriber",
@@ -1360,8 +1360,8 @@ dependencies = [
 "tokio",
 "tokio-postgres",
 "tokio-util",
- "toml 0.7.4",
- "toml_edit 0.19.10",
+ "toml",
+ "toml_edit",
 "tracing",
 "url",
 "utils",
@@ -1485,25 +1485,22 @@ checksum = "248e3bacc7dc6baa3b21e405ee045c3047101a49145e7e9eca583ab4c2ca5345"

 [[package]]
 name = "crossterm"
-version = "0.25.0"
+version = "0.27.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e64e6c0fbe2c17357405f7c758c1ef960fce08bdfb2c03d88d2a18d7e09c4b67"
+checksum = "f476fe445d41c9e991fd07515a6f463074b782242ccf4a5b7b1d1012e70824df"
 dependencies = [
- "bitflags 1.3.2",
+ "bitflags 2.4.1",
 "crossterm_winapi",
 "libc",
- "mio",
 "parking_lot 0.12.1",
- "signal-hook",
- "signal-hook-mio",
 "winapi",
 ]

 [[package]]
 name = "crossterm_winapi"
-version = "0.9.0"
+version = "0.9.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2ae1b35a484aa10e07fe0638d02301c5ad24de82d310ccbd2f3693da5f09bf1c"
+checksum = "acdd7c62a3665c7f6830a51635d9ac9b23ed385797f70a83bb8bafe9c572ab2b"
 dependencies = [
 "winapi",
 ]
@@ -3144,7 +3141,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "fd01039851e82f8799046eabbb354056283fb265c8ec0996af940f4e85a380ff"
 dependencies = [
 "serde",
- "toml 0.8.14",
+ "toml",
 ]

 [[package]]
@@ -3660,7 +3657,7 @@ dependencies = [
 "thiserror",
 "tokio",
 "tokio-util",
- "toml_edit 0.19.10",
+ "toml_edit",
 "utils",
 "workspace_hack",
 ]
@@ -3747,7 +3744,7 @@ dependencies = [
 "tokio-stream",
 "tokio-tar",
 "tokio-util",
- "toml_edit 0.19.10",
+ "toml_edit",
 "tracing",
 "twox-hash",
 "url",
@@ -3910,8 +3907,9 @@ dependencies = [

 [[package]]
 name = "parquet"
-version = "51.0.0"
-source = "git+https://github.com/apache/arrow-rs?branch=master#2534976a564be3d2d56312dc88fb1b6ed4cef829"
+version = "53.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f0fbf928021131daaa57d334ca8e3904fe9ae22f73c56244fc7db9b04eedc3d8"
 dependencies = [
 "ahash",
 "bytes",
@@ -3930,8 +3928,9 @@ dependencies = [

 [[package]]
 name = "parquet_derive"
-version = "51.0.0"
-source = "git+https://github.com/apache/arrow-rs?branch=master#2534976a564be3d2d56312dc88fb1b6ed4cef829"
+version = "53.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "86e9fcfae007533a06b580429a3f7e07cb833ec8aa37c041c16563e7918f057e"
 dependencies = [
 "parquet",
 "proc-macro2",
@@ -4121,7 +4120,7 @@ dependencies = [
 [[package]]
 name = "postgres"
 version = "0.19.4"
-source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#20031d7a9ee1addeae6e0968e3899ae6bf01cee2"
+source = "git+https://github.com/neondatabase/rust-postgres.git?rev=20031d7a9ee1addeae6e0968e3899ae6bf01cee2#20031d7a9ee1addeae6e0968e3899ae6bf01cee2"
 dependencies = [
 "bytes",
 "fallible-iterator",
@@ -4134,7 +4133,7 @@ dependencies = [
 [[package]]
 name = "postgres-protocol"
 version = "0.6.4"
-source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#20031d7a9ee1addeae6e0968e3899ae6bf01cee2"
+source = "git+https://github.com/neondatabase/rust-postgres.git?rev=20031d7a9ee1addeae6e0968e3899ae6bf01cee2#20031d7a9ee1addeae6e0968e3899ae6bf01cee2"
 dependencies = [
 "base64 0.20.0",
 "byteorder",
@@ -4153,7 +4152,7 @@ dependencies = [
 [[package]]
 name = "postgres-types"
 version = "0.2.4"
-source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#20031d7a9ee1addeae6e0968e3899ae6bf01cee2"
+source = "git+https://github.com/neondatabase/rust-postgres.git?rev=20031d7a9ee1addeae6e0968e3899ae6bf01cee2#20031d7a9ee1addeae6e0968e3899ae6bf01cee2"
 dependencies = [
 "bytes",
 "fallible-iterator",
@@ -4812,7 +4811,7 @@ dependencies = [
 "tokio",
 "tokio-stream",
 "tokio-util",
- "toml_edit 0.19.10",
+ "toml_edit",
 "tracing",
 "utils",
 ]
@@ -5322,7 +5321,7 @@ dependencies = [
 "tokio-stream",
 "tokio-tar",
 "tokio-util",
- "toml_edit 0.19.10",
+ "toml_edit",
 "tracing",
 "tracing-subscriber",
 "url",
@@ -5731,17 +5730,6 @@ dependencies = [
 "signal-hook-registry",
 ]

-[[package]]
-name = "signal-hook-mio"
-version = "0.2.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "29ad2e15f37ec9a6cc544097b78a1ec90001e9f71b81338ca39f430adaca99af"
-dependencies = [
- "libc",
- "mio",
- "signal-hook",
-]
-
 [[package]]
 name = "signal-hook-registry"
 version = "1.4.1"
@@ -6054,21 +6042,21 @@ checksum = "73473c0e59e6d5812c5dfe2a064a6444949f089e20eec9a2e5506596494e4623"

 [[package]]
 name = "strum"
-version = "0.24.1"
+version = "0.26.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "063e6045c0e62079840579a7e47a355ae92f60eb74daaf156fb1e84ba164e63f"
+checksum = "8fec0f0aef304996cf250b31b5a10dee7980c85da9d759361292b8bca5a18f06"

 [[package]]
 name = "strum_macros"
-version = "0.24.3"
+version = "0.26.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1e385be0d24f186b4ce2f9982191e7101bb737312ad61c1f2f984f34bcf85d59"
+checksum = "4c6bee85a5a24955dc440386795aa378cd9cf82acd5f764469152d2270e581be"
 dependencies = [
- "heck 0.4.1",
+ "heck 0.5.0",
 "proc-macro2",
 "quote",
 "rustversion",
- "syn 1.0.109",
+ "syn 2.0.52",
 ]

 [[package]]
@@ -6409,7 +6397,7 @@ dependencies = [
 [[package]]
 name = "tokio-postgres"
 version = "0.7.7"
-source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#20031d7a9ee1addeae6e0968e3899ae6bf01cee2"
+source = "git+https://github.com/neondatabase/rust-postgres.git?rev=20031d7a9ee1addeae6e0968e3899ae6bf01cee2#20031d7a9ee1addeae6e0968e3899ae6bf01cee2"
 dependencies = [
 "async-trait",
 "byteorder",
@@ -6520,18 +6508,6 @@ dependencies = [
 "tracing",
 ]

-[[package]]
-name = "toml"
-version = "0.7.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d6135d499e69981f9ff0ef2167955a5333c35e36f6937d382974566b3d5b94ec"
-dependencies = [
- "serde",
- "serde_spanned",
- "toml_datetime",
- "toml_edit 0.19.10",
-]
-
 [[package]]
 name = "toml"
 version = "0.8.14"
@@ -6541,7 +6517,7 @@ dependencies = [
 "serde",
 "serde_spanned",
 "toml_datetime",
- "toml_edit 0.22.14",
+ "toml_edit",
 ]

 [[package]]
@@ -6553,19 +6529,6 @@ dependencies = [
 "serde",
 ]

-[[package]]
-name = "toml_edit"
-version = "0.19.10"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2380d56e8670370eee6566b0bfd4265f65b3f432e8c6d85623f728d4fa31f739"
-dependencies = [
- "indexmap 1.9.3",
- "serde",
- "serde_spanned",
- "toml_datetime",
- "winnow 0.4.6",
-]
-
 [[package]]
 name = "toml_edit"
 version = "0.22.14"
@@ -6576,7 +6539,7 @@ dependencies = [
 "serde",
 "serde_spanned",
 "toml_datetime",
- "winnow 0.6.13",
+ "winnow",
 ]

 [[package]]
@@ -6989,7 +6952,7 @@ dependencies = [
 "tokio-stream",
 "tokio-tar",
 "tokio-util",
- "toml_edit 0.19.10",
+ "toml_edit",
 "tracing",
 "tracing-error",
 "tracing-subscriber",
@@ -7535,15 +7498,6 @@ version = "0.52.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "32b752e52a2da0ddfbdbcc6fceadfeede4c939ed16d13e648833a61dfb611ed8"

-[[package]]
-name = "winnow"
-version = "0.4.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "61de7bac303dc551fe038e2b3cef0f571087a47571ea6e79a87692ac99b99699"
-dependencies = [
- "memchr",
-]
-
 [[package]]
 name = "winnow"
 version = "0.6.13"
@@ -7651,6 +7605,7 @@ dependencies = [
 "tokio",
 "tokio-rustls 0.24.0",
 "tokio-util",
+ "toml_edit",
 "tonic",
 "tower",
 "tracing",
@@ -73,7 +73,7 @@ camino = "1.1.6"
 cfg-if = "1.0.0"
 chrono = { version = "0.4", default-features = false, features = ["clock"] }
 clap = { version = "4.0", features = ["derive"] }
-comfy-table = "6.1"
+comfy-table = "7.1"
 const_format = "0.2"
 crc32c = "0.6"
 crossbeam-deque = "0.8.5"
@@ -123,8 +123,8 @@ opentelemetry = "0.20.0"
 opentelemetry-otlp = { version = "0.13.0", default-features=false, features = ["http-proto", "trace", "http", "reqwest-client"] }
 opentelemetry-semantic-conventions = "0.12.0"
 parking_lot = "0.12"
-parquet = { version = "51.0.0", default-features = false, features = ["zstd"] }
-parquet_derive = "51.0.0"
+parquet = { version = "53", default-features = false, features = ["zstd"] }
+parquet_derive = "53"
 pbkdf2 = { version = "0.12.1", features = ["simple", "std"] }
 pin-project-lite = "0.2"
 procfs = "0.16"
@@ -158,8 +158,8 @@ signal-hook = "0.3"
 smallvec = "1.11"
 smol_str = { version = "0.2.0", features = ["serde"] }
 socket2 = "0.5"
-strum = "0.24"
-strum_macros = "0.24"
+strum = "0.26"
+strum_macros = "0.26"
 "subtle"  = "2.5.0"
 svg_fmt = "0.4.3"
 sync_wrapper = "0.1.2"
@@ -177,8 +177,8 @@ tokio-rustls = "0.25"
 tokio-stream = "0.1"
 tokio-tar = "0.3"
 tokio-util = { version = "0.7.10", features = ["io", "rt"] }
-toml = "0.7"
-toml_edit = "0.19"
+toml = "0.8"
+toml_edit = "0.22"
 tonic = {version = "0.9", features = ["tls", "tls-roots"]}
 tower-service = "0.3.2"
 tracing = "0.1"
@@ -201,10 +201,21 @@ env_logger = "0.10"
 log = "0.4"

 ## Libraries from neondatabase/ git forks, ideally with changes to be upstreamed
-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", branch="neon" }
-postgres-protocol = { git = "https://github.com/neondatabase/rust-postgres.git", branch="neon" }
-postgres-types = { git = "https://github.com/neondatabase/rust-postgres.git", branch="neon" }
-tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", branch="neon" }
+
+# We want to use the 'neon' branch for these, but there's currently one
+# incompatible change on the branch. See:
+#
+# - PR #8076 which contained changes that depended on the new changes in
+#   the rust-postgres crate, and
+# - PR #8654 which reverted those changes and made the code in proxy incompatible
+#   with the tip of the 'neon' branch again.
+#
+# When those proxy changes are re-applied (see PR #8747), we can switch using
+# the tip of the 'neon' branch again.
+postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev = "20031d7a9ee1addeae6e0968e3899ae6bf01cee2" }
+postgres-protocol = { git = "https://github.com/neondatabase/rust-postgres.git", rev = "20031d7a9ee1addeae6e0968e3899ae6bf01cee2" }
+postgres-types = { git = "https://github.com/neondatabase/rust-postgres.git", rev = "20031d7a9ee1addeae6e0968e3899ae6bf01cee2" }
+tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev = "20031d7a9ee1addeae6e0968e3899ae6bf01cee2" }

 ## Local libraries
 compute_api = { version = "0.1", path = "./libs/compute_api/" }
@@ -241,11 +252,7 @@ tonic-build = "0.9"
 [patch.crates-io]

 # Needed to get `tokio-postgres-rustls` to depend on our fork.
-tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", branch="neon" }
-
-# bug fixes for UUID
-parquet = { git = "https://github.com/apache/arrow-rs", branch = "master" }
-parquet_derive = { git = "https://github.com/apache/arrow-rs", branch = "master" }
+tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev = "20031d7a9ee1addeae6e0968e3899ae6bf01cee2" }

 ################# Binary contents sections

@@ -90,8 +90,11 @@ fn main() -> Result<()> {
        handle_init(sub_args).map(Some)
    } else {
        // all other commands need an existing config
-        let mut env =
-            LocalEnv::load_config(&local_env::base_path()).context("Error loading config")?;
+        let base_path = local_env::base_path();
+        let branch_mappings_path = local_env::branch_mappings_path();
+
+        let mut env = LocalEnv::load_config(&base_path, Some(&branch_mappings_path))
+            .context("Error loading config")?;
        let original_env = env.clone();

        let rt = tokio::runtime::Builder::new_current_thread()
@@ -264,7 +267,7 @@ async fn get_timeline_infos(
 fn get_tenant_id(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> anyhow::Result<TenantId> {
    if let Some(tenant_id_from_arguments) = parse_tenant_id(sub_match).transpose() {
        tenant_id_from_arguments
-    } else if let Some(default_id) = env.default_tenant_id {
+    } else if let Some(default_id) = env.branch_mappings.default_tenant_id {
        Ok(default_id)
    } else {
        anyhow::bail!("No tenant id. Use --tenant-id, or set a default tenant");
@@ -278,7 +281,7 @@ fn get_tenant_shard_id(
 ) -> anyhow::Result<TenantShardId> {
    if let Some(tenant_id_from_arguments) = parse_tenant_shard_id(sub_match).transpose() {
        tenant_id_from_arguments
-    } else if let Some(default_id) = env.default_tenant_id {
+    } else if let Some(default_id) = env.branch_mappings.default_tenant_id {
        Ok(TenantShardId::unsharded(default_id))
    } else {
        anyhow::bail!("No tenant shard id. Use --tenant-id, or set a default tenant");
@@ -360,7 +363,6 @@ fn handle_init(init_match: &ArgMatches) -> anyhow::Result<LocalEnv> {
                .collect(),
            pg_distrib_dir: None,
            neon_distrib_dir: None,
-            default_tenant_id: TenantId::from_array(std::array::from_fn(|_| 0)),
            storage_controller: None,
            control_plane_compute_hook_api: None,
        }
@@ -368,8 +370,12 @@ fn handle_init(init_match: &ArgMatches) -> anyhow::Result<LocalEnv> {

    LocalEnv::init(init_conf, force)
        .context("materialize initial neon_local environment on disk")?;
-    Ok(LocalEnv::load_config(&local_env::base_path())
-        .expect("freshly written config should be loadable"))
+    let base_path = local_env::base_path();
+    let branch_mappings_path = local_env::branch_mappings_path();
+    let env = LocalEnv::load_config(&base_path, Some(&branch_mappings_path))
+        .expect("freshly written config should be loadable");
+
+    Ok(env)
 }

 /// The default pageserver is the one where CLI tenant/timeline operations are sent by default.
@@ -525,14 +531,14 @@ async fn handle_tenant(

            if create_match.get_flag("set-default") {
                println!("Setting tenant {tenant_id} as a default one");
-                env.default_tenant_id = Some(tenant_id);
+                env.branch_mappings.default_tenant_id = Some(tenant_id);
            }
        }
        Some(("set-default", set_default_match)) => {
            let tenant_id =
                parse_tenant_id(set_default_match)?.context("No tenant id specified")?;
            println!("Setting tenant {tenant_id} as a default one");
-            env.default_tenant_id = Some(tenant_id);
+            env.branch_mappings.default_tenant_id = Some(tenant_id);
        }
        Some(("config", create_match)) => {
            let tenant_id = get_tenant_id(create_match, env)?;
@@ -640,6 +646,8 @@ async fn handle_timeline(timeline_match: &ArgMatches, env: &mut local_env::Local
        }
        Some(("branch", branch_match)) => {
            let tenant_id = get_tenant_id(branch_match, env)?;
+            let new_timeline_id =
+                parse_timeline_id(branch_match)?.unwrap_or(TimelineId::generate());
            let new_branch_name = branch_match
                .get_one::<String>("branch-name")
                .ok_or_else(|| anyhow!("No branch name provided"))?;
@@ -658,7 +666,6 @@ async fn handle_timeline(timeline_match: &ArgMatches, env: &mut local_env::Local
                .map(|lsn_str| Lsn::from_str(lsn_str))
                .transpose()
                .context("Failed to parse ancestor start Lsn from the request")?;
-            let new_timeline_id = TimelineId::generate();
            let storage_controller = StorageController::from_env(env);
            let create_req = TimelineCreateRequest {
                new_timeline_id,
@@ -692,6 +699,7 @@ async fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Re
        Some(ep_subcommand_data) => ep_subcommand_data,
        None => bail!("no endpoint subcommand provided"),
    };
+
    let mut cplane = ComputeControlPlane::load(env.clone())?;

    match sub_name {
@@ -1359,6 +1367,7 @@ async fn handle_stop_all(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> R

 async fn try_stop_all(env: &local_env::LocalEnv, immediate: bool) {
    // Stop all endpoints
+    // NOTE: This only knows about endpoints in the default endpoints dir
    match ComputeControlPlane::load(env.clone()) {
        Ok(cplane) => {
            for (_k, node) in cplane.endpoints {
@@ -1419,6 +1428,18 @@ fn cli() -> Command {
        .default_value("10s")
        .required(false);

+    let branch_mappings_arg = Arg::new("branch-mappings")
+        .long("branch-mappings")
+        .help("File holding all branch names. Default is <repo dir>/branches.toml")
+        .value_parser(value_parser!(PathBuf))
+        .required(false);
+
+    let endpoints_dir_arg = Arg::new("endpoints-dir")
+        .long("endpoints-dir")
+        .help("Path to directory holding all endpoints. Default is <repo dir>/endpoints")
+        .value_parser(value_parser!(PathBuf))
+        .required(false);
+
    let branch_name_arg = Arg::new("branch-name")
        .long("branch-name")
        .help("Name of the branch to be created or used as an alias for other services")
@@ -1559,6 +1580,8 @@ fn cli() -> Command {
    Command::new("Neon CLI")
        .arg_required_else_help(true)
        .version(GIT_VERSION)
+        .arg(branch_mappings_arg)
+        .arg(endpoints_dir_arg)
        .subcommand(
            Command::new("init")
                .about("Initialize a new Neon repository, preparing configs for services to start with")
@@ -1570,7 +1593,6 @@ fn cli() -> Command {
                        .value_parser(value_parser!(PathBuf))
                        .value_name("config")
                )
-                .arg(pg_version_arg.clone())
                .arg(force_arg)
        )
        .subcommand(
@@ -1583,6 +1605,7 @@ fn cli() -> Command {
            .subcommand(Command::new("branch")
                .about("Create a new timeline, using another timeline as a base, copying its data")
                .arg(tenant_id_arg.clone())
+                .arg(timeline_id_arg.clone())
                .arg(branch_name_arg.clone())
                .arg(Arg::new("ancestor-branch-name").long("ancestor-branch-name")
                    .help("Use last Lsn of another timeline (and its data) as base when creating the new timeline. The timeline gets resolved by its branch name.").required(false))
@@ -46,6 +46,11 @@ pub struct LocalEnv {
    // must be an absolute path. If the env var is not set, $PWD/.neon is used.
    pub base_data_dir: PathBuf,

+    // Similarly, path to branch mappings file. Not stored in the config file but
+    // read from the NEON_BRANCH_MAPPINGS env variable. "None" means no mappings
+    // are loaded, and they cannot be saved either.
+    pub branch_name_mappings_path: Option<PathBuf>,
+
    // Path to postgres distribution. It's expected that "bin", "include",
    // "lib", "share" from postgres distribution are there. If at some point
    // in time we will be able to run against vanilla postgres we may split that
@@ -55,10 +60,6 @@ pub struct LocalEnv {
    // Path to pageserver binary.
    pub neon_distrib_dir: PathBuf,

-    // Default tenant ID to use with the 'neon_local' command line utility, when
-    // --tenant_id is not explicitly specified.
-    pub default_tenant_id: Option<TenantId>,
-
    // used to issue tokens during e.g pg start
    pub private_key_path: PathBuf,

@@ -82,11 +83,7 @@ pub struct LocalEnv {
    // storage controller's configuration.
    pub control_plane_compute_hook_api: Option<Url>,

-    /// Keep human-readable aliases in memory (and persist them to config), to hide ZId hex strings from the user.
-    // A `HashMap<String, HashMap<TenantId, TimelineId>>` would be more appropriate here,
-    // but deserialization into a generic toml object as `toml::Value::try_from` fails with an error.
-    // https://toml.io/en/v1.0.0 does not contain a concept of "a table inside another table".
-    pub branch_name_mappings: HashMap<String, Vec<(TenantId, TimelineId)>>,
+    pub branch_mappings: BranchMappings,
 }

 /// On-disk state stored in `.neon/config`.
@@ -95,7 +92,6 @@ pub struct LocalEnv {
 pub struct OnDiskConfig {
    pub pg_distrib_dir: PathBuf,
    pub neon_distrib_dir: PathBuf,
-    pub default_tenant_id: Option<TenantId>,
    pub private_key_path: PathBuf,
    pub broker: NeonBroker,
    pub storage_controller: NeonStorageControllerConf,
@@ -107,7 +103,20 @@ pub struct OnDiskConfig {
    pub safekeepers: Vec<SafekeeperConf>,
    pub control_plane_api: Option<Url>,
    pub control_plane_compute_hook_api: Option<Url>,
-    branch_name_mappings: HashMap<String, Vec<(TenantId, TimelineId)>>,
+}
+
+#[derive(PartialEq, Eq, Clone, Debug, Default, Serialize, Deserialize)]
+#[serde(default, deny_unknown_fields)]
+pub struct BranchMappings {
+    // Default tenant ID to use with the 'neon_local' command line utility, when
+    // --tenant_id is not explicitly specified. This comes from the branches.
+    pub default_tenant_id: Option<TenantId>,
+
+    /// Keep human-readable aliases in memory (and persist them to config XXX), to hide ZId hex strings from the user.
+    // A `HashMap<String, HashMap<TenantId, TimelineId>>` would be more appropriate here,
+    // but deserialization into a generic toml object as `toml::Value::try_from` fails with an error.
+    // https://toml.io/en/v1.0.0 does not contain a concept of "a table inside another table".
+    pub mappings: HashMap<String, Vec<(TenantId, TimelineId)>>,
 }

 fn fail_if_pageservers_field_specified<'de, D>(_: D) -> Result<Vec<PageServerConf>, D::Error>
@@ -128,7 +137,6 @@ pub struct NeonLocalInitConf {
    pub pg_distrib_dir: Option<PathBuf>,
    // TODO: do we need this? Seems unused
    pub neon_distrib_dir: Option<PathBuf>,
-    pub default_tenant_id: TenantId,
    pub broker: NeonBroker,
    pub storage_controller: Option<NeonStorageControllerConf>,
    pub pageservers: Vec<NeonLocalInitPageserverConf>,
@@ -443,7 +451,8 @@ impl LocalEnv {
        timeline_id: TimelineId,
    ) -> anyhow::Result<()> {
        let existing_values = self
-            .branch_name_mappings
+            .branch_mappings
+            .mappings
            .entry(branch_name.clone())
            .or_default();

@@ -468,7 +477,8 @@ impl LocalEnv {
        branch_name: &str,
        tenant_id: TenantId,
    ) -> Option<TimelineId> {
-        self.branch_name_mappings
+        self.branch_mappings
+            .mappings
            .get(branch_name)?
            .iter()
            .find(|(mapped_tenant_id, _)| mapped_tenant_id == &tenant_id)
@@ -477,7 +487,8 @@ impl LocalEnv {
    }

    pub fn timeline_name_mappings(&self) -> HashMap<TenantTimelineId, String> {
-        self.branch_name_mappings
+        self.branch_mappings
+            .mappings
            .iter()
            .flat_map(|(name, tenant_timelines)| {
                tenant_timelines.iter().map(|&(tenant_id, timeline_id)| {
@@ -488,7 +499,10 @@ impl LocalEnv {
    }

    ///  Construct `Self` from on-disk state.
-    pub fn load_config(repopath: &Path) -> anyhow::Result<Self> {
+    pub fn load_config(
+        repopath: &Path,
+        branch_name_mappings_path: Option<&Path>,
+    ) -> anyhow::Result<Self> {
        if !repopath.exists() {
            bail!(
                "Neon config is not found in {}. You need to run 'neon_local init' first",
@@ -498,37 +512,43 @@ impl LocalEnv {

        // TODO: check that it looks like a neon repository

-        // load and parse file
+        // load and parse config file
        let config_file_contents = fs::read_to_string(repopath.join("config"))?;
        let on_disk_config: OnDiskConfig = toml::from_str(config_file_contents.as_str())?;
-        let mut env = {
-            let OnDiskConfig {
-                pg_distrib_dir,
-                neon_distrib_dir,
-                default_tenant_id,
-                private_key_path,
-                broker,
-                storage_controller,
-                pageservers,
-                safekeepers,
-                control_plane_api,
-                control_plane_compute_hook_api,
-                branch_name_mappings,
-            } = on_disk_config;
-            LocalEnv {
-                base_data_dir: repopath.to_owned(),
-                pg_distrib_dir,
-                neon_distrib_dir,
-                default_tenant_id,
-                private_key_path,
-                broker,
-                storage_controller,
-                pageservers,
-                safekeepers,
-                control_plane_api,
-                control_plane_compute_hook_api,
-                branch_name_mappings,
-            }
+        let OnDiskConfig {
+            pg_distrib_dir,
+            neon_distrib_dir,
+            private_key_path,
+            broker,
+            storage_controller,
+            pageservers,
+            safekeepers,
+            control_plane_api,
+            control_plane_compute_hook_api,
+        } = on_disk_config;
+
+        // load and parse "branches.toml" file
+        let branch_mappings = if let Some(path) = branch_name_mappings_path {
+            let contents = fs::read_to_string(path)
+                .context(format!("load branch mappings file {}", path.display()))?;
+            toml::from_str::<BranchMappings>(contents.as_str())?
+        } else {
+            BranchMappings::default()
+        };
+
+        let mut env = LocalEnv {
+            base_data_dir: repopath.to_owned(),
+            branch_name_mappings_path: branch_name_mappings_path.map(|p| p.to_owned()),
+            pg_distrib_dir,
+            neon_distrib_dir,
+            private_key_path,
+            broker,
+            storage_controller,
+            pageservers,
+            safekeepers,
+            control_plane_api,
+            control_plane_compute_hook_api,
+            branch_mappings,
        };

        // The source of truth for pageserver configuration is the pageserver.toml.
@@ -618,7 +638,6 @@ impl LocalEnv {
            &OnDiskConfig {
                pg_distrib_dir: self.pg_distrib_dir.clone(),
                neon_distrib_dir: self.neon_distrib_dir.clone(),
-                default_tenant_id: self.default_tenant_id,
                private_key_path: self.private_key_path.clone(),
                broker: self.broker.clone(),
                storage_controller: self.storage_controller.clone(),
@@ -626,9 +645,20 @@ impl LocalEnv {
                safekeepers: self.safekeepers.clone(),
                control_plane_api: self.control_plane_api.clone(),
                control_plane_compute_hook_api: self.control_plane_compute_hook_api.clone(),
-                branch_name_mappings: self.branch_name_mappings.clone(),
            },
-        )
+        )?;
+
+        if let Some(path) = &self.branch_name_mappings_path {
+            Self::persist_branches_impl(path, &self.branch_mappings)?;
+        } else {
+            if !self.branch_mappings.mappings.is_empty() {
+                tracing::warn!("command created a branch mapping, but it was not saved because no mappings file was configured")
+            } else if self.branch_mappings.default_tenant_id.is_some() {
+                tracing::warn!("command created a tenant default, but it was not saved because no mappings file was configured")
+            }
+        }
+
+        Ok(())
    }

    pub fn persist_config_impl(base_path: &Path, config: &OnDiskConfig) -> anyhow::Result<()> {
@@ -642,6 +672,19 @@ impl LocalEnv {
        })
    }

+    pub fn persist_branches_impl(
+        branch_name_mappings_path: &Path,
+        branch_mappings: &BranchMappings,
+    ) -> anyhow::Result<()> {
+        let content = &toml::to_string_pretty(branch_mappings)?;
+        fs::write(branch_name_mappings_path, content).with_context(|| {
+            format!(
+                "Failed to write branch information into path '{}'",
+                branch_name_mappings_path.display()
+            )
+        })
+    }
+
    // this function is used only for testing purposes in CLI e g generate tokens during init
    pub fn generate_auth_token(&self, claims: &Claims) -> anyhow::Result<String> {
        let private_key_path = self.get_private_key_path();
@@ -702,7 +745,6 @@ impl LocalEnv {
        let NeonLocalInitConf {
            pg_distrib_dir,
            neon_distrib_dir,
-            default_tenant_id,
            broker,
            storage_controller,
            pageservers,
@@ -746,9 +788,9 @@ impl LocalEnv {
        // TODO: refactor to avoid this, LocalEnv should only be constructed from on-disk state
        let env = LocalEnv {
            base_data_dir: base_path.clone(),
+            branch_name_mappings_path: Some(base_path.join("branches.toml")),
            pg_distrib_dir,
            neon_distrib_dir,
-            default_tenant_id: Some(default_tenant_id),
            private_key_path,
            broker,
            storage_controller: storage_controller.unwrap_or_default(),
@@ -756,10 +798,10 @@ impl LocalEnv {
            safekeepers,
            control_plane_api: control_plane_api.unwrap_or_default(),
            control_plane_compute_hook_api: control_plane_compute_hook_api.unwrap_or_default(),
-            branch_name_mappings: Default::default(),
+            branch_mappings: Default::default(),
        };

-        // create endpoints dir
+        // create the default endpoints dir
        fs::create_dir_all(env.endpoints_path())?;

        // create safekeeper dirs
@@ -806,6 +848,24 @@ pub fn base_path() -> PathBuf {
    path
 }

+pub fn branch_mappings_path() -> PathBuf {
+    let path = match std::env::var_os("NEON_BRANCH_MAPPINGS") {
+        Some(val) => {
+            let path = PathBuf::from(val);
+
+            // a relative path is relative to repo dir
+            if !path.is_absolute() {
+                base_path().join(path)
+            } else {
+                path
+            }
+        }
+        None => base_path().join("branches.toml"),
+    };
+    assert!(path.is_absolute());
+    path
+}
+
 /// Generate a public/private key pair for JWT authentication
 fn generate_auth_keys(private_key_path: &Path, public_key_path: &Path) -> anyhow::Result<()> {
    // Generate the key pair
@@ -75,14 +75,14 @@ impl PageServerNode {
        }
    }

-    fn pageserver_make_identity_toml(&self, node_id: NodeId) -> toml_edit::Document {
-        toml_edit::Document::from_str(&format!("id={node_id}")).unwrap()
+    fn pageserver_make_identity_toml(&self, node_id: NodeId) -> toml_edit::DocumentMut {
+        toml_edit::DocumentMut::from_str(&format!("id={node_id}")).unwrap()
    }

    fn pageserver_init_make_toml(
        &self,
        conf: NeonLocalInitPageserverConf,
-    ) -> anyhow::Result<toml_edit::Document> {
+    ) -> anyhow::Result<toml_edit::DocumentMut> {
        assert_eq!(&PageServerConf::from(&conf), &self.conf, "during neon_local init, we derive the runtime state of ps conf (self.conf) from the --config flag fully");

        // TODO(christian): instead of what we do here, create a pageserver_api::config::ConfigToml (PR #7656)
@@ -137,9 +137,9 @@ impl PageServerNode {

        // Turn `overrides` into a toml document.
        // TODO: above code is legacy code, it should be refactored to use toml_edit directly.
-        let mut config_toml = toml_edit::Document::new();
+        let mut config_toml = toml_edit::DocumentMut::new();
        for fragment_str in overrides {
-            let fragment = toml_edit::Document::from_str(&fragment_str)
+            let fragment = toml_edit::DocumentMut::from_str(&fragment_str)
                .expect("all fragments in `overrides` are valid toml documents, this function controls that");
            for (key, item) in fragment.iter() {
                config_toml.insert(key, item.clone());
@@ -64,6 +64,7 @@ pub struct ConfigToml {
    #[serde(with = "humantime_serde")]
    pub wal_redo_timeout: Duration,
    pub superuser: String,
+    pub initdb_cache_dir: Option<Utf8PathBuf>,
    pub page_cache_size: usize,
    pub max_file_descriptors: usize,
    pub pg_distrib_dir: Option<Utf8PathBuf>,
@@ -358,6 +359,7 @@ impl Default for ConfigToml {
            wal_redo_timeout: (humantime::parse_duration(DEFAULT_WAL_REDO_TIMEOUT)
                .expect("cannot parse default wal redo timeout")),
            superuser: (DEFAULT_SUPERUSER.to_string()),
+            initdb_cache_dir: None,
            page_cache_size: (DEFAULT_PAGE_CACHE_SIZE),
            max_file_descriptors: (DEFAULT_MAX_FILE_DESCRIPTORS),
            pg_distrib_dir: None, // Utf8PathBuf::from("./pg_install"), // TODO: formely, this was std::env::current_dir()
@@ -263,15 +263,6 @@ impl Key {
        field5: u8::MAX,
        field6: u32::MAX,
    };
-    /// A key slightly smaller than [`Key::MAX`] for use in layer key ranges to avoid them to be confused with L0 layers
-    pub const NON_L0_MAX: Key = Key {
-        field1: u8::MAX,
-        field2: u32::MAX,
-        field3: u32::MAX,
-        field4: u32::MAX,
-        field5: u8::MAX,
-        field6: u32::MAX - 1,
-    };

    pub fn from_hex(s: &str) -> Result<Self> {
        if s.len() != 36 {
@@ -62,7 +62,7 @@ use bytes::{Buf, BufMut, Bytes, BytesMut};
    serde::Serialize,
    serde::Deserialize,
    strum_macros::Display,
-    strum_macros::EnumVariantNames,
+    strum_macros::VariantNames,
    strum_macros::AsRefStr,
    strum_macros::IntoStaticStr,
 )]
@@ -89,8 +89,19 @@ impl PageserverUtilization {

    /// If a node is currently hosting more work than it can comfortably handle.  This does not indicate that
    /// it will fail, but it is a strong signal that more work should not be added unless there is no alternative.
+    ///
+    /// When a node is overloaded, we may override soft affinity preferences and do things like scheduling
+    /// into a node in a less desirable AZ, if all the nodes in the preferred AZ are overloaded.
    pub fn is_overloaded(score: RawScore) -> bool {
-        score >= Self::UTILIZATION_FULL
+        // Why the factor of two?  This is unscientific but reflects behavior of real systems:
+        // - In terms of shard counts, a node's preferred max count is a soft limit intended to keep
+        //   startup and housekeeping jobs nice and responsive.  We can go to double this limit if needed
+        //   until some more nodes are deployed.
+        // - In terms of disk space, the node's utilization heuristic assumes every tenant needs to
+        //   hold its biggest timeline fully on disk, which is tends to be an over estimate when
+        //   some tenants are very idle and have dropped layers from disk.  In practice going up to
+        //   double is generally better than giving up and scheduling in a sub-optimal AZ.
+        score >= 2 * Self::UTILIZATION_FULL
    }

    pub fn adjust_shard_count_max(&mut self, shard_count: u32) {
@@ -81,17 +81,16 @@ pub fn is_expected_io_error(e: &io::Error) -> bool {
    )
 }

-#[async_trait::async_trait]
 pub trait Handler<IO> {
    /// Handle single query.
    /// postgres_backend will issue ReadyForQuery after calling this (this
    /// might be not what we want after CopyData streaming, but currently we don't
    /// care). It will also flush out the output buffer.
-    async fn process_query(
+    fn process_query(
        &mut self,
        pgb: &mut PostgresBackend<IO>,
        query_string: &str,
-    ) -> Result<(), QueryError>;
+    ) -> impl Future<Output = Result<(), QueryError>>;

    /// Called on startup packet receival, allows to process params.
    ///
@@ -23,7 +23,6 @@ async fn make_tcp_pair() -> (TcpStream, TcpStream) {

 struct TestHandler {}

-#[async_trait::async_trait]
 impl<IO: AsyncRead + AsyncWrite + Unpin + Send> Handler<IO> for TestHandler {
    // return single col 'hey' for any query
    async fn process_query(
@@ -121,6 +121,7 @@ fn main() -> anyhow::Result<()> {
            .allowlist_type("XLogPageHeaderData")
            .allowlist_type("XLogLongPageHeaderData")
            .allowlist_var("XLOG_PAGE_MAGIC")
+            .allowlist_var("PG_MAJORVERSION_NUM")
            .allowlist_var("PG_CONTROL_FILE_SIZE")
            .allowlist_var("PG_CONTROLFILEDATA_OFFSETOF_CRC")
            .allowlist_type("PageHeaderData")
@@ -44,6 +44,9 @@ macro_rules! postgres_ffi {
            // Re-export some symbols from bindings
            pub use bindings::DBState_DB_SHUTDOWNED;
            pub use bindings::{CheckPoint, ControlFileData, XLogRecord};
+
+            pub const ZERO_CHECKPOINT: bytes::Bytes =
+                bytes::Bytes::from_static(&[0u8; xlog_utils::SIZEOF_CHECKPOINT]);
        }
    };
 }
@@ -106,6 +109,107 @@ macro_rules! dispatch_pgversion {
    };
 }

+#[macro_export]
+macro_rules! enum_pgversion_dispatch {
+    ($name:expr, $typ:ident, $bind:ident, $code:block) => {
+        enum_pgversion_dispatch!(
+            name = $name,
+            bind = $bind,
+            typ = $typ,
+            code = $code,
+            pgversions = [
+                V14 : v14,
+                V15 : v15,
+                V16 : v16,
+            ]
+        )
+    };
+    (name = $name:expr,
+     bind = $bind:ident,
+     typ = $typ:ident,
+     code = $code:block,
+     pgversions = [$($variant:ident : $md:ident),+ $(,)?]) => {
+        match $name {
+            $(
+            self::$typ::$variant($bind) => {
+                use $crate::$md as pgv;
+                $code
+            }
+            ),+,
+        }
+    };
+}
+
+#[macro_export]
+macro_rules! enum_pgversion {
+    {$name:ident, pgv :: $t:ident} => {
+        enum_pgversion!{
+            name = $name,
+            typ = $t,
+            pgversions = [
+                V14 : v14,
+                V15 : v15,
+                V16 : v16,
+            ]
+        }
+    };
+    {$name:ident, pgv :: $p:ident :: $t:ident} => {
+        enum_pgversion!{
+            name = $name,
+            path = $p,
+            typ = $t,
+            pgversions = [
+                V14 : v14,
+                V15 : v15,
+                V16 : v16,
+            ]
+        }
+    };
+    {name = $name:ident,
+     typ = $t:ident,
+     pgversions = [$($variant:ident : $md:ident),+ $(,)?]} => {
+        pub enum $name {
+            $($variant ( $crate::$md::$t )),+
+        }
+        impl self::$name {
+            pub fn pg_version(&self) -> u32 {
+                enum_pgversion_dispatch!(self, $name, _ign, {
+                    pgv::bindings::PG_MAJORVERSION_NUM
+                })
+            }
+        }
+        $(
+        impl Into<self::$name> for $crate::$md::$t {
+            fn into(self) -> self::$name {
+                self::$name::$variant (self)
+            }
+        }
+        )+
+    };
+    {name = $name:ident,
+     path = $p:ident,
+     typ = $t:ident,
+     pgversions = [$($variant:ident : $md:ident),+ $(,)?]} => {
+        pub enum $name {
+            $($variant ($crate::$md::$p::$t)),+
+        }
+        impl $name {
+            pub fn pg_version(&self) -> u32 {
+                enum_pgversion_dispatch!(self, $name, _ign, {
+                    pgv::bindings::PG_MAJORVERSION_NUM
+                })
+            }
+        }
+        $(
+        impl Into<$name> for $crate::$md::$p::$t {
+            fn into(self) -> $name {
+                $name::$variant (self)
+            }
+        }
+        )+
+    };
+}
+
 pub mod pg_constants;
 pub mod relfile_utils;

@@ -185,7 +185,7 @@ mod tests {
    use super::*;

    fn parse(input: &str) -> anyhow::Result<RemoteStorageConfig> {
-        let toml = input.parse::<toml_edit::Document>().unwrap();
+        let toml = input.parse::<toml_edit::DocumentMut>().unwrap();
        RemoteStorageConfig::from_toml(toml.as_item())
    }

@@ -3,11 +3,9 @@ use std::str::FromStr;
 use anyhow::Context;
 use metrics::{IntCounter, IntCounterVec};
 use once_cell::sync::Lazy;
-use strum_macros::{EnumString, EnumVariantNames};
+use strum_macros::{EnumString, VariantNames};

-#[derive(
-    EnumString, strum_macros::Display, EnumVariantNames, Eq, PartialEq, Debug, Clone, Copy,
-)]
+#[derive(EnumString, strum_macros::Display, VariantNames, Eq, PartialEq, Debug, Clone, Copy)]
 #[strum(serialize_all = "snake_case")]
 pub enum LogFormat {
    Plain,
@@ -10,7 +10,7 @@ pub fn deserialize_item<T>(item: &toml_edit::Item) -> Result<T, Error>
 where
    T: serde::de::DeserializeOwned,
 {
-    let document: toml_edit::Document = match item {
+    let document: toml_edit::DocumentMut = match item {
        toml_edit::Item::Table(toml) => toml.clone().into(),
        toml_edit::Item::Value(toml_edit::Value::InlineTable(toml)) => {
            toml.clone().into_table().into()
@@ -1,6 +1,6 @@
-use std::{pin::Pin, sync::Arc};
+use std::pin::Pin;

-use futures::{SinkExt, StreamExt};
+use futures::SinkExt;
 use pageserver_api::{
    models::{
        PagestreamBeMessage, PagestreamFeMessage, PagestreamGetPageRequest,
@@ -10,6 +10,7 @@ use pageserver_api::{
 };
 use tokio::task::JoinHandle;
 use tokio_postgres::CopyOutStream;
+use tokio_stream::StreamExt;
 use tokio_util::sync::CancellationToken;
 use utils::{
    id::{TenantId, TimelineId},
@@ -135,68 +136,18 @@ impl PagestreamClient {
        drop(copy_both);
    }

-    pub fn split(self) -> (PagestreamTx, PagestreamRx) {
-        let Self {
-            copy_both,
-            cancel_on_client_drop,
-            conn_task,
-        } = self;
-        let keep_client_alive = KeepClientAlive {
-            client: conn_task,
-            cancel_on_client_drop: cancel_on_client_drop.unwrap(),
-        };
-        let keep_client_alive = Arc::new(keep_client_alive);
-        let (sink, stream): (
-            futures::stream::SplitSink<
-                Pin<Box<tokio_postgres::CopyBothDuplex<bytes::Bytes>>>,
-                bytes::Bytes,
-            >,
-            futures::stream::SplitStream<Pin<Box<tokio_postgres::CopyBothDuplex<bytes::Bytes>>>>,
-        ) = copy_both.split();
-        (
-            PagestreamTx {
-                sink,
-                keep_client_alive: keep_client_alive.clone(),
-            },
-            PagestreamRx {
-                stream,
-                keep_client_alive,
-            },
-        )
-    }
-}
-
-struct KeepClientAlive {
-    client: JoinHandle<()>,
-    cancel_on_client_drop: tokio_util::sync::DropGuard,
-}
-
-pub struct PagestreamTx {
-    sink: futures::stream::SplitSink<
-        Pin<Box<tokio_postgres::CopyBothDuplex<bytes::Bytes>>>,
-        bytes::Bytes,
-    >,
-    keep_client_alive: Arc<KeepClientAlive>,
-}
-
-pub struct PagestreamRx {
-    stream: futures::stream::SplitStream<Pin<Box<tokio_postgres::CopyBothDuplex<bytes::Bytes>>>>,
-    keep_client_alive: Arc<KeepClientAlive>,
-}
-
-impl PagestreamTx {
-    pub async fn send_getpage(&mut self, req: PagestreamGetPageRequest) -> anyhow::Result<()> {
+    pub async fn getpage(
+        &mut self,
+        req: PagestreamGetPageRequest,
+    ) -> anyhow::Result<PagestreamGetPageResponse> {
        let req = PagestreamFeMessage::GetPage(req);
        let req: bytes::Bytes = req.serialize();
-        let mut req = tokio_stream::once(Ok(req.clone()));
-        self.sink.send_all(&mut req).await?;
-        Ok(())
-    }
-}
+        // let mut req = tokio_util::io::ReaderStream::new(&req);
+        let mut req = tokio_stream::once(Ok(req));

-impl PagestreamRx {
-    pub async fn recv_getpage(&mut self) -> anyhow::Result<PagestreamGetPageResponse> {
-        let next: Option<Result<bytes::Bytes, _>> = self.stream.next().await;
+        self.copy_both.send_all(&mut req).await?;
+
+        let next: Option<Result<bytes::Bytes, _>> = self.copy_both.next().await;
        let next: bytes::Bytes = next.unwrap()?;

        let msg = PagestreamBeMessage::deserialize(next)?;
@@ -108,7 +108,6 @@ fn parse_filename(name: &str) -> (Range<Key>, Range<Lsn>) {
 enum LineKind {
    GcCutoff,
    Branch,
-    KeyVertical,
 }

 impl From<LineKind> for Fill {
@@ -116,7 +115,6 @@ impl From<LineKind> for Fill {
        match value {
            LineKind::GcCutoff => Fill::Color(rgb(255, 0, 0)),
            LineKind::Branch => Fill::Color(rgb(0, 255, 0)),
-            LineKind::KeyVertical => Fill::Color(rgb(0, 0, 255)),
        }
    }
 }
@@ -128,7 +126,6 @@ impl FromStr for LineKind {
        Ok(match s {
            "gc_cutoff" => LineKind::GcCutoff,
            "branch" => LineKind::Branch,
-            "key" => LineKind::KeyVertical,
            _ => anyhow::bail!("unsupported linekind: {s}"),
        })
    }
@@ -145,31 +142,25 @@ pub fn main() -> Result<()> {
    let stdin = io::stdin();

    let mut lines: Vec<(Lsn, LineKind)> = vec![];
-    let mut vertical_lines: Vec<(Key, LineKind)> = vec![];

    for (lineno, line) in stdin.lock().lines().enumerate() {
        let lineno = lineno + 1;

        let line = line.unwrap();
-        if let Some((kind, what)) = line.split_once(':') {
-            (|| {
-                match LineKind::from_str(kind).context("parse kind")? {
-                    kind @ LineKind::Branch | kind @ LineKind::GcCutoff => {
-                        let lsn = if what.contains('/') {
-                            Lsn::from_str(what)?
-                        } else {
-                            Lsn::from_hex(what)?
-                        };
-                        lines.push((lsn, kind));
+        if let Some((kind, lsn)) = line.split_once(':') {
+            let (kind, lsn) = LineKind::from_str(kind)
+                .context("parse kind")
+                .and_then(|kind| {
+                    if lsn.contains('/') {
+                        Lsn::from_str(lsn)
+                    } else {
+                        Lsn::from_hex(lsn)
                    }
-                    kind @ LineKind::KeyVertical => {
-                        let key = Key::from_hex(what).context("parse key")?;
-                        vertical_lines.push((key, kind));
-                    }
-                }
-                anyhow::Ok(())
-            })()
-            .with_context(|| format!("parse {line:?} on {lineno}"))?;
+                    .map(|lsn| (kind, lsn))
+                    .context("parse lsn")
+                })
+                .with_context(|| format!("parse {line:?} on {lineno}"))?;
+            lines.push((lsn, kind));
            continue;
        }
        let line = PathBuf::from_str(&line).unwrap();
@@ -184,7 +175,7 @@ pub fn main() -> Result<()> {
    }

    // Collect all coordinates
-    let mut keys: Vec<Key> = Vec::with_capacity(files.len() + vertical_lines.len());
+    let mut keys: Vec<Key> = Vec::with_capacity(files.len());
    let mut lsns: Vec<Lsn> = Vec::with_capacity(files.len() + lines.len());

    for Layer {
@@ -201,8 +192,6 @@ pub fn main() -> Result<()> {

    lsns.extend(lines.iter().map(|(lsn, _)| *lsn));

-    keys.extend(vertical_lines.iter().map(|(key, _)| *key));
-
    // Analyze
    let key_map = build_coordinate_compression_map(keys);
    let lsn_map = build_coordinate_compression_map(lsns);
@@ -294,25 +283,6 @@ pub fn main() -> Result<()> {
        );
    }

-    for (key, kind) in vertical_lines {
-        let key = *key_map.get(&key).unwrap();
-        let stretch = 2.0;
-        let xmargin = 0.05;
-        let ymargin = 0.05;
-        let lsn_diff = 0.3;
-        let lsn_offset = -lsn_diff / 2.0;
-        println!(
-            "{}",
-            rectangle(
-                5.0 + key as f32,
-                0.0,
-                (key_map.len() + 10) as f32,
-                lsn_map.len() as f32,
-            )
-            .fill(kind)
-        );
-    }
-
    println!("{}", EndSvg);

    eprintln!("num_images: {}", num_images);
@@ -174,7 +174,7 @@ async fn main() -> anyhow::Result<()> {
                println!("specified prefix '{}' failed validation", cmd.prefix);
                return Ok(());
            };
-            let toml_document = toml_edit::Document::from_str(&cmd.config_toml_str)?;
+            let toml_document = toml_edit::DocumentMut::from_str(&cmd.config_toml_str)?;
            let toml_item = toml_document
                .get("remote_storage")
                .expect("need remote_storage");
@@ -13,7 +13,7 @@ use rand::prelude::*;
 use tokio::task::JoinSet;
 use tracing::info;

-use std::collections::{HashSet, VecDeque};
+use std::collections::HashSet;
 use std::future::Future;
 use std::num::NonZeroUsize;
 use std::pin::Pin;
@@ -295,58 +295,64 @@ async fn main_impl(
                .await
                .unwrap();

-            let (mut pagestream_tx, mut pagestream_rx) = client.split();
-
            start_work_barrier.wait().await;
            let client_start = Instant::now();
            let mut ticks_processed = 0;
-            let (rq_tx, mut rq_rx) = tokio::sync::mpsc::channel(4096);
-            let sender = tokio::spawn(async move {
-                while !cancel.is_cancelled() {
-                    let start = Instant::now();
-                    let req = {
-                        let mut rng = rand::thread_rng();
-                        let r = &ranges[weights.sample(&mut rng)];
-                        let key: i128 = rng.gen_range(r.start..r.end);
-                        let key = Key::from_i128(key);
-                        assert!(key.is_rel_block_key());
-                        let (rel_tag, block_no) = key
-                            .to_rel_block()
-                            .expect("we filter non-rel-block keys out above");
-                        PagestreamGetPageRequest {
-                            request_lsn: if rng.gen_bool(args.req_latest_probability) {
-                                Lsn::MAX
-                            } else {
-                                r.timeline_lsn
-                            },
-                            not_modified_since: r.timeline_lsn,
-                            rel: rel_tag,
-                            blkno: block_no,
-                        }
-                    };
-                    pagestream_tx.send_getpage(req).await.unwrap();
-                    rq_tx.send(start).await.unwrap();
-                }
-            });
-
-            let receiver = tokio::spawn(async move {
-                while let Some(start) = rq_rx.recv().await {
-                    let response = pagestream_rx.recv_getpage().await.unwrap();
-                    let end = Instant::now();
-                    live_stats.request_done();
-                    STATS.with(|stats| {
-                        stats
-                            .borrow()
-                            .lock()
-                            .unwrap()
-                            .observe(end.duration_since(start))
+            while !cancel.is_cancelled() {
+                // Detect if a request took longer than the RPS rate
+                if let Some(period) = &rps_period {
+                    let periods_passed_until_now =
+                        usize::try_from(client_start.elapsed().as_micros() / period.as_micros())
                            .unwrap();
-                    });
-                }
-            });

-            sender.await.unwrap();
-            receiver.await.unwrap();
+                    if periods_passed_until_now > ticks_processed {
+                        live_stats.missed((periods_passed_until_now - ticks_processed) as u64);
+                    }
+                    ticks_processed = periods_passed_until_now;
+                }
+
+                let start = Instant::now();
+                let req = {
+                    let mut rng = rand::thread_rng();
+                    let r = &ranges[weights.sample(&mut rng)];
+                    let key: i128 = rng.gen_range(r.start..r.end);
+                    let key = Key::from_i128(key);
+                    assert!(key.is_rel_block_key());
+                    let (rel_tag, block_no) = key
+                        .to_rel_block()
+                        .expect("we filter non-rel-block keys out above");
+                    PagestreamGetPageRequest {
+                        request_lsn: if rng.gen_bool(args.req_latest_probability) {
+                            Lsn::MAX
+                        } else {
+                            r.timeline_lsn
+                        },
+                        not_modified_since: r.timeline_lsn,
+                        rel: rel_tag,
+                        blkno: block_no,
+                    }
+                };
+                client.getpage(req).await.unwrap();
+                let end = Instant::now();
+                live_stats.request_done();
+                ticks_processed += 1;
+                STATS.with(|stats| {
+                    stats
+                        .borrow()
+                        .lock()
+                        .unwrap()
+                        .observe(end.duration_since(start))
+                        .unwrap();
+                });
+
+                if let Some(period) = &rps_period {
+                    let next_at = client_start
+                        + Duration::from_micros(
+                            (ticks_processed) as u64 * u64::try_from(period.as_micros()).unwrap(),
+                        );
+                    tokio::time::sleep_until(next_at.into()).await;
+                }
+            }
        })
    };

@@ -71,6 +71,8 @@ pub struct PageServerConf {

    pub superuser: String,

+    pub initdb_cache_dir: Option<Utf8PathBuf>,
+
    pub page_cache_size: usize,
    pub max_file_descriptors: usize,

@@ -309,6 +311,7 @@ impl PageServerConf {
            wait_lsn_timeout,
            wal_redo_timeout,
            superuser,
+            initdb_cache_dir,
            page_cache_size,
            max_file_descriptors,
            pg_distrib_dir,
@@ -356,6 +359,7 @@ impl PageServerConf {
            wait_lsn_timeout,
            wal_redo_timeout,
            superuser,
+            initdb_cache_dir,
            page_cache_size,
            max_file_descriptors,
            http_auth_type,
@@ -9,7 +9,7 @@ use metrics::{
 use once_cell::sync::Lazy;
 use pageserver_api::shard::TenantShardId;
 use strum::{EnumCount, VariantNames};
-use strum_macros::{EnumVariantNames, IntoStaticStr};
+use strum_macros::{IntoStaticStr, VariantNames};
 use tracing::warn;
 use utils::id::TimelineId;

@@ -27,7 +27,7 @@ const CRITICAL_OP_BUCKETS: &[f64] = &[
 ];

 // Metrics collected on operations on the storage repository.
-#[derive(Debug, EnumVariantNames, IntoStaticStr)]
+#[derive(Debug, VariantNames, IntoStaticStr)]
 #[strum(serialize_all = "kebab_case")]
 pub(crate) enum StorageTimeOperation {
    #[strum(serialize = "layer flush")]
@@ -1185,7 +1185,6 @@ struct GlobalAndPerTimelineHistogramTimer<'a, 'c> {
    ctx: &'c RequestContext,
    start: std::time::Instant,
    op: SmgrQueryType,
-    count: usize,
 }

 impl<'a, 'c> Drop for GlobalAndPerTimelineHistogramTimer<'a, 'c> {
@@ -1213,11 +1212,9 @@ impl<'a, 'c> Drop for GlobalAndPerTimelineHistogramTimer<'a, 'c> {
                elapsed
            }
        };
-        for _ in 0..self.count {
-            self.global_metric.observe(ex_throttled.as_secs_f64());
-            if let Some(timeline_metric) = self.timeline_metric {
-                timeline_metric.observe(ex_throttled.as_secs_f64());
-            }
+        self.global_metric.observe(ex_throttled.as_secs_f64());
+        if let Some(timeline_metric) = self.timeline_metric {
+            timeline_metric.observe(ex_throttled.as_secs_f64());
        }
    }
 }
@@ -1346,14 +1343,6 @@ impl SmgrQueryTimePerTimeline {
        &'a self,
        op: SmgrQueryType,
        ctx: &'c RequestContext,
-    ) -> Option<impl Drop + '_> {
-        self.start_timer_many(op, 1, ctx)
-    }
-    pub(crate) fn start_timer_many<'c: 'a, 'a>(
-        &'a self,
-        op: SmgrQueryType,
-        count: usize,
-        ctx: &'c RequestContext,
    ) -> Option<impl Drop + '_> {
        let global_metric = &self.global_metrics[op as usize];
        let start = Instant::now();
@@ -1387,7 +1376,6 @@ impl SmgrQueryTimePerTimeline {
            ctx,
            start,
            op,
-            count,
        })
    }
 }
@@ -3182,16 +3170,6 @@ static TOKIO_EXECUTOR_THREAD_COUNT: Lazy<UIntGaugeVec> = Lazy::new(|| {
    .unwrap()
 });

-pub(crate) static CONSECUTIVE_NONBLOCKING_GETPAGE_REQUESTS_HISTOGRAM: Lazy<Histogram> =
-    Lazy::new(|| {
-        register_histogram!(
-            "pageserver_consecutive_nonblocking_getpage_requests",
-            "Number of consecutive nonblocking getpage requests",
-            (0..=256).map(|x| x as f64).collect::<Vec<f64>>(),
-        )
-        .unwrap()
-    });
-
 pub(crate) fn set_tokio_runtime_setup(setup: &str, num_threads: NonZeroUsize) {
    static SERIALIZE: std::sync::Mutex<()> = std::sync::Mutex::new(());
    let _guard = SERIALIZE.lock().unwrap();
@@ -5,14 +5,14 @@ use anyhow::Context;
 use async_compression::tokio::write::GzipEncoder;
 use bytes::Buf;
 use futures::FutureExt;
-use once_cell::sync::{Lazy, OnceCell};
-use pageserver_api::models::{self, TenantState};
+use once_cell::sync::OnceCell;
+use pageserver_api::models::TenantState;
 use pageserver_api::models::{
    PagestreamBeMessage, PagestreamDbSizeRequest, PagestreamDbSizeResponse,
    PagestreamErrorResponse, PagestreamExistsRequest, PagestreamExistsResponse,
-    PagestreamFeMessage, PagestreamGetPageRequest, PagestreamGetSlruSegmentRequest,
-    PagestreamGetSlruSegmentResponse, PagestreamNblocksRequest, PagestreamNblocksResponse,
-    PagestreamProtocolVersion,
+    PagestreamFeMessage, PagestreamGetPageRequest, PagestreamGetPageResponse,
+    PagestreamGetSlruSegmentRequest, PagestreamGetSlruSegmentResponse, PagestreamNblocksRequest,
+    PagestreamNblocksResponse, PagestreamProtocolVersion,
 };
 use pageserver_api::shard::TenantShardId;
 use postgres_backend::{is_expected_io_error, AuthType, PostgresBackend, QueryError};
@@ -43,7 +43,7 @@ use crate::basebackup;
 use crate::basebackup::BasebackupError;
 use crate::config::PageServerConf;
 use crate::context::{DownloadBehavior, RequestContext};
-use crate::metrics::{self, CONSECUTIVE_NONBLOCKING_GETPAGE_REQUESTS_HISTOGRAM};
+use crate::metrics;
 use crate::metrics::{ComputeCommandKind, COMPUTE_COMMANDS_COUNTERS, LIVE_CONNECTIONS};
 use crate::pgdatadir_mapping::Version;
 use crate::span::debug_assert_current_span_has_tenant_and_timeline_id;
@@ -58,7 +58,7 @@ use crate::tenant::GetTimelineError;
 use crate::tenant::PageReconstructError;
 use crate::tenant::Timeline;
 use pageserver_api::key::rel_block_to_key;
-use pageserver_api::reltag::{BlockNumber, RelTag, SlruKind};
+use pageserver_api::reltag::SlruKind;
 use postgres_ffi::pg_constants::DEFAULTTABLESPACE_OID;
 use postgres_ffi::BLCKSZ;

@@ -577,326 +577,124 @@ impl PageServerHandler {
            }
        }

-        let mut batched = None;
-        'outer: loop {
-            enum DebouncedFeMessage {
-                Exists(models::PagestreamExistsRequest),
-                Nblocks(models::PagestreamNblocksRequest),
-                GetPage {
-                    span: Span,
-                    shard: timeline::handle::Handle<TenantManagerTypes>,
-                    effective_request_lsn: Lsn,
-                    pages: smallvec::SmallVec<[(RelTag, BlockNumber); 1]>,
-                },
-                DbSize(models::PagestreamDbSizeRequest),
-                GetSlruSegment(models::PagestreamGetSlruSegmentRequest),
-                RespondError(Span, PageStreamError),
-            }
-            let mut debounce: Option<std::time::Instant> = None;
-            // return or `?` on protocol error
-            // `break EXPR` to stop batching. The EXPR will be the first message in the next batch.
-            let next_batched: Option<DebouncedFeMessage> = loop {
-                static BOUNCE_TIMEOUT: Lazy<Duration> = Lazy::new(|| {
-                    utils::env::var::<humantime::Duration, _>("NEON_PAGESERVER_DEBOUNCE")
-                        .unwrap()
-                        .into()
-                });
-                let sleep_fut = if let Some(started_at) = debounce {
-                    futures::future::Either::Left(tokio::time::sleep_until(
-                        (started_at + *BOUNCE_TIMEOUT).into(),
-                    ))
-                } else {
-                    futures::future::Either::Right(futures::future::pending())
-                };
-                let msg = tokio::select! {
-                    biased;
-                    _ = self.cancel.cancelled() => {
-                        return Err(QueryError::Shutdown)
-                    }
-                    msg = pgb.read_message() => {
-                        msg
-                    }
-                    _ = sleep_fut => {
-                        assert!(batched.is_some());
-                        break None;
-                    }
-                };
-                let copy_data_bytes = match msg? {
-                    Some(FeMessage::CopyData(bytes)) => bytes,
-                    Some(FeMessage::Terminate) => break 'outer,
-                    Some(m) => {
-                        return Err(QueryError::Other(anyhow::anyhow!(
-                            "unexpected message: {m:?} during COPY"
-                        )));
-                    }
-                    None => break 'outer, // client disconnected
-                };
-                trace!("query: {copy_data_bytes:?}");
-                fail::fail_point!("ps::handle-pagerequest-message");
-
-                // parse request
-                let neon_fe_msg = PagestreamFeMessage::parse(&mut copy_data_bytes.reader())?;
-
-                let this_msg = match neon_fe_msg {
-                    PagestreamFeMessage::Exists(msg) => DebouncedFeMessage::Exists(msg),
-                    PagestreamFeMessage::Nblocks(msg) => DebouncedFeMessage::Nblocks(msg),
-                    PagestreamFeMessage::DbSize(msg) => DebouncedFeMessage::DbSize(msg),
-                    PagestreamFeMessage::GetSlruSegment(msg) => {
-                        DebouncedFeMessage::GetSlruSegment(msg)
-                    }
-                    PagestreamFeMessage::GetPage(PagestreamGetPageRequest {
-                        request_lsn,
-                        not_modified_since,
-                        rel,
-                        blkno,
-                    }) => {
-                        let span = tracing::info_span!("handle_get_page_at_lsn_request_batched", %tenant_id, %timeline_id, shard_id = tracing::field::Empty, req_lsn = %request_lsn, batch_size = tracing::field::Empty, batch_id = tracing::field::Empty);
-                        let key = rel_block_to_key(rel, blkno);
-                        let shard = match self
-                            .timeline_handles
-                            .get(tenant_id, timeline_id, ShardSelector::Page(key))
-                            .instrument(span.clone())
-                            .await
-                        {
-                            Ok(tl) => tl,
-                            Err(GetActiveTimelineError::Tenant(
-                                GetActiveTenantError::NotFound(_),
-                            )) => {
-                                // We already know this tenant exists in general, because we resolved it at
-                                // start of connection.  Getting a NotFound here indicates that the shard containing
-                                // the requested page is not present on this node: the client's knowledge of shard->pageserver
-                                // mapping is out of date.
-                                //
-                                // Closing the connection by returning ``::Reconnect` has the side effect of rate-limiting above message, via
-                                // client's reconnect backoff, as well as hopefully prompting the client to load its updated configuration
-                                // and talk to a different pageserver.
-                                break Some(DebouncedFeMessage::RespondError(
-                                    span,
-                                    PageStreamError::Reconnect(
-                                        "getpage@lsn request routed to wrong shard".into(),
-                                    ),
-                                ));
-                            }
-                            Err(e) => break Some(DebouncedFeMessage::RespondError(span, e.into())),
-                        };
-                        let effective_request_lsn = match Self::wait_or_get_last_lsn(
-                            &shard,
-                            request_lsn,
-                            not_modified_since,
-                            &shard.get_latest_gc_cutoff_lsn(),
-                            &ctx,
-                        )
-                        // TODO: if we actually need to wait for lsn here, it delays the entire batch which doesn't need to wait
-                        .await
-                        {
-                            Ok(lsn) => lsn,
-                            Err(e) => {
-                                break Some(DebouncedFeMessage::RespondError(span, e));
-                            }
-                        };
-                        DebouncedFeMessage::GetPage {
-                            span,
-                            shard,
-                            effective_request_lsn,
-                            pages: smallvec::smallvec![(rel, blkno)],
-                        }
-                    }
-                };
-
-                // check if we can debounce
-                match (&mut batched, this_msg) {
-                    (None, this_msg) => {
-                        batched = Some(this_msg);
-                    }
-                    (
-                        Some(DebouncedFeMessage::GetPage {
-                            span: _,
-                            shard: accum_shard,
-                            pages: accum_pages,
-                            effective_request_lsn: accum_lsn,
-                        }),
-                        DebouncedFeMessage::GetPage {
-                            span: _,
-                            shard: this_shard,
-                            pages: this_pages,
-                            effective_request_lsn: this_lsn,
-                        },
-                    ) if async {
-                        assert_eq!(this_pages.len(), 1);
-                        if accum_pages.len() >= Timeline::MAX_GET_VECTORED_KEYS as usize {
-                            assert_eq!(accum_pages.len(), Timeline::MAX_GET_VECTORED_KEYS as usize);
-                            return false;
-                        }
-                        if (accum_shard.tenant_shard_id, accum_shard.timeline_id)
-                            != (this_shard.tenant_shard_id, this_shard.timeline_id)
-                        {
-                            // TODO: we _could_ batch & execute each shard seperately (and in parallel).
-                            // But the current logig for keeping responses in order does not support that.
-                            return false;
-                        }
-                        // the vectored get currently only supports a single LSN, so, bounce as soon
-                        // as the effective request_lsn changes
-                        if (*accum_lsn != this_lsn) {
-                            return false;
-                        }
-                        return true;
-                    }
-                    .await =>
-                    {
-                        // ok to batch
-                        accum_pages.extend(this_pages);
-                    }
-                    (Some(_), this_msg) => {
-                        // by default, don't continue batching
-                        break Some(this_msg);
-                    }
+        loop {
+            // read request bytes (it's exactly 1 PagestreamFeMessage per CopyData)
+            let msg = tokio::select! {
+                biased;
+                _ = self.cancel.cancelled() => {
+                    return Err(QueryError::Shutdown)
                }
-
-                // debounce impl piece
-                let started_at = debounce.get_or_insert_with(Instant::now);
-                if started_at.elapsed() > *BOUNCE_TIMEOUT {
-                    break None;
+                msg = pgb.read_message() => { msg }
+            };
+            let copy_data_bytes = match msg? {
+                Some(FeMessage::CopyData(bytes)) => bytes,
+                Some(FeMessage::Terminate) => break,
+                Some(m) => {
+                    return Err(QueryError::Other(anyhow::anyhow!(
+                        "unexpected message: {m:?} during COPY"
+                    )));
                }
+                None => break, // client disconnected
            };

+            trace!("query: {copy_data_bytes:?}");
+            fail::fail_point!("ps::handle-pagerequest-message");
+
+            // parse request
+            let neon_fe_msg = PagestreamFeMessage::parse(&mut copy_data_bytes.reader())?;
+
            // invoke handler function
-            let (handler_results, span): (
-                smallvec::SmallVec<[Result<PagestreamBeMessage, PageStreamError>; 1]>,
-                _,
-            ) = match batched.take().expect("loop above ensures this") {
-                DebouncedFeMessage::Exists(req) => {
+            let (handler_result, span) = match neon_fe_msg {
+                PagestreamFeMessage::Exists(req) => {
                    fail::fail_point!("ps::handle-pagerequest-message::exists");
                    let span = tracing::info_span!("handle_get_rel_exists_request", rel = %req.rel, req_lsn = %req.request_lsn);
                    (
-                        smallvec::smallvec![
-                            self.handle_get_rel_exists_request(tenant_id, timeline_id, &req, &ctx)
-                                .instrument(span.clone())
-                                .await
-                        ],
+                        self.handle_get_rel_exists_request(tenant_id, timeline_id, &req, &ctx)
+                            .instrument(span.clone())
+                            .await,
                        span,
                    )
                }
-                DebouncedFeMessage::Nblocks(req) => {
+                PagestreamFeMessage::Nblocks(req) => {
                    fail::fail_point!("ps::handle-pagerequest-message::nblocks");
                    let span = tracing::info_span!("handle_get_nblocks_request", rel = %req.rel, req_lsn = %req.request_lsn);
                    (
-                        smallvec::smallvec![
-                            self.handle_get_nblocks_request(tenant_id, timeline_id, &req, &ctx)
-                                .instrument(span.clone())
-                                .await,
-                        ],
+                        self.handle_get_nblocks_request(tenant_id, timeline_id, &req, &ctx)
+                            .instrument(span.clone())
+                            .await,
                        span,
                    )
                }
-                DebouncedFeMessage::GetPage {
-                    span,
-                    shard,
-                    effective_request_lsn,
-                    pages,
-                } => {
-                    CONSECUTIVE_NONBLOCKING_GETPAGE_REQUESTS_HISTOGRAM.observe(pages.len() as f64);
-                    span.record("batch_size", pages.len() as u64);
-                    static BATCH_ID: Lazy<std::sync::atomic::AtomicUsize> =
-                        Lazy::new(|| std::sync::atomic::AtomicUsize::new(0));
-                    span.record(
-                        "batch_id",
-                        BATCH_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed) as u64,
-                    );
+                PagestreamFeMessage::GetPage(req) => {
                    fail::fail_point!("ps::handle-pagerequest-message::getpage");
                    // shard_id is filled in by the handler
+                    let span = tracing::info_span!("handle_get_page_at_lsn_request", rel = %req.rel, blkno = %req.blkno, req_lsn = %req.request_lsn);
                    (
-                        {
-                            let npages = pages.len();
-                            let res = self
-                                .handle_get_page_at_lsn_request_batched(
-                                    &shard,
-                                    effective_request_lsn,
-                                    pages,
-                                    &ctx,
-                                )
-                                .instrument(span.clone())
-                                .await;
-                            assert_eq!(res.len(), npages);
-                            res
-                        },
+                        self.handle_get_page_at_lsn_request(tenant_id, timeline_id, &req, &ctx)
+                            .instrument(span.clone())
+                            .await,
                        span,
                    )
                }
-                DebouncedFeMessage::DbSize(req) => {
+                PagestreamFeMessage::DbSize(req) => {
                    fail::fail_point!("ps::handle-pagerequest-message::dbsize");
                    let span = tracing::info_span!("handle_db_size_request", dbnode = %req.dbnode, req_lsn = %req.request_lsn);
                    (
-                        smallvec::smallvec![
-                            self.handle_db_size_request(tenant_id, timeline_id, &req, &ctx)
-                                .instrument(span.clone())
-                                .await
-                        ],
+                        self.handle_db_size_request(tenant_id, timeline_id, &req, &ctx)
+                            .instrument(span.clone())
+                            .await,
                        span,
                    )
                }
-                DebouncedFeMessage::GetSlruSegment(req) => {
+                PagestreamFeMessage::GetSlruSegment(req) => {
                    fail::fail_point!("ps::handle-pagerequest-message::slrusegment");
                    let span = tracing::info_span!("handle_get_slru_segment_request", kind = %req.kind, segno = %req.segno, req_lsn = %req.request_lsn);
                    (
-                        smallvec::smallvec![
-                            self.handle_get_slru_segment_request(
-                                tenant_id,
-                                timeline_id,
-                                &req,
-                                &ctx
-                            )
+                        self.handle_get_slru_segment_request(tenant_id, timeline_id, &req, &ctx)
                            .instrument(span.clone())
-                            .await
-                        ],
+                            .await,
                        span,
                    )
                }
-                DebouncedFeMessage::RespondError(span, e) => {
-                    // We've already decided to respond with an error, so we don't need to
-                    // call the handler.
-                    (smallvec::smallvec![Err(e)], span)
-                }
            };

            // Map handler result to protocol behavior.
            // 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 in handler_results {
-                let response_msg = match handler_result {
-                    Err(e) => match &e {
-                        PageStreamError::Shutdown => {
-                            // If we fail to fulfil a request during shutdown, which may be _because_ of
-                            // shutdown, then do not send the error to the client.  Instead just drop the
-                            // connection.
-                            span.in_scope(|| info!("dropping connection due to shutdown"));
-                            return Err(QueryError::Shutdown);
-                        }
-                        PageStreamError::Reconnect(reason) => {
-                            span.in_scope(|| info!("handler requested reconnect: {reason}"));
-                            return Err(QueryError::Reconnect);
-                        }
-                        PageStreamError::Read(_)
-                        | PageStreamError::LsnTimeout(_)
-                        | PageStreamError::NotFound(_)
-                        | PageStreamError::BadRequest(_) => {
-                            // print the all details to the log with {:#}, but for the client the
-                            // error message is enough.  Do not log if shutting down, as the anyhow::Error
-                            // here includes cancellation which is not an error.
-                            let full = utils::error::report_compact_sources(&e);
-                            span.in_scope(|| {
-                                error!("error reading relation or page version: {full:#}")
-                            });
-                            PagestreamBeMessage::Error(PagestreamErrorResponse {
-                                message: e.to_string(),
-                            })
-                        }
-                    },
-                    Ok(response_msg) => response_msg,
-                };
+            let response_msg = match handler_result {
+                Err(e) => match &e {
+                    PageStreamError::Shutdown => {
+                        // If we fail to fulfil a request during shutdown, which may be _because_ of
+                        // shutdown, then do not send the error to the client.  Instead just drop the
+                        // connection.
+                        span.in_scope(|| info!("dropping connection due to shutdown"));
+                        return Err(QueryError::Shutdown);
+                    }
+                    PageStreamError::Reconnect(reason) => {
+                        span.in_scope(|| info!("handler requested reconnect: {reason}"));
+                        return Err(QueryError::Reconnect);
+                    }
+                    PageStreamError::Read(_)
+                    | PageStreamError::LsnTimeout(_)
+                    | PageStreamError::NotFound(_)
+                    | PageStreamError::BadRequest(_) => {
+                        // print the all details to the log with {:#}, but for the client the
+                        // error message is enough.  Do not log if shutting down, as the anyhow::Error
+                        // here includes cancellation which is not an error.
+                        let full = utils::error::report_compact_sources(&e);
+                        span.in_scope(|| {
+                            error!("error reading relation or page version: {full:#}")
+                        });
+                        PagestreamBeMessage::Error(PagestreamErrorResponse {
+                            message: e.to_string(),
+                        })
+                    }
+                },
+                Ok(response_msg) => response_msg,
+            };

-                // marshal & transmit response message
-                pgb.write_message_noflush(&BeMessage::CopyData(&response_msg.serialize()))?;
-            }
+            // marshal & transmit response message
+            pgb.write_message_noflush(&BeMessage::CopyData(&response_msg.serialize()))?;
            tokio::select! {
                biased;
                _ = self.cancel.cancelled() => {
@@ -908,9 +706,6 @@ impl PageServerHandler {
                    res?;
                }
            }
-
-            assert!(batched.is_none(), "we take() earlier");
-            batched = next_batched;
        }
        Ok(())
    }
@@ -1154,30 +949,60 @@ impl PageServerHandler {
        }))
    }

-    #[instrument(skip_all)]
-    async fn handle_get_page_at_lsn_request_batched(
+    #[instrument(skip_all, fields(shard_id))]
+    async fn handle_get_page_at_lsn_request(
        &mut self,
-        timeline: &Timeline,
-        effective_lsn: Lsn,
-        pages: smallvec::SmallVec<[(RelTag, BlockNumber); 1]>,
+        tenant_id: TenantId,
+        timeline_id: TimelineId,
+        req: &PagestreamGetPageRequest,
        ctx: &RequestContext,
-    ) -> smallvec::SmallVec<[Result<PagestreamBeMessage, PageStreamError>; 1]> {
-        debug_assert_current_span_has_tenant_and_timeline_id();
-        let _timer = timeline.query_metrics.start_timer_many(
-            metrics::SmgrQueryType::GetPageAtLsn,
-            pages.len(),
+    ) -> Result<PagestreamBeMessage, PageStreamError> {
+        let timeline = match self
+            .timeline_handles
+            .get(
+                tenant_id,
+                timeline_id,
+                ShardSelector::Page(rel_block_to_key(req.rel, req.blkno)),
+            )
+            .await
+        {
+            Ok(tl) => tl,
+            Err(GetActiveTimelineError::Tenant(GetActiveTenantError::NotFound(_))) => {
+                // We already know this tenant exists in general, because we resolved it at
+                // start of connection.  Getting a NotFound here indicates that the shard containing
+                // the requested page is not present on this node: the client's knowledge of shard->pageserver
+                // mapping is out of date.
+                //
+                // Closing the connection by returning ``::Reconnect` has the side effect of rate-limiting above message, via
+                // client's reconnect backoff, as well as hopefully prompting the client to load its updated configuration
+                // and talk to a different pageserver.
+                return Err(PageStreamError::Reconnect(
+                    "getpage@lsn request routed to wrong shard".into(),
+                ));
+            }
+            Err(e) => return Err(e.into()),
+        };
+
+        let _timer = timeline
+            .query_metrics
+            .start_timer(metrics::SmgrQueryType::GetPageAtLsn, ctx);
+
+        let latest_gc_cutoff_lsn = timeline.get_latest_gc_cutoff_lsn();
+        let lsn = Self::wait_or_get_last_lsn(
+            &timeline,
+            req.request_lsn,
+            req.not_modified_since,
+            &latest_gc_cutoff_lsn,
            ctx,
-        );
+        )
+        .await?;

-        let pages = timeline
-            .get_rel_page_at_lsn_batched(pages, Version::Lsn(effective_lsn), ctx)
-            .await;
+        let page = timeline
+            .get_rel_page_at_lsn(req.rel, req.blkno, Version::Lsn(lsn), ctx)
+            .await?;

-        smallvec::SmallVec::from_iter(pages.into_iter().map(|page| {
-            page.map(|page| {
-                PagestreamBeMessage::GetPage(models::PagestreamGetPageResponse { page })
-            })
-            .map_err(PageStreamError::Read)
+        Ok(PagestreamBeMessage::GetPage(PagestreamGetPageResponse {
+            page,
        }))
    }

@@ -1374,7 +1199,6 @@ impl PageServerHandler {
    }
 }

-#[async_trait::async_trait]
 impl<IO> postgres_backend::Handler<IO> for PageServerHandler
 where
    IO: AsyncRead + AsyncWrite + Send + Sync + Unpin,
@@ -1674,10 +1498,3 @@ fn set_tracing_field_shard_id(timeline: &Timeline) {
    );
    debug_assert_current_span_has_tenant_and_timeline_id();
 }
-
-struct WaitedForLsn(Lsn);
-impl From<WaitedForLsn> for Lsn {
-    fn from(WaitedForLsn(lsn): WaitedForLsn) -> Self {
-        lsn
-    }
-}
@@ -9,17 +9,12 @@
 use super::tenant::{PageReconstructError, Timeline};
 use crate::context::RequestContext;
 use crate::keyspace::{KeySpace, KeySpaceAccum};
-use crate::span::{
-    debug_assert_current_span_has_tenant_and_timeline_id,
-    debug_assert_current_span_has_tenant_and_timeline_id_no_shard_id,
-};
-use crate::tenant::timeline::GetVectoredError;
+use crate::span::debug_assert_current_span_has_tenant_and_timeline_id_no_shard_id;
 use crate::walrecord::NeonWalRecord;
 use crate::{aux_file, repository::*};
 use anyhow::{ensure, Context};
 use bytes::{Buf, Bytes, BytesMut};
 use enum_map::Enum;
-use itertools::Itertools;
 use pageserver_api::key::{
    dbdir_key_range, rel_block_to_key, rel_dir_to_key, rel_key_range, rel_size_to_key,
    relmap_file_key, repl_origin_key, repl_origin_key_range, slru_block_to_key, slru_dir_to_key,
@@ -33,7 +28,7 @@ use postgres_ffi::relfile_utils::{FSM_FORKNUM, VISIBILITYMAP_FORKNUM};
 use postgres_ffi::BLCKSZ;
 use postgres_ffi::{Oid, RepOriginId, TimestampTz, TransactionId};
 use serde::{Deserialize, Serialize};
-use std::collections::{hash_map, BTreeMap, HashMap, HashSet};
+use std::collections::{hash_map, HashMap, HashSet};
 use std::ops::ControlFlow;
 use std::ops::Range;
 use strum::IntoEnumIterator;
@@ -196,184 +191,26 @@ impl Timeline {
        version: Version<'_>,
        ctx: &RequestContext,
    ) -> Result<Bytes, PageReconstructError> {
-        let pages = smallvec::smallvec![(tag, blknum)];
-        let res = self.get_rel_page_at_lsn_batched(pages, version, ctx).await;
-        assert_eq!(res.len(), 1);
-        res.into_iter().next().unwrap()
-    }
-
-    /// Like [`get_rel_page_at_lsn`], but returns a batch of pages.
-    pub(crate) async fn get_rel_page_at_lsn_batched(
-        &self,
-        pages: smallvec::SmallVec<[(RelTag, BlockNumber); 1]>,
-        version: Version<'_>,
-        ctx: &RequestContext,
-    ) -> smallvec::SmallVec<[Result<Bytes, PageReconstructError>; 1]> {
-        debug_assert_current_span_has_tenant_and_timeline_id();
-        let request_lsn = match version {
-            Version::Lsn(lsn) => lsn,
-            Version::Modified(_) => panic!("unsupported"),
-        };
-        enum KeyState {
-            NeedsVectoredGet,
-            Done(Result<Bytes, PageReconstructError>),
+        if tag.relnode == 0 {
+            return Err(PageReconstructError::Other(
+                RelationError::InvalidRelnode.into(),
+            ));
        }
-        let mut key_states = BTreeMap::new();
-        let mut vectored_gets: smallvec::SmallVec<[_; 1]> =
-            smallvec::SmallVec::with_capacity(pages.len());
-        for (response_order, (tag, blknum)) in pages.into_iter().enumerate() {
-            let key = rel_block_to_key(tag, blknum);
-            use std::collections::btree_map::Entry;
-            let key_state_slot = match key_states.entry((key, response_order)) {
-                Entry::Occupied(_entry) => unreachable!(
-                    "enumerate makes keys unique, even if batch contains same key twice"
-                ),
-                Entry::Vacant(entry) => entry,
-            };

-            if tag.relnode == 0 {
-                key_state_slot.insert(KeyState::Done(Err(PageReconstructError::Other(
-                    RelationError::InvalidRelnode.into(),
-                ))));
-                continue;
-            }
-
-            let nblocks = match self.get_rel_size(tag, version, ctx).await {
-                Ok(nblocks) => nblocks,
-                Err(err) => {
-                    key_state_slot.insert(KeyState::Done(Err(err)));
-                    continue;
-                }
-            };
-            if blknum >= nblocks {
-                debug!(
-                    "read beyond EOF at {} blk {} at {}, size is {}: returning all-zeros page",
-                    tag,
-                    blknum,
-                    version.get_lsn(),
-                    nblocks
-                );
-                key_state_slot.insert(KeyState::Done(Ok(ZERO_PAGE.clone())));
-                continue;
-            }
-
-            vectored_gets.push(key);
-            key_state_slot.insert(KeyState::NeedsVectoredGet);
+        let nblocks = self.get_rel_size(tag, version, ctx).await?;
+        if blknum >= nblocks {
+            debug!(
+                "read beyond EOF at {} blk {} at {}, size is {}: returning all-zeros page",
+                tag,
+                blknum,
+                version.get_lsn(),
+                nblocks
+            );
+            return Ok(ZERO_PAGE.clone());
        }
-        // turn vectored_gets into a keyspace
-        let keyspace = {
-            // add_key reuqires monotonicity
-            vectored_gets.sort_unstable();
-            let mut acc = KeySpaceAccum::new();
-            for key in vectored_gets
-                .into_iter()
-                // in fact it requires strong monotonicity
-                .dedup()
-            {
-                acc.add_key(key);
-            }
-            acc.to_keyspace()
-        };

-        match self.get_vectored(keyspace, request_lsn, ctx).await {
-            Ok(results) => {
-                for (key, res) in results {
-                    if let Err(err) = &res {
-                        warn!(%key, ?err, "a key inside get_vectored failed with a per-key error");
-                    }
-                    let mut interests = key_states.range_mut((key, 0)..(key.next(), 0)).peekable();
-                    let first_interest = interests.next().unwrap();
-                    let next_interest = interests.peek().is_some();
-                    if !next_interest {
-                        match first_interest.1 {
-                            KeyState::NeedsVectoredGet => {
-                                *first_interest.1 = KeyState::Done(res);
-                            }
-                            KeyState::Done(_) => unreachable!(),
-                        }
-                        continue;
-                    } else {
-                        for ((_, _), state) in [first_interest].into_iter().chain(interests) {
-                            match state {
-                                KeyState::NeedsVectoredGet => {
-                                    *state = KeyState::Done(match &res {
-                                        Ok(buf) => Ok(buf.clone()),
-                                        // this `match` is working around the fact that we cannot Clone the PageReconstructError
-                                        Err(err) => Err(match err {
-                                            PageReconstructError::Cancelled => {
-                                                PageReconstructError::Cancelled
-                                            }
-
-                                            x @ PageReconstructError::Other(_) |
-                                            x @ PageReconstructError::AncestorLsnTimeout(_) |
-                                            x @ PageReconstructError::WalRedo(_) |
-                                            x @ PageReconstructError::MissingKey(_) => {
-                                                PageReconstructError::Other(anyhow::anyhow!("there was more than one request for this key in the batch, error logged once: {x:?}"))
-                                            },
-                                        }),
-                                    });
-                                }
-                                KeyState::Done(_) => unreachable!(),
-                            }
-                        }
-                    }
-                }
-            }
-            Err(err) => {
-                warn!(?err, "get_vectored failed with a global error, mapping that error to per-key failure");
-                // this cannot really happen because get_vectored only errors globally on invalid LSN or too large batch size
-                for ((_, _), state) in key_states.iter_mut() {
-                    // this whole `match` is a lot like `From<GetVectoredError> for PageReconstructError`
-                    // but without taking ownership of the GetVectoredError
-                    match &err {
-                        GetVectoredError::Cancelled => {
-                            *state = KeyState::Done(Err(PageReconstructError::Cancelled));
-                        }
-                        // TODO: restructure get_vectored API to make this error per-key
-                        GetVectoredError::MissingKey(err) => {
-                            *state = KeyState::Done(Err(PageReconstructError::Other(anyhow::anyhow!("whole vectored get request failed because one or more of the requested keys were missing: {err:?}"))));
-                        }
-                        // TODO: restructure get_vectored API to make this error per-key
-                        GetVectoredError::GetReadyAncestorError(err) => {
-                            *state = KeyState::Done(Err(PageReconstructError::Other(anyhow::anyhow!("whole vectored get request failed because one or more key required ancestor that wasn't ready: {err:?}"))));
-                        }
-                        // TODO: restructure get_vectored API to make this error per-key
-                        GetVectoredError::Other(err) => {
-                            *state = KeyState::Done(Err(PageReconstructError::Other(
-                                anyhow::anyhow!("whole vectored get request failed: {err:?}"),
-                            )));
-                        }
-                        // TODO: we can prevent this error class by moving this check into the type system
-                        GetVectoredError::InvalidLsn(e) => {
-                            *state =
-                                KeyState::Done(Err(anyhow::anyhow!("invalid LSN: {e:?}").into()));
-                        }
-                        // NB: this should never happen in practice because we limit MAX_GET_VECTORED_KEYS
-                        // TODO: we can prevent this error class by moving this check into the type system
-                        GetVectoredError::Oversized(err) => {
-                            *state = KeyState::Done(Err(anyhow::anyhow!(
-                                "batching oversized: {err:?}"
-                            )
-                            .into()));
-                        }
-                    }
-                }
-            }
-        };
-
-        // get the results into the order in which they were requested
-        let mut return_order: smallvec::SmallVec<[_; Timeline::MAX_GET_VECTORED_KEYS as usize]> =
-            smallvec::SmallVec::with_capacity(key_states.len());
-        return_order.extend(key_states.keys().map(|(key, idx)| (*key, *idx)));
-        return_order.sort_unstable_by_key(|(_, idx)| *idx);
-        let mut res = smallvec::SmallVec::with_capacity(key_states.len());
-        res.extend(return_order.into_iter().map(|key_states_key| {
-            match key_states.remove(&key_states_key).unwrap() {
-                KeyState::Done(res) => res,
-                KeyState::NeedsVectoredGet => unreachable!(),
-            }
-        }));
-        res
+        let key = rel_block_to_key(tag, blknum);
+        version.get(self, key, ctx).await
    }

    // Get size of a database in blocks
@@ -1368,6 +1205,13 @@ impl<'a> DatadirModification<'a> {
        img: Bytes,
    ) -> anyhow::Result<()> {
        anyhow::ensure!(rel.relnode != 0, RelationError::InvalidRelnode);
+        let key = rel_block_to_key(rel, blknum);
+        if !key.is_valid_key_on_write_path() {
+            anyhow::bail!(
+                "the request contains data not supported by pageserver at {}",
+                key
+            );
+        }
        self.put(rel_block_to_key(rel, blknum), Value::Image(img));
        Ok(())
    }
@@ -1379,14 +1223,34 @@ impl<'a> DatadirModification<'a> {
        blknum: BlockNumber,
        img: Bytes,
    ) -> anyhow::Result<()> {
-        self.put(slru_block_to_key(kind, segno, blknum), Value::Image(img));
+        let key = slru_block_to_key(kind, segno, blknum);
+        if !key.is_valid_key_on_write_path() {
+            anyhow::bail!(
+                "the request contains data not supported by pageserver at {}",
+                key
+            );
+        }
+        self.put(key, Value::Image(img));
        Ok(())
    }

-    pub(crate) fn put_rel_page_image_zero(&mut self, rel: RelTag, blknum: BlockNumber) {
-        self.pending_zero_data_pages
-            .insert(rel_block_to_key(rel, blknum).to_compact());
+    pub(crate) fn put_rel_page_image_zero(
+        &mut self,
+        rel: RelTag,
+        blknum: BlockNumber,
+    ) -> anyhow::Result<()> {
+        anyhow::ensure!(rel.relnode != 0, RelationError::InvalidRelnode);
+        let key = rel_block_to_key(rel, blknum);
+        if !key.is_valid_key_on_write_path() {
+            anyhow::bail!(
+                "the request contains data not supported by pageserver: {} @ {}",
+                key,
+                self.lsn
+            );
+        }
+        self.pending_zero_data_pages.insert(key.to_compact());
        self.pending_bytes += ZERO_PAGE.len();
+        Ok(())
    }

    pub(crate) fn put_slru_page_image_zero(
@@ -1394,10 +1258,18 @@ impl<'a> DatadirModification<'a> {
        kind: SlruKind,
        segno: u32,
        blknum: BlockNumber,
-    ) {
-        self.pending_zero_data_pages
-            .insert(slru_block_to_key(kind, segno, blknum).to_compact());
+    ) -> anyhow::Result<()> {
+        let key = slru_block_to_key(kind, segno, blknum);
+        if !key.is_valid_key_on_write_path() {
+            anyhow::bail!(
+                "the request contains data not supported by pageserver: {} @ {}",
+                key,
+                self.lsn
+            );
+        }
+        self.pending_zero_data_pages.insert(key.to_compact());
        self.pending_bytes += ZERO_PAGE.len();
+        Ok(())
    }

    /// Call this at the end of each WAL record.
@@ -73,21 +73,6 @@ impl ValueBytes {

        Ok(raw[8] == 1)
    }
-
-    pub(crate) fn is_image(raw: &[u8]) -> Result<bool, InvalidInput> {
-        if raw.len() < 12 {
-            return Err(InvalidInput::TooShortValue);
-        }
-
-        let value_discriminator = &raw[0..4];
-
-        if value_discriminator == [0, 0, 0, 0] {
-            // Value::Image always initializes
-            return Ok(true);
-        }
-
-        Ok(false)
-    }
 }

 #[cfg(test)]
@@ -3491,7 +3491,7 @@ impl Tenant {
                .context("extract initdb tar")?;
        } else {
            // Init temporarily repo to get bootstrap data, this creates a directory in the `pgdata_path` path
-            run_initdb(self.conf, &pgdata_path, pg_version, &self.cancel).await?;
+            run_initdb_with_cache(self.conf, &pgdata_path, pg_version, &self.cancel).await?;

            // Upload the created data dir to S3
            if self.tenant_shard_id().is_shard_zero() {
@@ -3534,6 +3534,7 @@ impl Tenant {
        // Flush loop needs to be spawned in order to be able to flush.
        unfinished_timeline.maybe_spawn_flush_loop();

+        info!("starting to import from datadir");
        import_datadir::import_timeline_from_postgres_datadir(
            unfinished_timeline,
            &pgdata_path,
@@ -3549,6 +3550,7 @@ impl Tenant {
            anyhow::bail!("failpoint before-checkpoint-new-timeline");
        });

+        info!("calling freeze_and_flush");
        unfinished_timeline
            .freeze_and_flush()
            .await
@@ -3559,7 +3561,9 @@ impl Tenant {
            })?;

        // All done!
+        info!("calling finish_creation");
        let timeline = raw_timeline.finish_creation()?;
+        info!("call to  finish_creation done");

        Ok(timeline)
    }
@@ -3837,6 +3841,118 @@ impl Tenant {
    }
 }

+fn cached_initdb_dirname(initial_superuser_name: &str, pg_version: u32) -> String
+{
+    use std::hash::Hash;
+    use std::hash::Hasher;
+    use std::collections::hash_map::DefaultHasher;
+    let mut hasher = DefaultHasher::new();
+    initial_superuser_name.hash(&mut hasher);
+    let hash = hasher.finish();
+
+    format!("cached_initial_pgdata_{pg_version}_{:016}", hash)
+}
+
+fn copy_dir_all(src: impl AsRef<std::path::Path>, dst: impl AsRef<std::path::Path>) -> std::io::Result<()> {
+    for entry in fs::read_dir(src.as_ref())? {
+        let entry = entry?;
+        let subsrc = entry.path();
+        let subdst = dst.as_ref().join(&entry.file_name());
+
+        if entry.file_type()?.is_dir() {
+            std::fs::create_dir(&subdst)?;
+            copy_dir_all(&subsrc, &subdst)?;
+        } else {
+            std::fs::copy(&subsrc, &subdst)?;
+        }
+    }
+    Ok(())
+}
+
+fn restore_cached_initdb_dir(
+    cached_path: &Utf8Path,
+    target_path: &Utf8Path,
+) -> anyhow::Result<bool> {
+    if !cached_path.exists() {
+        info!("cached initdb dir \"{cached_path}\" does not exist yet");
+        return Ok(false);
+    }
+
+    std::fs::create_dir(target_path)?;
+    copy_dir_all(cached_path, target_path)?;
+    info!("restored initdb result from cache dir \"{cached_path}\"");
+    Ok(true)
+}
+
+fn save_cached_initdb_dir(
+    src_path: &Utf8Path,
+    cache_path: &Utf8Path,
+) -> anyhow::Result<()> {
+    match std::fs::create_dir(cache_path) {
+        Ok(()) => {
+            info!("saving initdb result to cache dir \"{cache_path}\"");
+        },
+        Err(err) if err.kind() == std::io::ErrorKind::AlreadyExists => {
+            info!("cache initdb dir \"{cache_path}\" already exists, not saving");
+            return Ok(())
+        },
+        Err(err) => { return Err(anyhow::Error::from(err))},
+    };
+    let cache_dir_guard = scopeguard::guard(cache_path, |cp| {
+        if let Err(err) = std::fs::remove_dir_all(&cp) {
+            error!("could not remove cached initdb directory {cp}: {err}");
+        }
+    });
+
+    let cache_parent_path = cache_path.parent().ok_or(anyhow::Error::msg("no cache parent path"))?;
+
+    let tmp_dirpath = camino_tempfile::tempdir_in(cache_parent_path)?;
+    copy_dir_all(src_path, &tmp_dirpath)?;
+    std::fs::rename(tmp_dirpath, &*cache_dir_guard)?;
+
+    // disarm the guard
+    scopeguard::ScopeGuard::into_inner(cache_dir_guard);
+
+    Ok(())
+}
+
+async fn run_initdb_with_cache(
+    conf: &'static PageServerConf,
+    initdb_target_dir: &Utf8Path,
+    pg_version: u32,
+    cancel: &CancellationToken,
+) -> Result<(), InitdbError> {
+
+    let cache_dir = conf.initdb_cache_dir.as_ref().map(|initdb_cache_dir| {
+        initdb_cache_dir.join(cached_initdb_dirname(&conf.superuser, pg_version))
+    });
+
+    if let Some(cache_dir) = &cache_dir {
+        match restore_cached_initdb_dir(&cache_dir, initdb_target_dir) {
+            Ok(true) => return Ok(()),
+            Ok(false) => {},
+            Err(err) => {
+                warn!("Error restoring from cached initdb directory \"{cache_dir}\": {err}");
+                if initdb_target_dir.exists() {
+                    if let Err(err) = std::fs::remove_dir_all(&initdb_target_dir) {
+                        error!("could not remove temporary initdb target directory {initdb_target_dir}: {err}");
+                    }
+                }
+            },
+        }
+    }
+
+    run_initdb(conf, initdb_target_dir, pg_version, cancel).await?;
+
+    if let Some(cache_dir) = &cache_dir {
+        if let Err(err) = save_cached_initdb_dir(initdb_target_dir, &cache_dir) {
+            warn!("error saving initdb result to cache directory \"{cache_dir}\": {err}");
+        }
+    }
+
+    Ok(())
+}
+
 /// Create the cluster temporarily in 'initdbpath' directory inside the repository
 /// to get bootstrap data for timeline initialization.
 async fn run_initdb(
@@ -7091,13 +7207,13 @@ mod tests {
            vec![
                // Image layer at GC horizon
                PersistentLayerKey {
-                    key_range: Key::MIN..Key::NON_L0_MAX,
+                    key_range: Key::MIN..Key::MAX,
                    lsn_range: Lsn(0x30)..Lsn(0x31),
                    is_delta: false
                },
-                // The delta layer covers the full range (with the layer key hack to avoid being recognized as L0)
+                // The delta layer below the horizon
                PersistentLayerKey {
-                    key_range: Key::MIN..Key::NON_L0_MAX,
+                    key_range: get_key(3)..get_key(4),
                    lsn_range: Lsn(0x30)..Lsn(0x48),
                    is_delta: true
                },
@@ -452,7 +452,8 @@ impl TryFrom<toml_edit::Item> for TenantConfOpt {
                    .map_err(|e| anyhow::anyhow!("{}: {}", e.path(), e.inner().message()));
            }
            toml_edit::Item::Table(table) => {
-                let deserializer = toml_edit::de::Deserializer::new(table.into());
+                let deserializer =
+                    toml_edit::de::Deserializer::from(toml_edit::DocumentMut::from(table));
                return serde_path_to_error::deserialize(deserializer)
                    .map_err(|e| anyhow::anyhow!("{}: {}", e.path(), e.inner().message()));
            }
@@ -8,18 +8,15 @@ mod layer_desc;
 mod layer_name;
 pub mod merge_iterator;

-use tokio::sync::{self};
-use tracing::{debug, Instrument};
-use utils::bin_ser::BeSer;
 pub mod split_writer;

 use crate::context::{AccessStatsBehavior, RequestContext};
-use crate::repository::{Value, ValueBytes};
+use crate::repository::Value;
 use crate::walrecord::NeonWalRecord;
 use bytes::Bytes;
-use pageserver_api::key::{Key, DBDIR_KEY};
+use pageserver_api::key::Key;
 use pageserver_api::keyspace::{KeySpace, KeySpaceRandomAccum};
-use std::cmp::Ordering;
+use std::cmp::{Ordering, Reverse};
 use std::collections::hash_map::Entry;
 use std::collections::{BinaryHeap, HashMap};
 use std::ops::Range;
@@ -82,57 +79,30 @@ pub(crate) enum ValueReconstructSituation {
 }

 /// Reconstruct data accumulated for a single key during a vectored get
-#[derive(Debug, Default)]
+#[derive(Debug, Default, Clone)]
 pub(crate) struct VectoredValueReconstructState {
-    pub(crate) records: Vec<(
-        Lsn,
-        tokio::sync::oneshot::Receiver<Result<Bytes, std::io::Error>>,
-    )>,
-    pub(crate) will_init_lsn: Option<Lsn>,
+    pub(crate) records: Vec<(Lsn, NeonWalRecord)>,
+    pub(crate) img: Option<(Lsn, Bytes)>,

-    pub(crate) situation: ValueReconstructSituation,
+    situation: ValueReconstructSituation,
 }

 impl VectoredValueReconstructState {
    fn get_cached_lsn(&self) -> Option<Lsn> {
-        self.will_init_lsn
+        self.img.as_ref().map(|img| img.0)
    }
 }

-pub(crate) async fn convert(
-    _key: Key,
-    from: VectoredValueReconstructState,
-) -> Result<ValueReconstructState, PageReconstructError> {
-    let mut to = ValueReconstructState::default();
+impl From<VectoredValueReconstructState> for ValueReconstructState {
+    fn from(mut state: VectoredValueReconstructState) -> Self {
+        // walredo expects the records to be descending in terms of Lsn
+        state.records.sort_by_key(|(lsn, _)| Reverse(*lsn));

-    for (lsn, fut) in from.records {
-        match fut.await {
-            Ok(res) => match res {
-                Ok(bytes) => {
-                    let value = Value::des(&bytes)
-                        .map_err(|err| PageReconstructError::Other(err.into()))?;
-
-                    match value {
-                        Value::WalRecord(rec) => {
-                            to.records.push((lsn, rec));
-                        }
-                        Value::Image(img) => {
-                            assert!(to.img.is_none());
-                            to.img = Some((lsn, img));
-                        }
-                    }
-                }
-                Err(err) => {
-                    return Err(PageReconstructError::Other(err.into()));
-                }
-            },
-            Err(err) => {
-                return Err(PageReconstructError::Other(err.into()));
-            }
+        ValueReconstructState {
+            records: state.records,
+            img: state.img,
        }
    }
-
-    Ok(to)
 }

 /// Bag of data accumulated during a vectored get..
@@ -149,47 +119,6 @@ pub(crate) struct ValuesReconstructState {
    // Statistics that are still accessible as a caller of `get_vectored_impl`.
    layers_visited: u32,
    delta_layers_visited: u32,
-
-    io_concurrency: IoConcurrency,
-}
-
-enum IoConcurrency {
-    Serial {
-        prev_io: Option<(usize, tokio::task::JoinHandle<()>)>,
-    },
-    Parallel,
-}
-
-impl IoConcurrency {
-    pub(crate) fn spawn_io<F>(&mut self, fut: F)
-    where
-        F: std::future::Future<Output = ()> + Send + 'static,
-    {
-        static IO_COUNTER: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
-        let io_id = IO_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-        let span = tracing::debug_span!("spawned_io", io_id,);
-        match self {
-            IoConcurrency::Serial { prev_io } => {
-                let prev = prev_io.take();
-                *prev_io = Some((
-                    io_id,
-                    tokio::spawn(
-                        async move {
-                            if let Some((prev_id, prev_task)) = prev {
-                                debug!(prev_io = prev_id, "Waiting for previous IO to complete");
-                                prev_task.await.unwrap();
-                            }
-                            fut.await;
-                        }
-                        .instrument(span),
-                    ),
-                ));
-            }
-            IoConcurrency::Parallel => {
-                tokio::spawn(fut);
-            }
-        }
-    }
 }

 impl ValuesReconstructState {
@@ -200,30 +129,9 @@ impl ValuesReconstructState {
            keys_with_image_coverage: None,
            layers_visited: 0,
            delta_layers_visited: 0,
-            io_concurrency: {
-                static IO_CONCURRENCY: once_cell::sync::Lazy<String> =
-                    once_cell::sync::Lazy::new(|| {
-                        std::env::var("NEON_PAGESERVER_VALUE_RECONSTRUCT_IO_CONCURRENCY").unwrap()
-                    });
-                match IO_CONCURRENCY.as_str() {
-                    "parallel" => IoConcurrency::Parallel,
-                    "serial" => IoConcurrency::Serial { prev_io: None },
-                    x => panic!(
-                        "Invalid value for NEON_PAGESERVER_VALUE_RECONSTRUCT_IO_CONCURRENCY: {}",
-                        x
-                    ),
-                }
-            },
        }
    }

-    pub(crate) fn spawn_io<F>(&mut self, fut: F)
-    where
-        F: std::future::Future<Output = ()> + Send + 'static,
-    {
-        self.io_concurrency.spawn_io(fut);
-    }
-
    /// Associate a key with the error which it encountered and mark it as done
    pub(crate) fn on_key_error(&mut self, key: Key, err: PageReconstructError) {
        let previous = self.keys.insert(key, Err(err));
@@ -292,8 +200,7 @@ impl ValuesReconstructState {
        &mut self,
        key: &Key,
        lsn: Lsn,
-        completes: bool,
-        value: sync::oneshot::Receiver<Result<Bytes, std::io::Error>>,
+        value: Value,
    ) -> ValueReconstructSituation {
        let state = self
            .keys
@@ -301,16 +208,31 @@ impl ValuesReconstructState {
            .or_insert(Ok(VectoredValueReconstructState::default()));

        if let Ok(state) = state {
-            match state.situation {
+            let key_done = match state.situation {
                ValueReconstructSituation::Complete => unreachable!(),
-                ValueReconstructSituation::Continue => {
-                    state.records.push((lsn, value));
-                }
-            }
+                ValueReconstructSituation::Continue => match value {
+                    Value::Image(img) => {
+                        state.img = Some((lsn, img));
+                        true
+                    }
+                    Value::WalRecord(rec) => {
+                        debug_assert!(
+                            Some(lsn) > state.get_cached_lsn(),
+                            "Attempt to collect a record below cached LSN for walredo: {} < {}",
+                            lsn,
+                            state
+                                .get_cached_lsn()
+                                .expect("Assertion can only fire if a cached lsn is present")
+                        );

-            if completes && state.situation == ValueReconstructSituation::Continue {
-                assert_eq!(state.will_init_lsn, None);
-                state.will_init_lsn = Some(lsn);
+                        let will_init = rec.will_init();
+                        state.records.push((lsn, rec));
+                        will_init
+                    }
+                },
+            };
+
+            if key_done && state.situation == ValueReconstructSituation::Continue {
                state.situation = ValueReconstructSituation::Complete;
                self.keys_done.add_key(*key);
            }
@@ -357,7 +279,7 @@ pub(crate) enum LayerId {
 /// Layer wrapper for the read path. Note that it is valid
 /// to use these layers even after external operations have
 /// been performed on them (compaction, freeze, etc.).
-#[derive(Debug, Clone)]
+#[derive(Debug)]
 pub(crate) enum ReadableLayer {
    PersistentLayer(Layer),
    InMemoryLayer(Arc<InMemoryLayer>),
@@ -370,8 +292,6 @@ struct ReadDesc {
    layer_id: LayerId,
    /// Lsn range for the read, used for selecting the next read
    lsn_range: Range<Lsn>,
-    /// This read's index in [`LayerKeyspace::target_keyspace`];
-    read_id: LayerKeyspaceReadId,
 }

 /// Data structure which maintains a fringe of layers for the
@@ -390,13 +310,9 @@ pub(crate) struct LayerFringe {
 #[derive(Debug)]
 struct LayerKeyspace {
    layer: ReadableLayer,
-    next_read_id: LayerKeyspaceReadId,
-    reads: HashMap<LayerKeyspaceReadId, (Range<Lsn>, KeySpace)>,
+    target_keyspace: KeySpaceRandomAccum,
 }

-#[derive(PartialEq, Eq, Hash, Debug, Clone, Copy)]
-struct LayerKeyspaceReadId(usize);
-
 impl LayerFringe {
    pub(crate) fn new() -> Self {
        LayerFringe {
@@ -411,24 +327,22 @@ impl LayerFringe {
            None => return None,
        };

-        let mut entry = match self.layers.entry(read_desc.layer_id) {
-            Entry::Occupied(o) => o,
-            Entry::Vacant(_) => unreachable!("fringe internals are always consistent"),
-        };
+        let removed = self.layers.remove_entry(&read_desc.layer_id);

-        let (lsn_range, keyspace) = entry
-            .get_mut()
-            .reads
-            .remove(&read_desc.read_id)
-            .expect("fringe internals are always consistent");
-
-        let layer = entry.get().layer.clone();
-
-        if entry.get().reads.is_empty() {
-            entry.remove();
+        match removed {
+            Some((
+                _,
+                LayerKeyspace {
+                    layer,
+                    mut target_keyspace,
+                },
+            )) => Some((
+                layer,
+                target_keyspace.consume_keyspace(),
+                read_desc.lsn_range,
+            )),
+            None => unreachable!("fringe internals are always consistent"),
        }
-
-        Some((layer, keyspace, lsn_range))
    }

    pub(crate) fn update(
@@ -441,31 +355,18 @@ impl LayerFringe {
        let entry = self.layers.entry(layer_id.clone());
        match entry {
            Entry::Occupied(mut entry) => {
-                let read_id = {
-                    let r = &mut entry.get_mut().next_read_id;
-                    let read_id = *r;
-                    *r = LayerKeyspaceReadId(r.0 + 1);
-                    read_id
-                };
-                self.planned_reads_by_lsn.push(ReadDesc {
-                    lsn_range: lsn_range.clone(),
-                    layer_id: layer_id.clone(),
-                    read_id,
-                });
-                let replaced = entry.get_mut().reads.insert(read_id, (lsn_range, keyspace));
-                assert!(replaced.is_none());
+                entry.get_mut().target_keyspace.add_keyspace(keyspace);
            }
            Entry::Vacant(entry) => {
-                let read_id = LayerKeyspaceReadId(0);
                self.planned_reads_by_lsn.push(ReadDesc {
-                    lsn_range: lsn_range.clone(),
+                    lsn_range,
                    layer_id: layer_id.clone(),
-                    read_id,
                });
+                let mut accum = KeySpaceRandomAccum::new();
+                accum.add_keyspace(keyspace);
                entry.insert(LayerKeyspace {
                    layer,
-                    next_read_id: LayerKeyspaceReadId(1),
-                    reads: [(read_id, (lsn_range, keyspace))].into(),
+                    target_keyspace: accum,
                });
            }
        }
@@ -42,12 +42,13 @@ use crate::tenant::vectored_blob_io::{
    BlobFlag, StreamingVectoredReadPlanner, VectoredBlobReader, VectoredRead,
    VectoredReadCoalesceMode, VectoredReadPlanner,
 };
+use crate::tenant::PageReconstructError;
 use crate::virtual_file::owned_buffers_io::io_buf_ext::{FullSlice, IoBufExt};
 use crate::virtual_file::{self, VirtualFile};
 use crate::{walrecord, TEMP_FILE_SUFFIX};
 use crate::{DELTA_FILE_MAGIC, STORAGE_FORMAT_VERSION};
-use anyhow::{bail, ensure, Context, Result};
-use bytes::{Bytes, BytesMut};
+use anyhow::{anyhow, bail, ensure, Context, Result};
+use bytes::BytesMut;
 use camino::{Utf8Path, Utf8PathBuf};
 use futures::StreamExt;
 use itertools::Itertools;
@@ -57,14 +58,14 @@ use pageserver_api::models::ImageCompressionAlgorithm;
 use pageserver_api::shard::TenantShardId;
 use rand::{distributions::Alphanumeric, Rng};
 use serde::{Deserialize, Serialize};
-use std::collections::{HashMap, VecDeque};
+use std::collections::VecDeque;
 use std::fs::File;
 use std::io::SeekFrom;
 use std::ops::Range;
 use std::os::unix::fs::FileExt;
 use std::str::FromStr;
 use std::sync::Arc;
-use tokio::sync::{self, OnceCell};
+use tokio::sync::OnceCell;
 use tokio_epoll_uring::IoBuf;
 use tracing::*;

@@ -223,7 +224,7 @@ pub struct DeltaLayerInner {
    index_start_blk: u32,
    index_root_blk: u32,

-    file: Arc<VirtualFile>,
+    file: VirtualFile,
    file_id: FileId,

    layer_key_range: Range<Key>,
@@ -787,11 +788,9 @@ impl DeltaLayerInner {
        max_vectored_read_bytes: Option<MaxVectoredReadBytes>,
        ctx: &RequestContext,
    ) -> anyhow::Result<Self> {
-        let file = Arc::new(
-            VirtualFile::open(path, ctx)
-                .await
-                .context("open layer file")?,
-        );
+        let file = VirtualFile::open(path, ctx)
+            .await
+            .context("open layer file")?;

        let file_id = page_cache::next_file_id();

@@ -842,7 +841,6 @@ impl DeltaLayerInner {
    // can be further optimised to visit the index only once.
    pub(super) async fn get_values_reconstruct_data(
        &self,
-        self_desc: PersistentLayerDesc,
        keyspace: KeySpace,
        lsn_range: Range<Lsn>,
        reconstruct_state: &mut ValuesReconstructState,
@@ -865,7 +863,6 @@ impl DeltaLayerInner {
        let data_end_offset = self.index_start_offset();

        let reads = Self::plan_reads(
-            self_desc,
            &keyspace,
            lsn_range.clone(),
            data_end_offset,
@@ -886,7 +883,6 @@ impl DeltaLayerInner {
    }

    async fn plan_reads<Reader>(
-        self_desc: PersistentLayerDesc,
        keyspace: &KeySpace,
        lsn_range: Range<Lsn>,
        data_end_offset: u64,
@@ -915,8 +911,6 @@ impl DeltaLayerInner {
                let lsn = DeltaKey::extract_lsn_from_buf(&raw_key);
                let blob_ref = BlobRef(value);

-                debug!(file = %self_desc.layer_name(), %key, %lsn, will_init = blob_ref.will_init(), "delta layer found key");
-
                // Lsns are not monotonically increasing across keys, so we don't assert on them.
                assert!(key >= range.start);

@@ -939,7 +933,7 @@ impl DeltaLayerInner {
                    range_end_handled = true;
                    break;
                } else {
-                    planner.handle(key, lsn, blob_ref.pos(), flag, blob_ref.will_init());
+                    planner.handle(key, lsn, blob_ref.pos(), flag);
                }
            }

@@ -986,59 +980,77 @@ impl DeltaLayerInner {
        reconstruct_state: &mut ValuesReconstructState,
        ctx: &RequestContext,
    ) {
+        let vectored_blob_reader = VectoredBlobReader::new(&self.file);
+        let mut ignore_key_with_err = None;
+
        let max_vectored_read_bytes = self
            .max_vectored_read_bytes
            .expect("Layer is loaded with max vectored bytes config")
            .0
            .into();
        let buf_size = Self::get_min_read_buffer_size(&reads, max_vectored_read_bytes);
+        let mut buf = Some(BytesMut::with_capacity(buf_size));

        // Note that reads are processed in reverse order (from highest key+lsn).
        // This is the order that `ReconstructState` requires such that it can
        // track when a key is done.
        for read in reads.into_iter().rev() {
-            let mut senders: HashMap<
-                (Key, Lsn),
-                sync::oneshot::Sender<Result<Bytes, std::io::Error>>,
-            > = Default::default();
-            for (_, blob_meta) in read.blobs_at.as_slice().iter().rev() {
-                let (tx, rx) = sync::oneshot::channel();
-                senders.insert((blob_meta.key, blob_meta.lsn), tx);
-                reconstruct_state.update_key(
-                    &blob_meta.key,
-                    blob_meta.lsn,
-                    blob_meta.will_init,
-                    rx,
-                );
+            let res = vectored_blob_reader
+                .read_blobs(&read, buf.take().expect("Should have a buffer"), ctx)
+                .await;
+
+            let blobs_buf = match res {
+                Ok(blobs_buf) => blobs_buf,
+                Err(err) => {
+                    let kind = err.kind();
+                    for (_, blob_meta) in read.blobs_at.as_slice() {
+                        reconstruct_state.on_key_error(
+                            blob_meta.key,
+                            PageReconstructError::Other(anyhow!(
+                                "Failed to read blobs from virtual file {}: {}",
+                                self.file.path,
+                                kind
+                            )),
+                        );
+                    }
+
+                    // We have "lost" the buffer since the lower level IO api
+                    // doesn't return the buffer on error. Allocate a new one.
+                    buf = Some(BytesMut::with_capacity(buf_size));
+
+                    continue;
+                }
+            };
+
+            for meta in blobs_buf.blobs.iter().rev() {
+                if Some(meta.meta.key) == ignore_key_with_err {
+                    continue;
+                }
+
+                let value = Value::des(&blobs_buf.buf[meta.start..meta.end]);
+                let value = match value {
+                    Ok(v) => v,
+                    Err(e) => {
+                        reconstruct_state.on_key_error(
+                            meta.meta.key,
+                            PageReconstructError::Other(anyhow!(e).context(format!(
+                                "Failed to deserialize blob from virtual file {}",
+                                self.file.path,
+                            ))),
+                        );
+
+                        ignore_key_with_err = Some(meta.meta.key);
+                        continue;
+                    }
+                };
+
+                // Invariant: once a key reaches [`ValueReconstructSituation::Complete`]
+                // state, no further updates shall be made to it. The call below will
+                // panic if the invariant is violated.
+                reconstruct_state.update_key(&meta.meta.key, meta.meta.lsn, value);
            }

-            let read_from = self.file.clone();
-            let read_ctx = ctx.attached_child();
-            reconstruct_state.spawn_io(async move {
-                let vectored_blob_reader = VectoredBlobReader::new(&read_from);
-                let buf = BytesMut::with_capacity(buf_size);
-
-                let res = vectored_blob_reader.read_blobs(&read, buf, &read_ctx).await;
-                match res {
-                    Ok(blobs_buf) => {
-                        for meta in blobs_buf.blobs.iter().rev() {
-                            let buf = &blobs_buf.buf[meta.start..meta.end];
-                            let sender = senders
-                                .remove(&(meta.meta.key, meta.meta.lsn))
-                                .expect("sender must exist");
-                            let _ = sender.send(Ok(Bytes::copy_from_slice(buf)));
-                        }
-
-                        assert!(senders.is_empty());
-                    }
-                    Err(err) => {
-                        for (_, sender) in senders {
-                            let _ = sender
-                                .send(Err(std::io::Error::new(err.kind(), "vec read failed")));
-                        }
-                    }
-                }
-            });
+            buf = Some(blobs_buf.buf);
        }
    }

@@ -1178,14 +1190,7 @@ impl DeltaLayerInner {
            let actionable = if let Some((key, lsn, start_offset)) = prev.take() {
                let end_offset = offset;

-                Some((
-                    BlobMeta {
-                        key,
-                        lsn,
-                        will_init: false,
-                    },
-                    start_offset..end_offset,
-                ))
+                Some((BlobMeta { key, lsn }, start_offset..end_offset))
            } else {
                None
            };
@@ -21,7 +21,7 @@
 //!
 //! Every image layer file consists of three parts: "summary",
 //! "index", and "values".  The summary is a fixed size header at the
-//! beginningof the file, and it contains basic information about the
+//! beginning of the file, and it contains basic information about the
 //! layer, and offsets to the other parts. The "index" is a B-tree,
 //! mapping from Key to an offset in the "values" part.  The
 //! actual page images are stored in the "values" part.
@@ -38,11 +38,11 @@ use crate::tenant::timeline::GetVectoredError;
 use crate::tenant::vectored_blob_io::{
    BlobFlag, StreamingVectoredReadPlanner, VectoredBlobReader, VectoredRead, VectoredReadPlanner,
 };
-use crate::tenant::Timeline;
+use crate::tenant::{PageReconstructError, Timeline};
 use crate::virtual_file::owned_buffers_io::io_buf_ext::IoBufExt;
 use crate::virtual_file::{self, VirtualFile};
 use crate::{IMAGE_FILE_MAGIC, STORAGE_FORMAT_VERSION, TEMP_FILE_SUFFIX};
-use anyhow::{bail, ensure, Context, Result};
+use anyhow::{anyhow, bail, ensure, Context, Result};
 use bytes::{Bytes, BytesMut};
 use camino::{Utf8Path, Utf8PathBuf};
 use hex;
@@ -52,14 +52,13 @@ use pageserver_api::keyspace::KeySpace;
 use pageserver_api::shard::{ShardIdentity, TenantShardId};
 use rand::{distributions::Alphanumeric, Rng};
 use serde::{Deserialize, Serialize};
-use std::collections::{HashMap, VecDeque};
+use std::collections::VecDeque;
 use std::fs::File;
 use std::io::SeekFrom;
 use std::ops::Range;
 use std::os::unix::prelude::FileExt;
 use std::str::FromStr;
 use std::sync::Arc;
-use tokio::sync::oneshot;
 use tokio::sync::OnceCell;
 use tokio_stream::StreamExt;
 use tracing::*;
@@ -164,7 +163,7 @@ pub struct ImageLayerInner {
    key_range: Range<Key>,
    lsn: Lsn,

-    file: Arc<VirtualFile>,
+    file: VirtualFile,
    file_id: FileId,

    max_vectored_read_bytes: Option<MaxVectoredReadBytes>,
@@ -391,11 +390,9 @@ impl ImageLayerInner {
        max_vectored_read_bytes: Option<MaxVectoredReadBytes>,
        ctx: &RequestContext,
    ) -> anyhow::Result<Self> {
-        let file = Arc::new(
-            VirtualFile::open(path, ctx)
-                .await
-                .context("open layer file")?,
-        );
+        let file = VirtualFile::open(path, ctx)
+            .await
+            .context("open layer file")?;
        let file_id = page_cache::next_file_id();
        let block_reader = FileBlockReader::new(&file, file_id);
        let summary_blk = block_reader
@@ -441,13 +438,12 @@ impl ImageLayerInner {
    // the reconstruct state with whatever is found.
    pub(super) async fn get_values_reconstruct_data(
        &self,
-        self_desc: PersistentLayerDesc,
        keyspace: KeySpace,
        reconstruct_state: &mut ValuesReconstructState,
        ctx: &RequestContext,
    ) -> Result<(), GetVectoredError> {
        let reads = self
-            .plan_reads(self_desc, keyspace, None, ctx)
+            .plan_reads(keyspace, None, ctx)
            .await
            .map_err(GetVectoredError::Other)?;

@@ -466,7 +462,6 @@ impl ImageLayerInner {
    /// this shard.
    async fn plan_reads(
        &self,
-        self_desc: PersistentLayerDesc,
        keyspace: KeySpace,
        shard_identity: Option<&ShardIdentity>,
        ctx: &RequestContext,
@@ -510,14 +505,12 @@ impl ImageLayerInner {
                    BlobFlag::None
                };

-                debug!(file = %self_desc.layer_name(), %key, %self.lsn, will_init=true, "image layer found key");
-
                if key >= range.end {
                    planner.handle_range_end(offset);
                    range_end_handled = true;
                    break;
                } else {
-                    planner.handle(key, self.lsn, offset, flag, true);
+                    planner.handle(key, self.lsn, offset, flag);
                }
            }

@@ -541,7 +534,6 @@ impl ImageLayerInner {
        // Fragment the range into the regions owned by this ShardIdentity
        let plan = self
            .plan_reads(
-                todo!(),
                KeySpace {
                    // If asked for the total key space, plan_reads will give us all the keys in the layer
                    ranges: vec![Key::MIN..Key::MAX],
@@ -587,16 +579,8 @@ impl ImageLayerInner {
            .0
            .into();

+        let vectored_blob_reader = VectoredBlobReader::new(&self.file);
        for read in reads.into_iter() {
-            let mut senders: HashMap<(Key, Lsn), oneshot::Sender<Result<Bytes, std::io::Error>>> =
-                Default::default();
-            for (_, blob_meta) in read.blobs_at.as_slice() {
-                let (tx, rx) = oneshot::channel();
-                senders.insert((blob_meta.key, blob_meta.lsn), tx);
-
-                reconstruct_state.update_key(&blob_meta.key, blob_meta.lsn, true, rx);
-            }
-
            let buf_size = read.size();

            if buf_size > max_vectored_read_bytes {
@@ -615,36 +599,36 @@ impl ImageLayerInner {
                );
            }

-            let read_from = self.file.clone();
-            let read_ctx = ctx.attached_child();
-            reconstruct_state.spawn_io(async move {
-                let buf = BytesMut::with_capacity(buf_size);
-                let vectored_blob_reader = VectoredBlobReader::new(&read_from);
-                let res = vectored_blob_reader.read_blobs(&read, buf, &read_ctx).await;
+            let buf = BytesMut::with_capacity(buf_size);
+            let res = vectored_blob_reader.read_blobs(&read, buf, ctx).await;

-                match res {
-                    Ok(blobs_buf) => {
-                        for meta in blobs_buf.blobs.iter() {
-                            let buf = &blobs_buf.buf[meta.start..meta.end];
-                            let sender = senders
-                                .remove(&(meta.meta.key, meta.meta.lsn))
-                                .expect("sender must exist");
-                            // TODO: this is silly - sort it out
-                            let bytes = Value::ser(&Value::Image(Bytes::copy_from_slice(buf)))
-                                .expect("stupid but correct");
-                            let _ = sender.send(Ok(bytes.into()));
-                        }
+            match res {
+                Ok(blobs_buf) => {
+                    let frozen_buf = blobs_buf.buf.freeze();

-                        assert!(senders.is_empty());
-                    }
-                    Err(err) => {
-                        for (_, sender) in senders {
-                            let _ = sender
-                                .send(Err(std::io::Error::new(err.kind(), "vec read failed")));
-                        }
+                    for meta in blobs_buf.blobs.iter() {
+                        let img_buf = frozen_buf.slice(meta.start..meta.end);
+                        reconstruct_state.update_key(
+                            &meta.meta.key,
+                            self.lsn,
+                            Value::Image(img_buf),
+                        );
                    }
                }
-            });
+                Err(err) => {
+                    let kind = err.kind();
+                    for (_, blob_meta) in read.blobs_at.as_slice() {
+                        reconstruct_state.on_key_error(
+                            blob_meta.key,
+                            PageReconstructError::from(anyhow!(
+                                "Failed to read blobs from virtual file {}: {}",
+                                self.file.path,
+                                kind
+                            )),
+                        );
+                    }
+                }
+            };
        }
    }

@@ -10,9 +10,10 @@ use crate::context::{PageContentKind, RequestContext, RequestContextBuilder};
 use crate::repository::{Key, Value};
 use crate::tenant::ephemeral_file::EphemeralFile;
 use crate::tenant::timeline::GetVectoredError;
+use crate::tenant::PageReconstructError;
 use crate::virtual_file::owned_buffers_io::io_buf_ext::IoBufExt;
 use crate::{l0_flush, page_cache};
-use anyhow::{Context, Result};
+use anyhow::{anyhow, Context, Result};
 use bytes::Bytes;
 use camino::Utf8PathBuf;
 use pageserver_api::key::CompactKey;
@@ -34,7 +35,9 @@ use std::sync::atomic::Ordering as AtomicOrdering;
 use std::sync::atomic::{AtomicU64, AtomicUsize};
 use tokio::sync::RwLock;

-use super::{DeltaLayerWriter, PersistentLayerDesc, ValuesReconstructState};
+use super::{
+    DeltaLayerWriter, PersistentLayerDesc, ValueReconstructSituation, ValuesReconstructState,
+};

 pub(crate) mod vectored_dio_read;

@@ -84,7 +87,7 @@ pub struct InMemoryLayerInner {
    /// The values are stored in a serialized format in this file.
    /// Each serialized Value is preceded by a 'u32' length field.
    /// PerSeg::page_versions map stores offsets into this file.
-    file: Arc<tokio::sync::RwLock<EphemeralFile>>,
+    file: EphemeralFile,

    resource_units: GlobalResourceUnits,
 }
@@ -378,11 +381,7 @@ impl InMemoryLayer {
    }

    pub(crate) fn try_len(&self) -> Option<u64> {
-        self.inner
-            .try_read()
-            .map(|i| i.file.try_read().map(|i| i.len()).ok())
-            .ok()
-            .flatten()
+        self.inner.try_read().map(|i| i.file.len()).ok()
    }

    pub(crate) fn assert_writable(&self) {
@@ -433,10 +432,6 @@ impl InMemoryLayer {
            read: vectored_dio_read::LogicalRead<Vec<u8>>,
        }
        let mut reads: HashMap<Key, Vec<ValueRead>> = HashMap::new();
-        let mut senders: HashMap<
-            (Key, Lsn),
-            tokio::sync::oneshot::Sender<Result<Bytes, std::io::Error>>,
-        > = Default::default();

        for range in keyspace.ranges.iter() {
            for (key, vec_map) in inner
@@ -464,13 +459,6 @@ impl InMemoryLayer {
                            Vec::with_capacity(len as usize),
                        ),
                    });
-
-                    let (tx, rx) = tokio::sync::oneshot::channel();
-                    senders.insert((key, *entry_lsn), tx);
-                    reconstruct_state.update_key(&key, *entry_lsn, will_init, rx);
-
-                    debug!(%key, %entry_lsn, will_init, "inmemory layer found key");
-
                    if will_init {
                        break;
                    }
@@ -478,39 +466,46 @@ impl InMemoryLayer {
            }
        }

-        let read_from = inner.file.clone();
-        let read_ctx = ctx.attached_child();
-        reconstruct_state.spawn_io(async move {
-            let locked = read_from.read().await;
-            let f = vectored_dio_read::execute(
-                &*locked,
-                reads
-                    .iter()
-                    .flat_map(|(_, value_reads)| value_reads.iter().map(|v| &v.read)),
-                &read_ctx,
-            );
-            send_future::SendFuture::send(f) // https://github.com/rust-lang/rust/issues/96865
-                .await;
+        // Execute the reads.

-            for (key, value_reads) in reads {
-                for ValueRead { entry_lsn, read } in value_reads {
-                    let sender = senders
-                        .remove(&(key, entry_lsn))
-                        .expect("sender must exist");
-                    match read.into_result().expect("we run execute() above") {
-                        Err(e) => {
-                            let _ = sender
-                                .send(Err(std::io::Error::new(e.kind(), "dio vec read failed")));
+        let f = vectored_dio_read::execute(
+            &inner.file,
+            reads
+                .iter()
+                .flat_map(|(_, value_reads)| value_reads.iter().map(|v| &v.read)),
+            &ctx,
+        );
+        send_future::SendFuture::send(f) // https://github.com/rust-lang/rust/issues/96865
+            .await;
+
+        // Process results into the reconstruct state
+        'next_key: for (key, value_reads) in reads {
+            for ValueRead { entry_lsn, read } in value_reads {
+                match read.into_result().expect("we run execute() above") {
+                    Err(e) => {
+                        reconstruct_state.on_key_error(key, PageReconstructError::from(anyhow!(e)));
+                        continue 'next_key;
+                    }
+                    Ok(value_buf) => {
+                        let value = Value::des(&value_buf);
+                        if let Err(e) = value {
+                            reconstruct_state
+                                .on_key_error(key, PageReconstructError::from(anyhow!(e)));
+                            continue 'next_key;
                        }
-                        Ok(value_buf) => {
-                            let _ = sender.send(Ok(value_buf.into()));
+
+                        let key_situation =
+                            reconstruct_state.update_key(&key, entry_lsn, value.unwrap());
+                        if key_situation == ValueReconstructSituation::Complete {
+                            // TODO: metric to see if we fetched more values than necessary
+                            continue 'next_key;
                        }
+
+                        // process the next value in the next iteration of the loop
                    }
                }
            }
-
-            assert!(senders.is_empty());
-        });
+        }

        reconstruct_state.on_lsn_advanced(&keyspace, self.start_lsn);

@@ -605,8 +600,7 @@ impl InMemoryLayer {
    /// Get layer size.
    pub async fn size(&self) -> Result<u64> {
        let inner = self.inner.read().await;
-        let locked = inner.file.try_read().expect("no contention");
-        Ok(locked.len())
+        Ok(inner.file.len())
    }

    /// Create a new, empty, in-memory layer
@@ -620,10 +614,9 @@ impl InMemoryLayer {
    ) -> Result<InMemoryLayer> {
        trace!("initializing new empty InMemoryLayer for writing on timeline {timeline_id} at {start_lsn}");

-        let file = Arc::new(tokio::sync::RwLock::new(
-            EphemeralFile::create(conf, tenant_shard_id, timeline_id, gate_guard, ctx).await?,
-        ));
-        let key = InMemoryLayerFileId(file.read().await.page_cache_file_id());
+        let file =
+            EphemeralFile::create(conf, tenant_shard_id, timeline_id, gate_guard, ctx).await?;
+        let key = InMemoryLayerFileId(file.page_cache_file_id());

        Ok(InMemoryLayer {
            file_id: key,
@@ -655,7 +648,7 @@ impl InMemoryLayer {
        let mut inner = self.inner.write().await;
        self.assert_writable();

-        let base_offset = inner.file.read().await.len();
+        let base_offset = inner.file.len();

        let SerializedBatch {
            raw,
@@ -679,13 +672,8 @@ impl InMemoryLayer {
        }

        // Write the batch to the file
-        // FIXME: can't borrow arc
-        let new_size = {
-            let mut locked = inner.file.write().await;
-            locked.write_raw(&raw, ctx).await?;
-            locked.len()
-        };
-
+        inner.file.write_raw(&raw, ctx).await?;
+        let new_size = inner.file.len();
        let expected_new_len = base_offset
            .checked_add(raw.len().into_u64())
            // write_raw would error if we were to overflow u64.
@@ -725,7 +713,7 @@ impl InMemoryLayer {

    pub(crate) async fn tick(&self) -> Option<u64> {
        let mut inner = self.inner.write().await;
-        let size = inner.file.read().await.len();
+        let size = inner.file.len();
        inner.resource_units.publish_size(size)
    }

@@ -821,7 +809,7 @@ impl InMemoryLayer {

        match l0_flush_global_state {
            l0_flush::Inner::Direct { .. } => {
-                let file_contents: Vec<u8> = inner.file.read().await.load_to_vec(ctx).await?;
+                let file_contents: Vec<u8> = inner.file.load_to_vec(ctx).await?;

                let file_contents = Bytes::from(file_contents);

@@ -1755,17 +1755,11 @@ impl DownloadedLayer {
            .map_err(GetVectoredError::Other)?
        {
            Delta(d) => {
-                d.get_values_reconstruct_data(
-                    owner.desc.clone(),
-                    keyspace,
-                    lsn_range,
-                    reconstruct_data,
-                    ctx,
-                )
-                .await
+                d.get_values_reconstruct_data(keyspace, lsn_range, reconstruct_data, ctx)
+                    .await
            }
            Image(i) => {
-                i.get_values_reconstruct_data(owner.desc.clone(), keyspace, reconstruct_data, ctx)
+                i.get_values_reconstruct_data(keyspace, reconstruct_data, ctx)
                    .await
            }
        }
@@ -107,8 +107,6 @@ async fn smoke_test() {
            .expect("tenant harness writes the control file")
    };

-    let img_before = (img_before.0, img_before.1.await.unwrap().unwrap());
-    let img_after = (img_after.0, img_after.1.await.unwrap().unwrap());
    assert_eq!(img_before, img_after);

    // evict_and_wait can timeout, but it doesn't cancel the evicting itself
@@ -188,7 +188,7 @@ impl SplitImageLayerWriter {
            .await
    }

-    /// When split writer fails, the caller should call this function and handle partially generated layers.
+    /// This function will be deprecated with #8841.
    pub(crate) fn take(self) -> anyhow::Result<(Vec<SplitWriterResult>, ImageLayerWriter)> {
        Ok((self.generated_layers, self.inner))
    }
@@ -204,7 +204,7 @@ impl SplitImageLayerWriter {
 /// will split them into multiple files based on size.
 #[must_use]
 pub struct SplitDeltaLayerWriter {
-    inner: DeltaLayerWriter,
+    inner: Option<(Key, DeltaLayerWriter)>,
    target_layer_size: u64,
    generated_layers: Vec<SplitWriterResult>,
    conf: &'static PageServerConf,
@@ -212,7 +212,6 @@ pub struct SplitDeltaLayerWriter {
    tenant_shard_id: TenantShardId,
    lsn_range: Range<Lsn>,
    last_key_written: Key,
-    start_key: Key,
 }

 impl SplitDeltaLayerWriter {
@@ -220,29 +219,18 @@ impl SplitDeltaLayerWriter {
        conf: &'static PageServerConf,
        timeline_id: TimelineId,
        tenant_shard_id: TenantShardId,
-        start_key: Key,
        lsn_range: Range<Lsn>,
        target_layer_size: u64,
-        ctx: &RequestContext,
    ) -> anyhow::Result<Self> {
        Ok(Self {
            target_layer_size,
-            inner: DeltaLayerWriter::new(
-                conf,
-                timeline_id,
-                tenant_shard_id,
-                start_key,
-                lsn_range.clone(),
-                ctx,
-            )
-            .await?,
+            inner: None,
            generated_layers: Vec::new(),
            conf,
            timeline_id,
            tenant_shard_id,
            lsn_range,
            last_key_written: Key::MIN,
-            start_key,
        })
    }

@@ -265,9 +253,26 @@ impl SplitDeltaLayerWriter {
        //
        // Also, keep all updates of a single key in a single file. TODO: split them using the legacy compaction
        // strategy. https://github.com/neondatabase/neon/issues/8837
+
+        if self.inner.is_none() {
+            self.inner = Some((
+                key,
+                DeltaLayerWriter::new(
+                    self.conf,
+                    self.timeline_id,
+                    self.tenant_shard_id,
+                    key,
+                    self.lsn_range.clone(),
+                    ctx,
+                )
+                .await?,
+            ));
+        }
+        let (_, inner) = self.inner.as_mut().unwrap();
+
        let addition_size_estimation = KEY_SIZE as u64 + 8 /* LSN u64 size */ + 80 /* value size estimation */;
-        if self.inner.num_keys() >= 1
-            && self.inner.estimated_size() + addition_size_estimation >= self.target_layer_size
+        if inner.num_keys() >= 1
+            && inner.estimated_size() + addition_size_estimation >= self.target_layer_size
        {
            if key != self.last_key_written {
                let next_delta_writer = DeltaLayerWriter::new(
@@ -279,13 +284,13 @@ impl SplitDeltaLayerWriter {
                    ctx,
                )
                .await?;
-                let prev_delta_writer = std::mem::replace(&mut self.inner, next_delta_writer);
+                let (start_key, prev_delta_writer) =
+                    std::mem::replace(&mut self.inner, Some((key, next_delta_writer))).unwrap();
                let layer_key = PersistentLayerKey {
-                    key_range: self.start_key..key,
+                    key_range: start_key..key,
                    lsn_range: self.lsn_range.clone(),
                    is_delta: true,
                };
-                self.start_key = key;
                if discard(&layer_key).await {
                    drop(prev_delta_writer);
                    self.generated_layers
@@ -296,17 +301,18 @@ impl SplitDeltaLayerWriter {
                    self.generated_layers
                        .push(SplitWriterResult::Produced(delta_layer));
                }
-            } else if self.inner.estimated_size() >= S3_UPLOAD_LIMIT {
+            } else if inner.estimated_size() >= S3_UPLOAD_LIMIT {
                // We have to produce a very large file b/c a key is updated too often.
                anyhow::bail!(
                    "a single key is updated too often: key={}, estimated_size={}, and the layer file cannot be produced",
                    key,
-                    self.inner.estimated_size()
+                    inner.estimated_size()
                );
            }
        }
        self.last_key_written = key;
-        self.inner.put_value(key, lsn, val, ctx).await
+        let (_, inner) = self.inner.as_mut().unwrap();
+        inner.put_value(key, lsn, val, ctx).await
    }

    pub async fn put_value(
@@ -325,7 +331,6 @@ impl SplitDeltaLayerWriter {
        self,
        tline: &Arc<Timeline>,
        ctx: &RequestContext,
-        end_key: Key,
        discard: D,
    ) -> anyhow::Result<Vec<SplitWriterResult>>
    where
@@ -337,11 +342,15 @@ impl SplitDeltaLayerWriter {
            inner,
            ..
        } = self;
+        let Some((start_key, inner)) = inner else {
+            return Ok(generated_layers);
+        };
        if inner.num_keys() == 0 {
            return Ok(generated_layers);
        }
+        let end_key = self.last_key_written.next();
        let layer_key = PersistentLayerKey {
-            key_range: self.start_key..end_key,
+            key_range: start_key..end_key,
            lsn_range: self.lsn_range.clone(),
            is_delta: true,
        };
@@ -360,15 +369,14 @@ impl SplitDeltaLayerWriter {
        self,
        tline: &Arc<Timeline>,
        ctx: &RequestContext,
-        end_key: Key,
    ) -> anyhow::Result<Vec<SplitWriterResult>> {
-        self.finish_with_discard_fn(tline, ctx, end_key, |_| async { false })
+        self.finish_with_discard_fn(tline, ctx, |_| async { false })
            .await
    }

-    /// When split writer fails, the caller should call this function and handle partially generated layers.
-    pub(crate) fn take(self) -> anyhow::Result<(Vec<SplitWriterResult>, DeltaLayerWriter)> {
-        Ok((self.generated_layers, self.inner))
+    /// This function will be deprecated with #8841.
+    pub(crate) fn take(self) -> anyhow::Result<(Vec<SplitWriterResult>, Option<DeltaLayerWriter>)> {
+        Ok((self.generated_layers, self.inner.map(|x| x.1)))
    }
 }

@@ -432,10 +440,8 @@ mod tests {
            tenant.conf,
            tline.timeline_id,
            tenant.tenant_shard_id,
-            get_key(0),
            Lsn(0x18)..Lsn(0x20),
            4 * 1024 * 1024,
-            &ctx,
        )
        .await
        .unwrap();
@@ -460,11 +466,22 @@ mod tests {
            )
            .await
            .unwrap();
-        let layers = delta_writer
-            .finish(&tline, &ctx, get_key(10))
-            .await
-            .unwrap();
+        let layers = delta_writer.finish(&tline, &ctx).await.unwrap();
        assert_eq!(layers.len(), 1);
+        assert_eq!(
+            layers
+                .into_iter()
+                .next()
+                .unwrap()
+                .into_resident_layer()
+                .layer_desc()
+                .key(),
+            PersistentLayerKey {
+                key_range: get_key(0)..get_key(1),
+                lsn_range: Lsn(0x18)..Lsn(0x20),
+                is_delta: true
+            }
+        );
    }

    #[tokio::test]
@@ -501,10 +518,8 @@ mod tests {
            tenant.conf,
            tline.timeline_id,
            tenant.tenant_shard_id,
-            get_key(0),
            Lsn(0x18)..Lsn(0x20),
            4 * 1024 * 1024,
-            &ctx,
        )
        .await
        .unwrap();
@@ -533,10 +548,7 @@ mod tests {
            .finish(&tline, &ctx, get_key(N as u32))
            .await
            .unwrap();
-        let delta_layers = delta_writer
-            .finish(&tline, &ctx, get_key(N as u32))
-            .await
-            .unwrap();
+        let delta_layers = delta_writer.finish(&tline, &ctx).await.unwrap();
        if discard {
            for layer in image_layers {
                layer.into_discarded_layer();
@@ -555,6 +567,14 @@ mod tests {
                .collect_vec();
            assert_eq!(image_layers.len(), N / 512 + 1);
            assert_eq!(delta_layers.len(), N / 512 + 1);
+            assert_eq!(
+                delta_layers.first().unwrap().layer_desc().key_range.start,
+                get_key(0)
+            );
+            assert_eq!(
+                delta_layers.last().unwrap().layer_desc().key_range.end,
+                get_key(N as u32)
+            );
            for idx in 0..image_layers.len() {
                assert_ne!(image_layers[idx].layer_desc().key_range.start, Key::MIN);
                assert_ne!(image_layers[idx].layer_desc().key_range.end, Key::MAX);
@@ -602,10 +622,8 @@ mod tests {
            tenant.conf,
            tline.timeline_id,
            tenant.tenant_shard_id,
-            get_key(0),
            Lsn(0x18)..Lsn(0x20),
            4 * 1024,
-            &ctx,
        )
        .await
        .unwrap();
@@ -644,11 +662,35 @@ mod tests {
            )
            .await
            .unwrap();
-        let layers = delta_writer
-            .finish(&tline, &ctx, get_key(10))
-            .await
-            .unwrap();
+        let layers = delta_writer.finish(&tline, &ctx).await.unwrap();
        assert_eq!(layers.len(), 2);
+        let mut layers_iter = layers.into_iter();
+        assert_eq!(
+            layers_iter
+                .next()
+                .unwrap()
+                .into_resident_layer()
+                .layer_desc()
+                .key(),
+            PersistentLayerKey {
+                key_range: get_key(0)..get_key(1),
+                lsn_range: Lsn(0x18)..Lsn(0x20),
+                is_delta: true
+            }
+        );
+        assert_eq!(
+            layers_iter
+                .next()
+                .unwrap()
+                .into_resident_layer()
+                .layer_desc()
+                .key(),
+            PersistentLayerKey {
+                key_range: get_key(1)..get_key(2),
+                lsn_range: Lsn(0x18)..Lsn(0x20),
+                is_delta: true
+            }
+        );
    }

    #[tokio::test]
@@ -668,10 +710,8 @@ mod tests {
            tenant.conf,
            tline.timeline_id,
            tenant.tenant_shard_id,
-            get_key(0),
            Lsn(0x10)..Lsn(N as u64 * 16 + 0x10),
            4 * 1024 * 1024,
-            &ctx,
        )
        .await
        .unwrap();
@@ -689,10 +729,20 @@ mod tests {
                .await
                .unwrap();
        }
-        let delta_layers = delta_writer
-            .finish(&tline, &ctx, get_key(N as u32))
-            .await
-            .unwrap();
+        let delta_layers = delta_writer.finish(&tline, &ctx).await.unwrap();
        assert_eq!(delta_layers.len(), 1);
+        let delta_layer = delta_layers
+            .into_iter()
+            .next()
+            .unwrap()
+            .into_resident_layer();
+        assert_eq!(
+            delta_layer.layer_desc().key(),
+            PersistentLayerKey {
+                key_range: get_key(0)..get_key(1),
+                lsn_range: Lsn(0x10)..Lsn(N as u64 * 16 + 0x10),
+                is_delta: true
+            }
+        );
    }
 }
@@ -18,7 +18,6 @@ use camino::Utf8Path;
 use chrono::{DateTime, Utc};
 use enumset::EnumSet;
 use fail::fail_point;
-use futures::{stream::FuturesUnordered, StreamExt};
 use handle::ShardTimelineId;
 use once_cell::sync::Lazy;
 use pageserver_api::{
@@ -49,6 +48,7 @@ use utils::{
    sync::gate::{Gate, GateGuard},
 };

+use std::pin::pin;
 use std::sync::atomic::Ordering as AtomicOrdering;
 use std::sync::{Arc, Mutex, RwLock, Weak};
 use std::time::{Duration, Instant, SystemTime};
@@ -62,16 +62,13 @@ use std::{
    collections::btree_map::Entry,
    ops::{Deref, Range},
 };
-use std::{pin::pin, sync::atomic::AtomicUsize};

 use crate::{
    aux_file::AuxFileSizeEstimator,
    tenant::{
        layer_map::{LayerMap, SearchResult},
        metadata::TimelineMetadata,
-        storage_layer::{
-            convert, inmemory_layer::IndexEntry, PersistentLayerDesc, ValueReconstructSituation,
-        },
+        storage_layer::{inmemory_layer::IndexEntry, PersistentLayerDesc},
    },
    walredo,
 };
@@ -568,7 +565,7 @@ impl From<layer_manager::Shutdown> for GetVectoredError {

 #[derive(thiserror::Error)]
 pub struct MissingKeyError {
-    keys: KeySpace,
+    key: Key,
    shard: ShardNumber,
    cont_lsn: Lsn,
    request_lsn: Lsn,
@@ -586,8 +583,8 @@ impl std::fmt::Display for MissingKeyError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
-            "could not find data for keys (shard {:?}) at LSN {}, request LSN {} keys {}",
-            self.shard, self.cont_lsn, self.request_lsn, self.keys
+            "could not find data for key {} (shard {:?}) at LSN {}, request LSN {}",
+            self.key, self.shard, self.cont_lsn, self.request_lsn
        )?;
        if let Some(ref ancestor_lsn) = self.ancestor_lsn {
            write!(f, ", ancestor {}", ancestor_lsn)?;
@@ -955,11 +952,7 @@ impl Timeline {
                }
            }
            None => Err(PageReconstructError::MissingKey(MissingKeyError {
-                keys: {
-                    let mut accum = KeySpaceAccum::new();
-                    accum.add_key(key);
-                    accum.to_keyspace()
-                },
+                key,
                shard: self.shard_identity.get_shard_number(&key),
                cont_lsn: Lsn(0),
                request_lsn: lsn,
@@ -1129,53 +1122,29 @@ impl Timeline {
        let get_data_timer = crate::metrics::GET_RECONSTRUCT_DATA_TIME
            .for_get_kind(get_kind)
            .start_timer();
-        static INVOCATION: Lazy<AtomicUsize> = Lazy::new(|| AtomicUsize::new(0));
-        let invocation = INVOCATION.fetch_add(1, AtomicOrdering::Relaxed);
        self.get_vectored_reconstruct_data(keyspace.clone(), lsn, reconstruct_state, ctx)
-            .instrument(debug_span!("get_vectored_reconstruct_data", invocation))
-            .await
-            .map_err(|err| {
-                anyhow::anyhow!(
-                    "get_vectored_reconstruct_data invocation {invocation} lsn={lsn} keyspace={keyspace} {err:?}",
-                )
-            })?;
+            .await?;
        get_data_timer.stop_and_record();

        let reconstruct_timer = crate::metrics::RECONSTRUCT_TIME
            .for_get_kind(get_kind)
            .start_timer();
+        let mut results: BTreeMap<Key, Result<Bytes, PageReconstructError>> = BTreeMap::new();
        let layers_visited = reconstruct_state.get_layers_visited();

-        let futs = FuturesUnordered::new();
        for (key, res) in std::mem::take(&mut reconstruct_state.keys) {
-            futs.push({
-                let walredo_self = self.myself.upgrade().expect("&self method holds the arc");
-                async move {
-                    let state = res.expect("Read path is infallible");
-                    assert!(matches!(
-                        state.situation,
-                        ValueReconstructSituation::Complete
-                    ));
-
-                    let converted = match convert(key, state).await {
-                        Ok(ok) => ok,
-                        Err(err) => {
-                            return (key, Err(err));
-                        }
-                    };
-
-                    (
-                        key,
-                        walredo_self.reconstruct_value(key, lsn, converted).await,
-                    )
+            match res {
+                Err(err) => {
+                    results.insert(key, Err(err));
                }
-            });
+                Ok(state) => {
+                    let state = ValueReconstructState::from(state);
+
+                    let reconstruct_res = self.reconstruct_value(key, lsn, state).await;
+                    results.insert(key, reconstruct_res);
+                }
+            }
        }
-
-        let results = futs
-            .collect::<BTreeMap<Key, Result<Bytes, PageReconstructError>>>()
-            .await;
-
        reconstruct_timer.stop_and_record();

        // For aux file keys (v1 or v2) the vectored read path does not return an error
@@ -3151,7 +3120,7 @@ impl Timeline {

        if let Some(missing_keyspace) = missing_keyspace {
            return Err(GetVectoredError::MissingKey(MissingKeyError {
-                keys: missing_keyspace.clone(),
+                key: missing_keyspace.start().unwrap(), /* better if we can store the full keyspace */
                shard: self
                    .shard_identity
                    .get_shard_number(&missing_keyspace.start().unwrap()),
@@ -5527,30 +5496,30 @@ impl Timeline {
    #[cfg(test)]
    pub(crate) async fn inspect_image_layers(
        self: &Arc<Timeline>,
-        _lsn: Lsn,
-        _ctx: &RequestContext,
+        lsn: Lsn,
+        ctx: &RequestContext,
    ) -> anyhow::Result<Vec<(Key, Bytes)>> {
-        // let mut all_data = Vec::new();
-        // let guard = self.layers.read().await;
-        // for layer in guard.layer_map()?.iter_historic_layers() {
-        //     if !layer.is_delta() && layer.image_layer_lsn() == lsn {
-        //         let layer = guard.get_from_desc(&layer);
-        //         let mut reconstruct_data = ValuesReconstructState::default();
-        //         layer
-        //             .get_values_reconstruct_data(
-        //                 KeySpace::single(Key::MIN..Key::MAX),
-        //                 lsn..Lsn(lsn.0 + 1),
-        //                 &mut reconstruct_data,
-        //                 ctx,
-        //             )
-        //             .await?;
-        //         for (k, v) in reconstruct_data.keys {
-        //             all_data.push((k, v?.img.unwrap().1));
-        //         }
-        //     }
-        // }
-        // all_data.sort();
-        Ok(Vec::new())
+        let mut all_data = Vec::new();
+        let guard = self.layers.read().await;
+        for layer in guard.layer_map()?.iter_historic_layers() {
+            if !layer.is_delta() && layer.image_layer_lsn() == lsn {
+                let layer = guard.get_from_desc(&layer);
+                let mut reconstruct_data = ValuesReconstructState::default();
+                layer
+                    .get_values_reconstruct_data(
+                        KeySpace::single(Key::MIN..Key::MAX),
+                        lsn..Lsn(lsn.0 + 1),
+                        &mut reconstruct_data,
+                        ctx,
+                    )
+                    .await?;
+                for (k, v) in reconstruct_data.keys {
+                    all_data.push((k, v?.img.unwrap().1));
+                }
+            }
+        }
+        all_data.sort();
+        Ok(all_data)
    }

    /// Get all historic layer descriptors in the layer map
@@ -1809,7 +1809,6 @@ impl Timeline {
            .unwrap();
        // We don't want any of the produced layers to cover the full key range (i.e., MIN..MAX) b/c it will then be recognized
        // as an L0 layer.
-        let hack_end_key = Key::NON_L0_MAX;
        let mut delta_layers = Vec::new();
        let mut image_layers = Vec::new();
        let mut downloaded_layers = Vec::new();
@@ -1855,10 +1854,8 @@ impl Timeline {
            self.conf,
            self.timeline_id,
            self.tenant_shard_id,
-            Key::MIN,
            lowest_retain_lsn..end_lsn,
            self.get_compaction_target_size(),
-            ctx,
        )
        .await?;

@@ -1965,7 +1962,7 @@ impl Timeline {
        let produced_image_layers = if let Some(writer) = image_layer_writer {
            if !dry_run {
                writer
-                    .finish_with_discard_fn(self, ctx, hack_end_key, discard)
+                    .finish_with_discard_fn(self, ctx, Key::MAX, discard)
                    .await?
            } else {
                let (layers, _) = writer.take()?;
@@ -1978,7 +1975,7 @@ impl Timeline {

        let produced_delta_layers = if !dry_run {
            delta_layer_writer
-                .finish_with_discard_fn(self, ctx, hack_end_key, discard)
+                .finish_with_discard_fn(self, ctx, discard)
                .await?
        } else {
            let (layers, _) = delta_layer_writer.take()?;
@@ -33,7 +33,6 @@ use crate::virtual_file::{self, VirtualFile};
 pub struct BlobMeta {
    pub key: Key,
    pub lsn: Lsn,
-    pub will_init: bool,
 }

 /// Blob offsets into [`VectoredBlobsBuf::buf`]
@@ -356,10 +355,9 @@ pub enum BlobFlag {
 /// * Iterate over the collected blobs and coalesce them into reads at the end
 pub struct VectoredReadPlanner {
    // Track all the blob offsets. Start offsets must be ordered.
-    // Note: last bool is will_init
-    blobs: BTreeMap<Key, Vec<(Lsn, u64, u64, bool)>>,
+    blobs: BTreeMap<Key, Vec<(Lsn, u64, u64)>>,
    // Arguments for previous blob passed into [`VectoredReadPlanner::handle`]
-    prev: Option<(Key, Lsn, u64, BlobFlag, bool)>,
+    prev: Option<(Key, Lsn, u64, BlobFlag)>,

    max_read_size: usize,

@@ -394,62 +392,40 @@ impl VectoredReadPlanner {
    ///   This is used for WAL records that `will_init`.
    /// * [`BlobFlag::Ignore`]: This blob should not be included in the read. This happens
    ///   if the blob is cached.
-    pub fn handle(&mut self, key: Key, lsn: Lsn, offset: u64, flag: BlobFlag, will_init: bool) {
+    pub fn handle(&mut self, key: Key, lsn: Lsn, offset: u64, flag: BlobFlag) {
        // Implementation note: internally lag behind by one blob such that
        // we have a start and end offset when initialising [`VectoredRead`]
-        let (prev_key, prev_lsn, prev_offset, prev_flag, prev_will_init) = match self.prev {
+        let (prev_key, prev_lsn, prev_offset, prev_flag) = match self.prev {
            None => {
-                self.prev = Some((key, lsn, offset, flag, will_init));
+                self.prev = Some((key, lsn, offset, flag));
                return;
            }
            Some(prev) => prev,
        };

-        self.add_blob(
-            prev_key,
-            prev_lsn,
-            prev_offset,
-            offset,
-            prev_flag,
-            prev_will_init,
-        );
+        self.add_blob(prev_key, prev_lsn, prev_offset, offset, prev_flag);

-        self.prev = Some((key, lsn, offset, flag, will_init));
+        self.prev = Some((key, lsn, offset, flag));
    }

    pub fn handle_range_end(&mut self, offset: u64) {
-        if let Some((prev_key, prev_lsn, prev_offset, prev_flag, prev_will_init)) = self.prev {
-            self.add_blob(
-                prev_key,
-                prev_lsn,
-                prev_offset,
-                offset,
-                prev_flag,
-                prev_will_init,
-            );
+        if let Some((prev_key, prev_lsn, prev_offset, prev_flag)) = self.prev {
+            self.add_blob(prev_key, prev_lsn, prev_offset, offset, prev_flag);
        }

        self.prev = None;
    }

-    fn add_blob(
-        &mut self,
-        key: Key,
-        lsn: Lsn,
-        start_offset: u64,
-        end_offset: u64,
-        flag: BlobFlag,
-        will_init: bool,
-    ) {
+    fn add_blob(&mut self, key: Key, lsn: Lsn, start_offset: u64, end_offset: u64, flag: BlobFlag) {
        match flag {
            BlobFlag::None => {
                let blobs_for_key = self.blobs.entry(key).or_default();
-                blobs_for_key.push((lsn, start_offset, end_offset, will_init));
+                blobs_for_key.push((lsn, start_offset, end_offset));
            }
            BlobFlag::ReplaceAll => {
                let blobs_for_key = self.blobs.entry(key).or_default();
                blobs_for_key.clear();
-                blobs_for_key.push((lsn, start_offset, end_offset, will_init));
+                blobs_for_key.push((lsn, start_offset, end_offset));
            }
            BlobFlag::Ignore => {}
        }
@@ -460,17 +436,11 @@ impl VectoredReadPlanner {
        let mut reads = Vec::new();

        for (key, blobs_for_key) in self.blobs {
-            for (lsn, start_offset, end_offset, will_init) in blobs_for_key {
+            for (lsn, start_offset, end_offset) in blobs_for_key {
                let extended = match &mut current_read_builder {
-                    Some(read_builder) => read_builder.extend(
-                        start_offset,
-                        end_offset,
-                        BlobMeta {
-                            key,
-                            lsn,
-                            will_init,
-                        },
-                    ),
+                    Some(read_builder) => {
+                        read_builder.extend(start_offset, end_offset, BlobMeta { key, lsn })
+                    }
                    None => VectoredReadExtended::No,
                };

@@ -478,11 +448,7 @@ impl VectoredReadPlanner {
                    let next_read_builder = VectoredReadBuilder::new(
                        start_offset,
                        end_offset,
-                        BlobMeta {
-                            key,
-                            lsn,
-                            will_init,
-                        },
+                        BlobMeta { key, lsn },
                        self.max_read_size,
                        self.mode,
                    );
@@ -702,15 +668,7 @@ impl StreamingVectoredReadPlanner {
    ) -> Option<VectoredRead> {
        match &mut self.read_builder {
            Some(read_builder) => {
-                let extended = read_builder.extend(
-                    start_offset,
-                    end_offset,
-                    BlobMeta {
-                        key,
-                        lsn,
-                        will_init: todo!(),
-                    },
-                );
+                let extended = read_builder.extend(start_offset, end_offset, BlobMeta { key, lsn });
                assert_eq!(extended, VectoredReadExtended::Yes);
            }
            None => {
@@ -718,11 +676,7 @@ impl StreamingVectoredReadPlanner {
                    Some(VectoredReadBuilder::new_streaming(
                        start_offset,
                        end_offset,
-                        BlobMeta {
-                            key,
-                            lsn,
-                            will_init: todo!(),
-                        },
+                        BlobMeta { key, lsn },
                        self.mode,
                    ))
                };
@@ -1054,7 +1008,6 @@ mod tests {
        let meta = BlobMeta {
            key: Key::MIN,
            lsn: Lsn(0),
-            will_init: false,
        };

        for (idx, (blob, offset)) in blobs.iter().zip(offsets.iter()).enumerate() {
@@ -25,9 +25,7 @@ use std::time::Duration;
 use std::time::SystemTime;

 use pageserver_api::shard::ShardIdentity;
-use postgres_ffi::v14::nonrelfile_utils::clogpage_precedes;
-use postgres_ffi::v14::nonrelfile_utils::slru_may_delete_clogsegment;
-use postgres_ffi::TimestampTz;
+use postgres_ffi::{dispatch_pgversion, enum_pgversion, enum_pgversion_dispatch, TimestampTz};
 use postgres_ffi::{fsm_logical_to_physical, page_is_new, page_set_lsn};

 use anyhow::{bail, Context, Result};
@@ -48,16 +46,31 @@ use pageserver_api::key::rel_block_to_key;
 use pageserver_api::reltag::{BlockNumber, RelTag, SlruKind};
 use postgres_ffi::pg_constants;
 use postgres_ffi::relfile_utils::{FSM_FORKNUM, INIT_FORKNUM, MAIN_FORKNUM, VISIBILITYMAP_FORKNUM};
-use postgres_ffi::v14::nonrelfile_utils::mx_offset_to_member_segment;
-use postgres_ffi::v14::xlog_utils::*;
-use postgres_ffi::v14::CheckPoint;
 use postgres_ffi::TransactionId;
 use postgres_ffi::BLCKSZ;
+use utils::bin_ser::SerializeError;
 use utils::lsn::Lsn;

+enum_pgversion! {CheckPoint, pgv::CheckPoint}
+
+impl CheckPoint {
+    fn encode(&self) -> Result<Bytes, SerializeError> {
+        enum_pgversion_dispatch!(self, CheckPoint, cp, { cp.encode() })
+    }
+
+    fn update_next_xid(&mut self, xid: u32) -> bool {
+        enum_pgversion_dispatch!(self, CheckPoint, cp, { cp.update_next_xid(xid) })
+    }
+
+    pub fn update_next_multixid(&mut self, multi_xid: u32, multi_offset: u32) -> bool {
+        enum_pgversion_dispatch!(self, CheckPoint, cp, {
+            cp.update_next_multixid(multi_xid, multi_offset)
+        })
+    }
+}
+
 pub struct WalIngest {
    shard: ShardIdentity,
-    pg_version: u32,
    checkpoint: CheckPoint,
    checkpoint_modified: bool,
    warn_ingest_lag: WarnIngestLag,
@@ -78,12 +91,16 @@ impl WalIngest {
        // Fetch the latest checkpoint into memory, so that we can compare with it
        // quickly in `ingest_record` and update it when it changes.
        let checkpoint_bytes = timeline.get_checkpoint(startpoint, ctx).await?;
-        let checkpoint = CheckPoint::decode(&checkpoint_bytes)?;
-        trace!("CheckPoint.nextXid = {}", checkpoint.nextXid.value);
+        let pgversion = timeline.pg_version;
+
+        let checkpoint = dispatch_pgversion!(pgversion, {
+            let checkpoint = pgv::CheckPoint::decode(&checkpoint_bytes)?;
+            trace!("CheckPoint.nextXid = {}", checkpoint.nextXid.value);
+            <pgv::CheckPoint as Into<CheckPoint>>::into(checkpoint)
+        });

        Ok(WalIngest {
            shard: *timeline.get_shard_identity(),
-            pg_version: timeline.pg_version,
            checkpoint,
            checkpoint_modified: false,
            warn_ingest_lag: WarnIngestLag {
@@ -117,7 +134,7 @@ impl WalIngest {

        modification.set_lsn(lsn)?;

-        if decoded.is_dbase_create_copy(self.pg_version) {
+        if decoded.is_dbase_create_copy(pg_version) {
            // Records of this type should always be preceded by a commit(), as they
            // rely on reading data pages back from the Timeline.
            assert!(!modification.has_dirty_data_pages());
@@ -337,70 +354,67 @@ impl WalIngest {
            pg_constants::RM_XLOG_ID => {
                let info = decoded.xl_info & pg_constants::XLR_RMGR_INFO_MASK;

-                if info == pg_constants::XLOG_NEXTOID {
-                    let next_oid = buf.get_u32_le();
-                    if self.checkpoint.nextOid != next_oid {
-                        self.checkpoint.nextOid = next_oid;
+                enum_pgversion_dispatch!(&mut self.checkpoint, CheckPoint, cp, {
+                    if info == pg_constants::XLOG_NEXTOID {
+                        let next_oid = buf.get_u32_le();
+                        if cp.nextOid != next_oid {
+                            cp.nextOid = next_oid;
+                            self.checkpoint_modified = true;
+                        }
+                    } else if info == pg_constants::XLOG_CHECKPOINT_ONLINE
+                        || info == pg_constants::XLOG_CHECKPOINT_SHUTDOWN
+                    {
+                        let mut checkpoint_bytes = [0u8; pgv::xlog_utils::SIZEOF_CHECKPOINT];
+                        buf.copy_to_slice(&mut checkpoint_bytes);
+                        let xlog_checkpoint = pgv::CheckPoint::decode(&checkpoint_bytes)?;
+                        trace!(
+                            "xlog_checkpoint.oldestXid={}, checkpoint.oldestXid={}",
+                            xlog_checkpoint.oldestXid,
+                            cp.oldestXid
+                        );
+                        if (cp.oldestXid.wrapping_sub(xlog_checkpoint.oldestXid) as i32) < 0 {
+                            cp.oldestXid = xlog_checkpoint.oldestXid;
+                        }
+                        trace!(
+                            "xlog_checkpoint.oldestActiveXid={}, checkpoint.oldestActiveXid={}",
+                            xlog_checkpoint.oldestActiveXid,
+                            cp.oldestActiveXid
+                        );
+
+                        // A shutdown checkpoint has `oldestActiveXid == InvalidTransactionid`,
+                        // because at shutdown, all in-progress transactions will implicitly
+                        // end. Postgres startup code knows that, and allows hot standby to start
+                        // immediately from a shutdown checkpoint.
+                        //
+                        // In Neon, Postgres hot standby startup always behaves as if starting from
+                        // an online checkpoint. It needs a valid `oldestActiveXid` value, so
+                        // instead of overwriting self.checkpoint.oldestActiveXid with
+                        // InvalidTransactionid from the checkpoint WAL record, update it to a
+                        // proper value, knowing that there are no in-progress transactions at this
+                        // point, except for prepared transactions.
+                        //
+                        // See also the neon code changes in the InitWalRecovery() function.
+                        if xlog_checkpoint.oldestActiveXid == pg_constants::INVALID_TRANSACTION_ID
+                            && info == pg_constants::XLOG_CHECKPOINT_SHUTDOWN
+                        {
+                            let mut oldest_active_xid = cp.nextXid.value as u32;
+                            for xid in modification.tline.list_twophase_files(lsn, ctx).await? {
+                                if (xid.wrapping_sub(oldest_active_xid) as i32) < 0 {
+                                    oldest_active_xid = xid;
+                                }
+                            }
+                            cp.oldestActiveXid = oldest_active_xid;
+                        } else {
+                            cp.oldestActiveXid = xlog_checkpoint.oldestActiveXid;
+                        }
+
+                        // Write a new checkpoint key-value pair on every checkpoint record, even
+                        // if nothing really changed. Not strictly required, but it seems nice to
+                        // have some trace of the checkpoint records in the layer files at the same
+                        // LSNs.
                        self.checkpoint_modified = true;
                    }
-                } else if info == pg_constants::XLOG_CHECKPOINT_ONLINE
-                    || info == pg_constants::XLOG_CHECKPOINT_SHUTDOWN
-                {
-                    let mut checkpoint_bytes = [0u8; SIZEOF_CHECKPOINT];
-                    buf.copy_to_slice(&mut checkpoint_bytes);
-                    let xlog_checkpoint = CheckPoint::decode(&checkpoint_bytes)?;
-                    trace!(
-                        "xlog_checkpoint.oldestXid={}, checkpoint.oldestXid={}",
-                        xlog_checkpoint.oldestXid,
-                        self.checkpoint.oldestXid
-                    );
-                    if (self
-                        .checkpoint
-                        .oldestXid
-                        .wrapping_sub(xlog_checkpoint.oldestXid) as i32)
-                        < 0
-                    {
-                        self.checkpoint.oldestXid = xlog_checkpoint.oldestXid;
-                    }
-                    trace!(
-                        "xlog_checkpoint.oldestActiveXid={}, checkpoint.oldestActiveXid={}",
-                        xlog_checkpoint.oldestActiveXid,
-                        self.checkpoint.oldestActiveXid
-                    );
-
-                    // A shutdown checkpoint has `oldestActiveXid == InvalidTransactionid`,
-                    // because at shutdown, all in-progress transactions will implicitly
-                    // end. Postgres startup code knows that, and allows hot standby to start
-                    // immediately from a shutdown checkpoint.
-                    //
-                    // In Neon, Postgres hot standby startup always behaves as if starting from
-                    // an online checkpoint. It needs a valid `oldestActiveXid` value, so
-                    // instead of overwriting self.checkpoint.oldestActiveXid with
-                    // InvalidTransactionid from the checkpoint WAL record, update it to a
-                    // proper value, knowing that there are no in-progress transactions at this
-                    // point, except for prepared transactions.
-                    //
-                    // See also the neon code changes in the InitWalRecovery() function.
-                    if xlog_checkpoint.oldestActiveXid == pg_constants::INVALID_TRANSACTION_ID
-                        && info == pg_constants::XLOG_CHECKPOINT_SHUTDOWN
-                    {
-                        let mut oldest_active_xid = self.checkpoint.nextXid.value as u32;
-                        for xid in modification.tline.list_twophase_files(lsn, ctx).await? {
-                            if (xid.wrapping_sub(oldest_active_xid) as i32) < 0 {
-                                oldest_active_xid = xid;
-                            }
-                        }
-                        self.checkpoint.oldestActiveXid = oldest_active_xid;
-                    } else {
-                        self.checkpoint.oldestActiveXid = xlog_checkpoint.oldestActiveXid;
-                    }
-
-                    // Write a new checkpoint key-value pair on every checkpoint record, even
-                    // if nothing really changed. Not strictly required, but it seems nice to
-                    // have some trace of the checkpoint records in the layer files at the same
-                    // LSNs.
-                    self.checkpoint_modified = true;
-                }
+                });
            }
            pg_constants::RM_LOGICALMSG_ID => {
                let info = decoded.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
@@ -424,7 +438,11 @@ impl WalIngest {
                let info = decoded.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
                if info == pg_constants::XLOG_RUNNING_XACTS {
                    let xlrec = crate::walrecord::XlRunningXacts::decode(&mut buf);
-                    self.checkpoint.oldestActiveXid = xlrec.oldest_running_xid;
+
+                    enum_pgversion_dispatch!(&mut self.checkpoint, CheckPoint, cp, {
+                        cp.oldestActiveXid = xlrec.oldest_running_xid;
+                    });
+
                    self.checkpoint_modified = true;
                }
            }
@@ -539,7 +557,7 @@ impl WalIngest {
            && blk.has_image
            && decoded.xl_rmid == pg_constants::RM_XLOG_ID
            && (decoded.xl_info == pg_constants::XLOG_FPI
-                || decoded.xl_info == pg_constants::XLOG_FPI_FOR_HINT)
+            || decoded.xl_info == pg_constants::XLOG_FPI_FOR_HINT)
            // compression of WAL is not yet supported: fall back to storing the original WAL record
            && !postgres_ffi::bkpimage_is_compressed(blk.bimg_info, modification.tline.pg_version)
            // do not materialize null pages because them most likely be soon replaced with real data
@@ -1204,7 +1222,7 @@ impl WalIngest {
            if rec.blkno % pg_constants::SLOTS_PER_FSM_PAGE != 0 {
                // Tail of last remaining FSM page has to be zeroed.
                // We are not precise here and instead of digging in FSM bitmap format just clear the whole page.
-                modification.put_rel_page_image_zero(rel, fsm_physical_page_no);
+                modification.put_rel_page_image_zero(rel, fsm_physical_page_no)?;
                fsm_physical_page_no += 1;
            }
            let nblocks = get_relsize(modification, rel, ctx).await?;
@@ -1226,7 +1244,7 @@ impl WalIngest {
            if rec.blkno % pg_constants::VM_HEAPBLOCKS_PER_PAGE != 0 {
                // Tail of last remaining vm page has to be zeroed.
                // We are not precise here and instead of digging in VM bitmap format just clear the whole page.
-                modification.put_rel_page_image_zero(rel, vm_page_no);
+                modification.put_rel_page_image_zero(rel, vm_page_no)?;
                vm_page_no += 1;
            }
            let nblocks = get_relsize(modification, rel, ctx).await?;
@@ -1242,12 +1260,17 @@ impl WalIngest {
    fn warn_on_ingest_lag(
        &mut self,
        conf: &crate::config::PageServerConf,
-        wal_timestmap: TimestampTz,
+        wal_timestamp: TimestampTz,
    ) {
        debug_assert_current_span_has_tenant_and_timeline_id();
        let now = SystemTime::now();
        let rate_limits = &mut self.warn_ingest_lag;
-        match try_from_pg_timestamp(wal_timestmap) {
+
+        let ts = enum_pgversion_dispatch!(&self.checkpoint, CheckPoint, _cp, {
+            pgv::xlog_utils::try_from_pg_timestamp(wal_timestamp)
+        });
+
+        match ts {
            Ok(ts) => {
                match now.duration_since(ts) {
                    Ok(lag) => {
@@ -1257,7 +1280,7 @@ impl WalIngest {
                                warn!(%rate_limit_stats, %lag, "ingesting record with timestamp lagging more than wait_lsn_timeout");
                            })
                        }
-                    },
+                    }
                    Err(e) => {
                        let delta_t = e.duration();
                        // determined by prod victoriametrics query: 1000 * (timestamp(node_time_seconds{neon_service="pageserver"}) - node_time_seconds)
@@ -1271,7 +1294,6 @@ impl WalIngest {
                        }
                    }
                };
-
            }
            Err(error) => {
                rate_limits.timestamp_invalid_msg_ratelimit.call2(|rate_limit_stats| {
@@ -1379,14 +1401,17 @@ impl WalIngest {
        // truncated, but a checkpoint record with the updated values isn't written until
        // later. In Neon, a server can start at any LSN, not just on a checkpoint record,
        // so we keep the oldestXid and oldestXidDB up-to-date.
-        self.checkpoint.oldestXid = xlrec.oldest_xid;
-        self.checkpoint.oldestXidDB = xlrec.oldest_xid_db;
+        enum_pgversion_dispatch!(&mut self.checkpoint, CheckPoint, cp, {
+            cp.oldestXid = xlrec.oldest_xid;
+            cp.oldestXidDB = xlrec.oldest_xid_db;
+        });
        self.checkpoint_modified = true;

        // TODO Treat AdvanceOldestClogXid() or write a comment why we don't need it

        let latest_page_number =
-            self.checkpoint.nextXid.value as u32 / pg_constants::CLOG_XACTS_PER_PAGE;
+            enum_pgversion_dispatch!(self.checkpoint, CheckPoint, cp, { cp.nextXid.value }) as u32
+                / pg_constants::CLOG_XACTS_PER_PAGE;

        // Now delete all segments containing pages between xlrec.pageno
        // and latest_page_number.
@@ -1394,7 +1419,9 @@ impl WalIngest {
        // First, make an important safety check:
        // the current endpoint page must not be eligible for removal.
        // See SimpleLruTruncate() in slru.c
-        if clogpage_precedes(latest_page_number, xlrec.pageno) {
+        if dispatch_pgversion!(modification.tline.pg_version, {
+            pgv::nonrelfile_utils::clogpage_precedes(latest_page_number, xlrec.pageno)
+        }) {
            info!("could not truncate directory pg_xact apparent wraparound");
            return Ok(());
        }
@@ -1411,7 +1438,12 @@ impl WalIngest {
            .await?
        {
            let segpage = segno * pg_constants::SLRU_PAGES_PER_SEGMENT;
-            if slru_may_delete_clogsegment(segpage, xlrec.pageno) {
+
+            let may_delete = dispatch_pgversion!(modification.tline.pg_version, {
+                pgv::nonrelfile_utils::slru_may_delete_clogsegment(segpage, xlrec.pageno)
+            });
+
+            if may_delete {
                modification
                    .drop_slru_segment(SlruKind::Clog, segno, ctx)
                    .await?;
@@ -1530,14 +1562,23 @@ impl WalIngest {
        xlrec: &XlMultiXactTruncate,
        ctx: &RequestContext,
    ) -> Result<()> {
-        self.checkpoint.oldestMulti = xlrec.end_trunc_off;
-        self.checkpoint.oldestMultiDB = xlrec.oldest_multi_db;
+        let (maxsegment, startsegment, endsegment) =
+            enum_pgversion_dispatch!(&mut self.checkpoint, CheckPoint, cp, {
+                cp.oldestMulti = xlrec.end_trunc_off;
+                cp.oldestMultiDB = xlrec.oldest_multi_db;
+                let maxsegment: i32 = pgv::nonrelfile_utils::mx_offset_to_member_segment(
+                    pg_constants::MAX_MULTIXACT_OFFSET,
+                );
+                let startsegment: i32 =
+                    pgv::nonrelfile_utils::mx_offset_to_member_segment(xlrec.start_trunc_memb);
+                let endsegment: i32 =
+                    pgv::nonrelfile_utils::mx_offset_to_member_segment(xlrec.end_trunc_memb);
+                (maxsegment, startsegment, endsegment)
+            });
+
        self.checkpoint_modified = true;

        // PerformMembersTruncation
-        let maxsegment: i32 = mx_offset_to_member_segment(pg_constants::MAX_MULTIXACT_OFFSET);
-        let startsegment: i32 = mx_offset_to_member_segment(xlrec.start_trunc_memb);
-        let endsegment: i32 = mx_offset_to_member_segment(xlrec.end_trunc_memb);
        let mut segment: i32 = startsegment;

        // Delete all the segments except the last one. The last segment can still
@@ -1696,7 +1737,7 @@ impl WalIngest {
                    continue;
                }

-                modification.put_rel_page_image_zero(rel, gap_blknum);
+                modification.put_rel_page_image_zero(rel, gap_blknum)?;
            }
        }
        Ok(())
@@ -1762,7 +1803,7 @@ impl WalIngest {

            // fill the gap with zeros
            for gap_blknum in old_nblocks..blknum {
-                modification.put_slru_page_image_zero(kind, segno, gap_blknum);
+                modification.put_slru_page_image_zero(kind, segno, gap_blknum)?;
            }
        }
        Ok(())
@@ -1811,11 +1852,23 @@ mod tests {
        // TODO
    }

-    static ZERO_CHECKPOINT: Bytes = Bytes::from_static(&[0u8; SIZEOF_CHECKPOINT]);
+    #[tokio::test]
+    async fn test_zeroed_checkpoint_decodes_correctly() -> Result<()> {
+        for i in 14..=16 {
+            dispatch_pgversion!(i, {
+                pgv::CheckPoint::decode(&pgv::ZERO_CHECKPOINT)?;
+            });
+        }
+
+        Ok(())
+    }

    async fn init_walingest_test(tline: &Timeline, ctx: &RequestContext) -> Result<WalIngest> {
        let mut m = tline.begin_modification(Lsn(0x10));
-        m.put_checkpoint(ZERO_CHECKPOINT.clone())?;
+        m.put_checkpoint(dispatch_pgversion!(
+            tline.pg_version,
+            pgv::ZERO_CHECKPOINT.clone()
+        ))?;
        m.put_relmap_file(0, 111, Bytes::from(""), ctx).await?; // dummy relmapper file
        m.commit(ctx).await?;
        let walingest = WalIngest::new(tline, Lsn(0x10), ctx).await?;
@@ -242,6 +242,6 @@ mod tests {
    #[test]
    fn test() {
        let s = "{\"branch_created\":null,\"endpoint_created\":{\"endpoint_id\":\"ep-rapid-thunder-w0qqw2q9\"},\"project_created\":null,\"type\":\"endpoint_created\"}";
-        let _: ControlPlaneEventKey = serde_json::from_str(s).unwrap();
+        serde_json::from_str::<ControlPlaneEventKey>(s).unwrap();
    }
 }
@@ -395,7 +395,7 @@ mod tests {
                }
            }
        });
-        let _: KickSession<'_> = serde_json::from_str(&json.to_string())?;
+        serde_json::from_str::<KickSession<'_>>(&json.to_string())?;

        Ok(())
    }
@@ -403,7 +403,7 @@ mod tests {
    #[test]
    fn parse_db_info() -> anyhow::Result<()> {
        // with password
-        let _: DatabaseInfo = serde_json::from_value(json!({
+        serde_json::from_value::<DatabaseInfo>(json!({
            "host": "localhost",
            "port": 5432,
            "dbname": "postgres",
@@ -413,7 +413,7 @@ mod tests {
        }))?;

        // without password
-        let _: DatabaseInfo = serde_json::from_value(json!({
+        serde_json::from_value::<DatabaseInfo>(json!({
            "host": "localhost",
            "port": 5432,
            "dbname": "postgres",
@@ -422,7 +422,7 @@ mod tests {
        }))?;

        // new field (forward compatibility)
-        let _: DatabaseInfo = serde_json::from_value(json!({
+        serde_json::from_value::<DatabaseInfo>(json!({
            "host": "localhost",
            "port": 5432,
            "dbname": "postgres",
@@ -441,7 +441,7 @@ mod tests {
            "address": "0.0.0.0",
            "aux": dummy_aux(),
        });
-        let _: WakeCompute = serde_json::from_str(&json.to_string())?;
+        serde_json::from_str::<WakeCompute>(&json.to_string())?;
        Ok(())
    }

@@ -451,18 +451,18 @@ mod tests {
        let json = json!({
            "role_secret": "secret",
        });
-        let _: GetRoleSecret = serde_json::from_str(&json.to_string())?;
+        serde_json::from_str::<GetRoleSecret>(&json.to_string())?;
        let json = json!({
            "role_secret": "secret",
            "allowed_ips": ["8.8.8.8"],
        });
-        let _: GetRoleSecret = serde_json::from_str(&json.to_string())?;
+        serde_json::from_str::<GetRoleSecret>(&json.to_string())?;
        let json = json!({
            "role_secret": "secret",
            "allowed_ips": ["8.8.8.8"],
            "project_id": "project",
        });
-        let _: GetRoleSecret = serde_json::from_str(&json.to_string())?;
+        serde_json::from_str::<GetRoleSecret>(&json.to_string())?;

        Ok(())
    }
@@ -78,7 +78,7 @@ pub(crate) type ComputeReady = DatabaseInfo;

 // TODO: replace with an http-based protocol.
 struct MgmtHandler;
-#[async_trait::async_trait]
+
 impl postgres_backend::Handler<tokio::net::TcpStream> for MgmtHandler {
    async fn process_query(
        &mut self,
@@ -6,7 +6,7 @@ use pq_proto::StartupMessageParams;
 use smol_str::SmolStr;
 use std::net::IpAddr;
 use tokio::sync::mpsc;
-use tracing::{field::display, info, info_span, Span};
+use tracing::{debug, field::display, info, info_span, Span};
 use try_lock::TryLock;
 use uuid::Uuid;

@@ -362,7 +362,9 @@ impl RequestMonitoringInner {
                });
        }
        if let Some(tx) = self.sender.take() {
-            let _: Result<(), _> = tx.send(RequestData::from(&*self));
+            tx.send(RequestData::from(&*self))
+                .inspect_err(|e| debug!("tx send failed: {e}"))
+                .ok();
        }
    }

@@ -371,7 +373,9 @@ impl RequestMonitoringInner {
        // Here we log the length of the session.
        self.disconnect_timestamp = Some(Utc::now());
        if let Some(tx) = self.disconnect_sender.take() {
-            let _: Result<(), _> = tx.send(RequestData::from(&*self));
+            tx.send(RequestData::from(&*self))
+                .inspect_err(|e| debug!("tx send failed: {e}"))
+                .ok();
        }
    }
 }
@@ -290,7 +290,7 @@ async fn worker_inner(
    }

    if !w.flushed_row_groups().is_empty() {
-        let _: Writer<BytesMut> = upload_parquet(w, len, &storage).await?;
+        let _rtchk: Writer<BytesMut> = upload_parquet(w, len, &storage).await?;
    }

    Ok(())
@@ -598,15 +598,15 @@ mod tests {
        assert_eq!(
            file_stats,
            [
-                (1315874, 3, 6000),
-                (1315867, 3, 6000),
-                (1315927, 3, 6000),
-                (1315884, 3, 6000),
-                (1316014, 3, 6000),
-                (1315856, 3, 6000),
-                (1315648, 3, 6000),
-                (1315884, 3, 6000),
-                (438913, 1, 2000)
+                (1312632, 3, 6000),
+                (1312621, 3, 6000),
+                (1312680, 3, 6000),
+                (1312637, 3, 6000),
+                (1312773, 3, 6000),
+                (1312610, 3, 6000),
+                (1312404, 3, 6000),
+                (1312639, 3, 6000),
+                (437848, 1, 2000)
            ]
        );

@@ -638,11 +638,11 @@ mod tests {
        assert_eq!(
            file_stats,
            [
-                (1208861, 5, 10000),
-                (1208592, 5, 10000),
-                (1208885, 5, 10000),
-                (1208873, 5, 10000),
-                (1209128, 5, 10000)
+                (1203465, 5, 10000),
+                (1203189, 5, 10000),
+                (1203490, 5, 10000),
+                (1203475, 5, 10000),
+                (1203729, 5, 10000)
            ]
        );

@@ -667,15 +667,15 @@ mod tests {
        assert_eq!(
            file_stats,
            [
-                (1315874, 3, 6000),
-                (1315867, 3, 6000),
-                (1315927, 3, 6000),
-                (1315884, 3, 6000),
-                (1316014, 3, 6000),
-                (1315856, 3, 6000),
-                (1315648, 3, 6000),
-                (1315884, 3, 6000),
-                (438913, 1, 2000)
+                (1312632, 3, 6000),
+                (1312621, 3, 6000),
+                (1312680, 3, 6000),
+                (1312637, 3, 6000),
+                (1312773, 3, 6000),
+                (1312610, 3, 6000),
+                (1312404, 3, 6000),
+                (1312639, 3, 6000),
+                (437848, 1, 2000)
            ]
        );

@@ -712,7 +712,7 @@ mod tests {
        // files are smaller than the size threshold, but they took too long to fill so were flushed early
        assert_eq!(
            file_stats,
-            [(659836, 2, 3001), (659550, 2, 3000), (659346, 2, 2999)]
+            [(657696, 2, 3001), (657410, 2, 3000), (657206, 2, 2999)]
        );

        tmpdir.close().unwrap();
@@ -3,7 +3,7 @@
 #![deny(
    deprecated,
    future_incompatible,
-    // TODO: consider let_underscore
+    let_underscore,
    nonstandard_style,
    rust_2024_compatibility
 )]
@@ -268,7 +268,7 @@ async fn keepalive_is_inherited() -> anyhow::Result<()> {
        anyhow::Ok(keepalive)
    });

-    let _ = TcpStream::connect(("127.0.0.1", port)).await?;
+    TcpStream::connect(("127.0.0.1", port)).await?;
    assert!(t.await??, "keepalive should be inherited");

    Ok(())
@@ -6,7 +6,7 @@ use redis::{
    ConnectionInfo, IntoConnectionInfo, RedisConnectionInfo, RedisResult,
 };
 use tokio::task::JoinHandle;
-use tracing::{error, info};
+use tracing::{debug, error, info};

 use super::elasticache::CredentialsProvider;

@@ -109,7 +109,10 @@ impl ConnectionWithCredentialsProvider {
            let credentials_provider = credentials_provider.clone();
            let con2 = con.clone();
            let f = tokio::spawn(async move {
-                let _ = Self::keep_connection(con2, credentials_provider).await;
+                Self::keep_connection(con2, credentials_provider)
+                    .await
+                    .inspect_err(|e| debug!("keep_connection failed: {e}"))
+                    .ok();
            });
            self.refresh_token_task = Some(f);
        }
@@ -12,6 +12,7 @@ use std::{io, task};
 use thiserror::Error;
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tokio_rustls::server::TlsStream;
+use tracing::debug;

 /// Stream wrapper which implements libpq's protocol.
 ///
@@ -138,9 +139,10 @@ impl<S: AsyncWrite + Unpin> PqStream<S> {
        );

        // already error case, ignore client IO error
-        let _: Result<_, std::io::Error> = self
-            .write_message(&BeMessage::ErrorResponse(msg, None))
-            .await;
+        self.write_message(&BeMessage::ErrorResponse(msg, None))
+            .await
+            .inspect_err(|e| debug!("write_message failed: {e}"))
+            .ok();

        Err(ReportedError {
            source: anyhow::anyhow!(msg),
@@ -164,9 +166,10 @@ impl<S: AsyncWrite + Unpin> PqStream<S> {
        );

        // already error case, ignore client IO error
-        let _: Result<_, std::io::Error> = self
-            .write_message(&BeMessage::ErrorResponse(&msg, None))
-            .await;
+        self.write_message(&BeMessage::ErrorResponse(&msg, None))
+            .await
+            .inspect_err(|e| debug!("write_message failed: {e}"))
+            .ok();

        Err(ReportedError {
            source: anyhow::anyhow!(error),
@@ -57,7 +57,7 @@ mod tests {
    fn bad_url() {
        let url = "test:foobar";
        url.parse::<url::Url>().expect("unexpected parsing failure");
-        let _ = url.parse::<ApiUrl>().expect_err("should not parse");
+        url.parse::<ApiUrl>().expect_err("should not parse");
    }

    #[test]
@@ -2,6 +2,7 @@
 //! protocol commands.

 use anyhow::Context;
+use std::future::Future;
 use std::str::{self, FromStr};
 use std::sync::Arc;
 use tokio::io::{AsyncRead, AsyncWrite};
@@ -95,7 +96,6 @@ fn cmd_to_string(cmd: &SafekeeperPostgresCommand) -> &str {
    }
 }

-#[async_trait::async_trait]
 impl<IO: AsyncRead + AsyncWrite + Unpin + Send> postgres_backend::Handler<IO>
    for SafekeeperPostgresHandler
 {
@@ -197,49 +197,51 @@ impl<IO: AsyncRead + AsyncWrite + Unpin + Send> postgres_backend::Handler<IO>
        Ok(())
    }

-    async fn process_query(
+    fn process_query(
        &mut self,
        pgb: &mut PostgresBackend<IO>,
        query_string: &str,
-    ) -> Result<(), QueryError> {
-        if query_string
-            .to_ascii_lowercase()
-            .starts_with("set datestyle to ")
-        {
-            // important for debug because psycopg2 executes "SET datestyle TO 'ISO'" on connect
-            pgb.write_message_noflush(&BeMessage::CommandComplete(b"SELECT 1"))?;
-            return Ok(());
-        }
-
-        let cmd = parse_cmd(query_string)?;
-        let cmd_str = cmd_to_string(&cmd);
-
-        let _guard = PG_QUERIES_GAUGE.with_label_values(&[cmd_str]).guard();
-
-        info!("got query {:?}", query_string);
-
-        let tenant_id = self.tenant_id.context("tenantid is required")?;
-        let timeline_id = self.timeline_id.context("timelineid is required")?;
-        self.check_permission(Some(tenant_id))?;
-        self.ttid = TenantTimelineId::new(tenant_id, timeline_id);
-
-        match cmd {
-            SafekeeperPostgresCommand::StartWalPush => {
-                self.handle_start_wal_push(pgb)
-                    .instrument(info_span!("WAL receiver"))
-                    .await
+    ) -> impl Future<Output = Result<(), QueryError>> {
+        Box::pin(async move {
+            if query_string
+                .to_ascii_lowercase()
+                .starts_with("set datestyle to ")
+            {
+                // important for debug because psycopg2 executes "SET datestyle TO 'ISO'" on connect
+                pgb.write_message_noflush(&BeMessage::CommandComplete(b"SELECT 1"))?;
+                return Ok(());
            }
-            SafekeeperPostgresCommand::StartReplication { start_lsn, term } => {
-                self.handle_start_replication(pgb, start_lsn, term)
-                    .instrument(info_span!("WAL sender"))
-                    .await
+
+            let cmd = parse_cmd(query_string)?;
+            let cmd_str = cmd_to_string(&cmd);
+
+            let _guard = PG_QUERIES_GAUGE.with_label_values(&[cmd_str]).guard();
+
+            info!("got query {:?}", query_string);
+
+            let tenant_id = self.tenant_id.context("tenantid is required")?;
+            let timeline_id = self.timeline_id.context("timelineid is required")?;
+            self.check_permission(Some(tenant_id))?;
+            self.ttid = TenantTimelineId::new(tenant_id, timeline_id);
+
+            match cmd {
+                SafekeeperPostgresCommand::StartWalPush => {
+                    self.handle_start_wal_push(pgb)
+                        .instrument(info_span!("WAL receiver"))
+                        .await
+                }
+                SafekeeperPostgresCommand::StartReplication { start_lsn, term } => {
+                    self.handle_start_replication(pgb, start_lsn, term)
+                        .instrument(info_span!("WAL sender"))
+                        .await
+                }
+                SafekeeperPostgresCommand::IdentifySystem => self.handle_identify_system(pgb).await,
+                SafekeeperPostgresCommand::TimelineStatus => self.handle_timeline_status(pgb).await,
+                SafekeeperPostgresCommand::JSONCtrl { ref cmd } => {
+                    handle_json_ctrl(self, pgb, cmd).await
+                }
            }
-            SafekeeperPostgresCommand::IdentifySystem => self.handle_identify_system(pgb).await,
-            SafekeeperPostgresCommand::TimelineStatus => self.handle_timeline_status(pgb).await,
-            SafekeeperPostgresCommand::JSONCtrl { ref cmd } => {
-                handle_json_ctrl(self, pgb, cmd).await
-            }
-        }
+        })
    }
 }

@@ -292,7 +292,7 @@ impl ComputeHook {
            );
            return Ok(());
        };
-        let env = match LocalEnv::load_config(repo_dir) {
+        let env = match LocalEnv::load_config(repo_dir, None) {
            Ok(e) => e,
            Err(e) => {
                tracing::warn!("Couldn't load neon_local config, skipping compute update ({e})");
@@ -18,8 +18,7 @@ Prerequisites:

 Regression tests are in the 'regress' directory. They can be run in
 parallel to minimize total runtime. Most regression test sets up their
-environment with its own pageservers and safekeepers (but see
-`TEST_SHARED_FIXTURES`).
+environment with its own pageservers and safekeepers.

 'pg_clients' contains tests for connecting with various client
 libraries. Each client test uses a Dockerfile that pulls an image that
@@ -74,7 +73,6 @@ This is used to construct full path to the postgres binaries.
 Format is 2-digit major version nubmer, i.e. `DEFAULT_PG_VERSION=16`
 `TEST_OUTPUT`: Set the directory where test state and test output files
 should go.
-`TEST_SHARED_FIXTURES`: Try to re-use a single pageserver for all the tests.
 `RUST_LOG`: logging configuration to pass into Neon CLI

 Useful parameters and commands:
@@ -259,11 +257,8 @@ compute Postgres nodes. The connections between them can be configured to use JW
 authentication tokens, and some other configuration options can be tweaked too.

 The easiest way to get access to a Neon Environment is by using the `neon_simple_env`
-fixture. The 'simple' env may be shared across multiple tests, so don't shut down the nodes
-or make other destructive changes in that environment. Also don't assume that
-there are no tenants or branches or data in the cluster. For convenience, there is a
-branch called `empty`, though. The convention is to create a test-specific branch of
-that and load any test data there, instead of the 'main' branch.
+fixture. For convenience, there is a branch called `main` in environments created with
+'neon_simple_env', ready to be used in the test.

 For more complicated cases, you can build a custom Neon Environment, with the `neon_env`
 fixture:
@@ -50,7 +50,7 @@ class NeonBroker:
                    raise RuntimeError(
                        f"timed out waiting {elapsed:.0f}s for storage_broker start: {e}"
                    ) from e
-                time.sleep(0.5)
+                time.sleep(0.1)
            else:
                break  # success

@@ -19,7 +19,7 @@ def pg_version() -> Optional[PgVersion]:
    return None


-@pytest.fixture(scope="function", autouse=True)
+@pytest.fixture(scope="session", autouse=True)
 def build_type() -> Optional[str]:
    return None

@@ -64,7 +64,7 @@ def pytest_generate_tests(metafunc: Metafunc):
    else:
        build_types = [bt.lower()]

-    metafunc.parametrize("build_type", build_types)
+    metafunc.parametrize("build_type", build_types, scope="session")

    if (v := os.getenv("DEFAULT_PG_VERSION")) is None:
        pg_versions = [version for version in PgVersion if version != PgVersion.NOT_SET]
@@ -22,10 +22,8 @@ if TYPE_CHECKING:
 def test_logical_replication(neon_simple_env: NeonEnv, pg_bin: PgBin, vanilla_pg):
    env = neon_simple_env

-    env.neon_cli.create_branch("test_logical_replication", "empty")
-    endpoint = env.endpoints.create_start("test_logical_replication")
+    endpoint = env.endpoints.create_start("main")

-    log.info("postgres is running on 'test_logical_replication' branch")
    pg_bin.run_capture(["pgbench", "-i", "-s10", endpoint.connstr()])

    endpoint.safe_psql("create publication pub1 for table pgbench_accounts, pgbench_history")
@@ -6,8 +6,8 @@ from fixtures.neon_fixtures import (
 )


-def test_aux_v2_config_switch(neon_env_builder: NeonEnvBuilder, vanilla_pg):
-    env = neon_env_builder.init_start()
+def test_aux_v2_config_switch(neon_shared_env: NeonEnv, vanilla_pg):
+    env = neon_shared_env
    endpoint = env.endpoints.create_start("main")
    client = env.pageserver.http_client()

@@ -8,11 +8,10 @@ from fixtures.neon_fixtures import NeonEnv
 #
 def test_basebackup_error(neon_simple_env: NeonEnv):
    env = neon_simple_env
-    env.neon_cli.create_branch("test_basebackup_error", "empty")
    pageserver_http = env.pageserver.http_client()

    # Introduce failpoint
    pageserver_http.configure_failpoints(("basebackup-before-control-file", "return"))

    with pytest.raises(Exception, match="basebackup-before-control-file"):
-        env.endpoints.create_start("test_basebackup_error")
+        env.endpoints.create_start("main")
@@ -46,11 +46,11 @@ from fixtures.utils import query_scalar
 # Because the delta layer D covering lsn1 is corrupted, creating a branch
 # starting from lsn1 should return an error as follows:
 #     could not find data for key ... at LSN ..., for request at LSN ...
-def test_branch_and_gc(neon_simple_env: NeonEnv, build_type: str):
+def test_branch_and_gc(neon_shared_env: NeonEnv, build_type: str):
    if build_type == "debug":
        pytest.skip("times out in debug builds")

-    env = neon_simple_env
+    env = neon_shared_env
    pageserver_http_client = env.pageserver.http_client()

    tenant, _ = env.neon_cli.create_tenant(
@@ -116,8 +116,8 @@ def test_branch_and_gc(neon_simple_env: NeonEnv, build_type: str):
 # and prevent creating branches with invalid starting LSNs.
 #
 # For more details, see discussion in https://github.com/neondatabase/neon/pull/2101#issuecomment-1185273447.
-def test_branch_creation_before_gc(neon_simple_env: NeonEnv):
-    env = neon_simple_env
+def test_branch_creation_before_gc(neon_shared_env: NeonEnv):
+    env = neon_shared_env
    pageserver_http_client = env.pageserver.http_client()

    error_regexes = [
@@ -33,7 +33,7 @@ from requests import RequestException
@pytest.mark.parametrize("scale", get_scales_matrix(1))
@pytest.mark.parametrize("ty", ["cascade", "flat"])
 def test_branching_with_pgbench(
-    neon_simple_env: NeonEnv, pg_bin: PgBin, n_branches: int, scale: int, ty: str
+    neon_shared_env: NeonEnv, pg_bin: PgBin, n_branches: int, scale: int, ty: str
 ):
    env = neon_simple_env

@@ -2,9 +2,8 @@ from fixtures.metrics import parse_metrics
 from fixtures.neon_fixtures import NeonEnvBuilder, NeonProxy


-def test_build_info_metric(neon_env_builder: NeonEnvBuilder, link_proxy: NeonProxy):
-    neon_env_builder.num_safekeepers = 1
-    env = neon_env_builder.init_start()
+def test_build_info_metric(neon_shared_env: NeonEnv, link_proxy: NeonProxy):
+    env = neon_shared_env

    parsed_metrics = {}

@@ -9,9 +9,8 @@ from fixtures.utils import query_scalar
 #
 # Test compute node start after clog truncation
 #
-def test_clog_truncate(neon_simple_env: NeonEnv):
-    env = neon_simple_env
-    env.neon_cli.create_branch("test_clog_truncate", "empty")
+def test_clog_truncate(neon_shared_env: NeonEnv):
+    env = neon_shared_env

    # set aggressive autovacuum to make sure that truncation will happen
    config = [
@@ -24,7 +23,7 @@ def test_clog_truncate(neon_simple_env: NeonEnv):
        "autovacuum_freeze_max_age=100000",
    ]

-    endpoint = env.endpoints.create_start("test_clog_truncate", config_lines=config)
+    endpoint = env.endpoints.create_start("main", config_lines=config)

    # Install extension containing function needed for test
    endpoint.safe_psql("CREATE EXTENSION neon_test_utils")
@@ -58,7 +57,7 @@ def test_clog_truncate(neon_simple_env: NeonEnv):
    # create new branch after clog truncation and start a compute node on it
    log.info(f"create branch at lsn_after_truncation {lsn_after_truncation}")
    env.neon_cli.create_branch(
-        "test_clog_truncate_new", "test_clog_truncate", ancestor_start_lsn=lsn_after_truncation
+        "test_clog_truncate_new", "main", ancestor_start_lsn=lsn_after_truncation
    )
    endpoint2 = env.endpoints.create_start("test_clog_truncate_new")

@@ -1,8 +1,8 @@
 from fixtures.neon_fixtures import NeonEnvBuilder, flush_ep_to_pageserver


-def do_combocid_op(neon_env_builder: NeonEnvBuilder, op):
-    env = neon_env_builder.init_start()
+def do_combocid_op(neon_shared_env: NeonEnv, op):
+    env = neon_shared_env
    endpoint = env.endpoints.create_start(
        "main",
        config_lines=[
@@ -49,20 +49,20 @@ def do_combocid_op(neon_env_builder: NeonEnvBuilder, op):
    )


-def test_combocid_delete(neon_env_builder: NeonEnvBuilder):
-    do_combocid_op(neon_env_builder, "delete from t")
+def test_combocid_delete(neon_shared_env: NeonEnv):
+    do_combocid_op(neon_env, "delete from t")


-def test_combocid_update(neon_env_builder: NeonEnvBuilder):
-    do_combocid_op(neon_env_builder, "update t set val=val+1")
+def test_combocid_update(neon_shared_env: NeonEnv):
+    do_combocid_op(neon_env, "update t set val=val+1")


-def test_combocid_lock(neon_env_builder: NeonEnvBuilder):
-    do_combocid_op(neon_env_builder, "select * from t for update")
+def test_combocid_lock(neon_shared_env: NeonEnv):
+    do_combocid_op(neon_env, "select * from t for update")


-def test_combocid_multi_insert(neon_env_builder: NeonEnvBuilder):
-    env = neon_env_builder.init_start()
+def test_combocid_multi_insert(neon_shared_env: NeonEnv):
+    env = neon_shared_env
    endpoint = env.endpoints.create_start(
        "main",
        config_lines=[
@@ -112,8 +112,8 @@ def test_combocid_multi_insert(neon_env_builder: NeonEnvBuilder):
    )


-def test_combocid(neon_env_builder: NeonEnvBuilder):
-    env = neon_env_builder.init_start()
+def test_combocid(neon_shared_env: NeonEnv):
+    env = neon_shared_env
    endpoint = env.endpoints.create_start("main")

    conn = endpoint.connect()
@@ -178,7 +178,7 @@ def test_backward_compatibility(
        neon_env_builder.num_safekeepers = 3
        env = neon_env_builder.from_repo_dir(compatibility_snapshot_dir / "repo")
        env.pageserver.allowed_errors.append(ingest_lag_log_line)
-        neon_env_builder.start()
+        env.start()

        check_neon_works(
            env,
@@ -265,7 +265,7 @@ def test_forward_compatibility(
        # does not include logs from previous runs
        assert not env.pageserver.log_contains("git-env:" + prev_pageserver_version)

-        neon_env_builder.start()
+        env.start()

        # ensure the specified pageserver is running
        assert env.pageserver.log_contains("git-env:" + prev_pageserver_version)
@@ -2,11 +2,10 @@ import requests
 from fixtures.neon_fixtures import NeonEnv


-def test_compute_catalog(neon_simple_env: NeonEnv):
-    env = neon_simple_env
-    env.neon_cli.create_branch("test_config", "empty")
+def test_compute_catalog(neon_shared_env: NeonEnv):
+    env = neon_shared_env

-    endpoint = env.endpoints.create_start("test_config", config_lines=["log_min_messages=debug1"])
+    endpoint = env.endpoints.create_start("main", config_lines=["log_min_messages=debug1"])
    client = endpoint.http_client()

    objects = client.dbs_and_roles()
@@ -9,10 +9,9 @@ from fixtures.neon_fixtures import NeonEnv, NeonEnvBuilder
 #
 def test_config(neon_simple_env: NeonEnv):
    env = neon_simple_env
-    env.neon_cli.create_branch("test_config", "empty")

    # change config
-    endpoint = env.endpoints.create_start("test_config", config_lines=["log_min_messages=debug1"])
+    endpoint = env.endpoints.create_start("main", config_lines=["log_min_messages=debug1"])

    with closing(endpoint.connect()) as conn:
        with conn.cursor() as cur:
@@ -30,6 +29,27 @@ def test_config(neon_simple_env: NeonEnv):
            # check that config change was applied
            assert cur.fetchone() == ("debug1",)

+def test_shared_config(neon_shared_env: NeonEnv):
+    env = neon_shared_env
+
+    # change config
+    endpoint = env.endpoints.create_start("main", config_lines=["log_min_messages=debug1"])
+
+    with closing(endpoint.connect()) as conn:
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                SELECT setting
+                FROM pg_settings
+                WHERE
+                    source != 'default'
+                    AND source != 'override'
+                    AND name = 'log_min_messages'
+            """
+            )
+
+            # check that config change was applied
+            assert cur.fetchone() == ("debug1",)

 #
 # Test that reordering of safekeepers does not restart walproposer
@@ -12,14 +12,12 @@ from fixtures.utils import query_scalar
 # Test CREATE DATABASE when there have been relmapper changes
 #
@pytest.mark.parametrize("strategy", ["file_copy", "wal_log"])
-def test_createdb(neon_simple_env: NeonEnv, strategy: str):
-    env = neon_simple_env
+def test_createdb(neon_shared_env: NeonEnv, strategy: str):
+    env = neon_shared_env
    if env.pg_version == PgVersion.V14 and strategy == "wal_log":
        pytest.skip("wal_log strategy not supported on PostgreSQL 14")

-    env.neon_cli.create_branch("test_createdb", "empty")
-
-    endpoint = env.endpoints.create_start("test_createdb")
+    endpoint = env.endpoints.create_start("main")

    with endpoint.cursor() as cur:
        # Cause a 'relmapper' change in the original branch
@@ -33,7 +31,7 @@ def test_createdb(neon_simple_env: NeonEnv, strategy: str):
        lsn = query_scalar(cur, "SELECT pg_current_wal_insert_lsn()")

    # Create a branch
-    env.neon_cli.create_branch("test_createdb2", "test_createdb", ancestor_start_lsn=lsn)
+    env.neon_cli.create_branch("test_createdb2", "main", ancestor_start_lsn=lsn)
    endpoint2 = env.endpoints.create_start("test_createdb2")

    # Test that you can connect to the new database on both branches
@@ -60,10 +58,9 @@ def test_createdb(neon_simple_env: NeonEnv, strategy: str):
 #
 # Test DROP DATABASE
 #
-def test_dropdb(neon_simple_env: NeonEnv, test_output_dir):
-    env = neon_simple_env
-    env.neon_cli.create_branch("test_dropdb", "empty")
-    endpoint = env.endpoints.create_start("test_dropdb")
+def test_dropdb(neon_shared_env: NeonEnv, test_output_dir):
+    env = neon_shared_env
+    endpoint = env.endpoints.create_start("main")

    with endpoint.cursor() as cur:
        cur.execute("CREATE DATABASE foodb")
@@ -80,14 +77,10 @@ def test_dropdb(neon_simple_env: NeonEnv, test_output_dir):
        lsn_after_drop = query_scalar(cur, "SELECT pg_current_wal_insert_lsn()")

    # Create two branches before and after database drop.
-    env.neon_cli.create_branch(
-        "test_before_dropdb", "test_dropdb", ancestor_start_lsn=lsn_before_drop
-    )
+    env.neon_cli.create_branch("test_before_dropdb", "main", ancestor_start_lsn=lsn_before_drop)
    endpoint_before = env.endpoints.create_start("test_before_dropdb")

-    env.neon_cli.create_branch(
-        "test_after_dropdb", "test_dropdb", ancestor_start_lsn=lsn_after_drop
-    )
+    env.neon_cli.create_branch("test_after_dropdb", "main", ancestor_start_lsn=lsn_after_drop)
    endpoint_after = env.endpoints.create_start("test_after_dropdb")

    # Test that database exists on the branch before drop
@@ -5,10 +5,9 @@ from fixtures.utils import query_scalar
 #
 # Test CREATE USER to check shared catalog restore
 #
-def test_createuser(neon_simple_env: NeonEnv):
-    env = neon_simple_env
-    env.neon_cli.create_branch("test_createuser", "empty")
-    endpoint = env.endpoints.create_start("test_createuser")
+def test_createuser(neon_shared_env: NeonEnv):
+    env = neon_shared_env
+    endpoint = env.endpoints.create_start("main")

    with endpoint.cursor() as cur:
        # Cause a 'relmapper' change in the original branch
@@ -19,7 +18,7 @@ def test_createuser(neon_simple_env: NeonEnv):
        lsn = query_scalar(cur, "SELECT pg_current_wal_insert_lsn()")

    # Create a branch
-    env.neon_cli.create_branch("test_createuser2", "test_createuser", ancestor_start_lsn=lsn)
+    env.neon_cli.create_branch("test_createuser2", "main", ancestor_start_lsn=lsn)
    endpoint2 = env.endpoints.create_start("test_createuser2")

    # Test that you can connect to new branch as a new user
@@ -290,9 +290,8 @@ def assert_db_connlimit(endpoint: Any, db_name: str, connlimit: int, msg: str):
 # Here we test the latter. The first one is tested in test_ddl_forwarding
 def test_ddl_forwarding_invalid_db(neon_simple_env: NeonEnv):
    env = neon_simple_env
-    env.neon_cli.create_branch("test_ddl_forwarding_invalid_db", "empty")
    endpoint = env.endpoints.create_start(
-        "test_ddl_forwarding_invalid_db",
+        "main",
        # Some non-existent url
        config_lines=["neon.console_url=http://localhost:9999/unknown/api/v0/roles_and_databases"],
    )
@@ -10,11 +10,11 @@ from fixtures.neon_fixtures import NeonEnvBuilder
        "💣",  # calls `trigger_segfault` internally
    ],
 )
-def test_endpoint_crash(neon_env_builder: NeonEnvBuilder, sql_func: str):
+def test_endpoint_crash(neon_shared_env: NeonEnv, sql_func: str):
    """
    Test that triggering crash from neon_test_utils crashes the endpoint
    """
-    env = neon_env_builder.init_start()
+    env = neon_shared_env
    env.neon_cli.create_branch("test_endpoint_crash")
    endpoint = env.endpoints.create_start("test_endpoint_crash")

@@ -10,11 +10,9 @@ def test_explain_with_lfc_stats(neon_simple_env: NeonEnv):
    cache_dir = Path(env.repo_dir) / "file_cache"
    cache_dir.mkdir(exist_ok=True)

-    branchname = "test_explain_with_lfc_stats"
-    env.neon_cli.create_branch(branchname, "empty")
-    log.info(f"Creating endopint with 1MB shared_buffers and 64 MB LFC for branch {branchname}")
+    log.info("Creating endpoint with 1MB shared_buffers and 64 MB LFC")
    endpoint = env.endpoints.create_start(
-        branchname,
+        "main",
        config_lines=[
            "shared_buffers='1MB'",
            f"neon.file_cache_path='{cache_dir}/file.cache'",
@@ -1,8 +1,8 @@
-from fixtures.neon_fixtures import NeonEnvBuilder
+from fixtures.neon_fixtures import NeonEnv


-def test_fsm_truncate(neon_env_builder: NeonEnvBuilder):
-    env = neon_env_builder.init_start()
+def test_fsm_truncate(neon_shared_env: NeonEnv):
+    env = neon_shared_env
    env.neon_cli.create_branch("test_fsm_truncate")
    endpoint = env.endpoints.create_start("test_fsm_truncate")
    endpoint.safe_psql(
@@ -16,12 +16,12 @@ num_rows = 1000

 # Ensure that regular postgres can start from fullbackup
 def test_fullbackup(
-    neon_env_builder: NeonEnvBuilder,
+    neon_shared_env: NeonEnv,
    pg_bin: PgBin,
    port_distributor: PortDistributor,
    test_output_dir: Path,
 ):
-    env = neon_env_builder.init_start()
+    env = neon_shared_env

    # endpoint needs to be alive until the fullbackup so that we have
    # prev_record_lsn for the vanilla_pg to start in read-write mode
@@ -6,8 +6,8 @@ from fixtures.neon_fixtures import NeonEnv, wait_replica_caughtup
 #
 # Test that redo of XLOG_GIN_VACUUM_PAGE doesn't produce error
 #
-def test_gin_redo(neon_simple_env: NeonEnv):
-    env = neon_simple_env
+def test_gin_redo(neon_shared_env: NeonEnv):
+    env = neon_shared_env

    primary = env.endpoints.create_start(branch_name="main", endpoint_id="primary")
    time.sleep(1)
@@ -87,8 +87,8 @@ def test_hot_standby(neon_simple_env: NeonEnv):
        sk_http.configure_failpoints(("sk-send-wal-replica-sleep", "off"))


-def test_2_replicas_start(neon_simple_env: NeonEnv):
-    env = neon_simple_env
+def test_2_replicas_start(neon_shared_env: NeonEnv):
+    env = neon_shared_env

    with env.endpoints.create_start(
        branch_name="main",
@@ -286,8 +286,8 @@ def test_hot_standby_feedback(neon_env_builder: NeonEnvBuilder, pg_bin: PgBin):

 # Test race condition between WAL replay and backends performing queries
 # https://github.com/neondatabase/neon/issues/7791
-def test_replica_query_race(neon_simple_env: NeonEnv):
-    env = neon_simple_env
+def test_replica_query_race(neon_shared_env: NeonEnv):
+    env = neon_shared_env

    primary_ep = env.endpoints.create_start(
        branch_name="main",
@@ -12,8 +12,8 @@ from fixtures.neon_fixtures import NeonEnvBuilder
 # to large (several gigabytes) layer files (both ephemeral and delta layers).
 # It may cause problems with uploading to S3 and also degrade performance because ephemeral file swapping.
 #
-def test_large_schema(neon_env_builder: NeonEnvBuilder):
-    env = neon_env_builder.init_start()
+def test_large_schema(neon_shared_env: NeonEnv):
+    env = neon_shared_env

    endpoint = env.endpoints.create_start("main")

@@ -14,11 +14,10 @@ from fixtures.neon_fixtures import NeonEnv, PgBin
 # Test branching, when a transaction is in prepared state
 #
@pytest.mark.timeout(600)
-def test_lfc_resize(neon_simple_env: NeonEnv, pg_bin: PgBin):
-    env = neon_simple_env
-    env.neon_cli.create_branch("test_lfc_resize", "empty")
+def test_lfc_resize(neon_shared_env: NeonEnv, pg_bin: PgBin):
+    env = neon_shared_env
    endpoint = env.endpoints.create_start(
-        "test_lfc_resize",
+        "main",
        config_lines=[
            "neon.file_cache_path='file.cache'",
            "neon.max_file_cache_size=512MB",
@@ -12,11 +12,9 @@ def test_lfc_working_set_approximation(neon_simple_env: NeonEnv):
    cache_dir = Path(env.repo_dir) / "file_cache"
    cache_dir.mkdir(exist_ok=True)

-    branchname = "test_approximate_working_set_size"
-    env.neon_cli.create_branch(branchname, "empty")
-    log.info(f"Creating endopint with 1MB shared_buffers and 64 MB LFC for branch {branchname}")
+    log.info("Creating endpoint with 1MB shared_buffers and 64 MB LFC")
    endpoint = env.endpoints.create_start(
-        branchname,
+        "main",
        config_lines=[
            "shared_buffers='1MB'",
            f"neon.file_cache_path='{cache_dir}/file.cache'",
@@ -5,21 +5,18 @@ import threading
 import time
 from typing import List

-from fixtures.neon_fixtures import DEFAULT_BRANCH_NAME, NeonEnvBuilder
+from fixtures.neon_fixtures import NeonEnvBuilder
 from fixtures.utils import query_scalar


-def test_local_file_cache_unlink(neon_env_builder: NeonEnvBuilder):
-    env = neon_env_builder.init_start()
+def test_local_file_cache_unlink(neon_shared_env: NeonEnv):
+    env = neon_shared_env

    cache_dir = os.path.join(env.repo_dir, "file_cache")
    os.mkdir(cache_dir)

-    env.neon_cli.create_branch("empty", ancestor_branch_name=DEFAULT_BRANCH_NAME)
-    env.neon_cli.create_branch("test_local_file_cache_unlink", "empty")
-
    endpoint = env.endpoints.create_start(
-        "test_local_file_cache_unlink",
+        "main",
        config_lines=[
            "shared_buffers='1MB'",
            f"neon.file_cache_path='{cache_dir}/file.cache'",
@@ -36,10 +36,8 @@ def test_logical_replication(neon_simple_env: NeonEnv, vanilla_pg):
    env = neon_simple_env

    tenant_id = env.initial_tenant
-    timeline_id = env.neon_cli.create_branch("test_logical_replication", "empty")
-    endpoint = env.endpoints.create_start(
-        "test_logical_replication", config_lines=["log_statement=all"]
-    )
+    timeline_id = env.initial_timeline
+    endpoint = env.endpoints.create_start("main", config_lines=["log_statement=all"])

    pg_conn = endpoint.connect()
    cur = pg_conn.cursor()
@@ -185,10 +183,9 @@ def test_obsolete_slot_drop(neon_simple_env: NeonEnv, vanilla_pg):

    env = neon_simple_env

-    env.neon_cli.create_branch("test_logical_replication", "empty")
    # set low neon.logical_replication_max_snap_files
    endpoint = env.endpoints.create_start(
-        "test_logical_replication",
+        "main",
        config_lines=["log_statement=all", "neon.logical_replication_max_snap_files=1"],
    )

@@ -472,7 +469,7 @@ def test_slots_and_branching(neon_simple_env: NeonEnv):
 def test_replication_shutdown(neon_simple_env: NeonEnv):
    # Ensure Postgres can exit without stuck when a replication job is active + neon extension installed
    env = neon_simple_env
-    env.neon_cli.create_branch("test_replication_shutdown_publisher", "empty")
+    env.neon_cli.create_branch("test_replication_shutdown_publisher", "main")
    pub = env.endpoints.create("test_replication_shutdown_publisher")

    env.neon_cli.create_branch("test_replication_shutdown_subscriber")
@@ -24,13 +24,13 @@ def assert_lsn_lease_granted(result, with_lease: bool):


@pytest.mark.parametrize("with_lease", [True, False])
-def test_lsn_mapping(neon_env_builder: NeonEnvBuilder, with_lease: bool):
+def test_lsn_mapping(neon_shared_env: NeonEnv, with_lease: bool):
    """
    Test pageserver get_lsn_by_timestamp API.

    :param with_lease: Whether to get a lease associated with returned LSN.
    """
-    env = neon_env_builder.init_start()
+    env = neon_shared_env

    tenant_id, _ = env.neon_cli.create_tenant(
        conf={
@@ -7,11 +7,10 @@ if TYPE_CHECKING:
    from fixtures.neon_fixtures import NeonEnv


-def test_migrations(neon_simple_env: NeonEnv):
-    env = neon_simple_env
-    env.neon_cli.create_branch("test_migrations", "empty")
+def test_migrations(neon_shared_env: NeonEnv):
+    env = neon_shared_env

-    endpoint = env.endpoints.create("test_migrations")
+    endpoint = env.endpoints.create("main")
    endpoint.respec(skip_pg_catalog_updates=False)
    endpoint.start()

@@ -12,10 +12,9 @@ from fixtures.utils import query_scalar
 # is enough to verify that the WAL records are handled correctly
 # in the pageserver.
 #
-def test_multixact(neon_simple_env: NeonEnv, test_output_dir):
-    env = neon_simple_env
-    env.neon_cli.create_branch("test_multixact", "empty")
-    endpoint = env.endpoints.create_start("test_multixact")
+def test_multixact(neon_shared_env: NeonEnv, test_output_dir):
+    env = neon_shared_env
+    endpoint = env.endpoints.create_start("main")

    cur = endpoint.connect().cursor()
    cur.execute(
@@ -73,7 +72,9 @@ def test_multixact(neon_simple_env: NeonEnv, test_output_dir):
    assert int(next_multixact_id) > int(next_multixact_id_old)

    # Branch at this point
-    env.neon_cli.create_branch("test_multixact_new", "test_multixact", ancestor_start_lsn=lsn)
+    env.neon_cli.create_branch(
+        "test_multixact_new", ancestor_branch_name="main", ancestor_start_lsn=lsn
+    )
    endpoint_new = env.endpoints.create_start("test_multixact_new")

    next_multixact_id_new = endpoint_new.safe_psql(
@@ -4,9 +4,9 @@ from fixtures.pg_version import PgVersion
 from fixtures.utils import wait_until


-def test_neon_superuser(neon_simple_env: NeonEnv, pg_version: PgVersion):
-    env = neon_simple_env
-    env.neon_cli.create_branch("test_neon_superuser_publisher", "empty")
+def test_neon_superuser(neon_shared_env: NeonEnv, pg_version: PgVersion):
+    env = neon_shared_env
+    env.neon_cli.create_branch("test_neon_superuser_publisher", "main")
    pub = env.endpoints.create("test_neon_superuser_publisher")

    env.neon_cli.create_branch("test_neon_superuser_subscriber")
@@ -2,8 +2,8 @@ from fixtures.log_helper import log
 from fixtures.neon_fixtures import NeonEnvBuilder


-def test_oid_overflow(neon_env_builder: NeonEnvBuilder):
-    env = neon_env_builder.init_start()
+def test_oid_overflow(neon_shared_env: NeonEnv):
+    env = neon_shared_env

    endpoint = env.endpoints.create_start("main")

@@ -56,8 +56,8 @@ def check_client(env: NeonEnv, client: PageserverHttpClient):
        assert TimelineId(timeline_details["timeline_id"]) == timeline_id


-def test_pageserver_http_get_wal_receiver_not_found(neon_simple_env: NeonEnv):
-    env = neon_simple_env
+def test_pageserver_http_get_wal_receiver_not_found(neon_shared_env: NeonEnv):
+    env = neon_shared_env
    with env.pageserver.http_client() as client:
        tenant_id, timeline_id = env.neon_cli.create_tenant()

@@ -105,8 +105,8 @@ def expect_updated_msg_lsn(
 #
 # These fields used to be returned by a separate API call, but they're part of
 # `timeline_details` now.
-def test_pageserver_http_get_wal_receiver_success(neon_simple_env: NeonEnv):
-    env = neon_simple_env
+def test_pageserver_http_get_wal_receiver_success(neon_shared_env: NeonEnv):
+    env = neon_shared_env
    with env.pageserver.http_client() as client:
        tenant_id, timeline_id = env.neon_cli.create_tenant()
        endpoint = env.endpoints.create_start(DEFAULT_BRANCH_NAME, tenant_id=tenant_id)
@@ -39,10 +39,9 @@ async def parallel_load_same_table(endpoint: Endpoint, n_parallel: int):


 # Load data into one table with COPY TO from 5 parallel connections
-def test_parallel_copy(neon_simple_env: NeonEnv, n_parallel=5):
-    env = neon_simple_env
-    env.neon_cli.create_branch("test_parallel_copy", "empty")
-    endpoint = env.endpoints.create_start("test_parallel_copy")
+def test_parallel_copy(neon_shared_env: NeonEnv, n_parallel=5):
+    env = neon_shared_env
+    endpoint = env.endpoints.create_start("main")

    # Create test table
    conn = endpoint.connect()
@@ -42,11 +42,9 @@ def test_cancellations(neon_simple_env: NeonEnv):
    ps_http = ps.http_client()
    ps_http.is_testing_enabled_or_skip()

-    env.neon_cli.create_branch("test_config", "empty")
-
    # We don't want to have any racy behaviour with autovacuum IOs
    ep = env.endpoints.create_start(
-        "test_config",
+        "main",
        config_lines=[
            "autovacuum = off",
            "shared_buffers = 128MB",
@@ -19,11 +19,11 @@ def check_wal_segment(pg_waldump_path: str, segment_path: str, test_output_dir):


 # Simple test to check that pg_waldump works with neon WAL files
-def test_pg_waldump(neon_simple_env: NeonEnv, test_output_dir, pg_bin: PgBin):
-    env = neon_simple_env
+def test_pg_waldump(neon_shared_env: NeonEnv, test_output_dir, pg_bin: PgBin):
+    env = neon_shared_env
    tenant_id = env.initial_tenant
-    timeline_id = env.neon_cli.create_branch("test_pg_waldump", "empty")
-    endpoint = env.endpoints.create_start("test_pg_waldump")
+    timeline_id = env.initial_timeline
+    endpoint = env.endpoints.create_start("main")

    cur = endpoint.connect().cursor()
    cur.execute(
@@ -13,14 +13,10 @@ extensions = ["pageinspect", "neon_test_utils", "pg_buffercache"]
 #
 # Validation of reading different page versions
 #
-def test_read_validation(neon_simple_env: NeonEnv):
-    env = neon_simple_env
-    env.neon_cli.create_branch("test_read_validation", "empty")
-
-    endpoint = env.endpoints.create_start(
-        "test_read_validation",
-    )
+def test_read_validation(neon_shared_env: NeonEnv):
+    env = neon_shared_env

+    endpoint = env.endpoints.create_start("main")
    with closing(endpoint.connect()) as con:
        with con.cursor() as c:
            for e in extensions:
@@ -129,15 +125,11 @@ def test_read_validation(neon_simple_env: NeonEnv):
                log.info(f"Caught an expected failure: {e}")


-def test_read_validation_neg(neon_simple_env: NeonEnv):
-    env = neon_simple_env
-    env.neon_cli.create_branch("test_read_validation_neg", "empty")
-
+def test_read_validation_neg(neon_shared_env: NeonEnv):
+    env = neon_shared_env
    env.pageserver.allowed_errors.append(".*invalid LSN\\(0\\) in request.*")

-    endpoint = env.endpoints.create_start(
-        "test_read_validation_neg",
-    )
+    endpoint = env.endpoints.create_start("main")

    with closing(endpoint.connect()) as con:
        with con.cursor() as c:
@@ -22,8 +22,7 @@ from fixtures.utils import query_scalar
 #
 def test_readonly_node(neon_simple_env: NeonEnv):
    env = neon_simple_env
-    env.neon_cli.create_branch("test_readonly_node", "empty")
-    endpoint_main = env.endpoints.create_start("test_readonly_node")
+    endpoint_main = env.endpoints.create_start("main")

    env.pageserver.allowed_errors.extend(
        [
@@ -74,12 +73,12 @@ def test_readonly_node(neon_simple_env: NeonEnv):

    # Create first read-only node at the point where only 100 rows were inserted
    endpoint_hundred = env.endpoints.create_start(
-        branch_name="test_readonly_node", endpoint_id="ep-readonly_node_hundred", lsn=lsn_a
+        branch_name="main", endpoint_id="ep-readonly_node_hundred", lsn=lsn_a
    )

    # And another at the point where 200100 rows were inserted
    endpoint_more = env.endpoints.create_start(
-        branch_name="test_readonly_node", endpoint_id="ep-readonly_node_more", lsn=lsn_b
+        branch_name="main", endpoint_id="ep-readonly_node_more", lsn=lsn_b
    )

    # On the 'hundred' node, we should see only 100 rows
@@ -100,7 +99,7 @@ def test_readonly_node(neon_simple_env: NeonEnv):

    # Check creating a node at segment boundary
    endpoint = env.endpoints.create_start(
-        branch_name="test_readonly_node",
+        branch_name="main",
        endpoint_id="ep-branch_segment_boundary",
        lsn=Lsn("0/3000000"),
    )
@@ -112,7 +111,7 @@ def test_readonly_node(neon_simple_env: NeonEnv):
    with pytest.raises(Exception, match="invalid basebackup lsn"):
        # compute node startup with invalid LSN should fail
        env.endpoints.create_start(
-            branch_name="test_readonly_node",
+            branch_name="main",
            endpoint_id="ep-readonly_node_preinitdb",
            lsn=Lsn("0/42"),
        )
@@ -216,16 +215,12 @@ def test_readonly_node_gc(neon_env_builder: NeonEnvBuilder):


 # Similar test, but with more data, and we force checkpoints
-def test_timetravel(neon_simple_env: NeonEnv):
-    env = neon_simple_env
-    pageserver_http_client = env.pageserver.http_client()
-    env.neon_cli.create_branch("test_timetravel", "empty")
-    endpoint = env.endpoints.create_start("test_timetravel")
-
+def test_timetravel(neon_shared_env: NeonEnv):
+    env = neon_shared_env
+    tenant_id = env.initial_tenant
+    timeline_id = env.initial_timeline
    client = env.pageserver.http_client()
-
-    tenant_id = endpoint.safe_psql("show neon.tenant_id")[0][0]
-    timeline_id = endpoint.safe_psql("show neon.timeline_id")[0][0]
+    endpoint = env.endpoints.create_start("main")

    lsns = []

@@ -249,7 +244,7 @@ def test_timetravel(neon_simple_env: NeonEnv):
        wait_for_last_record_lsn(client, tenant_id, timeline_id, current_lsn)

        # run checkpoint manually to force a new layer file
-        pageserver_http_client.timeline_checkpoint(tenant_id, timeline_id)
+        client.timeline_checkpoint(tenant_id, timeline_id)

    ##### Restart pageserver
    env.endpoints.stop_all()
@@ -258,7 +253,7 @@ def test_timetravel(neon_simple_env: NeonEnv):

    for i, lsn in lsns:
        endpoint_old = env.endpoints.create_start(
-            branch_name="test_timetravel", endpoint_id=f"ep-old_lsn_{i}", lsn=lsn
+            branch_name="main", endpoint_id=f"ep-old_lsn_{i}", lsn=lsn
        )
        with endpoint_old.cursor() as cur:
            assert query_scalar(cur, f"select count(*) from testtab where iteration={i}") == 100000
@@ -374,7 +374,7 @@ def test_sharding_split_smoke(
    non_default_tenant_config = {"gc_horizon": 77 * 1024 * 1024}

    env = neon_env_builder.init_configs(True)
-    neon_env_builder.start()
+    env.start()
    tenant_id = TenantId.generate()
    timeline_id = TimelineId.generate()
    env.neon_cli.create_tenant(
@@ -1436,7 +1436,7 @@ def test_sharding_unlogged_relation(neon_env_builder: NeonEnvBuilder):

    neon_env_builder.num_pageservers = 2
    env = neon_env_builder.init_configs()
-    neon_env_builder.start()
+    env.start()

    tenant_id = TenantId.generate()
    timeline_id = TimelineId.generate()
@@ -1475,7 +1475,7 @@ def test_top_tenants(neon_env_builder: NeonEnvBuilder):
    """

    env = neon_env_builder.init_configs()
-    neon_env_builder.start()
+    env.start()

    tenants = []
    n_tenants = 8
@@ -7,8 +7,8 @@ from fixtures.utils import wait_until

 # This test checks of logical replication subscriber is able to correctly restart replication without receiving duplicates.
 # It requires tracking information about replication origins at page server side
-def test_subscriber_restart(neon_simple_env: NeonEnv):
-    env = neon_simple_env
+def test_subscriber_restart(neon_shared_env: NeonEnv):
+    env = neon_shared_env
    env.neon_cli.create_branch("publisher")
    pub = env.endpoints.create("publisher")
    pub.start()
@@ -7,10 +7,9 @@ from fixtures.neon_fixtures import NeonEnv, check_restored_datadir_content
 # maintained in the pageserver, so subtransactions are not very exciting for
 # Neon. They are included in the commit record though and updated in the
 # CLOG.
-def test_subxacts(neon_simple_env: NeonEnv, test_output_dir):
-    env = neon_simple_env
-    env.neon_cli.create_branch("test_subxacts", "empty")
-    endpoint = env.endpoints.create_start("test_subxacts")
+def test_subxacts(neon_shared_env: NeonEnv, test_output_dir):
+    env = neon_shared_env
+    endpoint = env.endpoints.create_start("main")

    pg_conn = endpoint.connect()
    cur = pg_conn.cursor()
@@ -36,8 +36,8 @@ from fixtures.utils import query_scalar, run_pg_bench_small, wait_until
 from urllib3.util.retry import Retry


-def test_timeline_delete(neon_simple_env: NeonEnv):
-    env = neon_simple_env
+def test_timeline_delete(neon_shared_env: NeonEnv):
+    env = neon_shared_env

    env.pageserver.allowed_errors.extend(
        [
@@ -68,10 +68,13 @@ def test_timeline_delete(neon_simple_env: NeonEnv):

    # construct pair of branches to validate that pageserver prohibits
    # deletion of ancestor timelines when they have child branches
-    parent_timeline_id = env.neon_cli.create_branch("test_ancestor_branch_delete_parent", "empty")
+    parent_timeline_id = env.neon_cli.create_branch(
+        new_branch_name="test_ancestor_branch_delete_parent", ancestor_branch_name="main"
+    )

    leaf_timeline_id = env.neon_cli.create_branch(
-        "test_ancestor_branch_delete_branch1", "test_ancestor_branch_delete_parent"
+        new_branch_name="test_ancestor_branch_delete_branch1",
+        ancestor_branch_name="test_ancestor_branch_delete_parent",
    )

    timeline_path = env.pageserver.timeline_dir(env.initial_tenant, parent_timeline_id)
@@ -36,7 +36,7 @@ from fixtures.utils import get_timeline_dir_size, wait_until

 def test_timeline_size(neon_simple_env: NeonEnv):
    env = neon_simple_env
-    new_timeline_id = env.neon_cli.create_branch("test_timeline_size", "empty")
+    new_timeline_id = env.neon_cli.create_branch("test_timeline_size", "main")

    client = env.pageserver.http_client()
    client.timeline_wait_logical_size(env.initial_tenant, new_timeline_id)
@@ -68,7 +68,7 @@ def test_timeline_size(neon_simple_env: NeonEnv):

 def test_timeline_size_createdropdb(neon_simple_env: NeonEnv):
    env = neon_simple_env
-    new_timeline_id = env.neon_cli.create_branch("test_timeline_size_createdropdb", "empty")
+    new_timeline_id = env.neon_cli.create_branch("test_timeline_size_createdropdb", "main")

    client = env.pageserver.http_client()
    client.timeline_wait_logical_size(env.initial_tenant, new_timeline_id)
@@ -1,13 +1,13 @@
 import time

-from fixtures.neon_fixtures import NeonEnvBuilder
+from fixtures.neon_fixtures import NeonEnv


 #
 # Test truncation of FSM and VM forks of a relation
 #
-def test_truncate(neon_env_builder: NeonEnvBuilder, zenbenchmark):
-    env = neon_env_builder.init_start()
+def test_truncate(neon_shared_env: NeonEnv, zenbenchmark):
+    env = neon_shared_env
    n_records = 10000
    n_iter = 10

@@ -7,12 +7,9 @@ from fixtures.neon_fixtures import NeonEnv, fork_at_current_lsn
 #
 # Test branching, when a transaction is in prepared state
 #
-def test_twophase(neon_simple_env: NeonEnv):
-    env = neon_simple_env
-    env.neon_cli.create_branch("test_twophase", "empty")
-    endpoint = env.endpoints.create_start(
-        "test_twophase", config_lines=["max_prepared_transactions=5"]
-    )
+def test_twophase(neon_shared_env: NeonEnv):
+    env = neon_shared_env
+    endpoint = env.endpoints.create_start("main", config_lines=["max_prepared_transactions=5"])

    conn = endpoint.connect()
    cur = conn.cursor()
@@ -56,7 +53,7 @@ def test_twophase(neon_simple_env: NeonEnv):
    assert len(twophase_files) == 2

    # Create a branch with the transaction in prepared state
-    fork_at_current_lsn(env, endpoint, "test_twophase_prepared", "test_twophase")
+    fork_at_current_lsn(env, endpoint, "test_twophase_prepared", "main")

    # Start compute on the new branch
    endpoint2 = env.endpoints.create_start(
@@ -7,10 +7,9 @@ from fixtures.pg_version import PgVersion
 # fork to reset them during recovery. In Neon, pageserver directly sends init
 # fork contents as main fork during basebackup.
 #
-def test_unlogged(neon_simple_env: NeonEnv):
-    env = neon_simple_env
-    env.neon_cli.create_branch("test_unlogged", "empty")
-    endpoint = env.endpoints.create_start("test_unlogged")
+def test_unlogged(neon_shared_env: NeonEnv):
+    env = neon_shared_env
+    endpoint = env.endpoints.create_start("main")

    conn = endpoint.connect()
    cur = conn.cursor()
@@ -22,7 +21,7 @@ def test_unlogged(neon_simple_env: NeonEnv):
    cur.execute("INSERT INTO iut (id) values (42);")

    # create another compute to fetch inital empty contents from pageserver
-    fork_at_current_lsn(env, endpoint, "test_unlogged_basebackup", "test_unlogged")
+    fork_at_current_lsn(env, endpoint, "test_unlogged_basebackup", "main")
    endpoint2 = env.endpoints.create_start("test_unlogged_basebackup")

    conn2 = endpoint2.connect()
@@ -10,11 +10,10 @@ from fixtures.utils import query_scalar
 # Test that the VM bit is cleared correctly at a HEAP_DELETE and
 # HEAP_UPDATE record.
 #
-def test_vm_bit_clear(neon_simple_env: NeonEnv):
-    env = neon_simple_env
+def test_vm_bit_clear(neon_shared_env: NeonEnv):
+    env = neon_shared_env

-    env.neon_cli.create_branch("test_vm_bit_clear", "empty")
-    endpoint = env.endpoints.create_start("test_vm_bit_clear")
+    endpoint = env.endpoints.create_start("main")

    pg_conn = endpoint.connect()
    cur = pg_conn.cursor()
@@ -58,7 +57,7 @@ def test_vm_bit_clear(neon_simple_env: NeonEnv):
    cur.execute("UPDATE vmtest_cold_update2 SET id = 5000, filler=repeat('x', 200) WHERE id = 1")

    # Branch at this point, to test that later
-    fork_at_current_lsn(env, endpoint, "test_vm_bit_clear_new", "test_vm_bit_clear")
+    fork_at_current_lsn(env, endpoint, "test_vm_bit_clear_new", "main")

    # Clear the buffer cache, to force the VM page to be re-fetched from
    # the page server
@@ -115,13 +114,13 @@ def test_vm_bit_clear(neon_simple_env: NeonEnv):
    assert cur_new.fetchall() == []


-def test_vm_bit_clear_on_heap_lock_whitebox(neon_env_builder: NeonEnvBuilder):
+def test_vm_bit_clear_on_heap_lock_whitebox(neon_shared_env: NeonEnv):
    """
    Test that the ALL_FROZEN VM bit is cleared correctly at a HEAP_LOCK record.

    This is a repro for the bug fixed in commit 66fa176cc8.
    """
-    env = neon_env_builder.init_start()
+    env = neon_shared_env
    endpoint = env.endpoints.create_start(
        "main",
        config_lines=[
@@ -23,8 +23,7 @@ run_broken = pytest.mark.skipif(
 def test_broken(neon_simple_env: NeonEnv, pg_bin):
    env = neon_simple_env

-    env.neon_cli.create_branch("test_broken", "empty")
-    env.endpoints.create_start("test_broken")
+    env.endpoints.create_start("main")
    log.info("postgres is running")

    log.info("THIS NEXT COMMAND WILL FAIL:")
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
Heikki Linnakangas	87f270ee7a	wip	2024-09-12 01:36:11 +03:00
Heikki Linnakangas	bfae8ff23e	WIP	2024-09-12 01:20:59 +03:00
Heikki Linnakangas	a6e64fcc90	prefix endpoint ids with test name	2024-09-12 01:20:59 +03:00
Heikki Linnakangas	390235e095	wip add NeonEnv	2024-09-12 01:20:59 +03:00
Heikki Linnakangas	7565939dce	Store branch name mappings in separate branches.toml file	2024-09-12 01:20:59 +03:00
Heikki Linnakangas	cde4ea2b39	Add --timeline-id option to "neon_local timeline branch" command Makes it consistent with the "timeline create" and "timeline import" commands, which allowed you to pass the timeline id as argument. This also makes it unnecessary to parse the timeline ID from the output in the python function that calls it.	2024-09-12 01:20:59 +03:00
Heikki Linnakangas	60320e8e45	Rename test_name to env_name	2024-09-12 01:20:59 +03:00
Heikki Linnakangas	fe1fe213ba	Let NeonEnv handle starting and stopping NeonBroker	2024-09-12 01:20:59 +03:00
Heikki Linnakangas	67d48ac867	Remove silly NeonEnvBuilder.start function	2024-09-12 01:20:59 +03:00
Heikki Linnakangas	4da40cbbfb	initdb cache	2024-09-12 01:20:59 +03:00
Heikki Linnakangas	8e7f336540	Cache initdb output to speed up tenant creation in tests initdb takes about 1 s. Our tests create and destroy a lot of tenants, so that adds up. Cache the initdb result to speed it up. This is currently only enabled in tests. Out of caution, mostly. But also because when you reuse the initdb result, all the postgres clusters end up having the same system_identifier, which is supposed to be unique. It's not necessary for it to be unique for correctness, nothing critical relies on it and you can easily end up with duplicate system_identifiers in standalone PostgreSQL too, if you e.g. create a backup and restore it on a different system. But it is used in various checks to reduce the chance that you e.g. accidentally apply WAL belonging to a different cluster. Because this is aimed at tests, there are a few things that might be surprising: - The initdb cache directory is configurable, and can be outside the pageserver's repo directory. This allows reuse across different pageservers running on the same host. In production use, that'd be pointless, but our tests create a lot of pageservers. - The cache is not automatically purged at start / shutdown. For production use, we'd probably want that, so that we'd pick up any changes in what an empty cluster looks like after a Postgres minor version upgrade, for example. But again tests create and destroy a lot of pageservers, so it's important to retain the cache. - The locking on the cache directory relies purely on filesystem operations and atomic rename(). Using e.g. a rust Mutex() would be more straightforward, but that's not enough because the cache needs to be shared between different pageservers running on the same system.	2024-09-12 01:20:54 +03:00
John Spray	cb060548fb	libs: tweak PageserverUtilization::is_overloaded (#8946 ) ## Problem Having run in production for a while, we see that nodes are generally safely oversubscribed by about a factor of 2. ## Summary of changes Tweak the is_overloaded method to check for utililzation over 200% rather than over 100%	2024-09-11 18:45:34 +01:00
Folke Behrens	bae793ffcd	proxy: Handle all let underscore instances (#8898 ) * Most can be simply replaced * One instance renamed to _rtchk (return-type check)	2024-09-10 15:36:08 +02:00
John Spray	26b5fcdc50	reinstate write-path key check (#8973 ) ## Problem In https://github.com/neondatabase/neon/pull/8621, validation of keys during ingest was removed because the places where we actually store keys are now past the point where we have already converted them to CompactKey (i128) representation. ## Summary of changes Reinstate validation at an earlier stage in ingest. This doesn't cover literally every place we write a key, but it covers most cases where we're trusting postgres to give us a valid key (i.e. one that doesn't try and use a custom spacenode).	2024-09-10 12:54:25 +01:00
Arpad Müller	97582178cb	Remove async_trait from the Handler trait (#8958 ) Newest attempt to remove `async_trait` from the Handler trait. Earlier attempts were in #7301 and #8296 .	2024-09-10 02:40:00 +02:00
Matthias van de Meent	842be0ba74	Specialize WalIngest on PostgreSQL version (#8904 ) The current code assumes that most of this functionality is version-independent, which is only true up to v16 - PostgreSQL 17 has a new field in CheckPoint that we need to keep track of. This basically removes the file-level dependency on v14, and replaces it with switches that load the correct version dependencies where required.	2024-09-09 23:01:52 +01:00
Heikki Linnakangas	982b376ea2	Update parquet crate to a released version (#8961 ) PR #7782 set the dependency in Cargo.toml to 'master', and locked the version to commit that contained a specific fix, because we needed the fix before it was included in a versioned release. The fix was later included in parquet crate version 52.0.0, so we can now switch back to using a released version. The latest release is 53.0.0, switch straight to that. --------- Co-authored-by: Conrad Ludgate <conradludgate@gmail.com>	2024-09-10 00:04:00 +03:00
Alex Chi Z.	e158df4e86	feat(pageserver): split delta writer automatically determines key range (#8850 ) close https://github.com/neondatabase/neon/issues/8838 ## Summary of changes This patch modifies the split delta layer writer to avoid taking start_key and end_key when creating/finishing the layer writer. The start_key for the delta layers will be the first key provided to the layer writer, and the end_key would be the `last_key.next()`. This simplifies the delta layer writer API. On that, the layer key hack is removed. Image layers now use the full key range, and delta layers use the first/last key provided by the user. --------- Signed-off-by: Alex Chi Z <chi@neon.tech>	2024-09-09 22:03:27 +01:00
Heikki Linnakangas	723c0971e8	Don't create 'empty' branch in neon_simple_env (#8965 ) Now that we've given up hope on sharing the neon_simple_env between tests, there's no reason to not use the 'main' branch directly.	2024-09-09 12:38:34 +03:00
Heikki Linnakangas	c8f67eed8f	Remove TEST_SHARED_FIXTURES (#8965 ) I wish it worked, but it's been broken for a long time, so let's admit defeat and remove it. The idea of sharing the same pageserver and safekeeper environment between tests is still sound, and it could save a lot of time in our CI. We should perhaps put some time into doing that, but we're better off starting from scratch than trying to make TEST_SHARED_FIXTURES work in its current form.	2024-09-09 12:38:34 +03:00
Heikki Linnakangas	2d885ac07a	Update strum (#8962 ) I wanted to use some features from the newer version. The PR that needed the new version is not ready yet (and might never be), but seems nice to stay up in any case.	2024-09-08 21:47:57 +03:00
Heikki Linnakangas	89c5e80b3f	Update toml and toml_edit crates (#8963 ) Eliminates a few duplicate versions from the dependency tree.	2024-09-08 21:47:23 +03:00
Heikki Linnakangas	93ec7503e0	Lock the correct revision of rust-postgres crates (#8960 ) We modified the crate in an incompatible way and upgraded to the new version in PR #8076. However, it was reverted in #8654. The revert reverted the Cargo.lock reference to it, but since Cargo.toml still points to the (tip of the) 'neon' branch, every time you make any other unrelated changes to Cargo.toml, it also tries to update the rust-postgres crates to the tip of the 'neon' branch again, which doesn't work. To fix, lock the crates to the exact commit SHA that works.	2024-09-07 14:11:36 +01:00