Move check of page LSNB to heck_page_lsn() function

.github/actionlint.yml

@@ -31,7 +31,6 @@ config-variables:
   - NEON_PROD_AWS_ACCOUNT_ID
   - PGREGRESS_PG16_PROJECT_ID
   - PGREGRESS_PG17_PROJECT_ID
-  - PREWARM_PGBENCH_SIZE
   - REMOTE_STORAGE_AZURE_CONTAINER
   - REMOTE_STORAGE_AZURE_REGION
   - SLACK_CICD_CHANNEL_ID

.github/workflows/benchmarking.yml

@@ -219,7 +219,6 @@ jobs:
           --ignore test_runner/performance/test_cumulative_statistics_persistence.py
           --ignore test_runner/performance/test_perf_many_relations.py
           --ignore test_runner/performance/test_perf_oltp_large_tenant.py
-          --ignore test_runner/performance/test_lfc_prewarm.py
       env:
         BENCHMARK_CONNSTR: ${{ steps.create-neon-project.outputs.dsn }}
         VIP_VAP_ACCESS_TOKEN: "${{ secrets.VIP_VAP_ACCESS_TOKEN }}"
@@ -411,77 +410,6 @@ jobs:
       env:
         SLACK_BOT_TOKEN: ${{ secrets.SLACK_BOT_TOKEN }}
 
-  prewarm-test:
-    if: ${{ github.event.inputs.run_only_pgvector_tests == 'false' || github.event.inputs.run_only_pgvector_tests == null }}
-    permissions:
-      contents: write
-      statuses: write
-      id-token: write # aws-actions/configure-aws-credentials
-    env:
-      PGBENCH_SIZE: ${{ vars.PREWARM_PGBENCH_SIZE }}
-      POSTGRES_DISTRIB_DIR: /tmp/neon/pg_install
-      DEFAULT_PG_VERSION: 17
-      TEST_OUTPUT: /tmp/test_output
-      BUILD_TYPE: remote
-      SAVE_PERF_REPORT: ${{ github.event.inputs.save_perf_report || ( github.ref_name == 'main' ) }}
-      PLATFORM: "neon-staging"
-
-    runs-on: [ self-hosted, us-east-2, x64 ]
-    container:
-      image: ghcr.io/neondatabase/build-tools:pinned-bookworm
-      credentials:
-        username: ${{ github.actor }}
-        password: ${{ secrets.GITHUB_TOKEN }}
-      options: --init
-
-    steps:
-    - name: Harden the runner (Audit all outbound calls)
-      uses: step-security/harden-runner@4d991eb9b905ef189e4c376166672c3f2f230481 # v2.11.0
-      with:
-        egress-policy: audit
-
-    - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
-
-    - name: Configure AWS credentials
-      uses: aws-actions/configure-aws-credentials@e3dd6a429d7300a6a4c196c26e071d42e0343502 # v4.0.2
-      with:
-        aws-region: eu-central-1
-        role-to-assume: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-        role-duration-seconds: 18000 # 5 hours
-
-    - name: Download Neon artifact
-      uses: ./.github/actions/download
-      with:
-        name: neon-${{ runner.os }}-${{ runner.arch }}-release-artifact
-        path: /tmp/neon/
-        prefix: latest
-        aws-oidc-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-
-    - name: Run prewarm benchmark
-      uses: ./.github/actions/run-python-test-set
-      with:
-        build_type: ${{ env.BUILD_TYPE }}
-        test_selection: performance/test_lfc_prewarm.py
-        run_in_parallel: false
-        save_perf_report: ${{ env.SAVE_PERF_REPORT }}
-        extra_params: -m remote_cluster --timeout 5400
-        pg_version: ${{ env.DEFAULT_PG_VERSION }}
-        aws-oidc-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-      env:
-        VIP_VAP_ACCESS_TOKEN: "${{ secrets.VIP_VAP_ACCESS_TOKEN }}"
-        PERF_TEST_RESULT_CONNSTR: "${{ secrets.PERF_TEST_RESULT_CONNSTR }}"
-        NEON_API_KEY: ${{ secrets.NEON_STAGING_API_KEY }}
-
-    - name: Create Allure report
-      id: create-allure-report
-      if: ${{ !cancelled() }}
-      uses: ./.github/actions/allure-report-generate
-      with:
-        store-test-results-into-db: true
-        aws-oidc-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-      env:
-        REGRESS_TEST_RESULT_CONNSTR_NEW: ${{ secrets.REGRESS_TEST_RESULT_CONNSTR_NEW }}
-
   generate-matrices:
     if: ${{ github.event.inputs.run_only_pgvector_tests == 'false' || github.event.inputs.run_only_pgvector_tests == null }}
     # Create matrices for the benchmarking jobs, so we run benchmarks on rds only once a week (on Saturday)

Cargo.lock

