break e2e test

2026-05-21 07:00:38 +00:00 · 2024-10-05 11:30:19 +01:00
277 changed files with 1518 additions and 2581 deletions
--- a/.dockerignore
+++ b/.dockerignore
@@ -5,7 +5,9 @@
 !Cargo.toml
 !Makefile
 !rust-toolchain.toml
+!scripts/combine_control_files.py
 !scripts/ninstall.sh
+!vm-cgconfig.conf
 !docker-compose/run-tests.sh

 # Directories
@@ -15,12 +17,15 @@
 !compute_tools/
 !control_plane/
 !libs/
+!neon_local/
 !pageserver/
+!patches/
 !pgxn/
 !proxy/
 !storage_scrubber/
 !safekeeper/
 !storage_broker/
 !storage_controller/
+!trace/
 !vendor/postgres-*/
 !workspace_hack/
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -666,6 +666,34 @@ dependencies = [
 "tracing",
 ]

+[[package]]
+name = "axum"
+version = "0.6.20"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3b829e4e32b91e643de6eafe82b1d90675f5874230191a4ffbc1b336dec4d6bf"
+dependencies = [
+ "async-trait",
+ "axum-core 0.3.4",
+ "bitflags 1.3.2",
+ "bytes",
+ "futures-util",
+ "http 0.2.9",
+ "http-body 0.4.5",
+ "hyper 0.14.30",
+ "itoa",
+ "matchit 0.7.0",
+ "memchr",
+ "mime",
+ "percent-encoding",
+ "pin-project-lite",
+ "rustversion",
+ "serde",
+ "sync_wrapper 0.1.2",
+ "tower",
+ "tower-layer",
+ "tower-service",
+]
+
 [[package]]
 name = "axum"
 version = "0.7.5"
