fix clippy warnings

Signed-off-by: Alex Chi Z <chi@neon.tech>
more assertions
2026-07-20 04:20:38 +00:00 · 2023-08-04 15:19:41 -04:00 · 2023-08-04 14:55:25 -04:00 · 2023-08-04 14:50:55 -04:00 · 2023-07-21 15:23:28 -04:00 · 2023-07-20 16:40:52 -04:00
115 changed files with 2239 additions and 1235 deletions
--- a/.dockerignore
+++ b/.dockerignore
@@ -18,7 +18,6 @@
 !trace/
 !vendor/postgres-v14/
 !vendor/postgres-v15/
-!vendor/postgres-v16/
 !workspace_hack/
 !neon_local/
 !scripts/ninstall.sh
--- a/.github/actions/run-python-test-set/action.yml
+++ b/.github/actions/run-python-test-set/action.yml
@@ -150,6 +150,14 @@ runs:
          EXTRA_PARAMS="--flaky-tests-json $TEST_OUTPUT/flaky.json $EXTRA_PARAMS"
        fi

+        # We use pytest-split plugin to run benchmarks in parallel on different CI runners
+        if [ "${TEST_SELECTION}" = "test_runner/performance" ] && [ "${{ inputs.build_type }}" != "remote" ]; then
+          mkdir -p $TEST_OUTPUT
+          poetry run ./scripts/benchmark_durations.py "${TEST_RESULT_CONNSTR}" --days 10 --output "$TEST_OUTPUT/benchmark_durations.json"
+
+          EXTRA_PARAMS="--durations-path $TEST_OUTPUT/benchmark_durations.json $EXTRA_PARAMS"
+        fi
+
        if [[ "${{ inputs.build_type }}" == "debug" ]]; then
          cov_prefix=(scripts/coverage "--profraw-prefix=$GITHUB_JOB" --dir=/tmp/coverage run)
        elif [[ "${{ inputs.build_type }}" == "release" ]]; then
--- a/.github/workflows/build_and_test.yml
+++ b/.github/workflows/build_and_test.yml
@@ -396,13 +396,11 @@ jobs:
    strategy:
      fail-fast: false
      matrix:
+        pytest_split_group: [ 1, 2, 3, 4 ]
        build_type: [ release ]
    steps:
      - name: Checkout
        uses: actions/checkout@v3
-        with:
-          submodules: true
-          fetch-depth: 1

      - name: Pytest benchmarks
        uses: ./.github/actions/run-python-test-set
@@ -411,9 +409,11 @@ jobs:
          test_selection: performance
          run_in_parallel: false
          save_perf_report: ${{ github.ref_name == 'main' }}
+          extra_params: --splits ${{ strategy.job-total }} --group ${{ matrix.pytest_split_group }}
        env:
          VIP_VAP_ACCESS_TOKEN: "${{ secrets.VIP_VAP_ACCESS_TOKEN }}"
          PERF_TEST_RESULT_CONNSTR: "${{ secrets.PERF_TEST_RESULT_CONNSTR }}"
+          TEST_RESULT_CONNSTR: "${{ secrets.REGRESS_TEST_RESULT_CONNSTR }}"
      # XXX: no coverage data handling here, since benchmarks are run on release builds,
      # while coverage is currently collected for the debug ones

@@ -1007,6 +1007,8 @@ jobs:
          done

      - name: Upload postgres-extensions to S3
+        # TODO: Reenable step after switching to the new extensions format (tar-gzipped + index.json)
+        if: false
        run: |
          for BUCKET in $(echo ${S3_BUCKETS}); do
            aws s3 cp --recursive --only-show-errors ./extensions-to-upload s3://${BUCKET}/${{ needs.tag.outputs.build-tag }}/${{ matrix.version }}
--- a/.gitmodules
+++ b/.gitmodules
@@ -6,7 +6,3 @@
 	path = vendor/postgres-v15
 	url = https://github.com/neondatabase/postgres.git
 	branch = REL_15_STABLE_neon
-[submodule "vendor/postgres-v16"]
-	path = vendor/postgres-v16
-	url = https://github.com/neondatabase/postgres.git
-	branch = REL_16_STABLE_neon
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -110,6 +110,12 @@ dependencies = [
 "backtrace",
 ]

+[[package]]
+name = "arc-swap"
+version = "1.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bddcadddf5e9015d310179a59bb28c4d4b9920ad0f11e8e14dbadf654890c9a6"
+
 [[package]]
 name = "archery"
 version = "0.5.0"
