TEST ONLY Use branch fix_path_for_extension_request for vendor/postgres-v16

.github/workflows/build_and_test.yml

@@ -159,10 +159,6 @@ jobs:
       # This will catch compiler & clippy warnings in all feature combinations.
       # TODO: use cargo hack for build and test as well, but, that's quite expensive.
       # NB: keep clippy args in sync with ./run_clippy.sh
-      #
-      # The only difference between "clippy --debug" and "clippy --release" is that in --release mode,
-      # #[cfg(debug_assertions)] blocks are not built. It's not worth building everything for second
-      # time just for that, so skip "clippy --release".
       - run: |
           CLIPPY_COMMON_ARGS="$( source .neon_clippy_args; echo "$CLIPPY_COMMON_ARGS")"
           if [ "$CLIPPY_COMMON_ARGS" = "" ]; then
@@ -172,6 +168,8 @@ jobs:
           echo "CLIPPY_COMMON_ARGS=${CLIPPY_COMMON_ARGS}" >> $GITHUB_ENV
       - name: Run cargo clippy (debug)
         run: cargo hack --feature-powerset clippy $CLIPPY_COMMON_ARGS
+      - name: Run cargo clippy (release)
+        run: cargo hack --feature-powerset clippy --release $CLIPPY_COMMON_ARGS
 
       - name: Check documentation generation
         run: cargo doc --workspace --no-deps --document-private-items

libs/pageserver_api/src/config.rs

@@ -64,7 +64,6 @@ pub struct ConfigToml {
     #[serde(with = "humantime_serde")]
     pub wal_redo_timeout: Duration,
     pub superuser: String,
-    pub initdb_cache_dir: Option<Utf8PathBuf>,
     pub page_cache_size: usize,
     pub max_file_descriptors: usize,
     pub pg_distrib_dir: Option<Utf8PathBuf>,
@@ -174,6 +173,40 @@ impl Default for EvictionOrder {
     }
 }
 
+#[derive(
+    Eq,
+    PartialEq,
+    Debug,
+    Copy,
+    Clone,
+    strum_macros::EnumString,
+    strum_macros::Display,
+    serde_with::DeserializeFromStr,
+    serde_with::SerializeDisplay,
+)]
+#[strum(serialize_all = "kebab-case")]
+pub enum GetVectoredImpl {
+    Sequential,
+    Vectored,
+}
+
+#[derive(
+    Eq,
+    PartialEq,
+    Debug,
+    Copy,
+    Clone,
+    strum_macros::EnumString,
+    strum_macros::Display,
+    serde_with::DeserializeFromStr,
+    serde_with::SerializeDisplay,
+)]
+#[strum(serialize_all = "kebab-case")]
+pub enum GetImpl {
+    Legacy,
+    Vectored,
+}
+
 #[derive(Copy, Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 #[serde(transparent)]
 pub struct MaxVectoredReadBytes(pub NonZeroUsize);
