DNM: rip out throttling code (the recorder's borrowing of ctx obviously breaks &mut RequestContext passing)

## Problem
When creating a new shard the storage controller schedules via
Scheduler::schedule_shard. This does not take into account the number of
attached shards. What it does take into account is the node affinity:
when a shard is scheduled, all its nodes (primaries and secondaries) get
their affinity incremented.

For two node clusters and shards with one secondary we have a
pathological case where all primaries are scheduled on the same node.
Now that we track the count of attached shards per node, this is trivial
to fix. Still, the "proper" fix is to use the pageserver's utilization
score.

Closes https://github.com/neondatabase/neon/issues/8041

## Summary of changes
Use attached shard count when deciding which node to schedule a fresh
shard on.

.github/actions/run-python-test-set/action.yml

@@ -183,8 +183,7 @@ runs:
 
         # Run the tests.
         #
-        # The junit.xml file allows CI tools to display more fine-grained test information
-        # in its "Tests" tab in the results page.
+        # --alluredir saves test results in Allure format (in a specified directory)
         # --verbose prints name of each test (helpful when there are
         # multiple tests in one file)
         # -rA prints summary in the end
@@ -193,7 +192,6 @@ runs:
         #
         mkdir -p $TEST_OUTPUT/allure/results
         "${cov_prefix[@]}" ./scripts/pytest \
-          --junitxml=$TEST_OUTPUT/junit.xml \
           --alluredir=$TEST_OUTPUT/allure/results \
           --tb=short \
           --verbose \

.github/workflows/actionlint.yml

@@ -36,15 +36,16 @@ jobs:
           fail_on_error: true
           filter_mode: nofilter
           level: error
-      - run: |
+
+      - name: Disallow 'ubuntu-latest' runners
+        run: |
           PAT='^\s*runs-on:.*-latest'
-          if grep -ERq $PAT .github/workflows
-          then
+          if grep -ERq $PAT .github/workflows; then
             grep -ERl $PAT .github/workflows |\
             while read -r f
             do
               l=$(grep -nE $PAT .github/workflows/release.yml | awk -F: '{print $1}' | head -1)
-              echo "::error file=$f,line=$l::Please, do not use ubuntu-latest images to run on, use LTS instead."
+              echo "::error file=$f,line=$l::Please use 'ubuntu-22.04' instead of 'ubuntu-latest'"
             done
             exit 1
           fi

.github/workflows/build_and_test.yml

@@ -1023,6 +1023,18 @@ jobs:
         with:
           fetch-depth: 0
 
+      # Use custom DOCKER_CONFIG directory to avoid conflicts with default settings
+      # The default value is ~/.docker
+      - name: Set custom docker config directory
+        run: |
+          mkdir -p .docker-custom
+          echo DOCKER_CONFIG=$(pwd)/.docker-custom >> $GITHUB_ENV
+
+      - uses: docker/login-action@v3
+        with:
+          username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
+          password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+
       # `neondatabase/neon` contains multiple binaries, all of them use the same input for the version into the same version formatting library.
       # Pick pageserver as currently the only binary with extra "version" features printed in the string to verify.
       # Regular pageserver version string looks like
@@ -1057,6 +1069,11 @@ jobs:
           docker compose -f ./docker-compose/docker-compose.yml logs || 0
           docker compose -f ./docker-compose/docker-compose.yml down
 
+      - name: Remove custom docker config directory
+        if: always()
+        run: |
+          rm -rf .docker-custom
+
   promote-images:
     needs: [ check-permissions, tag, test-images, vm-compute-node-image ]
     runs-on: ubuntu-22.04

.github/workflows/release.yml

@@ -52,13 +52,15 @@ jobs:
       env:
         GH_TOKEN: ${{ secrets.CI_ACCESS_TOKEN }}
       run: |
+        TITLE="Storage & Compute release ${RELEASE_DATE}"
+
         cat << EOF > body.md
-          ## Storage & Compute release ${RELEASE_DATE}
+          ## ${TITLE}
 
           **Please merge this Pull Request using 'Create a merge commit' button**
         EOF
 
-        gh pr create --title "Release ${RELEASE_DATE}" \
+        gh pr create --title "${TITLE}" \
                      --body-file "body.md" \
                      --head "${RELEASE_BRANCH}" \
                      --base "release"
@@ -91,13 +93,15 @@ jobs:
       env:
         GH_TOKEN: ${{ secrets.CI_ACCESS_TOKEN }}
       run: |
+        TITLE="Proxy release ${RELEASE_DATE}"
+
         cat << EOF > body.md
-          ## Proxy release ${RELEASE_DATE}
+          ## ${TITLE}
 
           **Please merge this Pull Request using 'Create a merge commit' button**
         EOF
 
-        gh pr create --title "Proxy release ${RELEASE_DATE}" \
+        gh pr create --title "${TITLE}" \
                      --body-file "body.md" \
                      --head "${RELEASE_BRANCH}" \
                      --base "release-proxy"

docs/pageserver-pagecache.md

@@ -5,4 +5,3 @@ TODO:
 - shared across tenants
 - store pages from layer files
 - store pages from "in-memory layer"
-- store materialized pages

docs/settings.md

@@ -134,7 +134,7 @@ depends on that, so if you change it, bad things will happen.
 
 #### page_cache_size
 
-Size of the page cache, to hold materialized page versions. Unit is
+Size of the page cache. Unit is
 number of 8 kB blocks. The default is 8192, which means 64 MB.
 
 #### max_file_descriptors

libs/utils/src/backoff.rs

@@ -37,6 +37,20 @@ pub fn exponential_backoff_duration_seconds(n: u32, base_increment: f64, max_sec
     }
 }
 
+pub trait Op<T, E> {
+    async fn call(&mut self) -> Result<T, E>;
+}
+
+impl<T, E, F, Fut> Op<T, E> for F
+where
+    F: FnMut() -> Fut,
+    Fut: Future<Output = Result<T, E>>,
+{
+    async fn call(&mut self) -> Result<T, E> {
+        (&mut *self)().await
+    }
+}
+
 /// Retries passed operation until one of the following conditions are met:
 /// - encountered error is considered as permanent (non-retryable)
 /// - retries have been exhausted
