compress cache key

control_plane/src/bin/neon_local.rs

@@ -21,8 +21,10 @@ use pageserver_api::config::{
     DEFAULT_HTTP_LISTEN_PORT as DEFAULT_PAGESERVER_HTTP_PORT,
     DEFAULT_PG_LISTEN_PORT as DEFAULT_PAGESERVER_PG_PORT,
 };
-use pageserver_api::controller_api::{PlacementPolicy, TenantCreateRequest};
-use pageserver_api::models::{ShardParameters, TimelineCreateRequest, TimelineInfo};
+use pageserver_api::controller_api::PlacementPolicy;
+use pageserver_api::models::{
+    ShardParameters, TenantCreateRequest, TimelineCreateRequest, TimelineInfo,
+};
 use pageserver_api::shard::{ShardCount, ShardStripeSize, TenantShardId};
 use postgres_backend::AuthType;
 use postgres_connection::parse_host_port;

control_plane/src/pageserver.rs

@@ -17,7 +17,8 @@ use anyhow::{bail, Context};
 use camino::Utf8PathBuf;
 use futures::SinkExt;
 use pageserver_api::models::{
-    self, AuxFilePolicy, LocationConfig, TenantHistorySize, TenantInfo, TimelineInfo,
+    self, AuxFilePolicy, LocationConfig, ShardParameters, TenantHistorySize, TenantInfo,
+    TimelineInfo,
 };
 use pageserver_api::shard::TenantShardId;
 use pageserver_client::mgmt_api;
@@ -396,6 +397,28 @@ impl PageServerNode {
         }
     }
 