@@ -673,7 +701,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3a6c9af12842a67734c9a2e355436e5d03b22383ed60cf13cd0c18fbfe3dcbcf"
 dependencies = [
 "async-trait",
- "axum-core",
+ "axum-core 0.4.5",
 "base64 0.21.1",
 "bytes",
 "futures-util",
@@ -703,6 +731,23 @@ dependencies = [
 "tracing",
 ]

+[[package]]
+name = "axum-core"
+version = "0.3.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "759fa577a247914fd3f7f76d62972792636412fbfd634cd452f6a385a74d2d2c"
+dependencies = [
+ "async-trait",
+ "bytes",
+ "futures-util",
+ "http 0.2.9",
+ "http-body 0.4.5",
+ "mime",
+ "rustversion",
+ "tower-layer",
+ "tower-service",
+]
+
 [[package]]
 name = "axum-core"
 version = "0.4.5"
@@ -926,7 +971,7 @@ dependencies = [
 "clang-sys",
 "itertools 0.12.1",
 "log",
- "prettyplease",
+ "prettyplease 0.2.17",
 "proc-macro2",
 "quote",
 "regex",
@@ -2409,6 +2454,15 @@ dependencies = [
 "digest",
 ]

+[[package]]
+name = "home"
+version = "0.5.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e3d1354bf6b7235cb4a0576c2619fd4ed18183f689b12b006a0ee7329eeff9a5"
+dependencies = [
+ "windows-sys 0.52.0",
+]
+
 [[package]]
 name = "hostname"
 version = "0.4.0"
@@ -2603,15 +2657,14 @@ dependencies = [

 [[package]]
 name = "hyper-timeout"
-version = "0.5.1"
+version = "0.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3203a961e5c83b6f5498933e78b6b263e208c197b63e9c6c53cc82ffd3f63793"
+checksum = "bbb958482e8c7be4bc3cf272a766a2b0bf1a6755e7a6ae777f017a31d11b13b1"
 dependencies = [
- "hyper 1.4.1",
- "hyper-util",
+ "hyper 0.14.30",
 "pin-project-lite",
 "tokio",
- "tower-service",
+ "tokio-io-timeout",
 ]

 [[package]]
@@ -3417,7 +3470,7 @@ dependencies = [
 "opentelemetry-http",
 "opentelemetry-proto",
 "opentelemetry_sdk",
- "prost",
+ "prost 0.13.3",
 "reqwest 0.12.4",
 "thiserror",
 ]
@@ -3430,8 +3483,8 @@ checksum = "30ee9f20bff9c984511a02f082dc8ede839e4a9bf15cc2487c8d6fea5ad850d9"
 dependencies = [
 "opentelemetry",
 "opentelemetry_sdk",
- "prost",
- "tonic",
+ "prost 0.13.3",
+ "tonic 0.12.3",
 ]

 [[package]]
@@ -4125,6 +4178,16 @@ dependencies = [
 "tokio",
 ]

+[[package]]
+name = "prettyplease"
+version = "0.1.25"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6c8646e95016a7a6c4adea95bafa8a16baab64b583356217f2c85db4a39d9a86"
+dependencies = [
+ "proc-macro2",
+ "syn 1.0.109",
+]
+
 [[package]]
 name = "prettyplease"
 version = "0.2.17"
@@ -4195,6 +4258,16 @@ dependencies = [
 "thiserror",
 ]

+[[package]]
+name = "prost"
+version = "0.11.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0b82eaa1d779e9a4bc1c3217db8ffbeabaae1dca241bf70183242128d48681cd"
+dependencies = [
+ "bytes",
+ "prost-derive 0.11.9",
+]
+
 [[package]]
 name = "prost"
 version = "0.13.3"
@@ -4202,28 +4275,42 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7b0487d90e047de87f984913713b85c601c05609aad5b0df4b4573fbf69aa13f"
 dependencies = [
 "bytes",
- "prost-derive",
+ "prost-derive 0.13.3",
 ]

 [[package]]
 name = "prost-build"
-version = "0.13.3"
+version = "0.11.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0c1318b19085f08681016926435853bbf7858f9c082d0999b80550ff5d9abe15"
+checksum = "119533552c9a7ffacc21e099c24a0ac8bb19c2a2a3f363de84cd9b844feab270"
 dependencies = [
 "bytes",
- "heck 0.5.0",
- "itertools 0.12.1",
+ "heck 0.4.1",
+ "itertools 0.10.5",
+ "lazy_static",
 "log",
 "multimap",
- "once_cell",
 "petgraph",
- "prettyplease",
- "prost",
+ "prettyplease 0.1.25",
+ "prost 0.11.9",
 "prost-types",
 "regex",
- "syn 2.0.52",
+ "syn 1.0.109",
 "tempfile",
+ "which",
+]
+
+[[package]]
+name = "prost-derive"
+version = "0.11.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e5d2d8d10f3c6ded6da8b05b5fb3b8a5082514344d56c9f871412d29b4e075b4"
+dependencies = [
+ "anyhow",
+ "itertools 0.10.5",
+ "proc-macro2",
+ "quote",
+ "syn 1.0.109",
 ]

 [[package]]
@@ -4241,11 +4328,11 @@ dependencies = [

 [[package]]
 name = "prost-types"
-version = "0.13.3"
+version = "0.11.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4759aa0d3a6232fb8dbdb97b61de2c20047c68aca932c7ed76da9d788508d670"
+checksum = "213622a1460818959ac1181aaeb2dc9c7f63df720db7d788b3e24eacd1983e13"
 dependencies = [
- "prost",
+ "prost 0.11.9",
 ]

 [[package]]
@@ -5007,21 +5094,6 @@ dependencies = [
 "zeroize",
 ]

-[[package]]
-name = "rustls"
-version = "0.23.7"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ebbbdb961df0ad3f2652da8f3fdc4b36122f568f968f45ad3316f26c025c677b"
-dependencies = [
- "log",
- "once_cell",
- "ring",
- "rustls-pki-types",
- "rustls-webpki 0.102.2",
- "subtle",
- "zeroize",
-]
-
 [[package]]
 name = "rustls-native-certs"
 version = "0.6.2"
@@ -5047,19 +5119,6 @@ dependencies = [
 "security-framework",
 ]

-[[package]]
-name = "rustls-native-certs"
-version = "0.8.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fcaf18a4f2be7326cd874a5fa579fae794320a0f388d365dca7e480e55f83f8a"
-dependencies = [
- "openssl-probe",
- "rustls-pemfile 2.1.1",
- "rustls-pki-types",
- "schannel",
- "security-framework",
-]
-
 [[package]]
 name = "rustls-pemfile"
 version = "1.0.2"
@@ -5135,7 +5194,6 @@ dependencies = [
 "fail",
 "futures",
 "hex",
- "http 1.1.0",
 "humantime",
 "hyper 0.14.30",
 "metrics",
@@ -5692,22 +5750,19 @@ version = "0.1.0"
 dependencies = [
 "anyhow",
 "async-stream",
- "bytes",
 "clap",
 "const_format",
 "futures",
 "futures-core",
 "futures-util",
- "http-body-util",
 "humantime",
- "hyper 1.4.1",
- "hyper-util",
+ "hyper 0.14.30",
 "metrics",
 "once_cell",
 "parking_lot 0.12.1",
- "prost",
+ "prost 0.11.9",
 "tokio",
- "tonic",
+ "tonic 0.9.2",
 "tonic-build",
 "tracing",
 "utils",
@@ -6251,17 +6306,6 @@ dependencies = [
 "tokio",
 ]

-[[package]]
-name = "tokio-rustls"
-version = "0.26.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0c7bc40d0e5a97695bb96e27995cd3a08538541b0a846f65bba7a359f36700d4"
-dependencies = [
- "rustls 0.23.7",
- "rustls-pki-types",
- "tokio",
-]
-
 [[package]]
 name = "tokio-stream"
 version = "0.1.16"
@@ -6353,30 +6397,29 @@ dependencies = [

 [[package]]
 name = "tonic"
-version = "0.12.3"
+version = "0.9.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "877c5b330756d856ffcc4553ab34a5684481ade925ecc54bcd1bf02b1d0d4d52"
+checksum = "3082666a3a6433f7f511c7192923fa1fe07c69332d3c6a2e6bb040b569199d5a"
 dependencies = [
 "async-stream",
 "async-trait",
- "axum",
- "base64 0.22.1",
+ "axum 0.6.20",
+ "base64 0.21.1",
 "bytes",
- "h2 0.4.4",
- "http 1.1.0",
- "http-body 1.0.0",
- "http-body-util",
- "hyper 1.4.1",
+ "futures-core",
+ "futures-util",
+ "h2 0.3.26",
+ "http 0.2.9",
+ "http-body 0.4.5",
+ "hyper 0.14.30",
 "hyper-timeout",
- "hyper-util",
 "percent-encoding",
 "pin-project",
- "prost",
- "rustls-native-certs 0.8.0",
- "rustls-pemfile 2.1.1",
- "socket2",
+ "prost 0.11.9",
+ "rustls-native-certs 0.6.2",
+ "rustls-pemfile 1.0.2",
 "tokio",
- "tokio-rustls 0.26.0",
+ "tokio-rustls 0.24.0",
 "tokio-stream",
 "tower",
 "tower-layer",
@@ -6385,17 +6428,37 @@ dependencies = [
 ]

 [[package]]
-name = "tonic-build"
+name = "tonic"
 version = "0.12.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9557ce109ea773b399c9b9e5dca39294110b74f1f342cb347a80d1fce8c26a11"
+checksum = "877c5b330756d856ffcc4553ab34a5684481ade925ecc54bcd1bf02b1d0d4d52"
 dependencies = [
- "prettyplease",
+ "async-trait",
+ "base64 0.22.1",
+ "bytes",
+ "http 1.1.0",
+ "http-body 1.0.0",
+ "http-body-util",
+ "percent-encoding",
+ "pin-project",
+ "prost 0.13.3",
+ "tokio-stream",
+ "tower-layer",
+ "tower-service",
+ "tracing",
+]
+
+[[package]]
+name = "tonic-build"
+version = "0.9.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a6fdaae4c2c638bb70fe42803a26fbd6fc6ac8c72f5c59f67ecc2a2dcabf4b07"
+dependencies = [
+ "prettyplease 0.1.25",
 "proc-macro2",
 "prost-build",
- "prost-types",
 "quote",
- "syn 2.0.52",
+ "syn 1.0.109",
 ]

 [[package]]
@@ -6801,7 +6864,7 @@ name = "vm_monitor"
 version = "0.1.0"
 dependencies = [
 "anyhow",
- "axum",
+ "axum 0.7.5",
 "cgroups-rs",
 "clap",
 "futures",
@@ -7032,6 +7095,18 @@ dependencies = [
 "rustls-pki-types",
 ]

+[[package]]
+name = "which"
+version = "4.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "87ba24419a2078cd2b0f2ede2691b6c66d8e47836da3b6db8265ebad47afbfc7"
+dependencies = [
+ "either",
+ "home",
+ "once_cell",
+ "rustix",
+]
+
 [[package]]
 name = "whoami"
 version = "1.5.1"
@@ -7260,10 +7335,9 @@ version = "0.1.0"
 dependencies = [
 "ahash",
 "anyhow",
- "axum",
- "axum-core",
 "base64 0.21.1",
 "base64ct",
+ "bitflags 2.4.1",
 "bytes",
 "camino",
 "cc",
@@ -7291,6 +7365,7 @@ dependencies = [
 "hyper 1.4.1",
 "hyper-util",
 "indexmap 1.9.3",
+ "itertools 0.10.5",
 "itertools 0.12.1",
 "lazy_static",
 "libc",
@@ -7302,15 +7377,15 @@ dependencies = [
 "num-traits",
 "once_cell",
 "parquet",
- "prettyplease",
 "proc-macro2",
- "prost",
+ "prost 0.11.9",
 "quote",
 "rand 0.8.5",
 "regex",
 "regex-automata 0.4.3",
 "regex-syntax 0.8.2",
 "reqwest 0.12.4",
+ "rustls 0.21.11",
 "scopeguard",
 "serde",
 "serde_json",
@@ -7326,10 +7401,9 @@ dependencies = [
 "time",
 "time-macros",
 "tokio",
- "tokio-stream",
+ "tokio-rustls 0.24.0",
 "tokio-util",
 "toml_edit",
- "tonic",
 "tower",
 "tracing",
 "tracing-core",
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -130,7 +130,7 @@ pbkdf2 = { version = "0.12.1", features = ["simple", "std"] }
 pin-project-lite = "0.2"
 procfs = "0.16"
 prometheus = {version = "0.13", default-features=false, features = ["process"]} # removes protobuf dependency
-prost = "0.13"
+prost = "0.11"
 rand = "0.8"
 redis = { version = "0.25.2", features = ["tokio-rustls-comp", "keep-alive"] }
 regex = "1.10.2"
@@ -178,7 +178,7 @@ tokio-tar = "0.3"
 tokio-util = { version = "0.7.10", features = ["io", "rt"] }
 toml = "0.8"
 toml_edit = "0.22"
-tonic = {version = "0.12.3", features = ["tls", "tls-roots"]}
+tonic = {version = "0.9", features = ["tls", "tls-roots"]}
 tower-service = "0.3.2"
 tracing = "0.1"
 tracing-error = "0.2"
@@ -246,7 +246,7 @@ criterion = "0.5.1"
 rcgen = "0.12"
 rstest = "0.18"
 camino-tempfile = "1.0.2"
-tonic-build = "0.12"
+tonic-build = "0.9"

 [patch.crates-io]

--- a/14
+++ b/14
@@ -168,27 +168,27 @@ postgres-check-%: postgres-%
 neon-pg-ext-%: postgres-%
 	+@echo "Compiling neon $*"
 	mkdir -p $(POSTGRES_INSTALL_DIR)/build/neon-$*
-	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config COPT='$(COPT)' \
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config CFLAGS='$(PG_CFLAGS) $(COPT)' \
 		-C $(POSTGRES_INSTALL_DIR)/build/neon-$* \
 		-f $(ROOT_PROJECT_DIR)/pgxn/neon/Makefile install
 	+@echo "Compiling neon_walredo $*"
 	mkdir -p $(POSTGRES_INSTALL_DIR)/build/neon-walredo-$*
-	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config COPT='$(COPT)' \
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config CFLAGS='$(PG_CFLAGS) $(COPT)' \
 		-C $(POSTGRES_INSTALL_DIR)/build/neon-walredo-$* \
 		-f $(ROOT_PROJECT_DIR)/pgxn/neon_walredo/Makefile install
 	+@echo "Compiling neon_rmgr $*"
 	mkdir -p $(POSTGRES_INSTALL_DIR)/build/neon-rmgr-$*
-	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config COPT='$(COPT)' \
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config CFLAGS='$(PG_CFLAGS) $(COPT)' \
 		-C $(POSTGRES_INSTALL_DIR)/build/neon-rmgr-$* \
 		-f $(ROOT_PROJECT_DIR)/pgxn/neon_rmgr/Makefile install
 	+@echo "Compiling neon_test_utils $*"
 	mkdir -p $(POSTGRES_INSTALL_DIR)/build/neon-test-utils-$*
-	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config COPT='$(COPT)' \
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config CFLAGS='$(PG_CFLAGS) $(COPT)' \
 		-C $(POSTGRES_INSTALL_DIR)/build/neon-test-utils-$* \
 		-f $(ROOT_PROJECT_DIR)/pgxn/neon_test_utils/Makefile install
 	+@echo "Compiling neon_utils $*"
 	mkdir -p $(POSTGRES_INSTALL_DIR)/build/neon-utils-$*
-	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config COPT='$(COPT)' \
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config CFLAGS='$(PG_CFLAGS) $(COPT)' \
 		-C $(POSTGRES_INSTALL_DIR)/build/neon-utils-$* \
 		-f $(ROOT_PROJECT_DIR)/pgxn/neon_utils/Makefile install

@@ -220,7 +220,7 @@ neon-pg-clean-ext-%:
 walproposer-lib: neon-pg-ext-v17
 	+@echo "Compiling walproposer-lib"
 	mkdir -p $(POSTGRES_INSTALL_DIR)/build/walproposer-lib
-	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/v17/bin/pg_config COPT='$(COPT)' \
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/v17/bin/pg_config CFLAGS='$(PG_CFLAGS) $(COPT)' \
 		-C $(POSTGRES_INSTALL_DIR)/build/walproposer-lib \
 		-f $(ROOT_PROJECT_DIR)/pgxn/neon/Makefile walproposer-lib
 	cp $(POSTGRES_INSTALL_DIR)/v17/lib/libpgport.a $(POSTGRES_INSTALL_DIR)/build/walproposer-lib
@@ -333,7 +333,7 @@ postgres-%-pgindent: postgres-%-pg-bsd-indent postgres-%-typedefs.list
 # Indent pxgn/neon.
 .PHONY: neon-pgindent
 neon-pgindent: postgres-v17-pg-bsd-indent neon-pg-ext-v17
-	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/v17/bin/pg_config COPT='$(COPT)' \
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/v17/bin/pg_config CFLAGS='$(PG_CFLAGS) $(COPT)' \
 		FIND_TYPEDEF=$(ROOT_PROJECT_DIR)/vendor/postgres-v17/src/tools/find_typedef \
 		INDENT=$(POSTGRES_INSTALL_DIR)/build/v17/src/tools/pg_bsd_indent/pg_bsd_indent \
 		PGINDENT_SCRIPT=$(ROOT_PROJECT_DIR)/vendor/postgres-v17/src/tools/pgindent/pgindent \
--- a/compute_tools/src/bin/compute_ctl.rs
+++ b/compute_tools/src/bin/compute_ctl.rs
@@ -402,7 +402,8 @@ fn start_postgres(
 ) -> Result<(Option<PostgresHandle>, StartPostgresResult)> {
    // We got all we need, update the state.
    let mut state = compute.state.lock().unwrap();
-    state.set_status(ComputeStatus::Init, &compute.state_changed);
+    state.status = ComputeStatus::Init;
+    compute.state_changed.notify_all();

    info!(
        "running compute with features: {:?}",
--- a/compute_tools/src/compute.rs
+++ b/compute_tools/src/compute.rs
@@ -109,18 +109,6 @@ impl ComputeState {
            metrics: ComputeMetrics::default(),
        }
    }
-
-    pub fn set_status(&mut self, status: ComputeStatus, state_changed: &Condvar) {
-        let prev = self.status;
-        info!("Changing compute status from {} to {}", prev, status);
-        self.status = status;
-        state_changed.notify_all();
-    }
-
-    pub fn set_failed_status(&mut self, err: anyhow::Error, state_changed: &Condvar) {
-        self.error = Some(format!("{err:?}"));
-        self.set_status(ComputeStatus::Failed, state_changed);
-    }
 }

 impl Default for ComputeState {
@@ -315,12 +303,15 @@ impl ComputeNode {

    pub fn set_status(&self, status: ComputeStatus) {
        let mut state = self.state.lock().unwrap();
-        state.set_status(status, &self.state_changed);
+        state.status = status;
+        self.state_changed.notify_all();
    }

    pub fn set_failed_status(&self, err: anyhow::Error) {
        let mut state = self.state.lock().unwrap();
-        state.set_failed_status(err, &self.state_changed);
+        state.error = Some(format!("{err:?}"));
+        state.status = ComputeStatus::Failed;
+        self.state_changed.notify_all();
    }

    pub fn get_status(&self) -> ComputeStatus {
--- a/compute_tools/src/configurator.rs
+++ b/compute_tools/src/configurator.rs
@@ -24,7 +24,8 @@ fn configurator_main_loop(compute: &Arc<ComputeNode>) {
        // Re-check the status after waking up
        if state.status == ComputeStatus::ConfigurationPending {
            info!("got configuration request");
-            state.set_status(ComputeStatus::Configuration, &compute.state_changed);
+            state.status = ComputeStatus::Configuration;
+            compute.state_changed.notify_all();
            drop(state);

            let mut new_status = ComputeStatus::Failed;
--- a/compute_tools/src/http/api.rs
+++ b/compute_tools/src/http/api.rs
@@ -288,7 +288,8 @@ async fn handle_configure_request(
                return Err((msg, StatusCode::PRECONDITION_FAILED));
            }
            state.pspec = Some(parsed_spec);
-            state.set_status(ComputeStatus::ConfigurationPending, &compute.state_changed);
+            state.status = ComputeStatus::ConfigurationPending;
+            compute.state_changed.notify_all();
            drop(state);
            info!("set new spec and notified waiters");
        }
@@ -361,15 +362,15 @@ async fn handle_terminate_request(compute: &Arc<ComputeNode>) -> Result<(), (Str
        }
        if state.status != ComputeStatus::Empty && state.status != ComputeStatus::Running {
            let msg = format!(
-                "invalid compute status for termination request: {}",
-                state.status
+                "invalid compute status for termination request: {:?}",
+                state.status.clone()
            );
            return Err((msg, StatusCode::PRECONDITION_FAILED));
        }
-        state.set_status(ComputeStatus::TerminationPending, &compute.state_changed);
+        state.status = ComputeStatus::TerminationPending;
+        compute.state_changed.notify_all();
        drop(state);
    }
-
    forward_termination_signal();
    info!("sent signal and notified waiters");

@@ -383,8 +384,7 @@ async fn handle_terminate_request(compute: &Arc<ComputeNode>) -> Result<(), (Str
        while state.status != ComputeStatus::Terminated {
            state = c.state_changed.wait(state).unwrap();
            info!(
-                "waiting for compute to become {}, current status: {:?}",
-                ComputeStatus::Terminated,
+                "waiting for compute to become Terminated, current status: {:?}",
                state.status
            );
        }
--- a/compute_tools/src/spec.rs
+++ b/compute_tools/src/spec.rs
@@ -1,4 +1,3 @@
-use std::collections::HashSet;
 use std::fs::File;
 use std::path::Path;
 use std::str::FromStr;
@@ -190,15 +189,6 @@ pub fn handle_roles(spec: &ComputeSpec, client: &mut Client) -> Result<()> {
    let mut xact = client.transaction()?;
    let existing_roles: Vec<Role> = get_existing_roles(&mut xact)?;

-    let mut jwks_roles = HashSet::new();
-    if let Some(local_proxy) = &spec.local_proxy_config {
-        for jwks_setting in local_proxy.jwks.iter().flatten() {
-            for role_name in &jwks_setting.role_names {
-                jwks_roles.insert(role_name.clone());
-            }
-        }
-    }
-
    // Print a list of existing Postgres roles (only in debug mode)
    if span_enabled!(Level::INFO) {
        let mut vec = Vec::new();
@@ -318,9 +308,6 @@ pub fn handle_roles(spec: &ComputeSpec, client: &mut Client) -> Result<()> {
                    "CREATE ROLE {} INHERIT CREATEROLE CREATEDB BYPASSRLS REPLICATION IN ROLE neon_superuser",
                    name.pg_quote()
                );
-                if jwks_roles.contains(name.as_str()) {
-                    query = format!("CREATE ROLE {}", name.pg_quote());
-                }
                info!("running role create query: '{}'", &query);
                query.push_str(&role.to_pg_options());
                xact.execute(query.as_str(), &[])?;
--- a/libs/compute_api/src/responses.rs
+++ b/libs/compute_api/src/responses.rs
@@ -1,7 +1,5 @@
 //! Structs representing the JSON formats used in the compute_ctl's HTTP API.

-use std::fmt::Display;
-
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize, Serializer};

@@ -60,21 +58,6 @@ pub enum ComputeStatus {
    Terminated,
 }

-impl Display for ComputeStatus {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        match self {
-            ComputeStatus::Empty => f.write_str("empty"),
-            ComputeStatus::ConfigurationPending => f.write_str("configuration-pending"),
-            ComputeStatus::Init => f.write_str("init"),
-            ComputeStatus::Running => f.write_str("running"),
-            ComputeStatus::Configuration => f.write_str("configuration"),
-            ComputeStatus::Failed => f.write_str("failed"),
-            ComputeStatus::TerminationPending => f.write_str("termination-pending"),
-            ComputeStatus::Terminated => f.write_str("terminated"),
-        }
-    }
-}
-
 fn rfc3339_serialize<S>(x: &Option<DateTime<Utc>>, s: S) -> Result<S::Ok, S::Error>
 where
    S: Serializer,
--- a/libs/compute_api/src/spec.rs
+++ b/libs/compute_api/src/spec.rs
@@ -286,9 +286,7 @@ pub type GenericOptions = Option<Vec<GenericOption>>;
 /// use for authorizing connect requests using JWT.
 #[derive(Clone, Debug, Deserialize, Serialize)]
 pub struct LocalProxySpec {
-    #[serde(default)]
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub jwks: Option<Vec<JwksSettings>>,
+    pub jwks: Vec<JwksSettings>,
 }

 #[derive(Clone, Debug, Deserialize, Serialize)]
--- a/libs/vm_monitor/src/runner.rs
+++ b/libs/vm_monitor/src/runner.rs
@@ -79,7 +79,8 @@ pub struct Config {
    /// memory.
    ///
    /// The default value of `0.15` means that we *guarantee* sending upscale requests if the
-    /// cgroup is using more than 85% of total memory.
+    /// cgroup is using more than 85% of total memory (even if we're *not* separately reserving
+    /// memory for the file cache).
    cgroup_min_overhead_fraction: f64,

    cgroup_downscale_threshold_buffer_bytes: u64,
@@ -96,12 +97,24 @@ impl Default for Config {
 }

 impl Config {
-    fn cgroup_threshold(&self, total_mem: u64) -> u64 {
-        // We want our threshold to be met gracefully instead of letting postgres get OOM-killed
-        // (or if there's room, spilling to swap).
+    fn cgroup_threshold(&self, total_mem: u64, file_cache_disk_size: u64) -> u64 {
+        // If the file cache is in tmpfs, then it will count towards shmem usage of the cgroup,
+        // and thus be non-reclaimable, so we should allow for additional memory usage.
+        //
+        // If the file cache sits on disk, our desired stable system state is for it to be fully
+        // page cached (its contents should only be paged to/from disk in situations where we can't
+        // upscale fast enough). Page-cached memory is reclaimable, so we need to lower the
+        // threshold for non-reclaimable memory so we scale up *before* the kernel starts paging
+        // out the file cache.
+        let memory_remaining_for_cgroup = total_mem.saturating_sub(file_cache_disk_size);
+
+        // Even if we're not separately making room for the file cache (if it's in tmpfs), we still
+        // want our threshold to be met gracefully instead of letting postgres get OOM-killed.
        // So we guarantee that there's at least `cgroup_min_overhead_fraction` of total memory
        // remaining above the threshold.
-        (total_mem as f64 * (1.0 - self.cgroup_min_overhead_fraction)) as u64
+        let max_threshold = (total_mem as f64 * (1.0 - self.cgroup_min_overhead_fraction)) as u64;
+
+        memory_remaining_for_cgroup.min(max_threshold)
    }
 }

@@ -136,6 +149,11 @@ impl Runner {

        let mem = get_total_system_memory();

+        let mut file_cache_disk_size = 0;
+
+        // We need to process file cache initialization before cgroup initialization, so that the memory
+        // allocated to the file cache is appropriately taken into account when we decide the cgroup's
+        // memory limits.
        if let Some(connstr) = &args.pgconnstr {
            info!("initializing file cache");
            let config = FileCacheConfig::default();
@@ -166,6 +184,7 @@ impl Runner {
                info!("file cache size actually got set to {actual_size}")
            }

+            file_cache_disk_size = actual_size;
            state.filecache = Some(file_cache);
        }

@@ -188,7 +207,7 @@ impl Runner {
                cgroup.watch(hist_tx).await
            });

-            let threshold = state.config.cgroup_threshold(mem);
+            let threshold = state.config.cgroup_threshold(mem, file_cache_disk_size);
            info!(threshold, "set initial cgroup threshold",);

            state.cgroup = Some(CgroupState {
@@ -240,7 +259,9 @@ impl Runner {
                return Ok((false, status.to_owned()));
            }

-            let new_threshold = self.config.cgroup_threshold(usable_system_memory);
+            let new_threshold = self
+                .config
+                .cgroup_threshold(usable_system_memory, expected_file_cache_size);

            let current = last_history.avg_non_reclaimable;

@@ -261,11 +282,13 @@ impl Runner {

        // The downscaling has been approved. Downscale the file cache, then the cgroup.
        let mut status = vec![];
+        let mut file_cache_disk_size = 0;
        if let Some(file_cache) = &mut self.filecache {
            let actual_usage = file_cache
                .set_file_cache_size(expected_file_cache_size)
                .await
                .context("failed to set file cache size")?;
+            file_cache_disk_size = actual_usage;
            let message = format!(
                "set file cache size to {} MiB",
                bytes_to_mebibytes(actual_usage),
@@ -275,7 +298,9 @@ impl Runner {
        }

        if let Some(cgroup) = &mut self.cgroup {
-            let new_threshold = self.config.cgroup_threshold(usable_system_memory);
+            let new_threshold = self
+                .config
+                .cgroup_threshold(usable_system_memory, file_cache_disk_size);

            let message = format!(
                "set cgroup memory threshold from {} MiB to {} MiB, of new total {} MiB",
@@ -304,6 +329,7 @@ impl Runner {
        let new_mem = resources.mem;
        let usable_system_memory = new_mem.saturating_sub(self.config.sys_buffer_bytes);

+        let mut file_cache_disk_size = 0;
        if let Some(file_cache) = &mut self.filecache {
            let expected_usage = file_cache.config.calculate_cache_size(usable_system_memory);
            info!(
@@ -316,6 +342,7 @@ impl Runner {
                .set_file_cache_size(expected_usage)
                .await
                .context("failed to set file cache size")?;
+            file_cache_disk_size = actual_usage;

            if actual_usage != expected_usage {
                warn!(
@@ -327,7 +354,9 @@ impl Runner {
        }

        if let Some(cgroup) = &mut self.cgroup {
-            let new_threshold = self.config.cgroup_threshold(usable_system_memory);
+            let new_threshold = self
+                .config
+                .cgroup_threshold(usable_system_memory, file_cache_disk_size);

            info!(
                "set cgroup memory threshold from {} MiB to {} MiB of new total {} MiB",
--- a/pageserver/src/http/routes.rs
+++ b/pageserver/src/http/routes.rs
@@ -703,8 +703,6 @@ async fn timeline_archival_config_handler(
    let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
    let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;

-    let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
-
    let request_data: TimelineArchivalConfigRequest = json_request(&mut request).await?;
    check_permission(&request, Some(tenant_shard_id.tenant_id))?;
    let state = get_state(&request);
@@ -715,7 +713,7 @@ async fn timeline_archival_config_handler(
            .get_attached_tenant_shard(tenant_shard_id)?;

        tenant
-            .apply_timeline_archival_config(timeline_id, request_data.state, ctx)
+            .apply_timeline_archival_config(timeline_id, request_data.state)
            .await?;
        Ok::<_, ApiError>(())
    }
--- a/pageserver/src/tenant.rs
+++ b/pageserver/src/tenant.rs
@@ -38,7 +38,6 @@ use std::future::Future;
 use std::sync::Weak;
 use std::time::SystemTime;
 use storage_broker::BrokerClientChannel;
-use timeline::offload::offload_timeline;
 use tokio::io::BufReader;
 use tokio::sync::watch;
 use tokio::task::JoinSet;
@@ -288,13 +287,9 @@ pub struct Tenant {

    /// During timeline creation, we first insert the TimelineId to the
    /// creating map, then `timelines`, then remove it from the creating map.
-    /// **Lock order**: if acquiring both, acquire`timelines` before `timelines_creating`
+    /// **Lock order**: if acquring both, acquire`timelines` before `timelines_creating`
    timelines_creating: std::sync::Mutex<HashSet<TimelineId>>,

-    /// Possibly offloaded and archived timelines
-    /// **Lock order**: if acquiring both, acquire`timelines` before `timelines_offloaded`
-    timelines_offloaded: Mutex<HashMap<TimelineId, Arc<OffloadedTimeline>>>,
-
    // This mutex prevents creation of new timelines during GC.
    // Adding yet another mutex (in addition to `timelines`) is needed because holding
    // `timelines` mutex during all GC iteration
@@ -489,65 +484,6 @@ impl WalRedoManager {
    }
 }

-pub struct OffloadedTimeline {
-    pub tenant_shard_id: TenantShardId,
-    pub timeline_id: TimelineId,
-    pub ancestor_timeline_id: Option<TimelineId>,
-
-    // TODO: once we persist offloaded state, make this lazily constructed
-    pub remote_client: Arc<RemoteTimelineClient>,
-
-    /// Prevent two tasks from deleting the timeline at the same time. If held, the
-    /// timeline is being deleted. If 'true', the timeline has already been deleted.
-    pub delete_progress: Arc<tokio::sync::Mutex<DeleteTimelineFlow>>,
-}
-
-impl OffloadedTimeline {
-    fn from_timeline(timeline: &Timeline) -> Self {
-        Self {
-            tenant_shard_id: timeline.tenant_shard_id,
-            timeline_id: timeline.timeline_id,
-            ancestor_timeline_id: timeline.get_ancestor_timeline_id(),
-
-            remote_client: timeline.remote_client.clone(),
-            delete_progress: timeline.delete_progress.clone(),
-        }
-    }
-}
-
-#[derive(Clone)]
-pub enum TimelineOrOffloaded {
-    Timeline(Arc<Timeline>),
-    Offloaded(Arc<OffloadedTimeline>),
-}
-
-impl TimelineOrOffloaded {
-    pub fn tenant_shard_id(&self) -> TenantShardId {
-        match self {
-            TimelineOrOffloaded::Timeline(timeline) => timeline.tenant_shard_id,
-            TimelineOrOffloaded::Offloaded(offloaded) => offloaded.tenant_shard_id,
-        }
-    }
-    pub fn timeline_id(&self) -> TimelineId {
-        match self {
-            TimelineOrOffloaded::Timeline(timeline) => timeline.timeline_id,
-            TimelineOrOffloaded::Offloaded(offloaded) => offloaded.timeline_id,
-        }
-    }
-    pub fn delete_progress(&self) -> &Arc<tokio::sync::Mutex<DeleteTimelineFlow>> {
-        match self {
-            TimelineOrOffloaded::Timeline(timeline) => &timeline.delete_progress,
-            TimelineOrOffloaded::Offloaded(offloaded) => &offloaded.delete_progress,
-        }
-    }
-    pub fn remote_client(&self) -> &Arc<RemoteTimelineClient> {
-        match self {
-            TimelineOrOffloaded::Timeline(timeline) => &timeline.remote_client,
-            TimelineOrOffloaded::Offloaded(offloaded) => &offloaded.remote_client,
-        }
-    }
-}
-
 #[derive(Debug, thiserror::Error, PartialEq, Eq)]
 pub enum GetTimelineError {
    #[error("Timeline is shutting down")]
@@ -1470,192 +1406,52 @@ impl Tenant {
        }
    }

-    fn check_to_be_archived_has_no_unarchived_children(
-        timeline_id: TimelineId,
-        timelines: &std::sync::MutexGuard<'_, HashMap<TimelineId, Arc<Timeline>>>,
-    ) -> Result<(), TimelineArchivalError> {
-        let children: Vec<TimelineId> = timelines
-            .iter()
-            .filter_map(|(id, entry)| {
-                if entry.get_ancestor_timeline_id() != Some(timeline_id) {
-                    return None;
-                }
-                if entry.is_archived() == Some(true) {
-                    return None;
-                }
-                Some(*id)
-            })
-            .collect();
-
-        if !children.is_empty() {
-            return Err(TimelineArchivalError::HasUnarchivedChildren(children));
-        }
-        Ok(())
-    }
-
-    fn check_ancestor_of_to_be_unarchived_is_not_archived(
-        ancestor_timeline_id: TimelineId,
-        timelines: &std::sync::MutexGuard<'_, HashMap<TimelineId, Arc<Timeline>>>,
-        offloaded_timelines: &std::sync::MutexGuard<
-            '_,
-            HashMap<TimelineId, Arc<OffloadedTimeline>>,
-        >,
-    ) -> Result<(), TimelineArchivalError> {
-        let has_archived_parent =
-            if let Some(ancestor_timeline) = timelines.get(&ancestor_timeline_id) {
-                ancestor_timeline.is_archived() == Some(true)
-            } else if offloaded_timelines.contains_key(&ancestor_timeline_id) {
-                true
-            } else {
-                error!("ancestor timeline {ancestor_timeline_id} not found");
-                if cfg!(debug_assertions) {
-                    panic!("ancestor timeline {ancestor_timeline_id} not found");
-                }
-                return Err(TimelineArchivalError::NotFound);
-            };
-        if has_archived_parent {
-            return Err(TimelineArchivalError::HasArchivedParent(
-                ancestor_timeline_id,
-            ));
-        }
-        Ok(())
-    }
-
-    fn check_to_be_unarchived_timeline_has_no_archived_parent(
-        timeline: &Arc<Timeline>,
-    ) -> Result<(), TimelineArchivalError> {
-        if let Some(ancestor_timeline) = timeline.ancestor_timeline() {
-            if ancestor_timeline.is_archived() == Some(true) {
-                return Err(TimelineArchivalError::HasArchivedParent(
-                    ancestor_timeline.timeline_id,
-                ));
-            }
-        }
-        Ok(())
-    }
-
-    /// Loads the specified (offloaded) timeline from S3 and attaches it as a loaded timeline
-    async fn unoffload_timeline(
-        self: &Arc<Self>,
-        timeline_id: TimelineId,
-        ctx: RequestContext,
-    ) -> Result<Arc<Timeline>, TimelineArchivalError> {
-        let cancel = self.cancel.clone();
-        let timeline_preload = self
-            .load_timeline_metadata(timeline_id, self.remote_storage.clone(), cancel)
-            .await;
-
-        let index_part = match timeline_preload.index_part {
-            Ok(index_part) => {
-                debug!("remote index part exists for timeline {timeline_id}");
-                index_part
-            }
-            Err(DownloadError::NotFound) => {
-                error!(%timeline_id, "index_part not found on remote");
-                return Err(TimelineArchivalError::NotFound);
-            }
-            Err(e) => {
-                // Some (possibly ephemeral) error happened during index_part download.
-                warn!(%timeline_id, "Failed to load index_part from remote storage, failed creation? ({e})");
-                return Err(TimelineArchivalError::Other(
-                    anyhow::Error::new(e).context("downloading index_part from remote storage"),
-                ));
-            }
-        };
-        let index_part = match index_part {
-            MaybeDeletedIndexPart::IndexPart(index_part) => index_part,
-            MaybeDeletedIndexPart::Deleted(_index_part) => {
-                info!("timeline is deleted according to index_part.json");
-                return Err(TimelineArchivalError::NotFound);
-            }
-        };
-        let remote_metadata = index_part.metadata.clone();
-        let timeline_resources = self.build_timeline_resources(timeline_id);
-        self.load_remote_timeline(
-            timeline_id,
-            index_part,
-            remote_metadata,
-            timeline_resources,
-            &ctx,
-        )
-        .await
-        .with_context(|| {
-            format!(
-                "failed to load remote timeline {} for tenant {}",
-                timeline_id, self.tenant_shard_id
-            )
-        })?;
-        let timelines = self.timelines.lock().unwrap();
-        if let Some(timeline) = timelines.get(&timeline_id) {
-            let mut offloaded_timelines = self.timelines_offloaded.lock().unwrap();
-            if offloaded_timelines.remove(&timeline_id).is_none() {
-                warn!("timeline already removed from offloaded timelines");
-            }
-            Ok(Arc::clone(timeline))
-        } else {
-            warn!("timeline not available directly after attach");
-            Err(TimelineArchivalError::Other(anyhow::anyhow!(
-                "timeline not available directly after attach"
-            )))
-        }
-    }
-
    pub(crate) async fn apply_timeline_archival_config(
-        self: &Arc<Self>,
+        &self,
        timeline_id: TimelineId,
-        new_state: TimelineArchivalState,
-        ctx: RequestContext,
+        state: TimelineArchivalState,
    ) -> Result<(), TimelineArchivalError> {
        info!("setting timeline archival config");
-        // First part: figure out what is needed to do, and do validation
-        let timeline_or_unarchive_offloaded = 'outer: {
+        let timeline = {
            let timelines = self.timelines.lock().unwrap();

            let Some(timeline) = timelines.get(&timeline_id) else {
-                let offloaded_timelines = self.timelines_offloaded.lock().unwrap();
-                let Some(offloaded) = offloaded_timelines.get(&timeline_id) else {
-                    return Err(TimelineArchivalError::NotFound);
-                };
-                if new_state == TimelineArchivalState::Archived {
-                    // It's offloaded already, so nothing to do
-                    return Ok(());
-                }
-                if let Some(ancestor_timeline_id) = offloaded.ancestor_timeline_id {
-                    Self::check_ancestor_of_to_be_unarchived_is_not_archived(
-                        ancestor_timeline_id,
-                        &timelines,
-                        &offloaded_timelines,
-                    )?;
-                }
-                break 'outer None;
+                return Err(TimelineArchivalError::NotFound);
            };

-            // Do some validation. We release the timelines lock below, so there is potential
-            // for race conditions: these checks are more present to prevent misunderstandings of
-            // the API's capabilities, instead of serving as the sole way to defend their invariants.
-            match new_state {
-                TimelineArchivalState::Unarchived => {
-                    Self::check_to_be_unarchived_timeline_has_no_archived_parent(timeline)?
-                }
-                TimelineArchivalState::Archived => {
-                    Self::check_to_be_archived_has_no_unarchived_children(timeline_id, &timelines)?
+            if state == TimelineArchivalState::Unarchived {
+                if let Some(ancestor_timeline) = timeline.ancestor_timeline() {
+                    if ancestor_timeline.is_archived() == Some(true) {
+                        return Err(TimelineArchivalError::HasArchivedParent(
+                            ancestor_timeline.timeline_id,
+                        ));
+                    }
                }
            }
-            Some(Arc::clone(timeline))
+
+            // Ensure that there are no non-archived child timelines
+            let children: Vec<TimelineId> = timelines
+                .iter()
+                .filter_map(|(id, entry)| {
+                    if entry.get_ancestor_timeline_id() != Some(timeline_id) {
+                        return None;
+                    }
+                    if entry.is_archived() == Some(true) {
+                        return None;
+                    }
+                    Some(*id)
+                })
+                .collect();
+
+            if !children.is_empty() && state == TimelineArchivalState::Archived {
+                return Err(TimelineArchivalError::HasUnarchivedChildren(children));
+            }
+            Arc::clone(timeline)
        };

-        // Second part: unarchive timeline (if needed)
-        let timeline = if let Some(timeline) = timeline_or_unarchive_offloaded {
-            timeline
-        } else {
-            // Turn offloaded timeline into a non-offloaded one
-            self.unoffload_timeline(timeline_id, ctx).await?
-        };
-
-        // Third part: upload new timeline archival state and block until it is present in S3
        let upload_needed = timeline
            .remote_client
-            .schedule_index_upload_for_timeline_archival_state(new_state)?;
+            .schedule_index_upload_for_timeline_archival_state(state)?;

        if upload_needed {
            info!("Uploading new state");
@@ -2088,7 +1884,7 @@ impl Tenant {
    ///
    /// Returns whether we have pending compaction task.
    async fn compaction_iteration(
-        self: &Arc<Self>,
+        &self,
        cancel: &CancellationToken,
        ctx: &RequestContext,
    ) -> Result<bool, timeline::CompactionError> {
@@ -2109,28 +1905,21 @@ impl Tenant {
        // while holding the lock. Then drop the lock and actually perform the
        // compactions.  We don't want to block everything else while the
        // compaction runs.
-        let timelines_to_compact_or_offload;
-        {
+        let timelines_to_compact = {
            let timelines = self.timelines.lock().unwrap();
-            timelines_to_compact_or_offload = timelines
+            let timelines_to_compact = timelines
                .iter()
                .filter_map(|(timeline_id, timeline)| {
-                    let (is_active, can_offload) = (timeline.is_active(), timeline.can_offload());
-                    let has_no_unoffloaded_children = {
-                        !timelines
-                            .iter()
-                            .any(|(_id, tl)| tl.get_ancestor_timeline_id() == Some(*timeline_id))
-                    };
-                    let can_offload = can_offload && has_no_unoffloaded_children;
-                    if (is_active, can_offload) == (false, false) {
-                        None
+                    if timeline.is_active() {
+                        Some((*timeline_id, timeline.clone()))
                    } else {
-                        Some((*timeline_id, timeline.clone(), (is_active, can_offload)))
+                        None
                    }
                })
                .collect::<Vec<_>>();
            drop(timelines);
-        }
+            timelines_to_compact
+        };

        // Before doing any I/O work, check our circuit breaker
        if self.compaction_circuit_breaker.lock().unwrap().is_broken() {
@@ -2140,34 +1929,20 @@ impl Tenant {

        let mut has_pending_task = false;

-        for (timeline_id, timeline, (can_compact, can_offload)) in &timelines_to_compact_or_offload
-        {
-            let pending_task_left = if *can_compact {
-                Some(
-                    timeline
-                        .compact(cancel, EnumSet::empty(), ctx)
-                        .instrument(info_span!("compact_timeline", %timeline_id))
-                        .await
-                        .inspect_err(|e| match e {
-                            timeline::CompactionError::ShuttingDown => (),
-                            timeline::CompactionError::Other(e) => {
-                                self.compaction_circuit_breaker
-                                    .lock()
-                                    .unwrap()
-                                    .fail(&CIRCUIT_BREAKERS_BROKEN, e);
-                            }
-                        })?,
-                )
-            } else {
-                None
-            };
-            has_pending_task |= pending_task_left.unwrap_or(false);
-            if pending_task_left == Some(false) && *can_offload {
-                offload_timeline(self, timeline)
-                    .instrument(info_span!("offload_timeline", %timeline_id))
-                    .await
-                    .map_err(timeline::CompactionError::Other)?;
-            }
+        for (timeline_id, timeline) in &timelines_to_compact {
+            has_pending_task |= timeline
+                .compact(cancel, EnumSet::empty(), ctx)
+                .instrument(info_span!("compact_timeline", %timeline_id))
+                .await
+                .inspect_err(|e| match e {
+                    timeline::CompactionError::ShuttingDown => (),
+                    timeline::CompactionError::Other(e) => {
+                        self.compaction_circuit_breaker
+                            .lock()
+                            .unwrap()
+                            .fail(&CIRCUIT_BREAKERS_BROKEN, e);
+                    }
+                })?;
        }

        self.compaction_circuit_breaker
@@ -3077,7 +2852,6 @@ impl Tenant {
            constructed_at: Instant::now(),
            timelines: Mutex::new(HashMap::new()),
            timelines_creating: Mutex::new(HashSet::new()),
-            timelines_offloaded: Mutex::new(HashMap::new()),
            gc_cs: tokio::sync::Mutex::new(()),
            walredo_mgr,
            remote_storage,
--- a/pageserver/src/tenant/gc_block.rs
+++ b/pageserver/src/tenant/gc_block.rs
@@ -141,14 +141,14 @@ impl GcBlock {
        Ok(())
    }

-    pub(crate) fn before_delete(&self, timeline_id: &super::TimelineId) {
+    pub(crate) fn before_delete(&self, timeline: &super::Timeline) {
        let unblocked = {
            let mut g = self.reasons.lock().unwrap();
            if g.is_empty() {
                return;
            }

-            g.remove(timeline_id);
+            g.remove(&timeline.timeline_id);

            BlockingReasons::clean_and_summarize(g).is_none()
        };
--- a/pageserver/src/tenant/timeline.rs
+++ b/pageserver/src/tenant/timeline.rs
@@ -7,7 +7,6 @@ pub(crate) mod handle;
 mod init;
 pub mod layer_manager;
 pub(crate) mod logical_size;
-pub mod offload;
 pub mod span;
 pub mod uninit;
 mod walreceiver;
@@ -1557,17 +1556,6 @@ impl Timeline {
        }
    }

-    /// Checks if the internal state of the timeline is consistent with it being able to be offloaded.
-    /// This is neccessary but not sufficient for offloading of the timeline as it might have
-    /// child timelines that are not offloaded yet.
-    pub(crate) fn can_offload(&self) -> bool {
-        if self.remote_client.is_archived() != Some(true) {
-            return false;
-        }
-
-        true
-    }
-
    /// Outermost timeline compaction operation; downloads needed layers. Returns whether we have pending
    /// compaction tasks.
    pub(crate) async fn compact(
@@ -1830,6 +1818,7 @@ impl Timeline {
        self.current_state() == TimelineState::Active
    }

+    #[allow(unused)]
    pub(crate) fn is_archived(&self) -> Option<bool> {
        self.remote_client.is_archived()
    }
--- a/pageserver/src/tenant/timeline/delete.rs
+++ b/pageserver/src/tenant/timeline/delete.rs
@@ -15,7 +15,7 @@ use crate::{
    tenant::{
        metadata::TimelineMetadata,
        remote_timeline_client::{PersistIndexPartWithDeletedFlagError, RemoteTimelineClient},
-        CreateTimelineCause, DeleteTimelineError, Tenant, TimelineOrOffloaded,
+        CreateTimelineCause, DeleteTimelineError, Tenant,
    },
 };

@@ -24,14 +24,12 @@ use super::{Timeline, TimelineResources};
 /// Mark timeline as deleted in S3 so we won't pick it up next time
 /// during attach or pageserver restart.
 /// See comment in persist_index_part_with_deleted_flag.
-async fn set_deleted_in_remote_index(
-    timeline: &TimelineOrOffloaded,
-) -> Result<(), DeleteTimelineError> {
-    let res = timeline
-        .remote_client()
+async fn set_deleted_in_remote_index(timeline: &Timeline) -> Result<(), DeleteTimelineError> {
+    match timeline
+        .remote_client
        .persist_index_part_with_deleted_flag()
-        .await;
-    match res {
+        .await
+    {
        // If we (now, or already) marked it successfully as deleted, we can proceed
        Ok(()) | Err(PersistIndexPartWithDeletedFlagError::AlreadyDeleted(_)) => (),
        // Bail out otherwise
@@ -129,9 +127,9 @@ pub(super) async fn delete_local_timeline_directory(
 }

 /// Removes remote layers and an index file after them.
-async fn delete_remote_layers_and_index(timeline: &TimelineOrOffloaded) -> anyhow::Result<()> {
+async fn delete_remote_layers_and_index(timeline: &Timeline) -> anyhow::Result<()> {
    timeline
-        .remote_client()
+        .remote_client
        .delete_all()
        .await
        .context("delete_all")
@@ -139,41 +137,27 @@ async fn delete_remote_layers_and_index(timeline: &TimelineOrOffloaded) -> anyho

 /// It is important that this gets called when DeletionGuard is being held.
 /// For more context see comments in [`DeleteTimelineFlow::prepare`]
-async fn remove_maybe_offloaded_timeline_from_tenant(
+async fn remove_timeline_from_tenant(
    tenant: &Tenant,
-    timeline: &TimelineOrOffloaded,
+    timeline: &Timeline,
    _: &DeletionGuard, // using it as a witness
 ) -> anyhow::Result<()> {
    // Remove the timeline from the map.
-    // This observes the locking order between timelines and timelines_offloaded
    let mut timelines = tenant.timelines.lock().unwrap();
-    let mut timelines_offloaded = tenant.timelines_offloaded.lock().unwrap();
-    let offloaded_children_exist = timelines_offloaded
-        .iter()
-        .any(|(_, entry)| entry.ancestor_timeline_id == Some(timeline.timeline_id()));
    let children_exist = timelines
        .iter()
-        .any(|(_, entry)| entry.get_ancestor_timeline_id() == Some(timeline.timeline_id()));
-    // XXX this can happen because of race conditions with branch creation.
-    // We already deleted the remote layer files, so it's probably best to panic.
-    if children_exist || offloaded_children_exist {
+        .any(|(_, entry)| entry.get_ancestor_timeline_id() == Some(timeline.timeline_id));
+    // XXX this can happen because `branch_timeline` doesn't check `TimelineState::Stopping`.
+    // We already deleted the layer files, so it's probably best to panic.
+    // (Ideally, above remove_dir_all is atomic so we don't see this timeline after a restart)
+    if children_exist {
        panic!("Timeline grew children while we removed layer files");
    }

-    match timeline {
-        TimelineOrOffloaded::Timeline(timeline) => {
-            timelines.remove(&timeline.timeline_id).expect(
-                "timeline that we were deleting was concurrently removed from 'timelines' map",
-            );
-        }
-        TimelineOrOffloaded::Offloaded(timeline) => {
-            timelines_offloaded
-                .remove(&timeline.timeline_id)
-                .expect("timeline that we were deleting was concurrently removed from 'timelines_offloaded' map");
-        }
-    }
+    timelines
+        .remove(&timeline.timeline_id)
+        .expect("timeline that we were deleting was concurrently removed from 'timelines' map");

-    drop(timelines_offloaded);
    drop(timelines);

    Ok(())
@@ -223,11 +207,9 @@ impl DeleteTimelineFlow {
        guard.mark_in_progress()?;

        // Now that the Timeline is in Stopping state, request all the related tasks to shut down.
-        if let TimelineOrOffloaded::Timeline(timeline) = &timeline {
-            timeline.shutdown(super::ShutdownMode::Hard).await;
-        }
+        timeline.shutdown(super::ShutdownMode::Hard).await;

-        tenant.gc_block.before_delete(&timeline.timeline_id());
+        tenant.gc_block.before_delete(&timeline);

        fail::fail_point!("timeline-delete-before-index-deleted-at", |_| {
            Err(anyhow::anyhow!(
@@ -303,16 +285,15 @@ impl DeleteTimelineFlow {

        guard.mark_in_progress()?;

-        let timeline = TimelineOrOffloaded::Timeline(timeline);
        Self::schedule_background(guard, tenant.conf, tenant, timeline);

        Ok(())
    }

-    pub(super) fn prepare(
+    fn prepare(
        tenant: &Tenant,
        timeline_id: TimelineId,
-    ) -> Result<(TimelineOrOffloaded, DeletionGuard), DeleteTimelineError> {
+    ) -> Result<(Arc<Timeline>, DeletionGuard), DeleteTimelineError> {
        // Note the interaction between this guard and deletion guard.
        // Here we attempt to lock deletion guard when we're holding a lock on timelines.
        // This is important because when you take into account `remove_timeline_from_tenant`
@@ -326,14 +307,8 @@ impl DeleteTimelineFlow {
        let timelines = tenant.timelines.lock().unwrap();

        let timeline = match timelines.get(&timeline_id) {
-            Some(t) => TimelineOrOffloaded::Timeline(Arc::clone(t)),
-            None => {
-                let offloaded_timelines = tenant.timelines_offloaded.lock().unwrap();
-                match offloaded_timelines.get(&timeline_id) {
-                    Some(t) => TimelineOrOffloaded::Offloaded(Arc::clone(t)),
-                    None => return Err(DeleteTimelineError::NotFound),
-                }
-            }
+            Some(t) => t,
+            None => return Err(DeleteTimelineError::NotFound),
        };

        // Ensure that there are no child timelines **attached to that pageserver**,
@@ -359,32 +334,30 @@ impl DeleteTimelineFlow {
        // to remove the timeline from it.
        // Always if you have two locks that are taken in different order this can result in a deadlock.

-        let delete_progress = Arc::clone(timeline.delete_progress());
+        let delete_progress = Arc::clone(&timeline.delete_progress);
        let delete_lock_guard = match delete_progress.try_lock_owned() {
            Ok(guard) => DeletionGuard(guard),
            Err(_) => {
                // Unfortunately if lock fails arc is consumed.
                return Err(DeleteTimelineError::AlreadyInProgress(Arc::clone(
-                    timeline.delete_progress(),
+                    &timeline.delete_progress,
                )));
            }
        };

-        if let TimelineOrOffloaded::Timeline(timeline) = &timeline {
-            timeline.set_state(TimelineState::Stopping);
-        }
+        timeline.set_state(TimelineState::Stopping);

-        Ok((timeline, delete_lock_guard))
+        Ok((Arc::clone(timeline), delete_lock_guard))
    }

    fn schedule_background(
        guard: DeletionGuard,
        conf: &'static PageServerConf,
        tenant: Arc<Tenant>,
-        timeline: TimelineOrOffloaded,
+        timeline: Arc<Timeline>,
    ) {
-        let tenant_shard_id = timeline.tenant_shard_id();
-        let timeline_id = timeline.timeline_id();
+        let tenant_shard_id = timeline.tenant_shard_id;
+        let timeline_id = timeline.timeline_id;

        task_mgr::spawn(
            task_mgr::BACKGROUND_RUNTIME.handle(),
@@ -395,9 +368,7 @@ impl DeleteTimelineFlow {
            async move {
                if let Err(err) = Self::background(guard, conf, &tenant, &timeline).await {
                    error!("Error: {err:#}");
-                    if let TimelineOrOffloaded::Timeline(timeline) = timeline {
-                        timeline.set_broken(format!("{err:#}"))
-                    }
+                    timeline.set_broken(format!("{err:#}"))
                };
                Ok(())
            }
@@ -409,19 +380,15 @@ impl DeleteTimelineFlow {
        mut guard: DeletionGuard,
        conf: &PageServerConf,
        tenant: &Tenant,
-        timeline: &TimelineOrOffloaded,
+        timeline: &Timeline,
    ) -> Result<(), DeleteTimelineError> {
-        // Offloaded timelines have no local state
-        // TODO: once we persist offloaded information, delete the timeline from there, too
-        if let TimelineOrOffloaded::Timeline(timeline) = timeline {
-            delete_local_timeline_directory(conf, tenant.tenant_shard_id, timeline).await?;
-        }
+        delete_local_timeline_directory(conf, tenant.tenant_shard_id, timeline).await?;

        delete_remote_layers_and_index(timeline).await?;

        pausable_failpoint!("in_progress_delete");

-        remove_maybe_offloaded_timeline_from_tenant(tenant, timeline, &guard).await?;
+        remove_timeline_from_tenant(tenant, timeline, &guard).await?;

        *guard = Self::Finished;

@@ -433,7 +400,7 @@ impl DeleteTimelineFlow {
    }
 }

-pub(super) struct DeletionGuard(OwnedMutexGuard<DeleteTimelineFlow>);
+struct DeletionGuard(OwnedMutexGuard<DeleteTimelineFlow>);

 impl Deref for DeletionGuard {
    type Target = DeleteTimelineFlow;
--- a/pageserver/src/tenant/timeline/offload.rs
+++ b/pageserver/src/tenant/timeline/offload.rs
@@ -1,69 +0,0 @@
-use std::sync::Arc;
-
-use crate::tenant::{OffloadedTimeline, Tenant, TimelineOrOffloaded};
-
-use super::{
-    delete::{delete_local_timeline_directory, DeleteTimelineFlow, DeletionGuard},
-    Timeline,
-};
-
-pub(crate) async fn offload_timeline(
-    tenant: &Tenant,
-    timeline: &Arc<Timeline>,
-) -> anyhow::Result<()> {
-    tracing::info!("offloading archived timeline");
-    let (timeline, guard) = DeleteTimelineFlow::prepare(tenant, timeline.timeline_id)?;
-
-    let TimelineOrOffloaded::Timeline(timeline) = timeline else {
-        tracing::error!("timeline already offloaded, but given timeline object");
-        return Ok(());
-    };
-
-    // TODO extend guard mechanism above with method
-    // to make deletions possible while offloading is in progress
-
-    // TODO mark timeline as offloaded in S3
-
-    let conf = &tenant.conf;
-    delete_local_timeline_directory(conf, tenant.tenant_shard_id, &timeline).await?;
-
-    remove_timeline_from_tenant(tenant, &timeline, &guard).await?;
-
-    {
-        let mut offloaded_timelines = tenant.timelines_offloaded.lock().unwrap();
-        offloaded_timelines.insert(
-            timeline.timeline_id,
-            Arc::new(OffloadedTimeline::from_timeline(&timeline)),
-        );
-    }
-
-    Ok(())
-}
-
-/// It is important that this gets called when DeletionGuard is being held.
-/// For more context see comments in [`DeleteTimelineFlow::prepare`]
-async fn remove_timeline_from_tenant(
-    tenant: &Tenant,
-    timeline: &Timeline,
-    _: &DeletionGuard, // using it as a witness
-) -> anyhow::Result<()> {
-    // Remove the timeline from the map.
-    let mut timelines = tenant.timelines.lock().unwrap();
-    let children_exist = timelines
-        .iter()
-        .any(|(_, entry)| entry.get_ancestor_timeline_id() == Some(timeline.timeline_id));
-    // XXX this can happen because `branch_timeline` doesn't check `TimelineState::Stopping`.
-    // We already deleted the layer files, so it's probably best to panic.
-    // (Ideally, above remove_dir_all is atomic so we don't see this timeline after a restart)
-    if children_exist {
-        panic!("Timeline grew children while we removed layer files");
-    }
-
-    timelines
-        .remove(&timeline.timeline_id)
-        .expect("timeline that we were deleting was concurrently removed from 'timelines' map");
-
-    drop(timelines);
-
-    Ok(())
-}
--- a/pgxn/neon/control_plane_connector.c
+++ b/pgxn/neon/control_plane_connector.c
@@ -146,8 +146,6 @@ ConstructDeltaMessage()
 	if (RootTable.role_table)
 	{
 		JsonbValue	roles;
-		HASH_SEQ_STATUS status;
-		RoleEntry  *entry;

 		roles.type = jbvString;
 		roles.val.string.val = "roles";
@@ -155,6 +153,9 @@ ConstructDeltaMessage()
 		pushJsonbValue(&state, WJB_KEY, &roles);
 		pushJsonbValue(&state, WJB_BEGIN_ARRAY, NULL);

+		HASH_SEQ_STATUS status;
+		RoleEntry  *entry;
+
 		hash_seq_init(&status, RootTable.role_table);
 		while ((entry = hash_seq_search(&status)) != NULL)
 		{
@@ -189,12 +190,10 @@ ConstructDeltaMessage()
 		}
 		pushJsonbValue(&state, WJB_END_ARRAY, NULL);
 	}
-	{
-		JsonbValue *result = pushJsonbValue(&state, WJB_END_OBJECT, NULL);
-		Jsonb	   *jsonb = JsonbValueToJsonb(result);
+	JsonbValue *result = pushJsonbValue(&state, WJB_END_OBJECT, NULL);
+	Jsonb	   *jsonb = JsonbValueToJsonb(result);

-		return JsonbToCString(NULL, &jsonb->root, 0 /* estimated_len */ );
-	}
+	return JsonbToCString(NULL, &jsonb->root, 0 /* estimated_len */ );
 }

 #define ERROR_SIZE 1024
@@ -273,28 +272,32 @@ SendDeltasToControlPlane()
 		curl_easy_setopt(handle, CURLOPT_WRITEFUNCTION, ErrorWriteCallback);
 	}

+	char	   *message = ConstructDeltaMessage();
+	ErrorString str;
+
+	str.size = 0;
+
+	curl_easy_setopt(handle, CURLOPT_POSTFIELDS, message);
+	curl_easy_setopt(handle, CURLOPT_WRITEDATA, &str);
+
+	const int	num_retries = 5;
+	CURLcode	curl_status;
+
+	for (int i = 0; i < num_retries; i++)
+	{
+		if ((curl_status = curl_easy_perform(handle)) == 0)
+			break;
+		elog(LOG, "Curl request failed on attempt %d: %s", i, CurlErrorBuf);
+		pg_usleep(1000 * 1000);
+	}
+	if (curl_status != CURLE_OK)
+	{
+		elog(ERROR, "Failed to perform curl request: %s", CurlErrorBuf);
+	}
+	else
 	{
-		char	   *message = ConstructDeltaMessage();
-		ErrorString str;
-		const int	num_retries = 5;
-		CURLcode	curl_status;
 		long		response_code;

-		str.size = 0;
-
-		curl_easy_setopt(handle, CURLOPT_POSTFIELDS, message);
-		curl_easy_setopt(handle, CURLOPT_WRITEDATA, &str);
-
-		for (int i = 0; i < num_retries; i++)
-		{
-			if ((curl_status = curl_easy_perform(handle)) == 0)
-				break;
-			elog(LOG, "Curl request failed on attempt %d: %s", i, CurlErrorBuf);
-			pg_usleep(1000 * 1000);
-		}
-		if (curl_status != CURLE_OK)
-			elog(ERROR, "Failed to perform curl request: %s", CurlErrorBuf);
-
 		if (curl_easy_getinfo(handle, CURLINFO_RESPONSE_CODE, &response_code) != CURLE_UNKNOWN_OPTION)
 		{
 			if (response_code != 200)
@@ -373,11 +376,10 @@ MergeTable()

 	if (old_table->db_table)
 	{
+		InitDbTableIfNeeded();
 		DbEntry    *entry;
 		HASH_SEQ_STATUS status;

-		InitDbTableIfNeeded();
-
 		hash_seq_init(&status, old_table->db_table);
 		while ((entry = hash_seq_search(&status)) != NULL)
 		{
@@ -419,11 +421,10 @@ MergeTable()

 	if (old_table->role_table)
 	{
+		InitRoleTableIfNeeded();
 		RoleEntry  *entry;
 		HASH_SEQ_STATUS status;

-		InitRoleTableIfNeeded();
-
 		hash_seq_init(&status, old_table->role_table);
 		while ((entry = hash_seq_search(&status)) != NULL)
 		{
@@ -514,12 +515,9 @@ RoleIsNeonSuperuser(const char *role_name)
 static void
 HandleCreateDb(CreatedbStmt *stmt)
 {
+	InitDbTableIfNeeded();
 	DefElem    *downer = NULL;
 	ListCell   *option;
-	bool		found = false;
-	DbEntry    *entry;
-
-	InitDbTableIfNeeded();

 	foreach(option, stmt->options)
 	{
@@ -528,11 +526,13 @@ HandleCreateDb(CreatedbStmt *stmt)
 		if (strcmp(defel->defname, "owner") == 0)
 			downer = defel;
 	}
+	bool		found = false;
+	DbEntry    *entry = hash_search(
+									CurrentDdlTable->db_table,
+									stmt->dbname,
+									HASH_ENTER,
+									&found);

-	entry = hash_search(CurrentDdlTable->db_table,
-						stmt->dbname,
-						HASH_ENTER,
-						&found);
 	if (!found)
 		memset(entry->old_name, 0, sizeof(entry->old_name));

@@ -554,24 +554,21 @@ HandleCreateDb(CreatedbStmt *stmt)
 static void
 HandleAlterOwner(AlterOwnerStmt *stmt)
 {
-	const char *name;
-	bool		found = false;
-	DbEntry    *entry;
-	const char *new_owner;
-
 	if (stmt->objectType != OBJECT_DATABASE)
 		return;
 	InitDbTableIfNeeded();
+	const char *name = strVal(stmt->object);
+	bool		found = false;
+	DbEntry    *entry = hash_search(
+									CurrentDdlTable->db_table,
+									name,
+									HASH_ENTER,
+									&found);

-	name = strVal(stmt->object);
-	entry = hash_search(CurrentDdlTable->db_table,
-						name,
-						HASH_ENTER,
-						&found);
 	if (!found)
 		memset(entry->old_name, 0, sizeof(entry->old_name));
+	const char *new_owner = get_rolespec_name(stmt->newowner);

-	new_owner = get_rolespec_name(stmt->newowner);
 	if (RoleIsNeonSuperuser(new_owner))
 		elog(ERROR, "can't alter owner to neon_superuser");
 	entry->owner = get_role_oid(new_owner, false);
@@ -581,23 +578,21 @@ HandleAlterOwner(AlterOwnerStmt *stmt)
 static void
 HandleDbRename(RenameStmt *stmt)
 {
-	bool		found = false;
-	DbEntry    *entry;
-	DbEntry    *entry_for_new_name;
-
 	Assert(stmt->renameType == OBJECT_DATABASE);
 	InitDbTableIfNeeded();
-	entry = hash_search(CurrentDdlTable->db_table,
-						stmt->subname,
-						HASH_FIND,
-						&found);
+	bool		found = false;
+	DbEntry    *entry = hash_search(
+									CurrentDdlTable->db_table,
+									stmt->subname,
+									HASH_FIND,
+									&found);
+	DbEntry    *entry_for_new_name = hash_search(
+												 CurrentDdlTable->db_table,
+												 stmt->newname,
+												 HASH_ENTER,
+												 NULL);

-	entry_for_new_name = hash_search(CurrentDdlTable->db_table,
-									 stmt->newname,
-									 HASH_ENTER,
-									 NULL);
 	entry_for_new_name->type = Op_Set;
-
 	if (found)
 	{
 		if (entry->old_name[0] != '\0')
@@ -605,7 +600,8 @@ HandleDbRename(RenameStmt *stmt)
 		else
 			strlcpy(entry_for_new_name->old_name, entry->name, NAMEDATALEN);
 		entry_for_new_name->owner = entry->owner;
-		hash_search(CurrentDdlTable->db_table,
+		hash_search(
+					CurrentDdlTable->db_table,
 					stmt->subname,
 					HASH_REMOVE,
 					NULL);
@@ -620,15 +616,14 @@ HandleDbRename(RenameStmt *stmt)
 static void
 HandleDropDb(DropdbStmt *stmt)
 {
-	bool		found = false;
-	DbEntry    *entry;
-
 	InitDbTableIfNeeded();
+	bool		found = false;
+	DbEntry    *entry = hash_search(
+									CurrentDdlTable->db_table,
+									stmt->dbname,
+									HASH_ENTER,
+									&found);

-	entry = hash_search(CurrentDdlTable->db_table,
-						stmt->dbname,
-						HASH_ENTER,
-						&found);
 	entry->type = Op_Delete;
 	entry->owner = InvalidOid;
 	if (!found)
@@ -638,14 +633,16 @@ HandleDropDb(DropdbStmt *stmt)
 static void
 HandleCreateRole(CreateRoleStmt *stmt)
 {
+	InitRoleTableIfNeeded();
 	bool		found = false;
-	RoleEntry  *entry;
-	DefElem    *dpass;
+	RoleEntry  *entry = hash_search(
+									CurrentDdlTable->role_table,
+									stmt->role,
+									HASH_ENTER,
+									&found);
+	DefElem    *dpass = NULL;
 	ListCell   *option;

-	InitRoleTableIfNeeded();
-
-	dpass = NULL;
 	foreach(option, stmt->options)
 	{
 		DefElem    *defel = lfirst(option);
@@ -653,11 +650,6 @@ HandleCreateRole(CreateRoleStmt *stmt)
 		if (strcmp(defel->defname, "password") == 0)
 			dpass = defel;
 	}
-
-	entry = hash_search(CurrentDdlTable->role_table,
-						stmt->role,
-						HASH_ENTER,
-						&found);
 	if (!found)
 		memset(entry->old_name, 0, sizeof(entry->old_name));
 	if (dpass && dpass->arg)
@@ -670,18 +662,14 @@ HandleCreateRole(CreateRoleStmt *stmt)
 static void
 HandleAlterRole(AlterRoleStmt *stmt)
 {
-	const char *role_name = stmt->role->rolename;
-	DefElem    *dpass;
-	ListCell   *option;
-	bool		found = false;
-	RoleEntry  *entry;
-
 	InitRoleTableIfNeeded();
+	DefElem    *dpass = NULL;
+	ListCell   *option;
+	const char *role_name = stmt->role->rolename;

 	if (RoleIsNeonSuperuser(role_name) && !superuser())
 		elog(ERROR, "can't ALTER neon_superuser");

-	dpass = NULL;
 	foreach(option, stmt->options)
 	{
 		DefElem    *defel = lfirst(option);
@@ -692,11 +680,13 @@ HandleAlterRole(AlterRoleStmt *stmt)
 	/* We only care about updates to the password */
 	if (!dpass)
 		return;
+	bool		found = false;
+	RoleEntry  *entry = hash_search(
+									CurrentDdlTable->role_table,
+									role_name,
+									HASH_ENTER,
+									&found);

-	entry = hash_search(CurrentDdlTable->role_table,
-						role_name,
-						HASH_ENTER,
-						&found);
 	if (!found)
 		memset(entry->old_name, 0, sizeof(entry->old_name));
 	if (dpass->arg)
@@ -709,22 +699,20 @@ HandleAlterRole(AlterRoleStmt *stmt)
 static void
 HandleRoleRename(RenameStmt *stmt)
 {
-	bool		found = false;
-	RoleEntry  *entry;
-	RoleEntry  *entry_for_new_name;
-
-	Assert(stmt->renameType == OBJECT_ROLE);
 	InitRoleTableIfNeeded();
+	Assert(stmt->renameType == OBJECT_ROLE);
+	bool		found = false;
+	RoleEntry  *entry = hash_search(
+									CurrentDdlTable->role_table,
+									stmt->subname,
+									HASH_FIND,
+									&found);

-	entry = hash_search(CurrentDdlTable->role_table,
-						stmt->subname,
-						HASH_FIND,
-						&found);
-
-	entry_for_new_name = hash_search(CurrentDdlTable->role_table,
-									 stmt->newname,
-									 HASH_ENTER,
-									 NULL);
+	RoleEntry  *entry_for_new_name = hash_search(
+												 CurrentDdlTable->role_table,
+												 stmt->newname,
+												 HASH_ENTER,
+												 NULL);

 	entry_for_new_name->type = Op_Set;
 	if (found)
@@ -750,9 +738,8 @@ HandleRoleRename(RenameStmt *stmt)
 static void
 HandleDropRole(DropRoleStmt *stmt)
 {
-	ListCell   *item;
-
 	InitRoleTableIfNeeded();
+	ListCell   *item;

 	foreach(item, stmt->roles)
 	{
--- a/pgxn/neon/file_cache.c
+++ b/pgxn/neon/file_cache.c
@@ -170,14 +170,12 @@ lfc_disable(char const *op)

 		if (lfc_desc > 0)
 		{
-			int			rc;
-
 			/*
 			 * If the reason of error is ENOSPC, then truncation of file may
 			 * help to reclaim some space
 			 */
 			pgstat_report_wait_start(WAIT_EVENT_NEON_LFC_TRUNCATE);
-			rc = ftruncate(lfc_desc, 0);
+			int			rc = ftruncate(lfc_desc, 0);
 			pgstat_report_wait_end();

 			if (rc < 0)
@@ -618,7 +616,7 @@ lfc_evict(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno)
 	 */
 	if (entry->bitmap[chunk_offs >> 5] == 0)
 	{
-		bool		has_remaining_pages = false;
+		bool		has_remaining_pages;

 		for (int i = 0; i < CHUNK_BITMAP_SIZE; i++)
 		{
@@ -668,6 +666,7 @@ lfc_readv_select(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno,
 	BufferTag	tag;
 	FileCacheEntry *entry;
 	ssize_t		rc;
+	bool		result = true;
 	uint32		hash;
 	uint64		generation;
 	uint32		entry_offset;
@@ -926,10 +925,10 @@ lfc_writev(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno,
 				/* We can reuse a hole that was left behind when the LFC was shrunk previously */
 				FileCacheEntry *hole = dlist_container(FileCacheEntry, list_node, dlist_pop_head_node(&lfc_ctl->holes));
 				uint32		offset = hole->offset;
-				bool		hole_found;
+				bool		found;
 	
-				hash_search_with_hash_value(lfc_hash, &hole->key, hole->hash, HASH_REMOVE, &hole_found);
-				CriticalAssert(hole_found);
+				hash_search_with_hash_value(lfc_hash, &hole->key, hole->hash, HASH_REMOVE, &found);
+				CriticalAssert(found);
 	
 				lfc_ctl->used += 1;
 				entry->offset = offset;	/* reuse the hole */
@@ -1005,7 +1004,7 @@ neon_get_lfc_stats(PG_FUNCTION_ARGS)
 	Datum		result;
 	HeapTuple	tuple;
 	char const *key;
-	uint64		value = 0;
+	uint64		value;
 	Datum		values[NUM_NEON_GET_STATS_COLS];
 	bool		nulls[NUM_NEON_GET_STATS_COLS];

--- a/pgxn/neon/hll.c
+++ b/pgxn/neon/hll.c
@@ -116,6 +116,8 @@ addSHLL(HyperLogLogState *cState, uint32 hash)
 {
 	uint8		count;
 	uint32		index;
+	size_t		i;
+	size_t		j;

 	TimestampTz	now = GetCurrentTimestamp();
 	/* Use the first "k" (registerWidth) bits as a zero based index */
--- a/pgxn/neon/libpagestore.c
+++ b/pgxn/neon/libpagestore.c
@@ -89,6 +89,7 @@ typedef struct

 #if PG_VERSION_NUM >= 150000
 static shmem_request_hook_type prev_shmem_request_hook = NULL;
+static void walproposer_shmem_request(void);
 #endif
 static shmem_startup_hook_type prev_shmem_startup_hook;
 static PagestoreShmemState *pagestore_shared;
@@ -440,8 +441,8 @@ pageserver_connect(shardno_t shard_no, int elevel)
 			return false;
 		}
 		shard->state = PS_Connecting_Startup;
+		/* fallthrough */
 	}
-	/* FALLTHROUGH */
 	case PS_Connecting_Startup:
 	{
 		char	   *pagestream_query;
@@ -452,6 +453,8 @@ pageserver_connect(shardno_t shard_no, int elevel)

 		do
 		{
+			WaitEvent	event;
+
 			switch (poll_result)
 			{
 			default: /* unknown/unused states are handled as a failed connection */
@@ -582,8 +585,8 @@ pageserver_connect(shardno_t shard_no, int elevel)
 		}

 		shard->state = PS_Connecting_PageStream;
+		/* fallthrough */
 	}
-	/* FALLTHROUGH */
 	case PS_Connecting_PageStream:
 	{
 		neon_shard_log(shard_no, DEBUG5, "Connection state: Connecting_PageStream");
@@ -628,8 +631,8 @@ pageserver_connect(shardno_t shard_no, int elevel)
 		}

 		shard->state = PS_Connected;
+		/* fallthrough */
 	}
-	/* FALLTHROUGH */
 	case PS_Connected:
 		/*
 		 * We successfully connected. Future connections to this PageServer
--- a/pgxn/neon/neon_perf_counters.c
+++ b/pgxn/neon/neon_perf_counters.c
@@ -94,6 +94,7 @@ neon_perf_counters_to_metrics(neon_per_backend_counters *counters)
 	metric_t   *metrics = palloc((NUM_METRICS + 1) * sizeof(metric_t));
 	uint64		bucket_accum;
 	int			i = 0;
+	Datum		getpage_wait_str;

 	metrics[i].name = "getpage_wait_seconds_count";
 	metrics[i].is_bucket = false;
@@ -223,6 +224,7 @@ neon_get_perf_counters(PG_FUNCTION_ARGS)
 	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
 	Datum		values[3];
 	bool		nulls[3];
+	Datum		getpage_wait_str;
 	neon_per_backend_counters totals = {0};
 	metric_t   *metrics;

--- a/pgxn/neon/neon_pgversioncompat.h
+++ b/pgxn/neon/neon_pgversioncompat.h
@@ -7,7 +7,6 @@
 #define NEON_PGVERSIONCOMPAT_H

 #include "fmgr.h"
-#include "storage/buf_internals.h"

 #if PG_MAJORVERSION_NUM < 17
 #define NRelFileInfoBackendIsTemp(rinfo) (rinfo.backend != InvalidBackendId)
@@ -21,24 +20,11 @@
 	NInfoGetRelNumber(a) == NInfoGetRelNumber(b) \
 )

-/* These macros were turned into static inline functions in v16 */
+/* buftag population & RelFileNode/RelFileLocator rework */
 #if PG_MAJORVERSION_NUM < 16
-static inline bool
-BufferTagsEqual(const BufferTag *tag1, const BufferTag *tag2)
-{
-	return BUFFERTAGS_EQUAL(*tag1, *tag2);
-}

-static inline void
-InitBufferTag(BufferTag *tag, const RelFileNode *rnode,
-			  ForkNumber forkNum, BlockNumber blockNum)
-{
-	INIT_BUFFERTAG(*tag, *rnode, forkNum, blockNum);
-}
-#endif
+#define InitBufferTag(tag, rfn, fn, bn) INIT_BUFFERTAG(*tag, *rfn, fn, bn)

-/* RelFileNode -> RelFileLocator rework */
-#if PG_MAJORVERSION_NUM < 16
 #define USE_RELFILENODE

 #define RELFILEINFO_HDR "storage/relfilenode.h"
@@ -87,6 +73,8 @@ InitBufferTag(BufferTag *tag, const RelFileNode *rnode,

 #define USE_RELFILELOCATOR

+#define BUFFERTAGS_EQUAL(a, b) BufferTagsEqual(&(a), &(b))
+
 #define RELFILEINFO_HDR "storage/relfilelocator.h"

 #define NRelFileInfo RelFileLocator
--- a/pgxn/neon/pagestore_client.h
+++ b/pgxn/neon/pagestore_client.h
@@ -213,6 +213,32 @@ extern const f_smgr *smgr_neon(ProcNumber backend, NRelFileInfo rinfo);
 extern void smgr_init_neon(void);
 extern void readahead_buffer_resize(int newsize, void *extra);

+/* Neon storage manager functionality */
+
+extern void neon_init(void);
+extern void neon_open(SMgrRelation reln);
+extern void neon_close(SMgrRelation reln, ForkNumber forknum);
+extern void neon_create(SMgrRelation reln, ForkNumber forknum, bool isRedo);
+extern bool neon_exists(SMgrRelation reln, ForkNumber forknum);
+extern void neon_unlink(NRelFileInfoBackend rnode, ForkNumber forknum, bool isRedo);
+#if PG_MAJORVERSION_NUM < 16
+extern void neon_extend(SMgrRelation reln, ForkNumber forknum,
+						BlockNumber blocknum, char *buffer, bool skipFsync);
+#else
+extern void neon_extend(SMgrRelation reln, ForkNumber forknum,
+						BlockNumber blocknum, const void *buffer, bool skipFsync);
+extern void neon_zeroextend(SMgrRelation reln, ForkNumber forknum,
+							BlockNumber blocknum, int nbuffers, bool skipFsync);
+#endif
+
+#if PG_MAJORVERSION_NUM >=17
+extern bool neon_prefetch(SMgrRelation reln, ForkNumber forknum,
+						  BlockNumber blocknum, int nblocks);
+#else
+extern bool neon_prefetch(SMgrRelation reln, ForkNumber forknum,
+						  BlockNumber blocknum);
+#endif
+
 /*
 * LSN values associated with each request to the pageserver
 */
@@ -252,7 +278,13 @@ extern PGDLLEXPORT void neon_read_at_lsn(NRelFileInfo rnode, ForkNumber forkNum,
 extern PGDLLEXPORT void neon_read_at_lsn(NRelFileInfo rnode, ForkNumber forkNum, BlockNumber blkno,
 										 neon_request_lsns request_lsns, void *buffer);
 #endif
+extern void neon_writeback(SMgrRelation reln, ForkNumber forknum,
+						   BlockNumber blocknum, BlockNumber nblocks);
+extern BlockNumber neon_nblocks(SMgrRelation reln, ForkNumber forknum);
 extern int64 neon_dbsize(Oid dbNode);
+extern void neon_truncate(SMgrRelation reln, ForkNumber forknum,
+						  BlockNumber nblocks);
+extern void neon_immedsync(SMgrRelation reln, ForkNumber forknum);

 /* utils for neon relsize cache */
 extern void relsize_hash_init(void);
--- a/pgxn/neon/pagestore_smgr.c
+++ b/pgxn/neon/pagestore_smgr.c
@@ -118,8 +118,6 @@ static UnloggedBuildPhase unlogged_build_phase = UNLOGGED_BUILD_NOT_IN_PROGRESS;
 static bool neon_redo_read_buffer_filter(XLogReaderState *record, uint8 block_id);
 static bool (*old_redo_read_buffer_filter) (XLogReaderState *record, uint8 block_id) = NULL;

-static BlockNumber neon_nblocks(SMgrRelation reln, ForkNumber forknum);
-
 /*
 * Prefetch implementation:
 *
@@ -217,7 +215,7 @@ typedef struct PrfHashEntry
 	sizeof(BufferTag) \
 )

-#define SH_EQUAL(tb, a, b)	(BufferTagsEqual(&(a)->buftag, &(b)->buftag))
+#define SH_EQUAL(tb, a, b)	(BUFFERTAGS_EQUAL((a)->buftag, (b)->buftag))
 #define SH_SCOPE			static inline
 #define SH_DEFINE
 #define SH_DECLARE
@@ -738,7 +736,7 @@ static void
 prefetch_do_request(PrefetchRequest *slot, neon_request_lsns *force_request_lsns)
 {
 	bool		found;
-	uint64		mySlotNo PG_USED_FOR_ASSERTS_ONLY = slot->my_ring_index;
+	uint64		mySlotNo = slot->my_ring_index;

 	NeonGetPageRequest request = {
 		.req.tag = T_NeonGetPageRequest,
@@ -805,19 +803,15 @@ prefetch_register_bufferv(BufferTag tag, neon_request_lsns *frlsns,
 						  bool is_prefetch)
 {
 	uint64		min_ring_index;
-	PrefetchRequest hashkey;
+	PrefetchRequest req;
 #if USE_ASSERT_CHECKING
 	bool		any_hits = false;
 #endif
 	/* We will never read further ahead than our buffer can store. */
 	nblocks = Max(1, Min(nblocks, readahead_buffer_size));

-	/*
-	 * Use an intermediate PrefetchRequest struct as the hash key to ensure
-	 * correct alignment and that the padding bytes are cleared.
-	 */
-	memset(&hashkey.buftag, 0, sizeof(BufferTag));
-	hashkey.buftag = tag;
+	/* use an intermediate PrefetchRequest struct to ensure correct alignment */
+	req.buftag = tag;

 Retry:
 	min_ring_index = UINT64_MAX;
@@ -843,8 +837,8 @@ Retry:
 		slot = NULL;
 		entry = NULL;

-		hashkey.buftag.blockNum = tag.blockNum + i;
-		entry = prfh_lookup(MyPState->prf_hash, &hashkey);
+		req.buftag.blockNum = tag.blockNum + i;
+		entry = prfh_lookup(MyPState->prf_hash, (PrefetchRequest *) &req);

 		if (entry != NULL)
 		{
@@ -855,7 +849,7 @@ Retry:
 			Assert(slot->status != PRFS_UNUSED);
 			Assert(MyPState->ring_last <= ring_index &&
 				   ring_index < MyPState->ring_unused);
-			Assert(BufferTagsEqual(&slot->buftag, &hashkey.buftag));
+			Assert(BUFFERTAGS_EQUAL(slot->buftag, req.buftag));

 			/*
 			 * If the caller specified a request LSN to use, only accept
@@ -892,19 +886,12 @@ Retry:
 				{
 					min_ring_index = Min(min_ring_index, ring_index);
 					/* The buffered request is good enough, return that index */
-					if (is_prefetch)
-						pgBufferUsage.prefetch.duplicates++;
-					else
-						pgBufferUsage.prefetch.hits++;
+					pgBufferUsage.prefetch.duplicates++;
 					continue;
 				}
 			}
 		}
-		else if (!is_prefetch)
-		{
-			pgBufferUsage.prefetch.misses += 1;
-			MyNeonCounters->getpage_prefetch_misses_total++;
-		}
+
 		/*
 		 * We can only leave the block above by finding that there's
 		 * no entry that can satisfy this request, either because there
@@ -987,7 +974,7 @@ Retry:
 		 * We must update the slot data before insertion, because the hash
 		 * function reads the buffer tag from the slot.
 		 */
-		slot->buftag = hashkey.buftag;
+		slot->buftag = req.buftag;
 		slot->shard_no = get_shard_number(&tag);
 		slot->my_ring_index = ring_index;

@@ -1465,6 +1452,7 @@ log_newpages_copy(NRelFileInfo * rinfo, ForkNumber forkNum, BlockNumber blkno,
 	BlockNumber	blknos[XLR_MAX_BLOCK_ID];
 	Page		pageptrs[XLR_MAX_BLOCK_ID];
 	int			nregistered = 0;
+	XLogRecPtr	result = 0;

 	for (int i = 0; i < nblocks; i++)
 	{
@@ -1777,7 +1765,7 @@ neon_wallog_page(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, co
 /*
 *	neon_init() -- Initialize private state
 */
-static void
+void
 neon_init(void)
 {
 	Size		prfs_size;
@@ -2167,7 +2155,7 @@ neon_prefetch_response_usable(neon_request_lsns *request_lsns,
 /*
 *	neon_exists() -- Does the physical file exist?
 */
-static bool
+bool
 neon_exists(SMgrRelation reln, ForkNumber forkNum)
 {
 	bool		exists;
@@ -2273,7 +2261,7 @@ neon_exists(SMgrRelation reln, ForkNumber forkNum)
 *
 * If isRedo is true, it's okay for the relation to exist already.
 */
-static void
+void
 neon_create(SMgrRelation reln, ForkNumber forkNum, bool isRedo)
 {
 	switch (reln->smgr_relpersistence)
@@ -2349,7 +2337,7 @@ neon_create(SMgrRelation reln, ForkNumber forkNum, bool isRedo)
 * Note: any failure should be reported as WARNING not ERROR, because
 * we are usually not in a transaction anymore when this is called.
 */
-static void
+void
 neon_unlink(NRelFileInfoBackend rinfo, ForkNumber forkNum, bool isRedo)
 {
 	/*
@@ -2373,7 +2361,7 @@ neon_unlink(NRelFileInfoBackend rinfo, ForkNumber forkNum, bool isRedo)
 *		EOF).  Note that we assume writing a block beyond current EOF
 *		causes intervening file space to become filled with zeroes.
 */
-static void
+void
 #if PG_MAJORVERSION_NUM < 16
 neon_extend(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 			char *buffer, bool skipFsync)
@@ -2465,7 +2453,7 @@ neon_extend(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 }

 #if PG_MAJORVERSION_NUM >= 16
-static void
+void
 neon_zeroextend(SMgrRelation reln, ForkNumber forkNum, BlockNumber blocknum,
 				int nblocks, bool skipFsync)
 {
@@ -2561,7 +2549,7 @@ neon_zeroextend(SMgrRelation reln, ForkNumber forkNum, BlockNumber blocknum,
 /*
 *  neon_open() -- Initialize newly-opened relation.
 */
-static void
+void
 neon_open(SMgrRelation reln)
 {
 	/*
@@ -2579,7 +2567,7 @@ neon_open(SMgrRelation reln)
 /*
 *	neon_close() -- Close the specified relation, if it isn't closed already.
 */
-static void
+void
 neon_close(SMgrRelation reln, ForkNumber forknum)
 {
 	/*
@@ -2594,12 +2582,13 @@ neon_close(SMgrRelation reln, ForkNumber forknum)
 /*
 *	neon_prefetch() -- Initiate asynchronous read of the specified block of a relation
 */
-static bool
+bool
 neon_prefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 			  int nblocks)
 {
 	uint64		ring_index PG_USED_FOR_ASSERTS_ONLY;
 	BufferTag	tag;
+	bool		io_initiated = false;

 	switch (reln->smgr_relpersistence)
 	{
@@ -2623,6 +2612,7 @@ neon_prefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 	while (nblocks > 0)
 	{
 		int		iterblocks = Min(nblocks, PG_IOV_MAX);
+		int		seqlen = 0;
 		bits8		lfc_present[PG_IOV_MAX / 8];
 		memset(lfc_present, 0, sizeof(lfc_present));

@@ -2634,6 +2624,8 @@ neon_prefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 			continue;
 		}

+		io_initiated = true;
+
 		tag.blockNum = blocknum;
 		
 		for (int i = 0; i < PG_IOV_MAX / 8; i++)
@@ -2656,7 +2648,7 @@ neon_prefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 /*
 *	neon_prefetch() -- Initiate asynchronous read of the specified block of a relation
 */
-static bool
+bool
 neon_prefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum)
 {
 	uint64		ring_index PG_USED_FOR_ASSERTS_ONLY;
@@ -2700,7 +2692,7 @@ neon_prefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum)
 * This accepts a range of blocks because flushing several pages at once is
 * considerably more efficient than doing so individually.
 */
-static void
+void
 neon_writeback(SMgrRelation reln, ForkNumber forknum,
 			   BlockNumber blocknum, BlockNumber nblocks)
 {
@@ -2750,19 +2742,14 @@ neon_read_at_lsnv(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber base_block
 	uint64		ring_index;
 	PrfHashEntry *entry;
 	PrefetchRequest *slot;
-	PrefetchRequest hashkey;
+	BufferTag	buftag = {0};

 	Assert(PointerIsValid(request_lsns));
 	Assert(nblocks >= 1);

-	/*
-	 * Use an intermediate PrefetchRequest struct as the hash key to ensure
-	 * correct alignment and that the padding bytes are cleared.
-	 */
-	memset(&hashkey.buftag, 0, sizeof(BufferTag));
-	CopyNRelFileInfoToBufTag(hashkey.buftag, rinfo);
-	hashkey.buftag.forkNum = forkNum;
-	hashkey.buftag.blockNum = base_blockno;
+	CopyNRelFileInfoToBufTag(buftag, rinfo);
+	buftag.forkNum = forkNum;
+	buftag.blockNum = base_blockno;

 	/*
 	 * The redo process does not lock pages that it needs to replay but are
@@ -2780,7 +2767,7 @@ neon_read_at_lsnv(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber base_block
 	 * weren't for the behaviour of the LwLsn cache that uses the highest
 	 * value of the LwLsn cache when the entry is not found.
 	 */
-	prefetch_register_bufferv(hashkey.buftag, request_lsns, nblocks, mask, false);
+	prefetch_register_bufferv(buftag, request_lsns, nblocks, mask, false);

 	for (int i = 0; i < nblocks; i++)
 	{
@@ -2801,8 +2788,8 @@ neon_read_at_lsnv(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber base_block
 		 * Try to find prefetched page in the list of received pages.
 		 */
 Retry:
-		hashkey.buftag.blockNum = blockno;
-		entry = prfh_lookup(MyPState->prf_hash, &hashkey);
+		buftag.blockNum = blockno;
+		entry = prfh_lookup(MyPState->prf_hash, (PrefetchRequest *) &buftag);

 		if (entry != NULL)
 		{
@@ -2810,6 +2797,7 @@ Retry:
 			if (neon_prefetch_response_usable(reqlsns, slot))
 			{
 				ring_index = slot->my_ring_index;
+				pgBufferUsage.prefetch.hits += 1;
 			}
 			else
 			{
@@ -2839,7 +2827,10 @@ Retry:
 		{
 			if (entry == NULL)
 			{
-				ring_index = prefetch_register_bufferv(hashkey.buftag, reqlsns, 1, NULL, false);
+				pgBufferUsage.prefetch.misses += 1;
+				MyNeonCounters->getpage_prefetch_misses_total++;
+
+				ring_index = prefetch_register_bufferv(buftag, reqlsns, 1, NULL, false);
 				Assert(ring_index != UINT64_MAX);
 				slot = GetPrfSlot(ring_index);
 			}
@@ -2864,8 +2855,8 @@ Retry:
 		} while (!prefetch_wait_for(ring_index));

 		Assert(slot->status == PRFS_RECEIVED);
-		Assert(memcmp(&hashkey.buftag, &slot->buftag, sizeof(BufferTag)) == 0);
-		Assert(hashkey.buftag.blockNum == base_blockno + i);
+		Assert(memcmp(&buftag, &slot->buftag, sizeof(BufferTag)) == 0);
+		Assert(buftag.blockNum == base_blockno + i);

 		resp = slot->response;

@@ -2921,10 +2912,10 @@ neon_read_at_lsn(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno,
 *	neon_read() -- Read the specified block from a relation.
 */
 #if PG_MAJORVERSION_NUM < 16
-static void
+void
 neon_read(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno, char *buffer)
 #else
-static void
+void
 neon_read(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno, void *buffer)
 #endif
 {
@@ -3033,7 +3024,7 @@ neon_read(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno, void *buffer
 #endif /* PG_MAJORVERSION_NUM <= 16 */

 #if PG_MAJORVERSION_NUM >= 17
-static void
+void
 neon_readv(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 		void **buffers, BlockNumber nblocks)
 {
@@ -3068,9 +3059,6 @@ neon_readv(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 	lfc_result = lfc_readv_select(InfoFromSMgrRel(reln), forknum, blocknum, buffers,
 								  nblocks, read);

-	if (lfc_result > 0)
-		MyNeonCounters->file_cache_hits_total += lfc_result;
-
 	/* Read all blocks from LFC, so we're done */
 	if (lfc_result == nblocks)
 		return;
@@ -3197,7 +3185,6 @@ hexdump_page(char *page)
 }
 #endif

-#if PG_MAJORVERSION_NUM < 17
 /*
 *	neon_write() -- Write the supplied block at the appropriate location.
 *
@@ -3205,7 +3192,7 @@ hexdump_page(char *page)
 *		relation (ie, those before the current EOF).  To extend a relation,
 *		use mdextend().
 */
-static void
+void
 #if PG_MAJORVERSION_NUM < 16
 neon_write(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, char *buffer, bool skipFsync)
 #else
@@ -3271,12 +3258,11 @@ neon_write(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, const vo
 		#endif
 #endif
 }
-#endif



 #if PG_MAJORVERSION_NUM >= 17
-static void
+void
 neon_writev(SMgrRelation reln, ForkNumber forknum, BlockNumber blkno,
 			 const void **buffers, BlockNumber nblocks, bool skipFsync)
 {
@@ -3326,7 +3312,7 @@ neon_writev(SMgrRelation reln, ForkNumber forknum, BlockNumber blkno,
 /*
 *	neon_nblocks() -- Get the number of blocks stored in a relation.
 */
-static BlockNumber
+BlockNumber
 neon_nblocks(SMgrRelation reln, ForkNumber forknum)
 {
 	NeonResponse *resp;
@@ -3463,7 +3449,7 @@ neon_dbsize(Oid dbNode)
 /*
 *	neon_truncate() -- Truncate relation to specified number of blocks.
 */
-static void
+void
 neon_truncate(SMgrRelation reln, ForkNumber forknum, BlockNumber nblocks)
 {
 	XLogRecPtr	lsn;
@@ -3532,7 +3518,7 @@ neon_truncate(SMgrRelation reln, ForkNumber forknum, BlockNumber nblocks)
 * crash before the next checkpoint syncs the newly-inactive segment, that
 * segment may survive recovery, reintroducing unwanted data into the table.
 */
-static void
+void
 neon_immedsync(SMgrRelation reln, ForkNumber forknum)
 {
 	switch (reln->smgr_relpersistence)
@@ -3562,8 +3548,8 @@ neon_immedsync(SMgrRelation reln, ForkNumber forknum)
 }

 #if PG_MAJORVERSION_NUM >= 17
-static void
-neon_registersync(SMgrRelation reln, ForkNumber forknum)
+void
+neon_regisersync(SMgrRelation reln, ForkNumber forknum)
 {
 	switch (reln->smgr_relpersistence)
 	{
@@ -3747,8 +3733,6 @@ neon_read_slru_segment(SMgrRelation reln, const char* path, int segno, void* buf
 	SlruKind	kind;
 	int			n_blocks;
 	shardno_t	shard_no = 0; /* All SLRUs are at shard 0 */
-	NeonResponse *resp;
-	NeonGetSlruSegmentRequest request;

 	/*
 	 * Compute a request LSN to use, similar to neon_get_request_lsns() but the
@@ -3787,7 +3771,8 @@ neon_read_slru_segment(SMgrRelation reln, const char* path, int segno, void* buf
 	else
 		return -1;

-	request = (NeonGetSlruSegmentRequest) {
+	NeonResponse *resp;
+	NeonGetSlruSegmentRequest request = {
 		.req.tag = T_NeonGetSlruSegmentRequest,
 		.req.lsn = request_lsn,
 		.req.not_modified_since = not_modified_since,
@@ -3894,7 +3879,7 @@ static const struct f_smgr neon_smgr =
 	.smgr_truncate = neon_truncate,
 	.smgr_immedsync = neon_immedsync,
 #if PG_MAJORVERSION_NUM >= 17
-	.smgr_registersync = neon_registersync,
+	.smgr_registersync = neon_regisersync,
 #endif
 	.smgr_start_unlogged_build = neon_start_unlogged_build,
 	.smgr_finish_unlogged_build_phase_1 = neon_finish_unlogged_build_phase_1,
--- a/pgxn/neon/walproposer.c
+++ b/pgxn/neon/walproposer.c
@@ -252,6 +252,8 @@ WalProposerPoll(WalProposer *wp)
 		/* timeout expired: poll state */
 		if (rc == 0 || TimeToReconnect(wp, now) <= 0)
 		{
+			TimestampTz now;
+
 			/*
 			 * If no WAL was generated during timeout (and we have already
 			 * collected the quorum), then send empty keepalive message
@@ -267,7 +269,8 @@ WalProposerPoll(WalProposer *wp)
 			now = wp->api.get_current_timestamp(wp);
 			for (int i = 0; i < wp->n_safekeepers; i++)
 			{
-				sk = &wp->safekeeper[i];
+				Safekeeper *sk = &wp->safekeeper[i];
+
 				if (TimestampDifferenceExceeds(sk->latestMsgReceivedAt, now,
 											   wp->config->safekeeper_connection_timeout))
 				{
@@ -1077,7 +1080,7 @@ SendProposerElected(Safekeeper *sk)
 	ProposerElected msg;
 	TermHistory *th;
 	term_t		lastCommonTerm;
-	int			idx;
+	int			i;

 	/* Now that we are ready to send it's a good moment to create WAL reader */
 	wp->api.wal_reader_allocate(sk);
@@ -1096,15 +1099,15 @@ SendProposerElected(Safekeeper *sk)
 	/* We must start somewhere. */
 	Assert(wp->propTermHistory.n_entries >= 1);

-	for (idx = 0; idx < Min(wp->propTermHistory.n_entries, th->n_entries); idx++)
+	for (i = 0; i < Min(wp->propTermHistory.n_entries, th->n_entries); i++)
 	{
-		if (wp->propTermHistory.entries[idx].term != th->entries[idx].term)
+		if (wp->propTermHistory.entries[i].term != th->entries[i].term)
 			break;
 		/* term must begin everywhere at the same point */
-		Assert(wp->propTermHistory.entries[idx].lsn == th->entries[idx].lsn);
+		Assert(wp->propTermHistory.entries[i].lsn == th->entries[i].lsn);
 	}
-	idx--;						/* step back to the last common term */
-	if (idx < 0)
+	i--;						/* step back to the last common term */
+	if (i < 0)
 	{
 		/* safekeeper is empty or no common point, start from the beginning */
 		sk->startStreamingAt = wp->propTermHistory.entries[0].lsn;
@@ -1125,14 +1128,14 @@ SendProposerElected(Safekeeper *sk)
 		 * proposer, LSN it is currently writing, but then we just pick
 		 * safekeeper pos as it obviously can't be higher.
 		 */
-		if (wp->propTermHistory.entries[idx].term == wp->propTerm)
+		if (wp->propTermHistory.entries[i].term == wp->propTerm)
 		{
 			sk->startStreamingAt = sk->voteResponse.flushLsn;
 		}
 		else
 		{
-			XLogRecPtr	propEndLsn = wp->propTermHistory.entries[idx + 1].lsn;
-			XLogRecPtr	skEndLsn = (idx + 1 < th->n_entries ? th->entries[idx + 1].lsn : sk->voteResponse.flushLsn);
+			XLogRecPtr	propEndLsn = wp->propTermHistory.entries[i + 1].lsn;
+			XLogRecPtr	skEndLsn = (i + 1 < th->n_entries ? th->entries[i + 1].lsn : sk->voteResponse.flushLsn);

 			sk->startStreamingAt = Min(propEndLsn, skEndLsn);
 		}
@@ -1146,7 +1149,7 @@ SendProposerElected(Safekeeper *sk)
 	msg.termHistory = &wp->propTermHistory;
 	msg.timelineStartLsn = wp->timelineStartLsn;

-	lastCommonTerm = idx >= 0 ? wp->propTermHistory.entries[idx].term : 0;
+	lastCommonTerm = i >= 0 ? wp->propTermHistory.entries[i].term : 0;
 	wp_log(LOG,
 		   "sending elected msg to node " UINT64_FORMAT " term=" UINT64_FORMAT ", startStreamingAt=%X/%X (lastCommonTerm=" UINT64_FORMAT "), termHistory.n_entries=%u to %s:%s, timelineStartLsn=%X/%X",
 		   sk->greetResponse.nodeId, msg.term, LSN_FORMAT_ARGS(msg.startStreamingAt), lastCommonTerm, msg.termHistory->n_entries, sk->host, sk->port, LSN_FORMAT_ARGS(msg.timelineStartLsn));
@@ -1638,7 +1641,7 @@ UpdateDonorShmem(WalProposer *wp)
 * Process AppendResponse message from safekeeper.
 */
 static void
-HandleSafekeeperResponse(WalProposer *wp, Safekeeper *fromsk)
+HandleSafekeeperResponse(WalProposer *wp, Safekeeper *sk)
 {
 	XLogRecPtr	candidateTruncateLsn;
 	XLogRecPtr	newCommitLsn;
@@ -1657,7 +1660,7 @@ HandleSafekeeperResponse(WalProposer *wp, Safekeeper *fromsk)
 	 * and WAL is committed by the quorum. BroadcastAppendRequest() should be
 	 * called to notify safekeepers about the new commitLsn.
 	 */
-	wp->api.process_safekeeper_feedback(wp, fromsk);
+	wp->api.process_safekeeper_feedback(wp, sk);

 	/*
 	 * Try to advance truncateLsn -- the last record flushed to all
--- a/pgxn/neon/walproposer.h
+++ b/pgxn/neon/walproposer.h
@@ -725,7 +725,7 @@ extern void WalProposerBroadcast(WalProposer *wp, XLogRecPtr startpos, XLogRecPt
 extern void WalProposerPoll(WalProposer *wp);
 extern void WalProposerFree(WalProposer *wp);

-extern WalproposerShmemState *GetWalpropShmemState(void);
+extern WalproposerShmemState *GetWalpropShmemState();

 /*
 * WaitEventSet API doesn't allow to remove socket, so walproposer_pg uses it to
@@ -745,7 +745,7 @@ extern TimeLineID walprop_pg_get_timeline_id(void);
 * catch logging.
 */
 #ifdef WALPROPOSER_LIB
-extern void WalProposerLibLog(WalProposer *wp, int elevel, char *fmt,...) pg_attribute_printf(3, 4);
+extern void WalProposerLibLog(WalProposer *wp, int elevel, char *fmt,...);
 #define wp_log(elevel, fmt, ...) WalProposerLibLog(wp, elevel, fmt, ## __VA_ARGS__)
 #else
 #define wp_log(elevel, fmt, ...) elog(elevel, WP_LOG_PREFIX fmt, ## __VA_ARGS__)
--- a/pgxn/neon/walproposer_pg.c
+++ b/pgxn/neon/walproposer_pg.c
@@ -286,9 +286,6 @@ safekeepers_cmp(char *old, char *new)
 static void
 assign_neon_safekeepers(const char *newval, void *extra)
 {
-	char	   *newval_copy;
-	char	   *oldval;
-
 	if (!am_walproposer)
 		return;

@@ -298,8 +295,8 @@ assign_neon_safekeepers(const char *newval, void *extra)
 	}

 	/* Copy values because we will modify them in split_safekeepers_list() */
-	newval_copy = pstrdup(newval);
-	oldval = pstrdup(wal_acceptors_list);
+	char *newval_copy = pstrdup(newval);
+	char *oldval = pstrdup(wal_acceptors_list);

 	/* 
 	 * TODO: restarting through FATAL is stupid and introduces 1s delay before
@@ -541,7 +538,7 @@ nwp_shmem_startup_hook(void)
 }

 WalproposerShmemState *
-GetWalpropShmemState(void)
+GetWalpropShmemState()
 {
 	Assert(walprop_shared != NULL);
 	return walprop_shared;
--- a/pgxn/neon_rmgr/neon_rmgr_desc.c
+++ b/pgxn/neon_rmgr/neon_rmgr_desc.c
@@ -44,6 +44,27 @@ infobits_desc(StringInfo buf, uint8 infobits, const char *keyname)
 	appendStringInfoString(buf, "]");
 }

+static void
+truncate_flags_desc(StringInfo buf, uint8 flags)
+{
+	appendStringInfoString(buf, "flags: [");
+
+	if (flags & XLH_TRUNCATE_CASCADE)
+		appendStringInfoString(buf, "CASCADE, ");
+	if (flags & XLH_TRUNCATE_RESTART_SEQS)
+		appendStringInfoString(buf, "RESTART_SEQS, ");
+
+	if (buf->data[buf->len - 1] == ' ')
+	{
+		/* Truncate-away final unneeded ", "  */
+		Assert(buf->data[buf->len - 2] == ',');
+		buf->len -= 2;
+		buf->data[buf->len] = '\0';
+	}
+
+	appendStringInfoString(buf, "]");
+}
+
 void
 neon_rm_desc(StringInfo buf, XLogReaderState *record)
 {
--- a/pgxn/neon_walredo/walredoproc.c
+++ b/pgxn/neon_walredo/walredoproc.c
@@ -136,7 +136,7 @@ static bool redo_block_filter(XLogReaderState *record, uint8 block_id);
 static void GetPage(StringInfo input_message);
 static void Ping(StringInfo input_message);
 static ssize_t buffered_read(void *buf, size_t count);
-static void CreateFakeSharedMemoryAndSemaphores(void);
+static void CreateFakeSharedMemoryAndSemaphores();

 static BufferTag target_redo_tag;

@@ -170,40 +170,6 @@ close_range_syscall(unsigned int start_fd, unsigned int count, unsigned int flag
    return syscall(__NR_close_range, start_fd, count, flags);
 }

-
-static PgSeccompRule allowed_syscalls[] =
-{
-	/* Hard requirements */
-	PG_SCMP_ALLOW(exit_group),
-	PG_SCMP_ALLOW(pselect6),
-	PG_SCMP_ALLOW(read),
-	PG_SCMP_ALLOW(select),
-	PG_SCMP_ALLOW(write),
-
-	/* Memory allocation */
-	PG_SCMP_ALLOW(brk),
-#ifndef MALLOC_NO_MMAP
-	/* TODO: musl doesn't have mallopt */
-	PG_SCMP_ALLOW(mmap),
-	PG_SCMP_ALLOW(munmap),
-#endif
-	/*
-	 * getpid() is called on assertion failure, in ExceptionalCondition.
-	 * It's not really needed, but seems pointless to hide it either. The
-	 * system call unlikely to expose a kernel vulnerability, and the PID
-	 * is stored in MyProcPid anyway.
-	 */
-	PG_SCMP_ALLOW(getpid),
-
-	/* Enable those for a proper shutdown. */
-#if 0
-	   PG_SCMP_ALLOW(munmap),
-	   PG_SCMP_ALLOW(shmctl),
-	   PG_SCMP_ALLOW(shmdt),
-	   PG_SCMP_ALLOW(unlink),	/* shm_unlink */
-#endif
-};
-
 static void
 enter_seccomp_mode(void)
 {
@@ -217,12 +183,44 @@ enter_seccomp_mode(void)
 				(errcode(ERRCODE_SYSTEM_ERROR),
 				 errmsg("seccomp: could not close files >= fd 3")));

+	PgSeccompRule syscalls[] =
+	{
+		/* Hard requirements */
+		PG_SCMP_ALLOW(exit_group),
+		PG_SCMP_ALLOW(pselect6),
+		PG_SCMP_ALLOW(read),
+		PG_SCMP_ALLOW(select),
+		PG_SCMP_ALLOW(write),
+
+		/* Memory allocation */
+		PG_SCMP_ALLOW(brk),
+#ifndef MALLOC_NO_MMAP
+		/* TODO: musl doesn't have mallopt */
+		PG_SCMP_ALLOW(mmap),
+		PG_SCMP_ALLOW(munmap),
+#endif
+		/*
+		 * getpid() is called on assertion failure, in ExceptionalCondition.
+		 * It's not really needed, but seems pointless to hide it either. The
+		 * system call unlikely to expose a kernel vulnerability, and the PID
+		 * is stored in MyProcPid anyway.
+		 */
+		PG_SCMP_ALLOW(getpid),
+
+		/* Enable those for a proper shutdown.
+		PG_SCMP_ALLOW(munmap),
+		PG_SCMP_ALLOW(shmctl),
+		PG_SCMP_ALLOW(shmdt),
+		PG_SCMP_ALLOW(unlink), // shm_unlink
+	 */
+	};
+
 #ifdef MALLOC_NO_MMAP
 	/* Ask glibc not to use mmap() */
 	mallopt(M_MMAP_MAX, 0);
 #endif

-	seccomp_load_rules(allowed_syscalls, lengthof(allowed_syscalls));
+	seccomp_load_rules(syscalls, lengthof(syscalls));
 }
 #endif /* HAVE_LIBSECCOMP */

@@ -451,7 +449,7 @@ WalRedoMain(int argc, char *argv[])
 * half-initialized postgres.
 */
 static void
-CreateFakeSharedMemoryAndSemaphores(void)
+CreateFakeSharedMemoryAndSemaphores()
 {
 	PGShmemHeader *shim = NULL;
 	PGShmemHeader *hdr;
@@ -994,7 +992,7 @@ redo_block_filter(XLogReaderState *record, uint8 block_id)
 	 * 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);
+	return !BUFFERTAGS_EQUAL(target_tag, target_redo_tag);
 }

 /*
--- a/pre-commit.py
+++ b/pre-commit.py
@@ -1,12 +1,11 @@
 #!/usr/bin/env python3

-from __future__ import annotations
-
 import argparse
 import enum
 import os
 import subprocess
 import sys
+from typing import List


@enum.unique
@@ -56,12 +55,12 @@ def mypy() -> str:
    return "poetry run mypy"


-def get_commit_files() -> list[str]:
+def get_commit_files() -> List[str]:
    files = subprocess.check_output("git diff --cached --name-only --diff-filter=ACM".split())
    return files.decode().splitlines()


-def check(name: str, suffix: str, cmd: str, changed_files: list[str], no_color: bool = False):
+def check(name: str, suffix: str, cmd: str, changed_files: List[str], no_color: bool = False):
    print(f"Checking: {name} ", end="")
    applicable_files = list(filter(lambda fname: fname.strip().endswith(suffix), changed_files))
    if not applicable_files:
--- a/proxy/src/auth/mod.rs
+++ b/proxy/src/auth/mod.rs
@@ -18,7 +18,7 @@ pub(crate) use flow::*;
 use tokio::time::error::Elapsed;

 use crate::{
-    control_plane,
+    console,
    error::{ReportableError, UserFacingError},
 };
 use std::{io, net::IpAddr};
@@ -34,7 +34,7 @@ pub(crate) enum AuthErrorImpl {
    Web(#[from] backend::WebAuthError),

    #[error(transparent)]
-    GetAuthInfo(#[from] control_plane::errors::GetAuthInfoError),
+    GetAuthInfo(#[from] console::errors::GetAuthInfoError),

    /// SASL protocol errors (includes [SCRAM](crate::scram)).
    #[error(transparent)]
--- a/proxy/src/auth/backend/mod.rs
+++ b/proxy/src/auth/backend/mod.rs
@@ -1,27 +1,27 @@
 mod classic;
-mod console_redirect;
 mod hacks;
 pub mod jwt;
 pub mod local;
+mod web;

 use std::net::IpAddr;
 use std::sync::Arc;
 use std::time::Duration;

-pub(crate) use console_redirect::WebAuthError;
 use ipnet::{Ipv4Net, Ipv6Net};
 use local::LocalBackend;
 use tokio::io::{AsyncRead, AsyncWrite};
 use tokio_postgres::config::AuthKeys;
 use tracing::{info, warn};
+pub(crate) use web::WebAuthError;

 use crate::auth::credentials::check_peer_addr_is_in_list;
 use crate::auth::{validate_password_and_exchange, AuthError};
 use crate::cache::Cached;
+use crate::console::errors::GetAuthInfoError;
+use crate::console::provider::{CachedRoleSecret, ConsoleBackend};
+use crate::console::{AuthSecret, NodeInfo};
 use crate::context::RequestMonitoring;
-use crate::control_plane::errors::GetAuthInfoError;
-use crate::control_plane::provider::{CachedRoleSecret, ControlPlaneBackend};
-use crate::control_plane::{AuthSecret, NodeInfo};
 use crate::intern::EndpointIdInt;
 use crate::metrics::Metrics;
 use crate::proxy::connect_compute::ComputeConnectBackend;
@@ -31,7 +31,7 @@ use crate::stream::Stream;
 use crate::{
    auth::{self, ComputeUserInfoMaybeEndpoint},
    config::AuthenticationConfig,
-    control_plane::{
+    console::{
        self,
        provider::{CachedAllowedIps, CachedNodeInfo},
        Api,
@@ -67,19 +67,19 @@ impl<T> std::ops::Deref for MaybeOwned<'_, T> {
 ///   backends which require them for the authentication process.
 pub enum Backend<'a, T, D> {
    /// Cloud API (V2).
-    ControlPlane(MaybeOwned<'a, ControlPlaneBackend>, T),
+    Console(MaybeOwned<'a, ConsoleBackend>, T),
    /// Authentication via a web browser.
-    ConsoleRedirect(MaybeOwned<'a, url::ApiUrl>, D),
+    Web(MaybeOwned<'a, url::ApiUrl>, D),
    /// Local proxy uses configured auth credentials and does not wake compute
    Local(MaybeOwned<'a, LocalBackend>),
 }

 #[cfg(test)]
 pub(crate) trait TestBackend: Send + Sync + 'static {
-    fn wake_compute(&self) -> Result<CachedNodeInfo, control_plane::errors::WakeComputeError>;
+    fn wake_compute(&self) -> Result<CachedNodeInfo, console::errors::WakeComputeError>;
    fn get_allowed_ips_and_secret(
        &self,
-    ) -> Result<(CachedAllowedIps, Option<CachedRoleSecret>), control_plane::errors::GetAuthInfoError>;
+    ) -> Result<(CachedAllowedIps, Option<CachedRoleSecret>), console::errors::GetAuthInfoError>;
    fn dyn_clone(&self) -> Box<dyn TestBackend>;
 }

@@ -93,23 +93,18 @@ impl Clone for Box<dyn TestBackend> {
 impl std::fmt::Display for Backend<'_, (), ()> {
    fn fmt(&self, fmt: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
-            Self::ControlPlane(api, ()) => match &**api {
-                ControlPlaneBackend::Management(endpoint) => fmt
-                    .debug_tuple("ControlPlane::Management")
-                    .field(&endpoint.url())
-                    .finish(),
+            Self::Console(api, ()) => match &**api {
+                ConsoleBackend::Console(endpoint) => {
+                    fmt.debug_tuple("Console").field(&endpoint.url()).finish()
+                }
                #[cfg(any(test, feature = "testing"))]
-                ControlPlaneBackend::PostgresMock(endpoint) => fmt
-                    .debug_tuple("ControlPlane::PostgresMock")
-                    .field(&endpoint.url())
-                    .finish(),
+                ConsoleBackend::Postgres(endpoint) => {
+                    fmt.debug_tuple("Postgres").field(&endpoint.url()).finish()
+                }
                #[cfg(test)]
-                ControlPlaneBackend::Test(_) => fmt.debug_tuple("ControlPlane::Test").finish(),
+                ConsoleBackend::Test(_) => fmt.debug_tuple("Test").finish(),
            },
-            Self::ConsoleRedirect(url, ()) => fmt
-                .debug_tuple("ConsoleRedirect")
-                .field(&url.as_str())
-                .finish(),
+            Self::Web(url, ()) => fmt.debug_tuple("Web").field(&url.as_str()).finish(),
            Self::Local(_) => fmt.debug_tuple("Local").finish(),
        }
    }
@@ -120,8 +115,8 @@ impl<T, D> Backend<'_, T, D> {
    /// This helps us pass structured config to async tasks.
    pub(crate) fn as_ref(&self) -> Backend<'_, &T, &D> {
        match self {
-            Self::ControlPlane(c, x) => Backend::ControlPlane(MaybeOwned::Borrowed(c), x),
-            Self::ConsoleRedirect(c, x) => Backend::ConsoleRedirect(MaybeOwned::Borrowed(c), x),
+            Self::Console(c, x) => Backend::Console(MaybeOwned::Borrowed(c), x),
+            Self::Web(c, x) => Backend::Web(MaybeOwned::Borrowed(c), x),
            Self::Local(l) => Backend::Local(MaybeOwned::Borrowed(l)),
        }
    }
@@ -133,8 +128,8 @@ impl<'a, T, D> Backend<'a, T, D> {
    /// a function to a contained value.
    pub(crate) fn map<R>(self, f: impl FnOnce(T) -> R) -> Backend<'a, R, D> {
        match self {
-            Self::ControlPlane(c, x) => Backend::ControlPlane(c, f(x)),
-            Self::ConsoleRedirect(c, x) => Backend::ConsoleRedirect(c, x),
+            Self::Console(c, x) => Backend::Console(c, f(x)),
+            Self::Web(c, x) => Backend::Web(c, x),
            Self::Local(l) => Backend::Local(l),
        }
    }
@@ -144,8 +139,8 @@ impl<'a, T, D, E> Backend<'a, Result<T, E>, D> {
    /// This is most useful for error handling.
    pub(crate) fn transpose(self) -> Result<Backend<'a, T, D>, E> {
        match self {
-            Self::ControlPlane(c, x) => x.map(|x| Backend::ControlPlane(c, x)),
-            Self::ConsoleRedirect(c, x) => Ok(Backend::ConsoleRedirect(c, x)),
+            Self::Console(c, x) => x.map(|x| Backend::Console(c, x)),
+            Self::Web(c, x) => Ok(Backend::Web(c, x)),
            Self::Local(l) => Ok(Backend::Local(l)),
        }
    }
@@ -295,7 +290,7 @@ impl AuthenticationConfig {
 /// All authentication flows will emit an AuthenticationOk message if successful.
 async fn auth_quirks(
    ctx: &RequestMonitoring,
-    api: &impl control_plane::Api,
+    api: &impl console::Api,
    user_info: ComputeUserInfoMaybeEndpoint,
    client: &mut stream::PqStream<Stream<impl AsyncRead + AsyncWrite + Unpin>>,
    allow_cleartext: bool,
@@ -417,8 +412,8 @@ impl<'a> Backend<'a, ComputeUserInfoMaybeEndpoint, &()> {
    /// Get username from the credentials.
    pub(crate) fn get_user(&self) -> &str {
        match self {
-            Self::ControlPlane(_, user_info) => &user_info.user,
-            Self::ConsoleRedirect(_, ()) => "web",
+            Self::Console(_, user_info) => &user_info.user,
+            Self::Web(_, ()) => "web",
            Self::Local(_) => "local",
        }
    }
@@ -434,7 +429,7 @@ impl<'a> Backend<'a, ComputeUserInfoMaybeEndpoint, &()> {
        endpoint_rate_limiter: Arc<EndpointRateLimiter>,
    ) -> auth::Result<Backend<'a, ComputeCredentials, NodeInfo>> {
        let res = match self {
-            Self::ControlPlane(api, user_info) => {
+            Self::Console(api, user_info) => {
                info!(
                    user = &*user_info.user,
                    project = user_info.endpoint(),
@@ -451,15 +446,15 @@ impl<'a> Backend<'a, ComputeUserInfoMaybeEndpoint, &()> {
                    endpoint_rate_limiter,
                )
                .await?;
-                Backend::ControlPlane(api, credentials)
+                Backend::Console(api, credentials)
            }
            // NOTE: this auth backend doesn't use client credentials.
-            Self::ConsoleRedirect(url, ()) => {
+            Self::Web(url, ()) => {
                info!("performing web authentication");

-                let info = console_redirect::authenticate(ctx, config, &url, client).await?;
+                let info = web::authenticate(ctx, config, &url, client).await?;

-                Backend::ConsoleRedirect(url, info)
+                Backend::Web(url, info)
            }
            Self::Local(_) => {
                return Err(auth::AuthError::bad_auth_method("invalid for local proxy"))
@@ -477,8 +472,8 @@ impl Backend<'_, ComputeUserInfo, &()> {
        ctx: &RequestMonitoring,
    ) -> Result<CachedRoleSecret, GetAuthInfoError> {
        match self {
-            Self::ControlPlane(api, user_info) => api.get_role_secret(ctx, user_info).await,
-            Self::ConsoleRedirect(_, ()) => Ok(Cached::new_uncached(None)),
+            Self::Console(api, user_info) => api.get_role_secret(ctx, user_info).await,
+            Self::Web(_, ()) => Ok(Cached::new_uncached(None)),
            Self::Local(_) => Ok(Cached::new_uncached(None)),
        }
    }
@@ -488,10 +483,8 @@ impl Backend<'_, ComputeUserInfo, &()> {
        ctx: &RequestMonitoring,
    ) -> Result<(CachedAllowedIps, Option<CachedRoleSecret>), GetAuthInfoError> {
        match self {
-            Self::ControlPlane(api, user_info) => {
-                api.get_allowed_ips_and_secret(ctx, user_info).await
-            }
-            Self::ConsoleRedirect(_, ()) => Ok((Cached::new_uncached(Arc::new(vec![])), None)),
+            Self::Console(api, user_info) => api.get_allowed_ips_and_secret(ctx, user_info).await,
+            Self::Web(_, ()) => Ok((Cached::new_uncached(Arc::new(vec![])), None)),
            Self::Local(_) => Ok((Cached::new_uncached(Arc::new(vec![])), None)),
        }
    }
@@ -502,18 +495,18 @@ impl ComputeConnectBackend for Backend<'_, ComputeCredentials, NodeInfo> {
    async fn wake_compute(
        &self,
        ctx: &RequestMonitoring,
-    ) -> Result<CachedNodeInfo, control_plane::errors::WakeComputeError> {
+    ) -> Result<CachedNodeInfo, console::errors::WakeComputeError> {
        match self {
-            Self::ControlPlane(api, creds) => api.wake_compute(ctx, &creds.info).await,
-            Self::ConsoleRedirect(_, info) => Ok(Cached::new_uncached(info.clone())),
+            Self::Console(api, creds) => api.wake_compute(ctx, &creds.info).await,
+            Self::Web(_, info) => Ok(Cached::new_uncached(info.clone())),
            Self::Local(local) => Ok(Cached::new_uncached(local.node_info.clone())),
        }
    }

    fn get_keys(&self) -> &ComputeCredentialKeys {
        match self {
-            Self::ControlPlane(_, creds) => &creds.keys,
-            Self::ConsoleRedirect(_, _) => &ComputeCredentialKeys::None,
+            Self::Console(_, creds) => &creds.keys,
+            Self::Web(_, _) => &ComputeCredentialKeys::None,
            Self::Local(_) => &ComputeCredentialKeys::None,
        }
    }
@@ -524,10 +517,10 @@ impl ComputeConnectBackend for Backend<'_, ComputeCredentials, &()> {
    async fn wake_compute(
        &self,
        ctx: &RequestMonitoring,
-    ) -> Result<CachedNodeInfo, control_plane::errors::WakeComputeError> {
+    ) -> Result<CachedNodeInfo, console::errors::WakeComputeError> {
        match self {
-            Self::ControlPlane(api, creds) => api.wake_compute(ctx, &creds.info).await,
-            Self::ConsoleRedirect(_, ()) => {
+            Self::Console(api, creds) => api.wake_compute(ctx, &creds.info).await,
+            Self::Web(_, ()) => {
                unreachable!("web auth flow doesn't support waking the compute")
            }
            Self::Local(local) => Ok(Cached::new_uncached(local.node_info.clone())),
@@ -536,8 +529,8 @@ impl ComputeConnectBackend for Backend<'_, ComputeCredentials, &()> {

    fn get_keys(&self) -> &ComputeCredentialKeys {
        match self {
-            Self::ControlPlane(_, creds) => &creds.keys,
-            Self::ConsoleRedirect(_, ()) => &ComputeCredentialKeys::None,
+            Self::Console(_, creds) => &creds.keys,
+            Self::Web(_, ()) => &ComputeCredentialKeys::None,
            Self::Local(_) => &ComputeCredentialKeys::None,
        }
    }
@@ -560,12 +553,12 @@ mod tests {
    use crate::{
        auth::{backend::MaskedIp, ComputeUserInfoMaybeEndpoint, IpPattern},
        config::AuthenticationConfig,
-        context::RequestMonitoring,
-        control_plane::{
+        console::{
            self,
            provider::{self, CachedAllowedIps, CachedRoleSecret},
            CachedNodeInfo,
        },
+        context::RequestMonitoring,
        proxy::NeonOptions,
        rate_limiter::{EndpointRateLimiter, RateBucketInfo},
        scram::{threadpool::ThreadPool, ServerSecret},
@@ -579,12 +572,12 @@ mod tests {
        secret: AuthSecret,
    }

-    impl control_plane::Api for Auth {
+    impl console::Api for Auth {
        async fn get_role_secret(
            &self,
            _ctx: &RequestMonitoring,
            _user_info: &super::ComputeUserInfo,
-        ) -> Result<CachedRoleSecret, control_plane::errors::GetAuthInfoError> {
+        ) -> Result<CachedRoleSecret, console::errors::GetAuthInfoError> {
            Ok(CachedRoleSecret::new_uncached(Some(self.secret.clone())))
        }

@@ -592,10 +585,8 @@ mod tests {
            &self,
            _ctx: &RequestMonitoring,
            _user_info: &super::ComputeUserInfo,
-        ) -> Result<
-            (CachedAllowedIps, Option<CachedRoleSecret>),
-            control_plane::errors::GetAuthInfoError,
-        > {
+        ) -> Result<(CachedAllowedIps, Option<CachedRoleSecret>), console::errors::GetAuthInfoError>
+        {
            Ok((
                CachedAllowedIps::new_uncached(Arc::new(self.ips.clone())),
                Some(CachedRoleSecret::new_uncached(Some(self.secret.clone()))),
@@ -614,7 +605,7 @@ mod tests {
            &self,
            _ctx: &RequestMonitoring,
            _user_info: &super::ComputeUserInfo,
-        ) -> Result<CachedNodeInfo, control_plane::errors::WakeComputeError> {
+        ) -> Result<CachedNodeInfo, console::errors::WakeComputeError> {
            unimplemented!()
        }
    }
--- a/proxy/src/auth/backend/classic.rs
+++ b/proxy/src/auth/backend/classic.rs
@@ -3,8 +3,8 @@ use crate::{
    auth::{self, backend::ComputeCredentialKeys, AuthFlow},
    compute,
    config::AuthenticationConfig,
+    console::AuthSecret,
    context::RequestMonitoring,
-    control_plane::AuthSecret,
    sasl,
    stream::{PqStream, Stream},
 };
--- a/proxy/src/auth/backend/hacks.rs
+++ b/proxy/src/auth/backend/hacks.rs
@@ -2,8 +2,8 @@ use super::{ComputeCredentials, ComputeUserInfo, ComputeUserInfoNoEndpoint};
 use crate::{
    auth::{self, AuthFlow},
    config::AuthenticationConfig,
+    console::AuthSecret,
    context::RequestMonitoring,
-    control_plane::AuthSecret,
    intern::EndpointIdInt,
    sasl,
    stream::{self, Stream},
--- a/proxy/src/auth/backend/local.rs
+++ b/proxy/src/auth/backend/local.rs
@@ -5,11 +5,11 @@ use arc_swap::ArcSwapOption;

 use crate::{
    compute::ConnCfg,
-    context::RequestMonitoring,
-    control_plane::{
+    console::{
        messages::{ColdStartInfo, EndpointJwksResponse, MetricsAuxInfo},
        NodeInfo,
    },
+    context::RequestMonitoring,
    intern::{BranchIdTag, EndpointIdTag, InternId, ProjectIdTag},
    EndpointId,
 };
--- a/proxy/src/auth/backend/console_redirect.rs
+++ b/proxy/src/auth/backend/console_redirect.rs
@@ -1,8 +1,8 @@
 use crate::{
    auth, compute,
    config::AuthenticationConfig,
+    console::{self, provider::NodeInfo},
    context::RequestMonitoring,
-    control_plane::{self, provider::NodeInfo},
    error::{ReportableError, UserFacingError},
    stream::PqStream,
    waiters,
@@ -70,7 +70,7 @@ pub(super) async fn authenticate(
    let (psql_session_id, waiter) = loop {
        let psql_session_id = new_psql_session_id();

-        match control_plane::mgmt::get_waiter(&psql_session_id) {
+        match console::mgmt::get_waiter(&psql_session_id) {
            Ok(waiter) => break (psql_session_id, waiter),
            Err(_e) => continue,
        }
--- a/proxy/src/auth/flow.rs
+++ b/proxy/src/auth/flow.rs
@@ -3,8 +3,8 @@
 use super::{backend::ComputeCredentialKeys, AuthErrorImpl, PasswordHackPayload};
 use crate::{
    config::TlsServerEndPoint,
+    console::AuthSecret,
    context::RequestMonitoring,
-    control_plane::AuthSecret,
    intern::EndpointIdInt,
    sasl,
    scram::{self, threadpool::ThreadPool},
--- a/proxy/src/bin/local_proxy.rs
+++ b/proxy/src/bin/local_proxy.rs
@@ -12,7 +12,7 @@ use proxy::{
    },
    cancellation::CancellationHandlerMain,
    config::{self, AuthenticationConfig, HttpConfig, ProxyConfig, RetryConfig},
-    control_plane::{
+    console::{
        locks::ApiLocks,
        messages::{EndpointJwksResponse, JwksSettings},
    },
@@ -305,7 +305,7 @@ async fn refresh_config_inner(path: &Utf8Path) -> anyhow::Result<()> {

    let mut jwks_set = vec![];

-    for jwks in data.jwks.into_iter().flatten() {
+    for jwks in data.jwks {
        let mut jwks_url = url::Url::from_str(&jwks.jwks_url).context("parsing JWKS url")?;

        ensure!(
--- a/proxy/src/bin/proxy.rs
+++ b/proxy/src/bin/proxy.rs
@@ -19,8 +19,8 @@ use proxy::config::CacheOptions;
 use proxy::config::HttpConfig;
 use proxy::config::ProjectInfoCacheOptions;
 use proxy::config::ProxyProtocolV2;
+use proxy::console;
 use proxy::context::parquet::ParquetUploadArgs;
-use proxy::control_plane;
 use proxy::http;
 use proxy::http::health_server::AppMetrics;
 use proxy::metrics::Metrics;
@@ -495,7 +495,7 @@ async fn main() -> anyhow::Result<()> {
            proxy: proxy::metrics::Metrics::get(),
        },
    ));
-    maintenance_tasks.spawn(control_plane::mgmt::task_main(mgmt_listener));
+    maintenance_tasks.spawn(console::mgmt::task_main(mgmt_listener));

    if let Some(metrics_config) = &config.metric_collection {
        // TODO: Add gc regardles of the metric collection being enabled.
@@ -506,8 +506,8 @@ async fn main() -> anyhow::Result<()> {
        ));
    }

-    if let auth::Backend::ControlPlane(api, _) = &config.auth_backend {
-        if let proxy::control_plane::provider::ControlPlaneBackend::Management(api) = &**api {
+    if let auth::Backend::Console(api, _) = &config.auth_backend {
+        if let proxy::console::provider::ConsoleBackend::Console(api) = &**api {
            match (redis_notifications_client, regional_redis_client.clone()) {
                (None, None) => {}
                (client1, client2) => {
@@ -623,7 +623,7 @@ fn build_config(args: &ProxyCliArgs) -> anyhow::Result<&'static ProxyConfig> {
                "Using AllowedIpsCache (wake_compute) with options={project_info_cache_config:?}"
            );
            info!("Using EndpointCacheConfig with options={endpoint_cache_config:?}");
-            let caches = Box::leak(Box::new(control_plane::caches::ApiCaches::new(
+            let caches = Box::leak(Box::new(console::caches::ApiCaches::new(
                wake_compute_cache_config,
                project_info_cache_config,
                endpoint_cache_config,
@@ -636,7 +636,7 @@ fn build_config(args: &ProxyCliArgs) -> anyhow::Result<&'static ProxyConfig> {
                timeout,
            } = args.wake_compute_lock.parse()?;
            info!(?limiter, shards, ?epoch, "Using NodeLocks (wake_compute)");
-            let locks = Box::leak(Box::new(control_plane::locks::ApiLocks::new(
+            let locks = Box::leak(Box::new(console::locks::ApiLocks::new(
                "wake_compute_lock",
                limiter,
                shards,
@@ -653,27 +653,27 @@ fn build_config(args: &ProxyCliArgs) -> anyhow::Result<&'static ProxyConfig> {
            RateBucketInfo::validate(&mut wake_compute_rps_limit)?;
            let wake_compute_endpoint_rate_limiter =
                Arc::new(WakeComputeRateLimiter::new(wake_compute_rps_limit));
-            let api = control_plane::provider::neon::Api::new(
+            let api = console::provider::neon::Api::new(
                endpoint,
                caches,
                locks,
                wake_compute_endpoint_rate_limiter,
            );
-            let api = control_plane::provider::ControlPlaneBackend::Management(api);
-            auth::Backend::ControlPlane(MaybeOwned::Owned(api), ())
+            let api = console::provider::ConsoleBackend::Console(api);
+            auth::Backend::Console(MaybeOwned::Owned(api), ())
        }

        AuthBackendType::Web => {
            let url = args.uri.parse()?;
-            auth::Backend::ConsoleRedirect(MaybeOwned::Owned(url), ())
+            auth::Backend::Web(MaybeOwned::Owned(url), ())
        }

        #[cfg(feature = "testing")]
        AuthBackendType::Postgres => {
            let url = args.auth_endpoint.parse()?;
-            let api = control_plane::provider::mock::Api::new(url, !args.is_private_access_proxy);
-            let api = control_plane::provider::ControlPlaneBackend::PostgresMock(api);
-            auth::Backend::ControlPlane(MaybeOwned::Owned(api), ())
+            let api = console::provider::mock::Api::new(url, !args.is_private_access_proxy);
+            let api = console::provider::ConsoleBackend::Postgres(api);
+            auth::Backend::Console(MaybeOwned::Owned(api), ())
        }
    };

@@ -689,7 +689,7 @@ fn build_config(args: &ProxyCliArgs) -> anyhow::Result<&'static ProxyConfig> {
        ?epoch,
        "Using NodeLocks (connect_compute)"
    );
-    let connect_compute_locks = control_plane::locks::ApiLocks::new(
+    let connect_compute_locks = console::locks::ApiLocks::new(
        "connect_compute_lock",
        limiter,
        shards,
--- a/proxy/src/cache/mod.rs
+++ b/proxy/src/cache/mod.rs
--- a/proxy/src/cache/project_info.rs
+++ b/proxy/src/cache/project_info.rs
@@ -16,7 +16,7 @@ use tracing::{debug, info};
 use crate::{
    auth::IpPattern,
    config::ProjectInfoCacheOptions,
-    control_plane::AuthSecret,
+    console::AuthSecret,
    intern::{EndpointIdInt, ProjectIdInt, RoleNameInt},
    EndpointId, RoleName,
 };
--- a/proxy/src/compute.rs
+++ b/proxy/src/compute.rs
@@ -1,8 +1,8 @@
 use crate::{
    auth::parse_endpoint_param,
    cancellation::CancelClosure,
+    console::{errors::WakeComputeError, messages::MetricsAuxInfo, provider::ApiLockError},
    context::RequestMonitoring,
-    control_plane::{errors::WakeComputeError, messages::MetricsAuxInfo, provider::ApiLockError},
    error::{ReportableError, UserFacingError},
    metrics::{Metrics, NumDbConnectionsGuard},
    proxy::neon_option,
--- a/proxy/src/config.rs
+++ b/proxy/src/config.rs
@@ -3,7 +3,7 @@ use crate::{
        self,
        backend::{jwt::JwkCache, AuthRateLimiter},
    },
-    control_plane::locks::ApiLocks,
+    console::locks::ApiLocks,
    rate_limiter::{RateBucketInfo, RateLimitAlgorithm, RateLimiterConfig},
    scram::threadpool::ThreadPool,
    serverless::{cancel_set::CancelSet, GlobalConnPoolOptions},
@@ -372,7 +372,7 @@ pub struct EndpointCacheConfig {
 }

 impl EndpointCacheConfig {
-    /// Default options for [`crate::control_plane::provider::NodeInfoCache`].
+    /// Default options for [`crate::console::provider::NodeInfoCache`].
    /// Notice that by default the limiter is empty, which means that cache is disabled.
    pub const CACHE_DEFAULT_OPTIONS: &'static str =
        "initial_batch_size=1000,default_batch_size=10,xread_timeout=5m,stream_name=controlPlane,disable_cache=true,limiter_info=1000@1s,retry_interval=1s";
@@ -447,7 +447,7 @@ pub struct CacheOptions {
 }

 impl CacheOptions {
-    /// Default options for [`crate::control_plane::provider::NodeInfoCache`].
+    /// Default options for [`crate::console::provider::NodeInfoCache`].
    pub const CACHE_DEFAULT_OPTIONS: &'static str = "size=4000,ttl=4m";

    /// Parse cache options passed via cmdline.
@@ -503,7 +503,7 @@ pub struct ProjectInfoCacheOptions {
 }

 impl ProjectInfoCacheOptions {
-    /// Default options for [`crate::control_plane::provider::NodeInfoCache`].
+    /// Default options for [`crate::console::provider::NodeInfoCache`].
    pub const CACHE_DEFAULT_OPTIONS: &'static str =
        "size=10000,ttl=4m,max_roles=10,gc_interval=60m";

@@ -622,9 +622,9 @@ pub struct ConcurrencyLockOptions {
 }

 impl ConcurrencyLockOptions {
-    /// Default options for [`crate::control_plane::provider::ApiLocks`].
+    /// Default options for [`crate::console::provider::ApiLocks`].
    pub const DEFAULT_OPTIONS_WAKE_COMPUTE_LOCK: &'static str = "permits=0";
-    /// Default options for [`crate::control_plane::provider::ApiLocks`].
+    /// Default options for [`crate::console::provider::ApiLocks`].
    pub const DEFAULT_OPTIONS_CONNECT_COMPUTE_LOCK: &'static str =
        "shards=64,permits=100,epoch=10m,timeout=10ms";

--- a/proxy/src/control_plane/mod.rs
+++ b/proxy/src/control_plane/mod.rs
--- a/proxy/src/control_plane/messages.rs
+++ b/proxy/src/control_plane/messages.rs
@@ -10,14 +10,14 @@ use crate::proxy::retry::CouldRetry;
 /// Generic error response with human-readable description.
 /// Note that we can't always present it to user as is.
 #[derive(Debug, Deserialize, Clone)]
-pub(crate) struct ControlPlaneError {
+pub(crate) struct ConsoleError {
    pub(crate) error: Box<str>,
    #[serde(skip)]
    pub(crate) http_status_code: http::StatusCode,
    pub(crate) status: Option<Status>,
 }

-impl ControlPlaneError {
+impl ConsoleError {
    pub(crate) fn get_reason(&self) -> Reason {
        self.status
            .as_ref()
@@ -51,7 +51,7 @@ impl ControlPlaneError {
    }
 }

-impl Display for ControlPlaneError {
+impl Display for ConsoleError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let msg: &str = self
            .status
@@ -62,7 +62,7 @@ impl Display for ControlPlaneError {
    }
 }

-impl CouldRetry for ControlPlaneError {
+impl CouldRetry for ConsoleError {
    fn could_retry(&self) -> bool {
        // If the error message does not have a status,
        // the error is unknown and probably should not retry automatically
--- a/proxy/src/control_plane/mgmt.rs
+++ b/proxy/src/control_plane/mgmt.rs
@@ -1,5 +1,5 @@
 use crate::{
-    control_plane::messages::{DatabaseInfo, KickSession},
+    console::messages::{DatabaseInfo, KickSession},
    waiters::{self, Waiter, Waiters},
 };
 use anyhow::Context;
--- a/proxy/src/control_plane/provider/mod.rs
+++ b/proxy/src/control_plane/provider/mod.rs
@@ -2,7 +2,7 @@
 pub mod mock;
 pub mod neon;

-use super::messages::{ControlPlaneError, MetricsAuxInfo};
+use super::messages::{ConsoleError, MetricsAuxInfo};
 use crate::{
    auth::{
        backend::{
@@ -28,7 +28,7 @@ use tracing::info;

 pub(crate) mod errors {
    use crate::{
-        control_plane::messages::{self, ControlPlaneError, Reason},
+        console::messages::{self, ConsoleError, Reason},
        error::{io_error, ErrorKind, ReportableError, UserFacingError},
        proxy::retry::CouldRetry,
    };
@@ -44,7 +44,7 @@ pub(crate) mod errors {
    pub(crate) enum ApiError {
        /// Error returned by the console itself.
        #[error("{REQUEST_FAILED} with {0}")]
-        ControlPlane(ControlPlaneError),
+        Console(ConsoleError),

        /// Various IO errors like broken pipe or malformed payload.
        #[error("{REQUEST_FAILED}: {0}")]
@@ -55,7 +55,7 @@ pub(crate) mod errors {
        /// Returns HTTP status code if it's the reason for failure.
        pub(crate) fn get_reason(&self) -> messages::Reason {
            match self {
-                ApiError::ControlPlane(e) => e.get_reason(),
+                ApiError::Console(e) => e.get_reason(),
                ApiError::Transport(_) => messages::Reason::Unknown,
            }
        }
@@ -65,7 +65,7 @@ pub(crate) mod errors {
        fn to_string_client(&self) -> String {
            match self {
                // To minimize risks, only select errors are forwarded to users.
-                ApiError::ControlPlane(c) => c.get_user_facing_message(),
+                ApiError::Console(c) => c.get_user_facing_message(),
                ApiError::Transport(_) => REQUEST_FAILED.to_owned(),
            }
        }
@@ -74,7 +74,7 @@ pub(crate) mod errors {
    impl ReportableError for ApiError {
        fn get_error_kind(&self) -> crate::error::ErrorKind {
            match self {
-                ApiError::ControlPlane(e) => match e.get_reason() {
+                ApiError::Console(e) => match e.get_reason() {
                    Reason::RoleProtected => ErrorKind::User,
                    Reason::ResourceNotFound => ErrorKind::User,
                    Reason::ProjectNotFound => ErrorKind::User,
@@ -91,12 +91,12 @@ pub(crate) mod errors {
                    Reason::LockAlreadyTaken => ErrorKind::ControlPlane,
                    Reason::RunningOperations => ErrorKind::ControlPlane,
                    Reason::Unknown => match &e {
-                        ControlPlaneError {
+                        ConsoleError {
                            http_status_code:
                                http::StatusCode::NOT_FOUND | http::StatusCode::NOT_ACCEPTABLE,
                            ..
                        } => crate::error::ErrorKind::User,
-                        ControlPlaneError {
+                        ConsoleError {
                            http_status_code: http::StatusCode::UNPROCESSABLE_ENTITY,
                            error,
                            ..
@@ -105,7 +105,7 @@ pub(crate) mod errors {
                        {
                            crate::error::ErrorKind::User
                        }
-                        ControlPlaneError {
+                        ConsoleError {
                            http_status_code: http::StatusCode::LOCKED,
                            error,
                            ..
@@ -114,11 +114,11 @@ pub(crate) mod errors {
                        {
                            crate::error::ErrorKind::User
                        }
-                        ControlPlaneError {
+                        ConsoleError {
                            http_status_code: http::StatusCode::TOO_MANY_REQUESTS,
                            ..
                        } => crate::error::ErrorKind::ServiceRateLimit,
-                        ControlPlaneError { .. } => crate::error::ErrorKind::ControlPlane,
+                        ConsoleError { .. } => crate::error::ErrorKind::ControlPlane,
                    },
                },
                ApiError::Transport(_) => crate::error::ErrorKind::ControlPlane,
@@ -131,7 +131,7 @@ pub(crate) mod errors {
            match self {
                // retry some transport errors
                Self::Transport(io) => io.could_retry(),
-                Self::ControlPlane(e) => e.could_retry(),
+                Self::Console(e) => e.could_retry(),
            }
        }
    }
@@ -314,8 +314,7 @@ impl NodeInfo {
    }
 }

-pub(crate) type NodeInfoCache =
-    TimedLru<EndpointCacheKey, Result<NodeInfo, Box<ControlPlaneError>>>;
+pub(crate) type NodeInfoCache = TimedLru<EndpointCacheKey, Result<NodeInfo, Box<ConsoleError>>>;
 pub(crate) type CachedNodeInfo = Cached<&'static NodeInfoCache, NodeInfo>;
 pub(crate) type CachedRoleSecret = Cached<&'static ProjectInfoCacheImpl, Option<AuthSecret>>;
 pub(crate) type CachedAllowedIps = Cached<&'static ProjectInfoCacheImpl, Arc<Vec<IpPattern>>>;
@@ -354,28 +353,28 @@ pub(crate) trait Api {

 #[non_exhaustive]
 #[derive(Clone)]
-pub enum ControlPlaneBackend {
-    /// Current Management API (V2).
-    Management(neon::Api),
-    /// Local mock control plane.
+pub enum ConsoleBackend {
+    /// Current Cloud API (V2).
+    Console(neon::Api),
+    /// Local mock of Cloud API (V2).
    #[cfg(any(test, feature = "testing"))]
-    PostgresMock(mock::Api),
+    Postgres(mock::Api),
    /// Internal testing
    #[cfg(test)]
    #[allow(private_interfaces)]
    Test(Box<dyn crate::auth::backend::TestBackend>),
 }

-impl Api for ControlPlaneBackend {
+impl Api for ConsoleBackend {
    async fn get_role_secret(
        &self,
        ctx: &RequestMonitoring,
        user_info: &ComputeUserInfo,
    ) -> Result<CachedRoleSecret, errors::GetAuthInfoError> {
        match self {
-            Self::Management(api) => api.get_role_secret(ctx, user_info).await,
+            Self::Console(api) => api.get_role_secret(ctx, user_info).await,
            #[cfg(any(test, feature = "testing"))]
-            Self::PostgresMock(api) => api.get_role_secret(ctx, user_info).await,
+            Self::Postgres(api) => api.get_role_secret(ctx, user_info).await,
            #[cfg(test)]
            Self::Test(_) => {
                unreachable!("this function should never be called in the test backend")
@@ -389,9 +388,9 @@ impl Api for ControlPlaneBackend {
        user_info: &ComputeUserInfo,
    ) -> Result<(CachedAllowedIps, Option<CachedRoleSecret>), errors::GetAuthInfoError> {
        match self {
-            Self::Management(api) => api.get_allowed_ips_and_secret(ctx, user_info).await,
+            Self::Console(api) => api.get_allowed_ips_and_secret(ctx, user_info).await,
            #[cfg(any(test, feature = "testing"))]
-            Self::PostgresMock(api) => api.get_allowed_ips_and_secret(ctx, user_info).await,
+            Self::Postgres(api) => api.get_allowed_ips_and_secret(ctx, user_info).await,
            #[cfg(test)]
            Self::Test(api) => api.get_allowed_ips_and_secret(),
        }
@@ -403,9 +402,9 @@ impl Api for ControlPlaneBackend {
        endpoint: EndpointId,
    ) -> anyhow::Result<Vec<AuthRule>> {
        match self {
-            Self::Management(api) => api.get_endpoint_jwks(ctx, endpoint).await,
+            Self::Console(api) => api.get_endpoint_jwks(ctx, endpoint).await,
            #[cfg(any(test, feature = "testing"))]
-            Self::PostgresMock(api) => api.get_endpoint_jwks(ctx, endpoint).await,
+            Self::Postgres(api) => api.get_endpoint_jwks(ctx, endpoint).await,
            #[cfg(test)]
            Self::Test(_api) => Ok(vec![]),
        }
@@ -417,16 +416,16 @@ impl Api for ControlPlaneBackend {
        user_info: &ComputeUserInfo,
    ) -> Result<CachedNodeInfo, errors::WakeComputeError> {
        match self {
-            Self::Management(api) => api.wake_compute(ctx, user_info).await,
+            Self::Console(api) => api.wake_compute(ctx, user_info).await,
            #[cfg(any(test, feature = "testing"))]
-            Self::PostgresMock(api) => api.wake_compute(ctx, user_info).await,
+            Self::Postgres(api) => api.wake_compute(ctx, user_info).await,
            #[cfg(test)]
            Self::Test(api) => api.wake_compute(),
        }
    }
 }

-/// Various caches for [`control_plane`](super).
+/// Various caches for [`console`](super).
 pub struct ApiCaches {
    /// Cache for the `wake_compute` API method.
    pub(crate) node_info: NodeInfoCache,
@@ -455,7 +454,7 @@ impl ApiCaches {
    }
 }

-/// Various caches for [`control_plane`](super).
+/// Various caches for [`console`](super).
 pub struct ApiLocks<K> {
    name: &'static str,
    node_locks: DashMap<K, Arc<DynamicLimiter>>,
@@ -578,7 +577,7 @@ impl WakeComputePermit {
    }
 }

-impl FetchAuthRules for ControlPlaneBackend {
+impl FetchAuthRules for ConsoleBackend {
    async fn fetch_auth_rules(
        &self,
        ctx: &RequestMonitoring,
--- a/proxy/src/control_plane/provider/mock.rs
+++ b/proxy/src/control_plane/provider/mock.rs
@@ -10,7 +10,7 @@ use crate::{
 use crate::{auth::backend::ComputeUserInfo, compute, error::io_error, scram, url::ApiUrl};
 use crate::{auth::IpPattern, cache::Cached};
 use crate::{
-    control_plane::{
+    console::{
        messages::MetricsAuxInfo,
        provider::{CachedAllowedIps, CachedRoleSecret},
    },
@@ -166,7 +166,7 @@ impl Api {
                endpoint_id: (&EndpointId::from("endpoint")).into(),
                project_id: (&ProjectId::from("project")).into(),
                branch_id: (&BranchId::from("branch")).into(),
-                cold_start_info: crate::control_plane::messages::ColdStartInfo::Warm,
+                cold_start_info: crate::console::messages::ColdStartInfo::Warm,
            },
            allow_self_signed_compute: false,
        };
--- a/proxy/src/control_plane/provider/neon.rs
+++ b/proxy/src/control_plane/provider/neon.rs
@@ -1,7 +1,7 @@
 //! Production console backend.

 use super::{
-    super::messages::{ControlPlaneError, GetRoleSecret, WakeCompute},
+    super::messages::{ConsoleError, GetRoleSecret, WakeCompute},
    errors::{ApiError, GetAuthInfoError, WakeComputeError},
    ApiCaches, ApiLocks, AuthInfo, AuthSecret, CachedAllowedIps, CachedNodeInfo, CachedRoleSecret,
    NodeInfo,
@@ -9,7 +9,7 @@ use super::{
 use crate::{
    auth::backend::{jwt::AuthRule, ComputeUserInfo},
    compute,
-    control_plane::messages::{ColdStartInfo, EndpointJwksResponse, Reason},
+    console::messages::{ColdStartInfo, EndpointJwksResponse, Reason},
    http,
    metrics::{CacheOutcome, Metrics},
    rate_limiter::WakeComputeRateLimiter,
@@ -348,7 +348,7 @@ impl super::Api for Api {
                    let (cached, info) = cached.take_value();
                    let info = info.map_err(|c| {
                        info!(key = &*key, "found cached wake_compute error");
-                        WakeComputeError::ApiError(ApiError::ControlPlane(*c))
+                        WakeComputeError::ApiError(ApiError::Console(*c))
                    })?;

                    debug!(key = &*key, "found cached compute node info");
@@ -395,9 +395,9 @@ impl super::Api for Api {
                Ok(cached.map(|()| node))
            }
            Err(err) => match err {
-                WakeComputeError::ApiError(ApiError::ControlPlane(err)) => {
+                WakeComputeError::ApiError(ApiError::Console(err)) => {
                    let Some(status) = &err.status else {
-                        return Err(WakeComputeError::ApiError(ApiError::ControlPlane(err)));
+                        return Err(WakeComputeError::ApiError(ApiError::Console(err)));
                    };

                    let reason = status
@@ -407,7 +407,7 @@ impl super::Api for Api {

                    // if we can retry this error, do not cache it.
                    if reason.can_retry() {
-                        return Err(WakeComputeError::ApiError(ApiError::ControlPlane(err)));
+                        return Err(WakeComputeError::ApiError(ApiError::Console(err)));
                    }

                    // at this point, we should only have quota errors.
@@ -422,7 +422,7 @@ impl super::Api for Api {
                        Duration::from_secs(30),
                    );

-                    Err(WakeComputeError::ApiError(ApiError::ControlPlane(err)))
+                    Err(WakeComputeError::ApiError(ApiError::Console(err)))
                }
                err => return Err(err),
            },
@@ -448,7 +448,7 @@ async fn parse_body<T: for<'a> serde::Deserialize<'a>>(
    // as the fact that the request itself has failed.
    let mut body = serde_json::from_slice(&s).unwrap_or_else(|e| {
        warn!("failed to parse error body: {e}");
-        ControlPlaneError {
+        ConsoleError {
            error: "reason unclear (malformed error message)".into(),
            http_status_code: status,
            status: None,
@@ -457,7 +457,7 @@ async fn parse_body<T: for<'a> serde::Deserialize<'a>>(
    body.http_status_code = status;

    error!("console responded with an error ({status}): {body:?}");
-    Err(ApiError::ControlPlane(body))
+    Err(ApiError::Console(body))
 }

 fn parse_host_port(input: &str) -> Option<(&str, u16)> {
--- a/proxy/src/context/mod.rs
+++ b/proxy/src/context/mod.rs
@@ -11,7 +11,7 @@ use try_lock::TryLock;
 use uuid::Uuid;

 use crate::{
-    control_plane::messages::{ColdStartInfo, MetricsAuxInfo},
+    console::messages::{ColdStartInfo, MetricsAuxInfo},
    error::ErrorKind,
    intern::{BranchIdInt, ProjectIdInt},
    metrics::{ConnectOutcome, InvalidEndpointsGroup, LatencyTimer, Metrics, Protocol, Waiting},
--- a/proxy/src/http/mod.rs
+++ b/proxy/src/http/mod.rs
--- a/proxy/src/lib.rs
+++ b/proxy/src/lib.rs
@@ -97,8 +97,8 @@ pub mod cache;
 pub mod cancellation;
 pub mod compute;
 pub mod config;
+pub mod console;
 pub mod context;
-pub mod control_plane;
 pub mod error;
 pub mod http;
 pub mod intern;
--- a/proxy/src/metrics.rs
+++ b/proxy/src/metrics.rs
@@ -11,7 +11,7 @@ use metrics::{CounterPairAssoc, CounterPairVec, HyperLogLog, HyperLogLogVec};

 use tokio::time::{self, Instant};

-use crate::control_plane::messages::ColdStartInfo;
+use crate::console::messages::ColdStartInfo;

 #[derive(MetricGroup)]
 #[metric(new(thread_pool: Arc<ThreadPoolMetrics>))]
--- a/proxy/src/proxy/mod.rs
+++ b/proxy/src/proxy/mod.rs
--- a/proxy/src/proxy/connect_compute.rs
+++ b/proxy/src/proxy/connect_compute.rs
@@ -3,8 +3,8 @@ use crate::{
    compute::COULD_NOT_CONNECT,
    compute::{self, PostgresConnection},
    config::RetryConfig,
+    console::{self, errors::WakeComputeError, locks::ApiLocks, CachedNodeInfo, NodeInfo},
    context::RequestMonitoring,
-    control_plane::{self, errors::WakeComputeError, locks::ApiLocks, CachedNodeInfo, NodeInfo},
    error::ReportableError,
    metrics::{ConnectOutcome, ConnectionFailureKind, Metrics, RetriesMetricGroup, RetryType},
    proxy::{
@@ -26,7 +26,7 @@ const CONNECT_TIMEOUT: time::Duration = time::Duration::from_secs(2);
 /// (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(crate) fn invalidate_cache(node_info: control_plane::CachedNodeInfo) -> NodeInfo {
+pub(crate) fn invalidate_cache(node_info: console::CachedNodeInfo) -> NodeInfo {
    let is_cached = node_info.cached();
    if is_cached {
        warn!("invalidating stalled compute node info cache entry");
@@ -49,7 +49,7 @@ pub(crate) trait ConnectMechanism {
    async fn connect_once(
        &self,
        ctx: &RequestMonitoring,
-        node_info: &control_plane::CachedNodeInfo,
+        node_info: &console::CachedNodeInfo,
        timeout: time::Duration,
    ) -> Result<Self::Connection, Self::ConnectError>;

@@ -61,7 +61,7 @@ pub(crate) trait ComputeConnectBackend {
    async fn wake_compute(
        &self,
        ctx: &RequestMonitoring,
-    ) -> Result<CachedNodeInfo, control_plane::errors::WakeComputeError>;
+    ) -> Result<CachedNodeInfo, console::errors::WakeComputeError>;

    fn get_keys(&self) -> &ComputeCredentialKeys;
 }
@@ -84,7 +84,7 @@ impl ConnectMechanism for TcpMechanism<'_> {
    async fn connect_once(
        &self,
        ctx: &RequestMonitoring,
-        node_info: &control_plane::CachedNodeInfo,
+        node_info: &console::CachedNodeInfo,
        timeout: time::Duration,
    ) -> Result<PostgresConnection, Self::Error> {
        let host = node_info.config.get_host()?;
--- a/proxy/src/proxy/passthrough.rs
+++ b/proxy/src/proxy/passthrough.rs
@@ -1,7 +1,7 @@
 use crate::{
    cancellation,
    compute::PostgresConnection,
-    control_plane::messages::MetricsAuxInfo,
+    console::messages::MetricsAuxInfo,
    metrics::{Direction, Metrics, NumClientConnectionsGuard, NumConnectionRequestsGuard},
    stream::Stream,
    usage_metrics::{Ids, MetricCounterRecorder, USAGE_METRICS},
--- a/proxy/src/proxy/tests/mod.rs
+++ b/proxy/src/proxy/tests/mod.rs
@@ -11,11 +11,9 @@ use crate::auth::backend::{
    ComputeCredentialKeys, ComputeCredentials, ComputeUserInfo, MaybeOwned, TestBackend,
 };
 use crate::config::{CertResolver, RetryConfig};
-use crate::control_plane::messages::{ControlPlaneError, Details, MetricsAuxInfo, Status};
-use crate::control_plane::provider::{
-    CachedAllowedIps, CachedRoleSecret, ControlPlaneBackend, NodeInfoCache,
-};
-use crate::control_plane::{self, CachedNodeInfo, NodeInfo};
+use crate::console::messages::{ConsoleError, Details, MetricsAuxInfo, Status};
+use crate::console::provider::{CachedAllowedIps, CachedRoleSecret, ConsoleBackend, NodeInfoCache};
+use crate::console::{self, CachedNodeInfo, NodeInfo};
 use crate::error::ErrorKind;
 use crate::{sasl, scram, BranchId, EndpointId, ProjectId};
 use anyhow::{bail, Context};
@@ -461,7 +459,7 @@ impl ConnectMechanism for TestConnectMechanism {
    async fn connect_once(
        &self,
        _ctx: &RequestMonitoring,
-        _node_info: &control_plane::CachedNodeInfo,
+        _node_info: &console::CachedNodeInfo,
        _timeout: std::time::Duration,
    ) -> Result<Self::Connection, Self::ConnectError> {
        let mut counter = self.counter.lock().unwrap();
@@ -485,23 +483,23 @@ impl ConnectMechanism for TestConnectMechanism {
 }

 impl TestBackend for TestConnectMechanism {
-    fn wake_compute(&self) -> Result<CachedNodeInfo, control_plane::errors::WakeComputeError> {
+    fn wake_compute(&self) -> Result<CachedNodeInfo, console::errors::WakeComputeError> {
        let mut counter = self.counter.lock().unwrap();
        let action = self.sequence[*counter];
        *counter += 1;
        match action {
            ConnectAction::Wake => Ok(helper_create_cached_node_info(self.cache)),
            ConnectAction::WakeFail => {
-                let err = control_plane::errors::ApiError::ControlPlane(ControlPlaneError {
+                let err = console::errors::ApiError::Console(ConsoleError {
                    http_status_code: StatusCode::BAD_REQUEST,
                    error: "TEST".into(),
                    status: None,
                });
                assert!(!err.could_retry());
-                Err(control_plane::errors::WakeComputeError::ApiError(err))
+                Err(console::errors::WakeComputeError::ApiError(err))
            }
            ConnectAction::WakeRetry => {
-                let err = control_plane::errors::ApiError::ControlPlane(ControlPlaneError {
+                let err = console::errors::ApiError::Console(ConsoleError {
                    http_status_code: StatusCode::BAD_REQUEST,
                    error: "TEST".into(),
                    status: Some(Status {
@@ -509,15 +507,13 @@ impl TestBackend for TestConnectMechanism {
                        message: "error".into(),
                        details: Details {
                            error_info: None,
-                            retry_info: Some(control_plane::messages::RetryInfo {
-                                retry_delay_ms: 1,
-                            }),
+                            retry_info: Some(console::messages::RetryInfo { retry_delay_ms: 1 }),
                            user_facing_message: None,
                        },
                    }),
                });
                assert!(err.could_retry());
-                Err(control_plane::errors::WakeComputeError::ApiError(err))
+                Err(console::errors::WakeComputeError::ApiError(err))
            }
            x => panic!("expecting action {x:?}, wake_compute is called instead"),
        }
@@ -525,7 +521,7 @@ impl TestBackend for TestConnectMechanism {

    fn get_allowed_ips_and_secret(
        &self,
-    ) -> Result<(CachedAllowedIps, Option<CachedRoleSecret>), control_plane::errors::GetAuthInfoError>
+    ) -> Result<(CachedAllowedIps, Option<CachedRoleSecret>), console::errors::GetAuthInfoError>
    {
        unimplemented!("not used in tests")
    }
@@ -542,7 +538,7 @@ fn helper_create_cached_node_info(cache: &'static NodeInfoCache) -> CachedNodeIn
            endpoint_id: (&EndpointId::from("endpoint")).into(),
            project_id: (&ProjectId::from("project")).into(),
            branch_id: (&BranchId::from("branch")).into(),
-            cold_start_info: crate::control_plane::messages::ColdStartInfo::Warm,
+            cold_start_info: crate::console::messages::ColdStartInfo::Warm,
        },
        allow_self_signed_compute: false,
    };
@@ -553,8 +549,8 @@ fn helper_create_cached_node_info(cache: &'static NodeInfoCache) -> CachedNodeIn
 fn helper_create_connect_info(
    mechanism: &TestConnectMechanism,
 ) -> auth::Backend<'static, ComputeCredentials, &()> {
-    let user_info = auth::Backend::ControlPlane(
-        MaybeOwned::Owned(ControlPlaneBackend::Test(Box::new(mechanism.clone()))),
+    let user_info = auth::Backend::Console(
+        MaybeOwned::Owned(ConsoleBackend::Test(Box::new(mechanism.clone()))),
        ComputeCredentials {
            info: ComputeUserInfo {
                endpoint: "endpoint".into(),
--- a/proxy/src/proxy/wake_compute.rs
+++ b/proxy/src/proxy/wake_compute.rs
@@ -1,7 +1,7 @@
 use crate::config::RetryConfig;
+use crate::console::messages::{ConsoleError, Reason};
+use crate::console::{errors::WakeComputeError, provider::CachedNodeInfo};
 use crate::context::RequestMonitoring;
-use crate::control_plane::messages::{ControlPlaneError, Reason};
-use crate::control_plane::{errors::WakeComputeError, provider::CachedNodeInfo};
 use crate::metrics::{
    ConnectOutcome, ConnectionFailuresBreakdownGroup, Metrics, RetriesMetricGroup, RetryType,
    WakeupFailureKind,
@@ -59,11 +59,11 @@ pub(crate) async fn wake_compute<B: ComputeConnectBackend>(
 }

 fn report_error(e: &WakeComputeError, retry: bool) {
-    use crate::control_plane::errors::ApiError;
+    use crate::console::errors::ApiError;
    let kind = match e {
        WakeComputeError::BadComputeAddress(_) => WakeupFailureKind::BadComputeAddress,
        WakeComputeError::ApiError(ApiError::Transport(_)) => WakeupFailureKind::ApiTransportError,
-        WakeComputeError::ApiError(ApiError::ControlPlane(e)) => match e.get_reason() {
+        WakeComputeError::ApiError(ApiError::Console(e)) => match e.get_reason() {
            Reason::RoleProtected => WakeupFailureKind::ApiConsoleBadRequest,
            Reason::ResourceNotFound => WakeupFailureKind::ApiConsoleBadRequest,
            Reason::ProjectNotFound => WakeupFailureKind::ApiConsoleBadRequest,
@@ -80,7 +80,7 @@ fn report_error(e: &WakeComputeError, retry: bool) {
            Reason::LockAlreadyTaken => WakeupFailureKind::ApiConsoleLocked,
            Reason::RunningOperations => WakeupFailureKind::ApiConsoleLocked,
            Reason::Unknown => match e {
-                ControlPlaneError {
+                ConsoleError {
                    http_status_code: StatusCode::LOCKED,
                    ref error,
                    ..
@@ -89,27 +89,27 @@ fn report_error(e: &WakeComputeError, retry: bool) {
                {
                    WakeupFailureKind::QuotaExceeded
                }
-                ControlPlaneError {
+                ConsoleError {
                    http_status_code: StatusCode::UNPROCESSABLE_ENTITY,
                    ref error,
                    ..
                } if error.contains("compute time quota of non-primary branches is exceeded") => {
                    WakeupFailureKind::QuotaExceeded
                }
-                ControlPlaneError {
+                ConsoleError {
                    http_status_code: StatusCode::LOCKED,
                    ..
                } => WakeupFailureKind::ApiConsoleLocked,
-                ControlPlaneError {
+                ConsoleError {
                    http_status_code: StatusCode::BAD_REQUEST,
                    ..
                } => WakeupFailureKind::ApiConsoleBadRequest,
-                ControlPlaneError {
+                ConsoleError {
                    http_status_code, ..
                } if http_status_code.is_server_error() => {
                    WakeupFailureKind::ApiConsoleOtherServerError
                }
-                ControlPlaneError { .. } => WakeupFailureKind::ApiConsoleOtherError,
+                ConsoleError { .. } => WakeupFailureKind::ApiConsoleOtherError,
            },
        },
        WakeComputeError::TooManyConnections => WakeupFailureKind::ApiConsoleLocked,
--- a/proxy/src/rate_limiter/mod.rs
+++ b/proxy/src/rate_limiter/mod.rs
--- a/proxy/src/redis/mod.rs
+++ b/proxy/src/redis/mod.rs
--- a/proxy/src/sasl/mod.rs
+++ b/proxy/src/sasl/mod.rs
--- a/proxy/src/scram/mod.rs
+++ b/proxy/src/scram/mod.rs
--- a/proxy/src/serverless/mod.rs
+++ b/proxy/src/serverless/mod.rs
--- a/proxy/src/serverless/backend.rs
+++ b/proxy/src/serverless/backend.rs
@@ -12,13 +12,13 @@ use crate::{
    },
    compute,
    config::{AuthenticationConfig, ProxyConfig},
-    context::RequestMonitoring,
-    control_plane::{
+    console::{
        errors::{GetAuthInfoError, WakeComputeError},
        locks::ApiLocks,
        provider::ApiLockError,
        CachedNodeInfo,
    },
+    context::RequestMonitoring,
    error::{ErrorKind, ReportableError, UserFacingError},
    intern::EndpointIdInt,
    proxy::{
@@ -114,7 +114,7 @@ impl PoolingBackend {
        jwt: String,
    ) -> Result<(), AuthError> {
        match &self.config.auth_backend {
-            crate::auth::Backend::ControlPlane(console, ()) => {
+            crate::auth::Backend::Console(console, ()) => {
                config
                    .jwks_cache
                    .check_jwt(
@@ -129,7 +129,7 @@ impl PoolingBackend {

                Ok(())
            }
-            crate::auth::Backend::ConsoleRedirect(_, ()) => Err(AuthError::auth_failed(
+            crate::auth::Backend::Web(_, ()) => Err(AuthError::auth_failed(
                "JWT login over web auth proxy is not supported",
            )),
            crate::auth::Backend::Local(_) => {
--- a/proxy/src/serverless/conn_pool.rs
+++ b/proxy/src/serverless/conn_pool.rs
@@ -17,7 +17,7 @@ use tokio_postgres::tls::NoTlsStream;
 use tokio_postgres::{AsyncMessage, ReadyForQueryStatus, Socket};
 use tokio_util::sync::CancellationToken;

-use crate::control_plane::messages::{ColdStartInfo, MetricsAuxInfo};
+use crate::console::messages::{ColdStartInfo, MetricsAuxInfo};
 use crate::metrics::{HttpEndpointPoolsGuard, Metrics};
 use crate::usage_metrics::{Ids, MetricCounter, USAGE_METRICS};
 use crate::{
@@ -760,7 +760,7 @@ mod tests {
                endpoint_id: (&EndpointId::from("endpoint")).into(),
                project_id: (&ProjectId::from("project")).into(),
                branch_id: (&BranchId::from("branch")).into(),
-                cold_start_info: crate::control_plane::messages::ColdStartInfo::Warm,
+                cold_start_info: crate::console::messages::ColdStartInfo::Warm,
            },
            conn_id: uuid::Uuid::new_v4(),
        }
--- a/proxy/src/serverless/http_conn_pool.rs
+++ b/proxy/src/serverless/http_conn_pool.rs
@@ -8,7 +8,7 @@ use std::sync::atomic::{self, AtomicUsize};
 use std::{sync::Arc, sync::Weak};
 use tokio::net::TcpStream;

-use crate::control_plane::messages::{ColdStartInfo, MetricsAuxInfo};
+use crate::console::messages::{ColdStartInfo, MetricsAuxInfo};
 use crate::metrics::{HttpEndpointPoolsGuard, Metrics};
 use crate::usage_metrics::{Ids, MetricCounter, USAGE_METRICS};
 use crate::{context::RequestMonitoring, EndpointCacheKey};
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -97,8 +97,5 @@ select = [
    "I", # isort
    "W", # pycodestyle
    "B", # bugbear
-    "UP", # pyupgrade
+    "UP032", # f-string
 ]
-
-[tool.ruff.lint.pyupgrade]
-keep-runtime-typing = true # Remove this stanza when we require Python 3.10
--- a/safekeeper/Cargo.toml
+++ b/safekeeper/Cargo.toml
@@ -23,7 +23,6 @@ crc32c.workspace = true
 fail.workspace = true
 hex.workspace = true
 humantime.workspace = true
-http.workspace = true
 hyper0.workspace = true
 futures.workspace = true
 once_cell.workspace = true
--- a/safekeeper/tests/walproposer_sim/safekeeper.rs
+++ b/safekeeper/tests/walproposer_sim/safekeeper.rs
@@ -13,7 +13,7 @@ use desim::{
    node_os::NodeOs,
    proto::{AnyMessage, NetEvent, NodeEvent},
 };
-use http::Uri;
+use hyper0::Uri;
 use safekeeper::{
    safekeeper::{ProposerAcceptorMessage, SafeKeeper, ServerInfo, UNKNOWN_SERVER_VERSION},
    state::{TimelinePersistentState, TimelineState},
--- a/scripts/benchmark_durations.py
+++ b/scripts/benchmark_durations.py
@@ -1,10 +1,9 @@
 #! /usr/bin/env python3

-from __future__ import annotations
-
 import argparse
 import json
 import logging
+from typing import Dict

 import psycopg2
 import psycopg2.extras
@@ -111,7 +110,7 @@ def main(args: argparse.Namespace):
    output = args.output
    percentile = args.percentile

-    res: dict[str, float] = {}
+    res: Dict[str, float] = {}

    try:
        logging.info("connecting to the database...")
--- a/scripts/download_basebackup.py
+++ b/scripts/download_basebackup.py
@@ -4,9 +4,6 @@
 #
 # This can be useful in disaster recovery.
 #
-
-from __future__ import annotations
-
 import argparse

 import psycopg2
--- a/scripts/flaky_tests.py
+++ b/scripts/flaky_tests.py
@@ -1,21 +1,16 @@
 #! /usr/bin/env python3

-from __future__ import annotations
-
 import argparse
 import json
 import logging
 import os
 from collections import defaultdict
-from typing import TYPE_CHECKING
+from typing import Any, DefaultDict, Dict, Optional

 import psycopg2
 import psycopg2.extras
 import toml

-if TYPE_CHECKING:
-    from typing import Any, Optional
-
 FLAKY_TESTS_QUERY = """
    SELECT
        DISTINCT parent_suite, suite, name
@@ -38,7 +33,7 @@ def main(args: argparse.Namespace):
    build_type = args.build_type
    pg_version = args.pg_version

-    res: defaultdict[str, defaultdict[str, dict[str, bool]]]
+    res: DefaultDict[str, DefaultDict[str, Dict[str, bool]]]
    res = defaultdict(lambda: defaultdict(dict))

    try:
@@ -65,7 +60,7 @@ def main(args: argparse.Namespace):
        pageserver_virtual_file_io_engine_parameter = ""

    # re-use existing records of flaky tests from before parametrization by compaction_algorithm
-    def get_pageserver_default_tenant_config_compaction_algorithm() -> Optional[dict[str, Any]]:
+    def get_pageserver_default_tenant_config_compaction_algorithm() -> Optional[Dict[str, Any]]:
        """Duplicated from parametrize.py"""
        toml_table = os.getenv("PAGESERVER_DEFAULT_TENANT_CONFIG_COMPACTION_ALGORITHM")
        if toml_table is None:
--- a/scripts/force_layer_download.py
+++ b/scripts/force_layer_download.py
@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 import argparse
 import asyncio
 import json
@@ -7,15 +5,11 @@ import logging
 import signal
 import sys
 from collections import defaultdict
-from collections.abc import Awaitable
 from dataclasses import dataclass
-from typing import TYPE_CHECKING
+from typing import Any, Awaitable, Dict, List, Tuple

 import aiohttp

-if TYPE_CHECKING:
-    from typing import Any
-

 class ClientException(Exception):
    pass
@@ -95,7 +89,7 @@ class Client:
 class Completed:
    """The status dict returned by the API"""

-    status: dict[str, Any]
+    status: Dict[str, Any]


 sigint_received = asyncio.Event()
@@ -185,7 +179,7 @@ async def main_impl(args, report_out, client: Client):
    """
    Returns OS exit status.
    """
-    tenant_and_timline_ids: list[tuple[str, str]] = []
+    tenant_and_timline_ids: List[Tuple[str, str]] = []
    # fill tenant_and_timline_ids based on spec
    for spec in args.what:
        comps = spec.split(":")
@@ -221,14 +215,14 @@ async def main_impl(args, report_out, client: Client):
    tenant_and_timline_ids = tmp

    logging.info("create tasks and process them at specified concurrency")
-    task_q: asyncio.Queue[tuple[str, Awaitable[Any]]] = asyncio.Queue()
+    task_q: asyncio.Queue[Tuple[str, Awaitable[Any]]] = asyncio.Queue()
    tasks = {
        f"{tid}:{tlid}": do_timeline(client, tid, tlid) for tid, tlid in tenant_and_timline_ids
    }
    for task in tasks.items():
        task_q.put_nowait(task)

-    result_q: asyncio.Queue[tuple[str, Any]] = asyncio.Queue()
+    result_q: asyncio.Queue[Tuple[str, Any]] = asyncio.Queue()
    taskq_handlers = []
    for _ in range(0, args.concurrent_tasks):
        taskq_handlers.append(taskq_handler(task_q, result_q))
--- a/scripts/ingest_perf_test_result.py
+++ b/scripts/ingest_perf_test_result.py
@@ -1,7 +1,4 @@
 #!/usr/bin/env python3
-
-from __future__ import annotations
-
 import argparse
 import json
 import logging
--- a/scripts/ingest_regress_test_result-new-format.py
+++ b/scripts/ingest_regress_test_result-new-format.py
@@ -1,7 +1,5 @@
 #! /usr/bin/env python3

-from __future__ import annotations
-
 import argparse
 import dataclasses
 import json
@@ -13,6 +11,7 @@ from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
+from typing import Tuple

 import backoff
 import psycopg2
@@ -92,7 +91,7 @@ def create_table(cur):
    cur.execute(CREATE_TABLE)


-def parse_test_name(test_name: str) -> tuple[str, int, str]:
+def parse_test_name(test_name: str) -> Tuple[str, int, str]:
    build_type, pg_version = None, None
    if match := TEST_NAME_RE.search(test_name):
        found = match.groupdict()
--- a/scripts/sk_cleanup_tenants/script.py
+++ b/scripts/sk_cleanup_tenants/script.py
@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 import argparse
 import logging
 import os
--- a/storage_broker/Cargo.toml
+++ b/storage_broker/Cargo.toml
@@ -10,16 +10,13 @@ bench = []
 [dependencies]
 anyhow.workspace = true
 async-stream.workspace = true
-bytes.workspace = true
 clap = { workspace = true, features = ["derive"] }
 const_format.workspace = true
 futures.workspace = true
 futures-core.workspace = true
 futures-util.workspace = true
 humantime.workspace = true
-hyper = { workspace = true, features = ["full"] }
-http-body-util.workspace = true
-hyper-util = "0.1"
+hyper0 = { workspace = true, features = ["full"] }
 once_cell.workspace = true
 parking_lot.workspace = true
 prost.workspace = true
--- a/storage_broker/src/bin/storage_broker.rs
+++ b/storage_broker/src/bin/storage_broker.rs
@@ -10,15 +10,16 @@
 //!
 //! Only safekeeper message is supported, but it is not hard to add something
 //! else with generics.
+
+extern crate hyper0 as hyper;
+
 use clap::{command, Parser};
 use futures_core::Stream;
 use futures_util::StreamExt;
-use http_body_util::Full;
-use hyper::body::Incoming;
 use hyper::header::CONTENT_TYPE;
-use hyper::service::service_fn;
-use hyper::{Method, StatusCode};
-use hyper_util::rt::{TokioExecutor, TokioIo, TokioTimer};
+use hyper::server::conn::AddrStream;
+use hyper::service::{make_service_fn, service_fn};
+use hyper::{Body, Method, StatusCode};
 use parking_lot::RwLock;
 use std::collections::HashMap;
 use std::convert::Infallible;
@@ -26,11 +27,9 @@ use std::net::SocketAddr;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::time::Duration;
-use tokio::net::TcpListener;
 use tokio::sync::broadcast;
 use tokio::sync::broadcast::error::RecvError;
 use tokio::time;
-use tonic::body::{self, empty_body, BoxBody};
 use tonic::codegen::Service;
 use tonic::transport::server::Connected;
 use tonic::Code;
@@ -49,7 +48,9 @@ use storage_broker::proto::{
    FilterTenantTimelineId, MessageType, SafekeeperDiscoveryRequest, SafekeeperDiscoveryResponse,
    SafekeeperTimelineInfo, SubscribeByFilterRequest, SubscribeSafekeeperInfoRequest, TypedMessage,
 };
-use storage_broker::{parse_proto_ttid, DEFAULT_KEEPALIVE_INTERVAL, DEFAULT_LISTEN_ADDR};
+use storage_broker::{
+    parse_proto_ttid, EitherBody, DEFAULT_KEEPALIVE_INTERVAL, DEFAULT_LISTEN_ADDR,
+};
 use utils::id::TenantTimelineId;
 use utils::logging::{self, LogFormat};
 use utils::sentry_init::init_sentry;
@@ -601,8 +602,8 @@ impl BrokerService for Broker {

 // We serve only metrics and healthcheck through http1.
 async fn http1_handler(
-    req: hyper::Request<Incoming>,
-) -> Result<hyper::Response<BoxBody>, Infallible> {
+    req: hyper::Request<hyper::body::Body>,
+) -> Result<hyper::Response<Body>, Infallible> {
    let resp = match (req.method(), req.uri().path()) {
        (&Method::GET, "/metrics") => {
            let mut buffer = vec![];
@@ -613,16 +614,16 @@ async fn http1_handler(
            hyper::Response::builder()
                .status(StatusCode::OK)
                .header(CONTENT_TYPE, encoder.format_type())
-                .body(body::boxed(Full::new(bytes::Bytes::from(buffer))))
+                .body(Body::from(buffer))
                .unwrap()
        }
        (&Method::GET, "/status") => hyper::Response::builder()
            .status(StatusCode::OK)
-            .body(empty_body())
+            .body(Body::empty())
            .unwrap(),
        _ => hyper::Response::builder()
            .status(StatusCode::NOT_FOUND)
-            .body(empty_body())
+            .body(Body::empty())
            .unwrap(),
    };
    Ok(resp)
@@ -664,76 +665,52 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
    };
    let storage_broker_server = BrokerServiceServer::new(storage_broker_impl);

+    info!("listening on {}", &args.listen_addr);
+
    // grpc is served along with http1 for metrics on a single port, hence we
    // don't use tonic's Server.
-    let tcp_listener = TcpListener::bind(&args.listen_addr).await?;
-    info!("listening on {}", &args.listen_addr);
-    loop {
-        let (stream, addr) = match tcp_listener.accept().await {
-            Ok(v) => v,
-            Err(e) => {
-                info!("couldn't accept connection: {e}");
-                continue;
-            }
-        };
+    hyper::Server::bind(&args.listen_addr)
+        .http2_keep_alive_interval(Some(args.http2_keepalive_interval))
+        .serve(make_service_fn(move |conn: &AddrStream| {
+            let storage_broker_server_cloned = storage_broker_server.clone();
+            let connect_info = conn.connect_info();
+            async move {
+                Ok::<_, Infallible>(service_fn(move |mut req| {
+                    // That's what tonic's MakeSvc.call does to pass conninfo to
+                    // the request handler (and where its request.remote_addr()
+                    // expects it to find).
+                    req.extensions_mut().insert(connect_info.clone());

-        let mut builder = hyper_util::server::conn::auto::Builder::new(TokioExecutor::new());
-        builder.http1().timer(TokioTimer::new());
-        builder
-            .http2()
-            .timer(TokioTimer::new())
-            .keep_alive_interval(Some(args.http2_keepalive_interval))
-            // This matches the tonic server default. It allows us to support production-like workloads.
-            .max_concurrent_streams(None);
-
-        let storage_broker_server_cloned = storage_broker_server.clone();
-        let connect_info = stream.connect_info();
-        let service_fn_ = async move {
-            service_fn(move |mut req| {
-                // That's what tonic's MakeSvc.call does to pass conninfo to
-                // the request handler (and where its request.remote_addr()
-                // expects it to find).
-                req.extensions_mut().insert(connect_info.clone());
-
-                // Technically this second clone is not needed, but consume
-                // by async block is apparently unavoidable. BTW, error
-                // message is enigmatic, see
-                // https://github.com/rust-lang/rust/issues/68119
-                //
-                // We could get away without async block at all, but then we
-                // need to resort to futures::Either to merge the result,
-                // which doesn't caress an eye as well.
-                let mut storage_broker_server_svc = storage_broker_server_cloned.clone();
-                async move {
-                    if req.headers().get("content-type").map(|x| x.as_bytes())
-                        == Some(b"application/grpc")
-                    {
-                        let res_resp = storage_broker_server_svc.call(req).await;
-                        // Grpc and http1 handlers have slightly different
-                        // Response types: it is UnsyncBoxBody for the
-                        // former one (not sure why) and plain hyper::Body
-                        // for the latter. Both implement HttpBody though,
-                        // and `Either` is used to merge them.
-                        res_resp.map(|resp| resp.map(http_body_util::Either::Left))
-                    } else {
-                        let res_resp = http1_handler(req).await;
-                        res_resp.map(|resp| resp.map(http_body_util::Either::Right))
+                    // Technically this second clone is not needed, but consume
+                    // by async block is apparently unavoidable. BTW, error
+                    // message is enigmatic, see
+                    // https://github.com/rust-lang/rust/issues/68119
+                    //
+                    // We could get away without async block at all, but then we
+                    // need to resort to futures::Either to merge the result,
+                    // which doesn't caress an eye as well.
+                    let mut storage_broker_server_svc = storage_broker_server_cloned.clone();
+                    async move {
+                        if req.headers().get("content-type").map(|x| x.as_bytes())
+                            == Some(b"application/grpc")
+                        {
+                            let res_resp = storage_broker_server_svc.call(req).await;
+                            // Grpc and http1 handlers have slightly different
+                            // Response types: it is UnsyncBoxBody for the
+                            // former one (not sure why) and plain hyper::Body
+                            // for the latter. Both implement HttpBody though,
+                            // and EitherBody is used to merge them.
+                            res_resp.map(|resp| resp.map(EitherBody::Left))
+                        } else {
+                            let res_resp = http1_handler(req).await;
+                            res_resp.map(|resp| resp.map(EitherBody::Right))
+                        }
                    }
-                }
-            })
-        }
-        .await;
-
-        tokio::task::spawn(async move {
-            let res = builder
-                .serve_connection(TokioIo::new(stream), service_fn_)
-                .await;
-
-            if let Err(e) = res {
-                info!("error serving connection from {addr}: {e}");
+                }))
            }
-        });
-    }
+        }))
+        .await?;
+    Ok(())
 }

 #[cfg(test)]
--- a/storage_broker/src/lib.rs
+++ b/storage_broker/src/lib.rs
@@ -1,3 +1,8 @@
+extern crate hyper0 as hyper;
+
+use hyper::body::HttpBody;
+use std::pin::Pin;
+use std::task::{Context, Poll};
 use std::time::Duration;
 use tonic::codegen::StdError;
 use tonic::transport::{ClientTlsConfig, Endpoint};
@@ -91,3 +96,56 @@ pub fn parse_proto_ttid(proto_ttid: &ProtoTenantTimelineId) -> Result<TenantTime
        timeline_id,
    })
 }
+
+// These several usages don't justify anyhow dependency, though it would work as
+// well.
+type AnyError = Box<dyn std::error::Error + Send + Sync + 'static>;
+
+// Provides impl HttpBody for two different types implementing it. Inspired by
+// https://github.com/hyperium/tonic/blob/master/examples/src/hyper_warp/server.rs
+pub enum EitherBody<A, B> {
+    Left(A),
+    Right(B),
+}
+
+impl<A, B> HttpBody for EitherBody<A, B>
+where
+    A: HttpBody + Send + Unpin,
+    B: HttpBody<Data = A::Data> + Send + Unpin,
+    A::Error: Into<AnyError>,
+    B::Error: Into<AnyError>,
+{
+    type Data = A::Data;
+    type Error = Box<dyn std::error::Error + Send + Sync + 'static>;
+
+    fn is_end_stream(&self) -> bool {
+        match self {
+            EitherBody::Left(b) => b.is_end_stream(),
+            EitherBody::Right(b) => b.is_end_stream(),
+        }
+    }
+
+    fn poll_data(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
+        match self.get_mut() {
+            EitherBody::Left(b) => Pin::new(b).poll_data(cx).map(map_option_err),
+            EitherBody::Right(b) => Pin::new(b).poll_data(cx).map(map_option_err),
+        }
+    }
+
+    fn poll_trailers(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<Option<hyper::HeaderMap>, Self::Error>> {
+        match self.get_mut() {
+            EitherBody::Left(b) => Pin::new(b).poll_trailers(cx).map_err(Into::into),
+            EitherBody::Right(b) => Pin::new(b).poll_trailers(cx).map_err(Into::into),
+        }
+    }
+}
+
+fn map_option_err<T, U: Into<AnyError>>(err: Option<Result<T, U>>) -> Option<Result<T, AnyError>> {
+    err.map(|e| e.map_err(Into::into))
+}
--- a/storage_controller/src/service.rs
+++ b/storage_controller/src/service.rs
@@ -966,8 +966,6 @@ impl Service {

            let res = self.heartbeater.heartbeat(nodes).await;
            if let Ok(deltas) = res {
-                let mut to_handle = Vec::default();
-
                for (node_id, state) in deltas.0 {
                    let new_availability = match state {
                        PageserverState::Available { utilization, .. } => {
@@ -999,27 +997,14 @@ impl Service {
                        }
                    };

-                    let node_lock = trace_exclusive_lock(
-                        &self.node_op_locks,
-                        node_id,
-                        NodeOperations::Configure,
-                    )
-                    .await;
-
                    // This is the code path for geniune availability transitions (i.e node
                    // goes unavailable and/or comes back online).
                    let res = self
-                        .node_state_configure(node_id, Some(new_availability), None, &node_lock)
+                        .node_configure(node_id, Some(new_availability), None)
                        .await;

                    match res {
-                        Ok(transition) => {
-                            // Keep hold of the lock until the availability transitions
-                            // have been handled in
-                            // [`Service::handle_node_availability_transitions`] in order avoid
-                            // racing with [`Service::external_node_configure`].
-                            to_handle.push((node_id, node_lock, transition));
-                        }
+                        Ok(()) => {}
                        Err(ApiError::NotFound(_)) => {
                            // This should be rare, but legitimate since the heartbeats are done
                            // on a snapshot of the nodes.
@@ -1029,37 +1014,13 @@ impl Service {
                            // Transition to active involves reconciling: if a node responds to a heartbeat then
                            // becomes unavailable again, we may get an error here.
                            tracing::error!(
-                                "Failed to update node state {} after heartbeat round: {}",
+                                "Failed to update node {} after heartbeat round: {}",
                                node_id,
                                err
                            );
                        }
                    }
                }
-
-                // We collected all the transitions above and now we handle them.
-                let res = self.handle_node_availability_transitions(to_handle).await;
-                if let Err(errs) = res {
-                    for (node_id, err) in errs {
-                        match err {
-                            ApiError::NotFound(_) => {
-                                // This should be rare, but legitimate since the heartbeats are done
-                                // on a snapshot of the nodes.
-                                tracing::info!(
-                                    "Node {} was not found after heartbeat round",
-                                    node_id
-                                );
-                            }
-                            err => {
-                                tracing::error!(
-                                    "Failed to handle availability transition for {} after heartbeat round: {}",
-                                    node_id,
-                                    err
-                                );
-                            }
-                        }
-                    }
-                }
            }
        }
    }
@@ -5338,17 +5299,15 @@ impl Service {
        Ok(())
    }

-    /// Configure in-memory and persistent state of a node as requested
-    ///
-    /// Note that this function does not trigger any immediate side effects in response
-    /// to the changes. That part is handled by [`Self::handle_node_availability_transition`].
-    async fn node_state_configure(
+    pub(crate) async fn node_configure(
        &self,
        node_id: NodeId,
        availability: Option<NodeAvailability>,
        scheduling: Option<NodeSchedulingPolicy>,
-        node_lock: &TracingExclusiveGuard<NodeOperations>,
-    ) -> Result<AvailabilityTransition, ApiError> {
+    ) -> Result<(), ApiError> {
+        let _node_lock =
+            trace_exclusive_lock(&self.node_op_locks, node_id, NodeOperations::Configure).await;
+
        if let Some(scheduling) = scheduling {
            // Scheduling is a persistent part of Node: we must write updates to the database before
            // applying them in memory
@@ -5377,7 +5336,7 @@ impl Service {
            };

            if matches!(availability_transition, AvailabilityTransition::ToActive) {
-                self.node_activate_reconcile(activate_node, node_lock)
+                self.node_activate_reconcile(activate_node, &_node_lock)
                    .await?;
            }
            availability_transition
@@ -5387,7 +5346,7 @@ impl Service {

        // Apply changes from the request to our in-memory state for the Node
        let mut locked = self.inner.write().unwrap();
-        let (nodes, _tenants, scheduler) = locked.parts_mut();
+        let (nodes, tenants, scheduler) = locked.parts_mut();

        let mut new_nodes = (**nodes).clone();

@@ -5397,8 +5356,8 @@ impl Service {
            ));
        };

-        if let Some(availability) = availability {
-            node.set_availability(availability);
+        if let Some(availability) = availability.as_ref() {
+            node.set_availability(availability.clone());
        }

        if let Some(scheduling) = scheduling {
@@ -5409,30 +5368,11 @@ impl Service {
        scheduler.node_upsert(node);

        let new_nodes = Arc::new(new_nodes);
-        locked.nodes = new_nodes;

-        Ok(availability_transition)
-    }
-
-    /// Handle availability transition of one node
-    ///
-    /// Note that you should first call [`Self::node_state_configure`] to update
-    /// the in-memory state referencing that node. If you need to handle more than one transition
-    /// consider using [`Self::handle_node_availability_transitions`].
-    async fn handle_node_availability_transition(
-        &self,
-        node_id: NodeId,
-        transition: AvailabilityTransition,
-        _node_lock: &TracingExclusiveGuard<NodeOperations>,
-    ) -> Result<(), ApiError> {
        // Modify scheduling state for any Tenants that are affected by a change in the node's availability state.
-        match transition {
+        match availability_transition {
            AvailabilityTransition::ToOffline => {
                tracing::info!("Node {} transition to offline", node_id);
-
-                let mut locked = self.inner.write().unwrap();
-                let (nodes, tenants, scheduler) = locked.parts_mut();
-
                let mut tenants_affected: usize = 0;

                for (tenant_shard_id, tenant_shard) in tenants {
@@ -5442,14 +5382,14 @@ impl Service {
                        observed_loc.conf = None;
                    }

-                    if nodes.len() == 1 {
+                    if new_nodes.len() == 1 {
                        // Special case for single-node cluster: there is no point trying to reschedule
                        // any tenant shards: avoid doing so, in order to avoid spewing warnings about
                        // failures to schedule them.
                        continue;
                    }

-                    if !nodes
+                    if !new_nodes
                        .values()
                        .any(|n| matches!(n.may_schedule(), MaySchedule::Yes(_)))
                    {
@@ -5475,7 +5415,10 @@ impl Service {
                                tracing::warn!(%tenant_shard_id, "Scheduling error when marking pageserver {} offline: {e}", node_id);
                            }
                            Ok(()) => {
-                                if self.maybe_reconcile_shard(tenant_shard, nodes).is_some() {
+                                if self
+                                    .maybe_reconcile_shard(tenant_shard, &new_nodes)
+                                    .is_some()
+                                {
                                    tenants_affected += 1;
                                };
                            }
@@ -5490,13 +5433,9 @@ impl Service {
            }
            AvailabilityTransition::ToActive => {
                tracing::info!("Node {} transition to active", node_id);
-
-                let mut locked = self.inner.write().unwrap();
-                let (nodes, tenants, _scheduler) = locked.parts_mut();
-
                // When a node comes back online, we must reconcile any tenant that has a None observed
                // location on the node.
-                for tenant_shard in tenants.values_mut() {
+                for tenant_shard in locked.tenants.values_mut() {
                    // If a reconciliation is already in progress, rely on the previous scheduling
                    // decision and skip triggering a new reconciliation.
                    if tenant_shard.reconciler.is_some() {
@@ -5505,7 +5444,7 @@ impl Service {

                    if let Some(observed_loc) = tenant_shard.observed.locations.get_mut(&node_id) {
                        if observed_loc.conf.is_none() {
-                            self.maybe_reconcile_shard(tenant_shard, nodes);
+                            self.maybe_reconcile_shard(tenant_shard, &new_nodes);
                        }
                    }
                }
@@ -5526,54 +5465,11 @@ impl Service {
            }
        }

+        locked.nodes = new_nodes;
+
        Ok(())
    }

-    /// Handle availability transition for multiple nodes
-    ///
-    /// Note that you should first call [`Self::node_state_configure`] for
-    /// all nodes being handled here for the handling to use fresh in-memory state.
-    async fn handle_node_availability_transitions(
-        &self,
-        transitions: Vec<(
-            NodeId,
-            TracingExclusiveGuard<NodeOperations>,
-            AvailabilityTransition,
-        )>,
-    ) -> Result<(), Vec<(NodeId, ApiError)>> {
-        let mut errors = Vec::default();
-        for (node_id, node_lock, transition) in transitions {
-            let res = self
-                .handle_node_availability_transition(node_id, transition, &node_lock)
-                .await;
-            if let Err(err) = res {
-                errors.push((node_id, err));
-            }
-        }
-
-        if errors.is_empty() {
-            Ok(())
-        } else {
-            Err(errors)
-        }
-    }
-
-    pub(crate) async fn node_configure(
-        &self,
-        node_id: NodeId,
-        availability: Option<NodeAvailability>,
-        scheduling: Option<NodeSchedulingPolicy>,
-    ) -> Result<(), ApiError> {
-        let node_lock =
-            trace_exclusive_lock(&self.node_op_locks, node_id, NodeOperations::Configure).await;
-
-        let transition = self
-            .node_state_configure(node_id, availability, scheduling, &node_lock)
-            .await?;
-        self.handle_node_availability_transition(node_id, transition, &node_lock)
-            .await
-    }
-
    /// Wrapper around [`Self::node_configure`] which only allows changes while there is no ongoing
    /// operation for HTTP api.
    pub(crate) async fn external_node_configure(
--- a/test_runner/cloud_regress/test_cloud_regress.py
+++ b/test_runner/cloud_regress/test_cloud_regress.py
@@ -2,8 +2,6 @@
 Run the regression tests on the cloud instance of Neon
 """

-from __future__ import annotations
-
 from pathlib import Path
 from typing import Any

--- a/test_runner/conftest.py
+++ b/test_runner/conftest.py
@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 pytest_plugins = (
    "fixtures.pg_version",
    "fixtures.parametrize",
--- a/test_runner/fixtures/init.py
+++ b/test_runner/fixtures/init.py
@@ -1 +0,0 @@
-from __future__ import annotations
--- a/test_runner/fixtures/benchmark_fixture.py
+++ b/test_runner/fixtures/benchmark_fixture.py
@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 import calendar
 import dataclasses
 import enum
@@ -7,11 +5,12 @@ import json
 import os
 import re
 import timeit
-from collections.abc import Iterator
 from contextlib import contextmanager
 from datetime import datetime
 from pathlib import Path
-from typing import TYPE_CHECKING
+
+# Type-related stuff
+from typing import Callable, ClassVar, Dict, Iterator, Optional

 import allure
 import pytest
@@ -24,10 +23,6 @@ from fixtures.common_types import TenantId, TimelineId
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import NeonPageserver

-if TYPE_CHECKING:
-    from typing import Callable, ClassVar, Optional
-
-
 """
 This file contains fixtures for micro-benchmarks.

@@ -143,6 +138,18 @@ class PgBenchRunResult:

@dataclasses.dataclass
 class PgBenchInitResult:
+    # Taken from https://github.com/postgres/postgres/blob/REL_15_1/src/bin/pgbench/pgbench.c#L5144-L5171
+    EXTRACTORS: ClassVar[Dict[str, re.Pattern]] = {  # type: ignore[type-arg]
+        "drop_tables": re.compile(r"drop tables (\d+\.\d+) s"),
+        "create_tables": re.compile(r"create tables (\d+\.\d+) s"),
+        "client_side_generate": re.compile(r"client-side generate (\d+\.\d+) s"),
+        "server_side_generate": re.compile(r"server-side generate (\d+\.\d+) s"),
+        "vacuum": re.compile(r"vacuum (\d+\.\d+) s"),
+        "primary_keys": re.compile(r"primary keys (\d+\.\d+) s"),
+        "foreign_keys": re.compile(r"foreign keys (\d+\.\d+) s"),
+        "total": re.compile(r"done in (\d+\.\d+) s"),  # Total time printed by pgbench
+    }
+
    total: Optional[float]
    drop_tables: Optional[float]
    create_tables: Optional[float]
@@ -155,20 +162,6 @@ class PgBenchInitResult:
    start_timestamp: int
    end_timestamp: int

-    # Taken from https://github.com/postgres/postgres/blob/REL_15_1/src/bin/pgbench/pgbench.c#L5144-L5171
-    EXTRACTORS: ClassVar[dict[str, re.Pattern[str]]] = dataclasses.field(
-        default_factory=lambda: {
-            "drop_tables": re.compile(r"drop tables (\d+\.\d+) s"),
-            "create_tables": re.compile(r"create tables (\d+\.\d+) s"),
-            "client_side_generate": re.compile(r"client-side generate (\d+\.\d+) s"),
-            "server_side_generate": re.compile(r"server-side generate (\d+\.\d+) s"),
-            "vacuum": re.compile(r"vacuum (\d+\.\d+) s"),
-            "primary_keys": re.compile(r"primary keys (\d+\.\d+) s"),
-            "foreign_keys": re.compile(r"foreign keys (\d+\.\d+) s"),
-            "total": re.compile(r"done in (\d+\.\d+) s"),  # Total time printed by pgbench
-        }
-    )
-
    @classmethod
    def parse_from_stderr(
        cls,
@@ -182,7 +175,7 @@ class PgBenchInitResult:

        last_line = stderr.splitlines()[-1]

-        timings: dict[str, Optional[float]] = {}
+        timings: Dict[str, Optional[float]] = {}
        last_line_items = re.split(r"\(|\)|,", last_line)
        for item in last_line_items:
            for key, regex in cls.EXTRACTORS.items():
@@ -392,7 +385,7 @@ class NeonBenchmarker:
        self,
        pageserver: NeonPageserver,
        metric_name: str,
-        label_filters: Optional[dict[str, str]] = None,
+        label_filters: Optional[Dict[str, str]] = None,
    ) -> int:
        """Fetch the value of given int counter from pageserver metrics."""
        all_metrics = pageserver.http_client().get_metrics()
--- a/test_runner/fixtures/common_types.py
+++ b/test_runner/fixtures/common_types.py
@@ -1,16 +1,10 @@
-from __future__ import annotations
-
 import random
 from dataclasses import dataclass
 from enum import Enum
 from functools import total_ordering
-from typing import TYPE_CHECKING, TypeVar
-
-if TYPE_CHECKING:
-    from typing import Any, Union
-
-    T = TypeVar("T", bound="Id")
+from typing import Any, Dict, Type, TypeVar, Union

+T = TypeVar("T", bound="Id")

 DEFAULT_WAL_SEG_SIZE = 16 * 1024 * 1024

@@ -62,7 +56,7 @@ class Lsn:
            return NotImplemented
        return self.lsn_int - other.lsn_int

-    def __add__(self, other: Union[int, Lsn]) -> Lsn:
+    def __add__(self, other: Union[int, "Lsn"]) -> "Lsn":
        if isinstance(other, int):
            return Lsn(self.lsn_int + other)
        elif isinstance(other, Lsn):
@@ -76,7 +70,7 @@ class Lsn:
    def as_int(self) -> int:
        return self.lsn_int

-    def segment_lsn(self, seg_sz: int = DEFAULT_WAL_SEG_SIZE) -> Lsn:
+    def segment_lsn(self, seg_sz: int = DEFAULT_WAL_SEG_SIZE) -> "Lsn":
        return Lsn(self.lsn_int - (self.lsn_int % seg_sz))

    def segno(self, seg_sz: int = DEFAULT_WAL_SEG_SIZE) -> int:
@@ -133,7 +127,7 @@ class Id:
        return hash(str(self.id))

    @classmethod
-    def generate(cls: type[T]) -> T:
+    def generate(cls: Type[T]) -> T:
        """Generate a random ID"""
        return cls(random.randbytes(16).hex())

@@ -168,7 +162,7 @@ class TenantTimelineId:
    timeline_id: TimelineId

    @classmethod
-    def from_json(cls, d: dict[str, Any]) -> TenantTimelineId:
+    def from_json(cls, d: Dict[str, Any]) -> "TenantTimelineId":
        return TenantTimelineId(
            tenant_id=TenantId(d["tenant_id"]),
            timeline_id=TimelineId(d["timeline_id"]),
@@ -187,7 +181,7 @@ class TenantShardId:
        assert self.shard_number < self.shard_count or self.shard_count == 0

    @classmethod
-    def parse(cls: type[TTenantShardId], input) -> TTenantShardId:
+    def parse(cls: Type[TTenantShardId], input) -> TTenantShardId:
        if len(input) == 32:
            return cls(
                tenant_id=TenantId(input),
--- a/test_runner/fixtures/compare_fixtures.py
+++ b/test_runner/fixtures/compare_fixtures.py
@@ -1,13 +1,11 @@
-from __future__ import annotations
-
 import os
 import time
 from abc import ABC, abstractmethod
-from collections.abc import Iterator
 from contextlib import _GeneratorContextManager, contextmanager

 # Type-related stuff
 from pathlib import Path
+from typing import Dict, Iterator, List

 import pytest
 from _pytest.fixtures import FixtureRequest
@@ -74,7 +72,7 @@ class PgCompare(ABC):
        pass

    @contextmanager
-    def record_pg_stats(self, pg_stats: list[PgStatTable]) -> Iterator[None]:
+    def record_pg_stats(self, pg_stats: List[PgStatTable]) -> Iterator[None]:
        init_data = self._retrieve_pg_stats(pg_stats)

        yield
@@ -84,8 +82,8 @@ class PgCompare(ABC):
        for k in set(init_data) & set(data):
            self.zenbenchmark.record(k, data[k] - init_data[k], "", MetricReport.HIGHER_IS_BETTER)

-    def _retrieve_pg_stats(self, pg_stats: list[PgStatTable]) -> dict[str, int]:
-        results: dict[str, int] = {}
+    def _retrieve_pg_stats(self, pg_stats: List[PgStatTable]) -> Dict[str, int]:
+        results: Dict[str, int] = {}

        with self.pg.connect().cursor() as cur:
            for pg_stat in pg_stats:
--- a/test_runner/fixtures/compute_reconfigure.py
+++ b/test_runner/fixtures/compute_reconfigure.py
@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 import concurrent.futures
 from typing import Any

--- a/test_runner/fixtures/endpoint/init.py
+++ b/test_runner/fixtures/endpoint/init.py
@@ -1 +0,0 @@
-from __future__ import annotations
--- a/test_runner/fixtures/endpoint/http.py
+++ b/test_runner/fixtures/endpoint/http.py
@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 import requests
 from requests.adapters import HTTPAdapter

--- a/test_runner/fixtures/flaky.py
+++ b/test_runner/fixtures/flaky.py
@@ -1,9 +1,6 @@
-from __future__ import annotations
-
 import json
-from collections.abc import MutableMapping
 from pathlib import Path
-from typing import TYPE_CHECKING, cast
+from typing import Any, List, MutableMapping, cast

 import pytest
 from _pytest.config import Config
@@ -13,9 +10,6 @@ from allure_pytest.utils import allure_name, allure_suite_labels

 from fixtures.log_helper import log

-if TYPE_CHECKING:
-    from typing import Any
-
 """
 The plugin reruns flaky tests.
 It uses `pytest.mark.flaky` provided by `pytest-rerunfailures` plugin and flaky tests detected by `scripts/flaky_tests.py`
@@ -33,7 +27,7 @@ def pytest_addoption(parser: Parser):
    )


-def pytest_collection_modifyitems(config: Config, items: list[pytest.Item]):
+def pytest_collection_modifyitems(config: Config, items: List[pytest.Item]):
    if not config.getoption("--flaky-tests-json"):
        return

@@ -72,5 +66,5 @@ def pytest_collection_modifyitems(config: Config, items: list[pytest.Item]):
            # - [2] https://github.com/pytest-dev/pytest-timeout/issues/142
            timeout_marker = item.get_closest_marker("timeout")
            if timeout_marker is not None:
-                kwargs = cast("MutableMapping[str, Any]", timeout_marker.kwargs)
+                kwargs = cast(MutableMapping[str, Any], timeout_marker.kwargs)
                kwargs["func_only"] = True
--- a/test_runner/fixtures/httpserver.py
+++ b/test_runner/fixtures/httpserver.py
@@ -1,4 +1,4 @@
-from __future__ import annotations
+from typing import Tuple

 import pytest
 from pytest_httpserver import HTTPServer
@@ -40,6 +40,6 @@ def httpserver(make_httpserver):


@pytest.fixture(scope="function")
-def httpserver_listen_address(port_distributor) -> tuple[str, int]:
+def httpserver_listen_address(port_distributor) -> Tuple[str, int]:
    port = port_distributor.get_port()
    return ("localhost", port)
--- a/test_runner/fixtures/log_helper.py
+++ b/test_runner/fixtures/log_helper.py
@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 import logging
 import logging.config

--- a/test_runner/fixtures/metrics.py
+++ b/test_runner/fixtures/metrics.py
@@ -1,26 +1,21 @@
-from __future__ import annotations
-
 from collections import defaultdict
-from typing import TYPE_CHECKING
+from typing import Dict, List, Optional, Tuple

 from prometheus_client.parser import text_string_to_metric_families
 from prometheus_client.samples import Sample

 from fixtures.log_helper import log

-if TYPE_CHECKING:
-    from typing import Optional
-

 class Metrics:
-    metrics: dict[str, list[Sample]]
+    metrics: Dict[str, List[Sample]]
    name: str

    def __init__(self, name: str = ""):
        self.metrics = defaultdict(list)
        self.name = name

-    def query_all(self, name: str, filter: Optional[dict[str, str]] = None) -> list[Sample]:
+    def query_all(self, name: str, filter: Optional[Dict[str, str]] = None) -> List[Sample]:
        filter = filter or {}
        res = []

@@ -32,7 +27,7 @@ class Metrics:
                pass
        return res

-    def query_one(self, name: str, filter: Optional[dict[str, str]] = None) -> Sample:
+    def query_one(self, name: str, filter: Optional[Dict[str, str]] = None) -> Sample:
        res = self.query_all(name, filter or {})
        assert len(res) == 1, f"expected single sample for {name} {filter}, found {res}"
        return res[0]
@@ -48,7 +43,7 @@ class MetricsGetter:
        raise NotImplementedError()

    def get_metric_value(
-        self, name: str, filter: Optional[dict[str, str]] = None
+        self, name: str, filter: Optional[Dict[str, str]] = None
    ) -> Optional[float]:
        metrics = self.get_metrics()
        results = metrics.query_all(name, filter=filter)
@@ -59,8 +54,8 @@ class MetricsGetter:
        return results[0].value

    def get_metrics_values(
-        self, names: list[str], filter: Optional[dict[str, str]] = None, absence_ok=False
-    ) -> dict[str, float]:
+        self, names: list[str], filter: Optional[Dict[str, str]] = None, absence_ok=False
+    ) -> Dict[str, float]:
        """
        When fetching multiple named metrics, it is more efficient to use this
        than to call `get_metric_value` repeatedly.
@@ -102,7 +97,7 @@ def parse_metrics(text: str, name: str = "") -> Metrics:
    return metrics


-def histogram(prefix_without_trailing_underscore: str) -> list[str]:
+def histogram(prefix_without_trailing_underscore: str) -> List[str]:
    assert not prefix_without_trailing_underscore.endswith("_")
    return [f"{prefix_without_trailing_underscore}_{x}" for x in ["bucket", "count", "sum"]]

@@ -112,7 +107,7 @@ def counter(name: str) -> str:
    return f"{name}_total"


-PAGESERVER_PER_TENANT_REMOTE_TIMELINE_CLIENT_METRICS: tuple[str, ...] = (
+PAGESERVER_PER_TENANT_REMOTE_TIMELINE_CLIENT_METRICS: Tuple[str, ...] = (
    "pageserver_remote_timeline_client_calls_started_total",
    "pageserver_remote_timeline_client_calls_finished_total",
    "pageserver_remote_physical_size",
@@ -120,7 +115,7 @@ PAGESERVER_PER_TENANT_REMOTE_TIMELINE_CLIENT_METRICS: tuple[str, ...] = (
    "pageserver_remote_timeline_client_bytes_finished_total",
 )

-PAGESERVER_GLOBAL_METRICS: tuple[str, ...] = (
+PAGESERVER_GLOBAL_METRICS: Tuple[str, ...] = (
    "pageserver_storage_operations_seconds_global_count",
    "pageserver_storage_operations_seconds_global_sum",
    "pageserver_storage_operations_seconds_global_bucket",
@@ -152,7 +147,7 @@ PAGESERVER_GLOBAL_METRICS: tuple[str, ...] = (
    counter("pageserver_tenant_throttling_count_global"),
 )

-PAGESERVER_PER_TENANT_METRICS: tuple[str, ...] = (
+PAGESERVER_PER_TENANT_METRICS: Tuple[str, ...] = (
    "pageserver_current_logical_size",
    "pageserver_resident_physical_size",
    "pageserver_io_operations_bytes_total",
--- a/test_runner/fixtures/neon_api.py
+++ b/test_runner/fixtures/neon_api.py
@@ -6,12 +6,12 @@ from typing import TYPE_CHECKING, cast
 import requests

 if TYPE_CHECKING:
-    from typing import Any, Literal, Optional, Union
+    from typing import Any, Dict, Literal, Optional, Union

    from fixtures.pg_version import PgVersion


-def connection_parameters_to_env(params: dict[str, str]) -> dict[str, str]:
+def connection_parameters_to_env(params: Dict[str, str]) -> Dict[str, str]:
    return {
        "PGHOST": params["host"],
        "PGDATABASE": params["database"],
@@ -41,8 +41,8 @@ class NeonAPI:
        branch_name: Optional[str] = None,
        branch_role_name: Optional[str] = None,
        branch_database_name: Optional[str] = None,
-    ) -> dict[str, Any]:
-        data: dict[str, Any] = {
+    ) -> Dict[str, Any]:
+        data: Dict[str, Any] = {
            "project": {
                "branch": {},
            },
@@ -70,9 +70,9 @@ class NeonAPI:

        assert resp.status_code == 201

-        return cast("dict[str, Any]", resp.json())
+        return cast("Dict[str, Any]", resp.json())

-    def get_project_details(self, project_id: str) -> dict[str, Any]:
+    def get_project_details(self, project_id: str) -> Dict[str, Any]:
        resp = self.__request(
            "GET",
            f"/projects/{project_id}",
@@ -82,12 +82,12 @@ class NeonAPI:
            },
        )
        assert resp.status_code == 200
-        return cast("dict[str, Any]", resp.json())
+        return cast("Dict[str, Any]", resp.json())

    def delete_project(
        self,
        project_id: str,
-    ) -> dict[str, Any]:
+    ) -> Dict[str, Any]:
        resp = self.__request(
            "DELETE",
            f"/projects/{project_id}",
@@ -99,13 +99,13 @@ class NeonAPI:

        assert resp.status_code == 200

-        return cast("dict[str, Any]", resp.json())
+        return cast("Dict[str, Any]", resp.json())

    def start_endpoint(
        self,
        project_id: str,
        endpoint_id: str,
-    ) -> dict[str, Any]:
+    ) -> Dict[str, Any]:
        resp = self.__request(
            "POST",
            f"/projects/{project_id}/endpoints/{endpoint_id}/start",
@@ -116,13 +116,13 @@ class NeonAPI:

        assert resp.status_code == 200

-        return cast("dict[str, Any]", resp.json())
+        return cast("Dict[str, Any]", resp.json())

    def suspend_endpoint(
        self,
        project_id: str,
        endpoint_id: str,
-    ) -> dict[str, Any]:
+    ) -> Dict[str, Any]:
        resp = self.__request(
            "POST",
            f"/projects/{project_id}/endpoints/{endpoint_id}/suspend",
@@ -133,13 +133,13 @@ class NeonAPI:

        assert resp.status_code == 200

-        return cast("dict[str, Any]", resp.json())
+        return cast("Dict[str, Any]", resp.json())

    def restart_endpoint(
        self,
        project_id: str,
        endpoint_id: str,
-    ) -> dict[str, Any]:
+    ) -> Dict[str, Any]:
        resp = self.__request(
            "POST",
            f"/projects/{project_id}/endpoints/{endpoint_id}/restart",
@@ -150,16 +150,16 @@ class NeonAPI:

        assert resp.status_code == 200

-        return cast("dict[str, Any]", resp.json())
+        return cast("Dict[str, Any]", resp.json())

    def create_endpoint(
        self,
        project_id: str,
        branch_id: str,
        endpoint_type: Literal["read_write", "read_only"],
-        settings: dict[str, Any],
-    ) -> dict[str, Any]:
-        data: dict[str, Any] = {
+        settings: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        data: Dict[str, Any] = {
            "endpoint": {
                "branch_id": branch_id,
            },
@@ -182,7 +182,7 @@ class NeonAPI:

        assert resp.status_code == 201

-        return cast("dict[str, Any]", resp.json())
+        return cast("Dict[str, Any]", resp.json())

    def get_connection_uri(
        self,
@@ -192,7 +192,7 @@ class NeonAPI:
        database_name: str = "neondb",
        role_name: str = "neondb_owner",
        pooled: bool = True,
-    ) -> dict[str, Any]:
+    ) -> Dict[str, Any]:
        resp = self.__request(
            "GET",
            f"/projects/{project_id}/connection_uri",
@@ -210,9 +210,9 @@ class NeonAPI:

        assert resp.status_code == 200

-        return cast("dict[str, Any]", resp.json())
+        return cast("Dict[str, Any]", resp.json())

-    def get_branches(self, project_id: str) -> dict[str, Any]:
+    def get_branches(self, project_id: str) -> Dict[str, Any]:
        resp = self.__request(
            "GET",
            f"/projects/{project_id}/branches",
@@ -223,9 +223,9 @@ class NeonAPI:

        assert resp.status_code == 200

-        return cast("dict[str, Any]", resp.json())
+        return cast("Dict[str, Any]", resp.json())

-    def get_endpoints(self, project_id: str) -> dict[str, Any]:
+    def get_endpoints(self, project_id: str) -> Dict[str, Any]:
        resp = self.__request(
            "GET",
            f"/projects/{project_id}/endpoints",
@@ -236,9 +236,9 @@ class NeonAPI:

        assert resp.status_code == 200

-        return cast("dict[str, Any]", resp.json())
+        return cast("Dict[str, Any]", resp.json())

-    def get_operations(self, project_id: str) -> dict[str, Any]:
+    def get_operations(self, project_id: str) -> Dict[str, Any]:
        resp = self.__request(
            "GET",
            f"/projects/{project_id}/operations",
@@ -250,7 +250,7 @@ class NeonAPI:

        assert resp.status_code == 200

-        return cast("dict[str, Any]", resp.json())
+        return cast("Dict[str, Any]", resp.json())

    def wait_for_operation_to_finish(self, project_id: str):
        has_running = True
--- a/test_runner/fixtures/neon_cli.py
+++ b/test_runner/fixtures/neon_cli.py
@@ -9,7 +9,15 @@ import tempfile
 import textwrap
 from itertools import chain, product
 from pathlib import Path
-from typing import TYPE_CHECKING, cast
+from typing import (
+    Any,
+    Dict,
+    List,
+    Optional,
+    Tuple,
+    TypeVar,
+    cast,
+)

 import toml

@@ -19,15 +27,7 @@ from fixtures.pageserver.common_types import IndexPartDump
 from fixtures.pg_version import PgVersion
 from fixtures.utils import AuxFileStore

-if TYPE_CHECKING:
-    from typing import (
-        Any,
-        Optional,
-        TypeVar,
-        cast,
-    )
-
-    T = TypeVar("T")
+T = TypeVar("T")


 class AbstractNeonCli(abc.ABC):
@@ -37,7 +37,7 @@ class AbstractNeonCli(abc.ABC):
    Do not use directly, use specific subclasses instead.
    """

-    def __init__(self, extra_env: Optional[dict[str, str]], binpath: Path):
+    def __init__(self, extra_env: Optional[Dict[str, str]], binpath: Path):
        self.extra_env = extra_env
        self.binpath = binpath

@@ -45,11 +45,11 @@ class AbstractNeonCli(abc.ABC):

    def raw_cli(
        self,
-        arguments: list[str],
-        extra_env_vars: Optional[dict[str, str]] = None,
+        arguments: List[str],
+        extra_env_vars: Optional[Dict[str, str]] = None,
        check_return_code=True,
        timeout=None,
-    ) -> subprocess.CompletedProcess[str]:
+    ) -> "subprocess.CompletedProcess[str]":
        """
        Run the command with the specified arguments.

@@ -92,8 +92,9 @@ class AbstractNeonCli(abc.ABC):
                args,
                env=env_vars,
                check=False,
-                text=True,
-                capture_output=True,
+                universal_newlines=True,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
                timeout=timeout,
            )
        except subprocess.TimeoutExpired as e:
@@ -117,7 +118,7 @@ class AbstractNeonCli(abc.ABC):
            if len(lines) < 2:
                log.debug(f"Run {res.args} success: {stripped}")
            else:
-                log.debug("Run %s success:\n%s", res.args, textwrap.indent(stripped, indent))
+                log.debug("Run %s success:\n%s" % (res.args, textwrap.indent(stripped, indent)))
        elif check_return_code:
            # this way command output will be in recorded and shown in CI in failure message
            indent = indent * 2
@@ -174,7 +175,7 @@ class NeonLocalCli(AbstractNeonCli):

    def __init__(
        self,
-        extra_env: Optional[dict[str, str]],
+        extra_env: Optional[Dict[str, str]],
        binpath: Path,
        repo_dir: Path,
        pg_distrib_dir: Path,
@@ -196,7 +197,7 @@ class NeonLocalCli(AbstractNeonCli):
        tenant_id: TenantId,
        timeline_id: TimelineId,
        pg_version: PgVersion,
-        conf: Optional[dict[str, Any]] = None,
+        conf: Optional[Dict[str, Any]] = None,
        shard_count: Optional[int] = None,
        shard_stripe_size: Optional[int] = None,
        placement_policy: Optional[str] = None,
@@ -257,7 +258,7 @@ class NeonLocalCli(AbstractNeonCli):
        res = self.raw_cli(["tenant", "set-default", "--tenant-id", str(tenant_id)])
        res.check_returncode()

-    def tenant_config(self, tenant_id: TenantId, conf: dict[str, str]):
+    def tenant_config(self, tenant_id: TenantId, conf: Dict[str, str]):
        """
        Update tenant config.
        """
@@ -273,7 +274,7 @@ class NeonLocalCli(AbstractNeonCli):
        res = self.raw_cli(args)
        res.check_returncode()

-    def tenant_list(self) -> subprocess.CompletedProcess[str]:
+    def tenant_list(self) -> "subprocess.CompletedProcess[str]":
        res = self.raw_cli(["tenant", "list"])
        res.check_returncode()
        return res
@@ -367,7 +368,7 @@ class NeonLocalCli(AbstractNeonCli):
        res = self.raw_cli(cmd)
        res.check_returncode()

-    def timeline_list(self, tenant_id: TenantId) -> list[tuple[str, TimelineId]]:
+    def timeline_list(self, tenant_id: TenantId) -> List[Tuple[str, TimelineId]]:
        """
        Returns a list of (branch_name, timeline_id) tuples out of parsed `neon timeline list` CLI output.
        """
@@ -388,9 +389,9 @@ class NeonLocalCli(AbstractNeonCli):

    def init(
        self,
-        init_config: dict[str, Any],
+        init_config: Dict[str, Any],
        force: Optional[str] = None,
-    ) -> subprocess.CompletedProcess[str]:
+    ) -> "subprocess.CompletedProcess[str]":
        with tempfile.NamedTemporaryFile(mode="w+") as init_config_tmpfile:
            init_config_tmpfile.write(toml.dumps(init_config))
            init_config_tmpfile.flush()
@@ -433,15 +434,15 @@ class NeonLocalCli(AbstractNeonCli):
    def pageserver_start(
        self,
        id: int,
-        extra_env_vars: Optional[dict[str, str]] = None,
+        extra_env_vars: Optional[Dict[str, str]] = None,
        timeout_in_seconds: Optional[int] = None,
-    ) -> subprocess.CompletedProcess[str]:
+    ) -> "subprocess.CompletedProcess[str]":
        start_args = ["pageserver", "start", f"--id={id}"]
        if timeout_in_seconds is not None:
            start_args.append(f"--start-timeout={timeout_in_seconds}s")
        return self.raw_cli(start_args, extra_env_vars=extra_env_vars)

-    def pageserver_stop(self, id: int, immediate=False) -> subprocess.CompletedProcess[str]:
+    def pageserver_stop(self, id: int, immediate=False) -> "subprocess.CompletedProcess[str]":
        cmd = ["pageserver", "stop", f"--id={id}"]
        if immediate:
            cmd.extend(["-m", "immediate"])
@@ -452,10 +453,10 @@ class NeonLocalCli(AbstractNeonCli):
    def safekeeper_start(
        self,
        id: int,
-        extra_opts: Optional[list[str]] = None,
-        extra_env_vars: Optional[dict[str, str]] = None,
+        extra_opts: Optional[List[str]] = None,
+        extra_env_vars: Optional[Dict[str, str]] = None,
        timeout_in_seconds: Optional[int] = None,
-    ) -> subprocess.CompletedProcess[str]:
+    ) -> "subprocess.CompletedProcess[str]":
        if extra_opts is not None:
            extra_opts = [f"-e={opt}" for opt in extra_opts]
        else:
@@ -468,7 +469,7 @@ class NeonLocalCli(AbstractNeonCli):

    def safekeeper_stop(
        self, id: Optional[int] = None, immediate=False
-    ) -> subprocess.CompletedProcess[str]:
+    ) -> "subprocess.CompletedProcess[str]":
        args = ["safekeeper", "stop"]
        if id is not None:
            args.append(str(id))
@@ -478,13 +479,13 @@ class NeonLocalCli(AbstractNeonCli):

    def storage_broker_start(
        self, timeout_in_seconds: Optional[int] = None
-    ) -> subprocess.CompletedProcess[str]:
+    ) -> "subprocess.CompletedProcess[str]":
        cmd = ["storage_broker", "start"]
        if timeout_in_seconds is not None:
            cmd.append(f"--start-timeout={timeout_in_seconds}s")
        return self.raw_cli(cmd)

-    def storage_broker_stop(self) -> subprocess.CompletedProcess[str]:
+    def storage_broker_stop(self) -> "subprocess.CompletedProcess[str]":
        cmd = ["storage_broker", "stop"]
        return self.raw_cli(cmd)

@@ -500,7 +501,7 @@ class NeonLocalCli(AbstractNeonCli):
        lsn: Optional[Lsn] = None,
        pageserver_id: Optional[int] = None,
        allow_multiple=False,
-    ) -> subprocess.CompletedProcess[str]:
+    ) -> "subprocess.CompletedProcess[str]":
        args = [
            "endpoint",
            "create",
@@ -533,12 +534,12 @@ class NeonLocalCli(AbstractNeonCli):
    def endpoint_start(
        self,
        endpoint_id: str,
-        safekeepers: Optional[list[int]] = None,
+        safekeepers: Optional[List[int]] = None,
        remote_ext_config: Optional[str] = None,
        pageserver_id: Optional[int] = None,
        allow_multiple=False,
        basebackup_request_tries: Optional[int] = None,
-    ) -> subprocess.CompletedProcess[str]:
+    ) -> "subprocess.CompletedProcess[str]":
        args = [
            "endpoint",
            "start",
@@ -567,9 +568,9 @@ class NeonLocalCli(AbstractNeonCli):
        endpoint_id: str,
        tenant_id: Optional[TenantId] = None,
        pageserver_id: Optional[int] = None,
-        safekeepers: Optional[list[int]] = None,
+        safekeepers: Optional[List[int]] = None,
        check_return_code=True,
-    ) -> subprocess.CompletedProcess[str]:
+    ) -> "subprocess.CompletedProcess[str]":
        args = ["endpoint", "reconfigure", endpoint_id]
        if tenant_id is not None:
            args.extend(["--tenant-id", str(tenant_id)])
@@ -585,7 +586,7 @@ class NeonLocalCli(AbstractNeonCli):
        destroy=False,
        check_return_code=True,
        mode: Optional[str] = None,
-    ) -> subprocess.CompletedProcess[str]:
+    ) -> "subprocess.CompletedProcess[str]":
        args = [
            "endpoint",
            "stop",
@@ -601,7 +602,7 @@ class NeonLocalCli(AbstractNeonCli):

    def mappings_map_branch(
        self, name: str, tenant_id: TenantId, timeline_id: TimelineId
-    ) -> subprocess.CompletedProcess[str]:
+    ) -> "subprocess.CompletedProcess[str]":
        """
        Map tenant id and timeline id to a neon_local branch name. They do not have to exist.
        Usually needed when creating branches via PageserverHttpClient and not neon_local.
@@ -622,10 +623,10 @@ class NeonLocalCli(AbstractNeonCli):

        return self.raw_cli(args, check_return_code=True)

-    def start(self, check_return_code=True) -> subprocess.CompletedProcess[str]:
+    def start(self, check_return_code=True) -> "subprocess.CompletedProcess[str]":
        return self.raw_cli(["start"], check_return_code=check_return_code)

-    def stop(self, check_return_code=True) -> subprocess.CompletedProcess[str]:
+    def stop(self, check_return_code=True) -> "subprocess.CompletedProcess[str]":
        return self.raw_cli(["stop"], check_return_code=check_return_code)


@@ -637,7 +638,7 @@ class WalCraft(AbstractNeonCli):

    COMMAND = "wal_craft"

-    def postgres_config(self) -> list[str]:
+    def postgres_config(self) -> List[str]:
        res = self.raw_cli(["print-postgres-config"])
        res.check_returncode()
        return res.stdout.split("\n")
--- a/test_runner/fixtures/neon_fixtures.py
+++ b/test_runner/fixtures/neon_fixtures.py
@@ -13,7 +13,6 @@ import threading
 import time
 import uuid
 from collections import defaultdict
-from collections.abc import Iterable, Iterator
 from contextlib import closing, contextmanager
 from dataclasses import dataclass
 from datetime import datetime
@@ -22,7 +21,20 @@ from fcntl import LOCK_EX, LOCK_UN, flock
 from functools import cached_property
 from pathlib import Path
 from types import TracebackType
-from typing import TYPE_CHECKING, cast
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    Iterable,
+    Iterator,
+    List,
+    Optional,
+    Tuple,
+    Type,
+    TypeVar,
+    Union,
+    cast,
+)
 from urllib.parse import quote, urlparse

 import asyncpg
@@ -79,6 +91,7 @@ from fixtures.utils import (
    allure_attach_from_dir,
    assert_no_errors,
    get_dir_size,
+    get_self_dir,
    print_gc_result,
    subprocess_capture,
    wait_until,
@@ -87,17 +100,7 @@ from fixtures.utils import AuxFileStore as AuxFileStore  # reexport

 from .neon_api import NeonAPI, NeonApiEndpoint

-if TYPE_CHECKING:
-    from typing import (
-        Any,
-        Callable,
-        Optional,
-        TypeVar,
-        Union,
-    )
-
-    T = TypeVar("T")
-
+T = TypeVar("T")

 """
 This file contains pytest fixtures. A fixture is a test resource that can be
@@ -116,7 +119,7 @@ Don't import functions from this file, or pytest will emit warnings. Instead
 put directly-importable functions into utils.py or another separate file.
 """

-Env = dict[str, str]
+Env = Dict[str, str]

 DEFAULT_OUTPUT_DIR: str = "test_output"
 DEFAULT_BRANCH_NAME: str = "main"
@@ -127,7 +130,7 @@ BASE_PORT: int = 15000
@pytest.fixture(scope="session")
 def base_dir() -> Iterator[Path]:
    # find the base directory (currently this is the git root)
-    base_dir = Path(__file__).parents[2]
+    base_dir = get_self_dir().parent.parent
    log.info(f"base_dir is {base_dir}")

    yield base_dir
@@ -248,7 +251,7 @@ class PgProtocol:
        """
        return str(make_dsn(**self.conn_options(**kwargs)))

-    def conn_options(self, **kwargs: Any) -> dict[str, Any]:
+    def conn_options(self, **kwargs: Any) -> Dict[str, Any]:
        """
        Construct a dictionary of connection options from default values and extra parameters.
        An option can be dropped from the returning dictionary by None-valued extra parameter.
@@ -317,7 +320,7 @@ class PgProtocol:
                    conn_options["server_settings"] = {key: val}
        return await asyncpg.connect(**conn_options)

-    def safe_psql(self, query: str, **kwargs: Any) -> list[tuple[Any, ...]]:
+    def safe_psql(self, query: str, **kwargs: Any) -> List[Tuple[Any, ...]]:
        """
        Execute query against the node and return all rows.
        This method passes all extra params to connstr.
@@ -326,12 +329,12 @@ class PgProtocol:

    def safe_psql_many(
        self, queries: Iterable[str], log_query=True, **kwargs: Any
-    ) -> list[list[tuple[Any, ...]]]:
+    ) -> List[List[Tuple[Any, ...]]]:
        """
        Execute queries against the node and return all rows.
        This method passes all extra params to connstr.
        """
-        result: list[list[Any]] = []
+        result: List[List[Any]] = []
        with closing(self.connect(**kwargs)) as conn:
            with conn.cursor() as cur:
                for query in queries:
@@ -377,7 +380,7 @@ class NeonEnvBuilder:
        test_overlay_dir: Optional[Path] = None,
        pageserver_remote_storage: Optional[RemoteStorage] = None,
        # toml that will be decomposed into `--config-override` flags during `pageserver --init`
-        pageserver_config_override: Optional[str | Callable[[dict[str, Any]], None]] = None,
+        pageserver_config_override: Optional[str | Callable[[Dict[str, Any]], None]] = None,
        num_safekeepers: int = 1,
        num_pageservers: int = 1,
        # Use non-standard SK ids to check for various parsing bugs
@@ -392,7 +395,7 @@ class NeonEnvBuilder:
        initial_timeline: Optional[TimelineId] = None,
        pageserver_virtual_file_io_engine: Optional[str] = None,
        pageserver_aux_file_policy: Optional[AuxFileStore] = None,
-        pageserver_default_tenant_config_compaction_algorithm: Optional[dict[str, Any]] = None,
+        pageserver_default_tenant_config_compaction_algorithm: Optional[Dict[str, Any]] = None,
        safekeeper_extra_opts: Optional[list[str]] = None,
        storage_controller_port_override: Optional[int] = None,
        pageserver_io_buffer_alignment: Optional[int] = None,
@@ -427,7 +430,7 @@ class NeonEnvBuilder:
        self.enable_scrub_on_exit = True
        self.test_output_dir = test_output_dir
        self.test_overlay_dir = test_overlay_dir
-        self.overlay_mounts_created_by_us: list[tuple[str, Path]] = []
+        self.overlay_mounts_created_by_us: List[Tuple[str, Path]] = []
        self.config_init_force: Optional[str] = None
        self.top_output_dir = top_output_dir
        self.control_plane_compute_hook_api: Optional[str] = None
@@ -436,7 +439,7 @@ class NeonEnvBuilder:
        self.pageserver_virtual_file_io_engine: Optional[str] = pageserver_virtual_file_io_engine

        self.pageserver_default_tenant_config_compaction_algorithm: Optional[
-            dict[str, Any]
+            Dict[str, Any]
        ] = pageserver_default_tenant_config_compaction_algorithm
        if self.pageserver_default_tenant_config_compaction_algorithm is not None:
            log.debug(
@@ -466,7 +469,7 @@ class NeonEnvBuilder:

    def init_start(
        self,
-        initial_tenant_conf: Optional[dict[str, Any]] = None,
+        initial_tenant_conf: Optional[Dict[str, Any]] = None,
        default_remote_storage_if_missing: bool = True,
        initial_tenant_shard_count: Optional[int] = None,
        initial_tenant_shard_stripe_size: Optional[int] = None,
@@ -821,7 +824,7 @@ class NeonEnvBuilder:

        overlayfs_mounts = {mountpoint for _, mountpoint in self.overlay_mounts_created_by_us}

-        directories_to_clean: list[Path] = []
+        directories_to_clean: List[Path] = []
        for test_entry in Path(self.repo_dir).glob("**/*"):
            if test_entry in overlayfs_mounts:
                continue
@@ -852,12 +855,12 @@ class NeonEnvBuilder:
            if isinstance(x, S3Storage):
                x.do_cleanup()

-    def __enter__(self) -> NeonEnvBuilder:
+    def __enter__(self) -> "NeonEnvBuilder":
        return self

    def __exit__(
        self,
-        exc_type: Optional[type[BaseException]],
+        exc_type: Optional[Type[BaseException]],
        exc_value: Optional[BaseException],
        traceback: Optional[TracebackType],
    ):
@@ -968,8 +971,8 @@ class NeonEnv:
        self.port_distributor = config.port_distributor
        self.s3_mock_server = config.mock_s3_server
        self.endpoints = EndpointFactory(self)
-        self.safekeepers: list[Safekeeper] = []
-        self.pageservers: list[NeonPageserver] = []
+        self.safekeepers: List[Safekeeper] = []
+        self.pageservers: List[NeonPageserver] = []
        self.broker = NeonBroker(self)
        self.pageserver_remote_storage = config.pageserver_remote_storage
        self.safekeepers_remote_storage = config.safekeepers_remote_storage
@@ -1041,7 +1044,7 @@ class NeonEnv:
        self.pageserver_io_buffer_alignment = config.pageserver_io_buffer_alignment

        # Create the neon_local's `NeonLocalInitConf`
-        cfg: dict[str, Any] = {
+        cfg: Dict[str, Any] = {
            "default_tenant_id": str(self.initial_tenant),
            "broker": {
                "listen_addr": self.broker.listen_addr(),
@@ -1070,7 +1073,7 @@ class NeonEnv:
                http=self.port_distributor.get_port(),
            )

-            ps_cfg: dict[str, Any] = {
+            ps_cfg: Dict[str, Any] = {
                "id": ps_id,
                "listen_pg_addr": f"localhost:{pageserver_port.pg}",
                "listen_http_addr": f"localhost:{pageserver_port.http}",
@@ -1118,7 +1121,7 @@ class NeonEnv:
                http=self.port_distributor.get_port(),
            )
            id = config.safekeepers_id_start + i  # assign ids sequentially
-            sk_cfg: dict[str, Any] = {
+            sk_cfg: Dict[str, Any] = {
                "id": id,
                "pg_port": port.pg,
                "pg_tenant_only_port": port.pg_tenant_only,
@@ -1283,8 +1286,9 @@ class NeonEnv:
        res = subprocess.run(
            [bin_pageserver, "--version"],
            check=True,
-            text=True,
-            capture_output=True,
+            universal_newlines=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
        )
        return res.stdout

@@ -1327,13 +1331,13 @@ class NeonEnv:
        self,
        tenant_id: Optional[TenantId] = None,
        timeline_id: Optional[TimelineId] = None,
-        conf: Optional[dict[str, Any]] = None,
+        conf: Optional[Dict[str, Any]] = None,
        shard_count: Optional[int] = None,
        shard_stripe_size: Optional[int] = None,
        placement_policy: Optional[str] = None,
        set_default: bool = False,
        aux_file_policy: Optional[AuxFileStore] = None,
-    ) -> tuple[TenantId, TimelineId]:
+    ) -> Tuple[TenantId, TimelineId]:
        """
        Creates a new tenant, returns its id and its initial timeline's id.
        """
@@ -1354,7 +1358,7 @@ class NeonEnv:

        return tenant_id, timeline_id

-    def config_tenant(self, tenant_id: Optional[TenantId], conf: dict[str, str]):
+    def config_tenant(self, tenant_id: Optional[TenantId], conf: Dict[str, str]):
        """
        Update tenant config.
        """
@@ -1406,7 +1410,7 @@ def neon_simple_env(
    pg_version: PgVersion,
    pageserver_virtual_file_io_engine: str,
    pageserver_aux_file_policy: Optional[AuxFileStore],
-    pageserver_default_tenant_config_compaction_algorithm: Optional[dict[str, Any]],
+    pageserver_default_tenant_config_compaction_algorithm: Optional[Dict[str, Any]],
    pageserver_io_buffer_alignment: Optional[int],
 ) -> Iterator[NeonEnv]:
    """
@@ -1454,7 +1458,7 @@ def neon_env_builder(
    test_overlay_dir: Path,
    top_output_dir: Path,
    pageserver_virtual_file_io_engine: str,
-    pageserver_default_tenant_config_compaction_algorithm: Optional[dict[str, Any]],
+    pageserver_default_tenant_config_compaction_algorithm: Optional[Dict[str, Any]],
    pageserver_aux_file_policy: Optional[AuxFileStore],
    record_property: Callable[[str, object], None],
    pageserver_io_buffer_alignment: Optional[int],
@@ -1516,7 +1520,7 @@ class LogUtils:

    def assert_log_contains(
        self, pattern: str, offset: None | LogCursor = None
-    ) -> tuple[str, LogCursor]:
+    ) -> Tuple[str, LogCursor]:
        """Convenient for use inside wait_until()"""

        res = self.log_contains(pattern, offset=offset)
@@ -1525,7 +1529,7 @@ class LogUtils:

    def log_contains(
        self, pattern: str, offset: None | LogCursor = None
-    ) -> Optional[tuple[str, LogCursor]]:
+    ) -> Optional[Tuple[str, LogCursor]]:
        """Check that the log contains a line that matches the given regex"""
        logfile = self.logfile
        if not logfile.exists():
@@ -1606,7 +1610,7 @@ class NeonStorageController(MetricsGetter, LogUtils):
        self.running = True
        return self

-    def stop(self, immediate: bool = False) -> NeonStorageController:
+    def stop(self, immediate: bool = False) -> "NeonStorageController":
        if self.running:
            self.env.neon_cli.storage_controller_stop(immediate)
            self.running = False
@@ -1668,7 +1672,7 @@ class NeonStorageController(MetricsGetter, LogUtils):

        return resp

-    def headers(self, scope: Optional[TokenScope]) -> dict[str, str]:
+    def headers(self, scope: Optional[TokenScope]) -> Dict[str, str]:
        headers = {}
        if self.auth_enabled and scope is not None:
            jwt_token = self.env.auth_keys.generate_token(scope=scope)
@@ -1854,13 +1858,13 @@ class NeonStorageController(MetricsGetter, LogUtils):
        tenant_id: TenantId,
        shard_count: Optional[int] = None,
        shard_stripe_size: Optional[int] = None,
-        tenant_config: Optional[dict[Any, Any]] = None,
-        placement_policy: Optional[Union[dict[Any, Any] | str]] = None,
+        tenant_config: Optional[Dict[Any, Any]] = None,
+        placement_policy: Optional[Union[Dict[Any, Any] | str]] = None,
    ):
        """
        Use this rather than pageserver_api() when you need to include shard parameters
        """
-        body: dict[str, Any] = {"new_tenant_id": str(tenant_id)}
+        body: Dict[str, Any] = {"new_tenant_id": str(tenant_id)}

        if shard_count is not None:
            shard_params = {"count": shard_count}
@@ -2076,8 +2080,8 @@ class NeonStorageController(MetricsGetter, LogUtils):

                time.sleep(backoff)

-    def metadata_health_update(self, healthy: list[TenantShardId], unhealthy: list[TenantShardId]):
-        body: dict[str, Any] = {
+    def metadata_health_update(self, healthy: List[TenantShardId], unhealthy: List[TenantShardId]):
+        body: Dict[str, Any] = {
            "healthy_tenant_shards": [str(t) for t in healthy],
            "unhealthy_tenant_shards": [str(t) for t in unhealthy],
        }
@@ -2098,7 +2102,7 @@ class NeonStorageController(MetricsGetter, LogUtils):
        return response.json()

    def metadata_health_list_outdated(self, duration: str):
-        body: dict[str, Any] = {"not_scrubbed_for": duration}
+        body: Dict[str, Any] = {"not_scrubbed_for": duration}

        response = self.request(
            "POST",
@@ -2132,7 +2136,7 @@ class NeonStorageController(MetricsGetter, LogUtils):
        response.raise_for_status()
        return response.json()

-    def configure_failpoints(self, config_strings: tuple[str, str] | list[tuple[str, str]]):
+    def configure_failpoints(self, config_strings: Tuple[str, str] | List[Tuple[str, str]]):
        if isinstance(config_strings, tuple):
            pairs = [config_strings]
        else:
@@ -2149,13 +2153,13 @@ class NeonStorageController(MetricsGetter, LogUtils):
        log.info(f"Got failpoints request response code {res.status_code}")
        res.raise_for_status()

-    def get_tenants_placement(self) -> defaultdict[str, dict[str, Any]]:
+    def get_tenants_placement(self) -> defaultdict[str, Dict[str, Any]]:
        """
        Get the intent and observed placements of all tenants known to the storage controller.
        """
        tenants = self.tenant_list()

-        tenant_placement: defaultdict[str, dict[str, Any]] = defaultdict(
+        tenant_placement: defaultdict[str, Dict[str, Any]] = defaultdict(
            lambda: {
                "observed": {"attached": None, "secondary": []},
                "intent": {"attached": None, "secondary": []},
@@ -2262,12 +2266,12 @@ class NeonStorageController(MetricsGetter, LogUtils):
        response.raise_for_status()
        return [TenantShardId.parse(tid) for tid in response.json()["updated"]]

-    def __enter__(self) -> NeonStorageController:
+    def __enter__(self) -> "NeonStorageController":
        return self

    def __exit__(
        self,
-        exc_type: Optional[type[BaseException]],
+        exc_type: Optional[Type[BaseException]],
        exc: Optional[BaseException],
        tb: Optional[TracebackType],
    ):
@@ -2276,7 +2280,7 @@ class NeonStorageController(MetricsGetter, LogUtils):

 class NeonProxiedStorageController(NeonStorageController):
    def __init__(self, env: NeonEnv, proxy_port: int, auth_enabled: bool):
-        super().__init__(env, proxy_port, auth_enabled)
+        super(NeonProxiedStorageController, self).__init__(env, proxy_port, auth_enabled)
        self.instances: dict[int, dict[str, Any]] = {}

    def start(
@@ -2295,7 +2299,7 @@ class NeonProxiedStorageController(NeonStorageController):

    def stop_instance(
        self, immediate: bool = False, instance_id: Optional[int] = None
-    ) -> NeonStorageController:
+    ) -> "NeonStorageController":
        assert instance_id in self.instances
        if self.instances[instance_id]["running"]:
            self.env.neon_cli.storage_controller_stop(immediate, instance_id)
@@ -2304,7 +2308,7 @@ class NeonProxiedStorageController(NeonStorageController):
        self.running = any(meta["running"] for meta in self.instances.values())
        return self

-    def stop(self, immediate: bool = False) -> NeonStorageController:
+    def stop(self, immediate: bool = False) -> "NeonStorageController":
        for iid, details in self.instances.items():
            if details["running"]:
                self.env.neon_cli.storage_controller_stop(immediate, iid)
@@ -2323,7 +2327,7 @@ class NeonProxiedStorageController(NeonStorageController):

    def log_contains(
        self, pattern: str, offset: None | LogCursor = None
-    ) -> Optional[tuple[str, LogCursor]]:
+    ) -> Optional[Tuple[str, LogCursor]]:
        raise NotImplementedError()


@@ -2355,7 +2359,7 @@ class NeonPageserver(PgProtocol, LogUtils):
        # env.pageserver.allowed_errors.append(".*could not open garage door.*")
        #
        # The entries in the list are regular experessions.
-        self.allowed_errors: list[str] = list(DEFAULT_PAGESERVER_ALLOWED_ERRORS)
+        self.allowed_errors: List[str] = list(DEFAULT_PAGESERVER_ALLOWED_ERRORS)

    def timeline_dir(
        self,
@@ -2380,19 +2384,19 @@ class NeonPageserver(PgProtocol, LogUtils):
    def config_toml_path(self) -> Path:
        return self.workdir / "pageserver.toml"

-    def edit_config_toml(self, edit_fn: Callable[[dict[str, Any]], T]) -> T:
+    def edit_config_toml(self, edit_fn: Callable[[Dict[str, Any]], T]) -> T:
        """
        Edit the pageserver's config toml file in place.
        """
        path = self.config_toml_path
-        with open(path) as f:
+        with open(path, "r") as f:
            config = toml.load(f)
        res = edit_fn(config)
        with open(path, "w") as f:
            toml.dump(config, f)
        return res

-    def patch_config_toml_nonrecursive(self, patch: dict[str, Any]) -> dict[str, Any]:
+    def patch_config_toml_nonrecursive(self, patch: Dict[str, Any]) -> Dict[str, Any]:
        """
        Non-recursively merge the given `patch` dict into the existing config toml, using `dict.update()`.
        Returns the replaced values.
@@ -2401,7 +2405,7 @@ class NeonPageserver(PgProtocol, LogUtils):
        """
        replacements = {}

-        def doit(config: dict[str, Any]):
+        def doit(config: Dict[str, Any]):
            while len(patch) > 0:
                key, new = patch.popitem()
                old = config.get(key, None)
@@ -2413,9 +2417,9 @@ class NeonPageserver(PgProtocol, LogUtils):

    def start(
        self,
-        extra_env_vars: Optional[dict[str, str]] = None,
+        extra_env_vars: Optional[Dict[str, str]] = None,
        timeout_in_seconds: Optional[int] = None,
-    ) -> NeonPageserver:
+    ) -> "NeonPageserver":
        """
        Start the page server.
        `overrides` allows to add some config to this pageserver start.
@@ -2441,7 +2445,7 @@ class NeonPageserver(PgProtocol, LogUtils):

        return self

-    def stop(self, immediate: bool = False) -> NeonPageserver:
+    def stop(self, immediate: bool = False) -> "NeonPageserver":
        """
        Stop the page server.
        Returns self.
@@ -2489,12 +2493,12 @@ class NeonPageserver(PgProtocol, LogUtils):

        wait_until(20, 0.5, complete)

-    def __enter__(self) -> NeonPageserver:
+    def __enter__(self) -> "NeonPageserver":
        return self

    def __exit__(
        self,
-        exc_type: Optional[type[BaseException]],
+        exc_type: Optional[Type[BaseException]],
        exc: Optional[BaseException],
        tb: Optional[TracebackType],
    ):
@@ -2541,7 +2545,7 @@ class NeonPageserver(PgProtocol, LogUtils):
    def tenant_attach(
        self,
        tenant_id: TenantId,
-        config: None | dict[str, Any] = None,
+        config: None | Dict[str, Any] = None,
        generation: Optional[int] = None,
        override_storage_controller_generation: bool = False,
    ):
@@ -2580,7 +2584,7 @@ class NeonPageserver(PgProtocol, LogUtils):
    ) -> dict[str, Any]:
        path = self.tenant_dir(tenant_shard_id) / "config-v1"
        log.info(f"Reading location conf from {path}")
-        bytes = open(path).read()
+        bytes = open(path, "r").read()
        try:
            decoded: dict[str, Any] = toml.loads(bytes)
            return decoded
@@ -2591,7 +2595,7 @@ class NeonPageserver(PgProtocol, LogUtils):
    def tenant_create(
        self,
        tenant_id: TenantId,
-        conf: Optional[dict[str, Any]] = None,
+        conf: Optional[Dict[str, Any]] = None,
        auth_token: Optional[str] = None,
        generation: Optional[int] = None,
    ) -> TenantId:
@@ -2657,7 +2661,7 @@ class PgBin:
        self.env = os.environ.copy()
        self.env["LD_LIBRARY_PATH"] = str(self.pg_lib_dir)

-    def _fixpath(self, command: list[str]):
+    def _fixpath(self, command: List[str]):
        if "/" not in str(command[0]):
            command[0] = str(self.pg_bin_path / command[0])

@@ -2677,7 +2681,7 @@ class PgBin:

    def run_nonblocking(
        self,
-        command: list[str],
+        command: List[str],
        env: Optional[Env] = None,
        cwd: Optional[Union[str, Path]] = None,
    ) -> subprocess.Popen[Any]:
@@ -2701,7 +2705,7 @@ class PgBin:

    def run(
        self,
-        command: list[str],
+        command: List[str],
        env: Optional[Env] = None,
        cwd: Optional[Union[str, Path]] = None,
    ) -> None:
@@ -2724,7 +2728,7 @@ class PgBin:

    def run_capture(
        self,
-        command: list[str],
+        command: List[str],
        env: Optional[Env] = None,
        cwd: Optional[str] = None,
        with_command_header=True,
@@ -2837,14 +2841,14 @@ class VanillaPostgres(PgProtocol):
            ]
        )

-    def configure(self, options: list[str]):
+    def configure(self, options: List[str]):
        """Append lines into postgresql.conf file."""
        assert not self.running
        with open(os.path.join(self.pgdatadir, "postgresql.conf"), "a") as conf_file:
            conf_file.write("\n".join(options))
            conf_file.write("\n")

-    def edit_hba(self, hba: list[str]):
+    def edit_hba(self, hba: List[str]):
        """Prepend hba lines into pg_hba.conf file."""
        assert not self.running
        with open(os.path.join(self.pgdatadir, "pg_hba.conf"), "r+") as conf_file:
@@ -2872,12 +2876,12 @@ class VanillaPostgres(PgProtocol):
        """Return size of pgdatadir subdirectory in bytes."""
        return get_dir_size(self.pgdatadir / subdir)

-    def __enter__(self) -> VanillaPostgres:
+    def __enter__(self) -> "VanillaPostgres":
        return self

    def __exit__(
        self,
-        exc_type: Optional[type[BaseException]],
+        exc_type: Optional[Type[BaseException]],
        exc: Optional[BaseException],
        tb: Optional[TracebackType],
    ):
@@ -2907,7 +2911,7 @@ class RemotePostgres(PgProtocol):
        # The remote server is assumed to be running already
        self.running = True

-    def configure(self, options: list[str]):
+    def configure(self, options: List[str]):
        raise Exception("cannot change configuration of remote Posgres instance")

    def start(self):
@@ -2921,12 +2925,12 @@ class RemotePostgres(PgProtocol):
        # See https://www.postgresql.org/docs/14/functions-admin.html#FUNCTIONS-ADMIN-GENFILE
        raise Exception("cannot get size of a Postgres instance")

-    def __enter__(self) -> RemotePostgres:
+    def __enter__(self) -> "RemotePostgres":
        return self

    def __exit__(
        self,
-        exc_type: Optional[type[BaseException]],
+        exc_type: Optional[Type[BaseException]],
        exc: Optional[BaseException],
        tb: Optional[TracebackType],
    ):
@@ -3262,7 +3266,7 @@ class NeonProxy(PgProtocol):

    def __exit__(
        self,
-        exc_type: Optional[type[BaseException]],
+        exc_type: Optional[Type[BaseException]],
        exc: Optional[BaseException],
        tb: Optional[TracebackType],
    ):
@@ -3400,7 +3404,7 @@ class Endpoint(PgProtocol, LogUtils):
        self.http_port = http_port
        self.check_stop_result = check_stop_result
        # passed to endpoint create and endpoint reconfigure
-        self.active_safekeepers: list[int] = list(map(lambda sk: sk.id, env.safekeepers))
+        self.active_safekeepers: List[int] = list(map(lambda sk: sk.id, env.safekeepers))
        # path to conf is <repo_dir>/endpoints/<endpoint_id>/pgdata/postgresql.conf

        # Semaphore is set to 1 when we start, and acquire'd back to zero when we stop
@@ -3423,10 +3427,10 @@ class Endpoint(PgProtocol, LogUtils):
        endpoint_id: Optional[str] = None,
        hot_standby: bool = False,
        lsn: Optional[Lsn] = None,
-        config_lines: Optional[list[str]] = None,
+        config_lines: Optional[List[str]] = None,
        pageserver_id: Optional[int] = None,
        allow_multiple: bool = False,
-    ) -> Endpoint:
+    ) -> "Endpoint":
        """
        Create a new Postgres endpoint.
        Returns self.
@@ -3469,10 +3473,10 @@ class Endpoint(PgProtocol, LogUtils):
        self,
        remote_ext_config: Optional[str] = None,
        pageserver_id: Optional[int] = None,
-        safekeepers: Optional[list[int]] = None,
+        safekeepers: Optional[List[int]] = None,
        allow_multiple: bool = False,
        basebackup_request_tries: Optional[int] = None,
-    ) -> Endpoint:
+    ) -> "Endpoint":
        """
        Start the Postgres instance.
        Returns self.
@@ -3485,6 +3489,8 @@ class Endpoint(PgProtocol, LogUtils):
        if safekeepers is not None:
            self.active_safekeepers = safekeepers

+        log.info(f"Starting postgres endpoint {self.endpoint_id}")
+
        self.env.neon_cli.endpoint_start(
            self.endpoint_id,
            safekeepers=self.active_safekeepers,
@@ -3519,7 +3525,7 @@ class Endpoint(PgProtocol, LogUtils):
        """Path to the postgresql.conf in the endpoint directory (not the one in pgdata)"""
        return self.endpoint_path() / "postgresql.conf"

-    def config(self, lines: list[str]) -> Endpoint:
+    def config(self, lines: List[str]) -> "Endpoint":
        """
        Add lines to postgresql.conf.
        Lines should be an array of valid postgresql.conf rows.
@@ -3533,7 +3539,7 @@ class Endpoint(PgProtocol, LogUtils):

        return self

-    def edit_hba(self, hba: list[str]):
+    def edit_hba(self, hba: List[str]):
        """Prepend hba lines into pg_hba.conf file."""
        with open(os.path.join(self.pg_data_dir_path(), "pg_hba.conf"), "r+") as conf_file:
            data = conf_file.read()
@@ -3548,7 +3554,7 @@ class Endpoint(PgProtocol, LogUtils):
        return self._running._value > 0

    def reconfigure(
-        self, pageserver_id: Optional[int] = None, safekeepers: Optional[list[int]] = None
+        self, pageserver_id: Optional[int] = None, safekeepers: Optional[List[int]] = None
    ):
        assert self.endpoint_id is not None
        # If `safekeepers` is not None, they are remember them as active and use
@@ -3563,7 +3569,7 @@ class Endpoint(PgProtocol, LogUtils):
        """Update the endpoint.json file used by control_plane."""
        # Read config
        config_path = os.path.join(self.endpoint_path(), "endpoint.json")
-        with open(config_path) as f:
+        with open(config_path, "r") as f:
            data_dict: dict[str, Any] = json.load(f)

        # Write it back updated
@@ -3596,8 +3602,8 @@ class Endpoint(PgProtocol, LogUtils):
    def stop(
        self,
        mode: str = "fast",
-        sks_wait_walreceiver_gone: Optional[tuple[list[Safekeeper], TimelineId]] = None,
-    ) -> Endpoint:
+        sks_wait_walreceiver_gone: Optional[tuple[List[Safekeeper], TimelineId]] = None,
+    ) -> "Endpoint":
        """
        Stop the Postgres instance if it's running.

@@ -3631,7 +3637,7 @@ class Endpoint(PgProtocol, LogUtils):

        return self

-    def stop_and_destroy(self, mode: str = "immediate") -> Endpoint:
+    def stop_and_destroy(self, mode: str = "immediate") -> "Endpoint":
        """
        Stop the Postgres instance, then destroy the endpoint.
        Returns self.
@@ -3653,17 +3659,19 @@ class Endpoint(PgProtocol, LogUtils):
        endpoint_id: Optional[str] = None,
        hot_standby: bool = False,
        lsn: Optional[Lsn] = None,
-        config_lines: Optional[list[str]] = None,
+        config_lines: Optional[List[str]] = None,
        remote_ext_config: Optional[str] = None,
        pageserver_id: Optional[int] = None,
        allow_multiple=False,
        basebackup_request_tries: Optional[int] = None,
-    ) -> Endpoint:
+    ) -> "Endpoint":
        """
        Create an endpoint, apply config, and start Postgres.
        Returns self.
        """

+        started_at = time.time()
+
        self.create(
            branch_name=branch_name,
            endpoint_id=endpoint_id,
@@ -3679,14 +3687,16 @@ class Endpoint(PgProtocol, LogUtils):
            basebackup_request_tries=basebackup_request_tries,
        )

+        log.info(f"Postgres startup took {time.time() - started_at} seconds")
+
        return self

-    def __enter__(self) -> Endpoint:
+    def __enter__(self) -> "Endpoint":
        return self

    def __exit__(
        self,
-        exc_type: Optional[type[BaseException]],
+        exc_type: Optional[Type[BaseException]],
        exc: Optional[BaseException],
        tb: Optional[TracebackType],
    ):
@@ -3717,7 +3727,7 @@ class EndpointFactory:
    def __init__(self, env: NeonEnv):
        self.env = env
        self.num_instances: int = 0
-        self.endpoints: list[Endpoint] = []
+        self.endpoints: List[Endpoint] = []

    def create_start(
        self,
@@ -3726,7 +3736,7 @@ class EndpointFactory:
        tenant_id: Optional[TenantId] = None,
        lsn: Optional[Lsn] = None,
        hot_standby: bool = False,
-        config_lines: Optional[list[str]] = None,
+        config_lines: Optional[List[str]] = None,
        remote_ext_config: Optional[str] = None,
        pageserver_id: Optional[int] = None,
        basebackup_request_tries: Optional[int] = None,
@@ -3758,7 +3768,7 @@ class EndpointFactory:
        tenant_id: Optional[TenantId] = None,
        lsn: Optional[Lsn] = None,
        hot_standby: bool = False,
-        config_lines: Optional[list[str]] = None,
+        config_lines: Optional[List[str]] = None,
        pageserver_id: Optional[int] = None,
    ) -> Endpoint:
        ep = Endpoint(
@@ -3782,7 +3792,7 @@ class EndpointFactory:
            pageserver_id=pageserver_id,
        )

-    def stop_all(self, fail_on_error=True) -> EndpointFactory:
+    def stop_all(self, fail_on_error=True) -> "EndpointFactory":
        exception = None
        for ep in self.endpoints:
            try:
@@ -3797,7 +3807,7 @@ class EndpointFactory:
        return self

    def new_replica(
-        self, origin: Endpoint, endpoint_id: str, config_lines: Optional[list[str]] = None
+        self, origin: Endpoint, endpoint_id: str, config_lines: Optional[List[str]] = None
    ):
        branch_name = origin.branch_name
        assert origin in self.endpoints
@@ -3813,7 +3823,7 @@ class EndpointFactory:
        )

    def new_replica_start(
-        self, origin: Endpoint, endpoint_id: str, config_lines: Optional[list[str]] = None
+        self, origin: Endpoint, endpoint_id: str, config_lines: Optional[List[str]] = None
    ):
        branch_name = origin.branch_name
        assert origin in self.endpoints
@@ -3851,7 +3861,7 @@ class Safekeeper(LogUtils):
        port: SafekeeperPort,
        id: int,
        running: bool = False,
-        extra_opts: Optional[list[str]] = None,
+        extra_opts: Optional[List[str]] = None,
    ):
        self.env = env
        self.port = port
@@ -3877,8 +3887,8 @@ class Safekeeper(LogUtils):
        self.extra_opts = extra_opts

    def start(
-        self, extra_opts: Optional[list[str]] = None, timeout_in_seconds: Optional[int] = None
-    ) -> Safekeeper:
+        self, extra_opts: Optional[List[str]] = None, timeout_in_seconds: Optional[int] = None
+    ) -> "Safekeeper":
        if extra_opts is None:
            # Apply either the extra_opts passed in, or the ones from our constructor: we do not merge the two.
            extra_opts = self.extra_opts
@@ -3913,7 +3923,7 @@ class Safekeeper(LogUtils):
                break  # success
        return self

-    def stop(self, immediate: bool = False) -> Safekeeper:
+    def stop(self, immediate: bool = False) -> "Safekeeper":
        log.info(f"Stopping safekeeper {self.id}")
        self.env.neon_cli.safekeeper_stop(self.id, immediate)
        self.running = False
@@ -3925,8 +3935,8 @@ class Safekeeper(LogUtils):
        assert not self.log_contains("timeout while acquiring WalResidentTimeline guard")

    def append_logical_message(
-        self, tenant_id: TenantId, timeline_id: TimelineId, request: dict[str, Any]
-    ) -> dict[str, Any]:
+        self, tenant_id: TenantId, timeline_id: TimelineId, request: Dict[str, Any]
+    ) -> Dict[str, Any]:
        """
        Send JSON_CTRL query to append LogicalMessage to WAL and modify
        safekeeper state. It will construct LogicalMessage from provided
@@ -3979,7 +3989,7 @@ class Safekeeper(LogUtils):

    def pull_timeline(
        self, srcs: list[Safekeeper], tenant_id: TenantId, timeline_id: TimelineId
-    ) -> dict[str, Any]:
+    ) -> Dict[str, Any]:
        """
        pull_timeline from srcs to self.
        """
@@ -4015,7 +4025,7 @@ class Safekeeper(LogUtils):
        mysegs = [s for s in segs if f"sk{self.id}" in s]
        return mysegs

-    def list_segments(self, tenant_id, timeline_id) -> list[str]:
+    def list_segments(self, tenant_id, timeline_id) -> List[str]:
        """
        Get list of segment names of the given timeline.
        """
@@ -4120,7 +4130,7 @@ class StorageScrubber:
        self.log_dir = log_dir

    def scrubber_cli(
-        self, args: list[str], timeout, extra_env: Optional[dict[str, str]] = None
+        self, args: list[str], timeout, extra_env: Optional[Dict[str, str]] = None
    ) -> str:
        assert isinstance(self.env.pageserver_remote_storage, S3Storage)
        s3_storage = self.env.pageserver_remote_storage
@@ -4167,10 +4177,10 @@ class StorageScrubber:

    def scan_metadata_safekeeper(
        self,
-        timeline_lsns: list[dict[str, Any]],
+        timeline_lsns: List[Dict[str, Any]],
        cloud_admin_api_url: str,
        cloud_admin_api_token: str,
-    ) -> tuple[bool, Any]:
+    ) -> Tuple[bool, Any]:
        extra_env = {
            "CLOUD_ADMIN_API_URL": cloud_admin_api_url,
            "CLOUD_ADMIN_API_TOKEN": cloud_admin_api_token,
@@ -4183,9 +4193,9 @@ class StorageScrubber:
        self,
        post_to_storage_controller: bool = False,
        node_kind: NodeKind = NodeKind.PAGESERVER,
-        timeline_lsns: Optional[list[dict[str, Any]]] = None,
-        extra_env: Optional[dict[str, str]] = None,
-    ) -> tuple[bool, Any]:
+        timeline_lsns: Optional[List[Dict[str, Any]]] = None,
+        extra_env: Optional[Dict[str, str]] = None,
+    ) -> Tuple[bool, Any]:
        """
        Returns the health status and the metadata summary.
        """
@@ -4492,7 +4502,7 @@ def should_skip_file(filename: str) -> bool:
 #
 # Test helpers
 #
-def list_files_to_compare(pgdata_dir: Path) -> list[str]:
+def list_files_to_compare(pgdata_dir: Path) -> List[str]:
    pgdata_files = []
    for root, _dirs, filenames in os.walk(pgdata_dir):
        for filename in filenames:
--- a/test_runner/fixtures/overlayfs.py
+++ b/test_runner/fixtures/overlayfs.py
@@ -1,7 +1,5 @@
-from __future__ import annotations
-
-from collections.abc import Iterator
 from pathlib import Path
+from typing import Iterator

 import psutil

--- a/test_runner/fixtures/pageserver/init.py
+++ b/test_runner/fixtures/pageserver/init.py
@@ -1 +0,0 @@
-from __future__ import annotations
--- a/Show More
+++ b/Show More