WIP: Add pg_tracing extension

.dockerignore

@@ -19,7 +19,6 @@
 !pageserver/
 !pgxn/
 !proxy/
-!object_storage/
 !storage_scrubber/
 !safekeeper/
 !storage_broker/

.gitignore

@@ -1,4 +1,3 @@
-/artifact_cache
 /pg_install
 /target
 /tmp_check

Cargo.lock

@@ -2837,7 +2837,6 @@ dependencies = [
  "utils",
  "uuid",
  "workspace_hack",
- "x509-cert",
 ]
 
 [[package]]
@@ -3992,33 +3991,6 @@ dependencies = [
  "memchr",
 ]
 
-[[package]]
-name = "object_storage"
-version = "0.0.1"
-dependencies = [
- "anyhow",
- "axum",
- "axum-extra",
- "camino",
- "camino-tempfile",
- "futures",
- "http-body-util",
- "itertools 0.10.5",
- "jsonwebtoken",
- "prometheus",
- "rand 0.8.5",
- "remote_storage",
- "serde",
- "serde_json",
- "test-log",
- "tokio",
- "tokio-util",
- "tower 0.5.2",
- "tracing",
- "utils",
- "workspace_hack",
-]
-
 [[package]]
 name = "once_cell"
 version = "1.20.2"
