fix(compute_ctl): set stripe size before connstring during respec

Signed-off-by: Alex Chi Z <chi@neon.tech>
2026-07-08 14:40:37 +00:00 · 2025-07-09 11:15:36 -04:00
112 changed files with 1001 additions and 3387 deletions
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1009,12 +1009,6 @@ dependencies = [
 "generic-array",
 ]

-[[package]]
-name = "boxcar"
-version = "0.2.13"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "26c4925bc979b677330a8c7fe7a8c94af2dbb4a2d37b4a20a80d884400f46baa"
-
 [[package]]
 name = "bstr"
 version = "1.5.0"
@@ -1248,14 +1242,14 @@ checksum = "afb84c814227b90d6895e01398aee0d8033c00e7466aca416fb6a8e0eb19d8a7"

 [[package]]
 name = "clashmap"
-version = "1.2.0"
+version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e8a055b1f1bf558eae4959f6dd77cf2d7d50ae1483928e60ef21ca5a24fd4321"
+checksum = "93bd59c81e2bd87a775ae2de75f070f7e2bfe97363a6ad652f46824564c23e4d"
 dependencies = [
 "crossbeam-utils",
 "hashbrown 0.15.2",
 "lock_api",
- "parking_lot_core 0.9.10",
+ "parking_lot_core 0.9.8",
 "polonius-the-crab",
 "replace_with",
 ]
@@ -1759,6 +1753,19 @@ dependencies = [
 "syn 2.0.100",
 ]

+[[package]]
+name = "dashmap"
+version = "5.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6943ae99c34386c84a470c499d3414f66502a41340aa895406e0d2e4a207b91d"
+dependencies = [
+ "cfg-if",
+ "hashbrown 0.14.5",
+ "lock_api",
+ "once_cell",
+ "parking_lot_core 0.9.8",
+]
+
 [[package]]
 name = "dashmap"
 version = "6.1.0"
@@ -1770,7 +1777,7 @@ dependencies = [
 "hashbrown 0.14.5",
 "lock_api",
 "once_cell",
- "parking_lot_core 0.9.10",
+ "parking_lot_core 0.9.8",
 ]

 [[package]]
@@ -2336,12 +2343,6 @@ version = "1.0.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"

-[[package]]
-name = "foldhash"
-version = "0.1.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2"
-
 [[package]]
 name = "form_urlencoded"
 version = "1.2.1"