+    pub async fn tenant_create(
+        &self,
+        new_tenant_id: TenantId,
+        generation: Option<u32>,
+        settings: HashMap<&str, &str>,
+    ) -> anyhow::Result<TenantId> {
+        let config = Self::parse_config(settings.clone())?;
+
+        let request = models::TenantCreateRequest {
+            new_tenant_id: TenantShardId::unsharded(new_tenant_id),
+            generation,
+            config,
+            shard_parameters: ShardParameters::default(),
+            // Placement policy is not meaningful for creations not done via storage controller
+            placement_policy: None,
+        };
+        if !settings.is_empty() {
+            bail!("Unrecognized tenant settings: {settings:?}")
+        }
+        Ok(self.http_client.tenant_create(&request).await?)
+    }
+
     pub async fn tenant_config(
         &self,
         tenant_id: TenantId,

control_plane/src/storage_controller.rs

@@ -5,11 +5,12 @@ use crate::{
 use camino::{Utf8Path, Utf8PathBuf};
 use pageserver_api::{
     controller_api::{
-        NodeConfigureRequest, NodeRegisterRequest, TenantCreateRequest, TenantCreateResponse,
-        TenantLocateResponse, TenantShardMigrateRequest, TenantShardMigrateResponse,
+        NodeConfigureRequest, NodeRegisterRequest, TenantCreateResponse, TenantLocateResponse,
+        TenantShardMigrateRequest, TenantShardMigrateResponse,
     },
     models::{
-        TenantShardSplitRequest, TenantShardSplitResponse, TimelineCreateRequest, TimelineInfo,
+        TenantCreateRequest, TenantShardSplitRequest, TenantShardSplitResponse,
+        TimelineCreateRequest, TimelineInfo,
     },
     shard::{ShardStripeSize, TenantShardId},
 };

control_plane/storcon_cli/src/main.rs

@@ -4,13 +4,13 @@ use std::{str::FromStr, time::Duration};
 use clap::{Parser, Subcommand};
 use pageserver_api::{
     controller_api::{
-        NodeAvailabilityWrapper, NodeDescribeResponse, ShardSchedulingPolicy, TenantCreateRequest,
+        NodeAvailabilityWrapper, NodeDescribeResponse, ShardSchedulingPolicy,
         TenantDescribeResponse, TenantPolicyRequest,
     },
     models::{
         EvictionPolicy, EvictionPolicyLayerAccessThreshold, LocationConfigSecondary,
-        ShardParameters, TenantConfig, TenantConfigRequest, TenantShardSplitRequest,
-        TenantShardSplitResponse,
+        ShardParameters, TenantConfig, TenantConfigRequest, TenantCreateRequest,
+        TenantShardSplitRequest, TenantShardSplitResponse,
     },
     shard::{ShardStripeSize, TenantShardId},
 };
@@ -336,18 +336,14 @@ async fn main() -> anyhow::Result<()> {
                 .await?;
         }
         Command::TenantCreate { tenant_id } => {
-            storcon_client
-                .dispatch(
-                    Method::POST,
-                    "v1/tenant".to_string(),
-                    Some(TenantCreateRequest {
-                        new_tenant_id: TenantShardId::unsharded(tenant_id),
-                        generation: None,
-                        shard_parameters: ShardParameters::default(),
-                        placement_policy: Some(PlacementPolicy::Attached(1)),
-                        config: TenantConfig::default(),
-                    }),
-                )
+            vps_client
+                .tenant_create(&TenantCreateRequest {
+                    new_tenant_id: TenantShardId::unsharded(tenant_id),
+                    generation: None,
+                    shard_parameters: ShardParameters::default(),
+                    placement_policy: Some(PlacementPolicy::Attached(1)),
+                    config: TenantConfig::default(),
+                })
                 .await?;
         }
         Command::TenantDelete { tenant_id } => {

libs/pageserver_api/src/controller_api.rs

@@ -11,27 +11,6 @@ use crate::{
     shard::{ShardStripeSize, TenantShardId},
 };
 
-#[derive(Serialize, Deserialize, Debug)]
-#[serde(deny_unknown_fields)]
-pub struct TenantCreateRequest {
-    pub new_tenant_id: TenantShardId,
-    #[serde(default)]
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub generation: Option<u32>,
-
-    // If omitted, create a single shard with TenantShardId::unsharded()
-    #[serde(default)]
-    #[serde(skip_serializing_if = "ShardParameters::is_unsharded")]
-    pub shard_parameters: ShardParameters,
-
-    #[serde(default)]
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub placement_policy: Option<PlacementPolicy>,
-
-    #[serde(flatten)]
-    pub config: TenantConfig, // as we have a flattened field, we should reject all unknown fields in it
-}
-
 #[derive(Serialize, Deserialize)]
 pub struct TenantCreateResponseShard {
     pub shard_id: TenantShardId,
@@ -301,19 +280,4 @@ mod test {
         assert_eq!(serde_json::from_str::<PlacementPolicy>(&encoded)?, v);
         Ok(())
     }
-
-    #[test]
-    fn test_reject_unknown_field() {
-        let id = TenantId::generate();
-        let create_request = serde_json::json!({
-            "new_tenant_id": id.to_string(),
-            "unknown_field": "unknown_value".to_string(),
-        });
-        let err = serde_json::from_value::<TenantCreateRequest>(create_request).unwrap_err();
-        assert!(
-            err.to_string().contains("unknown field `unknown_field`"),
-            "expect unknown field `unknown_field` error, got: {}",
-            err
-        );
-    }
 }

libs/pageserver_api/src/models.rs

@@ -25,6 +25,7 @@ use utils::{
     serde_system_time,
 };
 
+use crate::controller_api::PlacementPolicy;
 use crate::{
     reltag::RelTag,
     shard::{ShardCount, ShardStripeSize, TenantShardId},
@@ -270,6 +271,28 @@ impl Default for ShardParameters {
     }
 }
 
+#[derive(Serialize, Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
+pub struct TenantCreateRequest {
+    pub new_tenant_id: TenantShardId,
+    #[serde(default)]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub generation: Option<u32>,
+
+    // If omitted, create a single shard with TenantShardId::unsharded()
+    #[serde(default)]
+    #[serde(skip_serializing_if = "ShardParameters::is_unsharded")]
+    pub shard_parameters: ShardParameters,
+
+    // This parameter is only meaningful in requests sent to the storage controller
+    #[serde(default)]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub placement_policy: Option<PlacementPolicy>,
+
+    #[serde(flatten)]
+    pub config: TenantConfig, // as we have a flattened field, we should reject all unknown fields in it
+}
+
 /// An alternative representation of `pageserver::tenant::TenantConf` with
 /// simpler types.
 #[derive(Serialize, Deserialize, Debug, Default, Clone, Eq, PartialEq)]
@@ -524,6 +547,10 @@ pub struct LocationConfigListResponse {
     pub tenant_shards: Vec<(TenantShardId, Option<LocationConfig>)>,
 }
 
+#[derive(Serialize, Deserialize)]
+#[serde(transparent)]
+pub struct TenantCreateResponse(pub TenantId);
+
 #[derive(Serialize)]
 pub struct StatusResponse {
     pub id: NodeId,
@@ -1480,6 +1507,18 @@ mod tests {
 
     #[test]
     fn test_reject_unknown_field() {
+        let id = TenantId::generate();
+        let create_request = json!({
+            "new_tenant_id": id.to_string(),
+            "unknown_field": "unknown_value".to_string(),
+        });
+        let err = serde_json::from_value::<TenantCreateRequest>(create_request).unwrap_err();
+        assert!(
+            err.to_string().contains("unknown field `unknown_field`"),
+            "expect unknown field `unknown_field` error, got: {}",
+            err
+        );
+
         let id = TenantId::generate();
         let config_request = json!({
             "tenant_id": id.to_string(),

pageserver/client/src/mgmt_api.rs

@@ -205,6 +205,15 @@ impl Client {
         Ok(())
     }
 
+    pub async fn tenant_create(&self, req: &TenantCreateRequest) -> Result<TenantId> {
+        let uri = format!("{}/v1/tenant", self.mgmt_api_endpoint);
+        self.request(Method::POST, &uri, req)
+            .await?
+            .json()
+            .await
+            .map_err(Error::ReceiveBody)
+    }
+
     /// The tenant deletion API can return 202 if deletion is incomplete, or
     /// 404 if it is complete.  Callers are responsible for checking the status
     /// code and retrying.  Error codes other than 404 will return Err().

pageserver/src/http/routes.rs

@@ -53,6 +53,7 @@ use utils::http::request::{get_request_param, must_get_query_param, parse_query_
 
 use crate::context::{DownloadBehavior, RequestContext};
 use crate::deletion_queue::DeletionQueueClient;
+use crate::metrics::{StorageTimeOperation, STORAGE_TIME_GLOBAL};
 use crate::pgdatadir_mapping::LsnForTimestamp;
 use crate::task_mgr::TaskKind;
 use crate::tenant::config::{LocationConf, TenantConfOpt};
@@ -74,12 +75,13 @@ use crate::tenant::timeline::CompactFlags;
 use crate::tenant::timeline::CompactionError;
 use crate::tenant::timeline::Timeline;
 use crate::tenant::GetTimelineError;
+use crate::tenant::SpawnMode;
 use crate::tenant::{LogicalSizeCalculationCause, PageReconstructError};
 use crate::{config::PageServerConf, tenant::mgr};
 use crate::{disk_usage_eviction_task, tenant};
 use pageserver_api::models::{
-    StatusResponse, TenantConfigRequest, TenantInfo, TimelineCreateRequest, TimelineGcRequest,
-    TimelineInfo,
+    StatusResponse, TenantConfigRequest, TenantCreateRequest, TenantCreateResponse, TenantInfo,
+    TimelineCreateRequest, TimelineGcRequest, TimelineInfo,
 };
 use utils::{
     auth::SwappableJwtAuth,
@@ -1235,6 +1237,75 @@ pub fn html_response(status: StatusCode, data: String) -> Result<Response<Body>,
     Ok(response)
 }
 
+/// Helper for requests that may take a generation, which is mandatory
+/// when control_plane_api is set, but otherwise defaults to Generation::none()
+fn get_request_generation(state: &State, req_gen: Option<u32>) -> Result<Generation, ApiError> {
+    if state.conf.control_plane_api.is_some() {
+        req_gen
+            .map(Generation::new)
+            .ok_or(ApiError::BadRequest(anyhow!(
+                "generation attribute missing"
+            )))
+    } else {
+        // Legacy mode: all tenants operate with no generation
+        Ok(Generation::none())
+    }
+}
+
+async fn tenant_create_handler(
+    mut request: Request<Body>,
+    _cancel: CancellationToken,
+) -> Result<Response<Body>, ApiError> {
+    let request_data: TenantCreateRequest = json_request(&mut request).await?;
+    let target_tenant_id = request_data.new_tenant_id;
+    check_permission(&request, None)?;
+
+    let _timer = STORAGE_TIME_GLOBAL
+        .get_metric_with_label_values(&[StorageTimeOperation::CreateTenant.into()])
+        .expect("bug")
+        .start_timer();
+
+    let tenant_conf =
+        TenantConfOpt::try_from(&request_data.config).map_err(ApiError::BadRequest)?;
+
+    let state = get_state(&request);
+
+    let generation = get_request_generation(state, request_data.generation)?;
+
+    let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
+
+    let location_conf =
+        LocationConf::attached_single(tenant_conf, generation, &request_data.shard_parameters);
+
+    let new_tenant = state
+        .tenant_manager
+        .upsert_location(
+            target_tenant_id,
+            location_conf,
+            None,
+            SpawnMode::Create,
+            &ctx,
+        )
+        .await?;
+
+    let Some(new_tenant) = new_tenant else {
+        // This should never happen: indicates a bug in upsert_location
+        return Err(ApiError::InternalServerError(anyhow::anyhow!(
+            "Upsert succeeded but didn't return tenant!"
+        )));
+    };
+    // We created the tenant. Existing API semantics are that the tenant
+    // is Active when this function returns.
+    new_tenant
+        .wait_to_become_active(ACTIVE_TENANT_TIMEOUT)
+        .await?;
+
+    json_response(
+        StatusCode::CREATED,
+        TenantCreateResponse(new_tenant.tenant_shard_id().tenant_id),
+    )
+}
+
 async fn get_tenant_config_handler(
     request: Request<Body>,
     _cancel: CancellationToken,
@@ -2540,6 +2611,7 @@ pub fn make_router(
             api_handler(r, reload_auth_validation_keys_handler)
         })
         .get("/v1/tenant", |r| api_handler(r, tenant_list_handler))
+        .post("/v1/tenant", |r| api_handler(r, tenant_create_handler))
         .get("/v1/tenant/:tenant_shard_id", |r| {
             api_handler(r, tenant_status)
         })

pageserver/src/metrics.rs

@@ -53,6 +53,9 @@ pub(crate) enum StorageTimeOperation {
 
     #[strum(serialize = "find gc cutoffs")]
     FindGcCutoffs,
+
+    #[strum(serialize = "create tenant")]
+    CreateTenant,
 }
 
 pub(crate) static STORAGE_TIME_SUM_PER_TIMELINE: Lazy<CounterVec> = Lazy::new(|| {

pageserver/src/tenant.rs

@@ -213,6 +213,8 @@ pub(crate) enum SpawnMode {
     Eager,
     /// Lazy activation in the background, with the option to skip the queue if the need comes up
     Lazy,
+    /// Tenant has been created during the lifetime of this process
+    Create,
 }
 
 ///
@@ -806,6 +808,9 @@ impl Tenant {
                 };
 
                 let preload = match &mode {
+                    SpawnMode::Create => {
+                        None
+                    },
                     SpawnMode::Eager | SpawnMode::Lazy => {
                         let _preload_timer = TENANT.preload.start_timer();
                         let res = tenant_clone
@@ -827,8 +832,11 @@ impl Tenant {
 
                 // We will time the duration of the attach phase unless this is a creation (attach will do no work)
                 let attached = {
-                    let _attach_timer = Some(TENANT.attach.start_timer());
-                    tenant_clone.attach(preload, &ctx).await
+                    let _attach_timer = match mode {
+                        SpawnMode::Create => None,
+                        SpawnMode::Eager | SpawnMode::Lazy => Some(TENANT.attach.start_timer()),
+                    };
+                    tenant_clone.attach(preload, mode, &ctx).await
                 };
 
                 match attached {
@@ -904,14 +912,21 @@ impl Tenant {
     async fn attach(
         self: &Arc<Tenant>,
         preload: Option<TenantPreload>,
+        mode: SpawnMode,
         ctx: &RequestContext,
     ) -> anyhow::Result<()> {
         span::debug_assert_current_span_has_tenant_id();
 
         failpoint_support::sleep_millis_async!("before-attaching-tenant");
 
-        let Some(preload) = preload else {
-            anyhow::bail!("local-only deployment is no longer supported, https://github.com/neondatabase/neon/issues/5624");
+        let preload = match (preload, mode) {
+            (Some(p), _) => p,
+            (None, SpawnMode::Create) => TenantPreload {
+                timelines: HashMap::new(),
+            },
+            (None, _) => {
+                anyhow::bail!("local-only deployment is no longer supported, https://github.com/neondatabase/neon/issues/5624");
+            }
         };
 
         let mut timelines_to_resume_deletions = vec![];
@@ -3826,7 +3841,7 @@ pub(crate) mod harness {
             let preload = tenant
                 .preload(&self.remote_storage, CancellationToken::new())
                 .await?;
-            tenant.attach(Some(preload), ctx).await?;
+            tenant.attach(Some(preload), SpawnMode::Eager, ctx).await?;
 
             tenant.state.send_replace(TenantState::Active);
             for timeline in tenant.timelines.lock().unwrap().values() {

pageserver/src/tenant/timeline.rs

@@ -3336,13 +3336,7 @@ impl Timeline {
                     .await
                 {
                     Ok(result) => result,
-                    Err(e) => {
-                        return Err(if self.cancel.is_cancelled() {
-                            PageReconstructError::Cancelled
-                        } else {
-                            PageReconstructError::Other(e)
-                        })
-                    }
+                    Err(e) => return Err(PageReconstructError::from(e)),
                 };
                 cont_lsn = lsn_floor;
                 *read_count += 1;

pageserver/src/tenant/vectored_blob_io.rs

@@ -20,7 +20,6 @@ use std::num::NonZeroUsize;
 
 use bytes::BytesMut;
 use pageserver_api::key::Key;
-use tokio_epoll_uring::BoundedBuf;
 use utils::lsn::Lsn;
 use utils::vec_map::VecMap;
 
@@ -317,9 +316,8 @@ impl<'a> VectoredBlobReader<'a> {
         );
         let buf = self
             .file
-            .read_exact_at(buf.slice(0..read.size()), read.start, ctx)
-            .await?
-            .into_inner();
+            .read_exact_at_n(buf, read.start, read.size(), ctx)
+            .await?;
 
         let blobs_at = read.blobs_at.as_slice();
         let start_offset = blobs_at.first().expect("VectoredRead is never empty").0;

pageserver/src/virtual_file.rs

@@ -13,7 +13,7 @@
 use crate::context::RequestContext;
 use crate::metrics::{StorageIoOperation, STORAGE_IO_SIZE, STORAGE_IO_TIME_METRIC};
 
-use crate::page_cache::{PageWriteGuard, PAGE_SZ};
+use crate::page_cache::PageWriteGuard;
 use crate::tenant::TENANTS_SEGMENT_NAME;
 use camino::{Utf8Path, Utf8PathBuf};
 use once_cell::sync::OnceCell;
@@ -48,7 +48,6 @@ pub(crate) mod owned_buffers_io {
     //! but for the time being we're proving out the primitives in the neon.git repo
     //! for faster iteration.
 
-    pub(crate) mod slice;
     pub(crate) mod write;
     pub(crate) mod util {
         pub(crate) mod size_tracking_writer;
@@ -144,17 +143,16 @@ struct SlotInner {
 /// Impl of [`tokio_epoll_uring::IoBuf`] and [`tokio_epoll_uring::IoBufMut`] for [`PageWriteGuard`].
 struct PageWriteGuardBuf {
     page: PageWriteGuard<'static>,
+    init_up_to: usize,
 }
 // Safety: the [`PageWriteGuard`] gives us exclusive ownership of the page cache slot,
 // and the location remains stable even if [`Self`] or the [`PageWriteGuard`] is moved.
-// Page cache pages are zero-initialized, so, wrt uninitialized memory we're good.
-// (Page cache tracks separately whether the contents are valid, see `PageWriteGuard::mark_valid`.)
 unsafe impl tokio_epoll_uring::IoBuf for PageWriteGuardBuf {
     fn stable_ptr(&self) -> *const u8 {
         self.page.as_ptr()
     }
     fn bytes_init(&self) -> usize {
-        self.page.len()
+        self.init_up_to
     }
     fn bytes_total(&self) -> usize {
         self.page.len()
@@ -168,8 +166,8 @@ unsafe impl tokio_epoll_uring::IoBufMut for PageWriteGuardBuf {
     }
 
     unsafe fn set_init(&mut self, pos: usize) {
-        // There shouldn't really be any reason to call this API since bytes_init() == bytes_total().
         assert!(pos <= self.page.len());
+        self.init_up_to = pos;
     }
 }
 
@@ -587,37 +585,37 @@ impl VirtualFile {
         Ok(self.pos)
     }
 
-    /// Read the file contents in range `offset..(offset + slice.bytes_total())` into `slice[0..slice.bytes_total()]`.
-    ///
-    /// The returned `Slice<Buf>` is equivalent to the input `slice`, i.e., it's the same view into the same buffer.
-    pub async fn read_exact_at<Buf>(
+    pub async fn read_exact_at<B>(
         &self,
-        slice: Slice<Buf>,
+        buf: B,
         offset: u64,
         ctx: &RequestContext,
-    ) -> Result<Slice<Buf>, Error>
+    ) -> Result<B, Error>
     where
-        Buf: IoBufMut + Send,
+        B: IoBufMut + Send,
     {
-        let assert_we_return_original_bounds = if cfg!(debug_assertions) {
-            Some((slice.stable_ptr() as usize, slice.bytes_total()))
-        } else {
-            None
-        };
+        let (buf, res) = read_exact_at_impl(buf, offset, None, |buf, offset| {
+            self.read_at(buf, offset, ctx)
+        })
+        .await;
+        res.map(|()| buf)
+    }
 
-        let original_bounds = slice.bounds();
-        let (buf, res) =
-            read_exact_at_impl(slice, offset, |buf, offset| self.read_at(buf, offset, ctx)).await;
-        let res = res.map(|_| buf.slice(original_bounds));
-
-        if let Some(original_bounds) = assert_we_return_original_bounds {
-            if let Ok(slice) = &res {
-                let returned_bounds = (slice.stable_ptr() as usize, slice.bytes_total());
-                assert_eq!(original_bounds, returned_bounds);
-            }
-        }
-
-        res
+    pub async fn read_exact_at_n<B>(
+        &self,
+        buf: B,
+        offset: u64,
+        count: usize,
+        ctx: &RequestContext,
+    ) -> Result<B, Error>
+    where
+        B: IoBufMut + Send,
+    {
+        let (buf, res) = read_exact_at_impl(buf, offset, Some(count), |buf, offset| {
+            self.read_at(buf, offset, ctx)
+        })
+        .await;
+        res.map(|()| buf)
     }
 
     /// Like [`Self::read_exact_at`] but for [`PageWriteGuard`].
@@ -627,11 +625,13 @@ impl VirtualFile {
         offset: u64,
         ctx: &RequestContext,
     ) -> Result<PageWriteGuard<'static>, Error> {
-        let buf = PageWriteGuardBuf { page }.slice_full();
-        debug_assert_eq!(buf.bytes_total(), PAGE_SZ);
-        self.read_exact_at(buf, offset, ctx)
-            .await
-            .map(|slice| slice.into_inner().page)
+        let buf = PageWriteGuardBuf {
+            page,
+            init_up_to: 0,
+        };
+        let res = self.read_exact_at(buf, offset, ctx).await;
+        res.map(|PageWriteGuardBuf { page, .. }| page)
+            .map_err(|e| Error::new(ErrorKind::Other, e))
     }
 
     // Copied from https://doc.rust-lang.org/1.72.0/src/std/os/unix/fs.rs.html#219-235
@@ -722,14 +722,14 @@ impl VirtualFile {
         (buf, Ok(n))
     }
 
-    pub(crate) async fn read_at<Buf>(
+    pub(crate) async fn read_at<B>(
         &self,
-        buf: tokio_epoll_uring::Slice<Buf>,
+        buf: B,
         offset: u64,
         _ctx: &RequestContext, /* TODO: use for metrics: https://github.com/neondatabase/neon/issues/6107 */
-    ) -> (tokio_epoll_uring::Slice<Buf>, Result<usize, Error>)
+    ) -> (B, Result<usize, Error>)
     where
-        Buf: tokio_epoll_uring::IoBufMut + Send,
+        B: tokio_epoll_uring::BoundedBufMut + Send,
     {
         let file_guard = match self.lock_file().await {
             Ok(file_guard) => file_guard,
@@ -781,16 +781,26 @@ impl VirtualFile {
 }
 
 // Adapted from https://doc.rust-lang.org/1.72.0/src/std/os/unix/fs.rs.html#117-135
-pub async fn read_exact_at_impl<Buf, F, Fut>(
-    mut buf: tokio_epoll_uring::Slice<Buf>,
+pub async fn read_exact_at_impl<B, F, Fut>(
+    buf: B,
     mut offset: u64,
+    count: Option<usize>,
     mut read_at: F,
-) -> (Buf, std::io::Result<()>)
+) -> (B, std::io::Result<()>)
 where
-    Buf: IoBufMut + Send,
-    F: FnMut(tokio_epoll_uring::Slice<Buf>, u64) -> Fut,
-    Fut: std::future::Future<Output = (tokio_epoll_uring::Slice<Buf>, std::io::Result<usize>)>,
+    B: IoBufMut + Send,
+    F: FnMut(tokio_epoll_uring::Slice<B>, u64) -> Fut,
+    Fut: std::future::Future<Output = (tokio_epoll_uring::Slice<B>, std::io::Result<usize>)>,
 {
+    let mut buf: tokio_epoll_uring::Slice<B> = match count {
+        Some(count) => {
+            assert!(count <= buf.bytes_total());
+            assert!(count > 0);
+            buf.slice(..count) // may include uninitialized memory
+        }
+        None => buf.slice_full(), // includes all the uninitialized memory
+    };
+
     while buf.bytes_total() != 0 {
         let res;
         (buf, res) = read_at(buf, offset).await;
@@ -872,7 +882,7 @@ mod test_read_exact_at_impl {
 
     #[tokio::test]
     async fn test_basic() {
-        let buf = Vec::with_capacity(5).slice_full();
+        let buf = Vec::with_capacity(5);
         let mock_read_at = Arc::new(tokio::sync::Mutex::new(MockReadAt {
             expectations: VecDeque::from(vec![Expectation {
                 offset: 0,
@@ -880,7 +890,7 @@ mod test_read_exact_at_impl {
                 result: Ok(vec![b'a', b'b', b'c', b'd', b'e']),
             }]),
         }));
-        let (buf, res) = read_exact_at_impl(buf, 0, |buf, offset| {
+        let (buf, res) = read_exact_at_impl(buf, 0, None, |buf, offset| {
             let mock_read_at = Arc::clone(&mock_read_at);
             async move { mock_read_at.lock().await.read_at(buf, offset).await }
         })
@@ -889,13 +899,33 @@ mod test_read_exact_at_impl {
         assert_eq!(buf, vec![b'a', b'b', b'c', b'd', b'e']);
     }
 
+    #[tokio::test]
+    async fn test_with_count() {
+        let buf = Vec::with_capacity(5);
+        let mock_read_at = Arc::new(tokio::sync::Mutex::new(MockReadAt {
+            expectations: VecDeque::from(vec![Expectation {
+                offset: 0,
+                bytes_total: 3,
+                result: Ok(vec![b'a', b'b', b'c']),
+            }]),
+        }));
+
+        let (buf, res) = read_exact_at_impl(buf, 0, Some(3), |buf, offset| {
+            let mock_read_at = Arc::clone(&mock_read_at);
+            async move { mock_read_at.lock().await.read_at(buf, offset).await }
+        })
+        .await;
+        assert!(res.is_ok());
+        assert_eq!(buf, vec![b'a', b'b', b'c']);
+    }
+
     #[tokio::test]
     async fn test_empty_buf_issues_no_syscall() {
-        let buf = Vec::new().slice_full();
+        let buf = Vec::new();
         let mock_read_at = Arc::new(tokio::sync::Mutex::new(MockReadAt {
             expectations: VecDeque::new(),
         }));
-        let (_buf, res) = read_exact_at_impl(buf, 0, |buf, offset| {
+        let (_buf, res) = read_exact_at_impl(buf, 0, None, |buf, offset| {
             let mock_read_at = Arc::clone(&mock_read_at);
             async move { mock_read_at.lock().await.read_at(buf, offset).await }
         })
@@ -905,7 +935,7 @@ mod test_read_exact_at_impl {
 
     #[tokio::test]
     async fn test_two_read_at_calls_needed_until_buf_filled() {
-        let buf = Vec::with_capacity(4).slice_full();
+        let buf = Vec::with_capacity(4);
         let mock_read_at = Arc::new(tokio::sync::Mutex::new(MockReadAt {
             expectations: VecDeque::from(vec![
                 Expectation {
@@ -920,7 +950,7 @@ mod test_read_exact_at_impl {
                 },
             ]),
         }));
-        let (buf, res) = read_exact_at_impl(buf, 0, |buf, offset| {
+        let (buf, res) = read_exact_at_impl(buf, 0, None, |buf, offset| {
             let mock_read_at = Arc::clone(&mock_read_at);
             async move { mock_read_at.lock().await.read_at(buf, offset).await }
         })
@@ -931,7 +961,7 @@ mod test_read_exact_at_impl {
 
     #[tokio::test]
     async fn test_eof_before_buffer_full() {
-        let buf = Vec::with_capacity(3).slice_full();
+        let buf = Vec::with_capacity(3);
         let mock_read_at = Arc::new(tokio::sync::Mutex::new(MockReadAt {
             expectations: VecDeque::from(vec![
                 Expectation {
@@ -951,7 +981,7 @@ mod test_read_exact_at_impl {
                 },
             ]),
         }));
-        let (_buf, res) = read_exact_at_impl(buf, 0, |buf, offset| {
+        let (_buf, res) = read_exact_at_impl(buf, 0, None, |buf, offset| {
             let mock_read_at = Arc::clone(&mock_read_at);
             async move { mock_read_at.lock().await.read_at(buf, offset).await }
         })
@@ -1021,29 +1051,27 @@ impl VirtualFile {
         ctx: &RequestContext,
     ) -> Result<crate::tenant::block_io::BlockLease<'_>, std::io::Error> {
         use crate::page_cache::PAGE_SZ;
-        let slice = Vec::with_capacity(PAGE_SZ).slice_full();
-        assert_eq!(slice.bytes_total(), PAGE_SZ);
-        let slice = self
-            .read_exact_at(slice, blknum as u64 * (PAGE_SZ as u64), ctx)
+        let buf = vec![0; PAGE_SZ];
+        let buf = self
+            .read_exact_at(buf, blknum as u64 * (PAGE_SZ as u64), ctx)
             .await?;
-        Ok(crate::tenant::block_io::BlockLease::Vec(slice.into_inner()))
+        Ok(crate::tenant::block_io::BlockLease::Vec(buf))
     }
 
     async fn read_to_end(&mut self, buf: &mut Vec<u8>, ctx: &RequestContext) -> Result<(), Error> {
         let mut tmp = vec![0; 128];
         loop {
-            let slice = tmp.slice(..128);
-            let (slice, res) = self.read_at(slice, self.pos, ctx).await;
+            let res;
+            (tmp, res) = self.read_at(tmp, self.pos, ctx).await;
             match res {
                 Ok(0) => return Ok(()),
                 Ok(n) => {
                     self.pos += n as u64;
-                    buf.extend_from_slice(&slice[..n]);
+                    buf.extend_from_slice(&tmp[..n]);
                 }
                 Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => {}
                 Err(e) => return Err(e),
             }
-            tmp = slice.into_inner();
         }
     }
 }
@@ -1157,7 +1185,6 @@ mod tests {
     use crate::task_mgr::TaskKind;
 
     use super::*;
-    use owned_buffers_io::slice::SliceExt;
     use rand::seq::SliceRandom;
     use rand::thread_rng;
     use rand::Rng;
@@ -1179,16 +1206,13 @@ mod tests {
     impl MaybeVirtualFile {
         async fn read_exact_at(
             &self,
-            mut slice: tokio_epoll_uring::Slice<Vec<u8>>,
+            mut buf: Vec<u8>,
             offset: u64,
             ctx: &RequestContext,
-        ) -> Result<tokio_epoll_uring::Slice<Vec<u8>>, Error> {
+        ) -> Result<Vec<u8>, Error> {
             match self {
-                MaybeVirtualFile::VirtualFile(file) => file.read_exact_at(slice, offset, ctx).await,
-                MaybeVirtualFile::File(file) => {
-                    let rust_slice: &mut [u8] = slice.as_mut_rust_slice_full_zeroed();
-                    file.read_exact_at(rust_slice, offset).map(|()| slice)
-                }
+                MaybeVirtualFile::VirtualFile(file) => file.read_exact_at(buf, offset, ctx).await,
+                MaybeVirtualFile::File(file) => file.read_exact_at(&mut buf, offset).map(|()| buf),
             }
         }
         async fn write_all_at<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
@@ -1262,12 +1286,9 @@ mod tests {
             len: usize,
             ctx: &RequestContext,
         ) -> Result<String, Error> {
-            let slice = Vec::with_capacity(len).slice_full();
-            assert_eq!(slice.bytes_total(), len);
-            let slice = self.read_exact_at(slice, pos, ctx).await?;
-            let vec = slice.into_inner();
-            assert_eq!(vec.len(), len);
-            Ok(String::from_utf8(vec).unwrap())
+            let buf = vec![0; len];
+            let buf = self.read_exact_at(buf, pos, ctx).await?;
+            Ok(String::from_utf8(buf).unwrap())
         }
     }
 
@@ -1486,11 +1507,7 @@ mod tests {
                 let mut rng = rand::rngs::OsRng;
                 for _ in 1..1000 {
                     let f = &files[rng.gen_range(0..files.len())];
-                    buf = f
-                        .read_exact_at(buf.slice_full(), 0, &ctx)
-                        .await
-                        .unwrap()
-                        .into_inner();
+                    buf = f.read_exact_at(buf, 0, &ctx).await.unwrap();
                     assert!(buf == SAMPLE);
                 }
             });

pageserver/src/virtual_file/io_engine.rs

@@ -107,7 +107,7 @@ use std::{
     sync::atomic::{AtomicU8, Ordering},
 };
 
-use super::{owned_buffers_io::slice::SliceExt, FileGuard, Metadata};
+use super::{FileGuard, Metadata};
 
 #[cfg(target_os = "linux")]
 fn epoll_uring_error_to_std(e: tokio_epoll_uring::Error<std::io::Error>) -> std::io::Error {
@@ -120,29 +120,38 @@ fn epoll_uring_error_to_std(e: tokio_epoll_uring::Error<std::io::Error>) -> std:
 }
 
 impl IoEngine {
-    pub(super) async fn read_at<Buf>(
+    pub(super) async fn read_at<B>(
         &self,
         file_guard: FileGuard,
         offset: u64,
-        mut slice: tokio_epoll_uring::Slice<Buf>,
-    ) -> (
-        (FileGuard, tokio_epoll_uring::Slice<Buf>),
-        std::io::Result<usize>,
-    )
+        mut buf: B,
+    ) -> ((FileGuard, B), std::io::Result<usize>)
     where
-        Buf: tokio_epoll_uring::IoBufMut + Send,
+        B: tokio_epoll_uring::BoundedBufMut + Send,
     {
         match self {
             IoEngine::NotSet => panic!("not initialized"),
             IoEngine::StdFs => {
-                let rust_slice = slice.as_mut_rust_slice_full_zeroed();
-                let res = file_guard.with_std_file(|std_file| std_file.read_at(rust_slice, offset));
-                ((file_guard, slice), res)
+                // SAFETY: `dst` only lives at most as long as this match arm, during which buf remains valid memory.
+                let dst = unsafe {
+                    std::slice::from_raw_parts_mut(buf.stable_mut_ptr(), buf.bytes_total())
+                };
+                let res = file_guard.with_std_file(|std_file| std_file.read_at(dst, offset));
+                if let Ok(nbytes) = &res {
+                    assert!(*nbytes <= buf.bytes_total());
+                    // SAFETY: see above assertion
+                    unsafe {
+                        buf.set_init(*nbytes);
+                    }
+                }
+                #[allow(dropping_references)]
+                drop(dst);
+                ((file_guard, buf), res)
             }
             #[cfg(target_os = "linux")]
             IoEngine::TokioEpollUring => {
                 let system = tokio_epoll_uring_ext::thread_local_system().await;
-                let (resources, res) = system.read(file_guard, offset, slice).await;
+                let (resources, res) = system.read(file_guard, offset, buf).await;
                 (resources, res.map_err(epoll_uring_error_to_std))
             }
         }

pageserver/src/virtual_file/owned_buffers_io/slice.rs

@@ -1,121 +0,0 @@
-use tokio_epoll_uring::BoundedBuf;
-use tokio_epoll_uring::BoundedBufMut;
-use tokio_epoll_uring::IoBufMut;
-use tokio_epoll_uring::Slice;
-
-pub(crate) trait SliceExt {
-    /// Get a `&mut[0..self.bytes_total()`] slice, for when you need to do borrow-based IO.
-    ///
-    /// See the test case `test_slice_full_zeroed` for the difference to just doing `&slice[..]`
-    fn as_mut_rust_slice_full_zeroed(&mut self) -> &mut [u8];
-}
-
-impl<B> SliceExt for Slice<B>
-where
-    B: IoBufMut,
-{
-    #[inline(always)]
-    fn as_mut_rust_slice_full_zeroed(&mut self) -> &mut [u8] {
-        // zero-initialize the uninitialized parts of the buffer so we can create a Rust slice
-        //
-        // SAFETY: we own `slice`, don't write outside the bounds
-        unsafe {
-            let to_init = self.bytes_total() - self.bytes_init();
-            self.stable_mut_ptr()
-                .add(self.bytes_init())
-                .write_bytes(0, to_init);
-            self.set_init(self.bytes_total());
-        };
-        let bytes_total = self.bytes_total();
-        &mut self[0..bytes_total]
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use std::io::Read;
-
-    use super::*;
-    use bytes::Buf;
-    use tokio_epoll_uring::Slice;
-
-    #[test]
-    fn test_slice_full_zeroed() {
-        let make_fake_file = || bytes::BytesMut::from(&b"12345"[..]).reader();
-
-        // before we start the test, let's make sure we have a shared understanding of what slice_full does
-        {
-            let buf = Vec::with_capacity(3);
-            let slice: Slice<_> = buf.slice_full();
-            assert_eq!(slice.bytes_init(), 0);
-            assert_eq!(slice.bytes_total(), 3);
-            let rust_slice = &slice[..];
-            assert_eq!(
-                rust_slice.len(),
-                0,
-                "Slice only derefs to a &[u8] of the initialized part"
-            );
-        }
-
-        // and also let's establish a shared understanding of .slice()
-        {
-            let buf = Vec::with_capacity(3);
-            let slice: Slice<_> = buf.slice(0..2);
-            assert_eq!(slice.bytes_init(), 0);
-            assert_eq!(slice.bytes_total(), 2);
-            let rust_slice = &slice[..];
-            assert_eq!(
-                rust_slice.len(),
-                0,
-                "Slice only derefs to a &[u8] of the initialized part"
-            );
-        }
-
-        // the above leads to the easy mistake of using slice[..] for borrow-based IO like so:
-        {
-            let buf = Vec::with_capacity(3);
-            let mut slice: Slice<_> = buf.slice_full();
-            assert_eq!(slice[..].len(), 0);
-            let mut file = make_fake_file();
-            file.read_exact(&mut slice[..]).unwrap(); // one might think this reads 3 bytes but it reads 0
-            assert_eq!(&slice[..] as &[u8], &[][..] as &[u8]);
-        }
-
-        // With owned buffers IO like with VirtualFilem, you could totally
-        // pass in a `Slice` with bytes_init()=0 but bytes_total()=5
-        // and it will read 5 bytes into the slice, and return a slice that has bytes_init()=5.
-        {
-            // TODO: demo
-        }
-
-        //
-        // Ok, now that we have a shared understanding let's demo how to use the extension trait.
-        //
-
-        // slice_full()
-        {
-            let buf = Vec::with_capacity(3);
-            let mut slice: Slice<_> = buf.slice_full();
-            let rust_slice = slice.as_mut_rust_slice_full_zeroed();
-            assert_eq!(rust_slice.len(), 3);
-            assert_eq!(rust_slice, &[0, 0, 0]);
-            let mut file = make_fake_file();
-            file.read_exact(rust_slice).unwrap();
-            assert_eq!(rust_slice, b"123");
-            assert_eq!(&slice[..], b"123");
-        }
-
-        // .slice(..)
-        {
-            let buf = Vec::with_capacity(3);
-            let mut slice: Slice<_> = buf.slice(0..2);
-            let rust_slice = slice.as_mut_rust_slice_full_zeroed();
-            assert_eq!(rust_slice.len(), 2);
-            assert_eq!(rust_slice, &[0, 0]);
-            let mut file = make_fake_file();
-            file.read_exact(rust_slice).unwrap();
-            assert_eq!(rust_slice, b"12");
-            assert_eq!(&slice[..], b"12");
-        }
-    }
-}

proxy/src/auth/backend.rs

@@ -153,7 +153,7 @@ pub struct ComputeUserInfo {
 
 impl ComputeUserInfo {
     pub fn endpoint_cache_key(&self) -> EndpointCacheKey {
-        self.options.get_cache_key(&self.endpoint)
+        self.options.get_cache_key((&self.endpoint).into())
     }
 }
 

proxy/src/auth/credentials.rs

@@ -241,6 +241,8 @@ fn project_name_valid(name: &str) -> bool {
 
 #[cfg(test)]
 mod tests {
+    use crate::intern::EndpointIdInt;
+
     use super::*;
     use serde_json::json;
     use ComputeUserInfoParseError::*;
@@ -284,7 +286,6 @@ mod tests {
             ComputeUserInfoMaybeEndpoint::parse(&mut ctx, &options, sni, common_names.as_ref())?;
         assert_eq!(user_info.user, "john_doe");
         assert_eq!(user_info.endpoint_id.as_deref(), Some("foo"));
-        assert_eq!(user_info.options.get_cache_key("foo"), "foo");
 
         Ok(())
     }
@@ -442,8 +443,9 @@ mod tests {
         let user_info =
             ComputeUserInfoMaybeEndpoint::parse(&mut ctx, &options, sni, common_names.as_ref())?;
         assert_eq!(user_info.endpoint_id.as_deref(), Some("project"));
+        let project = EndpointIdInt::from(EndpointId::from("project"));
         assert_eq!(
-            user_info.options.get_cache_key("project"),
+            user_info.options.get_cache_key(project).to_string(),
             "project endpoint_type:read_write lsn:0/2"
         );
 

proxy/src/cache/common.rs

@@ -43,6 +43,15 @@ impl<C: Cache, V> Cached<C, V> {
         Self { token: None, value }
     }
 
+    /// Place any entry into this wrapper; invalidation will be a no-op.
+    pub fn map<U>(self, f: impl FnOnce(V) -> U) -> Cached<C, U> {
+        let token = self.token;
+        Cached {
+            token,
+            value: f(self.value),
+        }
+    }
+
     pub fn take_value(self) -> (Cached<C, ()>, V) {
         (
             Cached {

proxy/src/compute.rs

@@ -93,7 +93,7 @@ pub type ScramKeys = tokio_postgres::config::ScramKeys<32>;
 /// Eventually, `tokio_postgres` will be replaced with something better.
 /// Newtype allows us to implement methods on top of it.
 #[derive(Clone, Default)]
-pub struct ConnCfg(Box<tokio_postgres::Config>);
+pub struct ConnCfg(tokio_postgres::Config);
 
 /// Creation and initialization routines.
 impl ConnCfg {

proxy/src/console/provider.rs

@@ -9,14 +9,14 @@ use crate::{
         IpPattern,
     },
     cache::{endpoints::EndpointsCache, project_info::ProjectInfoCacheImpl, Cached, TimedLru},
-    compute,
+    compute::{self, ConnCfg},
     config::{CacheOptions, EndpointCacheConfig, ProjectInfoCacheOptions},
     context::RequestMonitoring,
     error::ReportableError,
     intern::ProjectIdInt,
     metrics::ApiLockMetrics,
     rate_limiter::{DynamicLimiter, Outcome, RateLimiterConfig, Token},
-    scram, EndpointCacheKey,
+    scram, EndpointCacheKey, Host,
 };
 use dashmap::DashMap;
 use std::{hash::Hash, sync::Arc, time::Duration};
@@ -289,6 +289,33 @@ pub struct NodeInfo {
     pub allow_self_signed_compute: bool,
 }
 
+/// Cached info for establishing a connection to a compute node.
+#[derive(Clone)]
+pub struct NodeCachedInfo {
+    pub host: Host,
+    pub port: u16,
+
+    /// Labels for proxy's metrics.
+    pub aux: MetricsAuxInfo,
+
+    /// Whether we should accept self-signed certificates (for testing)
+    pub allow_self_signed_compute: bool,
+}
+
+impl NodeCachedInfo {
+    pub fn into_node_info(self) -> NodeInfo {
+        let mut config = ConnCfg::default();
+        config.ssl_mode(tokio_postgres::config::SslMode::Disable);
+        config.host(&self.host);
+        config.port(self.port);
+        NodeInfo {
+            config,
+            aux: self.aux,
+            allow_self_signed_compute: self.allow_self_signed_compute,
+        }
+    }
+}
+
 impl NodeInfo {
     pub async fn connect(
         &self,
@@ -317,8 +344,8 @@ impl NodeInfo {
     }
 }
 
-pub type NodeInfoCache = TimedLru<EndpointCacheKey, NodeInfo>;
-pub type CachedNodeInfo = Cached<&'static NodeInfoCache>;
+pub type NodeInfoCache = TimedLru<EndpointCacheKey, NodeCachedInfo>;
+pub type CachedNodeInfo = Cached<&'static NodeInfoCache, NodeInfo>;
 pub type CachedRoleSecret = Cached<&'static ProjectInfoCacheImpl, Option<AuthSecret>>;
 pub type CachedAllowedIps = Cached<&'static ProjectInfoCacheImpl, Arc<Vec<IpPattern>>>;
 

proxy/src/console/provider/neon.rs

@@ -4,22 +4,20 @@ use super::{
     super::messages::{ConsoleError, GetRoleSecret, WakeCompute},
     errors::{ApiError, GetAuthInfoError, WakeComputeError},
     ApiCaches, ApiLocks, AuthInfo, AuthSecret, CachedAllowedIps, CachedNodeInfo, CachedRoleSecret,
-    NodeInfo,
+    NodeCachedInfo,
 };
 use crate::{
     auth::backend::ComputeUserInfo,
-    compute,
     console::messages::ColdStartInfo,
     http,
     metrics::{CacheOutcome, Metrics},
     rate_limiter::EndpointRateLimiter,
-    scram, EndpointCacheKey,
+    scram, EndpointCacheKey, Host,
 };
 use crate::{cache::Cached, context::RequestMonitoring};
 use futures::TryFutureExt;
 use std::sync::Arc;
 use tokio::time::Instant;
-use tokio_postgres::config::SslMode;
 use tracing::{error, info, info_span, warn, Instrument};
 
 pub struct Api {
@@ -132,7 +130,7 @@ impl Api {
         &self,
         ctx: &mut RequestMonitoring,
         user_info: &ComputeUserInfo,
-    ) -> Result<NodeInfo, WakeComputeError> {
+    ) -> Result<NodeCachedInfo, WakeComputeError> {
         let request_id = ctx.session_id.to_string();
         let application_name = ctx.console_application_name();
         async {
@@ -167,15 +165,11 @@ impl Api {
                 None => return Err(WakeComputeError::BadComputeAddress(body.address)),
                 Some(x) => x,
             };
+            let host = Host(host.into());
 
-            // Don't set anything but host and port! This config will be cached.
-            // We'll set username and such later using the startup message.
-            // TODO: add more type safety (in progress).
-            let mut config = compute::ConnCfg::new();
-            config.host(host).port(port).ssl_mode(SslMode::Disable); // TLS is not configured on compute nodes.
-
-            let node = NodeInfo {
-                config,
+            let node = NodeCachedInfo {
+                host,
+                port,
                 aux: body.aux,
                 allow_self_signed_compute: false,
             };
@@ -278,9 +272,9 @@ impl super::Api for Api {
         // The connection info remains the same during that period of time,
         // which means that we might cache it to reduce the load and latency.
         if let Some(cached) = self.caches.node_info.get(&key) {
-            info!(key = &*key, "found cached compute node info");
+            info!(key = display(&key), "found cached compute node info");
             ctx.set_project(cached.aux.clone());
-            return Ok(cached);
+            return Ok(cached.map(NodeCachedInfo::into_node_info));
         }
 
         let permit = self.locks.get_permit(&key).await?;
@@ -289,9 +283,9 @@ impl super::Api for Api {
         // double check
         if permit.should_check_cache() {
             if let Some(cached) = self.caches.node_info.get(&key) {
-                info!(key = &*key, "found cached compute node info");
+                info!(key = display(&key), "found cached compute node info");
                 ctx.set_project(cached.aux.clone());
-                return Ok(cached);
+                return Ok(cached.map(NodeCachedInfo::into_node_info));
             }
         }
 
@@ -300,7 +294,7 @@ impl super::Api for Api {
             .wake_compute_endpoint_rate_limiter
             .check(user_info.endpoint.normalize_intern(), 1)
         {
-            info!(key = &*key, "found cached compute node info");
+            info!(key = display(&key), "found cached compute node info");
             return Err(WakeComputeError::TooManyConnections);
         }
 
@@ -314,9 +308,12 @@ impl super::Api for Api {
         let (_, mut cached) = self.caches.node_info.insert(key.clone(), node);
         cached.aux.cold_start_info = cold_start_info;
 
-        info!(key = &*key, "created a cache entry for compute node info");
+        info!(
+            key = display(&key),
+            "created a cache entry for compute node info"
+        );
 
-        Ok(cached)
+        Ok(cached.map(NodeCachedInfo::into_node_info))
     }
 }
 

proxy/src/lib.rs

@@ -157,8 +157,16 @@ smol_str_wrapper!(BranchId);
 // 90% of project strings are 23 characters or less.
 smol_str_wrapper!(ProjectId);
 
-// will usually equal endpoint ID
-smol_str_wrapper!(EndpointCacheKey);
+#[derive(PartialEq, Eq, Hash, Debug, Clone)]
+pub struct EndpointCacheKey {
+    pub id: EndpointIdInt,
+    pub extra: Box<str>,
+}
+impl std::fmt::Display for EndpointCacheKey {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}{}", &self.id, &self.extra)
+    }
+}
 
 smol_str_wrapper!(DbName);
 

proxy/src/proxy.rs

@@ -10,6 +10,7 @@ pub mod wake_compute;
 pub use copy_bidirectional::copy_bidirectional_client_compute;
 pub use copy_bidirectional::ErrorSource;
 
+use crate::intern::EndpointIdInt;
 use crate::{
     auth,
     cancellation::{self, CancellationHandlerMain, CancellationHandlerMainInternal},
@@ -404,13 +405,20 @@ impl NeonOptions {
         Self(options)
     }
 
-    pub fn get_cache_key(&self, prefix: &str) -> EndpointCacheKey {
-        // prefix + format!(" {k}:{v}")
-        // kinda jank because SmolStr is immutable
-        std::iter::once(prefix)
-            .chain(self.0.iter().flat_map(|(k, v)| [" ", &**k, ":", &**v]))
-            .collect::<SmolStr>()
-            .into()
+    pub fn get_cache_key(&self, endpoint: EndpointIdInt) -> EndpointCacheKey {
+        EndpointCacheKey {
+            id: endpoint,
+            extra: self.get_cache_key_extras(),
+        }
+    }
+
+    pub fn get_cache_key_extras(&self) -> Box<str> {
+        let mut extras = String::new();
+        for (k, v) in &self.0 {
+            use std::fmt::Write;
+            write!(&mut extras, " {k}:{v}").unwrap();
+        }
+        extras.into_boxed_str()
     }
 
     /// <https://swagger.io/docs/specification/serialization/> DeepObject format

proxy/src/proxy/connect_compute.rs

@@ -47,7 +47,7 @@ pub trait ConnectMechanism {
     async fn connect_once(
         &self,
         ctx: &mut RequestMonitoring,
-        node_info: &console::CachedNodeInfo,
+        node_info: &NodeInfo,
         timeout: time::Duration,
     ) -> Result<Self::Connection, Self::ConnectError>;
 
@@ -82,7 +82,7 @@ impl ConnectMechanism for TcpMechanism<'_> {
     async fn connect_once(
         &self,
         ctx: &mut RequestMonitoring,
-        node_info: &console::CachedNodeInfo,
+        node_info: &NodeInfo,
         timeout: time::Duration,
     ) -> Result<PostgresConnection, Self::Error> {
         let host = node_info.config.get_host()?;

proxy/src/proxy/tests.rs

@@ -13,8 +13,10 @@ use crate::auth::backend::{
 use crate::config::{CertResolver, RetryConfig};
 use crate::console::caches::NodeInfoCache;
 use crate::console::messages::{ConsoleError, Details, MetricsAuxInfo, Status};
-use crate::console::provider::{CachedAllowedIps, CachedRoleSecret, ConsoleBackend};
-use crate::console::{self, CachedNodeInfo, NodeInfo};
+use crate::console::provider::{
+    CachedAllowedIps, CachedRoleSecret, ConsoleBackend, NodeCachedInfo,
+};
+use crate::console::{self, CachedNodeInfo};
 use crate::error::ErrorKind;
 use crate::{http, sasl, scram, BranchId, EndpointId, ProjectId};
 use anyhow::{bail, Context};
@@ -458,7 +460,7 @@ impl ConnectMechanism for TestConnectMechanism {
     async fn connect_once(
         &self,
         _ctx: &mut RequestMonitoring,
-        _node_info: &console::CachedNodeInfo,
+        _node_info: &console::NodeInfo,
         _timeout: std::time::Duration,
     ) -> Result<Self::Connection, Self::ConnectError> {
         let mut counter = self.counter.lock().unwrap();
@@ -530,8 +532,9 @@ impl TestBackend for TestConnectMechanism {
 }
 
 fn helper_create_cached_node_info(cache: &'static NodeInfoCache) -> CachedNodeInfo {
-    let node = NodeInfo {
-        config: compute::ConnCfg::new(),
+    let node = NodeCachedInfo {
+        host: "localhost".into(),
+        port: 5432,
         aux: MetricsAuxInfo {
             endpoint_id: (&EndpointId::from("endpoint")).into(),
             project_id: (&ProjectId::from("project")).into(),
@@ -540,8 +543,12 @@ fn helper_create_cached_node_info(cache: &'static NodeInfoCache) -> CachedNodeIn
         },
         allow_self_signed_compute: false,
     };
-    let (_, node) = cache.insert("key".into(), node);
-    node
+    let key = EndpointCacheKey {
+        id: node.aux.endpoint_id,
+        extra: "".into(),
+    };
+    let (_, node) = cache.insert(key, node);
+    node.map(NodeCachedInfo::into_node_info)
 }
 
 fn helper_create_connect_info(

proxy/src/serverless/backend.rs

@@ -11,7 +11,7 @@ use crate::{
         errors::{GetAuthInfoError, WakeComputeError},
         locks::ApiLocks,
         provider::ApiLockError,
-        CachedNodeInfo,
+        NodeInfo,
     },
     context::RequestMonitoring,
     error::{ErrorKind, ReportableError, UserFacingError},
@@ -223,7 +223,7 @@ impl ConnectMechanism for TokioMechanism {
     async fn connect_once(
         &self,
         ctx: &mut RequestMonitoring,
-        node_info: &CachedNodeInfo,
+        node_info: &NodeInfo,
         timeout: Duration,
     ) -> Result<Self::Connection, Self::ConnectError> {
         let host = node_info.config.get_host()?;

proxy/src/serverless/conn_pool.rs

@@ -61,7 +61,7 @@ impl fmt::Display for ConnInfo {
             self.user_info.user,
             self.user_info.endpoint,
             self.dbname,
-            self.user_info.options.get_cache_key("")
+            self.user_info.options.get_cache_key_extras()
         )
     }
 }

safekeeper/src/metrics.rs

@@ -72,8 +72,7 @@ pub static WAL_STORAGE_OPERATION_SECONDS: Lazy<HistogramVec> = Lazy::new(|| {
     register_histogram_vec!(
         "safekeeper_wal_storage_operation_seconds",
         "Seconds spent on WAL storage operations",
-        &["operation"],
-        DISK_FSYNC_SECONDS_BUCKETS.to_vec()
+        &["operation"]
     )
     .expect("Failed to register safekeeper_wal_storage_operation_seconds histogram vec")
 });
@@ -81,8 +80,7 @@ pub static MISC_OPERATION_SECONDS: Lazy<HistogramVec> = Lazy::new(|| {
     register_histogram_vec!(
         "safekeeper_misc_operation_seconds",
         "Seconds spent on miscellaneous operations",
-        &["operation"],
-        DISK_FSYNC_SECONDS_BUCKETS.to_vec()
+        &["operation"]
     )
     .expect("Failed to register safekeeper_misc_operation_seconds histogram vec")
 });

storage_controller/src/http.rs

@@ -10,9 +10,8 @@ use hyper::header::CONTENT_TYPE;
 use hyper::{Body, Request, Response};
 use hyper::{StatusCode, Uri};
 use metrics::{BuildInfo, NeonMetrics};
-use pageserver_api::controller_api::TenantCreateRequest;
 use pageserver_api::models::{
-    TenantConfigRequest, TenantLocationConfigRequest, TenantShardSplitRequest,
+    TenantConfigRequest, TenantCreateRequest, TenantLocationConfigRequest, TenantShardSplitRequest,
     TenantTimeTravelRequest, TimelineCreateRequest,
 };
 use pageserver_api::shard::TenantShardId;

storage_controller/src/service.rs

@@ -32,10 +32,10 @@ use itertools::Itertools;
 use pageserver_api::{
     controller_api::{
         NodeAvailability, NodeRegisterRequest, NodeSchedulingPolicy, PlacementPolicy,
-        ShardSchedulingPolicy, TenantCreateRequest, TenantCreateResponse,
-        TenantCreateResponseShard, TenantDescribeResponse, TenantDescribeResponseShard,
-        TenantLocateResponse, TenantPolicyRequest, TenantShardMigrateRequest,
-        TenantShardMigrateResponse, UtilizationScore,
+        ShardSchedulingPolicy, TenantCreateResponse, TenantCreateResponseShard,
+        TenantDescribeResponse, TenantDescribeResponseShard, TenantLocateResponse,
+        TenantPolicyRequest, TenantShardMigrateRequest, TenantShardMigrateResponse,
+        UtilizationScore,
     },
     models::{SecondaryProgress, TenantConfigRequest, TopTenantShardsRequest},
 };
@@ -46,9 +46,10 @@ use crate::pageserver_client::PageserverClient;
 use pageserver_api::{
     models::{
         self, LocationConfig, LocationConfigListResponse, LocationConfigMode,
-        PageserverUtilization, ShardParameters, TenantConfig, TenantLocationConfigRequest,
-        TenantLocationConfigResponse, TenantShardLocation, TenantShardSplitRequest,
-        TenantShardSplitResponse, TenantTimeTravelRequest, TimelineCreateRequest, TimelineInfo,
+        PageserverUtilization, ShardParameters, TenantConfig, TenantCreateRequest,
+        TenantLocationConfigRequest, TenantLocationConfigResponse, TenantShardLocation,
+        TenantShardSplitRequest, TenantShardSplitResponse, TenantTimeTravelRequest,
+        TimelineCreateRequest, TimelineInfo,
     },
     shard::{ShardCount, ShardIdentity, ShardNumber, ShardStripeSize, TenantShardId},
     upcall_api::{

test_runner/fixtures/neon_fixtures.py

@@ -6,7 +6,6 @@ import concurrent.futures
 import filecmp
 import json
 import os
-import random
 import re
 import shutil
 import subprocess
@@ -2742,19 +2741,7 @@ class NeonPageserver(PgProtocol, LogUtils):
         if generation is None:
             generation = self.env.storage_controller.attach_hook_issue(tenant_id, self.id)
         client = self.http_client(auth_token=auth_token)
-
-        conf = conf or {}
-
-        client.tenant_location_conf(
-            tenant_id,
-            {
-                "mode": "AttachedSingle",
-                "generation": generation,
-                "tenant_conf": conf,
-                "secondary_conf": None,
-            },
-        )
-        return tenant_id
+        return client.tenant_create(tenant_id, conf, generation=generation)
 
     def list_layers(
         self, tenant_id: Union[TenantId, TenantShardId], timeline_id: TimelineId
@@ -2789,33 +2776,6 @@ class NeonPageserver(PgProtocol, LogUtils):
         layers = self.list_layers(tenant_id, timeline_id)
         return layer_name in [parse_layer_file_name(p.name) for p in layers]
 
-    def evict_random_layers(
-        self, rng: random.Random, tenant_id: TenantId, timeline_id: TimelineId, ratio: float
-    ):
-        """
-        Evict 50% of the layers on a pageserver
-        """
-        timeline_path = self.timeline_dir(tenant_id, timeline_id)
-        initial_local_layers = sorted(
-            list(filter(lambda path: path.name != "metadata", timeline_path.glob("*")))
-        )
-        client = self.http_client()
-        for layer in initial_local_layers:
-            if (
-                "ephemeral" in layer.name
-                or "temp_download" in layer.name
-                or "___temp" in layer.name
-            ):
-                continue
-
-            layer_name = parse_layer_file_name(layer.name)
-
-            if rng.uniform(0.0, 1.0) < ratio:
-                log.info(f"Evicting layer {tenant_id}/{timeline_id} {layer_name.to_str()}")
-                client.evict_layer(
-                    tenant_id=tenant_id, timeline_id=timeline_id, layer_name=layer_name.to_str()
-                )
-
 
 class PgBin:
     """A helper class for executing postgres binaries"""

test_runner/fixtures/pageserver/http.py

@@ -220,6 +220,34 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
         assert isinstance(res_json, list)
         return res_json
 
+    def tenant_create(
+        self,
+        new_tenant_id: Union[TenantId, TenantShardId],
+        conf: Optional[Dict[str, Any]] = None,
+        generation: Optional[int] = None,
+    ) -> TenantId:
+        if conf is not None:
+            assert "new_tenant_id" not in conf.keys()
+
+        body: Dict[str, Any] = {
+            "new_tenant_id": str(new_tenant_id),
+            **(conf or {}),
+        }
+
+        if generation is not None:
+            body.update({"generation": generation})
+
+        res = self.post(
+            f"http://localhost:{self.port}/v1/tenant",
+            json=body,
+        )
+        self.verbose_error(res)
+        if res.status_code == 409:
+            raise Exception(f"could not create tenant: already exists for id {new_tenant_id}")
+        new_tenant_id = res.json()
+        assert isinstance(new_tenant_id, str)
+        return TenantId(new_tenant_id)
+
     def tenant_attach(
         self,
         tenant_id: Union[TenantId, TenantShardId],

test_runner/regress/test_pageserver_api.py

@@ -85,10 +85,8 @@ def check_client(env: NeonEnv, client: PageserverHttpClient):
 
     # create new tenant and check it is also there
     tenant_id = TenantId.generate()
-    env.pageserver.tenant_create(
-        tenant_id,
-        generation=env.storage_controller.attach_hook_issue(tenant_id, env.pageserver.id),
-        auth_token=client.auth_token,
+    client.tenant_create(
+        tenant_id, generation=env.storage_controller.attach_hook_issue(tenant_id, env.pageserver.id)
     )
     assert tenant_id in {TenantId(t["id"]) for t in client.tenant_list()}
 

test_runner/regress/test_pageserver_restarts_under_workload.py

@@ -1,21 +1,16 @@
 # This test spawns pgbench in a thread in the background and concurrently restarts pageserver,
 # checking how client is able to transparently restore connection to pageserver
 #
-import random
 import threading
 import time
 
-import pytest
-from fixtures.common_types import TenantShardId
 from fixtures.log_helper import log
-from fixtures.neon_fixtures import NeonEnv, NeonEnvBuilder, PgBin
-from fixtures.remote_storage import RemoteStorageKind
-from fixtures.utils import wait_until
+from fixtures.neon_fixtures import NeonEnv, PgBin
 
 
 # Test restarting page server, while safekeeper and compute node keep
 # running.
-def test_pageserver_restarts_under_workload(neon_simple_env: NeonEnv, pg_bin: PgBin):
+def test_pageserver_restarts_under_worload(neon_simple_env: NeonEnv, pg_bin: PgBin):
     env = neon_simple_env
     env.neon_cli.create_branch("test_pageserver_restarts")
     endpoint = env.endpoints.create_start("test_pageserver_restarts")
@@ -37,149 +32,3 @@ def test_pageserver_restarts_under_workload(neon_simple_env: NeonEnv, pg_bin: Pg
         env.pageserver.start()
 
     thread.join()
-
-
-@pytest.mark.timeout(600)
-def test_pageserver_migration_under_workload(neon_env_builder: NeonEnvBuilder, pg_bin: PgBin):
-    tenant_conf = {
-        # We want layer rewrites to happen as soon as possible (this is the most stressful
-        # case for the system), so set PITR interval to something tiny.
-        "pitr_interval": "5s",
-        # Scaled down thresholds.  We will run at ~1GB scale but would like to emulate
-        # the behavior of a system running at ~100GB scale.
-        "checkpoint_distance": f"{1024 * 1024}",
-        "compaction_threshold": "1",
-        "compaction_target_size": f"{1024 * 1024}",
-        "image_creation_threshold": "2",
-        "image_layer_creation_check_threshold": "0",
-        # Do compaction & GC super frequently so that they are likely to be running while we are migrating
-        "compaction_period": "1s",
-        "gc_period": "1s",
-    }
-
-    neon_env_builder.num_pageservers = 2
-    neon_env_builder.num_safekeepers = 3
-    neon_env_builder.enable_pageserver_remote_storage(
-        RemoteStorageKind.MOCK_S3,
-    )
-    env = neon_env_builder.init_start(initial_tenant_conf=tenant_conf)
-
-    for ps in env.pageservers:
-        ps.allowed_errors.extend(
-            [
-                ".*Dropped remote consistent LSN updates.*",
-                ".*Dropping stale deletions.*",
-                # A stress test may hit the flush_ms timeout while transitioning a tenant to AttachedStale
-                ".*Timed out waiting for flush to remote storage, proceeding anyway.*",
-            ]
-        )
-
-    timeline_id = env.neon_cli.create_branch("branch")
-    endpoint = env.endpoints.create_start("branch")
-    n_migrations = 50
-    scale = 100
-
-    test_stop = threading.Event()
-
-    def run_pgbench(connstr: str):
-        ex = None
-        while not test_stop.is_set():
-            log.info(f"Starting pgbench on {connstr}")
-            try:
-                pg_bin.run_capture(["pgbench", "-i", f"-s{scale}", connstr])
-                pg_bin.run_capture(["pgbench", "-T10", connstr])
-            except Exception as e:
-                # We don't stop the loop running pgbench when it fails: we still want to run the test
-                # to completion to see if the pageserver itself hit any issues when under load, even if
-                # the client is not getting a clean experience.
-                ex = e
-                log.warning(f"pgbench failed: {e}")
-
-        if ex is not None:
-            raise ex
-
-    def safekeeper_restarts():
-        while not test_stop.is_set():
-            for safekeeper in env.safekeepers:
-                if test_stop.is_set():
-                    break
-
-                log.info(f"Restarting safekeeper {safekeeper.id}")
-                safekeeper.stop()
-                time.sleep(2)
-                safekeeper.start()
-                time.sleep(5)
-
-    # Run pgbench in the background to ensure some data is there to ingest, and some getpage requests are running
-    pgbench_thread = threading.Thread(target=run_pgbench, args=(endpoint.connstr(),), daemon=True)
-    pgbench_thread.start()
-
-    # Restart safekeepers in the background, so that pageserver's ingestion is not totally smooth, and sometimes
-    # during a tenant shutdown we might be in the process of finding a new safekeeper to download from
-    safekeeper_thread = threading.Thread(target=safekeeper_restarts, daemon=True)
-    safekeeper_thread.start()
-
-    # TODO: inject some randomized timing noise:
-    # - before-downloading-layer-stream-pausable
-    # - flush-frozen-pausable
-
-    def assert_getpage_reqs(ps_attached, init_getpage_reqs):
-        """
-        Assert that the getpage request counter has advanced on a particular node
-        """
-        getpage_reqs = (
-            env.get_pageserver(ps_attached)
-            .http_client()
-            .get_metric_value(
-                "pageserver_smgr_query_seconds_global_count", {"smgr_query_type": "get_page_at_lsn"}
-            )
-        )
-        log.info(f"getpage reqs({ps_attached}): {getpage_reqs} (vs init {init_getpage_reqs})")
-        assert getpage_reqs is not None
-        assert getpage_reqs > init_getpage_reqs
-
-    rng = random.Random(0xDEADBEEF)
-
-    # Ping-pong the tenant between pageservers: this repeatedly exercises the transitions between secondarry and
-    # attached, and implicitly exercises the tenant/timeline shutdown() methods under load
-    tenant_id = env.initial_tenant
-    ps_attached = env.get_tenant_pageserver(tenant_id).id
-    ps_secondary = [p for p in env.pageservers if p.id != ps_attached][0].id
-    for _ in range(n_migrations):
-        # Snapshot of request counter before migration
-        init_getpage_reqs = (
-            env.get_pageserver(ps_secondary)
-            .http_client()
-            .get_metric_value(
-                "pageserver_smgr_query_seconds_global_count", {"smgr_query_type": "get_page_at_lsn"}
-            )
-        )
-        if init_getpage_reqs is None:
-            init_getpage_reqs = 0
-
-        env.storage_controller.tenant_shard_migrate(TenantShardId(tenant_id, 0, 0), ps_secondary)
-        ps_attached, ps_secondary = ps_secondary, ps_attached
-
-        # Make sure we've seen some getpage requests: avoid migrating when a client isn't really doing anything
-        wait_until(
-            30,
-            1,
-            lambda ps_attached=ps_attached,  # type: ignore
-            init_getpage_reqs=init_getpage_reqs: assert_getpage_reqs(
-                ps_attached, init_getpage_reqs
-            ),
-        )
-
-        # Evict some layers, so that we're exercising getpage requests that involve a layer download
-        env.get_pageserver(ps_attached).evict_random_layers(rng, tenant_id, timeline_id, 0.1)
-
-        # Since this test exists to detach shutdown issues, be strict on any warnings logged about delays in shutdown.
-        # Do this each loop iteration so that on failures we don't bury the issue behind many more iterations of logs.
-        for ps in env.pageservers:
-            assert not ps.log_contains(
-                "(kept the gate from closing|closing is taking longer than expected)"
-            )
-
-    test_stop.set()
-    pgbench_thread.join()
-    safekeeper_thread.join()

test_runner/regress/test_pageserver_secondary.py

@@ -7,7 +7,7 @@ from typing import Any, Dict, Optional
 import pytest
 from fixtures.common_types import TenantId, TimelineId
 from fixtures.log_helper import log
-from fixtures.neon_fixtures import NeonEnvBuilder, StorageScrubber
+from fixtures.neon_fixtures import NeonEnvBuilder, NeonPageserver, StorageScrubber
 from fixtures.pageserver.common_types import parse_layer_file_name
 from fixtures.pageserver.utils import (
     assert_prefix_empty,
@@ -34,6 +34,30 @@ TENANT_CONF = {
 }
 
 
+def evict_random_layers(
+    rng: random.Random, pageserver: NeonPageserver, tenant_id: TenantId, timeline_id: TimelineId
+):
+    """
+    Evict 50% of the layers on a pageserver
+    """
+    timeline_path = pageserver.timeline_dir(tenant_id, timeline_id)
+    initial_local_layers = sorted(
+        list(filter(lambda path: path.name != "metadata", timeline_path.glob("*")))
+    )
+    client = pageserver.http_client()
+    for layer in initial_local_layers:
+        if "ephemeral" in layer.name or "temp_download" in layer.name:
+            continue
+
+        layer_name = parse_layer_file_name(layer.name)
+
+        if rng.choice([True, False]):
+            log.info(f"Evicting layer {tenant_id}/{timeline_id} {layer_name.to_str()}")
+            client.evict_layer(
+                tenant_id=tenant_id, timeline_id=timeline_id, layer_name=layer_name.to_str()
+            )
+
+
 @pytest.mark.parametrize("seed", [1, 2, 3])
 def test_location_conf_churn(neon_env_builder: NeonEnvBuilder, seed: int):
     """
@@ -112,7 +136,7 @@ def test_location_conf_churn(neon_env_builder: NeonEnvBuilder, seed: int):
         if mode == "_Evictions":
             if last_state_ps[0].startswith("Attached"):
                 log.info(f"Action: evictions on pageserver {pageserver.id}")
-                pageserver.evict_random_layers(rng, tenant_id, timeline_id, 0.5)
+                evict_random_layers(rng, pageserver, tenant_id, timeline_id)
             else:
                 log.info(
                     f"Action: skipping evictions on pageserver {pageserver.id}, is not attached"

test_runner/regress/test_storage_controller.py

@@ -315,7 +315,7 @@ def test_storage_controller_onboarding(neon_env_builder: NeonEnvBuilder, warm_up
     # Create a tenant directly via pageserver HTTP API, skipping the storage controller
     tenant_id = TenantId.generate()
     generation = 123
-    origin_ps.tenant_create(tenant_id, generation=generation)
+    origin_ps.http_client().tenant_create(tenant_id, generation=generation)
 
     # As if doing a live migration, first configure origin into stale mode
     r = origin_ps.http_client().tenant_location_conf(

vendor/revisions.json

@@ -1,5 +1,5 @@
 {
-  "v16": ["16.3", "e06bebc75306b583e758b52c95946d41109239b2"],
-  "v15": ["15.7", "f54d7373eb0de5a54bce2becdb1c801026c7edff"],
-  "v14": ["14.12", "223dd925959f8124711dd3d867dc8ba6629d52c0"]
+  "v16": ["16.3", "d55e0aca104af0b611cf5565f1033b2acd2dcc1c"],
+  "v15": ["15.7", "2ff5ecc67c64e5fe44b7dde598e64e4538e0c373"],
+  "v14": ["14.12", "7845c122d51d3ebb547a984a640ac0310a2fadce"]
 }