@@ -4721,7 +4693,7 @@ dependencies = [
 [[package]]
 name = "postgres-protocol"
 version = "0.6.6"
-source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#f3cf448febde5fd298071d54d568a9c875a7a62b"
+source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#1f21e7959a96a34dcfbfce1b14b73286cdadffe9"
 dependencies = [
  "base64 0.22.1",
  "byteorder",
@@ -4755,7 +4727,7 @@ dependencies = [
 [[package]]
 name = "postgres-types"
 version = "0.2.6"
-source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#f3cf448febde5fd298071d54d568a9c875a7a62b"
+source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#1f21e7959a96a34dcfbfce1b14b73286cdadffe9"
 dependencies = [
  "bytes",
  "chrono",
@@ -6953,28 +6925,6 @@ dependencies = [
  "syn 2.0.100",
 ]
 
-[[package]]
-name = "test-log"
-version = "0.2.17"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e7f46083d221181166e5b6f6b1e5f1d499f3a76888826e6cb1d057554157cd0f"
-dependencies = [
- "env_logger",
- "test-log-macros",
- "tracing-subscriber",
-]
-
-[[package]]
-name = "test-log-macros"
-version = "0.2.17"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "888d0c3c6db53c0fdab160d2ed5e12ba745383d3e85813f2ea0f2b1475ab553f"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.100",
-]
-
 [[package]]
 name = "thiserror"
 version = "1.0.69"
@@ -7222,7 +7172,7 @@ dependencies = [
 [[package]]
 name = "tokio-postgres"
 version = "0.7.10"
-source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#f3cf448febde5fd298071d54d568a9c875a7a62b"
+source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#1f21e7959a96a34dcfbfce1b14b73286cdadffe9"
 dependencies = [
  "async-trait",
  "byteorder",

Cargo.toml

@@ -40,7 +40,6 @@ members = [
     "libs/proxy/postgres-protocol2",
     "libs/proxy/postgres-types2",
     "libs/proxy/tokio-postgres2",
-    "object_storage",
 ]
 
 [workspace.package]
@@ -209,7 +208,6 @@ tracing-opentelemetry = "0.28"
 tracing-serde = "0.2.0"
 tracing-subscriber = { version = "0.3", default-features = false, features = ["smallvec", "fmt", "tracing-log", "std", "env-filter", "json"] }
 try-lock = "0.2.5"
-test-log = { version = "0.2.17", default-features = false, features = ["log"] }
 twox-hash = { version = "1.6.3", default-features = false }
 typed-json = "0.1"
 url = "2.2"

Dockerfile

@@ -89,7 +89,6 @@ RUN set -e \
       --bin storage_broker  \
       --bin storage_controller  \
       --bin proxy  \
-      --bin object_storage \
       --bin neon_local \
       --bin storage_scrubber \
       --locked --release
@@ -122,7 +121,6 @@ COPY --from=build --chown=neon:neon /home/nonroot/target/release/safekeeper
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/storage_broker      /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/storage_controller  /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/proxy               /usr/local/bin
-COPY --from=build --chown=neon:neon /home/nonroot/target/release/object_storage      /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/neon_local          /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/storage_scrubber    /usr/local/bin
 

compute/compute-node.Dockerfile

@@ -1388,6 +1388,38 @@ RUN make -j $(getconf _NPROCESSORS_ONLN) && \
     make -j $(getconf _NPROCESSORS_ONLN) install && \
     echo 'trusted = true' >> /usr/local/pgsql/share/extension/pg_partman.control
 
+#########################################################################################
+# Layer "pg_tracing"
+# compile pg_tracing extension
+#
+#########################################################################################
+FROM build-deps AS pg_tracing-src
+ARG PG_VERSION
+WORKDIR /ext-src
+RUN case "${PG_VERSION:?}" in \
+      "v14" | "v15") \
+        echo "pg_tracing not supported on this PostgreSQL version." && exit 0 \
+	;; \
+      *) \
+      ;; \
+    esac && \
+    wget https://github.com/DataDog/pg_tracing/archive/refs/tags/v0.1.3.tar.gz -O pg_tracing.tar.gz && \
+    echo "d0a7cca7279bb29601ba6c4c1aaeb3a44d71e6afa3b78aae1e3b7269e688f907 pg_tracing.tar.gz" | sha256sum --check && \
+    mkdir pg_tracing-src && cd pg_tracing-src && tar xzf ../pg_tracing.tar.gz --strip-components=1 -C .
+
+FROM pg-build AS pg_tracing-build
+COPY --from=pg_tracing-src /ext-src/ /ext-src/
+WORKDIR /ext-src/pg_tracing-src
+RUN case "${PG_VERSION:?}" in \
+      "v14" | "v15") \
+        echo "pg_tracing not supported on this PostgreSQL version." && exit 0 \
+        ;; \
+      *) \
+      ;; \
+    esac && \
+    make -j $(getconf _NPROCESSORS_ONLN) && \
+    make -j $(getconf _NPROCESSORS_ONLN) install
+
 #########################################################################################
 #
 # Layer "pg_mooncake"
@@ -1617,6 +1649,7 @@ COPY --from=pg_semver-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=wal2json-build /usr/local/pgsql /usr/local/pgsql
 COPY --from=pg_ivm-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg_partman-build /usr/local/pgsql/ /usr/local/pgsql/
+COPY --from=pg_tracing-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg_mooncake-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg_duckdb-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg_repack-build /usr/local/pgsql/ /usr/local/pgsql/
@@ -1677,7 +1710,7 @@ RUN set -e \
     && apt clean && rm -rf /var/lib/apt/lists/*
 
 # Use `dist_man_MANS=` to skip manpage generation (which requires python3/pandoc)
-ENV PGBOUNCER_TAG=pgbouncer_1_24_1
+ENV PGBOUNCER_TAG=pgbouncer_1_22_1
 RUN set -e \
     && git clone --recurse-submodules --depth 1 --branch ${PGBOUNCER_TAG} https://github.com/pgbouncer/pgbouncer.git pgbouncer \
     && cd pgbouncer \
@@ -1793,6 +1826,7 @@ COPY --from=pg_semver-src /ext-src/ /ext-src/
 #COPY --from=wal2json-src /ext-src/ /ext-src/
 COPY --from=pg_ivm-src /ext-src/ /ext-src/
 COPY --from=pg_partman-src /ext-src/ /ext-src/
+COPY --from=pg_tracing-src /ext-src/ /ext-src/
 #COPY --from=pg_mooncake-src /ext-src/ /ext-src/
 COPY --from=pg_repack-src /ext-src/ /ext-src/
 COPY --from=pg_repack-build /usr/local/pgsql/ /usr/local/pgsql/

compute_tools/src/bin/compute_ctl.rs

@@ -29,12 +29,13 @@
 //! ```sh
 //! compute_ctl -D /var/db/postgres/compute \
 //!             -C 'postgresql://cloud_admin@localhost/postgres' \
-//!             -c /var/db/postgres/configs/config.json \
+//!             -S /var/db/postgres/specs/current.json \
 //!             -b /usr/local/bin/postgres \
 //!             -r http://pg-ext-s3-gateway \
 //! ```
 use std::ffi::OsString;
 use std::fs::File;
+use std::path::Path;
 use std::process::exit;
 use std::sync::mpsc;
 use std::thread;
@@ -42,7 +43,8 @@ use std::time::Duration;
 
 use anyhow::{Context, Result};
 use clap::Parser;
-use compute_api::responses::ComputeConfig;
+use compute_api::responses::ComputeCtlConfig;
+use compute_api::spec::ComputeSpec;
 use compute_tools::compute::{
     BUILD_TAG, ComputeNode, ComputeNodeParams, forward_termination_signal,
 };
@@ -116,19 +118,16 @@ struct Cli {
     #[arg(long)]
     pub set_disk_quota_for_fs: Option<String>,
 
-    #[arg(short = 'c', long)]
-    pub config: Option<OsString>,
+    #[arg(short = 's', long = "spec", group = "spec")]
+    pub spec_json: Option<String>,
+
+    #[arg(short = 'S', long, group = "spec-path")]
+    pub spec_path: Option<OsString>,
 
     #[arg(short = 'i', long, group = "compute-id")]
     pub compute_id: String,
 
-    #[arg(
-        short = 'p',
-        long,
-        conflicts_with = "config",
-        value_name = "CONTROL_PLANE_API_BASE_URL",
-        requires = "compute-id"
-    )]
+    #[arg(short = 'p', long, conflicts_with_all = ["spec", "spec-path"], value_name = "CONTROL_PLANE_API_BASE_URL")]
     pub control_plane_uri: Option<String>,
 }
 
@@ -153,7 +152,7 @@ fn main() -> Result<()> {
 
     let connstr = Url::parse(&cli.connstr).context("cannot parse connstr as a URL")?;
 
-    let cli_spec = get_config(&cli)?;
+    let cli_spec = try_spec_from_cli(&cli)?;
 
     let compute_node = ComputeNode::new(
         ComputeNodeParams {
@@ -173,6 +172,7 @@ fn main() -> Result<()> {
             cgroup: cli.cgroup,
             #[cfg(target_os = "linux")]
             vm_monitor_addr: cli.vm_monitor_addr,
+            live_config_allowed: cli_spec.live_config_allowed,
         },
         cli_spec.spec,
         cli_spec.compute_ctl_config,
@@ -200,17 +200,37 @@ async fn init() -> Result<()> {
     Ok(())
 }
 
-fn get_config(cli: &Cli) -> Result<ComputeConfig> {
-    // First, read the config from the path if provided
-    if let Some(ref config) = cli.config {
-        let file = File::open(config)?;
-        return Ok(serde_json::from_reader(&file)?);
+fn try_spec_from_cli(cli: &Cli) -> Result<CliSpecParams> {
+    // First, try to get cluster spec from the cli argument
+    if let Some(ref spec_json) = cli.spec_json {
+        info!("got spec from cli argument {}", spec_json);
+        return Ok(CliSpecParams {
+            spec: Some(serde_json::from_str(spec_json)?),
+            compute_ctl_config: ComputeCtlConfig::default(),
+            live_config_allowed: false,
+        });
     }
 
-    // If the config wasn't provided in the CLI arguments, then retrieve it from
-    // the control plane
-    match get_config_from_control_plane(cli.control_plane_uri.as_ref().unwrap(), &cli.compute_id) {
-        Ok(config) => Ok(config),
+    // Second, try to read it from the file if path is provided
+    if let Some(ref spec_path) = cli.spec_path {
+        let file = File::open(Path::new(spec_path))?;
+        return Ok(CliSpecParams {
+            spec: Some(serde_json::from_reader(file)?),
+            compute_ctl_config: ComputeCtlConfig::default(),
+            live_config_allowed: true,
+        });
+    }
+
+    if cli.control_plane_uri.is_none() {
+        panic!("must specify --control-plane-uri");
+    };
+
+    match get_spec_from_control_plane(cli.control_plane_uri.as_ref().unwrap(), &cli.compute_id) {
+        Ok(resp) => Ok(CliSpecParams {
+            spec: resp.0,
+            compute_ctl_config: resp.1,
+            live_config_allowed: true,
+        }),
         Err(e) => {
             error!(
                 "cannot get response from control plane: {}\n\
@@ -222,6 +242,14 @@ fn get_config(cli: &Cli) -> Result<ComputeConfig> {
     }
 }
 
+struct CliSpecParams {
+    /// If a spec was provided via CLI or file, the [`ComputeSpec`]
+    spec: Option<ComputeSpec>,
+    #[allow(dead_code)]
+    compute_ctl_config: ComputeCtlConfig,
+    live_config_allowed: bool,
+}
+
 fn deinit_and_exit(exit_code: Option<i32>) -> ! {
     // Shutdown trace pipeline gracefully, so that it has a chance to send any
     // pending traces before we exit. Shutting down OTEL tracing provider may

compute_tools/src/catalog.rs

@@ -98,15 +98,13 @@ pub async fn get_database_schema(
         .kill_on_drop(true)
         .spawn()?;
 
-    let stdout = cmd
-        .stdout
-        .take()
-        .ok_or_else(|| std::io::Error::other("Failed to capture stdout."))?;
+    let stdout = cmd.stdout.take().ok_or_else(|| {
+        std::io::Error::new(std::io::ErrorKind::Other, "Failed to capture stdout.")
+    })?;
 
-    let stderr = cmd
-        .stderr
-        .take()
-        .ok_or_else(|| std::io::Error::other("Failed to capture stderr."))?;
+    let stderr = cmd.stderr.take().ok_or_else(|| {
+        std::io::Error::new(std::io::ErrorKind::Other, "Failed to capture stderr.")
+    })?;
 
     let mut stdout_reader = FramedRead::new(stdout, BytesCodec::new());
     let stderr_reader = BufReader::new(stderr);
@@ -130,7 +128,8 @@ pub async fn get_database_schema(
                 }
             });
 
-            return Err(SchemaDumpError::IO(std::io::Error::other(
+            return Err(SchemaDumpError::IO(std::io::Error::new(
+                std::io::ErrorKind::Other,
                 "failed to start pg_dump",
             )));
         }

compute_tools/src/compute.rs

@@ -93,6 +93,20 @@ pub struct ComputeNodeParams {
 
     /// the address of extension storage proxy gateway
     pub ext_remote_storage: Option<String>,
+
+    /// We should only allow live re- / configuration of the compute node if
+    /// it uses 'pull model', i.e. it can go to control-plane and fetch
+    /// the latest configuration. Otherwise, there could be a case:
+    /// - we start compute with some spec provided as argument
+    /// - we push new spec and it does reconfiguration
+    /// - but then something happens and compute pod / VM is destroyed,
+    ///   so k8s controller starts it again with the **old** spec
+    ///
+    /// and the same for empty computes:
+    /// - we started compute without any spec
+    /// - we push spec and it does configuration
+    /// - but then it is restarted without any spec again
+    pub live_config_allowed: bool,
 }
 
 /// Compute node info shared across several `compute_ctl` threads.
@@ -523,14 +537,11 @@ impl ComputeNode {
 
         let pspec = compute_state.pspec.as_ref().expect("spec must be set");
         info!(
-            "starting compute for project {}, operation {}, tenant {}, timeline {}, project {}, branch {}, endpoint {}, features {:?}, spec.remote_extensions {:?}",
+            "starting compute for project {}, operation {}, tenant {}, timeline {}, features {:?}, spec.remote_extensions {:?}",
             pspec.spec.cluster.cluster_id.as_deref().unwrap_or("None"),
             pspec.spec.operation_uuid.as_deref().unwrap_or("None"),
             pspec.tenant_id,
             pspec.timeline_id,
-            pspec.spec.project_id.as_deref().unwrap_or("None"),
-            pspec.spec.branch_id.as_deref().unwrap_or("None"),
-            pspec.spec.endpoint_id.as_deref().unwrap_or("None"),
             pspec.spec.features,
             pspec.spec.remote_extensions,
         );
@@ -634,28 +645,31 @@ impl ComputeNode {
             });
         }
 
-        // Configure and start rsyslog for compliance audit logging
-        match pspec.spec.audit_log_level {
-            ComputeAudit::Hipaa | ComputeAudit::Extended | ComputeAudit::Full => {
-                let remote_endpoint =
-                    std::env::var("AUDIT_LOGGING_ENDPOINT").unwrap_or("".to_string());
-                if remote_endpoint.is_empty() {
-                    anyhow::bail!("AUDIT_LOGGING_ENDPOINT is empty");
-                }
-
-                let log_directory_path = Path::new(&self.params.pgdata).join("log");
-                let log_directory_path = log_directory_path.to_string_lossy().to_string();
-                configure_audit_rsyslog(log_directory_path.clone(), "hipaa", &remote_endpoint)?;
-
-                // Launch a background task to clean up the audit logs
-                launch_pgaudit_gc(log_directory_path);
+        // Configure and start rsyslog for HIPAA if necessary
+        if let ComputeAudit::Hipaa = pspec.spec.audit_log_level {
+            let remote_endpoint = std::env::var("AUDIT_LOGGING_ENDPOINT").unwrap_or("".to_string());
+            if remote_endpoint.is_empty() {
+                anyhow::bail!("AUDIT_LOGGING_ENDPOINT is empty");
             }
-            _ => {}
+
+            let log_directory_path = Path::new(&self.params.pgdata).join("log");
+            let log_directory_path = log_directory_path.to_string_lossy().to_string();
+            configure_audit_rsyslog(log_directory_path.clone(), "hipaa", &remote_endpoint)?;
+
+            // Launch a background task to clean up the audit logs
+            launch_pgaudit_gc(log_directory_path);
         }
 
         // Configure and start rsyslog for Postgres logs export
-        let conf = PostgresLogsRsyslogConfig::new(pspec.spec.logs_export_host.as_deref());
-        configure_postgres_logs_export(conf)?;
+        if self.has_feature(ComputeFeature::PostgresLogsExport) {
+            if let Some(ref project_id) = pspec.spec.cluster.cluster_id {
+                let host = PostgresLogsRsyslogConfig::default_host(project_id);
+                let conf = PostgresLogsRsyslogConfig::new(Some(&host));
+                configure_postgres_logs_export(conf)?;
+            } else {
+                warn!("not configuring rsyslog for Postgres logs export: project ID is missing")
+            }
+        }
 
         // Launch remaining service threads
         let _monitor_handle = launch_monitor(self);
@@ -1559,10 +1573,6 @@ impl ComputeNode {
             });
         }
 
-        // Reconfigure rsyslog for Postgres logs export
-        let conf = PostgresLogsRsyslogConfig::new(spec.logs_export_host.as_deref());
-        configure_postgres_logs_export(conf)?;
-
         // Write new config
         let pgdata_path = Path::new(&self.params.pgdata);
         config::write_postgres_conf(

compute_tools/src/config.rs

@@ -7,7 +7,7 @@ use std::io::prelude::*;
 use std::path::Path;
 
 use compute_api::responses::TlsConfig;
-use compute_api::spec::{ComputeAudit, ComputeMode, ComputeSpec, GenericOption};
+use compute_api::spec::{ComputeAudit, ComputeFeature, ComputeMode, ComputeSpec, GenericOption};
 
 use crate::pg_helpers::{
     GenericOptionExt, GenericOptionsSearch, PgOptionsSerialize, escape_conf_value,
@@ -89,15 +89,6 @@ pub fn write_postgres_conf(
             escape_conf_value(&s.to_string())
         )?;
     }
-    if let Some(s) = &spec.project_id {
-        writeln!(file, "neon.project_id={}", escape_conf_value(s))?;
-    }
-    if let Some(s) = &spec.branch_id {
-        writeln!(file, "neon.branch_id={}", escape_conf_value(s))?;
-    }
-    if let Some(s) = &spec.endpoint_id {
-        writeln!(file, "neon.endpoint_id={}", escape_conf_value(s))?;
-    }
 
     // tls
     if let Some(tls_config) = tls_config {
@@ -178,7 +169,7 @@ pub fn write_postgres_conf(
     // and don't allow the user or the control plane admin to change them.
     match spec.audit_log_level {
         ComputeAudit::Disabled => {}
-        ComputeAudit::Log | ComputeAudit::Base => {
+        ComputeAudit::Log => {
             writeln!(file, "# Managed by compute_ctl base audit settings: start")?;
             writeln!(file, "pgaudit.log='ddl,role'")?;
             // Disable logging of catalog queries to reduce the noise
@@ -202,20 +193,16 @@ pub fn write_postgres_conf(
             }
             writeln!(file, "# Managed by compute_ctl base audit settings: end")?;
         }
-        ComputeAudit::Hipaa | ComputeAudit::Extended | ComputeAudit::Full => {
+        ComputeAudit::Hipaa => {
             writeln!(
                 file,
                 "# Managed by compute_ctl compliance audit settings: begin"
             )?;
-            // Enable logging of parameters.
-            // This is very verbose and may contain sensitive data.
-            if spec.audit_log_level == ComputeAudit::Full {
-                writeln!(file, "pgaudit.log_parameter=on")?;
-                writeln!(file, "pgaudit.log='all'")?;
-            } else {
-                writeln!(file, "pgaudit.log_parameter=off")?;
-                writeln!(file, "pgaudit.log='all, -misc'")?;
-            }
+            // This log level is very verbose
+            // but this is necessary for HIPAA compliance.
+            // Exclude 'misc' category, because it doesn't contain anythig relevant.
+            writeln!(file, "pgaudit.log='all, -misc'")?;
+            writeln!(file, "pgaudit.log_parameter=on")?;
             // Disable logging of catalog queries
             // The catalog doesn't contain sensitive data, so we don't need to audit it.
             writeln!(file, "pgaudit.log_catalog=off")?;
@@ -268,7 +255,7 @@ pub fn write_postgres_conf(
 
     // We need Postgres to send logs to rsyslog so that we can forward them
     // further to customers' log aggregation systems.
-    if spec.logs_export_host.is_some() {
+    if spec.features.contains(&ComputeFeature::PostgresLogsExport) {
         writeln!(file, "log_destination='stderr,syslog'")?;
     }
 

compute_tools/src/http/middleware/authorize.rs

@@ -6,15 +6,20 @@ use axum_extra::{
     TypedHeader,
     headers::{Authorization, authorization::Bearer},
 };
-use compute_api::requests::ComputeClaims;
 use futures::future::BoxFuture;
 use http::{Request, Response, StatusCode};
 use jsonwebtoken::{Algorithm, DecodingKey, TokenData, Validation, jwk::JwkSet};
+use serde::Deserialize;
 use tower_http::auth::AsyncAuthorizeRequest;
 use tracing::warn;
 
 use crate::http::{JsonResponse, extract::RequestId};
 
+#[derive(Clone, Debug, Deserialize)]
+pub(in crate::http) struct Claims {
+    compute_id: String,
+}
+
 #[derive(Clone, Debug)]
 pub(in crate::http) struct Authorize {
     compute_id: String,
@@ -107,11 +112,7 @@ impl AsyncAuthorizeRequest<Body> for Authorize {
 
 impl Authorize {
     /// Verify the token using the JSON Web Key set and return the token data.
-    fn verify(
-        jwks: &JwkSet,
-        token: &str,
-        validation: &Validation,
-    ) -> Result<TokenData<ComputeClaims>> {
+    fn verify(jwks: &JwkSet, token: &str, validation: &Validation) -> Result<TokenData<Claims>> {
         for jwk in jwks.keys.iter() {
             let decoding_key = match DecodingKey::from_jwk(jwk) {
                 Ok(key) => key,
@@ -126,7 +127,7 @@ impl Authorize {
                 }
             };
 
-            match jsonwebtoken::decode::<ComputeClaims>(token, &decoding_key, validation) {
+            match jsonwebtoken::decode::<Claims>(token, &decoding_key, validation) {
                 Ok(data) => return Ok(data),
                 Err(e) => {
                     warn!(

compute_tools/src/http/openapi_spec.yaml

@@ -306,6 +306,36 @@ paths:
               schema:
                 $ref: "#/components/schemas/GenericError"
 
+  /configure_telemetry:
+    post:
+      tags:
+        - Configure
+      summary: Configure rsyslog
+      description: |
+        This API endpoint configures rsyslog to forward Postgres logs
+        to a specified otel collector.
+      operationId: configureTelemetry
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                logs_export_host:
+                  type: string
+                  description: |
+                    Hostname and the port of the otel collector. Leave empty to disable logs forwarding.
+                    Example: config-shy-breeze-123-collector-monitoring.neon-telemetry.svc.cluster.local:54526
+      responses:
+        204:
+          description: "Telemetry configured successfully"
+        500:
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/GenericError"
+
 components:
   securitySchemes:
     JWT:

compute_tools/src/http/routes/configure.rs

@@ -1,9 +1,11 @@
 use std::sync::Arc;
 
+use axum::body::Body;
 use axum::extract::State;
 use axum::response::Response;
-use compute_api::requests::ConfigurationRequest;
+use compute_api::requests::{ConfigurationRequest, ConfigureTelemetryRequest};
 use compute_api::responses::{ComputeStatus, ComputeStatusResponse};
+use compute_api::spec::ComputeFeature;
 use http::StatusCode;
 use tokio::task;
 use tracing::info;
@@ -11,6 +13,7 @@ use tracing::info;
 use crate::compute::{ComputeNode, ParsedSpec};
 use crate::http::JsonResponse;
 use crate::http::extract::Json;
+use crate::rsyslog::{PostgresLogsRsyslogConfig, configure_postgres_logs_export};
 
 // Accept spec in JSON format and request compute configuration. If anything
 // goes wrong after we set the compute status to `ConfigurationPending` and
@@ -22,6 +25,13 @@ pub(in crate::http) async fn configure(
     State(compute): State<Arc<ComputeNode>>,
     request: Json<ConfigurationRequest>,
 ) -> Response {
+    if !compute.params.live_config_allowed {
+        return JsonResponse::error(
+            StatusCode::PRECONDITION_FAILED,
+            "live configuration is not allowed for this compute node".to_string(),
+        );
+    }
+
     let pspec = match ParsedSpec::try_from(request.spec.clone()) {
         Ok(p) => p,
         Err(e) => return JsonResponse::error(StatusCode::BAD_REQUEST, e),
@@ -85,3 +95,25 @@ pub(in crate::http) async fn configure(
 
     JsonResponse::success(StatusCode::OK, body)
 }
+
+pub(in crate::http) async fn configure_telemetry(
+    State(compute): State<Arc<ComputeNode>>,
+    request: Json<ConfigureTelemetryRequest>,
+) -> Response {
+    if !compute.has_feature(ComputeFeature::PostgresLogsExport) {
+        return JsonResponse::error(
+            StatusCode::PRECONDITION_FAILED,
+            "Postgres logs export feature is not enabled".to_string(),
+        );
+    }
+
+    let conf = PostgresLogsRsyslogConfig::new(request.logs_export_host.as_deref());
+    if let Err(err) = configure_postgres_logs_export(conf) {
+        return JsonResponse::error(StatusCode::INTERNAL_SERVER_ERROR, err.to_string());
+    }
+
+    Response::builder()
+        .status(StatusCode::NO_CONTENT)
+        .body(Body::from(""))
+        .unwrap()
+}

compute_tools/src/http/server.rs

@@ -87,6 +87,7 @@ impl From<&Server> for Router<Arc<ComputeNode>> {
                 let authenticated_router = Router::<Arc<ComputeNode>>::new()
                     .route("/check_writability", post(check_writability::is_writable))
                     .route("/configure", post(configure::configure))
+                    .route("/configure_telemetry", post(configure::configure_telemetry))
                     .route("/database_schema", get(database_schema::get_schema_dump))
                     .route("/dbs_and_roles", get(dbs_and_roles::get_catalog_objects))
                     .route("/insights", get(insights::get_insights))

compute_tools/src/metrics.rs

@@ -19,13 +19,13 @@ pub(crate) static INSTALLED_EXTENSIONS: Lazy<UIntGaugeVec> = Lazy::new(|| {
 // but for all our APIs we defined a 'slug'/method/operationId in the OpenAPI spec.
 // And it's fair to call it a 'RPC' (Remote Procedure Call).
 pub enum CPlaneRequestRPC {
-    GetConfig,
+    GetSpec,
 }
 
 impl CPlaneRequestRPC {
     pub fn as_str(&self) -> &str {
         match self {
-            CPlaneRequestRPC::GetConfig => "GetConfig",
+            CPlaneRequestRPC::GetSpec => "GetSpec",
         }
     }
 }

compute_tools/src/rsyslog.rs

@@ -119,9 +119,16 @@ impl<'a> PostgresLogsRsyslogConfig<'a> {
         };
         Ok(config_content)
     }
+
+    /// Returns the default host for otel collector that receives Postgres logs
+    pub fn default_host(project_id: &str) -> String {
+        format!(
+            "config-{}-collector.neon-telemetry.svc.cluster.local:10514",
+            project_id
+        )
+    }
 }
 
-/// Writes rsyslogd configuration for Postgres logs export and restarts rsyslog.
 pub fn configure_postgres_logs_export(conf: PostgresLogsRsyslogConfig) -> Result<()> {
     let new_config = conf.build()?;
     let current_config = PostgresLogsRsyslogConfig::current_config()?;
@@ -254,5 +261,16 @@ mod tests {
             let res = conf.build();
             assert!(res.is_err());
         }
+
+        {
+            // Verify config with default host
+            let host = PostgresLogsRsyslogConfig::default_host("shy-breeze-123");
+            let conf = PostgresLogsRsyslogConfig::new(Some(&host));
+            let res = conf.build();
+            assert!(res.is_ok());
+            let conf_str = res.unwrap();
+            assert!(conf_str.contains(r#"shy-breeze-123"#));
+            assert!(conf_str.contains(r#"port="10514""#));
+        }
     }
 }

compute_tools/src/spec.rs

@@ -3,8 +3,9 @@ use std::path::Path;
 
 use anyhow::{Result, anyhow, bail};
 use compute_api::responses::{
-    ComputeConfig, ControlPlaneComputeStatus, ControlPlaneConfigResponse,
+    ComputeCtlConfig, ControlPlaneComputeStatus, ControlPlaneSpecResponse,
 };
+use compute_api::spec::ComputeSpec;
 use reqwest::StatusCode;
 use tokio_postgres::Client;
 use tracing::{error, info, instrument};
@@ -20,7 +21,7 @@ use crate::params::PG_HBA_ALL_MD5;
 fn do_control_plane_request(
     uri: &str,
     jwt: &str,
-) -> Result<ControlPlaneConfigResponse, (bool, String, String)> {
+) -> Result<ControlPlaneSpecResponse, (bool, String, String)> {
     let resp = reqwest::blocking::Client::new()
         .get(uri)
         .header("Authorization", format!("Bearer {}", jwt))
@@ -28,14 +29,14 @@ fn do_control_plane_request(
         .map_err(|e| {
             (
                 true,
-                format!("could not perform request to control plane: {:?}", e),
+                format!("could not perform spec request to control plane: {:?}", e),
                 UNKNOWN_HTTP_STATUS.to_string(),
             )
         })?;
 
     let status = resp.status();
     match status {
-        StatusCode::OK => match resp.json::<ControlPlaneConfigResponse>() {
+        StatusCode::OK => match resp.json::<ControlPlaneSpecResponse>() {
             Ok(spec_resp) => Ok(spec_resp),
             Err(e) => Err((
                 true,
@@ -68,35 +69,40 @@ fn do_control_plane_request(
     }
 }
 
-/// Request config from the control-plane by compute_id. If
-/// `NEON_CONTROL_PLANE_TOKEN` env variable is set, it will be used for
-/// authorization.
-pub fn get_config_from_control_plane(base_uri: &str, compute_id: &str) -> Result<ComputeConfig> {
+/// Request spec from the control-plane by compute_id. If `NEON_CONTROL_PLANE_TOKEN`
+/// env variable is set, it will be used for authorization.
+pub fn get_spec_from_control_plane(
+    base_uri: &str,
+    compute_id: &str,
+) -> Result<(Option<ComputeSpec>, ComputeCtlConfig)> {
     let cp_uri = format!("{base_uri}/compute/api/v2/computes/{compute_id}/spec");
-    let jwt: String = std::env::var("NEON_CONTROL_PLANE_TOKEN").unwrap_or_default();
+    let jwt: String = match std::env::var("NEON_CONTROL_PLANE_TOKEN") {
+        Ok(v) => v,
+        Err(_) => "".to_string(),
+    };
     let mut attempt = 1;
 
-    info!("getting config from control plane: {}", cp_uri);
+    info!("getting spec from control plane: {}", cp_uri);
 
     // Do 3 attempts to get spec from the control plane using the following logic:
     // - network error -> then retry
     // - compute id is unknown or any other error -> bail out
     // - no spec for compute yet (Empty state) -> return Ok(None)
-    // - got config -> return Ok(Some(config))
+    // - got spec -> return Ok(Some(spec))
     while attempt < 4 {
         let result = match do_control_plane_request(&cp_uri, &jwt) {
-            Ok(config_resp) => {
+            Ok(spec_resp) => {
                 CPLANE_REQUESTS_TOTAL
                     .with_label_values(&[
-                        CPlaneRequestRPC::GetConfig.as_str(),
+                        CPlaneRequestRPC::GetSpec.as_str(),
                         &StatusCode::OK.to_string(),
                     ])
                     .inc();
-                match config_resp.status {
-                    ControlPlaneComputeStatus::Empty => Ok(config_resp.into()),
+                match spec_resp.status {
+                    ControlPlaneComputeStatus::Empty => Ok((None, spec_resp.compute_ctl_config)),
                     ControlPlaneComputeStatus::Attached => {
-                        if config_resp.spec.is_some() {
-                            Ok(config_resp.into())
+                        if let Some(spec) = spec_resp.spec {
+                            Ok((Some(spec), spec_resp.compute_ctl_config))
                         } else {
                             bail!("compute is attached, but spec is empty")
                         }
@@ -105,7 +111,7 @@ pub fn get_config_from_control_plane(base_uri: &str, compute_id: &str) -> Result
             }
             Err((retry, msg, status)) => {
                 CPLANE_REQUESTS_TOTAL
-                    .with_label_values(&[CPlaneRequestRPC::GetConfig.as_str(), &status])
+                    .with_label_values(&[CPlaneRequestRPC::GetSpec.as_str(), &status])
                     .inc();
                 if retry {
                     Err(anyhow!(msg))
@@ -116,7 +122,7 @@ pub fn get_config_from_control_plane(base_uri: &str, compute_id: &str) -> Result
         };
 
         if let Err(e) = &result {
-            error!("attempt {} to get config failed with: {}", attempt, e);
+            error!("attempt {} to get spec failed with: {}", attempt, e);
         } else {
             return result;
         }
@@ -127,13 +133,13 @@ pub fn get_config_from_control_plane(base_uri: &str, compute_id: &str) -> Result
 
     // All attempts failed, return error.
     Err(anyhow::anyhow!(
-        "Exhausted all attempts to retrieve the config from the control plane"
+        "Exhausted all attempts to retrieve the spec from the control plane"
     ))
 }
 
 /// Check `pg_hba.conf` and update if needed to allow external connections.
 pub fn update_pg_hba(pgdata_path: &Path) -> Result<()> {
-    // XXX: consider making it a part of config.json
+    // XXX: consider making it a part of spec.json
     let pghba_path = pgdata_path.join("pg_hba.conf");
 
     if config::line_in_file(&pghba_path, PG_HBA_ALL_MD5)? {
@@ -147,7 +153,7 @@ pub fn update_pg_hba(pgdata_path: &Path) -> Result<()> {
 
 /// Create a standby.signal file
 pub fn add_standby_signal(pgdata_path: &Path) -> Result<()> {
-    // XXX: consider making it a part of config.json
+    // XXX: consider making it a part of spec.json
     let signalfile = pgdata_path.join("standby.signal");
 
     if !signalfile.exists() {

compute_tools/src/spec_apply.rs

@@ -278,12 +278,12 @@ impl ComputeNode {
             // so that all config operations are audit logged.
             match spec.audit_log_level
             {
-                ComputeAudit::Hipaa | ComputeAudit::Extended | ComputeAudit::Full => {
+                ComputeAudit::Hipaa => {
                     phases.push(CreatePgauditExtension);
                     phases.push(CreatePgauditlogtofileExtension);
                     phases.push(DisablePostgresDBPgAudit);
                 }
-                ComputeAudit::Log | ComputeAudit::Base => {
+                ComputeAudit::Log => {
                     phases.push(CreatePgauditExtension);
                     phases.push(DisablePostgresDBPgAudit);
                 }

control_plane/src/bin/neon_local.rs

@@ -20,10 +20,8 @@ use compute_api::spec::ComputeMode;
 use control_plane::endpoint::ComputeControlPlane;
 use control_plane::local_env::{
     InitForceMode, LocalEnv, NeonBroker, NeonLocalInitConf, NeonLocalInitPageserverConf,
-    ObjectStorageConf, SafekeeperConf,
+    SafekeeperConf,
 };
-use control_plane::object_storage::OBJECT_STORAGE_DEFAULT_PORT;
-use control_plane::object_storage::ObjectStorage;
 use control_plane::pageserver::PageServerNode;
 use control_plane::safekeeper::SafekeeperNode;
 use control_plane::storage_controller::{
@@ -41,7 +39,7 @@ use pageserver_api::controller_api::{
 use pageserver_api::models::{
     ShardParameters, TenantConfigRequest, TimelineCreateRequest, TimelineInfo,
 };
-use pageserver_api::shard::{DEFAULT_STRIPE_SIZE, ShardCount, ShardStripeSize, TenantShardId};
+use pageserver_api::shard::{ShardCount, ShardStripeSize, TenantShardId};
 use postgres_backend::AuthType;
 use postgres_connection::parse_host_port;
 use safekeeper_api::membership::SafekeeperGeneration;
@@ -93,8 +91,6 @@ enum NeonLocalCmd {
     #[command(subcommand)]
     Safekeeper(SafekeeperCmd),
     #[command(subcommand)]
-    ObjectStorage(ObjectStorageCmd),
-    #[command(subcommand)]
     Endpoint(EndpointCmd),
     #[command(subcommand)]
     Mappings(MappingsCmd),
@@ -458,32 +454,6 @@ enum SafekeeperCmd {
     Restart(SafekeeperRestartCmdArgs),
 }
 
-#[derive(clap::Subcommand)]
-#[clap(about = "Manage object storage")]
-enum ObjectStorageCmd {
-    Start(ObjectStorageStartCmd),
-    Stop(ObjectStorageStopCmd),
-}
-
-#[derive(clap::Args)]
-#[clap(about = "Start object storage")]
-struct ObjectStorageStartCmd {
-    #[clap(short = 't', long, help = "timeout until we fail the command")]
-    #[arg(default_value = "10s")]
-    start_timeout: humantime::Duration,
-}
-
-#[derive(clap::Args)]
-#[clap(about = "Stop object storage")]
-struct ObjectStorageStopCmd {
-    #[arg(value_enum, default_value = "fast")]
-    #[clap(
-        short = 'm',
-        help = "If 'immediate', don't flush repository data at shutdown"
-    )]
-    stop_mode: StopMode,
-}
-
 #[derive(clap::Args)]
 #[clap(about = "Start local safekeeper")]
 struct SafekeeperStartCmdArgs {
@@ -789,7 +759,6 @@ fn main() -> Result<()> {
             }
             NeonLocalCmd::StorageBroker(subcmd) => rt.block_on(handle_storage_broker(&subcmd, env)),
             NeonLocalCmd::Safekeeper(subcmd) => rt.block_on(handle_safekeeper(&subcmd, env)),
-            NeonLocalCmd::ObjectStorage(subcmd) => rt.block_on(handle_object_storage(&subcmd, env)),
             NeonLocalCmd::Endpoint(subcmd) => rt.block_on(handle_endpoint(&subcmd, env)),
             NeonLocalCmd::Mappings(subcmd) => handle_mappings(&subcmd, env),
         };
@@ -1006,9 +975,6 @@ fn handle_init(args: &InitCmdArgs) -> anyhow::Result<LocalEnv> {
                     }
                 })
                 .collect(),
-            object_storage: ObjectStorageConf {
-                port: OBJECT_STORAGE_DEFAULT_PORT,
-            },
             pg_distrib_dir: None,
             neon_distrib_dir: None,
             default_tenant_id: TenantId::from_array(std::array::from_fn(|_| 0)),
@@ -1117,7 +1083,7 @@ async fn handle_tenant(subcmd: &TenantCmd, env: &mut local_env::LocalEnv) -> any
                         stripe_size: args
                             .shard_stripe_size
                             .map(ShardStripeSize)
-                            .unwrap_or(DEFAULT_STRIPE_SIZE),
+                            .unwrap_or(ShardParameters::DEFAULT_STRIPE_SIZE),
                     },
                     placement_policy: args.placement_policy.clone(),
                     config: tenant_conf,
@@ -1430,7 +1396,7 @@ async fn handle_endpoint(subcmd: &EndpointCmd, env: &local_env::LocalEnv) -> Res
                     vec![(parsed.0, parsed.1.unwrap_or(5432))],
                     // If caller is telling us what pageserver to use, this is not a tenant which is
                     // full managed by storage controller, therefore not sharded.
-                    DEFAULT_STRIPE_SIZE,
+                    ShardParameters::DEFAULT_STRIPE_SIZE,
                 )
             } else {
                 // Look up the currently attached location of the tenant, and its striping metadata,
@@ -1717,41 +1683,6 @@ async fn handle_safekeeper(subcmd: &SafekeeperCmd, env: &local_env::LocalEnv) ->
     Ok(())
 }
 
-async fn handle_object_storage(subcmd: &ObjectStorageCmd, env: &local_env::LocalEnv) -> Result<()> {
-    use ObjectStorageCmd::*;
-    let storage = ObjectStorage::from_env(env);
-
-    // In tests like test_forward_compatibility or test_graceful_cluster_restart
-    // old neon binaries (without object_storage) are present
-    if !storage.bin.exists() {
-        eprintln!(
-            "{} binary not found. Ignore if this is a compatibility test",
-            storage.bin
-        );
-        return Ok(());
-    }
-
-    match subcmd {
-        Start(ObjectStorageStartCmd { start_timeout }) => {
-            if let Err(e) = storage.start(start_timeout).await {
-                eprintln!("object_storage start failed: {e}");
-                exit(1);
-            }
-        }
-        Stop(ObjectStorageStopCmd { stop_mode }) => {
-            let immediate = match stop_mode {
-                StopMode::Fast => false,
-                StopMode::Immediate => true,
-            };
-            if let Err(e) = storage.stop(immediate) {
-                eprintln!("proxy stop failed: {e}");
-                exit(1);
-            }
-        }
-    };
-    Ok(())
-}
-
 async fn handle_storage_broker(subcmd: &StorageBrokerCmd, env: &local_env::LocalEnv) -> Result<()> {
     match subcmd {
         StorageBrokerCmd::Start(args) => {
@@ -1846,13 +1777,6 @@ async fn handle_start_all_impl(
                     .map_err(|e| e.context(format!("start safekeeper {}", safekeeper.id)))
             });
         }
-
-        js.spawn(async move {
-            ObjectStorage::from_env(env)
-                .start(&retry_timeout)
-                .await
-                .map_err(|e| e.context("start object_storage"))
-        });
     })();
 
     let mut errors = Vec::new();
@@ -1950,11 +1874,6 @@ async fn try_stop_all(env: &local_env::LocalEnv, immediate: bool) {
         }
     }
 
-    let storage = ObjectStorage::from_env(env);
-    if let Err(e) = storage.stop(immediate) {
-        eprintln!("object_storage stop failed: {:#}", e);
-    }
-
     for ps_conf in &env.pageservers {
         let pageserver = PageServerNode::from_env(env, ps_conf);
         if let Err(e) = pageserver.stop(immediate) {

control_plane/src/endpoint.rs

@@ -29,7 +29,7 @@
 //!     compute.log               - log output of `compute_ctl` and `postgres`
 //!     endpoint.json             - serialized `EndpointConf` struct
 //!     postgresql.conf           - postgresql settings
-//!     config.json                 - passed to `compute_ctl`
+//!     spec.json                 - passed to `compute_ctl`
 //!     pgdata/
 //!         postgresql.conf       - copy of postgresql.conf created by `compute_ctl`
 //!         zenith.signal
@@ -46,9 +46,7 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
 use anyhow::{Context, Result, anyhow, bail};
 use compute_api::requests::ConfigurationRequest;
-use compute_api::responses::{
-    ComputeConfig, ComputeCtlConfig, ComputeStatus, ComputeStatusResponse,
-};
+use compute_api::responses::{ComputeCtlConfig, ComputeStatus, ComputeStatusResponse};
 use compute_api::spec::{
     Cluster, ComputeAudit, ComputeFeature, ComputeMode, ComputeSpec, Database, PgIdent,
     RemoteExtSpec, Role,
@@ -621,97 +619,86 @@ impl Endpoint {
             remote_extensions = None;
         };
 
-        // Create config file
-        let config = {
-            let mut spec = ComputeSpec {
-                skip_pg_catalog_updates: self.skip_pg_catalog_updates,
-                format_version: 1.0,
-                operation_uuid: None,
-                features: self.features.clone(),
-                swap_size_bytes: None,
-                disk_quota_bytes: None,
-                disable_lfc_resizing: None,
-                cluster: Cluster {
-                    cluster_id: None, // project ID: not used
-                    name: None,       // project name: not used
-                    state: None,
-                    roles: if create_test_user {
-                        vec![Role {
-                            name: PgIdent::from_str("test").unwrap(),
-                            encrypted_password: None,
-                            options: None,
-                        }]
-                    } else {
-                        Vec::new()
-                    },
-                    databases: if create_test_user {
-                        vec![Database {
-                            name: PgIdent::from_str("neondb").unwrap(),
-                            owner: PgIdent::from_str("test").unwrap(),
-                            options: None,
-                            restrict_conn: false,
-                            invalid: false,
-                        }]
-                    } else {
-                        Vec::new()
-                    },
-                    settings: None,
-                    postgresql_conf: Some(postgresql_conf.clone()),
-                },
-                delta_operations: None,
-                tenant_id: Some(self.tenant_id),
-                timeline_id: Some(self.timeline_id),
-                project_id: None,
-                branch_id: None,
-                endpoint_id: Some(self.endpoint_id.clone()),
-                mode: self.mode,
-                pageserver_connstring: Some(pageserver_connstring),
-                safekeepers_generation: safekeepers_generation.map(|g| g.into_inner()),
-                safekeeper_connstrings,
-                storage_auth_token: auth_token.clone(),
-                remote_extensions,
-                pgbouncer_settings: None,
-                shard_stripe_size: Some(shard_stripe_size),
-                local_proxy_config: None,
-                reconfigure_concurrency: self.reconfigure_concurrency,
-                drop_subscriptions_before_start: self.drop_subscriptions_before_start,
-                audit_log_level: ComputeAudit::Disabled,
-                logs_export_host: None::<String>,
-            };
-
-            // this strange code is needed to support respec() in tests
-            if self.cluster.is_some() {
-                debug!("Cluster is already set in the endpoint spec, using it");
-                spec.cluster = self.cluster.clone().unwrap();
-
-                debug!("spec.cluster {:?}", spec.cluster);
-
-                // fill missing fields again
-                if create_test_user {
-                    spec.cluster.roles.push(Role {
+        // Create spec file
+        let mut spec = ComputeSpec {
+            skip_pg_catalog_updates: self.skip_pg_catalog_updates,
+            format_version: 1.0,
+            operation_uuid: None,
+            features: self.features.clone(),
+            swap_size_bytes: None,
+            disk_quota_bytes: None,
+            disable_lfc_resizing: None,
+            cluster: Cluster {
+                cluster_id: None, // project ID: not used
+                name: None,       // project name: not used
+                state: None,
+                roles: if create_test_user {
+                    vec![Role {
                         name: PgIdent::from_str("test").unwrap(),
                         encrypted_password: None,
                         options: None,
-                    });
-                    spec.cluster.databases.push(Database {
+                    }]
+                } else {
+                    Vec::new()
+                },
+                databases: if create_test_user {
+                    vec![Database {
                         name: PgIdent::from_str("neondb").unwrap(),
                         owner: PgIdent::from_str("test").unwrap(),
                         options: None,
                         restrict_conn: false,
                         invalid: false,
-                    });
-                }
-                spec.cluster.postgresql_conf = Some(postgresql_conf);
-            }
-
-            ComputeConfig {
-                spec: Some(spec),
-                compute_ctl_config: ComputeCtlConfig::default(),
-            }
+                    }]
+                } else {
+                    Vec::new()
+                },
+                settings: None,
+                postgresql_conf: Some(postgresql_conf.clone()),
+            },
+            delta_operations: None,
+            tenant_id: Some(self.tenant_id),
+            timeline_id: Some(self.timeline_id),
+            mode: self.mode,
+            pageserver_connstring: Some(pageserver_connstring),
+            safekeepers_generation: safekeepers_generation.map(|g| g.into_inner()),
+            safekeeper_connstrings,
+            storage_auth_token: auth_token.clone(),
+            remote_extensions,
+            pgbouncer_settings: None,
+            shard_stripe_size: Some(shard_stripe_size),
+            local_proxy_config: None,
+            reconfigure_concurrency: self.reconfigure_concurrency,
+            drop_subscriptions_before_start: self.drop_subscriptions_before_start,
+            audit_log_level: ComputeAudit::Disabled,
         };
 
-        let config_path = self.endpoint_path().join("config.json");
-        std::fs::write(config_path, serde_json::to_string_pretty(&config)?)?;
+        // this strange code is needed to support respec() in tests
+        if self.cluster.is_some() {
+            debug!("Cluster is already set in the endpoint spec, using it");
+            spec.cluster = self.cluster.clone().unwrap();
+
+            debug!("spec.cluster {:?}", spec.cluster);
+
+            // fill missing fields again
+            if create_test_user {
+                spec.cluster.roles.push(Role {
+                    name: PgIdent::from_str("test").unwrap(),
+                    encrypted_password: None,
+                    options: None,
+                });
+                spec.cluster.databases.push(Database {
+                    name: PgIdent::from_str("neondb").unwrap(),
+                    owner: PgIdent::from_str("test").unwrap(),
+                    options: None,
+                    restrict_conn: false,
+                    invalid: false,
+                });
+            }
+            spec.cluster.postgresql_conf = Some(postgresql_conf);
+        }
+
+        let spec_path = self.endpoint_path().join("spec.json");
+        std::fs::write(spec_path, serde_json::to_string_pretty(&spec)?)?;
 
         // Open log file. We'll redirect the stdout and stderr of `compute_ctl` to it.
         let logfile = std::fs::OpenOptions::new()
@@ -737,8 +724,10 @@ impl Endpoint {
         ])
         .args(["--pgdata", self.pgdata().to_str().unwrap()])
         .args(["--connstr", &conn_str])
-        .arg("--config")
-        .arg(self.endpoint_path().join("config.json").as_os_str())
+        .args([
+            "--spec-path",
+            self.endpoint_path().join("spec.json").to_str().unwrap(),
+        ])
         .args([
             "--pgbin",
             self.env
@@ -880,12 +869,10 @@ impl Endpoint {
         stripe_size: Option<ShardStripeSize>,
         safekeepers: Option<Vec<NodeId>>,
     ) -> Result<()> {
-        let (mut spec, compute_ctl_config) = {
-            let config_path = self.endpoint_path().join("config.json");
-            let file = std::fs::File::open(config_path)?;
-            let config: ComputeConfig = serde_json::from_reader(file)?;
-
-            (config.spec.unwrap(), config.compute_ctl_config)
+        let mut spec: ComputeSpec = {
+            let spec_path = self.endpoint_path().join("spec.json");
+            let file = std::fs::File::open(spec_path)?;
+            serde_json::from_reader(file)?
         };
 
         let postgresql_conf = self.read_postgresql_conf()?;
@@ -935,7 +922,7 @@ impl Endpoint {
             .body(
                 serde_json::to_string(&ConfigurationRequest {
                     spec,
-                    compute_ctl_config,
+                    compute_ctl_config: ComputeCtlConfig::default(),
                 })
                 .unwrap(),
             )

control_plane/src/lib.rs

@@ -10,7 +10,6 @@ mod background_process;
 pub mod broker;
 pub mod endpoint;
 pub mod local_env;
-pub mod object_storage;
 pub mod pageserver;
 pub mod postgresql_conf;
 pub mod safekeeper;

control_plane/src/local_env.rs

@@ -15,10 +15,9 @@ use clap::ValueEnum;
 use postgres_backend::AuthType;
 use reqwest::Url;
 use serde::{Deserialize, Serialize};
-use utils::auth::encode_from_key_file;
+use utils::auth::{Claims, encode_from_key_file};
 use utils::id::{NodeId, TenantId, TenantTimelineId, TimelineId};
 
-use crate::object_storage::{OBJECT_STORAGE_REMOTE_STORAGE_DIR, ObjectStorage};
 use crate::pageserver::{PAGESERVER_REMOTE_STORAGE_DIR, PageServerNode};
 use crate::safekeeper::SafekeeperNode;
 
@@ -56,7 +55,6 @@ pub struct LocalEnv {
 
     // used to issue tokens during e.g pg start
     pub private_key_path: PathBuf,
-    pub public_key_path: PathBuf,
 
     pub broker: NeonBroker,
 
@@ -70,8 +68,6 @@ pub struct LocalEnv {
 
     pub safekeepers: Vec<SafekeeperConf>,
 
-    pub object_storage: ObjectStorageConf,
-
     // Control plane upcall API for pageserver: if None, we will not run storage_controller  If set, this will
     // be propagated into each pageserver's configuration.
     pub control_plane_api: Url,
@@ -99,7 +95,6 @@ pub struct OnDiskConfig {
     pub neon_distrib_dir: PathBuf,
     pub default_tenant_id: Option<TenantId>,
     pub private_key_path: PathBuf,
-    pub public_key_path: PathBuf,
     pub broker: NeonBroker,
     pub storage_controller: NeonStorageControllerConf,
     #[serde(
@@ -108,7 +103,6 @@ pub struct OnDiskConfig {
     )]
     pub pageservers: Vec<PageServerConf>,
     pub safekeepers: Vec<SafekeeperConf>,
-    pub object_storage: ObjectStorageConf,
     pub control_plane_api: Option<Url>,
     pub control_plane_hooks_api: Option<Url>,
     pub control_plane_compute_hook_api: Option<Url>,
@@ -142,18 +136,11 @@ pub struct NeonLocalInitConf {
     pub storage_controller: Option<NeonStorageControllerConf>,
     pub pageservers: Vec<NeonLocalInitPageserverConf>,
     pub safekeepers: Vec<SafekeeperConf>,
-    pub object_storage: ObjectStorageConf,
     pub control_plane_api: Option<Url>,
     pub control_plane_hooks_api: Option<Url>,
     pub generate_local_ssl_certs: bool,
 }
 
-#[derive(Serialize, Default, Deserialize, PartialEq, Eq, Clone, Debug)]
-#[serde(default)]
-pub struct ObjectStorageConf {
-    pub port: u16,
-}
-
 /// Broker config for cluster internal communication.
 #[derive(Serialize, Deserialize, PartialEq, Eq, Clone, Debug)]
 #[serde(default)]
@@ -411,10 +398,6 @@ impl LocalEnv {
         self.pg_dir(pg_version, "lib")
     }
 
-    pub fn object_storage_bin(&self) -> PathBuf {
-        self.neon_distrib_dir.join("object_storage")
-    }
-
     pub fn pageserver_bin(&self) -> PathBuf {
         self.neon_distrib_dir.join("pageserver")
     }
@@ -448,10 +431,6 @@ impl LocalEnv {
         self.base_data_dir.join("safekeepers").join(data_dir_name)
     }
 
-    pub fn object_storage_data_dir(&self) -> PathBuf {
-        self.base_data_dir.join("object_storage")
-    }
-
     pub fn get_pageserver_conf(&self, id: NodeId) -> anyhow::Result<&PageServerConf> {
         if let Some(conf) = self.pageservers.iter().find(|node| node.id == id) {
             Ok(conf)
@@ -603,7 +582,6 @@ impl LocalEnv {
                 neon_distrib_dir,
                 default_tenant_id,
                 private_key_path,
-                public_key_path,
                 broker,
                 storage_controller,
                 pageservers,
@@ -613,7 +591,6 @@ impl LocalEnv {
                 control_plane_compute_hook_api: _,
                 branch_name_mappings,
                 generate_local_ssl_certs,
-                object_storage,
             } = on_disk_config;
             LocalEnv {
                 base_data_dir: repopath.to_owned(),
@@ -621,7 +598,6 @@ impl LocalEnv {
                 neon_distrib_dir,
                 default_tenant_id,
                 private_key_path,
-                public_key_path,
                 broker,
                 storage_controller,
                 pageservers,
@@ -630,7 +606,6 @@ impl LocalEnv {
                 control_plane_hooks_api,
                 branch_name_mappings,
                 generate_local_ssl_certs,
-                object_storage,
             }
         };
 
@@ -730,7 +705,6 @@ impl LocalEnv {
                 neon_distrib_dir: self.neon_distrib_dir.clone(),
                 default_tenant_id: self.default_tenant_id,
                 private_key_path: self.private_key_path.clone(),
-                public_key_path: self.public_key_path.clone(),
                 broker: self.broker.clone(),
                 storage_controller: self.storage_controller.clone(),
                 pageservers: vec![], // it's skip_serializing anyway
@@ -740,7 +714,6 @@ impl LocalEnv {
                 control_plane_compute_hook_api: None,
                 branch_name_mappings: self.branch_name_mappings.clone(),
                 generate_local_ssl_certs: self.generate_local_ssl_certs,
-                object_storage: self.object_storage.clone(),
             },
         )
     }
@@ -757,7 +730,7 @@ impl LocalEnv {
     }
 
     // this function is used only for testing purposes in CLI e g generate tokens during init
-    pub fn generate_auth_token<S: Serialize>(&self, claims: &S) -> anyhow::Result<String> {
+    pub fn generate_auth_token(&self, claims: &Claims) -> anyhow::Result<String> {
         let private_key_path = self.get_private_key_path();
         let key_data = fs::read(private_key_path)?;
         encode_from_key_file(claims, &key_data)
@@ -824,7 +797,6 @@ impl LocalEnv {
             control_plane_api,
             generate_local_ssl_certs,
             control_plane_hooks_api,
-            object_storage,
         } = conf;
 
         // Find postgres binaries.
@@ -856,7 +828,6 @@ impl LocalEnv {
         )
         .context("generate auth keys")?;
         let private_key_path = PathBuf::from("auth_private_key.pem");
-        let public_key_path = PathBuf::from("auth_public_key.pem");
 
         // create the runtime type because the remaining initialization code below needs
         // a LocalEnv instance op operation
@@ -867,7 +838,6 @@ impl LocalEnv {
             neon_distrib_dir,
             default_tenant_id: Some(default_tenant_id),
             private_key_path,
-            public_key_path,
             broker,
             storage_controller: storage_controller.unwrap_or_default(),
             pageservers: pageservers.iter().map(Into::into).collect(),
@@ -876,7 +846,6 @@ impl LocalEnv {
             control_plane_hooks_api,
             branch_name_mappings: Default::default(),
             generate_local_ssl_certs,
-            object_storage,
         };
 
         if generate_local_ssl_certs {
@@ -904,13 +873,8 @@ impl LocalEnv {
                 .context("pageserver init failed")?;
         }
 
-        ObjectStorage::from_env(&env)
-            .init()
-            .context("object storage init failed")?;
-
         // setup remote remote location for default LocalFs remote storage
         std::fs::create_dir_all(env.base_data_dir.join(PAGESERVER_REMOTE_STORAGE_DIR))?;
-        std::fs::create_dir_all(env.base_data_dir.join(OBJECT_STORAGE_REMOTE_STORAGE_DIR))?;
 
         env.persist_config()
     }

control_plane/src/object_storage.rs

@@ -1,107 +0,0 @@
-use crate::background_process::{self, start_process, stop_process};
-use crate::local_env::LocalEnv;
-use anyhow::anyhow;
-use anyhow::{Context, Result};
-use camino::Utf8PathBuf;
-use std::io::Write;
-use std::time::Duration;
-
-/// Directory within .neon which will be used by default for LocalFs remote storage.
-pub const OBJECT_STORAGE_REMOTE_STORAGE_DIR: &str = "local_fs_remote_storage/object_storage";
-pub const OBJECT_STORAGE_DEFAULT_PORT: u16 = 9993;
-
-pub struct ObjectStorage {
-    pub bin: Utf8PathBuf,
-    pub data_dir: Utf8PathBuf,
-    pub pemfile: Utf8PathBuf,
-    pub port: u16,
-}
-
-impl ObjectStorage {
-    pub fn from_env(env: &LocalEnv) -> ObjectStorage {
-        ObjectStorage {
-            bin: Utf8PathBuf::from_path_buf(env.object_storage_bin()).unwrap(),
-            data_dir: Utf8PathBuf::from_path_buf(env.object_storage_data_dir()).unwrap(),
-            pemfile: Utf8PathBuf::from_path_buf(env.public_key_path.clone()).unwrap(),
-            port: env.object_storage.port,
-        }
-    }
-
-    fn config_path(&self) -> Utf8PathBuf {
-        self.data_dir.join("object_storage.json")
-    }
-
-    fn listen_addr(&self) -> Utf8PathBuf {
-        format!("127.0.0.1:{}", self.port).into()
-    }
-
-    pub fn init(&self) -> Result<()> {
-        println!("Initializing object storage in {:?}", self.data_dir);
-        let parent = self.data_dir.parent().unwrap();
-
-        #[derive(serde::Serialize)]
-        struct Cfg {
-            listen: Utf8PathBuf,
-            pemfile: Utf8PathBuf,
-            local_path: Utf8PathBuf,
-            r#type: String,
-        }
-        let cfg = Cfg {
-            listen: self.listen_addr(),
-            pemfile: parent.join(self.pemfile.clone()),
-            local_path: parent.join(OBJECT_STORAGE_REMOTE_STORAGE_DIR),
-            r#type: "LocalFs".to_string(),
-        };
-        std::fs::create_dir_all(self.config_path().parent().unwrap())?;
-        std::fs::write(self.config_path(), serde_json::to_string(&cfg)?)
-            .context("write object storage config")?;
-        Ok(())
-    }
-
-    pub async fn start(&self, retry_timeout: &Duration) -> Result<()> {
-        println!("Starting s3 proxy at {}", self.listen_addr());
-        std::io::stdout().flush().context("flush stdout")?;
-
-        let process_status_check = || async {
-            tokio::time::sleep(Duration::from_millis(500)).await;
-            let res = reqwest::Client::new()
-                .get(format!("http://{}/metrics", self.listen_addr()))
-                .send()
-                .await;
-            match res {
-                Ok(response) if response.status().is_success() => Ok(true),
-                Ok(_) => Err(anyhow!("Failed to query /metrics")),
-                Err(e) => Err(anyhow!("Failed to check node status: {e}")),
-            }
-        };
-
-        let res = start_process(
-            "object_storage",
-            &self.data_dir.clone().into_std_path_buf(),
-            &self.bin.clone().into_std_path_buf(),
-            vec![self.config_path().to_string()],
-            vec![("RUST_LOG".into(), "debug".into())],
-            background_process::InitialPidFile::Create(self.pid_file()),
-            retry_timeout,
-            process_status_check,
-        )
-        .await;
-        if res.is_err() {
-            eprintln!("Logs:\n{}", std::fs::read_to_string(self.log_file())?);
-        }
-
-        res
-    }
-
-    pub fn stop(&self, immediate: bool) -> anyhow::Result<()> {
-        stop_process(immediate, "object_storage", &self.pid_file())
-    }
-
-    fn log_file(&self) -> Utf8PathBuf {
-        self.data_dir.join("object_storage.log")
-    }
-
-    fn pid_file(&self) -> Utf8PathBuf {
-        self.data_dir.join("object_storage.pid")
-    }
-}

control_plane/storcon_cli/src/main.rs

@@ -941,7 +941,7 @@ async fn main() -> anyhow::Result<()> {
             let mut node_to_fill_descs = Vec::new();
 
             for desc in node_descs {
-                let to_drain = nodes.contains(&desc.id);
+                let to_drain = nodes.iter().any(|id| *id == desc.id);
                 if to_drain {
                     node_to_drain_descs.push(desc);
                 } else {

docker-compose/compute_wrapper/shell/compute.sh

@@ -11,8 +11,8 @@ generate_id() {
 
 PG_VERSION=${PG_VERSION:-14}
 
-CONFIG_FILE_ORG=/var/db/postgres/configs/config.json
-CONFIG_FILE=/tmp/config.json
+SPEC_FILE_ORG=/var/db/postgres/specs/spec.json
+SPEC_FILE=/tmp/spec.json
 
 echo "Waiting pageserver become ready."
 while ! nc -z pageserver 6400; do
@@ -20,7 +20,7 @@ while ! nc -z pageserver 6400; do
 done
 echo "Page server is ready."
 
-cp ${CONFIG_FILE_ORG} ${CONFIG_FILE}
+cp ${SPEC_FILE_ORG} ${SPEC_FILE}
 
  if [ -n "${TENANT_ID:-}" ] && [ -n "${TIMELINE_ID:-}" ]; then
    tenant_id=${TENANT_ID}
@@ -73,17 +73,17 @@ else
   ulid_extension=ulid
 fi
 echo "Adding pgx_ulid"
-shared_libraries=$(jq -r '.spec.cluster.settings[] | select(.name=="shared_preload_libraries").value' ${CONFIG_FILE})
-sed -i "s/${shared_libraries}/${shared_libraries},${ulid_extension}/" ${CONFIG_FILE}
+shared_libraries=$(jq -r '.cluster.settings[] | select(.name=="shared_preload_libraries").value' ${SPEC_FILE})
+sed -i "s/${shared_libraries}/${shared_libraries},${ulid_extension}/" ${SPEC_FILE}
 echo "Overwrite tenant id and timeline id in spec file"
-sed -i "s/TENANT_ID/${tenant_id}/" ${CONFIG_FILE}
-sed -i "s/TIMELINE_ID/${timeline_id}/" ${CONFIG_FILE}
+sed -i "s/TENANT_ID/${tenant_id}/" ${SPEC_FILE}
+sed -i "s/TIMELINE_ID/${timeline_id}/" ${SPEC_FILE}
 
-cat ${CONFIG_FILE}
+cat ${SPEC_FILE}
 
 echo "Start compute node"
 /usr/local/bin/compute_ctl --pgdata /var/db/postgres/compute \
      -C "postgresql://cloud_admin@localhost:55433/postgres"  \
      -b /usr/local/bin/postgres                              \
      --compute-id "compute-$RANDOM"                          \
-     --config "$CONFIG_FILE"
+     -S ${SPEC_FILE}

docker-compose/compute_wrapper/var/db/postgres/configs/config.json

@@ -1,148 +0,0 @@
-{
-    "spec": {
-        "format_version": 1.0,
-
-        "timestamp": "2022-10-12T18:00:00.000Z",
-        "operation_uuid": "0f657b36-4b0f-4a2d-9c2e-1dcd615e7d8c",
-
-        "cluster": {
-            "cluster_id": "docker_compose",
-            "name": "docker_compose_test",
-            "state": "restarted",
-            "roles": [
-                {
-                    "name": "cloud_admin",
-                    "encrypted_password": "b093c0d3b281ba6da1eacc608620abd8",
-                    "options": null
-                }
-            ],
-            "databases": [
-            ],
-            "settings": [
-                {
-                    "name": "fsync",
-                    "value": "off",
-                    "vartype": "bool"
-                },
-                {
-                    "name": "wal_level",
-                    "value": "logical",
-                    "vartype": "enum"
-                },
-                {
-                    "name": "wal_log_hints",
-                    "value": "on",
-                    "vartype": "bool"
-                },
-                {
-                    "name": "log_connections",
-                    "value": "on",
-                    "vartype": "bool"
-                },
-                {
-                    "name": "port",
-                    "value": "55433",
-                    "vartype": "integer"
-                },
-                {
-                    "name": "shared_buffers",
-                    "value": "1MB",
-                    "vartype": "string"
-                },
-                {
-                    "name": "max_connections",
-                    "value": "100",
-                    "vartype": "integer"
-                },
-                {
-                    "name": "listen_addresses",
-                    "value": "0.0.0.0",
-                    "vartype": "string"
-                },
-                {
-                    "name": "max_wal_senders",
-                    "value": "10",
-                    "vartype": "integer"
-                },
-                {
-                    "name": "max_replication_slots",
-                    "value": "10",
-                    "vartype": "integer"
-                },
-                {
-                    "name": "wal_sender_timeout",
-                    "value": "5s",
-                    "vartype": "string"
-                },
-                {
-                    "name": "wal_keep_size",
-                    "value": "0",
-                    "vartype": "integer"
-                },
-                {
-                    "name": "password_encryption",
-                    "value": "md5",
-                    "vartype": "enum"
-                },
-                {
-                    "name": "restart_after_crash",
-                    "value": "off",
-                    "vartype": "bool"
-                },
-                {
-                    "name": "synchronous_standby_names",
-                    "value": "walproposer",
-                    "vartype": "string"
-                },
-                {
-                    "name": "shared_preload_libraries",
-                    "value": "neon,pg_cron,timescaledb,pg_stat_statements",
-                    "vartype": "string"
-                },
-                {
-                    "name": "neon.safekeepers",
-                    "value": "safekeeper1:5454,safekeeper2:5454,safekeeper3:5454",
-                    "vartype": "string"
-                },
-                {
-                    "name": "neon.timeline_id",
-                    "value": "TIMELINE_ID",
-                    "vartype": "string"
-                },
-                {
-                    "name": "neon.tenant_id",
-                    "value": "TENANT_ID",
-                    "vartype": "string"
-                },
-                {
-                    "name": "neon.pageserver_connstring",
-                    "value": "host=pageserver port=6400",
-                    "vartype": "string"
-                },
-                {
-                    "name": "max_replication_write_lag",
-                    "value": "500MB",
-                    "vartype": "string"
-                },
-                {
-                    "name": "max_replication_flush_lag",
-                    "value": "10GB",
-                    "vartype": "string"
-                },
-                {
-                    "name": "cron.database",
-                    "value": "postgres",
-                    "vartype": "string"
-                }
-            ]
-        },
-
-        "delta_operations": [
-        ]
-    },
-    "compute_ctl_config": {
-        "jwks": {
-            "keys": []
-        }
-    }
-}

docker-compose/compute_wrapper/var/db/postgres/specs/spec.json

@@ -0,0 +1,141 @@
+{
+    "format_version": 1.0,
+
+    "timestamp": "2022-10-12T18:00:00.000Z",
+    "operation_uuid": "0f657b36-4b0f-4a2d-9c2e-1dcd615e7d8c",
+
+    "cluster": {
+        "cluster_id": "docker_compose",
+        "name": "docker_compose_test",
+        "state": "restarted",
+        "roles": [
+            {
+                "name": "cloud_admin",
+                "encrypted_password": "b093c0d3b281ba6da1eacc608620abd8",
+                "options": null
+            }
+        ],
+        "databases": [
+        ],
+        "settings": [
+            {
+                "name": "fsync",
+                "value": "off",
+                "vartype": "bool"
+            },
+            {
+                "name": "wal_level",
+                "value": "logical",
+                "vartype": "enum"
+            },
+            {
+                "name": "wal_log_hints",
+                "value": "on",
+                "vartype": "bool"
+            },
+            {
+                "name": "log_connections",
+                "value": "on",
+                "vartype": "bool"
+            },
+            {
+                "name": "port",
+                "value": "55433",
+                "vartype": "integer"
+            },
+            {
+                "name": "shared_buffers",
+                "value": "1MB",
+                "vartype": "string"
+            },
+            {
+                "name": "max_connections",
+                "value": "100",
+                "vartype": "integer"
+            },
+            {
+                "name": "listen_addresses",
+                "value": "0.0.0.0",
+                "vartype": "string"
+            },
+            {
+                "name": "max_wal_senders",
+                "value": "10",
+                "vartype": "integer"
+            },
+            {
+                "name": "max_replication_slots",
+                "value": "10",
+                "vartype": "integer"
+            },
+            {
+                "name": "wal_sender_timeout",
+                "value": "5s",
+                "vartype": "string"
+            },
+            {
+                "name": "wal_keep_size",
+                "value": "0",
+                "vartype": "integer"
+            },
+            {
+                "name": "password_encryption",
+                "value": "md5",
+                "vartype": "enum"
+            },
+            {
+                "name": "restart_after_crash",
+                "value": "off",
+                "vartype": "bool"
+            },
+            {
+                "name": "synchronous_standby_names",
+                "value": "walproposer",
+                "vartype": "string"
+            },
+            {
+                "name": "shared_preload_libraries",
+                "value": "neon,pg_cron,timescaledb,pg_stat_statements",
+                "vartype": "string"
+            },
+            {
+                "name": "neon.safekeepers",
+                "value": "safekeeper1:5454,safekeeper2:5454,safekeeper3:5454",
+                "vartype": "string"
+            },
+            {
+                "name": "neon.timeline_id",
+                "value": "TIMELINE_ID",
+                "vartype": "string"
+            },
+            {
+                "name": "neon.tenant_id",
+                "value": "TENANT_ID",
+                "vartype": "string"
+            },
+            {
+                "name": "neon.pageserver_connstring",
+                "value": "host=pageserver port=6400",
+                "vartype": "string"
+            },
+            {
+                "name": "max_replication_write_lag",
+                "value": "500MB",
+                "vartype": "string"
+            },
+            {
+                "name": "max_replication_flush_lag",
+                "value": "10GB",
+                "vartype": "string"
+            },
+            {
+                "name": "cron.database",
+                "value": "postgres",
+                "vartype": "string"
+            }
+        ]
+    },
+
+    "delta_operations": [
+    ]
+}

docker-compose/docker-compose.yml

@@ -159,7 +159,7 @@ services:
       #- RUST_BACKTRACE=1
     # Mount the test files directly, for faster editing cycle.
     volumes:
-      - ./compute_wrapper/var/db/postgres/configs/:/var/db/postgres/configs/
+      - ./compute_wrapper/var/db/postgres/specs/:/var/db/postgres/specs/
       - ./compute_wrapper/shell/:/shell/
     ports:
       - 55433:55433 # pg protocol handler

docs/storage_controller.md

@@ -151,7 +151,7 @@ Example body:
 ```
 {
   "tenant_id": "1f359dd625e519a1a4e8d7509690f6fc",
-  "stripe_size": 2048,
+  "stripe_size": 32768,
   "shards": [
       {"node_id": 344, "shard_number": 0},
       {"node_id": 722, "shard_number": 1},

libs/compute_api/src/requests.rs

@@ -5,14 +5,6 @@ use crate::privilege::Privilege;
 use crate::responses::ComputeCtlConfig;
 use crate::spec::{ComputeSpec, ExtVersion, PgIdent};
 
-/// When making requests to the `compute_ctl` external HTTP server, the client
-/// must specify a set of claims in `Authorization` header JWTs such that
-/// `compute_ctl` can authorize the request.
-#[derive(Clone, Debug, Deserialize, Serialize)]
-pub struct ComputeClaims {
-    pub compute_id: String,
-}
-
 /// Request of the /configure API
 ///
 /// We now pass only `spec` in the configuration request, but later we can
@@ -38,3 +30,9 @@ pub struct SetRoleGrantsRequest {
     pub privileges: Vec<Privilege>,
     pub role: PgIdent,
 }
+
+/// Request of the /configure_telemetry API
+#[derive(Debug, Deserialize, Serialize)]
+pub struct ConfigureTelemetryRequest {
+    pub logs_export_host: Option<String>,
+}

libs/compute_api/src/responses.rs

@@ -14,32 +14,6 @@ pub struct GenericAPIError {
     pub error: String,
 }
 
-/// All configuration parameters necessary for a compute. When
-/// [`ComputeConfig::spec`] is provided, it means that the compute is attached
-/// to a tenant. [`ComputeConfig::compute_ctl_config`] will always be provided
-/// and contains parameters necessary for operating `compute_ctl` independently
-/// of whether a tenant is attached to the compute or not.
-///
-/// This also happens to be the body of `compute_ctl`'s /configure request.
-#[derive(Debug, Deserialize, Serialize)]
-pub struct ComputeConfig {
-    /// The compute spec
-    pub spec: Option<ComputeSpec>,
-
-    /// The compute_ctl configuration
-    #[allow(dead_code)]
-    pub compute_ctl_config: ComputeCtlConfig,
-}
-
-impl From<ControlPlaneConfigResponse> for ComputeConfig {
-    fn from(value: ControlPlaneConfigResponse) -> Self {
-        Self {
-            spec: value.spec,
-            compute_ctl_config: value.compute_ctl_config,
-        }
-    }
-}
-
 #[derive(Debug, Clone, Serialize)]
 pub struct ExtensionInstallResponse {
     pub extension: PgIdent,
@@ -187,7 +161,7 @@ pub struct TlsConfig {
 
 /// Response of the `/computes/{compute_id}/spec` control-plane API.
 #[derive(Deserialize, Debug)]
-pub struct ControlPlaneConfigResponse {
+pub struct ControlPlaneSpecResponse {
     pub spec: Option<ComputeSpec>,
     pub status: ControlPlaneComputeStatus,
     pub compute_ctl_config: ComputeCtlConfig,

libs/compute_api/src/spec.rs

@@ -1,8 +1,8 @@
-//! The ComputeSpec contains all the information needed to start up
-//! the right version of PostgreSQL, and connect it to the storage nodes.
-//! It can be passed as part of the `config.json`, or the control plane can
-//! provide it by calling the compute_ctl's `/compute_ctl` endpoint, or
-//! compute_ctl can fetch it by calling the control plane's API.
+//! `ComputeSpec` represents the contents of the spec.json file.
+//!
+//! The spec.json file is used to pass information to 'compute_ctl'. It contains
+//! all the information needed to start up the right version of PostgreSQL,
+//! and connect it to the storage nodes.
 use std::collections::HashMap;
 
 use indexmap::IndexMap;
@@ -104,12 +104,6 @@ pub struct ComputeSpec {
     pub timeline_id: Option<TimelineId>,
     pub pageserver_connstring: Option<String>,
 
-    // More neon ids that we expose to the compute_ctl
-    // and to postgres as neon extension GUCs.
-    pub project_id: Option<String>,
-    pub branch_id: Option<String>,
-    pub endpoint_id: Option<String>,
-
     /// Safekeeper membership config generation. It is put in
     /// neon.safekeepers GUC and serves two purposes:
     /// 1) Non zero value forces walproposer to use membership configurations.
@@ -165,13 +159,15 @@ pub struct ComputeSpec {
     #[serde(default)] // Default false
     pub drop_subscriptions_before_start: bool,
 
-    /// Log level for compute audit logging
+    /// Log level for audit logging:
+    ///
+    /// Disabled - no audit logging. This is the default.
+    /// log - log masked statements to the postgres log using pgaudit extension
+    /// hipaa - log unmasked statements to the file using pgaudit and pgauditlogtofile extension
+    ///
+    /// Extensions should be present in shared_preload_libraries
     #[serde(default)]
     pub audit_log_level: ComputeAudit,
-
-    /// Hostname and the port of the otel collector. Leave empty to disable Postgres logs forwarding.
-    /// Example: config-shy-breeze-123-collector-monitoring.neon-telemetry.svc.cluster.local:10514
-    pub logs_export_host: Option<String>,
 }
 
 /// Feature flag to signal `compute_ctl` to enable certain experimental functionality.
@@ -183,6 +179,9 @@ pub enum ComputeFeature {
     /// track short-lived connections as user activity.
     ActivityMonitorExperimental,
 
+    /// Allow to configure rsyslog for Postgres logs export
+    PostgresLogsExport,
+
     /// This is a special feature flag that is used to represent unknown feature flags.
     /// Basically all unknown to enum flags are represented as this one. See unit test
     /// `parse_unknown_features()` for more details.
@@ -289,25 +288,14 @@ impl ComputeMode {
 }
 
 /// Log level for audit logging
+/// Disabled, log, hipaa
+/// Default is Disabled
 #[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize, Serialize)]
 pub enum ComputeAudit {
     #[default]
     Disabled,
-    // Deprecated, use Base instead
     Log,
-    // (pgaudit.log = 'ddl', pgaudit.log_parameter='off')
-    // logged to the standard postgresql log stream
-    Base,
-    // Deprecated, use Full or Extended instead
     Hipaa,
-    // (pgaudit.log = 'all, -misc', pgaudit.log_parameter='off')
-    // logged to separate files collected by rsyslog
-    // into dedicated log storage with strict access
-    Extended,
-    // (pgaudit.log='all', pgaudit.log_parameter='on'),
-    // logged to separate files collected by rsyslog
-    // into dedicated log storage with strict access.
-    Full,
 }
 
 #[derive(Clone, Debug, Default, Deserialize, Serialize, PartialEq, Eq)]

libs/http-utils/Cargo.toml

@@ -30,7 +30,6 @@ tokio.workspace = true
 tracing.workspace = true
 url.workspace = true
 uuid.workspace = true
-x509-cert.workspace = true
 
 # to use tokio channels as streams, this is faster to compile than async_stream
 # why is it only here? no other crate should use it, streams are rarely needed.

libs/http-utils/src/server.rs

@@ -4,8 +4,6 @@ use futures::StreamExt;
 use futures::stream::FuturesUnordered;
 use hyper0::Body;
 use hyper0::server::conn::Http;
-use metrics::{IntCounterVec, register_int_counter_vec};
-use once_cell::sync::Lazy;
 use routerify::{RequestService, RequestServiceBuilder};
 use tokio::io::{AsyncRead, AsyncWrite};
 use tokio_rustls::TlsAcceptor;
@@ -28,24 +26,6 @@ pub struct Server {
     tls_acceptor: Option<TlsAcceptor>,
 }
 
-static CONNECTION_STARTED_COUNT: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "http_server_connection_started_total",
-        "Number of established http/https connections",
-        &["scheme"]
-    )
-    .expect("failed to define a metric")
-});
-
-static CONNECTION_ERROR_COUNT: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "http_server_connection_errors_total",
-        "Number of occured connection errors by type",
-        &["type"]
-    )
-    .expect("failed to define a metric")
-});
-
 impl Server {
     pub fn new(
         request_service: Arc<RequestServiceBuilder<Body, ApiError>>,
@@ -80,15 +60,6 @@ impl Server {
             false
         }
 
-        let tcp_error_cnt = CONNECTION_ERROR_COUNT.with_label_values(&["tcp"]);
-        let tls_error_cnt = CONNECTION_ERROR_COUNT.with_label_values(&["tls"]);
-        let http_error_cnt = CONNECTION_ERROR_COUNT.with_label_values(&["http"]);
-        let https_error_cnt = CONNECTION_ERROR_COUNT.with_label_values(&["https"]);
-        let panic_error_cnt = CONNECTION_ERROR_COUNT.with_label_values(&["panic"]);
-
-        let http_connection_cnt = CONNECTION_STARTED_COUNT.with_label_values(&["http"]);
-        let https_connection_cnt = CONNECTION_STARTED_COUNT.with_label_values(&["https"]);
-
         let mut connections = FuturesUnordered::new();
         loop {
             tokio::select! {
@@ -96,7 +67,6 @@ impl Server {
                     let (tcp_stream, remote_addr) = match stream {
                         Ok(stream) => stream,
                         Err(err) => {
-                            tcp_error_cnt.inc();
                             if !suppress_io_error(&err) {
                                 info!("Failed to accept TCP connection: {err:#}");
                             }
@@ -108,18 +78,11 @@ impl Server {
                     let tls_acceptor = self.tls_acceptor.clone();
                     let cancel = cancel.clone();
 
-                    let tls_error_cnt = tls_error_cnt.clone();
-                    let http_error_cnt = http_error_cnt.clone();
-                    let https_error_cnt = https_error_cnt.clone();
-                    let http_connection_cnt = http_connection_cnt.clone();
-                    let https_connection_cnt = https_connection_cnt.clone();
-
                     connections.push(tokio::spawn(
                         async move {
                             match tls_acceptor {
                                 Some(tls_acceptor) => {
                                     // Handle HTTPS connection.
-                                    https_connection_cnt.inc();
                                     let tls_stream = tokio::select! {
                                         tls_stream = tls_acceptor.accept(tcp_stream) => tls_stream,
                                         _ = cancel.cancelled() => return,
@@ -127,7 +90,6 @@ impl Server {
                                     let tls_stream = match tls_stream {
                                         Ok(tls_stream) => tls_stream,
                                         Err(err) => {
-                                            tls_error_cnt.inc();
                                             if !suppress_io_error(&err) {
                                                 info!(%remote_addr, "Failed to accept TLS connection: {err:#}");
                                             }
@@ -135,7 +97,6 @@ impl Server {
                                         }
                                     };
                                     if let Err(err) = Self::serve_connection(tls_stream, service, cancel).await {
-                                        https_error_cnt.inc();
                                         if !suppress_hyper_error(&err) {
                                             info!(%remote_addr, "Failed to serve HTTPS connection: {err:#}");
                                         }
@@ -143,9 +104,7 @@ impl Server {
                                 }
                                 None => {
                                     // Handle HTTP connection.
-                                    http_connection_cnt.inc();
                                     if let Err(err) = Self::serve_connection(tcp_stream, service, cancel).await {
-                                        http_error_cnt.inc();
                                         if !suppress_hyper_error(&err) {
                                             info!(%remote_addr, "Failed to serve HTTP connection: {err:#}");
                                         }
@@ -156,7 +115,6 @@ impl Server {
                  }
                 Some(conn) = connections.next() => {
                     if let Err(err) = conn {
-                        panic_error_cnt.inc();
                         error!("Connection panicked: {err:#}");
                     }
                 }
@@ -164,7 +122,6 @@ impl Server {
                     // Wait for graceful shutdown of all connections.
                     while let Some(conn) = connections.next().await {
                         if let Err(err) = conn {
-                            panic_error_cnt.inc();
                             error!("Connection panicked: {err:#}");
                         }
                     }

libs/http-utils/src/tls_certs.rs

@@ -3,14 +3,11 @@ use std::{sync::Arc, time::Duration};
 use anyhow::Context;
 use arc_swap::ArcSwap;
 use camino::Utf8Path;
-use metrics::{IntCounterVec, UIntGaugeVec, register_int_counter_vec, register_uint_gauge_vec};
-use once_cell::sync::Lazy;
 use rustls::{
-    pki_types::{CertificateDer, PrivateKeyDer, UnixTime},
+    pki_types::{CertificateDer, PrivateKeyDer},
     server::{ClientHello, ResolvesServerCert},
     sign::CertifiedKey,
 };
-use x509_cert::der::Reader;
 
 pub async fn load_cert_chain(filename: &Utf8Path) -> anyhow::Result<Vec<CertificateDer<'static>>> {
     let cert_data = tokio::fs::read(filename)
@@ -56,76 +53,6 @@ pub async fn load_certified_key(
     Ok(certified_key)
 }
 
-/// rustls's CertifiedKey with extra parsed fields used for metrics.
-struct ParsedCertifiedKey {
-    certified_key: CertifiedKey,
-    expiration_time: UnixTime,
-}
-
-/// Parse expiration time from an X509 certificate.
-fn parse_expiration_time(cert: &CertificateDer<'_>) -> anyhow::Result<UnixTime> {
-    let parsed_cert = x509_cert::der::SliceReader::new(cert)
-        .context("Failed to parse cerficiate")?
-        .decode::<x509_cert::Certificate>()
-        .context("Failed to parse cerficiate")?;
-
-    Ok(UnixTime::since_unix_epoch(
-        parsed_cert
-            .tbs_certificate
-            .validity
-            .not_after
-            .to_unix_duration(),
-    ))
-}
-
-async fn load_and_parse_certified_key(
-    key_filename: &Utf8Path,
-    cert_filename: &Utf8Path,
-) -> anyhow::Result<ParsedCertifiedKey> {
-    let certified_key = load_certified_key(key_filename, cert_filename).await?;
-    let expiration_time = parse_expiration_time(certified_key.end_entity_cert()?)?;
-    Ok(ParsedCertifiedKey {
-        certified_key,
-        expiration_time,
-    })
-}
-
-static CERT_EXPIRATION_TIME: Lazy<UIntGaugeVec> = Lazy::new(|| {
-    register_uint_gauge_vec!(
-        "tls_certs_expiration_time_seconds",
-        "Expiration time of the loaded certificate since unix epoch in seconds",
-        &["resolver_name"]
-    )
-    .expect("failed to define a metric")
-});
-
-static CERT_RELOAD_STARTED_COUNTER: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "tls_certs_reload_started_total",
-        "Number of certificate reload loop iterations started",
-        &["resolver_name"]
-    )
-    .expect("failed to define a metric")
-});
-
-static CERT_RELOAD_UPDATED_COUNTER: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "tls_certs_reload_updated_total",
-        "Number of times the certificate was updated to the new one",
-        &["resolver_name"]
-    )
-    .expect("failed to define a metric")
-});
-
-static CERT_RELOAD_FAILED_COUNTER: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "tls_certs_reload_failed_total",
-        "Number of times the certificate reload failed",
-        &["resolver_name"]
-    )
-    .expect("failed to define a metric")
-});
-
 /// Implementation of [`rustls::server::ResolvesServerCert`] which reloads certificates from
 /// the disk periodically.
 #[derive(Debug)]
@@ -136,28 +63,16 @@ pub struct ReloadingCertificateResolver {
 impl ReloadingCertificateResolver {
     /// Creates a new Resolver by loading certificate and private key from FS and
     /// creating tokio::task to reload them with provided reload_period.
-    /// resolver_name is used as metric's label.
     pub async fn new(
-        resolver_name: &str,
         key_filename: &Utf8Path,
         cert_filename: &Utf8Path,
         reload_period: Duration,
     ) -> anyhow::Result<Arc<Self>> {
-        // Create metrics for current resolver.
-        let cert_expiration_time = CERT_EXPIRATION_TIME.with_label_values(&[resolver_name]);
-        let cert_reload_started_counter =
-            CERT_RELOAD_STARTED_COUNTER.with_label_values(&[resolver_name]);
-        let cert_reload_updated_counter =
-            CERT_RELOAD_UPDATED_COUNTER.with_label_values(&[resolver_name]);
-        let cert_reload_failed_counter =
-            CERT_RELOAD_FAILED_COUNTER.with_label_values(&[resolver_name]);
-
-        let parsed_key = load_and_parse_certified_key(key_filename, cert_filename).await?;
-
         let this = Arc::new(Self {
-            certified_key: ArcSwap::from_pointee(parsed_key.certified_key),
+            certified_key: ArcSwap::from_pointee(
+                load_certified_key(key_filename, cert_filename).await?,
+            ),
         });
-        cert_expiration_time.set(parsed_key.expiration_time.as_secs());
 
         tokio::spawn({
             let weak_this = Arc::downgrade(&this);
@@ -173,22 +88,17 @@ impl ReloadingCertificateResolver {
                         Some(this) => this,
                         None => break, // Resolver has been destroyed, exit.
                     };
-                    cert_reload_started_counter.inc();
-
-                    match load_and_parse_certified_key(&key_filename, &cert_filename).await {
-                        Ok(parsed_key) => {
-                            if parsed_key.certified_key.cert == this.certified_key.load().cert {
+                    match load_certified_key(&key_filename, &cert_filename).await {
+                        Ok(new_certified_key) => {
+                            if new_certified_key.cert == this.certified_key.load().cert {
                                 tracing::debug!("Certificate has not changed since last reloading");
                             } else {
                                 tracing::info!("Certificate has been reloaded");
-                                this.certified_key.store(Arc::new(parsed_key.certified_key));
-                                cert_expiration_time.set(parsed_key.expiration_time.as_secs());
-                                cert_reload_updated_counter.inc();
+                                this.certified_key.store(Arc::new(new_certified_key));
                             }
                             last_reload_failed = false;
                         }
                         Err(err) => {
-                            cert_reload_failed_counter.inc();
                             // Note: Reloading certs may fail if it conflicts with the script updating
                             // the files at the same time. Warn only if the error is persistent.
                             if last_reload_failed {

libs/pageserver_api/src/config.rs

@@ -180,7 +180,6 @@ pub struct ConfigToml {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub generate_unarchival_heatmap: Option<bool>,
     pub tracing: Option<Tracing>,
-    pub enable_tls_page_service_api: bool,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
@@ -632,7 +631,6 @@ impl Default for ConfigToml {
             load_previous_heatmap: None,
             generate_unarchival_heatmap: None,
             tracing: None,
-            enable_tls_page_service_api: false,
         }
     }
 }

libs/pageserver_api/src/keyspace.rs

@@ -613,7 +613,8 @@ mod tests {
     use rand::{RngCore, SeedableRng};
 
     use super::*;
-    use crate::shard::{DEFAULT_STRIPE_SIZE, ShardCount, ShardNumber, ShardStripeSize};
+    use crate::models::ShardParameters;
+    use crate::shard::{ShardCount, ShardNumber};
 
     // Helper function to create a key range.
     //
@@ -963,8 +964,12 @@ mod tests {
     }
     #[test]
     fn sharded_range_relation_gap() {
-        let shard_identity =
-            ShardIdentity::new(ShardNumber(0), ShardCount::new(4), DEFAULT_STRIPE_SIZE).unwrap();
+        let shard_identity = ShardIdentity::new(
+            ShardNumber(0),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
+        )
+        .unwrap();
 
         let range = ShardedRange::new(
             Range {
@@ -980,8 +985,12 @@ mod tests {
 
     #[test]
     fn shard_identity_keyspaces_single_key() {
-        let shard_identity =
-            ShardIdentity::new(ShardNumber(1), ShardCount::new(4), DEFAULT_STRIPE_SIZE).unwrap();
+        let shard_identity = ShardIdentity::new(
+            ShardNumber(1),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
+        )
+        .unwrap();
 
         let range = ShardedRange::new(
             Range {
@@ -1025,8 +1034,12 @@ mod tests {
 
     #[test]
     fn shard_identity_keyspaces_forkno_gap() {
-        let shard_identity =
-            ShardIdentity::new(ShardNumber(1), ShardCount::new(4), DEFAULT_STRIPE_SIZE).unwrap();
+        let shard_identity = ShardIdentity::new(
+            ShardNumber(1),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
+        )
+        .unwrap();
 
         let range = ShardedRange::new(
             Range {
@@ -1048,7 +1061,7 @@ mod tests {
             let shard_identity = ShardIdentity::new(
                 ShardNumber(shard_number),
                 ShardCount::new(4),
-                DEFAULT_STRIPE_SIZE,
+                ShardParameters::DEFAULT_STRIPE_SIZE,
             )
             .unwrap();
 
@@ -1131,44 +1144,37 @@ mod tests {
     /// for a single tenant.
     #[test]
     fn sharded_range_fragment_simple() {
-        const SHARD_COUNT: u8 = 4;
-        const STRIPE_SIZE: u32 = DEFAULT_STRIPE_SIZE.0;
-
         let shard_identity = ShardIdentity::new(
             ShardNumber(0),
-            ShardCount::new(SHARD_COUNT),
-            ShardStripeSize(STRIPE_SIZE),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
         )
         .unwrap();
 
         // A range which we happen to know covers exactly one stripe which belongs to this shard
         let input_start = Key::from_hex("000000067f00000001000000ae0000000000").unwrap();
-        let mut input_end = input_start;
-        input_end.field6 += STRIPE_SIZE; // field6 is block number
+        let input_end = Key::from_hex("000000067f00000001000000ae0000008000").unwrap();
 
         // Ask for stripe_size blocks, we get the whole stripe
         assert_eq!(
-            do_fragment(input_start, input_end, &shard_identity, STRIPE_SIZE),
-            (STRIPE_SIZE, vec![(STRIPE_SIZE, input_start..input_end)])
+            do_fragment(input_start, input_end, &shard_identity, 32768),
+            (32768, vec![(32768, input_start..input_end)])
         );
 
         // Ask for more, we still get the whole stripe
         assert_eq!(
-            do_fragment(input_start, input_end, &shard_identity, 10 * STRIPE_SIZE),
-            (STRIPE_SIZE, vec![(STRIPE_SIZE, input_start..input_end)])
+            do_fragment(input_start, input_end, &shard_identity, 10000000),
+            (32768, vec![(32768, input_start..input_end)])
         );
 
         // Ask for target_nblocks of half the stripe size, we get two halves
         assert_eq!(
-            do_fragment(input_start, input_end, &shard_identity, STRIPE_SIZE / 2),
+            do_fragment(input_start, input_end, &shard_identity, 16384),
             (
-                STRIPE_SIZE,
+                32768,
                 vec![
-                    (
-                        STRIPE_SIZE / 2,
-                        input_start..input_start.add(STRIPE_SIZE / 2)
-                    ),
-                    (STRIPE_SIZE / 2, input_start.add(STRIPE_SIZE / 2)..input_end)
+                    (16384, input_start..input_start.add(16384)),
+                    (16384, input_start.add(16384)..input_end)
                 ]
             )
         );
@@ -1176,53 +1182,40 @@ mod tests {
 
     #[test]
     fn sharded_range_fragment_multi_stripe() {
-        const SHARD_COUNT: u8 = 4;
-        const STRIPE_SIZE: u32 = DEFAULT_STRIPE_SIZE.0;
-        const RANGE_SIZE: u32 = SHARD_COUNT as u32 * STRIPE_SIZE;
-
         let shard_identity = ShardIdentity::new(
             ShardNumber(0),
-            ShardCount::new(SHARD_COUNT),
-            ShardStripeSize(STRIPE_SIZE),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
         )
         .unwrap();
 
         // A range which covers multiple stripes, exactly one of which belongs to the current shard.
         let input_start = Key::from_hex("000000067f00000001000000ae0000000000").unwrap();
-        let mut input_end = input_start;
-        input_end.field6 += RANGE_SIZE; // field6 is block number
-
+        let input_end = Key::from_hex("000000067f00000001000000ae0000020000").unwrap();
         // Ask for all the blocks, get a fragment that covers the whole range but reports
         // its size to be just the blocks belonging to our shard.
         assert_eq!(
-            do_fragment(input_start, input_end, &shard_identity, RANGE_SIZE),
-            (STRIPE_SIZE, vec![(STRIPE_SIZE, input_start..input_end)])
+            do_fragment(input_start, input_end, &shard_identity, 131072),
+            (32768, vec![(32768, input_start..input_end)])
         );
 
-        // Ask for a sub-stripe quantity that results in 3 fragments.
-        let limit = STRIPE_SIZE / 3 + 1;
+        // Ask for a sub-stripe quantity
         assert_eq!(
-            do_fragment(input_start, input_end, &shard_identity, limit),
+            do_fragment(input_start, input_end, &shard_identity, 16000),
             (
-                STRIPE_SIZE,
+                32768,
                 vec![
-                    (limit, input_start..input_start.add(limit)),
-                    (limit, input_start.add(limit)..input_start.add(2 * limit)),
-                    (
-                        STRIPE_SIZE - 2 * limit,
-                        input_start.add(2 * limit)..input_end
-                    ),
+                    (16000, input_start..input_start.add(16000)),
+                    (16000, input_start.add(16000)..input_start.add(32000)),
+                    (768, input_start.add(32000)..input_end),
                 ]
             )
         );
 
         // Try on a range that starts slightly after our owned stripe
         assert_eq!(
-            do_fragment(input_start.add(1), input_end, &shard_identity, RANGE_SIZE),
-            (
-                STRIPE_SIZE - 1,
-                vec![(STRIPE_SIZE - 1, input_start.add(1)..input_end)]
-            )
+            do_fragment(input_start.add(1), input_end, &shard_identity, 131072),
+            (32767, vec![(32767, input_start.add(1)..input_end)])
         );
     }
 
@@ -1230,40 +1223,32 @@ mod tests {
     /// a previous relation.
     #[test]
     fn sharded_range_fragment_starting_from_logical_size() {
-        const SHARD_COUNT: u8 = 4;
-        const STRIPE_SIZE: u32 = DEFAULT_STRIPE_SIZE.0;
-        const RANGE_SIZE: u32 = SHARD_COUNT as u32 * STRIPE_SIZE;
-
         let input_start = Key::from_hex("000000067f00000001000000ae00ffffffff").unwrap();
-        let mut input_end = Key::from_hex("000000067f00000001000000ae0100000000").unwrap();
-        input_end.field6 += RANGE_SIZE; // field6 is block number
+        let input_end = Key::from_hex("000000067f00000001000000ae0100008000").unwrap();
 
         // Shard 0 owns the first stripe in the relation, and the preceding logical size is shard local too
         let shard_identity = ShardIdentity::new(
             ShardNumber(0),
-            ShardCount::new(SHARD_COUNT),
-            ShardStripeSize(STRIPE_SIZE),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
         )
         .unwrap();
         assert_eq!(
-            do_fragment(input_start, input_end, &shard_identity, 2 * STRIPE_SIZE),
-            (
-                STRIPE_SIZE + 1,
-                vec![(STRIPE_SIZE + 1, input_start..input_end)]
-            )
+            do_fragment(input_start, input_end, &shard_identity, 0x10000),
+            (0x8001, vec![(0x8001, input_start..input_end)])
         );
 
         // Shard 1 does not own the first stripe in the relation, but it does own the logical size (all shards
         // store all logical sizes)
         let shard_identity = ShardIdentity::new(
             ShardNumber(1),
-            ShardCount::new(SHARD_COUNT),
-            ShardStripeSize(STRIPE_SIZE),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
         )
         .unwrap();
         assert_eq!(
-            do_fragment(input_start, input_end, &shard_identity, 2 * STRIPE_SIZE),
-            (1, vec![(1, input_start..input_end)])
+            do_fragment(input_start, input_end, &shard_identity, 0x10000),
+            (0x1, vec![(0x1, input_start..input_end)])
         );
     }
 
@@ -1299,8 +1284,12 @@ mod tests {
         );
 
         // Same, but using a sharded identity
-        let shard_identity =
-            ShardIdentity::new(ShardNumber(0), ShardCount::new(4), DEFAULT_STRIPE_SIZE).unwrap();
+        let shard_identity = ShardIdentity::new(
+            ShardNumber(0),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
+        )
+        .unwrap();
         assert_eq!(
             do_fragment(input_start, input_end, &shard_identity, 0x8000),
             (u32::MAX, vec![(u32::MAX, input_start..input_end),])
@@ -1342,7 +1331,7 @@ mod tests {
                 ShardIdentity::new(
                     ShardNumber((prng.next_u32() % shard_count) as u8),
                     ShardCount::new(shard_count as u8),
-                    DEFAULT_STRIPE_SIZE,
+                    ShardParameters::DEFAULT_STRIPE_SIZE,
                 )
                 .unwrap()
             };

libs/pageserver_api/src/models.rs

@@ -26,7 +26,7 @@ use utils::{completion, serde_system_time};
 use crate::config::Ratio;
 use crate::key::{CompactKey, Key};
 use crate::reltag::RelTag;
-use crate::shard::{DEFAULT_STRIPE_SIZE, ShardCount, ShardStripeSize, TenantShardId};
+use crate::shard::{ShardCount, ShardStripeSize, TenantShardId};
 
 /// The state of a tenant in this pageserver.
 ///
@@ -80,22 +80,10 @@ pub enum TenantState {
     ///
     /// Transitions out of this state are possible through `set_broken()`.
     Stopping {
-        /// The barrier can be used to wait for shutdown to complete. The first caller to set
-        /// Some(Barrier) is responsible for driving shutdown to completion. Subsequent callers
-        /// will wait for the first caller's existing barrier.
-        ///
-        /// None is set when an attach is cancelled, to signal to shutdown that the attach has in
-        /// fact cancelled:
-        ///
-        /// 1. `shutdown` sees `TenantState::Attaching`, and cancels the tenant.
-        /// 2. `attach` sets `TenantState::Stopping(None)` and exits.
-        /// 3. `set_stopping` waits for `TenantState::Stopping(None)` and sets
-        ///    `TenantState::Stopping(Some)` to claim the barrier as the shutdown owner.
-        //
         // Because of https://github.com/serde-rs/serde/issues/2105 this has to be a named field,
         // otherwise it will not be skipped during deserialization
         #[serde(skip)]
-        progress: Option<completion::Barrier>,
+        progress: completion::Barrier,
     },
     /// The tenant is recognized by the pageserver, but can no longer be used for
     /// any operations.
@@ -438,6 +426,8 @@ pub struct ShardParameters {
 }
 
 impl ShardParameters {
+    pub const DEFAULT_STRIPE_SIZE: ShardStripeSize = ShardStripeSize(256 * 1024 / 8);
+
     pub fn is_unsharded(&self) -> bool {
         self.count.is_unsharded()
     }
@@ -447,7 +437,7 @@ impl Default for ShardParameters {
     fn default() -> Self {
         Self {
             count: ShardCount::new(0),
-            stripe_size: DEFAULT_STRIPE_SIZE,
+            stripe_size: Self::DEFAULT_STRIPE_SIZE,
         }
     }
 }
@@ -1678,7 +1668,6 @@ pub struct SecondaryProgress {
 pub struct TenantScanRemoteStorageShard {
     pub tenant_shard_id: TenantShardId,
     pub generation: Option<u32>,
-    pub stripe_size: Option<ShardStripeSize>,
 }
 
 #[derive(Serialize, Deserialize, Debug, Default)]
@@ -2730,15 +2719,10 @@ mod tests {
                 "Activating",
             ),
             (line!(), TenantState::Active, "Active"),
-            (
-                line!(),
-                TenantState::Stopping { progress: None },
-                "Stopping",
-            ),
             (
                 line!(),
                 TenantState::Stopping {
-                    progress: Some(completion::Barrier::default()),
+                    progress: utils::completion::Barrier::default(),
                 },
                 "Stopping",
             ),

libs/pageserver_api/src/record.rs

@@ -58,8 +58,6 @@ pub enum NeonWalRecord {
         /// to true. This record does not need the history WALs to reconstruct. See [`NeonWalRecord::will_init`] and
         /// its references in `timeline.rs`.
         will_init: bool,
-        /// Only append the record if the current image is the same as the one specified in this field.
-        only_if: Option<String>,
     },
 }
 
@@ -83,17 +81,6 @@ impl NeonWalRecord {
             append: s.as_ref().to_string(),
             clear: false,
             will_init: false,
-            only_if: None,
-        }
-    }
-
-    #[cfg(feature = "testing")]
-    pub fn wal_append_conditional(s: impl AsRef<str>, only_if: impl AsRef<str>) -> Self {
-        Self::Test {
-            append: s.as_ref().to_string(),
-            clear: false,
-            will_init: false,
-            only_if: Some(only_if.as_ref().to_string()),
         }
     }
 
@@ -103,7 +90,6 @@ impl NeonWalRecord {
             append: s.as_ref().to_string(),
             clear: true,
             will_init: false,
-            only_if: None,
         }
     }
 
@@ -113,7 +99,6 @@ impl NeonWalRecord {
             append: s.as_ref().to_string(),
             clear: true,
             will_init: true,
-            only_if: None,
         }
     }
 }

libs/pageserver_api/src/shard.rs

@@ -78,12 +78,6 @@ impl Default for ShardStripeSize {
     }
 }
 
-impl std::fmt::Display for ShardStripeSize {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        self.0.fmt(f)
-    }
-}
-
 /// Layout version: for future upgrades where we might change how the key->shard mapping works
 #[derive(Clone, Copy, Serialize, Deserialize, Eq, PartialEq, Hash, Debug)]
 pub struct ShardLayout(u8);
@@ -92,11 +86,8 @@ const LAYOUT_V1: ShardLayout = ShardLayout(1);
 /// ShardIdentity uses a magic layout value to indicate if it is unusable
 const LAYOUT_BROKEN: ShardLayout = ShardLayout(255);
 
-/// The default stripe size in pages. 16 MiB divided by 8 kiB page size.
-///
-/// A lower stripe size distributes ingest load better across shards, but reduces IO amortization.
-/// 16 MiB appears to be a reasonable balance: <https://github.com/neondatabase/neon/pull/10510>.
-pub const DEFAULT_STRIPE_SIZE: ShardStripeSize = ShardStripeSize(16 * 1024 / 8);
+/// Default stripe size in pages: 256MiB divided by 8kiB page size.
+const DEFAULT_STRIPE_SIZE: ShardStripeSize = ShardStripeSize(256 * 1024 / 8);
 
 #[derive(thiserror::Error, Debug, PartialEq, Eq)]
 pub enum ShardConfigError {
@@ -546,7 +537,7 @@ mod tests {
             field6: 0x7d06,
         };
 
-        let shard = key_to_shard_number(ShardCount(10), ShardStripeSize(32768), &key);
+        let shard = key_to_shard_number(ShardCount(10), DEFAULT_STRIPE_SIZE, &key);
         assert_eq!(shard, ShardNumber(8));
     }
 

libs/postgres_backend/src/lib.rs

@@ -5,6 +5,7 @@
 #![deny(unsafe_code)]
 #![deny(clippy::undocumented_unsafe_blocks)]
 use std::future::Future;
+use std::io::ErrorKind;
 use std::net::SocketAddr;
 use std::os::fd::{AsRawFd, RawFd};
 use std::pin::Pin;
@@ -226,7 +227,7 @@ impl<IO: AsyncRead + AsyncWrite + Unpin> MaybeWriteOnly<IO> {
         match self {
             MaybeWriteOnly::Full(framed) => framed.read_startup_message().await,
             MaybeWriteOnly::WriteOnly(_) => {
-                Err(io::Error::other("reading from write only half").into())
+                Err(io::Error::new(ErrorKind::Other, "reading from write only half").into())
             }
             MaybeWriteOnly::Broken => panic!("IO on invalid MaybeWriteOnly"),
         }
@@ -236,7 +237,7 @@ impl<IO: AsyncRead + AsyncWrite + Unpin> MaybeWriteOnly<IO> {
         match self {
             MaybeWriteOnly::Full(framed) => framed.read_message().await,
             MaybeWriteOnly::WriteOnly(_) => {
-                Err(io::Error::other("reading from write only half").into())
+                Err(io::Error::new(ErrorKind::Other, "reading from write only half").into())
             }
             MaybeWriteOnly::Broken => panic!("IO on invalid MaybeWriteOnly"),
         }
@@ -974,7 +975,7 @@ impl<IO: AsyncRead + AsyncWrite + Unpin> AsyncWrite for CopyDataWriter<'_, IO> {
             .write_message_noflush(&BeMessage::CopyData(buf))
             // write_message only writes to the buffer, so it can fail iff the
             // message is invaid, but CopyData can't be invalid.
-            .map_err(|_| io::Error::other("failed to serialize CopyData"))?;
+            .map_err(|_| io::Error::new(ErrorKind::Other, "failed to serialize CopyData"))?;
 
         Poll::Ready(Ok(buf.len()))
     }

libs/postgres_backend/tests/simple_select.rs

@@ -85,8 +85,8 @@ static KEY: Lazy<rustls::pki_types::PrivateKeyDer<'static>> = Lazy::new(|| {
 
 static CERT: Lazy<rustls::pki_types::CertificateDer<'static>> = Lazy::new(|| {
     let mut cursor = Cursor::new(include_bytes!("cert.pem"));
-
-    rustls_pemfile::certs(&mut cursor).next().unwrap().unwrap()
+    let cert = rustls_pemfile::certs(&mut cursor).next().unwrap().unwrap();
+    cert
 });
 
 // test that basic select with ssl works

libs/pq_proto/src/framed.rs

@@ -35,7 +35,7 @@ impl ConnectionError {
     pub fn into_io_error(self) -> io::Error {
         match self {
             ConnectionError::Io(io) => io,
-            ConnectionError::Protocol(pe) => io::Error::other(pe.to_string()),
+            ConnectionError::Protocol(pe) => io::Error::new(io::ErrorKind::Other, pe.to_string()),
         }
     }
 }

libs/pq_proto/src/lib.rs

@@ -257,7 +257,7 @@ pub enum ProtocolError {
 impl ProtocolError {
     /// Proxy stream.rs uses only io::Error; provide it.
     pub fn into_io_error(self) -> io::Error {
-        io::Error::other(self.to_string())
+        io::Error::new(io::ErrorKind::Other, self.to_string())
     }
 }
 

libs/proxy/postgres-protocol2/src/authentication/sasl.rs

@@ -212,7 +212,7 @@ impl ScramSha256 {
                     password,
                     channel_binding,
                 } => (nonce, password, channel_binding),
-                _ => return Err(io::Error::other("invalid SCRAM state")),
+                _ => return Err(io::Error::new(io::ErrorKind::Other, "invalid SCRAM state")),
             };
 
         let message =
@@ -291,7 +291,7 @@ impl ScramSha256 {
                 server_key,
                 auth_message,
             } => (server_key, auth_message),
-            _ => return Err(io::Error::other("invalid SCRAM state")),
+            _ => return Err(io::Error::new(io::ErrorKind::Other, "invalid SCRAM state")),
         };
 
         let message =
@@ -301,7 +301,10 @@ impl ScramSha256 {
 
         let verifier = match parsed {
             ServerFinalMessage::Error(e) => {
-                return Err(io::Error::other(format!("SCRAM error: {}", e)));
+                return Err(io::Error::new(
+                    io::ErrorKind::Other,
+                    format!("SCRAM error: {}", e),
+                ));
             }
             ServerFinalMessage::Verifier(verifier) => verifier,
         };

libs/remote_storage/Cargo.toml

@@ -28,7 +28,7 @@ toml_edit.workspace = true
 tracing.workspace = true
 scopeguard.workspace = true
 metrics.workspace = true
-utils = { path = "../utils", default-features = false }
+utils.workspace = true
 pin-project-lite.workspace = true
 
 azure_core.workspace = true

libs/remote_storage/src/azure_blob.rs

@@ -801,7 +801,8 @@ where
             // that support needs to be hacked in.
             //
             // including {self:?} into the message would be useful, but unsure how to unproject.
-            _ => std::task::Poll::Ready(Err(std::io::Error::other(
+            _ => std::task::Poll::Ready(Err(std::io::Error::new(
+                std::io::ErrorKind::Other,
                 "cloned or initial values cannot be read",
             ))),
         }
@@ -854,7 +855,7 @@ where
         };
         Err(azure_core::error::Error::new(
             azure_core::error::ErrorKind::Io,
-            std::io::Error::other(msg),
+            std::io::Error::new(std::io::ErrorKind::Other, msg),
         ))
     }
 

libs/utils/Cargo.toml

@@ -5,8 +5,7 @@ edition.workspace = true
 license.workspace = true
 
 [features]
-default = ["rename_noreplace"]
-rename_noreplace = []
+default = []
 # Enables test-only APIs, incuding failpoints. In particular, enables the `fail_point!` macro,
 # which adds some runtime cost to run tests on outage conditions
 testing = ["fail/failpoints"]
@@ -36,7 +35,7 @@ serde_with.workspace = true
 serde_json.workspace = true
 signal-hook.workspace = true
 thiserror.workspace = true
-tokio = { workspace = true, features = ["signal"] }
+tokio.workspace = true
 tokio-tar.workspace = true
 tokio-util.workspace = true
 toml_edit = { workspace = true, features = ["serde"] }

libs/utils/src/auth.rs

@@ -173,7 +173,7 @@ impl std::fmt::Debug for JwtAuth {
 }
 
 // this function is used only for testing purposes in CLI e g generate tokens during init
-pub fn encode_from_key_file<S: Serialize>(claims: &S, key_data: &[u8]) -> Result<String> {
+pub fn encode_from_key_file(claims: &Claims, key_data: &[u8]) -> Result<String> {
     let key = EncodingKey::from_ed_pem(key_data)?;
     Ok(encode(&Header::new(STORAGE_TOKEN_ALGORITHM), claims, &key)?)
 }

libs/utils/src/crashsafe.rs

@@ -81,9 +81,12 @@ pub fn path_with_suffix_extension(
 }
 
 pub fn fsync_file_and_parent(file_path: &Utf8Path) -> io::Result<()> {
-    let parent = file_path
-        .parent()
-        .ok_or_else(|| io::Error::other(format!("File {file_path:?} has no parent")))?;
+    let parent = file_path.parent().ok_or_else(|| {
+        io::Error::new(
+            io::ErrorKind::Other,
+            format!("File {file_path:?} has no parent"),
+        )
+    })?;
 
     fsync(file_path)?;
     fsync(parent)?;

libs/utils/src/fs_ext.rs

@@ -3,9 +3,7 @@ use std::{fs, io, path::Path};
 
 use anyhow::Context;
 
-#[cfg(feature = "rename_noreplace")]
 mod rename_noreplace;
-#[cfg(feature = "rename_noreplace")]
 pub use rename_noreplace::rename_noreplace;
 
 pub trait PathExt {

libs/utils/src/fs_ext/rename_noreplace.rs

@@ -8,7 +8,7 @@ pub fn rename_noreplace<P1: ?Sized + NixPath, P2: ?Sized + NixPath>(
     dst: &P2,
 ) -> nix::Result<()> {
     {
-        #[cfg(all(target_os = "linux", target_env = "gnu"))]
+        #[cfg(target_os = "linux")]
         {
             nix::fcntl::renameat2(
                 None,
@@ -29,7 +29,7 @@ pub fn rename_noreplace<P1: ?Sized + NixPath, P2: ?Sized + NixPath>(
             })??;
             nix::errno::Errno::result(res).map(drop)
         }
-        #[cfg(not(any(all(target_os = "linux", target_env = "gnu"), target_os = "macos")))]
+        #[cfg(not(any(target_os = "linux", target_os = "macos")))]
         {
             std::compile_error!("OS does not support no-replace renames");
         }

libs/utils/src/signals.rs

@@ -1,8 +1,6 @@
 pub use signal_hook::consts::TERM_SIGNALS;
 pub use signal_hook::consts::signal::*;
 use signal_hook::iterator::Signals;
-use tokio::signal::unix::{SignalKind, signal};
-use tracing::info;
 
 pub enum Signal {
     Quit,
@@ -38,30 +36,3 @@ impl ShutdownSignals {
         Ok(())
     }
 }
-
-/// Runs in a loop since we want to be responsive to multiple signals
-/// even after triggering shutdown (e.g. a SIGQUIT after a slow SIGTERM shutdown)
-/// <https://github.com/neondatabase/neon/issues/9740>
-pub async fn signal_handler(token: tokio_util::sync::CancellationToken) {
-    let mut sigint = signal(SignalKind::interrupt()).unwrap();
-    let mut sigterm = signal(SignalKind::terminate()).unwrap();
-    let mut sigquit = signal(SignalKind::quit()).unwrap();
-
-    loop {
-        let signal = tokio::select! {
-            _ = sigquit.recv() => {
-                info!("Got signal SIGQUIT. Terminating in immediate shutdown mode.");
-                std::process::exit(111);
-            }
-            _ = sigint.recv() => "SIGINT",
-            _ = sigterm.recv() => "SIGTERM",
-        };
-
-        if !token.is_cancelled() {
-            info!("Got signal {signal}. Terminating gracefully in fast shutdown mode.");
-            token.cancel();
-        } else {
-            info!("Got signal {signal}. Already shutting down.");
-        }
-    }
-}

object_storage/Cargo.toml

@@ -1,28 +0,0 @@
-[package]
-name = "object_storage"
-version = "0.0.1"
-edition.workspace = true
-license.workspace = true
-[dependencies]
-anyhow.workspace = true
-axum-extra.workspace = true
-axum.workspace = true
-camino.workspace = true
-futures.workspace = true
-jsonwebtoken.workspace = true
-prometheus.workspace = true
-remote_storage.workspace = true
-serde.workspace = true
-serde_json.workspace = true
-tokio-util.workspace = true
-tokio.workspace = true
-tracing.workspace = true
-utils = { path = "../libs/utils", default-features = false }
-workspace_hack.workspace = true
-[dev-dependencies]
-camino-tempfile.workspace = true
-http-body-util.workspace = true
-itertools.workspace = true
-rand.workspace = true
-test-log.workspace = true
-tower.workspace = true

object_storage/src/app.rs

@@ -1,561 +0,0 @@
-use anyhow::anyhow;
-use axum::body::{Body, Bytes};
-use axum::response::{IntoResponse, Response};
-use axum::{Router, http::StatusCode};
-use object_storage::{PrefixS3Path, S3Path, Storage, bad_request, internal_error, not_found, ok};
-use remote_storage::TimeoutOrCancel;
-use remote_storage::{DownloadError, DownloadOpts, GenericRemoteStorage, RemotePath};
-use std::{sync::Arc, time::SystemTime, time::UNIX_EPOCH};
-use tokio_util::sync::CancellationToken;
-use tracing::{error, info};
-use utils::backoff::retry;
-
-pub fn app(state: Arc<Storage>) -> Router<()> {
-    use axum::routing::{delete as _delete, get as _get};
-    let delete_prefix = _delete(delete_prefix);
-    Router::new()
-        .route(
-            "/{tenant_id}/{timeline_id}/{endpoint_id}/{*path}",
-            _get(get).put(set).delete(delete),
-        )
-        .route(
-            "/{tenant_id}/{timeline_id}/{endpoint_id}",
-            delete_prefix.clone(),
-        )
-        .route("/{tenant_id}/{timeline_id}", delete_prefix.clone())
-        .route("/{tenant_id}", delete_prefix)
-        .route("/metrics", _get(metrics))
-        .route("/status", _get(async || StatusCode::OK.into_response()))
-        .with_state(state)
-}
-
-type Result = anyhow::Result<Response, Response>;
-type State = axum::extract::State<Arc<Storage>>;
-
-const CONTENT_TYPE: &str = "content-type";
-const APPLICATION_OCTET_STREAM: &str = "application/octet-stream";
-const WARN_THRESHOLD: u32 = 3;
-const MAX_RETRIES: u32 = 10;
-
-async fn metrics() -> Result {
-    prometheus::TextEncoder::new()
-        .encode_to_string(&prometheus::gather())
-        .map(|s| s.into_response())
-        .map_err(|e| internal_error(e, "/metrics", "collecting metrics"))
-}
-
-async fn get(S3Path { path }: S3Path, state: State) -> Result {
-    info!(%path, "downloading");
-    let download_err = |e| {
-        if let DownloadError::NotFound = e {
-            info!(%path, %e, "downloading"); // 404 is not an issue of _this_ service
-            return not_found(&path);
-        }
-        internal_error(e, &path, "downloading")
-    };
-    let cancel = state.cancel.clone();
-    let opts = &DownloadOpts::default();
-
-    let stream = retry(
-        async || state.storage.download(&path, opts, &cancel).await,
-        DownloadError::is_permanent,
-        WARN_THRESHOLD,
-        MAX_RETRIES,
-        "downloading",
-        &cancel,
-    )
-    .await
-    .unwrap_or(Err(DownloadError::Cancelled))
-    .map_err(download_err)?
-    .download_stream;
-
-    Response::builder()
-        .status(StatusCode::OK)
-        .header(CONTENT_TYPE, APPLICATION_OCTET_STREAM)
-        .body(Body::from_stream(stream))
-        .map_err(|e| internal_error(e, path, "reading response"))
-}
-
-// Best solution for files is multipart upload, but remote_storage doesn't support it,
-// so we can either read Bytes in memory and push at once or forward BodyDataStream to
-// remote_storage. The latter may seem more peformant, but BodyDataStream doesn't have a
-// guaranteed size() which may produce issues while uploading to s3.
-// So, currently we're going with an in-memory copy plus a boundary to prevent uploading
-// very large files.
-async fn set(S3Path { path }: S3Path, state: State, bytes: Bytes) -> Result {
-    info!(%path, "uploading");
-    let request_len = bytes.len();
-    let max_len = state.max_upload_file_limit;
-    if request_len > max_len {
-        return Err(bad_request(
-            anyhow!("File size {request_len} exceeds max {max_len}"),
-            "uploading",
-        ));
-    }
-
-    let cancel = state.cancel.clone();
-    let fun = async || {
-        let stream = bytes_to_stream(bytes.clone());
-        state
-            .storage
-            .upload(stream, request_len, &path, None, &cancel)
-            .await
-    };
-    retry(
-        fun,
-        TimeoutOrCancel::caused_by_cancel,
-        WARN_THRESHOLD,
-        MAX_RETRIES,
-        "uploading",
-        &cancel,
-    )
-    .await
-    .unwrap_or(Err(anyhow!("uploading cancelled")))
-    .map_err(|e| internal_error(e, path, "reading response"))?;
-    Ok(ok())
-}
-
-async fn delete(S3Path { path }: S3Path, state: State) -> Result {
-    info!(%path, "deleting");
-    let cancel = state.cancel.clone();
-    retry(
-        async || state.storage.delete(&path, &cancel).await,
-        TimeoutOrCancel::caused_by_cancel,
-        WARN_THRESHOLD,
-        MAX_RETRIES,
-        "deleting",
-        &cancel,
-    )
-    .await
-    .unwrap_or(Err(anyhow!("deleting cancelled")))
-    .map_err(|e| internal_error(e, path, "deleting"))?;
-    Ok(ok())
-}
-
-async fn delete_prefix(PrefixS3Path { path }: PrefixS3Path, state: State) -> Result {
-    info!(%path, "deleting prefix");
-    let cancel = state.cancel.clone();
-    retry(
-        async || state.storage.delete_prefix(&path, &cancel).await,
-        TimeoutOrCancel::caused_by_cancel,
-        WARN_THRESHOLD,
-        MAX_RETRIES,
-        "deleting prefix",
-        &cancel,
-    )
-    .await
-    .unwrap_or(Err(anyhow!("deleting prefix cancelled")))
-    .map_err(|e| internal_error(e, path, "deleting prefix"))?;
-    Ok(ok())
-}
-
-pub async fn check_storage_permissions(
-    client: &GenericRemoteStorage,
-    cancel: CancellationToken,
-) -> anyhow::Result<()> {
-    info!("storage permissions check");
-
-    // as_nanos() as multiple instances proxying same bucket may be started at once
-    let now = SystemTime::now()
-        .duration_since(UNIX_EPOCH)?
-        .as_nanos()
-        .to_string();
-
-    let path = RemotePath::from_string(&format!("write_access_{now}"))?;
-    info!(%path, "uploading");
-
-    let body = now.to_string();
-    let stream = bytes_to_stream(Bytes::from(body.clone()));
-    client
-        .upload(stream, body.len(), &path, None, &cancel)
-        .await?;
-
-    use tokio::io::AsyncReadExt;
-    info!(%path, "downloading");
-    let download_opts = DownloadOpts {
-        kind: remote_storage::DownloadKind::Small,
-        ..Default::default()
-    };
-    let mut body_read_buf = Vec::new();
-    let stream = client
-        .download(&path, &download_opts, &cancel)
-        .await?
-        .download_stream;
-    tokio_util::io::StreamReader::new(stream)
-        .read_to_end(&mut body_read_buf)
-        .await?;
-    let body_read = String::from_utf8(body_read_buf)?;
-    if body != body_read {
-        error!(%body, %body_read, "File contents do not match");
-        anyhow::bail!("Read back file doesn't match original")
-    }
-
-    info!(%path, "removing");
-    client.delete(&path, &cancel).await
-}
-
-fn bytes_to_stream(bytes: Bytes) -> impl futures::Stream<Item = std::io::Result<Bytes>> {
-    futures::stream::once(futures::future::ready(Ok(bytes)))
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use axum::{body::Body, extract::Request, response::Response};
-    use http_body_util::BodyExt;
-    use itertools::iproduct;
-    use std::env::var;
-    use std::sync::Arc;
-    use std::time::Duration;
-    use test_log::test as testlog;
-    use tower::{Service, util::ServiceExt};
-    use utils::id::{TenantId, TimelineId};
-
-    // see libs/remote_storage/tests/test_real_s3.rs
-    const REAL_S3_ENV: &str = "ENABLE_REAL_S3_REMOTE_STORAGE";
-    const REAL_S3_BUCKET: &str = "REMOTE_STORAGE_S3_BUCKET";
-    const REAL_S3_REGION: &str = "REMOTE_STORAGE_S3_REGION";
-
-    async fn proxy() -> (Storage, Option<camino_tempfile::Utf8TempDir>) {
-        let cancel = CancellationToken::new();
-        let (dir, storage) = if var(REAL_S3_ENV).is_err() {
-            // tests execute in parallel and we need a new directory for each of them
-            let dir = camino_tempfile::tempdir().unwrap();
-            let fs =
-                remote_storage::LocalFs::new(dir.path().into(), Duration::from_secs(5)).unwrap();
-            (Some(dir), GenericRemoteStorage::LocalFs(fs))
-        } else {
-            // test_real_s3::create_s3_client is hard to reference, reimplementing here
-            let millis = SystemTime::now()
-                .duration_since(UNIX_EPOCH)
-                .unwrap()
-                .as_millis();
-            use rand::Rng;
-            let random = rand::thread_rng().r#gen::<u32>();
-
-            let s3_config = remote_storage::S3Config {
-                bucket_name: var(REAL_S3_BUCKET).unwrap(),
-                bucket_region: var(REAL_S3_REGION).unwrap(),
-                prefix_in_bucket: Some(format!("test_{millis}_{random:08x}/")),
-                endpoint: None,
-                concurrency_limit: std::num::NonZeroUsize::new(100).unwrap(),
-                max_keys_per_list_response: None,
-                upload_storage_class: None,
-            };
-            let bucket = remote_storage::S3Bucket::new(&s3_config, Duration::from_secs(1))
-                .await
-                .unwrap();
-            (None, GenericRemoteStorage::AwsS3(Arc::new(bucket)))
-        };
-
-        let proxy = Storage {
-            auth: object_storage::JwtAuth::new(TEST_PUB_KEY_ED25519).unwrap(),
-            storage,
-            cancel: cancel.clone(),
-            max_upload_file_limit: usize::MAX,
-        };
-        check_storage_permissions(&proxy.storage, cancel)
-            .await
-            .unwrap();
-        (proxy, dir)
-    }
-
-    // see libs/utils/src/auth.rs
-    const TEST_PUB_KEY_ED25519: &[u8] = b"
------BEGIN PUBLIC KEY-----
-MCowBQYDK2VwAyEARYwaNBayR+eGI0iXB4s3QxE3Nl2g1iWbr6KtLWeVD/w=
------END PUBLIC KEY-----
-";
-
-    const TEST_PRIV_KEY_ED25519: &[u8] = br#"
------BEGIN PRIVATE KEY-----
-MC4CAQAwBQYDK2VwBCIEID/Drmc1AA6U/znNRWpF3zEGegOATQxfkdWxitcOMsIH
------END PRIVATE KEY-----
-"#;
-
-    async fn request(req: Request<Body>) -> Response<Body> {
-        let (proxy, _) = proxy().await;
-        app(Arc::new(proxy))
-            .into_service()
-            .oneshot(req)
-            .await
-            .unwrap()
-    }
-
-    #[testlog(tokio::test)]
-    async fn status() {
-        let res = Request::builder()
-            .uri("/status")
-            .body(Body::empty())
-            .map(request)
-            .unwrap()
-            .await;
-        assert_eq!(res.status(), StatusCode::OK);
-    }
-
-    fn routes() -> impl Iterator<Item = (&'static str, &'static str)> {
-        iproduct!(
-            vec!["/1", "/1/2", "/1/2/3", "/1/2/3/4"],
-            vec!["GET", "PUT", "DELETE"]
-        )
-    }
-
-    #[testlog(tokio::test)]
-    async fn no_token() {
-        for (uri, method) in routes() {
-            info!(%uri, %method);
-            let res = Request::builder()
-                .uri(uri)
-                .method(method)
-                .body(Body::empty())
-                .map(request)
-                .unwrap()
-                .await;
-            assert!(matches!(
-                res.status(),
-                StatusCode::METHOD_NOT_ALLOWED | StatusCode::BAD_REQUEST
-            ));
-        }
-    }
-
-    #[testlog(tokio::test)]
-    async fn invalid_token() {
-        for (uri, method) in routes() {
-            info!(%uri, %method);
-            let status = Request::builder()
-                .uri(uri)
-                .header("Authorization", "Bearer 123")
-                .method(method)
-                .body(Body::empty())
-                .map(request)
-                .unwrap()
-                .await;
-            assert!(matches!(
-                status.status(),
-                StatusCode::METHOD_NOT_ALLOWED | StatusCode::BAD_REQUEST
-            ));
-        }
-    }
-
-    const TENANT_ID: TenantId =
-        TenantId::from_array([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 1, 2, 3, 4, 5, 6]);
-    const TIMELINE_ID: TimelineId =
-        TimelineId::from_array([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 1, 2, 3, 4, 5, 7]);
-    const ENDPOINT_ID: &str = "ep-winter-frost-a662z3vg";
-    fn token() -> String {
-        let claims = object_storage::Claims {
-            tenant_id: TENANT_ID,
-            timeline_id: TIMELINE_ID,
-            endpoint_id: ENDPOINT_ID.into(),
-            exp: u64::MAX,
-        };
-        let key = jsonwebtoken::EncodingKey::from_ed_pem(TEST_PRIV_KEY_ED25519).unwrap();
-        let header = jsonwebtoken::Header::new(object_storage::VALIDATION_ALGO);
-        jsonwebtoken::encode(&header, &claims, &key).unwrap()
-    }
-
-    #[testlog(tokio::test)]
-    async fn unauthorized() {
-        let (proxy, _) = proxy().await;
-        let mut app = app(Arc::new(proxy)).into_service();
-        let token = token();
-        let args = itertools::iproduct!(
-            vec![TENANT_ID.to_string(), TenantId::generate().to_string()],
-            vec![TIMELINE_ID.to_string(), TimelineId::generate().to_string()],
-            vec![ENDPOINT_ID, "ep-ololo"]
-        )
-        .skip(1);
-
-        for ((uri, method), (tenant, timeline, endpoint)) in iproduct!(routes(), args) {
-            info!(%uri, %method, %tenant, %timeline, %endpoint);
-            let request = Request::builder()
-                .uri(format!("/{tenant}/{timeline}/{endpoint}/sub/path/key"))
-                .method(method)
-                .header("Authorization", format!("Bearer {}", token))
-                .body(Body::empty())
-                .unwrap();
-            let status = ServiceExt::ready(&mut app)
-                .await
-                .unwrap()
-                .call(request)
-                .await
-                .unwrap()
-                .status();
-            assert_eq!(status, StatusCode::UNAUTHORIZED);
-        }
-    }
-
-    #[testlog(tokio::test)]
-    async fn method_not_allowed() {
-        let token = token();
-        let iter = iproduct!(vec!["", "/.."], vec!["GET", "PUT"]);
-        for (key, method) in iter {
-            let status = Request::builder()
-                .uri(format!("/{TENANT_ID}/{TIMELINE_ID}/{ENDPOINT_ID}{key}"))
-                .method(method)
-                .header("Authorization", format!("Bearer {token}"))
-                .body(Body::empty())
-                .map(request)
-                .unwrap()
-                .await
-                .status();
-            assert!(matches!(
-                status,
-                StatusCode::BAD_REQUEST | StatusCode::METHOD_NOT_ALLOWED
-            ));
-        }
-    }
-
-    async fn requests_chain(
-        chain: impl Iterator<Item = (String, &str, &'static str, StatusCode, bool)>,
-        token: impl Fn(&str) -> String,
-    ) {
-        let (proxy, _) = proxy().await;
-        let mut app = app(Arc::new(proxy)).into_service();
-        for (uri, method, body, expected_status, compare_body) in chain {
-            info!(%uri, %method, %body, %expected_status);
-            let bearer = format!("Bearer {}", token(&uri));
-            let request = Request::builder()
-                .uri(uri)
-                .method(method)
-                .header("Authorization", &bearer)
-                .body(Body::from(body))
-                .unwrap();
-            let response = ServiceExt::ready(&mut app)
-                .await
-                .unwrap()
-                .call(request)
-                .await
-                .unwrap();
-            assert_eq!(response.status(), expected_status);
-            if !compare_body {
-                continue;
-            }
-            let read_body = response.into_body().collect().await.unwrap().to_bytes();
-            assert_eq!(body, read_body);
-        }
-    }
-
-    #[testlog(tokio::test)]
-    async fn metrics() {
-        let uri = format!("/{TENANT_ID}/{TIMELINE_ID}/{ENDPOINT_ID}/key");
-        let req = vec![
-            (uri.clone(), "PUT", "body", StatusCode::OK, false),
-            (uri.clone(), "DELETE", "", StatusCode::OK, false),
-        ];
-        requests_chain(req.into_iter(), |_| token()).await;
-
-        let res = Request::builder()
-            .uri("/metrics")
-            .body(Body::empty())
-            .map(request)
-            .unwrap()
-            .await;
-        assert_eq!(res.status(), StatusCode::OK);
-        let body = res.into_body().collect().await.unwrap().to_bytes();
-        let body = String::from_utf8_lossy(&body);
-        tracing::debug!(%body);
-        // Storage metrics are not gathered for LocalFs
-        if var(REAL_S3_ENV).is_ok() {
-            assert!(body.contains("remote_storage_s3_deleted_objects_total"));
-        }
-        assert!(body.contains("process_threads"));
-    }
-
-    #[testlog(tokio::test)]
-    async fn insert_retrieve_remove() {
-        let uri = format!("/{TENANT_ID}/{TIMELINE_ID}/{ENDPOINT_ID}/key");
-        let chain = vec![
-            (uri.clone(), "GET", "", StatusCode::NOT_FOUND, false),
-            (uri.clone(), "PUT", "пыщьпыщь", StatusCode::OK, false),
-            (uri.clone(), "GET", "пыщьпыщь", StatusCode::OK, true),
-            (uri.clone(), "DELETE", "", StatusCode::OK, false),
-            (uri, "GET", "", StatusCode::NOT_FOUND, false),
-        ];
-        requests_chain(chain.into_iter(), |_| token()).await;
-    }
-
-    fn delete_prefix_token(uri: &str) -> String {
-        use serde::Serialize;
-        let parts = uri.split("/").collect::<Vec<&str>>();
-        #[derive(Serialize)]
-        struct PrefixClaims {
-            tenant_id: TenantId,
-            timeline_id: Option<TimelineId>,
-            endpoint_id: Option<object_storage::EndpointId>,
-            exp: u64,
-        }
-        let claims = PrefixClaims {
-            tenant_id: parts.get(1).map(|c| c.parse().unwrap()).unwrap(),
-            timeline_id: parts.get(2).map(|c| c.parse().unwrap()),
-            endpoint_id: parts.get(3).map(ToString::to_string),
-            exp: u64::MAX,
-        };
-        let key = jsonwebtoken::EncodingKey::from_ed_pem(TEST_PRIV_KEY_ED25519).unwrap();
-        let header = jsonwebtoken::Header::new(object_storage::VALIDATION_ALGO);
-        jsonwebtoken::encode(&header, &claims, &key).unwrap()
-    }
-
-    // Can't use single digit numbers as they won't be validated as TimelineId and EndpointId
-    #[testlog(tokio::test)]
-    async fn delete_prefix() {
-        let tenant_id =
-            TenantId::from_array([1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]).to_string();
-        let t2 = TimelineId::from_array([2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]);
-        let t3 = TimelineId::from_array([3, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]);
-        let t4 = TimelineId::from_array([4, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]);
-        let f = |timeline, path| format!("/{tenant_id}/{timeline}{path}");
-        // Why extra slash in string literals? Axum is weird with URIs:
-        // /1/2 and 1/2/ match different routes, thus first yields OK and second NOT_FOUND
-        //  as it matches /tenant/timeline/endpoint, see https://stackoverflow.com/a/75355932
-        // The cost of removing trailing slash is suprisingly hard:
-        // * Add tower dependency with NormalizePath layer
-        // * wrap Router<()> in this layer https://github.com/tokio-rs/axum/discussions/2377
-        // * Rewrite make_service() -> into_make_service()
-        // * Rewrite oneshot() (not available for NormalizePath)
-        // I didn't manage to get it working correctly
-        let chain = vec![
-            // create 1/2/3/4, 1/2/3/5, delete prefix 1/2/3 -> empty
-            (f(t2, "/3/4"), "PUT", "", StatusCode::OK, false),
-            (f(t2, "/3/4"), "PUT", "", StatusCode::OK, false), // we can override file contents
-            (f(t2, "/3/5"), "PUT", "", StatusCode::OK, false),
-            (f(t2, "/3"), "DELETE", "", StatusCode::OK, false),
-            (f(t2, "/3/4"), "GET", "", StatusCode::NOT_FOUND, false),
-            (f(t2, "/3/5"), "GET", "", StatusCode::NOT_FOUND, false),
-            // create 1/2/3/4, 1/2/5/6, delete prefix 1/2/3 -> 1/2/5/6
-            (f(t2, "/3/4"), "PUT", "", StatusCode::OK, false),
-            (f(t2, "/5/6"), "PUT", "", StatusCode::OK, false),
-            (f(t2, "/3"), "DELETE", "", StatusCode::OK, false),
-            (f(t2, "/3/4"), "GET", "", StatusCode::NOT_FOUND, false),
-            (f(t2, "/5/6"), "GET", "", StatusCode::OK, false),
-            // create 1/2/3/4, 1/2/7/8, delete prefix 1/2 -> empty
-            (f(t2, "/3/4"), "PUT", "", StatusCode::OK, false),
-            (f(t2, "/7/8"), "PUT", "", StatusCode::OK, false),
-            (f(t2, ""), "DELETE", "", StatusCode::OK, false),
-            (f(t2, "/3/4"), "GET", "", StatusCode::NOT_FOUND, false),
-            (f(t2, "/7/8"), "GET", "", StatusCode::NOT_FOUND, false),
-            // create 1/2/3/4, 1/2/5/6, 1/3/8/9, delete prefix 1/2/3 -> 1/2/5/6, 1/3/8/9
-            (f(t2, "/3/4"), "PUT", "", StatusCode::OK, false),
-            (f(t2, "/5/6"), "PUT", "", StatusCode::OK, false),
-            (f(t3, "/8/9"), "PUT", "", StatusCode::OK, false),
-            (f(t2, "/3"), "DELETE", "", StatusCode::OK, false),
-            (f(t2, "/3/4"), "GET", "", StatusCode::NOT_FOUND, false),
-            (f(t2, "/5/6"), "GET", "", StatusCode::OK, false),
-            (f(t3, "/8/9"), "GET", "", StatusCode::OK, false),
-            // create 1/4/5/6, delete prefix 1/2 -> 1/3/8/9, 1/4/5/6
-            (f(t4, "/5/6"), "PUT", "", StatusCode::OK, false),
-            (f(t2, ""), "DELETE", "", StatusCode::OK, false),
-            (f(t2, "/3/4"), "GET", "", StatusCode::NOT_FOUND, false),
-            (f(t2, "/5/6"), "GET", "", StatusCode::NOT_FOUND, false),
-            (f(t3, "/8/9"), "GET", "", StatusCode::OK, false),
-            (f(t4, "/5/6"), "GET", "", StatusCode::OK, false),
-            // delete prefix 1 -> empty
-            (format!("/{tenant_id}"), "DELETE", "", StatusCode::OK, false),
-            (f(t2, "/3/4"), "GET", "", StatusCode::NOT_FOUND, false),
-            (f(t2, "/5/6"), "GET", "", StatusCode::NOT_FOUND, false),
-            (f(t3, "/8/9"), "GET", "", StatusCode::NOT_FOUND, false),
-            (f(t4, "/5/6"), "GET", "", StatusCode::NOT_FOUND, false),
-        ];
-        requests_chain(chain.into_iter(), delete_prefix_token).await;
-    }
-}

object_storage/src/lib.rs

@@ -1,344 +0,0 @@
-use anyhow::Result;
-use axum::extract::{FromRequestParts, Path};
-use axum::response::{IntoResponse, Response};
-use axum::{RequestPartsExt, http::StatusCode, http::request::Parts};
-use axum_extra::TypedHeader;
-use axum_extra::headers::{Authorization, authorization::Bearer};
-use camino::Utf8PathBuf;
-use jsonwebtoken::{DecodingKey, Validation};
-use remote_storage::{GenericRemoteStorage, RemotePath};
-use serde::{Deserialize, Serialize};
-use std::fmt::Display;
-use std::result::Result as StdResult;
-use std::sync::Arc;
-use tokio_util::sync::CancellationToken;
-use tracing::{debug, error};
-use utils::id::{TenantId, TimelineId};
-
-// simplified version of utils::auth::JwtAuth
-pub struct JwtAuth {
-    decoding_key: DecodingKey,
-    validation: Validation,
-}
-
-pub const VALIDATION_ALGO: jsonwebtoken::Algorithm = jsonwebtoken::Algorithm::EdDSA;
-impl JwtAuth {
-    pub fn new(key: &[u8]) -> Result<Self> {
-        Ok(Self {
-            decoding_key: DecodingKey::from_ed_pem(key)?,
-            validation: Validation::new(VALIDATION_ALGO),
-        })
-    }
-
-    pub fn decode<T: serde::de::DeserializeOwned>(&self, token: &str) -> Result<T> {
-        Ok(jsonwebtoken::decode(token, &self.decoding_key, &self.validation).map(|t| t.claims)?)
-    }
-}
-
-fn normalize_key(key: &str) -> StdResult<Utf8PathBuf, String> {
-    let key = clean_utf8(&Utf8PathBuf::from(key));
-    if key.starts_with("..") || key == "." || key == "/" {
-        return Err(format!("invalid key {key}"));
-    }
-    match key.strip_prefix("/").map(Utf8PathBuf::from) {
-        Ok(p) => Ok(p),
-        _ => Ok(key),
-    }
-}
-
-// Copied from path_clean crate with PathBuf->Utf8PathBuf
-fn clean_utf8(path: &camino::Utf8Path) -> Utf8PathBuf {
-    use camino::Utf8Component as Comp;
-    let mut out = Vec::new();
-    for comp in path.components() {
-        match comp {
-            Comp::CurDir => (),
-            Comp::ParentDir => match out.last() {
-                Some(Comp::RootDir) => (),
-                Some(Comp::Normal(_)) => {
-                    out.pop();
-                }
-                None | Some(Comp::CurDir) | Some(Comp::ParentDir) | Some(Comp::Prefix(_)) => {
-                    out.push(comp)
-                }
-            },
-            comp => out.push(comp),
-        }
-    }
-    if !out.is_empty() {
-        out.iter().collect()
-    } else {
-        Utf8PathBuf::from(".")
-    }
-}
-
-pub struct Storage {
-    pub auth: JwtAuth,
-    pub storage: GenericRemoteStorage,
-    pub cancel: CancellationToken,
-    pub max_upload_file_limit: usize,
-}
-
-pub type EndpointId = String; // If needed, reuse small string from proxy/src/types.rc
-
-#[derive(Deserialize, Serialize, PartialEq)]
-pub struct Claims {
-    pub tenant_id: TenantId,
-    pub timeline_id: TimelineId,
-    pub endpoint_id: EndpointId,
-    pub exp: u64,
-}
-
-impl Display for Claims {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(
-            f,
-            "Claims(tenant_id {} timeline_id {} endpoint_id {} exp {})",
-            self.tenant_id, self.timeline_id, self.endpoint_id, self.exp
-        )
-    }
-}
-
-#[derive(Deserialize, Serialize)]
-struct KeyRequest {
-    tenant_id: TenantId,
-    timeline_id: TimelineId,
-    endpoint_id: EndpointId,
-    path: String,
-}
-
-#[derive(Debug, PartialEq)]
-pub struct S3Path {
-    pub path: RemotePath,
-}
-
-impl TryFrom<&KeyRequest> for S3Path {
-    type Error = String;
-    fn try_from(req: &KeyRequest) -> StdResult<Self, Self::Error> {
-        let KeyRequest {
-            tenant_id,
-            timeline_id,
-            endpoint_id,
-            path,
-        } = &req;
-        let prefix = format!("{tenant_id}/{timeline_id}/{endpoint_id}",);
-        let path = Utf8PathBuf::from(prefix).join(normalize_key(path)?);
-        let path = RemotePath::new(&path).unwrap(); // unwrap() because the path is already relative
-        Ok(S3Path { path })
-    }
-}
-
-fn unauthorized(route: impl Display, claims: impl Display) -> Response {
-    debug!(%route, %claims, "route doesn't match claims");
-    StatusCode::UNAUTHORIZED.into_response()
-}
-
-pub fn bad_request(err: impl Display, desc: &'static str) -> Response {
-    debug!(%err, desc);
-    (StatusCode::BAD_REQUEST, err.to_string()).into_response()
-}
-
-pub fn ok() -> Response {
-    StatusCode::OK.into_response()
-}
-
-pub fn internal_error(err: impl Display, path: impl Display, desc: &'static str) -> Response {
-    error!(%err, %path, desc);
-    StatusCode::INTERNAL_SERVER_ERROR.into_response()
-}
-
-pub fn not_found(key: impl ToString) -> Response {
-    (StatusCode::NOT_FOUND, key.to_string()).into_response()
-}
-
-impl FromRequestParts<Arc<Storage>> for S3Path {
-    type Rejection = Response;
-    async fn from_request_parts(
-        parts: &mut Parts,
-        state: &Arc<Storage>,
-    ) -> Result<Self, Self::Rejection> {
-        let Path(path): Path<KeyRequest> = parts
-            .extract()
-            .await
-            .map_err(|e| bad_request(e, "invalid route"))?;
-        let TypedHeader(Authorization(bearer)) = parts
-            .extract::<TypedHeader<Authorization<Bearer>>>()
-            .await
-            .map_err(|e| bad_request(e, "invalid token"))?;
-        let claims: Claims = state
-            .auth
-            .decode(bearer.token())
-            .map_err(|e| bad_request(e, "decoding token"))?;
-        let route = Claims {
-            tenant_id: path.tenant_id,
-            timeline_id: path.timeline_id,
-            endpoint_id: path.endpoint_id.clone(),
-            exp: claims.exp,
-        };
-        if route != claims {
-            return Err(unauthorized(route, claims));
-        }
-        (&path)
-            .try_into()
-            .map_err(|e| bad_request(e, "invalid route"))
-    }
-}
-
-#[derive(Deserialize, Serialize, PartialEq)]
-pub struct PrefixKeyPath {
-    pub tenant_id: TenantId,
-    pub timeline_id: Option<TimelineId>,
-    pub endpoint_id: Option<EndpointId>,
-}
-
-impl Display for PrefixKeyPath {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(
-            f,
-            "PrefixKeyPath(tenant_id {} timeline_id {} endpoint_id {})",
-            self.tenant_id,
-            self.timeline_id
-                .as_ref()
-                .map(ToString::to_string)
-                .unwrap_or("".to_string()),
-            self.endpoint_id
-                .as_ref()
-                .map(ToString::to_string)
-                .unwrap_or("".to_string())
-        )
-    }
-}
-
-#[derive(Debug, PartialEq)]
-pub struct PrefixS3Path {
-    pub path: RemotePath,
-}
-
-impl From<&PrefixKeyPath> for PrefixS3Path {
-    fn from(path: &PrefixKeyPath) -> Self {
-        let timeline_id = path
-            .timeline_id
-            .as_ref()
-            .map(ToString::to_string)
-            .unwrap_or("".to_string());
-        let endpoint_id = path
-            .endpoint_id
-            .as_ref()
-            .map(ToString::to_string)
-            .unwrap_or("".to_string());
-        let path = Utf8PathBuf::from(path.tenant_id.to_string())
-            .join(timeline_id)
-            .join(endpoint_id);
-        let path = RemotePath::new(&path).unwrap(); // unwrap() because the path is already relative
-        PrefixS3Path { path }
-    }
-}
-
-impl FromRequestParts<Arc<Storage>> for PrefixS3Path {
-    type Rejection = Response;
-    async fn from_request_parts(
-        parts: &mut Parts,
-        state: &Arc<Storage>,
-    ) -> Result<Self, Self::Rejection> {
-        let Path(path) = parts
-            .extract::<Path<PrefixKeyPath>>()
-            .await
-            .map_err(|e| bad_request(e, "invalid route"))?;
-        let TypedHeader(Authorization(bearer)) = parts
-            .extract::<TypedHeader<Authorization<Bearer>>>()
-            .await
-            .map_err(|e| bad_request(e, "invalid token"))?;
-        let claims: PrefixKeyPath = state
-            .auth
-            .decode(bearer.token())
-            .map_err(|e| bad_request(e, "invalid token"))?;
-        if path != claims {
-            return Err(unauthorized(path, claims));
-        }
-        Ok((&path).into())
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn normalize_key() {
-        let f = super::normalize_key;
-        assert_eq!(f("hello/world/..").unwrap(), Utf8PathBuf::from("hello"));
-        assert_eq!(
-            f("ololo/1/../../not_ololo").unwrap(),
-            Utf8PathBuf::from("not_ololo")
-        );
-        assert!(f("ololo/1/../../../").is_err());
-        assert!(f(".").is_err());
-        assert!(f("../").is_err());
-        assert!(f("").is_err());
-        assert_eq!(f("/1/2/3").unwrap(), Utf8PathBuf::from("1/2/3"));
-        assert!(f("/1/2/3/../../../").is_err());
-        assert!(f("/1/2/3/../../../../").is_err());
-    }
-
-    const TENANT_ID: TenantId =
-        TenantId::from_array([1, 1, 2, 3, 4, 5, 6, 7, 8, 9, 1, 2, 3, 4, 5, 6]);
-    const TIMELINE_ID: TimelineId =
-        TimelineId::from_array([1, 1, 2, 3, 4, 5, 6, 7, 8, 9, 1, 2, 3, 4, 5, 7]);
-    const ENDPOINT_ID: &str = "ep-winter-frost-a662z3vg";
-
-    #[test]
-    fn s3_path() {
-        let auth = Claims {
-            tenant_id: TENANT_ID,
-            timeline_id: TIMELINE_ID,
-            endpoint_id: ENDPOINT_ID.into(),
-            exp: u64::MAX,
-        };
-        let s3_path = |key| {
-            let path = &format!("{TENANT_ID}/{TIMELINE_ID}/{ENDPOINT_ID}/{key}");
-            let path = RemotePath::from_string(path).unwrap();
-            S3Path { path }
-        };
-
-        let path = "cache_key".to_string();
-        let mut key_path = KeyRequest {
-            path,
-            tenant_id: auth.tenant_id,
-            timeline_id: auth.timeline_id,
-            endpoint_id: auth.endpoint_id,
-        };
-        assert_eq!(S3Path::try_from(&key_path).unwrap(), s3_path(key_path.path));
-
-        key_path.path = "we/can/have/nested/paths".to_string();
-        assert_eq!(S3Path::try_from(&key_path).unwrap(), s3_path(key_path.path));
-
-        key_path.path = "../error/hello/../".to_string();
-        assert!(S3Path::try_from(&key_path).is_err());
-    }
-
-    #[test]
-    fn prefix_s3_path() {
-        let mut path = PrefixKeyPath {
-            tenant_id: TENANT_ID,
-            timeline_id: None,
-            endpoint_id: None,
-        };
-        let prefix_path = |s: String| RemotePath::from_string(&s).unwrap();
-        assert_eq!(
-            PrefixS3Path::from(&path).path,
-            prefix_path(format!("{TENANT_ID}"))
-        );
-
-        path.timeline_id = Some(TIMELINE_ID);
-        assert_eq!(
-            PrefixS3Path::from(&path).path,
-            prefix_path(format!("{TENANT_ID}/{TIMELINE_ID}"))
-        );
-
-        path.endpoint_id = Some(ENDPOINT_ID.into());
-        assert_eq!(
-            PrefixS3Path::from(&path).path,
-            prefix_path(format!("{TENANT_ID}/{TIMELINE_ID}/{ENDPOINT_ID}"))
-        );
-    }
-}

object_storage/src/main.rs

@@ -1,65 +0,0 @@
-//! `object_storage` is a service which provides API for uploading and downloading
-//! files. It is used by compute and control plane for accessing LFC prewarm data.
-//! This service is deployed either as a separate component or as part of compute image
-//! for large computes.
-mod app;
-use anyhow::Context;
-use tracing::info;
-use utils::logging;
-
-//see set()
-const fn max_upload_file_limit() -> usize {
-    100 * 1024 * 1024
-}
-
-#[derive(serde::Deserialize)]
-#[serde(tag = "type")]
-struct Config {
-    listen: std::net::SocketAddr,
-    pemfile: camino::Utf8PathBuf,
-    #[serde(flatten)]
-    storage_config: remote_storage::RemoteStorageConfig,
-    #[serde(default = "max_upload_file_limit")]
-    max_upload_file_limit: usize,
-}
-
-#[tokio::main]
-async fn main() -> anyhow::Result<()> {
-    logging::init(
-        logging::LogFormat::Plain,
-        logging::TracingErrorLayerEnablement::EnableWithRustLogFilter,
-        logging::Output::Stdout,
-    )?;
-
-    let config: String = std::env::args().skip(1).take(1).collect();
-    if config.is_empty() {
-        anyhow::bail!("Usage: object_storage config.json")
-    }
-    info!("Reading config from {config}");
-    let config = std::fs::read_to_string(config.clone())?;
-    let config: Config = serde_json::from_str(&config).context("parsing config")?;
-    info!("Reading pemfile from {}", config.pemfile.clone());
-    let pemfile = std::fs::read(config.pemfile.clone())?;
-    info!("Loading public key from {}", config.pemfile.clone());
-    let auth = object_storage::JwtAuth::new(&pemfile)?;
-
-    let listener = tokio::net::TcpListener::bind(config.listen).await.unwrap();
-    info!("listening on {}", listener.local_addr().unwrap());
-
-    let storage = remote_storage::GenericRemoteStorage::from_config(&config.storage_config).await?;
-    let cancel = tokio_util::sync::CancellationToken::new();
-    app::check_storage_permissions(&storage, cancel.clone()).await?;
-
-    let proxy = std::sync::Arc::new(object_storage::Storage {
-        auth,
-        storage,
-        cancel: cancel.clone(),
-        max_upload_file_limit: config.max_upload_file_limit,
-    });
-
-    tokio::spawn(utils::signals::signal_handler(cancel.clone()));
-    axum::serve(listener, app::app(proxy))
-        .with_graceful_shutdown(async move { cancel.cancelled().await })
-        .await?;
-    Ok(())
-}

pageserver/benches/bench_walredo.rs

@@ -65,7 +65,7 @@ use bytes::{Buf, Bytes};
 use criterion::{BenchmarkId, Criterion};
 use once_cell::sync::Lazy;
 use pageserver::config::PageServerConf;
-use pageserver::walredo::{PostgresRedoManager, RedoAttemptType};
+use pageserver::walredo::PostgresRedoManager;
 use pageserver_api::key::Key;
 use pageserver_api::record::NeonWalRecord;
 use pageserver_api::shard::TenantShardId;
@@ -223,14 +223,7 @@ impl Request {
 
         // TODO: avoid these clones
         manager
-            .request_redo(
-                *key,
-                *lsn,
-                base_img.clone(),
-                records.clone(),
-                *pg_version,
-                RedoAttemptType::ReadPage,
-            )
+            .request_redo(*key, *lsn, base_img.clone(), records.clone(), *pg_version)
             .await
             .context("request_redo")
     }

pageserver/src/bin/pageserver.rs

@@ -31,6 +31,7 @@ use pageserver::{
 };
 use postgres_backend::AuthType;
 use remote_storage::GenericRemoteStorage;
+use tokio::signal::unix::SignalKind;
 use tokio::time::Instant;
 use tokio_util::sync::CancellationToken;
 use tracing::*;
@@ -452,24 +453,6 @@ fn start_pageserver(
     info!("Using auth for http API: {:#?}", conf.http_auth_type);
     info!("Using auth for pg connections: {:#?}", conf.pg_auth_type);
 
-    let tls_server_config = if conf.listen_https_addr.is_some() || conf.enable_tls_page_service_api
-    {
-        let resolver = BACKGROUND_RUNTIME.block_on(ReloadingCertificateResolver::new(
-            "main",
-            &conf.ssl_key_file,
-            &conf.ssl_cert_file,
-            conf.ssl_cert_reload_period,
-        ))?;
-
-        let server_config = rustls::ServerConfig::builder()
-            .with_no_client_auth()
-            .with_cert_resolver(resolver);
-
-        Some(Arc::new(server_config))
-    } else {
-        None
-    };
-
     match var("NEON_AUTH_TOKEN") {
         Ok(v) => {
             info!("Loaded JWT token for authentication with Safekeeper");
@@ -688,11 +671,17 @@ fn start_pageserver(
 
         let https_task = match https_listener {
             Some(https_listener) => {
-                let tls_server_config = tls_server_config
-                    .clone()
-                    .expect("tls_server_config is set earlier if https is enabled");
+                let resolver = MGMT_REQUEST_RUNTIME.block_on(ReloadingCertificateResolver::new(
+                    &conf.ssl_key_file,
+                    &conf.ssl_cert_file,
+                    conf.ssl_cert_reload_period,
+                ))?;
 
-                let tls_acceptor = tokio_rustls::TlsAcceptor::from(tls_server_config);
+                let server_config = rustls::ServerConfig::builder()
+                    .with_no_client_auth()
+                    .with_cert_resolver(resolver);
+
+                let tls_acceptor = tokio_rustls::TlsAcceptor::from(Arc::new(server_config));
 
                 let server =
                     http_utils::server::Server::new(service, https_listener, Some(tls_acceptor))?;
@@ -748,11 +737,6 @@ fn start_pageserver(
             tokio::net::TcpListener::from_std(pageserver_listener)
                 .context("create tokio listener")?
         },
-        if conf.enable_tls_page_service_api {
-            tls_server_config
-        } else {
-            None
-        },
     );
 
     // All started up! Now just sit and wait for shutdown signal.
@@ -760,7 +744,32 @@ fn start_pageserver(
         let signal_token = CancellationToken::new();
         let signal_cancel = signal_token.child_token();
 
-        tokio::spawn(utils::signals::signal_handler(signal_token));
+        // Spawn signal handlers. Runs in a loop since we want to be responsive to multiple signals
+        // even after triggering shutdown (e.g. a SIGQUIT after a slow SIGTERM shutdown). See:
+        // https://github.com/neondatabase/neon/issues/9740.
+        tokio::spawn(async move {
+            let mut sigint = tokio::signal::unix::signal(SignalKind::interrupt()).unwrap();
+            let mut sigterm = tokio::signal::unix::signal(SignalKind::terminate()).unwrap();
+            let mut sigquit = tokio::signal::unix::signal(SignalKind::quit()).unwrap();
+
+            loop {
+                let signal = tokio::select! {
+                    _ = sigquit.recv() => {
+                        info!("Got signal SIGQUIT. Terminating in immediate shutdown mode.");
+                        std::process::exit(111);
+                    }
+                    _ = sigint.recv() => "SIGINT",
+                    _ = sigterm.recv() => "SIGTERM",
+                };
+
+                if !signal_token.is_cancelled() {
+                    info!("Got signal {signal}. Terminating gracefully in fast shutdown mode.");
+                    signal_token.cancel();
+                } else {
+                    info!("Got signal {signal}. Already shutting down.");
+                }
+            }
+        });
 
         // Wait for cancellation signal and shut down the pageserver.
         //

pageserver/src/config.rs

@@ -219,11 +219,6 @@ pub struct PageServerConf {
     pub generate_unarchival_heatmap: bool,
 
     pub tracing: Option<pageserver_api::config::Tracing>,
-
-    /// Enable TLS in page service API.
-    /// Does not force TLS: the client negotiates TLS usage during the handshake.
-    /// Uses key and certificate from ssl_key_file/ssl_cert_file.
-    pub enable_tls_page_service_api: bool,
 }
 
 /// Token for authentication to safekeepers
@@ -396,7 +391,6 @@ impl PageServerConf {
             load_previous_heatmap,
             generate_unarchival_heatmap,
             tracing,
-            enable_tls_page_service_api,
         } = config_toml;
 
         let mut conf = PageServerConf {
@@ -447,7 +441,6 @@ impl PageServerConf {
             page_service_pipelining,
             get_vectored_concurrent_io,
             tracing,
-            enable_tls_page_service_api,
 
             // ------------------------------------------------------------
             // fields that require additional validation or custom handling

pageserver/src/http/openapi_spec.yml

@@ -212,12 +212,6 @@ paths:
               schema:
                 type: string
                 format: date-time
-        "412":
-          description: No timestamp is found for given LSN, e.g. if there had been no commits till LSN
-          content:
-            application/json:
-              schema:
-                $ref: "#/components/schemas/PreconditionFailedError"
 
   /v1/tenant/{tenant_id}/timeline/{timeline_id}/get_lsn_by_timestamp:
     parameters:

pageserver/src/http/routes.rs

@@ -67,7 +67,7 @@ use crate::tenant::mgr::{
 };
 use crate::tenant::remote_timeline_client::index::GcCompactionState;
 use crate::tenant::remote_timeline_client::{
-    download_index_part, download_tenant_manifest, list_remote_tenant_shards, list_remote_timelines,
+    download_index_part, list_remote_tenant_shards, list_remote_timelines,
 };
 use crate::tenant::secondary::SecondaryController;
 use crate::tenant::size::ModelInputs;
@@ -989,7 +989,7 @@ async fn get_lsn_by_timestamp_handler(
     if !tenant_shard_id.is_shard_zero() {
         // Requires SLRU contents, which are only stored on shard zero
         return Err(ApiError::BadRequest(anyhow!(
-            "Lsn calculations by timestamp are only available on shard zero"
+            "Size calculations are only available on shard zero"
         )));
     }
 
@@ -1064,7 +1064,7 @@ async fn get_timestamp_of_lsn_handler(
     if !tenant_shard_id.is_shard_zero() {
         // Requires SLRU contents, which are only stored on shard zero
         return Err(ApiError::BadRequest(anyhow!(
-            "Timestamp calculations by lsn are only available on shard zero"
+            "Size calculations are only available on shard zero"
         )));
     }
 
@@ -1090,8 +1090,8 @@ async fn get_timestamp_of_lsn_handler(
             .to_string();
             json_response(StatusCode::OK, time)
         }
-        None => Err(ApiError::PreconditionFailed(
-            format!("Timestamp for lsn {} not found", lsn).into(),
+        None => Err(ApiError::NotFound(
+            anyhow::anyhow!("Timestamp for lsn {} not found", lsn).into(),
         )),
     }
 }
@@ -2274,7 +2274,6 @@ async fn timeline_compact_handler(
     if Some(true) == parse_query_param::<_, bool>(&request, "dry_run")? {
         flags |= CompactFlags::DryRun;
     }
-    // Manual compaction does not yield for L0.
 
     let wait_until_uploaded =
         parse_query_param::<_, bool>(&request, "wait_until_uploaded")?.unwrap_or(false);
@@ -2912,22 +2911,9 @@ async fn tenant_scan_remote_handler(
             };
         }
 
-        let result =
-            download_tenant_manifest(&state.remote_storage, &tenant_shard_id, generation, &cancel)
-                .instrument(info_span!("download_tenant_manifest",
-                            tenant_id=%tenant_shard_id.tenant_id,
-                            shard_id=%tenant_shard_id.shard_slug()))
-                .await;
-        let stripe_size = match result {
-            Ok((manifest, _, _)) => manifest.stripe_size,
-            Err(DownloadError::NotFound) => None,
-            Err(err) => return Err(ApiError::InternalServerError(anyhow!(err))),
-        };
-
         response.shards.push(TenantScanRemoteStorageShard {
             tenant_shard_id,
             generation: generation.into(),
-            stripe_size,
         });
     }
 
@@ -3382,11 +3368,11 @@ async fn put_tenant_timeline_import_basebackup(
 
         let broker_client = state.broker_client.clone();
 
-        let mut body = StreamReader::new(
-            request
-                .into_body()
-                .map(|res| res.map_err(|error| std::io::Error::other(anyhow::anyhow!(error)))),
-        );
+        let mut body = StreamReader::new(request.into_body().map(|res| {
+            res.map_err(|error| {
+                std::io::Error::new(std::io::ErrorKind::Other, anyhow::anyhow!(error))
+            })
+        }));
 
         tenant.wait_to_become_active(ACTIVE_TENANT_TIMEOUT).await?;
 
@@ -3460,7 +3446,7 @@ async fn put_tenant_timeline_import_wal(
 
         let mut body = StreamReader::new(request.into_body().map(|res| {
             res.map_err(|error| {
-                std::io::Error::other( anyhow::anyhow!(error))
+                std::io::Error::new(std::io::ErrorKind::Other, anyhow::anyhow!(error))
             })
         }));
 

pageserver/src/page_service.rs

@@ -105,7 +105,6 @@ pub fn spawn(
     pg_auth: Option<Arc<SwappableJwtAuth>>,
     perf_trace_dispatch: Option<Dispatch>,
     tcp_listener: tokio::net::TcpListener,
-    tls_config: Option<Arc<rustls::ServerConfig>>,
 ) -> Listener {
     let cancel = CancellationToken::new();
     let libpq_ctx = RequestContext::todo_child(
@@ -125,7 +124,6 @@ pub fn spawn(
             perf_trace_dispatch,
             tcp_listener,
             conf.pg_auth_type,
-            tls_config,
             conf.page_service_pipelining.clone(),
             libpq_ctx,
             cancel.clone(),
@@ -183,7 +181,6 @@ pub async fn libpq_listener_main(
     perf_trace_dispatch: Option<Dispatch>,
     listener: tokio::net::TcpListener,
     auth_type: AuthType,
-    tls_config: Option<Arc<rustls::ServerConfig>>,
     pipelining_config: PageServicePipeliningConfig,
     listener_ctx: RequestContext,
     listener_cancel: CancellationToken,
@@ -226,7 +223,6 @@ pub async fn libpq_listener_main(
                     local_auth,
                     socket,
                     auth_type,
-                    tls_config.clone(),
                     pipelining_config.clone(),
                     connection_ctx,
                     connections_cancel.child_token(),
@@ -268,7 +264,6 @@ async fn page_service_conn_main(
     auth: Option<Arc<SwappableJwtAuth>>,
     socket: tokio::net::TcpStream,
     auth_type: AuthType,
-    tls_config: Option<Arc<rustls::ServerConfig>>,
     pipelining_config: PageServicePipeliningConfig,
     connection_ctx: RequestContext,
     cancel: CancellationToken,
@@ -339,8 +334,7 @@ async fn page_service_conn_main(
         cancel.clone(),
         gate_guard,
     );
-    let pgbackend =
-        PostgresBackend::new_from_io(socket_fd, socket, peer_addr, auth_type, tls_config)?;
+    let pgbackend = PostgresBackend::new_from_io(socket_fd, socket, peer_addr, auth_type, None)?;
 
     match pgbackend.run(&mut conn_handler, &cancel).await {
         Ok(()) => {

pageserver/src/pgdatadir_mapping.rs

@@ -691,7 +691,7 @@ impl Timeline {
         Ok(buf.get_u32_le())
     }
 
-    /// Does the slru segment exist?
+    /// Get size of an SLRU segment
     pub(crate) async fn get_slru_segment_exists(
         &self,
         kind: SlruKind,
@@ -844,9 +844,9 @@ impl Timeline {
         .await
     }
 
-    /// Obtain the timestamp for the given lsn.
+    /// Obtain the possible timestamp range for the given lsn.
     ///
-    /// If the lsn has no timestamps (e.g. no commits), returns None.
+    /// If the lsn has no timestamps, returns None. returns `(min, max, median)` if it has timestamps.
     pub(crate) async fn get_timestamp_for_lsn(
         &self,
         probe_lsn: Lsn,

pageserver/src/tenant.rs

@@ -45,7 +45,6 @@ use remote_timeline_client::manifest::{
 };
 use remote_timeline_client::{
     FAILED_REMOTE_OP_RETRIES, FAILED_UPLOAD_WARN_THRESHOLD, UploadQueueNotReadyError,
-    download_tenant_manifest,
 };
 use secondary::heatmap::{HeatMapTenant, HeatMapTimeline};
 use storage_broker::BrokerClientChannel;
@@ -100,7 +99,7 @@ use crate::tenant::timeline::delete::DeleteTimelineFlow;
 use crate::tenant::timeline::uninit::cleanup_timeline_directory;
 use crate::virtual_file::VirtualFile;
 use crate::walingest::WalLagCooldown;
-use crate::walredo::{PostgresRedoManager, RedoAttemptType};
+use crate::walredo::PostgresRedoManager;
 use crate::{InitializationOrder, TEMP_FILE_SUFFIX, import_datadir, span, task_mgr, walredo};
 
 static INIT_DB_SEMAPHORE: Lazy<Semaphore> = Lazy::new(|| Semaphore::new(8));
@@ -227,8 +226,7 @@ struct TimelinePreload {
 }
 
 pub(crate) struct TenantPreload {
-    /// The tenant manifest from remote storage, or None if no manifest was found.
-    tenant_manifest: Option<TenantManifest>,
+    tenant_manifest: TenantManifest,
     /// Map from timeline ID to a possible timeline preload. It is None iff the timeline is offloaded according to the manifest.
     timelines: HashMap<TimelineId, Option<TimelinePreload>>,
 }
@@ -284,15 +282,12 @@ pub struct Tenant {
     /// **Lock order**: if acquiring all (or a subset), acquire them in order `timelines`, `timelines_offloaded`, `timelines_creating`
     timelines_offloaded: Mutex<HashMap<TimelineId, Arc<OffloadedTimeline>>>,
 
-    /// The last tenant manifest known to be in remote storage. None if the manifest has not yet
-    /// been either downloaded or uploaded. Always Some after tenant attach.
+    /// Serialize writes of the tenant manifest to remote storage.  If there are concurrent operations
+    /// affecting the manifest, such as timeline deletion and timeline offload, they must wait for
+    /// each other (this could be optimized to coalesce writes if necessary).
     ///
-    /// Initially populated during tenant attach, updated via `maybe_upload_tenant_manifest`.
-    ///
-    /// Do not modify this directly. It is used to check whether a new manifest needs to be
-    /// uploaded. The manifest is constructed in `build_tenant_manifest`, and uploaded via
-    /// `maybe_upload_tenant_manifest`.
-    remote_tenant_manifest: tokio::sync::Mutex<Option<TenantManifest>>,
+    /// The contents of the Mutex are the last manifest we successfully uploaded
+    tenant_manifest_upload: tokio::sync::Mutex<Option<TenantManifest>>,
 
     // This mutex prevents creation of new timelines during GC.
     // Adding yet another mutex (in addition to `timelines`) is needed because holding
@@ -473,16 +468,15 @@ impl WalRedoManager {
         base_img: Option<(Lsn, bytes::Bytes)>,
         records: Vec<(Lsn, pageserver_api::record::NeonWalRecord)>,
         pg_version: u32,
-        redo_attempt_type: RedoAttemptType,
     ) -> Result<bytes::Bytes, walredo::Error> {
         match self {
             Self::Prod(_, mgr) => {
-                mgr.request_redo(key, lsn, base_img, records, pg_version, redo_attempt_type)
+                mgr.request_redo(key, lsn, base_img, records, pg_version)
                     .await
             }
             #[cfg(test)]
             Self::Test(mgr) => {
-                mgr.request_redo(key, lsn, base_img, records, pg_version, redo_attempt_type)
+                mgr.request_redo(key, lsn, base_img, records, pg_version)
                     .await
             }
         }
@@ -921,7 +915,6 @@ enum StartCreatingTimelineResult {
     Idempotent(Arc<Timeline>),
 }
 
-#[allow(clippy::large_enum_variant, reason = "TODO")]
 enum TimelineInitAndSyncResult {
     ReadyToActivate(Arc<Timeline>),
     NeedsSpawnImportPgdata(TimelineInitAndSyncNeedsSpawnImportPgdata),
@@ -1008,7 +1001,6 @@ enum CreateTimelineCause {
     Delete,
 }
 
-#[allow(clippy::large_enum_variant, reason = "TODO")]
 enum LoadTimelineCause {
     Attach,
     Unoffload,
@@ -1362,41 +1354,36 @@ impl Tenant {
                     }
                 }
 
-                fn make_broken_or_stopping(t: &Tenant, err: anyhow::Error) {
-                    t.state.send_modify(|state| match state {
-                        // TODO: the old code alluded to DeleteTenantFlow sometimes setting
-                        // TenantState::Stopping before we get here, but this may be outdated.
-                        // Let's find out with a testing assertion. If this doesn't fire, and the
-                        // logs don't show this happening in production, remove the Stopping cases.
-                        TenantState::Stopping{..} if cfg!(any(test, feature = "testing")) => {
-                            panic!("unexpected TenantState::Stopping during attach")
-                        }
-                        // If the tenant is cancelled, assume the error was caused by cancellation.
-                        TenantState::Attaching if t.cancel.is_cancelled() => {
-                            info!("attach cancelled, setting tenant state to Stopping: {err}");
-                            // NB: progress None tells `set_stopping` that attach has cancelled.
-                            *state = TenantState::Stopping { progress: None };
-                        }
-                        // According to the old code, DeleteTenantFlow may already have set this to
-                        // Stopping. Retain its progress.
-                        // TODO: there is no DeleteTenantFlow. Is this still needed? See above.
-                        TenantState::Stopping { progress } if t.cancel.is_cancelled() => {
-                            assert!(progress.is_some(), "concurrent attach cancellation");
-                            info!("attach cancelled, already Stopping: {err}");
-                        }
-                        // Mark the tenant as broken.
-                        TenantState::Attaching | TenantState::Stopping { .. } => {
-                            error!("attach failed, setting tenant state to Broken (was {state}): {err:?}");
-                            *state = TenantState::broken_from_reason(err.to_string())
-                        }
-                        // The attach task owns the tenant state until activated.
-                        state => panic!("invalid tenant state {state} during attach: {err:?}"),
-                    });
+                // Ideally we should use Tenant::set_broken_no_wait, but it is not supposed to be used when tenant is in loading state.
+                enum BrokenVerbosity {
+                    Error,
+                    Info
                 }
+                let make_broken =
+                    |t: &Tenant, err: anyhow::Error, verbosity: BrokenVerbosity| {
+                        match verbosity {
+                            BrokenVerbosity::Info => {
+                                info!("attach cancelled, setting tenant state to Broken: {err}");
+                            },
+                            BrokenVerbosity::Error => {
+                                error!("attach failed, setting tenant state to Broken: {err:?}");
+                            }
+                        }
+                        t.state.send_modify(|state| {
+                            // The Stopping case is for when we have passed control on to DeleteTenantFlow:
+                            // if it errors, we will call make_broken when tenant is already in Stopping.
+                            assert!(
+                                matches!(*state, TenantState::Attaching | TenantState::Stopping { .. }),
+                                "the attach task owns the tenant state until activation is complete"
+                            );
+
+                            *state = TenantState::broken_from_reason(err.to_string());
+                        });
+                    };
 
                 // TODO: should also be rejecting tenant conf changes that violate this check.
                 if let Err(e) = crate::tenant::storage_layer::inmemory_layer::IndexEntry::validate_checkpoint_distance(tenant_clone.get_checkpoint_distance()) {
-                    make_broken_or_stopping(&tenant_clone, anyhow::anyhow!(e));
+                    make_broken(&tenant_clone, anyhow::anyhow!(e), BrokenVerbosity::Error);
                     return Ok(());
                 }
 
@@ -1448,8 +1435,10 @@ impl Tenant {
                             // stayed in Activating for such a long time that shutdown found it in
                             // that state.
                             tracing::info!(state=%tenant_clone.current_state(), "Tenant shut down before activation");
-                            // Set the tenant to Stopping to signal `set_stopping` that we're done.
-                            make_broken_or_stopping(&tenant_clone, anyhow::anyhow!("Shut down while Attaching"));
+                            // Make the tenant broken so that set_stopping will not hang waiting for it to leave
+                            // the Attaching state.  This is an over-reaction (nothing really broke, the tenant is
+                            // just shutting down), but ensures progress.
+                            make_broken(&tenant_clone, anyhow::anyhow!("Shut down while Attaching"), BrokenVerbosity::Info);
                             return Ok(());
                         },
                     )
@@ -1468,7 +1457,7 @@ impl Tenant {
                         match res {
                             Ok(p) => Some(p),
                             Err(e) => {
-                                make_broken_or_stopping(&tenant_clone, anyhow::anyhow!(e));
+                                make_broken(&tenant_clone, anyhow::anyhow!(e), BrokenVerbosity::Error);
                                 return Ok(());
                             }
                         }
@@ -1494,7 +1483,9 @@ impl Tenant {
                         info!("attach finished, activating");
                         tenant_clone.activate(broker_client, None, &ctx);
                     }
-                    Err(e) => make_broken_or_stopping(&tenant_clone, anyhow::anyhow!(e)),
+                    Err(e) => {
+                        make_broken(&tenant_clone, anyhow::anyhow!(e), BrokenVerbosity::Error);
+                    }
                 }
 
                 // If we are doing an opportunistic warmup attachment at startup, initialize
@@ -1534,27 +1525,28 @@ impl Tenant {
             cancel.clone(),
         )
         .await?;
-
-        let tenant_manifest = match download_tenant_manifest(
-            remote_storage,
-            &self.tenant_shard_id,
-            self.generation,
-            &cancel,
-        )
-        .await
-        {
-            Ok((tenant_manifest, _, _)) => Some(tenant_manifest),
-            Err(DownloadError::NotFound) => None,
-            Err(err) => return Err(err.into()),
-        };
+        let (offloaded_add, tenant_manifest) =
+            match remote_timeline_client::download_tenant_manifest(
+                remote_storage,
+                &self.tenant_shard_id,
+                self.generation,
+                &cancel,
+            )
+            .await
+            {
+                Ok((tenant_manifest, _generation, _manifest_mtime)) => (
+                    format!("{} offloaded", tenant_manifest.offloaded_timelines.len()),
+                    tenant_manifest,
+                ),
+                Err(DownloadError::NotFound) => {
+                    ("no manifest".to_string(), TenantManifest::empty())
+                }
+                Err(e) => Err(e)?,
+            };
 
         info!(
-            "found {} timelines ({} offloaded timelines)",
-            remote_timeline_ids.len(),
-            tenant_manifest
-                .as_ref()
-                .map(|m| m.offloaded_timelines.len())
-                .unwrap_or(0)
+            "found {} timelines, and {offloaded_add}",
+            remote_timeline_ids.len()
         );
 
         for k in other_keys {
@@ -1563,13 +1555,11 @@ impl Tenant {
 
         // Avoid downloading IndexPart of offloaded timelines.
         let mut offloaded_with_prefix = HashSet::new();
-        if let Some(tenant_manifest) = &tenant_manifest {
-            for offloaded in tenant_manifest.offloaded_timelines.iter() {
-                if remote_timeline_ids.remove(&offloaded.timeline_id) {
-                    offloaded_with_prefix.insert(offloaded.timeline_id);
-                } else {
-                    // We'll take care later of timelines in the manifest without a prefix
-                }
+        for offloaded in tenant_manifest.offloaded_timelines.iter() {
+            if remote_timeline_ids.remove(&offloaded.timeline_id) {
+                offloaded_with_prefix.insert(offloaded.timeline_id);
+            } else {
+                // We'll take care later of timelines in the manifest without a prefix
             }
         }
 
@@ -1643,14 +1633,12 @@ impl Tenant {
 
         let mut offloaded_timeline_ids = HashSet::new();
         let mut offloaded_timelines_list = Vec::new();
-        if let Some(tenant_manifest) = &preload.tenant_manifest {
-            for timeline_manifest in tenant_manifest.offloaded_timelines.iter() {
-                let timeline_id = timeline_manifest.timeline_id;
-                let offloaded_timeline =
-                    OffloadedTimeline::from_manifest(self.tenant_shard_id, timeline_manifest);
-                offloaded_timelines_list.push((timeline_id, Arc::new(offloaded_timeline)));
-                offloaded_timeline_ids.insert(timeline_id);
-            }
+        for timeline_manifest in preload.tenant_manifest.offloaded_timelines.iter() {
+            let timeline_id = timeline_manifest.timeline_id;
+            let offloaded_timeline =
+                OffloadedTimeline::from_manifest(self.tenant_shard_id, timeline_manifest);
+            offloaded_timelines_list.push((timeline_id, Arc::new(offloaded_timeline)));
+            offloaded_timeline_ids.insert(timeline_id);
         }
         // Complete deletions for offloaded timeline id's from manifest.
         // The manifest will be uploaded later in this function.
@@ -1808,21 +1796,15 @@ impl Tenant {
             .context("resume_deletion")
             .map_err(LoadLocalTimelineError::ResumeDeletion)?;
         }
+        let needs_manifest_upload =
+            offloaded_timelines_list.len() != preload.tenant_manifest.offloaded_timelines.len();
         {
             let mut offloaded_timelines_accessor = self.timelines_offloaded.lock().unwrap();
             offloaded_timelines_accessor.extend(offloaded_timelines_list.into_iter());
         }
-
-        // Stash the preloaded tenant manifest, and upload a new manifest if changed.
-        //
-        // NB: this must happen after the tenant is fully populated above. In particular the
-        // offloaded timelines, which are included in the manifest.
-        {
-            let mut guard = self.remote_tenant_manifest.lock().await;
-            assert!(guard.is_none(), "tenant manifest set before preload"); // first populated here
-            *guard = preload.tenant_manifest;
+        if needs_manifest_upload {
+            self.store_tenant_manifest().await?;
         }
-        self.maybe_upload_tenant_manifest().await?;
 
         // The local filesystem contents are a cache of what's in the remote IndexPart;
         // IndexPart is the source of truth.
@@ -2236,7 +2218,7 @@ impl Tenant {
         };
 
         // Upload new list of offloaded timelines to S3
-        self.maybe_upload_tenant_manifest().await?;
+        self.store_tenant_manifest().await?;
 
         // Activate the timeline (if it makes sense)
         if !(timeline.is_broken() || timeline.is_stopping()) {
@@ -3447,7 +3429,7 @@ impl Tenant {
             shutdown_mode
         };
 
-        match self.set_stopping(shutdown_progress).await {
+        match self.set_stopping(shutdown_progress, false, false).await {
             Ok(()) => {}
             Err(SetStoppingError::Broken) => {
                 // assume that this is acceptable
@@ -3527,13 +3509,25 @@ impl Tenant {
     /// This function waits for the tenant to become active if it isn't already, before transitioning it into Stopping state.
     ///
     /// This function is not cancel-safe!
-    async fn set_stopping(&self, progress: completion::Barrier) -> Result<(), SetStoppingError> {
+    ///
+    /// `allow_transition_from_loading` is needed for the special case of loading task deleting the tenant.
+    /// `allow_transition_from_attaching` is needed for the special case of attaching deleted tenant.
+    async fn set_stopping(
+        &self,
+        progress: completion::Barrier,
+        _allow_transition_from_loading: bool,
+        allow_transition_from_attaching: bool,
+    ) -> Result<(), SetStoppingError> {
         let mut rx = self.state.subscribe();
 
         // cannot stop before we're done activating, so wait out until we're done activating
         rx.wait_for(|state| match state {
+            TenantState::Attaching if allow_transition_from_attaching => true,
             TenantState::Activating(_) | TenantState::Attaching => {
-                info!("waiting for {state} to turn Active|Broken|Stopping");
+                info!(
+                    "waiting for {} to turn Active|Broken|Stopping",
+                    <&'static str>::from(state)
+                );
                 false
             }
             TenantState::Active | TenantState::Broken { .. } | TenantState::Stopping { .. } => true,
@@ -3544,24 +3538,25 @@ impl Tenant {
         // we now know we're done activating, let's see whether this task is the winner to transition into Stopping
         let mut err = None;
         let stopping = self.state.send_if_modified(|current_state| match current_state {
-            TenantState::Activating(_) | TenantState::Attaching => {
-                unreachable!("we ensured above that we're done with activation, and, there is no re-activation")
+            TenantState::Activating(_) => {
+                unreachable!("1we ensured above that we're done with activation, and, there is no re-activation")
+            }
+            TenantState::Attaching => {
+                if !allow_transition_from_attaching {
+                    unreachable!("2we ensured above that we're done with activation, and, there is no re-activation")
+                };
+                *current_state = TenantState::Stopping { progress };
+                true
             }
             TenantState::Active => {
                 // FIXME: due to time-of-check vs time-of-use issues, it can happen that new timelines
                 // are created after the transition to Stopping. That's harmless, as the Timelines
                 // won't be accessible to anyone afterwards, because the Tenant is in Stopping state.
-                *current_state = TenantState::Stopping { progress: Some(progress) };
+                *current_state = TenantState::Stopping { progress };
                 // Continue stopping outside the closure. We need to grab timelines.lock()
                 // and we plan to turn it into a tokio::sync::Mutex in a future patch.
                 true
             }
-            TenantState::Stopping { progress: None } => {
-                // An attach was cancelled, and the attach transitioned the tenant from Attaching to
-                // Stopping(None) to let us know it exited. Register our progress and continue.
-                *current_state = TenantState::Stopping { progress: Some(progress) };
-                true
-            }
             TenantState::Broken { reason, .. } => {
                 info!(
                     "Cannot set tenant to Stopping state, it is in Broken state due to: {reason}"
@@ -3569,7 +3564,7 @@ impl Tenant {
                 err = Some(SetStoppingError::Broken);
                 false
             }
-            TenantState::Stopping { progress: Some(progress) } => {
+            TenantState::Stopping { progress } => {
                 info!("Tenant is already in Stopping state");
                 err = Some(SetStoppingError::AlreadyStopping(progress.clone()));
                 false
@@ -4070,20 +4065,18 @@ impl Tenant {
 
     /// Generate an up-to-date TenantManifest based on the state of this Tenant.
     fn build_tenant_manifest(&self) -> TenantManifest {
-        // Collect the offloaded timelines, and sort them for deterministic output.
-        let offloaded_timelines = self
-            .timelines_offloaded
-            .lock()
-            .unwrap()
-            .values()
-            .map(|tli| tli.manifest())
-            .sorted_by_key(|m| m.timeline_id)
-            .collect_vec();
+        let timelines_offloaded = self.timelines_offloaded.lock().unwrap();
+
+        let mut timeline_manifests = timelines_offloaded
+            .iter()
+            .map(|(_timeline_id, offloaded)| offloaded.manifest())
+            .collect::<Vec<_>>();
+        // Sort the manifests so that our output is deterministic
+        timeline_manifests.sort_by_key(|timeline_manifest| timeline_manifest.timeline_id);
 
         TenantManifest {
             version: LATEST_TENANT_MANIFEST_VERSION,
-            stripe_size: Some(self.get_shard_stripe_size()),
-            offloaded_timelines,
+            offloaded_timelines: timeline_manifests,
         }
     }
 
@@ -4306,7 +4299,7 @@ impl Tenant {
             timelines: Mutex::new(HashMap::new()),
             timelines_creating: Mutex::new(HashSet::new()),
             timelines_offloaded: Mutex::new(HashMap::new()),
-            remote_tenant_manifest: Default::default(),
+            tenant_manifest_upload: Default::default(),
             gc_cs: tokio::sync::Mutex::new(()),
             walredo_mgr,
             remote_storage,
@@ -4402,7 +4395,10 @@ impl Tenant {
         .to_string();
 
         fail::fail_point!("tenant-config-before-write", |_| {
-            Err(std::io::Error::other("tenant-config-before-write"))
+            Err(std::io::Error::new(
+                std::io::ErrorKind::Other,
+                "tenant-config-before-write",
+            ))
         });
 
         // Convert the config to a toml file.
@@ -5536,35 +5532,27 @@ impl Tenant {
             .unwrap_or(0)
     }
 
-    /// Builds a new tenant manifest, and uploads it if it differs from the last-known tenant
-    /// manifest in `Self::remote_tenant_manifest`.
-    ///
-    /// TODO: instead of requiring callers to remember to call `maybe_upload_tenant_manifest` after
-    /// changing any `Tenant` state that's included in the manifest, consider making the manifest
-    /// the authoritative source of data with an API that automatically uploads on changes. Revisit
-    /// this when the manifest is more widely used and we have a better idea of the data model.
-    pub(crate) async fn maybe_upload_tenant_manifest(&self) -> Result<(), TenantManifestError> {
-        // Multiple tasks may call this function concurrently after mutating the Tenant runtime
-        // state, affecting the manifest generated by `build_tenant_manifest`. We use an async mutex
-        // to serialize these callers. `eq_ignoring_version` acts as a slightly inefficient but
-        // simple coalescing mechanism.
+    /// Serialize and write the latest TenantManifest to remote storage.
+    pub(crate) async fn store_tenant_manifest(&self) -> Result<(), TenantManifestError> {
+        // Only one manifest write may be done at at time, and the contents of the manifest
+        // must be loaded while holding this lock. This makes it safe to call this function
+        // from anywhere without worrying about colliding updates.
         let mut guard = tokio::select! {
-            guard = self.remote_tenant_manifest.lock() => guard,
-            _ = self.cancel.cancelled() => return Err(TenantManifestError::Cancelled),
+            g = self.tenant_manifest_upload.lock() => {
+                g
+            },
+            _ = self.cancel.cancelled() => {
+                return Err(TenantManifestError::Cancelled);
+            }
         };
 
-        // Build a new manifest.
         let manifest = self.build_tenant_manifest();
-
-        // Check if the manifest has changed. We ignore the version number here, to avoid
-        // uploading every manifest on version number bumps.
-        if let Some(old) = guard.as_ref() {
-            if manifest.eq_ignoring_version(old) {
-                return Ok(());
-            }
+        if Some(&manifest) == (*guard).as_ref() {
+            // Optimisation: skip uploads that don't change anything.
+            return Ok(());
         }
 
-        // Upload the manifest. Remote storage does no retries internally, so retry here.
+        // Remote storage does no retries internally, so wrap it
         match backoff::retry(
             || async {
                 upload_tenant_manifest(
@@ -5576,7 +5564,7 @@ impl Tenant {
                 )
                 .await
             },
-            |_| self.cancel.is_cancelled(),
+            |_e| self.cancel.is_cancelled(),
             FAILED_UPLOAD_WARN_THRESHOLD,
             FAILED_REMOTE_OP_RETRIES,
             "uploading tenant manifest",
@@ -5880,7 +5868,6 @@ pub(crate) mod harness {
             base_img: Option<(Lsn, Bytes)>,
             records: Vec<(Lsn, NeonWalRecord)>,
             _pg_version: u32,
-            _redo_attempt_type: RedoAttemptType,
         ) -> Result<Bytes, walredo::Error> {
             let records_neon = records.iter().all(|r| apply_neon::can_apply_in_neon(&r.1));
             if records_neon {
@@ -8735,21 +8722,6 @@ mod tests {
                 Lsn(0x20),
                 Value::WalRecord(NeonWalRecord::wal_init("i")),
             ),
-            (
-                get_key(4),
-                Lsn(0x30),
-                Value::WalRecord(NeonWalRecord::wal_append_conditional("j", "i")),
-            ),
-            (
-                get_key(5),
-                Lsn(0x20),
-                Value::WalRecord(NeonWalRecord::wal_init("1")),
-            ),
-            (
-                get_key(5),
-                Lsn(0x30),
-                Value::WalRecord(NeonWalRecord::wal_append_conditional("j", "2")),
-            ),
         ];
         let image1 = vec![(get_key(1), "0x10".into())];
 
@@ -8780,18 +8752,8 @@ mod tests {
 
         // Need to remove the limit of "Neon WAL redo requires base image".
 
-        assert_eq!(
-            tline.get(get_key(3), Lsn(0x50), &ctx).await?,
-            Bytes::from_static(b"c")
-        );
-        assert_eq!(
-            tline.get(get_key(4), Lsn(0x50), &ctx).await?,
-            Bytes::from_static(b"ij")
-        );
-
-        // Manual testing required: currently, read errors will panic the process in debug mode. So we
-        // cannot enable this assertion in the unit test.
-        // assert!(tline.get(get_key(5), Lsn(0x50), &ctx).await.is_err());
+        // assert_eq!(tline.get(get_key(3), Lsn(0x50), &ctx).await?, Bytes::new());
+        // assert_eq!(tline.get(get_key(4), Lsn(0x50), &ctx).await?, Bytes::new());
 
         Ok(())
     }
@@ -11571,99 +11533,6 @@ mod tests {
         Ok(())
     }
 
-    #[cfg(feature = "testing")]
-    #[tokio::test]
-    async fn test_bottom_most_compation_redo_failure() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_bottom_most_compation_redo_failure").await?;
-        let (tenant, ctx) = harness.load().await;
-
-        fn get_key(id: u32) -> Key {
-            // using aux key here b/c they are guaranteed to be inside `collect_keyspace`.
-            let mut key = Key::from_hex("620000000033333333444444445500000000").unwrap();
-            key.field6 = id;
-            key
-        }
-
-        let img_layer = (0..10)
-            .map(|id| (get_key(id), Bytes::from(format!("value {id}@0x10"))))
-            .collect_vec();
-
-        let delta1 = vec![
-            (
-                get_key(1),
-                Lsn(0x20),
-                Value::WalRecord(NeonWalRecord::wal_append("@0x20")),
-            ),
-            (
-                get_key(1),
-                Lsn(0x24),
-                Value::WalRecord(NeonWalRecord::wal_append("@0x24")),
-            ),
-            (
-                get_key(1),
-                Lsn(0x28),
-                // This record will fail to redo
-                Value::WalRecord(NeonWalRecord::wal_append_conditional("@0x28", "???")),
-            ),
-        ];
-
-        let tline = tenant
-            .create_test_timeline_with_layers(
-                TIMELINE_ID,
-                Lsn(0x10),
-                DEFAULT_PG_VERSION,
-                &ctx,
-                vec![], // in-memory layers
-                vec![DeltaLayerTestDesc::new_with_inferred_key_range(
-                    Lsn(0x20)..Lsn(0x30),
-                    delta1,
-                )], // delta layers
-                vec![(Lsn(0x10), img_layer)], // image layers
-                Lsn(0x50),
-            )
-            .await?;
-        {
-            tline
-                .applied_gc_cutoff_lsn
-                .lock_for_write()
-                .store_and_unlock(Lsn(0x30))
-                .wait()
-                .await;
-            // Update GC info
-            let mut guard = tline.gc_info.write().unwrap();
-            *guard = GcInfo {
-                retain_lsns: vec![],
-                cutoffs: GcCutoffs {
-                    time: Lsn(0x30),
-                    space: Lsn(0x30),
-                },
-                leases: Default::default(),
-                within_ancestor_pitr: false,
-            };
-        }
-
-        let cancel = CancellationToken::new();
-
-        // Compaction will fail, but should not fire any critical error.
-        // Gc-compaction currently cannot figure out what keys are not in the keyspace during the compaction
-        // process. It will always try to redo the logs it reads and if it doesn't work, fail the entire
-        // compaction job. Tracked in <https://github.com/neondatabase/neon/issues/10395>.
-        let res = tline
-            .compact_with_gc(
-                &cancel,
-                CompactOptions {
-                    compact_key_range: None,
-                    compact_lsn_range: None,
-                    ..Default::default()
-                },
-                &ctx,
-            )
-            .await;
-        assert!(res.is_err());
-
-        Ok(())
-    }
-
     #[cfg(feature = "testing")]
     #[tokio::test]
     async fn test_synthetic_size_calculation_with_invisible_branches() -> anyhow::Result<()> {

pageserver/src/tenant/blob_io.rs

@@ -15,7 +15,7 @@
 //! len >= 128: 1CCCXXXX XXXXXXXX XXXXXXXX XXXXXXXX
 //!
 use std::cmp::min;
-use std::io::Error;
+use std::io::{Error, ErrorKind};
 
 use async_compression::Level;
 use bytes::{BufMut, BytesMut};
@@ -331,7 +331,10 @@ impl<const BUFFERED: bool> BlobWriter<BUFFERED> {
                     return (
                         (
                             io_buf.slice_len(),
-                            Err(Error::other(format!("blob too large ({len} bytes)"))),
+                            Err(Error::new(
+                                ErrorKind::Other,
+                                format!("blob too large ({len} bytes)"),
+                            )),
                         ),
                         srcbuf,
                     );

pageserver/src/tenant/block_io.rs

@@ -216,8 +216,12 @@ impl<'a> FileBlockReader<'a> {
         match cache
             .read_immutable_buf(self.file_id, blknum, ctx)
             .await
-            .map_err(|e| std::io::Error::other(format!("Failed to read immutable buf: {e:#}")))?
-        {
+            .map_err(|e| {
+                std::io::Error::new(
+                    std::io::ErrorKind::Other,
+                    format!("Failed to read immutable buf: {e:#}"),
+                )
+            })? {
             ReadBufResult::Found(guard) => Ok(guard.into()),
             ReadBufResult::NotFound(write_guard) => {
                 // Read the page from disk into the buffer

pageserver/src/tenant/remote_timeline_client/manifest.rs

@@ -1,33 +1,21 @@
 use chrono::NaiveDateTime;
-use pageserver_api::shard::ShardStripeSize;
 use serde::{Deserialize, Serialize};
 use utils::id::TimelineId;
 use utils::lsn::Lsn;
 
-/// Tenant shard manifest, stored in remote storage. Contains offloaded timelines and other tenant
-/// shard-wide information that must be persisted in remote storage.
-///
-/// The manifest is always updated on tenant attach, and as needed.
-#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
+/// Tenant-shard scoped manifest
+#[derive(Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub struct TenantManifest {
-    /// The manifest version. Incremented on manifest format changes, even non-breaking ones.
-    /// Manifests must generally always be backwards and forwards compatible for one release, to
-    /// allow release rollbacks.
+    /// Debugging aid describing the version of this manifest.
+    /// Can also be used for distinguishing breaking changes later on.
     pub version: usize,
 
-    /// This tenant's stripe size. This is only advisory, and used to recover tenant data from
-    /// remote storage. The autoritative source is the storage controller. If None, assume the
-    /// original default value of 32768 blocks (256 MB).
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub stripe_size: Option<ShardStripeSize>,
-
     /// The list of offloaded timelines together with enough information
     /// to not have to actually load them.
     ///
     /// Note: the timelines mentioned in this list might be deleted, i.e.
     /// we don't hold an invariant that the references aren't dangling.
     /// Existence of index-part.json is the actual indicator of timeline existence.
-    #[serde(default)]
     pub offloaded_timelines: Vec<OffloadedTimelineManifest>,
 }
 
@@ -36,7 +24,7 @@ pub struct TenantManifest {
 /// Very similar to [`pageserver_api::models::OffloadedTimelineInfo`],
 /// but the two datastructures serve different needs, this is for a persistent disk format
 /// that must be backwards compatible, while the other is only for informative purposes.
-#[derive(Clone, Debug, Serialize, Deserialize, Copy, PartialEq, Eq)]
+#[derive(Clone, Serialize, Deserialize, Copy, PartialEq, Eq)]
 pub struct OffloadedTimelineManifest {
     pub timeline_id: TimelineId,
     /// Whether the timeline has a parent it has been branched off from or not
@@ -47,166 +35,20 @@ pub struct OffloadedTimelineManifest {
     pub archived_at: NaiveDateTime,
 }
 
-/// The newest manifest version. This should be incremented on changes, even non-breaking ones. We
-/// do not use deny_unknown_fields, so new fields are not breaking.
-///
-/// 1: initial version
-/// 2: +stripe_size
-///
-/// When adding new versions, also add a parse_vX test case below.
-pub const LATEST_TENANT_MANIFEST_VERSION: usize = 2;
+pub const LATEST_TENANT_MANIFEST_VERSION: usize = 1;
 
 impl TenantManifest {
-    /// Returns true if the manifests are equal, ignoring the version number. This avoids
-    /// re-uploading all manifests just because the version number is bumped.
-    pub fn eq_ignoring_version(&self, other: &Self) -> bool {
-        // Fast path: if the version is equal, just compare directly.
-        if self.version == other.version {
-            return self == other;
+    pub(crate) fn empty() -> Self {
+        Self {
+            version: LATEST_TENANT_MANIFEST_VERSION,
+            offloaded_timelines: vec![],
         }
-
-        // We could alternatively just clone and modify the version here.
-        let Self {
-            version: _, // ignore version
-            stripe_size,
-            offloaded_timelines,
-        } = self;
-
-        stripe_size == &other.stripe_size && offloaded_timelines == &other.offloaded_timelines
     }
-
-    /// Decodes a manifest from JSON.
     pub fn from_json_bytes(bytes: &[u8]) -> Result<Self, serde_json::Error> {
-        serde_json::from_slice(bytes)
+        serde_json::from_slice::<Self>(bytes)
     }
 
-    /// Encodes a manifest as JSON.
-    pub fn to_json_bytes(&self) -> serde_json::Result<Vec<u8>> {
+    pub(crate) fn to_json_bytes(&self) -> serde_json::Result<Vec<u8>> {
         serde_json::to_vec(self)
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use std::str::FromStr;
-
-    use utils::id::TimelineId;
-
-    use super::*;
-
-    /// Empty manifests should be parsed. Version is required.
-    #[test]
-    fn parse_empty() -> anyhow::Result<()> {
-        let json = r#"{
-             "version": 0
-         }"#;
-        let expected = TenantManifest {
-            version: 0,
-            stripe_size: None,
-            offloaded_timelines: Vec::new(),
-        };
-        assert_eq!(expected, TenantManifest::from_json_bytes(json.as_bytes())?);
-        Ok(())
-    }
-
-    /// Unknown fields should be ignored, for forwards compatibility.
-    #[test]
-    fn parse_unknown_fields() -> anyhow::Result<()> {
-        let json = r#"{
-             "version": 1,
-             "foo": "bar"
-         }"#;
-        let expected = TenantManifest {
-            version: 1,
-            stripe_size: None,
-            offloaded_timelines: Vec::new(),
-        };
-        assert_eq!(expected, TenantManifest::from_json_bytes(json.as_bytes())?);
-        Ok(())
-    }
-
-    /// v1 manifests should be parsed, for backwards compatibility.
-    #[test]
-    fn parse_v1() -> anyhow::Result<()> {
-        let json = r#"{
-             "version": 1,
-             "offloaded_timelines": [
-                 {
-                     "timeline_id": "5c4df612fd159e63c1b7853fe94d97da",
-                     "archived_at": "2025-03-07T11:07:11.373105434"
-                 },
-                 {
-                     "timeline_id": "f3def5823ad7080d2ea538d8e12163fa",
-                     "ancestor_timeline_id": "5c4df612fd159e63c1b7853fe94d97da",
-                     "ancestor_retain_lsn": "0/1F79038",
-                     "archived_at": "2025-03-05T11:10:22.257901390"
-                 }
-             ]
-         }"#;
-        let expected = TenantManifest {
-            version: 1,
-            stripe_size: None,
-            offloaded_timelines: vec![
-                OffloadedTimelineManifest {
-                    timeline_id: TimelineId::from_str("5c4df612fd159e63c1b7853fe94d97da")?,
-                    ancestor_timeline_id: None,
-                    ancestor_retain_lsn: None,
-                    archived_at: NaiveDateTime::from_str("2025-03-07T11:07:11.373105434")?,
-                },
-                OffloadedTimelineManifest {
-                    timeline_id: TimelineId::from_str("f3def5823ad7080d2ea538d8e12163fa")?,
-                    ancestor_timeline_id: Some(TimelineId::from_str(
-                        "5c4df612fd159e63c1b7853fe94d97da",
-                    )?),
-                    ancestor_retain_lsn: Some(Lsn::from_str("0/1F79038")?),
-                    archived_at: NaiveDateTime::from_str("2025-03-05T11:10:22.257901390")?,
-                },
-            ],
-        };
-        assert_eq!(expected, TenantManifest::from_json_bytes(json.as_bytes())?);
-        Ok(())
-    }
-
-    /// v2 manifests should be parsed, for backwards compatibility.
-    #[test]
-    fn parse_v2() -> anyhow::Result<()> {
-        let json = r#"{
-             "version": 2,
-             "stripe_size": 32768,
-             "offloaded_timelines": [
-                 {
-                     "timeline_id": "5c4df612fd159e63c1b7853fe94d97da",
-                     "archived_at": "2025-03-07T11:07:11.373105434"
-                 },
-                 {
-                     "timeline_id": "f3def5823ad7080d2ea538d8e12163fa",
-                     "ancestor_timeline_id": "5c4df612fd159e63c1b7853fe94d97da",
-                     "ancestor_retain_lsn": "0/1F79038",
-                     "archived_at": "2025-03-05T11:10:22.257901390"
-                 }
-             ]
-         }"#;
-        let expected = TenantManifest {
-            version: 2,
-            stripe_size: Some(ShardStripeSize(32768)),
-            offloaded_timelines: vec![
-                OffloadedTimelineManifest {
-                    timeline_id: TimelineId::from_str("5c4df612fd159e63c1b7853fe94d97da")?,
-                    ancestor_timeline_id: None,
-                    ancestor_retain_lsn: None,
-                    archived_at: NaiveDateTime::from_str("2025-03-07T11:07:11.373105434")?,
-                },
-                OffloadedTimelineManifest {
-                    timeline_id: TimelineId::from_str("f3def5823ad7080d2ea538d8e12163fa")?,
-                    ancestor_timeline_id: Some(TimelineId::from_str(
-                        "5c4df612fd159e63c1b7853fe94d97da",
-                    )?),
-                    ancestor_retain_lsn: Some(Lsn::from_str("0/1F79038")?),
-                    archived_at: NaiveDateTime::from_str("2025-03-05T11:10:22.257901390")?,
-                },
-            ],
-        };
-        assert_eq!(expected, TenantManifest::from_json_bytes(json.as_bytes())?);
-        Ok(())
-    }
-}

pageserver/src/tenant/remote_timeline_client/upload.rs

@@ -61,7 +61,6 @@ pub(crate) async fn upload_index_part(
         .await
         .with_context(|| format!("upload index part for '{tenant_shard_id} / {timeline_id}'"))
 }
-
 /// Serializes and uploads the given tenant manifest data to the remote storage.
 pub(crate) async fn upload_tenant_manifest(
     storage: &GenericRemoteStorage,
@@ -77,14 +76,16 @@ pub(crate) async fn upload_tenant_manifest(
     });
     pausable_failpoint!("before-upload-manifest-pausable");
 
-    let serialized = Bytes::from(tenant_manifest.to_json_bytes()?);
-    let tenant_manifest_size = serialized.len();
-    let remote_path = remote_tenant_manifest_path(tenant_shard_id, generation);
+    let serialized = tenant_manifest.to_json_bytes()?;
+    let serialized = Bytes::from(serialized);
 
+    let tenant_manifest_site = serialized.len();
+
+    let remote_path = remote_tenant_manifest_path(tenant_shard_id, generation);
     storage
         .upload_storage_object(
             futures::stream::once(futures::future::ready(Ok(serialized))),
-            tenant_manifest_size,
+            tenant_manifest_site,
             &remote_path,
             cancel,
         )

pageserver/src/tenant/storage_layer/batch_split_writer.rs

@@ -366,7 +366,7 @@ impl SplitDeltaLayerWriter {
                 )
                 .await?;
                 let (start_key, prev_delta_writer) =
-                    self.inner.replace((key, next_delta_writer)).unwrap();
+                    std::mem::replace(&mut self.inner, Some((key, next_delta_writer))).unwrap();
                 self.batches.add_unfinished_delta_writer(
                     prev_delta_writer,
                     start_key..key,

pageserver/src/tenant/storage_layer/image_layer.rs

@@ -1192,7 +1192,7 @@ mod test {
 
         // This key range contains several 0x8000 page stripes, only one of which belongs to shard zero
         let input_start = Key::from_hex("000000067f00000001000000ae0000000000").unwrap();
-        let input_end = Key::from_hex("000000067f00000001000000ae0000002000").unwrap();
+        let input_end = Key::from_hex("000000067f00000001000000ae0000020000").unwrap();
         let range = input_start..input_end;
 
         // Build an image layer to filter
@@ -1235,7 +1235,7 @@ mod test {
             let shard_identity = ShardIdentity::new(
                 ShardNumber(shard_number),
                 shard_count,
-                ShardStripeSize(0x800),
+                ShardStripeSize(0x8000),
             )
             .unwrap();
             let harness = TenantHarness::create_custom(
@@ -1287,12 +1287,12 @@ mod test {
 
             // This exact size and those below will need updating as/when the layer encoding changes, but
             // should be deterministic for a given version of the format, as we used no randomness generating the input.
-            assert_eq!(original_size, 122880);
+            assert_eq!(original_size, 1597440);
 
             match shard_number {
                 0 => {
                     // We should have written out just one stripe for our shard identity
-                    assert_eq!(wrote_keys, 0x800);
+                    assert_eq!(wrote_keys, 0x8000);
                     let replacement = replacement.unwrap();
 
                     // We should have dropped some of the data
@@ -1300,7 +1300,7 @@ mod test {
                     assert!(replacement.metadata().file_size > 0);
 
                     // Assert that we dropped ~3/4 of the data.
-                    assert_eq!(replacement.metadata().file_size, 49152);
+                    assert_eq!(replacement.metadata().file_size, 417792);
                 }
                 1 => {
                     // Shard 1 has no keys in our input range
@@ -1309,19 +1309,19 @@ mod test {
                 }
                 2 => {
                     // Shard 2 has one stripes in the input range
-                    assert_eq!(wrote_keys, 0x800);
+                    assert_eq!(wrote_keys, 0x8000);
                     let replacement = replacement.unwrap();
                     assert!(replacement.metadata().file_size < original_size);
                     assert!(replacement.metadata().file_size > 0);
-                    assert_eq!(replacement.metadata().file_size, 49152);
+                    assert_eq!(replacement.metadata().file_size, 417792);
                 }
                 3 => {
                     // Shard 3 has two stripes in the input range
-                    assert_eq!(wrote_keys, 0x1000);
+                    assert_eq!(wrote_keys, 0x10000);
                     let replacement = replacement.unwrap();
                     assert!(replacement.metadata().file_size < original_size);
                     assert!(replacement.metadata().file_size > 0);
-                    assert_eq!(replacement.metadata().file_size, 73728);
+                    assert_eq!(replacement.metadata().file_size, 811008);
                 }
                 _ => unreachable!(),
             }

pageserver/src/tenant/storage_layer/inmemory_layer/vectored_dio_read.rs

@@ -766,7 +766,7 @@ mod tests {
                     rand::Rng::fill(&mut rand::thread_rng(), &mut dst_slice[len..]); // to discover bugs
                     Ok((dst, len))
                 }
-                Err(e) => Err(std::io::Error::other(e)),
+                Err(e) => Err(std::io::Error::new(std::io::ErrorKind::Other, e)),
             }
         }
     }

pageserver/src/tenant/storage_layer/merge_iterator.rs

@@ -59,7 +59,6 @@ impl LayerIterRef<'_> {
 /// 1. Unified iterator for image and delta layers.
 /// 2. `Ord` for use in [`MergeIterator::heap`] (for the k-merge).
 /// 3. Lazy creation of the real delta/image iterator.
-#[allow(clippy::large_enum_variant, reason = "TODO")]
 pub(crate) enum IteratorWrapper<'a> {
     NotLoaded {
         ctx: &'a RequestContext,

pageserver/src/tenant/tasks.rs

@@ -268,12 +268,7 @@ async fn compaction_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
                 error_run += 1;
                 let backoff =
                     exponential_backoff_duration(error_run, BASE_BACKOFF_SECS, MAX_BACKOFF_SECS);
-                log_compaction_error(
-                    &err,
-                    Some((error_run, backoff)),
-                    cancel.is_cancelled(),
-                    false,
-                );
+                log_compaction_error(&err, Some((error_run, backoff)), cancel.is_cancelled());
                 continue;
             }
         }
@@ -290,7 +285,6 @@ pub(crate) fn log_compaction_error(
     err: &CompactionError,
     retry_info: Option<(u32, Duration)>,
     task_cancelled: bool,
-    degrade_to_warning: bool,
 ) {
     use CompactionError::*;
 
@@ -339,7 +333,6 @@ pub(crate) fn log_compaction_error(
         }
     } else {
         match level {
-            Level::ERROR if degrade_to_warning => warn!("Compaction failed and discarded: {err:#}"),
             Level::ERROR => error!("Compaction failed: {err:#}"),
             Level::INFO => info!("Compaction failed: {err:#}"),
             level => unimplemented!("unexpected level {level:?}"),

pageserver/src/tenant/timeline.rs

@@ -24,7 +24,6 @@ use std::sync::{Arc, Mutex, OnceLock, RwLock, Weak};
 use std::time::{Duration, Instant, SystemTime};
 
 use crate::PERF_TRACE_TARGET;
-use crate::walredo::RedoAttemptType;
 use anyhow::{Context, Result, anyhow, bail, ensure};
 use arc_swap::{ArcSwap, ArcSwapOption};
 use bytes::Bytes;
@@ -116,7 +115,7 @@ use crate::pgdatadir_mapping::{
 use crate::task_mgr::TaskKind;
 use crate::tenant::config::AttachmentMode;
 use crate::tenant::gc_result::GcResult;
-use crate::tenant::layer_map::LayerMap;
+use crate::tenant::layer_map::{LayerMap, SearchResult};
 use crate::tenant::metadata::TimelineMetadata;
 use crate::tenant::storage_layer::delta_layer::DeltaEntry;
 use crate::tenant::storage_layer::inmemory_layer::IndexEntry;
@@ -1040,7 +1039,6 @@ pub(crate) enum ShutdownMode {
     Hard,
 }
 
-#[allow(clippy::large_enum_variant, reason = "TODO")]
 enum ImageLayerCreationOutcome {
     /// We generated an image layer
     Generated {
@@ -1294,12 +1292,6 @@ impl Timeline {
         };
         reconstruct_state.read_path = read_path;
 
-        let redo_attempt_type = if ctx.task_kind() == TaskKind::Compaction {
-            RedoAttemptType::LegacyCompaction
-        } else {
-            RedoAttemptType::ReadPage
-        };
-
         let traversal_res: Result<(), _> = {
             let ctx = RequestContextBuilder::from(ctx)
                 .perf_span(|crnt_perf_span| {
@@ -1387,7 +1379,7 @@ impl Timeline {
 
                     let walredo_deltas = converted.num_deltas();
                     let walredo_res = walredo_self
-                        .reconstruct_value(key, lsn, converted, redo_attempt_type)
+                        .reconstruct_value(key, lsn, converted)
                         .maybe_perf_instrument(&ctx, |crnt_perf_span| {
                             info_span!(
                                 target: PERF_TRACE_TARGET,
@@ -1948,7 +1940,7 @@ impl Timeline {
             )
             .await;
         if let Err(err) = &res {
-            log_compaction_error(err, None, cancel.is_cancelled(), false);
+            log_compaction_error(err, None, cancel.is_cancelled());
         }
         res
     }
@@ -4112,6 +4104,12 @@ impl Timeline {
         cancel: &CancellationToken,
         ctx: &RequestContext,
     ) -> Result<TimelineVisitOutcome, GetVectoredError> {
+        let mut unmapped_keyspace = keyspace.clone();
+        let mut fringe = LayerFringe::new();
+
+        let mut completed_keyspace = KeySpace::default();
+        let mut image_covered_keyspace = KeySpaceRandomAccum::new();
+
         // Prevent GC from progressing while visiting the current timeline.
         // If we are GC-ing because a new image layer was added while traversing
         // the timeline, then it will remove layers that are required for fulfilling
@@ -4122,44 +4120,11 @@ impl Timeline {
         // See `compaction::compact_with_gc` for why we need this.
         let _guard = timeline.gc_compaction_layer_update_lock.read().await;
 
-        // Initialize the fringe
-        let mut fringe = {
-            let mut fringe = LayerFringe::new();
-
-            let guard = timeline.layers.read().await;
-            guard.update_search_fringe(&keyspace, cont_lsn, &mut fringe)?;
-
-            fringe
-        };
-
-        let mut completed_keyspace = KeySpace::default();
-        let mut image_covered_keyspace = KeySpaceRandomAccum::new();
-
-        while let Some((layer_to_read, keyspace_to_read, lsn_range)) = fringe.next_layer() {
+        loop {
             if cancel.is_cancelled() {
                 return Err(GetVectoredError::Cancelled);
             }
 
-            if let Some(ref mut read_path) = reconstruct_state.read_path {
-                read_path.record_layer_visit(&layer_to_read, &keyspace_to_read, &lsn_range);
-            }
-
-            // Visit the layer and plan IOs for it
-            let next_cont_lsn = lsn_range.start;
-            layer_to_read
-                .get_values_reconstruct_data(
-                    keyspace_to_read.clone(),
-                    lsn_range,
-                    reconstruct_state,
-                    ctx,
-                )
-                .await?;
-
-            let mut unmapped_keyspace = keyspace_to_read;
-            cont_lsn = next_cont_lsn;
-
-            reconstruct_state.on_layer_visited(&layer_to_read);
-
             let (keys_done_last_step, keys_with_image_coverage) =
                 reconstruct_state.consume_done_keys();
             unmapped_keyspace.remove_overlapping_with(&keys_done_last_step);
@@ -4170,15 +4135,31 @@ impl Timeline {
                 image_covered_keyspace.add_range(keys_with_image_coverage);
             }
 
-            // Query the layer map for the next layers to read.
-            //
             // Do not descent any further if the last layer we visited
             // completed all keys in the keyspace it inspected. This is not
             // required for correctness, but avoids visiting extra layers
             // which turns out to be a perf bottleneck in some cases.
             if !unmapped_keyspace.is_empty() {
                 let guard = timeline.layers.read().await;
-                guard.update_search_fringe(&unmapped_keyspace, cont_lsn, &mut fringe)?;
+                let layers = guard.layer_map()?;
+
+                for range in unmapped_keyspace.ranges.iter() {
+                    let results = layers.range_search(range.clone(), cont_lsn);
+
+                    results
+                        .found
+                        .into_iter()
+                        .map(|(SearchResult { layer, lsn_floor }, keyspace_accum)| {
+                            (
+                                guard.upgrade(layer),
+                                keyspace_accum.to_keyspace(),
+                                lsn_floor..cont_lsn,
+                            )
+                        })
+                        .for_each(|(layer, keyspace, lsn_range)| {
+                            fringe.update(layer, keyspace, lsn_range)
+                        });
+                }
 
                 // It's safe to drop the layer map lock after planning the next round of reads.
                 // The fringe keeps readable handles for the layers which are safe to read even
@@ -4192,6 +4173,28 @@ impl Timeline {
                 // at two different time points.
                 drop(guard);
             }
+
+            if let Some((layer_to_read, keyspace_to_read, lsn_range)) = fringe.next_layer() {
+                if let Some(ref mut read_path) = reconstruct_state.read_path {
+                    read_path.record_layer_visit(&layer_to_read, &keyspace_to_read, &lsn_range);
+                }
+                let next_cont_lsn = lsn_range.start;
+                layer_to_read
+                    .get_values_reconstruct_data(
+                        keyspace_to_read.clone(),
+                        lsn_range,
+                        reconstruct_state,
+                        ctx,
+                    )
+                    .await?;
+
+                unmapped_keyspace = keyspace_to_read;
+                cont_lsn = next_cont_lsn;
+
+                reconstruct_state.on_layer_visited(&layer_to_read);
+            } else {
+                break;
+            }
         }
 
         Ok(TimelineVisitOutcome {
@@ -6354,17 +6357,10 @@ impl Timeline {
         key: Key,
         request_lsn: Lsn,
         mut data: ValueReconstructState,
-        redo_attempt_type: RedoAttemptType,
     ) -> Result<Bytes, PageReconstructError> {
         // Perform WAL redo if needed
         data.records.reverse();
 
-        let fire_critical_error = match redo_attempt_type {
-            RedoAttemptType::ReadPage => true,
-            RedoAttemptType::LegacyCompaction => true,
-            RedoAttemptType::GcCompaction => false,
-        };
-
         // If we have a page image, and no WAL, we're all set
         if data.records.is_empty() {
             if let Some((img_lsn, img)) = &data.img {
@@ -6411,22 +6407,13 @@ impl Timeline {
                     .as_ref()
                     .context("timeline has no walredo manager")
                     .map_err(PageReconstructError::WalRedo)?
-                    .request_redo(
-                        key,
-                        request_lsn,
-                        data.img,
-                        data.records,
-                        self.pg_version,
-                        redo_attempt_type,
-                    )
+                    .request_redo(key, request_lsn, data.img, data.records, self.pg_version)
                     .await;
                 let img = match res {
                     Ok(img) => img,
                     Err(walredo::Error::Cancelled) => return Err(PageReconstructError::Cancelled),
                     Err(walredo::Error::Other(err)) => {
-                        if fire_critical_error {
-                            critical!("walredo failure during page reconstruction: {err:?}");
-                        }
+                        critical!("walredo failure during page reconstruction: {err:?}");
                         return Err(PageReconstructError::WalRedo(
                             err.context("reconstruct a page image"),
                         ));

pageserver/src/tenant/timeline/compaction.rs

@@ -7,7 +7,7 @@
 use std::collections::{BinaryHeap, HashMap, HashSet, VecDeque};
 use std::ops::{Deref, Range};
 use std::sync::Arc;
-use std::time::{Duration, Instant};
+use std::time::Instant;
 
 use super::layer_manager::LayerManager;
 use super::{
@@ -16,8 +16,6 @@ use super::{
     Timeline,
 };
 
-use crate::tenant::timeline::DeltaEntry;
-use crate::walredo::RedoAttemptType;
 use anyhow::{Context, anyhow};
 use bytes::Bytes;
 use enumset::EnumSet;
@@ -317,9 +315,6 @@ impl GcCompactionQueue {
                     flags: {
                         let mut flags = EnumSet::new();
                         flags |= CompactFlags::EnhancedGcBottomMostCompaction;
-                        if timeline.get_compaction_l0_first() {
-                            flags |= CompactFlags::YieldForL0;
-                        }
                         flags
                     },
                     sub_compaction: true,
@@ -453,7 +448,7 @@ impl GcCompactionQueue {
     ) -> Result<CompactionOutcome, CompactionError> {
         let res = self.iteration_inner(cancel, ctx, gc_block, timeline).await;
         if let Err(err) = &res {
-            log_compaction_error(err, None, cancel.is_cancelled(), true);
+            log_compaction_error(err, None, cancel.is_cancelled());
         }
         match res {
             Ok(res) => Ok(res),
@@ -824,16 +819,15 @@ pub struct CompactionStatistics {
     time_acquire_lock_secs: f64,
     time_analyze_secs: f64,
     time_download_layer_secs: f64,
-    time_to_first_kv_pair_secs: f64,
     time_main_loop_secs: f64,
     time_final_phase_secs: f64,
     time_total_secs: f64,
 
     // Summary
-    /// Ratio of the key-value size after/before gc-compaction.
-    uncompressed_retention_ratio: f64,
-    /// Ratio of the physical size after/before gc-compaction.
-    compressed_retention_ratio: f64,
+    /// Ratio of the key-value size before/after gc-compaction.
+    uncompressed_size_ratio: f64,
+    /// Ratio of the physical size before/after gc-compaction.
+    physical_size_ratio: f64,
 }
 
 impl CompactionStatistics {
@@ -902,15 +896,15 @@ impl CompactionStatistics {
     fn finalize(&mut self) {
         let original_key_value_size = self.image_keys_visited.size + self.wal_keys_visited.size;
         let produced_key_value_size = self.image_produced.size + self.wal_produced.size;
-        self.uncompressed_retention_ratio =
-            produced_key_value_size as f64 / (original_key_value_size as f64 + 1.0); // avoid div by 0
+        self.uncompressed_size_ratio =
+            original_key_value_size as f64 / (produced_key_value_size as f64 + 1.0); // avoid div by 0
         let original_physical_size = self.image_layer_visited.size + self.delta_layer_visited.size;
         let produced_physical_size = self.image_layer_produced.size
             + self.delta_layer_produced.size
             + self.image_layer_discarded.size
             + self.delta_layer_discarded.size; // Also include the discarded layers to make the ratio accurate
-        self.compressed_retention_ratio =
-            produced_physical_size as f64 / (original_physical_size as f64 + 1.0); // avoid div by 0
+        self.physical_size_ratio =
+            original_physical_size as f64 / (produced_physical_size as f64 + 1.0); // avoid div by 0
     }
 }
 
@@ -2416,9 +2410,7 @@ impl Timeline {
                 } else {
                     lsn_split_points[i]
                 };
-                let img = self
-                    .reconstruct_value(key, request_lsn, state, RedoAttemptType::GcCompaction)
-                    .await?;
+                let img = self.reconstruct_value(key, request_lsn, state).await?;
                 Some((request_lsn, img))
             } else {
                 None
@@ -3038,7 +3030,7 @@ impl Timeline {
         .map_err(CompactionError::Other)?;
 
         let time_download_layer = timer.elapsed();
-        let mut timer = Instant::now();
+        let timer = Instant::now();
 
         // Step 2: Produce images+deltas.
         let mut accumulated_values = Vec::new();
@@ -3113,7 +3105,8 @@ impl Timeline {
         // Actually, we can decide not to write to the image layer at all at this point because
         // the key and LSN range are determined. However, to keep things simple here, we still
         // create this writer, and discard the writer in the end.
-        let mut time_to_first_kv_pair = None;
+
+        let mut keys_processed = 0;
 
         while let Some(((key, lsn, val), desc)) = merge_iter
             .next_with_trace()
@@ -3121,16 +3114,13 @@ impl Timeline {
             .context("failed to get next key-value pair")
             .map_err(CompactionError::Other)?
         {
-            if time_to_first_kv_pair.is_none() {
-                time_to_first_kv_pair = Some(timer.elapsed());
-                timer = Instant::now();
-            }
-
             if cancel.is_cancelled() {
                 return Err(CompactionError::ShuttingDown);
             }
 
+            keys_processed += 1;
             let should_yield = yield_for_l0
+                && keys_processed % 1000 == 0
                 && self
                     .l0_compaction_trigger
                     .notified()
@@ -3461,9 +3451,6 @@ impl Timeline {
         let time_final_phase = timer.elapsed();
 
         stat.time_final_phase_secs = time_final_phase.as_secs_f64();
-        stat.time_to_first_kv_pair_secs = time_to_first_kv_pair
-            .unwrap_or(Duration::ZERO)
-            .as_secs_f64();
         stat.time_main_loop_secs = time_main_loop.as_secs_f64();
         stat.time_acquire_lock_secs = time_acquire_lock.as_secs_f64();
         stat.time_download_layer_secs = time_download_layer.as_secs_f64();
@@ -3924,6 +3911,8 @@ impl CompactionLayer<Key> for OwnArc<DeltaLayer> {
     }
 }
 
+use crate::tenant::timeline::DeltaEntry;
+
 impl CompactionLayer<Key> for ResidentDeltaLayer {
     fn key_range(&self) -> &Range<Key> {
         &self.0.layer_desc().key_range

pageserver/src/tenant/timeline/delete.rs

@@ -410,13 +410,10 @@ impl DeleteTimelineFlow {
         // So indeed, the tenant manifest might refer to an offloaded timeline which has already been deleted.
         // However, we handle this case in tenant loading code so the next time we attach, the issue is
         // resolved.
-        tenant
-            .maybe_upload_tenant_manifest()
-            .await
-            .map_err(|err| match err {
-                TenantManifestError::Cancelled => DeleteTimelineError::Cancelled,
-                err => DeleteTimelineError::Other(err.into()),
-            })?;
+        tenant.store_tenant_manifest().await.map_err(|e| match e {
+            TenantManifestError::Cancelled => DeleteTimelineError::Cancelled,
+            _ => DeleteTimelineError::Other(e.into()),
+        })?;
 
         *guard = Self::Finished;
 

pageserver/src/tenant/timeline/layer_manager.rs

@@ -3,18 +3,17 @@ use std::sync::Arc;
 
 use anyhow::{Context, bail, ensure};
 use itertools::Itertools;
-use pageserver_api::keyspace::KeySpace;
 use pageserver_api::shard::TenantShardId;
 use tokio_util::sync::CancellationToken;
 use tracing::trace;
 use utils::id::TimelineId;
 use utils::lsn::{AtomicLsn, Lsn};
 
-use super::{LayerFringe, ReadableLayer, TimelineWriterState};
+use super::{ReadableLayer, TimelineWriterState};
 use crate::config::PageServerConf;
 use crate::context::RequestContext;
 use crate::metrics::TimelineMetrics;
-use crate::tenant::layer_map::{BatchedUpdates, LayerMap, SearchResult};
+use crate::tenant::layer_map::{BatchedUpdates, LayerMap};
 use crate::tenant::storage_layer::{
     AsLayerDesc, InMemoryLayer, Layer, LayerVisibilityHint, PersistentLayerDesc,
     PersistentLayerKey, ReadableLayerWeak, ResidentLayer,
@@ -39,7 +38,7 @@ impl Default for LayerManager {
 }
 
 impl LayerManager {
-    fn upgrade(&self, weak: ReadableLayerWeak) -> ReadableLayer {
+    pub(crate) fn upgrade(&self, weak: ReadableLayerWeak) -> ReadableLayer {
         match weak {
             ReadableLayerWeak::PersistentLayer(desc) => {
                 ReadableLayer::PersistentLayer(self.get_from_desc(&desc))
@@ -148,36 +147,6 @@ impl LayerManager {
         self.layers().keys().cloned().collect_vec()
     }
 
-    /// Update the [`LayerFringe`] of a read request
-    ///
-    /// Take a key space at a given LSN and query the layer map below each range
-    /// of the key space to find the next layers to visit.
-    pub(crate) fn update_search_fringe(
-        &self,
-        keyspace: &KeySpace,
-        cont_lsn: Lsn,
-        fringe: &mut LayerFringe,
-    ) -> Result<(), Shutdown> {
-        let map = self.layer_map()?;
-
-        for range in keyspace.ranges.iter() {
-            let results = map.range_search(range.clone(), cont_lsn);
-            results
-                .found
-                .into_iter()
-                .map(|(SearchResult { layer, lsn_floor }, keyspace_accum)| {
-                    (
-                        self.upgrade(layer),
-                        keyspace_accum.to_keyspace(),
-                        lsn_floor..cont_lsn,
-                    )
-                })
-                .for_each(|(layer, keyspace, lsn_range)| fringe.update(layer, keyspace, lsn_range));
-        }
-
-        Ok(())
-    }
-
     fn layers(&self) -> &HashMap<PersistentLayerKey, Layer> {
         use LayerManager::*;
         match self {

pageserver/src/tenant/timeline/offload.rs

@@ -111,7 +111,7 @@ pub(crate) async fn offload_timeline(
     // at the next restart attach it again.
     // For that to happen, we'd need to make the manifest reflect our *intended* state,
     // not our actual state of offloaded timelines.
-    tenant.maybe_upload_tenant_manifest().await?;
+    tenant.store_tenant_manifest().await?;
 
     tracing::info!("Timeline offload complete (remaining arc refcount: {remaining_refcount})");
 

pageserver/src/tenant/timeline/walreceiver/walreceiver_connection.rs

@@ -445,7 +445,7 @@ pub(super) async fn handle_walreceiver_connection(
                         .inspect_err(|err| {
                             // TODO: we can't differentiate cancellation errors with
                             // anyhow::Error, so just ignore it if we're cancelled.
-                            if !cancellation.is_cancelled() && !timeline.is_stopping() {
+                            if !cancellation.is_cancelled() {
                                 critical!("{err:?}")
                             }
                         })?;
@@ -577,7 +577,7 @@ pub(super) async fn handle_walreceiver_connection(
                             .inspect_err(|err| {
                                 // TODO: we can't differentiate cancellation errors with
                                 // anyhow::Error, so just ignore it if we're cancelled.
-                                if !cancellation.is_cancelled() && !timeline.is_stopping() {
+                                if !cancellation.is_cancelled() {
                                     critical!("{err:?}")
                                 }
                             })?;

pageserver/src/tenant/upload_queue.rs

@@ -302,7 +302,6 @@ pub struct UploadQueueStoppedDeletable {
     pub(super) deleted_at: SetDeletedFlagProgress,
 }
 
-#[allow(clippy::large_enum_variant, reason = "TODO")]
 pub enum UploadQueueStopped {
     Deletable(UploadQueueStoppedDeletable),
     Uninitialized,

pageserver/src/walredo.rs

@@ -136,16 +136,6 @@ macro_rules! bail {
     }
 }
 
-#[derive(Debug, Clone, Copy)]
-pub enum RedoAttemptType {
-    /// Used for the read path. Will fire critical errors and retry twice if failure.
-    ReadPage,
-    // Used for legacy compaction (only used in image compaction). Will fire critical errors and retry once if failure.
-    LegacyCompaction,
-    // Used for gc compaction. Will not fire critical errors and not retry.
-    GcCompaction,
-}
-
 ///
 /// Public interface of WAL redo manager
 ///
@@ -166,18 +156,11 @@ impl PostgresRedoManager {
         base_img: Option<(Lsn, Bytes)>,
         records: Vec<(Lsn, NeonWalRecord)>,
         pg_version: u32,
-        redo_attempt_type: RedoAttemptType,
     ) -> Result<Bytes, Error> {
         if records.is_empty() {
             bail!("invalid WAL redo request with no records");
         }
 
-        let max_retry_attempts = match redo_attempt_type {
-            RedoAttemptType::ReadPage => 2,
-            RedoAttemptType::LegacyCompaction => 1,
-            RedoAttemptType::GcCompaction => 0,
-        };
-
         let base_img_lsn = base_img.as_ref().map(|p| p.0).unwrap_or(Lsn::INVALID);
         let mut img = base_img.map(|p| p.1);
         let mut batch_neon = apply_neon::can_apply_in_neon(&records[0].1);
@@ -197,7 +180,6 @@ impl PostgresRedoManager {
                         &records[batch_start..i],
                         self.conf.wal_redo_timeout,
                         pg_version,
-                        max_retry_attempts,
                     )
                     .await
                 };
@@ -219,7 +201,6 @@ impl PostgresRedoManager {
                 &records[batch_start..],
                 self.conf.wal_redo_timeout,
                 pg_version,
-                max_retry_attempts,
             )
             .await
         }
@@ -443,11 +424,11 @@ impl PostgresRedoManager {
         records: &[(Lsn, NeonWalRecord)],
         wal_redo_timeout: Duration,
         pg_version: u32,
-        max_retry_attempts: u32,
     ) -> Result<Bytes, Error> {
         *(self.last_redo_at.lock().unwrap()) = Some(Instant::now());
 
         let (rel, blknum) = key.to_rel_block().context("invalid record")?;
+        const MAX_RETRY_ATTEMPTS: u32 = 1;
         let mut n_attempts = 0u32;
         loop {
             let base_img = &base_img;
@@ -505,7 +486,7 @@ impl PostgresRedoManager {
                 info!(n_attempts, "retried walredo succeeded");
             }
             n_attempts += 1;
-            if n_attempts > max_retry_attempts || result.is_ok() {
+            if n_attempts > MAX_RETRY_ATTEMPTS || result.is_ok() {
                 return result;
             }
         }
@@ -579,7 +560,6 @@ mod tests {
 
     use super::PostgresRedoManager;
     use crate::config::PageServerConf;
-    use crate::walredo::RedoAttemptType;
 
     #[tokio::test]
     async fn test_ping() {
@@ -613,7 +593,6 @@ mod tests {
                 None,
                 short_records(),
                 14,
-                RedoAttemptType::ReadPage,
             )
             .instrument(h.span())
             .await
@@ -642,7 +621,6 @@ mod tests {
                 None,
                 short_records(),
                 14,
-                RedoAttemptType::ReadPage,
             )
             .instrument(h.span())
             .await
@@ -664,7 +642,6 @@ mod tests {
                 None,
                 short_records(),
                 16, /* 16 currently produces stderr output on startup, which adds a nice extra edge */
-                RedoAttemptType::ReadPage,
             )
             .instrument(h.span())
             .await

