Add missing semicolon to finalize_drop_subscriptions.sql

Signed-off-by: Tristan Partin <tristan@neon.tech>

Cargo.lock

@@ -2863,6 +2863,7 @@ version = "0.1.0"
 dependencies = [
  "anyhow",
  "bytes",
+ "camino",
  "fail",
  "futures",
  "hyper 0.14.30",
@@ -2873,6 +2874,7 @@ dependencies = [
  "pprof",
  "regex",
  "routerify",
+ "rustls-pemfile 2.1.1",
  "serde",
  "serde_json",
  "serde_path_to_error",
@@ -2902,9 +2904,9 @@ checksum = "c4a1e36c821dbe04574f602848a19f742f4fb3c98d40449f11bcad18d6b17421"
 
 [[package]]
 name = "humantime"
-version = "2.1.0"
+version = "2.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9a3a5bfb195931eeb336b2a7b4d761daec841b97f947d34394601737a7bba5e4"
+checksum = "9b112acc8b3adf4b107a8ec20977da0273a8c386765a3ec0229bd500a1443f9f"
 
 [[package]]
 name = "humantime-serde"
@@ -4327,8 +4329,6 @@ dependencies = [
  "reqwest",
  "rpds",
  "rustls 0.23.18",
- "rustls-pemfile 2.1.1",
- "rustls-pki-types",
  "scopeguard",
  "send-future",
  "serde",
@@ -6044,6 +6044,7 @@ dependencies = [
  "regex",
  "remote_storage",
  "reqwest",
+ "rustls 0.23.18",
  "safekeeper_api",
  "safekeeper_client",
  "scopeguard",
@@ -6060,6 +6061,7 @@ dependencies = [
  "tokio",
  "tokio-io-timeout",
  "tokio-postgres",
+ "tokio-rustls 0.26.0",
  "tokio-stream",
  "tokio-tar",
  "tokio-util",

Cargo.toml

@@ -106,7 +106,7 @@ hostname = "0.4"
 http = {version = "1.1.0", features = ["std"]}
 http-types = { version = "2", default-features = false }
 http-body-util = "0.1.2"
-humantime = "2.1"
+humantime = "2.2"
 humantime-serde = "1.1.1"
 hyper0 = { package = "hyper", version = "0.14" }
 hyper = "1.4"

compute/vm-image-spec-bookworm.yaml

@@ -145,7 +145,7 @@ merge: |
 
   COPY compute_rsyslog.conf /etc/compute_rsyslog.conf
   RUN chmod 0666 /etc/compute_rsyslog.conf
-  RUN chmod 0666 /var/log/
+  RUN mkdir /var/log/rsyslog && chown -R postgres /var/log/rsyslog
 
 
   COPY --from=libcgroup-builder /libcgroup-install/bin/*  /usr/bin/

compute/vm-image-spec-bullseye.yaml

@@ -140,7 +140,7 @@ merge: |
 
   COPY compute_rsyslog.conf /etc/compute_rsyslog.conf
   RUN chmod 0666 /etc/compute_rsyslog.conf
-  RUN chmod 0666 /var/log/
+  RUN mkdir /var/log/rsyslog && chown -R postgres /var/log/rsyslog
 
 
   COPY --from=libcgroup-builder /libcgroup-install/bin/*  /usr/bin/

compute_tools/src/compute.rs

@@ -37,7 +37,7 @@ use crate::logger::startup_context_from_env;
 use crate::lsn_lease::launch_lsn_lease_bg_task_for_static;
 use crate::monitor::launch_monitor;
 use crate::pg_helpers::*;
-use crate::rsyslog::configure_audit_rsyslog;
+use crate::rsyslog::{configure_audit_rsyslog, launch_pgaudit_gc};
 use crate::spec::*;
 use crate::swap::resize_swap;
 use crate::sync_sk::{check_if_synced, ping_safekeeper};
@@ -625,13 +625,11 @@ impl ComputeNode {
             }
 
             let log_directory_path = Path::new(&self.params.pgdata).join("log");
-            // TODO: make this more robust
-            // now rsyslog starts once and there is no monitoring or restart if it fails
-            configure_audit_rsyslog(
-                log_directory_path.to_str().unwrap(),
-                "hipaa",
-                &remote_endpoint,
-            )?;
+            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);
         }
 
         // Launch remaining service threads

compute_tools/src/config.rs

@@ -167,7 +167,8 @@ pub fn write_postgres_conf(
         writeln!(file, "# Managed by compute_ctl audit settings: begin")?;
         // This log level is very verbose
         // but this is necessary for HIPAA compliance.
-        writeln!(file, "pgaudit.log='all'")?;
+        // 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.

compute_tools/src/config_template/compute_audit_rsyslog_template.conf

@@ -4,7 +4,8 @@ module(load="imfile")
 # Input configuration for log files in the specified directory
 # Replace {log_directory} with the directory containing the log files
 input(type="imfile" File="{log_directory}/*.log" Tag="{tag}" Severity="info" Facility="local0")
-global(workDirectory="/var/log")
+# the directory to store rsyslog state files
+global(workDirectory="/var/log/rsyslog")
 
 # Forward logs to remote syslog server
 *.* @@{remote_endpoint}

compute_tools/src/metrics.rs

@@ -1,6 +1,8 @@
-use metrics::core::Collector;
+use metrics::core::{AtomicF64, Collector, GenericGauge};
 use metrics::proto::MetricFamily;
-use metrics::{IntCounterVec, UIntGaugeVec, register_int_counter_vec, register_uint_gauge_vec};
+use metrics::{
+    IntCounterVec, UIntGaugeVec, register_gauge, register_int_counter_vec, register_uint_gauge_vec,
+};
 use once_cell::sync::Lazy;
 
 pub(crate) static INSTALLED_EXTENSIONS: Lazy<UIntGaugeVec> = Lazy::new(|| {
@@ -59,10 +61,20 @@ pub(crate) static REMOTE_EXT_REQUESTS_TOTAL: Lazy<IntCounterVec> = Lazy::new(||
     .expect("failed to define a metric")
 });
 
+// Size of audit log directory in bytes
+pub(crate) static AUDIT_LOG_DIR_SIZE: Lazy<GenericGauge<AtomicF64>> = Lazy::new(|| {
+    register_gauge!(
+        "compute_audit_log_dir_size",
+        "Size of audit log directory in bytes",
+    )
+    .expect("failed to define a metric")
+});
+
 pub fn collect() -> Vec<MetricFamily> {
     let mut metrics = INSTALLED_EXTENSIONS.collect();
     metrics.extend(CPLANE_REQUESTS_TOTAL.collect());
     metrics.extend(REMOTE_EXT_REQUESTS_TOTAL.collect());
     metrics.extend(DB_MIGRATION_FAILED.collect());
+    metrics.extend(AUDIT_LOG_DIR_SIZE.collect());
     metrics
 }

compute_tools/src/rsyslog.rs

@@ -1,8 +1,11 @@
+use std::fs;
+use std::path::Path;
 use std::process::Command;
+use std::time::Duration;
 use std::{fs::OpenOptions, io::Write};
 
 use anyhow::{Context, Result};
-use tracing::info;
+use tracing::{error, info, instrument, warn};
 
 fn get_rsyslog_pid() -> Option<String> {
     let output = Command::new("pgrep")
@@ -43,7 +46,7 @@ fn restart_rsyslog() -> Result<()> {
 }
 
 pub fn configure_audit_rsyslog(
-    log_directory: &str,
+    log_directory: String,
     tag: &str,
     remote_endpoint: &str,
 ) -> Result<()> {
@@ -75,3 +78,61 @@ pub fn configure_audit_rsyslog(
 
     Ok(())
 }
+
+#[instrument(skip_all)]
+async fn pgaudit_gc_main_loop(log_directory: String) -> Result<()> {
+    info!("running pgaudit GC main loop");
+    loop {
+        // Check log_directory for old pgaudit logs and delete them.
+        // New log files are checked every 5 minutes, as set in pgaudit.log_rotation_age
+        // Find files that were not modified in the last 15 minutes and delete them.
+        // This should be enough time for rsyslog to process the logs and for us to catch the alerts.
+        //
+        // In case of a very high load, we might need to adjust this value and pgaudit.log_rotation_age.
+        //
+        // TODO: add some smarter logic to delete the files that are fully streamed according to rsyslog
+        // imfile-state files, but for now just do a simple GC to avoid filling up the disk.
+        let _ = Command::new("find")
+            .arg(&log_directory)
+            .arg("-name")
+            .arg("audit*.log")
+            .arg("-mmin")
+            .arg("+15")
+            .arg("-delete")
+            .output()?;
+
+        // also collect the metric for the size of the log directory
+        async fn get_log_files_size(path: &Path) -> Result<u64> {
+            let mut total_size = 0;
+
+            for entry in fs::read_dir(path)? {
+                let entry = entry?;
+                let entry_path = entry.path();
+
+                if entry_path.is_file() && entry_path.to_string_lossy().ends_with("log") {
+                    total_size += entry.metadata()?.len();
+                }
+            }
+
+            Ok(total_size)
+        }
+
+        let log_directory_size = get_log_files_size(Path::new(&log_directory))
+            .await
+            .unwrap_or_else(|e| {
+                warn!("Failed to get log directory size: {}", e);
+                0
+            });
+        crate::metrics::AUDIT_LOG_DIR_SIZE.set(log_directory_size as f64);
+        tokio::time::sleep(Duration::from_secs(60)).await;
+    }
+}
+
+// launch pgaudit GC thread to clean up the old pgaudit logs stored in the log_directory
+pub fn launch_pgaudit_gc(log_directory: String) {
+    tokio::spawn(async move {
+        if let Err(e) = pgaudit_gc_main_loop(log_directory).await {
+            error!("pgaudit GC main loop failed: {}", e);
+        }
+    });
+}

compute_tools/src/sql/finalize_drop_subscriptions.sql

@@ -17,5 +17,5 @@ BEGIN
     INSERT INTO neon.drop_subscriptions_done VALUES (1, current_setting('neon.timeline_id'))
     ON CONFLICT (id) DO UPDATE
     SET timeline_id = current_setting('neon.timeline_id');
-END
+END;
 $$

control_plane/src/local_env.rs

@@ -149,7 +149,7 @@ pub struct NeonBroker {
     pub listen_addr: SocketAddr,
 }
 
-/// Broker config for cluster internal communication.
+/// A part of storage controller's config the neon_local knows about.
 #[derive(Serialize, Deserialize, PartialEq, Eq, Clone, Debug)]
 #[serde(default)]
 pub struct NeonStorageControllerConf {
@@ -176,10 +176,11 @@ pub struct NeonStorageControllerConf {
     #[serde(with = "humantime_serde")]
     pub long_reconcile_threshold: Option<Duration>,
 
-    #[serde(default)]
     pub use_https_pageserver_api: bool,
 
     pub timelines_onto_safekeepers: bool,
+
+    pub use_https_safekeeper_api: bool,
 }
 
 impl NeonStorageControllerConf {
@@ -205,6 +206,7 @@ impl Default for NeonStorageControllerConf {
             long_reconcile_threshold: None,
             use_https_pageserver_api: false,
             timelines_onto_safekeepers: false,
+            use_https_safekeeper_api: false,
         }
     }
 }
@@ -302,6 +304,7 @@ pub struct SafekeeperConf {
     pub pg_port: u16,
     pub pg_tenant_only_port: Option<u16>,
     pub http_port: u16,
+    pub https_port: Option<u16>,
     pub sync: bool,
     pub remote_storage: Option<String>,
     pub backup_threads: Option<u32>,
@@ -316,6 +319,7 @@ impl Default for SafekeeperConf {
             pg_port: 0,
             pg_tenant_only_port: None,
             http_port: 0,
+            https_port: None,
             sync: true,
             remote_storage: None,
             backup_threads: None,
@@ -845,6 +849,9 @@ impl LocalEnv {
         // create safekeeper dirs
         for safekeeper in &env.safekeepers {
             fs::create_dir_all(SafekeeperNode::datadir_path_by_id(&env, safekeeper.id))?;
+            SafekeeperNode::from_env(&env, safekeeper)
+                .initialize()
+                .context("safekeeper init failed")?;
         }
 
         // initialize pageserver state

control_plane/src/safekeeper.rs

@@ -111,6 +111,18 @@ impl SafekeeperNode {
             .expect("non-Unicode path")
     }
 
+    /// Initializes a safekeeper node by creating all necessary files,
+    /// e.g. SSL certificates.
+    pub fn initialize(&self) -> anyhow::Result<()> {
+        if self.env.generate_local_ssl_certs {
+            self.env.generate_ssl_cert(
+                &self.datadir_path().join("server.crt"),
+                &self.datadir_path().join("server.key"),
+            )?;
+        }
+        Ok(())
+    }
+
     pub async fn start(
         &self,
         extra_opts: &[String],
@@ -196,6 +208,16 @@ impl SafekeeperNode {
             ]);
         }
 
+        if let Some(https_port) = self.conf.https_port {
+            args.extend([
+                "--listen-https".to_owned(),
+                format!("{}:{}", self.listen_addr, https_port),
+            ]);
+        }
+        if let Some(ssl_ca_file) = self.env.ssl_ca_cert_path() {
+            args.push(format!("--ssl-ca-file={}", ssl_ca_file.to_str().unwrap()));
+        }
+
         args.extend_from_slice(extra_opts);
 
         background_process::start_process(

control_plane/src/storage_controller.rs

@@ -538,6 +538,10 @@ impl StorageController {
             args.push("--use-https-pageserver-api".to_string());
         }
 
+        if self.config.use_https_safekeeper_api {
+            args.push("--use-https-safekeeper-api".to_string());
+        }
+
         if let Some(ssl_ca_file) = self.env.ssl_ca_cert_path() {
             args.push(format!("--ssl-ca-file={}", ssl_ca_file.to_str().unwrap()));
         }

deny.toml

@@ -31,10 +31,6 @@ reason = "the marvin attack only affects private key decryption, not public key
 id = "RUSTSEC-2024-0436"
 reason = "The paste crate is a build-only dependency with no runtime components. It is unlikely to have any security impact."
 
-[[advisories.ignore]]
-id = "RUSTSEC-2025-0014"
-reason = "The humantime is widely used and is not easy to replace right now. It is unmaintained, but it has no known vulnerabilities to care about. #11179"
-
 # This section is considered when running `cargo deny check licenses`
 # More documentation for the licenses section can be found here:
 # https://embarkstudios.github.io/cargo-deny/checks/licenses/cfg.html

libs/http-utils/Cargo.toml

@@ -7,6 +7,7 @@ license.workspace = true
 [dependencies]
 anyhow.workspace = true
 bytes.workspace = true
+camino.workspace = true
 fail.workspace = true
 futures.workspace = true
 hyper0.workspace = true
@@ -16,6 +17,7 @@ once_cell.workspace = true
 pprof.workspace = true
 regex.workspace = true
 routerify.workspace = true
+rustls-pemfile.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 serde_path_to_error.workspace = true

libs/http-utils/src/lib.rs

@@ -4,6 +4,7 @@ pub mod failpoints;
 pub mod json;
 pub mod request;
 pub mod server;
+pub mod tls_certs;
 
 extern crate hyper0 as hyper;
 

libs/http-utils/src/tls_certs.rs

@@ -0,0 +1,21 @@
+use camino::Utf8Path;
+use tokio_rustls::rustls::pki_types::{CertificateDer, PrivateKeyDer};
+
+pub fn load_cert_chain(filename: &Utf8Path) -> anyhow::Result<Vec<CertificateDer<'static>>> {
+    let file = std::fs::File::open(filename)?;
+    let mut reader = std::io::BufReader::new(file);
+
+    Ok(rustls_pemfile::certs(&mut reader).collect::<Result<Vec<_>, _>>()?)
+}
+
+pub fn load_private_key(filename: &Utf8Path) -> anyhow::Result<PrivateKeyDer<'static>> {
+    let file = std::fs::File::open(filename)?;
+    let mut reader = std::io::BufReader::new(file);
+
+    let key = rustls_pemfile::private_key(&mut reader)?;
+
+    key.ok_or(anyhow::anyhow!(
+        "no private key found in {}",
+        filename.as_str(),
+    ))
+}

libs/pageserver_api/src/config.rs

@@ -289,6 +289,8 @@ pub struct TenantConfigToml {
     /// If true, Level0 delta layer flushes will wait for S3 upload before flushing the next
     /// layer. This is a temporary backpressure mechanism which should be removed once
     /// l0_flush_{delay,stall}_threshold is fully enabled.
+    ///
+    /// TODO: this is no longer enabled, remove it when the config option is no longer set.
     pub l0_flush_wait_upload: bool,
     // Determines how much history is retained, to allow
     // branching and read replicas at an older point in time.
@@ -576,7 +578,7 @@ pub mod tenant_conf_defaults {
     pub const DEFAULT_COMPACTION_ALGORITHM: crate::models::CompactionAlgorithm =
         crate::models::CompactionAlgorithm::Legacy;
 
-    pub const DEFAULT_L0_FLUSH_WAIT_UPLOAD: bool = true;
+    pub const DEFAULT_L0_FLUSH_WAIT_UPLOAD: bool = false;
 
     pub const DEFAULT_GC_HORIZON: u64 = 64 * 1024 * 1024;
 

libs/pageserver_api/src/models.rs

@@ -176,6 +176,39 @@ impl LsnLease {
     }
 }
 
+/// Controls the detach ancestor behavior.
+/// - When set to `NoAncestorAndReparent`, we will only detach a branch if its ancestor is a root branch. It will automatically reparent any children of the ancestor before and at the branch point.
+/// - When set to `MultiLevelAndNoReparent`, we will detach a branch from multiple levels of ancestors, and no reparenting will happen at all.
+#[derive(Debug, Clone, Copy, Default)]
+pub enum DetachBehavior {
+    #[default]
+    NoAncestorAndReparent,
+    MultiLevelAndNoReparent,
+}
+
+impl std::str::FromStr for DetachBehavior {
+    type Err = &'static str;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "no_ancestor_and_reparent" => Ok(DetachBehavior::NoAncestorAndReparent),
+            "multi_level_and_no_reparent" => Ok(DetachBehavior::MultiLevelAndNoReparent),
+            "v1" => Ok(DetachBehavior::NoAncestorAndReparent),
+            "v2" => Ok(DetachBehavior::MultiLevelAndNoReparent),
+            _ => Err("cannot parse detach behavior"),
+        }
+    }
+}
+
+impl std::fmt::Display for DetachBehavior {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            DetachBehavior::NoAncestorAndReparent => write!(f, "no_ancestor_and_reparent"),
+            DetachBehavior::MultiLevelAndNoReparent => write!(f, "multi_level_and_no_reparent"),
+        }
+    }
+}
+
 /// The only [`TenantState`] variants we could be `TenantState::Activating` from.
 ///
 /// XXX: We used to have more variants here, but now it's just one, which makes this rather

libs/safekeeper_api/src/models.rs

@@ -221,6 +221,11 @@ pub struct TimelineMembershipSwitchResponse {
     pub current_conf: Configuration,
 }
 
+#[derive(Clone, Copy, Serialize, Deserialize)]
+pub struct TimelineDeleteResult {
+    pub dir_existed: bool,
+}
+
 fn lsn_invalid() -> Lsn {
     Lsn::INVALID
 }

pageserver/Cargo.toml

@@ -48,8 +48,6 @@ pprof.workspace = true
 rand.workspace = true
 range-set-blaze = { version = "0.1.16", features = ["alloc"] }
 regex.workspace = true
-rustls-pemfile.workspace = true
-rustls-pki-types.workspace = true
 rustls.workspace = true
 scopeguard.workspace = true
 send-future.workspace = true

pageserver/client/src/mgmt_api.rs

@@ -7,7 +7,7 @@ use http_utils::error::HttpErrorBody;
 use pageserver_api::models::*;
 use pageserver_api::shard::TenantShardId;
 pub use reqwest::Body as ReqwestBody;
-use reqwest::{Certificate, IntoUrl, Method, StatusCode};
+use reqwest::{Certificate, IntoUrl, Method, StatusCode, Url};
 use utils::id::{TenantId, TimelineId};
 use utils::lsn::Lsn;
 
@@ -458,13 +458,21 @@ impl Client {
         &self,
         tenant_shard_id: TenantShardId,
         timeline_id: TimelineId,
+        behavior: Option<DetachBehavior>,
     ) -> Result<AncestorDetached> {
         let uri = format!(
             "{}/v1/tenant/{tenant_shard_id}/timeline/{timeline_id}/detach_ancestor",
             self.mgmt_api_endpoint
         );
+        let mut uri = Url::parse(&uri)
+            .map_err(|e| Error::ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("{e}")))?;
 
-        self.request(Method::PUT, &uri, ())
+        if let Some(behavior) = behavior {
+            uri.query_pairs_mut()
+                .append_pair("detach_behavior", &behavior.to_string());
+        }
+
+        self.request(Method::PUT, uri, ())
             .await?
             .json()
             .await

pageserver/src/bin/pageserver.rs

@@ -30,7 +30,6 @@ use pageserver::{
 };
 use postgres_backend::AuthType;
 use remote_storage::GenericRemoteStorage;
-use rustls_pki_types::{CertificateDer, PrivateKeyDer};
 use tokio::signal::unix::SignalKind;
 use tokio::time::Instant;
 use tokio_util::sync::CancellationToken;
@@ -622,8 +621,8 @@ fn start_pageserver(
 
         let https_task = match https_listener {
             Some(https_listener) => {
-                let certs = load_certs(&conf.ssl_cert_file)?;
-                let key = load_private_key(&conf.ssl_key_file)?;
+                let certs = http_utils::tls_certs::load_cert_chain(&conf.ssl_cert_file)?;
+                let key = http_utils::tls_certs::load_private_key(&conf.ssl_key_file)?;
 
                 let server_config = rustls::ServerConfig::builder()
                     .with_no_client_auth()
@@ -735,25 +734,6 @@ fn start_pageserver(
     })
 }
 
-fn load_certs(filename: &Utf8Path) -> std::io::Result<Vec<CertificateDer<'static>>> {
-    let file = std::fs::File::open(filename)?;
-    let mut reader = std::io::BufReader::new(file);
-
-    rustls_pemfile::certs(&mut reader).collect()
-}
-
-fn load_private_key(filename: &Utf8Path) -> anyhow::Result<PrivateKeyDer<'static>> {
-    let file = std::fs::File::open(filename)?;
-    let mut reader = std::io::BufReader::new(file);
-
-    let key = rustls_pemfile::private_key(&mut reader)?;
-
-    key.ok_or(anyhow::anyhow!(
-        "no private key found in {}",
-        filename.as_str(),
-    ))
-}
-
 async fn create_remote_storage_client(
     conf: &'static PageServerConf,
 ) -> anyhow::Result<GenericRemoteStorage> {

pageserver/src/http/routes.rs

@@ -28,9 +28,9 @@ use hyper::{Body, Request, Response, StatusCode, Uri, header};
 use metrics::launch_timestamp::LaunchTimestamp;
 use pageserver_api::models::virtual_file::IoMode;
 use pageserver_api::models::{
-    DownloadRemoteLayersTaskSpawnRequest, IngestAuxFilesRequest, ListAuxFilesRequest,
-    LocationConfig, LocationConfigListResponse, LocationConfigMode, LsnLease, LsnLeaseRequest,
-    OffloadedTimelineInfo, PageTraceEvent, ShardParameters, StatusResponse,
+    DetachBehavior, DownloadRemoteLayersTaskSpawnRequest, IngestAuxFilesRequest,
+    ListAuxFilesRequest, LocationConfig, LocationConfigListResponse, LocationConfigMode, LsnLease,
+    LsnLeaseRequest, OffloadedTimelineInfo, PageTraceEvent, ShardParameters, StatusResponse,
     TenantConfigPatchRequest, TenantConfigRequest, TenantDetails, TenantInfo,
     TenantLocationConfigRequest, TenantLocationConfigResponse, TenantScanRemoteStorageResponse,
     TenantScanRemoteStorageShard, TenantShardLocation, TenantShardSplitRequest,
@@ -72,7 +72,6 @@ use crate::tenant::remote_timeline_client::{
 use crate::tenant::secondary::SecondaryController;
 use crate::tenant::size::ModelInputs;
 use crate::tenant::storage_layer::{IoConcurrency, LayerAccessStatsReset, LayerName};
-use crate::tenant::timeline::detach_ancestor::DetachBehavior;
 use crate::tenant::timeline::offload::{OffloadError, offload_timeline};
 use crate::tenant::timeline::{
     CompactFlags, CompactOptions, CompactRequest, CompactionError, Timeline, WaitLsnTimeout,
@@ -2508,6 +2507,7 @@ async fn timeline_detach_ancestor_handler(
     check_permission(&request, Some(tenant_shard_id.tenant_id))?;
     let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
     let behavior: Option<DetachBehavior> = parse_query_param(&request, "detach_behavior")?;
+
     let behavior = behavior.unwrap_or_default();
 
     let span = tracing::info_span!("detach_ancestor", tenant_id=%tenant_shard_id.tenant_id, shard_id=%tenant_shard_id.shard_slug(), %timeline_id);

pageserver/src/tenant.rs

@@ -7020,6 +7020,7 @@ mod tests {
         child_timeline.freeze_and_flush().await?;
         let mut flags = EnumSet::new();
         flags.insert(CompactFlags::ForceRepartition);
+        flags.insert(CompactFlags::NoYield);
         child_timeline
             .compact(&CancellationToken::new(), flags, &ctx)
             .await?;
@@ -7727,6 +7728,7 @@ mod tests {
                             let mut flags = EnumSet::new();
                             flags.insert(CompactFlags::ForceImageLayerCreation);
                             flags.insert(CompactFlags::ForceRepartition);
+                            flags.insert(CompactFlags::NoYield);
                             flags
                         } else {
                             EnumSet::empty()
@@ -7893,7 +7895,6 @@ mod tests {
             Ok((res, reconstruct_state.get_delta_layers_visited() as usize))
         }
 
-        #[allow(clippy::needless_range_loop)]
         for blknum in 0..NUM_KEYS {
             lsn = Lsn(lsn.0 + 0x10);
             test_key.field6 = (blknum * STEP) as u32;
@@ -7943,6 +7944,7 @@ mod tests {
                             let mut flags = EnumSet::new();
                             flags.insert(CompactFlags::ForceImageLayerCreation);
                             flags.insert(CompactFlags::ForceRepartition);
+                            flags.insert(CompactFlags::NoYield);
                             flags
                         },
                         &ctx,
@@ -8405,6 +8407,7 @@ mod tests {
                     let mut flags = EnumSet::new();
                     flags.insert(CompactFlags::ForceImageLayerCreation);
                     flags.insert(CompactFlags::ForceRepartition);
+                    flags.insert(CompactFlags::NoYield);
                     flags
                 },
                 &ctx,
@@ -8472,6 +8475,7 @@ mod tests {
                     let mut flags = EnumSet::new();
                     flags.insert(CompactFlags::ForceImageLayerCreation);
                     flags.insert(CompactFlags::ForceRepartition);
+                    flags.insert(CompactFlags::NoYield);
                     flags
                 },
                 &ctx,

pageserver/src/tenant/mgr.rs

@@ -14,7 +14,7 @@ use futures::StreamExt;
 use itertools::Itertools;
 use once_cell::sync::Lazy;
 use pageserver_api::key::Key;
-use pageserver_api::models::LocationConfigMode;
+use pageserver_api::models::{DetachBehavior, LocationConfigMode};
 use pageserver_api::shard::{
     ShardCount, ShardIdentity, ShardIndex, ShardNumber, ShardStripeSize, TenantShardId,
 };
@@ -1914,7 +1914,7 @@ impl TenantManager {
         tenant_shard_id: TenantShardId,
         timeline_id: TimelineId,
         prepared: PreparedTimelineDetach,
-        behavior: detach_ancestor::DetachBehavior,
+        behavior: DetachBehavior,
         mut attempt: detach_ancestor::Attempt,
         ctx: &RequestContext,
     ) -> Result<HashSet<TimelineId>, detach_ancestor::Error> {

pageserver/src/tenant/timeline.rs

@@ -45,8 +45,9 @@ use pageserver_api::key::{
 use pageserver_api::keyspace::{KeySpaceAccum, KeySpaceRandomAccum, SparseKeyPartitioning};
 use pageserver_api::models::{
     CompactKeyRange, CompactLsnRange, CompactionAlgorithm, CompactionAlgorithmSettings,
-    DownloadRemoteLayersTaskInfo, DownloadRemoteLayersTaskSpawnRequest, EvictionPolicy,
-    InMemoryLayerInfo, LayerMapInfo, LsnLease, PageTraceEvent, RelSizeMigration, TimelineState,
+    DetachBehavior, DownloadRemoteLayersTaskInfo, DownloadRemoteLayersTaskSpawnRequest,
+    EvictionPolicy, InMemoryLayerInfo, LayerMapInfo, LsnLease, PageTraceEvent, RelSizeMigration,
+    TimelineState,
 };
 use pageserver_api::reltag::{BlockNumber, RelTag};
 use pageserver_api::shard::{ShardIdentity, ShardIndex, ShardNumber, TenantShardId};
@@ -5445,7 +5446,7 @@ impl Timeline {
         self: &Arc<Timeline>,
         tenant: &crate::tenant::Tenant,
         options: detach_ancestor::Options,
-        behavior: detach_ancestor::DetachBehavior,
+        behavior: DetachBehavior,
         ctx: &RequestContext,
     ) -> Result<detach_ancestor::Progress, detach_ancestor::Error> {
         detach_ancestor::prepare(self, tenant, behavior, options, ctx).await
@@ -5466,7 +5467,7 @@ impl Timeline {
         prepared: detach_ancestor::PreparedTimelineDetach,
         ancestor_timeline_id: TimelineId,
         ancestor_lsn: Lsn,
-        behavior: detach_ancestor::DetachBehavior,
+        behavior: DetachBehavior,
         ctx: &RequestContext,
     ) -> Result<detach_ancestor::DetachingAndReparenting, detach_ancestor::Error> {
         detach_ancestor::detach_and_reparent(

pageserver/src/tenant/timeline/detach_ancestor.rs

@@ -3,6 +3,7 @@ use std::sync::Arc;
 
 use anyhow::Context;
 use http_utils::error::ApiError;
+use pageserver_api::models::DetachBehavior;
 use pageserver_api::models::detach_ancestor::AncestorDetached;
 use pageserver_api::shard::ShardIdentity;
 use tokio::sync::Semaphore;
@@ -139,30 +140,6 @@ pub(crate) struct Options {
     pub(crate) copy_concurrency: std::num::NonZeroUsize,
 }
 
-/// Controls the detach ancestor behavior.
-/// - When set to `NoAncestorAndReparent`, we will only detach a branch if its ancestor is a root branch. It will automatically reparent any children of the ancestor before and at the branch point.
-/// - When set to `MultiLevelAndNoReparent`, we will detach a branch from multiple levels of ancestors, and no reparenting will happen at all.
-#[derive(Debug, Clone, Copy, Default)]
-pub enum DetachBehavior {
-    #[default]
-    NoAncestorAndReparent,
-    MultiLevelAndNoReparent,
-}
-
-impl std::str::FromStr for DetachBehavior {
-    type Err = &'static str;
-
-    fn from_str(s: &str) -> Result<Self, Self::Err> {
-        match s {
-            "no_ancestor_and_reparent" => Ok(DetachBehavior::NoAncestorAndReparent),
-            "multi_level_and_no_reparent" => Ok(DetachBehavior::MultiLevelAndNoReparent),
-            "v1" => Ok(DetachBehavior::NoAncestorAndReparent),
-            "v2" => Ok(DetachBehavior::MultiLevelAndNoReparent),
-            _ => Err("cannot parse detach behavior"),
-        }
-    }
-}
-
 impl Default for Options {
     fn default() -> Self {
         Self {

proxy/src/metrics.rs

@@ -30,7 +30,16 @@ pub struct Metrics {
 static SELF: OnceLock<Metrics> = OnceLock::new();
 impl Metrics {
     pub fn install(thread_pool: Arc<ThreadPoolMetrics>) {
-        SELF.set(Metrics::new(thread_pool))
+        let mut metrics = Metrics::new(thread_pool);
+
+        metrics.proxy.errors_total.init_all_dense();
+        metrics.proxy.redis_errors_total.init_all_dense();
+        metrics.proxy.redis_events_count.init_all_dense();
+        metrics.proxy.retries_metric.init_all_dense();
+        metrics.proxy.invalid_endpoints_total.init_all_dense();
+        metrics.proxy.connection_failures_total.init_all_dense();
+
+        SELF.set(metrics)
             .ok()
             .expect("proxy metrics must not be installed more than once");
     }

proxy/src/proxy/passthrough.rs

@@ -10,7 +10,7 @@ use crate::config::ComputeConfig;
 use crate::control_plane::messages::MetricsAuxInfo;
 use crate::metrics::{Direction, Metrics, NumClientConnectionsGuard, NumConnectionRequestsGuard};
 use crate::stream::Stream;
-use crate::usage_metrics::{Ids, MetricCounterRecorder, TrafficDirection, USAGE_METRICS};
+use crate::usage_metrics::{Ids, MetricCounterRecorder, USAGE_METRICS};
 
 /// Forward bytes in both directions (client <-> compute).
 #[tracing::instrument(skip_all)]
@@ -24,7 +24,6 @@ pub(crate) async fn proxy_pass(
     let usage_tx = USAGE_METRICS.register(Ids {
         endpoint_id: aux.endpoint_id,
         branch_id: aux.branch_id,
-        direction: TrafficDirection::Egress,
         private_link_id,
     });
 
@@ -47,6 +46,7 @@ pub(crate) async fn proxy_pass(
         |cnt| {
             // Number of bytes the client sent to the compute node (inbound).
             metrics.get_metric(m_recv).inc_by(cnt as u64);
+            usage_tx.record_ingress(cnt as u64);
         },
     );
 

proxy/src/serverless/conn_pool_lib.rs

@@ -22,7 +22,7 @@ use crate::control_plane::messages::{ColdStartInfo, MetricsAuxInfo};
 use crate::metrics::{HttpEndpointPoolsGuard, Metrics};
 use crate::protocol2::ConnectionInfoExtra;
 use crate::types::{DbName, EndpointCacheKey, RoleName};
-use crate::usage_metrics::{Ids, MetricCounter, TrafficDirection, USAGE_METRICS};
+use crate::usage_metrics::{Ids, MetricCounter, USAGE_METRICS};
 
 #[derive(Debug, Clone)]
 pub(crate) struct ConnInfo {
@@ -639,11 +639,7 @@ impl<C: ClientInnerExt> Client<C> {
         (&mut inner.inner, Discard { conn_info, pool })
     }
 
-    pub(crate) fn metrics(
-        &self,
-        direction: TrafficDirection,
-        ctx: &RequestContext,
-    ) -> Arc<MetricCounter> {
+    pub(crate) fn metrics(&self, ctx: &RequestContext) -> Arc<MetricCounter> {
         let aux = &self
             .inner
             .as_ref()
@@ -659,7 +655,6 @@ impl<C: ClientInnerExt> Client<C> {
         USAGE_METRICS.register(Ids {
             endpoint_id: aux.endpoint_id,
             branch_id: aux.branch_id,
-            direction,
             private_link_id,
         })
     }

proxy/src/serverless/http_conn_pool.rs

@@ -19,7 +19,7 @@ use crate::control_plane::messages::{ColdStartInfo, MetricsAuxInfo};
 use crate::metrics::{HttpEndpointPoolsGuard, Metrics};
 use crate::protocol2::ConnectionInfoExtra;
 use crate::types::EndpointCacheKey;
-use crate::usage_metrics::{Ids, MetricCounter, TrafficDirection, USAGE_METRICS};
+use crate::usage_metrics::{Ids, MetricCounter, USAGE_METRICS};
 
 pub(crate) type Send = http2::SendRequest<hyper::body::Incoming>;
 pub(crate) type Connect = http2::Connection<TokioIo<AsyncRW>, hyper::body::Incoming, TokioExecutor>;
@@ -265,11 +265,7 @@ impl<C: ClientInnerExt + Clone> Client<C> {
         Self { inner }
     }
 
-    pub(crate) fn metrics(
-        &self,
-        direction: TrafficDirection,
-        ctx: &RequestContext,
-    ) -> Arc<MetricCounter> {
+    pub(crate) fn metrics(&self, ctx: &RequestContext) -> Arc<MetricCounter> {
         let aux = &self.inner.aux;
 
         let private_link_id = match ctx.extra() {
@@ -281,7 +277,6 @@ impl<C: ClientInnerExt + Clone> Client<C> {
         USAGE_METRICS.register(Ids {
             endpoint_id: aux.endpoint_id,
             branch_id: aux.branch_id,
-            direction,
             private_link_id,
         })
     }

proxy/src/serverless/sql_over_http.rs

@@ -42,7 +42,7 @@ use crate::metrics::{HttpDirection, Metrics};
 use crate::proxy::{NeonOptions, run_until_cancelled};
 use crate::serverless::backend::HttpConnError;
 use crate::types::{DbName, RoleName};
-use crate::usage_metrics::{MetricCounter, MetricCounterRecorder, TrafficDirection};
+use crate::usage_metrics::{MetricCounter, MetricCounterRecorder};
 
 #[derive(serde::Deserialize)]
 #[serde(rename_all = "camelCase")]
@@ -663,6 +663,7 @@ async fn handle_db_inner(
 
     let parsed_headers = HttpHeaders::try_parse(headers)?;
 
+    let mut request_len = 0;
     let fetch_and_process_request = Box::pin(
         async {
             let body = read_body_with_limit(
@@ -671,6 +672,8 @@ async fn handle_db_inner(
             )
             .await?;
 
+            request_len = body.len();
+
             Metrics::get()
                 .proxy
                 .http_conn_content_length_bytes
@@ -765,7 +768,7 @@ async fn handle_db_inner(
         }
     };
 
-    let metrics = client.metrics(TrafficDirection::Egress, ctx);
+    let metrics = client.metrics(ctx);
 
     let len = json_output.len();
     let response = response
@@ -781,6 +784,8 @@ async fn handle_db_inner(
     // count the egress bytes - we miss the TLS and header overhead but oh well...
     // moving this later in the stack is going to be a lot of effort and ehhhh
     metrics.record_egress(len as u64);
+    metrics.record_ingress(request_len as u64);
+
     Metrics::get()
         .proxy
         .http_conn_content_length_bytes
@@ -838,7 +843,7 @@ async fn handle_auth_broker_inner(
         .expect("all headers and params received via hyper should be valid for request");
 
     // todo: map body to count egress
-    let _metrics = client.metrics(TrafficDirection::Egress, ctx);
+    let _metrics = client.metrics(ctx);
 
     Ok(client
         .inner
@@ -1168,10 +1173,10 @@ enum Discard<'a> {
 }
 
 impl Client {
-    fn metrics(&self, direction: TrafficDirection, ctx: &RequestContext) -> Arc<MetricCounter> {
+    fn metrics(&self, ctx: &RequestContext) -> Arc<MetricCounter> {
         match self {
-            Client::Remote(client) => client.metrics(direction, ctx),
-            Client::Local(local_client) => local_client.metrics(direction, ctx),
+            Client::Remote(client) => client.metrics(ctx),
+            Client::Local(local_client) => local_client.metrics(ctx),
         }
     }
 

proxy/src/usage_metrics.rs

@@ -44,11 +44,17 @@ const HTTP_REPORTING_RETRY_DURATION: Duration = Duration::from_secs(60);
 pub(crate) struct Ids {
     pub(crate) endpoint_id: EndpointIdInt,
     pub(crate) branch_id: BranchIdInt,
-    pub(crate) direction: TrafficDirection,
     #[serde(with = "none_as_empty_string")]
     pub(crate) private_link_id: Option<SmolStr>,
 }
 
+#[derive(Eq, Hash, PartialEq, Serialize, Deserialize, Debug, Clone)]
+struct Extra {
+    #[serde(flatten)]
+    ids: Ids,
+    direction: TrafficDirection,
+}
+
 mod none_as_empty_string {
     use serde::Deserialize;
     use smol_str::SmolStr;
@@ -76,18 +82,23 @@ pub(crate) enum TrafficDirection {
 pub(crate) trait MetricCounterRecorder {
     /// Record that some bytes were sent from the proxy to the client
     fn record_egress(&self, bytes: u64);
+
+    /// Record that some bytes were sent from the client to the proxy
+    fn record_ingress(&self, bytes: u64);
+
     /// Record that some connections were opened
     fn record_connection(&self, count: usize);
 }
 
 trait MetricCounterReporter {
-    fn get_metrics(&mut self) -> (u64, usize);
-    fn move_metrics(&self) -> (u64, usize);
+    fn get_metrics(&mut self) -> MetricsData;
+    fn move_metrics(&self) -> MetricsData;
 }
 
 #[derive(Debug)]
 pub(crate) struct MetricCounter {
     transmitted: AtomicU64,
+    received: AtomicU64,
     opened_connections: AtomicUsize,
 }
 
@@ -97,6 +108,11 @@ impl MetricCounterRecorder for MetricCounter {
         self.transmitted.fetch_add(bytes, Ordering::Relaxed);
     }
 
+    /// Record that some bytes were sent from the proxy to the client
+    fn record_ingress(&self, bytes: u64) {
+        self.received.fetch_add(bytes, Ordering::Relaxed);
+    }
+
     /// Record that some connections were opened
     fn record_connection(&self, count: usize) {
         self.opened_connections.fetch_add(count, Ordering::Relaxed);
@@ -104,29 +120,43 @@ impl MetricCounterRecorder for MetricCounter {
 }
 
 impl MetricCounterReporter for MetricCounter {
-    fn get_metrics(&mut self) -> (u64, usize) {
-        (
-            *self.transmitted.get_mut(),
-            *self.opened_connections.get_mut(),
-        )
+    fn get_metrics(&mut self) -> MetricsData {
+        MetricsData {
+            received: *self.received.get_mut(),
+            transmitted: *self.transmitted.get_mut(),
+            connections: *self.opened_connections.get_mut(),
+        }
     }
-    fn move_metrics(&self) -> (u64, usize) {
-        (
-            self.transmitted.swap(0, Ordering::Relaxed),
-            self.opened_connections.swap(0, Ordering::Relaxed),
-        )
+
+    fn move_metrics(&self) -> MetricsData {
+        MetricsData {
+            received: self.received.swap(0, Ordering::Relaxed),
+            transmitted: self.transmitted.swap(0, Ordering::Relaxed),
+            connections: self.opened_connections.swap(0, Ordering::Relaxed),
+        }
     }
 }
 
+struct MetricsData {
+    transmitted: u64,
+    received: u64,
+    connections: usize,
+}
+
+struct BytesSent {
+    transmitted: u64,
+    received: u64,
+}
+
 trait Clearable {
     /// extract the value that should be reported
-    fn should_report(self: &Arc<Self>) -> Option<u64>;
+    fn should_report(self: &Arc<Self>) -> Option<BytesSent>;
     /// Determine whether the counter should be cleared from the global map.
     fn should_clear(self: &mut Arc<Self>) -> bool;
 }
 
 impl<C: MetricCounterReporter> Clearable for C {
-    fn should_report(self: &Arc<Self>) -> Option<u64> {
+    fn should_report(self: &Arc<Self>) -> Option<BytesSent> {
         // heuristic to see if the branch is still open
         // if a clone happens while we are observing, the heuristic will be incorrect.
         //
@@ -139,14 +169,21 @@ impl<C: MetricCounterReporter> Clearable for C {
         // (to avoid sending the same metrics twice)
         // see the relevant discussion on why to do so even if the status is not success:
         // https://github.com/neondatabase/neon/pull/4563#discussion_r1246710956
-        let (value, opened) = self.move_metrics();
+        let MetricsData {
+            transmitted,
+            received,
+            connections,
+        } = self.move_metrics();
 
         // Our only requirement is that we report in every interval if there was an open connection
         // if there were no opened connections since, then we don't need to report
-        if value == 0 && !is_open && opened == 0 {
+        if transmitted == 0 && received == 0 && !is_open && connections == 0 {
             None
         } else {
-            Some(value)
+            Some(BytesSent {
+                transmitted,
+                received,
+            })
         }
     }
     fn should_clear(self: &mut Arc<Self>) -> bool {
@@ -154,9 +191,13 @@ impl<C: MetricCounterReporter> Clearable for C {
         let Some(counter) = Arc::get_mut(self) else {
             return false;
         };
-        let (opened, value) = counter.get_metrics();
+        let MetricsData {
+            transmitted,
+            received,
+            connections,
+        } = counter.get_metrics();
         // clear if there's no data to report
-        value == 0 && opened == 0
+        transmitted == 0 && received == 0 && connections == 0
     }
 }
 
@@ -178,6 +219,7 @@ impl Metrics {
                 .entry(ids)
                 .or_insert_with(|| {
                     Arc::new(MetricCounter {
+                        received: AtomicU64::new(0),
                         transmitted: AtomicU64::new(0),
                         opened_connections: AtomicUsize::new(0),
                     })
@@ -242,10 +284,10 @@ pub async fn task_main(config: &MetricCollectionConfig) -> anyhow::Result<Infall
 
 fn collect_and_clear_metrics<C: Clearable>(
     endpoints: &ClashMap<Ids, Arc<C>, FastHasher>,
-) -> Vec<(Ids, u64)> {
+) -> Vec<(Ids, BytesSent)> {
     let mut metrics_to_clear = Vec::new();
 
-    let metrics_to_send: Vec<(Ids, u64)> = endpoints
+    let metrics_to_send: Vec<(Ids, BytesSent)> = endpoints
         .iter()
         .filter_map(|counter| {
             let key = counter.key().clone();
@@ -271,26 +313,46 @@ fn collect_and_clear_metrics<C: Clearable>(
 }
 
 fn create_event_chunks<'a>(
-    metrics_to_send: &'a [(Ids, u64)],
+    metrics_to_send: &'a [(Ids, BytesSent)],
     hostname: &'a str,
     prev: DateTime<Utc>,
     now: DateTime<Utc>,
     chunk_size: usize,
-) -> impl Iterator<Item = EventChunk<'a, Event<Ids, &'static str>>> + 'a {
+) -> impl Iterator<Item = EventChunk<'a, Event<Extra, &'static str>>> + 'a {
     metrics_to_send
         .chunks(chunk_size)
         .map(move |chunk| EventChunk {
             events: chunk
                 .iter()
-                .map(|(ids, value)| Event {
-                    kind: EventType::Incremental {
-                        start_time: prev,
-                        stop_time: now,
-                    },
-                    metric: PROXY_IO_BYTES_PER_CLIENT,
-                    idempotency_key: idempotency_key(hostname),
-                    value: *value,
-                    extra: ids.clone(),
+                .flat_map(|(ids, bytes)| {
+                    [
+                        Event {
+                            kind: EventType::Incremental {
+                                start_time: prev,
+                                stop_time: now,
+                            },
+                            metric: PROXY_IO_BYTES_PER_CLIENT,
+                            idempotency_key: idempotency_key(hostname),
+                            value: bytes.transmitted,
+                            extra: Extra {
+                                ids: ids.clone(),
+                                direction: TrafficDirection::Egress,
+                            },
+                        },
+                        Event {
+                            kind: EventType::Incremental {
+                                start_time: prev,
+                                stop_time: now,
+                            },
+                            metric: PROXY_IO_BYTES_PER_CLIENT,
+                            idempotency_key: idempotency_key(hostname),
+                            value: bytes.received,
+                            extra: Extra {
+                                ids: ids.clone(),
+                                direction: TrafficDirection::Ingress,
+                            },
+                        },
+                    ]
                 })
                 .collect(),
         })
@@ -350,7 +412,7 @@ fn create_remote_path_prefix(now: DateTime<Utc>) -> String {
 async fn upload_main_events_chunked(
     client: &http::ClientWithMiddleware,
     metric_collection_endpoint: &reqwest::Url,
-    chunk: &EventChunk<'_, Event<Ids, &str>>,
+    chunk: &EventChunk<'_, Event<Extra, &str>>,
     subchunk_size: usize,
 ) {
     // Split into smaller chunks to avoid exceeding the max request size
@@ -384,7 +446,7 @@ async fn upload_main_events_chunked(
 
 async fn upload_backup_events(
     storage: Option<&GenericRemoteStorage>,
-    chunk: &EventChunk<'_, Event<Ids, &'static str>>,
+    chunk: &EventChunk<'_, Event<Extra, &'static str>>,
     path_prefix: &str,
     cancel: &CancellationToken,
 ) -> anyhow::Result<()> {
@@ -461,7 +523,7 @@ mod tests {
 
     #[tokio::test]
     async fn metrics() {
-        type Report = EventChunk<'static, Event<Ids, String>>;
+        type Report = EventChunk<'static, Event<Extra, String>>;
         let reports: Arc<Mutex<Vec<Report>>> = Arc::default();
 
         let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
@@ -533,7 +595,6 @@ mod tests {
         let counter = metrics.register(Ids {
             endpoint_id: (&EndpointId::from("e1")).into(),
             branch_id: (&BranchId::from("b1")).into(),
-            direction: TrafficDirection::Egress,
             private_link_id: None,
         });
 
@@ -551,13 +612,19 @@ mod tests {
         .await;
         let r = std::mem::take(&mut *reports.lock().unwrap());
         assert_eq!(r.len(), 1);
-        assert_eq!(r[0].events.len(), 1);
+        assert_eq!(r[0].events.len(), 2);
         assert_eq!(r[0].events[0].value, 0);
+        assert_eq!(r[0].events[0].extra.direction, TrafficDirection::Egress);
+        assert_eq!(r[0].events[1].value, 0);
+        assert_eq!(r[0].events[1].extra.direction, TrafficDirection::Ingress);
         pushed_chunks.extend(r);
 
         // record egress
         counter.record_egress(1);
 
+        // record ingress
+        counter.record_ingress(2);
+
         // egress should be observered
         collect_metrics_iteration(
             &metrics.endpoints,
@@ -572,8 +639,11 @@ mod tests {
         .await;
         let r = std::mem::take(&mut *reports.lock().unwrap());
         assert_eq!(r.len(), 1);
-        assert_eq!(r[0].events.len(), 1);
+        assert_eq!(r[0].events.len(), 2);
         assert_eq!(r[0].events[0].value, 1);
+        assert_eq!(r[0].events[0].extra.direction, TrafficDirection::Egress);
+        assert_eq!(r[0].events[1].value, 2);
+        assert_eq!(r[0].events[1].extra.direction, TrafficDirection::Ingress);
         pushed_chunks.extend(r);
 
         // release counter

safekeeper/Cargo.toml

@@ -35,8 +35,9 @@ postgres-protocol.workspace = true
 pprof.workspace = true
 rand.workspace = true
 regex.workspace = true
-scopeguard.workspace = true
 reqwest = { workspace = true, features = ["json"] }
+rustls.workspace = true
+scopeguard.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 smallvec.workspace = true
@@ -45,10 +46,11 @@ strum_macros.workspace = true
 thiserror.workspace = true
 tikv-jemallocator.workspace = true
 tokio = { workspace = true, features = ["fs"] }
-tokio-util = { workspace = true }
 tokio-io-timeout.workspace = true
 tokio-postgres.workspace = true
+tokio-rustls.workspace = true
 tokio-tar.workspace = true
+tokio-util = { workspace = true }
 tracing.workspace = true
 url.workspace = true
 metrics.workspace = true

safekeeper/client/src/mgmt_api.rs

@@ -8,7 +8,7 @@ use std::error::Error as _;
 use http_utils::error::HttpErrorBody;
 use reqwest::{IntoUrl, Method, StatusCode};
 use safekeeper_api::models::{
-    PullTimelineRequest, PullTimelineResponse, SafekeeperUtilization, TimelineCreateRequest,
+    self, PullTimelineRequest, PullTimelineResponse, SafekeeperUtilization, TimelineCreateRequest,
     TimelineStatus,
 };
 use utils::id::{NodeId, TenantId, TimelineId};
@@ -96,11 +96,25 @@ impl Client {
         resp.json().await.map_err(Error::ReceiveBody)
     }
 
+    pub async fn exclude_timeline(
+        &self,
+        tenant_id: TenantId,
+        timeline_id: TimelineId,
+        req: &models::TimelineMembershipSwitchRequest,
+    ) -> Result<models::TimelineDeleteResult> {
+        let uri = format!(
+            "{}/v1/tenant/{}/timeline/{}/exclude",
+            self.mgmt_api_endpoint, tenant_id, timeline_id
+        );
+        let resp = self.put(&uri, req).await?;
+        resp.json().await.map_err(Error::ReceiveBody)
+    }
+
     pub async fn delete_timeline(
         &self,
         tenant_id: TenantId,
         timeline_id: TimelineId,
-    ) -> Result<TimelineStatus> {
+    ) -> Result<models::TimelineDeleteResult> {
         let uri = format!(
             "{}/v1/tenant/{}/timeline/{}",
             self.mgmt_api_endpoint, tenant_id, timeline_id
@@ -109,6 +123,20 @@ impl Client {
         resp.json().await.map_err(Error::ReceiveBody)
     }
 
+    pub async fn bump_timeline_term(
+        &self,
+        tenant_id: TenantId,
+        timeline_id: TimelineId,
+        req: &models::TimelineTermBumpRequest,
+    ) -> Result<models::TimelineTermBumpResponse> {
+        let uri = format!(
+            "{}/v1/tenant/{}/timeline/{}/term_bump",
+            self.mgmt_api_endpoint, tenant_id, timeline_id
+        );
+        let resp = self.post(&uri, req).await?;
+        resp.json().await.map_err(Error::ReceiveBody)
+    }
+
     pub async fn timeline_status(
         &self,
         tenant_id: TenantId,
@@ -149,6 +177,14 @@ impl Client {
         self.request(Method::POST, uri, body).await
     }
 
+    async fn put<B: serde::Serialize, U: IntoUrl>(
+        &self,
+        uri: U,
+        body: B,
+    ) -> Result<reqwest::Response> {
+        self.request(Method::PUT, uri, body).await
+    }
+
     async fn get<U: IntoUrl>(&self, uri: U) -> Result<reqwest::Response> {
         self.request(Method::GET, uri, ()).await
     }

safekeeper/src/bin/safekeeper.rs

@@ -16,10 +16,12 @@ use futures::stream::FuturesUnordered;
 use futures::{FutureExt, StreamExt};
 use metrics::set_build_info_metric;
 use remote_storage::RemoteStorageConfig;
+use reqwest::Certificate;
 use safekeeper::defaults::{
     DEFAULT_CONTROL_FILE_SAVE_INTERVAL, DEFAULT_EVICTION_MIN_RESIDENT, DEFAULT_HEARTBEAT_TIMEOUT,
     DEFAULT_HTTP_LISTEN_ADDR, DEFAULT_MAX_OFFLOADER_LAG_BYTES, DEFAULT_PARTIAL_BACKUP_CONCURRENCY,
-    DEFAULT_PARTIAL_BACKUP_TIMEOUT, DEFAULT_PG_LISTEN_ADDR,
+    DEFAULT_PARTIAL_BACKUP_TIMEOUT, DEFAULT_PG_LISTEN_ADDR, DEFAULT_SSL_CERT_FILE,
+    DEFAULT_SSL_KEY_FILE,
 };
 use safekeeper::{
     BROKER_RUNTIME, GlobalTimelines, HTTP_RUNTIME, SafeKeeperConf, WAL_SERVICE_RUNTIME, broker,
@@ -94,6 +96,9 @@ struct Args {
     /// Listen http endpoint for management and metrics in the form host:port.
     #[arg(long, default_value = DEFAULT_HTTP_LISTEN_ADDR)]
     listen_http: String,
+    /// Listen https endpoint for management and metrics in the form host:port.
+    #[arg(long, default_value = None)]
+    listen_https: Option<String>,
     /// Advertised endpoint for receiving/sending WAL in the form host:port. If not
     /// specified, listen_pg is used to advertise instead.
     #[arg(long, default_value = None)]
@@ -203,6 +208,15 @@ struct Args {
     /// and the current position of the reader is smaller than this value.
     #[arg(long)]
     max_delta_for_fanout: Option<u64>,
+    /// Path to a file with certificate's private key for https API.
+    #[arg(long, default_value = DEFAULT_SSL_KEY_FILE)]
+    ssl_key_file: Utf8PathBuf,
+    /// Path to a file with a X509 certificate for https API.
+    #[arg(long, default_value = DEFAULT_SSL_CERT_FILE)]
+    ssl_cert_file: Utf8PathBuf,
+    /// Trusted root CA certificate to use in https APIs.
+    #[arg(long)]
+    ssl_ca_file: Option<Utf8PathBuf>,
 }
 
 // Like PathBufValueParser, but allows empty string.
@@ -336,12 +350,22 @@ async fn main() -> anyhow::Result<()> {
         }
     };
 
+    let ssl_ca_cert = match args.ssl_ca_file.as_ref() {
+        Some(ssl_ca_file) => {
+            tracing::info!("Using ssl root CA file: {ssl_ca_file:?}");
+            let buf = tokio::fs::read(ssl_ca_file).await?;
+            Some(Certificate::from_pem(&buf)?)
+        }
+        None => None,
+    };
+
     let conf = Arc::new(SafeKeeperConf {
         workdir,
         my_id: id,
         listen_pg_addr: args.listen_pg,
         listen_pg_addr_tenant_only: args.listen_pg_tenant_only,
         listen_http_addr: args.listen_http,
+        listen_https_addr: args.listen_https,
         advertise_pg_addr: args.advertise_pg,
         availability_zone: args.availability_zone,
         no_sync: args.no_sync,
@@ -368,6 +392,9 @@ async fn main() -> anyhow::Result<()> {
         eviction_min_resident: args.eviction_min_resident,
         wal_reader_fanout: args.wal_reader_fanout,
         max_delta_for_fanout: args.max_delta_for_fanout,
+        ssl_key_file: args.ssl_key_file,
+        ssl_cert_file: args.ssl_cert_file,
+        ssl_ca_cert,
     });
 
     // initialize sentry if SENTRY_DSN is provided
@@ -428,6 +455,17 @@ async fn start_safekeeper(conf: Arc<SafeKeeperConf>) -> Result<()> {
         e
     })?;
 
+    let https_listener = match conf.listen_https_addr.as_ref() {
+        Some(listen_https_addr) => {
+            info!("starting safekeeper HTTPS service on {}", listen_https_addr);
+            Some(tcp_listener::bind(listen_https_addr).map_err(|e| {
+                error!("failed to bind to address {}: {}", listen_https_addr, e);
+                e
+            })?)
+        }
+        None => None,
+    };
+
     let global_timelines = Arc::new(GlobalTimelines::new(conf.clone()));
 
     // Register metrics collector for active timelines. It's important to do this
@@ -501,7 +539,7 @@ async fn start_safekeeper(conf: Arc<SafeKeeperConf>) -> Result<()> {
     let http_handle = current_thread_rt
         .as_ref()
         .unwrap_or_else(|| HTTP_RUNTIME.handle())
-        .spawn(http::task_main(
+        .spawn(http::task_main_http(
             conf.clone(),
             http_listener,
             global_timelines.clone(),
@@ -509,6 +547,19 @@ async fn start_safekeeper(conf: Arc<SafeKeeperConf>) -> Result<()> {
         .map(|res| ("HTTP service main".to_owned(), res));
     tasks_handles.push(Box::pin(http_handle));
 
+    if let Some(https_listener) = https_listener {
+        let https_handle = current_thread_rt
+            .as_ref()
+            .unwrap_or_else(|| HTTP_RUNTIME.handle())
+            .spawn(http::task_main_https(
+                conf.clone(),
+                https_listener,
+                global_timelines.clone(),
+            ))
+            .map(|res| ("HTTPS service main".to_owned(), res));
+        tasks_handles.push(Box::pin(https_handle));
+    }
+
     let broker_task_handle = current_thread_rt
         .as_ref()
         .unwrap_or_else(|| BROKER_RUNTIME.handle())

safekeeper/src/http/mod.rs

@@ -3,10 +3,11 @@ use std::sync::Arc;
 
 pub use routes::make_router;
 pub use safekeeper_api::models;
+use tokio_util::sync::CancellationToken;
 
 use crate::{GlobalTimelines, SafeKeeperConf};
 
-pub async fn task_main(
+pub async fn task_main_http(
     conf: Arc<SafeKeeperConf>,
     http_listener: std::net::TcpListener,
     global_timelines: Arc<GlobalTimelines>,
@@ -14,8 +15,37 @@ pub async fn task_main(
     let router = make_router(conf, global_timelines)
         .build()
         .map_err(|err| anyhow::anyhow!(err))?;
-    let service = http_utils::RouterService::new(router).unwrap();
-    let server = hyper::Server::from_tcp(http_listener)?;
-    server.serve(service).await?;
+
+    let service = Arc::new(
+        http_utils::RequestServiceBuilder::new(router).map_err(|err| anyhow::anyhow!(err))?,
+    );
+    let server = http_utils::server::Server::new(service, http_listener, None)?;
+    server.serve(CancellationToken::new()).await?;
+    Ok(()) // unreachable
+}
+
+pub async fn task_main_https(
+    conf: Arc<SafeKeeperConf>,
+    https_listener: std::net::TcpListener,
+    global_timelines: Arc<GlobalTimelines>,
+) -> anyhow::Result<()> {
+    let certs = http_utils::tls_certs::load_cert_chain(&conf.ssl_cert_file)?;
+    let key = http_utils::tls_certs::load_private_key(&conf.ssl_key_file)?;
+
+    let server_config = rustls::ServerConfig::builder()
+        .with_no_client_auth()
+        .with_single_cert(certs, key)?;
+
+    let tls_acceptor = tokio_rustls::TlsAcceptor::from(Arc::new(server_config));
+
+    let router = make_router(conf, global_timelines)
+        .build()
+        .map_err(|err| anyhow::anyhow!(err))?;
+
+    let service = Arc::new(
+        http_utils::RequestServiceBuilder::new(router).map_err(|err| anyhow::anyhow!(err))?,
+    );
+    let server = http_utils::server::Server::new(service, https_listener, Some(tls_acceptor))?;
+    server.serve(CancellationToken::new()).await?;
     Ok(()) // unreachable
 }

safekeeper/src/http/routes.rs

@@ -17,7 +17,8 @@ use hyper::{Body, Request, Response, StatusCode};
 use postgres_ffi::WAL_SEGMENT_SIZE;
 use safekeeper_api::models::{
     AcceptorStateStatus, PullTimelineRequest, SafekeeperStatus, SkTimelineInfo, TermSwitchApiEntry,
-    TimelineCopyRequest, TimelineCreateRequest, TimelineStatus, TimelineTermBumpRequest,
+    TimelineCopyRequest, TimelineCreateRequest, TimelineDeleteResult, TimelineStatus,
+    TimelineTermBumpRequest,
 };
 use safekeeper_api::{ServerInfo, membership, models};
 use storage_broker::proto::{SafekeeperTimelineInfo, TenantTimelineId as ProtoTenantTimelineId};
@@ -32,7 +33,7 @@ use utils::lsn::Lsn;
 
 use crate::debug_dump::TimelineDigestRequest;
 use crate::safekeeper::TermLsn;
-use crate::timelines_global_map::{DeleteOrExclude, TimelineDeleteResult};
+use crate::timelines_global_map::DeleteOrExclude;
 use crate::{
     GlobalTimelines, SafeKeeperConf, copy_timeline, debug_dump, patch_control_file, pull_timeline,
 };
@@ -231,9 +232,14 @@ async fn timeline_pull_handler(mut request: Request<Body>) -> Result<Response<Bo
     let conf = get_conf(&request);
     let global_timelines = get_global_timelines(&request);
 
-    let resp = pull_timeline::handle_request(data, conf.sk_auth_token.clone(), global_timelines)
-        .await
-        .map_err(ApiError::InternalServerError)?;
+    let resp = pull_timeline::handle_request(
+        data,
+        conf.sk_auth_token.clone(),
+        conf.ssl_ca_cert.clone(),
+        global_timelines,
+    )
+    .await
+    .map_err(ApiError::InternalServerError)?;
     json_response(StatusCode::OK, resp)
 }
 

safekeeper/src/lib.rs

@@ -7,6 +7,7 @@ use std::time::Duration;
 use camino::Utf8PathBuf;
 use once_cell::sync::Lazy;
 use remote_storage::RemoteStorageConfig;
+use reqwest::Certificate;
 use storage_broker::Uri;
 use tokio::runtime::Runtime;
 use utils::auth::SwappableJwtAuth;
@@ -69,6 +70,9 @@ pub mod defaults {
     // before uploading a partial segment, so that in normal operation the eviction can happen
     // as soon as we have done the partial segment upload.
     pub const DEFAULT_EVICTION_MIN_RESIDENT: &str = DEFAULT_PARTIAL_BACKUP_TIMEOUT;
+
+    pub const DEFAULT_SSL_KEY_FILE: &str = "server.key";
+    pub const DEFAULT_SSL_CERT_FILE: &str = "server.crt";
 }
 
 #[derive(Debug, Clone)]
@@ -84,6 +88,7 @@ pub struct SafeKeeperConf {
     pub listen_pg_addr: String,
     pub listen_pg_addr_tenant_only: Option<String>,
     pub listen_http_addr: String,
+    pub listen_https_addr: Option<String>,
     pub advertise_pg_addr: Option<String>,
     pub availability_zone: Option<String>,
     pub no_sync: bool,
@@ -111,6 +116,9 @@ pub struct SafeKeeperConf {
     pub eviction_min_resident: Duration,
     pub wal_reader_fanout: bool,
     pub max_delta_for_fanout: Option<u64>,
+    pub ssl_key_file: Utf8PathBuf,
+    pub ssl_cert_file: Utf8PathBuf,
+    pub ssl_ca_cert: Option<Certificate>,
 }
 
 impl SafeKeeperConf {
@@ -127,6 +135,7 @@ impl SafeKeeperConf {
             listen_pg_addr: defaults::DEFAULT_PG_LISTEN_ADDR.to_string(),
             listen_pg_addr_tenant_only: None,
             listen_http_addr: defaults::DEFAULT_HTTP_LISTEN_ADDR.to_string(),
+            listen_https_addr: None,
             advertise_pg_addr: None,
             availability_zone: None,
             remote_storage: None,
@@ -155,6 +164,9 @@ impl SafeKeeperConf {
             eviction_min_resident: Duration::ZERO,
             wal_reader_fanout: false,
             max_delta_for_fanout: None,
+            ssl_key_file: Utf8PathBuf::from(defaults::DEFAULT_SSL_KEY_FILE),
+            ssl_cert_file: Utf8PathBuf::from(defaults::DEFAULT_SSL_CERT_FILE),
+            ssl_ca_cert: None,
         }
     }
 }

safekeeper/src/pull_timeline.rs

@@ -8,6 +8,7 @@ use camino::Utf8PathBuf;
 use chrono::{DateTime, Utc};
 use futures::{SinkExt, StreamExt, TryStreamExt};
 use postgres_ffi::{PG_TLI, XLogFileName, XLogSegNo};
+use reqwest::Certificate;
 use safekeeper_api::Term;
 use safekeeper_api::models::{PullTimelineRequest, PullTimelineResponse, TimelineStatus};
 use safekeeper_client::mgmt_api;
@@ -392,6 +393,7 @@ pub struct DebugDumpResponse {
 pub async fn handle_request(
     request: PullTimelineRequest,
     sk_auth_token: Option<SecretString>,
+    ssl_ca_cert: Option<Certificate>,
     global_timelines: Arc<GlobalTimelines>,
 ) -> Result<PullTimelineResponse> {
     let existing_tli = global_timelines.get(TenantTimelineId::new(
@@ -402,9 +404,11 @@ pub async fn handle_request(
         bail!("Timeline {} already exists", request.timeline_id);
     }
 
-    // TODO(DimasKovas): add ssl root CA certificate when implementing safekeeper's
-    // part of https support (#24836).
-    let http_client = reqwest::Client::new();
+    let mut http_client = reqwest::Client::builder();
+    if let Some(ssl_ca_cert) = ssl_ca_cert {
+        http_client = http_client.add_root_certificate(ssl_ca_cert);
+    }
+    let http_client = http_client.build()?;
 
     let http_hosts = request.http_hosts.clone();
 
@@ -441,13 +445,21 @@ pub async fn handle_request(
     assert!(status.tenant_id == request.tenant_id);
     assert!(status.timeline_id == request.timeline_id);
 
-    pull_timeline(status, safekeeper_host, sk_auth_token, global_timelines).await
+    pull_timeline(
+        status,
+        safekeeper_host,
+        sk_auth_token,
+        http_client,
+        global_timelines,
+    )
+    .await
 }
 
 async fn pull_timeline(
     status: TimelineStatus,
     host: String,
     sk_auth_token: Option<SecretString>,
+    http_client: reqwest::Client,
     global_timelines: Arc<GlobalTimelines>,
 ) -> Result<PullTimelineResponse> {
     let ttid = TenantTimelineId::new(status.tenant_id, status.timeline_id);
@@ -464,9 +476,6 @@ async fn pull_timeline(
     let conf = &global_timelines.get_global_config();
 
     let (_tmp_dir, tli_dir_path) = create_temp_timeline_dir(conf, ttid).await?;
-    // TODO(DimasKovas): add ssl root CA certificate when implementing safekeeper's
-    // part of https support (#24836).
-    let http_client = reqwest::Client::new();
     let client = Client::new(http_client, host.clone(), sk_auth_token.clone());
     // Request stream with basebackup archive.
     let bb_resp = client

safekeeper/src/timelines_global_map.rs

@@ -11,9 +11,8 @@ use anyhow::{Context, Result, bail};
 use camino::Utf8PathBuf;
 use camino_tempfile::Utf8TempDir;
 use safekeeper_api::membership::Configuration;
-use safekeeper_api::models::SafekeeperUtilization;
+use safekeeper_api::models::{SafekeeperUtilization, TimelineDeleteResult};
 use safekeeper_api::{ServerInfo, membership};
-use serde::Serialize;
 use tokio::fs;
 use tracing::*;
 use utils::crashsafe::{durable_rename, fsync_async_opt};
@@ -579,11 +578,6 @@ impl GlobalTimelines {
     }
 }
 
-#[derive(Clone, Copy, Serialize)]
-pub struct TimelineDeleteResult {
-    pub dir_existed: bool,
-}
-
 /// Action for delete_or_exclude.
 #[derive(Clone, Debug)]
 pub enum DeleteOrExclude {

safekeeper/tests/walproposer_sim/safekeeper.rs

@@ -152,6 +152,7 @@ pub fn run_server(os: NodeOs, disk: Arc<SafekeeperDisk>) -> Result<()> {
         my_id: NodeId(os.id() as u64),
         listen_pg_addr: String::new(),
         listen_http_addr: String::new(),
+        listen_https_addr: None,
         no_sync: false,
         broker_endpoint: "/".parse::<Uri>().unwrap(),
         broker_keepalive_interval: Duration::from_secs(0),
@@ -179,6 +180,9 @@ pub fn run_server(os: NodeOs, disk: Arc<SafekeeperDisk>) -> Result<()> {
         eviction_min_resident: Duration::ZERO,
         wal_reader_fanout: false,
         max_delta_for_fanout: None,
+        ssl_key_file: Utf8PathBuf::from(""),
+        ssl_cert_file: Utf8PathBuf::from(""),
+        ssl_ca_cert: None,
     };
 
     let mut global = GlobalMap::new(disk, conf.clone())?;

storage_controller/src/http.rs

@@ -24,7 +24,7 @@ use pageserver_api::controller_api::{
     ShardsPreferredAzsRequest, TenantCreateRequest, TenantPolicyRequest, TenantShardMigrateRequest,
 };
 use pageserver_api::models::{
-    TenantConfigPatchRequest, TenantConfigRequest, TenantLocationConfigRequest,
+    DetachBehavior, TenantConfigPatchRequest, TenantConfigRequest, TenantLocationConfigRequest,
     TenantShardSplitRequest, TenantTimeTravelRequest, TimelineArchivalConfigRequest,
     TimelineCreateRequest,
 };
@@ -525,6 +525,7 @@ async fn handle_tenant_timeline_detach_ancestor(
 ) -> Result<Response<Body>, ApiError> {
     let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
     let timeline_id: TimelineId = parse_request_param(&req, "timeline_id")?;
+    let behavior: Option<DetachBehavior> = parse_query_param(&req, "detach_behavior")?;
 
     check_permissions(&req, Scope::PageServerApi)?;
     maybe_rate_limit(&req, tenant_id).await;
@@ -537,7 +538,7 @@ async fn handle_tenant_timeline_detach_ancestor(
     };
 
     let res = service
-        .tenant_timeline_detach_ancestor(tenant_id, timeline_id)
+        .tenant_timeline_detach_ancestor(tenant_id, timeline_id, behavior)
         .await?;
 
     json_response(StatusCode::OK, res)

storage_controller/src/pageserver_client.rs

@@ -1,9 +1,9 @@
 use pageserver_api::models::detach_ancestor::AncestorDetached;
 use pageserver_api::models::{
-    LocationConfig, LocationConfigListResponse, PageserverUtilization, SecondaryProgress,
-    TenantScanRemoteStorageResponse, TenantShardSplitRequest, TenantShardSplitResponse,
-    TenantWaitLsnRequest, TimelineArchivalConfigRequest, TimelineCreateRequest, TimelineInfo,
-    TopTenantShardsRequest, TopTenantShardsResponse,
+    DetachBehavior, LocationConfig, LocationConfigListResponse, PageserverUtilization,
+    SecondaryProgress, TenantScanRemoteStorageResponse, TenantShardSplitRequest,
+    TenantShardSplitResponse, TenantWaitLsnRequest, TimelineArchivalConfigRequest,
+    TimelineCreateRequest, TimelineInfo, TopTenantShardsRequest, TopTenantShardsResponse,
 };
 use pageserver_api::shard::TenantShardId;
 use pageserver_client::BlockUnblock;
@@ -252,13 +252,14 @@ impl PageserverClient {
         &self,
         tenant_shard_id: TenantShardId,
         timeline_id: TimelineId,
+        behavior: Option<DetachBehavior>,
     ) -> Result<AncestorDetached> {
         measured_request!(
             "timeline_detach_ancestor",
             crate::metrics::Method::Put,
             &self.node_id_label,
             self.inner
-                .timeline_detach_ancestor(tenant_shard_id, timeline_id)
+                .timeline_detach_ancestor(tenant_shard_id, timeline_id, behavior)
                 .await
         )
     }

storage_controller/src/safekeeper_client.rs

@@ -1,5 +1,5 @@
 use safekeeper_api::models::{
-    PullTimelineRequest, PullTimelineResponse, SafekeeperUtilization, TimelineCreateRequest,
+    self, PullTimelineRequest, PullTimelineResponse, SafekeeperUtilization, TimelineCreateRequest,
     TimelineStatus,
 };
 use safekeeper_client::mgmt_api::{Client, Result};
@@ -69,11 +69,28 @@ impl SafekeeperClient {
         )
     }
 
+    #[allow(unused)]
+    pub(crate) async fn exclude_timeline(
+        &self,
+        tenant_id: TenantId,
+        timeline_id: TimelineId,
+        req: &models::TimelineMembershipSwitchRequest,
+    ) -> Result<models::TimelineDeleteResult> {
+        measured_request!(
+            "exclude_timeline",
+            crate::metrics::Method::Post,
+            &self.node_id_label,
+            self.inner
+                .exclude_timeline(tenant_id, timeline_id, req)
+                .await
+        )
+    }
+
     pub(crate) async fn delete_timeline(
         &self,
         tenant_id: TenantId,
         timeline_id: TimelineId,
-    ) -> Result<TimelineStatus> {
+    ) -> Result<models::TimelineDeleteResult> {
         measured_request!(
             "delete_timeline",
             crate::metrics::Method::Delete,
@@ -94,6 +111,23 @@ impl SafekeeperClient {
         )
     }
 
+    #[allow(unused)]
+    pub(crate) async fn bump_timeline_term(
+        &self,
+        tenant_id: TenantId,
+        timeline_id: TimelineId,
+        req: &models::TimelineTermBumpRequest,
+    ) -> Result<models::TimelineTermBumpResponse> {
+        measured_request!(
+            "term_bump",
+            crate::metrics::Method::Post,
+            &self.node_id_label,
+            self.inner
+                .bump_timeline_term(tenant_id, timeline_id, req)
+                .await
+        )
+    }
+
     pub(crate) async fn get_utilization(&self) -> Result<SafekeeperUtilization> {
         measured_request!(
             "utilization",

storage_controller/src/service.rs

@@ -34,9 +34,9 @@ use pageserver_api::controller_api::{
     TenantShardMigrateResponse,
 };
 use pageserver_api::models::{
-    self, LocationConfig, LocationConfigListResponse, LocationConfigMode, PageserverUtilization,
-    SafekeeperInfo, SafekeepersInfo, SecondaryProgress, ShardParameters, TenantConfig,
-    TenantConfigPatchRequest, TenantConfigRequest, TenantLocationConfigRequest,
+    self, DetachBehavior, LocationConfig, LocationConfigListResponse, LocationConfigMode,
+    PageserverUtilization, SafekeeperInfo, SafekeepersInfo, SecondaryProgress, ShardParameters,
+    TenantConfig, TenantConfigPatchRequest, TenantConfigRequest, TenantLocationConfigRequest,
     TenantLocationConfigResponse, TenantShardLocation, TenantShardSplitRequest,
     TenantShardSplitResponse, TenantSorting, TenantTimeTravelRequest,
     TimelineArchivalConfigRequest, TimelineCreateRequest, TimelineCreateResponseStorcon,
@@ -4041,6 +4041,7 @@ impl Service {
         &self,
         tenant_id: TenantId,
         timeline_id: TimelineId,
+        behavior: Option<DetachBehavior>,
     ) -> Result<models::detach_ancestor::AncestorDetached, ApiError> {
         tracing::info!("Detaching timeline {tenant_id}/{timeline_id}",);
 
@@ -4064,6 +4065,7 @@ impl Service {
                 node: Node,
                 jwt: Option<String>,
                 ssl_ca_cert: Option<Certificate>,
+                behavior: Option<DetachBehavior>,
             ) -> Result<(ShardNumber, models::detach_ancestor::AncestorDetached), ApiError> {
                 tracing::info!(
                     "Detaching timeline on shard {tenant_shard_id}/{timeline_id}, attached to node {node}",
@@ -4073,7 +4075,7 @@ impl Service {
                     .map_err(|e| passthrough_api_error(&node, e))?;
 
                 client
-                    .timeline_detach_ancestor(tenant_shard_id, timeline_id)
+                    .timeline_detach_ancestor(tenant_shard_id, timeline_id, behavior)
                     .await
                     .map_err(|e| {
                         use mgmt_api::Error;
@@ -4111,6 +4113,7 @@ impl Service {
                         node,
                         self.config.pageserver_jwt_token.clone(),
                         self.config.ssl_ca_cert.clone(),
+                        behavior,
                     ))
                 })
                 .await?;
@@ -4265,7 +4268,8 @@ impl Service {
 
     /// Helper for concurrently calling a pageserver API on a number of shards, such as timeline creation.
     ///
-    /// On success, the returned vector contains exactly the same number of elements as the input `locations`.
+    /// On success, the returned vector contains exactly the same number of elements as the input `locations`
+    /// and returned element at index `i` is the result for `req_fn(op(locations[i])`.
     async fn tenant_for_shards<F, R>(
         &self,
         locations: Vec<(TenantShardId, Node)>,
@@ -4281,18 +4285,23 @@ impl Service {
         let mut futs = FuturesUnordered::new();
         let mut results = Vec::with_capacity(locations.len());
 
-        for (tenant_shard_id, node) in locations {
-            futs.push(req_fn(tenant_shard_id, node));
+        for (idx, (tenant_shard_id, node)) in locations.into_iter().enumerate() {
+            let fut = req_fn(tenant_shard_id, node);
+            futs.push(async move { (idx, fut.await) });
         }
 
-        while let Some(r) = futs.next().await {
-            results.push(r?);
+        while let Some((idx, r)) = futs.next().await {
+            results.push((idx, r?));
         }
 
-        Ok(results)
+        results.sort_by_key(|(idx, _)| *idx);
+        Ok(results.into_iter().map(|(_, r)| r).collect())
     }
 
-    /// Concurrently invoke a pageserver API call on many shards at once
+    /// Concurrently invoke a pageserver API call on many shards at once.
+    ///
+    /// The returned Vec has the same length as the `locations` Vec,
+    /// and returned element at index `i` is the result for `op(locations[i])`.
     pub(crate) async fn tenant_for_shards_api<T, O, F>(
         &self,
         locations: Vec<(TenantShardId, Node)>,
@@ -4309,27 +4318,29 @@ impl Service {
         let mut futs = FuturesUnordered::new();
         let mut results = Vec::with_capacity(locations.len());
 
-        for (tenant_shard_id, node) in locations {
+        for (idx, (tenant_shard_id, node)) in locations.into_iter().enumerate() {
             futs.push(async move {
-                node.with_client_retries(
-                    |client| op(tenant_shard_id, client),
-                    &self.config.pageserver_jwt_token,
-                    &self.config.ssl_ca_cert,
-                    warn_threshold,
-                    max_retries,
-                    timeout,
-                    cancel,
-                )
-                .await
+                let r = node
+                    .with_client_retries(
+                        |client| op(tenant_shard_id, client),
+                        &self.config.pageserver_jwt_token,
+                        &self.config.ssl_ca_cert,
+                        warn_threshold,
+                        max_retries,
+                        timeout,
+                        cancel,
+                    )
+                    .await;
+                (idx, r)
             });
         }
 
-        while let Some(r) = futs.next().await {
-            let r = r.unwrap_or(Err(mgmt_api::Error::Cancelled));
-            results.push(r);
+        while let Some((idx, r)) = futs.next().await {
+            results.push((idx, r.unwrap_or(Err(mgmt_api::Error::Cancelled))));
         }
 
-        results
+        results.sort_by_key(|(idx, _)| *idx);
+        results.into_iter().map(|(_, r)| r).collect()
     }
 
     /// Helper for safely working with the shards in a tenant remotely on pageservers, for example

test_runner/fixtures/neon_fixtures.py

@@ -466,6 +466,9 @@ class NeonEnvBuilder:
         # Flag to enable https listener in pageserver, generate local ssl certs,
         # and force storage controller to use https for pageserver api.
         self.use_https_pageserver_api: bool = False
+        # Flag to enable https listener in safekeeper, generate local ssl certs,
+        # and force storage controller to use https for safekeeper api.
+        self.use_https_safekeeper_api: bool = False
 
         self.pageserver_virtual_file_io_engine: str | None = pageserver_virtual_file_io_engine
         self.pageserver_get_vectored_concurrent_io: str | None = (
@@ -1063,7 +1066,9 @@ class NeonEnv:
         self.initial_tenant = config.initial_tenant
         self.initial_timeline = config.initial_timeline
 
-        self.generate_local_ssl_certs = config.use_https_pageserver_api
+        self.generate_local_ssl_certs = (
+            config.use_https_pageserver_api or config.use_https_safekeeper_api
+        )
         self.ssl_ca_file = (
             self.repo_dir.joinpath("rootCA.crt") if self.generate_local_ssl_certs else None
         )
@@ -1146,6 +1151,10 @@ class NeonEnv:
             storage_controller_config = storage_controller_config or {}
             storage_controller_config["use_https_pageserver_api"] = True
 
+        if config.use_https_safekeeper_api:
+            storage_controller_config = storage_controller_config or {}
+            storage_controller_config["use_https_safekeeper_api"] = True
+
         if storage_controller_config is not None:
             cfg["storage_controller"] = storage_controller_config
 
@@ -1248,6 +1257,7 @@ class NeonEnv:
                 pg=self.port_distributor.get_port(),
                 pg_tenant_only=self.port_distributor.get_port(),
                 http=self.port_distributor.get_port(),
+                https=self.port_distributor.get_port() if config.use_https_safekeeper_api else None,
             )
             id = config.safekeepers_id_start + i  # assign ids sequentially
             sk_cfg: dict[str, Any] = {
@@ -1255,6 +1265,7 @@ class NeonEnv:
                 "pg_port": port.pg,
                 "pg_tenant_only_port": port.pg_tenant_only,
                 "http_port": port.http,
+                "https_port": port.https,
                 "sync": config.safekeepers_enable_fsync,
             }
             if config.auth_enabled:
@@ -4475,6 +4486,7 @@ class SafekeeperPort:
     pg: int
     pg_tenant_only: int
     http: int
+    https: int | None
 
 
 @dataclass

test_runner/fixtures/pageserver/http.py

@@ -1070,14 +1070,14 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
         tenant_id: TenantId | TenantShardId,
         timeline_id: TimelineId,
         batch_size: int | None = None,
-        behavior_v2: bool = False,
+        detach_behavior: str | None = None,
         **kwargs,
     ) -> set[TimelineId]:
         params: dict[str, Any] = {}
         if batch_size is not None:
             params["batch_size"] = batch_size
-        if behavior_v2:
-            params["detach_behavior"] = "v2"
+        if detach_behavior:
+            params["detach_behavior"] = detach_behavior
         res = self.put(
             f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}/detach_ancestor",
             params=params,

test_runner/regress/test_attach_tenant_config.py

@@ -144,7 +144,7 @@ def test_fully_custom_config(positive_env: NeonEnv):
         "compaction_l0_semaphore": False,
         "l0_flush_delay_threshold": 25,
         "l0_flush_stall_threshold": 42,
-        "l0_flush_wait_upload": False,
+        "l0_flush_wait_upload": True,
         "compaction_target_size": 1048576,
         "checkpoint_distance": 10000,
         "checkpoint_timeout": "13m",

test_runner/regress/test_branching.py

@@ -19,6 +19,7 @@ from fixtures.pageserver.utils import wait_until_tenant_active
 from fixtures.utils import query_scalar
 from performance.test_perf_pgbench import get_scales_matrix
 from requests import RequestException
+from requests.exceptions import RetryError
 
 
 # Test branch creation
@@ -180,7 +181,6 @@ def test_cannot_create_endpoint_on_non_uploaded_timeline(neon_env_builder: NeonE
             env.endpoints.create_start(
                 initial_branch, tenant_id=env.initial_tenant, basebackup_request_tries=2
             )
-        ps_http.configure_failpoints(("before-upload-index-pausable", "off"))
     finally:
         env.pageserver.stop(immediate=True)
 
@@ -221,10 +221,7 @@ def test_cannot_branch_from_non_uploaded_branch(neon_env_builder: NeonEnvBuilder
 
         branch_id = TimelineId.generate()
 
-        with pytest.raises(
-            PageserverApiException,
-            match="Cannot branch off the timeline that's not present in pageserver",
-        ):
+        with pytest.raises(RetryError, match="too many 503 error responses"):
             ps_http.timeline_create(
                 env.pg_version,
                 env.initial_tenant,

test_runner/regress/test_remote_storage.py

@@ -29,7 +29,6 @@ from fixtures.remote_storage import (
 from fixtures.utils import (
     assert_eq,
     assert_ge,
-    assert_gt,
     print_gc_result,
     query_scalar,
     wait_until,
@@ -334,14 +333,12 @@ def test_remote_storage_upload_queue_retries(
     # Exponential back-off in upload queue, so, gracious timeouts.
 
     wait_until(
-        lambda: assert_gt(get_queued_count(file_kind="layer", op_kind="upload"), 0), timeout=30
+        lambda: assert_ge(get_queued_count(file_kind="layer", op_kind="upload"), 1), timeout=30
     )
     wait_until(
         lambda: assert_ge(get_queued_count(file_kind="index", op_kind="upload"), 1), timeout=30
     )
-    wait_until(
-        lambda: assert_eq(get_queued_count(file_kind="layer", op_kind="delete"), 0), timeout=30
-    )
+    # There may or may not be deletes queued up behind conflicting uploads; don't check.
 
     # unblock churn operations
     configure_storage_sync_failpoints("off")
@@ -786,54 +783,6 @@ def test_empty_branch_remote_storage_upload_on_restart(neon_env_builder: NeonEnv
         create_thread.join()
 
 
-def test_paused_upload_stalls_checkpoint(
-    neon_env_builder: NeonEnvBuilder,
-):
-    """
-    This test checks that checkpoints block on uploads to remote storage.
-    """
-    neon_env_builder.enable_pageserver_remote_storage(RemoteStorageKind.LOCAL_FS)
-
-    env = neon_env_builder.init_start(
-        initial_tenant_conf={
-            # Set a small compaction threshold
-            "compaction_threshold": "3",
-            # Disable GC
-            "gc_period": "0s",
-            # disable PITR
-            "pitr_interval": "0s",
-        }
-    )
-
-    env.pageserver.allowed_errors.append(
-        f".*PUT.* path=/v1/tenant/{env.initial_tenant}/timeline.* request was dropped before completing"
-    )
-
-    tenant_id = env.initial_tenant
-    timeline_id = env.initial_timeline
-
-    client = env.pageserver.http_client()
-    layers_at_creation = client.layer_map_info(tenant_id, timeline_id)
-    deltas_at_creation = len(layers_at_creation.delta_layers())
-    assert (
-        deltas_at_creation == 1
-    ), "are you fixing #5863? make sure we end up with 2 deltas at the end of endpoint lifecycle"
-
-    # Make new layer uploads get stuck.
-    # Note that timeline creation waits for the initial layers to reach remote storage.
-    # So at this point, the `layers_at_creation` are in remote storage.
-    client.configure_failpoints(("before-upload-layer-pausable", "pause"))
-
-    with env.endpoints.create_start("main", tenant_id=tenant_id) as endpoint:
-        # Build two tables with some data inside
-        endpoint.safe_psql("CREATE TABLE foo AS SELECT x FROM generate_series(1, 10000) g(x)")
-        wait_for_last_flush_lsn(env, endpoint, tenant_id, timeline_id)
-
-        with pytest.raises(ReadTimeout):
-            client.timeline_checkpoint(tenant_id, timeline_id, timeout=5)
-        client.configure_failpoints(("before-upload-layer-pausable", "off"))
-
-
 def wait_upload_queue_empty(
     client: PageserverHttpClient, tenant_id: TenantId, timeline_id: TimelineId
 ):

test_runner/regress/test_ssl.py

@@ -1,5 +1,7 @@
+import pytest
 import requests
-from fixtures.neon_fixtures import NeonEnvBuilder
+from fixtures.neon_fixtures import NeonEnvBuilder, StorageControllerApiException
+from fixtures.utils import wait_until
 
 
 def test_pageserver_https_api(neon_env_builder: NeonEnvBuilder):
@@ -13,3 +15,54 @@ def test_pageserver_https_api(neon_env_builder: NeonEnvBuilder):
 
     addr = f"https://localhost:{env.pageserver.service_port.https}/v1/status"
     requests.get(addr, verify=str(env.ssl_ca_file)).raise_for_status()
+
+
+def test_safekeeper_https_api(neon_env_builder: NeonEnvBuilder):
+    """
+    Test HTTPS safekeeper management API.
+    1. Make /v1/status request to HTTPS API to ensure it's appropriately configured.
+    2. Try to register safekeeper in storcon with https port missing.
+    3. Register safekeeper with https port.
+    4. Wait for a heartbeat round to complete.
+    """
+    neon_env_builder.use_https_safekeeper_api = True
+    env = neon_env_builder.init_start()
+
+    sk = env.safekeepers[0]
+
+    # 1. Make simple https request.
+    addr = f"https://localhost:{sk.port.https}/v1/status"
+    requests.get(addr, verify=str(env.ssl_ca_file)).raise_for_status()
+
+    # Note: http_port is intentionally wrong.
+    # Storcon should not use it if use_https is on.
+    http_port = 0
+
+    body = {
+        "active": True,
+        "id": sk.id,
+        "created_at": "2023-10-25T09:11:25Z",
+        "updated_at": "2024-08-28T11:32:43Z",
+        "region_id": "aws-us-east-2",
+        "host": "localhost",
+        "port": sk.port.pg,
+        "http_port": http_port,
+        "https_port": None,
+        "version": 5957,
+        "availability_zone_id": "us-east-2b",
+    }
+    # 2. Try register with https port missing.
+    with pytest.raises(StorageControllerApiException, match="https port is not specified"):
+        env.storage_controller.on_safekeeper_deploy(sk.id, body)
+
+    # 3. Register with https port.
+    body["https_port"] = sk.port.https
+    env.storage_controller.on_safekeeper_deploy(sk.id, body)
+
+    # 4. Wait for hearbeat round complete.
+    def storcon_heartbeat():
+        assert env.storage_controller.log_contains(
+            "Heartbeat round complete for 1 safekeepers, 0 offline"
+        )
+
+    wait_until(storcon_heartbeat)

test_runner/regress/test_timeline_detach_ancestor.py

@@ -407,7 +407,9 @@ def test_ancestor_detach_behavior_v2(neon_env_builder: NeonEnvBuilder):
 
     after = env.create_branch("after", ancestor_branch_name="main", ancestor_start_lsn=None)
 
-    all_reparented = client.detach_ancestor(env.initial_tenant, branch_to_detach, behavior_v2=True)
+    all_reparented = client.detach_ancestor(
+        env.initial_tenant, branch_to_detach, detach_behavior="v2"
+    )
     assert set(all_reparented) == set()
 
     env.pageserver.quiesce_tenants()
@@ -1350,8 +1352,10 @@ def test_sharded_tad_interleaved_after_partial_success(neon_env_builder: NeonEnv
         )
 
 
+@pytest.mark.parametrize("detach_behavior", ["default", "v1", "v2"])
 def test_retryable_500_hit_through_storcon_during_timeline_detach_ancestor(
     neon_env_builder: NeonEnvBuilder,
+    detach_behavior: str,
 ):
     shard_count = 2
     neon_env_builder.num_pageservers = shard_count
@@ -1390,7 +1394,11 @@ def test_retryable_500_hit_through_storcon_during_timeline_detach_ancestor(
     victim_http.configure_failpoints([(pausepoint, "pause"), (failpoint, "return")])
 
     def detach_timeline():
-        http.detach_ancestor(env.initial_tenant, detached_branch)
+        http.detach_ancestor(
+            env.initial_tenant,
+            detached_branch,
+            detach_behavior=detach_behavior if detach_behavior != "default" else None,
+        )
 
     def paused_at_failpoint():
         stuck.assert_log_contains(f"at failpoint {pausepoint}")

test_runner/regress/test_wal_acceptor.py

@@ -1427,6 +1427,7 @@ class SafekeeperEnv:
             pg=self.port_distributor.get_port(),
             pg_tenant_only=self.port_distributor.get_port(),
             http=self.port_distributor.get_port(),
+            https=None,
         )
 
         safekeeper_dir = self.repo_dir / f"sk{i}"