@@ -1185,15 +1191,15 @@ dependencies = [

 [[package]]
 name = "dashmap"
-version = "5.4.0"
+version = "5.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "907076dfda823b0b36d2a1bb5f90c96660a5bbcd7729e10727f07858f22c4edc"
+checksum = "6943ae99c34386c84a470c499d3414f66502a41340aa895406e0d2e4a207b91d"
 dependencies = [
 "cfg-if",
- "hashbrown 0.12.3",
+ "hashbrown 0.14.0",
 "lock_api",
 "once_cell",
- "parking_lot_core 0.9.7",
+ "parking_lot_core 0.9.8",
 ]

 [[package]]
@@ -1642,6 +1648,12 @@ dependencies = [
 "ahash",
 ]

+[[package]]
+name = "hashbrown"
+version = "0.14.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2c6201b9ff9fd90a5a3bac2e56a830d0caa509576f0e503818ee82c181b3437a"
+
 [[package]]
 name = "hashlink"
 version = "0.8.2"
@@ -2057,9 +2069,9 @@ checksum = "ef53942eb7bf7ff43a617b3e2c1c4a5ecf5944a7c1bc12d7ee39bbb15e5c1519"

 [[package]]
 name = "lock_api"
-version = "0.4.9"
+version = "0.4.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "435011366fe56583b16cf956f9df0095b405b82d76425bc8981c0e22e60ec4df"
+checksum = "c1cc9717a20b1bb222f333e6a92fd32f7d8a18ddc5a3191a11af45dcbf4dcd16"
 dependencies = [
 "autocfg",
 "scopeguard",
@@ -2323,9 +2335,9 @@ dependencies = [

 [[package]]
 name = "once_cell"
-version = "1.17.1"
+version = "1.18.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b7e5500299e16ebb147ae15a00a942af264cf3688f47923b8fc2cd5858f23ad3"
+checksum = "dd8b5dd2ae5ed71462c540258bedcb51965123ad7e7ccf4b9a8cafaa4a63576d"

 [[package]]
 name = "oorandom"
@@ -2515,6 +2527,7 @@ name = "pageserver"
 version = "0.1.0"
 dependencies = [
 "anyhow",
+ "arc-swap",
 "async-compression",
 "async-stream",
 "async-trait",
@@ -2528,6 +2541,7 @@ dependencies = [
 "crc32c",
 "criterion",
 "crossbeam-utils",
+ "dashmap",
 "either",
 "enum-map",
 "enumset",
@@ -2622,7 +2636,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3742b2c103b9f06bc9fff0a37ff4912935851bee6d36f3c02bcc755bcfec228f"
 dependencies = [
 "lock_api",
- "parking_lot_core 0.9.7",
+ "parking_lot_core 0.9.8",
 ]

 [[package]]
@@ -2641,15 +2655,15 @@ dependencies = [

 [[package]]
 name = "parking_lot_core"
-version = "0.9.7"
+version = "0.9.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9069cbb9f99e3a5083476ccb29ceb1de18b9118cafa53e90c9551235de2b9521"
+checksum = "93f00c865fe7cabf650081affecd3871070f26767e7b2070a3ffae14c654b447"
 dependencies = [
 "cfg-if",
 "libc",
- "redox_syscall 0.2.16",
+ "redox_syscall 0.3.5",
 "smallvec",
- "windows-sys 0.45.0",
+ "windows-targets 0.48.0",
 ]

 [[package]]
@@ -3854,7 +3868,8 @@ dependencies = [
 [[package]]
 name = "sharded-slab"
 version = "0.1.4"
-source = "git+https://github.com/neondatabase/sharded-slab.git?rev=98d16753ab01c61f0a028de44167307a00efea00#98d16753ab01c61f0a028de44167307a00efea00"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "900fba806f70c630b0a382d0d825e17a0f19fcd059a2ade1ff237bcddf446b31"
 dependencies = [
 "lazy_static",
 ]
@@ -4098,7 +4113,7 @@ checksum = "4b55807c0344e1e6c04d7c965f5289c39a8d94ae23ed5c0b57aabac549f871c6"
 dependencies = [
 "filetime",
 "libc",
- "xattr",
+ "xattr 0.2.3",
 ]

 [[package]]
@@ -4379,16 +4394,17 @@ dependencies = [

 [[package]]
 name = "tokio-tar"
-version = "0.3.0"
-source = "git+https://github.com/neondatabase/tokio-tar.git?rev=404df61437de0feef49ba2ccdbdd94eb8ad6e142#404df61437de0feef49ba2ccdbdd94eb8ad6e142"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9d5714c010ca3e5c27114c1cdeb9d14641ace49874aa5626d7149e47aedace75"
 dependencies = [
 "filetime",
 "futures-core",
 "libc",
- "redox_syscall 0.2.16",
+ "redox_syscall 0.3.5",
 "tokio",
 "tokio-stream",
- "xattr",
+ "xattr 1.0.0",
 ]

 [[package]]
@@ -5362,6 +5378,15 @@ dependencies = [
 "libc",
 ]

+[[package]]
+name = "xattr"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ea263437ca03c1522846a4ddafbca2542d0ad5ed9b784909d4b27b76f62bc34a"
+dependencies = [
+ "libc",
+]
+
 [[package]]
 name = "xmlparser"
 version = "0.13.5"
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -32,6 +32,7 @@ license = "Apache-2.0"
 ## All dependency versions, used in the project
 [workspace.dependencies]
 anyhow = { version = "1.0", features = ["backtrace"] }
+arc-swap = "1.6"
 async-compression = { version = "0.4.0", features = ["tokio", "gzip"] }
 flate2 = "1.0.26"
 async-stream = "0.3"
@@ -54,6 +55,7 @@ comfy-table = "6.1"
 const_format = "0.2"
 crc32c = "0.6"
 crossbeam-utils = "0.8.5"
+dashmap = "5.5"
 either = "1.8"
 enum-map = "2.4.2"
 enumset = "1.0.12"
@@ -124,6 +126,7 @@ tokio-io-timeout = "1.2.0"
 tokio-postgres-rustls = "0.9.0"
 tokio-rustls = "0.23"
 tokio-stream = "0.1"
+tokio-tar = "0.3"
 tokio-util = { version = "0.7", features = ["io"] }
 toml = "0.7"
 toml_edit = "0.19"
@@ -148,7 +151,6 @@ postgres-native-tls = { git = "https://github.com/neondatabase/rust-postgres.git
 postgres-protocol = { git = "https://github.com/neondatabase/rust-postgres.git", rev="1aaedab101b23f7612042850d8f2036810fa7c7f" }
 postgres-types = { git = "https://github.com/neondatabase/rust-postgres.git", rev="1aaedab101b23f7612042850d8f2036810fa7c7f" }
 tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev="1aaedab101b23f7612042850d8f2036810fa7c7f" }
-tokio-tar = { git = "https://github.com/neondatabase/tokio-tar.git", rev="404df61437de0feef49ba2ccdbdd94eb8ad6e142" }

 ## Other git libraries
 heapless = { default-features=false, features=[], git = "https://github.com/japaric/heapless.git", rev = "644653bf3b831c6bb4963be2de24804acf5e5001" } # upstream release pending
@@ -185,11 +187,6 @@ tonic-build = "0.9"
 # TODO: we should probably fork `tokio-postgres-rustls` instead.
 tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev="1aaedab101b23f7612042850d8f2036810fa7c7f" }

-# Changes the MAX_THREADS limit from 4096 to 32768.
-# This is a temporary workaround for using tracing from many threads in safekeepers code,
-# until async safekeepers patch is merged to the main.
-sharded-slab = { git = "https://github.com/neondatabase/sharded-slab.git", rev="98d16753ab01c61f0a028de44167307a00efea00" }
-
 ################# Binary contents sections

 [profile.release]
--- a/3
+++ b/3
@@ -12,7 +12,6 @@ WORKDIR /home/nonroot

 COPY --chown=nonroot vendor/postgres-v14 vendor/postgres-v14
 COPY --chown=nonroot vendor/postgres-v15 vendor/postgres-v15
-COPY --chown=nonroot vendor/postgres-v16 vendor/postgres-v16
 COPY --chown=nonroot pgxn pgxn
 COPY --chown=nonroot Makefile Makefile
 COPY --chown=nonroot scripts/ninstall.sh scripts/ninstall.sh
@@ -40,7 +39,6 @@ ARG CACHEPOT_BUCKET=neon-github-dev

 COPY --from=pg-build /home/nonroot/pg_install/v14/include/postgresql/server pg_install/v14/include/postgresql/server
 COPY --from=pg-build /home/nonroot/pg_install/v15/include/postgresql/server pg_install/v15/include/postgresql/server
-COPY --from=pg-build /home/nonroot/pg_install/v16/include/postgresql/server pg_install/v16/include/postgresql/server
 COPY --chown=nonroot . .

 # Show build caching stats to check if it was used in the end.
@@ -81,7 +79,6 @@ COPY --from=build --chown=neon:neon /home/nonroot/target/release/proxy

 COPY --from=pg-build /home/nonroot/pg_install/v14 /usr/local/v14/
 COPY --from=pg-build /home/nonroot/pg_install/v15 /usr/local/v15/
-COPY --from=pg-build /home/nonroot/pg_install/v16 /usr/local/v16/
 COPY --from=pg-build /home/nonroot/postgres_install.tar.gz /data/

 # By default, pageserver uses `.neon/` working directory in WORKDIR, so create one and fill it with the dummy config.
--- a/Dockerfile.compute-node
+++ b/Dockerfile.compute-node
@@ -535,10 +535,10 @@ FROM build-deps AS pg-embedding-pg-build
 COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/

 ENV PATH "/usr/local/pgsql/bin/:$PATH"
-# 2465f831ea1f8d49c1d74f8959adb7fc277d70cd made on 05/07/2023
+# eeb3ba7c3a60c95b2604dd543c64b2f1bb4a3703 made on 15/07/2023
 # There is no release tag yet
-RUN wget https://github.com/neondatabase/pg_embedding/archive/2465f831ea1f8d49c1d74f8959adb7fc277d70cd.tar.gz -O pg_embedding.tar.gz && \
-    echo "047af2b1f664a1e6e37867bd4eeaf5934fa27d6ba3d6c4461efa388ddf7cd1d5 pg_embedding.tar.gz" | sha256sum --check && \
+RUN wget https://github.com/neondatabase/pg_embedding/archive/eeb3ba7c3a60c95b2604dd543c64b2f1bb4a3703.tar.gz -O pg_embedding.tar.gz && \
+    echo "030846df723652f99a8689ce63b66fa0c23477a7fd723533ab8a6b28ab70730f pg_embedding.tar.gz" | sha256sum --check && \
    mkdir pg_embedding-src && cd pg_embedding-src && tar xvzf ../pg_embedding.tar.gz --strip-components=1 -C . && \
    make -j $(getconf _NPROCESSORS_ONLN) && \
    make -j $(getconf _NPROCESSORS_ONLN) install && \
--- a/17
+++ b/17
@@ -83,8 +83,6 @@ $(POSTGRES_INSTALL_DIR)/build/%/config.status:
 # I'm not sure why it wouldn't work, but this is the only place (apart from
 # the "build-all-versions" entry points) where direct mention of PostgreSQL
 # versions is used.
-.PHONY: postgres-configure-v16
-postgres-configure-v16: $(POSTGRES_INSTALL_DIR)/build/v16/config.status
 .PHONY: postgres-configure-v15
 postgres-configure-v15: $(POSTGRES_INSTALL_DIR)/build/v15/config.status
 .PHONY: postgres-configure-v14
@@ -167,33 +165,28 @@ neon-pg-ext-clean-%:
 .PHONY: neon-pg-ext
 neon-pg-ext: \
 	neon-pg-ext-v14 \
-	neon-pg-ext-v15 \
-	neon-pg-ext-v16
+	neon-pg-ext-v15

 .PHONY: neon-pg-ext-clean
 neon-pg-ext-clean: \
 	neon-pg-ext-clean-v14 \
-	neon-pg-ext-clean-v15 \
-	neon-pg-ext-clean-v16
+	neon-pg-ext-clean-v15

 # shorthand to build all Postgres versions
 .PHONY: postgres
 postgres: \
 	postgres-v14 \
-	postgres-v15 \
-	postgres-v16
+	postgres-v15

 .PHONY: postgres-headers
 postgres-headers: \
 	postgres-headers-v14 \
-	postgres-headers-v15 \
-	postgres-headers-v16
+	postgres-headers-v15

 .PHONY: postgres-clean
 postgres-clean: \
 	postgres-clean-v14 \
-	postgres-clean-v15 \
-	postgres-clean-v16
+	postgres-clean-v15

 # This doesn't remove the effects of 'configure'.
 .PHONY: clean
--- a/control_plane/src/endpoint.rs
+++ b/control_plane/src/endpoint.rs
@@ -289,7 +289,7 @@ impl Endpoint {
                        .env
                        .safekeepers
                        .iter()
-                        .map(|sk| format!("localhost:{}", sk.pg_port))
+                        .map(|sk| format!("localhost:{}", sk.get_compute_port()))
                        .collect::<Vec<String>>()
                        .join(",");
                    conf.append("neon.safekeepers", &safekeepers);
@@ -318,7 +318,7 @@ impl Endpoint {
                    .env
                    .safekeepers
                    .iter()
-                    .map(|x| x.pg_port.to_string())
+                    .map(|x| x.get_compute_port().to_string())
                    .collect::<Vec<_>>()
                    .join(",");
                let sk_hosts = vec!["localhost"; self.env.safekeepers.len()].join(",");
@@ -463,7 +463,7 @@ impl Endpoint {
                    .iter()
                    .find(|node| node.id == sk_id)
                    .ok_or_else(|| anyhow!("safekeeper {sk_id} does not exist"))?;
-                safekeeper_connstrings.push(format!("127.0.0.1:{}", sk.pg_port));
+                safekeeper_connstrings.push(format!("127.0.0.1:{}", sk.get_compute_port()));
            }
        }

--- a/control_plane/src/local_env.rs
+++ b/control_plane/src/local_env.rs
@@ -137,6 +137,7 @@ impl Default for PageServerConf {
 pub struct SafekeeperConf {
    pub id: NodeId,
    pub pg_port: u16,
+    pub pg_tenant_only_port: Option<u16>,
    pub http_port: u16,
    pub sync: bool,
    pub remote_storage: Option<String>,
@@ -149,6 +150,7 @@ impl Default for SafekeeperConf {
        Self {
            id: NodeId(0),
            pg_port: 0,
+            pg_tenant_only_port: None,
            http_port: 0,
            sync: true,
            remote_storage: None,
@@ -158,6 +160,14 @@ impl Default for SafekeeperConf {
    }
 }

+impl SafekeeperConf {
+    /// Compute is served by port on which only tenant scoped tokens allowed, if
+    /// it is configured.
+    pub fn get_compute_port(&self) -> u16 {
+        self.pg_tenant_only_port.unwrap_or(self.pg_port)
+    }
+}
+
 impl LocalEnv {
    pub fn pg_distrib_dir_raw(&self) -> PathBuf {
        self.pg_distrib_dir.clone()
@@ -169,7 +179,6 @@ impl LocalEnv {
        match pg_version {
            14 => Ok(path.join(format!("v{pg_version}"))),
            15 => Ok(path.join(format!("v{pg_version}"))),
-            16 => Ok(path.join(format!("v{pg_version}"))),
            _ => bail!("Unsupported postgres version: {}", pg_version),
        }
    }
@@ -178,7 +187,6 @@ impl LocalEnv {
        match pg_version {
            14 => Ok(self.pg_distrib_dir(pg_version)?.join("bin")),
            15 => Ok(self.pg_distrib_dir(pg_version)?.join("bin")),
-            16 => Ok(self.pg_distrib_dir(pg_version)?.join("bin")),
            _ => bail!("Unsupported postgres version: {}", pg_version),
        }
    }
@@ -186,7 +194,6 @@ impl LocalEnv {
        match pg_version {
            14 => Ok(self.pg_distrib_dir(pg_version)?.join("lib")),
            15 => Ok(self.pg_distrib_dir(pg_version)?.join("lib")),
-            16 => Ok(self.pg_distrib_dir(pg_version)?.join("lib")),
            _ => bail!("Unsupported postgres version: {}", pg_version),
        }
    }
--- a/control_plane/src/safekeeper.rs
+++ b/control_plane/src/safekeeper.rs
@@ -120,45 +120,55 @@ impl SafekeeperNode {
        let availability_zone = format!("sk-{}", id_string);

        let mut args = vec![
-            "-D",
-            datadir.to_str().with_context(|| {
-                format!("Datadir path {datadir:?} cannot be represented as a unicode string")
-            })?,
-            "--id",
-            &id_string,
-            "--listen-pg",
-            &listen_pg,
-            "--listen-http",
-            &listen_http,
-            "--availability-zone",
-            &availability_zone,
+            "-D".to_owned(),
+            datadir
+                .to_str()
+                .with_context(|| {
+                    format!("Datadir path {datadir:?} cannot be represented as a unicode string")
+                })?
+                .to_owned(),
+            "--id".to_owned(),
+            id_string,
+            "--listen-pg".to_owned(),
+            listen_pg,
+            "--listen-http".to_owned(),
+            listen_http,
+            "--availability-zone".to_owned(),
+            availability_zone,
        ];
+        if let Some(pg_tenant_only_port) = self.conf.pg_tenant_only_port {
+            let listen_pg_tenant_only = format!("127.0.0.1:{}", pg_tenant_only_port);
+            args.extend(["--listen-pg-tenant-only".to_owned(), listen_pg_tenant_only]);
+        }
        if !self.conf.sync {
-            args.push("--no-sync");
+            args.push("--no-sync".to_owned());
        }

        let broker_endpoint = format!("{}", self.env.broker.client_url());
-        args.extend(["--broker-endpoint", &broker_endpoint]);
+        args.extend(["--broker-endpoint".to_owned(), broker_endpoint]);

        let mut backup_threads = String::new();
        if let Some(threads) = self.conf.backup_threads {
            backup_threads = threads.to_string();
-            args.extend(["--backup-threads", &backup_threads]);
+            args.extend(["--backup-threads".to_owned(), backup_threads]);
        } else {
            drop(backup_threads);
        }

        if let Some(ref remote_storage) = self.conf.remote_storage {
-            args.extend(["--remote-storage", remote_storage]);
+            args.extend(["--remote-storage".to_owned(), remote_storage.clone()]);
        }

        let key_path = self.env.base_data_dir.join("auth_public_key.pem");
        if self.conf.auth_enabled {
            args.extend([
-                "--auth-validation-public-key-path",
-                key_path.to_str().with_context(|| {
-                    format!("Key path {key_path:?} cannot be represented as a unicode string")
-                })?,
+                "--auth-validation-public-key-path".to_owned(),
+                key_path
+                    .to_str()
+                    .with_context(|| {
+                        format!("Key path {key_path:?} cannot be represented as a unicode string")
+                    })?
+                    .to_owned(),
            ]);
        }

--- a/docs/pageserver-thread-mgmt.md
+++ b/docs/pageserver-thread-mgmt.md
@@ -30,8 +30,8 @@ or similar, to wake up on shutdown.

 In async Rust, futures can be "cancelled" at any await point, by
 dropping the Future. For example, `tokio::select!` returns as soon as
-one of the Futures returns, and drops the others. `tokio::timeout!` is
-another example. In the Rust ecosystem, some functions are
+one of the Futures returns, and drops the others. `tokio::time::timeout`
+is another example. In the Rust ecosystem, some functions are
 cancellation-safe, meaning they can be safely dropped without
 side-effects, while others are not. See documentation of
 `tokio::select!` for examples.
@@ -42,9 +42,9 @@ function that you call cannot be assumed to be async
 cancellation-safe, and must be polled to completion.

 The downside of non-cancellation safe code is that you have to be very
-careful when using `tokio::select!`, `tokio::timeout!`, and other such
-functions that can cause a Future to be dropped. They can only be used
-with functions that are explicitly documented to be cancellation-safe,
+careful when using `tokio::select!`, `tokio::time::timeout`, and other
+such functions that can cause a Future to be dropped. They can only be
+used with functions that are explicitly documented to be cancellation-safe,
 or you need to spawn a separate task to shield from the cancellation.

 At the entry points to the code, we also take care to poll futures to
--- a/libs/pageserver_api/src/models.rs
+++ b/libs/pageserver_api/src/models.rs
@@ -9,6 +9,7 @@ use serde::{Deserialize, Serialize};
 use serde_with::{serde_as, DisplayFromStr};
 use strum_macros;
 use utils::{
+    completion,
    history_buffer::HistoryBufferWithDropCounter,
    id::{NodeId, TenantId, TimelineId},
    lsn::Lsn,
@@ -76,7 +77,12 @@ pub enum TenantState {
    /// system is being shut down.
    ///
    /// Transitions out of this state are possible through `set_broken()`.
-    Stopping,
+    Stopping {
+        // Because of https://github.com/serde-rs/serde/issues/2105 this has to be a named field,
+        // otherwise it will not be skipped during deserialization
+        #[serde(skip)]
+        progress: completion::Barrier,
+    },
    /// The tenant is recognized by the pageserver, but can no longer be used for
    /// any operations.
    ///
@@ -118,7 +124,7 @@ impl TenantState {
            // Why is Stopping a Maybe case? Because, during pageserver shutdown,
            // we set the Stopping state irrespective of whether the tenant
            // has finished attaching or not.
-            Self::Stopping => Maybe,
+            Self::Stopping { .. } => Maybe,
        }
    }

@@ -928,7 +934,13 @@ mod tests {
                "Activating",
            ),
            (line!(), TenantState::Active, "Active"),
-            (line!(), TenantState::Stopping, "Stopping"),
+            (
+                line!(),
+                TenantState::Stopping {
+                    progress: utils::completion::Barrier::default(),
+                },
+                "Stopping",
+            ),
            (
                line!(),
                TenantState::Broken {
--- a/libs/postgres_ffi/build.rs
+++ b/libs/postgres_ffi/build.rs
@@ -56,7 +56,7 @@ fn main() -> anyhow::Result<()> {
        PathBuf::from("pg_install")
    };

-    for pg_version in &["v14", "v15", "v16"] {
+    for pg_version in &["v14", "v15"] {
        let mut pg_install_dir_versioned = pg_install_dir.join(pg_version);
        if pg_install_dir_versioned.is_relative() {
            let cwd = env::current_dir().context("Failed to get current_dir")?;
--- a/libs/postgres_ffi/src/lib.rs
+++ b/libs/postgres_ffi/src/lib.rs
@@ -51,7 +51,6 @@ macro_rules! for_all_postgres_versions {
    ($macro:tt) => {
        $macro!(v14);
        $macro!(v15);
-        $macro!(v16);
    };
 }

@@ -93,10 +92,9 @@ pub use v14::bindings::DBState_DB_SHUTDOWNED;
 pub fn bkpimage_is_compressed(bimg_info: u8, version: u32) -> anyhow::Result<bool> {
    match version {
        14 => Ok(bimg_info & v14::bindings::BKPIMAGE_IS_COMPRESSED != 0),
-        15 | 16 => Ok(bimg_info & v15::bindings::BKPIMAGE_COMPRESS_PGLZ != 0
+        15 => Ok(bimg_info & v15::bindings::BKPIMAGE_COMPRESS_PGLZ != 0
            || bimg_info & v15::bindings::BKPIMAGE_COMPRESS_LZ4 != 0
            || bimg_info & v15::bindings::BKPIMAGE_COMPRESS_ZSTD != 0),
-
        _ => anyhow::bail!("Unknown version {}", version),
    }
 }
@@ -112,7 +110,6 @@ pub fn generate_wal_segment(
    match pg_version {
        14 => v14::xlog_utils::generate_wal_segment(segno, system_id, lsn),
        15 => v15::xlog_utils::generate_wal_segment(segno, system_id, lsn),
-        16 => v16::xlog_utils::generate_wal_segment(segno, system_id, lsn),
        _ => Err(SerializeError::BadInput),
    }
 }
@@ -126,7 +123,6 @@ pub fn generate_pg_control(
    match pg_version {
        14 => v14::xlog_utils::generate_pg_control(pg_control_bytes, checkpoint_bytes, lsn),
        15 => v15::xlog_utils::generate_pg_control(pg_control_bytes, checkpoint_bytes, lsn),
-        16 => v16::xlog_utils::generate_pg_control(pg_control_bytes, checkpoint_bytes, lsn),
        _ => anyhow::bail!("Unknown version {}", pg_version),
    }
 }
@@ -201,7 +197,7 @@ pub fn fsm_logical_to_physical(addr: BlockNumber) -> BlockNumber {

 pub mod waldecoder {

-    use crate::{v14, v15, v16};
+    use crate::{v14, v15};
    use bytes::{Buf, Bytes, BytesMut};
    use std::num::NonZeroU32;
    use thiserror::Error;
@@ -263,10 +259,6 @@ pub mod waldecoder {
                    use self::v15::waldecoder_handler::WalStreamDecoderHandler;
                    self.poll_decode_internal()
                }
-                16 => {
-                    use self::v16::waldecoder_handler::WalStreamDecoderHandler;
-                    self.poll_decode_internal()
-                }
                _ => Err(WalDecodeError {
                    msg: format!("Unknown version {}", self.pg_version),
                    lsn: self.lsn,
--- a/libs/postgres_ffi/src/pg_constants_v16.rs
+++ b/libs/postgres_ffi/src/pg_constants_v16.rs
@@ -1 +0,0 @@
-
--- a/libs/postgres_ffi/wal_craft/src/lib.rs
+++ b/libs/postgres_ffi/wal_craft/src/lib.rs
@@ -52,7 +52,6 @@ impl Conf {
        match self.pg_version {
            14 => Ok(path.join(format!("v{}", self.pg_version))),
            15 => Ok(path.join(format!("v{}", self.pg_version))),
-            16 => Ok(path.join(format!("v{}", self.pg_version))),
            _ => bail!("Unsupported postgres version: {}", self.pg_version),
        }
    }
--- a/libs/utils/src/auth.rs
+++ b/libs/utils/src/auth.rs
@@ -16,7 +16,7 @@ use crate::id::TenantId;
 /// Algorithm to use. We require EdDSA.
 const STORAGE_TOKEN_ALGORITHM: Algorithm = Algorithm::EdDSA;

-#[derive(Debug, Serialize, Deserialize, Clone, PartialEq)]
+#[derive(Debug, Serialize, Deserialize, Clone, Copy, PartialEq)]
 #[serde(rename_all = "lowercase")]
 pub enum Scope {
    // Provides access to all data for a specific tenant (specified in `struct Claims` below)
--- a/libs/utils/src/completion.rs
+++ b/libs/utils/src/completion.rs
@@ -12,6 +12,13 @@ pub struct Completion(mpsc::Sender<()>);
 #[derive(Clone)]
 pub struct Barrier(Arc<Mutex<mpsc::Receiver<()>>>);

+impl Default for Barrier {
+    fn default() -> Self {
+        let (_, rx) = channel();
+        rx
+    }
+}
+
 impl Barrier {
    pub async fn wait(self) {
        self.0.lock().await.recv().await;
@@ -24,6 +31,15 @@ impl Barrier {
    }
 }

+impl PartialEq for Barrier {
+    fn eq(&self, other: &Self) -> bool {
+        // we don't use dyn so this is good
+        Arc::ptr_eq(&self.0, &other.0)
+    }
+}
+
+impl Eq for Barrier {}
+
 /// Create new Guard and Barrier pair.
 pub fn channel() -> (Completion, Barrier) {
    let (tx, rx) = mpsc::channel::<()>(1);
--- a/libs/utils/src/error.rs
+++ b/libs/utils/src/error.rs
@@ -0,0 +1,111 @@
+/// Create a reporter for an error that outputs similar to [`anyhow::Error`] with Display with alternative setting.
+///
+/// It can be used with `anyhow::Error` as well.
+///
+/// Why would one use this instead of converting to `anyhow::Error` on the spot? Because
+/// anyhow::Error would also capture a stacktrace on the spot, which you would later discard after
+/// formatting.
+///
+/// ## Usage
+///
+/// ```rust
+/// #[derive(Debug, thiserror::Error)]
+/// enum MyCoolError {
+///   #[error("should never happen")]
+///   Bad(#[source] std::io::Error),
+/// }
+///
+/// # fn failing_call() -> Result<(), MyCoolError> { Err(MyCoolError::Bad(std::io::ErrorKind::PermissionDenied.into())) }
+///
+/// # fn main() {
+/// use utils::error::report_compact_sources;
+///
+/// if let Err(e) = failing_call() {
+///     let e = report_compact_sources(&e);
+///     assert_eq!(format!("{e}"), "should never happen: permission denied");
+/// }
+/// # }
+/// ```
+///
+/// ## TODO
+///
+/// When we are able to describe return position impl trait in traits, this should of course be an
+/// extension trait. Until then avoid boxing with this more ackward interface.
+pub fn report_compact_sources<E: std::error::Error>(e: &E) -> impl std::fmt::Display + '_ {
+    struct AnyhowDisplayAlternateAlike<'a, E>(&'a E);
+
+    impl<E: std::error::Error> std::fmt::Display for AnyhowDisplayAlternateAlike<'_, E> {
+        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+            write!(f, "{}", self.0)?;
+
+            // why is E a generic parameter here? hope that rustc will see through a default
+            // Error::source implementation and leave the following out if there cannot be any
+            // sources:
+            Sources(self.0.source()).try_for_each(|src| write!(f, ": {}", src))
+        }
+    }
+
+    struct Sources<'a>(Option<&'a (dyn std::error::Error + 'static)>);
+
+    impl<'a> Iterator for Sources<'a> {
+        type Item = &'a (dyn std::error::Error + 'static);
+
+        fn next(&mut self) -> Option<Self::Item> {
+            let rem = self.0;
+
+            let next = self.0.and_then(|x| x.source());
+            self.0 = next;
+            rem
+        }
+    }
+
+    AnyhowDisplayAlternateAlike(e)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::report_compact_sources;
+
+    #[test]
+    fn report_compact_sources_examples() {
+        use std::fmt::Write;
+
+        #[derive(Debug, thiserror::Error)]
+        enum EvictionError {
+            #[error("cannot evict a remote layer")]
+            CannotEvictRemoteLayer,
+            #[error("stat failed")]
+            StatFailed(#[source] std::io::Error),
+            #[error("layer was no longer part of LayerMap")]
+            LayerNotFound(#[source] anyhow::Error),
+        }
+
+        let examples = [
+            (
+                line!(),
+                EvictionError::CannotEvictRemoteLayer,
+                "cannot evict a remote layer",
+            ),
+            (
+                line!(),
+                EvictionError::StatFailed(std::io::ErrorKind::PermissionDenied.into()),
+                "stat failed: permission denied",
+            ),
+            (
+                line!(),
+                EvictionError::LayerNotFound(anyhow::anyhow!("foobar")),
+                "layer was no longer part of LayerMap: foobar",
+            ),
+        ];
+
+        let mut s = String::new();
+
+        for (line, example, expected) in examples {
+            s.clear();
+
+            write!(s, "{}", report_compact_sources(&example)).expect("string grows");
+
+            assert_eq!(s, expected, "example on line {line}");
+        }
+    }
+}
--- a/libs/utils/src/lib.rs
+++ b/libs/utils/src/lib.rs
@@ -63,6 +63,9 @@ pub mod rate_limit;
 /// Simple once-barrier and a guard which keeps barrier awaiting.
 pub mod completion;

+/// Reporting utilities
+pub mod error;
+
 mod failpoint_macro_helpers {

    /// use with fail::cfg("$name", "return(2000)")
--- a/libs/utils/src/tracing_span_assert.rs
+++ b/libs/utils/src/tracing_span_assert.rs
@@ -164,9 +164,7 @@ fn tracing_subscriber_configured() -> bool {
    tracing::dispatcher::get_default(|d| {
        // it is possible that this closure will not be invoked, but the current implementation
        // always invokes it
-        noop_configured = d
-            .downcast_ref::<tracing::subscriber::NoSubscriber>()
-            .is_some();
+        noop_configured = d.is::<tracing::subscriber::NoSubscriber>();
    });

    !noop_configured
--- a/pageserver/Cargo.toml
+++ b/pageserver/Cargo.toml
@@ -12,6 +12,7 @@ testing = ["fail/failpoints"]

 [dependencies]
 anyhow.workspace = true
+arc-swap.workspace = true
 async-compression.workspace = true
 async-stream.workspace = true
 async-trait.workspace = true
@@ -24,6 +25,7 @@ const_format.workspace = true
 consumption_metrics.workspace = true
 crc32c.workspace = true
 crossbeam-utils.workspace = true
+dashmap.workspace = true
 either.workspace = true
 flate2.workspace = true
 fail.workspace = true
@@ -82,6 +84,7 @@ strum_macros.workspace = true
 criterion.workspace = true
 hex-literal.workspace = true
 tempfile.workspace = true
+tokio = { workspace = true, features = ["process", "sync", "fs", "rt", "io-util", "time", "test-util"] }

 [[bench]]
 name = "bench_layer_map"
--- a/pageserver/src/basebackup.rs
+++ b/pageserver/src/basebackup.rs
@@ -19,12 +19,6 @@ use tokio::io;
 use tokio::io::AsyncWrite;
 use tracing::*;

-/// NB: This relies on a modified version of tokio_tar that does *not* write the
-/// end-of-archive marker (1024 zero bytes), when the Builder struct is dropped
-/// without explicitly calling 'finish' or 'into_inner'!
-///
-/// See https://github.com/neondatabase/tokio-tar/pull/1
-///
 use tokio_tar::{Builder, EntryType, Header};

 use crate::context::RequestContext;
--- a/pageserver/src/bin/pageserver.rs
+++ b/pageserver/src/bin/pageserver.rs
@@ -396,8 +396,8 @@ fn start_pageserver(

            let guard = scopeguard::guard_on_success((), |_| tracing::info!("Cancelled before initial logical sizes completed"));

-            let init_sizes_done = tokio::select! {
-                _ = &mut init_sizes_done => {
+            let init_sizes_done = match tokio::time::timeout(timeout, &mut init_sizes_done).await {
+                Ok(_) => {
                    let now = std::time::Instant::now();
                    tracing::info!(
                        from_init_done_millis = (now - init_done).as_millis(),
@@ -406,7 +406,7 @@ fn start_pageserver(
                    );
                    None
                }
-                _ = tokio::time::sleep(timeout) => {
+                Err(_) => {
                    tracing::info!(
                        timeout_millis = timeout.as_millis(),
                        "Initial logical size timeout elapsed; starting background jobs"
--- a/pageserver/src/config.rs
+++ b/pageserver/src/config.rs
@@ -655,7 +655,6 @@ impl PageServerConf {
        match pg_version {
            14 => Ok(path.join(format!("v{pg_version}"))),
            15 => Ok(path.join(format!("v{pg_version}"))),
-            16 => Ok(path.join(format!("v{pg_version}"))),
            _ => bail!("Unsupported postgres version: {}", pg_version),
        }
    }
@@ -664,7 +663,6 @@ impl PageServerConf {
        match pg_version {
            14 => Ok(self.pg_distrib_dir(pg_version)?.join("bin")),
            15 => Ok(self.pg_distrib_dir(pg_version)?.join("bin")),
-            16 => Ok(self.pg_distrib_dir(pg_version)?.join("bin")),
            _ => bail!("Unsupported postgres version: {}", pg_version),
        }
    }
@@ -672,7 +670,6 @@ impl PageServerConf {
        match pg_version {
            14 => Ok(self.pg_distrib_dir(pg_version)?.join("lib")),
            15 => Ok(self.pg_distrib_dir(pg_version)?.join("lib")),
-            16 => Ok(self.pg_distrib_dir(pg_version)?.join("lib")),
            _ => bail!("Unsupported postgres version: {}", pg_version),
        }
    }
--- a/pageserver/src/disk_usage_eviction_task.rs
+++ b/pageserver/src/disk_usage_eviction_task.rs
@@ -60,7 +60,7 @@ use utils::serde_percent::Percent;
 use crate::{
    config::PageServerConf,
    task_mgr::{self, TaskKind, BACKGROUND_RUNTIME},
-    tenant::{self, storage_layer::PersistentLayer, Timeline},
+    tenant::{self, storage_layer::PersistentLayer, timeline::EvictionError, Timeline},
 };

 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -166,11 +166,11 @@ async fn disk_usage_eviction_task(
        .await;

        let sleep_until = start + task_config.period;
-        tokio::select! {
-            _ = tokio::time::sleep_until(sleep_until) => {},
-            _ = cancel.cancelled() => {
-                break
-            }
+        if tokio::time::timeout_at(sleep_until, cancel.cancelled())
+            .await
+            .is_ok()
+        {
+            break;
        }
    }
 }
@@ -390,13 +390,22 @@ pub async fn disk_usage_eviction_task_iteration_impl<U: Usage>(
                    assert_eq!(results.len(), batch.len());
                    for (result, layer) in results.into_iter().zip(batch.iter()) {
                        match result {
-                            Some(Ok(true)) => {
+                            Some(Ok(())) => {
                                usage_assumed.add_available_bytes(layer.file_size());
                            }
-                            Some(Ok(false)) => {
-                                // this is:
-                                // - Replacement::{NotFound, Unexpected}
-                                // - it cannot be is_remote_layer, filtered already
+                            Some(Err(EvictionError::CannotEvictRemoteLayer)) => {
+                                unreachable!("get_local_layers_for_disk_usage_eviction finds only local layers")
+                            }
+                            Some(Err(EvictionError::FileNotFound)) => {
+                                evictions_failed.file_sizes += layer.file_size();
+                                evictions_failed.count += 1;
+                            }
+                            Some(Err(
+                                e @ EvictionError::LayerNotFound(_)
+                                | e @ EvictionError::StatFailed(_),
+                            )) => {
+                                let e = utils::error::report_compact_sources(&e);
+                                warn!(%layer, "failed to evict layer: {e}");
                                evictions_failed.file_sizes += layer.file_size();
                                evictions_failed.count += 1;
                            }
@@ -404,10 +413,6 @@ pub async fn disk_usage_eviction_task_iteration_impl<U: Usage>(
                                assert!(cancel.is_cancelled());
                                return;
                            }
-                            Some(Err(e)) => {
-                                // we really shouldn't be getting this, precondition failure
-                                error!("failed to evict layer: {:#}", e);
-                            }
                        }
                    }
                }
--- a/pageserver/src/task_mgr.rs
+++ b/pageserver/src/task_mgr.rs
@@ -511,17 +511,13 @@ pub async fn shutdown_tasks(
                    warn!(name = task.name, tenant_id = ?tenant_id, timeline_id = ?timeline_id, kind = ?task_kind, "stopping left-over");
                }
            }
-            let join_handle = tokio::select! {
-                biased;
-                _ = &mut join_handle => { None },
-                _ = tokio::time::sleep(std::time::Duration::from_secs(1)) => {
-                    // allow some time to elapse before logging to cut down the number of log
-                    // lines.
-                    info!("waiting for {} to shut down", task.name);
-                    Some(join_handle)
-                }
-            };
-            if let Some(join_handle) = join_handle {
+            if tokio::time::timeout(std::time::Duration::from_secs(1), &mut join_handle)
+                .await
+                .is_err()
+            {
+                // allow some time to elapse before logging to cut down the number of log
+                // lines.
+                info!("waiting for {} to shut down", task.name);
                // we never handled this return value, but:
                // - we don't deschedule which would lead to is_cancelled
                // - panics are already logged (is_panicked)
--- a/pageserver/src/tenant.rs
+++ b/pageserver/src/tenant.rs
@@ -121,7 +121,7 @@ pub mod mgr;
 pub mod tasks;
 pub mod upload_queue;

-mod timeline;
+pub(crate) mod timeline;

 pub mod size;

@@ -281,7 +281,7 @@ pub enum DeleteTimelineError {
 }

 pub enum SetStoppingError {
-    AlreadyStopping,
+    AlreadyStopping(completion::Barrier),
    Broken,
 }

@@ -318,10 +318,6 @@ impl std::fmt::Display for WaitToBecomeActiveError {
    }
 }

-pub(crate) enum ShutdownError {
-    AlreadyStopping,
-}
-
 struct DeletionGuard(OwnedMutexGuard<bool>);

 impl DeletionGuard {
@@ -1172,7 +1168,7 @@ impl Tenant {
        )
    }

-    /// Helper for unit tests to create an emtpy timeline.
+    /// Helper for unit tests to create an empty timeline.
    ///
    /// The timeline is has state value `Active` but its background loops are not running.
    // This makes the various functions which anyhow::ensure! for Active state work in tests.
@@ -1721,7 +1717,7 @@ impl Tenant {
        self.state.send_modify(|current_state| {
            use pageserver_api::models::ActivatingFrom;
            match &*current_state {
-                TenantState::Activating(_) | TenantState::Active | TenantState::Broken { .. } | TenantState::Stopping => {
+                TenantState::Activating(_) | TenantState::Active | TenantState::Broken { .. } | TenantState::Stopping { .. } => {
                    panic!("caller is responsible for calling activate() only on Loading / Attaching tenants, got {state:?}", state = current_state);
                }
                TenantState::Loading => {
@@ -1785,7 +1781,16 @@ impl Tenant {
    /// - detach + ignore (freeze_and_flush == false)
    ///
    /// This will attempt to shutdown even if tenant is broken.
-    pub(crate) async fn shutdown(&self, freeze_and_flush: bool) -> Result<(), ShutdownError> {
+    ///
+    /// `shutdown_progress` is a [`completion::Barrier`] for the shutdown initiated by this call.
+    /// If the tenant is already shutting down, we return a clone of the first shutdown call's
+    /// `Barrier` as an `Err`. This not-first caller can use the returned barrier to join with
+    /// the ongoing shutdown.
+    async fn shutdown(
+        &self,
+        shutdown_progress: completion::Barrier,
+        freeze_and_flush: bool,
+    ) -> Result<(), completion::Barrier> {
        span::debug_assert_current_span_has_tenant_id();
        // Set tenant (and its timlines) to Stoppping state.
        //
@@ -1804,12 +1809,16 @@ impl Tenant {
        // But the tenant background loops are joined-on in our caller.
        // It's mesed up.
        // we just ignore the failure to stop
-        match self.set_stopping().await {
+
+        match self.set_stopping(shutdown_progress).await {
            Ok(()) => {}
            Err(SetStoppingError::Broken) => {
                // assume that this is acceptable
            }
-            Err(SetStoppingError::AlreadyStopping) => return Err(ShutdownError::AlreadyStopping),
+            Err(SetStoppingError::AlreadyStopping(other)) => {
+                // give caller the option to wait for this this shutdown
+                return Err(other);
+            }
        };

        if freeze_and_flush {
@@ -1841,7 +1850,7 @@ impl Tenant {
    /// This function waits for the tenant to become active if it isn't already, before transitioning it into Stopping state.
    ///
    /// This function is not cancel-safe!
-    async fn set_stopping(&self) -> Result<(), SetStoppingError> {
+    async fn set_stopping(&self, progress: completion::Barrier) -> Result<(), SetStoppingError> {
        let mut rx = self.state.subscribe();

        // cannot stop before we're done activating, so wait out until we're done activating
@@ -1853,7 +1862,7 @@ impl Tenant {
                );
                false
            }
-            TenantState::Active | TenantState::Broken { .. } | TenantState::Stopping {} => true,
+            TenantState::Active | TenantState::Broken { .. } | TenantState::Stopping { .. } => true,
        })
        .await
        .expect("cannot drop self.state while on a &self method");
@@ -1868,7 +1877,7 @@ impl Tenant {
                // FIXME: due to time-of-check vs time-of-use issues, it can happen that new timelines
                // are created after the transition to Stopping. That's harmless, as the Timelines
                // won't be accessible to anyone afterwards, because the Tenant is in Stopping state.
-                *current_state = TenantState::Stopping;
+                *current_state = TenantState::Stopping { progress };
                // Continue stopping outside the closure. We need to grab timelines.lock()
                // and we plan to turn it into a tokio::sync::Mutex in a future patch.
                true
@@ -1880,9 +1889,9 @@ impl Tenant {
                err = Some(SetStoppingError::Broken);
                false
            }
-            TenantState::Stopping => {
+            TenantState::Stopping { progress } => {
                info!("Tenant is already in Stopping state");
-                err = Some(SetStoppingError::AlreadyStopping);
+                err = Some(SetStoppingError::AlreadyStopping(progress.clone()));
                false
            }
        });
@@ -1926,7 +1935,7 @@ impl Tenant {
                );
                false
            }
-            TenantState::Active | TenantState::Broken { .. } | TenantState::Stopping {} => true,
+            TenantState::Active | TenantState::Broken { .. } | TenantState::Stopping { .. } => true,
        })
        .await
        .expect("cannot drop self.state while on a &self method");
@@ -1949,7 +1958,7 @@ impl Tenant {
                    warn!("Tenant is already in Broken state");
                }
                // This is the only "expected" path, any other path is a bug.
-                TenantState::Stopping => {
+                TenantState::Stopping { .. } => {
                    warn!(
                        "Marking Stopping tenant as Broken state, reason: {}",
                        reason
@@ -1982,7 +1991,7 @@ impl Tenant {
                TenantState::Active { .. } => {
                    return Ok(());
                }
-                TenantState::Broken { .. } | TenantState::Stopping => {
+                TenantState::Broken { .. } | TenantState::Stopping { .. } => {
                    // There's no chance the tenant can transition back into ::Active
                    return Err(WaitToBecomeActiveError::WillNotBecomeActive {
                        tenant_id: self.tenant_id,
@@ -3350,14 +3359,18 @@ pub mod harness {
        pub async fn load(&self) -> (Arc<Tenant>, RequestContext) {
            let ctx = RequestContext::new(TaskKind::UnitTest, DownloadBehavior::Error);
            (
-                self.try_load(&ctx)
+                self.try_load(&ctx, None)
                    .await
                    .expect("failed to load test tenant"),
                ctx,
            )
        }

-        pub async fn try_load(&self, ctx: &RequestContext) -> anyhow::Result<Arc<Tenant>> {
+        pub async fn try_load(
+            &self,
+            ctx: &RequestContext,
+            remote_storage: Option<remote_storage::GenericRemoteStorage>,
+        ) -> anyhow::Result<Arc<Tenant>> {
            let walredo_mgr = Arc::new(TestRedoManager);

            let tenant = Arc::new(Tenant::new(
@@ -3366,7 +3379,7 @@ pub mod harness {
                TenantConfOpt::from(self.tenant_conf),
                walredo_mgr,
                self.tenant_id,
-                None,
+                remote_storage,
            ));
            tenant
                .load(None, ctx)
@@ -3904,7 +3917,11 @@ mod tests {
        metadata_bytes[8] ^= 1;
        std::fs::write(metadata_path, metadata_bytes)?;

-        let err = harness.try_load(&ctx).await.err().expect("should fail");
+        let err = harness
+            .try_load(&ctx, None)
+            .await
+            .err()
+            .expect("should fail");
        // get all the stack with all .context, not tonly the last one
        let message = format!("{err:#}");
        let expected = "Failed to parse metadata bytes from path";
--- a/pageserver/src/tenant/layer_map.rs
+++ b/pageserver/src/tenant/layer_map.rs
@@ -65,7 +65,7 @@ use super::storage_layer::PersistentLayerDesc;
 ///
 /// LayerMap tracks what layers exist on a timeline.
 ///
-#[derive(Default)]
+#[derive(Default, Clone)]
 pub struct LayerMap {
    //
    // 'open_layer' holds the current InMemoryLayer that is accepting new
@@ -727,7 +727,7 @@ mod tests {
            // and can remove it in the future.
            let _map = LayerMap::default();

-            let mut mapping = TestLayerFileManager::new();
+            let mapping = TestLayerFileManager::new();

            mapping
                .replace_and_verify(not_found, new_version)
@@ -742,7 +742,7 @@ mod tests {
            let downloaded = Arc::new(LayerObject::new(skeleton));

            let mut map = LayerMap::default();
-            let mut mapping = LayerFileManager::new();
+            let mapping = LayerFileManager::new();

            // two disjoint Arcs in different lifecycle phases. even if it seems they must be the
            // same layer, we use LayerMap::compare_arced_layers as the identity of layers.
--- a/pageserver/src/tenant/layer_map/historic_layer_coverage.rs
+++ b/pageserver/src/tenant/layer_map/historic_layer_coverage.rs
@@ -60,6 +60,7 @@ impl From<&PersistentLayerDesc> for LayerKey {
 /// Allows answering layer map queries very efficiently,
 /// but doesn't allow retroactive insertion, which is
 /// sometimes necessary. See BufferedHistoricLayerCoverage.
+#[derive(Clone)]
 pub struct HistoricLayerCoverage<Value> {
    /// The latest state
    head: LayerCoverageTuple<Value>,
@@ -412,6 +413,7 @@ fn test_persistent_overlapping() {
 ///
 /// See this for more on persistent and retroactive techniques:
 /// <https://www.youtube.com/watch?v=WqCWghETNDc&t=581s>
+#[derive(Clone)]
 pub struct BufferedHistoricLayerCoverage<Value> {
    /// A persistent layer map that we rebuild when we need to retroactively update
    historic_coverage: HistoricLayerCoverage<Value>,
--- a/pageserver/src/tenant/layer_map/layer_coverage.rs
+++ b/pageserver/src/tenant/layer_map/layer_coverage.rs
@@ -15,6 +15,7 @@ use rpds::RedBlackTreeMapSync;
 ///
 /// NOTE The struct is parameterized over Value for easier
 ///      testing, but in practice it's some sort of layer.
+#[derive(Clone)]
 pub struct LayerCoverage<Value> {
    /// For every change in coverage (as we sweep the key space)
    /// we store (lsn.end, value).
@@ -138,6 +139,7 @@ impl<Value: Clone> LayerCoverage<Value> {
 }

 /// Image and delta coverage at a specific LSN.
+#[derive(Clone)]
 pub struct LayerCoverageTuple<Value> {
    pub image_coverage: LayerCoverage<Value>,
    pub delta_coverage: LayerCoverage<Value>,
--- a/pageserver/src/tenant/mgr.rs
+++ b/pageserver/src/tenant/mgr.rs
@@ -233,11 +233,17 @@ pub fn schedule_local_tenant_processing(
 /// That could be easily misinterpreted by control plane, the consumer of the
 /// management API. For example, it could attach the tenant on a different pageserver.
 /// We would then be in split-brain once this pageserver restarts.
-#[instrument]
+#[instrument(skip_all)]
 pub async fn shutdown_all_tenants() {
+    shutdown_all_tenants0(&TENANTS).await
+}
+
+async fn shutdown_all_tenants0(tenants: &tokio::sync::RwLock<TenantsMap>) {
+    use utils::completion;
+
    // Prevent new tenants from being created.
    let tenants_to_shut_down = {
-        let mut m = TENANTS.write().await;
+        let mut m = tenants.write().await;
        match &mut *m {
            TenantsMap::Initializing => {
                *m = TenantsMap::ShuttingDown(HashMap::default());
@@ -262,14 +268,41 @@ pub async fn shutdown_all_tenants() {
    for (tenant_id, tenant) in tenants_to_shut_down {
        join_set.spawn(
            async move {
-                let freeze_and_flush = true;
+                // ordering shouldn't matter for this, either we store true right away or never
+                let ordering = std::sync::atomic::Ordering::Relaxed;
+                let joined_other = std::sync::atomic::AtomicBool::new(false);

-                match tenant.shutdown(freeze_and_flush).await {
-                    Ok(()) => debug!("tenant successfully stopped"),
-                    Err(super::ShutdownError::AlreadyStopping) => {
-                        warn!("tenant was already shutting down")
+                let mut shutdown = std::pin::pin!(async {
+                    let freeze_and_flush = true;
+
+                    let res = {
+                        let (_guard, shutdown_progress) = completion::channel();
+                        tenant.shutdown(shutdown_progress, freeze_and_flush).await
+                    };
+
+                    if let Err(other_progress) = res {
+                        // join the another shutdown in progress
+                        joined_other.store(true, ordering);
+                        other_progress.wait().await;
                    }
-                }
+                });
+
+                // in practice we might not have a lot time to go, since systemd is going to
+                // SIGKILL us at 10s, but we can try. delete tenant might take a while, so put out
+                // a warning.
+                let warning = std::time::Duration::from_secs(5);
+                let mut warning = std::pin::pin!(tokio::time::sleep(warning));
+
+                tokio::select! {
+                    _ = &mut shutdown => {},
+                    _ = &mut warning => {
+                        let joined_other = joined_other.load(ordering);
+                        warn!(%joined_other, "waiting for the shutdown to complete");
+                        shutdown.await;
+                    }
+                };
+
+                debug!("tenant successfully stopped");
            }
            .instrument(info_span!("shutdown", %tenant_id)),
        );
@@ -413,6 +446,15 @@ pub async fn detach_tenant(
    conf: &'static PageServerConf,
    tenant_id: TenantId,
    detach_ignored: bool,
+) -> Result<(), TenantStateError> {
+    detach_tenant0(conf, &TENANTS, tenant_id, detach_ignored).await
+}
+
+async fn detach_tenant0(
+    conf: &'static PageServerConf,
+    tenants: &tokio::sync::RwLock<TenantsMap>,
+    tenant_id: TenantId,
+    detach_ignored: bool,
 ) -> Result<(), TenantStateError> {
    let local_files_cleanup_operation = |tenant_id_to_clean| async move {
        let local_tenant_directory = conf.tenant_path(&tenant_id_to_clean);
@@ -425,7 +467,8 @@ pub async fn detach_tenant(
    };

    let removal_result =
-        remove_tenant_from_memory(tenant_id, local_files_cleanup_operation(tenant_id)).await;
+        remove_tenant_from_memory(tenants, tenant_id, local_files_cleanup_operation(tenant_id))
+            .await;

    // Ignored tenants are not present in memory and will bail the removal from memory operation.
    // Before returning the error, check for ignored tenant removal case — we only need to clean its local files then.
@@ -472,7 +515,15 @@ pub async fn ignore_tenant(
    conf: &'static PageServerConf,
    tenant_id: TenantId,
 ) -> Result<(), TenantStateError> {
-    remove_tenant_from_memory(tenant_id, async {
+    ignore_tenant0(conf, &TENANTS, tenant_id).await
+}
+
+async fn ignore_tenant0(
+    conf: &'static PageServerConf,
+    tenants: &tokio::sync::RwLock<TenantsMap>,
+    tenant_id: TenantId,
+) -> Result<(), TenantStateError> {
+    remove_tenant_from_memory(tenants, tenant_id, async {
        let ignore_mark_file = conf.tenant_ignore_mark_file_path(&tenant_id);
        fs::File::create(&ignore_mark_file)
            .await
@@ -597,18 +648,21 @@ where
 /// If the cleanup fails, tenant will stay in memory in [`TenantState::Broken`] state, and another removal
 /// operation would be needed to remove it.
 async fn remove_tenant_from_memory<V, F>(
+    tenants: &tokio::sync::RwLock<TenantsMap>,
    tenant_id: TenantId,
    tenant_cleanup: F,
 ) -> Result<V, TenantStateError>
 where
    F: std::future::Future<Output = anyhow::Result<V>>,
 {
+    use utils::completion;
+
    // It's important to keep the tenant in memory after the final cleanup, to avoid cleanup races.
    // The exclusive lock here ensures we don't miss the tenant state updates before trying another removal.
    // tenant-wde cleanup operations may take some time (removing the entire tenant directory), we want to
    // avoid holding the lock for the entire process.
    let tenant = {
-        TENANTS
+        tenants
            .write()
            .await
            .get(&tenant_id)
@@ -616,14 +670,20 @@ where
            .ok_or(TenantStateError::NotFound(tenant_id))?
    };

+    // allow pageserver shutdown to await for our completion
+    let (_guard, progress) = completion::channel();
+
+    // whenever we remove a tenant from memory, we don't want to flush and wait for upload
    let freeze_and_flush = false;

    // shutdown is sure to transition tenant to stopping, and wait for all tasks to complete, so
    // that we can continue safely to cleanup.
-    match tenant.shutdown(freeze_and_flush).await {
+    match tenant.shutdown(progress, freeze_and_flush).await {
        Ok(()) => {}
-        Err(super::ShutdownError::AlreadyStopping) => {
-            return Err(TenantStateError::IsStopping(tenant_id))
+        Err(_other) => {
+            // if pageserver shutdown or other detach/ignore is already ongoing, we don't want to
+            // wait for it but return an error right away because these are distinct requests.
+            return Err(TenantStateError::IsStopping(tenant_id));
        }
    }

@@ -632,14 +692,14 @@ where
        .with_context(|| format!("Failed to run cleanup for tenant {tenant_id}"))
    {
        Ok(hook_value) => {
-            let mut tenants_accessor = TENANTS.write().await;
+            let mut tenants_accessor = tenants.write().await;
            if tenants_accessor.remove(&tenant_id).is_none() {
                warn!("Tenant {tenant_id} got removed from memory before operation finished");
            }
            Ok(hook_value)
        }
        Err(e) => {
-            let tenants_accessor = TENANTS.read().await;
+            let tenants_accessor = tenants.read().await;
            match tenants_accessor.get(&tenant_id) {
                Some(tenant) => {
                    tenant.set_broken(e.to_string()).await;
@@ -756,3 +816,109 @@ pub async fn immediate_compact(

    Ok(wait_task_done)
 }
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
+    use std::sync::Arc;
+    use tracing::{info_span, Instrument};
+
+    use super::{super::harness::TenantHarness, TenantsMap};
+
+    #[tokio::test(start_paused = true)]
+    async fn shutdown_joins_remove_tenant_from_memory() {
+        // the test is a bit ugly with the lockstep together with spawned tasks. the aim is to make
+        // sure `shutdown_all_tenants0` per-tenant processing joins in any active
+        // remove_tenant_from_memory calls, which is enforced by making the operation last until
+        // we've ran `shutdown_all_tenants0` for a long time.
+
+        let (t, _ctx) = TenantHarness::create("shutdown_joins_detach")
+            .unwrap()
+            .load()
+            .await;
+
+        // harness loads it to active, which is forced and nothing is running on the tenant
+
+        let id = t.tenant_id();
+
+        // tenant harness configures the logging and we cannot escape it
+        let _e = info_span!("testing", tenant_id = %id).entered();
+
+        let tenants = HashMap::from([(id, t.clone())]);
+        let tenants = Arc::new(tokio::sync::RwLock::new(TenantsMap::Open(tenants)));
+
+        let (until_cleanup_completed, can_complete_cleanup) = utils::completion::channel();
+        let (until_cleanup_started, cleanup_started) = utils::completion::channel();
+
+        // start a "detaching operation", which will take a while, until can_complete_cleanup
+        let cleanup_task = {
+            let jh = tokio::spawn({
+                let tenants = tenants.clone();
+                async move {
+                    let cleanup = async move {
+                        drop(until_cleanup_started);
+                        can_complete_cleanup.wait().await;
+                        anyhow::Ok(())
+                    };
+                    super::remove_tenant_from_memory(&tenants, id, cleanup).await
+                }
+                .instrument(info_span!("foobar", tenant_id = %id))
+            });
+
+            // now the long cleanup should be in place, with the stopping state
+            cleanup_started.wait().await;
+            jh
+        };
+
+        let mut cleanup_progress = std::pin::pin!(t
+            .shutdown(utils::completion::Barrier::default(), false)
+            .await
+            .unwrap_err()
+            .wait());
+
+        let mut shutdown_task = {
+            let (until_shutdown_started, shutdown_started) = utils::completion::channel();
+
+            let shutdown_task = tokio::spawn(async move {
+                drop(until_shutdown_started);
+                super::shutdown_all_tenants0(&tenants).await;
+            });
+
+            shutdown_started.wait().await;
+            shutdown_task
+        };
+
+        // if the joining in is removed from shutdown_all_tenants0, the shutdown_task should always
+        // get to complete within timeout and fail the test. it is expected to continue awaiting
+        // until completion or SIGKILL during normal shutdown.
+        //
+        // the timeout is long to cover anything that shutdown_task could be doing, but it is
+        // handled instantly because we use tokio's time pausing in this test. 100s is much more than
+        // what we get from systemd on shutdown (10s).
+        let long_time = std::time::Duration::from_secs(100);
+        tokio::select! {
+            _ = &mut shutdown_task => unreachable!("shutdown must continue, until_cleanup_completed is not dropped"),
+            _ = &mut cleanup_progress => unreachable!("cleanup progress must continue, until_cleanup_completed is not dropped"),
+            _ = tokio::time::sleep(long_time) => {},
+        }
+
+        // allow the remove_tenant_from_memory and thus eventually the shutdown to continue
+        drop(until_cleanup_completed);
+
+        let (je, ()) = tokio::join!(shutdown_task, cleanup_progress);
+        je.expect("Tenant::shutdown shutdown not have panicked");
+        cleanup_task
+            .await
+            .expect("no panicking")
+            .expect("remove_tenant_from_memory failed");
+
+        futures::future::poll_immediate(
+            t.shutdown(utils::completion::Barrier::default(), false)
+                .await
+                .unwrap_err()
+                .wait(),
+        )
+        .await
+        .expect("the stopping progress must still be complete");
+    }
+}
--- a/pageserver/src/tenant/storage_layer.rs
+++ b/pageserver/src/tenant/storage_layer.rs
@@ -41,7 +41,7 @@ pub use inmemory_layer::InMemoryLayer;
 pub use layer_desc::{PersistentLayerDesc, PersistentLayerKey};
 pub use remote_layer::RemoteLayer;

-use super::timeline::layer_manager::LayerManager;
+use super::timeline::layer_manager::LayerManagerWriteGuard;

 pub fn range_overlaps<T>(a: &Range<T>, b: &Range<T>) -> bool
 where
@@ -176,7 +176,7 @@ impl LayerAccessStats {
    /// [`LayerLoad`]: LayerResidenceEventReason::LayerLoad
    /// [`record_residence_event`]: Self::record_residence_event
    pub(crate) fn for_loading_layer(
-        layer_map_lock_held_witness: &LayerManager,
+        layer_map_lock_held_witness: &LayerManagerWriteGuard,
        status: LayerResidenceStatus,
    ) -> Self {
        let new = LayerAccessStats(Mutex::new(LayerAccessStatsLocked::default()));
@@ -197,7 +197,7 @@ impl LayerAccessStats {
    /// [`record_residence_event`]: Self::record_residence_event
    pub(crate) fn clone_for_residence_change(
        &self,
-        layer_map_lock_held_witness: &LayerManager,
+        layer_map_lock_held_witness: &LayerManagerWriteGuard,
        new_status: LayerResidenceStatus,
    ) -> LayerAccessStats {
        let clone = {
@@ -229,7 +229,7 @@ impl LayerAccessStats {
    ///
    pub(crate) fn record_residence_event(
        &self,
-        _layer_map_lock_held_witness: &LayerManager,
+        _layer_map_lock_held_witness: &LayerManagerWriteGuard,
        status: LayerResidenceStatus,
        reason: LayerResidenceEventReason,
    ) {
@@ -442,6 +442,10 @@ pub trait PersistentLayer: Layer + AsLayerDesc {
        None
    }

+    fn downcast_delta_layer(self: Arc<Self>) -> Option<std::sync::Arc<DeltaLayer>> {
+        None
+    }
+
    fn is_remote_layer(&self) -> bool {
        false
    }
--- a/pageserver/src/tenant/storage_layer/delta_layer.rs
+++ b/pageserver/src/tenant/storage_layer/delta_layer.rs
@@ -51,6 +51,7 @@ use std::io::{Seek, SeekFrom};
 use std::ops::Range;
 use std::os::unix::fs::FileExt;
 use std::path::{Path, PathBuf};
+use std::sync::Arc;
 use tracing::*;

 use utils::{
@@ -414,6 +415,10 @@ impl AsLayerDesc for DeltaLayer {
 }

 impl PersistentLayer for DeltaLayer {
+    fn downcast_delta_layer(self: Arc<Self>) -> Option<std::sync::Arc<DeltaLayer>> {
+        Some(self)
+    }
+
    fn local_path(&self) -> Option<PathBuf> {
        Some(self.path())
    }
--- a/pageserver/src/tenant/storage_layer/remote_layer.rs
+++ b/pageserver/src/tenant/storage_layer/remote_layer.rs
@@ -6,7 +6,7 @@ use crate::context::RequestContext;
 use crate::repository::Key;
 use crate::tenant::remote_timeline_client::index::LayerFileMetadata;
 use crate::tenant::storage_layer::{Layer, ValueReconstructResult, ValueReconstructState};
-use crate::tenant::timeline::layer_manager::LayerManager;
+use crate::tenant::timeline::layer_manager::LayerManagerWriteGuard;
 use anyhow::{bail, Result};
 use pageserver_api::models::HistoricLayerInfo;
 use std::ops::Range;
@@ -226,7 +226,7 @@ impl RemoteLayer {
    /// Create a Layer struct representing this layer, after it has been downloaded.
    pub fn create_downloaded_layer(
        &self,
-        layer_map_lock_held_witness: &LayerManager,
+        layer_map_lock_held_witness: &LayerManagerWriteGuard,
        conf: &'static PageServerConf,
        file_size: u64,
    ) -> Arc<dyn PersistentLayer> {
--- a/pageserver/src/tenant/tasks.rs
+++ b/pageserver/src/tenant/tasks.rs
@@ -122,12 +122,12 @@ async fn compaction_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
            warn_when_period_overrun(started_at.elapsed(), period, "compaction");

            // Sleep
-            tokio::select! {
-                _ = cancel.cancelled() => {
-                    info!("received cancellation request during idling");
-                    break;
-                },
-                _ = tokio::time::sleep(sleep_duration) => {},
+            if tokio::time::timeout(sleep_duration, cancel.cancelled())
+                .await
+                .is_ok()
+            {
+                info!("received cancellation request during idling");
+                break;
            }
        }
    }
@@ -196,12 +196,12 @@ async fn gc_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
            warn_when_period_overrun(started_at.elapsed(), period, "gc");

            // Sleep
-            tokio::select! {
-                _ = cancel.cancelled() => {
-                    info!("received cancellation request during idling");
-                    break;
-                },
-                _ = tokio::time::sleep(sleep_duration) => {},
+            if tokio::time::timeout(sleep_duration, cancel.cancelled())
+                .await
+                .is_ok()
+            {
+                info!("received cancellation request during idling");
+                break;
            }
        }
    }
@@ -263,9 +263,9 @@ pub(crate) async fn random_init_delay(
        rng.gen_range(Duration::ZERO..=period)
    };

-    tokio::select! {
-        _ = cancel.cancelled() => Err(Cancelled),
-        _ = tokio::time::sleep(d) => Ok(()),
+    match tokio::time::timeout(d, cancel.cancelled()).await {
+        Ok(_) => Err(Cancelled),
+        Err(_) => Ok(()),
    }
 }

--- a/pageserver/src/tenant/timeline.rs
+++ b/pageserver/src/tenant/timeline.rs
@@ -24,7 +24,7 @@ use tracing::*;
 use utils::id::TenantTimelineId;

 use std::cmp::{max, min, Ordering};
-use std::collections::{BinaryHeap, HashMap};
+use std::collections::{BinaryHeap, HashMap, HashSet};
 use std::fs;
 use std::ops::{Deref, Range};
 use std::path::{Path, PathBuf};
@@ -81,7 +81,7 @@ use crate::{is_temporary, task_mgr};

 pub(super) use self::eviction_task::EvictionTaskTenantState;
 use self::eviction_task::EvictionTaskTimelineState;
-use self::layer_manager::LayerManager;
+use self::layer_manager::{LayerManager, LayerManagerReadGuard, LayerManagerWriteGuard};
 use self::logical_size::LogicalSize;
 use self::walreceiver::{WalReceiver, WalReceiverConf};

@@ -125,13 +125,13 @@ impl PartialOrd for Hole {

 /// Temporary function for immutable storage state refactor, ensures we are dropping mutex guard instead of other things.
 /// Can be removed after all refactors are done.
-fn drop_rlock<T>(rlock: tokio::sync::OwnedRwLockReadGuard<T>) {
+fn drop_rlock(rlock: LayerManagerReadGuard) {
    drop(rlock)
 }

 /// Temporary function for immutable storage state refactor, ensures we are dropping mutex guard instead of other things.
 /// Can be removed after all refactors are done.
-fn drop_wlock<T>(rlock: tokio::sync::RwLockWriteGuard<'_, T>) {
+fn drop_wlock(rlock: LayerManagerWriteGuard) {
    drop(rlock)
 }
 pub struct Timeline {
@@ -162,7 +162,7 @@ pub struct Timeline {
    ///
    /// In the future, we'll be able to split up the tuple of LayerMap and `LayerFileManager`,
    /// so that e.g. on-demand-download/eviction, and layer spreading, can operate just on `LayerFileManager`.
-    pub(crate) layers: Arc<tokio::sync::RwLock<LayerManager>>,
+    pub(crate) layers: Arc<LayerManager>,

    /// Set of key ranges which should be covered by image layers to
    /// allow GC to remove old layers. This set is created by GC and its cutoff LSN is also stored.
@@ -1011,11 +1011,11 @@ impl Timeline {
            .evict_layer_batch(remote_client, &[local_layer], cancel)
            .await?;
        assert_eq!(results.len(), 1);
-        let result: Option<anyhow::Result<bool>> = results.into_iter().next().unwrap();
+        let result: Option<Result<(), EvictionError>> = results.into_iter().next().unwrap();
        match result {
            None => anyhow::bail!("task_mgr shutdown requested"),
-            Some(Ok(b)) => Ok(Some(b)),
-            Some(Err(e)) => Err(e),
+            Some(Ok(())) => Ok(Some(true)),
+            Some(Err(e)) => Err(anyhow::Error::new(e)),
        }
    }

@@ -1024,12 +1024,12 @@ impl Timeline {
    /// GenericRemoteStorage reference is required as a (witness)[witness_article] for "remote storage is configured."
    ///
    /// [witness_article]: https://willcrichton.net/rust-api-type-patterns/witnesses.html
-    pub async fn evict_layers(
+    pub(crate) async fn evict_layers(
        &self,
        _: &GenericRemoteStorage,
        layers_to_evict: &[Arc<dyn PersistentLayer>],
        cancel: CancellationToken,
-    ) -> anyhow::Result<Vec<Option<anyhow::Result<bool>>>> {
+    ) -> anyhow::Result<Vec<Option<Result<(), EvictionError>>>> {
        let remote_client = self.remote_client.clone().expect(
            "GenericRemoteStorage is configured, so timeline must have RemoteTimelineClient",
        );
@@ -1064,7 +1064,7 @@ impl Timeline {
        remote_client: &Arc<RemoteTimelineClient>,
        layers_to_evict: &[Arc<dyn PersistentLayer>],
        cancel: CancellationToken,
-    ) -> anyhow::Result<Vec<Option<anyhow::Result<bool>>>> {
+    ) -> anyhow::Result<Vec<Option<Result<(), EvictionError>>>> {
        // ensure that the layers have finished uploading
        // (don't hold the layer_removal_cs while we do it, we're not removing anything yet)
        remote_client
@@ -1109,12 +1109,10 @@ impl Timeline {
        &self,
        _layer_removal_cs: &tokio::sync::MutexGuard<'_, ()>,
        local_layer: &Arc<dyn PersistentLayer>,
-        layer_mgr: &mut LayerManager,
-    ) -> anyhow::Result<bool> {
+        layer_mgr: &mut LayerManagerWriteGuard,
+    ) -> Result<(), EvictionError> {
        if local_layer.is_remote_layer() {
-            // TODO(issue #3851): consider returning an err here instead of false,
-            // which is the same out the match later
-            return Ok(false);
+            return Err(EvictionError::CannotEvictRemoteLayer);
        }

        let layer_file_size = local_layer.file_size();
@@ -1123,13 +1121,22 @@ impl Timeline {
            .local_path()
            .expect("local layer should have a local path")
            .metadata()
-            .context("get local layer file stat")?
+            // when the eviction fails because we have already deleted the layer in compaction for
+            // example, a NotFound error bubbles up from here.
+            .map_err(|e| {
+                if e.kind() == std::io::ErrorKind::NotFound {
+                    EvictionError::FileNotFound
+                } else {
+                    EvictionError::StatFailed(e)
+                }
+            })?
            .modified()
-            .context("get mtime of layer file")?;
+            .map_err(EvictionError::StatFailed)?;
+
        let local_layer_residence_duration =
            match SystemTime::now().duration_since(local_layer_mtime) {
                Err(e) => {
-                    warn!("layer mtime is in the future: {}", e);
+                    warn!(layer = %local_layer, "layer mtime is in the future: {}", e);
                    None
                }
                Ok(delta) => Some(delta),
@@ -1160,54 +1167,65 @@ impl Timeline {

        assert_eq!(local_layer.layer_desc(), new_remote_layer.layer_desc());

-        let succeed = match layer_mgr.replace_and_verify(local_layer.clone(), new_remote_layer) {
-            Ok(()) => {
-                if let Err(e) = local_layer.delete_resident_layer_file() {
-                    error!("failed to remove layer file on evict after replacement: {e:#?}");
-                }
-                // Always decrement the physical size gauge, even if we failed to delete the file.
-                // Rationale: we already replaced the layer with a remote layer in the layer map,
-                // and any subsequent download_remote_layer will
-                // 1. overwrite the file on disk and
-                // 2. add the downloaded size to the resident size gauge.
-                //
-                // If there is no re-download, and we restart the pageserver, then load_layer_map
-                // will treat the file as a local layer again, count it towards resident size,
-                // and it'll be like the layer removal never happened.
-                // The bump in resident size is perhaps unexpected but overall a robust behavior.
-                self.metrics
-                    .resident_physical_size_gauge
-                    .sub(layer_file_size);
+        layer_mgr
+            .replace_and_verify(local_layer.clone(), new_remote_layer)
+            .map_err(EvictionError::LayerNotFound)?;

-                self.metrics.evictions.inc();
+        if let Err(e) = local_layer.delete_resident_layer_file() {
+            // this should never happen, because of layer_removal_cs usage and above stat
+            // access for mtime
+            error!("failed to remove layer file on evict after replacement: {e:#?}");
+        }
+        // Always decrement the physical size gauge, even if we failed to delete the file.
+        // Rationale: we already replaced the layer with a remote layer in the layer map,
+        // and any subsequent download_remote_layer will
+        // 1. overwrite the file on disk and
+        // 2. add the downloaded size to the resident size gauge.
+        //
+        // If there is no re-download, and we restart the pageserver, then load_layer_map
+        // will treat the file as a local layer again, count it towards resident size,
+        // and it'll be like the layer removal never happened.
+        // The bump in resident size is perhaps unexpected but overall a robust behavior.
+        self.metrics
+            .resident_physical_size_gauge
+            .sub(layer_file_size);

-                if let Some(delta) = local_layer_residence_duration {
-                    self.metrics
-                        .evictions_with_low_residence_duration
-                        .read()
-                        .unwrap()
-                        .observe(delta);
-                    info!(layer=%local_layer, residence_millis=delta.as_millis(), "evicted layer after known residence period");
-                } else {
-                    info!(layer=%local_layer, "evicted layer after unknown residence period");
-                }
+        self.metrics.evictions.inc();

-                true
-            }
-            Err(err) => {
-                if cfg!(debug_assertions) {
-                    panic!("failed to replace: {err}, evicted: {local_layer:?}");
-                } else {
-                    error!(evicted=?local_layer, "failed to replace: {err}");
-                }
-                false
-            }
-        };
+        if let Some(delta) = local_layer_residence_duration {
+            self.metrics
+                .evictions_with_low_residence_duration
+                .read()
+                .unwrap()
+                .observe(delta);
+            info!(layer=%local_layer, residence_millis=delta.as_millis(), "evicted layer after known residence period");
+        } else {
+            info!(layer=%local_layer, "evicted layer after unknown residence period");
+        }

-        Ok(succeed)
+        Ok(())
    }
 }

+#[derive(Debug, thiserror::Error)]
+pub(crate) enum EvictionError {
+    #[error("cannot evict a remote layer")]
+    CannotEvictRemoteLayer,
+    /// Most likely the to-be evicted layer has been deleted by compaction or gc which use the same
+    /// locks, so they got to execute before the eviction.
+    #[error("file backing the layer has been removed already")]
+    FileNotFound,
+    #[error("stat failed")]
+    StatFailed(#[source] std::io::Error),
+    /// In practice, this can be a number of things, but lets assume it means only this.
+    ///
+    /// This case includes situations such as the Layer was evicted and redownloaded in between,
+    /// because the file existed before an replacement attempt was made but now the Layers are
+    /// different objects in memory.
+    #[error("layer was no longer part of LayerMap")]
+    LayerNotFound(#[source] anyhow::Error),
+}
+
 /// Number of times we will compute partition within a checkpoint distance.
 const REPARTITION_FREQ_IN_CHECKPOINT_DISTANCE: u64 = 10;

@@ -1331,7 +1349,7 @@ impl Timeline {
                timeline_id,
                tenant_id,
                pg_version,
-                layers: Arc::new(tokio::sync::RwLock::new(LayerManager::create())),
+                layers: Arc::new(LayerManager::create()),
                wanted_image_layers: Mutex::new(None),

                walredo_mgr,
@@ -2543,13 +2561,15 @@ impl Timeline {
    ///
    async fn get_layer_for_write(&self, lsn: Lsn) -> anyhow::Result<Arc<InMemoryLayer>> {
        let mut guard = self.layers.write().await;
-        let layer = guard.get_layer_for_write(
-            lsn,
-            self.get_last_record_lsn(),
-            self.conf,
-            self.timeline_id,
-            self.tenant_id,
-        )?;
+        let layer = guard
+            .get_layer_for_write(
+                lsn,
+                self.get_last_record_lsn(),
+                self.conf,
+                self.timeline_id,
+                self.tenant_id,
+            )
+            .await?;
        Ok(layer)
    }

@@ -2582,7 +2602,9 @@ impl Timeline {
            Some(self.write_lock.lock().await)
        };
        let mut guard = self.layers.write().await;
-        guard.try_freeze_in_memory_layer(self.get_last_record_lsn(), &self.last_freeze_at);
+        guard
+            .try_freeze_in_memory_layer(self.get_last_record_lsn(), &self.last_freeze_at)
+            .await;
    }

    /// Layer flusher task's main loop.
@@ -2685,7 +2707,7 @@ impl Timeline {
        // files instead. This is possible as long as *all* the data imported into the
        // repository have the same LSN.
        let lsn_range = frozen_layer.get_lsn_range();
-        let layer_paths_to_upload =
+        let (layer_paths_to_upload, delta_layer_to_add) =
            if lsn_range.start == self.initdb_lsn && lsn_range.end == Lsn(self.initdb_lsn.0 + 1) {
                #[cfg(test)]
                match &mut *self.flush_loop_state.lock().unwrap() {
@@ -2704,8 +2726,12 @@ impl Timeline {
                let (partitioning, _lsn) = self
                    .repartition(self.initdb_lsn, self.get_compaction_target_size(), ctx)
                    .await?;
-                self.create_image_layers(&partitioning, self.initdb_lsn, true, ctx)
-                    .await?
+                // For image layers, we add them immediately into the layer map.
+                (
+                    self.create_image_layers(&partitioning, self.initdb_lsn, true, ctx)
+                        .await?,
+                    None,
+                )
            } else {
                #[cfg(test)]
                match &mut *self.flush_loop_state.lock().unwrap() {
@@ -2719,35 +2745,52 @@ impl Timeline {
                        assert!(!*expect_initdb_optimization, "expected initdb optimization");
                    }
                }
-                // normal case, write out a L0 delta layer file.
-                let (delta_path, metadata) = self.create_delta_layer(&frozen_layer).await?;
-                HashMap::from([(delta_path, metadata)])
+                // Normal case, write out a L0 delta layer file.
+                // `create_delta_layer` will not modify the layer map.
+                // We will remove frozen layer and add delta layer in one atomic operation later.
+                let layer = self.create_delta_layer(&frozen_layer).await?;
+                (
+                    HashMap::from([(layer.filename(), LayerFileMetadata::new(layer.file_size()))]),
+                    Some(layer),
+                )
            };

-        // FIXME: between create_delta_layer and the scheduling of the upload in `update_metadata_file`,
-        // a compaction can delete the file and then it won't be available for uploads any more.
-        // We still schedule the upload, resulting in an error, but ideally we'd somehow avoid this
-        // race situation.
-        // See https://github.com/neondatabase/neon/issues/4526
-
-        pausable_failpoint!("flush-frozen-before-sync");
-
        // The new on-disk layers are now in the layer map. We can remove the
        // in-memory layer from the map now. The flushed layer is stored in
        // the mapping in `create_delta_layer`.
        {
            let mut guard = self.layers.write().await;
-            let l = guard.layer_map_mut().frozen_layers.pop_front();

-            // Only one thread may call this function at a time (for this
-            // timeline). If two threads tried to flush the same frozen
-            // layer to disk at the same time, that would not work.
-            assert!(compare_arced_layers(&l.unwrap(), &frozen_layer));
+            if let Some(ref l) = delta_layer_to_add {
+                // TODO: move access stats, metrics update, etc. into layer manager.
+                l.access_stats().record_residence_event(
+                    &guard,
+                    LayerResidenceStatus::Resident,
+                    LayerResidenceEventReason::LayerCreate,
+                );

+                // update metrics
+                let sz = l.file_size();
+                self.metrics.resident_physical_size_gauge.add(sz);
+                self.metrics.num_persistent_files_created.inc_by(1);
+                self.metrics.persistent_bytes_written.inc_by(sz);
+            }
+
+            guard
+                .finish_flush_l0_layer(delta_layer_to_add, &frozen_layer)
+                .await;
            // release lock on 'layers'
        }

-        fail_point!("checkpoint-after-sync");
+        // FIXME: between create_delta_layer and the scheduling of the upload in `update_metadata_file`,
+        // a compaction can delete the file and then it won't be available for uploads any more.
+        // We still schedule the upload, resulting in an error, but ideally we'd somehow avoid this
+        // race situation.
+        // See https://github.com/neondatabase/neon/issues/4526
+        pausable_failpoint!("flush-frozen-pausable");
+
+        // This failpoint is used by another test case `test_pageserver_recovery`.
+        fail_point!("flush-frozen-exit");

        // Update the metadata file, with new 'disk_consistent_lsn'
        //
@@ -2829,11 +2872,12 @@ impl Timeline {
        Ok(())
    }

-    // Write out the given frozen in-memory layer as a new L0 delta file
+    // Write out the given frozen in-memory layer as a new L0 delta file. This L0 file will not be tracked
+    // in layer map immediately. The caller is responsible to put it into the layer map.
    async fn create_delta_layer(
        self: &Arc<Self>,
        frozen_layer: &Arc<InMemoryLayer>,
-    ) -> anyhow::Result<(LayerFileName, LayerFileMetadata)> {
+    ) -> anyhow::Result<DeltaLayer> {
        let span = tracing::info_span!("blocking");
        let new_delta: DeltaLayer = tokio::task::spawn_blocking({
            let _g = span.entered();
@@ -2870,25 +2914,8 @@ impl Timeline {
        })
        .await
        .context("spawn_blocking")??;
-        let new_delta_name = new_delta.filename();
-        let sz = new_delta.desc.file_size;

-        // Add it to the layer map
-        let l = Arc::new(new_delta);
-        let mut guard = self.layers.write().await;
-        l.access_stats().record_residence_event(
-            &guard,
-            LayerResidenceStatus::Resident,
-            LayerResidenceEventReason::LayerCreate,
-        );
-        guard.track_new_l0_delta_layer(l);
-
-        // update metrics
-        self.metrics.resident_physical_size_gauge.add(sz);
-        self.metrics.num_persistent_files_created.inc_by(1);
-        self.metrics.persistent_bytes_written.inc_by(sz);
-
-        Ok((new_delta_name, LayerFileMetadata::new(sz)))
+        Ok(new_delta)
    }

    async fn repartition(
@@ -3130,7 +3157,7 @@ impl Timeline {
                LayerResidenceEventReason::LayerCreate,
            );
        }
-        guard.track_new_image_layers(image_layers);
+        guard.track_new_image_layers(image_layers).await;
        drop_wlock(guard);
        timer.stop_and_record();

@@ -3140,7 +3167,7 @@ impl Timeline {

 #[derive(Default)]
 struct CompactLevel0Phase1Result {
-    new_layers: Vec<DeltaLayer>,
+    new_layers: Vec<Arc<DeltaLayer>>,
    deltas_to_compact: Vec<Arc<PersistentLayerDesc>>,
 }

@@ -3294,7 +3321,7 @@ impl Timeline {
    fn compact_level0_phase1(
        self: Arc<Self>,
        _layer_removal_cs: Arc<tokio::sync::OwnedMutexGuard<()>>,
-        guard: tokio::sync::OwnedRwLockReadGuard<LayerManager>,
+        guard: LayerManagerReadGuard,
        mut stats: CompactLevel0Phase1StatsBuilder,
        target_file_size: u64,
        ctx: &RequestContext,
@@ -3318,6 +3345,37 @@ impl Timeline {
            return Ok(CompactLevel0Phase1Result::default());
        }

+        // This failpoint is used together with `test_duplicate_layers` integration test.
+        // It returns the compaction result exactly the same layers as input to compaction.
+        // We want to ensure that this will not cause any problem when updating the layer map
+        // after the compaction is finished.
+        //
+        // Currently, there are two rare edge cases that will cause duplicated layers being
+        // inserted.
+        // 1. The compaction job is inturrupted / did not finish successfully. Assume we have file 1, 2, 3, 4, which
+        //    is compacted to 5, but the page server is shut down, next time we start page server we will get a layer
+        //    map containing 1, 2, 3, 4, and 5, whereas 5 has the same content as 4. If we trigger L0 compation at this
+        //    point again, it is likely that we will get a file 6 which has the same content and the key range as 5,
+        //    and this causes an overwrite. This is acceptable because the content is the same, and we should do a
+        //    layer replace instead of the normal remove / upload process.
+        // 2. The input workload pattern creates exactly n files that are sorted, non-overlapping and is of target file
+        //    size length. Compaction will likely create the same set of n files afterwards.
+        //
+        // This failpoint is a superset of both of the cases.
+        fail_point!("compact-level0-phase1-return-same", |_| {
+            println!("compact-level0-phase1-return-same"); // so that we can check if we hit the failpoint
+            Ok(CompactLevel0Phase1Result {
+                new_layers: level0_deltas
+                    .iter()
+                    .map(|x| x.clone().downcast_delta_layer().unwrap())
+                    .collect(),
+                deltas_to_compact: level0_deltas
+                    .iter()
+                    .map(|x| x.layer_desc().clone().into())
+                    .collect(),
+            })
+        });
+
        // Gather the files to compact in this iteration.
        //
        // Start with the oldest Level 0 delta file, and collect any other
@@ -3576,7 +3634,9 @@ impl Timeline {
                        || contains_hole
                    {
                        // ... if so, flush previous layer and prepare to write new one
-                        new_layers.push(writer.take().unwrap().finish(prev_key.unwrap().next())?);
+                        new_layers.push(Arc::new(
+                            writer.take().unwrap().finish(prev_key.unwrap().next())?,
+                        ));
                        writer = None;

                        if contains_hole {
@@ -3614,7 +3674,7 @@ impl Timeline {
            prev_key = Some(key);
        }
        if let Some(writer) = writer {
-            new_layers.push(writer.finish(prev_key.unwrap().next())?);
+            new_layers.push(Arc::new(writer.finish(prev_key.unwrap().next())?));
        }

        // Sync layers
@@ -3685,7 +3745,7 @@ impl Timeline {
            };

            let begin = tokio::time::Instant::now();
-            let phase1_layers_locked = Arc::clone(&self.layers).read_owned().await;
+            let phase1_layers_locked = self.layers.read().await;
            let now = tokio::time::Instant::now();
            stats.read_lock_acquisition_micros =
                DurationRecorder::Recorded(RecordedDuration(now - begin), now);
@@ -3720,9 +3780,14 @@ impl Timeline {
                .context("wait for layer upload ops to complete")?;
        }

-        let mut guard = self.layers.write().await;
+        let guard = self.layers.modify().await;
        let mut new_layer_paths = HashMap::with_capacity(new_layers.len());

+        // In some rare cases, we may generate a file with exactly the same key range / LSN as before the compaction.
+        // We should move to numbering the layer files instead of naming them using key range / LSN some day. But for
+        // now, we just skip the file to avoid unintentional modification to files on the disk and in the layer map.
+        let mut duplicated_layers = HashSet::new();
+
        let mut insert_layers = Vec::new();
        let mut remove_layers = Vec::new();

@@ -3749,31 +3814,43 @@ impl Timeline {
                .add(metadata.len());

            new_layer_paths.insert(new_delta_path, LayerFileMetadata::new(metadata.len()));
-            let x: Arc<dyn PersistentLayer + 'static> = Arc::new(l);
-            x.access_stats().record_residence_event(
+            l.access_stats().record_residence_event(
                &guard,
                LayerResidenceStatus::Resident,
                LayerResidenceEventReason::LayerCreate,
            );
-            insert_layers.push(x);
+            let l = l as Arc<dyn PersistentLayer>;
+            if guard.contains(&l) {
+                duplicated_layers.insert(l.layer_desc().key());
+            } else {
+                if LayerMap::is_l0(l.layer_desc()) {
+                    return Err(CompactionError::Other(anyhow!("compaction generates a L0 layer file as output, which will cause infinite compaction.")));
+                }
+                insert_layers.push(l);
+            }
        }

        // Now that we have reshuffled the data to set of new delta layers, we can
        // delete the old ones
        let mut layer_names_to_delete = Vec::with_capacity(deltas_to_compact.len());
-        for l in deltas_to_compact {
-            layer_names_to_delete.push(l.filename());
-            remove_layers.push(guard.get_from_desc(&l));
+        for ldesc in deltas_to_compact {
+            if duplicated_layers.contains(&ldesc.key()) {
+                // skip duplicated layers, they will not be removed; we have already overwritten them
+                // with new layers in the compaction phase 1.
+                continue;
+            }
+            layer_names_to_delete.push(ldesc.filename());
+            remove_layers.push(guard.get_from_desc(&ldesc));
        }

-        guard.finish_compact_l0(
-            layer_removal_cs,
-            remove_layers,
-            insert_layers,
-            &self.metrics,
-        )?;
-
-        drop_wlock(guard);
+        guard
+            .finish_compact_l0_consume_guard(
+                layer_removal_cs,
+                remove_layers,
+                insert_layers,
+                &self.metrics,
+            )
+            .await?;

        // Also schedule the deletions in remote storage
        if let Some(remote_client) = &self.remote_client {
@@ -4104,7 +4181,9 @@ impl Timeline {
                layer_names_to_delete.push(doomed_layer.filename());
                result.layers_removed += 1;
            }
-            let apply = guard.finish_gc_timeline(layer_removal_cs, gc_layers, &self.metrics)?;
+            guard
+                .finish_gc_timeline(layer_removal_cs, gc_layers, &self.metrics)
+                .await?;

            if result.layers_removed != 0 {
                fail_point!("after-timeline-gc-removed-layers");
@@ -4113,8 +4192,6 @@ impl Timeline {
            if let Some(remote_client) = &self.remote_client {
                remote_client.schedule_layer_file_deletion(&layer_names_to_delete)?;
            }
-
-            apply.flush();
        }

        info!(
@@ -4522,6 +4599,7 @@ impl LocalLayerInfoForDiskUsageEviction {
 }

 impl Timeline {
+    /// Returns non-remote layers for eviction.
    pub(crate) async fn get_local_layers_for_disk_usage_eviction(&self) -> DiskUsageEvictionInfo {
        let guard = self.layers.read().await;
        let layers = guard.layer_map();
@@ -4691,3 +4769,179 @@ pub fn compare_arced_layers<L: ?Sized>(left: &Arc<L>, right: &Arc<L>) -> bool {

    left == right
 }
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use utils::{id::TimelineId, lsn::Lsn};
+
+    use crate::tenant::{harness::TenantHarness, storage_layer::PersistentLayer};
+
+    use super::{EvictionError, Timeline};
+
+    #[tokio::test]
+    async fn two_layer_eviction_attempts_at_the_same_time() {
+        let harness =
+            TenantHarness::create("two_layer_eviction_attempts_at_the_same_time").unwrap();
+
+        let remote_storage = {
+            // this is never used for anything, because of how the create_test_timeline works, but
+            // it is with us in spirit and a Some.
+            use remote_storage::{GenericRemoteStorage, RemoteStorageConfig, RemoteStorageKind};
+            let path = harness.conf.workdir.join("localfs");
+            std::fs::create_dir_all(&path).unwrap();
+            let config = RemoteStorageConfig {
+                max_concurrent_syncs: std::num::NonZeroUsize::new(2_000_000).unwrap(),
+                max_sync_errors: std::num::NonZeroU32::new(3_000_000).unwrap(),
+                storage: RemoteStorageKind::LocalFs(path),
+            };
+            GenericRemoteStorage::from_config(&config).unwrap()
+        };
+
+        let ctx = any_context();
+        let tenant = harness.try_load(&ctx, Some(remote_storage)).await.unwrap();
+        let timeline = tenant
+            .create_test_timeline(TimelineId::generate(), Lsn(0x10), 14, &ctx)
+            .await
+            .unwrap();
+
+        let rc = timeline
+            .remote_client
+            .clone()
+            .expect("just configured this");
+
+        let layer = find_some_layer(&timeline).await;
+
+        let cancel = tokio_util::sync::CancellationToken::new();
+        let batch = [layer];
+
+        let first = {
+            let cancel = cancel.clone();
+            async {
+                timeline
+                    .evict_layer_batch(&rc, &batch, cancel)
+                    .await
+                    .unwrap()
+            }
+        };
+        let second = async {
+            timeline
+                .evict_layer_batch(&rc, &batch, cancel)
+                .await
+                .unwrap()
+        };
+
+        let (first, second) = tokio::join!(first, second);
+
+        let (first, second) = (only_one(first), only_one(second));
+
+        match (first, second) {
+            (Ok(()), Err(EvictionError::FileNotFound))
+            | (Err(EvictionError::FileNotFound), Ok(())) => {
+                // one of the evictions gets to do it,
+                // other one gets FileNotFound. all is good.
+            }
+            other => unreachable!("unexpected {:?}", other),
+        }
+    }
+
+    #[tokio::test]
+    async fn layer_eviction_aba_fails() {
+        let harness = TenantHarness::create("layer_eviction_aba_fails").unwrap();
+
+        let remote_storage = {
+            // this is never used for anything, because of how the create_test_timeline works, but
+            // it is with us in spirit and a Some.
+            use remote_storage::{GenericRemoteStorage, RemoteStorageConfig, RemoteStorageKind};
+            let path = harness.conf.workdir.join("localfs");
+            std::fs::create_dir_all(&path).unwrap();
+            let config = RemoteStorageConfig {
+                max_concurrent_syncs: std::num::NonZeroUsize::new(2_000_000).unwrap(),
+                max_sync_errors: std::num::NonZeroU32::new(3_000_000).unwrap(),
+                storage: RemoteStorageKind::LocalFs(path),
+            };
+            GenericRemoteStorage::from_config(&config).unwrap()
+        };
+
+        let ctx = any_context();
+        let tenant = harness.try_load(&ctx, Some(remote_storage)).await.unwrap();
+        let timeline = tenant
+            .create_test_timeline(TimelineId::generate(), Lsn(0x10), 14, &ctx)
+            .await
+            .unwrap();
+
+        let _e = tracing::info_span!("foobar", tenant_id = %tenant.tenant_id, timeline_id = %timeline.timeline_id).entered();
+
+        let rc = timeline.remote_client.clone().unwrap();
+
+        // TenantHarness allows uploads to happen given GenericRemoteStorage is configured
+        let layer = find_some_layer(&timeline).await;
+
+        let cancel = tokio_util::sync::CancellationToken::new();
+        let batch = [layer];
+
+        let first = {
+            let cancel = cancel.clone();
+            async {
+                timeline
+                    .evict_layer_batch(&rc, &batch, cancel)
+                    .await
+                    .unwrap()
+            }
+        };
+
+        // lets imagine this is stuck somehow, still referencing the original `Arc<dyn PersistentLayer>`
+        let second = {
+            let cancel = cancel.clone();
+            async {
+                timeline
+                    .evict_layer_batch(&rc, &batch, cancel)
+                    .await
+                    .unwrap()
+            }
+        };
+
+        // while it's stuck, we evict and end up redownloading it
+        only_one(first.await).expect("eviction succeeded");
+
+        let layer = find_some_layer(&timeline).await;
+        let layer = layer.downcast_remote_layer().unwrap();
+        timeline.download_remote_layer(layer).await.unwrap();
+
+        let res = only_one(second.await);
+
+        assert!(
+            matches!(res, Err(EvictionError::LayerNotFound(_))),
+            "{res:?}"
+        );
+
+        // no more specific asserting, outside of preconds this is the only valid replacement
+        // failure
+    }
+
+    fn any_context() -> crate::context::RequestContext {
+        use crate::context::*;
+        use crate::task_mgr::*;
+        RequestContext::new(TaskKind::UnitTest, DownloadBehavior::Error)
+    }
+
+    fn only_one<T>(mut input: Vec<Option<T>>) -> T {
+        assert_eq!(1, input.len());
+        input
+            .pop()
+            .expect("length just checked")
+            .expect("no cancellation")
+    }
+
+    async fn find_some_layer(timeline: &Timeline) -> Arc<dyn PersistentLayer> {
+        let layers = timeline.layers.read().await;
+        let desc = layers
+            .layer_map()
+            .iter_historic_layers()
+            .next()
+            .expect("must find one layer to evict");
+
+        layers.get_from_desc(&desc)
+    }
+}
--- a/pageserver/src/tenant/timeline/eviction_task.rs
+++ b/pageserver/src/tenant/timeline/eviction_task.rs
@@ -30,6 +30,7 @@ use crate::{
    tenant::{
        config::{EvictionPolicy, EvictionPolicyLayerAccessThreshold},
        storage_layer::PersistentLayer,
+        timeline::EvictionError,
        LogicalSizeCalculationCause, Tenant,
    },
 };
@@ -100,11 +101,11 @@ impl Timeline {
            match cf {
                ControlFlow::Break(()) => break,
                ControlFlow::Continue(sleep_until) => {
-                    tokio::select! {
-                        _ = cancel.cancelled() => {
-                            break;
-                        }
-                        _ = tokio::time::sleep_until(sleep_until) => { }
+                    if tokio::time::timeout_at(sleep_until, cancel.cancelled())
+                        .await
+                        .is_ok()
+                    {
+                        break;
                    }
                }
            }
@@ -270,20 +271,22 @@ impl Timeline {
                None => {
                    stats.skipped_for_shutdown += 1;
                }
-                Some(Ok(true)) => {
-                    debug!("evicted layer {l:?}");
+                Some(Ok(())) => {
                    stats.evicted += 1;
                }
-                Some(Ok(false)) => {
-                    debug!("layer is not evictable: {l:?}");
+                Some(Err(EvictionError::CannotEvictRemoteLayer)) => {
                    stats.not_evictable += 1;
                }
-                Some(Err(e)) => {
-                    // This variant is the case where an unexpected error happened during eviction.
-                    // Expected errors that result in non-eviction are `Some(Ok(false))`.
-                    // So, dump Debug here to gather as much info as possible in this rare case.
-                    warn!("failed to evict layer {l:?}: {e:?}");
-                    stats.errors += 1;
+                Some(Err(EvictionError::FileNotFound)) => {
+                    // compaction/gc removed the file while we were waiting on layer_removal_cs
+                    stats.not_evictable += 1;
+                }
+                Some(Err(
+                    e @ EvictionError::LayerNotFound(_) | e @ EvictionError::StatFailed(_),
+                )) => {
+                    let e = utils::error::report_compact_sources(&e);
+                    warn!(layer = %l, "failed to evict layer: {e}");
+                    stats.not_evictable += 1;
                }
            }
        }
--- a/pageserver/src/tenant/timeline/layer_manager.rs
+++ b/pageserver/src/tenant/timeline/layer_manager.rs
@@ -1,6 +1,9 @@
 use anyhow::{bail, ensure, Context, Result};
-use std::{collections::HashMap, sync::Arc};
-use tracing::trace;
+use arc_swap::ArcSwap;
+use dashmap::DashMap;
+use either::Either;
+use std::sync::Arc;
+use tracing::{log::warn, trace};
 use utils::{
    id::{TenantId, TimelineId},
    lsn::{AtomicLsn, Lsn},
@@ -21,94 +24,263 @@ use crate::{

 /// Provides semantic APIs to manipulate the layer map.
 pub struct LayerManager {
-    layer_map: LayerMap,
-    layer_fmgr: LayerFileManager,
+    /// The layer map that tracks all layers in the timeline at the current time.
+    layer_map: ArcSwap<LayerMap>,
+    /// Ensure there is only one thread that can modify the layer map at a time.
+    state_lock: Arc<tokio::sync::Mutex<()>>,
+    /// Layer file manager that manages the layer files in the local / remote file system.
+    layer_fmgr: Arc<LayerFileManager>,
+    /// The lock to mock the original behavior of `RwLock<LayerMap>`. Can be removed if
+    /// #4509 is implemented.
+    pseudo_lock: Arc<tokio::sync::RwLock<()>>,
 }

-/// After GC, the layer map changes will not be applied immediately. Users should manually apply the changes after
-/// scheduling deletes in remote client.
-pub struct ApplyGcResultGuard<'a>(BatchedUpdates<'a>);
+struct LayerSnapshot {
+    /// The current snapshot of the layer map. Immutable.
+    layer_map: Arc<LayerMap>,
+    /// Reference to the file manager. This is mutable and the content might change when
+    /// the snapshot is held.
+    layer_fmgr: Arc<LayerFileManager>,
+}

-impl ApplyGcResultGuard<'_> {
-    pub fn flush(self) {
-        self.0.flush();
-    }
+pub struct LayerManagerReadGuard {
+    snapshot: LayerSnapshot,
+    /// Mock the behavior of the layer map lock.
+    #[allow(dead_code)]
+    pseudo_lock: tokio::sync::OwnedRwLockReadGuard<()>,
+}
+
+pub struct LayerManagerWriteGuard {
+    snapshot: LayerSnapshot,
+    /// Semantic layer operations will need to modify the layer content.
+    layer_manager: Arc<LayerManager>,
+    /// Mock the behavior of the layer map lock.
+    #[allow(dead_code)]
+    pseudo_lock:
+        Either<tokio::sync::OwnedRwLockWriteGuard<()>, tokio::sync::OwnedRwLockReadGuard<()>>,
 }

 impl LayerManager {
    pub fn create() -> Self {
        Self {
-            layer_map: LayerMap::default(),
-            layer_fmgr: LayerFileManager::new(),
+            layer_map: ArcSwap::from(Arc::new(LayerMap::default())),
+            state_lock: Arc::new(tokio::sync::Mutex::new(())),
+            layer_fmgr: Arc::new(LayerFileManager::new()),
+            pseudo_lock: Arc::new(tokio::sync::RwLock::new(())),
        }
    }

-    pub fn get_from_desc(&self, desc: &PersistentLayerDesc) -> Arc<dyn PersistentLayer> {
+    /// Take the snapshot of the layer map and return a read guard. The read guard will prevent
+    /// the layer map from being modified.
+    pub async fn read(&self) -> LayerManagerReadGuard {
+        // take the lock before taking snapshot
+        let pseudo_lock = self.pseudo_lock.clone().read_owned().await;
+        LayerManagerReadGuard {
+            snapshot: LayerSnapshot {
+                layer_map: self.layer_map.load_full(),
+                layer_fmgr: Arc::clone(&self.layer_fmgr),
+            },
+            pseudo_lock,
+        }
+    }
+
+    /// Take the snapshot of the layer map and return a write guard. The read guard will prevent
+    /// the layer map from being read.
+    pub async fn write(self: &Arc<Self>) -> LayerManagerWriteGuard {
+        // take the lock before taking snapshot
+        let pseudo_lock = self.pseudo_lock.clone().write_owned().await;
+        LayerManagerWriteGuard {
+            snapshot: LayerSnapshot {
+                layer_map: self.layer_map.load_full(),
+                layer_fmgr: Arc::clone(&self.layer_fmgr),
+            },
+            pseudo_lock: Either::Left(pseudo_lock),
+            layer_manager: self.clone(),
+        }
+    }
+
+    /// Take the snapshot of the layer map and return a write guard. With the `modify` call, the guard
+    /// will only hold a read lock instead of write lock.
+    pub async fn modify(self: &Arc<Self>) -> LayerManagerWriteGuard {
+        // take the lock before taking snapshot
+        let pseudo_lock = self.pseudo_lock.clone().read_owned().await;
+        LayerManagerWriteGuard {
+            snapshot: LayerSnapshot {
+                layer_map: self.layer_map.load_full(),
+                layer_fmgr: Arc::clone(&self.layer_fmgr),
+            },
+            pseudo_lock: Either::Right(pseudo_lock),
+            layer_manager: self.clone(),
+        }
+    }
+
+    pub fn try_write(
+        self: &Arc<Self>,
+    ) -> Result<LayerManagerWriteGuard, tokio::sync::TryLockError> {
+        self.pseudo_lock
+            .clone()
+            .try_write_owned()
+            .map(|pseudo_lock| LayerManagerWriteGuard {
+                snapshot: LayerSnapshot {
+                    layer_map: self.layer_map.load_full(),
+                    layer_fmgr: Arc::clone(&self.layer_fmgr),
+                },
+                pseudo_lock: Either::Left(pseudo_lock),
+                layer_manager: self.clone(),
+            })
+    }
+
+    /// Make an update to the layer map. This function will NOT take the state lock and should ONLY
+    /// be used when there is not known concurrency when initializing the layer map. Error will be returned only when
+    /// the update function fails.
+    fn initialize_update(&self, f: impl FnOnce(LayerMap) -> Result<LayerMap>) -> Result<()> {
+        let snapshot = self.layer_map.load_full();
+        let new_layer_map = f(LayerMap::clone(&*snapshot))?;
+        let old_layer_map = self.layer_map.swap(Arc::new(new_layer_map));
+        debug_assert_eq!(
+            Arc::as_ptr(&snapshot) as usize,
+            Arc::as_ptr(&old_layer_map) as usize,
+            "race detected when modifying layer map, use `update` instead of `initialize_update`."
+        );
+        Ok(())
+    }
+
+    /// Make an update to the layer map. Error will be returned only when the update function fails.
+    async fn update<T>(&self, f: impl FnOnce(LayerMap) -> Result<(LayerMap, T)>) -> Result<T> {
+        let _guard = self.state_lock.lock().await;
+        let snapshot = self.layer_map.load_full();
+        let (new_layer_map, data) = f(LayerMap::clone(&*snapshot))?;
+        let old_layer_map = self.layer_map.swap(Arc::new(new_layer_map));
+        debug_assert_eq!(
+            Arc::as_ptr(&snapshot) as usize,
+            Arc::as_ptr(&old_layer_map) as usize,
+            "race detected when modifying layer map, please check if `initialize_update` is used on the `update` code path."
+        );
+        Ok(data)
+    }
+}
+
+impl LayerSnapshot {
+    fn get_from_desc(&self, desc: &PersistentLayerDesc) -> Arc<dyn PersistentLayer> {
        self.layer_fmgr.get_from_desc(desc)
    }

    /// Get an immutable reference to the layer map.
    ///
-    /// We expect users only to be able to get an immutable layer map. If users want to make modifications,
-    /// they should use the below semantic APIs. This design makes us step closer to immutable storage state.
-    pub fn layer_map(&self) -> &LayerMap {
+    /// If a user needs to modify the layer map, they should get a write guard and use the semantic
+    /// functions.
+    fn layer_map(&self) -> &LayerMap {
        &self.layer_map
    }

-    /// Get a mutable reference to the layer map. This function will be removed once `flush_frozen_layer`
-    /// gets a refactor.
-    pub fn layer_map_mut(&mut self) -> &mut LayerMap {
-        &mut self.layer_map
+    fn contains(&self, layer: &Arc<dyn PersistentLayer>) -> bool {
+        self.layer_fmgr.contains(layer)
+    }
+}
+
+impl LayerManagerReadGuard {
+    pub fn get_from_desc(&self, desc: &PersistentLayerDesc) -> Arc<dyn PersistentLayer> {
+        self.snapshot.get_from_desc(desc)
+    }
+
+    /// Get an immutable reference to the layer map.
+    pub fn layer_map(&self) -> &LayerMap {
+        self.snapshot.layer_map()
+    }
+}
+
+impl LayerManagerWriteGuard {
+    /// Check if the layer file manager contains a layer. This should ONLY be used in the compaction
+    /// code path where we need to check if a layer already exists on disk. With the immutable layer
+    /// map design, it is possible that the layer map snapshot does not contain the layer, but the layer file
+    /// manager does. Therefore, use this function with caution.
+    pub(crate) fn contains(&self, layer: &Arc<dyn PersistentLayer>) -> bool {
+        self.snapshot.contains(layer)
+    }
+
+    pub fn get_from_desc(&self, desc: &PersistentLayerDesc) -> Arc<dyn PersistentLayer> {
+        self.snapshot.get_from_desc(desc)
+    }
+
+    /// Get an immutable reference to the layer map.
+    pub fn layer_map(&self) -> &LayerMap {
+        self.snapshot.layer_map()
    }

    /// Replace layers in the layer file manager, used in evictions and layer downloads.
-    pub fn replace_and_verify(
+    pub(crate) fn replace_and_verify(
        &mut self,
        expected: Arc<dyn PersistentLayer>,
        new: Arc<dyn PersistentLayer>,
    ) -> Result<()> {
-        self.layer_fmgr.replace_and_verify(expected, new)
+        self.snapshot.layer_fmgr.replace_and_verify(expected, new)
    }

    /// Called from `load_layer_map`. Initialize the layer manager with:
    /// 1. all on-disk layers
    /// 2. next open layer (with disk disk_consistent_lsn LSN)
-    pub fn initialize_local_layers(
+    pub(crate) fn initialize_local_layers(
        &mut self,
        on_disk_layers: Vec<Arc<dyn PersistentLayer>>,
        next_open_layer_at: Lsn,
    ) {
-        let mut updates = self.layer_map.batch_update();
-        for layer in on_disk_layers {
-            Self::insert_historic_layer(layer, &mut updates, &mut self.layer_fmgr);
-        }
-        updates.flush();
-        self.layer_map.next_open_layer_at = Some(next_open_layer_at);
+        assert!(
+            self.pseudo_lock.is_left(),
+            "should use `write` guard for this function."
+        );
+
+        self.layer_manager
+            .initialize_update(|mut layer_map| {
+                let mut updates = layer_map.batch_update();
+                for layer in on_disk_layers {
+                    Self::insert_historic_layer(layer, &mut updates, &self.snapshot.layer_fmgr);
+                }
+                updates.flush();
+                layer_map.next_open_layer_at = Some(next_open_layer_at);
+                Ok(layer_map)
+            })
+            .unwrap();
    }

    /// Initialize when creating a new timeline, called in `init_empty_layer_map`.
-    pub fn initialize_empty(&mut self, next_open_layer_at: Lsn) {
-        self.layer_map.next_open_layer_at = Some(next_open_layer_at);
+    pub(crate) fn initialize_empty(&mut self, next_open_layer_at: Lsn) {
+        self.layer_manager
+            .initialize_update(|mut layer_map| {
+                layer_map.next_open_layer_at = Some(next_open_layer_at);
+                Ok(layer_map)
+            })
+            .unwrap();
    }

-    pub fn initialize_remote_layers(
+    pub(crate) fn initialize_remote_layers(
        &mut self,
        corrupted_local_layers: Vec<Arc<dyn PersistentLayer>>,
        remote_layers: Vec<Arc<RemoteLayer>>,
    ) {
-        let mut updates = self.layer_map.batch_update();
-        for layer in corrupted_local_layers {
-            Self::remove_historic_layer(layer, &mut updates, &mut self.layer_fmgr);
-        }
-        for layer in remote_layers {
-            Self::insert_historic_layer(layer, &mut updates, &mut self.layer_fmgr);
-        }
-        updates.flush();
+        assert!(
+            self.pseudo_lock.is_left(),
+            "should use `write` guard for this function."
+        );
+
+        self.layer_manager
+            .initialize_update(|mut layer_map| {
+                let mut updates = layer_map.batch_update();
+                for layer in corrupted_local_layers {
+                    Self::remove_historic_layer(layer, &mut updates, &self.snapshot.layer_fmgr);
+                }
+                for layer in remote_layers {
+                    Self::insert_historic_layer(layer, &mut updates, &self.snapshot.layer_fmgr);
+                }
+                updates.flush();
+                Ok(layer_map)
+            })
+            .unwrap();
    }

    /// Open a new writable layer to append data if there is no open layer, otherwise return the current open layer,
-    /// called within `get_layer_for_write`.
-    pub fn get_layer_for_write(
+    /// called within `get_layer_for_write`. Only ONE thread can call this function, which is guaranteed by `write_lock`
+    /// in `Timeline`.
+    pub(crate) async fn get_layer_for_write(
        &mut self,
        lsn: Lsn,
        last_record_lsn: Lsn,
@@ -116,6 +288,11 @@ impl LayerManager {
        timeline_id: TimelineId,
        tenant_id: TenantId,
    ) -> Result<Arc<InMemoryLayer>> {
+        assert!(
+            self.pseudo_lock.is_left(),
+            "should use `write` guard for this function."
+        );
+
        ensure!(lsn.is_aligned());

        ensure!(
@@ -127,7 +304,7 @@ impl LayerManager {
        );

        // Do we have a layer open for writing already?
-        let layer = if let Some(open_layer) = &self.layer_map.open_layer {
+        let layer = if let Some(open_layer) = &self.snapshot.layer_map.open_layer {
            if open_layer.get_lsn_range().start > lsn {
                bail!(
                    "unexpected open layer in the future: open layers starts at {}, write lsn {}",
@@ -138,122 +315,216 @@ impl LayerManager {

            Arc::clone(open_layer)
        } else {
-            // No writeable layer yet. Create one.
-            let start_lsn = self
-                .layer_map
-                .next_open_layer_at
-                .context("No next open layer found")?;
+            self.layer_manager
+                .update(|mut layer_map| {
+                    // No writeable layer yet. Create one.
+                    let start_lsn = layer_map
+                        .next_open_layer_at
+                        .context("No next open layer found")?;

-            trace!(
-                "creating in-memory layer at {}/{} for record at {}",
-                timeline_id,
-                start_lsn,
-                lsn
-            );
+                    trace!(
+                        "creating in-memory layer at {}/{} for record at {}",
+                        timeline_id,
+                        start_lsn,
+                        lsn
+                    );

-            let new_layer = InMemoryLayer::create(conf, timeline_id, tenant_id, start_lsn)?;
-            let layer = Arc::new(new_layer);
+                    let new_layer = InMemoryLayer::create(conf, timeline_id, tenant_id, start_lsn)?;
+                    let layer = Arc::new(new_layer);

-            self.layer_map.open_layer = Some(layer.clone());
-            self.layer_map.next_open_layer_at = None;
+                    layer_map.open_layer = Some(layer.clone());
+                    layer_map.next_open_layer_at = None;

-            layer
+                    Ok((layer_map, layer))
+                })
+                .await?
        };

        Ok(layer)
    }

    /// Called from `freeze_inmem_layer`, returns true if successfully frozen.
-    pub fn try_freeze_in_memory_layer(
+    pub(crate) async fn try_freeze_in_memory_layer(
        &mut self,
        Lsn(last_record_lsn): Lsn,
        last_freeze_at: &AtomicLsn,
    ) {
+        assert!(
+            self.pseudo_lock.is_left(),
+            "should use `write` guard for this function."
+        );
+
        let end_lsn = Lsn(last_record_lsn + 1);

-        if let Some(open_layer) = &self.layer_map.open_layer {
+        if let Some(open_layer) = &self.snapshot.layer_map.open_layer {
            let open_layer_rc = Arc::clone(open_layer);
            // Does this layer need freezing?
            open_layer.freeze(end_lsn);

            // The layer is no longer open, update the layer map to reflect this.
            // We will replace it with on-disk historics below.
-            self.layer_map.frozen_layers.push_back(open_layer_rc);
-            self.layer_map.open_layer = None;
-            self.layer_map.next_open_layer_at = Some(end_lsn);
+            self.layer_manager
+                .update(|mut layer_map| {
+                    layer_map.frozen_layers.push_back(open_layer_rc);
+                    layer_map.open_layer = None;
+                    layer_map.next_open_layer_at = Some(end_lsn);
+                    Ok((layer_map, ()))
+                })
+                .await
+                .unwrap();
+
            last_freeze_at.store(end_lsn);
        }
    }

    /// Add image layers to the layer map, called from `create_image_layers`.
-    pub fn track_new_image_layers(&mut self, image_layers: Vec<ImageLayer>) {
-        let mut updates = self.layer_map.batch_update();
-        for layer in image_layers {
-            Self::insert_historic_layer(Arc::new(layer), &mut updates, &mut self.layer_fmgr);
-        }
-        updates.flush();
+    pub(crate) async fn track_new_image_layers(&mut self, image_layers: Vec<ImageLayer>) {
+        assert!(
+            self.pseudo_lock.is_left(),
+            "should use `write` guard for this function."
+        );
+        self.layer_manager
+            .update(|mut layer_map| {
+                let mut updates: BatchedUpdates<'_> = layer_map.batch_update();
+                for layer in image_layers {
+                    Self::insert_historic_layer(
+                        Arc::new(layer),
+                        &mut updates,
+                        &self.snapshot.layer_fmgr,
+                    );
+                }
+                updates.flush();
+                Ok((layer_map, ()))
+            })
+            .await
+            .unwrap();
    }

-    /// Insert into the layer map when a new delta layer is created, called from `create_delta_layer`.
-    pub fn track_new_l0_delta_layer(&mut self, delta_layer: Arc<DeltaLayer>) {
-        let mut updates = self.layer_map.batch_update();
-        Self::insert_historic_layer(delta_layer, &mut updates, &mut self.layer_fmgr);
-        updates.flush();
+    /// Flush a frozen layer and add the written delta layer to the layer map.
+    pub(crate) async fn finish_flush_l0_layer(
+        &mut self,
+        delta_layer: Option<DeltaLayer>,
+        frozen_layer_for_check: &Arc<InMemoryLayer>,
+    ) {
+        assert!(
+            self.pseudo_lock.is_left(),
+            "should use `write` guard for this function."
+        );
+        self.layer_manager
+            .update(|mut layer_map| {
+                let l = layer_map.frozen_layers.pop_front();
+                let mut updates = layer_map.batch_update();
+
+                // Only one thread may call this function at a time (for this
+                // timeline). If two threads tried to flush the same frozen
+                // layer to disk at the same time, that would not work.
+                assert!(compare_arced_layers(&l.unwrap(), frozen_layer_for_check));
+
+                if let Some(delta_layer) = delta_layer {
+                    Self::insert_historic_layer(
+                        Arc::new(delta_layer),
+                        &mut updates,
+                        &self.snapshot.layer_fmgr,
+                    );
+                }
+                updates.flush();
+                Ok((layer_map, ()))
+            })
+            .await
+            .unwrap();
    }

    /// Called when compaction is completed.
-    pub fn finish_compact_l0(
-        &mut self,
-        layer_removal_cs: Arc<tokio::sync::OwnedMutexGuard<()>>,
+    pub(crate) async fn finish_compact_l0_consume_guard(
+        self,
+        _layer_removal_cs: Arc<tokio::sync::OwnedMutexGuard<()>>,
        compact_from: Vec<Arc<dyn PersistentLayer>>,
        compact_to: Vec<Arc<dyn PersistentLayer>>,
        metrics: &TimelineMetrics,
    ) -> Result<()> {
-        let mut updates = self.layer_map.batch_update();
-        for l in compact_to {
-            Self::insert_historic_layer(l, &mut updates, &mut self.layer_fmgr);
-        }
-        for l in compact_from {
+        assert!(
+            self.pseudo_lock.is_right(),
+            "should use `modify` guard for this function."
+        );
+        let compact_from = self
+            .layer_manager
+            .update(|mut layer_map| {
+                let mut updates = layer_map.batch_update();
+                for l in compact_to {
+                    Self::insert_historic_layer(l, &mut updates, &self.snapshot.layer_fmgr);
+                }
+                for l in &compact_from {
+                    // only remove from the layer map, not from file manager
+                    updates.remove_historic(l.layer_desc().clone());
+                }
+                updates.flush();
+                Ok((layer_map, Ok::<_, anyhow::Error>(compact_from)))
+            })
+            .await??;
+        drop(self.pseudo_lock);
+        // acquire the write lock so that all read threads are blocked, and once this lock is acquired,
+        // new reads will be based on the updated layer map.
+        let guard = self.layer_manager.pseudo_lock.write().await;
+        drop(guard);
+        // now that no one has access to the old layer map, we can safely remove the layers from disk.
+        for layer in compact_from {
+            self.snapshot.layer_fmgr.remove(layer.clone());
            // NB: the layer file identified by descriptor `l` is guaranteed to be present
            // in the LayerFileManager because compaction kept holding `layer_removal_cs` the entire
            // time, even though we dropped `Timeline::layers` inbetween.
-            Self::delete_historic_layer(
-                layer_removal_cs.clone(),
-                l,
-                &mut updates,
-                metrics,
-                &mut self.layer_fmgr,
-            )?;
+            if !layer.is_remote_layer() {
+                if let Err(e) = layer.delete_resident_layer_file() {
+                    warn!("Failed to delete resident layer file: {}", e);
+                } else {
+                    let layer_file_size = layer.file_size();
+                    metrics.resident_physical_size_gauge.sub(layer_file_size);
+                }
+            }
        }
-        updates.flush();
        Ok(())
    }

    /// Called when garbage collect the timeline. Returns a guard that will apply the updates to the layer map.
-    pub fn finish_gc_timeline(
+    pub(crate) async fn finish_gc_timeline(
        &mut self,
        layer_removal_cs: Arc<tokio::sync::OwnedMutexGuard<()>>,
        gc_layers: Vec<Arc<dyn PersistentLayer>>,
        metrics: &TimelineMetrics,
-    ) -> Result<ApplyGcResultGuard> {
-        let mut updates = self.layer_map.batch_update();
-        for doomed_layer in gc_layers {
-            Self::delete_historic_layer(
-                layer_removal_cs.clone(),
-                doomed_layer,
-                &mut updates,
-                metrics,
-                &mut self.layer_fmgr,
-            )?; // FIXME: schedule succeeded deletions in timeline.rs `gc_timeline` instead of in batch?
-        }
-        Ok(ApplyGcResultGuard(updates))
+    ) -> Result<()> {
+        assert!(
+            self.pseudo_lock.is_left(),
+            "should use `write` guard for this function."
+        );
+        self.layer_manager
+            .update(|mut layer_map| {
+                let mut updates = layer_map.batch_update();
+                for doomed_layer in gc_layers {
+                    // TODO: decouple deletion and layer map modification
+                    if let Err(e) = Self::delete_historic_layer(
+                        layer_removal_cs.clone(),
+                        doomed_layer,
+                        &mut updates,
+                        metrics,
+                        &self.snapshot.layer_fmgr,
+                    )
+                    // FIXME: schedule succeeded deletions in timeline.rs `gc_timeline` instead of in batch?
+                    {
+                        updates.flush();
+                        return Ok((layer_map, Err(e)));
+                    }
+                }
+                updates.flush();
+                Ok((layer_map, Ok(())))
+            })
+            .await
+            .unwrap() // unwrap first level error
    }

    /// Helper function to insert a layer into the layer map and file manager.
    fn insert_historic_layer(
        layer: Arc<dyn PersistentLayer>,
        updates: &mut BatchedUpdates<'_>,
-        mapping: &mut LayerFileManager,
+        mapping: &LayerFileManager,
    ) {
        updates.insert_historic(layer.layer_desc().clone());
        mapping.insert(layer);
@@ -263,7 +534,7 @@ impl LayerManager {
    fn remove_historic_layer(
        layer: Arc<dyn PersistentLayer>,
        updates: &mut BatchedUpdates<'_>,
-        mapping: &mut LayerFileManager,
+        mapping: &LayerFileManager,
    ) {
        updates.remove_historic(layer.layer_desc().clone());
        mapping.remove(layer);
@@ -277,7 +548,7 @@ impl LayerManager {
        layer: Arc<dyn PersistentLayer>,
        updates: &mut BatchedUpdates<'_>,
        metrics: &TimelineMetrics,
-        mapping: &mut LayerFileManager,
+        mapping: &LayerFileManager,
    ) -> anyhow::Result<()> {
        if !layer.is_remote_layer() {
            layer.delete_resident_layer_file()?;
@@ -297,12 +568,16 @@ impl LayerManager {
    }
 }

+/// Manages the layer files in the local / remote file system. This is a wrapper around `DashMap`.
+///
+/// Developer notes: dashmap will deadlock in some cases. Please ensure only one reference to the element
+/// in the dashmap is held in each of the functions.
 pub struct LayerFileManager<T: AsLayerDesc + ?Sized = dyn PersistentLayer>(
-    HashMap<PersistentLayerKey, Arc<T>>,
+    DashMap<PersistentLayerKey, Arc<T>>,
 );

 impl<T: AsLayerDesc + ?Sized> LayerFileManager<T> {
-    fn get_from_desc(&self, desc: &PersistentLayerDesc) -> Arc<T> {
+    pub(crate) fn get_from_desc(&self, desc: &PersistentLayerDesc) -> Arc<T> {
        // The assumption for the `expect()` is that all code maintains the following invariant:
        // A layer's descriptor is present in the LayerMap => the LayerFileManager contains a layer for the descriptor.
        self.0
@@ -312,18 +587,22 @@ impl<T: AsLayerDesc + ?Sized> LayerFileManager<T> {
            .clone()
    }

-    pub(crate) fn insert(&mut self, layer: Arc<T>) {
+    pub(crate) fn insert(&self, layer: Arc<T>) {
        let present = self.0.insert(layer.layer_desc().key(), layer.clone());
        if present.is_some() && cfg!(debug_assertions) {
            panic!("overwriting a layer: {:?}", layer.layer_desc())
        }
    }

-    pub(crate) fn new() -> Self {
-        Self(HashMap::new())
+    pub(crate) fn contains(&self, layer: &Arc<T>) -> bool {
+        self.0.contains_key(&layer.layer_desc().key())
    }

-    pub(crate) fn remove(&mut self, layer: Arc<T>) {
+    pub(crate) fn new() -> Self {
+        Self(DashMap::new())
+    }
+
+    pub(crate) fn remove(&self, layer: Arc<T>) {
        let present = self.0.remove(&layer.layer_desc().key());
        if present.is_none() && cfg!(debug_assertions) {
            panic!(
@@ -333,7 +612,7 @@ impl<T: AsLayerDesc + ?Sized> LayerFileManager<T> {
        }
    }

-    pub(crate) fn replace_and_verify(&mut self, expected: Arc<T>, new: Arc<T>) -> Result<()> {
+    pub(crate) fn replace_and_verify(&self, expected: Arc<T>, new: Arc<T>) -> Result<()> {
        let key = expected.layer_desc().key();
        let other = new.layer_desc().key();

@@ -354,12 +633,12 @@ impl<T: AsLayerDesc + ?Sized> LayerFileManager<T> {
            "one layer is l0 while the other is not: {expected_l0} != {new_l0}"
        );

-        if let Some(layer) = self.0.get_mut(&key) {
+        if let Some(mut layer) = self.0.get_mut(&key) {
            anyhow::ensure!(
-                compare_arced_layers(&expected, layer),
+                compare_arced_layers(&expected, &*layer),
                "another layer was found instead of expected, expected={expected:?}, new={new:?}",
                expected = Arc::as_ptr(&expected),
-                new = Arc::as_ptr(layer),
+                new = Arc::as_ptr(&*layer),
            );
            *layer = new;
            Ok(())
--- a/pageserver/src/walrecord.rs
+++ b/pageserver/src/walrecord.rs
@@ -360,7 +360,6 @@ impl XlXactParsedRecord {
            }
        }
        let mut xnodes = Vec::<RelFileNode>::new();
-        // In v16 this XACT_XINFO_HAS_RELFILENODES is renamed to XACT_XINFO_HAS_RELFILELOCATORS
        if xinfo & pg_constants::XACT_XINFO_HAS_RELFILENODES != 0 {
            let nrels = buf.get_i32_le();
            for _i in 0..nrels {
--- a/pgxn/hnsw/hnsw.control
+++ b/pgxn/hnsw/hnsw.control
@@ -1,4 +1,4 @@
-comment = 'hnsw index'
+comment = '** Deprecated ** Please use pg_embedding instead'
 default_version = '0.1.0'
 module_pathname = '$libdir/hnsw'
 relocatable = true
--- a/pgxn/neon/file_cache.c
+++ b/pgxn/neon/file_cache.c
@@ -25,11 +25,7 @@
 #include "pagestore_client.h"
 #include "access/parallel.h"
 #include "postmaster/bgworker.h"
-#if PG_VERSION_NUM >= 160000
-#include "storage/relfilelocator.h"
-#else
 #include "storage/relfilenode.h"
-#endif
 #include "storage/buf_internals.h"
 #include "storage/latch.h"
 #include "storage/ipc.h"
@@ -43,7 +39,6 @@
 #include "postmaster/bgworker.h"
 #include "postmaster/interrupt.h"

-
 /*
 * Local file cache is used to temporary store relations pages in local file system.
 * All blocks of all relations are stored inside one file and addressed using shared hash map.
@@ -365,12 +360,9 @@ lfc_cache_contains(RelFileNode rnode, ForkNumber forkNum, BlockNumber blkno)
 	if (lfc_size_limit == 0) /* fast exit if file cache is disabled */
 		return false;

-#if PG_VERSION_NUM >= 160000
-	InitBufferTag(&tag, &rnode, forkNum, (blkno & ~(BLOCKS_PER_CHUNK-1)));
-#else
-	INIT_BUFFERTAG(tag, rnode, forkNum, (blkno & ~(BLOCKS_PER_CHUNK-1)));
-#endif
-
+	tag.rnode = rnode;
+	tag.forkNum = forkNum;
+	tag.blockNum = blkno & ~(BLOCKS_PER_CHUNK-1);
 	hash = get_hash_value(lfc_hash, &tag);

 	LWLockAcquire(lfc_lock, LW_SHARED);
@@ -395,11 +387,7 @@ lfc_evict(RelFileNode rnode, ForkNumber forkNum, BlockNumber blkno)
 	if (lfc_size_limit == 0) /* fast exit if file cache is disabled */
 		return;

-#if PG_VERSION_NUM >= 160000
-	InitBufferTag(&tag, &rnode, forkNum, (blkno & ~(BLOCKS_PER_CHUNK-1)));
-#else
 	INIT_BUFFERTAG(tag, rnode, forkNum, (blkno & ~(BLOCKS_PER_CHUNK-1)));
-#endif

 	hash = get_hash_value(lfc_hash, &tag);

@@ -469,12 +457,10 @@ lfc_read(RelFileNode rnode, ForkNumber forkNum, BlockNumber blkno,

 	if (lfc_size_limit == 0) /* fast exit if file cache is disabled */
 		return false;
-#if PG_VERSION_NUM >= 160000
-	InitBufferTag(&tag, &rnode, forkNum, (blkno & ~(BLOCKS_PER_CHUNK-1)));
-#else
-	INIT_BUFFERTAG(tag, rnode, forkNum, (blkno & ~(BLOCKS_PER_CHUNK-1)));
-#endif

+	tag.rnode = rnode;
+	tag.forkNum = forkNum;
+	tag.blockNum = blkno & ~(BLOCKS_PER_CHUNK-1);
 	hash = get_hash_value(lfc_hash, &tag);

 	LWLockAcquire(lfc_lock, LW_EXCLUSIVE);
@@ -540,12 +526,9 @@ lfc_write(RelFileNode rnode, ForkNumber forkNum, BlockNumber blkno,
 	if (lfc_size_limit == 0) /* fast exit if file cache is disabled */
 		return;

-#if PG_VERSION_NUM >= 160000
-	InitBufferTag(&tag, &rnode, forkNum, (blkno & ~(BLOCKS_PER_CHUNK-1)));
-#else
-	INIT_BUFFERTAG(tag, rnode, forkNum, (blkno & ~(BLOCKS_PER_CHUNK-1)));
-#endif
-
+	tag.rnode = rnode;
+	tag.forkNum = forkNum;
+	tag.blockNum = blkno & ~(BLOCKS_PER_CHUNK-1);
 	hash = get_hash_value(lfc_hash, &tag);

 	LWLockAcquire(lfc_lock, LW_EXCLUSIVE);
@@ -739,16 +722,9 @@ local_cache_pages(PG_FUNCTION_ARGS)
 				if (entry->bitmap[i >> 5] & (1 << (i & 31)))
 				{
 					fctx->record[n_pages].pageoffs = entry->offset*BLOCKS_PER_CHUNK + i;
-
-#if PG_VERSION_NUM >= 160000
-					fctx->record[n_pages].relfilenode =  entry->key.relNumber;
-					fctx->record[n_pages].reltablespace = entry->key.spcOid;
-					fctx->record[n_pages].reldatabase = entry->key.dbOid;
-#else
 					fctx->record[n_pages].relfilenode = entry->key.rnode.relNode;
 					fctx->record[n_pages].reltablespace = entry->key.rnode.spcNode;
 					fctx->record[n_pages].reldatabase = entry->key.rnode.dbNode;
-#endif
 					fctx->record[n_pages].forknum = entry->key.forkNum;
 					fctx->record[n_pages].blocknum = entry->key.blockNum + i;
 					fctx->record[n_pages].accesscount = entry->access_count;
--- a/pgxn/neon/pagestore_client.h
+++ b/pgxn/neon/pagestore_client.h
@@ -16,11 +16,7 @@
 #include "postgres.h"

 #include "access/xlogdefs.h"
-#if PG_VERSION_NUM >= 160000
-#include "storage/relfilelocator.h"
-#else
 #include "storage/relfilenode.h"
-#endif
 #include "storage/block.h"
 #include "storage/smgr.h"
 #include "lib/stringinfo.h"
@@ -29,34 +25,6 @@

 #include "pg_config.h"

-// This is a hack to avoid too many ifdefs in the function definitions.
-#if PG_VERSION_NUM >= 160000
-typedef RelFileLocator RelFileNode;
-typedef RelFileLocatorBackend RelFileNodeBackend;
-#define RelFileNodeBackendIsTemp RelFileLocatorBackendIsTemp
-#endif
-
-#if PG_VERSION_NUM >= 160000
-#define RelnGetRnode(reln) (reln->smgr_rlocator.locator)
-#define RnodeGetSpcOid(rnode) (rnode.spcOid)
-#define RnodeGetDbOid(rnode) (rnode.dbOid)
-#define RnodeGetRelNumber(rnode) (rnode.relNumber)
-
-#define BufTagGetRnode(tag) (BufTagGetRelFileLocator(&tag))
-#else
-#define RelnGetRnode(reln) (reln->smgr_rnode.node)
-#define RnodeGetSpcOid(rnode) (rnode.spcNode)
-#define RnodeGetDbOid(rnode) (rnode.dbNode)
-#define RnodeGetRelNumber(rnode) (rnode.relNode)
-
-#define BufTagGetRnode(tag) (tag.rnode)
-
-#endif
-
-#define RelnGetSpcOid(reln) (RnodeGetRelNumber(RelnGetRnode(reln)))
-#define RelnGetDbOid(reln) (RnodeGetDbOid(RelnGetRnode(reln)))
-#define RelnGetRelNumber(reln) (RnodeGetRelNumber(RelnGetRnode(reln)))
-
 typedef enum
 {
 	/* pagestore_client -> pagestore */
@@ -117,7 +85,7 @@ typedef struct
 typedef struct
 {
 	NeonRequest req;
-	Oid			dbOid;
+	Oid			dbNode;
 }			NeonDbSizeRequest;

 typedef struct
--- a/pgxn/neon/pagestore_smgr.c
+++ b/pgxn/neon/pagestore_smgr.c
@@ -58,11 +58,7 @@
 #include "postmaster/autovacuum.h"
 #include "replication/walsender.h"
 #include "storage/bufmgr.h"
-#if PG_VERSION_NUM >= 160000
-#include "storage/relfilelocator.h"
-#else
 #include "storage/relfilenode.h"
-#endif
 #include "storage/buf_internals.h"
 #include "storage/smgr.h"
 #include "storage/md.h"
@@ -74,8 +70,6 @@
 #include "access/xlogrecovery.h"
 #endif

-
-
 /*
 * If DEBUG_COMPARE_LOCAL is defined, we pass through all the SMGR API
 * calls to md.c, and *also* do the calls to the Page Server. On every
@@ -92,10 +86,7 @@
 static char *hexdump_page(char *page);
 #endif

-
-#define IS_LOCAL_REL(reln) (RelnGetDbOid(reln) != 0 && RelnGetRelNumber(reln) > FirstNormalObjectId)
-
-
+#define IS_LOCAL_REL(reln) (reln->smgr_rnode.node.dbNode != 0 && reln->smgr_rnode.node.relNode > FirstNormalObjectId)

 const int	SmgrTrace = DEBUG5;

@@ -193,13 +184,7 @@ typedef struct PrfHashEntry {
 	sizeof(BufferTag) \
 )

-
-#if PG_VERSION_NUM >= 160000
-#define SH_EQUAL(tb, a, b)	(BufferTagsEqual(&((a)->buftag),&((b)->buftag)))
-#else
 #define SH_EQUAL(tb, a, b)	(BUFFERTAGS_EQUAL((a)->buftag, (b)->buftag))
-#endif
-
 #define SH_SCOPE			static inline
 #define SH_DEFINE
 #define SH_DECLARE
@@ -649,7 +634,7 @@ prefetch_do_request(PrefetchRequest *slot, bool *force_latest, XLogRecPtr *force
 		.req.tag = T_NeonGetPageRequest,
 		.req.latest = false,
 		.req.lsn = 0,
-		.rnode = BufTagGetRnode(slot->buftag),
+		.rnode = slot->buftag.rnode,
 		.forknum = slot->buftag.forkNum,
 		.blkno = slot->buftag.blockNum,
 	};
@@ -664,7 +649,7 @@ prefetch_do_request(PrefetchRequest *slot, bool *force_latest, XLogRecPtr *force
 	{
 		XLogRecPtr lsn = neon_get_request_lsn(
 			&request.req.latest,
-			BufTagGetRnode(slot->buftag),
+			slot->buftag.rnode,
 			slot->buftag.forkNum,
 			slot->buftag.blockNum
 		);
@@ -744,11 +729,8 @@ prefetch_register_buffer(BufferTag tag, bool *force_latest, XLogRecPtr *force_ls
 		Assert(slot->status != PRFS_UNUSED);
 		Assert(MyPState->ring_last <= ring_index &&
 			   ring_index < MyPState->ring_unused);
-#if PG_VERSION_NUM >= 160000
-		Assert(BufferTagsEqual(&slot->buftag, &tag));
-#else
 		Assert(BUFFERTAGS_EQUAL(slot->buftag, tag));
-#endif
+
 		/*
 		 * If we want a specific lsn, we do not accept requests that were made
 		 * with a potentially different LSN.
@@ -911,9 +893,9 @@ nm_pack_request(NeonRequest * msg)

 				pq_sendbyte(&s, msg_req->req.latest);
 				pq_sendint64(&s, msg_req->req.lsn);
-				pq_sendint32(&s, RnodeGetSpcOid(msg_req->rnode));
-				pq_sendint32(&s, RnodeGetDbOid(msg_req->rnode));
-				pq_sendint32(&s, RnodeGetRelNumber(msg_req->rnode));
+				pq_sendint32(&s, msg_req->rnode.spcNode);
+				pq_sendint32(&s, msg_req->rnode.dbNode);
+				pq_sendint32(&s, msg_req->rnode.relNode);
 				pq_sendbyte(&s, msg_req->forknum);

 				break;
@@ -924,9 +906,9 @@ nm_pack_request(NeonRequest * msg)

 				pq_sendbyte(&s, msg_req->req.latest);
 				pq_sendint64(&s, msg_req->req.lsn);
-				pq_sendint32(&s, RnodeGetSpcOid(msg_req->rnode));
-				pq_sendint32(&s, RnodeGetDbOid(msg_req->rnode));
-				pq_sendint32(&s, RnodeGetRelNumber(msg_req->rnode));
+				pq_sendint32(&s, msg_req->rnode.spcNode);
+				pq_sendint32(&s, msg_req->rnode.dbNode);
+				pq_sendint32(&s, msg_req->rnode.relNode);
 				pq_sendbyte(&s, msg_req->forknum);

 				break;
@@ -937,7 +919,7 @@ nm_pack_request(NeonRequest * msg)

 				pq_sendbyte(&s, msg_req->req.latest);
 				pq_sendint64(&s, msg_req->req.lsn);
-				pq_sendint32(&s, msg_req->dbOid);
+				pq_sendint32(&s, msg_req->dbNode);

 				break;
 			}
@@ -947,9 +929,9 @@ nm_pack_request(NeonRequest * msg)

 				pq_sendbyte(&s, msg_req->req.latest);
 				pq_sendint64(&s, msg_req->req.lsn);
-				pq_sendint32(&s, RnodeGetSpcOid(msg_req->rnode));
-				pq_sendint32(&s, RnodeGetDbOid(msg_req->rnode));
-				pq_sendint32(&s, RnodeGetRelNumber(msg_req->rnode));
+				pq_sendint32(&s, msg_req->rnode.spcNode);
+				pq_sendint32(&s, msg_req->rnode.dbNode);
+				pq_sendint32(&s, msg_req->rnode.relNode);
 				pq_sendbyte(&s, msg_req->forknum);
 				pq_sendint32(&s, msg_req->blkno);

@@ -1082,9 +1064,9 @@ nm_to_string(NeonMessage * msg)

 				appendStringInfoString(&s, "{\"type\": \"NeonExistsRequest\"");
 				appendStringInfo(&s, ", \"rnode\": \"%u/%u/%u\"",
-								 RnodeGetSpcOid(msg_req->rnode),
-								 RnodeGetDbOid(msg_req->rnode),
-								 RnodeGetRelNumber(msg_req->rnode));
+								 msg_req->rnode.spcNode,
+								 msg_req->rnode.dbNode,
+								 msg_req->rnode.relNode);
 				appendStringInfo(&s, ", \"forknum\": %d", msg_req->forknum);
 				appendStringInfo(&s, ", \"lsn\": \"%X/%X\"", LSN_FORMAT_ARGS(msg_req->req.lsn));
 				appendStringInfo(&s, ", \"latest\": %d", msg_req->req.latest);
@@ -1098,9 +1080,9 @@ nm_to_string(NeonMessage * msg)

 				appendStringInfoString(&s, "{\"type\": \"NeonNblocksRequest\"");
 				appendStringInfo(&s, ", \"rnode\": \"%u/%u/%u\"",
-								 RnodeGetSpcOid(msg_req->rnode),
-								 RnodeGetDbOid(msg_req->rnode),
-								 RnodeGetRelNumber(msg_req->rnode));
+								 msg_req->rnode.spcNode,
+								 msg_req->rnode.dbNode,
+								 msg_req->rnode.relNode);
 				appendStringInfo(&s, ", \"forknum\": %d", msg_req->forknum);
 				appendStringInfo(&s, ", \"lsn\": \"%X/%X\"", LSN_FORMAT_ARGS(msg_req->req.lsn));
 				appendStringInfo(&s, ", \"latest\": %d", msg_req->req.latest);
@@ -1114,9 +1096,9 @@ nm_to_string(NeonMessage * msg)

 				appendStringInfoString(&s, "{\"type\": \"NeonGetPageRequest\"");
 				appendStringInfo(&s, ", \"rnode\": \"%u/%u/%u\"",
-								 RnodeGetSpcOid(msg_req->rnode),
-								 RnodeGetDbOid(msg_req->rnode),
-								 RnodeGetRelNumber(msg_req->rnode));
+								 msg_req->rnode.spcNode,
+								 msg_req->rnode.dbNode,
+								 msg_req->rnode.relNode);
 				appendStringInfo(&s, ", \"forknum\": %d", msg_req->forknum);
 				appendStringInfo(&s, ", \"blkno\": %u", msg_req->blkno);
 				appendStringInfo(&s, ", \"lsn\": \"%X/%X\"", LSN_FORMAT_ARGS(msg_req->req.lsn));
@@ -1129,7 +1111,7 @@ nm_to_string(NeonMessage * msg)
 				NeonDbSizeRequest *msg_req = (NeonDbSizeRequest *) msg;

 				appendStringInfoString(&s, "{\"type\": \"NeonDbSizeRequest\"");
-				appendStringInfo(&s, ", \"dbnode\": \"%u\"", msg_req->dbOid);
+				appendStringInfo(&s, ", \"dbnode\": \"%u\"", msg_req->dbNode);
 				appendStringInfo(&s, ", \"lsn\": \"%X/%X\"", LSN_FORMAT_ARGS(msg_req->req.lsn));
 				appendStringInfo(&s, ", \"latest\": %d", msg_req->req.latest);
 				appendStringInfoChar(&s, '}');
@@ -1231,7 +1213,6 @@ static void
 neon_wallog_page(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, char *buffer, bool force)
 {
 	XLogRecPtr	lsn = PageGetLSN(buffer);
-	RelFileNode rnode = RelnGetRnode(reln);

 	if (ShutdownRequestPending)
 		return;
@@ -1251,16 +1232,15 @@ neon_wallog_page(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, ch
 		/* FSM is never WAL-logged and we don't care. */
 		XLogRecPtr	recptr;

-
-		recptr = log_newpage_copy(&rnode, forknum, blocknum, buffer, false);
+		recptr = log_newpage_copy(&reln->smgr_rnode.node, forknum, blocknum, buffer, false);
 		XLogFlush(recptr);
 		lsn = recptr;
 		ereport(SmgrTrace,
 				(errmsg("Page %u of relation %u/%u/%u.%u was force logged. Evicted at lsn=%X/%X",
 						blocknum,
-						RelnGetSpcOid(reln),
-						RelnGetDbOid(reln),
-						RelnGetRelNumber(reln),
+						reln->smgr_rnode.node.spcNode,
+						reln->smgr_rnode.node.dbNode,
+						reln->smgr_rnode.node.relNode,
 						forknum, LSN_FORMAT_ARGS(lsn))));
 	}
 	else if (lsn == InvalidXLogRecPtr)
@@ -1288,9 +1268,9 @@ neon_wallog_page(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, ch
 			ereport(SmgrTrace,
 					(errmsg("Page %u of relation %u/%u/%u.%u is all-zeros",
 							blocknum,
-							RelnGetSpcOid(reln),
-							RelnGetDbOid(reln),
-							RelnGetRelNumber(reln),
+							reln->smgr_rnode.node.spcNode,
+							reln->smgr_rnode.node.dbNode,
+							reln->smgr_rnode.node.relNode,
 							forknum)));
 		}
 		else if (PageIsEmptyHeapPage(buffer))
@@ -1298,9 +1278,9 @@ neon_wallog_page(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, ch
 			ereport(SmgrTrace,
 					(errmsg("Page %u of relation %u/%u/%u.%u is an empty heap page with no LSN",
 							blocknum,
-							RelnGetSpcOid(reln),
-							RelnGetDbOid(reln),
-							RelnGetRelNumber(reln),
+							reln->smgr_rnode.node.spcNode,
+							reln->smgr_rnode.node.dbNode,
+							reln->smgr_rnode.node.relNode,
 							forknum)));
 		}
 		else
@@ -1308,9 +1288,9 @@ neon_wallog_page(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, ch
 			ereport(PANIC,
 					(errmsg("Page %u of relation %u/%u/%u.%u is evicted with zero LSN",
 							blocknum,
-							RelnGetSpcOid(reln),
-							RelnGetDbOid(reln),
-							RelnGetRelNumber(reln),
+							reln->smgr_rnode.node.spcNode,
+							reln->smgr_rnode.node.dbNode,
+							reln->smgr_rnode.node.relNode,
 							forknum)));
 		}
 	}
@@ -1319,9 +1299,9 @@ neon_wallog_page(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, ch
 		ereport(SmgrTrace,
 				(errmsg("Page %u of relation %u/%u/%u.%u is already wal logged at lsn=%X/%X",
 						blocknum,
-						RelnGetSpcOid(reln),
-						RelnGetDbOid(reln),
-						RelnGetRelNumber(reln),
+						reln->smgr_rnode.node.spcNode,
+						reln->smgr_rnode.node.dbNode,
+						reln->smgr_rnode.node.relNode,
 						forknum, LSN_FORMAT_ARGS(lsn))));
 	}

@@ -1329,7 +1309,7 @@ neon_wallog_page(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, ch
 	 * Remember the LSN on this page. When we read the page again, we must
 	 * read the same or newer version of it.
 	 */
-	SetLastWrittenLSNForBlock(lsn, rnode, forknum, blocknum);
+	SetLastWrittenLSNForBlock(lsn, reln->smgr_rnode.node, forknum, blocknum);
 }

 /*
@@ -1479,7 +1459,6 @@ neon_exists(SMgrRelation reln, ForkNumber forkNum)
 	BlockNumber n_blocks;
 	bool		latest;
 	XLogRecPtr	request_lsn;
-	RelFileNode rnode = RelnGetRnode(reln);

 	switch (reln->smgr_relpersistence)
 	{
@@ -1506,7 +1485,7 @@ neon_exists(SMgrRelation reln, ForkNumber forkNum)
 			elog(ERROR, "unknown relpersistence '%c'", reln->smgr_relpersistence);
 	}

-	if (get_cached_relsize(RelnGetRnode(reln), forkNum, &n_blocks))
+	if (get_cached_relsize(reln->smgr_rnode.node, forkNum, &n_blocks))
 	{
 		return true;
 	}
@@ -1521,20 +1500,20 @@ neon_exists(SMgrRelation reln, ForkNumber forkNum)
 	 *
 	 * For now, handle that special case here.
 	 */
-	if (RelnGetSpcOid(reln) == 0 &&
-		RelnGetDbOid(reln) == 0 &&
-		RelnGetRelNumber(reln) == 0)
+	if (reln->smgr_rnode.node.spcNode == 0 &&
+		reln->smgr_rnode.node.dbNode == 0 &&
+		reln->smgr_rnode.node.relNode == 0)
 	{
 		return false;
 	}

-	request_lsn = neon_get_request_lsn(&latest, rnode, forkNum, REL_METADATA_PSEUDO_BLOCKNO);
+	request_lsn = neon_get_request_lsn(&latest, reln->smgr_rnode.node, forkNum, REL_METADATA_PSEUDO_BLOCKNO);
 	{
 		NeonExistsRequest request = {
 			.req.tag = T_NeonExistsRequest,
 			.req.latest = latest,
 			.req.lsn = request_lsn,
-			.rnode = rnode,
+			.rnode = reln->smgr_rnode.node,
 		.forknum = forkNum};

 		resp = page_server_request(&request);
@@ -1550,9 +1529,9 @@ neon_exists(SMgrRelation reln, ForkNumber forkNum)
 			ereport(ERROR,
 					(errcode(ERRCODE_IO_ERROR),
 					 errmsg("could not read relation existence of rel %u/%u/%u.%u from page server at lsn %X/%08X",
-							RelnGetSpcOid(reln),
-							RelnGetDbOid(reln),
-							RelnGetRelNumber(reln),
+							reln->smgr_rnode.node.spcNode,
+							reln->smgr_rnode.node.dbNode,
+							reln->smgr_rnode.node.relNode,
 							forkNum,
 							(uint32) (request_lsn >> 32), (uint32) request_lsn),
 					 errdetail("page server returned error: %s",
@@ -1574,8 +1553,6 @@ neon_exists(SMgrRelation reln, ForkNumber forkNum)
 void
 neon_create(SMgrRelation reln, ForkNumber forkNum, bool isRedo)
 {
-	RelFileNode rnode = RelnGetRnode(reln);
-
 	switch (reln->smgr_relpersistence)
 	{
 		case 0:
@@ -1594,8 +1571,9 @@ neon_create(SMgrRelation reln, ForkNumber forkNum, bool isRedo)
 	}

 	elog(SmgrTrace, "Create relation %u/%u/%u.%u",
-		 RelnGetSpcOid(reln),
-		 RelnGetDbOid(reln), RelnGetRelNumber(reln),
+		 reln->smgr_rnode.node.spcNode,
+		 reln->smgr_rnode.node.dbNode,
+		 reln->smgr_rnode.node.relNode,
 		 forkNum);

 	/*
@@ -1619,12 +1597,12 @@ neon_create(SMgrRelation reln, ForkNumber forkNum, bool isRedo)
 	 */
 	if (isRedo)
 	{
-		update_cached_relsize(rnode, forkNum, 0);
-		get_cached_relsize(rnode, forkNum,
+		update_cached_relsize(reln->smgr_rnode.node, forkNum, 0);
+		get_cached_relsize(reln->smgr_rnode.node, forkNum,
 						   &reln->smgr_cached_nblocks[forkNum]);
 	}
 	else
-		set_cached_relsize(rnode, forkNum, 0);
+		set_cached_relsize(reln->smgr_rnode.node, forkNum, 0);

 #ifdef DEBUG_COMPARE_LOCAL
 	if (IS_LOCAL_REL(reln))
@@ -1661,12 +1639,7 @@ neon_unlink(RelFileNodeBackend rnode, ForkNumber forkNum, bool isRedo)
 	mdunlink(rnode, forkNum, isRedo);
 	if (!RelFileNodeBackendIsTemp(rnode))
 	{
-
-#if PG_VERSION_NUM >= 160000
-		forget_cached_relsize(rnode.locator, forkNum);
-#else
 		forget_cached_relsize(rnode.node, forkNum);
-#endif
 	}
 }

@@ -1685,7 +1658,6 @@ neon_extend(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 {
 	XLogRecPtr	lsn;
 	BlockNumber	n_blocks = 0;
-	RelFileNode rnode = RelnGetRnode(reln);

 	switch (reln->smgr_relpersistence)
 	{
@@ -1735,16 +1707,17 @@ neon_extend(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 		neon_wallog_page(reln, forkNum, n_blocks++, buffer, true);

 	neon_wallog_page(reln, forkNum, blkno, buffer, false);
-	set_cached_relsize(rnode, forkNum, blkno + 1);
+	set_cached_relsize(reln->smgr_rnode.node, forkNum, blkno + 1);

 	lsn = PageGetLSN(buffer);
 	elog(SmgrTrace, "smgrextend called for %u/%u/%u.%u blk %u, page LSN: %X/%08X",
-		 RelnGetSpcOid(reln),
-		 RelnGetDbOid(reln), RelnGetRelNumber(reln),
+		 reln->smgr_rnode.node.spcNode,
+		 reln->smgr_rnode.node.dbNode,
+		 reln->smgr_rnode.node.relNode,
 		 forkNum, blkno,
 		 (uint32) (lsn >> 32), (uint32) lsn);

-	lfc_write(rnode, forkNum, blkno, buffer);
+	lfc_write(reln->smgr_rnode.node, forkNum, blkno, buffer);

 #ifdef DEBUG_COMPARE_LOCAL
 	if (IS_LOCAL_REL(reln))
@@ -1759,9 +1732,9 @@ neon_extend(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 	if (lsn == InvalidXLogRecPtr)
 	{
 		lsn = GetXLogInsertRecPtr();
-		SetLastWrittenLSNForBlock(lsn, rnode, forkNum, blkno);
+		SetLastWrittenLSNForBlock(lsn, reln->smgr_rnode.node, forkNum, blkno);
 	}
-	SetLastWrittenLSNForRelation(lsn, rnode, forkNum);
+	SetLastWrittenLSNForRelation(lsn, reln->smgr_rnode.node, forkNum);
 }

 /*
@@ -1805,8 +1778,6 @@ neon_prefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum)
 	BufferTag	tag;
 	uint64		ring_index PG_USED_FOR_ASSERTS_ONLY;

-	RelFileNode rnode = RelnGetRnode(reln);
-
 	switch (reln->smgr_relpersistence)
 	{
 		case 0: /* probably shouldn't happen, but ignore it */
@@ -1821,18 +1792,15 @@ neon_prefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum)
 			elog(ERROR, "unknown relpersistence '%c'", reln->smgr_relpersistence);
 	}

-	if (lfc_cache_contains(rnode, forknum, blocknum))
+	if (lfc_cache_contains(reln->smgr_rnode.node, forknum, blocknum))
 		return false;

-#if PG_VERSION_NUM >= 160000
-	InitBufferTag(&tag, &rnode, forknum, blocknum);
-#else
 	tag = (BufferTag) {
-		.rnode = rnode,
+		.rnode = reln->smgr_rnode.node,
 		.forkNum = forknum,
 		.blockNum = blocknum
 	};
-#endif
+
 	ring_index = prefetch_register_buffer(tag, NULL, NULL);

 	Assert(ring_index < MyPState->ring_unused &&
@@ -1893,15 +1861,11 @@ neon_read_at_lsn(RelFileNode rnode, ForkNumber forkNum, BlockNumber blkno,
 	PrfHashEntry *entry;
 	PrefetchRequest *slot;

-#if PG_VERSION_NUM >= 160000
-	InitBufferTag(&buftag, &rnode, forkNum, blkno);
-#else
 	buftag = (BufferTag) {
 		.rnode = rnode,
 		.forkNum = forkNum,
-		.blockNum = blkno
+		.blockNum = blkno,
 	};
-#endif

 	/*
 	 * The redo process does not lock pages that it needs to replay but are
@@ -2001,9 +1965,9 @@ neon_read_at_lsn(RelFileNode rnode, ForkNumber forkNum, BlockNumber blkno,
 					(errcode(ERRCODE_IO_ERROR),
 					 errmsg("could not read block %u in rel %u/%u/%u.%u from page server at lsn %X/%08X",
 							blkno,
-							RnodeGetSpcOid(rnode),
-							RnodeGetDbOid(rnode),
-							RnodeGetRelNumber(rnode),
+							rnode.spcNode,
+							rnode.dbNode,
+							rnode.relNode,
 							forkNum,
 							(uint32) (request_lsn >> 32), (uint32) request_lsn),
 					 errdetail("page server returned error: %s",
@@ -2027,7 +1991,6 @@ neon_read(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 {
 	bool		latest;
 	XLogRecPtr	request_lsn;
-	RelFileNode rnode = RelnGetRnode(reln);

 	switch (reln->smgr_relpersistence)
 	{
@@ -2047,13 +2010,13 @@ neon_read(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 	}

 	/* Try to read from local file cache */
-	if (lfc_read(RelnGetRnode(reln), forkNum, blkno, buffer))
+	if (lfc_read(reln->smgr_rnode.node, forkNum, blkno, buffer))
 	{
 		return;
 	}

-	request_lsn = neon_get_request_lsn(&latest, rnode, forkNum, blkno);
-	neon_read_at_lsn(rnode, forkNum, blkno, request_lsn, latest, buffer);
+	request_lsn = neon_get_request_lsn(&latest, reln->smgr_rnode.node, forkNum, blkno);
+	neon_read_at_lsn(reln->smgr_rnode.node, forkNum, blkno, request_lsn, latest, buffer);

 #ifdef DEBUG_COMPARE_LOCAL
 	if (forkNum == MAIN_FORKNUM && IS_LOCAL_REL(reln))
@@ -2073,9 +2036,9 @@ neon_read(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 			{
 				elog(PANIC, "page is new in MD but not in Page Server at blk %u in rel %u/%u/%u fork %u (request LSN %X/%08X):\n%s\n",
 					 blkno,
-					 RelnGetSpcOid(reln),
-					 RelnGetDbOid(reln),
-					 RelnGetRelNumber(reln),
+					 reln->smgr_rnode.node.spcNode,
+					 reln->smgr_rnode.node.dbNode,
+					 reln->smgr_rnode.node.relNode,
 					 forkNum,
 					 (uint32) (request_lsn >> 32), (uint32) request_lsn,
 					 hexdump_page(buffer));
@@ -2085,9 +2048,9 @@ neon_read(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 		{
 			elog(PANIC, "page is new in Page Server but not in MD at blk %u in rel %u/%u/%u fork %u (request LSN %X/%08X):\n%s\n",
 				 blkno,
-				 RelnGetSpcOid(reln),
-				 RelnGetDbOid(reln),
-				 RelnGetRelNumber(reln),
+				 reln->smgr_rnode.node.spcNode,
+				 reln->smgr_rnode.node.dbNode,
+				 reln->smgr_rnode.node.relNode,
 				 forkNum,
 				 (uint32) (request_lsn >> 32), (uint32) request_lsn,
 				 hexdump_page(mdbuf));
@@ -2102,9 +2065,9 @@ neon_read(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 			{
 				elog(PANIC, "heap buffers differ at blk %u in rel %u/%u/%u fork %u (request LSN %X/%08X):\n------ MD ------\n%s\n------ Page Server ------\n%s\n",
 					 blkno,
-					 RelnGetSpcOid(reln),
-					 RelnGetDbOid(reln),
-					 RelnGetRelNumber(reln),
+					 reln->smgr_rnode.node.spcNode,
+					 reln->smgr_rnode.node.dbNode,
+					 reln->smgr_rnode.node.relNode,
 					 forkNum,
 					 (uint32) (request_lsn >> 32), (uint32) request_lsn,
 					 hexdump_page(mdbuf_masked),
@@ -2123,9 +2086,9 @@ neon_read(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 				{
 					elog(PANIC, "btree buffers differ at blk %u in rel %u/%u/%u fork %u (request LSN %X/%08X):\n------ MD ------\n%s\n------ Page Server ------\n%s\n",
 						 blkno,
-						 RelnGetSpcOid(reln),
-						 RelnGetDbOid(reln),
-						 RelnGetRelNumber(reln),
+						 reln->smgr_rnode.node.spcNode,
+						 reln->smgr_rnode.node.dbNode,
+						 reln->smgr_rnode.node.relNode,
 						 forkNum,
 						 (uint32) (request_lsn >> 32), (uint32) request_lsn,
 						 hexdump_page(mdbuf_masked),
@@ -2170,7 +2133,7 @@ neon_write(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 		   char *buffer, bool skipFsync)
 {
 	XLogRecPtr	lsn;
-	RelFileNode rnode = RelnGetRnode(reln);
+
 	switch (reln->smgr_relpersistence)
 	{
 		case 0:
@@ -2207,12 +2170,13 @@ neon_write(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,

 	lsn = PageGetLSN(buffer);
 	elog(SmgrTrace, "smgrwrite called for %u/%u/%u.%u blk %u, page LSN: %X/%08X",
-		 RelnGetSpcOid(reln),
-		 RelnGetDbOid(reln), RelnGetRelNumber(reln),
+		 reln->smgr_rnode.node.spcNode,
+		 reln->smgr_rnode.node.dbNode,
+		 reln->smgr_rnode.node.relNode,
 		 forknum, blocknum,
 		 (uint32) (lsn >> 32), (uint32) lsn);

-	lfc_write(rnode, forknum, blocknum, buffer);
+	lfc_write(reln->smgr_rnode.node, forknum, blocknum, buffer);

 #ifdef DEBUG_COMPARE_LOCAL
 	if (IS_LOCAL_REL(reln))
@@ -2230,7 +2194,6 @@ neon_nblocks(SMgrRelation reln, ForkNumber forknum)
 	BlockNumber n_blocks;
 	bool		latest;
 	XLogRecPtr	request_lsn;
-	RelFileNode rnode = RelnGetRnode(reln);

 	switch (reln->smgr_relpersistence)
 	{
@@ -2249,23 +2212,23 @@ neon_nblocks(SMgrRelation reln, ForkNumber forknum)
 			elog(ERROR, "unknown relpersistence '%c'", reln->smgr_relpersistence);
 	}

-	if (get_cached_relsize(RelnGetRnode(reln), forknum, &n_blocks))
+	if (get_cached_relsize(reln->smgr_rnode.node, forknum, &n_blocks))
 	{
 		elog(SmgrTrace, "cached nblocks for %u/%u/%u.%u: %u blocks",
-			 RelnGetSpcOid(reln),
-			 RelnGetDbOid(reln),
-			 RelnGetRelNumber(reln),
+			 reln->smgr_rnode.node.spcNode,
+			 reln->smgr_rnode.node.dbNode,
+			 reln->smgr_rnode.node.relNode,
 			 forknum, n_blocks);
 		return n_blocks;
 	}

-	request_lsn = neon_get_request_lsn(&latest, rnode, forknum, REL_METADATA_PSEUDO_BLOCKNO);
+	request_lsn = neon_get_request_lsn(&latest, reln->smgr_rnode.node, forknum, REL_METADATA_PSEUDO_BLOCKNO);
 	{
 		NeonNblocksRequest request = {
 			.req.tag = T_NeonNblocksRequest,
 			.req.latest = latest,
 			.req.lsn = request_lsn,
-			.rnode = rnode,
+			.rnode = reln->smgr_rnode.node,
 			.forknum = forknum,
 		};

@@ -2282,9 +2245,9 @@ neon_nblocks(SMgrRelation reln, ForkNumber forknum)
 			ereport(ERROR,
 					(errcode(ERRCODE_IO_ERROR),
 					 errmsg("could not read relation size of rel %u/%u/%u.%u from page server at lsn %X/%08X",
-							RelnGetSpcOid(reln),
-							RelnGetDbOid(reln),
-							RelnGetRelNumber(reln),
+							reln->smgr_rnode.node.spcNode,
+							reln->smgr_rnode.node.dbNode,
+							reln->smgr_rnode.node.relNode,
 							forknum,
 							(uint32) (request_lsn >> 32), (uint32) request_lsn),
 					 errdetail("page server returned error: %s",
@@ -2294,11 +2257,12 @@ neon_nblocks(SMgrRelation reln, ForkNumber forknum)
 		default:
 			elog(ERROR, "unexpected response from page server with tag 0x%02x", resp->tag);
 	}
-	update_cached_relsize(rnode, forknum, n_blocks);
+	update_cached_relsize(reln->smgr_rnode.node, forknum, n_blocks);

 	elog(SmgrTrace, "neon_nblocks: rel %u/%u/%u fork %u (request LSN %X/%08X): %u blocks",
-		 RelnGetSpcOid(reln),
-		 RelnGetDbOid(reln), RelnGetRelNumber(reln),
+		 reln->smgr_rnode.node.spcNode,
+		 reln->smgr_rnode.node.dbNode,
+		 reln->smgr_rnode.node.relNode,
 		 forknum,
 		 (uint32) (request_lsn >> 32), (uint32) request_lsn,
 		 n_blocks);
@@ -2311,7 +2275,7 @@ neon_nblocks(SMgrRelation reln, ForkNumber forknum)
 *	neon_db_size() -- Get the size of the database in bytes.
 */
 int64
-neon_dbsize(Oid dbOid)
+neon_dbsize(Oid dbNode)
 {
 	NeonResponse *resp;
 	int64		db_size;
@@ -2325,7 +2289,7 @@ neon_dbsize(Oid dbOid)
 			.req.tag = T_NeonDbSizeRequest,
 			.req.latest = latest,
 			.req.lsn = request_lsn,
-			.dbOid = dbOid,
+			.dbNode = dbNode,
 		};

 		resp = page_server_request(&request);
@@ -2341,7 +2305,7 @@ neon_dbsize(Oid dbOid)
 			ereport(ERROR,
 					(errcode(ERRCODE_IO_ERROR),
 					 errmsg("could not read db size of db %u from page server at lsn %X/%08X",
-							dbOid,
+							dbNode,
 							(uint32) (request_lsn >> 32), (uint32) request_lsn),
 					 errdetail("page server returned error: %s",
 							   ((NeonErrorResponse *) resp)->message)));
@@ -2352,7 +2316,7 @@ neon_dbsize(Oid dbOid)
 	}

 	elog(SmgrTrace, "neon_dbsize: db %u (request LSN %X/%08X): %ld bytes",
-		 dbOid,
+		 dbNode,
 		 (uint32) (request_lsn >> 32), (uint32) request_lsn,
 		 db_size);

@@ -2367,7 +2331,6 @@ void
 neon_truncate(SMgrRelation reln, ForkNumber forknum, BlockNumber nblocks)
 {
 	XLogRecPtr	lsn;
-	RelFileNode rnode = RelnGetRnode(reln);

 	switch (reln->smgr_relpersistence)
 	{
@@ -2387,7 +2350,7 @@ neon_truncate(SMgrRelation reln, ForkNumber forknum, BlockNumber nblocks)
 			elog(ERROR, "unknown relpersistence '%c'", reln->smgr_relpersistence);
 	}

-	set_cached_relsize(rnode, forknum, nblocks);
+	set_cached_relsize(reln->smgr_rnode.node, forknum, nblocks);

 	/*
 	 * Truncating a relation drops all its buffers from the buffer cache
@@ -2415,7 +2378,7 @@ neon_truncate(SMgrRelation reln, ForkNumber forknum, BlockNumber nblocks)
 	 * for the extended pages, so there's no harm in leaving behind obsolete
 	 * entries for the truncated chunks.
 	 */
-	SetLastWrittenLSNForRelation(lsn, rnode, forknum);
+	SetLastWrittenLSNForRelation(lsn, reln->smgr_rnode.node, forknum);

 #ifdef DEBUG_COMPARE_LOCAL
 	if (IS_LOCAL_REL(reln))
@@ -2485,9 +2448,9 @@ neon_start_unlogged_build(SMgrRelation reln)

 	ereport(SmgrTrace,
 			(errmsg("starting unlogged build of relation %u/%u/%u",
-					RelnGetSpcOid(reln),
-					RelnGetDbOid(reln),
-					RelnGetRelNumber(reln))));
+					reln->smgr_rnode.node.spcNode,
+					reln->smgr_rnode.node.dbNode,
+					reln->smgr_rnode.node.relNode)));

 	switch (reln->smgr_relpersistence)
 	{
@@ -2537,9 +2500,9 @@ neon_finish_unlogged_build_phase_1(SMgrRelation reln)

 	ereport(SmgrTrace,
 			(errmsg("finishing phase 1 of unlogged build of relation %u/%u/%u",
-					RelnGetSpcOid(reln),
-					RelnGetDbOid(reln),
-					RelnGetRelNumber(reln))));
+					reln->smgr_rnode.node.spcNode,
+					reln->smgr_rnode.node.dbNode,
+					reln->smgr_rnode.node.relNode)));

 	if (unlogged_build_phase == UNLOGGED_BUILD_NOT_PERMANENT)
 		return;
@@ -2566,9 +2529,9 @@ neon_end_unlogged_build(SMgrRelation reln)

 	ereport(SmgrTrace,
 			(errmsg("ending unlogged build of relation %u/%u/%u",
-					RelnGetSpcOid(reln),
-					RelnGetDbOid(reln),
-					RelnGetRelNumber(reln))));
+					reln->smgr_rnode.node.spcNode,
+					reln->smgr_rnode.node.dbNode,
+					reln->smgr_rnode.node.relNode)));

 	if (unlogged_build_phase != UNLOGGED_BUILD_NOT_PERMANENT)
 	{
@@ -2581,24 +2544,16 @@ neon_end_unlogged_build(SMgrRelation reln)
 		reln->smgr_relpersistence = RELPERSISTENCE_PERMANENT;

 		/* Remove local copy */
-#if PG_VERSION_NUM >= 160000
-		rnode.locator = RelnGetRnode(reln);
-#else
-		rnode.node = RelnGetRnode(reln);
-#endif
+		rnode = reln->smgr_rnode;
 		for (int forknum = 0; forknum <= MAX_FORKNUM; forknum++)
 		{
 			elog(SmgrTrace, "forgetting cached relsize for %u/%u/%u.%u",
-				 RelnGetSpcOid(reln),
-				 RelnGetDbOid(reln),
-				 RelnGetRelNumber(reln),
+				 rnode.node.spcNode,
+				 rnode.node.dbNode,
+				 rnode.node.relNode,
 				 forknum);

-#if PG_VERSION_NUM >= 160000
-			forget_cached_relsize(rnode.locator, forknum);
-#else
 			forget_cached_relsize(rnode.node, forknum);
-#endif
 			mdclose(reln, forknum);
 			/* use isRedo == true, so that we drop it immediately */
 			mdunlink(rnode, forknum, true);
@@ -2751,16 +2706,10 @@ neon_redo_read_buffer_filter(XLogReaderState *record, uint8 block_id)
 	 * regardless of whether the block is stored in shared buffers.
 	 * See also this function's top comment.
 	 */
-
-	if (!OidIsValid(RnodeGetDbOid(rnode)))
+	if (!OidIsValid(rnode.dbNode))
 		return false;

-#if PG_VERSION_NUM >= 160000
-	InitBufferTag(&tag, &rnode, forknum, blkno);
-#else
 	INIT_BUFFERTAG(tag, rnode, forknum, blkno);
-#endif
-
 	hash = BufTableHashCode(&tag);
 	partitionLock = BufMappingPartitionLock(hash);

--- a/pgxn/neon/relsize_cache.c
+++ b/pgxn/neon/relsize_cache.c
@@ -15,11 +15,7 @@
 #include "postgres.h"

 #include "pagestore_client.h"
-#if PG_VERSION_NUM >= 160000
-#include "storage/relfilelocator.h"
-#else
 #include "storage/relfilenode.h"
-#endif
 #include "storage/smgr.h"
 #include "storage/lwlock.h"
 #include "storage/ipc.h"
@@ -32,7 +28,6 @@
 #include "miscadmin.h"
 #endif

-
 typedef struct
 {
 	RelFileNode rnode;
--- a/pgxn/neon/walproposer.c
+++ b/pgxn/neon/walproposer.c
@@ -1394,12 +1394,7 @@ WalProposerRecovery(int donor, TimeLineID timeline, XLogRecPtr startpos, XLogRec
 	WalReceiverConn *wrconn;
 	WalRcvStreamOptions options;

-#if PG_VERSION_NUM >= 160000
-	bool must_use_password = false;
-	wrconn = walrcv_connect(safekeeper[donor].conninfo, false, must_use_password, "wal_proposer_recovery", &err);
-#else
 	wrconn = walrcv_connect(safekeeper[donor].conninfo, false, "wal_proposer_recovery", &err);
-#endif
 	if (!wrconn)
 	{
 		ereport(WARNING,
--- a/pgxn/neon/walproposer_utils.c
+++ b/pgxn/neon/walproposer_utils.c
@@ -26,10 +26,6 @@
 #include "access/xlogrecovery.h"
 #endif

-#if PG_VERSION_NUM >= 160000
-#include "utils/guc.h"
-#endif
-
 /*
 * These variables are used similarly to openLogFile/SegNo,
 * but for walproposer to write the XLOG during recovery. walpropFileTLI is the TimeLineID
--- a/pgxn/neon_test_utils/neontest.c
+++ b/pgxn/neon_test_utils/neontest.c
@@ -128,11 +128,7 @@ clear_buffer_cache(PG_FUNCTION_ARGS)
 			else
 				isvalid = false;
 			bufferid = BufferDescriptorGetBuffer(bufHdr);
-#if PG_VERSION_NUM >= 160000
-			rnode = BufTagGetRelFileLocator(&bufHdr->tag);
-#else
 			rnode = bufHdr->tag.rnode;
-#endif
 			forknum = bufHdr->tag.forkNum;
 			blocknum = bufHdr->tag.blockNum;

@@ -242,7 +238,7 @@ get_raw_page_at_lsn(PG_FUNCTION_ARGS)
 	SET_VARSIZE(raw_page, BLCKSZ + VARHDRSZ);
 	raw_page_data = VARDATA(raw_page);

-	neon_read_at_lsn(RelnGetRnode(RelationGetSmgr(rel)), forknum, blkno, read_lsn, request_latest, raw_page_data);
+	neon_read_at_lsn(rel->rd_node, forknum, blkno, read_lsn, request_latest, raw_page_data);

 	relation_close(rel, AccessShareLock);

@@ -271,17 +267,11 @@ get_raw_page_at_lsn_ex(PG_FUNCTION_ARGS)
 		PG_RETURN_NULL();

 	{
-#if PG_VERSION_NUM >= 160000
-		RelFileLocator rnode = {
-			.spcOid = PG_GETARG_OID(0),
-			.dbOid = PG_GETARG_OID(1),
-		.relNumber = PG_GETARG_OID(2)};
-#else
 		RelFileNode rnode = {
 			.spcNode = PG_GETARG_OID(0),
 			.dbNode = PG_GETARG_OID(1),
 		.relNode = PG_GETARG_OID(2)};
-#endif
+
 		ForkNumber	forknum = PG_GETARG_UINT32(3);

 		uint32		blkno = PG_GETARG_UINT32(4);
--- a/pgxn/neon_walredo/inmem_smgr.c
+++ b/pgxn/neon_walredo/inmem_smgr.c
@@ -21,6 +21,7 @@
 #include "access/xlog.h"
 #include "storage/block.h"
 #include "storage/buf_internals.h"
+#include "storage/relfilenode.h"
 #include "storage/smgr.h"

 #if PG_VERSION_NUM >= 150000
@@ -29,7 +30,6 @@

 #include "inmem_smgr.h"

-
 /* Size of the in-memory smgr */
 #define MAX_PAGES 64

@@ -46,22 +46,12 @@ locate_page(SMgrRelation reln, ForkNumber forknum, BlockNumber blkno)
 	/* We only hold a small number of pages, so linear search */
 	for (int i = 0; i < used_pages; i++)
 	{
-
-#if PG_VERSION_NUM >= 160000
-		if (BufTagMatchesRelFileLocator(&page_tag[i], &reln->smgr_rlocator.locator)
+		if (RelFileNodeEquals(reln->smgr_rnode.node, page_tag[i].rnode)
 			&& forknum == page_tag[i].forkNum
 			&& blkno == page_tag[i].blockNum)
 		{
 			return i;
 		}
-#else
-		if (RelFileNodeEquals(RelnGetRnode(reln), page_tag[i].rnode)
-			&& forknum == page_tag[i].forkNum
-			&& blkno == page_tag[i].blockNum)
-		{
-			return i;
-		}
-#endif
 	}
 	return -1;
 }
@@ -107,12 +97,8 @@ inmem_exists(SMgrRelation reln, ForkNumber forknum)
 {
 	for (int i = 0; i < used_pages; i++)
 	{
-#if PG_VERSION_NUM >= 160000
-		if (BufTagMatchesRelFileLocator(&page_tag[i], &reln->smgr_rlocator.locator)
-#else
-		if (RelFileNodeEquals(RelnGetRnode(reln), page_tag[i].rnode)
-#endif			
-		&& forknum == page_tag[i].forkNum)
+		if (RelFileNodeEquals(reln->smgr_rnode.node, page_tag[i].rnode)
+			&& forknum == page_tag[i].forkNum)
 		{
 			return true;
 		}
@@ -230,9 +216,9 @@ inmem_write(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 		 */
 		elog(used_pages >= WARN_PAGES ? WARNING : DEBUG1,
 			 "inmem_write() called for %u/%u/%u.%u blk %u: used_pages %u",
-			 RelnGetSpcOid(reln),
-			 RelnGetDbOid(reln),
-			 RelnGetRelNumber(reln),
+			 reln->smgr_rnode.node.spcNode,
+			 reln->smgr_rnode.node.dbNode,
+			 reln->smgr_rnode.node.relNode,
 			 forknum,
 			 blocknum,
 			 used_pages);
@@ -241,19 +227,14 @@ inmem_write(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,

 		pg = used_pages;
 		used_pages++;
-
-#if PG_VERSION_NUM >= 160000
-	InitBufferTag(&page_tag[pg], &RelnGetRnode(reln), forknum, blocknum);
-#else
-	INIT_BUFFERTAG(page_tag[pg], RelnGetRnode(reln), forknum, blocknum);
-#endif
+		INIT_BUFFERTAG(page_tag[pg], reln->smgr_rnode.node, forknum, blocknum);
 	}
 	else
 	{
 		elog(DEBUG1, "inmem_write() called for %u/%u/%u.%u blk %u: found at %u",
-			 RelnGetSpcOid(reln),
-			 RelnGetDbOid(reln),
-			 RelnGetRelNumber(reln),
+			 reln->smgr_rnode.node.spcNode,
+			 reln->smgr_rnode.node.dbNode,
+			 reln->smgr_rnode.node.relNode,
 			 forknum,
 			 blocknum,
 			 used_pages);
--- a/pgxn/neon_walredo/inmem_smgr.h
+++ b/pgxn/neon_walredo/inmem_smgr.h
@@ -11,40 +11,6 @@
 #ifndef INMEM_SMGR_H
 #define INMEM_SMGR_H

-#if PG_VERSION_NUM >= 160000
-#include "storage/relfilelocator.h"
-#else
-#include "storage/relfilenode.h"
-#endif
-
-// This is a hack to avoid too many ifdefs in the function definitions.
-#if PG_VERSION_NUM >= 160000
-typedef RelFileLocator RelFileNode;
-typedef RelFileLocatorBackend RelFileNodeBackend;
-#define RelFileNodeBackendIsTemp RelFileLocatorBackendIsTemp
-#endif
-
-#if PG_VERSION_NUM >= 160000
-#define RelnGetRnode(reln) (reln->smgr_rlocator.locator)
-#define RnodeGetSpcOid(rnode) (rnode.spcOid)
-#define RnodeGetDbOid(rnode) (rnode.dbOid)
-#define RnodeGetRelNumber(rnode) (rnode.relNumber)
-
-#define BufTagGetRnode(tag) (BufTagGetRelFileLocator(&tag))
-#else
-#define RelnGetRnode(reln) (reln->smgr_rnode.node)
-#define RnodeGetSpcOid(rnode) (rnode.spcNode)
-#define RnodeGetDbOid(rnode) (rnode.dbNode)
-#define RnodeGetRelNumber(rnode) (rnode.relNode)
-
-#define BufTagGetRnode(tag) (tag.rnode)
-
-#endif
-
-#define RelnGetSpcOid(reln) (RnodeGetRelNumber(RelnGetRnode(reln)))
-#define RelnGetDbOid(reln) (RnodeGetDbOid(RelnGetRnode(reln)))
-#define RelnGetRelNumber(reln) (RnodeGetRelNumber(RelnGetRnode(reln)))
-
 extern const f_smgr *smgr_inmem(BackendId backend, RelFileNode rnode);
 extern void smgr_init_inmem(void);

--- a/pgxn/neon_walredo/walredoproc.c
+++ b/pgxn/neon_walredo/walredoproc.c
@@ -62,10 +62,8 @@
 #endif

 #ifndef HAVE_GETRUSAGE
-#if PG_VERSION_NUM < 160000
 #include "rusagestub.h"
 #endif
-#endif

 #include "access/clog.h"
 #include "access/commit_ts.h"
@@ -119,7 +117,6 @@
 #include "neon_seccomp.h"
 #endif

-
 PG_MODULE_MAGIC;

 static int	ReadRedoCommand(StringInfo inBuf);
@@ -665,31 +662,18 @@ BeginRedoForBlock(StringInfo input_message)
 	 * BlockNumber
 	 */
 	forknum = pq_getmsgbyte(input_message);
-#if PG_VERSION_NUM >= 160000
-	rnode.spcOid = pq_getmsgint(input_message, 4);
-	rnode.dbOid = pq_getmsgint(input_message, 4);
-	rnode.relNumber = pq_getmsgint(input_message, 4);
-#else
 	rnode.spcNode = pq_getmsgint(input_message, 4);
 	rnode.dbNode = pq_getmsgint(input_message, 4);
 	rnode.relNode = pq_getmsgint(input_message, 4);
-#endif
 	blknum = pq_getmsgint(input_message, 4);
 	wal_redo_buffer = InvalidBuffer;

-#if PG_VERSION_NUM >= 160000
-	InitBufferTag(&target_redo_tag, &rnode, forknum, blknum);
-#else
 	INIT_BUFFERTAG(target_redo_tag, rnode, forknum, blknum);
-#endif
-

 	elog(TRACE, "BeginRedoForBlock %u/%u/%u.%d blk %u",
-#if PG_VERSION_NUM >= 160000
-			target_redo_tag.spcOid, target_redo_tag.dbOid, target_redo_tag.relNumber,
-#else
-			target_redo_tag.rnode.spcNode, target_redo_tag.rnode.dbNode, target_redo_tag.rnode.relNode,
-#endif	
+		 target_redo_tag.rnode.spcNode,
+		 target_redo_tag.rnode.dbNode,
+		 target_redo_tag.rnode.relNode,
 		 target_redo_tag.forkNum,
 		 target_redo_tag.blockNum);

@@ -725,15 +709,9 @@ PushPage(StringInfo input_message)
 	 * 8k page content
 	 */
 	forknum = pq_getmsgbyte(input_message);
-#if PG_VERSION_NUM >= 160000
-	rnode.spcOid = pq_getmsgint(input_message, 4);
-	rnode.dbOid = pq_getmsgint(input_message, 4);
-	rnode.relNumber = pq_getmsgint(input_message, 4);
-#else
 	rnode.spcNode = pq_getmsgint(input_message, 4);
 	rnode.dbNode = pq_getmsgint(input_message, 4);
 	rnode.relNode = pq_getmsgint(input_message, 4);
-#endif
 	blknum = pq_getmsgint(input_message, 4);
 	content = pq_getmsgbytes(input_message, BLCKSZ);

@@ -853,12 +831,7 @@ ApplyRecord(StringInfo input_message)
 	 */
 	if (BufferIsInvalid(wal_redo_buffer))
 	{
-		wal_redo_buffer = NeonRedoReadBuffer(
-#if PG_VERSION_NUM >= 160000
-											 BufTagGetRelFileLocator(&target_redo_tag),
-#else
-											 target_redo_tag.rnode,
-#endif
+		wal_redo_buffer = NeonRedoReadBuffer(target_redo_tag.rnode,
 											 target_redo_tag.forkNum,
 											 target_redo_tag.blockNum,
 											 RBM_NORMAL);
@@ -900,43 +873,12 @@ apply_error_callback(void *arg)
 }


-#if PG_VERSION_NUM >= 160000

 static bool
 redo_block_filter(XLogReaderState *record, uint8 block_id)
 {
 	BufferTag	target_tag;

-	RelFileLocator rlocator;
-	XLogRecGetBlockTag(record, block_id,
-					   &rlocator, &target_tag.forkNum, &target_tag.blockNum);
-
-	target_tag.spcOid = rlocator.spcOid;
-	target_tag.dbOid = rlocator.dbOid;
-	target_tag.relNumber = rlocator.relNumber;
-
-	/*
-	 * Can a WAL redo function ever access a relation other than the one that
-	 * it modifies? I don't see why it would.
-	 */
-	if (RelFileLocatorEquals(BufTagGetRelFileLocator(&target_tag), BufTagGetRelFileLocator(&target_redo_tag)))
-		elog(WARNING, "REDO accessing unexpected page: %u/%u/%u.%u blk %u",
-			target_tag.spcOid, target_tag.dbOid, target_tag.relNumber,
-			target_tag.forkNum, target_tag.blockNum);
-
-	/*
-	 * If this block isn't one we are currently restoring, then return 'true'
-	 * so that this gets ignored
-	 */
-	return !BufferTagsEqual(&target_tag, &target_redo_tag);
-}
-#else
-static bool
-redo_block_filter(XLogReaderState *record, uint8 block_id)
-{
-	BufferTag	target_tag;
-
-
 #if PG_VERSION_NUM >= 150000
 	XLogRecGetBlockTag(record, block_id,
 					   &target_tag.rnode, &target_tag.forkNum, &target_tag.blockNum);
@@ -955,18 +897,14 @@ redo_block_filter(XLogReaderState *record, uint8 block_id)
 	 */
 	if (!RelFileNodeEquals(target_tag.rnode, target_redo_tag.rnode))
 		elog(WARNING, "REDO accessing unexpected page: %u/%u/%u.%u blk %u",
-			target_tag.rnode.spcNode, target_tag.rnode.dbNode, target_tag.rnode.relNode,
-			target_tag.forkNum, target_tag.blockNum);
+			 target_tag.rnode.spcNode, target_tag.rnode.dbNode, target_tag.rnode.relNode, target_tag.forkNum, target_tag.blockNum);

 	/*
 	 * If this block isn't one we are currently restoring, then return 'true'
 	 * so that this gets ignored
 	 */
-
 	return !BUFFERTAGS_EQUAL(target_tag, target_redo_tag);
 }
-#endif
-

 /*
 * Get a page image back from buffer cache.
@@ -993,15 +931,9 @@ GetPage(StringInfo input_message)
 	 * BlockNumber
 	 */
 	forknum = pq_getmsgbyte(input_message);
-#if PG_VERSION_NUM >= 160000
-	rnode.spcOid = pq_getmsgint(input_message, 4);
-	rnode.dbOid = pq_getmsgint(input_message, 4);
-	rnode.relNumber = pq_getmsgint(input_message, 4);
-#else
 	rnode.spcNode = pq_getmsgint(input_message, 4);
 	rnode.dbNode = pq_getmsgint(input_message, 4);
 	rnode.relNode = pq_getmsgint(input_message, 4);
-#endif
 	blknum = pq_getmsgint(input_message, 4);

 	/* FIXME: check that we got a BeginRedoForBlock message or this earlier */
@@ -1029,11 +961,7 @@ GetPage(StringInfo input_message)
 	} while (tot_written < BLCKSZ);

 	ReleaseBuffer(buf);
-#if PG_VERSION_NUM >= 160000
-	DropRelationAllLocalBuffers(rnode);
-#else
 	DropRelFileNodeAllLocalBuffers(rnode);
-#endif
 	wal_redo_buffer = InvalidBuffer;

 	elog(TRACE, "Page sent back for block %u", blknum);
--- a/poetry.lock
+++ b/poetry.lock
@@ -1868,6 +1868,20 @@ files = [
 packaging = ">=17.1"
 pytest = ">=5.3"

+[[package]]
+name = "pytest-split"
+version = "0.8.1"
+description = "Pytest plugin which splits the test suite to equally sized sub suites based on test execution time."
+optional = false
+python-versions = ">=3.7.1,<4.0"
+files = [
+    {file = "pytest_split-0.8.1-py3-none-any.whl", hash = "sha256:74b110ea091bd147cc1c5f9665a59506e5cedfa66f96a89fb03e4ab447c2c168"},
+    {file = "pytest_split-0.8.1.tar.gz", hash = "sha256:2d88bd3dc528689a7a3f58fc12ea165c3aa62e90795e420dfad920afe5612d6d"},
+]
+
+[package.dependencies]
+pytest = ">=5,<8"
+
 [[package]]
 name = "pytest-timeout"
 version = "2.1.0"
@@ -2513,4 +2527,4 @@ testing = ["func-timeout", "jaraco.itertools", "pytest (>=6)", "pytest-black (>=
 [metadata]
 lock-version = "2.0"
 python-versions = "^3.9"
-content-hash = "fe771b153ef7e308d6d04421d0eb3f97d00780882277d2b4fc1f296054d8db79"
+content-hash = "e16a65d8fdff4e2173610e552e0e7306e301de2c640ae6082ef6cc5755f566d2"
--- a/proxy/src/cache.rs
+++ b/proxy/src/cache.rs
@@ -262,24 +262,21 @@ pub mod timed_lru {
        token: Option<(C, C::LookupInfo<C::Key>)>,

        /// The value itself.
-        pub value: C::Value,
+        value: C::Value,
    }

    impl<C: Cache> Cached<C> {
        /// Place any entry into this wrapper; invalidation will be a no-op.
-        /// Unfortunately, rust doesn't let us implement [`From`] or [`Into`].
-        pub fn new_uncached(value: impl Into<C::Value>) -> Self {
-            Self {
-                token: None,
-                value: value.into(),
-            }
+        pub fn new_uncached(value: C::Value) -> Self {
+            Self { token: None, value }
        }

        /// Drop this entry from a cache if it's still there.
-        pub fn invalidate(&self) {
+        pub fn invalidate(self) -> C::Value {
            if let Some((cache, info)) = &self.token {
                cache.invalidate(info);
            }
+            self.value
        }

        /// Tell if this entry is actually cached.
--- a/proxy/src/compute.rs
+++ b/proxy/src/compute.rs
@@ -1,4 +1,9 @@
-use crate::{auth::parse_endpoint_param, cancellation::CancelClosure, error::UserFacingError};
+use crate::{
+    auth::parse_endpoint_param,
+    cancellation::CancelClosure,
+    console::errors::WakeComputeError,
+    error::{io_error, UserFacingError},
+};
 use futures::{FutureExt, TryFutureExt};
 use itertools::Itertools;
 use pq_proto::StartupMessageParams;
@@ -24,6 +29,12 @@ pub enum ConnectionError {
    TlsError(#[from] native_tls::Error),
 }

+impl From<WakeComputeError> for ConnectionError {
+    fn from(value: WakeComputeError) -> Self {
+        io_error(value).into()
+    }
+}
+
 impl UserFacingError for ConnectionError {
    fn to_string_client(&self) -> String {
        use ConnectionError::*;
--- a/proxy/src/console/provider.rs
+++ b/proxy/src/console/provider.rs
@@ -186,14 +186,14 @@ pub trait Api {
    async fn get_auth_info(
        &self,
        extra: &ConsoleReqExtra<'_>,
-        creds: &ClientCredentials<'_>,
+        creds: &ClientCredentials,
    ) -> Result<Option<AuthInfo>, errors::GetAuthInfoError>;

    /// Wake up the compute node and return the corresponding connection info.
    async fn wake_compute(
        &self,
        extra: &ConsoleReqExtra<'_>,
-        creds: &ClientCredentials<'_>,
+        creds: &ClientCredentials,
    ) -> Result<CachedNodeInfo, errors::WakeComputeError>;
 }

--- a/proxy/src/console/provider/mock.rs
+++ b/proxy/src/console/provider/mock.rs
@@ -106,7 +106,7 @@ impl super::Api for Api {
    async fn get_auth_info(
        &self,
        _extra: &ConsoleReqExtra<'_>,
-        creds: &ClientCredentials<'_>,
+        creds: &ClientCredentials,
    ) -> Result<Option<AuthInfo>, GetAuthInfoError> {
        self.do_get_auth_info(creds).await
    }
@@ -115,7 +115,7 @@ impl super::Api for Api {
    async fn wake_compute(
        &self,
        _extra: &ConsoleReqExtra<'_>,
-        _creds: &ClientCredentials<'_>,
+        _creds: &ClientCredentials,
    ) -> Result<CachedNodeInfo, WakeComputeError> {
        self.do_wake_compute()
            .map_ok(CachedNodeInfo::new_uncached)
--- a/proxy/src/console/provider/neon.rs
+++ b/proxy/src/console/provider/neon.rs
@@ -123,7 +123,7 @@ impl super::Api for Api {
    async fn get_auth_info(
        &self,
        extra: &ConsoleReqExtra<'_>,
-        creds: &ClientCredentials<'_>,
+        creds: &ClientCredentials,
    ) -> Result<Option<AuthInfo>, GetAuthInfoError> {
        self.do_get_auth_info(extra, creds).await
    }
@@ -132,7 +132,7 @@ impl super::Api for Api {
    async fn wake_compute(
        &self,
        extra: &ConsoleReqExtra<'_>,
-        creds: &ClientCredentials<'_>,
+        creds: &ClientCredentials,
    ) -> Result<CachedNodeInfo, WakeComputeError> {
        let key = creds.project().expect("impossible");

--- a/proxy/src/http/conn_pool.rs
+++ b/proxy/src/http/conn_pool.rs
@@ -1,19 +1,17 @@
+use anyhow::Context;
+use async_trait::async_trait;
 use parking_lot::Mutex;
 use pq_proto::StartupMessageParams;
 use std::fmt;
-use std::ops::ControlFlow;
 use std::{collections::HashMap, sync::Arc};
 use tokio::time;

-use crate::config;
 use crate::{auth, console};
+use crate::{compute, config};

 use super::sql_over_http::MAX_RESPONSE_SIZE;

-use crate::proxy::{
-    can_retry_tokio_postgres_error, invalidate_cache, retry_after, try_wake,
-    NUM_RETRIES_WAKE_COMPUTE,
-};
+use crate::proxy::ConnectMechanism;

 use tracing::error;
 use tracing::info;
@@ -187,6 +185,27 @@ impl GlobalConnPool {
    }
 }

+struct TokioMechanism<'a> {
+    conn_info: &'a ConnInfo,
+}
+
+#[async_trait]
+impl ConnectMechanism for TokioMechanism<'_> {
+    type Connection = tokio_postgres::Client;
+    type ConnectError = tokio_postgres::Error;
+    type Error = anyhow::Error;
+
+    async fn connect_once(
+        &self,
+        node_info: &console::CachedNodeInfo,
+        timeout: time::Duration,
+    ) -> Result<Self::Connection, Self::ConnectError> {
+        connect_to_compute_once(node_info, self.conn_info, timeout).await
+    }
+
+    fn update_connect_config(&self, _config: &mut compute::ConnCfg) {}
+}
+
 // Wake up the destination if needed. Code here is a bit involved because
 // we reuse the code from the usual proxy and we need to prepare few structures
 // that this code expects.
@@ -220,72 +239,18 @@ async fn connect_to_compute(
        application_name: Some(APP_NAME),
    };

-    let node_info = &mut creds.wake_compute(&extra).await?.expect("msg");
+    let node_info = creds
+        .wake_compute(&extra)
+        .await?
+        .context("missing cache entry from wake_compute")?;

-    let mut num_retries = 0;
-    let mut wait_duration = time::Duration::ZERO;
-    let mut should_wake_with_error = None;
-    loop {
-        if !wait_duration.is_zero() {
-            time::sleep(wait_duration).await;
-        }
-
-        // try wake the compute node if we have determined it's sensible to do so
-        if let Some(err) = should_wake_with_error.take() {
-            match try_wake(node_info, &extra, &creds).await {
-                // we can't wake up the compute node
-                Ok(None) => return Err(err),
-                // there was an error communicating with the control plane
-                Err(e) => return Err(e.into()),
-                // failed to wake up but we can continue to retry
-                Ok(Some(ControlFlow::Continue(()))) => {
-                    wait_duration = retry_after(num_retries);
-                    should_wake_with_error = Some(err);
-
-                    num_retries += 1;
-                    info!(num_retries, "retrying wake compute");
-                    continue;
-                }
-                // successfully woke up a compute node and can break the wakeup loop
-                Ok(Some(ControlFlow::Break(()))) => {}
-            }
-        }
-
-        match connect_to_compute_once(node_info, conn_info).await {
-            Ok(res) => return Ok(res),
-            Err(e) => {
-                error!(error = ?e, "could not connect to compute node");
-                if !can_retry_error(&e, num_retries) {
-                    return Err(e.into());
-                }
-                wait_duration = retry_after(num_retries);
-
-                // after the first connect failure,
-                // we should invalidate the cache and wake up a new compute node
-                if num_retries == 0 {
-                    invalidate_cache(node_info);
-                    should_wake_with_error = Some(e.into());
-                }
-            }
-        }
-
-        num_retries += 1;
-        info!(num_retries, "retrying connect");
-    }
-}
-
-fn can_retry_error(err: &tokio_postgres::Error, num_retries: u32) -> bool {
-    match err {
-        // retry all errors at least once
-        _ if num_retries == 0 => true,
-        _ if num_retries >= NUM_RETRIES_WAKE_COMPUTE => false,
-        err => can_retry_tokio_postgres_error(err),
-    }
+    crate::proxy::connect_to_compute(&TokioMechanism { conn_info }, node_info, &extra, &creds).await
 }

 async fn connect_to_compute_once(
    node_info: &console::CachedNodeInfo,
    conn_info: &ConnInfo,
+    timeout: time::Duration,
 ) -> Result<tokio_postgres::Client, tokio_postgres::Error> {
    let mut config = (*node_info.config).clone();

@@ -294,6 +259,7 @@ async fn connect_to_compute_once(
        .password(&conn_info.password)
        .dbname(&conn_info.dbname)
        .max_backend_message_size(MAX_RESPONSE_SIZE)
+        .connect_timeout(timeout)
        .connect(tokio_postgres::NoTls)
        .await?;

--- a/proxy/src/http/websocket.rs
+++ b/proxy/src/http/websocket.rs
@@ -1,5 +1,8 @@
 use crate::{
-    cancellation::CancelMap, config::ProxyConfig, error::io_error, proxy::handle_ws_client,
+    cancellation::CancelMap,
+    config::ProxyConfig,
+    error::io_error,
+    proxy::{handle_client, ClientMode},
 };
 use bytes::{Buf, Bytes};
 use futures::{Sink, Stream, StreamExt};
@@ -150,12 +153,12 @@ async fn serve_websocket(
    hostname: Option<String>,
 ) -> anyhow::Result<()> {
    let websocket = websocket.await?;
-    handle_ws_client(
+    handle_client(
        config,
        cancel_map,
        session_id,
        WebSocketRw::new(websocket),
-        hostname,
+        ClientMode::Websockets { hostname },
    )
    .await?;
    Ok(())
@@ -221,6 +224,18 @@ async fn ws_handler(
            );
            r
        })
+    } else if request.uri().path() == "/sql" && request.method() == Method::OPTIONS {
+        Response::builder()
+            .header("Allow", "OPTIONS, POST")
+            .header("Access-Control-Allow-Origin", "*")
+            .header(
+                "Access-Control-Allow-Headers",
+                "Neon-Connection-String, Neon-Raw-Text-Output, Neon-Array-Mode, Neon-Pool-Opt-In",
+            )
+            .header("Access-Control-Max-Age", "86400" /* 24 hours */)
+            .status(StatusCode::OK) // 204 is also valid, but see: https://developer.mozilla.org/en-US/docs/Web/HTTP/Methods/OPTIONS#status_code
+            .body(Body::empty())
+            .map_err(|e| ApiError::BadRequest(e.into()))
    } else {
        json_response(StatusCode::BAD_REQUEST, "query is not supported")
    }
--- a/proxy/src/proxy.rs
+++ b/proxy/src/proxy.rs
@@ -11,16 +11,16 @@ use crate::{
        errors::{ApiError, WakeComputeError},
        messages::MetricsAuxInfo,
    },
-    error::io_error,
    stream::{PqStream, Stream},
 };
 use anyhow::{bail, Context};
+use async_trait::async_trait;
 use futures::TryFutureExt;
 use hyper::StatusCode;
 use metrics::{register_int_counter, register_int_counter_vec, IntCounter, IntCounterVec};
 use once_cell::sync::Lazy;
 use pq_proto::{BeMessage as Be, FeStartupPacket, StartupMessageParams};
-use std::{error::Error, ops::ControlFlow, sync::Arc};
+use std::{error::Error, io, ops::ControlFlow, sync::Arc};
 use tokio::{
    io::{AsyncRead, AsyncWrite, AsyncWriteExt},
    time,
@@ -31,7 +31,8 @@ use utils::measured_stream::MeasuredStream;

 /// Number of times we should retry the `/proxy_wake_compute` http request.
 /// Retry duration is BASE_RETRY_WAIT_DURATION * 1.5^n
-pub const NUM_RETRIES_WAKE_COMPUTE: u32 = 10;
+const NUM_RETRIES_CONNECT: u32 = 10;
+const CONNECT_TIMEOUT: time::Duration = time::Duration::from_secs(2);
 const BASE_RETRY_WAIT_DURATION: time::Duration = time::Duration::from_millis(100);

 const ERR_INSECURE_CONNECTION: &str = "connection is insecure (try using `sslmode=require`)";
@@ -103,7 +104,8 @@ pub async fn task_main(
                            .set_nodelay(true)
                            .context("failed to set socket option")?;

-                        handle_client(config, &cancel_map, session_id, socket).await
+                        handle_client(config, &cancel_map, session_id, socket, ClientMode::Tcp)
+                        .await
                    }
                    .unwrap_or_else(move |e| {
                        // Acknowledge that the task has finished with an error.
@@ -128,14 +130,50 @@ pub async fn task_main(
    Ok(())
 }

-// TODO(tech debt): unite this with its twin below.
+pub enum ClientMode {
+    Tcp,
+    Websockets { hostname: Option<String> },
+}
+
+/// Abstracts the logic of handling TCP vs WS clients
+impl ClientMode {
+    fn allow_cleartext(&self) -> bool {
+        match self {
+            ClientMode::Tcp => false,
+            ClientMode::Websockets { .. } => true,
+        }
+    }
+
+    fn allow_self_signed_compute(&self, config: &ProxyConfig) -> bool {
+        match self {
+            ClientMode::Tcp => config.allow_self_signed_compute,
+            ClientMode::Websockets { .. } => false,
+        }
+    }
+
+    fn hostname<'a, S>(&'a self, s: &'a Stream<S>) -> Option<&'a str> {
+        match self {
+            ClientMode::Tcp => s.sni_hostname(),
+            ClientMode::Websockets { hostname } => hostname.as_deref(),
+        }
+    }
+
+    fn handshake_tls<'a>(&self, tls: Option<&'a TlsConfig>) -> Option<&'a TlsConfig> {
+        match self {
+            ClientMode::Tcp => tls,
+            // TLS is None here if using websockets, because the connection is already encrypted.
+            ClientMode::Websockets { .. } => None,
+        }
+    }
+}
+
 #[tracing::instrument(fields(session_id = ?session_id), skip_all)]
-pub async fn handle_ws_client(
+pub async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
    config: &'static ProxyConfig,
    cancel_map: &CancelMap,
    session_id: uuid::Uuid,
-    stream: impl AsyncRead + AsyncWrite + Unpin,
-    hostname: Option<String>,
+    stream: S,
+    mode: ClientMode,
 ) -> anyhow::Result<()> {
    // The `closed` counter will increase when this future is destroyed.
    NUM_CONNECTIONS_ACCEPTED_COUNTER.inc();
@@ -144,10 +182,8 @@ pub async fn handle_ws_client(
    }

    let tls = config.tls_config.as_ref();
-    let hostname = hostname.as_deref();

-    // TLS is None here, because the connection is already encrypted.
-    let do_handshake = handshake(stream, None, cancel_map);
+    let do_handshake = handshake(stream, mode.handshake_tls(tls), cancel_map);
    let (mut stream, params) = match do_handshake.await? {
        Some(x) => x,
        None => return Ok(()), // it's a cancellation request
@@ -155,6 +191,7 @@ pub async fn handle_ws_client(

    // Extract credentials which we're going to use for auth.
    let creds = {
+        let hostname = mode.hostname(stream.get_ref());
        let common_names = tls.and_then(|tls| tls.common_names.clone());
        let result = config
            .auth_backend
@@ -168,59 +205,15 @@ pub async fn handle_ws_client(
        }
    };

-    let client = Client::new(stream, creds, &params, session_id, false);
-    cancel_map
-        .with_session(|session| client.connect_to_db(session, true))
-        .await
-}
-
-#[tracing::instrument(fields(session_id = ?session_id), skip_all)]
-async fn handle_client(
-    config: &'static ProxyConfig,
-    cancel_map: &CancelMap,
-    session_id: uuid::Uuid,
-    stream: impl AsyncRead + AsyncWrite + Unpin,
-) -> anyhow::Result<()> {
-    // The `closed` counter will increase when this future is destroyed.
-    NUM_CONNECTIONS_ACCEPTED_COUNTER.inc();
-    scopeguard::defer! {
-        NUM_CONNECTIONS_CLOSED_COUNTER.inc();
-    }
-
-    let tls = config.tls_config.as_ref();
-    let do_handshake = handshake(stream, tls, cancel_map);
-    let (mut stream, params) = match do_handshake.await? {
-        Some(x) => x,
-        None => return Ok(()), // it's a cancellation request
-    };
-
-    // Extract credentials which we're going to use for auth.
-    let creds = {
-        let sni = stream.get_ref().sni_hostname();
-        let common_names = tls.and_then(|tls| tls.common_names.clone());
-        let result = config
-            .auth_backend
-            .as_ref()
-            .map(|_| auth::ClientCredentials::parse(&params, sni, common_names))
-            .transpose();
-
-        match result {
-            Ok(creds) => creds,
-            Err(e) => stream.throw_error(e).await?,
-        }
-    };
-
-    let allow_self_signed_compute = config.allow_self_signed_compute;
-
    let client = Client::new(
        stream,
        creds,
        &params,
        session_id,
-        allow_self_signed_compute,
+        mode.allow_self_signed_compute(config),
    );
    cancel_map
-        .with_session(|session| client.connect_to_db(session, false))
+        .with_session(|session| client.connect_to_db(session, mode.allow_cleartext()))
        .await
 }

@@ -303,18 +296,18 @@ async fn handshake<S: AsyncRead + AsyncWrite + Unpin>(
 /// (e.g. the compute node's address might've changed at the wrong time).
 /// Invalidate the cache entry (if any) to prevent subsequent errors.
 #[tracing::instrument(name = "invalidate_cache", skip_all)]
-pub fn invalidate_cache(node_info: &console::CachedNodeInfo) {
+pub fn invalidate_cache(node_info: console::CachedNodeInfo) -> compute::ConnCfg {
    let is_cached = node_info.cached();
    if is_cached {
        warn!("invalidating stalled compute node info cache entry");
-        node_info.invalidate();
    }
-
    let label = match is_cached {
        true => "compute_cached",
        false => "compute_uncached",
    };
    NUM_CONNECTION_FAILURES.with_label_values(&[label]).inc();
+
+    node_info.invalidate().config
 }

 /// Try to connect to the compute node once.
@@ -331,89 +324,118 @@ async fn connect_to_compute_once(
        .await
 }

+enum ConnectionState<E> {
+    Cached(console::CachedNodeInfo),
+    Invalid(compute::ConnCfg, E),
+}
+
+#[async_trait]
+pub trait ConnectMechanism {
+    type Connection;
+    type ConnectError;
+    type Error: From<Self::ConnectError>;
+    async fn connect_once(
+        &self,
+        node_info: &console::CachedNodeInfo,
+        timeout: time::Duration,
+    ) -> Result<Self::Connection, Self::ConnectError>;
+
+    fn update_connect_config(&self, conf: &mut compute::ConnCfg);
+}
+
+pub struct TcpMechanism<'a> {
+    /// KV-dictionary with PostgreSQL connection params.
+    pub params: &'a StartupMessageParams,
+}
+
+#[async_trait]
+impl ConnectMechanism for TcpMechanism<'_> {
+    type Connection = PostgresConnection;
+    type ConnectError = compute::ConnectionError;
+    type Error = compute::ConnectionError;
+
+    async fn connect_once(
+        &self,
+        node_info: &console::CachedNodeInfo,
+        timeout: time::Duration,
+    ) -> Result<PostgresConnection, Self::Error> {
+        connect_to_compute_once(node_info, timeout).await
+    }
+
+    fn update_connect_config(&self, config: &mut compute::ConnCfg) {
+        config.set_startup_params(self.params);
+    }
+}
+
 /// Try to connect to the compute node, retrying if necessary.
 /// This function might update `node_info`, so we take it by `&mut`.
 #[tracing::instrument(skip_all)]
-async fn connect_to_compute(
-    node_info: &mut console::CachedNodeInfo,
-    params: &StartupMessageParams,
+pub async fn connect_to_compute<M: ConnectMechanism>(
+    mechanism: &M,
+    mut node_info: console::CachedNodeInfo,
    extra: &console::ConsoleReqExtra<'_>,
    creds: &auth::BackendType<'_, auth::ClientCredentials<'_>>,
-) -> Result<PostgresConnection, compute::ConnectionError> {
+) -> Result<M::Connection, M::Error>
+where
+    M::ConnectError: ShouldRetry + std::fmt::Debug,
+    M::Error: From<WakeComputeError>,
+{
+    mechanism.update_connect_config(&mut node_info.config);
+
    let mut num_retries = 0;
-    let mut wait_duration = time::Duration::ZERO;
-    let mut should_wake_with_error = None;
+    let mut state = ConnectionState::<M::ConnectError>::Cached(node_info);
+
    loop {
-        // Apply startup params to the (possibly, cached) compute node info.
-        node_info.config.set_startup_params(params);
+        match state {
+            ConnectionState::Invalid(config, err) => {
+                match try_wake(&config, extra, creds).await {
+                    // we can't wake up the compute node
+                    Ok(None) => return Err(err.into()),
+                    // there was an error communicating with the control plane
+                    Err(e) => return Err(e.into()),
+                    // failed to wake up but we can continue to retry
+                    Ok(Some(ControlFlow::Continue(()))) => {
+                        state = ConnectionState::Invalid(config, err);
+                        let wait_duration = retry_after(num_retries);
+                        num_retries += 1;

-        if !wait_duration.is_zero() {
-            time::sleep(wait_duration).await;
-        }
-
-        // try wake the compute node if we have determined it's sensible to do so
-        if let Some(err) = should_wake_with_error.take() {
-            match try_wake(node_info, extra, creds).await {
-                // we can't wake up the compute node
-                Ok(None) => return Err(err),
-                // there was an error communicating with the control plane
-                Err(e) => return Err(io_error(e).into()),
-                // failed to wake up but we can continue to retry
-                Ok(Some(ControlFlow::Continue(()))) => {
-                    wait_duration = retry_after(num_retries);
-                    should_wake_with_error = Some(err);
-
-                    num_retries += 1;
-                    info!(num_retries, "retrying wake compute");
-                    continue;
+                        info!(num_retries, "retrying wake compute");
+                        time::sleep(wait_duration).await;
+                        continue;
+                    }
+                    // successfully woke up a compute node and can break the wakeup loop
+                    Ok(Some(ControlFlow::Break(mut node_info))) => {
+                        mechanism.update_connect_config(&mut node_info.config);
+                        state = ConnectionState::Cached(node_info)
+                    }
                }
-                // successfully woke up a compute node and can break the wakeup loop
-                Ok(Some(ControlFlow::Break(()))) => {}
            }
-        }
+            ConnectionState::Cached(node_info) => {
+                match mechanism.connect_once(&node_info, CONNECT_TIMEOUT).await {
+                    Ok(res) => return Ok(res),
+                    Err(e) => {
+                        error!(error = ?e, "could not connect to compute node");
+                        if !e.should_retry(num_retries) {
+                            return Err(e.into());
+                        }

-        // Set a shorter timeout for the initial connection attempt.
-        //
-        // In case we try to connect to an outdated address that is no longer valid, the
-        // default behavior of Kubernetes is to drop the packets, causing us to wait for
-        // the entire timeout period. We want to fail fast in such cases.
-        //
-        // A specific case to consider is when we have cached compute node information
-        // with a 4-minute TTL (Time To Live), but the user has executed a `/suspend` API
-        // call, resulting in the nonexistence of the compute node.
-        //
-        // We only use caching in case of scram proxy backed by the console, so reduce
-        // the timeout only in that case.
-        let is_scram_proxy = matches!(creds, auth::BackendType::Console(_, _));
-        let timeout = if is_scram_proxy && num_retries == 0 {
-            time::Duration::from_secs(2)
-        } else {
-            time::Duration::from_secs(10)
-        };
+                        // after the first connect failure,
+                        // we should invalidate the cache and wake up a new compute node
+                        if num_retries == 0 {
+                            state = ConnectionState::Invalid(invalidate_cache(node_info), e);
+                        } else {
+                            state = ConnectionState::Cached(node_info);
+                        }

-        // do this again to ensure we have username?
-        node_info.config.set_startup_params(params);
+                        let wait_duration = retry_after(num_retries);
+                        num_retries += 1;

-        match connect_to_compute_once(node_info, timeout).await {
-            Ok(res) => return Ok(res),
-            Err(e) => {
-                error!(error = ?e, "could not connect to compute node");
-                if !can_retry_error(&e, num_retries) {
-                    return Err(e);
-                }
-                wait_duration = retry_after(num_retries);
-
-                // after the first connect failure,
-                // we should invalidate the cache and wake up a new compute node
-                if num_retries == 0 {
-                    invalidate_cache(node_info);
-                    should_wake_with_error = Some(e);
+                        info!(num_retries, "retrying wake compute");
+                        time::sleep(wait_duration).await;
+                    }
                }
            }
        }
-
-        num_retries += 1;
-        info!(num_retries, "retrying connect");
    }
 }

@@ -421,11 +443,11 @@ async fn connect_to_compute(
 /// * Returns Ok(Some(true)) if there was an error waking but retries are acceptable
 /// * Returns Ok(Some(false)) if the wakeup succeeded
 /// * Returns Ok(None) or Err(e) if there was an error
-pub async fn try_wake(
-    node_info: &mut console::CachedNodeInfo,
+async fn try_wake(
+    config: &compute::ConnCfg,
    extra: &console::ConsoleReqExtra<'_>,
    creds: &auth::BackendType<'_, auth::ClientCredentials<'_>>,
-) -> Result<Option<ControlFlow<()>>, WakeComputeError> {
+) -> Result<Option<ControlFlow<console::CachedNodeInfo>>, WakeComputeError> {
    info!("compute node's state has likely changed; requesting a wake-up");
    match creds.wake_compute(extra).await {
        // retry wake if the compute was in an invalid state
@@ -435,53 +457,69 @@ pub async fn try_wake(
        })) => Ok(Some(ControlFlow::Continue(()))),
        // Update `node_info` and try again.
        Ok(Some(mut new)) => {
-            new.config.reuse_password(&node_info.config);
-            *node_info = new;
-            Ok(Some(ControlFlow::Break(())))
+            new.config.reuse_password(config);
+            Ok(Some(ControlFlow::Break(new)))
        }
        Err(e) => Err(e),
        Ok(None) => Ok(None),
    }
 }

-fn can_retry_error(err: &compute::ConnectionError, num_retries: u32) -> bool {
-    match err {
-        // retry all errors at least once
-        _ if num_retries == 0 => true,
-        _ if num_retries >= NUM_RETRIES_WAKE_COMPUTE => false,
-        compute::ConnectionError::Postgres(err) => can_retry_tokio_postgres_error(err),
-        compute::ConnectionError::CouldNotConnect(err) => is_io_connection_err(err),
-        _ => false,
+pub trait ShouldRetry {
+    fn could_retry(&self) -> bool;
+    fn should_retry(&self, num_retries: u32) -> bool {
+        match self {
+            // retry all errors at least once
+            _ if num_retries == 0 => true,
+            _ if num_retries >= NUM_RETRIES_CONNECT => false,
+            err => err.could_retry(),
+        }
    }
 }

-pub fn can_retry_tokio_postgres_error(err: &tokio_postgres::Error) -> bool {
-    if let Some(io_err) = err.source().and_then(|x| x.downcast_ref()) {
-        is_io_connection_err(io_err)
-    } else if let Some(db_err) = err.source().and_then(|x| x.downcast_ref()) {
-        is_sql_connection_err(db_err)
-    } else {
-        false
+impl ShouldRetry for io::Error {
+    fn could_retry(&self) -> bool {
+        use std::io::ErrorKind;
+        matches!(
+            self.kind(),
+            ErrorKind::ConnectionRefused | ErrorKind::AddrNotAvailable | ErrorKind::TimedOut
+        )
    }
 }

-fn is_sql_connection_err(err: &tokio_postgres::error::DbError) -> bool {
-    use tokio_postgres::error::SqlState;
-    matches!(
-        err.code(),
-        &SqlState::CONNECTION_FAILURE
-            | &SqlState::CONNECTION_EXCEPTION
-            | &SqlState::CONNECTION_DOES_NOT_EXIST
-            | &SqlState::SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION,
-    )
+impl ShouldRetry for tokio_postgres::error::DbError {
+    fn could_retry(&self) -> bool {
+        use tokio_postgres::error::SqlState;
+        matches!(
+            self.code(),
+            &SqlState::CONNECTION_FAILURE
+                | &SqlState::CONNECTION_EXCEPTION
+                | &SqlState::CONNECTION_DOES_NOT_EXIST
+                | &SqlState::SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION,
+        )
+    }
 }

-fn is_io_connection_err(err: &std::io::Error) -> bool {
-    use std::io::ErrorKind;
-    matches!(
-        err.kind(),
-        ErrorKind::ConnectionRefused | ErrorKind::AddrNotAvailable | ErrorKind::TimedOut
-    )
+impl ShouldRetry for tokio_postgres::Error {
+    fn could_retry(&self) -> bool {
+        if let Some(io_err) = self.source().and_then(|x| x.downcast_ref()) {
+            io::Error::could_retry(io_err)
+        } else if let Some(db_err) = self.source().and_then(|x| x.downcast_ref()) {
+            tokio_postgres::error::DbError::could_retry(db_err)
+        } else {
+            false
+        }
+    }
+}
+
+impl ShouldRetry for compute::ConnectionError {
+    fn could_retry(&self) -> bool {
+        match self {
+            compute::ConnectionError::Postgres(err) => err.could_retry(),
+            compute::ConnectionError::CouldNotConnect(err) => err.could_retry(),
+            _ => false,
+        }
+    }
 }

 pub fn retry_after(num_retries: u32) -> time::Duration {
@@ -637,7 +675,8 @@ impl<S: AsyncRead + AsyncWrite + Unpin> Client<'_, S> {

        node_info.allow_self_signed_compute = allow_self_signed_compute;

-        let mut node = connect_to_compute(&mut node_info, params, &extra, &creds)
+        let aux = node_info.aux.clone();
+        let mut node = connect_to_compute(&TcpMechanism { params }, node_info, &extra, &creds)
            .or_else(|e| stream.throw_error(e))
            .await?;

@@ -648,6 +687,6 @@ impl<S: AsyncRead + AsyncWrite + Unpin> Client<'_, S> {
        // immediately after opening the connection.
        let (stream, read_buf) = stream.into_inner();
        node.stream.write_all(&read_buf).await?;
-        proxy_pass(stream, node.stream, &node_info.aux).await
+        proxy_pass(stream, node.stream, &aux).await
    }
 }
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -36,6 +36,7 @@ pytest-httpserver = "^1.0.8"
 aiohttp = "3.7.4"
 pytest-rerunfailures = "^11.1.2"
 types-pytest-lazy-fixture = "^0.6.3.3"
+pytest-split = "^0.8.1"

 [tool.poetry.group.dev.dependencies]
 black = "^23.3.0"
@@ -78,6 +79,7 @@ module = [
 ignore_missing_imports = true

 [tool.ruff]
+target-version = "py39"
 extend-exclude = ["vendor/"]
 ignore = ["E501"]
 select = [
@@ -85,4 +87,5 @@ select = [
    "F", # Pyflakes
    "I", # isort
    "W", # pycodestyle
+    "B", # bugbear
 ]
--- a/safekeeper/src/bin/safekeeper.rs
+++ b/safekeeper/src/bin/safekeeper.rs
@@ -37,7 +37,7 @@ use safekeeper::{http, WAL_REMOVER_RUNTIME};
 use safekeeper::{remove_wal, WAL_BACKUP_RUNTIME};
 use safekeeper::{wal_backup, HTTP_RUNTIME};
 use storage_broker::DEFAULT_ENDPOINT;
-use utils::auth::JwtAuth;
+use utils::auth::{JwtAuth, Scope};
 use utils::{
    id::NodeId,
    logging::{self, LogFormat},
@@ -72,6 +72,10 @@ struct Args {
    /// Listen endpoint for receiving/sending WAL in the form host:port.
    #[arg(short, long, default_value = DEFAULT_PG_LISTEN_ADDR)]
    listen_pg: String,
+    /// Listen endpoint for receiving/sending WAL in the form host:port allowing
+    /// only tenant scoped auth tokens. Pointless if auth is disabled.
+    #[arg(long, default_value = None, verbatim_doc_comment)]
+    listen_pg_tenant_only: Option<String>,
    /// Listen http endpoint for management and metrics in the form host:port.
    #[arg(long, default_value = DEFAULT_HTTP_LISTEN_ADDR)]
    listen_http: String,
@@ -94,7 +98,7 @@ struct Args {
    broker_keepalive_interval: Duration,
    /// Peer safekeeper is considered dead after not receiving heartbeats from
    /// it during this period passed as a human readable duration.
-    #[arg(long, value_parser= humantime::parse_duration, default_value = DEFAULT_HEARTBEAT_TIMEOUT)]
+    #[arg(long, value_parser= humantime::parse_duration, default_value = DEFAULT_HEARTBEAT_TIMEOUT, verbatim_doc_comment)]
    heartbeat_timeout: Duration,
    /// Remote storage configuration for WAL backup (offloading to s3) as TOML
    /// inline table, e.g.
@@ -179,6 +183,7 @@ async fn main() -> anyhow::Result<()> {
        workdir,
        my_id: id,
        listen_pg_addr: args.listen_pg,
+        listen_pg_addr_tenant_only: args.listen_pg_tenant_only,
        listen_http_addr: args.listen_http,
        availability_zone: args.availability_zone,
        no_sync: args.no_sync,
@@ -222,6 +227,21 @@ async fn start_safekeeper(conf: SafeKeeperConf) -> Result<()> {
        e
    })?;

+    let pg_listener_tenant_only =
+        if let Some(listen_pg_addr_tenant_only) = &conf.listen_pg_addr_tenant_only {
+            info!(
+                "starting safekeeper tenant scoped WAL service on {}",
+                listen_pg_addr_tenant_only
+            );
+            let listener = tcp_listener::bind(listen_pg_addr_tenant_only.clone()).map_err(|e| {
+                error!("failed to bind to address {}: {}", conf.listen_pg_addr, e);
+                e
+            })?;
+            Some(listener)
+        } else {
+            None
+        };
+
    info!(
        "starting safekeeper HTTP service on {}",
        conf.listen_http_addr
@@ -253,14 +273,34 @@ async fn start_safekeeper(conf: SafeKeeperConf) -> Result<()> {
    let current_thread_rt = conf
        .current_thread_runtime
        .then(|| Handle::try_current().expect("no runtime in main"));
+
    let wal_service_handle = current_thread_rt
        .as_ref()
        .unwrap_or_else(|| WAL_SERVICE_RUNTIME.handle())
-        .spawn(wal_service::task_main(conf_, pg_listener))
+        .spawn(wal_service::task_main(
+            conf_,
+            pg_listener,
+            Some(Scope::SafekeeperData),
+        ))
        // wrap with task name for error reporting
        .map(|res| ("WAL service main".to_owned(), res));
    tasks_handles.push(Box::pin(wal_service_handle));

+    if let Some(pg_listener_tenant_only) = pg_listener_tenant_only {
+        let conf_ = conf.clone();
+        let wal_service_handle = current_thread_rt
+            .as_ref()
+            .unwrap_or_else(|| WAL_SERVICE_RUNTIME.handle())
+            .spawn(wal_service::task_main(
+                conf_,
+                pg_listener_tenant_only,
+                Some(Scope::Tenant),
+            ))
+            // wrap with task name for error reporting
+            .map(|res| ("WAL service tenant only main".to_owned(), res));
+        tasks_handles.push(Box::pin(wal_service_handle));
+    }
+
    let conf_ = conf.clone();
    let http_handle = current_thread_rt
        .as_ref()
--- a/safekeeper/src/handler.rs
+++ b/safekeeper/src/handler.rs
@@ -34,6 +34,8 @@ pub struct SafekeeperPostgresHandler {
    pub ttid: TenantTimelineId,
    /// Unique connection id is logged in spans for observability.
    pub conn_id: ConnectionId,
+    /// Auth scope allowed on the connections. None if auth is not configured.
+    allowed_auth_scope: Option<Scope>,
    claims: Option<Claims>,
    io_metrics: Option<TrafficMetrics>,
 }
@@ -147,6 +149,16 @@ impl<IO: AsyncRead + AsyncWrite + Unpin + Send> postgres_backend::Handler<IO>
            .unwrap()
            .decode(str::from_utf8(jwt_response).context("jwt response is not UTF-8")?)?;

+        let scope = self
+            .allowed_auth_scope
+            .expect("auth is enabled but scope is not configured");
+        // The handler might be configured to allow only tenant scope tokens.
+        if matches!(scope, Scope::Tenant) && !matches!(data.claims.scope, Scope::Tenant) {
+            return Err(QueryError::Other(anyhow::anyhow!(
+                "passed JWT token is for full access, but only tenant scope is allowed"
+            )));
+        }
+
        if matches!(data.claims.scope, Scope::Tenant) && data.claims.tenant_id.is_none() {
            return Err(QueryError::Other(anyhow::anyhow!(
                "jwt token scope is Tenant, but tenant id is missing"
@@ -215,7 +227,12 @@ impl<IO: AsyncRead + AsyncWrite + Unpin + Send> postgres_backend::Handler<IO>
 }

 impl SafekeeperPostgresHandler {
-    pub fn new(conf: SafeKeeperConf, conn_id: u32, io_metrics: Option<TrafficMetrics>) -> Self {
+    pub fn new(
+        conf: SafeKeeperConf,
+        conn_id: u32,
+        io_metrics: Option<TrafficMetrics>,
+        allowed_auth_scope: Option<Scope>,
+    ) -> Self {
        SafekeeperPostgresHandler {
            conf,
            appname: None,
@@ -224,6 +241,7 @@ impl SafekeeperPostgresHandler {
            ttid: TenantTimelineId::empty(),
            conn_id,
            claims: None,
+            allowed_auth_scope,
            io_metrics,
        }
    }
--- a/safekeeper/src/lib.rs
+++ b/safekeeper/src/lib.rs
@@ -53,6 +53,7 @@ pub struct SafeKeeperConf {
    pub workdir: PathBuf,
    pub my_id: NodeId,
    pub listen_pg_addr: String,
+    pub listen_pg_addr_tenant_only: Option<String>,
    pub listen_http_addr: String,
    pub availability_zone: Option<String>,
    pub no_sync: bool,
@@ -85,6 +86,7 @@ impl SafeKeeperConf {
            workdir: PathBuf::from("./"),
            no_sync: false,
            listen_pg_addr: defaults::DEFAULT_PG_LISTEN_ADDR.to_string(),
+            listen_pg_addr_tenant_only: None,
            listen_http_addr: defaults::DEFAULT_HTTP_LISTEN_ADDR.to_string(),
            availability_zone: None,
            remote_storage: None,
--- a/safekeeper/src/wal_service.rs
+++ b/safekeeper/src/wal_service.rs
@@ -8,7 +8,7 @@ use std::{future, time::Duration};
 use tokio::net::TcpStream;
 use tokio_io_timeout::TimeoutReader;
 use tracing::*;
-use utils::measured_stream::MeasuredStream;
+use utils::{auth::Scope, measured_stream::MeasuredStream};

 use crate::handler::SafekeeperPostgresHandler;
 use crate::metrics::TrafficMetrics;
@@ -19,6 +19,7 @@ use postgres_backend::{AuthType, PostgresBackend};
 pub async fn task_main(
    conf: SafeKeeperConf,
    pg_listener: std::net::TcpListener,
+    allowed_auth_scope: Option<Scope>,
 ) -> anyhow::Result<()> {
    // Tokio's from_std won't do this for us, per its comment.
    pg_listener.set_nonblocking(true)?;
@@ -33,7 +34,7 @@ pub async fn task_main(
        let conn_id = issue_connection_id(&mut connection_count);

        tokio::spawn(async move {
-            if let Err(err) = handle_socket(socket, conf, conn_id)
+            if let Err(err) = handle_socket(socket, conf, conn_id, allowed_auth_scope)
                .instrument(info_span!("", cid = %conn_id))
                .await
            {
@@ -49,6 +50,7 @@ async fn handle_socket(
    socket: TcpStream,
    conf: SafeKeeperConf,
    conn_id: ConnectionId,
+    allowed_auth_scope: Option<Scope>,
 ) -> Result<(), QueryError> {
    socket.set_nodelay(true)?;
    let peer_addr = socket.peer_addr()?;
@@ -84,8 +86,12 @@ async fn handle_socket(
        None => AuthType::Trust,
        Some(_) => AuthType::NeonJWT,
    };
-    let mut conn_handler =
-        SafekeeperPostgresHandler::new(conf, conn_id, Some(traffic_metrics.clone()));
+    let mut conn_handler = SafekeeperPostgresHandler::new(
+        conf,
+        conn_id,
+        Some(traffic_metrics.clone()),
+        allowed_auth_scope,
+    );
    let pgbackend = PostgresBackend::new_from_io(socket, peer_addr, auth_type, None)?;
    // libpq protocol between safekeeper and walproposer / pageserver
    // We don't use shutdown.
--- a/safekeeper/src/wal_storage.rs
+++ b/safekeeper/src/wal_storage.rs
@@ -149,11 +149,6 @@ impl PhysicalStorage {
                    wal_seg_size,
                    state.commit_lsn,
                )?,
-                16 => postgres_ffi::v16::xlog_utils::find_end_of_wal(
-                    &timeline_dir,
-                    wal_seg_size,
-                    state.commit_lsn,
-                )?,
                _ => bail!("unsupported postgres version: {}", state.server.pg_version),
            }
        };
--- a/scripts/benchmark_durations.py
+++ b/scripts/benchmark_durations.py
@@ -0,0 +1,177 @@
+#! /usr/bin/env python3
+
+import argparse
+import json
+import logging
+from typing import Dict
+
+import psycopg2
+import psycopg2.extras
+
+"""
+The script fetches the durations of benchmarks from the database and stores it in a file compatible with pytest-split plugin.
+"""
+
+
+BENCHMARKS_DURATION_QUERY = """
+    SELECT
+        DISTINCT parent_suite, suite, test,
+        PERCENTILE_DISC(%s) WITHIN GROUP (ORDER BY duration_ms) as percentile_ms
+    FROM
+        (
+            SELECT
+                jsonb_array_elements(data -> 'children') ->> 'name' as parent_suite,
+                jsonb_array_elements(jsonb_array_elements(data -> 'children') -> 'children') ->> 'name' as suite,
+                jsonb_array_elements(jsonb_array_elements(jsonb_array_elements(data -> 'children') -> 'children') -> 'children') ->> 'name' as test,
+                jsonb_array_elements(jsonb_array_elements(jsonb_array_elements(data -> 'children') -> 'children') -> 'children') ->> 'status' as status,
+                to_timestamp((jsonb_array_elements(jsonb_array_elements(jsonb_array_elements(data -> 'children') -> 'children') -> 'children') -> 'time' -> 'start')::bigint / 1000)::date as timestamp,
+                (jsonb_array_elements(jsonb_array_elements(jsonb_array_elements(data -> 'children') -> 'children') -> 'children') -> 'time' -> 'duration')::int as duration_ms
+            FROM
+                regress_test_results
+            WHERE
+                reference = 'refs/heads/main'
+        ) data
+    WHERE
+        timestamp > CURRENT_DATE - INTERVAL '%s' day
+        AND parent_suite = 'test_runner.performance'
+        AND status = 'passed'
+    GROUP BY
+        parent_suite, suite, test
+    ;
+"""
+
+# For out benchmarks the default distibution for 4 worked produces pretty uneven chunks,
+# the total duration varies from 8 to 40 minutes.
+# We use some pre-collected durations as a fallback to have a better distribution.
+FALLBACK_DURATION = {
+    "test_runner/performance/test_branch_creation.py::test_branch_creation_heavy_write[20]": 57.0,
+    "test_runner/performance/test_branch_creation.py::test_branch_creation_many_relations": 28.0,
+    "test_runner/performance/test_branch_creation.py::test_branch_creation_many[1024]": 71.0,
+    "test_runner/performance/test_branching.py::test_compare_child_and_root_pgbench_perf": 27.0,
+    "test_runner/performance/test_branching.py::test_compare_child_and_root_read_perf": 11.0,
+    "test_runner/performance/test_branching.py::test_compare_child_and_root_write_perf": 30.0,
+    "test_runner/performance/test_bulk_insert.py::test_bulk_insert[neon]": 40.0,
+    "test_runner/performance/test_bulk_insert.py::test_bulk_insert[vanilla]": 5.0,
+    "test_runner/performance/test_bulk_tenant_create.py::test_bulk_tenant_create[1]": 3.0,
+    "test_runner/performance/test_bulk_tenant_create.py::test_bulk_tenant_create[5]": 10.0,
+    "test_runner/performance/test_bulk_tenant_create.py::test_bulk_tenant_create[10]": 19.0,
+    "test_runner/performance/test_bulk_update.py::test_bulk_update[10]": 66.0,
+    "test_runner/performance/test_bulk_update.py::test_bulk_update[50]": 30.0,
+    "test_runner/performance/test_bulk_update.py::test_bulk_update[100]": 60.0,
+    "test_runner/performance/test_compaction.py::test_compaction": 77.0,
+    "test_runner/performance/test_compare_pg_stats.py::test_compare_pg_stats_ro_with_pgbench_select_only[neon-5-10-100]": 11.0,
+    "test_runner/performance/test_compare_pg_stats.py::test_compare_pg_stats_ro_with_pgbench_select_only[vanilla-5-10-100]": 16.0,
+    "test_runner/performance/test_compare_pg_stats.py::test_compare_pg_stats_rw_with_pgbench_default[neon-5-10-100]": 11.0,
+    "test_runner/performance/test_compare_pg_stats.py::test_compare_pg_stats_rw_with_pgbench_default[vanilla-5-10-100]": 18.0,
+    "test_runner/performance/test_compare_pg_stats.py::test_compare_pg_stats_wal_with_pgbench_default[neon-5-10-100]": 11.0,
+    "test_runner/performance/test_compare_pg_stats.py::test_compare_pg_stats_wal_with_pgbench_default[vanilla-5-10-100]": 16.0,
+    "test_runner/performance/test_compare_pg_stats.py::test_compare_pg_stats_wo_with_heavy_write[neon-10-1]": 11.0,
+    "test_runner/performance/test_compare_pg_stats.py::test_compare_pg_stats_wo_with_heavy_write[neon-10-10]": 11.0,
+    "test_runner/performance/test_compare_pg_stats.py::test_compare_pg_stats_wo_with_heavy_write[vanilla-10-1]": 10.0,
+    "test_runner/performance/test_compare_pg_stats.py::test_compare_pg_stats_wo_with_heavy_write[vanilla-10-10]": 10.0,
+    "test_runner/performance/test_compare_pg_stats.py::test_compare_pg_stats_wo_with_pgbench_simple_update[neon-5-10-100]": 11.0,
+    "test_runner/performance/test_compare_pg_stats.py::test_compare_pg_stats_wo_with_pgbench_simple_update[vanilla-5-10-100]": 16.0,
+    "test_runner/performance/test_copy.py::test_copy[neon]": 12.0,
+    "test_runner/performance/test_copy.py::test_copy[vanilla]": 10.0,
+    "test_runner/performance/test_gc_feedback.py::test_gc_feedback": 284.0,
+    "test_runner/performance/test_gist_build.py::test_gist_buffering_build[neon]": 11.0,
+    "test_runner/performance/test_gist_build.py::test_gist_buffering_build[vanilla]": 7.0,
+    "test_runner/performance/test_latency.py::test_measure_read_latency_heavy_write_workload[neon-1]": 85.0,
+    "test_runner/performance/test_latency.py::test_measure_read_latency_heavy_write_workload[vanilla-1]": 29.0,
+    "test_runner/performance/test_layer_map.py::test_layer_map": 44.0,
+    "test_runner/performance/test_parallel_copy_to.py::test_parallel_copy_different_tables[neon]": 16.0,
+    "test_runner/performance/test_parallel_copy_to.py::test_parallel_copy_different_tables[vanilla]": 67.0,
+    "test_runner/performance/test_parallel_copy_to.py::test_parallel_copy_same_table[neon]": 67.0,
+    "test_runner/performance/test_parallel_copy_to.py::test_parallel_copy_same_table[vanilla]": 80.0,
+    "test_runner/performance/test_perf_pgbench.py::test_pgbench[neon-45-10]": 102.0,
+    "test_runner/performance/test_perf_pgbench.py::test_pgbench[vanilla-45-10]": 99.0,
+    "test_runner/performance/test_random_writes.py::test_random_writes[neon]": 9.0,
+    "test_runner/performance/test_random_writes.py::test_random_writes[vanilla]": 2.0,
+    "test_runner/performance/test_seqscans.py::test_seqscans[neon-100000-100-0]": 4.0,
+    "test_runner/performance/test_seqscans.py::test_seqscans[neon-10000000-1-0]": 80.0,
+    "test_runner/performance/test_seqscans.py::test_seqscans[neon-10000000-1-4]": 68.0,
+    "test_runner/performance/test_seqscans.py::test_seqscans[vanilla-100000-100-0]": 0.0,
+    "test_runner/performance/test_seqscans.py::test_seqscans[vanilla-10000000-1-0]": 11.0,
+    "test_runner/performance/test_seqscans.py::test_seqscans[vanilla-10000000-1-4]": 10.0,
+    "test_runner/performance/test_startup.py::test_startup_simple": 2.0,
+    "test_runner/performance/test_startup.py::test_startup": 539.0,
+    "test_runner/performance/test_wal_backpressure.py::test_heavy_write_workload[neon_off-10-5-5]": 375.0,
+    "test_runner/performance/test_wal_backpressure.py::test_heavy_write_workload[neon_on-10-5-5]": 370.0,
+    "test_runner/performance/test_wal_backpressure.py::test_heavy_write_workload[vanilla-10-5-5]": 94.0,
+    "test_runner/performance/test_wal_backpressure.py::test_pgbench_intensive_init_workload[neon_off-1000]": 164.0,
+    "test_runner/performance/test_wal_backpressure.py::test_pgbench_intensive_init_workload[neon_on-1000]": 274.0,
+    "test_runner/performance/test_wal_backpressure.py::test_pgbench_intensive_init_workload[vanilla-1000]": 949.0,
+    "test_runner/performance/test_wal_backpressure.py::test_pgbench_simple_update_workload[neon_off-45-100]": 142.0,
+    "test_runner/performance/test_wal_backpressure.py::test_pgbench_simple_update_workload[neon_on-45-100]": 151.0,
+    "test_runner/performance/test_wal_backpressure.py::test_pgbench_simple_update_workload[vanilla-45-100]": 182.0,
+    "test_runner/performance/test_write_amplification.py::test_write_amplification[neon]": 13.0,
+    "test_runner/performance/test_write_amplification.py::test_write_amplification[vanilla]": 16.0,
+}
+
+
+def main(args: argparse.Namespace):
+    connstr = args.connstr
+    interval_days = args.days
+    output = args.output
+    percentile = args.percentile
+
+    res: Dict[str, float] = {}
+
+    try:
+        logging.info("connecting to the database...")
+        with psycopg2.connect(connstr, connect_timeout=30) as conn:
+            with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
+                logging.info("fetching benchmarks...")
+                cur.execute(BENCHMARKS_DURATION_QUERY, (percentile, interval_days))
+                rows = cur.fetchall()
+    except psycopg2.OperationalError as exc:
+        logging.error("cannot fetch benchmarks duration from the DB due to an error", exc)
+        rows = []
+        res = FALLBACK_DURATION
+
+    for row in rows:
+        pytest_name = f"{row['parent_suite'].replace('.', '/')}/{row['suite']}.py::{row['test']}"
+        duration = row["percentile_ms"] / 1000
+        logging.info(f"\t{pytest_name}: {duration}")
+        res[pytest_name] = duration
+
+    logging.info(f"saving results to {output.name}")
+    json.dump(res, output, indent=2)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        description="Get <percentile> of benchmarks duration for the last <N> days"
+    )
+    parser.add_argument(
+        "--output",
+        type=argparse.FileType("w"),
+        default=".test_durations",
+        help="path to output json file (default: .test_durations)",
+    )
+    parser.add_argument(
+        "--percentile",
+        type=float,
+        default="0.99",
+        help="percentile (default: 0.99)",
+    )
+    parser.add_argument(
+        "--days",
+        required=False,
+        default=10,
+        type=int,
+        help="how many days to look back for (default: 10)",
+    )
+    parser.add_argument(
+        "connstr",
+        help="connection string to the test results database",
+    )
+    args = parser.parse_args()
+
+    level = logging.INFO
+    logging.basicConfig(
+        format="%(message)s",
+        level=level,
+    )
+
+    main(args)
--- a/scripts/export_import_between_pageservers.py
+++ b/scripts/export_import_between_pageservers.py
@@ -214,8 +214,7 @@ class VanillaPostgres(PgProtocol):
        assert not self.running
        self.running = True

-        if log_path is None:
-            log_path = os.path.join(self.pgdatadir, "pg.log")
+        log_path = log_path or os.path.join(self.pgdatadir, "pg.log")

        self.pg_bin.run_capture(
            ["pg_ctl", "-w", "-D", str(self.pgdatadir), "-l", log_path, "start"]
@@ -396,7 +395,7 @@ def reconstruct_paths(log_dir, pg_bin, base_tar, port: int):

                query = "select relname, pg_relation_filepath(oid) from pg_class"
                result = vanilla_pg.safe_psql(query, user="cloud_admin", dbname=database)
-                for relname, filepath in result:
+                for _relname, filepath in result:
                    if filepath is not None:
                        if database == "template0copy":
                            # Add all template0copy paths to template0
--- a/test_runner/fixtures/benchmark_fixture.py
+++ b/test_runner/fixtures/benchmark_fixture.py
@@ -5,7 +5,6 @@ import json
 import os
 import re
 import timeit
-import warnings
 from contextlib import contextmanager
 from datetime import datetime
 from pathlib import Path
@@ -18,6 +17,7 @@ from _pytest.config import Config
 from _pytest.config.argparsing import Parser
 from _pytest.terminal import TerminalReporter

+from fixtures.log_helper import log
 from fixtures.neon_fixtures import NeonPageserver
 from fixtures.types import TenantId, TimelineId

@@ -385,7 +385,7 @@ class NeonBenchmarker:
        path = f"{repo_dir}/tenants/{tenant_id}/timelines/{timeline_id}"

        totalbytes = 0
-        for root, dirs, files in os.walk(path):
+        for root, _dirs, files in os.walk(path):
            for name in files:
                totalbytes += os.path.getsize(os.path.join(root, name))

@@ -492,7 +492,7 @@ def pytest_terminal_summary(
        return

    if not result:
-        warnings.warn("no results to store (no passed test suites)")
+        log.warning("no results to store (no passed test suites)")
        return

    get_out_path(Path(out_dir), revision=revision).write_text(
--- a/test_runner/fixtures/neon_fixtures.py
+++ b/test_runner/fixtures/neon_fixtures.py
@@ -213,7 +213,7 @@ def worker_base_port(worker_seq_no: int) -> int:
 def get_dir_size(path: str) -> int:
    """Return size in bytes."""
    totalbytes = 0
-    for root, dirs, files in os.walk(path):
+    for root, _dirs, files in os.walk(path):
        for name in files:
            totalbytes += os.path.getsize(os.path.join(root, name))

@@ -459,6 +459,7 @@ class AuthKeys:
    def generate_safekeeper_token(self) -> str:
        return self.generate_token(scope="safekeeperdata")

+    # generate token giving access to only one tenant
    def generate_tenant_token(self, tenant_id: TenantId) -> str:
        return self.generate_token(scope="tenant", tenant_id=str(tenant_id))

@@ -965,6 +966,7 @@ class NeonEnv:
        for i in range(1, config.num_safekeepers + 1):
            port = SafekeeperPort(
                pg=self.port_distributor.get_port(),
+                pg_tenant_only=self.port_distributor.get_port(),
                http=self.port_distributor.get_port(),
            )
            id = config.safekeepers_id_start + i  # assign ids sequentially
@@ -973,6 +975,7 @@ class NeonEnv:
                [[safekeepers]]
                id = {id}
                pg_port = {port.pg}
+                pg_tenant_only_port = {port.pg_tenant_only}
                http_port = {port.http}
                sync = {'true' if config.safekeepers_enable_fsync else 'false'}"""
            )
@@ -1231,7 +1234,7 @@ class AbstractNeonCli(abc.ABC):
              stderr: {res.stderr}
            """
            log.info(msg)
-            raise Exception(msg) from subprocess.CalledProcessError(
+            raise RuntimeError(msg) from subprocess.CalledProcessError(
                res.returncode, res.args, res.stdout, res.stderr
            )
        return res
@@ -1255,10 +1258,8 @@ class NeonCli(AbstractNeonCli):
        """
        Creates a new tenant, returns its id and its initial timeline's id.
        """
-        if tenant_id is None:
-            tenant_id = TenantId.generate()
-        if timeline_id is None:
-            timeline_id = TimelineId.generate()
+        tenant_id = tenant_id or TenantId.generate()
+        timeline_id = timeline_id or TimelineId.generate()

        args = [
            "tenant",
@@ -1885,8 +1886,7 @@ class VanillaPostgres(PgProtocol):
        assert not self.running
        self.running = True

-        if log_path is None:
-            log_path = os.path.join(self.pgdatadir, "pg.log")
+        log_path = log_path or os.path.join(self.pgdatadir, "pg.log")

        self.pg_bin.run_capture(
            ["pg_ctl", "-w", "-D", str(self.pgdatadir), "-l", log_path, "start"]
@@ -2346,8 +2346,7 @@ class Endpoint(PgProtocol):
        if not config_lines:
            config_lines = []

-        if endpoint_id is None:
-            endpoint_id = self.env.generate_endpoint_id()
+        endpoint_id = endpoint_id or self.env.generate_endpoint_id()
        self.endpoint_id = endpoint_id
        self.branch_name = branch_name

@@ -2363,8 +2362,7 @@ class Endpoint(PgProtocol):
        path = Path("endpoints") / self.endpoint_id / "pgdata"
        self.pgdata_dir = os.path.join(self.env.repo_dir, path)

-        if config_lines is None:
-            config_lines = []
+        config_lines = config_lines or []

        # set small 'max_replication_write_lag' to enable backpressure
        # and make tests more stable.
@@ -2560,8 +2558,7 @@ class EndpointFactory:
            http_port=self.env.port_distributor.get_port(),
        )

-        if endpoint_id is None:
-            endpoint_id = self.env.generate_endpoint_id()
+        endpoint_id = endpoint_id or self.env.generate_endpoint_id()

        self.num_instances += 1
        self.endpoints.append(ep)
@@ -2614,6 +2611,7 @@ class EndpointFactory:
@dataclass
 class SafekeeperPort:
    pg: int
+    pg_tenant_only: int
    http: int


@@ -2641,7 +2639,7 @@ class Safekeeper:
                if elapsed > 3:
                    raise RuntimeError(
                        f"timed out waiting {elapsed:.0f}s for wal acceptor start: {e}"
-                    )
+                    ) from e
                time.sleep(0.5)
            else:
                break  # success
@@ -2721,7 +2719,8 @@ class SafekeeperHttpClient(requests.Session):
    def check_status(self):
        self.get(f"http://localhost:{self.port}/v1/status").raise_for_status()

-    def debug_dump(self, params: Dict[str, str] = {}) -> Dict[str, Any]:
+    def debug_dump(self, params: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
+        params = params or {}
        res = self.get(f"http://localhost:{self.port}/v1/debug_dump", params=params)
        res.raise_for_status()
        res_json = res.json()
@@ -2861,7 +2860,7 @@ class NeonBroker:
                if elapsed > 5:
                    raise RuntimeError(
                        f"timed out waiting {elapsed:.0f}s for storage_broker start: {e}"
-                    )
+                    ) from e
                time.sleep(0.5)
            else:
                break  # success
@@ -2977,7 +2976,7 @@ def should_skip_file(filename: str) -> bool:
 #
 def list_files_to_compare(pgdata_dir: Path) -> List[str]:
    pgdata_files = []
-    for root, _file, filenames in os.walk(pgdata_dir):
+    for root, _dirs, filenames in os.walk(pgdata_dir):
        for filename in filenames:
            rel_dir = os.path.relpath(root, pgdata_dir)
            # Skip some dirs and files we don't want to compare
--- a/test_runner/fixtures/pageserver/http.py
+++ b/test_runner/fixtures/pageserver/http.py
@@ -193,8 +193,7 @@ class PageserverHttpClient(requests.Session):
            body = "null"
        else:
            # null-config is prohibited by the API
-            if config is None:
-                config = {}
+            config = config or {}
            body = json.dumps({"config": config})
        res = self.post(
            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/attach",
--- a/test_runner/fixtures/pg_version.py
+++ b/test_runner/fixtures/pg_version.py
@@ -17,7 +17,6 @@ This fixture is used to determine which version of Postgres to use for tests.
 class PgVersion(str, enum.Enum):
    V14 = "14"
    V15 = "15"
-    V16 = "16"
    # Instead of making version an optional parameter in methods, we can use this fake entry
    # to explicitly rely on the default server version (could be different from pg_version fixture value)
    NOT_SET = "<-POSTRGRES VERSION IS NOT SET->"
--- a/test_runner/fixtures/utils.py
+++ b/test_runner/fixtures/utils.py
@@ -95,7 +95,7 @@ def query_scalar(cur: cursor, query: str) -> Any:
 def get_dir_size(path: str) -> int:
    """Return size in bytes."""
    totalbytes = 0
-    for root, dirs, files in os.walk(path):
+    for root, _dirs, files in os.walk(path):
        for name in files:
            try:
                totalbytes += os.path.getsize(os.path.join(root, name))
--- a/test_runner/performance/test_gc_feedback.py
+++ b/test_runner/performance/test_gc_feedback.py
@@ -47,7 +47,7 @@ def test_gc_feedback(neon_env_builder: NeonEnvBuilder, zenbenchmark: NeonBenchma
        # without modifying the earlier parts of the table.
        for step in range(n_steps):
            cur.execute(f"INSERT INTO t (step) SELECT {step} FROM generate_series(1, {step_size})")
-            for i in range(n_update_iters):
+            for _ in range(n_update_iters):
                cur.execute(f"UPDATE t set count=count+1 where step = {step}")
                cur.execute("vacuum t")

--- a/test_runner/performance/test_hot_table.py
+++ b/test_runner/performance/test_hot_table.py
@@ -33,6 +33,6 @@ def test_hot_table(env: PgCompare):

            # Read the table
            with env.record_duration("read"):
-                for i in range(num_reads):
+                for _ in range(num_reads):
                    cur.execute("select * from t;")
                    cur.fetchall()
--- a/test_runner/performance/test_layer_map.py
+++ b/test_runner/performance/test_layer_map.py
@@ -28,7 +28,7 @@ def test_layer_map(neon_env_builder: NeonEnvBuilder, zenbenchmark):
    endpoint = env.endpoints.create_start("test_layer_map", tenant_id=tenant)
    cur = endpoint.connect().cursor()
    cur.execute("create table t(x integer)")
-    for i in range(n_iters):
+    for _ in range(n_iters):
        cur.execute(f"insert into t values (generate_series(1,{n_records}))")
        time.sleep(1)

--- a/test_runner/performance/test_parallel_copy_to.py
+++ b/test_runner/performance/test_parallel_copy_to.py
@@ -6,7 +6,7 @@ from fixtures.neon_fixtures import PgProtocol


 async def repeat_bytes(buf, repetitions: int):
-    for i in range(repetitions):
+    for _ in range(repetitions):
        yield buf


--- a/test_runner/performance/test_random_writes.py
+++ b/test_runner/performance/test_random_writes.py
@@ -77,8 +77,8 @@ def test_random_writes(neon_with_baseline: PgCompare):

            # Update random keys
            with env.record_duration("run"):
-                for it in range(n_iterations):
-                    for i in range(n_writes):
+                for _ in range(n_iterations):
+                    for _ in range(n_writes):
                        key = random.randint(1, n_rows)
                        cur.execute(f"update Big set count=count+1 where pk={key}")
                    env.flush()
--- a/test_runner/performance/test_seqscans.py
+++ b/test_runner/performance/test_seqscans.py
@@ -61,5 +61,5 @@ def test_seqscans(env: PgCompare, scale: int, rows: int, iters: int, workers: in
            cur.execute(f"set max_parallel_workers_per_gather = {workers}")

            with env.record_duration("run"):
-                for i in range(iters):
+                for _ in range(iters):
                    cur.execute("select count(*) from t;")
--- a/test_runner/performance/test_startup.py
+++ b/test_runner/performance/test_startup.py
@@ -43,7 +43,17 @@ def test_startup_simple(neon_env_builder: NeonEnvBuilder, zenbenchmark: NeonBenc
            if endpoint:
                endpoint.start()
            else:
-                endpoint = env.endpoints.create_start("test_startup")
+                endpoint = env.endpoints.create_start(
+                    "test_startup",
+                    # Shared buffers need to be allocated during startup, so they
+                    # impact startup time. This is the default value we use for
+                    # 1CPU pods (maybe different for VMs).
+                    #
+                    # TODO extensions also contribute to shared memory allocation,
+                    #      and this test doesn't include all default extensions we
+                    #      load.
+                    config_lines=["shared_buffers=262144"],
+                )
            endpoint.safe_psql("select 1;")

        # Get metrics
--- a/test_runner/pg_clients/python/asyncpg/asyncpg_example.py
+++ b/test_runner/pg_clients/python/asyncpg/asyncpg_example.py
@@ -19,7 +19,7 @@ async def run(**kwargs) -> asyncpg.Record:

 if __name__ == "__main__":
    kwargs = {
-        k.lstrip("NEON_").lower(): v
+        k.removeprefix("NEON_").lower(): v
        for k in ("NEON_HOST", "NEON_DATABASE", "NEON_USER", "NEON_PASSWORD")
        if (v := os.environ.get(k, None)) is not None
    }
--- a/test_runner/pg_clients/python/pg8000/pg8000_example.py
+++ b/test_runner/pg_clients/python/pg8000/pg8000_example.py
@@ -6,7 +6,7 @@ import pg8000.dbapi

 if __name__ == "__main__":
    kwargs = {
-        k.lstrip("NEON_").lower(): v
+        k.removeprefix("NEON_").lower(): v
        for k in ("NEON_HOST", "NEON_DATABASE", "NEON_USER", "NEON_PASSWORD")
        if (v := os.environ.get(k, None)) is not None
    }
--- a/test_runner/regress/test_ancestor_branch.py
+++ b/test_runner/regress/test_ancestor_branch.py
@@ -20,7 +20,7 @@ def test_ancestor_branch(neon_env_builder: NeonEnvBuilder):
        }
    )

-    pageserver_http.configure_failpoints(("flush-frozen-before-sync", "sleep(10000)"))
+    pageserver_http.configure_failpoints(("flush-frozen-pausable", "sleep(10000)"))

    endpoint_branch0 = env.endpoints.create_start("main", tenant_id=tenant)
    branch0_cur = endpoint_branch0.connect().cursor()
--- a/test_runner/regress/test_auth.py
+++ b/test_runner/regress/test_auth.py
@@ -1,5 +1,6 @@
 from contextlib import closing

+import psycopg2
 import pytest
 from fixtures.neon_fixtures import NeonEnvBuilder, PgProtocol
 from fixtures.pageserver.http import PageserverApiException
@@ -106,7 +107,7 @@ def test_auth_failures(neon_env_builder: NeonEnvBuilder, auth_enabled: bool):
        if expect_success:
            op()
        else:
-            with pytest.raises(Exception):
+            with pytest.raises(psycopg2.Error):
                op()

    def check_pageserver(expect_success: bool, **conn_kwargs):
--- a/test_runner/regress/test_backpressure.py
+++ b/test_runner/regress/test_backpressure.py
@@ -141,13 +141,13 @@ def test_backpressure_received_lsn_lag(neon_env_builder: NeonEnvBuilder):
                    log.info("stopping check thread")
                    check_stop_event.set()
                    check_thread.join()
-                    assert (
-                        False
-                    ), f"Exception {e} while inserting rows, but WAL lag is within configured threshold. That means backpressure is not tuned properly"
+                    raise AssertionError(
+                        f"Exception {e} while inserting rows, but WAL lag is within configured threshold. That means backpressure is not tuned properly"
+                    ) from e
                else:
-                    assert (
-                        False
-                    ), f"Exception {e} while inserting rows and WAL lag overflowed configured threshold. That means backpressure doesn't work."
+                    raise AssertionError(
+                        f"Exception {e} while inserting rows and WAL lag overflowed configured threshold. That means backpressure doesn't work."
+                    ) from e

        log.info(f"inserted {rows_inserted} rows")

@@ -157,9 +157,9 @@ def test_backpressure_received_lsn_lag(neon_env_builder: NeonEnvBuilder):
        check_thread.join()
        log.info("check thread stopped")
    else:
-        assert (
-            False
-        ), "WAL lag overflowed configured threshold. That means backpressure doesn't work."
+        raise AssertionError(
+            "WAL lag overflowed configured threshold. That means backpressure doesn't work."
+        )


 # TODO test_backpressure_disk_consistent_lsn_lag. Play with pageserver's checkpoint settings
--- a/test_runner/regress/test_broken_timeline.py
+++ b/test_runner/regress/test_broken_timeline.py
@@ -26,7 +26,7 @@ def test_broken_timeline(neon_env_builder: NeonEnvBuilder):

    tenant_timelines: List[Tuple[TenantId, TimelineId, Endpoint]] = []

-    for n in range(4):
+    for _ in range(4):
        tenant_id, timeline_id = env.neon_cli.create_tenant()

        endpoint = env.endpoints.create_start("main", tenant_id=tenant_id)
--- a/test_runner/regress/test_build_info_metric.py
+++ b/test_runner/regress/test_build_info_metric.py
@@ -12,7 +12,7 @@ def test_build_info_metric(neon_env_builder: NeonEnvBuilder, link_proxy: NeonPro
    parsed_metrics["safekeeper"] = parse_metrics(env.safekeepers[0].http_client().get_metrics_str())
    parsed_metrics["proxy"] = parse_metrics(link_proxy.get_metrics())

-    for component, metrics in parsed_metrics.items():
+    for _component, metrics in parsed_metrics.items():
        sample = metrics.query_one("libmetrics_build_info")

        assert "revision" in sample.labels
--- a/test_runner/regress/test_duplicate_layers.py
+++ b/test_runner/regress/test_duplicate_layers.py
@@ -0,0 +1,36 @@
+import time
+
+import pytest
+from fixtures.neon_fixtures import NeonEnvBuilder, PgBin
+
+
+# Test duplicate layer detection
+#
+# This test sets fail point at the end of first compaction phase:
+# after flushing new L1 layers but before deletion of L0 layers
+# it should cause generation of duplicate L1 layer by compaction after restart.
+@pytest.mark.timeout(600)
+def test_duplicate_layers(neon_env_builder: NeonEnvBuilder, pg_bin: PgBin):
+    env = neon_env_builder.init_start()
+    pageserver_http = env.pageserver.http_client()
+
+    # Use aggressive compaction and checkpoint settings
+    tenant_id, _ = env.neon_cli.create_tenant(
+        conf={
+            "checkpoint_distance": f"{1024 ** 2}",
+            "compaction_target_size": f"{1024 ** 2}",
+            "compaction_period": "5 s",
+            "compaction_threshold": "3",
+        }
+    )
+
+    pageserver_http.configure_failpoints(("compact-level0-phase1-return-same", "return"))
+
+    endpoint = env.endpoints.create_start("main", tenant_id=tenant_id)
+    connstr = endpoint.connstr(options="-csynchronous_commit=off")
+    pg_bin.run_capture(["pgbench", "-i", "-s1", connstr])
+
+    time.sleep(10)  # let compaction to be performed
+    assert env.pageserver.log_contains("compact-level0-phase1-return-same")
+
+    pg_bin.run_capture(["pgbench", "-P1", "-N", "-c5", "-T500", "-Mprepared", connstr])
--- a/test_runner/regress/test_gc_aggressive.py
+++ b/test_runner/regress/test_gc_aggressive.py
@@ -54,7 +54,7 @@ async def gc(env: NeonEnv, timeline: TimelineId):
 # At the same time, run UPDATEs and GC
 async def update_and_gc(env: NeonEnv, endpoint: Endpoint, timeline: TimelineId):
    workers = []
-    for worker_id in range(num_connections):
+    for _ in range(num_connections):
        workers.append(asyncio.create_task(update_table(endpoint)))
    workers.append(asyncio.create_task(gc(env, timeline)))

--- a/test_runner/regress/test_gc_cutoff.py
+++ b/test_runner/regress/test_gc_cutoff.py
@@ -1,3 +1,5 @@
+import subprocess
+
 import pytest
 from fixtures.neon_fixtures import NeonEnvBuilder, PgBin

@@ -38,7 +40,7 @@ def test_gc_cutoff(neon_env_builder: NeonEnvBuilder, pg_bin: PgBin):
    pageserver_http.configure_failpoints(("after-timeline-gc-removed-layers", "exit"))

    for _ in range(5):
-        with pytest.raises(Exception):
+        with pytest.raises(subprocess.SubprocessError):
            pg_bin.run_capture(["pgbench", "-P1", "-N", "-c5", "-T500", "-Mprepared", connstr])
        env.pageserver.stop()
        env.pageserver.start()
--- a/test_runner/regress/test_import.py
+++ b/test_runner/regress/test_import.py
@@ -135,11 +135,11 @@ def test_import_from_vanilla(test_output_dir, pg_bin, vanilla_pg, neon_env_build
    # Importing empty file fails
    empty_file = os.path.join(test_output_dir, "empty_file")
    with open(empty_file, "w") as _:
-        with pytest.raises(Exception):
+        with pytest.raises(RuntimeError):
            import_tar(empty_file, empty_file)

    # Importing corrupt backup fails
-    with pytest.raises(Exception):
+    with pytest.raises(RuntimeError):
        import_tar(corrupt_base_tar, wal_tar)

    # A tar with trailing garbage is currently accepted. It prints a warnings
--- a/test_runner/regress/test_layer_writers_fail.py
+++ b/test_runner/regress/test_layer_writers_fail.py
@@ -1,5 +1,6 @@
 import pytest
 from fixtures.neon_fixtures import NeonEnv, NeonPageserver
+from fixtures.pageserver.http import PageserverApiException


@pytest.mark.skip("See https://github.com/neondatabase/neon/issues/2703")
@@ -77,7 +78,7 @@ def test_delta_layer_writer_fail_before_finish(neon_simple_env: NeonEnv):
    pageserver_http.configure_failpoints(("delta-layer-writer-fail-before-finish", "return"))
    # Note: we cannot test whether the exception is exactly 'delta-layer-writer-fail-before-finish'
    # since our code does it in loop, we cannot get this exact error for our request.
-    with pytest.raises(Exception):
+    with pytest.raises(PageserverApiException):
        pageserver_http.timeline_checkpoint(tenant_id, timeline_id)

    new_temp_layer_files = list(
--- a/test_runner/regress/test_multixact.py
+++ b/test_runner/regress/test_multixact.py
@@ -30,7 +30,7 @@ def test_multixact(neon_simple_env: NeonEnv, test_output_dir):
    # Lock entries using parallel connections in a round-robin fashion.
    nclients = 20
    connections = []
-    for i in range(nclients):
+    for _ in range(nclients):
        # Do not turn on autocommit. We want to hold the key-share locks.
        conn = endpoint.connect(autocommit=False)
        connections.append(conn)
--- a/test_runner/regress/test_old_request_lsn.py
+++ b/test_runner/regress/test_old_request_lsn.py
@@ -58,12 +58,12 @@ def test_old_request_lsn(neon_env_builder: NeonEnvBuilder):

    # Make a lot of updates on a single row, generating a lot of WAL. Trigger
    # garbage collections so that the page server will remove old page versions.
-    for i in range(10):
+    for _ in range(10):
        pageserver_http.timeline_checkpoint(env.initial_tenant, timeline)
        gc_result = pageserver_http.timeline_gc(env.initial_tenant, timeline, 0)
        print_gc_result(gc_result)

-        for j in range(100):
+        for _ in range(100):
            cur.execute("UPDATE foo SET val = val + 1 WHERE id = 1;")

    # All (or at least most of) the updates should've been on the same page, so
--- a/test_runner/regress/test_pageserver_restarts_under_workload.py
+++ b/test_runner/regress/test_pageserver_restarts_under_workload.py
@@ -25,7 +25,7 @@ def test_pageserver_restarts_under_worload(neon_simple_env: NeonEnv, pg_bin: PgB
    thread = threading.Thread(target=run_pgbench, args=(endpoint.connstr(),), daemon=True)
    thread.start()

-    for i in range(n_restarts):
+    for _ in range(n_restarts):
        # Stop the pageserver gracefully and restart it.
        time.sleep(1)
        env.pageserver.stop()
--- a/test_runner/regress/test_parallel_copy.py
+++ b/test_runner/regress/test_parallel_copy.py
@@ -6,7 +6,7 @@ from fixtures.neon_fixtures import Endpoint, NeonEnv


 async def repeat_bytes(buf, repetitions: int):
-    for i in range(repetitions):
+    for _ in range(repetitions):
        yield buf


--- a/test_runner/regress/test_proxy.py
+++ b/test_runner/regress/test_proxy.py
@@ -1,6 +1,6 @@
 import json
 import subprocess
-from typing import Any, List
+from typing import Any, List, Optional

 import psycopg2
 import pytest
@@ -179,7 +179,8 @@ def test_close_on_connections_exit(static_proxy: NeonProxy):
 def test_sql_over_http(static_proxy: NeonProxy):
    static_proxy.safe_psql("create role http with login password 'http' superuser")

-    def q(sql: str, params: List[Any] = []) -> Any:
+    def q(sql: str, params: Optional[List[Any]] = None) -> Any:
+        params = params or []
        connstr = f"postgresql://http:http@{static_proxy.domain}:{static_proxy.proxy_port}/postgres"
        response = requests.post(
            f"https://{static_proxy.domain}:{static_proxy.external_http_port}/sql",
@@ -229,7 +230,8 @@ def test_sql_over_http(static_proxy: NeonProxy):
 def test_sql_over_http_output_options(static_proxy: NeonProxy):
    static_proxy.safe_psql("create role http2 with login password 'http2' superuser")

-    def q(sql: str, raw_text: bool, array_mode: bool, params: List[Any] = []) -> Any:
+    def q(sql: str, raw_text: bool, array_mode: bool, params: Optional[List[Any]] = None) -> Any:
+        params = params or []
        connstr = (
            f"postgresql://http2:http2@{static_proxy.domain}:{static_proxy.proxy_port}/postgres"
        )
--- a/test_runner/regress/test_read_validation.py
+++ b/test_runner/regress/test_read_validation.py
@@ -133,7 +133,7 @@ def test_read_validation(neon_simple_env: NeonEnv):
            log.info("Validation page inspect won't allow reading pages of dropped relations")
            try:
                c.execute("select * from page_header(get_raw_page('foo', 'main', 0));")
-                assert False, "query should have failed"
+                raise AssertionError("query should have failed")
            except UndefinedTable as e:
                log.info("Caught an expected failure: {}".format(e))

@@ -157,7 +157,7 @@ def test_read_validation_neg(neon_simple_env: NeonEnv):
                c.execute(
                    "select lsn, lower, upper from page_header(get_raw_page_at_lsn('Unknown', 'main', 0, '0/0'))"
                )
-                assert False, "query should have failed"
+                raise AssertionError("query should have failed")
            except UndefinedTable as e:
                log.info("Caught an expected failure: {}".format(e))

@@ -169,7 +169,7 @@ def test_read_validation_neg(neon_simple_env: NeonEnv):
                c.execute(
                    "select lsn, lower, upper from page_header(get_raw_page_at_lsn('foo', 'main', 0, '0/0'))"
                )
-                assert False, "query should have failed"
+                raise AssertionError("query should have failed")
            except IoError as e:
                log.info("Caught an expected failure: {}".format(e))

--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
Alex Chi Z	84ab5e4101	fix clippy warnings Signed-off-by: Alex Chi Z <chi@neon.tech>	2023-08-04 15:19:41 -04:00
Alex Chi Z	607f341e24	more assertions Signed-off-by: Alex Chi Z <chi@neon.tech>	2023-08-04 14:55:25 -04:00
Alex Chi Z	2275b74a75	use read lock for compaction Signed-off-by: Alex Chi Z <chi@neon.tech>	2023-08-04 14:50:55 -04:00
Alex Chi Z	09c33fd0a4	fix lock order, take snapshot after taking lock Signed-off-by: Alex Chi Z <chi@neon.tech>	2023-07-21 15:23:28 -04:00
Alex Chi Z	f726063d2f	fix test compile Signed-off-by: Alex Chi Z <chi@neon.tech>	2023-07-20 16:40:52 -04:00
Alex Chi Z	91c7bbb230	switch to immutable layer map implementation Signed-off-by: Alex Chi Z <chi@neon.tech>	2023-07-20 16:35:26 -04:00
Alex Chi Z	3fc3666df7	make flush frozen layer an atomic operation (#4720 ) ## Problem close https://github.com/neondatabase/neon/issues/4712 ## Summary of changes Previously, when flushing frozen layers, it was split into two operations: add delta layer to disk + remove frozen layer from memory. This would cause a short period of time where we will have the same data both in frozen and delta layer. In this PR, we merge them into one atomic operation in layer map manager, therefore simplifying the code. Note that if we decide to create image layers for L0 flush, it will still be split into two operations on layer map. --------- Signed-off-by: Alex Chi Z <chi@neon.tech> Co-authored-by: Joonas Koivunen <joonas@neon.tech>	2023-07-20 13:39:19 -04:00
Joonas Koivunen	89746a48c6	chore: fix copypaste caused flakyness (#4763 ) I introduced a copypaste error leading to flaky [test failure][report] in #4737. Solution is to use correct/unique test name. I also looked into providing a proper fn name via macro but ... Yeah, it's probably not a great idea. [report]: https://github.com/neondatabase/neon/actions/runs/5612473297/job/15206293430#step:15:197	2023-07-20 19:55:40 +03:00
Joonas Koivunen	8d27a9c54e	Less verbose eviction failures (#4737 ) As seen in staging logs with some massive compactions (create_image_layer), in addition to racing with compaction or gc or even between two invocations to `evict_layer_batch`. Cc: #4745 Fixes: #3851 (organic tech debt reduction) Solution is not to log the Not Found in such cases; it is perfectly natural to happen. Route to this is quite long, but implemented two cases of "race between two eviction processes" which are like our disk usage based eviction and eviction_task, both have the separate "lets figure out what to evict" and "lets evict" phases.	2023-07-20 17:45:10 +03:00
arpad-m	d98cb39978	pageserver: use tokio::time::timeout where possible (#4756 ) Removes a bunch of cases which used `tokio::select` to emulate the `tokio::time::timeout` function. I've done an additional review on the cancellation safety of these futures, all of them seem to be cancellation safe (not that `select!` allows non-cancellation-safe futures, but as we touch them, such a review makes sense). Furthermore, I correct a few mentions of a non-existent `tokio::timeout!` macro in the docs to the `tokio::time::timeout` function.	2023-07-20 16:19:38 +02:00
Alexander Bayandin	27c73c8740	Bump pg_embedding extension (#4758 ) ``` % git log --pretty=oneline 2465f831ea1f8d49c1d74f8959adb7fc277d70cd..eeb3ba7c3a60c95b2604dd543c64b2f1bb4a3703 eeb3ba7c3a60c95b2604dd543c64b2f1bb4a3703 (HEAD -> main, origin/main) Fixc in-mmeory index rebuild after TRUNCATE 1d7cfcfe3d58e2cf4566900437c609725448d14b Correctly handle truncate forin-0memory HNSW index 8fd2a4a191f67858498d876ec378b58e76b5874a :Fix empty index search issue 30e9ef4064cff40c60ff2f78afeac6c296722757 Fix extensiomn name in makefile 23bb5d504aa21b1663719739f6eedfdcb139d948 Fix getting memory size at Mac OS/X 39193a38d6ad8badd2a8d1dce2dd999e1b86885d Update a comment for the extension bf3b0d62a7df56a5e4db9d9e62dc535794c425bc Merge branch 'main' of https://github.com/neondatabase/pg_embedding c2142d514280e14322d1026f0c811876ccf7a91f Update README.md 53b641880f786d2b69a75941c49e569018e8e97e Create LICENSE 093aaa36d5af183831bf370c97b563c12d15f23a Update README.md 91f0bb84d14cb26fd8b452bf9e1ecea026ac5cbc Update README.md 7f7efa38015f24ee9a09beca3009b8d0497a40b4 Update README.md 71defdd4143ecf35489d93289f6cdfa2545fbd36 Merge pull request #4 from neondatabase/danieltprice-patch-1 e06c228b99c6b7c47ebce3bb7c97dbd494088b0a Update README.md d7e52b576b47d9023743b124bdd0360a9fc98f59 Update README.md 70ab399c861330b50a9aff9ab9edc7044942a65b Merge pull request #5 from neondatabase/oom_error_reporting 0aee1d937997198fa2d2b2ed7a0886d1075fa790 Fix OOM error reporting and support vectprization for ARM 18d80079ce60b2aa81d58cefdf42fc09d2621fc1 Update README.md ```	2023-07-20 12:32:57 +01:00
Joonas Koivunen	9e871318a0	Wait detaches or ignores on pageserver shutdown (#4678 ) Adds in a barrier for the duration of the `Tenant::shutdown`. `pageserver_shutdown` will join this await, `detach`es and `ignore`s will not. Fixes #4429. --------- Co-authored-by: Christian Schwarz <christian@neon.tech>	2023-07-20 13:14:13 +03:00
bojanserafimov	e1061879aa	Improve startup python test (#4757 )	2023-07-19 23:46:16 -04:00
Daniel	f09e82270e	Update comment for hnsw extension (#4755 ) Updated the description that appears for hnsw when you query extensions: ``` neondb=> SELECT * FROM pg_available_extensions WHERE name = 'hnsw'; name \| default_version \| installed_version \| comment ----------------------+-----------------+-------------------+-------------------------------------------------- hnsw \| 0.1.0 \| \| Deprecated Please use pg_embedding instead (1 row) ``` --------- Co-authored-by: Alexander Bayandin <alexander@neon.tech>	2023-07-19 19:08:25 +01:00
Alexander Bayandin	d4a5fd5258	Disable extension uploading to S3 (#4751 ) ## Problem We're going to reset S3 buckets for extensions (https://github.com/neondatabase/aws/pull/413), and as soon as we're going to change the format we store extensions on S3. Let's stop uploading extensions in the old format. ## Summary of changes - Disable `aws s3 cp` step for extensions	2023-07-19 15:44:14 +01:00
Arseny Sher	921bb86909	Use safekeeper tenant only port in all tests and actually test it. Compute now uses special safekeeper WAL service port allowing auth tokens with only tenant scope. Adds understanding of this port to neon_local and fixtures, as well as test of both ports behaviour with different tokens. ref https://github.com/neondatabase/neon/issues/4730	2023-07-19 06:03:51 +04:00
Arseny Sher	1e7db5458f	Add one more WAL service port allowing only tenant scoped auth tokens. It will make it easier to limit access at network level, with e.g. k8s network policies. ref https://github.com/neondatabase/neon/issues/4730	2023-07-19 06:03:51 +04:00
Alexander Bayandin	b4d36f572d	Use sharded-slab from crates (#4729 ) ## Problem We use a patched version of `sharded-slab` with increased MAX_THREADS [1]. It is not required anymore because safekeepers are async now. A valid comment from the original PR tho [1]: > Note that patch can affect other rust services, not only the safekeeper binary. - [1] https://github.com/neondatabase/neon/pull/4122 ## Summary of changes - Remove patch for `sharded-slab`	2023-07-18 13:50:44 +01:00
Joonas Koivunen	762a8a7bb5	python: more linting (#4734 ) Ruff has "B" class of lints, including B018 which will nag on useless expressions, related to #4719. Enable such lints and fix the existing issues. Most notably: - https://beta.ruff.rs/docs/rules/mutable-argument-default/ - https://beta.ruff.rs/docs/rules/assert-false/ --------- Co-authored-by: Alexander Bayandin <alexander@neon.tech>	2023-07-18 12:56:40 +03:00
Conrad Ludgate	2e8a3afab1	proxy: merge handle_client (#4740 ) ## Problem Second half of #4699. we were maintaining 2 implementations of handle_client. ## Summary of changes Merge the handle_client code, but abstract some of the details. ## Checklist before requesting a review - [X] I have performed a self-review of my code. - [ ] If it is a core feature, I have added thorough tests. - [ ] Do we need to implement analytics? if so did you add the relevant metrics to the dashboard? - [ ] If this PR requires public announcement, mark it with /release-notes label and add several sentences in this section. ## Checklist before merging - [ ] Do not forget to reformat commit message to not include the above checklist	2023-07-17 22:20:23 +01:00
Alexander Bayandin	4580f5085a	test_runner: run benchmarks in parallel (#4683 ) ## Problem Benchmarks run takes about an hour on main branch (in a single job), which delays pipeline results. And it takes another hour if we want to restart the job due to some failures. ## Summary of changes - Use `pytest-split` plugin to run benchmarks on separate CI runners in 4 parallel jobs - Add `scripts/benchmark_durations.py` for getting benchmark durations from the database to help `pytest-split` schedule tests more evenly. It uses p99 for the last 10 days' results (durations). The current distribution could be better; each worker's durations vary from 9m to 35m, but this could be improved in consequent PRs.	2023-07-17 20:09:45 +01:00
Conrad Ludgate	e074ccf170	reduce proxy timeouts (#4708 ) ## Problem 10 retries * 10 second timeouts makes for a very long retry window. ## Summary of changes Adds a 2s timeout to sql_over_http connections, and also reduces the 10s timeout in TCP.	2023-07-17 20:05:26 +01:00
George MacKerron	196943c78f	CORS preflight OPTIONS support for /sql (http fetch) endpoint (#4706 ) ## Problem HTTP fetch can't be used from browsers because proxy doesn't support [CORS 'preflight' `OPTIONS` requests](https://developer.mozilla.org/en-US/docs/Web/HTTP/CORS#preflighted_requests). ## Summary of changes Added a simple `OPTIONS` endpoint for `/sql`.	2023-07-17 20:01:25 +01:00
bojanserafimov	149dd36b6b	Update pg: add startup logs (#4736 )	2023-07-17 14:47:08 -04:00
Kirill Bulatov	be271e3edf	Use upstream version of tokio-tar (#4722 ) tokio-tar 0.3.1 got released, including all changes from the fork currently used, switch over to that one.	2023-07-17 17:18:33 +01:00
Conrad Ludgate	7c85c7ea91	proxy: merge connect compute (#4713 ) ## Problem Half of #4699. TCP/WS have one implementation of `connect_to_compute`, HTTP has another implementation of `connect_to_compute`. Having both is annoying to deal with. ## Summary of changes Creates a set of traits `ConnectMechanism` and `ShouldError` that allows the `connect_to_compute` to be generic over raw TCP stream or tokio_postgres based connections. I'm not super happy with this. I think it would be nice to remove tokio_postgres entirely but that will need a lot more thought to be put into it. I have also slightly refactored the caching to use fewer references. Instead using ownership to ensure the state of retrying is encoded in the type system.	2023-07-17 15:53:01 +01:00
Alex Chi Z	1066bca5e3	compaction: allow duplicated layers and skip in replacement (#4696 ) ## Problem Compactions might generate files of exactly the same name as before compaction due to our naming of layer files. This could have already caused some mess in the system, and is known to cause some issues like https://github.com/neondatabase/neon/issues/4088. Therefore, we now consider duplicated layers in the post-compaction process to avoid violating the layer map duplicate checks. related previous works: close https://github.com/neondatabase/neon/pull/4094 error reported in: https://github.com/neondatabase/neon/issues/4690, https://github.com/neondatabase/neon/issues/4088 ## Summary of changes If a file already exists in the layer map before the compaction, do not modify the layer map and do not delete the file. The file on disk at that time should be the new one overwritten by the compaction process. This PR also adds a test case with a fail point that produces exactly the same set of files. This bypassing behavior is safe because the produced layer files have the same content / are the same representation of the original file. An alternative might be directly removing the duplicate check in the layer map, but I feel it would be good if we can prevent that in the first place. --------- Signed-off-by: Alex Chi Z <chi@neon.tech> Co-authored-by: Konstantin Knizhnik <knizhnik@garret.ru> Co-authored-by: Heikki Linnakangas <heikki@neon.tech> Co-authored-by: Joonas Koivunen <joonas@neon.tech>	2023-07-17 17:26:29 +03:00