@@ -1290,14 +1290,8 @@ dependencies = [
 name = "communicator"
 version = "0.1.0"
 dependencies = [
- "axum",
  "cbindgen",
- "http 1.1.0",
- "measured",
- "tokio",
- "tracing",
- "tracing-subscriber",
- "utils",
+ "neon-shmem",
  "workspace_hack",
 ]
 
@@ -1341,9 +1335,6 @@ dependencies = [
  "futures",
  "hostname-validator",
  "http 1.1.0",
- "http-body-util",
- "hyper 1.4.1",
- "hyper-util",
  "indexmap 2.9.0",
  "itertools 0.10.5",
  "jsonwebtoken",
@@ -1366,7 +1357,6 @@ dependencies = [
  "ring",
  "rlimit",
  "rust-ini",
- "scopeguard",
  "serde",
  "serde_json",
  "serde_with",
@@ -4306,7 +4296,6 @@ dependencies = [
  "pageserver_client",
  "pageserver_client_grpc",
  "pageserver_page_api",
- "pprof",
  "rand 0.8.5",
  "reqwest",
  "serde",
@@ -7574,7 +7563,6 @@ dependencies = [
  "futures-core",
  "pin-project-lite",
  "tokio",
- "tokio-util",
 ]
 
 [[package]]

Cargo.toml

@@ -201,7 +201,7 @@ tokio-epoll-uring = { git = "https://github.com/neondatabase/tokio-epoll-uring.g
 tokio-io-timeout = "1.2.0"
 tokio-postgres-rustls = "0.12.0"
 tokio-rustls = { version = "0.26.0", default-features = false, features = ["tls12", "ring"]}
-tokio-stream = { version = "0.1", features = ["sync"] }
+tokio-stream = "0.1"
 tokio-tar = "0.3"
 tokio-util = { version = "0.7.10", features = ["io", "io-util", "rt"] }
 toml = "0.8"

compute_tools/Cargo.toml

@@ -27,10 +27,7 @@ fail.workspace = true
 flate2.workspace = true
 futures.workspace = true
 http.workspace = true
-http-body-util.workspace = true
 hostname-validator = "1.1"
-hyper.workspace = true
-hyper-util.workspace = true
 indexmap.workspace = true
 itertools.workspace = true
 jsonwebtoken.workspace = true
@@ -47,7 +44,6 @@ postgres.workspace = true
 regex.workspace = true
 reqwest = { workspace = true, features = ["json"] }
 ring = "0.17"
-scopeguard.workspace = true
 serde.workspace = true
 serde_with.workspace = true
 serde_json.workspace = true

compute_tools/src/compute.rs

@@ -2487,7 +2487,7 @@ pub async fn installed_extensions(conf: tokio_postgres::Config) -> Result<()> {
                 serde_json::to_string(&extensions).expect("failed to serialize extensions list")
             );
         }
-        Err(err) => error!("could not get installed extensions: {err}"),
+        Err(err) => error!("could not get installed extensions: {err:?}"),
     }
     Ok(())
 }

compute_tools/src/http/routes/metrics.rs

@@ -1,20 +1,12 @@
-use std::path::Path;
-use std::sync::Arc;
-
-use anyhow::Context;
 use axum::body::Body;
-use axum::extract::State;
 use axum::response::Response;
+use http::StatusCode;
 use http::header::CONTENT_TYPE;
-use http_body_util::BodyExt;
-use hyper::{Request, StatusCode};
 use metrics::proto::MetricFamily;
 use metrics::{Encoder, TextEncoder};
 
-use crate::compute::ComputeNode;
 use crate::http::JsonResponse;
 use crate::metrics::collect;
-use crate::postgres_metrics_client::connect_postgres_metrics_socket;
 
 /// Expose Prometheus metrics.
 pub(in crate::http) async fn get_metrics() -> Response {
@@ -39,42 +31,3 @@ pub(in crate::http) async fn get_metrics() -> Response {
         .body(Body::from(buffer))
         .unwrap()
 }
-
-/// Fetch and forward metrics from the Postgres neon extension's metrics
-/// exporter that are used by autoscaling-agent.
-///
-/// The neon extension exposes these metrics over a Unix domain socket
-/// in the data directory. That's not accessible directly from the outside
-/// world, so we have this endpoint in compute_ctl to expose it
-pub(in crate::http) async fn get_autoscaling_metrics(
-    State(compute): State<Arc<ComputeNode>>,
-) -> Result<Response, Response> {
-    let pgdata = Path::new(&compute.params.pgdata);
-
-    // Connect to the communicator process's metrics socket
-    let mut metrics_client = connect_postgres_metrics_socket(pgdata)
-        .await
-        .map_err(|e| JsonResponse::error(StatusCode::INTERNAL_SERVER_ERROR, format!("{e:#}")))?;
-
-    // Make a request for /autoscaling_metrics
-    let request = Request::builder()
-        .method("GET")
-        .uri("/autoscaling_metrics")
-        .header("Host", "localhost") // hyper requires Host, even though the server won't care
-        .body(Body::from(""))
-        .unwrap();
-    let resp = metrics_client
-        .send_request(request)
-        .await
-        .context("fetching metrics from Postgres metrics service")
-        .map_err(|e| JsonResponse::error(StatusCode::INTERNAL_SERVER_ERROR, format!("{e:#}")))?;
-
-    // Build a response that just forwards the response we got.
-    let mut response = Response::builder();
-    response = response.status(resp.status());
-    if let Some(content_type) = resp.headers().get(CONTENT_TYPE) {
-        response = response.header(CONTENT_TYPE, content_type);
-    }
-    let body = tonic::service::AxumBody::from_stream(resp.into_body().into_data_stream());
-    Ok(response.body(body).unwrap())
-}

compute_tools/src/http/server.rs

@@ -81,12 +81,8 @@ impl From<&Server> for Router<Arc<ComputeNode>> {
             Server::External {
                 config, compute_id, ..
             } => {
-                let unauthenticated_router = Router::<Arc<ComputeNode>>::new()
-                    .route("/metrics", get(metrics::get_metrics))
-                    .route(
-                        "/autoscaling_metrics",
-                        get(metrics::get_autoscaling_metrics),
-                    );
+                let unauthenticated_router =
+                    Router::<Arc<ComputeNode>>::new().route("/metrics", get(metrics::get_metrics));
 
                 let authenticated_router = Router::<Arc<ComputeNode>>::new()
                     .route("/lfc/prewarm", get(lfc::prewarm_state).post(lfc::prewarm))

compute_tools/src/installed_extensions.rs

@@ -2,7 +2,6 @@ use std::collections::HashMap;
 
 use anyhow::Result;
 use compute_api::responses::{InstalledExtension, InstalledExtensions};
-use tokio_postgres::error::Error as PostgresError;
 use tokio_postgres::{Client, Config, NoTls};
 
 use crate::metrics::INSTALLED_EXTENSIONS;
@@ -11,7 +10,7 @@ use crate::metrics::INSTALLED_EXTENSIONS;
 /// and to make database listing query here more explicit.
 ///
 /// Limit the number of databases to 500 to avoid excessive load.
-async fn list_dbs(client: &mut Client) -> Result<Vec<String>, PostgresError> {
+async fn list_dbs(client: &mut Client) -> Result<Vec<String>> {
     // `pg_database.datconnlimit = -2` means that the database is in the
     // invalid state
     let databases = client
@@ -38,9 +37,7 @@ async fn list_dbs(client: &mut Client) -> Result<Vec<String>, PostgresError> {
 /// Same extension can be installed in multiple databases with different versions,
 /// so we report a separate metric (number of databases where it is installed)
 /// for each extension version.
-pub async fn get_installed_extensions(
-    mut conf: Config,
-) -> Result<InstalledExtensions, PostgresError> {
+pub async fn get_installed_extensions(mut conf: Config) -> Result<InstalledExtensions> {
     conf.application_name("compute_ctl:get_installed_extensions");
     let databases: Vec<String> = {
         let (mut client, connection) = conf.connect(NoTls).await?;

compute_tools/src/lib.rs

@@ -24,7 +24,6 @@ pub mod monitor;
 pub mod params;
 pub mod pg_helpers;
 pub mod pgbouncer;
-pub mod postgres_metrics_client;
 pub mod rsyslog;
 pub mod spec;
 mod spec_apply;

compute_tools/src/postgres_metrics_client.rs

@@ -1,97 +0,0 @@
-//! Client for making request to a running Postgres server's metrics service
-//!
-//! The storage communicator process that runs inside Postgres exposes
-//! an HTTP endpoint in a Unix Domain Socket in the Postgres data
-//! directory. This provides access to it.
-
-use std::path::Path;
-
-use anyhow::Context;
-use hyper::client::conn::http1::SendRequest;
-use hyper_util::rt::TokioIo;
-
-/// Name of the socket within the Postgres data directory. This better match that in
-/// `pgxn/neon/communicator/src/worker_process/metrics_exporter.rs`.
-const NEON_COMMUNICATOR_SOCKET_NAME: &str = "neon-communicator.socket";
-
-/// Open a connection to the metrics exporter's socket, prepare to send requests to it
-/// with hyper.
-pub async fn connect_postgres_metrics_socket<B>(pgdata: &Path) -> anyhow::Result<SendRequest<B>>
-where
-    B: hyper::body::Body + 'static + Send,
-    B::Data: Send,
-    B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
-{
-    let socket_path = pgdata.join(NEON_COMMUNICATOR_SOCKET_NAME);
-    let socket_path_len = socket_path.display().to_string().len();
-
-    // There is a limit of around 100 bytes (108 on Linux?) on the length of the path to a
-    // Unix Domain socket. The limit is on the connect(2) function used to open the
-    // socket, not on the absolute path itself. Postgres changes the current directory to
-    // the data directory and uses a relative path to bind to the socket, and the relative
-    // path "./neon-communicator.socket" is always short, but when compute_ctl needs to
-    // open the socket, we need to use a full path, which can be arbitrarily long.
-    //
-    // There are a few ways we could work around this:
-    //
-    // 1. Change the current directory to the Postgres data directory and use a relative
-    //    path  in the connect(2) call. That's problematic because the current directory
-    //    applies to the whole process. We could change the current directory early in
-    //    compute_ctl startup, and that might be a good idea anyway for other reasons too:
-    //    it would be more robust if the data directory is moved around or unlinked for some
-    //    reason, and you would be less likely to accidentally litter other parts of the
-    //    filesystem with e.g. temporary files. However, that's a pretty invasive change.
-    //
-    // 2. On Linux, you could open() the data directory, and refer to the the socket inside it
-    //    as "/proc/self/fd/<fd>/neon-communicator.socket". But that's Linux-only.
-    //
-    // 3. Create a symbolic link to the socket with a shorter path, and use that.
-    //
-    // We use the symbolic link approach here. Hopefully the paths we use in production
-    // are shorter, so that we can open the socket directly, so that this hack is needed
-    // only in development.
-    let connect_result = if socket_path_len < 100 {
-        // We can open the path directly with no hacks.
-        tokio::net::UnixStream::connect(socket_path).await
-    } else {
-        // The path to the socket is too long. Create a symlink to it with a shorter path.
-        let short_path = std::env::temp_dir().join(format!(
-            "compute_ctl.short-socket.{}.{}",
-            std::process::id(),
-            tokio::task::id()
-        ));
-        std::os::unix::fs::symlink(&socket_path, &short_path)?;
-
-        // Delete the symlink as soon as we have connected to it. There's a small chance
-        // of leaking if the process dies before we remove it, so try to keep that window
-        // as small as possible.
-        scopeguard::defer! {
-            if let Err(err) = std::fs::remove_file(&short_path) {
-                tracing::warn!("could not remove symlink \"{}\" created for socket: {}",
-                               short_path.display(), err);
-            }
-        }
-
-        tracing::info!(
-            "created symlink \"{}\" for socket \"{}\", opening it now",
-            short_path.display(),
-            socket_path.display()
-        );
-
-        tokio::net::UnixStream::connect(&short_path).await
-    };
-
-    let stream = connect_result.context("opening postgres metrics socket")?;
-
-    let io = TokioIo::new(stream);
-    let (request_sender, connection) = hyper::client::conn::http1::handshake(io).await.unwrap();
-
-    // spawn a task to poll the connection and drive the HTTP state
-    tokio::spawn(async move {
-        if let Err(err) = connection.await {
-            eprintln!("Error in connection: {err}");
-        }
-    });
-
-    Ok(request_sender)
-}

control_plane/storcon_cli/src/main.rs

@@ -476,7 +476,6 @@ async fn main() -> anyhow::Result<()> {
                         listen_http_port,
                         listen_https_port,
                         availability_zone_id: AvailabilityZone(availability_zone_id),
-                        node_ip_addr: None,
                     }),
                 )
                 .await?;

libs/pageserver_api/src/controller_api.rs

@@ -1,6 +1,5 @@
 use std::collections::{HashMap, HashSet};
 use std::fmt::Display;
-use std::net::IpAddr;
 use std::str::FromStr;
 use std::time::{Duration, Instant};
 
@@ -61,11 +60,6 @@ pub struct NodeRegisterRequest {
     pub listen_https_port: Option<u16>,
 
     pub availability_zone_id: AvailabilityZone,
-
-    // Reachable IP address of the PS/SK registering, if known.
-    // Hadron Cluster Coordiantor will update the DNS record of the registering node
-    // with this IP address.
-    pub node_ip_addr: Option<IpAddr>,
 }
 
 #[derive(Serialize, Deserialize)]
@@ -551,39 +545,6 @@ pub struct SafekeeperDescribeResponse {
     pub scheduling_policy: SkSchedulingPolicy,
 }
 
-#[derive(Serialize, Deserialize, Clone, Debug)]
-pub struct TimelineSafekeeperPeer {
-    pub node_id: NodeId,
-    pub listen_http_addr: String,
-    pub http_port: i32,
-}
-
-#[derive(Serialize, Deserialize, Clone, Debug)]
-pub struct SCSafekeeperTimeline {
-    // SC does not know the tenant id.
-    pub timeline_id: TimelineId,
-    pub peers: Vec<NodeId>,
-}
-
-#[derive(Serialize, Deserialize, Clone, Debug)]
-pub struct SCSafekeeperTimelinesResponse {
-    pub timelines: Vec<SCSafekeeperTimeline>,
-    pub safekeeper_peers: Vec<TimelineSafekeeperPeer>,
-}
-
-#[derive(Serialize, Deserialize, Clone, Debug)]
-pub struct SafekeeperTimeline {
-    pub tenant_id: TenantId,
-    pub timeline_id: TimelineId,
-    pub peers: Vec<NodeId>,
-}
-
-#[derive(Serialize, Deserialize, Clone, Debug)]
-pub struct SafekeeperTimelinesResponse {
-    pub timelines: Vec<SafekeeperTimeline>,
-    pub safekeeper_peers: Vec<TimelineSafekeeperPeer>,
-}
-
 #[derive(Serialize, Deserialize, Clone)]
 pub struct SafekeeperSchedulingPolicyRequest {
     pub scheduling_policy: SkSchedulingPolicy,

libs/utils/src/ip_address.rs

@@ -1,73 +0,0 @@
-use std::env::{VarError, var};
-use std::error::Error;
-use std::net::IpAddr;
-use std::str::FromStr;
-
-/// Name of the environment variable containing the reachable IP address of the node. If set, the IP address contained in this
-/// environment variable is used as the reachable IP address of the pageserver or safekeeper node during node registration.
-/// In a Kubernetes environment, this environment variable should be set by Kubernetes to the Pod IP (specified in the Pod
-/// template).
-pub const HADRON_NODE_IP_ADDRESS: &str = "HADRON_NODE_IP_ADDRESS";
-
-/// Read the reachable IP address of this page server from env var HADRON_NODE_IP_ADDRESS.
-/// In Kubernetes this environment variable is set to the Pod IP (specified in the Pod template).
-pub fn read_node_ip_addr_from_env() -> Result<Option<IpAddr>, Box<dyn Error>> {
-    match var(HADRON_NODE_IP_ADDRESS) {
-        Ok(v) => {
-            if let Ok(addr) = IpAddr::from_str(&v) {
-                Ok(Some(addr))
-            } else {
-                Err(format!("Invalid IP address string: {v}. Cannot be parsed as either an IPv4 or an IPv6 address.").into())
-            }
-        }
-        Err(VarError::NotPresent) => Ok(None),
-        Err(e) => Err(e.into()),
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use std::env;
-    use std::net::{Ipv4Addr, Ipv6Addr};
-
-    #[test]
-    fn test_read_node_ip_addr_from_env() {
-        // SAFETY: test code
-        unsafe {
-            // Test with a valid IPv4 address
-            env::set_var(HADRON_NODE_IP_ADDRESS, "192.168.1.1");
-            let result = read_node_ip_addr_from_env().unwrap();
-            assert_eq!(result, Some(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1))));
-
-            // Test with a valid IPv6 address
-            env::set_var(
-                HADRON_NODE_IP_ADDRESS,
-                "2001:0db8:85a3:0000:0000:8a2e:0370:7334",
-            );
-        }
-        let result = read_node_ip_addr_from_env().unwrap();
-        assert_eq!(
-            result,
-            Some(IpAddr::V6(
-                Ipv6Addr::from_str("2001:0db8:85a3:0000:0000:8a2e:0370:7334").unwrap()
-            ))
-        );
-
-        // Test with an invalid IP address
-        // SAFETY: test code
-        unsafe {
-            env::set_var(HADRON_NODE_IP_ADDRESS, "invalid_ip");
-        }
-        let result = read_node_ip_addr_from_env();
-        assert!(result.is_err());
-
-        // Test with no environment variable set
-        // SAFETY: test code
-        unsafe {
-            env::remove_var(HADRON_NODE_IP_ADDRESS);
-        }
-        let result = read_node_ip_addr_from_env().unwrap();
-        assert_eq!(result, None);
-    }
-}

libs/utils/src/lib.rs

@@ -26,9 +26,6 @@ pub mod auth;
 // utility functions and helper traits for unified unique id generation/serialization etc.
 pub mod id;
 
-// utility functions to obtain reachable IP addresses in PS/SK nodes.
-pub mod ip_address;
-
 pub mod shard;
 
 mod hex;

libs/utils/src/logging.rs

@@ -1,5 +1,4 @@
 use std::future::Future;
-use std::pin::Pin;
 use std::str::FromStr;
 use std::time::Duration;
 
@@ -8,7 +7,7 @@ use metrics::{IntCounter, IntCounterVec};
 use once_cell::sync::Lazy;
 use strum_macros::{EnumString, VariantNames};
 use tokio::time::Instant;
-use tracing::{info, warn};
+use tracing::info;
 
 /// Logs a critical error, similarly to `tracing::error!`. This will:
 ///
@@ -378,11 +377,10 @@ impl std::fmt::Debug for SecretString {
 ///
 /// TODO: consider upgrading this to a warning, but currently it fires too often.
 #[inline]
-pub async fn log_slow<O>(
-    name: &str,
-    threshold: Duration,
-    f: Pin<&mut impl Future<Output = O>>,
-) -> O {
+pub async fn log_slow<F, O>(name: &str, threshold: Duration, f: std::pin::Pin<&mut F>) -> O
+where
+    F: Future<Output = O>,
+{
     monitor_slow_future(
         threshold,
         threshold, // period = threshold
@@ -396,42 +394,16 @@ pub async fn log_slow<O>(
             if !is_slow {
                 return;
             }
-            let elapsed = elapsed_total.as_secs_f64();
             if ready {
-                info!("slow {name} completed after {elapsed:.3}s");
+                info!(
+                    "slow {name} completed after {:.3}s",
+                    elapsed_total.as_secs_f64()
+                );
             } else {
-                info!("slow {name} still running after {elapsed:.3}s");
-            }
-        },
-    )
-    .await
-}
-
-/// Logs a periodic warning if a future is slow to complete.
-#[inline]
-pub async fn warn_slow<O>(
-    name: &str,
-    threshold: Duration,
-    f: Pin<&mut impl Future<Output = O>>,
-) -> O {
-    monitor_slow_future(
-        threshold,
-        threshold, // period = threshold
-        f,
-        |MonitorSlowFutureCallback {
-             ready,
-             is_slow,
-             elapsed_total,
-             elapsed_since_last_callback: _,
-         }| {
-            if !is_slow {
-                return;
-            }
-            let elapsed = elapsed_total.as_secs_f64();
-            if ready {
-                warn!("slow {name} completed after {elapsed:.3}s");
-            } else {
-                warn!("slow {name} still running after {elapsed:.3}s");
+                info!(
+                    "slow {name} still running after {:.3}s",
+                    elapsed_total.as_secs_f64()
+                );
             }
         },
     )
@@ -444,7 +416,7 @@ pub async fn warn_slow<O>(
 pub async fn monitor_slow_future<F, O>(
     threshold: Duration,
     period: Duration,
-    mut fut: Pin<&mut F>,
+    mut fut: std::pin::Pin<&mut F>,
     mut cb: impl FnMut(MonitorSlowFutureCallback),
 ) -> O
 where

pageserver/client_grpc/src/client.rs

@@ -1,16 +1,13 @@
 use std::collections::HashMap;
 use std::num::NonZero;
-use std::pin::pin;
 use std::sync::Arc;
-use std::time::{Duration, Instant};
 
 use anyhow::anyhow;
 use arc_swap::ArcSwap;
 use futures::stream::FuturesUnordered;
 use futures::{FutureExt as _, StreamExt as _};
 use tonic::codec::CompressionEncoding;
-use tracing::{debug, instrument};
-use utils::logging::warn_slow;
+use tracing::instrument;
 
 use crate::pool::{ChannelPool, ClientGuard, ClientPool, StreamGuard, StreamPool};
 use crate::retry::Retry;
@@ -24,40 +21,28 @@ 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.
 ///
-/// Normal requests are small, and we don't pipeline them, so we can afford a large number of
-/// streams per connection.
-///
 /// TODO: tune all of these constants, and consider making them configurable.
-const MAX_CLIENTS_PER_CHANNEL: NonZero<usize> = NonZero::new(64).unwrap();
+/// 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 bulk GetPage streams per channel (i.e. TCP connection). These use a
-/// dedicated channel pool with a lower client limit, to avoid TCP-level head-of-line blocking and
-/// transmission delays. This also concentrates large window sizes on a smaller set of
-/// streams/connections, presumably reducing memory use.
-const MAX_BULK_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();
 
-/// The batch size threshold at which a GetPage request will use the bulk stream pool.
-///
-/// The gRPC initial window size is 64 KB. Each page is 8 KB, so let's avoid increasing the window
-/// size for the normal stream pool, and route requests for >= 5 pages (>32 KB) to the bulk pool.
-const BULK_THRESHOLD_BATCH_SIZE: usize = 5;
+/// 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();
 
-/// The overall request call timeout, including retries and pool acquisition.
-/// TODO: should we retry forever? Should the caller decide?
-const CALL_TIMEOUT: Duration = Duration::from_secs(60);
+/// Max number of pipelined requests per stream.
+const MAX_STREAM_QUEUE_DEPTH: NonZero<usize> = NonZero::new(2).unwrap();
 
-/// The per-request (retry attempt) timeout, including any lazy connection establishment.
-const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
+/// 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();
 
-/// The initial request retry backoff duration. The first retry does not back off.
-/// TODO: use a different backoff for ResourceExhausted (rate limiting)? Needs server support.
-const BASE_BACKOFF: Duration = Duration::from_millis(5);
-
-/// The maximum request retry backoff duration.
-const MAX_BACKOFF: Duration = Duration::from_secs(5);
-
-/// Threshold and interval for warning about slow operation.
-const SLOW_THRESHOLD: Duration = Duration::from_secs(3);
+/// 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:
@@ -65,19 +50,10 @@ const SLOW_THRESHOLD: Duration = Duration::from_secs(3);
 /// * Sharded tenants across multiple Pageservers.
 /// * Pooling of connections, clients, and streams for efficient resource use.
 /// * Concurrent use by many callers.
-/// * Internal handling of GetPage bidirectional streams.
+/// * Internal handling of GetPage bidirectional streams, with pipelining and error handling.
 /// * Automatic retries.
 /// * Observability.
 ///
-/// The client has dedicated connection/client/stream pools per shard, for resource reuse. These
-/// pools are unbounded: we allow scaling out as many concurrent streams as needed to serve all
-/// concurrent callers, which mostly eliminates head-of-line blocking. Idle streams are fairly
-/// cheap: the server task currently uses 26 KB of memory, so we can comfortably fit 100,000
-/// concurrent idle streams (2.5 GB memory). The worst case degenerates to the old libpq case with
-/// one stream per backend, but without the TCP connection overhead. In the common case we expect
-/// significantly lower stream counts due to stream sharing, driven e.g. by idle backends, LFC hits,
-/// read coalescing, sharding (backends typically only talk to one shard at a time), etc.
-///
 /// 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 {
@@ -91,6 +67,8 @@ pub struct PageserverClient {
     compression: Option<CompressionEncoding>,
     /// The shards for this tenant.
     shards: ArcSwap<Shards>,
+    /// The retry configuration.
+    retry: Retry,
 }
 
 impl PageserverClient {
@@ -116,6 +94,7 @@ impl PageserverClient {
             auth_token,
             compression,
             shards: ArcSwap::new(Arc::new(shards)),
+            retry: Retry,
         })
     }
 
@@ -163,15 +142,13 @@ impl PageserverClient {
         &self,
         req: page_api::CheckRelExistsRequest,
     ) -> tonic::Result<page_api::CheckRelExistsResponse> {
-        debug!("sending request: {req:?}");
-        let resp = Self::with_retries(CALL_TIMEOUT, async |_| {
-            // Relation metadata is only available on shard 0.
-            let mut client = self.shards.load_full().get_zero().client().await?;
-            Self::with_timeout(REQUEST_TIMEOUT, client.check_rel_exists(req)).await
-        })
-        .await?;
-        debug!("received response: {resp:?}");
-        Ok(resp)
+        self.retry
+            .with(async |_| {
+                // Relation metadata is only available on shard 0.
+                let mut client = self.shards.load_full().get_zero().client().await?;
+                client.check_rel_exists(req).await
+            })
+            .await
     }
 
     /// Returns the total size of a database, as # of bytes.
@@ -180,15 +157,13 @@ impl PageserverClient {
         &self,
         req: page_api::GetDbSizeRequest,
     ) -> tonic::Result<page_api::GetDbSizeResponse> {
-        debug!("sending request: {req:?}");
-        let resp = Self::with_retries(CALL_TIMEOUT, async |_| {
-            // Relation metadata is only available on shard 0.
-            let mut client = self.shards.load_full().get_zero().client().await?;
-            Self::with_timeout(REQUEST_TIMEOUT, client.get_db_size(req)).await
-        })
-        .await?;
-        debug!("received response: {resp:?}");
-        Ok(resp)
+        self.retry
+            .with(async |_| {
+                // Relation metadata is only available on shard 0.
+                let mut client = self.shards.load_full().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
@@ -218,8 +193,6 @@ impl PageserverClient {
             return Err(tonic::Status::invalid_argument("request attempt must be 0"));
         }
 
-        debug!("sending request: {req:?}");
-
         // 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
@@ -228,16 +201,13 @@ impl PageserverClient {
         //
         // TODO: the gRPC server and client doesn't yet properly support shard splits. Revisit this
         // once we figure out how to handle these.
-        let resp = Self::with_retries(CALL_TIMEOUT, async |attempt| {
-            let mut req = req.clone();
-            req.request_id.attempt = attempt as u32;
-            let shards = self.shards.load_full();
-            Self::with_timeout(REQUEST_TIMEOUT, Self::get_page_with_shards(req, &shards)).await
-        })
-        .await?;
-
-        debug!("received response: {resp:?}");
-        Ok(resp)
+        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
@@ -276,7 +246,7 @@ impl PageserverClient {
         req: page_api::GetPageRequest,
         shard: &Shard,
     ) -> tonic::Result<page_api::GetPageResponse> {
-        let mut stream = shard.stream(Self::is_bulk(&req)).await?;
+        let stream = shard.stream(req.request_class.is_bulk()).await;
         let resp = stream.send(req.clone()).await?;
 
         // Convert per-request errors into a tonic::Status.
@@ -320,15 +290,13 @@ impl PageserverClient {
         &self,
         req: page_api::GetRelSizeRequest,
     ) -> tonic::Result<page_api::GetRelSizeResponse> {
-        debug!("sending request: {req:?}");
-        let resp = Self::with_retries(CALL_TIMEOUT, async |_| {
-            // Relation metadata is only available on shard 0.
-            let mut client = self.shards.load_full().get_zero().client().await?;
-            Self::with_timeout(REQUEST_TIMEOUT, client.get_rel_size(req)).await
-        })
-        .await?;
-        debug!("received response: {resp:?}");
-        Ok(resp)
+        self.retry
+            .with(async |_| {
+                // Relation metadata is only available on shard 0.
+                let mut client = self.shards.load_full().get_zero().client().await?;
+                client.get_rel_size(req).await
+            })
+            .await
     }
 
     /// Fetches an SLRU segment.
@@ -337,50 +305,13 @@ impl PageserverClient {
         &self,
         req: page_api::GetSlruSegmentRequest,
     ) -> tonic::Result<page_api::GetSlruSegmentResponse> {
-        debug!("sending request: {req:?}");
-        let resp = Self::with_retries(CALL_TIMEOUT, async |_| {
-            // SLRU segments are only available on shard 0.
-            let mut client = self.shards.load_full().get_zero().client().await?;
-            Self::with_timeout(REQUEST_TIMEOUT, client.get_slru_segment(req)).await
-        })
-        .await?;
-        debug!("received response: {resp:?}");
-        Ok(resp)
-    }
-
-    /// Runs the given async closure with retries up to the given timeout. Only certain gRPC status
-    /// codes are retried, see [`Retry::should_retry`]. Returns `DeadlineExceeded` on timeout.
-    async fn with_retries<T, F, O>(timeout: Duration, f: F) -> tonic::Result<T>
-    where
-        F: FnMut(usize) -> O, // pass attempt number, starting at 0
-        O: Future<Output = tonic::Result<T>>,
-    {
-        Retry {
-            timeout: Some(timeout),
-            base_backoff: BASE_BACKOFF,
-            max_backoff: MAX_BACKOFF,
-        }
-        .with(f)
-        .await
-    }
-
-    /// Runs the given future with a timeout. Returns `DeadlineExceeded` on timeout.
-    async fn with_timeout<T>(
-        timeout: Duration,
-        f: impl Future<Output = tonic::Result<T>>,
-    ) -> tonic::Result<T> {
-        let started = Instant::now();
-        tokio::time::timeout(timeout, f).await.map_err(|_| {
-            tonic::Status::deadline_exceeded(format!(
-                "request timed out after {:.3}s",
-                started.elapsed().as_secs_f64()
-            ))
-        })?
-    }
-
-    /// Returns true if the request is considered a bulk request and should use the bulk pool.
-    fn is_bulk(req: &page_api::GetPageRequest) -> bool {
-        req.block_numbers.len() >= BULK_THRESHOLD_BATCH_SIZE
+        self.retry
+            .with(async |_| {
+                // SLRU segments are only available on shard 0.
+                let mut client = self.shards.load_full().get_zero().client().await?;
+                client.get_slru_segment(req).await
+            })
+            .await
     }
 }
 
@@ -509,23 +440,15 @@ impl Shards {
     }
 }
 
-/// A single shard. Has dedicated resource pools with the following structure:
+/// A single shard. Uses dedicated resource pools with the following structure:
 ///
-/// * Channel pool: MAX_CLIENTS_PER_CHANNEL.
-///   * Client pool: unbounded.
-///     * Stream pool: unbounded.
-/// * Bulk channel pool: MAX_BULK_CLIENTS_PER_CHANNEL.
+/// * 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: unbounded.
-///
-/// We use a separate bulk channel pool with a lower concurrency limit for large batch requests.
-/// This avoids TCP-level head-of-line blocking, and also concentrates large window sizes on a
-/// smaller set of streams/connections, which presumably reduces memory use. Neither of these pools
-/// are bounded, nor do they pipeline requests, so the latency characteristics should be mostly
-/// similar (except for TCP transmission time).
-///
-/// TODO: since we never use bounded pools, we could consider removing the pool limiters. However,
-/// the code is fairly trivial, so we may as well keep them around for now in case we need them.
+///     * Bulk stream pool: MAX_BULK_STREAMS and MAX_BULK_STREAM_QUEUE_DEPTH.
 struct Shard {
     /// The shard ID.
     id: ShardIndex,
@@ -533,7 +456,7 @@ struct Shard {
     client_pool: Arc<ClientPool>,
     /// GetPage stream pool.
     stream_pool: Arc<StreamPool>,
-    /// GetPage stream pool for bulk requests.
+    /// GetPage stream pool for bulk requests, e.g. prefetches.
     bulk_stream_pool: Arc<StreamPool>,
 }
 
@@ -547,30 +470,50 @@ impl Shard {
         auth_token: Option<String>,
         compression: Option<CompressionEncoding>,
     ) -> anyhow::Result<Self> {
-        // Shard pools for unary requests and non-bulk GetPage requests.
+        // Common channel pool for unary and stream requests. Bounded by client/stream pools.
+        let channel_pool = ChannelPool::new(url.clone(), MAX_CLIENTS_PER_CHANNEL)?;
+
+        // Client pool for unary requests.
         let client_pool = ClientPool::new(
-            ChannelPool::new(url.clone(), MAX_CLIENTS_PER_CHANNEL)?,
+            channel_pool.clone(),
             tenant_id,
             timeline_id,
             shard_id,
             auth_token.clone(),
             compression,
-            None, // unbounded
+            Some(MAX_UNARY_CLIENTS),
         );
-        let stream_pool = StreamPool::new(client_pool.clone(), None); // unbounded
 
-        // Bulk GetPage stream pool for large batches (prefetches, sequential scans, vacuum, etc.).
+        // 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_BULK_CLIENTS_PER_CHANNEL)?,
+                ChannelPool::new(url, MAX_CLIENTS_PER_CHANNEL)?,
                 tenant_id,
                 timeline_id,
                 shard_id,
                 auth_token,
                 compression,
-                None, // unbounded,
+                None, // unbounded, limited by stream pool
             ),
-            None, // unbounded
+            Some(MAX_BULK_STREAMS),
+            MAX_BULK_STREAM_QUEUE_DEPTH,
         );
 
         Ok(Self {
@@ -582,23 +525,19 @@ impl Shard {
     }
 
     /// Returns a pooled client for this shard.
-    #[instrument(skip_all)]
     async fn client(&self) -> tonic::Result<ClientGuard> {
-        warn_slow(
-            "client pool acquisition",
-            SLOW_THRESHOLD,
-            pin!(self.client_pool.get()),
-        )
-        .await
+        self.client_pool
+            .get()
+            .await
+            .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 pool.
-    #[instrument(skip_all, fields(bulk))]
-    async fn stream(&self, bulk: bool) -> tonic::Result<StreamGuard> {
-        let pool = match bulk {
-            false => &self.stream_pool,
-            true => &self.bulk_stream_pool,
-        };
-        warn_slow("stream pool acquisition", SLOW_THRESHOLD, pin!(pool.get())).await
+    /// 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,
+        }
     }
 }

pageserver/client_grpc/src/pool.rs

@@ -9,36 +9,19 @@
 //!
 //! * ChannelPool: manages gRPC channels (TCP connections) to a single Pageserver. Multiple clients
 //!   can acquire and use the same channel concurrently (via HTTP/2 stream multiplexing), up to a
-//!   per-channel client limit. Channels are closed immediately when empty, and indirectly rely on
-//!   client/stream idle timeouts.
+//!   per-channel client limit. Channels may be closed when they are no longer used by any clients.
 //!
 //! * ClientPool: manages gRPC clients for a single tenant shard. Each client acquires a (shared)
 //!   channel from the ChannelPool for the client's lifetime. A client can only be acquired by a
-//!   single caller at a time, and is returned to the pool when dropped. Idle clients are removed
-//!   from the pool after a while to free up resources.
+//!   single caller at a time, and is returned to the pool when dropped. Idle clients may be removed
+//!   from the pool after some time, to free up the channel.
 //!
 //! * StreamPool: manages bidirectional gRPC GetPage streams. Each stream acquires a client from the
-//!   ClientPool for the stream's lifetime. A stream can only be acquired by a single caller at a
-//!   time, and is returned to the pool when dropped. Idle streams are removed from the pool after
-//!   a while to free up resources.
-//!
-//!   The stream only supports sending a single, synchronous request at a time, and does not support
-//!   pipelining multiple requests from different callers onto the same stream -- instead, we scale
-//!   out concurrent streams to improve throughput. There are many reasons for this design choice:
-//!
-//!     * It (mostly) eliminates head-of-line blocking. A single stream is processed sequentially by
-//!       a single server task, which may block e.g. on layer downloads, LSN waits, etc.
-//!
-//!     * Cancellation becomes trivial, by closing the stream. Otherwise, if a caller goes away
-//!       (e.g. because of a timeout), the request would still be processed by the server and block
-//!       requests behind it in the stream. It might even block its own timeout retry.
-//!
-//!     * Stream scheduling becomes significantly simpler and cheaper.
-//!
-//!     * Individual callers can still use client-side batching for pipelining.
-//!
-//!     * Idle streams are cheap. Benchmarks show that an idle GetPage stream takes up about 26 KB
-//!       per stream (2.5 GB for 100,000 streams), so we can afford to scale out.
+//!   ClientPool for the stream's lifetime. Internal streams are not exposed to callers; instead, it
+//!   returns a guard that can be used to send a single request, to properly enforce queue depth and
+//!   route responses. Internally, the pool will reuse or spin up a suitable stream for the request,
+//!   possibly pipelining multiple requests from multiple callers on the same stream (up to some
+//!   queue depth). Idle streams may be removed from the pool after a while to free up the client.
 //!
 //! Each channel corresponds to one TCP connection. Each client unary request and each stream
 //! corresponds to one HTTP/2 stream and server task.
@@ -46,31 +29,33 @@
 //! TODO: error handling (including custom error types).
 //! TODO: observability.
 
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, HashMap};
 use std::num::NonZero;
 use std::ops::{Deref, DerefMut};
-use std::pin::Pin;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex, Weak};
 use std::time::{Duration, Instant};
 
-use futures::{Stream, StreamExt as _};
-use tokio::sync::{OwnedSemaphorePermit, Semaphore, watch};
-use tokio_stream::wrappers::WatchStream;
+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};
 
 use pageserver_page_api as page_api;
 use utils::id::{TenantId, TimelineId};
 use utils::shard::ShardIndex;
 
-/// Reap clients/streams that have been idle for this long. Channels are reaped immediately when
-/// empty, and indirectly rely on the client/stream idle timeouts.
+/// Reap channels/clients/streams that have been idle for this long.
 ///
-/// A stream's client will be reaped after 2x the idle threshold (first stream the client), but
-/// that's okay -- if the stream closes abruptly (e.g. due to timeout or cancellation), we want to
-/// keep its client around in the pool for a while.
+/// 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
@@ -98,6 +83,8 @@ pub struct ChannelPool {
     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,
 }
@@ -109,6 +96,9 @@ 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 {
@@ -118,12 +108,15 @@ impl ChannelPool {
         E: TryInto<Endpoint> + Send + Sync + 'static,
         <E as TryInto<Endpoint>>::Error: std::error::Error + Send + Sync,
     {
-        Ok(Arc::new(Self {
+        let pool = 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.
@@ -144,17 +137,22 @@ impl ChannelPool {
         let mut channels = self.channels.lock().unwrap();
 
         // Try to find an existing channel with available capacity. We check entries in BTreeMap
-        // order, to fill up the lower-ordered channels first. The client/stream pools also prefer
-        // clients with lower-ordered channel IDs first. This will cluster clients in lower-ordered
+        // order, to fill up the lower-ordered channels first. The ClientPool also prefers clients
+        // 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_ne!(entry.clients, 0, "empty channel not reaped");
+            assert_eq!(
+                entry.idle_since.is_some(),
+                entry.clients == 0,
+                "incorrect channel idle state"
+            );
             if entry.clients < self.max_clients_per_channel.get() {
                 entry.clients += 1;
+                entry.idle_since = None;
                 return ChannelGuard {
                     pool: Arc::downgrade(self),
                     id,
@@ -171,6 +169,7 @@ impl ChannelPool {
         let entry = ChannelEntry {
             channel: channel.clone(),
             clients: 1, // account for the guard below
+            idle_since: None,
         };
         channels.insert(id, entry);
 
@@ -182,6 +181,20 @@ 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 {
@@ -198,7 +211,7 @@ impl ChannelGuard {
     }
 }
 
-/// Returns the channel to the pool. The channel is closed when empty.
+/// Returns the channel to the pool.
 impl Drop for ChannelGuard {
     fn drop(&mut self) {
         let Some(pool) = self.pool.upgrade() else {
@@ -207,12 +220,11 @@ impl Drop for ChannelGuard {
 
         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;
-
-        // Reap empty channels immediately.
         if entry.clients == 0 {
-            channels.remove(&self.id);
+            entry.idle_since = Some(Instant::now()); // mark channel as idle
         }
     }
 }
@@ -241,7 +253,8 @@ pub struct ClientPool {
     ///
     /// The first client in the map will be acquired next. The map is sorted by client ID, which in
     /// turn is sorted by its channel ID, such that we prefer acquiring idle clients from
-    /// lower-ordered channels. This allows us to free up and reap higher-ordered channels.
+    /// 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,
@@ -297,7 +310,7 @@ impl ClientPool {
     /// 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>) -> tonic::Result<ClientGuard> {
+    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() {
@@ -315,7 +328,7 @@ impl ClientPool {
             });
         }
 
-        // Construct a new client.
+        // Slow path: construct a new client.
         let mut channel_guard = self.channel_pool.get();
         let client = page_api::Client::new(
             channel_guard.take(),
@@ -324,8 +337,7 @@ impl ClientPool {
             self.shard_id,
             self.auth_token.clone(),
             self.compression,
-        )
-        .map_err(|err| tonic::Status::internal(format!("failed to create client: {err}")))?;
+        )?;
 
         Ok(ClientGuard {
             pool: Arc::downgrade(self),
@@ -395,187 +407,287 @@ impl Drop for ClientGuard {
 /// A pool of bidirectional gRPC streams. Currently only used for GetPage streams. Each stream
 /// acquires a client from the inner `ClientPool` for the stream's lifetime.
 ///
-/// Individual streams only send a single request at a time, and do not pipeline multiple callers
-/// onto the same stream. Instead, we scale out the number of concurrent streams. This is primarily
-/// to eliminate head-of-line blocking. See the module documentation for more details.
+/// Individual streams are not exposed to callers -- instead, the returned guard can be used to send
+/// 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: 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.
     client_pool: Arc<ClientPool>,
-    /// Idle pooled streams. Acquired streams are removed from here and returned on drop.
+    /// All pooled streams.
     ///
-    /// The first stream in the map will be acquired next. The map is sorted by stream ID, which is
-    /// equivalent to the client ID and in turn sorted by its channel ID. This way we prefer
-    /// acquiring idle streams from lower-ordered channels, which allows us to free up and reap
-    /// higher-ordered channels.
-    idle: Mutex<BTreeMap<StreamID, StreamEntry>>,
-    /// Limits the max number of concurrent streams. None if the pool is unbounded.
+    /// 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,
+    /// Stream ID generator.
+    next_stream_id: AtomicUsize,
 }
 
-/// The stream ID. Reuses the inner client ID.
-type StreamID = ClientID;
+type StreamID = usize;
+type RequestSender = Sender<(page_api::GetPageRequest, ResponseSender)>;
+type RequestReceiver = Receiver<(page_api::GetPageRequest, ResponseSender)>;
+type ResponseSender = oneshot::Sender<tonic::Result<page_api::GetPageResponse>>;
 
-/// A pooled stream.
 struct StreamEntry {
-    /// The bidirectional stream.
-    stream: BiStream,
-    /// The time when this stream was last used, i.e. when it was put back into `StreamPool::idle`.
-    idle_since: Instant,
-}
-
-/// A bidirectional GetPage stream and its client. Can send requests and receive responses.
-struct BiStream {
-    /// The owning client. Holds onto the channel slot while the stream is alive.
-    client: ClientGuard,
-    /// Stream for sending requests. Uses a watch channel, so it can only send a single request at a
-    /// time, and the caller must await the response before sending another request. This is
-    /// enforced by `StreamGuard::send`.
-    sender: watch::Sender<page_api::GetPageRequest>,
-    /// Stream for receiving responses.
-    receiver: Pin<Box<dyn Stream<Item = tonic::Result<page_api::GetPageResponse>> + Send>>,
+    /// 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>,
 }
 
 impl StreamPool {
-    /// Creates a new stream pool, using the given client pool. It will use up to `max_streams`
-    /// concurrent streams.
+    /// 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.
     ///
     /// 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>>) -> Arc<Self> {
+    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 {
             client_pool,
-            idle: Mutex::default(),
-            limiter: max_streams.map(|max_streams| Arc::new(Semaphore::new(max_streams.get()))),
+            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),
+            next_stream_id: AtomicUsize::default(),
         });
         pool.idle_reaper.spawn(&pool);
         pool
     }
 
-    /// Acquires an available stream from the pool, or spins up a new stream if all streams are
-    /// full. Returns a guard that can be used to send requests and await the responses. Blocks if
-    /// the pool is full.
+    /// Acquires an available stream from the pool, or spins up a new stream async if all streams
+    /// are full. Returns a guard that can be used to send a single request on the stream and await
+    /// the response, with queue depth quota already acquired. Blocks if the pool is at capacity
+    /// (i.e. `CLIENT_LIMIT * STREAM_QUEUE_DEPTH` requests in flight).
     ///
     /// This is very performance-sensitive, as it is on the GetPage hot path.
     ///
-    /// TODO: is a `Mutex<BTreeMap>` performant enough? Will it become too contended? We can't
-    /// trivially use e.g. DashMap or sharding, because we want to pop lower-ordered streams first
-    /// to free up higher-ordered channels.
-    pub async fn get(self: &Arc<Self>) -> tonic::Result<StreamGuard> {
+    /// TODO: this must do something more sophisticated for performance. We want:
+    ///
+    /// * Cheap, concurrent access in the common case where we can use a pooled stream.
+    /// * Quick acquisition of pooled streams with available capacity.
+    /// * Prefer streams that belong to lower-numbered channels, to reap idle channels.
+    /// * Prefer filling up existing streams' queue depth before spinning up new streams.
+    /// * Don't hold a lock while spinning up new streams.
+    /// * 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"));
         }
+        let mut streams = self.streams.lock().unwrap();
 
-        // Fast path: acquire an idle stream from the pool.
-        if let Some((_, entry)) = self.idle.lock().unwrap().pop_first() {
-            return Ok(StreamGuard {
-                pool: Arc::downgrade(self),
-                stream: Some(entry.stream),
-                can_reuse: true,
-                permit,
-            });
+        // Look for a pooled stream with available capacity.
+        for (&id, entry) in streams.iter_mut() {
+            assert!(
+                entry.queue_depth <= self.max_queue_depth.get(),
+                "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;
+                return StreamGuard {
+                    pool: Arc::downgrade(self),
+                    id,
+                    sender: entry.sender.clone(),
+                    permit,
+                };
+            }
         }
 
-        // Spin up a new stream. Uses a watch channel to send a single request at a time, since
-        // `StreamGuard::send` enforces this anyway and it avoids unnecessary channel overhead.
-        let mut client = self.client_pool.get().await?;
+        // No available stream, spin up a new one. We install the stream entry in the pool first and
+        // 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 entry = StreamEntry {
+            sender: req_tx.clone(),
+            queue_depth: 1, // reserve quota for this caller
+            idle_since: None,
+        };
+        streams.insert(id, entry);
 
-        let (req_tx, req_rx) = watch::channel(page_api::GetPageRequest::default());
-        let req_stream = WatchStream::from_changes(req_rx);
-        let resp_stream = client.get_pages(req_stream).await?;
+        if let Some(max_streams) = self.max_streams {
+            assert!(streams.len() <= max_streams.get(), "stream overflow");
+        };
 
-        Ok(StreamGuard {
+        let client_pool = self.client_pool.clone();
+        let pool = Arc::downgrade(self);
+
+        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}");
+            }
+        });
+
+        StreamGuard {
             pool: Arc::downgrade(self),
-            stream: Some(BiStream {
-                client,
-                sender: req_tx,
-                receiver: Box::pin(resp_stream),
-            }),
-            can_reuse: true,
+            id,
+            sender: req_tx,
             permit,
-        })
+        }
+    }
+
+    /// Runs a stream task. This acquires a client from the `ClientPool` and establishes a
+    /// bidirectional GetPage stream, then forwards requests and responses between callers and the
+    /// stream. It does not track or enforce queue depths -- that's done by `get()` since it must be
+    /// atomic with pool stream acquisition.
+    ///
+    /// The task exits when the request channel is closed, or on a stream error. The caller is
+    /// responsible for removing the stream from the pool on exit.
+    async fn run_stream(
+        client_pool: Arc<ClientPool>,
+        mut caller_rx: RequestReceiver,
+    ) -> anyhow::Result<()> {
+        // 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 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();
+
+        // Process requests and responses.
+        loop {
+            tokio::select! {
+                // Receive requests from callers and send them to the stream.
+                req = caller_rx.recv() => {
+                    // Shut down if request channel is closed.
+                    let Some((req, resp_tx)) = req else {
+                        return Ok(());
+                    };
+
+                    // Store the response channel by request ID.
+                    if callers.contains_key(&req.request_id) {
+                        // Error on request ID duplicates. Ignore callers that went away.
+                        _ = resp_tx.send(Err(tonic::Status::invalid_argument(
+                            format!("duplicate request ID: {}", req.request_id),
+                        )));
+                        continue;
+                    }
+                    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(|_| {
+                        tonic::Status::unavailable("stream closed")
+                    })?;
+                }
+
+                // Receive responses from the stream and send them to callers.
+                resp = resp_stream.next() => {
+                    // Shut down if the stream is closed, and bail out on stream errors.
+                    let Some(resp) = resp.transpose()? else {
+                        return Ok(())
+                    };
+
+                    // Send the response to the caller. Ignore errors if the caller went away.
+                    let Some(resp_tx) = callers.remove(&resp.request_id) else {
+                        warn!("received response for unknown request ID: {}", resp.request_id);
+                        continue;
+                    };
+                    _ = resp_tx.send(Ok(resp));
+                }
+            }
+        }
     }
 }
 
 impl Reapable for StreamPool {
     /// Reaps streams that have been idle since before the cutoff.
     fn reap_idle(&self, cutoff: Instant) {
-        self.idle
-            .lock()
-            .unwrap()
-            .retain(|_, entry| entry.idle_since >= cutoff);
+        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 stream acquired from the pool. Returned to the pool when dropped, unless there are still
-/// in-flight requests on the stream, or the stream failed.
+/// 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>,
-    stream: Option<BiStream>,             // Some until dropped
-    can_reuse: bool,                      // returned to pool if true
+    id: StreamID,
+    sender: RequestSender,
     permit: Option<OwnedSemaphorePermit>, // None if pool is unbounded
 }
 
 impl StreamGuard {
-    /// Sends a request on the stream and awaits the response. If the future is dropped before it
-    /// resolves (e.g. due to a timeout or cancellation), the stream will be closed to cancel the
-    /// request and is not returned to the pool. The same is true if the stream errors, in which
-    /// case the caller can't send further requests on the stream.
+    /// Sends a request on the stream and awaits the response. Consumes the guard, since it's only
+    /// valid for a single request (to enforce queue depth). This also drops the guard on return and
+    /// returns the queue depth quota to the pool.
     ///
-    /// We only support sending a single request at a time, to eliminate head-of-line blocking. See
-    /// module documentation for details.
+    /// The `GetPageRequest::request_id` must be unique across in-flight requests.
     ///
     /// NB: errors are often returned as `GetPageResponse::status_code` instead of `tonic::Status`
     /// to avoid tearing down the stream for per-request errors. Callers must check this.
     pub async fn send(
-        &mut self,
+        self,
         req: page_api::GetPageRequest,
     ) -> tonic::Result<page_api::GetPageResponse> {
-        let req_id = req.request_id;
-        let stream = self.stream.as_mut().expect("not dropped");
+        let (resp_tx, resp_rx) = oneshot::channel();
 
-        // Mark the stream as not reusable while the request is in flight. We can't return the
-        // stream to the pool until we receive the response, to avoid head-of-line blocking and
-        // stale responses. Failed streams can't be reused either.
-        if !self.can_reuse {
-            return Err(tonic::Status::internal("stream can't be reused"));
-        }
-        self.can_reuse = false;
-
-        // Send the request and receive the response.
-        //
-        // NB: this uses a watch channel, so it's unsafe to change this code to pipeline requests.
-        stream
-            .sender
-            .send(req)
+        self.sender
+            .send((req, resp_tx))
+            .await
             .map_err(|_| tonic::Status::unavailable("stream closed"))?;
 
-        let resp = stream
-            .receiver
-            .next()
+        resp_rx
             .await
-            .ok_or_else(|| tonic::Status::unavailable("stream closed"))??;
-
-        if resp.request_id != req_id {
-            return Err(tonic::Status::internal(format!(
-                "response ID {} does not match request ID {}",
-                resp.request_id, req_id
-            )));
-        }
-
-        // Success, mark the stream as reusable.
-        self.can_reuse = true;
-
-        Ok(resp)
+            .map_err(|_| tonic::Status::unavailable("stream closed"))?
     }
 }
 
@@ -585,21 +697,26 @@ impl Drop for StreamGuard {
             return; // pool was dropped
         };
 
-        // If the stream isn't reusable, it can't be returned to the pool.
-        if !self.can_reuse {
-            return;
+        // 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
         }
 
-        // Place the idle stream back into the pool.
-        let entry = StreamEntry {
-            stream: self.stream.take().expect("dropped once"),
-            idle_since: Instant::now(),
-        };
-        pool.idle
-            .lock()
-            .unwrap()
-            .insert(entry.stream.client.id, entry);
-
         _ = self.permit; // returned on drop, referenced for visibility
     }
 }

pageserver/client_grpc/src/retry.rs

@@ -1,6 +1,5 @@
 use std::time::Duration;
 
-use futures::future::pending;
 use tokio::time::Instant;
 use tracing::{error, info, warn};
 
@@ -9,54 +8,60 @@ use utils::backoff::exponential_backoff_duration;
 /// A retry handler for Pageserver gRPC requests.
 ///
 /// This is used instead of backoff::retry for better control and observability.
-pub struct Retry {
-    /// Timeout across all retry attempts. If None, retries forever.
-    pub timeout: Option<Duration>,
-    /// The initial backoff duration. The first retry does not use a backoff.
-    pub base_backoff: Duration,
-    /// The maximum backoff duration.
-    pub max_backoff: Duration,
-}
+pub struct Retry;
 
 impl Retry {
-    /// Runs the given async closure with timeouts and retries (exponential backoff). Logs errors,
-    /// using the current tracing span for context.
+    /// The per-request timeout.
+    // TODO: tune these, and/or make them configurable. Should we retry forever?
+    const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
+    /// The total timeout across all attempts
+    const TOTAL_TIMEOUT: Duration = Duration::from_secs(60);
+    /// The initial backoff duration.
+    const BASE_BACKOFF: Duration = Duration::from_millis(10);
+    /// The maximum backoff duration.
+    const MAX_BACKOFF: Duration = Duration::from_secs(10);
+    /// 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.
     ///
-    /// Only certain gRPC status codes are retried, see [`Self::should_retry`].
+    /// 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, // pass attempt number, starting at 0
+        F: FnMut(usize) -> O, // takes attempt number, starting at 0
         O: Future<Output = tonic::Result<T>>,
     {
         let started = Instant::now();
-        let deadline = self.timeout.map(|timeout| started + timeout);
+        let deadline = started + Self::TOTAL_TIMEOUT;
         let mut last_error = None;
         let mut retries = 0;
         loop {
-            // Set up a future to wait for the backoff, if any, and run the closure.
+            // Set up a future to wait for the backoff (if any) and run the request with a timeout.
             let backoff_and_try = async {
                 // NB: sleep() always sleeps 1ms, even when given a 0 argument. See:
                 // https://github.com/tokio-rs/tokio/issues/6866
-                if let Some(backoff) = self.backoff_duration(retries) {
+                if let Some(backoff) = Self::backoff_duration(retries) {
                     tokio::time::sleep(backoff).await;
                 }
 
-                f(retries).await
+                let request_started = Instant::now();
+                tokio::time::timeout(Self::REQUEST_TIMEOUT, f(retries))
+                    .await
+                    .map_err(|_| {
+                        tonic::Status::deadline_exceeded(format!(
+                            "request timed out after {:.3}s",
+                            request_started.elapsed().as_secs_f64()
+                        ))
+                    })?
             };
 
-            // Set up a future for the timeout, if any.
-            let timeout = async {
-                match deadline {
-                    Some(deadline) => tokio::time::sleep_until(deadline).await,
-                    None => pending().await,
-                }
-            };
-
-            // Wait for the backoff and request, or bail out if the timeout is exceeded.
+            // Wait for the backoff and request, or bail out if the total timeout is exceeded.
             let result = tokio::select! {
                 result = backoff_and_try => result,
 
-                _ = timeout => {
+                _ = tokio::time::sleep_until(deadline) => {
                     let last_error = last_error.unwrap_or_else(|| {
                         tonic::Status::deadline_exceeded(format!(
                             "request timed out after {:.3}s",
@@ -74,7 +79,7 @@ impl Retry {
             match result {
                 // Success, return the result.
                 Ok(result) => {
-                    if retries > 0 {
+                    if retries > 0 || Self::LOG_SUCCESS {
                         info!(
                             "request succeeded after {retries} retries in {:.3}s",
                             started.elapsed().as_secs_f64(),
@@ -107,13 +112,12 @@ impl Retry {
         }
     }
 
-    /// Returns the backoff duration for the given retry attempt, or None for no backoff. The first
-    /// attempt and first retry never backs off, so this returns None for 0 and 1 retries.
-    fn backoff_duration(&self, retries: usize) -> Option<Duration> {
+    /// Returns the backoff duration for the given retry attempt, or None for no backoff.
+    fn backoff_duration(retry: usize) -> Option<Duration> {
         let backoff = exponential_backoff_duration(
-            (retries as u32).saturating_sub(1), // first retry does not back off
-            self.base_backoff.as_secs_f64(),
-            self.max_backoff.as_secs_f64(),
+            retry as u32,
+            Self::BASE_BACKOFF.as_secs_f64(),
+            Self::MAX_BACKOFF.as_secs_f64(),
         );
         (!backoff.is_zero()).then_some(backoff)
     }

pageserver/page_api/src/model.rs

@@ -49,7 +49,7 @@ impl From<ProtocolError> for tonic::Status {
 }
 
 /// The LSN a request should read at.
-#[derive(Clone, Copy, Debug, Default)]
+#[derive(Clone, Copy, Debug)]
 pub struct ReadLsn {
     /// The request's read LSN.
     pub request_lsn: Lsn,
@@ -329,7 +329,7 @@ impl From<GetDbSizeResponse> for proto::GetDbSizeResponse {
 }
 
 /// Requests one or more pages.
-#[derive(Clone, Debug, Default)]
+#[derive(Clone, Debug)]
 pub struct GetPageRequest {
     /// A request ID. Will be included in the response. Should be unique for in-flight requests on
     /// the stream.
@@ -430,13 +430,12 @@ impl From<RequestID> for proto::RequestId {
 }
 
 /// A GetPage request class.
-#[derive(Clone, Copy, Debug, Default, strum_macros::Display)]
+#[derive(Clone, Copy, Debug, strum_macros::Display)]
 pub enum GetPageClass {
     /// Unknown class. For backwards compatibility: used when an older client version sends a class
     /// that a newer server version has removed.
     Unknown,
     /// A normal request. This is the default.
-    #[default]
     Normal,
     /// A prefetch request. NB: can only be classified on pg < 18.
     Prefetch,
@@ -444,6 +443,19 @@ 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 {

pageserver/pagebench/Cargo.toml

@@ -16,7 +16,6 @@ futures.workspace = true
 hdrhistogram.workspace = true
 humantime.workspace = true
 humantime-serde.workspace = true
-pprof.workspace = true
 rand.workspace = true
 reqwest.workspace = true
 serde.workspace = true

pageserver/pagebench/src/cmd/idle_streams.rs

@@ -1,127 +0,0 @@
-use std::sync::Arc;
-
-use anyhow::anyhow;
-use futures::StreamExt;
-use tonic::transport::Endpoint;
-use tracing::info;
-
-use pageserver_page_api::{GetPageClass, GetPageRequest, GetPageStatusCode, ReadLsn, RelTag};
-use utils::id::TenantTimelineId;
-use utils::lsn::Lsn;
-use utils::shard::ShardIndex;
-
-/// Starts a large number of idle gRPC GetPage streams.
-#[derive(clap::Parser)]
-pub(crate) struct Args {
-    /// The Pageserver to connect to. Must use grpc://.
-    #[clap(long, default_value = "grpc://localhost:51051")]
-    server: String,
-    /// The Pageserver HTTP API.
-    #[clap(long, default_value = "http://localhost:9898")]
-    http_server: String,
-    /// The number of streams to open.
-    #[clap(long, default_value = "100000")]
-    count: usize,
-    /// Number of streams per connection.
-    #[clap(long, default_value = "100")]
-    per_connection: usize,
-    /// Send a single GetPage request on each stream.
-    #[clap(long, default_value_t = false)]
-    send_request: bool,
-}
-
-pub(crate) fn main(args: Args) -> anyhow::Result<()> {
-    let rt = tokio::runtime::Builder::new_multi_thread()
-        .enable_all()
-        .build()?;
-
-    rt.block_on(main_impl(args))
-}
-
-async fn main_impl(args: Args) -> anyhow::Result<()> {
-    // Discover a tenant and timeline to use.
-    let mgmt_api_client = Arc::new(pageserver_client::mgmt_api::Client::new(
-        reqwest::Client::new(),
-        args.http_server.clone(),
-        None,
-    ));
-    let timelines: Vec<TenantTimelineId> = crate::util::cli::targets::discover(
-        &mgmt_api_client,
-        crate::util::cli::targets::Spec {
-            limit_to_first_n_targets: Some(1),
-            targets: None,
-        },
-    )
-    .await?;
-    let ttid = timelines
-        .first()
-        .ok_or_else(|| anyhow!("no timelines found"))?;
-
-    // Set up the initial client.
-    let endpoint = Endpoint::from_shared(args.server.clone())?;
-
-    let connect = async || {
-        pageserver_page_api::Client::new(
-            endpoint.connect().await?,
-            ttid.tenant_id,
-            ttid.timeline_id,
-            ShardIndex::unsharded(),
-            None,
-            None,
-        )
-    };
-
-    let mut client = connect().await?;
-    let mut streams = Vec::with_capacity(args.count);
-
-    // Create streams.
-    for i in 0..args.count {
-        if i % 100 == 0 {
-            info!("opened {}/{} streams", i, args.count);
-        }
-        if i % args.per_connection == 0 && i > 0 {
-            client = connect().await?;
-        }
-
-        let (req_tx, req_rx) = tokio::sync::mpsc::unbounded_channel();
-        let req_stream = tokio_stream::wrappers::UnboundedReceiverStream::new(req_rx);
-        let mut resp_stream = client.get_pages(req_stream).await?;
-
-        // Send request if specified.
-        if args.send_request {
-            req_tx.send(GetPageRequest {
-                request_id: 1.into(),
-                request_class: GetPageClass::Normal,
-                read_lsn: ReadLsn {
-                    request_lsn: Lsn::MAX,
-                    not_modified_since_lsn: Some(Lsn(1)),
-                },
-                rel: RelTag {
-                    spcnode: 1664, // pg_global
-                    dbnode: 0,     // shared database
-                    relnode: 1262, // pg_authid
-                    forknum: 0,    // init
-                },
-                block_numbers: vec![0],
-            })?;
-            let resp = resp_stream
-                .next()
-                .await
-                .transpose()?
-                .ok_or_else(|| anyhow!("no response"))?;
-            if resp.status_code != GetPageStatusCode::Ok {
-                return Err(anyhow!("{} response", resp.status_code));
-            }
-        }
-
-        // Hold onto streams to avoid closing them.
-        streams.push((req_tx, resp_stream));
-    }
-
-    info!("opened {} streams, sleeping", args.count);
-
-    // Block forever, to hold the idle streams open for inspection.
-    futures::future::pending::<()>().await;
-
-    Ok(())
-}

pageserver/pagebench/src/main.rs

@@ -1,7 +1,4 @@
-use std::fs::File;
-
 use clap::Parser;
-use tracing::info;
 use utils::logging;
 
 /// Re-usable pieces of code that aren't CLI-specific.
@@ -20,73 +17,38 @@ mod cmd {
     pub(super) mod aux_files;
     pub(super) mod basebackup;
     pub(super) mod getpage_latest_lsn;
-    pub(super) mod idle_streams;
     pub(super) mod ondemand_download_churn;
     pub(super) mod trigger_initial_size_calculation;
 }
 
 /// Component-level performance test for pageserver.
 #[derive(clap::Parser)]
-struct Args {
-    /// Takes a client CPU profile into profile.svg. The benchmark must exit cleanly before it's
-    /// written, e.g. via --runtime.
-    #[arg(long)]
-    profile: bool,
-
-    #[command(subcommand)]
-    subcommand: Subcommand,
-}
-
-#[derive(clap::Subcommand)]
-enum Subcommand {
+enum Args {
     Basebackup(cmd::basebackup::Args),
     GetPageLatestLsn(cmd::getpage_latest_lsn::Args),
     TriggerInitialSizeCalculation(cmd::trigger_initial_size_calculation::Args),
     OndemandDownloadChurn(cmd::ondemand_download_churn::Args),
     AuxFiles(cmd::aux_files::Args),
-    IdleStreams(cmd::idle_streams::Args),
 }
 
-fn main() -> anyhow::Result<()> {
+fn main() {
     logging::init(
         logging::LogFormat::Plain,
         logging::TracingErrorLayerEnablement::Disabled,
         logging::Output::Stderr,
-    )?;
+    )
+    .unwrap();
     logging::replace_panic_hook_with_tracing_panic_hook().forget();
 
     let args = Args::parse();
-
-    // Start a CPU profile if requested.
-    let mut profiler = None;
-    if args.profile {
-        profiler = Some(
-            pprof::ProfilerGuardBuilder::default()
-                .frequency(1000)
-                .blocklist(&["libc", "libgcc", "pthread", "vdso"])
-                .build()?,
-        );
-    }
-
-    match args.subcommand {
-        Subcommand::Basebackup(args) => cmd::basebackup::main(args),
-        Subcommand::GetPageLatestLsn(args) => cmd::getpage_latest_lsn::main(args),
-        Subcommand::TriggerInitialSizeCalculation(args) => {
+    match args {
+        Args::Basebackup(args) => cmd::basebackup::main(args),
+        Args::GetPageLatestLsn(args) => cmd::getpage_latest_lsn::main(args),
+        Args::TriggerInitialSizeCalculation(args) => {
             cmd::trigger_initial_size_calculation::main(args)
         }
-        Subcommand::OndemandDownloadChurn(args) => cmd::ondemand_download_churn::main(args),
-        Subcommand::AuxFiles(args) => cmd::aux_files::main(args),
-        Subcommand::IdleStreams(args) => cmd::idle_streams::main(args),
-    }?;
-
-    // Generate a CPU flamegraph if requested.
-    if let Some(profiler) = profiler {
-        let report = profiler.report().build()?;
-        drop(profiler); // stop profiling
-        let file = File::create("profile.svg")?;
-        report.flamegraph(file)?;
-        info!("wrote CPU profile flamegraph to profile.svg")
+        Args::OndemandDownloadChurn(args) => cmd::ondemand_download_churn::main(args),
+        Args::AuxFiles(args) => cmd::aux_files::main(args),
     }
-
-    Ok(())
+    .unwrap()
 }

pageserver/src/controller_upcall_client.rs

@@ -194,7 +194,6 @@ impl StorageControllerUpcallApi for StorageControllerUpcallClient {
                         listen_http_port: m.http_port,
                         listen_https_port: m.https_port,
                         availability_zone_id: az_id.expect("Checked above"),
-                        node_ip_addr: None,
                     })
                 }
                 Err(e) => {

pageserver/src/http/routes.rs

@@ -10,7 +10,6 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use anyhow::{Context, Result, anyhow};
-use bytes::Bytes;
 use enumset::EnumSet;
 use futures::future::join_all;
 use futures::{StreamExt, TryFutureExt};
@@ -47,7 +46,6 @@ use pageserver_api::shard::{ShardCount, TenantShardId};
 use postgres_ffi::PgMajorVersion;
 use remote_storage::{DownloadError, GenericRemoteStorage, TimeTravelError};
 use scopeguard::defer;
-use serde::{Deserialize, Serialize};
 use serde_json::json;
 use tenant_size_model::svg::SvgBranchKind;
 use tenant_size_model::{SizeResult, StorageModel};
@@ -59,7 +57,6 @@ use utils::auth::SwappableJwtAuth;
 use utils::generation::Generation;
 use utils::id::{TenantId, TimelineId};
 use utils::lsn::Lsn;
-use wal_decoder::models::record::NeonWalRecord;
 
 use crate::config::PageServerConf;
 use crate::context;
@@ -80,7 +77,6 @@ use crate::tenant::remote_timeline_client::{
 };
 use crate::tenant::secondary::SecondaryController;
 use crate::tenant::size::ModelInputs;
-use crate::tenant::storage_layer::ValuesReconstructState;
 use crate::tenant::storage_layer::{IoConcurrency, LayerAccessStatsReset, LayerName};
 use crate::tenant::timeline::layer_manager::LayerManagerLockHolder;
 use crate::tenant::timeline::offload::{OffloadError, offload_timeline};
@@ -2712,16 +2708,6 @@ async fn deletion_queue_flush(
     }
 }
 
-/// Try if `GetPage@Lsn` is successful, useful for manual debugging.
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
-struct GetPageResponse {
-    pub page: Bytes,
-    pub layers_visited: u32,
-    pub delta_layers_visited: u32,
-    pub records: Vec<(Lsn, NeonWalRecord)>,
-    pub img: Option<(Lsn, Bytes)>,
-}
-
 async fn getpage_at_lsn_handler(
     request: Request<Body>,
     cancel: CancellationToken,
@@ -2772,24 +2758,21 @@ async fn getpage_at_lsn_handler_inner(
 
         // Use last_record_lsn if no lsn is provided
         let lsn = lsn.unwrap_or_else(|| timeline.get_last_record_lsn());
+        let page = timeline.get(key.0, lsn, &ctx).await?;
 
         if touch {
             json_response(StatusCode::OK, ())
         } else {
-            let mut reconstruct_state = ValuesReconstructState::new_with_debug(IoConcurrency::sequential());
-            let page = timeline.debug_get(key.0, lsn, &ctx, &mut reconstruct_state).await?;
-            let response = GetPageResponse {
-                page,
-                layers_visited: reconstruct_state.get_layers_visited(),
-                delta_layers_visited: reconstruct_state.get_delta_layers_visited(),
-                records: reconstruct_state.debug_state.records.clone(),
-                img: reconstruct_state.debug_state.img.clone(),
-            };
-
-            json_response(StatusCode::OK, response)
+            Result::<_, ApiError>::Ok(
+                Response::builder()
+                    .status(StatusCode::OK)
+                    .header(header::CONTENT_TYPE, "application/octet-stream")
+                    .body(hyper::Body::from(page))
+                    .unwrap(),
+            )
         }
     }
-    .instrument(info_span!("timeline_debug_get", tenant_id = %tenant_shard_id.tenant_id, shard_id = %tenant_shard_id.shard_slug(), %timeline_id))
+    .instrument(info_span!("timeline_get", tenant_id = %tenant_shard_id.tenant_id, shard_id = %tenant_shard_id.shard_slug(), %timeline_id))
     .await
 }
 

pageserver/src/page_service.rs

@@ -2167,7 +2167,7 @@ impl PageServerHandler {
     fn effective_request_lsn(
         timeline: &Timeline,
         last_record_lsn: Lsn,
-        request_lsn: Lsn,
+        mut request_lsn: Lsn,
         not_modified_since: Lsn,
         latest_gc_cutoff_lsn: &RcuReadGuard<Lsn>,
     ) -> Result<Lsn, PageStreamError> {
@@ -2195,12 +2195,16 @@ impl PageServerHandler {
         if request_lsn < **latest_gc_cutoff_lsn && !timeline.is_gc_blocked_by_lsn_lease_deadline() {
             let gc_info = &timeline.gc_info.read().unwrap();
             if !gc_info.lsn_covered_by_lease(request_lsn) {
-                return Err(
-                    PageStreamError::BadRequest(format!(
-                        "tried to request a page version that was garbage collected. requested at {} gc cutoff {}",
-                        request_lsn, **latest_gc_cutoff_lsn
-                    ).into())
+                // While request was in flight, replica apply_lsn may be advanced.
+                // latest_gc_cutoff_lsn is conservative estimation for min(redo_lsn) for all replicas,
+                // so it is safe to move request_lsn forward to latest_gc_cutoff_lsn.
+                // If replica lease is expired and latest_gc_cutoff_lsn>redo_lsn for this replica,
+                // then check of page LSN at replia protects it from getting too new version of the page.
+                warn!(
+                    "Tried to request a page version that was garbage collected. requested at {} gc cutoff {}",
+                    request_lsn, **latest_gc_cutoff_lsn
                 );
+                request_lsn = **latest_gc_cutoff_lsn;
             }
         }
 

pageserver/src/tenant.rs

@@ -3393,13 +3393,7 @@ impl TenantShard {
                 .collect_vec();
 
             for timeline in timelines {
-                // Include a span with the timeline ID. The parent span already has the tenant ID.
-                let span =
-                    info_span!("maybe_freeze_ephemeral_layer", timeline_id = %timeline.timeline_id);
-                timeline
-                    .maybe_freeze_ephemeral_layer()
-                    .instrument(span)
-                    .await;
+                timeline.maybe_freeze_ephemeral_layer().await;
             }
         }
 

pageserver/src/tenant/storage_layer.rs

@@ -75,7 +75,7 @@ where
 /// the same ValueReconstructState struct in the next 'get_value_reconstruct_data'
 /// call, to collect more records.
 ///
-#[derive(Debug, Default, Clone)]
+#[derive(Debug, Default)]
 pub(crate) struct ValueReconstructState {
     pub(crate) records: Vec<(Lsn, NeonWalRecord)>,
     pub(crate) img: Option<(Lsn, Bytes)>,
@@ -308,9 +308,6 @@ pub struct ValuesReconstructState {
     layers_visited: u32,
     delta_layers_visited: u32,
 
-    pub(crate) enable_debug: bool,
-    pub(crate) debug_state: ValueReconstructState,
-
     pub(crate) io_concurrency: IoConcurrency,
     num_active_ios: Arc<AtomicUsize>,
 
@@ -660,23 +657,6 @@ impl ValuesReconstructState {
             layers_visited: 0,
             delta_layers_visited: 0,
             io_concurrency,
-            enable_debug: false,
-            debug_state: ValueReconstructState::default(),
-            num_active_ios: Arc::new(AtomicUsize::new(0)),
-            read_path: None,
-        }
-    }
-
-    pub(crate) fn new_with_debug(io_concurrency: IoConcurrency) -> Self {
-        Self {
-            keys: HashMap::new(),
-            keys_done: KeySpaceRandomAccum::new(),
-            keys_with_image_coverage: None,
-            layers_visited: 0,
-            delta_layers_visited: 0,
-            io_concurrency,
-            enable_debug: true,
-            debug_state: ValueReconstructState::default(),
             num_active_ios: Arc::new(AtomicUsize::new(0)),
             read_path: None,
         }
@@ -690,12 +670,6 @@ impl ValuesReconstructState {
         self.io_concurrency.spawn_io(fut).await;
     }
 
-    pub(crate) fn set_debug_state(&mut self, debug_state: &ValueReconstructState) {
-        if self.enable_debug {
-            self.debug_state = debug_state.clone();
-        }
-    }
-
     pub(crate) fn on_layer_visited(&mut self, layer: &ReadableLayer) {
         self.layers_visited += 1;
         if let ReadableLayer::PersistentLayer(layer) = layer {

pageserver/src/tenant/timeline.rs

@@ -1253,57 +1253,6 @@ impl Timeline {
         }
     }
 
-    #[inline(always)]
-    pub(crate) async fn debug_get(
-        &self,
-        key: Key,
-        lsn: Lsn,
-        ctx: &RequestContext,
-        reconstruct_state: &mut ValuesReconstructState,
-    ) -> Result<Bytes, PageReconstructError> {
-        if !lsn.is_valid() {
-            return Err(PageReconstructError::Other(anyhow::anyhow!("Invalid LSN")));
-        }
-
-        // This check is debug-only because of the cost of hashing, and because it's a double-check: we
-        // already checked the key against the shard_identity when looking up the Timeline from
-        // page_service.
-        debug_assert!(!self.shard_identity.is_key_disposable(&key));
-
-        let query = VersionedKeySpaceQuery::uniform(KeySpace::single(key..key.next()), lsn);
-        let vectored_res = self
-            .debug_get_vectored_impl(query, reconstruct_state, ctx)
-            .await;
-
-        let key_value = vectored_res?.pop_first();
-        match key_value {
-            Some((got_key, value)) => {
-                if got_key != key {
-                    error!(
-                        "Expected {}, but singular vectored get returned {}",
-                        key, got_key
-                    );
-                    Err(PageReconstructError::Other(anyhow!(
-                        "Singular vectored get returned wrong key"
-                    )))
-                } else {
-                    value
-                }
-            }
-            None => Err(PageReconstructError::MissingKey(Box::new(
-                MissingKeyError {
-                    keyspace: KeySpace::single(key..key.next()),
-                    shard: self.shard_identity.get_shard_number(&key),
-                    original_hwm_lsn: lsn,
-                    ancestor_lsn: None,
-                    backtrace: None,
-                    read_path: None,
-                    query: None,
-                },
-            ))),
-        }
-    }
-
     pub(crate) const LAYERS_VISITED_WARN_THRESHOLD: u32 = 100;
 
     /// Look up multiple page versions at a given LSN
@@ -1598,98 +1547,6 @@ impl Timeline {
         Ok(results)
     }
 
-    // A copy of the get_vectored_impl method except that we store the image and wal records into `reconstruct_state`.
-    // This is only used in the http getpage call for debugging purpose.
-    pub(super) async fn debug_get_vectored_impl(
-        &self,
-        query: VersionedKeySpaceQuery,
-        reconstruct_state: &mut ValuesReconstructState,
-        ctx: &RequestContext,
-    ) -> Result<BTreeMap<Key, Result<Bytes, PageReconstructError>>, GetVectoredError> {
-        if query.is_empty() {
-            return Ok(BTreeMap::default());
-        }
-
-        let read_path = if self.conf.enable_read_path_debugging || ctx.read_path_debug() {
-            Some(ReadPath::new(
-                query.total_keyspace(),
-                query.high_watermark_lsn()?,
-            ))
-        } else {
-            None
-        };
-
-        reconstruct_state.read_path = read_path;
-
-        let traversal_res: Result<(), _> = self
-            .get_vectored_reconstruct_data(query.clone(), reconstruct_state, ctx)
-            .await;
-
-        if let Err(err) = traversal_res {
-            // Wait for all the spawned IOs to complete.
-            // See comments on `spawn_io` inside `storage_layer` for more details.
-            let mut collect_futs = std::mem::take(&mut reconstruct_state.keys)
-                .into_values()
-                .map(|state| state.collect_pending_ios())
-                .collect::<FuturesUnordered<_>>();
-            while collect_futs.next().await.is_some() {}
-            return Err(err);
-        };
-
-        let reconstruct_state = Arc::new(Mutex::new(reconstruct_state));
-        let futs = FuturesUnordered::new();
-
-        for (key, state) in std::mem::take(&mut reconstruct_state.lock().unwrap().keys) {
-            let req_lsn_for_key = query.map_key_to_lsn(&key);
-            futs.push({
-                let walredo_self = self.myself.upgrade().expect("&self method holds the arc");
-                let rc_clone = Arc::clone(&reconstruct_state);
-
-                async move {
-                    assert_eq!(state.situation, ValueReconstructSituation::Complete);
-
-                    let converted = match state.collect_pending_ios().await {
-                        Ok(ok) => ok,
-                        Err(err) => {
-                            return (key, Err(err));
-                        }
-                    };
-                    DELTAS_PER_READ_GLOBAL.observe(converted.num_deltas() as f64);
-
-                    // The walredo module expects the records to be descending in terms of Lsn.
-                    // And we submit the IOs in that order, so, there shuold be no need to sort here.
-                    debug_assert!(
-                        converted
-                            .records
-                            .is_sorted_by_key(|(lsn, _)| std::cmp::Reverse(*lsn)),
-                        "{converted:?}"
-                    );
-                    {
-                        let mut guard = rc_clone.lock().unwrap();
-                        guard.set_debug_state(&converted);
-                    }
-                    (
-                        key,
-                        walredo_self
-                            .reconstruct_value(
-                                key,
-                                req_lsn_for_key,
-                                converted,
-                                RedoAttemptType::ReadPage,
-                            )
-                            .await,
-                    )
-                }
-            });
-        }
-
-        let results = futs
-            .collect::<BTreeMap<Key, Result<Bytes, PageReconstructError>>>()
-            .await;
-
-        Ok(results)
-    }
-
     /// Get last or prev record separately. Same as get_last_record_rlsn().last/prev.
     pub(crate) fn get_last_record_lsn(&self) -> Lsn {
         self.last_record_lsn.load().last
@@ -2036,8 +1893,6 @@ impl Timeline {
     // an ephemeral layer open forever when idle.  It also freezes layers if the global limit on
     // ephemeral layer bytes has been breached.
     pub(super) async fn maybe_freeze_ephemeral_layer(&self) {
-        debug_assert_current_span_has_tenant_and_timeline_id();
-
         let Ok(mut write_guard) = self.write_lock.try_lock() else {
             // If the write lock is held, there is an active wal receiver: rolling open layers
             // is their responsibility while they hold this lock.

pgxn/neon/Makefile

@@ -5,7 +5,6 @@ MODULE_big = neon
 OBJS = \
 	$(WIN32RES) \
 	communicator.o \
-	communicator_process.o \
 	extension_server.o \
 	file_cache.o \
 	hll.o \
@@ -30,11 +29,6 @@ PG_CPPFLAGS = -I$(libpq_srcdir)
 SHLIB_LINK_INTERNAL = $(libpq)
 SHLIB_LINK = -lcurl
 
-UNAME_S := $(shell uname -s)
-ifeq ($(UNAME_S), Darwin)
-    SHLIB_LINK += -framework Security -framework CoreFoundation -framework SystemConfiguration
-endif
-
 EXTENSION = neon
 DATA = \
 	neon--1.0.sql \
@@ -63,8 +57,7 @@ WALPROP_OBJS = \
 
 # libcommunicator.a is built by cargo from the Rust sources under communicator/
 # subdirectory. `cargo build` also generates communicator_bindings.h.
-communicator_process.o: communicator/communicator_bindings.h
-file_cache.o: communicator/communicator_bindings.h
+neon.o: communicator/communicator_bindings.h
 
 $(NEON_CARGO_ARTIFACT_TARGET_DIR)/libcommunicator.a communicator/communicator_bindings.h &:
 	(cd $(srcdir)/communicator && cargo build $(CARGO_BUILD_FLAGS) $(CARGO_PROFILE))

pgxn/neon/communicator.c

@@ -421,7 +421,7 @@ check_getpage_response(PrefetchRequest* slot, NeonResponse* resp)
 {
 	if (resp->tag != T_NeonGetPageResponse && resp->tag != T_NeonErrorResponse)
 	{
-		neon_shard_log(slot->shard_no, PANIC, "Unexpected prefetch response %d, ring_receive=" UINT64_FORMAT ", ring_flush=" UINT64_FORMAT ", ring_unused=" UINT64_FORMAT "",
+		neon_shard_log(slot->shard_no, PANIC, "Unexpected prefetch response %d, ring_receive=%ld, ring_flush=%ld, ring_unused=%ld",
 					   resp->tag, MyPState->ring_receive, MyPState->ring_flush, MyPState->ring_unused);
 	}
 	if (neon_protocol_version >= 3)
@@ -438,7 +438,7 @@ check_getpage_response(PrefetchRequest* slot, NeonResponse* resp)
 				getpage_resp->req.blkno != slot->buftag.blockNum)
 			{
 				NEON_PANIC_CONNECTION_STATE(slot->shard_no, PANIC,
-											"Receive unexpected getpage response {reqid=" UINT64_HEX_FORMAT ",lsn=%X/%08X, since=%X/%08X, rel=%u/%u/%u.%u, block=%u} to get page request {reqid=" UINT64_HEX_FORMAT ",lsn=%X/%08X, since=%X/%08X, rel=%u/%u/%u.%u, block=%u}",
+											"Receive unexpected getpage response {reqid=%lx,lsn=%X/%08X, since=%X/%08X, rel=%u/%u/%u.%u, block=%u} to get page request {reqid=%lx,lsn=%X/%08X, since=%X/%08X, rel=%u/%u/%u.%u, block=%u}",
 											resp->reqid, LSN_FORMAT_ARGS(resp->lsn), LSN_FORMAT_ARGS(resp->not_modified_since), RelFileInfoFmt(getpage_resp->req.rinfo), getpage_resp->req.forknum, getpage_resp->req.blkno,
 											slot->reqid, LSN_FORMAT_ARGS(slot->request_lsns.request_lsn), LSN_FORMAT_ARGS(slot->request_lsns.not_modified_since), RelFileInfoFmt(rinfo), slot->buftag.forkNum, slot->buftag.blockNum);
 			}
@@ -447,7 +447,7 @@ check_getpage_response(PrefetchRequest* slot, NeonResponse* resp)
 				 resp->lsn != slot->request_lsns.request_lsn ||
 				 resp->not_modified_since != slot->request_lsns.not_modified_since)
 		{
-			elog(WARNING, NEON_TAG "Error message {reqid=" UINT64_HEX_FORMAT ",lsn=%X/%08X, since=%X/%08X} doesn't match exists request {reqid=" UINT64_HEX_FORMAT ",lsn=%X/%08X, since=%X/%08X}",
+			elog(WARNING, NEON_TAG "Error message {reqid=%lx,lsn=%X/%08X, since=%X/%08X} doesn't match exists request {reqid=%lx,lsn=%X/%08X, since=%X/%08X}",
 				 resp->reqid, LSN_FORMAT_ARGS(resp->lsn), LSN_FORMAT_ARGS(resp->not_modified_since),
 				 slot->reqid, LSN_FORMAT_ARGS(slot->request_lsns.request_lsn), LSN_FORMAT_ARGS(slot->request_lsns.not_modified_since));
 		}
@@ -496,9 +496,9 @@ communicator_prefetch_pump_state(void)
 			slot->my_ring_index != MyPState->ring_receive)
 		{
 			neon_shard_log(slot->shard_no, PANIC,
-						   "Incorrect prefetch slot state after receive: status=%d response=%p my=" UINT64_FORMAT " receive=" UINT64_FORMAT "",
+						   "Incorrect prefetch slot state after receive: status=%d response=%p my=%lu receive=%lu",
 						   slot->status, slot->response,
-						   slot->my_ring_index, MyPState->ring_receive);
+						   (long) slot->my_ring_index, (long) MyPState->ring_receive);
 		}
 		/* update prefetch state */
 		MyPState->n_responses_buffered += 1;
@@ -789,9 +789,9 @@ prefetch_read(PrefetchRequest *slot)
 		slot->my_ring_index != MyPState->ring_receive)
 	{
 		neon_shard_log(slot->shard_no, PANIC,
-					   "Incorrect prefetch read: status=%d response=%p my=" UINT64_FORMAT " receive=" UINT64_FORMAT "",
+					   "Incorrect prefetch read: status=%d response=%p my=%lu receive=%lu",
 					   slot->status, slot->response,
-					   slot->my_ring_index, MyPState->ring_receive);
+					   (long)slot->my_ring_index, (long)MyPState->ring_receive);
 	}
 
 	/*
@@ -816,9 +816,9 @@ prefetch_read(PrefetchRequest *slot)
 			slot->my_ring_index != MyPState->ring_receive)
 		{
 			neon_shard_log(shard_no, PANIC,
-						   "Incorrect prefetch slot state after receive: status=%d response=%p my=" UINT64_FORMAT " receive=" UINT64_FORMAT "",
+						   "Incorrect prefetch slot state after receive: status=%d response=%p my=%lu receive=%lu",
 						   slot->status, slot->response,
-						   slot->my_ring_index, MyPState->ring_receive);
+						   (long) slot->my_ring_index, (long) MyPState->ring_receive);
 		}
 
 		/* update prefetch state */
@@ -852,8 +852,8 @@ prefetch_read(PrefetchRequest *slot)
 		 * and the prefetch queue was flushed during the receive call
 		 */
 		neon_shard_log(shard_no, LOG,
-					   "No response from reading prefetch entry " UINT64_FORMAT ": %u/%u/%u.%u block %u. This can be caused by a concurrent disconnect",
-					   my_ring_index,
+					   "No response from reading prefetch entry %lu: %u/%u/%u.%u block %u. This can be caused by a concurrent disconnect",
+					   (long) my_ring_index,
 					   RelFileInfoFmt(BufTagGetNRelFileInfo(buftag)),
 					   buftag.forkNum, buftag.blockNum);
 		return false;
@@ -1045,6 +1045,34 @@ prefetch_do_request(PrefetchRequest *slot, neon_request_lsns *force_request_lsns
 	Assert(!found);
 }
 
+/*
+ * Check that pahge LSN returned by PS to replica is not beyand replay LSN.
+ * It can happen only in case of deteriorated lease.
+ */
+static bool
+check_page_lsn(NeonGetPageResponse* resp, XLogRecPtr* replay_lsn_ptr)
+{
+	if (RecoveryInProgress())
+	{
+		XLogRecPtr page_lsn = PageGetLSN((Page)resp->page);
+#if PG_VERSION_NUM >= 150000
+		XLogRecPtr replay_lsn = GetCurrentReplayRecPtr(NULL);
+#else
+		/*
+		 * PG14 doesn't have GetCurrentReplayRecPtr() function which returns end of currently applied record.
+		 * And GetXLogReplayRecPtr returns end of WAL records which was already applied.
+		 * So we have to use this hack with resp->req.lsn which is expected to contain end record ptr in this case.
+		 * But it works onlyfor v3 protocol version.
+		 */
+		XLogRecPtr replay_lsn = Max(GetXLogReplayRecPtr(NULL), resp->req.hdr.lsn);
+#endif
+		if (replay_lsn_ptr)
+			*replay_lsn_ptr = replay_lsn;
+		return replay_lsn == 0 || page_lsn <= replay_lsn;
+	}
+	return true;
+}
+
 /*
  * Lookup of already received prefetch requests. Only already received responses matching required LSNs are accepted.
  * Present pages are marked in "mask" bitmap and total number of such pages is returned.
@@ -1068,7 +1096,7 @@ communicator_prefetch_lookupv(NRelFileInfo rinfo, ForkNumber forknum, BlockNumbe
 	for (int i = 0; i < nblocks; i++)
 	{
 		PrfHashEntry *entry;
-
+		NeonGetPageResponse* resp;
 		hashkey.buftag.blockNum = blocknum + i;
 		entry = prfh_lookup(MyPState->prf_hash, &hashkey);
 
@@ -1101,8 +1129,16 @@ communicator_prefetch_lookupv(NRelFileInfo rinfo, ForkNumber forknum, BlockNumbe
 				continue;
 			}
 			Assert(slot->response->tag == T_NeonGetPageResponse); /* checked by check_getpage_response when response was assigned to the slot */
-			memcpy(buffers[i], ((NeonGetPageResponse*)slot->response)->page, BLCKSZ);
+			resp = (NeonGetPageResponse*)slot->response;
 
+			/*
+			 * Ignore "in-future" responses caused by deteriorated lease
+			 */
+			if (!check_page_lsn(resp, NULL))
+			{
+				continue;
+			}
+			memcpy(buffers[i], resp->page, BLCKSZ);
 
 			/*
 			 * With lfc_store_prefetch_result=true prefetch result is stored in LFC in prefetch_pump_state when response is received
@@ -1844,7 +1880,7 @@ nm_to_string(NeonMessage *msg)
 				NeonDbSizeResponse *msg_resp = (NeonDbSizeResponse *) msg;
 
 				appendStringInfoString(&s, "{\"type\": \"NeonDbSizeResponse\"");
-				appendStringInfo(&s, ", \"db_size\": " INT64_FORMAT "}",
+				appendStringInfo(&s, ", \"db_size\": %ld}",
 								 msg_resp->db_size);
 				appendStringInfoChar(&s, '}');
 
@@ -2045,7 +2081,7 @@ communicator_exists(NRelFileInfo rinfo, ForkNumber forkNum, neon_request_lsns *r
 						exists_resp->req.forknum != request.forknum)
 					{
 						NEON_PANIC_CONNECTION_STATE(0, PANIC,
-													"Unexpect response {reqid=" UINT64_HEX_FORMAT ",lsn=%X/%08X, since=%X/%08X, rel=%u/%u/%u.%u} to exits request {reqid=" UINT64_HEX_FORMAT ",lsn=%X/%08X, since=%X/%08X, rel=%u/%u/%u.%u}",
+													"Unexpect response {reqid=%lx,lsn=%X/%08X, since=%X/%08X, rel=%u/%u/%u.%u} to exits request {reqid=%lx,lsn=%X/%08X, since=%X/%08X, rel=%u/%u/%u.%u}",
 													resp->reqid, LSN_FORMAT_ARGS(resp->lsn), LSN_FORMAT_ARGS(resp->not_modified_since), RelFileInfoFmt(exists_resp->req.rinfo), exists_resp->req.forknum,
 													request.hdr.reqid, LSN_FORMAT_ARGS(request.hdr.lsn), LSN_FORMAT_ARGS(request.hdr.not_modified_since), RelFileInfoFmt(request.rinfo), request.forknum);
 					}
@@ -2058,14 +2094,14 @@ communicator_exists(NRelFileInfo rinfo, ForkNumber forkNum, neon_request_lsns *r
 				{
 					if (!equal_requests(resp, &request.hdr))
 					{
-						elog(WARNING, NEON_TAG "Error message {reqid=" UINT64_HEX_FORMAT ",lsn=%X/%08X, since=%X/%08X} doesn't match exists request {reqid=" UINT64_HEX_FORMAT ",lsn=%X/%08X, since=%X/%08X}",
+						elog(WARNING, NEON_TAG "Error message {reqid=%lx,lsn=%X/%08X, since=%X/%08X} doesn't match exists request {reqid=%lx,lsn=%X/%08X, since=%X/%08X}",
 							 resp->reqid, LSN_FORMAT_ARGS(resp->lsn), LSN_FORMAT_ARGS(resp->not_modified_since),
 							 request.hdr.reqid, LSN_FORMAT_ARGS(request.hdr.lsn), LSN_FORMAT_ARGS(request.hdr.not_modified_since));
 					}
 				}
 				ereport(ERROR,
 						(errcode(ERRCODE_IO_ERROR),
-						 errmsg(NEON_TAG "[reqid " UINT64_HEX_FORMAT "] could not read relation existence of rel %u/%u/%u.%u from page server at lsn %X/%08X",
+						 errmsg(NEON_TAG "[reqid %lx] could not read relation existence of rel %u/%u/%u.%u from page server at lsn %X/%08X",
 								resp->reqid,
 								RelFileInfoFmt(rinfo),
 								forkNum,
@@ -2227,6 +2263,15 @@ Retry:
 			case T_NeonGetPageResponse:
 			{
 				NeonGetPageResponse* getpage_resp = (NeonGetPageResponse *) resp;
+				XLogRecPtr replay_lsn;
+				if (!check_page_lsn(getpage_resp, &replay_lsn))
+				{
+					/* Alternative to throw error is to repeat the query with request_lsn=replay_lsn */
+					ereport(ERROR,
+							(errcode(ERRCODE_IO_ERROR),
+							 errmsg("There is no more version of page %u of relation %u/%u/%u.%u at LSN %X/%X at page server, request LSN %X/%X, latest version is at LSN %X/%X",
+									blockno, RelFileInfoFmt(rinfo), forkNum, LSN_FORMAT_ARGS(replay_lsn), LSN_FORMAT_ARGS(resp->lsn), LSN_FORMAT_ARGS(PageGetLSN((Page)getpage_resp->page)))));
+				}
 				memcpy(buffer, getpage_resp->page, BLCKSZ);
 
 				/*
@@ -2241,7 +2286,7 @@ Retry:
 			case T_NeonErrorResponse:
 				ereport(ERROR,
 						(errcode(ERRCODE_IO_ERROR),
-						 errmsg(NEON_TAG "[shard %d, reqid " UINT64_HEX_FORMAT "] could not read block %u in rel %u/%u/%u.%u from page server at lsn %X/%08X",
+						 errmsg(NEON_TAG "[shard %d, reqid %lx] could not read block %u in rel %u/%u/%u.%u from page server at lsn %X/%08X",
 								slot->shard_no, resp->reqid, blockno, RelFileInfoFmt(rinfo),
 								forkNum, LSN_FORMAT_ARGS(reqlsns->effective_request_lsn)),
 						 errdetail("page server returned error: %s",
@@ -2294,7 +2339,7 @@ communicator_nblocks(NRelFileInfo rinfo, ForkNumber forknum, neon_request_lsns *
 						relsize_resp->req.forknum != forknum)
 					{
 						NEON_PANIC_CONNECTION_STATE(0, PANIC,
-													"Unexpect response {reqid=" UINT64_HEX_FORMAT ",lsn=%X/%08X, since=%X/%08X, rel=%u/%u/%u.%u} to get relsize request {reqid=" UINT64_HEX_FORMAT ",lsn=%X/%08X, since=%X/%08X, rel=%u/%u/%u.%u}",
+													"Unexpect response {reqid=%lx,lsn=%X/%08X, since=%X/%08X, rel=%u/%u/%u.%u} to get relsize request {reqid=%lx,lsn=%X/%08X, since=%X/%08X, rel=%u/%u/%u.%u}",
 													resp->reqid, LSN_FORMAT_ARGS(resp->lsn), LSN_FORMAT_ARGS(resp->not_modified_since), RelFileInfoFmt(relsize_resp->req.rinfo), relsize_resp->req.forknum,
 													request.hdr.reqid, LSN_FORMAT_ARGS(request.hdr.lsn), LSN_FORMAT_ARGS(request.hdr.not_modified_since), RelFileInfoFmt(request.rinfo), forknum);
 					}
@@ -2307,14 +2352,14 @@ communicator_nblocks(NRelFileInfo rinfo, ForkNumber forknum, neon_request_lsns *
 				{
 					if (!equal_requests(resp, &request.hdr))
 					{
-						elog(WARNING, NEON_TAG "Error message {reqid=" UINT64_HEX_FORMAT ",lsn=%X/%08X, since=%X/%08X} doesn't match get relsize request {reqid=" UINT64_HEX_FORMAT ",lsn=%X/%08X, since=%X/%08X}",
+						elog(WARNING, NEON_TAG "Error message {reqid=%lx,lsn=%X/%08X, since=%X/%08X} doesn't match get relsize request {reqid=%lx,lsn=%X/%08X, since=%X/%08X}",
 							 resp->reqid, LSN_FORMAT_ARGS(resp->lsn), LSN_FORMAT_ARGS(resp->not_modified_since),
 							 request.hdr.reqid, LSN_FORMAT_ARGS(request.hdr.lsn), LSN_FORMAT_ARGS(request.hdr.not_modified_since));
 					}
 				}
 				ereport(ERROR,
 						(errcode(ERRCODE_IO_ERROR),
-						 errmsg(NEON_TAG "[reqid " UINT64_HEX_FORMAT "] could not read relation size of rel %u/%u/%u.%u from page server at lsn %X/%08X",
+						 errmsg(NEON_TAG "[reqid %lx] could not read relation size of rel %u/%u/%u.%u from page server at lsn %X/%08X",
 								resp->reqid,
 								RelFileInfoFmt(rinfo),
 								forknum,
@@ -2364,7 +2409,7 @@ communicator_dbsize(Oid dbNode, neon_request_lsns *request_lsns)
 						dbsize_resp->req.dbNode != dbNode)
 					{
 						NEON_PANIC_CONNECTION_STATE(0, PANIC,
-													"Unexpect response {reqid=" UINT64_HEX_FORMAT ",lsn=%X/%08X, since=%X/%08X, dbNode=%u} to get DB size request {reqid=" UINT64_HEX_FORMAT ",lsn=%X/%08X, since=%X/%08X, dbNode=%u}",
+													"Unexpect response {reqid=%lx,lsn=%X/%08X, since=%X/%08X, dbNode=%u} to get DB size request {reqid=%lx,lsn=%X/%08X, since=%X/%08X, dbNode=%u}",
 													resp->reqid, LSN_FORMAT_ARGS(resp->lsn), LSN_FORMAT_ARGS(resp->not_modified_since), dbsize_resp->req.dbNode,
 													request.hdr.reqid, LSN_FORMAT_ARGS(request.hdr.lsn), LSN_FORMAT_ARGS(request.hdr.not_modified_since), dbNode);
 					}
@@ -2377,14 +2422,14 @@ communicator_dbsize(Oid dbNode, neon_request_lsns *request_lsns)
 				{
 					if (!equal_requests(resp, &request.hdr))
 					{
-						elog(WARNING, NEON_TAG "Error message {reqid=" UINT64_HEX_FORMAT ",lsn=%X/%08X, since=%X/%08X} doesn't match get DB size request {reqid=" UINT64_HEX_FORMAT ",lsn=%X/%08X, since=%X/%08X}",
+						elog(WARNING, NEON_TAG "Error message {reqid=%lx,lsn=%X/%08X, since=%X/%08X} doesn't match get DB size request {reqid=%lx,lsn=%X/%08X, since=%X/%08X}",
 							 resp->reqid, LSN_FORMAT_ARGS(resp->lsn), LSN_FORMAT_ARGS(resp->not_modified_since),
 							 request.hdr.reqid, LSN_FORMAT_ARGS(request.hdr.lsn), LSN_FORMAT_ARGS(request.hdr.not_modified_since));
 					}
 				}
 				ereport(ERROR,
 						(errcode(ERRCODE_IO_ERROR),
-						 errmsg(NEON_TAG "[reqid " UINT64_HEX_FORMAT "] could not read db size of db %u from page server at lsn %X/%08X",
+						 errmsg(NEON_TAG "[reqid %lx] could not read db size of db %u from page server at lsn %X/%08X",
 								resp->reqid,
 								dbNode, LSN_FORMAT_ARGS(request_lsns->effective_request_lsn)),
 						 errdetail("page server returned error: %s",
@@ -2455,7 +2500,7 @@ communicator_read_slru_segment(SlruKind kind, int64 segno, neon_request_lsns *re
 					slru_resp->req.segno != segno)
 				{
 					NEON_PANIC_CONNECTION_STATE(0, PANIC,
-												"Unexpect response {reqid=" UINT64_HEX_FORMAT ",lsn=%X/%08X, since=%X/%08X, kind=%u, segno=%u} to get SLRU segment request {reqid=" UINT64_HEX_FORMAT ",lsn=%X/%08X, since=%X/%08X, kind=%u, segno=%lluu}",
+												"Unexpect response {reqid=%lx,lsn=%X/%08X, since=%X/%08X, kind=%u, segno=%u} to get SLRU segment request {reqid=%lx,lsn=%X/%08X, since=%X/%08X, kind=%u, segno=%lluu}",
 												resp->reqid, LSN_FORMAT_ARGS(resp->lsn), LSN_FORMAT_ARGS(resp->not_modified_since), slru_resp->req.kind, slru_resp->req.segno,
 												request.hdr.reqid, LSN_FORMAT_ARGS(request.hdr.lsn), LSN_FORMAT_ARGS(request.hdr.not_modified_since), kind, (unsigned long long) segno);
 				}
@@ -2469,14 +2514,14 @@ communicator_read_slru_segment(SlruKind kind, int64 segno, neon_request_lsns *re
 			{
 				if (!equal_requests(resp, &request.hdr))
 				{
-					elog(WARNING, NEON_TAG "Error message {reqid=" UINT64_HEX_FORMAT ",lsn=%X/%08X, since=%X/%08X} doesn't match get SLRU segment request {reqid=" UINT64_HEX_FORMAT ",lsn=%X/%08X, since=%X/%08X}",
+					elog(WARNING, NEON_TAG "Error message {reqid=%lx,lsn=%X/%08X, since=%X/%08X} doesn't match get SLRU segment request {reqid=%lx,lsn=%X/%08X, since=%X/%08X}",
 						 resp->reqid, LSN_FORMAT_ARGS(resp->lsn), LSN_FORMAT_ARGS(resp->not_modified_since),
 						 request.hdr.reqid, LSN_FORMAT_ARGS(request.hdr.lsn), LSN_FORMAT_ARGS(request.hdr.not_modified_since));
 				}
 			}
 			ereport(ERROR,
 					(errcode(ERRCODE_IO_ERROR),
-					 errmsg(NEON_TAG "[reqid " UINT64_HEX_FORMAT "] could not read SLRU %d segment %llu at lsn %X/%08X",
+					 errmsg(NEON_TAG "[reqid %lx] could not read SLRU %d segment %llu at lsn %X/%08X",
 							resp->reqid,
 							kind,
 							(unsigned long long) segno,

pgxn/neon/communicator/Cargo.toml

@@ -13,14 +13,7 @@ crate-type = ["staticlib"]
 testing = []
 
 [dependencies]
-axum.workspace = true
-http.workspace = true
-tokio = { version = "1.43.1", features = ["macros", "net", "io-util", "rt", "rt-multi-thread"] }
-tracing.workspace = true
-tracing-subscriber.workspace = true
-
-measured.workspace = true
-utils.workspace = true
+neon-shmem.workspace = true
 workspace_hack = { version = "0.1", path = "../../../workspace_hack" }
 
 [build-dependencies]

pgxn/neon/communicator/README.md

@@ -1,22 +1,7 @@
-# Communicator
-
-This package provides the so-called "compute-pageserver communicator",
-or just "communicator" in short. The communicator is a separate
-background worker process that runs in the PostgreSQL server. It's
-part of the neon extension. Currently, it only provides an HTTP
-endpoint for metrics, but in the future it will evolve to handle all
-communications with the pageservers.
-
-## Source code view
-
-pgxn/neon/communicator_process.c
-	Contains code needed to start up the communicator process, and
-	the glue that interacts with PostgreSQL code and the Rust
-	code in the communicator process.
-
-
-pgxn/neon/communicator/src/worker_process/
-    Worker process main loop and glue code
+This package will evolve into a "compute-pageserver communicator"
+process and machinery. For now, it's just a dummy that doesn't do
+anything interesting, but it allows us to test the compilation and
+linking of Rust code into the Postgres extensions.
 
 At compilation time, pgxn/neon/communicator/ produces a static
 library, libcommunicator.a. It is linked to the neon.so extension

pgxn/neon/communicator/src/lib.rs

@@ -1 +1,6 @@
-mod worker_process;
+/// dummy function, just to test linking Rust functions into the C
+/// extension
+#[unsafe(no_mangle)]
+pub extern "C" fn communicator_dummy(arg: u32) -> u32 {
+    arg + 1
+}

pgxn/neon/communicator/src/worker_process/callbacks.rs

@@ -1,49 +0,0 @@
-//! C callbacks to PostgreSQL facilities that the neon extension needs to provide. These are
-//! implemented in `neon/pgxn/communicator_process.c`.  The function signatures better match!
-//!
-//! These are called from the communicator threads! Careful what you do, most Postgres functions are
-//! not safe to call in that context.
-
-#[cfg(not(test))]
-unsafe extern "C" {
-    pub fn callback_set_my_latch_unsafe();
-    pub fn callback_get_lfc_metrics_unsafe() -> LfcMetrics;
-}
-
-// Compile unit tests with dummy versions of the functions. Unit tests
-// cannot call back into the C code. (As of this writing, no unit tests
-// even exists in the communicator package, but the code coverage build
-// still builds these and tries to link with the external C code.)
-#[cfg(test)]
-unsafe fn callback_set_my_latch_unsafe() {
-    panic!("not usable in unit tests");
-}
-#[cfg(test)]
-unsafe fn callback_get_lfc_metrics_unsafe() -> LfcMetrics {
-    panic!("not usable in unit tests");
-}
-
-// safe wrappers
-
-pub(super) fn callback_set_my_latch() {
-    unsafe { callback_set_my_latch_unsafe() };
-}
-
-#[repr(C)]
-pub struct LfcMetrics {
-    pub lfc_cache_size_limit: i64,
-    pub lfc_hits: i64,
-    pub lfc_misses: i64,
-    pub lfc_used: i64,
-    pub lfc_writes: i64,
-
-    // working set size looking back 1..60 minutes.
-    //
-    // Index 0 is size of working set accessed within last 1 minute,
-    // index 59 is size of working set accessed within last 60 minutes.
-    pub lfc_approximate_working_set_size_windows: [i64; 60],
-}
-
-pub fn callback_get_lfc_metrics() -> LfcMetrics {
-    unsafe { callback_get_lfc_metrics_unsafe() }
-}

pgxn/neon/communicator/src/worker_process/lfc_metrics.rs

@@ -1,89 +0,0 @@
-use measured::{
-    FixedCardinalityLabel, Gauge, GaugeVec, LabelGroup, MetricGroup,
-    label::{LabelName, LabelValue, StaticLabelSet},
-    metric::{MetricEncoding, gauge::GaugeState, group::Encoding},
-};
-
-use super::callbacks::callback_get_lfc_metrics;
-
-pub(crate) struct LfcMetricsCollector;
-
-impl LfcMetricsCollector {
-    pub(crate) fn new() -> Self {
-        Self
-    }
-}
-
-#[derive(MetricGroup)]
-#[metric(new())]
-struct LfcMetricsGroup {
-    /// LFC cache size limit in bytes
-    lfc_cache_size_limit: Gauge,
-    /// LFC cache hits
-    lfc_hits: Gauge,
-    /// LFC cache misses
-    lfc_misses: Gauge,
-    /// LFC chunks used (chunk = 1MB)
-    lfc_used: Gauge,
-    /// LFC cache writes
-    lfc_writes: Gauge,
-    /// Approximate working set size in pages of 8192 bytes
-    #[metric(init = GaugeVec::dense())]
-    lfc_approximate_working_set_size_windows: GaugeVec<StaticLabelSet<MinuteAsSeconds>>,
-}
-
-impl<T: Encoding> MetricGroup<T> for LfcMetricsCollector
-where
-    GaugeState: MetricEncoding<T>,
-{
-    fn collect_group_into(&self, enc: &mut T) -> Result<(), <T as Encoding>::Err> {
-        let g = LfcMetricsGroup::new();
-
-        let lfc_metrics = callback_get_lfc_metrics();
-
-        g.lfc_cache_size_limit.set(lfc_metrics.lfc_cache_size_limit);
-        g.lfc_hits.set(lfc_metrics.lfc_hits);
-        g.lfc_misses.set(lfc_metrics.lfc_misses);
-        g.lfc_used.set(lfc_metrics.lfc_used);
-        g.lfc_writes.set(lfc_metrics.lfc_writes);
-
-        for i in 0..60 {
-            let val = lfc_metrics.lfc_approximate_working_set_size_windows[i];
-            g.lfc_approximate_working_set_size_windows
-                .set(MinuteAsSeconds(i), val);
-        }
-
-        g.collect_group_into(enc)
-    }
-}
-
-/// This stores the values in range 0..60,
-/// encodes them as seconds (0, 60, 120, 180, ..., 3540)
-#[derive(Clone, Copy)]
-struct MinuteAsSeconds(usize);
-
-impl FixedCardinalityLabel for MinuteAsSeconds {
-    fn cardinality() -> usize {
-        60
-    }
-
-    fn encode(&self) -> usize {
-        self.0
-    }
-
-    fn decode(value: usize) -> Self {
-        Self(value)
-    }
-}
-
-impl LabelValue for MinuteAsSeconds {
-    fn visit<V: measured::label::LabelVisitor>(&self, v: V) -> V::Output {
-        v.write_int(self.0 as i64 * 60)
-    }
-}
-
-impl LabelGroup for MinuteAsSeconds {
-    fn visit_values(&self, v: &mut impl measured::label::LabelGroupVisitor) {
-        v.write_value(LabelName::from_str("duration_seconds"), self);
-    }
-}

pgxn/neon/communicator/src/worker_process/logging.rs

@@ -1,228 +0,0 @@
-//! Glue code to hook up Rust logging with the `tracing` crate to the PostgreSQL log
-//!
-//! In the Rust threads, the log messages are written to a mpsc Channel, and the Postgres
-//! process latch is raised. That wakes up the loop in the main thread, see
-//! `communicator_new_bgworker_main()`. It reads the message from the channel and
-//! ereport()s it. This ensures that only one thread, the main thread, calls the
-//! PostgreSQL logging routines at any time.
-
-use std::sync::mpsc::sync_channel;
-use std::sync::mpsc::{Receiver, SyncSender};
-use std::sync::mpsc::{TryRecvError, TrySendError};
-
-use tracing::info;
-use tracing::{Event, Level, Metadata, Subscriber};
-use tracing_subscriber::filter::LevelFilter;
-use tracing_subscriber::fmt::format::Writer;
-use tracing_subscriber::fmt::{FmtContext, FormatEvent, FormatFields, FormattedFields, MakeWriter};
-use tracing_subscriber::registry::LookupSpan;
-
-use crate::worker_process::callbacks::callback_set_my_latch;
-
-pub struct LoggingState {
-    receiver: Receiver<FormattedEventWithMeta>,
-}
-
-/// Called once, at worker process startup. The returned LoggingState is passed back
-/// in the subsequent calls to `pump_logging`. It is opaque to the C code.
-#[unsafe(no_mangle)]
-pub extern "C" fn configure_logging() -> Box<LoggingState> {
-    let (sender, receiver) = sync_channel(1000);
-
-    let maker = Maker { channel: sender };
-
-    use tracing_subscriber::prelude::*;
-    let r = tracing_subscriber::registry();
-
-    let r = r.with(
-        tracing_subscriber::fmt::layer()
-            .with_ansi(false)
-            .event_format(SimpleFormatter::new())
-            .with_writer(maker)
-            // TODO: derive this from log_min_messages?
-            .with_filter(LevelFilter::from_level(Level::INFO)),
-    );
-    r.init();
-
-    info!("communicator process logging started");
-
-    let state = LoggingState { receiver };
-
-    Box::new(state)
-}
-
-/// Read one message from the logging queue. This is essentially a wrapper to Receiver,
-/// with a C-friendly signature.
-///
-/// The message is copied into *errbuf, which is a caller-supplied buffer of size
-/// `errbuf_len`.  If the message doesn't fit in the buffer, it is truncated. It is always
-/// NULL-terminated.
-///
-/// The error level is returned *elevel_p. It's one of the PostgreSQL error levels, see
-/// elog.h
-#[unsafe(no_mangle)]
-pub extern "C" fn pump_logging(
-    state: &mut LoggingState,
-    errbuf: *mut u8,
-    errbuf_len: u32,
-    elevel_p: &mut i32,
-) -> i32 {
-    let msg = match state.receiver.try_recv() {
-        Err(TryRecvError::Empty) => return 0,
-        Err(TryRecvError::Disconnected) => return -1,
-        Ok(msg) => msg,
-    };
-
-    let src: &[u8] = &msg.message;
-    let dst = errbuf;
-    let len = std::cmp::min(src.len(), errbuf_len as usize - 1);
-    unsafe {
-        std::ptr::copy_nonoverlapping(src.as_ptr(), dst, len);
-        *(errbuf.add(len)) = b'\0'; // NULL terminator
-    }
-
-    // XXX: these levels are copied from PostgreSQL's elog.h. Introduce another enum to
-    // hide these?
-    *elevel_p = match msg.level {
-        Level::TRACE => 10, // DEBUG5
-        Level::DEBUG => 14, // DEBUG1
-        Level::INFO => 17,  // INFO
-        Level::WARN => 19,  // WARNING
-        Level::ERROR => 21, // ERROR
-    };
-
-    1
-}
-
-//---- The following functions can be called from any thread ----
-
-#[derive(Clone)]
-struct FormattedEventWithMeta {
-    message: Vec<u8>,
-    level: tracing::Level,
-}
-
-impl Default for FormattedEventWithMeta {
-    fn default() -> Self {
-        FormattedEventWithMeta {
-            message: Vec::new(),
-            level: tracing::Level::DEBUG,
-        }
-    }
-}
-
-struct EventBuilder<'a> {
-    event: FormattedEventWithMeta,
-
-    maker: &'a Maker,
-}
-
-impl std::io::Write for EventBuilder<'_> {
-    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
-        self.event.message.write(buf)
-    }
-    fn flush(&mut self) -> std::io::Result<()> {
-        self.maker.send_event(self.event.clone());
-        Ok(())
-    }
-}
-
-impl Drop for EventBuilder<'_> {
-    fn drop(&mut self) {
-        let maker = self.maker;
-        let event = std::mem::take(&mut self.event);
-
-        maker.send_event(event);
-    }
-}
-
-struct Maker {
-    channel: SyncSender<FormattedEventWithMeta>,
-}
-
-impl<'a> MakeWriter<'a> for Maker {
-    type Writer = EventBuilder<'a>;
-
-    fn make_writer(&'a self) -> Self::Writer {
-        panic!("not expected to be called when make_writer_for is implemented");
-    }
-
-    fn make_writer_for(&'a self, meta: &Metadata<'_>) -> Self::Writer {
-        EventBuilder {
-            event: FormattedEventWithMeta {
-                message: Vec::new(),
-                level: *meta.level(),
-            },
-            maker: self,
-        }
-    }
-}
-
-impl Maker {
-    fn send_event(&self, e: FormattedEventWithMeta) {
-        match self.channel.try_send(e) {
-            Ok(()) => {
-                // notify the main thread
-                callback_set_my_latch();
-            }
-            Err(TrySendError::Disconnected(_)) => {}
-            Err(TrySendError::Full(_)) => {
-                // TODO: record that some messages were lost
-            }
-        }
-    }
-}
-
-/// Simple formatter implementation for tracing_subscriber, which prints the log spans and
-/// message part like the default formatter, but no timestamp or error level. The error
-/// level is captured separately by `FormattedEventWithMeta', and when the error is
-/// printed by the main thread, with PostgreSQL ereport(), it gets a timestamp at that
-/// point. (The timestamp printed will therefore lag behind the timestamp on the event
-/// here, if the main thread doesn't process the log message promptly)
-struct SimpleFormatter;
-
-impl<S, N> FormatEvent<S, N> for SimpleFormatter
-where
-    S: Subscriber + for<'a> LookupSpan<'a>,
-    N: for<'a> FormatFields<'a> + 'static,
-{
-    fn format_event(
-        &self,
-        ctx: &FmtContext<'_, S, N>,
-        mut writer: Writer<'_>,
-        event: &Event<'_>,
-    ) -> std::fmt::Result {
-        // Format all the spans in the event's span context.
-        if let Some(scope) = ctx.event_scope() {
-            for span in scope.from_root() {
-                write!(writer, "{}", span.name())?;
-
-                // `FormattedFields` is a formatted representation of the span's fields,
-                // which is stored in its extensions by the `fmt` layer's `new_span`
-                // method. The fields will have been formatted by the same field formatter
-                // that's provided to the event formatter in the `FmtContext`.
-                let ext = span.extensions();
-                let fields = &ext
-                    .get::<FormattedFields<N>>()
-                    .expect("will never be `None`");
-
-                // Skip formatting the fields if the span had no fields.
-                if !fields.is_empty() {
-                    write!(writer, "{{{fields}}}")?;
-                }
-                write!(writer, ": ")?;
-            }
-        }
-
-        // Write fields on the event
-        ctx.field_format().format_fields(writer.by_ref(), event)?;
-
-        writeln!(writer)
-    }
-}
-
-impl SimpleFormatter {
-    fn new() -> Self {
-        SimpleFormatter {}
-    }
-}

pgxn/neon/communicator/src/worker_process/main_loop.rs

@@ -1,26 +0,0 @@
-use std::str::FromStr as _;
-
-use crate::worker_process::lfc_metrics::LfcMetricsCollector;
-
-use measured::MetricGroup;
-use utils::id::{TenantId, TimelineId};
-
-#[derive(MetricGroup)]
-pub struct CommunicatorWorkerProcessStruct {
-    /*** Metrics ***/
-    #[metric(flatten)]
-    pub(crate) lfc_metrics: LfcMetricsCollector,
-}
-
-pub(super) async fn init(
-    tenant_id: Option<&str>,
-    timeline_id: Option<&str>,
-) -> CommunicatorWorkerProcessStruct {
-    let _tenant_id = tenant_id.map(|s| TenantId::from_str(s).expect("invalid tenant ID"));
-    let _timeline_id = timeline_id.map(|s| TimelineId::from_str(s).expect("invalid timeline ID"));
-
-    CommunicatorWorkerProcessStruct {
-        // metrics
-        lfc_metrics: LfcMetricsCollector::new(),
-    }
-}

pgxn/neon/communicator/src/worker_process/metrics_exporter.rs

@@ -1,80 +0,0 @@
-//! Export information about Postgres, the communicator process, file cache etc. as
-//! prometheus metrics.
-//!
-//! The exporter speaks HTTP, listens on a Unix Domain Socket under the Postgres
-//! data directory. For debugging, you can access it with curl:
-//!
-//!    curl --unix-socket neon-communicator.socket http://localhost/metrics
-//!
-use axum::Router;
-use axum::body::Body;
-use axum::extract::State;
-use axum::response::Response;
-use http::StatusCode;
-use http::header::CONTENT_TYPE;
-
-use measured::MetricGroup;
-use measured::text::BufferedTextEncoder;
-
-use std::path::PathBuf;
-
-use tokio::net::UnixListener;
-
-use crate::worker_process::main_loop::CommunicatorWorkerProcessStruct;
-
-const NEON_COMMUNICATOR_SOCKET_NAME: &str = "neon-communicator.socket";
-
-impl CommunicatorWorkerProcessStruct {
-    /// Launch the metrics exporter
-    pub(crate) async fn launch_metrics_exporter(&'static self) {
-        use axum::routing::get;
-        let app = Router::new()
-            .route("/metrics", get(get_metrics))
-            .route("/autoscaling_metrics", get(get_autoscaling_metrics))
-            .route("/debug/panic", get(handle_debug_panic))
-            .with_state(self);
-
-        // Listen on unix domain socket, in the data directory. That should be unique.
-        let path = PathBuf::from(NEON_COMMUNICATOR_SOCKET_NAME);
-        let listener = UnixListener::bind(path.clone()).unwrap();
-
-        tokio::spawn(async {
-            tracing::info!("metrics listener spawned");
-            axum::serve(listener, app).await.unwrap()
-        });
-    }
-}
-
-/// Expose all Prometheus metrics.
-async fn get_metrics(State(state): State<&CommunicatorWorkerProcessStruct>) -> Response {
-    tracing::trace!("/metrics requested");
-    metrics_to_response(&state).await
-}
-
-/// Expose Prometheus metrics, for use by the autoscaling agent.
-///
-/// This is a subset of all the metrics.
-async fn get_autoscaling_metrics(
-    State(state): State<&CommunicatorWorkerProcessStruct>,
-) -> Response {
-    tracing::trace!("/metrics requested");
-    metrics_to_response(&state.lfc_metrics).await
-}
-
-async fn handle_debug_panic(State(_state): State<&CommunicatorWorkerProcessStruct>) -> Response {
-    panic!("test HTTP handler task panic");
-}
-
-/// Helper function to convert prometheus metrics to a text response
-async fn metrics_to_response(metrics: &(dyn MetricGroup<BufferedTextEncoder> + Sync)) -> Response {
-    let mut enc = BufferedTextEncoder::new();
-    metrics
-        .collect_group_into(&mut enc)
-        .unwrap_or_else(|never| match never {});
-
-    Response::builder()
-        .status(StatusCode::INTERNAL_SERVER_ERROR)
-        .header(CONTENT_TYPE, "application/text")
-        .body(Body::from(enc.finish()))
-        .unwrap()
-}

pgxn/neon/communicator/src/worker_process/mod.rs

@@ -1,13 +0,0 @@
-//! This code runs in the communicator worker process. This provides
-//! the glue code to:
-//!
-//! - launch the main loop,
-//! - receive IO requests from backends and process them,
-//! - write results back to backends.
-
-mod callbacks;
-mod lfc_metrics;
-mod logging;
-mod main_loop;
-mod metrics_exporter;
-mod worker_interface;

pgxn/neon/communicator/src/worker_process/worker_interface.rs

@@ -1,44 +0,0 @@
-//! Functions called from the C code in the worker process
-
-use std::ffi::{CStr, c_char};
-
-use crate::worker_process::main_loop;
-use crate::worker_process::main_loop::CommunicatorWorkerProcessStruct;
-
-/// Launch the communicator's tokio tasks, which do most of the work.
-///
-/// The caller has initialized the process as a regular PostgreSQL background worker
-/// process.
-#[unsafe(no_mangle)]
-pub extern "C" fn communicator_worker_process_launch(
-    tenant_id: *const c_char,
-    timeline_id: *const c_char,
-) -> &'static CommunicatorWorkerProcessStruct {
-    // Convert the arguments into more convenient Rust types
-    let tenant_id = if tenant_id.is_null() {
-        None
-    } else {
-        Some(unsafe { CStr::from_ptr(tenant_id) }.to_str().unwrap())
-    };
-    let timeline_id = if timeline_id.is_null() {
-        None
-    } else {
-        Some(unsafe { CStr::from_ptr(timeline_id) }.to_str().unwrap())
-    };
-
-    let runtime = tokio::runtime::Builder::new_multi_thread()
-        .enable_all()
-        .thread_name("communicator thread")
-        .build()
-        .unwrap();
-
-    let worker_struct = runtime.block_on(main_loop::init(tenant_id, timeline_id));
-    let worker_struct = Box::leak(Box::new(worker_struct));
-
-    runtime.block_on(worker_struct.launch_metrics_exporter());
-
-    // keep the runtime running after we exit this function
-    Box::leak(Box::new(runtime));
-
-    worker_struct
-}

pgxn/neon/communicator_process.c

@@ -1,179 +0,0 @@
-/*-------------------------------------------------------------------------
- *
- * communicator_process.c
- *	  Functions for starting up the communicator background worker process.
- *
- * Currently, the communicator process only functions as a metrics
- * exporter. It provides an HTTP endpoint for polling a limited set of
- * metrics. TODO: In the future, it will do much more, i.e. handle all
- * the communications with the pageservers.
- *
- * Portions Copyright (c) 1996-2021, PostgreSQL Global Development Group
- * Portions Copyright (c) 1994, Regents of the University of California
- *
- *-------------------------------------------------------------------------
- */
-#include "postgres.h"
-
-#include <unistd.h>
-
-#include "miscadmin.h"
-#include "postmaster/bgworker.h"
-#include "postmaster/interrupt.h"
-#include "postmaster/postmaster.h"
-#include "replication/walsender.h"
-#include "storage/ipc.h"
-#include "storage/latch.h"
-#include "storage/pmsignal.h"
-#include "storage/procsignal.h"
-#include "tcop/tcopprot.h"
-
-#include "communicator_process.h"
-#include "file_cache.h"
-#include "neon.h"
-#include "neon_perf_counters.h"
-
-/* the rust bindings, generated by cbindgen */
-#include "communicator/communicator_bindings.h"
-
-PGDLLEXPORT void communicator_new_bgworker_main(Datum main_arg);
-
-/**** Initialization functions. These run in postmaster ****/
-
-void
-register_communicator_bgworker(void)
-{
-	BackgroundWorker bgw;
-
-	/* Initialize the background worker process */
-	memset(&bgw, 0, sizeof(bgw));
-	bgw.bgw_flags = BGWORKER_SHMEM_ACCESS;
-	bgw.bgw_start_time = BgWorkerStart_PostmasterStart;
-	snprintf(bgw.bgw_library_name, BGW_MAXLEN, "neon");
-	snprintf(bgw.bgw_function_name, BGW_MAXLEN, "communicator_new_bgworker_main");
-	snprintf(bgw.bgw_name, BGW_MAXLEN, "Storage communicator process");
-	snprintf(bgw.bgw_type, BGW_MAXLEN, "Storage communicator process");
-	bgw.bgw_restart_time = 5;
-	bgw.bgw_notify_pid = 0;
-	bgw.bgw_main_arg = (Datum) 0;
-
-	RegisterBackgroundWorker(&bgw);
-}
-
-/**** Worker process functions. These run in the communicator worker process ****/
-
-/* Entry point for the communicator bgworker process */
-void
-communicator_new_bgworker_main(Datum main_arg)
-{
-	struct LoggingState *logging;
-	char		errbuf[1000];
-	int			elevel;
-	const struct CommunicatorWorkerProcessStruct *proc_handle;
-
-	/*
-	 * Pretend that this process is a WAL sender. That affects the shutdown
-	 * sequence: WAL senders are shut down last, after the final checkpoint
-	 * has been written. That's what we want for the communicator process too.
-	 */
-	am_walsender = true;
-	MarkPostmasterChildWalSender();
-
-	/* Establish signal handlers. */
-	pqsignal(SIGUSR1, procsignal_sigusr1_handler);
-	/*
-	 * Postmaster sends us SIGUSR2 when all regular backends and bgworkers
-	 * have exited, and it's time for us to exit too
-	 */
-	pqsignal(SIGUSR2, die);
-	pqsignal(SIGHUP, SignalHandlerForConfigReload);
-	pqsignal(SIGTERM, die);
-
-	BackgroundWorkerUnblockSignals();
-
-	logging = configure_logging();
-
-	proc_handle = communicator_worker_process_launch(
-		neon_tenant[0] == '\0' ? NULL : neon_tenant,
-		neon_timeline[0] == '\0' ? NULL : neon_timeline
-		);
-
-	/* proc_handle is not currently used, but will be in the future */
-	(void) proc_handle;
-
-	/*
-	 * The Rust tokio runtime has been launched, and it's running in the
-	 * background now. This process is now multi-threaded! The Rust threads do
-	 * not call into any Postgres functions.
-	 *
-	 * This loop in the main thread handles any interactions we need with the
-	 * rest of PostgreSQL.
-	 */
-	elog(LOG, "communicator threads started");
-	for (;;)
-	{
-		int32		rc;
-
-		/*
-		 * Check interrupts like system shutdown or config reload
-		 */
-		CHECK_FOR_INTERRUPTS();
-		if (ConfigReloadPending)
-		{
-			ConfigReloadPending = false;
-			ProcessConfigFile(PGC_SIGHUP);
-		}
-
-		/*
-		 * Forward any log messages from the Rust threads into the normal Postgres
-		 * logging facility.
-		 */
-		for (;;)
-		{
-			rc = pump_logging(logging, (uint8 *) errbuf, sizeof(errbuf), &elevel);
-			if (rc == 0)
-			{
-				/* nothing to do */
-				break;
-			}
-			else if (rc == 1)
-			{
-				/* Because we don't want to exit on error */
-				if (elevel == ERROR)
-					elevel = LOG;
-				if (elevel == INFO)
-					elevel = LOG;
-				elog(elevel, "[COMMUNICATOR] %s", errbuf);
-			}
-			else if (rc == -1)
-			{
-				elog(ERROR, "logging channel was closed unexpectedly");
-			}
-		}
-
-		/*
-		 * Wait until we are woken up. The rust threads will set the latch if
-		 * there's log message to forward.
-		 */
-		(void) WaitLatch(MyLatch,
-						 WL_LATCH_SET | WL_EXIT_ON_PM_DEATH,
-						 0,
-						 PG_WAIT_EXTENSION);
-		ResetLatch(MyLatch);
-	}
-}
-
-/****
- * Callbacks from the rust code, in the communicator process.
- *
- * NOTE: These must be thread-safe! It's very limited which PostgreSQL
- * functions you can use!!!
- *
- * The signatures of these need to match those in the Rust code.
- */
-
-void
-callback_set_my_latch_unsafe(void)
-{
-	SetLatch(MyLatch);
-}

pgxn/neon/communicator_process.h

@@ -1,17 +0,0 @@
-/*-------------------------------------------------------------------------
- *
- * communicator_process.h
- *		Communicator process
- *
- *
- * Portions Copyright (c) 1996-2021, PostgreSQL Global Development Group
- * Portions Copyright (c) 1994, Regents of the University of California
- *
- *-------------------------------------------------------------------------
- */
-#ifndef COMMUNICATOR_PROCESS_H
-#define COMMUNICATOR_PROCESS_H
-
-extern void register_communicator_bgworker(void);
-
-#endif			/* COMMUNICATOR_PROCESS_H */

pgxn/neon/file_cache.c

@@ -52,8 +52,6 @@
 #include "pagestore_client.h"
 #include "communicator.h"
 
-#include "communicator/communicator_bindings.h"
-
 #define CriticalAssert(cond) do if (!(cond)) elog(PANIC, "LFC: assertion %s failed at %s:%d: ", #cond, __FILE__, __LINE__); while (0)
 
 /*
@@ -2181,38 +2179,6 @@ lfc_approximate_working_set_size_seconds(time_t duration, bool reset)
 	return dc;
 }
 
-/*
- * Get metrics, for the built-in metrics exporter that's part of the communicator
- * process.
- *
- * NB: This is called from a Rust tokio task inside the communicator process.
- * Acquiring lwlocks, elog(), allocating memory etc. or anything else
- * non-trivial is strictly prohibited here!
- */
-struct LfcMetrics
-callback_get_lfc_metrics_unsafe(void)
-{
-	struct LfcMetrics result = {
-		.lfc_cache_size_limit = lfc_size_limit,
-		.lfc_hits = lfc_ctl ? lfc_ctl->hits : 0,
-		.lfc_misses = lfc_ctl ? lfc_ctl->misses : 0,
-		.lfc_used = lfc_ctl ? lfc_ctl->used : 0,
-		.lfc_writes = lfc_ctl ? lfc_ctl->writes : 0,
-	};
-
-	if (lfc_ctl)
-	{
-		for (int minutes = 1; minutes <= 60; minutes++)
-		{
-			result.lfc_approximate_working_set_size_windows[minutes - 1] =
-				lfc_approximate_working_set_size_seconds(minutes * 60, false);
-		}
-	}
-
-	return result;
-}
-
-
 PG_FUNCTION_INFO_V1(get_local_cache_state);
 
 Datum

pgxn/neon/neon.c

@@ -30,7 +30,6 @@
 #include "utils/guc_tables.h"
 
 #include "communicator.h"
-#include "communicator_process.h"
 #include "extension_server.h"
 #include "file_cache.h"
 #include "neon.h"
@@ -44,6 +43,9 @@
 #include "storage/ipc.h"
 #endif
 
+/* the rust bindings, generated by cbindgen */
+#include "communicator/communicator_bindings.h"
+
 PG_MODULE_MAGIC;
 void		_PG_init(void);
 
@@ -453,13 +455,14 @@ _PG_init(void)
 	shmem_startup_hook = neon_shmem_startup_hook;
 #endif
 
+	/* dummy call to a Rust function in the communicator library, to check that it works */
+	(void) communicator_dummy(123);
+
 	pg_init_libpagestore();
 	lfc_init();
 	pg_init_walproposer();
 	init_lwlsncache();
 
-	register_communicator_bgworker();
-
 	pg_init_communicator();
 	Custom_XLogReaderRoutines = NeonOnDemandXLogReaderRoutines;
 

pgxn/neon/neon_pgversioncompat.h

@@ -165,8 +165,4 @@ extern void InitMaterializedSRF(FunctionCallInfo fcinfo, bits32 flags);
 extern TimeLineID GetWALInsertionTimeLine(void);
 #endif
 
-/* format codes not present in PG17-; but available in PG18+ */
-#define INT64_HEX_FORMAT "%" INT64_MODIFIER "x"
-#define UINT64_HEX_FORMAT "%" INT64_MODIFIER "x"
-
 #endif							/* NEON_PGVERSIONCOMPAT_H */

poetry.lock

@@ -1,4 +1,4 @@
-# This file is automatically @generated by Poetry 2.1.3 and should not be changed by hand.
+# This file is automatically @generated by Poetry 2.1.1 and should not be changed by hand.
 
 [[package]]
 name = "aiohappyeyeballs"
@@ -3071,21 +3071,6 @@ urllib3 = ">=1.21.1,<3"
 socks = ["PySocks (>=1.5.6,!=1.5.7)"]
 use-chardet-on-py3 = ["chardet (>=3.0.2,<6)"]
 
-[[package]]
-name = "requests-unixsocket"
-version = "0.4.1"
-description = "Use requests to talk HTTP via a UNIX domain socket"
-optional = false
-python-versions = ">=3.9"
-groups = ["main"]
-files = [
-    {file = "requests_unixsocket-0.4.1-py3-none-any.whl", hash = "sha256:60c4942e9dbecc2f64d611039fb1dfc25da382083c6434ac0316dca3ff908f4d"},
-    {file = "requests_unixsocket-0.4.1.tar.gz", hash = "sha256:b2596158c356ecee68d27ba469a52211230ac6fb0cde8b66afb19f0ed47a1995"},
-]
-
-[package.dependencies]
-requests = ">=1.1"
-
 [[package]]
 name = "responses"
 version = "0.25.3"
@@ -3862,4 +3847,4 @@ cffi = ["cffi (>=1.11)"]
 [metadata]
 lock-version = "2.1"
 python-versions = "^3.11"
-content-hash = "b741d0b6f7cd3a062dedb8896471b6e7ba20ab1caef82c060506562e19380ad5"
+content-hash = "bd93313f110110aa53b24a3ed47ba2d7f60e2c658a79cdff7320fed1bb1b57b5"

proxy/src/cache/timed_lru.rs

@@ -14,8 +14,8 @@ use std::time::{Duration, Instant};
 use hashlink::{LruCache, linked_hash_map::RawEntryMut};
 use tracing::debug;
 
-use super::Cache;
 use super::common::Cached;
+use super::{Cache, timed_lru};
 
 /// An implementation of timed LRU cache with fixed capacity.
 /// Key properties:
@@ -30,7 +30,7 @@ use super::common::Cached;
 ///
 /// * There's an API for immediate invalidation (removal) of a cache entry;
 ///   It's useful in case we know for sure that the entry is no longer correct.
-///   See [`Cached`] for more information.
+///   See [`timed_lru::Cached`] for more information.
 ///
 /// * Expired entries are kept in the cache, until they are evicted by the LRU policy,
 ///   or by a successful lookup (i.e. the entry hasn't expired yet).
@@ -217,18 +217,15 @@ impl<K: Hash + Eq + Clone, V: Clone> TimedLru<K, V> {
 }
 
 impl<K: Hash + Eq, V: Clone> TimedLru<K, V> {
-    /// Retrieve a cached entry in convenient wrapper, alongside timing information.
-    pub(crate) fn get_with_created_at<Q>(
-        &self,
-        key: &Q,
-    ) -> Option<Cached<&Self, (<Self as Cache>::Value, Instant)>>
+    /// Retrieve a cached entry in convenient wrapper.
+    pub(crate) fn get<Q>(&self, key: &Q) -> Option<timed_lru::Cached<&Self>>
     where
         K: Borrow<Q> + Clone,
         Q: Hash + Eq + ?Sized,
     {
         self.get_raw(key, |key, entry| Cached {
             token: Some((self, key.clone())),
-            value: (entry.value.clone(), entry.created_at),
+            value: entry.value.clone(),
         })
     }
 }

proxy/src/cancellation.rs

@@ -28,28 +28,23 @@ use crate::pqproto::CancelKeyData;
 use crate::rate_limiter::LeakyBucketRateLimiter;
 use crate::redis::keys::KeyPrefix;
 use crate::redis::kv_ops::{RedisKVClient, RedisKVClientError};
-use crate::util::run_until;
 
 type IpSubnetKey = IpNet;
 
-const CANCEL_KEY_TTL: Duration = Duration::from_secs(600);
-const CANCEL_KEY_REFRESH: Duration = Duration::from_secs(570);
+const CANCEL_KEY_TTL: std::time::Duration = std::time::Duration::from_secs(600);
+const CANCEL_KEY_REFRESH: std::time::Duration = std::time::Duration::from_secs(570);
 
 // Message types for sending through mpsc channel
 pub enum CancelKeyOp {
-    Store {
+    StoreCancelKey {
         key: CancelKeyData,
         value: Box<str>,
-        expire: Duration,
+        expire: std::time::Duration,
     },
-    Refresh {
-        key: CancelKeyData,
-        expire: Duration,
-    },
-    Get {
+    GetCancelData {
         key: CancelKeyData,
     },
-    GetOld {
+    GetCancelDataOld {
         key: CancelKeyData,
     },
 }
@@ -112,7 +107,7 @@ impl Pipeline {
 impl CancelKeyOp {
     fn register(&self, pipe: &mut Pipeline) {
         match self {
-            CancelKeyOp::Store { key, value, expire } => {
+            CancelKeyOp::StoreCancelKey { key, value, expire } => {
                 let key = KeyPrefix::Cancel(*key).build_redis_key();
                 pipe.add_command(Cmd::set_options(
                     &key,
@@ -120,15 +115,11 @@ impl CancelKeyOp {
                     SetOptions::default().with_expiration(SetExpiry::EX(expire.as_secs())),
                 ));
             }
-            CancelKeyOp::Refresh { key, expire } => {
-                let key = KeyPrefix::Cancel(*key).build_redis_key();
-                pipe.add_command(Cmd::expire(&key, expire.as_secs() as i64));
-            }
-            CancelKeyOp::GetOld { key } => {
+            CancelKeyOp::GetCancelDataOld { key } => {
                 let key = KeyPrefix::Cancel(*key).build_redis_key();
                 pipe.add_command(Cmd::hget(key, "data"));
             }
-            CancelKeyOp::Get { key } => {
+            CancelKeyOp::GetCancelData { key } => {
                 let key = KeyPrefix::Cancel(*key).build_redis_key();
                 pipe.add_command(Cmd::get(key));
             }
@@ -272,7 +263,7 @@ impl CancellationHandler {
             .proxy
             .cancel_channel_size
             .guard(RedisMsgKind::Get);
-        let op = CancelKeyOp::Get { key };
+        let op = CancelKeyOp::GetCancelData { key };
         let result = timeout(
             TIMEOUT,
             tx.call((guard, op), std::future::pending::<Infallible>()),
@@ -297,7 +288,7 @@ impl CancellationHandler {
                 .proxy
                 .cancel_channel_size
                 .guard(RedisMsgKind::HGet);
-            let op = CancelKeyOp::GetOld { key };
+            let op = CancelKeyOp::GetCancelDataOld { key };
             timeout(
                 TIMEOUT,
                 tx.call((guard, op), std::future::pending::<Infallible>()),
@@ -482,95 +473,40 @@ impl Session {
 
         let mut cancel = pin!(cancel);
 
-        enum State {
-            Set,
-            Refresh,
-        }
-        let mut state = State::Set;
-
         loop {
-            let guard_op = match state {
-                State::Set => {
-                    let guard = Metrics::get()
-                        .proxy
-                        .cancel_channel_size
-                        .guard(RedisMsgKind::Set);
-                    let op = CancelKeyOp::Store {
-                        key: self.key,
-                        value: closure_json.clone(),
-                        expire: CANCEL_KEY_TTL,
-                    };
-                    tracing::debug!(
-                        src=%self.key,
-                        dest=?cancel_closure.cancel_token,
-                        "registering cancellation key"
-                    );
-                    (guard, op)
-                }
-
-                State::Refresh => {
-                    let guard = Metrics::get()
-                        .proxy
-                        .cancel_channel_size
-                        .guard(RedisMsgKind::Expire);
-                    let op = CancelKeyOp::Refresh {
-                        key: self.key,
-                        expire: CANCEL_KEY_TTL,
-                    };
-                    tracing::debug!(
-                        src=%self.key,
-                        dest=?cancel_closure.cancel_token,
-                        "refreshing cancellation key"
-                    );
-                    (guard, op)
-                }
+            let guard = Metrics::get()
+                .proxy
+                .cancel_channel_size
+                .guard(RedisMsgKind::Set);
+            let op = CancelKeyOp::StoreCancelKey {
+                key: self.key,
+                value: closure_json.clone(),
+                expire: CANCEL_KEY_TTL,
             };
 
-            match tx.call(guard_op, cancel.as_mut()).await {
-                // SET returns OK
-                Ok(Value::Okay) => {
+            tracing::debug!(
+                src=%self.key,
+                dest=?cancel_closure.cancel_token,
+                "registering cancellation key"
+            );
+
+            match tx.call((guard, op), cancel.as_mut()).await {
+                Ok(_) => {
                     tracing::debug!(
                         src=%self.key,
                         dest=?cancel_closure.cancel_token,
                         "registered cancellation key"
                     );
-                    state = State::Refresh;
-                }
 
-                // EXPIRE returns 1
-                Ok(Value::Int(1)) => {
-                    tracing::debug!(
-                        src=%self.key,
-                        dest=?cancel_closure.cancel_token,
-                        "refreshed cancellation key"
-                    );
+                    // wait before continuing.
+                    tokio::time::sleep(CANCEL_KEY_REFRESH).await;
                 }
-
-                Ok(_) => {
-                    // Any other response likely means the key expired.
-                    tracing::warn!(src=%self.key, "refreshing cancellation key failed");
-                    // Re-enter the SET loop to repush full data.
-                    state = State::Set;
-                }
-
                 // retry immediately.
                 Err(BatchQueueError::Result(error)) => {
-                    tracing::warn!(?error, "error refreshing cancellation key");
-                    // Small delay to prevent busy loop with high cpu and logging.
-                    tokio::time::sleep(Duration::from_millis(10)).await;
-                    continue;
+                    tracing::warn!(?error, "error registering cancellation key");
                 }
-
                 Err(BatchQueueError::Cancelled(Err(_cancelled))) => break,
             }
-
-            // wait before continuing. break immediately if cancelled.
-            if run_until(tokio::time::sleep(CANCEL_KEY_REFRESH), cancel.as_mut())
-                .await
-                .is_err()
-            {
-                break;
-            }
         }
 
         if let Err(err) = cancel_closure

proxy/src/control_plane/client/cplane_proxy_v1.rs

@@ -23,13 +23,12 @@ use crate::control_plane::errors::{
     ControlPlaneError, GetAuthInfoError, GetEndpointJwksError, WakeComputeError,
 };
 use crate::control_plane::locks::ApiLocks;
-use crate::control_plane::messages::{ColdStartInfo, EndpointJwksResponse};
+use crate::control_plane::messages::{ColdStartInfo, EndpointJwksResponse, Reason};
 use crate::control_plane::{
     AccessBlockerFlags, AuthInfo, AuthSecret, CachedNodeInfo, EndpointAccessControl, NodeInfo,
     RoleAccessControl,
 };
 use crate::metrics::Metrics;
-use crate::proxy::retry::CouldRetry;
 use crate::rate_limiter::WakeComputeRateLimiter;
 use crate::types::{EndpointCacheKey, EndpointId, RoleName};
 use crate::{compute, http, scram};
@@ -383,31 +382,16 @@ impl super::ControlPlaneApi for NeonControlPlaneClient {
 
         macro_rules! check_cache {
             () => {
-                if let Some(cached) = self.caches.node_info.get_with_created_at(&key) {
-                    let (cached, (info, created_at)) = cached.take_value();
-                    return match info {
-                        Err(mut msg) => {
-                            info!(key = &*key, "found cached wake_compute error");
+                if let Some(cached) = self.caches.node_info.get(&key) {
+                    let (cached, info) = cached.take_value();
+                    let info = info.map_err(|c| {
+                        info!(key = &*key, "found cached wake_compute error");
+                        WakeComputeError::ControlPlane(ControlPlaneError::Message(Box::new(*c)))
+                    })?;
 
-                            // if retry_delay_ms is set, reduce it by the amount of time it spent in cache
-                            if let Some(status) = &mut msg.status {
-                                if let Some(retry_info) = &mut status.details.retry_info {
-                                    retry_info.retry_delay_ms = retry_info
-                                        .retry_delay_ms
-                                        .saturating_sub(created_at.elapsed().as_millis() as u64)
-                                }
-                            }
-
-                            Err(WakeComputeError::ControlPlane(ControlPlaneError::Message(
-                                msg,
-                            )))
-                        }
-                        Ok(info) => {
-                            debug!(key = &*key, "found cached compute node info");
-                            ctx.set_project(info.aux.clone());
-                            Ok(cached.map(|()| info))
-                        }
-                    };
+                    debug!(key = &*key, "found cached compute node info");
+                    ctx.set_project(info.aux.clone());
+                    return Ok(cached.map(|()| info));
                 }
             };
         }
@@ -450,29 +434,42 @@ impl super::ControlPlaneApi for NeonControlPlaneClient {
                 Ok(cached.map(|()| node))
             }
             Err(err) => match err {
-                WakeComputeError::ControlPlane(ControlPlaneError::Message(ref msg)) => {
-                    let retry_info = msg.status.as_ref().and_then(|s| s.details.retry_info);
+                WakeComputeError::ControlPlane(ControlPlaneError::Message(err)) => {
+                    let Some(status) = &err.status else {
+                        return Err(WakeComputeError::ControlPlane(ControlPlaneError::Message(
+                            err,
+                        )));
+                    };
 
-                    // If we can retry this error, do not cache it,
-                    // unless we were given a retry delay.
-                    if msg.could_retry() && retry_info.is_none() {
-                        return Err(err);
+                    let reason = status
+                        .details
+                        .error_info
+                        .map_or(Reason::Unknown, |x| x.reason);
+
+                    // if we can retry this error, do not cache it.
+                    if reason.can_retry() {
+                        return Err(WakeComputeError::ControlPlane(ControlPlaneError::Message(
+                            err,
+                        )));
                     }
 
+                    // at this point, we should only have quota errors.
                     debug!(
                         key = &*key,
                         "created a cache entry for the wake compute error"
                     );
 
-                    let ttl = retry_info.map_or(Duration::from_secs(30), |r| {
-                        Duration::from_millis(r.retry_delay_ms)
-                    });
+                    self.caches.node_info.insert_ttl(
+                        key,
+                        Err(err.clone()),
+                        Duration::from_secs(30),
+                    );
 
-                    self.caches.node_info.insert_ttl(key, Err(msg.clone()), ttl);
-
-                    Err(err)
+                    Err(WakeComputeError::ControlPlane(ControlPlaneError::Message(
+                        err,
+                    )))
                 }
-                err => Err(err),
+                err => return Err(err),
             },
         }
     }

proxy/src/control_plane/errors.rs

@@ -43,35 +43,28 @@ impl UserFacingError for ControlPlaneError {
 }
 
 impl ReportableError for ControlPlaneError {
-    fn get_error_kind(&self) -> ErrorKind {
+    fn get_error_kind(&self) -> crate::error::ErrorKind {
         match self {
             ControlPlaneError::Message(e) => match e.get_reason() {
-                Reason::RoleProtected
-                | Reason::ResourceNotFound
-                | Reason::ProjectNotFound
-                | Reason::EndpointNotFound
-                | Reason::EndpointDisabled
-                | Reason::BranchNotFound
-                | Reason::InvalidEphemeralEndpointOptions => ErrorKind::User,
-
+                Reason::RoleProtected => ErrorKind::User,
+                Reason::ResourceNotFound => ErrorKind::User,
+                Reason::ProjectNotFound => ErrorKind::User,
+                Reason::EndpointNotFound => ErrorKind::User,
+                Reason::BranchNotFound => ErrorKind::User,
                 Reason::RateLimitExceeded => ErrorKind::ServiceRateLimit,
-
-                Reason::NonDefaultBranchComputeTimeExceeded
-                | Reason::ActiveTimeQuotaExceeded
-                | Reason::ComputeTimeQuotaExceeded
-                | Reason::WrittenDataQuotaExceeded
-                | Reason::DataTransferQuotaExceeded
-                | Reason::LogicalSizeQuotaExceeded
-                | Reason::ActiveEndpointsLimitExceeded => ErrorKind::Quota,
-
-                Reason::ConcurrencyLimitReached
-                | Reason::LockAlreadyTaken
-                | Reason::RunningOperations
-                | Reason::EndpointIdle
-                | Reason::ProjectUnderMaintenance
-                | Reason::Unknown => ErrorKind::ControlPlane,
+                Reason::NonDefaultBranchComputeTimeExceeded => ErrorKind::Quota,
+                Reason::ActiveTimeQuotaExceeded => ErrorKind::Quota,
+                Reason::ComputeTimeQuotaExceeded => ErrorKind::Quota,
+                Reason::WrittenDataQuotaExceeded => ErrorKind::Quota,
+                Reason::DataTransferQuotaExceeded => ErrorKind::Quota,
+                Reason::LogicalSizeQuotaExceeded => ErrorKind::Quota,
+                Reason::ConcurrencyLimitReached => ErrorKind::ControlPlane,
+                Reason::LockAlreadyTaken => ErrorKind::ControlPlane,
+                Reason::RunningOperations => ErrorKind::ControlPlane,
+                Reason::ActiveEndpointsLimitExceeded => ErrorKind::ControlPlane,
+                Reason::Unknown => ErrorKind::ControlPlane,
             },
-            ControlPlaneError::Transport(_) => ErrorKind::ControlPlane,
+            ControlPlaneError::Transport(_) => crate::error::ErrorKind::ControlPlane,
         }
     }
 }
@@ -127,10 +120,10 @@ impl UserFacingError for GetAuthInfoError {
 }
 
 impl ReportableError for GetAuthInfoError {
-    fn get_error_kind(&self) -> ErrorKind {
+    fn get_error_kind(&self) -> crate::error::ErrorKind {
         match self {
-            Self::BadSecret => ErrorKind::ControlPlane,
-            Self::ApiError(_) => ErrorKind::ControlPlane,
+            Self::BadSecret => crate::error::ErrorKind::ControlPlane,
+            Self::ApiError(_) => crate::error::ErrorKind::ControlPlane,
         }
     }
 }

proxy/src/control_plane/messages.rs

@@ -126,16 +126,10 @@ pub(crate) enum Reason {
     /// or that the subject doesn't have enough permissions to access the requested endpoint.
     #[serde(rename = "ENDPOINT_NOT_FOUND")]
     EndpointNotFound,
-    /// EndpointDisabled indicates that the endpoint has been disabled and does not accept connections.
-    #[serde(rename = "ENDPOINT_DISABLED")]
-    EndpointDisabled,
     /// BranchNotFound indicates that the branch wasn't found, usually due to the provided ID not being correct,
     /// or that the subject doesn't have enough permissions to access the requested branch.
     #[serde(rename = "BRANCH_NOT_FOUND")]
     BranchNotFound,
-    /// InvalidEphemeralEndpointOptions indicates that the specified LSN or timestamp are wrong.
-    #[serde(rename = "INVALID_EPHEMERAL_OPTIONS")]
-    InvalidEphemeralEndpointOptions,
     /// RateLimitExceeded indicates that the rate limit for the operation has been exceeded.
     #[serde(rename = "RATE_LIMIT_EXCEEDED")]
     RateLimitExceeded,
@@ -158,9 +152,6 @@ pub(crate) enum Reason {
     /// LogicalSizeQuotaExceeded indicates that the logical size quota was exceeded.
     #[serde(rename = "LOGICAL_SIZE_QUOTA_EXCEEDED")]
     LogicalSizeQuotaExceeded,
-    /// ActiveEndpointsLimitExceeded indicates that the limit of concurrently active endpoints was exceeded.
-    #[serde(rename = "ACTIVE_ENDPOINTS_LIMIT_EXCEEDED")]
-    ActiveEndpointsLimitExceeded,
     /// RunningOperations indicates that the project already has some running operations
     /// and scheduling of new ones is prohibited.
     #[serde(rename = "RUNNING_OPERATIONS")]
@@ -171,13 +162,9 @@ pub(crate) enum Reason {
     /// LockAlreadyTaken indicates that the we attempted to take a lock that was already taken.
     #[serde(rename = "LOCK_ALREADY_TAKEN")]
     LockAlreadyTaken,
-    /// EndpointIdle indicates that the endpoint cannot become active, because it's idle.
-    #[serde(rename = "ENDPOINT_IDLE")]
-    EndpointIdle,
-    /// ProjectUnderMaintenance indicates that the project is currently ongoing maintenance,
-    /// and thus cannot accept connections.
-    #[serde(rename = "PROJECT_UNDER_MAINTENANCE")]
-    ProjectUnderMaintenance,
+    /// ActiveEndpointsLimitExceeded indicates that the limit of concurrently active endpoints was exceeded.
+    #[serde(rename = "ACTIVE_ENDPOINTS_LIMIT_EXCEEDED")]
+    ActiveEndpointsLimitExceeded,
     #[default]
     #[serde(other)]
     Unknown,
@@ -197,15 +184,13 @@ impl Reason {
     pub(crate) fn can_retry(self) -> bool {
         match self {
             // do not retry role protected errors
-            // not a transient error
+            // not a transitive error
             Reason::RoleProtected => false,
-            // on retry, it will still not be found or valid
+            // on retry, it will still not be found
             Reason::ResourceNotFound
             | Reason::ProjectNotFound
             | Reason::EndpointNotFound
-            | Reason::EndpointDisabled
-            | Reason::BranchNotFound
-            | Reason::InvalidEphemeralEndpointOptions => false,
+            | Reason::BranchNotFound => false,
             // we were asked to go away
             Reason::RateLimitExceeded
             | Reason::NonDefaultBranchComputeTimeExceeded
@@ -215,13 +200,11 @@ impl Reason {
             | Reason::DataTransferQuotaExceeded
             | Reason::LogicalSizeQuotaExceeded
             | Reason::ActiveEndpointsLimitExceeded => false,
-            // transient error. control plane is currently busy
+            // transitive error. control plane is currently busy
             // but might be ready soon
             Reason::RunningOperations
             | Reason::ConcurrencyLimitReached
-            | Reason::LockAlreadyTaken
-            | Reason::EndpointIdle
-            | Reason::ProjectUnderMaintenance => true,
+            | Reason::LockAlreadyTaken => true,
             // unknown error. better not retry it.
             Reason::Unknown => false,
         }

proxy/src/metrics.rs

@@ -376,7 +376,6 @@ pub enum Waiting {
 pub enum RedisMsgKind {
     Set,
     Get,
-    Expire,
     HGet,
 }
 

proxy/src/proxy/mod.rs

@@ -195,18 +195,15 @@ impl NeonOptions {
     // proxy options:
 
     /// `PARAMS_COMPAT` allows opting in to forwarding all startup parameters from client to compute.
-    pub const PARAMS_COMPAT: &'static str = "proxy_params_compat";
+    pub const PARAMS_COMPAT: &str = "proxy_params_compat";
 
     // cplane options:
 
     /// `LSN` allows provisioning an ephemeral compute with time-travel to the provided LSN.
-    const LSN: &'static str = "lsn";
-
-    /// `TIMESTAMP` allows provisioning an ephemeral compute with time-travel to the provided timestamp.
-    const TIMESTAMP: &'static str = "timestamp";
+    const LSN: &str = "lsn";
 
     /// `ENDPOINT_TYPE` allows configuring an ephemeral compute to be read_only or read_write.
-    const ENDPOINT_TYPE: &'static str = "endpoint_type";
+    const ENDPOINT_TYPE: &str = "endpoint_type";
 
     pub(crate) fn parse_params(params: &StartupMessageParams) -> Self {
         params
@@ -231,7 +228,6 @@ impl NeonOptions {
             // This is not a cplane option, we know it does not create ephemeral computes.
             Self::PARAMS_COMPAT => false,
             Self::LSN => true,
-            Self::TIMESTAMP => true,
             Self::ENDPOINT_TYPE => true,
             // err on the side of caution. any cplane options we don't know about
             // might lead to ephemeral computes.

proxy/src/util.rs

@@ -7,16 +7,8 @@ pub async fn run_until_cancelled<F: Future>(
     f: F,
     cancellation_token: &CancellationToken,
 ) -> Option<F::Output> {
-    run_until(f, cancellation_token.cancelled()).await.ok()
-}
-
-/// Runs the future `f` unless interrupted by future `condition`.
-pub async fn run_until<F1: Future, F2: Future>(
-    f: F1,
-    condition: F2,
-) -> Result<F1::Output, F2::Output> {
-    match select(pin!(f), pin!(condition)).await {
-        Either::Left((f1, _)) => Ok(f1),
-        Either::Right((f2, _)) => Err(f2),
+    match select(pin!(f), pin!(cancellation_token.cancelled())).await {
+        Either::Left((f, _)) => Some(f),
+        Either::Right(((), _)) => None,
     }
 }

pyproject.toml

@@ -50,7 +50,6 @@ types-pyyaml = "^6.0.12.20240917"
 testcontainers = "^4.9.0"
 # Install a release candidate of `jsonnet`, as it supports Python 3.13
 jsonnet = "^0.21.0-rc2"
-requests-unixsocket = "^0.4.1"
 
 [tool.poetry.group.dev.dependencies]
 mypy = "==1.13.0"

safekeeper/client/src/mgmt_api.rs

@@ -6,10 +6,10 @@
 use std::error::Error as _;
 
 use http_utils::error::HttpErrorBody;
-use reqwest::{IntoUrl, Method, Response, StatusCode};
+use reqwest::{IntoUrl, Method, StatusCode};
 use safekeeper_api::models::{
     self, PullTimelineRequest, PullTimelineResponse, SafekeeperStatus, SafekeeperUtilization,
-    TimelineCreateRequest,
+    TimelineCreateRequest, TimelineStatus,
 };
 use utils::id::{NodeId, TenantId, TimelineId};
 use utils::logging::SecretString;
@@ -161,12 +161,13 @@ impl Client {
         &self,
         tenant_id: TenantId,
         timeline_id: TimelineId,
-    ) -> Result<Response> {
+    ) -> Result<TimelineStatus> {
         let uri = format!(
             "{}/v1/tenant/{}/timeline/{}",
             self.mgmt_api_endpoint, tenant_id, timeline_id
         );
-        self.get(&uri).await
+        let resp = self.get(&uri).await?;
+        resp.json().await.map_err(Error::ReceiveBody)
     }
 
     pub async fn snapshot(

safekeeper/src/bin/safekeeper.rs

@@ -23,7 +23,6 @@ use safekeeper::defaults::{
     DEFAULT_PARTIAL_BACKUP_CONCURRENCY, DEFAULT_PARTIAL_BACKUP_TIMEOUT, DEFAULT_PG_LISTEN_ADDR,
     DEFAULT_SSL_CERT_FILE, DEFAULT_SSL_CERT_RELOAD_PERIOD, DEFAULT_SSL_KEY_FILE,
 };
-use safekeeper::hadron;
 use safekeeper::wal_backup::WalBackup;
 use safekeeper::{
     BACKGROUND_RUNTIME, BROKER_RUNTIME, GlobalTimelines, HTTP_RUNTIME, SafeKeeperConf,
@@ -253,10 +252,6 @@ struct Args {
     /// Run in development mode (disables security checks)
     #[arg(long, help = "Run in development mode (disables security checks)")]
     dev: bool,
-    /* BEGIN_HADRON */
-    #[arg(long)]
-    enable_pull_timeline_on_startup: bool,
-    /* END_HADRON */
 }
 
 // Like PathBufValueParser, but allows empty string.
@@ -440,11 +435,6 @@ async fn main() -> anyhow::Result<()> {
         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,
-        /* BEGIN_HADRON */
-        advertise_pg_addr_tenant_only: None,
-        enable_pull_timeline_on_startup: args.enable_pull_timeline_on_startup,
-        hcc_base_url: None,
-        /* END_HADRON */
     });
 
     // initialize sentry if SENTRY_DSN is provided
@@ -539,20 +529,6 @@ async fn start_safekeeper(conf: Arc<SafeKeeperConf>) -> Result<()> {
     // Load all timelines from disk to memory.
     global_timelines.init().await?;
 
-    /* BEGIN_HADRON */
-    if conf.enable_pull_timeline_on_startup && global_timelines.timelines_count() == 0 {
-        match hadron::hcc_pull_timelines(&conf, global_timelines.clone()).await {
-            Ok(_) => {
-                info!("Successfully pulled all timelines from peer safekeepers");
-            }
-            Err(e) => {
-                error!("Failed to pull timelines from peer safekeepers: {:?}", e);
-                return Err(e);
-            }
-        }
-    }
-    /* END_HADRON */
-
     // Run everything in current thread rt, if asked.
     if conf.current_thread_runtime {
         info!("running in current thread runtime");

safekeeper/src/hadron.rs

@@ -1,388 +0,0 @@
-use pem::Pem;
-use safekeeper_api::models::PullTimelineRequest;
-use std::{collections::HashMap, env::VarError, net::IpAddr, sync::Arc, time::Duration};
-use tokio::time::sleep;
-use tokio_util::sync::CancellationToken;
-use url::Url;
-use utils::{backoff, id::TenantTimelineId, ip_address};
-
-use anyhow::Result;
-use pageserver_api::controller_api::{
-    AvailabilityZone, NodeRegisterRequest, SafekeeperTimeline, SafekeeperTimelinesResponse,
-};
-
-use crate::{
-    GlobalTimelines, SafeKeeperConf,
-    metrics::{
-        SK_RECOVERY_PULL_TIMELINE_ERRORS, SK_RECOVERY_PULL_TIMELINE_OKS,
-        SK_RECOVERY_PULL_TIMELINE_SECONDS, SK_RECOVERY_PULL_TIMELINES_SECONDS,
-    },
-    pull_timeline,
-    timelines_global_map::DeleteOrExclude,
-};
-
-// Extract information in the SafeKeeperConf to build a NodeRegisterRequest used to register the safekeeper with the HCC.
-fn build_node_registeration_request(
-    conf: &SafeKeeperConf,
-    node_ip_addr: Option<IpAddr>,
-) -> Result<NodeRegisterRequest> {
-    let advertise_pg_addr_with_port = conf
-        .advertise_pg_addr_tenant_only
-        .as_deref()
-        .expect("advertise_pg_addr_tenant_only is required to register with HCC");
-
-    // Extract host/port from the string.
-    let (advertise_host_addr, pg_port_str) = advertise_pg_addr_with_port.split_at(
-        advertise_pg_addr_with_port
-            .rfind(':')
-            .ok_or(anyhow::anyhow!("Invalid advertise_pg_addr"))?,
-    );
-    // Need the `[1..]` to remove the leading ':'.
-    let pg_port = pg_port_str[1..]
-        .parse::<u16>()
-        .map_err(|e| anyhow::anyhow!("Cannot parse PG port: {}", e))?;
-
-    let (_, http_port_str) = conf.listen_http_addr.split_at(
-        conf.listen_http_addr
-            .rfind(':')
-            .ok_or(anyhow::anyhow!("Invalid listen_http_addr"))?,
-    );
-    let http_port = http_port_str[1..]
-        .parse::<u16>()
-        .map_err(|e| anyhow::anyhow!("Cannot parse HTTP port: {}", e))?;
-
-    Ok(NodeRegisterRequest {
-        node_id: conf.my_id,
-        listen_pg_addr: advertise_host_addr.to_string(),
-        listen_pg_port: pg_port,
-        listen_http_addr: advertise_host_addr.to_string(),
-        listen_http_port: http_port,
-        node_ip_addr,
-        availability_zone_id: AvailabilityZone("todo".to_string()),
-        listen_grpc_addr: None,
-        listen_grpc_port: None,
-        listen_https_port: None,
-    })
-}
-
-// Retrieve the JWT token used for authenticating with HCC from the environment variable.
-// Returns None if the token cannot be retrieved.
-fn get_hcc_auth_token() -> Option<String> {
-    match std::env::var("HCC_AUTH_TOKEN") {
-        Ok(v) => {
-            tracing::info!("Loaded JWT token for authentication with HCC");
-            Some(v)
-        }
-        Err(VarError::NotPresent) => {
-            tracing::info!("No JWT token for authentication with HCC detected");
-            None
-        }
-        Err(_) => {
-            tracing::info!(
-                "Failed to either load to detect non-present HCC_AUTH_TOKEN environment variable"
-            );
-            None
-        }
-    }
-}
-
-async fn send_safekeeper_register_request(
-    request_url: &Url,
-    auth_token: &Option<String>,
-    request: &NodeRegisterRequest,
-) -> Result<()> {
-    let client = reqwest::Client::new();
-    let mut req_builder = client
-        .post(request_url.clone())
-        .header("Content-Type", "application/json");
-    if let Some(token) = auth_token {
-        req_builder = req_builder.bearer_auth(token);
-    }
-    req_builder
-        .json(&request)
-        .send()
-        .await?
-        .error_for_status()?;
-    Ok(())
-}
-
-/// Registers this safe keeper with the HCC.
-pub async fn register(conf: &SafeKeeperConf) -> Result<()> {
-    match conf.hcc_base_url.as_ref() {
-        None => {
-            tracing::info!("HCC base URL is not set, skipping registration");
-            Ok(())
-        }
-        Some(hcc_base_url) => {
-            // The following operations acquiring the auth token and the node IP address both read environment
-            // variables. It's fine for now as this `register()` function is only called once during startup.
-            // If we start to talk to HCC more regularly in the safekeeper we should probably consider
-            // refactoring things into a "HadronClusterCoordinatorClient" struct.
-            let auth_token = get_hcc_auth_token();
-            let node_ip_addr =
-                ip_address::read_node_ip_addr_from_env().expect("Error reading node IP address.");
-
-            let request = build_node_registeration_request(conf, node_ip_addr)?;
-            let cancel = CancellationToken::new();
-            let request_url = hcc_base_url.clone().join("/hadron-internal/v1/sk")?;
-
-            backoff::retry(
-                || async {
-                    send_safekeeper_register_request(&request_url, &auth_token, &request).await
-                },
-                |_| false,
-                3,
-                u32::MAX,
-                "Calling the HCC safekeeper register API",
-                &cancel,
-            )
-            .await
-            .ok_or(anyhow::anyhow!(
-                "Error in forever retry loop. This error should never be surfaced."
-            ))?
-        }
-    }
-}
-
-async fn safekeeper_list_timelines_request(
-    conf: &SafeKeeperConf,
-) -> Result<pageserver_api::controller_api::SafekeeperTimelinesResponse> {
-    if conf.hcc_base_url.is_none() {
-        tracing::info!("HCC base URL is not set, skipping registration");
-        return Err(anyhow::anyhow!("HCC base URL is not set"));
-    }
-
-    // The following operations acquiring the auth token and the node IP address both read environment
-    // variables. It's fine for now as this `register()` function is only called once during startup.
-    // If we start to talk to HCC more regularly in the safekeeper we should probably consider
-    // refactoring things into a "HadronClusterCoordinatorClient" struct.
-    let auth_token = get_hcc_auth_token();
-    let method = format!("/control/v1/safekeeper/{}/timelines", conf.my_id.0);
-    let request_url = conf.hcc_base_url.as_ref().unwrap().clone().join(&method)?;
-
-    let client = reqwest::Client::new();
-    let mut req_builder = client
-        .get(request_url.clone())
-        .header("Content-Type", "application/json")
-        .query(&[("id", conf.my_id.0)]);
-    if let Some(token) = auth_token {
-        req_builder = req_builder.bearer_auth(token);
-    }
-    let response = req_builder
-        .send()
-        .await?
-        .error_for_status()?
-        .json::<pageserver_api::controller_api::SafekeeperTimelinesResponse>()
-        .await?;
-    Ok(response)
-}
-
-// Returns true on success, false otherwise.
-pub async fn hcc_pull_timeline(
-    timeline: SafekeeperTimeline,
-    conf: &SafeKeeperConf,
-    global_timelines: Arc<GlobalTimelines>,
-    nodeid_http: &HashMap<u64, String>,
-) -> bool {
-    let mut request = PullTimelineRequest {
-        tenant_id: timeline.tenant_id,
-        timeline_id: timeline.timeline_id,
-        http_hosts: Vec::new(),
-        ignore_tombstone: None,
-    };
-    for host in timeline.peers {
-        if host.0 == conf.my_id.0 {
-            continue;
-        }
-        if let Some(http_host) = nodeid_http.get(&host.0) {
-            request.http_hosts.push(http_host.clone());
-        }
-    }
-
-    let ca_certs = match conf
-        .ssl_ca_certs
-        .iter()
-        .map(Pem::contents)
-        .map(reqwest::Certificate::from_der)
-        .collect::<Result<Vec<_>, _>>()
-    {
-        Ok(result) => result,
-        Err(_) => {
-            return false;
-        }
-    };
-    match pull_timeline::handle_request(
-        request,
-        conf.sk_auth_token.clone(),
-        ca_certs,
-        global_timelines.clone(),
-        true,
-    )
-    .await
-    {
-        Ok(resp) => {
-            tracing::info!(
-                "Completed pulling tenant {} timeline {} from SK {:?}",
-                timeline.tenant_id,
-                timeline.timeline_id,
-                resp.safekeeper_host
-            );
-            return true;
-        }
-        Err(e) => {
-            tracing::error!(
-                "Failed to pull tenant {} timeline {} from SK {}",
-                timeline.tenant_id,
-                timeline.timeline_id,
-                e
-            );
-
-            let ttid = TenantTimelineId {
-                tenant_id: timeline.tenant_id,
-                timeline_id: timeline.timeline_id,
-            };
-            // Revert the failed timeline pull.
-            // Notice that not found timeline returns OK also.
-            match global_timelines
-                .delete_or_exclude(&ttid, DeleteOrExclude::DeleteLocal)
-                .await
-            {
-                Ok(dr) => {
-                    tracing::info!(
-                        "Deleted tenant {} timeline {} DirExists: {}",
-                        timeline.tenant_id,
-                        timeline.timeline_id,
-                        dr.dir_existed,
-                    );
-                }
-                Err(e) => {
-                    tracing::error!(
-                        "Failed to delete tenant {} timeline {} from global_timelines: {}",
-                        timeline.tenant_id,
-                        timeline.timeline_id,
-                        e
-                    );
-                }
-            }
-        }
-    }
-    false
-}
-
-pub async fn hcc_pull_timeline_till_success(
-    timeline: SafekeeperTimeline,
-    conf: &SafeKeeperConf,
-    global_timelines: Arc<GlobalTimelines>,
-    nodeid_http: &HashMap<u64, String>,
-) {
-    const MAX_PULL_TIMELINE_RETRIES: u64 = 100;
-    for i in 0..MAX_PULL_TIMELINE_RETRIES {
-        if hcc_pull_timeline(
-            timeline.clone(),
-            conf,
-            global_timelines.clone(),
-            nodeid_http,
-        )
-        .await
-        {
-            SK_RECOVERY_PULL_TIMELINE_OKS.inc();
-            return;
-        }
-        tracing::error!(
-            "Failed to pull timeline {} from SK peers, retrying {}/{}",
-            timeline.timeline_id,
-            i + 1,
-            MAX_PULL_TIMELINE_RETRIES
-        );
-        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
-    }
-    SK_RECOVERY_PULL_TIMELINE_ERRORS.inc();
-}
-
-pub async fn hcc_pull_timelines(
-    conf: &SafeKeeperConf,
-    global_timelines: Arc<GlobalTimelines>,
-) -> Result<()> {
-    let _timer = SK_RECOVERY_PULL_TIMELINES_SECONDS.start_timer();
-    tracing::info!("Start pulling timelines from SK peers");
-
-    let mut response = SafekeeperTimelinesResponse {
-        timelines: Vec::new(),
-        safekeeper_peers: Vec::new(),
-    };
-    for i in 0..100 {
-        match safekeeper_list_timelines_request(conf).await {
-            Ok(timelines) => {
-                response = timelines;
-            }
-            Err(e) => {
-                tracing::error!("Failed to list timelines from HCC: {}", e);
-                if i == 99 {
-                    return Err(e);
-                }
-            }
-        }
-        sleep(Duration::from_millis(100)).await;
-    }
-
-    let mut nodeid_http = HashMap::new();
-    for sk in response.safekeeper_peers {
-        nodeid_http.insert(
-            sk.node_id.0,
-            format!("http://{}:{}", sk.listen_http_addr, sk.http_port),
-        );
-    }
-    tracing::info!("Received {} timelines from HCC", response.timelines.len());
-    for timeline in response.timelines {
-        let _timer = SK_RECOVERY_PULL_TIMELINE_SECONDS
-            .with_label_values(&[
-                &timeline.tenant_id.to_string(),
-                &timeline.timeline_id.to_string(),
-            ])
-            .start_timer();
-        hcc_pull_timeline_till_success(timeline, conf, global_timelines.clone(), &nodeid_http)
-            .await;
-    }
-    Ok(())
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use utils::id::NodeId;
-
-    #[test]
-    fn test_build_node_registeration_request() {
-        // Test that:
-        // 1. We always extract the host name and port used to register with the HCC from the
-        //    `advertise_pg_addr` if it is set.
-        // 2. The correct ports are extracted from `advertise_pg_addr` and `listen_http_addr`.
-        let mut conf = SafeKeeperConf::dummy();
-        conf.my_id = NodeId(1);
-        conf.advertise_pg_addr_tenant_only =
-            Some("safe-keeper-1.safe-keeper.hadron.svc.cluster.local:5454".to_string());
-        // `listen_pg_addr` and `listen_pg_addr_tenant_only` are not used for node registration. Set them to a different
-        // host and port values and make sure that they don't show up in the node registration request.
-        conf.listen_pg_addr = "0.0.0.0:5456".to_string();
-        conf.listen_pg_addr_tenant_only = Some("0.0.0.0:5456".to_string());
-        conf.listen_http_addr = "0.0.0.0:7676".to_string();
-        let node_ip_addr: Option<IpAddr> = Some("127.0.0.1".parse().unwrap());
-
-        let request = build_node_registeration_request(&conf, node_ip_addr).unwrap();
-        assert_eq!(request.node_id, NodeId(1));
-        assert_eq!(
-            request.listen_pg_addr,
-            "safe-keeper-1.safe-keeper.hadron.svc.cluster.local"
-        );
-        assert_eq!(request.listen_pg_port, 5454);
-        assert_eq!(
-            request.listen_http_addr,
-            "safe-keeper-1.safe-keeper.hadron.svc.cluster.local"
-        );
-        assert_eq!(request.listen_http_port, 7676);
-        assert_eq!(
-            request.node_ip_addr,
-            Some(IpAddr::V4("127.0.0.1".parse().unwrap()))
-        );
-    }
-}

safekeeper/src/http/routes.rs

@@ -241,14 +241,9 @@ async fn timeline_pull_handler(mut request: Request<Body>) -> Result<Response<Bo
             ApiError::InternalServerError(anyhow::anyhow!("failed to parse CA certs: {e}"))
         })?;
 
-    let resp = pull_timeline::handle_request(
-        data,
-        conf.sk_auth_token.clone(),
-        ca_certs,
-        global_timelines,
-        false,
-    )
-    .await?;
+    let resp =
+        pull_timeline::handle_request(data, conf.sk_auth_token.clone(), ca_certs, global_timelines)
+            .await?;
     json_response(StatusCode::OK, resp)
 }
 

safekeeper/src/lib.rs

@@ -10,7 +10,6 @@ use pem::Pem;
 use remote_storage::RemoteStorageConfig;
 use storage_broker::Uri;
 use tokio::runtime::Runtime;
-use url::Url;
 use utils::auth::SwappableJwtAuth;
 use utils::id::NodeId;
 use utils::logging::SecretString;
@@ -21,7 +20,6 @@ pub mod control_file;
 pub mod control_file_upgrade;
 pub mod copy_timeline;
 pub mod debug_dump;
-pub mod hadron;
 pub mod handler;
 pub mod http;
 pub mod metrics;
@@ -102,11 +100,6 @@ pub struct SafeKeeperConf {
     pub advertise_pg_addr: Option<String>,
     pub availability_zone: Option<String>,
     pub no_sync: bool,
-    /* BEGIN_HADRON */
-    pub advertise_pg_addr_tenant_only: Option<String>,
-    pub enable_pull_timeline_on_startup: bool,
-    pub hcc_base_url: Option<Url>,
-    /* END_HADRON */
     pub broker_endpoint: Uri,
     pub broker_keepalive_interval: Duration,
     pub heartbeat_timeout: Duration,
@@ -192,11 +185,6 @@ impl SafeKeeperConf {
             use_https_safekeeper_api: false,
             enable_tls_wal_service_api: false,
             force_metric_collection_on_scrape: true,
-            /* BEGIN_HADRON */
-            advertise_pg_addr_tenant_only: None,
-            enable_pull_timeline_on_startup: false,
-            hcc_base_url: None,
-            /* END_HADRON */
         }
     }
 }

safekeeper/src/metrics.rs

@@ -85,43 +85,6 @@ pub static WAL_STORAGE_LIMIT_ERRORS: Lazy<IntCounter> = Lazy::new(|| {
     )
     .expect("Failed to register safekeeper_wal_storage_limit_errors counter")
 });
-pub static SK_RECOVERY_PULL_TIMELINE_ERRORS: Lazy<IntCounter> = Lazy::new(|| {
-    register_int_counter!(
-        "safekeeper_recovery_pull_timeline_errors",
-        concat!(
-            "Number of errors due to pull_timeline errors during SK lost disk recovery.",
-            "An increase in this metric indicates pull timelines runs into error."
-        )
-    )
-    .expect("Failed to register safekeeper_recovery_pull_timeline_errors counter")
-});
-pub static SK_RECOVERY_PULL_TIMELINE_OKS: Lazy<IntCounter> = Lazy::new(|| {
-    register_int_counter!(
-        "safekeeper_recovery_pull_timeline_oks",
-        concat!(
-            "Number of successful pull_timeline during SK lost disk recovery.",
-            "An increase in this metric indicates pull timelines is successful."
-        )
-    )
-    .expect("Failed to register safekeeper_recovery_pull_timeline_oks counter")
-});
-pub static SK_RECOVERY_PULL_TIMELINES_SECONDS: Lazy<Histogram> = Lazy::new(|| {
-    register_histogram!(
-        "safekeeper_recovery_pull_timelines_seconds",
-        "Seconds to pull timelines",
-        DISK_FSYNC_SECONDS_BUCKETS.to_vec()
-    )
-    .expect("Failed to register safekeeper_recovery_pull_timelines_seconds histogram")
-});
-pub static SK_RECOVERY_PULL_TIMELINE_SECONDS: Lazy<HistogramVec> = Lazy::new(|| {
-    register_histogram_vec!(
-        "safekeeper_recovery_pull_timeline_seconds",
-        "Seconds to pull timeline",
-        &["tenant_id", "timeline_id"],
-        DISK_FSYNC_SECONDS_BUCKETS.to_vec()
-    )
-    .expect("Failed to register safekeeper_recovery_pull_timeline_seconds histogram vec")
-});
 /* END_HADRON */
 pub static PERSIST_CONTROL_FILE_SECONDS: Lazy<Histogram> = Lazy::new(|| {
     register_histogram!(

safekeeper/src/pull_timeline.rs

@@ -8,7 +8,6 @@ use bytes::Bytes;
 use camino::Utf8PathBuf;
 use chrono::{DateTime, Utc};
 use futures::{SinkExt, StreamExt, TryStreamExt};
-use http::StatusCode;
 use http_utils::error::ApiError;
 use postgres_ffi::{PG_TLI, XLogFileName, XLogSegNo};
 use remote_storage::GenericRemoteStorage;
@@ -22,11 +21,10 @@ use tokio::fs::OpenOptions;
 use tokio::io::AsyncWrite;
 use tokio::sync::mpsc;
 use tokio::task;
-use tokio::time::sleep;
 use tokio_tar::{Archive, Builder, Header};
 use tokio_util::io::{CopyToBytes, SinkWriter};
 use tokio_util::sync::PollSender;
-use tracing::{error, info, instrument, warn};
+use tracing::{error, info, instrument};
 use utils::crashsafe::fsync_async_opt;
 use utils::id::{NodeId, TenantTimelineId};
 use utils::logging::SecretString;
@@ -451,7 +449,6 @@ pub async fn handle_request(
     sk_auth_token: Option<SecretString>,
     ssl_ca_certs: Vec<Certificate>,
     global_timelines: Arc<GlobalTimelines>,
-    wait_for_peer_timeline_status: bool,
 ) -> Result<PullTimelineResponse, ApiError> {
     let existing_tli = global_timelines.get(TenantTimelineId::new(
         request.tenant_id,
@@ -475,100 +472,37 @@ pub async fn handle_request(
     let http_hosts = request.http_hosts.clone();
 
     // Figure out statuses of potential donors.
+    let responses: Vec<Result<TimelineStatus, mgmt_api::Error>> =
+        futures::future::join_all(http_hosts.iter().map(|url| async {
+            let cclient = Client::new(http_client.clone(), url.clone(), sk_auth_token.clone());
+            let info = cclient
+                .timeline_status(request.tenant_id, request.timeline_id)
+                .await?;
+            Ok(info)
+        }))
+        .await;
+
     let mut statuses = Vec::new();
-    if !wait_for_peer_timeline_status {
-        let responses: Vec<Result<TimelineStatus, mgmt_api::Error>> =
-            futures::future::join_all(http_hosts.iter().map(|url| async {
-                let cclient = Client::new(http_client.clone(), url.clone(), sk_auth_token.clone());
-                let resp = cclient
-                    .timeline_status(request.tenant_id, request.timeline_id)
-                    .await?;
-                let info: TimelineStatus = resp
-                    .json()
-                    .await
-                    .context("Failed to deserialize timeline status")
-                    .map_err(|e| mgmt_api::Error::ReceiveErrorBody(e.to_string()))?;
-                Ok(info)
-            }))
-            .await;
-
-        for (i, response) in responses.into_iter().enumerate() {
-            match response {
-                Ok(status) => {
-                    statuses.push((status, i));
-                }
-                Err(e) => {
-                    info!("error fetching status from {}: {e}", http_hosts[i]);
-                }
+    for (i, response) in responses.into_iter().enumerate() {
+        match response {
+            Ok(status) => {
+                statuses.push((status, i));
+            }
+            Err(e) => {
+                info!("error fetching status from {}: {e}", http_hosts[i]);
             }
         }
+    }
 
-        // Allow missing responses from up to one safekeeper (say due to downtime)
-        // e.g. if we created a timeline on PS A and B, with C being offline. Then B goes
-        // offline and C comes online. Then we want a pull on C with A and B as hosts to work.
-        let min_required_successful = (http_hosts.len() - 1).max(1);
-        if statuses.len() < min_required_successful {
-            return Err(ApiError::InternalServerError(anyhow::anyhow!(
-                "only got {} successful status responses. required: {min_required_successful}",
-                statuses.len()
-            )));
-        }
-    } else {
-        let mut retry = true;
-        // We must get status from all other peers.
-        // Otherwise, we may run into split-brain scenario.
-        while retry {
-            statuses.clear();
-            retry = false;
-            for (i, url) in http_hosts.iter().enumerate() {
-                let cclient = Client::new(http_client.clone(), url.clone(), sk_auth_token.clone());
-                match cclient
-                    .timeline_status(request.tenant_id, request.timeline_id)
-                    .await
-                {
-                    Ok(resp) => {
-                        if resp.status() == StatusCode::NOT_FOUND {
-                            warn!(
-                                "Timeline {} not found on peer SK {}, no need to pull it",
-                                TenantTimelineId::new(request.tenant_id, request.timeline_id),
-                                url
-                            );
-                            return Ok(PullTimelineResponse {
-                                safekeeper_host: None,
-                            });
-                        }
-                        let info: TimelineStatus = resp
-                            .json()
-                            .await
-                            .context("Failed to deserialize timeline status")
-                            .map_err(ApiError::InternalServerError)?;
-                        statuses.push((info, i));
-                    }
-                    Err(e) => {
-                        match e {
-                            // If we get a 404, it means the timeline doesn't exist on this safekeeper.
-                            // We can ignore this error.
-                            mgmt_api::Error::ApiError(status, _)
-                                if status == StatusCode::NOT_FOUND =>
-                            {
-                                warn!(
-                                    "Timeline {} not found on peer SK {}, no need to pull it",
-                                    TenantTimelineId::new(request.tenant_id, request.timeline_id),
-                                    url
-                                );
-                                return Ok(PullTimelineResponse {
-                                    safekeeper_host: None,
-                                });
-                            }
-                            _ => {}
-                        }
-                        retry = true;
-                        error!("Failed to get timeline status from {}: {:#}", url, e);
-                    }
-                }
-            }
-            sleep(std::time::Duration::from_millis(100)).await;
-        }
+    // Allow missing responses from up to one safekeeper (say due to downtime)
+    // e.g. if we created a timeline on PS A and B, with C being offline. Then B goes
+    // offline and C comes online. Then we want a pull on C with A and B as hosts to work.
+    let min_required_successful = (http_hosts.len() - 1).max(1);
+    if statuses.len() < min_required_successful {
+        return Err(ApiError::InternalServerError(anyhow::anyhow!(
+            "only got {} successful status responses. required: {min_required_successful}",
+            statuses.len()
+        )));
     }
 
     // Find the most advanced safekeeper
@@ -577,12 +511,6 @@ pub async fn handle_request(
         .max_by_key(|(status, _)| {
             (
                 status.acceptor_state.epoch,
-                /* BEGIN_HADRON */
-                // We need to pull from the SK with the highest term.
-                // This is because another compute may come online and vote the same highest term again on the other two SKs.
-                // Then, there will be 2 computes running on the same term.
-                status.acceptor_state.term,
-                /* END_HADRON */
                 status.flush_lsn,
                 status.commit_lsn,
             )

safekeeper/tests/walproposer_sim/safekeeper.rs

@@ -191,11 +191,6 @@ pub fn run_server(os: NodeOs, disk: Arc<SafekeeperDisk>) -> Result<()> {
         use_https_safekeeper_api: false,
         enable_tls_wal_service_api: false,
         force_metric_collection_on_scrape: true,
-        /* BEGIN_HADRON */
-        enable_pull_timeline_on_startup: false,
-        advertise_pg_addr_tenant_only: None,
-        hcc_base_url: None,
-        /* END_HADRON */
     };
 
     let mut global = GlobalMap::new(disk, conf.clone())?;

test_runner/fixtures/endpoint/http.py

@@ -66,12 +66,6 @@ class EndpointHttpClient(requests.Session):
         res.raise_for_status()
         return res.json()
 
-    def autoscaling_metrics(self):
-        res = self.get(f"http://localhost:{self.external_port}/autoscaling_metrics")
-        res.raise_for_status()
-        log.debug("raw compute metrics: %s", res.text)
-        return res.text
-
     def prewarm_lfc_status(self) -> dict[str, str]:
         res = self.get(self.prewarm_url)
         res.raise_for_status()

test_runner/fixtures/neon_api.py

@@ -34,9 +34,7 @@ class NeonAPI:
         self.retries524 = 0
         self.retries4xx = 0
 
-    def __request(
-        self, method: str | bytes, endpoint: str, retry404: bool = False, **kwargs: Any
-    ) -> requests.Response:
+    def __request(self, method: str | bytes, endpoint: str, **kwargs: Any) -> requests.Response:
         kwargs["headers"] = kwargs.get("headers", {})
         kwargs["headers"]["Authorization"] = f"Bearer {self.__neon_api_key}"
 
@@ -57,12 +55,10 @@ class NeonAPI:
                 resp.raise_for_status()
                 break
             elif resp.status_code >= 400:
-                if resp.status_code == 404 and retry404:
-                    retry = True
-                    self.retries4xx += 1
-                elif resp.status_code == 422 and resp.json()["message"] == "branch not ready yet":
-                    retry = True
-                    self.retries4xx += 1
+                if resp.status_code == 422:
+                    if resp.json()["message"] == "branch not ready yet":
+                        retry = True
+                        self.retries4xx += 1
                 elif resp.status_code == 423 and resp.json()["message"] in {
                     "endpoint is in some transitive state, could not suspend",
                     "project already has running conflicting operations, scheduling of new ones is prohibited",
@@ -70,7 +66,7 @@ class NeonAPI:
                     retry = True
                     self.retries4xx += 1
                 elif resp.status_code == 524:
-                    log.info("The request was timed out")
+                    log.info("The request was timed out, trying to get operations")
                     retry = True
                     self.retries524 += 1
             if retry:
@@ -207,9 +203,6 @@ class NeonAPI:
         resp = self.__request(
             "GET",
             f"/projects/{project_id}/branches/{branch_id}",
-            # XXX Retry get parent details to work around the issue
-            # https://databricks.atlassian.net/browse/LKB-279
-            retry404=True,
             headers={
                 "Accept": "application/json",
             },
@@ -314,10 +307,6 @@ class NeonAPI:
         if endpoint_type:
             data["endpoint"]["type"] = endpoint_type
         if settings:
-            # otherwise we get 400 "settings must not be nil"
-            # TODO(myrrc): fix on cplane side
-            if "pg_settings" not in settings:
-                settings["pg_settings"] = {}
             data["endpoint"]["settings"] = settings
 
         resp = self.__request(

test_runner/fixtures/neon_fixtures.py

@@ -5417,7 +5417,6 @@ SKIP_FILES = frozenset(
         "postmaster.pid",
         "pg_control",
         "pg_dynshmem",
-        "neon-communicator.socket",
     )
 )
 

test_runner/performance/test_lfc_prewarm.py

@@ -1,168 +0,0 @@
-from __future__ import annotations
-
-import os
-import timeit
-import traceback
-from concurrent.futures import ThreadPoolExecutor as Exec
-from pathlib import Path
-from time import sleep
-from typing import TYPE_CHECKING, Any, cast
-
-import pytest
-from fixtures.benchmark_fixture import NeonBenchmarker, PgBenchRunResult
-from fixtures.log_helper import log
-from fixtures.neon_api import NeonAPI, connection_parameters_to_env
-
-if TYPE_CHECKING:
-    from fixtures.compare_fixtures import NeonCompare
-    from fixtures.neon_fixtures import Endpoint, PgBin
-    from fixtures.pg_version import PgVersion
-
-from performance.test_perf_pgbench import utc_now_timestamp
-
-# These tests compare performance for a write-heavy and read-heavy workloads of an ordinary endpoint
-# compared to the endpoint which saves its LFC and prewarms using it on startup.
-
-
-def test_compare_prewarmed_pgbench_perf(neon_compare: NeonCompare):
-    env = neon_compare.env
-    env.create_branch("normal")
-    env.create_branch("prewarmed")
-    pg_bin = neon_compare.pg_bin
-    ep_normal: Endpoint = env.endpoints.create_start("normal")
-    ep_prewarmed: Endpoint = env.endpoints.create_start("prewarmed", autoprewarm=True)
-
-    for ep in [ep_normal, ep_prewarmed]:
-        connstr: str = ep.connstr()
-        pg_bin.run(["pgbench", "-i", "-I", "dtGvp", connstr, "-s100"])
-        ep.safe_psql("CREATE EXTENSION neon")
-        client = ep.http_client()
-        client.offload_lfc()
-        ep.stop()
-        ep.start()
-        client.prewarm_lfc_wait()
-
-        run_start_timestamp = utc_now_timestamp()
-        t0 = timeit.default_timer()
-        out = pg_bin.run_capture(["pgbench", "-c10", "-T10", connstr])
-        run_duration = timeit.default_timer() - t0
-        run_end_timestamp = utc_now_timestamp()
-
-        stdout = Path(f"{out}.stdout").read_text()
-        res = PgBenchRunResult.parse_from_stdout(
-            stdout=stdout,
-            run_duration=run_duration,
-            run_start_timestamp=run_start_timestamp,
-            run_end_timestamp=run_end_timestamp,
-        )
-        name: str = cast("str", ep.branch_name)
-        neon_compare.zenbenchmark.record_pg_bench_result(name, res)
-
-
-@pytest.mark.remote_cluster
-@pytest.mark.timeout(2 * 60 * 60)
-def test_compare_prewarmed_pgbench_perf_benchmark(
-    pg_bin: PgBin,
-    neon_api: NeonAPI,
-    pg_version: PgVersion,
-    zenbenchmark: NeonBenchmarker,
-):
-    name = f"Test prewarmed pgbench performance, GITHUB_RUN_ID={os.getenv('GITHUB_RUN_ID')}"
-    project = neon_api.create_project(pg_version, name)
-    project_id = project["project"]["id"]
-    neon_api.wait_for_operation_to_finish(project_id)
-    err = False
-    try:
-        benchmark_impl(pg_bin, neon_api, project, zenbenchmark)
-    except Exception as e:
-        err = True
-        log.error(f"Caught exception: {e}")
-        log.error(traceback.format_exc())
-    finally:
-        assert not err
-        neon_api.delete_project(project_id)
-
-
-def benchmark_impl(
-    pg_bin: PgBin, neon_api: NeonAPI, project: dict[str, Any], zenbenchmark: NeonBenchmarker
-):
-    pgbench_size = int(os.getenv("PGBENCH_SIZE") or "3424")  # 50GB
-    offload_secs = 20
-    test_duration_min = 5
-    pgbench_duration = f"-T{test_duration_min * 60}"
-    # prewarm API is not publicly exposed. In order to test performance of a
-    # fully prewarmed endpoint, wait after it restarts.
-    # The number here is empirical, based on manual runs on staging
-    prewarmed_sleep_secs = 180
-
-    branch_id = project["branch"]["id"]
-    project_id = project["project"]["id"]
-    normal_env = connection_parameters_to_env(
-        project["connection_uris"][0]["connection_parameters"]
-    )
-    normal_id = project["endpoints"][0]["id"]
-
-    prewarmed_branch_id = neon_api.create_branch(
-        project_id, "prewarmed", parent_id=branch_id, add_endpoint=False
-    )["branch"]["id"]
-    neon_api.wait_for_operation_to_finish(project_id)
-
-    ep_prewarmed = neon_api.create_endpoint(
-        project_id,
-        prewarmed_branch_id,
-        endpoint_type="read_write",
-        settings={"autoprewarm": True, "offload_lfc_interval_seconds": offload_secs},
-    )
-    neon_api.wait_for_operation_to_finish(project_id)
-
-    prewarmed_env = normal_env.copy()
-    prewarmed_env["PGHOST"] = ep_prewarmed["endpoint"]["host"]
-    prewarmed_id = ep_prewarmed["endpoint"]["id"]
-
-    def bench(endpoint_name, endpoint_id, env):
-        pg_bin.run(["pgbench", "-i", "-I", "dtGvp", f"-s{pgbench_size}"], env)
-        sleep(offload_secs * 2)  # ensure LFC is offloaded after pgbench finishes
-        neon_api.restart_endpoint(project_id, endpoint_id)
-        sleep(prewarmed_sleep_secs)
-
-        run_start_timestamp = utc_now_timestamp()
-        t0 = timeit.default_timer()
-        out = pg_bin.run_capture(["pgbench", "-c10", pgbench_duration, "-Mprepared"], env)
-        run_duration = timeit.default_timer() - t0
-        run_end_timestamp = utc_now_timestamp()
-
-        stdout = Path(f"{out}.stdout").read_text()
-        res = PgBenchRunResult.parse_from_stdout(
-            stdout=stdout,
-            run_duration=run_duration,
-            run_start_timestamp=run_start_timestamp,
-            run_end_timestamp=run_end_timestamp,
-        )
-        zenbenchmark.record_pg_bench_result(endpoint_name, res)
-
-    with Exec(max_workers=2) as exe:
-        exe.submit(bench, "normal", normal_id, normal_env)
-        exe.submit(bench, "prewarmed", prewarmed_id, prewarmed_env)
-
-
-def test_compare_prewarmed_read_perf(neon_compare: NeonCompare):
-    env = neon_compare.env
-    env.create_branch("normal")
-    env.create_branch("prewarmed")
-    ep_normal: Endpoint = env.endpoints.create_start("normal")
-    ep_prewarmed: Endpoint = env.endpoints.create_start("prewarmed", autoprewarm=True)
-
-    sql = [
-        "CREATE EXTENSION neon",
-        "CREATE TABLE foo(key serial primary key, t text default 'foooooooooooooooooooooooooooooooooooooooooooooooooooo')",
-        "INSERT INTO foo SELECT FROM generate_series(1,1000000)",
-    ]
-    for ep in [ep_normal, ep_prewarmed]:
-        ep.safe_psql_many(sql)
-        client = ep.http_client()
-        client.offload_lfc()
-        ep.stop()
-        ep.start()
-        client.prewarm_lfc_wait()
-        with neon_compare.record_duration(f"{ep.branch_name}_run_duration"):
-            ep.safe_psql("SELECT count(*) from foo")

test_runner/random_ops/test_random_ops.py

@@ -13,6 +13,7 @@ from typing import TYPE_CHECKING, Any
 
 import pytest
 from fixtures.log_helper import log
+from requests import HTTPError
 
 if TYPE_CHECKING:
     from pathlib import Path
@@ -152,11 +153,26 @@ 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, self.neon_api.get_branch_details(self.project_id, parent_id), True
-        )
+        parent = NeonBranch(self.project, parent_def, True)
         self.project.branches[parent_id] = parent
         self.parent = parent
         parent.children[self.id] = self

test_runner/regress/test_communicator_metrics_exporter.py

@@ -1,54 +0,0 @@
-from __future__ import annotations
-
-import os
-from typing import TYPE_CHECKING
-
-import pytest
-import requests
-import requests_unixsocket  # type: ignore [import-untyped]
-from fixtures.metrics import parse_metrics
-
-if TYPE_CHECKING:
-    from fixtures.neon_fixtures import NeonEnv
-
-NEON_COMMUNICATOR_SOCKET_NAME = "neon-communicator.socket"
-
-
-def test_communicator_metrics(neon_simple_env: NeonEnv):
-    """
-    Test the communicator's built-in HTTP prometheus exporter
-    """
-    env = neon_simple_env
-
-    endpoint = env.endpoints.create("main")
-    endpoint.start()
-
-    # Change current directory to the data directory, so that we can use
-    # a short relative path to refer to the socket. (There's a 100 char
-    # limitation on the path.)
-    os.chdir(str(endpoint.pgdata_dir))
-    session = requests_unixsocket.Session()
-    r = session.get(f"http+unix://{NEON_COMMUNICATOR_SOCKET_NAME}/metrics")
-    assert r.status_code == 200
-
-    # quick test that the endpoint returned something expected. (We don't validate
-    # that the metrics returned are sensible.)
-    m = parse_metrics(r.text)
-    m.query_one("lfc_hits")
-    m.query_one("lfc_misses")
-
-    # Test panic handling. The /debug/panic endpoint raises a Rust panic. It's
-    # expected to unwind and drop the HTTP connection without response, but not
-    # kill the process or the server.
-    with pytest.raises(
-        requests.ConnectionError, match="Remote end closed connection without response"
-    ):
-        r = session.get(f"http+unix://{NEON_COMMUNICATOR_SOCKET_NAME}/debug/panic")
-        assert r.status_code == 500
-
-    # Test that subsequent requests after the panic still work.
-    r = session.get(f"http+unix://{NEON_COMMUNICATOR_SOCKET_NAME}/metrics")
-    assert r.status_code == 200
-    m = parse_metrics(r.text)
-    m.query_one("lfc_hits")
-    m.query_one("lfc_misses")

test_runner/regress/test_compatibility.py

@@ -197,7 +197,7 @@ def test_create_snapshot(
     shutil.copytree(
         test_output_dir,
         compatibility_snapshot_dir,
-        ignore=shutil.ignore_patterns("pg_dynshmem", "neon-communicator.socket"),
+        ignore=shutil.ignore_patterns("pg_dynshmem"),
     )
 
 

test_runner/regress/test_lfc_working_set_approximation.py

@@ -6,7 +6,6 @@ from typing import TYPE_CHECKING
 
 import pytest
 from fixtures.log_helper import log
-from fixtures.metrics import parse_metrics
 from fixtures.utils import USE_LFC, query_scalar
 
 if TYPE_CHECKING:
@@ -76,24 +75,10 @@ WITH (fillfactor='100');
     cur.execute("SELECT abalance FROM pgbench_accounts WHERE aid = 104242")
     cur.execute("SELECT abalance FROM pgbench_accounts WHERE aid = 204242")
     # verify working set size after some index access of a few select pages only
-    blocks = query_scalar(cur, "select approximate_working_set_size(false)")
+    blocks = query_scalar(cur, "select approximate_working_set_size(true)")
     log.info(f"working set size after some index access of a few select pages only {blocks}")
     assert blocks < 20
 
-    # Also test the metrics from the /autoscaling_metrics endpoint
-    autoscaling_metrics = endpoint.http_client().autoscaling_metrics()
-    log.debug(f"Raw metrics: {autoscaling_metrics}")
-    m = parse_metrics(autoscaling_metrics)
-
-    http_estimate = m.query_one(
-        "lfc_approximate_working_set_size_windows",
-        {
-            "duration_seconds": "60",
-        },
-    ).value
-    log.info(f"http estimate: {http_estimate}, blocks: {blocks}")
-    assert http_estimate > 0 and http_estimate < 20
-
 
 @pytest.mark.skipif(not USE_LFC, reason="LFC is disabled, skipping")
 def test_sliding_working_set_approximation(neon_simple_env: NeonEnv):

test_runner/regress/test_wal_restore.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 import sys
 import tarfile
 import tempfile
-from pathlib import Path
 from typing import TYPE_CHECKING
 
 import pytest
@@ -199,115 +198,3 @@ def test_wal_restore_http(neon_env_builder: NeonEnvBuilder, broken_tenant: bool)
     # the table is back now!
     restored = env.endpoints.create_start("main")
     assert restored.safe_psql("select count(*) from t", user="cloud_admin") == [(300000,)]
-
-
-# BEGIN_HADRON
-# TODO: re-enable once CM python is integreated.
-# def clear_directory(directory):
-#     for item in os.listdir(directory):
-#         item_path = os.path.join(directory, item)
-#         if os.path.isdir(item_path):
-#             log.info(f"removing SK directory: {item_path}")
-#             shutil.rmtree(item_path)
-#         else:
-#             log.info(f"removing SK file: {item_path}")
-#             os.remove(item_path)
-
-
-# def test_sk_pull_timelines(
-#     neon_env_builder: NeonEnvBuilder,
-# ):
-#     DBNAME = "regression"
-#     superuser_name = "databricks_superuser"
-#     neon_env_builder.num_safekeepers = 3
-#     neon_env_builder.num_pageservers = 4
-#     neon_env_builder.safekeeper_extra_opts = ["--enable-pull-timeline-on-startup"]
-#     neon_env_builder.enable_safekeeper_remote_storage(s3_storage())
-
-#     env = neon_env_builder.init_start(initial_tenant_shard_count=4)
-
-#     env.compute_manager.start(base_port=env.compute_manager_port)
-
-#     test_creator = "test_creator"
-#     test_metastore_id = uuid4()
-#     test_account_id = uuid4()
-#     test_workspace_id = 1
-#     test_workspace_url = "http://test_workspace_url"
-#     test_metadata_version = 1
-#     test_metadata = {
-#         "state": "INSTANCE_PROVISIONING",
-#         "admin_rolename": "admin",
-#         "admin_password_scram": "abc123456",
-#     }
-
-#     test_instance_name_1 = "test_instance_1"
-#     test_instance_read_write_compute_pool_1 = {
-#         "instance_name": test_instance_name_1,
-#         "compute_pool_name": "compute_pool_1",
-#         "creator": test_creator,
-#         "capacity": 2.0,
-#         "node_count": 1,
-#         "metadata_version": 0,
-#         "metadata": {
-#             "state": "INSTANCE_PROVISIONING",
-#         },
-#     }
-
-#     test_instance_1_readable_secondaries_enabled = False
-
-#     # Test creation
-#     create_instance_with_retries(
-#         env,
-#         test_instance_name_1,
-#         test_creator,
-#         test_metastore_id,
-#         test_account_id,
-#         test_workspace_id,
-#         test_workspace_url,
-#         test_instance_read_write_compute_pool_1,
-#         test_metadata_version,
-#         test_metadata,
-#         test_instance_1_readable_secondaries_enabled,
-#     )
-#     instance = env.compute_manager.get_instance_by_name(test_instance_name_1, test_workspace_id)
-#     log.info(f"haoyu Instance created: {instance}")
-#     assert instance["instance_name"] == test_instance_name_1
-#     test_instance_id = instance["instance_id"]
-#     instance_detail = env.compute_manager.describe_instance(test_instance_id)
-#     log.info(f"haoyu Instance detail: {instance_detail}")
-
-#     env.initial_tenant = instance_detail[0]["tenant_id"]
-#     env.initial_timeline = instance_detail[0]["timeline_id"]
-
-#     # Connect to postgres and create a database called "regression".
-#     endpoint = env.endpoints.create_start("main")
-#     endpoint.safe_psql(f"CREATE ROLE {superuser_name}")
-#     endpoint.safe_psql(f"CREATE DATABASE {DBNAME}")
-
-#     endpoint.safe_psql("CREATE TABLE usertable ( YCSB_KEY INT, FIELD0 TEXT);")
-#     # Write some data. ~20 MB.
-#     num_rows = 0
-#     for _i in range(0, 20000):
-#         endpoint.safe_psql(
-#             "INSERT INTO usertable SELECT random(), repeat('a', 1000);", log_query=False
-#         )
-#         num_rows += 1
-
-#     log.info(f"SKs {env.storage_controller.hcc_sk_node_list()}")
-
-#     env.safekeepers[0].stop(immediate=True)
-#     clear_directory(env.safekeepers[0].data_dir)
-#     env.safekeepers[0].start()
-
-#     # PG can still write data. ~20 MB.
-#     for _i in range(0, 20000):
-#         endpoint.safe_psql(
-#             "INSERT INTO usertable SELECT random(), repeat('a', 1000);", log_query=False
-#         )
-#         num_rows += 1
-
-#     tuples = endpoint.safe_psql("SELECT COUNT(*) FROM usertable;")
-#     assert tuples[0][0] == num_rows
-#     endpoint.stop_and_destroy()
-
-# END_HADRON

workspace_hack/Cargo.toml

@@ -98,7 +98,7 @@ tikv-jemalloc-sys = { version = "0.6", features = ["profiling", "stats", "unpref
 time = { version = "0.3", features = ["macros", "serde-well-known"] }
 tokio = { version = "1", features = ["full", "test-util"] }
 tokio-rustls = { version = "0.26", default-features = false, features = ["logging", "ring", "tls12"] }
-tokio-stream = { version = "0.1", features = ["net", "sync"] }
+tokio-stream = { version = "0.1", features = ["net"] }
 tokio-util = { version = "0.7", features = ["codec", "compat", "io-util", "rt"] }
 toml_edit = { version = "0.22", features = ["serde"] }
 tower = { version = "0.5", default-features = false, features = ["balance", "buffer", "limit", "log"] }