pageserver/src/walredo/apply_neon.rs

@@ -276,7 +276,6 @@ pub(crate) fn apply_in_neon(
             append,
             clear,
             will_init,
-            only_if,
         } => {
             use bytes::BufMut;
             if *will_init {
@@ -289,13 +288,6 @@ pub(crate) fn apply_in_neon(
             if *clear {
                 page.clear();
             }
-            if let Some(only_if) = only_if {
-                if page != only_if.as_bytes() {
-                    return Err(anyhow::anyhow!(
-                        "the current image does not match the expected image, cannot append"
-                    ));
-                }
-            }
             page.put_slice(append.as_bytes());
         }
     }

pgxn/neon/Makefile

@@ -4,7 +4,6 @@
 MODULE_big = neon
 OBJS = \
 	$(WIN32RES) \
-	communicator.o \
 	extension_server.o \
 	file_cache.o \
 	hll.o \

pgxn/neon/communicator.h

@@ -1,48 +0,0 @@
-/*-------------------------------------------------------------------------
- *
- * communicator.h
- *	  internal interface for communicating with remote pageservers
- *
- *
- * Portions Copyright (c) 1996-2021, PostgreSQL Global Development Group
- * Portions Copyright (c) 1994, Regents of the University of California
- *
- *-------------------------------------------------------------------------
- */
-#ifndef COMMUNICATOR_h
-#define COMMUNICATOR_h
-
-#include "neon_pgversioncompat.h"
-
-#include "storage/buf_internals.h"
-
-#include "pagestore_client.h"
-
-/* initialization at postmaster startup */
-extern void pg_init_communicator(void);
-
-/* initialization at backend startup */
-extern void communicator_init(void);
-
-extern bool communicator_exists(NRelFileInfo rinfo, ForkNumber forkNum,
-								neon_request_lsns *request_lsns);
-extern BlockNumber communicator_nblocks(NRelFileInfo rinfo, ForkNumber forknum,
-										neon_request_lsns *request_lsns);
-extern int64 communicator_dbsize(Oid dbNode, neon_request_lsns *request_lsns);
-extern void communicator_read_at_lsnv(NRelFileInfo rinfo, ForkNumber forkNum,
-									  BlockNumber base_blockno, neon_request_lsns *request_lsns,
-									  void **buffers, BlockNumber nblocks, const bits8 *mask);
-extern int communicator_prefetch_lookupv(NRelFileInfo rinfo, ForkNumber forknum, BlockNumber blocknum,
-										 neon_request_lsns *lsns,
-										 BlockNumber nblocks, void **buffers, bits8 *mask);
-extern void communicator_prefetch_register_bufferv(BufferTag tag, neon_request_lsns *frlsns,
-												   BlockNumber nblocks, const bits8 *mask);
-extern int communicator_read_slru_segment(SlruKind kind, int64 segno,
-										  neon_request_lsns *request_lsns,
-										  void *buffer);
-
-extern void communicator_reconfigure_timeout_if_needed(void);
-extern void communicator_prefetch_pump_state(bool IsHandlingInterrupts);
-
-
-#endif

pgxn/neon/file_cache.c

@@ -21,6 +21,7 @@
 #include "access/xlog.h"
 #include "funcapi.h"
 #include "miscadmin.h"
+#include "pagestore_client.h"
 #include "common/hashfn.h"
 #include "pgstat.h"
 #include "port/pg_iovec.h"
@@ -42,7 +43,6 @@
 
 #include "hll.h"
 #include "bitmap.h"
-#include "file_cache.h"
 #include "neon.h"
 #include "neon_lwlsncache.h"
 #include "neon_perf_counters.h"
@@ -1563,12 +1563,8 @@ local_cache_pages(PG_FUNCTION_ARGS)
 				hash_seq_init(&status, lfc_hash);
 				while ((entry = hash_seq_search(&status)) != NULL)
 				{
-					/* Skip hole tags */
-					if (NInfoGetRelNumber(BufTagGetNRelFileInfo(entry->key)) != 0)
-					{
-						for (int i = 0; i < BLOCKS_PER_CHUNK; i++)
-							n_pages += GET_STATE(entry, i) == AVAILABLE;
-					}
+					for (int i = 0; i < BLOCKS_PER_CHUNK; i++)
+						n_pages += GET_STATE(entry, i) == AVAILABLE;
 				}
 			}
 		}
