Commit latest script updates

Cargo.lock

@@ -1348,7 +1348,6 @@ dependencies = [
  "tokio-postgres",
  "tokio-util",
  "toml",
- "toml_edit",
  "tracing",
  "url",
  "utils",
@@ -5195,6 +5194,7 @@ dependencies = [
  "rand 0.8.5",
  "remote_storage",
  "reqwest 0.12.4",
+ "safekeeper",
  "serde",
  "serde_json",
  "serde_with",
@@ -5857,8 +5857,6 @@ dependencies = [
  "routerify",
  "serde",
  "serde_json",
- "strum",
- "strum_macros",
  "thiserror",
  "tokio",
  "tokio-util",

Dockerfile.build-tools

@@ -65,7 +65,7 @@ RUN curl -sL "https://github.com/peak/s5cmd/releases/download/v${S5CMD_VERSION}/
     && mv s5cmd /usr/local/bin/s5cmd
 
 # LLVM
-ENV LLVM_VERSION=18
+ENV LLVM_VERSION=17
 RUN curl -fsSL 'https://apt.llvm.org/llvm-snapshot.gpg.key' | apt-key add - \
     && echo "deb http://apt.llvm.org/bullseye/ llvm-toolchain-bullseye-${LLVM_VERSION} main" > /etc/apt/sources.list.d/llvm.stable.list \
     && apt update \
@@ -141,7 +141,7 @@ WORKDIR /home/nonroot
 
 # Rust
 # Please keep the version of llvm (installed above) in sync with rust llvm (`rustc --version --verbose | grep LLVM`)
-ENV RUSTC_VERSION=1.78.0
+ENV RUSTC_VERSION=1.77.0
 ENV RUSTUP_HOME="/home/nonroot/.rustup"
 ENV PATH="/home/nonroot/.cargo/bin:${PATH}"
 RUN curl -sSO https://static.rust-lang.org/rustup/dist/$(uname -m)-unknown-linux-gnu/rustup-init && whoami && \

compute_tools/src/bin/compute_ctl.rs

@@ -47,7 +47,7 @@ use chrono::Utc;
 use clap::Arg;
 use signal_hook::consts::{SIGQUIT, SIGTERM};
 use signal_hook::{consts::SIGINT, iterator::Signals};
-use tracing::{error, info, warn};
+use tracing::{error, info};
 use url::Url;
 
 use compute_api::responses::ComputeStatus;
@@ -62,7 +62,6 @@ use compute_tools::logger::*;
 use compute_tools::monitor::launch_monitor;
 use compute_tools::params::*;
 use compute_tools::spec::*;
-use compute_tools::swap::resize_swap;
 
 // this is an arbitrary build tag. Fine as a default / for testing purposes
 // in-case of not-set environment var
@@ -111,7 +110,6 @@ fn main() -> Result<()> {
         .expect("Postgres connection string is required");
     let spec_json = matches.get_one::<String>("spec");
     let spec_path = matches.get_one::<String>("spec-path");
-    let resize_swap_on_bind = matches.get_flag("resize-swap-on-bind");
 
     // Extract OpenTelemetry context for the startup actions from the
     // TRACEPARENT and TRACESTATE env variables, and attach it to the current
@@ -228,14 +226,14 @@ fn main() -> Result<()> {
 
     // If this is a pooled VM, prewarm before starting HTTP server and becoming
     // available for binding. Prewarming helps Postgres start quicker later,
-    // because QEMU will already have its memory allocated from the host, and
+    // because QEMU will already have it's memory allocated from the host, and
     // the necessary binaries will already be cached.
     if !spec_set {
         compute.prewarm_postgres()?;
     }
 
-    // Launch http service first, so that we can serve control-plane requests
-    // while configuration is still in progress.
+    // Launch http service first, so we were able to serve control-plane
+    // requests, while configuration is still in progress.
     let _http_handle =
         launch_http_server(http_port, &compute).expect("cannot launch http endpoint thread");
 
@@ -255,22 +253,21 @@ fn main() -> Result<()> {
                 break;
             }
         }
-
-        // Record for how long we slept waiting for the spec.
-        let now = Utc::now();
-        state.metrics.wait_for_spec_ms = now
-            .signed_duration_since(state.start_time)
-            .to_std()
-            .unwrap()
-            .as_millis() as u64;
-
-        // Reset start time, so that the total startup time that is calculated later will
-        // not include the time that we waited for the spec.
-        state.start_time = now;
     }
 
     // We got all we need, update the state.
     let mut state = compute.state.lock().unwrap();
+
+    // Record for how long we slept waiting for the spec.
+    state.metrics.wait_for_spec_ms = Utc::now()
+        .signed_duration_since(state.start_time)
+        .to_std()
+        .unwrap()
+        .as_millis() as u64;
+    // Reset start time to the actual start of the configuration, so that
+    // total startup time was properly measured at the end.
+    state.start_time = Utc::now();
+
     state.status = ComputeStatus::Init;
     compute.state_changed.notify_all();
 
@@ -278,72 +275,33 @@ fn main() -> Result<()> {
         "running compute with features: {:?}",
         state.pspec.as_ref().unwrap().spec.features
     );
-    // before we release the mutex, fetch the swap size (if any) for later.
-    let swap_size_bytes = state.pspec.as_ref().unwrap().spec.swap_size_bytes;
     drop(state);
 
     // Launch remaining service threads
     let _monitor_handle = launch_monitor(&compute);
     let _configurator_handle = launch_configurator(&compute);
 
-    let mut prestartup_failed = false;
-    let mut delay_exit = false;
-
-    // Resize swap to the desired size if the compute spec says so
-    if let (Some(size_bytes), true) = (swap_size_bytes, resize_swap_on_bind) {
-        // To avoid 'swapoff' hitting postgres startup, we need to run resize-swap to completion
-        // *before* starting postgres.
-        //
-        // In theory, we could do this asynchronously if SkipSwapon was enabled for VMs, but this
-        // carries a risk of introducing hard-to-debug issues - e.g. if postgres sometimes gets
-        // OOM-killed during startup because swap wasn't available yet.
-        match resize_swap(size_bytes) {
-            Ok(()) => {
-                let size_gib = size_bytes as f32 / (1 << 20) as f32; // just for more coherent display.
-                info!(%size_bytes, %size_gib, "resized swap");
-            }
-            Err(err) => {
-                let err = err.context("failed to resize swap");
-                error!("{err:#}");
-
-                // Mark compute startup as failed; don't try to start postgres, and report this
-                // error to the control plane when it next asks.
-                prestartup_failed = true;
-                let mut state = compute.state.lock().unwrap();
-                state.error = Some(format!("{err:?}"));
-                state.status = ComputeStatus::Failed;
-                compute.state_changed.notify_all();
-                delay_exit = true;
-            }
-        }
-    }
-
     // Start Postgres
-    let mut pg = None;
+    let mut delay_exit = false;
     let mut exit_code = None;
-
-    if !prestartup_failed {
-        pg = match compute.start_compute(extension_server_port) {
-            Ok(pg) => Some(pg),
-            Err(err) => {
-                error!("could not start the compute node: {:#}", err);
-                let mut state = compute.state.lock().unwrap();
-                state.error = Some(format!("{:?}", err));
-                state.status = ComputeStatus::Failed;
-                // Notify others that Postgres failed to start. In case of configuring the
-                // empty compute, it's likely that API handler is still waiting for compute
-                // state change. With this we will notify it that compute is in Failed state,
-                // so control plane will know about it earlier and record proper error instead
-                // of timeout.
-                compute.state_changed.notify_all();
-                drop(state); // unlock
-                delay_exit = true;
-                None
-            }
-        };
-    } else {
-        warn!("skipping postgres startup because pre-startup step failed");
-    }
+    let pg = match compute.start_compute(extension_server_port) {
+        Ok(pg) => Some(pg),
+        Err(err) => {
+            error!("could not start the compute node: {:#}", err);
+            let mut state = compute.state.lock().unwrap();
+            state.error = Some(format!("{:?}", err));
+            state.status = ComputeStatus::Failed;
+            // Notify others that Postgres failed to start. In case of configuring the
+            // empty compute, it's likely that API handler is still waiting for compute
+            // state change. With this we will notify it that compute is in Failed state,
+            // so control plane will know about it earlier and record proper error instead
+            // of timeout.
+            compute.state_changed.notify_all();
+            drop(state); // unlock
+            delay_exit = true;
+            None
+        }
+    };
 
     // Start the vm-monitor if directed to. The vm-monitor only runs on linux
     // because it requires cgroups.
@@ -568,11 +526,6 @@ fn cli() -> clap::Command {
                 )
                 .value_name("FILECACHE_CONNSTR"),
         )
-        .arg(
-            Arg::new("resize-swap-on-bind")
-                .long("resize-swap-on-bind")
-                .action(clap::ArgAction::SetTrue),
-        )
 }
 
 /// When compute_ctl is killed, send also termination signal to sync-safekeepers

compute_tools/src/lib.rs

@@ -14,5 +14,4 @@ pub mod monitor;
 pub mod params;
 pub mod pg_helpers;
 pub mod spec;
-pub mod swap;
 pub mod sync_sk;

compute_tools/src/spec.rs

@@ -490,7 +490,7 @@ pub fn handle_databases(spec: &ComputeSpec, client: &mut Client) -> Result<()> {
                 "rename_db" => {
                     let new_name = op.new_name.as_ref().unwrap();
 
-                    if existing_dbs.contains_key(&op.name) {
+                    if existing_dbs.get(&op.name).is_some() {
                         let query: String = format!(
                             "ALTER DATABASE {} RENAME TO {}",
                             op.name.pg_quote(),

compute_tools/src/swap.rs

@@ -1,36 +0,0 @@
-use anyhow::{anyhow, Context};
-use tracing::warn;
-
-pub const RESIZE_SWAP_BIN: &str = "/neonvm/bin/resize-swap";
-
-pub fn resize_swap(size_bytes: u64) -> anyhow::Result<()> {
-    // run `/neonvm/bin/resize-swap --once {size_bytes}`
-    //
-    // Passing '--once' causes resize-swap to delete itself after successful completion, which
-    // means that if compute_ctl restarts later, we won't end up calling 'swapoff' while
-    // postgres is running.
-    //
-    // NOTE: resize-swap is not very clever. If present, --once MUST be the first arg.
-    let child_result = std::process::Command::new("/usr/bin/sudo")
-        .arg(RESIZE_SWAP_BIN)
-        .arg("--once")
-        .arg(size_bytes.to_string())
-        .spawn();
-
-    if matches!(&child_result, Err(e) if e.kind() == std::io::ErrorKind::NotFound) {
-        warn!("ignoring \"not found\" error from resize-swap to avoid swapoff while compute is running");
-        return Ok(());
-    }
-
-    child_result
-        .context("spawn() failed")
-        .and_then(|mut child| child.wait().context("wait() failed"))
-        .and_then(|status| match status.success() {
-            true => Ok(()),
-            false => Err(anyhow!("process exited with {status}")),
-        })
-        // wrap any prior error with the overall context that we couldn't run the command
-        .with_context(|| {
-            format!("could not run `/usr/bin/sudo {RESIZE_SWAP_BIN} --once {size_bytes}`")
-        })
-}

control_plane/Cargo.toml

@@ -28,7 +28,6 @@ serde_with.workspace = true
 tar.workspace = true
 thiserror.workspace = true
 toml.workspace = true
-toml_edit.workspace = true
 tokio.workspace = true
 tokio-postgres.workspace = true
 tokio-util.workspace = true

control_plane/src/bin/neon_local.rs

@@ -14,15 +14,15 @@ use control_plane::pageserver::{PageServerNode, PAGESERVER_REMOTE_STORAGE_DIR};
 use control_plane::safekeeper::SafekeeperNode;
 use control_plane::storage_controller::StorageController;
 use control_plane::{broker, local_env};
-use pageserver_api::config::{
-    DEFAULT_HTTP_LISTEN_PORT as DEFAULT_PAGESERVER_HTTP_PORT,
-    DEFAULT_PG_LISTEN_PORT as DEFAULT_PAGESERVER_PG_PORT,
-};
 use pageserver_api::controller_api::PlacementPolicy;
 use pageserver_api::models::{
     ShardParameters, TenantCreateRequest, TimelineCreateRequest, TimelineInfo,
 };
 use pageserver_api::shard::{ShardCount, ShardStripeSize, TenantShardId};
+use pageserver_api::{
+    DEFAULT_HTTP_LISTEN_PORT as DEFAULT_PAGESERVER_HTTP_PORT,
+    DEFAULT_PG_LISTEN_PORT as DEFAULT_PAGESERVER_PG_PORT,
+};
 use postgres_backend::AuthType;
 use postgres_connection::parse_host_port;
 use safekeeper_api::{
@@ -133,7 +133,7 @@ fn main() -> Result<()> {
         let subcommand_result = match sub_name {
             "tenant" => rt.block_on(handle_tenant(sub_args, &mut env)),
             "timeline" => rt.block_on(handle_timeline(sub_args, &mut env)),
-            "start" => rt.block_on(handle_start_all(&env)),
+            "start" => rt.block_on(handle_start_all(sub_args, &env)),
             "stop" => rt.block_on(handle_stop_all(sub_args, &env)),
             "pageserver" => rt.block_on(handle_pageserver(sub_args, &env)),
             "storage_controller" => rt.block_on(handle_storage_controller(sub_args, &env)),
@@ -358,13 +358,6 @@ fn handle_init(init_match: &ArgMatches) -> anyhow::Result<LocalEnv> {
         default_conf(*num_pageservers)
     };
 
-    let pageserver_config: toml_edit::Document =
-        if let Some(path) = init_match.get_one::<PathBuf>("pageserver-config") {
-            std::fs::read_to_string(path)?.parse()?
-        } else {
-            toml_edit::Document::new()
-        };
-
     let pg_version = init_match
         .get_one::<u32>("pg-version")
         .copied()
@@ -382,7 +375,7 @@ fn handle_init(init_match: &ArgMatches) -> anyhow::Result<LocalEnv> {
     // Initialize pageserver, create initial tenant and timeline.
     for ps_conf in &env.pageservers {
         PageServerNode::from_env(&env, ps_conf)
-            .initialize(&pageserver_config)
+            .initialize(&pageserver_config_overrides(init_match))
             .unwrap_or_else(|e| {
                 eprintln!("pageserver init failed: {e:?}");
                 exit(1);
@@ -404,6 +397,15 @@ fn get_default_pageserver(env: &local_env::LocalEnv) -> PageServerNode {
     PageServerNode::from_env(env, ps_conf)
 }
 
+fn pageserver_config_overrides(init_match: &ArgMatches) -> Vec<&str> {
+    init_match
+        .get_many::<String>("pageserver-config-override")
+        .into_iter()
+        .flatten()
+        .map(String::as_str)
+        .collect()
+}
+
 async fn handle_tenant(
     tenant_match: &ArgMatches,
     env: &mut local_env::LocalEnv,
@@ -1066,7 +1068,10 @@ fn get_pageserver(env: &local_env::LocalEnv, args: &ArgMatches) -> Result<PageSe
 async fn handle_pageserver(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<()> {
     match sub_match.subcommand() {
         Some(("start", subcommand_args)) => {
-            if let Err(e) = get_pageserver(env, subcommand_args)?.start().await {
+            if let Err(e) = get_pageserver(env, subcommand_args)?
+                .start(&pageserver_config_overrides(subcommand_args))
+                .await
+            {
                 eprintln!("pageserver start failed: {e}");
                 exit(1);
             }
@@ -1092,7 +1097,10 @@ async fn handle_pageserver(sub_match: &ArgMatches, env: &local_env::LocalEnv) ->
                 exit(1);
             }
 
-            if let Err(e) = pageserver.start().await {
+            if let Err(e) = pageserver
+                .start(&pageserver_config_overrides(subcommand_args))
+                .await
+            {
                 eprintln!("pageserver start failed: {e}");
                 exit(1);
             }
@@ -1219,7 +1227,7 @@ async fn handle_safekeeper(sub_match: &ArgMatches, env: &local_env::LocalEnv) ->
     Ok(())
 }
 
-async fn handle_start_all(env: &local_env::LocalEnv) -> anyhow::Result<()> {
+async fn handle_start_all(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> anyhow::Result<()> {
     // Endpoints are not started automatically
 
     broker::start_broker_process(env).await?;
@@ -1236,7 +1244,10 @@ async fn handle_start_all(env: &local_env::LocalEnv) -> anyhow::Result<()> {
 
     for ps_conf in &env.pageservers {
         let pageserver = PageServerNode::from_env(env, ps_conf);
-        if let Err(e) = pageserver.start().await {
+        if let Err(e) = pageserver
+            .start(&pageserver_config_overrides(sub_match))
+            .await
+        {
             eprintln!("pageserver {} start failed: {:#}", ps_conf.id, e);
             try_stop_all(env, true).await;
             exit(1);
@@ -1377,6 +1388,13 @@ fn cli() -> Command {
         .required(false)
         .value_name("stop-mode");
 
+    let pageserver_config_args = Arg::new("pageserver-config-override")
+        .long("pageserver-config-override")
+        .num_args(1)
+        .action(ArgAction::Append)
+        .help("Additional pageserver's configuration options or overrides, refer to pageserver's 'config-override' CLI parameter docs for more")
+        .required(false);
+
     let remote_ext_config_args = Arg::new("remote-ext-config")
         .long("remote-ext-config")
         .num_args(1)
@@ -1432,21 +1450,14 @@ fn cli() -> Command {
         .subcommand(
             Command::new("init")
                 .about("Initialize a new Neon repository, preparing configs for services to start with")
+                .arg(pageserver_config_args.clone())
                 .arg(num_pageservers_arg.clone())
                 .arg(
                     Arg::new("config")
                         .long("config")
                         .required(false)
                         .value_parser(value_parser!(PathBuf))
-                        .value_name("config")
-                )
-                .arg(
-                    Arg::new("pageserver-config")
-                        .long("pageserver-config")
-                        .required(false)
-                        .value_parser(value_parser!(PathBuf))
-                        .value_name("pageserver-config")
-                        .help("Merge the provided pageserver config into the one generated by neon_local."),
+                        .value_name("config"),
                 )
                 .arg(pg_version_arg.clone())
                 .arg(force_arg)
@@ -1528,6 +1539,7 @@ fn cli() -> Command {
                 .subcommand(Command::new("status"))
                 .subcommand(Command::new("start")
                     .about("Start local pageserver")
+                    .arg(pageserver_config_args.clone())
                 )
                 .subcommand(Command::new("stop")
                     .about("Stop local pageserver")
@@ -1535,6 +1547,7 @@ fn cli() -> Command {
                 )
                 .subcommand(Command::new("restart")
                     .about("Restart local pageserver")
+                    .arg(pageserver_config_args.clone())
                 )
         )
         .subcommand(
@@ -1647,6 +1660,7 @@ fn cli() -> Command {
         .subcommand(
             Command::new("start")
                 .about("Start page server and safekeepers")
+                .arg(pageserver_config_args)
         )
         .subcommand(
             Command::new("stop")

control_plane/src/endpoint.rs

@@ -554,7 +554,6 @@ impl Endpoint {
             format_version: 1.0,
             operation_uuid: None,
             features: self.features.clone(),
-            swap_size_bytes: None,
             cluster: Cluster {
                 cluster_id: None, // project ID: not used
                 name: None,       // project name: not used

control_plane/src/local_env.rs

@@ -382,10 +382,7 @@ impl LocalEnv {
 
         // Find neon binaries.
         if env.neon_distrib_dir == Path::new("") {
-            env::current_exe()?
-                .parent()
-                .unwrap()
-                .clone_into(&mut env.neon_distrib_dir);
+            env.neon_distrib_dir = env::current_exe()?.parent().unwrap().to_owned();
         }
 
         if env.pageservers.is_empty() {

control_plane/src/pageserver.rs

@@ -4,6 +4,7 @@
 //!
 //!   .neon/
 //!
+use std::borrow::Cow;
 use std::collections::HashMap;
 
 use std::io;
@@ -76,7 +77,7 @@ impl PageServerNode {
     /// Merge overrides provided by the user on the command line with our default overides derived from neon_local configuration.
     ///
     /// These all end up on the command line of the `pageserver` binary.
-    fn neon_local_overrides(&self, cli_overrides: &toml_edit::Document) -> Vec<String> {
+    fn neon_local_overrides(&self, cli_overrides: &[&str]) -> Vec<String> {
         // FIXME: the paths should be shell-escaped to handle paths with spaces, quotas etc.
         let pg_distrib_dir_param = format!(
             "pg_distrib_dir='{}'",
@@ -156,7 +157,10 @@ impl PageServerNode {
             }
         }
 
-        if !cli_overrides.contains_key("remote_storage") {
+        if !cli_overrides
+            .iter()
+            .any(|c| c.starts_with("remote_storage"))
+        {
             overrides.push(format!(
                 "remote_storage={{local_path='../{PAGESERVER_REMOTE_STORAGE_DIR}'}}"
             ));
@@ -169,13 +173,13 @@ impl PageServerNode {
         }
 
         // Apply the user-provided overrides
-        overrides.push(cli_overrides.to_string());
+        overrides.extend(cli_overrides.iter().map(|&c| c.to_owned()));
 
         overrides
     }
 
     /// Initializes a pageserver node by creating its config with the overrides provided.
-    pub fn initialize(&self, config_overrides: &toml_edit::Document) -> anyhow::Result<()> {
+    pub fn initialize(&self, config_overrides: &[&str]) -> anyhow::Result<()> {
         // First, run `pageserver --init` and wait for it to write a config into FS and exit.
         self.pageserver_init(config_overrides)
             .with_context(|| format!("Failed to run init for pageserver node {}", self.conf.id))
@@ -193,11 +197,11 @@ impl PageServerNode {
             .expect("non-Unicode path")
     }
 
-    pub async fn start(&self) -> anyhow::Result<()> {
-        self.start_node().await
+    pub async fn start(&self, config_overrides: &[&str]) -> anyhow::Result<()> {
+        self.start_node(config_overrides, false).await
     }
 
-    fn pageserver_init(&self, config_overrides: &toml_edit::Document) -> anyhow::Result<()> {
+    fn pageserver_init(&self, config_overrides: &[&str]) -> anyhow::Result<()> {
         let datadir = self.repo_path();
         let node_id = self.conf.id;
         println!(
@@ -215,18 +219,11 @@ impl PageServerNode {
         let datadir_path_str = datadir.to_str().with_context(|| {
             format!("Cannot start pageserver node {node_id} in path that has no string representation: {datadir:?}")
         })?;
+        let mut args = self.pageserver_basic_args(config_overrides, datadir_path_str);
+        args.push(Cow::Borrowed("--init"));
 
-        // `pageserver --init` merges the `--config-override`s into a built-in default config,
-        // then writes out the merged product to `pageserver.toml`.
-        // TODO: just write the full `pageserver.toml` and get rid of `--config-override`.
-        let mut args = vec!["--init", "--workdir", datadir_path_str];
-        let overrides = self.neon_local_overrides(config_overrides);
-        for piece in &overrides {
-            args.push("--config-override");
-            args.push(piece);
-        }
         let init_output = Command::new(self.env.pageserver_bin())
-            .args(args)
+            .args(args.iter().map(Cow::as_ref))
             .envs(self.pageserver_env_variables()?)
             .output()
             .with_context(|| format!("Failed to run pageserver init for node {node_id}"))?;
@@ -251,13 +248,12 @@ impl PageServerNode {
         // situation: the metadata is written by some other script.
         std::fs::write(
             metadata_path,
-            serde_json::to_vec(&pageserver_api::config::NodeMetadata {
-                postgres_host: "localhost".to_string(),
-                postgres_port: self.pg_connection_config.port(),
-                http_host: "localhost".to_string(),
-                http_port,
-                other: HashMap::new(),
-            })
+            serde_json::to_vec(&serde_json::json!({
+                "host": "localhost",
+                "port": self.pg_connection_config.port(),
+                "http_host": "localhost",
+                "http_port": http_port,
+            }))
             .unwrap(),
         )
         .expect("Failed to write metadata file");
@@ -265,7 +261,11 @@ impl PageServerNode {
         Ok(())
     }
 
-    async fn start_node(&self) -> anyhow::Result<()> {
+    async fn start_node(
+        &self,
+        config_overrides: &[&str],
+        update_config: bool,
+    ) -> anyhow::Result<()> {
         // TODO: using a thread here because start_process() is not async but we need to call check_status()
         let datadir = self.repo_path();
         print!(
@@ -282,12 +282,15 @@ impl PageServerNode {
                 self.conf.id, datadir,
             )
         })?;
-        let args = vec!["-D", datadir_path_str];
+        let mut args = self.pageserver_basic_args(config_overrides, datadir_path_str);
+        if update_config {
+            args.push(Cow::Borrowed("--update-config"));
+        }
         background_process::start_process(
             "pageserver",
             &datadir,
             &self.env.pageserver_bin(),
-            args,
+            args.iter().map(Cow::as_ref),
             self.pageserver_env_variables()?,
             background_process::InitialPidFile::Expect(self.pid_file()),
             || async {
@@ -304,6 +307,22 @@ impl PageServerNode {
         Ok(())
     }
 
+    fn pageserver_basic_args<'a>(
+        &self,
+        config_overrides: &'a [&'a str],
+        datadir_path_str: &'a str,
+    ) -> Vec<Cow<'a, str>> {
+        let mut args = vec![Cow::Borrowed("-D"), Cow::Borrowed(datadir_path_str)];
+
+        let overrides = self.neon_local_overrides(config_overrides);
+        for config_override in overrides {
+            args.push(Cow::Borrowed("-c"));
+            args.push(Cow::Owned(config_override));
+        }
+
+        args
+    }
+
     fn pageserver_env_variables(&self) -> anyhow::Result<Vec<(String, String)>> {
         // FIXME: why is this tied to pageserver's auth type? Whether or not the safekeeper
         // needs a token, and how to generate that token, seems independent to whether

libs/compute_api/src/spec.rs

@@ -33,23 +33,6 @@ pub struct ComputeSpec {
     #[serde(default)]
     pub features: Vec<ComputeFeature>,
 
-    /// If compute_ctl was passed `--resize-swap-on-bind`, a value of `Some(_)` instructs
-    /// compute_ctl to `/neonvm/bin/resize-swap` with the given size, when the spec is first
-    /// received.
-    ///
-    /// Both this field and `--resize-swap-on-bind` are required, so that the control plane's
-    /// spec generation doesn't need to be aware of the actual compute it's running on, while
-    /// guaranteeing gradual rollout of swap. Otherwise, without `--resize-swap-on-bind`, we could
-    /// end up trying to resize swap in VMs without it -- or end up *not* resizing swap, thus
-    /// giving every VM much more swap than it should have (32GiB).
-    ///
-    /// Eventually we may remove `--resize-swap-on-bind` and exclusively use `swap_size_bytes` for
-    /// enabling the swap resizing behavior once rollout is complete.
-    ///
-    /// See neondatabase/cloud#12047 for more.
-    #[serde(default)]
-    pub swap_size_bytes: Option<u64>,
-
     /// Expected cluster state at the end of transition process.
     pub cluster: Cluster,
     pub delta_operations: Option<Vec<DeltaOp>>,

libs/pageserver_api/src/config.rs

@@ -1,31 +0,0 @@
-use std::collections::HashMap;
-
-use const_format::formatcp;
-
-#[cfg(test)]
-mod tests;
-
-pub const DEFAULT_PG_LISTEN_PORT: u16 = 64000;
-pub const DEFAULT_PG_LISTEN_ADDR: &str = formatcp!("127.0.0.1:{DEFAULT_PG_LISTEN_PORT}");
-pub const DEFAULT_HTTP_LISTEN_PORT: u16 = 9898;
-pub const DEFAULT_HTTP_LISTEN_ADDR: &str = formatcp!("127.0.0.1:{DEFAULT_HTTP_LISTEN_PORT}");
-
-// Certain metadata (e.g. externally-addressable name, AZ) is delivered
-// as a separate structure.  This information is not neeed by the pageserver
-// itself, it is only used for registering the pageserver with the control
-// plane and/or storage controller.
-//
-#[derive(PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
-pub struct NodeMetadata {
-    #[serde(rename = "host")]
-    pub postgres_host: String,
-    #[serde(rename = "port")]
-    pub postgres_port: u16,
-    pub http_host: String,
-    pub http_port: u16,
-
-    // Deployment tools may write fields to the metadata file beyond what we
-    // use in this type: this type intentionally only names fields that require.
-    #[serde(flatten)]
-    pub other: HashMap<String, serde_json::Value>,
-}

libs/pageserver_api/src/config/tests.rs

@@ -1,22 +0,0 @@
-use super::*;
-
-#[test]
-fn test_node_metadata_v1_backward_compatibilty() {
-    let v1 = serde_json::to_vec(&serde_json::json!({
-        "host": "localhost",
-        "port": 23,
-        "http_host": "localhost",
-        "http_port": 42,
-    }));
-
-    assert_eq!(
-        serde_json::from_slice::<NodeMetadata>(&v1.unwrap()).unwrap(),
-        NodeMetadata {
-            postgres_host: "localhost".to_string(),
-            postgres_port: 23,
-            http_host: "localhost".to_string(),
-            http_port: 42,
-            other: HashMap::new(),
-        }
-    )
-}

libs/pageserver_api/src/key.rs

@@ -80,7 +80,7 @@ impl Key {
     }
 
     /// Get the range of metadata keys.
-    pub const fn metadata_key_range() -> Range<Self> {
+    pub fn metadata_key_range() -> Range<Self> {
         Key {
             field1: METADATA_KEY_BEGIN_PREFIX,
             field2: 0,
@@ -572,17 +572,14 @@ pub const AUX_FILES_KEY: Key = Key {
 // Reverse mappings for a few Keys.
 // These are needed by WAL redo manager.
 
-/// Non inherited range for vectored get.
 pub const NON_INHERITED_RANGE: Range<Key> = AUX_FILES_KEY..AUX_FILES_KEY.next();
-/// Sparse keyspace range for vectored get. Missing key error will be ignored for this range.
-pub const NON_INHERITED_SPARSE_RANGE: Range<Key> = Key::metadata_key_range();
 
 // AUX_FILES currently stores only data for logical replication (slots etc), and
 // we don't preserve these on a branch because safekeepers can't follow timeline
 // switch (and generally it likely should be optional), so ignore these.
 #[inline(always)]
 pub fn is_inherited_key(key: Key) -> bool {
-    !NON_INHERITED_RANGE.contains(&key) && !NON_INHERITED_SPARSE_RANGE.contains(&key)
+    !NON_INHERITED_RANGE.contains(&key)
 }
 
 #[inline(always)]

libs/pageserver_api/src/lib.rs

@@ -1,5 +1,6 @@
 #![deny(unsafe_code)]
 #![deny(clippy::undocumented_unsafe_blocks)]
+use const_format::formatcp;
 
 pub mod controller_api;
 pub mod key;
@@ -10,4 +11,7 @@ pub mod shard;
 /// Public API types
 pub mod upcall_api;
 
-pub mod config;
+pub const DEFAULT_PG_LISTEN_PORT: u16 = 64000;
+pub const DEFAULT_PG_LISTEN_ADDR: &str = formatcp!("127.0.0.1:{DEFAULT_PG_LISTEN_PORT}");
+pub const DEFAULT_HTTP_LISTEN_PORT: u16 = 9898;
+pub const DEFAULT_HTTP_LISTEN_ADDR: &str = formatcp!("127.0.0.1:{DEFAULT_HTTP_LISTEN_PORT}");

libs/pageserver_api/src/shard.rs

@@ -97,7 +97,7 @@ impl ShardCount {
 
     /// The internal value of a ShardCount may be zero, which means "1 shard, but use
     /// legacy format for TenantShardId that excludes the shard suffix", also known
-    /// as [`TenantShardId::unsharded`].
+    /// as `TenantShardId::unsharded`.
     ///
     /// This method returns the actual number of shards, i.e. if our internal value is
     /// zero, we return 1 (unsharded tenants have 1 shard).
@@ -116,9 +116,7 @@ impl ShardCount {
         self.0
     }
 
-    /// Whether the `ShardCount` is for an unsharded tenant, so uses one shard but
-    /// uses the legacy format for `TenantShardId`. See also the documentation for
-    /// [`Self::count`].
+    ///
     pub fn is_unsharded(&self) -> bool {
         self.0 == 0
     }

libs/remote_storage/src/lib.rs

@@ -55,11 +55,11 @@ pub use error::{DownloadError, TimeTravelError, TimeoutOrCancel};
 /// ~3500 PUT/COPY/POST/DELETE or 5500 GET/HEAD S3 requests
 /// <https://aws.amazon.com/premiumsupport/knowledge-center/s3-request-limit-avoid-throttling/>
 pub const DEFAULT_REMOTE_STORAGE_S3_CONCURRENCY_LIMIT: usize = 100;
-/// Set this limit analogously to the S3 limit
+/// We set this a little bit low as we currently buffer the entire file into RAM
 ///
 /// Here, a limit of max 20k concurrent connections was noted.
 /// <https://learn.microsoft.com/en-us/answers/questions/1301863/is-there-any-limitation-to-concurrent-connections>
-pub const DEFAULT_REMOTE_STORAGE_AZURE_CONCURRENCY_LIMIT: usize = 100;
+pub const DEFAULT_REMOTE_STORAGE_AZURE_CONCURRENCY_LIMIT: usize = 30;
 /// No limits on the client side, which currenltly means 1000 for AWS S3.
 /// <https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListObjectsV2.html#API_ListObjectsV2_RequestSyntax>
 pub const DEFAULT_MAX_KEYS_PER_LIST_RESPONSE: Option<i32> = None;

pageserver/client/src/mgmt_api.rs

@@ -284,34 +284,6 @@ impl Client {
         Ok((status, progress))
     }
 
-    pub async fn tenant_secondary_status(
-        &self,
-        tenant_shard_id: TenantShardId,
-    ) -> Result<SecondaryProgress> {
-        let path = reqwest::Url::parse(&format!(
-            "{}/v1/tenant/{}/secondary/status",
-            self.mgmt_api_endpoint, tenant_shard_id
-        ))
-        .expect("Cannot build URL");
-
-        self.request(Method::GET, path, ())
-            .await?
-            .json()
-            .await
-            .map_err(Error::ReceiveBody)
-    }
-
-    pub async fn tenant_heatmap_upload(&self, tenant_id: TenantShardId) -> Result<()> {
-        let path = reqwest::Url::parse(&format!(
-            "{}/v1/tenant/{}/heatmap_upload",
-            self.mgmt_api_endpoint, tenant_id
-        ))
-        .expect("Cannot build URL");
-
-        self.request(Method::POST, path, ()).await?;
-        Ok(())
-    }
-
     pub async fn location_config(
         &self,
         tenant_shard_id: TenantShardId,

pageserver/src/bin/pageserver.rs

@@ -3,7 +3,6 @@
 //! Main entry point for the Page Server executable.
 
 use std::env::{var, VarError};
-use std::io::Read;
 use std::sync::Arc;
 use std::time::Duration;
 use std::{env, ops::ControlFlow, str::FromStr};
@@ -152,34 +151,37 @@ fn initialize_config(
     workdir: &Utf8Path,
 ) -> anyhow::Result<ControlFlow<(), &'static PageServerConf>> {
     let init = arg_matches.get_flag("init");
+    let update_config = init || arg_matches.get_flag("update-config");
 
-    let file_contents: Option<toml_edit::Document> = match std::fs::File::open(cfg_file_path) {
-        Ok(mut f) => {
-            if init {
-                anyhow::bail!("config file already exists: {cfg_file_path}");
-            }
-            let md = f.metadata().context("stat config file")?;
-            if md.is_file() {
-                let mut s = String::new();
-                f.read_to_string(&mut s).context("read config file")?;
-                Some(s.parse().context("parse config file toml")?)
-            } else {
-                anyhow::bail!("directory entry exists but is not a file: {cfg_file_path}");
-            }
-        }
-        Err(e) if e.kind() == std::io::ErrorKind::NotFound => None,
-        Err(e) => {
-            anyhow::bail!("open pageserver config: {e}: {cfg_file_path}");
+    let (mut toml, config_file_exists) = if cfg_file_path.is_file() {
+        if init {
+            anyhow::bail!(
+                "Config file '{cfg_file_path}' already exists, cannot init it, use --update-config to update it",
+            );
         }
+        // Supplement the CLI arguments with the config file
+        let cfg_file_contents = std::fs::read_to_string(cfg_file_path)
+            .with_context(|| format!("Failed to read pageserver config at '{cfg_file_path}'"))?;
+        (
+            cfg_file_contents
+                .parse::<toml_edit::Document>()
+                .with_context(|| {
+                    format!("Failed to parse '{cfg_file_path}' as pageserver config")
+                })?,
+            true,
+        )
+    } else if cfg_file_path.exists() {
+        anyhow::bail!("Config file '{cfg_file_path}' exists but is not a regular file");
+    } else {
+        // We're initializing the tenant, so there's no config file yet
+        (
+            DEFAULT_CONFIG_FILE
+                .parse::<toml_edit::Document>()
+                .context("could not parse built-in config file")?,
+            false,
+        )
     };
 
-    let mut effective_config = file_contents.unwrap_or_else(|| {
-        DEFAULT_CONFIG_FILE
-            .parse()
-            .expect("unit tests ensure this works")
-    });
-
-    // Patch with overrides from the command line
     if let Some(values) = arg_matches.get_many::<String>("config-override") {
         for option_line in values {
             let doc = toml_edit::Document::from_str(option_line).with_context(|| {
@@ -187,21 +189,22 @@ fn initialize_config(
             })?;
 
             for (key, item) in doc.iter() {
-                effective_config.insert(key, item.clone());
+                if config_file_exists && update_config && key == "id" && toml.contains_key(key) {
+                    anyhow::bail!("Pageserver config file exists at '{cfg_file_path}' and has node id already, it cannot be overridden");
+                }
+                toml.insert(key, item.clone());
             }
         }
     }
 
-    debug!("Resulting toml: {effective_config}");
-
-    // Construct the runtime representation
-    let conf = PageServerConf::parse_and_validate(&effective_config, workdir)
+    debug!("Resulting toml: {toml}");
+    let conf = PageServerConf::parse_and_validate(&toml, workdir)
         .context("Failed to parse pageserver configuration")?;
 
-    if init {
+    if update_config {
         info!("Writing pageserver config to '{cfg_file_path}'");
 
-        std::fs::write(cfg_file_path, effective_config.to_string())
+        std::fs::write(cfg_file_path, toml.to_string())
             .with_context(|| format!("Failed to write pageserver config to '{cfg_file_path}'"))?;
         info!("Config successfully written to '{cfg_file_path}'")
     }
@@ -755,13 +758,18 @@ fn cli() -> Command {
         // See `settings.md` for more details on the extra configuration patameters pageserver can process
         .arg(
             Arg::new("config-override")
-                .long("config-override")
                 .short('c')
                 .num_args(1)
                 .action(ArgAction::Append)
                 .help("Additional configuration overrides of the ones from the toml config file (or new ones to add there). \
                 Any option has to be a valid toml document, example: `-c=\"foo='hey'\"` `-c=\"foo={value=1}\"`"),
         )
+        .arg(
+            Arg::new("update-config")
+                .long("update-config")
+                .action(ArgAction::SetTrue)
+                .help("Update the config file when started"),
+        )
         .arg(
             Arg::new("enabled-features")
                 .long("enabled-features")

pageserver/src/config.rs

@@ -9,7 +9,7 @@ use pageserver_api::shard::TenantShardId;
 use remote_storage::{RemotePath, RemoteStorageConfig};
 use serde;
 use serde::de::IntoDeserializer;
-use std::env;
+use std::{collections::HashMap, env};
 use storage_broker::Uri;
 use utils::crashsafe::path_with_suffix_extension;
 use utils::id::ConnectionId;
@@ -51,7 +51,7 @@ pub mod defaults {
     use crate::tenant::config::defaults::*;
     use const_format::formatcp;
 
-    pub use pageserver_api::config::{
+    pub use pageserver_api::{
         DEFAULT_HTTP_LISTEN_ADDR, DEFAULT_HTTP_LISTEN_PORT, DEFAULT_PG_LISTEN_ADDR,
         DEFAULT_PG_LISTEN_PORT,
     };
@@ -335,6 +335,26 @@ impl<T: Clone> BuilderValue<T> {
     }
 }
 
+// Certain metadata (e.g. externally-addressable name, AZ) is delivered
+// as a separate structure.  This information is not neeed by the pageserver
+// itself, it is only used for registering the pageserver with the control
+// plane and/or storage controller.
+//
+#[derive(serde::Deserialize)]
+pub(crate) struct NodeMetadata {
+    #[serde(rename = "host")]
+    pub(crate) postgres_host: String,
+    #[serde(rename = "port")]
+    pub(crate) postgres_port: u16,
+    pub(crate) http_host: String,
+    pub(crate) http_port: u16,
+
+    // Deployment tools may write fields to the metadata file beyond what we
+    // use in this type: this type intentionally only names fields that require.
+    #[serde(flatten)]
+    pub(crate) other: HashMap<String, serde_json::Value>,
+}
+
 // needed to simplify config construction
 #[derive(Default)]
 struct PageServerConfigBuilder {

pageserver/src/control_plane_client.rs

@@ -14,8 +14,10 @@ use tokio_util::sync::CancellationToken;
 use url::Url;
 use utils::{backoff, failpoint_support, generation::Generation, id::NodeId};
 
-use crate::{config::PageServerConf, virtual_file::on_fatal_io_error};
-use pageserver_api::config::NodeMetadata;
+use crate::{
+    config::{NodeMetadata, PageServerConf},
+    virtual_file::on_fatal_io_error,
+};
 
 /// The Pageserver's client for using the control plane API: this is a small subset
 /// of the overall control plane API, for dealing with generations (see docs/rfcs/025-generation-numbers.md)

pageserver/src/http/routes.rs

@@ -2160,27 +2160,6 @@ async fn secondary_download_handler(
     json_response(status, progress)
 }
 
-async fn secondary_status_handler(
-    request: Request<Body>,
-    _cancel: CancellationToken,
-) -> Result<Response<Body>, ApiError> {
-    let state = get_state(&request);
-    let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
-
-    let Some(secondary_tenant) = state
-        .tenant_manager
-        .get_secondary_tenant_shard(tenant_shard_id)
-    else {
-        return Err(ApiError::NotFound(
-            anyhow::anyhow!("Shard {} not found", tenant_shard_id).into(),
-        ));
-    };
-
-    let progress = secondary_tenant.progress.lock().unwrap().clone();
-
-    json_response(StatusCode::OK, progress)
-}
-
 async fn handler_404(_: Request<Body>) -> Result<Response<Body>, ApiError> {
     json_response(
         StatusCode::NOT_FOUND,
@@ -2542,9 +2521,6 @@ pub fn make_router(
         .put("/v1/deletion_queue/flush", |r| {
             api_handler(r, deletion_queue_flush)
         })
-        .get("/v1/tenant/:tenant_shard_id/secondary/status", |r| {
-            api_handler(r, secondary_status_handler)
-        })
         .post("/v1/tenant/:tenant_shard_id/secondary/download", |r| {
             api_handler(r, secondary_download_handler)
         })

pageserver/src/metrics.rs

@@ -51,8 +51,8 @@ pub(crate) enum StorageTimeOperation {
     #[strum(serialize = "gc")]
     Gc,
 
-    #[strum(serialize = "find gc cutoffs")]
-    FindGcCutoffs,
+    #[strum(serialize = "update gc info")]
+    UpdateGcInfo,
 
     #[strum(serialize = "create tenant")]
     CreateTenant,
@@ -194,11 +194,6 @@ pub(crate) struct GetVectoredLatency {
     map: EnumMap<TaskKind, Option<Histogram>>,
 }
 
-#[allow(dead_code)]
-pub(crate) struct ScanLatency {
-    map: EnumMap<TaskKind, Option<Histogram>>,
-}
-
 impl GetVectoredLatency {
     // Only these task types perform vectored gets. Filter all other tasks out to reduce total
     // cardinality of the metric.
@@ -209,48 +204,6 @@ impl GetVectoredLatency {
     }
 }
 
-impl ScanLatency {
-    // Only these task types perform vectored gets. Filter all other tasks out to reduce total
-    // cardinality of the metric.
-    const TRACKED_TASK_KINDS: [TaskKind; 1] = [TaskKind::PageRequestHandler];
-
-    pub(crate) fn for_task_kind(&self, task_kind: TaskKind) -> Option<&Histogram> {
-        self.map[task_kind].as_ref()
-    }
-}
-
-pub(crate) struct ScanLatencyOngoingRecording<'a> {
-    parent: &'a Histogram,
-    start: std::time::Instant,
-}
-
-impl<'a> ScanLatencyOngoingRecording<'a> {
-    pub(crate) fn start_recording(parent: &'a Histogram) -> ScanLatencyOngoingRecording<'a> {
-        let start = Instant::now();
-        ScanLatencyOngoingRecording { parent, start }
-    }
-
-    pub(crate) fn observe(self, throttled: Option<Duration>) {
-        let elapsed = self.start.elapsed();
-        let ex_throttled = if let Some(throttled) = throttled {
-            elapsed.checked_sub(throttled)
-        } else {
-            Some(elapsed)
-        };
-        if let Some(ex_throttled) = ex_throttled {
-            self.parent.observe(ex_throttled.as_secs_f64());
-        } else {
-            use utils::rate_limit::RateLimit;
-            static LOGGED: Lazy<Mutex<RateLimit>> =
-                Lazy::new(|| Mutex::new(RateLimit::new(Duration::from_secs(10))));
-            let mut rate_limit = LOGGED.lock().unwrap();
-            rate_limit.call(|| {
-                warn!("error deducting time spent throttled; this message is logged at a global rate limit");
-            });
-        }
-    }
-}
-
 pub(crate) static GET_VECTORED_LATENCY: Lazy<GetVectoredLatency> = Lazy::new(|| {
     let inner = register_histogram_vec!(
         "pageserver_get_vectored_seconds",
@@ -274,29 +227,6 @@ pub(crate) static GET_VECTORED_LATENCY: Lazy<GetVectoredLatency> = Lazy::new(||
     }
 });
 
-pub(crate) static SCAN_LATENCY: Lazy<ScanLatency> = Lazy::new(|| {
-    let inner = register_histogram_vec!(
-        "pageserver_scan_seconds",
-        "Time spent in scan, excluding time spent in timeline_get_throttle.",
-        &["task_kind"],
-        CRITICAL_OP_BUCKETS.into(),
-    )
-    .expect("failed to define a metric");
-
-    ScanLatency {
-        map: EnumMap::from_array(std::array::from_fn(|task_kind_idx| {
-            let task_kind = <TaskKind as enum_map::Enum>::from_usize(task_kind_idx);
-
-            if ScanLatency::TRACKED_TASK_KINDS.contains(&task_kind) {
-                let task_kind = task_kind.into();
-                Some(inner.with_label_values(&[task_kind]))
-            } else {
-                None
-            }
-        })),
-    }
-});
-
 pub(crate) struct PageCacheMetricsForTaskKind {
     pub read_accesses_materialized_page: IntCounter,
     pub read_accesses_immutable: IntCounter,
@@ -2059,7 +1989,7 @@ pub(crate) struct TimelineMetrics {
     pub imitate_logical_size_histo: StorageTimeMetrics,
     pub load_layer_map_histo: StorageTimeMetrics,
     pub garbage_collect_histo: StorageTimeMetrics,
-    pub find_gc_cutoffs_histo: StorageTimeMetrics,
+    pub update_gc_info_histo: StorageTimeMetrics,
     pub last_record_gauge: IntGauge,
     resident_physical_size_gauge: UIntGauge,
     /// copy of LayeredTimeline.current_logical_size
@@ -2120,8 +2050,8 @@ impl TimelineMetrics {
             &shard_id,
             &timeline_id,
         );
-        let find_gc_cutoffs_histo = StorageTimeMetrics::new(
-            StorageTimeOperation::FindGcCutoffs,
+        let update_gc_info_histo = StorageTimeMetrics::new(
+            StorageTimeOperation::UpdateGcInfo,
             &tenant_id,
             &shard_id,
             &timeline_id,
@@ -2168,7 +2098,7 @@ impl TimelineMetrics {
             logical_size_histo,
             imitate_logical_size_histo,
             garbage_collect_histo,
-            find_gc_cutoffs_histo,
+            update_gc_info_histo,
             load_layer_map_histo,
             last_record_gauge,
             resident_physical_size_gauge,

pageserver/src/pgdatadir_mapping.rs

@@ -279,7 +279,7 @@ impl Timeline {
 
         match RelDirectory::des(&buf).context("deserialization failure") {
             Ok(dir) => {
-                let exists = dir.rels.contains(&(tag.relnode, tag.forknum));
+                let exists = dir.rels.get(&(tag.relnode, tag.forknum)).is_some();
                 Ok(exists)
             }
             Err(e) => Err(PageReconstructError::from(e)),
@@ -379,7 +379,7 @@ impl Timeline {
 
         match SlruSegmentDirectory::des(&buf).context("deserialization failure") {
             Ok(dir) => {
-                let exists = dir.segments.contains(&segno);
+                let exists = dir.segments.get(&segno).is_some();
                 Ok(exists)
             }
             Err(e) => Err(PageReconstructError::from(e)),
@@ -1143,22 +1143,21 @@ impl<'a> DatadirModification<'a> {
         let mut dbdir = DbDirectory::des(&self.get(DBDIR_KEY, ctx).await.context("read db")?)
             .context("deserialize db")?;
         let rel_dir_key = rel_dir_to_key(rel.spcnode, rel.dbnode);
-        let mut rel_dir =
-            if let hash_map::Entry::Vacant(e) = dbdir.dbdirs.entry((rel.spcnode, rel.dbnode)) {
-                // Didn't exist. Update dbdir
-                e.insert(false);
-                let buf = DbDirectory::ser(&dbdir).context("serialize db")?;
-                self.pending_directory_entries
-                    .push((DirectoryKind::Db, dbdir.dbdirs.len()));
-                self.put(DBDIR_KEY, Value::Image(buf.into()));
+        let mut rel_dir = if dbdir.dbdirs.get(&(rel.spcnode, rel.dbnode)).is_none() {
+            // Didn't exist. Update dbdir
+            dbdir.dbdirs.insert((rel.spcnode, rel.dbnode), false);
+            let buf = DbDirectory::ser(&dbdir).context("serialize db")?;
+            self.pending_directory_entries
+                .push((DirectoryKind::Db, dbdir.dbdirs.len()));
+            self.put(DBDIR_KEY, Value::Image(buf.into()));
 
-                // and create the RelDirectory
-                RelDirectory::default()
-            } else {
-                // reldir already exists, fetch it
-                RelDirectory::des(&self.get(rel_dir_key, ctx).await.context("read db")?)
-                    .context("deserialize db")?
-            };
+            // and create the RelDirectory
+            RelDirectory::default()
+        } else {
+            // reldir already exists, fetch it
+            RelDirectory::des(&self.get(rel_dir_key, ctx).await.context("read db")?)
+                .context("deserialize db")?
+        };
 
         // Add the new relation to the rel directory entry, and write it back
         if !rel_dir.rels.insert((rel.relnode, rel.forknum)) {

pageserver/src/tenant.rs

@@ -64,7 +64,6 @@ use self::timeline::uninit::UninitializedTimeline;
 use self::timeline::EvictionTaskTenantState;
 use self::timeline::TimelineResources;
 use self::timeline::WaitLsnError;
-use self::timeline::{GcCutoffs, GcInfo};
 use crate::config::PageServerConf;
 use crate::context::{DownloadBehavior, RequestContext};
 use crate::deletion_queue::DeletionQueueClient;
@@ -87,6 +86,7 @@ use crate::tenant::remote_timeline_client::INITDB_PATH;
 use crate::tenant::storage_layer::DeltaLayer;
 use crate::tenant::storage_layer::ImageLayer;
 use crate::InitializationOrder;
+use std::cmp::min;
 use std::collections::hash_map::Entry;
 use std::collections::BTreeSet;
 use std::collections::HashMap;
@@ -2812,48 +2812,7 @@ impl Tenant {
         cancel: &CancellationToken,
         ctx: &RequestContext,
     ) -> anyhow::Result<Vec<Arc<Timeline>>> {
-        // before taking the gc_cs lock, do the heavier weight finding of gc_cutoff points for
-        // currently visible timelines.
-        let timelines = self
-            .timelines
-            .lock()
-            .unwrap()
-            .values()
-            .filter(|tl| match target_timeline_id.as_ref() {
-                Some(target) => &tl.timeline_id == target,
-                None => true,
-            })
-            .cloned()
-            .collect::<Vec<_>>();
-
-        let mut gc_cutoffs: HashMap<TimelineId, GcCutoffs> =
-            HashMap::with_capacity(timelines.len());
-
-        for timeline in timelines.iter() {
-            let cutoff = timeline
-                .get_last_record_lsn()
-                .checked_sub(horizon)
-                .unwrap_or(Lsn(0));
-
-            let res = timeline.find_gc_cutoffs(cutoff, pitr, cancel, ctx).await;
-
-            match res {
-                Ok(cutoffs) => {
-                    let old = gc_cutoffs.insert(timeline.timeline_id, cutoffs);
-                    assert!(old.is_none());
-                }
-                Err(e) => {
-                    tracing::warn!(timeline_id = %timeline.timeline_id, "ignoring failure to find gc cutoffs: {e:#}");
-                }
-            }
-        }
-
-        if !self.is_active() {
-            anyhow::bail!("shutting down");
-        }
-
-        // grab mutex to prevent new timelines from being created here; avoid doing long operations
-        // because that will stall branch creation.
+        // grab mutex to prevent new timelines from being created here.
         let gc_cs = self.gc_cs.lock().await;
 
         // Scan all timelines. For each timeline, remember the timeline ID and
@@ -2915,6 +2874,11 @@ impl Tenant {
                 }
             }
 
+            let cutoff = timeline
+                .get_last_record_lsn()
+                .checked_sub(horizon)
+                .unwrap_or(Lsn(0));
+
             let branchpoints: Vec<Lsn> = all_branchpoints
                 .range((
                     Included((timeline_id, Lsn(0))),
@@ -2922,27 +2886,9 @@ impl Tenant {
                 ))
                 .map(|&x| x.1)
                 .collect();
-
-            {
-                let mut target = timeline.gc_info.write().unwrap();
-
-                match gc_cutoffs.remove(&timeline_id) {
-                    Some(cutoffs) => {
-                        *target = GcInfo {
-                            retain_lsns: branchpoints,
-                            cutoffs,
-                        };
-                    }
-                    None => {
-                        // reasons for this being unavailable:
-                        // - this timeline was created while we were finding cutoffs
-                        // - lsn for timestamp search fails for this timeline repeatedly
-                        //
-                        // in both cases, refreshing the branchpoints is correct.
-                        target.retain_lsns = branchpoints;
-                    }
-                };
-            }
+            timeline
+                .update_gc_info(branchpoints, cutoff, pitr, cancel, ctx)
+                .await?;
 
             gc_timelines.push(timeline);
         }
@@ -3031,7 +2977,7 @@ impl Tenant {
         // and then the planned GC cutoff
         {
             let gc_info = src_timeline.gc_info.read().unwrap();
-            let cutoff = gc_info.min_cutoff();
+            let cutoff = min(gc_info.pitr_cutoff, gc_info.horizon_cutoff);
             if start_lsn < cutoff {
                 return Err(CreateTimelineError::AncestorLsn(anyhow::anyhow!(
                     "invalid branch start lsn: less than planned GC cutoff {cutoff}"
@@ -3925,7 +3871,7 @@ mod tests {
     use crate::DEFAULT_PG_VERSION;
     use bytes::BytesMut;
     use hex_literal::hex;
-    use pageserver_api::key::{AUX_KEY_PREFIX, NON_INHERITED_RANGE};
+    use pageserver_api::key::NON_INHERITED_RANGE;
     use pageserver_api::keyspace::KeySpace;
     use pageserver_api::models::CompactionAlgorithm;
     use rand::{thread_rng, Rng};
@@ -4567,20 +4513,18 @@ mod tests {
     }
 
     async fn bulk_insert_compact_gc(
-        tenant: &Tenant,
-        timeline: &Arc<Timeline>,
+        timeline: Arc<Timeline>,
         ctx: &RequestContext,
         lsn: Lsn,
         repeat: usize,
         key_count: usize,
     ) -> anyhow::Result<()> {
         let compact = true;
-        bulk_insert_maybe_compact_gc(tenant, timeline, ctx, lsn, repeat, key_count, compact).await
+        bulk_insert_maybe_compact_gc(timeline, ctx, lsn, repeat, key_count, compact).await
     }
 
     async fn bulk_insert_maybe_compact_gc(
-        tenant: &Tenant,
-        timeline: &Arc<Timeline>,
+        timeline: Arc<Timeline>,
         ctx: &RequestContext,
         mut lsn: Lsn,
         repeat: usize,
@@ -4593,8 +4537,6 @@ mod tests {
         // Enforce that key range is monotonously increasing
         let mut keyspace = KeySpaceAccum::new();
 
-        let cancel = CancellationToken::new();
-
         for _ in 0..repeat {
             for _ in 0..key_count {
                 test_key.field6 = blknum;
@@ -4616,19 +4558,24 @@ mod tests {
                 blknum += 1;
             }
 
+            let cutoff = timeline.get_last_record_lsn();
+
+            timeline
+                .update_gc_info(
+                    Vec::new(),
+                    cutoff,
+                    Duration::ZERO,
+                    &CancellationToken::new(),
+                    ctx,
+                )
+                .await?;
             timeline.freeze_and_flush().await?;
             if compact {
-                // this requires timeline to be &Arc<Timeline>
-                timeline.compact(&cancel, EnumSet::empty(), ctx).await?;
+                timeline
+                    .compact(&CancellationToken::new(), EnumSet::empty(), ctx)
+                    .await?;
             }
-
-            // this doesn't really need to use the timeline_id target, but it is closer to what it
-            // originally was.
-            let res = tenant
-                .gc_iteration(Some(timeline.timeline_id), 0, Duration::ZERO, &cancel, ctx)
-                .await?;
-
-            assert_eq!(res.layers_removed, 0, "this never removes anything");
+            timeline.gc().await?;
         }
 
         Ok(())
@@ -4647,7 +4594,7 @@ mod tests {
             .await?;
 
         let lsn = Lsn(0x10);
-        bulk_insert_compact_gc(&tenant, &tline, &ctx, lsn, 50, 10000).await?;
+        bulk_insert_compact_gc(tline.clone(), &ctx, lsn, 50, 10000).await?;
 
         Ok(())
     }
@@ -4678,7 +4625,7 @@ mod tests {
             .await?;
 
         let lsn = Lsn(0x10);
-        bulk_insert_compact_gc(&tenant, &tline, &ctx, lsn, 50, 10000).await?;
+        bulk_insert_compact_gc(tline.clone(), &ctx, lsn, 50, 10000).await?;
 
         let guard = tline.layers.read().await;
         guard.layer_map().dump(true, &ctx).await?;
@@ -4791,7 +4738,15 @@ mod tests {
             .await;
 
         let images = vectored_res?;
-        assert!(images.is_empty());
+        let mut key = NON_INHERITED_RANGE.start;
+        while key < NON_INHERITED_RANGE.end {
+            assert!(matches!(
+                images[&key],
+                Err(PageReconstructError::MissingKey(_))
+            ));
+            key = key.next();
+        }
+
         Ok(())
     }
 
@@ -5124,7 +5079,6 @@ mod tests {
             .await?;
 
         const NUM_KEYS: usize = 1000;
-        let cancel = CancellationToken::new();
 
         let mut test_key = Key::from_hex("010000000033333333444444445500000000").unwrap();
 
@@ -5184,10 +5138,18 @@ mod tests {
             }
 
             // Perform a cycle of flush, and GC
-            tline.freeze_and_flush().await?;
-            tenant
-                .gc_iteration(Some(tline.timeline_id), 0, Duration::ZERO, &cancel, &ctx)
+            let cutoff = tline.get_last_record_lsn();
+            tline
+                .update_gc_info(
+                    Vec::new(),
+                    cutoff,
+                    Duration::ZERO,
+                    &CancellationToken::new(),
+                    &ctx,
+                )
                 .await?;
+            tline.freeze_and_flush().await?;
+            tline.gc().await?;
         }
 
         Ok(())
@@ -5208,8 +5170,6 @@ mod tests {
 
         let mut keyspace = KeySpaceAccum::new();
 
-        let cancel = CancellationToken::new();
-
         // Track when each page was last modified. Used to assert that
         // a read sees the latest page version.
         let mut updated = [Lsn(0); NUM_KEYS];
@@ -5273,11 +5233,21 @@ mod tests {
             }
 
             // Perform a cycle of flush, compact, and GC
-            tline.freeze_and_flush().await?;
-            tline.compact(&cancel, EnumSet::empty(), &ctx).await?;
-            tenant
-                .gc_iteration(Some(tline.timeline_id), 0, Duration::ZERO, &cancel, &ctx)
+            let cutoff = tline.get_last_record_lsn();
+            tline
+                .update_gc_info(
+                    Vec::new(),
+                    cutoff,
+                    Duration::ZERO,
+                    &CancellationToken::new(),
+                    &ctx,
+                )
                 .await?;
+            tline.freeze_and_flush().await?;
+            tline
+                .compact(&CancellationToken::new(), EnumSet::empty(), &ctx)
+                .await?;
+            tline.gc().await?;
         }
 
         Ok(())
@@ -5482,7 +5452,7 @@ mod tests {
 
         let lsn = Lsn(0x10);
         let compact = false;
-        bulk_insert_maybe_compact_gc(&tenant, &tline, &ctx, lsn, 50, 10000, compact).await?;
+        bulk_insert_maybe_compact_gc(tline.clone(), &ctx, lsn, 50, 10000, compact).await?;
 
         let test_key = Key::from_hex("010000000033333333444444445500000000").unwrap();
         let read_lsn = Lsn(u64::MAX - 1);
@@ -5492,108 +5462,4 @@ mod tests {
 
         Ok(())
     }
-
-    #[tokio::test]
-    async fn test_metadata_scan() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_metadata_scan")?;
-        let (tenant, ctx) = harness.load().await;
-        let tline = tenant
-            .create_test_timeline(TIMELINE_ID, Lsn(0x10), DEFAULT_PG_VERSION, &ctx)
-            .await?;
-
-        const NUM_KEYS: usize = 1000;
-        const STEP: usize = 100; // random update + scan base_key + idx * STEP
-
-        let cancel = CancellationToken::new();
-
-        let mut base_key = Key::from_hex("000000000033333333444444445500000000").unwrap();
-        base_key.field1 = AUX_KEY_PREFIX;
-        let mut test_key = base_key;
-
-        // Track when each page was last modified. Used to assert that
-        // a read sees the latest page version.
-        let mut updated = [Lsn(0); NUM_KEYS];
-
-        let mut lsn = Lsn(0x10);
-        #[allow(clippy::needless_range_loop)]
-        for blknum in 0..NUM_KEYS {
-            lsn = Lsn(lsn.0 + 0x10);
-            test_key.field6 = (blknum * STEP) as u32;
-            let mut writer = tline.writer().await;
-            writer
-                .put(
-                    test_key,
-                    lsn,
-                    &Value::Image(test_img(&format!("{} at {}", blknum, lsn))),
-                    &ctx,
-                )
-                .await?;
-            writer.finish_write(lsn);
-            updated[blknum] = lsn;
-            drop(writer);
-        }
-
-        let keyspace = KeySpace::single(base_key..base_key.add((NUM_KEYS * STEP) as u32));
-
-        for _ in 0..10 {
-            // Read all the blocks
-            for (blknum, last_lsn) in updated.iter().enumerate() {
-                test_key.field6 = (blknum * STEP) as u32;
-                assert_eq!(
-                    tline.get(test_key, lsn, &ctx).await?,
-                    test_img(&format!("{} at {}", blknum, last_lsn))
-                );
-            }
-
-            let mut cnt = 0;
-            for (key, value) in tline
-                .get_vectored_impl(
-                    keyspace.clone(),
-                    lsn,
-                    ValuesReconstructState::default(),
-                    &ctx,
-                )
-                .await?
-            {
-                let blknum = key.field6 as usize;
-                let value = value?;
-                assert!(blknum % STEP == 0);
-                let blknum = blknum / STEP;
-                assert_eq!(
-                    value,
-                    test_img(&format!("{} at {}", blknum, updated[blknum]))
-                );
-                cnt += 1;
-            }
-
-            assert_eq!(cnt, NUM_KEYS);
-
-            for _ in 0..NUM_KEYS {
-                lsn = Lsn(lsn.0 + 0x10);
-                let blknum = thread_rng().gen_range(0..NUM_KEYS);
-                test_key.field6 = (blknum * STEP) as u32;
-                let mut writer = tline.writer().await;
-                writer
-                    .put(
-                        test_key,
-                        lsn,
-                        &Value::Image(test_img(&format!("{} at {}", blknum, lsn))),
-                        &ctx,
-                    )
-                    .await?;
-                writer.finish_write(lsn);
-                drop(writer);
-                updated[blknum] = lsn;
-            }
-
-            // Perform a cycle of flush, compact, and GC
-            tline.freeze_and_flush().await?;
-            tline.compact(&cancel, EnumSet::empty(), &ctx).await?;
-            tenant
-                .gc_iteration(Some(tline.timeline_id), 0, Duration::ZERO, &cancel, &ctx)
-                .await?;
-        }
-
-        Ok(())
-    }
 }

pageserver/src/tenant/layer_map.rs

@@ -588,7 +588,7 @@ impl LayerMap {
             let kr = Key::from_i128(current_key)..Key::from_i128(change_key);
             coverage.push((kr, current_val.take()));
             current_key = change_key;
-            current_val.clone_from(&change_val);
+            current_val = change_val.clone();
         }
 
         // Add the final interval
@@ -672,12 +672,12 @@ impl LayerMap {
         // Loop through the delta coverage and recurse on each part
         for (change_key, change_val) in version.delta_coverage.range(start..end) {
             // If there's a relevant delta in this part, add 1 and recurse down
-            if let Some(val) = &current_val {
+            if let Some(val) = current_val {
                 if val.get_lsn_range().end > lsn.start {
                     let kr = Key::from_i128(current_key)..Key::from_i128(change_key);
                     let lr = lsn.start..val.get_lsn_range().start;
                     if !kr.is_empty() {
-                        let base_count = Self::is_reimage_worthy(val, key) as usize;
+                        let base_count = Self::is_reimage_worthy(&val, key) as usize;
                         let new_limit = limit.map(|l| l - base_count);
                         let max_stacked_deltas_underneath = self.count_deltas(&kr, &lr, new_limit);
                         max_stacked_deltas = std::cmp::max(
@@ -689,17 +689,17 @@ impl LayerMap {
             }
 
             current_key = change_key;
-            current_val.clone_from(&change_val);
+            current_val = change_val.clone();
         }
 
         // Consider the last part
-        if let Some(val) = &current_val {
+        if let Some(val) = current_val {
             if val.get_lsn_range().end > lsn.start {
                 let kr = Key::from_i128(current_key)..Key::from_i128(end);
                 let lr = lsn.start..val.get_lsn_range().start;
 
                 if !kr.is_empty() {
-                    let base_count = Self::is_reimage_worthy(val, key) as usize;
+                    let base_count = Self::is_reimage_worthy(&val, key) as usize;
                     let new_limit = limit.map(|l| l - base_count);
                     let max_stacked_deltas_underneath = self.count_deltas(&kr, &lr, new_limit);
                     max_stacked_deltas = std::cmp::max(

pageserver/src/tenant/remote_timeline_client/download.rs

@@ -136,7 +136,7 @@ async fn download_object<'a>(
     src_path: &RemotePath,
     dst_path: &Utf8PathBuf,
     cancel: &CancellationToken,
-    #[cfg_attr(target_os = "macos", allow(unused_variables))] ctx: &RequestContext,
+    ctx: &RequestContext,
 ) -> Result<u64, DownloadError> {
     let res = match crate::virtual_file::io_engine::get() {
         crate::virtual_file::io_engine::IoEngine::NotSet => panic!("unset"),

pageserver/src/tenant/secondary/downloader.rs

@@ -31,10 +31,7 @@ use crate::{
 
 use super::{
     heatmap::HeatMapLayer,
-    scheduler::{
-        self, period_jitter, period_warmup, Completion, JobGenerator, SchedulingResult,
-        TenantBackgroundJobs,
-    },
+    scheduler::{self, Completion, JobGenerator, SchedulingResult, TenantBackgroundJobs},
     SecondaryTenant,
 };
 
@@ -48,6 +45,7 @@ use chrono::format::{DelayedFormat, StrftimeItems};
 use futures::Future;
 use pageserver_api::models::SecondaryProgress;
 use pageserver_api::shard::TenantShardId;
+use rand::Rng;
 use remote_storage::{DownloadError, Etag, GenericRemoteStorage};
 
 use tokio_util::sync::CancellationToken;
@@ -276,7 +274,7 @@ impl JobGenerator<PendingDownload, RunningDownload, CompleteDownload, DownloadCo
         // Update freshened_at even if there was an error: we don't want errored tenants to implicitly
         // take priority to run again.
         let mut detail = secondary_state.detail.lock().unwrap();
-        detail.next_download = Some(Instant::now() + period_jitter(DOWNLOAD_FRESHEN_INTERVAL, 5));
+        detail.next_download = Some(Instant::now() + DOWNLOAD_FRESHEN_INTERVAL);
     }
 
     async fn schedule(&mut self) -> SchedulingResult<PendingDownload> {
@@ -307,9 +305,11 @@ impl JobGenerator<PendingDownload, RunningDownload, CompleteDownload, DownloadCo
                     }
 
                     if detail.next_download.is_none() {
-                        // Initialize randomly in the range from 0 to our interval: this uniformly spreads the start times.  Subsequent
-                        // rounds will use a smaller jitter to avoid accidentally synchronizing later.
-                        detail.next_download = Some(now.checked_add(period_warmup(DOWNLOAD_FRESHEN_INTERVAL)).expect(
+                        // Initialize with a jitter: this spreads initial downloads on startup
+                        // or mass-attach across our freshen interval.
+                        let jittered_period =
+                            rand::thread_rng().gen_range(Duration::ZERO..DOWNLOAD_FRESHEN_INTERVAL);
+                        detail.next_download = Some(now.checked_add(jittered_period).expect(
                         "Using our constant, which is known to be small compared with clock range",
                     ));
                     }

pageserver/src/tenant/secondary/heatmap_uploader.rs

@@ -20,14 +20,12 @@ use crate::{
 
 use futures::Future;
 use pageserver_api::shard::TenantShardId;
+use rand::Rng;
 use remote_storage::{GenericRemoteStorage, TimeoutOrCancel};
 
 use super::{
     heatmap::HeatMapTenant,
-    scheduler::{
-        self, period_jitter, period_warmup, JobGenerator, RunningJob, SchedulingResult,
-        TenantBackgroundJobs,
-    },
+    scheduler::{self, JobGenerator, RunningJob, SchedulingResult, TenantBackgroundJobs},
     CommandRequest, UploadCommand,
 };
 use tokio_util::sync::CancellationToken;
@@ -183,11 +181,15 @@ impl JobGenerator<UploadPending, WriteInProgress, WriteComplete, UploadCommand>
             let state = self
                 .tenants
                 .entry(*tenant.get_tenant_shard_id())
-                .or_insert_with(|| UploaderTenantState {
-                    tenant: Arc::downgrade(&tenant),
-                    last_upload: None,
-                    next_upload: Some(now.checked_add(period_warmup(period)).unwrap_or(now)),
-                    last_digest: None,
+                .or_insert_with(|| {
+                    let jittered_period = rand::thread_rng().gen_range(Duration::ZERO..period);
+
+                    UploaderTenantState {
+                        tenant: Arc::downgrade(&tenant),
+                        last_upload: None,
+                        next_upload: Some(now.checked_add(jittered_period).unwrap_or(now)),
+                        last_digest: None,
+                    }
                 });
 
             // Decline to do the upload if insufficient time has passed
@@ -272,7 +274,7 @@ impl JobGenerator<UploadPending, WriteInProgress, WriteComplete, UploadCommand>
 
             let next_upload = tenant
                 .get_heatmap_period()
-                .and_then(|period| now.checked_add(period_jitter(period, 5)));
+                .and_then(|period| now.checked_add(period));
 
             WriteComplete {
                     tenant_shard_id: *tenant.get_tenant_shard_id(),

pageserver/src/tenant/secondary/scheduler.rs

@@ -1,5 +1,4 @@
 use futures::Future;
-use rand::Rng;
 use std::{
     collections::HashMap,
     marker::PhantomData,
@@ -20,26 +19,6 @@ use super::{CommandRequest, CommandResponse};
 const MAX_SCHEDULING_INTERVAL: Duration = Duration::from_secs(10);
 const MIN_SCHEDULING_INTERVAL: Duration = Duration::from_secs(1);
 
-/// Jitter a Duration by an integer percentage.  Returned values are uniform
-/// in the range 100-pct..100+pct (i.e. a 5% jitter is 5% either way: a ~10% range)
-pub(super) fn period_jitter(d: Duration, pct: u32) -> Duration {
-    if d == Duration::ZERO {
-        d
-    } else {
-        rand::thread_rng().gen_range((d * (100 - pct)) / 100..(d * (100 + pct)) / 100)
-    }
-}
-
-/// When a periodic task first starts, it should wait for some time in the range 0..period, so
-/// that starting many such tasks at the same time spreads them across the time range.
-pub(super) fn period_warmup(period: Duration) -> Duration {
-    if period == Duration::ZERO {
-        period
-    } else {
-        rand::thread_rng().gen_range(Duration::ZERO..period)
-    }
-}
-
 /// Scheduling helper for background work across many tenants.
 ///
 /// Systems that need to run background work across many tenants may use this type

pageserver/src/tenant/size.rs

@@ -118,6 +118,9 @@ pub(super) async fn gather_inputs(
     ctx: &RequestContext,
 ) -> anyhow::Result<ModelInputs> {
     // refresh is needed to update gc related pitr_cutoff and horizon_cutoff
+    //
+    // FIXME: if a single timeline is deleted while refresh gc info is ongoing, we will fail the
+    // whole computation. It does not make sense from the billing perspective.
     tenant
         .refresh_gc_info(cancel, ctx)
         .await
@@ -189,9 +192,7 @@ pub(super) async fn gather_inputs(
         // than a space bound (horizon cutoff).  This means that if someone drops a database and waits for their
         // PITR interval, they will see synthetic size decrease, even if we are still storing data inside
         // horizon_cutoff.
-        let pitr_cutoff = gc_info.cutoffs.pitr;
-        let horizon_cutoff = gc_info.cutoffs.horizon;
-        let mut next_gc_cutoff = pitr_cutoff;
+        let mut next_gc_cutoff = gc_info.pitr_cutoff;
 
         // If the caller provided a shorter retention period, use that instead of the GC cutoff.
         let retention_param_cutoff = if let Some(max_retention_period) = max_retention_period {
@@ -218,8 +219,6 @@ pub(super) async fn gather_inputs(
             .map(|lsn| (lsn, LsnKind::BranchPoint))
             .collect::<Vec<_>>();
 
-        drop(gc_info);
-
         // Add branch points we collected earlier, just in case there were any that were
         // not present in retain_lsns. We will remove any duplicates below later.
         if let Some(this_branchpoints) = branchpoints.get(&timeline_id) {
@@ -298,8 +297,8 @@ pub(super) async fn gather_inputs(
             last_record: last_record_lsn,
             // this is not used above, because it might not have updated recently enough
             latest_gc_cutoff: *timeline.get_latest_gc_cutoff_lsn(),
-            horizon_cutoff,
-            pitr_cutoff,
+            horizon_cutoff: gc_info.horizon_cutoff,
+            pitr_cutoff: gc_info.pitr_cutoff,
             next_gc_cutoff,
             retention_param_cutoff,
         });

pageserver/src/tenant/timeline.rs

@@ -16,10 +16,7 @@ use enumset::EnumSet;
 use fail::fail_point;
 use once_cell::sync::Lazy;
 use pageserver_api::{
-    key::{
-        AUX_FILES_KEY, METADATA_KEY_BEGIN_PREFIX, METADATA_KEY_END_PREFIX, NON_INHERITED_RANGE,
-        NON_INHERITED_SPARSE_RANGE,
-    },
+    key::{AUX_FILES_KEY, NON_INHERITED_RANGE},
     keyspace::{KeySpaceAccum, SparseKeyPartitioning},
     models::{
         CompactionAlgorithm, DownloadRemoteLayersTaskInfo, DownloadRemoteLayersTaskSpawnRequest,
@@ -58,6 +55,7 @@ use std::{
     ops::ControlFlow,
 };
 
+use crate::tenant::timeline::logical_size::CurrentLogicalSize;
 use crate::tenant::{
     layer_map::{LayerMap, SearchResult},
     metadata::TimelineMetadata,
@@ -79,9 +77,6 @@ use crate::{
 use crate::{
     disk_usage_eviction_task::EvictionCandidate, tenant::storage_layer::delta_layer::DeltaEntry,
 };
-use crate::{
-    metrics::ScanLatencyOngoingRecording, tenant::timeline::logical_size::CurrentLogicalSize,
-};
 use crate::{pgdatadir_mapping::LsnForTimestamp, tenant::tasks::BackgroundLoopKind};
 use crate::{
     pgdatadir_mapping::{AuxFilesDirectory, DirectoryKind},
@@ -330,7 +325,7 @@ pub struct Timeline {
 
     // List of child timelines and their branch points. This is needed to avoid
     // garbage collecting data that is still needed by the child timelines.
-    pub(crate) gc_info: std::sync::RwLock<GcInfo>,
+    pub gc_info: std::sync::RwLock<GcInfo>,
 
     // It may change across major versions so for simplicity
     // keep it after running initdb for a timeline.
@@ -414,59 +409,33 @@ pub struct WalReceiverInfo {
     pub last_received_msg_ts: u128,
 }
 
+///
 /// Information about how much history needs to be retained, needed by
 /// Garbage Collection.
-#[derive(Default)]
-pub(crate) struct GcInfo {
+///
+pub struct GcInfo {
     /// Specific LSNs that are needed.
     ///
     /// Currently, this includes all points where child branches have
     /// been forked off from. In the future, could also include
     /// explicit user-defined snapshot points.
-    pub(crate) retain_lsns: Vec<Lsn>,
+    pub retain_lsns: Vec<Lsn>,
 
-    /// The cutoff coordinates, which are combined by selecting the minimum.
-    pub(crate) cutoffs: GcCutoffs,
-}
-
-impl GcInfo {
-    pub(crate) fn min_cutoff(&self) -> Lsn {
-        self.cutoffs.select_min()
-    }
-}
-
-/// The `GcInfo` component describing which Lsns need to be retained.
-#[derive(Debug)]
-pub(crate) struct GcCutoffs {
-    /// Keep everything newer than this point.
+    /// In addition to 'retain_lsns', keep everything newer than this
+    /// point.
     ///
     /// This is calculated by subtracting 'gc_horizon' setting from
     /// last-record LSN
     ///
     /// FIXME: is this inclusive or exclusive?
-    pub(crate) horizon: Lsn,
+    pub horizon_cutoff: Lsn,
 
     /// In addition to 'retain_lsns' and 'horizon_cutoff', keep everything newer than this
     /// point.
     ///
     /// This is calculated by finding a number such that a record is needed for PITR
     /// if only if its LSN is larger than 'pitr_cutoff'.
-    pub(crate) pitr: Lsn,
-}
-
-impl Default for GcCutoffs {
-    fn default() -> Self {
-        Self {
-            horizon: Lsn::INVALID,
-            pitr: Lsn::INVALID,
-        }
-    }
-}
-
-impl GcCutoffs {
-    fn select_min(&self) -> Lsn {
-        std::cmp::min(self.horizon, self.pitr)
-    }
+    pub pitr_cutoff: Lsn,
 }
 
 /// An error happened in a get() operation.
@@ -890,15 +859,16 @@ impl Timeline {
                             value
                         }
                     }
-                    None => Err(PageReconstructError::MissingKey(MissingKeyError {
-                        key,
-                        shard: self.shard_identity.get_shard_number(&key),
-                        cont_lsn: Lsn(0),
-                        request_lsn: lsn,
-                        ancestor_lsn: None,
-                        traversal_path: Vec::new(),
-                        backtrace: None,
-                    })),
+                    None => {
+                        error!(
+                            "Expected {}, but singular vectored get returned nothing",
+                            key
+                        );
+                        Err(PageReconstructError::Other(anyhow!(
+                            "Singular vectored get did not return a value for {}",
+                            key
+                        )))
+                    }
                 }
             }
         }
@@ -1048,70 +1018,6 @@ impl Timeline {
         res
     }
 
-    /// Scan the keyspace and return all existing key-values in the keyspace. This currently uses vectored
-    /// get underlying. Normal vectored get would throw an error when a key in the keyspace is not found
-    /// during the search, but for the scan interface, it returns all existing key-value pairs, and does
-    /// not expect each single key in the key space will be found. The semantics is closer to the RocksDB
-    /// scan iterator interface. We could optimize this interface later to avoid some checks in the vectored
-    /// get path to maintain and split the probing and to-be-probe keyspace. We also need to ensure that
-    /// the scan operation will not cause OOM in the future.
-    #[allow(dead_code)]
-    pub(crate) async fn scan(
-        &self,
-        keyspace: KeySpace,
-        lsn: Lsn,
-        ctx: &RequestContext,
-    ) -> Result<BTreeMap<Key, Result<Bytes, PageReconstructError>>, GetVectoredError> {
-        if !lsn.is_valid() {
-            return Err(GetVectoredError::InvalidLsn(lsn));
-        }
-
-        trace!(
-            "key-value scan request for {:?}@{} from task kind {:?}",
-            keyspace,
-            lsn,
-            ctx.task_kind()
-        );
-
-        // We should generalize this into Keyspace::contains in the future.
-        for range in &keyspace.ranges {
-            if range.start.field1 < METADATA_KEY_BEGIN_PREFIX
-                || range.end.field1 >= METADATA_KEY_END_PREFIX
-            {
-                return Err(GetVectoredError::Other(anyhow::anyhow!(
-                    "only metadata keyspace can be scanned"
-                )));
-            }
-        }
-
-        let start = crate::metrics::SCAN_LATENCY
-            .for_task_kind(ctx.task_kind())
-            .map(ScanLatencyOngoingRecording::start_recording);
-
-        // start counting after throttle so that throttle time
-        // is always less than observation time
-        let throttled = self
-            .timeline_get_throttle
-            // assume scan = 1 quota for now until we find a better way to process this
-            .throttle(ctx, 1)
-            .await;
-
-        let vectored_res = self
-            .get_vectored_impl(
-                keyspace.clone(),
-                lsn,
-                ValuesReconstructState::default(),
-                ctx,
-            )
-            .await;
-
-        if let Some(recording) = start {
-            recording.observe(throttled);
-        }
-
-        vectored_res
-    }
-
     /// Not subject to [`Self::timeline_get_throttle`].
     pub(super) async fn get_vectored_sequential_impl(
         &self,
@@ -1120,7 +1026,6 @@ impl Timeline {
         ctx: &RequestContext,
     ) -> Result<BTreeMap<Key, Result<Bytes, PageReconstructError>>, GetVectoredError> {
         let mut values = BTreeMap::new();
-
         for range in keyspace.ranges {
             let mut key = range.start;
             while key != range.end {
@@ -1133,16 +1038,12 @@ impl Timeline {
                     Err(Cancelled | AncestorStopping(_)) => {
                         return Err(GetVectoredError::Cancelled)
                     }
-                    Err(MissingKey(_))
-                        if NON_INHERITED_RANGE.contains(&key)
-                            || NON_INHERITED_SPARSE_RANGE.contains(&key) =>
-                    {
-                        // Ignore missing key error for aux key range. TODO: currently, we assume non_inherited_range == aux_key_range.
-                        // When we add more types of keys into the page server, we should revisit this part of code and throw errors
-                        // accordingly.
-                        key = key.next();
-                    }
-                    Err(MissingKey(err)) => {
+                    // we only capture stuck_at_lsn=false now until we figure out https://github.com/neondatabase/neon/issues/7380
+                    Err(MissingKey(err)) if !NON_INHERITED_RANGE.contains(&key) => {
+                        // The vectored read path handles non inherited keys specially.
+                        // If such a a key cannot be reconstructed from the current timeline,
+                        // the vectored read path returns a key level error as opposed to a top
+                        // level error.
                         return Err(GetVectoredError::MissingKey(err));
                     }
                     Err(Other(err))
@@ -1230,11 +1131,6 @@ impl Timeline {
         lsn: Lsn,
         ctx: &RequestContext,
     ) {
-        if keyspace.overlaps(&Key::metadata_key_range()) {
-            // skip validation for metadata key range
-            return;
-        }
-
         let sequential_res = self
             .get_vectored_sequential_impl(keyspace.clone(), lsn, ctx)
             .await;
@@ -1259,7 +1155,7 @@ impl Timeline {
                                " - keyspace={:?} lsn={}"),
                        seq_err, keyspace, lsn) },
             (Ok(_), Err(GetVectoredError::GetReadyAncestorError(GetReadyAncestorError::AncestorLsnTimeout(_)))) => {
-                // Sequential get runs after vectored get, so it is possible for the later
+                // Sequential get runs after vectored get, so it is possible for the later 
                 // to time out while waiting for its ancestor's Lsn to become ready and for the
                 // former to succeed (it essentially has a doubled wait time).
             },
@@ -2201,7 +2097,11 @@ impl Timeline {
 
                 write_lock: tokio::sync::Mutex::new(None),
 
-                gc_info: std::sync::RwLock::new(GcInfo::default()),
+                gc_info: std::sync::RwLock::new(GcInfo {
+                    retain_lsns: Vec::new(),
+                    horizon_cutoff: Lsn(0),
+                    pitr_cutoff: Lsn(0),
+                }),
 
                 latest_gc_cutoff_lsn: Rcu::new(metadata.latest_gc_cutoff_lsn()),
                 initdb_lsn: metadata.initdb_lsn(),
@@ -3287,12 +3187,36 @@ impl Timeline {
             // Do not descend into the ancestor timeline for aux files.
             // We don't return a blanket [`GetVectoredError::MissingKey`] to avoid
             // stalling compaction.
-            keyspace.remove_overlapping_with(&KeySpace {
-                ranges: vec![NON_INHERITED_RANGE, NON_INHERITED_SPARSE_RANGE],
-            });
+            // TODO(chi): this will need to be updated for aux files v2 storage
+            if keyspace.overlaps(&NON_INHERITED_RANGE) {
+                let removed = keyspace.remove_overlapping_with(&KeySpace {
+                    ranges: vec![NON_INHERITED_RANGE],
+                });
+
+                for range in removed.ranges {
+                    let mut key = range.start;
+                    while key < range.end {
+                        reconstruct_state.on_key_error(
+                            key,
+                            PageReconstructError::MissingKey(MissingKeyError {
+                                key,
+                                shard: self.shard_identity.get_shard_number(&key),
+                                cont_lsn,
+                                request_lsn,
+                                ancestor_lsn: None,
+                                traversal_path: Vec::default(),
+                                backtrace: if cfg!(test) {
+                                    Some(std::backtrace::Backtrace::force_capture())
+                                } else {
+                                    None
+                                },
+                            }),
+                        );
+                        key = key.next();
+                    }
+                }
+            }
 
-            // Keyspace is fully retrieved, no ancestor timeline, or metadata scan (where we do not look
-            // into ancestor timelines). TODO: is there any other metadata which we want to inherit?
             if keyspace.total_raw_size() == 0 || timeline.ancestor_timeline.is_none() {
                 break;
             }
@@ -4459,7 +4383,7 @@ impl Timeline {
         Ok(())
     }
 
-    /// Find the Lsns above which layer files need to be retained on
+    /// Update information about which layer files need to be retained on
     /// garbage collection. This is separate from actually performing the GC,
     /// and is updated more frequently, so that compaction can remove obsolete
     /// page versions more aggressively.
@@ -4467,6 +4391,17 @@ impl Timeline {
     /// TODO: that's wishful thinking, compaction doesn't actually do that
     /// currently.
     ///
+    /// The caller specifies how much history is needed with the 3 arguments:
+    ///
+    /// retain_lsns: keep a version of each page at these LSNs
+    /// cutoff_horizon: also keep everything newer than this LSN
+    /// pitr: the time duration required to keep data for PITR
+    ///
+    /// The 'retain_lsns' list is currently used to prevent removing files that
+    /// are needed by child timelines. In the future, the user might be able to
+    /// name additional points in time to retain. The caller is responsible for
+    /// collecting that information.
+    ///
     /// The 'cutoff_horizon' point is used to retain recent versions that might still be
     /// needed by read-only nodes. (As of this writing, the caller just passes
     /// the latest LSN subtracted by a constant, and doesn't do anything smart
@@ -4474,22 +4409,29 @@ impl Timeline {
     ///
     /// The 'pitr' duration is used to calculate a 'pitr_cutoff', which can be used to determine
     /// whether a record is needed for PITR.
+    ///
+    /// NOTE: This function holds a short-lived lock to protect the 'gc_info'
+    /// field, so that the three values passed as argument are stored
+    /// atomically. But the caller is responsible for ensuring that no new
+    /// branches are created that would need to be included in 'retain_lsns',
+    /// for example. The caller should hold `Tenant::gc_cs` lock to ensure
+    /// that.
+    ///
     #[instrument(skip_all, fields(timeline_id=%self.timeline_id))]
-    pub(super) async fn find_gc_cutoffs(
+    pub(super) async fn update_gc_info(
         &self,
+        retain_lsns: Vec<Lsn>,
         cutoff_horizon: Lsn,
         pitr: Duration,
         cancel: &CancellationToken,
         ctx: &RequestContext,
-    ) -> anyhow::Result<GcCutoffs> {
+    ) -> anyhow::Result<()> {
         let _timer = self
             .metrics
-            .find_gc_cutoffs_histo
+            .update_gc_info_histo
             .start_timer()
             .record_on_drop();
 
-        pausable_failpoint!("Timeline::find_gc_cutoffs-pausable");
-
         // First, calculate pitr_cutoff_timestamp and then convert it to LSN.
         //
         // Some unit tests depend on garbage-collection working even when
@@ -4539,10 +4481,14 @@ impl Timeline {
             self.get_last_record_lsn()
         };
 
-        Ok(GcCutoffs {
-            horizon: cutoff_horizon,
-            pitr: pitr_cutoff,
-        })
+        // Grab the lock and update the values
+        *self.gc_info.write().unwrap() = GcInfo {
+            retain_lsns,
+            horizon_cutoff: cutoff_horizon,
+            pitr_cutoff,
+        };
+
+        Ok(())
     }
 
     /// Garbage collect layer files on a timeline that are no longer needed.
@@ -4571,8 +4517,8 @@ impl Timeline {
         let (horizon_cutoff, pitr_cutoff, retain_lsns) = {
             let gc_info = self.gc_info.read().unwrap();
 
-            let horizon_cutoff = min(gc_info.cutoffs.horizon, self.get_disk_consistent_lsn());
-            let pitr_cutoff = gc_info.cutoffs.pitr;
+            let horizon_cutoff = min(gc_info.horizon_cutoff, self.get_disk_consistent_lsn());
+            let pitr_cutoff = gc_info.pitr_cutoff;
             let retain_lsns = gc_info.retain_lsns.clone();
             (horizon_cutoff, pitr_cutoff, retain_lsns)
         };

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

@@ -1535,7 +1535,7 @@ mod tests {
 
         let harness = TenantHarness::create("switch_to_same_availability_zone")?;
         let mut state = dummy_state(&harness).await;
-        state.conf.availability_zone.clone_from(&test_az);
+        state.conf.availability_zone = test_az.clone();
         let current_lsn = Lsn(100_000).align();
         let now = Utc::now().naive_utc();
 
@@ -1568,7 +1568,7 @@ mod tests {
         // We have another safekeeper with the same commit_lsn, and it have the same availability zone as
         // the current pageserver.
         let mut same_az_sk = dummy_broker_sk_timeline(current_lsn.0, "same_az", now);
-        same_az_sk.timeline.availability_zone.clone_from(&test_az);
+        same_az_sk.timeline.availability_zone = test_az.clone();
 
         state.wal_stream_candidates = HashMap::from([
             (

pageserver/src/virtual_file/owned_buffers_io/write.rs

@@ -76,11 +76,14 @@ where
     }
 
     #[cfg_attr(target_os = "macos", allow(dead_code))]
-    pub async fn write_buffered<S: IoBuf + Send>(
+    pub async fn write_buffered<S: IoBuf>(
         &mut self,
         chunk: Slice<S>,
         ctx: &RequestContext,
-    ) -> std::io::Result<(usize, S)> {
+    ) -> std::io::Result<(usize, S)>
+    where
+        S: IoBuf + Send,
+    {
         let chunk_len = chunk.len();
         // avoid memcpy for the middle of the chunk
         if chunk.len() >= self.buf().cap() {

proxy/src/bin/proxy.rs

@@ -118,11 +118,8 @@ struct ProxyCliArgs {
     #[clap(long, default_value = config::CacheOptions::CACHE_DEFAULT_OPTIONS)]
     wake_compute_cache: String,
     /// lock for `wake_compute` api method. example: "shards=32,permits=4,epoch=10m,timeout=1s". (use `permits=0` to disable).
-    #[clap(long, default_value = config::ConcurrencyLockOptions::DEFAULT_OPTIONS_WAKE_COMPUTE_LOCK)]
+    #[clap(long, default_value = config::WakeComputeLockOptions::DEFAULT_OPTIONS_WAKE_COMPUTE_LOCK)]
     wake_compute_lock: String,
-    /// lock for `connect_compute` api method. example: "shards=32,permits=4,epoch=10m,timeout=1s". (use `permits=0` to disable).
-    #[clap(long, default_value = config::ConcurrencyLockOptions::DEFAULT_OPTIONS_CONNECT_COMPUTE_LOCK)]
-    connect_compute_lock: String,
     /// Allow self-signed certificates for compute nodes (for testing)
     #[clap(long, default_value_t = false, value_parser = clap::builder::BoolishValueParser::new(), action = clap::ArgAction::Set)]
     allow_self_signed_compute: bool,
@@ -532,21 +529,24 @@ fn build_config(args: &ProxyCliArgs) -> anyhow::Result<&'static ProxyConfig> {
                 endpoint_cache_config,
             )));
 
-            let config::ConcurrencyLockOptions {
+            let config::WakeComputeLockOptions {
                 shards,
                 permits,
                 epoch,
                 timeout,
             } = args.wake_compute_lock.parse()?;
             info!(permits, shards, ?epoch, "Using NodeLocks (wake_compute)");
-            let locks = Box::leak(Box::new(console::locks::ApiLocks::new(
-                "wake_compute_lock",
-                permits,
-                shards,
-                timeout,
-                epoch,
-                &Metrics::get().wake_compute_lock,
-            )?));
+            let locks = Box::leak(Box::new(
+                console::locks::ApiLocks::new(
+                    "wake_compute_lock",
+                    permits,
+                    shards,
+                    timeout,
+                    epoch,
+                    &Metrics::get().wake_compute_lock,
+                )
+                .unwrap(),
+            ));
             tokio::spawn(locks.garbage_collect_worker());
 
             let url = args.auth_endpoint.parse()?;
@@ -572,23 +572,6 @@ fn build_config(args: &ProxyCliArgs) -> anyhow::Result<&'static ProxyConfig> {
             auth::BackendType::Link(MaybeOwned::Owned(url), ())
         }
     };
-
-    let config::ConcurrencyLockOptions {
-        shards,
-        permits,
-        epoch,
-        timeout,
-    } = args.connect_compute_lock.parse()?;
-    info!(permits, shards, ?epoch, "Using NodeLocks (connect_compute)");
-    let connect_compute_locks = console::locks::ApiLocks::new(
-        "connect_compute_lock",
-        permits,
-        shards,
-        timeout,
-        epoch,
-        &Metrics::get().proxy.connect_compute_lock,
-    )?;
-
     let http_config = HttpConfig {
         request_timeout: args.sql_over_http.sql_over_http_timeout,
         pool_options: GlobalConnPoolOptions {
@@ -624,14 +607,11 @@ fn build_config(args: &ProxyCliArgs) -> anyhow::Result<&'static ProxyConfig> {
         region: args.region.clone(),
         aws_region: args.aws_region.clone(),
         wake_compute_retry_config: config::RetryConfig::parse(&args.wake_compute_retry)?,
-        connect_compute_locks,
         connect_to_compute_retry_config: config::RetryConfig::parse(
             &args.connect_to_compute_retry,
         )?,
     }));
 
-    tokio::spawn(config.connect_compute_locks.garbage_collect_worker());
-
     Ok(config)
 }
 

proxy/src/cache/project_info.rs

@@ -5,11 +5,9 @@ use std::{
     time::Duration,
 };
 
-use async_trait::async_trait;
 use dashmap::DashMap;
 use rand::{thread_rng, Rng};
 use smol_str::SmolStr;
-use tokio::sync::Mutex;
 use tokio::time::Instant;
 use tracing::{debug, info};
 
@@ -23,12 +21,11 @@ use crate::{
 
 use super::{Cache, Cached};
 
-#[async_trait]
 pub trait ProjectInfoCache {
     fn invalidate_allowed_ips_for_project(&self, project_id: ProjectIdInt);
     fn invalidate_role_secret_for_project(&self, project_id: ProjectIdInt, role_name: RoleNameInt);
-    async fn decrement_active_listeners(&self);
-    async fn increment_active_listeners(&self);
+    fn enable_ttl(&self);
+    fn disable_ttl(&self);
 }
 
 struct Entry<T> {
@@ -119,10 +116,8 @@ pub struct ProjectInfoCacheImpl {
 
     start_time: Instant,
     ttl_disabled_since_us: AtomicU64,
-    active_listeners_lock: Mutex<usize>,
 }
 
-#[async_trait]
 impl ProjectInfoCache for ProjectInfoCacheImpl {
     fn invalidate_allowed_ips_for_project(&self, project_id: ProjectIdInt) {
         info!("invalidating allowed ips for project `{}`", project_id);
@@ -153,27 +148,15 @@ impl ProjectInfoCache for ProjectInfoCacheImpl {
             }
         }
     }
-    async fn decrement_active_listeners(&self) {
-        let mut listeners_guard = self.active_listeners_lock.lock().await;
-        if *listeners_guard == 0 {
-            tracing::error!("active_listeners count is already 0, something is broken");
-            return;
-        }
-        *listeners_guard -= 1;
-        if *listeners_guard == 0 {
-            self.ttl_disabled_since_us
-                .store(u64::MAX, std::sync::atomic::Ordering::SeqCst);
-        }
+    fn enable_ttl(&self) {
+        self.ttl_disabled_since_us
+            .store(u64::MAX, std::sync::atomic::Ordering::Relaxed);
     }
 
-    async fn increment_active_listeners(&self) {
-        let mut listeners_guard = self.active_listeners_lock.lock().await;
-        *listeners_guard += 1;
-        if *listeners_guard == 1 {
-            let new_ttl = (self.start_time.elapsed() + self.config.ttl).as_micros() as u64;
-            self.ttl_disabled_since_us
-                .store(new_ttl, std::sync::atomic::Ordering::SeqCst);
-        }
+    fn disable_ttl(&self) {
+        let new_ttl = (self.start_time.elapsed() + self.config.ttl).as_micros() as u64;
+        self.ttl_disabled_since_us
+            .store(new_ttl, std::sync::atomic::Ordering::Relaxed);
     }
 }
 
@@ -185,7 +168,6 @@ impl ProjectInfoCacheImpl {
             config,
             ttl_disabled_since_us: AtomicU64::new(u64::MAX),
             start_time: Instant::now(),
-            active_listeners_lock: Mutex::new(0),
         }
     }
 
@@ -450,7 +432,7 @@ mod tests {
             ttl: Duration::from_secs(1),
             gc_interval: Duration::from_secs(600),
         }));
-        cache.clone().increment_active_listeners().await;
+        cache.clone().disable_ttl();
         tokio::time::advance(Duration::from_secs(2)).await;
 
         let project_id: ProjectId = "project".into();
@@ -507,7 +489,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_increment_active_listeners_invalidate_added_before() {
+    async fn test_disable_ttl_invalidate_added_before() {
         tokio::time::pause();
         let cache = Arc::new(ProjectInfoCacheImpl::new(ProjectInfoCacheOptions {
             size: 2,
@@ -532,7 +514,7 @@ mod tests {
             (&user1).into(),
             secret1.clone(),
         );
-        cache.clone().increment_active_listeners().await;
+        cache.clone().disable_ttl();
         tokio::time::advance(Duration::from_millis(100)).await;
         cache.insert_role_secret(
             (&project_id).into(),

proxy/src/compute.rs

@@ -6,7 +6,6 @@ use crate::{
     error::{ReportableError, UserFacingError},
     metrics::{Metrics, NumDbConnectionsGuard},
     proxy::neon_option,
-    Host,
 };
 use futures::{FutureExt, TryFutureExt};
 use itertools::Itertools;
@@ -102,16 +101,6 @@ impl ConnCfg {
         }
     }
 
-    pub fn get_host(&self) -> Result<Host, WakeComputeError> {
-        match self.0.get_hosts() {
-            [tokio_postgres::config::Host::Tcp(s)] => Ok(s.into()),
-            // we should not have multiple address or unix addresses.
-            _ => Err(WakeComputeError::BadComputeAddress(
-                "invalid compute address".into(),
-            )),
-        }
-    }
-
     /// Apply startup message params to the connection config.
     pub fn set_startup_params(&mut self, params: &StartupMessageParams) {
         // Only set `user` if it's not present in the config.

proxy/src/config.rs

@@ -1,9 +1,7 @@
 use crate::{
     auth::{self, backend::AuthRateLimiter},
-    console::locks::ApiLocks,
     rate_limiter::RateBucketInfo,
     serverless::GlobalConnPoolOptions,
-    Host,
 };
 use anyhow::{bail, ensure, Context, Ok};
 use itertools::Itertools;
@@ -36,7 +34,6 @@ pub struct ProxyConfig {
     pub handshake_timeout: Duration,
     pub aws_region: String,
     pub wake_compute_retry_config: RetryConfig,
-    pub connect_compute_locks: ApiLocks<Host>,
     pub connect_to_compute_retry_config: RetryConfig,
 }
 
@@ -576,7 +573,7 @@ impl RetryConfig {
 }
 
 /// Helper for cmdline cache options parsing.
-pub struct ConcurrencyLockOptions {
+pub struct WakeComputeLockOptions {
     /// The number of shards the lock map should have
     pub shards: usize,
     /// The number of allowed concurrent requests for each endpoitn
@@ -587,12 +584,9 @@ pub struct ConcurrencyLockOptions {
     pub timeout: Duration,
 }
 
-impl ConcurrencyLockOptions {
+impl WakeComputeLockOptions {
     /// Default options for [`crate::console::provider::ApiLocks`].
     pub const DEFAULT_OPTIONS_WAKE_COMPUTE_LOCK: &'static str = "permits=0";
-    /// Default options for [`crate::console::provider::ApiLocks`].
-    pub const DEFAULT_OPTIONS_CONNECT_COMPUTE_LOCK: &'static str =
-        "shards=64,permits=50,epoch=10m,timeout=500ms";
 
     // pub const DEFAULT_OPTIONS_WAKE_COMPUTE_LOCK: &'static str = "shards=32,permits=4,epoch=10m,timeout=1s";
 
@@ -642,7 +636,7 @@ impl ConcurrencyLockOptions {
     }
 }
 
-impl FromStr for ConcurrencyLockOptions {
+impl FromStr for WakeComputeLockOptions {
     type Err = anyhow::Error;
 
     fn from_str(options: &str) -> Result<Self, Self::Err> {
@@ -678,7 +672,7 @@ mod tests {
 
     #[test]
     fn test_parse_lock_options() -> anyhow::Result<()> {
-        let ConcurrencyLockOptions {
+        let WakeComputeLockOptions {
             epoch,
             permits,
             shards,
@@ -689,7 +683,7 @@ mod tests {
         assert_eq!(shards, 32);
         assert_eq!(permits, 4);
 
-        let ConcurrencyLockOptions {
+        let WakeComputeLockOptions {
             epoch,
             permits,
             shards,
@@ -700,7 +694,7 @@ mod tests {
         assert_eq!(shards, 16);
         assert_eq!(permits, 8);
 
-        let ConcurrencyLockOptions {
+        let WakeComputeLockOptions {
             epoch,
             permits,
             shards,

proxy/src/console/provider.rs

@@ -17,7 +17,7 @@ use crate::{
     scram, EndpointCacheKey,
 };
 use dashmap::DashMap;
-use std::{hash::Hash, sync::Arc, time::Duration};
+use std::{sync::Arc, time::Duration};
 use tokio::sync::{OwnedSemaphorePermit, Semaphore};
 use tokio::time::Instant;
 use tracing::info;
@@ -447,16 +447,16 @@ impl ApiCaches {
 }
 
 /// Various caches for [`console`](super).
-pub struct ApiLocks<K> {
+pub struct ApiLocks {
     name: &'static str,
-    node_locks: DashMap<K, Arc<Semaphore>>,
+    node_locks: DashMap<EndpointCacheKey, Arc<Semaphore>>,
     permits: usize,
     timeout: Duration,
     epoch: std::time::Duration,
     metrics: &'static ApiLockMetrics,
 }
 
-impl<K: Hash + Eq + Clone> ApiLocks<K> {
+impl ApiLocks {
     pub fn new(
         name: &'static str,
         permits: usize,
@@ -475,7 +475,10 @@ impl<K: Hash + Eq + Clone> ApiLocks<K> {
         })
     }
 
-    pub async fn get_permit(&self, key: &K) -> Result<WakeComputePermit, errors::WakeComputeError> {
+    pub async fn get_wake_compute_permit(
+        &self,
+        key: &EndpointCacheKey,
+    ) -> Result<WakeComputePermit, errors::WakeComputeError> {
         if self.permits == 0 {
             return Ok(WakeComputePermit { permit: None });
         }

proxy/src/console/provider/neon.rs

@@ -13,7 +13,7 @@ use crate::{
     http,
     metrics::{CacheOutcome, Metrics},
     rate_limiter::EndpointRateLimiter,
-    scram, EndpointCacheKey, Normalize,
+    scram, Normalize,
 };
 use crate::{cache::Cached, context::RequestMonitoring};
 use futures::TryFutureExt;
@@ -25,7 +25,7 @@ use tracing::{error, info, info_span, warn, Instrument};
 pub struct Api {
     endpoint: http::Endpoint,
     pub caches: &'static ApiCaches,
-    pub locks: &'static ApiLocks<EndpointCacheKey>,
+    pub locks: &'static ApiLocks,
     pub endpoint_rate_limiter: Arc<EndpointRateLimiter>,
     jwt: String,
 }
@@ -35,7 +35,7 @@ impl Api {
     pub fn new(
         endpoint: http::Endpoint,
         caches: &'static ApiCaches,
-        locks: &'static ApiLocks<EndpointCacheKey>,
+        locks: &'static ApiLocks,
         endpoint_rate_limiter: Arc<EndpointRateLimiter>,
     ) -> Self {
         let jwt: String = match std::env::var("NEON_PROXY_TO_CONTROLPLANE_TOKEN") {
@@ -289,7 +289,7 @@ impl super::Api for Api {
             return Err(WakeComputeError::TooManyConnections);
         }
 
-        let permit = self.locks.get_permit(&key).await?;
+        let permit = self.locks.get_wake_compute_permit(&key).await?;
 
         // after getting back a permit - it's possible the cache was filled
         // double check

proxy/src/lib.rs

@@ -159,9 +159,6 @@ smol_str_wrapper!(EndpointCacheKey);
 
 smol_str_wrapper!(DbName);
 
-// postgres hostname, will likely be a port:ip addr
-smol_str_wrapper!(Host);
-
 // Endpoints are a bit tricky. Rare they might be branches or projects.
 impl EndpointId {
     pub fn is_endpoint(&self) -> bool {

proxy/src/metrics.rs

@@ -126,9 +126,6 @@ pub struct ProxyMetrics {
 
     /// Number of events consumed from redis (per event type).
     pub redis_events_count: CounterVec<StaticLabelSet<RedisEventsCount>>,
-
-    #[metric(namespace = "connect_compute_lock")]
-    pub connect_compute_lock: ApiLockMetrics,
 }
 
 #[derive(MetricGroup)]
@@ -152,12 +149,6 @@ impl Default for ProxyMetrics {
     }
 }
 
-impl Default for ApiLockMetrics {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
 #[derive(FixedCardinalityLabel, Copy, Clone)]
 #[label(singleton = "direction")]
 pub enum HttpDirection {

proxy/src/proxy.rs

@@ -301,10 +301,7 @@ pub async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
 
     let mut node = connect_to_compute(
         ctx,
-        &TcpMechanism {
-            params: &params,
-            locks: &config.connect_compute_locks,
-        },
+        &TcpMechanism { params: &params },
         &user_info,
         mode.allow_self_signed_compute(config),
         config.wake_compute_retry_config,

proxy/src/proxy/connect_compute.rs

@@ -2,7 +2,7 @@ use crate::{
     auth::backend::ComputeCredentialKeys,
     compute::{self, PostgresConnection},
     config::RetryConfig,
-    console::{self, errors::WakeComputeError, locks::ApiLocks, CachedNodeInfo, NodeInfo},
+    console::{self, errors::WakeComputeError, CachedNodeInfo, NodeInfo},
     context::RequestMonitoring,
     error::ReportableError,
     metrics::{ConnectOutcome, ConnectionFailureKind, Metrics, RetriesMetricGroup, RetryType},
@@ -10,7 +10,6 @@ use crate::{
         retry::{retry_after, ShouldRetry},
         wake_compute::wake_compute,
     },
-    Host,
 };
 use async_trait::async_trait;
 use pq_proto::StartupMessageParams;
@@ -65,9 +64,6 @@ pub trait ComputeConnectBackend {
 pub struct TcpMechanism<'a> {
     /// KV-dictionary with PostgreSQL connection params.
     pub params: &'a StartupMessageParams,
-
-    /// connect_to_compute concurrency lock
-    pub locks: &'static ApiLocks<Host>,
 }
 
 #[async_trait]
@@ -83,8 +79,6 @@ impl ConnectMechanism for TcpMechanism<'_> {
         node_info: &console::CachedNodeInfo,
         timeout: time::Duration,
     ) -> Result<PostgresConnection, Self::Error> {
-        let host = node_info.config.get_host()?;
-        let _permit = self.locks.get_permit(&host).await?;
         node_info.connect(ctx, timeout).await
     }
 

proxy/src/redis/notifications.rs

@@ -4,7 +4,6 @@ use futures::StreamExt;
 use pq_proto::CancelKeyData;
 use redis::aio::PubSub;
 use serde::{Deserialize, Serialize};
-use tokio_util::sync::CancellationToken;
 use uuid::Uuid;
 
 use super::connection_with_credentials_provider::ConnectionWithCredentialsProvider;
@@ -78,16 +77,6 @@ struct MessageHandler<C: ProjectInfoCache + Send + Sync + 'static> {
     region_id: String,
 }
 
-impl<C: ProjectInfoCache + Send + Sync + 'static> Clone for MessageHandler<C> {
-    fn clone(&self) -> Self {
-        Self {
-            cache: self.cache.clone(),
-            cancellation_handler: self.cancellation_handler.clone(),
-            region_id: self.region_id.clone(),
-        }
-    }
-}
-
 impl<C: ProjectInfoCache + Send + Sync + 'static> MessageHandler<C> {
     pub fn new(
         cache: Arc<C>,
@@ -100,11 +89,11 @@ impl<C: ProjectInfoCache + Send + Sync + 'static> MessageHandler<C> {
             region_id,
         }
     }
-    pub async fn increment_active_listeners(&self) {
-        self.cache.increment_active_listeners().await;
+    pub fn disable_ttl(&self) {
+        self.cache.disable_ttl();
     }
-    pub async fn decrement_active_listeners(&self) {
-        self.cache.decrement_active_listeners().await;
+    pub fn enable_ttl(&self) {
+        self.cache.enable_ttl();
     }
     #[tracing::instrument(skip(self, msg), fields(session_id = tracing::field::Empty))]
     async fn handle_message(&self, msg: redis::Msg) -> anyhow::Result<()> {
@@ -193,24 +182,37 @@ fn invalidate_cache<C: ProjectInfoCache>(cache: Arc<C>, msg: Notification) {
     }
 }
 
-async fn handle_messages<C: ProjectInfoCache + Send + Sync + 'static>(
-    handler: MessageHandler<C>,
+/// Handle console's invalidation messages.
+#[tracing::instrument(name = "console_notifications", skip_all)]
+pub async fn task_main<C>(
     redis: ConnectionWithCredentialsProvider,
-    cancellation_token: CancellationToken,
-) -> anyhow::Result<()> {
+    cache: Arc<C>,
+    cancel_map: CancelMap,
+    region_id: String,
+) -> anyhow::Result<Infallible>
+where
+    C: ProjectInfoCache + Send + Sync + 'static,
+{
+    cache.enable_ttl();
+    let handler = MessageHandler::new(
+        cache,
+        Arc::new(CancellationHandler::<()>::new(
+            cancel_map,
+            crate::metrics::CancellationSource::FromRedis,
+        )),
+        region_id,
+    );
+
     loop {
-        if cancellation_token.is_cancelled() {
-            return Ok(());
-        }
         let mut conn = match try_connect(&redis).await {
             Ok(conn) => {
-                handler.increment_active_listeners().await;
+                handler.disable_ttl();
                 conn
             }
             Err(e) => {
                 tracing::error!(
-            "failed to connect to redis: {e}, will try to reconnect in {RECONNECT_TIMEOUT:#?}"
-        );
+                    "failed to connect to redis: {e}, will try to reconnect in {RECONNECT_TIMEOUT:#?}"
+                );
                 tokio::time::sleep(RECONNECT_TIMEOUT).await;
                 continue;
             }
@@ -224,47 +226,8 @@ async fn handle_messages<C: ProjectInfoCache + Send + Sync + 'static>(
                     break;
                 }
             }
-            if cancellation_token.is_cancelled() {
-                handler.decrement_active_listeners().await;
-                return Ok(());
-            }
         }
-        handler.decrement_active_listeners().await;
-    }
-}
-
-/// Handle console's invalidation messages.
-#[tracing::instrument(name = "redis_notifications", skip_all)]
-pub async fn task_main<C>(
-    redis: ConnectionWithCredentialsProvider,
-    cache: Arc<C>,
-    cancel_map: CancelMap,
-    region_id: String,
-) -> anyhow::Result<Infallible>
-where
-    C: ProjectInfoCache + Send + Sync + 'static,
-{
-    let cancellation_handler = Arc::new(CancellationHandler::<()>::new(
-        cancel_map,
-        crate::metrics::CancellationSource::FromRedis,
-    ));
-    let handler = MessageHandler::new(cache, cancellation_handler, region_id);
-    // 6h - 1m.
-    // There will be 1 minute overlap between two tasks. But at least we can be sure that no message is lost.
-    let mut interval = tokio::time::interval(std::time::Duration::from_secs(6 * 60 * 60 - 60));
-    loop {
-        let cancellation_token = CancellationToken::new();
-        interval.tick().await;
-
-        tokio::spawn(handle_messages(
-            handler.clone(),
-            redis.clone(),
-            cancellation_token.clone(),
-        ));
-        tokio::spawn(async move {
-            tokio::time::sleep(std::time::Duration::from_secs(6 * 60 * 60)).await; // 6h.
-            cancellation_token.cancel();
-        });
+        handler.enable_ttl();
     }
 }
 

proxy/src/serverless/backend.rs

@@ -9,13 +9,11 @@ use crate::{
     config::{AuthenticationConfig, ProxyConfig},
     console::{
         errors::{GetAuthInfoError, WakeComputeError},
-        locks::ApiLocks,
         CachedNodeInfo,
     },
     context::RequestMonitoring,
     error::{ErrorKind, ReportableError, UserFacingError},
-    proxy::{connect_compute::ConnectMechanism, retry::ShouldRetry},
-    Host,
+    proxy::connect_compute::ConnectMechanism,
 };
 
 use super::conn_pool::{poll_client, Client, ConnInfo, GlobalConnPool};
@@ -107,7 +105,6 @@ impl PoolingBackend {
                 conn_id,
                 conn_info,
                 pool: self.pool.clone(),
-                locks: &self.config.connect_compute_locks,
             },
             &backend,
             false, // do not allow self signed compute for http flow
@@ -157,31 +154,16 @@ impl UserFacingError for HttpConnError {
     }
 }
 
-impl ShouldRetry for HttpConnError {
-    fn could_retry(&self) -> bool {
-        match self {
-            HttpConnError::ConnectionError(e) => e.could_retry(),
-            HttpConnError::ConnectionClosedAbruptly(_) => false,
-            HttpConnError::GetAuthInfo(_) => false,
-            HttpConnError::AuthError(_) => false,
-            HttpConnError::WakeCompute(_) => false,
-        }
-    }
-}
-
 struct TokioMechanism {
     pool: Arc<GlobalConnPool<tokio_postgres::Client>>,
     conn_info: ConnInfo,
     conn_id: uuid::Uuid,
-
-    /// connect_to_compute concurrency lock
-    locks: &'static ApiLocks<Host>,
 }
 
 #[async_trait]
 impl ConnectMechanism for TokioMechanism {
     type Connection = Client<tokio_postgres::Client>;
-    type ConnectError = HttpConnError;
+    type ConnectError = tokio_postgres::Error;
     type Error = HttpConnError;
 
     async fn connect_once(
@@ -190,9 +172,6 @@ impl ConnectMechanism for TokioMechanism {
         node_info: &CachedNodeInfo,
         timeout: Duration,
     ) -> Result<Self::Connection, Self::ConnectError> {
-        let host = node_info.config.get_host()?;
-        let permit = self.locks.get_permit(&host).await?;
-
         let mut config = (*node_info.config).clone();
         let config = config
             .user(&self.conn_info.user_info.user)
@@ -203,7 +182,6 @@ impl ConnectMechanism for TokioMechanism {
         let pause = ctx.latency_timer.pause(crate::metrics::Waiting::Compute);
         let (client, connection) = config.connect(tokio_postgres::NoTls).await?;
         drop(pause);
-        drop(permit);
 
         tracing::Span::current().record("pid", &tracing::field::display(client.get_process_id()));
         Ok(poll_client(

rust-toolchain.toml

@@ -1,5 +1,5 @@
 [toolchain]
-channel = "1.78.0"
+channel = "1.77.0"
 profile = "default"
 # The default profile includes rustc, rust-std, cargo, rust-docs, rustfmt and clippy.
 # https://rust-lang.github.io/rustup/concepts/profiles.html

s3_scrubber/Cargo.toml

@@ -40,6 +40,7 @@ aws-config = { workspace = true, default-features = false, features = ["rustls",
 pageserver = { path = "../pageserver" }
 pageserver_api = { path = "../libs/pageserver_api" }
 remote_storage = { path = "../libs/remote_storage" }
+safekeeper = { path = "../safekeeper" }
 
 tracing.workspace = true
 tracing-subscriber.workspace = true

s3_scrubber/src/lib.rs

@@ -7,6 +7,7 @@ pub mod metadata_stream;
 pub mod scan_pageserver_metadata;
 pub mod scan_safekeeper_metadata;
 pub mod tenant_snapshot;
+pub mod validate_safekeeper_timeline;
 
 use std::env;
 use std::fmt::Display;
@@ -36,7 +37,7 @@ use tracing::error;
 use tracing_appender::non_blocking::WorkerGuard;
 use tracing_subscriber::{fmt, prelude::*, EnvFilter};
 use utils::fs_ext;
-use utils::id::{TenantId, TimelineId};
+use utils::id::{TenantId, TimelineId, TenantTimelineId};
 
 const MAX_RETRIES: usize = 20;
 const CLOUD_ADMIN_API_TOKEN_ENV_VAR: &str = "CLOUD_ADMIN_API_TOKEN";
@@ -184,6 +185,12 @@ impl RootTarget {
             .with_sub_segment(&id.timeline_id.to_string())
     }
 
+    pub fn safekeeper_timeline_root(&self, id: &TenantTimelineId) -> S3Target {
+        self.tenants_root()
+            .with_sub_segment(&id.tenant_id.to_string())
+            .with_sub_segment(&id.timeline_id.to_string())
+    }
+
     pub fn bucket_name(&self) -> &str {
         match self {
             Self::Pageserver(root) => &root.bucket_name,
@@ -256,6 +263,25 @@ impl ConsoleConfig {
     }
 }
 
+pub struct SafekeeperApiConfig {
+    pub token: String,
+    pub base_url: Url,
+}
+
+impl SafekeeperApiConfig {
+    pub fn from_env() -> anyhow::Result<Self> {
+        let base_url: Url = env::var("SAFEKEEPER_API_URL")
+            .context("'SAFEKEEPER_API_URL' param retrieval")?
+            .parse()
+            .context("'SAFEKEEPER_API_URL' param parsing")?;
+
+        let token = env::var("SAFEKEEPER_API_TOKEN")
+            .context("'SAFEKEEPER_API_TOKEN' environment variable fetch")?;
+
+        Ok(Self { base_url, token })
+    }
+}
+
 pub fn init_logging(file_name: &str) -> WorkerGuard {
     let (file_writer, guard) =
         tracing_appender::non_blocking(tracing_appender::rolling::never("./logs/", file_name));
@@ -266,6 +292,7 @@ pub fn init_logging(file_name: &str) -> WorkerGuard {
         .with_writer(file_writer);
     let stderr_logs = fmt::Layer::new()
         .with_target(false)
+        .with_ansi(false)
         .with_writer(std::io::stderr);
     tracing_subscriber::registry()
         .with(EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")))

s3_scrubber/src/main.rs

@@ -4,6 +4,7 @@ use pageserver_api::shard::TenantShardId;
 use s3_scrubber::garbage::{find_garbage, purge_garbage, PurgeMode};
 use s3_scrubber::scan_pageserver_metadata::scan_metadata;
 use s3_scrubber::tenant_snapshot::SnapshotDownloader;
+use s3_scrubber::validate_safekeeper_timeline::{fix_local_start_lsn, validate_timelines};
 use s3_scrubber::{
     init_logging, scan_safekeeper_metadata::scan_safekeeper_metadata, BucketConfig, ConsoleConfig,
     NodeKind, TraversingDepth,
@@ -62,6 +63,20 @@ enum Command {
         #[arg(short, long)]
         output_path: Utf8PathBuf,
     },
+    #[command(verbatim_doc_comment)]
+    ValidateTimelines {
+        // points to db with debug dump
+        dump_db_connstr: String,
+        // table in the db with debug dump
+        dump_db_table: String,
+        // validation script will be written to this file
+        script_file: String,
+    },
+    #[command(verbatim_doc_comment)]
+    FixLocalStartLsn {
+        // file with timelines to fix in "<tenant>/<timeline>" format, on each line
+        tli_list_file: String,
+    },
 }
 
 #[tokio::main]
@@ -75,6 +90,8 @@ async fn main() -> anyhow::Result<()> {
         Command::FindGarbage { .. } => "find-garbage",
         Command::PurgeGarbage { .. } => "purge-garbage",
         Command::TenantSnapshot { .. } => "tenant-snapshot",
+        Command::ValidateTimelines { .. } => "validate-timelines",
+        Command::FixLocalStartLsn { .. } => "fix-local-start-lsn",
     };
     let _guard = init_logging(&format!(
         "{}_{}_{}_{}.log",
@@ -178,5 +195,25 @@ async fn main() -> anyhow::Result<()> {
                 SnapshotDownloader::new(bucket_config, tenant_id, output_path, concurrency)?;
             downloader.download().await
         }
+        Command::ValidateTimelines {
+            dump_db_connstr,
+            dump_db_table,
+            script_file,
+        } => {
+            validate_timelines(
+                bucket_config.clone(),
+                dump_db_connstr,
+                dump_db_table,
+                script_file,
+            )
+            .await
+        }
+        Command::FixLocalStartLsn {
+            tli_list_file,
+        } => {
+            fix_local_start_lsn(
+                tli_list_file,
+            ).await
+        }
     }
 }

s3_scrubber/src/validate_safekeeper_timeline.rs

@@ -0,0 +1,542 @@
+use std::{
+    cmp::{max, min},
+    str::FromStr,
+    sync::{Arc, Mutex}, time::Duration,
+};
+
+use aws_sdk_s3::{types::ObjectIdentifier, Client};
+use camino::Utf8Path;
+use futures::stream::{StreamExt, TryStreamExt};
+
+use postgres_ffi::{XLogFileName, PG_TLI};
+use reqwest::Url;
+
+use safekeeper::patch_control_file;
+use serde_json::json;
+use tokio::{fs::File, io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, process::Command};
+use tokio_postgres::types::PgLsn;
+use tracing::{error, info, info_span, Instrument};
+use utils::{
+    id::{TenantId, TenantTimelineId, TimelineId},
+    lsn::Lsn,
+};
+
+use crate::{
+    download_object_to_file, init_remote, metadata_stream::stream_listing, BucketConfig, NodeKind, RootTarget, SafekeeperApiConfig
+};
+
+/// Generally we should ask safekeepers, but so far we use everywhere default 16MB.
+const WAL_SEGSIZE: usize = 16 * 1024 * 1024;
+
+pub struct SafekeeperClient {
+    token: String,
+    base_url: Url,
+    http_client: reqwest::Client,
+}
+
+impl SafekeeperClient {
+    pub fn new(config: SafekeeperApiConfig) -> Self {
+        Self {
+            token: config.token,
+            base_url: config.base_url,
+            http_client: reqwest::Client::new(),
+        }
+    }
+
+    pub async fn timeline_status(
+        &self,
+        ttid: TenantTimelineId,
+    ) -> anyhow::Result<safekeeper::http::routes::TimelineStatus> {
+        // /v1/tenant/:tenant_id/timeline/:timeline_id
+        let req = self
+            .http_client
+            .get(self.append_url(format!(
+                "v1/tenant/{}/timeline/{}",
+                ttid.tenant_id, ttid.timeline_id
+            )))
+            .bearer_auth(&self.token);
+
+        let response = req.send().await?;
+        let mut response: safekeeper::http::routes::TimelineStatus = response.json().await?;
+
+        // this field is noisy
+        response.acceptor_state.term_history.clear();
+
+        Ok(response)
+    }
+
+    pub async fn patch_control_file(
+        &self,
+        ttid: TenantTimelineId,
+        request: patch_control_file::Request,
+    ) -> anyhow::Result<patch_control_file::Response> {
+        // /v1/tenant/:tenant_id/timeline/:timeline_id/control_file
+
+        let req = self
+            .http_client
+            .patch(self.append_url(format!(
+                "v1/tenant/{}/timeline/{}/control_file",
+                ttid.tenant_id, ttid.timeline_id,
+            )))
+            .json(&request)
+            .bearer_auth(&self.token);
+
+        let response = req.send().await?;
+        let mut response: patch_control_file::Response = response.json().await?;
+
+        // this field is noisy
+        response.old_control_file.acceptor_state.term_history.0.clear();
+        response.new_control_file.acceptor_state.term_history.0.clear();
+
+        Ok(response)
+    }
+
+    fn append_url(&self, subpath: String) -> Url {
+        // TODO fugly, but `.join` does not work when called
+        (self.base_url.to_string() + &subpath)
+            .parse()
+            .unwrap_or_else(|e| panic!("Could not append {subpath} to base url: {e}"))
+    }
+}
+
+#[derive(Debug)]
+struct DumpedTimeline {
+    ttid: TenantTimelineId,
+    timeline_start_lsn: Lsn,
+    local_start_lsn: Lsn,
+    backup_lsn: Lsn,
+    sk_id: u64,
+}
+
+fn parse_commit_lsn(formatted_string: &str) -> Option<Lsn> {
+    // Split the string by the underscore '_'
+    let parts: Vec<&str> = formatted_string.split('_').collect();
+    
+    // The fourth element (index 3) is `commit_lsn.0`, formatted as a hexadecimal string
+    if parts.len() > 3 {
+        Some(Lsn(u64::from_str_radix(parts[3], 16).ok()?))
+    } else {
+        None
+    }
+}
+
+/// Check a single safekeeper timeline with local_start_lsn != timeline_start_lsn.
+/// The function will find the segment at local_start_lsn and will try to generate a script to verify it.
+/// 
+/// If backup_lsn > local_start_lsn, this segment is no longer available locally and we have only full segment in S3.
+/// In this case, segment will be downloaded with `aws s3 cp` and checked with pg_waldump, zero exitcode means success.
+/// 
+/// Otherwise, when backup_lsn < local_start_lsn, this segment should be partial and present on 3 safekeepers.
+/// Script will download partial segment from S3 for the current safekeeper and other safekeeper. Successful validation
+/// means that partial segment on current safekeeper is equal to the partial segment on the other safekeeper.
+async fn validate_timeline(
+    s3_client: &Client,
+    root: &RootTarget,
+    tli: DumpedTimeline,
+    api_client: &SafekeeperClient,
+    shared: Arc<SharedTimelines>,
+) -> anyhow::Result<()> {
+    info!("found timeline {tli:?}");
+
+    // fetching current status from safekeeper HTTP API
+    let res = api_client.timeline_status(tli.ttid).await;
+    if res.is_err() {
+        info!("skipping, failed to fetch info about timeline: {res:?}");
+    }
+    let status = res?;
+    info!("status from sk: {status:?}");
+
+    // Path to the timeline directory in S3
+    let timeline_dir_target = root.safekeeper_timeline_root(&tli.ttid);
+
+    assert!(status.backup_lsn >= tli.backup_lsn);
+    assert!(status.timeline_start_lsn == tli.timeline_start_lsn);
+
+    if status.timeline_start_lsn == status.local_start_lsn {
+        info!("nothing to do, LSNs are equal");
+        return Ok(());
+    }
+
+    assert!(status.local_start_lsn == tli.local_start_lsn);
+    let timeline_start_lsn = status.timeline_start_lsn;
+    let local_start_lsn = status.local_start_lsn;
+
+    let segno = local_start_lsn.segment_number(WAL_SEGSIZE);
+    let segfile = XLogFileName(PG_TLI, segno, WAL_SEGSIZE);
+
+    if status.backup_lsn <= status.local_start_lsn {
+        // we have partial segments, let's find them in S3 and compare in script
+        info!("timeline without full backed up segment");
+        let mut target = timeline_dir_target;
+        target.delimiter = "".to_string();
+        target.prefix_in_bucket += &segfile;
+        let vec: Vec<ObjectIdentifier> = stream_listing(s3_client, &target).try_collect().await?;
+        info!("found partial files: {:?}", vec);
+
+        let expected_suffix = format!("_sk{}.partial", tli.sk_id);
+        let segment_of_interest = vec.iter().find(|obj| obj.key.ends_with(&expected_suffix));
+
+        let segment_of_interest = match segment_of_interest {
+            Some(seg) => seg,
+            None => {
+                info!("haven't found a partial segment, skipping");
+                return Ok(());
+            }
+        };
+
+        let partial_prefix = segment_of_interest
+            .key
+            .strip_suffix(&expected_suffix)
+            .unwrap();
+
+        let another_segment = vec.iter().find(|obj| {
+            obj.key != segment_of_interest.key
+        });
+
+        if another_segment.is_none() {
+            info!("haven't found another partial segment to compare to");
+            return Ok(());
+        }
+        let another_segment = another_segment.unwrap();
+
+        let ours_commit_lsn = parse_commit_lsn(&segment_of_interest.key);
+        if ours_commit_lsn.is_none() {
+            info!("failed to parse commit_lsn from {}", &segment_of_interest.key);
+            return Ok(());
+        }
+        let ours_commit_lsn = ours_commit_lsn.unwrap();
+
+        let their_commit_lsn = parse_commit_lsn(&another_segment.key);
+        if their_commit_lsn.is_none() {
+            info!("failed to parse commit_lsn from {}", &another_segment.key);
+            return Ok(());
+        }
+        let their_commit_lsn = their_commit_lsn.unwrap();
+
+        let min_commit_lsn = min(ours_commit_lsn, their_commit_lsn);
+        let strip_offset = min_commit_lsn.segment_offset(WAL_SEGSIZE);
+
+        info!("found ours_commit_lsn={ours_commit_lsn}, their_commit_lsn={their_commit_lsn}, min={min_commit_lsn}, offset={strip_offset}");
+
+        let ours_str = format!("./tmp/ours");
+        let ours = Utf8Path::new(&ours_str);
+        download_object_to_file(
+            s3_client,
+            root.bucket_name(),
+            &segment_of_interest.key,
+            None,
+            &ours,
+        ).await?;
+
+        let their_str = format!("./tmp/their");
+        let their = Utf8Path::new(&their_str);
+        download_object_to_file(
+            s3_client,
+            root.bucket_name(),
+            &another_segment.key,
+            None,
+            &their,
+        ).await?;
+
+        let cmd = format!(
+            "truncate --size={} ./tmp/ours && truncate --size={} ./tmp/their && cmp -l ./tmp/ours ./tmp/their",
+            strip_offset,
+            strip_offset,
+        );
+
+        // Execute the command using `bash -c`
+        let mut process = Command::new("bash")
+            .arg("-c")
+            .arg(&cmd)
+            .spawn()?;
+
+        // Await the process to finish and check the exit status
+        let status = process.wait().await?;
+
+        // let's delete the file
+        tokio::fs::remove_file(ours).await?;
+        tokio::fs::remove_file(their).await?;
+
+        if status.success() {
+            info!("Command executed successfully.");
+            shared.append_script(&[
+                format!("echo '{}' >> valid.log", tli.ttid),
+            ]);
+        } else {
+            shared.append_script(&[
+                format!("echo '{}' >> invalid.log", tli.ttid),
+            ]);
+
+            info!("Command failed with status: {}", status);
+            if let Some(code) = status.code() {
+                info!("Exit code: {}", code);
+            } else {
+                info!("Process terminated by signal");
+            }
+        }
+
+        return Ok(());
+    }
+
+    // we have only full segment, let's download it and run pg_waldump
+    let key = timeline_dir_target.prefix_in_bucket + &segfile;
+    info!(
+        "starting s3 download from bucket {}, key {}",
+        root.bucket_name(),
+        key
+    );
+
+    let segment_start_lsn = local_start_lsn.segment_lsn(WAL_SEGSIZE);
+
+    let path_str = format!("./tmp/{}", segfile);
+    let path = Utf8Path::new(&path_str);
+    download_object_to_file(
+        s3_client,
+        root.bucket_name(),
+        &key,
+        None,
+        &path,
+    ).await?;
+
+    info!("downloaded file to {}", path);
+
+    let waldump_cmd =         format!(
+        "./pg_install/v{}/bin/pg_waldump -i{} ./tmp/{}",
+        status.pg_info.pg_version / 10000,
+        if segment_start_lsn < timeline_start_lsn {
+            format!(" -s {}", timeline_start_lsn)
+        } else {
+            "".to_string()
+        },
+        segfile,
+    );
+
+    info!("running {}", waldump_cmd);
+
+    // Execute the command using `bash -c`
+    let mut process = Command::new("bash")
+        .arg("-c")
+        .arg(&waldump_cmd)
+        .spawn()?;
+
+    // Await the process to finish and check the exit status
+    let status = process.wait().await?;
+
+    // let's delete the file
+    tokio::fs::remove_file(path).await?;
+
+    if status.success() {
+        info!("Command executed successfully.");
+        shared.append_script(&[
+            format!("echo '{}' >> valid.log", tli.ttid),
+        ]);
+    } else {
+        shared.append_script(&[
+            format!("echo '{}' >> invalid.log", tli.ttid),
+        ]);
+
+        info!("Command failed with status: {}", status);
+        if let Some(code) = status.code() {
+            info!("Exit code: {}", code);
+        } else {
+            info!("Process terminated by signal");
+        }
+    }
+
+    // shared.append_script(&[
+    //     String::new(),
+    //     format!("aws s3 cp s3://{}/{} {}", root.bucket_name(), key, segfile),
+        // format!(
+        //     "/usr/local/v{}/bin/pg_waldump{} {} > /dev/null",
+        //     status.pg_info.pg_version / 10000,
+        //     if segment_start_lsn < timeline_start_lsn {
+        //         format!(" -s {}", timeline_start_lsn)
+        //     } else {
+        //         "".to_string()
+        //     },
+        //     segfile,
+        // ),
+    //     "if [ $? -ne 0 ]; then".to_string(),
+    //     format!("   echo '{}' >> invalid.log", tli.ttid),
+    //     "else".to_string(),
+    //     format!("   echo '{}' >> valid.log", tli.ttid),
+    //     "fi".to_string(),
+    //     format!("rm {}", segfile),
+    // ]);
+
+    Ok(())
+}
+
+pub struct InternalTimelines {
+    validation_script: Vec<String>,
+}
+
+// Shared memory for validation tasks, generated script will be inserted here.
+pub struct SharedTimelines {
+    data: Mutex<InternalTimelines>,
+}
+
+impl SharedTimelines {
+    fn append_script(&self, arr: &[String]) {
+        let mut lock = self.data.lock().unwrap();
+        lock.validation_script.extend_from_slice(arr);
+    }
+}
+
+pub async fn validate_timelines(
+    bucket_config: BucketConfig,
+    dump_db_connstr: String,
+    dump_db_table: String,
+    script_file: String,
+) -> anyhow::Result<()> {
+    info!(
+        "checking bucket {}, region {}, dump_db_table {}",
+        bucket_config.bucket, bucket_config.region, dump_db_table
+    );
+
+    let shared = Arc::new(SharedTimelines {
+        data: Mutex::new(InternalTimelines {
+            validation_script: vec![
+                "#!/bin/bash".to_string(),
+                "echo 'Starting the validation script'".to_string(),
+            ],
+        }),
+    });
+
+    // Use the native TLS implementation (Neon requires TLS)
+    let tls_connector =
+        postgres_native_tls::MakeTlsConnector::new(native_tls::TlsConnector::new().unwrap());
+    let (client, connection) = tokio_postgres::connect(&dump_db_connstr, tls_connector).await?;
+    // The connection object performs the actual communication with the database,
+    // so spawn it off to run on its own.
+    tokio::spawn(async move {
+        if let Err(e) = connection.await {
+            eprintln!("connection error: {}", e);
+        }
+    });
+
+    let sk_api_config = SafekeeperApiConfig::from_env()?;
+    let sk_api_client = SafekeeperClient::new(sk_api_config);
+
+    let query = format!(
+        "
+        select
+            tenant_id, timeline_id, timeline_start_lsn, local_start_lsn, backup_lsn, sk_id
+        from \"{}\"
+        where
+        timeline_start_lsn != local_start_lsn
+AND sk_id = 51
+AND timeline_id != '13a865b39537d5538e0ea74c926d9c6f'
+AND timeline_id != 'c5e944b5c13628ba8fe128b01e7e663d'
+AND timeline_id != 'e7cfa4a2bd15c88ff011d69feef4b076';",
+        dump_db_table,
+    );
+
+    info!("query is {}", query);
+    let timelines = client.query(&query, &[]).await?;
+    info!("loaded {} timelines", timelines.len());
+
+    let (s3_client, target) = init_remote(bucket_config, NodeKind::Safekeeper)?;
+
+    let checks = futures::stream::iter(timelines.iter().map(Ok)).map_ok(|row| {
+        let tenant_id = TenantId::from_str(row.get(0)).expect("failed to parse tenant_id");
+        let timeline_id = TimelineId::from_str(row.get(1)).expect("failed to parse tenant_id");
+        let timeline_start_lsn_pg: PgLsn = row.get(2);
+        let local_start_lsn_pg: PgLsn = row.get(3);
+        let backup_lsn_pg: PgLsn = row.get(4);
+        let sk_id: i64 = row.get(5);
+        let sk_id = sk_id as u64;
+
+        let ttid = TenantTimelineId::new(tenant_id, timeline_id);
+        let shared = shared.clone();
+
+        let dumped_tli = DumpedTimeline {
+            ttid,
+            timeline_start_lsn: Lsn(u64::from(timeline_start_lsn_pg)),
+            local_start_lsn: Lsn(u64::from(local_start_lsn_pg)),
+            backup_lsn: Lsn(u64::from(backup_lsn_pg)),
+            sk_id,
+        };
+
+        validate_timeline(&s3_client, &target, dumped_tli, &sk_api_client, shared)
+            .instrument(info_span!("validate", ttid=%ttid))
+    });
+
+    // Run tasks concurrently.
+    const CONCURRENCY: usize = 1;
+    let mut timelines = checks.try_buffered(CONCURRENCY);
+
+    while let Some(r) = timelines.next().await {
+        if r.is_err() {
+            error!("failed to process the timeline, error: {:?}", r);
+        }
+    }
+
+    // Save resulting script to the file.
+    let mut file = File::create(script_file).await?;
+    for line in &shared.data.lock().unwrap().validation_script {
+        file.write_all(line.as_bytes()).await?;
+        file.write_all(b"\n").await?;
+    }
+    file.flush().await?;
+    drop(file);
+
+    Ok(())
+}
+
+pub async fn fix_local_start_lsn(tli_list_file: String) -> anyhow::Result<()> {
+    let sk_api_config = SafekeeperApiConfig::from_env()?;
+    let sk_api_client = SafekeeperClient::new(sk_api_config);
+
+    let mut tlis: Vec<TenantTimelineId> = Vec::new();
+
+    let file = File::open(tli_list_file).await?;
+    let reader = BufReader::new(file);
+    let mut lines = reader.lines();
+    
+    while let Some(line) = lines.next_line().await? {
+        let ttid = TenantTimelineId::from_str(&line)?;
+        tlis.push(ttid);
+    }
+    drop(lines);
+    
+    info!("Read {} timelines from the file, starting applying new local_start_lsn", tlis.len());
+
+    for ttid in tlis {
+        let res = fix_one_timeline(&sk_api_client, ttid).await;
+        if res.is_err() {
+            error!("error while fixing timeline {}: {:?}", ttid, res);
+        }
+    }
+
+    Ok(())
+}
+
+async fn fix_one_timeline(api_client: &SafekeeperClient, ttid: TenantTimelineId) -> anyhow::Result<()> {
+    // fetching current status from safekeeper HTTP API
+    let res = api_client.timeline_status(ttid).await;
+    if res.is_err() {
+        info!("skipping, failed to fetch info about timeline: {res:?}");
+    }
+    
+    let status = res?;
+    info!("status from sk: {status:?}");
+
+    if status.timeline_start_lsn == status.local_start_lsn {
+        info!("nothing to do, LSNs are equal");
+        return Ok(());
+    }
+
+    let request = patch_control_file::Request {
+        updates: json!({
+            "local_start_lsn": status.timeline_start_lsn.to_string(),
+        }),
+        apply_fields: vec!["local_start_lsn".to_string()],
+    };
+
+    info!("sending request {:?}", request);
+    let res = api_client.patch_control_file(ttid, request).await?;
+    info!("patch finished: {:?}", res);
+
+    Ok(())
+}

safekeeper/src/patch_control_file.rs

@@ -6,7 +6,7 @@ use tracing::info;
 
 use crate::{state::TimelinePersistentState, timeline::Timeline};
 
-#[derive(Deserialize, Debug, Clone)]
+#[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct Request {
     /// JSON object with fields to update
     pub updates: serde_json::Value,
@@ -14,7 +14,7 @@ pub struct Request {
     pub apply_fields: Vec<String>,
 }
 
-#[derive(Serialize)]
+#[derive(Serialize, Deserialize, Debug)]
 pub struct Response {
     pub old_control_file: TimelinePersistentState,
     pub new_control_file: TimelinePersistentState,

storage_controller/Cargo.toml

@@ -40,8 +40,6 @@ tokio.workspace = true
 tokio-util.workspace = true
 tracing.workspace = true
 measured.workspace = true
-strum.workspace = true
-strum_macros.workspace = true
 
 diesel = { version = "2.1.4", features = ["serde_json", "postgres", "r2d2"] }
 diesel_migrations = { version = "2.1.0" }

storage_controller/src/http.rs

@@ -912,7 +912,7 @@ pub fn make_router(
                 RequestName("v1_tenant_timeline"),
             )
         })
-        // Tenant detail GET passthrough to shard zero:
+        // Tenant detail GET passthrough to shard zero
         .get("/v1/tenant/:tenant_id", |r| {
             tenant_service_handler(
                 r,
@@ -920,14 +920,13 @@ pub fn make_router(
                 RequestName("v1_tenant_passthrough"),
             )
         })
-        // The `*` in the  URL is a wildcard: any tenant/timeline GET APIs on the pageserver
-        // are implicitly exposed here.  This must be last in the list to avoid
-        // taking precedence over other GET methods we might implement by hand.
-        .get("/v1/tenant/:tenant_id/*", |r| {
+        // Timeline GET passthrough to shard zero.  Note that the `*` in the URL is a wildcard: any future
+        // timeline GET APIs will be implicitly included.
+        .get("/v1/tenant/:tenant_id/timeline*", |r| {
             tenant_service_handler(
                 r,
                 handle_tenant_timeline_passthrough,
-                RequestName("v1_tenant_passthrough"),
+                RequestName("v1_tenant_timeline_passthrough"),
             )
         })
 }

storage_controller/src/id_lock_map.rs

@@ -1,64 +1,25 @@
-use std::fmt::Display;
-use std::time::Instant;
 use std::{collections::HashMap, sync::Arc};
 
-use std::time::Duration;
-
-use crate::service::RECONCILE_TIMEOUT;
-
-const LOCK_TIMEOUT_ALERT_THRESHOLD: Duration = RECONCILE_TIMEOUT;
-
-/// A wrapper around `OwnedRwLockWriteGuard` that when dropped changes the
-/// current holding operation in lock.
-pub struct WrappedWriteGuard<T: Display> {
-    guard: tokio::sync::OwnedRwLockWriteGuard<Option<T>>,
-    start: Instant,
-}
-
-impl<T: Display> WrappedWriteGuard<T> {
-    pub fn new(guard: tokio::sync::OwnedRwLockWriteGuard<Option<T>>) -> Self {
-        Self {
-            guard,
-            start: Instant::now(),
-        }
-    }
-}
-
-impl<T: Display> Drop for WrappedWriteGuard<T> {
-    fn drop(&mut self) {
-        let duration = self.start.elapsed();
-        if duration > LOCK_TIMEOUT_ALERT_THRESHOLD {
-            tracing::warn!(
-                "Lock on {} was held for {:?}",
-                self.guard.as_ref().unwrap(),
-                duration
-            );
-        }
-        *self.guard = None;
-    }
-}
-
 /// A map of locks covering some arbitrary identifiers. Useful if you have a collection of objects but don't
 /// want to embed a lock in each one, or if your locking granularity is different to your object granularity.
 /// For example, used in the storage controller where the objects are tenant shards, but sometimes locking
 /// is needed at a tenant-wide granularity.
-pub(crate) struct IdLockMap<T, I>
+pub(crate) struct IdLockMap<T>
 where
     T: Eq + PartialEq + std::hash::Hash,
 {
     /// A synchronous lock for getting/setting the async locks that our callers will wait on.
-    entities: std::sync::Mutex<std::collections::HashMap<T, Arc<tokio::sync::RwLock<Option<I>>>>>,
+    entities: std::sync::Mutex<std::collections::HashMap<T, Arc<tokio::sync::RwLock<()>>>>,
 }
 
-impl<T, I> IdLockMap<T, I>
+impl<T> IdLockMap<T>
 where
     T: Eq + PartialEq + std::hash::Hash,
-    I: Display,
 {
     pub(crate) fn shared(
         &self,
         key: T,
-    ) -> impl std::future::Future<Output = tokio::sync::OwnedRwLockReadGuard<Option<I>>> {
+    ) -> impl std::future::Future<Output = tokio::sync::OwnedRwLockReadGuard<()>> {
         let mut locked = self.entities.lock().unwrap();
         let entry = locked.entry(key).or_default();
         entry.clone().read_owned()
@@ -67,26 +28,21 @@ where
     pub(crate) fn exclusive(
         &self,
         key: T,
-        operation: I,
-    ) -> impl std::future::Future<Output = WrappedWriteGuard<I>> {
+    ) -> impl std::future::Future<Output = tokio::sync::OwnedRwLockWriteGuard<()>> {
         let mut locked = self.entities.lock().unwrap();
-        let entry = locked.entry(key).or_default().clone();
-        async move {
-            let mut guard = WrappedWriteGuard::new(entry.clone().write_owned().await);
-            *guard.guard = Some(operation);
-            guard
-        }
+        let entry = locked.entry(key).or_default();
+        entry.clone().write_owned()
     }
 
     /// Rather than building a lock guard that re-takes the [`Self::entities`] lock, we just do
     /// periodic housekeeping to avoid the map growing indefinitely
     pub(crate) fn housekeeping(&self) {
         let mut locked = self.entities.lock().unwrap();
-        locked.retain(|_k, entry| entry.try_write().is_err())
+        locked.retain(|_k, lock| lock.try_write().is_err())
     }
 }
 
-impl<T, I> Default for IdLockMap<T, I>
+impl<T> Default for IdLockMap<T>
 where
     T: Eq + PartialEq + std::hash::Hash,
 {
@@ -96,94 +52,3 @@ where
         }
     }
 }
-
-pub async fn trace_exclusive_lock<
-    T: Clone + Display + Eq + PartialEq + std::hash::Hash,
-    I: Display + Clone,
->(
-    op_locks: &IdLockMap<T, I>,
-    key: T,
-    operation: I,
-) -> WrappedWriteGuard<I> {
-    let start = Instant::now();
-    let guard = op_locks.exclusive(key.clone(), operation.clone()).await;
-
-    let duration = start.elapsed();
-    if duration > LOCK_TIMEOUT_ALERT_THRESHOLD {
-        tracing::warn!(
-            "Operation {} on key {} has waited {:?} for exclusive lock",
-            operation,
-            key,
-            duration
-        );
-    }
-
-    guard
-}
-
-pub async fn trace_shared_lock<
-    T: Clone + Display + Eq + PartialEq + std::hash::Hash,
-    I: Display,
->(
-    op_locks: &IdLockMap<T, I>,
-    key: T,
-    operation: I,
-) -> tokio::sync::OwnedRwLockReadGuard<Option<I>> {
-    let start = Instant::now();
-    let guard = op_locks.shared(key.clone()).await;
-
-    let duration = start.elapsed();
-    if duration > LOCK_TIMEOUT_ALERT_THRESHOLD {
-        tracing::warn!(
-            "Operation {} on key {} has waited {:?} for shared lock",
-            operation,
-            key,
-            duration
-        );
-    }
-
-    guard
-}
-
-#[cfg(test)]
-mod tests {
-    use super::IdLockMap;
-
-    #[derive(Clone, Debug, strum_macros::Display, PartialEq)]
-    enum Operations {
-        Op1,
-        Op2,
-    }
-
-    #[tokio::test]
-    async fn multiple_shared_locks() {
-        let id_lock_map: IdLockMap<i32, Operations> = IdLockMap::default();
-
-        let shared_lock_1 = id_lock_map.shared(1).await;
-        let shared_lock_2 = id_lock_map.shared(1).await;
-
-        assert!(shared_lock_1.is_none());
-        assert!(shared_lock_2.is_none());
-    }
-
-    #[tokio::test]
-    async fn exclusive_locks() {
-        let id_lock_map = IdLockMap::default();
-        let resource_id = 1;
-
-        {
-            let _ex_lock = id_lock_map.exclusive(resource_id, Operations::Op1).await;
-            assert_eq!(_ex_lock.guard.clone().unwrap(), Operations::Op1);
-
-            let _ex_lock_2 = tokio::time::timeout(
-                tokio::time::Duration::from_millis(1),
-                id_lock_map.exclusive(resource_id, Operations::Op2),
-            )
-            .await;
-            assert!(_ex_lock_2.is_err());
-        }
-
-        let shared_lock_1 = id_lock_map.shared(resource_id).await;
-        assert!(shared_lock_1.is_none());
-    }
-}

storage_controller/src/pageserver_client.rs

@@ -114,27 +114,6 @@ impl PageserverClient {
         )
     }
 
-    pub(crate) async fn tenant_secondary_status(
-        &self,
-        tenant_shard_id: TenantShardId,
-    ) -> Result<SecondaryProgress> {
-        measured_request!(
-            "tenant_secondary_status",
-            crate::metrics::Method::Get,
-            &self.node_id_label,
-            self.inner.tenant_secondary_status(tenant_shard_id).await
-        )
-    }
-
-    pub(crate) async fn tenant_heatmap_upload(&self, tenant_id: TenantShardId) -> Result<()> {
-        measured_request!(
-            "tenant_heatmap_upload",
-            crate::metrics::Method::Post,
-            &self.node_id_label,
-            self.inner.tenant_heatmap_upload(tenant_id).await
-        )
-    }
-
     pub(crate) async fn location_config(
         &self,
         tenant_shard_id: TenantShardId,

storage_controller/src/service.rs

@@ -9,13 +9,11 @@ use std::{
 
 use crate::{
     compute_hook::NotifyError,
-    id_lock_map::{trace_exclusive_lock, trace_shared_lock, IdLockMap, WrappedWriteGuard},
+    id_lock_map::IdLockMap,
     persistence::{AbortShardSplitStatus, TenantFilter},
     reconciler::{ReconcileError, ReconcileUnits},
     scheduler::{ScheduleContext, ScheduleMode},
-    tenant_shard::{
-        MigrateAttachment, ReconcileNeeded, ScheduleOptimization, ScheduleOptimizationAction,
-    },
+    tenant_shard::ReconcileNeeded,
 };
 use anyhow::Context;
 use control_plane::storage_controller::{
@@ -35,7 +33,6 @@ use pageserver_api::{
     models::{SecondaryProgress, TenantConfigRequest},
 };
 use reqwest::StatusCode;
-use tracing::instrument;
 
 use crate::pageserver_client::PageserverClient;
 use pageserver_api::{
@@ -53,11 +50,11 @@ use pageserver_api::{
     },
 };
 use pageserver_client::mgmt_api;
-use tokio::sync::mpsc::error::TrySendError;
+use tokio::sync::{mpsc::error::TrySendError, OwnedRwLockWriteGuard};
 use tokio_util::sync::CancellationToken;
+use tracing::instrument;
 use utils::{
     completion::Barrier,
-    failpoint_support,
     generation::Generation,
     http::error::ApiError,
     id::{NodeId, TenantId, TimelineId},
@@ -82,7 +79,7 @@ const SHORT_RECONCILE_TIMEOUT: Duration = Duration::from_secs(5);
 
 // For operations that might be slow, like migrating a tenant with
 // some data in it.
-pub const RECONCILE_TIMEOUT: Duration = Duration::from_secs(30);
+const RECONCILE_TIMEOUT: Duration = Duration::from_secs(30);
 
 // If we receive a call using Secondary mode initially, it will omit generation.  We will initialize
 // tenant shards into this generation, and as long as it remains in this generation, we will accept
@@ -99,26 +96,6 @@ pub(crate) const STARTUP_RECONCILE_TIMEOUT: Duration = Duration::from_secs(30);
 /// (`<https://github.com/neondatabase/neon/issues/7552>`)
 pub const MAX_UNAVAILABLE_INTERVAL_DEFAULT: Duration = Duration::from_secs(300);
 
-#[derive(Clone, strum_macros::Display)]
-enum TenantOperations {
-    Create,
-    LocationConfig,
-    ConfigSet,
-    TimeTravelRemoteStorage,
-    Delete,
-    UpdatePolicy,
-    ShardSplit,
-    SecondaryDownload,
-    TimelineCreate,
-    TimelineDelete,
-}
-
-#[derive(Clone, strum_macros::Display)]
-enum NodeOperations {
-    Register,
-    Configure,
-}
-
 pub const RECONCILER_CONCURRENCY_DEFAULT: usize = 128;
 
 // Depth of the channel used to enqueue shards for reconciliation when they can't do it immediately.
@@ -254,11 +231,11 @@ pub struct Service {
     // Locking on a tenant granularity (covers all shards in the tenant):
     // - Take exclusively for rare operations that mutate the tenant's persistent state (e.g. create/delete/split)
     // - Take in shared mode for operations that need the set of shards to stay the same to complete reliably (e.g. timeline CRUD)
-    tenant_op_locks: IdLockMap<TenantId, TenantOperations>,
+    tenant_op_locks: IdLockMap<TenantId>,
 
     // Locking for node-mutating operations: take exclusively for operations that modify the node's persistent state, or
     // that transition it to/from Active.
-    node_op_locks: IdLockMap<NodeId, NodeOperations>,
+    node_op_locks: IdLockMap<NodeId>,
 
     // Limit how many Reconcilers we will spawn concurrently
     reconciler_concurrency: Arc<tokio::sync::Semaphore>,
@@ -330,7 +307,7 @@ struct TenantShardSplitAbort {
     new_shard_count: ShardCount,
     new_stripe_size: Option<ShardStripeSize>,
     /// Until this abort op is complete, no other operations may be done on the tenant
-    _tenant_lock: WrappedWriteGuard<TenantOperations>,
+    _tenant_lock: tokio::sync::OwnedRwLockWriteGuard<()>,
 }
 
 #[derive(thiserror::Error, Debug)]
@@ -711,7 +688,7 @@ impl Service {
                 let reconciles_spawned = self.reconcile_all();
                 if reconciles_spawned == 0 {
                     // Run optimizer only when we didn't find any other work to do
-                    self.optimize_all().await;
+                    self.optimize_all();
                 }
             }
               _ = self.cancel.cancelled() => return
@@ -1363,7 +1340,7 @@ impl Service {
     async fn node_activate_reconcile(
         &self,
         mut node: Node,
-        _lock: &WrappedWriteGuard<NodeOperations>,
+        _lock: &OwnedRwLockWriteGuard<()>,
     ) -> Result<(), ApiError> {
         // This Node is a mutable local copy: we will set it active so that we can use its
         // API client to reconcile with the node.  The Node in [`Self::nodes`] will get updated
@@ -1609,12 +1586,11 @@ impl Service {
         let tenant_id = create_req.new_tenant_id.tenant_id;
 
         // Exclude any concurrent attempts to create/access the same tenant ID
-        let _tenant_lock = trace_exclusive_lock(
-            &self.tenant_op_locks,
-            create_req.new_tenant_id.tenant_id,
-            TenantOperations::Create,
-        )
-        .await;
+        let _tenant_lock = self
+            .tenant_op_locks
+            .exclusive(create_req.new_tenant_id.tenant_id)
+            .await;
+
         let (response, waiters) = self.do_tenant_create(create_req).await?;
 
         if let Err(e) = self.await_waiters(waiters, RECONCILE_TIMEOUT).await {
@@ -1953,12 +1929,10 @@ impl Service {
         req: TenantLocationConfigRequest,
     ) -> Result<TenantLocationConfigResponse, ApiError> {
         // We require an exclusive lock, because we are updating both persistent and in-memory state
-        let _tenant_lock = trace_exclusive_lock(
-            &self.tenant_op_locks,
-            tenant_shard_id.tenant_id,
-            TenantOperations::LocationConfig,
-        )
-        .await;
+        let _tenant_lock = self
+            .tenant_op_locks
+            .exclusive(tenant_shard_id.tenant_id)
+            .await;
 
         if !tenant_shard_id.is_unsharded() {
             return Err(ApiError::BadRequest(anyhow::anyhow!(
@@ -2076,12 +2050,7 @@ impl Service {
 
     pub(crate) async fn tenant_config_set(&self, req: TenantConfigRequest) -> Result<(), ApiError> {
         // We require an exclusive lock, because we are updating persistent and in-memory state
-        let _tenant_lock = trace_exclusive_lock(
-            &self.tenant_op_locks,
-            req.tenant_id,
-            TenantOperations::ConfigSet,
-        )
-        .await;
+        let _tenant_lock = self.tenant_op_locks.exclusive(req.tenant_id).await;
 
         let tenant_id = req.tenant_id;
         let config = req.config;
@@ -2170,12 +2139,7 @@ impl Service {
         timestamp: Cow<'_, str>,
         done_if_after: Cow<'_, str>,
     ) -> Result<(), ApiError> {
-        let _tenant_lock = trace_exclusive_lock(
-            &self.tenant_op_locks,
-            tenant_id,
-            TenantOperations::TimeTravelRemoteStorage,
-        )
-        .await;
+        let _tenant_lock = self.tenant_op_locks.exclusive(tenant_id).await;
 
         let node = {
             let locked = self.inner.read().unwrap();
@@ -2266,12 +2230,7 @@ impl Service {
         tenant_id: TenantId,
         wait: Option<Duration>,
     ) -> Result<(StatusCode, SecondaryProgress), ApiError> {
-        let _tenant_lock = trace_shared_lock(
-            &self.tenant_op_locks,
-            tenant_id,
-            TenantOperations::SecondaryDownload,
-        )
-        .await;
+        let _tenant_lock = self.tenant_op_locks.shared(tenant_id).await;
 
         // Acquire lock and yield the collection of shard-node tuples which we will send requests onward to
         let targets = {
@@ -2365,8 +2324,7 @@ impl Service {
     }
 
     pub(crate) async fn tenant_delete(&self, tenant_id: TenantId) -> Result<StatusCode, ApiError> {
-        let _tenant_lock =
-            trace_exclusive_lock(&self.tenant_op_locks, tenant_id, TenantOperations::Delete).await;
+        let _tenant_lock = self.tenant_op_locks.exclusive(tenant_id).await;
 
         self.ensure_attached_wait(tenant_id).await?;
 
@@ -2466,14 +2424,7 @@ impl Service {
         req: TenantPolicyRequest,
     ) -> Result<(), ApiError> {
         // We require an exclusive lock, because we are updating persistent and in-memory state
-        let _tenant_lock = trace_exclusive_lock(
-            &self.tenant_op_locks,
-            tenant_id,
-            TenantOperations::UpdatePolicy,
-        )
-        .await;
-
-        failpoint_support::sleep_millis_async!("tenant-update-policy-exclusive-lock");
+        let _tenant_lock = self.tenant_op_locks.exclusive(tenant_id).await;
 
         let TenantPolicyRequest {
             placement,
@@ -2527,12 +2478,7 @@ impl Service {
             create_req.new_timeline_id,
         );
 
-        let _tenant_lock = trace_shared_lock(
-            &self.tenant_op_locks,
-            tenant_id,
-            TenantOperations::TimelineCreate,
-        )
-        .await;
+        let _tenant_lock = self.tenant_op_locks.shared(tenant_id).await;
 
         self.ensure_attached_wait(tenant_id).await?;
 
@@ -2641,57 +2587,13 @@ impl Service {
         Ok(results)
     }
 
-    /// Concurrently invoke a pageserver API call on many shards at once
-    pub(crate) async fn tenant_for_shards_api<T, O, F>(
-        &self,
-        locations: Vec<(TenantShardId, Node)>,
-        op: O,
-        warn_threshold: u32,
-        max_retries: u32,
-        timeout: Duration,
-        cancel: &CancellationToken,
-    ) -> Vec<mgmt_api::Result<T>>
-    where
-        O: Fn(TenantShardId, PageserverClient) -> F + Copy,
-        F: std::future::Future<Output = mgmt_api::Result<T>>,
-    {
-        let mut futs = FuturesUnordered::new();
-        let mut results = Vec::with_capacity(locations.len());
-
-        for (tenant_shard_id, node) in locations {
-            futs.push(async move {
-                node.with_client_retries(
-                    |client| op(tenant_shard_id, client),
-                    &self.config.jwt_token,
-                    warn_threshold,
-                    max_retries,
-                    timeout,
-                    cancel,
-                )
-                .await
-            });
-        }
-
-        while let Some(r) = futs.next().await {
-            let r = r.unwrap_or(Err(mgmt_api::Error::Cancelled));
-            results.push(r);
-        }
-
-        results
-    }
-
     pub(crate) async fn tenant_timeline_delete(
         &self,
         tenant_id: TenantId,
         timeline_id: TimelineId,
     ) -> Result<StatusCode, ApiError> {
         tracing::info!("Deleting timeline {}/{}", tenant_id, timeline_id,);
-        let _tenant_lock = trace_shared_lock(
-            &self.tenant_op_locks,
-            tenant_id,
-            TenantOperations::TimelineDelete,
-        )
-        .await;
+        let _tenant_lock = self.tenant_op_locks.shared(tenant_id).await;
 
         self.ensure_attached_wait(tenant_id).await?;
 
@@ -3129,14 +3031,11 @@ impl Service {
     ) -> (
         TenantShardSplitResponse,
         Vec<(TenantShardId, NodeId, ShardStripeSize)>,
-        Vec<ReconcilerWaiter>,
     ) {
         let mut response = TenantShardSplitResponse {
             new_shards: Vec::new(),
         };
         let mut child_locations = Vec::new();
-        let mut waiters = Vec::new();
-
         {
             let mut locked = self.inner.write().unwrap();
 
@@ -3215,112 +3114,14 @@ impl Service {
                         tracing::warn!("Failed to schedule child shard {child}: {e}");
                     }
                     // In the background, attach secondary locations for the new shards
-                    if let Some(waiter) = self.maybe_reconcile_shard(&mut child_state, nodes) {
-                        waiters.push(waiter);
-                    }
+                    self.maybe_reconcile_shard(&mut child_state, nodes);
 
                     tenants.insert(child, child_state);
                     response.new_shards.push(child);
                 }
             }
-            (response, child_locations, waiters)
-        }
-    }
 
-    async fn tenant_shard_split_start_secondaries(
-        &self,
-        tenant_id: TenantId,
-        waiters: Vec<ReconcilerWaiter>,
-    ) {
-        // Wait for initial reconcile of child shards, this creates the secondary locations
-        if let Err(e) = self.await_waiters(waiters, RECONCILE_TIMEOUT).await {
-            // This is not a failure to split: it's some issue reconciling the new child shards, perhaps
-            // their secondaries couldn't be attached.
-            tracing::warn!("Failed to reconcile after split: {e}");
-            return;
-        }
-
-        // Take the state lock to discover the attached & secondary intents for all shards
-        let (attached, secondary) = {
-            let locked = self.inner.read().unwrap();
-            let mut attached = Vec::new();
-            let mut secondary = Vec::new();
-
-            for (tenant_shard_id, shard) in
-                locked.tenants.range(TenantShardId::tenant_range(tenant_id))
-            {
-                let Some(node_id) = shard.intent.get_attached() else {
-                    // Unexpected.  Race with a PlacementPolicy change?
-                    tracing::warn!(
-                        "No attached node on {tenant_shard_id} immediately after shard split!"
-                    );
-                    continue;
-                };
-
-                let Some(secondary_node_id) = shard.intent.get_secondary().first() else {
-                    // No secondary location.  Nothing for us to do.
-                    continue;
-                };
-
-                let attached_node = locked
-                    .nodes
-                    .get(node_id)
-                    .expect("Pageservers may not be deleted while referenced");
-
-                let secondary_node = locked
-                    .nodes
-                    .get(secondary_node_id)
-                    .expect("Pageservers may not be deleted while referenced");
-
-                attached.push((*tenant_shard_id, attached_node.clone()));
-                secondary.push((*tenant_shard_id, secondary_node.clone()));
-            }
-            (attached, secondary)
-        };
-
-        if secondary.is_empty() {
-            // No secondary locations; nothing for us to do
-            return;
-        }
-
-        for result in self
-            .tenant_for_shards_api(
-                attached,
-                |tenant_shard_id, client| async move {
-                    client.tenant_heatmap_upload(tenant_shard_id).await
-                },
-                1,
-                1,
-                SHORT_RECONCILE_TIMEOUT,
-                &self.cancel,
-            )
-            .await
-        {
-            if let Err(e) = result {
-                tracing::warn!("Error calling heatmap upload after shard split: {e}");
-                return;
-            }
-        }
-
-        for result in self
-            .tenant_for_shards_api(
-                secondary,
-                |tenant_shard_id, client| async move {
-                    client
-                        .tenant_secondary_download(tenant_shard_id, Some(Duration::ZERO))
-                        .await
-                },
-                1,
-                1,
-                SHORT_RECONCILE_TIMEOUT,
-                &self.cancel,
-            )
-            .await
-        {
-            if let Err(e) = result {
-                tracing::warn!("Error calling secondary download after shard split: {e}");
-                return;
-            }
+            (response, child_locations)
         }
     }
 
@@ -3331,12 +3132,7 @@ impl Service {
     ) -> Result<TenantShardSplitResponse, ApiError> {
         // TODO: return 503 if we get stuck waiting for this lock
         // (issue https://github.com/neondatabase/neon/issues/7108)
-        let _tenant_lock = trace_exclusive_lock(
-            &self.tenant_op_locks,
-            tenant_id,
-            TenantOperations::ShardSplit,
-        )
-        .await;
+        let _tenant_lock = self.tenant_op_locks.exclusive(tenant_id).await;
 
         let new_shard_count = ShardCount::new(split_req.new_shard_count);
         let new_stripe_size = split_req.new_stripe_size;
@@ -3354,8 +3150,8 @@ impl Service {
             .do_tenant_shard_split(tenant_id, shard_split_params)
             .await;
 
-        let (response, waiters) = match r {
-            Ok(r) => r,
+        match r {
+            Ok(r) => Ok(r),
             Err(e) => {
                 // Split might be part-done, we must do work to abort it.
                 tracing::warn!("Enqueuing background abort of split on {tenant_id}");
@@ -3368,17 +3164,9 @@ impl Service {
                     })
                     // Ignore error sending: that just means we're shutting down: aborts are ephemeral so it's fine to drop it.
                     .ok();
-                return Err(e);
+                Err(e)
             }
-        };
-
-        // The split is now complete.  As an optimization, we will trigger all the child shards to upload
-        // a heatmap immediately, and all their secondary locations to start downloading: this avoids waiting
-        // for the background heatmap/download interval before secondaries get warm enough to migrate shards
-        // in [`Self::optimize_all`]
-        self.tenant_shard_split_start_secondaries(tenant_id, waiters)
-            .await;
-        Ok(response)
+        }
     }
 
     fn prepare_tenant_shard_split(
@@ -3528,7 +3316,7 @@ impl Service {
         &self,
         tenant_id: TenantId,
         params: ShardSplitParams,
-    ) -> Result<(TenantShardSplitResponse, Vec<ReconcilerWaiter>), ApiError> {
+    ) -> Result<TenantShardSplitResponse, ApiError> {
         // FIXME: we have dropped self.inner lock, and not yet written anything to the database: another
         // request could occur here, deleting or mutating the tenant.  begin_shard_split checks that the
         // parent shards exist as expected, but it would be neater to do the above pre-checks within the
@@ -3730,7 +3518,7 @@ impl Service {
         ));
 
         // Replace all the shards we just split with their children: this phase is infallible.
-        let (response, child_locations, waiters) =
+        let (response, child_locations) =
             self.tenant_shard_split_commit_inmem(tenant_id, new_shard_count, new_stripe_size);
 
         // Send compute notifications for all the new shards
@@ -3757,7 +3545,7 @@ impl Service {
             }
         }
 
-        Ok((response, waiters))
+        Ok(response)
     }
 
     pub(crate) async fn tenant_shard_migrate(
@@ -4105,13 +3893,9 @@ impl Service {
         &self,
         register_req: NodeRegisterRequest,
     ) -> Result<(), ApiError> {
-        let _node_lock = trace_exclusive_lock(
-            &self.node_op_locks,
-            register_req.node_id,
-            NodeOperations::Register,
-        )
-        .await;
+        let _node_lock = self.node_op_locks.exclusive(register_req.node_id).await;
 
+        // Pre-check for an already-existing node
         {
             let locked = self.inner.read().unwrap();
             if let Some(node) = locked.nodes.get(&register_req.node_id) {
@@ -4198,8 +3982,7 @@ impl Service {
         availability: Option<NodeAvailability>,
         scheduling: Option<NodeSchedulingPolicy>,
     ) -> Result<(), ApiError> {
-        let _node_lock =
-            trace_exclusive_lock(&self.node_op_locks, node_id, NodeOperations::Configure).await;
+        let _node_lock = self.node_op_locks.exclusive(node_id).await;
 
         if let Some(scheduling) = scheduling {
             // Scheduling is a persistent part of Node: we must write updates to the database before
@@ -4523,68 +4306,25 @@ impl Service {
     /// To put it more briefly: whereas the scheduler respects soft constraints in a ScheduleContext at
     /// the time of scheduling, this function looks for cases where a better-scoring location is available
     /// according to those same soft constraints.
-    async fn optimize_all(&self) -> usize {
+    fn optimize_all(&self) -> usize {
+        let mut locked = self.inner.write().unwrap();
+        let (nodes, tenants, scheduler) = locked.parts_mut();
+        let pageservers = nodes.clone();
+
+        let mut schedule_context = ScheduleContext::default();
+
+        let mut reconciles_spawned = 0;
+
+        let mut tenant_shards: Vec<&TenantShard> = Vec::new();
+
         // Limit on how many shards' optmizations each call to this function will execute.  Combined
         // with the frequency of background calls, this acts as an implicit rate limit that runs a small
         // trickle of optimizations in the background, rather than executing a large number in parallel
         // when a change occurs.
-        const MAX_OPTIMIZATIONS_EXEC_PER_PASS: usize = 2;
-
-        // Synchronous prepare: scan shards for possible scheduling optimizations
-        let candidate_work = self.optimize_all_plan();
-        let candidate_work_len = candidate_work.len();
-
-        // Asynchronous validate: I/O to pageservers to make sure shards are in a good state to apply validation
-        let validated_work = self.optimize_all_validate(candidate_work).await;
-
-        let was_work_filtered = validated_work.len() != candidate_work_len;
-
-        // Synchronous apply: update the shards' intent states according to validated optimisations
-        let mut reconciles_spawned = 0;
-        let mut optimizations_applied = 0;
-        let mut locked = self.inner.write().unwrap();
-        let (nodes, tenants, scheduler) = locked.parts_mut();
-        for (tenant_shard_id, optimization) in validated_work {
-            let Some(shard) = tenants.get_mut(&tenant_shard_id) else {
-                // Shard was dropped between planning and execution;
-                continue;
-            };
-            if shard.apply_optimization(scheduler, optimization) {
-                optimizations_applied += 1;
-                if self.maybe_reconcile_shard(shard, nodes).is_some() {
-                    reconciles_spawned += 1;
-                }
-            }
-
-            if optimizations_applied >= MAX_OPTIMIZATIONS_EXEC_PER_PASS {
-                break;
-            }
-        }
-
-        if was_work_filtered {
-            // If we filtered any work out during validation, ensure we return a nonzero value to indicate
-            // to callers that the system is not in a truly quiet state, it's going to do some work as soon
-            // as these validations start passing.
-            reconciles_spawned = std::cmp::max(reconciles_spawned, 1);
-        }
-
-        reconciles_spawned
-    }
-
-    fn optimize_all_plan(&self) -> Vec<(TenantShardId, ScheduleOptimization)> {
-        let mut schedule_context = ScheduleContext::default();
-
-        let mut tenant_shards: Vec<&TenantShard> = Vec::new();
-
-        // How many candidate optimizations we will generate, before evaluating them for readniess: setting
-        // this higher than the execution limit gives us a chance to execute some work even if the first
-        // few optimizations we find are not ready.
-        const MAX_OPTIMIZATIONS_PLAN_PER_PASS: usize = 8;
+        const MAX_OPTIMIZATIONS_PER_PASS: usize = 2;
 
         let mut work = Vec::new();
 
-        let mut locked = self.inner.write().unwrap();
-        let (nodes, tenants, scheduler) = locked.parts_mut();
         for (tenant_shard_id, shard) in tenants.iter() {
             if tenant_shard_id.is_shard_zero() {
                 // Reset accumulators on the first shard in a tenant
@@ -4593,7 +4333,7 @@ impl Service {
                 tenant_shards.clear();
             }
 
-            if work.len() >= MAX_OPTIMIZATIONS_PLAN_PER_PASS {
+            if work.len() >= MAX_OPTIMIZATIONS_PER_PASS {
                 break;
             }
 
@@ -4665,105 +4405,18 @@ impl Service {
             }
         }
 
-        work
-    }
+        for (tenant_shard_id, optimization) in work {
+            let shard = tenants
+                .get_mut(&tenant_shard_id)
+                .expect("We held lock from place we got this ID");
+            shard.apply_optimization(scheduler, optimization);
 
-    async fn optimize_all_validate(
-        &self,
-        candidate_work: Vec<(TenantShardId, ScheduleOptimization)>,
-    ) -> Vec<(TenantShardId, ScheduleOptimization)> {
-        // Take a clone of the node map to use outside the lock in async validation phase
-        let validation_nodes = { self.inner.read().unwrap().nodes.clone() };
-
-        let mut want_secondary_status = Vec::new();
-
-        // Validate our plans: this is an async phase where we may do I/O to pageservers to
-        // check that the state of locations is acceptable to run the optimization, such as
-        // checking that a secondary location is sufficiently warmed-up to cleanly cut over
-        // in a live migration.
-        let mut validated_work = Vec::new();
-        for (tenant_shard_id, optimization) in candidate_work {
-            match optimization.action {
-                ScheduleOptimizationAction::MigrateAttachment(MigrateAttachment {
-                    old_attached_node_id: _,
-                    new_attached_node_id,
-                }) => {
-                    match validation_nodes.get(&new_attached_node_id) {
-                        None => {
-                            // Node was dropped between planning and validation
-                        }
-                        Some(node) => {
-                            if !node.is_available() {
-                                tracing::info!("Skipping optimization migration of {tenant_shard_id} to {new_attached_node_id} because node unavailable");
-                            } else {
-                                // Accumulate optimizations that require fetching secondary status, so that we can execute these
-                                // remote API requests concurrently.
-                                want_secondary_status.push((
-                                    tenant_shard_id,
-                                    node.clone(),
-                                    optimization,
-                                ));
-                            }
-                        }
-                    }
-                }
-                ScheduleOptimizationAction::ReplaceSecondary(_) => {
-                    // No extra checks needed to replace a secondary: this does not interrupt client access
-                    validated_work.push((tenant_shard_id, optimization))
-                }
-            };
-        }
-
-        // Call into pageserver API to find out if the destination secondary location is warm enough for a reasonably smooth migration: we
-        // do this so that we avoid spawning a Reconciler that would have to wait minutes/hours for a destination to warm up: that reconciler
-        // would hold a precious reconcile semaphore unit the whole time it was waiting for the destination to warm up.
-        let results = self
-            .tenant_for_shards_api(
-                want_secondary_status
-                    .iter()
-                    .map(|i| (i.0, i.1.clone()))
-                    .collect(),
-                |tenant_shard_id, client| async move {
-                    client.tenant_secondary_status(tenant_shard_id).await
-                },
-                1,
-                1,
-                SHORT_RECONCILE_TIMEOUT,
-                &self.cancel,
-            )
-            .await;
-
-        for ((tenant_shard_id, node, optimization), secondary_status) in
-            want_secondary_status.into_iter().zip(results.into_iter())
-        {
-            match secondary_status {
-                Err(e) => {
-                    tracing::info!("Skipping migration of {tenant_shard_id} to {node}, error querying secondary: {e}");
-                }
-                Ok(progress) => {
-                    // We require secondary locations to have less than 10GiB of downloads pending before we will use
-                    // them in an optimization
-                    const DOWNLOAD_FRESHNESS_THRESHOLD: u64 = 10 * 1024 * 1024 * 1024;
-
-                    if progress.bytes_total == 0
-                        || progress.bytes_total < DOWNLOAD_FRESHNESS_THRESHOLD
-                            && progress.bytes_downloaded != progress.bytes_total
-                        || progress.bytes_total - progress.bytes_downloaded
-                            > DOWNLOAD_FRESHNESS_THRESHOLD
-                    {
-                        tracing::info!("Skipping migration of {tenant_shard_id} to {node} because secondary isn't ready: {progress:?}");
-                    } else {
-                        // Location looks ready: proceed
-                        tracing::info!(
-                            "{tenant_shard_id} secondary on {node} is warm enough for migration: {progress:?}"
-                        );
-                        validated_work.push((tenant_shard_id, optimization))
-                    }
-                }
+            if self.maybe_reconcile_shard(shard, &pageservers).is_some() {
+                reconciles_spawned += 1;
             }
         }
 
-        validated_work
+        reconciles_spawned
     }
 
     /// Useful for tests: run whatever work a background [`Self::reconcile_all`] would have done, but
@@ -4771,12 +4424,10 @@ impl Service {
     /// put the system into a quiescent state where future background reconciliations won't do anything.
     pub(crate) async fn reconcile_all_now(&self) -> Result<usize, ReconcileWaitError> {
         let reconciles_spawned = self.reconcile_all();
-        let reconciles_spawned = if reconciles_spawned == 0 {
+        if reconciles_spawned == 0 {
             // Only optimize when we are otherwise idle
-            self.optimize_all().await
-        } else {
-            reconciles_spawned
-        };
+            self.optimize_all();
+        }
 
         let waiters = {
             let mut waiters = Vec::new();

storage_controller/src/tenant_shard.rs

@@ -325,28 +325,18 @@ pub(crate) struct ReplaceSecondary {
 
 #[derive(Eq, PartialEq, Debug)]
 pub(crate) struct MigrateAttachment {
-    pub(crate) old_attached_node_id: NodeId,
-    pub(crate) new_attached_node_id: NodeId,
+    old_attached_node_id: NodeId,
+    new_attached_node_id: NodeId,
 }
 
 #[derive(Eq, PartialEq, Debug)]
-pub(crate) enum ScheduleOptimizationAction {
+pub(crate) enum ScheduleOptimization {
     // Replace one of our secondary locations with a different node
     ReplaceSecondary(ReplaceSecondary),
     // Migrate attachment to an existing secondary location
     MigrateAttachment(MigrateAttachment),
 }
 
-#[derive(Eq, PartialEq, Debug)]
-pub(crate) struct ScheduleOptimization {
-    // What was the reconcile sequence when we generated this optimization?  The optimization
-    // should only be applied if the shard's sequence is still at this value, in case other changes
-    // happened between planning the optimization and applying it.
-    sequence: Sequence,
-
-    pub(crate) action: ScheduleOptimizationAction,
-}
-
 impl ReconcilerWaiter {
     pub(crate) async fn wait_timeout(&self, timeout: Duration) -> Result<(), ReconcileWaitError> {
         tokio::select! {
@@ -685,13 +675,10 @@ impl TenantShard {
                         "Identified optimization: migrate attachment {attached}->{preferred_node} (secondaries {:?})",
                         self.intent.get_secondary()
                     );
-                    return Some(ScheduleOptimization {
-                        sequence: self.sequence,
-                        action: ScheduleOptimizationAction::MigrateAttachment(MigrateAttachment {
-                            old_attached_node_id: attached,
-                            new_attached_node_id: *preferred_node,
-                        }),
-                    });
+                    return Some(ScheduleOptimization::MigrateAttachment(MigrateAttachment {
+                        old_attached_node_id: attached,
+                        new_attached_node_id: *preferred_node,
+                    }));
                 }
             } else {
                 tracing::debug!(
@@ -749,37 +736,28 @@ impl TenantShard {
                     "Identified optimization: replace secondary {secondary}->{candidate_node} (current secondaries {:?})",
                     self.intent.get_secondary()
                 );
-                return Some(ScheduleOptimization {
-                    sequence: self.sequence,
-                    action: ScheduleOptimizationAction::ReplaceSecondary(ReplaceSecondary {
-                        old_node_id: *secondary,
-                        new_node_id: candidate_node,
-                    }),
-                });
+                return Some(ScheduleOptimization::ReplaceSecondary(ReplaceSecondary {
+                    old_node_id: *secondary,
+                    new_node_id: candidate_node,
+                }));
             }
         }
 
         None
     }
 
-    /// Return true if the optimization was really applied: it will not be applied if the optimization's
-    /// sequence is behind this tenant shard's
     pub(crate) fn apply_optimization(
         &mut self,
         scheduler: &mut Scheduler,
         optimization: ScheduleOptimization,
-    ) -> bool {
-        if optimization.sequence != self.sequence {
-            return false;
-        }
-
+    ) {
         metrics::METRICS_REGISTRY
             .metrics_group
             .storage_controller_schedule_optimization
             .inc();
 
-        match optimization.action {
-            ScheduleOptimizationAction::MigrateAttachment(MigrateAttachment {
+        match optimization {
+            ScheduleOptimization::MigrateAttachment(MigrateAttachment {
                 old_attached_node_id,
                 new_attached_node_id,
             }) => {
@@ -787,7 +765,7 @@ impl TenantShard {
                 self.intent
                     .promote_attached(scheduler, new_attached_node_id);
             }
-            ScheduleOptimizationAction::ReplaceSecondary(ReplaceSecondary {
+            ScheduleOptimization::ReplaceSecondary(ReplaceSecondary {
                 old_node_id,
                 new_node_id,
             }) => {
@@ -795,8 +773,6 @@ impl TenantShard {
                 self.intent.push_secondary(scheduler, new_node_id);
             }
         }
-
-        true
     }
 
     /// Query whether the tenant's observed state for attached node matches its intent state, and if so,
@@ -1452,13 +1428,10 @@ pub(crate) mod tests {
         // would be no other shards from the same tenant, and request to do so.
         assert_eq!(
             optimization_a,
-            Some(ScheduleOptimization {
-                sequence: shard_a.sequence,
-                action: ScheduleOptimizationAction::MigrateAttachment(MigrateAttachment {
-                    old_attached_node_id: NodeId(1),
-                    new_attached_node_id: NodeId(2)
-                })
-            })
+            Some(ScheduleOptimization::MigrateAttachment(MigrateAttachment {
+                old_attached_node_id: NodeId(1),
+                new_attached_node_id: NodeId(2)
+            }))
         );
 
         // Note that these optimizing two shards in the same tenant with the same ScheduleContext is
@@ -1469,13 +1442,10 @@ pub(crate) mod tests {
         let optimization_b = shard_b.optimize_attachment(&nodes, &schedule_context);
         assert_eq!(
             optimization_b,
-            Some(ScheduleOptimization {
-                sequence: shard_b.sequence,
-                action: ScheduleOptimizationAction::MigrateAttachment(MigrateAttachment {
-                    old_attached_node_id: NodeId(1),
-                    new_attached_node_id: NodeId(3)
-                })
-            })
+            Some(ScheduleOptimization::MigrateAttachment(MigrateAttachment {
+                old_attached_node_id: NodeId(1),
+                new_attached_node_id: NodeId(3)
+            }))
         );
 
         // Applying these optimizations should result in the end state proposed
@@ -1519,13 +1489,10 @@ pub(crate) mod tests {
         // same tenant should generate an optimization to move one away
         assert_eq!(
             optimization_a,
-            Some(ScheduleOptimization {
-                sequence: shard_a.sequence,
-                action: ScheduleOptimizationAction::ReplaceSecondary(ReplaceSecondary {
-                    old_node_id: NodeId(3),
-                    new_node_id: NodeId(4)
-                })
-            })
+            Some(ScheduleOptimization::ReplaceSecondary(ReplaceSecondary {
+                old_node_id: NodeId(3),
+                new_node_id: NodeId(4)
+            }))
         );
 
         shard_a.apply_optimization(&mut scheduler, optimization_a.unwrap());

test_runner/README.md

@@ -76,10 +76,13 @@ you can use `--pg-version` argument.
 `TEST_OUTPUT`: Set the directory where test state and test output files
 should go.
 `TEST_SHARED_FIXTURES`: Try to re-use a single pageserver for all the tests.
+`NEON_PAGESERVER_OVERRIDES`: add a `;`-separated set of configs that will be passed as
 `RUST_LOG`: logging configuration to pass into Neon CLI
 
 Useful parameters and commands:
 
+`--pageserver-config-override=${value}` `-c` values to pass into pageserver through neon_local cli
+
 `--preserve-database-files` to preserve pageserver (layer) and safekeer (segment) timeline files on disk
 after running a test suite. Such files might be large, so removed by default; but might be useful for debugging or creation of svg images with layer file contents.
 

test_runner/fixtures/neon_fixtures.py

@@ -14,7 +14,7 @@ import textwrap
 import threading
 import time
 import uuid
-from contextlib import ExitStack, closing, contextmanager
+from contextlib import closing, contextmanager
 from dataclasses import dataclass
 from datetime import datetime
 from enum import Enum
@@ -450,7 +450,6 @@ class NeonEnvBuilder:
         test_output_dir: Path,
         test_overlay_dir: Optional[Path] = None,
         pageserver_remote_storage: Optional[RemoteStorage] = None,
-        # toml that will be decomposed into `--config-override` flags during `pageserver --init`
         pageserver_config_override: Optional[str] = None,
         num_safekeepers: int = 1,
         num_pageservers: int = 1,
@@ -1022,6 +1021,7 @@ class NeonEnv:
         self.neon_local_binpath = config.neon_binpath
         self.pg_distrib_dir = config.pg_distrib_dir
         self.endpoint_counter = 0
+        self.pageserver_config_override = config.pageserver_config_override
         self.storage_controller_config = config.storage_controller_config
 
         # generate initial tenant ID here instead of letting 'neon init' generate it,
@@ -1104,6 +1104,7 @@ class NeonEnv:
                     self,
                     ps_id,
                     port=pageserver_port,
+                    config_override=self.pageserver_config_override,
                 )
             )
             cfg["pageservers"].append(ps_cfg)
@@ -1131,11 +1132,7 @@ class NeonEnv:
             cfg["safekeepers"].append(sk_cfg)
 
         log.info(f"Config: {cfg}")
-        self.neon_cli.init(
-            cfg,
-            force=config.config_init_force,
-            pageserver_config_override=config.pageserver_config_override,
-        )
+        self.neon_cli.init(cfg, force=config.config_init_force)
 
     def start(self):
         # Storage controller starts first, so that pageserver /re-attach calls don't
@@ -1707,47 +1704,30 @@ class NeonCli(AbstractNeonCli):
         self,
         config: Dict[str, Any],
         force: Optional[str] = None,
-        pageserver_config_override: Optional[str] = None,
     ) -> "subprocess.CompletedProcess[str]":
-        remote_storage = self.env.pageserver_remote_storage
+        with tempfile.NamedTemporaryFile(mode="w+") as tmp:
+            tmp.write(toml.dumps(config))
+            tmp.flush()
 
-        ps_config = {}
-        if remote_storage is not None:
-            remote_storage_toml_table = remote_storage_to_toml_inline_table(remote_storage)
-            ps_config["remote_storage"] = remote_storage_toml_table
-
-        if pageserver_config_override is not None:
-            for o in pageserver_config_override.split(";"):
-                override = toml.loads(o)
-                for key, value in override.items():
-                    ps_config[key] = value
-
-        with ExitStack() as stack:
-            ps_config_file = stack.enter_context(tempfile.NamedTemporaryFile(mode="w+"))
-            ps_config_file.write(toml.dumps(ps_config))
-            ps_config_file.flush()
-
-            neon_local_config = stack.enter_context(tempfile.NamedTemporaryFile(mode="w+"))
-            neon_local_config.write(toml.dumps(config))
-            neon_local_config.flush()
-
-            cmd = [
-                "init",
-                f"--config={neon_local_config.name}",
-                "--pg-version",
-                self.env.pg_version,
-                f"--pageserver-config={ps_config_file.name}",
-            ]
+            cmd = ["init", f"--config={tmp.name}", "--pg-version", self.env.pg_version]
 
             if force is not None:
                 cmd.extend(["--force", force])
 
+            storage = self.env.pageserver_remote_storage
+
+            append_pageserver_param_overrides(
+                params_to_update=cmd,
+                remote_storage=storage,
+                pageserver_config_override=self.env.pageserver_config_override,
+            )
+
             s3_env_vars = None
-            if isinstance(remote_storage, S3Storage):
-                s3_env_vars = remote_storage.access_env_vars()
+            if isinstance(storage, S3Storage):
+                s3_env_vars = storage.access_env_vars()
             res = self.raw_cli(cmd, extra_env_vars=s3_env_vars)
             res.check_returncode()
-        return res
+            return res
 
     def storage_controller_start(self):
         cmd = ["storage_controller", "start"]
@@ -1762,10 +1742,16 @@ class NeonCli(AbstractNeonCli):
     def pageserver_start(
         self,
         id: int,
+        overrides: Tuple[str, ...] = (),
         extra_env_vars: Optional[Dict[str, str]] = None,
     ) -> "subprocess.CompletedProcess[str]":
-        start_args = ["pageserver", "start", f"--id={id}"]
+        start_args = ["pageserver", "start", f"--id={id}", *overrides]
         storage = self.env.pageserver_remote_storage
+        append_pageserver_param_overrides(
+            params_to_update=start_args,
+            remote_storage=storage,
+            pageserver_config_override=self.env.pageserver_config_override,
+        )
 
         if isinstance(storage, S3Storage):
             s3_env_vars = storage.access_env_vars()
@@ -1973,55 +1959,6 @@ class Pagectl(AbstractNeonCli):
         return IndexPartDump.from_json(parsed)
 
 
-class LogUtils:
-    """
-    A mixin class which provides utilities for inspecting the logs of a service.
-    """
-
-    def __init__(self, logfile: Path) -> None:
-        self.logfile = logfile
-
-    def assert_log_contains(
-        self, pattern: str, offset: None | LogCursor = None
-    ) -> Tuple[str, LogCursor]:
-        """Convenient for use inside wait_until()"""
-
-        res = self.log_contains(pattern, offset=offset)
-        assert res is not None
-        return res
-
-    def log_contains(
-        self, pattern: str, offset: None | LogCursor = None
-    ) -> Optional[Tuple[str, LogCursor]]:
-        """Check that the log contains a line that matches the given regex"""
-        logfile = self.logfile
-        if not logfile.exists():
-            log.warning(f"Skipping log check: {logfile} does not exist")
-            return None
-
-        contains_re = re.compile(pattern)
-
-        # XXX: Our rust logging machinery buffers the messages, so if you
-        # call this function immediately after it's been logged, there is
-        # no guarantee it is already present in the log file. This hasn't
-        # been a problem in practice, our python tests are not fast enough
-        # to hit that race condition.
-        skip_until_line_no = 0 if offset is None else offset._line_no
-        cur_line_no = 0
-        with logfile.open("r") as f:
-            for line in f:
-                if cur_line_no < skip_until_line_no:
-                    cur_line_no += 1
-                    continue
-                elif contains_re.search(line):
-                    # found it!
-                    cur_line_no += 1
-                    return (line, LogCursor(cur_line_no))
-                else:
-                    cur_line_no += 1
-        return None
-
-
 class StorageControllerApiException(Exception):
     def __init__(self, message, status_code: int):
         super().__init__(message)
@@ -2029,13 +1966,12 @@ class StorageControllerApiException(Exception):
         self.status_code = status_code
 
 
-class NeonStorageController(MetricsGetter, LogUtils):
+class NeonStorageController(MetricsGetter):
     def __init__(self, env: NeonEnv, auth_enabled: bool):
         self.env = env
         self.running = False
         self.auth_enabled = auth_enabled
         self.allowed_errors: list[str] = DEFAULT_STORAGE_CONTROLLER_ALLOWED_ERRORS
-        self.logfile = self.workdir / "storage_controller.log"
 
     def start(self):
         assert not self.running
@@ -2359,10 +2295,6 @@ class NeonStorageController(MetricsGetter, LogUtils):
         log.info(f"Got failpoints request response code {res.status_code}")
         res.raise_for_status()
 
-    @property
-    def workdir(self) -> Path:
-        return self.env.repo_dir
-
     def __enter__(self) -> "NeonStorageController":
         return self
 
@@ -2380,21 +2312,24 @@ class LogCursor:
     _line_no: int
 
 
-class NeonPageserver(PgProtocol, LogUtils):
+class NeonPageserver(PgProtocol):
     """
     An object representing a running pageserver.
     """
 
     TEMP_FILE_SUFFIX = "___temp"
 
-    def __init__(self, env: NeonEnv, id: int, port: PageserverPort):
+    def __init__(
+        self, env: NeonEnv, id: int, port: PageserverPort, config_override: Optional[str] = None
+    ):
         super().__init__(host="localhost", port=port.pg, user="cloud_admin")
         self.env = env
         self.id = id
         self.running = False
         self.service_port = port
+        self.config_override = config_override
         self.version = env.get_binary_version("pageserver")
-        self.logfile = self.workdir / "pageserver.log"
+
         # After a test finishes, we will scrape the log to see if there are any
         # unexpected error messages. If your test expects an error, add it to
         # 'allowed_errors' in the test with something like:
@@ -2423,47 +2358,9 @@ class NeonPageserver(PgProtocol, LogUtils):
             return self.workdir / "tenants"
         return self.workdir / "tenants" / str(tenant_shard_id)
 
-    @property
-    def config_toml_path(self) -> Path:
-        return self.workdir / "pageserver.toml"
-
-    def edit_config_toml(self, edit_fn: Callable[[Dict[str, Any]], bool]):
-        """
-        Edit the pageserver's config toml file in place.
-
-        The `edit_fn` is to manipulate the dict, and if it returns True, the file will be written.
-        If it returns False, no changes are made to the file system.
-        """
-        path = self.config_toml_path
-        with open(path, "r") as f:
-            config = toml.load(f)
-        save = edit_fn(config)
-        if save:
-            with open(path, "w") as f:
-                toml.dump(config, f)
-
-    def patch_config_toml_nonrecursive(self, patch: Dict[str, Any]) -> Dict[str, Any]:
-        """
-        Non-recursively merge the given `patch` dict into the existing config toml, using `dict.update()`.
-        Returns the replaced values.
-        If there was no previous value, the key is mapped to None.
-        This allows to restore the original value by calling this method with the returned dict.
-        """
-        replacements = {}
-
-        def doit(config: Dict[str, Any]) -> bool:
-            while len(patch) > 0:
-                key, new = patch.popitem()
-                old = config.get(key, None)
-                config[key] = new
-                replacements[key] = old
-            return True
-
-        self.edit_config_toml(doit)
-        return replacements
-
     def start(
         self,
+        overrides: Tuple[str, ...] = (),
         extra_env_vars: Optional[Dict[str, str]] = None,
     ) -> "NeonPageserver":
         """
@@ -2473,7 +2370,9 @@ class NeonPageserver(PgProtocol, LogUtils):
         """
         assert self.running is False
 
-        self.env.neon_cli.pageserver_start(self.id, extra_env_vars=extra_env_vars)
+        self.env.neon_cli.pageserver_start(
+            self.id, overrides=overrides, extra_env_vars=extra_env_vars
+        )
         self.running = True
         return self
 
@@ -2570,6 +2469,46 @@ class NeonPageserver(PgProtocol, LogUtils):
             value = self.http_client().get_metric_value(metric)
             assert value == 0, f"Nonzero {metric} == {value}"
 
+    def assert_log_contains(
+        self, pattern: str, offset: None | LogCursor = None
+    ) -> Tuple[str, LogCursor]:
+        """Convenient for use inside wait_until()"""
+
+        res = self.log_contains(pattern, offset=offset)
+        assert res is not None
+        return res
+
+    def log_contains(
+        self, pattern: str, offset: None | LogCursor = None
+    ) -> Optional[Tuple[str, LogCursor]]:
+        """Check that the pageserver log contains a line that matches the given regex"""
+        logfile = self.workdir / "pageserver.log"
+        if not logfile.exists():
+            log.warning(f"Skipping log check: {logfile} does not exist")
+            return None
+
+        contains_re = re.compile(pattern)
+
+        # XXX: Our rust logging machinery buffers the messages, so if you
+        # call this function immediately after it's been logged, there is
+        # no guarantee it is already present in the log file. This hasn't
+        # been a problem in practice, our python tests are not fast enough
+        # to hit that race condition.
+        skip_until_line_no = 0 if offset is None else offset._line_no
+        cur_line_no = 0
+        with logfile.open("r") as f:
+            for line in f:
+                if cur_line_no < skip_until_line_no:
+                    cur_line_no += 1
+                    continue
+                elif contains_re.search(line):
+                    # found it!
+                    cur_line_no += 1
+                    return (line, LogCursor(cur_line_no))
+                else:
+                    cur_line_no += 1
+        return None
+
     def tenant_attach(
         self,
         tenant_id: TenantId,
@@ -2636,6 +2575,33 @@ class NeonPageserver(PgProtocol, LogUtils):
         )
 
 
+def append_pageserver_param_overrides(
+    params_to_update: List[str],
+    remote_storage: Optional[RemoteStorage],
+    pageserver_config_override: Optional[str] = None,
+):
+    if remote_storage is not None:
+        remote_storage_toml_table = remote_storage_to_toml_inline_table(remote_storage)
+
+        params_to_update.append(
+            f"--pageserver-config-override=remote_storage={remote_storage_toml_table}"
+        )
+    else:
+        params_to_update.append('--pageserver-config-override=remote_storage=""')
+
+    env_overrides = os.getenv("NEON_PAGESERVER_OVERRIDES")
+    if env_overrides is not None:
+        params_to_update += [
+            f"--pageserver-config-override={o.strip()}" for o in env_overrides.split(";")
+        ]
+
+    if pageserver_config_override is not None:
+        params_to_update += [
+            f"--pageserver-config-override={o.strip()}"
+            for o in pageserver_config_override.split(";")
+        ]
+
+
 class PgBin:
     """A helper class for executing postgres binaries"""
 

test_runner/performance/test_branch_creation.py

@@ -140,13 +140,10 @@ def test_branch_creation_many(neon_compare: NeonCompare, n_branches: int, shape:
 
     # start without gc so we can time compaction with less noise; use shorter
     # period for compaction so it starts earlier
-    def patch_default_tenant_config(config):
-        config["compaction_period"] = "3s"
-        config["gc_period"] = "0s"
-        return True
-
-    env.pageserver.edit_config_toml(patch_default_tenant_config)
     env.pageserver.start(
+        overrides=(
+            "--pageserver-config-override=tenant_config={ compaction_period = '3s', gc_period = '0s' }",
+        ),
         # this does print more than we want, but the number should be comparable between runs
         extra_env_vars={
             "RUST_LOG": f"[compaction_loop{{tenant_id={env.initial_tenant}}}]=debug,info"

test_runner/regress/test_branching.py

@@ -1,7 +1,6 @@
 import random
 import threading
 import time
-from concurrent.futures import ThreadPoolExecutor
 from typing import List
 
 import pytest
@@ -406,29 +405,6 @@ def test_duplicate_creation(neon_env_builder: NeonEnvBuilder):
     assert len(ps_http.timeline_list(tenant_id=env.initial_tenant)) == 1
 
 
-def test_branching_while_stuck_find_gc_cutoffs(neon_env_builder: NeonEnvBuilder):
-    env = neon_env_builder.init_start()
-
-    client = env.pageserver.http_client()
-
-    failpoint = "Timeline::find_gc_cutoffs-pausable"
-
-    client.configure_failpoints((failpoint, "pause"))
-
-    with ThreadPoolExecutor(max_workers=1) as exec:
-        completion = exec.submit(client.timeline_gc, env.initial_tenant, env.initial_timeline, None)
-
-        wait_until_paused(env, failpoint)
-
-        env.neon_cli.create_branch(
-            tenant_id=env.initial_tenant, ancestor_branch_name="main", new_branch_name="branch"
-        )
-
-        client.configure_failpoints((failpoint, "off"))
-
-        completion.result()
-
-
 def wait_until_paused(env: NeonEnv, failpoint: str):
     found = False
     msg = f"at failpoint {failpoint}"

test_runner/regress/test_disk_usage_eviction.py

@@ -5,6 +5,7 @@ from dataclasses import dataclass
 from typing import Any, Dict, Iterable, Tuple
 
 import pytest
+import toml
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import (
     NeonEnv,
@@ -44,15 +45,17 @@ def test_min_resident_size_override_handling(
         ps_http.set_tenant_config(tenant_id, {})
         assert_config(tenant_id, None, default_tenant_conf_value)
 
-    if config_level_override is not None:
-
-        def set_min_resident_size(config):
-            config["tenant_config"] = {"min_resident_size": config_level_override}
-            return True
-
-        env.pageserver.edit_config_toml(set_min_resident_size)
     env.pageserver.stop()
-    env.pageserver.start()
+    if config_level_override is not None:
+        env.pageserver.start(
+            overrides=(
+                "--pageserver-config-override=tenant_config={ min_resident_size_override =  "
+                + str(config_level_override)
+                + " }",
+            )
+        )
+    else:
+        env.pageserver.start()
 
     tenant_id, _ = env.neon_cli.create_tenant()
     assert_overrides(tenant_id, config_level_override)
@@ -161,32 +164,34 @@ class EvictionEnv:
         usage eviction task is unknown; it might need to run one more iteration
         before assertions can be made.
         """
+        disk_usage_config = {
+            "period": period,
+            "max_usage_pct": max_usage_pct,
+            "min_avail_bytes": min_avail_bytes,
+            "mock_statvfs": mock_behavior,
+            "eviction_order": eviction_order.config(),
+        }
+
+        enc = toml.TomlEncoder()
 
         # these can sometimes happen during startup before any tenants have been
         # loaded, so nothing can be evicted, we just wait for next iteration which
         # is able to evict.
         pageserver.allowed_errors.append(".*WARN.* disk usage still high.*")
 
-        pageserver.patch_config_toml_nonrecursive(
-            {
-                "disk_usage_based_eviction": {
-                    "period": period,
-                    "max_usage_pct": max_usage_pct,
-                    "min_avail_bytes": min_avail_bytes,
-                    "mock_statvfs": mock_behavior,
-                    "eviction_order": eviction_order.config(),
-                },
+        pageserver.start(
+            overrides=(
+                "--pageserver-config-override=disk_usage_based_eviction="
+                + enc.dump_inline_table(disk_usage_config).replace("\n", " "),
                 # Disk usage based eviction runs as a background task.
                 # But pageserver startup delays launch of background tasks for some time, to prioritize initial logical size calculations during startup.
                 # But, initial logical size calculation may not be triggered if safekeepers don't publish new broker messages.
                 # But, we only have a 10-second-timeout in this test.
                 # So, disable the delay for this test.
-                "background_task_maximum_delay": "0s",
-            }
+                "--pageserver-config-override=background_task_maximum_delay='0s'",
+            ),
         )
 
-        pageserver.start()
-
         # we now do initial logical size calculation on startup, which on debug builds can fight with disk usage based eviction
         for tenant_id, timeline_id in self.timelines:
             tenant_ps = self.neon_env.get_tenant_pageserver(tenant_id)

test_runner/regress/test_ondemand_download.py

@@ -333,17 +333,6 @@ def test_download_remote_layers_api(
         }
     )
 
-    # This test triggers layer download failures on demand. It is possible to modify the failpoint
-    # during a `Timeline::get_vectored` right between the vectored read and it's validation read.
-    # This means that one of the reads can fail while the other one succeeds and vice versa.
-    # TODO(vlad): Remove this block once the vectored read path validation goes away.
-    env.pageserver.allowed_errors.extend(
-        [
-            ".*initial_size_calculation.*Vectored get failed with downloading evicted layer file failed, but sequential get did not.*"
-            ".*initial_size_calculation.*Sequential get failed with downloading evicted layer file failed, but vectored get did not.*"
-        ]
-    )
-
     endpoint = env.endpoints.create_start("main")
 
     client = env.pageserver.http_client()

test_runner/regress/test_pageserver_api.py

@@ -12,6 +12,7 @@ from fixtures.types import Lsn, TenantId, TimelineId
 from fixtures.utils import wait_until
 
 
+# test that we cannot override node id after init
 def test_pageserver_init_node_id(
     neon_simple_env: NeonEnv, neon_binpath: Path, pg_distrib_dir: Path
 ):
@@ -48,7 +49,11 @@ def test_pageserver_init_node_id(
 
     bad_reinit = run_pageserver(good_init_cmd)
     assert bad_reinit.returncode == 1, "pageserver refuses to init if already exists"
-    assert "config file already exists" in bad_reinit.stderr
+    assert "already exists, cannot init it" in bad_reinit.stderr
+
+    bad_update = run_pageserver(["--update-config", "-c", "id = 3"])
+    assert bad_update.returncode == 1, "pageserver should not allow updating node id"
+    assert "has node id already, it cannot be overridden" in bad_update.stderr
 
 
 def check_client(env: NeonEnv, client: PageserverHttpClient):

test_runner/regress/test_pageserver_generations.py

@@ -220,12 +220,7 @@ def test_generations_upgrade(neon_env_builder: NeonEnvBuilder):
     # We will start a pageserver with no control_plane_api set, so it won't be able to self-register
     env.storage_controller.node_register(env.pageserver)
 
-    replaced_config = env.pageserver.patch_config_toml_nonrecursive(
-        {
-            "control_plane_api": "",
-        }
-    )
-    env.pageserver.start()
+    env.pageserver.start(overrides=('--pageserver-config-override=control_plane_api=""',))
     env.storage_controller.node_configure(env.pageserver.id, {"availability": "Active"})
 
     env.neon_cli.create_tenant(
@@ -256,8 +251,8 @@ def test_generations_upgrade(neon_env_builder: NeonEnvBuilder):
         assert parse_generation_suffix(key) is None
 
     env.pageserver.stop()
+
     # Starting without the override that disabled control_plane_api
-    env.pageserver.patch_config_toml_nonrecursive(replaced_config)
     env.pageserver.start()
 
     generate_uploads_and_deletions(env, pageserver=env.pageserver, init=False)
@@ -530,12 +525,9 @@ def test_emergency_mode(neon_env_builder: NeonEnvBuilder, pg_bin: PgBin):
     # incident, but it might be unavoidable: if so, we want to be able to start up
     # and serve clients.
     env.pageserver.stop()  # Non-immediate: implicitly checking that shutdown doesn't hang waiting for CP
-    replaced = env.pageserver.patch_config_toml_nonrecursive(
-        {
-            "control_plane_emergency_mode": True,
-        }
+    env.pageserver.start(
+        overrides=("--pageserver-config-override=control_plane_emergency_mode=true",),
     )
-    env.pageserver.start()
 
     # The pageserver should provide service to clients
     generate_uploads_and_deletions(env, init=False, pageserver=env.pageserver)
@@ -557,7 +549,6 @@ def test_emergency_mode(neon_env_builder: NeonEnvBuilder, pg_bin: PgBin):
 
     # The pageserver should work fine when subsequently restarted in non-emergency mode
     env.pageserver.stop()  # Non-immediate: implicitly checking that shutdown doesn't hang waiting for CP
-    env.pageserver.patch_config_toml_nonrecursive(replaced)
     env.pageserver.start()
 
     generate_uploads_and_deletions(env, init=False, pageserver=env.pageserver)

test_runner/regress/test_remote_storage.py

@@ -1,3 +1,6 @@
+# It's possible to run any regular test with the local fs remote storage via
+# env NEON_PAGESERVER_OVERRIDES="remote_storage={local_path='/tmp/neon_zzz/'}" poetry ......
+
 import os
 import queue
 import shutil

test_runner/regress/test_sharding.py

@@ -287,11 +287,6 @@ def test_sharding_split_smoke(
         == shard_count
     )
 
-    # Make secondary downloads slow: this exercises the storage controller logic for not migrating an attachment
-    # during post-split optimization until the secondary is ready
-    for ps in env.pageservers:
-        ps.http_client().configure_failpoints([("secondary-layer-download-sleep", "return(1000)")])
-
     env.storage_controller.tenant_shard_split(tenant_id, shard_count=split_shard_count)
 
     post_split_pageserver_ids = [loc["node_id"] for loc in env.storage_controller.locate(tenant_id)]
@@ -305,7 +300,7 @@ def test_sharding_split_smoke(
 
     # Enough background reconciliations should result in the shards being properly distributed.
     # Run this before the workload, because its LSN-waiting code presumes stable locations.
-    env.storage_controller.reconcile_until_idle(timeout_secs=60)
+    env.storage_controller.reconcile_until_idle()
 
     workload.validate()
 
@@ -347,10 +342,6 @@ def test_sharding_split_smoke(
     assert cancelled_reconciles is not None and int(cancelled_reconciles) == 0
     assert errored_reconciles is not None and int(errored_reconciles) == 0
 
-    # We should see that the migration of shards after the split waited for secondaries to warm up
-    # before happening
-    assert env.storage_controller.log_contains(".*Skipping.*because secondary isn't ready.*")
-
     env.storage_controller.consistency_check()
 
     def get_node_shard_counts(env: NeonEnv, tenant_ids):
@@ -1080,17 +1071,6 @@ def test_sharding_split_failures(
         finish_split()
         assert_split_done()
 
-    if isinstance(failure, StorageControllerFailpoint) and "post-complete" in failure.failpoint:
-        # On a post-complete failure, the controller will recover the post-split state
-        # after restart, but it will have missed the optimization part of the split function
-        # where secondary downloads are kicked off.  This means that reconcile_until_idle
-        # will take a very long time if we wait for all optimizations to complete, because
-        # those optimizations will wait for secondary downloads.
-        #
-        # Avoid that by configuring the tenant into Essential scheduling mode, so that it will
-        # skip optimizations when we're exercising this particular failpoint.
-        env.storage_controller.tenant_policy_update(tenant_id, {"scheduling": "Essential"})
-
     # Having completed the split, pump the background reconciles to ensure that
     # the scheduler reaches an idle state
     env.storage_controller.reconcile_until_idle(timeout_secs=30)

test_runner/regress/test_storage_controller.py

@@ -1,5 +1,4 @@
 import json
-import threading
 import time
 from collections import defaultdict
 from datetime import datetime, timezone
@@ -230,10 +229,6 @@ def test_storage_controller_passthrough(
     }
     assert status["state"]["slug"] == "Active"
 
-    (synthetic_size, size_inputs) = client.tenant_size_and_modelinputs(env.initial_tenant)
-    assert synthetic_size > 0
-    assert "segments" in size_inputs
-
     env.storage_controller.consistency_check()
 
 
@@ -290,12 +285,9 @@ def test_storage_controller_onboarding(neon_env_builder: NeonEnvBuilder, warm_up
     # This is the pageserver where we'll initially create the tenant.  Run it in emergency
     # mode so that it doesn't talk to storage controller, and do not register it.
     env.pageservers[0].allowed_errors.append(".*Emergency mode!.*")
-    env.pageservers[0].patch_config_toml_nonrecursive(
-        {
-            "control_plane_emergency_mode": True,
-        }
+    env.pageservers[0].start(
+        overrides=("--pageserver-config-override=control_plane_emergency_mode=true",),
     )
-    env.pageservers[0].start()
     origin_ps = env.pageservers[0]
 
     # These are the pageservers managed by the sharding service, where the tenant
@@ -1267,53 +1259,6 @@ def test_storcon_cli(neon_env_builder: NeonEnvBuilder):
     env.storage_controller.consistency_check()
 
 
-def test_lock_time_tracing(neon_env_builder: NeonEnvBuilder):
-    """
-    Check that when lock on resource (tenants, nodes) is held for too long it is
-    traced in logs.
-    """
-    env = neon_env_builder.init_start()
-    tenant_id = env.initial_tenant
-    env.storage_controller.allowed_errors.extend(
-        [
-            ".*Lock on.*",
-            ".*Scheduling is disabled by policy.*",
-            f".*Operation TimelineCreate on key {tenant_id} has waited.*",
-        ]
-    )
-
-    # Apply failpoint
-    env.storage_controller.configure_failpoints(
-        ("tenant-update-policy-exclusive-lock", "return(31000)")
-    )
-
-    # This will hold the exclusive for enough time to cause an warning
-    def update_tenent_policy():
-        env.storage_controller.tenant_policy_update(
-            tenant_id=tenant_id,
-            body={
-                "scheduling": "Stop",
-            },
-        )
-
-    thread_update_tenant_policy = threading.Thread(target=update_tenent_policy)
-    thread_update_tenant_policy.start()
-
-    # Make sure the update policy thread has started
-    time.sleep(1)
-    # This will not be able to access and will log a warning
-    timeline_id = TimelineId.generate()
-    env.storage_controller.pageserver_api().timeline_create(
-        pg_version=PgVersion.NOT_SET, tenant_id=tenant_id, new_timeline_id=timeline_id
-    )
-    thread_update_tenant_policy.join(timeout=10)
-
-    env.storage_controller.assert_log_contains("Lock on UpdatePolicy was held for")
-    env.storage_controller.assert_log_contains(
-        f"Operation TimelineCreate on key {tenant_id} has waited"
-    )
-
-
 @pytest.mark.parametrize("remote_storage", [RemoteStorageKind.LOCAL_FS, s3_storage()])
 @pytest.mark.parametrize("shard_count", [None, 4])
 def test_tenant_import(neon_env_builder: NeonEnvBuilder, shard_count, remote_storage):

test_runner/regress/test_tenant_size.py

@@ -1,5 +1,4 @@
 import os
-from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 from typing import List, Tuple
 
@@ -12,15 +11,13 @@ from fixtures.neon_fixtures import (
     wait_for_last_flush_lsn,
     wait_for_wal_insert_lsn,
 )
-from fixtures.pageserver.http import PageserverApiException, PageserverHttpClient
+from fixtures.pageserver.http import PageserverHttpClient
 from fixtures.pageserver.utils import (
-    tenant_delete_wait_completed,
     timeline_delete_wait_completed,
     wait_until_tenant_active,
 )
 from fixtures.pg_version import PgVersion
 from fixtures.types import Lsn, TenantId, TimelineId
-from fixtures.utils import wait_until
 
 
 def test_empty_tenant_size(neon_env_builder: NeonEnvBuilder):
@@ -619,68 +616,6 @@ def test_get_tenant_size_with_multiple_branches(
     size_debug_file.write(size_debug)
 
 
-def test_synthetic_size_while_deleting(neon_env_builder: NeonEnvBuilder):
-    """
-    Makes sure synthetic size can still be calculated even if one of the
-    timelines is deleted or the tenant is deleted.
-    """
-
-    env = neon_env_builder.init_start()
-    failpoint = "Timeline::find_gc_cutoffs-pausable"
-    client = env.pageserver.http_client()
-
-    orig_size = client.tenant_size(env.initial_tenant)
-
-    branch_id = env.neon_cli.create_branch(
-        tenant_id=env.initial_tenant, ancestor_branch_name="main", new_branch_name="branch"
-    )
-    client.configure_failpoints((failpoint, "pause"))
-
-    with ThreadPoolExecutor(max_workers=1) as exec:
-        completion = exec.submit(client.tenant_size, env.initial_tenant)
-        _, last_offset = wait_until(
-            10, 1.0, lambda: env.pageserver.assert_log_contains(f"at failpoint {failpoint}")
-        )
-
-        timeline_delete_wait_completed(client, env.initial_tenant, branch_id)
-
-        client.configure_failpoints((failpoint, "off"))
-        size = completion.result()
-
-        assert_size_approx_equal(orig_size, size)
-
-    branch_id = env.neon_cli.create_branch(
-        tenant_id=env.initial_tenant, ancestor_branch_name="main", new_branch_name="branch2"
-    )
-    client.configure_failpoints((failpoint, "pause"))
-
-    with ThreadPoolExecutor(max_workers=1) as exec:
-        completion = exec.submit(client.tenant_size, env.initial_tenant)
-        wait_until(
-            10,
-            1.0,
-            lambda: env.pageserver.assert_log_contains(
-                f"at failpoint {failpoint}", offset=last_offset
-            ),
-        )
-
-        tenant_delete_wait_completed(client, env.initial_tenant, 10)
-
-        client.configure_failpoints((failpoint, "off"))
-
-        with pytest.raises(
-            PageserverApiException, match="Failed to refresh gc_info before gathering inputs"
-        ):
-            completion.result()
-
-    # this happens on both cases
-    env.pageserver.allowed_errors.append(
-        ".*ignoring failure to find gc cutoffs: timeline shutting down.*"
-    )
-    # this happens only in the case of deletion (http response logging)
-    env.pageserver.allowed_errors.append(".*Failed to refresh gc_info before gathering inputs.*")
-
-
 # Helper for tests that compare timeline_inputs
 # We don't want to compare the exact values, because they can be unstable
 # and cause flaky tests. So replace the values with useful invariants.

vm-image-spec.yaml

@@ -5,12 +5,6 @@ commands:
     user: root
     sysvInitAction: sysinit
     shell: 'cgconfigparser -l /etc/cgconfig.conf -s 1664'
-  # restrict permissions on /neonvm/bin/resize-swap, because we grant access to compute_ctl for
-  # running it as root.
-  - name: chmod-resize-swap
-    user: root
-    sysvInitAction: sysinit
-    shell: 'chmod 711 /neonvm/bin/resize-swap'
   - name: pgbouncer
     user: postgres
     sysvInitAction: respawn
@@ -30,11 +24,6 @@ commands:
 shutdownHook: |
   su -p postgres --session-command '/usr/local/bin/pg_ctl stop -D /var/db/postgres/compute/pgdata -m fast --wait -t 10'
 files:
-  - filename: compute_ctl-resize-swap
-    content: |
-      # Allow postgres user (which is what compute_ctl runs as) to run /neonvm/bin/resize-swap
-      # as root without requiring entering a password (NOPASSWD), regardless of hostname (ALL)
-      postgres ALL=(root) NOPASSWD: /neonvm/bin/resize-swap
   - filename: pgbouncer.ini
     content: |
       [databases]
@@ -364,17 +353,6 @@ merge: |
       && echo 'root - nofile 1048576' >>/etc/security/limits.conf \
          )
 
-  # Allow postgres user (compute_ctl) to run swap resizer.
-  # Need to install sudo in order to allow this.
-  #
-  # Also, remove the 'read' permission from group/other on /neonvm/bin/resize-swap, just to be safe.
-  RUN set -e \
-      && apt update \
-      && apt install --no-install-recommends -y \
-             sudo \
-      && rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/*
-  COPY compute_ctl-resize-swap /etc/sudoers.d/compute_ctl-resize-swap
-
   COPY cgconfig.conf /etc/cgconfig.conf
   COPY pgbouncer.ini /etc/pgbouncer.ini
   COPY sql_exporter.yml /etc/sql_exporter.yml