CI: Disable cleanup of docker/setup-buildx-action

It started to fail with the following error (for no obvious reason): ``` /nvme/actions-runner/_work/_actions/docker/setup-buildx-action/v3/webpack:/docker-setup-buildx/node_modules/@actions/cache/lib/cache.js:175 throw new Error(`Path Validation Error: Path(s) specified in the action for caching do(es) not exist, hence no cache is being saved.`); ^ Error: Path Validation Error: Path(s) specified in the action for caching do(es) not exist, hence no cache is being saved. at Object.rejected (/nvme/actions-runner/_work/_actions/docker/setup-buildx-action/v3/webpack:/docker-setup-buildx/node_modules/@actions/cache/lib/cache.js:175:1) at Generator.next (<anonymous>) at fulfilled (/nvme/actions-runner/_work/_actions/docker/setup-buildx-action/v3/webpack:/docker-setup-buildx/node_modules/@actions/cache/lib/cache.js:29:1) ```
update rustls
2026-05-23 16:10:37 +00:00 · 2024-04-21 22:04:21 +03:00 · 2024-04-20 12:37:58 +03:00
34 changed files with 213 additions and 832 deletions
--- a/.github/workflows/build_and_test.yml
+++ b/.github/workflows/build_and_test.yml
@@ -735,7 +735,12 @@ jobs:
        run: |
          mkdir -p .docker-custom
          echo DOCKER_CONFIG=$(pwd)/.docker-custom >> $GITHUB_ENV
-      - uses: docker/setup-buildx-action@v2
+      - uses: docker/setup-buildx-action@v3
+        with:
+          # Disable cleanup to work-around failure at post-run stage:
+          # "Path Validation Error: Path(s) specified in the action for caching do(es) not exist, hence no cache is being saved."
+          # See also https://github.com/neondatabase/neon/pull/7445
+          cleanup: false

      - uses: docker/login-action@v3
        with:
@@ -792,13 +797,17 @@ jobs:
        run: |
          mkdir -p .docker-custom
          echo DOCKER_CONFIG=$(pwd)/.docker-custom >> $GITHUB_ENV
-      - uses: docker/setup-buildx-action@v2
+      - uses: docker/setup-buildx-action@v3
        with:
          # Disable parallelism for docker buildkit.
          # As we already build everything with `make -j$(nproc)`, running it in additional level of parallelisam blows up the Runner.
          config-inline: |
            [worker.oci]
              max-parallelism = 1
+          # Disable cleanup to work-around failure at post-run stage:
+          # "Path Validation Error: Path(s) specified in the action for caching do(es) not exist, hence no cache is being saved."
+          # See also https://github.com/neondatabase/neon/pull/7445
+          cleanup: false

      - uses: docker/login-action@v3
        with:
@@ -865,7 +874,7 @@ jobs:
      run:
        shell: sh -eu {0}
    env:
-      VM_BUILDER_VERSION: v0.28.1
+      VM_BUILDER_VERSION: v0.23.2

    steps:
      - name: Checkout
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -5830,7 +5830,8 @@ checksum = "81cdd64d312baedb58e21336b31bc043b77e01cc99033ce76ef539f78e965ebc"
 [[package]]
 name = "svg_fmt"
 version = "0.4.2"
-source = "git+https://github.com/neondatabase/fork--nical--rust_debug?branch=neon#b9501105e746629004bc6d0473639320939dbe10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f83ba502a3265efb76efb89b0a2f7782ad6f2675015d4ce37e4b547dda42b499"

 [[package]]
 name = "syn"
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -157,8 +157,7 @@ socket2 = "0.5"
 strum = "0.24"
 strum_macros = "0.24"
 "subtle"  = "2.5.0"
-# https://github.com/nical/rust_debug/pull/4
-svg_fmt = { git = "https://github.com/neondatabase/fork--nical--rust_debug", branch = "neon" }
+svg_fmt = "0.4.1"
 sync_wrapper = "0.1.2"
 tar = "0.4"
 task-local-extensions = "0.1.4"
--- a/libs/pageserver_api/src/shard.rs
+++ b/libs/pageserver_api/src/shard.rs
@@ -5,7 +5,6 @@ use crate::{
    models::ShardParameters,
 };
 use hex::FromHex;
-use postgres_ffi::relfile_utils::INIT_FORKNUM;
 use serde::{Deserialize, Serialize};
 use utils::id::TenantId;

@@ -538,24 +537,6 @@ impl ShardIdentity {
        }
    }

-    /// Special case for issue `<https://github.com/neondatabase/neon/issues/7451>`
-    ///
-    /// When we fail to read a forknum block, this function tells us whether we may ignore the error
-    /// as a symptom of that issue.
-    pub fn is_key_buggy_forknum(&self, key: &Key) -> bool {
-        if !is_rel_block_key(key) || key.field5 != INIT_FORKNUM {
-            return false;
-        }
-
-        let mut hash = murmurhash32(key.field4);
-        hash = hash_combine(hash, murmurhash32(key.field6 / self.stripe_size.0));
-        let mapped_shard = ShardNumber((hash % self.count.0 as u32) as u8);
-
-        // The key may be affected by issue #7454: it is an initfork and it would not
-        // have mapped to shard 0 until we fixed that issue.
-        mapped_shard != ShardNumber(0)
-    }
-
    /// Return true if the key should be discarded if found in this shard's
    /// data store, e.g. during compaction after a split.
    ///
@@ -668,13 +649,7 @@ fn key_is_shard0(key: &Key) -> bool {
    // relation pages are distributed to shards other than shard zero. Everything else gets
    // stored on shard 0.  This guarantees that shard 0 can independently serve basebackup
    // requests, and any request other than those for particular blocks in relations.
-    //
-    // The only exception to this rule is "initfork" data -- this relates to postgres's UNLOGGED table
-    // type. These are special relations, usually with only 0 or 1 blocks, and we store them on shard 0
-    // because they must be included in basebackups.
-    let is_initfork = key.field5 == INIT_FORKNUM;
-
-    !is_rel_block_key(key) || is_initfork
+    !is_rel_block_key(key)
 }

 /// Provide the same result as the function in postgres `hashfn.h` with the same name
--- a/libs/postgres_ffi/src/lib.rs
+++ b/libs/postgres_ffi/src/lib.rs
@@ -118,9 +118,7 @@ pub use v14::bindings::{TimeLineID, TimestampTz, XLogRecPtr, XLogSegNo};
 // Likewise for these, although the assumption that these don't change is a little more iffy.
 pub use v14::bindings::{MultiXactOffset, MultiXactStatus};
 pub use v14::bindings::{PageHeaderData, XLogRecord};
-pub use v14::xlog_utils::{
-    XLOG_SIZE_OF_XLOG_LONG_PHD, XLOG_SIZE_OF_XLOG_RECORD, XLOG_SIZE_OF_XLOG_SHORT_PHD,
-};
+pub use v14::xlog_utils::{XLOG_SIZE_OF_XLOG_RECORD, XLOG_SIZE_OF_XLOG_SHORT_PHD};

 pub use v14::bindings::{CheckPoint, ControlFileData};

--- a/libs/postgres_ffi/wal_craft/src/lib.rs
+++ b/libs/postgres_ffi/wal_craft/src/lib.rs
@@ -4,9 +4,7 @@ use log::*;
 use postgres::types::PgLsn;
 use postgres::Client;
 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 postgres_ffi::{XLOG_SIZE_OF_XLOG_RECORD, XLOG_SIZE_OF_XLOG_SHORT_PHD};
 use std::path::{Path, PathBuf};
 use std::process::Command;
 use std::time::{Duration, Instant};
@@ -264,21 +262,11 @@ fn craft_internal<C: postgres::GenericClient>(
        intermediate_lsns.insert(0, initial_lsn);
    }

-    // Some records may be not flushed, e.g. non-transactional logical messages. Flush now.
+    // Some records may be not flushed, e.g. non-transactional logical messages.
    //
-    // If the previous WAL record ended exactly at page boundary, pg_current_wal_insert_lsn
-    // returns the position just after the page header on the next page. That's where the next
-    // record will be inserted. But the page header hasn't actually been written to the WAL
-    // yet, and if you try to flush it, you get a "request to flush past end of generated WAL"
-    // error. Because of that, if the insert location is just after a page header, back off to
-    // previous page boundary.
-    let mut lsn = u64::from(client.pg_current_wal_insert_lsn()?);
-    if lsn % WAL_SEGMENT_SIZE as u64 == XLOG_SIZE_OF_XLOG_LONG_PHD as u64 {
-        lsn -= XLOG_SIZE_OF_XLOG_LONG_PHD as u64;
-    } else if lsn % XLOG_BLCKSZ as u64 == XLOG_SIZE_OF_XLOG_SHORT_PHD as u64 {
-        lsn -= XLOG_SIZE_OF_XLOG_SHORT_PHD as u64;
-    }
-    client.execute("select neon_xlogflush($1)", &[&PgLsn::from(lsn)])?;
+    // Note: this is broken if pg_current_wal_insert_lsn is at page boundary
+    // because pg_current_wal_insert_lsn skips page headers.
+    client.execute("select neon_xlogflush(pg_current_wal_insert_lsn())", &[])?;
    Ok(intermediate_lsns)
 }