@@ -1596,19 +1592,16 @@ local_cache_pages(PG_FUNCTION_ARGS)
 			{
 				for (int i = 0; i < BLOCKS_PER_CHUNK; i++)
 				{
-					if (NInfoGetRelNumber(BufTagGetNRelFileInfo(entry->key)) != 0)
+					if (GET_STATE(entry, i) == AVAILABLE)
 					{
-						if (GET_STATE(entry, i) == AVAILABLE)
-						{
-							fctx->record[n].pageoffs = entry->offset * BLOCKS_PER_CHUNK + i;
-							fctx->record[n].relfilenode = NInfoGetRelNumber(BufTagGetNRelFileInfo(entry->key));
-							fctx->record[n].reltablespace = NInfoGetSpcOid(BufTagGetNRelFileInfo(entry->key));
-							fctx->record[n].reldatabase = NInfoGetDbOid(BufTagGetNRelFileInfo(entry->key));
-							fctx->record[n].forknum = entry->key.forkNum;
-							fctx->record[n].blocknum = entry->key.blockNum + i;
-							fctx->record[n].accesscount = entry->access_count;
-							n += 1;
-						}
+						fctx->record[n].pageoffs = entry->offset * BLOCKS_PER_CHUNK + i;
+						fctx->record[n].relfilenode = NInfoGetRelNumber(BufTagGetNRelFileInfo(entry->key));
+						fctx->record[n].reltablespace = NInfoGetSpcOid(BufTagGetNRelFileInfo(entry->key));
+						fctx->record[n].reldatabase = NInfoGetDbOid(BufTagGetNRelFileInfo(entry->key));
+						fctx->record[n].forknum = entry->key.forkNum;
+						fctx->record[n].blocknum = entry->key.blockNum + i;
+						fctx->record[n].accesscount = entry->access_count;
+						n += 1;
 					}
 				}
 			}

