[test_runner/performance] scale: Increase data volumes in PostgreSQL regression tests by 10x

Co-Authored-By: alexeymasterov@neon.tech <alexeymasterov@neon.tech>
2026-05-14 11:40:38 +00:00 · 2025-05-19 10:06:26 +00:00
75 changed files with 845 additions and 2901 deletions
--- a/.github/workflows/build_and_test.yml
+++ b/.github/workflows/build_and_test.yml
@@ -314,8 +314,7 @@ jobs:
          test_selection: performance
          run_in_parallel: false
          save_perf_report: ${{ github.ref_name == 'main' }}
-          # test_pageserver_max_throughput_getpage_at_latest_lsn is run in separate workflow periodic_pagebench.yml because it needs snapshots
-          extra_params: --splits 5 --group ${{ matrix.pytest_split_group }} --ignore=test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py
+          extra_params: --splits 5 --group ${{ matrix.pytest_split_group }}
          benchmark_durations: ${{ needs.get-benchmarks-durations.outputs.json }}
          pg_version: v16
          aws-oidc-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
--- a/.github/workflows/periodic_pagebench.yml
+++ b/.github/workflows/periodic_pagebench.yml
@@ -1,4 +1,4 @@
-name: Periodic pagebench performance test on unit-perf hetzner runner
+name: Periodic pagebench performance test on dedicated EC2 machine in eu-central-1 region

 on:
  schedule:
@@ -8,7 +8,7 @@ on:
    #        │   │ ┌───────────── day of the month (1 - 31)
    #        │   │ │ ┌───────────── month (1 - 12 or JAN-DEC)
    #        │   │ │ │ ┌───────────── day of the week (0 - 6 or SUN-SAT)
-    - cron: '0 */4 * * *' # Runs every 4 hours
+    - cron: '0 */3 * * *' # Runs every 3 hours
  workflow_dispatch: # Allows manual triggering of the workflow
    inputs:
      commit_hash:
@@ -16,11 +16,6 @@ on:
        description: 'The long neon repo commit hash for the system under test (pageserver) to be tested.'
        required: false
        default: ''
-      recreate_snapshots:
-        type: boolean
-        description: 'Recreate snapshots - !!!WARNING!!! We should only recreate snapshots if the previous ones are no longer compatible. Otherwise benchmarking results are not comparable across runs.'
-        required: false
-        default: false

 defaults:
  run:
@@ -34,13 +29,13 @@ permissions:
  contents: read

 jobs:
-  run_periodic_pagebench_test:
+  trigger_bench_on_ec2_machine_in_eu_central_1:
    permissions:
      id-token: write # aws-actions/configure-aws-credentials
      statuses: write
      contents: write
      pull-requests: write
-    runs-on: [ self-hosted, unit-perf ]
+    runs-on: [ self-hosted, small ]
    container:
      image: ghcr.io/neondatabase/build-tools:pinned-bookworm
      credentials:
@@ -49,13 +44,10 @@ jobs:
      options: --init
    timeout-minutes: 360  # Set the timeout to 6 hours
    env:
+      API_KEY: ${{ secrets.PERIODIC_PAGEBENCH_EC2_RUNNER_API_KEY }}
      RUN_ID: ${{ github.run_id }}
-      DEFAULT_PG_VERSION: 16
-      BUILD_TYPE: release
-      RUST_BACKTRACE: 1
-      # NEON_ENV_BUILDER_USE_OVERLAYFS_FOR_SNAPSHOTS: 1 - doesn't work without root in container
-      S3_BUCKET: neon-github-public-dev
-      PERF_TEST_RESULT_CONNSTR: "${{ secrets.PERF_TEST_RESULT_CONNSTR }}"
+      AWS_DEFAULT_REGION : "eu-central-1"
+      AWS_INSTANCE_ID : "i-02a59a3bf86bc7e74"
    steps:
    # we don't need the neon source code because we run everything remotely
    # however we still need the local github actions to run the allure step below
@@ -64,194 +56,99 @@ jobs:
      with:
        egress-policy: audit

-    - name: Set up the environment which depends on $RUNNER_TEMP on nvme drive
-      id: set-env
-      shell: bash -euxo pipefail {0}
-      run: |
-        {
-          echo "NEON_DIR=${RUNNER_TEMP}/neon"
-          echo "NEON_BIN=${RUNNER_TEMP}/neon/bin"
-          echo "POSTGRES_DISTRIB_DIR=${RUNNER_TEMP}/neon/pg_install"
-          echo "LD_LIBRARY_PATH=${RUNNER_TEMP}/neon/pg_install/v${DEFAULT_PG_VERSION}/lib"
-          echo "BACKUP_DIR=${RUNNER_TEMP}/instance_store/saved_snapshots"
-          echo "TEST_OUTPUT=${RUNNER_TEMP}/neon/test_output"
-          echo "PERF_REPORT_DIR=${RUNNER_TEMP}/neon/test_output/perf-report-local"
-          echo "ALLURE_DIR=${RUNNER_TEMP}/neon/test_output/allure-results"
-          echo "ALLURE_RESULTS_DIR=${RUNNER_TEMP}/neon/test_output/allure-results/results"
-        } >> "$GITHUB_ENV"
+    - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2

-        echo "allure_results_dir=${RUNNER_TEMP}/neon/test_output/allure-results/results" >> "$GITHUB_OUTPUT"
+    - name: Show my own (github runner) external IP address - usefull for IP allowlisting
+      run: curl https://ifconfig.me

-    - uses: aws-actions/configure-aws-credentials@e3dd6a429d7300a6a4c196c26e071d42e0343502 # v4.0.2
+    - name: Assume AWS OIDC role that allows to manage (start/stop/describe... EC machine)
+      uses: aws-actions/configure-aws-credentials@e3dd6a429d7300a6a4c196c26e071d42e0343502 # v4.0.2
      with:
        aws-region: eu-central-1
-        role-to-assume: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-        role-duration-seconds: 18000 # max 5 hours (needed in case commit hash is still being built)
+        role-to-assume: ${{ vars.DEV_AWS_OIDC_ROLE_MANAGE_BENCHMARK_EC2_VMS_ARN }}
+        role-duration-seconds: 3600
+
+    - name: Start EC2 instance and wait for the instance to boot up
+      run: |
+        aws ec2 start-instances --instance-ids $AWS_INSTANCE_ID
+        aws ec2 wait instance-running --instance-ids $AWS_INSTANCE_ID
+        sleep 60 # sleep some time to allow cloudinit and our API server to start up
+
+    - name: Determine public IP of the EC2 instance and set env variable EC2_MACHINE_URL_US
+      run: |
+        public_ip=$(aws ec2 describe-instances --instance-ids $AWS_INSTANCE_ID --query 'Reservations[*].Instances[*].PublicIpAddress' --output text)
+        echo "Public IP of the EC2 instance: $public_ip"
+        echo "EC2_MACHINE_URL_US=https://${public_ip}:8443" >> $GITHUB_ENV
+
    - name: Determine commit hash
-      id: commit_hash
-      shell: bash -euxo pipefail {0}
      env:
        INPUT_COMMIT_HASH: ${{ github.event.inputs.commit_hash }}
      run: |
