v16-lr-worker-reply-after-processing

.devcontainer/Dockerfile.devcontainer

@@ -1 +0,0 @@
-FROM neondatabase/build-tools:pinned

.devcontainer/devcontainer.json

@@ -1,23 +0,0 @@
-// https://containers.dev/implementors/json_reference/
-{
-  "name": "Neon",
-  "build": {
-    "context": "..",
-    "dockerfile": "Dockerfile.devcontainer"
-  },
-
-  "postCreateCommand": {
-    "build neon": "BUILD_TYPE=debug CARGO_BUILD_FLAGS='--features=testing' mold -run make -s -j`nproc`",
-    "install python deps": "./scripts/pysync"
-  },
-
-  "customizations": {
-    "vscode": {
-      "extensions": [
-        "charliermarsh.ruff",
-        "github.vscode-github-actions",
-        "rust-lang.rust-analyzer"
-      ]
-    }
-  }
-}

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,
@@ -1571,6 +1570,7 @@ fn cli() -> Command {
                         .value_parser(value_parser!(PathBuf))
                         .value_name("config")
                 )
+                .arg(pg_version_arg.clone())
                 .arg(force_arg)
         )
         .subcommand(
@@ -1583,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))

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(

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/walingest.rs

@@ -1222,7 +1222,7 @@ impl WalIngest {
             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?;
@@ -1244,7 +1244,7 @@ impl WalIngest {
             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?;
@@ -1737,7 +1737,7 @@ impl WalIngest {
                     continue;
                 }
 
-                modification.put_rel_page_image_zero(rel, gap_blknum)?;
+                modification.put_rel_page_image_zero(rel, gap_blknum);
             }
         }
         Ok(())
@@ -1803,7 +1803,7 @@ impl WalIngest {
 
             // 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(())

proxy/src/auth/backend.rs

@@ -311,9 +311,7 @@ async fn auth_quirks(
     let (allowed_ips, maybe_secret) = api.get_allowed_ips_and_secret(ctx, &info).await?;
 
     // check allowed list
-    if config.ip_allowlist_check_enabled
-        && !check_peer_addr_is_in_list(&ctx.peer_addr(), &allowed_ips)
-    {
+    if !check_peer_addr_is_in_list(&ctx.peer_addr(), &allowed_ips) {
         return Err(auth::AuthError::ip_address_not_allowed(ctx.peer_addr()));
     }
 
@@ -605,7 +603,6 @@ mod tests {
         rate_limiter_enabled: true,
         rate_limiter: AuthRateLimiter::new(&RateBucketInfo::DEFAULT_AUTH_SET),
         rate_limit_ip_subnet: 64,
-        ip_allowlist_check_enabled: true,
     });
 
     async fn read_message(r: &mut (impl AsyncRead + Unpin), b: &mut BytesMut) -> PgMessage {

proxy/src/auth/credentials.rs

@@ -538,17 +538,4 @@ mod tests {
         ));
         Ok(())
     }
-
-    #[test]
-    fn test_connection_blocker() {
-        fn check(v: serde_json::Value) -> bool {
-            let peer_addr = IpAddr::from([127, 0, 0, 1]);
-            let ip_list: Vec<IpPattern> = serde_json::from_value(v).unwrap();
-            check_peer_addr_is_in_list(&peer_addr, &ip_list)
-        }
-
-        assert!(check(json!([])));
-        assert!(check(json!(["127.0.0.1"])));
-        assert!(!check(json!(["255.255.255.255"])));
-    }
 }

proxy/src/bin/local_proxy.rs

@@ -224,7 +224,6 @@ fn build_config(args: &LocalProxyCliArgs) -> anyhow::Result<&'static ProxyConfig
             rate_limiter_enabled: false,
             rate_limiter: BucketRateLimiter::new(vec![]),
             rate_limit_ip_subnet: 64,
-            ip_allowlist_check_enabled: true,
         },
         require_client_ip: false,
         handshake_timeout: Duration::from_secs(10),

proxy/src/bin/proxy.rs

@@ -224,10 +224,6 @@ struct ProxyCliArgs {
     /// Whether to retry the wake_compute request
     #[clap(long, default_value = config::RetryConfig::WAKE_COMPUTE_DEFAULT_VALUES)]
     wake_compute_retry: String,
-
-    /// Configure if this is a private access proxy for the POC: In that case the proxy will ignore the IP allowlist
-    #[clap(long, default_value_t = false, value_parser = clap::builder::BoolishValueParser::new(), action = clap::ArgAction::Set)]
-    is_private_access_proxy: bool,
 }
 
 #[derive(clap::Args, Clone, Copy, Debug)]
@@ -686,7 +682,6 @@ fn build_config(args: &ProxyCliArgs) -> anyhow::Result<&'static ProxyConfig> {
         rate_limiter_enabled: args.auth_rate_limit_enabled,
         rate_limiter: AuthRateLimiter::new(args.auth_rate_limit.clone()),
         rate_limit_ip_subnet: args.auth_rate_limit_ip_subnet,
-        ip_allowlist_check_enabled: !args.is_private_access_proxy,
     };
 
     let config = Box::leak(Box::new(ProxyConfig {

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/config.rs

@@ -64,7 +64,6 @@ pub struct AuthenticationConfig {
     pub rate_limiter_enabled: bool,
     pub rate_limiter: AuthRateLimiter,
     pub rate_limit_ip_subnet: u8,
-    pub ip_allowlist_check_enabled: bool,
 }
 
 impl TlsConfig {

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/serverless/backend.rs

@@ -50,9 +50,7 @@ impl PoolingBackend {
             .as_ref()
             .map(|()| user_info.clone());
         let (allowed_ips, maybe_secret) = backend.get_allowed_ips_and_secret(ctx).await?;
-        if config.ip_allowlist_check_enabled
-            && !check_peer_addr_is_in_list(&ctx.peer_addr(), &allowed_ips)
-        {
+        if !check_peer_addr_is_in_list(&ctx.peer_addr(), &allowed_ips) {
             return Err(AuthError::ip_address_not_allowed(ctx.peer_addr()));
         }
         if !self

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
+            }
+        }
     }
 }
 

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

vendor/revisions.json

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