pgxn/neon/file_cache.h

@@ -1,52 +0,0 @@
-/*-------------------------------------------------------------------------
- *
- * file_cache.h
- *	  Local File Cache definitions
- *
- * Portions Copyright (c) 1996-2021, PostgreSQL Global Development Group
- * Portions Copyright (c) 1994, Regents of the University of California
- *
- *-------------------------------------------------------------------------
- */
-#ifndef FILE_CACHE_h
-#define FILE_CACHE_h
-
-#include "neon_pgversioncompat.h"
-
-/* GUCs */
-extern bool lfc_store_prefetch_result;
-
-/* functions for local file cache */
-extern void lfc_writev(NRelFileInfo rinfo, ForkNumber forkNum,
-					   BlockNumber blkno, const void *const *buffers,
-					   BlockNumber nblocks);
-/* returns number of blocks read, with one bit set in *read for each  */
-extern int lfc_readv_select(NRelFileInfo rinfo, ForkNumber forkNum,
-							BlockNumber blkno, void **buffers,
-							BlockNumber nblocks, bits8 *mask);
-
-extern bool lfc_cache_contains(NRelFileInfo rinfo, ForkNumber forkNum,
-							   BlockNumber blkno);
-extern int lfc_cache_containsv(NRelFileInfo rinfo, ForkNumber forkNum,
-							   BlockNumber blkno, int nblocks, bits8 *bitmap);
-extern void lfc_init(void);
-extern bool lfc_prefetch(NRelFileInfo rinfo, ForkNumber forknum, BlockNumber blkno,
-						 const void* buffer, XLogRecPtr lsn);
-
-
-static inline bool
-lfc_read(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno,
-		 void *buffer)
-{
-	bits8		rv = 0;
-	return lfc_readv_select(rinfo, forkNum, blkno, &buffer, 1, &rv) == 1;
-}
-
-static inline void
-lfc_write(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno,
-		  const void *buffer)
-{
-	return lfc_writev(rinfo, forkNum, blkno, &buffer, 1);
-}
-
-#endif							/* FILE_CACHE_H */