@@ -2581,7 +2582,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "842dc78579ce01e6a1576ad896edc92fca002dd60c9c3746b7fc2bec6fb429d0"
 dependencies = [
 "cfg-if",
- "dashmap",
+ "dashmap 6.1.0",
 "futures-sink",
 "futures-timer",
 "futures-util",
@@ -3287,7 +3288,7 @@ dependencies = [
 "clap",
 "crossbeam-channel",
 "crossbeam-utils",
- "dashmap",
+ "dashmap 6.1.0",
 "env_logger",
 "indexmap 2.9.0",
 "itoa",
@@ -3544,6 +3545,16 @@ dependencies = [
 "libc",
 ]

+[[package]]
+name = "lasso"
+version = "0.7.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4644821e1c3d7a560fe13d842d13f587c07348a1a05d3a797152d41c90c56df2"
+dependencies = [
+ "dashmap 5.5.0",
+ "hashbrown 0.13.2",
+]
+
 [[package]]
 name = "lazy_static"
 version = "1.5.0"
@@ -3663,17 +3674,17 @@ checksum = "490cc448043f947bae3cbee9c203358d62dbee0db12107a74be5c30ccfd09771"

 [[package]]
 name = "measured"
-version = "0.0.23"
+version = "0.0.22"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d22ae866c28b9c59afaeb488ad6e3bd148570cf5a2bacf6c4845def5b9a03470"
+checksum = "3051f3a030d55d680cdef6ca50e80abd1182f8da29f2344a7c9cb575721138f0"
 dependencies = [
 "bytes",
 "crossbeam-utils",
 "hashbrown 0.14.5",
 "itoa",
+ "lasso",
 "measured-derive",
 "memchr",
- "paracord",
 "parking_lot 0.12.1",
 "rustc-hash 1.1.0",
 "ryu",
@@ -3681,9 +3692,9 @@ dependencies = [

 [[package]]
 name = "measured-derive"
-version = "0.0.23"
+version = "0.0.22"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "16d734ed9dbca22d87d56b54d220f254ce921cb5cce97d4a960075af0131d076"
+checksum = "b9e6777fc80a575f9503d908c8b498782a6c3ee88a06cb416dc3941401e43b94"
 dependencies = [
 "heck",
 "proc-macro2",
@@ -3693,26 +3704,15 @@ dependencies = [

 [[package]]
 name = "measured-process"
-version = "0.0.23"
+version = "0.0.22"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f71a318d2b9edcded1e5e0ccf3c9e1e1614217d7f07933631c771daa717743aa"
+checksum = "7c4b80445aeb08e832d87bf1830049a924cdc1d6b7ef40b6b9b365bff17bf8ec"
 dependencies = [
 "libc",
 "measured",
 "procfs",
 ]

-[[package]]
-name = "measured-tokio"
-version = "0.0.23"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5e79a936051c484268e2d71a5dd01219096c62c8ff09afad95f07c14a3b0c580"
-dependencies = [
- "itoa",
- "measured",
- "tokio",
-]
-
 [[package]]
 name = "memchr"
 version = "2.6.4"
@@ -4294,7 +4294,6 @@ dependencies = [
 "humantime-serde",
 "pageserver_api",
 "pageserver_client",
- "pageserver_client_grpc",
 "pageserver_page_api",
 "rand 0.8.5",
 "reqwest",
@@ -4324,7 +4323,6 @@ dependencies = [
 "pageserver_api",
 "postgres_ffi",
 "remote_storage",
- "serde",
 "serde_json",
 "svg_fmt",
 "thiserror 1.0.69",
@@ -4501,7 +4499,6 @@ name = "pageserver_client_grpc"
 version = "0.1.0"
 dependencies = [
 "anyhow",
- "arc-swap",
 "bytes",
 "compute_api",
 "futures",
@@ -4509,7 +4506,6 @@ dependencies = [
 "pageserver_page_api",
 "tokio",
 "tokio-stream",
- "tokio-util",
 "tonic 0.13.1",
 "tracing",
 "utils",
@@ -4571,21 +4567,6 @@ dependencies = [
 "seize",
 ]

-[[package]]
-name = "paracord"
-version = "0.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5e41e924113f9a05eecd561c6e695648f243a03b3ad8d9b3eff689342b95023b"
-dependencies = [
- "boxcar",
- "clashmap",
- "foldhash",
- "hashbrown 0.15.2",
- "serde",
- "sync_wrapper 1.0.1",
- "typed-arena",
-]
-
 [[package]]
 name = "parking"
 version = "2.1.1"
@@ -4610,7 +4591,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3742b2c103b9f06bc9fff0a37ff4912935851bee6d36f3c02bcc755bcfec228f"
 dependencies = [
 "lock_api",
- "parking_lot_core 0.9.10",
+ "parking_lot_core 0.9.8",
 ]

 [[package]]
@@ -4629,15 +4610,15 @@ dependencies = [

 [[package]]
 name = "parking_lot_core"
-version = "0.9.10"
+version = "0.9.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1e401f977ab385c9e4e3ab30627d6f26d00e2c73eef317493c4ec6d468726cf8"
+checksum = "93f00c865fe7cabf650081affecd3871070f26767e7b2070a3ffae14c654b447"
 dependencies = [
 "cfg-if",
 "libc",
- "redox_syscall 0.5.10",
+ "redox_syscall 0.3.5",
 "smallvec",
- "windows-targets 0.52.6",
+ "windows-targets 0.48.0",
 ]

 [[package]]
@@ -5343,14 +5324,13 @@ dependencies = [
 "itoa",
 "jose-jwa",
 "jose-jwk",
+ "lasso",
 "measured",
- "measured-tokio",
 "metrics",
 "once_cell",
 "opentelemetry",
 "p256 0.13.2",
 "papaya",
- "paracord",
 "parking_lot 0.12.1",
 "parquet",
 "parquet_derive",
@@ -5653,15 +5633,6 @@ dependencies = [
 "bitflags 1.3.2",
 ]

-[[package]]
-name = "redox_syscall"
-version = "0.5.10"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0b8c0c260b63a8219631167be35e6a988e9554dbd323f8bd08439c8ed1302bd1"
-dependencies = [
- "bitflags 2.8.0",
-]
-
 [[package]]
 name = "regex"
 version = "1.10.2"
@@ -6922,12 +6893,12 @@ dependencies = [
 "hyper 0.14.30",
 "itertools 0.10.5",
 "json-structural-diff",
+ "lasso",
 "measured",
 "metrics",
 "once_cell",
 "pageserver_api",
 "pageserver_client",
- "paracord",
 "postgres_connection",
 "posthog_client_lite",
 "rand 0.8.5",
@@ -7016,7 +6987,6 @@ dependencies = [
 "pageserver_api",
 "pageserver_client",
 "reqwest",
- "safekeeper_api",
 "serde_json",
 "storage_controller_client",
 "tokio",
@@ -8042,12 +8012,6 @@ dependencies = [
 "static_assertions",
 ]

-[[package]]
-name = "typed-arena"
-version = "2.0.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6af6ae20167a9ece4bcb41af5b80f8a1f1df981f6391189ce00fd257af04126a"
-
 [[package]]
 name = "typed-json"
 version = "0.1.1"
@@ -8792,7 +8756,6 @@ dependencies = [
 "num-traits",
 "once_cell",
 "p256 0.13.2",
- "paracord",
 "parquet",
 "prettyplease",
 "proc-macro2",
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -128,11 +128,11 @@ itertools = "0.10"
 itoa = "1.0.11"
 jemalloc_pprof = { version = "0.7", features = ["symbolize", "flamegraph"] }
 jsonwebtoken = "9"
+lasso = "0.7"
 libc = "0.2"
 md5 = "0.7.0"
-measured = { version = "0.0.23", features = ["paracord"] }
-measured-process = { version = "0.0.23" }
-measured-tokio = { version = "0.0.23" }
+measured = { version = "0.0.22", features=["lasso"] }
+measured-process = { version = "0.0.22" }
 memoffset = "0.9"
 nix = { version = "0.30.1", features = ["dir", "fs", "mman", "process", "socket", "signal", "poll"] }
 # Do not update to >= 7.0.0, at least. The update will have a significant impact
@@ -145,7 +145,6 @@ opentelemetry = "0.27"
 opentelemetry_sdk = "0.27"
 opentelemetry-otlp = { version = "0.27", default-features = false, features = ["http-proto", "trace", "http", "reqwest-client"] }
 opentelemetry-semantic-conventions = "0.27"
-paracord = { version = "0.1.0", features = ["serde"] }
 parking_lot = "0.12"
 parquet = { version = "53", default-features = false, features = ["zstd"] }
 parquet_derive = "53"
@@ -263,7 +262,6 @@ neon-shmem = { version = "0.1", path = "./libs/neon-shmem/" }
 pageserver = { path = "./pageserver" }
 pageserver_api = { version = "0.1", path = "./libs/pageserver_api/" }
 pageserver_client = { path = "./pageserver/client" }
-pageserver_client_grpc = { path = "./pageserver/client_grpc" }
 pageserver_compaction = { version = "0.1", path = "./pageserver/compaction/" }
 pageserver_page_api = { path = "./pageserver/page_api" }
 postgres_backend = { version = "0.1", path = "./libs/postgres_backend/" }
--- a/compute_tools/README.md
+++ b/compute_tools/README.md
@@ -46,14 +46,11 @@ stateDiagram-v2
  Configuration --> Failed : Failed to configure the compute
  Configuration --> Running : Compute has been configured
  Empty --> Init : Compute spec is immediately available
-  Empty --> TerminationPendingFast : Requested termination
-  Empty --> TerminationPendingImmediate : Requested termination
+  Empty --> TerminationPending : Requested termination
  Init --> Failed : Failed to start Postgres
  Init --> Running : Started Postgres
-  Running --> TerminationPendingFast : Requested termination
-  Running --> TerminationPendingImmediate : Requested termination
-  TerminationPendingFast --> Terminated compute with 30s delay for cplane to inspect status
-  TerminationPendingImmediate --> Terminated : Terminated compute immediately
+  Running --> TerminationPending : Requested termination
+  TerminationPending --> Terminated : Terminated compute
  Failed --> [*] : Compute exited
  Terminated --> [*] : Compute exited
 ```
--- a/compute_tools/src/compute.rs
+++ b/compute_tools/src/compute.rs
@@ -956,20 +956,14 @@ impl ComputeNode {
            None
        };

+        let mut delay_exit = false;
        let mut state = self.state.lock().unwrap();
        state.terminate_flush_lsn = lsn;
-
-        let delay_exit = state.status == ComputeStatus::TerminationPendingFast;
-        if state.status == ComputeStatus::TerminationPendingFast
-            || state.status == ComputeStatus::TerminationPendingImmediate
-        {
-            info!(
-                "Changing compute status from {} to {}",
-                state.status,
-                ComputeStatus::Terminated
-            );
+        if let ComputeStatus::TerminationPending { mode } = state.status {
            state.status = ComputeStatus::Terminated;
            self.state_changed.notify_all();
+            // we were asked to terminate gracefully, don't exit to avoid restart
+            delay_exit = mode == compute_api::responses::TerminateMode::Fast
        }
        drop(state);

@@ -1811,8 +1805,6 @@ impl ComputeNode {
            tls_config,
        )?;

-        self.pg_reload_conf()?;
-
        if !spec.skip_pg_catalog_updates {
            let max_concurrent_connections = spec.reconfigure_concurrency;
            // Temporarily reset max_cluster_size in config
@@ -1832,9 +1824,10 @@ impl ComputeNode {

                Ok(())
            })?;
-            self.pg_reload_conf()?;
        }

+        self.pg_reload_conf()?;
+
        let unknown_op = "unknown".to_string();
        let op_id = spec.operation_uuid.as_ref().unwrap_or(&unknown_op);
        info!(
@@ -1907,8 +1900,7 @@ impl ComputeNode {

                            // exit loop
                            ComputeStatus::Failed
-                            | ComputeStatus::TerminationPendingFast
-                            | ComputeStatus::TerminationPendingImmediate
+                            | ComputeStatus::TerminationPending { .. }
                            | ComputeStatus::Terminated => break 'cert_update,

                            // wait
--- a/compute_tools/src/compute_prewarm.rs
+++ b/compute_tools/src/compute_prewarm.rs
@@ -70,7 +70,7 @@ impl ComputeNode {
            }
        };
        let row = match client
-            .query_one("select * from neon.get_prewarm_info()", &[])
+            .query_one("select * from get_prewarm_info()", &[])
            .await
        {
            Ok(row) => row,
@@ -146,7 +146,7 @@ impl ComputeNode {
        ComputeNode::get_maintenance_client(&self.tokio_conn_conf)
            .await
            .context("connecting to postgres")?
-            .query_one("select neon.prewarm_local_cache($1)", &[&uncompressed])
+            .query_one("select prewarm_local_cache($1)", &[&uncompressed])
            .await
            .context("loading LFC state into postgres")
            .map(|_| ())
@@ -196,7 +196,7 @@ impl ComputeNode {
        ComputeNode::get_maintenance_client(&self.tokio_conn_conf)
            .await
            .context("connecting to postgres")?
-            .query_one("select neon.get_local_cache_state()", &[])
+            .query_one("select get_local_cache_state()", &[])
            .await
            .context("querying LFC state")?
            .try_get::<usize, &[u8]>(0)
--- a/compute_tools/src/config.rs
+++ b/compute_tools/src/config.rs
@@ -56,12 +56,15 @@ pub fn write_postgres_conf(

    // Add options for connecting to storage
    writeln!(file, "# Neon storage settings")?;
-    if let Some(s) = &spec.pageserver_connstring {
-        writeln!(file, "neon.pageserver_connstring={}", escape_conf_value(s))?;
-    }
+    // Setting stripe size before connstring is required to avoid compute routing keys to the wrong shard.
+    // Otherwise, if we set the connstring first and then stripe size, there will be a very small window
+    // where the default stripe size is used and the connstring is set.
    if let Some(stripe_size) = spec.shard_stripe_size {
        writeln!(file, "neon.stripe_size={stripe_size}")?;
    }
+    if let Some(s) = &spec.pageserver_connstring {
+        writeln!(file, "neon.pageserver_connstring={}", escape_conf_value(s))?;
+    }
    if !spec.safekeeper_connstrings.is_empty() {
        let mut neon_safekeepers_value = String::new();
        tracing::info!(
--- a/compute_tools/src/http/openapi_spec.yaml
+++ b/compute_tools/src/http/openapi_spec.yaml
@@ -371,28 +371,9 @@ paths:
      summary: Terminate Postgres and wait for it to exit
      description: ""
      operationId: terminate
-      parameters:
-        - name: mode
-          in: query
-          description: "Terminate mode: fast (wait 30s before returning) and immediate"
-          required: false
-          schema:
-            type: string
-            enum: ["fast", "immediate"]
-            default: fast
      responses:
        200:
          description: Result
-          content:
-            application/json:
-              schema:
-                $ref: "#/components/schemas/TerminateResponse"
-        201:
-          description: Result if compute is already terminated
-          content:
-            application/json:
-              schema:
-                $ref: "#/components/schemas/TerminateResponse"
        412:
          description: "wrong state"
          content:
@@ -549,14 +530,11 @@ components:
      type: string
      enum:
        - empty
-        - configuration_pending
        - init
-        - running
-        - configuration
        - failed
-        - termination_pending_fast
-        - termination_pending_immediate
-        - terminated
+        - running
+        - configuration_pending
+        - configuration
      example: running

    ExtensionInstallRequest:
@@ -682,17 +660,6 @@ components:
          description: Role name.
          example: "neon"

-    TerminateResponse:
-      type: object
-      required:
-        - lsn
-      properties:
-        lsn:
-          type: string
-          nullable: true
-          description: "last WAL flush LSN"
-          example: "0/028F10D8"
-
    SetRoleGrantsResponse:
      type: object
      required:
--- a/compute_tools/src/http/routes/terminate.rs
+++ b/compute_tools/src/http/routes/terminate.rs
@@ -3,7 +3,7 @@ use crate::http::JsonResponse;
 use axum::extract::State;
 use axum::response::Response;
 use axum_extra::extract::OptionalQuery;
-use compute_api::responses::{ComputeStatus, TerminateMode, TerminateResponse};
+use compute_api::responses::{ComputeStatus, TerminateResponse};
 use http::StatusCode;
 use serde::Deserialize;
 use std::sync::Arc;
@@ -12,7 +12,7 @@ use tracing::info;

 #[derive(Deserialize, Default)]
 pub struct TerminateQuery {
-    mode: TerminateMode,
+    mode: compute_api::responses::TerminateMode,
 }

 /// Terminate the compute.
@@ -24,16 +24,16 @@ pub(in crate::http) async fn terminate(
    {
        let mut state = compute.state.lock().unwrap();
        if state.status == ComputeStatus::Terminated {
-            let response = TerminateResponse {
-                lsn: state.terminate_flush_lsn,
-            };
-            return JsonResponse::success(StatusCode::CREATED, response);
+            return JsonResponse::success(StatusCode::CREATED, state.terminate_flush_lsn);
        }

        if !matches!(state.status, ComputeStatus::Empty | ComputeStatus::Running) {
            return JsonResponse::invalid_status(state.status);
        }
-        state.set_status(mode.into(), &compute.state_changed);
+        state.set_status(
+            ComputeStatus::TerminationPending { mode },
+            &compute.state_changed,
+        );
    }

    forward_termination_signal(false);
--- a/compute_tools/src/metrics.rs
+++ b/compute_tools/src/metrics.rs
@@ -108,7 +108,7 @@ pub(crate) static LFC_PREWARMS: Lazy<IntCounter> = Lazy::new(|| {
 pub(crate) static LFC_PREWARM_ERRORS: Lazy<IntCounter> = Lazy::new(|| {
    register_int_counter!(
        "compute_ctl_lfc_prewarm_errors_total",
-        "Total number of LFC prewarm errors",
+        "Total number of LFC prewarms errors requested by compute_ctl or autoprewarm option",
    )
    .expect("failed to define a metric")
 });
@@ -124,7 +124,7 @@ pub(crate) static LFC_OFFLOADS: Lazy<IntCounter> = Lazy::new(|| {
 pub(crate) static LFC_OFFLOAD_ERRORS: Lazy<IntCounter> = Lazy::new(|| {
    register_int_counter!(
        "compute_ctl_lfc_offload_errors_total",
-        "Total number of LFC offload errors",
+        "Total number of LFC offload errors requested by compute_ctl or lfc_offload_period_seconds option",
    )
    .expect("failed to define a metric")
 });
--- a/compute_tools/src/migrations/0002-alter_roles.sql
+++ b/compute_tools/src/migrations/0002-alter_roles.sql
@@ -1,16 +1,3 @@
-- On December 8th, 2023, an engineering escalation (INC-110) was opened after
-- it was found that BYPASSRLS was being applied to all roles.
--
-- PR that introduced the issue: https://github.com/neondatabase/neon/pull/5657
-- Subsequent commit on main: https://github.com/neondatabase/neon/commit/ad99fa5f0393e2679e5323df653c508ffa0ac072
--
-- NOBYPASSRLS and INHERIT are the defaults for a Postgres role, but because it
-- isn't easy to know if a Postgres cluster is affected by the issue, we need to
-- keep the migration around for a long time, if not indefinitely, so any
-- cluster can be fixed.
--
-- Branching is the gift that keeps on giving...
-
 DO $$
 DECLARE
    role_name text;
--- a/compute_tools/src/migrations/0012-grant_pg_signal_backend_to_neon_superuser.sql
+++ b/compute_tools/src/migrations/0012-grant_pg_signal_backend_to_neon_superuser.sql
@@ -1 +0,0 @@
-GRANT pg_signal_backend TO neon_superuser WITH ADMIN OPTION;
--- a/compute_tools/src/migrations/tests/0004-grant_pg_monitor_to_neon_superuser.sql
+++ b/compute_tools/src/migrations/tests/0004-grant_pg_monitor_to_neon_superuser.sql
@@ -7,17 +7,13 @@ BEGIN
        INTO monitor
        FROM pg_auth_members
        WHERE roleid = 'pg_monitor'::regrole
-            AND member = 'neon_superuser'::regrole;
+            AND member = 'pg_monitor'::regrole;

-    IF monitor IS NULL THEN
-        RAISE EXCEPTION 'no entry in pg_auth_members for neon_superuser and pg_monitor';
-    END IF;
-
-    IF monitor.admin IS NULL OR NOT monitor.member THEN
+    IF NOT monitor.member THEN
        RAISE EXCEPTION 'neon_superuser is not a member of pg_monitor';
    END IF;

-    IF monitor.admin IS NULL OR NOT monitor.admin THEN
+    IF NOT monitor.admin THEN
        RAISE EXCEPTION 'neon_superuser cannot grant pg_monitor';
    END IF;
 END $$;
--- a/compute_tools/src/migrations/tests/0012-grant_pg_signal_backend_to_neon_superuser.sql
+++ b/compute_tools/src/migrations/tests/0012-grant_pg_signal_backend_to_neon_superuser.sql
@@ -1,23 +0,0 @@
-DO $$
-DECLARE
-    signal_backend record;
-BEGIN
-    SELECT pg_has_role('neon_superuser', 'pg_signal_backend', 'member') AS member,
-            admin_option AS admin
-        INTO signal_backend
-        FROM pg_auth_members
-        WHERE roleid = 'pg_signal_backend'::regrole
-            AND member = 'neon_superuser'::regrole;
-
-    IF signal_backend IS NULL THEN
-        RAISE EXCEPTION 'no entry in pg_auth_members for neon_superuser and pg_signal_backend';
-    END IF;
-
-    IF signal_backend.member IS NULL OR NOT signal_backend.member THEN
-        RAISE EXCEPTION 'neon_superuser is not a member of pg_signal_backend';
-    END IF;
-
-    IF signal_backend.admin IS NULL OR NOT signal_backend.admin THEN
-        RAISE EXCEPTION 'neon_superuser cannot grant pg_signal_backend';
-    END IF;
-END $$;
--- a/compute_tools/src/monitor.rs
+++ b/compute_tools/src/monitor.rs
@@ -84,8 +84,7 @@ impl ComputeMonitor {
        if matches!(
            compute_status,
            ComputeStatus::Terminated
-                | ComputeStatus::TerminationPendingFast
-                | ComputeStatus::TerminationPendingImmediate
+                | ComputeStatus::TerminationPending { .. }
                | ComputeStatus::Failed
        ) {
            info!(
--- a/compute_tools/src/spec.rs
+++ b/compute_tools/src/spec.rs
@@ -197,7 +197,6 @@ pub async fn handle_migrations(client: &mut Client) -> Result<()> {
        include_str!(
            "./migrations/0011-grant_pg_show_replication_origin_status_to_neon_superuser.sql"
        ),
-        include_str!("./migrations/0012-grant_pg_signal_backend_to_neon_superuser.sql"),
    ];

    MigrationRunner::new(client, &migrations)
--- a/control_plane/src/broker.rs
+++ b/control_plane/src/broker.rs
@@ -36,7 +36,7 @@ impl StorageBroker {
    pub async fn start(&self, retry_timeout: &Duration) -> anyhow::Result<()> {
        let broker = &self.env.broker;

-        println!("Starting neon broker at {}", broker.client_url());
+        print!("Starting neon broker at {}", broker.client_url());

        let mut args = Vec::new();

--- a/control_plane/src/endpoint.rs
+++ b/control_plane/src/endpoint.rs
@@ -922,8 +922,7 @@ impl Endpoint {
                        ComputeStatus::Empty
                        | ComputeStatus::ConfigurationPending
                        | ComputeStatus::Configuration
-                        | ComputeStatus::TerminationPendingFast
-                        | ComputeStatus::TerminationPendingImmediate
+                        | ComputeStatus::TerminationPending { .. }
                        | ComputeStatus::Terminated => {
                            bail!("unexpected compute status: {:?}", state.status)
                        }
--- a/control_plane/src/pageserver.rs
+++ b/control_plane/src/pageserver.rs
@@ -303,7 +303,7 @@ impl PageServerNode {
    async fn start_node(&self, retry_timeout: &Duration) -> anyhow::Result<()> {
        // TODO: using a thread here because start_process() is not async but we need to call check_status()
        let datadir = self.repo_path();
-        println!(
+        print!(
            "Starting pageserver node {} at '{}' in {:?}, retrying for {:?}",
            self.conf.id,
            self.pg_connection_config.raw_address(),
@@ -452,12 +452,6 @@ impl PageServerNode {
                .map(|x| x.parse::<usize>())
                .transpose()
                .context("Failed to parse 'image_creation_threshold' as non zero integer")?,
-            // HADRON
-            image_layer_force_creation_period: settings
-                .remove("image_layer_force_creation_period")
-                .map(humantime::parse_duration)
-                .transpose()
-                .context("Failed to parse 'image_layer_force_creation_period' as duration")?,
            image_layer_creation_check_threshold: settings
                .remove("image_layer_creation_check_threshold")
                .map(|x| x.parse::<u8>())
--- a/control_plane/src/safekeeper.rs
+++ b/control_plane/src/safekeeper.rs
@@ -127,7 +127,7 @@ impl SafekeeperNode {
        extra_opts: &[String],
        retry_timeout: &Duration,
    ) -> anyhow::Result<()> {
-        println!(
+        print!(
            "Starting safekeeper at '{}' in '{}', retrying for {:?}",
            self.pg_connection_config.raw_address(),
            self.datadir_path().display(),
--- a/control_plane/src/storage_controller.rs
+++ b/control_plane/src/storage_controller.rs
@@ -660,7 +660,7 @@ impl StorageController {
            ));
        }

-        println!("Starting storage controller at {scheme}://{host}:{listen_port}");
+        println!("Starting storage controller");

        background_process::start_process(
            COMMAND,
--- a/control_plane/storcon_cli/Cargo.toml
+++ b/control_plane/storcon_cli/Cargo.toml
@@ -14,7 +14,6 @@ humantime.workspace = true
 pageserver_api.workspace = true
 pageserver_client.workspace = true
 reqwest.workspace = true
-safekeeper_api.workspace=true
 serde_json = { workspace = true, features = ["raw_value"] }
 storage_controller_client.workspace = true
 tokio.workspace = true
--- a/control_plane/storcon_cli/src/main.rs
+++ b/control_plane/storcon_cli/src/main.rs
@@ -11,7 +11,7 @@ use pageserver_api::controller_api::{
    PlacementPolicy, SafekeeperDescribeResponse, SafekeeperSchedulingPolicyRequest,
    ShardSchedulingPolicy, ShardsPreferredAzsRequest, ShardsPreferredAzsResponse,
    SkSchedulingPolicy, TenantCreateRequest, TenantDescribeResponse, TenantPolicyRequest,
-    TenantShardMigrateRequest, TenantShardMigrateResponse, TimelineSafekeeperMigrateRequest,
+    TenantShardMigrateRequest, TenantShardMigrateResponse,
 };
 use pageserver_api::models::{
    EvictionPolicy, EvictionPolicyLayerAccessThreshold, ShardParameters, TenantConfig,
@@ -21,7 +21,6 @@ use pageserver_api::models::{
 use pageserver_api::shard::{ShardStripeSize, TenantShardId};
 use pageserver_client::mgmt_api::{self};
 use reqwest::{Certificate, Method, StatusCode, Url};
-use safekeeper_api::models::TimelineLocateResponse;
 use storage_controller_client::control_api::Client;
 use utils::id::{NodeId, TenantId, TimelineId};

@@ -280,23 +279,6 @@ enum Command {
        #[arg(long)]
        concurrency: Option<usize>,
    },
-    /// Locate safekeepers for a timeline from the storcon DB.
-    TimelineLocate {
-        #[arg(long)]
-        tenant_id: TenantId,
-        #[arg(long)]
-        timeline_id: TimelineId,
-    },
-    /// Migrate a timeline to a new set of safekeepers
-    TimelineSafekeeperMigrate {
-        #[arg(long)]
-        tenant_id: TenantId,
-        #[arg(long)]
-        timeline_id: TimelineId,
-        /// Example: --new-sk-set 1,2,3
-        #[arg(long, required = true, value_delimiter = ',')]
-        new_sk_set: Vec<NodeId>,
-    },
 }

 #[derive(Parser)]
@@ -1342,7 +1324,7 @@ async fn main() -> anyhow::Result<()> {
            concurrency,
        } => {
            let mut path = format!(
-                "v1/tenant/{tenant_shard_id}/timeline/{timeline_id}/download_heatmap_layers",
+                "/v1/tenant/{tenant_shard_id}/timeline/{timeline_id}/download_heatmap_layers",
            );

            if let Some(c) = concurrency {
@@ -1353,41 +1335,6 @@ async fn main() -> anyhow::Result<()> {
                .dispatch::<(), ()>(Method::POST, path, None)
                .await?;
        }
-        Command::TimelineLocate {
-            tenant_id,
-            timeline_id,
-        } => {
-            let path = format!("debug/v1/tenant/{tenant_id}/timeline/{timeline_id}/locate");
-
-            let resp = storcon_client
-                .dispatch::<(), TimelineLocateResponse>(Method::GET, path, None)
-                .await?;
-
-            let sk_set = resp.sk_set.iter().map(|id| id.0 as i64).collect::<Vec<_>>();
-            let new_sk_set = resp
-                .new_sk_set
-                .as_ref()
-                .map(|ids| ids.iter().map(|id| id.0 as i64).collect::<Vec<_>>());
-
-            println!("generation = {}", resp.generation);
-            println!("sk_set = {sk_set:?}");
-            println!("new_sk_set = {new_sk_set:?}");
-        }
-        Command::TimelineSafekeeperMigrate {
-            tenant_id,
-            timeline_id,
-            new_sk_set,
-        } => {
-            let path = format!("v1/tenant/{tenant_id}/timeline/{timeline_id}/safekeeper_migrate");
-
-            storcon_client
-                .dispatch::<_, ()>(
-                    Method::POST,
-                    path,
-                    Some(TimelineSafekeeperMigrateRequest { new_sk_set }),
-                )
-                .await?;
-        }
    }

    Ok(())
--- a/deny.toml
+++ b/deny.toml
@@ -52,7 +52,6 @@ exceptions = [
    # Zlib license has some restrictions if we decide to change sth
    { allow = ["Zlib"], name = "const_format_proc_macros", version = "*" },
    { allow = ["Zlib"], name = "const_format", version = "*" },
-    { allow = ["Zlib"], name = "foldhash", version = "*" },
 ]

 [licenses.private]
--- a/libs/compute_api/src/responses.rs
+++ b/libs/compute_api/src/responses.rs
@@ -121,15 +121,6 @@ pub enum TerminateMode {
    Immediate,
 }

-impl From<TerminateMode> for ComputeStatus {
-    fn from(mode: TerminateMode) -> Self {
-        match mode {
-            TerminateMode::Fast => ComputeStatus::TerminationPendingFast,
-            TerminateMode::Immediate => ComputeStatus::TerminationPendingImmediate,
-        }
-    }
-}
-
 #[derive(Serialize, Clone, Copy, Debug, Deserialize, PartialEq, Eq)]
 #[serde(rename_all = "snake_case")]
 pub enum ComputeStatus {
@@ -150,9 +141,7 @@ pub enum ComputeStatus {
    // control-plane to terminate it.
    Failed,
    // Termination requested
-    TerminationPendingFast,
-    // Termination requested, without waiting 30s before returning from /terminate
-    TerminationPendingImmediate,
+    TerminationPending { mode: TerminateMode },
    // Terminated Postgres
    Terminated,
 }
@@ -171,10 +160,7 @@ impl Display for ComputeStatus {
            ComputeStatus::Running => f.write_str("running"),
            ComputeStatus::Configuration => f.write_str("configuration"),
            ComputeStatus::Failed => f.write_str("failed"),
-            ComputeStatus::TerminationPendingFast => f.write_str("termination-pending-fast"),
-            ComputeStatus::TerminationPendingImmediate => {
-                f.write_str("termination-pending-immediate")
-            }
+            ComputeStatus::TerminationPending { .. } => f.write_str("termination-pending"),
            ComputeStatus::Terminated => f.write_str("terminated"),
        }
    }
--- a/libs/http-utils/src/endpoint.rs
+++ b/libs/http-utils/src/endpoint.rs
@@ -20,7 +20,6 @@ use tokio_stream::wrappers::ReceiverStream;
 use tokio_util::io::ReaderStream;
 use tracing::{Instrument, debug, info, info_span, warn};
 use utils::auth::{AuthError, Claims, SwappableJwtAuth};
-use utils::metrics_collector::{METRICS_COLLECTOR, METRICS_STALE_MILLIS};

 use crate::error::{ApiError, api_error_handler, route_error_handler};
 use crate::request::{get_query_param, parse_query_param};
@@ -251,28 +250,9 @@ impl std::io::Write for ChannelWriter {
    }
 }

-pub async fn prometheus_metrics_handler(
-    req: Request<Body>,
-    force_metric_collection_on_scrape: bool,
-) -> Result<Response<Body>, ApiError> {
+pub async fn prometheus_metrics_handler(_req: Request<Body>) -> Result<Response<Body>, ApiError> {
    SERVE_METRICS_COUNT.inc();

-    // HADRON
-    let requested_use_latest = parse_query_param(&req, "use_latest")?;
-
-    let use_latest = match requested_use_latest {
-        None => force_metric_collection_on_scrape,
-        Some(true) => true,
-        Some(false) => {
-            if force_metric_collection_on_scrape {
-                // We don't cache in this case
-                true
-            } else {
-                false
-            }
-        }
-    };
-
    let started_at = std::time::Instant::now();

    let (tx, rx) = mpsc::channel(1);
@@ -297,18 +277,12 @@ pub async fn prometheus_metrics_handler(

        let _span = span.entered();

-        // HADRON
-        let collected = if use_latest {
-            // Skip caching the results if we always force metric collection on scrape.
-            METRICS_COLLECTOR.run_once(!force_metric_collection_on_scrape)
-        } else {
-            METRICS_COLLECTOR.last_collected()
-        };
+        let metrics = metrics::gather();

        let gathered_at = std::time::Instant::now();

        let res = encoder
-            .encode(&collected.metrics, &mut writer)
+            .encode(&metrics, &mut writer)
            .and_then(|_| writer.flush().map_err(|e| e.into()));

        // this instant is not when we finally got the full response sent, sending is done by hyper
@@ -321,10 +295,6 @@ pub async fn prometheus_metrics_handler(
        let encoded_in = encoded_at - gathered_at - writer.wait_time();
        let total = encoded_at - started_at;

-        // HADRON
-        let staleness_ms = (encoded_at - collected.collected_at).as_millis();
-        METRICS_STALE_MILLIS.set(staleness_ms as i64);
-
        match res {
            Ok(()) => {
                tracing::info!(
@@ -333,7 +303,6 @@ pub async fn prometheus_metrics_handler(
                    spawning_ms = spawned_in.as_millis(),
                    collection_ms = collected_in.as_millis(),
                    encoding_ms = encoded_in.as_millis(),
-                    stalenss_ms = staleness_ms,
                    "responded /metrics"
                );
            }
--- a/libs/pageserver_api/src/config.rs
+++ b/libs/pageserver_api/src/config.rs
@@ -226,7 +226,6 @@ pub struct ConfigToml {
    pub synthetic_size_calculation_interval: Duration,
    pub disk_usage_based_eviction: DiskUsageEvictionTaskConfig,
    pub test_remote_failures: u64,
-    pub test_remote_failures_probability: u64,
    pub ondemand_download_behavior_treat_error_as_warn: bool,
    #[serde(with = "humantime_serde")]
    pub background_task_maximum_delay: Duration,
@@ -272,9 +271,6 @@ pub struct ConfigToml {
    pub timeline_import_config: TimelineImportConfig,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub basebackup_cache_config: Option<BasebackupCacheConfig>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub image_layer_generation_large_timeline_threshold: Option<u64>,
-    pub force_metric_collection_on_scrape: bool,
 }

 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
@@ -564,11 +560,6 @@ pub struct TenantConfigToml {
    pub gc_period: Duration,
    // Delta layer churn threshold to create L1 image layers.
    pub image_creation_threshold: usize,
-    // HADRON
-    // When the timeout is reached, PageServer will (1) force compact any remaining L0 deltas and
-    // (2) create image layers if there are any L1 deltas.
-    #[serde(with = "humantime_serde")]
-    pub image_layer_force_creation_period: Option<Duration>,
    // Determines how much history is retained, to allow
    // branching and read replicas at an older point in time.
    // The unit is time.
@@ -767,7 +758,6 @@ impl Default for ConfigToml {
            disk_usage_based_eviction: DiskUsageEvictionTaskConfig::default(),

            test_remote_failures: (0),
-            test_remote_failures_probability: (100),

            ondemand_download_behavior_treat_error_as_warn: (false),

@@ -831,8 +821,6 @@ impl Default for ConfigToml {
            },
            basebackup_cache_config: None,
            posthog_config: None,
-            image_layer_generation_large_timeline_threshold: Some(2 * 1024 * 1024 * 1024),
-            force_metric_collection_on_scrape: true,
        }
    }
 }
@@ -926,7 +914,6 @@ impl Default for TenantConfigToml {
            gc_period: humantime::parse_duration(DEFAULT_GC_PERIOD)
                .expect("cannot parse default gc period"),
            image_creation_threshold: DEFAULT_IMAGE_CREATION_THRESHOLD,
-            image_layer_force_creation_period: None,
            pitr_interval: humantime::parse_duration(DEFAULT_PITR_INTERVAL)
                .expect("cannot parse default PITR interval"),
            walreceiver_connect_timeout: humantime::parse_duration(
--- a/libs/pageserver_api/src/controller_api.rs
+++ b/libs/pageserver_api/src/controller_api.rs
@@ -10,7 +10,7 @@ use serde::{Deserialize, Serialize};
 use utils::id::{NodeId, TenantId, TimelineId};
 use utils::lsn::Lsn;

-use crate::models::{PageserverUtilization, ShardParameters, TenantConfig, TimelineInfo};
+use crate::models::{PageserverUtilization, ShardParameters, TenantConfig};
 use crate::shard::{ShardStripeSize, TenantShardId};

 #[derive(Serialize, Deserialize, Debug)]
@@ -126,13 +126,6 @@ pub struct TenantDescribeResponse {
    pub config: TenantConfig,
 }

-#[derive(Serialize, Deserialize, Debug)]
-pub struct TenantTimelineDescribeResponse {
-    pub shards: Vec<TimelineInfo>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub image_consistent_lsn: Option<Lsn>,
-}
-
 #[derive(Serialize, Deserialize, Debug)]
 pub struct NodeShardResponse {
    pub node_id: NodeId,
--- a/libs/pageserver_api/src/models.rs
+++ b/libs/pageserver_api/src/models.rs
@@ -597,9 +597,6 @@ pub struct TenantConfigPatch {
    pub gc_period: FieldPatch<String>,
    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
    pub image_creation_threshold: FieldPatch<usize>,
-    // HADRON
-    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
-    pub image_layer_force_creation_period: FieldPatch<String>,
    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
    pub pitr_interval: FieldPatch<String>,
    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
@@ -703,11 +700,6 @@ pub struct TenantConfig {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub image_creation_threshold: Option<usize>,

-    // HADRON
-    #[serde(skip_serializing_if = "Option::is_none")]
-    #[serde(with = "humantime_serde")]
-    pub image_layer_force_creation_period: Option<Duration>,
-
    #[serde(skip_serializing_if = "Option::is_none")]
    #[serde(with = "humantime_serde")]
    pub pitr_interval: Option<Duration>,
@@ -806,7 +798,6 @@ impl TenantConfig {
            mut gc_horizon,
            mut gc_period,
            mut image_creation_threshold,
-            mut image_layer_force_creation_period,
            mut pitr_interval,
            mut walreceiver_connect_timeout,
            mut lagging_wal_timeout,
@@ -870,11 +861,6 @@ impl TenantConfig {
        patch
            .image_creation_threshold
            .apply(&mut image_creation_threshold);
-        // HADRON
-        patch
-            .image_layer_force_creation_period
-            .map(|v| humantime::parse_duration(&v))?
-            .apply(&mut image_layer_force_creation_period);
        patch
            .pitr_interval
            .map(|v| humantime::parse_duration(&v))?
@@ -956,7 +942,6 @@ impl TenantConfig {
            gc_horizon,
            gc_period,
            image_creation_threshold,
-            image_layer_force_creation_period,
            pitr_interval,
            walreceiver_connect_timeout,
            lagging_wal_timeout,
@@ -1031,9 +1016,6 @@ impl TenantConfig {
            image_creation_threshold: self
                .image_creation_threshold
                .unwrap_or(global_conf.image_creation_threshold),
-            image_layer_force_creation_period: self
-                .image_layer_force_creation_period
-                .or(global_conf.image_layer_force_creation_period),
            pitr_interval: self.pitr_interval.unwrap_or(global_conf.pitr_interval),
            walreceiver_connect_timeout: self
                .walreceiver_connect_timeout
@@ -1622,9 +1604,6 @@ pub struct TimelineInfo {

    /// Whether the timeline is invisible in synthetic size calculations.
    pub is_invisible: Option<bool>,
-    // HADRON: the largest LSN below which all page updates have been included in the image layers.
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub image_consistent_lsn: Option<Lsn>,
 }

 #[derive(Debug, Clone, Serialize, Deserialize)]
--- a/libs/remote_storage/Cargo.toml
+++ b/libs/remote_storage/Cargo.toml
@@ -43,7 +43,6 @@ itertools.workspace = true
 sync_wrapper = { workspace = true, features = ["futures"] }

 byteorder = "1.4"
-rand = "0.8.5"

 [dev-dependencies]
 camino-tempfile.workspace = true
--- a/libs/remote_storage/src/lib.rs
+++ b/libs/remote_storage/src/lib.rs
@@ -732,15 +732,9 @@ impl GenericRemoteStorage {
        })
    }

-    /* BEGIN_HADRON */
-    pub fn unreliable_wrapper(s: Self, fail_first: u64, fail_probability: u64) -> Self {
-        Self::Unreliable(Arc::new(UnreliableWrapper::new(
-            s,
-            fail_first,
-            fail_probability,
-        )))
+    pub fn unreliable_wrapper(s: Self, fail_first: u64) -> Self {
+        Self::Unreliable(Arc::new(UnreliableWrapper::new(s, fail_first)))
    }
-    /* END_HADRON */

    /// See [`RemoteStorage::upload`], which this method calls with `None` as metadata.
    pub async fn upload_storage_object(
--- a/libs/remote_storage/src/simulate_failures.rs
+++ b/libs/remote_storage/src/simulate_failures.rs
@@ -1,8 +1,6 @@
 //! This module provides a wrapper around a real RemoteStorage implementation that
 //! causes the first N attempts at each upload or download operatio to fail. For
 //! testing purposes.
-use rand::Rng;
-use std::cmp;
 use std::collections::HashMap;
 use std::collections::hash_map::Entry;
 use std::num::NonZeroU32;
@@ -27,12 +25,6 @@ pub struct UnreliableWrapper {

    // Tracks how many failed attempts of each operation has been made.
    attempts: Mutex<HashMap<RemoteOp, u64>>,
-
-    /* BEGIN_HADRON */
-    // This the probability of failure for each operation, ranged from [0, 100].
-    // The probability is default to 100, which means that all operations will fail.
-    attempt_failure_probability: u64,
-    /* END_HADRON */
 }

 /// Used to identify retries of different unique operation.
@@ -48,11 +40,7 @@ enum RemoteOp {
 }

 impl UnreliableWrapper {
-    pub fn new(
-        inner: crate::GenericRemoteStorage,
-        attempts_to_fail: u64,
-        attempt_failure_probability: u64,
-    ) -> Self {
+    pub fn new(inner: crate::GenericRemoteStorage, attempts_to_fail: u64) -> Self {
        assert!(attempts_to_fail > 0);
        let inner = match inner {
            GenericRemoteStorage::AwsS3(s) => GenericRemoteStorage::AwsS3(s),
@@ -63,11 +51,9 @@ impl UnreliableWrapper {
                panic!("Can't wrap unreliable wrapper unreliably")
            }
        };
-        let actual_attempt_failure_probability = cmp::min(attempt_failure_probability, 100);
        UnreliableWrapper {
            inner,
            attempts_to_fail,
-            attempt_failure_probability: actual_attempt_failure_probability,
            attempts: Mutex::new(HashMap::new()),
        }
    }
@@ -80,7 +66,6 @@ impl UnreliableWrapper {
    ///
    fn attempt(&self, op: RemoteOp) -> anyhow::Result<u64> {
        let mut attempts = self.attempts.lock().unwrap();
-        let mut rng = rand::thread_rng();

        match attempts.entry(op) {
            Entry::Occupied(mut e) => {
@@ -90,19 +75,15 @@ impl UnreliableWrapper {
                    *p
                };

-                /* BEGIN_HADRON */
-                // If there are more attempts to fail, fail the request by probability.
-                if (attempts_before_this < self.attempts_to_fail)
-                    && (rng.gen_range(0..=100) < self.attempt_failure_probability)
-                {
+                if attempts_before_this >= self.attempts_to_fail {
+                    // let it succeed
+                    e.remove();
+                    Ok(attempts_before_this)
+                } else {
                    let error =
                        anyhow::anyhow!("simulated failure of remote operation {:?}", e.key());
                    Err(error)
-                } else {
-                    e.remove();
-                    Ok(attempts_before_this)
                }
-                /* END_HADRON */
            }
            Entry::Vacant(e) => {
                let error = anyhow::anyhow!("simulated failure of remote operation {:?}", e.key());
--- a/libs/safekeeper_api/src/models.rs
+++ b/libs/safekeeper_api/src/models.rs
@@ -11,7 +11,7 @@ use utils::id::{NodeId, TenantId, TenantTimelineId, TimelineId};
 use utils::lsn::Lsn;
 use utils::pageserver_feedback::PageserverFeedback;

-use crate::membership::{Configuration, SafekeeperGeneration};
+use crate::membership::Configuration;
 use crate::{ServerInfo, Term};

 #[derive(Debug, Serialize, Deserialize)]
@@ -311,12 +311,3 @@ pub struct PullTimelineResponse {
    pub safekeeper_host: Option<String>,
    // TODO: add more fields?
 }
-
-/// Response to a timeline locate request.
-/// Storcon-only API.
-#[derive(Serialize, Deserialize, Clone, Debug)]
-pub struct TimelineLocateResponse {
-    pub generation: SafekeeperGeneration,
-    pub sk_set: Vec<NodeId>,
-    pub new_sk_set: Option<Vec<NodeId>>,
-}
--- a/libs/utils/src/env.rs
+++ b/libs/utils/src/env.rs
@@ -44,62 +44,3 @@ where
        }
    }
 }
-
-/* BEGIN_HADRON */
-pub enum DeploymentMode {
-    Dev,
-    Staging,
-    Prod,
-}
-
-pub fn get_deployment_mode() -> Option<DeploymentMode> {
-    match std::env::var("DEPLOYMENT_MODE") {
-        Ok(env) => match env.as_str() {
-            "development" => Some(DeploymentMode::Dev),
-            "staging" => Some(DeploymentMode::Staging),
-            "production" => Some(DeploymentMode::Prod),
-            _ => {
-                tracing::error!("Unexpected DEPLOYMENT_MODE: {}", env);
-                None
-            }
-        },
-        Err(_) => {
-            tracing::error!("DEPLOYMENT_MODE not set");
-            None
-        }
-    }
-}
-
-pub fn is_dev_or_staging() -> bool {
-    matches!(
-        get_deployment_mode(),
-        Some(DeploymentMode::Dev) | Some(DeploymentMode::Staging)
-    )
-}
-
-pub enum TestingMode {
-    Chaos,
-    Stress,
-}
-
-pub fn get_test_mode() -> Option<TestingMode> {
-    match std::env::var("HADRON_TEST_MODE") {
-        Ok(env) => match env.as_str() {
-            "chaos" => Some(TestingMode::Chaos),
-            "stress" => Some(TestingMode::Stress),
-            _ => {
-                tracing::error!("Unexpected HADRON_TEST_MODE: {}", env);
-                None
-            }
-        },
-        Err(_) => {
-            tracing::error!("HADRON_TEST_MODE not set");
-            None
-        }
-    }
-}
-
-pub fn is_chaos_testing() -> bool {
-    matches!(get_test_mode(), Some(TestingMode::Chaos))
-}
-/* END_HADRON */
--- a/libs/utils/src/lib.rs
+++ b/libs/utils/src/lib.rs
@@ -99,8 +99,6 @@ pub mod elapsed_accum;
 #[cfg(target_os = "linux")]
 pub mod linux_socket_ioctl;

-pub mod metrics_collector;
-
 // Re-export used in macro. Avoids adding git-version as dep in target crates.
 #[doc(hidden)]
 pub use git_version;
--- a/libs/utils/src/metrics_collector.rs
+++ b/libs/utils/src/metrics_collector.rs
@@ -1,75 +0,0 @@
-use std::{
-    sync::{Arc, RwLock},
-    time::{Duration, Instant},
-};
-
-use metrics::{IntGauge, proto::MetricFamily, register_int_gauge};
-use once_cell::sync::Lazy;
-
-pub static METRICS_STALE_MILLIS: Lazy<IntGauge> = Lazy::new(|| {
-    register_int_gauge!(
-        "metrics_metrics_stale_milliseconds",
-        "The current metrics stale time in milliseconds"
-    )
-    .expect("failed to define a metric")
-});
-
-#[derive(Debug)]
-pub struct CollectedMetrics {
-    pub metrics: Vec<MetricFamily>,
-    pub collected_at: Instant,
-}
-
-impl CollectedMetrics {
-    fn new(metrics: Vec<MetricFamily>) -> Self {
-        Self {
-            metrics,
-            collected_at: Instant::now(),
-        }
-    }
-}
-
-#[derive(Debug)]
-pub struct MetricsCollector {
-    last_collected: RwLock<Arc<CollectedMetrics>>,
-}
-
-impl MetricsCollector {
-    pub fn new() -> Self {
-        Self {
-            last_collected: RwLock::new(Arc::new(CollectedMetrics::new(vec![]))),
-        }
-    }
-
-    #[tracing::instrument(name = "metrics_collector", skip_all)]
-    pub fn run_once(&self, cache_metrics: bool) -> Arc<CollectedMetrics> {
-        let started = Instant::now();
-        let metrics = metrics::gather();
-        let collected = Arc::new(CollectedMetrics::new(metrics));
-        if cache_metrics {
-            let mut guard = self.last_collected.write().unwrap();
-            *guard = collected.clone();
-        }
-        tracing::info!(
-            "Collected {} metric families in {} ms",
-            collected.metrics.len(),
-            started.elapsed().as_millis()
-        );
-        collected
-    }
-
-    pub fn last_collected(&self) -> Arc<CollectedMetrics> {
-        self.last_collected.read().unwrap().clone()
-    }
-}
-
-impl Default for MetricsCollector {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-// Interval for metrics collection. Currently hard-coded to be the same as the metrics scape interval from the obs agent
-pub static METRICS_COLLECTION_INTERVAL: Duration = Duration::from_secs(30);
-
-pub static METRICS_COLLECTOR: Lazy<MetricsCollector> = Lazy::new(MetricsCollector::default);
--- a/libs/walproposer/src/api_bindings.rs
+++ b/libs/walproposer/src/api_bindings.rs
@@ -428,12 +428,6 @@ pub fn empty_shmem() -> crate::bindings::WalproposerShmemState {
        shard_number: 0,
    };

-    let empty_wal_rate_limiter = crate::bindings::WalRateLimiter {
-        should_limit: crate::bindings::pg_atomic_uint32 { value: 0 },
-        sent_bytes: 0,
-        last_recorded_time_us: 0,
-    };
-
    crate::bindings::WalproposerShmemState {
        propEpochStartLsn: crate::bindings::pg_atomic_uint64 { value: 0 },
        donor_name: [0; 64],
@@ -447,7 +441,6 @@ pub fn empty_shmem() -> crate::bindings::WalproposerShmemState {
        num_shards: 0,
        replica_promote: false,
        min_ps_feedback: empty_feedback,
-        wal_rate_limiter: empty_wal_rate_limiter,
    }
 }

--- a/pageserver/client/src/mgmt_api.rs
+++ b/pageserver/client/src/mgmt_api.rs
@@ -1,4 +1,4 @@
-use std::collections::{BTreeMap, HashMap};
+use std::collections::HashMap;
 use std::error::Error as _;
 use std::time::Duration;

@@ -251,70 +251,6 @@ impl Client {
        Ok(())
    }

-    pub async fn tenant_timeline_compact(
-        &self,
-        tenant_shard_id: TenantShardId,
-        timeline_id: TimelineId,
-        force_image_layer_creation: bool,
-        must_force_image_layer_creation: bool,
-        scheduled: bool,
-        wait_until_done: bool,
-    ) -> Result<()> {
-        let mut path = reqwest::Url::parse(&format!(
-            "{}/v1/tenant/{tenant_shard_id}/timeline/{timeline_id}/compact",
-            self.mgmt_api_endpoint
-        ))
-        .expect("Cannot build URL");
-
-        if force_image_layer_creation {
-            path.query_pairs_mut()
-                .append_pair("force_image_layer_creation", "true");
-        }
-
-        if must_force_image_layer_creation {
-            path.query_pairs_mut()
-                .append_pair("must_force_image_layer_creation", "true");
-        }
-
-        if scheduled {
-            path.query_pairs_mut().append_pair("scheduled", "true");
-        }
-        if wait_until_done {
-            path.query_pairs_mut()
-                .append_pair("wait_until_scheduled_compaction_done", "true");
-            path.query_pairs_mut()
-                .append_pair("wait_until_uploaded", "true");
-        }
-        self.request(Method::PUT, path, ()).await?;
-        Ok(())
-    }
-
-    /* BEGIN_HADRON */
-    pub async fn tenant_timeline_describe(
-        &self,
-        tenant_shard_id: &TenantShardId,
-        timeline_id: &TimelineId,
-    ) -> Result<TimelineInfo> {
-        let mut path = reqwest::Url::parse(&format!(
-            "{}/v1/tenant/{tenant_shard_id}/timeline/{timeline_id}",
-            self.mgmt_api_endpoint
-        ))
-        .expect("Cannot build URL");
-        path.query_pairs_mut()
-            .append_pair("include-image-consistent-lsn", "true");
-
-        let response: reqwest::Response = self.request(Method::GET, path, ()).await?;
-        let body = response.json().await.map_err(Error::ReceiveBody)?;
-        Ok(body)
-    }
-
-    pub async fn list_tenant_visible_size(&self) -> Result<BTreeMap<TenantShardId, u64>> {
-        let uri = format!("{}/v1/list_tenant_visible_size", self.mgmt_api_endpoint);
-        let resp = self.get(&uri).await?;
-        resp.json().await.map_err(Error::ReceiveBody)
-    }
-    /* END_HADRON */
-
    pub async fn tenant_scan_remote_storage(
        &self,
        tenant_id: TenantId,
--- a/pageserver/client_grpc/Cargo.toml
+++ b/pageserver/client_grpc/Cargo.toml
@@ -4,12 +4,8 @@ version = "0.1.0"
 edition.workspace = true
 license.workspace = true

-[features]
-testing = ["pageserver_api/testing"]
-
 [dependencies]
 anyhow.workspace = true
-arc-swap.workspace = true
 bytes.workspace = true
 compute_api.workspace = true
 futures.workspace = true
@@ -17,7 +13,6 @@ pageserver_api.workspace = true
 pageserver_page_api.workspace = true
 tokio.workspace = true
 tokio-stream.workspace = true
-tokio-util.workspace = true
 tonic.workspace = true
 tracing.workspace = true
 utils.workspace = true
--- a/pageserver/client_grpc/src/client.rs
+++ b/pageserver/client_grpc/src/client.rs
@@ -1,12 +1,9 @@
 use std::collections::HashMap;
-use std::num::NonZero;
 use std::sync::Arc;

 use anyhow::anyhow;
-use arc_swap::ArcSwap;
 use futures::stream::FuturesUnordered;
 use futures::{FutureExt as _, StreamExt as _};
-use tonic::codec::CompressionEncoding;
 use tracing::instrument;

 use crate::pool::{ChannelPool, ClientGuard, ClientPool, StreamGuard, StreamPool};
@@ -18,32 +15,6 @@ use pageserver_page_api as page_api;
 use utils::id::{TenantId, TimelineId};
 use utils::shard::{ShardCount, ShardIndex, ShardNumber};

-/// Max number of concurrent clients per channel (i.e. TCP connection). New channels will be spun up
-/// when full.
-///
-/// TODO: tune all of these constants, and consider making them configurable.
-/// TODO: consider separate limits for unary and streaming clients, so we don't fill up channels
-/// with only streams.
-const MAX_CLIENTS_PER_CHANNEL: NonZero<usize> = NonZero::new(16).unwrap();
-
-/// Max number of concurrent unary request clients per shard.
-const MAX_UNARY_CLIENTS: NonZero<usize> = NonZero::new(64).unwrap();
-
-/// Max number of concurrent GetPage streams per shard. The max number of concurrent GetPage
-/// requests is given by `MAX_STREAMS * MAX_STREAM_QUEUE_DEPTH`.
-const MAX_STREAMS: NonZero<usize> = NonZero::new(64).unwrap();
-
-/// Max number of pipelined requests per stream.
-const MAX_STREAM_QUEUE_DEPTH: NonZero<usize> = NonZero::new(2).unwrap();
-
-/// Max number of concurrent bulk GetPage streams per shard, used e.g. for prefetches. Because these
-/// are more throughput-oriented, we have a smaller limit but higher queue depth.
-const MAX_BULK_STREAMS: NonZero<usize> = NonZero::new(16).unwrap();
-
-/// Max number of pipelined requests per bulk stream. These are more throughput-oriented and thus
-/// get a larger queue depth.
-const MAX_BULK_STREAM_QUEUE_DEPTH: NonZero<usize> = NonZero::new(4).unwrap();
-
 /// A rich Pageserver gRPC client for a single tenant timeline. This client is more capable than the
 /// basic `page_api::Client` gRPC client, and supports:
 ///
@@ -57,85 +28,28 @@ const MAX_BULK_STREAM_QUEUE_DEPTH: NonZero<usize> = NonZero::new(4).unwrap();
 /// TODO: this client does not support base backups or LSN leases, as these are only used by
 /// compute_ctl. Consider adding this, but LSN leases need concurrent requests on all shards.
 pub struct PageserverClient {
-    /// The tenant ID.
-    tenant_id: TenantId,
-    /// The timeline ID.
-    timeline_id: TimelineId,
-    /// The JWT auth token for this tenant, if any.
-    auth_token: Option<String>,
-    /// The compression to use, if any.
-    compression: Option<CompressionEncoding>,
-    /// The shards for this tenant.
-    shards: ArcSwap<Shards>,
-    /// The retry configuration.
+    // TODO: support swapping out the shard map, e.g. via an ArcSwap.
+    shards: Shards,
    retry: Retry,
 }

 impl PageserverClient {
    /// Creates a new Pageserver client for a given tenant and timeline. Uses the Pageservers given
-    /// in the shard spec, which must be complete and must use gRPC URLs.
+    /// in the shard map, which must be complete and must use gRPC URLs.
    pub fn new(
        tenant_id: TenantId,
        timeline_id: TimelineId,
-        shard_spec: ShardSpec,
+        shard_map: HashMap<ShardIndex, String>,
+        stripe_size: ShardStripeSize,
        auth_token: Option<String>,
-        compression: Option<CompressionEncoding>,
    ) -> anyhow::Result<Self> {
-        let shards = Shards::new(
-            tenant_id,
-            timeline_id,
-            shard_spec,
-            auth_token.clone(),
-            compression,
-        )?;
+        let shards = Shards::new(tenant_id, timeline_id, shard_map, stripe_size, auth_token)?;
        Ok(Self {
-            tenant_id,
-            timeline_id,
-            auth_token,
-            compression,
-            shards: ArcSwap::new(Arc::new(shards)),
+            shards,
            retry: Retry,
        })
    }

-    /// Updates the shards from the given shard spec. In-flight requests will complete using the
-    /// existing shards, but may retry with the new shards if they fail.
-    ///
-    /// TODO: verify that in-flight requests are allowed to complete, and that the old pools are
-    /// properly spun down and dropped afterwards.
-    pub fn update_shards(&self, shard_spec: ShardSpec) -> anyhow::Result<()> {
-        // Validate the shard spec. We should really use `ArcSwap::rcu` for this, to avoid races
-        // with concurrent updates, but that involves creating a new `Shards` on every attempt,
-        // which spins up a bunch of Tokio tasks and such. These should already be checked elsewhere
-        // in the stack, and if they're violated then we already have problems elsewhere, so a
-        // best-effort but possibly-racy check is okay here.
-        let old = self.shards.load_full();
-        if shard_spec.count < old.count {
-            return Err(anyhow!(
-                "can't reduce shard count from {} to {}",
-                old.count,
-                shard_spec.count
-            ));
-        }
-        if !old.count.is_unsharded() && shard_spec.stripe_size != old.stripe_size {
-            return Err(anyhow!(
-                "can't change stripe size from {} to {}",
-                old.stripe_size,
-                shard_spec.stripe_size
-            ));
-        }
-
-        let shards = Shards::new(
-            self.tenant_id,
-            self.timeline_id,
-            shard_spec,
-            self.auth_token.clone(),
-            self.compression,
-        )?;
-        self.shards.store(Arc::new(shards));
-        Ok(())
-    }
-
    /// Returns whether a relation exists.
    #[instrument(skip_all, fields(rel=%req.rel, lsn=%req.read_lsn))]
    pub async fn check_rel_exists(
@@ -143,9 +57,9 @@ impl PageserverClient {
        req: page_api::CheckRelExistsRequest,
    ) -> tonic::Result<page_api::CheckRelExistsResponse> {
        self.retry
-            .with(async |_| {
+            .with(async || {
                // Relation metadata is only available on shard 0.
-                let mut client = self.shards.load_full().get_zero().client().await?;
+                let mut client = self.shards.get_zero().client().await?;
                client.check_rel_exists(req).await
            })
            .await
@@ -158,23 +72,21 @@ impl PageserverClient {
        req: page_api::GetDbSizeRequest,
    ) -> tonic::Result<page_api::GetDbSizeResponse> {
        self.retry
-            .with(async |_| {
+            .with(async || {
                // Relation metadata is only available on shard 0.
-                let mut client = self.shards.load_full().get_zero().client().await?;
+                let mut client = self.shards.get_zero().client().await?;
                client.get_db_size(req).await
            })
            .await
    }

-    /// Fetches pages. The `request_id` must be unique across all in-flight requests, and the
-    /// `attempt` must be 0 (incremented on retry). Automatically splits requests that straddle
-    /// shard boundaries, and assembles the responses.
+    /// Fetches pages. The `request_id` must be unique across all in-flight requests. Automatically
+    /// splits requests that straddle shard boundaries, and assembles the responses.
    ///
    /// Unlike `page_api::Client`, this automatically converts `status_code` into `tonic::Status`
    /// errors. All responses will have `GetPageStatusCode::Ok`.
    #[instrument(skip_all, fields(
        req_id = %req.request_id,
-        class = %req.request_class,
        rel = %req.rel,
        blkno = %req.block_numbers[0],
        blks = %req.block_numbers.len(),
@@ -188,96 +100,68 @@ impl PageserverClient {
        if req.block_numbers.is_empty() {
            return Err(tonic::Status::invalid_argument("no block number"));
        }
-        // The request attempt must be 0. The client will increment it internally.
-        if req.request_id.attempt != 0 {
-            return Err(tonic::Status::invalid_argument("request attempt must be 0"));
-        }

-        // The shards may change while we're fetching pages. We execute the request using a stable
-        // view of the shards (especially important for requests that span shards), but retry the
-        // top-level (pre-split) request to pick up shard changes. This can lead to unnecessary
-        // retries and re-splits in some cases where requests span shards, but these are expected to
-        // be rare.
-        //
-        // TODO: the gRPC server and client doesn't yet properly support shard splits. Revisit this
-        // once we figure out how to handle these.
-        self.retry
-            .with(async |attempt| {
-                let mut req = req.clone();
-                req.request_id.attempt = attempt as u32;
-                Self::get_page_with_shards(req, &self.shards.load_full()).await
-            })
-            .await
-    }
-
-    /// Fetches pages using the given shards. This uses a stable view of the shards, regardless of
-    /// concurrent shard updates. Does not retry internally, but is retried by `get_page()`.
-    async fn get_page_with_shards(
-        req: page_api::GetPageRequest,
-        shards: &Shards,
-    ) -> tonic::Result<page_api::GetPageResponse> {
        // Fast path: request is for a single shard.
        if let Some(shard_id) =
-            GetPageSplitter::for_single_shard(&req, shards.count, shards.stripe_size)
+            GetPageSplitter::is_single_shard(&req, self.shards.count, self.shards.stripe_size)
        {
-            return Self::get_page_with_shard(req, shards.get(shard_id)?).await;
+            return self.get_page_for_shard(shard_id, req).await;
        }

        // Request spans multiple shards. Split it, dispatch concurrent per-shard requests, and
        // reassemble the responses.
-        let mut splitter = GetPageSplitter::split(req, shards.count, shards.stripe_size);
+        //
+        // TODO: when we support shard map updates, we need to detect when it changes and re-split
+        // the request on errors.
+        let mut splitter = GetPageSplitter::split(req, self.shards.count, self.shards.stripe_size);

-        let mut shard_requests = FuturesUnordered::new();
-        for (shard_id, shard_req) in splitter.drain_requests() {
-            let future = Self::get_page_with_shard(shard_req, shards.get(shard_id)?)
-                .map(move |result| result.map(|resp| (shard_id, resp)));
-            shard_requests.push(future);
-        }
+        let mut shard_requests: FuturesUnordered<_> = splitter
+            .drain_requests()
+            .map(|(shard_id, shard_req)| {
+                // NB: each request will retry internally.
+                self.get_page_for_shard(shard_id, shard_req)
+                    .map(move |result| result.map(|resp| (shard_id, resp)))
+            })
+            .collect();

        while let Some((shard_id, shard_response)) = shard_requests.next().await.transpose()? {
            splitter.add_response(shard_id, shard_response)?;
        }

-        splitter.get_response()
+        splitter.assemble_response()
    }

-    /// Fetches pages on the given shard. Does not retry internally.
-    async fn get_page_with_shard(
+    /// Fetches pages that belong to the given shard.
+    #[instrument(skip_all, fields(shard = %shard_id))]
+    async fn get_page_for_shard(
+        &self,
+        shard_id: ShardIndex,
        req: page_api::GetPageRequest,
-        shard: &Shard,
    ) -> tonic::Result<page_api::GetPageResponse> {
-        let stream = shard.stream(req.request_class.is_bulk()).await;
-        let resp = stream.send(req.clone()).await?;
+        let resp = self
+            .retry
+            .with(async || {
+                let stream = self.shards.get(shard_id)?.stream().await;
+                let resp = stream.send(req.clone()).await?;

-        // Convert per-request errors into a tonic::Status.
-        if resp.status_code != page_api::GetPageStatusCode::Ok {
-            return Err(tonic::Status::new(
-                resp.status_code.into(),
-                resp.reason.unwrap_or_else(|| String::from("unknown error")),
-            ));
-        }
+                // Convert per-request errors into a tonic::Status.
+                if resp.status_code != page_api::GetPageStatusCode::Ok {
+                    return Err(tonic::Status::new(
+                        resp.status_code.into(),
+                        resp.reason.unwrap_or_else(|| String::from("unknown error")),
+                    ));
+                }

-        // Check that we received the expected pages.
-        if req.rel != resp.rel {
+                Ok(resp)
+            })
+            .await?;
+
+        // Make sure we got the right number of pages.
+        // NB: check outside of the retry loop, since we don't want to retry this.
+        let (expected, actual) = (req.block_numbers.len(), resp.page_images.len());
+        if expected != actual {
            return Err(tonic::Status::internal(format!(
-                "shard {} returned wrong relation, expected {} got {}",
-                shard.id, req.rel, resp.rel
-            )));
-        }
-        if !req
-            .block_numbers
-            .iter()
-            .copied()
-            .eq(resp.pages.iter().map(|p| p.block_number))
-        {
-            return Err(tonic::Status::internal(format!(
-                "shard {} returned wrong pages, expected {:?} got {:?}",
-                shard.id,
-                req.block_numbers,
-                resp.pages
-                    .iter()
-                    .map(|page| page.block_number)
-                    .collect::<Vec<_>>()
+                "expected {expected} pages for shard {shard_id}, got {actual}",
            )));
        }

@@ -291,9 +175,9 @@ impl PageserverClient {
        req: page_api::GetRelSizeRequest,
    ) -> tonic::Result<page_api::GetRelSizeResponse> {
        self.retry
-            .with(async |_| {
+            .with(async || {
                // Relation metadata is only available on shard 0.
-                let mut client = self.shards.load_full().get_zero().client().await?;
+                let mut client = self.shards.get_zero().client().await?;
                client.get_rel_size(req).await
            })
            .await
@@ -306,53 +190,50 @@ impl PageserverClient {
        req: page_api::GetSlruSegmentRequest,
    ) -> tonic::Result<page_api::GetSlruSegmentResponse> {
        self.retry
-            .with(async |_| {
+            .with(async || {
                // SLRU segments are only available on shard 0.
-                let mut client = self.shards.load_full().get_zero().client().await?;
+                let mut client = self.shards.get_zero().client().await?;
                client.get_slru_segment(req).await
            })
            .await
    }
 }

-/// Shard specification for a PageserverClient.
-pub struct ShardSpec {
-    /// Maps shard indices to gRPC URLs.
-    ///
-    /// INVARIANT: every shard 0..count is present, and shard 0 is always present.
-    /// INVARIANT: every URL is valid and uses grpc:// scheme.
-    urls: HashMap<ShardIndex, String>,
+/// Tracks the tenant's shards.
+struct Shards {
    /// The shard count.
    ///
    /// NB: this is 0 for unsharded tenants, following `ShardIndex::unsharded()` convention.
    count: ShardCount,
-    /// The stripe size for these shards.
+    /// The stripe size. Only used for sharded tenants.
    stripe_size: ShardStripeSize,
+    /// Shards by shard index.
+    ///
+    /// NB: unsharded tenants use count 0, like `ShardIndex::unsharded()`.
+    ///
+    /// INVARIANT: every shard 0..count is present.
+    /// INVARIANT: shard 0 is always present.
+    map: HashMap<ShardIndex, Shard>,
 }

-impl ShardSpec {
-    /// Creates a new shard spec with the given URLs and stripe size. All shards must be given.
-    /// The stripe size may be omitted for unsharded tenants.
-    pub fn new(
-        urls: HashMap<ShardIndex, String>,
-        stripe_size: Option<ShardStripeSize>,
+impl Shards {
+    /// Creates a new set of shards based on a shard map.
+    fn new(
+        tenant_id: TenantId,
+        timeline_id: TimelineId,
+        shard_map: HashMap<ShardIndex, String>,
+        stripe_size: ShardStripeSize,
+        auth_token: Option<String>,
    ) -> anyhow::Result<Self> {
-        // Compute the shard count.
-        let count = match urls.len() {
+        let count = match shard_map.len() {
            0 => return Err(anyhow!("no shards provided")),
            1 => ShardCount::new(0), // NB: unsharded tenants use 0, like `ShardIndex::unsharded()`
            n if n > u8::MAX as usize => return Err(anyhow!("too many shards: {n}")),
            n => ShardCount::new(n as u8),
        };

-        // Determine the stripe size. It doesn't matter for unsharded tenants.
-        if stripe_size.is_none() && !count.is_unsharded() {
-            return Err(anyhow!("stripe size must be given for sharded tenants"));
-        }
-        let stripe_size = stripe_size.unwrap_or_default();
-
-        // Validate the shard spec.
-        for (shard_id, url) in &urls {
+        let mut map = HashMap::new();
+        for (shard_id, url) in shard_map {
            // The shard index must match the computed shard count, even for unsharded tenants.
            if shard_id.shard_count != count {
                return Err(anyhow!("invalid shard index {shard_id}, expected {count}"));
@@ -363,72 +244,21 @@ impl ShardSpec {
            }
            // The above conditions guarantee that we have all shards 0..count: len() matches count,
            // shard number < count, and numbers are unique (via hashmap).
-
-            // Validate the URL.
-            if PageserverProtocol::from_connstring(url)? != PageserverProtocol::Grpc {
-                return Err(anyhow!("invalid shard URL {url}: must use gRPC"));
-            }
+            let shard = Shard::new(url, tenant_id, timeline_id, shard_id, auth_token.clone())?;
+            map.insert(shard_id, shard);
        }

        Ok(Self {
-            urls,
            count,
            stripe_size,
-        })
-    }
-}
-
-/// Tracks the tenant's shards.
-struct Shards {
-    /// Shards by shard index.
-    ///
-    /// INVARIANT: every shard 0..count is present.
-    /// INVARIANT: shard 0 is always present.
-    by_index: HashMap<ShardIndex, Shard>,
-    /// The shard count.
-    ///
-    /// NB: this is 0 for unsharded tenants, following `ShardIndex::unsharded()` convention.
-    count: ShardCount,
-    /// The stripe size. Only used for sharded tenants.
-    stripe_size: ShardStripeSize,
-}
-
-impl Shards {
-    /// Creates a new set of shards based on a shard spec.
-    fn new(
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-        shard_spec: ShardSpec,
-        auth_token: Option<String>,
-        compression: Option<CompressionEncoding>,
-    ) -> anyhow::Result<Self> {
-        // NB: the shard spec has already been validated when constructed.
-        let mut shards = HashMap::with_capacity(shard_spec.urls.len());
-        for (shard_id, url) in shard_spec.urls {
-            shards.insert(
-                shard_id,
-                Shard::new(
-                    url,
-                    tenant_id,
-                    timeline_id,
-                    shard_id,
-                    auth_token.clone(),
-                    compression,
-                )?,
-            );
-        }
-
-        Ok(Self {
-            by_index: shards,
-            count: shard_spec.count,
-            stripe_size: shard_spec.stripe_size,
+            map,
        })
    }

    /// Looks up the given shard.
    #[allow(clippy::result_large_err)] // TODO: check perf impact
    fn get(&self, shard_id: ShardIndex) -> tonic::Result<&Shard> {
-        self.by_index
+        self.map
            .get(&shard_id)
            .ok_or_else(|| tonic::Status::not_found(format!("unknown shard {shard_id}")))
    }
@@ -440,24 +270,17 @@ impl Shards {
    }
 }

-/// A single shard. Uses dedicated resource pools with the following structure:
+/// A single shard.
 ///
-/// * Channel pool: unbounded.
-///   * Unary client pool: MAX_UNARY_CLIENTS.
-///   * Stream client pool: unbounded.
-///     * Stream pool: MAX_STREAMS and MAX_STREAM_QUEUE_DEPTH.
-/// * Bulk channel pool: unbounded.
-///   * Bulk client pool: unbounded.
-///     * Bulk stream pool: MAX_BULK_STREAMS and MAX_BULK_STREAM_QUEUE_DEPTH.
+/// TODO: consider separate pools for normal and bulk traffic, with different settings.
 struct Shard {
-    /// The shard ID.
-    id: ShardIndex,
-    /// Unary gRPC client pool.
+    /// Dedicated channel pool for this shard. Shared by all clients/streams in this shard.
+    _channel_pool: Arc<ChannelPool>,
+    /// Unary gRPC client pool for this shard. Uses the shared channel pool.
    client_pool: Arc<ClientPool>,
-    /// GetPage stream pool.
+    /// GetPage stream pool for this shard. Uses a dedicated client pool, but shares the channel
+    /// pool with unary clients.
    stream_pool: Arc<StreamPool>,
-    /// GetPage stream pool for bulk requests, e.g. prefetches.
-    bulk_stream_pool: Arc<StreamPool>,
 }

 impl Shard {
@@ -468,59 +291,40 @@ impl Shard {
        timeline_id: TimelineId,
        shard_id: ShardIndex,
        auth_token: Option<String>,
-        compression: Option<CompressionEncoding>,
    ) -> anyhow::Result<Self> {
-        // Common channel pool for unary and stream requests. Bounded by client/stream pools.
-        let channel_pool = ChannelPool::new(url.clone(), MAX_CLIENTS_PER_CHANNEL)?;
+        // Sanity-check that the URL uses gRPC.
+        if PageserverProtocol::from_connstring(&url)? != PageserverProtocol::Grpc {
+            return Err(anyhow!("invalid shard URL {url}: must use gRPC"));
+        }

-        // Client pool for unary requests.
+        // Use a common channel pool for all clients, to multiplex unary and stream requests across
+        // the same TCP connections. The channel pool is unbounded (but client pools are bounded).
+        let channel_pool = ChannelPool::new(url)?;
+
+        // Dedicated client pool for unary requests.
        let client_pool = ClientPool::new(
            channel_pool.clone(),
            tenant_id,
            timeline_id,
            shard_id,
            auth_token.clone(),
-            compression,
-            Some(MAX_UNARY_CLIENTS),
        );

-        // GetPage stream pool. Uses a dedicated client pool to avoid starving out unary clients,
-        // but shares a channel pool with it (as it's unbounded).
-        let stream_pool = StreamPool::new(
-            ClientPool::new(
-                channel_pool.clone(),
-                tenant_id,
-                timeline_id,
-                shard_id,
-                auth_token.clone(),
-                compression,
-                None, // unbounded, limited by stream pool
-            ),
-            Some(MAX_STREAMS),
-            MAX_STREAM_QUEUE_DEPTH,
-        );
-
-        // Bulk GetPage stream pool, e.g. for prefetches. Uses dedicated channel/client/stream pools
-        // to avoid head-of-line blocking of latency-sensitive requests.
-        let bulk_stream_pool = StreamPool::new(
-            ClientPool::new(
-                ChannelPool::new(url, MAX_CLIENTS_PER_CHANNEL)?,
-                tenant_id,
-                timeline_id,
-                shard_id,
-                auth_token,
-                compression,
-                None, // unbounded, limited by stream pool
-            ),
-            Some(MAX_BULK_STREAMS),
-            MAX_BULK_STREAM_QUEUE_DEPTH,
-        );
+        // Stream pool with dedicated client pool. If this shared a client pool with unary requests,
+        // long-lived streams could fill up the client pool and starve out unary requests. It shares
+        // the same underlying channel pool with unary clients though, which is unbounded.
+        let stream_pool = StreamPool::new(ClientPool::new(
+            channel_pool.clone(),
+            tenant_id,
+            timeline_id,
+            shard_id,
+            auth_token,
+        ));

        Ok(Self {
-            id: shard_id,
+            _channel_pool: channel_pool,
            client_pool,
            stream_pool,
-            bulk_stream_pool,
        })
    }

@@ -532,12 +336,8 @@ impl Shard {
            .map_err(|err| tonic::Status::internal(format!("failed to get client: {err}")))
    }

-    /// Returns a pooled stream for this shard. If `bulk` is `true`, uses the dedicated bulk stream
-    /// pool (e.g. for prefetches).
-    async fn stream(&self, bulk: bool) -> StreamGuard {
-        match bulk {
-            false => self.stream_pool.get().await,
-            true => self.bulk_stream_pool.get().await,
-        }
+    /// Returns a pooled stream for this shard.
+    async fn stream(&self) -> StreamGuard {
+        self.stream_pool.get().await
    }
 }
--- a/pageserver/client_grpc/src/lib.rs
+++ b/pageserver/client_grpc/src/lib.rs
@@ -3,4 +3,4 @@ mod pool;
 mod retry;
 mod split;

-pub use client::{PageserverClient, ShardSpec};
+pub use client::PageserverClient;
--- a/pageserver/client_grpc/src/pool.rs
+++ b/pageserver/client_grpc/src/pool.rs
@@ -30,17 +30,13 @@
 //! TODO: observability.

 use std::collections::{BTreeMap, HashMap};
-use std::num::NonZero;
 use std::ops::{Deref, DerefMut};
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex, Weak};
-use std::time::{Duration, Instant};

 use futures::StreamExt as _;
 use tokio::sync::mpsc::{Receiver, Sender};
 use tokio::sync::{OwnedSemaphorePermit, Semaphore, mpsc, oneshot};
-use tokio_util::sync::CancellationToken;
-use tonic::codec::CompressionEncoding;
 use tonic::transport::{Channel, Endpoint};
 use tracing::{error, warn};

@@ -48,43 +44,33 @@ use pageserver_page_api as page_api;
 use utils::id::{TenantId, TimelineId};
 use utils::shard::ShardIndex;

-/// Reap channels/clients/streams that have been idle for this long.
+/// Max number of concurrent clients per channel.
 ///
-/// TODO: this is per-pool. For nested pools, it can take up to 3x as long for a TCP connection to
-/// be reaped. First, we must wait for an idle stream to be reaped, which marks its client as idle.
-/// Then, we must wait for the idle client to be reaped, which marks its channel as idle. Then, we
-/// must wait for the idle channel to be reaped. Is that a problem? Maybe not, we just have to
-/// account for it when setting the reap threshold. Alternatively, we can immediately reap empty
-/// channels, and/or stream pool clients.
-const REAP_IDLE_THRESHOLD: Duration = match cfg!(any(test, feature = "testing")) {
-    false => Duration::from_secs(180),
-    true => Duration::from_secs(1), // exercise reaping in tests
-};
+/// TODO: tune these constants, and make them configurable.
+/// TODO: consider separate limits for unary and streaming clients, so we don't fill up channels
+/// with only streams.
+const CLIENTS_PER_CHANNEL: usize = 16;

-/// Reap idle resources with this interval.
-const REAP_IDLE_INTERVAL: Duration = match cfg!(any(test, feature = "testing")) {
-    false => Duration::from_secs(10),
-    true => Duration::from_secs(1), // exercise reaping in tests
-};
+/// Maximum number of concurrent clients per `ClientPool`.
+const CLIENT_LIMIT: usize = 64;
+
+/// Max number of pipelined requests per gRPC GetPage stream.
+const STREAM_QUEUE_DEPTH: usize = 2;

 /// A gRPC channel pool, for a single Pageserver. A channel is shared by many clients (via HTTP/2
-/// stream multiplexing), up to `clients_per_channel` -- a new channel will be spun up beyond this.
-/// The pool does not limit the number of channels, and instead relies on `ClientPool` or
-/// `StreamPool` to limit the number of concurrent clients.
+/// stream multiplexing), up to `CLIENTS_PER_CHANNEL`. The pool does not limit the number of
+/// channels, and instead relies on `ClientPool` to limit the number of concurrent clients.
 ///
 /// The pool is always wrapped in an outer `Arc`, to allow long-lived guards across tasks/threads.
 ///
+/// TODO: reap idle channels.
 /// TODO: consider prewarming a set of channels, to avoid initial connection latency.
 /// TODO: consider adding a circuit breaker for errors and fail fast.
 pub struct ChannelPool {
    /// Pageserver endpoint to connect to.
    endpoint: Endpoint,
-    /// Max number of clients per channel. Beyond this, a new channel will be created.
-    max_clients_per_channel: NonZero<usize>,
    /// Open channels.
    channels: Mutex<BTreeMap<ChannelID, ChannelEntry>>,
-    /// Reaps idle channels.
-    idle_reaper: Reaper,
    /// Channel ID generator.
    next_channel_id: AtomicUsize,
 }
@@ -96,27 +82,20 @@ struct ChannelEntry {
    channel: Channel,
    /// Number of clients using this channel.
    clients: usize,
-    /// The channel has been idle (no clients) since this time. None if channel is in use.
-    /// INVARIANT: Some if clients == 0, otherwise None.
-    idle_since: Option<Instant>,
 }

 impl ChannelPool {
    /// Creates a new channel pool for the given Pageserver endpoint.
-    pub fn new<E>(endpoint: E, max_clients_per_channel: NonZero<usize>) -> anyhow::Result<Arc<Self>>
+    pub fn new<E>(endpoint: E) -> anyhow::Result<Arc<Self>>
    where
        E: TryInto<Endpoint> + Send + Sync + 'static,
        <E as TryInto<Endpoint>>::Error: std::error::Error + Send + Sync,
    {
-        let pool = Arc::new(Self {
+        Ok(Arc::new(Self {
            endpoint: endpoint.try_into()?,
-            max_clients_per_channel,
            channels: Mutex::default(),
-            idle_reaper: Reaper::new(REAP_IDLE_THRESHOLD, REAP_IDLE_INTERVAL),
            next_channel_id: AtomicUsize::default(),
-        });
-        pool.idle_reaper.spawn(&pool);
-        Ok(pool)
+        }))
    }

    /// Acquires a gRPC channel for a client. Multiple clients may acquire the same channel.
@@ -141,18 +120,9 @@ impl ChannelPool {
        // with lower-ordered channel IDs first. This will cluster clients in lower-ordered
        // channels, and free up higher-ordered channels such that they can be reaped.
        for (&id, entry) in channels.iter_mut() {
-            assert!(
-                entry.clients <= self.max_clients_per_channel.get(),
-                "channel overflow"
-            );
-            assert_eq!(
-                entry.idle_since.is_some(),
-                entry.clients == 0,
-                "incorrect channel idle state"
-            );
-            if entry.clients < self.max_clients_per_channel.get() {
+            assert!(entry.clients <= CLIENTS_PER_CHANNEL, "channel overflow");
+            if entry.clients < CLIENTS_PER_CHANNEL {
                entry.clients += 1;
-                entry.idle_since = None;
                return ChannelGuard {
                    pool: Arc::downgrade(self),
                    id,
@@ -169,7 +139,6 @@ impl ChannelPool {
        let entry = ChannelEntry {
            channel: channel.clone(),
            clients: 1, // account for the guard below
-            idle_since: None,
        };
        channels.insert(id, entry);

@@ -181,20 +150,6 @@ impl ChannelPool {
    }
 }

-impl Reapable for ChannelPool {
-    /// Reaps channels that have been idle since before the cutoff.
-    fn reap_idle(&self, cutoff: Instant) {
-        self.channels.lock().unwrap().retain(|_, entry| {
-            let Some(idle_since) = entry.idle_since else {
-                assert_ne!(entry.clients, 0, "empty channel not marked idle");
-                return true;
-            };
-            assert_eq!(entry.clients, 0, "idle channel has clients");
-            idle_since >= cutoff
-        })
-    }
-}
-
 /// Tracks a channel acquired from the pool. The owned inner channel can be obtained with `take()`,
 /// since the gRPC client requires an owned `Channel`.
 pub struct ChannelGuard {
@@ -217,23 +172,20 @@ impl Drop for ChannelGuard {
        let Some(pool) = self.pool.upgrade() else {
            return; // pool was dropped
        };
-
        let mut channels = pool.channels.lock().unwrap();
        let entry = channels.get_mut(&self.id).expect("unknown channel");
-        assert!(entry.idle_since.is_none(), "active channel marked idle");
        assert!(entry.clients > 0, "channel underflow");
        entry.clients -= 1;
-        if entry.clients == 0 {
-            entry.idle_since = Some(Instant::now()); // mark channel as idle
-        }
    }
 }

 /// A pool of gRPC clients for a single tenant shard. Each client acquires a channel from the inner
 /// `ChannelPool`. A client is only given out to single caller at a time. The pool limits the total
-/// number of concurrent clients to `max_clients` via semaphore.
+/// number of concurrent clients to `CLIENT_LIMIT` via semaphore.
 ///
 /// The pool is always wrapped in an outer `Arc`, to allow long-lived guards across tasks/threads.
+///
+/// TODO: reap idle clients.
 pub struct ClientPool {
    /// Tenant ID.
    tenant_id: TenantId,
@@ -243,12 +195,10 @@ pub struct ClientPool {
    shard_id: ShardIndex,
    /// Authentication token, if any.
    auth_token: Option<String>,
-    /// Compression to use.
-    compression: Option<CompressionEncoding>,
    /// Channel pool to acquire channels from.
    channel_pool: Arc<ChannelPool>,
-    /// Limits the max number of concurrent clients for this pool. None if the pool is unbounded.
-    limiter: Option<Arc<Semaphore>>,
+    /// Limits the max number of concurrent clients for this pool.
+    limiter: Arc<Semaphore>,
    /// Idle pooled clients. Acquired clients are removed from here and returned on drop.
    ///
    /// The first client in the map will be acquired next. The map is sorted by client ID, which in
@@ -256,8 +206,6 @@ pub struct ClientPool {
    /// lower-ordered channels. This allows us to free up and reap higher-numbered channels as idle
    /// clients are reaped.
    idle: Mutex<BTreeMap<ClientID, ClientEntry>>,
-    /// Reaps idle clients.
-    idle_reaper: Reaper,
    /// Unique client ID generator.
    next_client_id: AtomicUsize,
 }
@@ -269,53 +217,44 @@ struct ClientEntry {
    client: page_api::Client,
    /// The channel guard for the channel used by the client.
    channel_guard: ChannelGuard,
-    /// The client has been idle since this time. All clients in `ClientPool::idle` are idle by
-    /// definition, so this is the time when it was added back to the pool.
-    idle_since: Instant,
 }

 impl ClientPool {
    /// Creates a new client pool for the given tenant shard. Channels are acquired from the given
-    /// `ChannelPool`, which must point to a Pageserver that hosts the tenant shard. Allows up to
-    /// `max_clients` concurrent clients, or unbounded if None.
+    /// `ChannelPool`, which must point to a Pageserver that hosts the tenant shard.
    pub fn new(
        channel_pool: Arc<ChannelPool>,
        tenant_id: TenantId,
        timeline_id: TimelineId,
        shard_id: ShardIndex,
        auth_token: Option<String>,
-        compression: Option<CompressionEncoding>,
-        max_clients: Option<NonZero<usize>>,
    ) -> Arc<Self> {
-        let pool = Arc::new(Self {
+        Arc::new(Self {
            tenant_id,
            timeline_id,
            shard_id,
            auth_token,
-            compression,
            channel_pool,
            idle: Mutex::default(),
-            idle_reaper: Reaper::new(REAP_IDLE_THRESHOLD, REAP_IDLE_INTERVAL),
-            limiter: max_clients.map(|max| Arc::new(Semaphore::new(max.get()))),
+            limiter: Arc::new(Semaphore::new(CLIENT_LIMIT)),
            next_client_id: AtomicUsize::default(),
-        });
-        pool.idle_reaper.spawn(&pool);
-        pool
+        })
    }

    /// Gets a client from the pool, or creates a new one if necessary. Connections are established
-    /// lazily and do not block, but this call can block if the pool is at `max_clients`. The client
-    /// is returned to the pool when the guard is dropped.
+    /// lazily and do not block, but this call can block if the pool is at `CLIENT_LIMIT`. The
+    /// client is returned to the pool when the guard is dropped.
    ///
    /// This is moderately performance-sensitive. It is called for every unary request, but these
    /// establish a new gRPC stream per request so they're already expensive. GetPage requests use
    /// the `StreamPool` instead.
    pub async fn get(self: &Arc<Self>) -> anyhow::Result<ClientGuard> {
-        // Acquire a permit if the pool is bounded.
-        let mut permit = None;
-        if let Some(limiter) = self.limiter.clone() {
-            permit = Some(limiter.acquire_owned().await.expect("never closed"));
-        }
+        let permit = self
+            .limiter
+            .clone()
+            .acquire_owned()
+            .await
+            .expect("never closed");

        // Fast path: acquire an idle client from the pool.
        if let Some((id, entry)) = self.idle.lock().unwrap().pop_first() {
@@ -336,7 +275,7 @@ impl ClientPool {
            self.timeline_id,
            self.shard_id,
            self.auth_token.clone(),
-            self.compression,
+            None,
        )?;

        Ok(ClientGuard {
@@ -352,24 +291,14 @@ impl ClientPool {
    }
 }

-impl Reapable for ClientPool {
-    /// Reaps clients that have been idle since before the cutoff.
-    fn reap_idle(&self, cutoff: Instant) {
-        self.idle
-            .lock()
-            .unwrap()
-            .retain(|_, entry| entry.idle_since >= cutoff)
-    }
-}
-
 /// A client acquired from the pool. The inner client can be accessed via Deref. The client is
 /// returned to the pool when dropped.
 pub struct ClientGuard {
    pool: Weak<ClientPool>,
    id: ClientID,
-    client: Option<page_api::Client>,     // Some until dropped
-    channel_guard: Option<ChannelGuard>,  // Some until dropped
-    permit: Option<OwnedSemaphorePermit>, // None if pool is unbounded
+    client: Option<page_api::Client>,    // Some until dropped
+    channel_guard: Option<ChannelGuard>, // Some until dropped
+    permit: OwnedSemaphorePermit,
 }

 impl Deref for ClientGuard {
@@ -392,11 +321,9 @@ impl Drop for ClientGuard {
        let Some(pool) = self.pool.upgrade() else {
            return; // pool was dropped
        };
-
        let entry = ClientEntry {
            client: self.client.take().expect("dropped once"),
            channel_guard: self.channel_guard.take().expect("dropped once"),
-            idle_since: Instant::now(),
        };
        pool.idle.lock().unwrap().insert(self.id, entry);

@@ -411,25 +338,19 @@ impl Drop for ClientGuard {
 /// a single request and await the response. Internally, requests are multiplexed across streams and
 /// channels. This allows proper queue depth enforcement and response routing.
 ///
+/// TODO: reap idle streams.
 /// TODO: consider making this generic over request and response types; not currently needed.
 pub struct StreamPool {
-    /// The client pool to acquire clients from. Must be unbounded.
+    /// The client pool to acquire clients from.
    client_pool: Arc<ClientPool>,
    /// All pooled streams.
    ///
    /// Incoming requests will be sent over an existing stream with available capacity. If all
-    /// streams are full, a new one is spun up and added to the pool (up to `max_streams`). Each
-    /// stream has an associated Tokio task that processes requests and responses.
-    streams: Mutex<HashMap<StreamID, StreamEntry>>,
-    /// The max number of concurrent streams, or None if unbounded.
-    max_streams: Option<NonZero<usize>>,
-    /// The max number of concurrent requests per stream.
-    max_queue_depth: NonZero<usize>,
-    /// Limits the max number of concurrent requests, given by `max_streams * max_queue_depth`.
-    /// None if the pool is unbounded.
-    limiter: Option<Arc<Semaphore>>,
-    /// Reaps idle streams.
-    idle_reaper: Reaper,
+    /// streams are full, a new one is spun up and added to the pool (up to the `ClientPool` limit).
+    /// Each stream has an associated Tokio task that processes requests and responses.
+    streams: Arc<Mutex<HashMap<StreamID, StreamEntry>>>,
+    /// Limits the max number of concurrent requests (not streams).
+    limiter: Arc<Semaphore>,
    /// Stream ID generator.
    next_stream_id: AtomicUsize,
 }
@@ -442,40 +363,24 @@ type ResponseSender = oneshot::Sender<tonic::Result<page_api::GetPageResponse>>;
 struct StreamEntry {
    /// Sends caller requests to the stream task. The stream task exits when this is dropped.
    sender: RequestSender,
-    /// Number of in-flight requests on this stream.
-    queue_depth: usize,
-    /// The time when this stream went idle (queue_depth == 0).
-    /// INVARIANT: Some if queue_depth == 0, otherwise None.
-    idle_since: Option<Instant>,
+    /// Number of in-flight requests on this stream. This is an atomic to allow decrementing it on
+    /// completion without acquiring the `StreamPool::streams` lock.
+    queue_depth: Arc<AtomicUsize>,
 }

 impl StreamPool {
-    /// Creates a new stream pool, using the given client pool. It will send up to `max_queue_depth`
-    /// concurrent requests on each stream, and use up to `max_streams` concurrent streams.
+    /// Creates a new stream pool, using the given client pool.
    ///
-    /// The client pool must be unbounded. The stream pool will enforce its own limits, and because
-    /// streams are long-lived they can cause persistent starvation if they exhaust the client pool.
-    /// The stream pool should generally have its own dedicated client pool (but it can share a
-    /// channel pool with others since these are always unbounded).
-    pub fn new(
-        client_pool: Arc<ClientPool>,
-        max_streams: Option<NonZero<usize>>,
-        max_queue_depth: NonZero<usize>,
-    ) -> Arc<Self> {
-        assert!(client_pool.limiter.is_none(), "bounded client pool");
-        let pool = Arc::new(Self {
+    /// NB: the stream pool should use a dedicated client pool. Otherwise, long-lived streams may
+    /// fill up the client pool and starve out unary requests. Client pools can share the same
+    /// `ChannelPool` though, since the channel pool is unbounded.
+    pub fn new(client_pool: Arc<ClientPool>) -> Arc<Self> {
+        Arc::new(Self {
            client_pool,
-            streams: Mutex::default(),
-            limiter: max_streams.map(|max_streams| {
-                Arc::new(Semaphore::new(max_streams.get() * max_queue_depth.get()))
-            }),
-            max_streams,
-            max_queue_depth,
-            idle_reaper: Reaper::new(REAP_IDLE_THRESHOLD, REAP_IDLE_INTERVAL),
+            streams: Arc::default(),
+            limiter: Arc::new(Semaphore::new(CLIENT_LIMIT * STREAM_QUEUE_DEPTH)),
            next_stream_id: AtomicUsize::default(),
-        });
-        pool.idle_reaper.spawn(&pool);
-        pool
+        })
    }

    /// Acquires an available stream from the pool, or spins up a new stream async if all streams
@@ -495,33 +400,34 @@ impl StreamPool {
    /// * Allow concurrent clients to join onto streams while they're spun up.
    /// * Allow spinning up multiple streams concurrently, but don't overshoot limits.
    ///
-    /// For now, we just do something simple but inefficient (linear scan under mutex).
-    pub async fn get(self: &Arc<Self>) -> StreamGuard {
-        // Acquire a permit if the pool is bounded.
-        let mut permit = None;
-        if let Some(limiter) = self.limiter.clone() {
-            permit = Some(limiter.acquire_owned().await.expect("never closed"));
-        }
+    /// For now, we just do something simple and functional, but very inefficient (linear scan).
+    pub async fn get(&self) -> StreamGuard {
+        let permit = self
+            .limiter
+            .clone()
+            .acquire_owned()
+            .await
+            .expect("never closed");
        let mut streams = self.streams.lock().unwrap();

        // Look for a pooled stream with available capacity.
-        for (&id, entry) in streams.iter_mut() {
+        for entry in streams.values() {
            assert!(
-                entry.queue_depth <= self.max_queue_depth.get(),
+                entry.queue_depth.load(Ordering::Relaxed) <= STREAM_QUEUE_DEPTH,
                "stream queue overflow"
            );
-            assert_eq!(
-                entry.idle_since.is_some(),
-                entry.queue_depth == 0,
-                "incorrect stream idle state"
-            );
-            if entry.queue_depth < self.max_queue_depth.get() {
-                entry.queue_depth += 1;
-                entry.idle_since = None;
+            if entry
+                .queue_depth
+                .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |queue_depth| {
+                    // Increment the queue depth via compare-and-swap.
+                    // TODO: review ordering.
+                    (queue_depth < STREAM_QUEUE_DEPTH).then_some(queue_depth + 1)
+                })
+                .is_ok()
+            {
                return StreamGuard {
-                    pool: Arc::downgrade(self),
-                    id,
                    sender: entry.sender.clone(),
+                    queue_depth: entry.queue_depth.clone(),
                    permit,
                };
            }
@@ -531,36 +437,35 @@ impl StreamPool {
        // return the guard, while spinning up the stream task async. This allows other callers to
        // join onto this stream and also create additional streams concurrently if this fills up.
        let id = self.next_stream_id.fetch_add(1, Ordering::Relaxed);
-        let (req_tx, req_rx) = mpsc::channel(self.max_queue_depth.get());
+        let queue_depth = Arc::new(AtomicUsize::new(1)); // reserve quota for this caller
+        let (req_tx, req_rx) = mpsc::channel(STREAM_QUEUE_DEPTH);
        let entry = StreamEntry {
            sender: req_tx.clone(),
-            queue_depth: 1, // reserve quota for this caller
-            idle_since: None,
+            queue_depth: queue_depth.clone(),
        };
        streams.insert(id, entry);

-        if let Some(max_streams) = self.max_streams {
-            assert!(streams.len() <= max_streams.get(), "stream overflow");
-        };
+        // NB: make sure we don't overshoot the client limit. The semaphore limit is CLIENT_LIMIT *
+        // STREAM_QUEUE_DEPTH, but if we were to misaccount queue depth we'd try to spin up more
+        // streams than CLIENT_LIMIT and block on the client pool ~forever. This should not happen
+        // because we only acquire queue depth under lock and after acquiring a semaphore permit.
+        assert!(streams.len() <= CLIENT_LIMIT, "stream overflow");

        let client_pool = self.client_pool.clone();
-        let pool = Arc::downgrade(self);
+        let streams = self.streams.clone();

        tokio::spawn(async move {
            if let Err(err) = Self::run_stream(client_pool, req_rx).await {
                error!("stream failed: {err}");
            }
-            // Remove stream from pool on exit. Weak reference to avoid holding the pool alive.
-            if let Some(pool) = pool.upgrade() {
-                let entry = pool.streams.lock().unwrap().remove(&id);
-                assert!(entry.is_some(), "unknown stream ID: {id}");
-            }
+            // Remove stream from pool on exit.
+            let entry = streams.lock().unwrap().remove(&id);
+            assert!(entry.is_some(), "unknown stream ID: {id}");
        });

        StreamGuard {
-            pool: Arc::downgrade(self),
-            id,
            sender: req_tx,
+            queue_depth,
            permit,
        }
    }
@@ -579,26 +484,19 @@ impl StreamPool {
        // Acquire a client from the pool and create a stream.
        let mut client = client_pool.get().await?;

-        // NB: use an unbounded channel such that the stream send never blocks. Otherwise, we could
-        // theoretically deadlock if both the client and server block on sends (since we're not
-        // reading responses while sending). This is unlikely to happen due to gRPC/TCP buffers and
-        // low queue depths, but it was seen to happen with the libpq protocol so better safe than
-        // sorry. It should never buffer more than the queue depth anyway, but using an unbounded
-        // channel guarantees that it will never block.
-        let (req_tx, req_rx) = mpsc::unbounded_channel();
-        let req_stream = tokio_stream::wrappers::UnboundedReceiverStream::new(req_rx);
+        let (req_tx, req_rx) = mpsc::channel(STREAM_QUEUE_DEPTH);
+        let req_stream = tokio_stream::wrappers::ReceiverStream::new(req_rx);
        let mut resp_stream = client.get_pages(req_stream).await?;

        // Track caller response channels by request ID. If the task returns early, these response
        // channels will be dropped and the waiting callers will receive an error.
-        //
-        // NB: this will leak entries if the server doesn't respond to a request (by request ID).
-        // It shouldn't happen, and if it does it will often hold onto queue depth quota anyway and
-        // block further use. But we could consider reaping closed channels after some time.
-        let mut callers = HashMap::new();
+        let mut callers = HashMap::with_capacity(STREAM_QUEUE_DEPTH);

        // Process requests and responses.
        loop {
+            // NB: this can trip if the server doesn't respond to a request, so only debug_assert.
+            debug_assert!(callers.len() <= STREAM_QUEUE_DEPTH, "stream queue overflow");
+
            tokio::select! {
                // Receive requests from callers and send them to the stream.
                req = caller_rx.recv() => {
@@ -617,8 +515,8 @@ impl StreamPool {
                    }
                    callers.insert(req.request_id, resp_tx);

-                    // Send the request on the stream. Bail out if the stream is closed.
-                    req_tx.send(req).map_err(|_| {
+                    // Send the request on the stream. Bail out if the send fails.
+                    req_tx.send(req).await.map_err(|_| {
                        tonic::Status::unavailable("stream closed")
                    })?;
                }
@@ -642,27 +540,12 @@ impl StreamPool {
    }
 }

-impl Reapable for StreamPool {
-    /// Reaps streams that have been idle since before the cutoff.
-    fn reap_idle(&self, cutoff: Instant) {
-        self.streams.lock().unwrap().retain(|_, entry| {
-            let Some(idle_since) = entry.idle_since else {
-                assert_ne!(entry.queue_depth, 0, "empty stream not marked idle");
-                return true;
-            };
-            assert_eq!(entry.queue_depth, 0, "idle stream has requests");
-            idle_since >= cutoff
-        });
-    }
-}
-
 /// A pooled stream reference. Can be used to send a single request, to properly enforce queue
 /// depth. Queue depth is already reserved and will be returned on drop.
 pub struct StreamGuard {
-    pool: Weak<StreamPool>,
-    id: StreamID,
    sender: RequestSender,
-    permit: Option<OwnedSemaphorePermit>, // None if pool is unbounded
+    queue_depth: Arc<AtomicUsize>,
+    permit: OwnedSemaphorePermit,
 }

 impl StreamGuard {
@@ -693,87 +576,11 @@ impl StreamGuard {

 impl Drop for StreamGuard {
    fn drop(&mut self) {
-        let Some(pool) = self.pool.upgrade() else {
-            return; // pool was dropped
-        };
-
        // Release the queue depth reservation on drop. This can prematurely decrement it if dropped
        // before the response is received, but that's okay.
-        //
-        // TODO: actually, it's probably not okay. Queue depth release should be moved into the
-        // stream task, such that it continues to account for the queue depth slot until the server
-        // responds. Otherwise, if a slow request times out and keeps blocking the stream, the
-        // server will keep waiting on it and we can pile on subsequent requests (including the
-        // timeout retry) in the same stream and get blocked. But we may also want to avoid blocking
-        // requests on e.g. LSN waits and layer downloads, instead returning early to free up the
-        // stream. Or just scale out streams with a queue depth of 1 to sidestep all head-of-line
-        // blocking. TBD.
-        let mut streams = pool.streams.lock().unwrap();
-        let entry = streams.get_mut(&self.id).expect("unknown stream");
-        assert!(entry.idle_since.is_none(), "active stream marked idle");
-        assert!(entry.queue_depth > 0, "stream queue underflow");
-        entry.queue_depth -= 1;
-        if entry.queue_depth == 0 {
-            entry.idle_since = Some(Instant::now()); // mark stream as idle
-        }
+        let prev_queue_depth = self.queue_depth.fetch_sub(1, Ordering::SeqCst);
+        assert!(prev_queue_depth > 0, "stream queue underflow");

        _ = self.permit; // returned on drop, referenced for visibility
    }
 }
-
-/// Periodically reaps idle resources from a pool.
-struct Reaper {
-    /// The task check interval.
-    interval: Duration,
-    /// The threshold for reaping idle resources.
-    threshold: Duration,
-    /// Cancels the reaper task. Cancelled when the reaper is dropped.
-    cancel: CancellationToken,
-}
-
-impl Reaper {
-    /// Creates a new reaper.
-    pub fn new(threshold: Duration, interval: Duration) -> Self {
-        Self {
-            cancel: CancellationToken::new(),
-            threshold,
-            interval,
-        }
-    }
-
-    /// Spawns a task to periodically reap idle resources from the given task pool. The task is
-    /// cancelled when the reaper is dropped.
-    pub fn spawn(&self, pool: &Arc<impl Reapable>) {
-        // NB: hold a weak pool reference, otherwise the task will prevent dropping the pool.
-        let pool = Arc::downgrade(pool);
-        let cancel = self.cancel.clone();
-        let (interval, threshold) = (self.interval, self.threshold);
-
-        tokio::spawn(async move {
-            loop {
-                tokio::select! {
-                    _ = tokio::time::sleep(interval) => {
-                        let Some(pool) = pool.upgrade() else {
-                            return; // pool was dropped
-                        };
-                        pool.reap_idle(Instant::now() - threshold);
-                    }
-
-                    _ = cancel.cancelled() => return,
-                }
-            }
-        });
-    }
-}
-
-impl Drop for Reaper {
-    fn drop(&mut self) {
-        self.cancel.cancel(); // cancel reaper task
-    }
-}
-
-/// A reapable resource pool.
-trait Reapable: Send + Sync + 'static {
-    /// Reaps resources that have been idle since before the given cutoff.
-    fn reap_idle(&self, cutoff: Instant);
-}
--- a/pageserver/client_grpc/src/retry.rs
+++ b/pageserver/client_grpc/src/retry.rs
@@ -23,14 +23,14 @@ impl Retry {
    /// If true, log successful requests. For debugging.
    const LOG_SUCCESS: bool = false;

-    /// Runs the given async closure with timeouts and retries (exponential backoff), passing the
-    /// attempt number starting at 0. Logs errors, using the current tracing span for context.
+    /// Runs the given async closure with timeouts and retries (exponential backoff). Logs errors,
+    /// using the current tracing span for context.
    ///
    /// Only certain gRPC status codes are retried, see [`Self::should_retry`]. For default
    /// timeouts, see [`Self::REQUEST_TIMEOUT`] and [`Self::TOTAL_TIMEOUT`].
    pub async fn with<T, F, O>(&self, mut f: F) -> tonic::Result<T>
    where
-        F: FnMut(usize) -> O, // takes attempt number, starting at 0
+        F: FnMut() -> O,
        O: Future<Output = tonic::Result<T>>,
    {
        let started = Instant::now();
@@ -47,7 +47,7 @@ impl Retry {
                }

                let request_started = Instant::now();
-                tokio::time::timeout(Self::REQUEST_TIMEOUT, f(retries))
+                tokio::time::timeout(Self::REQUEST_TIMEOUT, f())
                    .await
                    .map_err(|_| {
                        tonic::Status::deadline_exceeded(format!(
@@ -131,6 +131,7 @@ impl Retry {
            tonic::Code::Aborted => true,
            tonic::Code::Cancelled => true,
            tonic::Code::DeadlineExceeded => true, // maybe transient slowness
+            tonic::Code::Internal => true,         // maybe transient failure?
            tonic::Code::ResourceExhausted => true,
            tonic::Code::Unavailable => true,

@@ -138,10 +139,6 @@ impl Retry {
            tonic::Code::AlreadyExists => false,
            tonic::Code::DataLoss => false,
            tonic::Code::FailedPrecondition => false,
-            // NB: don't retry Internal. It is intended for serious errors such as invariant
-            // violations, and is also used for client-side invariant checks that would otherwise
-            // result in retry loops.
-            tonic::Code::Internal => false,
            tonic::Code::InvalidArgument => false,
            tonic::Code::NotFound => false,
            tonic::Code::OutOfRange => false,
--- a/pageserver/client_grpc/src/split.rs
+++ b/pageserver/client_grpc/src/split.rs
@@ -5,24 +5,27 @@ use bytes::Bytes;
 use pageserver_api::key::rel_block_to_key;
 use pageserver_api::shard::{ShardStripeSize, key_to_shard_number};
 use pageserver_page_api as page_api;
-use utils::shard::{ShardCount, ShardIndex, ShardNumber};
+use utils::shard::{ShardCount, ShardIndex};

 /// Splits GetPageRequests that straddle shard boundaries and assembles the responses.
 /// TODO: add tests for this.
 pub struct GetPageSplitter {
+    /// The original request ID. Used for all shard requests.
+    request_id: page_api::RequestID,
    /// Split requests by shard index.
    requests: HashMap<ShardIndex, page_api::GetPageRequest>,
-    /// The response being assembled. Preallocated with empty pages, to be filled in.
-    response: page_api::GetPageResponse,
-    /// Maps the offset in `request.block_numbers` and `response.pages` to the owning shard. Used
-    /// to assemble the response pages in the same order as the original request.
+    /// Maps the offset in `GetPageRequest::block_numbers` to the owning shard. Used to assemble
+    /// the response pages in the same order as the original request.
    block_shards: Vec<ShardIndex>,
+    /// Page responses by shard index. Will be assembled into a single response.
+    responses: HashMap<ShardIndex, Vec<Bytes>>,
 }

 impl GetPageSplitter {
    /// Checks if the given request only touches a single shard, and returns the shard ID. This is
    /// the common case, so we check first in order to avoid unnecessary allocations and overhead.
-    pub fn for_single_shard(
+    /// The caller must ensure that the request has at least one block number, or this will panic.
+    pub fn is_single_shard(
        req: &page_api::GetPageRequest,
        count: ShardCount,
        stripe_size: ShardStripeSize,
@@ -32,12 +35,8 @@ impl GetPageSplitter {
            return Some(ShardIndex::unsharded());
        }

-        // Find the first page's shard, for comparison. If there are no pages, just return the first
-        // shard (caller likely checked already, otherwise the server will reject it).
-        let Some(&first_page) = req.block_numbers.first() else {
-            return Some(ShardIndex::new(ShardNumber(0), count));
-        };
-        let key = rel_block_to_key(req.rel, first_page);
+        // Find the base shard index for the first page, and compare with the rest.
+        let key = rel_block_to_key(req.rel, *req.block_numbers.first().expect("no pages"));
        let shard_number = key_to_shard_number(count, stripe_size, &key);

        req.block_numbers
@@ -58,19 +57,19 @@ impl GetPageSplitter {
    ) -> Self {
        // The caller should make sure we don't split requests unnecessarily.
        debug_assert!(
-            Self::for_single_shard(&req, count, stripe_size).is_none(),
+            Self::is_single_shard(&req, count, stripe_size).is_none(),
            "unnecessary request split"
        );

        // Split the requests by shard index.
        let mut requests = HashMap::with_capacity(2); // common case
        let mut block_shards = Vec::with_capacity(req.block_numbers.len());
-        for &blkno in &req.block_numbers {
+        for blkno in req.block_numbers {
            let key = rel_block_to_key(req.rel, blkno);
            let shard_number = key_to_shard_number(count, stripe_size, &key);
            let shard_id = ShardIndex::new(shard_number, count);

-            requests
+            let shard_req = requests
                .entry(shard_id)
                .or_insert_with(|| page_api::GetPageRequest {
                    request_id: req.request_id,
@@ -78,47 +77,27 @@ impl GetPageSplitter {
                    rel: req.rel,
                    read_lsn: req.read_lsn,
                    block_numbers: Vec::new(),
-                })
-                .block_numbers
-                .push(blkno);
+                });
+            shard_req.block_numbers.push(blkno);
            block_shards.push(shard_id);
        }

-        // Construct a response to be populated by shard responses. Preallocate empty page slots
-        // with the expected block numbers.
-        let response = page_api::GetPageResponse {
-            request_id: req.request_id,
-            status_code: page_api::GetPageStatusCode::Ok,
-            reason: None,
-            rel: req.rel,
-            pages: req
-                .block_numbers
-                .into_iter()
-                .map(|block_number| {
-                    page_api::Page {
-                        block_number,
-                        image: Bytes::new(), // empty page slot to be filled in
-                    }
-                })
-                .collect(),
-        };
-
        Self {
+            request_id: req.request_id,
+            responses: HashMap::with_capacity(requests.len()),
            requests,
-            response,
            block_shards,
        }
    }

-    /// Drains the per-shard requests, moving them out of the splitter to avoid extra allocations.
+    /// Drains the per-shard requests, moving them out of the hashmap to avoid extra allocations.
    pub fn drain_requests(
        &mut self,
    ) -> impl Iterator<Item = (ShardIndex, page_api::GetPageRequest)> {
        self.requests.drain()
    }

-    /// Adds a response from the given shard. The response must match the request ID and have an OK
-    /// status code. A response must not already exist for the given shard ID.
+    /// Adds a response from the given shard.
    #[allow(clippy::result_large_err)]
    pub fn add_response(
        &mut self,
@@ -126,84 +105,68 @@ impl GetPageSplitter {
        response: page_api::GetPageResponse,
    ) -> tonic::Result<()> {
        // The caller should already have converted status codes into tonic::Status.
-        if response.status_code != page_api::GetPageStatusCode::Ok {
+        assert_eq!(response.status_code, page_api::GetPageStatusCode::Ok);
+
+        // Make sure the response matches the request ID.
+        if response.request_id != self.request_id {
            return Err(tonic::Status::internal(format!(
-                "unexpected non-OK response for shard {shard_id}: {} {}",
-                response.status_code,
-                response.reason.unwrap_or_default()
+                "response ID {} does not match request ID {}",
+                response.request_id, self.request_id
            )));
        }

-        if response.request_id != self.response.request_id {
+        // Add the response data to the map.
+        let old = self.responses.insert(shard_id, response.page_images);
+
+        if old.is_some() {
            return Err(tonic::Status::internal(format!(
-                "response ID mismatch for shard {shard_id}: expected {}, got {}",
-                self.response.request_id, response.request_id
-            )));
-        }
-
-        // Place the shard response pages into the assembled response, in request order.
-        let mut pages = response.pages.into_iter();
-
-        for (i, &s) in self.block_shards.iter().enumerate() {
-            if shard_id != s {
-                continue;
-            }
-
-            let Some(slot) = self.response.pages.get_mut(i) else {
-                return Err(tonic::Status::internal(format!(
-                    "no block_shards slot {i} for shard {shard_id}"
-                )));
-            };
-            let Some(page) = pages.next() else {
-                return Err(tonic::Status::internal(format!(
-                    "missing page {} in shard {shard_id} response",
-                    slot.block_number
-                )));
-            };
-            if page.block_number != slot.block_number {
-                return Err(tonic::Status::internal(format!(
-                    "shard {shard_id} returned wrong page at index {i}, expected {} got {}",
-                    slot.block_number, page.block_number
-                )));
-            }
-            if !slot.image.is_empty() {
-                return Err(tonic::Status::internal(format!(
-                    "shard {shard_id} returned duplicate page {} at index {i}",
-                    slot.block_number
-                )));
-            }
-
-            *slot = page;
-        }
-
-        // Make sure we've consumed all pages from the shard response.
-        if let Some(extra_page) = pages.next() {
-            return Err(tonic::Status::internal(format!(
-                "shard {shard_id} returned extra page: {}",
-                extra_page.block_number
+                "duplicate response for shard {shard_id}",
            )));
        }

        Ok(())
    }

-    /// Fetches the final, assembled response.
+    /// Assembles the shard responses into a single response. Responses must be present for all
+    /// relevant shards, and the total number of pages must match the original request.
    #[allow(clippy::result_large_err)]
-    pub fn get_response(self) -> tonic::Result<page_api::GetPageResponse> {
-        // Check that the response is complete.
-        for (i, page) in self.response.pages.iter().enumerate() {
-            if page.image.is_empty() {
+    pub fn assemble_response(self) -> tonic::Result<page_api::GetPageResponse> {
+        let mut response = page_api::GetPageResponse {
+            request_id: self.request_id,
+            status_code: page_api::GetPageStatusCode::Ok,
+            reason: None,
+            page_images: Vec::with_capacity(self.block_shards.len()),
+        };
+
+        // Set up per-shard page iterators we can pull from.
+        let mut shard_responses = HashMap::with_capacity(self.responses.len());
+        for (shard_id, responses) in self.responses {
+            shard_responses.insert(shard_id, responses.into_iter());
+        }
+
+        // Reassemble the responses in the same order as the original request.
+        for shard_id in &self.block_shards {
+            let page = shard_responses
+                .get_mut(shard_id)
+                .ok_or_else(|| {
+                    tonic::Status::internal(format!("missing response for shard {shard_id}"))
+                })?
+                .next()
+                .ok_or_else(|| {
+                    tonic::Status::internal(format!("missing page from shard {shard_id}"))
+                })?;
+            response.page_images.push(page);
+        }
+
+        // Make sure there are no additional pages.
+        for (shard_id, mut pages) in shard_responses {
+            if pages.next().is_some() {
                return Err(tonic::Status::internal(format!(
-                    "missing page {} for shard {}",
-                    page.block_number,
-                    self.block_shards
-                        .get(i)
-                        .map(|s| s.to_string())
-                        .unwrap_or_else(|| "?".to_string())
+                    "extra pages returned from shard {shard_id}"
                )));
            }
        }

-        Ok(self.response)
+        Ok(response)
    }
 }
--- a/pageserver/ctl/Cargo.toml
+++ b/pageserver/ctl/Cargo.toml
@@ -17,7 +17,6 @@ pageserver = { path = ".." }
 pageserver_api.workspace = true
 remote_storage = { path = "../../libs/remote_storage" }
 postgres_ffi.workspace = true
-serde.workspace = true
 thiserror.workspace = true
 tokio.workspace = true
 tokio-util.workspace = true
--- a/pageserver/ctl/src/download_remote_object.rs
+++ b/pageserver/ctl/src/download_remote_object.rs
@@ -1,85 +0,0 @@
-use camino::Utf8PathBuf;
-use clap::Parser;
-use tokio_util::sync::CancellationToken;
-
-/// Download a specific object from remote storage to a local file.
-///
-/// The remote storage configuration is supplied via the `REMOTE_STORAGE_CONFIG` environment
-/// variable, in the same TOML format that the pageserver itself understands. This allows the
-/// command to work with any cloud supported by the `remote_storage` crate (currently AWS S3,
-/// Azure Blob Storage and local files), as long as the credentials are available via the
-/// standard environment variables expected by the underlying SDKs.
-///
-/// Examples for setting the environment variable:
-///
-/// ```bash
-/// # AWS S3 (region can also be provided via AWS_REGION)
-/// export REMOTE_STORAGE_CONFIG='remote_storage = { bucket_name = "my-bucket", bucket_region = "us-east-2" }'
-///
-/// # Azure Blob Storage (account key picked up from AZURE_STORAGE_ACCOUNT_KEY)
-/// export REMOTE_STORAGE_CONFIG='remote_storage = { container = "my-container", account = "my-account" }'
-/// ```
-#[derive(Parser)]
-pub(crate) struct DownloadRemoteObjectCmd {
-    /// Key / path of the object to download (relative to the remote storage prefix).
-    ///
-    /// Examples:
-    ///   "wal/3aa8f.../00000001000000000000000A"
-    ///   "pageserver/v1/tenants/<tenant_id>/timelines/<timeline_id>/layer_12345"
-    pub remote_path: String,
-
-    /// Path of the local file to create. Existing file will be overwritten.
-    ///
-    /// Examples:
-    ///   "./segment"
-    ///   "/tmp/layer_12345.parquet"
-    pub output_file: Utf8PathBuf,
-}
-
-pub(crate) async fn main(cmd: &DownloadRemoteObjectCmd) -> anyhow::Result<()> {
-    use remote_storage::{DownloadOpts, GenericRemoteStorage, RemotePath, RemoteStorageConfig};
-
-    // Fetch remote storage configuration from the environment
-    let config_str = std::env::var("REMOTE_STORAGE_CONFIG").map_err(|_| {
-        anyhow::anyhow!(
-            "'REMOTE_STORAGE_CONFIG' environment variable must be set to a valid remote storage TOML config"
-        )
-    })?;
-
-    let config = RemoteStorageConfig::from_toml_str(&config_str)?;
-
-    // Initialise remote storage client
-    let storage = GenericRemoteStorage::from_config(&config).await?;
-
-    // RemotePath must be relative – leading slashes confuse the parser.
-    let remote_path_str = cmd.remote_path.trim_start_matches('/');
-    let remote_path = RemotePath::from_string(remote_path_str)?;
-
-    let cancel = CancellationToken::new();
-
-    println!(
-        "Downloading '{remote_path}' from remote storage bucket {:?} ...",
-        config.storage.bucket_name()
-    );
-
-    // Start the actual download
-    let download = storage
-        .download(&remote_path, &DownloadOpts::default(), &cancel)
-        .await?;
-
-    // Stream to file
-    let mut reader = tokio_util::io::StreamReader::new(download.download_stream);
-    let tmp_path = cmd.output_file.with_extension("tmp");
-    let mut file = tokio::fs::File::create(&tmp_path).await?;
-    tokio::io::copy(&mut reader, &mut file).await?;
-    file.sync_all().await?;
-    // Atomically move into place
-    tokio::fs::rename(&tmp_path, &cmd.output_file).await?;
-
-    println!(
-        "Downloaded to '{}'. Last modified: {:?}, etag: {}",
-        cmd.output_file, download.last_modified, download.etag
-    );
-
-    Ok(())
-}
--- a/pageserver/ctl/src/index_part.rs
+++ b/pageserver/ctl/src/index_part.rs
@@ -1,16 +1,14 @@
 use std::str::FromStr;

-use anyhow::{Context, Ok};
+use anyhow::Context;
 use camino::Utf8PathBuf;
 use pageserver::tenant::{
    IndexPart,
    layer_map::{LayerMap, SearchResult},
-    remote_timeline_client::{index::LayerFileMetadata, remote_layer_path},
-    storage_layer::{LayerName, LayerVisibilityHint, PersistentLayerDesc, ReadableLayerWeak},
+    remote_timeline_client::remote_layer_path,
+    storage_layer::{PersistentLayerDesc, ReadableLayerWeak},
 };
 use pageserver_api::key::Key;
-use serde::Serialize;
-use std::collections::BTreeMap;
 use utils::{
    id::{TenantId, TimelineId},
    lsn::Lsn,
@@ -35,31 +33,6 @@ pub(crate) enum IndexPartCmd {
        #[arg(long)]
        lsn: String,
    },
-    /// List all visible delta and image layers at the latest LSN.
-    ListVisibleLayers {
-        #[arg(long)]
-        path: Utf8PathBuf,
-    },
-}
-
-fn create_layer_map_from_index_part(
-    index_part: &IndexPart,
-    tenant_shard_id: TenantShardId,
-    timeline_id: TimelineId,
-) -> LayerMap {
-    let mut layer_map = LayerMap::default();
-    {
-        let mut updates = layer_map.batch_update();
-        for (key, value) in index_part.layer_metadata.iter() {
-            updates.insert_historic(PersistentLayerDesc::from_filename(
-                tenant_shard_id,
-                timeline_id,
-                key.clone(),
-                value.file_size,
-            ));
-        }
-    }
-    layer_map
 }

 async fn search_layers(
@@ -76,7 +49,18 @@ async fn search_layers(
        let bytes = tokio::fs::read(path).await?;
        IndexPart::from_json_bytes(&bytes).unwrap()
    };
-    let layer_map = create_layer_map_from_index_part(&index_json, tenant_shard_id, timeline_id);
+    let mut layer_map = LayerMap::default();
+    {
+        let mut updates = layer_map.batch_update();
+        for (key, value) in index_json.layer_metadata.iter() {
+            updates.insert_historic(PersistentLayerDesc::from_filename(
+                tenant_shard_id,
+                timeline_id,
+                key.clone(),
+                value.file_size,
+            ));
+        }
+    }
    let key = Key::from_hex(key)?;

    let lsn = Lsn::from_str(lsn).unwrap();
@@ -114,69 +98,6 @@ async fn search_layers(
    Ok(())
 }

-#[derive(Debug, Clone, Serialize)]
-struct VisibleLayers {
-    pub total_images: u64,
-    pub total_image_bytes: u64,
-    pub total_deltas: u64,
-    pub total_delta_bytes: u64,
-    pub layer_metadata: BTreeMap<LayerName, LayerFileMetadata>,
-}
-
-impl VisibleLayers {
-    pub fn new() -> Self {
-        Self {
-            layer_metadata: BTreeMap::new(),
-            total_images: 0,
-            total_image_bytes: 0,
-            total_deltas: 0,
-            total_delta_bytes: 0,
-        }
-    }
-
-    pub fn add_layer(&mut self, name: LayerName, layer: LayerFileMetadata) {
-        match name {
-            LayerName::Image(_) => {
-                self.total_images += 1;
-                self.total_image_bytes += layer.file_size;
-            }
-            LayerName::Delta(_) => {
-                self.total_deltas += 1;
-                self.total_delta_bytes += layer.file_size;
-            }
-        }
-        self.layer_metadata.insert(name, layer);
-    }
-}
-
-async fn list_visible_layers(path: &Utf8PathBuf) -> anyhow::Result<()> {
-    let tenant_id = TenantId::generate();
-    let tenant_shard_id = TenantShardId::unsharded(tenant_id);
-    let timeline_id = TimelineId::generate();
-
-    let bytes = tokio::fs::read(path).await.context("read file")?;
-    let index_part = IndexPart::from_json_bytes(&bytes).context("deserialize")?;
-    let layer_map = create_layer_map_from_index_part(&index_part, tenant_shard_id, timeline_id);
-    let mut visible_layers = VisibleLayers::new();
-    let (layers, _key_space) = layer_map.get_visibility(Vec::new());
-    for (layer, visibility) in layers {
-        if visibility == LayerVisibilityHint::Visible {
-            visible_layers.add_layer(
-                layer.layer_name(),
-                index_part
-                    .layer_metadata
-                    .get(&layer.layer_name())
-                    .unwrap()
-                    .clone(),
-            );
-        }
-    }
-    let output = serde_json::to_string_pretty(&visible_layers).context("serialize output")?;
-    println!("{output}");
-
-    Ok(())
-}
-
 pub(crate) async fn main(cmd: &IndexPartCmd) -> anyhow::Result<()> {
    match cmd {
        IndexPartCmd::Dump { path } => {
@@ -193,6 +114,5 @@ pub(crate) async fn main(cmd: &IndexPartCmd) -> anyhow::Result<()> {
            key,
            lsn,
        } => search_layers(tenant_id, timeline_id, path, key, lsn).await,
-        IndexPartCmd::ListVisibleLayers { path } => list_visible_layers(path).await,
    }
 }
--- a/pageserver/ctl/src/main.rs
+++ b/pageserver/ctl/src/main.rs
@@ -4,7 +4,6 @@
 //!
 //! Separate, `metadata` subcommand allows to print and update pageserver's metadata file.

-mod download_remote_object;
 mod draw_timeline_dir;
 mod index_part;
 mod key;
@@ -17,7 +16,6 @@ use std::time::{Duration, SystemTime};

 use camino::{Utf8Path, Utf8PathBuf};
 use clap::{Parser, Subcommand};
-use download_remote_object::DownloadRemoteObjectCmd;
 use index_part::IndexPartCmd;
 use layers::LayerCmd;
 use page_trace::PageTraceCmd;
@@ -65,7 +63,6 @@ enum Commands {
    /// Debug print a hex key found from logs
    Key(key::DescribeKeyCommand),
    PageTrace(PageTraceCmd),
-    DownloadRemoteObject(DownloadRemoteObjectCmd),
 }

 /// Read and update pageserver metadata file
@@ -188,9 +185,6 @@ async fn main() -> anyhow::Result<()> {
        }
        Commands::Key(dkc) => dkc.execute(),
        Commands::PageTrace(cmd) => page_trace::main(&cmd)?,
-        Commands::DownloadRemoteObject(cmd) => {
-            download_remote_object::main(&cmd).await?;
-        }
    };
    Ok(())
 }
--- a/pageserver/page_api/proto/page_service.proto
+++ b/pageserver/page_api/proto/page_service.proto
@@ -153,7 +153,7 @@ message GetDbSizeResponse {
 message GetPageRequest {
  // A request ID. Will be included in the response. Should be unique for
  // in-flight requests on the stream.
-  RequestID request_id = 1;
+  uint64 request_id = 1;
  // The request class.
  GetPageClass request_class = 2;
  // The LSN to read at.
@@ -177,14 +177,6 @@ message GetPageRequest {
  repeated uint32 block_number = 5;
 }

-// A Request ID. Should be unique for in-flight requests on a stream. Included in the response.
-message RequestID {
-  // The base request ID.
-  uint64 id = 1;
-  // The request attempt. Starts at 0, incremented on each retry.
-  uint32 attempt = 2;
-}
-
 // A GetPageRequest class. Primarily intended for observability, but may also be
 // used for prioritization in the future.
 enum GetPageClass {
@@ -207,26 +199,13 @@ enum GetPageClass {
 // the entire batch is ready, so no one can make use of the individual pages.
 message GetPageResponse {
  // The original request's ID.
-  RequestID request_id = 1;
-  // The response status code. If not OK, the rel and page fields will be empty.
+  uint64 request_id = 1;
+  // The response status code.
  GetPageStatusCode status_code = 2;
  // A string describing the status, if any.
  string reason = 3;
-  // The relation that the pages belong to.
-  RelTag rel = 4;
-  // The page(s), in the same order as the request.
-  repeated Page page = 5;
-}
-
-// A page.
-//
-// TODO: it would be slightly more efficient (but less convenient) to have separate arrays of block
-// numbers and images, but given the 8KB page size it's probably negligible. Benchmark it anyway.
-message Page {
-  // The page number.
-  uint32 block_number = 1;
-  // The materialized page image, as an 8KB byte vector.
-  bytes image = 2;
+  // The 8KB page images, in the same order as the request. Empty if status_code != OK.
+  repeated bytes page_image = 4;
 }

 // A GetPageResponse status code.
--- a/pageserver/page_api/src/client.rs
+++ b/pageserver/page_api/src/client.rs
@@ -1,5 +1,4 @@
 use anyhow::Context as _;
-use futures::future::ready;
 use futures::{Stream, StreamExt as _, TryStreamExt as _};
 use tokio::io::AsyncRead;
 use tokio_util::io::StreamReader;
@@ -111,7 +110,7 @@ impl Client {
    ) -> tonic::Result<impl Stream<Item = tonic::Result<GetPageResponse>> + Send + 'static> {
        let reqs = reqs.map(proto::GetPageRequest::from);
        let resps = self.inner.get_pages(reqs).await?.into_inner();
-        Ok(resps.and_then(|resp| ready(GetPageResponse::try_from(resp).map_err(|err| err.into()))))
+        Ok(resps.map_ok(GetPageResponse::from))
    }

    /// Returns the size of a relation, as # of blocks.
--- a/pageserver/page_api/src/model.rs
+++ b/pageserver/page_api/src/model.rs
@@ -356,10 +356,7 @@ impl TryFrom<proto::GetPageRequest> for GetPageRequest {
            return Err(ProtocolError::Missing("block_number"));
        }
        Ok(Self {
-            request_id: pb
-                .request_id
-                .ok_or(ProtocolError::Missing("request_id"))?
-                .into(),
+            request_id: pb.request_id,
            request_class: pb.request_class.into(),
            read_lsn: pb
                .read_lsn
@@ -374,7 +371,7 @@ impl TryFrom<proto::GetPageRequest> for GetPageRequest {
 impl From<GetPageRequest> for proto::GetPageRequest {
    fn from(request: GetPageRequest) -> Self {
        Self {
-            request_id: Some(request.request_id.into()),
+            request_id: request.request_id,
            request_class: request.request_class.into(),
            read_lsn: Some(request.read_lsn.into()),
            rel: Some(request.rel.into()),
@@ -383,54 +380,11 @@ impl From<GetPageRequest> for proto::GetPageRequest {
    }
 }

-/// A GetPage request ID and retry attempt. Should be unique for in-flight requests on a stream.
-#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash, PartialOrd, Ord)]
-pub struct RequestID {
-    /// The base request ID.
-    pub id: u64,
-    // The request attempt. Starts at 0, incremented on each retry.
-    pub attempt: u32,
-}
-
-impl RequestID {
-    /// Creates a new RequestID with the given ID and an initial attempt of 0.
-    pub fn new(id: u64) -> Self {
-        Self { id, attempt: 0 }
-    }
-}
-
-impl Display for RequestID {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}.{}", self.id, self.attempt)
-    }
-}
-
-impl From<proto::RequestId> for RequestID {
-    fn from(pb: proto::RequestId) -> Self {
-        Self {
-            id: pb.id,
-            attempt: pb.attempt,
-        }
-    }
-}
-
-impl From<u64> for RequestID {
-    fn from(id: u64) -> Self {
-        Self::new(id)
-    }
-}
-
-impl From<RequestID> for proto::RequestId {
-    fn from(request_id: RequestID) -> Self {
-        Self {
-            id: request_id.id,
-            attempt: request_id.attempt,
-        }
-    }
-}
+/// A GetPage request ID.
+pub type RequestID = u64;

 /// A GetPage request class.
-#[derive(Clone, Copy, Debug, strum_macros::Display)]
+#[derive(Clone, Copy, Debug)]
 pub enum GetPageClass {
    /// Unknown class. For backwards compatibility: used when an older client version sends a class
    /// that a newer server version has removed.
@@ -443,19 +397,6 @@ pub enum GetPageClass {
    Background,
 }

-impl GetPageClass {
-    /// Returns true if this is considered a bulk request (i.e. more throughput-oriented rather than
-    /// latency-sensitive).
-    pub fn is_bulk(&self) -> bool {
-        match self {
-            Self::Unknown => false,
-            Self::Normal => false,
-            Self::Prefetch => true,
-            Self::Background => true,
-        }
-    }
-}
-
 impl From<proto::GetPageClass> for GetPageClass {
    fn from(pb: proto::GetPageClass) -> Self {
        match pb {
@@ -502,41 +443,32 @@ impl From<GetPageClass> for i32 {
 pub struct GetPageResponse {
    /// The original request's ID.
    pub request_id: RequestID,
-    /// The response status code. If not OK, the `rel` and `pages` fields will be empty.
+    /// The response status code.
    pub status_code: GetPageStatusCode,
    /// A string describing the status, if any.
    pub reason: Option<String>,
-    /// The relation that the pages belong to.
-    pub rel: RelTag,
-    // The page(s), in the same order as the request.
-    pub pages: Vec<Page>,
+    /// The 8KB page images, in the same order as the request. Empty if status != OK.
+    pub page_images: Vec<Bytes>,
 }

-impl TryFrom<proto::GetPageResponse> for GetPageResponse {
-    type Error = ProtocolError;
-
-    fn try_from(pb: proto::GetPageResponse) -> Result<Self, ProtocolError> {
-        Ok(Self {
-            request_id: pb
-                .request_id
-                .ok_or(ProtocolError::Missing("request_id"))?
-                .into(),
+impl From<proto::GetPageResponse> for GetPageResponse {
+    fn from(pb: proto::GetPageResponse) -> Self {
+        Self {
+            request_id: pb.request_id,
            status_code: pb.status_code.into(),
            reason: Some(pb.reason).filter(|r| !r.is_empty()),
-            rel: pb.rel.ok_or(ProtocolError::Missing("rel"))?.try_into()?,
-            pages: pb.page.into_iter().map(Page::from).collect(),
-        })
+            page_images: pb.page_image,
+        }
    }
 }

 impl From<GetPageResponse> for proto::GetPageResponse {
    fn from(response: GetPageResponse) -> Self {
        Self {
-            request_id: Some(response.request_id.into()),
+            request_id: response.request_id,
            status_code: response.status_code.into(),
            reason: response.reason.unwrap_or_default(),
-            rel: Some(response.rel.into()),
-            page: response.pages.into_iter().map(proto::Page::from).collect(),
+            page_image: response.page_images,
        }
    }
 }
@@ -569,39 +501,11 @@ impl GetPageResponse {
            request_id,
            status_code,
            reason: Some(status.message().to_string()),
-            rel: RelTag::default(),
-            pages: Vec::new(),
+            page_images: Vec::new(),
        })
    }
 }

-// A page.
-#[derive(Clone, Debug)]
-pub struct Page {
-    /// The page number.
-    pub block_number: u32,
-    /// The materialized page image, as an 8KB byte vector.
-    pub image: Bytes,
-}
-
-impl From<proto::Page> for Page {
-    fn from(pb: proto::Page) -> Self {
-        Self {
-            block_number: pb.block_number,
-            image: pb.image,
-        }
-    }
-}
-
-impl From<Page> for proto::Page {
-    fn from(page: Page) -> Self {
-        Self {
-            block_number: page.block_number,
-            image: page.image,
-        }
-    }
-}
-
 /// A GetPage response status code.
 ///
 /// These are effectively equivalent to gRPC statuses. However, we use a bidirectional stream
--- a/pageserver/pagebench/Cargo.toml
+++ b/pageserver/pagebench/Cargo.toml
@@ -27,9 +27,8 @@ tokio-util.workspace = true
 tonic.workspace = true
 url.workspace = true

-pageserver_api.workspace = true
 pageserver_client.workspace = true
-pageserver_client_grpc.workspace = true
+pageserver_api.workspace = true
 pageserver_page_api.workspace = true
 utils = { path = "../../libs/utils/" }
 workspace_hack = { version = "0.1", path = "../../workspace_hack" }
--- a/pageserver/pagebench/src/cmd/getpage_latest_lsn.rs
+++ b/pageserver/pagebench/src/cmd/getpage_latest_lsn.rs
@@ -10,14 +10,12 @@ use anyhow::Context;
 use async_trait::async_trait;
 use bytes::Bytes;
 use camino::Utf8PathBuf;
-use futures::stream::FuturesUnordered;
 use futures::{Stream, StreamExt as _};
 use pageserver_api::key::Key;
 use pageserver_api::keyspace::KeySpaceAccum;
 use pageserver_api::pagestream_api::{PagestreamGetPageRequest, PagestreamRequest};
 use pageserver_api::reltag::RelTag;
 use pageserver_api::shard::TenantShardId;
-use pageserver_client_grpc::{self as client_grpc, ShardSpec};
 use pageserver_page_api as page_api;
 use rand::prelude::*;
 use tokio::task::JoinSet;
@@ -39,10 +37,6 @@ pub(crate) struct Args {
    /// Pageserver connection string. Supports postgresql:// and grpc:// protocols.
    #[clap(long, default_value = "postgres://postgres@localhost:64000")]
    page_service_connstring: String,
-    /// Use the rich gRPC Pageserver client `client_grpc::PageserverClient`, rather than the basic
-    /// no-frills `page_api::Client`. Only valid with grpc:// connstrings.
-    #[clap(long)]
-    rich_client: bool,
    #[clap(long)]
    pageserver_jwt: Option<String>,
    #[clap(long, default_value = "1")]
@@ -338,7 +332,6 @@ async fn main_impl(
            let client: Box<dyn Client> = match scheme.as_str() {
                "postgresql" | "postgres" => {
                    assert!(!args.compression, "libpq does not support compression");
-                    assert!(!args.rich_client, "rich client requires grpc://");
                    Box::new(
                        LibpqClient::new(&args.page_service_connstring, worker_id.timeline)
                            .await
@@ -346,16 +339,6 @@ async fn main_impl(
                    )
                }

-                "grpc" if args.rich_client => Box::new(
-                    RichGrpcClient::new(
-                        &args.page_service_connstring,
-                        worker_id.timeline,
-                        args.compression,
-                    )
-                    .await
-                    .unwrap(),
-                ),
-
                "grpc" => Box::new(
                    GrpcClient::new(
                        &args.page_service_connstring,
@@ -674,7 +657,7 @@ impl Client for GrpcClient {
        blks: Vec<u32>,
    ) -> anyhow::Result<()> {
        let req = page_api::GetPageRequest {
-            request_id: req_id.into(),
+            request_id: req_id,
            request_class: page_api::GetPageClass::Normal,
            read_lsn: page_api::ReadLsn {
                request_lsn: req_lsn,
@@ -694,79 +677,6 @@ impl Client for GrpcClient {
            "unexpected status code: {}",
            resp.status_code,
        );
-        Ok((
-            resp.request_id.id,
-            resp.pages.into_iter().map(|p| p.image).collect(),
-        ))
-    }
-}
-
-/// A rich gRPC Pageserver client.
-struct RichGrpcClient {
-    inner: Arc<client_grpc::PageserverClient>,
-    requests: FuturesUnordered<
-        Pin<Box<dyn Future<Output = anyhow::Result<page_api::GetPageResponse>> + Send>>,
-    >,
-}
-
-impl RichGrpcClient {
-    async fn new(
-        connstring: &str,
-        ttid: TenantTimelineId,
-        compression: bool,
-    ) -> anyhow::Result<Self> {
-        let inner = Arc::new(client_grpc::PageserverClient::new(
-            ttid.tenant_id,
-            ttid.timeline_id,
-            ShardSpec::new(
-                [(ShardIndex::unsharded(), connstring.to_string())].into(),
-                None,
-            )?,
-            None,
-            compression.then_some(tonic::codec::CompressionEncoding::Zstd),
-        )?);
-        Ok(Self {
-            inner,
-            requests: FuturesUnordered::new(),
-        })
-    }
-}
-
-#[async_trait]
-impl Client for RichGrpcClient {
-    async fn send_get_page(
-        &mut self,
-        req_id: u64,
-        req_lsn: Lsn,
-        mod_lsn: Lsn,
-        rel: RelTag,
-        blks: Vec<u32>,
-    ) -> anyhow::Result<()> {
-        let req = page_api::GetPageRequest {
-            request_id: req_id.into(),
-            request_class: page_api::GetPageClass::Normal,
-            read_lsn: page_api::ReadLsn {
-                request_lsn: req_lsn,
-                not_modified_since_lsn: Some(mod_lsn),
-            },
-            rel,
-            block_numbers: blks,
-        };
-        let inner = self.inner.clone();
-        self.requests.push(Box::pin(async move {
-            inner
-                .get_page(req)
-                .await
-                .map_err(|err| anyhow::anyhow!("{err}"))
-        }));
-        Ok(())
-    }
-
-    async fn recv_get_page(&mut self) -> anyhow::Result<(u64, Vec<Bytes>)> {
-        let resp = self.requests.next().await.unwrap()?;
-        Ok((
-            resp.request_id.id,
-            resp.pages.into_iter().map(|p| p.image).collect(),
-        ))
+        Ok((resp.request_id, resp.page_images))
    }
 }
--- a/pageserver/src/bin/pageserver.rs
+++ b/pageserver/src/bin/pageserver.rs
@@ -29,8 +29,8 @@ use pageserver::task_mgr::{
 };
 use pageserver::tenant::{TenantSharedResources, mgr, secondary};
 use pageserver::{
-    CancellableTask, ConsumptionMetricsTasks, HttpEndpointListener, HttpsEndpointListener,
-    MetricsCollectionTask, http, page_cache, page_service, task_mgr, virtual_file,
+    CancellableTask, ConsumptionMetricsTasks, HttpEndpointListener, HttpsEndpointListener, http,
+    page_cache, page_service, task_mgr, virtual_file,
 };
 use postgres_backend::AuthType;
 use remote_storage::GenericRemoteStorage;
@@ -41,7 +41,6 @@ use tracing_utils::OtelGuard;
 use utils::auth::{JwtAuth, SwappableJwtAuth};
 use utils::crashsafe::syncfs;
 use utils::logging::TracingErrorLayerEnablement;
-use utils::metrics_collector::{METRICS_COLLECTION_INTERVAL, METRICS_COLLECTOR};
 use utils::sentry_init::init_sentry;
 use utils::{failpoint_support, logging, project_build_tag, project_git_version, tcp_listener};

@@ -764,41 +763,6 @@ fn start_pageserver(
        (http_task, https_task)
    };

-    /* BEGIN_HADRON */
-    let metrics_collection_task = {
-        let cancel = shutdown_pageserver.child_token();
-        let task = crate::BACKGROUND_RUNTIME.spawn({
-            let cancel = cancel.clone();
-            let background_jobs_barrier = background_jobs_barrier.clone();
-            async move {
-                if conf.force_metric_collection_on_scrape {
-                    return;
-                }
-
-                // first wait until background jobs are cleared to launch.
-                tokio::select! {
-                    _ = cancel.cancelled() => { return; },
-                    _ = background_jobs_barrier.wait() => {}
-                };
-                let mut interval = tokio::time::interval(METRICS_COLLECTION_INTERVAL);
-                loop {
-                    tokio::select! {
-                        _ = cancel.cancelled() => {
-                            tracing::info!("cancelled metrics collection task, exiting...");
-                             break;
-                        },
-                        _ = interval.tick() => {}
-                    }
-                    tokio::task::spawn_blocking(|| {
-                        METRICS_COLLECTOR.run_once(true);
-                    });
-                }
-            }
-        });
-        MetricsCollectionTask(CancellableTask { task, cancel })
-    };
-    /* END_HADRON */
-
    let consumption_metrics_tasks = {
        let cancel = shutdown_pageserver.child_token();
        let task = crate::BACKGROUND_RUNTIME.spawn({
@@ -880,7 +844,6 @@ fn start_pageserver(
            https_endpoint_listener,
            page_service,
            page_service_grpc,
-            metrics_collection_task,
            consumption_metrics_tasks,
            disk_usage_eviction_task,
            &tenant_manager,
@@ -926,11 +889,8 @@ async fn create_remote_storage_client(
            "Simulating remote failures for first {} attempts of each op",
            conf.test_remote_failures
        );
-        remote_storage = GenericRemoteStorage::unreliable_wrapper(
-            remote_storage,
-            conf.test_remote_failures,
-            conf.test_remote_failures_probability,
-        );
+        remote_storage =
+            GenericRemoteStorage::unreliable_wrapper(remote_storage, conf.test_remote_failures);
    }

    Ok(remote_storage)
--- a/pageserver/src/config.rs
+++ b/pageserver/src/config.rs
@@ -147,11 +147,7 @@ pub struct PageServerConf {

    pub disk_usage_based_eviction: DiskUsageEvictionTaskConfig,

-    // The number of allowed failures in remote storage operations.
    pub test_remote_failures: u64,
-    // The probability of failure in remote storage operations. Only works when test_remote_failures > 1.
-    // Use 100 for 100% failure, 0 for no failure.
-    pub test_remote_failures_probability: u64,

    pub ondemand_download_behavior_treat_error_as_warn: bool,

@@ -252,14 +248,6 @@ pub struct PageServerConf {
    pub timeline_import_config: pageserver_api::config::TimelineImportConfig,

    pub basebackup_cache_config: Option<pageserver_api::config::BasebackupCacheConfig>,
-
-    /// Defines what is a big tenant for the purpose of image layer generation.
-    /// See Timeline::should_check_if_image_layers_required
-    pub image_layer_generation_large_timeline_threshold: Option<u64>,
-
-    /// Controls whether to collect all metrics on each scrape or to return potentially stale
-    /// results.
-    pub force_metric_collection_on_scrape: bool,
 }

 /// Token for authentication to safekeepers
@@ -404,7 +392,6 @@ impl PageServerConf {
            synthetic_size_calculation_interval,
            disk_usage_based_eviction,
            test_remote_failures,
-            test_remote_failures_probability,
            ondemand_download_behavior_treat_error_as_warn,
            background_task_maximum_delay,
            control_plane_api,
@@ -440,8 +427,6 @@ impl PageServerConf {
            posthog_config,
            timeline_import_config,
            basebackup_cache_config,
-            image_layer_generation_large_timeline_threshold,
-            force_metric_collection_on_scrape,
        } = config_toml;

        let mut conf = PageServerConf {
@@ -476,7 +461,6 @@ impl PageServerConf {
            synthetic_size_calculation_interval,
            disk_usage_based_eviction,
            test_remote_failures,
-            test_remote_failures_probability,
            ondemand_download_behavior_treat_error_as_warn,
            background_task_maximum_delay,
            control_plane_api: control_plane_api
@@ -500,8 +484,6 @@ impl PageServerConf {
            dev_mode,
            timeline_import_config,
            basebackup_cache_config,
-            image_layer_generation_large_timeline_threshold,
-            force_metric_collection_on_scrape,

            // ------------------------------------------------------------
            // fields that require additional validation or custom handling
--- a/pageserver/src/feature_resolver.rs
+++ b/pageserver/src/feature_resolver.rs
@@ -409,12 +409,11 @@ impl TenantFeatureResolver {

    /// Refresh the cached properties and flags on the critical path.
    pub fn refresh_properties_and_flags(&self, tenant_shard: &TenantShard) {
-        let mut remote_size_mb = Some(0.0);
+        let mut remote_size_mb = None;
        for timeline in tenant_shard.list_timelines() {
            let size = timeline.metrics.resident_physical_size_get();
            if size == 0 {
                remote_size_mb = None;
-                break;
            }
            if let Some(ref mut remote_size_mb) = remote_size_mb {
                *remote_size_mb += size as f64 / 1024.0 / 1024.0;
--- a/pageserver/src/http/routes.rs
+++ b/pageserver/src/http/routes.rs
@@ -2,9 +2,7 @@
 //! Management HTTP API
 //!
 use std::cmp::Reverse;
-use std::collections::BTreeMap;
-use std::collections::BinaryHeap;
-use std::collections::HashMap;
+use std::collections::{BinaryHeap, HashMap};
 use std::str::FromStr;
 use std::sync::Arc;
 use std::time::Duration;
@@ -81,8 +79,8 @@ use crate::tenant::storage_layer::{IoConcurrency, LayerAccessStatsReset, LayerNa
 use crate::tenant::timeline::layer_manager::LayerManagerLockHolder;
 use crate::tenant::timeline::offload::{OffloadError, offload_timeline};
 use crate::tenant::timeline::{
-    CompactFlags, CompactOptions, CompactRequest, MarkInvisibleRequest, Timeline, WaitLsnTimeout,
-    WaitLsnWaiter, import_pgdata,
+    CompactFlags, CompactOptions, CompactRequest, CompactionError, MarkInvisibleRequest, Timeline,
+    WaitLsnTimeout, WaitLsnWaiter, import_pgdata,
 };
 use crate::tenant::{
    GetTimelineError, LogicalSizeCalculationCause, OffloadedTimeline, PageReconstructError,
@@ -397,7 +395,6 @@ async fn build_timeline_info(
    timeline: &Arc<Timeline>,
    include_non_incremental_logical_size: bool,
    force_await_initial_logical_size: bool,
-    include_image_consistent_lsn: bool,
    ctx: &RequestContext,
 ) -> anyhow::Result<TimelineInfo> {
    crate::tenant::debug_assert_current_span_has_tenant_and_timeline_id();
@@ -422,10 +419,6 @@ async fn build_timeline_info(
                .await?,
        );
    }
-    // HADRON
-    if include_image_consistent_lsn {
-        info.image_consistent_lsn = Some(timeline.compute_image_consistent_lsn().await?);
-    }
    Ok(info)
 }

@@ -515,8 +508,6 @@ async fn build_timeline_info_common(
        is_invisible: Some(is_invisible),

        walreceiver_status,
-        // HADRON
-        image_consistent_lsn: None,
    };
    Ok(info)
 }
@@ -719,8 +710,6 @@ async fn timeline_list_handler(
        parse_query_param(&request, "include-non-incremental-logical-size")?;
    let force_await_initial_logical_size: Option<bool> =
        parse_query_param(&request, "force-await-initial-logical-size")?;
-    let include_image_consistent_lsn: Option<bool> =
-        parse_query_param(&request, "include-image-consistent-lsn")?;
    check_permission(&request, Some(tenant_shard_id.tenant_id))?;

    let state = get_state(&request);
@@ -741,7 +730,6 @@ async fn timeline_list_handler(
                &timeline,
                include_non_incremental_logical_size.unwrap_or(false),
                force_await_initial_logical_size.unwrap_or(false),
-                include_image_consistent_lsn.unwrap_or(false),
                &ctx,
            )
            .instrument(info_span!("build_timeline_info", timeline_id = %timeline.timeline_id))
@@ -770,9 +758,6 @@ async fn timeline_and_offloaded_list_handler(
        parse_query_param(&request, "include-non-incremental-logical-size")?;
    let force_await_initial_logical_size: Option<bool> =
        parse_query_param(&request, "force-await-initial-logical-size")?;
-    let include_image_consistent_lsn: Option<bool> =
-        parse_query_param(&request, "include-image-consistent-lsn")?;
-
    check_permission(&request, Some(tenant_shard_id.tenant_id))?;

    let state = get_state(&request);
@@ -793,7 +778,6 @@ async fn timeline_and_offloaded_list_handler(
                &timeline,
                include_non_incremental_logical_size.unwrap_or(false),
                force_await_initial_logical_size.unwrap_or(false),
-                include_image_consistent_lsn.unwrap_or(false),
                &ctx,
            )
            .instrument(info_span!("build_timeline_info", timeline_id = %timeline.timeline_id))
@@ -978,9 +962,6 @@ async fn timeline_detail_handler(
        parse_query_param(&request, "include-non-incremental-logical-size")?;
    let force_await_initial_logical_size: Option<bool> =
        parse_query_param(&request, "force-await-initial-logical-size")?;
-    // HADRON
-    let include_image_consistent_lsn: Option<bool> =
-        parse_query_param(&request, "include-image-consistent-lsn")?;
    check_permission(&request, Some(tenant_shard_id.tenant_id))?;

    // Logical size calculation needs downloading.
@@ -1001,7 +982,6 @@ async fn timeline_detail_handler(
            &timeline,
            include_non_incremental_logical_size.unwrap_or(false),
            force_await_initial_logical_size.unwrap_or(false),
-            include_image_consistent_lsn.unwrap_or(false),
            ctx,
        )
        .await
@@ -2520,10 +2500,9 @@ async fn timeline_checkpoint_handler(
                .compact(&cancel, flags, &ctx)
                .await
                .map_err(|e|
-                    if e.is_cancel() {
-                        ApiError::ShuttingDown
-                    } else {
-                        ApiError::InternalServerError(e.into_anyhow())
+                    match e {
+                        CompactionError::ShuttingDown => ApiError::ShuttingDown,
+                        CompactionError::Other(e) => ApiError::InternalServerError(e),
                    }
                )?;
        }
@@ -3234,30 +3213,6 @@ async fn get_utilization(
        .map_err(ApiError::InternalServerError)
 }

-/// HADRON
-async fn list_tenant_visible_size_handler(
-    request: Request<Body>,
-    _cancel: CancellationToken,
-) -> Result<Response<Body>, ApiError> {
-    check_permission(&request, None)?;
-    let state = get_state(&request);
-
-    let mut map = BTreeMap::new();
-    for (tenant_shard_id, slot) in state.tenant_manager.list() {
-        match slot {
-            TenantSlot::Attached(tenant) => {
-                let visible_size = tenant.get_visible_size();
-                map.insert(tenant_shard_id, visible_size);
-            }
-            TenantSlot::Secondary(_) | TenantSlot::InProgress(_) => {
-                continue;
-            }
-        }
-    }
-
-    json_response(StatusCode::OK, map)
-}
-
 async fn list_aux_files(
    mut request: Request<Body>,
    _cancel: CancellationToken,
@@ -3661,7 +3616,6 @@ async fn activate_post_import_handler(
        let timeline_info = build_timeline_info(
            &timeline, false, // include_non_incremental_logical_size,
            false, // force_await_initial_logical_size
-            false, // include_image_consistent_lsn
            &ctx,
        )
        .await
@@ -3737,23 +3691,6 @@ async fn read_tar_eof(mut reader: (impl tokio::io::AsyncRead + Unpin)) -> anyhow
    Ok(())
 }

-async fn force_refresh_feature_flag(
-    request: Request<Body>,
-    _cancel: CancellationToken,
-) -> Result<Response<Body>, ApiError> {
-    let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
-    check_permission(&request, Some(tenant_shard_id.tenant_id))?;
-
-    let state = get_state(&request);
-    let tenant = state
-        .tenant_manager
-        .get_attached_tenant_shard(tenant_shard_id)?;
-    tenant
-        .feature_resolver
-        .refresh_properties_and_flags(&tenant);
-    json_response(StatusCode::OK, ())
-}
-
 async fn tenant_evaluate_feature_flag(
    request: Request<Body>,
    _cancel: CancellationToken,
@@ -3983,14 +3920,9 @@ pub fn make_router(
        .expect("construct launch timestamp header middleware"),
    );

-    let force_metric_collection_on_scrape = state.conf.force_metric_collection_on_scrape;
-
-    let prometheus_metrics_handler_wrapper =
-        move |req| prometheus_metrics_handler(req, force_metric_collection_on_scrape);
-
    Ok(router
        .data(state)
-        .get("/metrics", move |r| request_span(r, prometheus_metrics_handler_wrapper))
+        .get("/metrics", |r| request_span(r, prometheus_metrics_handler))
        .get("/profile/cpu", |r| request_span(r, profile_cpu_handler))
        .get("/profile/heap", |r| request_span(r, profile_heap_handler))
        .get("/v1/status", |r| api_handler(r, status_handler))
@@ -4196,7 +4128,6 @@ pub fn make_router(
        .put("/v1/io_engine", |r| api_handler(r, put_io_engine_handler))
        .put("/v1/io_mode", |r| api_handler(r, put_io_mode_handler))
        .get("/v1/utilization", |r| api_handler(r, get_utilization))
-        .get("/v1/list_tenant_visible_size", |r| api_handler(r, list_tenant_visible_size_handler))
        .post(
            "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/ingest_aux_files",
            |r| testing_api_handler("ingest_aux_files", r, ingest_aux_files),
@@ -4225,9 +4156,6 @@ pub fn make_router(
        .get("/v1/tenant/:tenant_shard_id/feature_flag/:flag_key", |r| {
            api_handler(r, tenant_evaluate_feature_flag)
        })
-        .post("/v1/tenant/:tenant_shard_id/force_refresh_feature_flag", |r| {
-            api_handler(r, force_refresh_feature_flag)
-        })
        .put("/v1/feature_flag/:flag_key", |r| {
            testing_api_handler("force override feature flag - put", r, force_override_feature_flag_for_testing_put)
        })
--- a/pageserver/src/lib.rs
+++ b/pageserver/src/lib.rs
@@ -73,9 +73,6 @@ pub struct HttpEndpointListener(pub CancellableTask);
 pub struct HttpsEndpointListener(pub CancellableTask);
 pub struct ConsumptionMetricsTasks(pub CancellableTask);
 pub struct DiskUsageEvictionTask(pub CancellableTask);
-// HADRON
-pub struct MetricsCollectionTask(pub CancellableTask);
-
 impl CancellableTask {
    pub async fn shutdown(self) {
        self.cancel.cancel();
@@ -90,7 +87,6 @@ pub async fn shutdown_pageserver(
    https_listener: Option<HttpsEndpointListener>,
    page_service: page_service::Listener,
    grpc_task: Option<CancellableTask>,
-    metrics_collection_task: MetricsCollectionTask,
    consumption_metrics_worker: ConsumptionMetricsTasks,
    disk_usage_eviction_task: Option<DiskUsageEvictionTask>,
    tenant_manager: &TenantManager,
@@ -215,14 +211,6 @@ pub async fn shutdown_pageserver(
    // Best effort to persist any outstanding deletions, to avoid leaking objects
    deletion_queue.shutdown(Duration::from_secs(5)).await;

-    // HADRON
-    timed(
-        metrics_collection_task.0.shutdown(),
-        "shutdown metrics collections metrics",
-        Duration::from_secs(1),
-    )
-    .await;
-
    timed(
        consumption_metrics_worker.0.shutdown(),
        "shutdown consumption metrics",
--- a/pageserver/src/metrics.rs
+++ b/pageserver/src/metrics.rs
@@ -2847,24 +2847,6 @@ pub(crate) static MISROUTED_PAGESTREAM_REQUESTS: Lazy<IntCounter> = Lazy::new(||
    .expect("failed to define a metric")
 });

-// Global counter for PageStream request results by outcome. Outcomes are divided into 3 categories:
-// - success
-// - internal_error: errors that indicate bugs in the storage cluster (e.g. page reconstruction errors, misrouted requests, LSN timeout errors)
-// - other_error: transient error conditions that are expected in normal operation or indicate bugs with other parts of the system (e.g. error due to pageserver shutdown, malformed requests etc.)
-pub(crate) static PAGESTREAM_HANDLER_RESULTS_TOTAL: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "pageserver_pagestream_handler_results_total",
-        "Number of pageserver pagestream handler results by outcome (success, internal_error, other_error)",
-        &["outcome"]
-    )
-    .expect("failed to define a metric")
-});
-
-// Constants for pageserver_pagestream_handler_results_total's outcome labels
-pub(crate) const PAGESTREAM_HANDLER_OUTCOME_SUCCESS: &str = "success";
-pub(crate) const PAGESTREAM_HANDLER_OUTCOME_INTERNAL_ERROR: &str = "internal_error";
-pub(crate) const PAGESTREAM_HANDLER_OUTCOME_OTHER_ERROR: &str = "other_error";
-
 // Metrics collected on WAL redo operations
 //
 // We collect the time spent in actual WAL redo ('redo'), and time waiting
--- a/pageserver/src/page_service.rs
+++ b/pageserver/src/page_service.rs
@@ -70,7 +70,7 @@ use crate::context::{
 };
 use crate::metrics::{
    self, COMPUTE_COMMANDS_COUNTERS, ComputeCommandKind, GetPageBatchBreakReason, LIVE_CONNECTIONS,
-    MISROUTED_PAGESTREAM_REQUESTS, PAGESTREAM_HANDLER_RESULTS_TOTAL, SmgrOpTimer, TimelineMetrics,
+    MISROUTED_PAGESTREAM_REQUESTS, SmgrOpTimer, TimelineMetrics,
 };
 use crate::pgdatadir_mapping::{LsnRange, Version};
 use crate::span::{
@@ -1441,57 +1441,20 @@ impl PageServerHandler {
            let (response_msg, ctx) = match handler_result {
                Err(e) => match &e.err {
                    PageStreamError::Shutdown => {
-                        // BEGIN HADRON
-                        PAGESTREAM_HANDLER_RESULTS_TOTAL
-                            .with_label_values(&[metrics::PAGESTREAM_HANDLER_OUTCOME_OTHER_ERROR])
-                            .inc();
-                        // END HADRON
-
                        // If we fail to fulfil a request during shutdown, which may be _because_ of
                        // shutdown, then do not send the error to the client.  Instead just drop the
                        // connection.
                        span.in_scope(|| info!("dropping connection due to shutdown"));
                        return Err(QueryError::Shutdown);
                    }
-                    PageStreamError::Reconnect(_reason) => {
-                        span.in_scope(|| {
-                            // BEGIN HADRON
-                            // We can get here because the compute node is pointing at the wrong PS. We
-                            // already have a metric to keep track of this so suppressing this log to
-                            // reduce log spam. The information in this log message is not going to be that
-                            // helpful given the volume of logs that can be generated.
-                            // info!("handler requested reconnect: {reason}")
-                            // END HADRON
-                        });
-                        // BEGIN HADRON
-                        PAGESTREAM_HANDLER_RESULTS_TOTAL
-                            .with_label_values(&[
-                                metrics::PAGESTREAM_HANDLER_OUTCOME_INTERNAL_ERROR,
-                            ])
-                            .inc();
-                        // END HADRON
+                    PageStreamError::Reconnect(reason) => {
+                        span.in_scope(|| info!("handler requested reconnect: {reason}"));
                        return Err(QueryError::Reconnect);
                    }
                    PageStreamError::Read(_)
                    | PageStreamError::LsnTimeout(_)
                    | PageStreamError::NotFound(_)
                    | PageStreamError::BadRequest(_) => {
-                        // BEGIN HADRON
-                        if let PageStreamError::Read(_) | PageStreamError::LsnTimeout(_) = &e.err {
-                            PAGESTREAM_HANDLER_RESULTS_TOTAL
-                                .with_label_values(&[
-                                    metrics::PAGESTREAM_HANDLER_OUTCOME_INTERNAL_ERROR,
-                                ])
-                                .inc();
-                        } else {
-                            PAGESTREAM_HANDLER_RESULTS_TOTAL
-                                .with_label_values(&[
-                                    metrics::PAGESTREAM_HANDLER_OUTCOME_OTHER_ERROR,
-                                ])
-                                .inc();
-                        }
-                        // END HADRON
-
                        // print the all details to the log with {:#}, but for the client the
                        // error message is enough.  Do not log if shutting down, as the anyhow::Error
                        // here includes cancellation which is not an error.
@@ -1509,15 +1472,7 @@ impl PageServerHandler {
                        )
                    }
                },
-                Ok((response_msg, _op_timer_already_observed, ctx)) => {
-                    // BEGIN HADRON
-                    PAGESTREAM_HANDLER_RESULTS_TOTAL
-                        .with_label_values(&[metrics::PAGESTREAM_HANDLER_OUTCOME_SUCCESS])
-                        .inc();
-                    // END HADRON
-
-                    (response_msg, Some(ctx))
-                }
+                Ok((response_msg, _op_timer_already_observed, ctx)) => (response_msg, Some(ctx)),
            };

            let ctx = ctx.map(|req_ctx| {
@@ -3338,12 +3293,9 @@ impl GrpcPageServiceHandler {
    }

    /// Generates a PagestreamRequest header from a ReadLsn and request ID.
-    fn make_hdr(
-        read_lsn: page_api::ReadLsn,
-        req_id: Option<page_api::RequestID>,
-    ) -> PagestreamRequest {
+    fn make_hdr(read_lsn: page_api::ReadLsn, req_id: u64) -> PagestreamRequest {
        PagestreamRequest {
-            reqid: req_id.map(|r| r.id).unwrap_or_default(),
+            reqid: req_id,
            request_lsn: read_lsn.request_lsn,
            not_modified_since: read_lsn
                .not_modified_since_lsn
@@ -3453,7 +3405,7 @@ impl GrpcPageServiceHandler {

            batch.push(BatchedGetPageRequest {
                req: PagestreamGetPageRequest {
-                    hdr: Self::make_hdr(req.read_lsn, Some(req.request_id)),
+                    hdr: Self::make_hdr(req.read_lsn, req.request_id),
                    rel: req.rel,
                    blkno,
                },
@@ -3483,16 +3435,12 @@ impl GrpcPageServiceHandler {
            request_id: req.request_id,
            status_code: page_api::GetPageStatusCode::Ok,
            reason: None,
-            rel: req.rel,
-            pages: Vec::with_capacity(results.len()),
+            page_images: Vec::with_capacity(results.len()),
        };

        for result in results {
            match result {
-                Ok((PagestreamBeMessage::GetPage(r), _, _)) => resp.pages.push(page_api::Page {
-                    block_number: r.req.blkno,
-                    image: r.page,
-                }),
+                Ok((PagestreamBeMessage::GetPage(r), _, _)) => resp.page_images.push(r.page),
                Ok((resp, _, _)) => {
                    return Err(tonic::Status::internal(format!(
                        "unexpected response: {resp:?}"
@@ -3535,7 +3483,7 @@ impl proto::PageService for GrpcPageServiceHandler {
        span_record!(rel=%req.rel, lsn=%req.read_lsn);

        let req = PagestreamExistsRequest {
-            hdr: Self::make_hdr(req.read_lsn, None),
+            hdr: Self::make_hdr(req.read_lsn, 0),
            rel: req.rel,
        };

@@ -3685,7 +3633,7 @@ impl proto::PageService for GrpcPageServiceHandler {
        span_record!(db_oid=%req.db_oid, lsn=%req.read_lsn);

        let req = PagestreamDbSizeRequest {
-            hdr: Self::make_hdr(req.read_lsn, None),
+            hdr: Self::make_hdr(req.read_lsn, 0),
            dbnode: req.db_oid,
        };

@@ -3735,7 +3683,7 @@ impl proto::PageService for GrpcPageServiceHandler {
                .await?
                .downgrade();
            while let Some(req) = reqs.message().await? {
-                let req_id = req.request_id.map(page_api::RequestID::from).unwrap_or_default();
+                let req_id = req.request_id;
                let result = Self::get_page(&ctx, &timeline, req, io_concurrency.clone())
                    .instrument(span.clone()) // propagate request span
                    .await;
@@ -3774,7 +3722,7 @@ impl proto::PageService for GrpcPageServiceHandler {
        span_record!(rel=%req.rel, lsn=%req.read_lsn);

        let req = PagestreamNblocksRequest {
-            hdr: Self::make_hdr(req.read_lsn, None),
+            hdr: Self::make_hdr(req.read_lsn, 0),
            rel: req.rel,
        };

@@ -3807,7 +3755,7 @@ impl proto::PageService for GrpcPageServiceHandler {
        span_record!(kind=%req.kind, segno=%req.segno, lsn=%req.read_lsn);

        let req = PagestreamGetSlruSegmentRequest {
-            hdr: Self::make_hdr(req.read_lsn, None),
+            hdr: Self::make_hdr(req.read_lsn, 0),
            kind: req.kind as u8,
            segno: req.segno,
        };
--- a/pageserver/src/tenant.rs
+++ b/pageserver/src/tenant.rs
@@ -3291,7 +3291,7 @@ impl TenantShard {
                        // Ignore this, we likely raced with unarchival.
                        OffloadError::NotArchived => Ok(()),
                        OffloadError::AlreadyInProgress => Ok(()),
-                        OffloadError::Cancelled => Err(CompactionError::new_cancelled()),
+                        OffloadError::Cancelled => Err(CompactionError::ShuttingDown),
                        // don't break the anyhow chain
                        OffloadError::Other(err) => Err(CompactionError::Other(err)),
                    })?;
@@ -3321,13 +3321,16 @@ impl TenantShard {

    /// Trips the compaction circuit breaker if appropriate.
    pub(crate) fn maybe_trip_compaction_breaker(&self, err: &CompactionError) {
-        if err.is_cancel() {
-            return;
+        match err {
+            err if err.is_cancel() => {}
+            CompactionError::ShuttingDown => (),
+            CompactionError::Other(err) => {
+                self.compaction_circuit_breaker
+                    .lock()
+                    .unwrap()
+                    .fail(&CIRCUIT_BREAKERS_BROKEN, err);
+            }
        }
-        self.compaction_circuit_breaker
-            .lock()
-            .unwrap()
-            .fail(&CIRCUIT_BREAKERS_BROKEN, err);
    }

    /// Cancel scheduled compaction tasks
@@ -4171,15 +4174,6 @@ impl TenantShard {
            .unwrap_or(self.conf.default_tenant_conf.image_creation_threshold)
    }

-    // HADRON
-    pub fn get_image_creation_timeout(&self) -> Option<Duration> {
-        let tenant_conf = self.tenant_conf.load().tenant_conf.clone();
-        tenant_conf.image_layer_force_creation_period.or(self
-            .conf
-            .default_tenant_conf
-            .image_layer_force_creation_period)
-    }
-
    pub fn get_pitr_interval(&self) -> Duration {
        let tenant_conf = self.tenant_conf.load().tenant_conf.clone();
        tenant_conf
@@ -5719,16 +5713,6 @@ impl TenantShard {
            .unwrap_or(0)
    }

-    /// HADRON
-    /// Return the visible size of all timelines in this tenant.
-    pub(crate) fn get_visible_size(&self) -> u64 {
-        let timelines = self.timelines.lock().unwrap();
-        timelines
-            .values()
-            .map(|t| t.metrics.visible_physical_size_gauge.get())
-            .sum()
-    }
-
    /// Builds a new tenant manifest, and uploads it if it differs from the last-known tenant
    /// manifest in `Self::remote_tenant_manifest`.
    ///
@@ -12816,40 +12800,6 @@ mod tests {
                },
            ]
        );
-
-        Ok(())
-    }
-
-    #[tokio::test]
-    async fn test_get_force_image_creation_lsn() -> anyhow::Result<()> {
-        let tenant_conf = pageserver_api::models::TenantConfig {
-            pitr_interval: Some(Duration::from_secs(7 * 3600)),
-            image_layer_force_creation_period: Some(Duration::from_secs(3600)),
-            ..Default::default()
-        };
-
-        let tenant_id = TenantId::generate();
-
-        let harness = TenantHarness::create_custom(
-            "test_get_force_image_creation_lsn",
-            tenant_conf,
-            tenant_id,
-            ShardIdentity::unsharded(),
-            Generation::new(1),
-        )
-        .await?;
-        let (tenant, ctx) = harness.load().await;
-        let timeline = tenant
-            .create_test_timeline(TIMELINE_ID, Lsn(0x10), DEFAULT_PG_VERSION, &ctx)
-            .await?;
-        timeline.gc_info.write().unwrap().cutoffs.time = Some(Lsn(100));
-        {
-            let writer = timeline.writer().await;
-            writer.finish_write(Lsn(5000));
-        }
-
-        let image_creation_lsn = timeline.get_force_image_creation_lsn().unwrap();
-        assert_eq!(image_creation_lsn, Lsn(4300));
        Ok(())
    }
 }
--- a/pageserver/src/tenant/layer_map.rs
+++ b/pageserver/src/tenant/layer_map.rs
@@ -46,11 +46,10 @@
 mod historic_layer_coverage;
 mod layer_coverage;

-use std::collections::{BTreeMap, HashMap, VecDeque};
+use std::collections::{HashMap, VecDeque};
 use std::iter::Peekable;
 use std::ops::Range;
 use std::sync::Arc;
-use std::time::Instant;

 use anyhow::Result;
 use historic_layer_coverage::BufferedHistoricLayerCoverage;
@@ -905,103 +904,6 @@ impl LayerMap {
        max_stacked_deltas
    }

-    /* BEGIN_HADRON */
-    /**
-     * Compute the image consistent LSN, the largest LSN below which all pages have been redone successfully.
-     * It works by first finding the latest image layers and store them into a map. Then for each delta layer,
-     * find all overlapping image layers in order to potentially increase the image LSN in case there are gaps
-     * (e.g., if an image is created at LSN 100 but the delta layer spans LSN [150, 200], then we can increase
-     * image LSN to 150 because there is no WAL record in between).
-     * Finally, the image consistent LSN is computed by taking the minimum of all image layers.
-     */
-    pub fn compute_image_consistent_lsn(&self, disk_consistent_lsn: Lsn) -> Lsn {
-        struct ImageLayerInfo {
-            // creation LSN of the image layer
-            image_lsn: Lsn,
-            // the current minimum LSN of newer delta layers with overlapping key ranges
-            min_delta_lsn: Lsn,
-        }
-        let started_at = Instant::now();
-
-        let min_l0_deltas_lsn = {
-            let l0_deltas = self.level0_deltas();
-            l0_deltas
-                .iter()
-                .map(|layer| layer.get_lsn_range().start)
-                .min()
-                .unwrap_or(disk_consistent_lsn)
-        };
-        let global_key_range = Key::MIN..Key::MAX;
-
-        // step 1: collect all most recent image layers into a map
-        // map: end key to image_layer_info
-        let mut image_map: BTreeMap<Key, ImageLayerInfo> = BTreeMap::new();
-        for (img_range, img) in self.image_coverage(&global_key_range, disk_consistent_lsn) {
-            let img_lsn = img.map(|layer| layer.get_lsn_range().end).unwrap_or(Lsn(0));
-            image_map.insert(
-                img_range.end,
-                ImageLayerInfo {
-                    image_lsn: img_lsn,
-                    min_delta_lsn: min_l0_deltas_lsn,
-                },
-            );
-        }
-
-        // step 2: go through all delta layers, and update the image layer info with overlapping
-        // key ranges
-        for layer in self.historic.iter() {
-            if !layer.is_delta {
-                continue;
-            }
-            let delta_key_range = layer.get_key_range();
-            let delta_lsn_range = layer.get_lsn_range();
-            for (img_end_key, img_info) in image_map.range_mut(delta_key_range.start..Key::MAX) {
-                debug_assert!(img_end_key >= &delta_key_range.start);
-                if delta_lsn_range.end > img_info.image_lsn {
-                    // the delta layer includes WAL records after the image
-                    // it's possibel that the delta layer's start LSN < image LSN, which will be simply ignored by step 3
-                    img_info.min_delta_lsn =
-                        std::cmp::min(img_info.min_delta_lsn, delta_lsn_range.start);
-                }
-                if img_end_key >= &delta_key_range.end {
-                    // we have fully processed all overlapping image layers
-                    break;
-                }
-            }
-        }
-
-        // step 3, go through all image layers and find the image consistent LSN
-        let mut img_consistent_lsn = min_l0_deltas_lsn.checked_sub(Lsn(1)).unwrap();
-        let mut prev_key = Key::MIN;
-        for (img_key, img_info) in image_map {
-            tracing::debug!(
-                "Image layer {:?}:{} has min delta lsn {}",
-                Range {
-                    start: prev_key,
-                    end: img_key,
-                },
-                img_info.image_lsn,
-                img_info.min_delta_lsn,
-            );
-            let image_lsn = std::cmp::max(
-                img_info.image_lsn,
-                img_info.min_delta_lsn.checked_sub(Lsn(1)).unwrap_or(Lsn(0)),
-            );
-            img_consistent_lsn = std::cmp::min(img_consistent_lsn, image_lsn);
-            prev_key = img_key;
-        }
-        tracing::info!(
-            "computed image_consistent_lsn {} for disk_consistent_lsn {} in {}ms. Processed {} layrs in total.",
-            img_consistent_lsn,
-            disk_consistent_lsn,
-            started_at.elapsed().as_millis(),
-            self.historic.len()
-        );
-        img_consistent_lsn
-    }
-
-    /* END_HADRON */
-
    /// Return all L0 delta layers
    pub fn level0_deltas(&self) -> &Vec<Arc<PersistentLayerDesc>> {
        &self.l0_delta_layers
@@ -1677,138 +1579,6 @@ mod tests {
            LayerVisibilityHint::Visible
        ));
    }
-
-    /* BEGIN_HADRON */
-    #[test]
-    fn test_compute_image_consistent_lsn() {
-        let mut layer_map = LayerMap::default();
-
-        let disk_consistent_lsn = Lsn(1000);
-        // case 1: empty layer map
-        let image_consistent_lsn = layer_map.compute_image_consistent_lsn(disk_consistent_lsn);
-        assert_eq!(
-            disk_consistent_lsn.checked_sub(Lsn(1)).unwrap(),
-            image_consistent_lsn
-        );
-
-        // case 2: only L0 delta layer
-        {
-            let mut updates = layer_map.batch_update();
-            updates.insert_historic(PersistentLayerDesc::new_test(
-                Key::from_i128(0)..Key::from_i128(100),
-                Lsn(900)..Lsn(990),
-                true,
-            ));
-
-            updates.insert_historic(PersistentLayerDesc::new_test(
-                Key::from_i128(0)..Key::from_i128(100),
-                Lsn(850)..Lsn(899),
-                true,
-            ));
-        }
-
-        // should use min L0 delta LSN - 1 as image consistent LSN
-        let image_consistent_lsn = layer_map.compute_image_consistent_lsn(disk_consistent_lsn);
-        assert_eq!(Lsn(849), image_consistent_lsn);
-
-        // case 3: 3 images, no L1 delta
-        {
-            let mut updates = layer_map.batch_update();
-            updates.insert_historic(PersistentLayerDesc::new_test(
-                Key::from_i128(0)..Key::from_i128(40),
-                Lsn(100)..Lsn(100),
-                false,
-            ));
-
-            updates.insert_historic(PersistentLayerDesc::new_test(
-                Key::from_i128(40)..Key::from_i128(70),
-                Lsn(200)..Lsn(200),
-                false,
-            ));
-
-            updates.insert_historic(PersistentLayerDesc::new_test(
-                Key::from_i128(70)..Key::from_i128(100),
-                Lsn(150)..Lsn(150),
-                false,
-            ));
-        }
-        // should use min L0 delta LSN - 1 as image consistent LSN
-        let image_consistent_lsn = layer_map.compute_image_consistent_lsn(disk_consistent_lsn);
-        assert_eq!(Lsn(849), image_consistent_lsn);
-
-        // case 4: 3 images with 1 L1 delta
-        {
-            let mut updates = layer_map.batch_update();
-            updates.insert_historic(PersistentLayerDesc::new_test(
-                Key::from_i128(0)..Key::from_i128(50),
-                Lsn(300)..Lsn(350),
-                true,
-            ));
-        }
-        let image_consistent_lsn = layer_map.compute_image_consistent_lsn(disk_consistent_lsn);
-        assert_eq!(Lsn(299), image_consistent_lsn);
-
-        // case 5: 3 images with 1 more L1 delta with smaller LSN
-        {
-            let mut updates = layer_map.batch_update();
-            updates.insert_historic(PersistentLayerDesc::new_test(
-                Key::from_i128(50)..Key::from_i128(72),
-                Lsn(200)..Lsn(300),
-                true,
-            ));
-        }
-        let image_consistent_lsn = layer_map.compute_image_consistent_lsn(disk_consistent_lsn);
-        assert_eq!(Lsn(199), image_consistent_lsn);
-
-        // case 6: 3 images with more newer L1 deltas (no impact on final results)
-        {
-            let mut updates = layer_map.batch_update();
-            updates.insert_historic(PersistentLayerDesc::new_test(
-                Key::from_i128(0)..Key::from_i128(30),
-                Lsn(400)..Lsn(500),
-                true,
-            ));
-            updates.insert_historic(PersistentLayerDesc::new_test(
-                Key::from_i128(35)..Key::from_i128(100),
-                Lsn(450)..Lsn(600),
-                true,
-            ));
-        }
-        let image_consistent_lsn = layer_map.compute_image_consistent_lsn(disk_consistent_lsn);
-        assert_eq!(Lsn(199), image_consistent_lsn);
-
-        // case 7: 3 images with more older L1 deltas (no impact on final results)
-        {
-            let mut updates = layer_map.batch_update();
-            updates.insert_historic(PersistentLayerDesc::new_test(
-                Key::from_i128(0)..Key::from_i128(40),
-                Lsn(0)..Lsn(50),
-                true,
-            ));
-
-            updates.insert_historic(PersistentLayerDesc::new_test(
-                Key::from_i128(50)..Key::from_i128(100),
-                Lsn(10)..Lsn(60),
-                true,
-            ));
-        }
-        let image_consistent_lsn = layer_map.compute_image_consistent_lsn(disk_consistent_lsn);
-        assert_eq!(Lsn(199), image_consistent_lsn);
-
-        // case 8: 3 images with one more L1 delta with overlapping LSN range
-        {
-            let mut updates = layer_map.batch_update();
-            updates.insert_historic(PersistentLayerDesc::new_test(
-                Key::from_i128(0)..Key::from_i128(50),
-                Lsn(50)..Lsn(250),
-                true,
-            ));
-        }
-        let image_consistent_lsn = layer_map.compute_image_consistent_lsn(disk_consistent_lsn);
-        assert_eq!(Lsn(100), image_consistent_lsn);
-    }
-
-    /* END_HADRON */
 }

 #[cfg(test)]
--- a/pageserver/src/tenant/storage_layer/layer_name.rs
+++ b/pageserver/src/tenant/storage_layer/layer_name.rs
@@ -225,7 +225,7 @@ impl fmt::Display for ImageLayerName {
 /// storage and object names in remote storage consist of the LayerName plus some extra qualifiers
 /// that uniquely identify the physical incarnation of a layer (see [crate::tenant::remote_timeline_client::remote_layer_path])
 /// and [`crate::tenant::storage_layer::layer::local_layer_path`])
-#[derive(Debug, PartialEq, Eq, Hash, Clone, Ord, PartialOrd)]
+#[derive(Debug, PartialEq, Eq, Hash, Clone)]
 pub enum LayerName {
    Image(ImageLayerName),
    Delta(DeltaLayerName),
--- a/pageserver/src/tenant/tasks.rs
+++ b/pageserver/src/tenant/tasks.rs
@@ -17,14 +17,17 @@ use tracing::*;
 use utils::backoff::exponential_backoff_duration;
 use utils::completion::Barrier;
 use utils::pausable_failpoint;
+use utils::sync::gate::GateError;

 use crate::context::{DownloadBehavior, RequestContext};
 use crate::metrics::{self, BackgroundLoopSemaphoreMetricsRecorder, TENANT_TASK_EVENTS};
 use crate::task_mgr::{self, BACKGROUND_RUNTIME, TOKIO_WORKER_THREADS, TaskKind};
+use crate::tenant::blob_io::WriteBlobError;
 use crate::tenant::throttle::Stats;
 use crate::tenant::timeline::CompactionError;
 use crate::tenant::timeline::compaction::CompactionOutcome;
 use crate::tenant::{TenantShard, TenantState};
+use crate::virtual_file::owned_buffers_io::write::FlushTaskError;

 /// Semaphore limiting concurrent background tasks (across all tenants).
 ///
@@ -307,12 +310,45 @@ pub(crate) fn log_compaction_error(
    task_cancelled: bool,
    degrade_to_warning: bool,
 ) {
-    let is_cancel = err.is_cancel();
+    use CompactionError::*;

-    let level = if is_cancel || task_cancelled {
-        Level::INFO
-    } else {
-        Level::ERROR
+    use crate::tenant::PageReconstructError;
+    use crate::tenant::upload_queue::NotInitialized;
+
+    let level = match err {
+        e if e.is_cancel() => return,
+        ShuttingDown => return,
+        _ if task_cancelled => Level::INFO,
+        Other(err) => {
+            let root_cause = err.root_cause();
+
+            let upload_queue = root_cause
+                .downcast_ref::<NotInitialized>()
+                .is_some_and(|e| e.is_stopping());
+            let timeline = root_cause
+                .downcast_ref::<PageReconstructError>()
+                .is_some_and(|e| e.is_cancel());
+            let buffered_writer_flush_task_canelled = root_cause
+                .downcast_ref::<FlushTaskError>()
+                .is_some_and(|e| e.is_cancel());
+            let write_blob_cancelled = root_cause
+                .downcast_ref::<WriteBlobError>()
+                .is_some_and(|e| e.is_cancel());
+            let gate_closed = root_cause
+                .downcast_ref::<GateError>()
+                .is_some_and(|e| e.is_cancel());
+            let is_stopping = upload_queue
+                || timeline
+                || buffered_writer_flush_task_canelled
+                || write_blob_cancelled
+                || gate_closed;
+
+            if is_stopping {
+                Level::INFO
+            } else {
+                Level::ERROR
+            }
+        }
    };

    if let Some((error_count, sleep_duration)) = retry_info {
--- a/pageserver/src/tenant/timeline.rs
+++ b/pageserver/src/tenant/timeline.rs
@@ -1002,7 +1002,7 @@ impl From<WaitLsnError> for tonic::Status {
 impl From<CreateImageLayersError> for CompactionError {
    fn from(e: CreateImageLayersError) -> Self {
        match e {
-            CreateImageLayersError::Cancelled => CompactionError::new_cancelled(),
+            CreateImageLayersError::Cancelled => CompactionError::ShuttingDown,
            CreateImageLayersError::Other(e) => {
                CompactionError::Other(e.context("create image layers"))
            }
@@ -2117,7 +2117,12 @@ impl Timeline {
        match &result {
            Ok(_) => self.compaction_failed.store(false, AtomicOrdering::Relaxed),
            Err(e) if e.is_cancel() => {}
-            Err(_) => self.compaction_failed.store(true, AtomicOrdering::Relaxed),
+            Err(CompactionError::ShuttingDown) => {
+                // Covered by the `Err(e) if e.is_cancel()` branch.
+            }
+            Err(CompactionError::Other(_)) => {
+                self.compaction_failed.store(true, AtomicOrdering::Relaxed)
+            }
        };

        result
@@ -2846,18 +2851,6 @@ impl Timeline {
            .unwrap_or(self.conf.default_tenant_conf.image_creation_threshold)
    }

-    // HADRON
-    fn get_image_layer_force_creation_period(&self) -> Option<Duration> {
-        let tenant_conf = self.tenant_conf.load();
-        tenant_conf
-            .tenant_conf
-            .image_layer_force_creation_period
-            .or(self
-                .conf
-                .default_tenant_conf
-                .image_layer_force_creation_period)
-    }
-
    fn get_compaction_algorithm_settings(&self) -> CompactionAlgorithmSettings {
        let tenant_conf = &self.tenant_conf.load();
        tenant_conf
@@ -3127,6 +3120,7 @@ impl Timeline {
                repartition_threshold: 0,
                last_image_layer_creation_check_at: AtomicLsn::new(0),
                last_image_layer_creation_check_instant: Mutex::new(None),
+
                last_received_wal: Mutex::new(None),
                rel_size_latest_cache: RwLock::new(HashMap::new()),
                rel_size_snapshot_cache: Mutex::new(LruCache::new(relsize_snapshot_cache_capacity)),
@@ -5047,7 +5041,6 @@ impl Timeline {
                .create_image_layers(
                    &partitions,
                    self.initdb_lsn,
-                    None,
                    ImageLayerCreationMode::Initial,
                    ctx,
                    LastImageLayerCreationStatus::Initial,
@@ -5319,19 +5312,14 @@ impl Timeline {
    }

    // Is it time to create a new image layer for the given partition? True if we want to generate.
-    async fn time_for_new_image_layer(
-        &self,
-        partition: &KeySpace,
-        lsn: Lsn,
-        force_image_creation_lsn: Option<Lsn>,
-    ) -> bool {
+    async fn time_for_new_image_layer(&self, partition: &KeySpace, lsn: Lsn) -> bool {
        let threshold = self.get_image_creation_threshold();

        let guard = self.layers.read(LayerManagerLockHolder::Compaction).await;
        let Ok(layers) = guard.layer_map() else {
            return false;
        };
-        let mut min_image_lsn: Lsn = Lsn::MAX;
+
        let mut max_deltas = 0;
        for part_range in &partition.ranges {
            let image_coverage = layers.image_coverage(part_range, lsn);
@@ -5366,25 +5354,9 @@ impl Timeline {
                        return true;
                    }
                }
-                min_image_lsn = min(min_image_lsn, img_lsn);
            }
        }

-        // HADRON
-        // for child timelines, we consider all pages up to ancestor_LSN are redone successfully by the parent timeline
-        min_image_lsn = min_image_lsn.max(self.get_ancestor_lsn());
-        if min_image_lsn < force_image_creation_lsn.unwrap_or(Lsn(0)) && max_deltas > 0 {
-            info!(
-                "forcing image creation for partitioned range {}-{}. Min image LSN: {}, force image creation LSN: {}, num deltas: {}",
-                partition.ranges[0].start,
-                partition.ranges[0].end,
-                min_image_lsn,
-                force_image_creation_lsn.unwrap(),
-                max_deltas
-            );
-            return true;
-        }
-
        debug!(
            max_deltas,
            "none of the partitioned ranges had >= {threshold} deltas"
@@ -5610,7 +5582,7 @@ impl Timeline {
    ///        suffer from the lack of image layers
    ///     2. For small tenants (that can mostly fit in RAM), we use a much longer interval
    fn should_check_if_image_layers_required(self: &Arc<Timeline>, lsn: Lsn) -> bool {
-        let large_timeline_threshold = self.conf.image_layer_generation_large_timeline_threshold;
+        const LARGE_TENANT_THRESHOLD: u64 = 2 * 1024 * 1024 * 1024;

        let last_checks_at = self.last_image_layer_creation_check_at.load();
        let distance = lsn
@@ -5624,12 +5596,12 @@ impl Timeline {
        let mut time_based_decision = false;
        let mut last_check_instant = self.last_image_layer_creation_check_instant.lock().unwrap();
        if let CurrentLogicalSize::Exact(logical_size) = self.current_logical_size.current_size() {
-            let check_required_after =
-                if Some(Into::<u64>::into(&logical_size)) >= large_timeline_threshold {
-                    self.get_checkpoint_timeout()
-                } else {
-                    Duration::from_secs(3600 * 48)
-                };
+            let check_required_after = if Into::<u64>::into(&logical_size) >= LARGE_TENANT_THRESHOLD
+            {
+                self.get_checkpoint_timeout()
+            } else {
+                Duration::from_secs(3600 * 48)
+            };

            time_based_decision = match *last_check_instant {
                Some(last_check) => {
@@ -5657,12 +5629,10 @@ impl Timeline {
    /// true = we have generate all image layers, false = we preempt the process for L0 compaction.
    ///
    /// `partition_mode` is only for logging purpose and is not used anywhere in this function.
-    #[allow(clippy::too_many_arguments)]
    async fn create_image_layers(
        self: &Arc<Timeline>,
        partitioning: &KeyPartitioning,
        lsn: Lsn,
-        force_image_creation_lsn: Option<Lsn>,
        mode: ImageLayerCreationMode,
        ctx: &RequestContext,
        last_status: LastImageLayerCreationStatus,
@@ -5766,11 +5736,7 @@ impl Timeline {
            } else if let ImageLayerCreationMode::Try = mode {
                // check_for_image_layers = false -> skip
                // check_for_image_layers = true -> check time_for_new_image_layer -> skip/generate
-                if !check_for_image_layers
-                    || !self
-                        .time_for_new_image_layer(partition, lsn, force_image_creation_lsn)
-                        .await
-                {
+                if !check_for_image_layers || !self.time_for_new_image_layer(partition, lsn).await {
                    start = img_range.end;
                    continue;
                }
@@ -6091,88 +6057,26 @@ impl Drop for Timeline {
    }
 }

-pub(crate) use compaction_error::CompactionError;
-/// In a private mod to enforce that [`CompactionError::is_cancel`] is used
-/// instead of `match`ing on [`CompactionError::ShuttingDown`].
-mod compaction_error {
-    use utils::sync::gate::GateError;
+/// Top-level failure to compact.
+#[derive(Debug, thiserror::Error)]
+pub(crate) enum CompactionError {
+    #[error("The timeline or pageserver is shutting down")]
+    ShuttingDown,
+    #[error(transparent)]
+    Other(anyhow::Error),
+}

-    use crate::{
-        pgdatadir_mapping::CollectKeySpaceError,
-        tenant::{PageReconstructError, blob_io::WriteBlobError, upload_queue::NotInitialized},
-        virtual_file::owned_buffers_io::write::FlushTaskError,
-    };
-
-    /// Top-level failure to compact. Use [`Self::is_cancel`].
-    #[derive(Debug, thiserror::Error)]
-    pub(crate) enum CompactionError {
-        /// Use [`Self::is_cancel`] instead of checking for this variant.
-        #[error("The timeline or pageserver is shutting down")]
-        #[allow(private_interfaces)]
-        ShuttingDown(ForbidMatching), // private ForbidMatching enforces use of [`Self::is_cancel`].
-        #[error(transparent)]
-        Other(anyhow::Error),
+impl CompactionError {
+    /// Errors that can be ignored, i.e., cancel and shutdown.
+    pub fn is_cancel(&self) -> bool {
+        matches!(self, Self::ShuttingDown)
    }

-    #[derive(Debug)]
-    struct ForbidMatching;
-
-    impl CompactionError {
-        pub fn new_cancelled() -> Self {
-            Self::ShuttingDown(ForbidMatching)
-        }
-        /// Errors that can be ignored, i.e., cancel and shutdown.
-        pub fn is_cancel(&self) -> bool {
-            let other = match self {
-                CompactionError::ShuttingDown(_) => return true,
-                CompactionError::Other(other) => other,
-            };
-
-            // The write path of compaction in particular often lacks differentiated
-            // handling errors stemming from cancellation from other errors.
-            // So, if requested, we also check the ::Other variant by downcasting.
-            // The list below has been found empirically from flaky tests and production logs.
-            // The process is simple: on ::Other(), compaction will print the enclosed
-            // anyhow::Error in debug mode, i.e., with backtrace. That backtrace contains the
-            // line where the write path / compaction code does undifferentiated error handling
-            // from a non-anyhow type to an anyhow type. Add the type to the list of downcasts
-            // below, following the same is_cancel() pattern.
-
-            let root_cause = other.root_cause();
-
-            let upload_queue = root_cause
-                .downcast_ref::<NotInitialized>()
-                .is_some_and(|e| e.is_stopping());
-            let timeline = root_cause
-                .downcast_ref::<PageReconstructError>()
-                .is_some_and(|e| e.is_cancel());
-            let buffered_writer_flush_task_canelled = root_cause
-                .downcast_ref::<FlushTaskError>()
-                .is_some_and(|e| e.is_cancel());
-            let write_blob_cancelled = root_cause
-                .downcast_ref::<WriteBlobError>()
-                .is_some_and(|e| e.is_cancel());
-            let gate_closed = root_cause
-                .downcast_ref::<GateError>()
-                .is_some_and(|e| e.is_cancel());
-            upload_queue
-                || timeline
-                || buffered_writer_flush_task_canelled
-                || write_blob_cancelled
-                || gate_closed
-        }
-        pub fn into_anyhow(self) -> anyhow::Error {
-            match self {
-                CompactionError::ShuttingDown(ForbidMatching) => anyhow::Error::new(self),
-                CompactionError::Other(e) => e,
-            }
-        }
-        pub fn from_collect_keyspace(err: CollectKeySpaceError) -> Self {
-            if err.is_cancel() {
-                Self::new_cancelled()
-            } else {
-                Self::Other(err.into_anyhow())
-            }
+    pub fn from_collect_keyspace(err: CollectKeySpaceError) -> Self {
+        if err.is_cancel() {
+            Self::ShuttingDown
+        } else {
+            Self::Other(err.into_anyhow())
        }
    }
 }
@@ -6184,7 +6088,7 @@ impl From<super::upload_queue::NotInitialized> for CompactionError {
                CompactionError::Other(anyhow::anyhow!(value))
            }
            super::upload_queue::NotInitialized::ShuttingDown
-            | super::upload_queue::NotInitialized::Stopped => CompactionError::new_cancelled(),
+            | super::upload_queue::NotInitialized::Stopped => CompactionError::ShuttingDown,
        }
    }
 }
@@ -6194,7 +6098,7 @@ impl From<super::storage_layer::layer::DownloadError> for CompactionError {
        match e {
            super::storage_layer::layer::DownloadError::TimelineShutdown
            | super::storage_layer::layer::DownloadError::DownloadCancelled => {
-                CompactionError::new_cancelled()
+                CompactionError::ShuttingDown
            }
            super::storage_layer::layer::DownloadError::ContextAndConfigReallyDeniesDownloads
            | super::storage_layer::layer::DownloadError::DownloadRequired
@@ -6213,14 +6117,14 @@ impl From<super::storage_layer::layer::DownloadError> for CompactionError {

 impl From<layer_manager::Shutdown> for CompactionError {
    fn from(_: layer_manager::Shutdown) -> Self {
-        CompactionError::new_cancelled()
+        CompactionError::ShuttingDown
    }
 }

 impl From<super::storage_layer::errors::PutError> for CompactionError {
    fn from(e: super::storage_layer::errors::PutError) -> Self {
        if e.is_cancel() {
-            CompactionError::new_cancelled()
+            CompactionError::ShuttingDown
        } else {
            CompactionError::Other(e.into_anyhow())
        }
@@ -6319,7 +6223,7 @@ impl Timeline {
        let mut guard = tokio::select! {
            guard = self.layers.write(LayerManagerLockHolder::Compaction) => guard,
            _ = self.cancel.cancelled() => {
-                return Err(CompactionError::new_cancelled());
+                return Err(CompactionError::ShuttingDown);
            }
        };

@@ -7146,19 +7050,6 @@ impl Timeline {
            .unwrap()
            .clone()
    }
-
-    /* BEGIN_HADRON */
-    pub(crate) async fn compute_image_consistent_lsn(&self) -> anyhow::Result<Lsn> {
-        let guard = self
-            .layers
-            .read(LayerManagerLockHolder::ComputeImageConsistentLsn)
-            .await;
-        let layer_map = guard.layer_map()?;
-        let disk_consistent_lsn = self.get_disk_consistent_lsn();
-
-        Ok(layer_map.compute_image_consistent_lsn(disk_consistent_lsn))
-    }
-    /* END_HADRON */
 }

 impl Timeline {
--- a/pageserver/src/tenant/timeline/compaction.rs
+++ b/pageserver/src/tenant/timeline/compaction.rs
@@ -4,7 +4,6 @@
 //!
 //! The old legacy algorithm is implemented directly in `timeline.rs`.

-use std::cmp::min;
 use std::collections::{BinaryHeap, HashMap, HashSet, VecDeque};
 use std::ops::{Deref, Range};
 use std::sync::Arc;
@@ -573,8 +572,8 @@ impl GcCompactionQueue {
        }
        match res {
            Ok(res) => Ok(res),
-            Err(e) if e.is_cancel() => Err(e),
-            Err(_) => {
+            Err(CompactionError::ShuttingDown) => Err(CompactionError::ShuttingDown),
+            Err(CompactionError::Other(_)) => {
                // There are some cases where traditional gc might collect some layer
                // files causing gc-compaction cannot read the full history of the key.
                // This needs to be resolved in the long-term by improving the compaction
@@ -1261,16 +1260,13 @@ impl Timeline {
        // Is the timeline being deleted?
        if self.is_stopping() {
            trace!("Dropping out of compaction on timeline shutdown");
-            return Err(CompactionError::new_cancelled());
+            return Err(CompactionError::ShuttingDown);
        }

        let target_file_size = self.get_checkpoint_distance();

        // Define partitioning schema if needed

-        // HADRON
-        let force_image_creation_lsn = self.get_force_image_creation_lsn();
-
        // 1. L0 Compact
        let l0_outcome = {
            let timer = self.metrics.compact_time_histo.start_timer();
@@ -1278,7 +1274,6 @@ impl Timeline {
                .compact_level0(
                    target_file_size,
                    options.flags.contains(CompactFlags::ForceL0Compaction),
-                    force_image_creation_lsn,
                    ctx,
                )
                .await?;
@@ -1381,7 +1376,6 @@ impl Timeline {
                    .create_image_layers(
                        &partitioning,
                        lsn,
-                        force_image_creation_lsn,
                        mode,
                        &image_ctx,
                        self.last_image_layer_creation_status
@@ -1478,41 +1472,6 @@ impl Timeline {
        Ok(CompactionOutcome::Done)
    }

-    /* BEGIN_HADRON */
-    // Get the force image creation LSN based on gc_cutoff_lsn.
-    // Note that this is an estimation and the workload rate may suddenly change. When that happens,
-    // the force image creation may be too early or too late, but eventually it should be able to catch up.
-    pub(crate) fn get_force_image_creation_lsn(self: &Arc<Self>) -> Option<Lsn> {
-        let image_creation_period = self.get_image_layer_force_creation_period()?;
-        let current_lsn = self.get_last_record_lsn();
-        let pitr_lsn = self.gc_info.read().unwrap().cutoffs.time?;
-        let pitr_interval = self.get_pitr_interval();
-        if pitr_lsn == Lsn::INVALID || pitr_interval.is_zero() {
-            tracing::warn!(
-                "pitr LSN/interval not found, skipping force image creation LSN calculation"
-            );
-            return None;
-        }
-
-        let delta_lsn = current_lsn.checked_sub(pitr_lsn).unwrap().0
-            * image_creation_period.as_secs()
-            / pitr_interval.as_secs();
-        let force_image_creation_lsn = current_lsn.checked_sub(delta_lsn).unwrap_or(Lsn(0));
-
-        tracing::info!(
-            "Tenant shard {} computed force_image_creation_lsn: {}. Current lsn: {}, image_layer_force_creation_period: {:?}, GC cutoff: {}, PITR interval: {:?}",
-            self.tenant_shard_id,
-            force_image_creation_lsn,
-            current_lsn,
-            image_creation_period,
-            pitr_lsn,
-            pitr_interval
-        );
-
-        Some(force_image_creation_lsn)
-    }
-    /* END_HADRON */
-
    /// Check for layers that are elegible to be rewritten:
    /// - Shard splitting: After a shard split, ancestor layers beyond pitr_interval, so that
    ///   we don't indefinitely retain keys in this shard that aren't needed.
@@ -1665,7 +1624,7 @@ impl Timeline {

        for (i, layer) in layers_to_rewrite.into_iter().enumerate() {
            if self.cancel.is_cancelled() {
-                return Err(CompactionError::new_cancelled());
+                return Err(CompactionError::ShuttingDown);
            }

            info!(layer=%layer, "rewriting layer after shard split: {}/{}", i, total);
@@ -1763,7 +1722,7 @@ impl Timeline {
                    Ok(()) => {},
                    Err(WaitCompletionError::NotInitialized(ni)) => return Err(CompactionError::from(ni)),
                    Err(WaitCompletionError::UploadQueueShutDownOrStopped) => {
-                        return Err(CompactionError::new_cancelled());
+                        return Err(CompactionError::ShuttingDown);
                    }
                },
                // Don't wait if there's L0 compaction to do. We don't need to update the outcome
@@ -1842,7 +1801,6 @@ impl Timeline {
        self: &Arc<Self>,
        target_file_size: u64,
        force_compaction_ignore_threshold: bool,
-        force_compaction_lsn: Option<Lsn>,
        ctx: &RequestContext,
    ) -> Result<CompactionOutcome, CompactionError> {
        let CompactLevel0Phase1Result {
@@ -1863,7 +1821,6 @@ impl Timeline {
                stats,
                target_file_size,
                force_compaction_ignore_threshold,
-                force_compaction_lsn,
                &ctx,
            )
            .instrument(phase1_span)
@@ -1886,7 +1843,6 @@ impl Timeline {
        mut stats: CompactLevel0Phase1StatsBuilder,
        target_file_size: u64,
        force_compaction_ignore_threshold: bool,
-        force_compaction_lsn: Option<Lsn>,
        ctx: &RequestContext,
    ) -> Result<CompactLevel0Phase1Result, CompactionError> {
        let begin = tokio::time::Instant::now();
@@ -1916,28 +1872,11 @@ impl Timeline {
                    return Ok(CompactLevel0Phase1Result::default());
                }
            } else {
-                // HADRON
-                let min_lsn = level0_deltas
-                    .iter()
-                    .map(|a| a.get_lsn_range().start)
-                    .reduce(min);
-                if force_compaction_lsn.is_some()
-                    && min_lsn.is_some()
-                    && min_lsn.unwrap() < force_compaction_lsn.unwrap()
-                {
-                    info!(
-                        "forcing L0 compaction of {} L0 deltas. Min lsn: {}, force compaction lsn: {}",
-                        level0_deltas.len(),
-                        min_lsn.unwrap(),
-                        force_compaction_lsn.unwrap()
-                    );
-                } else {
-                    debug!(
-                        level0_deltas = level0_deltas.len(),
-                        threshold, "too few deltas to compact"
-                    );
-                    return Ok(CompactLevel0Phase1Result::default());
-                }
+                debug!(
+                    level0_deltas = level0_deltas.len(),
+                    threshold, "too few deltas to compact"
+                );
+                return Ok(CompactLevel0Phase1Result::default());
            }
        }

@@ -2046,7 +1985,7 @@ impl Timeline {
            let mut all_keys = Vec::new();
            for l in deltas_to_compact.iter() {
                if self.cancel.is_cancelled() {
-                    return Err(CompactionError::new_cancelled());
+                    return Err(CompactionError::ShuttingDown);
                }
                let delta = l.get_as_delta(ctx).await.map_err(CompactionError::Other)?;
                let keys = delta
@@ -2139,7 +2078,7 @@ impl Timeline {
        stats.read_lock_held_compute_holes_micros = stats.read_lock_held_key_sort_micros.till_now();

        if self.cancel.is_cancelled() {
-            return Err(CompactionError::new_cancelled());
+            return Err(CompactionError::ShuttingDown);
        }

        stats.read_lock_drop_micros = stats.read_lock_held_compute_holes_micros.till_now();
@@ -2247,7 +2186,7 @@ impl Timeline {
                // avoid hitting the cancellation token on every key. in benches, we end up
                // shuffling an order of million keys per layer, this means we'll check it
                // around tens of times per layer.
-                return Err(CompactionError::new_cancelled());
+                return Err(CompactionError::ShuttingDown);
            }

            let same_key = prev_key == Some(key);
@@ -2332,7 +2271,7 @@ impl Timeline {
                if writer.is_none() {
                    if self.cancel.is_cancelled() {
                        // to be somewhat responsive to cancellation, check for each new layer
-                        return Err(CompactionError::new_cancelled());
+                        return Err(CompactionError::ShuttingDown);
                    }
                    // Create writer if not initiaized yet
                    writer = Some(
@@ -2588,7 +2527,7 @@ impl Timeline {
        // Is the timeline being deleted?
        if self.is_stopping() {
            trace!("Dropping out of compaction on timeline shutdown");
-            return Err(CompactionError::new_cancelled());
+            return Err(CompactionError::ShuttingDown);
        }

        let (dense_ks, _sparse_ks) = self
@@ -3250,7 +3189,7 @@ impl Timeline {
        let gc_lock = async {
            tokio::select! {
                guard = self.gc_lock.lock() => Ok(guard),
-                _ = cancel.cancelled() => Err(CompactionError::new_cancelled()),
+                _ = cancel.cancelled() => Err(CompactionError::ShuttingDown),
            }
        };

@@ -3523,7 +3462,7 @@ impl Timeline {
            }
            total_layer_size += layer.layer_desc().file_size;
            if cancel.is_cancelled() {
-                return Err(CompactionError::new_cancelled());
+                return Err(CompactionError::ShuttingDown);
            }
            let should_yield = yield_for_l0
                && self
@@ -3670,7 +3609,7 @@ impl Timeline {
            }

            if cancel.is_cancelled() {
-                return Err(CompactionError::new_cancelled());
+                return Err(CompactionError::ShuttingDown);
            }

            let should_yield = yield_for_l0
--- a/pageserver/src/tenant/timeline/layer_manager.rs
+++ b/pageserver/src/tenant/timeline/layer_manager.rs
@@ -47,7 +47,6 @@ pub(crate) enum LayerManagerLockHolder {
    ImportPgData,
    DetachAncestor,
    Eviction,
-    ComputeImageConsistentLsn,
    #[cfg(test)]
    Testing,
 }
--- a/pgxn/neon/libpagestore.c
+++ b/pgxn/neon/libpagestore.c
@@ -1410,7 +1410,7 @@ pg_init_libpagestore(void)
 							"sharding stripe size",
 							NULL,
 							&stripe_size,
-							2048, 1, INT_MAX,
+							32768, 1, INT_MAX,
 							PGC_SIGHUP,
 							GUC_UNIT_BLOCKS,
 							NULL, NULL, NULL);
--- a/pgxn/neon/neon_ddl_handler.c
+++ b/pgxn/neon/neon_ddl_handler.c
@@ -953,9 +953,7 @@ neon_fmgr_hook(FmgrHookEventType event, FmgrInfo *flinfo, Datum *private)

 	/*
 	 * Fire Event Trigger if both function owner and current user are
-	 * superuser. Allow executing Event Trigger function that belongs to a
-	 * superuser when connected as a non-superuser, even when the function is
-	 * SECURITY DEFINER.
+	 * superuser, or none of them are.
 	 */
    else if (event == FHET_START
 		/* still enable it to pass pg_regress tests */
@@ -978,7 +976,32 @@ neon_fmgr_hook(FmgrHookEventType event, FmgrInfo *flinfo, Datum *private)
 		function_is_owned_by_super = superuser_arg(function_owner);

 		/*
-		 * Refuse to run functions that belongs to a non-superuser when the
+		 * 1. Refuse to run SECURITY DEFINER function that belongs to a
+		 * superuser when the current user is not a superuser itself.
+		 */
+		if (!role_is_super
+			&& function_is_owned_by_super
+			&& function_is_secdef)
+		{
+			char *func_name = get_func_name(flinfo->fn_oid);
+
+			ereport(WARNING,
+					(errmsg("Skipping Event Trigger"),
+					 errdetail("Event Trigger function \"%s\" is owned by \"%s\" "
+							   "and is SECURITY DEFINER",
+							   func_name,
+							   GetUserNameFromId(function_owner, false))));
+
+			/*
+			 * we can't skip execution directly inside the fmgr_hook so
+			 * instead we change the event trigger function to a noop
+			 * function.
+			 */
+			force_noop(flinfo);
+		}
+
+		/*
+		 * 2. Refuse to run functions that belongs to a non-superuser when the
 		 * current user is a superuser.
 		 *
 		 * We could run a SECURITY DEFINER user-function here and be safe with
@@ -986,7 +1009,7 @@ neon_fmgr_hook(FmgrHookEventType event, FmgrInfo *flinfo, Datum *private)
 		 * infrastructure maintenance operations, where we prefer to skip
 		 * running user-defined code.
 		 */
-		if (role_is_super && !function_is_owned_by_super)
+		else if (role_is_super && !function_is_owned_by_super)
 		{
 			char *func_name = get_func_name(flinfo->fn_oid);

--- a/pgxn/neon/walproposer.h
+++ b/pgxn/neon/walproposer.h
@@ -376,18 +376,6 @@ typedef struct PageserverFeedback
 	uint32		shard_number;
 } PageserverFeedback;

-/* BEGIN_HADRON */
-typedef struct WalRateLimiter
-{
-	/* If the value is 1, PG backends will hit backpressure. */
-	pg_atomic_uint32 should_limit;
-	/* The number of bytes sent in the current second. */
-	uint64		sent_bytes;
-	/* The last recorded time in microsecond. */
-	TimestampTz last_recorded_time_us;
-} WalRateLimiter;
-/* END_HADRON */
-
 typedef struct WalproposerShmemState
 {
 	pg_atomic_uint64 propEpochStartLsn;
@@ -407,11 +395,6 @@ typedef struct WalproposerShmemState

 	/* aggregated feedback with min LSNs across shards */
 	PageserverFeedback min_ps_feedback;
-
-	/* BEGIN_HADRON */
-	/* The WAL rate limiter */
-	WalRateLimiter wal_rate_limiter;
-	/* END_HADRON */
 } WalproposerShmemState;

 /*
--- a/pgxn/neon/walproposer_pg.c
+++ b/pgxn/neon/walproposer_pg.c
@@ -66,9 +66,6 @@ int			wal_acceptor_reconnect_timeout = 1000;
 int			wal_acceptor_connection_timeout = 10000;
 int			safekeeper_proto_version = 3;
 char	   *safekeeper_conninfo_options = "";
-/* BEGIN_HADRON */
-int         databricks_max_wal_mb_per_second = -1;
-/* END_HADRON */

 /* Set to true in the walproposer bgw. */
 static bool am_walproposer;
@@ -255,18 +252,6 @@ nwp_register_gucs(void)
 							PGC_POSTMASTER,
 							0,
 							NULL, NULL, NULL);
-
-    /* BEGIN_HADRON */
-    DefineCustomIntVariable(
-                            "databricks.max_wal_mb_per_second",
-                            "The maximum WAL MB per second allowed. If breached, sending WAL hit the backpressure. Setting to -1 disables the limit.",
-                            NULL,
-                            &databricks_max_wal_mb_per_second,
-                            -1, -1, INT_MAX,
-                            PGC_SUSET,
-                            GUC_UNIT_MB,
-                            NULL, NULL, NULL);
-    /* END_HADRON */
 }


@@ -408,7 +393,6 @@ assign_neon_safekeepers(const char *newval, void *extra)
 static uint64
 backpressure_lag_impl(void)
 {
-	struct WalproposerShmemState* state = NULL;
 	if (max_replication_apply_lag > 0 || max_replication_flush_lag > 0 || max_replication_write_lag > 0)
 	{
 		XLogRecPtr	writePtr;
@@ -442,18 +426,6 @@ backpressure_lag_impl(void)
 			return (myFlushLsn - applyPtr - max_replication_apply_lag * MB);
 		}
 	}
-
-	/* BEGIN_HADRON */
-	if (databricks_max_wal_mb_per_second == -1) {
-		return 0;
-	}
-
-	state = GetWalpropShmemState();
-	if (state != NULL && pg_atomic_read_u32(&state->wal_rate_limiter.should_limit) == 1)
-	{
-		return 1;
-	}
-	/* END_HADRON */
 	return 0;
 }

@@ -500,9 +472,6 @@ WalproposerShmemInit(void)
 		pg_atomic_init_u64(&walprop_shared->mineLastElectedTerm, 0);
 		pg_atomic_init_u64(&walprop_shared->backpressureThrottlingTime, 0);
 		pg_atomic_init_u64(&walprop_shared->currentClusterSize, 0);
-		/* BEGIN_HADRON */
-		pg_atomic_init_u32(&walprop_shared->wal_rate_limiter.should_limit, 0);
-		/* END_HADRON */
 	}
 	LWLockRelease(AddinShmemInitLock);

@@ -518,9 +487,6 @@ WalproposerShmemInit_SyncSafekeeper(void)
 	pg_atomic_init_u64(&walprop_shared->propEpochStartLsn, 0);
 	pg_atomic_init_u64(&walprop_shared->mineLastElectedTerm, 0);
 	pg_atomic_init_u64(&walprop_shared->backpressureThrottlingTime, 0);
-	/* BEGIN_HADRON */
-	pg_atomic_init_u32(&walprop_shared->wal_rate_limiter.should_limit, 0);
-	/* END_HADRON */
 }

 #define BACK_PRESSURE_DELAY 10000L // 0.01 sec
@@ -555,6 +521,7 @@ backpressure_throttling_impl(void)
 	if (lag == 0)
 		return retry;

+
 	old_status = get_ps_display(&len);
 	new_status = (char *) palloc(len + 64 + 1);
 	memcpy(new_status, old_status, len);
@@ -1491,8 +1458,6 @@ XLogBroadcastWalProposer(WalProposer *wp)
 {
 	XLogRecPtr	startptr;
 	XLogRecPtr	endptr;
-	struct WalproposerShmemState *state = NULL;
-	TimestampTz now = 0;

 	/* Start from the last sent position */
 	startptr = sentPtr;
@@ -1537,36 +1502,13 @@ XLogBroadcastWalProposer(WalProposer *wp)
 	 * that arbitrary LSN is eventually reported as written, flushed and
 	 * applied, so that it can measure the elapsed time.
 	 */
-	now = GetCurrentTimestamp();
-	LagTrackerWrite(endptr, now);
+	LagTrackerWrite(endptr, GetCurrentTimestamp());

 	/* Do we have any work to do? */
 	Assert(startptr <= endptr);
 	if (endptr <= startptr)
 		return;

-	/* BEGIN_HADRON */
-	state = GetWalpropShmemState();
-	if (databricks_max_wal_mb_per_second != -1 && state != NULL)
-	{
-		uint64 max_wal_bytes = (uint64) databricks_max_wal_mb_per_second * 1024 * 1024;
-		struct WalRateLimiter *limiter = &state->wal_rate_limiter;
-
-		if (now - limiter->last_recorded_time_us > USECS_PER_SEC)
-		{
-			/* Reset the rate limiter */
-			limiter->last_recorded_time_us = now;
-			limiter->sent_bytes = 0;
-			pg_atomic_exchange_u32(&limiter->should_limit, 0);
-		}
-		limiter->sent_bytes += (endptr - startptr);
-		if (limiter->sent_bytes > max_wal_bytes)
-		{
-			pg_atomic_exchange_u32(&limiter->should_limit, 1);
-		}
-	}
-	/* END_HADRON */
-
 	WalProposerBroadcast(wp, startptr, endptr);
 	sentPtr = endptr;

--- a/proxy/Cargo.toml
+++ b/proxy/Cargo.toml
@@ -48,13 +48,12 @@ indexmap = { workspace = true, features = ["serde"] }
 ipnet.workspace = true
 itertools.workspace = true
 itoa.workspace = true
-measured = { workspace = true, features = ["paracord"] }
-measured-tokio.workspace = true
+lasso = { workspace = true, features = ["multi-threaded"] }
+measured = { workspace = true, features = ["lasso"] }
 metrics.workspace = true
 once_cell.workspace = true
 opentelemetry = { workspace = true, features = ["trace"] }
 papaya = "0.2.0"
-paracord.workspace = true
 parking_lot.workspace = true
 parquet.workspace = true
 parquet_derive.workspace = true
@@ -128,4 +127,4 @@ rstest.workspace = true
 walkdir.workspace = true
 rand_distr = "0.4"
 tokio-postgres.workspace = true
-tracing-test = "0.2"
+tracing-test = "0.2"
--- a/proxy/src/auth/backend/jwt.rs
+++ b/proxy/src/auth/backend/jwt.rs
@@ -87,12 +87,7 @@ impl JwkCacheEntry {
        self.key_sets
            .values()
            // make sure our requested role has access to the key set
-            .filter(|key_set| {
-                key_set
-                    .role_names
-                    .iter()
-                    .any(|role| *role.as_str() == **role_name)
-            })
+            .filter(|key_set| key_set.role_names.iter().any(|role| **role == **role_name))
            // try and find the requested key-id in the key set
            .find_map(|key_set| {
                key_set
--- a/proxy/src/auth/backend/local.rs
+++ b/proxy/src/auth/backend/local.rs
@@ -12,7 +12,7 @@ use crate::context::RequestContext;
 use crate::control_plane::NodeInfo;
 use crate::control_plane::messages::{ColdStartInfo, EndpointJwksResponse, MetricsAuxInfo};
 use crate::http;
-use crate::intern::{BranchIdInt, EndpointIdInt, ProjectIdInt};
+use crate::intern::{BranchIdTag, EndpointIdTag, InternId, ProjectIdTag};
 use crate::types::EndpointId;
 use crate::url::ApiUrl;

@@ -38,9 +38,9 @@ impl LocalBackend {
                },
                // TODO(conrad): make this better reflect compute info rather than endpoint info.
                aux: MetricsAuxInfo {
-                    endpoint_id: EndpointIdInt::new("local"),
-                    project_id: ProjectIdInt::new("local"),
-                    branch_id: BranchIdInt::new("local"),
+                    endpoint_id: EndpointIdTag::get_interner().get_or_intern("local"),
+                    project_id: ProjectIdTag::get_interner().get_or_intern("local"),
+                    branch_id: BranchIdTag::get_interner().get_or_intern("local"),
                    compute_id: "local".into(),
                    cold_start_info: ColdStartInfo::WarmCached,
                },
--- a/proxy/src/binary/local_proxy.rs
+++ b/proxy/src/binary/local_proxy.rs
@@ -192,7 +192,6 @@ pub async fn run() -> anyhow::Result<()> {
            jemalloc,
            neon_metrics,
            proxy: crate::metrics::Metrics::get(),
-            tokio: measured_tokio::RuntimeCollector::current(),
        },
    ));

--- a/proxy/src/binary/proxy.rs
+++ b/proxy/src/binary/proxy.rs
@@ -513,7 +513,6 @@ pub async fn run() -> anyhow::Result<()> {
            jemalloc,
            neon_metrics,
            proxy: crate::metrics::Metrics::get(),
-            tokio: measured_tokio::RuntimeCollector::current(),
        },
    ));
    maintenance_tasks.spawn(control_plane::mgmt::task_main(mgmt_listener));
--- a/proxy/src/cache/project_info.rs
+++ b/proxy/src/cache/project_info.rs
@@ -194,7 +194,7 @@ impl ProjectInfoCacheImpl {
        &self,
        endpoint_id: &EndpointId,
    ) -> Option<Ref<'_, EndpointIdInt, EndpointInfo>> {
-        let endpoint_id = EndpointIdInt::try_new_existing(endpoint_id)?;
+        let endpoint_id = EndpointIdInt::get(endpoint_id)?;
        self.cache.get(&endpoint_id)
    }

@@ -204,7 +204,7 @@ impl ProjectInfoCacheImpl {
        role_name: &RoleName,
    ) -> Option<RoleAccessControl> {
        let valid_since = self.get_cache_times();
-        let role_name = RoleNameInt::try_new_existing(role_name)?;
+        let role_name = RoleNameInt::get(role_name)?;
        let endpoint_info = self.get_endpoint_cache(endpoint_id)?;
        endpoint_info.get_role_secret(role_name, valid_since)
    }
@@ -297,10 +297,10 @@ impl ProjectInfoCacheImpl {
    }

    pub fn maybe_invalidate_role_secret(&self, endpoint_id: &EndpointId, role_name: &RoleName) {
-        let Some(endpoint_id) = EndpointIdInt::try_new_existing(endpoint_id) else {
+        let Some(endpoint_id) = EndpointIdInt::get(endpoint_id) else {
            return;
        };
-        let Some(role_name) = RoleNameInt::try_new_existing(role_name) else {
+        let Some(role_name) = RoleNameInt::get(role_name) else {
            return;
        };

--- a/proxy/src/context/parquet.rs
+++ b/proxy/src/context/parquet.rs
@@ -267,7 +267,7 @@ async fn worker_inner(
 ) -> anyhow::Result<()> {
    #[cfg(any(test, feature = "testing"))]
    let storage = if config.test_remote_failures > 0 {
-        GenericRemoteStorage::unreliable_wrapper(storage, config.test_remote_failures, 100)
+        GenericRemoteStorage::unreliable_wrapper(storage, config.test_remote_failures)
    } else {
        storage
    };
--- a/proxy/src/http/health_server.rs
+++ b/proxy/src/http/health_server.rs
@@ -71,8 +71,6 @@ pub struct AppMetrics {
    pub neon_metrics: NeonMetrics,
    #[metric(flatten)]
    pub proxy: &'static crate::metrics::Metrics,
-    #[metric(namespace = "tokio")]
-    pub tokio: measured_tokio::RuntimeCollector,
 }

 async fn prometheus_metrics_handler(
--- a/proxy/src/intern.rs
+++ b/proxy/src/intern.rs
@@ -1,59 +1,272 @@
-use paracord::custom_key;
+use std::hash::BuildHasherDefault;
+use std::marker::PhantomData;
+use std::num::NonZeroUsize;
+use std::ops::Index;
+use std::sync::OnceLock;
+
+use lasso::{Capacity, MemoryLimits, Spur, ThreadedRodeo};
+use rustc_hash::FxHasher;

 use crate::types::{AccountId, BranchId, EndpointId, ProjectId, RoleName};

-custom_key!(pub struct RoleNameInt);
-custom_key!(pub struct EndpointIdInt);
-custom_key!(pub struct BranchIdInt);
-custom_key!(pub struct ProjectIdInt);
-custom_key!(pub struct AccountIdInt);
+pub trait InternId: Sized + 'static {
+    fn get_interner() -> &'static StringInterner<Self>;
+}

-impl From<&RoleName> for RoleNameInt {
-    fn from(value: &RoleName) -> Self {
-        RoleNameInt::new(value)
+pub struct StringInterner<Id> {
+    inner: ThreadedRodeo<Spur, BuildHasherDefault<FxHasher>>,
+    _id: PhantomData<Id>,
+}
+
+#[derive(PartialEq, Debug, Clone, Copy, Eq, Hash)]
+pub struct InternedString<Id> {
+    inner: Spur,
+    _id: PhantomData<Id>,
+}
+
+impl<Id: InternId> std::fmt::Display for InternedString<Id> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        self.as_str().fmt(f)
    }
 }

+impl<Id: InternId> InternedString<Id> {
+    pub(crate) fn as_str(&self) -> &'static str {
+        Id::get_interner().inner.resolve(&self.inner)
+    }
+    pub(crate) fn get(s: &str) -> Option<Self> {
+        Id::get_interner().get(s)
+    }
+}
+
+impl<Id: InternId> AsRef<str> for InternedString<Id> {
+    fn as_ref(&self) -> &str {
+        self.as_str()
+    }
+}
+
+impl<Id: InternId> std::ops::Deref for InternedString<Id> {
+    type Target = str;
+    fn deref(&self) -> &str {
+        self.as_str()
+    }
+}
+
+impl<'de, Id: InternId> serde::de::Deserialize<'de> for InternedString<Id> {
+    fn deserialize<D: serde::de::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
+        struct Visitor<Id>(PhantomData<Id>);
+        impl<Id: InternId> serde::de::Visitor<'_> for Visitor<Id> {
+            type Value = InternedString<Id>;
+
+            fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+                formatter.write_str("a string")
+            }
+
+            fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                Ok(Id::get_interner().get_or_intern(v))
+            }
+        }
+        d.deserialize_str(Visitor::<Id>(PhantomData))
+    }
+}
+
+impl<Id: InternId> serde::Serialize for InternedString<Id> {
+    fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
+        self.as_str().serialize(s)
+    }
+}
+
+impl<Id: InternId> StringInterner<Id> {
+    pub(crate) fn new() -> Self {
+        StringInterner {
+            inner: ThreadedRodeo::with_capacity_memory_limits_and_hasher(
+                Capacity::new(2500, NonZeroUsize::new(1 << 16).expect("value is nonzero")),
+                // unbounded
+                MemoryLimits::for_memory_usage(usize::MAX),
+                BuildHasherDefault::<FxHasher>::default(),
+            ),
+            _id: PhantomData,
+        }
+    }
+
+    #[cfg(test)]
+    fn len(&self) -> usize {
+        self.inner.len()
+    }
+
+    #[cfg(test)]
+    fn current_memory_usage(&self) -> usize {
+        self.inner.current_memory_usage()
+    }
+
+    pub(crate) fn get_or_intern(&self, s: &str) -> InternedString<Id> {
+        InternedString {
+            inner: self.inner.get_or_intern(s),
+            _id: PhantomData,
+        }
+    }
+
+    pub(crate) fn get(&self, s: &str) -> Option<InternedString<Id>> {
+        Some(InternedString {
+            inner: self.inner.get(s)?,
+            _id: PhantomData,
+        })
+    }
+}
+
+impl<Id: InternId> Index<InternedString<Id>> for StringInterner<Id> {
+    type Output = str;
+
+    fn index(&self, index: InternedString<Id>) -> &Self::Output {
+        self.inner.resolve(&index.inner)
+    }
+}
+
+impl<Id: InternId> Default for StringInterner<Id> {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
+pub struct RoleNameTag;
+impl InternId for RoleNameTag {
+    fn get_interner() -> &'static StringInterner<Self> {
+        static ROLE_NAMES: OnceLock<StringInterner<RoleNameTag>> = OnceLock::new();
+        ROLE_NAMES.get_or_init(Default::default)
+    }
+}
+pub type RoleNameInt = InternedString<RoleNameTag>;
+impl From<&RoleName> for RoleNameInt {
+    fn from(value: &RoleName) -> Self {
+        RoleNameTag::get_interner().get_or_intern(value)
+    }
+}
+
+#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
+pub struct EndpointIdTag;
+impl InternId for EndpointIdTag {
+    fn get_interner() -> &'static StringInterner<Self> {
+        static ROLE_NAMES: OnceLock<StringInterner<EndpointIdTag>> = OnceLock::new();
+        ROLE_NAMES.get_or_init(Default::default)
+    }
+}
+pub type EndpointIdInt = InternedString<EndpointIdTag>;
 impl From<&EndpointId> for EndpointIdInt {
    fn from(value: &EndpointId) -> Self {
-        EndpointIdInt::new(value)
+        EndpointIdTag::get_interner().get_or_intern(value)
    }
 }
 impl From<EndpointId> for EndpointIdInt {
    fn from(value: EndpointId) -> Self {
-        EndpointIdInt::new(&value)
+        EndpointIdTag::get_interner().get_or_intern(&value)
    }
 }

+#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
+pub struct BranchIdTag;
+impl InternId for BranchIdTag {
+    fn get_interner() -> &'static StringInterner<Self> {
+        static ROLE_NAMES: OnceLock<StringInterner<BranchIdTag>> = OnceLock::new();
+        ROLE_NAMES.get_or_init(Default::default)
+    }
+}
+pub type BranchIdInt = InternedString<BranchIdTag>;
 impl From<&BranchId> for BranchIdInt {
    fn from(value: &BranchId) -> Self {
-        BranchIdInt::new(value)
+        BranchIdTag::get_interner().get_or_intern(value)
    }
 }
 impl From<BranchId> for BranchIdInt {
    fn from(value: BranchId) -> Self {
-        BranchIdInt::new(&value)
+        BranchIdTag::get_interner().get_or_intern(&value)
    }
 }

+#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
+pub struct ProjectIdTag;
+impl InternId for ProjectIdTag {
+    fn get_interner() -> &'static StringInterner<Self> {
+        static ROLE_NAMES: OnceLock<StringInterner<ProjectIdTag>> = OnceLock::new();
+        ROLE_NAMES.get_or_init(Default::default)
+    }
+}
+pub type ProjectIdInt = InternedString<ProjectIdTag>;
 impl From<&ProjectId> for ProjectIdInt {
    fn from(value: &ProjectId) -> Self {
-        ProjectIdInt::new(value)
+        ProjectIdTag::get_interner().get_or_intern(value)
    }
 }
 impl From<ProjectId> for ProjectIdInt {
    fn from(value: ProjectId) -> Self {
-        ProjectIdInt::new(&value)
+        ProjectIdTag::get_interner().get_or_intern(&value)
    }
 }

+#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
+pub struct AccountIdTag;
+impl InternId for AccountIdTag {
+    fn get_interner() -> &'static StringInterner<Self> {
+        static ROLE_NAMES: OnceLock<StringInterner<AccountIdTag>> = OnceLock::new();
+        ROLE_NAMES.get_or_init(Default::default)
+    }
+}
+pub type AccountIdInt = InternedString<AccountIdTag>;
 impl From<&AccountId> for AccountIdInt {
    fn from(value: &AccountId) -> Self {
-        AccountIdInt::new(value)
+        AccountIdTag::get_interner().get_or_intern(value)
    }
 }
 impl From<AccountId> for AccountIdInt {
    fn from(value: AccountId) -> Self {
-        AccountIdInt::new(&value)
+        AccountIdTag::get_interner().get_or_intern(&value)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::OnceLock;
+
+    use super::InternId;
+    use crate::intern::StringInterner;
+
+    struct MyId;
+    impl InternId for MyId {
+        fn get_interner() -> &'static StringInterner<Self> {
+            pub(crate) static ROLE_NAMES: OnceLock<StringInterner<MyId>> = OnceLock::new();
+            ROLE_NAMES.get_or_init(Default::default)
+        }
+    }
+
+    #[test]
+    fn push_many_strings() {
+        use rand::rngs::StdRng;
+        use rand::{Rng, SeedableRng};
+        use rand_distr::Zipf;
+
+        let endpoint_dist = Zipf::new(500000, 0.8).unwrap();
+        let endpoints = StdRng::seed_from_u64(272488357).sample_iter(endpoint_dist);
+
+        let interner = MyId::get_interner();
+
+        const N: usize = 100_000;
+        let mut verify = Vec::with_capacity(N);
+        for endpoint in endpoints.take(N) {
+            let endpoint = format!("ep-string-interning-{endpoint}");
+            let key = interner.get_or_intern(&endpoint);
+            verify.push((endpoint, key));
+        }
+
+        for (s, key) in verify {
+            assert_eq!(interner[key], s);
+        }
+
+        // 2031616/59861 = 34 bytes per string
+        assert_eq!(interner.len(), 59_861);
+        // will have other overhead for the internal hashmaps that are not accounted for.
+        assert_eq!(interner.current_memory_usage(), 2_031_616);
    }
 }
--- a/proxy/src/metrics.rs
+++ b/proxy/src/metrics.rs
@@ -1,5 +1,6 @@
 use std::sync::{Arc, OnceLock};

+use lasso::ThreadedRodeo;
 use measured::label::{
    FixedCardinalitySet, LabelGroupSet, LabelName, LabelSet, LabelValue, StaticLabelSet,
 };
@@ -10,7 +11,6 @@ use measured::{
    MetricGroup,
 };
 use metrics::{CounterPairAssoc, CounterPairVec, HyperLogLogVec};
-use paracord::ParaCord;
 use tokio::time::{self, Instant};

 use crate::control_plane::messages::ColdStartInfo;
@@ -222,7 +222,7 @@ pub enum CacheOutcome {
 #[derive(LabelGroup)]
 #[label(set = ConsoleRequestSet)]
 pub struct ConsoleRequest<'a> {
-    #[label(dynamic_with = ParaCord, default)]
+    #[label(dynamic_with = ThreadedRodeo, default)]
    pub request: &'a str,
 }

@@ -345,7 +345,7 @@ pub struct ConnectionFailuresBreakdownGroup {
 #[derive(LabelGroup, Copy, Clone)]
 #[label(set = RedisErrorsSet)]
 pub struct RedisErrors<'a> {
-    #[label(dynamic_with = ParaCord, default)]
+    #[label(dynamic_with = ThreadedRodeo, default)]
    pub channel: &'a str,
 }

--- a/proxy/src/types.rs
+++ b/proxy/src/types.rs
@@ -1,4 +1,4 @@
-use crate::intern::EndpointIdInt;
+use crate::intern::{EndpointIdInt, EndpointIdTag, InternId};

 macro_rules! smol_str_wrapper {
    ($name:ident) => {
@@ -85,7 +85,7 @@ impl EndpointId {

    #[must_use]
    pub fn normalize_intern(&self) -> EndpointIdInt {
-        EndpointIdInt::new(self.normalize_str())
+        EndpointIdTag::get_interner().get_or_intern(self.normalize_str())
    }
 }

--- a/safekeeper/src/bin/safekeeper.rs
+++ b/safekeeper/src/bin/safekeeper.rs
@@ -37,7 +37,6 @@ use tracing::*;
 use utils::auth::{JwtAuth, Scope, SwappableJwtAuth};
 use utils::id::NodeId;
 use utils::logging::{self, LogFormat, SecretString};
-use utils::metrics_collector::{METRICS_COLLECTION_INTERVAL, METRICS_COLLECTOR};
 use utils::sentry_init::init_sentry;
 use utils::{pid_file, project_build_tag, project_git_version, tcp_listener};

@@ -244,11 +243,6 @@ struct Args {
    #[arg(long)]
    enable_tls_wal_service_api: bool,

-    /// Controls whether to collect all metrics on each scrape or to return potentially stale
-    /// results.
-    #[arg(long, default_value_t = true)]
-    force_metric_collection_on_scrape: bool,
-
    /// Run in development mode (disables security checks)
    #[arg(long, help = "Run in development mode (disables security checks)")]
    dev: bool,
@@ -434,7 +428,6 @@ async fn main() -> anyhow::Result<()> {
        ssl_ca_certs,
        use_https_safekeeper_api: args.use_https_safekeeper_api,
        enable_tls_wal_service_api: args.enable_tls_wal_service_api,
-        force_metric_collection_on_scrape: args.force_metric_collection_on_scrape,
    });

    // initialize sentry if SENTRY_DSN is provided
@@ -647,26 +640,6 @@ async fn start_safekeeper(conf: Arc<SafeKeeperConf>) -> Result<()> {
        .map(|res| ("broker main".to_owned(), res));
    tasks_handles.push(Box::pin(broker_task_handle));

-    /* BEGIN_HADRON */
-    if conf.force_metric_collection_on_scrape {
-        let metrics_handle = current_thread_rt
-            .as_ref()
-            .unwrap_or_else(|| BACKGROUND_RUNTIME.handle())
-            .spawn(async move {
-                let mut interval: tokio::time::Interval =
-                    tokio::time::interval(METRICS_COLLECTION_INTERVAL);
-                loop {
-                    interval.tick().await;
-                    tokio::task::spawn_blocking(|| {
-                        METRICS_COLLECTOR.run_once(true);
-                    });
-                }
-            })
-            .map(|res| ("broker main".to_owned(), res));
-        tasks_handles.push(Box::pin(metrics_handle));
-    }
-    /* END_HADRON */
-
    set_build_info_metric(GIT_VERSION, BUILD_TAG);

    // TODO: update tokio-stream, convert to real async Stream with
--- a/safekeeper/src/http/routes.rs
+++ b/safekeeper/src/http/routes.rs
@@ -699,11 +699,6 @@ pub fn make_router(
        }))
    }

-    let force_metric_collection_on_scrape = conf.force_metric_collection_on_scrape;
-
-    let prometheus_metrics_handler_wrapper =
-        move |req| prometheus_metrics_handler(req, force_metric_collection_on_scrape);
-
    // NB: on any changes do not forget to update the OpenAPI spec
    // located nearby (/safekeeper/src/http/openapi_spec.yaml).
    let auth = conf.http_auth.clone();
@@ -711,9 +706,7 @@ pub fn make_router(
        .data(conf)
        .data(global_timelines)
        .data(auth)
-        .get("/metrics", move |r| {
-            request_span(r, prometheus_metrics_handler_wrapper)
-        })
+        .get("/metrics", |r| request_span(r, prometheus_metrics_handler))
        .get("/profile/cpu", |r| request_span(r, profile_cpu_handler))
        .get("/profile/heap", |r| request_span(r, profile_heap_handler))
        .get("/v1/status", |r| request_span(r, status_handler))
--- a/safekeeper/src/lib.rs
+++ b/safekeeper/src/lib.rs
@@ -134,7 +134,6 @@ pub struct SafeKeeperConf {
    pub ssl_ca_certs: Vec<Pem>,
    pub use_https_safekeeper_api: bool,
    pub enable_tls_wal_service_api: bool,
-    pub force_metric_collection_on_scrape: bool,
 }

 impl SafeKeeperConf {
@@ -184,7 +183,6 @@ impl SafeKeeperConf {
            ssl_ca_certs: Vec::new(),
            use_https_safekeeper_api: false,
            enable_tls_wal_service_api: false,
-            force_metric_collection_on_scrape: true,
        }
    }
 }
--- a/safekeeper/src/metrics.rs
+++ b/safekeeper/src/metrics.rs
@@ -59,15 +59,6 @@ pub static FLUSH_WAL_SECONDS: Lazy<Histogram> = Lazy::new(|| {
    .expect("Failed to register safekeeper_flush_wal_seconds histogram")
 });
 /* BEGIN_HADRON */
-// Counter of all ProposerAcceptorMessage requests received
-pub static PROPOSER_ACCEPTOR_MESSAGES_TOTAL: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "safekeeper_proposer_acceptor_messages_total",
-        "Total number of ProposerAcceptorMessage requests received by the Safekeeper.",
-        &["outcome"]
-    )
-    .expect("Failed to register safekeeper_proposer_acceptor_messages_total counter")
-});
 pub static WAL_DISK_IO_ERRORS: Lazy<IntCounter> = Lazy::new(|| {
    register_int_counter!(
        "safekeeper_wal_disk_io_errors",
--- a/safekeeper/src/safekeeper.rs
+++ b/safekeeper/src/safekeeper.rs
@@ -24,7 +24,7 @@ use utils::id::{NodeId, TenantId, TimelineId};
 use utils::lsn::Lsn;
 use utils::pageserver_feedback::PageserverFeedback;

-use crate::metrics::{MISC_OPERATION_SECONDS, PROPOSER_ACCEPTOR_MESSAGES_TOTAL};
+use crate::metrics::MISC_OPERATION_SECONDS;
 use crate::state::TimelineState;
 use crate::{control_file, wal_storage};

@@ -938,7 +938,7 @@ where
        &mut self,
        msg: &ProposerAcceptorMessage,
    ) -> Result<Option<AcceptorProposerMessage>> {
-        let res = match msg {
+        match msg {
            ProposerAcceptorMessage::Greeting(msg) => self.handle_greeting(msg).await,
            ProposerAcceptorMessage::VoteRequest(msg) => self.handle_vote_request(msg).await,
            ProposerAcceptorMessage::Elected(msg) => self.handle_elected(msg).await,
@@ -949,20 +949,7 @@ where
                self.handle_append_request(msg, false).await
            }
            ProposerAcceptorMessage::FlushWAL => self.handle_flush().await,
-        };
-
-        // BEGIN HADRON
-        match &res {
-            Ok(_) => PROPOSER_ACCEPTOR_MESSAGES_TOTAL
-                .with_label_values(&["success"])
-                .inc(),
-            Err(_) => PROPOSER_ACCEPTOR_MESSAGES_TOTAL
-                .with_label_values(&["error"])
-                .inc(),
-        };
-
-        res
-        // END HADRON
+        }
    }

    /// Handle initial message from proposer: check its sanity and send my
--- a/safekeeper/src/wal_backup.rs
+++ b/safekeeper/src/wal_backup.rs
@@ -166,7 +166,7 @@ fn hadron_determine_offloader(mgr: &Manager, state: &StateSnapshot) -> (Option<N

    let backup_lag = state.commit_lsn.checked_sub(state.backup_lsn);
    if backup_lag.is_none() {
-        debug!("Backup lag is None. Skipping re-election.");
+        info!("Backup lag is None. Skipping re-election.");
        return (offloader, election_dbg_str);
    }

--- a/safekeeper/tests/walproposer_sim/safekeeper.rs
+++ b/safekeeper/tests/walproposer_sim/safekeeper.rs
@@ -190,7 +190,6 @@ pub fn run_server(os: NodeOs, disk: Arc<SafekeeperDisk>) -> Result<()> {
        ssl_ca_certs: Vec::new(),
        use_https_safekeeper_api: false,
        enable_tls_wal_service_api: false,
-        force_metric_collection_on_scrape: true,
    };

    let mut global = GlobalMap::new(disk, conf.clone())?;
--- a/storage_controller/Cargo.toml
+++ b/storage_controller/Cargo.toml
@@ -31,10 +31,10 @@ humantime.workspace = true
 humantime-serde.workspace = true
 itertools.workspace = true
 json-structural-diff.workspace = true
+lasso.workspace = true
 once_cell.workspace = true
 pageserver_api.workspace = true
 pageserver_client.workspace = true
-paracord.workspace = true
 postgres_connection.workspace = true
 posthog_client_lite.workspace = true
 rand.workspace = true
@@ -72,4 +72,4 @@ http-utils = { path = "../libs/http-utils/" }
 utils = { path = "../libs/utils/" }
 metrics = { path = "../libs/metrics/" }
 control_plane = { path = "../control_plane" }
-workspace_hack = { version = "0.1", path = "../workspace_hack" }
+workspace_hack = { version = "0.1", path = "../workspace_hack" }
--- a/storage_controller/src/http.rs
+++ b/storage_controller/src/http.rs
@@ -850,31 +850,6 @@ async fn handle_tenant_describe(
    json_response(StatusCode::OK, service.tenant_describe(tenant_id)?)
 }

-/* BEGIN_HADRON */
-async fn handle_tenant_timeline_describe(
-    service: Arc<Service>,
-    req: Request<Body>,
-) -> Result<Response<Body>, ApiError> {
-    check_permissions(&req, Scope::Scrubber)?;
-
-    let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
-    let timeline_id: TimelineId = parse_request_param(&req, "timeline_id")?;
-    match maybe_forward(req).await {
-        ForwardOutcome::Forwarded(res) => {
-            return res;
-        }
-        ForwardOutcome::NotForwarded(_req) => {}
-    };
-
-    json_response(
-        StatusCode::OK,
-        service
-            .tenant_timeline_describe(tenant_id, timeline_id)
-            .await?,
-    )
-}
-/* END_HADRON */
-
 async fn handle_tenant_list(
    service: Arc<Service>,
    req: Request<Body>,
@@ -2505,13 +2480,6 @@ pub fn make_router(
            )
        })
        // Timeline operations
-        .get("/control/v1/tenant/:tenant_id/timeline/:timeline_id", |r| {
-            tenant_service_handler(
-                r,
-                handle_tenant_timeline_describe,
-                RequestName("v1_tenant_timeline_describe"),
-            )
-        })
        .delete("/v1/tenant/:tenant_id/timeline/:timeline_id", |r| {
            tenant_service_handler(
                r,
--- a/storage_controller/src/metrics.rs
+++ b/storage_controller/src/metrics.rs
@@ -183,9 +183,9 @@ impl Default for StorageControllerMetrics {
 #[derive(measured::LabelGroup, Clone)]
 #[label(set = NodeLabelGroupSet)]
 pub(crate) struct NodeLabelGroup<'a> {
-    #[label(dynamic_with = paracord::ParaCord, default)]
+    #[label(dynamic_with = lasso::ThreadedRodeo, default)]
    pub(crate) az: &'a str,
-    #[label(dynamic_with = paracord::ParaCord, default)]
+    #[label(dynamic_with = lasso::ThreadedRodeo, default)]
    pub(crate) node_id: &'a str,
 }

@@ -198,7 +198,7 @@ pub(crate) struct ReconcileCompleteLabelGroup {
 #[derive(measured::LabelGroup)]
 #[label(set = HttpRequestStatusLabelGroupSet)]
 pub(crate) struct HttpRequestStatusLabelGroup<'a> {
-    #[label(dynamic_with = paracord::ParaCord, default)]
+    #[label(dynamic_with = lasso::ThreadedRodeo, default)]
    pub(crate) path: &'a str,
    pub(crate) method: Method,
    pub(crate) status: StatusCode,
@@ -207,7 +207,7 @@ pub(crate) struct HttpRequestStatusLabelGroup<'a> {
 #[derive(measured::LabelGroup)]
 #[label(set = HttpRequestLatencyLabelGroupSet)]
 pub(crate) struct HttpRequestLatencyLabelGroup<'a> {
-    #[label(dynamic_with = paracord::ParaCord, default)]
+    #[label(dynamic_with = lasso::ThreadedRodeo, default)]
    pub(crate) path: &'a str,
    pub(crate) method: Method,
 }
@@ -215,9 +215,9 @@ pub(crate) struct HttpRequestLatencyLabelGroup<'a> {
 #[derive(measured::LabelGroup, Clone)]
 #[label(set = PageserverRequestLabelGroupSet)]
 pub(crate) struct PageserverRequestLabelGroup<'a> {
-    #[label(dynamic_with = paracord::ParaCord, default)]
+    #[label(dynamic_with = lasso::ThreadedRodeo, default)]
    pub(crate) pageserver_id: &'a str,
-    #[label(dynamic_with = paracord::ParaCord, default)]
+    #[label(dynamic_with = lasso::ThreadedRodeo, default)]
    pub(crate) path: &'a str,
    pub(crate) method: Method,
 }
@@ -225,9 +225,9 @@ pub(crate) struct PageserverRequestLabelGroup<'a> {
 #[derive(measured::LabelGroup, Clone)]
 #[label(set = SafekeeperRequestLabelGroupSet)]
 pub(crate) struct SafekeeperRequestLabelGroup<'a> {
-    #[label(dynamic_with = paracord::ParaCord, default)]
+    #[label(dynamic_with = lasso::ThreadedRodeo, default)]
    pub(crate) safekeeper_id: &'a str,
-    #[label(dynamic_with = paracord::ParaCord, default)]
+    #[label(dynamic_with = lasso::ThreadedRodeo, default)]
    pub(crate) path: &'a str,
    pub(crate) method: Method,
 }
@@ -254,11 +254,11 @@ pub(crate) struct LeadershipStatusGroup {
 #[derive(measured::LabelGroup, Clone)]
 #[label(set = ReconcileLongRunningLabelGroupSet)]
 pub(crate) struct ReconcileLongRunningLabelGroup<'a> {
-    #[label(dynamic_with = paracord::ParaCord, default)]
+    #[label(dynamic_with = lasso::ThreadedRodeo, default)]
    pub(crate) tenant_id: &'a str,
-    #[label(dynamic_with = paracord::ParaCord, default)]
+    #[label(dynamic_with = lasso::ThreadedRodeo, default)]
    pub(crate) shard_number: &'a str,
-    #[label(dynamic_with = paracord::ParaCord, default)]
+    #[label(dynamic_with = lasso::ThreadedRodeo, default)]
    pub(crate) sequence: &'a str,
 }

@@ -282,11 +282,11 @@ pub(crate) enum Method {
 #[derive(measured::LabelGroup, Clone)]
 #[label(set = SafekeeperReconcilerLabelGroupSet)]
 pub(crate) struct SafekeeperReconcilerLabelGroup<'a> {
-    #[label(dynamic_with = paracord::ParaCord, default)]
+    #[label(dynamic_with = lasso::ThreadedRodeo, default)]
    pub(crate) sk_az: &'a str,
-    #[label(dynamic_with = paracord::ParaCord, default)]
+    #[label(dynamic_with = lasso::ThreadedRodeo, default)]
    pub(crate) sk_node_id: &'a str,
-    #[label(dynamic_with = paracord::ParaCord, default)]
+    #[label(dynamic_with = lasso::ThreadedRodeo, default)]
    pub(crate) sk_hostname: &'a str,
 }

--- a/storage_controller/src/pageserver_client.rs
+++ b/storage_controller/src/pageserver_client.rs
@@ -86,23 +86,6 @@ impl PageserverClient {
        )
    }

-    /* BEGIN_HADRON */
-    pub(crate) async fn tenant_timeline_describe(
-        &self,
-        tenant_shard_id: &TenantShardId,
-        timeline_id: &TimelineId,
-    ) -> Result<TimelineInfo> {
-        measured_request!(
-            "tenant_timeline_describe",
-            crate::metrics::Method::Get,
-            &self.node_id_label,
-            self.inner
-                .tenant_timeline_describe(tenant_shard_id, timeline_id,)
-                .await
-        )
-    }
-    /* END_HADRON */
-
    pub(crate) async fn tenant_scan_remote_storage(
        &self,
        tenant_id: TenantId,
--- a/storage_controller/src/service.rs
+++ b/storage_controller/src/service.rs
@@ -32,7 +32,7 @@ use pageserver_api::controller_api::{
    ShardSchedulingPolicy, ShardsPreferredAzsRequest, ShardsPreferredAzsResponse,
    SkSchedulingPolicy, TenantCreateRequest, TenantCreateResponse, TenantCreateResponseShard,
    TenantDescribeResponse, TenantDescribeResponseShard, TenantLocateResponse, TenantPolicyRequest,
-    TenantShardMigrateRequest, TenantShardMigrateResponse, TenantTimelineDescribeResponse,
+    TenantShardMigrateRequest, TenantShardMigrateResponse,
 };
 use pageserver_api::models::{
    self, DetachBehavior, LocationConfig, LocationConfigListResponse, LocationConfigMode, LsnLease,
@@ -1984,14 +1984,11 @@ impl Service {
        });

        // Check that there is enough safekeepers configured that we can create new timelines
-        let test_sk_res_str = match this.safekeepers_for_new_timeline().await {
-            Ok(v) => format!("Ok({v:?})"),
-            Err(v) => format!("Err({v:})"),
-        };
+        let test_sk_res = this.safekeepers_for_new_timeline().await;
        tracing::info!(
            timeline_safekeeper_count = config.timeline_safekeeper_count,
            timelines_onto_safekeepers = config.timelines_onto_safekeepers,
-            "viability test result (test timeline creation on safekeepers): {test_sk_res_str}",
+            "viability test result (test timeline creation on safekeepers): {test_sk_res:?}",
        );

        Ok(this)
@@ -4431,7 +4428,7 @@ impl Service {
                .await;

            let mut failed = 0;
-            for (tid, (_, result)) in targeted_tenant_shards.iter().zip(results.into_iter()) {
+            for (tid, result) in targeted_tenant_shards.iter().zip(results.into_iter()) {
                match result {
                    Ok(ok) => {
                        if tid.is_shard_zero() {
@@ -4761,7 +4758,6 @@ impl Service {
        )
        .await;

-        let mut retry_if_not_attached = false;
        let targets = {
            let locked = self.inner.read().unwrap();
            let mut targets = Vec::new();
@@ -4778,24 +4774,6 @@ impl Service {
                        .expect("Pageservers may not be deleted while referenced");

                    targets.push((*tenant_shard_id, node.clone()));
-
-                    if let Some(location) = shard.observed.locations.get(node_id) {
-                        if let Some(ref conf) = location.conf {
-                            if conf.mode != LocationConfigMode::AttachedSingle
-                                && conf.mode != LocationConfigMode::AttachedMulti
-                            {
-                                // If the shard is attached as secondary, we need to retry if 404.
-                                retry_if_not_attached = true;
-                            }
-                            // If the shard is attached as primary, we should succeed.
-                        } else {
-                            // Location conf is not available yet, retry if 404.
-                            retry_if_not_attached = true;
-                        }
-                    } else {
-                        // The shard is not attached to the intended pageserver yet, retry if 404.
-                        retry_if_not_attached = true;
-                    }
                }
            }
            targets
@@ -4817,7 +4795,7 @@ impl Service {
            .await;

        let mut valid_until = None;
-        for (node, r) in res {
+        for r in res {
            match r {
                Ok(lease) => {
                    if let Some(ref mut valid_until) = valid_until {
@@ -4826,20 +4804,8 @@ impl Service {
                        valid_until = Some(lease.valid_until);
                    }
                }
-                Err(mgmt_api::Error::ApiError(StatusCode::NOT_FOUND, _))
-                    if retry_if_not_attached =>
-                {
-                    // This is expected if the attach is not finished yet. Return 503 so that the client can retry.
-                    return Err(ApiError::ResourceUnavailable(
-                        format!(
-                            "Timeline is not attached to the pageserver {} yet, please retry",
-                            node.get_id()
-                        )
-                        .into(),
-                    ));
-                }
                Err(e) => {
-                    return Err(passthrough_api_error(&node, e));
+                    return Err(ApiError::InternalServerError(anyhow::anyhow!(e)));
                }
            }
        }
@@ -4953,7 +4919,7 @@ impl Service {
        max_retries: u32,
        timeout: Duration,
        cancel: &CancellationToken,
-    ) -> Vec<(Node, mgmt_api::Result<T>)>
+    ) -> Vec<mgmt_api::Result<T>>
    where
        O: Fn(TenantShardId, PageserverClient) -> F + Copy,
        F: std::future::Future<Output = mgmt_api::Result<T>>,
@@ -4974,16 +4940,16 @@ impl Service {
                        cancel,
                    )
                    .await;
-                (idx, node, r)
+                (idx, r)
            });
        }

-        while let Some((idx, node, r)) = futs.next().await {
-            results.push((idx, node, r.unwrap_or(Err(mgmt_api::Error::Cancelled))));
+        while let Some((idx, r)) = futs.next().await {
+            results.push((idx, r.unwrap_or(Err(mgmt_api::Error::Cancelled))));
        }

-        results.sort_by_key(|(idx, _, _)| *idx);
-        results.into_iter().map(|(_, node, r)| (node, r)).collect()
+        results.sort_by_key(|(idx, _)| *idx);
+        results.into_iter().map(|(_, r)| r).collect()
    }

    /// Helper for safely working with the shards in a tenant remotely on pageservers, for example
@@ -5486,92 +5452,6 @@ impl Service {
        .ok_or_else(|| ApiError::NotFound(anyhow::anyhow!("Tenant {tenant_id} not found").into()))
    }

-    /* BEGIN_HADRON */
-    pub(crate) async fn tenant_timeline_describe(
-        &self,
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-    ) -> Result<TenantTimelineDescribeResponse, ApiError> {
-        self.tenant_remote_mutation(tenant_id, |locations| async move {
-            if locations.0.is_empty() {
-                return Err(ApiError::NotFound(
-                    anyhow::anyhow!("Tenant not found").into(),
-                ));
-            };
-
-            let locations: Vec<(TenantShardId, Node)> = locations
-                .0
-                .iter()
-                .map(|t| (*t.0, t.1.latest.node.clone()))
-                .collect();
-            let mut futs = FuturesUnordered::new();
-
-            for (shard_id, node) in locations {
-                futs.push({
-                    async move {
-                        let result = node
-                            .with_client_retries(
-                                |client| async move {
-                                    client
-                                        .tenant_timeline_describe(&shard_id, &timeline_id)
-                                        .await
-                                },
-                                &self.http_client,
-                                &self.config.pageserver_jwt_token,
-                                3,
-                                3,
-                                Duration::from_secs(30),
-                                &self.cancel,
-                            )
-                            .await;
-                        (result, shard_id, node.get_id())
-                    }
-                });
-            }
-
-            let mut results: Vec<TimelineInfo> = Vec::new();
-            while let Some((result, tenant_shard_id, node_id)) = futs.next().await {
-                match result {
-                    Some(Ok(timeline_info)) => results.push(timeline_info),
-                    Some(Err(e)) => {
-                        tracing::warn!(
-                            "Failed to describe tenant {} timeline {} for pageserver {}: {e}",
-                            tenant_shard_id,
-                            timeline_id,
-                            node_id,
-                        );
-                        return Err(ApiError::ResourceUnavailable(format!("{e}").into()));
-                    }
-                    None => return Err(ApiError::Cancelled),
-                }
-            }
-            let mut image_consistent_lsn: Option<Lsn> = Some(Lsn::MAX);
-            for timeline_info in &results {
-                if let Some(tline_image_consistent_lsn) = timeline_info.image_consistent_lsn {
-                    image_consistent_lsn = Some(std::cmp::min(
-                        image_consistent_lsn.unwrap(),
-                        tline_image_consistent_lsn,
-                    ));
-                } else {
-                    tracing::warn!(
-                        "Timeline {} on shard {} does not have image consistent lsn",
-                        timeline_info.timeline_id,
-                        timeline_info.tenant_id
-                    );
-                    image_consistent_lsn = None;
-                    break;
-                }
-            }
-
-            Ok(TenantTimelineDescribeResponse {
-                shards: results,
-                image_consistent_lsn,
-            })
-        })
-        .await?
-    }
-    /* END_HADRON */
-
    /// limit & offset are pagination parameters. Since we are walking an in-memory HashMap, `offset` does not
    /// avoid traversing data, it just avoid returning it. This is suitable for our purposes, since our in memory
    /// maps are small enough to traverse fast, our pagination is just to avoid serializing huge JSON responses
@@ -5982,7 +5862,7 @@ impl Service {
            return;
        }

-        for (_, result) in self
+        for result in self
            .tenant_for_shards_api(
                attached,
                |tenant_shard_id, client| async move {
@@ -6001,7 +5881,7 @@ impl Service {
            }
        }

-        for (_, result) in self
+        for result in self
            .tenant_for_shards_api(
                secondary,
                |tenant_shard_id, client| async move {
@@ -8888,7 +8768,7 @@ impl Service {
            )
            .await;

-        for ((tenant_shard_id, node, optimization), (_, secondary_status)) in
+        for ((tenant_shard_id, node, optimization), secondary_status) in
            want_secondary_status.into_iter().zip(results.into_iter())
        {
            match secondary_status {
--- a/storage_controller/src/service/safekeeper_service.rs
+++ b/storage_controller/src/service/safekeeper_service.rs
@@ -25,8 +25,7 @@ use pageserver_api::models::{SafekeeperInfo, SafekeepersInfo, TimelineInfo};
 use safekeeper_api::PgVersionId;
 use safekeeper_api::membership::{self, MemberSet, SafekeeperGeneration};
 use safekeeper_api::models::{
-    PullTimelineRequest, TimelineLocateResponse, TimelineMembershipSwitchRequest,
-    TimelineMembershipSwitchResponse,
+    PullTimelineRequest, TimelineMembershipSwitchRequest, TimelineMembershipSwitchResponse,
 };
 use safekeeper_api::{INITIAL_TERM, Term};
 use safekeeper_client::mgmt_api;
@@ -38,6 +37,13 @@ use utils::lsn::Lsn;

 use super::Service;

+#[derive(serde::Serialize, serde::Deserialize, Clone)]
+pub struct TimelineLocateResponse {
+    pub generation: SafekeeperGeneration,
+    pub sk_set: Vec<NodeId>,
+    pub new_sk_set: Option<Vec<NodeId>>,
+}
+
 impl Service {
    fn make_member_set(safekeepers: &[Safekeeper]) -> Result<MemberSet, ApiError> {
        let members = safekeepers
--- a/test_runner/fixtures/neon_fixtures.py
+++ b/test_runner/fixtures/neon_fixtures.py
@@ -1795,33 +1795,6 @@ def neon_env_builder(
        record_property("preserve_database_files", builder.preserve_database_files)


-@pytest.fixture(scope="function")
-def neon_env_builder_local(
-    neon_env_builder: NeonEnvBuilder,
-    test_output_dir: Path,
-    pg_distrib_dir: Path,
-) -> NeonEnvBuilder:
-    """
-    Fixture to create a Neon environment for test with its own pg_install copy.
-
-    This allows the test to edit the list of available extensions in the
-    local instance of Postgres used for the test, and install extensions via
-    downloading them when a remote extension is tested, for instance, or
-    copying files around for local extension testing.
-    """
-    test_local_pginstall = test_output_dir / "pg_install"
-    log.info(f"copy {pg_distrib_dir} to {test_local_pginstall}")
-
-    # We can't copy only the version that we are currently testing because other
-    # binaries like the storage controller need specific Postgres versions.
-    shutil.copytree(pg_distrib_dir, test_local_pginstall)
-
-    neon_env_builder.pg_distrib_dir = test_local_pginstall
-    log.info(f"local neon_env_builder.pg_distrib_dir: {neon_env_builder.pg_distrib_dir}")
-
-    return neon_env_builder
-
-
@dataclass
 class PageserverPort:
    pg: int
@@ -2342,20 +2315,6 @@ class NeonStorageController(MetricsGetter, LogUtils):
        response.raise_for_status()
        return response.json()

-    # HADRON
-    def tenant_timeline_describe(
-        self,
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-    ):
-        response = self.request(
-            "GET",
-            f"{self.api}/control/v1/tenant/{tenant_id}/timeline/{timeline_id}",
-            headers=self.headers(TokenScope.ADMIN),
-        )
-        response.raise_for_status()
-        return response.json()
-
    def nodes(self):
        """
        :return: list of {"id": ""}
--- a/test_runner/fixtures/pageserver/http.py
+++ b/test_runner/fixtures/pageserver/http.py
@@ -333,13 +333,6 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
        res = self.post(f"http://localhost:{self.port}/v1/reload_auth_validation_keys")
        self.verbose_error(res)

-    def list_tenant_visible_size(self) -> dict[TenantShardId, int]:
-        res = self.get(f"http://localhost:{self.port}/v1/list_tenant_visible_size")
-        self.verbose_error(res)
-        res_json = res.json()
-        assert isinstance(res_json, dict)
-        return res_json
-
    def tenant_list(self) -> list[dict[Any, Any]]:
        res = self.get(f"http://localhost:{self.port}/v1/tenant")
        self.verbose_error(res)
@@ -1009,7 +1002,7 @@ class PageserverHttpClient(requests.Session, MetricsGetter):

    def get_metrics_str(self) -> str:
        """You probably want to use get_metrics() instead."""
-        res = self.get(f"http://localhost:{self.port}/metrics?use_latest=true")
+        res = self.get(f"http://localhost:{self.port}/metrics")
        self.verbose_error(res)
        return res.text

@@ -1254,10 +1247,3 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
        )
        self.verbose_error(res)
        return res.json()
-
-    def force_refresh_feature_flag(self, tenant_id: TenantId | TenantShardId):
-        res = self.post(
-            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/force_refresh_feature_flag",
-        )
-        self.verbose_error(res)
-        return res.json()
--- a/test_runner/fixtures/port_distributor.py
+++ b/test_runner/fixtures/port_distributor.py
@@ -3,7 +3,6 @@ from __future__ import annotations
 import re
 import socket
 from contextlib import closing
-from itertools import cycle

 from fixtures.log_helper import log

@@ -35,23 +34,15 @@ def can_bind(host: str, port: int) -> bool:

 class PortDistributor:
    def __init__(self, base_port: int, port_number: int):
-        self.base_port = base_port
-        self.port_number = port_number
-        self.cycle = cycle(range(base_port, base_port + port_number))
+        self.iterator = iter(range(base_port, base_port + port_number))
        self.port_map: dict[int, int] = {}

    def get_port(self) -> int:
-        checked = 0
-        for port in self.cycle:
+        for port in self.iterator:
            if can_bind("localhost", port):
                return port
-            elif checked < self.port_number:
-                checked += 1
-            else:
-                break
-
        raise RuntimeError(
-            f"port range ({self.base_port}..{self.base_port + self.port_number}) configured for test is exhausted, consider enlarging the range"
+            "port range configured for test is exhausted, consider enlarging the range"
        )

    def replace_with_new_port(self, value: int | str) -> int | str:
--- a/test_runner/fixtures/safekeeper/http.py
+++ b/test_runner/fixtures/safekeeper/http.py
@@ -143,7 +143,7 @@ class SafekeeperHttpClient(requests.Session, MetricsGetter):

    def get_metrics_str(self) -> str:
        """You probably want to use get_metrics() instead."""
-        request_result = self.get(f"http://localhost:{self.port}/metrics?use_latest=true")
+        request_result = self.get(f"http://localhost:{self.port}/metrics")
        request_result.raise_for_status()
        return request_result.text

--- a/test_runner/random_ops/test_random_ops.py
+++ b/test_runner/random_ops/test_random_ops.py
@@ -117,9 +117,7 @@ class NeonBranch:
    def create_child_branch(self) -> NeonBranch | None:
        return self.project.create_branch(self.id)

-    def create_ro_endpoint(self) -> NeonEndpoint | None:
-        if not self.project.check_limit_endpoints():
-            return None
+    def create_ro_endpoint(self) -> NeonEndpoint:
        return NeonEndpoint(
            self.project,
            self.neon_api.create_endpoint(self.project_id, self.id, "read_only", {})["endpoint"],
@@ -153,26 +151,11 @@ class NeonBranch:
            return
        self.updated_at = datetime.fromisoformat(res["branch"]["updated_at"])
        parent_id: str = res["branch"]["parent_id"]
-        # XXX Retry get parent details to work around the issue
-        # https://databricks.atlassian.net/browse/LKB-279
-        target_time = datetime.now() + timedelta(seconds=30)
-        while datetime.now() < target_time:
-            try:
-                parent_def = self.neon_api.get_branch_details(self.project_id, parent_id)
-            except HTTPError as he:
-                if he.response.status_code == 404:
-                    log.info("Branch not found, waiting...")
-                    time.sleep(1)
-                else:
-                    raise HTTPError(he) from he
-            else:
-                break
-        else:
-            raise RuntimeError(f"Branch {parent_id} not found")
-
        # Creates an object for the parent branch
        # After the reset operation a new parent branch is created
-        parent = NeonBranch(self.project, parent_def, True)
+        parent = NeonBranch(
+            self.project, self.neon_api.get_branch_details(self.project_id, parent_id), True
+        )
        self.project.branches[parent_id] = parent
        self.parent = parent
        parent.children[self.id] = self
@@ -185,21 +168,29 @@ class NeonBranch:
        source_timestamp: str | None = None,
        preserve_under_name: str | None = None,
    ) -> dict[str, Any] | None:
-        if not self.project.check_limit_branches():
-            return None
        endpoints = [ep for ep in self.endpoints.values() if ep.type == "read_only"]
        # Terminate all the benchmarks running to prevent errors. Errors in benchmark during pgbench are expected
        for ep in endpoints:
            ep.terminate_benchmark()
        self.terminate_benchmark()
-        res: dict[str, Any] = self.neon_api.restore_branch(
-            self.project_id,
-            self.id,
-            source_branch_id,
-            source_lsn,
-            source_timestamp,
-            preserve_under_name,
-        )
+        try:
+            res: dict[str, Any] = self.neon_api.restore_branch(
+                self.project_id,
+                self.id,
+                source_branch_id,
+                source_lsn,
+                source_timestamp,
+                preserve_under_name,
+            )
+        except HTTPError as he:
+            if (
+                he.response.status_code == 422
+                and he.response.json()["code"] == "BRANCHES_LIMIT_EXCEEDED"
+            ):
+                log.info("Branch limit exceeded, skipping")
+                return None
+            else:
+                raise HTTPError(he) from he
        self.project.wait()
        self.start_benchmark()
        for ep in endpoints:
@@ -248,30 +239,19 @@ class NeonProject:
    def delete(self) -> None:
        self.neon_api.delete_project(self.id)

-    def check_limit_branches(self) -> bool:
-        if self.limits["max_branches"] == -1 or len(self.branches) < self.limits["max_branches"]:
-            return True
-        log.info("branch limit exceeded (%s/%s)", len(self.branches), self.limits["max_branches"])
-        return False
-
-    def check_limit_endpoints(self) -> bool:
-        if (
-            self.limits["max_read_only_endpoints"] == -1
-            or self.read_only_endpoints_total < self.limits["max_read_only_endpoints"]
-        ):
-            return True
-        log.info(
-            "Maximum read only endpoint limit exceeded (%s/%s)",
-            self.read_only_endpoints_total,
-            self.limits["max_read_only_endpoints"],
-        )
-        return False
-
    def create_branch(self, parent_id: str | None = None) -> NeonBranch | None:
        self.wait()
-        if not self.check_limit_branches():
-            return None
-        branch_def = self.neon_api.create_branch(self.id, parent_id=parent_id)
+        try:
+            branch_def = self.neon_api.create_branch(self.id, parent_id=parent_id)
+        except HTTPError as he:
+            if (
+                he.response.status_code == 422
+                and he.response.json()["code"] == "BRANCHES_LIMIT_EXCEEDED"
+            ):
+                log.info("Branch limit exceeded, skipping")
+                return None
+            else:
+                raise HTTPError(he) from he
        new_branch = NeonBranch(self, branch_def)
        self.wait()
        return new_branch
@@ -408,9 +388,17 @@ def do_action(project: NeonProject, action: str) -> bool:
    log.info("Action: %s", action)
    if action == "new_branch":
        log.info("Trying to create a new branch")
+        if 0 <= project.limits["max_branches"] <= len(project.branches):
+            log.info(
+                "Maximum branch limit exceeded (%s of %s)",
+                len(project.branches),
+                project.limits["max_branches"],
+            )
+            return False
        parent = project.branches[
            random.choice(list(set(project.branches.keys()) - project.reset_branches))
        ]
+        log.info("Parent: %s", parent)
        child = parent.create_child_branch()
        if child is None:
            return False
@@ -425,11 +413,16 @@ def do_action(project: NeonProject, action: str) -> bool:
            log.info("Leaf branches not found, skipping")
            return False
    elif action == "new_ro_endpoint":
+        if 0 <= project.limits["max_read_only_endpoints"] <= project.read_only_endpoints_total:
+            log.info(
+                "Maximum read only endpoint limit exceeded (%s of %s)",
+                project.read_only_endpoints_total,
+                project.limits["max_read_only_endpoints"],
+            )
+            return False
        ep = random.choice(
            [br for br in project.branches.values() if br.id not in project.reset_branches]
        ).create_ro_endpoint()
-        if ep is None:
-            return False
        log.info("Created the RO endpoint with id %s branch: %s", ep.id, ep.branch.id)
        ep.start_benchmark()
    elif action == "delete_ro_endpoint":
--- a/test_runner/regress/data/test_event_trigger_extension/test_event_trigger_extension--1.0.sql
+++ b/test_runner/regress/data/test_event_trigger_extension/test_event_trigger_extension--1.0.sql
@@ -1,32 +0,0 @@
-\echo Use "CREATE EXTENSION test_event_trigger_extension" to load this file. \quit
-
-CREATE SCHEMA event_trigger;
-
-create sequence if not exists event_trigger.seq_schema_version as int cycle;
-
-create or replace function event_trigger.increment_schema_version()
-    returns event_trigger
-    security definer
-    language plpgsql
-as $$
-begin
-    perform pg_catalog.nextval('event_trigger.seq_schema_version');
-end;
-$$;
-
-create or replace function event_trigger.get_schema_version()
-    returns int
-    security definer
-    language sql
-as $$
-    select last_value from event_trigger.seq_schema_version;
-$$;
-
-- On DDL event, increment the schema version number
-create event trigger event_trigger_watch_ddl
-    on ddl_command_end
-    execute procedure event_trigger.increment_schema_version();
-
-create event trigger event_trigger_watch_drop
-    on sql_drop
-    execute procedure event_trigger.increment_schema_version();
--- a/Show More
+++ b/Show More
				`@@ -1 +0,0 @@`
				`GRANT pg_signal_backend TO neon_superuser WITH ADMIN OPTION;`