@@ -332,49 +320,38 @@ impl Crafter for LastWalRecordXlogSwitchEndsOnPageBoundary {

        client.execute("CREATE table t(x int)", &[])?;

-        // Add padding so the XLOG_SWITCH record ends exactly on XLOG_BLCKSZ boundary.  We
-        // will use carefully-sized logical messages to advance WAL insert location such
-        // that there is just enough space on the page for the XLOG_SWITCH record.
-        loop {
-            // We start with measuring how much WAL it takes for one logical message,
-            // considering all alignments and headers.
+        // Add padding so the XLOG_SWITCH record ends exactly on XLOG_BLCKSZ boundary.
+        // We will use logical message as the padding. We start with detecting how much WAL
+        // it takes for one logical message, considering all alignments and headers.
+        let base_wal_advance = {
            let before_lsn = client.pg_current_wal_insert_lsn()?;
+            // Small non-empty message bigger than few bytes is more likely than an empty
+            // message to have the same format as the big padding message.
            client.execute(
                "SELECT pg_logical_emit_message(false, 'swch', REPEAT('a', 10))",
                &[],
            )?;
-            let after_lsn = client.pg_current_wal_insert_lsn()?;
-
-            // Did the record cross a page boundary? If it did, start over. Crossing a
-            // page boundary adds to the apparent size of the record because of the page
-            // header, which throws off the calculation.
-            if u64::from(before_lsn) / XLOG_BLCKSZ as u64
-                != u64::from(after_lsn) / XLOG_BLCKSZ as u64
-            {
-                continue;
-            }
-            // base_size is the size of a logical message without the payload
-            let base_size = u64::from(after_lsn) - u64::from(before_lsn) - 10;
-
-            // Is there enough space on the page for another logical message and an
-            // XLOG_SWITCH? If not, start over.
-            let page_remain = XLOG_BLCKSZ as u64 - u64::from(after_lsn) % XLOG_BLCKSZ as u64;
-            if page_remain < base_size - XLOG_SIZE_OF_XLOG_RECORD as u64 {
-                continue;
-            }
-
-            // We will write another logical message, such that after the logical message
-            // record, there will be space for exactly one XLOG_SWITCH. How large should
-            // the logical message's payload be? An XLOG_SWITCH record has no data => its
-            // size is exactly XLOG_SIZE_OF_XLOG_RECORD.
-            let repeats = page_remain - base_size - XLOG_SIZE_OF_XLOG_RECORD as u64;
-
-            client.execute(
-                "SELECT pg_logical_emit_message(false, 'swch', REPEAT('a', $1))",
-                &[&(repeats as i32)],
-            )?;
-            break;
+            // The XLOG_SWITCH record has no data => its size is exactly XLOG_SIZE_OF_XLOG_RECORD.
+            (u64::from(client.pg_current_wal_insert_lsn()?) - u64::from(before_lsn)) as usize
+                + XLOG_SIZE_OF_XLOG_RECORD
+        };
+        let mut remaining_lsn =
+            XLOG_BLCKSZ - u64::from(client.pg_current_wal_insert_lsn()?) as usize % XLOG_BLCKSZ;
+        if remaining_lsn < base_wal_advance {
+            remaining_lsn += XLOG_BLCKSZ;
        }
+        let repeats = 10 + remaining_lsn - base_wal_advance;
+        info!(
+            "current_wal_insert_lsn={}, remaining_lsn={}, base_wal_advance={}, repeats={}",
+            client.pg_current_wal_insert_lsn()?,
+            remaining_lsn,
+            base_wal_advance,
+            repeats
+        );
+        client.execute(
+            "SELECT pg_logical_emit_message(false, 'swch', REPEAT('a', $1))",
+            &[&(repeats as i32)],
+        )?;
        info!(
            "current_wal_insert_lsn={}, XLOG_SIZE_OF_XLOG_RECORD={}",
            client.pg_current_wal_insert_lsn()?,
--- a/libs/utils/src/rate_limit.rs
+++ b/libs/utils/src/rate_limit.rs
@@ -5,7 +5,6 @@ use std::time::{Duration, Instant};
 pub struct RateLimit {
    last: Option<Instant>,
    interval: Duration,
-    rate_limited: u32,
 }

 impl RateLimit {
@@ -13,16 +12,9 @@ impl RateLimit {
        Self {
            last: None,
            interval,
-            rate_limited: 0,
        }
    }

-    /// Returns the number of calls that were rate limited
-    /// since the last one was allowed.
-    pub fn rate_limited(&self) -> u32 {
-        self.rate_limited
-    }
-
    /// Call `f` if the rate limit allows.
    /// Don't call it otherwise.
    pub fn call<F: FnOnce()>(&mut self, f: F) {
@@ -30,13 +22,9 @@ impl RateLimit {
        match self.last {
            Some(last) if now - last <= self.interval => {
                // ratelimit
-                if let Some(updated) = self.rate_limited.checked_add(1) {
-                    self.rate_limited = updated;
-                }
            }
            _ => {
                self.last = Some(now);
-                self.rate_limited = 0;
                f();
            }
        }
@@ -62,24 +50,17 @@ mod tests {

        f.call(cl);
        assert_eq!(called.load(Relaxed), 1);
-        assert_eq!(f.rate_limited(), 0);
        f.call(cl);
        assert_eq!(called.load(Relaxed), 1);
-        assert_eq!(f.rate_limited(), 1);
        f.call(cl);
        assert_eq!(called.load(Relaxed), 1);
-        assert_eq!(f.rate_limited(), 2);
        std::thread::sleep(Duration::from_millis(100));
        f.call(cl);
        assert_eq!(called.load(Relaxed), 2);
-        assert_eq!(f.rate_limited(), 0);
        f.call(cl);
        assert_eq!(called.load(Relaxed), 2);
-        assert_eq!(f.rate_limited(), 1);
-        f.call(cl);
        std::thread::sleep(Duration::from_millis(100));
        f.call(cl);
        assert_eq!(called.load(Relaxed), 3);
-        assert_eq!(f.rate_limited(), 0);
    }
 }
--- a/pageserver/ctl/src/draw_timeline_dir.rs
+++ b/pageserver/ctl/src/draw_timeline_dir.rs
@@ -9,45 +9,18 @@
 //! Coordinates in both axis are compressed for better readability.
 //! (see <https://medium.com/algorithms-digest/coordinate-compression-2fff95326fb>)
 //!
-//! The plain text API was chosen so that we can easily work with filenames from various
-//! sources; see the Usage section below for examples.
-//!
-//! # Usage
-//!
-//! ## Producing the SVG
-//!
+//! Example use:
 //! ```bash
-//!
-//! # local timeline dir
-//! ls test_output/test_pgbench\[neon-45-684\]/repo/tenants/$TENANT/timelines/$TIMELINE | \
-//!     grep "__" | cargo run --release --bin pagectl draw-timeline-dir > out.svg
-//!
-//! # Layer map dump from `/v1/tenant/$TENANT/timeline/$TIMELINE/layer`
-//! (jq -r '.historic_layers[] | .layer_file_name' | cargo  run -p pagectl draw-timeline) < layer-map.json > out.svg
-//!
-//! # From an `index_part.json` in S3
-//! (jq -r '.layer_metadata | keys[]' | cargo  run -p pagectl draw-timeline ) < index_part.json-00000016 > out.svg
-//!
+//! $ ls test_output/test_pgbench\[neon-45-684\]/repo/tenants/$TENANT/timelines/$TIMELINE | \
+//! $   grep "__" | cargo run --release --bin pagectl draw-timeline-dir > out.svg
+//! $ firefox out.svg
 //! ```
 //!
-//! ## Viewing
+//! This API was chosen so that we can easily work with filenames extracted from ssh,
+//! or from pageserver log files.
 //!
-//! **Inkscape** is better than the built-in viewers in browsers.
-//!
-//! After selecting a layer file rectangle, use "Open XML Editor" (Ctrl|Cmd + Shift + X)
-//! to see the layer file name in the comment field.
-//!
-//! ```bash
-//!
-//! # Linux
-//! inkscape out.svg
-//!
-//! # macOS
-//! /Applications/Inkscape.app/Contents/MacOS/inkscape out.svg
-//!
-//! ```
-//!
-
+//! TODO Consider shipping this as a grafana panel plugin:
+//!      <https://grafana.com/tutorials/build-a-panel-plugin/>
 use anyhow::Result;
 use pageserver::repository::Key;
 use pageserver::METADATA_FILE_NAME;
@@ -92,12 +65,7 @@ fn parse_filename(name: &str) -> (Range<Key>, Range<Lsn>) {

 pub fn main() -> Result<()> {
    // Parse layer filenames from stdin
-    struct Layer {
-        filename: String,
-        key_range: Range<Key>,
-        lsn_range: Range<Lsn>,
-    }
-    let mut files: Vec<Layer> = vec![];
+    let mut ranges: Vec<(Range<Key>, Range<Lsn>)> = vec![];
    let stdin = io::stdin();
    for line in stdin.lock().lines() {
        let line = line.unwrap();
@@ -108,23 +76,14 @@ pub fn main() -> Result<()> {
            // Don't try and parse "metadata" like a key-lsn range
            continue;
        }
-        let (key_range, lsn_range) = parse_filename(filename);
-        files.push(Layer {
-            filename: filename.to_owned(),
-            key_range,
-            lsn_range,
-        });
+        let range = parse_filename(filename);
+        ranges.push(range);
    }

    // Collect all coordinates
    let mut keys: Vec<Key> = vec![];
    let mut lsns: Vec<Lsn> = vec![];
-    for Layer {
-        key_range: keyr,
-        lsn_range: lsnr,
-        ..
-    } in &files
-    {
+    for (keyr, lsnr) in &ranges {
        keys.push(keyr.start);
        keys.push(keyr.end);
        lsns.push(lsnr.start);
@@ -148,12 +107,7 @@ pub fn main() -> Result<()> {
            h: stretch * lsn_map.len() as f32
        }
    );
-    for Layer {
-        filename,
-        key_range: keyr,
-        lsn_range: lsnr,
-    } in &files
-    {
+    for (keyr, lsnr) in &ranges {
        let key_start = *key_map.get(&keyr.start).unwrap();
        let key_end = *key_map.get(&keyr.end).unwrap();
        let key_diff = key_end - key_start;
@@ -197,7 +151,6 @@ pub fn main() -> Result<()> {
            .fill(fill)
            .stroke(Stroke::Color(rgb(0, 0, 0), 0.1))
            .border_radius(0.4)
-            .comment(filename)
        );
    }
    println!("{}", EndSvg);
--- a/pageserver/src/basebackup.rs
+++ b/pageserver/src/basebackup.rs
@@ -13,7 +13,7 @@
 use anyhow::{anyhow, bail, ensure, Context};
 use bytes::{BufMut, Bytes, BytesMut};
 use fail::fail_point;
-use pageserver_api::key::{key_to_slru_block, rel_block_to_key, Key};
+use pageserver_api::key::{key_to_slru_block, Key};
 use postgres_ffi::pg_constants;
 use std::fmt::Write as FmtWrite;
 use std::time::SystemTime;
@@ -297,20 +297,7 @@ where
                if rel.forknum == INIT_FORKNUM {
                    // I doubt we need _init fork itself, but having it at least
                    // serves as a marker relation is unlogged.
-                    if let Err(_e) = self.add_rel(rel, rel).await {
-                        if self
-                            .timeline
-                            .get_shard_identity()
-                            .is_key_buggy_forknum(&rel_block_to_key(rel, 0x0))
-                        {
-                            // Workaround https://github.com/neondatabase/neon/issues/7451 -- if we have an unlogged relation
-                            // whose INIT_FORKNUM is not correctly on shard zero, then omit it in the basebackup.  This allows
-                            // postgres to start up.  The relation won't work, but it will be possible to DROP TABLE on it and
-                            // recreate.
-                            tracing::warn!("Omitting relation {rel} for issue #7451: drop and recreate this unlogged relation");
-                            continue;
-                        }
-                    };
+                    self.add_rel(rel, rel).await?;
                    self.add_rel(rel, rel.with_forknum(MAIN_FORKNUM)).await?;
                    continue;
                }
--- a/pageserver/src/http/routes.rs
+++ b/pageserver/src/http/routes.rs
@@ -160,9 +160,6 @@ impl From<PageReconstructError> for ApiError {
    fn from(pre: PageReconstructError) -> ApiError {
        match pre {
            PageReconstructError::Other(pre) => ApiError::InternalServerError(pre),
-            PageReconstructError::MissingKey(e) => {
-                ApiError::InternalServerError(anyhow::anyhow!("{e}"))
-            }
            PageReconstructError::Cancelled => {
                ApiError::InternalServerError(anyhow::anyhow!("request was cancelled"))
            }
--- a/pageserver/src/metrics.rs
+++ b/pageserver/src/metrics.rs
@@ -86,20 +86,11 @@ pub(crate) static STORAGE_TIME_GLOBAL: Lazy<HistogramVec> = Lazy::new(|| {
    .expect("failed to define a metric")
 });

-pub(crate) static READ_NUM_LAYERS_VISITED: Lazy<Histogram> = Lazy::new(|| {
+pub(crate) static READ_NUM_FS_LAYERS: Lazy<Histogram> = Lazy::new(|| {
    register_histogram!(
-        "pageserver_layers_visited_per_read_global",
-        "Number of layers visited to reconstruct one key",
-        vec![1.0, 4.0, 8.0, 16.0, 32.0, 64.0, 128.0, 256.0, 512.0, 1024.0],
-    )
-    .expect("failed to define a metric")
-});
-
-pub(crate) static VEC_READ_NUM_LAYERS_VISITED: Lazy<Histogram> = Lazy::new(|| {
-    register_histogram!(
-        "pageserver_layers_visited_per_vectored_read_global",
-        "Average number of layers visited to reconstruct one key",
-        vec![1.0, 4.0, 8.0, 16.0, 32.0, 64.0, 128.0, 256.0, 512.0, 1024.0],
+        "pageserver_read_num_fs_layers",
+        "Number of persistent layers accessed for processing a read request, including those in the cache",
+        vec![1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 10.0, 20.0, 50.0, 100.0],
    )
    .expect("failed to define a metric")
 });
@@ -2780,8 +2771,7 @@ pub fn preinitialize_metrics() {

    // histograms
    [
-        &READ_NUM_LAYERS_VISITED,
-        &VEC_READ_NUM_LAYERS_VISITED,
+        &READ_NUM_FS_LAYERS,
        &WAIT_LSN_TIME,
        &WAL_REDO_TIME,
        &WAL_REDO_RECORDS_HISTOGRAM,
--- a/pageserver/src/pgdatadir_mapping.rs
+++ b/pageserver/src/pgdatadir_mapping.rs
@@ -1446,14 +1446,10 @@ impl<'a> DatadirModification<'a> {
                    // reset the map.
                    return Err(e.into());
                }
-                // Note: we added missing key error variant in https://github.com/neondatabase/neon/pull/7393 but
-                // the original code assumes all other errors are missing keys. Therefore, we keep the code path
-                // the same for now, though in theory, we should only match the `MissingKey` variant.
-                Err(
-                    PageReconstructError::Other(_)
-                    | PageReconstructError::WalRedo(_)
-                    | PageReconstructError::MissingKey { .. },
-                ) => {
+                // FIXME: PageReconstructError doesn't have an explicit variant for key-not-found, so
+                // we are assuming that all _other_ possible errors represents a missing key.  If some
+                // other error occurs, we may incorrectly reset the map of aux files.
+                Err(PageReconstructError::Other(_) | PageReconstructError::WalRedo(_)) => {
                    // Key is missing, we must insert an image as the basis for subsequent deltas.

                    let mut dir = AuxFilesDirectory {
--- a/pageserver/src/tenant.rs
+++ b/pageserver/src/tenant.rs
@@ -559,10 +559,9 @@ impl Tenant {
            // By doing what we do here, the index part upload is retried.
            // If control plane retries timeline creation in the meantime, the mgmt API handler
            // for timeline creation will coalesce on the upload we queue here.
-            // FIXME: this branch should be dead code as we no longer write local metadata.
            let rtc = timeline.remote_client.as_ref().unwrap();
            rtc.init_upload_queue_for_empty_remote(&metadata)?;
-            rtc.schedule_index_upload_for_full_metadata_update(&metadata)?;
+            rtc.schedule_index_upload_for_metadata_update(&metadata)?;
        }

        timeline
@@ -3028,7 +3027,7 @@ impl Tenant {
        // See also https://github.com/neondatabase/neon/issues/3865
        if let Some(remote_client) = new_timeline.remote_client.as_ref() {
            remote_client
-                .schedule_index_upload_for_full_metadata_update(&metadata)
+                .schedule_index_upload_for_metadata_update(&metadata)
                .context("branch initial metadata upload")?;
        }

--- a/pageserver/src/tenant/metadata.rs
+++ b/pageserver/src/tenant/metadata.rs
@@ -235,12 +235,6 @@ impl TimelineMetadata {
        let bytes = instance.to_bytes().unwrap();
        Self::from_bytes(&bytes).unwrap()
    }
-
-    pub(crate) fn apply(&mut self, update: &MetadataUpdate) {
-        self.body.disk_consistent_lsn = update.disk_consistent_lsn;
-        self.body.prev_record_lsn = update.prev_record_lsn;
-        self.body.latest_gc_cutoff_lsn = update.latest_gc_cutoff_lsn;
-    }
 }

 impl<'de> Deserialize<'de> for TimelineMetadata {
@@ -265,27 +259,6 @@ impl Serialize for TimelineMetadata {
    }
 }

-/// Parts of the metadata which are regularly modified.
-pub(crate) struct MetadataUpdate {
-    disk_consistent_lsn: Lsn,
-    prev_record_lsn: Option<Lsn>,
-    latest_gc_cutoff_lsn: Lsn,
-}
-
-impl MetadataUpdate {
-    pub(crate) fn new(
-        disk_consistent_lsn: Lsn,
-        prev_record_lsn: Option<Lsn>,
-        latest_gc_cutoff_lsn: Lsn,
-    ) -> Self {
-        Self {
-            disk_consistent_lsn,
-            prev_record_lsn,
-            latest_gc_cutoff_lsn,
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
    use super::*;
--- a/pageserver/src/tenant/remote_timeline_client.rs
+++ b/pageserver/src/tenant/remote_timeline_client.rs
@@ -236,7 +236,6 @@ use utils::id::{TenantId, TimelineId};

 use self::index::IndexPart;

-use super::metadata::MetadataUpdate;
 use super::storage_layer::{Layer, LayerFileName, ResidentLayer};
 use super::upload_queue::SetDeletedFlagProgress;
 use super::Generation;
@@ -537,10 +536,9 @@ impl RemoteTimelineClient {
    // Upload operations.
    //

-    /// Launch an index-file upload operation in the background, with
-    /// fully updated metadata.
    ///
-    /// This should only be used to upload initial metadata to remote storage.
+    /// Launch an index-file upload operation in the background, with
+    /// updated metadata.
    ///
    /// The upload will be added to the queue immediately, but it
    /// won't be performed until all previously scheduled layer file
@@ -552,7 +550,7 @@ impl RemoteTimelineClient {
    /// If there were any changes to the list of files, i.e. if any
    /// layer file uploads were scheduled, since the last index file
    /// upload, those will be included too.
-    pub fn schedule_index_upload_for_full_metadata_update(
+    pub fn schedule_index_upload_for_metadata_update(
        self: &Arc<Self>,
        metadata: &TimelineMetadata,
    ) -> anyhow::Result<()> {
@@ -568,27 +566,6 @@ impl RemoteTimelineClient {
        Ok(())
    }

-    /// Launch an index-file upload operation in the background, with only parts of the metadata
-    /// updated.
-    ///
-    /// This is the regular way of updating metadata on layer flushes or Gc.
-    ///
-    /// Using this lighter update mechanism allows for reparenting and detaching without changes to
-    /// `index_part.json`, while being more clear on what values update regularly.
-    pub(crate) fn schedule_index_upload_for_metadata_update(
-        self: &Arc<Self>,
-        update: &MetadataUpdate,
-    ) -> anyhow::Result<()> {
-        let mut guard = self.upload_queue.lock().unwrap();
-        let upload_queue = guard.initialized_mut()?;
-
-        upload_queue.latest_metadata.apply(update);
-
-        self.schedule_index_upload(upload_queue, upload_queue.latest_metadata.clone());
-
-        Ok(())
-    }
-
    ///
    /// Launch an index-file upload operation in the background, if necessary.
    ///
@@ -2047,7 +2024,7 @@ mod tests {
        // Schedule upload of index. Check that it is queued
        let metadata = dummy_metadata(Lsn(0x20));
        client
-            .schedule_index_upload_for_full_metadata_update(&metadata)
+            .schedule_index_upload_for_metadata_update(&metadata)
            .unwrap();
        {
            let mut guard = client.upload_queue.lock().unwrap();
--- a/pageserver/src/tenant/storage_layer.rs
+++ b/pageserver/src/tenant/storage_layer.rs
@@ -118,7 +118,6 @@ pub(crate) struct ValuesReconstructState {
    pub(crate) keys: HashMap<Key, Result<VectoredValueReconstructState, PageReconstructError>>,

    keys_done: KeySpaceRandomAccum,
-    layers_visited: u32,
 }

 impl ValuesReconstructState {
@@ -126,7 +125,6 @@ impl ValuesReconstructState {
        Self {
            keys: HashMap::new(),
            keys_done: KeySpaceRandomAccum::new(),
-            layers_visited: 0,
        }
    }

@@ -140,14 +138,6 @@ impl ValuesReconstructState {
        }
    }

-    pub(crate) fn on_layer_visited(&mut self) {
-        self.layers_visited += 1;
-    }
-
-    pub(crate) fn get_layers_visited(&self) -> u32 {
-        self.layers_visited
-    }
-
    /// Update the state collected for a given key.
    /// Returns true if this was the last value needed for the key and false otherwise.
    ///
--- a/pageserver/src/tenant/storage_layer/delta_layer.rs
+++ b/pageserver/src/tenant/storage_layer/delta_layer.rs
@@ -728,9 +728,6 @@ impl DeltaLayerInner {
            // production code path
            expected_summary.index_start_blk = actual_summary.index_start_blk;
            expected_summary.index_root_blk = actual_summary.index_root_blk;
-            // mask out the timeline_id, but still require the layers to be from the same tenant
-            expected_summary.timeline_id = actual_summary.timeline_id;
-
            if actual_summary != expected_summary {
                bail!(
                    "in-file summary does not match expected summary. actual = {:?} expected = {:?}",
--- a/pageserver/src/tenant/storage_layer/image_layer.rs
+++ b/pageserver/src/tenant/storage_layer/image_layer.rs
@@ -396,8 +396,6 @@ impl ImageLayerInner {
            // production code path
            expected_summary.index_start_blk = actual_summary.index_start_blk;
            expected_summary.index_root_blk = actual_summary.index_root_blk;
-            // mask out the timeline_id, but still require the layers to be from the same tenant
-            expected_summary.timeline_id = actual_summary.timeline_id;

            if actual_summary != expected_summary {
                bail!(
--- a/pageserver/src/tenant/storage_layer/inmemory_layer.rs
+++ b/pageserver/src/tenant/storage_layer/inmemory_layer.rs
@@ -26,7 +26,7 @@ use utils::{bin_ser::BeSer, id::TimelineId, lsn::Lsn, vec_map::VecMap};
 // while being able to use std::fmt::Write's methods
 use crate::metrics::TIMELINE_EPHEMERAL_BYTES;
 use std::cmp::Ordering;
-use std::fmt::Write;
+use std::fmt::Write as _;
 use std::ops::Range;
 use std::sync::atomic::Ordering as AtomicOrdering;
 use std::sync::atomic::{AtomicU64, AtomicUsize};
@@ -54,12 +54,6 @@ pub struct InMemoryLayer {
    /// Writes are only allowed when this is `None`.
    end_lsn: OnceLock<Lsn>,

-    /// Used for traversal path. Cached representation of the in-memory layer before frozen.
-    local_path_str: Arc<str>,
-
-    /// Used for traversal path. Cached representation of the in-memory layer after frozen.
-    frozen_local_path_str: OnceLock<Arc<str>>,
-
    opened_at: Instant,

    /// The above fields never change, except for `end_lsn`, which is only set once.
@@ -247,12 +241,6 @@ impl InMemoryLayer {
        self.start_lsn..self.end_lsn_or_max()
    }

-    pub(crate) fn local_path_str(&self) -> &Arc<str> {
-        self.frozen_local_path_str
-            .get()
-            .unwrap_or(&self.local_path_str)
-    }
-
    /// debugging function to print out the contents of the layer
    ///
    /// this is likely completly unused
@@ -442,24 +430,10 @@ impl InMemoryLayer {
    }
 }

-fn inmem_layer_display(mut f: impl Write, start_lsn: Lsn, end_lsn: Lsn) -> std::fmt::Result {
-    write!(f, "inmem-{:016X}-{:016X}", start_lsn.0, end_lsn.0)
-}
-
-fn inmem_layer_log_display(
-    mut f: impl Write,
-    timeline: TimelineId,
-    start_lsn: Lsn,
-    end_lsn: Lsn,
-) -> std::fmt::Result {
-    write!(f, "timeline {} in-memory ", timeline)?;
-    inmem_layer_display(f, start_lsn, end_lsn)
-}
-
 impl std::fmt::Display for InMemoryLayer {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let end_lsn = self.end_lsn_or_max();
-        inmem_layer_display(f, self.start_lsn, end_lsn)
+        write!(f, "inmem-{:016X}-{:016X}", self.start_lsn.0, end_lsn.0)
    }
 }

@@ -484,12 +458,6 @@ impl InMemoryLayer {

        Ok(InMemoryLayer {
            file_id: key,
-            local_path_str: {
-                let mut buf = String::new();
-                inmem_layer_log_display(&mut buf, timeline_id, start_lsn, Lsn::MAX).unwrap();
-                buf.into()
-            },
-            frozen_local_path_str: OnceLock::new(),
            conf,
            timeline_id,
            tenant_shard_id,
@@ -584,15 +552,6 @@ impl InMemoryLayer {
        );
        self.end_lsn.set(end_lsn).expect("end_lsn set only once");

-        self.frozen_local_path_str
-            .set({
-                let mut buf = String::new();
-                inmem_layer_log_display(&mut buf, self.get_timeline_id(), self.start_lsn, end_lsn)
-                    .unwrap();
-                buf.into()
-            })
-            .expect("frozen_local_path_str set only once");
-
        for vec_map in inner.index.values() {
            for (lsn, _pos) in vec_map.as_slice() {
                assert!(*lsn < end_lsn);
--- a/pageserver/src/tenant/storage_layer/layer.rs
+++ b/pageserver/src/tenant/storage_layer/layer.rs
@@ -395,10 +395,6 @@ impl Layer {
        &self.0.path
    }

-    pub(crate) fn local_path_str(&self) -> &Arc<str> {
-        &self.0.path_str
-    }
-
    pub(crate) fn metadata(&self) -> LayerFileMetadata {
        self.0.metadata()
    }
@@ -521,9 +517,6 @@ struct LayerInner {
    /// Full path to the file; unclear if this should exist anymore.
    path: Utf8PathBuf,

-    /// String representation of the full path, used for traversal id.
-    path_str: Arc<str>,
-
    desc: PersistentLayerDesc,

    /// Timeline access is needed for remote timeline client and metrics.
@@ -729,7 +722,6 @@ impl LayerInner {

        LayerInner {
            conf,
-            path_str: path.to_string().into(),
            path,
            desc,
            timeline: Arc::downgrade(timeline),
--- a/pageserver/src/tenant/storage_layer/layer/tests.rs
+++ b/pageserver/src/tenant/storage_layer/layer/tests.rs
@@ -818,13 +818,11 @@ async fn eviction_cancellation_on_drop() {
    }
 }

-/// A test case to remind you the cost of these structures. You can bump the size limit
-/// below if it is really necessary to add more fields to the structures.
 #[test]
 fn layer_size() {
    assert_eq!(std::mem::size_of::<LayerAccessStats>(), 2040);
    assert_eq!(std::mem::size_of::<PersistentLayerDesc>(), 104);
-    assert_eq!(std::mem::size_of::<LayerInner>(), 2344);
+    assert_eq!(std::mem::size_of::<LayerInner>(), 2328);
    // it also has the utf8 path
 }

--- a/pageserver/src/tenant/timeline.rs
+++ b/pageserver/src/tenant/timeline.rs
@@ -23,7 +23,7 @@ use pageserver_api::{
        EvictionPolicy, InMemoryLayerInfo, LayerMapInfo, TimelineState,
    },
    reltag::BlockNumber,
-    shard::{ShardIdentity, ShardNumber, TenantShardId},
+    shard::{ShardIdentity, TenantShardId},
 };
 use rand::Rng;
 use serde_with::serde_as;
@@ -428,62 +428,6 @@ pub(crate) enum PageReconstructError {
    /// An error happened replaying WAL records
    #[error(transparent)]
    WalRedo(anyhow::Error),
-
-    #[error("{0}")]
-    MissingKey(MissingKeyError),
-}
-
-#[derive(Debug)]
-pub struct MissingKeyError {
-    stuck_at_lsn: bool,
-    key: Key,
-    shard: ShardNumber,
-    cont_lsn: Lsn,
-    request_lsn: Lsn,
-    ancestor_lsn: Option<Lsn>,
-    traversal_path: Vec<TraversalPathItem>,
-    backtrace: Option<std::backtrace::Backtrace>,
-}
-
-impl std::fmt::Display for MissingKeyError {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        if self.stuck_at_lsn {
-            // Records are found in this timeline but no image layer or initial delta record was found.
-            write!(
-                f,
-                "could not find layer with more data for key {} (shard {:?}) at LSN {}, request LSN {}",
-                self.key, self.shard, self.cont_lsn, self.request_lsn
-            )?;
-            if let Some(ref ancestor_lsn) = self.ancestor_lsn {
-                write!(f, ", ancestor {}", ancestor_lsn)?;
-            }
-        } else {
-            // No records in this timeline.
-            write!(
-                f,
-                "could not find data for key {} (shard {:?}) at LSN {}, for request at LSN {}",
-                self.key, self.shard, self.cont_lsn, self.request_lsn
-            )?;
-        }
-
-        if !self.traversal_path.is_empty() {
-            writeln!(f)?;
-        }
-
-        for (r, c, l) in &self.traversal_path {
-            writeln!(
-                f,
-                "layer traversal: result {:?}, cont_lsn {}, layer: {}",
-                r, c, l,
-            )?;
-        }
-
-        if let Some(ref backtrace) = self.backtrace {
-            write!(f, "\n{}", backtrace)?;
-        }
-
-        Ok(())
-    }
 }

 impl PageReconstructError {
@@ -495,7 +439,6 @@ impl PageReconstructError {
            AncestorLsnTimeout(_) => false,
            Cancelled | AncestorStopping(_) => true,
            WalRedo(_) => false,
-            MissingKey { .. } => false,
        }
    }
 }
@@ -810,7 +753,7 @@ impl Timeline {
                writeln!(
                    msg,
                    "- layer traversal: result {res:?}, cont_lsn {cont_lsn}, layer: {}",
-                    layer,
+                    layer(),
                )
                .expect("string grows")
            });
@@ -824,10 +767,6 @@ impl Timeline {
    }

    pub(crate) const MAX_GET_VECTORED_KEYS: u64 = 32;
-    // If we are visiting more than 50 layers to reconstruct one value,
-    // it's likely that something is wrong with compaction. We print a
-    // rate limited warning in that case.
-    pub(crate) const LAYERS_VISITED_WARN_THRESHOLD: u64 = 50;

    /// Look up multiple page versions at a given LSN
    ///
@@ -933,11 +872,9 @@ impl Timeline {
                    Err(Cancelled | AncestorStopping(_)) => {
                        return Err(GetVectoredError::Cancelled)
                    }
-                    // we only capture stuck_at_lsn=false now until we figure out https://github.com/neondatabase/neon/issues/7380
-                    Err(MissingKey(MissingKeyError {
-                        stuck_at_lsn: false,
-                        ..
-                    })) => return Err(GetVectoredError::MissingKey(key)),
+                    Err(Other(err)) if err.to_string().contains("could not find data for key") => {
+                        return Err(GetVectoredError::MissingKey(key))
+                    }
                    _ => {
                        values.insert(key, block);
                        key = key.next();
@@ -961,7 +898,6 @@ impl Timeline {
            .await?;

        let mut results: BTreeMap<Key, Result<Bytes, PageReconstructError>> = BTreeMap::new();
-        let layers_visited = reconstruct_state.get_layers_visited();
        for (key, res) in reconstruct_state.keys {
            match res {
                Err(err) => {
@@ -976,25 +912,6 @@ impl Timeline {
            }
        }

-        // Note that this is an approximation. Tracking the exact number of layers visited
-        // per key requires virtually unbounded memory usage and is inefficient
-        // (i.e. segment tree tracking each range queried from a layer)
-        let average_layers_visited = layers_visited as f64 / results.len() as f64;
-
-        crate::metrics::VEC_READ_NUM_LAYERS_VISITED.observe(average_layers_visited);
-
-        if average_layers_visited >= Self::LAYERS_VISITED_WARN_THRESHOLD as f64 {
-            use utils::rate_limit::RateLimit;
-            static LOGGED: Lazy<Mutex<RateLimit>> =
-                Lazy::new(|| Mutex::new(RateLimit::new(Duration::from_secs(10))));
-            let mut rate_limit = LOGGED.lock().unwrap();
-            let rate_limited = rate_limit.rate_limited();
-            rate_limit.call(|| {
-                warn!("Too many layers visited by vectored read: {} >= {}; rate limited {} occurences",
-                      average_layers_visited, Self::LAYERS_VISITED_WARN_THRESHOLD, rate_limited);
-            });
-        }
-
        Ok(results)
    }

@@ -2775,7 +2692,7 @@ impl Timeline {
    }
 }

-type TraversalId = Arc<str>;
+type TraversalId = String;

 trait TraversalLayerExt {
    fn traversal_id(&self) -> TraversalId;
@@ -2783,13 +2700,13 @@ trait TraversalLayerExt {

 impl TraversalLayerExt for Layer {
    fn traversal_id(&self) -> TraversalId {
-        Arc::clone(self.local_path_str())
+        self.local_path().to_string()
    }
 }

 impl TraversalLayerExt for Arc<InMemoryLayer> {
    fn traversal_id(&self) -> TraversalId {
-        Arc::clone(self.local_path_str())
+        format!("timeline {} in-memory {self}", self.get_timeline_id())
    }
 }

@@ -2818,18 +2735,7 @@ impl Timeline {
        let mut timeline = self;

        let mut read_count = scopeguard::guard(0, |cnt| {
-            crate::metrics::READ_NUM_LAYERS_VISITED.observe(cnt as f64);
-            if cnt >= Self::LAYERS_VISITED_WARN_THRESHOLD {
-                use utils::rate_limit::RateLimit;
-                static LOGGED: Lazy<Mutex<RateLimit>> =
-                    Lazy::new(|| Mutex::new(RateLimit::new(Duration::from_secs(10))));
-                let mut rate_limit = LOGGED.lock().unwrap();
-                let rate_limited = rate_limit.rate_limited();
-                rate_limit.call(|| {
-                    warn!("Too many layers visited by non-vectored read: {} >= {}; rate limited {} occurences",
-                          cnt, Self::LAYERS_VISITED_WARN_THRESHOLD, rate_limited);
-                });
-            }
+            crate::metrics::READ_NUM_FS_LAYERS.observe(cnt as f64)
        });

        // For debugging purposes, collect the path of layers that we traversed
@@ -2869,35 +2775,32 @@ impl Timeline {
                        if prev <= cont_lsn {
                            // Didn't make any progress in last iteration. Error out to avoid
                            // getting stuck in the loop.
-                            return Err(PageReconstructError::MissingKey(MissingKeyError {
-                                stuck_at_lsn: true,
+                            return Err(layer_traversal_error(format!(
+                                "could not find layer with more data for key {} at LSN {}, request LSN {}, ancestor {}",
                                key,
-                                shard: self.shard_identity.get_shard_number(&key),
-                                cont_lsn: Lsn(cont_lsn.0 - 1),
+                                Lsn(cont_lsn.0 - 1),
                                request_lsn,
-                                ancestor_lsn: Some(timeline.ancestor_lsn),
-                                traversal_path,
-                                backtrace: None,
-                            }));
+                                timeline.ancestor_lsn
+                            ), traversal_path));
                        }
                    }
                    prev_lsn = Some(cont_lsn);
                }
                ValueReconstructResult::Missing => {
-                    return Err(PageReconstructError::MissingKey(MissingKeyError {
-                        stuck_at_lsn: false,
-                        key,
-                        shard: self.shard_identity.get_shard_number(&key),
-                        cont_lsn,
-                        request_lsn,
-                        ancestor_lsn: None,
-                        traversal_path,
-                        backtrace: if cfg!(test) {
-                            Some(std::backtrace::Backtrace::force_capture())
+                    return Err(layer_traversal_error(
+                        if cfg!(test) {
+                            format!(
+                                "could not find data for key {} (shard {:?}) at LSN {}, for request at LSN {}\n{}",
+                                key, self.shard_identity.get_shard_number(&key), cont_lsn, request_lsn, std::backtrace::Backtrace::force_capture(),
+                            )
                        } else {
-                            None
+                            format!(
+                                "could not find data for key {} (shard {:?}) at LSN {}, for request at LSN {}",
+                                key, self.shard_identity.get_shard_number(&key), cont_lsn, request_lsn
+                            )
                        },
-                    }));
+                        traversal_path,
+                    ));
                }
            }

@@ -2944,8 +2847,12 @@ impl Timeline {
                        Err(e) => return Err(PageReconstructError::from(e)),
                    };
                    cont_lsn = lsn_floor;
-                    *read_count += 1;
-                    traversal_path.push((result, cont_lsn, open_layer.traversal_id()));
+                    // metrics: open_layer does not count as fs access, so we are not updating `read_count`
+                    traversal_path.push((
+                        result,
+                        cont_lsn,
+                        Box::new(move || open_layer.traversal_id()),
+                    ));
                    continue 'outer;
                }
            }
@@ -2971,8 +2878,12 @@ impl Timeline {
                        Err(e) => return Err(PageReconstructError::from(e)),
                    };
                    cont_lsn = lsn_floor;
-                    *read_count += 1;
-                    traversal_path.push((result, cont_lsn, frozen_layer.traversal_id()));
+                    // metrics: open_layer does not count as fs access, so we are not updating `read_count`
+                    traversal_path.push((
+                        result,
+                        cont_lsn,
+                        Box::new(move || frozen_layer.traversal_id()),
+                    ));
                    continue 'outer;
                }
            }
@@ -2993,7 +2904,14 @@ impl Timeline {
                };
                cont_lsn = lsn_floor;
                *read_count += 1;
-                traversal_path.push((result, cont_lsn, layer.traversal_id()));
+                traversal_path.push((
+                    result,
+                    cont_lsn,
+                    Box::new({
+                        let layer = layer.to_owned();
+                        move || layer.traversal_id()
+                    }),
+                ));
                continue 'outer;
            } else if timeline.ancestor_timeline.is_some() {
                // Nothing on this timeline. Traverse to parent
@@ -3164,8 +3082,6 @@ impl Timeline {

                unmapped_keyspace = keyspace_to_read;
                cont_lsn = next_cont_lsn;
-
-                reconstruct_state.on_layer_visited();
            } else {
                break;
            }
@@ -3609,7 +3525,7 @@ impl Timeline {
        &self,
        disk_consistent_lsn: Lsn,
        layers_to_upload: impl IntoIterator<Item = ResidentLayer>,
-    ) -> anyhow::Result<()> {
+    ) -> anyhow::Result<TimelineMetadata> {
        // We can only save a valid 'prev_record_lsn' value on disk if we
        // flushed *all* in-memory changes to disk. We only track
        // 'prev_record_lsn' in memory for the latest processed record, so we
@@ -3626,10 +3542,19 @@ impl Timeline {
            None
        };

-        let update = crate::tenant::metadata::MetadataUpdate::new(
+        let ancestor_timeline_id = self
+            .ancestor_timeline
+            .as_ref()
+            .map(|ancestor| ancestor.timeline_id);
+
+        let metadata = TimelineMetadata::new(
            disk_consistent_lsn,
            ondisk_prev_record_lsn,
+            ancestor_timeline_id,
+            self.ancestor_lsn,
            *self.latest_gc_cutoff_lsn.read(),
+            self.initdb_lsn,
+            self.pg_version,
        );

        fail_point!("checkpoint-before-saving-metadata", |x| bail!(
@@ -3641,10 +3566,10 @@ impl Timeline {
            for layer in layers_to_upload {
                remote_client.schedule_layer_file_upload(layer)?;
            }
-            remote_client.schedule_index_upload_for_metadata_update(&update)?;
+            remote_client.schedule_index_upload_for_metadata_update(&metadata)?;
        }

-        Ok(())
+        Ok(metadata)
    }

    pub(crate) async fn preserve_initdb_archive(&self) -> anyhow::Result<()> {
@@ -4740,7 +4665,35 @@ impl Timeline {
    }
 }

-type TraversalPathItem = (ValueReconstructResult, Lsn, TraversalId);
+type TraversalPathItem = (
+    ValueReconstructResult,
+    Lsn,
+    Box<dyn Send + FnOnce() -> TraversalId>,
+);
+
+/// Helper function for get_reconstruct_data() to add the path of layers traversed
+/// to an error, as anyhow context information.
+fn layer_traversal_error(msg: String, path: Vec<TraversalPathItem>) -> PageReconstructError {
+    // We want the original 'msg' to be the outermost context. The outermost context
+    // is the most high-level information, which also gets propagated to the client.
+    let mut msg_iter = path
+        .into_iter()
+        .map(|(r, c, l)| {
+            format!(
+                "layer traversal: result {:?}, cont_lsn {}, layer: {}",
+                r,
+                c,
+                l(),
+            )
+        })
+        .chain(std::iter::once(msg));
+    // Construct initial message from the first traversed layer
+    let err = anyhow!(msg_iter.next().unwrap());
+
+    // Append all subsequent traversals, and the error message 'msg', as contexts.
+    let msg = msg_iter.fold(err, |err, msg| err.context(msg));
+    PageReconstructError::from(msg)
+}

 struct TimelineWriterState {
    open_layer: Arc<InMemoryLayer>,
--- a/proxy/src/bin/proxy.rs
+++ b/proxy/src/bin/proxy.rs
@@ -200,12 +200,6 @@ struct ProxyCliArgs {
    /// Size of each event is no more than 400 bytes, so 2**22 is about 200MB before the compression.
    #[clap(long, default_value = "4194304")]
    metric_backup_collection_chunk_size: usize,
-    /// Whether to retry the connection to the compute node
-    #[clap(long, default_value = config::RetryConfig::CONNECT_TO_COMPUTE_DEFAULT_VALUES)]
-    connect_to_compute_retry: String,
-    /// Whether to retry the wake_compute request
-    #[clap(long, default_value = config::RetryConfig::WAKE_COMPUTE_DEFAULT_VALUES)]
-    wake_compute_retry: String,
 }

 #[derive(clap::Args, Clone, Copy, Debug)]
@@ -590,10 +584,6 @@ fn build_config(args: &ProxyCliArgs) -> anyhow::Result<&'static ProxyConfig> {
        handshake_timeout: args.handshake_timeout,
        region: args.region.clone(),
        aws_region: args.aws_region.clone(),
-        wake_compute_retry_config: config::RetryConfig::parse(&args.wake_compute_retry)?,
-        connect_to_compute_retry_config: config::RetryConfig::parse(
-            &args.connect_to_compute_retry,
-        )?,
    }));

    Ok(config)
--- a/proxy/src/config.rs
+++ b/proxy/src/config.rs
@@ -33,8 +33,6 @@ pub struct ProxyConfig {
    pub region: String,
    pub handshake_timeout: Duration,
    pub aws_region: String,
-    pub wake_compute_retry_config: RetryConfig,
-    pub connect_to_compute_retry_config: RetryConfig,
 }

 #[derive(Debug)]
@@ -519,59 +517,6 @@ impl FromStr for ProjectInfoCacheOptions {
    }
 }

-/// This is a config for connect to compute and wake compute.
-#[derive(Clone, Copy, Debug)]
-pub struct RetryConfig {
-    /// Number of times we should retry.
-    pub max_retries: u32,
-    /// Retry duration is base_delay * backoff_factor ^ n, where n starts at 0
-    pub base_delay: tokio::time::Duration,
-    /// Exponential base for retry wait duration
-    pub backoff_factor: f64,
-}
-
-impl RetryConfig {
-    /// Default options for RetryConfig.
-
-    /// Total delay for 4 retries with 1s base delay and 2.0 backoff factor is 7s.
-    pub const CONNECT_TO_COMPUTE_DEFAULT_VALUES: &'static str =
-        "num_retries=4,base_retry_wait_duration=1s,retry_wait_exponent_base=2.0";
-    /// Total delay for 4 retries with 1s base delay and 2.0 backoff factor is 7s.
-    /// Cplane has timeout of 60s on each request.
-    pub const WAKE_COMPUTE_DEFAULT_VALUES: &'static str =
-        "num_retries=4,base_retry_wait_duration=1s,retry_wait_exponent_base=2.0";
-
-    /// Parse retry options passed via cmdline.
-    /// Example: [`Self::CONNECT_TO_COMPUTE_DEFAULT_VALUES`].
-    pub fn parse(options: &str) -> anyhow::Result<Self> {
-        let mut num_retries = None;
-        let mut base_retry_wait_duration = None;
-        let mut retry_wait_exponent_base = None;
-
-        for option in options.split(',') {
-            let (key, value) = option
-                .split_once('=')
-                .with_context(|| format!("bad key-value pair: {option}"))?;
-
-            match key {
-                "num_retries" => num_retries = Some(value.parse()?),
-                "base_retry_wait_duration" => {
-                    base_retry_wait_duration = Some(humantime::parse_duration(value)?)
-                }
-                "retry_wait_exponent_base" => retry_wait_exponent_base = Some(value.parse()?),
-                unknown => bail!("unknown key: {unknown}"),
-            }
-        }
-
-        Ok(Self {
-            max_retries: num_retries.context("missing `num_retries`")?,
-            base_delay: base_retry_wait_duration.context("missing `base_retry_wait_duration`")?,
-            backoff_factor: retry_wait_exponent_base
-                .context("missing `retry_wait_exponent_base`")?,
-        })
-    }
-}
-
 /// Helper for cmdline cache options parsing.
 pub struct WakeComputeLockOptions {
    /// The number of shards the lock map should have
--- a/proxy/src/metrics.rs
+++ b/proxy/src/metrics.rs
@@ -119,10 +119,6 @@ pub struct ProxyMetrics {

    /// Number of invalid endpoints (per protocol, per rejected).
    pub invalid_endpoints_total: CounterVec<InvalidEndpointsSet>,
-
-    /// Number of retries (per outcome, per retry_type).
-    #[metric(metadata = Thresholds::with_buckets([0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0]))]
-    pub retries_metric: HistogramVec<RetriesMetricSet, 9>,
 }

 #[derive(MetricGroup)]
@@ -484,16 +480,3 @@ pub struct InvalidEndpointsGroup {
    pub rejected: Bool,
    pub outcome: ConnectOutcome,
 }
-
-#[derive(LabelGroup)]
-#[label(set = RetriesMetricSet)]
-pub struct RetriesMetricGroup {
-    pub outcome: ConnectOutcome,
-    pub retry_type: RetryType,
-}
-
-#[derive(FixedCardinalityLabel, Clone, Copy, Debug)]
-pub enum RetryType {
-    WakeCompute,
-    ConnectToCompute,
-}
--- a/proxy/src/proxy.rs
+++ b/proxy/src/proxy.rs
@@ -308,8 +308,6 @@ pub async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
        &TcpMechanism { params: &params },
        &user_info,
        mode.allow_self_signed_compute(config),
-        config.wake_compute_retry_config,
-        config.connect_to_compute_retry_config,
    )
    .or_else(|e| stream.throw_error(e))
    .await?;
--- a/proxy/src/proxy/connect_compute.rs
+++ b/proxy/src/proxy/connect_compute.rs
@@ -1,11 +1,10 @@
 use crate::{
    auth::backend::ComputeCredentialKeys,
    compute::{self, PostgresConnection},
-    config::RetryConfig,
    console::{self, errors::WakeComputeError, CachedNodeInfo, NodeInfo},
    context::RequestMonitoring,
    error::ReportableError,
-    metrics::{ConnectOutcome, ConnectionFailureKind, Metrics, RetriesMetricGroup, RetryType},
+    metrics::{ConnectionFailureKind, Metrics},
    proxy::{
        retry::{retry_after, ShouldRetry},
        wake_compute::wake_compute,
@@ -94,23 +93,19 @@ pub async fn connect_to_compute<M: ConnectMechanism, B: ComputeConnectBackend>(
    mechanism: &M,
    user_info: &B,
    allow_self_signed_compute: bool,
-    wake_compute_retry_config: RetryConfig,
-    connect_to_compute_retry_config: RetryConfig,
 ) -> Result<M::Connection, M::Error>
 where
    M::ConnectError: ShouldRetry + std::fmt::Debug,
    M::Error: From<WakeComputeError>,
 {
    let mut num_retries = 0;
-    let mut node_info =
-        wake_compute(&mut num_retries, ctx, user_info, wake_compute_retry_config).await?;
+    let mut node_info = wake_compute(&mut num_retries, ctx, user_info).await?;
    if let Some(keys) = user_info.get_keys() {
        node_info.set_keys(keys);
    }
    node_info.allow_self_signed_compute = allow_self_signed_compute;
    // let mut node_info = credentials.get_node_info(ctx, user_info).await?;
    mechanism.update_connect_config(&mut node_info.config);
-    let retry_type = RetryType::ConnectToCompute;

    // try once
    let err = match mechanism
@@ -119,13 +114,6 @@ where
    {
        Ok(res) => {
            ctx.latency_timer.success();
-            Metrics::get().proxy.retries_metric.observe(
-                RetriesMetricGroup {
-                    outcome: ConnectOutcome::Success,
-                    retry_type,
-                },
-                num_retries.into(),
-            );
            return Ok(res);
        }
        Err(e) => e,
@@ -136,7 +124,7 @@ where
    let node_info = if !node_info.cached() {
        // If we just recieved this from cplane and dodn't get it from cache, we shouldn't retry.
        // Do not need to retrieve a new node_info, just return the old one.
-        if !err.should_retry(num_retries, connect_to_compute_retry_config) {
+        if !err.should_retry(num_retries) {
            return Err(err.into());
        }
        node_info
@@ -144,8 +132,7 @@ where
        // if we failed to connect, it's likely that the compute node was suspended, wake a new compute node
        info!("compute node's state has likely changed; requesting a wake-up");
        let old_node_info = invalidate_cache(node_info);
-        let mut node_info =
-            wake_compute(&mut num_retries, ctx, user_info, wake_compute_retry_config).await?;
+        let mut node_info = wake_compute(&mut num_retries, ctx, user_info).await?;
        node_info.reuse_settings(old_node_info);

        mechanism.update_connect_config(&mut node_info.config);
@@ -164,34 +151,19 @@ where
        {
            Ok(res) => {
                ctx.latency_timer.success();
-                Metrics::get().proxy.retries_metric.observe(
-                    RetriesMetricGroup {
-                        outcome: ConnectOutcome::Success,
-                        retry_type,
-                    },
-                    num_retries.into(),
-                );
-                info!(?num_retries, "connected to compute node after");
                return Ok(res);
            }
            Err(e) => {
-                let retriable = e.should_retry(num_retries, connect_to_compute_retry_config);
+                let retriable = e.should_retry(num_retries);
                if !retriable {
                    error!(error = ?e, num_retries, retriable, "couldn't connect to compute node");
-                    Metrics::get().proxy.retries_metric.observe(
-                        RetriesMetricGroup {
-                            outcome: ConnectOutcome::Failed,
-                            retry_type,
-                        },
-                        num_retries.into(),
-                    );
                    return Err(e.into());
                }
                warn!(error = ?e, num_retries, retriable, "couldn't connect to compute node");
            }
        }

-        let wait_duration = retry_after(num_retries, connect_to_compute_retry_config);
+        let wait_duration = retry_after(num_retries);
        num_retries += 1;

        time::sleep(wait_duration).await;
--- a/proxy/src/proxy/retry.rs
+++ b/proxy/src/proxy/retry.rs
@@ -1,12 +1,18 @@
-use crate::{compute, config::RetryConfig};
+use crate::compute;
 use std::{error::Error, io};
 use tokio::time;

+/// Number of times we should retry the `/proxy_wake_compute` http request.
+/// Retry duration is BASE_RETRY_WAIT_DURATION * RETRY_WAIT_EXPONENT_BASE ^ n, where n starts at 0
+pub const NUM_RETRIES_CONNECT: u32 = 16;
+const BASE_RETRY_WAIT_DURATION: time::Duration = time::Duration::from_millis(25);
+const RETRY_WAIT_EXPONENT_BASE: f64 = std::f64::consts::SQRT_2;
+
 pub trait ShouldRetry {
    fn could_retry(&self) -> bool;
-    fn should_retry(&self, num_retries: u32, config: RetryConfig) -> bool {
+    fn should_retry(&self, num_retries: u32) -> bool {
        match self {
-            _ if num_retries >= config.max_retries => false,
+            _ if num_retries >= NUM_RETRIES_CONNECT => false,
            err => err.could_retry(),
        }
    }
@@ -57,8 +63,6 @@ impl ShouldRetry for compute::ConnectionError {
    }
 }

-pub fn retry_after(num_retries: u32, config: RetryConfig) -> time::Duration {
-    config
-        .base_delay
-        .mul_f64(config.backoff_factor.powi((num_retries as i32) - 1))
+pub fn retry_after(num_retries: u32) -> time::Duration {
+    BASE_RETRY_WAIT_DURATION.mul_f64(RETRY_WAIT_EXPONENT_BASE.powi((num_retries as i32) - 1))
 }
--- a/proxy/src/proxy/tests.rs
+++ b/proxy/src/proxy/tests.rs
@@ -10,13 +10,13 @@ use super::*;
 use crate::auth::backend::{
    ComputeCredentialKeys, ComputeCredentials, ComputeUserInfo, MaybeOwned, TestBackend,
 };
-use crate::config::{CertResolver, RetryConfig};
+use crate::config::CertResolver;
 use crate::console::caches::NodeInfoCache;
 use crate::console::messages::MetricsAuxInfo;
 use crate::console::provider::{CachedAllowedIps, CachedRoleSecret, ConsoleBackend};
 use crate::console::{self, CachedNodeInfo, NodeInfo};
 use crate::error::ErrorKind;
-use crate::proxy::retry::retry_after;
+use crate::proxy::retry::{retry_after, NUM_RETRIES_CONNECT};
 use crate::{http, sasl, scram, BranchId, EndpointId, ProjectId};
 use anyhow::{bail, Context};
 use async_trait::async_trait;
@@ -361,15 +361,11 @@ async fn scram_auth_mock() -> anyhow::Result<()> {
 #[test]
 fn connect_compute_total_wait() {
    let mut total_wait = tokio::time::Duration::ZERO;
-    let config = RetryConfig {
-        base_delay: Duration::from_secs(1),
-        max_retries: 5,
-        backoff_factor: 2.0,
-    };
-    for num_retries in 1..config.max_retries {
-        total_wait += retry_after(num_retries, config);
+    for num_retries in 1..NUM_RETRIES_CONNECT {
+        total_wait += retry_after(num_retries);
    }
-    assert!(f64::abs(total_wait.as_secs_f64() - 15.0) < 0.1);
+    assert!(total_wait < tokio::time::Duration::from_secs(12));
+    assert!(total_wait > tokio::time::Duration::from_secs(10));
 }

 #[derive(Clone, Copy, Debug)]
@@ -553,12 +549,7 @@ async fn connect_to_compute_success() {
    let mut ctx = RequestMonitoring::test();
    let mechanism = TestConnectMechanism::new(vec![Wake, Connect]);
    let user_info = helper_create_connect_info(&mechanism);
-    let config = RetryConfig {
-        base_delay: Duration::from_secs(1),
-        max_retries: 5,
-        backoff_factor: 2.0,
-    };
-    connect_to_compute(&mut ctx, &mechanism, &user_info, false, config, config)
+    connect_to_compute(&mut ctx, &mechanism, &user_info, false)
        .await
        .unwrap();
    mechanism.verify();
@@ -571,12 +562,7 @@ async fn connect_to_compute_retry() {
    let mut ctx = RequestMonitoring::test();
    let mechanism = TestConnectMechanism::new(vec![Wake, Retry, Wake, Connect]);
    let user_info = helper_create_connect_info(&mechanism);
-    let config = RetryConfig {
-        base_delay: Duration::from_secs(1),
-        max_retries: 5,
-        backoff_factor: 2.0,
-    };
-    connect_to_compute(&mut ctx, &mechanism, &user_info, false, config, config)
+    connect_to_compute(&mut ctx, &mechanism, &user_info, false)
        .await
        .unwrap();
    mechanism.verify();
@@ -590,12 +576,7 @@ async fn connect_to_compute_non_retry_1() {
    let mut ctx = RequestMonitoring::test();
    let mechanism = TestConnectMechanism::new(vec![Wake, Retry, Wake, Fail]);
    let user_info = helper_create_connect_info(&mechanism);
-    let config = RetryConfig {
-        base_delay: Duration::from_secs(1),
-        max_retries: 5,
-        backoff_factor: 2.0,
-    };
-    connect_to_compute(&mut ctx, &mechanism, &user_info, false, config, config)
+    connect_to_compute(&mut ctx, &mechanism, &user_info, false)
        .await
        .unwrap_err();
    mechanism.verify();
@@ -609,12 +590,7 @@ async fn connect_to_compute_non_retry_2() {
    let mut ctx = RequestMonitoring::test();
    let mechanism = TestConnectMechanism::new(vec![Wake, Fail, Wake, Connect]);
    let user_info = helper_create_connect_info(&mechanism);
-    let config = RetryConfig {
-        base_delay: Duration::from_secs(1),
-        max_retries: 5,
-        backoff_factor: 2.0,
-    };
-    connect_to_compute(&mut ctx, &mechanism, &user_info, false, config, config)
+    connect_to_compute(&mut ctx, &mechanism, &user_info, false)
        .await
        .unwrap();
    mechanism.verify();
@@ -624,32 +600,17 @@ async fn connect_to_compute_non_retry_2() {
 #[tokio::test]
 async fn connect_to_compute_non_retry_3() {
    let _ = env_logger::try_init();
-    tokio::time::pause();
+    assert_eq!(NUM_RETRIES_CONNECT, 16);
    use ConnectAction::*;
    let mut ctx = RequestMonitoring::test();
-    let mechanism =
-        TestConnectMechanism::new(vec![Wake, Retry, Wake, Retry, Retry, Retry, Retry, Retry]);
+    let mechanism = TestConnectMechanism::new(vec![
+        Wake, Retry, Wake, Retry, Retry, Retry, Retry, Retry, Retry, Retry, Retry, Retry, Retry,
+        Retry, Retry, Retry, Retry, Retry, /* the 17th time */ Retry,
+    ]);
    let user_info = helper_create_connect_info(&mechanism);
-    let wake_compute_retry_config = RetryConfig {
-        base_delay: Duration::from_secs(1),
-        max_retries: 1,
-        backoff_factor: 2.0,
-    };
-    let connect_to_compute_retry_config = RetryConfig {
-        base_delay: Duration::from_secs(1),
-        max_retries: 5,
-        backoff_factor: 2.0,
-    };
-    connect_to_compute(
-        &mut ctx,
-        &mechanism,
-        &user_info,
-        false,
-        wake_compute_retry_config,
-        connect_to_compute_retry_config,
-    )
-    .await
-    .unwrap_err();
+    connect_to_compute(&mut ctx, &mechanism, &user_info, false)
+        .await
+        .unwrap_err();
    mechanism.verify();
 }

@@ -661,12 +622,7 @@ async fn wake_retry() {
    let mut ctx = RequestMonitoring::test();
    let mechanism = TestConnectMechanism::new(vec![WakeRetry, Wake, Connect]);
    let user_info = helper_create_connect_info(&mechanism);
-    let config = RetryConfig {
-        base_delay: Duration::from_secs(1),
-        max_retries: 5,
-        backoff_factor: 2.0,
-    };
-    connect_to_compute(&mut ctx, &mechanism, &user_info, false, config, config)
+    connect_to_compute(&mut ctx, &mechanism, &user_info, false)
        .await
        .unwrap();
    mechanism.verify();
@@ -680,12 +636,7 @@ async fn wake_non_retry() {
    let mut ctx = RequestMonitoring::test();
    let mechanism = TestConnectMechanism::new(vec![WakeRetry, WakeFail]);
    let user_info = helper_create_connect_info(&mechanism);
-    let config = RetryConfig {
-        base_delay: Duration::from_secs(1),
-        max_retries: 5,
-        backoff_factor: 2.0,
-    };
-    connect_to_compute(&mut ctx, &mechanism, &user_info, false, config, config)
+    connect_to_compute(&mut ctx, &mechanism, &user_info, false)
        .await
        .unwrap_err();
    mechanism.verify();
--- a/proxy/src/proxy/wake_compute.rs
+++ b/proxy/src/proxy/wake_compute.rs
@@ -1,14 +1,10 @@
-use crate::config::RetryConfig;
 use crate::console::{errors::WakeComputeError, provider::CachedNodeInfo};
 use crate::context::RequestMonitoring;
-use crate::metrics::{
-    ConnectOutcome, ConnectionFailuresBreakdownGroup, Metrics, RetriesMetricGroup, RetryType,
-    WakeupFailureKind,
-};
+use crate::metrics::{ConnectionFailuresBreakdownGroup, Metrics, WakeupFailureKind};
 use crate::proxy::retry::retry_after;
 use hyper::StatusCode;
 use std::ops::ControlFlow;
-use tracing::{error, info, warn};
+use tracing::{error, warn};

 use super::connect_compute::ComputeConnectBackend;
 use super::retry::ShouldRetry;
@@ -17,42 +13,23 @@ pub async fn wake_compute<B: ComputeConnectBackend>(
    num_retries: &mut u32,
    ctx: &mut RequestMonitoring,
    api: &B,
-    config: RetryConfig,
 ) -> Result<CachedNodeInfo, WakeComputeError> {
-    let retry_type = RetryType::WakeCompute;
    loop {
        let wake_res = api.wake_compute(ctx).await;
-        match handle_try_wake(wake_res, *num_retries, config) {
+        match handle_try_wake(wake_res, *num_retries) {
            Err(e) => {
                error!(error = ?e, num_retries, retriable = false, "couldn't wake compute node");
                report_error(&e, false);
-                Metrics::get().proxy.retries_metric.observe(
-                    RetriesMetricGroup {
-                        outcome: ConnectOutcome::Failed,
-                        retry_type,
-                    },
-                    (*num_retries).into(),
-                );
                return Err(e);
            }
            Ok(ControlFlow::Continue(e)) => {
                warn!(error = ?e, num_retries, retriable = true, "couldn't wake compute node");
                report_error(&e, true);
            }
-            Ok(ControlFlow::Break(n)) => {
-                Metrics::get().proxy.retries_metric.observe(
-                    RetriesMetricGroup {
-                        outcome: ConnectOutcome::Success,
-                        retry_type,
-                    },
-                    (*num_retries).into(),
-                );
-                info!(?num_retries, "compute node woken up after");
-                return Ok(n);
-            }
+            Ok(ControlFlow::Break(n)) => return Ok(n),
        }

-        let wait_duration = retry_after(*num_retries, config);
+        let wait_duration = retry_after(*num_retries);
        *num_retries += 1;
        tokio::time::sleep(wait_duration).await;
    }
@@ -65,11 +42,10 @@ pub async fn wake_compute<B: ComputeConnectBackend>(
 pub fn handle_try_wake(
    result: Result<CachedNodeInfo, WakeComputeError>,
    num_retries: u32,
-    config: RetryConfig,
 ) -> Result<ControlFlow<CachedNodeInfo, WakeComputeError>, WakeComputeError> {
    match result {
        Err(err) => match &err {
-            WakeComputeError::ApiError(api) if api.should_retry(num_retries, config) => {
+            WakeComputeError::ApiError(api) if api.should_retry(num_retries) => {
                Ok(ControlFlow::Continue(err))
            }
            _ => Err(err),
--- a/proxy/src/serverless/backend.rs
+++ b/proxy/src/serverless/backend.rs
@@ -108,8 +108,6 @@ impl PoolingBackend {
            },
            &backend,
            false, // do not allow self signed compute for http flow
-            self.config.wake_compute_retry_config,
-            self.config.connect_to_compute_retry_config,
        )
        .await
    }
--- a/test_runner/fixtures/metrics.py
+++ b/test_runner/fixtures/metrics.py
@@ -129,7 +129,7 @@ PAGESERVER_GLOBAL_METRICS: Tuple[str, ...] = (
    "pageserver_getpage_reconstruct_seconds_sum",
    *[f"pageserver_basebackup_query_seconds_{x}" for x in ["bucket", "count", "sum"]],
    *histogram("pageserver_smgr_query_seconds_global"),
-    *histogram("pageserver_layers_visited_per_read_global"),
+    *histogram("pageserver_read_num_fs_layers"),
    *histogram("pageserver_getpage_get_reconstruct_data_seconds"),
    *histogram("pageserver_wait_lsn_seconds"),
    *histogram("pageserver_remote_operation_seconds"),
--- a/test_runner/regress/test_compaction.py
+++ b/test_runner/regress/test_compaction.py
@@ -1,93 +0,0 @@
-import os
-
-import pytest
-from fixtures.log_helper import log
-from fixtures.neon_fixtures import NeonEnvBuilder
-from fixtures.workload import Workload
-
-AGGRESIVE_COMPACTION_TENANT_CONF = {
-    # Disable gc and compaction. The test runs compaction manually.
-    "gc_period": "0s",
-    "compaction_period": "0s",
-    # Small checkpoint distance to create many layers
-    "checkpoint_distance": 1024**2,
-    # Compact small layers
-    "compaction_target_size": 1024**2,
-    "image_creation_threshold": 2,
-    # INC-186: remove when merging the fix
-    "image_layer_creation_check_threshold": 0,
-}
-
-
-@pytest.mark.skipif(os.environ.get("BUILD_TYPE") == "debug", reason="only run with release build")
-def test_pageserver_compaction_smoke(neon_env_builder: NeonEnvBuilder):
-    """
-    This is a smoke test that compaction kicks in. The workload repeatedly churns
-    a small number of rows and manually instructs the pageserver to run compaction
-    between iterations. At the end of the test validate that the average number of
-    layers visited to gather reconstruct data for a given key is within the empirically
-    observed bounds.
-    """
-
-    # Effectively disable the page cache to rely only on image layers
-    # to shorten reads.
-    neon_env_builder.pageserver_config_override = """
-page_cache_size=10
-"""
-
-    env = neon_env_builder.init_start(initial_tenant_conf=AGGRESIVE_COMPACTION_TENANT_CONF)
-
-    tenant_id = env.initial_tenant
-    timeline_id = env.initial_timeline
-
-    row_count = 10000
-    churn_rounds = 100
-
-    ps_http = env.pageserver.http_client()
-
-    workload = Workload(env, tenant_id, timeline_id)
-    workload.init(env.pageserver.id)
-
-    log.info("Writing initial data ...")
-    workload.write_rows(row_count, env.pageserver.id)
-
-    for i in range(1, churn_rounds + 1):
-        if i % 10 == 0:
-            log.info(f"Running churn round {i}/{churn_rounds} ...")
-
-        workload.churn_rows(row_count, env.pageserver.id)
-        ps_http.timeline_compact(tenant_id, timeline_id)
-
-    log.info("Validating at workload end ...")
-    workload.validate(env.pageserver.id)
-
-    log.info("Checking layer access metrics ...")
-
-    layer_access_metric_names = [
-        "pageserver_layers_visited_per_read_global_sum",
-        "pageserver_layers_visited_per_read_global_count",
-        "pageserver_layers_visited_per_read_global_bucket",
-        "pageserver_layers_visited_per_vectored_read_global_sum",
-        "pageserver_layers_visited_per_vectored_read_global_count",
-        "pageserver_layers_visited_per_vectored_read_global_bucket",
-    ]
-
-    metrics = env.pageserver.http_client().get_metrics()
-    for name in layer_access_metric_names:
-        layer_access_metrics = metrics.query_all(name)
-        log.info(f"Got metrics: {layer_access_metrics}")
-
-    non_vectored_sum = metrics.query_one("pageserver_layers_visited_per_read_global_sum")
-    non_vectored_count = metrics.query_one("pageserver_layers_visited_per_read_global_count")
-    non_vectored_average = non_vectored_sum / non_vectored_count
-
-    vectored_sum = metrics.query_one("pageserver_layers_visited_per_vectored_read_global_sum")
-    vectored_count = metrics.query_one("pageserver_layers_visited_per_vectored_read_global_count")
-    vectored_average = vectored_sum / vectored_count
-
-    log.info(f"{non_vectored_average=} {vectored_average=}")
-
-    # The upper bound for average number of layer visits below (8)
-    # was chosen empirically for this workload.
-    assert non_vectored_average < 8
-    assert vectored_average < 8
--- a/test_runner/regress/test_sharding.py
+++ b/test_runner/regress/test_sharding.py
@@ -1201,45 +1201,3 @@ def test_sharding_backpressure(neon_env_builder: NeonEnvBuilder):
        max_lsn = max(Lsn(info["last_record_lsn"]) for info in infos)
        diff = max_lsn - min_lsn
        assert diff < 2 * 1024 * 1024, f"LSN diff={diff}, expected diff < 2MB due to backpressure"
-
-
-def test_sharding_unlogged_relation(neon_env_builder: NeonEnvBuilder):
-    """
-    Check that an unlogged relation is handled properly on a sharded tenant
-
-    Reproducer for https://github.com/neondatabase/neon/issues/7451
-    """
-
-    neon_env_builder.num_pageservers = 2
-    env = neon_env_builder.init_configs()
-    neon_env_builder.start()
-
-    tenant_id = TenantId.generate()
-    timeline_id = TimelineId.generate()
-    env.neon_cli.create_tenant(tenant_id, timeline_id, shard_count=8)
-
-    # We will create many tables to ensure it's overwhelmingly likely that at least one
-    # of them doesn't land on shard 0
-    table_names = [f"my_unlogged_{i}" for i in range(0, 16)]
-
-    with env.endpoints.create_start("main", tenant_id=tenant_id) as ep:
-        for table_name in table_names:
-            ep.safe_psql(f"CREATE UNLOGGED TABLE {table_name} (id integer, value varchar(64));")
-            ep.safe_psql(f"INSERT INTO {table_name} VALUES (1, 'foo')")
-            result = ep.safe_psql(f"SELECT * from {table_name};")
-            assert result == [(1, "foo")]
-            ep.safe_psql(f"CREATE INDEX ON {table_name} USING btree (value);")
-
-        wait_for_last_flush_lsn(env, ep, tenant_id, timeline_id)
-
-    with env.endpoints.create_start("main", tenant_id=tenant_id) as ep:
-        for table_name in table_names:
-            # Check that table works: we can select and insert
-            result = ep.safe_psql(f"SELECT * from {table_name};")
-            assert result == []
-            ep.safe_psql(f"INSERT INTO {table_name} VALUES (2, 'bar');")
-            result = ep.safe_psql(f"SELECT * from {table_name};")
-            assert result == [(2, "bar")]
-
-        # Ensure that post-endpoint-restart modifications are ingested happily by pageserver
-        wait_for_last_flush_lsn(env, ep, tenant_id, timeline_id)