pgxn/neon/libpagestore.c

@@ -65,9 +65,6 @@ static const struct config_enum_entry neon_compute_modes[] = {
 /* GUCs */
 char	   *neon_timeline;
 char	   *neon_tenant;
-char	   *neon_project_id;
-char	   *neon_branch_id;
-char	   *neon_endpoint_id;
 int32		max_cluster_size;
 char	   *page_server_connstring;
 char	   *neon_auth_token;
@@ -1355,31 +1352,6 @@ pg_init_libpagestore(void)
 							   0,	/* no flags required */
 							   check_neon_id, NULL, NULL);
 
-	DefineCustomStringVariable("neon.project_id",
-							   "Neon project_id the server is running on",
-							   NULL,
-							   &neon_project_id,
-							   "",
-							   PGC_POSTMASTER,
-							   0,	/* no flags required */
-							   NULL, NULL, NULL);
-	DefineCustomStringVariable("neon.branch_id",
-							   "Neon branch_id the server is running on",
-							   NULL,
-							   &neon_branch_id,
-							   "",
-							   PGC_POSTMASTER,
-							   0,	/* no flags required */
-							   NULL, NULL, NULL);
-	DefineCustomStringVariable("neon.endpoint_id",
-							   "Neon endpoint_id the server is running on",
-							   NULL,
-							   &neon_endpoint_id,
-							   "",
-							   PGC_POSTMASTER,
-							   0,	/* no flags required */
-							   NULL, NULL, NULL);
-
 	DefineCustomIntVariable("neon.stripe_size",
 							"sharding stripe size",
 							NULL,
@@ -1503,4 +1475,6 @@ pg_init_libpagestore(void)
 	}
 
 	memset(page_servers, 0, sizeof(page_servers));