@@ -51,8 +65,8 @@ pub fn exponential_backoff_duration_seconds(n: u32, base_increment: f64, max_sec
 /// for any other error type. Final failed attempt is logged with `{:?}`.
 ///
 /// Returns `None` if cancellation was noticed during backoff or the terminal result.
-pub async fn retry<T, O, F, E>(
-    mut op: O,
+pub async fn retry<T, E>(
+    mut op: impl Op<T, E>,
     is_permanent: impl Fn(&E) -> bool,
     warn_threshold: u32,
     max_retries: u32,
@@ -63,8 +77,6 @@ where
     // Not std::error::Error because anyhow::Error doesnt implement it.
     // For context see https://github.com/dtolnay/anyhow/issues/63
     E: Display + Debug + 'static,
-    O: FnMut() -> F,
-    F: Future<Output = Result<T, E>>,
 {
     let mut attempts = 0;
     loop {
@@ -72,7 +84,7 @@ where
             return None;
         }
 
-        let result = op().await;
+        let result = op.call().await;
         match &result {
             Ok(_) => {
                 if attempts > 0 {

pageserver/ctl/src/layer_map_analyzer.rs

@@ -99,7 +99,11 @@ pub(crate) fn parse_filename(name: &str) -> Option<LayerFile> {
 }
 
 // Finds the max_holes largest holes, ignoring any that are smaller than MIN_HOLE_LENGTH"
-async fn get_holes(path: &Utf8Path, max_holes: usize, ctx: &RequestContext) -> Result<Vec<Hole>> {
+async fn get_holes(
+    path: &Utf8Path,
+    max_holes: usize,
+    ctx: &mut RequestContext,
+) -> Result<Vec<Hole>> {
     let file = VirtualFile::open(path, ctx).await?;
     let file_id = page_cache::next_file_id();
     let block_reader = FileBlockReader::new(&file, file_id);

pageserver/ctl/src/layers.rs

@@ -57,7 +57,7 @@ pub(crate) enum LayerCmd {
     },
 }
 
-async fn read_delta_file(path: impl AsRef<Path>, ctx: &RequestContext) -> Result<()> {
+async fn read_delta_file(path: impl AsRef<Path>, ctx: &mut RequestContext) -> Result<()> {
     let path = Utf8Path::from_path(path.as_ref()).expect("non-Unicode path");
     virtual_file::init(10, virtual_file::api::IoEngineKind::StdFs);
     page_cache::init(100);

pageserver/src/basebackup.rs

@@ -60,7 +60,7 @@ pub async fn send_basebackup_tarball<'a, W>(
     req_lsn: Option<Lsn>,
     prev_lsn: Option<Lsn>,
     full_backup: bool,
-    ctx: &'a RequestContext,
+    ctx: &'a mut RequestContext,
 ) -> Result<(), BasebackupError>
 where
     W: AsyncWrite + Send + Sync + Unpin,
@@ -141,7 +141,7 @@ where
     lsn: Lsn,
     prev_record_lsn: Lsn,
     full_backup: bool,
-    ctx: &'a RequestContext,
+    ctx: &'a mut RequestContext,
 }
 
 /// A sink that accepts SLRU blocks ordered by key and forwards

pageserver/src/consumption_metrics.rs

@@ -51,7 +51,7 @@ pub async fn collect_metrics(
     node_id: NodeId,
     local_disk_storage: Utf8PathBuf,
     cancel: CancellationToken,
-    ctx: RequestContext,
+    mut ctx: RequestContext,
 ) -> anyhow::Result<()> {
     if _cached_metric_collection_interval != Duration::ZERO {
         tracing::warn!(
@@ -60,7 +60,7 @@ pub async fn collect_metrics(
     }
 
     // spin up background worker that caclulates tenant sizes
-    let worker_ctx =
+    let mut worker_ctx =
         ctx.detached_child(TaskKind::CalculateSyntheticSize, DownloadBehavior::Download);
     task_mgr::spawn(
         BACKGROUND_RUNTIME.handle(),
@@ -76,7 +76,7 @@ pub async fn collect_metrics(
                     tenant_manager,
                     synthetic_size_calculation_interval,
                     &cancel,
-                    &worker_ctx,
+                    &mut worker_ctx,
                 )
                 .instrument(info_span!("synthetic_size_worker"))
                 .await?;
@@ -122,7 +122,8 @@ pub async fn collect_metrics(
         let started_at = Instant::now();
 
         // these are point in time, with variable "now"
-        let metrics = metrics::collect_all_metrics(&tenant_manager, &cached_metrics, &ctx).await;
+        let metrics =
+            metrics::collect_all_metrics(&tenant_manager, &cached_metrics, &mut ctx).await;
 
         let metrics = Arc::new(metrics);
 
@@ -280,7 +281,7 @@ async fn calculate_synthetic_size_worker(
     tenant_manager: Arc<TenantManager>,
     synthetic_size_calculation_interval: Duration,
     cancel: &CancellationToken,
-    ctx: &RequestContext,
+    ctx: &mut RequestContext,
 ) -> anyhow::Result<()> {
     info!("starting calculate_synthetic_size_worker");
     scopeguard::defer! {
@@ -340,7 +341,7 @@ async fn calculate_synthetic_size_worker(
     }
 }
 
-async fn calculate_and_log(tenant: &Tenant, cancel: &CancellationToken, ctx: &RequestContext) {
+async fn calculate_and_log(tenant: &Tenant, cancel: &CancellationToken, ctx: &mut RequestContext) {
     const CAUSE: LogicalSizeCalculationCause =
         LogicalSizeCalculationCause::ConsumptionMetricsSyntheticSize;
 

pageserver/src/consumption_metrics/metrics.rs

@@ -184,7 +184,7 @@ impl MetricsKey {
 pub(super) async fn collect_all_metrics(
     tenant_manager: &Arc<TenantManager>,
     cached_metrics: &Cache,
-    ctx: &RequestContext,
+    ctx: &mut RequestContext,
 ) -> Vec<RawMetric> {
     use pageserver_api::models::TenantState;
 
@@ -220,7 +220,7 @@ pub(super) async fn collect_all_metrics(
     res
 }
 
-async fn collect<S>(tenants: S, cache: &Cache, ctx: &RequestContext) -> Vec<RawMetric>
+async fn collect<S>(tenants: S, cache: &Cache, ctx: &mut RequestContext) -> Vec<RawMetric>
 where
     S: futures::stream::Stream<Item = (TenantId, Arc<crate::tenant::Tenant>)>,
 {
@@ -342,7 +342,7 @@ impl TimelineSnapshot {
     /// [`Timeline::get_current_logical_size`]: crate::tenant::Timeline::get_current_logical_size
     fn collect(
         t: &Arc<crate::tenant::Timeline>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<Option<Self>> {
         if !t.is_active() {
             // no collection for broken or stopping needed, we will still keep the cached values

pageserver/src/context.rs

@@ -97,9 +97,10 @@ pub struct RequestContext {
     download_behavior: DownloadBehavior,
     access_stats_behavior: AccessStatsBehavior,
     page_content_kind: PageContentKind,
-    pub micros_spent_throttled: optional_counter::MicroSecondsCounterU32,
 }
 
+pub(crate) struct MicrosSpentThrottled(optional_counter::MicroSecondsCounterU32);
+
 /// The kind of access to the page cache.
 #[derive(Clone, Copy, PartialEq, Eq, Debug, enum_map::Enum, strum_macros::IntoStaticStr)]
 pub enum PageContentKind {
@@ -158,7 +159,7 @@ impl RequestContextBuilder {
         }
     }
 
-    pub fn extend(original: &RequestContext) -> Self {
+    pub fn extend(original: &mut RequestContext) -> Self {
         Self {
             // This is like a Copy, but avoid implementing Copy because ordinary users of
             // RequestContext should always move or ref it.

pageserver/src/http/routes.rs

@@ -352,7 +352,7 @@ async fn build_timeline_info(
     timeline: &Arc<Timeline>,
     include_non_incremental_logical_size: bool,
     force_await_initial_logical_size: bool,
-    ctx: &RequestContext,
+    ctx: &mut RequestContext,
 ) -> anyhow::Result<TimelineInfo> {
     crate::tenant::debug_assert_current_span_has_tenant_and_timeline_id();
 
@@ -381,7 +381,7 @@ async fn build_timeline_info(
 
 async fn build_timeline_info_common(
     timeline: &Arc<Timeline>,
-    ctx: &RequestContext,
+    ctx: &mut RequestContext,
     logical_size_task_priority: tenant::timeline::GetLogicalSizePriority,
 ) -> anyhow::Result<TimelineInfo> {
     crate::tenant::debug_assert_current_span_has_tenant_and_timeline_id();
@@ -502,7 +502,7 @@ async fn timeline_create_handler(
 
     let new_timeline_id = request_data.new_timeline_id;
 
-    let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Error);
+    let mut ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Error);
 
     let state = get_state(&request);
 
@@ -527,7 +527,7 @@ async fn timeline_create_handler(
                 request_data.pg_version.unwrap_or(crate::DEFAULT_PG_VERSION),
                 request_data.existing_initdb_timeline_id,
                 state.broker_client.clone(),
-                &ctx,
+                &mut ctx,
             )
             .await
         {
@@ -535,7 +535,7 @@ async fn timeline_create_handler(
                 // Created. Construct a TimelineInfo for it.
                 let timeline_info = build_timeline_info_common(
                     &new_timeline,
-                    &ctx,
+                    &mut ctx,
                     tenant::timeline::GetLogicalSizePriority::User,
                 )
                 .await
@@ -593,7 +593,7 @@ async fn timeline_list_handler(
     check_permission(&request, Some(tenant_shard_id.tenant_id))?;
 
     let state = get_state(&request);
-    let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
+    let mut ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
 
     let response_data = async {
         let tenant = state
@@ -610,7 +610,7 @@ async fn timeline_list_handler(
                 &timeline,
                 include_non_incremental_logical_size.unwrap_or(false),
                 force_await_initial_logical_size.unwrap_or(false),
-                &ctx,
+                &mut ctx,
             )
             .instrument(info_span!("build_timeline_info", timeline_id = %timeline.timeline_id))
             .await
@@ -680,7 +680,7 @@ async fn timeline_detail_handler(
     check_permission(&request, Some(tenant_shard_id.tenant_id))?;
 
     // Logical size calculation needs downloading.
-    let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
+    let mut ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
     let state = get_state(&request);
 
     let timeline_info = async {
@@ -696,7 +696,7 @@ async fn timeline_detail_handler(
             &timeline,
             include_non_incremental_logical_size.unwrap_or(false),
             force_await_initial_logical_size.unwrap_or(false),
-            &ctx,
+            &mut ctx,
         )
         .await
         .context("get local timeline info")
@@ -735,13 +735,13 @@ async fn get_lsn_by_timestamp_handler(
         .map_err(ApiError::BadRequest)?;
     let timestamp_pg = postgres_ffi::to_pg_timestamp(timestamp);
 
-    let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
+    let mut ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
 
     let timeline =
         active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
             .await?;
     let result = timeline
-        .find_lsn_for_timestamp(timestamp_pg, &cancel, &ctx)
+        .find_lsn_for_timestamp(timestamp_pg, &cancel, &mut ctx)
         .await?;
     #[derive(serde::Serialize, Debug)]
     struct Result {
@@ -786,11 +786,11 @@ async fn get_timestamp_of_lsn_handler(
         .with_context(|| format!("Invalid LSN: {lsn_str:?}"))
         .map_err(ApiError::BadRequest)?;
 
-    let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
+    let  mut ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
     let timeline =
         active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
             .await?;
-    let result = timeline.get_timestamp_for_lsn(lsn, &ctx).await?;
+    let result = timeline.get_timestamp_for_lsn(lsn, &mut ctx).await?;
 
     match result {
         Some(time) => {
@@ -816,7 +816,7 @@ async fn tenant_attach_handler(
         None => TenantConfOpt::default(),
     };
 
-    let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
+    let mut ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
 
     info!("Handling tenant attach {tenant_id}");
 
@@ -830,7 +830,7 @@ async fn tenant_attach_handler(
 
     let tenant = state
         .tenant_manager
-        .upsert_location(tenant_shard_id, location_conf, None, SpawnMode::Eager, &ctx)
+        .upsert_location(tenant_shard_id, location_conf, None, SpawnMode::Eager, &mut ctx)
         .await?;
 
     let Some(tenant) = tenant else {
@@ -921,11 +921,11 @@ async fn tenant_reset_handler(
 
     let drop_cache: Option<bool> = parse_query_param(&request, "drop_cache")?;
 
-    let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
+    let mut ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
     let state = get_state(&request);
     state
         .tenant_manager
-        .reset_tenant(tenant_shard_id, drop_cache.unwrap_or(false), &ctx)
+        .reset_tenant(tenant_shard_id, drop_cache.unwrap_or(false), &mut ctx)
         .await
         .map_err(ApiError::InternalServerError)?;
 
@@ -939,7 +939,7 @@ async fn tenant_load_handler(
     let tenant_id: TenantId = parse_request_param(&request, "tenant_id")?;
     check_permission(&request, Some(tenant_id))?;
 
-    let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
+    let mut ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
 
     let maybe_body: Option<TenantLoadRequest> = json_request_or_empty_body(&mut request).await?;
 
@@ -956,7 +956,7 @@ async fn tenant_load_handler(
         state.broker_client.clone(),
         state.remote_storage.clone(),
         state.deletion_queue_client.clone(),
-        &ctx,
+        &mut ctx,
     )
     .instrument(info_span!("load", %tenant_id))
     .await?;
@@ -1120,7 +1120,7 @@ async fn tenant_size_handler(
         )));
     }
 
-    let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
+    let mut ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
     let tenant = state
         .tenant_manager
         .get_attached_tenant_shard(tenant_shard_id)?;
@@ -1132,7 +1132,7 @@ async fn tenant_size_handler(
             retention_period,
             LogicalSizeCalculationCause::TenantSizeHandler,
             &cancel,
-            &ctx,
+            &mut ctx,
         )
         .await
         .map_err(|e| match e {
@@ -1193,7 +1193,7 @@ async fn tenant_shard_split_handler(
 
     let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
     let state = get_state(&request);
-    let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
+    let mut ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
 
     let tenant = state
         .tenant_manager
@@ -1206,7 +1206,7 @@ async fn tenant_shard_split_handler(
             tenant,
             ShardCount::new(req.new_shard_count),
             req.new_stripe_size,
-            &ctx,
+            &mut ctx,
         )
         .await
         .map_err(ApiError::InternalServerError)?;
@@ -1386,7 +1386,7 @@ async fn tenant_create_handler(
 
     let generation = get_request_generation(state, request_data.generation)?;
 
-    let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
+    let  mut ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
 
     let location_conf =
         LocationConf::attached_single(tenant_conf, generation, &request_data.shard_parameters);
@@ -1398,7 +1398,7 @@ async fn tenant_create_handler(
             location_conf,
             None,
             SpawnMode::Create,
-            &ctx,
+            &mut ctx,
         )
         .await?;
 
@@ -1498,7 +1498,7 @@ async fn put_tenant_location_config_handler(
     let lazy = parse_query_param(&request, "lazy")?.unwrap_or(false);
     check_permission(&request, Some(tenant_shard_id.tenant_id))?;
 
-    let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
+    let mut ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
     let state = get_state(&request);
     let conf = state.conf;
 
@@ -1537,7 +1537,7 @@ async fn put_tenant_location_config_handler(
 
     let tenant = state
         .tenant_manager
-        .upsert_location(tenant_shard_id, location_conf, flush, spawn_mode, &ctx)
+        .upsert_location(tenant_shard_id, location_conf, flush, spawn_mode, &mut ctx)
         .await?;
     let stripe_size = tenant.as_ref().map(|t| t.get_shard_stripe_size());
     let attached = tenant.is_some();
@@ -1722,7 +1722,7 @@ async fn lsn_lease_handler(
     let lsn: Lsn = parse_query_param(&request, "lsn")?
         .ok_or_else(|| ApiError::BadRequest(anyhow!("missing 'lsn' query parameter")))?;
 
-    let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
+    let mut ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
 
     let state = get_state(&request);
 
@@ -1730,7 +1730,7 @@ async fn lsn_lease_handler(
         active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
             .await?;
     let result = timeline
-        .make_lsn_lease(lsn, timeline.get_lsn_lease_length(), &ctx)
+        .make_lsn_lease(lsn, timeline.get_lsn_lease_length(), &mut ctx)
         .map_err(|e| ApiError::InternalServerError(e.context("lsn lease http handler")))?;
 
     json_response(StatusCode::OK, result)
@@ -1747,8 +1747,8 @@ async fn timeline_gc_handler(
 
     let gc_req: TimelineGcRequest = json_request(&mut request).await?;
 
-    let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
-    let gc_result = mgr::immediate_gc(tenant_shard_id, timeline_id, gc_req, cancel, &ctx).await?;
+    let mut ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
+    let gc_result = mgr::immediate_gc(tenant_shard_id, timeline_id, gc_req, cancel, &mut ctx).await?;
 
     json_response(StatusCode::OK, gc_result)
 }
@@ -1775,10 +1775,10 @@ async fn timeline_compact_handler(
         parse_query_param::<_, bool>(&request, "wait_until_uploaded")?.unwrap_or(false);
 
     async {
-        let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
+        let mut ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
         let timeline = active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id).await?;
         timeline
-            .compact(&cancel, flags, &ctx)
+            .compact(&cancel, flags, &mut ctx)
             .await
             .map_err(|e| ApiError::InternalServerError(e.into()))?;
         if wait_until_uploaded {
@@ -1812,7 +1812,7 @@ async fn timeline_checkpoint_handler(
         parse_query_param::<_, bool>(&request, "wait_until_uploaded")?.unwrap_or(false);
 
     async {
-        let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
+        let mut ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
         let timeline = active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id).await?;
         timeline
             .freeze_and_flush()
@@ -1825,7 +1825,7 @@ async fn timeline_checkpoint_handler(
                 }
             })?;
         timeline
-            .compact(&cancel, flags, &ctx)
+            .compact(&cancel, flags, &mut ctx)
             .await
             .map_err(|e|
                 match e {
@@ -1918,8 +1918,8 @@ async fn timeline_detach_ancestor_handler(
 
         tenant.wait_to_become_active(ACTIVE_TENANT_TIMEOUT).await?;
 
-        let ctx = RequestContext::new(TaskKind::DetachAncestor, DownloadBehavior::Download);
-        let ctx = &ctx;
+        let mut ctx = RequestContext::new(TaskKind::DetachAncestor, DownloadBehavior::Download);
+        let ctx = &mut ctx;
 
         let timeline = tenant.get_timeline(timeline_id, true)?;
 
@@ -2003,10 +2003,10 @@ async fn getpage_at_lsn_handler(
         .ok_or_else(|| ApiError::BadRequest(anyhow!("missing 'lsn' query parameter")))?;
 
     async {
-        let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
+        let mut ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
         let timeline = active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id).await?;
 
-        let page = timeline.get(key.0, lsn, &ctx).await?;
+        let page = timeline.get(key.0, lsn, &mut ctx).await?;
 
         Result::<_, ApiError>::Ok(
             Response::builder()
@@ -2032,11 +2032,11 @@ async fn timeline_collect_keyspace(
     let at_lsn: Option<Lsn> = parse_query_param(&request, "at_lsn")?;
 
     async {
-        let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
+        let mut ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
         let timeline = active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id).await?;
         let at_lsn = at_lsn.unwrap_or_else(|| timeline.get_last_record_lsn());
         let (dense_ks, sparse_ks) = timeline
-            .collect_keyspace(at_lsn, &ctx)
+            .collect_keyspace(at_lsn, &mut ctx)
             .await
             .map_err(|e| ApiError::InternalServerError(e.into()))?;
 
@@ -2425,8 +2425,8 @@ async fn list_aux_files(
         active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
             .await?;
 
-    let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
-    let files = timeline.list_aux_files(body.lsn, &ctx).await?;
+    let mut ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
+    let files = timeline.list_aux_files(body.lsn, &mut ctx).await?;
     json_response(StatusCode::OK, files)
 }
 
@@ -2467,15 +2467,15 @@ async fn ingest_aux_files(
     let mut modification = timeline.begin_modification(
         Lsn(timeline.get_last_record_lsn().0 + 8), /* advance LSN by 8 */
     );
-    let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
+    let mut ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
     for (fname, content) in body.aux_files {
         modification
-            .put_file(&fname, content.as_bytes(), &ctx)
+            .put_file(&fname, content.as_bytes(), &mut ctx)
             .await
             .map_err(ApiError::InternalServerError)?;
     }
     modification
-        .commit(&ctx)
+        .commit(&mut ctx)
         .await
         .map_err(ApiError::InternalServerError)?;
 

pageserver/src/import_datadir.rs

@@ -53,7 +53,7 @@ pub async fn import_timeline_from_postgres_datadir(
     tline: &Timeline,
     pgdata_path: &Utf8Path,
     pgdata_lsn: Lsn,
-    ctx: &RequestContext,
+    ctx: &mut RequestContext,
 ) -> Result<()> {
     let mut pg_control: Option<ControlFileData> = None;
 
@@ -121,7 +121,7 @@ async fn import_rel(
     dboid: Oid,
     reader: &mut (impl AsyncRead + Unpin),
     len: usize,
-    ctx: &RequestContext,
+    ctx: &mut RequestContext,
 ) -> anyhow::Result<()> {
     // Does it look like a relation file?
     trace!("importing rel file {}", path.display());
@@ -210,7 +210,7 @@ async fn import_slru(
     path: &Path,
     reader: &mut (impl AsyncRead + Unpin),
     len: usize,
-    ctx: &RequestContext,
+    ctx: &mut RequestContext,
 ) -> anyhow::Result<()> {
     info!("importing slru file {path:?}");
 
@@ -268,7 +268,7 @@ async fn import_wal(
     tline: &Timeline,
     startpoint: Lsn,
     endpoint: Lsn,
-    ctx: &RequestContext,
+    ctx: &mut RequestContext,
 ) -> anyhow::Result<()> {
     let mut waldecoder = WalStreamDecoder::new(startpoint, tline.pg_version);
 
@@ -346,7 +346,7 @@ pub async fn import_basebackup_from_tar(
     tline: &Timeline,
     reader: &mut (impl AsyncRead + Send + Sync + Unpin),
     base_lsn: Lsn,
-    ctx: &RequestContext,
+    ctx: &mut RequestContext,
 ) -> Result<()> {
     info!("importing base at {base_lsn}");
     let mut modification = tline.begin_modification(base_lsn);
@@ -397,7 +397,7 @@ pub async fn import_wal_from_tar(
     reader: &mut (impl AsyncRead + Send + Sync + Unpin),
     start_lsn: Lsn,
     end_lsn: Lsn,
-    ctx: &RequestContext,
+    ctx: &mut RequestContext,
 ) -> Result<()> {
     // Set up walingest mutable state
     let mut waldecoder = WalStreamDecoder::new(start_lsn, tline.pg_version);
@@ -489,7 +489,7 @@ async fn import_file(
     file_path: &Path,
     reader: &mut (impl AsyncRead + Send + Sync + Unpin),
     len: usize,
-    ctx: &RequestContext,
+    ctx: &mut RequestContext,
 ) -> Result<Option<ControlFileData>> {
     let file_name = match file_path.file_name() {
         Some(name) => name.to_string_lossy(),

pageserver/src/metrics.rs

@@ -145,14 +145,6 @@ impl ReconstructTimeMetrics {
     }
 }
 
-pub(crate) static MATERIALIZED_PAGE_CACHE_HIT_DIRECT: Lazy<IntCounter> = Lazy::new(|| {
-    register_int_counter!(
-        "pageserver_materialized_cache_hits_direct_total",
-        "Number of cache hits from materialized page cache without redo",
-    )
-    .expect("failed to define a metric")
-});
-
 pub(crate) struct ReconstructDataTimeMetrics {
     singular: Histogram,
     vectored: Histogram,
@@ -182,14 +174,6 @@ pub(crate) static GET_RECONSTRUCT_DATA_TIME: Lazy<ReconstructDataTimeMetrics> =
     }
 });
 
-pub(crate) static MATERIALIZED_PAGE_CACHE_HIT: Lazy<IntCounter> = Lazy::new(|| {
-    register_int_counter!(
-        "pageserver_materialized_cache_hits_total",
-        "Number of cache hits from materialized page cache",
-    )
-    .expect("failed to define a metric")
-});
-
 pub(crate) struct GetVectoredLatency {
     map: EnumMap<TaskKind, Option<Histogram>>,
 }
@@ -298,12 +282,8 @@ pub(crate) static SCAN_LATENCY: Lazy<ScanLatency> = Lazy::new(|| {
 });
 
 pub(crate) struct PageCacheMetricsForTaskKind {
-    pub read_accesses_materialized_page: IntCounter,
     pub read_accesses_immutable: IntCounter,
-
     pub read_hits_immutable: IntCounter,
-    pub read_hits_materialized_page_exact: IntCounter,
-    pub read_hits_materialized_page_older_lsn: IntCounter,
 }
 
 pub(crate) struct PageCacheMetrics {
@@ -336,16 +316,6 @@ pub(crate) static PAGE_CACHE: Lazy<PageCacheMetrics> = Lazy::new(|| PageCacheMet
             let content_kind = <PageContentKind as enum_map::Enum>::from_usize(content_kind);
             let content_kind: &'static str = content_kind.into();
             PageCacheMetricsForTaskKind {
-                read_accesses_materialized_page: {
-                    PAGE_CACHE_READ_ACCESSES
-                        .get_metric_with_label_values(&[
-                            task_kind,
-                            "materialized_page",
-                            content_kind,
-                        ])
-                        .unwrap()
-                },
-
                 read_accesses_immutable: {
                     PAGE_CACHE_READ_ACCESSES
                         .get_metric_with_label_values(&[task_kind, "immutable", content_kind])
@@ -357,35 +327,13 @@ pub(crate) static PAGE_CACHE: Lazy<PageCacheMetrics> = Lazy::new(|| PageCacheMet
                         .get_metric_with_label_values(&[task_kind, "immutable", content_kind, "-"])
                         .unwrap()
                 },
-
-                read_hits_materialized_page_exact: {
-                    PAGE_CACHE_READ_HITS
-                        .get_metric_with_label_values(&[
-                            task_kind,
-                            "materialized_page",
-                            content_kind,
-                            "exact",
-                        ])
-                        .unwrap()
-                },
-
-                read_hits_materialized_page_older_lsn: {
-                    PAGE_CACHE_READ_HITS
-                        .get_metric_with_label_values(&[
-                            task_kind,
-                            "materialized_page",
-                            content_kind,
-                            "older_lsn",
-                        ])
-                        .unwrap()
-                },
             }
         }))
     })),
 });
 
 impl PageCacheMetrics {
-    pub(crate) fn for_ctx(&self, ctx: &RequestContext) -> &PageCacheMetricsForTaskKind {
+    pub(crate) fn for_ctx(&self, ctx: &mut RequestContext) -> &PageCacheMetricsForTaskKind {
         &self.map[ctx.task_kind()][ctx.page_content_kind()]
     }
 }
@@ -394,7 +342,6 @@ pub(crate) struct PageCacheSizeMetrics {
     pub max_bytes: UIntGauge,
 
     pub current_bytes_immutable: UIntGauge,
-    pub current_bytes_materialized_page: UIntGauge,
 }
 
 static PAGE_CACHE_SIZE_CURRENT_BYTES: Lazy<UIntGaugeVec> = Lazy::new(|| {
@@ -420,11 +367,6 @@ pub(crate) static PAGE_CACHE_SIZE: Lazy<PageCacheSizeMetrics> =
                 .get_metric_with_label_values(&["immutable"])
                 .unwrap()
         },
-        current_bytes_materialized_page: {
-            PAGE_CACHE_SIZE_CURRENT_BYTES
-                .get_metric_with_label_values(&["materialized_page"])
-                .unwrap()
-        },
     });
 
 pub(crate) mod page_cache_eviction_metrics {
@@ -1143,7 +1085,6 @@ impl GlobalAndPerTimelineHistogram {
 
 struct GlobalAndPerTimelineHistogramTimer<'a, 'c> {
     h: &'a GlobalAndPerTimelineHistogram,
-    ctx: &'c RequestContext,
     start: std::time::Instant,
     op: SmgrQueryType,
 }
@@ -1151,32 +1092,11 @@ struct GlobalAndPerTimelineHistogramTimer<'a, 'c> {
 impl<'a, 'c> Drop for GlobalAndPerTimelineHistogramTimer<'a, 'c> {
     fn drop(&mut self) {
         let elapsed = self.start.elapsed();
-        let ex_throttled = self
-            .ctx
-            .micros_spent_throttled
-            .close_and_checked_sub_from(elapsed);
-        let ex_throttled = match ex_throttled {
-            Ok(res) => res,
-            Err(error) => {
-                use utils::rate_limit::RateLimit;
-                static LOGGED: Lazy<Mutex<enum_map::EnumMap<SmgrQueryType, RateLimit>>> =
-                    Lazy::new(|| {
-                        Mutex::new(enum_map::EnumMap::from_array(std::array::from_fn(|_| {
-                            RateLimit::new(Duration::from_secs(10))
-                        })))
-                    });
-                let mut guard = LOGGED.lock().unwrap();
-                let rate_limit = &mut guard[self.op];
-                rate_limit.call(|| {
-                    warn!(op=?self.op, error, "error deducting time spent throttled; this message is logged at a global rate limit");
-                });
-                elapsed
-            }
-        };
-        self.h.observe(ex_throttled.as_secs_f64());
+        self.h.observe(elapsed.as_secs_f64());
     }
 }
 
+
 #[derive(
     Debug,
     Clone,
@@ -1291,33 +1211,11 @@ impl SmgrQueryTimePerTimeline {
         });
         Self { metrics }
     }
-    pub(crate) fn start_timer<'c: 'a, 'a>(
-        &'a self,
-        op: SmgrQueryType,
-        ctx: &'c RequestContext,
-    ) -> impl Drop + '_ {
+    pub(crate) fn start_timer<'a>(&'a self, op: SmgrQueryType) -> impl Drop + 'a {
         let metric = &self.metrics[op as usize];
         let start = Instant::now();
-        match ctx.micros_spent_throttled.open() {
-            Ok(()) => (),
-            Err(error) => {
-                use utils::rate_limit::RateLimit;
-                static LOGGED: Lazy<Mutex<enum_map::EnumMap<SmgrQueryType, RateLimit>>> =
-                    Lazy::new(|| {
-                        Mutex::new(enum_map::EnumMap::from_array(std::array::from_fn(|_| {
-                            RateLimit::new(Duration::from_secs(10))
-                        })))
-                    });
-                let mut guard = LOGGED.lock().unwrap();
-                let rate_limit = &mut guard[op];
-                rate_limit.call(|| {
-                    warn!(?op, error, "error opening micros_spent_throttled; this message is logged at a global rate limit");
-                });
-            }
-        }
         GlobalAndPerTimelineHistogramTimer {
             h: metric,
-            ctx,
             start,
             op,
         }
@@ -1384,7 +1282,7 @@ mod smgr_query_time_tests {
             assert_eq!(pre_per_tenant_timeline, 0);
 
             let ctx = RequestContext::new(TaskKind::UnitTest, DownloadBehavior::Download);
-            let timer = metrics.start_timer(*op, &ctx);
+            let timer = metrics.start_timer(*op);
             drop(timer);
 
             let (post_global, post_per_tenant_timeline) = get_counts();
@@ -1405,17 +1303,23 @@ static COMPUTE_STARTUP_BUCKETS: Lazy<[f64; 28]> = Lazy::new(|| {
     .map(|ms| (ms as f64) / 1000.0)
 });
 
-pub(crate) struct BasebackupQueryTime(HistogramVec);
+pub(crate) struct BasebackupQueryTime {
+    ok: Histogram,
+    error: Histogram,
+}
+
 pub(crate) static BASEBACKUP_QUERY_TIME: Lazy<BasebackupQueryTime> = Lazy::new(|| {
-    BasebackupQueryTime({
-        register_histogram_vec!(
-            "pageserver_basebackup_query_seconds",
-            "Histogram of basebackup queries durations, by result type",
-            &["result"],
-            COMPUTE_STARTUP_BUCKETS.to_vec(),
-        )
-        .expect("failed to define a metric")
-    })
+    let vec = register_histogram_vec!(
+        "pageserver_basebackup_query_seconds",
+        "Histogram of basebackup queries durations, by result type",
+        &["result"],
+        COMPUTE_STARTUP_BUCKETS.to_vec(),
+    )
+    .expect("failed to define a metric");
+    BasebackupQueryTime {
+        ok: vec.get_metric_with_label_values(&["ok"]).unwrap(),
+        error: vec.get_metric_with_label_values(&["error"]).unwrap(),
+    }
 });
 
 pub(crate) struct BasebackupQueryTimeOngoingRecording<'a, 'c> {
@@ -1470,12 +1374,11 @@ impl<'a, 'c> BasebackupQueryTimeOngoingRecording<'a, 'c> {
                 elapsed
             }
         };
-        let label_value = if res.is_ok() { "ok" } else { "error" };
-        let metric = self
-            .parent
-            .0
-            .get_metric_with_label_values(&[label_value])
-            .unwrap();
+        let metric = if res.is_ok() {
+            &self.parent.ok
+        } else {
+            &self.parent.error
+        };
         metric.observe(ex_throttled.as_secs_f64());
     }
 }
@@ -2918,13 +2821,11 @@ pub fn preinitialize_metrics() {
     // FIXME(4813): make it so that we have no top level metrics as this fn will easily fall out of
     // order:
     // - global metrics reside in a Lazy<PageserverMetrics>
-    //   - access via crate::metrics::PS_METRICS.materialized_page_cache_hit.inc()
+    //   - access via crate::metrics::PS_METRICS.some_metric.inc()
     // - could move the statics into TimelineMetrics::new()?
 
     // counters
     [
-        &MATERIALIZED_PAGE_CACHE_HIT,
-        &MATERIALIZED_PAGE_CACHE_HIT_DIRECT,
         &UNEXPECTED_ONDEMAND_DOWNLOADS,
         &WALRECEIVER_STARTED_CONNECTIONS,
         &WALRECEIVER_BROKER_UPDATES,
@@ -2986,4 +2887,5 @@ pub fn preinitialize_metrics() {
     // Custom
     Lazy::force(&RECONSTRUCT_TIME);
     Lazy::force(&tenant_throttling::TIMELINE_GET);
+    Lazy::force(&BASEBACKUP_QUERY_TIME);
 }

pageserver/src/page_cache.rs

@@ -17,7 +17,6 @@
 //!
 //! Two types of pages are supported:
 //!
-//! * **Materialized pages**, filled & used by page reconstruction
 //! * **Immutable File pages**, filled & used by [`crate::tenant::block_io`] and [`crate::tenant::ephemeral_file`].
 //!
 //! Note that [`crate::tenant::ephemeral_file::EphemeralFile`] is generally mutable, but, it's append-only.
@@ -28,9 +27,6 @@
 //! Page cache maps from a cache key to a buffer slot.
 //! The cache key uniquely identifies the piece of data that is being cached.
 //!
-//! The cache key for **materialized pages** is  [`TenantShardId`], [`TimelineId`], [`Key`], and [`Lsn`].
-//! Use [`PageCache::memorize_materialized_page`] and [`PageCache::lookup_materialized_page`] for fill & access.
-//!
 //! The cache key for **immutable file** pages is [`FileId`] and a block number.
 //! Users of page cache that wish to page-cache an arbitrary (immutable!) on-disk file do the following:
 //! * Have a mechanism to deterministically associate the on-disk file with a [`FileId`].
@@ -82,13 +78,10 @@ use std::{
 
 use anyhow::Context;
 use once_cell::sync::OnceCell;
-use pageserver_api::shard::TenantShardId;
-use utils::{id::TimelineId, lsn::Lsn};
 
 use crate::{
     context::RequestContext,
     metrics::{page_cache_eviction_metrics, PageCacheSizeMetrics},
-    repository::Key,
 };
 
 static PAGE_CACHE: OnceCell<PageCache> = OnceCell::new();
@@ -139,33 +132,7 @@ pub fn next_file_id() -> FileId {
 #[derive(Debug, PartialEq, Eq, Clone)]
 #[allow(clippy::enum_variant_names)]
 enum CacheKey {
-    MaterializedPage {
-        hash_key: MaterializedPageHashKey,
-        lsn: Lsn,
-    },
-    ImmutableFilePage {
-        file_id: FileId,
-        blkno: u32,
-    },
-}
-
-#[derive(Debug, PartialEq, Eq, Hash, Clone)]
-struct MaterializedPageHashKey {
-    /// Why is this TenantShardId rather than TenantId?
-    ///
-    /// Usually, the materialized value of a page@lsn is identical on any shard in the same tenant.  However, this
-    /// this not the case for certain internally-generated pages (e.g. relation sizes).  In future, we may make this
-    /// key smaller by omitting the shard, if we ensure that reads to such pages always skip the cache, or are
-    /// special-cased in some other way.
-    tenant_shard_id: TenantShardId,
-    timeline_id: TimelineId,
-    key: Key,
-}
-
-#[derive(Clone)]
-struct Version {
-    lsn: Lsn,
-    slot_idx: usize,
+    ImmutableFilePage { file_id: FileId, blkno: u32 },
 }
 
 struct Slot {
@@ -236,17 +203,6 @@ impl SlotInner {
 }
 
 pub struct PageCache {
-    /// This contains the mapping from the cache key to buffer slot that currently
-    /// contains the page, if any.
-    ///
-    /// TODO: This is protected by a single lock. If that becomes a bottleneck,
-    /// this HashMap can be replaced with a more concurrent version, there are
-    /// plenty of such crates around.
-    ///
-    /// If you add support for caching different kinds of objects, each object kind
-    /// can have a separate mapping map, next to this field.
-    materialized_page_map: std::sync::RwLock<HashMap<MaterializedPageHashKey, Vec<Version>>>,
-
     immutable_page_map: std::sync::RwLock<HashMap<(FileId, u32), usize>>,
 
     /// The actual buffers with their metadata.
@@ -371,175 +327,14 @@ pub enum ReadBufResult<'a> {
 }
 
 impl PageCache {
-    //
-    // Section 1.1: Public interface functions for looking up and memorizing materialized page
-    // versions in the page cache
-    //
-
-    /// Look up a materialized page version.
-    ///
-    /// The 'lsn' is an upper bound, this will return the latest version of
-    /// the given block, but not newer than 'lsn'. Returns the actual LSN of the
-    /// returned page.
-    pub async fn lookup_materialized_page(
-        &self,
-        tenant_shard_id: TenantShardId,
-        timeline_id: TimelineId,
-        key: &Key,
-        lsn: Lsn,
-        ctx: &RequestContext,
-    ) -> Option<(Lsn, PageReadGuard)> {
-        let Ok(permit) = self.try_get_pinned_slot_permit().await else {
-            return None;
-        };
-
-        crate::metrics::PAGE_CACHE
-            .for_ctx(ctx)
-            .read_accesses_materialized_page
-            .inc();
-
-        let mut cache_key = CacheKey::MaterializedPage {
-            hash_key: MaterializedPageHashKey {
-                tenant_shard_id,
-                timeline_id,
-                key: *key,
-            },
-            lsn,
-        };
-
-        if let Some(guard) = self
-            .try_lock_for_read(&mut cache_key, &mut Some(permit))
-            .await
-        {
-            if let CacheKey::MaterializedPage {
-                hash_key: _,
-                lsn: available_lsn,
-            } = cache_key
-            {
-                if available_lsn == lsn {
-                    crate::metrics::PAGE_CACHE
-                        .for_ctx(ctx)
-                        .read_hits_materialized_page_exact
-                        .inc();
-                } else {
-                    crate::metrics::PAGE_CACHE
-                        .for_ctx(ctx)
-                        .read_hits_materialized_page_older_lsn
-                        .inc();
-                }
-                Some((available_lsn, guard))
-            } else {
-                panic!("unexpected key type in slot");
-            }
-        } else {
-            None
-        }
-    }
-
-    ///
-    /// Store an image of the given page in the cache.
-    ///
-    pub async fn memorize_materialized_page(
-        &self,
-        tenant_shard_id: TenantShardId,
-        timeline_id: TimelineId,
-        key: Key,
-        lsn: Lsn,
-        img: &[u8],
-    ) -> anyhow::Result<()> {
-        let cache_key = CacheKey::MaterializedPage {
-            hash_key: MaterializedPageHashKey {
-                tenant_shard_id,
-                timeline_id,
-                key,
-            },
-            lsn,
-        };
-
-        let mut permit = Some(self.try_get_pinned_slot_permit().await?);
-        loop {
-            // First check if the key already exists in the cache.
-            if let Some(slot_idx) = self.search_mapping_exact(&cache_key) {
-                // The page was found in the mapping. Lock the slot, and re-check
-                // that it's still what we expected (because we don't released the mapping
-                // lock already, another thread could have evicted the page)
-                let slot = &self.slots[slot_idx];
-                let inner = slot.inner.write().await;
-                if inner.key.as_ref() == Some(&cache_key) {
-                    slot.inc_usage_count();
-                    debug_assert!(
-                        {
-                            let guard = inner.permit.lock().unwrap();
-                            guard.upgrade().is_none()
-                        },
-                        "we hold a write lock, so, no one else should have a permit"
-                    );
-                    debug_assert_eq!(inner.buf.len(), img.len());
-                    // We already had it in cache. Another thread must've put it there
-                    // concurrently. Check that it had the same contents that we
-                    // replayed.
-                    assert!(inner.buf == img);
-                    return Ok(());
-                }
-            }
-            debug_assert!(permit.is_some());
-
-            // Not found. Find a victim buffer
-            let (slot_idx, mut inner) = self
-                .find_victim(permit.as_ref().unwrap())
-                .await
-                .context("Failed to find evict victim")?;
-
-            // Insert mapping for this. At this point, we may find that another
-            // thread did the same thing concurrently. In that case, we evicted
-            // our victim buffer unnecessarily. Put it into the free list and
-            // continue with the slot that the other thread chose.
-            if let Some(_existing_slot_idx) = self.try_insert_mapping(&cache_key, slot_idx) {
-                // TODO: put to free list
-
-                // We now just loop back to start from beginning. This is not
-                // optimal, we'll perform the lookup in the mapping again, which
-                // is not really necessary because we already got
-                // 'existing_slot_idx'.  But this shouldn't happen often enough
-                // to matter much.
-                continue;
-            }
-
-            // Make the slot ready
-            let slot = &self.slots[slot_idx];
-            inner.key = Some(cache_key.clone());
-            slot.set_usage_count(1);
-            // Create a write guard for the slot so we go through the expected motions.
-            debug_assert!(
-                {
-                    let guard = inner.permit.lock().unwrap();
-                    guard.upgrade().is_none()
-                },
-                "we hold a write lock, so, no one else should have a permit"
-            );
-            let mut write_guard = PageWriteGuard {
-                state: PageWriteGuardState::Invalid {
-                    _permit: permit.take().unwrap(),
-                    inner,
-                },
-            };
-            write_guard.copy_from_slice(img);
-            let _ = write_guard.mark_valid();
-            return Ok(());
-        }
-    }
-
-    // Section 1.2: Public interface functions for working with immutable file pages.
-
     pub async fn read_immutable_buf(
         &self,
         file_id: FileId,
         blkno: u32,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<ReadBufResult> {
-        let mut cache_key = CacheKey::ImmutableFilePage { file_id, blkno };
-
-        self.lock_for_read(&mut cache_key, ctx).await
+        self.lock_for_read(&(CacheKey::ImmutableFilePage { file_id, blkno }), ctx)
+            .await
     }
 
     //
@@ -573,19 +368,11 @@ impl PageCache {
 
     /// Look up a page in the cache.
     ///
-    /// If the search criteria is not exact, *cache_key is updated with the key
-    /// for exact key of the returned page. (For materialized pages, that means
-    /// that the LSN in 'cache_key' is updated with the LSN of the returned page
-    /// version.)
-    ///
-    /// If no page is found, returns None and *cache_key is left unmodified.
-    ///
     async fn try_lock_for_read(
         &self,
-        cache_key: &mut CacheKey,
+        cache_key: &CacheKey,
         permit: &mut Option<PinnedSlotsPermit>,
     ) -> Option<PageReadGuard> {
-        let cache_key_orig = cache_key.clone();
         if let Some(slot_idx) = self.search_mapping(cache_key) {
             // The page was found in the mapping. Lock the slot, and re-check
             // that it's still what we expected (because we released the mapping
@@ -598,9 +385,6 @@ impl PageCache {
                     _permit: inner.coalesce_readers_permit(permit.take().unwrap()),
                     slot_guard: inner,
                 });
-            } else {
-                // search_mapping might have modified the search key; restore it.
-                *cache_key = cache_key_orig;
             }
         }
         None
@@ -637,15 +421,12 @@ impl PageCache {
     ///
     async fn lock_for_read(
         &self,
-        cache_key: &mut CacheKey,
-        ctx: &RequestContext,
+        cache_key: &CacheKey,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<ReadBufResult> {
         let mut permit = Some(self.try_get_pinned_slot_permit().await?);
 
         let (read_access, hit) = match cache_key {
-            CacheKey::MaterializedPage { .. } => {
-                unreachable!("Materialized pages use lookup_materialized_page")
-            }
             CacheKey::ImmutableFilePage { .. } => (
                 &crate::metrics::PAGE_CACHE
                     .for_ctx(ctx)
@@ -717,52 +498,15 @@ impl PageCache {
 
     /// Search for a page in the cache using the given search key.
     ///
-    /// Returns the slot index, if any. If the search criteria is not exact,
-    /// *cache_key is updated with the actual key of the found page.
+    /// Returns the slot index, if any.
     ///
     /// NOTE: We don't hold any lock on the mapping on return, so the slot might
     /// get recycled for an unrelated page immediately after this function
     /// returns.  The caller is responsible for re-checking that the slot still
     /// contains the page with the same key before using it.
     ///
-    fn search_mapping(&self, cache_key: &mut CacheKey) -> Option<usize> {
+    fn search_mapping(&self, cache_key: &CacheKey) -> Option<usize> {
         match cache_key {
-            CacheKey::MaterializedPage { hash_key, lsn } => {
-                let map = self.materialized_page_map.read().unwrap();
-                let versions = map.get(hash_key)?;
-
-                let version_idx = match versions.binary_search_by_key(lsn, |v| v.lsn) {
-                    Ok(version_idx) => version_idx,
-                    Err(0) => return None,
-                    Err(version_idx) => version_idx - 1,
-                };
-                let version = &versions[version_idx];
-                *lsn = version.lsn;
-                Some(version.slot_idx)
-            }
-            CacheKey::ImmutableFilePage { file_id, blkno } => {
-                let map = self.immutable_page_map.read().unwrap();
-                Some(*map.get(&(*file_id, *blkno))?)
-            }
-        }
-    }
-
-    /// Search for a page in the cache using the given search key.
-    ///
-    /// Like 'search_mapping, but performs an "exact" search. Used for
-    /// allocating a new buffer.
-    fn search_mapping_exact(&self, key: &CacheKey) -> Option<usize> {
-        match key {
-            CacheKey::MaterializedPage { hash_key, lsn } => {
-                let map = self.materialized_page_map.read().unwrap();
-                let versions = map.get(hash_key)?;
-
-                if let Ok(version_idx) = versions.binary_search_by_key(lsn, |v| v.lsn) {
-                    Some(versions[version_idx].slot_idx)
-                } else {
-                    None
-                }
-            }
             CacheKey::ImmutableFilePage { file_id, blkno } => {
                 let map = self.immutable_page_map.read().unwrap();
                 Some(*map.get(&(*file_id, *blkno))?)
@@ -775,27 +519,6 @@ impl PageCache {
     ///
     fn remove_mapping(&self, old_key: &CacheKey) {
         match old_key {
-            CacheKey::MaterializedPage {
-                hash_key: old_hash_key,
-                lsn: old_lsn,
-            } => {
-                let mut map = self.materialized_page_map.write().unwrap();
-                if let Entry::Occupied(mut old_entry) = map.entry(old_hash_key.clone()) {
-                    let versions = old_entry.get_mut();
-
-                    if let Ok(version_idx) = versions.binary_search_by_key(old_lsn, |v| v.lsn) {
-                        versions.remove(version_idx);
-                        self.size_metrics
-                            .current_bytes_materialized_page
-                            .sub_page_sz(1);
-                        if versions.is_empty() {
-                            old_entry.remove_entry();
-                        }
-                    }
-                } else {
-                    panic!("could not find old key in mapping")
-                }
-            }
             CacheKey::ImmutableFilePage { file_id, blkno } => {
                 let mut map = self.immutable_page_map.write().unwrap();
                 map.remove(&(*file_id, *blkno))
@@ -812,30 +535,6 @@ impl PageCache {
     /// of the existing mapping and leaves it untouched.
     fn try_insert_mapping(&self, new_key: &CacheKey, slot_idx: usize) -> Option<usize> {
         match new_key {
-            CacheKey::MaterializedPage {
-                hash_key: new_key,
-                lsn: new_lsn,
-            } => {
-                let mut map = self.materialized_page_map.write().unwrap();
-                let versions = map.entry(new_key.clone()).or_default();
-                match versions.binary_search_by_key(new_lsn, |v| v.lsn) {
-                    Ok(version_idx) => Some(versions[version_idx].slot_idx),
-                    Err(version_idx) => {
-                        versions.insert(
-                            version_idx,
-                            Version {
-                                lsn: *new_lsn,
-                                slot_idx,
-                            },
-                        );
-                        self.size_metrics
-                            .current_bytes_materialized_page
-                            .add_page_sz(1);
-                        None
-                    }
-                }
-            }
-
             CacheKey::ImmutableFilePage { file_id, blkno } => {
                 let mut map = self.immutable_page_map.write().unwrap();
                 match map.entry((*file_id, *blkno)) {
@@ -949,7 +648,6 @@ impl PageCache {
         let size_metrics = &crate::metrics::PAGE_CACHE_SIZE;
         size_metrics.max_bytes.set_page_sz(num_pages);
         size_metrics.current_bytes_immutable.set_page_sz(0);
-        size_metrics.current_bytes_materialized_page.set_page_sz(0);
 
         let slots = page_buffer
             .chunks_exact_mut(PAGE_SZ)
@@ -968,7 +666,6 @@ impl PageCache {
             .collect();
 
         Self {
-            materialized_page_map: Default::default(),
             immutable_page_map: Default::default(),
             slots,
             next_evict_slot: AtomicUsize::new(0),

pageserver/src/page_service.rs

@@ -554,7 +554,7 @@ impl PageServerHandler {
         tenant_id: TenantId,
         timeline_id: TimelineId,
         protocol_version: PagestreamProtocolVersion,
-        ctx: RequestContext,
+        mut ctx: RequestContext,
     ) -> Result<(), QueryError>
     where
         IO: AsyncRead + AsyncWrite + Send + Sync + Unpin,
@@ -624,7 +624,7 @@ impl PageServerHandler {
                     fail::fail_point!("ps::handle-pagerequest-message::exists");
                     let span = tracing::info_span!("handle_get_rel_exists_request", rel = %req.rel, req_lsn = %req.request_lsn);
                     (
-                        self.handle_get_rel_exists_request(tenant_id, timeline_id, &req, &ctx)
+                        self.handle_get_rel_exists_request(tenant_id, timeline_id, &req, &mut ctx)
                             .instrument(span.clone())
                             .await,
                         span,
@@ -634,7 +634,7 @@ impl PageServerHandler {
                     fail::fail_point!("ps::handle-pagerequest-message::nblocks");
                     let span = tracing::info_span!("handle_get_nblocks_request", rel = %req.rel, req_lsn = %req.request_lsn);
                     (
-                        self.handle_get_nblocks_request(tenant_id, timeline_id, &req, &ctx)
+                        self.handle_get_nblocks_request(tenant_id, timeline_id, &req, &mut ctx)
                             .instrument(span.clone())
                             .await,
                         span,
@@ -645,7 +645,7 @@ impl PageServerHandler {
                     // shard_id is filled in by the handler
                     let span = tracing::info_span!("handle_get_page_at_lsn_request", rel = %req.rel, blkno = %req.blkno, req_lsn = %req.request_lsn);
                     (
-                        self.handle_get_page_at_lsn_request(tenant_id, timeline_id, &req, &ctx)
+                        self.handle_get_page_at_lsn_request(tenant_id, timeline_id, &req, &mut ctx)
                             .instrument(span.clone())
                             .await,
                         span,
@@ -655,7 +655,7 @@ impl PageServerHandler {
                     fail::fail_point!("ps::handle-pagerequest-message::dbsize");
                     let span = tracing::info_span!("handle_db_size_request", dbnode = %req.dbnode, req_lsn = %req.request_lsn);
                     (
-                        self.handle_db_size_request(tenant_id, timeline_id, &req, &ctx)
+                        self.handle_db_size_request(tenant_id, timeline_id, &req, &mut ctx)
                             .instrument(span.clone())
                             .await,
                         span,
@@ -665,7 +665,7 @@ impl PageServerHandler {
                     fail::fail_point!("ps::handle-pagerequest-message::slrusegment");
                     let span = tracing::info_span!("handle_get_slru_segment_request", kind = %req.kind, segno = %req.segno, req_lsn = %req.request_lsn);
                     (
-                        self.handle_get_slru_segment_request(tenant_id, timeline_id, &req, &ctx)
+                        self.handle_get_slru_segment_request(tenant_id, timeline_id, &req, &mut ctx)
                             .instrument(span.clone())
                             .await,
                         span,
@@ -728,7 +728,7 @@ impl PageServerHandler {
         base_lsn: Lsn,
         _end_lsn: Lsn,
         pg_version: u32,
-        ctx: RequestContext,
+        mut ctx: RequestContext,
     ) -> Result<(), QueryError>
     where
         IO: AsyncRead + AsyncWrite + Send + Sync + Unpin,
@@ -741,7 +741,7 @@ impl PageServerHandler {
             .get_active_tenant_with_timeout(tenant_id, ShardSelector::Zero, ACTIVE_TENANT_TIMEOUT)
             .await?;
         let timeline = tenant
-            .create_empty_timeline(timeline_id, base_lsn, pg_version, &ctx)
+            .create_empty_timeline(timeline_id, base_lsn, pg_version, &mut ctx)
             .await?;
 
         // TODO mark timeline as not ready until it reaches end_lsn.
@@ -766,7 +766,7 @@ impl PageServerHandler {
                 &mut copyin_reader,
                 base_lsn,
                 self.broker_client.clone(),
-                &ctx,
+                &mut ctx,
             )
             .await?;
 
@@ -791,7 +791,7 @@ impl PageServerHandler {
         timeline_id: TimelineId,
         start_lsn: Lsn,
         end_lsn: Lsn,
-        ctx: RequestContext,
+        mut ctx: RequestContext,
     ) -> Result<(), QueryError>
     where
         IO: AsyncRead + AsyncWrite + Send + Sync + Unpin,
@@ -814,7 +814,7 @@ impl PageServerHandler {
         pgb.write_message_noflush(&BeMessage::CopyInResponse)?;
         self.flush_cancellable(pgb, &timeline.cancel).await?;
         let mut copyin_reader = pin!(StreamReader::new(self.copyin_stream(pgb, &timeline.cancel)));
-        import_wal_from_tar(&timeline, &mut copyin_reader, start_lsn, end_lsn, &ctx).await?;
+        import_wal_from_tar(&timeline, &mut copyin_reader, start_lsn, end_lsn, &mut ctx).await?;
         info!("wal import complete");
 
         // Read the end of the tar archive.
@@ -867,7 +867,7 @@ impl PageServerHandler {
         request_lsn: Lsn,
         not_modified_since: Lsn,
         latest_gc_cutoff_lsn: &RcuReadGuard<Lsn>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Lsn, PageStreamError> {
         let last_record_lsn = timeline.get_last_record_lsn();
 
@@ -926,7 +926,7 @@ impl PageServerHandler {
         tenant_shard_id: TenantShardId,
         timeline_id: TimelineId,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), QueryError>
     where
         IO: AsyncRead + AsyncWrite + Send + Sync + Unpin,
@@ -958,12 +958,12 @@ impl PageServerHandler {
         tenant_id: TenantId,
         timeline_id: TimelineId,
         req: &PagestreamExistsRequest,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<PagestreamBeMessage, PageStreamError> {
         let timeline = self.get_timeline_shard_zero(tenant_id, timeline_id).await?;
         let _timer = timeline
             .query_metrics
-            .start_timer(metrics::SmgrQueryType::GetRelExists, ctx);
+            .start_timer(metrics::SmgrQueryType::GetRelExists);
 
         let latest_gc_cutoff_lsn = timeline.get_latest_gc_cutoff_lsn();
         let lsn = Self::wait_or_get_last_lsn(
@@ -990,13 +990,13 @@ impl PageServerHandler {
         tenant_id: TenantId,
         timeline_id: TimelineId,
         req: &PagestreamNblocksRequest,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<PagestreamBeMessage, PageStreamError> {
         let timeline = self.get_timeline_shard_zero(tenant_id, timeline_id).await?;
 
         let _timer = timeline
             .query_metrics
-            .start_timer(metrics::SmgrQueryType::GetRelSize, ctx);
+            .start_timer(metrics::SmgrQueryType::GetRelSize);
 
         let latest_gc_cutoff_lsn = timeline.get_latest_gc_cutoff_lsn();
         let lsn = Self::wait_or_get_last_lsn(
@@ -1023,13 +1023,13 @@ impl PageServerHandler {
         tenant_id: TenantId,
         timeline_id: TimelineId,
         req: &PagestreamDbSizeRequest,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<PagestreamBeMessage, PageStreamError> {
         let timeline = self.get_timeline_shard_zero(tenant_id, timeline_id).await?;
 
         let _timer = timeline
             .query_metrics
-            .start_timer(metrics::SmgrQueryType::GetDbSize, ctx);
+            .start_timer(metrics::SmgrQueryType::GetDbSize);
 
         let latest_gc_cutoff_lsn = timeline.get_latest_gc_cutoff_lsn();
         let lsn = Self::wait_or_get_last_lsn(
@@ -1173,7 +1173,7 @@ impl PageServerHandler {
         tenant_id: TenantId,
         timeline_id: TimelineId,
         req: &PagestreamGetPageRequest,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<PagestreamBeMessage, PageStreamError> {
         let timeline = match self.get_cached_timeline_for_page(req) {
             Ok(tl) => {
@@ -1206,7 +1206,7 @@ impl PageServerHandler {
 
         let _timer = timeline
             .query_metrics
-            .start_timer(metrics::SmgrQueryType::GetPageAtLsn, ctx);
+            .start_timer(metrics::SmgrQueryType::GetPageAtLsn);
 
         let latest_gc_cutoff_lsn = timeline.get_latest_gc_cutoff_lsn();
         let lsn = Self::wait_or_get_last_lsn(
@@ -1233,13 +1233,13 @@ impl PageServerHandler {
         tenant_id: TenantId,
         timeline_id: TimelineId,
         req: &PagestreamGetSlruSegmentRequest,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<PagestreamBeMessage, PageStreamError> {
         let timeline = self.get_timeline_shard_zero(tenant_id, timeline_id).await?;
 
         let _timer = timeline
             .query_metrics
-            .start_timer(metrics::SmgrQueryType::GetSlruSegment, ctx);
+            .start_timer(metrics::SmgrQueryType::GetSlruSegment);
 
         let latest_gc_cutoff_lsn = timeline.get_latest_gc_cutoff_lsn();
         let lsn = Self::wait_or_get_last_lsn(
@@ -1275,7 +1275,7 @@ impl PageServerHandler {
         prev_lsn: Option<Lsn>,
         full_backup: bool,
         gzip: bool,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), QueryError>
     where
         IO: AsyncRead + AsyncWrite + Send + Sync + Unpin,
@@ -1534,7 +1534,7 @@ where
 
         fail::fail_point!("ps::connection-start::process-query");
 
-        let ctx = self.connection_ctx.attached_child();
+        let mut ctx = self.connection_ctx.attached_child();
         debug!("process query {query_string:?}");
         let parts = query_string.split_whitespace().collect::<Vec<_>>();
         if let Some(params) = parts.strip_prefix(&["pagestream_v2"]) {
@@ -1624,7 +1624,7 @@ where
                 }
             };
 
-            let metric_recording = metrics::BASEBACKUP_QUERY_TIME.start_recording(&ctx);
+            let metric_recording = metrics::BASEBACKUP_QUERY_TIME.start_recording(&mut ctx);
             let res = async {
                 self.handle_basebackup_request(
                     pgb,
@@ -1634,7 +1634,7 @@ where
                     None,
                     false,
                     gzip,
-                    &ctx,
+                    &mut ctx,
                 )
                 .await?;
                 pgb.write_message_noflush(&BeMessage::CommandComplete(b"SELECT 1"))?;
@@ -1732,7 +1732,7 @@ where
                 prev_lsn,
                 true,
                 false,
-                &ctx,
+                &mut ctx,
             )
             .await?;
             pgb.write_message_noflush(&BeMessage::CommandComplete(b"SELECT 1"))?;
@@ -1860,7 +1860,7 @@ where
                 .with_context(|| format!("Failed to parse Lsn from {}", params[2]))?;
 
             match self
-                .handle_make_lsn_lease(pgb, tenant_shard_id, timeline_id, lsn, &ctx)
+                .handle_make_lsn_lease(pgb, tenant_shard_id, timeline_id, lsn, &mut ctx)
                 .await
             {
                 Ok(()) => pgb.write_message_noflush(&BeMessage::CommandComplete(b"SELECT 1"))?,

pageserver/src/pgdatadir_mapping.rs

@@ -188,7 +188,7 @@ impl Timeline {
         tag: RelTag,
         blknum: BlockNumber,
         version: Version<'_>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Bytes, PageReconstructError> {
         if tag.relnode == 0 {
             return Err(PageReconstructError::Other(
@@ -218,7 +218,7 @@ impl Timeline {
         spcnode: Oid,
         dbnode: Oid,
         version: Version<'_>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<usize, PageReconstructError> {
         let mut total_blocks = 0;
 
@@ -236,7 +236,7 @@ impl Timeline {
         &self,
         tag: RelTag,
         version: Version<'_>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<BlockNumber, PageReconstructError> {
         if tag.relnode == 0 {
             return Err(PageReconstructError::Other(
@@ -272,7 +272,7 @@ impl Timeline {
         &self,
         tag: RelTag,
         version: Version<'_>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<bool, PageReconstructError> {
         if tag.relnode == 0 {
             return Err(PageReconstructError::Other(
@@ -307,7 +307,7 @@ impl Timeline {
         spcnode: Oid,
         dbnode: Oid,
         version: Version<'_>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<HashSet<RelTag>, PageReconstructError> {
         // fetch directory listing
         let key = rel_dir_to_key(spcnode, dbnode);
@@ -335,7 +335,7 @@ impl Timeline {
         kind: SlruKind,
         segno: u32,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Bytes, PageReconstructError> {
         let n_blocks = self
             .get_slru_segment_size(kind, segno, Version::Lsn(lsn), ctx)
@@ -357,7 +357,7 @@ impl Timeline {
         segno: u32,
         blknum: BlockNumber,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Bytes, PageReconstructError> {
         let key = slru_block_to_key(kind, segno, blknum);
         self.get(key, lsn, ctx).await
@@ -369,7 +369,7 @@ impl Timeline {
         kind: SlruKind,
         segno: u32,
         version: Version<'_>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<BlockNumber, PageReconstructError> {
         let key = slru_segment_size_to_key(kind, segno);
         let mut buf = version.get(self, key, ctx).await?;
@@ -382,7 +382,7 @@ impl Timeline {
         kind: SlruKind,
         segno: u32,
         version: Version<'_>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<bool, PageReconstructError> {
         // fetch directory listing
         let key = slru_dir_to_key(kind);
@@ -408,7 +408,7 @@ impl Timeline {
         &self,
         search_timestamp: TimestampTz,
         cancel: &CancellationToken,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<LsnForTimestamp, PageReconstructError> {
         pausable_failpoint!("find-lsn-for-timestamp-pausable");
 
@@ -499,7 +499,7 @@ impl Timeline {
         probe_lsn: Lsn,
         found_smaller: &mut bool,
         found_larger: &mut bool,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<bool, PageReconstructError> {
         self.map_all_timestamps(probe_lsn, ctx, |timestamp| {
             if timestamp >= search_timestamp {
@@ -519,7 +519,7 @@ impl Timeline {
     pub(crate) async fn get_timestamp_for_lsn(
         &self,
         probe_lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Option<TimestampTz>, PageReconstructError> {
         let mut max: Option<TimestampTz> = None;
         self.map_all_timestamps(probe_lsn, ctx, |timestamp| {
@@ -542,7 +542,7 @@ impl Timeline {
     async fn map_all_timestamps<T: Default>(
         &self,
         probe_lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
         mut f: impl FnMut(TimestampTz) -> ControlFlow<T>,
     ) -> Result<T, PageReconstructError> {
         for segno in self
@@ -575,7 +575,7 @@ impl Timeline {
     pub(crate) async fn get_slru_keyspace(
         &self,
         version: Version<'_>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<KeySpace, PageReconstructError> {
         let mut accum = KeySpaceAccum::new();
 
@@ -604,7 +604,7 @@ impl Timeline {
         &self,
         kind: SlruKind,
         version: Version<'_>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<HashSet<u32>, PageReconstructError> {
         // fetch directory entry
         let key = slru_dir_to_key(kind);
@@ -621,7 +621,7 @@ impl Timeline {
         spcnode: Oid,
         dbnode: Oid,
         version: Version<'_>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Bytes, PageReconstructError> {
         let key = relmap_file_key(spcnode, dbnode);
 
@@ -632,7 +632,7 @@ impl Timeline {
     pub(crate) async fn list_dbdirs(
         &self,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<HashMap<(Oid, Oid), bool>, PageReconstructError> {
         // fetch directory entry
         let buf = self.get(DBDIR_KEY, lsn, ctx).await?;
@@ -647,7 +647,7 @@ impl Timeline {
         &self,
         xid: TransactionId,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Bytes, PageReconstructError> {
         let key = twophase_file_key(xid);
         let buf = self.get(key, lsn, ctx).await?;
@@ -657,7 +657,7 @@ impl Timeline {
     pub(crate) async fn list_twophase_files(
         &self,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<HashSet<TransactionId>, PageReconstructError> {
         // fetch directory entry
         let buf = self.get(TWOPHASEDIR_KEY, lsn, ctx).await?;
@@ -671,7 +671,7 @@ impl Timeline {
     pub(crate) async fn get_control_file(
         &self,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Bytes, PageReconstructError> {
         self.get(CONTROLFILE_KEY, lsn, ctx).await
     }
@@ -679,7 +679,7 @@ impl Timeline {
     pub(crate) async fn get_checkpoint(
         &self,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Bytes, PageReconstructError> {
         self.get(CHECKPOINT_KEY, lsn, ctx).await
     }
@@ -687,7 +687,7 @@ impl Timeline {
     async fn list_aux_files_v1(
         &self,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<HashMap<String, Bytes>, PageReconstructError> {
         match self.get(AUX_FILES_KEY, lsn, ctx).await {
             Ok(buf) => match AuxFilesDirectory::des(&buf).context("deserialization failure") {
@@ -705,7 +705,7 @@ impl Timeline {
     async fn list_aux_files_v2(
         &self,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<HashMap<String, Bytes>, PageReconstructError> {
         let kv = self
             .scan(KeySpace::single(Key::metadata_aux_key_range()), lsn, ctx)
@@ -729,7 +729,7 @@ impl Timeline {
     pub(crate) async fn trigger_aux_file_size_computation(
         &self,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), PageReconstructError> {
         let current_policy = self.last_aux_file_policy.load();
         if let Some(AuxFilePolicy::V2) | Some(AuxFilePolicy::CrossValidation) = current_policy {
@@ -741,7 +741,7 @@ impl Timeline {
     pub(crate) async fn list_aux_files(
         &self,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<HashMap<String, Bytes>, PageReconstructError> {
         let current_policy = self.last_aux_file_policy.load();
         match current_policy {
@@ -779,7 +779,7 @@ impl Timeline {
     pub(crate) async fn get_replorigins(
         &self,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<HashMap<RepOriginId, Lsn>, PageReconstructError> {
         let kv = self
             .scan(KeySpace::single(repl_origin_key_range()), lsn, ctx)
@@ -809,7 +809,7 @@ impl Timeline {
     pub(crate) async fn get_current_logical_size_non_incremental(
         &self,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<u64, CalculateLogicalSizeError> {
         debug_assert_current_span_has_tenant_and_timeline_id_no_shard_id();
 
@@ -845,7 +845,7 @@ impl Timeline {
     pub(crate) async fn collect_keyspace(
         &self,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(KeySpace, SparseKeySpace), CollectKeySpaceError> {
         // Iterate through key ranges, greedily packing them into partitions
         let mut result = KeySpaceAccum::new();
@@ -1145,7 +1145,7 @@ impl<'a> DatadirModification<'a> {
         spcnode: Oid,
         dbnode: Oid,
         img: Bytes,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         // Add it to the directory (if it doesn't exist already)
         let buf = self.get(DBDIR_KEY, ctx).await?;
@@ -1182,7 +1182,7 @@ impl<'a> DatadirModification<'a> {
         &mut self,
         xid: TransactionId,
         img: Bytes,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         // Add it to the directory entry
         let buf = self.get(TWOPHASEDIR_KEY, ctx).await?;
@@ -1229,7 +1229,7 @@ impl<'a> DatadirModification<'a> {
         &mut self,
         spcnode: Oid,
         dbnode: Oid,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         let total_blocks = self
             .tline
@@ -1266,7 +1266,7 @@ impl<'a> DatadirModification<'a> {
         &mut self,
         rel: RelTag,
         nblocks: BlockNumber,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), RelationError> {
         if rel.relnode == 0 {
             return Err(RelationError::InvalidRelnode);
@@ -1328,7 +1328,7 @@ impl<'a> DatadirModification<'a> {
         &mut self,
         rel: RelTag,
         nblocks: BlockNumber,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         anyhow::ensure!(rel.relnode != 0, RelationError::InvalidRelnode);
         if self
@@ -1362,7 +1362,7 @@ impl<'a> DatadirModification<'a> {
         &mut self,
         rel: RelTag,
         nblocks: BlockNumber,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         anyhow::ensure!(rel.relnode != 0, RelationError::InvalidRelnode);
 
@@ -1384,7 +1384,11 @@ impl<'a> DatadirModification<'a> {
     }
 
     /// Drop a relation.
-    pub async fn put_rel_drop(&mut self, rel: RelTag, ctx: &RequestContext) -> anyhow::Result<()> {
+    pub async fn put_rel_drop(
+        &mut self,
+        rel: RelTag,
+        ctx: &mut RequestContext,
+    ) -> anyhow::Result<()> {
         anyhow::ensure!(rel.relnode != 0, RelationError::InvalidRelnode);
 
         // Remove it from the directory entry
@@ -1420,7 +1424,7 @@ impl<'a> DatadirModification<'a> {
         kind: SlruKind,
         segno: u32,
         nblocks: BlockNumber,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         // Add it to the directory entry
         let dir_key = slru_dir_to_key(kind);
@@ -1466,7 +1470,7 @@ impl<'a> DatadirModification<'a> {
         &mut self,
         kind: SlruKind,
         segno: u32,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         // Remove it from the directory entry
         let dir_key = slru_dir_to_key(kind);
@@ -1499,7 +1503,7 @@ impl<'a> DatadirModification<'a> {
     pub async fn drop_twophase_file(
         &mut self,
         xid: TransactionId,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         // Remove it from the directory entry
         let buf = self.get(TWOPHASEDIR_KEY, ctx).await?;
@@ -1538,7 +1542,7 @@ impl<'a> DatadirModification<'a> {
         &mut self,
         path: &str,
         content: &[u8],
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         let switch_policy = self.tline.get_switch_aux_file_policy();
 
@@ -1731,7 +1735,7 @@ impl<'a> DatadirModification<'a> {
     /// retains all the metadata, but data pages are flushed. That's again OK
     /// for bulk import, where you are just loading data pages and won't try to
     /// modify the same pages twice.
-    pub async fn flush(&mut self, ctx: &RequestContext) -> anyhow::Result<()> {
+    pub async fn flush(&mut self, ctx: &mut RequestContext) -> anyhow::Result<()> {
         // Unless we have accumulated a decent amount of changes, it's not worth it
         // to scan through the pending_updates list.
         let pending_nblocks = self.pending_nblocks;
@@ -1777,7 +1781,7 @@ impl<'a> DatadirModification<'a> {
     /// underlying timeline.
     /// All the modifications in this atomic update are stamped by the specified LSN.
     ///
-    pub async fn commit(&mut self, ctx: &RequestContext) -> anyhow::Result<()> {
+    pub async fn commit(&mut self, ctx: &mut RequestContext) -> anyhow::Result<()> {
         let mut writer = self.tline.writer().await;
 
         let pending_nblocks = self.pending_nblocks;
@@ -1828,7 +1832,7 @@ impl<'a> DatadirModification<'a> {
 
     // Internal helper functions to batch the modifications
 
-    async fn get(&self, key: Key, ctx: &RequestContext) -> Result<Bytes, PageReconstructError> {
+    async fn get(&self, key: Key, ctx: &mut RequestContext) -> Result<Bytes, PageReconstructError> {
         // Have we already updated the same key? Read the latest pending updated
         // version in that case.
         //
@@ -1895,7 +1899,7 @@ impl<'a> Version<'a> {
         &self,
         timeline: &Timeline,
         key: Key,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Bytes, PageReconstructError> {
         match self {
             Version::Lsn(lsn) => timeline.get(key, *lsn, ctx).await,

pageserver/src/tenant.rs

@@ -547,7 +547,7 @@ impl Tenant {
         metadata: TimelineMetadata,
         ancestor: Option<Arc<Timeline>>,
         last_aux_file_policy: Option<AuxFilePolicy>,
-        _ctx: &RequestContext,
+        _ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         let tenant_id = self.tenant_shard_id;
 
@@ -656,7 +656,7 @@ impl Tenant {
         init_order: Option<InitializationOrder>,
         tenants: &'static std::sync::RwLock<TenantsMap>,
         mode: SpawnMode,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<Arc<Tenant>> {
         let wal_redo_manager = Arc::new(WalRedoManager::from(PostgresRedoManager::new(
             conf,
@@ -965,7 +965,7 @@ impl Tenant {
         self: &Arc<Tenant>,
         preload: Option<TenantPreload>,
         mode: SpawnMode,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         span::debug_assert_current_span_has_tenant_id();
 
@@ -1175,7 +1175,7 @@ impl Tenant {
         index_part: IndexPart,
         remote_metadata: TimelineMetadata,
         resources: TimelineResources,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         span::debug_assert_current_span_has_tenant_id();
 
@@ -1358,7 +1358,7 @@ impl Tenant {
         new_timeline_id: TimelineId,
         initdb_lsn: Lsn,
         pg_version: u32,
-        _ctx: &RequestContext,
+        _ctx: &mut RequestContext,
     ) -> anyhow::Result<UninitializedTimeline> {
         anyhow::ensure!(
             self.is_active(),
@@ -1401,7 +1401,7 @@ impl Tenant {
         new_timeline_id: TimelineId,
         initdb_lsn: Lsn,
         pg_version: u32,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<Arc<Timeline>> {
         let uninit_tl = self
             .create_empty_timeline(new_timeline_id, initdb_lsn, pg_version, ctx)
@@ -1440,7 +1440,7 @@ impl Tenant {
         new_timeline_id: TimelineId,
         initdb_lsn: Lsn,
         pg_version: u32,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
         delta_layer_desc: Vec<Vec<(pageserver_api::key::Key, Lsn, crate::repository::Value)>>,
         image_layer_desc: Vec<(Lsn, Vec<(pageserver_api::key::Key, bytes::Bytes)>)>,
         end_lsn: Lsn,
@@ -1477,7 +1477,7 @@ impl Tenant {
         pg_version: u32,
         load_existing_initdb: Option<TimelineId>,
         broker_client: storage_broker::BrokerClientChannel,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Arc<Timeline>, CreateTimelineError> {
         if !self.is_active() {
             if matches!(self.current_state(), TenantState::Stopping { .. }) {
@@ -1650,7 +1650,7 @@ impl Tenant {
         horizon: u64,
         pitr: Duration,
         cancel: &CancellationToken,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<GcResult, GcError> {
         // Don't start doing work during shutdown
         if let TenantState::Stopping { .. } = self.current_state() {
@@ -1682,7 +1682,7 @@ impl Tenant {
     async fn compaction_iteration(
         &self,
         cancel: &CancellationToken,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<(), timeline::CompactionError> {
         // Don't start doing work during shutdown, or when broken, we do not need those in the logs
         if !self.is_active() {
@@ -1779,7 +1779,7 @@ impl Tenant {
         self: &Arc<Self>,
         broker_client: BrokerClientChannel,
         background_jobs_can_start: Option<&completion::Barrier>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) {
         span::debug_assert_current_span_has_tenant_id();
 
@@ -2834,7 +2834,7 @@ impl Tenant {
         horizon: u64,
         pitr: Duration,
         cancel: &CancellationToken,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<GcResult, GcError> {
         let mut totals: GcResult = Default::default();
         let now = Instant::now();
@@ -2894,7 +2894,7 @@ impl Tenant {
     pub(crate) async fn refresh_gc_info(
         &self,
         cancel: &CancellationToken,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Vec<Arc<Timeline>>, GcError> {
         // since this method can now be called at different rates than the configured gc loop, it
         // might be that these configuration values get applied faster than what it was previously,
@@ -2915,7 +2915,7 @@ impl Tenant {
         horizon: u64,
         pitr: Duration,
         cancel: &CancellationToken,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Vec<Arc<Timeline>>, GcError> {
         // before taking the gc_cs lock, do the heavier weight finding of gc_cutoff points for
         // currently visible timelines.
@@ -3053,7 +3053,7 @@ impl Tenant {
         src_timeline: &Arc<Timeline>,
         dst_id: TimelineId,
         ancestor_lsn: Option<Lsn>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Arc<Timeline>, CreateTimelineError> {
         let create_guard = self.create_timeline_create_guard(dst_id).unwrap();
         let tl = self
@@ -3071,7 +3071,7 @@ impl Tenant {
         src_timeline: &Arc<Timeline>,
         dst_id: TimelineId,
         ancestor_lsn: Option<Lsn>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
         delta_layer_desc: Vec<Vec<(pageserver_api::key::Key, Lsn, crate::repository::Value)>>,
         image_layer_desc: Vec<(Lsn, Vec<(pageserver_api::key::Key, bytes::Bytes)>)>,
         end_lsn: Lsn,
@@ -3108,7 +3108,7 @@ impl Tenant {
         dst_id: TimelineId,
         start_lsn: Option<Lsn>,
         timeline_create_guard: TimelineCreateGuard<'_>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Arc<Timeline>, CreateTimelineError> {
         self.branch_timeline_impl(src_timeline, dst_id, start_lsn, timeline_create_guard, ctx)
             .await
@@ -3120,7 +3120,7 @@ impl Tenant {
         dst_id: TimelineId,
         start_lsn: Option<Lsn>,
         timeline_create_guard: TimelineCreateGuard<'_>,
-        _ctx: &RequestContext,
+        _ctx: &mut RequestContext,
     ) -> Result<Arc<Timeline>, CreateTimelineError> {
         let src_id = src_timeline.timeline_id;
 
@@ -3233,7 +3233,7 @@ impl Tenant {
         timeline_id: TimelineId,
         pg_version: u32,
         load_existing_initdb: Option<TimelineId>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<Arc<Timeline>> {
         let create_guard = self.create_timeline_create_guard(timeline_id).unwrap();
         self.bootstrap_timeline(
@@ -3305,7 +3305,7 @@ impl Tenant {
         pg_version: u32,
         load_existing_initdb: Option<TimelineId>,
         timeline_create_guard: TimelineCreateGuard<'_>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<Arc<Timeline>> {
         // create a `tenant/{tenant_id}/timelines/basebackup-{timeline_id}.{TEMP_FILE_SUFFIX}/`
         // temporary directory for basebackup files for the given timeline.
@@ -3563,7 +3563,7 @@ impl Tenant {
         max_retention_period: Option<u64>,
         cause: LogicalSizeCalculationCause,
         cancel: &CancellationToken,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<size::ModelInputs, size::CalculateSyntheticSizeError> {
         let logical_sizes_at_once = self
             .conf
@@ -3603,7 +3603,7 @@ impl Tenant {
         &self,
         cause: LogicalSizeCalculationCause,
         cancel: &CancellationToken,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<u64, size::CalculateSyntheticSizeError> {
         let inputs = self.gather_size_inputs(None, cause, cancel, ctx).await?;
 
@@ -3756,7 +3756,7 @@ async fn run_initdb(
 pub async fn dump_layerfile_from_path(
     path: &Utf8Path,
     verbose: bool,
-    ctx: &RequestContext,
+    ctx: &mut RequestContext,
 ) -> anyhow::Result<()> {
     use std::os::unix::fs::FileExt;
 
@@ -3960,7 +3960,7 @@ pub(crate) mod harness {
         #[instrument(skip_all, fields(tenant_id=%self.tenant_shard_id.tenant_id, shard_id=%self.tenant_shard_id.shard_slug()))]
         pub(crate) async fn do_try_load(
             &self,
-            ctx: &RequestContext,
+            ctx: &mut RequestContext,
         ) -> anyhow::Result<Arc<Tenant>> {
             let walredo_mgr = Arc::new(WalRedoManager::from(TestRedoManager));
 
@@ -4219,7 +4219,7 @@ mod tests {
     async fn make_some_layers(
         tline: &Timeline,
         start_lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         let mut lsn = start_lsn;
         {
@@ -4707,7 +4707,7 @@ mod tests {
     async fn bulk_insert_compact_gc(
         tenant: &Tenant,
         timeline: &Arc<Timeline>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
         lsn: Lsn,
         repeat: usize,
         key_count: usize,
@@ -4719,7 +4719,7 @@ mod tests {
     async fn bulk_insert_maybe_compact_gc(
         tenant: &Tenant,
         timeline: &Arc<Timeline>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
         mut lsn: Lsn,
         repeat: usize,
         key_count: usize,
@@ -6249,7 +6249,7 @@ mod tests {
             tline: &Timeline,
             keyspace: &KeySpace,
             lsn: Lsn,
-            ctx: &RequestContext,
+            ctx: &mut RequestContext,
         ) -> anyhow::Result<(BTreeMap<Key, Result<Bytes, PageReconstructError>>, usize)> {
             let mut reconstruct_state = ValuesReconstructState::default();
             let res = tline
@@ -6365,7 +6365,7 @@ mod tests {
             tline: &Arc<Timeline>,
             key: Key,
             lsn: Lsn,
-            ctx: &RequestContext,
+            ctx: &mut RequestContext,
         ) -> Result<Option<Bytes>, GetVectoredError> {
             let mut reconstruct_state = ValuesReconstructState::new();
             let mut res = tline
@@ -6461,7 +6461,7 @@ mod tests {
             tline: &Arc<Timeline>,
             key: Key,
             lsn: Lsn,
-            ctx: &RequestContext,
+            ctx: &mut RequestContext,
         ) -> Result<Option<Bytes>, GetVectoredError> {
             let mut reconstruct_state = ValuesReconstructState::new();
             let mut res = tline
@@ -6515,7 +6515,7 @@ mod tests {
         tline: &Arc<Timeline>,
         key: Key,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Option<Bytes>, GetVectoredError> {
         let mut reconstruct_state = ValuesReconstructState::new();
         let mut res = tline

pageserver/src/tenant/blob_io.rs

@@ -26,7 +26,7 @@ impl<'a> BlockCursor<'a> {
     pub async fn read_blob(
         &self,
         offset: u64,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Vec<u8>, std::io::Error> {
         let mut buf = Vec::new();
         self.read_blob_into_buf(offset, &mut buf, ctx).await?;
@@ -38,7 +38,7 @@ impl<'a> BlockCursor<'a> {
         &self,
         offset: u64,
         dstbuf: &mut Vec<u8>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), std::io::Error> {
         let mut blknum = (offset / PAGE_SZ as u64) as u32;
         let mut off = (offset % PAGE_SZ as u64) as usize;
@@ -130,7 +130,7 @@ impl<const BUFFERED: bool> BlobWriter<BUFFERED> {
     async fn write_all_unbuffered<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
         &mut self,
         src_buf: B,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> (B::Buf, Result<(), Error>) {
         let (src_buf, res) = self.inner.write_all(src_buf, ctx).await;
         let nbytes = match res {
@@ -143,7 +143,7 @@ impl<const BUFFERED: bool> BlobWriter<BUFFERED> {
 
     #[inline(always)]
     /// Flushes the internal buffer to the underlying `VirtualFile`.
-    pub async fn flush_buffer(&mut self, ctx: &RequestContext) -> Result<(), Error> {
+    pub async fn flush_buffer(&mut self, ctx: &mut RequestContext) -> Result<(), Error> {
         let buf = std::mem::take(&mut self.buf);
         let (mut buf, res) = self.inner.write_all(buf, ctx).await;
         res?;
@@ -166,7 +166,7 @@ impl<const BUFFERED: bool> BlobWriter<BUFFERED> {
     async fn write_all<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
         &mut self,
         src_buf: B,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> (B::Buf, Result<(), Error>) {
         if !BUFFERED {
             assert!(self.buf.is_empty());
@@ -218,7 +218,7 @@ impl<const BUFFERED: bool> BlobWriter<BUFFERED> {
     pub async fn write_blob<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
         &mut self,
         srcbuf: B,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> (B::Buf, Result<u64, Error>) {
         let offset = self.offset;
 
@@ -267,7 +267,7 @@ impl BlobWriter<true> {
     ///
     /// This function flushes the internal buffer before giving access
     /// to the underlying `VirtualFile`.
-    pub async fn into_inner(mut self, ctx: &RequestContext) -> Result<VirtualFile, Error> {
+    pub async fn into_inner(mut self, ctx: &mut RequestContext) -> Result<VirtualFile, Error> {
         self.flush_buffer(ctx).await?;
         Ok(self.inner)
     }

pageserver/src/tenant/block_io.rs

@@ -92,7 +92,7 @@ impl<'a> BlockReaderRef<'a> {
     async fn read_blk(
         &self,
         blknum: u32,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<BlockLease, std::io::Error> {
         use BlockReaderRef::*;
         match self {
@@ -150,7 +150,7 @@ impl<'a> BlockCursor<'a> {
     pub async fn read_blk(
         &self,
         blknum: u32,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<BlockLease, std::io::Error> {
         self.reader.read_blk(blknum, ctx).await
     }
@@ -177,7 +177,7 @@ impl<'a> FileBlockReader<'a> {
         &self,
         buf: PageWriteGuard<'static>,
         blkno: u32,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<PageWriteGuard<'static>, std::io::Error> {
         assert!(buf.len() == PAGE_SZ);
         self.file
@@ -192,7 +192,7 @@ impl<'a> FileBlockReader<'a> {
     pub async fn read_blk<'b>(
         &self,
         blknum: u32,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<BlockLease<'b>, std::io::Error> {
         let cache = page_cache::get();
         match cache

pageserver/src/tenant/delete.rs

@@ -404,7 +404,7 @@ impl DeleteTenantFlow {
         tenant: &Arc<Tenant>,
         preload: Option<TenantPreload>,
         tenants: &'static std::sync::RwLock<TenantsMap>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), DeleteTenantError> {
         let (_, progress) = completion::channel();
 

pageserver/src/tenant/disk_btree.rs

@@ -242,7 +242,7 @@ where
     ///
     /// Read the value for given key. Returns the value, or None if it doesn't exist.
     ///
-    pub async fn get(&self, search_key: &[u8; L], ctx: &RequestContext) -> Result<Option<u64>> {
+    pub async fn get(&self, search_key: &[u8; L], ctx: &mut RequestContext) -> Result<Option<u64>> {
         let mut result: Option<u64> = None;
         self.visit(
             search_key,
@@ -278,7 +278,7 @@ where
     pub fn get_stream_from<'a>(
         &'a self,
         start_key: &'a [u8; L],
-        ctx: &'a RequestContext,
+        ctx: &'a mut RequestContext,
     ) -> impl Stream<Item = std::result::Result<(Vec<u8>, u64), DiskBtreeError>> + 'a {
         try_stream! {
             let mut stack = Vec::new();
@@ -363,7 +363,7 @@ where
         search_key: &[u8; L],
         dir: VisitDirection,
         mut visitor: V,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<bool>
     where
         V: FnMut(&[u8], u64) -> bool,

pageserver/src/tenant/ephemeral_file.rs

@@ -28,7 +28,7 @@ impl EphemeralFile {
         conf: &PageServerConf,
         tenant_shard_id: TenantShardId,
         timeline_id: TimelineId,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<EphemeralFile, io::Error> {
         static NEXT_FILENAME: AtomicU64 = AtomicU64::new(1);
         let filename_disambiguator =
@@ -68,7 +68,7 @@ impl EphemeralFile {
     pub(crate) async fn read_blk(
         &self,
         blknum: u32,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<BlockLease, io::Error> {
         self.rw.read_blk(blknum, ctx).await
     }
@@ -76,7 +76,7 @@ impl EphemeralFile {
     pub(crate) async fn write_blob(
         &mut self,
         srcbuf: &[u8],
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<u64, io::Error> {
         let pos = self.rw.bytes_written();
 

pageserver/src/tenant/ephemeral_file/page_caching.rs

@@ -38,7 +38,7 @@ impl RW {
     pub(crate) async fn write_all_borrowed(
         &mut self,
         srcbuf: &[u8],
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<usize, io::Error> {
         // It doesn't make sense to proactively fill the page cache on the Pageserver write path
         // because Compute is unlikely to access recently written data.
@@ -52,7 +52,7 @@ impl RW {
     pub(crate) async fn read_blk(
         &self,
         blknum: u32,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<BlockLease, io::Error> {
         match self.rw.read_blk(blknum).await? {
             zero_padded_read_write::ReadResult::NeedsReadFromWriter { writer } => {
@@ -138,7 +138,7 @@ impl crate::virtual_file::owned_buffers_io::write::OwnedAsyncWriter for PreWarmi
     >(
         &mut self,
         buf: B,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> std::io::Result<(usize, B::Buf)> {
         let buf = buf.slice(..);
         let saved_bounds = buf.bounds(); // save for reconstructing the Slice from iobuf after the IO is done

pageserver/src/tenant/ephemeral_file/zero_padded_read_write.rs

@@ -64,7 +64,7 @@ where
     pub async fn write_all_borrowed(
         &mut self,
         buf: &[u8],
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> std::io::Result<usize> {
         self.buffered_writer.write_buffered_borrowed(buf, ctx).await
     }

pageserver/src/tenant/layer_map.rs

@@ -850,7 +850,7 @@ impl LayerMap {
 
     /// debugging function to print out the contents of the layer map
     #[allow(unused)]
-    pub async fn dump(&self, verbose: bool, ctx: &RequestContext) -> Result<()> {
+    pub async fn dump(&self, verbose: bool, ctx: &mut RequestContext) -> Result<()> {
         println!("Begin dump LayerMap");
 
         println!("open_layer:");

pageserver/src/tenant/mgr.rs

@@ -696,7 +696,7 @@ fn tenant_spawn(
     init_order: Option<InitializationOrder>,
     tenants: &'static std::sync::RwLock<TenantsMap>,
     mode: SpawnMode,
-    ctx: &RequestContext,
+    ctx: &mut RequestContext,
 ) -> anyhow::Result<Arc<Tenant>> {
     anyhow::ensure!(
         tenant_path.is_dir(),
@@ -956,7 +956,7 @@ impl TenantManager {
         new_location_config: LocationConf,
         flush: Option<Duration>,
         mut spawn_mode: SpawnMode,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Option<Arc<Tenant>>, UpsertLocationError> {
         debug_assert_current_span_has_tenant_id();
         info!("configuring tenant location to state {new_location_config:?}");
@@ -1247,7 +1247,7 @@ impl TenantManager {
         &self,
         tenant_shard_id: TenantShardId,
         drop_cache: bool,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         let mut slot_guard = tenant_map_acquire_slot(&tenant_shard_id, TenantSlotAcquireMode::Any)?;
         let Some(old_slot) = slot_guard.get_old_value() else {
@@ -1509,7 +1509,7 @@ impl TenantManager {
         tenant: Arc<Tenant>,
         new_shard_count: ShardCount,
         new_stripe_size: Option<ShardStripeSize>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<Vec<TenantShardId>> {
         let tenant_shard_id = *tenant.get_tenant_shard_id();
         let r = self
@@ -1539,7 +1539,7 @@ impl TenantManager {
         tenant: Arc<Tenant>,
         new_shard_count: ShardCount,
         new_stripe_size: Option<ShardStripeSize>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<Vec<TenantShardId>> {
         let tenant_shard_id = *tenant.get_tenant_shard_id();
 
@@ -1994,7 +1994,7 @@ impl TenantManager {
         tenant_shard_id: TenantShardId,
         timeline_id: TimelineId,
         prepared: PreparedTimelineDetach,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Vec<TimelineId>, anyhow::Error> {
         struct RevertOnDropSlot(Option<SlotGuard>);
 
@@ -2229,7 +2229,7 @@ pub(crate) async fn load_tenant(
     broker_client: storage_broker::BrokerClientChannel,
     remote_storage: GenericRemoteStorage,
     deletion_queue_client: DeletionQueueClient,
-    ctx: &RequestContext,
+    ctx: &mut RequestContext,
 ) -> Result<(), TenantMapInsertError> {
     // This is a legacy API (replaced by `/location_conf`).  It does not support sharding
     let tenant_shard_id = TenantShardId::unsharded(tenant_id);
@@ -2837,7 +2837,7 @@ pub(crate) async fn immediate_gc(
     timeline_id: TimelineId,
     gc_req: TimelineGcRequest,
     cancel: CancellationToken,
-    ctx: &RequestContext,
+    ctx: &mut RequestContext,
 ) -> Result<GcResult, ApiError> {
     let tenant = {
         let guard = TENANTS.read().unwrap();

pageserver/src/tenant/remote_timeline_client.rs

@@ -518,7 +518,7 @@ impl RemoteTimelineClient {
         layer_metadata: &LayerFileMetadata,
         local_path: &Utf8Path,
         cancel: &CancellationToken,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<u64> {
         let downloaded_size = {
             let _unfinished_gauge_guard = self.metrics.call_begin(

pageserver/src/tenant/remote_timeline_client/download.rs

@@ -52,7 +52,7 @@ pub async fn download_layer_file<'a>(
     layer_metadata: &'a LayerFileMetadata,
     local_path: &Utf8Path,
     cancel: &CancellationToken,
-    ctx: &RequestContext,
+    ctx: &mut RequestContext,
 ) -> Result<u64, DownloadError> {
     debug_assert_current_span_has_tenant_and_timeline_id();
 
@@ -78,8 +78,33 @@ pub async fn download_layer_file<'a>(
     // If pageserver crashes the temp file will be deleted on startup and re-downloaded.
     let temp_file_path = path_with_suffix_extension(local_path, TEMP_DOWNLOAD_EXTENSION);
 
+    struct DownloadObjectClosure<'a> {
+        storage: &'a GenericRemoteStorage,
+        remote_path: &'a RemotePath,
+        temp_file_path: &'a Utf8PathBuf,
+        cancel: &'a CancellationToken,
+        ctx: &'a mut RequestContext,
+    }
+    impl backoff::Op<u64, DownloadError> for DownloadObjectClosure<'_> {
+        async fn call(&mut self) -> Result<u64, DownloadError> {
+            let DownloadObjectClosure {
+                storage,
+                remote_path,
+                temp_file_path,
+                cancel,
+                ctx,
+            } = self;
+            download_object(storage, remote_path, temp_file_path, cancel, ctx).await
+        }
+    }
     let bytes_amount = download_retry(
-        || async { download_object(storage, &remote_path, &temp_file_path, cancel, ctx).await },
+        DownloadObjectClosure {
+            storage,
+            remote_path: &remote_path,
+            temp_file_path: &temp_file_path,
+            cancel,
+            ctx,
+        },
         &format!("download {remote_path:?}"),
         cancel,
     )
@@ -107,9 +132,9 @@ pub async fn download_layer_file<'a>(
     // the in-memory state of the filesystem already has the layer file in its final place,
     // and subsequent pageserver code could think it's durable while it really isn't.
     let work = {
-        let ctx = ctx.detached_child(ctx.task_kind(), ctx.download_behavior());
+        let mut ctx = ctx.detached_child(ctx.task_kind(), ctx.download_behavior());
         async move {
-            let timeline_dir = VirtualFile::open(&timeline_path, &ctx)
+            let timeline_dir = VirtualFile::open(&timeline_path, &mut ctx)
                 .await
                 .fatal_err("VirtualFile::open for timeline dir fsync");
             timeline_dir
@@ -140,7 +165,7 @@ async fn download_object<'a>(
     src_path: &RemotePath,
     dst_path: &Utf8PathBuf,
     cancel: &CancellationToken,
-    #[cfg_attr(target_os = "macos", allow(unused_variables))] ctx: &RequestContext,
+    #[cfg_attr(target_os = "macos", allow(unused_variables))] ctx: &mut RequestContext,
 ) -> Result<u64, DownloadError> {
     let res = match crate::virtual_file::io_engine::get() {
         crate::virtual_file::io_engine::IoEngine::NotSet => panic!("unset"),
@@ -568,15 +593,11 @@ pub(crate) async fn download_initdb_tar_zst(
 /// with backoff.
 ///
 /// (See similar logic for uploads in `perform_upload_task`)
-pub(super) async fn download_retry<T, O, F>(
-    op: O,
+pub(super) async fn download_retry<T>(
+    op: impl backoff::Op<T, DownloadError>,
     description: &str,
     cancel: &CancellationToken,
-) -> Result<T, DownloadError>
-where
-    O: FnMut() -> F,
-    F: Future<Output = Result<T, DownloadError>>,
-{
+) -> Result<T, DownloadError> {
     backoff::retry(
         op,
         DownloadError::is_permanent,

pageserver/src/tenant/secondary/downloader.rs

@@ -506,7 +506,7 @@ impl<'a> TenantDownloader<'a> {
         }
     }
 
-    async fn download(&self, ctx: &RequestContext) -> Result<(), UpdateError> {
+    async fn download(&self, ctx: &mut RequestContext) -> Result<(), UpdateError> {
         debug_assert_current_span_has_tenant_id();
 
         // For the duration of a download, we must hold the SecondaryTenant::gate, to ensure
@@ -831,7 +831,7 @@ impl<'a> TenantDownloader<'a> {
         &self,
         timeline: HeatMapTimeline,
         timeline_state: SecondaryDetailTimeline,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), UpdateError> {
         debug_assert_current_span_has_tenant_and_timeline_id();
         let tenant_shard_id = self.secondary_state.get_tenant_shard_id();
@@ -978,7 +978,7 @@ impl<'a> TenantDownloader<'a> {
         tenant_shard_id: &TenantShardId,
         timeline_id: &TimelineId,
         layer: HeatMapLayer,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Option<HeatMapLayer>, UpdateError> {
         // Failpoint for simulating slow remote storage
         failpoint_support::sleep_millis_async!(

pageserver/src/tenant/size.rs

@@ -148,7 +148,7 @@ pub(super) async fn gather_inputs(
     logical_size_cache: &mut HashMap<(TimelineId, Lsn), u64>,
     cause: LogicalSizeCalculationCause,
     cancel: &CancellationToken,
-    ctx: &RequestContext,
+    ctx: &mut RequestContext,
 ) -> Result<ModelInputs, CalculateSyntheticSizeError> {
     // refresh is needed to update gc related pitr_cutoff and horizon_cutoff
     tenant.refresh_gc_info(cancel, ctx).await?;
@@ -379,7 +379,7 @@ async fn fill_logical_sizes(
     limit: &Arc<Semaphore>,
     logical_size_cache: &mut HashMap<(TimelineId, Lsn), u64>,
     cause: LogicalSizeCalculationCause,
-    ctx: &RequestContext,
+    ctx: &mut RequestContext,
 ) -> Result<(), CalculateSyntheticSizeError> {
     let timeline_hash: HashMap<TimelineId, Arc<Timeline>> = HashMap::from_iter(
         timelines

pageserver/src/tenant/storage_layer.rs

@@ -425,7 +425,7 @@ impl ReadableLayer {
         keyspace: KeySpace,
         lsn_range: Range<Lsn>,
         reconstruct_state: &mut ValuesReconstructState,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), GetVectoredError> {
         match self {
             ReadableLayer::PersistentLayer(layer) => {
@@ -574,7 +574,7 @@ impl LayerAccessStats {
         });
     }
 
-    fn record_access(&self, access_kind: LayerAccessKind, ctx: &RequestContext) {
+    fn record_access(&self, access_kind: LayerAccessKind, ctx: &mut RequestContext) {
         if ctx.access_stats_behavior() == AccessStatsBehavior::Skip {
             return;
         }

pageserver/src/tenant/storage_layer/delta_layer.rs

@@ -249,7 +249,7 @@ impl AsLayerDesc for DeltaLayer {
 }
 
 impl DeltaLayer {
-    pub(crate) async fn dump(&self, verbose: bool, ctx: &RequestContext) -> Result<()> {
+    pub(crate) async fn dump(&self, verbose: bool, ctx: &mut RequestContext) -> Result<()> {
         self.desc.dump();
 
         if !verbose {
@@ -292,7 +292,7 @@ impl DeltaLayer {
     async fn load(
         &self,
         access_kind: LayerAccessKind,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<&Arc<DeltaLayerInner>> {
         self.access_stats.record_access(access_kind, ctx);
         // Quick exit if already loaded
@@ -302,7 +302,7 @@ impl DeltaLayer {
             .with_context(|| format!("Failed to load delta layer {}", self.path()))
     }
 
-    async fn load_inner(&self, ctx: &RequestContext) -> Result<Arc<DeltaLayerInner>> {
+    async fn load_inner(&self, ctx: &mut RequestContext) -> Result<Arc<DeltaLayerInner>> {
         let path = self.path();
 
         let loaded = DeltaLayerInner::load(&path, None, None, ctx)
@@ -393,7 +393,7 @@ impl DeltaLayerWriterInner {
         tenant_shard_id: TenantShardId,
         key_start: Key,
         lsn_range: Range<Lsn>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<Self> {
         // Create the file initially with a temporary filename. We don't know
         // the end key yet, so we cannot form the final filename yet. We will
@@ -435,7 +435,7 @@ impl DeltaLayerWriterInner {
         key: Key,
         lsn: Lsn,
         val: Value,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         let (_, res) = self
             .put_value_bytes(key, lsn, Value::ser(&val)?, val.will_init(), ctx)
@@ -449,7 +449,7 @@ impl DeltaLayerWriterInner {
         lsn: Lsn,
         val: Vec<u8>,
         will_init: bool,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> (Vec<u8>, anyhow::Result<()>) {
         assert!(self.lsn_range.start <= lsn);
         let (val, res) = self.blob_writer.write_blob(val, ctx).await;
@@ -476,7 +476,7 @@ impl DeltaLayerWriterInner {
         self,
         key_end: Key,
         timeline: &Arc<Timeline>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<ResidentLayer> {
         let temp_path = self.path.clone();
         let result = self.finish0(key_end, timeline, ctx).await;
@@ -493,7 +493,7 @@ impl DeltaLayerWriterInner {
         self,
         key_end: Key,
         timeline: &Arc<Timeline>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<ResidentLayer> {
         let index_start_blk =
             ((self.blob_writer.size() + PAGE_SZ as u64 - 1) / PAGE_SZ as u64) as u32;
@@ -603,7 +603,7 @@ impl DeltaLayerWriter {
         tenant_shard_id: TenantShardId,
         key_start: Key,
         lsn_range: Range<Lsn>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<Self> {
         Ok(Self {
             inner: Some(
@@ -630,7 +630,7 @@ impl DeltaLayerWriter {
         key: Key,
         lsn: Lsn,
         val: Value,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         self.inner
             .as_mut()
@@ -645,7 +645,7 @@ impl DeltaLayerWriter {
         lsn: Lsn,
         val: Vec<u8>,
         will_init: bool,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> (Vec<u8>, anyhow::Result<()>) {
         self.inner
             .as_mut()
@@ -665,7 +665,7 @@ impl DeltaLayerWriter {
         mut self,
         key_end: Key,
         timeline: &Arc<Timeline>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<ResidentLayer> {
         self.inner
             .take()
@@ -704,7 +704,7 @@ impl DeltaLayer {
     pub async fn rewrite_summary<F>(
         path: &Utf8Path,
         rewrite: F,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), RewriteSummaryError>
     where
         F: Fn(Summary) -> Summary,
@@ -744,7 +744,7 @@ impl DeltaLayerInner {
         path: &Utf8Path,
         summary: Option<Summary>,
         max_vectored_read_bytes: Option<MaxVectoredReadBytes>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Result<Self, anyhow::Error>, anyhow::Error> {
         let file = match VirtualFile::open(path, ctx).await {
             Ok(file) => file,
@@ -793,7 +793,7 @@ impl DeltaLayerInner {
         key: Key,
         lsn_range: Range<Lsn>,
         reconstruct_state: &mut ValueReconstructState,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<ValueReconstructResult> {
         let mut need_image = true;
         // Scan the page versions backwards, starting from `lsn`.
@@ -824,13 +824,13 @@ impl DeltaLayerInner {
 
                     !blob_ref.will_init()
                 },
-                &RequestContextBuilder::extend(ctx)
+                &mut RequestContextBuilder::extend(ctx)
                     .page_content_kind(PageContentKind::DeltaLayerBtreeNode)
                     .build(),
             )
             .await?;
 
-        let ctx = &RequestContextBuilder::extend(ctx)
+        let ctx = &mut RequestContextBuilder::extend(ctx)
             .page_content_kind(PageContentKind::DeltaLayerValue)
             .build();
 
@@ -889,7 +889,7 @@ impl DeltaLayerInner {
         keyspace: KeySpace,
         lsn_range: Range<Lsn>,
         reconstruct_state: &mut ValuesReconstructState,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), GetVectoredError> {
         let block_reader = FileBlockReader::new(&self.file, self.file_id);
         let index_reader = DiskBtreeReader::<_, DELTA_KEY_SIZE>::new(
@@ -931,7 +931,7 @@ impl DeltaLayerInner {
     #[cfg(test)]
     pub(super) async fn load_key_values(
         &self,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<Vec<(Key, Lsn, Value)>> {
         let block_reader = FileBlockReader::new(&self.file, self.file_id);
         let index_reader = DiskBtreeReader::<_, DELTA_KEY_SIZE>::new(
@@ -973,12 +973,12 @@ impl DeltaLayerInner {
         index_reader: DiskBtreeReader<Reader, DELTA_KEY_SIZE>,
         mut planner: VectoredReadPlanner,
         reconstruct_state: &mut ValuesReconstructState,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<Vec<VectoredRead>>
     where
         Reader: BlockReader,
     {
-        let ctx = RequestContextBuilder::extend(ctx)
+        let mut ctx = RequestContextBuilder::extend(ctx)
             .page_content_kind(PageContentKind::DeltaLayerBtreeNode)
             .build();
 
@@ -986,7 +986,7 @@ impl DeltaLayerInner {
             let mut range_end_handled = false;
 
             let start_key = DeltaKey::from_key_lsn(&range.start, lsn_range.start);
-            let index_stream = index_reader.get_stream_from(&start_key.0, &ctx);
+            let index_stream = index_reader.get_stream_from(&start_key.0, &mut ctx);
             let mut index_stream = std::pin::pin!(index_stream);
 
             while let Some(index_entry) = index_stream.next().await {
@@ -1062,7 +1062,7 @@ impl DeltaLayerInner {
         &self,
         reads: Vec<VectoredRead>,
         reconstruct_state: &mut ValuesReconstructState,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) {
         let vectored_blob_reader = VectoredBlobReader::new(&self.file);
         let mut ignore_key_with_err = None;
@@ -1140,7 +1140,7 @@ impl DeltaLayerInner {
 
     pub(super) async fn load_keys<'a>(
         &'a self,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Vec<DeltaEntry<'a>>> {
         let block_reader = FileBlockReader::new(&self.file, self.file_id);
         let tree_reader = DiskBtreeReader::<_, DELTA_KEY_SIZE>::new(
@@ -1179,7 +1179,7 @@ impl DeltaLayerInner {
                     all_keys.push(entry);
                     true
                 },
-                &RequestContextBuilder::extend(ctx)
+                &mut RequestContextBuilder::extend(ctx)
                     .page_content_kind(PageContentKind::DeltaLayerBtreeNode)
                     .build(),
             )
@@ -1199,7 +1199,7 @@ impl DeltaLayerInner {
         &self,
         writer: &mut DeltaLayerWriter,
         until: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<usize> {
         use crate::tenant::vectored_blob_io::{
             BlobMeta, VectoredReadBuilder, VectoredReadExtended,
@@ -1387,7 +1387,7 @@ impl DeltaLayerInner {
         Ok(records)
     }
 
-    pub(super) async fn dump(&self, ctx: &RequestContext) -> anyhow::Result<()> {
+    pub(super) async fn dump(&self, ctx: &mut RequestContext) -> anyhow::Result<()> {
         println!(
             "index_start_blk: {}, root {}",
             self.index_start_blk, self.index_root_blk
@@ -1404,7 +1404,7 @@ impl DeltaLayerInner {
 
         let keys = self.load_keys(ctx).await?;
 
-        async fn dump_blob(val: &ValueRef<'_>, ctx: &RequestContext) -> anyhow::Result<String> {
+        async fn dump_blob(val: &ValueRef<'_>, ctx: &mut RequestContext) -> anyhow::Result<String> {
             let buf = val.reader.read_blob(val.blob_ref.pos(), ctx).await?;
             let val = Value::des(&buf)?;
             let desc = match val {
@@ -1513,7 +1513,7 @@ pub struct ValueRef<'a> {
 
 impl<'a> ValueRef<'a> {
     /// Loads the value from disk
-    pub async fn load(&self, ctx: &RequestContext) -> Result<Value> {
+    pub async fn load(&self, ctx: &mut RequestContext) -> Result<Value> {
         // theoretically we *could* record an access time for each, but it does not really matter
         let buf = self.reader.read_blob(self.blob_ref.pos(), ctx).await?;
         let val = Value::des(&buf)?;
@@ -1527,7 +1527,7 @@ impl<T: AsRef<DeltaLayerInner>> Adapter<T> {
     pub(crate) async fn read_blk(
         &self,
         blknum: u32,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<BlockLease, std::io::Error> {
         let block_reader = FileBlockReader::new(&self.0.as_ref().file, self.0.as_ref().file_id);
         block_reader.read_blk(blknum, ctx).await
@@ -2060,7 +2060,7 @@ mod test {
         source: &DeltaLayerInner,
         truncated: &DeltaLayerInner,
         truncated_at: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) {
         use futures::future::ready;
         use futures::stream::TryStreamExt;

pageserver/src/tenant/storage_layer/image_layer.rs

@@ -177,7 +177,7 @@ impl std::fmt::Debug for ImageLayerInner {
 }
 
 impl ImageLayerInner {
-    pub(super) async fn dump(&self, ctx: &RequestContext) -> anyhow::Result<()> {
+    pub(super) async fn dump(&self, ctx: &mut RequestContext) -> anyhow::Result<()> {
         let block_reader = FileBlockReader::new(&self.file, self.file_id);
         let tree_reader = DiskBtreeReader::<_, KEY_SIZE>::new(
             self.index_start_blk,
@@ -217,7 +217,7 @@ impl AsLayerDesc for ImageLayer {
 }
 
 impl ImageLayer {
-    pub(crate) async fn dump(&self, verbose: bool, ctx: &RequestContext) -> Result<()> {
+    pub(crate) async fn dump(&self, verbose: bool, ctx: &mut RequestContext) -> Result<()> {
         self.desc.dump();
 
         if !verbose {
@@ -254,7 +254,7 @@ impl ImageLayer {
     async fn load(
         &self,
         access_kind: LayerAccessKind,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<&ImageLayerInner> {
         self.access_stats.record_access(access_kind, ctx);
         self.inner
@@ -263,7 +263,7 @@ impl ImageLayer {
             .with_context(|| format!("Failed to load image layer {}", self.path()))
     }
 
-    async fn load_inner(&self, ctx: &RequestContext) -> Result<ImageLayerInner> {
+    async fn load_inner(&self, ctx: &mut RequestContext) -> Result<ImageLayerInner> {
         let path = self.path();
 
         let loaded = ImageLayerInner::load(&path, self.desc.image_layer_lsn(), None, None, ctx)
@@ -336,7 +336,7 @@ impl ImageLayer {
     pub async fn rewrite_summary<F>(
         path: &Utf8Path,
         rewrite: F,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), RewriteSummaryError>
     where
         F: Fn(Summary) -> Summary,
@@ -377,7 +377,7 @@ impl ImageLayerInner {
         lsn: Lsn,
         summary: Option<Summary>,
         max_vectored_read_bytes: Option<MaxVectoredReadBytes>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Result<Self, anyhow::Error>, anyhow::Error> {
         let file = match VirtualFile::open(path, ctx).await {
             Ok(file) => file,
@@ -428,7 +428,7 @@ impl ImageLayerInner {
         &self,
         key: Key,
         reconstruct_state: &mut ValueReconstructState,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<ValueReconstructResult> {
         let block_reader = FileBlockReader::new(&self.file, self.file_id);
         let tree_reader =
@@ -439,7 +439,7 @@ impl ImageLayerInner {
         if let Some(offset) = tree_reader
             .get(
                 &keybuf,
-                &RequestContextBuilder::extend(ctx)
+                &mut RequestContextBuilder::extend(ctx)
                     .page_content_kind(PageContentKind::ImageLayerBtreeNode)
                     .build(),
             )
@@ -449,7 +449,7 @@ impl ImageLayerInner {
                 .block_cursor()
                 .read_blob(
                     offset,
-                    &RequestContextBuilder::extend(ctx)
+                    &mut RequestContextBuilder::extend(ctx)
                         .page_content_kind(PageContentKind::ImageLayerValue)
                         .build(),
                 )
@@ -470,7 +470,7 @@ impl ImageLayerInner {
         &self,
         keyspace: KeySpace,
         reconstruct_state: &mut ValuesReconstructState,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), GetVectoredError> {
         let reads = self
             .plan_reads(keyspace, None, ctx)
@@ -489,7 +489,7 @@ impl ImageLayerInner {
     #[cfg(test)]
     pub(super) async fn load_key_values(
         &self,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<Vec<(Key, Lsn, Value)>> {
         let block_reader = FileBlockReader::new(&self.file, self.file_id);
         let tree_reader =
@@ -522,7 +522,7 @@ impl ImageLayerInner {
         &self,
         keyspace: KeySpace,
         shard_identity: Option<&ShardIdentity>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<Vec<VectoredRead>> {
         let mut planner = VectoredReadPlanner::new(
             self.max_vectored_read_bytes
@@ -535,7 +535,7 @@ impl ImageLayerInner {
         let tree_reader =
             DiskBtreeReader::new(self.index_start_blk, self.index_root_blk, block_reader);
 
-        let ctx = RequestContextBuilder::extend(ctx)
+        let mut ctx = RequestContextBuilder::extend(ctx)
             .page_content_kind(PageContentKind::ImageLayerBtreeNode)
             .build();
 
@@ -544,7 +544,7 @@ impl ImageLayerInner {
             let mut search_key: [u8; KEY_SIZE] = [0u8; KEY_SIZE];
             range.start.write_to_byte_slice(&mut search_key);
 
-            let index_stream = tree_reader.get_stream_from(&search_key, &ctx);
+            let index_stream = tree_reader.get_stream_from(&search_key, &mut ctx);
             let mut index_stream = std::pin::pin!(index_stream);
 
             while let Some(index_entry) = index_stream.next().await {
@@ -587,7 +587,7 @@ impl ImageLayerInner {
         &self,
         shard_identity: &ShardIdentity,
         writer: &mut ImageLayerWriter,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<usize> {
         // Fragment the range into the regions owned by this ShardIdentity
         let plan = self
@@ -629,7 +629,7 @@ impl ImageLayerInner {
         &self,
         reads: Vec<VectoredRead>,
         reconstruct_state: &mut ValuesReconstructState,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) {
         let max_vectored_read_bytes = self
             .max_vectored_read_bytes
@@ -724,7 +724,7 @@ impl ImageLayerWriterInner {
         tenant_shard_id: TenantShardId,
         key_range: &Range<Key>,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<Self> {
         // Create the file initially with a temporary filename.
         // We'll atomically rename it to the final name when we're done.
@@ -779,7 +779,7 @@ impl ImageLayerWriterInner {
         &mut self,
         key: Key,
         img: Bytes,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         ensure!(self.key_range.contains(&key));
         let (_img, res) = self.blob_writer.write_blob(img, ctx).await;
@@ -799,7 +799,7 @@ impl ImageLayerWriterInner {
     async fn finish(
         self,
         timeline: &Arc<Timeline>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<ResidentLayer> {
         let index_start_blk =
             ((self.blob_writer.size() + PAGE_SZ as u64 - 1) / PAGE_SZ as u64) as u32;
@@ -899,7 +899,7 @@ impl ImageLayerWriter {
         tenant_shard_id: TenantShardId,
         key_range: &Range<Key>,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<ImageLayerWriter> {
         Ok(Self {
             inner: Some(
@@ -918,7 +918,7 @@ impl ImageLayerWriter {
         &mut self,
         key: Key,
         img: Bytes,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         self.inner.as_mut().unwrap().put_image(key, img, ctx).await
     }
@@ -929,7 +929,7 @@ impl ImageLayerWriter {
     pub(crate) async fn finish(
         mut self,
         timeline: &Arc<Timeline>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<super::ResidentLayer> {
         self.inner.take().unwrap().finish(timeline, ctx).await
     }

pageserver/src/tenant/storage_layer/inmemory_layer.rs

@@ -256,7 +256,7 @@ impl InMemoryLayer {
     /// debugging function to print out the contents of the layer
     ///
     /// this is likely completly unused
-    pub async fn dump(&self, verbose: bool, ctx: &RequestContext) -> Result<()> {
+    pub async fn dump(&self, verbose: bool, ctx: &mut RequestContext) -> Result<()> {
         let inner = self.inner.read().await;
 
         let end_str = self.end_lsn_or_max();
@@ -308,12 +308,12 @@ impl InMemoryLayer {
         key: Key,
         lsn_range: Range<Lsn>,
         reconstruct_state: &mut ValueReconstructState,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<ValueReconstructResult> {
         ensure!(lsn_range.start >= self.start_lsn);
         let mut need_image = true;
 
-        let ctx = RequestContextBuilder::extend(ctx)
+        let mut ctx = RequestContextBuilder::extend(ctx)
             .page_content_kind(PageContentKind::InMemoryLayer)
             .build();
 
@@ -325,7 +325,7 @@ impl InMemoryLayer {
         if let Some(vec_map) = inner.index.get(&key) {
             let slice = vec_map.slice_range(lsn_range);
             for (entry_lsn, pos) in slice.iter().rev() {
-                let buf = reader.read_blob(*pos, &ctx).await?;
+                let buf = reader.read_blob(*pos, &mut ctx).await?;
                 let value = Value::des(&buf)?;
                 match value {
                     Value::Image(img) => {
@@ -365,9 +365,9 @@ impl InMemoryLayer {
         keyspace: KeySpace,
         end_lsn: Lsn,
         reconstruct_state: &mut ValuesReconstructState,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), GetVectoredError> {
-        let ctx = RequestContextBuilder::extend(ctx)
+        let mut ctx = RequestContextBuilder::extend(ctx)
             .page_content_kind(PageContentKind::InMemoryLayer)
             .build();
 
@@ -410,7 +410,7 @@ impl InMemoryLayer {
                 continue;
             }
 
-            let buf = reader.read_blob(block_read.block_offset, &ctx).await;
+            let buf = reader.read_blob(block_read.block_offset, &mut ctx).await;
             if let Err(e) = buf {
                 reconstruct_state
                     .on_key_error(block_read.key, PageReconstructError::from(anyhow!(e)));
@@ -473,7 +473,7 @@ impl InMemoryLayer {
         timeline_id: TimelineId,
         tenant_shard_id: TenantShardId,
         start_lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<InMemoryLayer> {
         trace!("initializing new empty InMemoryLayer for writing on timeline {timeline_id} at {start_lsn}");
 
@@ -512,7 +512,7 @@ impl InMemoryLayer {
         key: Key,
         lsn: Lsn,
         buf: &[u8],
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<()> {
         let mut inner = self.inner.write().await;
         self.assert_writable();
@@ -525,7 +525,7 @@ impl InMemoryLayer {
         key: Key,
         lsn: Lsn,
         buf: &[u8],
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<()> {
         trace!("put_value key {} at {}/{}", key, self.timeline_id, lsn);
 
@@ -534,7 +534,7 @@ impl InMemoryLayer {
                 .file
                 .write_blob(
                     buf,
-                    &RequestContextBuilder::extend(ctx)
+                    &mut RequestContextBuilder::extend(ctx)
                         .page_content_kind(PageContentKind::InMemoryLayer)
                         .build(),
                 )
@@ -606,7 +606,7 @@ impl InMemoryLayer {
     pub(crate) async fn write_to_disk(
         &self,
         timeline: &Arc<Timeline>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
         key_range: Option<Range<Key>>,
     ) -> Result<Option<ResidentLayer>> {
         // Grab the lock in read-mode. We hold it over the I/O, but because this

pageserver/src/tenant/storage_layer/layer.rs

@@ -331,7 +331,7 @@ impl Layer {
         key: Key,
         lsn_range: Range<Lsn>,
         reconstruct_data: &mut ValueReconstructState,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<ValueReconstructResult> {
         use anyhow::ensure;
 
@@ -361,7 +361,7 @@ impl Layer {
         keyspace: KeySpace,
         lsn_range: Range<Lsn>,
         reconstruct_data: &mut ValuesReconstructState,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), GetVectoredError> {
         let layer = self
             .0
@@ -392,7 +392,7 @@ impl Layer {
     #[cfg(test)]
     pub(crate) async fn load_key_values(
         &self,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<Vec<(Key, Lsn, crate::repository::Value)>> {
         let layer = self
             .0
@@ -479,7 +479,7 @@ impl Layer {
 
     /// Traditional debug dumping facility
     #[allow(unused)]
-    pub(crate) async fn dump(&self, verbose: bool, ctx: &RequestContext) -> anyhow::Result<()> {
+    pub(crate) async fn dump(&self, verbose: bool, ctx: &mut RequestContext) -> anyhow::Result<()> {
         self.0.desc.dump();
 
         if verbose {
@@ -898,7 +898,7 @@ impl LayerInner {
     async fn get_or_maybe_download(
         self: &Arc<Self>,
         allow_download: bool,
-        ctx: Option<&RequestContext>,
+        ctx: Option<&mut RequestContext>,
     ) -> Result<Arc<DownloadedLayer>, DownloadError> {
         let (weak, permit) = {
             // get_or_init_detached can:
@@ -988,7 +988,7 @@ impl LayerInner {
             return Err(DownloadError::NotFile(ft));
         }
 
-        if let Some(ctx) = ctx {
+        if let Some(ref ctx) = ctx {
             self.check_expected_download(ctx)?;
         }
 
@@ -1049,7 +1049,7 @@ impl LayerInner {
         self: &Arc<Self>,
         timeline: Arc<Timeline>,
         permit: heavier_once_cell::InitPermit,
-        ctx: RequestContext,
+        mut ctx: RequestContext,
     ) -> Result<Arc<DownloadedLayer>, DownloadError> {
         debug_assert_current_span_has_tenant_and_timeline_id();
 
@@ -1079,7 +1079,7 @@ impl LayerInner {
                     .await
                     .unwrap();
 
-                let res = this.download_and_init(timeline, permit, &ctx).await;
+                let res = this.download_and_init(timeline, permit, &mut ctx).await;
 
                 if let Err(res) = tx.send(res) {
                     match res {
@@ -1122,7 +1122,7 @@ impl LayerInner {
         self: &Arc<LayerInner>,
         timeline: Arc<Timeline>,
         permit: heavier_once_cell::InitPermit,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<Arc<DownloadedLayer>> {
         let result = timeline
             .remote_client
@@ -1662,9 +1662,9 @@ impl DownloadedLayer {
     async fn get<'a>(
         &'a self,
         owner: &Arc<LayerInner>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<&'a LayerKind> {
-        let init = || async {
+        let init = async {
             assert_eq!(
                 Weak::as_ptr(&self.owner),
                 Arc::as_ptr(owner),
@@ -1719,7 +1719,7 @@ impl DownloadedLayer {
             }
         };
         self.kind
-            .get_or_try_init(init)
+            .get_or_try_init(move || init)
             // return transient errors using `?`
             .await?
             .as_ref()
@@ -1736,7 +1736,7 @@ impl DownloadedLayer {
         lsn_range: Range<Lsn>,
         reconstruct_data: &mut ValueReconstructState,
         owner: &Arc<LayerInner>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<ValueReconstructResult> {
         use LayerKind::*;
 
@@ -1758,7 +1758,7 @@ impl DownloadedLayer {
         lsn_range: Range<Lsn>,
         reconstruct_data: &mut ValuesReconstructState,
         owner: &Arc<LayerInner>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), GetVectoredError> {
         use LayerKind::*;
 
@@ -1778,7 +1778,7 @@ impl DownloadedLayer {
     async fn load_key_values(
         &self,
         owner: &Arc<LayerInner>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<Vec<(Key, Lsn, crate::repository::Value)>> {
         use LayerKind::*;
 
@@ -1788,7 +1788,7 @@ impl DownloadedLayer {
         }
     }
 
-    async fn dump(&self, owner: &Arc<LayerInner>, ctx: &RequestContext) -> anyhow::Result<()> {
+    async fn dump(&self, owner: &Arc<LayerInner>, ctx: &mut RequestContext) -> anyhow::Result<()> {
         use LayerKind::*;
         match self.get(owner, ctx).await? {
             Delta(d) => d.dump(ctx).await?,
@@ -1837,7 +1837,7 @@ impl ResidentLayer {
     #[tracing::instrument(level = tracing::Level::DEBUG, skip_all, fields(layer=%self))]
     pub(crate) async fn load_keys<'a>(
         &'a self,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<Vec<DeltaEntry<'a>>> {
         use LayerKind::*;
 
@@ -1866,7 +1866,7 @@ impl ResidentLayer {
         &'a self,
         shard_identity: &ShardIdentity,
         writer: &mut ImageLayerWriter,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<usize> {
         use LayerKind::*;
 
@@ -1881,7 +1881,7 @@ impl ResidentLayer {
         &self,
         writer: &mut super::delta_layer::DeltaLayerWriter,
         until: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<usize> {
         use LayerKind::*;
 
@@ -1907,7 +1907,7 @@ impl ResidentLayer {
     #[cfg(test)]
     pub(crate) async fn as_delta(
         &self,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<&delta_layer::DeltaLayerInner> {
         use LayerKind::*;
         match self.downloaded.get(&self.owner.0, ctx).await? {

pageserver/src/tenant/tasks.rs

@@ -73,7 +73,7 @@ static PERMIT_GAUGES: once_cell::sync::Lazy<
 /// Cancellation safe.
 pub(crate) async fn concurrent_background_tasks_rate_limit_permit(
     loop_kind: BackgroundLoopKind,
-    _ctx: &RequestContext,
+    _ctx: &mut RequestContext,
 ) -> tokio::sync::SemaphorePermit<'static> {
     let _guard = PERMIT_GAUGES[loop_kind].guard();
 

pageserver/src/tenant/throttle.rs

@@ -157,19 +157,6 @@ where
                 .fetch_add(wait_time.as_micros() as u64, Ordering::Relaxed);
             let observation = Observation { wait_time };
             self.metric.observe_throttling(&observation);
-            match ctx.micros_spent_throttled.add(wait_time) {
-                Ok(res) => res,
-                Err(error) => {
-                    use once_cell::sync::Lazy;
-                    use utils::rate_limit::RateLimit;
-                    static WARN_RATE_LIMIT: Lazy<Mutex<RateLimit>> =
-                        Lazy::new(|| Mutex::new(RateLimit::new(Duration::from_secs(10))));
-                    let mut guard = WARN_RATE_LIMIT.lock().unwrap();
-                    guard.call(move || {
-                        warn!(error, "error adding time spent throttled; this message is logged at a global rate limit");
-                    });
-                }
-            }
             Some(wait_time)
         } else {
             None

pageserver/src/tenant/timeline.rs

@@ -101,9 +101,7 @@ use crate::{
 
 use crate::config::PageServerConf;
 use crate::keyspace::{KeyPartitioning, KeySpace};
-use crate::metrics::{
-    TimelineMetrics, MATERIALIZED_PAGE_CACHE_HIT, MATERIALIZED_PAGE_CACHE_HIT_DIRECT,
-};
+use crate::metrics::TimelineMetrics;
 use crate::pgdatadir_mapping::CalculateLogicalSizeError;
 use crate::tenant::config::TenantConfOpt;
 use pageserver_api::reltag::RelTag;
@@ -120,7 +118,6 @@ use utils::{
     simple_rcu::{Rcu, RcuReadGuard},
 };
 
-use crate::page_cache;
 use crate::repository::GcResult;
 use crate::repository::{Key, Value};
 use crate::task_mgr;
@@ -134,7 +131,7 @@ use self::layer_manager::LayerManager;
 use self::logical_size::LogicalSize;
 use self::walreceiver::{WalReceiver, WalReceiverConf};
 
-use super::{config::TenantConf, storage_layer::VectoredValueReconstructState};
+use super::config::TenantConf;
 use super::{debug_assert_current_span_has_tenant_and_timeline_id, AttachedTenantConf};
 use super::{remote_timeline_client::index::IndexPart, storage_layer::LayerFringe};
 use super::{remote_timeline_client::RemoteTimelineClient, storage_layer::ReadableLayer};
@@ -874,7 +871,7 @@ impl Timeline {
         &self,
         key: Key,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Bytes, PageReconstructError> {
         if !lsn.is_valid() {
             return Err(PageReconstructError::Other(anyhow::anyhow!("Invalid LSN")));
@@ -885,34 +882,11 @@ impl Timeline {
         // page_service.
         debug_assert!(!self.shard_identity.is_key_disposable(&key));
 
-        self.timeline_get_throttle.throttle(ctx, 1).await;
-
-        // Check the page cache. We will get back the most recent page with lsn <= `lsn`.
-        // The cached image can be returned directly if there is no WAL between the cached image
-        // and requested LSN. The cached image can also be used to reduce the amount of WAL needed
-        // for redo.
-        let cached_page_img = match self.lookup_cached_page(&key, lsn, ctx).await {
-            Some((cached_lsn, cached_img)) => {
-                match cached_lsn.cmp(&lsn) {
-                    Ordering::Less => {} // there might be WAL between cached_lsn and lsn, we need to check
-                    Ordering::Equal => {
-                        MATERIALIZED_PAGE_CACHE_HIT_DIRECT.inc();
-                        return Ok(cached_img); // exact LSN match, return the image
-                    }
-                    Ordering::Greater => {
-                        unreachable!("the returned lsn should never be after the requested lsn")
-                    }
-                }
-                Some((cached_lsn, cached_img))
-            }
-            None => None,
-        };
-
         match self.conf.get_impl {
             GetImpl::Legacy => {
                 let reconstruct_state = ValueReconstructState {
                     records: Vec::new(),
-                    img: cached_page_img,
+                    img: None,
                 };
 
                 self.get_impl(key, lsn, reconstruct_state, ctx).await
@@ -926,13 +900,6 @@ impl Timeline {
                 // entry returned above.
                 let mut reconstruct_state = ValuesReconstructState::new();
 
-                // Only add the cached image to the reconstruct state when it exists.
-                if cached_page_img.is_some() {
-                    let mut key_state = VectoredValueReconstructState::default();
-                    key_state.img = cached_page_img;
-                    reconstruct_state.keys.insert(key, Ok(key_state));
-                }
-
                 let vectored_res = self
                     .get_vectored_impl(keyspace.clone(), lsn, &mut reconstruct_state, ctx)
                     .await;
@@ -977,7 +944,7 @@ impl Timeline {
         key: Key,
         lsn: Lsn,
         mut reconstruct_state: ValueReconstructState,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Bytes, PageReconstructError> {
         // XXX: structured stats collection for layer eviction here.
         trace!(
@@ -1035,7 +1002,7 @@ impl Timeline {
         &self,
         keyspace: KeySpace,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<BTreeMap<Key, Result<Bytes, PageReconstructError>>, GetVectoredError> {
         if !lsn.is_valid() {
             return Err(GetVectoredError::InvalidLsn(lsn));
@@ -1066,12 +1033,7 @@ impl Timeline {
             .for_task_kind(ctx.task_kind())
             .map(|metric| (metric, Instant::now()));
 
-        // start counting after throttle so that throttle time
-        // is always less than observation time
-        let throttled = self
-            .timeline_get_throttle
-            .throttle(ctx, key_count as usize)
-            .await;
+        let throttled = None;
 
         let res = match self.conf.get_vectored_impl {
             GetVectoredImpl::Sequential => {
@@ -1132,7 +1094,7 @@ impl Timeline {
         &self,
         keyspace: KeySpace,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<BTreeMap<Key, Result<Bytes, PageReconstructError>>, GetVectoredError> {
         if !lsn.is_valid() {
             return Err(GetVectoredError::InvalidLsn(lsn));
@@ -1160,13 +1122,7 @@ impl Timeline {
             .for_task_kind(ctx.task_kind())
             .map(ScanLatencyOngoingRecording::start_recording);
 
-        // start counting after throttle so that throttle time
-        // is always less than observation time
-        let throttled = self
-            .timeline_get_throttle
-            // assume scan = 1 quota for now until we find a better way to process this
-            .throttle(ctx, 1)
-            .await;
+        let throttled = None;
 
         let vectored_res = self
             .get_vectored_impl(
@@ -1189,7 +1145,7 @@ impl Timeline {
         &self,
         keyspace: KeySpace,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<BTreeMap<Key, Result<Bytes, PageReconstructError>>, GetVectoredError> {
         let mut values = BTreeMap::new();
 
@@ -1248,7 +1204,7 @@ impl Timeline {
         keyspace: KeySpace,
         lsn: Lsn,
         reconstruct_state: &mut ValuesReconstructState,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<BTreeMap<Key, Result<Bytes, PageReconstructError>>, GetVectoredError> {
         let get_kind = if keyspace.total_raw_size() == 1 {
             GetKind::Singular
@@ -1305,7 +1261,7 @@ impl Timeline {
         vectored_res: &Result<BTreeMap<Key, Result<Bytes, PageReconstructError>>, GetVectoredError>,
         keyspace: KeySpace,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) {
         if keyspace.overlaps(&Key::metadata_key_range()) {
             // skip validation for metadata key range
@@ -1475,7 +1431,7 @@ impl Timeline {
         &self,
         lsn: Lsn,
         who_is_waiting: WaitLsnWaiter<'_>,
-        ctx: &RequestContext, /* Prepare for use by cancellation */
+        ctx: &mut RequestContext, /* Prepare for use by cancellation */
     ) -> Result<(), WaitLsnError> {
         let state = self.current_state();
         if self.cancel.is_cancelled() || matches!(state, TimelineState::Stopping) {
@@ -1571,7 +1527,7 @@ impl Timeline {
         &self,
         lsn: Lsn,
         length: Duration,
-        _ctx: &RequestContext,
+        _ctx: &mut RequestContext,
     ) -> anyhow::Result<LsnLease> {
         let lease = {
             let mut gc_info = self.gc_info.write().unwrap();
@@ -1744,7 +1700,7 @@ impl Timeline {
         self: &Arc<Self>,
         cancel: &CancellationToken,
         flags: EnumSet<CompactFlags>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), CompactionError> {
         // most likely the cancellation token is from background task, but in tests it could be the
         // request task as well.
@@ -1796,7 +1752,7 @@ impl Timeline {
         parent: Arc<crate::tenant::Tenant>,
         broker_client: BrokerClientChannel,
         background_jobs_can_start: Option<&completion::Barrier>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) {
         if self.tenant_shard_id.is_shard_zero() {
             // Logical size is only maintained accurately on shard zero.
@@ -1977,7 +1933,7 @@ impl Timeline {
 
     pub(crate) async fn wait_to_become_active(
         &self,
-        _ctx: &RequestContext, // Prepare for use by cancellation
+        _ctx: &mut RequestContext, // Prepare for use by cancellation
     ) -> Result<(), TimelineState> {
         let mut receiver = self.state.subscribe();
         loop {
@@ -2479,7 +2435,7 @@ impl Timeline {
     /// when the timeline is activated.
     fn launch_wal_receiver(
         self: &Arc<Self>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
         broker_client: BrokerClientChannel,
     ) {
         info!(
@@ -2709,7 +2665,7 @@ impl Timeline {
     pub(crate) fn get_current_logical_size(
         self: &Arc<Self>,
         priority: GetLogicalSizePriority,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> logical_size::CurrentLogicalSize {
         if !self.tenant_shard_id.is_shard_zero() {
             // Logical size is only accurately maintained on shard zero: when called elsewhere, for example
@@ -2776,7 +2732,7 @@ impl Timeline {
         current_size
     }
 
-    fn spawn_initial_logical_size_computation_task(self: &Arc<Self>, ctx: &RequestContext) {
+    fn spawn_initial_logical_size_computation_task(self: &Arc<Self>, ctx: &mut RequestContext) {
         let Some(initial_part_end) = self.current_logical_size.initial_part_end else {
             // nothing to do for freshly created timelines;
             assert_eq!(
@@ -2994,7 +2950,7 @@ impl Timeline {
         self: &Arc<Self>,
         lsn: Lsn,
         cause: LogicalSizeCalculationCause,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<u64, CalculateLogicalSizeError> {
         crate::span::debug_assert_current_span_has_tenant_and_timeline_id();
         // We should never be calculating logical sizes on shard !=0, because these shards do not have
@@ -3037,7 +2993,7 @@ impl Timeline {
         up_to_lsn: Lsn,
         cause: LogicalSizeCalculationCause,
         _guard: &GateGuard,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<u64, CalculateLogicalSizeError> {
         info!(
             "Calculating logical size for timeline {} at {}",
@@ -3200,7 +3156,7 @@ impl Timeline {
         key: Key,
         request_lsn: Lsn,
         reconstruct_state: &mut ValueReconstructState,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Vec<TraversalPathItem>, PageReconstructError> {
         // Start from the current timeline.
         let mut timeline_owned;
@@ -3240,7 +3196,6 @@ impl Timeline {
                 ValueReconstructResult::Continue => {
                     // If we reached an earlier cached page image, we're done.
                     if cont_lsn == cached_lsn + 1 {
-                        MATERIALIZED_PAGE_CACHE_HIT.inc_by(1);
                         return Ok(traversal_path);
                     }
                     if let Some(prev) = prev_lsn {
@@ -3402,7 +3357,7 @@ impl Timeline {
         mut keyspace: KeySpace,
         request_lsn: Lsn,
         reconstruct_state: &mut ValuesReconstructState,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), GetVectoredError> {
         let mut timeline_owned: Arc<Timeline>;
         let mut timeline = self;
@@ -3509,7 +3464,7 @@ impl Timeline {
         mut cont_lsn: Lsn,
         reconstruct_state: &mut ValuesReconstructState,
         cancel: &CancellationToken,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<TimelineVisitOutcome, GetVectoredError> {
         let mut unmapped_keyspace = keyspace.clone();
         let mut fringe = LayerFringe::new();
@@ -3614,30 +3569,10 @@ impl Timeline {
         })
     }
 
-    /// # Cancel-safety
-    ///
-    /// This method is cancellation-safe.
-    async fn lookup_cached_page(
-        &self,
-        key: &Key,
-        lsn: Lsn,
-        ctx: &RequestContext,
-    ) -> Option<(Lsn, Bytes)> {
-        let cache = page_cache::get();
-
-        // FIXME: It's pointless to check the cache for things that are not 8kB pages.
-        // We should look at the key to determine if it's a cacheable object
-        let (lsn, read_guard) = cache
-            .lookup_materialized_page(self.tenant_shard_id, self.timeline_id, key, lsn, ctx)
-            .await?;
-        let img = Bytes::from(read_guard.to_vec());
-        Some((lsn, img))
-    }
-
     async fn get_ready_ancestor_timeline(
         &self,
         ancestor: &Arc<Timeline>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Arc<Timeline>, GetReadyAncestorError> {
         // It's possible that the ancestor timeline isn't active yet, or
         // is active but hasn't yet caught up to the branch point. Wait
@@ -3705,7 +3640,7 @@ impl Timeline {
     async fn get_layer_for_write(
         &self,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<Arc<InMemoryLayer>> {
         let mut guard = self.layers.write().await;
         let layer = guard
@@ -3749,7 +3684,7 @@ impl Timeline {
     async fn flush_loop(
         self: &Arc<Self>,
         mut layer_flush_start_rx: tokio::sync::watch::Receiver<(u64, Lsn)>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) {
         info!("started flush loop");
         loop {
@@ -3908,7 +3843,7 @@ impl Timeline {
     async fn flush_frozen_layer(
         self: &Arc<Self>,
         frozen_layer: Arc<InMemoryLayer>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Lsn, FlushLayerError> {
         debug_assert_current_span_has_tenant_and_timeline_id();
 
@@ -4137,7 +4072,7 @@ impl Timeline {
         self: &Arc<Self>,
         frozen_layer: &Arc<InMemoryLayer>,
         key_range: Option<Range<Key>>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<Option<ResidentLayer>> {
         let self_clone = Arc::clone(self);
         let frozen_layer = Arc::clone(frozen_layer);
@@ -4194,7 +4129,7 @@ impl Timeline {
         lsn: Lsn,
         partition_size: u64,
         flags: EnumSet<CompactFlags>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<((KeyPartitioning, SparseKeyPartitioning), Lsn)> {
         let Ok(mut partitioning_guard) = self.partitioning.try_lock() else {
             // NB: there are two callers, one is the compaction task, of which there is only one per struct Tenant and hence Timeline.
@@ -4291,7 +4226,7 @@ impl Timeline {
         partition: &KeySpace,
         mut image_layer_writer: ImageLayerWriter,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
         img_range: Range<Key>,
         start: Key,
     ) -> Result<ImageLayerCreationOutcome, CreateImageLayersError> {
@@ -4391,7 +4326,7 @@ impl Timeline {
         partition: &KeySpace,
         mut image_layer_writer: ImageLayerWriter,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
         img_range: Range<Key>,
         mode: ImageLayerCreationMode,
         start: Key,
@@ -4475,7 +4410,7 @@ impl Timeline {
         partitioning: &KeyPartitioning,
         lsn: Lsn,
         mode: ImageLayerCreationMode,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Vec<ResidentLayer>, CreateImageLayersError> {
         let timer = self.metrics.create_images_time_histo.start_timer();
         let mut image_layers = Vec::new();
@@ -4676,7 +4611,7 @@ impl Timeline {
         self: &Arc<Timeline>,
         tenant: &crate::tenant::Tenant,
         options: detach_ancestor::Options,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<
         (
             completion::Completion,
@@ -4696,7 +4631,7 @@ impl Timeline {
         self: &Arc<Timeline>,
         tenant: &crate::tenant::Tenant,
         prepared: detach_ancestor::PreparedTimelineDetach,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Vec<TimelineId>, anyhow::Error> {
         detach_ancestor::complete(self, tenant, prepared, ctx).await
     }
@@ -4874,7 +4809,7 @@ impl Timeline {
         cutoff_horizon: Lsn,
         pitr: Duration,
         cancel: &CancellationToken,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<GcCutoffs, PageReconstructError> {
         let _timer = self
             .metrics
@@ -5280,8 +5215,6 @@ impl Timeline {
                     trace!("found {} WAL records that will init the page for {} at {}, performing WAL redo", data.records.len(), key, request_lsn);
                 };
 
-                let last_rec_lsn = data.records.last().unwrap().0;
-
                 let img = match self
                     .walredo_mgr
                     .as_ref()
@@ -5295,23 +5228,6 @@ impl Timeline {
                     Err(e) => return Err(PageReconstructError::WalRedo(e)),
                 };
 
-                if img.len() == page_cache::PAGE_SZ {
-                    let cache = page_cache::get();
-                    if let Err(e) = cache
-                        .memorize_materialized_page(
-                            self.tenant_shard_id,
-                            self.timeline_id,
-                            key,
-                            last_rec_lsn,
-                            &img,
-                        )
-                        .await
-                        .context("Materialized page memoization failed")
-                    {
-                        return Err(PageReconstructError::from(e));
-                    }
-                }
-
                 Ok(img)
             }
         }
@@ -5540,7 +5456,7 @@ impl Timeline {
         lsn: Lsn,
         mut images: Vec<(Key, Bytes)>,
         check_start_lsn: Option<Lsn>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         let last_record_lsn = self.get_last_record_lsn();
         assert!(
@@ -5584,7 +5500,7 @@ impl Timeline {
         self: &Arc<Timeline>,
         mut deltas: Vec<(Key, Lsn, Value)>,
         check_start_lsn: Option<Lsn>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         let last_record_lsn = self.get_last_record_lsn();
         deltas.sort_unstable_by(|(ka, la, _), (kb, lb, _)| (ka, la).cmp(&(kb, lb)));
@@ -5627,7 +5543,7 @@ impl Timeline {
     pub(crate) async fn inspect_image_layers(
         self: &Arc<Timeline>,
         lsn: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<Vec<(Key, Bytes)>> {
         let mut all_data = Vec::new();
         let guard = self.layers.read().await;
@@ -5735,7 +5651,7 @@ impl<'a> TimelineWriter<'a> {
         key: Key,
         lsn: Lsn,
         value: &Value,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         // Avoid doing allocations for "small" values.
         // In the regression test suite, the limit of 256 avoided allocations in 95% of cases:
@@ -5768,7 +5684,7 @@ impl<'a> TimelineWriter<'a> {
         &mut self,
         at: Lsn,
         action: OpenLayerAction,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<&Arc<InMemoryLayer>> {
         match action {
             OpenLayerAction::Roll => {
@@ -5785,7 +5701,7 @@ impl<'a> TimelineWriter<'a> {
         Ok(&self.write_guard.as_ref().unwrap().open_layer)
     }
 
-    async fn open_layer(&mut self, at: Lsn, ctx: &RequestContext) -> anyhow::Result<()> {
+    async fn open_layer(&mut self, at: Lsn, ctx: &mut RequestContext) -> anyhow::Result<()> {
         let layer = self.tl.get_layer_for_write(at, ctx).await?;
         let initial_size = layer.size().await?;
 
@@ -5871,7 +5787,7 @@ impl<'a> TimelineWriter<'a> {
     pub(crate) async fn put_batch(
         &mut self,
         batch: VecMap<Lsn, (Key, Value)>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         for (lsn, (key, val)) in batch {
             self.put(key, lsn, &val, ctx).await?
@@ -5883,7 +5799,7 @@ impl<'a> TimelineWriter<'a> {
     pub(crate) async fn delete_batch(
         &mut self,
         batch: &[(Range<Key>, Lsn)],
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         if let Some((_, lsn)) = batch.first() {
             let action = self.get_open_layer_action(*lsn, 0);

pageserver/src/tenant/timeline/compaction.rs

@@ -49,7 +49,7 @@ impl Timeline {
         self: &Arc<Self>,
         _cancel: &CancellationToken,
         flags: EnumSet<CompactFlags>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), CompactionError> {
         // High level strategy for compaction / image creation:
         //
@@ -175,7 +175,7 @@ impl Timeline {
     async fn compact_shard_ancestors(
         self: &Arc<Self>,
         rewrite_max: usize,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         let mut drop_layers = Vec::new();
         let mut layers_to_rewrite: Vec<Layer> = Vec::new();
@@ -359,7 +359,7 @@ impl Timeline {
     async fn compact_level0(
         self: &Arc<Self>,
         target_file_size: u64,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), CompactionError> {
         let CompactLevel0Phase1Result {
             new_layers,
@@ -400,7 +400,7 @@ impl Timeline {
         guard: tokio::sync::OwnedRwLockReadGuard<LayerManager>,
         mut stats: CompactLevel0Phase1StatsBuilder,
         target_file_size: u64,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<CompactLevel0Phase1Result, CompactionError> {
         stats.read_lock_held_spawn_blocking_startup_micros =
             stats.read_lock_acquisition_micros.till_now(); // set by caller
@@ -907,7 +907,7 @@ impl Timeline {
     pub(crate) async fn compact_tiered(
         self: &Arc<Self>,
         _cancel: &CancellationToken,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), CompactionError> {
         let fanout = self.get_compaction_threshold() as u64;
         let target_file_size = self.get_checkpoint_distance();
@@ -963,7 +963,7 @@ impl Timeline {
     pub(crate) async fn compact_with_gc(
         self: &Arc<Self>,
         _cancel: &CancellationToken,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), CompactionError> {
         use crate::tenant::storage_layer::ValueReconstructState;
         // Step 0: pick all delta layers + image layers below/intersect with the GC horizon.
@@ -1190,7 +1190,7 @@ impl CompactionJobExecutor for TimelineAdaptor {
         &mut self,
         key_range: &Range<Key>,
         lsn_range: &Range<Lsn>,
-        _ctx: &RequestContext,
+        _ctx: &mut RequestContext,
     ) -> anyhow::Result<Vec<OwnArc<PersistentLayerDesc>>> {
         self.flush_updates().await?;
 
@@ -1211,7 +1211,7 @@ impl CompactionJobExecutor for TimelineAdaptor {
         &mut self,
         key_range: &Range<Key>,
         lsn: Lsn,
-        _ctx: &RequestContext,
+        _ctx: &mut RequestContext,
     ) -> anyhow::Result<Vec<Range<Key>>> {
         if lsn == self.keyspace.0 {
             Ok(pageserver_compaction::helpers::intersect_keyspace(
@@ -1247,7 +1247,7 @@ impl CompactionJobExecutor for TimelineAdaptor {
         &mut self,
         lsn: Lsn,
         key_range: &Range<Key>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         Ok(self.create_image_impl(lsn, key_range, ctx).await?)
     }
@@ -1257,7 +1257,7 @@ impl CompactionJobExecutor for TimelineAdaptor {
         lsn_range: &Range<Lsn>,
         key_range: &Range<Key>,
         input_layers: &[ResidentDeltaLayer],
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         debug!("Create new layer {}..{}", lsn_range.start, lsn_range.end);
 
@@ -1329,7 +1329,7 @@ impl CompactionJobExecutor for TimelineAdaptor {
     async fn delete_layer(
         &mut self,
         layer: &OwnArc<PersistentLayerDesc>,
-        _ctx: &RequestContext,
+        _ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         self.layers_to_delete.push(layer.clone().0);
         Ok(())
@@ -1341,7 +1341,7 @@ impl TimelineAdaptor {
         &mut self,
         lsn: Lsn,
         key_range: &Range<Key>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), CreateImageLayersError> {
         let timer = self.timeline.metrics.create_images_time_histo.start_timer();
 
@@ -1468,7 +1468,7 @@ impl CompactionLayer<Key> for ResidentDeltaLayer {
 impl CompactionDeltaLayer<TimelineAdaptor> for ResidentDeltaLayer {
     type DeltaEntry<'a> = DeltaEntry<'a>;
 
-    async fn load_keys<'a>(&self, ctx: &RequestContext) -> anyhow::Result<Vec<DeltaEntry<'_>>> {
+    async fn load_keys<'a>(&self, ctx: &mut RequestContext) -> anyhow::Result<Vec<DeltaEntry<'_>>> {
         self.0.load_keys(ctx).await
     }
 }

pageserver/src/tenant/timeline/detach_ancestor.rs

@@ -87,7 +87,7 @@ pub(super) async fn prepare(
     detached: &Arc<Timeline>,
     tenant: &Tenant,
     options: Options,
-    ctx: &RequestContext,
+    ctx: &mut RequestContext,
 ) -> Result<(completion::Completion, PreparedTimelineDetach), Error> {
     use Error::*;
 
@@ -325,7 +325,7 @@ async fn upload_rewritten_layer(
     layer: &Layer,
     target: &Arc<Timeline>,
     cancel: &CancellationToken,
-    ctx: &RequestContext,
+    ctx: &mut RequestContext,
 ) -> Result<Option<Layer>, Error> {
     use Error::UploadRewritten;
     let copied = copy_lsn_prefix(end_lsn, layer, target, ctx).await?;
@@ -348,7 +348,7 @@ async fn copy_lsn_prefix(
     end_lsn: Lsn,
     layer: &Layer,
     target_timeline: &Arc<Timeline>,
-    ctx: &RequestContext,
+    ctx: &mut RequestContext,
 ) -> Result<Option<ResidentLayer>, Error> {
     use Error::{CopyDeltaPrefix, RewrittenDeltaDownloadFailed};
 
@@ -437,7 +437,7 @@ pub(super) async fn complete(
     detached: &Arc<Timeline>,
     tenant: &Tenant,
     prepared: PreparedTimelineDetach,
-    _ctx: &RequestContext,
+    _ctx: &mut RequestContext,
 ) -> Result<Vec<TimelineId>, anyhow::Error> {
     let PreparedTimelineDetach { layers } = prepared;
 

pageserver/src/tenant/timeline/eviction_task.rs

@@ -127,7 +127,7 @@ impl Timeline {
         policy: &EvictionPolicy,
         cancel: &CancellationToken,
         gate: &GateGuard,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> ControlFlow<(), Instant> {
         debug!("eviction iteration: {policy:?}");
         let start = Instant::now();
@@ -184,7 +184,7 @@ impl Timeline {
         p: &EvictionPolicyLayerAccessThreshold,
         cancel: &CancellationToken,
         gate: &GateGuard,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> ControlFlow<()> {
         let now = SystemTime::now();
 
@@ -309,7 +309,7 @@ impl Timeline {
         p: &EvictionPolicyLayerAccessThreshold,
         cancel: &CancellationToken,
         gate: &GateGuard,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> ControlFlow<()> {
         let permit = self.acquire_imitation_permit(cancel, ctx).await?;
 
@@ -320,7 +320,7 @@ impl Timeline {
     async fn acquire_imitation_permit(
         &self,
         cancel: &CancellationToken,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> ControlFlow<(), tokio::sync::SemaphorePermit<'static>> {
         let acquire_permit = crate::tenant::tasks::concurrent_background_tasks_rate_limit_permit(
             BackgroundLoopKind::Eviction,
@@ -366,7 +366,7 @@ impl Timeline {
         cancel: &CancellationToken,
         gate: &GateGuard,
         permit: tokio::sync::SemaphorePermit<'static>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> ControlFlow<()> {
         if !self.tenant_shard_id.is_shard_zero() {
             // Shards !=0 do not maintain accurate relation sizes, and do not need to calculate logical size
@@ -442,7 +442,7 @@ impl Timeline {
     async fn imitate_timeline_cached_layer_accesses(
         &self,
         guard: &GateGuard,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) {
         let lsn = self.get_last_record_lsn();
 
@@ -499,7 +499,7 @@ impl Timeline {
         &self,
         tenant: &Tenant,
         cancel: &CancellationToken,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) {
         if self.conf.metric_collection_endpoint.is_none() {
             // We don't start the consumption metrics task if this is not set in the config.

pageserver/src/tenant/timeline/layer_manager.rs

@@ -73,7 +73,7 @@ impl LayerManager {
         conf: &'static PageServerConf,
         timeline_id: TimelineId,
         tenant_shard_id: TenantShardId,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<Arc<InMemoryLayer>> {
         ensure!(lsn.is_aligned());
 

pageserver/src/tenant/timeline/uninit.rs

@@ -90,7 +90,7 @@ impl<'t> UninitializedTimeline<'t> {
         copyin_read: &mut (impl tokio::io::AsyncRead + Send + Sync + Unpin),
         base_lsn: Lsn,
         broker_client: storage_broker::BrokerClientChannel,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<Arc<Timeline>> {
         let raw_timeline = self.raw_timeline()?;
 

pageserver/src/tenant/timeline/walreceiver.rs

@@ -68,7 +68,7 @@ impl WalReceiver {
         timeline: Arc<Timeline>,
         conf: WalReceiverConf,
         mut broker_client: BrokerClientChannel,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Self {
         let tenant_shard_id = timeline.tenant_shard_id;
         let timeline_id = timeline.timeline_id;

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

@@ -59,7 +59,7 @@ pub(crate) struct Cancelled;
 pub(super) async fn connection_manager_loop_step(
     broker_client: &mut BrokerClientChannel,
     connection_manager_state: &mut ConnectionManagerState,
-    ctx: &RequestContext,
+    ctx: &mut RequestContext,
     cancel: &CancellationToken,
     manager_status: &std::sync::RwLock<Option<ConnectionManagerStatus>>,
 ) -> Result<(), Cancelled> {
@@ -523,7 +523,11 @@ impl ConnectionManagerState {
     }
 
     /// Shuts down the current connection (if any) and immediately starts another one with the given connection string.
-    async fn change_connection(&mut self, new_sk: NewWalConnectionCandidate, ctx: &RequestContext) {
+    async fn change_connection(
+        &mut self,
+        new_sk: NewWalConnectionCandidate,
+        ctx: &mut RequestContext,
+    ) {
         WALRECEIVER_SWITCHES
             .with_label_values(&[new_sk.reason.name()])
             .inc();

pageserver/src/tenant/vectored_blob_io.rs

@@ -286,7 +286,7 @@ impl<'a> VectoredBlobReader<'a> {
         &self,
         read: &VectoredRead,
         buf: BytesMut,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<VectoredBlobsBuf, std::io::Error> {
         assert!(read.size() > 0);
         assert!(

pageserver/src/virtual_file.rs

@@ -346,7 +346,7 @@ impl VirtualFile {
     /// Open a file in read-only mode. Like File::open.
     pub async fn open<P: AsRef<Utf8Path>>(
         path: P,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<VirtualFile, std::io::Error> {
         Self::open_with_options(path.as_ref(), OpenOptions::new().read(true), ctx).await
     }
@@ -355,7 +355,7 @@ impl VirtualFile {
     /// Like File::create.
     pub async fn create<P: AsRef<Utf8Path>>(
         path: P,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<VirtualFile, std::io::Error> {
         Self::open_with_options(
             path.as_ref(),
@@ -373,7 +373,7 @@ impl VirtualFile {
     pub async fn open_with_options<P: AsRef<Utf8Path>>(
         path: P,
         open_options: &OpenOptions,
-        _ctx: &RequestContext, /* TODO: carry a pointer to the metrics in the RequestContext instead of the parsing https://github.com/neondatabase/neon/issues/6107 */
+        _ctx: &mut RequestContext, /* TODO: carry a pointer to the metrics in the RequestContext instead of the parsing https://github.com/neondatabase/neon/issues/6107 */
     ) -> Result<VirtualFile, std::io::Error> {
         let path_ref = path.as_ref();
         let path_str = path_ref.to_string();
@@ -589,15 +589,13 @@ impl VirtualFile {
         &self,
         buf: B,
         offset: u64,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<B, Error>
     where
         B: IoBufMut + Send,
     {
-        let (buf, res) = read_exact_at_impl(buf, offset, None, |buf, offset| {
-            self.read_at(buf, offset, ctx)
-        })
-        .await;
+        let (buf, res) =
+            read_exact_at_impl(buf, offset, None, VirtualFileReadAt { file: self, ctx }).await;
         res.map(|()| buf)
     }
 
@@ -606,14 +604,17 @@ impl VirtualFile {
         buf: B,
         offset: u64,
         count: usize,
-        ctx: &RequestContext,
+        ctx: &mut 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)
-        })
+        let (buf, res) = read_exact_at_impl(
+            buf,
+            offset,
+            Some(count),
+            VirtualFileReadAt { file: self, ctx },
+        )
         .await;
         res.map(|()| buf)
     }
@@ -623,7 +624,7 @@ impl VirtualFile {
         &self,
         page: PageWriteGuard<'static>,
         offset: u64,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<PageWriteGuard<'static>, Error> {
         let buf = PageWriteGuardBuf {
             page,
@@ -639,7 +640,7 @@ impl VirtualFile {
         &self,
         buf: B,
         mut offset: u64,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> (B::Buf, Result<(), Error>) {
         let buf_len = buf.bytes_init();
         if buf_len == 0 {
@@ -677,7 +678,7 @@ impl VirtualFile {
     pub async fn write_all<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
         &mut self,
         buf: B,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> (B::Buf, Result<usize, Error>) {
         let nbytes = buf.bytes_init();
         if nbytes == 0 {
@@ -710,7 +711,7 @@ impl VirtualFile {
     async fn write<B: IoBuf + Send>(
         &mut self,
         buf: Slice<B>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> (Slice<B>, Result<usize, std::io::Error>) {
         let pos = self.pos;
         let (buf, res) = self.write_at(buf, pos, ctx).await;
@@ -726,7 +727,7 @@ impl VirtualFile {
         &self,
         buf: B,
         offset: u64,
-        _ctx: &RequestContext, /* TODO: use for metrics: https://github.com/neondatabase/neon/issues/6107 */
+        _ctx: &mut RequestContext, /* TODO: use for metrics: https://github.com/neondatabase/neon/issues/6107 */
     ) -> (B, Result<usize, Error>)
     where
         B: tokio_epoll_uring::BoundedBufMut + Send,
@@ -756,7 +757,7 @@ impl VirtualFile {
         &self,
         buf: Slice<B>,
         offset: u64,
-        _ctx: &RequestContext, /* TODO: use for metrics: https://github.com/neondatabase/neon/issues/6107 */
+        _ctx: &mut RequestContext, /* TODO: use for metrics: https://github.com/neondatabase/neon/issues/6107 */
     ) -> (Slice<B>, Result<usize, Error>) {
         let file_guard = match self.lock_file().await {
             Ok(file_guard) => file_guard,
@@ -780,8 +781,35 @@ impl VirtualFile {
     }
 }
 
+trait AsyncClosureThatDoesReadAt {
+    async fn call<B>(
+        &mut self,
+        buf: tokio_epoll_uring::Slice<B>,
+        offset: u64,
+    ) -> (tokio_epoll_uring::Slice<B>, std::io::Result<usize>)
+    where
+        B: IoBufMut + Send;
+}
+
+struct VirtualFileReadAt<'a> {
+    file: &'a VirtualFile,
+    ctx: &'a mut RequestContext,
+}
+impl<'a> AsyncClosureThatDoesReadAt for VirtualFileReadAt<'a> {
+    async fn call<B>(
+        &mut self,
+        buf: tokio_epoll_uring::Slice<B>,
+        offset: u64,
+    ) -> (tokio_epoll_uring::Slice<B>, std::io::Result<usize>)
+    where
+        B: IoBufMut + Send,
+    {
+        self.file.read_at(buf, offset, self.ctx).await
+    }
+}
+
 // 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<B, F, Fut>(
+pub async fn read_exact_at_impl<B, F>(
     buf: B,
     mut offset: u64,
     count: Option<usize>,
@@ -789,8 +817,7 @@ pub async fn read_exact_at_impl<B, F, Fut>(
 ) -> (B, std::io::Result<()>)
 where
     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>)>,
+    F: AsyncClosureThatDoesReadAt,
 {
     let mut buf: tokio_epoll_uring::Slice<B> = match count {
         Some(count) => {
@@ -802,8 +829,8 @@ where
     };
 
     while buf.bytes_total() != 0 {
-        let res;
-        (buf, res) = read_at(buf, offset).await;
+        let res: std::io::Result<usize>;
+        (buf, res) = read_at.call(buf, offset).await;
         match res {
             Ok(0) => break,
             Ok(n) => {
@@ -1048,7 +1075,7 @@ impl VirtualFile {
     pub(crate) async fn read_blk(
         &self,
         blknum: u32,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<crate::tenant::block_io::BlockLease<'_>, std::io::Error> {
         use crate::page_cache::PAGE_SZ;
         let buf = vec![0; PAGE_SZ];
@@ -1058,7 +1085,11 @@ impl VirtualFile {
         Ok(crate::tenant::block_io::BlockLease::Vec(buf))
     }
 
-    async fn read_to_end(&mut self, buf: &mut Vec<u8>, ctx: &RequestContext) -> Result<(), Error> {
+    async fn read_to_end(
+        &mut self,
+        buf: &mut Vec<u8>,
+        ctx: &mut RequestContext,
+    ) -> Result<(), Error> {
         let mut tmp = vec![0; 128];
         loop {
             let res;
@@ -1122,7 +1153,7 @@ impl OwnedAsyncWriter for VirtualFile {
     async fn write_all<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
         &mut self,
         buf: B,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> std::io::Result<(usize, B::Buf)> {
         let (buf, res) = VirtualFile::write_all(self, buf, ctx).await;
         res.map(move |v| (v, buf))
@@ -1208,7 +1239,7 @@ mod tests {
             &self,
             mut buf: Vec<u8>,
             offset: u64,
-            ctx: &RequestContext,
+            ctx: &mut RequestContext,
         ) -> Result<Vec<u8>, Error> {
             match self {
                 MaybeVirtualFile::VirtualFile(file) => file.read_exact_at(buf, offset, ctx).await,
@@ -1219,7 +1250,7 @@ mod tests {
             &self,
             buf: B,
             offset: u64,
-            ctx: &RequestContext,
+            ctx: &mut RequestContext,
         ) -> Result<(), Error> {
             match self {
                 MaybeVirtualFile::VirtualFile(file) => {
@@ -1244,7 +1275,7 @@ mod tests {
         async fn write_all<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
             &mut self,
             buf: B,
-            ctx: &RequestContext,
+            ctx: &mut RequestContext,
         ) -> Result<(), Error> {
             match self {
                 MaybeVirtualFile::VirtualFile(file) => {
@@ -1263,7 +1294,7 @@ mod tests {
 
         // Helper function to slurp contents of a file, starting at the current position,
         // into a string
-        async fn read_string(&mut self, ctx: &RequestContext) -> Result<String, Error> {
+        async fn read_string(&mut self, ctx: &mut RequestContext) -> Result<String, Error> {
             use std::io::Read;
             let mut buf = String::new();
             match self {
@@ -1284,7 +1315,7 @@ mod tests {
             &mut self,
             pos: u64,
             len: usize,
-            ctx: &RequestContext,
+            ctx: &mut RequestContext,
         ) -> Result<String, Error> {
             let buf = vec![0; len];
             let buf = self.read_exact_at(buf, pos, ctx).await?;
@@ -1307,7 +1338,7 @@ mod tests {
             async fn open(
                 path: Utf8PathBuf,
                 opts: OpenOptions,
-                ctx: &RequestContext,
+                ctx: &mut RequestContext,
             ) -> Result<MaybeVirtualFile, anyhow::Error> {
                 let vf = VirtualFile::open_with_options(&path, &opts, ctx).await?;
                 Ok(MaybeVirtualFile::VirtualFile(vf))
@@ -1324,7 +1355,7 @@ mod tests {
             async fn open(
                 path: Utf8PathBuf,
                 opts: OpenOptions,
-                _ctx: &RequestContext,
+                _ctx: &mut RequestContext,
             ) -> Result<MaybeVirtualFile, anyhow::Error> {
                 Ok(MaybeVirtualFile::File({
                     let owned_fd = opts.open(path.as_std_path()).await?;
@@ -1343,7 +1374,7 @@ mod tests {
         async fn open(
             path: Utf8PathBuf,
             opts: OpenOptions,
-            ctx: &RequestContext,
+            ctx: &mut RequestContext,
         ) -> Result<MaybeVirtualFile, anyhow::Error>;
     }
 

pageserver/src/virtual_file/owned_buffers_io/util/size_tracking_writer.rs

@@ -38,7 +38,7 @@ where
     async fn write_all<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
         &mut self,
         buf: B,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> std::io::Result<(usize, B::Buf)> {
         let (nwritten, buf) = self.dst.write_all(buf, ctx).await?;
         self.bytes_amount += u64::try_from(nwritten).unwrap();

pageserver/src/virtual_file/owned_buffers_io/write.rs

@@ -9,7 +9,7 @@ pub trait OwnedAsyncWriter {
     async fn write_all<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
         &mut self,
         buf: B,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> std::io::Result<(usize, B::Buf)>;
 }
 
@@ -60,7 +60,7 @@ where
     }
 
     #[cfg_attr(target_os = "macos", allow(dead_code))]
-    pub async fn flush_and_into_inner(mut self, ctx: &RequestContext) -> std::io::Result<W> {
+    pub async fn flush_and_into_inner(mut self, ctx: &mut RequestContext) -> std::io::Result<W> {
         self.flush(ctx).await?;
 
         let Self { buf, writer } = self;
@@ -79,7 +79,7 @@ where
     pub async fn write_buffered<S: IoBuf + Send>(
         &mut self,
         chunk: Slice<S>,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> std::io::Result<(usize, S)> {
         let chunk_len = chunk.len();
         // avoid memcpy for the middle of the chunk
@@ -124,7 +124,7 @@ where
     pub async fn write_buffered_borrowed(
         &mut self,
         mut chunk: &[u8],
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> std::io::Result<usize> {
         let chunk_len = chunk.len();
         while !chunk.is_empty() {
@@ -142,7 +142,7 @@ where
         Ok(chunk_len)
     }
 
-    async fn flush(&mut self, ctx: &RequestContext) -> std::io::Result<()> {
+    async fn flush(&mut self, ctx: &mut RequestContext) -> std::io::Result<()> {
         let buf = self.buf.take().expect("must not use after an error");
         let buf_len = buf.pending();
         if buf_len == 0 {
@@ -215,7 +215,7 @@ impl OwnedAsyncWriter for Vec<u8> {
     async fn write_all<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
         &mut self,
         buf: B,
-        _: &RequestContext,
+        _: &mut RequestContext,
     ) -> std::io::Result<(usize, B::Buf)> {
         let nbytes = buf.bytes_init();
         if nbytes == 0 {
@@ -243,7 +243,7 @@ mod tests {
         async fn write_all<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
             &mut self,
             buf: B,
-            _: &RequestContext,
+            _: &mut RequestContext,
         ) -> std::io::Result<(usize, B::Buf)> {
             let nbytes = buf.bytes_init();
             if nbytes == 0 {

pageserver/src/walingest.rs

@@ -59,7 +59,7 @@ impl WalIngest {
     pub async fn new(
         timeline: &Timeline,
         startpoint: Lsn,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<WalIngest> {
         // Fetch the latest checkpoint into memory, so that we can compare with it
         // quickly in `ingest_record` and update it when it changes.
@@ -90,7 +90,7 @@ impl WalIngest {
         lsn: Lsn,
         modification: &mut DatadirModification<'_>,
         decoded: &mut DecodedWALRecord,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<bool> {
         WAL_INGEST.records_received.inc();
         let pg_version = modification.tline.pg_version;
@@ -449,7 +449,7 @@ impl WalIngest {
         &mut self,
         modification: &mut DatadirModification<'_>,
         blk: &DecodedBkpBlock,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), PageReconstructError> {
         let rel = RelTag {
             spcnode: blk.rnode_spcnode,
@@ -467,7 +467,7 @@ impl WalIngest {
         lsn: Lsn,
         decoded: &DecodedWALRecord,
         blk: &DecodedBkpBlock,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), PageReconstructError> {
         let rel = RelTag {
             spcnode: blk.rnode_spcnode,
@@ -530,7 +530,7 @@ impl WalIngest {
         buf: &mut Bytes,
         modification: &mut DatadirModification<'_>,
         decoded: &DecodedWALRecord,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         // Handle VM bit updates that are implicitly part of heap records.
 
@@ -836,7 +836,7 @@ impl WalIngest {
         buf: &mut Bytes,
         modification: &mut DatadirModification<'_>,
         decoded: &DecodedWALRecord,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         // Handle VM bit updates that are implicitly part of heap records.
 
@@ -1007,7 +1007,7 @@ impl WalIngest {
         &mut self,
         modification: &mut DatadirModification<'_>,
         rec: &XlCreateDatabase,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         let db_id = rec.db_id;
         let tablespace_id = rec.tablespace_id;
@@ -1102,7 +1102,7 @@ impl WalIngest {
         &mut self,
         modification: &mut DatadirModification<'_>,
         rec: &XlSmgrCreate,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         let rel = RelTag {
             spcnode: rec.rnode.spcnode,
@@ -1121,7 +1121,7 @@ impl WalIngest {
         &mut self,
         modification: &mut DatadirModification<'_>,
         rec: &XlSmgrTruncate,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         let spcnode = rec.rnode.spcnode;
         let dbnode = rec.rnode.dbnode;
@@ -1193,7 +1193,7 @@ impl WalIngest {
         parsed: &XlXactParsedRecord,
         is_commit: bool,
         origin_id: u16,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         // Record update of CLOG pages
         let mut pageno = parsed.xid / pg_constants::CLOG_XACTS_PER_PAGE;
@@ -1270,7 +1270,7 @@ impl WalIngest {
         &mut self,
         modification: &mut DatadirModification<'_>,
         xlrec: &XlClogTruncate,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         info!(
             "RM_CLOG_ID truncate pageno {} oldestXid {} oldestXidDB {}",
@@ -1416,7 +1416,7 @@ impl WalIngest {
         &mut self,
         modification: &mut DatadirModification<'_>,
         xlrec: &XlMultiXactTruncate,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<()> {
         self.checkpoint.oldestMulti = xlrec.end_trunc_off;
         self.checkpoint.oldestMultiDB = xlrec.oldest_multi_db;
@@ -1454,7 +1454,7 @@ impl WalIngest {
         modification: &mut DatadirModification<'_>,
         xlrec: &XlRelmapUpdate,
         decoded: &DecodedWALRecord,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<()> {
         let mut buf = decoded.record.clone();
         buf.advance(decoded.main_data_offset);
@@ -1475,7 +1475,7 @@ impl WalIngest {
         &mut self,
         modification: &mut DatadirModification<'_>,
         rel: RelTag,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<()> {
         modification.put_rel_creation(rel, 0, ctx).await?;
         Ok(())
@@ -1487,7 +1487,7 @@ impl WalIngest {
         rel: RelTag,
         blknum: BlockNumber,
         img: Bytes,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), PageReconstructError> {
         self.handle_rel_extend(modification, rel, blknum, ctx)
             .await?;
@@ -1501,7 +1501,7 @@ impl WalIngest {
         rel: RelTag,
         blknum: BlockNumber,
         rec: NeonWalRecord,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<()> {
         self.handle_rel_extend(modification, rel, blknum, ctx)
             .await?;
@@ -1514,7 +1514,7 @@ impl WalIngest {
         modification: &mut DatadirModification<'_>,
         rel: RelTag,
         nblocks: BlockNumber,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         modification.put_rel_truncation(rel, nblocks, ctx).await?;
         Ok(())
@@ -1524,7 +1524,7 @@ impl WalIngest {
         &mut self,
         modification: &mut DatadirModification<'_>,
         rel: RelTag,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<()> {
         modification.put_rel_drop(rel, ctx).await?;
         Ok(())
@@ -1535,7 +1535,7 @@ impl WalIngest {
         modification: &mut DatadirModification<'_>,
         rel: RelTag,
         blknum: BlockNumber,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<(), PageReconstructError> {
         let new_nblocks = blknum + 1;
         // Check if the relation exists. We implicitly create relations on first
@@ -1597,7 +1597,7 @@ impl WalIngest {
         segno: u32,
         blknum: BlockNumber,
         img: Bytes,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> Result<()> {
         self.handle_slru_extend(modification, kind, segno, blknum, ctx)
             .await?;
@@ -1611,7 +1611,7 @@ impl WalIngest {
         kind: SlruKind,
         segno: u32,
         blknum: BlockNumber,
-        ctx: &RequestContext,
+        ctx: &mut RequestContext,
     ) -> anyhow::Result<()> {
         // we don't use a cache for this like we do for relations. SLRUS are explcitly
         // extended with ZEROPAGE records, not with commit records, so it happens
@@ -1660,7 +1660,7 @@ impl WalIngest {
 async fn get_relsize(
     modification: &DatadirModification<'_>,
     rel: RelTag,
-    ctx: &RequestContext,
+    ctx: &mut RequestContext,
 ) -> anyhow::Result<BlockNumber> {
     let nblocks = if !modification
         .tline
@@ -1701,7 +1701,7 @@ mod tests {
 
     static ZERO_CHECKPOINT: Bytes = Bytes::from_static(&[0u8; SIZEOF_CHECKPOINT]);
 
-    async fn init_walingest_test(tline: &Timeline, ctx: &RequestContext) -> Result<WalIngest> {
+    async fn init_walingest_test(tline: &Timeline, ctx: &mut RequestContext) -> Result<WalIngest> {
         let mut m = tline.begin_modification(Lsn(0x10));
         m.put_checkpoint(ZERO_CHECKPOINT.clone())?;
         m.put_relmap_file(0, 111, Bytes::from(""), ctx).await?; // dummy relmapper file

storage_controller/src/scheduler.rs

@@ -391,7 +391,7 @@ impl Scheduler {
             return Err(ScheduleError::NoPageservers);
         }
 
-        let mut scores: Vec<(NodeId, AffinityScore, usize)> = self
+        let mut scores: Vec<(NodeId, AffinityScore, usize, usize)> = self
             .nodes
             .iter()
             .filter_map(|(k, v)| {
@@ -402,6 +402,7 @@ impl Scheduler {
                         *k,
                         context.nodes.get(k).copied().unwrap_or(AffinityScore::FREE),
                         v.shard_count,
+                        v.attached_shard_count,
                     ))
                 }
             })
@@ -409,9 +410,12 @@ impl Scheduler {
 
         // Sort by, in order of precedence:
         //  1st: Affinity score.  We should never pick a higher-score node if a lower-score node is available
-        //  2nd: Utilization.  Within nodes with the same affinity, use the least loaded nodes.
-        //  3rd: Node ID.  This is a convenience to make selection deterministic in tests and empty systems.
-        scores.sort_by_key(|i| (i.1, i.2, i.0));
+        //  2nd: Attached shard count.  Within nodes with the same affinity, we always pick the node with
+        //  the least number of attached shards.
+        //  3rd: Total shard count.  Within nodes with the same affinity and attached shard count, use nodes
+        //  with the lower total shard count.
+        //  4th: Node ID.  This is a convenience to make selection deterministic in tests and empty systems.
+        scores.sort_by_key(|i| (i.1, i.3, i.2, i.0));
 
         if scores.is_empty() {
             // After applying constraints, no pageservers were left.

storage_controller/src/tenant_shard.rs

@@ -1632,14 +1632,10 @@ pub(crate) mod tests {
 
         // We should see equal number of locations on the two nodes.
         assert_eq!(scheduler.get_node_shard_count(NodeId(1)), 4);
-        // Scheduling does not consider the number of attachments picking the initial
-        // pageserver to attach to (hence the assertion that all primaries are on the
-        // same node)
-        // TODO: Tweak the scheduling to evenly distribute attachments for new shards.
-        assert_eq!(scheduler.get_node_attached_shard_count(NodeId(1)), 4);
+        assert_eq!(scheduler.get_node_attached_shard_count(NodeId(1)), 2);
 
         assert_eq!(scheduler.get_node_shard_count(NodeId(2)), 4);
-        assert_eq!(scheduler.get_node_attached_shard_count(NodeId(2)), 0);
+        assert_eq!(scheduler.get_node_attached_shard_count(NodeId(2)), 2);
 
         // Add another two nodes: we should see the shards spread out when their optimize
         // methods are called

test_runner/fixtures/metrics.py

@@ -118,8 +118,6 @@ PAGESERVER_GLOBAL_METRICS: Tuple[str, ...] = (
     "libmetrics_launch_timestamp",
     "libmetrics_build_info",
     "libmetrics_tracing_event_count_total",
-    "pageserver_materialized_cache_hits_total",
-    "pageserver_materialized_cache_hits_direct_total",
     "pageserver_page_cache_read_hits_total",
     "pageserver_page_cache_read_accesses_total",
     "pageserver_page_cache_size_current_bytes",

test_runner/fixtures/neon_fixtures.py

@@ -3446,11 +3446,12 @@ class Endpoint(PgProtocol, LogUtils):
         self.active_safekeepers: List[int] = list(map(lambda sk: sk.id, env.safekeepers))
         # path to conf is <repo_dir>/endpoints/<endpoint_id>/pgdata/postgresql.conf
 
-        # This lock prevents concurrent start & stop operations, keeping `self.running` consistent
-        # with whether we're really running.  Tests generally wouldn't try and do these concurrently,
-        # but endpoints are also stopped during test teardown, which might happen concurrently with
-        # destruction of objects in tests.
-        self.lock = threading.Lock()
+        # Semaphore is set to 1 when we start, and acquire'd back to zero when we stop
+        #
+        # We use a semaphore rather than a bool so that racing calls to stop() don't
+        # try and stop the same process twice, as stop() is called by test teardown and
+        # potentially by some __del__ chains in other threads.
+        self._running = threading.Semaphore(0)
 
     def http_client(
         self, auth_token: Optional[str] = None, retries: Optional[Retry] = None
@@ -3522,15 +3523,14 @@ class Endpoint(PgProtocol, LogUtils):
 
         log.info(f"Starting postgres endpoint {self.endpoint_id}")
 
-        with self.lock:
-            self.env.neon_cli.endpoint_start(
-                self.endpoint_id,
-                safekeepers=self.active_safekeepers,
-                remote_ext_config=remote_ext_config,
-                pageserver_id=pageserver_id,
-                allow_multiple=allow_multiple,
-            )
-            self.running = True
+        self.env.neon_cli.endpoint_start(
+            self.endpoint_id,
+            safekeepers=self.active_safekeepers,
+            remote_ext_config=remote_ext_config,
+            pageserver_id=pageserver_id,
+            allow_multiple=allow_multiple,
+        )
+        self._running.release(1)
 
         return self
 
@@ -3578,9 +3578,12 @@ class Endpoint(PgProtocol, LogUtils):
             conf_file.write("\n".join(hba) + "\n")
             conf_file.write(data)
 
-        if self.running:
+        if self.is_running():
             self.safe_psql("SELECT pg_reload_conf()")
 
+    def is_running(self):
+        return self._running._value > 0
+
     def reconfigure(self, pageserver_id: Optional[int] = None):
         assert self.endpoint_id is not None
         self.env.neon_cli.endpoint_reconfigure(self.endpoint_id, self.tenant_id, pageserver_id)
@@ -3629,13 +3632,12 @@ class Endpoint(PgProtocol, LogUtils):
         Returns self.
         """
 
-        with self.lock:
-            if self.running:
-                assert self.endpoint_id is not None
-                self.env.neon_cli.endpoint_stop(
-                    self.endpoint_id, check_return_code=self.check_stop_result, mode=mode
-                )
-                self.running = False
+        running = self._running.acquire(blocking=False)
+        if running:
+            assert self.endpoint_id is not None
+            self.env.neon_cli.endpoint_stop(
+                self.endpoint_id, check_return_code=self.check_stop_result, mode=mode
+            )
 
         return self
 
@@ -3645,13 +3647,13 @@ class Endpoint(PgProtocol, LogUtils):
         Returns self.
         """
 
-        with self.lock:
+        running = self._running.acquire(blocking=False)
+        if running:
             assert self.endpoint_id is not None
             self.env.neon_cli.endpoint_stop(
                 self.endpoint_id, True, check_return_code=self.check_stop_result, mode=mode
             )
             self.endpoint_id = None
-            self.running = False
 
         return self
 

test_runner/regress/test_local_file_cache.py

@@ -1,4 +1,5 @@
 import os
+import queue
 import random
 import threading
 import time
@@ -8,11 +9,7 @@ from fixtures.neon_fixtures import DEFAULT_BRANCH_NAME, NeonEnvBuilder
 from fixtures.utils import query_scalar
 
 
-def test_local_file_cache_unlink(neon_env_builder: NeonEnvBuilder, build_type: str):
-    if build_type == "debug":
-        # Disable vectored read path cross validation since it makes the test time out.
-        neon_env_builder.pageserver_config_override = "validate_vectored_get=false"
-
+def test_local_file_cache_unlink(neon_env_builder: NeonEnvBuilder):
     env = neon_env_builder.init_start()
 
     cache_dir = os.path.join(env.repo_dir, "file_cache")
@@ -33,11 +30,10 @@ def test_local_file_cache_unlink(neon_env_builder: NeonEnvBuilder, build_type: s
 
     cur = endpoint.connect().cursor()
 
+    stop = threading.Event()
     n_rows = 100000
     n_threads = 20
-    n_updates_per_thread = 10000
     n_updates_per_connection = 1000
-    n_total_updates = n_threads * n_updates_per_thread
 
     cur.execute("CREATE TABLE lfctest (id int4 PRIMARY KEY, n int) WITH (fillfactor=10)")
     cur.execute(f"INSERT INTO lfctest SELECT g, 1 FROM generate_series(1, {n_rows}) g")
@@ -48,11 +44,11 @@ def test_local_file_cache_unlink(neon_env_builder: NeonEnvBuilder, build_type: s
     # performed (plus the initial 1 on each row).
     #
     # Furthermore, each thread will reconnect between every 1000 updates.
-    def run_updates():
+    def run_updates(n_updates_performed_q: queue.Queue[int]):
         n_updates_performed = 0
         conn = endpoint.connect()
         cur = conn.cursor()
-        for _ in range(n_updates_per_thread):
+        while not stop.is_set():
             id = random.randint(1, n_rows)
             cur.execute(f"UPDATE lfctest SET n = n + 1 WHERE id = {id}")
             n_updates_performed += 1
@@ -61,19 +57,28 @@ def test_local_file_cache_unlink(neon_env_builder: NeonEnvBuilder, build_type: s
                 conn.close()
                 conn = endpoint.connect()
                 cur = conn.cursor()
+        n_updates_performed_q.put(n_updates_performed)
 
+    n_updates_performed_q: queue.Queue[int] = queue.Queue()
     threads: List[threading.Thread] = []
     for _i in range(n_threads):
-        thread = threading.Thread(target=run_updates, args=(), daemon=True)
+        thread = threading.Thread(target=run_updates, args=(n_updates_performed_q,), daemon=True)
         thread.start()
         threads.append(thread)
 
     time.sleep(5)
 
+    # unlink, this is what we're actually testing
     new_cache_dir = os.path.join(env.repo_dir, "file_cache_new")
     os.rename(cache_dir, new_cache_dir)
 
+    time.sleep(10)
+
+    stop.set()
+
+    n_updates_performed = 0
     for thread in threads:
         thread.join()
+        n_updates_performed += n_updates_performed_q.get()
 
-    assert query_scalar(cur, "SELECT SUM(n) FROM lfctest") == n_total_updates + n_rows
+    assert query_scalar(cur, "SELECT SUM(n) FROM lfctest") == n_rows + n_updates_performed

test_runner/regress/test_pg_regress.py

@@ -23,11 +23,11 @@ if TYPE_CHECKING:
 
 # Run the main PostgreSQL regression tests, in src/test/regress.
 #
+@pytest.mark.timeout(600)
 @pytest.mark.parametrize("shard_count", [None, 4])
 def test_pg_regress(
     neon_env_builder: NeonEnvBuilder,
     test_output_dir: Path,
-    build_type: str,
     pg_bin: PgBin,
     capsys: CaptureFixture[str],
     base_dir: Path,
@@ -43,10 +43,6 @@ def test_pg_regress(
     if shard_count is not None:
         neon_env_builder.num_pageservers = shard_count
 
-    if build_type == "debug":
-        # Disable vectored read path cross validation since it makes the test time out.
-        neon_env_builder.pageserver_config_override = "validate_vectored_get=false"
-
     neon_env_builder.enable_pageserver_remote_storage(s3_storage())
     neon_env_builder.enable_scrub_on_exit()
     env = neon_env_builder.init_start(initial_tenant_shard_count=shard_count)

test_runner/regress/test_physical_replication.py

@@ -6,7 +6,6 @@ from fixtures.neon_fixtures import NeonEnv
 
 def test_physical_replication(neon_simple_env: NeonEnv):
     env = neon_simple_env
-    n_records = 100000
     with env.endpoints.create_start(
         branch_name="main",
         endpoint_id="primary",
@@ -22,8 +21,20 @@ def test_physical_replication(neon_simple_env: NeonEnv):
                 with p_con.cursor() as p_cur:
                     with secondary.connect() as s_con:
                         with s_con.cursor() as s_cur:
-                            for pk in range(n_records):
+                            runtime_secs = 30
+                            started_at = time.time()
+                            pk = 0
+                            while True:
+                                pk += 1
+                                now = time.time()
+                                if now - started_at > runtime_secs:
+                                    break
                                 p_cur.execute("insert into t (pk) values (%s)", (pk,))
+                                # an earlier version of this test was based on a fixed number of loop iterations
+                                # and selected for pk=(random.randrange(1, fixed number of loop iterations)).
+                                # => the probability of selection for a value that was never inserted changed from 99.9999% to 0% over the course of the test.
+                                #
+                                # We changed the test to where=(random.randrange(1, 2*pk)), which means the probability is now fixed to 50%.
                                 s_cur.execute(
-                                    "select * from t where pk=%s", (random.randrange(1, n_records),)
+                                    "select * from t where pk=%s", (random.randrange(1, 2 * pk),)
                                 )

test_runner/regress/test_storage_controller.py

@@ -1527,13 +1527,7 @@ def test_graceful_cluster_restart(neon_env_builder: NeonEnvBuilder):
         )
 
     # Give things a chance to settle.
-    # A call to `reconcile_until_idle` could be used here instead,
-    # however since all attachments are placed on the same node,
-    # we'd have to wait for a long time (2 minutes-ish) for optimizations
-    # to quiesce.
-    # TODO: once the initial attachment selection is fixed, update this
-    # to use `reconcile_until_idle`.
-    time.sleep(2)
+    env.storage_controller.reconcile_until_idle(timeout_secs=30)
 
     nodes = env.storage_controller.node_list()
     assert len(nodes) == 2