-        if [[ -z "${INPUT_COMMIT_HASH}" ]]; then
-          COMMIT_HASH=$(curl -s https://api.github.com/repos/neondatabase/neon/commits/main | jq -r '.sha')
-          echo "COMMIT_HASH=$COMMIT_HASH" >> $GITHUB_ENV
-          echo "commit_hash=$COMMIT_HASH" >> "$GITHUB_OUTPUT"
+        if [ -z "$INPUT_COMMIT_HASH" ]; then
+          echo "COMMIT_HASH=$(curl -s https://api.github.com/repos/neondatabase/neon/commits/main | jq -r '.sha')" >> $GITHUB_ENV
          echo "COMMIT_HASH_TYPE=latest" >> $GITHUB_ENV
        else
-          COMMIT_HASH="${INPUT_COMMIT_HASH}"
-          echo "COMMIT_HASH=$COMMIT_HASH" >> $GITHUB_ENV
-          echo "commit_hash=$COMMIT_HASH" >> "$GITHUB_OUTPUT"
+          echo "COMMIT_HASH=$INPUT_COMMIT_HASH" >> $GITHUB_ENV
          echo "COMMIT_HASH_TYPE=manual" >> $GITHUB_ENV
        fi
-    - name: Checkout the neon repository at given commit hash
-      uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
-      with:
-        ref: ${{ steps.commit_hash.outputs.commit_hash }}

-    # does not reuse ./.github/actions/download because we need to download the artifact for the given commit hash
-    # example artifact
-    # s3://neon-github-public-dev/artifacts/48b870bc078bd2c450eb7b468e743b9c118549bf/15036827400/1/neon-Linux-X64-release-artifact.tar.zst /instance_store/artifacts/neon-Linux-release-artifact.tar.zst
-    - name: Determine artifact S3_KEY for given commit hash and download and extract artifact
-      id: artifact_prefix
-      shell: bash -euxo pipefail {0}
-      env:
-        ARCHIVE: ${{ runner.temp }}/downloads/neon-${{ runner.os }}-${{ runner.arch }}-release-artifact.tar.zst
-        COMMIT_HASH: ${{ env.COMMIT_HASH }}
-        COMMIT_HASH_TYPE: ${{ env.COMMIT_HASH_TYPE }}
+    - name: Start Bench with run_id
      run: |
-        attempt=0
-        max_attempts=24 # 5 minutes * 24 = 2 hours
+        curl -k -X 'POST' \
+        "${EC2_MACHINE_URL_US}/start_test/${GITHUB_RUN_ID}" \
+        -H 'accept: application/json' \
+        -H 'Content-Type: application/json' \
+        -H "Authorization: Bearer $API_KEY" \
+        -d "{\"neonRepoCommitHash\": \"${COMMIT_HASH}\", \"neonRepoCommitHashType\": \"${COMMIT_HASH_TYPE}\"}"

-        while [[ $attempt -lt $max_attempts ]]; do
-          # the following command will fail until the artifacts are available ...
-          S3_KEY=$(aws s3api list-objects-v2 --bucket "$S3_BUCKET" --prefix "artifacts/$COMMIT_HASH/" \
-            | jq -r '.Contents[]?.Key' \
-            | grep "neon-${{ runner.os }}-${{ runner.arch }}-release-artifact.tar.zst" \
-            | sort --version-sort \
-            | tail -1) || true # ... thus ignore errors from the command
-          if [[ -n "${S3_KEY}" ]]; then
-            echo "Artifact found: $S3_KEY"
-            echo "S3_KEY=$S3_KEY" >> $GITHUB_ENV
+    - name: Poll Test Status
+      id: poll_step
+      run: |
+        status=""
+        while [[ "$status" != "failure" && "$status" != "success" ]]; do
+          response=$(curl -k -X 'GET' \
+          "${EC2_MACHINE_URL_US}/test_status/${GITHUB_RUN_ID}" \
+          -H 'accept: application/json' \
+          -H "Authorization: Bearer $API_KEY")
+          echo "Response: $response"
+          set +x
+          status=$(echo $response | jq -r '.status')
+          echo "Test status: $status"
+          if [[ "$status" == "failure" ]]; then
+            echo "Test failed"
+            exit 1 # Fail the job step if status is failure
+          elif [[ "$status" == "success" || "$status" == "null" ]]; then
            break
+          elif [[ "$status" == "too_many_runs" ]]; then
+            echo "Too many runs already running"
+            echo "too_many_runs=true" >> "$GITHUB_OUTPUT"
+            exit 1
          fi
-          
-          # Increment attempt counter and sleep for 5 minutes
-          attempt=$((attempt + 1))
-          echo "Attempt $attempt of $max_attempts to find artifacts in S3 bucket s3://$S3_BUCKET/artifacts/$COMMIT_HASH failed. Retrying in 5 minutes..."
-          sleep 300 # Sleep for 5 minutes
+
+          sleep 60 # Poll every 60 seconds
        done

-        if [[ -z "${S3_KEY}" ]]; then
-          echo "Error: artifact not found in S3 bucket s3://$S3_BUCKET/artifacts/$COMMIT_HASH" after 2 hours
-        else
-          mkdir -p $(dirname $ARCHIVE)
-          time aws s3 cp --only-show-errors s3://$S3_BUCKET/${S3_KEY} ${ARCHIVE}
-          mkdir -p ${NEON_DIR}
-          time tar -xf ${ARCHIVE} -C ${NEON_DIR}
-          rm -f ${ARCHIVE}
-        fi
-
-    - name: Download snapshots from S3
-      if: ${{ github.event_name != 'workflow_dispatch' || github.event.inputs.recreate_snapshots == 'false' || github.event.inputs.recreate_snapshots == '' }}
-      id: download_snapshots
-      shell: bash -euxo pipefail {0}
+    - name: Retrieve Test Logs
+      if: always() && steps.poll_step.outputs.too_many_runs != 'true'
      run: |
-        # Download the snapshots from S3
-        mkdir -p ${TEST_OUTPUT}
-        mkdir -p $BACKUP_DIR
-        cd $BACKUP_DIR
-        mkdir parts
-        cd parts
-        PART=$(aws s3api list-objects-v2 --bucket $S3_BUCKET --prefix performance/pagebench/ \
-          | jq -r '.Contents[]?.Key' \
-          | grep -E 'shared-snapshots-[0-9]{4}-[0-9]{2}-[0-9]{2}' \
-          | sort \
-          | tail -1)
-        echo "Latest PART: $PART"
-        if [[ -z "$PART" ]]; then
-          echo "ERROR: No matching S3 key found" >&2
-          exit 1
-        fi
-        S3_KEY=$(dirname $PART)
-        time aws s3 cp --only-show-errors --recursive s3://${S3_BUCKET}/$S3_KEY/ .
-        cd $TEST_OUTPUT
-        time cat $BACKUP_DIR/parts/* | zstdcat | tar --extract --preserve-permissions
-        rm -rf ${BACKUP_DIR}
+        curl -k -X 'GET' \
+        "${EC2_MACHINE_URL_US}/test_log/${GITHUB_RUN_ID}" \
+        -H 'accept: application/gzip' \
+        -H "Authorization: Bearer $API_KEY" \
+        --output "test_log_${GITHUB_RUN_ID}.gz"

-    - name: Cache poetry deps
-      uses: actions/cache@v4
-      with:
-        path: ~/.cache/pypoetry/virtualenvs
-        key: v2-${{ runner.os }}-${{ runner.arch }}-python-deps-bookworm-${{ hashFiles('poetry.lock') }}
-
-    - name: Install Python deps
-      shell: bash -euxo pipefail {0}
-      run: ./scripts/pysync
-
-    # we need high number of open files for pagebench
-    - name: show ulimits
-      shell: bash -euxo pipefail {0}
+    - name: Unzip Test Log and Print it into this job's log
+      if: always() && steps.poll_step.outputs.too_many_runs != 'true'
      run: |
-        ulimit -a
-
-    - name: Run pagebench testcase
-      shell: bash -euxo pipefail {0}
-      env:
-        CI: false  # need to override this env variable set by github to enforce using snapshots
-      run: |
-        export PLATFORM=hetzner-unit-perf-${COMMIT_HASH_TYPE}
-        # report the commit hash of the neon repository in the revision of the test results
-        export GITHUB_SHA=${COMMIT_HASH}
-        rm -rf ${PERF_REPORT_DIR}
-        rm -rf ${ALLURE_RESULTS_DIR}
-        mkdir -p ${PERF_REPORT_DIR}
-        mkdir -p ${ALLURE_RESULTS_DIR}
-        PARAMS="--alluredir=${ALLURE_RESULTS_DIR} --tb=short --verbose -rA"
-        EXTRA_PARAMS="--out-dir ${PERF_REPORT_DIR} --durations-path $TEST_OUTPUT/benchmark_durations.json"
-        # run only two selected tests
-        # environment set by parent:
-        # RUST_BACKTRACE=1 DEFAULT_PG_VERSION=16 BUILD_TYPE=release
-        ./scripts/pytest ${PARAMS} test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py::test_pageserver_characterize_throughput_with_n_tenants ${EXTRA_PARAMS}
-        ./scripts/pytest ${PARAMS} test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py::test_pageserver_characterize_latencies_with_1_client_and_throughput_with_many_clients_one_tenant ${EXTRA_PARAMS}
-
-    - name: upload the performance metrics to the Neon performance database which is used by grafana dashboards to display the results
-      shell: bash -euxo pipefail {0}
-      run: |
-        export REPORT_FROM="$PERF_REPORT_DIR"
-        export GITHUB_SHA=${COMMIT_HASH}
-        time ./scripts/generate_and_push_perf_report.sh
-
-    - name: Upload test results
-      if: ${{ !cancelled() }}
-      uses: ./.github/actions/allure-report-store
-      with:
-        report-dir:  ${{ steps.set-env.outputs.allure_results_dir }}
-        unique-key: ${{ env.BUILD_TYPE }}-${{ env.DEFAULT_PG_VERSION }}-${{ runner.arch }}
-        aws-oidc-role-arn:  ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
+        gzip -d "test_log_${GITHUB_RUN_ID}.gz"
+        cat "test_log_${GITHUB_RUN_ID}"

    - name: Create Allure report
-      id: create-allure-report
      if: ${{ !cancelled() }}
      uses: ./.github/actions/allure-report-generate
      with:
        aws-oidc-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}

-    - name: Upload snapshots
-      if: ${{ github.event_name == 'workflow_dispatch' && github.event.inputs.recreate_snapshots != 'false' && github.event.inputs.recreate_snapshots != '' }}
-      id: upload_snapshots
-      shell: bash -euxo pipefail {0}
-      run: |
-        mkdir -p $BACKUP_DIR
-        cd $TEST_OUTPUT
-        tar --create --preserve-permissions --file - shared-snapshots | zstd -o $BACKUP_DIR/shared_snapshots.tar.zst
-        cd $BACKUP_DIR
-        mkdir parts
-        split -b 1G shared_snapshots.tar.zst ./parts/shared_snapshots.tar.zst.part.
-        SNAPSHOT_DATE=$(date +%F)  # YYYY-MM-DD
-        cd parts
-        time aws s3 cp --recursive . s3://${S3_BUCKET}/performance/pagebench/shared-snapshots-${SNAPSHOT_DATE}/
-
    - name: Post to a Slack channel
      if: ${{ github.event.schedule && failure() }}
      uses: slackapi/slack-github-action@fcfb566f8b0aab22203f066d80ca1d7e4b5d05b3 # v1.27.1
@@ -260,22 +157,26 @@ jobs:
        slack-message: "Periodic pagebench testing on dedicated hardware: ${{ job.status }}\n${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
      env:
        SLACK_BOT_TOKEN: ${{ secrets.SLACK_BOT_TOKEN }}
-        
+
    - name: Cleanup Test Resources
      if: always()
-      shell: bash -euxo pipefail {0}
-      env:
-        ARCHIVE: ${{ runner.temp }}/downloads/neon-${{ runner.os }}-${{ runner.arch }}-release-artifact.tar.zst
      run: |
-        # Cleanup the test resources
-        if [[ -d "${BACKUP_DIR}" ]]; then
-          rm -rf ${BACKUP_DIR}
-        fi
-        if [[ -d "${TEST_OUTPUT}" ]]; then
-          rm -rf ${TEST_OUTPUT}
-        fi
-        if [[ -d "${NEON_DIR}" ]]; then
-          rm -rf ${NEON_DIR}
-        fi
-        rm -rf $(dirname $ARCHIVE)
+        curl -k -X 'POST' \
+        "${EC2_MACHINE_URL_US}/cleanup_test/${GITHUB_RUN_ID}" \
+        -H 'accept: application/json' \
+        -H "Authorization: Bearer $API_KEY" \
+        -d ''

+    - name: Assume AWS OIDC role that allows to manage (start/stop/describe... EC machine)
+      if: always() && steps.poll_step.outputs.too_many_runs != 'true'
+      uses: aws-actions/configure-aws-credentials@e3dd6a429d7300a6a4c196c26e071d42e0343502 # v4.0.2
+      with:
+        aws-region: eu-central-1
+        role-to-assume: ${{ vars.DEV_AWS_OIDC_ROLE_MANAGE_BENCHMARK_EC2_VMS_ARN }}
+        role-duration-seconds: 3600
+
+    - name: Stop EC2 instance and wait for the instance to be stopped
+      if: always() && steps.poll_step.outputs.too_many_runs != 'true'
+      run: |
+        aws ec2 stop-instances --instance-ids $AWS_INSTANCE_ID
+        aws ec2 wait instance-stopped --instance-ids $AWS_INSTANCE_ID
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1276,7 +1276,7 @@ version = "0.1.0"
 dependencies = [
 "anyhow",
 "chrono",
- "indexmap 2.9.0",
+ "indexmap 2.0.1",
 "jsonwebtoken",
 "regex",
 "remote_storage",
@@ -1308,7 +1308,7 @@ dependencies = [
 "flate2",
 "futures",
 "http 1.1.0",
- "indexmap 2.9.0",
+ "indexmap 2.0.1",
 "itertools 0.10.5",
 "jsonwebtoken",
 "metrics",
@@ -2597,7 +2597,7 @@ dependencies = [
 "futures-sink",
 "futures-util",
 "http 0.2.9",
- "indexmap 2.9.0",
+ "indexmap 2.0.1",
 "slab",
 "tokio",
 "tokio-util",
@@ -2616,7 +2616,7 @@ dependencies = [
 "futures-sink",
 "futures-util",
 "http 1.1.0",
- "indexmap 2.9.0",
+ "indexmap 2.0.1",
 "slab",
 "tokio",
 "tokio-util",
@@ -2863,14 +2863,14 @@ dependencies = [
 "pprof",
 "regex",
 "routerify",
- "rustls 0.23.27",
+ "rustls 0.23.18",
 "rustls-pemfile 2.1.1",
 "serde",
 "serde_json",
 "serde_path_to_error",
 "thiserror 1.0.69",
 "tokio",
- "tokio-rustls 0.26.2",
+ "tokio-rustls 0.26.0",
 "tokio-stream",
 "tokio-util",
 "tracing",
@@ -3200,12 +3200,12 @@ dependencies = [

 [[package]]
 name = "indexmap"
-version = "2.9.0"
+version = "2.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cea70ddb795996207ad57735b50c5982d8844f38ba9ee5f1aedcfb708a2aa11e"
+checksum = "ad227c3af19d4914570ad36d30409928b75967c298feb9ea1969db3a610bb14e"
 dependencies = [
 "equivalent",
- "hashbrown 0.15.2",
+ "hashbrown 0.14.5",
 "serde",
 ]

@@ -3228,7 +3228,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "232929e1d75fe899576a3d5c7416ad0d88dbfbb3c3d6aa00873a7408a50ddb88"
 dependencies = [
 "ahash",
- "indexmap 2.9.0",
+ "indexmap 2.0.1",
 "is-terminal",
 "itoa",
 "log",
@@ -3251,7 +3251,7 @@ dependencies = [
 "crossbeam-utils",
 "dashmap 6.1.0",
 "env_logger",
- "indexmap 2.9.0",
+ "indexmap 2.0.1",
 "itoa",
 "log",
 "num-format",
@@ -3898,16 +3898,6 @@ dependencies = [
 "winapi",
 ]

-[[package]]
-name = "nu-ansi-term"
-version = "0.46.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "77a8165726e8236064dbb45459242600304b42a5ea24ee2948e18e023bf7ba84"
-dependencies = [
- "overload",
- "winapi",
-]
-
 [[package]]
 name = "num"
 version = "0.4.1"
@@ -4112,7 +4102,7 @@ dependencies = [
 "opentelemetry-http",
 "opentelemetry-proto",
 "opentelemetry_sdk",
- "prost 0.13.5",
+ "prost 0.13.3",
 "reqwest",
 "thiserror 1.0.69",
 ]
@@ -4125,8 +4115,8 @@ checksum = "a6e05acbfada5ec79023c85368af14abd0b307c015e9064d249b2a950ef459a6"
 dependencies = [
 "opentelemetry",
 "opentelemetry_sdk",
- "prost 0.13.5",
- "tonic 0.12.3",
+ "prost 0.13.3",
+ "tonic",
 ]

 [[package]]
@@ -4192,12 +4182,6 @@ version = "0.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4030760ffd992bef45b0ae3f10ce1aba99e33464c90d14dd7c039884963ddc7a"

-[[package]]
-name = "overload"
-version = "0.1.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b15813163c1d831bf4a13c3610c05c0d03b39feb07f7e09fa234dac9b15aaf39"
-
 [[package]]
 name = "p256"
 version = "0.11.1"
@@ -4302,7 +4286,6 @@ dependencies = [
 "enumset",
 "fail",
 "futures",
- "hashlink",
 "hex",
 "hex-literal",
 "http-utils",
@@ -4339,7 +4322,7 @@ dependencies = [
 "reqwest",
 "rpds",
 "rstest",
- "rustls 0.23.27",
+ "rustls 0.23.18",
 "scopeguard",
 "send-future",
 "serde",
@@ -4358,7 +4341,7 @@ dependencies = [
 "tokio-epoll-uring",
 "tokio-io-timeout",
 "tokio-postgres",
- "tokio-rustls 0.26.2",
+ "tokio-rustls 0.26.0",
 "tokio-stream",
 "tokio-tar",
 "tokio-util",
@@ -4455,8 +4438,8 @@ dependencies = [
 name = "pageserver_page_api"
 version = "0.1.0"
 dependencies = [
- "prost 0.13.5",
- "tonic 0.13.1",
+ "prost 0.13.3",
+ "tonic",
 "tonic-build",
 "workspace_hack",
 ]
@@ -4837,14 +4820,14 @@ dependencies = [
 "bytes",
 "once_cell",
 "pq_proto",
- "rustls 0.23.27",
+ "rustls 0.23.18",
 "rustls-pemfile 2.1.1",
 "serde",
 "thiserror 1.0.69",
 "tokio",
 "tokio-postgres",
 "tokio-postgres-rustls",
- "tokio-rustls 0.26.2",
+ "tokio-rustls 0.26.0",
 "tokio-util",
 "tracing",
 ]
@@ -4951,7 +4934,7 @@ dependencies = [
 "inferno 0.12.0",
 "num",
 "paste",
- "prost 0.13.5",
+ "prost 0.13.3",
 ]

 [[package]]
@@ -5056,12 +5039,12 @@ dependencies = [

 [[package]]
 name = "prost"
-version = "0.13.5"
+version = "0.13.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2796faa41db3ec313a31f7624d9286acf277b52de526150b7e69f3debf891ee5"
+checksum = "7b0487d90e047de87f984913713b85c601c05609aad5b0df4b4573fbf69aa13f"
 dependencies = [
 "bytes",
- "prost-derive 0.13.5",
+ "prost-derive 0.13.3",
 ]

 [[package]]
@@ -5099,7 +5082,7 @@ dependencies = [
 "once_cell",
 "petgraph",
 "prettyplease",
- "prost 0.13.5",
+ "prost 0.13.3",
 "prost-types 0.13.3",
 "regex",
 "syn 2.0.100",
@@ -5121,9 +5104,9 @@ dependencies = [

 [[package]]
 name = "prost-derive"
-version = "0.13.5"
+version = "0.13.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d"
+checksum = "e9552f850d5f0964a4e4d0bf306459ac29323ddfbae05e35a7c0d35cb0803cc5"
 dependencies = [
 "anyhow",
 "itertools 0.12.1",
@@ -5147,7 +5130,7 @@ version = "0.13.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4759aa0d3a6232fb8dbdb97b61de2c20047c68aca932c7ed76da9d788508d670"
 dependencies = [
- "prost 0.13.5",
+ "prost 0.13.3",
 ]

 [[package]]
@@ -5195,7 +5178,7 @@ dependencies = [
 "hyper 0.14.30",
 "hyper 1.4.1",
 "hyper-util",
- "indexmap 2.9.0",
+ "indexmap 2.0.1",
 "ipnet",
 "itertools 0.10.5",
 "itoa",
@@ -5229,7 +5212,7 @@ dependencies = [
 "rsa",
 "rstest",
 "rustc-hash 1.1.0",
- "rustls 0.23.27",
+ "rustls 0.23.18",
 "rustls-native-certs 0.8.0",
 "rustls-pemfile 2.1.1",
 "scopeguard",
@@ -5248,14 +5231,13 @@ dependencies = [
 "tokio",
 "tokio-postgres",
 "tokio-postgres2",
- "tokio-rustls 0.26.2",
+ "tokio-rustls 0.26.0",
 "tokio-tungstenite 0.21.0",
 "tokio-util",
 "tracing",
 "tracing-log",
 "tracing-opentelemetry",
 "tracing-subscriber",
- "tracing-test",
 "tracing-utils",
 "try-lock",
 "typed-json",
@@ -5472,13 +5454,13 @@ dependencies = [
 "num-bigint",
 "percent-encoding",
 "pin-project-lite",
- "rustls 0.23.27",
+ "rustls 0.23.18",
 "rustls-native-certs 0.8.0",
 "ryu",
 "sha1_smol",
 "socket2",
 "tokio",
- "tokio-rustls 0.26.2",
+ "tokio-rustls 0.26.0",
 "tokio-util",
 "url",
 ]
@@ -5926,15 +5908,15 @@ dependencies = [

 [[package]]
 name = "rustls"
-version = "0.23.27"
+version = "0.23.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "730944ca083c1c233a75c09f199e973ca499344a2b7ba9e755c457e86fb4a321"
+checksum = "9c9cc1d47e243d655ace55ed38201c19ae02c148ae56412ab8750e8f0166ab7f"
 dependencies = [
 "log",
 "once_cell",
 "ring",
 "rustls-pki-types",
- "rustls-webpki 0.103.3",
+ "rustls-webpki 0.102.8",
 "subtle",
 "zeroize",
 ]
@@ -6023,17 +6005,6 @@ dependencies = [
 "untrusted",
 ]

-[[package]]
-name = "rustls-webpki"
-version = "0.103.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e4a72fe2bcf7a6ac6fd7d0b9e5cb68aeb7d4c0a0271730218b3e92d43b4eb435"
-dependencies = [
- "ring",
- "rustls-pki-types",
- "untrusted",
-]
-
 [[package]]
 name = "rustversion"
 version = "1.0.12"
@@ -6085,7 +6056,7 @@ dependencies = [
 "regex",
 "remote_storage",
 "reqwest",
- "rustls 0.23.27",
+ "rustls 0.23.18",
 "safekeeper_api",
 "safekeeper_client",
 "scopeguard",
@@ -6102,7 +6073,7 @@ dependencies = [
 "tokio",
 "tokio-io-timeout",
 "tokio-postgres",
- "tokio-rustls 0.26.2",
+ "tokio-rustls 0.26.0",
 "tokio-stream",
 "tokio-tar",
 "tokio-util",
@@ -6274,7 +6245,7 @@ checksum = "255914a8e53822abd946e2ce8baa41d4cded6b8e938913b7f7b9da5b7ab44335"
 dependencies = [
 "httpdate",
 "reqwest",
- "rustls 0.23.27",
+ "rustls 0.23.18",
 "sentry-backtrace",
 "sentry-contexts",
 "sentry-core",
@@ -6703,11 +6674,11 @@ dependencies = [
 "metrics",
 "once_cell",
 "parking_lot 0.12.1",
- "prost 0.13.5",
- "rustls 0.23.27",
+ "prost 0.13.3",
+ "rustls 0.23.18",
 "tokio",
- "tokio-rustls 0.26.2",
- "tonic 0.13.1",
+ "tokio-rustls 0.26.0",
+ "tonic",
 "tonic-build",
 "tracing",
 "utils",
@@ -6749,7 +6720,7 @@ dependencies = [
 "regex",
 "reqwest",
 "routerify",
- "rustls 0.23.27",
+ "rustls 0.23.18",
 "rustls-native-certs 0.8.0",
 "safekeeper_api",
 "safekeeper_client",
@@ -6764,7 +6735,7 @@ dependencies = [
 "tokio",
 "tokio-postgres",
 "tokio-postgres-rustls",
- "tokio-rustls 0.26.2",
+ "tokio-rustls 0.26.0",
 "tokio-util",
 "tracing",
 "utils",
@@ -6802,7 +6773,7 @@ dependencies = [
 "postgres_ffi",
 "remote_storage",
 "reqwest",
- "rustls 0.23.27",
+ "rustls 0.23.18",
 "rustls-native-certs 0.8.0",
 "serde",
 "serde_json",
@@ -7336,10 +7307,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "04fb792ccd6bbcd4bba408eb8a292f70fc4a3589e5d793626f45190e6454b6ab"
 dependencies = [
 "ring",
- "rustls 0.23.27",
+ "rustls 0.23.18",
 "tokio",
 "tokio-postgres",
- "tokio-rustls 0.26.2",
+ "tokio-rustls 0.26.0",
 "x509-certificate",
 ]

@@ -7383,11 +7354,12 @@ dependencies = [

 [[package]]
 name = "tokio-rustls"
-version = "0.26.2"
+version = "0.26.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8e727b36a1a0e8b74c376ac2211e40c2c8af09fb4013c60d910495810f008e9b"
+checksum = "0c7bc40d0e5a97695bb96e27995cd3a08538541b0a846f65bba7a359f36700d4"
 dependencies = [
- "rustls 0.23.27",
+ "rustls 0.23.18",
+ "rustls-pki-types",
 "tokio",
 ]

@@ -7485,7 +7457,7 @@ version = "0.22.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f21c7aaf97f1bd9ca9d4f9e73b0a6c74bd5afef56f2bc931943a6e1c37e04e38"
 dependencies = [
- "indexmap 2.9.0",
+ "indexmap 2.0.1",
 "serde",
 "serde_spanned",
 "toml_datetime",
@@ -7497,27 +7469,6 @@ name = "tonic"
 version = "0.12.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "877c5b330756d856ffcc4553ab34a5684481ade925ecc54bcd1bf02b1d0d4d52"
-dependencies = [
- "async-trait",
- "base64 0.22.1",
- "bytes",
- "http 1.1.0",
- "http-body 1.0.0",
- "http-body-util",
- "percent-encoding",
- "pin-project",
- "prost 0.13.5",
- "tokio-stream",
- "tower-layer",
- "tower-service",
- "tracing",
-]
-
-[[package]]
-name = "tonic"
-version = "0.13.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7e581ba15a835f4d9ea06c55ab1bd4dce26fc53752c69a04aac00703bfb49ba9"
 dependencies = [
 "async-trait",
 "base64 0.22.1",
@@ -7530,12 +7481,13 @@ dependencies = [
 "hyper-util",
 "percent-encoding",
 "pin-project",
- "prost 0.13.5",
+ "prost 0.13.3",
 "rustls-native-certs 0.8.0",
+ "rustls-pemfile 2.1.1",
 "tokio",
- "tokio-rustls 0.26.2",
+ "tokio-rustls 0.26.0",
 "tokio-stream",
- "tower 0.5.2",
+ "tower 0.4.13",
 "tower-layer",
 "tower-service",
 "tracing",
@@ -7543,9 +7495,9 @@ dependencies = [

 [[package]]
 name = "tonic-build"
-version = "0.13.1"
+version = "0.12.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "eac6f67be712d12f0b41328db3137e0d0757645d8904b4cb7d51cd9c2279e847"
+checksum = "9557ce109ea773b399c9b9e5dca39294110b74f1f342cb347a80d1fce8c26a11"
 dependencies = [
 "prettyplease",
 "proc-macro2",
@@ -7563,11 +7515,16 @@ checksum = "b8fa9be0de6cf49e536ce1851f987bd21a43b771b09473c3549a6c853db37c1c"
 dependencies = [
 "futures-core",
 "futures-util",
+ "indexmap 1.9.3",
 "pin-project",
 "pin-project-lite",
+ "rand 0.8.5",
+ "slab",
 "tokio",
+ "tokio-util",
 "tower-layer",
 "tower-service",
+ "tracing",
 ]

 [[package]]
@@ -7578,12 +7535,9 @@ checksum = "d039ad9159c98b70ecfd540b2573b97f7f52c3e8d9f8ad57a24b916a536975f9"
 dependencies = [
 "futures-core",
 "futures-util",
- "indexmap 2.9.0",
 "pin-project-lite",
- "slab",
 "sync_wrapper 1.0.1",
 "tokio",
- "tokio-util",
 "tower-layer",
 "tower-service",
 "tracing",
@@ -7734,7 +7688,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e8189decb5ac0fa7bc8b96b7cb9b2701d60d48805aca84a238004d665fcc4008"
 dependencies = [
 "matchers",
- "nu-ansi-term",
 "once_cell",
 "regex",
 "serde",
@@ -7748,27 +7701,6 @@ dependencies = [
 "tracing-serde",
 ]

-[[package]]
-name = "tracing-test"
-version = "0.2.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "557b891436fe0d5e0e363427fc7f217abf9ccd510d5136549847bdcbcd011d68"
-dependencies = [
- "tracing-core",
- "tracing-subscriber",
- "tracing-test-macro",
-]
-
-[[package]]
-name = "tracing-test-macro"
-version = "0.2.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "04659ddb06c87d233c566112c1c9c5b9e98256d9af50ec3bc9c8327f873a7568"
-dependencies = [
- "quote",
- "syn 2.0.100",
-]
-
 [[package]]
 name = "tracing-utils"
 version = "0.1.0"
@@ -7911,7 +7843,7 @@ dependencies = [
 "base64 0.22.1",
 "log",
 "once_cell",
- "rustls 0.23.27",
+ "rustls 0.23.18",
 "rustls-pki-types",
 "url",
 "webpki-roots",
@@ -8106,7 +8038,7 @@ dependencies = [
 "pageserver_api",
 "postgres_ffi",
 "pprof",
- "prost 0.13.5",
+ "prost 0.13.3",
 "remote_storage",
 "serde",
 "serde_json",
@@ -8562,7 +8494,8 @@ dependencies = [
 "hyper 0.14.30",
 "hyper 1.4.1",
 "hyper-util",
- "indexmap 2.9.0",
+ "indexmap 1.9.3",
+ "indexmap 2.0.1",
 "itertools 0.12.1",
 "lazy_static",
 "libc",
@@ -8584,16 +8517,16 @@ dependencies = [
 "percent-encoding",
 "prettyplease",
 "proc-macro2",
- "prost 0.13.5",
+ "prost 0.13.3",
 "quote",
 "rand 0.8.5",
 "regex",
 "regex-automata 0.4.3",
 "regex-syntax 0.8.2",
 "reqwest",
- "rustls 0.23.27",
+ "rustls 0.23.18",
 "rustls-pki-types",
- "rustls-webpki 0.103.3",
+ "rustls-webpki 0.102.8",
 "scopeguard",
 "sec1 0.7.3",
 "serde",
@@ -8611,15 +8544,15 @@ dependencies = [
 "time",
 "time-macros",
 "tokio",
- "tokio-rustls 0.26.2",
+ "tokio-rustls 0.26.0",
 "tokio-stream",
 "tokio-util",
 "toml_edit",
- "tower 0.5.2",
+ "tonic",
+ "tower 0.4.13",
 "tracing",
 "tracing-core",
 "tracing-log",
- "tracing-subscriber",
 "url",
 "uuid",
 "zeroize",
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -149,7 +149,7 @@ pin-project-lite = "0.2"
 pprof = { version = "0.14", features = ["criterion", "flamegraph", "frame-pointer", "prost-codec"] }
 procfs = "0.16"
 prometheus = {version = "0.13", default-features=false, features = ["process"]} # removes protobuf dependency
-prost = "0.13.5"
+prost = "0.13"
 rand = "0.8"
 redis = { version = "0.29.2", features = ["tokio-rustls-comp", "keep-alive"] }
 regex = "1.10.2"
@@ -199,7 +199,7 @@ tokio-tar = "0.3"
 tokio-util = { version = "0.7.10", features = ["io", "rt"] }
 toml = "0.8"
 toml_edit = "0.22"
-tonic = { version = "0.13.1", default-features = false, features = ["channel", "codegen", "prost", "tls-ring", "tls-native-roots"] }
+tonic = {version = "0.12.3", default-features = false, features = ["channel", "tls", "tls-roots"]}
 tower = { version = "0.5.2", default-features = false }
 tower-http = { version = "0.6.2", features = ["auth", "request-id", "trace"] }

@@ -280,7 +280,7 @@ criterion = "0.5.1"
 rcgen = "0.13"
 rstest = "0.18"
 camino-tempfile = "1.0.2"
-tonic-build = "0.13.1"
+tonic-build = "0.12"

 [patch.crates-io]

--- a/compute/compute-node.Dockerfile
+++ b/compute/compute-node.Dockerfile
@@ -582,38 +582,6 @@ RUN make -j $(getconf _NPROCESSORS_ONLN) && \
    make -j $(getconf _NPROCESSORS_ONLN) install && \
    echo 'trusted = true' >> /usr/local/pgsql/share/extension/hypopg.control

-#########################################################################################
-#
-# Layer "online_advisor-build"
-# compile online_advisor extension
-#
-#########################################################################################
-FROM build-deps AS online_advisor-src
-ARG PG_VERSION
-
-# online_advisor supports all Postgres version starting from PG14, but prior to PG17 has to be included in preload_shared_libraries
-# last release 1.0 - May 15, 2025
-WORKDIR /ext-src
-RUN case "${PG_VERSION:?}" in \
-    "v17") \
-        ;; \
-    *) \
-        echo "skipping the version of online_advistor for $PG_VERSION" && exit 0 \
-        ;; \
-    esac && \
-	wget https://github.com/knizhnik/online_advisor/archive/refs/tags/1.0.tar.gz -O online_advisor.tar.gz && \
-    echo "059b7d9e5a90013a58bdd22e9505b88406ce05790675eb2d8434e5b215652d54 online_advisor.tar.gz" | sha256sum --check && \
-    mkdir online_advisor-src && cd online_advisor-src && tar xzf ../online_advisor.tar.gz --strip-components=1 -C .
-
-FROM pg-build AS online_advisor-build
-COPY --from=online_advisor-src /ext-src/ /ext-src/
-WORKDIR /ext-src/
-RUN if [ -d online_advisor-src ]; then \
-	    cd online_advisor-src && \
-        make -j install && \
-        echo 'trusted = true' >> /usr/local/pgsql/share/extension/online_advisor.control; \
-    fi
-
 #########################################################################################
 #
 # Layer "pg_hashids-build"
@@ -1680,7 +1648,6 @@ COPY --from=pg_jsonschema-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg_graphql-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg_tiktoken-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=hypopg-build /usr/local/pgsql/ /usr/local/pgsql/
-COPY --from=online_advisor-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg_hashids-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=rum-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pgtap-build /usr/local/pgsql/ /usr/local/pgsql/
@@ -1856,7 +1823,6 @@ COPY --from=pgjwt-src /ext-src/ /ext-src/
 COPY --from=pg_graphql-src /ext-src/ /ext-src/
 #COPY --from=pg_tiktoken-src /ext-src/ /ext-src/
 COPY --from=hypopg-src /ext-src/ /ext-src/
-COPY --from=online_advisor-src /ext-src/ /ext-src/
 COPY --from=pg_hashids-src /ext-src/ /ext-src/
 COPY --from=rum-src /ext-src/ /ext-src/
 COPY --from=pgtap-src /ext-src/ /ext-src/
@@ -1882,7 +1848,6 @@ COPY --from=pg_partman-src /ext-src/ /ext-src/
 #COPY --from=pg_mooncake-src /ext-src/ /ext-src/
 COPY --from=pg_repack-src /ext-src/ /ext-src/
 COPY --from=pg_repack-build /usr/local/pgsql/ /usr/local/pgsql/
-COPY --from=pg_duckdb-src /ext-src/ /ext-src/
 COPY compute/patches/pg_repack.patch /ext-src
 RUN cd /ext-src/pg_repack-src && patch -p1 </ext-src/pg_repack.patch && rm -f /ext-src/pg_repack.patch

--- a/compute_tools/src/pg_helpers.rs
+++ b/compute_tools/src/pg_helpers.rs
@@ -213,10 +213,8 @@ impl Escaping for PgIdent {

        // Find the first suitable tag that is not present in the string.
        // Postgres' max role/DB name length is 63 bytes, so even in the
-        // worst case it won't take long. Outer tag is always `tag + "x"`,
-        // so if `tag` is not present in the string, `outer_tag` is not
-        // present in the string either.
-        while self.contains(&tag.to_string()) {
+        // worst case it won't take long.
+        while self.contains(&format!("${tag}$")) || self.contains(&format!("${outer_tag}$")) {
            tag += "x";
            outer_tag = tag.clone() + "x";
        }
--- a/compute_tools/src/rsyslog.rs
+++ b/compute_tools/src/rsyslog.rs
@@ -27,40 +27,6 @@ fn get_rsyslog_pid() -> Option<String> {
    }
 }

-fn wait_for_rsyslog_pid() -> Result<String, anyhow::Error> {
-    const MAX_WAIT: Duration = Duration::from_secs(5);
-    const INITIAL_SLEEP: Duration = Duration::from_millis(2);
-
-    let mut sleep_duration = INITIAL_SLEEP;
-    let start = std::time::Instant::now();
-    let mut attempts = 1;
-
-    for attempt in 1.. {
-        attempts = attempt;
-        match get_rsyslog_pid() {
-            Some(pid) => return Ok(pid),
-            None => {
-                if start.elapsed() >= MAX_WAIT {
-                    break;
-                }
-                info!(
-                    "rsyslogd is not running, attempt {}. Sleeping for {} ms",
-                    attempt,
-                    sleep_duration.as_millis()
-                );
-                std::thread::sleep(sleep_duration);
-                sleep_duration *= 2;
-            }
-        }
-    }
-
-    Err(anyhow::anyhow!(
-        "rsyslogd is not running after waiting for {} seconds and {} attempts",
-        attempts,
-        start.elapsed().as_secs()
-    ))
-}
-
 // Restart rsyslogd to apply the new configuration.
 // This is necessary, because there is no other way to reload the rsyslog configuration.
 //
@@ -70,14 +36,14 @@ fn wait_for_rsyslog_pid() -> Result<String, anyhow::Error> {
 // TODO: test it properly
 //
 fn restart_rsyslog() -> Result<()> {
+    let old_pid = get_rsyslog_pid().context("rsyslogd is not running")?;
+    info!("rsyslogd is running with pid: {}, restart it", old_pid);
+
    // kill it to restart
    let _ = Command::new("pkill")
        .arg("rsyslogd")
        .output()
-        .context("Failed to restart rsyslogd")?;
-
-    // ensure rsyslogd is running
-    wait_for_rsyslog_pid()?;
+        .context("Failed to stop rsyslogd")?;

    Ok(())
 }
@@ -165,11 +131,15 @@ pub fn configure_postgres_logs_export(conf: PostgresLogsRsyslogConfig) -> Result
        return Ok(());
    }

-    // Nothing to configure
+    // When new config is empty we can simply remove the configuration file.
    if new_config.is_empty() {
-        // When the configuration is removed, PostgreSQL will stop sending data
-        // to the files watched by rsyslog, so restarting rsyslog is more effort
-        // than just ignoring this change.
+        info!("removing rsyslog config file: {}", POSTGRES_LOGS_CONF_PATH);
+        match std::fs::remove_file(POSTGRES_LOGS_CONF_PATH) {
+            Ok(_) => {}
+            Err(err) if err.kind() == ErrorKind::NotFound => {}
+            Err(err) => return Err(err.into()),
+        }
+        restart_rsyslog()?;
        return Ok(());
    }

--- a/compute_tools/tests/pg_helpers_tests.rs
+++ b/compute_tools/tests/pg_helpers_tests.rs
@@ -71,14 +71,6 @@ test.escaping = 'here''s a backslash \\ and a quote '' and a double-quote " hoor
            ("name$$$", ("$x$name$$$$x$", "xx")),
            ("name$$$$", ("$x$name$$$$$x$", "xx")),
            ("name$x$", ("$xx$name$x$$xx$", "xxx")),
-            ("x", ("$xx$x$xx$", "xxx")),
-            ("xx", ("$xxx$xx$xxx$", "xxxx")),
-            ("$x", ("$xx$$x$xx$", "xxx")),
-            ("x$", ("$xx$x$$xx$", "xxx")),
-            ("$x$", ("$xx$$x$$xx$", "xxx")),
-            ("xx$", ("$xxx$xx$$xxx$", "xxxx")),
-            ("$xx", ("$xxx$$xx$xxx$", "xxxx")),
-            ("$xx$", ("$xxx$$xx$$xxx$", "xxxx")),
        ];

        for (input, expected) in test_cases {
--- a/control_plane/src/pageserver.rs
+++ b/control_plane/src/pageserver.rs
@@ -546,16 +546,6 @@ impl PageServerNode {
                .map(serde_json::from_str)
                .transpose()
                .context("Falied to parse 'sampling_ratio'")?,
-            relsize_snapshot_cache_capacity: settings
-                .remove("relsize snapshot cache capacity")
-                .map(|x| x.parse::<usize>())
-                .transpose()
-                .context("Falied to parse 'relsize_snapshot_cache_capacity' as integer")?,
-            basebackup_cache_enabled: settings
-                .remove("basebackup_cache_enabled")
-                .map(|x| x.parse::<bool>())
-                .transpose()
-                .context("Failed to parse 'basebackup_cache_enabled' as bool")?,
        };
        if !settings.is_empty() {
            bail!("Unrecognized tenant settings: {settings:?}")
--- a/endpoint_storage/src/app.rs
+++ b/endpoint_storage/src/app.rs
@@ -462,8 +462,6 @@ MC4CAQAwBQYDK2VwBCIEID/Drmc1AA6U/znNRWpF3zEGegOATQxfkdWxitcOMsIH
        if var(REAL_S3_ENV).is_ok() {
            assert!(body.contains("remote_storage_s3_deleted_objects_total"));
        }
-
-        #[cfg(target_os = "linux")]
        assert!(body.contains("process_threads"));
    }

--- a/libs/pageserver_api/src/config.rs
+++ b/libs/pageserver_api/src/config.rs
@@ -183,8 +183,6 @@ pub struct ConfigToml {
    pub enable_tls_page_service_api: bool,
    pub dev_mode: bool,
    pub timeline_import_config: TimelineImportConfig,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub basebackup_cache_config: Option<BasebackupCacheConfig>,
 }

 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
@@ -237,7 +235,7 @@ pub enum PageServiceProtocolPipelinedBatchingStrategy {
    ScatteredLsn,
 }

-#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 #[serde(tag = "mode", rename_all = "kebab-case")]
 pub enum GetVectoredConcurrentIo {
    /// The read path is fully sequential: layers are visited
@@ -310,26 +308,6 @@ pub struct TimelineImportConfig {
    pub import_job_checkpoint_threshold: NonZeroUsize,
 }

-#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
-#[serde(default)]
-pub struct BasebackupCacheConfig {
-    #[serde(with = "humantime_serde")]
-    pub cleanup_period: Duration,
-    // FIXME: Support max_size_bytes.
-    // pub max_size_bytes: usize,
-    pub max_size_entries: i64,
-}
-
-impl Default for BasebackupCacheConfig {
-    fn default() -> Self {
-        Self {
-            cleanup_period: Duration::from_secs(60),
-            // max_size_bytes: 1024 * 1024 * 1024, // 1 GiB
-            max_size_entries: 1000,
-        }
-    }
-}
-
 pub mod statvfs {
    pub mod mock {
        #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
@@ -513,14 +491,6 @@ pub struct TenantConfigToml {
    /// Tenant level performance sampling ratio override. Controls the ratio of get page requests
    /// that will get perf sampling for the tenant.
    pub sampling_ratio: Option<Ratio>,
-
-    /// Capacity of relsize snapshot cache (used by replicas).
-    pub relsize_snapshot_cache_capacity: usize,
-
-    /// Enable preparing basebackup on XLOG_CHECKPOINT_SHUTDOWN and using it in basebackup requests.
-    // FIXME: Remove skip_serializing_if when the feature is stable.
-    #[serde(skip_serializing_if = "std::ops::Not::not")]
-    pub basebackup_cache_enabled: bool,
 }

 pub mod defaults {
@@ -694,7 +664,6 @@ impl Default for ConfigToml {
                import_job_soft_size_limit: NonZeroUsize::new(1024 * 1024 * 1024).unwrap(),
                import_job_checkpoint_threshold: NonZeroUsize::new(128).unwrap(),
            },
-            basebackup_cache_config: None,
        }
    }
 }
@@ -761,7 +730,6 @@ pub mod tenant_conf_defaults {
    pub const DEFAULT_GC_COMPACTION_VERIFICATION: bool = true;
    pub const DEFAULT_GC_COMPACTION_INITIAL_THRESHOLD_KB: u64 = 5 * 1024 * 1024; // 5GB
    pub const DEFAULT_GC_COMPACTION_RATIO_PERCENT: u64 = 100;
-    pub const DEFAULT_RELSIZE_SNAPSHOT_CACHE_CAPACITY: usize = 1000;
 }

 impl Default for TenantConfigToml {
@@ -819,8 +787,6 @@ impl Default for TenantConfigToml {
            gc_compaction_initial_threshold_kb: DEFAULT_GC_COMPACTION_INITIAL_THRESHOLD_KB,
            gc_compaction_ratio_percent: DEFAULT_GC_COMPACTION_RATIO_PERCENT,
            sampling_ratio: None,
-            relsize_snapshot_cache_capacity: DEFAULT_RELSIZE_SNAPSHOT_CACHE_CAPACITY,
-            basebackup_cache_enabled: false,
        }
    }
 }
--- a/libs/pageserver_api/src/models.rs
+++ b/libs/pageserver_api/src/models.rs
@@ -630,10 +630,6 @@ pub struct TenantConfigPatch {
    pub gc_compaction_ratio_percent: FieldPatch<u64>,
    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
    pub sampling_ratio: FieldPatch<Option<Ratio>>,
-    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
-    pub relsize_snapshot_cache_capacity: FieldPatch<usize>,
-    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
-    pub basebackup_cache_enabled: FieldPatch<bool>,
 }

 /// Like [`crate::config::TenantConfigToml`], but preserves the information
@@ -763,12 +759,6 @@ pub struct TenantConfig {

    #[serde(skip_serializing_if = "Option::is_none")]
    pub sampling_ratio: Option<Option<Ratio>>,
-
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub relsize_snapshot_cache_capacity: Option<usize>,
-
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub basebackup_cache_enabled: Option<bool>,
 }

 impl TenantConfig {
@@ -814,8 +804,6 @@ impl TenantConfig {
            mut gc_compaction_initial_threshold_kb,
            mut gc_compaction_ratio_percent,
            mut sampling_ratio,
-            mut relsize_snapshot_cache_capacity,
-            mut basebackup_cache_enabled,
        } = self;

        patch.checkpoint_distance.apply(&mut checkpoint_distance);
@@ -917,12 +905,6 @@ impl TenantConfig {
            .gc_compaction_ratio_percent
            .apply(&mut gc_compaction_ratio_percent);
        patch.sampling_ratio.apply(&mut sampling_ratio);
-        patch
-            .relsize_snapshot_cache_capacity
-            .apply(&mut relsize_snapshot_cache_capacity);
-        patch
-            .basebackup_cache_enabled
-            .apply(&mut basebackup_cache_enabled);

        Ok(Self {
            checkpoint_distance,
@@ -962,8 +944,6 @@ impl TenantConfig {
            gc_compaction_initial_threshold_kb,
            gc_compaction_ratio_percent,
            sampling_ratio,
-            relsize_snapshot_cache_capacity,
-            basebackup_cache_enabled,
        })
    }

@@ -1072,12 +1052,6 @@ impl TenantConfig {
                .gc_compaction_ratio_percent
                .unwrap_or(global_conf.gc_compaction_ratio_percent),
            sampling_ratio: self.sampling_ratio.unwrap_or(global_conf.sampling_ratio),
-            relsize_snapshot_cache_capacity: self
-                .relsize_snapshot_cache_capacity
-                .unwrap_or(global_conf.relsize_snapshot_cache_capacity),
-            basebackup_cache_enabled: self
-                .basebackup_cache_enabled
-                .unwrap_or(global_conf.basebackup_cache_enabled),
        }
    }
 }
--- a/libs/proxy/tokio-postgres2/src/error/mod.rs
+++ b/libs/proxy/tokio-postgres2/src/error/mod.rs
@@ -86,27 +86,6 @@ pub struct DbError {
 }

 impl DbError {
-    pub fn new_test_error(code: SqlState, message: String) -> Self {
-        DbError {
-            severity: "ERROR".to_string(),
-            parsed_severity: Some(Severity::Error),
-            code,
-            message,
-            detail: None,
-            hint: None,
-            position: None,
-            where_: None,
-            schema: None,
-            table: None,
-            column: None,
-            datatype: None,
-            constraint: None,
-            file: None,
-            line: None,
-            routine: None,
-        }
-    }
-
    pub(crate) fn parse(fields: &mut ErrorFields<'_>) -> io::Result<DbError> {
        let mut severity = None;
        let mut parsed_severity = None;
--- a/pageserver/Cargo.toml
+++ b/pageserver/Cargo.toml
@@ -30,7 +30,6 @@ crc32c.workspace = true
 either.workspace = true
 fail.workspace = true
 futures.workspace = true
-hashlink.workspace = true
 hex.workspace = true
 humantime.workspace = true
 humantime-serde.workspace = true
--- a/pageserver/page_api/build.rs
+++ b/pageserver/page_api/build.rs
@@ -1,13 +1,7 @@
-use std::env;
-use std::path::PathBuf;
-
-/// Generates Rust code from .proto Protobuf schemas, along with a binary file
-/// descriptor set for Protobuf schema reflection.
 fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let out_dir = PathBuf::from(env::var("OUT_DIR")?);
+    // Generates Rust code from .proto Protobuf schemas.
    tonic_build::configure()
        .bytes(["."])
-        .file_descriptor_set_path(out_dir.join("page_api_descriptor.bin"))
        .compile_protos(&["proto/page_service.proto"], &["proto"])
        .map_err(|err| err.into())
 }
--- a/pageserver/page_api/proto/page_service.proto
+++ b/pageserver/page_api/proto/page_service.proto
@@ -11,19 +11,6 @@
 // - neon-shard-id: shard ID, as <number><count> in hex ("0b10" = shard 11 of 16, 0-based)
 // - neon-timeline-id: timeline ID ("f08c4e9a2d5f76b1e3a7c2d8910f4b3e")
 //
-// The service can be accessed via e.g. grpcurl:
-//
-//    ```
-//    grpcurl \
-//      -plaintext \
-//      -H "neon-tenant-id: 7c4a1f9e3bd6470c8f3e21a65bd2e980" \
-//      -H "neon-shard-id: 0b10" \
-//      -H "neon-timeline-id: f08c4e9a2d5f76b1e3a7c2d8910f4b3e" \
-//      -H "authorization: Bearer $JWT" \
-//      -d '{"read_lsn": {"request_lsn": 1234567890}, "rel": {"spc_oid": 1663, "db_oid": 1234, "rel_number": 5678, "fork_number": 0}}'
-//      localhost:51051 page_api.PageService/CheckRelExists
-//    ```
-//
 // TODO: consider adding neon-compute-mode ("primary", "static", "replica").
 // However, this will require reconnecting when changing modes.
 //
@@ -33,7 +20,7 @@
 // - Compression

 syntax = "proto3";
-package page_api;
+package page_service;

 service PageService {
  // Returns whether a relation exists.
--- a/pageserver/page_api/src/lib.rs
+++ b/pageserver/page_api/src/lib.rs
@@ -7,12 +7,7 @@

 // Code generated by protobuf.
 pub mod proto {
-    tonic::include_proto!("page_api");
-
-    /// File descriptor set for Protobuf schema reflection. This allows using
-    /// e.g. grpcurl with the API.
-    pub const FILE_DESCRIPTOR_SET: &[u8] =
-        tonic::include_file_descriptor_set!("page_api_descriptor");
+    tonic::include_proto!("page_service");

    pub use page_service_client::PageServiceClient;
    pub use page_service_server::{PageService, PageServiceServer};
--- a/pageserver/src/basebackup.rs
+++ b/pageserver/src/basebackup.rs
@@ -144,7 +144,7 @@ where
        replica,
        ctx,
        io_concurrency: IoConcurrency::spawn_from_conf(
-            timeline.conf.get_vectored_concurrent_io,
+            timeline.conf,
            timeline
                .gate
                .enter()
@@ -343,7 +343,7 @@ where
            // Gather non-relational files from object storage pages.
            let slru_partitions = self
                .timeline
-                .get_slru_keyspace(Version::at(self.lsn), self.ctx)
+                .get_slru_keyspace(Version::Lsn(self.lsn), self.ctx)
                .await?
                .partition(
                    self.timeline.get_shard_identity(),
@@ -378,7 +378,7 @@ where
            // Otherwise only include init forks of unlogged relations.
            let rels = self
                .timeline
-                .list_rels(spcnode, dbnode, Version::at(self.lsn), self.ctx)
+                .list_rels(spcnode, dbnode, Version::Lsn(self.lsn), self.ctx)
                .await?;
            for &rel in rels.iter() {
                // Send init fork as main fork to provide well formed empty
@@ -517,7 +517,7 @@ where
    async fn add_rel(&mut self, src: RelTag, dst: RelTag) -> Result<(), BasebackupError> {
        let nblocks = self
            .timeline
-            .get_rel_size(src, Version::at(self.lsn), self.ctx)
+            .get_rel_size(src, Version::Lsn(self.lsn), self.ctx)
            .await?;

        // If the relation is empty, create an empty file
@@ -577,7 +577,7 @@ where
        let relmap_img = if has_relmap_file {
            let img = self
                .timeline
-                .get_relmap_file(spcnode, dbnode, Version::at(self.lsn), self.ctx)
+                .get_relmap_file(spcnode, dbnode, Version::Lsn(self.lsn), self.ctx)
                .await?;

            if img.len()
@@ -631,7 +631,7 @@ where
            if !has_relmap_file
                && self
                    .timeline
-                    .list_rels(spcnode, dbnode, Version::at(self.lsn), self.ctx)
+                    .list_rels(spcnode, dbnode, Version::Lsn(self.lsn), self.ctx)
                    .await?
                    .is_empty()
            {
--- a/pageserver/src/basebackup_cache.rs
+++ b/pageserver/src/basebackup_cache.rs
@@ -1,518 +0,0 @@
-use std::{collections::HashMap, sync::Arc};
-
-use async_compression::tokio::write::GzipEncoder;
-use camino::{Utf8Path, Utf8PathBuf};
-use metrics::core::{AtomicU64, GenericCounter};
-use pageserver_api::{config::BasebackupCacheConfig, models::TenantState};
-use tokio::{
-    io::{AsyncWriteExt, BufWriter},
-    sync::mpsc::{UnboundedReceiver, UnboundedSender},
-};
-use tokio_util::sync::CancellationToken;
-use utils::{
-    id::{TenantId, TenantTimelineId, TimelineId},
-    lsn::Lsn,
-    shard::TenantShardId,
-};
-
-use crate::{
-    basebackup::send_basebackup_tarball,
-    context::{DownloadBehavior, RequestContext},
-    metrics::{BASEBACKUP_CACHE_ENTRIES, BASEBACKUP_CACHE_PREPARE, BASEBACKUP_CACHE_READ},
-    task_mgr::TaskKind,
-    tenant::{
-        Timeline,
-        mgr::{TenantManager, TenantSlot},
-    },
-};
-
-pub struct BasebackupPrepareRequest {
-    pub tenant_shard_id: TenantShardId,
-    pub timeline_id: TimelineId,
-    pub lsn: Lsn,
-}
-
-pub type BasebackupPrepareSender = UnboundedSender<BasebackupPrepareRequest>;
-pub type BasebackupPrepareReceiver = UnboundedReceiver<BasebackupPrepareRequest>;
-
-type BasebackupRemoveEntrySender = UnboundedSender<Utf8PathBuf>;
-type BasebackupRemoveEntryReceiver = UnboundedReceiver<Utf8PathBuf>;
-
-/// BasebackupCache stores cached basebackup archives for timelines on local disk.
-///
-/// The main purpose of this cache is to speed up the startup process of compute nodes
-/// after scaling to zero.
-/// Thus, the basebackup is stored only for the latest LSN of the timeline and with
-/// fixed set of parameters (gzip=true, full_backup=false, replica=false, prev_lsn=none).
-///
-/// The cache receives prepare requests through the `BasebackupPrepareSender` channel,
-/// generates a basebackup from the timeline in the background, and stores it on disk.
-///
-/// Basebackup requests are pretty rare. We expect ~thousands of entries in the cache
-/// and ~1 RPS for get requests.
-pub struct BasebackupCache {
-    data_dir: Utf8PathBuf,
-    config: BasebackupCacheConfig,
-    tenant_manager: Arc<TenantManager>,
-    remove_entry_sender: BasebackupRemoveEntrySender,
-
-    entries: std::sync::Mutex<HashMap<TenantTimelineId, Lsn>>,
-
-    cancel: CancellationToken,
-
-    read_hit_count: GenericCounter<AtomicU64>,
-    read_miss_count: GenericCounter<AtomicU64>,
-    read_err_count: GenericCounter<AtomicU64>,
-
-    prepare_ok_count: GenericCounter<AtomicU64>,
-    prepare_skip_count: GenericCounter<AtomicU64>,
-    prepare_err_count: GenericCounter<AtomicU64>,
-}
-
-impl BasebackupCache {
-    /// Creates a BasebackupCache and spawns the background task.
-    /// The initialization of the cache is performed in the background and does not
-    /// block the caller. The cache will return `None` for any get requests until
-    /// initialization is complete.
-    pub fn spawn(
-        runtime_handle: &tokio::runtime::Handle,
-        data_dir: Utf8PathBuf,
-        config: Option<BasebackupCacheConfig>,
-        prepare_receiver: BasebackupPrepareReceiver,
-        tenant_manager: Arc<TenantManager>,
-        cancel: CancellationToken,
-    ) -> Arc<Self> {
-        let (remove_entry_sender, remove_entry_receiver) = tokio::sync::mpsc::unbounded_channel();
-
-        let enabled = config.is_some();
-
-        let cache = Arc::new(BasebackupCache {
-            data_dir,
-            config: config.unwrap_or_default(),
-            tenant_manager,
-            remove_entry_sender,
-
-            entries: std::sync::Mutex::new(HashMap::new()),
-
-            cancel,
-
-            read_hit_count: BASEBACKUP_CACHE_READ.with_label_values(&["hit"]),
-            read_miss_count: BASEBACKUP_CACHE_READ.with_label_values(&["miss"]),
-            read_err_count: BASEBACKUP_CACHE_READ.with_label_values(&["error"]),
-
-            prepare_ok_count: BASEBACKUP_CACHE_PREPARE.with_label_values(&["ok"]),
-            prepare_skip_count: BASEBACKUP_CACHE_PREPARE.with_label_values(&["skip"]),
-            prepare_err_count: BASEBACKUP_CACHE_PREPARE.with_label_values(&["error"]),
-        });
-
-        if enabled {
-            runtime_handle.spawn(
-                cache
-                    .clone()
-                    .background(prepare_receiver, remove_entry_receiver),
-            );
-        }
-
-        cache
-    }
-
-    /// Gets a basebackup entry from the cache.
-    /// If the entry is found, opens a file with the basebackup archive and returns it.
-    /// The open file descriptor will prevent the file system from deleting the file
-    /// even if the entry is removed from the cache in the background.
-    pub async fn get(
-        &self,
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-        lsn: Lsn,
-    ) -> Option<tokio::fs::File> {
-        // Fast path. Check if the entry exists using the in-memory state.
-        let tti = TenantTimelineId::new(tenant_id, timeline_id);
-        if self.entries.lock().unwrap().get(&tti) != Some(&lsn) {
-            self.read_miss_count.inc();
-            return None;
-        }
-
-        let path = self.entry_path(tenant_id, timeline_id, lsn);
-
-        match tokio::fs::File::open(path).await {
-            Ok(file) => {
-                self.read_hit_count.inc();
-                Some(file)
-            }
-            Err(e) => {
-                if e.kind() == std::io::ErrorKind::NotFound {
-                    // We may end up here if the basebackup was concurrently removed by the cleanup task.
-                    self.read_miss_count.inc();
-                } else {
-                    self.read_err_count.inc();
-                    tracing::warn!("Unexpected error opening basebackup cache file: {:?}", e);
-                }
-                None
-            }
-        }
-    }
-
-    // Private methods.
-
-    fn entry_filename(tenant_id: TenantId, timeline_id: TimelineId, lsn: Lsn) -> String {
-        // The default format for LSN is 0/ABCDEF.
-        // The backslash is not filename friendly, so serialize it as plain hex.
-        let lsn = lsn.0;
-        format!("basebackup_{tenant_id}_{timeline_id}_{lsn:016X}.tar.gz")
-    }
-
-    fn entry_path(&self, tenant_id: TenantId, timeline_id: TimelineId, lsn: Lsn) -> Utf8PathBuf {
-        self.data_dir
-            .join(Self::entry_filename(tenant_id, timeline_id, lsn))
-    }
-
-    fn entry_tmp_path(
-        &self,
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-        lsn: Lsn,
-    ) -> Utf8PathBuf {
-        self.data_dir
-            .join("tmp")
-            .join(Self::entry_filename(tenant_id, timeline_id, lsn))
-    }
-
-    fn parse_entry_filename(filename: &str) -> Option<(TenantId, TimelineId, Lsn)> {
-        let parts: Vec<&str> = filename
-            .strip_prefix("basebackup_")?
-            .strip_suffix(".tar.gz")?
-            .split('_')
-            .collect();
-        if parts.len() != 3 {
-            return None;
-        }
-        let tenant_id = parts[0].parse::<TenantId>().ok()?;
-        let timeline_id = parts[1].parse::<TimelineId>().ok()?;
-        let lsn = Lsn(u64::from_str_radix(parts[2], 16).ok()?);
-
-        Some((tenant_id, timeline_id, lsn))
-    }
-
-    async fn cleanup(&self) -> anyhow::Result<()> {
-        // Cleanup tmp directory.
-        let tmp_dir = self.data_dir.join("tmp");
-        let mut tmp_dir = tokio::fs::read_dir(&tmp_dir).await?;
-        while let Some(dir_entry) = tmp_dir.next_entry().await? {
-            if let Err(e) = tokio::fs::remove_file(dir_entry.path()).await {
-                tracing::warn!("Failed to remove basebackup cache tmp file: {:#}", e);
-            }
-        }
-
-        // Remove outdated entries.
-        let entries_old = self.entries.lock().unwrap().clone();
-        let mut entries_new = HashMap::new();
-        for (tenant_shard_id, tenant_slot) in self.tenant_manager.list() {
-            if !tenant_shard_id.is_shard_zero() {
-                continue;
-            }
-            let TenantSlot::Attached(tenant) = tenant_slot else {
-                continue;
-            };
-            let tenant_id = tenant_shard_id.tenant_id;
-
-            for timeline in tenant.list_timelines() {
-                let tti = TenantTimelineId::new(tenant_id, timeline.timeline_id);
-                if let Some(&entry_lsn) = entries_old.get(&tti) {
-                    if timeline.get_last_record_lsn() <= entry_lsn {
-                        entries_new.insert(tti, entry_lsn);
-                    }
-                }
-            }
-        }
-
-        for (&tti, &lsn) in entries_old.iter() {
-            if !entries_new.contains_key(&tti) {
-                self.remove_entry_sender
-                    .send(self.entry_path(tti.tenant_id, tti.timeline_id, lsn))
-                    .unwrap();
-            }
-        }
-
-        BASEBACKUP_CACHE_ENTRIES.set(entries_new.len() as i64);
-        *self.entries.lock().unwrap() = entries_new;
-
-        Ok(())
-    }
-
-    async fn on_startup(&self) -> anyhow::Result<()> {
-        // Create data_dir and tmp directory if they do not exist.
-        tokio::fs::create_dir_all(&self.data_dir.join("tmp"))
-            .await
-            .map_err(|e| {
-                anyhow::anyhow!(
-                    "Failed to create basebackup cache data_dir {:?}: {:?}",
-                    self.data_dir,
-                    e
-                )
-            })?;
-
-        // Read existing entries from the data_dir and add them to in-memory state.
-        let mut entries = HashMap::new();
-        let mut dir = tokio::fs::read_dir(&self.data_dir).await?;
-        while let Some(dir_entry) = dir.next_entry().await? {
-            let filename = dir_entry.file_name();
-
-            if filename == "tmp" {
-                // Skip the tmp directory.
-                continue;
-            }
-
-            let parsed = Self::parse_entry_filename(filename.to_string_lossy().as_ref());
-            let Some((tenant_id, timeline_id, lsn)) = parsed else {
-                tracing::warn!("Invalid basebackup cache file name: {:?}", filename);
-                continue;
-            };
-
-            let tti = TenantTimelineId::new(tenant_id, timeline_id);
-
-            use std::collections::hash_map::Entry::*;
-
-            match entries.entry(tti) {
-                Occupied(mut entry) => {
-                    let entry_lsn = *entry.get();
-                    // Leave only the latest entry, remove the old one.
-                    if lsn < entry_lsn {
-                        self.remove_entry_sender.send(self.entry_path(
-                            tenant_id,
-                            timeline_id,
-                            lsn,
-                        ))?;
-                    } else if lsn > entry_lsn {
-                        self.remove_entry_sender.send(self.entry_path(
-                            tenant_id,
-                            timeline_id,
-                            entry_lsn,
-                        ))?;
-                        entry.insert(lsn);
-                    } else {
-                        // Two different filenames parsed to the same timline_id and LSN.
-                        // Should never happen.
-                        return Err(anyhow::anyhow!(
-                            "Duplicate basebackup cache entry with the same LSN: {:?}",
-                            filename
-                        ));
-                    }
-                }
-                Vacant(entry) => {
-                    entry.insert(lsn);
-                }
-            }
-        }
-
-        BASEBACKUP_CACHE_ENTRIES.set(entries.len() as i64);
-        *self.entries.lock().unwrap() = entries;
-
-        Ok(())
-    }
-
-    async fn background(
-        self: Arc<Self>,
-        mut prepare_receiver: BasebackupPrepareReceiver,
-        mut remove_entry_receiver: BasebackupRemoveEntryReceiver,
-    ) {
-        // Panic in the background is a safe fallback.
-        // It will drop receivers and the cache will be effectively disabled.
-        self.on_startup()
-            .await
-            .expect("Failed to initialize basebackup cache");
-
-        let mut cleanup_ticker = tokio::time::interval(self.config.cleanup_period);
-        cleanup_ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
-
-        loop {
-            tokio::select! {
-                Some(req) = prepare_receiver.recv() => {
-                    if let Err(err) = self.prepare_basebackup(
-                        req.tenant_shard_id,
-                        req.timeline_id,
-                        req.lsn,
-                    ).await {
-                        tracing::info!("Failed to prepare basebackup: {:#}", err);
-                        self.prepare_err_count.inc();
-                        continue;
-                    }
-                }
-                Some(req) = remove_entry_receiver.recv() => {
-                    if let Err(e) = tokio::fs::remove_file(req).await {
-                        tracing::warn!("Failed to remove basebackup cache file: {:#}", e);
-                    }
-                }
-                _ = cleanup_ticker.tick() => {
-                    self.cleanup().await.unwrap_or_else(|e| {
-                        tracing::warn!("Failed to clean up basebackup cache: {:#}", e);
-                    });
-                }
-                _ = self.cancel.cancelled() => {
-                    tracing::info!("BasebackupCache background task cancelled");
-                    break;
-                }
-            }
-        }
-    }
-
-    /// Prepare a basebackup for the given timeline.
-    ///
-    /// If the basebackup already exists with a higher LSN or the timeline already
-    /// has a higher last_record_lsn, skip the preparation.
-    ///
-    /// The basebackup is prepared in a temporary directory and then moved to the final
-    /// location to make the operation atomic.
-    async fn prepare_basebackup(
-        &self,
-        tenant_shard_id: TenantShardId,
-        timeline_id: TimelineId,
-        req_lsn: Lsn,
-    ) -> anyhow::Result<()> {
-        tracing::info!(
-            tenant_id = %tenant_shard_id.tenant_id,
-            %timeline_id,
-            %req_lsn,
-            "Preparing basebackup for timeline",
-        );
-
-        let tti = TenantTimelineId::new(tenant_shard_id.tenant_id, timeline_id);
-
-        {
-            let entries = self.entries.lock().unwrap();
-            if let Some(&entry_lsn) = entries.get(&tti) {
-                if entry_lsn >= req_lsn {
-                    tracing::info!(
-                        %timeline_id,
-                        %req_lsn,
-                        %entry_lsn,
-                        "Basebackup entry already exists for timeline with higher LSN, skipping basebackup",
-                    );
-                    self.prepare_skip_count.inc();
-                    return Ok(());
-                }
-            }
-
-            if entries.len() as i64 >= self.config.max_size_entries {
-                tracing::info!(
-                    %timeline_id,
-                    %req_lsn,
-                    "Basebackup cache is full, skipping basebackup",
-                );
-                self.prepare_skip_count.inc();
-                return Ok(());
-            }
-        }
-
-        let tenant = self
-            .tenant_manager
-            .get_attached_tenant_shard(tenant_shard_id)?;
-
-        let tenant_state = tenant.current_state();
-        if tenant_state != TenantState::Active {
-            anyhow::bail!(
-                "Tenant {} is not active, current state: {:?}",
-                tenant_shard_id.tenant_id,
-                tenant_state
-            )
-        }
-
-        let timeline = tenant.get_timeline(timeline_id, true)?;
-
-        let last_record_lsn = timeline.get_last_record_lsn();
-        if last_record_lsn > req_lsn {
-            tracing::info!(
-                %timeline_id,
-                %req_lsn,
-                %last_record_lsn,
-                "Timeline has a higher LSN than the requested one, skipping basebackup",
-            );
-            self.prepare_skip_count.inc();
-            return Ok(());
-        }
-
-        let entry_tmp_path = self.entry_tmp_path(tenant_shard_id.tenant_id, timeline_id, req_lsn);
-
-        let res = self
-            .prepare_basebackup_tmp(&entry_tmp_path, &timeline, req_lsn)
-            .await;
-
-        if let Err(err) = res {
-            tracing::info!("Failed to prepare basebackup tmp file: {:#}", err);
-            // Try to clean up tmp file. If we fail, the background clean up task will take care of it.
-            match tokio::fs::remove_file(&entry_tmp_path).await {
-                Ok(_) => {}
-                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
-                Err(e) => {
-                    tracing::info!("Failed to remove basebackup tmp file: {:?}", e);
-                }
-            }
-            return Err(err);
-        }
-
-        // Move the tmp file to the final location atomically.
-        let entry_path = self.entry_path(tenant_shard_id.tenant_id, timeline_id, req_lsn);
-        tokio::fs::rename(&entry_tmp_path, &entry_path).await?;
-
-        let mut entries = self.entries.lock().unwrap();
-        if let Some(old_lsn) = entries.insert(tti, req_lsn) {
-            // Remove the old entry if it exists.
-            self.remove_entry_sender
-                .send(self.entry_path(tenant_shard_id.tenant_id, timeline_id, old_lsn))
-                .unwrap();
-        }
-        BASEBACKUP_CACHE_ENTRIES.set(entries.len() as i64);
-
-        self.prepare_ok_count.inc();
-        Ok(())
-    }
-
-    /// Prepares a basebackup in a temporary file.
-    async fn prepare_basebackup_tmp(
-        &self,
-        emptry_tmp_path: &Utf8Path,
-        timeline: &Arc<Timeline>,
-        req_lsn: Lsn,
-    ) -> anyhow::Result<()> {
-        let ctx = RequestContext::new(TaskKind::BasebackupCache, DownloadBehavior::Download);
-        let ctx = ctx.with_scope_timeline(timeline);
-
-        let file = tokio::fs::File::create(emptry_tmp_path).await?;
-        let mut writer = BufWriter::new(file);
-
-        let mut encoder = GzipEncoder::with_quality(
-            &mut writer,
-            // Level::Best because compression is not on the hot path of basebackup requests.
-            // The decompression is almost not affected by the compression level.
-            async_compression::Level::Best,
-        );
-
-        // We may receive a request before the WAL record is applied to the timeline.
-        // Wait for the requested LSN to be applied.
-        timeline
-            .wait_lsn(
-                req_lsn,
-                crate::tenant::timeline::WaitLsnWaiter::BaseBackupCache,
-                crate::tenant::timeline::WaitLsnTimeout::Default,
-                &ctx,
-            )
-            .await?;
-
-        send_basebackup_tarball(
-            &mut encoder,
-            timeline,
-            Some(req_lsn),
-            None,
-            false,
-            false,
-            &ctx,
-        )
-        .await?;
-
-        encoder.shutdown().await?;
-        writer.flush().await?;
-        writer.into_inner().sync_all().await?;
-
-        Ok(())
-    }
-}
--- a/pageserver/src/bin/pageserver.rs
+++ b/pageserver/src/bin/pageserver.rs
@@ -16,7 +16,6 @@ use http_utils::tls_certs::ReloadingCertificateResolver;
 use metrics::launch_timestamp::{LaunchTimestamp, set_launch_timestamp_metric};
 use metrics::set_build_info_metric;
 use nix::sys::socket::{setsockopt, sockopt};
-use pageserver::basebackup_cache::BasebackupCache;
 use pageserver::config::{PageServerConf, PageserverIdentity, ignored_fields};
 use pageserver::controller_upcall_client::StorageControllerUpcallClient;
 use pageserver::deletion_queue::DeletionQueue;
@@ -542,8 +541,6 @@ fn start_pageserver(
        pageserver::l0_flush::L0FlushGlobalState::new(conf.l0_flush.clone());

    // Scan the local 'tenants/' directory and start loading the tenants
-    let (basebackup_prepare_sender, basebackup_prepare_receiver) =
-        tokio::sync::mpsc::unbounded_channel();
    let deletion_queue_client = deletion_queue.new_client();
    let background_purges = mgr::BackgroundPurges::default();
    let tenant_manager = BACKGROUND_RUNTIME.block_on(mgr::init_tenant_mgr(
@@ -554,22 +551,12 @@ fn start_pageserver(
            remote_storage: remote_storage.clone(),
            deletion_queue_client,
            l0_flush_global_state,
-            basebackup_prepare_sender,
        },
        order,
        shutdown_pageserver.clone(),
    ))?;
    let tenant_manager = Arc::new(tenant_manager);

-    let basebackup_cache = BasebackupCache::spawn(
-        BACKGROUND_RUNTIME.handle(),
-        conf.basebackup_cache_dir(),
-        conf.basebackup_cache_config.clone(),
-        basebackup_prepare_receiver,
-        Arc::clone(&tenant_manager),
-        shutdown_pageserver.child_token(),
-    );
-
    BACKGROUND_RUNTIME.spawn({
        let shutdown_pageserver = shutdown_pageserver.clone();
        let drive_init = async move {
@@ -776,7 +763,6 @@ fn start_pageserver(
        } else {
            None
        },
-        basebackup_cache,
    );

    // All started up! Now just sit and wait for shutdown signal.
--- a/pageserver/src/config.rs
+++ b/pageserver/src/config.rs
@@ -232,8 +232,6 @@ pub struct PageServerConf {
    pub dev_mode: bool,

    pub timeline_import_config: pageserver_api::config::TimelineImportConfig,
-
-    pub basebackup_cache_config: Option<pageserver_api::config::BasebackupCacheConfig>,
 }

 /// Token for authentication to safekeepers
@@ -263,10 +261,6 @@ impl PageServerConf {
        self.workdir.join("metadata.json")
    }

-    pub fn basebackup_cache_dir(&self) -> Utf8PathBuf {
-        self.workdir.join("basebackup_cache")
-    }
-
    pub fn deletion_list_path(&self, sequence: u64) -> Utf8PathBuf {
        // Encode a version in the filename, so that if we ever switch away from JSON we can
        // increment this.
@@ -413,7 +407,6 @@ impl PageServerConf {
            enable_tls_page_service_api,
            dev_mode,
            timeline_import_config,
-            basebackup_cache_config,
        } = config_toml;

        let mut conf = PageServerConf {
@@ -468,7 +461,6 @@ impl PageServerConf {
            enable_tls_page_service_api,
            dev_mode,
            timeline_import_config,
-            basebackup_cache_config,

            // ------------------------------------------------------------
            // fields that require additional validation or custom handling
@@ -552,23 +544,6 @@ impl PageServerConf {
                    ratio.numerator, ratio.denominator
                )
            );
-
-            let url = Url::parse(&tracing_config.export_config.endpoint)
-                .map_err(anyhow::Error::msg)
-                .with_context(|| {
-                    format!(
-                        "tracing endpoint URL is invalid : {}",
-                        tracing_config.export_config.endpoint
-                    )
-                })?;
-
-            ensure!(
-                url.scheme() == "http" || url.scheme() == "https",
-                format!(
-                    "tracing endpoint URL must start with http:// or https://: {}",
-                    tracing_config.export_config.endpoint
-                )
-            );
        }

        IndexEntry::validate_checkpoint_distance(conf.default_tenant_conf.checkpoint_distance)
@@ -685,25 +660,4 @@ mod tests {
        PageServerConf::parse_and_validate(NodeId(0), config_toml, &workdir)
            .expect("parse_and_validate");
    }
-
-    #[test]
-    fn test_config_tracing_endpoint_is_invalid() {
-        let input = r#"
-            control_plane_api = "http://localhost:6666"
-
-            [tracing]
-
-            sampling_ratio = { numerator = 1, denominator = 0 }
-
-            [tracing.export_config]
-            endpoint = "localhost:4317"
-            protocol = "http-binary"
-            timeout = "1ms"
-        "#;
-        let config_toml = toml_edit::de::from_str::<pageserver_api::config::ConfigToml>(input)
-            .expect("config has valid fields");
-        let workdir = Utf8PathBuf::from("/nonexistent");
-        PageServerConf::parse_and_validate(NodeId(0), config_toml, &workdir)
-            .expect_err("parse_and_validate should fail for endpoint without scheme");
-    }
 }
--- a/pageserver/src/consumption_metrics/metrics.rs
+++ b/pageserver/src/consumption_metrics/metrics.rs
@@ -18,25 +18,12 @@ use crate::tenant::timeline::logical_size::CurrentLogicalSize;
 // management.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
 pub(super) enum Name {
-    /// Timeline last_record_lsn, absolute.
+    /// Timeline last_record_lsn, absolute
    #[serde(rename = "written_size")]
    WrittenSize,
    /// Timeline last_record_lsn, incremental
    #[serde(rename = "written_data_bytes_delta")]
    WrittenSizeDelta,
-    /// Written bytes only on this timeline (not including ancestors):
-    /// written_size - ancestor_lsn
-    ///
-    /// On the root branch, this is equivalent to `written_size`.
-    #[serde(rename = "written_size_since_parent")]
-    WrittenSizeSinceParent,
-    /// PITR history size only on this timeline (not including ancestors):
-    /// last_record_lsn - max(pitr_cutoff, ancestor_lsn).
-    ///
-    /// On the root branch, this is its entire PITR history size. Not emitted if GC hasn't computed
-    /// the PITR cutoff yet. 0 if PITR is disabled.
-    #[serde(rename = "pitr_history_size_since_parent")]
-    PitrHistorySizeSinceParent,
    /// Timeline logical size
    #[serde(rename = "timeline_logical_size")]
    LogicalSize,
@@ -170,32 +157,6 @@ impl MetricsKey {
        .incremental_values()
    }

-    /// `written_size` - `ancestor_lsn`.
-    const fn written_size_since_parent(
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-    ) -> AbsoluteValueFactory {
-        MetricsKey {
-            tenant_id,
-            timeline_id: Some(timeline_id),
-            metric: Name::WrittenSizeSinceParent,
-        }
-        .absolute_values()
-    }
-
-    /// `written_size` - max(`pitr_cutoff`, `ancestor_lsn`).
-    const fn pitr_history_size_since_parent(
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-    ) -> AbsoluteValueFactory {
-        MetricsKey {
-            tenant_id,
-            timeline_id: Some(timeline_id),
-            metric: Name::PitrHistorySizeSinceParent,
-        }
-        .absolute_values()
-    }
-
    /// Exact [`Timeline::get_current_logical_size`].
    ///
    /// [`Timeline::get_current_logical_size`]: crate::tenant::Timeline::get_current_logical_size
@@ -373,13 +334,7 @@ impl TenantSnapshot {
 struct TimelineSnapshot {
    loaded_at: (Lsn, SystemTime),
    last_record_lsn: Lsn,
-    ancestor_lsn: Lsn,
    current_exact_logical_size: Option<u64>,
-    /// Whether PITR is enabled (pitr_interval > 0).
-    pitr_enabled: bool,
-    /// The PITR cutoff LSN. None if not yet initialized. If PITR is disabled, this is approximately
-    /// Some(last_record_lsn), but may lag behind it since it's computed periodically.
-    pitr_cutoff: Option<Lsn>,
 }

 impl TimelineSnapshot {
@@ -399,9 +354,6 @@ impl TimelineSnapshot {
        } else {
            let loaded_at = t.loaded_at;
            let last_record_lsn = t.get_last_record_lsn();
-            let ancestor_lsn = t.get_ancestor_lsn();
-            let pitr_enabled = !t.get_pitr_interval().is_zero();
-            let pitr_cutoff = t.gc_info.read().unwrap().cutoffs.time;

            let current_exact_logical_size = {
                let span = tracing::info_span!("collect_metrics_iteration", tenant_id = %t.tenant_shard_id.tenant_id, timeline_id = %t.timeline_id);
@@ -421,10 +373,7 @@ impl TimelineSnapshot {
            Ok(Some(TimelineSnapshot {
                loaded_at,
                last_record_lsn,
-                ancestor_lsn,
                current_exact_logical_size,
-                pitr_enabled,
-                pitr_cutoff,
            }))
        }
    }
@@ -475,8 +424,6 @@ impl TimelineSnapshot {

        let up_to = now;

-        let written_size_last = written_size_now.value.max(prev.1); // don't regress
-
        if let Some(delta) = written_size_now.value.checked_sub(prev.1) {
            let key_value = written_size_delta_key.from_until(prev.0, up_to, delta);
            // written_size_delta
@@ -494,27 +441,6 @@ impl TimelineSnapshot {
            });
        }

-        // Compute the branch-local written size.
-        let written_size_since_parent_key =
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id);
-        metrics.push(
-            written_size_since_parent_key
-                .at(now, written_size_last.saturating_sub(self.ancestor_lsn.0)),
-        );
-
-        // Compute the branch-local PITR history size. Not emitted if GC hasn't yet computed the
-        // PITR cutoff. 0 if PITR is disabled.
-        let pitr_history_size_since_parent_key =
-            MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id);
-        if !self.pitr_enabled {
-            metrics.push(pitr_history_size_since_parent_key.at(now, 0));
-        } else if let Some(pitr_cutoff) = self.pitr_cutoff {
-            metrics.push(pitr_history_size_since_parent_key.at(
-                now,
-                written_size_last.saturating_sub(pitr_cutoff.max(self.ancestor_lsn).0),
-            ));
-        }
-
        {
            let factory = MetricsKey::timeline_logical_size(tenant_id, timeline_id);
            let current_or_previous = self
--- a/pageserver/src/consumption_metrics/metrics/tests.rs
+++ b/pageserver/src/consumption_metrics/metrics/tests.rs
@@ -12,17 +12,12 @@ fn startup_collected_timeline_metrics_before_advancing() {
    let cache = HashMap::new();

    let initdb_lsn = Lsn(0x10000);
-    let pitr_cutoff = Lsn(0x11000);
    let disk_consistent_lsn = Lsn(initdb_lsn.0 * 2);
-    let logical_size = 0x42000;

    let snap = TimelineSnapshot {
        loaded_at: (disk_consistent_lsn, SystemTime::now()),
        last_record_lsn: disk_consistent_lsn,
-        ancestor_lsn: Lsn(0),
-        current_exact_logical_size: Some(logical_size),
-        pitr_enabled: true,
-        pitr_cutoff: Some(pitr_cutoff),
+        current_exact_logical_size: Some(0x42000),
    };

    let now = DateTime::<Utc>::from(SystemTime::now());
@@ -38,11 +33,7 @@ fn startup_collected_timeline_metrics_before_advancing() {
                0
            ),
            MetricsKey::written_size(tenant_id, timeline_id).at(now, disk_consistent_lsn.0),
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id)
-                .at(now, disk_consistent_lsn.0),
-            MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id)
-                .at(now, disk_consistent_lsn.0 - pitr_cutoff.0),
-            MetricsKey::timeline_logical_size(tenant_id, timeline_id).at(now, logical_size)
+            MetricsKey::timeline_logical_size(tenant_id, timeline_id).at(now, 0x42000)
        ]
    );
 }
@@ -58,9 +49,7 @@ fn startup_collected_timeline_metrics_second_round() {
    let before = DateTime::<Utc>::from(before);

    let initdb_lsn = Lsn(0x10000);
-    let pitr_cutoff = Lsn(0x11000);
    let disk_consistent_lsn = Lsn(initdb_lsn.0 * 2);
-    let logical_size = 0x42000;

    let mut metrics = Vec::new();
    let cache = HashMap::from([MetricsKey::written_size(tenant_id, timeline_id)
@@ -70,10 +59,7 @@ fn startup_collected_timeline_metrics_second_round() {
    let snap = TimelineSnapshot {
        loaded_at: (disk_consistent_lsn, init),
        last_record_lsn: disk_consistent_lsn,
-        ancestor_lsn: Lsn(0),
-        current_exact_logical_size: Some(logical_size),
-        pitr_enabled: true,
-        pitr_cutoff: Some(pitr_cutoff),
+        current_exact_logical_size: Some(0x42000),
    };

    snap.to_metrics(tenant_id, timeline_id, now, &mut metrics, &cache);
@@ -83,11 +69,7 @@ fn startup_collected_timeline_metrics_second_round() {
        &[
            MetricsKey::written_size_delta(tenant_id, timeline_id).from_until(before, now, 0),
            MetricsKey::written_size(tenant_id, timeline_id).at(now, disk_consistent_lsn.0),
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id)
-                .at(now, disk_consistent_lsn.0),
-            MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id)
-                .at(now, disk_consistent_lsn.0 - pitr_cutoff.0),
-            MetricsKey::timeline_logical_size(tenant_id, timeline_id).at(now, logical_size)
+            MetricsKey::timeline_logical_size(tenant_id, timeline_id).at(now, 0x42000)
        ]
    );
 }
@@ -104,9 +86,7 @@ fn startup_collected_timeline_metrics_nth_round_at_same_lsn() {
    let before = DateTime::<Utc>::from(before);

    let initdb_lsn = Lsn(0x10000);
-    let pitr_cutoff = Lsn(0x11000);
    let disk_consistent_lsn = Lsn(initdb_lsn.0 * 2);
-    let logical_size = 0x42000;

    let mut metrics = Vec::new();
    let cache = HashMap::from([
@@ -123,10 +103,7 @@ fn startup_collected_timeline_metrics_nth_round_at_same_lsn() {
    let snap = TimelineSnapshot {
        loaded_at: (disk_consistent_lsn, init),
        last_record_lsn: disk_consistent_lsn,
-        ancestor_lsn: Lsn(0),
-        current_exact_logical_size: Some(logical_size),
-        pitr_enabled: true,
-        pitr_cutoff: Some(pitr_cutoff),
+        current_exact_logical_size: Some(0x42000),
    };

    snap.to_metrics(tenant_id, timeline_id, now, &mut metrics, &cache);
@@ -136,18 +113,16 @@ fn startup_collected_timeline_metrics_nth_round_at_same_lsn() {
        &[
            MetricsKey::written_size_delta(tenant_id, timeline_id).from_until(just_before, now, 0),
            MetricsKey::written_size(tenant_id, timeline_id).at(now, disk_consistent_lsn.0),
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id)
-                .at(now, disk_consistent_lsn.0),
-            MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id)
-                .at(now, disk_consistent_lsn.0 - pitr_cutoff.0),
-            MetricsKey::timeline_logical_size(tenant_id, timeline_id).at(now, logical_size)
+            MetricsKey::timeline_logical_size(tenant_id, timeline_id).at(now, 0x42000)
        ]
    );
 }

-/// Tests that written sizes do not regress across restarts.
 #[test]
 fn post_restart_written_sizes_with_rolled_back_last_record_lsn() {
+    // it can happen that we lose the inmemorylayer but have previously sent metrics and we
+    // should never go backwards
+
    let tenant_id = TenantId::generate();
    let timeline_id = TimelineId::generate();

@@ -165,10 +140,7 @@ fn post_restart_written_sizes_with_rolled_back_last_record_lsn() {
    let snap = TimelineSnapshot {
        loaded_at: (Lsn(50), at_restart),
        last_record_lsn: Lsn(50),
-        ancestor_lsn: Lsn(0),
        current_exact_logical_size: None,
-        pitr_enabled: true,
-        pitr_cutoff: Some(Lsn(20)),
    };

    let mut cache = HashMap::from([
@@ -197,8 +169,6 @@ fn post_restart_written_sizes_with_rolled_back_last_record_lsn() {
                0
            ),
            MetricsKey::written_size(tenant_id, timeline_id).at(now, 100),
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id).at(now, 100),
-            MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id).at(now, 80),
        ]
    );

@@ -213,157 +183,6 @@ fn post_restart_written_sizes_with_rolled_back_last_record_lsn() {
        &[
            MetricsKey::written_size_delta(tenant_id, timeline_id).from_until(now, later, 0),
            MetricsKey::written_size(tenant_id, timeline_id).at(later, 100),
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id).at(later, 100),
-            MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id).at(later, 80),
-        ]
-    );
-}
-
-/// Tests that written sizes do not regress across restarts, even on child branches.
-#[test]
-fn post_restart_written_sizes_with_rolled_back_last_record_lsn_and_ancestor_lsn() {
-    let tenant_id = TenantId::generate();
-    let timeline_id = TimelineId::generate();
-
-    let [later, now, at_restart] = time_backwards();
-
-    // FIXME: tests would be so much easier if we did not need to juggle back and forth
-    // SystemTime and DateTime::<Utc> ... Could do the conversion only at upload time?
-    let now = DateTime::<Utc>::from(now);
-    let later = DateTime::<Utc>::from(later);
-    let before_restart = at_restart - std::time::Duration::from_secs(5 * 60);
-    let way_before = before_restart - std::time::Duration::from_secs(10 * 60);
-    let before_restart = DateTime::<Utc>::from(before_restart);
-    let way_before = DateTime::<Utc>::from(way_before);
-
-    let snap = TimelineSnapshot {
-        loaded_at: (Lsn(50), at_restart),
-        last_record_lsn: Lsn(50),
-        ancestor_lsn: Lsn(40),
-        current_exact_logical_size: None,
-        pitr_enabled: true,
-        pitr_cutoff: Some(Lsn(20)),
-    };
-
-    let mut cache = HashMap::from([
-        MetricsKey::written_size(tenant_id, timeline_id)
-            .at(before_restart, 100)
-            .to_kv_pair(),
-        MetricsKey::written_size_delta(tenant_id, timeline_id)
-            .from_until(
-                way_before,
-                before_restart,
-                // not taken into account, but the timestamps are important
-                999_999_999,
-            )
-            .to_kv_pair(),
-    ]);
-
-    let mut metrics = Vec::new();
-    snap.to_metrics(tenant_id, timeline_id, now, &mut metrics, &cache);
-
-    assert_eq!(
-        metrics,
-        &[
-            MetricsKey::written_size_delta(tenant_id, timeline_id).from_until(
-                before_restart,
-                now,
-                0
-            ),
-            MetricsKey::written_size(tenant_id, timeline_id).at(now, 100),
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id).at(now, 60),
-            MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id).at(now, 60),
-        ]
-    );
-
-    // now if we cache these metrics, and re-run while "still in recovery"
-    cache.extend(metrics.drain(..).map(|x| x.to_kv_pair()));
-
-    // "still in recovery", because our snapshot did not change
-    snap.to_metrics(tenant_id, timeline_id, later, &mut metrics, &cache);
-
-    assert_eq!(
-        metrics,
-        &[
-            MetricsKey::written_size_delta(tenant_id, timeline_id).from_until(now, later, 0),
-            MetricsKey::written_size(tenant_id, timeline_id).at(later, 100),
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id).at(later, 60),
-            MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id).at(later, 60),
-        ]
-    );
-}
-
-/// Tests that written sizes do not regress across restarts, even on child branches and
-/// with a PITR cutoff after the branch point.
-#[test]
-fn post_restart_written_sizes_with_rolled_back_last_record_lsn_and_ancestor_lsn_and_pitr_cutoff() {
-    let tenant_id = TenantId::generate();
-    let timeline_id = TimelineId::generate();
-
-    let [later, now, at_restart] = time_backwards();
-
-    // FIXME: tests would be so much easier if we did not need to juggle back and forth
-    // SystemTime and DateTime::<Utc> ... Could do the conversion only at upload time?
-    let now = DateTime::<Utc>::from(now);
-    let later = DateTime::<Utc>::from(later);
-    let before_restart = at_restart - std::time::Duration::from_secs(5 * 60);
-    let way_before = before_restart - std::time::Duration::from_secs(10 * 60);
-    let before_restart = DateTime::<Utc>::from(before_restart);
-    let way_before = DateTime::<Utc>::from(way_before);
-
-    let snap = TimelineSnapshot {
-        loaded_at: (Lsn(50), at_restart),
-        last_record_lsn: Lsn(50),
-        ancestor_lsn: Lsn(30),
-        current_exact_logical_size: None,
-        pitr_enabled: true,
-        pitr_cutoff: Some(Lsn(40)),
-    };
-
-    let mut cache = HashMap::from([
-        MetricsKey::written_size(tenant_id, timeline_id)
-            .at(before_restart, 100)
-            .to_kv_pair(),
-        MetricsKey::written_size_delta(tenant_id, timeline_id)
-            .from_until(
-                way_before,
-                before_restart,
-                // not taken into account, but the timestamps are important
-                999_999_999,
-            )
-            .to_kv_pair(),
-    ]);
-
-    let mut metrics = Vec::new();
-    snap.to_metrics(tenant_id, timeline_id, now, &mut metrics, &cache);
-
-    assert_eq!(
-        metrics,
-        &[
-            MetricsKey::written_size_delta(tenant_id, timeline_id).from_until(
-                before_restart,
-                now,
-                0
-            ),
-            MetricsKey::written_size(tenant_id, timeline_id).at(now, 100),
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id).at(now, 70),
-            MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id).at(now, 60),
-        ]
-    );
-
-    // now if we cache these metrics, and re-run while "still in recovery"
-    cache.extend(metrics.drain(..).map(|x| x.to_kv_pair()));
-
-    // "still in recovery", because our snapshot did not change
-    snap.to_metrics(tenant_id, timeline_id, later, &mut metrics, &cache);
-
-    assert_eq!(
-        metrics,
-        &[
-            MetricsKey::written_size_delta(tenant_id, timeline_id).from_until(now, later, 0),
-            MetricsKey::written_size(tenant_id, timeline_id).at(later, 100),
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id).at(later, 70),
-            MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id).at(later, 60),
        ]
    );
 }
@@ -382,10 +201,7 @@ fn post_restart_current_exact_logical_size_uses_cached() {
    let snap = TimelineSnapshot {
        loaded_at: (Lsn(50), at_restart),
        last_record_lsn: Lsn(50),
-        ancestor_lsn: Lsn(0),
        current_exact_logical_size: None,
-        pitr_enabled: true,
-        pitr_cutoff: None,
    };

    let cache = HashMap::from([MetricsKey::timeline_logical_size(tenant_id, timeline_id)
@@ -470,101 +286,16 @@ fn time_backwards<const N: usize>() -> [std::time::SystemTime; N] {
    times
 }

-/// Tests that disabled PITR history does not yield any history size, even when the PITR cutoff
-/// indicates otherwise.
-#[test]
-fn pitr_disabled_yields_no_history_size() {
-    let tenant_id = TenantId::generate();
-    let timeline_id = TimelineId::generate();
-
-    let mut metrics = Vec::new();
-    let cache = HashMap::new();
-
-    let initdb_lsn = Lsn(0x10000);
-    let pitr_cutoff = Lsn(0x11000);
-    let disk_consistent_lsn = Lsn(initdb_lsn.0 * 2);
-
-    let snap = TimelineSnapshot {
-        loaded_at: (disk_consistent_lsn, SystemTime::now()),
-        last_record_lsn: disk_consistent_lsn,
-        ancestor_lsn: Lsn(0),
-        current_exact_logical_size: None,
-        pitr_enabled: false,
-        pitr_cutoff: Some(pitr_cutoff),
-    };
-
-    let now = DateTime::<Utc>::from(SystemTime::now());
-
-    snap.to_metrics(tenant_id, timeline_id, now, &mut metrics, &cache);
-
-    assert_eq!(
-        metrics,
-        &[
-            MetricsKey::written_size_delta(tenant_id, timeline_id).from_until(
-                snap.loaded_at.1.into(),
-                now,
-                0
-            ),
-            MetricsKey::written_size(tenant_id, timeline_id).at(now, disk_consistent_lsn.0),
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id)
-                .at(now, disk_consistent_lsn.0),
-            MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id).at(now, 0),
-        ]
-    );
-}
-
-/// Tests that uninitialized PITR cutoff does not emit any history size metric at all.
-#[test]
-fn pitr_uninitialized_does_not_emit_history_size() {
-    let tenant_id = TenantId::generate();
-    let timeline_id = TimelineId::generate();
-
-    let mut metrics = Vec::new();
-    let cache = HashMap::new();
-
-    let initdb_lsn = Lsn(0x10000);
-    let disk_consistent_lsn = Lsn(initdb_lsn.0 * 2);
-
-    let snap = TimelineSnapshot {
-        loaded_at: (disk_consistent_lsn, SystemTime::now()),
-        last_record_lsn: disk_consistent_lsn,
-        ancestor_lsn: Lsn(0),
-        current_exact_logical_size: None,
-        pitr_enabled: true,
-        pitr_cutoff: None,
-    };
-
-    let now = DateTime::<Utc>::from(SystemTime::now());
-
-    snap.to_metrics(tenant_id, timeline_id, now, &mut metrics, &cache);
-
-    assert_eq!(
-        metrics,
-        &[
-            MetricsKey::written_size_delta(tenant_id, timeline_id).from_until(
-                snap.loaded_at.1.into(),
-                now,
-                0
-            ),
-            MetricsKey::written_size(tenant_id, timeline_id).at(now, disk_consistent_lsn.0),
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id)
-                .at(now, disk_consistent_lsn.0),
-        ]
-    );
-}
-
 pub(crate) const fn metric_examples_old(
    tenant_id: TenantId,
    timeline_id: TimelineId,
    now: DateTime<Utc>,
    before: DateTime<Utc>,
-) -> [RawMetric; 7] {
+) -> [RawMetric; 5] {
    [
        MetricsKey::written_size(tenant_id, timeline_id).at_old_format(now, 0),
        MetricsKey::written_size_delta(tenant_id, timeline_id)
            .from_until_old_format(before, now, 0),
-        MetricsKey::written_size_since_parent(tenant_id, timeline_id).at_old_format(now, 0),
-        MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id).at_old_format(now, 0),
        MetricsKey::timeline_logical_size(tenant_id, timeline_id).at_old_format(now, 0),
        MetricsKey::remote_storage_size(tenant_id).at_old_format(now, 0),
        MetricsKey::synthetic_size(tenant_id).at_old_format(now, 1),
@@ -576,12 +307,10 @@ pub(crate) const fn metric_examples(
    timeline_id: TimelineId,
    now: DateTime<Utc>,
    before: DateTime<Utc>,
-) -> [NewRawMetric; 7] {
+) -> [NewRawMetric; 5] {
    [
        MetricsKey::written_size(tenant_id, timeline_id).at(now, 0),
        MetricsKey::written_size_delta(tenant_id, timeline_id).from_until(before, now, 0),
-        MetricsKey::written_size_since_parent(tenant_id, timeline_id).at(now, 0),
-        MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id).at(now, 0),
        MetricsKey::timeline_logical_size(tenant_id, timeline_id).at(now, 0),
        MetricsKey::remote_storage_size(tenant_id).at(now, 0),
        MetricsKey::synthetic_size(tenant_id).at(now, 1),
--- a/pageserver/src/consumption_metrics/upload.rs
+++ b/pageserver/src/consumption_metrics/upload.rs
@@ -513,14 +513,6 @@ mod tests {
                line!(),
                r#"{"type":"incremental","start_time":"2023-09-14T00:00:00.123456789Z","stop_time":"2023-09-15T00:00:00.123456789Z","metric":"written_data_bytes_delta","idempotency_key":"2023-09-15 00:00:00.123456789 UTC-1-0000","value":0,"tenant_id":"00000000000000000000000000000000","timeline_id":"ffffffffffffffffffffffffffffffff"}"#,
            ),
-            (
-                line!(),
-                r#"{"type":"absolute","time":"2023-09-15T00:00:00.123456789Z","metric":"written_size_since_parent","idempotency_key":"2023-09-15 00:00:00.123456789 UTC-1-0000","value":0,"tenant_id":"00000000000000000000000000000000","timeline_id":"ffffffffffffffffffffffffffffffff"}"#,
-            ),
-            (
-                line!(),
-                r#"{"type":"absolute","time":"2023-09-15T00:00:00.123456789Z","metric":"pitr_history_size_since_parent","idempotency_key":"2023-09-15 00:00:00.123456789 UTC-1-0000","value":0,"tenant_id":"00000000000000000000000000000000","timeline_id":"ffffffffffffffffffffffffffffffff"}"#,
-            ),
            (
                line!(),
                r#"{"type":"absolute","time":"2023-09-15T00:00:00.123456789Z","metric":"timeline_logical_size","idempotency_key":"2023-09-15 00:00:00.123456789 UTC-1-0000","value":0,"tenant_id":"00000000000000000000000000000000","timeline_id":"ffffffffffffffffffffffffffffffff"}"#,
@@ -568,7 +560,7 @@ mod tests {
        assert_eq!(upgraded_samples, new_samples);
    }

-    fn metric_samples_old() -> [RawMetric; 7] {
+    fn metric_samples_old() -> [RawMetric; 5] {
        let tenant_id = TenantId::from_array([0; 16]);
        let timeline_id = TimelineId::from_array([0xff; 16]);

@@ -580,7 +572,7 @@ mod tests {
        super::super::metrics::metric_examples_old(tenant_id, timeline_id, now, before)
    }

-    fn metric_samples() -> [NewRawMetric; 7] {
+    fn metric_samples() -> [NewRawMetric; 5] {
        let tenant_id = TenantId::from_array([0; 16]);
        let timeline_id = TimelineId::from_array([0xff; 16]);

--- a/pageserver/src/http/routes.rs
+++ b/pageserver/src/http/routes.rs
@@ -449,7 +449,7 @@ async fn build_timeline_info_common(
    // Internally we distinguish between the planned GC cutoff (PITR point) and the "applied" GC cutoff (where we
    // actually trimmed data to), which can pass each other when PITR is changed.
    let min_readable_lsn = std::cmp::max(
-        timeline.get_gc_cutoff_lsn().unwrap_or_default(),
+        timeline.get_gc_cutoff_lsn(),
        *timeline.get_applied_gc_cutoff_lsn(),
    );

@@ -3199,7 +3199,7 @@ async fn list_aux_files(
            .await?;

    let io_concurrency = IoConcurrency::spawn_from_conf(
-        state.conf.get_vectored_concurrent_io,
+        state.conf,
        timeline.gate.enter().map_err(|_| ApiError::Cancelled)?,
    );

--- a/pageserver/src/lib.rs
+++ b/pageserver/src/lib.rs
@@ -3,7 +3,6 @@

 mod auth;
 pub mod basebackup;
-pub mod basebackup_cache;
 pub mod config;
 pub mod consumption_metrics;
 pub mod context;
--- a/pageserver/src/metrics.rs
+++ b/pageserver/src/metrics.rs
@@ -843,50 +843,23 @@ pub(crate) static COMPRESSION_IMAGE_OUTPUT_BYTES: Lazy<IntCounter> = Lazy::new(|
    .expect("failed to define a metric")
 });

-pub(crate) static RELSIZE_LATEST_CACHE_ENTRIES: Lazy<UIntGauge> = Lazy::new(|| {
+pub(crate) static RELSIZE_CACHE_ENTRIES: Lazy<UIntGauge> = Lazy::new(|| {
    register_uint_gauge!(
-        "pageserver_relsize_latest_cache_entries",
-        "Number of entries in the latest relation size cache",
+        "pageserver_relsize_cache_entries",
+        "Number of entries in the relation size cache",
    )
    .expect("failed to define a metric")
 });

-pub(crate) static RELSIZE_LATEST_CACHE_HITS: Lazy<IntCounter> = Lazy::new(|| {
+pub(crate) static RELSIZE_CACHE_HITS: Lazy<IntCounter> = Lazy::new(|| {
+    register_int_counter!("pageserver_relsize_cache_hits", "Relation size cache hits",)
+        .expect("failed to define a metric")
+});
+
+pub(crate) static RELSIZE_CACHE_MISSES: Lazy<IntCounter> = Lazy::new(|| {
    register_int_counter!(
-        "pageserver_relsize_latest_cache_hits",
-        "Latest relation size cache hits",
-    )
-    .expect("failed to define a metric")
-});
-
-pub(crate) static RELSIZE_LATEST_CACHE_MISSES: Lazy<IntCounter> = Lazy::new(|| {
-    register_int_counter!(
-        "pageserver_relsize_latest_cache_misses",
-        "Relation size latest cache misses",
-    )
-    .expect("failed to define a metric")
-});
-
-pub(crate) static RELSIZE_SNAPSHOT_CACHE_ENTRIES: Lazy<UIntGauge> = Lazy::new(|| {
-    register_uint_gauge!(
-        "pageserver_relsize_snapshot_cache_entries",
-        "Number of entries in the pitr relation size cache",
-    )
-    .expect("failed to define a metric")
-});
-
-pub(crate) static RELSIZE_SNAPSHOT_CACHE_HITS: Lazy<IntCounter> = Lazy::new(|| {
-    register_int_counter!(
-        "pageserver_relsize_snapshot_cache_hits",
-        "Pitr relation size cache hits",
-    )
-    .expect("failed to define a metric")
-});
-
-pub(crate) static RELSIZE_SNAPSHOT_CACHE_MISSES: Lazy<IntCounter> = Lazy::new(|| {
-    register_int_counter!(
-        "pageserver_relsize_snapshot_cache_misses",
-        "Relation size snapshot cache misses",
+        "pageserver_relsize_cache_misses",
+        "Relation size cache misses",
    )
    .expect("failed to define a metric")
 });
@@ -1066,15 +1039,6 @@ pub(crate) static TENANT_SYNTHETIC_SIZE_METRIC: Lazy<UIntGaugeVec> = Lazy::new(|
    .expect("Failed to register pageserver_tenant_synthetic_cached_size_bytes metric")
 });

-pub(crate) static TENANT_OFFLOADED_TIMELINES: Lazy<UIntGaugeVec> = Lazy::new(|| {
-    register_uint_gauge_vec!(
-        "pageserver_tenant_offloaded_timelines",
-        "Number of offloaded timelines of a tenant",
-        &["tenant_id", "shard_id"]
-    )
-    .expect("Failed to register pageserver_tenant_offloaded_timelines metric")
-});
-
 pub(crate) static EVICTION_ITERATION_DURATION: Lazy<HistogramVec> = Lazy::new(|| {
    register_histogram_vec!(
        "pageserver_eviction_iteration_duration_seconds_global",
@@ -3560,14 +3524,11 @@ impl TimelineMetrics {
 }

 pub(crate) fn remove_tenant_metrics(tenant_shard_id: &TenantShardId) {
-    let tid = tenant_shard_id.tenant_id.to_string();
-    let shard_id = tenant_shard_id.shard_slug().to_string();
-
    // Only shard zero deals in synthetic sizes
    if tenant_shard_id.is_shard_zero() {
+        let tid = tenant_shard_id.tenant_id.to_string();
        let _ = TENANT_SYNTHETIC_SIZE_METRIC.remove_label_values(&[&tid]);
    }
-    let _ = TENANT_OFFLOADED_TIMELINES.remove_label_values(&[&tid, &shard_id]);

    tenant_throttling::remove_tenant_metrics(tenant_shard_id);

@@ -4359,42 +4320,6 @@ pub(crate) fn set_tokio_runtime_setup(setup: &str, num_threads: NonZeroUsize) {
        .set(u64::try_from(num_threads.get()).unwrap());
 }

-pub(crate) static BASEBACKUP_CACHE_READ: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "pageserver_basebackup_cache_read_total",
-        "Number of read accesses to the basebackup cache grouped by hit/miss/error",
-        &["result"]
-    )
-    .expect("failed to define a metric")
-});
-
-pub(crate) static BASEBACKUP_CACHE_PREPARE: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "pageserver_basebackup_cache_prepare_total",
-        "Number of prepare requests processed by the basebackup cache grouped by ok/skip/error",
-        &["result"]
-    )
-    .expect("failed to define a metric")
-});
-
-pub(crate) static BASEBACKUP_CACHE_ENTRIES: Lazy<IntGauge> = Lazy::new(|| {
-    register_int_gauge!(
-        "pageserver_basebackup_cache_entries_total",
-        "Number of entries in the basebackup cache"
-    )
-    .expect("failed to define a metric")
-});
-
-// FIXME: Support basebackup cache size metrics.
-#[allow(dead_code)]
-pub(crate) static BASEBACKUP_CACHE_SIZE: Lazy<IntGauge> = Lazy::new(|| {
-    register_int_gauge!(
-        "pageserver_basebackup_cache_size_bytes",
-        "Total size of all basebackup cache entries on disk in bytes"
-    )
-    .expect("failed to define a metric")
-});
-
 static PAGESERVER_CONFIG_IGNORED_ITEMS: Lazy<UIntGaugeVec> = Lazy::new(|| {
    register_uint_gauge_vec!(
        "pageserver_config_ignored_items",
--- a/pageserver/src/page_service.rs
+++ b/pageserver/src/page_service.rs
@@ -9,6 +9,7 @@ use std::sync::Arc;
 use std::time::{Duration, Instant, SystemTime};
 use std::{io, str};

+use crate::PERF_TRACE_TARGET;
 use anyhow::{Context, bail};
 use async_compression::tokio::write::GzipEncoder;
 use bytes::Buf;
@@ -17,7 +18,7 @@ use itertools::Itertools;
 use jsonwebtoken::TokenData;
 use once_cell::sync::OnceCell;
 use pageserver_api::config::{
-    GetVectoredConcurrentIo, PageServicePipeliningConfig, PageServicePipeliningConfigPipelined,
+    PageServicePipeliningConfig, PageServicePipeliningConfigPipelined,
    PageServiceProtocolPipelinedBatchingStrategy, PageServiceProtocolPipelinedExecutionStrategy,
 };
 use pageserver_api::key::rel_block_to_key;
@@ -51,10 +52,8 @@ use utils::simple_rcu::RcuReadGuard;
 use utils::sync::gate::{Gate, GateGuard};
 use utils::sync::spsc_fold;

-use crate::PERF_TRACE_TARGET;
 use crate::auth::check_permission;
 use crate::basebackup::BasebackupError;
-use crate::basebackup_cache::BasebackupCache;
 use crate::config::PageServerConf;
 use crate::context::{
    DownloadBehavior, PerfInstrumentFutureExt, RequestContext, RequestContextBuilder,
@@ -63,7 +62,7 @@ use crate::metrics::{
    self, COMPUTE_COMMANDS_COUNTERS, ComputeCommandKind, GetPageBatchBreakReason, LIVE_CONNECTIONS,
    SmgrOpTimer, TimelineMetrics,
 };
-use crate::pgdatadir_mapping::{LsnRange, Version};
+use crate::pgdatadir_mapping::Version;
 use crate::span::{
    debug_assert_current_span_has_tenant_and_timeline_id,
    debug_assert_current_span_has_tenant_and_timeline_id_no_shard_id,
@@ -108,7 +107,6 @@ pub fn spawn(
    perf_trace_dispatch: Option<Dispatch>,
    tcp_listener: tokio::net::TcpListener,
    tls_config: Option<Arc<rustls::ServerConfig>>,
-    basebackup_cache: Arc<BasebackupCache>,
 ) -> Listener {
    let cancel = CancellationToken::new();
    let libpq_ctx = RequestContext::todo_child(
@@ -130,7 +128,6 @@ pub fn spawn(
            conf.pg_auth_type,
            tls_config,
            conf.page_service_pipelining.clone(),
-            basebackup_cache,
            libpq_ctx,
            cancel.clone(),
        )
@@ -189,7 +186,6 @@ pub async fn libpq_listener_main(
    auth_type: AuthType,
    tls_config: Option<Arc<rustls::ServerConfig>>,
    pipelining_config: PageServicePipeliningConfig,
-    basebackup_cache: Arc<BasebackupCache>,
    listener_ctx: RequestContext,
    listener_cancel: CancellationToken,
 ) -> Connections {
@@ -233,7 +229,6 @@ pub async fn libpq_listener_main(
                    auth_type,
                    tls_config.clone(),
                    pipelining_config.clone(),
-                    Arc::clone(&basebackup_cache),
                    connection_ctx,
                    connections_cancel.child_token(),
                    gate_guard,
@@ -276,7 +271,6 @@ async fn page_service_conn_main(
    auth_type: AuthType,
    tls_config: Option<Arc<rustls::ServerConfig>>,
    pipelining_config: PageServicePipeliningConfig,
-    basebackup_cache: Arc<BasebackupCache>,
    connection_ctx: RequestContext,
    cancel: CancellationToken,
    gate_guard: GateGuard,
@@ -337,12 +331,11 @@ async fn page_service_conn_main(
    // But it's in a shared crate, so, we store connection_ctx inside PageServerHandler
    // and create the per-query context in process_query ourselves.
    let mut conn_handler = PageServerHandler::new(
+        conf,
        tenant_manager,
        auth,
        pipelining_config,
-        conf.get_vectored_concurrent_io,
        perf_span_fields,
-        basebackup_cache,
        connection_ctx,
        cancel.clone(),
        gate_guard,
@@ -378,6 +371,7 @@ async fn page_service_conn_main(
 }

 struct PageServerHandler {
+    conf: &'static PageServerConf,
    auth: Option<Arc<SwappableJwtAuth>>,
    claims: Option<Claims>,

@@ -395,9 +389,6 @@ struct PageServerHandler {
    timeline_handles: Option<TimelineHandles>,

    pipelining_config: PageServicePipeliningConfig,
-    get_vectored_concurrent_io: GetVectoredConcurrentIo,
-
-    basebackup_cache: Arc<BasebackupCache>,

    gate_guard: GateGuard,
 }
@@ -651,7 +642,7 @@ impl std::fmt::Display for BatchedPageStreamError {
 struct BatchedGetPageRequest {
    req: PagestreamGetPageRequest,
    timer: SmgrOpTimer,
-    lsn_range: LsnRange,
+    effective_request_lsn: Lsn,
    ctx: RequestContext,
 }

@@ -773,12 +764,12 @@ impl BatchedFeMessage {
                match batching_strategy {
                    PageServiceProtocolPipelinedBatchingStrategy::UniformLsn => {
                        if let Some(last_in_batch) = accum_pages.last() {
-                            if last_in_batch.lsn_range.effective_lsn
-                                != this_pages[0].lsn_range.effective_lsn
+                            if last_in_batch.effective_request_lsn
+                                != this_pages[0].effective_request_lsn
                            {
                                trace!(
-                                    accum_lsn = %last_in_batch.lsn_range.effective_lsn,
-                                    this_lsn = %this_pages[0].lsn_range.effective_lsn,
+                                    accum_lsn = %last_in_batch.effective_request_lsn,
+                                    this_lsn = %this_pages[0].effective_request_lsn,
                                    "stopping batching because LSN changed"
                                );

@@ -793,15 +784,15 @@ impl BatchedFeMessage {
                        let same_page_different_lsn = accum_pages.iter().any(|batched| {
                            batched.req.rel == this_pages[0].req.rel
                                && batched.req.blkno == this_pages[0].req.blkno
-                                && batched.lsn_range.effective_lsn
-                                    != this_pages[0].lsn_range.effective_lsn
+                                && batched.effective_request_lsn
+                                    != this_pages[0].effective_request_lsn
                        });

                        if same_page_different_lsn {
                            trace!(
                                rel=%this_pages[0].req.rel,
                                blkno=%this_pages[0].req.blkno,
-                                lsn=%this_pages[0].lsn_range.effective_lsn,
+                                lsn=%this_pages[0].effective_request_lsn,
                                "stopping batching because same page was requested at different LSNs"
                            );

@@ -853,17 +844,17 @@ impl BatchedFeMessage {
 impl PageServerHandler {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
+        conf: &'static PageServerConf,
        tenant_manager: Arc<TenantManager>,
        auth: Option<Arc<SwappableJwtAuth>>,
        pipelining_config: PageServicePipeliningConfig,
-        get_vectored_concurrent_io: GetVectoredConcurrentIo,
        perf_span_fields: ConnectionPerfSpanFields,
-        basebackup_cache: Arc<BasebackupCache>,
        connection_ctx: RequestContext,
        cancel: CancellationToken,
        gate_guard: GateGuard,
    ) -> Self {
        PageServerHandler {
+            conf,
            auth,
            claims: None,
            connection_ctx,
@@ -871,8 +862,6 @@ impl PageServerHandler {
            timeline_handles: Some(TimelineHandles::new(tenant_manager)),
            cancel,
            pipelining_config,
-            get_vectored_concurrent_io,
-            basebackup_cache,
            gate_guard,
        }
    }
@@ -1169,7 +1158,7 @@ impl PageServerHandler {
                .await?;

                // We're holding the Handle
-                let effective_lsn = match Self::effective_request_lsn(
+                let effective_request_lsn = match Self::effective_request_lsn(
                    &shard,
                    shard.get_last_record_lsn(),
                    req.hdr.request_lsn,
@@ -1188,10 +1177,7 @@ impl PageServerHandler {
                    pages: smallvec::smallvec![BatchedGetPageRequest {
                        req,
                        timer,
-                        lsn_range: LsnRange {
-                            effective_lsn,
-                            request_lsn: req.hdr.request_lsn
-                        },
+                        effective_request_lsn,
                        ctx,
                    }],
                    // The executor grabs the batch when it becomes idle.
@@ -1637,7 +1623,7 @@ impl PageServerHandler {
        }

        let io_concurrency = IoConcurrency::spawn_from_conf(
-            self.get_vectored_concurrent_io,
+            self.conf,
            match self.gate_guard.try_clone() {
                Ok(guard) => guard,
                Err(_) => {
@@ -2141,14 +2127,7 @@ impl PageServerHandler {
        .await?;

        let exists = timeline
-            .get_rel_exists(
-                req.rel,
-                Version::LsnRange(LsnRange {
-                    effective_lsn: lsn,
-                    request_lsn: req.hdr.request_lsn,
-                }),
-                ctx,
-            )
+            .get_rel_exists(req.rel, Version::Lsn(lsn), ctx)
            .await?;

        Ok(PagestreamBeMessage::Exists(PagestreamExistsResponse {
@@ -2175,14 +2154,7 @@ impl PageServerHandler {
        .await?;

        let n_blocks = timeline
-            .get_rel_size(
-                req.rel,
-                Version::LsnRange(LsnRange {
-                    effective_lsn: lsn,
-                    request_lsn: req.hdr.request_lsn,
-                }),
-                ctx,
-            )
+            .get_rel_size(req.rel, Version::Lsn(lsn), ctx)
            .await?;

        Ok(PagestreamBeMessage::Nblocks(PagestreamNblocksResponse {
@@ -2209,15 +2181,7 @@ impl PageServerHandler {
        .await?;

        let total_blocks = timeline
-            .get_db_size(
-                DEFAULTTABLESPACE_OID,
-                req.dbnode,
-                Version::LsnRange(LsnRange {
-                    effective_lsn: lsn,
-                    request_lsn: req.hdr.request_lsn,
-                }),
-                ctx,
-            )
+            .get_db_size(DEFAULTTABLESPACE_OID, req.dbnode, Version::Lsn(lsn), ctx)
            .await?;
        let db_size = total_blocks as i64 * BLCKSZ as i64;

@@ -2250,7 +2214,7 @@ impl PageServerHandler {
                // Ignore error (trace buffer may be full or tracer may have disconnected).
                _ = page_trace.try_send(PageTraceEvent {
                    key,
-                    effective_lsn: batch.lsn_range.effective_lsn,
+                    effective_lsn: batch.effective_request_lsn,
                    time,
                });
            }
@@ -2265,7 +2229,7 @@ impl PageServerHandler {
                    perf_instrument = true;
                }

-                req.lsn_range.effective_lsn
+                req.effective_request_lsn
            })
            .max()
            .expect("batch is never empty");
@@ -2319,7 +2283,7 @@ impl PageServerHandler {
                    (
                        &p.req.rel,
                        &p.req.blkno,
-                        p.lsn_range,
+                        p.effective_request_lsn,
                        p.ctx.attached_child(),
                    )
                }),
@@ -2504,8 +2468,6 @@ impl PageServerHandler {
            .map_err(QueryError::Disconnected)?;
        self.flush_cancellable(pgb, &self.cancel).await?;

-        let mut from_cache = false;
-
        // Send a tarball of the latest layer on the timeline. Compress if not
        // fullbackup. TODO Compress in that case too (tests need to be updated)
        if full_backup {
@@ -2523,33 +2485,7 @@ impl PageServerHandler {
            .map_err(map_basebackup_error)?;
        } else {
            let mut writer = BufWriter::new(pgb.copyout_writer());
-
-            let cached = {
-                // Basebackup is cached only for this combination of parameters.
-                if timeline.is_basebackup_cache_enabled()
-                    && gzip
-                    && lsn.is_some()
-                    && prev_lsn.is_none()
-                {
-                    self.basebackup_cache
-                        .get(tenant_id, timeline_id, lsn.unwrap())
-                        .await
-                } else {
-                    None
-                }
-            };
-
-            if let Some(mut cached) = cached {
-                from_cache = true;
-                tokio::io::copy(&mut cached, &mut writer)
-                    .await
-                    .map_err(|e| {
-                        map_basebackup_error(BasebackupError::Client(
-                            e,
-                            "handle_basebackup_request,cached,copy",
-                        ))
-                    })?;
-            } else if gzip {
+            if gzip {
                let mut encoder = GzipEncoder::with_quality(
                    &mut writer,
                    // NOTE using fast compression because it's on the critical path
@@ -2608,7 +2544,6 @@ impl PageServerHandler {
        info!(
            lsn_await_millis = lsn_awaited_after.as_millis(),
            basebackup_millis = basebackup_after.as_millis(),
-            %from_cache,
            "basebackup complete"
        );

--- a/pageserver/src/pgdatadir_mapping.rs
+++ b/pageserver/src/pgdatadir_mapping.rs
@@ -43,9 +43,7 @@ use crate::aux_file;
 use crate::context::{PerfInstrumentFutureExt, RequestContext, RequestContextBuilder};
 use crate::keyspace::{KeySpace, KeySpaceAccum};
 use crate::metrics::{
-    RELSIZE_CACHE_MISSES_OLD, RELSIZE_LATEST_CACHE_ENTRIES, RELSIZE_LATEST_CACHE_HITS,
-    RELSIZE_LATEST_CACHE_MISSES, RELSIZE_SNAPSHOT_CACHE_ENTRIES, RELSIZE_SNAPSHOT_CACHE_HITS,
-    RELSIZE_SNAPSHOT_CACHE_MISSES,
+    RELSIZE_CACHE_ENTRIES, RELSIZE_CACHE_HITS, RELSIZE_CACHE_MISSES, RELSIZE_CACHE_MISSES_OLD,
 };
 use crate::span::{
    debug_assert_current_span_has_tenant_and_timeline_id,
@@ -92,28 +90,6 @@ pub enum LsnForTimestamp {
    NoData(Lsn),
 }

-/// Each request to page server contains LSN range: `not_modified_since..request_lsn`.
-/// See comments libs/pageserver_api/src/models.rs.
-/// Based on this range and `last_record_lsn` PS calculates `effective_lsn`.
-/// But to distinguish requests from primary and replicas we need also to pass `request_lsn`.
-#[derive(Debug, Clone, Copy, Default)]
-pub struct LsnRange {
-    pub effective_lsn: Lsn,
-    pub request_lsn: Lsn,
-}
-
-impl LsnRange {
-    pub fn at(lsn: Lsn) -> LsnRange {
-        LsnRange {
-            effective_lsn: lsn,
-            request_lsn: lsn,
-        }
-    }
-    pub fn is_latest(&self) -> bool {
-        self.request_lsn == Lsn::MAX
-    }
-}
-
 #[derive(Debug, thiserror::Error)]
 pub(crate) enum CalculateLogicalSizeError {
    #[error("cancelled")]
@@ -226,13 +202,13 @@ impl Timeline {
        io_concurrency: IoConcurrency,
    ) -> Result<Bytes, PageReconstructError> {
        match version {
-            Version::LsnRange(lsns) => {
+            Version::Lsn(effective_lsn) => {
                let pages: smallvec::SmallVec<[_; 1]> = smallvec::smallvec![(tag, blknum)];
                let res = self
                    .get_rel_page_at_lsn_batched(
-                        pages
-                            .iter()
-                            .map(|(tag, blknum)| (tag, blknum, lsns, ctx.attached_child())),
+                        pages.iter().map(|(tag, blknum)| {
+                            (tag, blknum, effective_lsn, ctx.attached_child())
+                        }),
                        io_concurrency.clone(),
                        ctx,
                    )
@@ -270,7 +246,7 @@ impl Timeline {
    /// The ordering of the returned vec corresponds to the ordering of `pages`.
    pub(crate) async fn get_rel_page_at_lsn_batched(
        &self,
-        pages: impl ExactSizeIterator<Item = (&RelTag, &BlockNumber, LsnRange, RequestContext)>,
+        pages: impl ExactSizeIterator<Item = (&RelTag, &BlockNumber, Lsn, RequestContext)>,
        io_concurrency: IoConcurrency,
        ctx: &RequestContext,
    ) -> Vec<Result<Bytes, PageReconstructError>> {
@@ -289,7 +265,7 @@ impl Timeline {
        let mut req_keyspaces: HashMap<Lsn, KeySpaceRandomAccum> =
            HashMap::with_capacity(pages.len());

-        for (response_slot_idx, (tag, blknum, lsns, ctx)) in pages.enumerate() {
+        for (response_slot_idx, (tag, blknum, lsn, ctx)) in pages.enumerate() {
            if tag.relnode == 0 {
                result_slots[response_slot_idx].write(Err(PageReconstructError::Other(
                    RelationError::InvalidRelnode.into(),
@@ -298,7 +274,7 @@ impl Timeline {
                slots_filled += 1;
                continue;
            }
-            let lsn = lsns.effective_lsn;
+
            let nblocks = {
                let ctx = RequestContextBuilder::from(&ctx)
                    .perf_span(|crnt_perf_span| {
@@ -313,7 +289,7 @@ impl Timeline {
                    .attached_child();

                match self
-                    .get_rel_size(*tag, Version::LsnRange(lsns), &ctx)
+                    .get_rel_size(*tag, Version::Lsn(lsn), &ctx)
                    .maybe_perf_instrument(&ctx, |crnt_perf_span| crnt_perf_span.clone())
                    .await
                {
@@ -494,7 +470,7 @@ impl Timeline {
            ));
        }

-        if let Some(nblocks) = self.get_cached_rel_size(&tag, version) {
+        if let Some(nblocks) = self.get_cached_rel_size(&tag, version.get_lsn()) {
            return Ok(nblocks);
        }

@@ -512,7 +488,7 @@ impl Timeline {
        let mut buf = version.get(self, key, ctx).await?;
        let nblocks = buf.get_u32_le();

-        self.update_cached_rel_size(tag, version, nblocks);
+        self.update_cached_rel_size(tag, version.get_lsn(), nblocks);

        Ok(nblocks)
    }
@@ -534,7 +510,7 @@ impl Timeline {
        }

        // first try to lookup relation in cache
-        if let Some(_nblocks) = self.get_cached_rel_size(&tag, version) {
+        if let Some(_nblocks) = self.get_cached_rel_size(&tag, version.get_lsn()) {
            return Ok(true);
        }
        // then check if the database was already initialized.
@@ -610,7 +586,7 @@ impl Timeline {
        // scan directory listing (new), merge with the old results
        let key_range = rel_tag_sparse_key_range(spcnode, dbnode);
        let io_concurrency = IoConcurrency::spawn_from_conf(
-            self.conf.get_vectored_concurrent_io,
+            self.conf,
            self.gate
                .enter()
                .map_err(|_| PageReconstructError::Cancelled)?,
@@ -656,7 +632,7 @@ impl Timeline {
    ) -> Result<Bytes, PageReconstructError> {
        assert!(self.tenant_shard_id.is_shard_zero());
        let n_blocks = self
-            .get_slru_segment_size(kind, segno, Version::at(lsn), ctx)
+            .get_slru_segment_size(kind, segno, Version::Lsn(lsn), ctx)
            .await?;

        let keyspace = KeySpace::single(
@@ -669,7 +645,7 @@ impl Timeline {
        );

        let io_concurrency = IoConcurrency::spawn_from_conf(
-            self.conf.get_vectored_concurrent_io,
+            self.conf,
            self.gate
                .enter()
                .map_err(|_| PageReconstructError::Cancelled)?,
@@ -891,11 +867,11 @@ impl Timeline {
        mut f: impl FnMut(TimestampTz) -> ControlFlow<T>,
    ) -> Result<T, PageReconstructError> {
        for segno in self
-            .list_slru_segments(SlruKind::Clog, Version::at(probe_lsn), ctx)
+            .list_slru_segments(SlruKind::Clog, Version::Lsn(probe_lsn), ctx)
            .await?
        {
            let nblocks = self
-                .get_slru_segment_size(SlruKind::Clog, segno, Version::at(probe_lsn), ctx)
+                .get_slru_segment_size(SlruKind::Clog, segno, Version::Lsn(probe_lsn), ctx)
                .await?;

            let keyspace = KeySpace::single(
@@ -909,7 +885,7 @@ impl Timeline {
            );

            let io_concurrency = IoConcurrency::spawn_from_conf(
-                self.conf.get_vectored_concurrent_io,
+                self.conf,
                self.gate
                    .enter()
                    .map_err(|_| PageReconstructError::Cancelled)?,
@@ -1161,7 +1137,7 @@ impl Timeline {
        let mut total_size: u64 = 0;
        for (spcnode, dbnode) in dbdir.dbdirs.keys() {
            for rel in self
-                .list_rels(*spcnode, *dbnode, Version::at(lsn), ctx)
+                .list_rels(*spcnode, *dbnode, Version::Lsn(lsn), ctx)
                .await?
            {
                if self.cancel.is_cancelled() {
@@ -1236,7 +1212,7 @@ impl Timeline {
            result.add_key(rel_dir_to_key(spcnode, dbnode));

            let mut rels: Vec<RelTag> = self
-                .list_rels(spcnode, dbnode, Version::at(lsn), ctx)
+                .list_rels(spcnode, dbnode, Version::Lsn(lsn), ctx)
                .await?
                .into_iter()
                .collect();
@@ -1353,75 +1329,59 @@ impl Timeline {
        Ok((dense_keyspace, sparse_keyspace))
    }

-    /// Get cached size of relation. There are two caches: one for primary updates, it captures the latest state of
-    /// of the timeline and snapshot cache, which key includes LSN and so can be used by replicas to get relation size
-    /// at the particular LSN (snapshot).
-    pub fn get_cached_rel_size(&self, tag: &RelTag, version: Version<'_>) -> Option<BlockNumber> {
-        let lsn = version.get_lsn();
-        {
-            let rel_size_cache = self.rel_size_latest_cache.read().unwrap();
-            if let Some((cached_lsn, nblocks)) = rel_size_cache.get(tag) {
-                if lsn >= *cached_lsn {
-                    RELSIZE_LATEST_CACHE_HITS.inc();
-                    return Some(*nblocks);
-                }
-                RELSIZE_CACHE_MISSES_OLD.inc();
+    /// Get cached size of relation if it not updated after specified LSN
+    pub fn get_cached_rel_size(&self, tag: &RelTag, lsn: Lsn) -> Option<BlockNumber> {
+        let rel_size_cache = self.rel_size_cache.read().unwrap();
+        if let Some((cached_lsn, nblocks)) = rel_size_cache.map.get(tag) {
+            if lsn >= *cached_lsn {
+                RELSIZE_CACHE_HITS.inc();
+                return Some(*nblocks);
            }
+            RELSIZE_CACHE_MISSES_OLD.inc();
        }
-        {
-            let mut rel_size_cache = self.rel_size_snapshot_cache.lock().unwrap();
-            if let Some(nblock) = rel_size_cache.get(&(lsn, *tag)) {
-                RELSIZE_SNAPSHOT_CACHE_HITS.inc();
-                return Some(*nblock);
-            }
-        }
-        if version.is_latest() {
-            RELSIZE_LATEST_CACHE_MISSES.inc();
-        } else {
-            RELSIZE_SNAPSHOT_CACHE_MISSES.inc();
-        }
+        RELSIZE_CACHE_MISSES.inc();
        None
    }

    /// Update cached relation size if there is no more recent update
-    pub fn update_cached_rel_size(&self, tag: RelTag, version: Version<'_>, nblocks: BlockNumber) {
-        let lsn = version.get_lsn();
-        if version.is_latest() {
-            let mut rel_size_cache = self.rel_size_latest_cache.write().unwrap();
-            match rel_size_cache.entry(tag) {
-                hash_map::Entry::Occupied(mut entry) => {
-                    let cached_lsn = entry.get_mut();
-                    if lsn >= cached_lsn.0 {
-                        *cached_lsn = (lsn, nblocks);
-                    }
-                }
-                hash_map::Entry::Vacant(entry) => {
-                    entry.insert((lsn, nblocks));
-                    RELSIZE_LATEST_CACHE_ENTRIES.inc();
+    pub fn update_cached_rel_size(&self, tag: RelTag, lsn: Lsn, nblocks: BlockNumber) {
+        let mut rel_size_cache = self.rel_size_cache.write().unwrap();
+
+        if lsn < rel_size_cache.complete_as_of {
+            // Do not cache old values. It's safe to cache the size on read, as long as
+            // the read was at an LSN since we started the WAL ingestion. Reasoning: we
+            // never evict values from the cache, so if the relation size changed after
+            // 'lsn', the new value is already in the cache.
+            return;
+        }
+
+        match rel_size_cache.map.entry(tag) {
+            hash_map::Entry::Occupied(mut entry) => {
+                let cached_lsn = entry.get_mut();
+                if lsn >= cached_lsn.0 {
+                    *cached_lsn = (lsn, nblocks);
                }
            }
-        } else {
-            let mut rel_size_cache = self.rel_size_snapshot_cache.lock().unwrap();
-            if rel_size_cache.capacity() != 0 {
-                rel_size_cache.insert((lsn, tag), nblocks);
-                RELSIZE_SNAPSHOT_CACHE_ENTRIES.set(rel_size_cache.len() as u64);
+            hash_map::Entry::Vacant(entry) => {
+                entry.insert((lsn, nblocks));
+                RELSIZE_CACHE_ENTRIES.inc();
            }
        }
    }

    /// Store cached relation size
    pub fn set_cached_rel_size(&self, tag: RelTag, lsn: Lsn, nblocks: BlockNumber) {
-        let mut rel_size_cache = self.rel_size_latest_cache.write().unwrap();
-        if rel_size_cache.insert(tag, (lsn, nblocks)).is_none() {
-            RELSIZE_LATEST_CACHE_ENTRIES.inc();
+        let mut rel_size_cache = self.rel_size_cache.write().unwrap();
+        if rel_size_cache.map.insert(tag, (lsn, nblocks)).is_none() {
+            RELSIZE_CACHE_ENTRIES.inc();
        }
    }

    /// Remove cached relation size
    pub fn remove_cached_rel_size(&self, tag: &RelTag) {
-        let mut rel_size_cache = self.rel_size_latest_cache.write().unwrap();
-        if rel_size_cache.remove(tag).is_some() {
-            RELSIZE_LATEST_CACHE_ENTRIES.dec();
+        let mut rel_size_cache = self.rel_size_cache.write().unwrap();
+        if rel_size_cache.map.remove(tag).is_some() {
+            RELSIZE_CACHE_ENTRIES.dec();
        }
    }
 }
@@ -1625,10 +1585,7 @@ impl DatadirModification<'_> {
        //       check the cache too. This is because eagerly checking the cache results in
        //       less work overall and 10% better performance. It's more work on cache miss
        //       but cache miss is rare.
-        if let Some(nblocks) = self
-            .tline
-            .get_cached_rel_size(&rel, Version::Modified(self))
-        {
+        if let Some(nblocks) = self.tline.get_cached_rel_size(&rel, self.get_lsn()) {
            Ok(nblocks)
        } else if !self
            .tline
@@ -2710,7 +2667,7 @@ pub struct DatadirModificationStats {
 /// timeline to not miss the latest updates.
 #[derive(Clone, Copy)]
 pub enum Version<'a> {
-    LsnRange(LsnRange),
+    Lsn(Lsn),
    Modified(&'a DatadirModification<'a>),
 }

@@ -2722,7 +2679,7 @@ impl Version<'_> {
        ctx: &RequestContext,
    ) -> Result<Bytes, PageReconstructError> {
        match self {
-            Version::LsnRange(lsns) => timeline.get(key, lsns.effective_lsn, ctx).await,
+            Version::Lsn(lsn) => timeline.get(key, *lsn, ctx).await,
            Version::Modified(modification) => modification.get(key, ctx).await,
        }
    }
@@ -2744,26 +2701,12 @@ impl Version<'_> {
        }
    }

-    pub fn is_latest(&self) -> bool {
+    fn get_lsn(&self) -> Lsn {
        match self {
-            Version::LsnRange(lsns) => lsns.is_latest(),
-            Version::Modified(_) => true,
-        }
-    }
-
-    pub fn get_lsn(&self) -> Lsn {
-        match self {
-            Version::LsnRange(lsns) => lsns.effective_lsn,
+            Version::Lsn(lsn) => *lsn,
            Version::Modified(modification) => modification.lsn,
        }
    }
-
-    pub fn at(lsn: Lsn) -> Self {
-        Version::LsnRange(LsnRange {
-            effective_lsn: lsn,
-            request_lsn: lsn,
-        })
-    }
 }

 //--- Metadata structs stored in key-value pairs in the repository.
--- a/pageserver/src/task_mgr.rs
+++ b/pageserver/src/task_mgr.rs
@@ -380,10 +380,6 @@ pub enum TaskKind {
    DetachAncestor,

    ImportPgdata,
-
-    /// Background task of [`crate::basebackup_cache::BasebackupCache`].
-    /// Prepares basebackups and clears outdated entries.
-    BasebackupCache,
 }

 #[derive(Default)]
--- a/pageserver/src/tenant.rs
+++ b/pageserver/src/tenant.rs
@@ -78,7 +78,6 @@ use self::timeline::uninit::{TimelineCreateGuard, TimelineExclusionError, Uninit
 use self::timeline::{
    EvictionTaskTenantState, GcCutoffs, TimelineDeleteProgress, TimelineResources, WaitLsnError,
 };
-use crate::basebackup_cache::BasebackupPrepareSender;
 use crate::config::PageServerConf;
 use crate::context;
 use crate::context::RequestContextBuilder;
@@ -87,8 +86,8 @@ use crate::deletion_queue::{DeletionQueueClient, DeletionQueueError};
 use crate::l0_flush::L0FlushGlobalState;
 use crate::metrics::{
    BROKEN_TENANTS_SET, CIRCUIT_BREAKERS_BROKEN, CIRCUIT_BREAKERS_UNBROKEN, CONCURRENT_INITDBS,
-    INITDB_RUN_TIME, INITDB_SEMAPHORE_ACQUISITION_TIME, TENANT, TENANT_OFFLOADED_TIMELINES,
-    TENANT_STATE_METRIC, TENANT_SYNTHETIC_SIZE_METRIC, remove_tenant_metrics,
+    INITDB_RUN_TIME, INITDB_SEMAPHORE_ACQUISITION_TIME, TENANT, TENANT_STATE_METRIC,
+    TENANT_SYNTHETIC_SIZE_METRIC, remove_tenant_metrics,
 };
 use crate::task_mgr::TaskKind;
 use crate::tenant::config::LocationMode;
@@ -158,7 +157,6 @@ pub struct TenantSharedResources {
    pub remote_storage: GenericRemoteStorage,
    pub deletion_queue_client: DeletionQueueClient,
    pub l0_flush_global_state: L0FlushGlobalState,
-    pub basebackup_prepare_sender: BasebackupPrepareSender,
 }

 /// A [`TenantShard`] is really an _attached_ tenant.  The configuration
@@ -319,15 +317,12 @@ pub struct TenantShard {
    gc_cs: tokio::sync::Mutex<()>,
    walredo_mgr: Option<Arc<WalRedoManager>>,

-    /// Provides access to timeline data sitting in the remote storage.
+    // provides access to timeline data sitting in the remote storage
    pub(crate) remote_storage: GenericRemoteStorage,

-    /// Access to global deletion queue for when this tenant wants to schedule a deletion.
+    // Access to global deletion queue for when this tenant wants to schedule a deletion
    deletion_queue_client: DeletionQueueClient,

-    /// A channel to send async requests to prepare a basebackup for the basebackup cache.
-    basebackup_prepare_sender: BasebackupPrepareSender,
-
    /// Cached logical sizes updated updated on each [`TenantShard::gather_size_inputs`].
    cached_logical_sizes: tokio::sync::Mutex<HashMap<(TimelineId, Lsn), u64>>,
    cached_synthetic_tenant_size: Arc<AtomicU64>,
@@ -1291,7 +1286,6 @@ impl TenantShard {
            remote_storage,
            deletion_queue_client,
            l0_flush_global_state,
-            basebackup_prepare_sender,
        } = resources;

        let attach_mode = attached_conf.location.attach_mode;
@@ -1307,7 +1301,6 @@ impl TenantShard {
            remote_storage.clone(),
            deletion_queue_client,
            l0_flush_global_state,
-            basebackup_prepare_sender,
        ));

        // The attach task will carry a GateGuard, so that shutdown() reliably waits for it to drop out if
@@ -3355,13 +3348,6 @@ impl TenantShard {
                activated_timelines += 1;
            }

-            let tid = self.tenant_shard_id.tenant_id.to_string();
-            let shard_id = self.tenant_shard_id.shard_slug().to_string();
-            let offloaded_timeline_count = timelines_offloaded_accessor.len();
-            TENANT_OFFLOADED_TIMELINES
-                .with_label_values(&[&tid, &shard_id])
-                .set(offloaded_timeline_count as u64);
-
            self.state.send_modify(move |current_state| {
                assert!(
                    matches!(current_state, TenantState::Activating(_)),
@@ -4246,7 +4232,6 @@ impl TenantShard {
        remote_storage: GenericRemoteStorage,
        deletion_queue_client: DeletionQueueClient,
        l0_flush_global_state: L0FlushGlobalState,
-        basebackup_prepare_sender: BasebackupPrepareSender,
    ) -> TenantShard {
        assert!(!attached_conf.location.generation.is_none());

@@ -4350,7 +4335,6 @@ impl TenantShard {
            ongoing_timeline_detach: std::sync::Mutex::default(),
            gc_block: Default::default(),
            l0_flush_global_state,
-            basebackup_prepare_sender,
        }
    }

@@ -4603,7 +4587,7 @@ impl TenantShard {

            target.cutoffs = GcCutoffs {
                space: space_cutoff,
-                time: None,
+                time: Lsn::INVALID,
            };
        }
    }
@@ -4687,7 +4671,7 @@ impl TenantShard {
                if let Some(ancestor_id) = timeline.get_ancestor_timeline_id() {
                    if let Some(ancestor_gc_cutoffs) = gc_cutoffs.get(&ancestor_id) {
                        target.within_ancestor_pitr =
-                            Some(timeline.get_ancestor_lsn()) >= ancestor_gc_cutoffs.time;
+                            timeline.get_ancestor_lsn() >= ancestor_gc_cutoffs.time;
                    }
                }

@@ -4700,15 +4684,13 @@ impl TenantShard {
                    } else {
                        0
                    });
-                if let Some(time_cutoff) = target.cutoffs.time {
-                    timeline.metrics.pitr_history_size.set(
-                        timeline
-                            .get_last_record_lsn()
-                            .checked_sub(time_cutoff)
-                            .unwrap_or_default()
-                            .0,
-                    );
-                }
+                timeline.metrics.pitr_history_size.set(
+                    timeline
+                        .get_last_record_lsn()
+                        .checked_sub(target.cutoffs.time)
+                        .unwrap_or(Lsn(0))
+                        .0,
+                );

                // Apply the cutoffs we found to the Timeline's GcInfo.  Why might we _not_ have cutoffs for a timeline?
                // - this timeline was created while we were finding cutoffs
@@ -4717,8 +4699,8 @@ impl TenantShard {
                    let original_cutoffs = target.cutoffs.clone();
                    // GC cutoffs should never go back
                    target.cutoffs = GcCutoffs {
-                        space: cutoffs.space.max(original_cutoffs.space),
-                        time: cutoffs.time.max(original_cutoffs.time),
+                        space: Lsn(cutoffs.space.0.max(original_cutoffs.space.0)),
+                        time: Lsn(cutoffs.time.0.max(original_cutoffs.time.0)),
                    }
                }
            }
@@ -5270,7 +5252,6 @@ impl TenantShard {
            pagestream_throttle_metrics: self.pagestream_throttle_metrics.clone(),
            l0_compaction_trigger: self.l0_compaction_trigger.clone(),
            l0_flush_global_state: self.l0_flush_global_state.clone(),
-            basebackup_prepare_sender: self.basebackup_prepare_sender.clone(),
        }
    }

@@ -5579,14 +5560,6 @@ impl TenantShard {
            }
        }

-        // Update metrics
-        let tid = self.tenant_shard_id.to_string();
-        let shard_id = self.tenant_shard_id.shard_slug().to_string();
-        let set_key = &[tid.as_str(), shard_id.as_str()][..];
-        TENANT_OFFLOADED_TIMELINES
-            .with_label_values(set_key)
-            .set(manifest.offloaded_timelines.len() as u64);
-
        // Upload the manifest. Remote storage does no retries internally, so retry here.
        match backoff::retry(
            || async {
@@ -5853,8 +5826,6 @@ pub(crate) mod harness {
        ) -> anyhow::Result<Arc<TenantShard>> {
            let walredo_mgr = Arc::new(WalRedoManager::from(TestRedoManager));

-            let (basebackup_requst_sender, _) = tokio::sync::mpsc::unbounded_channel();
-
            let tenant = Arc::new(TenantShard::new(
                TenantState::Attaching,
                self.conf,
@@ -5872,7 +5843,6 @@ pub(crate) mod harness {
                self.deletion_queue.new_client(),
                // TODO: ideally we should run all unit tests with both configs
                L0FlushGlobalState::new(L0FlushConfig::default()),
-                basebackup_requst_sender,
            ));

            let preload = tenant
@@ -8626,10 +8596,8 @@ mod tests {
        lsn: Lsn,
        ctx: &RequestContext,
    ) -> Result<Option<Bytes>, GetVectoredError> {
-        let io_concurrency = IoConcurrency::spawn_from_conf(
-            tline.conf.get_vectored_concurrent_io,
-            tline.gate.enter().unwrap(),
-        );
+        let io_concurrency =
+            IoConcurrency::spawn_from_conf(tline.conf, tline.gate.enter().unwrap());
        let mut reconstruct_state = ValuesReconstructState::new(io_concurrency);
        let query = VersionedKeySpaceQuery::uniform(KeySpace::single(key..key.next()), lsn);
        let mut res = tline
@@ -8967,7 +8935,7 @@ mod tests {
                .await;
            // Update GC info
            let mut guard = tline.gc_info.write().unwrap();
-            guard.cutoffs.time = Some(Lsn(0x30));
+            guard.cutoffs.time = Lsn(0x30);
            guard.cutoffs.space = Lsn(0x30);
        }

@@ -9075,7 +9043,7 @@ mod tests {
                .await;
            // Update GC info
            let mut guard = tline.gc_info.write().unwrap();
-            guard.cutoffs.time = Some(Lsn(0x40));
+            guard.cutoffs.time = Lsn(0x40);
            guard.cutoffs.space = Lsn(0x40);
        }
        tline
@@ -9493,7 +9461,7 @@ mod tests {
            *guard = GcInfo {
                retain_lsns: vec![],
                cutoffs: GcCutoffs {
-                    time: Some(Lsn(0x30)),
+                    time: Lsn(0x30),
                    space: Lsn(0x30),
                },
                leases: Default::default(),
@@ -9577,7 +9545,7 @@ mod tests {
                .await;
            // Update GC info
            let mut guard = tline.gc_info.write().unwrap();
-            guard.cutoffs.time = Some(Lsn(0x40));
+            guard.cutoffs.time = Lsn(0x40);
            guard.cutoffs.space = Lsn(0x40);
        }
        tline
@@ -10048,7 +10016,7 @@ mod tests {
                    (Lsn(0x20), tline.timeline_id, MaybeOffloaded::No),
                ],
                cutoffs: GcCutoffs {
-                    time: Some(Lsn(0x30)),
+                    time: Lsn(0x30),
                    space: Lsn(0x30),
                },
                leases: Default::default(),
@@ -10111,7 +10079,7 @@ mod tests {
        let verify_result = || async {
            let gc_horizon = {
                let gc_info = tline.gc_info.read().unwrap();
-                gc_info.cutoffs.time.unwrap_or_default()
+                gc_info.cutoffs.time
            };
            for idx in 0..10 {
                assert_eq!(
@@ -10189,7 +10157,7 @@ mod tests {
                .await;
            // Update GC info
            let mut guard = tline.gc_info.write().unwrap();
-            guard.cutoffs.time = Some(Lsn(0x38));
+            guard.cutoffs.time = Lsn(0x38);
            guard.cutoffs.space = Lsn(0x38);
        }
        tline
@@ -10297,7 +10265,7 @@ mod tests {
                    (Lsn(0x20), tline.timeline_id, MaybeOffloaded::No),
                ],
                cutoffs: GcCutoffs {
-                    time: Some(Lsn(0x30)),
+                    time: Lsn(0x30),
                    space: Lsn(0x30),
                },
                leases: Default::default(),
@@ -10360,7 +10328,7 @@ mod tests {
        let verify_result = || async {
            let gc_horizon = {
                let gc_info = tline.gc_info.read().unwrap();
-                gc_info.cutoffs.time.unwrap_or_default()
+                gc_info.cutoffs.time
            };
            for idx in 0..10 {
                assert_eq!(
@@ -10546,7 +10514,7 @@ mod tests {
            *guard = GcInfo {
                retain_lsns: vec![(Lsn(0x18), branch_tline.timeline_id, MaybeOffloaded::No)],
                cutoffs: GcCutoffs {
-                    time: Some(Lsn(0x10)),
+                    time: Lsn(0x10),
                    space: Lsn(0x10),
                },
                leases: Default::default(),
@@ -10566,7 +10534,7 @@ mod tests {
            *guard = GcInfo {
                retain_lsns: vec![(Lsn(0x40), branch_tline.timeline_id, MaybeOffloaded::No)],
                cutoffs: GcCutoffs {
-                    time: Some(Lsn(0x50)),
+                    time: Lsn(0x50),
                    space: Lsn(0x50),
                },
                leases: Default::default(),
@@ -11287,7 +11255,7 @@ mod tests {
            *guard = GcInfo {
                retain_lsns: vec![(Lsn(0x20), tline.timeline_id, MaybeOffloaded::No)],
                cutoffs: GcCutoffs {
-                    time: Some(Lsn(0x30)),
+                    time: Lsn(0x30),
                    space: Lsn(0x30),
                },
                leases: Default::default(),
@@ -11676,7 +11644,7 @@ mod tests {
                    (Lsn(0x20), tline.timeline_id, MaybeOffloaded::No),
                ],
                cutoffs: GcCutoffs {
-                    time: Some(Lsn(0x30)),
+                    time: Lsn(0x30),
                    space: Lsn(0x30),
                },
                leases: Default::default(),
@@ -11739,7 +11707,7 @@ mod tests {
        let verify_result = || async {
            let gc_horizon = {
                let gc_info = tline.gc_info.read().unwrap();
-                gc_info.cutoffs.time.unwrap_or_default()
+                gc_info.cutoffs.time
            };
            for idx in 0..10 {
                assert_eq!(
@@ -11928,7 +11896,7 @@ mod tests {
                    (Lsn(0x20), tline.timeline_id, MaybeOffloaded::No),
                ],
                cutoffs: GcCutoffs {
-                    time: Some(Lsn(0x30)),
+                    time: Lsn(0x30),
                    space: Lsn(0x30),
                },
                leases: Default::default(),
@@ -11991,7 +11959,7 @@ mod tests {
        let verify_result = || async {
            let gc_horizon = {
                let gc_info = tline.gc_info.read().unwrap();
-                gc_info.cutoffs.time.unwrap_or_default()
+                gc_info.cutoffs.time
            };
            for idx in 0..10 {
                assert_eq!(
@@ -12254,7 +12222,7 @@ mod tests {
            *guard = GcInfo {
                retain_lsns: vec![],
                cutoffs: GcCutoffs {
-                    time: Some(Lsn(0x30)),
+                    time: Lsn(0x30),
                    space: Lsn(0x30),
                },
                leases: Default::default(),
--- a/pageserver/src/tenant/size.rs
+++ b/pageserver/src/tenant/size.rs
@@ -235,7 +235,7 @@ pub(super) async fn gather_inputs(
        // than our internal space cutoff.  This means that if someone drops a database and waits for their
        // PITR interval, they will see synthetic size decrease, even if we are still storing data inside
        // the space cutoff.
-        let mut next_pitr_cutoff = gc_info.cutoffs.time.unwrap_or_default(); // TODO: handle None
+        let mut next_pitr_cutoff = gc_info.cutoffs.time;

        // If the caller provided a shorter retention period, use that instead of the GC cutoff.
        let retention_param_cutoff = if let Some(max_retention_period) = max_retention_period {
--- a/pageserver/src/tenant/storage_layer.rs
+++ b/pageserver/src/tenant/storage_layer.rs
@@ -31,7 +31,6 @@ pub use inmemory_layer::InMemoryLayer;
 pub(crate) use layer::{EvictionError, Layer, ResidentLayer};
 pub use layer_desc::{PersistentLayerDesc, PersistentLayerKey};
 pub use layer_name::{DeltaLayerName, ImageLayerName, LayerName};
-use pageserver_api::config::GetVectoredConcurrentIo;
 use pageserver_api::key::Key;
 use pageserver_api::keyspace::{KeySpace, KeySpaceRandomAccum};
 use pageserver_api::record::NeonWalRecord;
@@ -44,6 +43,7 @@ use self::inmemory_layer::InMemoryLayerFileId;
 use super::PageReconstructError;
 use super::layer_map::InMemoryLayerDesc;
 use super::timeline::{GetVectoredError, ReadPath};
+use crate::config::PageServerConf;
 use crate::context::{
    AccessStatsBehavior, PerfInstrumentFutureExt, RequestContext, RequestContextBuilder,
 };
@@ -318,10 +318,11 @@ impl IoConcurrency {
    }

    pub(crate) fn spawn_from_conf(
-        conf: GetVectoredConcurrentIo,
+        conf: &'static PageServerConf,
        gate_guard: GateGuard,
    ) -> IoConcurrency {
-        let selected = match conf {
+        use pageserver_api::config::GetVectoredConcurrentIo;
+        let selected = match conf.get_vectored_concurrent_io {
            GetVectoredConcurrentIo::Sequential => SelectedIoConcurrency::Sequential,
            GetVectoredConcurrentIo::SidecarTask => SelectedIoConcurrency::SidecarTask(gate_guard),
        };
--- a/pageserver/src/tenant/storage_layer/inmemory_layer.rs
+++ b/pageserver/src/tenant/storage_layer/inmemory_layer.rs
@@ -63,28 +63,7 @@ pub struct InMemoryLayer {

    opened_at: Instant,

-    /// All versions of all pages in the layer are kept here. Indexed
-    /// by block number and LSN. The [`IndexEntry`] is an offset into the
-    /// ephemeral file where the page version is stored.
-    ///
-    /// We use a separate lock for the index to reduce the critical section
-    /// during which reads cannot be planned.
-    ///
-    /// If you need access to both the index and the underlying file at the same time,
-    /// respect the following locking order to avoid deadlocks:
-    /// 1. [`InMemoryLayer::inner`]
-    /// 2. [`InMemoryLayer::index`]
-    ///
-    /// Note that the file backing [`InMemoryLayer::inner`] is append-only,
-    /// so it is not necessary to hold simultaneous locks on index.
-    /// This avoids holding index locks across IO, and is crucial for avoiding read tail latency.
-    /// In particular:
-    /// 1. It is safe to read and release [`InMemoryLayer::index`] before locking and reading from [`InMemoryLayer::inner`].
-    /// 2. It is safe to write and release [`InMemoryLayer::inner`] before locking and updating [`InMemoryLayer::index`].
-    index: RwLock<BTreeMap<CompactKey, VecMap<Lsn, IndexEntry>>>,
-
-    /// The above fields never change, except for `end_lsn`, which is only set once,
-    /// and `index` (see rationale there).
+    /// The above fields never change, except for `end_lsn`, which is only set once.
    /// All other changing parts are in `inner`, and protected by a mutex.
    inner: RwLock<InMemoryLayerInner>,

@@ -102,6 +81,11 @@ impl std::fmt::Debug for InMemoryLayer {
 }

 pub struct InMemoryLayerInner {
+    /// All versions of all pages in the layer are kept here. Indexed
+    /// by block number and LSN. The [`IndexEntry`] is an offset into the
+    /// ephemeral file where the page version is stored.
+    index: BTreeMap<CompactKey, VecMap<Lsn, IndexEntry>>,
+
    /// The values are stored in a serialized format in this file.
    /// Each serialized Value is preceded by a 'u32' length field.
    /// PerSeg::page_versions map stores offsets into this file.
@@ -121,7 +105,7 @@ const MAX_SUPPORTED_BLOB_LEN_BITS: usize = {
    trailing_ones
 };

-/// See [`InMemoryLayer::index`].
+/// See [`InMemoryLayerInner::index`].
 ///
 /// For memory efficiency, the data is packed into a u64.
 ///
@@ -441,7 +425,7 @@ impl InMemoryLayer {
            .page_content_kind(PageContentKind::InMemoryLayer)
            .attached_child();

-        let index = self.index.read().await;
+        let inner = self.inner.read().await;

        struct ValueRead {
            entry_lsn: Lsn,
@@ -451,7 +435,10 @@ impl InMemoryLayer {
        let mut ios: HashMap<(Key, Lsn), OnDiskValueIo> = Default::default();

        for range in keyspace.ranges.iter() {
-            for (key, vec_map) in index.range(range.start.to_compact()..range.end.to_compact()) {
+            for (key, vec_map) in inner
+                .index
+                .range(range.start.to_compact()..range.end.to_compact())
+            {
                let key = Key::from_compact(*key);
                let slice = vec_map.slice_range(lsn_range.clone());

@@ -479,7 +466,7 @@ impl InMemoryLayer {
                }
            }
        }
-        drop(index); // release the lock before we spawn the IO; if it's serial-mode IO we will deadlock on the read().await below
+        drop(inner); // release the lock before we spawn the IO; if it's serial-mode IO we will deadlock on the read().await below
        let read_from = Arc::clone(self);
        let read_ctx = ctx.attached_child();
        reconstruct_state
@@ -586,8 +573,8 @@ impl InMemoryLayer {
            start_lsn,
            end_lsn: OnceLock::new(),
            opened_at: Instant::now(),
-            index: RwLock::new(BTreeMap::new()),
            inner: RwLock::new(InMemoryLayerInner {
+                index: BTreeMap::new(),
                file,
                resource_units: GlobalResourceUnits::new(),
            }),
@@ -605,39 +592,31 @@ impl InMemoryLayer {
        serialized_batch: SerializedValueBatch,
        ctx: &RequestContext,
    ) -> anyhow::Result<()> {
-        let (base_offset, metadata) = {
-            let mut inner = self.inner.write().await;
-            self.assert_writable();
+        let mut inner = self.inner.write().await;
+        self.assert_writable();

-            let base_offset = inner.file.len();
+        let base_offset = inner.file.len();

-            let SerializedValueBatch {
-                raw,
-                metadata,
-                max_lsn: _,
-                len: _,
-            } = serialized_batch;
+        let SerializedValueBatch {
+            raw,
+            metadata,
+            max_lsn: _,
+            len: _,
+        } = serialized_batch;

-            // Write the batch to the file
-            inner.file.write_raw(&raw, ctx).await?;
-            let new_size = inner.file.len();
+        // Write the batch to the file
+        inner.file.write_raw(&raw, ctx).await?;
+        let new_size = inner.file.len();

-            let expected_new_len = base_offset
-                .checked_add(raw.len().into_u64())
-                // write_raw would error if we were to overflow u64.
-                // also IndexEntry and higher levels in
-                //the code don't allow the file to grow that large
-                .unwrap();
-            assert_eq!(new_size, expected_new_len);
-
-            inner.resource_units.maybe_publish_size(new_size);
-
-            (base_offset, metadata)
-        };
+        let expected_new_len = base_offset
+            .checked_add(raw.len().into_u64())
+            // write_raw would error if we were to overflow u64.
+            // also IndexEntry and higher levels in
+            //the code don't allow the file to grow that large
+            .unwrap();
+        assert_eq!(new_size, expected_new_len);

        // Update the index with the new entries
-        let mut index = self.index.write().await;
-
        for meta in metadata {
            let SerializedValueMeta {
                key,
@@ -660,7 +639,7 @@ impl InMemoryLayer {
                will_init,
            })?;

-            let vec_map = index.entry(key).or_default();
+            let vec_map = inner.index.entry(key).or_default();
            let old = vec_map.append_or_update_last(lsn, index_entry).unwrap().0;
            if old.is_some() {
                // This should not break anything, but is unexpected: ingestion code aims to filter out
@@ -679,6 +658,8 @@ impl InMemoryLayer {
            );
        }

+        inner.resource_units.maybe_publish_size(new_size);
+
        Ok(())
    }

@@ -699,18 +680,6 @@ impl InMemoryLayer {

    /// Records the end_lsn for non-dropped layers.
    /// `end_lsn` is exclusive
-    ///
-    /// A note on locking:
-    /// The current API of [`InMemoryLayer`] does not ensure that there's no ongoing
-    /// writes while freezing the layer. This is enforced at a higher level via
-    /// [`crate::tenant::Timeline::write_lock`]. Freeze might be called via two code paths:
-    /// 1. Via the active [`crate::tenant::timeline::TimelineWriter`]. This holds the
-    ///    Timeline::write_lock for its lifetime. The rolling is handled in
-    ///    [`crate::tenant::timeline::TimelineWriter::put_batch`]. It's a &mut self function
-    ///    so can't be called from different threads.
-    /// 2. In the background via [`crate::tenant::Timeline::maybe_freeze_ephemeral_layer`].
-    ///    This only proceeds if try_lock on Timeline::write_lock succeeds (i.e. there's no active writer),
-    ///    hence there can be no concurrent writes
    pub async fn freeze(&self, end_lsn: Lsn) {
        assert!(
            self.start_lsn < end_lsn,
@@ -731,8 +700,8 @@ impl InMemoryLayer {

        #[cfg(debug_assertions)]
        {
-            let index = self.index.read().await;
-            for vec_map in index.values() {
+            let inner = self.inner.write().await;
+            for vec_map in inner.index.values() {
                for (lsn, _) in vec_map.as_slice() {
                    assert!(*lsn < end_lsn);
                }
@@ -755,11 +724,14 @@ impl InMemoryLayer {
    ) -> Result<Option<(PersistentLayerDesc, Utf8PathBuf)>> {
        // Grab the lock in read-mode. We hold it over the I/O, but because this
        // layer is not writeable anymore, no one should be trying to acquire the
-        // write lock on it, so we shouldn't block anyone. See the comment on
-        // [`InMemoryLayer::freeze`] to understand how locking between the append path
-        // and layer flushing works.
+        // write lock on it, so we shouldn't block anyone. There's one exception
+        // though: another thread might have grabbed a reference to this layer
+        // in `get_layer_for_write' just before the checkpointer called
+        // `freeze`, and then `write_to_disk` on it. When the thread gets the
+        // lock, it will see that it's not writeable anymore and retry, but it
+        // would have to wait until we release it. That race condition is very
+        // rare though, so we just accept the potential latency hit for now.
        let inner = self.inner.read().await;
-        let index = self.index.read().await;

        use l0_flush::Inner;
        let _concurrency_permit = match l0_flush_global_state {
@@ -771,9 +743,13 @@ impl InMemoryLayer {
        let key_count = if let Some(key_range) = key_range {
            let key_range = key_range.start.to_compact()..key_range.end.to_compact();

-            index.iter().filter(|(k, _)| key_range.contains(k)).count()
+            inner
+                .index
+                .iter()
+                .filter(|(k, _)| key_range.contains(k))
+                .count()
        } else {
-            index.len()
+            inner.index.len()
        };
        if key_count == 0 {
            return Ok(None);
@@ -796,7 +772,7 @@ impl InMemoryLayer {
                let file_contents = inner.file.load_to_io_buf(ctx).await?;
                let file_contents = file_contents.freeze();

-                for (key, vec_map) in index.iter() {
+                for (key, vec_map) in inner.index.iter() {
                    // Write all page versions
                    for (lsn, entry) in vec_map
                        .as_slice()
--- a/pageserver/src/tenant/timeline.rs
+++ b/pageserver/src/tenant/timeline.rs
@@ -14,7 +14,6 @@ pub mod span;
 pub mod uninit;
 mod walreceiver;

-use hashlink::LruCache;
 use std::array;
 use std::cmp::{max, min};
 use std::collections::btree_map::Entry;
@@ -24,6 +23,8 @@ use std::sync::atomic::{AtomicBool, AtomicU64, Ordering as AtomicOrdering};
 use std::sync::{Arc, Mutex, OnceLock, RwLock, Weak};
 use std::time::{Duration, Instant, SystemTime};

+use crate::PERF_TRACE_TARGET;
+use crate::walredo::RedoAttemptType;
 use anyhow::{Context, Result, anyhow, bail, ensure};
 use arc_swap::{ArcSwap, ArcSwapOption};
 use bytes::Bytes;
@@ -92,12 +93,10 @@ use super::storage_layer::{LayerFringe, LayerVisibilityHint, ReadableLayer};
 use super::tasks::log_compaction_error;
 use super::upload_queue::NotInitialized;
 use super::{
-    AttachedTenantConf, BasebackupPrepareSender, GcError, HeatMapTimeline, MaybeOffloaded,
+    AttachedTenantConf, GcError, HeatMapTimeline, MaybeOffloaded,
    debug_assert_current_span_has_tenant_and_timeline_id,
 };
-use crate::PERF_TRACE_TARGET;
 use crate::aux_file::AuxFileSizeEstimator;
-use crate::basebackup_cache::BasebackupPrepareRequest;
 use crate::config::PageServerConf;
 use crate::context::{
    DownloadBehavior, PerfInstrumentFutureExt, RequestContext, RequestContextBuilder,
@@ -131,7 +130,6 @@ use crate::tenant::tasks::BackgroundLoopKind;
 use crate::tenant::timeline::logical_size::CurrentLogicalSize;
 use crate::virtual_file::{MaybeFatalIo, VirtualFile};
 use crate::walingest::WalLagCooldown;
-use crate::walredo::RedoAttemptType;
 use crate::{ZERO_PAGE, task_mgr, walredo};

 #[derive(Debug, PartialEq, Eq, Clone, Copy)]
@@ -197,7 +195,16 @@ pub struct TimelineResources {
    pub pagestream_throttle_metrics: Arc<crate::metrics::tenant_throttling::Pagestream>,
    pub l0_compaction_trigger: Arc<Notify>,
    pub l0_flush_global_state: l0_flush::L0FlushGlobalState,
-    pub basebackup_prepare_sender: BasebackupPrepareSender,
+}
+
+/// The relation size cache caches relation sizes at the end of the timeline. It speeds up WAL
+/// ingestion considerably, because WAL ingestion needs to check on most records if the record
+/// implicitly extends the relation.  At startup, `complete_as_of` is initialized to the current end
+/// of the timeline (disk_consistent_lsn).  It's used on reads of relation sizes to check if the
+/// value can be used to also update the cache, see [`Timeline::update_cached_rel_size`].
+pub(crate) struct RelSizeCache {
+    pub(crate) complete_as_of: Lsn,
+    pub(crate) map: HashMap<RelTag, (Lsn, BlockNumber)>,
 }

 pub struct Timeline {
@@ -358,8 +365,7 @@ pub struct Timeline {
    pub walreceiver: Mutex<Option<WalReceiver>>,

    /// Relation size cache
-    pub(crate) rel_size_latest_cache: RwLock<HashMap<RelTag, (Lsn, BlockNumber)>>,
-    pub(crate) rel_size_snapshot_cache: Mutex<LruCache<(Lsn, RelTag), BlockNumber>>,
+    pub(crate) rel_size_cache: RwLock<RelSizeCache>,

    download_all_remote_layers_task_info: RwLock<Option<DownloadRemoteLayersTaskInfo>>,

@@ -441,9 +447,6 @@ pub struct Timeline {
    pub(crate) rel_size_v2_status: ArcSwapOption<RelSizeMigration>,

    wait_lsn_log_slow: tokio::sync::Semaphore,
-
-    /// A channel to send async requests to prepare a basebackup for the basebackup cache.
-    basebackup_prepare_sender: BasebackupPrepareSender,
 }

 pub(crate) enum PreviousHeatmap {
@@ -534,24 +537,29 @@ impl GcInfo {
 /// The `GcInfo` component describing which Lsns need to be retained.  Functionally, this
 /// is a single number (the oldest LSN which we must retain), but it internally distinguishes
 /// between time-based and space-based retention for observability and consumption metrics purposes.
-#[derive(Clone, Debug, Default)]
+#[derive(Debug, Clone)]
 pub(crate) struct GcCutoffs {
    /// Calculated from the [`pageserver_api::models::TenantConfig::gc_horizon`], this LSN indicates how much
    /// history we must keep to retain a specified number of bytes of WAL.
    pub(crate) space: Lsn,

-    /// Calculated from [`pageserver_api::models::TenantConfig::pitr_interval`], this LSN indicates
-    /// how much history we must keep to enable reading back at least the PITR interval duration.
-    ///
-    /// None indicates that the PITR cutoff has not been computed. A PITR interval of 0 will yield
-    /// Some(last_record_lsn).
-    pub(crate) time: Option<Lsn>,
+    /// Calculated from [`pageserver_api::models::TenantConfig::pitr_interval`], this LSN indicates how much
+    /// history we must keep to enable reading back at least the PITR interval duration.
+    pub(crate) time: Lsn,
+}
+
+impl Default for GcCutoffs {
+    fn default() -> Self {
+        Self {
+            space: Lsn::INVALID,
+            time: Lsn::INVALID,
+        }
+    }
 }

 impl GcCutoffs {
    fn select_min(&self) -> Lsn {
-        // NB: if we haven't computed the PITR cutoff yet, we can't GC anything.
-        self.space.min(self.time.unwrap_or_default())
+        std::cmp::min(self.space, self.time)
    }
 }

@@ -1033,7 +1041,6 @@ pub(crate) enum WaitLsnWaiter<'a> {
    Tenant,
    PageService,
    HttpEndpoint,
-    BaseBackupCache,
 }

 /// Argument to [`Timeline::shutdown`].
@@ -1089,14 +1096,11 @@ impl Timeline {
    /// Get the bytes written since the PITR cutoff on this branch, and
    /// whether this branch's ancestor_lsn is within its parent's PITR.
    pub(crate) fn get_pitr_history_stats(&self) -> (u64, bool) {
-        // TODO: for backwards compatibility, we return the full history back to 0 when the PITR
-        // cutoff has not yet been initialized. This should return None instead, but this is exposed
-        // in external HTTP APIs and callers may not handle a null value.
        let gc_info = self.gc_info.read().unwrap();
        let history = self
            .get_last_record_lsn()
-            .checked_sub(gc_info.cutoffs.time.unwrap_or_default())
-            .unwrap_or_default()
+            .checked_sub(gc_info.cutoffs.time)
+            .unwrap_or(Lsn(0))
            .0;
        (history, gc_info.within_ancestor_pitr)
    }
@@ -1106,10 +1110,9 @@ impl Timeline {
        self.applied_gc_cutoff_lsn.read()
    }

-    /// Read timeline's planned GC cutoff: this is the logical end of history that users are allowed
-    /// to read (based on configured PITR), even if physically we have more history. Returns None
-    /// if the PITR cutoff has not yet been initialized.
-    pub(crate) fn get_gc_cutoff_lsn(&self) -> Option<Lsn> {
+    /// Read timeline's planned GC cutoff: this is the logical end of history that users
+    /// are allowed to read (based on configured PITR), even if physically we have more history.
+    pub(crate) fn get_gc_cutoff_lsn(&self) -> Lsn {
        self.gc_info.read().unwrap().cutoffs.time
    }

@@ -1560,8 +1563,7 @@ impl Timeline {
                        }
                        WaitLsnWaiter::Tenant
                        | WaitLsnWaiter::PageService
-                        | WaitLsnWaiter::HttpEndpoint
-                        | WaitLsnWaiter::BaseBackupCache => unreachable!(
+                        | WaitLsnWaiter::HttpEndpoint => unreachable!(
                            "tenant or page_service context are not expected to have task kind {:?}",
                            ctx.task_kind()
                        ),
@@ -2466,41 +2468,6 @@ impl Timeline {
            false
        }
    }
-
-    pub(crate) fn is_basebackup_cache_enabled(&self) -> bool {
-        let tenant_conf = self.tenant_conf.load();
-        tenant_conf
-            .tenant_conf
-            .basebackup_cache_enabled
-            .unwrap_or(self.conf.default_tenant_conf.basebackup_cache_enabled)
-    }
-
-    /// Prepare basebackup for the given LSN and store it in the basebackup cache.
-    /// The method is asynchronous and returns immediately.
-    /// The actual basebackup preparation is performed in the background
-    /// by the basebackup cache on a best-effort basis.
-    pub(crate) fn prepare_basebackup(&self, lsn: Lsn) {
-        if !self.is_basebackup_cache_enabled() {
-            return;
-        }
-        if !self.tenant_shard_id.is_shard_zero() {
-            // In theory we should never get here, but just in case check it.
-            // Preparing basebackup doesn't make sense for shards other than shard zero.
-            return;
-        }
-
-        let res = self
-            .basebackup_prepare_sender
-            .send(BasebackupPrepareRequest {
-                tenant_shard_id: self.tenant_shard_id,
-                timeline_id: self.timeline_id,
-                lsn,
-            });
-        if let Err(e) = res {
-            // May happen during shutdown, it's not critical.
-            info!("Failed to send shutdown checkpoint: {e:#}");
-        }
-    }
 }

 /// Number of times we will compute partition within a checkpoint distance.
@@ -2578,13 +2545,6 @@ impl Timeline {
            .unwrap_or(self.conf.default_tenant_conf.checkpoint_timeout)
    }

-    pub(crate) fn get_pitr_interval(&self) -> Duration {
-        let tenant_conf = &self.tenant_conf.load().tenant_conf;
-        tenant_conf
-            .pitr_interval
-            .unwrap_or(self.conf.default_tenant_conf.pitr_interval)
-    }
-
    fn get_compaction_period(&self) -> Duration {
        let tenant_conf = self.tenant_conf.load().tenant_conf.clone();
        tenant_conf
@@ -2860,13 +2820,6 @@ impl Timeline {

            self.remote_client.update_config(&new_conf.location);

-            let mut rel_size_cache = self.rel_size_snapshot_cache.lock().unwrap();
-            if let Some(new_capacity) = new_conf.tenant_conf.relsize_snapshot_cache_capacity {
-                if new_capacity != rel_size_cache.capacity() {
-                    rel_size_cache.set_capacity(new_capacity);
-                }
-            }
-
            self.metrics
                .evictions_with_low_residence_duration
                .write()
@@ -2925,14 +2878,6 @@ impl Timeline {
            ancestor_gc_info.insert_child(timeline_id, metadata.ancestor_lsn(), is_offloaded);
        }

-        let relsize_snapshot_cache_capacity = {
-            let loaded_tenant_conf = tenant_conf.load();
-            loaded_tenant_conf
-                .tenant_conf
-                .relsize_snapshot_cache_capacity
-                .unwrap_or(conf.default_tenant_conf.relsize_snapshot_cache_capacity)
-        };
-
        Arc::new_cyclic(|myself| {
            let metrics = Arc::new(TimelineMetrics::new(
                &tenant_shard_id,
@@ -3024,8 +2969,10 @@ impl Timeline {
                last_image_layer_creation_check_instant: Mutex::new(None),

                last_received_wal: Mutex::new(None),
-                rel_size_latest_cache: RwLock::new(HashMap::new()),
-                rel_size_snapshot_cache: Mutex::new(LruCache::new(relsize_snapshot_cache_capacity)),
+                rel_size_cache: RwLock::new(RelSizeCache {
+                    complete_as_of: disk_consistent_lsn,
+                    map: HashMap::new(),
+                }),

                download_all_remote_layers_task_info: RwLock::new(None),

@@ -3070,8 +3017,6 @@ impl Timeline {
                rel_size_v2_status: ArcSwapOption::from_pointee(rel_size_v2_status),

                wait_lsn_log_slow: tokio::sync::Semaphore::new(1),
-
-                basebackup_prepare_sender: resources.basebackup_prepare_sender,
            };

            result.repartition_threshold =
@@ -3585,7 +3530,7 @@ impl Timeline {
                };

                let io_concurrency = IoConcurrency::spawn_from_conf(
-                    self_ref.conf.get_vectored_concurrent_io,
+                    self_ref.conf,
                    self_ref
                        .gate
                        .enter()
@@ -5614,7 +5559,7 @@ impl Timeline {
            });

            let io_concurrency = IoConcurrency::spawn_from_conf(
-                self.conf.get_vectored_concurrent_io,
+                self.conf,
                self.gate
                    .enter()
                    .map_err(|_| CreateImageLayersError::Cancelled)?,
@@ -6285,12 +6230,14 @@ impl Timeline {

        pausable_failpoint!("Timeline::find_gc_cutoffs-pausable");

-        if cfg!(test) && pitr == Duration::ZERO {
+        if cfg!(test) {
            // Unit tests which specify zero PITR interval expect to avoid doing any I/O for timestamp lookup
-            return Ok(GcCutoffs {
-                time: Some(self.get_last_record_lsn()),
-                space: space_cutoff,
-            });
+            if pitr == Duration::ZERO {
+                return Ok(GcCutoffs {
+                    time: self.get_last_record_lsn(),
+                    space: space_cutoff,
+                });
+            }
        }

        // Calculate a time-based limit on how much to retain:
@@ -6304,14 +6251,14 @@ impl Timeline {
                // PITR is not set. Retain the size-based limit, or the default time retention,
                // whichever requires less data.
                GcCutoffs {
-                    time: Some(self.get_last_record_lsn()),
+                    time: self.get_last_record_lsn(),
                    space: std::cmp::max(time_cutoff, space_cutoff),
                }
            }
            (Duration::ZERO, None) => {
                // PITR is not set, and time lookup failed
                GcCutoffs {
-                    time: Some(self.get_last_record_lsn()),
+                    time: self.get_last_record_lsn(),
                    space: space_cutoff,
                }
            }
@@ -6319,7 +6266,7 @@ impl Timeline {
                // PITR interval is set & we didn't look up a timestamp successfully.  Conservatively assume PITR
                // cannot advance beyond what was already GC'd, and respect space-based retention
                GcCutoffs {
-                    time: Some(*self.get_applied_gc_cutoff_lsn()),
+                    time: *self.get_applied_gc_cutoff_lsn(),
                    space: space_cutoff,
                }
            }
@@ -6327,7 +6274,7 @@ impl Timeline {
                // PITR interval is set and we looked up timestamp successfully.  Ignore
                // size based retention and make time cutoff authoritative
                GcCutoffs {
-                    time: Some(time_cutoff),
+                    time: time_cutoff,
                    space: time_cutoff,
                }
            }
@@ -6380,7 +6327,7 @@ impl Timeline {
            )
        };

-        let mut new_gc_cutoff = space_cutoff.min(time_cutoff.unwrap_or_default());
+        let mut new_gc_cutoff = Lsn::min(space_cutoff, time_cutoff);
        let standby_horizon = self.standby_horizon.load();
        // Hold GC for the standby, but as a safety guard do it only within some
        // reasonable lag.
@@ -6429,7 +6376,7 @@ impl Timeline {
    async fn gc_timeline(
        &self,
        space_cutoff: Lsn,
-        time_cutoff: Option<Lsn>, // None if uninitialized
+        time_cutoff: Lsn,
        retain_lsns: Vec<Lsn>,
        max_lsn_with_valid_lease: Option<Lsn>,
        new_gc_cutoff: Lsn,
@@ -6448,12 +6395,6 @@ impl Timeline {
            return Ok(result);
        }

-        let Some(time_cutoff) = time_cutoff else {
-            // The GC cutoff should have been computed by now, but let's be defensive.
-            info!("Nothing to GC: time_cutoff not yet computed");
-            return Ok(result);
-        };
-
        // We need to ensure that no one tries to read page versions or create
        // branches at a point before latest_gc_cutoff_lsn. See branch_timeline()
        // for details. This will block until the old value is no longer in use.
--- a/pageserver/src/tenant/timeline/compaction.rs
+++ b/pageserver/src/tenant/timeline/compaction.rs
@@ -1526,7 +1526,7 @@ impl Timeline {
        info!(
            "starting shard ancestor compaction, rewriting {} layers and dropping {} layers, \
                checked {layers_checked}/{layers_total} layers \
-                (latest_gc_cutoff={} pitr_cutoff={:?})",
+                (latest_gc_cutoff={} pitr_cutoff={})",
            layers_to_rewrite.len(),
            drop_layers.len(),
            *latest_gc_cutoff,
--- a/pageserver/src/tenant/timeline/detach_ancestor.rs
+++ b/pageserver/src/tenant/timeline/detach_ancestor.rs
@@ -188,7 +188,7 @@ pub(crate) async fn generate_tombstone_image_layer(
        "removing non-inherited keys by writing an image layer with tombstones at the detach LSN"
    );
    let io_concurrency = IoConcurrency::spawn_from_conf(
-        detached.conf.get_vectored_concurrent_io,
+        detached.conf,
        detached.gate.enter().map_err(|_| Error::ShuttingDown)?,
    );
    let mut reconstruct_state = ValuesReconstructState::new(io_concurrency);
--- a/pageserver/src/walingest.rs
+++ b/pageserver/src/walingest.rs
@@ -1316,10 +1316,6 @@ impl WalIngest {
            }
        });

-        if info == pg_constants::XLOG_CHECKPOINT_SHUTDOWN {
-            modification.tline.prepare_basebackup(lsn);
-        }
-
        Ok(())
    }

@@ -1688,31 +1684,31 @@ mod tests {
        // The relation was created at LSN 2, not visible at LSN 1 yet.
        assert_eq!(
            tline
-                .get_rel_exists(TESTREL_A, Version::at(Lsn(0x10)), &ctx)
+                .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x10)), &ctx)
                .await?,
            false
        );
        assert!(
            tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x10)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x10)), &ctx)
                .await
                .is_err()
        );
        assert_eq!(
            tline
-                .get_rel_exists(TESTREL_A, Version::at(Lsn(0x20)), &ctx)
+                .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
                .await?,
            true
        );
        assert_eq!(
            tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x20)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
                .await?,
            1
        );
        assert_eq!(
            tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x50)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x50)), &ctx)
                .await?,
            3
        );
@@ -1723,7 +1719,7 @@ mod tests {
                .get_rel_page_at_lsn(
                    TESTREL_A,
                    0,
-                    Version::at(Lsn(0x20)),
+                    Version::Lsn(Lsn(0x20)),
                    &ctx,
                    io_concurrency.clone()
                )
@@ -1737,7 +1733,7 @@ mod tests {
                .get_rel_page_at_lsn(
                    TESTREL_A,
                    0,
-                    Version::at(Lsn(0x30)),
+                    Version::Lsn(Lsn(0x30)),
                    &ctx,
                    io_concurrency.clone()
                )
@@ -1751,7 +1747,7 @@ mod tests {
                .get_rel_page_at_lsn(
                    TESTREL_A,
                    0,
-                    Version::at(Lsn(0x40)),
+                    Version::Lsn(Lsn(0x40)),
                    &ctx,
                    io_concurrency.clone()
                )
@@ -1764,7 +1760,7 @@ mod tests {
                .get_rel_page_at_lsn(
                    TESTREL_A,
                    1,
-                    Version::at(Lsn(0x40)),
+                    Version::Lsn(Lsn(0x40)),
                    &ctx,
                    io_concurrency.clone()
                )
@@ -1778,7 +1774,7 @@ mod tests {
                .get_rel_page_at_lsn(
                    TESTREL_A,
                    0,
-                    Version::at(Lsn(0x50)),
+                    Version::Lsn(Lsn(0x50)),
                    &ctx,
                    io_concurrency.clone()
                )
@@ -1791,7 +1787,7 @@ mod tests {
                .get_rel_page_at_lsn(
                    TESTREL_A,
                    1,
-                    Version::at(Lsn(0x50)),
+                    Version::Lsn(Lsn(0x50)),
                    &ctx,
                    io_concurrency.clone()
                )
@@ -1804,7 +1800,7 @@ mod tests {
                .get_rel_page_at_lsn(
                    TESTREL_A,
                    2,
-                    Version::at(Lsn(0x50)),
+                    Version::Lsn(Lsn(0x50)),
                    &ctx,
                    io_concurrency.clone()
                )
@@ -1824,7 +1820,7 @@ mod tests {
        // Check reported size and contents after truncation
        assert_eq!(
            tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x60)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x60)), &ctx)
                .await?,
            2
        );
@@ -1833,7 +1829,7 @@ mod tests {
                .get_rel_page_at_lsn(
                    TESTREL_A,
                    0,
-                    Version::at(Lsn(0x60)),
+                    Version::Lsn(Lsn(0x60)),
                    &ctx,
                    io_concurrency.clone()
                )
@@ -1846,7 +1842,7 @@ mod tests {
                .get_rel_page_at_lsn(
                    TESTREL_A,
                    1,
-                    Version::at(Lsn(0x60)),
+                    Version::Lsn(Lsn(0x60)),
                    &ctx,
                    io_concurrency.clone()
                )
@@ -1858,7 +1854,7 @@ mod tests {
        // should still see the truncated block with older LSN
        assert_eq!(
            tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x50)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x50)), &ctx)
                .await?,
            3
        );
@@ -1867,7 +1863,7 @@ mod tests {
                .get_rel_page_at_lsn(
                    TESTREL_A,
                    2,
-                    Version::at(Lsn(0x50)),
+                    Version::Lsn(Lsn(0x50)),
                    &ctx,
                    io_concurrency.clone()
                )
@@ -1884,7 +1880,7 @@ mod tests {
        m.commit(&ctx).await?;
        assert_eq!(
            tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x68)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x68)), &ctx)
                .await?,
            0
        );
@@ -1897,7 +1893,7 @@ mod tests {
        m.commit(&ctx).await?;
        assert_eq!(
            tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x70)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x70)), &ctx)
                .await?,
            2
        );
@@ -1906,7 +1902,7 @@ mod tests {
                .get_rel_page_at_lsn(
                    TESTREL_A,
                    0,
-                    Version::at(Lsn(0x70)),
+                    Version::Lsn(Lsn(0x70)),
                    &ctx,
                    io_concurrency.clone()
                )
@@ -1919,7 +1915,7 @@ mod tests {
                .get_rel_page_at_lsn(
                    TESTREL_A,
                    1,
-                    Version::at(Lsn(0x70)),
+                    Version::Lsn(Lsn(0x70)),
                    &ctx,
                    io_concurrency.clone()
                )
@@ -1936,7 +1932,7 @@ mod tests {
        m.commit(&ctx).await?;
        assert_eq!(
            tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x80)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x80)), &ctx)
                .await?,
            1501
        );
@@ -1946,7 +1942,7 @@ mod tests {
                    .get_rel_page_at_lsn(
                        TESTREL_A,
                        blk,
-                        Version::at(Lsn(0x80)),
+                        Version::Lsn(Lsn(0x80)),
                        &ctx,
                        io_concurrency.clone()
                    )
@@ -1960,7 +1956,7 @@ mod tests {
                .get_rel_page_at_lsn(
                    TESTREL_A,
                    1500,
-                    Version::at(Lsn(0x80)),
+                    Version::Lsn(Lsn(0x80)),
                    &ctx,
                    io_concurrency.clone()
                )
@@ -1994,13 +1990,13 @@ mod tests {
        // Check that rel exists and size is correct
        assert_eq!(
            tline
-                .get_rel_exists(TESTREL_A, Version::at(Lsn(0x20)), &ctx)
+                .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
                .await?,
            true
        );
        assert_eq!(
            tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x20)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
                .await?,
            1
        );
@@ -2015,7 +2011,7 @@ mod tests {
        // Check that rel is not visible anymore
        assert_eq!(
            tline
-                .get_rel_exists(TESTREL_A, Version::at(Lsn(0x30)), &ctx)
+                .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x30)), &ctx)
                .await?,
            false
        );
@@ -2033,13 +2029,13 @@ mod tests {
        // Check that rel exists and size is correct
        assert_eq!(
            tline
-                .get_rel_exists(TESTREL_A, Version::at(Lsn(0x40)), &ctx)
+                .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x40)), &ctx)
                .await?,
            true
        );
        assert_eq!(
            tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x40)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x40)), &ctx)
                .await?,
            1
        );
@@ -2081,26 +2077,26 @@ mod tests {
        // The relation was created at LSN 20, not visible at LSN 1 yet.
        assert_eq!(
            tline
-                .get_rel_exists(TESTREL_A, Version::at(Lsn(0x10)), &ctx)
+                .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x10)), &ctx)
                .await?,
            false
        );
        assert!(
            tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x10)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x10)), &ctx)
                .await
                .is_err()
        );

        assert_eq!(
            tline
-                .get_rel_exists(TESTREL_A, Version::at(Lsn(0x20)), &ctx)
+                .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
                .await?,
            true
        );
        assert_eq!(
            tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x20)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
                .await?,
            relsize
        );
@@ -2114,7 +2110,7 @@ mod tests {
                    .get_rel_page_at_lsn(
                        TESTREL_A,
                        blkno,
-                        Version::at(lsn),
+                        Version::Lsn(lsn),
                        &ctx,
                        io_concurrency.clone()
                    )
@@ -2135,7 +2131,7 @@ mod tests {
        // Check reported size and contents after truncation
        assert_eq!(
            tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x60)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x60)), &ctx)
                .await?,
            1
        );
@@ -2148,7 +2144,7 @@ mod tests {
                    .get_rel_page_at_lsn(
                        TESTREL_A,
                        blkno,
-                        Version::at(Lsn(0x60)),
+                        Version::Lsn(Lsn(0x60)),
                        &ctx,
                        io_concurrency.clone()
                    )
@@ -2161,7 +2157,7 @@ mod tests {
        // should still see all blocks with older LSN
        assert_eq!(
            tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x50)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x50)), &ctx)
                .await?,
            relsize
        );
@@ -2173,7 +2169,7 @@ mod tests {
                    .get_rel_page_at_lsn(
                        TESTREL_A,
                        blkno,
-                        Version::at(Lsn(0x50)),
+                        Version::Lsn(Lsn(0x50)),
                        &ctx,
                        io_concurrency.clone()
                    )
@@ -2197,13 +2193,13 @@ mod tests {

        assert_eq!(
            tline
-                .get_rel_exists(TESTREL_A, Version::at(Lsn(0x80)), &ctx)
+                .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x80)), &ctx)
                .await?,
            true
        );
        assert_eq!(
            tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x80)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x80)), &ctx)
                .await?,
            relsize
        );
@@ -2216,7 +2212,7 @@ mod tests {
                    .get_rel_page_at_lsn(
                        TESTREL_A,
                        blkno,
-                        Version::at(Lsn(0x80)),
+                        Version::Lsn(Lsn(0x80)),
                        &ctx,
                        io_concurrency.clone()
                    )
@@ -2254,7 +2250,7 @@ mod tests {

        assert_eq!(
            tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(lsn)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(lsn)), &ctx)
                .await?,
            RELSEG_SIZE + 1
        );
@@ -2268,7 +2264,7 @@ mod tests {
        m.commit(&ctx).await?;
        assert_eq!(
            tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(lsn)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(lsn)), &ctx)
                .await?,
            RELSEG_SIZE
        );
@@ -2283,7 +2279,7 @@ mod tests {
        m.commit(&ctx).await?;
        assert_eq!(
            tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(lsn)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(lsn)), &ctx)
                .await?,
            RELSEG_SIZE - 1
        );
@@ -2301,7 +2297,7 @@ mod tests {
            m.commit(&ctx).await?;
            assert_eq!(
                tline
-                    .get_rel_size(TESTREL_A, Version::at(Lsn(lsn)), &ctx)
+                    .get_rel_size(TESTREL_A, Version::Lsn(Lsn(lsn)), &ctx)
                    .await?,
                size as BlockNumber
            );
--- a/pgxn/neon/communicator.c
+++ b/pgxn/neon/communicator.c
@@ -717,7 +717,7 @@ prefetch_read(PrefetchRequest *slot)
 	Assert(slot->status == PRFS_REQUESTED);
 	Assert(slot->response == NULL);
 	Assert(slot->my_ring_index == MyPState->ring_receive);
-	Assert(readpage_reentrant_guard || AmPrewarmWorker);
+	Assert(readpage_reentrant_guard);

 	if (slot->status != PRFS_REQUESTED ||
 		slot->response != NULL ||
@@ -800,7 +800,7 @@ communicator_prefetch_receive(BufferTag tag)
 	PrfHashEntry *entry;
 	PrefetchRequest hashkey;

-	Assert(readpage_reentrant_guard || AmPrewarmWorker); /* do not pump prefetch state in prewarm worker */
+	Assert(readpage_reentrant_guard);
 	hashkey.buftag = tag;
 	entry = prfh_lookup(MyPState->prf_hash, &hashkey);
 	if (entry != NULL && prefetch_wait_for(entry->slot->my_ring_index))
@@ -2450,7 +2450,6 @@ void
 communicator_reconfigure_timeout_if_needed(void)
 {
 	bool	needs_set = MyPState->ring_receive != MyPState->ring_unused &&
-						!AmPrewarmWorker && /* do not pump prefetch state in prewarm worker */
 						readahead_getpage_pull_timeout_ms > 0;

 	if (needs_set != timeout_set)
--- a/pgxn/neon/file_cache.c
+++ b/pgxn/neon/file_cache.c
@@ -201,8 +201,6 @@ static shmem_request_hook_type prev_shmem_request_hook;
 bool lfc_store_prefetch_result;
 bool lfc_prewarm_update_ws_estimation;

-bool AmPrewarmWorker;
-
 #define LFC_ENABLED() (lfc_ctl->limit != 0)

 /*
@@ -847,8 +845,6 @@ lfc_prewarm_main(Datum main_arg)
 	PrewarmWorkerState* ws;
 	uint32 worker_id = DatumGetInt32(main_arg);

-	AmPrewarmWorker = true;
-
 	pqsignal(SIGTERM, die);
 	BackgroundWorkerUnblockSignals();

--- a/pgxn/neon/neon.h
+++ b/pgxn/neon/neon.h
@@ -23,8 +23,6 @@ extern int	wal_acceptor_connection_timeout;
 extern int	readahead_getpage_pull_timeout_ms;
 extern bool	disable_wal_prev_lsn_checks;

-extern bool AmPrewarmWorker;
-
 #if PG_MAJORVERSION_NUM >= 17
 extern uint32		WAIT_EVENT_NEON_LFC_MAINTENANCE;
 extern uint32		WAIT_EVENT_NEON_LFC_READ;
--- a/poetry.lock
+++ b/poetry.lock
@@ -3170,24 +3170,19 @@ pbr = "*"

 [[package]]
 name = "setuptools"
-version = "78.1.1"
+version = "70.0.0"
 description = "Easily download, build, install, upgrade, and uninstall Python packages"
 optional = false
-python-versions = ">=3.9"
+python-versions = ">=3.8"
 groups = ["main"]
 files = [
-    {file = "setuptools-78.1.1-py3-none-any.whl", hash = "sha256:c3a9c4211ff4c309edb8b8c4f1cbfa7ae324c4ba9f91ff254e3d305b9fd54561"},
-    {file = "setuptools-78.1.1.tar.gz", hash = "sha256:fcc17fd9cd898242f6b4adfaca46137a9edef687f43e6f78469692a5e70d851d"},
+    {file = "setuptools-70.0.0-py3-none-any.whl", hash = "sha256:54faa7f2e8d2d11bcd2c07bed282eef1046b5c080d1c32add737d7b5817b1ad4"},
+    {file = "setuptools-70.0.0.tar.gz", hash = "sha256:f211a66637b8fa059bb28183da127d4e86396c991a942b028c6650d4319c3fd0"},
 ]

 [package.extras]
-check = ["pytest-checkdocs (>=2.4)", "pytest-ruff (>=0.2.1) ; sys_platform != \"cygwin\"", "ruff (>=0.8.0) ; sys_platform != \"cygwin\""]
-core = ["importlib_metadata (>=6) ; python_version < \"3.10\"", "jaraco.functools (>=4)", "jaraco.text (>=3.7)", "more_itertools", "more_itertools (>=8.8)", "packaging (>=24.2)", "platformdirs (>=4.2.2)", "tomli (>=2.0.1) ; python_version < \"3.11\"", "wheel (>=0.43.0)"]
-cover = ["pytest-cov"]
-doc = ["furo", "jaraco.packaging (>=9.3)", "jaraco.tidelift (>=1.4)", "pygments-github-lexers (==0.0.5)", "pyproject-hooks (!=1.1)", "rst.linker (>=1.9)", "sphinx (>=3.5)", "sphinx-favicon", "sphinx-inline-tabs", "sphinx-lint", "sphinx-notfound-page (>=1,<2)", "sphinx-reredirects", "sphinxcontrib-towncrier", "towncrier (<24.7)"]
-enabler = ["pytest-enabler (>=2.2)"]
-test = ["build[virtualenv] (>=1.0.3)", "filelock (>=3.4.0)", "ini2toml[lite] (>=0.14)", "jaraco.develop (>=7.21) ; python_version >= \"3.9\" and sys_platform != \"cygwin\"", "jaraco.envs (>=2.2)", "jaraco.path (>=3.7.2)", "jaraco.test (>=5.5)", "packaging (>=24.2)", "pip (>=19.1)", "pyproject-hooks (!=1.1)", "pytest (>=6,!=8.1.*)", "pytest-home (>=0.5)", "pytest-perf ; sys_platform != \"cygwin\"", "pytest-subprocess", "pytest-timeout", "pytest-xdist (>=3)", "tomli-w (>=1.0.0)", "virtualenv (>=13.0.0)", "wheel (>=0.44.0)"]
-type = ["importlib_metadata (>=7.0.2) ; python_version < \"3.10\"", "jaraco.develop (>=7.21) ; sys_platform != \"cygwin\"", "mypy (==1.14.*)", "pytest-mypy"]
+docs = ["furo", "jaraco.packaging (>=9.3)", "jaraco.tidelift (>=1.4)", "pygments-github-lexers (==0.0.5)", "pyproject-hooks (!=1.1)", "rst.linker (>=1.9)", "sphinx (>=3.5)", "sphinx-favicon", "sphinx-inline-tabs", "sphinx-lint", "sphinx-notfound-page (>=1,<2)", "sphinx-reredirects", "sphinxcontrib-towncrier"]
+testing = ["build[virtualenv] (>=1.0.3)", "filelock (>=3.4.0)", "importlib-metadata", "ini2toml[lite] (>=0.14)", "jaraco.develop (>=7.21) ; python_version >= \"3.9\" and sys_platform != \"cygwin\"", "jaraco.envs (>=2.2)", "jaraco.path (>=3.2.0)", "mypy (==1.9)", "packaging (>=23.2)", "pip (>=19.1)", "pyproject-hooks (!=1.1)", "pytest (>=6,!=8.1.1)", "pytest-checkdocs (>=2.4)", "pytest-cov ; platform_python_implementation != \"PyPy\"", "pytest-enabler (>=2.2)", "pytest-home (>=0.5)", "pytest-mypy", "pytest-perf ; sys_platform != \"cygwin\"", "pytest-ruff (>=0.2.1) ; sys_platform != \"cygwin\"", "pytest-subprocess", "pytest-timeout", "pytest-xdist (>=3)", "tomli", "tomli-w (>=1.0.0)", "virtualenv (>=13.0.0)", "wheel"]

 [[package]]
 name = "six"
--- a/proxy/Cargo.toml
+++ b/proxy/Cargo.toml
@@ -127,4 +127,3 @@ rstest.workspace = true
 walkdir.workspace = true
 rand_distr = "0.4"
 tokio-postgres.workspace = true
-tracing-test = "0.2"
--- a/proxy/src/auth/backend/mod.rs
+++ b/proxy/src/auth/backend/mod.rs
@@ -80,22 +80,10 @@ impl std::fmt::Display for Backend<'_, ()> {
                    .field(&endpoint.url())
                    .finish(),
                #[cfg(any(test, feature = "testing"))]
-                ControlPlaneClient::PostgresMock(endpoint) => {
-                    let url = endpoint.url();
-                    match url::Url::parse(url) {
-                        Ok(mut url) => {
-                            let _ = url.set_password(Some("_redacted_"));
-                            let url = url.as_str();
-                            fmt.debug_tuple("ControlPlane::PostgresMock")
-                                .field(&url)
-                                .finish()
-                        }
-                        Err(_) => fmt
-                            .debug_tuple("ControlPlane::PostgresMock")
-                            .field(&url)
-                            .finish(),
-                    }
-                }
+                ControlPlaneClient::PostgresMock(endpoint) => fmt
+                    .debug_tuple("ControlPlane::PostgresMock")
+                    .field(&endpoint.url())
+                    .finish(),
                #[cfg(test)]
                ControlPlaneClient::Test(_) => fmt.debug_tuple("ControlPlane::Test").finish(),
            },
--- a/proxy/src/binary/proxy.rs
+++ b/proxy/src/binary/proxy.rs
@@ -1,13 +1,9 @@
-#[cfg(any(test, feature = "testing"))]
-use std::env;
 use std::net::SocketAddr;
 use std::path::PathBuf;
 use std::pin::pin;
 use std::sync::Arc;
 use std::time::Duration;

-#[cfg(any(test, feature = "testing"))]
-use anyhow::Context;
 use anyhow::{bail, ensure};
 use arc_swap::ArcSwapOption;
 use futures::future::Either;
@@ -39,8 +35,6 @@ use crate::scram::threadpool::ThreadPool;
 use crate::serverless::GlobalConnPoolOptions;
 use crate::serverless::cancel_set::CancelSet;
 use crate::tls::client_config::compute_client_config_with_root_certs;
-#[cfg(any(test, feature = "testing"))]
-use crate::url::ApiUrl;
 use crate::{auth, control_plane, http, serverless, usage_metrics};

 project_git_version!(GIT_VERSION);
@@ -167,11 +161,8 @@ struct ProxyCliArgs {
    #[clap(long, default_values_t = RateBucketInfo::DEFAULT_REDIS_SET)]
    redis_rps_limit: Vec<RateBucketInfo>,
    /// Cancellation channel size (max queue size for redis kv client)
-    #[clap(long, default_value_t = 1024)]
+    #[clap(long, default_value = "1024")]
    cancellation_ch_size: usize,
-    /// Cancellation ops batch size for redis
-    #[clap(long, default_value_t = 8)]
-    cancellation_batch_size: usize,
    /// cache for `allowed_ips` (use `size=0` to disable)
    #[clap(long, default_value = config::CacheOptions::CACHE_DEFAULT_OPTIONS)]
    allowed_ips_cache: String,
@@ -551,12 +542,7 @@ pub async fn run() -> anyhow::Result<()> {
            if let Some(mut redis_kv_client) = redis_kv_client {
                maintenance_tasks.spawn(async move {
                    redis_kv_client.try_connect().await?;
-                    handle_cancel_messages(
-                        &mut redis_kv_client,
-                        rx_cancel,
-                        args.cancellation_batch_size,
-                    )
-                    .await?;
+                    handle_cancel_messages(&mut redis_kv_client, rx_cancel).await?;

                    drop(redis_kv_client);

@@ -783,13 +769,7 @@ fn build_auth_backend(

        #[cfg(any(test, feature = "testing"))]
        AuthBackendType::Postgres => {
-            let mut url: ApiUrl = args.auth_endpoint.parse()?;
-            if url.password().is_none() {
-                let password = env::var("PGPASSWORD")
-                    .with_context(|| "auth-endpoint does not contain a password and environment variable `PGPASSWORD` is not set")?;
-                url.set_password(Some(&password))
-                    .expect("Failed to set password");
-            }
+            let url = args.auth_endpoint.parse()?;
            let api = control_plane::client::mock::MockControlPlane::new(
                url,
                !args.is_private_access_proxy,
--- a/proxy/src/cancellation.rs
+++ b/proxy/src/cancellation.rs
@@ -30,6 +30,8 @@ use crate::tls::postgres_rustls::MakeRustlsConnect;
 type IpSubnetKey = IpNet;

 const CANCEL_KEY_TTL: i64 = 1_209_600; // 2 weeks cancellation key expire time
+const REDIS_SEND_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(10);
+const BATCH_SIZE: usize = 8;

 // Message types for sending through mpsc channel
 pub enum CancelKeyOp {
@@ -229,13 +231,12 @@ impl CancelReplyOp {
 pub async fn handle_cancel_messages(
    client: &mut RedisKVClient,
    mut rx: mpsc::Receiver<CancelKeyOp>,
-    batch_size: usize,
 ) -> anyhow::Result<()> {
-    let mut batch = Vec::with_capacity(batch_size);
-    let mut pipeline = Pipeline::with_capacity(batch_size);
+    let mut batch = Vec::with_capacity(BATCH_SIZE);
+    let mut pipeline = Pipeline::with_capacity(BATCH_SIZE);

    loop {
-        if rx.recv_many(&mut batch, batch_size).await == 0 {
+        if rx.recv_many(&mut batch, BATCH_SIZE).await == 0 {
            warn!("shutting down cancellation queue");
            break Ok(());
        }
@@ -366,7 +367,8 @@ impl CancellationHandler {
            return Err(CancelError::InternalError);
        };

-        tx.try_send(op)
+        tx.send_timeout(op, REDIS_SEND_TIMEOUT)
+            .await
            .map_err(|e| {
                tracing::warn!("failed to send GetCancelData for {key}: {e}");
            })
@@ -568,7 +570,7 @@ impl Session {
    }

    // Send the store key op to the cancellation handler and set TTL for the key
-    pub(crate) fn write_cancel_key(
+    pub(crate) async fn write_cancel_key(
        &self,
        cancel_closure: CancelClosure,
    ) -> Result<(), CancelError> {
@@ -594,14 +596,14 @@ impl Session {
            expire: CANCEL_KEY_TTL,
        };

-        let _ = tx.try_send(op).map_err(|e| {
+        let _ = tx.send_timeout(op, REDIS_SEND_TIMEOUT).await.map_err(|e| {
            let key = self.key;
            tracing::warn!("failed to send StoreCancelKey for {key}: {e}");
        });
        Ok(())
    }

-    pub(crate) fn remove_cancel_key(&self) -> Result<(), CancelError> {
+    pub(crate) async fn remove_cancel_key(&self) -> Result<(), CancelError> {
        let Some(tx) = &self.cancellation_handler.tx else {
            tracing::warn!("cancellation handler is not available");
            return Err(CancelError::InternalError);
@@ -617,7 +619,7 @@ impl Session {
                .guard(RedisMsgKind::HDel),
        };

-        let _ = tx.try_send(op).map_err(|e| {
+        let _ = tx.send_timeout(op, REDIS_SEND_TIMEOUT).await.map_err(|e| {
            let key = self.key;
            tracing::warn!("failed to send RemoveCancelKey for {key}: {e}");
        });
--- a/proxy/src/console_redirect_proxy.rs
+++ b/proxy/src/console_redirect_proxy.rs
@@ -244,7 +244,9 @@ pub(crate) async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
    let cancellation_handler_clone = Arc::clone(&cancellation_handler);
    let session = cancellation_handler_clone.get_key();

-    session.write_cancel_key(node.cancel_closure.clone())?;
+    session
+        .write_cancel_key(node.cancel_closure.clone())
+        .await?;

    prepare_client_connection(&node, *session.key(), &mut stream).await?;

--- a/proxy/src/logging.rs
+++ b/proxy/src/logging.rs
@@ -1,11 +1,13 @@
-use std::cell::RefCell;
+use std::cell::{Cell, RefCell};
 use std::collections::HashMap;
-use std::sync::Arc;
+use std::hash::BuildHasher;
 use std::sync::atomic::{AtomicU32, Ordering};
-use std::{env, io};
+use std::{array, env, fmt, io};

 use chrono::{DateTime, Utc};
+use indexmap::IndexSet;
 use opentelemetry::trace::TraceContextExt;
+use scopeguard::defer;
 use serde::ser::{SerializeMap, Serializer};
 use tracing::subscriber::Interest;
 use tracing::{Event, Metadata, Span, Subscriber, callsite, span};
@@ -17,6 +19,7 @@ use tracing_subscriber::fmt::{FormatEvent, FormatFields};
 use tracing_subscriber::layer::{Context, Layer};
 use tracing_subscriber::prelude::*;
 use tracing_subscriber::registry::{LookupSpan, SpanRef};
+use try_lock::TryLock;

 /// Initialize logging and OpenTelemetry tracing and exporter.
 ///
@@ -52,7 +55,7 @@ pub async fn init() -> anyhow::Result<LoggingGuard> {
            StderrWriter {
                stderr: std::io::stderr(),
            },
-            &["request_id", "session_id", "conn_id"],
+            ["request_id", "session_id", "conn_id"],
        ))
    } else {
        None
@@ -180,65 +183,50 @@ impl Clock for RealClock {
 /// Name of the field used by tracing crate to store the event message.
 const MESSAGE_FIELD: &str = "message";

-/// Tracing used to enforce that spans/events have no more than 32 fields.
-/// It seems this is no longer the case, but it's still documented in some places.
-/// Generally, we shouldn't expect more than 32 fields anyway, so we can try and
-/// rely on it for some (minor) performance gains.
-const MAX_TRACING_FIELDS: usize = 32;
-
 thread_local! {
+    /// Protects against deadlocks and double panics during log writing.
+    /// The current panic handler will use tracing to log panic information.
+    static REENTRANCY_GUARD: Cell<bool> = const { Cell::new(false) };
    /// Thread-local instance with per-thread buffer for log writing.
-    static EVENT_FORMATTER: RefCell<EventFormatter> = const { RefCell::new(EventFormatter::new()) };
+    static EVENT_FORMATTER: RefCell<EventFormatter> = RefCell::new(EventFormatter::new());
    /// Cached OS thread ID.
    static THREAD_ID: u64 = gettid::gettid();
 }

-/// Map for values fixed at callsite registration.
-// We use papaya here because registration rarely happens post-startup.
-// papaya is good for read-heavy workloads.
-//
-// We use rustc_hash here because callsite::Identifier will always be an integer with low-bit entropy,
-// since it's always a pointer to static mutable data. rustc_hash was designed for low-bit entropy.
-type CallsiteMap<T> =
-    papaya::HashMap<callsite::Identifier, T, std::hash::BuildHasherDefault<rustc_hash::FxHasher>>;
-
 /// Implements tracing layer to handle events specific to logging.
-struct JsonLoggingLayer<C: Clock, W: MakeWriter> {
+struct JsonLoggingLayer<C: Clock, W: MakeWriter, const F: usize> {
    clock: C,
+    skipped_field_indices: papaya::HashMap<callsite::Identifier, SkippedFieldIndices>,
+    callsite_ids: papaya::HashMap<callsite::Identifier, CallsiteId>,
    writer: W,
-
-    /// tracks which fields of each **event** are duplicates
-    skipped_field_indices: CallsiteMap<SkippedFieldIndices>,
-
-    span_info: CallsiteMap<CallsiteSpanInfo>,
-
-    /// Fields we want to keep track of in a separate json object.
-    extract_fields: &'static [&'static str],
+    // We use a const generic and arrays to bypass one heap allocation.
+    extract_fields: IndexSet<&'static str>,
+    _marker: std::marker::PhantomData<[&'static str; F]>,
 }

-impl<C: Clock, W: MakeWriter> JsonLoggingLayer<C, W> {
-    fn new(clock: C, writer: W, extract_fields: &'static [&'static str]) -> Self {
+impl<C: Clock, W: MakeWriter, const F: usize> JsonLoggingLayer<C, W, F> {
+    fn new(clock: C, writer: W, extract_fields: [&'static str; F]) -> Self {
        JsonLoggingLayer {
            clock,
-            skipped_field_indices: CallsiteMap::default(),
-            span_info: CallsiteMap::default(),
+            skipped_field_indices: papaya::HashMap::default(),
+            callsite_ids: papaya::HashMap::default(),
            writer,
-            extract_fields,
+            extract_fields: IndexSet::from_iter(extract_fields),
+            _marker: std::marker::PhantomData,
        }
    }

    #[inline]
-    fn span_info(&self, metadata: &'static Metadata<'static>) -> CallsiteSpanInfo {
-        self.span_info
+    fn callsite_id(&self, cs: callsite::Identifier) -> CallsiteId {
+        *self
+            .callsite_ids
            .pin()
-            .get_or_insert_with(metadata.callsite(), || {
-                CallsiteSpanInfo::new(metadata, self.extract_fields)
-            })
-            .clone()
+            .get_or_insert_with(cs, CallsiteId::next)
    }
 }

-impl<S, C: Clock + 'static, W: MakeWriter + 'static> Layer<S> for JsonLoggingLayer<C, W>
+impl<S, C: Clock + 'static, W: MakeWriter + 'static, const F: usize> Layer<S>
+    for JsonLoggingLayer<C, W, F>
 where
    S: Subscriber + for<'a> LookupSpan<'a>,
 {
@@ -249,25 +237,35 @@ where
        //       early, before OTel machinery, and add as event extension.
        let now = self.clock.now();

-        let res: io::Result<()> = EVENT_FORMATTER.with(|f| {
-            let mut borrow = f.try_borrow_mut();
-            let formatter = match borrow.as_deref_mut() {
-                Ok(formatter) => formatter,
-                // If the thread local formatter is borrowed,
-                // then we likely hit an edge case were we panicked during formatting.
-                // We allow the logging to proceed with an uncached formatter.
-                Err(_) => &mut EventFormatter::new(),
-            };
+        let res: io::Result<()> = REENTRANCY_GUARD.with(move |entered| {
+            if entered.get() {
+                let mut formatter = EventFormatter::new();
+                formatter.format::<S, F>(
+                    now,
+                    event,
+                    &ctx,
+                    &self.skipped_field_indices,
+                    &self.callsite_ids,
+                    &self.extract_fields,
+                )?;
+                self.writer.make_writer().write_all(formatter.buffer())
+            } else {
+                entered.set(true);
+                defer!(entered.set(false););

-            formatter.reset();
-            formatter.format(
-                now,
-                event,
-                &ctx,
-                &self.skipped_field_indices,
-                self.extract_fields,
-            )?;
-            self.writer.make_writer().write_all(formatter.buffer())
+                EVENT_FORMATTER.with_borrow_mut(move |formatter| {
+                    formatter.reset();
+                    formatter.format::<S, F>(
+                        now,
+                        event,
+                        &ctx,
+                        &self.skipped_field_indices,
+                        &self.callsite_ids,
+                        &self.extract_fields,
+                    )?;
+                    self.writer.make_writer().write_all(formatter.buffer())
+                })
+            }
        });

        // In case logging fails we generate a simpler JSON object.
@@ -289,48 +287,50 @@ where
    /// Registers a SpanFields instance as span extension.
    fn on_new_span(&self, attrs: &span::Attributes<'_>, id: &span::Id, ctx: Context<'_, S>) {
        let span = ctx.span(id).expect("span must exist");
+        let fields = SpanFields::default();
+        fields.record_fields(attrs);

-        let mut fields = SpanFields::new(self.span_info(span.metadata()));
-        attrs.record(&mut fields);
+        // This could deadlock when there's a panic somewhere in the tracing
+        // event handling and a read or write guard is still held. This includes
+        // the OTel subscriber.
+        let mut exts = span.extensions_mut();

-        // This is a new span: the extensions should not be locked
-        // unless some layer spawned a thread to process this span.
-        // I don't think any layers do that.
-        span.extensions_mut().insert(fields);
+        exts.insert(fields);
    }

    fn on_record(&self, id: &span::Id, values: &span::Record<'_>, ctx: Context<'_, S>) {
        let span = ctx.span(id).expect("span must exist");
-
-        // assumption: `on_record` is rarely called.
-        // assumption: a span being updated by one thread,
-        //             and formatted by another thread is even rarer.
-        let mut ext = span.extensions_mut();
-        if let Some(fields) = ext.get_mut::<SpanFields>() {
-            values.record(fields);
+        let ext = span.extensions();
+        if let Some(data) = ext.get::<SpanFields>() {
+            data.record_fields(values);
        }
    }

-    /// Called (lazily) roughly once per event/span instance. We quickly check
-    /// for duplicate field names and record duplicates as skippable. Last field wins.
+    /// Called (lazily) whenever a new log call is executed. We quickly check
+    /// for duplicate field names and record duplicates as skippable. Last one
+    /// wins.
    fn register_callsite(&self, metadata: &'static Metadata<'static>) -> Interest {
-        debug_assert!(
-            metadata.fields().len() <= MAX_TRACING_FIELDS,
-            "callsite {metadata:?} has too many fields."
-        );
-
        if !metadata.is_event() {
-            // register the span info.
-            self.span_info(metadata);
+            self.callsite_id(metadata.callsite());
            // Must not be never because we wouldn't get trace and span data.
            return Interest::always();
        }

        let mut field_indices = SkippedFieldIndices::default();
-        let mut seen_fields = HashMap::new();
+        let mut seen_fields = HashMap::<&'static str, usize>::new();
        for field in metadata.fields() {
-            if let Some(old_index) = seen_fields.insert(field.name(), field.index()) {
-                field_indices.set(old_index);
+            use std::collections::hash_map::Entry;
+            match seen_fields.entry(field.name()) {
+                Entry::Vacant(entry) => {
+                    // field not seen yet
+                    entry.insert(field.index());
+                }
+                Entry::Occupied(mut entry) => {
+                    // replace currently stored index
+                    let old_index = entry.insert(field.index());
+                    // ... and append it to list of skippable indices
+                    field_indices.push(old_index);
+                }
            }
        }

@@ -344,113 +344,110 @@ where
    }
 }

-/// Any span info that is fixed to a particular callsite. Not variable between span instances.
-#[derive(Clone)]
-struct CallsiteSpanInfo {
-    /// index of each field to extract. usize::MAX if not found.
-    extract: Arc<[usize]>,
+#[derive(Copy, Clone, Debug, Default)]
+#[repr(transparent)]
+struct CallsiteId(u32);

-    /// tracks the fixed "callsite ID" for each span.
-    /// note: this is not stable between runs.
-    normalized_name: Arc<str>,
-}
-
-impl CallsiteSpanInfo {
-    fn new(metadata: &'static Metadata<'static>, extract_fields: &[&'static str]) -> Self {
+impl CallsiteId {
+    #[inline]
+    fn next() -> Self {
        // Start at 1 to reserve 0 for default.
        static COUNTER: AtomicU32 = AtomicU32::new(1);
+        CallsiteId(COUNTER.fetch_add(1, Ordering::Relaxed))
+    }
+}

-        let names: Vec<&'static str> = metadata.fields().iter().map(|f| f.name()).collect();
-
-        // get all the indices of span fields we want to focus
-        let extract = extract_fields
-            .iter()
-            // use rposition, since we want last match wins.
-            .map(|f1| names.iter().rposition(|f2| f1 == f2).unwrap_or(usize::MAX))
-            .collect();
-
-        // normalized_name is unique for each callsite, but it is not
-        // unified across separate proxy instances.
-        // todo: can we do better here?
-        let cid = COUNTER.fetch_add(1, Ordering::Relaxed);
-        let normalized_name = format!("{}#{cid}", metadata.name()).into();
-
-        Self {
-            extract,
-            normalized_name,
-        }
+impl fmt::Display for CallsiteId {
+    #[inline]
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.0.fmt(f)
    }
 }

 /// Stores span field values recorded during the spans lifetime.
+#[derive(Default)]
 struct SpanFields {
-    values: [serde_json::Value; MAX_TRACING_FIELDS],
-
-    /// cached span info so we can avoid extra hashmap lookups in the hot path.
-    span_info: CallsiteSpanInfo,
+    // TODO: Switch to custom enum with lasso::Spur for Strings?
+    fields: papaya::HashMap<&'static str, serde_json::Value>,
 }

 impl SpanFields {
-    fn new(span_info: CallsiteSpanInfo) -> Self {
-        Self {
-            span_info,
-            values: [const { serde_json::Value::Null }; MAX_TRACING_FIELDS],
-        }
+    #[inline]
+    fn record_fields<R: tracing_subscriber::field::RecordFields>(&self, fields: R) {
+        fields.record(&mut SpanFieldsRecorder {
+            fields: self.fields.pin(),
+        });
    }
 }

-impl tracing::field::Visit for SpanFields {
+/// Implements a tracing field visitor to convert and store values.
+struct SpanFieldsRecorder<'m, S, G> {
+    fields: papaya::HashMapRef<'m, &'static str, serde_json::Value, S, G>,
+}
+
+impl<S: BuildHasher, G: papaya::Guard> tracing::field::Visit for SpanFieldsRecorder<'_, S, G> {
    #[inline]
    fn record_f64(&mut self, field: &tracing::field::Field, value: f64) {
-        self.values[field.index()] = serde_json::Value::from(value);
+        self.fields
+            .insert(field.name(), serde_json::Value::from(value));
    }

    #[inline]
    fn record_i64(&mut self, field: &tracing::field::Field, value: i64) {
-        self.values[field.index()] = serde_json::Value::from(value);
+        self.fields
+            .insert(field.name(), serde_json::Value::from(value));
    }

    #[inline]
    fn record_u64(&mut self, field: &tracing::field::Field, value: u64) {
-        self.values[field.index()] = serde_json::Value::from(value);
+        self.fields
+            .insert(field.name(), serde_json::Value::from(value));
    }

    #[inline]
    fn record_i128(&mut self, field: &tracing::field::Field, value: i128) {
        if let Ok(value) = i64::try_from(value) {
-            self.values[field.index()] = serde_json::Value::from(value);
+            self.fields
+                .insert(field.name(), serde_json::Value::from(value));
        } else {
-            self.values[field.index()] = serde_json::Value::from(format!("{value}"));
+            self.fields
+                .insert(field.name(), serde_json::Value::from(format!("{value}")));
        }
    }

    #[inline]
    fn record_u128(&mut self, field: &tracing::field::Field, value: u128) {
        if let Ok(value) = u64::try_from(value) {
-            self.values[field.index()] = serde_json::Value::from(value);
+            self.fields
+                .insert(field.name(), serde_json::Value::from(value));
        } else {
-            self.values[field.index()] = serde_json::Value::from(format!("{value}"));
+            self.fields
+                .insert(field.name(), serde_json::Value::from(format!("{value}")));
        }
    }

    #[inline]
    fn record_bool(&mut self, field: &tracing::field::Field, value: bool) {
-        self.values[field.index()] = serde_json::Value::from(value);
+        self.fields
+            .insert(field.name(), serde_json::Value::from(value));
    }

    #[inline]
    fn record_bytes(&mut self, field: &tracing::field::Field, value: &[u8]) {
-        self.values[field.index()] = serde_json::Value::from(value);
+        self.fields
+            .insert(field.name(), serde_json::Value::from(value));
    }

    #[inline]
    fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
-        self.values[field.index()] = serde_json::Value::from(value);
+        self.fields
+            .insert(field.name(), serde_json::Value::from(value));
    }

    #[inline]
    fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
-        self.values[field.index()] = serde_json::Value::from(format!("{value:?}"));
+        self.fields
+            .insert(field.name(), serde_json::Value::from(format!("{value:?}")));
    }

    #[inline]
@@ -459,33 +456,38 @@ impl tracing::field::Visit for SpanFields {
        field: &tracing::field::Field,
        value: &(dyn std::error::Error + 'static),
    ) {
-        self.values[field.index()] = serde_json::Value::from(format!("{value}"));
+        self.fields
+            .insert(field.name(), serde_json::Value::from(format!("{value}")));
    }
 }

 /// List of field indices skipped during logging. Can list duplicate fields or
 /// metafields not meant to be logged.
-#[derive(Copy, Clone, Default)]
+#[derive(Clone, Default)]
 struct SkippedFieldIndices {
-    // 32-bits is large enough for `MAX_TRACING_FIELDS`
-    bits: u32,
+    bits: u64,
 }

 impl SkippedFieldIndices {
    #[inline]
-    fn is_empty(self) -> bool {
+    fn is_empty(&self) -> bool {
        self.bits == 0
    }

    #[inline]
-    fn set(&mut self, index: usize) {
-        debug_assert!(index <= 32, "index out of bounds of 32-bit set");
-        self.bits |= 1 << index;
+    fn push(&mut self, index: usize) {
+        self.bits |= 1u64
+            .checked_shl(index as u32)
+            .expect("field index too large");
    }

    #[inline]
-    fn contains(self, index: usize) -> bool {
-        self.bits & (1 << index) != 0
+    fn contains(&self, index: usize) -> bool {
+        self.bits
+            & 1u64
+                .checked_shl(index as u32)
+                .expect("field index too large")
+            != 0
    }
 }

@@ -497,7 +499,7 @@ struct EventFormatter {

 impl EventFormatter {
    #[inline]
-    const fn new() -> Self {
+    fn new() -> Self {
        EventFormatter {
            logline_buffer: Vec::new(),
        }
@@ -513,13 +515,14 @@ impl EventFormatter {
        self.logline_buffer.clear();
    }

-    fn format<S>(
+    fn format<S, const F: usize>(
        &mut self,
        now: DateTime<Utc>,
        event: &Event<'_>,
        ctx: &Context<'_, S>,
-        skipped_field_indices: &CallsiteMap<SkippedFieldIndices>,
-        extract_fields: &'static [&'static str],
+        skipped_field_indices: &papaya::HashMap<callsite::Identifier, SkippedFieldIndices>,
+        callsite_ids: &papaya::HashMap<callsite::Identifier, CallsiteId>,
+        extract_fields: &IndexSet<&'static str>,
    ) -> io::Result<()>
    where
        S: Subscriber + for<'a> LookupSpan<'a>,
@@ -530,11 +533,8 @@ impl EventFormatter {
        let normalized_meta = event.normalized_metadata();
        let meta = normalized_meta.as_ref().unwrap_or_else(|| event.metadata());

-        let skipped_field_indices = skipped_field_indices
-            .pin()
-            .get(&meta.callsite())
-            .copied()
-            .unwrap_or_default();
+        let skipped_field_indices = skipped_field_indices.pin();
+        let skipped_field_indices = skipped_field_indices.get(&meta.callsite());

        let mut serialize = || {
            let mut serializer = serde_json::Serializer::new(&mut self.logline_buffer);
@@ -565,11 +565,9 @@ impl EventFormatter {
            }

            let spans = SerializableSpans {
-                // collect all spans from parent to root.
-                spans: ctx
-                    .event_span(event)
-                    .map_or(vec![], |parent| parent.scope().collect()),
-                extracted: ExtractedSpanFields::new(extract_fields),
+                ctx,
+                callsite_ids,
+                extract: ExtractedSpanFields::<'_, F>::new(extract_fields),
            };
            serializer.serialize_entry("spans", &spans)?;

@@ -622,9 +620,9 @@ impl EventFormatter {
                }
            }

-            if spans.extracted.has_values() {
+            if spans.extract.has_values() {
                // TODO: add fields from event, too?
-                serializer.serialize_entry("extract", &spans.extracted)?;
+                serializer.serialize_entry("extract", &spans.extract)?;
            }

            serializer.end()
@@ -637,15 +635,15 @@ impl EventFormatter {
 }

 /// Extracts the message field that's mixed will other fields.
-struct MessageFieldExtractor<S: serde::ser::SerializeMap> {
+struct MessageFieldExtractor<'a, S: serde::ser::SerializeMap> {
    serializer: S,
-    skipped_field_indices: SkippedFieldIndices,
+    skipped_field_indices: Option<&'a SkippedFieldIndices>,
    state: Option<Result<(), S::Error>>,
 }

-impl<S: serde::ser::SerializeMap> MessageFieldExtractor<S> {
+impl<'a, S: serde::ser::SerializeMap> MessageFieldExtractor<'a, S> {
    #[inline]
-    fn new(serializer: S, skipped_field_indices: SkippedFieldIndices) -> Self {
+    fn new(serializer: S, skipped_field_indices: Option<&'a SkippedFieldIndices>) -> Self {
        Self {
            serializer,
            skipped_field_indices,
@@ -667,11 +665,13 @@ impl<S: serde::ser::SerializeMap> MessageFieldExtractor<S> {
    fn accept_field(&self, field: &tracing::field::Field) -> bool {
        self.state.is_none()
            && field.name() == MESSAGE_FIELD
-            && !self.skipped_field_indices.contains(field.index())
+            && !self
+                .skipped_field_indices
+                .is_some_and(|i| i.contains(field.index()))
    }
 }

-impl<S: serde::ser::SerializeMap> tracing::field::Visit for MessageFieldExtractor<S> {
+impl<S: serde::ser::SerializeMap> tracing::field::Visit for MessageFieldExtractor<'_, S> {
    #[inline]
    fn record_f64(&mut self, field: &tracing::field::Field, value: f64) {
        if self.accept_field(field) {
@@ -751,14 +751,14 @@ impl<S: serde::ser::SerializeMap> tracing::field::Visit for MessageFieldExtracto
 /// can be skipped.
 // This is entirely optional and only cosmetic, though maybe helps a
 // bit during log parsing in dashboards when there's no field with empty object.
-struct FieldsPresent(pub bool, SkippedFieldIndices);
+struct FieldsPresent<'a>(pub bool, Option<&'a SkippedFieldIndices>);

 // Even though some methods have an overhead (error, bytes) it is assumed the
 // compiler won't include this since we ignore the value entirely.
-impl tracing::field::Visit for FieldsPresent {
+impl tracing::field::Visit for FieldsPresent<'_> {
    #[inline]
    fn record_debug(&mut self, field: &tracing::field::Field, _: &dyn std::fmt::Debug) {
-        if !self.1.contains(field.index())
+        if !self.1.is_some_and(|i| i.contains(field.index()))
            && field.name() != MESSAGE_FIELD
            && !field.name().starts_with("log.")
        {
@@ -768,7 +768,10 @@ impl tracing::field::Visit for FieldsPresent {
 }

 /// Serializes the fields directly supplied with a log event.
-struct SerializableEventFields<'a, 'event>(&'a tracing::Event<'event>, SkippedFieldIndices);
+struct SerializableEventFields<'a, 'event>(
+    &'a tracing::Event<'event>,
+    Option<&'a SkippedFieldIndices>,
+);

 impl serde::ser::Serialize for SerializableEventFields<'_, '_> {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
@@ -785,15 +788,15 @@ impl serde::ser::Serialize for SerializableEventFields<'_, '_> {
 }

 /// A tracing field visitor that skips the message field.
-struct MessageFieldSkipper<S: serde::ser::SerializeMap> {
+struct MessageFieldSkipper<'a, S: serde::ser::SerializeMap> {
    serializer: S,
-    skipped_field_indices: SkippedFieldIndices,
+    skipped_field_indices: Option<&'a SkippedFieldIndices>,
    state: Result<(), S::Error>,
 }

-impl<S: serde::ser::SerializeMap> MessageFieldSkipper<S> {
+impl<'a, S: serde::ser::SerializeMap> MessageFieldSkipper<'a, S> {
    #[inline]
-    fn new(serializer: S, skipped_field_indices: SkippedFieldIndices) -> Self {
+    fn new(serializer: S, skipped_field_indices: Option<&'a SkippedFieldIndices>) -> Self {
        Self {
            serializer,
            skipped_field_indices,
@@ -806,7 +809,9 @@ impl<S: serde::ser::SerializeMap> MessageFieldSkipper<S> {
        self.state.is_ok()
            && field.name() != MESSAGE_FIELD
            && !field.name().starts_with("log.")
-            && !self.skipped_field_indices.contains(field.index())
+            && !self
+                .skipped_field_indices
+                .is_some_and(|i| i.contains(field.index()))
    }

    #[inline]
@@ -816,7 +821,7 @@ impl<S: serde::ser::SerializeMap> MessageFieldSkipper<S> {
    }
 }

-impl<S: serde::ser::SerializeMap> tracing::field::Visit for MessageFieldSkipper<S> {
+impl<S: serde::ser::SerializeMap> tracing::field::Visit for MessageFieldSkipper<'_, S> {
    #[inline]
    fn record_f64(&mut self, field: &tracing::field::Field, value: f64) {
        if self.accept_field(field) {
@@ -900,17 +905,18 @@ impl<S: serde::ser::SerializeMap> tracing::field::Visit for MessageFieldSkipper<
 /// with the span names as keys. To prevent collision we append a numberic value
 /// to the name. Also, collects any span fields we're interested in. Last one
 /// wins.
-struct SerializableSpans<'ctx, S>
+struct SerializableSpans<'a, 'ctx, Span, const F: usize>
 where
-    S: for<'lookup> LookupSpan<'lookup>,
+    Span: Subscriber + for<'lookup> LookupSpan<'lookup>,
 {
-    spans: Vec<SpanRef<'ctx, S>>,
-    extracted: ExtractedSpanFields,
+    ctx: &'a Context<'ctx, Span>,
+    callsite_ids: &'a papaya::HashMap<callsite::Identifier, CallsiteId>,
+    extract: ExtractedSpanFields<'a, F>,
 }

-impl<S> serde::ser::Serialize for SerializableSpans<'_, S>
+impl<Span, const F: usize> serde::ser::Serialize for SerializableSpans<'_, '_, Span, F>
 where
-    S: for<'lookup> LookupSpan<'lookup>,
+    Span: Subscriber + for<'lookup> LookupSpan<'lookup>,
 {
    fn serialize<Ser>(&self, serializer: Ser) -> Result<Ser::Ok, Ser::Error>
    where
@@ -918,22 +924,25 @@ where
    {
        let mut serializer = serializer.serialize_map(None)?;

-        for span in self.spans.iter().rev() {
-            let ext = span.extensions();
+        if let Some(leaf_span) = self.ctx.lookup_current() {
+            for span in leaf_span.scope().from_root() {
+                // Append a numeric callsite ID to the span name to keep the name unique
+                // in the JSON object.
+                let cid = self
+                    .callsite_ids
+                    .pin()
+                    .get(&span.metadata().callsite())
+                    .copied()
+                    .unwrap_or_default();

-            // all spans should have this extension.
-            let Some(fields) = ext.get() else { continue };
+                // Loki turns the # into an underscore during field name concatenation.
+                serializer.serialize_key(&format_args!("{}#{}", span.metadata().name(), &cid))?;

-            self.extracted.layer_span(fields);
-
-            let SpanFields { values, span_info } = fields;
-            serializer.serialize_entry(
-                &*span_info.normalized_name,
-                &SerializableSpanFields {
-                    fields: span.metadata().fields(),
-                    values,
-                },
-            )?;
+                serializer.serialize_value(&SerializableSpanFields {
+                    span: &span,
+                    extract: &self.extract,
+                })?;
+            }
        }

        serializer.end()
@@ -941,77 +950,80 @@ where
 }

 /// Serializes the span fields as object.
-struct SerializableSpanFields<'span> {
-    fields: &'span tracing::field::FieldSet,
-    values: &'span [serde_json::Value; MAX_TRACING_FIELDS],
+struct SerializableSpanFields<'a, 'span, Span, const F: usize>
+where
+    Span: for<'lookup> LookupSpan<'lookup>,
+{
+    span: &'a SpanRef<'span, Span>,
+    extract: &'a ExtractedSpanFields<'a, F>,
 }

-impl serde::ser::Serialize for SerializableSpanFields<'_> {
+impl<Span, const F: usize> serde::ser::Serialize for SerializableSpanFields<'_, '_, Span, F>
+where
+    Span: for<'lookup> LookupSpan<'lookup>,
+{
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::ser::Serializer,
    {
        let mut serializer = serializer.serialize_map(None)?;

-        for (field, value) in std::iter::zip(self.fields, self.values) {
-            if value.is_null() {
-                continue;
+        let ext = self.span.extensions();
+        if let Some(data) = ext.get::<SpanFields>() {
+            for (name, value) in &data.fields.pin() {
+                serializer.serialize_entry(name, value)?;
+                // TODO: replace clone with reference, if possible.
+                self.extract.set(name, value.clone());
            }
-            serializer.serialize_entry(field.name(), value)?;
        }

        serializer.end()
    }
 }

-struct ExtractedSpanFields {
-    names: &'static [&'static str],
-    values: RefCell<Vec<serde_json::Value>>,
+struct ExtractedSpanFields<'a, const F: usize> {
+    names: &'a IndexSet<&'static str>,
+    // TODO: replace TryLock with something local thread and interior mutability.
+    //       serde API doesn't let us use `mut`.
+    values: TryLock<([Option<serde_json::Value>; F], bool)>,
 }

-impl ExtractedSpanFields {
-    fn new(names: &'static [&'static str]) -> Self {
+impl<'a, const F: usize> ExtractedSpanFields<'a, F> {
+    fn new(names: &'a IndexSet<&'static str>) -> Self {
        ExtractedSpanFields {
            names,
-            values: RefCell::new(vec![serde_json::Value::Null; names.len()]),
+            values: TryLock::new((array::from_fn(|_| Option::default()), false)),
        }
    }

-    fn layer_span(&self, fields: &SpanFields) {
-        let mut v = self.values.borrow_mut();
-        let SpanFields { values, span_info } = fields;
-
-        // extract the fields
-        for (i, &j) in span_info.extract.iter().enumerate() {
-            let Some(value) = values.get(j) else { continue };
-
-            if !value.is_null() {
-                // TODO: replace clone with reference, if possible.
-                v[i] = value.clone();
-            }
+    #[inline]
+    fn set(&self, name: &'static str, value: serde_json::Value) {
+        if let Some((index, _)) = self.names.get_full(name) {
+            let mut fields = self.values.try_lock().expect("thread-local use");
+            fields.0[index] = Some(value);
+            fields.1 = true;
        }
    }

    #[inline]
    fn has_values(&self) -> bool {
-        self.values.borrow().iter().any(|v| !v.is_null())
+        self.values.try_lock().expect("thread-local use").1
    }
 }

-impl serde::ser::Serialize for ExtractedSpanFields {
+impl<const F: usize> serde::ser::Serialize for ExtractedSpanFields<'_, F> {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::ser::Serializer,
    {
        let mut serializer = serializer.serialize_map(None)?;

-        let values = self.values.borrow();
-        for (key, value) in std::iter::zip(self.names, &*values) {
-            if value.is_null() {
-                continue;
+        let values = self.values.try_lock().expect("thread-local use");
+        for (i, value) in values.0.iter().enumerate() {
+            if let Some(value) = value {
+                let key = self.names[i];
+                serializer.serialize_entry(key, value)?;
            }
-
-            serializer.serialize_entry(key, value)?;
        }

        serializer.end()
@@ -1020,6 +1032,7 @@ impl serde::ser::Serialize for ExtractedSpanFields {

 #[cfg(test)]
 mod tests {
+    use std::marker::PhantomData;
    use std::sync::{Arc, Mutex, MutexGuard};

    use assert_json_diff::assert_json_eq;
@@ -1068,9 +1081,10 @@ mod tests {
        let log_layer = JsonLoggingLayer {
            clock: clock.clone(),
            skipped_field_indices: papaya::HashMap::default(),
-            span_info: papaya::HashMap::default(),
+            callsite_ids: papaya::HashMap::default(),
            writer: buffer.clone(),
-            extract_fields: &["x"],
+            extract_fields: IndexSet::from_iter(["x"]),
+            _marker: PhantomData::<[&'static str; 1]>,
        };

        let registry = tracing_subscriber::Registry::default().with(log_layer);
--- a/proxy/src/proxy/mod.rs
+++ b/proxy/src/proxy/mod.rs
@@ -383,7 +383,9 @@ pub(crate) async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
    let cancellation_handler_clone = Arc::clone(&cancellation_handler);
    let session = cancellation_handler_clone.get_key();

-    session.write_cancel_key(node.cancel_closure.clone())?;
+    session
+        .write_cancel_key(node.cancel_closure.clone())
+        .await?;

    prepare_client_connection(&node, *session.key(), &mut stream).await?;

--- a/proxy/src/proxy/passthrough.rs
+++ b/proxy/src/proxy/passthrough.rs
@@ -94,7 +94,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin> ProxyPassthrough<S> {
            tracing::warn!(session_id = ?self.session_id, ?err, "could not cancel the query in the database");
        }

-        drop(self.cancel.remove_cancel_key()); // we don't need a result. If the queue is full, we just log the error
+        drop(self.cancel.remove_cancel_key().await); // we don't need a result. If the queue is full, we just log the error

        res
    }
--- a/proxy/src/proxy/retry.rs
+++ b/proxy/src/proxy/retry.rs
@@ -48,7 +48,7 @@ impl ShouldRetryWakeCompute for postgres_client::error::DbError {
        use postgres_client::error::SqlState;
        // Here are errors that happens after the user successfully authenticated to the database.
        // TODO: there are pgbouncer errors that should be retried, but they are not listed here.
-        let non_retriable_pg_errors = matches!(
+        !matches!(
            self.code(),
            &SqlState::TOO_MANY_CONNECTIONS
                | &SqlState::OUT_OF_MEMORY
@@ -56,20 +56,8 @@ impl ShouldRetryWakeCompute for postgres_client::error::DbError {
                | &SqlState::T_R_SERIALIZATION_FAILURE
                | &SqlState::INVALID_CATALOG_NAME
                | &SqlState::INVALID_SCHEMA_NAME
-                | &SqlState::INVALID_PARAMETER_VALUE,
-        );
-        if non_retriable_pg_errors {
-            return false;
-        }
-        // PGBouncer errors that should not trigger a wake_compute retry.
-        if self.code() == &SqlState::PROTOCOL_VIOLATION {
-            // Source for the error message:
-            // https://github.com/pgbouncer/pgbouncer/blob/f15997fe3effe3a94ba8bcc1ea562e6117d1a131/src/client.c#L1070
-            return !self
-                .message()
-                .contains("no more connections allowed (max_client_conn)");
-        }
-        true
+                | &SqlState::INVALID_PARAMETER_VALUE
+        )
    }
 }

@@ -122,55 +110,3 @@ pub(crate) fn retry_after(num_retries: u32, config: RetryConfig) -> time::Durati
        .base_delay
        .mul_f64(config.backoff_factor.powi((num_retries as i32) - 1))
 }
-
-#[cfg(test)]
-mod tests {
-    use super::ShouldRetryWakeCompute;
-    use postgres_client::error::{DbError, SqlState};
-
-    #[test]
-    fn should_retry_wake_compute_for_db_error() {
-        // These SQLStates should NOT trigger a wake_compute retry.
-        let non_retry_states = [
-            SqlState::TOO_MANY_CONNECTIONS,
-            SqlState::OUT_OF_MEMORY,
-            SqlState::SYNTAX_ERROR,
-            SqlState::T_R_SERIALIZATION_FAILURE,
-            SqlState::INVALID_CATALOG_NAME,
-            SqlState::INVALID_SCHEMA_NAME,
-            SqlState::INVALID_PARAMETER_VALUE,
-        ];
-        for state in non_retry_states {
-            let err = DbError::new_test_error(state.clone(), "oops".to_string());
-            assert!(
-                !err.should_retry_wake_compute(),
-                "State {state:?} unexpectedly retried"
-            );
-        }
-
-        // Errors coming from pgbouncer should not trigger a wake_compute retry
-        let non_retry_pgbouncer_errors = ["no more connections allowed (max_client_conn)"];
-        for error in non_retry_pgbouncer_errors {
-            let err = DbError::new_test_error(SqlState::PROTOCOL_VIOLATION, error.to_string());
-            assert!(
-                !err.should_retry_wake_compute(),
-                "PGBouncer error {error:?} unexpectedly retried"
-            );
-        }
-
-        // These SQLStates should trigger a wake_compute retry.
-        let retry_states = [
-            SqlState::CONNECTION_FAILURE,
-            SqlState::CONNECTION_EXCEPTION,
-            SqlState::CONNECTION_DOES_NOT_EXIST,
-            SqlState::SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION,
-        ];
-        for state in retry_states {
-            let err = DbError::new_test_error(state.clone(), "oops".to_string());
-            assert!(
-                err.should_retry_wake_compute(),
-                "State {state:?} unexpectedly skipped retry"
-            );
-        }
-    }
-}
--- a/proxy/src/proxy/tests/mod.rs
+++ b/proxy/src/proxy/tests/mod.rs
@@ -15,7 +15,6 @@ use rstest::rstest;
 use rustls::crypto::ring;
 use rustls::pki_types;
 use tokio::io::DuplexStream;
-use tracing_test::traced_test;

 use super::connect_compute::ConnectMechanism;
 use super::retry::CouldRetry;
@@ -382,14 +381,8 @@ enum ConnectAction {
    WakeFail,
    WakeRetry,
    Connect,
-    // connect_once -> Err, could_retry = true, should_retry_wake_compute = true
    Retry,
-    // connect_once -> Err, could_retry = true, should_retry_wake_compute = false
-    RetryNoWake,
-    // connect_once -> Err, could_retry = false, should_retry_wake_compute = true
    Fail,
-    // connect_once -> Err, could_retry = false, should_retry_wake_compute = false
-    FailNoWake,
 }

 #[derive(Clone)]
@@ -431,7 +424,6 @@ struct TestConnection;
 #[derive(Debug)]
 struct TestConnectError {
    retryable: bool,
-    wakeable: bool,
    kind: crate::error::ErrorKind,
 }

@@ -456,7 +448,7 @@ impl CouldRetry for TestConnectError {
 }
 impl ShouldRetryWakeCompute for TestConnectError {
    fn should_retry_wake_compute(&self) -> bool {
-        self.wakeable
+        true
    }
 }

@@ -479,22 +471,10 @@ impl ConnectMechanism for TestConnectMechanism {
            ConnectAction::Connect => Ok(TestConnection),
            ConnectAction::Retry => Err(TestConnectError {
                retryable: true,
-                wakeable: true,
-                kind: ErrorKind::Compute,
-            }),
-            ConnectAction::RetryNoWake => Err(TestConnectError {
-                retryable: true,
-                wakeable: false,
                kind: ErrorKind::Compute,
            }),
            ConnectAction::Fail => Err(TestConnectError {
                retryable: false,
-                wakeable: true,
-                kind: ErrorKind::Compute,
-            }),
-            ConnectAction::FailNoWake => Err(TestConnectError {
-                retryable: false,
-                wakeable: false,
                kind: ErrorKind::Compute,
            }),
            x => panic!("expecting action {x:?}, connect is called instead"),
@@ -729,92 +709,3 @@ async fn wake_non_retry() {
        .unwrap_err();
    mechanism.verify();
 }
-
-#[tokio::test]
-#[traced_test]
-async fn fail_but_wake_invalidates_cache() {
-    let ctx = RequestContext::test();
-    let mech = TestConnectMechanism::new(vec![
-        ConnectAction::Wake,
-        ConnectAction::Fail,
-        ConnectAction::Wake,
-        ConnectAction::Connect,
-    ]);
-    let user = helper_create_connect_info(&mech);
-    let cfg = config();
-
-    connect_to_compute(&ctx, &mech, &user, cfg.retry, &cfg)
-        .await
-        .unwrap();
-
-    assert!(logs_contain(
-        "invalidating stalled compute node info cache entry"
-    ));
-}
-
-#[tokio::test]
-#[traced_test]
-async fn fail_no_wake_skips_cache_invalidation() {
-    let ctx = RequestContext::test();
-    let mech = TestConnectMechanism::new(vec![
-        ConnectAction::Wake,
-        ConnectAction::FailNoWake,
-        ConnectAction::Connect,
-    ]);
-    let user = helper_create_connect_info(&mech);
-    let cfg = config();
-
-    connect_to_compute(&ctx, &mech, &user, cfg.retry, &cfg)
-        .await
-        .unwrap();
-
-    assert!(!logs_contain(
-        "invalidating stalled compute node info cache entry"
-    ));
-}
-
-#[tokio::test]
-#[traced_test]
-async fn retry_but_wake_invalidates_cache() {
-    let _ = env_logger::try_init();
-    use ConnectAction::*;
-
-    let ctx = RequestContext::test();
-    // Wake → Retry (retryable + wakeable) → Wake → Connect
-    let mechanism = TestConnectMechanism::new(vec![Wake, Retry, Wake, Connect]);
-    let user_info = helper_create_connect_info(&mechanism);
-    let cfg = config();
-
-    connect_to_compute(&ctx, &mechanism, &user_info, cfg.retry, &cfg)
-        .await
-        .unwrap();
-    mechanism.verify();
-
-    // Because Retry has wakeable=true, we should see invalidate_cache
-    assert!(logs_contain(
-        "invalidating stalled compute node info cache entry"
-    ));
-}
-
-#[tokio::test]
-#[traced_test]
-async fn retry_no_wake_skips_invalidation() {
-    let _ = env_logger::try_init();
-    use ConnectAction::*;
-
-    let ctx = RequestContext::test();
-    // Wake → RetryNoWake (retryable + NOT wakeable)
-    let mechanism = TestConnectMechanism::new(vec![Wake, RetryNoWake]);
-    let user_info = helper_create_connect_info(&mechanism);
-    let cfg = config();
-
-    connect_to_compute(&ctx, &mechanism, &user_info, cfg.retry, &cfg)
-        .await
-        .unwrap_err();
-    mechanism.verify();
-
-    // Because RetryNoWake has wakeable=false, we must NOT see invalidate_cache
-    assert!(!logs_contain(
-        "invalidating stalled compute node info cache entry"
-    ));
-}
--- a/proxy/src/scram/pbkdf2.rs
+++ b/proxy/src/scram/pbkdf2.rs
@@ -13,19 +13,22 @@ pub(crate) struct Pbkdf2 {
 // inspired from <https://github.com/neondatabase/rust-postgres/blob/20031d7a9ee1addeae6e0968e3899ae6bf01cee2/postgres-protocol/src/authentication/sasl.rs#L36-L61>
 impl Pbkdf2 {
    pub(crate) fn start(str: &[u8], salt: &[u8], iterations: u32) -> Self {
-        // key the HMAC and derive the first block in-place
-        let mut hmac =
+        let hmac =
            Hmac::<Sha256>::new_from_slice(str).expect("HMAC is able to accept all key sizes");
-        hmac.update(salt);
-        hmac.update(&1u32.to_be_bytes());
-        let init_block = hmac.finalize_reset().into_bytes();
+
+        let prev = hmac
+            .clone()
+            .chain_update(salt)
+            .chain_update(1u32.to_be_bytes())
+            .finalize()
+            .into_bytes();

        Self {
            hmac,
-            // one iteration spent above
+            // one consumed for the hash above
            iterations: iterations - 1,
-            hi: init_block,
-            prev: init_block,
+            hi: prev,
+            prev,
        }
    }

@@ -41,17 +44,14 @@ impl Pbkdf2 {
            iterations,
        } = self;

-        // only do up to 4096 iterations per turn for fairness
+        // only do 4096 iterations per turn before sharing the thread for fairness
        let n = (*iterations).clamp(0, 4096);
        for _ in 0..n {
-            hmac.update(prev);
-            let block = hmac.finalize_reset().into_bytes();
+            *prev = hmac.clone().chain_update(*prev).finalize().into_bytes();

-            for (hi_byte, &b) in hi.iter_mut().zip(block.iter()) {
-                *hi_byte ^= b;
+            for (hi, prev) in hi.iter_mut().zip(*prev) {
+                *hi ^= prev;
            }
-
-            *prev = block;
        }

        *iterations -= n;
--- a/proxy/src/url.rs
+++ b/proxy/src/url.rs
@@ -43,12 +43,6 @@ impl std::ops::Deref for ApiUrl {
    }
 }

-impl std::ops::DerefMut for ApiUrl {
-    fn deref_mut(&mut self) -> &mut Self::Target {
-        &mut self.0
-    }
-}
-
 impl std::fmt::Display for ApiUrl {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        self.0.fmt(f)
--- a/scripts/benchmark_durations.py
+++ b/scripts/benchmark_durations.py
@@ -32,6 +32,12 @@ BENCHMARKS_DURATION_QUERY = """
 # the total duration varies from 8 to 40 minutes.
 # We use some pre-collected durations as a fallback to have a better distribution.
 FALLBACK_DURATION = {
+    "test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py::test_pageserver_max_throughput_getpage_at_latest_lsn[1-13-30]": 400.15,
+    "test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py::test_pageserver_max_throughput_getpage_at_latest_lsn[1-6-30]": 372.521,
+    "test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py::test_pageserver_max_throughput_getpage_at_latest_lsn[10-13-30]": 420.017,
+    "test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py::test_pageserver_max_throughput_getpage_at_latest_lsn[10-6-30]": 373.769,
+    "test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py::test_pageserver_max_throughput_getpage_at_latest_lsn[100-13-30]": 678.742,
+    "test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py::test_pageserver_max_throughput_getpage_at_latest_lsn[100-6-30]": 512.135,
    "test_runner/performance/test_branch_creation.py::test_branch_creation_heavy_write[20]": 58.036,
    "test_runner/performance/test_branch_creation.py::test_branch_creation_many_relations": 22.104,
    "test_runner/performance/test_branch_creation.py::test_branch_creation_many[1024]": 126.073,
--- a/storage_broker/src/bin/storage_broker.rs
+++ b/storage_broker/src/bin/storage_broker.rs
@@ -17,14 +17,12 @@ use std::pin::Pin;
 use std::sync::Arc;
 use std::time::Duration;

-use bytes::Bytes;
 use camino::Utf8PathBuf;
 use clap::{Parser, command};
 use futures::future::OptionFuture;
 use futures_core::Stream;
 use futures_util::StreamExt;
-use http_body_util::combinators::BoxBody;
-use http_body_util::{Empty, Full};
+use http_body_util::Full;
 use http_utils::tls_certs::ReloadingCertificateResolver;
 use hyper::body::Incoming;
 use hyper::header::CONTENT_TYPE;
@@ -48,6 +46,7 @@ use tokio::net::TcpListener;
 use tokio::sync::broadcast;
 use tokio::sync::broadcast::error::RecvError;
 use tokio::time;
+use tonic::body::{self, BoxBody, empty_body};
 use tonic::codegen::Service;
 use tonic::{Code, Request, Response, Status};
 use tracing::*;
@@ -635,7 +634,7 @@ impl BrokerService for Broker {
 // We serve only metrics and healthcheck through http1.
 async fn http1_handler(
    req: hyper::Request<Incoming>,
-) -> Result<hyper::Response<BoxBody<Bytes, Infallible>>, Infallible> {
+) -> Result<hyper::Response<BoxBody>, Infallible> {
    let resp = match (req.method(), req.uri().path()) {
        (&Method::GET, "/metrics") => {
            let mut buffer = vec![];
@@ -646,16 +645,16 @@ async fn http1_handler(
            hyper::Response::builder()
                .status(StatusCode::OK)
                .header(CONTENT_TYPE, encoder.format_type())
-                .body(BoxBody::new(Full::new(Bytes::from(buffer))))
+                .body(body::boxed(Full::new(bytes::Bytes::from(buffer))))
                .unwrap()
        }
        (&Method::GET, "/status") => hyper::Response::builder()
            .status(StatusCode::OK)
-            .body(BoxBody::new(Empty::new()))
+            .body(empty_body())
            .unwrap(),
        _ => hyper::Response::builder()
            .status(StatusCode::NOT_FOUND)
-            .body(BoxBody::new(Empty::new()))
+            .body(empty_body())
            .unwrap(),
    };
    Ok(resp)
--- a/test_runner/fixtures/metrics.py
+++ b/test_runner/fixtures/metrics.py
@@ -184,7 +184,6 @@ PAGESERVER_PER_TENANT_METRICS: tuple[str, ...] = (
    "pageserver_evictions_with_low_residence_duration_total",
    "pageserver_aux_file_estimated_size",
    "pageserver_valid_lsn_lease_count",
-    "pageserver_tenant_offloaded_timelines",
    counter("pageserver_tenant_throttling_count_accounted_start"),
    counter("pageserver_tenant_throttling_count_accounted_finish"),
    counter("pageserver_tenant_throttling_wait_usecs_sum"),
--- a/test_runner/fixtures/neon_fixtures.py
+++ b/test_runner/fixtures/neon_fixtures.py
@@ -682,7 +682,7 @@ class NeonEnvBuilder:
                log.info(
                    f"Copying pageserver tenants directory {tenants_from_dir} to {tenants_to_dir}"
                )
-                subprocess.run(["cp", "-a", tenants_from_dir, tenants_to_dir], check=True)
+                shutil.copytree(tenants_from_dir, tenants_to_dir)
            else:
                log.info(
                    f"Creating overlayfs mount of pageserver tenants directory {tenants_from_dir} to {tenants_to_dir}"
@@ -698,9 +698,8 @@ class NeonEnvBuilder:
        shutil.rmtree(self.repo_dir / "local_fs_remote_storage", ignore_errors=True)
        if self.test_overlay_dir is None:
            log.info("Copying local_fs_remote_storage directory from snapshot")
-            subprocess.run(
-                ["cp", "-a", f"{repo_dir / 'local_fs_remote_storage'}", f"{self.repo_dir}"],
-                check=True,
+            shutil.copytree(
+                repo_dir / "local_fs_remote_storage", self.repo_dir / "local_fs_remote_storage"
            )
        else:
            log.info("Creating overlayfs mount of local_fs_remote_storage directory from snapshot")
--- a/test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py
+++ b/test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py
@@ -14,7 +14,7 @@ from fixtures.neon_fixtures import (
    PgBin,
    wait_for_last_flush_lsn,
 )
-from fixtures.utils import get_scale_for_db, humantime_to_ms
+from fixtures.utils import get_scale_for_db, humantime_to_ms, skip_on_ci

 from performance.pageserver.util import setup_pageserver_with_tenants

@@ -36,6 +36,9 @@ if TYPE_CHECKING:
@pytest.mark.parametrize("pgbench_scale", [get_scale_for_db(200)])
@pytest.mark.parametrize("n_tenants", [500])
@pytest.mark.timeout(10000)
+@skip_on_ci(
+    "This test needs lot of resources and should run on dedicated HW, not in github action runners as part of CI"
+)
 def test_pageserver_characterize_throughput_with_n_tenants(
    neon_env_builder: NeonEnvBuilder,
    zenbenchmark: NeonBenchmarker,
@@ -60,6 +63,9 @@ def test_pageserver_characterize_throughput_with_n_tenants(
@pytest.mark.parametrize("n_clients", [1, 64])
@pytest.mark.parametrize("n_tenants", [1])
@pytest.mark.timeout(2400)
+@skip_on_ci(
+    "This test needs lot of resources and should run on dedicated HW, not in github action runners as part of CI"
+)
 def test_pageserver_characterize_latencies_with_1_client_and_throughput_with_many_clients_one_tenant(
    neon_env_builder: NeonEnvBuilder,
    zenbenchmark: NeonBenchmarker,
--- a/test_runner/performance/test_branching.py
+++ b/test_runner/performance/test_branching.py
@@ -70,9 +70,9 @@ def test_compare_child_and_root_write_perf(neon_compare: NeonCompare):
    endpoint_child = env.endpoints.create_start("child")

    with neon_compare.record_duration("root_run_duration"):
-        endpoint_root.safe_psql("INSERT INTO foo SELECT FROM generate_series(1,1000000)")
+        endpoint_root.safe_psql("INSERT INTO foo SELECT FROM generate_series(1,10000000)")  # 10x increase from 1000000
    with neon_compare.record_duration("child_run_duration"):
-        endpoint_child.safe_psql("INSERT INTO foo SELECT FROM generate_series(1,1000000)")
+        endpoint_child.safe_psql("INSERT INTO foo SELECT FROM generate_series(1,10000000)")  # 10x increase from 1000000


 def test_compare_child_and_root_read_perf(neon_compare: NeonCompare):
@@ -83,7 +83,7 @@ def test_compare_child_and_root_read_perf(neon_compare: NeonCompare):
    endpoint_root.safe_psql_many(
        [
            "CREATE TABLE foo(key serial primary key, t text default 'foooooooooooooooooooooooooooooooooooooooooooooooooooo')",
-            "INSERT INTO foo SELECT FROM generate_series(1,1000000)",
+            "INSERT INTO foo SELECT FROM generate_series(1,10000000)",  # 10x increase from 1000000
        ]
    )

--- a/test_runner/performance/test_copy.py
+++ b/test_runner/performance/test_copy.py
@@ -59,7 +59,7 @@ def test_copy(neon_with_baseline: PgCompare):
            # Since there's no data in the table previously, this extends it.
            with env.record_pageserver_writes("copy_extend_pageserver_writes"):
                with env.record_duration("copy_extend"):
-                    cur.copy_from(copy_test_data(1000000), "copytest")
+                    cur.copy_from(copy_test_data(10000000), "copytest")  # 10x increase from 1000000
                    env.flush()

            # Delete most rows, and VACUUM to make the space available for reuse.
@@ -79,7 +79,7 @@ def test_copy(neon_with_baseline: PgCompare):
            # This will also clear all the VM bits.
            with env.record_pageserver_writes("copy_reuse_pageserver_writes"):
                with env.record_duration("copy_reuse"):
-                    cur.copy_from(copy_test_data(1000000), "copytest")
+                    cur.copy_from(copy_test_data(10000000), "copytest")  # 10x increase from 1000000
                    env.flush()

            env.report_peak_memory_use()
--- a/test_runner/performance/test_cumulative_statistics_persistence.py
+++ b/test_runner/performance/test_cumulative_statistics_persistence.py
@@ -101,8 +101,8 @@ def test_cumulative_statistics_persistence(
    try:
        connstr = project["connection_uris"][0]["connection_uri"]
        env = connection_parameters_to_env(project["connection_uris"][0]["connection_parameters"])
-        # seed about 1 GiB of data into pgbench_accounts
-        pg_bin.run_capture(["pgbench", "-i", "-s68"], env=env)
+        # seed about 10 GiB of data into pgbench_accounts
+        pg_bin.run_capture(["pgbench", "-i", "-s680"], env=env)  # 10x increase from 68

        # assert rows in pgbench_accounts is 6800000 rows
        conn = psycopg2.connect(connstr)
--- a/test_runner/performance/test_latency.py
+++ b/test_runner/performance/test_latency.py
@@ -13,7 +13,7 @@ if TYPE_CHECKING:
    from fixtures.neon_fixtures import PgProtocol


-def start_write_workload(pg: PgProtocol, scale: int = 10):
+def start_write_workload(pg: PgProtocol, scale: int = 100):  # 10x increase from 10
    with pg.connect().cursor() as cur:
        cur.execute(f"create table big as select generate_series(1,{scale * 100_000})")

--- a/test_runner/performance/test_parallel_copy.py
+++ b/test_runner/performance/test_parallel_copy.py
@@ -17,13 +17,13 @@ async def repeat_bytes(buf, repetitions: int):

 async def copy_test_data_to_table(endpoint: Endpoint, worker_id: int, table_name: str):
    buf = BytesIO()
-    for i in range(1000):
+    for i in range(10000):  # 10x increase from 1000
        buf.write(
            f"{i}\tLoaded by worker {worker_id}. Long string to consume some space.\n".encode()
        )
    buf.seek(0)

-    copy_input = repeat_bytes(buf.read(), 5000)
+    copy_input = repeat_bytes(buf.read(), 50000)  # 10x increase from 5000

    pg_conn = await endpoint.connect_async()

--- a/test_runner/performance/test_parallel_copy_to.py
+++ b/test_runner/performance/test_parallel_copy_to.py
@@ -16,13 +16,13 @@ async def repeat_bytes(buf, repetitions: int):

 async def copy_test_data_to_table(pg: PgProtocol, worker_id: int, table_name: str):
    buf = BytesIO()
-    for i in range(1000):
+    for i in range(10000):  # 10x increase from 1000
        buf.write(
            f"{i}\tLoaded by worker {worker_id}. Long string to consume some space.\n".encode()
        )
    buf.seek(0)

-    copy_input = repeat_bytes(buf.read(), 5000)
+    copy_input = repeat_bytes(buf.read(), 50000)  # 10x increase from 5000

    pg_conn = await pg.connect_async()
    await pg_conn.copy_to_table(table_name, source=copy_input)
--- a/test_runner/performance/test_perf_pgbench.py
+++ b/test_runner/performance/test_perf_pgbench.py
@@ -181,7 +181,7 @@ def run_test_pgbench(env: PgCompare, scale: int, duration: int, workload_type: P
    env.report_size()


-def get_durations_matrix(default: int = 45) -> list[int]:
+def get_durations_matrix(default: int = 450) -> list[int]:  # 10x increase from 45
    durations = os.getenv("TEST_PG_BENCH_DURATIONS_MATRIX", default=str(default))
    rv = []
    for d in durations.split(","):
@@ -197,7 +197,7 @@ def get_durations_matrix(default: int = 45) -> list[int]:
    return rv


-def get_scales_matrix(default: int = 10) -> list[int]:
+def get_scales_matrix(default: int = 100) -> list[int]:  # 10x increase from 10
    scales = os.getenv("TEST_PG_BENCH_SCALES_MATRIX", default=str(default))
    rv = []
    for s in scales.split(","):
--- a/test_runner/performance/test_seqscans.py
+++ b/test_runner/performance/test_seqscans.py
@@ -18,13 +18,13 @@ if TYPE_CHECKING:
@pytest.mark.parametrize(
    "rows,iters,workers",
    [
-        # The test table is large enough (3-4 MB) that it doesn't fit in the compute node
+        # The test table is large enough (30-40 MB) that it doesn't fit in the compute node
        # cache, so the seqscans go to the page server. But small enough that it fits
        # into memory in the page server.
-        pytest.param(100000, 100, 0),
+        pytest.param(1000000, 100, 0),  # 10x increase from 100000
        # Also test with a larger table, with and without parallelism
-        pytest.param(10000000, 1, 0),
-        pytest.param(10000000, 1, 4),
+        pytest.param(100000000, 1, 0),  # 10x increase from 10000000
+        pytest.param(100000000, 1, 4),  # 10x increase from 10000000
    ],
 )
@pytest.mark.parametrize(
--- a/test_runner/performance/test_wal_backpressure.py
+++ b/test_runner/performance/test_wal_backpressure.py
@@ -69,9 +69,9 @@ def start_heavy_write_workload(env: PgCompare, n_tables: int, scale: int, num_it

    ## Single table workload:
    At each step, insert new `new_rows_each_update` rows.
-    The variable `new_rows_each_update` is equal to `scale * 100_000`.
+    The variable `new_rows_each_update` is equal to `scale * 1_000_000`.
    The number of steps is determined by `num_iters` variable."""
-    new_rows_each_update = scale * 100_000
+    new_rows_each_update = scale * 1_000_000  # 10x increase from 100_000

    def start_single_table_workload(table_id: int):
        for _ in range(num_iters):
--- a/test_runner/regress/test_attach_tenant_config.py
+++ b/test_runner/regress/test_attach_tenant_config.py
@@ -187,7 +187,6 @@ def test_fully_custom_config(positive_env: NeonEnv):
            "args": {"format": "bincode", "compression": {"zstd": {"level": 1}}},
        },
        "rel_size_v2_enabled": True,
-        "relsize_snapshot_cache_capacity": 10000,
        "gc_compaction_enabled": True,
        "gc_compaction_verification": False,
        "gc_compaction_initial_threshold_kb": 1024000,
--- a/test_runner/regress/test_basebackup.py
+++ b/test_runner/regress/test_basebackup.py
@@ -1,77 +0,0 @@
-from __future__ import annotations
-
-from typing import TYPE_CHECKING
-
-from fixtures.utils import wait_until
-
-if TYPE_CHECKING:
-    from fixtures.neon_fixtures import NeonEnvBuilder
-
-
-def test_basebackup_cache(neon_env_builder: NeonEnvBuilder):
-    """
-    Simple test for basebackup cache.
-    1. Check that we always hit the cache after compute restart.
-    2. Check that we eventually delete old basebackup files, but not the latest one.
-    3. Check that we delete basebackup file for timeline with active compute.
-    """
-
-    neon_env_builder.pageserver_config_override = """
-        tenant_config = { basebackup_cache_enabled = true }
-        basebackup_cache_config = { cleanup_period = '1s' }
-    """
-
-    env = neon_env_builder.init_start()
-    ep = env.endpoints.create("main")
-    ps = env.pageserver
-    ps_http = ps.http_client()
-
-    # 1. Check that we always hit the cache after compute restart.
-    for i in range(3):
-        ep.start()
-        ep.stop()
-
-        def check_metrics(i=i):
-            metrics = ps_http.get_metrics()
-            # Never miss.
-            # The first time compute_ctl sends `get_basebackup` with lsn=None, we do not cache such requests.
-            # All other requests should be a hit
-            assert (
-                metrics.query_one(
-                    "pageserver_basebackup_cache_read_total", {"result": "miss"}
-                ).value
-                == 0
-            )
-            # All but the first requests are hits.
-            assert (
-                metrics.query_one("pageserver_basebackup_cache_read_total", {"result": "hit"}).value
-                == i
-            )
-            # Every compute shut down should trigger a prepare reuest.
-            assert (
-                metrics.query_one(
-                    "pageserver_basebackup_cache_prepare_total", {"result": "ok"}
-                ).value
-                == i + 1
-            )
-
-        wait_until(check_metrics)
-
-    # 2. Check that we eventually delete old basebackup files, but not the latest one.
-    def check_bb_file_count():
-        bb_files = list(ps.workdir.joinpath("basebackup_cache").iterdir())
-        # tmp dir + 1 basebackup file.
-        assert len(bb_files) == 2
-
-    wait_until(check_bb_file_count)
-
-    # 3. Check that we delete basebackup file for timeline with active compute.
-    ep.start()
-    ep.safe_psql("create table t1 as select generate_series(1, 10) as n")
-
-    def check_bb_dir_empty():
-        bb_files = list(ps.workdir.joinpath("basebackup_cache").iterdir())
-        # only tmp dir.
-        assert len(bb_files) == 1
-
-    wait_until(check_bb_dir_empty)
--- a/test_runner/regress/test_compute_catalog.py
+++ b/test_runner/regress/test_compute_catalog.py
@@ -19,16 +19,6 @@ TEST_ROLE_NAMES = [
    {"name": "role$"},
    {"name": "role$$"},
    {"name": "role$x$"},
-    {"name": "x"},
-    {"name": "xx"},
-    {"name": "$x"},
-    {"name": "x$"},
-    {"name": "$x$"},
-    {"name": "xx$"},
-    {"name": "$xx"},
-    {"name": "$xx$"},
-    # 63 bytes is the limit for role/DB names in Postgres
-    {"name": "x" * 63},
 ]

 TEST_DB_NAMES = [
@@ -84,43 +74,6 @@ TEST_DB_NAMES = [
        "name": "db name$x$",
        "owner": "role$x$",
    },
-    {
-        "name": "x",
-        "owner": "x",
-    },
-    {
-        "name": "xx",
-        "owner": "xx",
-    },
-    {
-        "name": "$x",
-        "owner": "$x",
-    },
-    {
-        "name": "x$",
-        "owner": "x$",
-    },
-    {
-        "name": "$x$",
-        "owner": "$x$",
-    },
-    {
-        "name": "xx$",
-        "owner": "xx$",
-    },
-    {
-        "name": "$xx",
-        "owner": "$xx",
-    },
-    {
-        "name": "$xx$",
-        "owner": "$xx$",
-    },
-    # 63 bytes is the limit for role/DB names in Postgres
-    {
-        "name": "x" * 63,
-        "owner": "x" * 63,
-    },
 ]


@@ -193,10 +146,6 @@ def test_compute_create_drop_dbs_and_roles(neon_simple_env: NeonEnv):
    """
    Test that compute_ctl can create and work with databases and roles
    with special characters (whitespaces, %, tabs, etc.) in the name.
-    Also use `drop_subscriptions_before_start: true`. We do not actually
-    have any subscriptions in this test, so it should be no-op, but it
-    i) simulates the case when we create a second dev branch together with
-    a new project creation, and ii) just generally stresses more code paths.
    """
    env = neon_simple_env

@@ -210,7 +159,6 @@ def test_compute_create_drop_dbs_and_roles(neon_simple_env: NeonEnv):
        **{
            "spec": {
                "skip_pg_catalog_updates": False,
-                "drop_subscriptions_before_start": True,
                "cluster": {
                    "roles": TEST_ROLE_NAMES,
                    "databases": TEST_DB_NAMES,
@@ -254,7 +202,6 @@ def test_compute_create_drop_dbs_and_roles(neon_simple_env: NeonEnv):
        **{
            "spec": {
                "skip_pg_catalog_updates": False,
-                "drop_subscriptions_before_start": True,
                "cluster": {
                    "roles": [],
                    "databases": [],
--- a/test_runner/regress/test_pageserver_metric_collection.py
+++ b/test_runner/regress/test_pageserver_metric_collection.py
@@ -508,9 +508,6 @@ PER_METRIC_VERIFIERS = {
    "remote_storage_size": CannotVerifyAnything,
    "written_size": WrittenDataVerifier,
    "written_data_bytes_delta": WrittenDataDeltaVerifier,
-    "written_size_since_parent": WrittenDataVerifier,  # same as written_size on root
-    "pitr_cutoff": CannotVerifyAnything,
-    "pitr_history_size_since_parent": WrittenDataVerifier,  # same as written_size on root w/o GC
    "timeline_logical_size": CannotVerifyAnything,
    "synthetic_storage_size": SyntheticSizeVerifier,
 }
--- a/test_runner/regress/test_replica_start.py
+++ b/test_runner/regress/test_replica_start.py
@@ -27,9 +27,8 @@ from contextlib import closing

 import psycopg2
 import pytest
-from fixtures.common_types import Lsn
 from fixtures.log_helper import log
-from fixtures.neon_fixtures import NeonEnv, PgBin, wait_for_last_flush_lsn, wait_replica_caughtup
+from fixtures.neon_fixtures import NeonEnv, wait_for_last_flush_lsn, wait_replica_caughtup
 from fixtures.pg_version import PgVersion
 from fixtures.utils import query_scalar, skip_on_postgres, wait_until

@@ -696,110 +695,3 @@ def test_replica_start_with_too_many_unused_xids(neon_simple_env: NeonEnv):
    with secondary.cursor() as secondary_cur:
        secondary_cur.execute("select count(*) from t")
        assert secondary_cur.fetchone() == (n_restarts,)
-
-
-def test_ephemeral_endpoints_vacuum(neon_simple_env: NeonEnv, pg_bin: PgBin):
-    env = neon_simple_env
-    endpoint = env.endpoints.create_start("main")
-
-    sql = """
-CREATE TABLE CHAR_TBL(f1 char(4));
-CREATE TABLE FLOAT8_TBL(f1 float8);
-CREATE TABLE INT2_TBL(f1 int2);
-CREATE TABLE INT4_TBL(f1 int4);
-CREATE TABLE INT8_TBL(q1 int8, q2 int8);
-CREATE TABLE POINT_TBL(f1 point);
-CREATE TABLE TEXT_TBL (f1 text);
-CREATE TABLE VARCHAR_TBL(f1 varchar(4));
-CREATE TABLE onek (unique1		int4);
-CREATE TABLE onek2 AS SELECT * FROM onek;
-CREATE TABLE tenk1 (unique1		int4);
-CREATE TABLE tenk2 AS SELECT * FROM tenk1;
-CREATE TABLE person (name text, age int4,location point);
-CREATE TABLE emp (salary int4, manager name) INHERITS (person);
-CREATE TABLE student (gpa float8) INHERITS (person);
-CREATE TABLE stud_emp (	percent 	int4) INHERITS (emp, student);
-CREATE TABLE road (name		text,thepath 	path);
-CREATE TABLE ihighway () INHERITS (road);
-CREATE TABLE shighway(surface		text) INHERITS (road);
-CREATE TABLE BOOLTBL3 (d text, b bool, o int);
-CREATE TABLE booltbl4(isfalse bool, istrue bool, isnul bool);
-DROP TABLE BOOLTBL3;
-DROP TABLE BOOLTBL4;
-CREATE TABLE ceil_floor_round (a numeric);
-DROP TABLE ceil_floor_round;
-CREATE TABLE width_bucket_test (operand_num numeric, operand_f8 float8);
-DROP TABLE width_bucket_test;
-CREATE TABLE num_input_test (n1 numeric);
-CREATE TABLE num_variance (a numeric);
-INSERT INTO num_variance VALUES (0);
-CREATE TABLE snapshot_test (nr	integer, snap	txid_snapshot);
-CREATE TABLE guid1(guid_field UUID, text_field TEXT DEFAULT(now()));
-CREATE TABLE guid2(guid_field UUID, text_field TEXT DEFAULT(now()));
-CREATE INDEX guid1_btree ON guid1 USING BTREE (guid_field);
-CREATE INDEX guid1_hash  ON guid1 USING HASH  (guid_field);
-TRUNCATE guid1;
-DROP TABLE guid1;
-DROP TABLE guid2 CASCADE;
-CREATE TABLE numrange_test (nr NUMRANGE);
-CREATE INDEX numrange_test_btree on numrange_test(nr);
-CREATE TABLE numrange_test2(nr numrange);
-CREATE INDEX numrange_test2_hash_idx on numrange_test2 using hash (nr);
-INSERT INTO numrange_test2 VALUES('[, 5)');
-CREATE TABLE textrange_test (tr text);
-CREATE INDEX textrange_test_btree on textrange_test(tr);
-CREATE TABLE test_range_gist(ir int4range);
-CREATE INDEX test_range_gist_idx on test_range_gist using gist (ir);
-DROP INDEX test_range_gist_idx;
-CREATE INDEX test_range_gist_idx on test_range_gist using gist (ir);
-CREATE TABLE test_range_spgist(ir int4range);
-CREATE INDEX test_range_spgist_idx on test_range_spgist using spgist (ir);
-DROP INDEX test_range_spgist_idx;
-CREATE INDEX test_range_spgist_idx on test_range_spgist using spgist (ir);
-CREATE TABLE test_range_elem(i int4);
-CREATE INDEX test_range_elem_idx on test_range_elem (i);
-CREATE INDEX ON test_range_elem using spgist(int4range(i,i+10));
-DROP TABLE test_range_elem;
-CREATE TABLE test_range_excl(room int4range, speaker int4range, during tsrange, exclude using gist (room with =, during with &&), exclude using gist (speaker with =, during with &&));
-CREATE TABLE f_test(f text, i int);
-CREATE TABLE i8r_array (f1 int, f2 text);
-CREATE TYPE arrayrange as range (subtype=int4[]);
-CREATE TYPE two_ints as (a int, b int);
-DROP TYPE two_ints cascade;
-CREATE TABLE text_support_test (t text);
-CREATE TABLE TEMP_FLOAT (f1 FLOAT8);
-CREATE TABLE TEMP_INT4 (f1 INT4);
-CREATE TABLE TEMP_INT2 (f1 INT2);
-CREATE TABLE TEMP_GROUP (f1 INT4, f2 INT4, f3 FLOAT8);
-CREATE TABLE POLYGON_TBL(f1 polygon);
-CREATE TABLE quad_poly_tbl (id int, p polygon);
-INSERT INTO quad_poly_tbl SELECT (x - 1) * 100 + y, polygon(circle(point(x * 10, y * 10), 1 + (x + y) % 10)) FROM generate_series(1, 200) x, generate_series(1, 100) y;
-CREATE TABLE quad_poly_tbl_ord_seq2 AS SELECT 1 FROM quad_poly_tbl;
-CREATE TABLE quad_poly_tbl_ord_idx2 AS SELECT 1 FROM quad_poly_tbl;
-"""
-
-    with endpoint.cursor() as cur:
-        lsn = Lsn(query_scalar(cur, "SELECT pg_current_wal_flush_lsn()"))
-        env.endpoints.create_start(branch_name="main", lsn=lsn)
-        log.info(f"lsn: {lsn}")
-
-        for line in sql.split("\n"):
-            if len(line.strip()) == 0 or line.startswith("--"):
-                continue
-            cur.execute(line)
-
-        lsn = Lsn(query_scalar(cur, "SELECT pg_current_wal_flush_lsn()"))
-        env.endpoints.create_start(branch_name="main", lsn=lsn)
-        log.info(f"lsn: {lsn}")
-
-        cur.execute("VACUUM FULL pg_class;")
-
-    for ep in env.endpoints.endpoints:
-        log.info(f"{ep.endpoint_id} / {ep.pg_port}")
-        pg_dump_command = ["pg_dumpall", "-f", f"/tmp/dump-{ep.endpoint_id}.sql"]
-        env_vars = {
-            "PGPORT": str(ep.pg_port),
-            "PGUSER": endpoint.default_options["user"],
-            "PGHOST": endpoint.default_options["host"],
-        }
-        pg_bin.run_capture(pg_dump_command, env=env_vars)
--- a/test_runner/regress/test_timeline_archive.py
+++ b/test_runner/regress/test_timeline_archive.py
@@ -193,11 +193,6 @@ def test_timeline_offloading(neon_env_builder: NeonEnvBuilder, manual_offload: b
        "test_ancestor_branch_archive_branch1", tenant_id, "test_ancestor_branch_archive_parent"
    )

-    offloaded_count = ps_http.get_metric_value(
-        "pageserver_tenant_offloaded_timelines", {"tenant_id": f"{tenant_id}"}
-    )
-    assert offloaded_count == 0
-
    ps_http.timeline_archival_config(
        tenant_id,
        leaf_timeline_id,
@@ -249,11 +244,6 @@ def test_timeline_offloading(neon_env_builder: NeonEnvBuilder, manual_offload: b
    wait_until(leaf_offloaded)
    wait_until(parent_offloaded)

-    offloaded_count = ps_http.get_metric_value(
-        "pageserver_tenant_offloaded_timelines", {"tenant_id": f"{tenant_id}"}
-    )
-    assert offloaded_count == 2
-
    # Offloaded child timelines should still prevent deletion
    with pytest.raises(
        PageserverApiException,
--- a/workspace_hack/Cargo.toml
+++ b/workspace_hack/Cargo.toml
@@ -53,7 +53,8 @@ hmac = { version = "0.12", default-features = false, features = ["reset"] }
 hyper-582f2526e08bb6a0 = { package = "hyper", version = "0.14", features = ["client", "http1", "http2", "runtime", "server", "stream"] }
 hyper-dff4ba8e3ae991db = { package = "hyper", version = "1", features = ["full"] }
 hyper-util = { version = "0.1", features = ["client-legacy", "http1", "http2", "server", "service"] }
-indexmap = { version = "2", features = ["serde"] }
+indexmap-dff4ba8e3ae991db = { package = "indexmap", version = "1", default-features = false, features = ["std"] }
+indexmap-f595c2ba2a3f28df = { package = "indexmap", version = "2", features = ["serde"] }
 itertools = { version = "0.12" }
 lazy_static = { version = "1", default-features = false, features = ["spin_no_std"] }
 libc = { version = "0.2", features = ["extra_traits", "use_std"] }
@@ -81,7 +82,7 @@ regex-syntax = { version = "0.8" }
 reqwest = { version = "0.12", default-features = false, features = ["blocking", "json", "rustls-tls", "rustls-tls-native-roots", "stream"] }
 rustls = { version = "0.23", default-features = false, features = ["logging", "ring", "std", "tls12"] }
 rustls-pki-types = { version = "1", features = ["std"] }
-rustls-webpki = { version = "0.103", default-features = false, features = ["ring", "std"] }
+rustls-webpki = { version = "0.102", default-features = false, features = ["ring", "std"] }
 scopeguard = { version = "1" }
 sec1 = { version = "0.7", features = ["pem", "serde", "std", "subtle"] }
 serde = { version = "1", features = ["alloc", "derive"] }
@@ -101,11 +102,11 @@ tokio-rustls = { version = "0.26", default-features = false, features = ["loggin
 tokio-stream = { version = "0.1" }
 tokio-util = { version = "0.7", features = ["codec", "compat", "io", "rt"] }
 toml_edit = { version = "0.22", features = ["serde"] }
-tower = { version = "0.5", default-features = false, features = ["balance", "buffer", "limit", "log"] }
+tonic = { version = "0.12", default-features = false, features = ["codegen", "prost", "tls-roots"] }
+tower = { version = "0.4", default-features = false, features = ["balance", "buffer", "limit", "util"] }
 tracing = { version = "0.1", features = ["log"] }
 tracing-core = { version = "0.1" }
 tracing-log = { version = "0.2" }
-tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
 url = { version = "2", features = ["serde"] }
 uuid = { version = "1", features = ["serde", "v4", "v7"] }
 zeroize = { version = "1", features = ["derive", "serde"] }
@@ -123,7 +124,8 @@ either = { version = "1" }
 getrandom = { version = "0.2", default-features = false, features = ["std"] }
 half = { version = "2", default-features = false, features = ["num-traits"] }
 hashbrown = { version = "0.14", features = ["raw"] }
-indexmap = { version = "2", features = ["serde"] }
+indexmap-dff4ba8e3ae991db = { package = "indexmap", version = "1", default-features = false, features = ["std"] }
+indexmap-f595c2ba2a3f28df = { package = "indexmap", version = "2", features = ["serde"] }
 itertools = { version = "0.12" }
 libc = { version = "0.2", features = ["extra_traits", "use_std"] }
 log = { version = "0.4", default-features = false, features = ["std"] }