+
+	lfc_init();
 }

pgxn/neon/neon.c

@@ -28,9 +28,7 @@
 #include "utils/guc.h"
 #include "utils/guc_tables.h"
 
-#include "communicator.h"
 #include "extension_server.h"
-#include "file_cache.h"
 #include "neon.h"
 #include "neon_lwlsncache.h"
 #include "control_plane_connector.h"
@@ -436,11 +434,10 @@ _PG_init(void)
 #endif
 
 	pg_init_libpagestore();
-	lfc_init();
 	pg_init_walproposer();
 	init_lwlsncache();
 
-	pg_init_communicator();
+	pagestore_smgr_init();
 	Custom_XLogReaderRoutines = NeonOnDemandXLogReaderRoutines;
 
 	InitUnstableExtensionsSupport();

pgxn/neon/neon.h

@@ -47,18 +47,9 @@ extern uint32		WAIT_EVENT_NEON_WAL_DL;
 #define WAIT_EVENT_NEON_WAL_DL			WAIT_EVENT_WAL_READ
 #endif
 
-
-#define NEON_TAG "[NEON_SMGR] "
-#define neon_log(tag, fmt, ...) ereport(tag,                                  \
-										(errmsg(NEON_TAG fmt, ##__VA_ARGS__), \
-										 errhidestmt(true), errhidecontext(true), errposition(0), internalerrposition(0)))
-#define neon_shard_log(shard_no, tag, fmt, ...) ereport(tag,	\
-														(errmsg(NEON_TAG "[shard %d] " fmt, shard_no, ##__VA_ARGS__), \
-														 errhidestmt(true), errhidecontext(true), errposition(0), internalerrposition(0)))
-
-
 extern void pg_init_libpagestore(void);
 extern void pg_init_walproposer(void);
+extern void pagestore_smgr_init(void);
 
 extern uint64 BackpressureThrottlingTime(void);
 extern void SetNeonCurrentClusterSize(uint64 size);

pgxn/neon/pagestore_client.h

@@ -58,6 +58,14 @@ typedef struct
 
 #define messageTag(m) (((const NeonMessage *)(m))->tag)
 