@@ -305,6 +338,8 @@ pub mod defaults {
     pub const DEFAULT_IMAGE_COMPRESSION: ImageCompressionAlgorithm =
         ImageCompressionAlgorithm::Zstd { level: Some(1) };
 
+    pub const DEFAULT_VALIDATE_VECTORED_GET: bool = false;
+
     pub const DEFAULT_EPHEMERAL_BYTES_PER_MEMORY_KB: usize = 0;
 
     pub const DEFAULT_IO_BUFFER_ALIGNMENT: usize = 512;
@@ -323,7 +358,6 @@ impl Default for ConfigToml {
             wal_redo_timeout: (humantime::parse_duration(DEFAULT_WAL_REDO_TIMEOUT)
                 .expect("cannot parse default wal redo timeout")),
             superuser: (DEFAULT_SUPERUSER.to_string()),
-            initdb_cache_dir: None,
             page_cache_size: (DEFAULT_PAGE_CACHE_SIZE),
             max_file_descriptors: (DEFAULT_MAX_FILE_DESCRIPTORS),
             pg_distrib_dir: None, // Utf8PathBuf::from("./pg_install"), // TODO: formely, this was std::env::current_dir()
@@ -342,10 +376,7 @@ impl Default for ConfigToml {
 
             concurrent_tenant_warmup: (NonZeroUsize::new(DEFAULT_CONCURRENT_TENANT_WARMUP)
                 .expect("Invalid default constant")),
-            concurrent_tenant_size_logical_size_queries: NonZeroUsize::new(
+            concurrent_tenant_size_logical_size_queries: NonZeroUsize::new(1).unwrap(),
-                DEFAULT_CONCURRENT_TENANT_SIZE_LOGICAL_SIZE_QUERIES,
-            )
-            .unwrap(),
             metric_collection_interval: (humantime::parse_duration(
                 DEFAULT_METRIC_COLLECTION_INTERVAL,
             )
@@ -436,6 +467,8 @@ pub mod tenant_conf_defaults {
     // By default ingest enough WAL for two new L0 layers before checking if new image
     // image layers should be created.
     pub const DEFAULT_IMAGE_LAYER_CREATION_CHECK_THRESHOLD: u8 = 2;
+
+    pub const DEFAULT_INGEST_BATCH_SIZE: u64 = 100;
 }
 
 impl Default for TenantConfigToml {

libs/postgres_ffi/src/xlog_utils.rs

@@ -26,7 +26,6 @@ use bytes::{Buf, Bytes};
 use log::*;
 
 use serde::Serialize;
-use std::ffi::OsStr;
 use std::fs::File;
 use std::io::prelude::*;
 use std::io::ErrorKind;
@@ -79,34 +78,19 @@ pub fn XLogFileName(tli: TimeLineID, logSegNo: XLogSegNo, wal_segsz_bytes: usize
     )
 }
 
-pub fn XLogFromFileName(
+pub fn XLogFromFileName(fname: &str, wal_seg_size: usize) -> (XLogSegNo, TimeLineID) {
-    fname: &OsStr,
+    let tli = u32::from_str_radix(&fname[0..8], 16).unwrap();
-    wal_seg_size: usize,
+    let log = u32::from_str_radix(&fname[8..16], 16).unwrap() as XLogSegNo;
-) -> anyhow::Result<(XLogSegNo, TimeLineID)> {
+    let seg = u32::from_str_radix(&fname[16..24], 16).unwrap() as XLogSegNo;
-    if let Some(fname_str) = fname.to_str() {
+    (log * XLogSegmentsPerXLogId(wal_seg_size) + seg, tli)
-        let tli = u32::from_str_radix(&fname_str[0..8], 16)?;
-        let log = u32::from_str_radix(&fname_str[8..16], 16)? as XLogSegNo;
-        let seg = u32::from_str_radix(&fname_str[16..24], 16)? as XLogSegNo;
-        Ok((log * XLogSegmentsPerXLogId(wal_seg_size) + seg, tli))
-    } else {
-        anyhow::bail!("non-ut8 filename: {:?}", fname);
-    }
 }
 
-pub fn IsXLogFileName(fname: &OsStr) -> bool {
+pub fn IsXLogFileName(fname: &str) -> bool {
-    if let Some(fname) = fname.to_str() {
+    return fname.len() == XLOG_FNAME_LEN && fname.chars().all(|c| c.is_ascii_hexdigit());
-        fname.len() == XLOG_FNAME_LEN && fname.chars().all(|c| c.is_ascii_hexdigit())
-    } else {
-        false
-    }
 }
 
-pub fn IsPartialXLogFileName(fname: &OsStr) -> bool {
+pub fn IsPartialXLogFileName(fname: &str) -> bool {
-    if let Some(fname) = fname.to_str() {
+    fname.ends_with(".partial") && IsXLogFileName(&fname[0..fname.len() - 8])
-        fname.ends_with(".partial") && IsXLogFileName(OsStr::new(&fname[0..fname.len() - 8]))
-    } else {
-        false
-    }
 }
 
 /// If LSN points to the beginning of the page, then shift it to first record,

libs/postgres_ffi/wal_craft/src/lib.rs

@@ -7,7 +7,6 @@ use postgres_ffi::{WAL_SEGMENT_SIZE, XLOG_BLCKSZ};
 use postgres_ffi::{
     XLOG_SIZE_OF_XLOG_LONG_PHD, XLOG_SIZE_OF_XLOG_RECORD, XLOG_SIZE_OF_XLOG_SHORT_PHD,
 };
-use std::ffi::OsStr;
 use std::path::{Path, PathBuf};
 use std::process::Command;
 use std::time::{Duration, Instant};
@@ -136,8 +135,8 @@ impl Conf {
 
     pub fn pg_waldump(
         &self,
-        first_segment_name: &OsStr,
+        first_segment_name: &str,
-        last_segment_name: &OsStr,
+        last_segment_name: &str,
     ) -> anyhow::Result<std::process::Output> {
         let first_segment_file = self.datadir.join(first_segment_name);
         let last_segment_file = self.datadir.join(last_segment_name);

libs/postgres_ffi/wal_craft/src/xlog_utils_test.rs

@@ -4,7 +4,6 @@ use super::*;
 use crate::{error, info};
 use regex::Regex;
 use std::cmp::min;
-use std::ffi::OsStr;
 use std::fs::{self, File};
 use std::io::Write;
 use std::{env, str::FromStr};
@@ -55,7 +54,7 @@ fn test_end_of_wal<C: crate::Crafter>(test_name: &str) {
         .wal_dir()
         .read_dir()
         .unwrap()
-        .map(|f| f.unwrap().file_name())
+        .map(|f| f.unwrap().file_name().into_string().unwrap())
         .filter(|fname| IsXLogFileName(fname))
         .max()
         .unwrap();
@@ -71,11 +70,11 @@ fn test_end_of_wal<C: crate::Crafter>(test_name: &str) {
             start_lsn
         );
         for file in fs::read_dir(cfg.wal_dir()).unwrap().flatten() {
-            let fname = file.file_name();
+            let fname = file.file_name().into_string().unwrap();
             if !IsXLogFileName(&fname) {
                 continue;
             }
-            let (segno, _) = XLogFromFileName(&fname, WAL_SEGMENT_SIZE).unwrap();
+            let (segno, _) = XLogFromFileName(&fname, WAL_SEGMENT_SIZE);
             let seg_start_lsn = XLogSegNoOffsetToRecPtr(segno, 0, WAL_SEGMENT_SIZE);
             if seg_start_lsn > u64::from(*start_lsn) {
                 continue;
@@ -94,10 +93,10 @@ fn test_end_of_wal<C: crate::Crafter>(test_name: &str) {
     }
 }
 
-fn find_pg_waldump_end_of_wal(cfg: &crate::Conf, last_segment: &OsStr) -> Lsn {
+fn find_pg_waldump_end_of_wal(cfg: &crate::Conf, last_segment: &str) -> Lsn {
     // Get the actual end of WAL by pg_waldump
     let waldump_output = cfg
-        .pg_waldump(OsStr::new("000000010000000000000001"), last_segment)
+        .pg_waldump("000000010000000000000001", last_segment)
         .unwrap()
         .stderr;
     let waldump_output = std::str::from_utf8(&waldump_output).unwrap();
@@ -118,7 +117,7 @@ fn find_pg_waldump_end_of_wal(cfg: &crate::Conf, last_segment: &OsStr) -> Lsn {
 
 fn check_end_of_wal(
     cfg: &crate::Conf,
-    last_segment: &OsStr,
+    last_segment: &str,
     start_lsn: Lsn,
     expected_end_of_wal: Lsn,
 ) {
@@ -133,8 +132,7 @@ fn check_end_of_wal(
     // Rename file to partial to actually find last valid lsn, then rename it back.
     fs::rename(
         cfg.wal_dir().join(last_segment),
-        cfg.wal_dir()
+        cfg.wal_dir().join(format!("{}.partial", last_segment)),
-            .join(format!("{}.partial", last_segment.to_str().unwrap())),
     )
     .unwrap();
     let wal_end = find_end_of_wal(&cfg.wal_dir(), WAL_SEGMENT_SIZE, start_lsn).unwrap();
@@ -144,8 +142,7 @@ fn check_end_of_wal(
     );
     assert_eq!(wal_end, expected_end_of_wal);
     fs::rename(
-        cfg.wal_dir()
+        cfg.wal_dir().join(format!("{}.partial", last_segment)),
-            .join(format!("{}.partial", last_segment.to_str().unwrap())),
         cfg.wal_dir().join(last_segment),
     )
     .unwrap();

libs/utils/src/http/error.rs

@@ -82,7 +82,7 @@ impl ApiError {
                 StatusCode::INTERNAL_SERVER_ERROR,
             ),
             ApiError::InternalServerError(err) => HttpErrorBody::response_from_msg_and_status(
-                format!("{err:#}"), // use alternative formatting so that we give the cause without backtrace
+                err.to_string(),
                 StatusCode::INTERNAL_SERVER_ERROR,
             ),
         }

libs/utils/src/leaky_bucket.rs

@@ -21,13 +21,7 @@
 //!
 //! Another explaination can be found here: <https://brandur.org/rate-limiting>
 
-use std::{
+use std::{sync::Mutex, time::Duration};
-    sync::{
-        atomic::{AtomicU64, Ordering},
-        Mutex,
-    },
-    time::Duration,
-};
 
 use tokio::{sync::Notify, time::Instant};
 
@@ -134,7 +128,6 @@ impl LeakyBucketState {
 
 pub struct RateLimiter {
     pub config: LeakyBucketConfig,
-    pub sleep_counter: AtomicU64,
     pub state: Mutex<LeakyBucketState>,
     /// a queue to provide this fair ordering.
     pub queue: Notify,
@@ -151,7 +144,6 @@ impl Drop for Requeue<'_> {
 impl RateLimiter {
     pub fn with_initial_tokens(config: LeakyBucketConfig, initial_tokens: f64) -> Self {
         RateLimiter {
-            sleep_counter: AtomicU64::new(0),
             state: Mutex::new(LeakyBucketState::with_initial_tokens(
                 &config,
                 initial_tokens,
@@ -171,16 +163,15 @@ impl RateLimiter {
 
     /// returns true if we did throttle
     pub async fn acquire(&self, count: usize) -> bool {
-        let start = tokio::time::Instant::now();
+        let mut throttled = false;
 
-        let start_count = self.sleep_counter.load(Ordering::Acquire);
+        let start = tokio::time::Instant::now();
-        let mut end_count = start_count;
 
         // wait until we are the first in the queue
         let mut notified = std::pin::pin!(self.queue.notified());
         if !notified.as_mut().enable() {
+            throttled = true;
             notified.await;
-            end_count = self.sleep_counter.load(Ordering::Acquire);
         }
 
         // notify the next waiter in the queue when we are done.
@@ -193,22 +184,9 @@ impl RateLimiter {
                 .unwrap()
                 .add_tokens(&self.config, start, count as f64);
             match res {
-                Ok(()) => return end_count > start_count,
+                Ok(()) => return throttled,
                 Err(ready_at) => {
-                    struct Increment<'a>(&'a AtomicU64);
+                    throttled = true;
-
-                    impl Drop for Increment<'_> {
-                        fn drop(&mut self) {
-                            self.0.fetch_add(1, Ordering::AcqRel);
-                        }
-                    }
-
-                    // increment the counter after we finish sleeping (or cancel this task).
-                    // this ensures that tasks that have already started the acquire will observe
-                    // the new sleep count when they are allowed to resume on the notify.
-                    let _inc = Increment(&self.sleep_counter);
-                    end_count += 1;
-
                     tokio::time::sleep_until(ready_at).await;
                 }
             }

pageserver/src/config.rs

@@ -70,8 +70,6 @@ pub struct PageServerConf {
 
     pub superuser: String,
 
-    pub initdb_cache_dir: Option<Utf8PathBuf>,
-
     pub page_cache_size: usize,
     pub max_file_descriptors: usize,
 
@@ -299,7 +297,6 @@ impl PageServerConf {
             wait_lsn_timeout,
             wal_redo_timeout,
             superuser,
-            initdb_cache_dir,
             page_cache_size,
             max_file_descriptors,
             pg_distrib_dir,
@@ -347,7 +344,6 @@ impl PageServerConf {
             wait_lsn_timeout,
             wal_redo_timeout,
             superuser,
-            initdb_cache_dir,
             page_cache_size,
             max_file_descriptors,
             http_auth_type,
@@ -483,6 +479,11 @@ pub struct ConfigurableSemaphore {
 }
 
 impl ConfigurableSemaphore {
+    pub const DEFAULT_INITIAL: NonZeroUsize = match NonZeroUsize::new(1) {
+        Some(x) => x,
+        None => panic!("const unwrap is not yet stable"),
+    };
+
     /// Initializse using a non-zero amount of permits.
     ///
     /// Require a non-zero initial permits, because using permits == 0 is a crude way to disable a
@@ -503,6 +504,12 @@ impl ConfigurableSemaphore {
     }
 }
 
+impl Default for ConfigurableSemaphore {
+    fn default() -> Self {
+        Self::new(Self::DEFAULT_INITIAL)
+    }
+}
+
 impl PartialEq for ConfigurableSemaphore {
     fn eq(&self, other: &Self) -> bool {
         // the number of permits can be increased at runtime, so we cannot really fulfill the

pageserver/src/metrics.rs

@@ -1177,10 +1177,10 @@ pub(crate) mod virtual_file_io_engine {
 }
 
 struct GlobalAndPerTimelineHistogramTimer<'a, 'c> {
-    global_latency_histo: &'a Histogram,
+    global_metric: &'a Histogram,
 
     // Optional because not all op types are tracked per-timeline
-    per_timeline_latency_histo: Option<&'a Histogram>,
+    timeline_metric: Option<&'a Histogram>,
 
     ctx: &'c RequestContext,
     start: std::time::Instant,
@@ -1212,10 +1212,9 @@ impl<'a, 'c> Drop for GlobalAndPerTimelineHistogramTimer<'a, 'c> {
                 elapsed
             }
         };
-        self.global_latency_histo
+        self.global_metric.observe(ex_throttled.as_secs_f64());
-            .observe(ex_throttled.as_secs_f64());
+        if let Some(timeline_metric) = self.timeline_metric {
-        if let Some(per_timeline_getpage_histo) = self.per_timeline_latency_histo {
+            timeline_metric.observe(ex_throttled.as_secs_f64());
-            per_timeline_getpage_histo.observe(ex_throttled.as_secs_f64());
         }
     }
 }
@@ -1241,32 +1240,10 @@ pub enum SmgrQueryType {
 
 #[derive(Debug)]
 pub(crate) struct SmgrQueryTimePerTimeline {
-    global_started: [IntCounter; SmgrQueryType::COUNT],
+    global_metrics: [Histogram; SmgrQueryType::COUNT],
-    global_latency: [Histogram; SmgrQueryType::COUNT],
+    per_timeline_getpage: Histogram,
-    per_timeline_getpage_started: IntCounter,
-    per_timeline_getpage_latency: Histogram,
 }
 
-static SMGR_QUERY_STARTED_GLOBAL: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        // it's a counter, but, name is prepared to extend it to a histogram of queue depth
-        "pageserver_smgr_query_started_global_count",
-        "Number of smgr queries started, aggregated by query type.",
-        &["smgr_query_type"],
-    )
-    .expect("failed to define a metric")
-});
-
-static SMGR_QUERY_STARTED_PER_TENANT_TIMELINE: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        // it's a counter, but, name is prepared to extend it to a histogram of queue depth
-        "pageserver_smgr_query_started_count",
-        "Number of smgr queries started, aggregated by query type and tenant/timeline.",
-        &["smgr_query_type", "tenant_id", "shard_id", "timeline_id"],
-    )
-    .expect("failed to define a metric")
-});
-
 static SMGR_QUERY_TIME_PER_TENANT_TIMELINE: Lazy<HistogramVec> = Lazy::new(|| {
     register_histogram_vec!(
         "pageserver_smgr_query_seconds",
@@ -1342,20 +1319,14 @@ impl SmgrQueryTimePerTimeline {
         let tenant_id = tenant_shard_id.tenant_id.to_string();
         let shard_slug = format!("{}", tenant_shard_id.shard_slug());
         let timeline_id = timeline_id.to_string();
-        let global_started = std::array::from_fn(|i| {
+        let global_metrics = std::array::from_fn(|i| {
-            let op = SmgrQueryType::from_repr(i).unwrap();
-            SMGR_QUERY_STARTED_GLOBAL
-                .get_metric_with_label_values(&[op.into()])
-                .unwrap()
-        });
-        let global_latency = std::array::from_fn(|i| {
             let op = SmgrQueryType::from_repr(i).unwrap();
             SMGR_QUERY_TIME_GLOBAL
                 .get_metric_with_label_values(&[op.into()])
                 .unwrap()
         });
 
-        let per_timeline_getpage_started = SMGR_QUERY_STARTED_PER_TENANT_TIMELINE
+        let per_timeline_getpage = SMGR_QUERY_TIME_PER_TENANT_TIMELINE
             .get_metric_with_label_values(&[
                 SmgrQueryType::GetPageAtLsn.into(),
                 &tenant_id,
@@ -1363,20 +1334,9 @@ impl SmgrQueryTimePerTimeline {
                 &timeline_id,
             ])
             .unwrap();
-        let per_timeline_getpage_latency = SMGR_QUERY_TIME_PER_TENANT_TIMELINE
-            .get_metric_with_label_values(&[
-                SmgrQueryType::GetPageAtLsn.into(),
-                &tenant_id,
-                &shard_slug,
-                &timeline_id,
-            ])
-            .unwrap();
-
         Self {
-            global_started,
+            global_metrics,
-            global_latency,
+            per_timeline_getpage,
-            per_timeline_getpage_latency,
-            per_timeline_getpage_started,
         }
     }
     pub(crate) fn start_timer<'c: 'a, 'a>(
@@ -1384,11 +1344,8 @@ impl SmgrQueryTimePerTimeline {
         op: SmgrQueryType,
         ctx: &'c RequestContext,
     ) -> Option<impl Drop + '_> {
+        let global_metric = &self.global_metrics[op as usize];
         let start = Instant::now();
-
-        self.global_started[op as usize].inc();
-
-        // We subtract time spent throttled from the observed latency.
         match ctx.micros_spent_throttled.open() {
             Ok(()) => (),
             Err(error) => {
@@ -1407,16 +1364,15 @@ impl SmgrQueryTimePerTimeline {
             }
         }
 
-        let per_timeline_latency_histo = if matches!(op, SmgrQueryType::GetPageAtLsn) {
+        let timeline_metric = if matches!(op, SmgrQueryType::GetPageAtLsn) {
-            self.per_timeline_getpage_started.inc();
+            Some(&self.per_timeline_getpage)
-            Some(&self.per_timeline_getpage_latency)
         } else {
             None
         };
 
         Some(GlobalAndPerTimelineHistogramTimer {
-            global_latency_histo: &self.global_latency[op as usize],
+            global_metric,
-            per_timeline_latency_histo,
+            timeline_metric,
             ctx,
             start,
             op,
@@ -1467,12 +1423,9 @@ mod smgr_query_time_tests {
             let get_counts = || {
                 let global: u64 = ops
                     .iter()
-                    .map(|op| metrics.global_latency[*op as usize].get_sample_count())
+                    .map(|op| metrics.global_metrics[*op as usize].get_sample_count())
                     .sum();
-                (
+                (global, metrics.per_timeline_getpage.get_sample_count())
-                    global,
-                    metrics.per_timeline_getpage_latency.get_sample_count(),
-                )
             };
 
             let (pre_global, pre_per_tenant_timeline) = get_counts();
@@ -2623,12 +2576,6 @@ impl TimelineMetrics {
             let _ = STORAGE_IO_SIZE.remove_label_values(&[op, tenant_id, shard_id, timeline_id]);
         }
 
-        let _ = SMGR_QUERY_STARTED_PER_TENANT_TIMELINE.remove_label_values(&[
-            SmgrQueryType::GetPageAtLsn.into(),
-            tenant_id,
-            shard_id,
-            timeline_id,
-        ]);
         let _ = SMGR_QUERY_TIME_PER_TENANT_TIMELINE.remove_label_values(&[
             SmgrQueryType::GetPageAtLsn.into(),
             tenant_id,
@@ -2645,8 +2592,6 @@ pub(crate) fn remove_tenant_metrics(tenant_shard_id: &TenantShardId) {
         let _ = TENANT_SYNTHETIC_SIZE_METRIC.remove_label_values(&[&tid]);
     }
 
-    tenant_throttling::remove_tenant_metrics(tenant_shard_id);
-
     // we leave the BROKEN_TENANTS_SET entry if any
 }
 
@@ -3110,180 +3055,41 @@ pub mod tokio_epoll_uring {
 pub(crate) mod tenant_throttling {
     use metrics::{register_int_counter_vec, IntCounter};
     use once_cell::sync::Lazy;
-    use utils::shard::TenantShardId;
 
     use crate::tenant::{self, throttle::Metric};
 
-    struct GlobalAndPerTenantIntCounter {
-        global: IntCounter,
-        per_tenant: IntCounter,
-    }
-
-    impl GlobalAndPerTenantIntCounter {
-        #[inline(always)]
-        pub(crate) fn inc(&self) {
-            self.inc_by(1)
-        }
-        #[inline(always)]
-        pub(crate) fn inc_by(&self, n: u64) {
-            self.global.inc_by(n);
-            self.per_tenant.inc_by(n);
-        }
-    }
-
     pub(crate) struct TimelineGet {
-        count_accounted_start: GlobalAndPerTenantIntCounter,
+        wait_time: IntCounter,
-        count_accounted_finish: GlobalAndPerTenantIntCounter,
+        count: IntCounter,
-        wait_time: GlobalAndPerTenantIntCounter,
-        count_throttled: GlobalAndPerTenantIntCounter,
     }
 
-    static COUNT_ACCOUNTED_START: Lazy<metrics::IntCounterVec> = Lazy::new(|| {
+    pub(crate) static TIMELINE_GET: Lazy<TimelineGet> = Lazy::new(|| {
-        register_int_counter_vec!(
+        static WAIT_USECS: Lazy<metrics::IntCounterVec> = Lazy::new(|| {
-            "pageserver_tenant_throttling_count_accounted_start_global",
+            register_int_counter_vec!(
-            "Count of tenant throttling starts, by kind of throttle.",
-            &["kind"]
-        )
-        .unwrap()
-    });
-    static COUNT_ACCOUNTED_START_PER_TENANT: Lazy<metrics::IntCounterVec> = Lazy::new(|| {
-        register_int_counter_vec!(
-            "pageserver_tenant_throttling_count_accounted_start",
-            "Count of tenant throttling starts, by kind of throttle.",
-            &["kind", "tenant_id", "shard_id"]
-        )
-        .unwrap()
-    });
-    static COUNT_ACCOUNTED_FINISH: Lazy<metrics::IntCounterVec> = Lazy::new(|| {
-        register_int_counter_vec!(
-            "pageserver_tenant_throttling_count_accounted_finish_global",
-            "Count of tenant throttling finishes, by kind of throttle.",
-            &["kind"]
-        )
-        .unwrap()
-    });
-    static COUNT_ACCOUNTED_FINISH_PER_TENANT: Lazy<metrics::IntCounterVec> = Lazy::new(|| {
-        register_int_counter_vec!(
-            "pageserver_tenant_throttling_count_accounted_finish",
-            "Count of tenant throttling finishes, by kind of throttle.",
-            &["kind", "tenant_id", "shard_id"]
-        )
-        .unwrap()
-    });
-    static WAIT_USECS: Lazy<metrics::IntCounterVec> = Lazy::new(|| {
-        register_int_counter_vec!(
             "pageserver_tenant_throttling_wait_usecs_sum_global",
-            "Sum of microseconds that spent waiting throttle by kind of throttle.",
+            "Sum of microseconds that tenants spent waiting for a tenant throttle of a given kind.",
             &["kind"]
         )
-        .unwrap()
+            .unwrap()
-    });
+        });
-    static WAIT_USECS_PER_TENANT: Lazy<metrics::IntCounterVec> = Lazy::new(|| {
+
-        register_int_counter_vec!(
+        static WAIT_COUNT: Lazy<metrics::IntCounterVec> = Lazy::new(|| {
-            "pageserver_tenant_throttling_wait_usecs_sum",
+            register_int_counter_vec!(
-            "Sum of microseconds that spent waiting throttle by kind of throttle.",
+                "pageserver_tenant_throttling_count_global",
-            &["kind", "tenant_id", "shard_id"]
+                "Count of tenant throttlings, by kind of throttle.",
-        )
+                &["kind"]
-        .unwrap()
+            )
+            .unwrap()
+        });
+
+        let kind = "timeline_get";
+        TimelineGet {
+            wait_time: WAIT_USECS.with_label_values(&[kind]),
+            count: WAIT_COUNT.with_label_values(&[kind]),
+        }
     });
 
-    static WAIT_COUNT: Lazy<metrics::IntCounterVec> = Lazy::new(|| {
+    impl Metric for &'static TimelineGet {
-        register_int_counter_vec!(
-            "pageserver_tenant_throttling_count_global",
-            "Count of tenant throttlings, by kind of throttle.",
-            &["kind"]
-        )
-        .unwrap()
-    });
-    static WAIT_COUNT_PER_TENANT: Lazy<metrics::IntCounterVec> = Lazy::new(|| {
-        register_int_counter_vec!(
-            "pageserver_tenant_throttling_count",
-            "Count of tenant throttlings, by kind of throttle.",
-            &["kind", "tenant_id", "shard_id"]
-        )
-        .unwrap()
-    });
-
-    const KIND: &str = "timeline_get";
-
-    impl TimelineGet {
-        pub(crate) fn new(tenant_shard_id: &TenantShardId) -> Self {
-            TimelineGet {
-                count_accounted_start: {
-                    GlobalAndPerTenantIntCounter {
-                        global: COUNT_ACCOUNTED_START.with_label_values(&[KIND]),
-                        per_tenant: COUNT_ACCOUNTED_START_PER_TENANT.with_label_values(&[
-                            KIND,
-                            &tenant_shard_id.tenant_id.to_string(),
-                            &tenant_shard_id.shard_slug().to_string(),
-                        ]),
-                    }
-                },
-                count_accounted_finish: {
-                    GlobalAndPerTenantIntCounter {
-                        global: COUNT_ACCOUNTED_FINISH.with_label_values(&[KIND]),
-                        per_tenant: COUNT_ACCOUNTED_FINISH_PER_TENANT.with_label_values(&[
-                            KIND,
-                            &tenant_shard_id.tenant_id.to_string(),
-                            &tenant_shard_id.shard_slug().to_string(),
-                        ]),
-                    }
-                },
-                wait_time: {
-                    GlobalAndPerTenantIntCounter {
-                        global: WAIT_USECS.with_label_values(&[KIND]),
-                        per_tenant: WAIT_USECS_PER_TENANT.with_label_values(&[
-                            KIND,
-                            &tenant_shard_id.tenant_id.to_string(),
-                            &tenant_shard_id.shard_slug().to_string(),
-                        ]),
-                    }
-                },
-                count_throttled: {
-                    GlobalAndPerTenantIntCounter {
-                        global: WAIT_COUNT.with_label_values(&[KIND]),
-                        per_tenant: WAIT_COUNT_PER_TENANT.with_label_values(&[
-                            KIND,
-                            &tenant_shard_id.tenant_id.to_string(),
-                            &tenant_shard_id.shard_slug().to_string(),
-                        ]),
-                    }
-                },
-            }
-        }
-    }
-
-    pub(crate) fn preinitialize_global_metrics() {
-        Lazy::force(&COUNT_ACCOUNTED_START);
-        Lazy::force(&COUNT_ACCOUNTED_FINISH);
-        Lazy::force(&WAIT_USECS);
-        Lazy::force(&WAIT_COUNT);
-    }
-
-    pub(crate) fn remove_tenant_metrics(tenant_shard_id: &TenantShardId) {
-        for m in &[
-            &COUNT_ACCOUNTED_START_PER_TENANT,
-            &COUNT_ACCOUNTED_FINISH_PER_TENANT,
-            &WAIT_USECS_PER_TENANT,
-            &WAIT_COUNT_PER_TENANT,
-        ] {
-            let _ = m.remove_label_values(&[
-                KIND,
-                &tenant_shard_id.tenant_id.to_string(),
-                &tenant_shard_id.shard_slug().to_string(),
-            ]);
-        }
-    }
-
-    impl Metric for TimelineGet {
-        #[inline(always)]
-        fn accounting_start(&self) {
-            self.count_accounted_start.inc();
-        }
-        #[inline(always)]
-        fn accounting_finish(&self) {
-            self.count_accounted_finish.inc();
-        }
         #[inline(always)]
         fn observe_throttling(
             &self,
@@ -3291,7 +3097,7 @@ pub(crate) mod tenant_throttling {
         ) {
             let val = u64::try_from(wait_time.as_micros()).unwrap();
             self.wait_time.inc_by(val);
-            self.count_throttled.inc();
+            self.count.inc();
         }
     }
 }
@@ -3421,14 +3227,11 @@ pub fn preinitialize_metrics() {
     }
 
     // countervecs
-    [
+    [&BACKGROUND_LOOP_PERIOD_OVERRUN_COUNT]
-        &BACKGROUND_LOOP_PERIOD_OVERRUN_COUNT,
+        .into_iter()
-        &SMGR_QUERY_STARTED_GLOBAL,
+        .for_each(|c| {
-    ]
+            Lazy::force(c);
-    .into_iter()
+        });
-    .for_each(|c| {
-        Lazy::force(c);
-    });
 
     // gauges
     WALRECEIVER_ACTIVE_MANAGERS.get();
@@ -3450,8 +3253,7 @@ pub fn preinitialize_metrics() {
 
     // Custom
     Lazy::force(&RECONSTRUCT_TIME);
+    Lazy::force(&tenant_throttling::TIMELINE_GET);
     Lazy::force(&BASEBACKUP_QUERY_TIME);
     Lazy::force(&COMPUTE_COMMANDS_COUNTERS);
-
-    tenant_throttling::preinitialize_global_metrics();
 }

pageserver/src/tenant.rs

@@ -140,7 +140,6 @@ pub mod metadata;
 pub mod remote_timeline_client;
 pub mod storage_layer;
 
-pub mod checks;
 pub mod config;
 pub mod mgr;
 pub mod secondary;
@@ -302,7 +301,7 @@ pub struct Tenant {
     /// Throttle applied at the top of [`Timeline::get`].
     /// All [`Tenant::timelines`] of a given [`Tenant`] instance share the same [`throttle::Throttle`] instance.
     pub(crate) timeline_get_throttle:
-        Arc<throttle::Throttle<crate::metrics::tenant_throttling::TimelineGet>>,
+        Arc<throttle::Throttle<&'static crate::metrics::tenant_throttling::TimelineGet>>,
 
     /// An ongoing timeline detach concurrency limiter.
     ///
@@ -1574,9 +1573,6 @@ impl Tenant {
         image_layer_desc: Vec<(Lsn, Vec<(pageserver_api::key::Key, bytes::Bytes)>)>,
         end_lsn: Lsn,
     ) -> anyhow::Result<Arc<Timeline>> {
-        use checks::check_valid_layermap;
-        use itertools::Itertools;
-
         let tline = self
             .create_test_timeline(new_timeline_id, initdb_lsn, pg_version, ctx)
             .await?;
@@ -1591,18 +1587,6 @@ impl Tenant {
                 .force_create_image_layer(lsn, images, Some(initdb_lsn), ctx)
                 .await?;
         }
-        let layer_names = tline
-            .layers
-            .read()
-            .await
-            .layer_map()
-            .unwrap()
-            .iter_historic_layers()
-            .map(|layer| layer.layer_name())
-            .collect_vec();
-        if let Some(err) = check_valid_layermap(&layer_names) {
-            bail!("invalid layermap: {err}");
-        }
         Ok(tline)
     }
 
@@ -2831,7 +2815,7 @@ impl Tenant {
             gate: Gate::default(),
             timeline_get_throttle: Arc::new(throttle::Throttle::new(
                 Tenant::get_timeline_get_throttle_config(conf, &attached_conf.tenant_conf),
-                crate::metrics::tenant_throttling::TimelineGet::new(&tenant_shard_id),
+                &crate::metrics::tenant_throttling::TIMELINE_GET,
             )),
             tenant_conf: Arc::new(ArcSwap::from_pointee(attached_conf)),
             ongoing_timeline_detach: std::sync::Mutex::default(),
@@ -3213,9 +3197,6 @@ impl Tenant {
         image_layer_desc: Vec<(Lsn, Vec<(pageserver_api::key::Key, bytes::Bytes)>)>,
         end_lsn: Lsn,
     ) -> anyhow::Result<Arc<Timeline>> {
-        use checks::check_valid_layermap;
-        use itertools::Itertools;
-
         let tline = self
             .branch_timeline_test(src_timeline, dst_id, ancestor_lsn, ctx)
             .await?;
@@ -3236,18 +3217,6 @@ impl Tenant {
                 .force_create_image_layer(lsn, images, Some(ancestor_lsn), ctx)
                 .await?;
         }
-        let layer_names = tline
-            .layers
-            .read()
-            .await
-            .layer_map()
-            .unwrap()
-            .iter_historic_layers()
-            .map(|layer| layer.layer_name())
-            .collect_vec();
-        if let Some(err) = check_valid_layermap(&layer_names) {
-            bail!("invalid layermap: {err}");
-        }
         Ok(tline)
     }
 
@@ -3522,7 +3491,7 @@ impl Tenant {
                 .context("extract initdb tar")?;
         } else {
             // Init temporarily repo to get bootstrap data, this creates a directory in the `pgdata_path` path
-            run_initdb_with_cache(self.conf, &pgdata_path, pg_version, &self.cancel).await?;
+            run_initdb(self.conf, &pgdata_path, pg_version, &self.cancel).await?;
 
             // Upload the created data dir to S3
             if self.tenant_shard_id().is_shard_zero() {
@@ -3868,118 +3837,6 @@ impl Tenant {
     }
 }
 
-fn cached_initdb_dirname(initial_superuser_name: &str, pg_version: u32) -> String
-{
-    use std::hash::Hash;
-    use std::hash::Hasher;
-    use std::collections::hash_map::DefaultHasher;
-    let mut hasher = DefaultHasher::new();
-    initial_superuser_name.hash(&mut hasher);
-    let hash = hasher.finish();
-
-    format!("cached_initial_pgdata_{pg_version}_{:016}", hash)
-}
-
-fn copy_dir_all(src: impl AsRef<std::path::Path>, dst: impl AsRef<std::path::Path>) -> std::io::Result<()> {
-    for entry in fs::read_dir(src.as_ref())? {
-        let entry = entry?;
-        let subsrc = entry.path();
-        let subdst = dst.as_ref().join(&entry.file_name());
-
-        if entry.file_type()?.is_dir() {
-            std::fs::create_dir(&subdst)?;
-            copy_dir_all(&subsrc, &subdst)?;
-        } else {
-            std::fs::copy(&subsrc, &subdst)?;
-        }
-    }
-    Ok(())
-}
-
-fn restore_cached_initdb_dir(
-    cached_path: &Utf8Path,
-    target_path: &Utf8Path,
-) -> anyhow::Result<bool> {
-    if !cached_path.exists() {
-        info!("cached initdb dir \"{cached_path}\" does not exist yet");
-        return Ok(false);
-    }
-
-    std::fs::create_dir(target_path)?;
-    copy_dir_all(cached_path, target_path)?;
-    info!("restored initdb result from cache dir \"{cached_path}\"");
-    Ok(true)
-}
-
-fn save_cached_initdb_dir(
-    src_path: &Utf8Path,
-    cache_path: &Utf8Path,
-) -> anyhow::Result<()> {
-    match std::fs::create_dir(cache_path) {
-        Ok(()) => {
-            info!("saving initdb result to cache dir \"{cache_path}\"");
-        },
-        Err(err) if err.kind() == std::io::ErrorKind::AlreadyExists => {
-            info!("cache initdb dir \"{cache_path}\" already exists, not saving");
-            return Ok(())
-        },
-        Err(err) => { return Err(anyhow::Error::from(err))},
-    };
-    let cache_dir_guard = scopeguard::guard(cache_path, |cp| {
-        if let Err(err) = std::fs::remove_dir_all(&cp) {
-            error!("could not remove cached initdb directory {cp}: {err}");
-        }
-    });
-
-    let cache_parent_path = cache_path.parent().ok_or(anyhow::Error::msg("no cache parent path"))?;
-
-    let tmp_dirpath = camino_tempfile::tempdir_in(cache_parent_path)?;
-    copy_dir_all(src_path, &tmp_dirpath)?;
-    std::fs::rename(tmp_dirpath, &*cache_dir_guard)?;
-
-    // disarm the guard
-    scopeguard::ScopeGuard::into_inner(cache_dir_guard);
-
-    Ok(())
-}
-
-async fn run_initdb_with_cache(
-    conf: &'static PageServerConf,
-    initdb_target_dir: &Utf8Path,
-    pg_version: u32,
-    cancel: &CancellationToken,
-) -> Result<(), InitdbError> {
-
-    let cache_dir = conf.initdb_cache_dir.as_ref().map(|initdb_cache_dir| {
-        initdb_cache_dir.join(cached_initdb_dirname(&conf.superuser, pg_version))
-    });
-
-    if let Some(cache_dir) = &cache_dir {
-        match restore_cached_initdb_dir(&cache_dir, initdb_target_dir) {
-            Ok(true) => return Ok(()),
-            Ok(false) => {},
-            Err(err) => {
-                warn!("Error restoring from cached initdb directory \"{cache_dir}\": {err}");
-                if initdb_target_dir.exists() {
-                    if let Err(err) = std::fs::remove_dir_all(&initdb_target_dir) {
-                        error!("could not remove temporary initdb target directory {initdb_target_dir}: {err}");
-                    }
-                }
-            },
-        }
-    }
-
-    run_initdb(conf, initdb_target_dir, pg_version, cancel).await?;
-
-    if let Some(cache_dir) = &cache_dir {
-        if let Err(err) = save_cached_initdb_dir(initdb_target_dir, &cache_dir) {
-            warn!("error saving initdb result to cache directory \"{cache_dir}\": {err}");
-        }
-    }
-
-    Ok(())
-}
-
 /// Create the cluster temporarily in 'initdbpath' directory inside the repository
 /// to get bootstrap data for timeline initialization.
 async fn run_initdb(
@@ -4307,18 +4164,9 @@ pub(crate) mod harness {
             let records_neon = records.iter().all(|r| apply_neon::can_apply_in_neon(&r.1));
             if records_neon {
                 // For Neon wal records, we can decode without spawning postgres, so do so.
-                let mut page = match (base_img, records.first()) {
+                let base_img = base_img.expect("Neon WAL redo requires base image").1;
-                    (Some((_lsn, img)), _) => {
+                let mut page = BytesMut::new();
-                        let mut page = BytesMut::new();
+                page.extend_from_slice(&base_img);
-                        page.extend_from_slice(&img);
-                        page
-                    }
-                    (_, Some((_lsn, rec))) if rec.will_init() => BytesMut::new(),
-                    _ => {
-                        panic!("Neon WAL redo requires base image or will init record");
-                    }
-                };
-
                 for (record_lsn, record) in records {
                     apply_neon::apply_in_neon(&record, record_lsn, key, &mut page)?;
                 }
@@ -8622,135 +8470,4 @@ mod tests {
 
         Ok(())
     }
-
-    // Regression test for https://github.com/neondatabase/neon/issues/9012
-    // Create an image arrangement where we have to read at different LSN ranges
-    // from a delta layer. This is achieved by overlapping an image layer on top of
-    // a delta layer. Like so:
-    //
-    //     A      B
-    // +----------------+ -> delta_layer
-    // |                |                           ^ lsn
-    // |       =========|-> nested_image_layer      |
-    // |       C        |                           |
-    // +----------------+                           |
-    // ======== -> baseline_image_layer             +-------> key
-    //
-    //
-    // When querying the key range [A, B) we need to read at different LSN ranges
-    // for [A, C) and [C, B). This test checks that the described edge case is handled correctly.
-    #[tokio::test]
-    async fn test_vectored_read_with_nested_image_layer() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_vectored_read_with_nested_image_layer").await?;
-        let (tenant, ctx) = harness.load().await;
-
-        let will_init_keys = [2, 6];
-        fn get_key(id: u32) -> Key {
-            let mut key = Key::from_hex("110000000033333333444444445500000000").unwrap();
-            key.field6 = id;
-            key
-        }
-
-        let mut expected_key_values = HashMap::new();
-
-        let baseline_image_layer_lsn = Lsn(0x10);
-        let mut baseline_img_layer = Vec::new();
-        for i in 0..5 {
-            let key = get_key(i);
-            let value = format!("value {i}@{baseline_image_layer_lsn}");
-
-            let removed = expected_key_values.insert(key, value.clone());
-            assert!(removed.is_none());
-
-            baseline_img_layer.push((key, Bytes::from(value)));
-        }
-
-        let nested_image_layer_lsn = Lsn(0x50);
-        let mut nested_img_layer = Vec::new();
-        for i in 5..10 {
-            let key = get_key(i);
-            let value = format!("value {i}@{nested_image_layer_lsn}");
-
-            let removed = expected_key_values.insert(key, value.clone());
-            assert!(removed.is_none());
-
-            nested_img_layer.push((key, Bytes::from(value)));
-        }
-
-        let mut delta_layer_spec = Vec::default();
-        let delta_layer_start_lsn = Lsn(0x20);
-        let mut delta_layer_end_lsn = delta_layer_start_lsn;
-
-        for i in 0..10 {
-            let key = get_key(i);
-            let key_in_nested = nested_img_layer
-                .iter()
-                .any(|(key_with_img, _)| *key_with_img == key);
-            let lsn = {
-                if key_in_nested {
-                    Lsn(nested_image_layer_lsn.0 + 0x10)
-                } else {
-                    delta_layer_start_lsn
-                }
-            };
-
-            let will_init = will_init_keys.contains(&i);
-            if will_init {
-                delta_layer_spec.push((key, lsn, Value::WalRecord(NeonWalRecord::wal_init())));
-
-                expected_key_values.insert(key, "".to_string());
-            } else {
-                let delta = format!("@{lsn}");
-                delta_layer_spec.push((
-                    key,
-                    lsn,
-                    Value::WalRecord(NeonWalRecord::wal_append(&delta)),
-                ));
-
-                expected_key_values
-                    .get_mut(&key)
-                    .expect("An image exists for each key")
-                    .push_str(delta.as_str());
-            }
-            delta_layer_end_lsn = std::cmp::max(delta_layer_start_lsn, lsn);
-        }
-
-        delta_layer_end_lsn = Lsn(delta_layer_end_lsn.0 + 1);
-
-        assert!(
-            nested_image_layer_lsn > delta_layer_start_lsn
-                && nested_image_layer_lsn < delta_layer_end_lsn
-        );
-
-        let tline = tenant
-            .create_test_timeline_with_layers(
-                TIMELINE_ID,
-                baseline_image_layer_lsn,
-                DEFAULT_PG_VERSION,
-                &ctx,
-                vec![DeltaLayerTestDesc::new_with_inferred_key_range(
-                    delta_layer_start_lsn..delta_layer_end_lsn,
-                    delta_layer_spec,
-                )], // delta layers
-                vec![
-                    (baseline_image_layer_lsn, baseline_img_layer),
-                    (nested_image_layer_lsn, nested_img_layer),
-                ], // image layers
-                delta_layer_end_lsn,
-            )
-            .await?;
-
-        let keyspace = KeySpace::single(get_key(0)..get_key(10));
-        let results = tline
-            .get_vectored(keyspace, delta_layer_end_lsn, &ctx)
-            .await
-            .expect("No vectored errors");
-        for (key, res) in results {
-            let value = res.expect("No key errors");
-            let expected_value = expected_key_values.remove(&key).expect("No unknown keys");
-            assert_eq!(value, Bytes::from(expected_value));
-        }
-
-        Ok(())
-    }
 }

pageserver/src/tenant/checks.rs

@@ -1,55 +0,0 @@
-use std::collections::BTreeSet;
-
-use itertools::Itertools;
-
-use super::storage_layer::LayerName;
-
-/// Checks whether a layer map is valid (i.e., is a valid result of the current compaction algorithm if nothing goes wrong).
-/// The function checks if we can split the LSN range of a delta layer only at the LSNs of the delta layers. For example,
-///
-/// ```plain
-/// |       |                 |       |
-/// |   1   |    |   2   |    |   3   |
-/// |       |    |       |    |       |
-/// ```
-///
-/// This is not a valid layer map because the LSN range of layer 1 intersects with the LSN range of layer 2. 1 and 2 should have
-/// the same LSN range.
-///
-/// The exception is that when layer 2 only contains a single key, it could be split over the LSN range. For example,
-///
-/// ```plain
-/// |       |    |   2   |    |       |
-/// |   1   |    |-------|    |   3   |
-/// |       |    |   4   |    |       |
-///
-/// If layer 2 and 4 contain the same single key, this is also a valid layer map.
-pub fn check_valid_layermap(metadata: &[LayerName]) -> Option<String> {
-    let mut lsn_split_point = BTreeSet::new(); // TODO: use a better data structure (range tree / range set?)
-    let mut all_delta_layers = Vec::new();
-    for name in metadata {
-        if let LayerName::Delta(layer) = name {
-            if layer.key_range.start.next() != layer.key_range.end {
-                all_delta_layers.push(layer.clone());
-            }
-        }
-    }
-    for layer in &all_delta_layers {
-        let lsn_range = &layer.lsn_range;
-        lsn_split_point.insert(lsn_range.start);
-        lsn_split_point.insert(lsn_range.end);
-    }
-    for layer in &all_delta_layers {
-        let lsn_range = layer.lsn_range.clone();
-        let intersects = lsn_split_point.range(lsn_range).collect_vec();
-        if intersects.len() > 1 {
-            let err = format!(
-                "layer violates the layer map LSN split assumption: layer {} intersects with LSN [{}]",
-                layer,
-                intersects.into_iter().map(|lsn| lsn.to_string()).join(", ")
-            );
-            return Some(err);
-        }
-    }
-    None
-}

pageserver/src/tenant/gc_block.rs

@@ -1,29 +1,11 @@
-use std::{collections::HashMap, time::Duration};
+use std::collections::HashMap;
 
-use super::remote_timeline_client::index::GcBlockingReason;
-use tokio::time::Instant;
 use utils::id::TimelineId;
 
-type TimelinesBlocked = HashMap<TimelineId, enumset::EnumSet<GcBlockingReason>>;
+use super::remote_timeline_client::index::GcBlockingReason;
 
-#[derive(Default)]
+type Storage = HashMap<TimelineId, enumset::EnumSet<GcBlockingReason>>;
-struct Storage {
-    timelines_blocked: TimelinesBlocked,
-    /// The deadline before which we are blocked from GC so that
-    /// leases have a chance to be renewed.
-    lsn_lease_deadline: Option<Instant>,
-}
 
-impl Storage {
-    fn is_blocked_by_lsn_lease_deadline(&self) -> bool {
-        self.lsn_lease_deadline
-            .map(|d| Instant::now() < d)
-            .unwrap_or(false)
-    }
-}
-
-/// GcBlock provides persistent (per-timeline) gc blocking and facilitates transient time based gc
-/// blocking.
 #[derive(Default)]
 pub(crate) struct GcBlock {
     /// The timelines which have current reasons to block gc.
@@ -31,12 +13,6 @@ pub(crate) struct GcBlock {
     /// LOCK ORDER: this is held locked while scheduling the next index_part update. This is done
     /// to keep the this field up to date with RemoteTimelineClient `upload_queue.dirty`.
     reasons: std::sync::Mutex<Storage>,
-
-    /// GC background task or manually run `Tenant::gc_iteration` holds a lock on this.
-    ///
-    /// Do not add any more features taking and forbidding taking this lock. It should be
-    /// `tokio::sync::Notify`, but that is rarely used. On the other side, [`GcBlock::insert`]
-    /// synchronizes with gc attempts by locking and unlocking this mutex.
     blocking: tokio::sync::Mutex<()>,
 }
 
@@ -66,20 +42,6 @@ impl GcBlock {
         }
     }
 
-    /// Sets a deadline before which we cannot proceed to GC due to lsn lease.
-    ///
-    /// We do this as the leases mapping are not persisted to disk. By delaying GC by lease
-    /// length, we guarantee that all the leases we granted before will have a chance to renew
-    /// when we run GC for the first time after restart / transition from AttachedMulti to AttachedSingle.
-    pub(super) fn set_lsn_lease_deadline(&self, lsn_lease_length: Duration) {
-        let deadline = Instant::now() + lsn_lease_length;
-        let mut g = self.reasons.lock().unwrap();
-        g.lsn_lease_deadline = Some(deadline);
-    }
-
-    /// Describe the current gc blocking reasons.
-    ///
-    /// TODO: make this json serializable.
     pub(crate) fn summary(&self) -> Option<BlockingReasons> {
         let g = self.reasons.lock().unwrap();
 
@@ -102,7 +64,7 @@ impl GcBlock {
     ) -> anyhow::Result<bool> {
         let (added, uploaded) = {
             let mut g = self.reasons.lock().unwrap();
-            let set = g.timelines_blocked.entry(timeline.timeline_id).or_default();
+            let set = g.entry(timeline.timeline_id).or_default();
             let added = set.insert(reason);
 
             // LOCK ORDER: intentionally hold the lock, see self.reasons.
@@ -133,7 +95,7 @@ impl GcBlock {
 
         let (remaining_blocks, uploaded) = {
             let mut g = self.reasons.lock().unwrap();
-            match g.timelines_blocked.entry(timeline.timeline_id) {
+            match g.entry(timeline.timeline_id) {
                 Entry::Occupied(mut oe) => {
                     let set = oe.get_mut();
                     set.remove(reason);
@@ -147,7 +109,7 @@ impl GcBlock {
                 }
             }
 
-            let remaining_blocks = g.timelines_blocked.len();
+            let remaining_blocks = g.len();
 
             // LOCK ORDER: intentionally hold the lock while scheduling; see self.reasons
             let uploaded = timeline
@@ -172,11 +134,11 @@ impl GcBlock {
     pub(crate) fn before_delete(&self, timeline: &super::Timeline) {
         let unblocked = {
             let mut g = self.reasons.lock().unwrap();
-            if g.timelines_blocked.is_empty() {
+            if g.is_empty() {
                 return;
             }
 
-            g.timelines_blocked.remove(&timeline.timeline_id);
+            g.remove(&timeline.timeline_id);
 
             BlockingReasons::clean_and_summarize(g).is_none()
         };
@@ -187,11 +149,10 @@ impl GcBlock {
     }
 
     /// Initialize with the non-deleted timelines of this tenant.
-    pub(crate) fn set_scanned(&self, scanned: TimelinesBlocked) {
+    pub(crate) fn set_scanned(&self, scanned: Storage) {
         let mut g = self.reasons.lock().unwrap();
-        assert!(g.timelines_blocked.is_empty());
+        assert!(g.is_empty());
-        g.timelines_blocked
+        g.extend(scanned.into_iter().filter(|(_, v)| !v.is_empty()));
-            .extend(scanned.into_iter().filter(|(_, v)| !v.is_empty()));
 
         if let Some(reasons) = BlockingReasons::clean_and_summarize(g) {
             tracing::info!(summary=?reasons, "initialized with gc blocked");
@@ -205,7 +166,6 @@ pub(super) struct Guard<'a> {
 
 #[derive(Debug)]
 pub(crate) struct BlockingReasons {
-    tenant_blocked_by_lsn_lease_deadline: bool,
     timelines: usize,
     reasons: enumset::EnumSet<GcBlockingReason>,
 }
@@ -214,8 +174,8 @@ impl std::fmt::Display for BlockingReasons {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(
             f,
-            "tenant_blocked_by_lsn_lease_deadline: {}, {} timelines block for {:?}",
+            "{} timelines block for {:?}",
-            self.tenant_blocked_by_lsn_lease_deadline, self.timelines, self.reasons
+            self.timelines, self.reasons
         )
     }
 }
@@ -223,15 +183,13 @@ impl std::fmt::Display for BlockingReasons {
 impl BlockingReasons {
     fn clean_and_summarize(mut g: std::sync::MutexGuard<'_, Storage>) -> Option<Self> {
         let mut reasons = enumset::EnumSet::empty();
-        g.timelines_blocked.retain(|_key, value| {
+        g.retain(|_key, value| {
             reasons = reasons.union(*value);
             !value.is_empty()
         });
-        let blocked_by_lsn_lease_deadline = g.is_blocked_by_lsn_lease_deadline();
+        if !g.is_empty() {
-        if !g.timelines_blocked.is_empty() || blocked_by_lsn_lease_deadline {
             Some(BlockingReasons {
-                tenant_blocked_by_lsn_lease_deadline: blocked_by_lsn_lease_deadline,
+                timelines: g.len(),
-                timelines: g.timelines_blocked.len(),
                 reasons,
             })
         } else {
@@ -240,17 +198,14 @@ impl BlockingReasons {
     }
 
     fn summarize(g: &std::sync::MutexGuard<'_, Storage>) -> Option<Self> {
-        let blocked_by_lsn_lease_deadline = g.is_blocked_by_lsn_lease_deadline();
+        if g.is_empty() {
-        if g.timelines_blocked.is_empty() && !blocked_by_lsn_lease_deadline {
             None
         } else {
             let reasons = g
-                .timelines_blocked
                 .values()
                 .fold(enumset::EnumSet::empty(), |acc, next| acc.union(*next));
             Some(BlockingReasons {
-                tenant_blocked_by_lsn_lease_deadline: blocked_by_lsn_lease_deadline,
+                timelines: g.len(),
-                timelines: g.timelines_blocked.len(),
                 reasons,
             })
         }

pageserver/src/tenant/mgr.rs

@@ -949,12 +949,6 @@ impl TenantManager {
                 (LocationMode::Attached(attach_conf), Some(TenantSlot::Attached(tenant))) => {
                     match attach_conf.generation.cmp(&tenant.generation) {
                         Ordering::Equal => {
-                            if attach_conf.attach_mode == AttachmentMode::Single {
-                                tenant
-                                    .gc_block
-                                    .set_lsn_lease_deadline(tenant.get_lsn_lease_length());
-                            }
-
                             // A transition from Attached to Attached in the same generation, we may
                             // take our fast path and just provide the updated configuration
                             // to the tenant.

pageserver/src/tenant/storage_layer.rs

@@ -276,16 +276,6 @@ pub(crate) enum LayerId {
     InMemoryLayerId(InMemoryLayerFileId),
 }
 
-/// Uniquely identify a layer visit by the layer
-/// and LSN floor (or start LSN) of the reads.
-/// The layer itself is not enough since we may
-/// have different LSN lower bounds for delta layer reads.
-#[derive(Debug, PartialEq, Eq, Clone, Hash)]
-struct LayerToVisitId {
-    layer_id: LayerId,
-    lsn_floor: Lsn,
-}
-
 /// Layer wrapper for the read path. Note that it is valid
 /// to use these layers even after external operations have
 /// been performed on them (compaction, freeze, etc.).
@@ -297,9 +287,9 @@ pub(crate) enum ReadableLayer {
 
 /// A partial description of a read to be done.
 #[derive(Debug, Clone)]
-struct LayerVisit {
+struct ReadDesc {
     /// An id used to resolve the readable layer within the fringe
-    layer_to_visit_id: LayerToVisitId,
+    layer_id: LayerId,
     /// Lsn range for the read, used for selecting the next read
     lsn_range: Range<Lsn>,
 }
@@ -313,12 +303,12 @@ struct LayerVisit {
 /// a two layer indexing scheme.
 #[derive(Debug)]
 pub(crate) struct LayerFringe {
-    planned_visits_by_lsn: BinaryHeap<LayerVisit>,
+    planned_reads_by_lsn: BinaryHeap<ReadDesc>,
-    visit_reads: HashMap<LayerToVisitId, LayerVisitReads>,
+    layers: HashMap<LayerId, LayerKeyspace>,
 }
 
 #[derive(Debug)]
-struct LayerVisitReads {
+struct LayerKeyspace {
     layer: ReadableLayer,
     target_keyspace: KeySpaceRandomAccum,
 }
@@ -326,23 +316,23 @@ struct LayerVisitReads {
 impl LayerFringe {
     pub(crate) fn new() -> Self {
         LayerFringe {
-            planned_visits_by_lsn: BinaryHeap::new(),
+            planned_reads_by_lsn: BinaryHeap::new(),
-            visit_reads: HashMap::new(),
+            layers: HashMap::new(),
         }
     }
 
     pub(crate) fn next_layer(&mut self) -> Option<(ReadableLayer, KeySpace, Range<Lsn>)> {
-        let read_desc = match self.planned_visits_by_lsn.pop() {
+        let read_desc = match self.planned_reads_by_lsn.pop() {
             Some(desc) => desc,
             None => return None,
         };
 
-        let removed = self.visit_reads.remove_entry(&read_desc.layer_to_visit_id);
+        let removed = self.layers.remove_entry(&read_desc.layer_id);
 
         match removed {
             Some((
                 _,
-                LayerVisitReads {
+                LayerKeyspace {
                     layer,
                     mut target_keyspace,
                 },
@@ -361,24 +351,20 @@ impl LayerFringe {
         keyspace: KeySpace,
         lsn_range: Range<Lsn>,
     ) {
-        let layer_to_visit_id = LayerToVisitId {
+        let layer_id = layer.id();
-            layer_id: layer.id(),
+        let entry = self.layers.entry(layer_id.clone());
-            lsn_floor: lsn_range.start,
-        };
-
-        let entry = self.visit_reads.entry(layer_to_visit_id.clone());
         match entry {
             Entry::Occupied(mut entry) => {
                 entry.get_mut().target_keyspace.add_keyspace(keyspace);
             }
             Entry::Vacant(entry) => {
-                self.planned_visits_by_lsn.push(LayerVisit {
+                self.planned_reads_by_lsn.push(ReadDesc {
                     lsn_range,
-                    layer_to_visit_id: layer_to_visit_id.clone(),
+                    layer_id: layer_id.clone(),
                 });
                 let mut accum = KeySpaceRandomAccum::new();
                 accum.add_keyspace(keyspace);
-                entry.insert(LayerVisitReads {
+                entry.insert(LayerKeyspace {
                     layer,
                     target_keyspace: accum,
                 });
@@ -393,7 +379,7 @@ impl Default for LayerFringe {
     }
 }
 
-impl Ord for LayerVisit {
+impl Ord for ReadDesc {
     fn cmp(&self, other: &Self) -> Ordering {
         let ord = self.lsn_range.end.cmp(&other.lsn_range.end);
         if ord == std::cmp::Ordering::Equal {
@@ -404,19 +390,19 @@ impl Ord for LayerVisit {
     }
 }
 
-impl PartialOrd for LayerVisit {
+impl PartialOrd for ReadDesc {
     fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
         Some(self.cmp(other))
     }
 }
 
-impl PartialEq for LayerVisit {
+impl PartialEq for ReadDesc {
     fn eq(&self, other: &Self) -> bool {
         self.lsn_range == other.lsn_range
     }
 }
 
-impl Eq for LayerVisit {}
+impl Eq for ReadDesc {}
 
 impl ReadableLayer {
     pub(crate) fn id(&self) -> LayerId {

pageserver/src/tenant/storage_layer/layer.rs

@@ -439,30 +439,11 @@ impl Layer {
 
     fn record_access(&self, ctx: &RequestContext) {
         if self.0.access_stats.record_access(ctx) {
-            // Visibility was modified to Visible: maybe log about this
+            // Visibility was modified to Visible
-            match ctx.task_kind() {
+            tracing::info!(
-                TaskKind::CalculateSyntheticSize
+                "Layer {} became visible as a result of access",
-                | TaskKind::GarbageCollector
+                self.0.desc.key()
-                | TaskKind::MgmtRequest => {
+            );
-                    // This situation is expected in code paths do binary searches of the LSN space to resolve
-                    // an LSN to a timestamp, which happens during GC, during GC cutoff calculations in synthetic size,
-                    // and on-demand for certain HTTP API requests.
-                }
-                _ => {
-                    // In all other contexts, it is unusual to do I/O involving layers which are not visible at
-                    // some branch tip, so we log the fact that we are accessing something that the visibility
-                    // calculation thought should not be visible.
-                    //
-                    // This case is legal in brief time windows: for example an in-flight getpage request can hold on to a layer object
-                    // which was covered by a concurrent compaction.
-                    tracing::info!(
-                        "Layer {} became visible as a result of access",
-                        self.0.desc.key()
-                    );
-                }
-            }
-
-            // Update the timeline's visible bytes count
             if let Some(tl) = self.0.timeline.upgrade() {
                 tl.metrics
                     .visible_physical_size_gauge

pageserver/src/tenant/storage_layer/layer/tests.rs

@@ -1025,15 +1025,6 @@ fn access_stats() {
     assert_eq!(access_stats.latest_activity(), lowres_time(atime));
     access_stats.set_visibility(LayerVisibilityHint::Visible);
     assert_eq!(access_stats.latest_activity(), lowres_time(atime));
-
-    // Recording access implicitly makes layer visible, if it wasn't already
-    let atime = UNIX_EPOCH + Duration::from_secs(2200000000);
-    access_stats.set_visibility(LayerVisibilityHint::Covered);
-    assert_eq!(access_stats.visibility(), LayerVisibilityHint::Covered);
-    assert!(access_stats.record_access_at(atime));
-    access_stats.set_visibility(LayerVisibilityHint::Visible);
-    assert!(!access_stats.record_access_at(atime));
-    access_stats.set_visibility(LayerVisibilityHint::Visible);
 }
 
 #[test]

pageserver/src/tenant/tasks.rs

@@ -163,6 +163,8 @@ async fn compaction_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
     // How many errors we have seen consequtively
     let mut error_run_count = 0;
 
+    let mut last_throttle_flag_reset_at = Instant::now();
+
     TENANT_TASK_EVENTS.with_label_values(&["start"]).inc();
     async {
         let ctx = RequestContext::todo_child(TaskKind::Compaction, DownloadBehavior::Download);
@@ -189,6 +191,8 @@ async fn compaction_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
                 }
             }
 
+
+
             let sleep_duration;
             if period == Duration::ZERO {
                 #[cfg(not(feature = "testing"))]
@@ -203,18 +207,12 @@ async fn compaction_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
                 };
 
                 // Run compaction
-                let IterationResult { output, elapsed } = iteration
+                let IterationResult { output, elapsed } = iteration.run(tenant.compaction_iteration(&cancel, &ctx)).await;
-                    .run(tenant.compaction_iteration(&cancel, &ctx))
-                    .await;
                 match output {
                     Ok(has_pending_task) => {
                         error_run_count = 0;
                         // schedule the next compaction immediately in case there is a pending compaction task
-                        sleep_duration = if has_pending_task {
+                        sleep_duration = if has_pending_task { Duration::ZERO } else { period };
-                            Duration::ZERO
-                        } else {
-                            period
-                        };
                     }
                     Err(e) => {
                         let wait_duration = backoff::exponential_backoff_duration_seconds(
@@ -235,20 +233,38 @@ async fn compaction_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
                 }
 
                 // the duration is recorded by performance tests by enabling debug in this function
-                tracing::debug!(
+                tracing::debug!(elapsed_ms=elapsed.as_millis(), "compaction iteration complete");
-                    elapsed_ms = elapsed.as_millis(),
-                    "compaction iteration complete"
-                );
             };
 
+
             // Perhaps we did no work and the walredo process has been idle for some time:
             // give it a chance to shut down to avoid leaving walredo process running indefinitely.
-            // TODO: move this to a separate task (housekeeping loop) that isn't affected by the back-off,
-            // so we get some upper bound guarantee on when walredo quiesce / this throttling reporting here happens.
             if let Some(walredo_mgr) = &tenant.walredo_mgr {
                 walredo_mgr.maybe_quiesce(period * 10);
             }
 
+            // TODO: move this (and walredo quiesce) to a separate task that isn't affected by the back-off,
+            // so we get some upper bound guarantee on when walredo quiesce / this throttling reporting here happens.
+            info_span!(parent: None, "timeline_get_throttle", tenant_id=%tenant.tenant_shard_id, shard_id=%tenant.tenant_shard_id.shard_slug()).in_scope(|| {
+                let now = Instant::now();
+                let prev = std::mem::replace(&mut last_throttle_flag_reset_at, now);
+                let Stats { count_accounted, count_throttled, sum_throttled_usecs } = tenant.timeline_get_throttle.reset_stats();
+                if count_throttled == 0 {
+                    return;
+                }
+                let allowed_rps = tenant.timeline_get_throttle.steady_rps();
+                let delta = now - prev;
+                info!(
+                    n_seconds=%format_args!("{:.3}",
+                    delta.as_secs_f64()),
+                    count_accounted,
+                    count_throttled,
+                    sum_throttled_usecs,
+                    allowed_rps=%format_args!("{allowed_rps:.0}"),
+                    "shard was throttled in the last n_seconds"
+                );
+            });
+
             // Sleep
             if tokio::time::timeout(sleep_duration, cancel.cancelled())
                 .await
@@ -330,7 +346,6 @@ async fn gc_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
             RequestContext::todo_child(TaskKind::GarbageCollector, DownloadBehavior::Download);
 
         let mut first = true;
-        tenant.gc_block.set_lsn_lease_deadline(tenant.get_lsn_lease_length());
         loop {
             tokio::select! {
                 _ = cancel.cancelled() => {
@@ -348,6 +363,7 @@ async fn gc_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
                 first = false;
 
                 let delays = async {
+                    delay_by_lease_length(tenant.get_lsn_lease_length(), &cancel).await?;
                     random_init_delay(period, &cancel).await?;
                     Ok::<_, Cancelled>(())
                 };
@@ -421,7 +437,6 @@ async fn gc_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
 async fn ingest_housekeeping_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
     TENANT_TASK_EVENTS.with_label_values(&["start"]).inc();
     async {
-    let mut last_throttle_flag_reset_at = Instant::now();
         loop {
             tokio::select! {
                 _ = cancel.cancelled() => {
@@ -468,29 +483,6 @@ async fn ingest_housekeeping_loop(tenant: Arc<Tenant>, cancel: CancellationToken
                 kind: BackgroundLoopKind::IngestHouseKeeping,
             };
             iteration.run(tenant.ingest_housekeeping()).await;
-
-            // TODO: rename the background loop kind to something more generic, like, tenant housekeeping.
-            // Or just spawn another background loop for this throttle, it's not like it's super costly.
-            info_span!(parent: None, "timeline_get_throttle", tenant_id=%tenant.tenant_shard_id, shard_id=%tenant.tenant_shard_id.shard_slug()).in_scope(|| {
-                let now = Instant::now();
-                let prev = std::mem::replace(&mut last_throttle_flag_reset_at, now);
-                let Stats { count_accounted_start, count_accounted_finish, count_throttled, sum_throttled_usecs} = tenant.timeline_get_throttle.reset_stats();
-                if count_throttled == 0 {
-                    return;
-                }
-                let allowed_rps = tenant.timeline_get_throttle.steady_rps();
-                let delta = now - prev;
-                info!(
-                    n_seconds=%format_args!("{:.3}",
-                    delta.as_secs_f64()),
-                    count_accounted = count_accounted_finish,  // don't break existing log scraping
-                    count_throttled,
-                    sum_throttled_usecs,
-                    count_accounted_start, // log after pre-existing fields to not break existing log scraping
-                    allowed_rps=%format_args!("{allowed_rps:.0}"),
-                    "shard was throttled in the last n_seconds"
-                );
-            });
         }
     }
     .await;
@@ -546,12 +538,28 @@ pub(crate) async fn random_init_delay(
         let mut rng = rand::thread_rng();
         rng.gen_range(Duration::ZERO..=period)
     };
+
     match tokio::time::timeout(d, cancel.cancelled()).await {
         Ok(_) => Err(Cancelled),
         Err(_) => Ok(()),
     }
 }
 
+/// Delays GC by defaul lease length at restart.
+///
+/// We do this as the leases mapping are not persisted to disk. By delaying GC by default
+/// length, we gurantees that all the leases we granted before the restart will expire
+/// when we run GC for the first time after the restart.
+pub(crate) async fn delay_by_lease_length(
+    length: Duration,
+    cancel: &CancellationToken,
+) -> Result<(), Cancelled> {
+    match tokio::time::timeout(length, cancel.cancelled()).await {
+        Ok(_) => Err(Cancelled),
+        Err(_) => Ok(()),
+    }
+}
+
 struct Iteration {
     started_at: Instant,
     period: Duration,

pageserver/src/tenant/throttle.rs

@@ -24,10 +24,8 @@ use crate::{context::RequestContext, task_mgr::TaskKind};
 pub struct Throttle<M: Metric> {
     inner: ArcSwap<Inner>,
     metric: M,
-    /// will be turned into [`Stats::count_accounted_start`]
+    /// will be turned into [`Stats::count_accounted`]
-    count_accounted_start: AtomicU64,
+    count_accounted: AtomicU64,
-    /// will be turned into [`Stats::count_accounted_finish`]
-    count_accounted_finish: AtomicU64,
     /// will be turned into [`Stats::count_throttled`]
     count_throttled: AtomicU64,
     /// will be turned into [`Stats::sum_throttled_usecs`]
@@ -45,21 +43,17 @@ pub struct Observation {
     pub wait_time: Duration,
 }
 pub trait Metric {
-    fn accounting_start(&self);
-    fn accounting_finish(&self);
     fn observe_throttling(&self, observation: &Observation);
 }
 
 /// See [`Throttle::reset_stats`].
 pub struct Stats {
-    /// Number of requests that started [`Throttle::throttle`] calls.
+    // Number of requests that were subject to throttling, i.e., requests of the configured [`Config::task_kinds`].
-    pub count_accounted_start: u64,
+    pub count_accounted: u64,
-    /// Number of requests that finished [`Throttle::throttle`] calls.
+    // Subset of the `accounted` requests that were actually throttled.
-    pub count_accounted_finish: u64,
+    // Note that the numbers are stored as two independent atomics, so, there might be a slight drift.
-    /// Subset of the `accounted` requests that were actually throttled.
-    /// Note that the numbers are stored as two independent atomics, so, there might be a slight drift.
     pub count_throttled: u64,
-    /// Sum of microseconds that throttled requests spent waiting for throttling.
+    // Sum of microseconds that throttled requests spent waiting for throttling.
     pub sum_throttled_usecs: u64,
 }
 
@@ -71,8 +65,7 @@ where
         Self {
             inner: ArcSwap::new(Arc::new(Self::new_inner(config))),
             metric,
-            count_accounted_start: AtomicU64::new(0),
+            count_accounted: AtomicU64::new(0),
-            count_accounted_finish: AtomicU64::new(0),
             count_throttled: AtomicU64::new(0),
             sum_throttled_usecs: AtomicU64::new(0),
         }
@@ -124,13 +117,11 @@ where
     /// This method allows retrieving & resetting that flag.
     /// Useful for periodic reporting.
     pub fn reset_stats(&self) -> Stats {
-        let count_accounted_start = self.count_accounted_start.swap(0, Ordering::Relaxed);
+        let count_accounted = self.count_accounted.swap(0, Ordering::Relaxed);
-        let count_accounted_finish = self.count_accounted_finish.swap(0, Ordering::Relaxed);
         let count_throttled = self.count_throttled.swap(0, Ordering::Relaxed);
         let sum_throttled_usecs = self.sum_throttled_usecs.swap(0, Ordering::Relaxed);
         Stats {
-            count_accounted_start,
+            count_accounted,
-            count_accounted_finish,
             count_throttled,
             sum_throttled_usecs,
         }
@@ -148,12 +139,9 @@ where
         };
         let start = std::time::Instant::now();
 
-        self.metric.accounting_start();
-        self.count_accounted_start.fetch_add(1, Ordering::Relaxed);
         let did_throttle = inner.rate_limiter.acquire(key_count).await;
-        self.count_accounted_finish.fetch_add(1, Ordering::Relaxed);
-        self.metric.accounting_finish();
 
+        self.count_accounted.fetch_add(1, Ordering::Relaxed);
         if did_throttle {
             self.count_throttled.fetch_add(1, Ordering::Relaxed);
             let now = Instant::now();

pageserver/src/tenant/timeline.rs

@@ -196,8 +196,9 @@ fn drop_wlock<T>(rlock: tokio::sync::RwLockWriteGuard<'_, T>) {
 /// The outward-facing resources required to build a Timeline
 pub struct TimelineResources {
     pub remote_client: RemoteTimelineClient,
-    pub timeline_get_throttle:
+    pub timeline_get_throttle: Arc<
-        Arc<crate::tenant::throttle::Throttle<crate::metrics::tenant_throttling::TimelineGet>>,
+        crate::tenant::throttle::Throttle<&'static crate::metrics::tenant_throttling::TimelineGet>,
+    >,
     pub l0_flush_global_state: l0_flush::L0FlushGlobalState,
 }
 
@@ -405,8 +406,9 @@ pub struct Timeline {
     gc_lock: tokio::sync::Mutex<()>,
 
     /// Cloned from [`super::Tenant::timeline_get_throttle`] on construction.
-    timeline_get_throttle:
+    timeline_get_throttle: Arc<
-        Arc<crate::tenant::throttle::Throttle<crate::metrics::tenant_throttling::TimelineGet>>,
+        crate::tenant::throttle::Throttle<&'static crate::metrics::tenant_throttling::TimelineGet>,
+    >,
 
     /// Keep aux directory cache to avoid it's reconstruction on each update
     pub(crate) aux_files: tokio::sync::Mutex<AuxFilesState>,
@@ -4013,7 +4015,6 @@ impl Timeline {
             // partition, so flush it to disk.
             let (desc, path) = image_layer_writer.finish(ctx).await?;
             let image_layer = Layer::finish_creating(self.conf, self, desc, &path)?;
-            info!("created image layer for rel {}", image_layer.local_path());
             Ok(ImageLayerCreationOutcome {
                 image: Some(image_layer),
                 next_start_key: img_range.end,
@@ -4103,10 +4104,6 @@ impl Timeline {
             // partition, so flush it to disk.
             let (desc, path) = image_layer_writer.finish(ctx).await?;
             let image_layer = Layer::finish_creating(self.conf, self, desc, &path)?;
-            info!(
-                "created image layer for metadata {}",
-                image_layer.local_path()
-            );
             Ok(ImageLayerCreationOutcome {
                 image: Some(image_layer),
                 next_start_key: img_range.end,
@@ -4314,9 +4311,7 @@ impl Timeline {
         timer.stop_and_record();
 
         // Creating image layers may have caused some previously visible layers to be covered
-        if !image_layers.is_empty() {
+        self.update_layer_visibility().await?;
-            self.update_layer_visibility().await?;
-        }
 
         Ok(image_layers)
     }
@@ -5378,8 +5373,7 @@ impl Timeline {
     /// Force create an image layer and place it into the layer map.
     ///
     /// DO NOT use this function directly. Use [`Tenant::branch_timeline_test_with_layers`]
-    /// or [`Tenant::create_test_timeline_with_layers`] to ensure all these layers are
+    /// or [`Tenant::create_test_timeline_with_layers`] to ensure all these layers are placed into the layer map in one run.
-    /// placed into the layer map in one run AND be validated.
     #[cfg(test)]
     pub(super) async fn force_create_image_layer(
         self: &Arc<Timeline>,
@@ -5413,7 +5407,7 @@ impl Timeline {
         }
         let (desc, path) = image_layer_writer.finish(ctx).await?;
         let image_layer = Layer::finish_creating(self.conf, self, desc, &path)?;
-        info!("force created image layer {}", image_layer.local_path());
+
         {
             let mut guard = self.layers.write().await;
             guard.open_mut().unwrap().force_insert_layer(image_layer);
@@ -5425,8 +5419,7 @@ impl Timeline {
     /// Force create a delta layer and place it into the layer map.
     ///
     /// DO NOT use this function directly. Use [`Tenant::branch_timeline_test_with_layers`]
-    /// or [`Tenant::create_test_timeline_with_layers`] to ensure all these layers are
+    /// or [`Tenant::create_test_timeline_with_layers`] to ensure all these layers are placed into the layer map in one run.
-    /// placed into the layer map in one run AND be validated.
     #[cfg(test)]
     pub(super) async fn force_create_delta_layer(
         self: &Arc<Timeline>,
@@ -5452,6 +5445,33 @@ impl Timeline {
         if let Some(check_start_lsn) = check_start_lsn {
             assert!(deltas.lsn_range.start >= check_start_lsn);
         }
+        // check if the delta layer does not violate the LSN invariant, the legacy compaction should always produce a batch of
+        // layers of the same start/end LSN, and so should the force inserted layer
+        {
+            /// Checks if a overlaps with b, assume a/b = [start, end).
+            pub fn overlaps_with<T: Ord>(a: &Range<T>, b: &Range<T>) -> bool {
+                !(a.end <= b.start || b.end <= a.start)
+            }
+
+            if deltas.key_range.start.next() != deltas.key_range.end {
+                let guard = self.layers.read().await;
+                let mut invalid_layers =
+                    guard.layer_map()?.iter_historic_layers().filter(|layer| {
+                        layer.is_delta()
+                        && overlaps_with(&layer.lsn_range, &deltas.lsn_range)
+                        && layer.lsn_range != deltas.lsn_range
+                        // skip single-key layer files
+                        && layer.key_range.start.next() != layer.key_range.end
+                    });
+                if let Some(layer) = invalid_layers.next() {
+                    // If a delta layer overlaps with another delta layer AND their LSN range is not the same, panic
+                    panic!(
+                        "inserted layer violates delta layer LSN invariant: current_lsn_range={}..{}, conflict_lsn_range={}..{}",
+                        deltas.lsn_range.start, deltas.lsn_range.end, layer.lsn_range.start, layer.lsn_range.end
+                    );
+                }
+            }
+        }
         let mut delta_layer_writer = DeltaLayerWriter::new(
             self.conf,
             self.timeline_id,
@@ -5466,7 +5486,7 @@ impl Timeline {
         }
         let (desc, path) = delta_layer_writer.finish(deltas.key_range.end, ctx).await?;
         let delta_layer = Layer::finish_creating(self.conf, self, desc, &path)?;
-        info!("force created delta layer {}", delta_layer.local_path());
+
         {
             let mut guard = self.layers.write().await;
             guard.open_mut().unwrap().force_insert_layer(delta_layer);

pageserver/src/tenant/timeline/compaction.rs

@@ -29,7 +29,6 @@ use utils::id::TimelineId;
 
 use crate::context::{AccessStatsBehavior, RequestContext, RequestContextBuilder};
 use crate::page_cache;
-use crate::tenant::checks::check_valid_layermap;
 use crate::tenant::remote_timeline_client::WaitCompletionError;
 use crate::tenant::storage_layer::merge_iterator::MergeIterator;
 use crate::tenant::storage_layer::split_writer::{
@@ -1789,12 +1788,20 @@ impl Timeline {
                 stat.visit_image_layer(desc.file_size());
             }
         }
-        let layer_names: Vec<crate::tenant::storage_layer::LayerName> = layer_selection
+        for layer in &layer_selection {
-            .iter()
+            let desc = layer.layer_desc();
-            .map(|layer| layer.layer_desc().layer_name())
+            let key_range = &desc.key_range;
-            .collect_vec();
+            if desc.is_delta() && key_range.start.next() != key_range.end {
-        if let Some(err) = check_valid_layermap(&layer_names) {
+                let lsn_range = desc.lsn_range.clone();
-            bail!("cannot run gc-compaction because {}", err);
+                let intersects = lsn_split_point.range(lsn_range).collect_vec();
+                if intersects.len() > 1 {
+                    bail!(
+                        "cannot run gc-compaction because it violates the layer map LSN split assumption: layer {} intersects with LSN [{}]",
+                        desc.key(),
+                        intersects.into_iter().map(|lsn| lsn.to_string()).join(", ")
+                    );
+                }
+            }
         }
         // The maximum LSN we are processing in this compaction loop
         let end_lsn = layer_selection

proxy/src/auth/backend.rs

@@ -163,7 +163,6 @@ impl ComputeUserInfo {
 }
 
 pub(crate) enum ComputeCredentialKeys {
-    #[cfg(any(test, feature = "testing"))]
     Password(Vec<u8>),
     AuthKeys(AuthKeys),
     None,
@@ -294,10 +293,16 @@ async fn auth_quirks(
     // We now expect to see a very specific payload in the place of password.
     let (info, unauthenticated_password) = match user_info.try_into() {
         Err(info) => {
-            let (info, password) =
+            let res = hacks::password_hack_no_authentication(ctx, info, client).await?;
-                hacks::password_hack_no_authentication(ctx, info, client).await?;
+
-            ctx.set_endpoint_id(info.endpoint.clone());
+            ctx.set_endpoint_id(res.info.endpoint.clone());
-            (info, Some(password))
+            let password = match res.keys {
+                ComputeCredentialKeys::Password(p) => p,
+                ComputeCredentialKeys::AuthKeys(_) | ComputeCredentialKeys::None => {
+                    unreachable!("password hack should return a password")
+                }
+            };
+            (res.info, Some(password))
         }
         Ok(info) => (info, None),
     };

proxy/src/auth/backend/hacks.rs

@@ -1,4 +1,6 @@
-use super::{ComputeCredentials, ComputeUserInfo, ComputeUserInfoNoEndpoint};
+use super::{
+    ComputeCredentialKeys, ComputeCredentials, ComputeUserInfo, ComputeUserInfoNoEndpoint,
+};
 use crate::{
     auth::{self, AuthFlow},
     config::AuthenticationConfig,
@@ -61,7 +63,7 @@ pub(crate) async fn password_hack_no_authentication(
     ctx: &RequestMonitoring,
     info: ComputeUserInfoNoEndpoint,
     client: &mut stream::PqStream<Stream<impl AsyncRead + AsyncWrite + Unpin>>,
-) -> auth::Result<(ComputeUserInfo, Vec<u8>)> {
+) -> auth::Result<ComputeCredentials> {
     warn!("project not specified, resorting to the password hack auth flow");
     ctx.set_auth_method(crate::context::AuthMethod::Cleartext);
 
@@ -77,12 +79,12 @@ pub(crate) async fn password_hack_no_authentication(
     info!(project = &*payload.endpoint, "received missing parameter");
 
     // Report tentative success; compute node will check the password anyway.
-    Ok((
+    Ok(ComputeCredentials {
-        ComputeUserInfo {
+        info: ComputeUserInfo {
             user: info.user,
             options: info.options,
             endpoint: payload.endpoint,
         },
-        payload.password,
+        keys: ComputeCredentialKeys::Password(payload.password),
-    ))
+    })
 }

proxy/src/auth/backend/jwt.rs

@@ -25,8 +25,6 @@ const MAX_JWK_BODY_SIZE: usize = 64 * 1024;
 pub(crate) trait FetchAuthRules: Clone + Send + Sync + 'static {
     fn fetch_auth_rules(
         &self,
-        ctx: &RequestMonitoring,
-        endpoint: EndpointId,
         role_name: RoleName,
     ) -> impl Future<Output = anyhow::Result<Vec<AuthRule>>> + Send;
 }
@@ -103,9 +101,7 @@ impl JwkCacheEntryLock {
     async fn renew_jwks<F: FetchAuthRules>(
         &self,
         _permit: JwkRenewalPermit<'_>,
-        ctx: &RequestMonitoring,
         client: &reqwest::Client,
-        endpoint: EndpointId,
         role_name: RoleName,
         auth_rules: &F,
     ) -> anyhow::Result<Arc<JwkCacheEntry>> {
@@ -119,9 +115,7 @@ impl JwkCacheEntryLock {
             }
         }
 
-        let rules = auth_rules
+        let rules = auth_rules.fetch_auth_rules(role_name).await?;
-            .fetch_auth_rules(ctx, endpoint, role_name)
-            .await?;
         let mut key_sets =
             ahash::HashMap::with_capacity_and_hasher(rules.len(), ahash::RandomState::new());
         // TODO(conrad): run concurrently
@@ -172,7 +166,6 @@ impl JwkCacheEntryLock {
         self: &Arc<Self>,
         ctx: &RequestMonitoring,
         client: &reqwest::Client,
-        endpoint: EndpointId,
         role_name: RoleName,
         fetch: &F,
     ) -> Result<Arc<JwkCacheEntry>, anyhow::Error> {
@@ -183,9 +176,7 @@ impl JwkCacheEntryLock {
         let Some(cached) = guard else {
             let _paused = ctx.latency_timer_pause(crate::metrics::Waiting::Compute);
             let permit = self.acquire_permit().await;
-            return self
+            return self.renew_jwks(permit, client, role_name, fetch).await;
-                .renew_jwks(permit, ctx, client, endpoint, role_name, fetch)
-                .await;
         };
 
         let last_update = now.duration_since(cached.last_retrieved);
@@ -196,9 +187,7 @@ impl JwkCacheEntryLock {
             let permit = self.acquire_permit().await;
 
             // it's been too long since we checked the keys. wait for them to update.
-            return self
+            return self.renew_jwks(permit, client, role_name, fetch).await;
-                .renew_jwks(permit, ctx, client, endpoint, role_name, fetch)
-                .await;
         }
 
         // every 5 minutes we should spawn a job to eagerly update the token.
@@ -209,12 +198,8 @@ impl JwkCacheEntryLock {
                 let entry = self.clone();
                 let client = client.clone();
                 let fetch = fetch.clone();
-                let ctx = ctx.clone();
                 tokio::spawn(async move {
-                    if let Err(e) = entry
+                    if let Err(e) = entry.renew_jwks(permit, &client, role_name, &fetch).await {
-                        .renew_jwks(permit, &ctx, &client, endpoint, role_name, &fetch)
-                        .await
-                    {
                         tracing::warn!(error=?e, "could not fetch JWKs in background job");
                     }
                 });
@@ -231,7 +216,6 @@ impl JwkCacheEntryLock {
         ctx: &RequestMonitoring,
         jwt: &str,
         client: &reqwest::Client,
-        endpoint: EndpointId,
         role_name: RoleName,
         fetch: &F,
     ) -> Result<(), anyhow::Error> {
@@ -258,7 +242,7 @@ impl JwkCacheEntryLock {
         let kid = header.key_id.context("missing key id")?;
 
         let mut guard = self
-            .get_or_update_jwk_cache(ctx, client, endpoint.clone(), role_name.clone(), fetch)
+            .get_or_update_jwk_cache(ctx, client, role_name.clone(), fetch)
             .await?;
 
         // get the key from the JWKs if possible. If not, wait for the keys to update.
@@ -270,14 +254,7 @@ impl JwkCacheEntryLock {
 
                     let permit = self.acquire_permit().await;
                     guard = self
-                        .renew_jwks(
+                        .renew_jwks(permit, client, role_name.clone(), fetch)
-                            permit,
-                            ctx,
-                            client,
-                            endpoint.clone(),
-                            role_name.clone(),
-                            fetch,
-                        )
                         .await?;
                 }
                 _ => {
@@ -341,7 +318,7 @@ impl JwkCache {
         jwt: &str,
     ) -> Result<(), anyhow::Error> {
         // try with just a read lock first
-        let key = (endpoint.clone(), role_name.clone());
+        let key = (endpoint, role_name.clone());
         let entry = self.map.get(&key).as_deref().map(Arc::clone);
         let entry = entry.unwrap_or_else(|| {
             // acquire a write lock after to insert.
@@ -350,7 +327,7 @@ impl JwkCache {
         });
 
         entry
-            .check_jwt(ctx, jwt, &self.client, endpoint, role_name, fetch)
+            .check_jwt(ctx, jwt, &self.client, role_name, fetch)
             .await
     }
 }
@@ -711,8 +688,6 @@ X0n5X2/pBLJzxZc62ccvZYVnctBiFs6HbSnxpuMQCfkt/BcR/ttIepBQQIW86wHL
         impl FetchAuthRules for Fetch {
             async fn fetch_auth_rules(
                 &self,
-                _ctx: &RequestMonitoring,
-                _endpoint: EndpointId,
                 _role_name: RoleName,
             ) -> anyhow::Result<Vec<AuthRule>> {
                 Ok(vec![
@@ -731,7 +706,6 @@ X0n5X2/pBLJzxZc62ccvZYVnctBiFs6HbSnxpuMQCfkt/BcR/ttIepBQQIW86wHL
         }
 
         let role_name = RoleName::from("user");
-        let endpoint = EndpointId::from("ep");
 
         let jwk_cache = Arc::new(JwkCacheEntryLock::default());
 
@@ -741,7 +715,6 @@ X0n5X2/pBLJzxZc62ccvZYVnctBiFs6HbSnxpuMQCfkt/BcR/ttIepBQQIW86wHL
                     &RequestMonitoring::test(),
                     &token,
                     &client,
-                    endpoint.clone(),
                     role_name.clone(),
                     &Fetch(addr),
                 )

proxy/src/auth/backend/local.rs

@@ -9,9 +9,8 @@ use crate::{
         messages::{ColdStartInfo, EndpointJwksResponse, MetricsAuxInfo},
         NodeInfo,
     },
-    context::RequestMonitoring,
     intern::{BranchIdInt, BranchIdTag, EndpointIdTag, InternId, ProjectIdInt, ProjectIdTag},
-    EndpointId, RoleName,
+    RoleName,
 };
 
 use super::jwt::{AuthRule, FetchAuthRules, JwkCache};
@@ -58,12 +57,7 @@ pub struct JwksRoleSettings {
 }
 
 impl FetchAuthRules for StaticAuthRules {
-    async fn fetch_auth_rules(
+    async fn fetch_auth_rules(&self, role_name: RoleName) -> anyhow::Result<Vec<AuthRule>> {
-        &self,
-        _ctx: &RequestMonitoring,
-        _endpoint: EndpointId,
-        role_name: RoleName,
-    ) -> anyhow::Result<Vec<AuthRule>> {
         let mappings = JWKS_ROLE_MAP.load();
         let role_mappings = mappings
             .as_deref()

proxy/src/console/provider.rs

@@ -303,7 +303,6 @@ impl NodeInfo {
 
     pub(crate) fn set_keys(&mut self, keys: &ComputeCredentialKeys) {
         match keys {
-            #[cfg(any(test, feature = "testing"))]
             ComputeCredentialKeys::Password(password) => self.config.password(password),
             ComputeCredentialKeys::AuthKeys(auth_keys) => self.config.auth_keys(*auth_keys),
             ComputeCredentialKeys::None => &mut self.config,

proxy/src/context.rs

@@ -79,40 +79,6 @@ pub(crate) enum AuthMethod {
     Cleartext,
 }
 
-impl Clone for RequestMonitoring {
-    fn clone(&self) -> Self {
-        let inner = self.0.try_lock().expect("should not deadlock");
-        let new = RequestMonitoringInner {
-            peer_addr: inner.peer_addr,
-            session_id: inner.session_id,
-            protocol: inner.protocol,
-            first_packet: inner.first_packet,
-            region: inner.region,
-            span: info_span!("background_task"),
-
-            project: inner.project,
-            branch: inner.branch,
-            endpoint_id: inner.endpoint_id.clone(),
-            dbname: inner.dbname.clone(),
-            user: inner.user.clone(),
-            application: inner.application.clone(),
-            error_kind: inner.error_kind,
-            auth_method: inner.auth_method.clone(),
-            success: inner.success,
-            rejected: inner.rejected,
-            cold_start_info: inner.cold_start_info,
-            pg_options: inner.pg_options.clone(),
-
-            sender: None,
-            disconnect_sender: None,
-            latency_timer: LatencyTimer::noop(inner.protocol),
-            disconnect_timestamp: inner.disconnect_timestamp,
-        };
-
-        Self(TryLock::new(new))
-    }
-}
-
 impl RequestMonitoring {
     pub fn new(
         session_id: Uuid,

proxy/src/metrics.rs

@@ -397,8 +397,6 @@ pub struct LatencyTimer {
     protocol: Protocol,
     cold_start_info: ColdStartInfo,
     outcome: ConnectOutcome,
-
-    skip_reporting: bool,
 }
 
 impl LatencyTimer {
@@ -411,20 +409,6 @@ impl LatencyTimer {
             cold_start_info: ColdStartInfo::Unknown,
             // assume failed unless otherwise specified
             outcome: ConnectOutcome::Failed,
-            skip_reporting: false,
-        }
-    }
-
-    pub(crate) fn noop(protocol: Protocol) -> Self {
-        Self {
-            start: time::Instant::now(),
-            stop: None,
-            accumulated: Accumulated::default(),
-            protocol,
-            cold_start_info: ColdStartInfo::Unknown,
-            // assume failed unless otherwise specified
-            outcome: ConnectOutcome::Failed,
-            skip_reporting: true,
         }
     }
 
@@ -459,10 +443,6 @@ pub enum ConnectOutcome {
 
 impl Drop for LatencyTimer {
     fn drop(&mut self) {
-        if self.skip_reporting {
-            return;
-        }
-
         let duration = self
             .stop
             .unwrap_or_else(time::Instant::now)

proxy/src/serverless/backend.rs

@@ -27,7 +27,7 @@ use crate::{
     Host,
 };
 
-use super::conn_pool::{poll_client, Client, ConnInfo, GlobalConnPool};
+use super::conn_pool::{poll_client, AuthData, Client, ConnInfo, GlobalConnPool};
 
 pub(crate) struct PoolingBackend {
     pub(crate) pool: Arc<GlobalConnPool<tokio_postgres::Client>>,
@@ -274,6 +274,13 @@ impl ConnectMechanism for TokioMechanism {
             .dbname(&self.conn_info.dbname)
             .connect_timeout(timeout);
 
+        match &self.conn_info.auth {
+            AuthData::Jwt(_) => {}
+            AuthData::Password(pw) => {
+                config.password(pw);
+            }
+        }
+
         let pause = ctx.latency_timer_pause(crate::metrics::Waiting::Compute);
         let res = config.connect(tokio_postgres::NoTls).await;
         drop(pause);

proxy/src/serverless/conn_pool.rs

@@ -29,16 +29,11 @@ use tracing::{info, info_span, Instrument};
 
 use super::backend::HttpConnError;
 
-#[derive(Debug, Clone)]
-pub(crate) struct ConnInfoWithAuth {
-    pub(crate) conn_info: ConnInfo,
-    pub(crate) auth: AuthData,
-}
-
 #[derive(Debug, Clone)]
 pub(crate) struct ConnInfo {
     pub(crate) user_info: ComputeUserInfo,
     pub(crate) dbname: DbName,
+    pub(crate) auth: AuthData,
 }
 
 #[derive(Debug, Clone)]
@@ -792,6 +787,7 @@ mod tests {
                 options: NeonOptions::default(),
             },
             dbname: "dbname".into(),
+            auth: AuthData::Password("password".as_bytes().into()),
         };
         let ep_pool = Arc::downgrade(
             &pool.get_or_create_endpoint_pool(&conn_info.endpoint_cache_key().unwrap()),
@@ -849,6 +845,7 @@ mod tests {
                 options: NeonOptions::default(),
             },
             dbname: "dbname".into(),
+            auth: AuthData::Password("password".as_bytes().into()),
         };
         let ep_pool = Arc::downgrade(
             &pool.get_or_create_endpoint_pool(&conn_info.endpoint_cache_key().unwrap()),

proxy/src/serverless/sql_over_http.rs

@@ -60,7 +60,6 @@ use super::backend::PoolingBackend;
 use super::conn_pool::AuthData;
 use super::conn_pool::Client;
 use super::conn_pool::ConnInfo;
-use super::conn_pool::ConnInfoWithAuth;
 use super::http_util::json_response;
 use super::json::json_to_pg_text;
 use super::json::pg_text_row_to_json;
@@ -149,7 +148,7 @@ fn get_conn_info(
     ctx: &RequestMonitoring,
     headers: &HeaderMap,
     tls: Option<&TlsConfig>,
-) -> Result<ConnInfoWithAuth, ConnInfoError> {
+) -> Result<ConnInfo, ConnInfoError> {
     // HTTP only uses cleartext (for now and likely always)
     ctx.set_auth_method(crate::context::AuthMethod::Cleartext);
 
@@ -236,8 +235,11 @@ fn get_conn_info(
         options: options.unwrap_or_default(),
     };
 
-    let conn_info = ConnInfo { user_info, dbname };
+    Ok(ConnInfo {
-    Ok(ConnInfoWithAuth { conn_info, auth })
+        user_info,
+        dbname,
+        auth,
+    })
 }
 
 // TODO: return different http error codes
@@ -521,10 +523,7 @@ async fn handle_inner(
 
     // TLS config should be there.
     let conn_info = get_conn_info(ctx, headers, config.tls_config.as_ref())?;
-    info!(
+    info!(user = conn_info.user_info.user.as_str(), "credentials");
-        user = conn_info.conn_info.user_info.user.as_str(),
-        "credentials"
-    );
 
     // Allow connection pooling only if explicitly requested
     // or if we have decided that http pool is no longer opt-in
@@ -569,20 +568,20 @@ async fn handle_inner(
                         .authenticate_with_password(
                             ctx,
                             &config.authentication_config,
-                            &conn_info.conn_info.user_info,
+                            &conn_info.user_info,
                             pw,
                         )
                         .await?
                 }
                 AuthData::Jwt(jwt) => {
                     backend
-                        .authenticate_with_jwt(ctx, &conn_info.conn_info.user_info, jwt)
+                        .authenticate_with_jwt(ctx, &conn_info.user_info, jwt)
                         .await?
                 }
             };
 
             let client = backend
-                .connect_to_compute(ctx, conn_info.conn_info, keys, !allow_pool)
+                .connect_to_compute(ctx, conn_info, keys, !allow_pool)
                 .await?;
             // not strictly necessary to mark success here,
             // but it's just insurance for if we forget it somewhere else

safekeeper/src/debug_dump.rs

@@ -17,7 +17,6 @@ use postgres_ffi::MAX_SEND_SIZE;
 use serde::Deserialize;
 use serde::Serialize;
 
-use postgres_ffi::v14::xlog_utils::{IsPartialXLogFileName, IsXLogFileName};
 use sha2::{Digest, Sha256};
 use utils::id::NodeId;
 use utils::id::TenantTimelineId;
@@ -52,9 +51,6 @@ pub struct Args {
     /// Dump full term history. True by default.
     pub dump_term_history: bool,
 
-    /// Dump last modified time of WAL segments. Uses value of `dump_all` by default.
-    pub dump_wal_last_modified: bool,
-
     /// Filter timelines by tenant_id.
     pub tenant_id: Option<TenantId>,
 
@@ -132,19 +128,12 @@ async fn build_from_tli_dump(
         None
     };
 
-    let wal_last_modified = if args.dump_wal_last_modified {
-        get_wal_last_modified(timeline_dir).ok().flatten()
-    } else {
-        None
-    };
-
     Timeline {
         tenant_id: timeline.ttid.tenant_id,
         timeline_id: timeline.ttid.timeline_id,
         control_file,
         memory,
         disk_content,
-        wal_last_modified,
     }
 }
 
@@ -167,7 +156,6 @@ pub struct Timeline {
     pub control_file: Option<TimelinePersistentState>,
     pub memory: Option<Memory>,
     pub disk_content: Option<DiskContent>,
-    pub wal_last_modified: Option<DateTime<Utc>>,
 }
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -314,27 +302,6 @@ fn build_file_info(entry: DirEntry) -> Result<FileInfo> {
     })
 }
 
-/// Get highest modified time of WAL segments in the directory.
-fn get_wal_last_modified(path: &Utf8Path) -> Result<Option<DateTime<Utc>>> {
-    let mut res = None;
-    for entry in fs::read_dir(path)? {
-        if entry.is_err() {
-            continue;
-        }
-        let entry = entry?;
-        /* Ignore files that are not XLOG segments */
-        let fname = entry.file_name();
-        if !IsXLogFileName(&fname) && !IsPartialXLogFileName(&fname) {
-            continue;
-        }
-
-        let metadata = entry.metadata()?;
-        let modified: DateTime<Utc> = DateTime::from(metadata.modified()?);
-        res = std::cmp::max(res, Some(modified));
-    }
-    Ok(res)
-}
-
 /// Converts SafeKeeperConf to Config, filtering out the fields that are not
 /// supposed to be exposed.
 fn build_config(config: SafeKeeperConf) -> Config {

safekeeper/src/http/openapi_spec.yaml

@@ -1,11 +1,7 @@
 openapi: "3.0.2"
 info:
   title: Safekeeper control API
-  description: Neon Safekeeper API
   version: "1.0"
-  license:
-    name: "Apache"
-    url: https://github.com/neondatabase/neon/blob/main/LICENSE
 
 
 servers:
@@ -390,12 +386,6 @@ components:
         msg:
           type: string
 
-    NotFoundError:
-      type: object
-      properties:
-        msg:
-          type: string
-
   responses:
 
     #

safekeeper/src/http/routes.rs

@@ -481,7 +481,6 @@ async fn dump_debug_handler(mut request: Request<Body>) -> Result<Response<Body>
     let mut dump_memory: Option<bool> = None;
     let mut dump_disk_content: Option<bool> = None;
     let mut dump_term_history: Option<bool> = None;
-    let mut dump_wal_last_modified: Option<bool> = None;
     let mut tenant_id: Option<TenantId> = None;
     let mut timeline_id: Option<TimelineId> = None;
 
@@ -495,7 +494,6 @@ async fn dump_debug_handler(mut request: Request<Body>) -> Result<Response<Body>
             "dump_memory" => dump_memory = Some(parse_kv_str(&k, &v)?),
             "dump_disk_content" => dump_disk_content = Some(parse_kv_str(&k, &v)?),
             "dump_term_history" => dump_term_history = Some(parse_kv_str(&k, &v)?),
-            "dump_wal_last_modified" => dump_wal_last_modified = Some(parse_kv_str(&k, &v)?),
             "tenant_id" => tenant_id = Some(parse_kv_str(&k, &v)?),
             "timeline_id" => timeline_id = Some(parse_kv_str(&k, &v)?),
             _ => Err(ApiError::BadRequest(anyhow::anyhow!(
@@ -510,7 +508,6 @@ async fn dump_debug_handler(mut request: Request<Body>) -> Result<Response<Body>
     let dump_memory = dump_memory.unwrap_or(dump_all);
     let dump_disk_content = dump_disk_content.unwrap_or(dump_all);
     let dump_term_history = dump_term_history.unwrap_or(true);
-    let dump_wal_last_modified = dump_wal_last_modified.unwrap_or(dump_all);
 
     let args = debug_dump::Args {
         dump_all,
@@ -518,7 +515,6 @@ async fn dump_debug_handler(mut request: Request<Body>) -> Result<Response<Body>
         dump_memory,
         dump_disk_content,
         dump_term_history,
-        dump_wal_last_modified,
         tenant_id,
         timeline_id,
     };

safekeeper/src/wal_storage.rs

@@ -539,17 +539,20 @@ async fn remove_segments_from_disk(
     while let Some(entry) = entries.next_entry().await? {
         let entry_path = entry.path();
         let fname = entry_path.file_name().unwrap();
-        /* Ignore files that are not XLOG segments */
+
-        if !IsXLogFileName(fname) && !IsPartialXLogFileName(fname) {
+        if let Some(fname_str) = fname.to_str() {
-            continue;
+            /* Ignore files that are not XLOG segments */
-        }
+            if !IsXLogFileName(fname_str) && !IsPartialXLogFileName(fname_str) {
-        let (segno, _) = XLogFromFileName(fname, wal_seg_size)?;
+                continue;
-        if remove_predicate(segno) {
+            }
-            remove_file(entry_path).await?;
+            let (segno, _) = XLogFromFileName(fname_str, wal_seg_size);
-            n_removed += 1;
+            if remove_predicate(segno) {
-            min_removed = min(min_removed, segno);
+                remove_file(entry_path).await?;
-            max_removed = max(max_removed, segno);
+                n_removed += 1;
-            REMOVED_WAL_SEGMENTS.inc();
+                min_removed = min(min_removed, segno);
+                max_removed = max(max_removed, segno);
+                REMOVED_WAL_SEGMENTS.inc();
+            }
         }
     }
 

storage_controller/src/service.rs

@@ -3,7 +3,6 @@ use std::{
     borrow::Cow,
     cmp::Ordering,
     collections::{BTreeMap, HashMap, HashSet},
-    error::Error,
     ops::Deref,
     path::PathBuf,
     str::FromStr,
@@ -219,16 +218,9 @@ fn passthrough_api_error(node: &Node, e: mgmt_api::Error) -> ApiError {
                 format!("{node} error receiving error body: {str}").into(),
             )
         }
-        mgmt_api::Error::ReceiveBody(err) if err.is_decode() => {
+        mgmt_api::Error::ReceiveBody(str) => {
-            // Return 500 for decoding errors.
+            // Presume errors receiving body are connectivity/availability issues
-            ApiError::InternalServerError(anyhow::Error::from(err).context("error decoding body"))
+            ApiError::ResourceUnavailable(format!("{node} error receiving body: {str}").into())
-        }
-        mgmt_api::Error::ReceiveBody(err) => {
-            // Presume errors receiving body are connectivity/availability issues except for decoding errors
-            let src_str = err.source().map(|e| e.to_string()).unwrap_or_default();
-            ApiError::ResourceUnavailable(
-                format!("{node} error receiving error body: {err} {}", src_str).into(),
-            )
         }
         mgmt_api::Error::ApiError(StatusCode::NOT_FOUND, msg) => {
             ApiError::NotFound(anyhow::anyhow!(format!("{node}: {msg}")).into())

storage_scrubber/src/checks.rs

@@ -1,8 +1,7 @@
-use std::collections::{HashMap, HashSet};
+use std::collections::{BTreeSet, HashMap, HashSet};
 
 use anyhow::Context;
 use itertools::Itertools;
-use pageserver::tenant::checks::check_valid_layermap;
 use pageserver::tenant::layer_map::LayerMap;
 use pageserver::tenant::remote_timeline_client::index::LayerFileMetadata;
 use pageserver_api::shard::ShardIndex;
@@ -49,6 +48,56 @@ impl TimelineAnalysis {
     }
 }
 
+/// Checks whether a layer map is valid (i.e., is a valid result of the current compaction algorithm if nothing goes wrong).
+/// The function checks if we can split the LSN range of a delta layer only at the LSNs of the delta layers. For example,
+///
+/// ```plain
+/// |       |                 |       |
+/// |   1   |    |   2   |    |   3   |
+/// |       |    |       |    |       |
+/// ```
+///
+/// This is not a valid layer map because the LSN range of layer 1 intersects with the LSN range of layer 2. 1 and 2 should have
+/// the same LSN range.
+///
+/// The exception is that when layer 2 only contains a single key, it could be split over the LSN range. For example,
+///
+/// ```plain
+/// |       |    |   2   |    |       |
+/// |   1   |    |-------|    |   3   |
+/// |       |    |   4   |    |       |
+///
+/// If layer 2 and 4 contain the same single key, this is also a valid layer map.
+fn check_valid_layermap(metadata: &HashMap<LayerName, LayerFileMetadata>) -> Option<String> {
+    let mut lsn_split_point = BTreeSet::new(); // TODO: use a better data structure (range tree / range set?)
+    let mut all_delta_layers = Vec::new();
+    for (name, _) in metadata.iter() {
+        if let LayerName::Delta(layer) = name {
+            if layer.key_range.start.next() != layer.key_range.end {
+                all_delta_layers.push(layer.clone());
+            }
+        }
+    }
+    for layer in &all_delta_layers {
+        let lsn_range = &layer.lsn_range;
+        lsn_split_point.insert(lsn_range.start);
+        lsn_split_point.insert(lsn_range.end);
+    }
+    for layer in &all_delta_layers {
+        let lsn_range = layer.lsn_range.clone();
+        let intersects = lsn_split_point.range(lsn_range).collect_vec();
+        if intersects.len() > 1 {
+            let err = format!(
+                        "layer violates the layer map LSN split assumption: layer {} intersects with LSN [{}]",
+                        layer,
+                        intersects.into_iter().map(|lsn| lsn.to_string()).join(", ")
+                    );
+            return Some(err);
+        }
+    }
+    None
+}
+
 pub(crate) async fn branch_cleanup_and_check_errors(
     remote_client: &GenericRemoteStorage,
     id: &TenantShardTimelineId,
@@ -128,8 +177,7 @@ pub(crate) async fn branch_cleanup_and_check_errors(
                         }
                     }
 
-                    let layer_names = index_part.layer_metadata.keys().cloned().collect_vec();
+                    if let Some(err) = check_valid_layermap(&index_part.layer_metadata) {
-                    if let Some(err) = check_valid_layermap(&layer_names) {
                         result.errors.push(format!(
                             "index_part.json contains invalid layer map structure: {err}"
                         ));

test_runner/fixtures/metrics.py

@@ -102,11 +102,6 @@ def histogram(prefix_without_trailing_underscore: str) -> List[str]:
     return [f"{prefix_without_trailing_underscore}_{x}" for x in ["bucket", "count", "sum"]]
 
 
-def counter(name: str) -> str:
-    # the prometheus_client package appends _total to all counters client-side
-    return f"{name}_total"
-
-
 PAGESERVER_PER_TENANT_REMOTE_TIMELINE_CLIENT_METRICS: Tuple[str, ...] = (
     "pageserver_remote_timeline_client_calls_started_total",
     "pageserver_remote_timeline_client_calls_finished_total",
@@ -137,14 +132,9 @@ PAGESERVER_GLOBAL_METRICS: Tuple[str, ...] = (
     *histogram("pageserver_wait_lsn_seconds"),
     *histogram("pageserver_remote_operation_seconds"),
     *histogram("pageserver_io_operations_seconds"),
-    "pageserver_smgr_query_started_global_count_total",
     "pageserver_tenant_states_count",
     "pageserver_circuit_breaker_broken_total",
     "pageserver_circuit_breaker_unbroken_total",
-    counter("pageserver_tenant_throttling_count_accounted_start_global"),
-    counter("pageserver_tenant_throttling_count_accounted_finish_global"),
-    counter("pageserver_tenant_throttling_wait_usecs_sum_global"),
-    counter("pageserver_tenant_throttling_count_global"),
 )
 
 PAGESERVER_PER_TENANT_METRICS: Tuple[str, ...] = (
@@ -156,7 +146,6 @@ PAGESERVER_PER_TENANT_METRICS: Tuple[str, ...] = (
     "pageserver_smgr_query_seconds_bucket",
     "pageserver_smgr_query_seconds_count",
     "pageserver_smgr_query_seconds_sum",
-    "pageserver_smgr_query_started_count_total",
     "pageserver_archive_size",
     "pageserver_pitr_history_size",
     "pageserver_layer_bytes",
@@ -168,10 +157,6 @@ PAGESERVER_PER_TENANT_METRICS: Tuple[str, ...] = (
     "pageserver_evictions_with_low_residence_duration_total",
     "pageserver_aux_file_estimated_size",
     "pageserver_valid_lsn_lease_count",
-    counter("pageserver_tenant_throttling_count_accounted_start"),
-    counter("pageserver_tenant_throttling_count_accounted_finish"),
-    counter("pageserver_tenant_throttling_wait_usecs_sum"),
-    counter("pageserver_tenant_throttling_count"),
     *PAGESERVER_PER_TENANT_REMOTE_TIMELINE_CLIENT_METRICS,
     # "pageserver_directory_entries_count", -- only used if above a certain threshold
     # "pageserver_broken_tenants_count" -- used only for broken

test_runner/fixtures/neon_fixtures.py

@@ -181,17 +181,6 @@ def top_output_dir(base_dir: Path) -> Iterator[Path]:
     log.info(f"top_output_dir is {output_dir}")
     yield output_dir
 
-@pytest.fixture(scope="session", autouse=True)
-def shared_initdb_cache_dir(top_output_dir: Path) -> Iterator[Path]:
-    log.info("Creating shared initdb cache directory")
-
-    cache_dir = top_output_dir / "shared_initdb_cache"
-
-    shutil.rmtree(cache_dir, ignore_errors=True)
-    cache_dir.mkdir(exist_ok=True)
-
-    yield cache_dir
-
 
 @pytest.fixture(scope="session")
 def neon_api_key() -> str:
@@ -412,7 +401,6 @@ class NeonEnvBuilder:
         safekeeper_extra_opts: Optional[list[str]] = None,
         storage_controller_port_override: Optional[int] = None,
         pageserver_io_buffer_alignment: Optional[int] = None,
-        shared_initdb_cache_dir: Optional[Path] = None,
     ):
         self.repo_dir = repo_dir
         self.rust_log_override = rust_log_override
@@ -444,7 +432,6 @@ class NeonEnvBuilder:
         self.enable_scrub_on_exit = True
         self.test_output_dir = test_output_dir
         self.test_overlay_dir = test_overlay_dir
-        self.shared_initdb_cache_dir = shared_initdb_cache_dir
         self.overlay_mounts_created_by_us: List[Tuple[str, Path]] = []
         self.config_init_force: Optional[str] = None
         self.top_output_dir = top_output_dir
@@ -982,7 +969,6 @@ class NeonEnv:
 
     def __init__(self, config: NeonEnvBuilder):
         self.repo_dir = config.repo_dir
-        self.shared_initdb_cache_dir = config.shared_initdb_cache_dir
         self.rust_log_override = config.rust_log_override
         self.port_distributor = config.port_distributor
         self.s3_mock_server = config.mock_s3_server
@@ -1085,10 +1071,6 @@ class NeonEnv:
                 # Default which can be overriden with `NeonEnvBuilder.pageserver_config_override`
                 "availability_zone": "us-east-2a",
             }
-
-            if self.shared_initdb_cache_dir is not None:
-                ps_cfg["initdb_cache_dir"] = str(self.shared_initdb_cache_dir)
-
             if self.pageserver_virtual_file_io_engine is not None:
                 ps_cfg["virtual_file_io_engine"] = self.pageserver_virtual_file_io_engine
             if config.pageserver_default_tenant_config_compaction_algorithm is not None:
@@ -1351,7 +1333,6 @@ def neon_simple_env(
     pageserver_aux_file_policy: Optional[AuxFileStore],
     pageserver_default_tenant_config_compaction_algorithm: Optional[Dict[str, Any]],
     pageserver_io_buffer_alignment: Optional[int],
-    shared_initdb_cache_dir: Optional[Path],
 ) -> Iterator[NeonEnv]:
     """
     Simple Neon environment, with no authentication and no safekeepers.
@@ -1378,7 +1359,6 @@ def neon_simple_env(
         pageserver_aux_file_policy=pageserver_aux_file_policy,
         pageserver_default_tenant_config_compaction_algorithm=pageserver_default_tenant_config_compaction_algorithm,
         pageserver_io_buffer_alignment=pageserver_io_buffer_alignment,
-        shared_initdb_cache_dir=shared_initdb_cache_dir
     ) as builder:
         env = builder.init_start()
 
@@ -1403,7 +1383,6 @@ def neon_env_builder(
     pageserver_aux_file_policy: Optional[AuxFileStore],
     record_property: Callable[[str, object], None],
     pageserver_io_buffer_alignment: Optional[int],
-    shared_initdb_cache_dir: Optional[Path],
 ) -> Iterator[NeonEnvBuilder]:
     """
     Fixture to create a Neon environment for test.
@@ -1439,7 +1418,6 @@ def neon_env_builder(
         pageserver_aux_file_policy=pageserver_aux_file_policy,
         pageserver_default_tenant_config_compaction_algorithm=pageserver_default_tenant_config_compaction_algorithm,
         pageserver_io_buffer_alignment=pageserver_io_buffer_alignment,
-        shared_initdb_cache_dir=shared_initdb_cache_dir
     ) as builder:
         yield builder
         # Propogate `preserve_database_files` to make it possible to use in other fixtures,
@@ -3885,6 +3863,9 @@ def static_proxy(
     dbname = vanilla_pg.default_options["dbname"]
     auth_endpoint = f"postgres://proxy:password@{host}:{port}/{dbname}"
 
+    # require password for 'http_auth' user
+    vanilla_pg.edit_hba([f"host {dbname} http_auth {host} password"])
+
     # For simplicity, we use the same user for both `--auth-endpoint` and `safe_psql`
     vanilla_pg.start()
     vanilla_pg.safe_psql("create user proxy with login superuser password 'password'")

test_runner/regress/test_branch_and_gc.py

@@ -142,7 +142,6 @@ def test_branch_creation_before_gc(neon_simple_env: NeonEnv):
             "image_creation_threshold": "1",
             # set PITR interval to be small, so we can do GC
             "pitr_interval": "0 s",
-            "lsn_lease_length": "0s",
         }
     )
 

test_runner/regress/test_branch_behind.py

@@ -11,9 +11,7 @@ from fixtures.utils import print_gc_result, query_scalar
 #
 def test_branch_behind(neon_env_builder: NeonEnvBuilder):
     # Disable pitr, because here we want to test branch creation after GC
-    env = neon_env_builder.init_start(
+    env = neon_env_builder.init_start(initial_tenant_conf={"pitr_interval": "0 sec"})
-        initial_tenant_conf={"pitr_interval": "0 sec", "lsn_lease_length": "0s"}
-    )
 
     error_regexes = [
         ".*invalid branch start lsn.*",

test_runner/regress/test_branching.py

@@ -419,7 +419,7 @@ def test_duplicate_creation(neon_env_builder: NeonEnvBuilder):
 
 
 def test_branching_while_stuck_find_gc_cutoffs(neon_env_builder: NeonEnvBuilder):
-    env = neon_env_builder.init_start(initial_tenant_conf={"lsn_lease_length": "0s"})
+    env = neon_env_builder.init_start()
 
     client = env.pageserver.http_client()
 

test_runner/regress/test_compaction.py

@@ -240,7 +240,6 @@ def test_uploads_and_deletions(
         "image_creation_threshold": "1",
         "image_layer_creation_check_threshold": "0",
         "compaction_algorithm": json.dumps({"kind": compaction_algorithm.value}),
-        "lsn_lease_length": "0s",
     }
     env = neon_env_builder.init_start(initial_tenant_conf=tenant_conf)
 

test_runner/regress/test_hot_standby.py

@@ -222,7 +222,7 @@ def pgbench_accounts_initialized(ep):
 # Without hs feedback enabled we'd see 'User query might have needed to see row
 # versions that must be removed.' errors.
 def test_hot_standby_feedback(neon_env_builder: NeonEnvBuilder, pg_bin: PgBin):
-    env = neon_env_builder.init_start(initial_tenant_conf={"lsn_lease_length": "0s"})
+    env = neon_env_builder.init_start()
     agressive_vacuum_conf = [
         "log_autovacuum_min_duration = 0",
         "autovacuum_naptime = 10s",

test_runner/regress/test_layer_eviction.py

@@ -173,7 +173,6 @@ def test_gc_of_remote_layers(neon_env_builder: NeonEnvBuilder):
         # "image_creation_threshold": set at runtime
         "compaction_target_size": f"{128 * (1024**2)}",  # make it so that we only have 1 partition => image coverage for delta layers => enables gc of delta layers
         "image_layer_creation_check_threshold": "0",  # always check if a new image layer can be created
-        "lsn_lease_length": "0s",
     }
 
     def tenant_update_config(changes):

test_runner/regress/test_pageserver_generations.py

@@ -53,7 +53,6 @@ TENANT_CONF = {
     # create image layers eagerly, so that GC can remove some layers
     "image_creation_threshold": "1",
     "image_layer_creation_check_threshold": "0",
-    "lsn_lease_length": "0s",
 }
 
 

test_runner/regress/test_remote_storage.py

@@ -244,7 +244,6 @@ def test_remote_storage_upload_queue_retries(
             # create image layers eagerly, so that GC can remove some layers
             "image_creation_threshold": "1",
             "image_layer_creation_check_threshold": "0",
-            "lsn_lease_length": "0s",
         }
     )
 
@@ -392,7 +391,6 @@ def test_remote_timeline_client_calls_started_metric(
             # disable background compaction and GC. We invoke it manually when we want it to happen.
             "gc_period": "0s",
             "compaction_period": "0s",
-            "lsn_lease_length": "0s",
         }
     )
 

test_runner/regress/test_sharding.py

@@ -200,7 +200,6 @@ def test_sharding_split_compaction(neon_env_builder: NeonEnvBuilder, failpoint:
         # Disable automatic creation of image layers, as we will create them explicitly when we want them
         "image_creation_threshold": 9999,
         "image_layer_creation_check_threshold": 0,
-        "lsn_lease_length": "0s",
     }
 
     neon_env_builder.storage_controller_config = {

test_runner/regress/test_storage_controller.py

@@ -485,7 +485,7 @@ def test_storage_controller_compute_hook(
     httpserver.expect_request("/notify", method="PUT").respond_with_handler(handler)
 
     # Start running
-    env = neon_env_builder.init_start(initial_tenant_conf={"lsn_lease_length": "0s"})
+    env = neon_env_builder.init_start()
 
     # Initial notification from tenant creation
     assert len(notifications) == 1

test_runner/regress/test_storage_scrubber.py

@@ -204,7 +204,6 @@ def test_scrubber_physical_gc_ancestors(
             # No PITR, so that as soon as child shards generate an image layer, it covers ancestor deltas
             # and makes them GC'able
             "pitr_interval": "0s",
-            "lsn_lease_length": "0s",
         },
     )
 

test_runner/regress/test_tenant_detach.py

@@ -266,13 +266,13 @@ def test_tenant_reattach_while_busy(
 
 
 def test_tenant_detach_smoke(neon_env_builder: NeonEnvBuilder):
-    env = neon_env_builder.init_start(initial_tenant_conf={"lsn_lease_length": "0s"})
+    env = neon_env_builder.init_start()
     pageserver_http = env.pageserver.http_client()
 
     env.pageserver.allowed_errors.extend(PERMIT_PAGE_SERVICE_ERRORS)
 
     # create new nenant
-    tenant_id, timeline_id = env.initial_tenant, env.initial_timeline
+    tenant_id, timeline_id = env.neon_cli.create_tenant()
 
     # assert tenant exists on disk
     assert env.pageserver.tenant_dir(tenant_id).exists()

test_runner/regress/test_timeline_gc_blocking.py

@@ -45,10 +45,7 @@ def test_gc_blocking_by_timeline(neon_env_builder: NeonEnvBuilder, sharded: bool
     tenant_after = http.tenant_status(env.initial_tenant)
     assert tenant_before != tenant_after
     gc_blocking = tenant_after["gc_blocking"]
-    assert (
+    assert gc_blocking == "BlockingReasons { timelines: 1, reasons: EnumSet(Manual) }"
-        gc_blocking
-        == "BlockingReasons { tenant_blocked_by_lsn_lease_deadline: false, timelines: 1, reasons: EnumSet(Manual) }"
-    )
 
     wait_for_another_gc_round()
     pss.assert_log_contains(gc_skipped_line)

test_runner/regress/test_wal_acceptor.py

@@ -892,7 +892,6 @@ def test_timeline_status(neon_env_builder: NeonEnvBuilder, auth_enabled: bool):
     log.info(f"debug_dump before reboot {debug_dump_0}")
     assert debug_dump_0["timelines_count"] == 1
     assert debug_dump_0["timelines"][0]["timeline_id"] == str(timeline_id)
-    assert debug_dump_0["timelines"][0]["wal_last_modified"] != ""
 
     endpoint.safe_psql("create table t(i int)")
 

vendor/revisions.json

@@ -1,18 +1,18 @@
 {
   "v17": [
     "17rc1",
-    "dadd6fe208bb906cc0a48980f2ab4e13c47ba3ad"
+    "9156d63ce253bed9d1f76355ceec610e444eaffa"
   ],
   "v16": [
     "16.4",
-    "d009084a745cb4d5e6de222c778b2a562c8b2767"
+    "0baa7346dfd42d61912eeca554c9bb0a190f0a1e"
   ],
   "v15": [
     "15.8",
-    "1dbd6f316416c8360bbd4f3d6db956cf70937cf0"
+    "6f6d77fb5960602fcd3fd130aca9f99ecb1619c9"
   ],
   "v14": [
     "14.13",
-    "f9c51c12438b20049b6905eb4e43d321defd6ff2"
+    "a317b9b5b96978b49e78986697f3dd80d06f99a7"
   ]
 }