v16-lr-worker-reply-after-processing

.gitmodules

@@ -5,8 +5,8 @@
 [submodule "vendor/postgres-v15"]
 	path = vendor/postgres-v15
 	url = https://github.com/neondatabase/postgres.git
-	branch = tristan957/15/pg_upgrade
+	branch = REL_15_STABLE_neon
 [submodule "vendor/postgres-v16"]
 	path = vendor/postgres-v16
 	url = https://github.com/neondatabase/postgres.git
-	branch = tristan957/pg_upgrade
+	branch = REL_16_STABLE_neon

Cargo.lock

@@ -1220,7 +1220,6 @@ dependencies = [
  "anyhow",
  "async-compression",
  "bytes",
- "camino",
  "cfg-if",
  "chrono",
  "clap",
@@ -1238,7 +1237,6 @@ dependencies = [
  "reqwest 0.12.4",
  "rlimit",
  "rust-ini",
- "scopeguard",
  "serde",
  "serde_json",
  "signal-hook",

compute_tools/Cargo.toml

@@ -12,7 +12,6 @@ testing = []
 [dependencies]
 anyhow.workspace = true
 async-compression.workspace = true
-camino.workspace = true
 chrono.workspace = true
 cfg-if.workspace = true
 clap.workspace = true
@@ -25,7 +24,6 @@ num_cpus.workspace = true
 opentelemetry.workspace = true
 postgres.workspace = true
 regex.workspace = true
-scopeguard.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 signal-hook.workspace = true

compute_tools/src/bin/compute_ctl.rs

@@ -43,7 +43,7 @@ use std::{thread, time::Duration};
 
 use anyhow::{Context, Result};
 use chrono::Utc;
-use clap::{Arg, ArgAction};
+use clap::Arg;
 use compute_tools::lsn_lease::launch_lsn_lease_bg_task_for_static;
 use signal_hook::consts::{SIGQUIT, SIGTERM};
 use signal_hook::{consts::SIGINT, iterator::Signals};
@@ -51,12 +51,13 @@ use tracing::{error, info, warn};
 use url::Url;
 
 use compute_api::responses::ComputeStatus;
-use compute_api::spec::{ComputeMode, ComputeSpec};
+use compute_api::spec::ComputeSpec;
 
 use compute_tools::compute::{
     forward_termination_signal, ComputeNode, ComputeState, ParsedSpec, PG_PID,
 };
 use compute_tools::configurator::launch_configurator;
+use compute_tools::extension_server::get_pg_version;
 use compute_tools::http::api::launch_http_server;
 use compute_tools::logger::*;
 use compute_tools::monitor::launch_monitor;
@@ -69,14 +70,13 @@ use rlimit::{setrlimit, Resource};
 // in-case of not-set environment var
 const BUILD_TAG_DEFAULT: &str = "latest";
 
-#[tokio::main]
-async fn main() -> Result<()> {
+fn main() -> Result<()> {
     let (build_tag, clap_args) = init()?;
 
     // enable core dumping for all child processes
     setrlimit(Resource::CORE, rlimit::INFINITY, rlimit::INFINITY)?;
 
-    let (http_handle, (pg_handle, start_pg_result)) = {
+    let (pg_handle, start_pg_result) = {
         // Enter startup tracing context
         let _startup_context_guard = startup_context_from_env();
 
@@ -84,61 +84,13 @@ async fn main() -> Result<()> {
 
         let cli_spec = try_spec_from_cli(&clap_args, &cli_args)?;
 
-        let compute = Arc::new(ComputeNode {
-            connstr: Url::parse(cli_args.connstr).context("cannot parse connstr as a URL")?,
-            pgdata: cli_args.pgdata.to_string(),
-            pgroot: cli_args.pgroot.to_string(),
-            pgversion: cli_args.pgversion.to_string(),
-            http_port: cli_args.http_port,
-            live_config_allowed: cli_spec.live_config_allowed,
-            state: Mutex::new({
-                let mut state = ComputeState::new();
+        let wait_spec_result = wait_spec(build_tag, cli_args, cli_spec)?;
 
-                if let Some(spec) = cli_spec.spec {
-                    let pspec = ParsedSpec::try_from(spec).map_err(|msg| anyhow::anyhow!(msg))?;
-                    info!("new pspec.spec: {:?}", pspec.spec);
-                    state.pspec = Some(pspec);
-                }
-
-                state
-            }),
-            state_changed: Condvar::new(),
-            ext_remote_storage: cli_args.ext_remote_storage.map(|s| s.to_string()),
-            ext_download_progress: RwLock::new(HashMap::new()),
-            build_tag: build_tag.clone(),
-        });
-
-        // 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
-        // the necessary binaries will already be cached.
-        if compute.state.lock().unwrap().pspec.is_none() {
-            compute.prewarm_postgres()?;
-        }
-
-        // Launch http service first, so that we can serve control-plane requests
-        // while configuration is still in progress.
-        let http_handle = launch_http_server(cli_args.http_port, &compute)
-            .expect("cannot launch http endpoint thread");
-
-        wait_spec(&compute)?;
-
-        (
-            http_handle,
-            start_postgres(
-                compute,
-                #[cfg(target_os = "linux")]
-                &clap_args,
-                cli_args.resize_swap_on_bind,
-            )
-            .await?,
-        )
+        start_postgres(&clap_args, wait_spec_result)?
 
         // Startup is finished, exit the startup tracing span
     };
 
-    let _ = http_handle.join();
-
     // PostgreSQL is now running, if startup was successful. Wait until it exits.
     let wait_pg_result = wait_postgres(pg_handle)?;
 
@@ -168,14 +120,11 @@ fn init() -> Result<(String, clap::ArgMatches)> {
 }
 
 fn process_cli(matches: &clap::ArgMatches) -> Result<ProcessCliResult> {
-    let pgroot = matches
-        .get_one::<String>("pgroot")
+    let pgbin_default = "postgres";
+    let pgbin = matches
+        .get_one::<String>("pgbin")
         .map(|s| s.as_str())
-        .expect("pgroot is required");
-    let pgversion = matches
-        .get_one::<String>("pgversion")
-        .map(|s| s.as_str())
-        .expect("pgversion is required");
+        .unwrap_or(pgbin_default);
 
     let ext_remote_storage = matches
         .get_one::<String>("remote-ext-config")
@@ -206,8 +155,7 @@ fn process_cli(matches: &clap::ArgMatches) -> Result<ProcessCliResult> {
     Ok(ProcessCliResult {
         connstr,
         pgdata,
-        pgroot,
-        pgversion,
+        pgbin,
         ext_remote_storage,
         http_port,
         spec_json,
@@ -219,8 +167,7 @@ fn process_cli(matches: &clap::ArgMatches) -> Result<ProcessCliResult> {
 struct ProcessCliResult<'clap> {
     connstr: &'clap str,
     pgdata: &'clap str,
-    pgroot: &'clap str,
-    pgversion: &'clap str,
+    pgbin: &'clap str,
     ext_remote_storage: Option<&'clap str>,
     http_port: u16,
     spec_json: Option<&'clap String>,
@@ -338,8 +285,61 @@ struct CliSpecParams {
     live_config_allowed: bool,
 }
 
-fn wait_spec(compute: &Arc<ComputeNode>) -> Result<()> {
-    if compute.state.lock().unwrap().pspec.is_none() {
+fn wait_spec(
+    build_tag: String,
+    ProcessCliResult {
+        connstr,
+        pgdata,
+        pgbin,
+        ext_remote_storage,
+        resize_swap_on_bind,
+        http_port,
+        ..
+    }: ProcessCliResult,
+    CliSpecParams {
+        spec,
+        live_config_allowed,
+    }: CliSpecParams,
+) -> Result<WaitSpecResult> {
+    let mut new_state = ComputeState::new();
+    let spec_set;
+
+    if let Some(spec) = spec {
+        let pspec = ParsedSpec::try_from(spec).map_err(|msg| anyhow::anyhow!(msg))?;
+        info!("new pspec.spec: {:?}", pspec.spec);
+        new_state.pspec = Some(pspec);
+        spec_set = true;
+    } else {
+        spec_set = false;
+    }
+    let compute_node = ComputeNode {
+        connstr: Url::parse(connstr).context("cannot parse connstr as a URL")?,
+        pgdata: pgdata.to_string(),
+        pgbin: pgbin.to_string(),
+        pgversion: get_pg_version(pgbin),
+        live_config_allowed,
+        state: Mutex::new(new_state),
+        state_changed: Condvar::new(),
+        ext_remote_storage: ext_remote_storage.map(|s| s.to_string()),
+        ext_download_progress: RwLock::new(HashMap::new()),
+        build_tag,
+    };
+    let compute = Arc::new(compute_node);
+
+    // 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
+    // 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.
+    let _http_handle =
+        launch_http_server(http_port, &compute).expect("cannot launch http endpoint thread");
+
+    if !spec_set {
         // No spec provided, hang waiting for it.
         info!("no compute spec provided, waiting");
 
@@ -369,13 +369,28 @@ fn wait_spec(compute: &Arc<ComputeNode>) -> Result<()> {
 
     launch_lsn_lease_bg_task_for_static(&compute);
 
-    Ok(())
+    Ok(WaitSpecResult {
+        compute,
+        http_port,
+        resize_swap_on_bind,
+    })
 }
 
-async fn start_postgres(
+struct WaitSpecResult {
     compute: Arc<ComputeNode>,
-    #[cfg(target_os = "linux")] matches: &clap::ArgMatches,
+    // passed through from ProcessCliResult
+    http_port: u16,
     resize_swap_on_bind: bool,
+}
+
+fn start_postgres(
+    // need to allow unused because `matches` is only used if target_os = "linux"
+    #[allow(unused_variables)] matches: &clap::ArgMatches,
+    WaitSpecResult {
+        compute,
+        http_port,
+        resize_swap_on_bind,
+    }: WaitSpecResult,
 ) -> Result<(Option<PostgresHandle>, StartPostgresResult)> {
     // We got all we need, update the state.
     let mut state = compute.state.lock().unwrap();
@@ -426,32 +441,27 @@ async fn start_postgres(
         }
     }
 
-    compute.prepare_compute().await?;
+    let extension_server_port: u16 = http_port;
 
+    // Start Postgres
     let mut pg = None;
     if !prestartup_failed {
-        match compute.get_mode() {
-            ComputeMode::Upgrade => {}
-            _ => {
-                // Start Postgres
-                pg = match compute.start_compute() {
-                    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
-                    }
-                }
+        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 {
@@ -676,17 +686,11 @@ fn cli() -> clap::Command {
                 .required(true),
         )
         .arg(
-            Arg::new("pgroot")
-                .short('R')
-                .long("pgroot")
-                .value_name("POSTGRES_ROOT")
-                .required(true),
-        )
-        .arg(
-            Arg::new("pgversion")
-                .long("pgversion")
-                .value_name("POSTGRES_VERSION")
-                .required(true),
+            Arg::new("pgbin")
+                .short('b')
+                .long("pgbin")
+                .default_value("postgres")
+                .value_name("POSTGRES_PATH"),
         )
         .arg(
             Arg::new("spec")
@@ -746,11 +750,6 @@ fn cli() -> clap::Command {
                 .long("resize-swap-on-bind")
                 .action(clap::ArgAction::SetTrue),
         )
-        .arg(
-            Arg::new("no-postgres")
-                .long("no-postgres")
-                .action(ArgAction::SetTrue),
-        )
 }
 
 /// When compute_ctl is killed, send also termination signal to sync-safekeepers

compute_tools/src/catalog.rs

@@ -4,7 +4,7 @@ use compute_api::{
 };
 use futures::Stream;
 use postgres::{Client, NoTls};
-use std::{process::Stdio, result::Result, sync::Arc};
+use std::{path::Path, process::Stdio, result::Result, sync::Arc};
 use tokio::{
     io::{AsyncBufReadExt, BufReader},
     process::Command,
@@ -55,7 +55,9 @@ pub async fn get_database_schema(
     compute: &Arc<ComputeNode>,
     dbname: &str,
 ) -> Result<impl Stream<Item = Result<bytes::Bytes, std::io::Error>>, SchemaDumpError> {
-    let pgdump = compute.get_my_pg_binary("pg_dump");
+    let pgbin = &compute.pgbin;
+    let basepath = Path::new(pgbin).parent().unwrap();
+    let pgdump = basepath.join("pg_dump");
     let mut connstr = compute.connstr.clone();
     connstr.set_path(dbname);
     let mut cmd = Command::new(pgdump)

compute_tools/src/compute.rs

@@ -1,9 +1,9 @@
 use std::collections::HashMap;
 use std::env;
 use std::fs;
+use std::io::BufRead;
 use std::os::unix::fs::{symlink, PermissionsExt};
 use std::path::Path;
-use std::path::PathBuf;
 use std::process::{Command, Stdio};
 use std::str::FromStr;
 use std::sync::atomic::AtomicU32;
@@ -13,8 +13,6 @@ use std::thread;
 use std::time::Instant;
 
 use anyhow::{Context, Result};
-use bytes::{Buf, BufMut};
-use camino::Utf8Path;
 use chrono::{DateTime, Utc};
 use futures::future::join_all;
 use futures::stream::FuturesUnordered;
@@ -22,15 +20,13 @@ use futures::StreamExt;
 use nix::unistd::Pid;
 use postgres::error::SqlState;
 use postgres::{Client, NoTls};
-use tokio;
-use tokio_postgres;
 use tracing::{debug, error, info, instrument, warn};
 use utils::id::{TenantId, TimelineId};
 use utils::lsn::Lsn;
-use utils::zstd::create_zst_tarball;
 
 use compute_api::responses::{ComputeMetrics, ComputeStatus};
 use compute_api::spec::{ComputeFeature, ComputeMode, ComputeSpec};
+use utils::measured_stream::MeasuredReader;
 
 use nix::sys::signal::{kill, Signal};
 
@@ -51,9 +47,8 @@ pub struct ComputeNode {
     // Url type maintains proper escaping
     pub connstr: url::Url,
     pub pgdata: String,
-    pub pgroot: String,
+    pub pgbin: String,
     pub pgversion: String,
-    pub http_port: u16,
     /// We should only allow live re- / configuration of the compute node if
     /// it uses 'pull model', i.e. it can go to control-plane and fetch
     /// the latest configuration. Otherwise, there could be a case:
@@ -314,13 +309,6 @@ impl ComputeNode {
         self.state.lock().unwrap().status
     }
 
-    /// Get the mode of this compute.
-    pub fn get_mode(&self) -> ComputeMode {
-        let state = self.state.lock().unwrap();
-
-        state.pspec.as_ref().unwrap().spec.mode
-    }
-
     // Remove `pgdata` directory and create it again with right permissions.
     fn create_pgdata(&self) -> Result<()> {
         // Ignore removal error, likely it is a 'No such file or directory (os error 2)'.
@@ -332,44 +320,15 @@ impl ComputeNode {
         Ok(())
     }
 
-    /// Get path pointing to requested binary directory.
-    pub fn get_pg_bindir(&self, version: &str) -> PathBuf {
-        Path::new(&self.pgroot)
-            .join(format!("v{}", version))
-            .join("bin")
-    }
-
-    /// Get path to requested Postgres binary.
-    pub fn get_pg_binary(&self, version: &str, binary: &str) -> String {
-        self.get_pg_bindir(version)
-            .join(binary)
-            .into_os_string()
-            .into_string()
-            .expect(&format!(
-                "path to {}-{} cannot be represented as UTF-8",
-                binary, version
-            ))
-    }
-
-    /// Get path to Postgres binary directory of this compute.
-    pub fn get_my_pg_bindir(&self) -> PathBuf {
-        self.get_pg_bindir(&self.pgversion)
-    }
-
-    /// Get path to specified Postgres binary of this compute.
-    pub fn get_my_pg_binary(&self, binary: &str) -> String {
-        self.get_pg_binary(&self.pgversion, binary)
-    }
-
     // Get basebackup from the libpq connection to pageserver using `connstr` and
     // unarchive it to `pgdata` directory overriding all its previous content.
     #[instrument(skip_all, fields(%lsn))]
-    async fn try_get_basebackup(&self, compute_state: &ComputeState, lsn: Lsn) -> Result<()> {
+    fn try_get_basebackup(&self, compute_state: &ComputeState, lsn: Lsn) -> Result<()> {
         let spec = compute_state.pspec.as_ref().expect("spec must be set");
         let start_time = Instant::now();
 
         let shard0_connstr = spec.pageserver_connstr.split(',').next().unwrap();
-        let mut config = tokio_postgres::Config::from_str(shard0_connstr)?;
+        let mut config = postgres::Config::from_str(shard0_connstr)?;
 
         // Use the storage auth token from the config file, if given.
         // Note: this overrides any password set in the connection string.
@@ -381,12 +340,7 @@ impl ComputeNode {
         }
 
         // Connect to pageserver
-        let (client, connection) = config.connect(tokio_postgres::NoTls).await?;
-        tokio::spawn(async move {
-            if let Err(e) = connection.await {
-                eprintln!("connection error: {}", e);
-            }
-        });
+        let mut client = config.connect(NoTls)?;
         let pageserver_connect_micros = start_time.elapsed().as_micros() as u64;
 
         let basebackup_cmd = match lsn {
@@ -398,18 +352,8 @@ impl ComputeNode {
             ),
         };
 
-        let mut copystream = std::pin::pin!(client.copy_out(basebackup_cmd.as_str()).await?);
-        let mut buf = bytes::BytesMut::with_capacity(1024);
-        while let Some(i) = copystream.next().await {
-            match i {
-                Ok(b) => {
-                    buf.put(b);
-                }
-                Err(e) => return Err(anyhow::anyhow!(e)),
-            }
-        }
-
-        let basebackup_bytes = buf.len();
+        let copyreader = client.copy_out(basebackup_cmd.as_str())?;
+        let mut measured_reader = MeasuredReader::new(copyreader);
 
         // Check the magic number to see if it's a gzip or not. Even though
         // we might explicitly ask for gzip, an old pageserver with no implementation
@@ -422,9 +366,11 @@ impl ComputeNode {
         // and 0x1f and 0x8b are unlikely first characters for any filename. Moreover,
         // we send the "global" directory first from the pageserver, so it definitely
         // won't be recognized as gzip.
-        let gzip = buf[0] == 0x1f && buf[1] == 0x8b;
-
-        let mut bufreader = buf.reader();
+        let mut bufreader = std::io::BufReader::new(&mut measured_reader);
+        let gzip = {
+            let peek = bufreader.fill_buf().unwrap();
+            peek[0] == 0x1f && peek[1] == 0x8b
+        };
 
         // Read the archive directly from the `CopyOutReader`
         //
@@ -444,14 +390,14 @@ impl ComputeNode {
         // Report metrics
         let mut state = self.state.lock().unwrap();
         state.metrics.pageserver_connect_micros = pageserver_connect_micros;
-        state.metrics.basebackup_bytes = basebackup_bytes as u64;
+        state.metrics.basebackup_bytes = measured_reader.get_byte_count() as u64;
         state.metrics.basebackup_ms = start_time.elapsed().as_millis() as u64;
         Ok(())
     }
 
     // Gets the basebackup in a retry loop
     #[instrument(skip_all, fields(%lsn))]
-    pub async fn get_basebackup(&self, compute_state: &ComputeState, lsn: Lsn) -> Result<()> {
+    pub fn get_basebackup(&self, compute_state: &ComputeState, lsn: Lsn) -> Result<()> {
         let mut retry_period_ms = 500.0;
         let mut attempts = 0;
         const DEFAULT_ATTEMPTS: u16 = 10;
@@ -464,7 +410,7 @@ impl ComputeNode {
         #[cfg(not(feature = "testing"))]
         let max_attempts = DEFAULT_ATTEMPTS;
         loop {
-            let result = self.try_get_basebackup(compute_state, lsn).await;
+            let result = self.try_get_basebackup(compute_state, lsn);
             match result {
                 Ok(_) => {
                     return result;
@@ -485,14 +431,10 @@ impl ComputeNode {
         }
     }
 
-    // Fast path for sync_safekeepers. If they're already synced we get the lsn
-    // in one roundtrip. If not, we should do a full sync_safekeepers.
-    pub async fn check_safekeepers_synced(
+    pub async fn check_safekeepers_synced_async(
         &self,
         compute_state: &ComputeState,
     ) -> Result<Option<Lsn>> {
-        let start_time = Utc::now();
-
         // Construct a connection config for each safekeeper
         let pspec: ParsedSpec = compute_state
             .pspec
@@ -561,7 +503,20 @@ impl ComputeNode {
             return Ok(None);
         }
 
-        let lsn = check_if_synced(responses);
+        Ok(check_if_synced(responses))
+    }
+
+    // Fast path for sync_safekeepers. If they're already synced we get the lsn
+    // in one roundtrip. If not, we should do a full sync_safekeepers.
+    pub fn check_safekeepers_synced(&self, compute_state: &ComputeState) -> Result<Option<Lsn>> {
+        let start_time = Utc::now();
+
+        // Run actual work with new tokio runtime
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("failed to create rt");
+        let result = rt.block_on(self.check_safekeepers_synced_async(compute_state));
 
         // Record runtime
         self.state.lock().unwrap().metrics.sync_sk_check_ms = Utc::now()
@@ -569,8 +524,7 @@ impl ComputeNode {
             .to_std()
             .unwrap()
             .as_millis() as u64;
-
-        Ok(lsn)
+        result
     }
 
     // Run `postgres` in a special mode with `--sync-safekeepers` argument
@@ -579,8 +533,7 @@ impl ComputeNode {
     pub fn sync_safekeepers(&self, storage_auth_token: Option<String>) -> Result<Lsn> {
         let start_time = Utc::now();
 
-        let postgres = self.get_my_pg_binary("postgres");
-        let mut sync_handle = maybe_cgexec(&postgres)
+        let mut sync_handle = maybe_cgexec(&self.pgbin)
             .args(["--sync-safekeepers"])
             .env("PGDATA", &self.pgdata) // we cannot use -D in this mode
             .envs(if let Some(storage_auth_token) = &storage_auth_token {
@@ -636,7 +589,11 @@ impl ComputeNode {
     /// Do all the preparations like PGDATA directory creation, configuration,
     /// safekeepers sync, basebackup, etc.
     #[instrument(skip_all)]
-    pub async fn prepare_pgdata(&self, compute_state: &ComputeState) -> Result<()> {
+    pub fn prepare_pgdata(
+        &self,
+        compute_state: &ComputeState,
+        extension_server_port: u16,
+    ) -> Result<()> {
         let pspec = compute_state.pspec.as_ref().expect("spec must be set");
         let spec = &pspec.spec;
         let pgdata_path = Path::new(&self.pgdata);
@@ -646,7 +603,7 @@ impl ComputeNode {
         config::write_postgres_conf(
             &pgdata_path.join("postgresql.conf"),
             &pspec.spec,
-            Some(self.http_port),
+            Some(extension_server_port),
         )?;
 
         // Syncing safekeepers is only safe with primary nodes: if a primary
@@ -655,8 +612,7 @@ impl ComputeNode {
         let lsn = match spec.mode {
             ComputeMode::Primary => {
                 info!("checking if safekeepers are synced");
-                let lsn = if let Ok(Some(lsn)) = self.check_safekeepers_synced(compute_state).await
-                {
+                let lsn = if let Ok(Some(lsn)) = self.check_safekeepers_synced(compute_state) {
                     lsn
                 } else {
                     info!("starting safekeepers syncing");
@@ -674,24 +630,18 @@ impl ComputeNode {
                 info!("Initializing standby from latest Pageserver LSN");
                 Lsn(0)
             }
-            ComputeMode::Upgrade => {
-                info!("Starting upgrade node at latest Pageserver LSN");
-                Lsn(0)
-            }
         };
 
         info!(
             "getting basebackup@{} from pageserver {}",
             lsn, &pspec.pageserver_connstr
         );
-        self.get_basebackup(compute_state, lsn)
-            .await
-            .with_context(|| {
-                format!(
-                    "failed to get basebackup@{} from pageserver {}",
-                    lsn, &pspec.pageserver_connstr
-                )
-            })?;
+        self.get_basebackup(compute_state, lsn).with_context(|| {
+            format!(
+                "failed to get basebackup@{} from pageserver {}",
+                lsn, &pspec.pageserver_connstr
+            )
+        })?;
 
         // Update pg_hba.conf received with basebackup.
         update_pg_hba(pgdata_path)?;
@@ -739,7 +689,7 @@ impl ComputeNode {
         symlink("/dev/shm/", pgdata_path.join("pg_dynshmem"))?;
 
         match spec.mode {
-            ComputeMode::Primary | ComputeMode::Upgrade => {}
+            ComputeMode::Primary => {}
             ComputeMode::Replica | ComputeMode::Static(..) => {
                 add_standby_signal(pgdata_path)?;
             }
@@ -758,7 +708,7 @@ impl ComputeNode {
 
         // Run initdb to completion
         info!("running initdb");
-        let initdb_bin = self.get_my_pg_binary("initdb");
+        let initdb_bin = Path::new(&self.pgbin).parent().unwrap().join("initdb");
         Command::new(initdb_bin)
             .args(["-D", pgdata])
             .output()
@@ -774,8 +724,7 @@ impl ComputeNode {
 
         // Start postgres
         info!("starting postgres");
-        let postgres = self.get_my_pg_binary("postgres");
-        let mut pg = maybe_cgexec(&postgres)
+        let mut pg = maybe_cgexec(&self.pgbin)
             .args(["-D", pgdata])
             .spawn()
             .expect("cannot start postgres process");
@@ -808,8 +757,7 @@ impl ComputeNode {
         let pgdata_path = Path::new(&self.pgdata);
 
         // Run postgres as a child process.
-        let postgres = self.get_my_pg_binary("postgres");
-        let mut pg = maybe_cgexec(&postgres)
+        let mut pg = maybe_cgexec(&self.pgbin)
             .args(["-D", &self.pgdata])
             .envs(if let Some(storage_auth_token) = &storage_auth_token {
                 vec![("NEON_AUTH_TOKEN", storage_auth_token)]
@@ -830,152 +778,6 @@ impl ComputeNode {
         Ok((pg, logs_handle))
     }
 
-    pub async fn upgrade(&self, pg_version: &str) -> Result<()> {
-        let old_bindir = self.get_my_pg_bindir();
-        let new_bindir = self.get_pg_bindir(pg_version);
-
-        let old_datadir = Utf8Path::new(&self.pgdata);
-        let parent_dir = old_datadir.parent().unwrap();
-        let new_datadir = parent_dir.join("new-pgdata");
-
-        // Delete the new data directory before attempting, just in case it exists
-        let _ = std::fs::remove_dir_all(&new_datadir);
-
-        // Step 1: Create new cluster
-        info!(
-            "Running initdb to start a cluster upgrade from v{} to v{}",
-            self.pgversion, pg_version
-        );
-
-        let initdb_bin = self.get_pg_binary(pg_version, "initdb");
-        let mut initdb_cmd = Command::new(&initdb_bin);
-        initdb_cmd
-            .args(["--pgdata", new_datadir.as_str()])
-            .args(["--username", "cloud_admin"])
-            .args(["--encoding", "utf8"])
-            .args(["--auth-local", "trust"])
-            .env_clear()
-            .stdout(Stdio::inherit())
-            .stderr(Stdio::inherit());
-
-        match initdb_cmd.status() {
-            Ok(status) => {
-                if !status.success() {
-                    return Err(anyhow::anyhow!("failed to initialize the new database"));
-                }
-
-                info!("Initialized v{} database", pg_version);
-            }
-            Err(_) => {
-                return Err(anyhow::anyhow!(
-                    "failed to spawn initdb for the new database"
-                ))
-            }
-        };
-
-        // Step 2: Run pg_upgrade
-        info!(
-            "Running pg_upgrade to upgrade from v{} to v{}",
-            self.pgversion, pg_version
-        );
-
-        let pg_upgrade_bin = self.get_pg_binary(pg_version, "pg_upgrade");
-        let mut pg_upgrade_cmd = Command::new(&pg_upgrade_bin);
-        let mut child = pg_upgrade_cmd
-            .args([
-                "--old-bindir",
-                &old_bindir.into_os_string().into_string().unwrap(),
-            ])
-            .args(["--old-datadir", old_datadir.as_str()])
-            .args(["--old-options", "-c neon.safekeepers=''"])
-            .args([
-                "--new-bindir",
-                &new_bindir.into_os_string().into_string().unwrap(),
-            ])
-            .args(["--new-datadir", new_datadir.as_str()])
-            .args(["--new-options", "-c neon.safekeepers=''"])
-            .args(["--username", "cloud_admin"])
-            .arg("--no-transfer")
-            .stdout(Stdio::inherit())
-            .stderr(Stdio::inherit())
-            .spawn()?;
-
-        let status = child.wait()?;
-        if status.success() {
-            info!("pg_upgrade was successful");
-        } else {
-            return Err(anyhow::anyhow!(
-                "pg_upgrade failed with exit code {}",
-                status.code().unwrap()
-            ));
-        }
-
-        /* Step 3: Delete the script that pg_upgrade generates, which is created in the current
-         * working directory
-         */
-        // TODO: We should write a patch for upstream to not generate this file
-        if cfg!(windows) {
-            let _ = std::fs::remove_file("delete_old_cluster.bat");
-        } else {
-            let _ = std::fs::remove_file("delete_old_cluster.sh");
-        }
-
-        /* Step 4: Re-prepare the pgdata directory to work with the latest basebackup from the
-         * pageserver
-         */
-        {
-            let state = self.state.lock().unwrap().clone();
-
-            self.prepare_pgdata(&state).await?;
-        }
-
-        // Step 5: Create tarball minus things like pg_dynshm, etc.
-        info!("Creating tarball of upgraded initdb directory, minus some files");
-
-        let initdb_tar_path = parent_dir.join("initdb.tar.zst");
-        let _ = std::fs::remove_file(&initdb_tar_path);
-        create_zst_tarball(&new_datadir, &initdb_tar_path).await?;
-
-        // Step 6: Write new postgresql.conf file for upgraded initdb
-        std::fs::copy(
-            old_datadir.join("postgresql.conf"),
-            new_datadir.join("postgresql.conf"),
-        )?;
-
-        // Step 7: Write the tarball into the Postgres WAL
-        info!("Writing {} to WAL", initdb_tar_path.as_str());
-
-        let postgres_bin = self.get_my_pg_binary("postgres");
-        let mut wal_log_cmd = Command::new(&postgres_bin);
-        child = wal_log_cmd
-            .args(["--wal-log", initdb_tar_path.as_str()])
-            .env_clear()
-            .env("PGDATA", old_datadir.as_str())
-            .env("NEON_PURPOSE", "upgrade")
-            .stdout(Stdio::inherit())
-            .stderr(Stdio::inherit())
-            .spawn()
-            .expect("postgres --wal-log failed to start");
-        match child.wait() {
-            Ok(s) => {
-                if !s.success() {
-                    return Err(anyhow::anyhow!("Could not WAL log upgrade tarball"));
-                }
-            }
-            Err(e) => return Err(e.into()),
-        };
-
-        // Step 8: Sync the safekeepers to push WAL record to Neon
-        self.sync_safekeepers(None)?;
-
-        /* Note that whether any errors occur after this are unimportant. ALWAYS return success
-         * after this point. The compute will be terminated immediately after the upgrade. Remember
-         * that this is an upgrade-only compute, and it will not accept connections from users.
-         */
-
-        Ok(())
-    }
-
     /// Do post configuration of the already started Postgres. This function spawns a background thread to
     /// configure the database after applying the compute spec. Currently, it upgrades the neon extension
     /// version. In the future, it may upgrade all 3rd-party extensions.
@@ -1093,8 +895,8 @@ impl ComputeNode {
     // `pg_ctl` for start / stop.
     #[instrument(skip_all)]
     fn pg_reload_conf(&self) -> Result<()> {
-        let pgctl_bin = self.get_my_pg_binary("pg_ctl");
-        Command::new(&pgctl_bin)
+        let pgctl_bin = Path::new(&self.pgbin).parent().unwrap().join("pg_ctl");
+        Command::new(pgctl_bin)
             .args(["reload", "-D", &self.pgdata])
             .output()
             .expect("cannot run pg_ctl process");
@@ -1178,25 +980,43 @@ impl ComputeNode {
         Ok(())
     }
 
-    /// Prepares the compute for Postgres operations, including downloading
-    /// remote extensions and preparing the pgdata directory.
-    ///
-    /// The caller must hold the state mutex.
     #[instrument(skip_all)]
-    pub async fn prepare_compute(&self) -> Result<()> {
-        let state = self.state.lock().unwrap().clone();
-        let pspec = state.pspec.as_ref().expect("spec must be set");
-
+    pub fn start_compute(
+        &self,
+        extension_server_port: u16,
+    ) -> Result<(std::process::Child, std::thread::JoinHandle<()>)> {
+        let compute_state = self.state.lock().unwrap().clone();
+        let pspec = compute_state.pspec.as_ref().expect("spec must be set");
         info!(
-            "preparing compute for project {}, operation {}, tenant {}, timeline {}",
+            "starting compute for project {}, operation {}, tenant {}, timeline {}",
             pspec.spec.cluster.cluster_id.as_deref().unwrap_or("None"),
             pspec.spec.operation_uuid.as_deref().unwrap_or("None"),
             pspec.tenant_id,
             pspec.timeline_id,
         );
 
+        // tune pgbouncer
+        if let Some(pgbouncer_settings) = &pspec.spec.pgbouncer_settings {
+            info!("tuning pgbouncer");
+
+            let rt = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .expect("failed to create rt");
+
+            // Spawn a thread to do the tuning,
+            // so that we don't block the main thread that starts Postgres.
+            let pgbouncer_settings = pgbouncer_settings.clone();
+            let _handle = thread::spawn(move || {
+                let res = rt.block_on(tune_pgbouncer(pgbouncer_settings));
+                if let Err(err) = res {
+                    error!("error while tuning pgbouncer: {err:?}");
+                }
+            });
+        }
+
         info!(
-            "prepare_compute spec.remote_extensions {:?}",
+            "start_compute spec.remote_extensions {:?}",
             pspec.spec.remote_extensions
         );
 
@@ -1204,8 +1024,7 @@ impl ComputeNode {
         // remote shared_preload_libraries before postgres start (if any)
         if let Some(remote_extensions) = &pspec.spec.remote_extensions {
             // First, create control files for all availale extensions
-            let postgres_bin = self.get_my_pg_binary("postgres");
-            extension_server::create_control_files(remote_extensions, &postgres_bin);
+            extension_server::create_control_files(remote_extensions, &self.pgbin);
 
             let library_load_start_time = Utc::now();
             let remote_ext_metrics = self.prepare_preload_libraries(&pspec.spec)?;
@@ -1227,17 +1046,7 @@ impl ComputeNode {
             info!("{:?}", remote_ext_metrics);
         }
 
-        self.prepare_pgdata(&state).await?;
-
-        self.set_status(ComputeStatus::Prepared);
-
-        Ok(())
-    }
-
-    #[instrument(skip_all)]
-    pub fn start_compute(&self) -> Result<(std::process::Child, std::thread::JoinHandle<()>)> {
-        let compute_state = self.state.lock().unwrap().clone();
-        let pspec = compute_state.pspec.as_ref().expect("spec must be set");
+        self.prepare_pgdata(&compute_state, extension_server_port)?;
 
         let start_time = Utc::now();
         let pg_process = self.start_postgres(pspec.storage_auth_token.clone())?;
@@ -1343,11 +1152,9 @@ impl ComputeNode {
                 core_path.display()
             );
 
-            let postgres_bin = self.get_my_pg_binary("postgres");
-
             // Try first with gdb
             let backtrace = Command::new("gdb")
-                .args(["--batch", "-q", "-ex", "bt", &postgres_bin])
+                .args(["--batch", "-q", "-ex", "bt", &self.pgbin])
                 .arg(&core_path)
                 .output();
 
@@ -1480,12 +1287,11 @@ LIMIT 100",
         // then we try to download it here
         info!("downloading new extension {ext_archive_name}");
 
-        let postgres_bin = self.get_my_pg_binary("postgres");
         let download_size = extension_server::download_extension(
             &real_ext_name,
             &ext_path,
             ext_remote_storage,
-            &postgres_bin,
+            &self.pgbin,
         )
         .await
         .map_err(DownloadError::Other);

compute_tools/src/config.rs

@@ -74,7 +74,7 @@ pub fn write_postgres_conf(
     }
 
     match spec.mode {
-        ComputeMode::Primary | ComputeMode::Upgrade => {}
+        ComputeMode::Primary => {}
         ComputeMode::Static(lsn) => {
             // hot_standby is 'on' by default, but let's be explicit
             writeln!(file, "hot_standby=on")?;

compute_tools/src/extension_server.rs

@@ -75,6 +75,7 @@ use anyhow::Result;
 use anyhow::{bail, Context};
 use bytes::Bytes;
 use compute_api::spec::RemoteExtSpec;
+use regex::Regex;
 use remote_storage::*;
 use reqwest::StatusCode;
 use std::path::Path;
@@ -102,6 +103,34 @@ fn get_pg_config(argument: &str, pgbin: &str) -> String {
         .to_string()
 }
 
+pub fn get_pg_version(pgbin: &str) -> String {
+    // pg_config --version returns a (platform specific) human readable string
+    // such as "PostgreSQL 15.4". We parse this to v14/v15/v16 etc.
+    let human_version = get_pg_config("--version", pgbin);
+    return parse_pg_version(&human_version).to_string();
+}
+
+fn parse_pg_version(human_version: &str) -> &str {
+    // Normal releases have version strings like "PostgreSQL 15.4". But there
+    // are also pre-release versions like "PostgreSQL 17devel" or "PostgreSQL
+    // 16beta2" or "PostgreSQL 17rc1". And with the --with-extra-version
+    // configure option, you can tack any string to the version number,
+    // e.g. "PostgreSQL 15.4foobar".
+    match Regex::new(r"^PostgreSQL (?<major>\d+).+")
+        .unwrap()
+        .captures(human_version)
+    {
+        Some(captures) if captures.len() == 2 => match &captures["major"] {
+            "14" => return "v14",
+            "15" => return "v15",
+            "16" => return "v16",
+            _ => {}
+        },
+        _ => {}
+    }
+    panic!("Unsuported postgres version {human_version}");
+}
+
 // download the archive for a given extension,
 // unzip it, and place files in the appropriate locations (share/lib)
 pub async fn download_extension(
@@ -226,3 +255,42 @@ async fn download_extension_tar(ext_remote_storage: &str, ext_path: &str) -> Res
         ),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::parse_pg_version;
+
+    #[test]
+    fn test_parse_pg_version() {
+        assert_eq!(parse_pg_version("PostgreSQL 15.4"), "v15");
+        assert_eq!(parse_pg_version("PostgreSQL 15.14"), "v15");
+        assert_eq!(
+            parse_pg_version("PostgreSQL 15.4 (Ubuntu 15.4-0ubuntu0.23.04.1)"),
+            "v15"
+        );
+
+        assert_eq!(parse_pg_version("PostgreSQL 14.15"), "v14");
+        assert_eq!(parse_pg_version("PostgreSQL 14.0"), "v14");
+        assert_eq!(
+            parse_pg_version("PostgreSQL 14.9 (Debian 14.9-1.pgdg120+1"),
+            "v14"
+        );
+
+        assert_eq!(parse_pg_version("PostgreSQL 16devel"), "v16");
+        assert_eq!(parse_pg_version("PostgreSQL 16beta1"), "v16");
+        assert_eq!(parse_pg_version("PostgreSQL 16rc2"), "v16");
+        assert_eq!(parse_pg_version("PostgreSQL 16extra"), "v16");
+    }
+
+    #[test]
+    #[should_panic]
+    fn test_parse_pg_unsupported_version() {
+        parse_pg_version("PostgreSQL 13.14");
+    }
+
+    #[test]
+    #[should_panic]
+    fn test_parse_pg_incorrect_version_format() {
+        parse_pg_version("PostgreSQL 14");
+    }
+}

compute_tools/src/http/api.rs

@@ -10,7 +10,6 @@ use crate::catalog::{get_database_schema, get_dbs_and_roles};
 use crate::compute::forward_termination_signal;
 use crate::compute::{ComputeNode, ComputeState, ParsedSpec};
 use compute_api::requests::ConfigurationRequest;
-use compute_api::requests::UpgradeRequest;
 use compute_api::responses::{ComputeStatus, ComputeStatusResponse, GenericAPIError};
 
 use anyhow::Result;
@@ -138,20 +137,6 @@ async fn routes(req: Request<Body>, compute: &Arc<ComputeNode>) -> Response<Body
             }
         }
 
-        (&Method::POST, "/upgrade") => {
-            info!("serving /upgrade POST request");
-            match handle_upgrade_request(req, compute).await {
-                Ok(_) => Response::builder()
-                    .status(StatusCode::ACCEPTED)
-                    .body(Body::from("Starting upgrade"))
-                    .unwrap(),
-                Err((e, status)) => {
-                    error!("error handling /upgrade request: {e}");
-                    render_json_error(&format!("{}", e), status)
-                }
-            }
-        }
-
         (&Method::GET, "/dbs_and_roles") => {
             info!("serving /dbs_and_roles GET request",);
             match get_dbs_and_roles(compute).await {
@@ -412,59 +397,6 @@ async fn handle_terminate_request(compute: &Arc<ComputeNode>) -> Result<(), (Str
     Ok(())
 }
 
-async fn handle_upgrade_request(
-    req: Request<Body>,
-    compute: &Arc<ComputeNode>,
-) -> Result<(), (anyhow::Error, StatusCode)> {
-    let body_bytes = hyper::body::to_bytes(req.into_body()).await.unwrap();
-    let body_str = String::from_utf8(body_bytes.to_vec()).unwrap();
-
-    let body = match serde_json::from_str::<UpgradeRequest>(&body_str) {
-        Ok(r) => r,
-        Err(e) => return Err((Into::into(e), StatusCode::BAD_REQUEST)),
-    };
-
-    // No sense in trying to upgrade to the same version.
-    let curr_version = compute.pgversion.clone();
-    let new_version = body.pg_version;
-    if curr_version == new_version {
-        return Err((
-            anyhow::anyhow!("cannot upgrade endpoint to the same version"),
-            StatusCode::UNPROCESSABLE_ENTITY,
-        ));
-    }
-
-    // Check that we are in the running state before trying to upgrade.
-    match compute.get_status() {
-        ComputeStatus::Prepared => (),
-        ComputeStatus::Upgrading => {
-            return Err((
-                anyhow::anyhow!("upgrade already in progress"),
-                StatusCode::CONFLICT,
-            ));
-        }
-        _ => {
-            return Err((
-                anyhow::anyhow!("expected compute to be in the prepared state"),
-                StatusCode::CONFLICT,
-            ));
-        }
-    }
-
-    compute.set_status(ComputeStatus::Upgrading);
-
-    let c = compute.clone();
-    tokio::spawn(async move {
-        if let Err(e) = c.upgrade(&new_version).await {
-            error!("Failed to upgrade database: {}", e);
-        }
-
-        c.set_status(ComputeStatus::Running);
-    });
-
-    Ok(())
-}
-
 // Main Hyper HTTP server function that runs it and blocks waiting on it forever.
 #[tokio::main]
 async fn serve(port: u16, state: Arc<ComputeNode>) {

compute_tools/src/http/openapi_spec.yaml

@@ -212,21 +212,6 @@ paths:
             application/json:
               schema:
                 $ref: "#/components/schemas/GenericError"
-  /upgrade:
-    post:
-      tags:
-        - Upgrade
-      summary: Upgrade project to another Postgres major version.
-      operationId: upgradePostgres
-      responses:
-        202:
-          description: Upgrade request in progress.
-        409:
-          description: Upgrade already in progress.
-        422:
-          description: Upgrade request could not be processed.
-        500:
-          description: Upgrade request failed.
 
   /terminate:
     post:

control_plane/src/bin/neon_local.rs

@@ -640,8 +640,6 @@ async fn handle_timeline(timeline_match: &ArgMatches, env: &mut local_env::Local
         }
         Some(("branch", branch_match)) => {
             let tenant_id = get_tenant_id(branch_match, env)?;
-            let new_timeline_id =
-                parse_timeline_id(branch_match)?.unwrap_or(TimelineId::generate());
             let new_branch_name = branch_match
                 .get_one::<String>("branch-name")
                 .ok_or_else(|| anyhow!("No branch name provided"))?;
@@ -660,6 +658,7 @@ async fn handle_timeline(timeline_match: &ArgMatches, env: &mut local_env::Local
                 .map(|lsn_str| Lsn::from_str(lsn_str))
                 .transpose()
                 .context("Failed to parse ancestor start Lsn from the request")?;
+            let new_timeline_id = TimelineId::generate();
             let storage_controller = StorageController::from_env(env);
             let create_req = TimelineCreateRequest {
                 new_timeline_id,
@@ -798,19 +797,14 @@ async fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Re
                 .get_one::<bool>("hot-standby")
                 .copied()
                 .unwrap_or(false);
-            let upgrade_only = sub_args
-                .get_one::<bool>("upgrade-only")
-                .copied()
-                .unwrap_or(false);
+
             let allow_multiple = sub_args.get_flag("allow-multiple");
 
-            let mode = match (lsn, hot_standby, upgrade_only) {
-                (Some(lsn), false, false) => ComputeMode::Static(lsn),
-                (None, true, false) => ComputeMode::Replica,
-                (None, false, false) => ComputeMode::Primary,
-                (None, false, true) => ComputeMode::Upgrade,
-                // Seeing this message means we aren't setting conflicts_with on clap arguments.
-                _ => anyhow::bail!("Invalid command line invocation"),
+            let mode = match (lsn, hot_standby) {
+                (Some(lsn), false) => ComputeMode::Static(lsn),
+                (None, true) => ComputeMode::Replica,
+                (None, false) => ComputeMode::Primary,
+                (Some(_), true) => anyhow::bail!("cannot specify both lsn and hot-standby"),
             };
 
             match (mode, hot_standby) {
@@ -1506,8 +1500,7 @@ fn cli() -> Command {
     let lsn_arg = Arg::new("lsn")
         .long("lsn")
         .help("Specify Lsn on the timeline to start from. By default, end of the timeline would be used.")
-        .required(false)
-        .conflicts_with("hot-standby");
+        .required(false);
 
     let hot_standby_arg = Arg::new("hot-standby")
         .value_parser(value_parser!(bool))
@@ -1577,6 +1570,7 @@ fn cli() -> Command {
                         .value_parser(value_parser!(PathBuf))
                         .value_name("config")
                 )
+                .arg(pg_version_arg.clone())
                 .arg(force_arg)
         )
         .subcommand(
@@ -1589,7 +1583,6 @@ fn cli() -> Command {
             .subcommand(Command::new("branch")
                 .about("Create a new timeline, using another timeline as a base, copying its data")
                 .arg(tenant_id_arg.clone())
-                .arg(timeline_id_arg.clone())
                 .arg(branch_name_arg.clone())
                 .arg(Arg::new("ancestor-branch-name").long("ancestor-branch-name")
                     .help("Use last Lsn of another timeline (and its data) as base when creating the new timeline. The timeline gets resolved by its branch name.").required(false))
@@ -1724,18 +1717,6 @@ fn cli() -> Command {
                     .arg(hot_standby_arg.clone())
                     .arg(update_catalog)
                     .arg(allow_multiple.clone())
-                    .arg(
-                        Arg::new("upgrade-only")
-                            .help("Mark this compute as an upgrade compute")
-                            .long("upgrade-only")
-                            .action(ArgAction::SetTrue)
-                            .conflicts_with_all(&[
-                                "config-only",
-                                "hot-standby",
-                                // Perhaps we could offer upgrades at a specific LSN in the future.
-                                "lsn",
-                            ])
-                    )
                 )
                 .subcommand(Command::new("start")
                     .about("Start postgres.\n If the endpoint doesn't exist yet, it is created.")

control_plane/src/endpoint.rs

@@ -182,7 +182,6 @@ impl ComputeControlPlane {
         tenant_id: TenantId,
         timeline_id: TimelineId,
     ) -> Result<()> {
-        // TODO: It really feels like I need to do some protection here
         if matches!(mode, ComputeMode::Primary) {
             // this check is not complete, as you could have a concurrent attempt at
             // creating another primary, both reading the state before checking it here,
@@ -394,7 +393,6 @@ impl Endpoint {
                     conf.append("recovery_prefetch", "off");
                 }
             }
-            ComputeMode::Upgrade => {}
         }
 
         Ok(conf)
@@ -504,7 +502,7 @@ impl Endpoint {
     /// Map safekeepers ids to the actual connection strings.
     fn build_safekeepers_connstrs(&self, sk_ids: Vec<NodeId>) -> Result<Vec<String>> {
         let mut safekeeper_connstrings = Vec::new();
-        if matches!(self.mode, ComputeMode::Primary | ComputeMode::Upgrade) {
+        if self.mode == ComputeMode::Primary {
             for sk_id in sk_ids {
                 let sk = self
                     .env
@@ -626,16 +624,13 @@ impl Endpoint {
                 self.endpoint_path().join("spec.json").to_str().unwrap(),
             ])
             .args([
-                "--pgroot",
-                &self
-                    .env
-                    .pg_distrib_dir
-                    .clone()
-                    .into_os_string()
-                    .into_string()
+                "--pgbin",
+                self.env
+                    .pg_bin_dir(self.pg_version)?
+                    .join("postgres")
+                    .to_str()
                     .unwrap(),
             ])
-            .args(["--pgversion", &self.pg_version.to_string()])
             .stdin(std::process::Stdio::null())
             .stderr(logfile.try_clone()?)
             .stdout(logfile);
@@ -679,7 +674,7 @@ impl Endpoint {
                             }
                             // keep retrying
                         }
-                        ComputeStatus::Running | ComputeStatus::Prepared => {
+                        ComputeStatus::Running => {
                             // All good!
                             break;
                         }
@@ -693,7 +688,6 @@ impl Endpoint {
                             );
                         }
                         ComputeStatus::Empty
-                        | ComputeStatus::Upgrading
                         | ComputeStatus::ConfigurationPending
                         | ComputeStatus::Configuration
                         | ComputeStatus::TerminationPending

libs/compute_api/src/requests.rs

@@ -12,9 +12,3 @@ use serde::Deserialize;
 pub struct ConfigurationRequest {
     pub spec: ComputeSpec,
 }
-
-/// Request body of the /upgrade API
-#[derive(Deserialize, Debug)]
-pub struct UpgradeRequest {
-    pub pg_version: String,
-}

libs/compute_api/src/responses.rs

@@ -44,15 +44,10 @@ pub enum ComputeStatus {
     // Compute node has spec and initial startup and
     // configuration is in progress.
     Init,
-    // Compute has been prepared, meaning that remote extensions have been
-    // downloaded and the data directory has been prepared.
-    Prepared,
     // Compute is configured and running.
     Running,
     // New spec is being applied.
     Configuration,
-    // Compute is upgrading Postgres.
-    Upgrading,
     // Either startup or configuration failed,
     // compute will exit soon or is waiting for
     // control-plane to terminate it.

libs/compute_api/src/spec.rs

@@ -199,11 +199,6 @@ pub enum ComputeMode {
     /// Future versions may want to distinguish between replicas with hot standby
     /// feedback and other kinds of replication configurations.
     Replica,
-    /// An upgrade-only node
-    ///
-    /// This node will not accept remote Postgres connections. It's only
-    /// purpose is to upgrade a timeline.
-    Upgrade,
 }
 
 #[derive(Clone, Debug, Default, Deserialize, Serialize)]

libs/pageserver_api/src/models/utilization.rs

@@ -89,19 +89,8 @@ impl PageserverUtilization {
 
     /// If a node is currently hosting more work than it can comfortably handle.  This does not indicate that
     /// it will fail, but it is a strong signal that more work should not be added unless there is no alternative.
-    ///
-    /// When a node is overloaded, we may override soft affinity preferences and do things like scheduling
-    /// into a node in a less desirable AZ, if all the nodes in the preferred AZ are overloaded.
     pub fn is_overloaded(score: RawScore) -> bool {
-        // Why the factor of two?  This is unscientific but reflects behavior of real systems:
-        // - In terms of shard counts, a node's preferred max count is a soft limit intended to keep
-        //   startup and housekeeping jobs nice and responsive.  We can go to double this limit if needed
-        //   until some more nodes are deployed.
-        // - In terms of disk space, the node's utilization heuristic assumes every tenant needs to
-        //   hold its biggest timeline fully on disk, which is tends to be an over estimate when
-        //   some tenants are very idle and have dropped layers from disk.  In practice going up to
-        //   double is generally better than giving up and scheduling in a sub-optimal AZ.
-        score >= 2 * Self::UTILIZATION_FULL
+        score >= Self::UTILIZATION_FULL
     }
 
     pub fn adjust_shard_count_max(&mut self, shard_count: u32) {

libs/postgres_backend/src/lib.rs

@@ -81,16 +81,17 @@ pub fn is_expected_io_error(e: &io::Error) -> bool {
     )
 }
 
+#[async_trait::async_trait]
 pub trait Handler<IO> {
     /// Handle single query.
     /// postgres_backend will issue ReadyForQuery after calling this (this
     /// might be not what we want after CopyData streaming, but currently we don't
     /// care). It will also flush out the output buffer.
-    fn process_query(
+    async fn process_query(
         &mut self,
         pgb: &mut PostgresBackend<IO>,
         query_string: &str,
-    ) -> impl Future<Output = Result<(), QueryError>>;
+    ) -> Result<(), QueryError>;
 
     /// Called on startup packet receival, allows to process params.
     ///

libs/postgres_backend/tests/simple_select.rs

@@ -23,6 +23,7 @@ async fn make_tcp_pair() -> (TcpStream, TcpStream) {
 
 struct TestHandler {}
 
+#[async_trait::async_trait]
 impl<IO: AsyncRead + AsyncWrite + Unpin + Send> Handler<IO> for TestHandler {
     // return single col 'hey' for any query
     async fn process_query(

libs/postgres_ffi/src/pg_constants.rs

@@ -167,16 +167,7 @@ pub const RM_RELMAP_ID: u8 = 7;
 pub const RM_STANDBY_ID: u8 = 8;
 pub const RM_HEAP2_ID: u8 = 9;
 pub const RM_HEAP_ID: u8 = 10;
-pub const RM_BTREE_ID: u8 = 11;
-pub const RM_HASH_ID: u8 = 12;
-pub const RM_GIN_ID: u8 = 13;
-pub const RM_GIST_ID: u8 = 14;
-pub const RM_SEQ_ID: u8 = 15;
-pub const RM_SPGIST_ID: u8 = 16;
-pub const RM_BRIN_ID: u8 = 17;
-pub const RM_COMMIT_TS_ID: u8 = 18;
 pub const RM_REPLORIGIN_ID: u8 = 19;
-pub const RM_GENERIC_ID: u8 = 20;
 pub const RM_LOGICALMSG_ID: u8 = 21;
 
 // from neon_rmgr.h
@@ -190,27 +181,9 @@ pub const XLOG_NEON_HEAP_UPDATE: u8 = 0x20;
 pub const XLOG_NEON_HEAP_HOT_UPDATE: u8 = 0x30;
 pub const XLOG_NEON_HEAP_LOCK: u8 = 0x40;
 pub const XLOG_NEON_HEAP_MULTI_INSERT: u8 = 0x50;
-pub const XLOG_NEON_FILE: u8 = 0x60;
 
 pub const XLOG_NEON_HEAP_VISIBLE: u8 = 0x40;
 
-#[repr(C)]
-#[derive(Debug)]
-pub enum XlNeonFileFiletype {
-    UPGRADE_TARBALL,
-}
-
-impl TryFrom<u8> for XlNeonFileFiletype {
-    type Error = ();
-
-    fn try_from(value: u8) -> Result<Self, Self::Error> {
-        match value {
-            0 => Ok(XlNeonFileFiletype::UPGRADE_TARBALL),
-            _ => Err(()),
-        }
-    }
-}
-
 // from xlogreader.h
 pub const XLR_INFO_MASK: u8 = 0x0F;
 pub const XLR_RMGR_INFO_MASK: u8 = 0xF0;

pageserver/src/import_datadir.rs

@@ -506,10 +506,6 @@ async fn import_file(
         return Ok(None);
     }
 
-    if file_name == "pg_internal.init" {
-        return Ok(None);
-    }
-
     if file_path.starts_with("global") {
         let spcnode = postgres_ffi::pg_constants::GLOBALTABLESPACE_OID;
         let dbnode = 0;

pageserver/src/page_service.rs

@@ -1199,6 +1199,7 @@ impl PageServerHandler {
     }
 }
 
+#[async_trait::async_trait]
 impl<IO> postgres_backend::Handler<IO> for PageServerHandler
 where
     IO: AsyncRead + AsyncWrite + Send + Sync + Unpin,

pageserver/src/pgdatadir_mapping.rs

@@ -1205,13 +1205,6 @@ impl<'a> DatadirModification<'a> {
         img: Bytes,
     ) -> anyhow::Result<()> {
         anyhow::ensure!(rel.relnode != 0, RelationError::InvalidRelnode);
-        let key = rel_block_to_key(rel, blknum);
-        if !key.is_valid_key_on_write_path() {
-            anyhow::bail!(
-                "the request contains data not supported by pageserver at {}",
-                key
-            );
-        }
         self.put(rel_block_to_key(rel, blknum), Value::Image(img));
         Ok(())
     }
@@ -1223,34 +1216,14 @@ impl<'a> DatadirModification<'a> {
         blknum: BlockNumber,
         img: Bytes,
     ) -> anyhow::Result<()> {
-        let key = slru_block_to_key(kind, segno, blknum);
-        if !key.is_valid_key_on_write_path() {
-            anyhow::bail!(
-                "the request contains data not supported by pageserver at {}",
-                key
-            );
-        }
-        self.put(key, Value::Image(img));
+        self.put(slru_block_to_key(kind, segno, blknum), Value::Image(img));
         Ok(())
     }
 
-    pub(crate) fn put_rel_page_image_zero(
-        &mut self,
-        rel: RelTag,
-        blknum: BlockNumber,
-    ) -> anyhow::Result<()> {
-        anyhow::ensure!(rel.relnode != 0, RelationError::InvalidRelnode);
-        let key = rel_block_to_key(rel, blknum);
-        if !key.is_valid_key_on_write_path() {
-            anyhow::bail!(
-                "the request contains data not supported by pageserver: {} @ {}",
-                key,
-                self.lsn
-            );
-        }
-        self.pending_zero_data_pages.insert(key.to_compact());
+    pub(crate) fn put_rel_page_image_zero(&mut self, rel: RelTag, blknum: BlockNumber) {
+        self.pending_zero_data_pages
+            .insert(rel_block_to_key(rel, blknum).to_compact());
         self.pending_bytes += ZERO_PAGE.len();
-        Ok(())
     }
 
     pub(crate) fn put_slru_page_image_zero(
@@ -1258,18 +1231,10 @@ impl<'a> DatadirModification<'a> {
         kind: SlruKind,
         segno: u32,
         blknum: BlockNumber,
-    ) -> anyhow::Result<()> {
-        let key = slru_block_to_key(kind, segno, blknum);
-        if !key.is_valid_key_on_write_path() {
-            anyhow::bail!(
-                "the request contains data not supported by pageserver: {} @ {}",
-                key,
-                self.lsn
-            );
-        }
-        self.pending_zero_data_pages.insert(key.to_compact());
+    ) {
+        self.pending_zero_data_pages
+            .insert(slru_block_to_key(kind, segno, blknum).to_compact());
         self.pending_bytes += ZERO_PAGE.len();
-        Ok(())
     }
 
     /// Call this at the end of each WAL record.

pageserver/src/tenant/timeline.rs

@@ -14,7 +14,7 @@ mod walreceiver;
 use anyhow::{anyhow, bail, ensure, Context, Result};
 use arc_swap::ArcSwap;
 use bytes::Bytes;
-use camino::{Utf8Path, Utf8PathBuf};
+use camino::Utf8Path;
 use chrono::{DateTime, Utc};
 use enumset::EnumSet;
 use fail::fail_point;
@@ -4357,10 +4357,6 @@ impl Timeline {
         )
     }
 
-    pub fn get_path(&self) -> Utf8PathBuf {
-        self.conf.timelines_path(&self.tenant_shard_id)
-    }
-
     /// Detach this timeline from its ancestor by copying all of ancestors layers as this
     /// Timelines layers up to the ancestor_lsn.
     ///

pageserver/src/walingest.rs

@@ -24,7 +24,6 @@
 use std::time::Duration;
 use std::time::SystemTime;
 
-use camino::Utf8PathBuf;
 use pageserver_api::shard::ShardIdentity;
 use postgres_ffi::{dispatch_pgversion, enum_pgversion, enum_pgversion_dispatch, TimestampTz};
 use postgres_ffi::{fsm_logical_to_physical, page_is_new, page_set_lsn};
@@ -34,10 +33,8 @@ use bytes::{Buf, Bytes, BytesMut};
 use tracing::*;
 use utils::failpoint_support;
 use utils::rate_limit::RateLimit;
-use utils::zstd::extract_zst_tarball;
 
 use crate::context::RequestContext;
-use crate::import_datadir;
 use crate::metrics::WAL_INGEST;
 use crate::pgdatadir_mapping::{DatadirModification, Version};
 use crate::span::debug_assert_current_span_has_tenant_and_timeline_id;
@@ -72,9 +69,7 @@ impl CheckPoint {
     }
 }
 
-pub struct WalIngest<'t> {
-    timeline: &'t Timeline,
-    timeline_path: Utf8PathBuf,
+pub struct WalIngest {
     shard: ShardIdentity,
     checkpoint: CheckPoint,
     checkpoint_modified: bool,
@@ -87,12 +82,12 @@ struct WarnIngestLag {
     timestamp_invalid_msg_ratelimit: RateLimit,
 }
 
-impl<'t> WalIngest<'t> {
+impl WalIngest {
     pub async fn new(
-        timeline: &'t Timeline,
+        timeline: &Timeline,
         startpoint: Lsn,
         ctx: &RequestContext,
-    ) -> anyhow::Result<WalIngest<'t>> {
+    ) -> anyhow::Result<WalIngest> {
         // Fetch the latest checkpoint into memory, so that we can compare with it
         // quickly in `ingest_record` and update it when it changes.
         let checkpoint_bytes = timeline.get_checkpoint(startpoint, ctx).await?;
@@ -105,8 +100,6 @@ impl<'t> WalIngest<'t> {
         });
 
         Ok(WalIngest {
-            timeline,
-            timeline_path: timeline.get_path(),
             shard: *timeline.get_shard_identity(),
             checkpoint,
             checkpoint_modified: false,
@@ -465,17 +458,6 @@ impl<'t> WalIngest<'t> {
                     modification.drop_replorigin(xlrec.node_id).await?
                 }
             }
-            pg_constants::RM_BTREE_ID
-            | pg_constants::RM_HASH_ID
-            | pg_constants::RM_GIN_ID
-            | pg_constants::RM_GIST_ID
-            | pg_constants::RM_SEQ_ID
-            | pg_constants::RM_SPGIST_ID
-            | pg_constants::RM_BRIN_ID
-            | pg_constants::RM_COMMIT_TS_ID
-            | pg_constants::RM_GENERIC_ID => {
-                // No special handling currently for these resource managers
-            }
             _x => {
                 // TODO: should probably log & fail here instead of blindly
                 // doing something without understanding the protocol
@@ -941,33 +923,6 @@ impl<'t> WalIngest<'t> {
         assert_eq!(decoded.xl_rmid, pg_constants::RM_NEON_ID);
 
         match pg_version {
-            15 => {
-                let info = decoded.xl_info;
-
-                match info {
-                    pg_constants::XLOG_NEON_FILE => {
-                        info!(
-                            "tristan: last_record_lsn={}",
-                            self.timeline.get_last_record_lsn()
-                        );
-                        let xlrec = v16::rm_neon::XlNeonFile::decode(buf);
-                        let pgdata_path = self.timeline_path.join("new-pgdata");
-
-                        extract_zst_tarball(&pgdata_path, &*xlrec.data).await?;
-
-                        let lsn = import_datadir::get_lsn_from_controlfile(&pgdata_path)?;
-                        info!("LSN from pg_upgraded controlfile: {lsn}");
-                        Box::pin(import_datadir::import_timeline_from_postgres_datadir(
-                            self.timeline,
-                            &pgdata_path,
-                            lsn,
-                            ctx,
-                        ))
-                        .await?;
-                    }
-                    _ => return Err(anyhow::anyhow!("Unknown XLOG xl_info field: {}", info)),
-                }
-            }
             16 => {
                 let info = decoded.xl_info & pg_constants::XLOG_HEAP_OPMASK;
 
@@ -1267,7 +1222,7 @@ impl<'t> WalIngest<'t> {
             if rec.blkno % pg_constants::SLOTS_PER_FSM_PAGE != 0 {
                 // Tail of last remaining FSM page has to be zeroed.
                 // We are not precise here and instead of digging in FSM bitmap format just clear the whole page.
-                modification.put_rel_page_image_zero(rel, fsm_physical_page_no)?;
+                modification.put_rel_page_image_zero(rel, fsm_physical_page_no);
                 fsm_physical_page_no += 1;
             }
             let nblocks = get_relsize(modification, rel, ctx).await?;
@@ -1289,7 +1244,7 @@ impl<'t> WalIngest<'t> {
             if rec.blkno % pg_constants::VM_HEAPBLOCKS_PER_PAGE != 0 {
                 // Tail of last remaining vm page has to be zeroed.
                 // We are not precise here and instead of digging in VM bitmap format just clear the whole page.
-                modification.put_rel_page_image_zero(rel, vm_page_no)?;
+                modification.put_rel_page_image_zero(rel, vm_page_no);
                 vm_page_no += 1;
             }
             let nblocks = get_relsize(modification, rel, ctx).await?;
@@ -1782,7 +1737,7 @@ impl<'t> WalIngest<'t> {
                     continue;
                 }
 
-                modification.put_rel_page_image_zero(rel, gap_blknum)?;
+                modification.put_rel_page_image_zero(rel, gap_blknum);
             }
         }
         Ok(())
@@ -1848,7 +1803,7 @@ impl<'t> WalIngest<'t> {
 
             // fill the gap with zeros
             for gap_blknum in old_nblocks..blknum {
-                modification.put_slru_page_image_zero(kind, segno, gap_blknum)?;
+                modification.put_slru_page_image_zero(kind, segno, gap_blknum);
             }
         }
         Ok(())

pageserver/src/walrecord.rs

@@ -526,29 +526,6 @@ pub mod v16 {
                 }
             }
         }
-
-        #[repr(C)]
-        #[derive(Debug)]
-        pub struct XlNeonFile {
-            pub filetype: u8,
-            pub size: u32,
-            pub data: Bytes,
-        }
-
-        impl XlNeonFile {
-            pub fn decode(buf: &mut Bytes) -> Self {
-                let filetype = buf.get_u8();
-                // Skip the padding
-                buf.advance(std::mem::size_of::<u8>() * 3);
-                let size = buf.get_u32_le();
-
-                Self {
-                    filetype,
-                    size,
-                    data: buf.copy_to_bytes(buf.remaining()),
-                }
-            }
-        }
     }
 }
 

pgxn/neon/Makefile

@@ -23,18 +23,7 @@ SHLIB_LINK_INTERNAL = $(libpq)
 SHLIB_LINK = -lcurl
 
 EXTENSION = neon
-DATA = \
-	neon--1.0.sql \
-	neon--1.0--1.1.sql \
-	neon--1.1--1.2.sql \
-	neon--1.2--1.3.sql \
-	neon--1.3--1.4.sql \
-	neon--1.4--1.5.sql \
-	neon--1.5--1.4.sql \
-	neon--1.4--1.3.sql \
-	neon--1.3--1.2.sql \
-	neon--1.2--1.1.sql \
-	neon--1.1--1.0.sql
+DATA = neon--1.0.sql neon--1.0--1.1.sql neon--1.1--1.2.sql neon--1.2--1.3.sql neon--1.3--1.2.sql neon--1.2--1.1.sql neon--1.1--1.0.sql  neon--1.3--1.4.sql neon--1.4--1.3.sql
 PGFILEDESC = "neon - cloud storage for PostgreSQL"
 
 EXTRA_CLEAN = \

pgxn/neon/neon--1.3--1.4.sql

@@ -7,7 +7,3 @@ LANGUAGE C PARALLEL SAFE;
 
 GRANT EXECUTE ON FUNCTION approximate_working_set_size_seconds(integer) TO pg_monitor;
 
-CREATE FUNCTION wal_log_file(path text)
-RETURNS pg_lsn
-AS 'MODULE_PATHNAME', 'wal_log_file'
-LANGUAGE C STRICT PARALLEL UNSAFE;

pgxn/neon/neon--1.4--1.5.sql

@@ -1,8 +0,0 @@
-\echo Use "ALTER EXTENSION neon UPDATE TO '1.5'" to load this file. \quit
-
-CREATE FUNCTION wal_log_file(path text)
-RETURNS pg_lsn
-AS 'MODULE_PATHNAME', 'wal_log_file'
-LANGUAGE C STRICT PARALLEL UNSAFE;
-
-GRANT EXECUTE ON FUNCTION wal_log_file TO pg_monitor;

pgxn/neon/neon--1.5--1.4.sql

@@ -1 +0,0 @@
-DROP FUNCTION IF EXISTS wal_log_file(text) CASCADE;

pgxn/neon/neon.c

@@ -11,8 +11,6 @@
 #include "postgres.h"
 #include "fmgr.h"
 
-#include <sys/stat.h>
-
 #include "miscadmin.h"
 #include "access/subtrans.h"
 #include "access/twophase.h"
@@ -31,19 +29,11 @@
 #include "tcop/tcopprot.h"
 #include "funcapi.h"
 #include "access/htup_details.h"
-#include "access/xloginsert.h"
 #include "utils/builtins.h"
 #include "utils/pg_lsn.h"
 #include "utils/guc.h"
 #include "utils/guc_tables.h"
-#include "utils/timeout.h"
 #include "utils/wait_event.h"
-#include "storage/ipc.h"
-#include "storage/proc.h"
-
-#if PG_MAJORVERSION_NUM >= 15
-#include "access/neon_xlog.h"
-#endif
 
 #include "extension_server.h"
 #include "neon.h"
@@ -639,17 +629,10 @@ ReportSearchPath(void)
 void
 _PG_init(void)
 {
-	const char *purpose;
-
-	purpose = getenv("NEON_PURPOSE");
-	if (purpose && strcmp(purpose, "upgrade") == 0)
-		return;
-
 	/*
 	 * Also load 'neon_rmgr'. This makes it unnecessary to list both 'neon'
 	 * and 'neon_rmgr' in shared_preload_libraries.
 	 */
-
 #if PG_VERSION_NUM >= 160000
 	load_file("$libdir/neon_rmgr", false);
 #endif
@@ -693,9 +676,6 @@ _PG_init(void)
 PG_FUNCTION_INFO_V1(pg_cluster_size);
 PG_FUNCTION_INFO_V1(backpressure_lsns);
 PG_FUNCTION_INFO_V1(backpressure_throttling_time);
-#if PG_MAJORVERSION_NUM >= 15
-PG_FUNCTION_INFO_V1(wal_log_file);
-#endif
 
 Datum
 pg_cluster_size(PG_FUNCTION_ARGS)
@@ -741,200 +721,3 @@ backpressure_throttling_time(PG_FUNCTION_ARGS)
 {
 	PG_RETURN_UINT64(BackpressureThrottlingTime());
 }
-
-#if PG_MAJORVERSION_NUM >= 15
-
-Datum
-wal_log_file(PG_FUNCTION_ARGS)
-{
-	int rc;
-	int fd;
-	ssize_t n;
-	text *path;
-	size_t off;
-	char *data;
-	short nargs;
-	struct stat st;
-	XLogRecPtr lsn;
-	size_t path_len;
-	xl_neon_file xlrec;
-	char file[MAXPGPATH];
-
-#if defined(WAL_DEBUG) && PG_MAJORVERSION_NUM < 16
-	bool wal_debug;
-#endif
-
-	path = PG_GETARG_TEXT_PP(0);
-	path_len = VARSIZE(path) - VARHDRSZ;
-
-	memcpy(file, VARDATA(path), path_len);
-	file[path_len] = '\0';
-
-	/* Get the size of the file. Note that stat(2) follows symlinks. */
-	rc = stat(file, &st);
-	if (rc != 0)
-		ereport(ERROR,
-				(errmsg("failed to get size of file (%s): %m", file)));
-
-	xlrec.size = (size_t) st.st_size;
-
-	fd = open(file, O_RDONLY);
-	if (fd == -1)
-		ereport(ERROR,
-				(errmsg("could not open %s: %m", file)));
-
-	/* If the file is too large, error out. */
-
-	data = palloc(xlrec.size);
-
-	/* Copy the file contents */
-	off = 0;
-	while (true) {
-		n = read(fd, data + off, xlrec.size - off);
-		if (n == EOF)
-		{
-			close(fd);
-			ereport(ERROR,
-					(errmsg("failed to read %s: %m", file)));
-		}
-
-		off += n;
-
-		if (xlrec.size - off == 0)
-			break;
-	}
-
-	close(fd);
-
-	xlrec.filetype = XL_NEON_FILE_UPGRADE_TARBALL;
-
-	XLogBeginInsert();
-	XLogRegisterData((char *) &xlrec, SizeOfNeonFile);
-	XLogRegisterData((char *) data, xlrec.size);
-
-	/*
-	 * We must turn debugging off on anything where the Neon RMGR is not
-	 * registered. Stash the original value for restoration later.
-	 */
-#if defined(WAL_DEBUG) && PG_MAJORVERSION_NUM < 16
-	wal_debug = XLOG_DEBUG;
-	XLOG_DEBUG = false;
-#endif
-
-	lsn = XLogInsert(RM_NEON_ID, XLOG_NEON_FILE);
-
-#if defined(WAL_DEBUG) && PG_MAJORVERSION_NUM < 16
-	XLOG_DEBUG = wal_debug;
-#endif
-
-	PG_RETURN_LSN(lsn);
-}
-
-#endif
-
-#if PG_MAJORVERSION_NUM >= 15
-
-/*
- * Entry point for `postgres --wal-log`.
- */
-PGDLLEXPORT void
-WalLog(int argc, char *argv[])
-{
-	int			rc;
-	int			fd;
-	off_t		off;
-	struct stat	st;
-	XLogRecPtr	lsn;
-	void	   *data;
-	xl_neon_file xlrec;
-	/* TODO: should this be PATH_MAX? should we require an absolute path? */
-	char		file[MAXPGPATH];
-
-#if defined(WAL_DEBUG) && PG_MAJORVERSION_NUM < 16
-	bool wal_debug;
-#endif
-
-	if (argc != 3)
-		ereport(ERROR, errmsg("wrong number of arguments passed to --wal-log"));
-
-	if (!realpath(argv[2], file))
-		ereport(ERROR, errmsg("failed to resolve path: %m"));
-
-	ereport(LOG, errmsg("writing %s to WAL", file));
-
-	ChangeToDataDir();
-	CreateDataDirLockFile(false);
-	LocalProcessControlFile(false);
-	InitializeMaxBackends();
-	CreateSharedMemoryAndSemaphores();
-	InitializeTimeouts();
-	InitProcess();
-	BaseInit();
-	CreateAuxProcessResourceOwner();
-	StartupXLOG();
-
-	/* Get the size of the file. Note that stat(2) follows symlinks. */
-	rc = stat(file, &st);
-	if (rc != 0)
-		ereport(ERROR,
-				(errmsg("failed to get size of file (%s): %m", file)));
-
-	xlrec.size = (size_t) st.st_size;
-
-	fd = open(file, O_RDONLY);
-	if (fd == -1)
-		ereport(ERROR,
-				(errmsg("could not open %s: %m", file)));
-
-	/* If the file is too large, error out. */
-
-	data = palloc(xlrec.size);
-
-	/* Copy the file contents */
-	off = 0;
-	while (true) {
-		ssize_t		n;
-
-		n = read(fd, data + off, xlrec.size - off);
-		if (n == EOF)
-		{
-			close(fd);
-			ereport(ERROR,
-					(errmsg("failed to read %s: %m", file)));
-		}
-
-		off += n;
-
-		if (xlrec.size - off == 0)
-			break;
-	}
-
-	close(fd);
-
-	xlrec.filetype = XL_NEON_FILE_UPGRADE_TARBALL;
-
-	/* ereport(LOG, errmsg("Current LSN: %X/%X" , LSN_FORMAT_ARGS(GetXLogWriteRecPtr()))); */
-
-	XLogBeginInsert();
-	XLogRegisterData((char *) &xlrec, SizeOfNeonFile);
-	XLogRegisterData(data, xlrec.size);
-
-	/*
-	 * We must turn debugging off on anything where the Neon RMGR is not
-	 * registered. Stash the original value for restoration later.
-	 */
-#if defined(WAL_DEBUG) && PG_MAJORVERSION_NUM < 16
-	wal_debug = XLOG_DEBUG;
-	XLOG_DEBUG = false;
-#endif
-
-	lsn = XLogInsert(RM_NEON_ID, XLOG_NEON_FILE);
-
-#if defined(WAL_DEBUG) && PG_MAJORVERSION_NUM < 16
-	XLOG_DEBUG = wal_debug;
-#endif
-
-	exit(0);
-}
-
-#endif

pgxn/neon/relsize_cache.c

@@ -78,16 +78,11 @@ neon_smgr_shmem_startup(void)
 	if (prev_shmem_startup_hook)
 		prev_shmem_startup_hook();
 
-	if (relsize_hash_size <= 0)
-		return;
-
 	LWLockAcquire(AddinShmemInitLock, LW_EXCLUSIVE);
 	relsize_ctl = (RelSizeHashControl *) ShmemInitStruct("relsize_hash", sizeof(RelSizeHashControl), &found);
 	if (!found)
 	{
-		elog(LOG, "neon_relsize: %d", relsize_hash_size);
 		relsize_lock = (LWLockId) GetNamedLWLockTranche("neon_relsize");
-		elog(LOG, "neon_relsize");
 		info.keysize = sizeof(RelTag);
 		info.entrysize = sizeof(RelSizeEntry);
 		relsize_hash = ShmemInitHash("neon_relsize",

pgxn/neon_rmgr/neon_rmgr.c

@@ -76,8 +76,6 @@ neon_rm_redo(XLogReaderState *record)
 		case XLOG_NEON_HEAP_MULTI_INSERT:
 			redo_neon_heap_multi_insert(record);
 			break;
-		case XLOG_NEON_FILE:
-			break;
 		default:
 			elog(PANIC, "neon_rm_redo: unknown op code %u", info);
 	}

pgxn/neon_rmgr/neon_rmgr_decode.c

@@ -57,8 +57,6 @@ neon_rm_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 			if (SnapBuildProcessChange(builder, xid, buf->origptr))
 				DecodeNeonMultiInsert(ctx, buf);
 			break;
-		case XLOG_NEON_FILE:
-			break;
 		default:
 			elog(ERROR, "unexpected RM_HEAP_ID record type: %u", info);
 			break;
@@ -403,4 +401,4 @@ DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tuple)
 }
 
 
-#endif
+#endif

pgxn/neon_rmgr/neon_rmgr_desc.c

@@ -134,16 +134,6 @@ neon_rm_desc(StringInfo buf, XLogReaderState *record)
 					   xlrec->ntuples, &offset_elem_desc, NULL);
 		}
 	}
-	else if (info == XLOG_NEON_FILE)
-	{
-		const xl_neon_file *xlrec = (xl_neon_file *) rec;
-		switch ((xl_neon_file_filetype) xlrec->filetype)
-		{
-			case XL_NEON_FILE_UPGRADE_TARBALL:
-				appendStringInfo(buf, "filetype: upgrade tarball, size: %zu", xlrec->size);
-				break;
-		}
-	}
 }
 
 const char *
@@ -183,9 +173,6 @@ neon_rm_identify(uint8 info)
 		case XLOG_NEON_HEAP_MULTI_INSERT | XLOG_NEON_INIT_PAGE:
 			id = "MULTI_INSERT+INIT";
 			break;
-		case XLOG_NEON_FILE:
-			id = "FILE";
-			break;
 	}
 
 	return id;

proxy/src/cache/endpoints.rs

@@ -242,6 +242,6 @@ mod tests {
     #[test]
     fn test() {
         let s = "{\"branch_created\":null,\"endpoint_created\":{\"endpoint_id\":\"ep-rapid-thunder-w0qqw2q9\"},\"project_created\":null,\"type\":\"endpoint_created\"}";
-        serde_json::from_str::<ControlPlaneEventKey>(s).unwrap();
+        let _: ControlPlaneEventKey = serde_json::from_str(s).unwrap();
     }
 }

proxy/src/console/messages.rs

@@ -395,7 +395,7 @@ mod tests {
                 }
             }
         });
-        serde_json::from_str::<KickSession<'_>>(&json.to_string())?;
+        let _: KickSession<'_> = serde_json::from_str(&json.to_string())?;
 
         Ok(())
     }
@@ -403,7 +403,7 @@ mod tests {
     #[test]
     fn parse_db_info() -> anyhow::Result<()> {
         // with password
-        serde_json::from_value::<DatabaseInfo>(json!({
+        let _: DatabaseInfo = serde_json::from_value(json!({
             "host": "localhost",
             "port": 5432,
             "dbname": "postgres",
@@ -413,7 +413,7 @@ mod tests {
         }))?;
 
         // without password
-        serde_json::from_value::<DatabaseInfo>(json!({
+        let _: DatabaseInfo = serde_json::from_value(json!({
             "host": "localhost",
             "port": 5432,
             "dbname": "postgres",
@@ -422,7 +422,7 @@ mod tests {
         }))?;
 
         // new field (forward compatibility)
-        serde_json::from_value::<DatabaseInfo>(json!({
+        let _: DatabaseInfo = serde_json::from_value(json!({
             "host": "localhost",
             "port": 5432,
             "dbname": "postgres",
@@ -441,7 +441,7 @@ mod tests {
             "address": "0.0.0.0",
             "aux": dummy_aux(),
         });
-        serde_json::from_str::<WakeCompute>(&json.to_string())?;
+        let _: WakeCompute = serde_json::from_str(&json.to_string())?;
         Ok(())
     }
 
@@ -451,18 +451,18 @@ mod tests {
         let json = json!({
             "role_secret": "secret",
         });
-        serde_json::from_str::<GetRoleSecret>(&json.to_string())?;
+        let _: GetRoleSecret = serde_json::from_str(&json.to_string())?;
         let json = json!({
             "role_secret": "secret",
             "allowed_ips": ["8.8.8.8"],
         });
-        serde_json::from_str::<GetRoleSecret>(&json.to_string())?;
+        let _: GetRoleSecret = serde_json::from_str(&json.to_string())?;
         let json = json!({
             "role_secret": "secret",
             "allowed_ips": ["8.8.8.8"],
             "project_id": "project",
         });
-        serde_json::from_str::<GetRoleSecret>(&json.to_string())?;
+        let _: GetRoleSecret = serde_json::from_str(&json.to_string())?;
 
         Ok(())
     }

proxy/src/console/mgmt.rs

@@ -78,7 +78,7 @@ pub(crate) type ComputeReady = DatabaseInfo;
 
 // TODO: replace with an http-based protocol.
 struct MgmtHandler;
-
+#[async_trait::async_trait]
 impl postgres_backend::Handler<tokio::net::TcpStream> for MgmtHandler {
     async fn process_query(
         &mut self,

proxy/src/context.rs

@@ -6,7 +6,7 @@ use pq_proto::StartupMessageParams;
 use smol_str::SmolStr;
 use std::net::IpAddr;
 use tokio::sync::mpsc;
-use tracing::{debug, field::display, info, info_span, Span};
+use tracing::{field::display, info, info_span, Span};
 use try_lock::TryLock;
 use uuid::Uuid;
 
@@ -362,9 +362,7 @@ impl RequestMonitoringInner {
                 });
         }
         if let Some(tx) = self.sender.take() {
-            tx.send(RequestData::from(&*self))
-                .inspect_err(|e| debug!("tx send failed: {e}"))
-                .ok();
+            let _: Result<(), _> = tx.send(RequestData::from(&*self));
         }
     }
 
@@ -373,9 +371,7 @@ impl RequestMonitoringInner {
         // Here we log the length of the session.
         self.disconnect_timestamp = Some(Utc::now());
         if let Some(tx) = self.disconnect_sender.take() {
-            tx.send(RequestData::from(&*self))
-                .inspect_err(|e| debug!("tx send failed: {e}"))
-                .ok();
+            let _: Result<(), _> = tx.send(RequestData::from(&*self));
         }
     }
 }

proxy/src/context/parquet.rs

@@ -290,7 +290,7 @@ async fn worker_inner(
     }
 
     if !w.flushed_row_groups().is_empty() {
-        let _rtchk: Writer<BytesMut> = upload_parquet(w, len, &storage).await?;
+        let _: Writer<BytesMut> = upload_parquet(w, len, &storage).await?;
     }
 
     Ok(())

proxy/src/lib.rs

@@ -3,7 +3,7 @@
 #![deny(
     deprecated,
     future_incompatible,
-    let_underscore,
+    // TODO: consider let_underscore
     nonstandard_style,
     rust_2024_compatibility
 )]

proxy/src/proxy/tests.rs

@@ -268,7 +268,7 @@ async fn keepalive_is_inherited() -> anyhow::Result<()> {
         anyhow::Ok(keepalive)
     });
 
-    TcpStream::connect(("127.0.0.1", port)).await?;
+    let _ = TcpStream::connect(("127.0.0.1", port)).await?;
     assert!(t.await??, "keepalive should be inherited");
 
     Ok(())

proxy/src/redis/connection_with_credentials_provider.rs

@@ -6,7 +6,7 @@ use redis::{
     ConnectionInfo, IntoConnectionInfo, RedisConnectionInfo, RedisResult,
 };
 use tokio::task::JoinHandle;
-use tracing::{debug, error, info};
+use tracing::{error, info};
 
 use super::elasticache::CredentialsProvider;
 
@@ -109,10 +109,7 @@ impl ConnectionWithCredentialsProvider {
             let credentials_provider = credentials_provider.clone();
             let con2 = con.clone();
             let f = tokio::spawn(async move {
-                Self::keep_connection(con2, credentials_provider)
-                    .await
-                    .inspect_err(|e| debug!("keep_connection failed: {e}"))
-                    .ok();
+                let _ = Self::keep_connection(con2, credentials_provider).await;
             });
             self.refresh_token_task = Some(f);
         }

proxy/src/stream.rs

@@ -12,7 +12,6 @@ use std::{io, task};
 use thiserror::Error;
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tokio_rustls::server::TlsStream;
-use tracing::debug;
 
 /// Stream wrapper which implements libpq's protocol.
 ///
@@ -139,10 +138,9 @@ impl<S: AsyncWrite + Unpin> PqStream<S> {
         );
 
         // already error case, ignore client IO error
-        self.write_message(&BeMessage::ErrorResponse(msg, None))
-            .await
-            .inspect_err(|e| debug!("write_message failed: {e}"))
-            .ok();
+        let _: Result<_, std::io::Error> = self
+            .write_message(&BeMessage::ErrorResponse(msg, None))
+            .await;
 
         Err(ReportedError {
             source: anyhow::anyhow!(msg),
@@ -166,10 +164,9 @@ impl<S: AsyncWrite + Unpin> PqStream<S> {
         );
 
         // already error case, ignore client IO error
-        self.write_message(&BeMessage::ErrorResponse(&msg, None))
-            .await
-            .inspect_err(|e| debug!("write_message failed: {e}"))
-            .ok();
+        let _: Result<_, std::io::Error> = self
+            .write_message(&BeMessage::ErrorResponse(&msg, None))
+            .await;
 
         Err(ReportedError {
             source: anyhow::anyhow!(error),

proxy/src/url.rs

@@ -57,7 +57,7 @@ mod tests {
     fn bad_url() {
         let url = "test:foobar";
         url.parse::<url::Url>().expect("unexpected parsing failure");
-        url.parse::<ApiUrl>().expect_err("should not parse");
+        let _ = url.parse::<ApiUrl>().expect_err("should not parse");
     }
 
     #[test]

safekeeper/src/handler.rs

@@ -2,7 +2,6 @@
 //! protocol commands.
 
 use anyhow::Context;
-use std::future::Future;
 use std::str::{self, FromStr};
 use std::sync::Arc;
 use tokio::io::{AsyncRead, AsyncWrite};
@@ -96,6 +95,7 @@ fn cmd_to_string(cmd: &SafekeeperPostgresCommand) -> &str {
     }
 }
 
+#[async_trait::async_trait]
 impl<IO: AsyncRead + AsyncWrite + Unpin + Send> postgres_backend::Handler<IO>
     for SafekeeperPostgresHandler
 {
@@ -197,51 +197,49 @@ impl<IO: AsyncRead + AsyncWrite + Unpin + Send> postgres_backend::Handler<IO>
         Ok(())
     }
 
-    fn process_query(
+    async fn process_query(
         &mut self,
         pgb: &mut PostgresBackend<IO>,
         query_string: &str,
-    ) -> impl Future<Output = Result<(), QueryError>> {
-        Box::pin(async move {
-            if query_string
-                .to_ascii_lowercase()
-                .starts_with("set datestyle to ")
-            {
-                // important for debug because psycopg2 executes "SET datestyle TO 'ISO'" on connect
-                pgb.write_message_noflush(&BeMessage::CommandComplete(b"SELECT 1"))?;
-                return Ok(());
+    ) -> Result<(), QueryError> {
+        if query_string
+            .to_ascii_lowercase()
+            .starts_with("set datestyle to ")
+        {
+            // important for debug because psycopg2 executes "SET datestyle TO 'ISO'" on connect
+            pgb.write_message_noflush(&BeMessage::CommandComplete(b"SELECT 1"))?;
+            return Ok(());
+        }
+
+        let cmd = parse_cmd(query_string)?;
+        let cmd_str = cmd_to_string(&cmd);
+
+        let _guard = PG_QUERIES_GAUGE.with_label_values(&[cmd_str]).guard();
+
+        info!("got query {:?}", query_string);
+
+        let tenant_id = self.tenant_id.context("tenantid is required")?;
+        let timeline_id = self.timeline_id.context("timelineid is required")?;
+        self.check_permission(Some(tenant_id))?;
+        self.ttid = TenantTimelineId::new(tenant_id, timeline_id);
+
+        match cmd {
+            SafekeeperPostgresCommand::StartWalPush => {
+                self.handle_start_wal_push(pgb)
+                    .instrument(info_span!("WAL receiver"))
+                    .await
             }
-
-            let cmd = parse_cmd(query_string)?;
-            let cmd_str = cmd_to_string(&cmd);
-
-            let _guard = PG_QUERIES_GAUGE.with_label_values(&[cmd_str]).guard();
-
-            info!("got query {:?}", query_string);
-
-            let tenant_id = self.tenant_id.context("tenantid is required")?;
-            let timeline_id = self.timeline_id.context("timelineid is required")?;
-            self.check_permission(Some(tenant_id))?;
-            self.ttid = TenantTimelineId::new(tenant_id, timeline_id);
-
-            match cmd {
-                SafekeeperPostgresCommand::StartWalPush => {
-                    self.handle_start_wal_push(pgb)
-                        .instrument(info_span!("WAL receiver"))
-                        .await
-                }
-                SafekeeperPostgresCommand::StartReplication { start_lsn, term } => {
-                    self.handle_start_replication(pgb, start_lsn, term)
-                        .instrument(info_span!("WAL sender"))
-                        .await
-                }
-                SafekeeperPostgresCommand::IdentifySystem => self.handle_identify_system(pgb).await,
-                SafekeeperPostgresCommand::TimelineStatus => self.handle_timeline_status(pgb).await,
-                SafekeeperPostgresCommand::JSONCtrl { ref cmd } => {
-                    handle_json_ctrl(self, pgb, cmd).await
-                }
+            SafekeeperPostgresCommand::StartReplication { start_lsn, term } => {
+                self.handle_start_replication(pgb, start_lsn, term)
+                    .instrument(info_span!("WAL sender"))
+                    .await
             }
-        })
+            SafekeeperPostgresCommand::IdentifySystem => self.handle_identify_system(pgb).await,
+            SafekeeperPostgresCommand::TimelineStatus => self.handle_timeline_status(pgb).await,
+            SafekeeperPostgresCommand::JSONCtrl { ref cmd } => {
+                handle_json_ctrl(self, pgb, cmd).await
+            }
+        }
     }
 }
 

safekeeper/src/safekeeper.rs

@@ -580,15 +580,15 @@ where
          * because safekeepers parse WAL headers and the format
          * may change between versions.
          */
-        // if msg.pg_version / 10000 != self.state.server.pg_version / 10000
-        //     && self.state.server.pg_version != UNKNOWN_SERVER_VERSION
-        // {
-        //     bail!(
-        //         "incompatible server version {}, expected {}",
-        //         msg.pg_version,
-        //         self.state.server.pg_version
-        //     );
-        // }
+        if msg.pg_version / 10000 != self.state.server.pg_version / 10000
+            && self.state.server.pg_version != UNKNOWN_SERVER_VERSION
+        {
+            bail!(
+                "incompatible server version {}, expected {}",
+                msg.pg_version,
+                self.state.server.pg_version
+            );
+        }
 
         if msg.tenant_id != self.state.tenant_id {
             bail!(

test.sh

@@ -1,12 +0,0 @@
-#!/bin/sh
-
-cargo neon endpoint stop ep-main 2>/dev/null
-kill $(pgrep compute_ctl)
-cargo neon stop 2>/dev/null
-rm -rf .neon
-cargo neon init
-cargo neon start
-cargo neon tenant create --pg-version 15 --set-default
-cargo neon endpoint create --pg-version 15 --upgrade-only
-cargo neon endpoint start ep-main
-curl -i -X POST http://localhost:55433/upgrade -d '{"pg_version": "16"}'

test_runner/fixtures/neon_fixtures.py

@@ -553,6 +553,10 @@ class NeonEnvBuilder:
         self.env = NeonEnv(self)
         return self.env
 
+    def start(self):
+        assert self.env is not None, "environment is not already initialized, call init() first"
+        self.env.start()
+
     def init_start(
         self,
         initial_tenant_conf: Optional[Dict[str, Any]] = None,
@@ -568,7 +572,7 @@ class NeonEnvBuilder:
         Configuring pageserver with remote storage is now the default. There will be a warning if pageserver is created without one.
         """
         env = self.init_configs(default_remote_storage_if_missing=default_remote_storage_if_missing)
-        env.start()
+        self.start()
 
         # Prepare the default branch to start the postgres on later.
         # Pageserver itself does not create tenants and timelines, until started first and asked via HTTP API.
@@ -1069,6 +1073,9 @@ class NeonEnv:
         self.pg_distrib_dir = config.pg_distrib_dir
         self.endpoint_counter = 0
         self.storage_controller_config = config.storage_controller_config
+
+        # generate initial tenant ID here instead of letting 'neon init' generate it,
+        # so that we don't need to dig it out of the config file afterwards.
         self.initial_tenant = config.initial_tenant
         self.initial_timeline = config.initial_timeline
 
@@ -1514,6 +1521,14 @@ class PageserverPort:
     http: int
 
 
+CREATE_TIMELINE_ID_EXTRACTOR: re.Pattern = re.compile(  # type: ignore[type-arg]
+    r"^Created timeline '(?P<timeline_id>[^']+)'", re.MULTILINE
+)
+TIMELINE_DATA_EXTRACTOR: re.Pattern = re.compile(  # type: ignore[type-arg]
+    r"\s?(?P<branch_name>[^\s]+)\s\[(?P<timeline_id>[^\]]+)\]", re.MULTILINE
+)
+
+
 class AbstractNeonCli(abc.ABC):
     """
     A typed wrapper around an arbitrary Neon CLI tool.
@@ -1742,9 +1757,6 @@ class NeonCli(AbstractNeonCli):
         tenant_id: Optional[TenantId] = None,
         timeline_id: Optional[TimelineId] = None,
     ) -> TimelineId:
-        if timeline_id is None:
-            timeline_id = TimelineId.generate()
-
         cmd = [
             "timeline",
             "create",
@@ -1752,16 +1764,23 @@ class NeonCli(AbstractNeonCli):
             new_branch_name,
             "--tenant-id",
             str(tenant_id or self.env.initial_tenant),
-            "--timeline-id",
-            str(timeline_id),
             "--pg-version",
             self.env.pg_version,
         ]
 
+        if timeline_id is not None:
+            cmd.extend(["--timeline-id", str(timeline_id)])
+
         res = self.raw_cli(cmd)
         res.check_returncode()
 
-        return timeline_id
+        matches = CREATE_TIMELINE_ID_EXTRACTOR.search(res.stdout)
+
+        created_timeline_id = None
+        if matches is not None:
+            created_timeline_id = matches.group("timeline_id")
+
+        return TimelineId(str(created_timeline_id))
 
     def create_branch(
         self,
@@ -1769,17 +1788,12 @@ class NeonCli(AbstractNeonCli):
         ancestor_branch_name: Optional[str] = None,
         tenant_id: Optional[TenantId] = None,
         ancestor_start_lsn: Optional[Lsn] = None,
-        new_timeline_id: Optional[TimelineId] = None,
     ) -> TimelineId:
-        if new_timeline_id is None:
-            new_timeline_id = TimelineId.generate()
         cmd = [
             "timeline",
             "branch",
             "--branch-name",
             new_branch_name,
-            "--timeline-id",
-            str(new_timeline_id),
             "--tenant-id",
             str(tenant_id or self.env.initial_tenant),
         ]
@@ -1791,7 +1805,16 @@ class NeonCli(AbstractNeonCli):
         res = self.raw_cli(cmd)
         res.check_returncode()
 
-        return TimelineId(str(new_timeline_id))
+        matches = CREATE_TIMELINE_ID_EXTRACTOR.search(res.stdout)
+
+        created_timeline_id = None
+        if matches is not None:
+            created_timeline_id = matches.group("timeline_id")
+
+        if created_timeline_id is None:
+            raise Exception("could not find timeline id after `neon timeline create` invocation")
+        else:
+            return TimelineId(str(created_timeline_id))
 
     def list_timelines(self, tenant_id: Optional[TenantId] = None) -> List[Tuple[str, TimelineId]]:
         """
@@ -1800,9 +1823,6 @@ class NeonCli(AbstractNeonCli):
 
         # main [b49f7954224a0ad25cc0013ea107b54b]
         # ┣━ @0/16B5A50: test_cli_branch_list_main [20f98c79111b9015d84452258b7d5540]
-        TIMELINE_DATA_EXTRACTOR: re.Pattern = re.compile(  # type: ignore[type-arg]
-            r"\s?(?P<branch_name>[^\s]+)\s\[(?P<timeline_id>[^\]]+)\]", re.MULTILINE
-        )
         res = self.raw_cli(
             ["timeline", "list", "--tenant-id", str(tenant_id or self.env.initial_tenant)]
         )

test_runner/performance/test_storage_controller_scale.py

@@ -84,7 +84,7 @@ def test_storage_controller_many_tenants(
     compute_reconfigure_listener.register_on_notify(lambda body: time.sleep(0.01))
 
     env = neon_env_builder.init_configs()
-    env.start()
+    neon_env_builder.start()
 
     # We will intentionally stress reconciler concurrrency, which triggers a warning when lots
     # of shards are hitting the delayed path.

test_runner/regress/test_compatibility.py

@@ -178,7 +178,7 @@ def test_backward_compatibility(
         neon_env_builder.num_safekeepers = 3
         env = neon_env_builder.from_repo_dir(compatibility_snapshot_dir / "repo")
         env.pageserver.allowed_errors.append(ingest_lag_log_line)
-        env.start()
+        neon_env_builder.start()
 
         check_neon_works(
             env,
@@ -265,7 +265,7 @@ def test_forward_compatibility(
         # does not include logs from previous runs
         assert not env.pageserver.log_contains("git-env:" + prev_pageserver_version)
 
-        env.start()
+        neon_env_builder.start()
 
         # ensure the specified pageserver is running
         assert env.pageserver.log_contains("git-env:" + prev_pageserver_version)

test_runner/regress/test_sharding.py

@@ -374,7 +374,7 @@ def test_sharding_split_smoke(
     non_default_tenant_config = {"gc_horizon": 77 * 1024 * 1024}
 
     env = neon_env_builder.init_configs(True)
-    env.start()
+    neon_env_builder.start()
     tenant_id = TenantId.generate()
     timeline_id = TimelineId.generate()
     env.neon_cli.create_tenant(
@@ -1436,7 +1436,7 @@ def test_sharding_unlogged_relation(neon_env_builder: NeonEnvBuilder):
 
     neon_env_builder.num_pageservers = 2
     env = neon_env_builder.init_configs()
-    env.start()
+    neon_env_builder.start()
 
     tenant_id = TenantId.generate()
     timeline_id = TimelineId.generate()
@@ -1475,7 +1475,7 @@ def test_top_tenants(neon_env_builder: NeonEnvBuilder):
     """
 
     env = neon_env_builder.init_configs()
-    env.start()
+    neon_env_builder.start()
 
     tenants = []
     n_tenants = 8

test_runner/regress/test_upgrade.py

@@ -1,54 +0,0 @@
-from time import sleep
-
-import pytest
-import requests
-from fixtures.neon_fixtures import NeonEnv
-from fixtures.pg_version import PgVersion
-
-
-def test_upgrade(pg_version: PgVersion, neon_simple_env: NeonEnv):
-    env = neon_simple_env
-
-    upgrade_to: PgVersion
-    if pg_version == PgVersion.V14:
-        upgrade_to = PgVersion.V15
-    elif pg_version == PgVersion.V15:
-        upgrade_to = PgVersion.V16
-    else:
-        pytest.skip("Nothing to upgrade")
-
-    env.neon_cli.create_timeline("test_upgrade")
-    endpoint = env.endpoints.create_start("test_upgrade")
-    resp = requests.post(
-        f"http://localhost:{endpoint.http_port}/upgrade",
-        json={
-            "pg_version": upgrade_to,
-        },
-    )
-    assert resp.status_code == 202
-
-    while True:
-        resp = requests.get(f"http://localhost:{endpoint.http_port}/status")
-        assert resp.status_code == 200
-
-        data = resp.json()
-        if data["status"] == "upgrading":
-            sleep(1)
-            continue
-        elif data["status"] == "running":
-            break
-        else:
-            pytest.fail(f"Unexpected compute state during upgrade: {data['status']}")
-
-    endpoint.stop_and_destroy()
-
-
-def test_upgrade_bad_request(neon_simple_env: NeonEnv):
-    env = neon_simple_env
-
-    env.neon_cli.create_timeline("test_upgrade_bad_request")
-    endpoint = env.endpoints.create_start("test_upgrade_bad_request")
-    resp = requests.post(f"http://localhost:{endpoint.http_port}/upgrade")
-    assert resp.status_code == 400
-
-    # TODO: Use postgres versions that are out of range.

vendor/revisions.json

@@ -1,11 +1,11 @@
 {
   "v16": [
     "16.4",
-    "0baa7346dfd42d61912eeca554c9bb0a190f0a1e"
+    "6e9a4ff6249ac02b8175054b7b3f7dfb198be48b"
   ],
   "v15": [
     "15.8",
-    "6f6d77fb5960602fcd3fd130aca9f99ecb1619c9"
+    "49d5e576a56e4cc59cd6a6a0791b2324b9fa675e"
   ],
   "v14": [
     "14.13",