+#define NEON_TAG "[NEON_SMGR] "
+#define neon_log(tag, fmt, ...) ereport(tag,                                  \
+										(errmsg(NEON_TAG fmt, ##__VA_ARGS__), \
+										 errhidestmt(true), errhidecontext(true), errposition(0), internalerrposition(0)))
+#define neon_shard_log(shard_no, tag, fmt, ...) ereport(tag,	\
+														(errmsg(NEON_TAG "[shard %d] " fmt, shard_no, ##__VA_ARGS__), \
+														 errhidestmt(true), errhidecontext(true), errposition(0), internalerrposition(0)))
+
 /* SLRUs downloadable from page server */
 typedef enum {
 	SLRU_CLOG,
@@ -226,6 +234,7 @@ extern char *neon_timeline;
 extern char *neon_tenant;
 extern int32 max_cluster_size;
 extern int  neon_protocol_version;
+extern bool lfc_store_prefetch_result;
 
 extern shardno_t get_shard_number(BufferTag* tag);
 
@@ -233,7 +242,6 @@ extern const f_smgr *smgr_neon(ProcNumber backend, NRelFileInfo rinfo);
 extern void smgr_init_neon(void);
 extern void readahead_buffer_resize(int newsize, void *extra);
 
-
 /*
  * LSN values associated with each request to the pageserver
  */
@@ -270,10 +278,6 @@ extern PGDLLEXPORT void neon_read_at_lsn(NRelFileInfo rnode, ForkNumber forkNum,
 										 neon_request_lsns request_lsns, void *buffer);
 extern int64 neon_dbsize(Oid dbNode);
 
-extern void neon_get_request_lsns(NRelFileInfo rinfo, ForkNumber forknum,
-								  BlockNumber blkno, neon_request_lsns *output,
-								  BlockNumber nblocks);
-
 /* utils for neon relsize cache */
 extern void relsize_hash_init(void);
 extern bool get_cached_relsize(NRelFileInfo rinfo, ForkNumber forknum, BlockNumber *size);
@@ -281,4 +285,37 @@ extern void set_cached_relsize(NRelFileInfo rinfo, ForkNumber forknum, BlockNumb
 extern void update_cached_relsize(NRelFileInfo rinfo, ForkNumber forknum, BlockNumber size);
 extern void forget_cached_relsize(NRelFileInfo rinfo, ForkNumber forknum);
 
+/* functions for local file cache */
+extern void lfc_writev(NRelFileInfo rinfo, ForkNumber forkNum,
+					   BlockNumber blkno, const void *const *buffers,
+					   BlockNumber nblocks);
+/* returns number of blocks read, with one bit set in *read for each  */
+extern int lfc_readv_select(NRelFileInfo rinfo, ForkNumber forkNum,
+							BlockNumber blkno, void **buffers,
+							BlockNumber nblocks, bits8 *mask);
+
+extern bool lfc_cache_contains(NRelFileInfo rinfo, ForkNumber forkNum,
+							   BlockNumber blkno);
+extern int lfc_cache_containsv(NRelFileInfo rinfo, ForkNumber forkNum,
+							   BlockNumber blkno, int nblocks, bits8 *bitmap);
+extern void lfc_init(void);
+extern bool lfc_prefetch(NRelFileInfo rinfo, ForkNumber forknum, BlockNumber blkno,
+						 const void* buffer, XLogRecPtr lsn);
+
+
+static inline bool
+lfc_read(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno,
+		 void *buffer)
+{
+	bits8		rv = 0;
+	return lfc_readv_select(rinfo, forkNum, blkno, &buffer, 1, &rv) == 1;
+}
+
+static inline void
+lfc_write(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno,
+		  const void *buffer)
+{
+	return lfc_writev(rinfo, forkNum, blkno, &buffer, 1);
+}
+
 #endif							/* PAGESTORE_CLIENT_H */

pgxn/neon/walproposer.c

@@ -99,9 +99,6 @@ WalProposerCreate(WalProposerConfig *config, walproposer_api api)
 	wp->config = config;
 	wp->api = api;
 	wp->state = WPS_COLLECTING_TERMS;
-	wp->mconf.generation = INVALID_GENERATION;
-	wp->mconf.members.len = 0;
-	wp->mconf.new_members.len = 0;
 
 	wp_log(LOG, "neon.safekeepers=%s", wp->config->safekeepers_list);
 
@@ -173,8 +170,6 @@ WalProposerCreate(WalProposerConfig *config, walproposer_api api)
 
 	if (wp->config->proto_version != 2 && wp->config->proto_version != 3)
 		wp_log(FATAL, "unsupported safekeeper protocol version %d", wp->config->proto_version);
-	if (wp->safekeepers_generation > INVALID_GENERATION && wp->config->proto_version < 3)
-		wp_log(FATAL, "enabling generations requires protocol version 3");
 	wp_log(LOG, "using safekeeper protocol version %d", wp->config->proto_version);
 
 	/* Fill the greeting package */
@@ -219,7 +214,7 @@ WalProposerFree(WalProposer *wp)
 static bool
 WalProposerGenerationsEnabled(WalProposer *wp)
 {
-	return wp->safekeepers_generation != INVALID_GENERATION;
+	return wp->safekeepers_generation != 0;
 }
 
 /*
@@ -728,176 +723,13 @@ SendProposerGreeting(Safekeeper *sk)
 	BlockingWrite(sk, sk->outbuf.data, sk->outbuf.len, SS_HANDSHAKE_RECV);
 }
 
-/*
- * Assuming `sk` sent its node id, find such member(s) in wp->mconf and set ptr in
- * members_safekeepers & new_members_safekeepers to sk.
- */
-static void
-UpdateMemberSafekeeperPtr(WalProposer *wp, Safekeeper *sk)
-{
-	/* members_safekeepers etc are fixed size, sanity check mconf size */
-	if (wp->mconf.members.len > MAX_SAFEKEEPERS)
-		wp_log(FATAL, "too many members %d in mconf", wp->mconf.members.len);
-	if (wp->mconf.new_members.len > MAX_SAFEKEEPERS)
-		wp_log(FATAL, "too many new_members %d in mconf", wp->mconf.new_members.len);
-
-	/* node id is not known until greeting is received */
-	if (sk->state < SS_WAIT_VOTING)
-		return;
-
-	/* 0 is assumed to be invalid node id, should never happen */
-	if (sk->greetResponse.nodeId == 0)
-	{
-		wp_log(WARNING, "safekeeper %s:%s sent zero node id", sk->host, sk->port);
-		return;
-	}
-
-	for (uint32 i = 0; i < wp->mconf.members.len; i++)
-	{
-		SafekeeperId *sk_id = &wp->mconf.members.m[i];
-
-		if (wp->mconf.members.m[i].node_id == sk->greetResponse.nodeId)
-		{
-			/*
-			 * If mconf or list of safekeepers to connect to changed (the
-			 * latter always currently goes through restart though),
-			 * ResetMemberSafekeeperPtrs is expected to be called before
-			 * UpdateMemberSafekeeperPtr. So, other value suggests that we are
-			 * connected to the same sk under different host name, complain
-			 * about that.
-			 */
-			if (wp->members_safekeepers[i] != NULL && wp->members_safekeepers[i] != sk)
-			{
-				wp_log(WARNING, "safekeeper {id = %lu, ep = %s:%u } in members[%u] is already mapped to connection slot %lu",
-					   sk_id->node_id, sk_id->host, sk_id->port, i, wp->members_safekeepers[i] - wp->safekeeper);
-			}
-			wp_log(LOG, "safekeeper {id = %lu, ep = %s:%u } in members[%u] mapped to connection slot %lu",
-				   sk_id->node_id, sk_id->host, sk_id->port, i, sk - wp->safekeeper);
-			wp->members_safekeepers[i] = sk;
-		}
-	}
-	/* repeat for new_members */
-	for (uint32 i = 0; i < wp->mconf.new_members.len; i++)
-	{
-		SafekeeperId *sk_id = &wp->mconf.new_members.m[i];
-
-		if (wp->mconf.new_members.m[i].node_id == sk->greetResponse.nodeId)
-		{
-			if (wp->new_members_safekeepers[i] != NULL && wp->new_members_safekeepers[i] != sk)
-			{
-				wp_log(WARNING, "safekeeper {id = %lu, ep = %s:%u } in new_members[%u] is already mapped to connection slot %lu",
-					   sk_id->node_id, sk_id->host, sk_id->port, i, wp->new_members_safekeepers[i] - wp->safekeeper);
-			}
-			wp_log(LOG, "safekeeper {id = %lu, ep = %s:%u } in new_members[%u] mapped to connection slot %lu",
-				   sk_id->node_id, sk_id->host, sk_id->port, i, sk - wp->safekeeper);
-			wp->new_members_safekeepers[i] = sk;
-		}
-	}
-}
-
-/*
- * Reset wp->members_safekeepers & new_members_safekeepers and refill them.
- * Called after wp changes mconf.
- */
-static void
-ResetMemberSafekeeperPtrs(WalProposer *wp)
-{
-	memset(&wp->members_safekeepers, 0, sizeof(Safekeeper *) * MAX_SAFEKEEPERS);
-	memset(&wp->new_members_safekeepers, 0, sizeof(Safekeeper *) * MAX_SAFEKEEPERS);
-	for (int i = 0; i < wp->n_safekeepers; i++)
-	{
-		if (wp->safekeeper[i].state >= SS_WAIT_VOTING)
-			UpdateMemberSafekeeperPtr(wp, &wp->safekeeper[i]);
-	}
-}
-
-static uint32
-MsetQuorum(MemberSet *mset)
-{
-	Assert(mset->len > 0);
-	return mset->len / 2 + 1;
-}
-
-/* Does n forms quorum in mset? */
-static bool
-MsetHasQuorum(MemberSet *mset, uint32 n)
-{
-	return n >= MsetQuorum(mset);
-}
-
-/*
- * TermsCollected helper for a single member set `mset`.
- *
- * `msk` is the member -> safekeeper mapping for mset, i.e. members_safekeepers
- * or new_members_safekeepers.
- */
-static bool
-TermsCollectedMset(WalProposer *wp, MemberSet *mset, Safekeeper **msk, StringInfo s)
-{
-	uint32		n_greeted = 0;
-
-	for (uint32 i = 0; i < wp->mconf.members.len; i++)
-	{
-		Safekeeper *sk = msk[i];
-
-		if (sk != NULL && sk->state == SS_WAIT_VOTING)
-		{
-			if (n_greeted > 0)
-				appendStringInfoString(s, ", ");
-			appendStringInfo(s, "{id = %lu, ep = %s:%s}", sk->greetResponse.nodeId, sk->host, sk->port);
-			n_greeted++;
-		}
-	}
-	appendStringInfo(s, ", %u/%u total", n_greeted, mset->len);
-	return MsetHasQuorum(mset, n_greeted);
-}
-
 /*
  * Have we received greeting from enough (quorum) safekeepers to start voting?
  */
 static bool
 TermsCollected(WalProposer *wp)
 {
-	StringInfoData s;			/* str for logging */
-	bool		collected = false;
-
-	/* legacy: generations disabled */
-	if (!WalProposerGenerationsEnabled(wp) && wp->mconf.generation == INVALID_GENERATION)
-	{
-		collected = wp->n_connected >= wp->quorum;
-		if (collected)
-		{
-			wp->propTerm++;
-			wp_log(LOG, "walproposer connected to quorum (%d) safekeepers, propTerm=" INT64_FORMAT ", starting voting", wp->quorum, wp->propTerm);
-		}
-		return collected;
-	}
-
-	/*
-	 * With generations enabled, we start campaign only when 1) some mconf is
-	 * actually received 2) we have greetings from majority of members as well
-	 * as from majority of new_members if it exists.
-	 */
-	if (wp->mconf.generation == INVALID_GENERATION)
-		return false;
-
-	initStringInfo(&s);
-	appendStringInfoString(&s, "mset greeters: ");
-	if (!TermsCollectedMset(wp, &wp->mconf.members, wp->members_safekeepers, &s))
-		goto res;
-	if (wp->mconf.new_members.len > 0)
-	{
-		appendStringInfoString(&s, ", new_mset greeters: ");
-		if (!TermsCollectedMset(wp, &wp->mconf.new_members, wp->new_members_safekeepers, &s))
-			goto res;
-	}
-	wp->propTerm++;
-	wp_log(LOG, "walproposer connected to quorum of safekeepers: %s, propTerm=" INT64_FORMAT ", starting voting", s.data, wp->propTerm);
-	collected = true;
-
-res:
-	pfree(s.data);
-	return collected;
+	return wp->n_connected >= wp->quorum;
 }
 
 static void
@@ -921,41 +753,13 @@ RecvAcceptorGreeting(Safekeeper *sk)
 	pfree(mconf_toml);
 
 	/*
-	 * Adopt mconf of safekeepers if it is higher.
+	 * Adopt mconf of safekeepers if it is higher. TODO: mconf change should
+	 * restart wp if it started voting.
 	 */
 	if (sk->greetResponse.mconf.generation > wp->mconf.generation)
 	{
-		/* sanity check before adopting, should never happen */
-		if (sk->greetResponse.mconf.members.len == 0)
-		{
-			wp_log(FATAL, "mconf %u has zero members", sk->greetResponse.mconf.generation);
-		}
-
-		/*
-		 * If we at least started campaign, restart wp to get elected in the
-		 * new mconf. Note: in principle once wp is already elected
-		 * re-election is not required, but being conservative here is not
-		 * bad.
-		 *
-		 * TODO: put mconf to shmem to immediately pick it up on start,
-		 * otherwise if some safekeeper(s) misses latest mconf and gets
-		 * connected the first, it may cause redundant restarts here.
-		 *
-		 * More generally, it would be nice to restart walproposer (wiping
-		 * election state) without restarting the process. In particular, that
-		 * would allow sync-safekeepers not to die here if it intersected with
-		 * sk migration (as well as remove 1s delay).
-		 *
-		 * Note that assign_neon_safekeepers also currently restarts the
-		 * process, so during normal migration walproposer may restart twice.
-		 */
-		if (wp->state >= WPS_CAMPAIGN)
-		{
-			wp_log(FATAL, "restarting to adopt mconf generation %d", sk->greetResponse.mconf.generation);
-		}
 		MembershipConfigurationFree(&wp->mconf);
 		MembershipConfigurationCopy(&sk->greetResponse.mconf, &wp->mconf);
-		ResetMemberSafekeeperPtrs(wp);
 		/* full conf was just logged above */
 		wp_log(LOG, "changed mconf to generation %u", wp->mconf.generation);
 	}
@@ -963,9 +767,6 @@ RecvAcceptorGreeting(Safekeeper *sk)
 	/* Protocol is all good, move to voting. */
 	sk->state = SS_WAIT_VOTING;
 
-	/* In greeting safekeeper sent its id; update mappings accordingly. */
-	UpdateMemberSafekeeperPtr(wp, sk);
-
 	/*
 	 * Note: it would be better to track the counter on per safekeeper basis,
 	 * but at worst walproposer would restart with 'term rejected', so leave
@@ -977,9 +778,12 @@ RecvAcceptorGreeting(Safekeeper *sk)
 		/* We're still collecting terms from the majority. */
 		wp->propTerm = Max(sk->greetResponse.term, wp->propTerm);
 
-		/* Quorum is acquired, prepare the vote request. */
+		/* Quorum is acquried, prepare the vote request. */
 		if (TermsCollected(wp))
 		{
+			wp->propTerm++;
+			wp_log(LOG, "proposer connected to quorum (%d) safekeepers, propTerm=" INT64_FORMAT, wp->quorum, wp->propTerm);
+
 			wp->state = WPS_CAMPAIGN;
 			wp->voteRequest.pam.tag = 'v';
 			wp->voteRequest.generation = wp->mconf.generation;
@@ -1028,8 +832,8 @@ SendVoteRequest(Safekeeper *sk)
 					   &sk->outbuf, wp->config->proto_version);
 
 	/* We have quorum for voting, send our vote request */
-	wp_log(LOG, "requesting vote from sk {id = %lu, ep = %s:%s} for generation %u term " UINT64_FORMAT,
-		   sk->greetResponse.nodeId, sk->host, sk->port, wp->voteRequest.generation, wp->voteRequest.term);
+	wp_log(LOG, "requesting vote from %s:%s for generation %u term " UINT64_FORMAT, sk->host, sk->port,
+		   wp->voteRequest.generation, wp->voteRequest.term);
 	/* On failure, logging & resetting is handled */
 	BlockingWrite(sk, sk->outbuf.data, sk->outbuf.len, SS_WAIT_VERDICT);
 	/* If successful, wait for read-ready with SS_WAIT_VERDICT */
@@ -1047,8 +851,8 @@ RecvVoteResponse(Safekeeper *sk)
 		return;
 
 	wp_log(LOG,
-		   "got VoteResponse from sk {id = %lu, ep = %s:%s}, generation=%u, term=%lu, voteGiven=%u, last_log_term=" UINT64_FORMAT ", flushLsn=%X/%X, truncateLsn=%X/%X",
-		   sk->greetResponse.nodeId, sk->host, sk->port, sk->voteResponse.generation, sk->voteResponse.term,
+		   "got VoteResponse from acceptor %s:%s, generation=%u, term=%lu, voteGiven=%u, last_log_term=" UINT64_FORMAT ", flushLsn=%X/%X, truncateLsn=%X/%X",
+		   sk->host, sk->port, sk->voteResponse.generation, sk->voteResponse.term,
 		   sk->voteResponse.voteGiven,
 		   GetHighestTerm(&sk->voteResponse.termHistory),
 		   LSN_FORMAT_ARGS(sk->voteResponse.flushLsn),
@@ -1095,53 +899,6 @@ RecvVoteResponse(Safekeeper *sk)
 	}
 }
 
-/*
- * VotesCollected helper for a single member set `mset`.
- *
- * `msk` is the member -> safekeeper mapping for mset, i.e. members_safekeepers
- * or new_members_safekeepers.
- */
-static bool
-VotesCollectedMset(WalProposer *wp, MemberSet *mset, Safekeeper **msk, StringInfo s)
-{
-	uint32		n_votes = 0;
-
-	for (uint32 i = 0; i < wp->mconf.members.len; i++)
-	{
-		Safekeeper *sk = msk[i];
-
-		if (sk != NULL && sk->state == SS_WAIT_ELECTED)
-		{
-			Assert(sk->voteResponse.voteGiven);
-
-			/*
-			 * Find the highest vote. NULL check is for the legacy case where
-			 * safekeeper might be not initialized with LSN at all and return
-			 * 0 LSN in the vote response; we still want to set donor to
-			 * something in this case.
-			 */
-			if (GetLastLogTerm(sk) > wp->donorLastLogTerm ||
-				(GetLastLogTerm(sk) == wp->donorLastLogTerm &&
-				 sk->voteResponse.flushLsn > wp->propTermStartLsn) ||
-				wp->donor == NULL)
-			{
-				wp->donorLastLogTerm = GetLastLogTerm(sk);
-				wp->propTermStartLsn = sk->voteResponse.flushLsn;
-				wp->donor = sk;
-			}
-			wp->truncateLsn = Max(wp->safekeeper[i].voteResponse.truncateLsn, wp->truncateLsn);
-
-			if (n_votes > 0)
-				appendStringInfoString(s, ", ");
-			appendStringInfo(s, "{id = %lu, ep = %s:%s}", sk->greetResponse.nodeId, sk->host, sk->port);
-			n_votes++;
-		}
-	}
-	appendStringInfo(s, ", %u/%u total", n_votes, mset->len);
-	return MsetHasQuorum(mset, n_votes);
-}
-
-
 /*
  * Checks if enough votes has been collected to get elected and if that's the
  * case finds the highest vote, setting donor, donorLastLogTerm,
@@ -1150,8 +907,7 @@ VotesCollectedMset(WalProposer *wp, MemberSet *mset, Safekeeper **msk, StringInf
 static bool
 VotesCollected(WalProposer *wp)
 {
-	StringInfoData s;			/* str for logging */
-	bool		collected = false;
+	int			n_ready = 0;
 
 	/* assumed to be called only when not elected yet */
 	Assert(wp->state == WPS_CAMPAIGN);
@@ -1160,62 +916,25 @@ VotesCollected(WalProposer *wp)
 	wp->donorLastLogTerm = 0;
 	wp->truncateLsn = InvalidXLogRecPtr;
 
-	/* legacy: generations disabled */
-	if (!WalProposerGenerationsEnabled(wp) && wp->mconf.generation == INVALID_GENERATION)
+	for (int i = 0; i < wp->n_safekeepers; i++)
 	{
-		int			n_ready = 0;
-
-		for (int i = 0; i < wp->n_safekeepers; i++)
+		if (wp->safekeeper[i].state == SS_WAIT_ELECTED)
 		{
-			if (wp->safekeeper[i].state == SS_WAIT_ELECTED)
+			n_ready++;
+
+			if (GetLastLogTerm(&wp->safekeeper[i]) > wp->donorLastLogTerm ||
+				(GetLastLogTerm(&wp->safekeeper[i]) == wp->donorLastLogTerm &&
+				 wp->safekeeper[i].voteResponse.flushLsn > wp->propTermStartLsn))
 			{
-				n_ready++;
-
-				if (GetLastLogTerm(&wp->safekeeper[i]) > wp->donorLastLogTerm ||
-					(GetLastLogTerm(&wp->safekeeper[i]) == wp->donorLastLogTerm &&
-					 wp->safekeeper[i].voteResponse.flushLsn > wp->propTermStartLsn) ||
-					wp->donor == NULL)
-				{
-					wp->donorLastLogTerm = GetLastLogTerm(&wp->safekeeper[i]);
-					wp->propTermStartLsn = wp->safekeeper[i].voteResponse.flushLsn;
-					wp->donor = &wp->safekeeper[i];
-				}
-				wp->truncateLsn = Max(wp->safekeeper[i].voteResponse.truncateLsn, wp->truncateLsn);
+				wp->donorLastLogTerm = GetLastLogTerm(&wp->safekeeper[i]);
+				wp->propTermStartLsn = wp->safekeeper[i].voteResponse.flushLsn;
+				wp->donor = i;
 			}
+			wp->truncateLsn = Max(wp->safekeeper[i].voteResponse.truncateLsn, wp->truncateLsn);
 		}
-		collected = n_ready >= wp->quorum;
-		if (collected)
-		{
-			wp_log(LOG, "walproposer elected with %d/%d votes", n_ready, wp->n_safekeepers);
-		}
-		return collected;
 	}
 
-	/*
-	 * if generations are enabled we're expected to get to voting only when
-	 * mconf is established.
-	 */
-	Assert(wp->mconf.generation != INVALID_GENERATION);
-
-	/*
-	 * We must get votes from both msets if both are present.
-	 */
-	initStringInfo(&s);
-	appendStringInfoString(&s, "mset voters: ");
-	if (!VotesCollectedMset(wp, &wp->mconf.members, wp->members_safekeepers, &s))
-		goto res;
-	if (wp->mconf.new_members.len > 0)
-	{
-		appendStringInfoString(&s, ", new_mset voters: ");
-		if (!VotesCollectedMset(wp, &wp->mconf.new_members, wp->new_members_safekeepers, &s))
-			goto res;
-	}
-	wp_log(LOG, "walproposer elected, %s", s.data);
-	collected = true;
-
-res:
-	pfree(s.data);
-	return collected;
+	return n_ready >= wp->quorum;
 }
 
 /*
@@ -1236,7 +955,7 @@ HandleElectedProposer(WalProposer *wp)
 	 * that only for logical replication (and switching logical walsenders to
 	 * neon_walreader is a todo.)
 	 */
-	if (!wp->api.recovery_download(wp, wp->donor))
+	if (!wp->api.recovery_download(wp, &wp->safekeeper[wp->donor]))
 	{
 		wp_log(FATAL, "failed to download WAL for logical replicaiton");
 	}
@@ -1359,7 +1078,7 @@ ProcessPropStartPos(WalProposer *wp)
 	/*
 	 * Proposer's term history is the donor's + its own entry.
 	 */
-	dth = &wp->donor->voteResponse.termHistory;
+	dth = &wp->safekeeper[wp->donor].voteResponse.termHistory;
 	wp->propTermHistory.n_entries = dth->n_entries + 1;
 	wp->propTermHistory.entries = palloc(sizeof(TermSwitchEntry) * wp->propTermHistory.n_entries);
 	if (dth->n_entries > 0)
@@ -1367,10 +1086,11 @@ ProcessPropStartPos(WalProposer *wp)
 	wp->propTermHistory.entries[wp->propTermHistory.n_entries - 1].term = wp->propTerm;
 	wp->propTermHistory.entries[wp->propTermHistory.n_entries - 1].lsn = wp->propTermStartLsn;
 
-	wp_log(LOG, "walproposer elected in term " UINT64_FORMAT ", epochStartLsn %X/%X, donor %s:%s, truncate_lsn %X/%X",
+	wp_log(LOG, "got votes from majority (%d) of nodes, term " UINT64_FORMAT ", epochStartLsn %X/%X, donor %s:%s, truncate_lsn %X/%X",
+		   wp->quorum,
 		   wp->propTerm,
 		   LSN_FORMAT_ARGS(wp->propTermStartLsn),
-		   wp->donor->host, wp->donor->port,
+		   wp->safekeeper[wp->donor].host, wp->safekeeper[wp->donor].port,
 		   LSN_FORMAT_ARGS(wp->truncateLsn));
 
 	/*
@@ -1788,14 +1508,6 @@ RecvAppendResponses(Safekeeper *sk)
 
 		readAnything = true;
 
-		/* should never happen: sk is expected to send ERROR instead */
-		if (sk->appendResponse.generation != wp->mconf.generation)
-		{
-			wp_log(FATAL, "safekeeper {id = %lu, ep = %s:%s} sent response with generation %u, expected %u",
-				   sk->greetResponse.nodeId, sk->host, sk->port,
-				   sk->appendResponse.generation, wp->mconf.generation);
-		}
-
 		if (sk->appendResponse.term > wp->propTerm)
 		{
 			/*
@@ -1912,101 +1624,30 @@ CalculateMinFlushLsn(WalProposer *wp)
 }
 
 /*
- * GetAcknowledgedByQuorumWALPosition for a single member set `mset`.
- *
- * `msk` is the member -> safekeeper mapping for mset, i.e. members_safekeepers
- * or new_members_safekeepers.
- */
-static XLogRecPtr
-GetCommittedMset(WalProposer *wp, MemberSet *mset, Safekeeper **msk)
-{
-	XLogRecPtr	responses[MAX_SAFEKEEPERS];
-
-	/*
-	 * Ascending sort acknowledged LSNs.
-	 */
-	Assert(mset->len <= MAX_SAFEKEEPERS);
-	for (uint32 i = 0; i < mset->len; i++)
-	{
-		Safekeeper *sk = msk[i];
-
-		/*
-		 * Like in Raft, we aren't allowed to commit entries from previous
-		 * terms, so ignore reported LSN until it gets to propTermStartLsn.
-		 *
-		 * Note: we ignore sk state, which is ok: before first ack flushLsn is
-		 * 0, and later we just preserve value across reconnections. It would
-		 * be ok to check for SS_ACTIVE as well.
-		 */
-		if (sk != NULL && sk->appendResponse.flushLsn >= wp->propTermStartLsn)
-		{
-			responses[i] = sk->appendResponse.flushLsn;
-		}
-		else
-		{
-			responses[i] = 0;
-		}
-	}
-	qsort(responses, mset->len, sizeof(XLogRecPtr), CompareLsn);
-
-	/*
-	 * And get value committed by the quorum. A way to view this: to get the
-	 * highest value committed on the quorum, in the ordered array we skip n -
-	 * n_quorum elements to get to the first (lowest) value present on all sks
-	 * of the highest quorum.
-	 */
-	return responses[mset->len - MsetQuorum(mset)];
-}
-
-/*
- * Calculate WAL position acknowledged by quorum, i.e. which may be regarded
- * committed.
- *
- * Zero may be returned when there is no quorum of nodes recovered to term start
- * lsn which sent feedback yet.
+ * Calculate WAL position acknowledged by quorum
  */
 static XLogRecPtr
 GetAcknowledgedByQuorumWALPosition(WalProposer *wp)
 {
-	XLogRecPtr	committed;
+	XLogRecPtr	responses[MAX_SAFEKEEPERS];
 
-	/* legacy: generations disabled */
-	if (!WalProposerGenerationsEnabled(wp) && wp->mconf.generation == INVALID_GENERATION)
+	/*
+	 * Sort acknowledged LSNs
+	 */
+	for (int i = 0; i < wp->n_safekeepers; i++)
 	{
-		XLogRecPtr	responses[MAX_SAFEKEEPERS];
-
 		/*
-		 * Sort acknowledged LSNs
+		 * Like in Raft, we aren't allowed to commit entries from previous
+		 * terms, so ignore reported LSN until it gets to epochStartLsn.
 		 */
-		for (int i = 0; i < wp->n_safekeepers; i++)
-		{
-			/*
-			 * Like in Raft, we aren't allowed to commit entries from previous
-			 * terms, so ignore reported LSN until it gets to
-			 * propTermStartLsn.
-			 *
-			 * Note: we ignore sk state, which is ok: before first ack
-			 * flushLsn is 0, and later we just preserve value across
-			 * reconnections. It would be ok to check for SS_ACTIVE as well.
-			 */
-			responses[i] = wp->safekeeper[i].appendResponse.flushLsn >= wp->propTermStartLsn ? wp->safekeeper[i].appendResponse.flushLsn : 0;
-		}
-		qsort(responses, wp->n_safekeepers, sizeof(XLogRecPtr), CompareLsn);
-
-		/*
-		 * Get the smallest LSN committed by quorum
-		 */
-		return responses[wp->n_safekeepers - wp->quorum];
+		responses[i] = wp->safekeeper[i].appendResponse.flushLsn >= wp->propTermStartLsn ? wp->safekeeper[i].appendResponse.flushLsn : 0;
 	}
+	qsort(responses, wp->n_safekeepers, sizeof(XLogRecPtr), CompareLsn);
 
-	committed = GetCommittedMset(wp, &wp->mconf.members, wp->members_safekeepers);
-	if (wp->mconf.new_members.len > 0)
-	{
-		XLogRecPtr	new_mset_committed = GetCommittedMset(wp, &wp->mconf.new_members, wp->new_members_safekeepers);
-
-		committed = Min(committed, new_mset_committed);
-	}
-	return committed;
+	/*
+	 * Get the smallest LSN committed by quorum
+	 */
+	return responses[wp->n_safekeepers - wp->quorum];
 }
 
 /*
@@ -2021,7 +1662,7 @@ UpdateDonorShmem(WalProposer *wp)
 	int			i;
 	XLogRecPtr	donor_lsn = InvalidXLogRecPtr;
 
-	if (wp->state < WPS_ELECTED)
+	if (wp->n_votes < wp->quorum)
 	{
 		wp_log(WARNING, "UpdateDonorShmem called before elections are won");
 		return;
@@ -2032,9 +1673,9 @@ UpdateDonorShmem(WalProposer *wp)
 	 * about its position immediately after election before any feedbacks are
 	 * sent.
 	 */
-	if (wp->donor->state >= SS_WAIT_ELECTED)
+	if (wp->safekeeper[wp->donor].state >= SS_WAIT_ELECTED)
 	{
-		donor = wp->donor;
+		donor = &wp->safekeeper[wp->donor];
 		donor_lsn = wp->propTermStartLsn;
 	}
 
@@ -2105,13 +1746,13 @@ HandleSafekeeperResponse(WalProposer *wp, Safekeeper *fromsk)
 	}
 
 	/*
-	 * Generally sync is done when majority reached propTermStartLsn so we
-	 * committed it and made the majority aware of it, ensuring they are ready
-	 * to give all WAL to pageserver. It would mean whichever majority is
-	 * alive, there will be at least one safekeeper who is able to stream WAL
-	 * to pageserver to make basebackup possible. However, since at the moment
-	 * we don't have any good mechanism of defining the healthy and most
-	 * advanced safekeeper who should push the wal into pageserver and
+	 * Generally sync is done when majority switched the epoch so we committed
+	 * epochStartLsn and made the majority aware of it, ensuring they are
+	 * ready to give all WAL to pageserver. It would mean whichever majority
+	 * is alive, there will be at least one safekeeper who is able to stream
+	 * WAL to pageserver to make basebackup possible. However, since at the
+	 * moment we don't have any good mechanism of defining the healthy and
+	 * most advanced safekeeper who should push the wal into pageserver and
 	 * basically the random one gets connected, to prevent hanging basebackup
 	 * (due to pageserver connecting to not-synced-safekeeper) we currently
 	 * wait for all seemingly alive safekeepers to get synced.
@@ -2133,7 +1774,7 @@ HandleSafekeeperResponse(WalProposer *wp, Safekeeper *fromsk)
 				n_synced++;
 		}
 
-		if (newCommitLsn >= wp->propTermStartLsn)
+		if (n_synced >= wp->quorum)
 		{
 			/* A quorum of safekeepers has been synced! */
 

pgxn/neon/walproposer.h

@@ -145,7 +145,6 @@ typedef uint64 NNodeId;
  * This and following structs pair ones in membership.rs.
  */
 typedef uint32 Generation;
-#define INVALID_GENERATION 0
 
 typedef struct SafekeeperId
 {
@@ -772,17 +771,7 @@ typedef struct WalProposer
 	/* Current walproposer membership configuration */
 	MembershipConfiguration mconf;
 
-	/*
-	 * Parallels mconf.members with pointers to the member's slot in
-	 * safekeepers array of connections, or NULL if such member is not
-	 * connected. Helps to avoid looking slot per id through all
-	 * .safekeepers[] when doing quorum checks.
-	 */
-	Safekeeper *members_safekeepers[MAX_SAFEKEEPERS];
-	/* As above, but for new_members. */
-	Safekeeper *new_members_safekeepers[MAX_SAFEKEEPERS];
-
-	/* (n_safekeepers / 2) + 1. Used for static pre-generations quorum checks. */
+	/* (n_safekeepers / 2) + 1 */
 	int			quorum;
 
 	/*
@@ -840,7 +829,7 @@ typedef struct WalProposer
 	term_t		donorLastLogTerm;
 
 	/* Most advanced acceptor */
-	Safekeeper *donor;
+	int			donor;
 
 	/* timeline globally starts at this LSN */
 	XLogRecPtr	timelineStartLsn;

proxy/src/cancellation.rs

@@ -425,7 +425,12 @@ impl CancelClosure {
             &mut mk_tls,
             &self.hostname,
         )
-        .map_err(|e| CancelError::IO(std::io::Error::other(e.to_string())))?;
+        .map_err(|e| {
+            CancelError::IO(std::io::Error::new(
+                std::io::ErrorKind::Other,
+                e.to_string(),
+            ))
+        })?;
 
         self.cancel_token.cancel_query_raw(socket, tls).await?;
         debug!("query was cancelled");

proxy/src/proxy/tests/mod.rs

@@ -568,7 +568,7 @@ fn helper_create_cached_node_info(cache: &'static NodeInfoCache) -> CachedNodeIn
 fn helper_create_connect_info(
     mechanism: &TestConnectMechanism,
 ) -> auth::Backend<'static, ComputeCredentials> {
-    auth::Backend::ControlPlane(
+    let user_info = auth::Backend::ControlPlane(
         MaybeOwned::Owned(ControlPlaneClient::Test(Box::new(mechanism.clone()))),
         ComputeCredentials {
             info: ComputeUserInfo {
@@ -578,7 +578,8 @@ fn helper_create_connect_info(
             },
             keys: ComputeCredentialKeys::Password("password".into()),
         },
-    )
+    );
+    user_info
 }
 
 fn config() -> ComputeConfig {

proxy/src/serverless/conn_pool_lib.rs

@@ -47,7 +47,6 @@ impl ConnInfo {
 }
 
 #[derive(Clone)]
-#[allow(clippy::large_enum_variant, reason = "TODO")]
 pub(crate) enum ClientDataEnum {
     Remote(ClientDataRemote),
     Local(ClientDataLocal),

safekeeper/src/http/mod.rs

@@ -31,7 +31,6 @@ pub async fn task_main_https(
     global_timelines: Arc<GlobalTimelines>,
 ) -> anyhow::Result<()> {
     let cert_resolver = ReloadingCertificateResolver::new(
-        "main",
         &conf.ssl_key_file,
         &conf.ssl_cert_file,
         conf.ssl_cert_reload_period,

safekeeper/src/timeline.rs

@@ -138,7 +138,6 @@ impl Drop for WriteGuardSharedState<'_> {
 /// Usually it holds SafeKeeper, but it also supports offloaded timeline state. In this
 /// case, SafeKeeper is not available (because WAL is not present on disk) and all
 /// operations can be done only with control file.
-#[allow(clippy::large_enum_variant, reason = "TODO")]
 pub enum StateSK {
     Loaded(SafeKeeper<control_file::FileStorage, wal_storage::PhysicalStorage>),
     Offloaded(Box<TimelineState<control_file::FileStorage>>),